#!/usr/bin/env python3
"""
live_signal_monitor_v2.py
==========================
Unified live execution monitor — EURUSD + USDJPY macro mean reversion models.

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 (validated model).
  • USDJPY uses the simple spread z-score (US-JP 2Y, z30).
  • 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
  USDJPY  z±2.00  TP=0.70%  SL=0.40%  hold=24h  z_exit=off

Run from project root:
  python src/execution/live_signal_monitor_v2.py

PATCHED 2026-05-06:
  • _get_closed_pnl now retries up to 6× with 1s backoff to handle MT5
    history table population delay. Looks specifically for the EXIT deal
    (entry == DEAL_ENTRY_OUT) before computing P&L. Fixes the bug where
    close notifications reported $0.00 on real losses.
  • New /announcement <message> admin command broadcasts to all recipients
    with a "📣 ANNOUNCEMENT FROM ADMIN" header and confirms delivery count.
  • New send_telegram_with_results helper returns (success, fail) counts.
"""

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":  7,            # EET hour
        "session_end":    17,
        "z_window":       60,
        "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",  # uses rolling OLS
        "bars_1h":        500,
        "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)
        "session_end":    17,
        "z_window":       30,
        "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(),
        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

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(
                "🔐 <b>Admin commands available</b>\n"
                "────────────────\n"
                "/list — show all recipients\n"
                "/add &lt;chat_id&gt; [name] — grant access\n"
                "/remove &lt;chat_id&gt; — revoke access\n"
                "/pending — see access requests\n"
                "/announcement &lt;message&gt; — broadcast to everyone\n"
                "/whoami — your chat ID\n"
                "────────────────\n"
                "<i>Commands must come from your admin chat only.</i>",
                chat_ids=[chat_id]
            )
        else:
            in_list = chat_id in _load_recipients()
            disclaimer = (
                "\n────────────────\n"
                "⚠️ <b>DISCLAIMER</b>\n\n"
                "This bot is provided for <b>educational and informational "
                "purposes only</b>.\n\n"
                "The signals, alerts, and commentary shared here are the "
                "output of a systematic quantitative model and <b>do not "
                "constitute financial advice</b>, 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. "
                "<b>You are solely responsible for your own trading "
                "decisions.</b>\n\n"
                "By using this bot you acknowledge and accept these terms."
            )
            if in_list:
                send_telegram(
                    "✅ <b>You're subscribed to alerts.</b>\n"
                    "────────────────\n"
                    "Available commands:\n"
                    "/whoami — see your chat ID\n"
                    "/leave — stop receiving alerts\n"
                    "/help — show this message again"
                    + disclaimer,
                    chat_ids=[chat_id]
                )
            else:
                send_telegram(
                    "👋 <b>Welcome</b>\n"
                    "────────────────\n"
                    "You are NOT currently authorised to receive alerts.\n\n"
                    "Available commands:\n"
                    "/requestaccess — request access from the owner\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: <code>{chat_id}</code>\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"📥 <b>ACCESS REQUEST</b>\n"
                f"────────────────\n"
                f"From: {display_name}\n"
                f"Chat ID: <code>{chat_id}</code>\n"
                f"────────────────\n"
                f"To approve: <code>/add {chat_id} {display_name}</code>\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

    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} (<code>{chat_id}</code>) 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 = ["👥 <b>ACTIVE RECIPIENTS</b>", "────────────────"]
            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"<code>{cid}</code>"
                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 = ["📥 <b>PENDING REQUESTS</b>", "────────────────"]
            for req_cid, info in _pending_access_requests.items():
                name = info.get("name", "")
                when = info.get("requested_at", "")[:16].replace("T", " ")
                lines.append(f"<code>{req_cid}</code> · {name} · {when}")
                lines.append(f"  Approve: <code>/add {req_cid} {name}</code>")
            send_telegram("\n".join(lines), chat_ids=[chat_id])
        return

    if cmd == "/add":
        if not args:
            send_telegram(
                "Usage: <code>/add &lt;chat_id&gt; [name]</code>\n"
                "Example: <code>/add 987654321 John</code>",
                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"ℹ️ <code>{new_id}</code> 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 <code>{new_id}</code>" + (f" ({name})" if name else "") + " to the alert list.",
                chat_ids=[chat_id]
            )
            # Notify the newly added user
            send_telegram(
                "🎉 <b>You're in!</b>\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 &amp; disclaimer\n"
                "/whoami — your chat ID\n"
                "/leave — stop receiving alerts\n"
                "────────────────\n"
                "⚠️ <b>DISCLAIMER</b>\n\n"
                "This bot is provided for <b>educational and informational "
                "purposes only</b>. Alerts do <b>not constitute financial "
                "advice</b> 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. <b>You are solely responsible for your own trading "
                "decisions.</b> 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: <code>/remove &lt;chat_id&gt;</code>",
                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"ℹ️ <code>{rm_id}</code> is not on the list.", chat_ids=[chat_id])
                return
            _save_recipients(data["recipients"])
            send_telegram(f"✅ Removed <code>{rm_id}</code> 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: <code>/announcement &lt;your message&gt;</code>\n"
                "Example: <code>/announcement Model paused for 30 min for maintenance</code>\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"📣 <b>ANNOUNCEMENT FROM ADMIN</b>\n"
            f"────────────────\n"
            f"{message_text}"
        )
        # Tell admin we're sending
        send_telegram(
            f"📢 Broadcasting to <b>{len(recipients)}</b> 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 <b>{success}</b> recipients.",
                chat_ids=[chat_id]
            )
        else:
            send_telegram(
                f"⚠️ Announcement sent to <b>{success}/{success + fail}</b> recipients.\n"
                f"<b>{fail}</b> 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:
        data = json.loads(p.read_text(encoding="utf-8"))
        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"⚠️ <b>[{pair}] STATE RESTORED AFTER REBOOT</b>\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_1h):
    """Dispatch to correct z-score engine per pair."""
    cfg = PAIR_CONFIG[pair]
    if cfg["signal_type"] == "beta_model":
        return _compute_eurusd_zscore(prices_1h, cfg)
    else:
        return _compute_spread_zscore(prices_1h, cfg)

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

# ─── 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):
    timer = threading.Timer(5.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"])
            send_telegram(
                f"📈 <b>[{pair}] ORDER PLACED</b>\n"
                f"Direction: <b>{'LONG' if direction==1 else 'SHORT'}</b>\n"
                f"Entry: {exec_price:.{dp}f}\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

        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 = []
            existing_tickets = {int(t.get("ticket", 0)) for t in existing}

            to_add = []
            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 int(pos_id) in existing_tickets:
                    total_already_synced += 1
                    continue

                # 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

                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   = sum(d.profit + d.swap + d.commission for d in pair_deals)
                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,
                })

            if to_add:
                combined = existing + 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)
                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:
            log.info(f"[RECONCILE] Back-filled {total_reconciled} missing trade(s) "
                     f"from MT5 history ({total_already_synced} already in sync)")
            try:
                send_telegram(
                    f"🔄 <b>TRADE HISTORY RECONCILED</b>\n"
                    f"────────────────\n"
                    f"Back-filled <b>{total_reconciled}</b> missing trade(s)\n"
                    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 in_session(pair):
    cfg  = PAIR_CONFIG[pair]
    hour = (datetime.now(timezone.utc).hour + 2) % 24  # UTC → EET approx
    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).

    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.

    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 = 6
        wait_seconds = 1.0
        pos_deals = []
        exit_deal = None

        # Retry loop: wait for exit deal to appear in MT5 history
        for attempt in range(1, max_attempts + 1):
            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"

        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)
        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

        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}")
        return round(net_pnl, 2), close_px, reason

    except Exception as e:
        log.warning(f"[{pair}] _get_closed_pnl error: {e}")
        return 0.0, None, "TP/SL"

def run_signal_check_pair(pair):
    """Hourly signal check for one pair."""
    cfg   = PAIR_CONFIG[pair]
    state = PAIR_STATE[pair]
    armed = PAIR_ARMED[pair]
    symbol= cfg["symbol"]
    dp    = cfg["dp"]
    log.info(f"[{pair}] ── Signal check ──")

    # Get 1H bars
    prices_1h = get_bars_1h(symbol, cfg["bars_1h"])
    if prices_1h is None:
        log.warning(f"[{pair}] No 1H bars — skipping")
        return

    # Compute z-score
    df = compute_zscore(pair, prices_1h)
    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 = _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  = "✅ PROFIT" if net_pnl > 0 else "❌ LOSS" if net_pnl < 0 else "🔵 BREAK EVEN"
            dir_str = "LONG" if entry_dir == 1 else "SHORT"
            close_str = f"{close_px:.{dp}f}" if close_px else "—"

            send_telegram(
                f"🏁 <b>[{pair}] POSITION CLOSED</b>\n"
                f"────────────────\n"
                f"Result: <b>{result}</b>\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: <b>{pnl_str}</b>"
            )
            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"]:
                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_position_market(pair, ticket_h, "time_limit")
                net_pnl_h, close_px_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"]
                pnl_str_h = f"+${net_pnl_h:,.2f}" if net_pnl_h >= 0 else f"-${abs(net_pnl_h):,.2f}"
                send_telegram(
                    f"⏱️ <b>[{pair}] POSITION CLOSED</b> — hold limit {cfg['hold_hours']}h\n"
                    f"Net P&L: <b>{pnl_str_h}</b>"
                )
                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"]):
                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_position_market(pair, ticket_z, "zscore_reversal")
                net_pnl_z, close_px_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)

                pnl_str_z = f"+${net_pnl_z:,.2f}" if net_pnl_z >= 0 else f"-${abs(net_pnl_z):,.2f}"
                send_telegram(
                    f"🔄 <b>[{pair}] Z-EXIT TRIGGERED</b>\n"
                    f"Z-score: {current_z:.3f}\n"
                    f"Net P&L: <b>{pnl_str_z}</b>")
                return

        log.info(f"[{pair}] Position open — holding | "
                 f"ticket={state['position_ticket']} | "
                 f"entry={state['entry_price']}")
        return

    # ── Check for new signal ─────────────────────────────────────────────
    if abs(current_z) < cfg["threshold"]:
        if not in_session(pair):
            cfg2 = PAIR_CONFIG[pair]
            hour_eet = (datetime.now(timezone.utc).hour + 2) % 24
            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 calculation from latest 1H bar
    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: return
        target = bc - cfg["fib"] * pull
    else:
        pull = bh - bc
        if pull <= 0.0001: 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"🎯 <b>[{pair}] SIGNAL DETECTED</b>\n"
        f"Z-score: <b>{current_z:.4f}</b> (threshold ±{cfg['threshold']})\n"
        f"Direction: <b>{'LONG' if direction==1 else 'SHORT'}</b>\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
        total_pnl = balance - STARTING_BALANCE   # authoritative total

        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)

        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)

        # Sanity: if sum of per-pair P&L differs substantially from total balance
        # delta, prefer the total (which is indisputable from account balance)
        sum_pair = sum(final_pair_pnl.values())
        if abs(sum_pair - total_pnl) > 5.0:
            # Scale per-pair proportionally to match total, preserving ratio
            if sum_pair != 0:
                scale = total_pnl / sum_pair
                final_pair_pnl = {k: round(v * scale, 2) for k, v in final_pair_pnl.items()}

        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"📅 <b>WEEK AHEAD — {pair_label}</b>",
            "────────────────",
        ]
        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"]) + " <b>" + d["day"] + "</b>  " +
                str(d["prob"]) + "%  " + _prob_bar(d["prob"]) + " " + star +
                "\n    └ " + evs
            )
        lines += [
            "────────────────",
            "⭐ Best day: <b>" + best["day"] + "</b> 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📅 <b>TODAY'S " + pair_label + " — " + td["day"] + "</b>\n"
            "────────────────\n" +
            _prob_emoji(td["prob"]) + " Signal probability: <b>" +
            str(td["prob"]) + "%</b>\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"<b>{header}</b>\n"
        f"────────────────\n"
        f"🟢 <b>EURUSD</b>\n"
        f"Z-score: <b>{eu_z_s}</b>  ({eu_pct})\n"
        f"Position: {_pair_position_str('EURUSD')}\n"
        f"Trades: {eu_n}  |  PnL: {_format_pnl(eu_tot)}\n"
        f"────────────────\n"
        f"🔵 <b>USDJPY</b>\n"
        f"Z-score: <b>{uj_z_s}</b>  ({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📅 <b>WEEK AHEAD — {label}</b>\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"📆 <b>WEEKLY OUTLOOK</b>\n"
        f"────────────────\n"
        f"🟢 <b>EURUSD</b>  z={eu_z_s}\n"
        f"🔵 <b>USDJPY</b>  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"🌅 <b>MORNING BRIEFING — {day_name}</b>\n"
        f"────────────────\n"
        f"🟢 <b>EURUSD</b>  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"🔵 <b>USDJPY</b>  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"⏰ <b>EVENT IN ~{int(round(mins))} MIN</b>\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 v2 — 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():
        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"⚠️ <b>[WATCHDOG] {pair} STATE FILE STALE</b>\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"⚠️ <b>[WATCHDOG] RATE DATA STALE</b>\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(
                            "⚠️ <b>[WATCHDOG] MT5 DISCONNECTED</b>\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"🔴 <b>[WATCHDOG] {pair} ORPHANED POSITION</b>\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")

    schedule.every().hour.at(":01").do(_safe_signal_check)
    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 — signal check every hour at :01")
    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)")

    # 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"),
        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"⚠️ <b>MONITOR WARNING</b>\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()