""" Refresh stale metric columns in MLP winner CSVs. A winner CSV stores the optimizer's params PLUS the metric columns recorded at promotion time. If the data is re-exported (or the simulation changes) after a winner is written, those metric columns go stale — they no longer reproduce when re-evaluated against the current data/weights/code. `mlp_results_table.py` recomputes live and is unaffected, but anything that reads the CSV metric columns directly (dashboards, commit-message tables, eyeballing the file) then shows wrong numbers and looks like a TV-parity failure. This tool recomputes the canonical IS-window metrics for the EXISTING params (params untouched) using the exact same path the sweep uses to record them (`load_scored_frame` -> `_score_to_signals` with `_pine_time_start=SCORE_START` -> filter `<= TRAIN_END` -> `calculate_metrics(score_start=SCORE_START)`), then rewrites only the metric columns. Dry-run by default. Use --apply to write. python3 tools/refresh_winner_metrics.py # dry run, BTC all TFs python3 tools/refresh_winner_metrics.py --apply # write python3 tools/refresh_winner_metrics.py --asset COINBASE_ETHUSD --tfs 4H 6H """ from __future__ import annotations import argparse import sys from pathlib import Path import pandas as pd REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) from config import SCORE_START, TRAIN_END # noqa: E402 import tools.run_mlp_deep_sweep as sweep # noqa: E402 # Numeric param cols that arrive from csv.DictReader as strings and must be # coerced before they reach _score_to_signals / calculate_metrics. _FLOAT_PARAMS = ( "i_long_entry_activation_threshold", "i_long_exit_activation_threshold", "i_long_exit_activation_confirmation_threshold", "i_trailing_stop_threshold", "i_regime_entry_min_score", ) _INT_PARAMS = ( "i_regime_window", "i_exit_score_window", "i_entry_score_window", "i_use_long_exit_confirmation", ) def _coerce(params: dict) -> dict: p = dict(params) for k in _FLOAT_PARAMS: if k in p and p[k] != "": p[k] = float(p[k]) for k in _INT_PARAMS: if k in p and p[k] != "": p[k] = int(float(p[k])) # bools already coerced by sweep.load_winner; mlp_weights_file stays a str return p def recompute_metrics(tf: str, params: dict) -> dict: """Canonical IS-window metric recompute — mirrors sweep.evaluate() steps 1-5.""" weights = params["mlp_weights_file"] scored, stoch_peak, close_arr, high_arr = sweep.load_scored_frame(tf, weights) signal_params = _coerce(params) signal_params["_pine_time_start"] = SCORE_START sigs = sweep._score_to_signals(scored.copy(), signal_params, stoch_peak, close_arr, high_arr) train_sigs = sigs[sigs["time"] <= pd.Timestamp(TRAIN_END)].copy() return sweep.calculate_metrics(train_sigs, score_start=SCORE_START) def _fmt(v) -> str: try: return f"{float(v):,.2f}" except (TypeError, ValueError): return str(v) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--asset", default="COINBASE_BTCUSD") ap.add_argument("--tfs", nargs="+", default=sweep.TFS) ap.add_argument("--apply", action="store_true", help="write changes (default: dry run)") ap.add_argument("--check", action="store_true", help="guard mode: never write; exit non-zero if any combo is stale " "(drift > threshold). Wire into pre-sweep / data-export checks.") ap.add_argument("--threshold", type=float, default=5.0, help="flag a combo as STALE if |Total P&L %% drift| exceeds this %% (relative)") args = ap.parse_args() if args.check: args.apply = False sweep.ASSET = args.asset # winner_path/data_path read this module global # --- Data-integrity preflight ---------------------------------------------- # A deficient/truncated TV export silently produces wrong metrics (the 34× # 4H drift on 2026-06-23). Refusing to refresh/guard on ERROR-level data # stops us from storing garbage. Warnings are allowed through. from tools.validate_chart_data import validate_file data_errors = False for tf in args.tfs: dp = sweep.data_path(tf) if not dp.exists(): continue errs, _warns, _ = validate_file(str(dp)) if errs: data_errors = True print(f" {tf:>4}: DATA INTEGRITY FAIL — {dp.name}") for e in errs: print(f" ERROR: {e}") if data_errors: print("\nAborting: chart data failed integrity checks. Re-export full, clean " "history from TradingView, then re-run. (tools/validate_chart_data.py)") return 1 any_stale = False print(f"Refreshing metric columns for {args.asset} ({'APPLY' if args.apply else 'dry run'})") print(f"Window: SCORE_START={SCORE_START} -> TRAIN_END={TRAIN_END}\n") for tf in args.tfs: wp = sweep.winner_path(tf) if not wp.exists(): print(f" {tf:>4}: no winner CSV — skip") continue row = sweep.load_winner(tf) if row is None: print(f" {tf:>4}: empty winner CSV — skip") continue try: new = recompute_metrics(tf, row) except Exception as exc: # noqa: BLE001 print(f" {tf:>4}: ERROR recomputing ({exc})") any_stale = True continue old_pnl = float(row.get("Total P&L %", 0.0) or 0.0) new_pnl = float(new.get("Total P&L %", 0.0) or 0.0) drift = abs(new_pnl - old_pnl) / max(1.0, abs(old_pnl)) * 100.0 stale = drift > args.threshold any_stale = any_stale or stale flag = " <-- STALE" if stale else "" print(f" {tf:>4}: P&L%% {_fmt(old_pnl):>14} -> {_fmt(new_pnl):>14} " f"(drift {drift:5.1f}%) trades {row.get('Total Trades')}->{int(new.get('Total Trades',0))}" f" Calmar {_fmt(row.get('Calmar Ratio'))}->{_fmt(new.get('Calmar Ratio'))}{flag}") if args.apply: merged = dict(row) for k in sweep.METRIC_COLS: if k in new: merged[k] = new[k] sweep.write_winner(tf, merged) print() if args.check: if any_stale: print("FAIL: stale winner metrics detected (drift > threshold). " "Run `refresh_winner_metrics.py --apply` after confirming data is current.") return 1 print("OK: all winner metrics reproduce within threshold.") return 0 if args.apply: print("Applied. Re-run mlp_results_table.py to confirm against TV.") else: print("Dry run only. Re-run with --apply to write.") if any_stale: print("STALE combos detected (drift > threshold).") return 0 if __name__ == "__main__": raise SystemExit(main())