""" Recover best WFO winners from sweep DB for all BTC timeframes. Usage: python3 tools/recover_winners_from_db.py [--dry-run] For each TF, finds the DB row with the best WFO score (best mean OOS Calmar across folds), re-runs WFO evaluation to get fold-level details, then writes the winner CSV, cheatsheet, and Pine snippet if the DB result beats the current winner file. Run this after discovering that the winner files were downgraded by the IS-composite/WFO mismatch bug in auto_optimize_loop.py. """ import argparse import io import os import sys import pandas as pd import sqlite3 # Ensure project root is on path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import ( WINNERS_DIR, SCORE_START, TRAIN_START, TRAIN_END, WFO_FOLDS, WFO_MIN_OOS_TRADES, WFO_MIN_VALID_FOLDS, composite_score as _composite_score, ) from tools.auto_optimize_loop import _wfo_score_one, _verify_one, write_winner_readable DB_PATH = "results/sweep_database.db" STRATEGY_NAME = "activation_scores" ASSET = "COINBASE_BTCUSD" TIMEFRAMES = ["4H", "6H", "8H", "12H", "1D"] DATA_FILES = { "4H": "data/mlp/COINBASE_BTCUSD, 240.csv", "6H": "data/mlp/COINBASE_BTCUSD, 360.csv", "8H": "data/mlp/COINBASE_BTCUSD, 480.csv", "12H": "data/mlp/COINBASE_BTCUSD, 720.csv", "1D": "data/mlp/COINBASE_BTCUSD, 1D.csv", } # DB column → winner dict key mapping for metric columns _DB_TO_WINNER = { "calmar_ratio": "Calmar Ratio", "sortino_ratio": "Sortino Ratio", "sharpe_ratio": "Sharpe Ratio", "total_trades": "Total Trades", "total_pnl_pct": "Total P&L %", "max_drawdown": "Max Drawdown %", "pct_in_market": "% In Market", "composite_score": "Composite", "wfo_score": "WFO_Score", "pnl_dd_ratio": "P&L/DD Ratio", "gpu_score": "GPU_Score", "subperiod_consistent": "subperiod_consistent", } def load_data(data_file): df = pd.read_csv(data_file) df.columns = df.columns.str.lower() if "time" in df.columns: df["time"] = pd.to_datetime(df["time"], utc=True).dt.tz_localize(None) mask = (df["time"] >= TRAIN_START) & (df["time"] <= TRAIN_END) df = df.loc[mask].copy() return df def db_row_to_params_dict(row: dict) -> dict: """Extract only i_* param keys from a DB row dict.""" return {k: v for k, v in row.items() if k.startswith("i_")} def current_winner_wfo_score(tf: str) -> float: """Return current winner WFO_Score from CSV, or -inf if missing.""" path = os.path.join(WINNERS_DIR, f"optimization_winner_{STRATEGY_NAME}_{ASSET}_{tf}.csv") if not os.path.exists(path): return float("-inf") try: df = pd.read_csv(path) return float(df.iloc[0].get("WFO_Score", -float("inf"))) except Exception: return float("-inf") def current_winner_wfo_min(tf: str) -> float: """Return current winner WFO_Min_Fold from CSV, or -inf if missing.""" path = os.path.join(WINNERS_DIR, f"optimization_winner_{STRATEGY_NAME}_{ASSET}_{tf}.csv") if not os.path.exists(path): return float("-inf") try: df = pd.read_csv(path) return float(df.iloc[0].get("WFO_Min_Fold", -float("inf"))) except Exception: return float("-inf") def recover_tf(tf: str, dry_run: bool) -> None: print(f"\n{'='*60}") print(f" {tf}") print(f"{'='*60}") data_file = DATA_FILES[tf] if not os.path.exists(data_file): print(f" [SKIP] Data file not found: {data_file}") return # --- Query DB for best WFO row --- conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute( """SELECT * FROM sweep_results WHERE asset=? AND timeframe=? AND schema_ver=3 AND wfo_score > 0 ORDER BY wfo_score DESC LIMIT 1""", (ASSET, tf), ) db_row = cur.fetchone() conn.close() if db_row is None: print(f" [SKIP] No WFO rows in DB for {tf}") return db_dict = dict(db_row) db_wfo_score = float(db_dict.get("wfo_score", 0.0)) db_composite = float(db_dict.get("composite_score", 0.0)) cur_wfo_score = current_winner_wfo_score(tf) cur_wfo_min = current_winner_wfo_min(tf) print(f" DB best WFO Score: {db_wfo_score:.4f} (Composite={db_composite:.4f})") print(f" Current winner WFO: {cur_wfo_score:.4f} (WFO_Min={cur_wfo_min:.4f})") if db_wfo_score <= cur_wfo_score: print(f" [OK] Current winner already has best or equal WFO score. No action needed.") return print(f" [RECOVER] DB has better WFO (+{db_wfo_score - cur_wfo_score:.4f}). Rebuilding winner...") # --- Load and prepare data --- df_data = load_data(data_file) is_years = 8.0 try: df_sw = df_data[df_data["time"] >= pd.to_datetime(SCORE_START)] if len(df_sw) > 1: is_years = max(0.5, (df_sw["time"].iloc[-1] - df_sw["time"].iloc[0]).days / 365.25) except Exception: pass buf = io.BytesIO() df_data.to_pickle(buf) df_bytes = buf.getvalue() # --- Build params dict from DB row --- params = db_row_to_params_dict(db_dict) # --- CPU verify to get accurate IS metrics --- print(f" Running CPU verification...") verify_result = _verify_one((params, df_bytes, SCORE_START, "Composite")) if verify_result is None: print(f" [ERROR] CPU verification failed. Aborting {tf}.") return calmar = verify_result.get("Calmar Ratio", 0.0) sortino = verify_result.get("Sortino Ratio", 0.0) trades = int(verify_result.get("Total Trades", 0)) pnl = verify_result.get("Total P&L %", 0.0) composite = _composite_score(calmar, sortino, trades, is_years, pnl) if not verify_result.get("subperiod_consistent", True): composite = 0.0 verify_result["Composite"] = composite verify_result["CPU_Score"] = verify_result.get("Composite", 0.0) verify_result["GPU_Score"] = db_dict.get("gpu_score", 0.0) verify_result["verified"] = True print(f" CPU IS: Composite={composite:.4f} Calmar={calmar:.4f} Sortino={sortino:.4f} Trades={trades}") # --- Re-run WFO to get fold details --- print(f" Running WFO fold evaluation...") wfo_score, fold_calmars, wfo_min, wfo_neg = _wfo_score_one( (params, df_bytes, WFO_FOLDS, WFO_MIN_OOS_TRADES, WFO_MIN_VALID_FOLDS) ) print(f" WFO: score={wfo_score:.4f} min={wfo_min:.4f} neg_folds={wfo_neg}") print(f" Fold Calmars: {[round(c, 4) for c in fold_calmars]}") if wfo_score <= 0: print(f" [WARN] Re-run WFO score is 0 (not enough qualifying folds). " f"Proceeding with DB wfo_score={db_wfo_score:.4f} and fold details unavailable.") wfo_score = db_wfo_score # --- Build complete winner row --- winner_row = verify_result.copy() winner_row.update(params) winner_row["WFO_Score"] = wfo_score winner_row["WFO_Fold_Calmars"] = fold_calmars winner_row["WFO_Min_Fold"] = wfo_min winner_row["WFO_Neg_Folds"] = wfo_neg winner_row["_pnl_dd_percentile"] = float(db_dict.get("pnl_dd_percentile", 0.0)) if dry_run: print(f"\n [DRY RUN] Would write winner for {tf}:") print(f" Composite={composite:.4f} WFO={wfo_score:.4f} WFO_Min={wfo_min:.4f}") return # --- Write winner files --- winner_csv_path = os.path.join(WINNERS_DIR, f"optimization_winner_{STRATEGY_NAME}_{ASSET}_{tf}.csv") pd.DataFrame([winner_row]).to_csv(winner_csv_path, index=False) write_winner_readable(winner_row, STRATEGY_NAME, WINNERS_DIR, asset=ASSET, timeframe=tf) # Also update _wfo.csv wfo_csv_path = os.path.join(WINNERS_DIR, f"optimization_winner_{STRATEGY_NAME}_{ASSET}_{tf}_wfo.csv") pd.DataFrame([winner_row]).to_csv(wfo_csv_path, index=False) print(f" [DONE] Winner files updated for {ASSET} {tf}") def main(): parser = argparse.ArgumentParser(description="Recover best WFO winners from sweep DB") parser.add_argument("--dry-run", action="store_true", help="Show what would change without writing files") parser.add_argument("--tf", nargs="+", choices=TIMEFRAMES, default=TIMEFRAMES, help="Timeframes to check") args = parser.parse_args() print(f"\nRecover Winners from DB {'[DRY RUN]' if args.dry_run else ''}") print(f"Checking TFs: {args.tf}") for tf in args.tf: recover_tf(tf, args.dry_run) print(f"\n{'='*60}") print("Done. Re-run tools/oos_dashboard.py to refresh the dashboard.") if __name__ == "__main__": main()