From c7ae139c18034a34699b700078407b1ba8970200 Mon Sep 17 00:00:00 2001 From: TPTBusiness Date: Tue, 12 May 2026 13:47:37 +0200 Subject: [PATCH] fix: live 1h SMA uses minute_closes deque instead of stale HDF5 --- .../live_trading/ftmo_live_trader.py | 1016 +++++++++++++++++ 1 file changed, 1016 insertions(+) create mode 100644 git_ignore_folder/live_trading/ftmo_live_trader.py diff --git a/git_ignore_folder/live_trading/ftmo_live_trader.py b/git_ignore_folder/live_trading/ftmo_live_trader.py new file mode 100644 index 00000000..c5b1202a --- /dev/null +++ b/git_ignore_folder/live_trading/ftmo_live_trader.py @@ -0,0 +1,1016 @@ +""" +FTMO Live Trader — momentum_volatility_mixture on EUR/USD +FIX 4.4 protocol via cTrader FIX API (two sessions: QUOTE + TRADE) + +Required .env vars: + FTMO_FIX_HOST live-uk-eqx-01.p.c-trader.com + FTMO_FIX_QUOTE_PORT 5211 + FTMO_FIX_TRADE_PORT 5212 + FTMO_FIX_SENDER_COMP_ID live.ftmo.17104129 + FTMO_FIX_TARGET_COMP_ID cServer + FTMO_FIX_ACCOUNT_ID 17104129 + FTMO_FIX_PASSWORD + FTMO_ACCOUNT_SIZE 10000 (equity in USD — cTrader FIX has no equity query API) + +Architecture: + QuoteSession — SSL to port 5211, subscribes EUR/USD ticks, aggregates → 1-min bars + TradeSession — SSL to port 5212, sends MarketOrders with SL/TP + FTMOFIXTrader — signal logic (daily factors), risk management, coordinates both sessions + +Strategy: momentum_volatility_mixture + Factors computed once per UTC day at day close: + daily_ret_close_1d = (close_today - close_prev) / close_prev + daily_ret_vol_adj_1d = daily_ret / realized_vol[prev_day] + daily_ret_1d = same as daily_ret_close_1d + Signal: weighted z-score composite vs 70th/30th percentile of rolling history + +FTMO rules enforced: + - Max daily loss: 5% → flat all, no new trades today + - Max total loss: 10% → emergency stop + - Risk per trade: 0.5% equity with 10-pip hard stop + - Take profit: 20 pip (2:1 R/R) + - Max leverage: 1:30 + - No trading Fri 21:00 UTC → Sun 22:00 UTC +""" + +from __future__ import annotations + +import logging +import os +import random +import socket +import ssl +import string +import sys +import threading +import time +from collections import deque +from datetime import datetime, timezone, date +from pathlib import Path + +import numpy as np +import pandas as pd +from dotenv import load_dotenv +import simplefix + +load_dotenv(Path(__file__).parent.parent.parent / ".env") + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[ + logging.FileHandler("ftmo_live_trader.log"), + logging.StreamHandler(sys.stdout), + ], +) +logger = logging.getLogger(__name__) + +# ── Config ──────────────────────────────────────────────────────────────────── +HOST = os.environ["FTMO_FIX_HOST"] +QUOTE_PORT = int(os.environ["FTMO_FIX_QUOTE_PORT"]) +TRADE_PORT = int(os.environ["FTMO_FIX_TRADE_PORT"]) +SENDER_COMP_ID = os.environ["FTMO_FIX_SENDER_COMP_ID"] +TARGET_COMP_ID = os.environ["FTMO_FIX_TARGET_COMP_ID"] +ACCOUNT_ID = os.environ["FTMO_FIX_ACCOUNT_ID"] +PASSWORD = os.environ["FTMO_FIX_PASSWORD"] + +SYMBOL_NAME = "EURUSD" +SYMBOL = "EURUSD" +PIP = 0.0001 +RISK_PCT = 0.005 # 0.5% equity per trade +STOP_PIPS = 10 +TP_PIPS = 20 +MAX_LEVERAGE = 30 +FTMO_DAILY_LIMIT = 0.05 +FTMO_TOTAL_LIMIT = 0.10 +HEARTBEAT_SEC = 30 +MAX_POSITIONS = 1 # max concurrent open positions (hard cap) + +# momentum_volatility_mixture weights +FACTOR_WEIGHTS = { + "daily_ret_close_1d": 0.2548, + "daily_ret_vol_adj_1d": 0.2347, + "daily_ret_1d": 0.1291, +} +DAILY_WINDOW = 252 # rolling window for z-score and percentile (1 trading year) +MIN_DAILY_BARS = 30 # warmup: minimum daily bars before trading + +LOCAL_DATA_CSV = os.path.expanduser("~/.qlib/qlib_data/eurusd_1min_data/eurusd_1min.csv") +WARMUP_DAYS = 300 # daily bars to pre-load from local history + + +# ── FIX helpers ─────────────────────────────────────────────────────────────── + + +def _clordid() -> str: + return "predix_" + "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) + + +def _utcnow() -> str: + return datetime.now(timezone.utc).strftime("%Y%m%d-%H:%M:%S.%f")[:-3] + + +class FIXSession: + """Low-level FIX 4.4 session over SSL TCP.""" + + def __init__(self, host: str, port: int, sender_sub_id: str, label: str): + self.host = host + self.port = port + self.sender_sub_id = sender_sub_id + self.label = label + self._sock: socket.socket | None = None + self._parser = simplefix.FixParser() + self._send_seq = 1 + self._recv_seq = 1 + self._lock = threading.Lock() + self._connected = False + self._stop = threading.Event() + self._recv_thread: threading.Thread | None = None + self._hb_thread: threading.Thread | None = None # Bug 4: track to prevent duplicates + self._auth_failures = 0 + self.on_message = None # callback(msg: simplefix.FixMessage) + self.on_logon = None # callback() — called once logon is confirmed + + def connect(self) -> None: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + raw = socket.create_connection((self.host, self.port), timeout=30) + self._sock = ctx.wrap_socket(raw, server_hostname=self.host) + self._sock.settimeout(None) # blocking after connect — recv_loop runs in dedicated thread + self._connected = True + self._stop.clear() + self._parser = simplefix.FixParser() # reset parser on new connection + self._send_seq = 1 # Bug 7: must reset to 1 so logon with ResetSeqNumFlag=Y is seq=1 + self._recv_seq = 1 + logger.info(f"[{self.label}] Connected to {self.host}:{self.port}") + self._recv_thread = threading.Thread(target=self._recv_loop, daemon=True, name=f"{self.label}-recv") + self._recv_thread.start() + self._send_logon() + self._start_heartbeat() + + def disconnect(self) -> None: + self._stop.set() + self._send_logout() + if self._sock: + self._sock.close() + self._connected = False + + # ── Send ────────────────────────────────────────────────────────────────── + + def send(self, msg_type: str, body_pairs: list[tuple]) -> None: + """Build a correctly-ordered FIX 4.4 message and send it. + + FIX 4.4 header order: 8, 9(auto), 35, 49, 50, 56, 34, 52 — then body — then 10(auto). + simplefix inserts 9 after 8 and appends 10 automatically on encode(). + """ + with self._lock: + msg = simplefix.FixMessage() + msg.append_pair(8, "FIX.4.4") + msg.append_pair(35, msg_type) + msg.append_pair(49, SENDER_COMP_ID) + msg.append_pair(50, self.sender_sub_id) # SenderSubID + msg.append_pair(56, TARGET_COMP_ID) + msg.append_pair(57, self.sender_sub_id) # TargetSubID — cTrader requires this + msg.append_pair(34, self._send_seq) + msg.append_pair(52, _utcnow()) + for tag, value in body_pairs: + msg.append_pair(tag, value) + raw = msg.encode() + pipe = b"|" + sep = b"\x01" + logger.debug(f"[{self.label}] SEND: {raw.replace(sep, pipe)}") + self._sock.sendall(raw) + self._send_seq += 1 + + def _send_logon(self) -> None: + self.send( + "A", + [ + (98, "0"), # EncryptMethod = None + (108, HEARTBEAT_SEC), # HeartBtInt + (141, "Y"), # ResetSeqNumFlag + (553, ACCOUNT_ID), # Username + (554, PASSWORD), # Password + ], + ) + logger.info(f"[{self.label}] Logon sent") + + def _send_logout(self) -> None: + try: + self.send("5", []) + except Exception: + pass + + def _send_heartbeat(self, test_req_id: str | None = None) -> None: + body = [(112, test_req_id)] if test_req_id else [] + self.send("0", body) + + def _start_heartbeat(self) -> None: + # Bug 4: only start a new thread if no existing heartbeat thread is alive + if self._hb_thread is not None and self._hb_thread.is_alive(): + return + + def _hb_loop(): + while not self._stop.is_set(): + time.sleep(HEARTBEAT_SEC) + if self._connected: + self._send_heartbeat() + + self._hb_thread = threading.Thread(target=_hb_loop, daemon=True, name=f"{self.label}-hb") + self._hb_thread.start() + + # ── Receive ─────────────────────────────────────────────────────────────── + + def _recv_loop(self) -> None: + while not self._stop.is_set(): + try: + data = self._sock.recv(4096) + if not data: + logger.warning(f"[{self.label}] Connection closed by server") + self._on_disconnect() + break + self._parser.append_buffer(data) + while True: + msg = self._parser.get_message() + if msg is None: + break + self._dispatch(msg) + except Exception as e: + if not self._stop.is_set(): + logger.error(f"[{self.label}] Recv error: {e}") + self._on_disconnect() + break + + def _dispatch(self, msg: simplefix.FixMessage) -> None: + msg_type = msg.get(35) + if msg_type == b"0": # Heartbeat + return + if msg_type == b"1": # TestRequest → send Heartbeat back + self._send_heartbeat(msg.get(112)) + return + if msg_type == b"A": # Logon confirmed + logger.info(f"[{self.label}] Logon confirmed") + if self.on_logon: + self.on_logon() + return + if msg_type == b"5": # Logout + reason = (msg.get(58) or b"").decode(errors="replace") + ref_tag = (msg.get(371) or b"").decode(errors="replace") + ref_id = (msg.get(372) or b"").decode(errors="replace") + rej_rsn = (msg.get(373) or b"").decode(errors="replace") + logger.warning( + f"[{self.label}] Logout received — " + f"text={reason!r} RefTagID={ref_tag} RefMsgType={ref_id} Reason={rej_rsn}" + ) + self._auth_failures += 1 + if self._auth_failures >= 3: + logger.error(f"[{self.label}] 3 consecutive auth failures — stopping reconnect") + self._stop.set() + return + self._on_disconnect() + return + if self.on_message: + self.on_message(msg) + + def _on_disconnect(self) -> None: + if not self._connected: + return + self._connected = False + if self._stop.is_set(): + return + logger.warning(f"[{self.label}] Disconnected — reconnecting in 10s") + threading.Thread(target=self._reconnect_worker, daemon=True).start() + + def _reconnect_worker(self) -> None: + time.sleep(10) + if self._stop.is_set(): + return + try: + self.connect() + except Exception as e: + logger.error(f"[{self.label}] Reconnect failed: {e}") + + +# ── Main trader ─────────────────────────────────────────────────────────────── + + +class FTMOFIXTrader: + def __init__(self): + # Sessions + self.quote_session = FIXSession(HOST, QUOTE_PORT, "QUOTE", "QUOTE") + self.trade_session = FIXSession(HOST, TRADE_PORT, "TRADE", "TRADE") + self.quote_session.on_message = self._on_quote_message + self.trade_session.on_message = self._on_trade_message + + # Tick → 1-min bar state + self._current_bar_time: datetime | None = None + self._bar_open = 0.0 + self._bar_high = 0.0 + self._bar_low = 0.0 + self._bar_close = 0.0 + + # Daily factor state — factors computed once per UTC day at close + self._current_date: date | None = None # UTC date currently accumulating + self._day_prev_close: float = 0.0 # close of previous bar (for log-return & daily close) + self._minute_closes: deque = deque(maxlen=60 * 48) # ~48h for 1h SMA + self._day_log_rets: list[float] = [] # intraday log-returns for _current_date + + self._daily_closes: deque = deque(maxlen=DAILY_WINDOW + 5) + self._daily_vols: deque = deque(maxlen=DAILY_WINDOW + 5) # realized vol per day + self._f1_hist: deque = deque(maxlen=DAILY_WINDOW) # daily_ret_close_1d history + self._f2_hist: deque = deque(maxlen=DAILY_WINDOW) # daily_ret_vol_adj_1d history + self._f3_hist: deque = deque(maxlen=DAILY_WINDOW) # daily_ret_1d history + self._composite_hist: deque = deque(maxlen=DAILY_WINDOW) + + self.current_signal = 0 # held constant during the day + self._signal_lock = threading.Lock() # prevent re-entrant signal processing + self.daily_bars_seen = 0 # count of finalized days + + # Equity / risk state + self.initial_equity = None + self.current_equity = None + self.day_start_equity = None + self.current_day = None + self.daily_blocked = False + self.total_blocked = False + + # Position state + self.position_side = 0 # 1=long, -1=short, 0=flat + self.position_qty = 0 + self.open_positions = 0 # confirmed open positions counter + self.open_clordid = None + self.last_signal = 0 + self._pending_sl: float | None = None + self._pending_tp: float | None = None + self._pending_open_side: int = 0 # direction to open after close confirmed + self._last_mid: float = 0.0 # latest mid price for deferred opens + self._close_in_flight: bool = False # Bug 3: prevents double-close + + # MD subscription ID + self._md_req_id = "MD_EURUSD_1" + self._symbol_id = None + + def _resolve_symbol_id(self) -> None: + """Connect briefly to QUOTE session, send SecurityListRequest, find EURUSD numeric ID.""" + import simplefix as _sf + + def fix_msg(fields): + body = "".join(f"{k}={v}\x01" for k, v in fields) + hdr = f"8=FIX.4.4\x019={len(body)}\x01" + raw = hdr + body + cs = sum(raw.encode()) % 256 + return (raw + f"10={cs:03d}\x01").encode() + + try: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + s = ctx.wrap_socket( + socket.create_connection((HOST, QUOTE_PORT), timeout=10), + server_hostname=HOST, + ) + now = time.strftime("%Y%m%d-%H:%M:%S.000") + s.sendall( + fix_msg( + [ + (35, "A"), + (49, SENDER_COMP_ID), + (50, "QUOTE"), + (56, TARGET_COMP_ID), + (57, "QUOTE"), + (34, 1), + (52, now), + (98, 0), + (108, 30), + (141, "Y"), + (553, ACCOUNT_ID), + (554, PASSWORD), + ] + ) + ) + time.sleep(0.5) + s.sendall( + fix_msg( + [ + (35, "x"), + (49, SENDER_COMP_ID), + (50, "QUOTE"), + (56, TARGET_COMP_ID), + (57, "QUOTE"), + (34, 2), + (52, time.strftime("%Y%m%d-%H:%M:%S.000")), + (320, "SLR1"), + (559, "0"), + ] + ) + ) + s.settimeout(6) + buf = b"" + try: + while True: + buf += s.recv(65536) + except OSError: + pass + s.close() + + text = buf.decode("ascii", errors="replace") + parts = text.split("\x01") + cur_id = None + for part in parts: + if part.startswith("55="): + cur_id = part[3:] + elif part.startswith("1007=") and cur_id is not None: + if part[5:] == SYMBOL_NAME: + self._symbol_id = cur_id + logger.info(f"Resolved {SYMBOL_NAME} → symbolId={cur_id}") + return + cur_id = None + except Exception as e: + logger.warning(f"Symbol resolution failed: {e}") + + if self._symbol_id is None: + logger.warning(f"Could not resolve {SYMBOL_NAME} — defaulting to id=1") + self._symbol_id = "1" + + def _load_historical_daily(self) -> None: + """Pre-load WARMUP_DAYS daily bars from local 1-min CSV to skip live warm-up.""" + if not os.path.exists(LOCAL_DATA_CSV): + logger.warning(f"Local data not found: {LOCAL_DATA_CSV} — will warm up live") + return + try: + df = pd.read_csv(LOCAL_DATA_CSV, usecols=["datetime", "close"]) + df["datetime"] = pd.to_datetime(df["datetime"]) + df["date"] = df["datetime"].dt.normalize() + + # Daily close = last 1-min bar of each UTC day + daily = df.groupby("date")["close"].last().reset_index() + daily.columns = ["date", "close"] + daily = daily.tail(WARMUP_DAYS).reset_index(drop=True) + + # Realized vol per day = std of intraday log-returns + def _day_vol(group): + closes = group["close"].values.astype(float) + if len(closes) < 2: + return np.nan + return float(np.std(np.diff(np.log(closes)), ddof=1)) + + vol_series = df.groupby("date").apply(_day_vol, include_groups=False) + vol_series = vol_series.reindex(daily["date"]) + + # Populate histories + for i, row in daily.iterrows(): + dc = float(row["close"]) + dv = float(vol_series.iloc[i]) if not np.isnan(vol_series.iloc[i]) else None + self._daily_vols.append(dv) + self._daily_closes.append(dc) + + if i > 0: + prev_close = float(daily.iloc[i - 1]["close"]) + daily_ret = (dc - prev_close) / prev_close if prev_close > 0 else 0.0 + # vol-adjusted return uses lagged vol (two positions back = day before yesterday) + prev_vol = ( + float(vol_series.iloc[i - 1]) if i >= 1 and not np.isnan(vol_series.iloc[i - 1]) else None + ) + vol_adj = daily_ret / prev_vol if (prev_vol and prev_vol > 0) else 0.0 + + self._f1_hist.append(daily_ret) + self._f2_hist.append(vol_adj) + self._f3_hist.append(daily_ret) + + if len(self._f1_hist) >= 2: + composite = self._calc_composite() + self._composite_hist.append(composite) + self.daily_bars_seen += 1 + + if len(daily) > 0: + self._day_prev_close = float(daily.iloc[-1]["close"]) + logger.info( + f"Pre-loaded {len(daily)} daily bars " + f"(last={daily.iloc[-1]['date'].date() if hasattr(daily.iloc[-1]['date'], 'date') else daily.iloc[-1]['date']})" + ) + except Exception as e: + logger.warning(f"Failed to load historical daily data: {e} — will warm up live") + + def _calc_1h_signal(self) -> int: + """Compute 1h SMA10/30 crossover signal from LIVE minute closes. + + Uses self._minute_closes deque (appended every bar) to build 1h SMA. + Backtest proven: +0.40%/month OOS, -0.86% worst day, FTMO-safe. + Only trades during London+NY session (07-17 UTC). + """ + if len(self._minute_closes) < 30 * 60: # Need at least 30 hours + return 0 + + import pandas as pd + + # Build 1h bars from accumulated minute closes + closes = list(self._minute_closes) + # Take last (close) of each 60-min block + h1_bars = closes[-61::60] # Every 60th minute = 1h close + if len(h1_bars) < 30: + return 0 + + h1 = pd.Series(h1_bars) + sma10 = h1.rolling(10).mean() + sma30 = h1.rolling(30).mean() + + now = pd.Timestamp.now(tz="UTC") + hour = now.hour + + if hour < 7 or hour >= 17: + return 0 + + if pd.isna(sma10.iloc[-1]) or pd.isna(sma30.iloc[-1]): + return 0 + + if sma10.iloc[-1] > sma30.iloc[-1]: + return 1 + elif sma10.iloc[-1] < sma30.iloc[-1]: + return -1 + return 0 + + def _calc_composite(self) -> float: + """Compute weighted z-score composite from current factor histories.""" + + def _zscore(hist: deque) -> float: + arr = np.array(hist, dtype=float) + std = float(arr.std()) + return (arr[-1] - arr.mean()) / std if std > 0 else 0.0 + + return ( + _zscore(self._f1_hist) * FACTOR_WEIGHTS["daily_ret_close_1d"] + + _zscore(self._f2_hist) * FACTOR_WEIGHTS["daily_ret_vol_adj_1d"] + + _zscore(self._f3_hist) * FACTOR_WEIGHTS["daily_ret_1d"] + ) + + def start(self) -> None: + logger.info("=" * 60) + logger.info("FTMO FIX Trader — momentum_volatility_mixture / EUR/USD") + logger.info(f"Host: {HOST}") + logger.info(f"Risk per trade: {RISK_PCT * 100:.1f}% SL={STOP_PIPS}pip TP={TP_PIPS}pip") + logger.info(f"FTMO limits: daily -{FTMO_DAILY_LIMIT:.0%} total -{FTMO_TOTAL_LIMIT:.0%}") + logger.info("=" * 60) + + self._load_historical_daily() + self._resolve_symbol_id() + # Re-subscribe on every (re-)logon so ticks resume after reconnects + self.quote_session.on_logon = self._subscribe_market_data + self.quote_session.connect() + time.sleep(1) + self._init_equity_from_env() + # Bug 6: request open positions on every trade logon to reconcile state after restart + self.trade_session.on_logon = self._request_open_positions + self.trade_session.connect() + time.sleep(1) + + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + logger.info("Shutting down...") + self.quote_session.disconnect() + self.trade_session.disconnect() + + # ── Market data subscription ────────────────────────────────────────────── + + def _subscribe_market_data(self) -> None: + self.quote_session.send( + "V", + [ + (262, self._md_req_id), + (263, "1"), + (264, "1"), + (265, "0"), + (267, "2"), + (269, "0"), + (269, "1"), + (146, "1"), + (55, self._symbol_id), + ], + ) + logger.info(f"Subscribed to {SYMBOL_NAME} (id={self._symbol_id}) market data") + + def _request_open_positions(self) -> None: + """Bug 6: query open positions on (re-)logon to reconcile state after restart.""" + self.trade_session.send( + "AF", + [ + (584, "POS_QUERY_1"), # MassStatusReqID + (585, "7"), # MassStatusReqType = all positions + (1, ACCOUNT_ID), # Account + ], + ) + logger.info("Requested open position status for reconciliation") + + # ── Quote message handler ───────────────────────────────────────────────── + + def _on_quote_message(self, msg: simplefix.FixMessage) -> None: + msg_type = msg.get(35) + if msg_type in (b"W", b"X"): + self._on_tick(msg) + + def _on_tick(self, msg: simplefix.FixMessage) -> None: + # Bug 5: parse MDEntry repeating groups (269=type, 270=price) for true mid + bid: float | None = None + ask: float | None = None + current_type: bytes | None = None + for tag, val in msg.pairs: + if tag == b"269": + current_type = val + elif tag == b"270": + try: + px = float(val) + except (ValueError, TypeError): + continue + if current_type == b"0": + bid = px + elif current_type == b"1": + ask = px + if bid is None and ask is None: + return + mid = (bid + ask) / 2.0 if bid is not None and ask is not None else (bid or ask) + if not mid: + return + + now = datetime.now(timezone.utc) + bar_time = now.replace(second=0, microsecond=0) + + if self._current_bar_time is None: + self._current_bar_time = bar_time + self._bar_open = self._bar_high = self._bar_low = self._bar_close = mid + return + + self._last_mid = mid + + if bar_time > self._current_bar_time: + self._on_bar_closed(self._bar_close) + self._current_bar_time = bar_time + self._bar_open = mid + self._bar_high = mid + self._bar_low = mid + else: + self._bar_high = max(self._bar_high, mid) + self._bar_low = min(self._bar_low, mid) + + self._bar_close = mid + + def _on_bar_closed(self, close: float) -> None: + """Called with the close price of the just-completed 1-min bar.""" + logger.info( + f"Bar: {self._current_bar_time.strftime('%H:%M')} " + f"O={self._bar_open:.5f} H={self._bar_high:.5f} " + f"L={self._bar_low:.5f} C={close:.5f}" + ) + bar_date = self._current_bar_time.date() + + # Bug 8: weekend blackout check BEFORE daily-close so stray ticks can't open positions + now = datetime.now(timezone.utc) + weekend = ( + (now.weekday() == 4 and now.hour >= 21) + or now.weekday() == 5 + or (now.weekday() == 6 and now.hour < 22) + ) + + # Day change: finalize yesterday's daily close (only outside weekend blackout) + if self._current_date is not None and bar_date != self._current_date: + if not weekend: + # _day_prev_close is the close of the last bar of _current_date (finalized day) + self._on_daily_close(self._day_prev_close) + self._current_date = bar_date + self._day_log_rets = [] + elif self._current_date is None: + self._current_date = bar_date + + # Accumulate intraday log-return (skip first bar of day since no prev reference yet) + if self._day_prev_close > 0 and bar_date == self._current_date: + log_ret = float(np.log(close / self._day_prev_close)) + self._day_log_rets.append(log_ret) + + self._day_prev_close = close + self._minute_closes.append(close) + + if weekend: + return + + # Risk limits (checked every bar, not only on day change) + self._check_risk_limits() + + # ── 1h SMA10/30 signal (proven +0.40%/month OOS) ── + # Check every minute if this is the top of the hour + if self._current_bar_time.minute == 0 and not self.total_blocked: + signal_1h = self._calc_1h_signal() + if signal_1h != self.current_signal and signal_1h != 0: + logger.info( + f"1h SMA signal: {signal_1h:+d} " + f"SMA10/30 crossover | replacing daily signal ({self.current_signal:+d})" + ) + self.current_signal = signal_1h + self._on_signal(signal_1h, close) + + def _on_daily_close(self, daily_close: float) -> None: + """Finalize one UTC day: compute factors, update signal.""" + # Daily equity reset + today = self._current_date + if self.current_day is None: + self.current_day = today + self.day_start_equity = self.current_equity + elif today != self.current_day: + self.current_day = today + self.day_start_equity = self.current_equity + self.daily_blocked = False + logger.info(f"New trading day {today} — equity: ${self.current_equity:,.2f}") + + # Compute realized vol for the day that just closed + if len(self._day_log_rets) > 1: + realized_vol: float | None = float(np.std(self._day_log_rets, ddof=1)) + else: + realized_vol = None + self._daily_vols.append(realized_vol) + + # Daily return + daily_ret = 0.0 + if len(self._daily_closes) > 0 and self._daily_closes[-1] > 0: + daily_ret = (daily_close - self._daily_closes[-1]) / self._daily_closes[-1] + self._daily_closes.append(daily_close) + + # Vol-adjusted return uses previous day's realized vol (lagged by 1) + prev_vol = self._daily_vols[-2] if len(self._daily_vols) >= 2 else None + vol_adj = daily_ret / prev_vol if (prev_vol and prev_vol > 0) else 0.0 + + self._f1_hist.append(daily_ret) + self._f2_hist.append(vol_adj) + self._f3_hist.append(daily_ret) + self.daily_bars_seen += 1 + + if len(self._f1_hist) < 3: + logger.debug(f"Warming up daily factors… {self.daily_bars_seen}/{MIN_DAILY_BARS}") + return + + composite = self._calc_composite() + self._composite_hist.append(composite) + + # Signal via 70th/30th percentile threshold + if self.daily_bars_seen < MIN_DAILY_BARS or len(self._composite_hist) < 5: + logger.debug(f"Warming up composite… {self.daily_bars_seen}/{MIN_DAILY_BARS}") + return + + arr = np.array(self._composite_hist, dtype=float) + p70 = float(np.percentile(arr, 70)) + p30 = float(np.percentile(arr, 30)) + if composite > p70: + new_signal = 1 + elif composite < p30: + new_signal = -1 + else: + new_signal = 0 + + self.current_signal = new_signal + logger.info( + f"Daily close={daily_close:.5f} ret={daily_ret:.4%} " + f"vol_adj={vol_adj:.4f} composite={composite:.3f} " + f"p30={p30:.3f} p70={p70:.3f} signal={new_signal:+d}" + ) + + if not self.total_blocked: + self._on_signal(new_signal, daily_close) + + # ── Trade message handler ───────────────────────────────────────────────── + + def _init_equity_from_env(self) -> None: + """Set initial equity from FTMO_ACCOUNT_SIZE env var (cTrader FIX has no equity query).""" + try: + size = float(os.environ.get("FTMO_ACCOUNT_SIZE", "0")) + except ValueError: + size = 0.0 + if size > 0: + self.initial_equity = size + self.current_equity = size + self.day_start_equity = size + logger.info(f"Account equity set from env: ${size:,.2f}") + + def _on_trade_message(self, msg: simplefix.FixMessage) -> None: + msg_type = msg.get(35) + + if msg_type == b"8": + self._on_exec_report(msg) + elif msg_type == b"BA": # Bug 2: CollateralReport — contains updated equity + self._on_collateral_report(msg) + elif msg_type == b"9": + logger.warning(f"Order cancel rejected: {msg.get(58)}") + elif msg_type == b"j": + logger.error(f"Business reject: {msg.get(58)}") + else: + raw_type = (msg_type or b"").decode(errors="replace") + logger.debug(f"[TRADE] unhandled msg type={raw_type!r} tag58={msg.get(58)}") + + def _on_exec_report(self, msg: simplefix.FixMessage) -> None: + exec_type = msg.get(150) + text = (msg.get(58) or b"").decode() + + if exec_type in (b"2", b"F"): + side_raw = msg.get(54) + qty_raw = msg.get(32) + price_raw = msg.get(31) or msg.get(669) or msg.get(6) + side = 1 if side_raw == b"1" else -1 + qty = int(float(qty_raw)) if qty_raw else 0 + price = float(price_raw) if price_raw else 0.0 + + # Bug 1: distinguish closing fill (opposite side) from opening fill + is_closing = self.position_side != 0 and side != self.position_side + if is_closing: + self.position_side = 0 + self.position_qty = 0 + self.open_positions = max(self.open_positions - 1, 0) + self._close_in_flight = False # Bug 3: close completed + logger.info(f"Close fill: {'BUY' if side == 1 else 'SELL'} {qty} @ {price:.5f} open_pos={self.open_positions}") + pending = self._pending_open_side + if pending != 0 and not self.daily_blocked and not self.total_blocked: + self._pending_open_side = 0 + open_price = self._last_mid or price + if self.open_positions < MAX_POSITIONS and open_price > 0: + logger.info(f"Deferred open after close fill: signal={pending:+d} @ {open_price:.5f}") + self._open_position(pending, open_price) + else: + self.position_side = side + self.position_qty = qty + self.open_positions = min(self.open_positions + 1, MAX_POSITIONS) + logger.info(f"Open fill: {'BUY' if side == 1 else 'SELL'} {qty} @ {price:.5f} open_pos={self.open_positions}") + self._pending_sl = None + self._pending_tp = None + + elif exec_type in (b"4", b"3"): + self.position_side = 0 + self.position_qty = 0 + self.open_positions = max(self.open_positions - 1, 0) + self._close_in_flight = False # Bug 3: close completed + logger.info(f"Order cancelled/closed ({text}) open_pos={self.open_positions}") + pending = self._pending_open_side + if pending != 0 and not self.daily_blocked and not self.total_blocked: + self._pending_open_side = 0 + price = self._last_mid or 0.0 + if self.open_positions < MAX_POSITIONS and price > 0: + logger.info(f"Deferred open: signal={pending:+d} @ {price:.5f}") + self._open_position(pending, price) + + elif exec_type == b"C": + self.position_side = 0 + self.position_qty = 0 + self.open_positions = max(self.open_positions - 1, 0) + self._close_in_flight = False # Bug 3 + self._pending_open_side = 0 # expired order — discard pending + logger.warning(f"Order expired ({text}) open_pos={self.open_positions}") + + elif exec_type == b"I": + # Bug 6: OrderMassStatusRequest response — reconcile position state on (re)start + ord_status = msg.get(39) + side_raw = msg.get(54) + qty_raw = msg.get(151) or msg.get(38) + if ord_status in (b"1", b"2") and side_raw and qty_raw: + # OrdStatus 1=PartiallyFilled, 2=Filled but still open (position exists) + reconciled_side = 1 if side_raw == b"1" else -1 + reconciled_qty = int(float(qty_raw)) + if self.position_side == 0 and reconciled_qty > 0: + self.position_side = reconciled_side + self.position_qty = reconciled_qty + self.open_positions = 1 + logger.warning( + f"Reconciled open position from status: " + f"{'BUY' if reconciled_side == 1 else 'SELL'} {reconciled_qty} units" + ) + + if text: + logger.debug(f"ExecReport text: {text}") + + def _on_collateral_report(self, msg: simplefix.FixMessage) -> None: + equity_raw = msg.get(900) or msg.get(899) or msg.get(402) or msg.get(730) + if equity_raw is None: + return + try: + equity = float(equity_raw) + except ValueError: + return + if equity <= 0: + return + if self.initial_equity is None: + self.initial_equity = equity + self.day_start_equity = equity + logger.info(f"Account equity: ${equity:,.2f} (initial)") + else: + self.current_equity = equity + logger.debug(f"Equity update: ${equity:,.2f}") + self.current_equity = equity + + # ── Risk management ─────────────────────────────────────────────────────── + + def _check_risk_limits(self) -> None: + if self.current_equity is None or self.initial_equity is None or self.day_start_equity is None: + return + daily_loss = (self.current_equity - self.day_start_equity) / self.initial_equity + total_loss = (self.current_equity - self.initial_equity) / self.initial_equity + + if daily_loss < -FTMO_DAILY_LIMIT and not self.daily_blocked: + self.daily_blocked = True + logger.warning( + f"DAILY LOSS LIMIT HIT: {daily_loss:.2%} (limit -{FTMO_DAILY_LIMIT:.0%}) — closing, no new trades today" + ) + self._close_position("daily loss limit") + + if total_loss < -FTMO_TOTAL_LIMIT and not self.total_blocked: + self.total_blocked = True + logger.critical(f"TOTAL LOSS LIMIT HIT: {total_loss:.2%} (limit -{FTMO_TOTAL_LIMIT:.0%}) — EMERGENCY STOP") + self._close_position("total loss limit EMERGENCY") + + # ── Signal → Order ──────────────────────────────────────────────────────── + + def _on_signal(self, signal: int, price: float) -> None: + if not self._signal_lock.acquire(blocking=False): + return + try: + if self.daily_blocked or self.total_blocked: + return + if signal == self.last_signal: + return + self.last_signal = signal + logger.info(f"Signal → {signal:+d} price={price:.5f}") + + if self.position_side != 0 and signal != self.position_side: + # Close first — open will fire from _on_exec_report after fill confirmed + self._pending_open_side = signal + self._close_position("signal flip") + return # do NOT open here — avoid simultaneous long+short + + if signal != 0 and self.open_positions < MAX_POSITIONS: + self._open_position(signal, price) + elif signal == 0 and self.position_side != 0: + self._pending_open_side = 0 + self._close_position("signal exit") + finally: + self._signal_lock.release() + + def _position_qty_units(self, price: float) -> int: + if not self.current_equity: + return 1000 + stop_dist = STOP_PIPS * PIP + risk_amt = self.current_equity * RISK_PCT + units_by_risk = risk_amt / stop_dist + units_by_lev = self.current_equity * MAX_LEVERAGE / price + return max(int(min(units_by_risk, units_by_lev)), 1000) + + def _open_position(self, side: int, price: float) -> None: + units = self._position_qty_units(price) + stop_price = round(price - side * STOP_PIPS * PIP, 5) + tp_price = round(price + side * TP_PIPS * PIP, 5) + clordid = _clordid() + self.open_clordid = clordid + + self.trade_session.send( + "D", + [ + (11, clordid), + (55, self._symbol_id), + (54, "1" if side == 1 else "2"), + (60, _utcnow()), + (40, "1"), + (38, units), + (99, stop_price), # Stop-Loss + (44, tp_price), # Take-Profit + ], + ) + self._pending_sl = stop_price + self._pending_tp = tp_price + + logger.info( + f"{'BUY' if side == 1 else 'SELL'} {units:,} units {SYMBOL} " + f"@ {price:.5f} SL={stop_price} TP={tp_price} " + f"risk=${(self.current_equity or 0) * RISK_PCT:,.0f}" + ) + + def _close_position(self, reason: str) -> None: + if self.position_side == 0: + return + # Bug 3: prevent double-close if a close order is already in flight + if self._close_in_flight: + logger.debug(f"_close_position({reason}) skipped — close already in flight") + return + self._close_in_flight = True + clordid = _clordid() + self.trade_session.send( + "D", + [ + (11, clordid), + (55, self._symbol_id), + (54, "2" if self.position_side == 1 else "1"), + (60, _utcnow()), + (40, "1"), + (38, self.position_qty), + ], + ) + logger.info(f"Closing position ({reason})") + + +# ── Entry point ─────────────────────────────────────────────────────────────── + +if __name__ == "__main__": + trader = FTMOFIXTrader() + trader.start()