""" Backfill sweep_database.db from existing GPU sweep CSVs in results/sweeps/. For each sweep CSV that has no DB rows yet, this script: 1. CPU-verifies the top 5000 results using the current strategy code 2. Computes composite scores and subperiod consistency 3. Writes passing rows to sweep_database.db Idempotent: skips combos that already have DB rows (checks by asset+timeframe count). Use --force to re-verify and overwrite existing rows for a combo. Usage: python3 tools/backfill_sweep_db.py # all missing combos python3 tools/backfill_sweep_db.py --tfs 12H 8H 1D # specific TFs python3 tools/backfill_sweep_db.py --force # re-verify even if rows exist """ import os, sys, io, re, sqlite3, datetime, argparse sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pandas as pd from concurrent.futures import ProcessPoolExecutor, as_completed from config import ( TRAIN_START, TRAIN_END, SCORE_START, SUBPERIOD_SPLIT, WEIGHT_COLS, SWEEPS_DIR, composite_score as _composite_score, mlp_data_path, ) SWEEP_DB_FILE = "results/sweep_database.db" DB_SCHEMA_VER = 3 TOP_N = 5000 _NON_WEIGHT_PARAM_COLS = { "i_long_entry_activation_threshold", "i_long_exit_activation_threshold", "i_long_exit_activation_confirmation_threshold", "i_use_long_exit_confirmation", "i_use_long_entry_confirmation", "i_trailing_stop_threshold", "i_m3_momentum_period", "i_cs_body_quality_ratio", "i_cs_confidence_scaling_factor", "i_cs_bull_eng_scaling_factor", "i_regime_window", "i_regime_entry_min_score", "i_mvrv_suppress_bear", "i_div_window", } _PARAM_COLS = set(WEIGHT_COLS) | _NON_WEIGHT_PARAM_COLS _METRIC_TO_COL = { "Total P&L %": "total_pnl_pct", "Max Drawdown %": "max_drawdown", "Total Trades": "total_trades", "% In Market": "pct_in_market", "Calmar Ratio": "calmar_ratio", "Sortino Ratio": "sortino_ratio", "Sharpe Ratio": "sharpe_ratio", "P&L/DD Ratio": "pnl_dd_ratio", } def _verify_one(args): params_dict, df_data_bytes, score_start, split_ts_str = args import io as _io, pandas as _pd import strategies.strategy_activation_scores as _strat try: df = _pd.read_pickle(_io.BytesIO(df_data_bytes)) df_res = _strat.generate_signals(df, **params_dict) metrics = _strat.calculate_metrics(df_res, score_start=score_start) # Subperiod consistency: Calmar > 0 in both IS halves split_ts = _pd.to_datetime(split_ts_str) train_end = _pd.to_datetime(TRAIN_END) if 'time' in df.columns: df_p1 = df[df['time'] <= split_ts] df_p2 = df[(df['time'] > split_ts) & (df['time'] <= train_end)] consistent = True for part in [df_p1, df_p2]: if len(part) > 0: r1 = _strat.generate_signals(part.copy(), **params_dict) m1 = _strat.calculate_metrics(r1, score_start=score_start) if m1.get("Calmar Ratio", -10.0) <= 0: consistent = False break else: consistent = True result = params_dict.copy() result.update(metrics) result['subperiod_consistent'] = consistent return result except Exception: return None def get_db_counts(): if not os.path.exists(SWEEP_DB_FILE): return {} conn = sqlite3.connect(SWEEP_DB_FILE) rows = conn.execute( "SELECT asset, timeframe, COUNT(*) FROM sweep_results GROUP BY asset, timeframe" ).fetchall() conn.close() return {(r[0], r[1]): r[2] for r in rows} def write_results(verified_results, asset, timeframe, run_id): if not verified_results: return 0 positive = [r for r in verified_results if r.get("Total Trades", 0) > 0 and r.get("Calmar Ratio", -10.0) > 0] if not positive: return 0 scores = sorted(r.get("Calmar Ratio", 0.0) for r in positive) n = max(len(scores) - 1, 1) for r in positive: s = r.get("Calmar Ratio", 0.0) idx = scores.index(s) r["_pnl_dd_percentile"] = 100.0 * idx / n now = datetime.datetime.now().isoformat(timespec="seconds") rows_to_insert = [] for r in positive: row = { "asset": asset, "timeframe": timeframe, "schema_ver": DB_SCHEMA_VER, "recorded_at": now, "run_id": run_id, "train_start": TRAIN_START, "train_end": TRAIN_END, "score_start": SCORE_START, "pnl_dd_percentile": r.get("_pnl_dd_percentile"), "iteration_number": None, "gpu_score": r.get("GPU_Score"), "composite_score": r.get("Composite"), "subperiod_consistent": (1 if r.get("subperiod_consistent") else 0) if "subperiod_consistent" in r else None, } for src, col in _METRIC_TO_COL.items(): val = r.get(src) row[col] = float(val) if val is not None else None for col in _PARAM_COLS: val = r.get(col) row[col] = float(val) if val is not None else None rows_to_insert.append(row) cols = list(rows_to_insert[0].keys()) placeholders = ", ".join("?" * len(cols)) col_names = ", ".join(cols) sql = f"INSERT INTO sweep_results ({col_names}) VALUES ({placeholders})" values = [tuple(row[c] for c in cols) for row in rows_to_insert] conn = sqlite3.connect(SWEEP_DB_FILE) # Ensure all columns exist (migration) existing = {r[1] for r in conn.execute("PRAGMA table_info(sweep_results)").fetchall()} for col in cols: if col not in existing and col not in ("id",): try: conn.execute(f"ALTER TABLE sweep_results ADD COLUMN {col} REAL") except Exception: pass conn.executemany(sql, values) conn.commit() conn.close() return len(rows_to_insert) def backfill_combo(asset, tf, sweep_csv, force=False): data_csv = mlp_data_path(asset, tf) if not os.path.exists(data_csv): print(f" SKIP {asset}-{tf}: data CSV not found ({data_csv})") return db_counts = get_db_counts() existing = db_counts.get((asset, tf), 0) if existing > 0 and not force: print(f" SKIP {asset}-{tf}: already has {existing:,} rows in DB (use --force to overwrite)") return print(f"\n{'='*60}") print(f" {asset}-{tf} [{sweep_csv}]") print(f"{'='*60}") df_sweep = pd.read_csv(sweep_csv) if df_sweep.empty: print(" SKIP: empty sweep CSV") return df_sweep = df_sweep.sort_values("Calmar Ratio", ascending=False).head(TOP_N) print(f" Loaded {len(df_sweep)} candidates (top {TOP_N} by GPU Calmar)") df_data = pd.read_csv(data_csv) df_data.columns = df_data.columns.str.lower() if 'time' in df_data.columns: df_data['time'] = pd.to_datetime(df_data['time'], utc=True).dt.tz_localize(None) mask = (df_data['time'] >= TRAIN_START) & (df_data['time'] <= TRAIN_END) df_data = df_data.loc[mask].copy() print(f" IS data rows: {len(df_data)}") buf = io.BytesIO() df_data.to_pickle(buf) df_data_bytes = buf.getvalue() candidate_params = [row.to_dict() for _, row in df_sweep.iterrows()] worker_args = [(p, df_data_bytes, SCORE_START, SUBPERIOD_SPLIT) for p in candidate_params] cpu_count = os.cpu_count() or 4 max_workers = max(1, min(6, cpu_count - 4)) print(f" Verifying {len(candidate_params)} candidates with {max_workers} workers...") # Compute is_years from scoring window df_score = df_data[df_data['time'] >= pd.to_datetime(SCORE_START)] if 'time' in df_data.columns else df_data is_years = 8.0 if len(df_score) > 1: t0 = df_score['time'].iloc[0] t1 = df_score['time'].iloc[-1] is_years = max(0.5, (t1 - t0).days / 365.25) print(f" IS years: {is_years:.2f}") verified_results = [] with ProcessPoolExecutor(max_workers=max_workers) as executor: futures = {executor.submit(_verify_one, a): a[0] for a in worker_args} for fut in as_completed(futures): result = fut.result() if result is not None: orig = futures[fut] result['GPU_Score'] = orig.get("Calmar Ratio", 0) calmar = result.get("Calmar Ratio", 0.0) sortino = result.get("Sortino Ratio", 0.0) trades = int(result.get("Total Trades", 0)) pnl = result.get("Total P&L %", 0.0) composite = _composite_score(calmar, sortino, trades, is_years, pnl) if not result.get('subperiod_consistent', True): composite = 0.0 result['Composite'] = composite verified_results.append(result) passing = [r for r in verified_results if r.get("Calmar Ratio", -10.0) > 0] print(f" CPU verified: {len(verified_results)} returned results, {len(passing)} with Calmar > 0") if force and existing > 0: conn = sqlite3.connect(SWEEP_DB_FILE) conn.execute("DELETE FROM sweep_results WHERE asset=? AND timeframe=?", (asset, tf)) conn.commit() conn.close() print(f" Deleted {existing} existing rows") run_id = f"backfill_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}_{tf}" written = write_results(verified_results, asset, tf, run_id) print(f" Written {written} rows to DB") def main(): parser = argparse.ArgumentParser(description="Backfill sweep DB from GPU sweep CSVs") parser.add_argument("--tfs", nargs="*", metavar="TF", help="Timeframes to process (default: all found in results/sweeps/)") parser.add_argument("--asset", type=str, default="COINBASE_BTCUSD", help="Asset to backfill (default: COINBASE_BTCUSD)") parser.add_argument("--force", action="store_true", help="Re-verify and overwrite even if rows already exist") args = parser.parse_args() pattern = re.compile( rf"optimization_sweep_activation_scores_{re.escape(args.asset)}_(\w+)\.csv" ) sweep_files = {} for fname in os.listdir(SWEEPS_DIR): m = pattern.match(fname) if m: tf = m.group(1) sweep_files[tf] = os.path.join(SWEEPS_DIR, fname) if not sweep_files: print(f"No sweep CSVs found for {args.asset} in {SWEEPS_DIR}/") return tfs_to_run = [tf.upper() for tf in args.tfs] if args.tfs else sorted(sweep_files.keys()) missing = [tf for tf in tfs_to_run if tf not in sweep_files] if missing: print(f"No sweep CSV found for: {missing}") print(f"Processing: {args.asset} TFs: {tfs_to_run}") print(f"DB: {SWEEP_DB_FILE}") print(f"Current DB counts: {get_db_counts()}") for tf in tfs_to_run: if tf in sweep_files: backfill_combo(args.asset, tf, sweep_files[tf], force=args.force) print(f"\nDone. Final DB counts: {get_db_counts()}") if __name__ == "__main__": main()