""" Out-of-Sample Performance Dashboard ===================================== Reads all winner CSVs from results/ and runs each through the full data (in-sample + OOS) to produce a combined dashboard at results/oos_dashboard.md. Usage: python3 tools/oos_dashboard.py # all assets/TFs found in results/ python3 tools/oos_dashboard.py --asset COINBASE_BTCUSD python3 tools/oos_dashboard.py --asset COINBASE_BTCUSD --timeframe 4H """ import argparse import os import sys import re from datetime import datetime import math import pandas as pd import warnings warnings.filterwarnings("ignore") sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import strategies.strategy_activation_scores as strategy_module from config import SCORE_START, TRAIN_END, OOS_START, RESULTS_DIR, WINNERS_DIR DATA_DIR = "data" OUTPUT_FILE = "results/oos_dashboard.md" # Used for dynamic few-trades threshold: IS window length in years IS_YEARS = (pd.Timestamp(TRAIN_END) - pd.Timestamp(SCORE_START)).days / 365.25 OOS_YEARS = (datetime.now() - pd.Timestamp(OOS_START)).days / 365.25 # Penalty sentinel returned by calculate_metrics when trade count < 30 _SORTINO_PENALTY = -10.0 # TradingView exports use minute counts for sub-daily timeframes. _TF_TO_MINUTES = {"4H": "240", "6H": "360", "8H": "480", "12H": "720"} def _find_data_file(asset: str, tf: str) -> str | None: """Resolve data file path supporting both dash and TradingView comma-space naming. Tries in order: 1. data/{asset}-{tf}.csv (legacy dash format) 2. data/{asset}, {tf}.csv (TV export with TF label, e.g. '1D') 3. data/{asset}, {minutes}.csv (TV export with minute count, e.g. '360' for 6H) """ dash = os.path.join(DATA_DIR, f"{asset}-{tf}.csv") if os.path.exists(dash): return dash comma_tf = os.path.join(DATA_DIR, f"{asset}, {tf}.csv") if os.path.exists(comma_tf): return comma_tf minutes = _TF_TO_MINUTES.get(tf) if minutes: comma_min = os.path.join(DATA_DIR, f"{asset}, {minutes}.csv") if os.path.exists(comma_min): return comma_min return None def _oos_regime_context(df_full, oos_start_ts): """Return (bear_pct, neutral_pct, bull_pct, label) for the OOS window. label examples: '99% Bear/1% Neutral', '60% Bull', None if column missing. """ if 'mvrv_regime' not in df_full.columns: return None, None, None, None df_oos = df_full[df_full['time'] >= oos_start_ts] if len(df_oos) == 0: return None, None, None, None total = len(df_oos) bear_pct = float((df_oos['mvrv_regime'] == -1).sum() / total * 100) neutral_pct = float((df_oos['mvrv_regime'] == 0).sum() / total * 100) bull_pct = float((df_oos['mvrv_regime'] == 1).sum() / total * 100) if bear_pct >= 5 and neutral_pct >= 5: label = f"{bear_pct:.0f}% Bear/{neutral_pct:.0f}% Neutral" elif bear_pct + neutral_pct >= 60: label = f"{bear_pct + neutral_pct:.0f}% Bear+Neutral" else: label = f"{bull_pct:.0f}% Bull" return bear_pct, neutral_pct, bull_pct, label def _parse_fold_calmars(raw): """Parse WFO_Fold_Calmars string → compact '[0.86, 1.29, ...]' or '—'. Handles both 'np.float64(0.86)' and plain '[0.86, 1.29]' formats.""" if raw is None or (isinstance(raw, float) and math.isnan(raw)): return "—" # Match decimal numbers only (Calmars always have a decimal point; this avoids matching '64' # from 'np.float64' type annotations in older numpy repr strings). vals = re.findall(r"-?\d+\.\d+(?:[eE][-+]?\d+)?", str(raw)) if not vals: return "—" return "[" + ", ".join(f"{float(v):.2f}" for v in vals) + "]" def _fmt_wfo(val, fmt=".4f"): """Format a WFO scalar value, returning '—' if missing/NaN.""" if val is None or (isinstance(val, float) and math.isnan(val)): return "—" try: return format(float(val), fmt) except (TypeError, ValueError): return "—" def _freshness_label(dt): """🟢 < 24 h | 🟡 1–5 d | 🔴 > 5 d""" age = datetime.now() - dt hours = age.total_seconds() / 3600 days = age.days if hours < 1: return "🟢 just now" elif hours < 24: return f"🟢 {int(hours)}h ago" elif days < 5: return f"🟡 {days}d ago" else: return f"🔴 {days}d ago" def find_winner_files(asset_filter=None, tf_filter=None): pattern = re.compile( r"optimization_winner_activation_scores_([A-Z0-9_]+)_([A-Z0-9HD]+)\.csv$" ) matches = [] for fname in sorted(os.listdir(WINNERS_DIR)): m = pattern.match(fname) if not m: continue asset, tf = m.group(1), m.group(2) if asset_filter and asset != asset_filter: continue if tf_filter and tf != tf_filter: continue matches.append((asset, tf, os.path.join(WINNERS_DIR, fname))) return matches def run_combo(asset, tf, winner_file): data_file = _find_data_file(asset, tf) if not data_file: return None, f" {asset} {tf}: data file missing (tried data/{asset}-{tf}.csv and comma variants)" row = pd.read_csv(winner_file).iloc[0] params = row.to_dict() is_composite = float(row.get("Composite", float('nan'))) wfo_score_d = _fmt_wfo(row.get("WFO_Score")) wfo_min_d = _fmt_wfo(row.get("WFO_Min_Fold")) wfo_neg_raw = row.get("WFO_Neg_Folds") wfo_neg_d = str(int(wfo_neg_raw)) if wfo_neg_raw is not None and not (isinstance(wfo_neg_raw, float) and math.isnan(wfo_neg_raw)) else "—" wfo_folds = _parse_fold_calmars(row.get("WFO_Fold_Calmars")) # Winner file modification date — shows when params were last updated winner_date = datetime.fromtimestamp(os.path.getmtime(winner_file)).strftime('%m-%d') df_full = pd.read_csv(data_file) df_full.columns = df_full.columns.str.lower() if 'time' in df_full.columns: df_full['time'] = pd.to_datetime(df_full['time'], utc=True).dt.tz_localize(None) df_signals = strategy_module.generate_signals(df_full.copy(), **params) train_end_ts = pd.to_datetime(TRAIN_END) df_is = df_signals[df_signals['time'] <= train_end_ts].copy() is_m = strategy_module.calculate_metrics(df_is, score_start=SCORE_START) itr = int(is_m.get('Total Trades', 0)) # Compute scaled OOS minimum before calling calculate_metrics so we can pass it in, # bypassing the internal MIN_SCORABLE_TRADES (10) for short OOS windows. min_oos_trades = max(3, int(itr / IS_YEARS * OOS_YEARS * 0.8)) if itr > 0 else 3 oos_m = strategy_module.calculate_metrics(df_signals, score_start=OOS_START, min_trades=min_oos_trades) # 0 OOS trades is neutral for a long-only strategy; replace penalty sentinel with 0.0. if int(oos_m.get('Total Trades', 0)) == 0: for _k in ('Calmar Ratio', 'Sortino Ratio', 'Sharpe Ratio'): if oos_m.get(_k) == _SORTINO_PENALTY: oos_m[_k] = 0.0 oos_last = df_full['time'].iloc[-1].strftime('%Y-%m-%d') oos_start_ts = pd.to_datetime(OOS_START) oos_bear_pct, oos_neutral_pct, oos_bull_pct, oos_regime_label = _oos_regime_context(df_full, oos_start_ts) df_oos_raw = df_full[df_full['time'] >= oos_start_ts] if len(df_oos_raw) >= 2 and df_oos_raw.iloc[0]['open'] > 0: bh_pnl = (df_oos_raw.iloc[-1]['close'] / df_oos_raw.iloc[0]['open'] - 1) * 100 else: bh_pnl = None iso = is_m.get('Sortino Ratio', 0) oso = oos_m.get('Sortino Ratio', 0) otr = int(oos_m.get('Total Trades', 0)) oos_calmar = oos_m.get('Calmar Ratio', -10.0) # When calculate_metrics returns the penalty sentinel, values are meaningless — display as "—" oso_display = "—" if oso == _SORTINO_PENALTY else f"{oso:.4f}" calmar_display = "—" if oos_calmar == _SORTINO_PENALTY else f"{oos_calmar:.2f}" oos_pnl_val = oos_m.get('Total P&L %', 0) pnl_str = f"+{oos_pnl_val:.1f}%" if oos_pnl_val >= 0 else f"{oos_pnl_val:.1f}%" if itr == 0 or iso <= 0: verdict_short = "⚠ N/A" elif otr == 0: # Distinguish regime-adverse (neutral) from opportunity-rich (suspicious). adverse = (oos_bear_pct is not None) and ((oos_bear_pct + oos_neutral_pct) >= 60) if oos_regime_label and adverse: verdict_short = f"⚠ No trades — {oos_regime_label} (neutral)" elif oos_regime_label: verdict_short = f"⚠ No trades — {oos_regime_label} (suspicious)" else: verdict_short = "⚠ No OOS trades" elif otr < min_oos_trades: verdict_short = f"⚠ FEW TRADES ({otr}) P&L={pnl_str}" elif oso == _SORTINO_PENALTY: # Penalty fired for a reason other than trade count (e.g. threshold proximity) verdict_short = f"⚠ PENALTY ({otr} trades) P&L={pnl_str}" elif oso < 0: verdict_short = f"❌ NEGATIVE OOS (Sortino={oso:.2f})" else: ratio = oso / iso if ratio >= 0.8: verdict_short = f"✅ EXCELLENT ({ratio:.0%})" elif ratio >= 0.5: verdict_short = f"✅ ACCEPT ({ratio:.0%})" elif ratio >= 0.25: verdict_short = f"⚠ WEAK ({ratio:.0%})" else: verdict_short = f"❌ POOR ({ratio:.0%})" return { "asset": asset, "tf": tf, "winner_date": winner_date, "oos_last": oos_last, "is_composite": is_composite, "is_pnl": is_m.get('Total P&L %', 0), "is_dd": is_m.get('Max Drawdown %', 0), "is_sort": iso, "is_tr": itr, "wfo_score_d": wfo_score_d, "wfo_min_d": wfo_min_d, "wfo_neg_d": wfo_neg_d, "wfo_folds": wfo_folds, "oos_pnl": oos_pnl_val, "bh_pnl": bh_pnl, "oos_dd": oos_m.get('Max Drawdown %', 0), "oos_sort": oso, "oos_sort_display": oso_display, "oos_calmar_display": calmar_display, "oos_tr": otr, "min_oos_trades": min_oos_trades, "oos_regime_label": oos_regime_label or "—", "verdict": verdict_short, }, None def main(): parser = argparse.ArgumentParser( description="OOS performance dashboard across all optimized strategies", epilog=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--asset", help="Filter to specific asset") parser.add_argument("--timeframe", help="Filter to specific timeframe") args = parser.parse_args() combos = find_winner_files(args.asset, args.timeframe) if not combos: print("No winner files found.") sys.exit(1) print(f"Found {len(combos)} winner file(s). Running OOS evaluation...\n") rows = [] for asset, tf, winner_file in combos: print(f" {asset} {tf}...", end=" ", flush=True) try: result, err = run_combo(asset, tf, winner_file) if err: print(err) continue rows.append(result) print(f"IS Sortino={result['is_sort']:.4f} OOS Sortino={result['oos_sort']:.4f} {result['verdict']}") except Exception as e: print(f"ERROR: {e}") if not rows: print("No results generated.") sys.exit(1) lines = [ "# Out-of-Sample Performance Dashboard", f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", f"Training window: {SCORE_START} → {TRAIN_END} | OOS window: {OOS_START} → (latest data)", "", "| Asset | TF | IS Composite | Updated | IS P&L | IS DD | IS Sortino | IS Tr | WFO Score | WFO Min | WFO Neg | WFO Folds | OOS P&L | BH P&L | OOS DD | OOS Sortino | OOS Calmar | OOS Tr | OOS Regime | OOS end | Verdict |", "|-------|-----|-------------|---------|--------|-------|------------|-------|-----------|---------|---------|-----------|---------|--------|--------|-------------|------------|--------|------------|---------|---------|", ] for r in rows: bh = r['bh_pnl'] bh_str = ("—" if bh is None else f"+{bh:.1f}%" if bh >= 0 else f"{bh:.1f}%") lines.append( f"| {r['asset']} | {r['tf']} | {r['is_composite']:.4f} | {r['winner_date']} " f"| {r['is_pnl']:,.1f}% | {r['is_dd']:.1f}% | {r['is_sort']:.4f} | {r['is_tr']} " f"| {r['wfo_score_d']} | {r['wfo_min_d']} | {r['wfo_neg_d']} | {r['wfo_folds']} " f"| {r['oos_pnl']:,.1f}% | {bh_str} | {r['oos_dd']:.1f}% | {r['oos_sort_display']} | {r['oos_calmar_display']} | {r['oos_tr']} " f"| {r['oos_regime_label']} | {r['oos_last']} | {r['verdict']} |" ) lines += [ "", "---", "**Verdict key:** ✅ EXCELLENT ≥80% of IS Sortino | ✅ ACCEPT ≥50% | ⚠ WEAK ≥25% | ❌ POOR <25% or negative", "**OOS Regime:** MVRV-based regime composition of the OOS window (Bear/Neutral = adverse for longs; Bull = opportunity-rich). " "When '0 OOS trades' in a Bear+Neutral market this is neutral/correct behavior, not failure.", "**WFO columns:** WFO Score = mean fold Calmar | WFO Min = worst-fold Calmar | WFO Neg = negative fold count | WFO Folds = per-fold Calmars", "*Generated by tools/oos_dashboard.py*", ] os.makedirs(RESULTS_DIR, exist_ok=True) with open(OUTPUT_FILE, "w") as f: f.write("\n".join(lines) + "\n") print(f"\n[Written to {OUTPUT_FILE}]") if __name__ == "__main__": main()