diff --git a/examples/02_strategy.py b/examples/02_strategy.py new file mode 100644 index 0000000..c3d123c --- /dev/null +++ b/examples/02_strategy.py @@ -0,0 +1,156 @@ +""" +02_strategy.py +============== + +A complete moving-average crossover strategy using mt5bridge-ccxt. +Drop this into Freqtrade / Jesse / your own runner. +""" + +import time +import signal +import sys +from datetime import datetime + +import pandas as pd + +import mt5bridge_ccxt + + +class MAStrategy: + """Simple MA crossover strategy. + + Buy when fast MA crosses above slow MA (golden cross). + Sell when fast MA crosses below slow MA (death cross). + """ + + def __init__(self, exchange, symbol, timeframe="1h", fast=20, slow=60, + volume=0.01, sl_distance=5.0, tp_distance=10.0, magic=12345): + self.exchange = exchange + self.symbol = symbol + self.timeframe = timeframe + self.fast = fast + self.slow = slow + self.volume = volume + self.sl_distance = sl_distance + self.tp_distance = tp_distance + self.magic = magic + + def fetch_bars(self, n=None): + """Fetch OHLCV bars and compute MAs.""" + n = n or (self.slow + 5) + ohlcv = self.exchange.fetch_ohlcv(self.symbol, self.timeframe, limit=n) + df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"]) + df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True) + df.set_index("datetime", inplace=True) + df["ma_fast"] = df["close"].rolling(self.fast).mean() + df["ma_slow"] = df["close"].rolling(self.slow).mean() + return df + + def signal(self, df): + """Return 1 (buy), -1 (sell), or 0 (no signal) using closed bars.""" + if len(df) < self.slow + 2: + return 0 + prev = df.iloc[-3] + curr = df.iloc[-2] + if pd.isna(prev["ma_fast"]) or pd.isna(curr["ma_fast"]): + return 0 + if prev["ma_fast"] <= prev["ma_slow"] and curr["ma_fast"] > curr["ma_slow"]: + return 1 + if prev["ma_fast"] >= prev["ma_slow"] and curr["ma_fast"] < curr["ma_slow"]: + return -1 + return 0 + + def has_position(self): + """Check if we already have a position in this symbol.""" + positions = self.exchange.fetch_positions([self.symbol]) + return len(positions) > 0 + + def run_once(self): + """Run one iteration of the strategy.""" + df = self.fetch_bars() + sig = self.signal(df) + ticker = self.exchange.fetch_ticker(self.symbol) + bid, ask = ticker["bid"], ticker["ask"] + balance = self.exchange.fetch_balance() + equity = list(balance["total"].values())[0] + + print(f"[{datetime.now():%Y-%m-%d %H:%M:%S}] {self.symbol} " + f"bid={bid} ask={ask} equity={equity:.2f} signal={sig}") + + if sig == 0: + return + + if self.has_position(): + return + + if sig == 1: + order = self.exchange.create_order( + self.symbol, "market", "buy", + amount=self.volume, + price=ask, + params={ + "sl": ask - self.sl_distance, + "tp": ask + self.tp_distance, + "magic": self.magic, + "comment": "MA cross buy", + }, + ) + print(f" -> BUY order placed #{order['id']}") + elif sig == -1: + order = self.exchange.create_order( + self.symbol, "market", "sell", + amount=self.volume, + price=bid, + params={ + "sl": bid + self.sl_distance, + "tp": bid - self.tp_distance, + "magic": self.magic, + "comment": "MA cross sell", + }, + ) + print(f" -> SELL order placed #{order['id']}") + + def run_forever(self, interval=60): + """Run the strategy in a loop.""" + stop = {"flag": False} + + def _stop(signum, frame): + stop["flag"] = True + print("\nStopping...") + + signal.signal(signal.SIGINT, _stop) + signal.signal(signal.SIGTERM, _stop) + + print(f"Starting strategy on {self.symbol} {self.timeframe}, " + f"fast={self.fast} slow={self.slow}, poll every {interval}s") + while not stop["flag"]: + try: + self.run_once() + except Exception as e: + print(f" Error: {e}") + time.sleep(interval) + + +if __name__ == "__main__": + exchange = mt5bridge_ccxt.mt5bridge({ + "apiKey": "your-api-key", + "host": "http://localhost:8080", + "symbols": {"XAU/USD": "XAUUSDc"}, + }) + + strategy = MAStrategy( + exchange=exchange, + symbol="XAU/USD", + timeframe="1h", + fast=20, + slow=60, + volume=0.01, + sl_distance=5.0, + tp_distance=10.0, + ) + + # Single iteration (for testing) + strategy.run_once() + + # Continuous loop (uncomment for live trading) + # strategy.run_forever(interval=60)