157 lines
4.8 KiB
Python
157 lines
4.8 KiB
Python
"""
|
|
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)
|