""" ═══════════════════════════════════════════════════════════════════════════════ build_equity_curve.py — generate real equity curve & metrics from backtest ═══════════════════════════════════════════════════════════════════════════════ Reads the canonical backtest trade tables that produced the v4 documentation numbers (Sharpe 5.26, Max DD 2.61%) and converts them into a daily equity curve + drawdown series + summary metrics. Inputs (must exist): data/processed/trades/trades_real_costs.csv — EURUSD trades data/processed/usdjpy_trades_real_costs.csv — USDJPY trades Outputs: data/output/dashboard_equity.json — for dashboard to read data/output/equity_summary.txt — printable summary Run from project root: python src/research/build_equity_curve.py ═══════════════════════════════════════════════════════════════════════════════ """ from __future__ import annotations import json import sys from pathlib import Path from datetime import datetime import pandas as pd import numpy as np # ── Paths ──────────────────────────────────────────────────────────────────── SCRIPT_PATH = Path(__file__).resolve() PROJECT_ROOT = SCRIPT_PATH.parent.parent.parent DATA_DIR = PROJECT_ROOT / "data" PROC_DIR = DATA_DIR / "processed" OUT_DIR = DATA_DIR / "output" OUT_DIR.mkdir(parents=True, exist_ok=True) EU_TRADES = PROC_DIR / "trades" / "trades_real_costs.csv" UJ_TRADES = PROC_DIR / "usdjpy_trades_real_costs.csv" JSON_OUT = OUT_DIR / "dashboard_equity.json" SUMM_OUT = OUT_DIR / "equity_summary.txt" STARTING_BALANCE = 100_000.0 # ── Loaders ────────────────────────────────────────────────────────────────── # ── Sizing tier configuration ──────────────────────────────────────────────── # These match the live monitor's risk tiers (live_signal_monitor_v2.py). # Each band maps |z-score| to a position-size multiplier. EU_TIERS = [(2.75, 3.5, 1.0), (3.5, 4.5, 1.5), (4.5, 999.0, 2.0)] UJ_TIERS = [(2.0, 2.5, 1.0), (2.5, 3.5, 1.5), (3.5, 999.0, 2.0)] def tier_multiplier(z_abs: float, tiers: list) -> float: """Return the sizing multiplier for a given |z-score| based on the tier table.""" for lo, hi, mult in tiers: if lo <= z_abs < hi: return mult return 0.0 # below threshold = no trade (shouldn't happen for valid signals) # ── Loaders ────────────────────────────────────────────────────────────────── def load_eurusd(): """ Load EURUSD trades. Schema: entry_time, exit_time, ..., dollar_pnl_real, return_real, zscore_abs """ if not EU_TRADES.exists(): raise FileNotFoundError(f"EURUSD trade file missing: {EU_TRADES}") df = pd.read_csv(EU_TRADES, parse_dates=["entry_time", "exit_time"]) z_abs = df["zscore_abs"].astype(float) multiplier = z_abs.apply(lambda z: tier_multiplier(z, EU_TIERS)) out = pd.DataFrame({ "pair": "EURUSD", "entry_date": df["entry_time"].dt.normalize(), "exit_date": df["exit_time"].dt.normalize(), "pnl": df["dollar_pnl_real"].astype(float), "pnl_dynamic": df["dollar_pnl_real"].astype(float) * multiplier, "tier": multiplier, "zscore_abs": z_abs, "return_pct": df["return_real"].astype(float) * 100, "win": (df["dollar_pnl_real"] > 0).astype(int), }) print(f" EURUSD trades loaded: {len(out):,} | " f"{out['exit_date'].min().date()} → {out['exit_date'].max().date()}") print(f" Tier 1.0x: {(out.tier==1.0).sum():>5} trades | " f"sum=${out.loc[out.tier==1.0,'pnl'].sum():>12,.0f} (1.0x) " f"=${out.loc[out.tier==1.0,'pnl_dynamic'].sum():>12,.0f} (dyn)") print(f" Tier 1.5x: {(out.tier==1.5).sum():>5} trades | " f"sum=${out.loc[out.tier==1.5,'pnl'].sum():>12,.0f} (1.0x) " f"=${out.loc[out.tier==1.5,'pnl_dynamic'].sum():>12,.0f} (dyn)") print(f" Tier 2.0x: {(out.tier==2.0).sum():>5} trades | " f"sum=${out.loc[out.tier==2.0,'pnl'].sum():>12,.0f} (1.0x) " f"=${out.loc[out.tier==2.0,'pnl_dynamic'].sum():>12,.0f} (dyn)") return out def load_usdjpy(): """ Load USDJPY trades. Schema: entry_time, exit_reason, ..., pnl_real, zscore_abs No exit_time column — most trades close within 24h hold so use entry date as a daily-resolution proxy. The dashboard chart is daily anyway. """ if not UJ_TRADES.exists(): raise FileNotFoundError(f"USDJPY trade file missing: {UJ_TRADES}") df = pd.read_csv(UJ_TRADES, parse_dates=["entry_time"]) z_abs = df["zscore_abs"].astype(float) multiplier = z_abs.apply(lambda z: tier_multiplier(z, UJ_TIERS)) out = pd.DataFrame({ "pair": "USDJPY", "entry_date": df["entry_time"].dt.normalize(), "exit_date": df["entry_time"].dt.normalize(), "pnl": df["pnl_real"].astype(float), "pnl_dynamic": df["pnl_real"].astype(float) * multiplier, "tier": multiplier, "zscore_abs": z_abs, "return_pct": (df["pnl_real"].astype(float) / 300_000.0) * 100, "win": (df["pnl_real"] > 0).astype(int), }) print(f" USDJPY trades loaded: {len(out):,} | " f"{out['entry_date'].min().date()} → {out['entry_date'].max().date()}") print(f" Tier 1.0x: {(out.tier==1.0).sum():>5} trades | " f"sum=${out.loc[out.tier==1.0,'pnl'].sum():>12,.0f} (1.0x) " f"=${out.loc[out.tier==1.0,'pnl_dynamic'].sum():>12,.0f} (dyn)") print(f" Tier 1.5x: {(out.tier==1.5).sum():>5} trades | " f"sum=${out.loc[out.tier==1.5,'pnl'].sum():>12,.0f} (1.0x) " f"=${out.loc[out.tier==1.5,'pnl_dynamic'].sum():>12,.0f} (dyn)") print(f" Tier 2.0x: {(out.tier==2.0).sum():>5} trades | " f"sum=${out.loc[out.tier==2.0,'pnl'].sum():>12,.0f} (1.0x) " f"=${out.loc[out.tier==2.0,'pnl_dynamic'].sum():>12,.0f} (dyn)") return out # ── Equity curve construction ──────────────────────────────────────────────── def _build_curve_from_pnl(daily_pnl_series: pd.Series, compound: bool) -> pd.DataFrame: """ Generic equity curve builder. Takes a date-indexed series of daily $ P&L and returns a DataFrame with date, daily_pnl, balance, peak, drawdown_pct, drawdown_dollar columns. If compound=True, each day's P&L is scaled by (current_balance / starting), capped at 5× to prevent multi-decade overflow. """ daily = pd.DataFrame({ "date": daily_pnl_series.index, "daily_pnl": daily_pnl_series.values, }).sort_values("date").reset_index(drop=True) if not compound: daily["balance"] = STARTING_BALANCE + daily["daily_pnl"].cumsum() else: MAX_SCALE = 5.0 balances = [] bal = STARTING_BALANCE for pnl in daily["daily_pnl"].values: scale = min(bal / STARTING_BALANCE, MAX_SCALE) today_pnl = pnl * scale bal += today_pnl balances.append(bal) daily["balance"] = balances daily["daily_pnl"] = daily["balance"].diff().fillna(daily["balance"] - STARTING_BALANCE) daily["peak"] = daily["balance"].cummax() daily["drawdown_dollar"] = daily["balance"] - daily["peak"] daily["drawdown_pct"] = (daily["drawdown_dollar"] / daily["peak"]) * 100 return daily def build_daily_equity(eu_df: pd.DataFrame, uj_df: pd.DataFrame, compound: bool, pnl_col: str = "pnl"): """ Build the COMBINED portfolio equity curve (EURUSD + USDJPY). Returns (daily_dataframe, final_balance, max_dd_pct, max_dd_amount). pnl_col selects which column to aggregate: "pnl" → fixed 1.0× sizing (baseline backtest) "pnl_dynamic" → dynamic tier sizing matching the live monitor's risk logic Daily dataframe columns: date, eu_pnl, uj_pnl, daily_pnl, balance, peak, drawdown_pct, drawdown_dollar """ # Group each pair's P&L by exit date eu_daily = (eu_df.groupby("exit_date")[pnl_col].sum() .rename("eu_pnl")) uj_daily = (uj_df.groupby("exit_date")[pnl_col].sum() .rename("uj_pnl")) # Build complete date range (zero P&L on idle days for chart continuity) start = min(eu_daily.index.min(), uj_daily.index.min()) end = max(eu_daily.index.max(), uj_daily.index.max()) all_dates = pd.date_range(start=start, end=end, freq="D") eu_full = eu_daily.reindex(all_dates, fill_value=0.0) uj_full = uj_daily.reindex(all_dates, fill_value=0.0) combined_pnl = (eu_full + uj_full).rename("combined_pnl") # Build the combined curve curve = _build_curve_from_pnl(combined_pnl, compound) curve["eu_pnl"] = eu_full.values curve["uj_pnl"] = uj_full.values return ( curve, curve["balance"].iloc[-1], curve["drawdown_pct"].min(), curve["drawdown_dollar"].min(), ) def build_pair_equity(pair_df: pd.DataFrame, compound: bool, pnl_col: str = "pnl"): """ Build a SINGLE-PAIR equity curve from one pair's trades. pnl_col: "pnl" for fixed, "pnl_dynamic" for tier-scaled. """ pair_daily = pair_df.groupby("exit_date")[pnl_col].sum() if len(pair_daily) == 0: return pd.DataFrame(columns=["date", "balance", "drawdown_pct"]) all_dates = pd.date_range(start=pair_daily.index.min(), end=pair_daily.index.max(), freq="D") pair_full = pair_daily.reindex(all_dates, fill_value=0.0) return _build_curve_from_pnl(pair_full, compound) # ── Performance metrics ────────────────────────────────────────────────────── def compute_yearly_dd_view(eu_df: pd.DataFrame, uj_df: pd.DataFrame, pnl_col: str = "pnl") -> pd.DataFrame: """ Compute per-year drawdown as if each year started fresh with $100k. This is the FTMO-relevant view: each year is a "fresh challenge" perspective. Within-year DD is calculated against running peak inside that year only — not against all-time peak. Returns a DataFrame with columns: year, trades, total_pnl, dd_pct, dd_dollar, worst_trade, end_balance. """ eu_t = eu_df[["exit_date", pnl_col]].rename(columns={"exit_date": "time", pnl_col: "pnl"}) uj_t = uj_df[["exit_date", pnl_col]].rename(columns={"exit_date": "time", pnl_col: "pnl"}) all_t = pd.concat([eu_t, uj_t]).sort_values("time").reset_index(drop=True) all_t["year"] = all_t["time"].dt.year rows = [] for yr, sub in all_t.groupby("year"): if len(sub) == 0: continue # Each year starts at $100k (FTMO-fresh-account view) balance = STARTING_BALANCE + sub["pnl"].cumsum() peak = balance.cummax() dd_dollar = (balance - peak).min() dd_pct = (dd_dollar / STARTING_BALANCE) * 100 rows.append({ "year": int(yr), "trades": int(len(sub)), "total_pnl": round(sub["pnl"].sum(), 2), "dd_pct": round(dd_pct, 3), "dd_dollar": round(dd_dollar, 2), "worst_trade": round(sub["pnl"].min(), 2), "end_balance": round(balance.iloc[-1], 2), }) return pd.DataFrame(rows) def compute_daily_pnl_records(eu_df: pd.DataFrame, uj_df: pd.DataFrame, pnl_col: str = "pnl", eu_live: pd.DataFrame | None = None, uj_live: pd.DataFrame | None = None) -> dict: """ Build a compact daily P&L record for the dashboard's monthly drilldown. Returns a dict with parallel arrays: dates : ISO date strings 'YYYY-MM-DD' for every active trading day eu_pnl : EURUSD $ P&L on that day (0 if no EU trade closed that day) uj_pnl : USDJPY $ P&L on that day eu_trades: # of EURUSD trades that closed that day uj_trades: # of USDJPY trades that closed that day Only active days (≥1 trade) are emitted. The dashboard fills in the missing calendar days client-side when rendering the monthly view. pnl_col selects which P&L column to aggregate ("pnl" for 1.0× baseline, "pnl_dynamic" for tier-scaled). eu_live / uj_live: optional live-trade DataFrames (same schema as eu_df / uj_df). If provided, their rows are concatenated onto the historic data before aggregation. This is how live trades flow into the drilldown without contaminating the equity curves / hero stats. """ # Concat historic + live before grouping. Live trades use exit_time-derived # exit_date just like historic, so the daily grouping works uniformly. eu_full = eu_df if eu_live is None or eu_live.empty else pd.concat([eu_df, eu_live], ignore_index=True) uj_full = uj_df if uj_live is None or uj_live.empty else pd.concat([uj_df, uj_live], ignore_index=True) # EURUSD daily aggregation eu_grouped = (eu_full.groupby("exit_date") .agg(eu_pnl=(pnl_col, "sum"), eu_trades=(pnl_col, "count"))) # USDJPY daily aggregation uj_grouped = (uj_full.groupby("exit_date") .agg(uj_pnl=(pnl_col, "sum"), uj_trades=(pnl_col, "count"))) # Outer join — keep any day that had EU OR UJ activity daily = eu_grouped.join(uj_grouped, how="outer").fillna(0.0).sort_index() return { "dates": [d.strftime("%Y-%m-%d") for d in daily.index], "eu_pnl": [round(float(v), 2) for v in daily["eu_pnl"]], "uj_pnl": [round(float(v), 2) for v in daily["uj_pnl"]], "eu_trades": [int(v) for v in daily["eu_trades"]], "uj_trades": [int(v) for v in daily["uj_trades"]], } def load_live_trades(symbol: str) -> pd.DataFrame: """ Load live-trade CSV produced by extract_live_trades.py. Returns a DataFrame in the same schema as load_eurusd() / load_usdjpy() so it can be concatenated for daily aggregation. Columns: pair, entry_date, exit_date, pnl, pnl_dynamic, tier, win Returns an EMPTY DataFrame if the file doesn't exist or has no rows. Critical: live trades feed ONLY the drilldown's daily records — they do NOT enter the equity curve / yearly DD / metrics calculations, so the historical backtest remains the canonical "track record" for FTMO sizing. """ fname = "trades_live_eurusd.csv" if symbol == "EURUSD" else "trades_live_usdjpy.csv" path = PROJECT_ROOT / "data" / "live" / fname empty = pd.DataFrame(columns=["pair","entry_date","exit_date","pnl", "pnl_dynamic","tier","zscore_abs", "return_pct","win"]) if not path.exists(): return empty try: raw = pd.read_csv(path, parse_dates=["entry_time","exit_time"]) except Exception as e: print(f" WARN: could not read {fname}: {e}") return empty if raw.empty: return empty # Live "net_pnl" already includes the dynamic tier sizing (the live monitor # scales lots by tier before placing the order). So: # pnl_dynamic = net_pnl as recorded (real broker P&L with swap/comm) # pnl = pnl_dynamic / tier (back-out the tier multiplier to get # the equivalent 1.0× baseline P&L) pnl_dyn = raw["net_pnl"].astype(float) tier = raw["tier"].astype(float).clip(lower=0.5) # safety floor pnl_base = pnl_dyn / tier out = pd.DataFrame({ "pair": symbol, "entry_date": raw["entry_time"].dt.normalize(), "exit_date": raw["exit_time"].dt.normalize(), "pnl": pnl_base, "pnl_dynamic": pnl_dyn, "tier": tier, "zscore_abs": 0.0, # not recorded for live trades — set 0 (unused for drilldown) "return_pct": 0.0, # not used by drilldown aggregation "win": (pnl_dyn > 0).astype(int), }) print(f" {symbol} live trades: {len(out):>3} | sum 1.0×=${out['pnl'].sum():>9,.2f} dyn=${pnl_dyn.sum():>9,.2f}") return out def compute_metrics(eu_df: pd.DataFrame, uj_df: pd.DataFrame, daily_fixed: pd.DataFrame, daily_comp: pd.DataFrame, eu_curve_fixed=None, eu_curve_comp=None, uj_curve_fixed=None, uj_curve_comp=None, # Dynamic-sizing curves daily_dyn_fixed=None, daily_dyn_comp=None, # Yearly DD views yearly_dd_fixed=None, yearly_dd_dynamic=None) -> dict: """Compute all the headline statistics for the v4 doc replacement. Per-pair curves are optional — if supplied, their max-DD numbers are attached to the per-pair stats dict so the dashboard can show real per-tab DD badges. Dynamic-sizing curves (daily_dyn_*) when supplied surface the realistic "live monitor with tier multipliers" Sharpe / DD / final balance. yearly_dd_* DataFrames provide the FTMO-relevant fresh-account view. """ combined = pd.concat([eu_df, uj_df], ignore_index=True) # Per-pair stats def pair_stats(df, label, curve_fixed=None, curve_comp=None): if len(df) == 0: return {"label": label, "trades": 0} wins = df["win"].sum() losses = len(df) - wins avg_win = df.loc[df["win"] == 1, "pnl"].mean() if wins > 0 else 0 avg_loss = df.loc[df["win"] == 0, "pnl"].mean() if losses > 0 else 0 gross_win = df.loc[df["win"] == 1, "pnl"].sum() gross_loss = abs(df.loc[df["win"] == 0, "pnl"].sum()) pf = (gross_win / gross_loss) if gross_loss > 0 else float("inf") out = { "label": label, "trades": int(len(df)), "wins": int(wins), "losses": int(losses), "win_rate": round(wins / len(df) * 100, 2), "total_pnl": round(df["pnl"].sum(), 2), "avg_win": round(avg_win, 2), "avg_loss": round(avg_loss, 2), "best_trade": round(df["pnl"].max(), 2), "worst_trade": round(df["pnl"].min(), 2), "profit_factor": round(pf, 3) if pf != float("inf") else None, } # Attach per-pair drawdown numbers if curves were supplied if curve_fixed is not None and len(curve_fixed) > 0: out["max_dd_pct_fixed"] = round(abs(curve_fixed["drawdown_pct"].min()), 3) out["final_balance_fixed"] = round(curve_fixed["balance"].iloc[-1], 2) if curve_comp is not None and len(curve_comp) > 0: out["max_dd_pct_comp"] = round(abs(curve_comp["drawdown_pct"].min()), 3) out["final_balance_comp"] = round(curve_comp["balance"].iloc[-1], 2) return out eu_stats = pair_stats(eu_df, "EURUSD", eu_curve_fixed, eu_curve_comp) uj_stats = pair_stats(uj_df, "USDJPY", uj_curve_fixed, uj_curve_comp) comb_stats = pair_stats(combined, "COMBINED") # Sharpe — based on daily P&L, fixed-size curve (the honest metric) daily_pnl = daily_fixed["daily_pnl"].values daily_ret = daily_pnl / STARTING_BALANCE # daily return on starting equity if daily_ret.std() > 0: sharpe = (daily_ret.mean() / daily_ret.std()) * np.sqrt(252) else: sharpe = 0.0 # Sortino — only downside deviation downside = daily_ret[daily_ret < 0] if len(downside) > 0 and downside.std() > 0: sortino = (daily_ret.mean() / downside.std()) * np.sqrt(252) else: sortino = 0.0 # Calmar — annualized return / max drawdown days = (daily_fixed["date"].iloc[-1] - daily_fixed["date"].iloc[0]).days years = days / 365.25 final_bal = daily_fixed["balance"].iloc[-1] total_return_pct = (final_bal / STARTING_BALANCE - 1) * 100 cagr_fixed = ((final_bal / STARTING_BALANCE) ** (1/years) - 1) * 100 if years > 0 else 0 final_bal_comp = daily_comp["balance"].iloc[-1] cagr_comp = ((final_bal_comp / STARTING_BALANCE) ** (1/years) - 1) * 100 if years > 0 else 0 max_dd_fixed = abs(daily_fixed["drawdown_pct"].min()) max_dd_comp = abs(daily_comp["drawdown_pct"].min()) calmar = cagr_fixed / max_dd_fixed if max_dd_fixed > 0 else 0 # Annual P&L breakdown daily_fixed["year"] = daily_fixed["date"].dt.year annual_pnl = (daily_fixed.groupby("year")["daily_pnl"].sum() .round(2).to_dict()) out = { "starting_balance": STARTING_BALANCE, "final_balance_fixed": round(final_bal, 2), "final_balance_comp": round(final_bal_comp, 2), "total_return_pct": round(total_return_pct, 2), "years": round(years, 2), "cagr_fixed": round(cagr_fixed, 2), "cagr_compounded": round(cagr_comp, 2), "sharpe": round(sharpe, 3), "sortino": round(sortino, 3), "calmar": round(calmar, 3), "max_dd_pct_fixed": round(max_dd_fixed, 3), "max_dd_pct_comp": round(max_dd_comp, 3), "max_dd_dollar_fixed": round(daily_fixed["drawdown_dollar"].min(), 2), "max_dd_dollar_comp": round(daily_comp["drawdown_dollar"].min(), 2), "eurusd": eu_stats, "usdjpy": uj_stats, "combined": comb_stats, "annual_pnl": annual_pnl, } # ── Dynamic-sizing aggregate stats (matches live monitor's tier sizing) ── if daily_dyn_fixed is not None and len(daily_dyn_fixed) > 0: dyn_pnl = daily_dyn_fixed["daily_pnl"].values dyn_ret = dyn_pnl / STARTING_BALANCE dyn_sharpe = (dyn_ret.mean() / dyn_ret.std() * np.sqrt(252)) if dyn_ret.std() > 0 else 0 out["dynamic"] = { "sharpe": round(dyn_sharpe, 3), "final_balance_fixed": round(daily_dyn_fixed["balance"].iloc[-1], 2), "max_dd_pct_fixed": round(abs(daily_dyn_fixed["drawdown_pct"].min()), 3), "max_dd_dollar_fixed": round(daily_dyn_fixed["drawdown_dollar"].min(), 2), "total_pnl": round(daily_dyn_fixed["balance"].iloc[-1] - STARTING_BALANCE, 2), } if daily_dyn_comp is not None and len(daily_dyn_comp) > 0: out["dynamic"]["final_balance_comp"] = round(daily_dyn_comp["balance"].iloc[-1], 2) out["dynamic"]["max_dd_pct_comp"] = round(abs(daily_dyn_comp["drawdown_pct"].min()), 3) # ── Worst-year DD (FTMO fresh-account perspective) ─────────────────────── if yearly_dd_fixed is not None and len(yearly_dd_fixed) > 0: worst_idx = yearly_dd_fixed["dd_pct"].idxmin() wr = yearly_dd_fixed.loc[worst_idx] out["worst_year_fixed"] = { "year": int(wr["year"]), "dd_pct": float(wr["dd_pct"]), "dd_dollar": float(wr["dd_dollar"]), } if yearly_dd_dynamic is not None and len(yearly_dd_dynamic) > 0: worst_idx = yearly_dd_dynamic["dd_pct"].idxmin() wr = yearly_dd_dynamic.loc[worst_idx] out["worst_year_dynamic"] = { "year": int(wr["year"]), "dd_pct": float(wr["dd_pct"]), "dd_dollar": float(wr["dd_dollar"]), } return out # ── Output writers ─────────────────────────────────────────────────────────── def downsample_for_chart(daily: pd.DataFrame, target_points: int = 800): """ Downsample to a manageable number of points for the dashboard chart. 23 years × 365 days = ~8400 points. Downsample to ~800 points for fast render. """ if len(daily) <= target_points: return daily step = max(1, len(daily) // target_points) sampled = daily.iloc[::step].copy() # Always include the very last point if sampled.iloc[-1]["date"] != daily.iloc[-1]["date"]: sampled = pd.concat([sampled, daily.iloc[[-1]]], ignore_index=True) return sampled def _series_to_payload(daily: pd.DataFrame) -> dict: """Downsample and serialize one equity curve to JSON-friendly dict.""" if daily is None or len(daily) == 0: return {"dates": [], "balance": [], "drawdown_pct": []} sampled = downsample_for_chart(daily) return { "dates": sampled["date"].dt.strftime("%Y-%m-%d").tolist(), "balance": sampled["balance"].round(2).tolist(), "drawdown_pct": sampled["drawdown_pct"].round(3).tolist(), } def write_json(daily_fixed, daily_comp, metrics, eu_fixed=None, eu_comp=None, uj_fixed=None, uj_comp=None, # Dynamic-sizing variants (tier-scaled) daily_dyn_fixed=None, daily_dyn_comp=None, eu_dyn_fixed=None, eu_dyn_comp=None, uj_dyn_fixed=None, uj_dyn_comp=None, # Yearly DD view (FTMO-relevant fresh-account perspective) yearly_dd_fixed=None, yearly_dd_dynamic=None, # Daily P&L records for the monthly drilldown chart daily_pnl_fixed=None, daily_pnl_dynamic=None): """Compact JSON for the dashboard to fetch. Includes: - Combined + per-pair curves (1.0× baseline sizing) - Combined + per-pair curves (dynamic tier sizing matching live monitor) - Per-year DD view (each year as a fresh $100k account — FTMO-relevant) - Per-day P&L records (active trading days only) for monthly drilldown """ def yearly_to_dict(df): """Convert yearly-DD DataFrame to dashboard-friendly dict.""" if df is None or len(df) == 0: return {"years": [], "dd_pct": [], "dd_dollar": [], "pnl": []} return { "years": df["year"].tolist(), "dd_pct": df["dd_pct"].tolist(), "dd_dollar": df["dd_dollar"].tolist(), "pnl": df["total_pnl"].tolist(), } def daily_to_dict(d): if d is None: return {"dates": [], "eu_pnl": [], "uj_pnl": [], "eu_trades": [], "uj_trades": []} return d payload = { "generated_at": datetime.now().isoformat(), "source_files": { "eurusd": str(EU_TRADES.relative_to(PROJECT_ROOT)), "usdjpy": str(UJ_TRADES.relative_to(PROJECT_ROOT)), }, "metrics": metrics, # Combined portfolio — 1.0× baseline (default view) "equity_curve_fixed": _series_to_payload(daily_fixed), "equity_curve_compounded": _series_to_payload(daily_comp), # Per-pair — 1.0× baseline "equity_curve_eurusd_fixed": _series_to_payload(eu_fixed), "equity_curve_eurusd_compounded": _series_to_payload(eu_comp), "equity_curve_usdjpy_fixed": _series_to_payload(uj_fixed), "equity_curve_usdjpy_compounded": _series_to_payload(uj_comp), # Combined — dynamic tier sizing "equity_curve_dynamic_fixed": _series_to_payload(daily_dyn_fixed), "equity_curve_dynamic_compounded": _series_to_payload(daily_dyn_comp), # Per-pair — dynamic tier sizing "equity_curve_eurusd_dynamic_fixed": _series_to_payload(eu_dyn_fixed), "equity_curve_eurusd_dynamic_compounded": _series_to_payload(eu_dyn_comp), "equity_curve_usdjpy_dynamic_fixed": _series_to_payload(uj_dyn_fixed), "equity_curve_usdjpy_dynamic_compounded": _series_to_payload(uj_dyn_comp), # Per-year DD bars (FTMO fresh-account view) "yearly_dd_fixed": yearly_to_dict(yearly_dd_fixed), "yearly_dd_dynamic": yearly_to_dict(yearly_dd_dynamic), # Daily P&L records (compact: only active days emitted) "daily_pnl_fixed": daily_to_dict(daily_pnl_fixed), "daily_pnl_dynamic": daily_to_dict(daily_pnl_dynamic), } JSON_OUT.write_text(json.dumps(payload, indent=2), encoding="utf-8") return payload def write_summary(metrics, daily_fixed, daily_comp): lines = [] P = lines.append P("═" * 72) P(" DUAL FX MACRO MODEL — REAL BACKTEST METRICS") P(" Generated from canonical trade tables (real costs applied)") P("═" * 72) P(f" Source EURUSD : {EU_TRADES.relative_to(PROJECT_ROOT)}") P(f" Source USDJPY : {UJ_TRADES.relative_to(PROJECT_ROOT)}") P(f" Period : {metrics['years']} years") P(f" Starting bal : ${metrics['starting_balance']:,.2f}") P("") P("─" * 72) P(" HEADLINE METRICS") P("─" * 72) P(f" Sharpe Ratio : {metrics['sharpe']}") P(f" Sortino Ratio : {metrics['sortino']}") P(f" Calmar Ratio : {metrics['calmar']}") P("") P(f" Max DD (fixed) : {metrics['max_dd_pct_fixed']}% " f"(${metrics['max_dd_dollar_fixed']:,.2f})") P(f" Max DD (compound) : {metrics['max_dd_pct_comp']}% " f"(${metrics['max_dd_dollar_comp']:,.2f})") P("") P(f" Final balance (fixed) : ${metrics['final_balance_fixed']:,.2f}") P(f" Final balance (compound) : ${metrics['final_balance_comp']:,.2f}") P(f" Total return (fixed) : {metrics['total_return_pct']}%") P(f" CAGR (fixed) : {metrics['cagr_fixed']}%") P(f" CAGR (compounded) : {metrics['cagr_compounded']}%") P("") # ── DYNAMIC SIZING COMPARISON (matches live monitor's tier system) ── if "dynamic" in metrics: d = metrics["dynamic"] P("─" * 72) P(" DYNAMIC TIER SIZING (matches live monitor's risk logic)") P("─" * 72) P(f" Sharpe Ratio : {d['sharpe']} " f"(vs {metrics['sharpe']} at 1.0×)") P(f" Max DD (fixed) : {d['max_dd_pct_fixed']}% " f"(${d['max_dd_dollar_fixed']:,.2f})") if "max_dd_pct_comp" in d: P(f" Max DD (compound) : {d['max_dd_pct_comp']}%") P(f" Final balance (fixed) : ${d['final_balance_fixed']:,.2f}") if "final_balance_comp" in d: P(f" Final balance (compound) : ${d['final_balance_comp']:,.2f}") P(f" Total P&L : ${d['total_pnl']:,.2f} " f"(vs ${metrics['final_balance_fixed'] - 100000:,.0f} at 1.0×)") P("") # ── WORST-YEAR DD (FTMO-relevant fresh-account perspective) ── if "worst_year_fixed" in metrics: wf = metrics["worst_year_fixed"] P("─" * 72) P(" WORST CALENDAR YEAR (FTMO fresh-$100k-account perspective)") P("─" * 72) P(f" Worst year (1.0×) : {wf['year']} DD {wf['dd_pct']:.2f}% " f"(${wf['dd_dollar']:,.2f})") if "worst_year_dynamic" in metrics: wd = metrics["worst_year_dynamic"] P(f" Worst year (dyn) : {wd['year']} DD {wd['dd_pct']:.2f}% " f"(${wd['dd_dollar']:,.2f})") P("") P(" This is the perspective FTMO judges: each challenge is a fresh") P(" $100k account. The historical worst-year DD is your best estimate") P(" of likely worst case during a challenge.") P("") P("─" * 72) P(" PER-PAIR BREAKDOWN") P("─" * 72) for stats in [metrics["eurusd"], metrics["usdjpy"], metrics["combined"]]: P(f" {stats['label']:10s} trades={stats['trades']:>4} " f"wins={stats['wins']:>4} losses={stats['losses']:>4} " f"WR={stats['win_rate']:>5.1f}%") P(f" total=${stats['total_pnl']:>12,.2f} " f"avg_win=${stats['avg_win']:>8,.2f} " f"avg_loss=${stats['avg_loss']:>9,.2f}") P(f" best=${stats['best_trade']:>10,.2f} " f"worst=${stats['worst_trade']:>10,.2f} " f"PF={stats['profit_factor']}") P("") P("─" * 72) P(" ANNUAL P&L (fixed-size)") P("─" * 72) for year, pnl in sorted(metrics["annual_pnl"].items()): bar = "█" * min(40, max(1, int(abs(pnl) / 5000))) sign = "+" if pnl >= 0 else "-" P(f" {year} {sign}${abs(pnl):>10,.2f} {bar}") P("") P("═" * 72) P(f" Equity curve JSON written to: {JSON_OUT.relative_to(PROJECT_ROOT)}") P(f" This summary at: {SUMM_OUT.relative_to(PROJECT_ROOT)}") P("═" * 72) text = "\n".join(lines) SUMM_OUT.write_text(text, encoding="utf-8") print() print(text) # ── Main ───────────────────────────────────────────────────────────────────── def main(): print("═" * 72) print(" Building real equity curve from backtest trade tables") print("═" * 72) print() print("Loading trade data...") eu_df = load_eurusd() uj_df = load_usdjpy() print() print("Building combined equity curves (1.0× baseline)...") daily_fixed, _, _, _ = build_daily_equity(eu_df, uj_df, compound=False, pnl_col="pnl") daily_comp, _, _, _ = build_daily_equity(eu_df, uj_df, compound=True, pnl_col="pnl") print(f" Combined fixed: {len(daily_fixed):,} daily points " f"(final ${daily_fixed['balance'].iloc[-1]:,.0f})") print(f" Combined compounded: {len(daily_comp):,} daily points " f"(final ${daily_comp['balance'].iloc[-1]:,.0f})") print("Building per-pair equity curves (1.0× baseline)...") eu_fixed = build_pair_equity(eu_df, compound=False, pnl_col="pnl") eu_comp = build_pair_equity(eu_df, compound=True, pnl_col="pnl") uj_fixed = build_pair_equity(uj_df, compound=False, pnl_col="pnl") uj_comp = build_pair_equity(uj_df, compound=True, pnl_col="pnl") print(f" EURUSD fixed: final ${eu_fixed['balance'].iloc[-1]:,.0f}, " f"max DD {abs(eu_fixed['drawdown_pct'].min()):.2f}%") print(f" USDJPY fixed: final ${uj_fixed['balance'].iloc[-1]:,.0f}, " f"max DD {abs(uj_fixed['drawdown_pct'].min()):.2f}%") print("Building combined equity curves (DYNAMIC tier sizing)...") daily_dyn_fixed, _, _, _ = build_daily_equity(eu_df, uj_df, compound=False, pnl_col="pnl_dynamic") daily_dyn_comp, _, _, _ = build_daily_equity(eu_df, uj_df, compound=True, pnl_col="pnl_dynamic") print(f" Combined dyn fixed: final ${daily_dyn_fixed['balance'].iloc[-1]:,.0f}, " f"max DD {abs(daily_dyn_fixed['drawdown_pct'].min()):.2f}%") print(f" Combined dyn compounded: final ${daily_dyn_comp['balance'].iloc[-1]:,.0f}, " f"max DD {abs(daily_dyn_comp['drawdown_pct'].min()):.2f}%") print("Building per-pair equity curves (DYNAMIC)...") eu_dyn_fixed = build_pair_equity(eu_df, compound=False, pnl_col="pnl_dynamic") eu_dyn_comp = build_pair_equity(eu_df, compound=True, pnl_col="pnl_dynamic") uj_dyn_fixed = build_pair_equity(uj_df, compound=False, pnl_col="pnl_dynamic") uj_dyn_comp = build_pair_equity(uj_df, compound=True, pnl_col="pnl_dynamic") print(f" EURUSD dyn fixed: final ${eu_dyn_fixed['balance'].iloc[-1]:,.0f}, " f"max DD {abs(eu_dyn_fixed['drawdown_pct'].min()):.2f}%") print(f" USDJPY dyn fixed: final ${uj_dyn_fixed['balance'].iloc[-1]:,.0f}, " f"max DD {abs(uj_dyn_fixed['drawdown_pct'].min()):.2f}%") print("Computing per-year DD view (FTMO fresh-account perspective)...") yearly_dd_fixed = compute_yearly_dd_view(eu_df, uj_df, pnl_col="pnl") yearly_dd_dynamic = compute_yearly_dd_view(eu_df, uj_df, pnl_col="pnl_dynamic") worst_fix = yearly_dd_fixed.loc[yearly_dd_fixed["dd_pct"].idxmin()] worst_dyn = yearly_dd_dynamic.loc[yearly_dd_dynamic["dd_pct"].idxmin()] print(f" Worst year (1.0×): {int(worst_fix['year'])} " f"DD {worst_fix['dd_pct']:.2f}% (${worst_fix['dd_dollar']:,.0f})") print(f" Worst year (dyn): {int(worst_dyn['year'])} " f"DD {worst_dyn['dd_pct']:.2f}% (${worst_dyn['dd_dollar']:,.0f})") print() print("Computing daily P&L records (for monthly drilldown chart)...") print(" Loading live trades (if available)...") eu_live = load_live_trades("EURUSD") uj_live = load_live_trades("USDJPY") daily_pnl_fixed = compute_daily_pnl_records(eu_df, uj_df, pnl_col="pnl", eu_live=eu_live, uj_live=uj_live) daily_pnl_dynamic = compute_daily_pnl_records(eu_df, uj_df, pnl_col="pnl_dynamic", eu_live=eu_live, uj_live=uj_live) n_live = len(eu_live) + len(uj_live) print(f" Active trading days: {len(daily_pnl_fixed['dates']):,} " f"(historic + {n_live} live trade{'s' if n_live != 1 else ''})") print() print("Computing metrics...") metrics = compute_metrics(eu_df, uj_df, daily_fixed, daily_comp, eu_fixed, eu_comp, uj_fixed, uj_comp, daily_dyn_fixed, daily_dyn_comp, yearly_dd_fixed, yearly_dd_dynamic) print("Writing dashboard JSON...") write_json(daily_fixed, daily_comp, metrics, eu_fixed, eu_comp, uj_fixed, uj_comp, daily_dyn_fixed, daily_dyn_comp, eu_dyn_fixed, eu_dyn_comp, uj_dyn_fixed, uj_dyn_comp, yearly_dd_fixed, yearly_dd_dynamic, daily_pnl_fixed, daily_pnl_dynamic) print("Writing summary report...") write_summary(metrics, daily_fixed, daily_comp) if __name__ == "__main__": try: main() except Exception as e: print(f"\n[ERROR] {e}") import traceback traceback.print_exc() sys.exit(1)