""" MLP Parity Checker — Python vs TradingView ========================================== Compares Python mlp_forward scores against the mlp_score column exported from TradingView, to verify that the Pine float literals and forward-pass loop reproduce the same values as the Python implementation. Pass criteria ------------- - max |Δ| < 0.01 (9-sig-digit literals + summation drift ≈ 1e-3 on ±1000 scale) - zero threshold-side disagreements (no bar where Python and TV sit on opposite sides of entry OR exit threshold) Warns on bars where |score − threshold| < 0.05 (fragile, could flip on next export). Usage ----- python tools/check_mlp_parity.py \\ --data data/COINBASE_BTCUSD-6H.csv \\ --weights strategies/params/mlp/mlp_weights_COINBASE_BTCUSD_6H.json \\ [--params results/winners/optimization_winner_strategy_mlp_scores_COINBASE_BTCUSD_6H.csv] \\ [--date 2024-01-15] # show bar-level detail around this date \\ [--max-delta N] # show top-N worst-delta bars (default 10) \\ [--crossunders-only] # only show entry/exit bars \\ [--export /tmp/parity.csv] """ import argparse import json import os import sys from pathlib import Path import numpy as np import pandas as pd REPO = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO)) from config import TRAIN_START from strategies.strategy_mlp_scores import ( load_mlp_artifact, mlp_forward, FEATURE_COLS, ) from strategies.strategy_activation_scores import _prepare_features as _prepare_features_base def _prepare_features(df, feature_cols=None): """Wrapper that uses artifact feature_cols when provided (MLP path).""" return _prepare_features_base(df, feature_cols=feature_cols) # ── Param loader (mirrors diagnose_tv_parity.py) ───────────────────────────── def load_params(params_path: str | None) -> dict: if not params_path or not os.path.exists(params_path): return {} if params_path.endswith(".csv"): row = pd.read_csv(params_path).iloc[0].to_dict() drop = { "Sharpe Ratio", "Sortino Ratio", "Calmar Ratio", "Total Trades", "Total P&L %", "P&L/DD Ratio", "Max Drawdown %", "% In Market", "verified", "GPU_Score", "Composite", "CPU_Score", "_pnl_dd_percentile", "composite_score", "subperiod_consistent", "WFO_Score", "WFO_Fold_Calmars", } return {k: v for k, v in row.items() if k not in drop} if params_path.endswith(".json"): with open(params_path) as f: raw = json.load(f) flat = {} for k, v in raw.items(): if isinstance(v, dict): flat[k] = v.get("values", [None])[0] if "values" in v else v.get("start", 0) else: flat[k] = v return flat return {} # ── Core comparison ─────────────────────────────────────────────────────────── def run_parity(data_path: str, weights_path: str, params: dict) -> pd.DataFrame: """ Returns a DataFrame with one row per bar (from TRAIN_START onward) containing: py_score — Python mlp_forward score tv_score — TV-exported mlp_score column (NaN if column missing) delta — py_score - tv_score execute_entry, execute_exit — Python signals """ art = load_mlp_artifact(weights_path) layers = art["layers"] df = pd.read_csv(data_path) df.columns = df.columns.str.lower().str.strip() df["time"] = pd.to_datetime(df["time"], utc=True).dt.tz_localize(None) df = df.loc[df["time"] >= pd.Timestamp(TRAIN_START)].copy().reset_index(drop=True) # Python MLP scores — use artifact's feature_cols so col count matches weights art_cols = art["feature_cols"] X = _prepare_features(df, feature_cols=art_cols) # (T, F) py_scores = mlp_forward(X, layers) # (T,) # TV exported score (column name may vary slightly) tv_col = next( (c for c in df.columns if c in ("mlp_score", "mlp_score_poc")), None, ) tv_scores = df[tv_col].values if tv_col else np.full(len(df), np.nan) delta = py_scores - tv_scores # Python signals (entry/exit) from generate_signals from strategies.strategy_mlp_scores import generate_signals sigs = generate_signals(df.copy(), **params) result = df[["time"]].copy() result["py_score"] = py_scores result["tv_score"] = tv_scores result["delta"] = delta result["execute_entry"] = sigs.get("execute_entry", pd.Series(False, index=df.index)) result["execute_exit"] = sigs.get("execute_exit", pd.Series(False, index=df.index)) return result # ── Threshold-side disagreement check ──────────────────────────────────────── def threshold_disagreements(result: pd.DataFrame, params: dict) -> pd.DataFrame: """ Returns rows where Python and TV sit on opposite sides of entry OR exit threshold. A NaN tv_score is skipped (TV column not present). """ entry_thr = float(params.get("i_long_entry_activation_threshold", 0.0)) exit_thr = float(params.get("i_long_exit_activation_threshold", 0.0)) valid = result.dropna(subset=["tv_score"]) # Side = True means score >= threshold (above) py_above_entry = valid["py_score"] >= entry_thr tv_above_entry = valid["tv_score"] >= entry_thr py_above_exit = valid["py_score"] >= exit_thr tv_above_exit = valid["tv_score"] >= exit_thr disagree = (py_above_entry != tv_above_entry) | (py_above_exit != tv_above_exit) return valid[disagree].copy() # ── Fragile-bar warning ─────────────────────────────────────────────────────── def fragile_bars(result: pd.DataFrame, params: dict, margin: float = 0.05) -> pd.DataFrame: """Bars where |py_score - threshold| < margin for any threshold.""" entry_thr = float(params.get("i_long_entry_activation_threshold", 0.0)) exit_thr = float(params.get("i_long_exit_activation_threshold", 0.0)) conf_thr = float(params.get("i_long_exit_activation_confirmation_threshold", 0.0)) near = ( (np.abs(result["py_score"] - entry_thr) < margin) | (np.abs(result["py_score"] - exit_thr) < margin) | (np.abs(result["py_score"] - conf_thr) < margin) ) return result[near].copy() # ── Auto-detection helpers ──────────────────────────────────────────────────── _TF_TO_MINUTES = {"4H": "240", "6H": "360", "8H": "480", "12H": "720", "1D": "1D"} def _auto_resolve(asset: str, tf: str) -> tuple[str, str, str]: """Return (data_path, weights_path, params_path) from asset+TF using project conventions.""" minutes = _TF_TO_MINUTES.get(tf.upper()) if minutes is None: sys.exit(f"Unknown TF '{tf}'. Choices: {list(_TF_TO_MINUTES)}") data = REPO / "data" / "mlp" / f"{asset}, {minutes}.csv" if not data.exists(): sys.exit(f"Data file not found: {data}\nRe-export from TradingView and save to data/mlp/") params_csv = REPO / "results" / "winners" / f"optimization_winner_strategy_mlp_scores_{asset}_{tf.upper()}.csv" if not params_csv.exists(): sys.exit(f"No winner CSV for {asset} {tf}: {params_csv}") weights = pd.read_csv(params_csv).iloc[0]["mlp_weights_file"] if not Path(weights).exists(): sys.exit(f"Weights file from winner CSV not found: {weights}") return str(data), weights, str(params_csv) # ── CLI ─────────────────────────────────────────────────────────────────────── def main(): ap = argparse.ArgumentParser( description="Check Python vs TradingView MLP score parity", epilog="With no --data/--weights, pass --asset and --tf to auto-resolve paths.", ) ap.add_argument("--asset", default="COINBASE_BTCUSD", help="Asset (default: COINBASE_BTCUSD)") ap.add_argument("--tf", default="6H", help="Timeframe (default: 6H)") ap.add_argument("--data", default=None, help="Path to TV-exported CSV (auto-detected from --asset/--tf if omitted)") ap.add_argument("--weights", default=None, help="Path to MLP artifact JSON (auto-detected from --asset/--tf if omitted)") ap.add_argument("--params", default=None, help="Winner CSV or params JSON (auto-detected from --asset/--tf if omitted)") ap.add_argument("--date", default=None, help="Show bar-level detail around this date (YYYY-MM-DD)") ap.add_argument("--window", default=5, type=int, help="Bars each side of --date (default 5)") ap.add_argument("--max-delta", default=10, type=int, help="Show top-N worst-delta bars (default 10)") ap.add_argument("--crossunders-only",action="store_true", help="Only show entry/exit bars in --date table") ap.add_argument("--export", default=None, help="Export full result to CSV") args = ap.parse_args() if args.data is None or args.weights is None: auto_data, auto_weights, auto_params = _auto_resolve(args.asset, args.tf) args.data = args.data or auto_data args.weights = args.weights or auto_weights args.params = args.params or auto_params params = load_params(args.params) print(f"Loading artifact: {args.weights}") print(f"Loading data: {args.data}") if args.params: print(f"Loading params: {args.params}") result = run_parity(args.data, args.weights, params) n_total = len(result) has_tv = result["tv_score"].notna().sum() print(f"\nBars compared: {has_tv}/{n_total} have TV score exported") if has_tv == 0: print("\n⚠️ No mlp_score column found in CSV.") print(" Re-export from TradingView with the MLP strategy active and mlp_score plotted.") sys.exit(1) valid = result.dropna(subset=["tv_score"]) deltas = valid["delta"].abs() max_delta = deltas.max() mean_delta = deltas.mean() p95_delta = deltas.quantile(0.95) print(f"\nScore delta (py - tv):") print(f" max |Δ| = {max_delta:.6f}") print(f" mean |Δ| = {mean_delta:.6f}") print(f" p95 |Δ| = {p95_delta:.6f}") # ── Threshold-side disagreements ── disagrees = threshold_disagreements(result, params) n_disagree = len(disagrees) if n_disagree: print(f"\n❌ THRESHOLD-SIDE DISAGREEMENTS: {n_disagree}") print(disagrees[["time", "py_score", "tv_score", "delta"]].to_string(index=False)) else: print(f"\n✅ Zero threshold-side disagreements") # ── Fragile bars warning ── frags = fragile_bars(result, params) if len(frags): print(f"\n⚠️ {len(frags)} fragile bars (|score - threshold| < 0.05):") print(frags[["time", "py_score", "delta"]].head(20).to_string(index=False)) # ── Overall pass/fail ── passed = max_delta < 0.01 and n_disagree == 0 print(f"\n{'✅ PASS' if passed else '❌ FAIL'} — max|Δ|={max_delta:.6f} {'<' if max_delta < 0.01 else '>='} 0.01, " f"{n_disagree} side-disagreements") # ── Top-N worst bars ── if args.max_delta > 0: worst = valid.nlargest(args.max_delta, "delta") if len(worst): print(f"\nTop {args.max_delta} worst-delta bars:") print(worst[["time", "py_score", "tv_score", "delta"]].to_string(index=False)) # ── Date window ── if args.date: centre = pd.Timestamp(args.date) mask = (result["time"] >= centre - pd.Timedelta(days=args.window)) & \ (result["time"] <= centre + pd.Timedelta(days=args.window)) window_df = result[mask].copy() if args.crossunders_only: window_df = window_df[window_df["execute_entry"] | window_df["execute_exit"]] if len(window_df): print(f"\nBar detail around {args.date} (±{args.window} days):") print(window_df[["time", "py_score", "tv_score", "delta", "execute_entry", "execute_exit"]].to_string(index=False)) else: print(f"\nNo bars found in window around {args.date}") # ── Export ── if args.export: result.to_csv(args.export, index=False) print(f"\nFull result exported to {args.export}") sys.exit(0 if passed else 1) if __name__ == "__main__": main()