#!/usr/bin/env python3 """ live_signal_monitor_v3.py ========================== Unified live execution monitor — EURUSD + USDJPY macro mean reversion models. CHANGES FROM v2 (2026-05-13): EURUSD signal generation switched from 1H formula to validated 15M formula: • Signal cadence: every 15 minutes (was hourly) • Z-score computation: on 15M bars, pct_change(24) = 6h return, rolling(60) = 15h window (was 24h return / 60h window on 1H data) • Session window adjusted to UTC 7-16 (matches backtest) • All other infrastructure unchanged: MT5, telegram, watchdog, position management, dashboard hooks, daily summaries USDJPY UNCHANGED: USDJPY uses daily spread z-score (already matches backtest in v2). USDJPY signal check remains hourly. Architecture: PAIR_CONFIG drives everything. • Each pair has its own parameters, state, armed signal, and trade log. • EURUSD uses the rolling-beta lag z-score on 15M bars (Formula A). • USDJPY uses the simple spread z-score (US-JP 2Y, z30) — unchanged. • Both share: MT5 connection, rate scheduler, Telegram, dashboard state. Pairs: EURUSD z±2.75 TP=0.20% SL=0.25% hold=52h z_exit=±1.5 signal every 15m USDJPY z±2.00 TP=0.70% SL=0.40% hold=24h z_exit=off signal hourly Run from project root: python src/execution/live_signal_monitor_v3.py """ import os import sys import time import json import logging import threading import urllib.parse import urllib.request from pathlib import Path from datetime import datetime, timezone, timedelta, date import numpy as np import pandas as pd import schedule BASE_PATH = Path(__file__).resolve().parents[2] if str(BASE_PATH / "src") not in sys.path: sys.path.insert(0, str(BASE_PATH / "src")) # ═══════════════════════════════════════════════════════════════════════════ # PAIR CONFIGURATION — all pair-specific parameters in one place # ═══════════════════════════════════════════════════════════════════════════ PAIR_CONFIG = { "EURUSD": { "symbol": "EURUSD", "threshold": 2.75, "fib": 0.786, "hold_hours": 52, "stop_pct": 0.0025, "tp_pct": 0.0020, "z_exit": 1.5, # reversion threshold (None = off) "session_start": 9, # EET hour — EET 9-18 = UTC 7-16 matches backtest "session_end": 18, "z_window": 60, # 60 bars on 15M = 15h normalisation window "return_lookback": 24, # pct_change(24) on 15M = 6h return (validated model) "signal_timeframe": "15m", # SIGNAL CADENCE — every 15 min on 15M bars "sizing_bands": [(2.75, 3.50, 1.0), (3.50, 4.50, 1.5), (4.50, 99.0, 2.0)], "pip_size": 0.00010, # 4-decimal pair "rate_files": ["us2y.csv", "de2y.csv", "us10y.csv", "de10y.csv"], "spread_cols": ("us2y", "de2y"), # for simple z-score fallback "signal_type": "beta_model_15m", # uses rolling OLS, signals on 15M bars "bars_15m": 15000, # 15000 × 15M = ~6 months of trading days # gives ~125+ daily closes for the 120-day OLS rolling window "trade_log": "trade_history_eurusd.json", "state_file": "state_eurusd.json", "dp": 5, # decimal places for price display }, "USDJPY": { "symbol": "USDJPY", "threshold": 2.0, "fib": 0.786, "hold_hours": 24, "stop_pct": 0.0040, "tp_pct": 0.0070, "z_exit": None, # confirmed not needed "session_start": 7, # EET hour (London session) — unchanged from v2 "session_end": 17, "z_window": 30, "signal_timeframe": "1h", # SIGNAL CADENCE — hourly (unchanged from v2) "sizing_bands": [(2.0, 2.5, 1.0), (2.5, 3.5, 1.5), (3.5, 99.0, 2.0)], "pip_size": 0.010, # JPY pair (2 decimal places) "rate_files": ["us2y.csv", "jp2y.csv"], "spread_cols": ("us2y", "jp2y"), "signal_type": "spread_zscore", # simple z-score of spread level "bars_1h": 300, "trade_log": "trade_history_usdjpy.json", "state_file": "state_usdjpy.json", "dp": 3, # decimal places for price display }, } # ── Shared constants ─────────────────────────────────────────────────────── BASE_RISK_PCT = 0.0075 BETA_WINDOW = 120 SMOOTH_SPAN = 20 RATES_DIR = BASE_PATH / "data" / "raw" / "rates" # ── Logging ──────────────────────────────────────────────────────────────── LOG_DIR = BASE_PATH / "data" / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) # Daily summary deduplification sentinel _LAST_DAILY_SUMMARY_SENT = None logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(), # Write to same log file as v2 so dashboard_server.py picks up v3 events # without changes. v3 messages will appear inline with v2 history. logging.FileHandler(LOG_DIR / "live_monitor_v2.log", encoding="utf-8"), ], ) log = logging.getLogger(__name__) # ── Telegram ─────────────────────────────────────────────────────────────── TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", "8794831706:AAGzqHNhyvfWL2lG_kCQUupacilRf7GfNSI") # TELEGRAM_CHAT_ID is the ADMIN/OWNER chat ID — the only one allowed to run # /add /remove /list commands. Extra recipients are managed via the bot commands # and stored in data/logs/telegram_recipients.json. TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "1031071259") # Recipients file location — set once LOG_DIR is defined, below _TG_RECIPIENTS_FILE = None def _get_recipients_file(): """Resolve the recipients file path (needs LOG_DIR which is set later in module).""" global _TG_RECIPIENTS_FILE if _TG_RECIPIENTS_FILE is None: _TG_RECIPIENTS_FILE = LOG_DIR / "telegram_recipients.json" return _TG_RECIPIENTS_FILE def _load_recipients(): """ Return list of active chat IDs (strings). Always includes admin. File format: {"recipients": [{"chat_id": "123", "added_at": "...", "name": "..."}]} """ admin = str(TELEGRAM_CHAT_ID).strip() if TELEGRAM_CHAT_ID else None # Support legacy comma-separated env var as well env_list = [c.strip() for c in str(TELEGRAM_CHAT_ID).split(",") if c.strip()] if TELEGRAM_CHAT_ID else [] all_ids = set(env_list) try: path = _get_recipients_file() if path.exists(): data = json.loads(path.read_text(encoding="utf-8")) for r in data.get("recipients", []): cid = str(r.get("chat_id", "")).strip() if cid: all_ids.add(cid) except Exception as e: log.warning(f"Recipients file read failed: {e}") return sorted(all_ids) def _save_recipients(recipients_list): """Write recipients file. Atomic write to avoid partial reads.""" try: path = _get_recipients_file() payload = {"recipients": recipients_list, "updated_at": datetime.now().isoformat()} tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(payload, indent=2), encoding="utf-8") import os as _os _os.replace(str(tmp), str(path)) except Exception as e: log.warning(f"Recipients file save failed: {e}") def _is_admin(chat_id): """Only the configured admin chat ID can manage recipients.""" admin = str(TELEGRAM_CHAT_ID).split(",")[0].strip() if TELEGRAM_CHAT_ID else None return admin and str(chat_id).strip() == admin # ─── Dashboard access tokens (Option B) ────────────────────────────────── # Per-user random tokens stored in dashboard_tokens.json. Dashboard server # checks ?key= on each request and serves the page only if the token # is valid. Admin grants tokens via /grantdashboard, revokes via # /revokedashboard, lists via /listdashboard. Users request access via # /requestdashboard which notifies the admin. DASHBOARD_BASE_URL = os.getenv("DASHBOARD_BASE_URL", "http://193.38.138.152:8765/dashboard.html") _TG_DASHBOARD_TOKENS_FILE = None _pending_dashboard_requests = {} # chat_id -> {name, requested_at} def _get_dashboard_tokens_file(): global _TG_DASHBOARD_TOKENS_FILE if _TG_DASHBOARD_TOKENS_FILE is None: _TG_DASHBOARD_TOKENS_FILE = LOG_DIR / "dashboard_tokens.json" return _TG_DASHBOARD_TOKENS_FILE def _load_dashboard_tokens(): """Return list of dicts: {chat_id, name, token, granted_at, is_admin}.""" try: path = _get_dashboard_tokens_file() if not path.exists(): return [] data = json.loads(path.read_text(encoding="utf-8")) return data.get("tokens", []) except Exception as e: log.warning(f"Dashboard tokens file read failed: {e}") return [] def _save_dashboard_tokens(tokens_list): """Atomic write of dashboard tokens file.""" try: path = _get_dashboard_tokens_file() payload = {"tokens": tokens_list, "updated_at": datetime.now().isoformat()} tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(payload, indent=2), encoding="utf-8") import os as _os _os.replace(str(tmp), str(path)) except Exception as e: log.warning(f"Dashboard tokens save failed: {e}") def _generate_dashboard_token(): """Generate a 16-char URL-safe random token.""" import secrets return secrets.token_urlsafe(12) # ~16 chars def _grant_dashboard_token(chat_id, name, is_admin=False): """Grant (or refresh) a dashboard token for a chat_id. Returns the token.""" tokens = _load_dashboard_tokens() chat_id = str(chat_id).strip() # If already has a token, reuse it (idempotent grant) for t in tokens: if str(t.get("chat_id")) == chat_id: return t.get("token") token = _generate_dashboard_token() tokens.append({ "chat_id": chat_id, "name": name or "", "token": token, "granted_at": datetime.now().isoformat(), "is_admin": bool(is_admin), }) _save_dashboard_tokens(tokens) return token def _revoke_dashboard_token(chat_id): """Remove a chat_id's dashboard token. Returns True if removed.""" tokens = _load_dashboard_tokens() chat_id = str(chat_id).strip() new_tokens = [t for t in tokens if str(t.get("chat_id")) != chat_id] if len(new_tokens) < len(tokens): _save_dashboard_tokens(new_tokens) return True return False def send_telegram(message: str, chat_ids=None): """ Send message to all registered recipients. Pass explicit chat_ids (list or single) to target specific recipients instead — used by command handlers. """ if not TELEGRAM_TOKEN: return if chat_ids is None: targets = _load_recipients() elif isinstance(chat_ids, (list, tuple, set)): targets = [str(c) for c in chat_ids] else: targets = [str(chat_ids)] if not targets: return try: import ssl url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage" # SSL bypass — required on Windows Python 3.14 (self-signed cert chain) ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE for cid in targets: try: data = (f"chat_id={cid}" f"&text={urllib.parse.quote(message)}" f"&parse_mode=HTML") req = urllib.request.Request(url, data=data.encode(), method="POST") with urllib.request.urlopen(req, timeout=10, context=ctx): pass except Exception as e: log.warning(f"Telegram send to {cid} failed: {e}") except Exception as e: log.warning(f"Telegram send failed: {e}") def send_telegram_with_results(message: str, chat_ids): """ Like send_telegram but returns (success_count, fail_count). Used by /announcement command to confirm delivery to admin. """ if not TELEGRAM_TOKEN or not chat_ids: return 0, 0 targets = [str(c) for c in chat_ids] if isinstance(chat_ids, (list, tuple, set)) else [str(chat_ids)] success = 0 fail = 0 try: import ssl url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE for cid in targets: try: data = (f"chat_id={cid}" f"&text={urllib.parse.quote(message)}" f"&parse_mode=HTML") req = urllib.request.Request(url, data=data.encode(), method="POST") with urllib.request.urlopen(req, timeout=10, context=ctx): pass success += 1 except Exception as e: fail += 1 log.warning(f"Telegram send to {cid} failed: {e}") except Exception as e: log.warning(f"Telegram send failed: {e}") return success, fail # ── Telegram command handler ─────────────────────────────────────────────── # Polls getUpdates every 30s, processes incoming commands, replies to sender only. _telegram_update_offset = 0 _pending_access_requests = {} # chat_id -> {name, requested_at} def poll_telegram_commands(): """ Poll Telegram's getUpdates for incoming messages and process commands. Runs every 30 seconds in the scheduler. Always non-blocking. """ global _telegram_update_offset if not TELEGRAM_TOKEN: return try: import ssl url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/getUpdates" params = f"?timeout=1&offset={_telegram_update_offset + 1}&limit=20" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE req = urllib.request.Request(url + params, method="GET") with urllib.request.urlopen(req, timeout=5, context=ctx) as resp: body = json.loads(resp.read().decode("utf-8")) if not body.get("ok"): return for update in body.get("result", []): _telegram_update_offset = max(_telegram_update_offset, update.get("update_id", 0)) msg = update.get("message") or update.get("channel_post") if not msg: continue chat_id = str(msg.get("chat", {}).get("id", "")) text = (msg.get("text") or "").strip() first_name = msg.get("from", {}).get("first_name", "") username = msg.get("from", {}).get("username", "") if chat_id and text: _handle_telegram_command(chat_id, text, first_name, username) except Exception as e: # Log quietly — getUpdates is noisy if network blips log.debug(f"Telegram poll failed: {e}") def _handle_telegram_command(chat_id, text, first_name="", username=""): """Process a single incoming Telegram command.""" text = text.strip() parts = text.split() cmd = parts[0].lower() if parts else "" args = parts[1:] is_admin = _is_admin(chat_id) # Build a friendly name for logging display_name = username or first_name or "unknown" # ── Public commands (anyone) ─────────────────────────────────────────── if cmd in ("/start", "/help"): if is_admin: send_telegram( "🔐 Admin commands available\n" "────────────────\n" "/list — show all recipients\n" "/add <chat_id> [name] — grant alert access\n" "/remove <chat_id> — revoke alert access\n" "/pending — see alert requests\n" "/announcement <message> — broadcast to everyone\n" "────────────────\n" "📺 Dashboard access\n" "/listdashboard — show dashboard recipients\n" "/grantdashboard <chat_id> [name] — grant dashboard URL\n" "/revokedashboard <chat_id> — revoke dashboard access\n" "────────────────\n" "/whoami — your chat ID\n" "────────────────\n" "Commands must come from your admin chat only.", chat_ids=[chat_id] ) else: in_list = chat_id in _load_recipients() disclaimer = ( "\n────────────────\n" "⚠️ DISCLAIMER\n\n" "This bot is provided for educational and informational " "purposes only.\n\n" "The signals, alerts, and commentary shared here are the " "output of a systematic quantitative model and do not " "constitute financial advice, investment advice, " "trading advice, or a recommendation to buy or sell any " "security or currency pair.\n\n" "Past performance of the model in backtesting or paper " "trading is not indicative of future results. Systematic " "trading involves substantial risk of loss. You should " "consult with a licensed financial advisor before making " "any investment decisions.\n\n" "The bot operator accepts no liability for any losses " "incurred from acting on information shared via this bot. " "You are solely responsible for your own trading " "decisions.\n\n" "By using this bot you acknowledge and accept these terms." ) if in_list: send_telegram( "✅ You're subscribed to alerts.\n" "────────────────\n" "Available commands:\n" "/whoami — see your chat ID\n" "/leave — stop receiving alerts\n" "/requestdashboard — request dashboard URL\n" "/help — show this message again" + disclaimer, chat_ids=[chat_id] ) else: send_telegram( "👋 Welcome\n" "────────────────\n" "You are NOT currently authorised to receive alerts.\n\n" "Available commands:\n" "/requestaccess — request access to trade alerts\n" "/requestdashboard — request dashboard URL\n" "/whoami — see your chat ID\n" "/help — show this message again" + disclaimer, chat_ids=[chat_id] ) return if cmd == "/whoami": send_telegram( f"Your chat ID: {chat_id}\n" f"Name: {display_name}", chat_ids=[chat_id] ) return if cmd == "/requestaccess": _pending_access_requests[chat_id] = { "name": display_name, "requested_at": datetime.now().isoformat(), } # Notify admin of the request admin_id = str(TELEGRAM_CHAT_ID).split(",")[0].strip() if admin_id: send_telegram( f"📥 ACCESS REQUEST\n" f"────────────────\n" f"From: {display_name}\n" f"Chat ID: {chat_id}\n" f"────────────────\n" f"To approve: /add {chat_id} {display_name}\n" f"To ignore: do nothing", chat_ids=[admin_id] ) send_telegram( "✅ Access request sent to the owner.\n" "You'll be notified if approved.", chat_ids=[chat_id] ) return # ── /requestdashboard — request dashboard URL access (anyone) ───────── if cmd == "/requestdashboard": # If already granted, just resend the URL silently existing = next((t for t in _load_dashboard_tokens() if str(t.get("chat_id")) == chat_id), None) if existing: url = f"{DASHBOARD_BASE_URL}?key={existing['token']}" send_telegram( f"✅ You already have dashboard access\n" f"────────────────\n" f"Your private link:\n" f"{url}\n\n" f"This link is personal to you. Do not share it — " f"if it leaks, /requestdashboard again and I can issue a new one.", chat_ids=[chat_id] ) return # Not already granted — log the request and notify admin _pending_dashboard_requests[chat_id] = { "name": display_name, "requested_at": datetime.now().isoformat(), } admin_id = str(TELEGRAM_CHAT_ID).split(",")[0].strip() if admin_id: send_telegram( f"📺 DASHBOARD REQUEST\n" f"────────────────\n" f"From: {display_name}\n" f"Chat ID: {chat_id}\n" f"────────────────\n" f"To approve: /grantdashboard {chat_id} {display_name}\n" f"To ignore: do nothing", chat_ids=[admin_id] ) send_telegram( "✅ Dashboard access request sent to the owner.\n" "You'll be notified if approved.", chat_ids=[chat_id] ) return # ── /grantdashboard [name] — admin only ───────────────────── if cmd == "/grantdashboard": if not is_admin: send_telegram("Admin command only.", chat_ids=[chat_id]) return if not args: send_telegram( "Usage: /grantdashboard <chat_id> [name]", chat_ids=[chat_id] ) return target_id = args[0].strip() name = " ".join(args[1:]) if len(args) > 1 else "" # Prefer name from pending requests if available if not name and target_id in _pending_dashboard_requests: name = _pending_dashboard_requests[target_id].get("name", "") token = _grant_dashboard_token(target_id, name, is_admin=False) _pending_dashboard_requests.pop(target_id, None) url = f"{DASHBOARD_BASE_URL}?key={token}" # Notify the granted user send_telegram( f"🎉 Dashboard access granted\n" f"────────────────\n" f"Your private link:\n" f"{url}\n\n" f"This link is personal to you. Do not share it — " f"if it leaks, /requestdashboard again and I can issue a new one.", chat_ids=[target_id] ) # Confirm to admin send_telegram( f"✅ Dashboard access granted to {name or target_id}\n" f"Token: {token}", chat_ids=[chat_id] ) return # ── /revokedashboard — admin only ─────────────────────────── if cmd == "/revokedashboard": if not is_admin: send_telegram("Admin command only.", chat_ids=[chat_id]) return if not args: send_telegram( "Usage: /revokedashboard <chat_id>", chat_ids=[chat_id] ) return target_id = args[0].strip() if _revoke_dashboard_token(target_id): send_telegram( f"✅ Dashboard access revoked for {target_id}", chat_ids=[chat_id] ) else: send_telegram( f"⚠️ No dashboard token found for {target_id}", chat_ids=[chat_id] ) return # ── /listdashboard — admin only ─────────────────────────────────────── if cmd == "/listdashboard": if not is_admin: send_telegram("Admin command only.", chat_ids=[chat_id]) return tokens = _load_dashboard_tokens() if not tokens: send_telegram("No dashboard tokens granted.", chat_ids=[chat_id]) return lines = ["📺 Dashboard access", "────────────────"] for t in tokens: admin_tag = " (admin)" if t.get("is_admin") else "" lines.append( f"• {t.get('name') or '(no name)'}{admin_tag}\n" f" ID: {t.get('chat_id')}\n" f" Granted: {t.get('granted_at', '')[:10]}" ) send_telegram("\n".join(lines), chat_ids=[chat_id]) return if cmd == "/leave": recipients = _load_recipients() if chat_id in recipients and not is_admin: # Rewrite file without this chat_id try: path = _get_recipients_file() data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {"recipients": []} data["recipients"] = [r for r in data.get("recipients", []) if str(r.get("chat_id")) != chat_id] _save_recipients(data["recipients"]) send_telegram( "👋 You've been removed from the alert list.\n" "Use /requestaccess to rejoin later.", chat_ids=[chat_id] ) # Notify admin admin_id = str(TELEGRAM_CHAT_ID).split(",")[0].strip() if admin_id: send_telegram( f"ℹ️ {display_name} ({chat_id}) left the alert list.", chat_ids=[admin_id] ) except Exception as e: log.warning(f"Leave command failed: {e}") elif is_admin: send_telegram("Admin cannot /leave. Edit TELEGRAM_CHAT_ID env var if needed.", chat_ids=[chat_id]) else: send_telegram("You're not currently on the alert list.", chat_ids=[chat_id]) return # ── Admin-only commands ──────────────────────────────────────────────── if cmd in ("/add", "/remove", "/list", "/pending", "/announcement"): if not is_admin: send_telegram( "🚫 That command is admin-only.\n" "Use /requestaccess to request access.", chat_ids=[chat_id] ) return if cmd == "/list": recipients = _load_recipients() try: path = _get_recipients_file() data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {"recipients": []} rich = {str(r.get("chat_id")): r for r in data.get("recipients", [])} except Exception: rich = {} if not recipients: send_telegram("No recipients configured.", chat_ids=[chat_id]) else: lines = ["👥 ACTIVE RECIPIENTS", "────────────────"] for cid in recipients: admin_tag = " 👑 ADMIN" if _is_admin(cid) else "" info = rich.get(cid, {}) name = info.get("name", "") added = info.get("added_at", "")[:10] if info.get("added_at") else "" line = f"{cid}" if name: line += f" · {name}" if added: line += f" · added {added}" line += admin_tag lines.append(line) lines.append("────────────────") lines.append(f"Total: {len(recipients)}") send_telegram("\n".join(lines), chat_ids=[chat_id]) return if cmd == "/pending": if not _pending_access_requests: send_telegram("📭 No pending access requests.", chat_ids=[chat_id]) else: lines = ["📥 PENDING REQUESTS", "────────────────"] for req_cid, info in _pending_access_requests.items(): name = info.get("name", "") when = info.get("requested_at", "")[:16].replace("T", " ") lines.append(f"{req_cid} · {name} · {when}") lines.append(f" Approve: /add {req_cid} {name}") send_telegram("\n".join(lines), chat_ids=[chat_id]) return if cmd == "/add": if not args: send_telegram( "Usage: /add <chat_id> [name]\n" "Example: /add 987654321 John", chat_ids=[chat_id] ) return new_id = args[0].strip() name = " ".join(args[1:]) if len(args) > 1 else "" try: path = _get_recipients_file() data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {"recipients": []} existing = data.get("recipients", []) if any(str(r.get("chat_id")) == new_id for r in existing): send_telegram(f"ℹ️ {new_id} is already on the list.", chat_ids=[chat_id]) return # If name not provided, use pending request data if available if not name and new_id in _pending_access_requests: name = _pending_access_requests[new_id].get("name", "") existing.append({ "chat_id": new_id, "name": name, "added_at": datetime.now().isoformat(), "added_by": chat_id, }) _save_recipients(existing) _pending_access_requests.pop(new_id, None) send_telegram( f"✅ Added {new_id}" + (f" ({name})" if name else "") + " to the alert list.", chat_ids=[chat_id] ) # Notify the newly added user send_telegram( "🎉 You're in!\n" "────────────────\n" "The owner has granted you access to the macro model alerts. " "You'll start receiving them on the next signal check.\n\n" "Commands:\n" "/help — show commands & disclaimer\n" "/whoami — your chat ID\n" "/leave — stop receiving alerts\n" "────────────────\n" "⚠️ DISCLAIMER\n\n" "This bot is provided for educational and informational " "purposes only. Alerts do not constitute financial " "advice or a recommendation to buy or sell any security. " "Past backtest performance is not indicative of future " "results. Systematic trading involves substantial risk of " "loss. You are solely responsible for your own trading " "decisions. By using this bot you acknowledge and " "accept these terms. Type /help to see the full disclaimer.", chat_ids=[new_id] ) except Exception as e: send_telegram(f"❌ Add failed: {e}", chat_ids=[chat_id]) return if cmd == "/remove": if not args: send_telegram( "Usage: /remove <chat_id>", chat_ids=[chat_id] ) return rm_id = args[0].strip() if _is_admin(rm_id): send_telegram("🚫 Cannot remove the admin chat ID.", chat_ids=[chat_id]) return try: path = _get_recipients_file() data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {"recipients": []} before = len(data.get("recipients", [])) data["recipients"] = [r for r in data.get("recipients", []) if str(r.get("chat_id")) != rm_id] after = len(data["recipients"]) if before == after: send_telegram(f"ℹ️ {rm_id} is not on the list.", chat_ids=[chat_id]) return _save_recipients(data["recipients"]) send_telegram(f"✅ Removed {rm_id} from the alert list.", chat_ids=[chat_id]) # Notify the removed user send_telegram( "👋 You've been removed from the macro model alerts.\n" "Use /requestaccess if you'd like to rejoin.", chat_ids=[rm_id] ) except Exception as e: send_telegram(f"❌ Remove failed: {e}", chat_ids=[chat_id]) return # ── /announcement — broadcast a message to all recipients (admin only) ── if cmd == "/announcement": # Extract message text — everything after the command message_text = text[len(cmd):].strip() if len(message_text) < 5: send_telegram( "Usage: /announcement <your message>\n" "Example: /announcement Model paused for 30 min for maintenance\n\n" "Message must be at least 5 characters.", chat_ids=[chat_id] ) return recipients = _load_recipients() if not recipients: send_telegram("No recipients to send to.", chat_ids=[chat_id]) return # Build the broadcast message broadcast = ( f"📣 ANNOUNCEMENT FROM ADMIN\n" f"────────────────\n" f"{message_text}" ) # Tell admin we're sending send_telegram( f"📢 Broadcasting to {len(recipients)} recipient(s)...", chat_ids=[chat_id] ) # Send with result tracking success, fail = send_telegram_with_results(broadcast, recipients) # Report back to admin if fail == 0: send_telegram( f"✅ Announcement sent to all {success} recipients.", chat_ids=[chat_id] ) else: send_telegram( f"⚠️ Announcement sent to {success}/{success + fail} recipients.\n" f"{fail} delivery failure(s) — check log for details.", chat_ids=[chat_id] ) log.info(f"Announcement broadcast: {success} success, {fail} fail | text: {message_text[:80]}") return # Unknown command — only respond if it starts with / if text.startswith("/"): send_telegram(f"Unknown command: {cmd}\nType /help for options.", chat_ids=[chat_id]) # ── Per-pair state ───────────────────────────────────────────────────────── def _blank_state(): return { "position_open": False, "position_ticket": None, "position_signal": None, "entry_price": None, "entry_time": None, "position_lot": None, "last_zscore": None, } def _blank_armed(): return { "active": False, "direction": None, "target_price": None, "armed_time": None, "zscore_abs": None, } PAIR_STATE = {k: _blank_state() for k in PAIR_CONFIG} PAIR_ARMED = {k: _blank_armed() for k in PAIR_CONFIG} LAST_RATE_UPDATE = {"date": None} _fast_poll_timer = {} # {pair_key: threading.Timer} # ─── State persistence ───────────────────────────────────────────────────── def _state_path(pair): return LOG_DIR / PAIR_CONFIG[pair]["state_file"] def save_state(pair): try: data = dict(PAIR_STATE[pair]) if data.get("entry_time"): data["entry_time"] = data["entry_time"].isoformat() armed = dict(PAIR_ARMED[pair]) if armed.get("armed_time"): armed["armed_time"] = armed["armed_time"].isoformat() data["_armed"] = armed target = _state_path(pair) # Atomic write: write to .tmp then rename — bypasses OneDrive file locks tmp = target.with_suffix(".tmp") tmp.write_text(json.dumps(data, indent=2), encoding="utf-8") import os as _os _os.replace(str(tmp), str(target)) except Exception as e: log.warning(f"[{pair}] State save failed: {e}") def load_state(pair): p = _state_path(pair) if not p.exists(): return try: # Use utf-8-sig encoding to tolerate UTF-8 BOM that Windows PowerShell # Set-Content -Encoding utf8 silently adds. Manual state-file edits # written from PowerShell would otherwise fail to parse and v3 would # fall back to default state, losing track of open positions. data = json.loads(p.read_text(encoding="utf-8-sig")) if data.get("position_open"): for k in ["position_open","position_ticket","position_signal", "entry_price","position_lot","last_zscore"]: PAIR_STATE[pair][k] = data.get(k) if data.get("entry_time"): PAIR_STATE[pair]["entry_time"] = datetime.fromisoformat(data["entry_time"]) log.info(f"[{pair}] State restored — ticket={PAIR_STATE[pair]['position_ticket']}") send_telegram( f"⚠️ [{pair}] STATE RESTORED AFTER REBOOT\n" f"Position was open at restart\n" f"Ticket: {PAIR_STATE[pair]['position_ticket']}\n" f"Entry: {PAIR_STATE[pair]['entry_price']}\n" f"Resuming monitoring..." ) armed_data = data.get("_armed", {}) if armed_data.get("active"): at_str = armed_data.get("armed_time") if at_str: armed_time = datetime.fromisoformat(at_str) hours_since = (datetime.utcnow() - armed_time).total_seconds() / 3600 if hours_since < 6.0: PAIR_ARMED[pair]["active"] = True PAIR_ARMED[pair]["direction"] = armed_data.get("direction") PAIR_ARMED[pair]["target_price"] = armed_data.get("target_price") PAIR_ARMED[pair]["armed_time"] = armed_time PAIR_ARMED[pair]["zscore_abs"] = armed_data.get("zscore_abs") remaining = round(6.0 - hours_since, 1) log.info(f"[{pair}] Armed signal restored — {remaining:.1f}h remaining") except Exception as e: log.warning(f"[{pair}] State restore failed: {e}") # ─── Rate data update ────────────────────────────────────────────────────── def update_rate_data(): today = date.today() if LAST_RATE_UPDATE["date"] == today: log.info("Rate data already updated today — skipping") return True try: # Single call — FinanceFlow primary, FRED/Bundesbank/MoF/ECB fallbacks from ingestion.auto_rates_loader import update_all_rates log.info("Updating rate data (FinanceFlow primary)...") ok = update_all_rates() LAST_RATE_UPDATE["date"] = today log.info("Rate data updated successfully (US + DE + JP)") return ok except KeyboardInterrupt: raise except Exception as e: log.error(f"Rate update failed: {e}") # Non-fatal — model uses last known rates (changes slowly) return False def _update_jp2y_basic(): """ Basic JP2Y fallback — used only if fetch_jp2y_robust.py is not deployed. Prefer fetch_jp2y_robust.update_jp2y_if_stale() which has all EU bugs fixed. """ import ssl jp2y_path = RATES_DIR / "jp2y.csv" try: url = "https://www.mof.go.jp/english/policy/jgbs/reference/interest_rate/historical/jgbcme_all.csv" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=30, context=ctx) as resp: raw = resp.read().decode("utf-8", errors="replace") lines = [l for l in raw.splitlines() if l.strip() and not l.startswith("#")] rows = [] for line in lines[1:]: parts = [p.strip() for p in line.split(",")] if len(parts) < 3: continue try: dt = pd.to_datetime(parts[0], format="%Y/%m/%d") val = float(parts[2]) rows.append({"date": dt.strftime("%Y-%m-%d"), "jp2y": val}) except (ValueError, IndexError): continue if rows: df = pd.DataFrame(rows).drop_duplicates("date").sort_values("date") df.to_csv(jp2y_path, index=False) log.info(f"JP2Y updated: {len(df)} rows → jp2y.csv") except Exception as e: log.warning(f"JP2Y update failed (using cached): {e}") # ─── MT5 helpers (parametrised by symbol) ────────────────────────────────── def _refresh_eurusd_recent_csv(n_bars=5000): """Refresh EURUSD_1H_recent.csv with fresh bars from MT5. Used by get_bars_1h for EURUSD. Failure to refresh is non-fatal — caller will still read whatever's in the CSV from the last successful refresh. Matches the format produced by src/research/export_recent_bars.py exactly. """ try: import MetaTrader5 as mt5 out_path = BASE_PATH / "data" / "raw" / "prices" / "EURUSD" / "EURUSD_1H_recent.csv" out_path.parent.mkdir(parents=True, exist_ok=True) rates = mt5.copy_rates_from_pos("EURUSD", mt5.TIMEFRAME_H1, 0, n_bars) if rates is None or len(rates) == 0: log.warning(f"[EURUSD] recent CSV refresh: MT5 returned no bars ({mt5.last_error()})") return False df = pd.DataFrame(rates) df["datetime"] = pd.to_datetime(df["time"], unit="s", utc=True).dt.tz_localize(None) df = df.rename(columns={"tick_volume": "volume"}) df = df[["datetime","open","high","low","close","volume"]].sort_values("datetime") df.to_csv(out_path, index=False) return True except Exception as e: log.warning(f"[EURUSD] recent CSV refresh failed (non-fatal): {e}") return False def _load_eurusd_canonical_bars(): """Load the canonical 1H EURUSD price history (2003 → today). Combines: EURUSD_1H_2003_2026.csv (full historical archive — 'Time (EET)' column, 'YYYY.MM.DD HH:MM:SS' format, 142k+ bars) EURUSD_1H_recent.csv (rolling window kept fresh by MT5 — 'datetime' column, ISO format, ~5k bars) Both files are treated as EET (validated convention — matches the trading session filter and the original backtest). No timezone conversion. Returns None if neither file is readable — caller should fall back to MT5. """ try: price_dir = BASE_PATH / "data" / "raw" / "prices" / "EURUSD" archive = price_dir / "EURUSD_1H_2003_2026.csv" recent = price_dir / "EURUSD_1H_recent.csv" # Column-name candidates we'll accept for the timestamp column. # Includes the archive's 'Time (EET)' format and standard names. TIMESTAMP_COL_CANDIDATES = [ "datetime", "date", "time", "timestamp", "time (eet)", "time(eet)", "time_eet", "time eet", ] frames = [] for path in [archive, recent]: if not path.exists(): continue try: df = pd.read_csv(path) # Strip whitespace from headers (archive has 'Volume ' with trailing space) df.columns = [c.strip().lower() for c in df.columns] # Find a timestamp column dt_col = None for cand in TIMESTAMP_COL_CANDIDATES: if cand in df.columns: dt_col = cand break if dt_col is None: log.warning(f"[EURUSD] {path.name}: no timestamp column found " f"(have: {list(df.columns)})") continue # Parse timestamps — handle both 'YYYY.MM.DD HH:MM:SS' (archive) # and 'YYYY-MM-DD HH:MM:SS' (recent) formats automatically df["datetime"] = pd.to_datetime(df[dt_col], format="mixed", errors="coerce") df["close"] = pd.to_numeric(df["close"], errors="coerce") df = df.dropna(subset=["datetime","close"]) # Keep just the columns we need downstream cols = ["datetime", "close"] for opt in ["open", "high", "low", "volume"]: if opt in df.columns: df[opt] = pd.to_numeric(df[opt], errors="coerce") cols.append(opt) frames.append(df[cols]) except Exception as e: log.warning(f"[EURUSD] could not read {path.name}: {e}") continue if not frames: return None combined = (pd.concat(frames, ignore_index=True) .drop_duplicates(subset=["datetime"], keep="last") .sort_values("datetime") .reset_index(drop=True)) return combined except Exception as e: log.error(f"[EURUSD] _load_eurusd_canonical_bars failed: {e}") return None def get_bars_1h(symbol, n_bars=500): """Fetch 1H bars for a symbol. For EURUSD: reads canonical CSV combination (gives ~143k bars, 5.7k daily, enough to run the rolling OLS beta model the backtest validates). Refreshes the recent CSV from MT5 inline before reading. Falls back to MT5-only path on any failure. For other symbols: pulls n_bars hourly from MT5 (original behaviour). """ # ── EURUSD: canonical CSV path (full history → real OLS model) ───────── if symbol == "EURUSD": try: refreshed = _refresh_eurusd_recent_csv(n_bars=5000) df = _load_eurusd_canonical_bars() if df is not None and len(df) > 1000: log.info(f"[EURUSD] 1H bars: canonical CSV ({len(df):,} bars, " f"refresh: {'ok' if refreshed else 'stale'})") return df log.warning(f"[EURUSD] canonical CSV unavailable — falling back to MT5-only") except Exception as e: log.warning(f"[EURUSD] canonical-CSV path failed (falling back to MT5): {e}") # Fall through to MT5-only path on any failure # ── Standard MT5 path (USDJPY always; EURUSD only if canonical failed) ─ try: import MetaTrader5 as mt5 tf = mt5.TIMEFRAME_H1 rates = mt5.copy_rates_from_pos(symbol, tf, 0, n_bars) if rates is None or len(rates) == 0: log.error(f"[{symbol}] 1H bars failed: {mt5.last_error()}") return None df = pd.DataFrame(rates) df["datetime"] = pd.to_datetime(df["time"], unit="s") return df[["datetime","open","high","low","close","tick_volume"]].rename( columns={"tick_volume": "volume"}).sort_values("datetime").reset_index(drop=True) except Exception as e: log.error(f"[{symbol}] get_bars_1h error: {e}") return None def get_bars_15m(symbol, n_bars=200): try: import MetaTrader5 as mt5 tf = mt5.TIMEFRAME_M15 rates = mt5.copy_rates_from_pos(symbol, tf, 0, n_bars) if rates is None or len(rates) == 0: log.error(f"[{symbol}] 15M bars failed: {mt5.last_error()}") return None df = pd.DataFrame(rates) df["datetime"] = pd.to_datetime(df["time"], unit="s") return df[["datetime","open","high","low","close"]].sort_values( "datetime").reset_index(drop=True) except Exception as e: log.error(f"[{symbol}] get_bars_15m error: {e}") return None def get_live_price(symbol): try: import MetaTrader5 as mt5 tick = mt5.symbol_info_tick(symbol) return (tick.bid, tick.ask) if tick else None except Exception as e: log.error(f"[{symbol}] get_live_price error: {e}") return None # ─── Z-score computation ─────────────────────────────────────────────────── def compute_zscore(pair, prices_df): """Dispatch to correct z-score engine per pair. `prices_df` is 15M bars for EURUSD (beta_model_15m), 1H bars for USDJPY (spread_zscore). """ cfg = PAIR_CONFIG[pair] if cfg["signal_type"] == "beta_model_15m": return _compute_eurusd_zscore_15m(prices_df, cfg) elif cfg["signal_type"] == "beta_model": # Legacy 1H path — kept for backwards compatibility; not used in v3 return _compute_eurusd_zscore(prices_df, cfg) else: return _compute_spread_zscore(prices_df, cfg) def _compute_eurusd_zscore_15m(prices_15m, cfg): """EURUSD VALIDATED FORMULA (v3 — Formula A): Z-score computed on 15M bars matching backtest exactly: - lag_gap = predicted_return_1d (daily, OLS) - eurusd_return_24bar (15M pct_change) - eurusd_return_24bar = 24 × 15M bars = 6-HOUR rolling return - z-score normalisation: rolling(60) on 15M = 15-HOUR window - Signal evaluation: every 15M bar The daily predicted_return_1d series comes from the same rolling-OLS beta model as v2 (120-day window, 20-day EWM smooth, ±clip, shift 1). It's forward-filled onto every 15M bar within its day. """ try: import statsmodels.api as sm prices_15m = prices_15m.copy() prices_15m["date"] = prices_15m["datetime"].dt.normalize() # Build daily close series from 15M bars daily_px = (prices_15m.groupby("date", as_index=False) .agg(close=("close", "last")) .sort_values("date").reset_index(drop=True)) daily_px["eurusd_return_1d"] = daily_px["close"].pct_change() daily_px["date"] = pd.to_datetime(daily_px["date"]) def load_rate(fname, col): path = RATES_DIR / fname df = pd.read_csv(path, parse_dates=["date"]) df.columns = [c.lower() for c in df.columns] df[col] = pd.to_numeric(df[col], errors="coerce") return df[["date", col]].dropna().sort_values("date") us2y = load_rate("us2y.csv", "us2y") us10y = load_rate("us10y.csv", "us10y") de2y = load_rate("de2y.csv", "de2y") de10y = load_rate("de10y.csv", "de10y") date_range = pd.DataFrame({"date": pd.date_range( daily_px["date"].min(), daily_px["date"].max(), freq="D")}) def ffill(rate_df, col): left = date_range.copy().astype({"date": "datetime64[us]"}) right = rate_df.copy().astype({"date": "datetime64[us]"}) return pd.merge_asof(left, right, on="date", direction="backward") spreads = ffill(us2y,"us2y").merge(ffill(us10y,"us10y"),on="date") \ .merge(ffill(de2y,"de2y"),on="date").merge(ffill(de10y,"de10y"),on="date") spreads["spread_2y"] = spreads["us2y"] - spreads["de2y"] spreads["spread_10y"] = spreads["us10y"] - spreads["de10y"] spreads["spread_2y_change_1d"] = spreads["spread_2y"].diff(1) spreads["spread_10y_change_1d"] = spreads["spread_10y"].diff(1) spreads = spreads[["date","spread_2y_change_1d","spread_10y_change_1d"]].dropna() model_df = daily_px.merge(spreads, on="date", how="left").dropna( subset=["eurusd_return_1d","spread_2y_change_1d"]) n_daily = len(model_df) if n_daily >= BETA_WINDOW + 5: log.info(f"[EURUSD/15m] z-score path: OLS (full model) | daily bars: {n_daily}") n = n_daily b2y_raw = np.full(n, np.nan); b10y_raw = np.full(n, np.nan) for i in range(BETA_WINDOW, n): sample = model_df.iloc[i-BETA_WINDOW:i].copy() sample = sample[(sample["spread_2y_change_1d"].abs() > 0) | (sample["spread_10y_change_1d"].abs() > 0)] if len(sample) < 30: continue X = sm.add_constant(sample[["spread_2y_change_1d","spread_10y_change_1d"]]) try: res = sm.OLS(sample["eurusd_return_1d"], X).fit() b2y_raw[i] = res.params.get("spread_2y_change_1d", np.nan) b10y_raw[i] = res.params.get("spread_10y_change_1d", np.nan) except Exception: pass model_df["beta_2y"] = pd.Series(b2y_raw).ewm(span=SMOOTH_SPAN).mean().shift(1).clip(-0.10,0.10).values model_df["beta_10y"] = pd.Series(b10y_raw).ewm(span=SMOOTH_SPAN).mean().shift(1).clip(-0.08,0.08).values model_df["predicted_return_1d"] = ( model_df["beta_2y"] * model_df["spread_2y_change_1d"] + model_df["beta_10y"] * model_df["spread_10y_change_1d"]) else: log.warning(f"[EURUSD/15m] z-score path: FALLBACK (hardcoded -0.03) | daily bars: {n_daily} (need {BETA_WINDOW + 5}+)") model_df["predicted_return_1d"] = model_df["spread_2y_change_1d"] * -0.03 model_df = model_df[["date","predicted_return_1d","spread_2y_change_1d","spread_10y_change_1d"]] # Forward-fill daily predicted return onto every 15M bar within its day df = prices_15m.merge(model_df, on="date", how="left") df[["predicted_return_1d","spread_2y_change_1d","spread_10y_change_1d"]] = \ df[["predicted_return_1d","spread_2y_change_1d","spread_10y_change_1d"]].ffill() # CRITICAL FORMULA A PARAMETERS: # pct_change(24) on 15M = 6-hour rolling return # rolling(60) on 15M = 15-hour normalisation window lookback = cfg.get("return_lookback", 24) df["eurusd_return_24bar"] = df["close"].pct_change(lookback) df["lag_gap_24bar"] = df["predicted_return_1d"] - df["eurusd_return_24bar"] w = cfg["z_window"] mean = df["lag_gap_24bar"].rolling(w, min_periods=w).mean() std = df["lag_gap_24bar"].rolling(w, min_periods=w).std() df["zscore"] = (df["lag_gap_24bar"] - mean) / std return df.dropna(subset=["zscore"]).reset_index(drop=True) except Exception as e: log.error(f"[EURUSD/15m] zscore error: {e}") import traceback; traceback.print_exc() return None def _compute_eurusd_zscore(prices_1h, cfg): """EURUSD: rolling-beta lag z-score (original validated model).""" try: import statsmodels.api as sm prices_1h = prices_1h.copy() prices_1h["date"] = prices_1h["datetime"].dt.normalize() daily_px = (prices_1h.groupby("date", as_index=False) .agg(close=("close","last")) .sort_values("date").reset_index(drop=True)) daily_px["eurusd_return_1d"] = daily_px["close"].pct_change() daily_px["date"] = pd.to_datetime(daily_px["date"]) def load_rate(fname, col): path = RATES_DIR / fname df = pd.read_csv(path, parse_dates=["date"]) df.columns = [c.lower() for c in df.columns] df[col] = pd.to_numeric(df[col], errors="coerce") return df[["date", col]].dropna().sort_values("date") us2y = load_rate("us2y.csv", "us2y") us10y = load_rate("us10y.csv", "us10y") de2y = load_rate("de2y.csv", "de2y") de10y = load_rate("de10y.csv", "de10y") date_range = pd.DataFrame({"date": pd.date_range( daily_px["date"].min(), daily_px["date"].max(), freq="D")}) def ffill(rate_df, col): left = date_range.copy().astype({"date": "datetime64[us]"}) right = rate_df.copy().astype({"date": "datetime64[us]"}) return pd.merge_asof(left, right, on="date", direction="backward") spreads = ffill(us2y,"us2y").merge(ffill(us10y,"us10y"),on="date") \ .merge(ffill(de2y,"de2y"),on="date").merge(ffill(de10y,"de10y"),on="date") spreads["spread_2y"] = spreads["us2y"] - spreads["de2y"] spreads["spread_10y"] = spreads["us10y"] - spreads["de10y"] spreads["spread_2y_change_1d"] = spreads["spread_2y"].diff(1) spreads["spread_10y_change_1d"] = spreads["spread_10y"].diff(1) spreads = spreads[["date","spread_2y_change_1d","spread_10y_change_1d"]].dropna() model_df = daily_px.merge(spreads, on="date", how="left").dropna( subset=["eurusd_return_1d","spread_2y_change_1d"]) n_daily = len(model_df) if n_daily >= BETA_WINDOW + 5: log.info(f"[EURUSD] z-score path: OLS (full model) | daily bars: {n_daily}") n = n_daily b2y_raw = np.full(n, np.nan); b10y_raw = np.full(n, np.nan) for i in range(BETA_WINDOW, n): sample = model_df.iloc[i-BETA_WINDOW:i].copy() sample = sample[(sample["spread_2y_change_1d"].abs() > 0) | (sample["spread_10y_change_1d"].abs() > 0)] if len(sample) < 30: continue X = sm.add_constant(sample[["spread_2y_change_1d","spread_10y_change_1d"]]) try: res = sm.OLS(sample["eurusd_return_1d"], X).fit() b2y_raw[i] = res.params.get("spread_2y_change_1d", np.nan) b10y_raw[i] = res.params.get("spread_10y_change_1d", np.nan) except Exception: pass model_df["beta_2y"] = pd.Series(b2y_raw).ewm(span=SMOOTH_SPAN).mean().shift(1).clip(-0.10,0.10).values model_df["beta_10y"] = pd.Series(b10y_raw).ewm(span=SMOOTH_SPAN).mean().shift(1).clip(-0.08,0.08).values model_df["predicted_return_1d"] = ( model_df["beta_2y"] * model_df["spread_2y_change_1d"] + model_df["beta_10y"] * model_df["spread_10y_change_1d"]) else: log.warning(f"[EURUSD] z-score path: FALLBACK (hardcoded -0.03) | daily bars: {n_daily} (need {BETA_WINDOW + 5}+)") model_df["predicted_return_1d"] = model_df["spread_2y_change_1d"] * -0.03 model_df = model_df[["date","predicted_return_1d","spread_2y_change_1d","spread_10y_change_1d"]] df = prices_1h.merge(model_df, on="date", how="left") df[["predicted_return_1d","spread_2y_change_1d","spread_10y_change_1d"]] = \ df[["predicted_return_1d","spread_2y_change_1d","spread_10y_change_1d"]].ffill() df["eurusd_return_24h"] = df["close"].pct_change(24) df["lag_gap_24h"] = df["predicted_return_1d"] - df["eurusd_return_24h"] w = cfg["z_window"] mean = df["lag_gap_24h"].rolling(w, min_periods=w).mean() std = df["lag_gap_24h"].rolling(w, min_periods=w).std() df["zscore"] = (df["lag_gap_24h"] - mean) / std return df.dropna(subset=["zscore"]).reset_index(drop=True) except Exception as e: log.error(f"[EURUSD] zscore error: {e}") import traceback; traceback.print_exc() return None def _compute_spread_zscore(prices_1h, cfg): """ USDJPY: z-score of US-JP 2Y spread level. CRITICAL: z_window=30 means 30 CALENDAR DAYS on daily spread data — this matches the backtest exactly. Rolling on hourly bars would give a 30-hour window which produces a completely wrong signal. Method: 1. Build daily spread series (us2y - jp2y) 2. Compute 30-day rolling z-score on DAILY data 3. Forward-fill daily z-score onto hourly bars """ try: spread_col1, spread_col2 = cfg["spread_cols"] w = cfg["z_window"] # 30 = 30 DAYS on daily data def load_rate(fname, col): path = RATES_DIR / fname df = pd.read_csv(path, parse_dates=["date"]) df.columns = [c.lower() for c in df.columns] df[col] = pd.to_numeric(df[col], errors="coerce") return df[["date", col]].dropna().sort_values("date") r1 = load_rate(f"{spread_col1}.csv", spread_col1) r2 = load_rate(f"{spread_col2}.csv", spread_col2) rates = pd.merge(r1, r2, on="date", how="outer").sort_values("date") rates = rates.set_index("date") rates = rates.reindex( pd.date_range(rates.index.min(), rates.index.max(), freq="D") ).ffill().dropna() rates["spread"] = rates[spread_col1] - rates[spread_col2] # ── Step 1: Compute z-score on DAILY spread data ────────────────── rates["z_mean"] = rates["spread"].rolling(w, min_periods=w).mean() rates["z_std"] = rates["spread"].rolling(w, min_periods=w).std() rates["zscore"] = (rates["spread"] - rates["z_mean"]) / rates["z_std"] rates = rates.dropna(subset=["zscore"]) # ── Step 2: Forward-fill daily z-score onto hourly bars ─────────── # datetime64[us] on both sides — required by pandas merge_asof on Python 3.14 daily_z = rates["zscore"].reset_index() daily_z.columns = ["date", "zscore"] daily_z["date"] = pd.to_datetime(daily_z["date"]).astype("datetime64[us]") prices_1h = prices_1h.copy() prices_1h["datetime"] = pd.to_datetime( prices_1h["datetime"]).astype("datetime64[us]") df = pd.merge_asof( prices_1h.sort_values("datetime"), daily_z.sort_values("date"), left_on="datetime", right_on="date", direction="backward", ) current_z = float(df["zscore"].iloc[-1]) if not df.empty else 0.0 log.info(f"[USDJPY] Spread z-score computed on {len(rates)} daily bars | " f"current z={current_z:.4f} | " f"spread={rates['spread'].iloc[-1]:.4f}% " f"(mean={rates['z_mean'].iloc[-1]:.4f} " f"std={rates['z_std'].iloc[-1]:.4f})") return df.dropna(subset=["zscore"]).reset_index(drop=True) except Exception as e: log.error(f"[USDJPY] zscore error: {e}") import traceback; traceback.print_exc() return None # ─── Position sizing ─────────────────────────────────────────────────────── def get_multiplier(pair, zscore_abs): for lo, hi, mult in PAIR_CONFIG[pair]["sizing_bands"]: if lo <= zscore_abs < hi: return mult return PAIR_CONFIG[pair]["sizing_bands"][-1][2] def calculate_lot_size(pair, zscore_abs): try: import MetaTrader5 as mt5 cfg = PAIR_CONFIG[pair] account = mt5.account_info() if account is None: return 0.01 balance = account.balance multiplier = get_multiplier(pair, zscore_abs) risk_dollar= balance * BASE_RISK_PCT * multiplier symbol_info = mt5.symbol_info(cfg["symbol"]) stop_pct = cfg["stop_pct"] pip_size = cfg["pip_size"] if pair == "USDJPY": # USDJPY: pip = 0.01 JPY # pip_value per standard lot = 0.01 / current_price * 100000 tick = mt5.symbol_info_tick(cfg["symbol"]) price = tick.bid if tick else 150.0 pip_value = pip_size * 100_000 / price # USD per pip per standard lot stop_pips = stop_pct * price / pip_size # pips from price % else: # EURUSD: pip = 0.0001, pip_value ≈ $10/lot pip_value = symbol_info.trade_tick_value * 10 if symbol_info else 10.0 stop_pips = stop_pct / pip_size lot_size = risk_dollar / (stop_pips * pip_value) min_lot = symbol_info.volume_min if symbol_info else 0.01 max_lot = symbol_info.volume_max if symbol_info else 100.0 lot_step = symbol_info.volume_step if symbol_info else 0.01 lot_size = max(min_lot, min(max_lot, round(lot_size / lot_step) * lot_step)) log.info(f"[{pair}] Lot size: {lot_size:.2f} | Risk: ${risk_dollar:,.0f} | " f"Mult: {multiplier}x | Balance: ${balance:,.0f}") return lot_size except Exception as e: log.error(f"[{pair}] calculate_lot_size error: {e}") return 0.01 # ─── Order management ────────────────────────────────────────────────────── def place_order(pair, signal, entry_price, lot_size, zscore_abs): try: import MetaTrader5 as mt5 cfg = PAIR_CONFIG[pair] symbol = cfg["symbol"] tp_pct = cfg["tp_pct"] sl_pct = cfg["stop_pct"] dp = cfg["dp"] if signal == 1: order_type = mt5.ORDER_TYPE_BUY tp = round(entry_price * (1 + tp_pct), dp) sl = round(entry_price * (1 - sl_pct), dp) else: order_type = mt5.ORDER_TYPE_SELL tp = round(entry_price * (1 - tp_pct), dp) sl = round(entry_price * (1 + sl_pct), dp) request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": lot_size, "type": order_type, "price": entry_price, "sl": sl, "tp": tp, "magic": 20240101, "comment": f"z-score {zscore_abs:.2f}", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, "deviation": 20, } result = mt5.order_send(request) if result is None: log.error(f"[{pair}] order_send returned None") return None if result.retcode != mt5.TRADE_RETCODE_DONE: log.error(f"[{pair}] order_send failed: {result.retcode} {result.comment}") return None log.info(f"[{pair}] Order placed: ticket={result.order} " f"dir={'BUY' if signal==1 else 'SELL'} " f"lots={lot_size} entry={entry_price} tp={tp} sl={sl}") return result.order except Exception as e: log.error(f"[{pair}] place_order error: {e}") return None def close_position_market(pair, ticket, reason): try: import MetaTrader5 as mt5 symbol = PAIR_CONFIG[pair]["symbol"] positions = mt5.positions_get(ticket=ticket) if not positions: log.info(f"[{pair}] Position {ticket} not found — already closed") return True pos = positions[0] side = mt5.ORDER_TYPE_SELL if pos.type == mt5.POSITION_TYPE_BUY else mt5.ORDER_TYPE_BUY tick = mt5.symbol_info_tick(symbol) price= tick.ask if side == mt5.ORDER_TYPE_BUY else tick.bid req = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": pos.volume, "type": side, "position": ticket, "price": price, "magic": 20240101, "comment": reason, "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, "deviation": 20, } result = mt5.order_send(req) if result and result.retcode == mt5.TRADE_RETCODE_DONE: log.info(f"[{pair}] Position {ticket} closed — reason: {reason}") return True log.error(f"[{pair}] Close failed: {result.retcode if result else 'None'}") return False except Exception as e: log.error(f"[{pair}] close_position_market error: {e}") return False def check_position_status(pair): try: import MetaTrader5 as mt5 ticket = PAIR_STATE[pair]["position_ticket"] if ticket is None: return False positions = mt5.positions_get(ticket=ticket) return bool(positions) except Exception: return False def _is_market_tradeable(pair): """Return True if the broker market is open and fully tradeable. Used by hold-limit, z-exit, and broker-detected-close paths to avoid attempting to close positions when markets are closed. Saturday's bug: v3 attempted hold-limit close while UJ was closed weekend, MT5 returned error 10018 (TRADE_RETCODE_MARKET_CLOSED), but v3 didn't check and proceeded to mark the position closed in state anyway. Result: state diverged from broker truth and v3 lost track of the open position. """ try: import MetaTrader5 as mt5 symbol = PAIR_CONFIG[pair]["symbol"] si = mt5.symbol_info(symbol) if si is None: return False # trade_mode == FULL means orders can be placed return si.trade_mode == mt5.SYMBOL_TRADE_MODE_FULL except Exception: return False # ─── Fast poll (entry detection) ─────────────────────────────────────────── def arm_fast_polling(pair, direction, target_price, zscore_abs): PAIR_ARMED[pair]["active"] = True PAIR_ARMED[pair]["direction"] = direction PAIR_ARMED[pair]["target_price"] = target_price PAIR_ARMED[pair]["armed_time"] = datetime.now(timezone.utc) PAIR_ARMED[pair]["zscore_abs"] = zscore_abs save_state(pair) log.info(f"[{pair}] Armed | dir={'LONG' if direction==1 else 'SHORT'} " f"target={target_price:.{PAIR_CONFIG[pair]['dp']}f}") _schedule_fast_poll(pair) def disarm_fast_polling(pair, reason=""): PAIR_ARMED[pair]["active"] = False if pair in _fast_poll_timer and _fast_poll_timer[pair]: _fast_poll_timer[pair].cancel() _fast_poll_timer[pair] = None save_state(pair) log.info(f"[{pair}] Disarmed — {reason}") def _schedule_fast_poll(pair): # 1-second polling cadence. Was previously 5 seconds, which on raw-spread # accounts produced 3-5 pip slip vs. the Fib target during fast moves # because price could dip below target and rebound between two 5-second # checks. 1 second is still safe (get_live_price hits the local MT5 # terminal cache, not the broker API directly) but catches taps much # more reliably. timer = threading.Timer(1.0, lambda: _run_fast_poll(pair)) timer.daemon = True timer.start() _fast_poll_timer[pair] = timer def _run_fast_poll(pair): if not PAIR_ARMED[pair]["active"]: return try: armed = PAIR_ARMED[pair] state = PAIR_STATE[pair] cfg = PAIR_CONFIG[pair] now = datetime.now(timezone.utc) elapsed_hours = (now - armed["armed_time"]).total_seconds() / 3600 if elapsed_hours >= 6.0: disarm_fast_polling(pair, "6h expiry") return if state["position_open"]: return prices = get_live_price(cfg["symbol"]) if prices is None: _schedule_fast_poll(pair) return bid, ask = prices direction = armed["direction"] target = armed["target_price"] if direction == 1: entry_hit = ask <= target exec_price = ask else: entry_hit = bid >= target exec_price = bid if not entry_hit: _schedule_fast_poll(pair) return elapsed_min = elapsed_hours * 60 log.info(f"[{pair}] TARGET HIT | {'LONG' if direction==1 else 'SHORT'} | " f"price={exec_price:.{cfg['dp']}f} | " f"{elapsed_min:.0f}m since signal") zscore_abs = armed["zscore_abs"] lot_size = calculate_lot_size(pair, zscore_abs) if lot_size < 0.01: disarm_fast_polling(pair, "lot size error") return ticket = place_order(pair, direction, exec_price, lot_size, zscore_abs) if ticket: state["position_open"] = True state["position_ticket"] = ticket state["position_signal"] = direction state["entry_price"] = exec_price state["entry_time"] = now.replace(tzinfo=None) state["position_lot"] = lot_size disarm_fast_polling(pair, "order placed") dp = cfg["dp"] tp_px = exec_price*(1+cfg["tp_pct"]) if direction==1 else exec_price*(1-cfg["tp_pct"]) sl_px = exec_price*(1-cfg["stop_pct"]) if direction==1 else exec_price*(1+cfg["stop_pct"]) # Slip diagnostic: how far the actual fill was from the Fib target. # For LONG, slip = target - exec_price (positive = filled cheaper # than target). For SHORT, slip = exec_price - target (positive # = filled higher than target). Convert to pips for legibility. # EU/UJ use 0.0001 / 0.01 as pip size (one less than dp 5/3). pip_size = 10 ** -(dp - 1) raw_slip = (target - exec_price) if direction == 1 else (exec_price - target) slip_pips = raw_slip / pip_size slip_str = f"{slip_pips:+.1f}p" if slip_pips != 0 else "0.0p" log.info(f"[{pair}] Entry slip vs target: {slip_str} " f"(target={target:.{dp}f}, fill={exec_price:.{dp}f})") send_telegram( f"📈 [{pair}] ORDER PLACED\n" f"Direction: {'LONG' if direction==1 else 'SHORT'}\n" f"Entry: {exec_price:.{dp}f}\n" f"Target was: {target:.{dp}f} (slip {slip_str})\n" f"Lots: {lot_size:.2f}\n" f"TP: {tp_px:.{dp}f}\n" f"SL: {sl_px:.{dp}f}\n" f"Ticket: {ticket}" ) else: disarm_fast_polling(pair, "order failed") except Exception as e: log.error(f"[{pair}] fast poll error: {e}") _schedule_fast_poll(pair) # ─── Trade history ───────────────────────────────────────────────────────── def save_trade_history(pair, ticket, entry_price, direction, lot_size, net_pnl): try: path = LOG_DIR / PAIR_CONFIG[pair]["trade_log"] trades = json.loads(path.read_text(encoding="utf-8")) if path.exists() else [] trades.append({ "pair": pair, "ticket": ticket, "entry_price": entry_price, "direction": direction, "lot_size": lot_size, "net_pnl": round(net_pnl, 2), "closed_at": datetime.now().isoformat(), }) path.write_text(json.dumps(trades, indent=2), encoding="utf-8") total = sum(t["net_pnl"] for t in trades) log.info(f"[{pair}] Trade history: {len(trades)} trades | total=${total:.2f}") except Exception as e: log.warning(f"[{pair}] trade history save failed: {e}") # ─── Startup reconciliation ──────────────────────────────────────────────── # Scans MT5 deal history for ALL closed trades in the past 90 days and # ensures every one is reflected in trade_history_{pair}.json. Catches the # "_get_closed_pnl returned $0 or missed the deal entirely" silent data loss # that happens when the monitor is offline during TP/SL close. def reconcile_trade_history_from_mt5(lookback_days=90): """ Startup-time reconciliation. Scans MT5 deal history and back-fills any closed trades missing from the trade_history JSON files. Logic: 1. Load existing tickets from each pair's trade_history file 2. Query MT5 history_deals_get() for the last N days 3. Group deals by position_id (ticket) + symbol 4. For any ticket not already in the JSON, compute net P&L from its deals (entry + exit + commission + swap) and append to JSON 5. Report: "reconciled N missing trades" or "all X trades in sync" """ try: import MetaTrader5 as mt5 except ImportError: log.warning("Reconciliation: MT5 not available, skipping") return try: now = datetime.now(timezone.utc) start = now - timedelta(days=lookback_days) deals = mt5.history_deals_get(start, now) if deals is None: log.warning("Reconciliation: MT5 deal history returned None") return # Group deals by position_id. Only BUY/SELL deals count; balance/credit # entries (like the initial demo deposit) are excluded. DEAL_BUY = 0 # mt5.DEAL_TYPE_BUY DEAL_SELL = 1 # mt5.DEAL_TYPE_SELL by_position = {} for d in deals: if d.type not in (DEAL_BUY, DEAL_SELL): continue if not d.position_id: continue by_position.setdefault(d.position_id, []).append(d) total_reconciled = 0 total_already_synced = 0 total_healed = 0 # entries that had net_pnl=0.0 and got corrected for pair, cfg in PAIR_CONFIG.items(): symbol = cfg["symbol"] path = LOG_DIR / cfg["trade_log"] try: existing = json.loads(path.read_text(encoding="utf-8")) if path.exists() else [] except Exception: existing = [] # Index existing by ticket for both lookup AND in-place updates existing_by_ticket = {int(t.get("ticket", 0)): t for t in existing} existing_tickets = set(existing_by_ticket.keys()) to_add = [] healed_in_pair = 0 for pos_id, pos_deals in by_position.items(): pair_deals = [d for d in pos_deals if d.symbol == symbol] if not pair_deals: continue # If the date-range scan saw fewer than 2 deals (e.g. only # the entry — a known IC Markets quirk, see 2026-06-01 root # cause), re-fetch via the position-specific query which # reliably returns both entry and exit deals. This must run # BEFORE the len<2 guard, otherwise single-deal positions get # skipped and never healed. if len(pair_deals) < 2: pos_specific = mt5.history_deals_get(position=int(pos_id)) if pos_specific and len(pos_specific) > len(pair_deals): pair_deals = list(pos_specific) # Fully closed position needs entry + exit (2+ deals) pair_deals = sorted(pair_deals, key=lambda d: d.time) if len(pair_deals) < 2: continue # still open or incomplete # Compute the authoritative net P&L from MT5 deals mt5_net_pnl = sum(d.profit + d.swap + d.commission for d in pair_deals) if int(pos_id) in existing_tickets: # ── Self-healing branch ───────────────────────────────── # If the existing entry has net_pnl == 0.0 (the stale-zero # symptom from when _get_closed_pnl gave up before MT5 # populated the exit deal), AND MT5 now has a non-zero # value, update the entry in place. This catches the # entire class of pnl_confident=False trades that v3 # previously had no way to self-correct. existing_entry = existing_by_ticket[int(pos_id)] existing_pnl = float(existing_entry.get("net_pnl", 0.0)) if existing_pnl == 0.0 and abs(mt5_net_pnl) > 0.01: existing_entry["net_pnl"] = round(mt5_net_pnl, 2) # Also fix close timestamp to match MT5's real one existing_entry["closed_at"] = ( datetime.fromtimestamp(pair_deals[-1].time).isoformat() ) healed_in_pair += 1 log.info(f"[RECONCILE-HEAL] {pair} ticket {pos_id} — " f"net_pnl was $0.00, MT5 says " f"${mt5_net_pnl:+,.2f} — corrected") total_already_synced += 1 continue entry_deal = pair_deals[0] close_deal = pair_deals[-1] entry_price = entry_deal.price # Entry type BUY → LONG (+1) ; SELL → SHORT (-1) direction = 1 if entry_deal.type == DEAL_BUY else -1 lot_size = entry_deal.volume net_pnl = mt5_net_pnl closed_at = datetime.fromtimestamp(close_deal.time).isoformat() to_add.append({ "pair": pair, "ticket": int(pos_id), "entry_price": round(entry_price, cfg["dp"]), "direction": direction, "lot_size": round(lot_size, 2), "net_pnl": round(net_pnl, 2), "closed_at": closed_at, }) # Write back if anything changed (new additions OR healed entries) if to_add or healed_in_pair: combined = list(existing_by_ticket.values()) + to_add combined.sort(key=lambda t: t.get("closed_at", "")) path.write_text(json.dumps(combined, indent=2), encoding="utf-8") total_reconciled += len(to_add) total_healed += healed_in_pair for t in to_add: sign = "+" if t["net_pnl"] >= 0 else "" log.info(f"[RECONCILE] {pair} ticket {t['ticket']} — " f"{'LONG' if t['direction']==1 else 'SHORT'} " f"@ {t['entry_price']} · {sign}${t['net_pnl']:,.2f} · " f"closed {t['closed_at'][:10]}") if total_reconciled > 0 or total_healed > 0: log.info(f"[RECONCILE] Back-filled {total_reconciled} missing trade(s), " f"healed {total_healed} stale-zero entries " f"({total_already_synced} already in sync)") try: healed_line = (f"Healed: {total_healed} stale-zero entries\n" if total_healed > 0 else "") send_telegram( f"🔄 TRADE HISTORY RECONCILED\n" f"────────────────\n" f"Back-filled {total_reconciled} missing trade(s)\n" f"{healed_line}" f"from MT5 deal history on startup.\n" f"────────────────\n" f"Already in sync: {total_already_synced}\n" f"Lookback: {lookback_days} days" ) except Exception: pass else: log.info(f"[RECONCILE] Trade history in sync " f"({total_already_synced} trades verified)") except Exception as e: log.warning(f"Reconciliation failed: {e}") import traceback; traceback.print_exc() # ─── In-session check ────────────────────────────────────────────────────── def _eet_now(): """Return current datetime in Eastern European Time (EET in winter, EEST in summer). Uses the Europe/Helsinki zone — both EE timezone variants follow DST consistently with that anchor. Previous implementation used a naive `(UTC.hour + 2) % 24` which was correct in winter but 1 hour off in summer (DST), causing v3 to skip legitimate signals during the first hour of EEST sessions. Caught 2026-05-18 when EU broke threshold at z=-3.12 inside the validated 09:00-18:00 EET window but v3 logged "outside session — skipping". """ try: from zoneinfo import ZoneInfo return datetime.now(ZoneInfo("Europe/Helsinki")) except Exception: # Fallback if zoneinfo unavailable: assume EEST (UTC+3) May–Oct # crudely by month. Better than the previous fixed +2 offset. utc = datetime.now(timezone.utc) offset = 3 if 3 <= utc.month <= 10 else 2 return utc + timedelta(hours=offset) def in_session(pair): cfg = PAIR_CONFIG[pair] hour = _eet_now().hour return cfg["session_start"] <= hour <= cfg["session_end"] # ─── Main signal check ───────────────────────────────────────────────────── def _get_closed_pnl(pair, ticket): """ Pull closed trade P&L from MT5 deal history. Returns (net_pnl, close_price, reason, pnl_confident). pnl_confident is True when we found the exit deal cleanly and net_pnl reflects real broker numbers. False when retries were exhausted and net_pnl may be approximate or zero-by-default. PATCHED 2026-05-06: Retries with backoff to handle MT5 history table population delay. The MT5 history_deals_get() result updates asynchronously after a position closes — typical delay 100-2000ms. Without retry, we sometimes find only the entry deal (profit=0) or partial data and report wrong P&L on closes. PATCHED 2026-05-14: Added confidence flag so the Telegram close message can avoid the misleading "BREAK EVEN" label when P&L is really just "unknown — exit deal wasn't in MT5 history table yet". Strategy: • Filter strictly by position_id (no fallback to symbol-wide search, which previously caused cross-ticket P&L contamination) • Look specifically for the EXIT deal (entry == DEAL_ENTRY_OUT) • Retry up to 6 times with 1-second waits (max 6s wait total) • Sum profit/swap/commission across all deals for this ticket only """ try: import MetaTrader5 as mt5 import time as _time from datetime import timezone DEAL_ENTRY_OUT = 1 # MT5 constant: this deal closed a position now = datetime.now(timezone.utc) start = now - timedelta(days=7) max_attempts = 30 wait_seconds = 2.0 # Total wait time: 60s. IC Markets demo has been observed taking # 6-15s to populate exit deals; longer waits tolerate worse broker # latency. Cost is only paid on slow closes — fast closes still # break out of the loop on attempt 1. Self-healing reconcile catches # any cases where even 60s isn't enough. pos_deals = [] exit_deal = None # Retry loop: wait for exit deal to appear in MT5 history for attempt in range(1, max_attempts + 1): # ROOT-CAUSE FIX (2026-06-01): query by POSITION first. The # date-range query is unreliable on IC Markets demo and omits # recent exit deals; the position query returns them reliably. deals = mt5.history_deals_get(position=ticket) if not deals: deals = mt5.history_deals_get(start, now) if deals is None: log.warning(f"[{pair}] history_deals_get returned None on attempt {attempt}") _time.sleep(wait_seconds) continue # Filter strictly by ticket — never aggregate across positions pos_deals = [d for d in deals if d.position_id == ticket] # Look specifically for the exit deal (the one carrying real P&L) exit_deals = [d for d in pos_deals if getattr(d, 'entry', None) == DEAL_ENTRY_OUT] if exit_deals: exit_deal = exit_deals[0] if attempt > 1: log.info(f"[{pair}] Exit deal found on attempt {attempt}/{max_attempts}") break log.info(f"[{pair}] Exit deal not yet in history " f"(attempt {attempt}/{max_attempts}, " f"found {len(pos_deals)} deals for ticket) — waiting {wait_seconds}s") _time.sleep(wait_seconds) # If we never found any deals, fall back gracefully if not pos_deals: log.warning(f"[{pair}] No deals found for ticket {ticket} after {max_attempts} attempts") return 0.0, None, "TP/SL", False if exit_deal is None: log.warning(f"[{pair}] Exit deal never appeared after " f"{max_attempts}s — using all matched deals as-is " f"(P&L may be 0 if exit deal still pending)") net_pnl = sum(d.profit + d.swap + d.commission for d in pos_deals) close_deal = max(pos_deals, key=lambda d: d.time) pnl_confident = False else: # Sum across all deals for this ticket only (entry profit=0, # exit carries P&L; both contribute swap/commission) net_pnl = sum(d.profit + d.swap + d.commission for d in pos_deals) close_deal = exit_deal pnl_confident = True close_px = close_deal.price # Determine close reason from deal comment comment = close_deal.comment.lower() if close_deal.comment else "" if "tp" in comment or "take profit" in comment: reason = "Take Profit ✅" elif "sl" in comment or "stop loss" in comment: reason = "Stop Loss ❌" else: reason = "TP/SL" log.info(f"[{pair}] Close P&L from MT5: ${net_pnl:.2f} | " f"close_px={close_px} | reason={reason} | " f"confident={pnl_confident}") return round(net_pnl, 2), close_px, reason, pnl_confident except Exception as e: log.warning(f"[{pair}] _get_closed_pnl error: {e}") return 0.0, None, "TP/SL", False def run_signal_check_pair(pair): """Signal check for one pair. Cadence is per-pair (EU=15m, UJ=1h).""" cfg = PAIR_CONFIG[pair] state = PAIR_STATE[pair] armed = PAIR_ARMED[pair] symbol= cfg["symbol"] dp = cfg["dp"] log.info(f"[{pair}] ── Signal check ── (cadence: {cfg.get('signal_timeframe', '1h')})") # Get bars based on signal_timeframe — EU on 15M, UJ on 1H timeframe = cfg.get("signal_timeframe", "1h") if timeframe == "15m": prices_df = get_bars_15m(symbol, cfg.get("bars_15m", 600)) if prices_df is None: log.warning(f"[{pair}] No 15M bars — skipping") return else: prices_df = get_bars_1h(symbol, cfg.get("bars_1h", 500)) if prices_df is None: log.warning(f"[{pair}] No 1H bars — skipping") return # Compute z-score df = compute_zscore(pair, prices_df) if df is None or len(df) < 5: log.warning(f"[{pair}] Z-score computation failed") return current_z = float(df["zscore"].iloc[-1]) state["last_zscore"] = current_z log.info(f"[{pair}] Z-score: {current_z:.4f} (threshold ±{cfg['threshold']})") # ── Manage open position ───────────────────────────────────────────── if state["position_open"] and state["position_ticket"] is not None: still_open = check_position_status(pair) if not still_open: # Closed by broker (TP or SL hit) — pull P&L from MT5 deal history ticket = state["position_ticket"] entry_px = state["entry_price"] entry_dir = state["position_signal"] entry_lots = state.get("position_lot", 0) log.info(f"[{pair}] Position closed by broker (TP/SL) ticket={ticket}") net_pnl, close_px, close_reason, pnl_confident = _get_closed_pnl(pair, ticket) state["position_open"] = False state["position_ticket"] = None state["position_signal"] = None state["entry_price"] = None state["entry_time"] = None save_state(pair) if entry_px and entry_dir is not None: save_trade_history(pair, ticket, entry_px, entry_dir, entry_lots, net_pnl) dp = PAIR_CONFIG[pair]["dp"] pnl_str = f"+${net_pnl:,.2f}" if net_pnl >= 0 else f"-${abs(net_pnl):,.2f}" # Result classification: # The model's exit paths are TP / SL / Z-exit (EU only) and # time-based. NONE of these produce literal $0.00 P&L by # design — any "zero" reading is almost certainly stale MT5 # history data, not a true break-even outcome. So we never # label a result BREAK EVEN. If P&L is zero we treat it as # unknown and direct the user to check MT5 history. if net_pnl > 0: result = "✅ PROFIT" elif net_pnl < 0: result = "❌ LOSS" else: # Zero P&L is suspicious — either pnl_confident is False # (MT5 history wasn't ready yet) or something genuinely # rare happened. Either way, surface it as UNKNOWN. result = "⚠️ UNKNOWN — check MT5 history" pnl_str = "Pending — see broker" dir_str = "LONG" if entry_dir == 1 else "SHORT" close_str = f"{close_px:.{dp}f}" if close_px else "—" send_telegram( f"🏁 [{pair}] POSITION CLOSED\n" f"────────────────\n" f"Result: {result}\n" f"Direction: {dir_str}\n" f"Entry: {entry_px:.{dp}f}\n" f"Close: {close_str}\n" f"Reason: {close_reason}\n" f"Lots: {entry_lots:.2f}\n" f"────────────────\n" f"Net P&L: {pnl_str}" ) return # Check hold time limit if state["entry_time"] is not None: elapsed = (datetime.now() - state["entry_time"]).total_seconds() / 3600 if elapsed >= cfg["hold_hours"]: # Market-open check FIRST. If market is closed, just log and # skip — we'll retry on the next signal check. Previously v3 # attempted close anyway, got MT5 error 10018, and then ate # its own state. This fix prevents that whole failure cascade. if not _is_market_tradeable(pair): log.info(f"[{pair}] Hold limit reached ({elapsed:.1f}h) " f"but market closed — will retry next check") # Save state so the watchdog doesn't fire "state file stale" # while we're correctly waiting for the market to reopen. # last_zscore was updated earlier in this function; this # persists it to disk. save_state(pair) return log.info(f"[{pair}] Hold limit reached ({elapsed:.1f}h) — closing") ticket_h = state["position_ticket"] entry_px_h = state["entry_price"] entry_dir_h = state["position_signal"] entry_lots_h= state.get("position_lot", 0) close_ok = close_position_market(pair, ticket_h, "time_limit") # Verify broker actually has no position before mutating state. # close_position_market can return False on rejection, and even # if it returned True we double-check via positions_get to # avoid race conditions. still_open = check_position_status(pair) if (not close_ok) or still_open: log.warning(f"[{pair}] Hold-limit close did NOT succeed " f"(close_ok={close_ok}, still_open={still_open}) " f"— leaving state intact, will retry") # Save state so watchdog doesn't fire stale alerts while # we're correctly waiting for retry conditions. Belt and # braces alongside the market-tradeable early return — # some brokers (IC Markets demo) leave symbol_info().trade_mode # as FULL on weekends even though orders return MARKET_CLOSED. save_state(pair) return net_pnl_h, close_px_h, _, pnl_confident_h = _get_closed_pnl(pair, ticket_h) state["position_open"] = False state["position_ticket"] = None state["position_signal"] = None state["entry_price"] = None state["entry_time"] = None save_state(pair) if entry_px_h and entry_dir_h is not None: save_trade_history(pair, ticket_h, entry_px_h, entry_dir_h, entry_lots_h, net_pnl_h) dp_h = PAIR_CONFIG[pair]["dp"] # P&L classification: PROFIT / LOSS / UNKNOWN. No BREAK EVEN — # the model doesn't produce literal $0.00 by design; any zero # reading almost certainly means MT5 history wasn't ready yet. if net_pnl_h > 0: pnl_str_h = f"+${net_pnl_h:,.2f}" result_h = "✅ PROFIT" elif net_pnl_h < 0: pnl_str_h = f"-${abs(net_pnl_h):,.2f}" result_h = "❌ LOSS" else: pnl_str_h = "Pending — see broker" result_h = "⚠️ UNKNOWN — check MT5 history" send_telegram( f"⏱️ [{pair}] POSITION CLOSED — hold limit {cfg['hold_hours']}h\n" f"Result: {result_h}\n" f"Net P&L: {pnl_str_h}" ) return # Check z-exit (EURUSD only — USDJPY has z_exit=None) if cfg["z_exit"] is not None: dir_open = state["position_signal"] if (dir_open == 1 and current_z <= -cfg["z_exit"]) or \ (dir_open == -1 and current_z >= cfg["z_exit"]): if not _is_market_tradeable(pair): log.info(f"[{pair}] Z-exit triggered at z={current_z:.3f} " f"but market closed — will retry next check") save_state(pair) return log.info(f"[{pair}] Z-exit triggered at z={current_z:.3f}") ticket_z = state["position_ticket"] entry_px_z = state["entry_price"] entry_dir_z = state["position_signal"] entry_lots_z= state.get("position_lot", 0) close_ok_z = close_position_market(pair, ticket_z, "zscore_reversal") still_open_z = check_position_status(pair) if (not close_ok_z) or still_open_z: log.warning(f"[{pair}] Z-exit close did NOT succeed " f"(close_ok={close_ok_z}, still_open={still_open_z}) " f"— leaving state intact, will retry") save_state(pair) return net_pnl_z, close_px_z, _, pnl_confident_z = _get_closed_pnl(pair, ticket_z) state["position_open"] = False state["position_ticket"] = None state["position_signal"] = None state["entry_price"] = None state["entry_time"] = None save_state(pair) if entry_px_z and entry_dir_z is not None: save_trade_history(pair, ticket_z, entry_px_z, entry_dir_z, entry_lots_z, net_pnl_z) # P&L classification — same PROFIT/LOSS/UNKNOWN scheme if net_pnl_z > 0: pnl_str_z = f"+${net_pnl_z:,.2f}" result_z = "✅ PROFIT" elif net_pnl_z < 0: pnl_str_z = f"-${abs(net_pnl_z):,.2f}" result_z = "❌ LOSS" else: pnl_str_z = "Pending — see broker" result_z = "⚠️ UNKNOWN — check MT5 history" send_telegram( f"🔄 [{pair}] Z-EXIT TRIGGERED\n" f"Z-score: {current_z:.3f}\n" f"Result: {result_z}\n" f"Net P&L: {pnl_str_z}") return log.info(f"[{pair}] Position open — holding | " f"ticket={state['position_ticket']} | " f"entry={state['entry_price']}") # Persist the fresh last_zscore even when holding — otherwise the # state file goes stale through long holds and the dashboard # displays a frozen z-score reading. save_state(pair) return # ── Check for new signal ───────────────────────────────────────────── if abs(current_z) < cfg["threshold"]: if not in_session(pair): cfg2 = PAIR_CONFIG[pair] hour_eet = _eet_now().hour log.info(f"[{pair}] Outside trading window ({hour_eet:02d}:00 EET | " f"session {cfg2['session_start']:02d}:00–{cfg2['session_end']:02d}:00 EET) " f"| z={current_z:.3f}") else: log.info(f"[{pair}] No signal (|z|={abs(current_z):.3f} < {cfg['threshold']})") save_state(pair) return if not in_session(pair): log.info(f"[{pair}] Signal |z|={abs(current_z):.3f} outside session — skipping") save_state(pair) # still persist the updated z-score return if armed["active"]: log.info(f"[{pair}] Already armed — fast polling running") return direction = 1 if current_z >= cfg["threshold"] else -1 zscore_abs = abs(current_z) # Fib entry target derivation: # For EU (15M cadence): use the SIGNAL bar's OHLC (the 15M bar that just # crossed the threshold = the last row of df from compute_zscore). # This matches backtest's find_entry_pullback which uses signal row. # For UJ (hourly cadence): use the latest 15M bar (subdivision of the # hourly signal bar), matching v2's existing behaviour. if cfg.get("signal_timeframe") == "15m": # df is 15M bars with z-score. Use df.iloc[-2] = LAST COMPLETED bar # for Fib target, NOT df.iloc[-1] = in-progress current bar. # Bug found 2026-05-27: z=+2.81 (11:31 UTC) and z=+3.70 (16:01 UTC) # signals silently didn't arm because the in-progress bar (1 minute # of tick data) had C-L < 1 pip, hitting the Fib pull guard. Using # the completed prior bar gives stable H/L/C from a full 15 minutes # of price action — same source the backtest uses. signal_bar = df.iloc[-2] bh = float(signal_bar["high"]) bl = float(signal_bar["low"]) bc = float(signal_bar["close"]) else: bars_15m = get_bars_15m(symbol, 10) if bars_15m is None or bars_15m.empty: log.warning(f"[{pair}] No 15M bars for Fib entry") return last_bar = bars_15m.iloc[-1] bh = float(last_bar["high"]) bl = float(last_bar["low"]) bc = float(last_bar["close"]) if direction == 1: pull = bc - bl if pull <= 0.0001: # Diagnostic logging — previously this guard silent-returned with # no log entry, making it impossible to tell why a threshold- # crossing signal didn't arm. Bug found 2026-05-27. log.info(f"[{pair}] SIGNAL z={current_z:+.3f} dir=LONG NOT ARMED: " f"bar pull (C-L) = {pull*10000:.2f}p < 1.0p Fib guard | " f"bar H={bh:.5f} L={bl:.5f} C={bc:.5f}") save_state(pair) return target = bc - cfg["fib"] * pull else: pull = bh - bc if pull <= 0.0001: log.info(f"[{pair}] SIGNAL z={current_z:+.3f} dir=SHORT NOT ARMED: " f"bar pull (H-C) = {pull*10000:.2f}p < 1.0p Fib guard | " f"bar H={bh:.5f} L={bl:.5f} C={bc:.5f}") save_state(pair) return target = bc + cfg["fib"] * pull log.info(f"[{pair}] SIGNAL DETECTED | " f"z={current_z:.4f} | dir={'LONG' if direction==1 else 'SHORT'} | " f"target={target:.{dp}f}") send_telegram( f"🎯 [{pair}] SIGNAL DETECTED\n" f"Z-score: {current_z:.4f} (threshold ±{cfg['threshold']})\n" f"Direction: {'LONG' if direction==1 else 'SHORT'}\n" f"Fib target: {target:.{dp}f}\n" f"Watching for entry (6h window)..." ) arm_fast_polling(pair, direction, target, zscore_abs) save_state(pair) def run_signal_check(): """Run signal check for all configured pairs. Fully exception-safe.""" for pair in PAIR_CONFIG: try: run_signal_check_pair(pair) except KeyboardInterrupt: raise # let Ctrl+C propagate cleanly except Exception as e: log.error(f"[{pair}] Signal check exception: {e}") import traceback; traceback.print_exc() # Don't let one pair failure block the other pair # After each pair's check, refresh live P&L for any open positions # so the dashboard can display floating unrealized P&L try: update_live_position_pnl() except Exception as e: log.warning(f"Live P&L update failed: {e}") # Also write authoritative account snapshot for the dashboard server try: update_account_snapshot() except Exception as e: log.warning(f"Account snapshot update failed: {e}") def update_account_snapshot(): """ Write authoritative MT5 account state + per-pair trade stats to data/logs/account_snapshot.json so the dashboard server can display real balance-derived P&L without needing its own MT5 connection. Strategy — combines two sources to avoid MT5 deal-history quirks: • Total P&L from MT5 balance − starting balance (perfectly accurate, net of fees) • Per-pair trade count from the JSON trade history files (monitor's authoritative record of which positions it executed — immune to MT5 occasionally missing a position in history_deals_get grouping) • Per-pair P&L: prefer MT5 deal-grouped sum if it matches the JSON count, else use the JSON-derived sum (gross) and log a warning """ try: import MetaTrader5 as mt5 from datetime import timezone as _tz STARTING_BALANCE = 100_000.0 acct = mt5.account_info() if acct is None: return balance = float(acct.balance) equity = float(acct.equity) floating = equity - balance # NOTE: balance - STARTING_BALANCE captures activity on ALL instruments # ever traded on this account, not just our two strategy pairs. If the # operator places ad-hoc trades on other symbols (e.g. USTEC tests on # the demo account), those flows would contaminate strategy reporting. # account_balance_delta is kept for diagnostic logging only — the # snapshot's pnl_total is computed below from EU+UJ deals only. account_balance_delta = balance - STARTING_BALANCE now_utc = datetime.now(_tz.utc) start = now_utc - timedelta(days=90) deals = mt5.history_deals_get(start, now_utc) if deals is None: deals = [] DEAL_BUY, DEAL_SELL = 0, 1 our_symbols = {"EURUSD", "USDJPY"} # Group deals by position_id for per-pair P&L computation by_position = {} for d in deals: if d.type not in (DEAL_BUY, DEAL_SELL): continue if d.symbol not in our_symbols: continue if not d.position_id: continue by_position.setdefault((d.symbol, d.position_id), []).append(d) # Augment incomplete positions (date-range query sometimes omits # the exit deal on IC Markets demo - 2026-06-01 root cause). For # any position with <2 deals, re-fetch via the position query. for (_sym, _pid), _dl in list(by_position.items()): if len(_dl) < 2: _ps = mt5.history_deals_get(position=_pid) if _ps and len(_ps) > len(_dl): by_position[(_sym, _pid)] = [d for d in _ps if d.symbol == _sym] mt5_pair_pnl = {"EURUSD": 0.0, "USDJPY": 0.0} mt5_pair_trades = {"EURUSD": 0, "USDJPY": 0} today_pnl_mt5 = 0.0 today = now_utc.date() for (symbol, _pid), pos_deals in by_position.items(): if len(pos_deals) < 2: continue # position still open or partial net = sum(d.profit + d.swap + d.commission for d in pos_deals) mt5_pair_pnl[symbol] += net mt5_pair_trades[symbol] += 1 close_time = datetime.fromtimestamp(max(d.time for d in pos_deals)) if close_time.date() == today: today_pnl_mt5 += net # JSON trade counts — authoritative for "how many trades did we execute" json_pair_pnl = {"EURUSD": 0.0, "USDJPY": 0.0} json_pair_trades = {"EURUSD": 0, "USDJPY": 0} for pair in ("EURUSD", "USDJPY"): try: path = LOG_DIR / PAIR_CONFIG[pair]["trade_log"] if path.exists(): trades = json.loads(path.read_text(encoding="utf-8")) json_pair_trades[pair] = len(trades) json_pair_pnl[pair] = sum(t.get("net_pnl", 0) for t in trades) except Exception: pass # Reconcile: for each pair, use MT5 sum if count matches JSON, else JSON # (MT5 can silently miss a position in its grouping — we trust the JSON # count because it's written on every trade close by this same monitor) final_pair_pnl = {} final_pair_trades = {} for pair in ("EURUSD", "USDJPY"): if mt5_pair_trades[pair] == json_pair_trades[pair] and json_pair_trades[pair] > 0: # MT5 sees same count — use its net-of-fees P&L final_pair_pnl[pair] = mt5_pair_pnl[pair] final_pair_trades[pair] = mt5_pair_trades[pair] else: # Counts disagree (MT5 missed one) — trust JSON final_pair_pnl[pair] = json_pair_pnl[pair] final_pair_trades[pair] = json_pair_trades[pair] if json_pair_trades[pair] > mt5_pair_trades[pair]: log.info(f"[{pair}] Snapshot: using JSON counts " f"(JSON={json_pair_trades[pair]} vs MT5={mt5_pair_trades[pair]})") total_trades = sum(final_pair_trades.values()) daily_loss = max(0.0, -today_pnl_mt5) # Strategy-only total: sum of EU + UJ closed-position P&L. Does NOT # include any activity on other instruments (USTEC, gold, etc.) the # operator may have on the same demo account. The previous version # used `balance - STARTING_BALANCE` and then rescaled per-pair sums # to match — which silently attributed non-strategy losses to the # smaller-volume pair (bug found 2026-05-23: USDJPY showed -$5,816 # in snapshot when actual EU+UJ closed P&L was -$2,677). total_pnl = sum(final_pair_pnl.values()) # Diagnostic: log the gap between account balance delta and our # strategy-only P&L. Non-zero values indicate trades on other # instruments — useful info, not an error. non_strategy_pnl = account_balance_delta - total_pnl if abs(non_strategy_pnl) > 5.0: log.info(f"[snapshot] account balance delta ${account_balance_delta:+,.2f} " f"vs strategy P&L ${total_pnl:+,.2f} — " f"non-strategy activity ${non_strategy_pnl:+,.2f}") snapshot = { "snapshot_at" : now_utc.isoformat(), "balance" : round(balance, 2), "equity" : round(equity, 2), "floating_pnl" : round(floating, 2), "pnl_total" : round(total_pnl, 2), "daily_loss" : round(daily_loss, 2), "trade_count" : total_trades, "pair_pnl" : {k: round(v, 2) for k, v in final_pair_pnl.items()}, "pair_trades" : final_pair_trades, "starting_balance": STARTING_BALANCE, } path = LOG_DIR / "account_snapshot.json" # Atomic write (same pattern as save_state) to avoid partial reads tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(snapshot, indent=2), encoding="utf-8") import os as _os _os.replace(str(tmp), str(path)) log.info(f"Account snapshot: balance=${balance:,.2f} · " f"P&L ${total_pnl:+,.2f} · " f"trades={total_trades} (EU:{final_pair_trades['EURUSD']} " f"UJ:{final_pair_trades['USDJPY']}) · " f"daily_loss=${daily_loss:,.2f}") except ImportError: pass except Exception as e: log.warning(f"update_account_snapshot failed: {e}") def update_live_position_pnl(): """ For any open position, query MT5 for current price + unrealized P&L and persist to the state file so the dashboard can display floating P&L. Writes these fields into state_{pair}.json when position_open=True: current_price : latest bid/ask floating_pnl : unrealized P&L in account currency (from MT5 position.profit) floating_pips : distance from entry in pips bars_held : hours held so far bars_remaining : hours until hold_hours limit Removes these fields when no position is open (keeps state clean). """ try: import MetaTrader5 as mt5 except ImportError: return for pair, state in PAIR_STATE.items(): try: if not state.get("position_open") or state.get("position_ticket") is None: # Clear stale live fields when no position open for k in ["current_price", "floating_pnl", "floating_pips", "bars_held", "bars_remaining"]: state.pop(k, None) save_state(pair) continue cfg = PAIR_CONFIG[pair] ticket = state["position_ticket"] positions = mt5.positions_get(ticket=ticket) if not positions: # Position no longer exists — let next signal check detect close continue pos = positions[0] entry_px = state.get("entry_price") or pos.price_open direction = state.get("position_signal") # 1 long, -1 short tick = mt5.symbol_info_tick(cfg["symbol"]) if not tick: continue current_px = tick.bid if direction == 1 else tick.ask pip_size = cfg["pip_size"] floating_pips = ((current_px - entry_px) / pip_size) if direction == 1 \ else ((entry_px - current_px) / pip_size) # Hours held bars_held = None bars_remaining = None entry_time = state.get("entry_time") if entry_time: if isinstance(entry_time, str): try: entry_time = datetime.fromisoformat(entry_time) except Exception: entry_time = None if entry_time: bars_held = (datetime.now() - entry_time).total_seconds() / 3600 bars_remaining = max(0, cfg["hold_hours"] - bars_held) state["current_price"] = round(current_px, cfg["dp"]) state["floating_pnl"] = round(float(pos.profit), 2) state["floating_pips"] = round(floating_pips, 1) state["bars_held"] = round(bars_held, 2) if bars_held is not None else None state["bars_remaining"] = round(bars_remaining, 2) if bars_remaining is not None else None save_state(pair) log.info(f"[{pair}] Live P&L: ${pos.profit:+.2f} | " f"{floating_pips:+.1f} pips | " f"held={bars_held:.1f}h/{cfg['hold_hours']}h") except Exception as e: log.warning(f"[{pair}] Live P&L update failed: {e}") # ─── Combined Telegram daily summary ─────────────────────────────────────── def _pair_trade_stats(pair): """ Return (trade_count, pnl_total) for this pair. Prefers account_snapshot.json (MT5-derived, includes swap/commission/spread) over the raw trade history log (which records gross profit only). Falls back to the log if snapshot is missing or stale. """ # Try authoritative snapshot first try: snap_path = LOG_DIR / "account_snapshot.json" if snap_path.exists(): snap = json.loads(snap_path.read_text("utf-8")) # Snapshot freshness check — snapshot_at less than 2 hours old snap_at = snap.get("snapshot_at", "") if snap_at: try: snap_dt = datetime.fromisoformat(snap_at.replace("Z", "+00:00")) age_h = (datetime.now(snap_dt.tzinfo) - snap_dt).total_seconds() / 3600 if age_h <= 2.0: pair_pnl = snap.get("pair_pnl", {}).get(pair, 0.0) pair_cnt = snap.get("pair_trades", {}).get(pair, 0) return pair_cnt, pair_pnl except Exception: pass except Exception: pass # Fallback: raw JSON trade log (gross P&L only) try: path = LOG_DIR / PAIR_CONFIG[pair]["trade_log"] trades = json.loads(path.read_text("utf-8")) if path.exists() else [] return len(trades), sum(t["net_pnl"] for t in trades) except Exception: return 0, 0.0 def _format_pnl(tot): return f"+${tot:,.2f}" if tot >= 0 else f"-${abs(tot):,.2f}" def _pair_position_str(pair): state = PAIR_STATE[pair] if state.get("position_open"): ep = state.get("entry_price") d = state.get("position_signal") dp = PAIR_CONFIG[pair]["dp"] return f"🔵 {'LONG' if d==1 else 'SHORT'} @ {ep:.{dp}f}" if ep else "🔵 Open" return "⚪ No position" def _prob_bar(p): return "█"*round(p/10)+"░"*(10-round(p/10)) def _prob_emoji(p): return "🟢" if p>=60 else "🟡" if p>=35 else "⚪" def _build_outlook_section(pair, outlook, is_monday, today_str): """Build the signal outlook section in the exact user-preferred format.""" if not outlook: return "" pair_label = "EURUSD SIGNAL OUTLOOK" if pair == "EURUSD" else "USDJPY SIGNAL OUTLOOK" if is_monday: best = max(outlook, key=lambda x: x["prob"]) lines = [ "", f"📅 WEEK AHEAD — {pair_label}", "────────────────", ] for d in outlook: star = "⭐" if d["date"] == best["date"] and d["prob"] > 20 else "" evs = ", ".join(e["event"] for e in d["events"][:2]) if d["events"] else "No key events" lines.append( _prob_emoji(d["prob"]) + " " + d["day"] + " " + str(d["prob"]) + "% " + _prob_bar(d["prob"]) + " " + star + "\n └ " + evs ) lines += [ "────────────────", "⭐ Best day: " + best["day"] + " at " + str(best["prob"]) + "%", ] return "\n".join(lines) else: td = next((d for d in outlook if d["date"] == today_str), None) if not td: # Weekend or holiday — show next available trading day td = next((d for d in outlook if d["prob"] is not None), None) if not td: return "" day_name = datetime.now().strftime("%a") # Relabel as "next trading day" rather than "today" td = dict(td) # copy so we don't mutate td["day"] = td["day"] + " (next trading day)" evs = "\n".join( " • " + e["event"] + " (" + e["country"] + ")" for e in td["events"][:3] ) if td["events"] else " • No key macro events today" return ( "\n📅 TODAY'S " + pair_label + " — " + td["day"] + "\n" "────────────────\n" + _prob_emoji(td["prob"]) + " Signal probability: " + str(td["prob"]) + "%\n" + _prob_bar(td["prob"]) + "\n" "Key events:\n" + evs + "\n" "────────────────" ) def send_daily_summary(header="📊 DAILY SUMMARY"): global _LAST_DAILY_SUMMARY_SENT # Dedup: never send two summaries within 30 minutes of each other _now = datetime.now(timezone.utc) if (_LAST_DAILY_SUMMARY_SENT is not None and (_now - _LAST_DAILY_SUMMARY_SENT).total_seconds() < 1800 and header == "📊 DAILY SUMMARY"): log.info("Daily summary skipped — sent within last 30 minutes") return _LAST_DAILY_SUMMARY_SENT = _now today_str = datetime.now().strftime("%Y-%m-%d") is_monday = datetime.now().weekday() == 0 eu_z = PAIR_STATE["EURUSD"].get("last_zscore") eu_z_s = f"{eu_z:.3f}" if eu_z is not None else "—" eu_pct = f"{abs(eu_z)/2.75*100:.0f}% of threshold" if eu_z else "" eu_n, eu_tot = _pair_trade_stats("EURUSD") uj_z = PAIR_STATE["USDJPY"].get("last_zscore") uj_z_s = f"{uj_z:.3f}" if uj_z is not None else "—" uj_pct = f"{abs(uj_z)/2.0*100:.0f}% of threshold" if uj_z else "" uj_n, uj_tot = _pair_trade_stats("USDJPY") co_n = eu_n + uj_n co_tot = eu_tot + uj_tot eu_outlook_text = "" uj_outlook_text = "" try: # Import ONCE — importing from two threads simultaneously triggers # Python's import lock and one thread silently fails. # send_daily_summary already runs in a background thread so sequential # calls here are fine — they never block the scheduler. from execution.dashboard_server import ( get_weekly_outlook, get_usdjpy_weekly_outlook ) eu_out = get_weekly_outlook(float(eu_z or 0)) uj_out = get_usdjpy_weekly_outlook(float(uj_z or 0)) eu_outlook_text = _build_outlook_section( "EURUSD", eu_out, is_monday, today_str) uj_outlook_text = _build_outlook_section( "USDJPY", uj_out, is_monday, today_str) except Exception as _e: log.warning(f"Outlook fetch failed: {_e}") # Fallback — show section header even if calendar unavailable day_name = datetime.now().strftime("%a") eu_outlook_text = ( f"\n📅 TODAY'S EURUSD SIGNAL OUTLOOK — {day_name}\n" f"────────────────\n" f"Calendar unavailable — check log\n" f"────────────────" ) uj_outlook_text = "" send_telegram( f"{header}\n" f"────────────────\n" f"🟢 EURUSD\n" f"Z-score: {eu_z_s} ({eu_pct})\n" f"Position: {_pair_position_str('EURUSD')}\n" f"Trades: {eu_n} | PnL: {_format_pnl(eu_tot)}\n" f"────────────────\n" f"🔵 USDJPY\n" f"Z-score: {uj_z_s} ({uj_pct})\n" f"Position: {_pair_position_str('USDJPY')}\n" f"Trades: {uj_n} | PnL: {_format_pnl(uj_tot)}\n" f"────────────────\n" f"Combined: {co_n} trades | {_format_pnl(co_tot)}\n" f"System: ✅ Running\n" f"────────────────\n" f"Thresholds: EURUSD ±2.75 · USDJPY ±2.00" f"{eu_outlook_text}" f"{uj_outlook_text}" ) # ─── Weekly Telegram summary (Sundays) ───────────────────────────────────── def send_weekly_summary(): """ Sunday 06:05 UTC — week-ahead outlook replacing the daily summary. Shows full 5-day signal probability calendar for the coming week. """ try: from execution.dashboard_server import ( get_weekly_outlook, get_usdjpy_weekly_outlook ) except Exception: # Fall back to daily if outlook unavailable send_daily_summary(header="📊 WEEKLY OUTLOOK") return eu_z = PAIR_STATE["EURUSD"].get("last_zscore") or 0.0 uj_z = PAIR_STATE["USDJPY"].get("last_zscore") or 0.0 try: eu_out = get_weekly_outlook(float(eu_z)) uj_out = get_usdjpy_weekly_outlook(float(uj_z)) except Exception: send_daily_summary(header="📊 WEEKLY OUTLOOK") return def _bar(pct): filled = int(pct / 10) return "█" * filled + "░" * (10 - filled) def _icon(pct): if pct >= 60: return "🟢" if pct >= 35: return "🟡" return "⚪" def _build_week_block(days, label, threshold_label): if not days: return "" best_day = max(days, key=lambda d: d.get("signal_probability", 0)) best_pct = best_day.get("signal_probability", 0) lines = [f"\n📅 WEEK AHEAD — {label}\n────────────────"] for d in days: pct = d.get("signal_probability", 0) date = d.get("date","") day = d.get("day","") top = d.get("top_event","No key events") icon = _icon(pct) star = " ⭐" if d is best_day and best_pct >= 30 else "" lines.append( f"{icon} {day} {date[5:]} {pct}% {_bar(pct)}{star}\n" f" └ {top}" ) lines.append(f"────────────────") if best_pct >= 30: lines.append(f"⭐ Best day: {best_day.get('day')} at {best_pct}%") lines.append(f"% = prob z-score crosses {threshold_label}") return "\n".join(lines) eu_z_s = f"{eu_z:.3f}" if eu_z else "—" uj_z_s = f"{uj_z:.3f}" if uj_z else "—" eu_n, eu_tot = _pair_trade_stats("EURUSD") uj_n, uj_tot = _pair_trade_stats("USDJPY") co_n = eu_n + uj_n co_tot = eu_tot + uj_tot eu_week = _build_week_block(eu_out, "EURUSD SIGNAL OUTLOOK", "±2.75") uj_week = _build_week_block(uj_out, "USDJPY SIGNAL OUTLOOK", "±2.00") send_telegram( f"📆 WEEKLY OUTLOOK\n" f"────────────────\n" f"🟢 EURUSD z={eu_z_s}\n" f"🔵 USDJPY z={uj_z_s}\n" f"Trades: {co_n} | PnL: {_format_pnl(co_tot)}\n" f"System: ✅ Running\n" f"{eu_week}" f"{uj_week}" ) # ─── Morning briefing (London open, 07:00 UTC) ───────────────────────────── def send_morning_briefing(): """ Fires at 07:00 UTC (London open). Unlike the 06:05 daily summary which is a full system report, this is a focused 'today's plan' briefing: • Current z-scores with sizing tier info • Top 3 scheduled events for each pair today (with times/forecasts) • Position status with floating P&L if open """ eu_z = PAIR_STATE["EURUSD"].get("last_zscore") uj_z = PAIR_STATE["USDJPY"].get("last_zscore") eu_z_s = f"{eu_z:+.3f}" if eu_z is not None else "—" uj_z_s = f"{uj_z:+.3f}" if uj_z is not None else "—" # Sizing tier for each pair given current z def _tier_text(pair, z): if z is None: return "—" absZ = abs(z) cfg = PAIR_CONFIG[pair] if absZ < cfg["threshold"]: pct = absZ / cfg["threshold"] * 100 return f"{pct:.0f}% to signal (base 1.0× if triggered)" for lo, hi, mult in cfg["sizing_bands"]: if lo <= absZ < hi: return f"would size {mult}× at current z" return f"would size {cfg['sizing_bands'][-1][2]}× (max tier) at current z" eu_tier = _tier_text("EURUSD", eu_z) uj_tier = _tier_text("USDJPY", uj_z) # Position info def _pos_line(pair): state = PAIR_STATE[pair] if not state.get("position_open"): return "No position" dp = PAIR_CONFIG[pair]["dp"] dir_str = "LONG" if state.get("position_signal") == 1 else "SHORT" entry = state.get("entry_price") fpnl = state.get("floating_pnl") fpips = state.get("floating_pips") held = state.get("bars_held") hold_limit = PAIR_CONFIG[pair]["hold_hours"] parts = [f"{dir_str} @ {entry:.{dp}f}"] if fpnl is not None: sign = "+" if fpnl >= 0 else "" parts.append(f"P&L {sign}${fpnl:,.2f}") if fpips is not None: sign = "+" if fpips >= 0 else "" parts.append(f"{sign}{fpips:.1f}p") if held is not None: parts.append(f"{held:.1f}h/{hold_limit}h") return " · ".join(parts) # Today's events from weekly outlook cache today_str = datetime.now().strftime("%Y-%m-%d") eu_today_events = "No calendar data" uj_today_events = "No calendar data" try: from execution.dashboard_server import ( get_weekly_outlook, get_usdjpy_weekly_outlook ) eu_out = get_weekly_outlook(float(eu_z or 0)) uj_out = get_usdjpy_weekly_outlook(float(uj_z or 0)) def _format_today(outlook): td = next((d for d in outlook if d.get("date") == today_str), None) if not td or not td.get("events"): return "No key events today" lines = [] for e in td["events"][:3]: time_str = e.get("time", "") or "--:--" fc = e.get("forecast", "") fc_part = f" · Fcst {fc}" if fc else "" lines.append(f" {time_str} {e['event']} ({e.get('country','')[:2]}){fc_part}") return "\n".join(lines) eu_today_events = _format_today(eu_out) uj_today_events = _format_today(uj_out) except Exception as e: log.warning(f"Morning briefing outlook fetch failed: {e}") day_name = datetime.now().strftime("%A, %d %B") send_telegram( f"🌅 MORNING BRIEFING — {day_name}\n" f"────────────────\n" f"🟢 EURUSD z={eu_z_s}\n" f" {eu_tier}\n" f" Position: {_pos_line('EURUSD')}\n" f" Events today:\n{eu_today_events}\n" f"────────────────\n" f"🔵 USDJPY z={uj_z_s}\n" f" {uj_tier}\n" f" Position: {_pos_line('USDJPY')}\n" f" Events today:\n{uj_today_events}\n" f"────────────────\n" f"London session: 07:00–17:00 EET · good luck" ) # ─── Pre-event alerts (30min before high-impact releases) ────────────────── _pre_event_sent = set() # keys: "YYYY-MM-DD HH:MM|EVENT" — prevent duplicate alerts def _minutes_until_event_uk(event_date_str, event_time_str): """ Calendar times from FinanceFlow are already converted to UK time by the dashboard_server's _fetch_te_calendar. Compare to current UK time. Returns minutes until event (negative if already past), or None if unparseable. """ try: if not event_time_str or not event_date_str: return None hh, mm = event_time_str.split(":") event_dt = datetime.strptime(event_date_str, "%Y-%m-%d").replace( hour=int(hh), minute=int(mm) ) # UK time approximation: UTC + 1 Apr-Oct, UTC + 0 Nov-Mar now_utc = datetime.now(timezone.utc).replace(tzinfo=None) uk_offset = 1 if 3 < now_utc.month < 11 else 0 now_uk = now_utc + timedelta(hours=uk_offset) delta_min = (event_dt - now_uk).total_seconds() / 60 return delta_min except Exception: return None def check_pre_event_alerts(): """ Run every 5 minutes. Scans today's cached weekly outlook for any event with score >= 80 (FOMC, ECB rate decision, CPI, NFP, BoJ) that's scheduled to release in the next 30-35 minutes. Fires one Telegram alert per event (deduplicated via _pre_event_sent). """ try: from execution.dashboard_server import ( get_weekly_outlook, get_usdjpy_weekly_outlook ) except Exception as e: log.warning(f"Pre-event alert import failed: {e}") return today_str = datetime.now().strftime("%Y-%m-%d") eu_z = PAIR_STATE["EURUSD"].get("last_zscore") or 0.0 uj_z = PAIR_STATE["USDJPY"].get("last_zscore") or 0.0 def _scan(outlook, pair_label): try: td = next((d for d in outlook if d.get("date") == today_str), None) if not td or not td.get("events"): return for ev in td["events"]: score = ev.get("score", 0) if score < 80: continue # only high-impact events ev_time = ev.get("time", "") if not ev_time: continue mins = _minutes_until_event_uk(today_str, ev_time) if mins is None: continue # Fire 30-35 minute window (caught by any 5-min check in that window) if not (25 <= mins <= 35): continue key = f"{today_str} {ev_time}|{ev.get('event','')}" if key in _pre_event_sent: continue _pre_event_sent.add(key) # Current z context for pair if pair_label == "EURUSD": z, th = eu_z, 2.75 else: z, th = uj_z, 2.0 z_pct = abs(z) / th * 100 if z else 0 gap = th - abs(z) if z else th z_context = (f"z = {z:+.3f} ({z_pct:.0f}% of threshold, {gap:+.2f}σ away)" if z else "z unavailable") fc = ev.get("forecast", "") prev = ev.get("previous", "") ctx_parts = [] if fc: ctx_parts.append(f"Fcst {fc}") if prev: ctx_parts.append(f"Prev {prev}") ctx = " · ".join(ctx_parts) if ctx_parts else "No consensus" send_telegram( f"⏰ EVENT IN ~{int(round(mins))} MIN\n" f"────────────────\n" f"{ev.get('event','?')} ({ev.get('country','?')})\n" f"Time: {ev_time} UK · Impact: {score}/100\n" f"{ctx}\n" f"────────────────\n" f"{pair_label} current: {z_context}" ) log.info(f"Pre-event alert fired: {pair_label} {ev.get('event','')} in {mins:.0f}min") except Exception as e: log.warning(f"Pre-event scan failed for {pair_label}: {e}") # Scan both pairs' outlooks try: eu_out = get_weekly_outlook(eu_z) _scan(eu_out, "EURUSD") except Exception as e: log.warning(f"EURUSD pre-event scan failed: {e}") try: uj_out = get_usdjpy_weekly_outlook(uj_z) _scan(uj_out, "USDJPY") except Exception as e: log.warning(f"USDJPY pre-event scan failed: {e}") # ─── Main ────────────────────────────────────────────────────────────────── def main(): log.info("="*60) log.info("LIVE SIGNAL MONITOR v3 — EURUSD + USDJPY") for pair, cfg in PAIR_CONFIG.items(): log.info(f" {pair}: threshold=±{cfg['threshold']} " f"TP={cfg['tp_pct']:.2%} SL={cfg['stop_pct']:.2%} " f"hold={cfg['hold_hours']}h") log.info("="*60) # Init MT5 try: import MetaTrader5 as mt5 if not mt5.initialize(): log.error(f"MT5 init failed: {mt5.last_error()}") sys.exit(1) log.info(f"MT5 connected | terminal={mt5.terminal_info().name}") except ImportError: log.warning("MetaTrader5 not installed — running in simulation mode") # Restore state for both pairs for pair in PAIR_CONFIG: load_state(pair) # Reconcile trade history from MT5 deal history (back-fill any missing closes) # Runs at startup before any scheduler jobs; safe because all it does is # read MT5 history and append missing records to trade_history_*.json. reconcile_trade_history_from_mt5(lookback_days=90) # Initial rate update update_rate_data() # Schedule def _safe_signal_check_pair(pair): """Run signal check for a single pair, exception-safe. Also refreshes the account snapshot after each signal check so the dashboard never sees more than ~15 minutes of staleness. Previously update_account_snapshot() was only called from the legacy combined run_signal_check() function which only ran on startup, leading to 15+ hour stale snapshots on the dashboard. """ try: run_signal_check_pair(pair) except KeyboardInterrupt: raise except Exception as e: log.error(f"[{pair}] signal_check job error: {e}") # Refresh account snapshot after each per-pair check so the dashboard # keeps current balance, equity, and live trade P&L visible. try: update_account_snapshot() except Exception as e: log.warning(f"Account snapshot update failed: {e}") def _safe_signal_check_eurusd(): _safe_signal_check_pair("EURUSD") def _safe_signal_check_usdjpy(): _safe_signal_check_pair("USDJPY") def _safe_signal_check(): """Legacy combined check — kept for startup priming only.""" try: run_signal_check() except KeyboardInterrupt: raise except Exception as e: log.error(f"signal_check job error: {e}") def _safe_rate_update(): try: update_rate_data() except KeyboardInterrupt: raise except Exception as e: log.error(f"rate_update job error: {e}") _summary_lock = __import__('threading').Lock() def _safe_daily_summary(): # Sunday → send week-ahead outlook; Mon–Sat → send daily summary import threading as _t2 def _run(): with _summary_lock: if datetime.now().weekday() == 6: # Sunday send_weekly_summary() else: send_daily_summary() _t2.Thread(target=_run, daemon=True).start() # ── Inline watchdog health check ─────────────────────────────────────── _last_watchdog_alert = {} # track last alert time per check to avoid spam def _watchdog_health_check(): """ Runs every 5 minutes inside the existing monitor process. Checks: 1. State files updated recently (detects hung/frozen state) 2. Rate CSVs updated within 36h (detects data pipeline failure) 3. MT5 connection still alive (detects broker disconnect) 4. Open positions still exist in MT5 (detects orphaned state) Sends Telegram alerts on issues — no separate process needed. """ import threading as _wt def _run_watchdog(): now = datetime.now(timezone.utc) # ── Check 1: State file freshness ────────────────────────────── for pair in PAIR_CONFIG: sf = LOG_DIR / PAIR_CONFIG[pair]["state_file"] if not sf.exists(): continue age_h = (now.timestamp() - sf.stat().st_mtime) / 3600 if age_h > 2.0: key = f"stale_state_{pair}" last = _last_watchdog_alert.get(key) if last is None or (now - last).total_seconds() > 7200: _last_watchdog_alert[key] = now log.warning(f"[WATCHDOG] {pair} state file stale ({age_h:.1f}h)") send_telegram( f"⚠️ [WATCHDOG] {pair} STATE FILE STALE\n" f"Last updated: {age_h:.1f}h ago\n" f"Monitor may be running but not writing state.\n" f"Check live_monitor.log for errors." ) # ── Check 2: Rate file freshness ─────────────────────────────── rate_files = {"us2y": "us2y.csv", "de2y": "de2y.csv", "jp2y": "jp2y.csv"} stale_rates = [] for name, fname in rate_files.items(): rf = RATES_DIR / fname if not rf.exists(): stale_rates.append(f"{name} (missing)") continue age_h = (now.timestamp() - rf.stat().st_mtime) / 3600 if age_h > 36: stale_rates.append(f"{name} ({age_h:.0f}h old)") if stale_rates: key = "stale_rates" last = _last_watchdog_alert.get(key) if last is None or (now - last).total_seconds() > 3600: _last_watchdog_alert[key] = now log.warning(f"[WATCHDOG] Stale rate files: {stale_rates}") send_telegram( f"⚠️ [WATCHDOG] RATE DATA STALE\n" f"Files: {', '.join(stale_rates)}\n" f"FinanceFlow pipeline may have failed.\n" f"Signals continue on last known rates — monitor closely." ) # ── Check 3: MT5 connection ──────────────────────────────────── try: import MetaTrader5 as mt5 if not mt5.terminal_info(): key = "mt5_disconnect" last = _last_watchdog_alert.get(key) if last is None or (now - last).total_seconds() > 1800: _last_watchdog_alert[key] = now log.warning("[WATCHDOG] MT5 terminal not responding") send_telegram( "⚠️ [WATCHDOG] MT5 DISCONNECTED\n" "Terminal not responding.\n" "Attempting reconnect on next signal check.\n" "Check open positions in MT5 manually." ) except Exception: pass # ── Check 4: Orphaned state (state says open, MT5 says closed) ─ # Only alert if state hasn't been refreshed by the hourly signal check # for 90+ minutes. TP/SL closes happen mid-hour and the next hourly # check will reconcile cleanly — we don't want to spam alerts for # perfectly normal trade closes between watchdog runs. try: import MetaTrader5 as mt5 for pair in PAIR_CONFIG: state = PAIR_STATE[pair] if state.get("position_open") and state.get("position_ticket"): ticket = state["position_ticket"] positions = mt5.positions_get(ticket=ticket) if not positions: # Check state file modification time — if it was # updated in the last 90 min, the hourly signal check # is still reconciling things normally. Stay quiet. sf = LOG_DIR / PAIR_CONFIG[pair]["state_file"] if sf.exists(): sf_age_min = (now.timestamp() - sf.stat().st_mtime) / 60 if sf_age_min < 90: # State is fresh — hourly check will handle it continue # Truly orphaned: state file stale AND MT5 has no position key = f"orphan_{pair}_{ticket}" last = _last_watchdog_alert.get(key) if last is None or (now - last).total_seconds() > 3600: _last_watchdog_alert[key] = now log.warning( f"[WATCHDOG] {pair} orphaned state — " f"ticket {ticket} not in MT5" ) send_telegram( f"🔴 [WATCHDOG] {pair} ORPHANED POSITION\n" f"State file says open (ticket {ticket})\n" f"MT5 has no matching position.\n" f"The trade was likely closed externally.\n" f"Monitor will detect this on next hourly check." ) except Exception: pass # Run in daemon thread so it never blocks the scheduler _wt.Thread(target=_run_watchdog, daemon=True).start() schedule.every(5).minutes.do(_watchdog_health_check) log.info("Inline watchdog enabled — health checks every 5 minutes") # ── Signal check schedules — per-pair cadence ────────────────────────── # EURUSD: every 15 minutes at :01, :16, :31, :46 — Formula A 15M cadence # USDJPY: every hour at :01 — daily spread z-score (no benefit to faster) schedule.every().hour.at(":01").do(_safe_signal_check_eurusd) schedule.every().hour.at(":16").do(_safe_signal_check_eurusd) schedule.every().hour.at(":31").do(_safe_signal_check_eurusd) schedule.every().hour.at(":46").do(_safe_signal_check_eurusd) schedule.every().hour.at(":02").do(_safe_signal_check_usdjpy) schedule.every().day.at("06:00").do(_safe_rate_update) schedule.every().day.at("06:05").do(_safe_daily_summary) # Morning briefing at London open (07:00 UTC) def _safe_morning_briefing(): import threading as _t3 def _run(): try: send_morning_briefing() except Exception as e: log.error(f"morning briefing error: {e}") _t3.Thread(target=_run, daemon=True).start() schedule.every().day.at("07:00").do(_safe_morning_briefing) # Pre-event alerts every 5 min, ~30 min before score ≥ 80 releases def _safe_pre_event_check(): try: check_pre_event_alerts() except Exception as e: log.error(f"pre-event alert error: {e}") schedule.every(5).minutes.do(_safe_pre_event_check) # Telegram command polling — listens for /add /remove /list etc. # Runs every 30s in a daemon thread so command responses are quick. def _safe_telegram_poll(): import threading as _tpt def _run(): try: poll_telegram_commands() except Exception as e: log.debug(f"Telegram poll error: {e}") _tpt.Thread(target=_run, daemon=True).start() schedule.every(30).seconds.do(_safe_telegram_poll) log.info("Scheduler running") log.info(" EURUSD signal check: every 15min (:01, :16, :31, :46) — Formula A validated") log.info(" USDJPY signal check: every hour at :02 — daily spread z-score") log.info(" Rate update daily at 06:00 UTC") log.info(" Daily Telegram summary at 06:05 UTC") log.info(" Morning briefing daily at 07:00 UTC (London open)") log.info(" Pre-event alerts: every 5min, 30min before score≥80 releases") log.info(" Telegram commands: polled every 30s " "(/add /remove /list /requestaccess /announcement " "/requestdashboard /grantdashboard /revokedashboard /listdashboard)") # Run first signal check so z-scores are populated before startup message run_signal_check() # Startup Telegram — run in background thread so it never blocks the # scheduler even if FinanceFlow API is slow on startup import threading as _threading _t = _threading.Thread( target=lambda: send_daily_summary(header="🚀 MONITOR STARTED (v3 — Formula A)"), daemon=True ) _t.start() # ── Main scheduler loop — bulletproof ────────────────────────────────── # Each job wrapped in exception handler so one failure never kills the loop. # Consecutive error counter resets on any successful run. _consecutive_errors = 0 _MAX_ERRORS = 10 # restart-worthy if this many consecutive failures while True: try: schedule.run_pending() _consecutive_errors = 0 except Exception as _loop_err: _consecutive_errors += 1 log.error(f"Scheduler loop error #{_consecutive_errors}: {_loop_err}") if _consecutive_errors >= _MAX_ERRORS: log.error("Too many consecutive errors — sending alert and continuing") send_telegram( f"⚠️ MONITOR WARNING\n" f"Scheduler loop error ×{_consecutive_errors}\n" f"Last error: {str(_loop_err)[:100]}\n" f"Monitor is still running — investigate log." ) _consecutive_errors = 0 # reset after alert time.sleep(1) if __name__ == "__main__": main()