""" dashboard_server.py ==================== Reads the live_monitor.log file every 30 seconds and writes dashboard_data.json which the dashboard.html reads automatically. Run this alongside the live signal monitor: python src/execution/dashboard_server.py Then open dashboard.html in any browser. No internet required — runs entirely on your machine. """ import json import os import re import sys import time import subprocess import http.server import threading from pathlib import Path from datetime import datetime BASE_PATH = Path(__file__).resolve().parents[2] LOG_FILE = BASE_PATH / "data" / "logs" / "live_monitor_v2.log" DATA_FILE = BASE_PATH / "dashboard_data.json" PORT = 8765 # ────────────────────────────────────────────────────────────────────── # Regime classification (added v6 — 2026-05-14) # Computed on each refresh cycle from rate CSVs + trade history. # Lightweight (~50ms). Falls back gracefully if regime_live import fails # or regime_thresholds.json doesn't exist yet. # ────────────────────────────────────────────────────────────────────── _REGIME_AVAILABLE = False try: _src_path = str(BASE_PATH / "src" / "execution") if _src_path not in sys.path: sys.path.insert(0, _src_path) import regime_live # noqa: E402 _REGIME_AVAILABLE = True except Exception as _e: print(f"[regime] module not available: {_e} — regime cards will show Unknown") # ────────────────────────────────────────────────────────────────────── # Live-trade auto-refresh (Phase 4) # ────────────────────────────────────────────────────────────────────── # Periodically runs extract_live_trades.py + build_equity_curve.py to pull # fresh broker deals and rebuild the dashboard JSON. Runs in a background # thread; never blocks request handling. If it fails, server keeps serving # stale (but still valid) data. # # Set LIVE_REFRESH_ENABLED=False to disable entirely (server reverts to # pre-Phase-4 behaviour: dashboard JSON only updates when build script is # run manually). LIVE_REFRESH_ENABLED = True LIVE_REFRESH_INTERVAL_SEC = 3600 # max one refresh per hour LIVE_REFRESH_EXTRACT_TIMEOUT = 60 # seconds — MT5 connect + fetch LIVE_REFRESH_BUILD_TIMEOUT = 180 # seconds — pandas-heavy build EXTRACT_SCRIPT = BASE_PATH / "src" / "research" / "extract_live_trades.py" BUILD_SCRIPT = BASE_PATH / "src" / "research" / "build_equity_curve.py" _last_live_refresh: datetime | None = None _last_live_refresh_attempt: datetime | None = None # for failure-throttling _live_refresh_running: bool = False _live_refresh_lock = threading.Lock() def get_mt5_trade_history(): """ Read trade history from the pair-specific trade logs written by live_signal_monitor. Reads BOTH trade_history_eurusd.json and trade_history_usdjpy.json from data/logs/ and combines them. """ result = { "trade_count": 0, "pnl_total" : 0.0, "p1_pnl" : 0.0, "wins" : 0, "losses" : 0, "win_rate" : 0.0, } try: import json all_trades = [] for pair_file in ("trade_history_eurusd.json", "trade_history_usdjpy.json"): path = BASE_PATH / "data" / "logs" / pair_file if not path.exists(): continue try: pair_trades = json.loads(path.read_text(encoding="utf-8")) if isinstance(pair_trades, list): all_trades.extend(pair_trades) except Exception: continue if not all_trades: return result net = [t.get("net_pnl", 0) for t in all_trades] result["trade_count"] = len(net) result["pnl_total"] = round(sum(net), 2) result["p1_pnl"] = round(max(0, sum(net)), 2) result["wins"] = sum(1 for n in net if n > 0) result["losses"] = sum(1 for n in net if n <= 0) result["win_rate"] = round( result["wins"] / len(net) * 100, 1) if net else 0.0 result["trade_results"] = net except Exception: pass return result def _load_all_trades(): """Helper: load combined trades from both pair-specific files.""" import json all_trades = [] for pair_file in ("trade_history_eurusd.json", "trade_history_usdjpy.json"): path = BASE_PATH / "data" / "logs" / pair_file if not path.exists(): continue try: pair_trades = json.loads(path.read_text(encoding="utf-8")) if isinstance(pair_trades, list): all_trades.extend(pair_trades) except Exception: continue return all_trades # Starting account balance — FTMO $100k challenge STARTING_BALANCE = 100_000.0 def get_mt5_account_snapshot(): """ Authoritative source of truth for P&L and DD. Reads the account snapshot file written by live_signal_monitor_v2 every hour (monitor has MT5 connected; server can't share that connection since MT5 terminal only allows one). If snapshot exists and is fresh, returns its contents. If missing or stale, returns None and callers fall back to JSON trade history. Snapshot fields: pnl_total, balance, equity, floating_pnl, daily_loss, trade_count, pair_pnl, pair_trades """ try: import json as _json snap_path = BASE_PATH / "data" / "logs" / "account_snapshot.json" if not snap_path.exists(): return None snap = _json.loads(snap_path.read_text(encoding="utf-8")) # Sanity check that it's recent — if snapshot is > 2 hours old the # monitor is probably not running, fall back to JSON. if "snapshot_at" in snap: try: snap_time = datetime.fromisoformat(snap["snapshot_at"].replace("Z", "+00:00")) age_hours = (datetime.now(snap_time.tzinfo) - snap_time).total_seconds() / 3600 if age_hours > 2.0: print(f"[Account snapshot] stale ({age_hours:.1f}h old), falling back") return None except Exception: pass # Normalize into the shape the caller expects return { "source" : "monitor_snapshot", "pnl_total" : snap.get("pnl_total", 0.0), "current_balance": snap.get("balance", 0.0), "current_equity" : snap.get("equity", 0.0), "floating_pnl" : snap.get("floating_pnl", 0.0), "daily_loss" : snap.get("daily_loss", 0.0), "trade_count" : snap.get("trade_count", 0), "pair_pnl" : snap.get("pair_pnl", {"EURUSD": 0.0, "USDJPY": 0.0}), "pair_trades" : snap.get("pair_trades", {"EURUSD": 0, "USDJPY": 0}), } except Exception as e: print(f"[Account snapshot] read failed: {e}") return None # ── Weekly Signal Outlook ───────────────────────────────────────────────────── TE_API_KEY = "2d9e4c37c81f1b4c970d7b1f8d7d382ff450312cd0adf57117f28d5708e6e38e" # FinanceFlow API key FRED_API_KEY = "b4e39ea4507e1ceca1a389cfacee8a00" TE_US_SCORES = { "core inflation rate month": 85, "core inflation rate year": 85, "inflation rate month": 85, "inflation rate year": 80, "consumer price index": 85, "core pce": 80, "pce price index": 80, "personal spending": 75, "personal income": 75, "nonfarm payrolls": 75, "non-farm payrolls": 75, "adp employment change": 55, # monthly ADP only "unemployment rate": 70, "fomc meeting minutes": 75, "fed minutes": 75, "fomc rate": 95, "federal funds rate": 95, "gdp growth rate": 65, "gdp": 60, "retail sales month": 45, "ism manufacturing pmi": 45, "ism services pmi": 45, "ism non-manufacturing": 45, "durable goods orders": 35, "consumer confidence": 30, "michigan consumer": 35, "jolts job openings": 40, "producer price index": 40, "10-year note auction": 45, "30-year bond auction": 40, "treasury": 30, } TE_DE_EU_SCORES = { "ecb interest rate": 90, "ecb rate decision": 90, "ecb monetary policy": 85, "ecb president": 70, "lagarde": 70, "ecb minutes": 65, "inflation rate month": 65, "inflation rate year": 60, "consumer price": 65, "zew economic": 50, "ifo business": 50, "gdp growth": 60, "gdp": 55, "industrial production": 40, "factory orders": 40, "unemployment rate": 40, "manufacturing pmi": 40, "services pmi": 40, "retail sales": 35, } # Events to explicitly score as 0 (noise — no meaningful 2Y spread impact) TE_EXCLUSIONS = [ "mortgage", "redbook", "weekly earnings", "weekly jobless", "adp employment change weekly", "adp weekly", "3-month treasury", "6-month treasury", "4-week", "8-week", "bill auction", "t-bill", "bubill", "non-monetary", "speaks", "speech", "interview", "testimony", "index of coincident", "index of leading", "wholesale inventories", "building permits", "housing starts", "existing home", "pending home", "construction spending", "natural gas", "crude oil", "eia", "api weekly", "milk", "grain", ] # ECB 2026 rate decision dates (verified from ECB/Morningstar/skilling.com Apr 2026) # Announcements typically at 14:15 CET, press conference 14:45 CET ECB_MEETINGS_2026 = [ "2026-03-19","2026-04-30","2026-06-11","2026-07-23", "2026-09-10","2026-10-29","2026-12-17" ] # FOMC 2026 rate decision dates (verified from federalreserve.gov) # Second day of each two-day meeting, statement at 14:00 ET FOMC_MEETINGS_2026 = [ "2026-01-28","2026-03-18","2026-04-29","2026-06-17", "2026-07-29","2026-09-16","2026-10-28","2026-12-09" ] FOMC_MINUTES_2026 = [ (datetime.strptime(d,"%Y-%m-%d")+__import__("datetime").timedelta(weeks=3)).strftime("%Y-%m-%d") for d in FOMC_MEETINGS_2026 ] _outlook_cache = {"data": None, "fetched": None} def _score_te_event(event): country = event.get("Country","").lower() category = event.get("Category","").lower() importance = int(event.get("Importance", 1)) is_us = "united states" in country is_de = "germany" in country is_eu = "euro" in country is_jp = "japan" in country if not (is_us or is_de or is_eu or is_jp): return 0 # Exclude noise events regardless of importance if any(excl in category for excl in TE_EXCLUSIONS): return 0 if is_jp: # Japan event scoring JP_SCORES = { "boj": 90, "bank of japan": 90, "interest rate": 90, "monetary policy": 85, "inflation": 75, "cpi": 75, "gdp": 70, "trade balance": 65, "current account":60, "employment": 60, "industrial": 50, "retail": 45, "tankan": 70, "pmi": 45, } base = max((v for k,v in JP_SCORES.items() if k in category), default=0) if base == 0 and importance == 3: base = 40 elif base == 0 and importance == 2: base = 20 return base scores = TE_US_SCORES if is_us else TE_DE_EU_SCORES base = max((v for k,v in scores.items() if k in category), default=0) # Only use importance fallback for Importance=3 (High) events not matched if base == 0 and importance == 3: base = 35 # Cap ECB non-policy events if is_eu and base > 0 and "non-monetary" in category: base = 5 return base def _fetch_te_calendar(start, end): """Fetch calendar from FinanceFlow API.""" try: import requests as _req import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) except ImportError: print("[WARNING] requests not installed — run: python -m pip install requests") return [] events = [] # FinanceFlow API: /financial-calendar?api_key=KEY&country=NAME&date_from=X&date_to=Y # Response fields: country, report_name, actual, previous, consensus, # economicImpact (High/Moderate/Low), datetime for country_name, label in [("United States","US"),("Germany","DE"),("Euro Area","EU"),("Japan","JP")]: try: url = "https://financeflowapi.com/api/v1/financial-calendar" resp = _req.get(url, params={ "api_key": TE_API_KEY, "country": country_name, "date_from": start, "date_to": end, }, timeout=10, verify=False) if resp.status_code == 200: data = resp.json() if data.get("success") and isinstance(data.get("data"), list): raw = data["data"] print(f"[INFO] FinanceFlow {label}: {len(raw)} events ({start} → {end})") for e in raw: # Convert UTC → UK time (BST=UTC+1 Apr-Oct, GMT=UTC+0 Nov-Mar) dt_str = e.get("datetime","") uk_time = "" if len(dt_str) >= 16: try: dt_utc = datetime.strptime(dt_str[:16], "%Y-%m-%d %H:%M") offset = 1 if 3 < dt_utc.month < 11 else 0 uk_h = (dt_utc.hour + offset) % 24 uk_time = f"{uk_h:02d}:{dt_utc.minute:02d}" except Exception: uk_time = dt_str[11:16] events.append({ "Country": e.get("country", country_name), "Event": e.get("report_name",""), "Category": e.get("report_name",""), "Date": dt_str[:10], "Time": uk_time, "Importance": {"High":3,"Moderate":2,"Low":1}.get( e.get("economicImpact","Low"), 1), "Forecast": e.get("consensus",""), "Previous": e.get("previous",""), "Actual": e.get("actual",""), }) else: print(f"[WARNING] FinanceFlow {label}: {str(data)[:100]}") else: print(f"[WARNING] FinanceFlow {label} HTTP {resp.status_code}: {resp.text[:80]}") except Exception as e: print(f"[WARNING] FinanceFlow {label} failed: {e}") print(f"[INFO] FinanceFlow total: {len(events)} events fetched") return events def get_weekly_outlook(current_z=0.0): """Build weekly signal outlook. Caches 6 hours.""" now = datetime.now() cache = _outlook_cache # Calculate monday FIRST so cache check can compare week day = now.weekday() if day >= 5: # Saturday=5, Sunday=6 — show NEXT week monday = now + __import__("datetime").timedelta(days=(7 - day)) else: monday = now - __import__("datetime").timedelta(days=day) cache_age = (now - cache["fetched"]).total_seconds() if cache["fetched"] else 99999 #print(f"Signal outlook: cache_age={cache_age:.0f}s") # Invalidate if week has rolled over if cache.get("week_start") and cache["week_start"] != monday.strftime("%Y-%m-%d"): cache["fetched"] = None #print("Signal outlook: new week detected — forcing refresh") if cache["fetched"] and cache_age < 3600: # 1 hour (actuals appear faster) return _recalc_probs(cache["data"], current_z) start = monday.strftime("%Y-%m-%d") end = (monday + __import__("datetime").timedelta(days=4)).strftime("%Y-%m-%d") week_events = { (monday + __import__("datetime").timedelta(days=i)).strftime("%Y-%m-%d"): [] for i in range(5) } # Primary: Trading Economics API te_events = _fetch_te_calendar_cached(start, end) te_used = len(te_events) > 0 for e in te_events: date = e.get("Date","")[:10] country = e.get("Country","").strip().lower() # EU outlook: exclude Japan events — only US/DE/EU drivers if "japan" in country: continue score = _score_te_event(e) if date in week_events and score > 0: week_events[date].append({ "event": e.get("Event", e.get("Category","")), "country": e.get("Country",""), "score": score, "time": e.get("Time",""), "forecast": e.get("Forecast",""), "actual": e.get("Actual",""), "previous": e.get("Previous",""), "source": "FF", }) # Always add hardcoded ECB/FOMC (most critical, confirmed dates) for date in week_events: existing = [ev["event"] for ev in week_events[date]] if date in ECB_MEETINGS_2026 and not any("ECB" in ev and "Rate" in ev for ev in existing): week_events[date].append({"event":"ECB Rate Decision","country":"Euro Area","score":90,"source":"hardcoded"}) if date in FOMC_MEETINGS_2026 and not any("FOMC" in ev and "Rate" in ev for ev in existing): week_events[date].append({"event":"FOMC Rate Decision","country":"United States","score":95,"source":"hardcoded"}) if date in FOMC_MINUTES_2026 and not any("Minutes" in ev for ev in existing): week_events[date].append({"event":"FOMC Meeting Minutes","country":"United States","score":75,"source":"hardcoded"}) # Fallback: FRED if TE returned nothing if not te_used: print("TE unavailable — falling back to FRED") try: import urllib.request as _u url = (f"https://api.stlouisfed.org/fred/releases/dates" f"?realtime_start={start}&realtime_end={end}" f"&include_release_dates_with_no_data=false" f"&api_key={FRED_API_KEY}&file_type=json") data = json.loads(_u.urlopen(url, timeout=8).read()) FRED_IDS = { 21:("Consumer Price Index (CPI)",85), 50:("Personal Income & PCE",80), 51:("Personal Income & Outlays",80), 10:("Employment Situation (NFP)",75), 46:("Gross Domestic Product",60), 14:("Retail Sales",45), 101:("ISM Manufacturing PMI",45), 102:("ISM Services PMI",45), 52:("Durable Goods Orders",35), 55:("PPI",40), } for rd in data.get("release_dates",[]): rid, date = rd.get("release_id"), rd.get("date","")[:10] if date in week_events and rid in FRED_IDS: n, s = FRED_IDS[rid] week_events[date].append({"event":n,"country":"United States","score":s,"source":"FRED"}) except Exception as fe: print(f"[WARNING] FRED fallback failed: {fe}") result = [] for i in range(5): date = (monday + __import__("datetime").timedelta(days=i)).strftime("%Y-%m-%d") events = sorted(week_events.get(date,[]), key=lambda x: x["score"], reverse=True) result.append({ "date": date, "day": ["Mon","Tue","Wed","Thu","Fri"][i], "events": events[:3], "top_event": events[0]["event"] if events else "No key events", "source": "FF" if te_used else "FRED", }) cache["data"] = result cache["fetched"] = now cache["week_start"] = monday.strftime("%Y-%m-%d") return _recalc_probs(result, current_z) def _recalc_probs(days_data, current_z): """Recalculate signal probabilities with current z-score.""" THRESHOLD = 2.75 out = [] for d in days_data: events = d.get("events", []) if not events: prob = 3 else: top = events[0]["score"] sec = events[1]["score"] * 0.4 if len(events) > 1 else 0 ep = min((top + sec) / 100 * 50, 50) dist = THRESHOLD - abs(current_z) zp = (30 if dist <= 0 else 25 if dist < 0.5 else 20 if dist < 1.0 else 12 if dist < 1.5 else 6 if dist < 2.0 else 3) bonus = sum(5 for e in events[:2] if any(k in e["event"].upper() for k in ["CPI","PCE","FOMC","NFP", "ECB RATE","MINUTES"])) prob = min(int(ep + zp + bonus), 85) out.append({**d, "prob": prob, "signal_probability": prob}) return out _uj_outlook_cache = {"data": None, "fetched": None, "week_start": None} _calendar_cache = {"events": None, "fetched": None, "start": None, "end": None} def _fetch_te_calendar_cached(start, end): """Wrapper that caches calendar results for 6 hours to avoid double API calls.""" now = datetime.now() cache = _calendar_cache if (cache["events"] is not None and cache["start"] == start and cache["end"] == end and cache["fetched"] and (now - cache["fetched"]).total_seconds() < 3600): return cache["events"] events = _fetch_te_calendar(start, end) cache["events"] = events cache["fetched"] = now cache["start"] = start cache["end"] = end return events def get_usdjpy_weekly_outlook(current_z=0.0): """Weekly outlook for USDJPY — US + Japan events. Caches 6 hours.""" import datetime as _dt now = _dt.datetime.now() cache = _uj_outlook_cache day = now.weekday() if day >= 5: monday = now + _dt.timedelta(days=(7 - day)) else: monday = now - _dt.timedelta(days=day) cache_age = (now - cache["fetched"]).total_seconds() if cache["fetched"] else 99999 if cache.get("week_start") and cache["week_start"] != monday.strftime("%Y-%m-%d"): cache["fetched"] = None if cache["fetched"] and cache_age < 3600: # 1 hour return _recalc_probs_uj(cache["data"], current_z) start = monday.strftime("%Y-%m-%d") end = (monday + _dt.timedelta(days=4)).strftime("%Y-%m-%d") week_events = { (monday + _dt.timedelta(days=i)).strftime("%Y-%m-%d"): [] for i in range(5) } # Fetch events — filter for US and Japan te_events = _fetch_te_calendar_cached(start, end) for e in te_events: date = e.get("Date", "")[:10] country = e.get("Country", "").strip().lower() score = _score_te_event(e) if date in week_events and score > 0: if any(c in country for c in ["united states", "japan"]): week_events[date].append({ "event": e.get("Event", e.get("Category", "")), "country": e.get("Country", ""), "score": score, "time": e.get("Time", ""), "forecast": e.get("Forecast", ""), "actual": e.get("Actual", ""), "previous": e.get("Previous", ""), "source": "FF", }) # Add hardcoded FOMC dates for date in week_events: existing = [ev["event"] for ev in week_events[date]] if date in FOMC_MEETINGS_2026 and not any("FOMC" in ev and "Rate" in ev for ev in existing): week_events[date].append({"event": "FOMC Rate Decision", "country": "United States", "score": 95, "source": "hardcoded"}) if date in FOMC_MINUTES_2026 and not any("Minutes" in ev for ev in existing): week_events[date].append({"event": "FOMC Meeting Minutes", "country": "United States", "score": 75, "source": "hardcoded"}) result = [] for i in range(5): date = (monday + _dt.timedelta(days=i)).strftime("%Y-%m-%d") events = sorted(week_events.get(date, []), key=lambda x: x["score"], reverse=True) result.append({ "date": date, "day": ["Mon","Tue","Wed","Thu","Fri"][i], "events": events[:3], "top_event": events[0]["event"] if events else "No key events", }) cache["data"] = result cache["fetched"] = now cache["week_start"] = monday.strftime("%Y-%m-%d") return _recalc_probs_uj(result, current_z) def _recalc_probs_uj(days_data, current_z): """Recalculate USDJPY signal probabilities.""" THRESHOLD = 2.0 out = [] for d in days_data: events = d.get("events", []) if not events: prob = 5 else: top = events[0]["score"] sec = events[1]["score"] * 0.4 if len(events) > 1 else 0 ep = min((top + sec) / 100 * 45, 45) dist = THRESHOLD - abs(current_z) zp = (28 if dist <= 0 else 22 if dist < 0.3 else 15 if dist < 0.7 else 8 if dist < 1.2 else 4) bonus = sum(5 for e in events[:2] if any(k in e["event"].upper() for k in ["BOJ","FOMC","CPI","NFP","INFLATION", "MONETARY","RATE DECISION"])) prob = min(int(ep + zp + bonus), 80) out.append({**d, "prob": prob, "signal_probability": prob}) return out def parse_log(): """Parse live_monitor.log and extract current state.""" if not LOG_FILE.exists(): return None try: # errors="ignore" handles Windows cp1252 log files containing ± symbol # (Python logging on Windows may write ±2.75 as cp1252 byte 0xB1 which # is invalid UTF-8, causing silent return None and empty zscore_history) lines = LOG_FILE.read_text(encoding="utf-8", errors="ignore").splitlines() except Exception: try: lines = LOG_FILE.read_text(encoding="cp1252", errors="ignore").splitlines() except Exception: return None data = { "timestamp" : datetime.now().isoformat(), "zscore" : 0.0, "position" : "None", "rate_status" : "Updated", "trade_count" : 0, "pnl_total" : 0.0, "p1_pnl" : 0.0, "wins" : 0, "losses" : 0, "win_rate" : 0.0, } # Read z-scores from state files — most reliable source try: import json as _json eu_state_path = BASE_PATH / "data" / "logs" / "state_eurusd.json" if eu_state_path.exists(): eu_state = _json.loads(eu_state_path.read_text(encoding="utf-8-sig")) eu_z = eu_state.get("last_zscore") if eu_z is not None: data["zscore"] = float(eu_z) data["eu_zscore"] = float(eu_z) except Exception: pass try: import json as _json uj_state_path = BASE_PATH / "data" / "logs" / "state_usdjpy.json" if uj_state_path.exists(): uj_state = _json.loads(uj_state_path.read_text(encoding="utf-8-sig")) uj_z = uj_state.get("last_zscore") if uj_z is not None: data["uj_zscore"] = float(uj_z) except Exception: pass # uj_zscore state read complete # ── Z-score history — parse both log files (old + new format) ─────────── try: eu_history = [] uj_history = [] cutoff = (datetime.now() - __import__('datetime').timedelta(days=60)).isoformat()[:10] # Combine old log (live_monitor.log) + new log (live_monitor_v2.log) all_lines = list(lines) old_log = BASE_PATH / "data" / "logs" / "live_monitor.log" if old_log.exists(): try: all_lines = old_log.read_text(encoding="utf-8", errors="ignore").splitlines() + all_lines except Exception: pass for line in all_lines: # Rate status (check all lines regardless of date) if "Rate data updated successfully" in line: data["rate_status"] = "Updated" elif "Rate update failed" in line: if data.get("rate_status") not in ("Updated",): data["rate_status"] = "Failed" m_t = re.search(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2})", line) if not m_t: continue ts = m_t.group(1) if ts[:10] < cutoff: continue # New format: "2026-04-12 22:33:48,107 [INFO] [EURUSD] Z-score: 0.5998" if "] [EURUSD] Z-score:" in line: m_z = re.search(r"Z-score:\s*([-\d.]+)", line) if m_z: eu_history.append({"t": ts, "z": round(float(m_z.group(1)), 3)}) elif "] [USDJPY] Z-score:" in line: m_z = re.search(r"Z-score:\s*([-\d.]+)", line) if m_z: uj_history.append({"t": ts, "z": round(float(m_z.group(1)), 3)}) # Old format: "Current z-score: 0.599" (EURUSD only, pre-v2 monitor) elif "Current z-score:" in line: m_z = re.search(r"Current z-score:\s*([-\d.]+)", line) if m_z: eu_history.append({"t": ts, "z": round(float(m_z.group(1)), 3)}) # Deduplicate by timestamp eu_history = list({e["t"]: e for e in eu_history}.values()) uj_history = list({e["t"]: e for e in uj_history}.values()) eu_history.sort(key=lambda x: x["t"]) uj_history.sort(key=lambda x: x["t"]) data["zscore_history"] = eu_history[-500:] data["uj_zscore_history"] = uj_history[-500:] except Exception: data["zscore_history"] = [] data["uj_zscore_history"] = [] # ── Live 2026 heatmap row from trade history ────────────────────────────── try: import calendar trades = _load_all_trades() if trades: monthly_2026 = [0] * 12 for t in trades: closed = t.get("closed_at", "")[:7] # YYYY-MM if closed.startswith("2026-"): month_idx = int(closed[5:7]) - 1 if 0 <= month_idx < 12: monthly_2026[month_idx] += round(t.get("net_pnl", 0)) # Future months as None now_month = datetime.now().month for i in range(now_month, 12): monthly_2026[i] = None data["heatmap_2026"] = monthly_2026 except Exception: data["heatmap_2026"] = None # Override position from state file — more reliable than log parsing try: import json state_path = BASE_PATH / "data" / "logs" / "state_eurusd.json" if state_path.exists(): state = json.loads(state_path.read_text(encoding="utf-8-sig")) if state.get("position_open"): entry = state.get("entry_price", "") direction = "LONG" if state.get("position_signal") == 1 else "SHORT" data["position"] = f"{direction} @ {entry}" else: # Check armed state from state file armed = state.get("_armed", {}) if armed.get("active", False): target = armed.get("target_price", "") direction = "LONG" if armed.get("direction") == 1 else "SHORT" data["position"] = f"Armed {direction} → {target:.5f}" if target else "Armed" else: data["position"] = "None" except Exception: pass # ── USDJPY position from state file (mirrors EURUSD block above) ──────── # Dashboard HTML reads `data.uj_position` (with fallback to # `data.usdjpy.position`). Without this read, the UJ position card # always showed "None" regardless of actual broker position. try: import json as _json_uj uj_state_path = BASE_PATH / "data" / "logs" / "state_usdjpy.json" if uj_state_path.exists(): uj_state = _json_uj.loads(uj_state_path.read_text(encoding="utf-8-sig")) if uj_state.get("position_open"): uj_entry = uj_state.get("entry_price", "") uj_direction = "LONG" if uj_state.get("position_signal") == 1 else "SHORT" data["uj_position"] = f"{uj_direction} @ {uj_entry}" else: uj_armed = uj_state.get("_armed", {}) if uj_armed.get("active", False): uj_target = uj_armed.get("target_price", "") uj_direction = "LONG" if uj_armed.get("direction") == 1 else "SHORT" data["uj_position"] = (f"Armed {uj_direction} → {uj_target:.3f}" if uj_target else "Armed") else: data["uj_position"] = "None" except Exception: pass # Pull trade stats from trade history file (JSON fallback) mt5_stats = get_mt5_trade_history() data.update(mt5_stats) # Try MT5 live snapshot — authoritative source of truth when available. # This overrides the JSON-based stats because MT5 reflects actual account # state (including swap/commission), while the JSON can drift if the # monitor missed logging a trade or double-wrote one. snap = get_mt5_account_snapshot() if snap is not None: data["pnl_total"] = snap["pnl_total"] data["daily_loss"] = snap["daily_loss"] # Use higher count to avoid undercounting if MT5 missed a position data["trade_count"] = max(snap["trade_count"], mt5_stats.get("trade_count", 0)) data["current_balance"] = snap["current_balance"] data["current_equity"] = snap["current_equity"] data["floating_pnl"] = snap["floating_pnl"] # Phase 1 progress uses live balance delta — always accurate data["p1_pnl"] = max(0.0, snap["pnl_total"]) data["pnl_source"] = "mt5_live" else: data["pnl_source"] = "json_fallback" # Build per-pair trade stats for HTML (reads d.usdjpy_trades and d.portfolio) try: all_trades = _load_all_trades() eu_trades_raw = [t for t in all_trades if t.get("pair","").upper() == "EURUSD"] uj_trades_raw = [t for t in all_trades if t.get("pair","").upper() == "USDJPY"] def _trade_stats(trades): if not trades: return {"trade_count":0,"pnl_total":0.0,"wins":0,"losses":0,"win_rate":0.0} net = [t.get("net_pnl",0) for t in trades] wins = sum(1 for n in net if n > 0) return {"trade_count": len(net), "pnl_total": round(sum(net),2), "wins": wins, "losses": len(net)-wins, "win_rate": round(wins/len(net)*100,1) if net else 0.0} uj_stats = _trade_stats(uj_trades_raw) eu_stats = _trade_stats(eu_trades_raw) # If we have a live MT5 snapshot, use its per-pair P&L (balance-accurate, # includes swap/commission). For trade COUNT use max(JSON, MT5) — MT5 can # occasionally miss a position in its deal grouping (weekend holds, gaps). # The JSON is the monitor's authoritative record of what it executed. if snap is not None: eu_stats["pnl_total"] = snap["pair_pnl"]["EURUSD"] uj_stats["pnl_total"] = snap["pair_pnl"]["USDJPY"] # Use higher of the two counts to avoid undercounting eu_stats["trade_count"] = max(eu_stats["trade_count"], snap["pair_trades"]["EURUSD"]) uj_stats["trade_count"] = max(uj_stats["trade_count"], snap["pair_trades"]["USDJPY"]) data["usdjpy_trades"] = uj_stats data["eurusd_trades"] = eu_stats # NEW — was previously top-level which # was actually showing COMBINED counts # in the EU card (real bug found # 2026-05-23: EU showed 11 trades when # actual EU=6, combined=11). data["portfolio"] = { "eu_pnl": eu_stats["pnl_total"], "uj_pnl": uj_stats["pnl_total"], "combined_pnl": round(eu_stats["pnl_total"] + uj_stats["pnl_total"], 2), "total_trades": eu_stats["trade_count"] + uj_stats["trade_count"], } except Exception as _e: pass # Weekly signal outlook — EURUSD and USDJPY try: eu_z = float(data.get("eu_zscore") or data.get("zscore") or 0.0) uj_z = float(data.get("uj_zscore") or 0.0) data["signal_outlook"] = get_weekly_outlook(eu_z) data["usdjpy_outlook"] = get_usdjpy_weekly_outlook(uj_z) except Exception as _e: print("[WARNING]", f"Signal outlook failed: {_e}") data["signal_outlook"] = [] data["usdjpy_outlook"] = [] # ── Build uj sub-object — HTML reads from d.uj.* ───────────────────────── try: import csv as _csv rates_dir = BASE_PATH / "data" / "raw" / "rates" uj_jp2y = None uj_jp2y_ser = [] uj_spread = None uj_spr_hist = [] jp2y_path = rates_dir / "jp2y.csv" us2y_path = rates_dir / "us2y.csv" def _safe_rows(path): """Read CSV rows, skipping header and any non-numeric value rows.""" result = [] for r in _csv.reader(open(path)): if len(r) < 2: continue try: float(r[1]) # will raise if header or '.' result.append(r) except (ValueError, TypeError): continue return result if jp2y_path.exists(): rows = _safe_rows(jp2y_path) if rows: uj_jp2y = round(float(rows[-1][1]), 3) uj_jp2y_ser = [{"date": r[0], "jp2y": round(float(r[1]),3)} for r in rows[-60:]] if us2y_path.exists() and uj_jp2y is not None: us_rows = _safe_rows(us2y_path) if us_rows: us_last = round(float(us_rows[-1][1]), 3) uj_spread = round(us_last - uj_jp2y, 3) us_dict = {r[0]: float(r[1]) for r in us_rows} uj_spr_hist = [ {"date": e["date"], "spread": round(us_dict[e["date"]] - e["jp2y"], 3)} for e in uj_jp2y_ser if e["date"] in us_dict ] data["usdjpy"] = { "current_z": data.get("uj_zscore"), "z_history": data.get("uj_zscore_history", []), "jp2y": uj_jp2y, "jp2y_series": uj_jp2y_ser, "current_spread": uj_spread, "spread_history": uj_spr_hist, } except Exception as _e: print("[WARNING] uj block:", _e) data["usdjpy"] = { "current_z": data.get("uj_zscore"), "z_history": data.get("uj_zscore_history", []), } return data def fetch_yield_data(): """ Fetches yield data for dashboard display. US 2Y and DE 2Y: read from local CSV files maintained by live_signal_monitor. SOFR and FFR: fetched from FRED API and cached locally. No yfinance proxies — those return wrong rates (^IRX is 13-week T-bill, not 2Y). """ yields = {} rates_dir = BASE_PATH / "data" / "raw" / "rates" sofr_path = rates_dir / "sofr.csv" ffr_path = rates_dir / "ffr.csv" # ── Read local CSV files (maintained by live_signal_monitor) ───────────── try: import csv HISTORY_DAYS = 60 # show last 60 trading days on charts # US 2Y us2y_path = rates_dir / "us2y.csv" if us2y_path.exists(): rows = [r for r in csv.reader(open(us2y_path)) if len(r) == 2 and r[1] not in ('.', '', 'VALUE')] if rows: yields["us2y_daily"] = float(rows[-1][1]) yields["us2y_date"] = rows[-1][0] hist = rows[-HISTORY_DAYS:] yields["us2y_series"] = [round(float(r[1]), 3) for r in hist] yields["us2y_times"] = [r[0] for r in hist] # DE 2Y de2y_path = rates_dir / "de2y.csv" if de2y_path.exists(): rows = [r for r in csv.reader(open(de2y_path)) if len(r) == 2 and r[1] not in ('.', '', 'VALUE')] if rows: yields["de2y_daily"] = float(rows[-1][1]) yields["de2y_date"] = rows[-1][0] hist = rows[-HISTORY_DAYS:] yields["de2y_series"] = [round(float(r[1]), 3) for r in hist] yields["de2y_times"] = [r[0] for r in hist] # JP 2Y jp2y_path = rates_dir / "jp2y.csv" if jp2y_path.exists(): rows = [r for r in csv.reader(open(jp2y_path)) if len(r) == 2 and r[1] not in ('.', '', 'VALUE')] if rows: yields["jp2y_daily"] = float(rows[-1][1]) yields["jp2y_date"] = rows[-1][0] hist = rows[-HISTORY_DAYS:] yields["jp2y_series"] = [{"date": r[0], "jp2y": round(float(r[1]), 3)} for r in hist] except Exception as e: yields["csv_error"] = str(e)[:100] # ── SOFR + FFR from FRED API (api.stlouisfed.org) ───────────────────────── # Switched from fred.stlouisfed.org/graph/fredgraph.csv to the official API # because the graph endpoint blocks/throttles the VPS IP. The api.* host # works fine. Same caching: writes a 2-column CSV (date,value) for the # downstream chart-builder block. Uses the module-level FRED_API_KEY. def _fred_api_fetch(series_id: str, csv_path): """ Fetch a FRED series via the official API and persist as 2-column CSV (date, value) in chronological order. Returns (latest_value, latest_date) or (None, None) on failure (no logging spam — caller decides how loud). """ try: import requests as _rq url = "https://api.stlouisfed.org/fred/series/observations" params = { "series_id": series_id, "api_key": FRED_API_KEY, "file_type": "json", # Last ~6 months is plenty for the 60-row chart and gives some # buffer for weekends/holidays. SOFR/FFR are daily-banking-day series. "limit": 180, "sort_order": "desc", # newest first; we'll reverse before saving } r = _rq.get(url, params=params, timeout=15) if r.status_code != 200: print(f" [fred] {series_id} HTTP {r.status_code}: {r.text[:120]}") return (None, None) obs = r.json().get("observations", []) if not obs: print(f" [fred] {series_id} returned no observations") return (None, None) # Filter out FRED's "no value" placeholder (".") and reverse to ascending date clean = [(o["date"], o["value"]) for o in obs if o.get("value") not in (".", "", None)] clean.reverse() if not clean: return (None, None) # Write CSV (no header — matches existing format the chart code expects) with csv_path.open("w", encoding="utf-8", newline="") as f: for d, v in clean: f.write(f"{d},{v}\n") latest_date, latest_val = clean[-1] return (float(latest_val), latest_date) except Exception as _e: # Surface failure so we'd notice if FRED ever changes/breaks again, # but don't crash the dashboard request. print(f" [fred] {series_id} fetch failed: {_e}") return (None, None) # SOFR — Secured Overnight Financing Rate sofr_val, sofr_date = _fred_api_fetch("SOFR", sofr_path) if sofr_val is not None: yields["sofr_daily"] = sofr_val yields["sofr_date"] = sofr_date else: # Fallback: read whatever is in the cached CSV (may be stale) try: import csv if sofr_path.exists(): rows = [r for r in csv.reader(open(sofr_path)) if len(r) == 2 and r[1] not in (".", "", "VALUE")] if rows: yields["sofr_daily"] = float(rows[-1][1]) yields["sofr_date"] = rows[-1][0] except Exception: pass # EFFR — Effective Federal Funds Rate ffr_val, ffr_date = _fred_api_fetch("EFFR", ffr_path) if ffr_val is not None: yields["ffr_daily"] = ffr_val else: try: import csv if ffr_path.exists(): rows = [r for r in csv.reader(open(ffr_path)) if len(r) == 2 and r[1] not in (".", "", "VALUE")] if rows: yields["ffr_daily"] = float(rows[-1][1]) except Exception: pass # ── Compute spreads ─────────────────────────────────────────────────────── if "us2y_daily" in yields and "de2y_daily" in yields: yields["spread_daily"] = round( yields["us2y_daily"] - yields["de2y_daily"], 3) if "sofr_daily" in yields and "ffr_daily" in yields: yields["sofr_ffr_spread_bp"] = round( (yields["sofr_daily"] - yields["ffr_daily"]) * 100, 1) if "us2y_daily" in yields and "sofr_daily" in yields: yields["sofr_vs_2y"] = round( yields["us2y_daily"] - yields["sofr_daily"], 3) # Build SOFR series for chart (HTML expects yields.sofr_series + yields.sofr_times) try: import csv as _sc from datetime import timedelta as _td sofr_path2 = BASE_PATH / "data" / "raw" / "rates" / "sofr.csv" if sofr_path2.exists(): sofr_rows = [] for r in _sc.reader(open(sofr_path2)): if len(r) < 2: continue try: float(r[1]); sofr_rows.append(r) except (ValueError, TypeError): continue if len(sofr_rows) > 1: hist = sofr_rows[-60:] yields["sofr_series"] = [round(float(r[1]),3) for r in hist] yields["sofr_times"] = [r[0] for r in hist] # Fallback: flat series from single value so chart renders if "sofr_series" not in yields and "sofr_daily" in yields: val = yields["sofr_daily"] from datetime import datetime as _dt2 dates = [(_dt2.now() - _td(days=i)).strftime("%Y-%m-%d") for i in range(30, -1, -1)] yields["sofr_series"] = [val] * len(dates) yields["sofr_times"] = dates except Exception: pass # US-JP 2Y spread series for USDJPY chart if "us2y_daily" in yields and "jp2y_daily" in yields: yields["usjp_spread_daily"] = round( yields["us2y_daily"] - yields["jp2y_daily"], 3) if "us2y_series" in yields and "jp2y_series" in yields: jp_dict = {r["date"]: r["jp2y"] for r in yields["jp2y_series"]} yields["usjp_spread_series"] = [ {"date": d, "spread": round(v - jp_dict[d], 3)} for d, v in zip(yields["us2y_times"], yields["us2y_series"]) if d in jp_dict ] return yields def write_data(): """Parse log and write JSON file for dashboard. VPS detection: state files exist on VPS. Home machine just serves existing JSON. Yield/outlook fetches run regardless of whether log file exists. """ eu_state = BASE_PATH / "data" / "logs" / "state_eurusd.json" uj_state = BASE_PATH / "data" / "logs" / "state_usdjpy.json" is_vps = eu_state.exists() or uj_state.exists() or LOG_FILE.exists() if not is_vps: return data = parse_log() if LOG_FILE.exists() else {} if data is None: data = {} # Read z-scores from state files if log parsing didn't provide them try: import json as _j2 if eu_state.exists(): st = _j2.loads(eu_state.read_text(encoding="utf-8")) if data.get("eu_zscore") is None: data["eu_zscore"] = st.get("last_zscore", 0.0) data["zscore"] = st.get("last_zscore", 0.0) if uj_state.exists(): st = _j2.loads(uj_state.read_text(encoding="utf-8")) if data.get("uj_zscore") is None: data["uj_zscore"] = st.get("last_zscore", 0.0) except Exception: pass data["yields"] = fetch_yield_data() data["timestamp"] = data.get("timestamp", datetime.now().isoformat()) # Load real backtest equity curve + metrics if available # (generated by src/research/build_equity_curve.py) try: eq_path = BASE_PATH / "data" / "output" / "dashboard_equity.json" if eq_path.exists(): with eq_path.open(encoding="utf-8") as f: eq_data = json.load(f) # Pass through everything the dashboard needs: # - metrics (incl. dynamic block + worst_year_fixed/dynamic) # - all 12 equity curves (combined / per-pair × fixed/compounded × baseline/dynamic) # - per-year DD arrays (fixed and dynamic) for the FTMO bar chart data["backtest"] = { "metrics": eq_data.get("metrics", {}), "generated_at": eq_data.get("generated_at"), # Combined portfolio (1.0× baseline) — historical default tab "equity_fixed": eq_data.get("equity_curve_fixed", {}), "equity_compounded": eq_data.get("equity_curve_compounded", {}), # Per-pair (1.0× baseline) "equity_eurusd_fixed": eq_data.get("equity_curve_eurusd_fixed", {}), "equity_eurusd_compounded": eq_data.get("equity_curve_eurusd_compounded", {}), "equity_usdjpy_fixed": eq_data.get("equity_curve_usdjpy_fixed", {}), "equity_usdjpy_compounded": eq_data.get("equity_curve_usdjpy_compounded", {}), # Combined portfolio (dynamic tier sizing) "equity_curve_dynamic_fixed": eq_data.get("equity_curve_dynamic_fixed", {}), "equity_curve_dynamic_compounded": eq_data.get("equity_curve_dynamic_compounded", {}), # Per-pair (dynamic tier sizing) "equity_curve_eurusd_dynamic_fixed": eq_data.get("equity_curve_eurusd_dynamic_fixed", {}), "equity_curve_eurusd_dynamic_compounded": eq_data.get("equity_curve_eurusd_dynamic_compounded", {}), "equity_curve_usdjpy_dynamic_fixed": eq_data.get("equity_curve_usdjpy_dynamic_fixed", {}), "equity_curve_usdjpy_dynamic_compounded": eq_data.get("equity_curve_usdjpy_dynamic_compounded", {}), # Per-year DD bar data (FTMO fresh-account perspective) "yearly_dd_fixed": eq_data.get("yearly_dd_fixed", {}), "yearly_dd_dynamic": eq_data.get("yearly_dd_dynamic", {}), # Daily P&L records (for monthly drilldown chart — Phase 2) "daily_pnl_fixed": eq_data.get("daily_pnl_fixed", {}), "daily_pnl_dynamic": eq_data.get("daily_pnl_dynamic", {}), } except Exception as _e: print(f"[WARNING] Could not load backtest equity JSON: {_e}") eu_z = float(data.get("eu_zscore") or data.get("zscore") or 0.0) uj_z = float(data.get("uj_zscore") or 0.0) if not data.get("signal_outlook"): data["signal_outlook"] = get_weekly_outlook(eu_z) if not data.get("usdjpy_outlook"): data["usdjpy_outlook"] = get_usdjpy_weekly_outlook(uj_z) # ── Regime classification (Activity + Macro tiers per pair) ───────── # Live-JSON-only: regime_live reads data/logs/trade_history_*.json directly # for activity counts and data/raw/rates/* for macro composite scores. # Thresholds come from data/research/regime_thresholds.json (calibrated # once by src/research/regime_classifier_v1.py). if _REGIME_AVAILABLE: try: data["regime"] = regime_live.compute_regimes(BASE_PATH) except Exception as _re: print(f"[regime] compute failed: {_re}") data["regime"] = {"errors": [str(_re)]} DATA_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") eu_sig = "SIGNAL" if abs(eu_z) >= 2.75 else "no signal" uj_sig = "SIGNAL" if abs(uj_z) >= 2.00 else "no signal" tc = data.get("trade_count", 0) pos = data.get("position", "None") print(f" [{datetime.now().strftime('%H:%M:%S')}] " f"EU z={eu_z:.3f} {eu_sig} | " f"UJ z={uj_z:.3f} {uj_sig} | " f"trades={tc} | pos={pos}") class DashboardHandler(http.server.SimpleHTTPRequestHandler): """Serves dashboard.html and dashboard_data.json from BASE_PATH. Protected resources (the dashboard HTML and its data JSON) require a valid ?key= query parameter. Tokens are managed by the live signal monitor via Telegram (/grantdashboard, /revokedashboard). Static assets (favicon etc.) are served without a token check. """ # Protected paths — require a valid token. Anything else is public. _PROTECTED_PREFIXES = ("/dashboard.html", "/dashboard_data.json", "/dashboard_equity.json") def __init__(self, *args, **kwargs): super().__init__(*args, directory=str(BASE_PATH), **kwargs) def log_message(self, format, *args): pass # suppress request logs def _load_valid_tokens(self): """Read dashboard_tokens.json and return set of valid token strings. File is read fresh on each request — overhead is minimal (file is small) and means token grants/revokes take effect immediately without restarting the server. """ try: p = BASE_PATH / "data" / "logs" / "dashboard_tokens.json" if not p.exists(): return set() data = json.loads(p.read_text(encoding="utf-8")) return {str(t.get("token", "")) for t in data.get("tokens", []) if t.get("token")} except Exception: return set() def _path_requires_token(self, path): """Return True if path is a protected resource.""" path_lower = path.lower().split("?", 1)[0] return any(path_lower.startswith(p) for p in self._PROTECTED_PREFIXES) def _extract_token(self, path): """Pull ?key=X from the URL. Returns empty string if absent.""" if "?" not in path: return "" from urllib.parse import parse_qs qs = path.split("?", 1)[1] params = parse_qs(qs) return params.get("key", [""])[0].strip() def _serve_access_denied(self): """Send a friendly access-denied HTML page (HTTP 403).""" body = ( "" "Access required" "" "

🔒 Private dashboard

" "

This dashboard requires a personal access link.

" "

Message the bot on Telegram and use " "/requestdashboard to request access.

" "" ).encode("utf-8") self.send_response(403) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() try: self.wfile.write(body) except Exception: pass def do_GET(self): # Public paths fall through to the default static-file handler if not self._path_requires_token(self.path): return super().do_GET() # Protected path — validate token token = self._extract_token(self.path) if token and token in self._load_valid_tokens(): # Strip the query string before passing to the file handler so # SimpleHTTPRequestHandler doesn't try to open "dashboard.html?key=X" self.path = self.path.split("?", 1)[0] return super().do_GET() # Token missing or invalid — deny return self._serve_access_denied() def run_server(): with http.server.HTTPServer(("", PORT), DashboardHandler) as httpd: print(f" Dashboard server running at http://localhost:{PORT}/dashboard.html") httpd.serve_forever() def _do_refresh_live_trades(): """Background worker: runs extract + build subprocesses sequentially. Runs in a daemon thread, never raises (failures are logged + swallowed). On success, the build script overwrites dashboard_equity.json and the HTTP handler picks it up on the next browser request automatically. """ global _last_live_refresh, _last_live_refresh_attempt, _live_refresh_running started = datetime.now() last_str = _last_live_refresh.strftime("%H:%M:%S") if _last_live_refresh else "never" print(f" [live-refresh] Starting (last success: {last_str})") # Force UTF-8 on the subprocess so Unicode characters in script output # (→, ═, etc.) don't blow up Windows' default cp1252 pipe encoding. child_env = os.environ.copy() child_env["PYTHONIOENCODING"] = "utf-8" run_kwargs = dict( capture_output=True, text=True, encoding="utf-8", errors="replace", cwd=str(BASE_PATH), env=child_env, ) # Mark this attempt regardless of outcome so a failing refresh doesn't # retry every 30 seconds — it'll wait the full interval before trying again. _last_live_refresh_attempt = datetime.now() try: # Step 1: extract live trades from MT5 broker history if not EXTRACT_SCRIPT.exists(): print(f" [live-refresh] WARN: extract script missing at {EXTRACT_SCRIPT}") return try: r1 = subprocess.run( [sys.executable, str(EXTRACT_SCRIPT)], timeout=LIVE_REFRESH_EXTRACT_TIMEOUT, **run_kwargs, ) if r1.returncode != 0: # Don't fail the whole refresh — extract may fail if MT5 momentarily # disconnected. Log and continue to build with whatever live CSV exists. tail = (r1.stderr or r1.stdout or "")[-200:].strip() print(f" [live-refresh] extract returned {r1.returncode}: {tail}") except subprocess.TimeoutExpired: print(f" [live-refresh] extract timeout ({LIVE_REFRESH_EXTRACT_TIMEOUT}s) — skipping rebuild") return # Step 2: rebuild dashboard equity JSON (always runs, even if extract had issues) if not BUILD_SCRIPT.exists(): print(f" [live-refresh] WARN: build script missing at {BUILD_SCRIPT}") return try: r2 = subprocess.run( [sys.executable, str(BUILD_SCRIPT)], timeout=LIVE_REFRESH_BUILD_TIMEOUT, **run_kwargs, ) if r2.returncode != 0: tail = (r2.stderr or r2.stdout or "")[-200:].strip() print(f" [live-refresh] build returned {r2.returncode}: {tail}") return except subprocess.TimeoutExpired: print(f" [live-refresh] build timeout ({LIVE_REFRESH_BUILD_TIMEOUT}s) — aborted") return _last_live_refresh = datetime.now() elapsed = (_last_live_refresh - started).total_seconds() # Try to surface a useful one-line summary from the build output # (looks for the "live trades:" line emitted by load_live_trades()) summary = "" for line in (r2.stdout or "").splitlines(): line = line.strip() if "live trades:" in line and "EURUSD" in line: summary = line[:120] break if not summary: summary = "OK" print(f" [live-refresh] Complete in {elapsed:.1f}s — {summary}") except Exception as e: print(f" [live-refresh] Failed (non-fatal): {e}") finally: with _live_refresh_lock: _live_refresh_running = False def maybe_refresh_live_trades(): """Decide whether a live-data refresh is due. Non-blocking. Called from the main update loop. Spawns a daemon thread to run the actual extract + build if (a) feature is enabled, (b) no refresh is already running, and (c) enough time has elapsed since last refresh ATTEMPT (success or failure). Throttling on failure prevents log spam when MT5 is momentarily disconnected or scripts have issues. """ if not LIVE_REFRESH_ENABLED: return global _live_refresh_running with _live_refresh_lock: if _live_refresh_running: return # Use last ATTEMPT (not last success) for throttling — prevents tight # retry loops on persistent failures last = _last_live_refresh_attempt or _last_live_refresh if last is not None: elapsed = (datetime.now() - last).total_seconds() if elapsed < LIVE_REFRESH_INTERVAL_SEC: return _live_refresh_running = True threading.Thread(target=_do_refresh_live_trades, daemon=True).start() def main(): print("=" * 60) print("DASHBOARD SERVER") print(f" Log file : {LOG_FILE}") print(f" Data file: {DATA_FILE}") print(f" Port : {PORT}") eu_st = BASE_PATH / "data" / "logs" / "state_eurusd.json" if LOG_FILE.exists() or eu_st.exists(): print(f" Mode : VPS (writing live data)") else: print(f" Mode : Home machine (read-only — serving VPS data via OneDrive)") # Calendar feed status try: if TE_API_KEY and len(TE_API_KEY) > 8: print(f" Calendar : FinanceFlow API enabled (key ...{TE_API_KEY[-6:]})") else: print(f" Calendar : FinanceFlow DISABLED (no API key)") except NameError: print(f" Calendar : FinanceFlow DISABLED (no API key set)") # Live refresh status if LIVE_REFRESH_ENABLED: mins = LIVE_REFRESH_INTERVAL_SEC // 60 ok = EXTRACT_SCRIPT.exists() and BUILD_SCRIPT.exists() if ok: print(f" Live-refresh: ENABLED (every {mins} min) → fires on first request after interval") else: missing = [] if not EXTRACT_SCRIPT.exists(): missing.append("extract_live_trades.py") if not BUILD_SCRIPT.exists(): missing.append("build_equity_curve.py") print(f" Live-refresh: DISABLED (missing: {', '.join(missing)})") else: print(f" Live-refresh: DISABLED (set LIVE_REFRESH_ENABLED=True in server file to enable)") print("=" * 60) # Start HTTP server in background thread server_thread = threading.Thread(target=run_server, daemon=True) server_thread.start() print(f"\n Open in browser: http://localhost:{PORT}/dashboard.html\n") # Update loop print(" Updating every 30 seconds (Ctrl+C to stop):\n") while True: try: write_data() # Phase 4: trigger live-refresh if interval has elapsed (non-blocking) maybe_refresh_live_trades() time.sleep(30) except KeyboardInterrupt: print("\n Dashboard server stopped.") break if __name__ == "__main__": main()