#!/usr/bin/env python3 """ build_heatmap_data.py ===================== Builds heatmap_data.json from REAL sources only — no hardcoded or fabricated numbers. Cleanly separates backtest period from live paper-trading period. SOURCES: EU backtest: data/processed/trades/trades_real_costs.csv (P&L col = dollar_pnl_real, dates derived from exit_time) UJ backtest: data/processed/usdjpy_trades_real_costs.csv (P&L col = pnl_real, year/month columns present) EU live: data/logs/trade_history_eurusd.json UJ live: data/logs/trade_history_usdjpy.json CUTOVER: Backtest data is used THROUGH 2026-03 (March 2026). Live data is used FROM 2026-04 (April 2026) onward. This avoids any overlap/double-counting in the live-launch period. OUTPUT: data/logs/heatmap_data.json { "cutover_month": "2026-04", "generated_at": "...", "eurusd": {"YYYY": [12 monthly values or null], ...}, "usdjpy": {"YYYY": [...], ...}, "source_meta": {... which file fed which year ...} } Each 2026 array uses backtest for Jan-Mar, live for Apr-current, null for future. """ import csv import json from pathlib import Path from datetime import datetime from collections import defaultdict # ---- Resolve base path (works on VPS or home machine) -------------------- candidates_base = [ Path(r"C:\Users\Administrator\OneDrive\fx_macro_intraday"), Path(r"C:\Users\paul_\OneDrive\fx_macro_intraday"), ] BASE = next((b for b in candidates_base if b.exists()), None) if BASE is None: raise SystemExit("ERROR: could not find project base path") print(f"Base: {BASE}") EU_BT = BASE / "data" / "processed" / "trades" / "trades_real_costs.csv" UJ_BT = BASE / "data" / "processed" / "usdjpy_trades_real_costs.csv" EU_LIVE = BASE / "data" / "logs" / "trade_history_eurusd.json" UJ_LIVE = BASE / "data" / "logs" / "trade_history_usdjpy.json" OUT = BASE / "data" / "logs" / "heatmap_data.json" CUTOVER_YEAR = 2026 CUTOVER_MONTH = 4 # April — first clean live month. Backtest is used < this. # ---- Helpers -------------------------------------------------------------- def blank_year(): return [0.0] * 12 def add_month(store, year, month, value): if month < 1 or month > 12: return store.setdefault(year, blank_year()) store[year][month - 1] += value # ---- EU backtest: derive year/month from exit_time, sum dollar_pnl_real --- def load_eu_backtest(): store = {} with open(EU_BT, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: try: # exit_time like "2005-03-04 16:15:00" dt = datetime.strptime(row["exit_time"][:19], "%Y-%m-%d %H:%M:%S") pnl = float(row["dollar_pnl_real"]) except (ValueError, KeyError): continue # Cutover: backtest only BEFORE April 2026 if dt.year > CUTOVER_YEAR or (dt.year == CUTOVER_YEAR and dt.month >= CUTOVER_MONTH): continue add_month(store, dt.year, dt.month, pnl) return store # ---- UJ backtest: use year/month cols, sum pnl_real ----------------------- def load_uj_backtest(): store = {} with open(UJ_BT, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: try: y = int(row["year"]); m = int(row["month"]) pnl = float(row["pnl_real"]) except (ValueError, KeyError): continue if y > CUTOVER_YEAR or (y == CUTOVER_YEAR and m >= CUTOVER_MONTH): continue add_month(store, y, m, pnl) return store # ---- Live JSON: bucket by close month, FROM April 2026 -------------------- def load_live(path): store = {} if not path.exists(): return store trades = json.loads(path.read_text(encoding="utf-8")) for t in trades: closed = t.get("closed_at", "") if len(closed) < 7: continue try: y = int(closed[0:4]); m = int(closed[5:7]) pnl = float(t.get("net_pnl", 0)) except ValueError: continue # Live only FROM April 2026 onward if y < CUTOVER_YEAR or (y == CUTOVER_YEAR and m < CUTOVER_MONTH): continue add_month(store, y, m, pnl) return store # ---- Merge backtest + live into one per-pair structure -------------------- def merge(backtest, live): """Combine; round; mark future months of the current year as null.""" all_years = sorted(set(backtest) | set(live)) out = {} now = datetime.now() for y in all_years: arr = blank_year() if y in backtest: for i in range(12): arr[i] += backtest[y][i] if y in live: for i in range(12): arr[i] += live[y][i] arr = [round(v) for v in arr] # Null out future months of the current calendar year if y == now.year: for i in range(now.month, 12): arr[i] = None out[str(y)] = arr return out def main(): print("Loading EU backtest...") eu_bt = load_eu_backtest() print(f" EU backtest years: {min(eu_bt)}-{max(eu_bt)}") print("Loading UJ backtest...") uj_bt = load_uj_backtest() print(f" UJ backtest years: {min(uj_bt)}-{max(uj_bt)}") print("Loading EU live...") eu_live = load_live(EU_LIVE) print(f" EU live years: {sorted(eu_live) or 'none'}") print("Loading UJ live...") uj_live = load_live(UJ_LIVE) print(f" UJ live years: {sorted(uj_live) or 'none'}") eu = merge(eu_bt, eu_live) uj = merge(uj_bt, uj_live) payload = { "cutover_month": f"{CUTOVER_YEAR}-{CUTOVER_MONTH:02d}", "generated_at": datetime.now().isoformat(), "note": "Backtest data through 2026-03; live paper-trade data from 2026-04 onward. P&L is cost-adjusted.", "eurusd": eu, "usdjpy": uj, } OUT.write_text(json.dumps(payload, indent=2), encoding="utf-8") print(f"\nWrote {OUT}") # Sanity print: 2026 rows print("\n2026 EU:", eu.get("2026")) print("2026 UJ:", uj.get("2026")) print("\nReminder: EU 2026 should be Jan-Mar backtest, Apr+ live.") print("Live EU April should match your trade JSON April closes.") if __name__ == "__main__": main()