""" Diagnose missing or stale indicator columns in a TradingView data CSV. Checks that every column read by strategy_activation_scores.py via get_col() is present in the data CSV. If a params file is provided, also flags which missing score columns have non-zero weights (those will cause score differences). Usage: python tools/diagnose_missing_data.py data/COINBASE_BTCUSD-4H.csv python tools/diagnose_missing_data.py data/COINBASE_BTCUSD-4H.csv \\ --params results/winners/optimization_winner_activation_scores_COINBASE_BTCUSD_4H.csv Exit codes: 0 all clear (or only locked=0 columns missing) 1 gate column missing (HIGH SEVERITY) or non-zero weight column missing """ import argparse import json import os import sys import pandas as pd # ── Score columns ───────────────────────────────────────────────────────────── # Maps CSV column name (lowercase) → weight param name. # Source: strategy_activation_scores.py get_col() calls. # # SINGLE-SOURCE-OF-TRUTH CONTRACT: # set(SCORE_COLUMNS.values()) ∪ {gate weight params} must equal config.WEIGHT_COLS. # Validated at startup by _validate_column_coverage() below — any mismatch prints # a loud WARNING so new signals are never silently ignored by this tool. # Gate-only weight: i_w_stoch_peaking → its CSV column lives in GATE_COLUMNS. SCORE_COLUMNS = { # Core technical signals "stoch_norm": "i_w_stoch", "macd_pred_norm": "i_w_macd_pred", "osc_norm": "i_w_osc", "macd_bullish_norm": "i_w_macd_bullish", "m3_momentum_norm": "i_w_m3_momentum", "m2_tiny_norm": "i_w_m2_tiny", "rsid_norm": "i_w_rsid_osc", "stoch_div_norm": "i_w_stoch_div_osc", "vwap_div_norm": "i_w_vwap_div_osc", "stoch_bot_norm": "i_w_stoch_bottoming", "m3_div_norm": "i_w_m3_div_osc", "m2_div_norm": "i_w_m2_div_osc", "m2_nooff_norm": "i_w_m2_div_osc_noOffset", # Candlestick patterns "bearish_engulfing_score": "i_w_bearish_engulfing", "bullish_hammer_score": "i_w_bullish_hammer", "bullish_engulfing_score": "i_w_bullish_engulfing", "shooting_star_score": "i_w_shooting_star", # Macro / cross-asset "btc_spx_corr_30": "i_w_btc_spx_corr", "dxy_roc_norm": "i_w_dxy", "vix_pctrank_inv": "i_w_vix", "btc_dom_roc_sign": "i_w_btc_dom", "us10y_roc_inv_sign": "i_w_us10y", "spy_above_200ema": "i_w_spy", "gold_roc_pctrank": "i_w_gold", # On-chain / regime "mvrv_zscore_value": "i_w_mvrv", "mvrv_zscore_cont": "i_w_mvrv_cont", "nupl_norm": "i_w_nupl", "fed_net_liq_sign": "i_w_fed_net_liq", "gc_position": "i_w_gc_position", # Rates / macro "us2y_roc_inv_sign": "i_w_us2y", "yield_curve_sign": "i_w_yield_curve", "qqq_spy_roc_sign": "i_w_qqq_spy_ratio", # RSI divergence "rsid_reg_bull_norm": "i_w_rsid_reg_bull", "rsid_reg_bear_norm": "i_w_rsid_reg_bear", "rsid_hid_bull_norm": "i_w_rsid_hid_bull", "rsid_hid_bear_norm": "i_w_rsid_hid_bear", "rsid_rt_bull_norm": "i_w_rsid_rt_bull", "rsid_rt_bear_norm": "i_w_rsid_rt_bear", "rsid_slow_bull_norm": "i_w_rsid_slow_bull", "rsid_slow_bear_norm": "i_w_rsid_slow_bear", "rsid_delayed_peak_norm": "i_w_rsid_delayed_peak", "rsid_delayed_dip_norm": "i_w_rsid_delayed_dip", # Derivatives / market structure "oi_roc_norm": "i_w_oi_roc", "usdt_d_norm": "i_w_usdt_d", "basis_norm": "i_w_basis", # Sentiment / sub-TF "fear_greed_norm": "i_w_fear_greed", "btc_gold_norm": "i_w_btc_gold", "rsi_subtf_norm": "i_w_rsi_subtf", } # Condition gate columns — used in entry/exit logic beyond their score weight. # CRITICAL: missing these causes Python/TV exit or entry divergence regardless of weight. GATE_COLUMNS = { "stoch_peak_norm": ( "EXIT GATE: longExitCondition requires stoch_is_peaking.\n" " Missing → Python never exits via score crossunder → holds far longer than TV.\n" " See docs/tv_parity_workflow.md §F." ), } # Weight params whose CSV column lives in GATE_COLUMNS (not SCORE_COLUMNS). # These are intentionally excluded from SCORE_COLUMNS because they act as # condition gates only (their score weight is always locked=0). _GATE_WEIGHT_PARAMS = {"i_w_stoch_peaking"} # TV score reference column (informational — needed for compare_tv_trades.py score deltas). REFERENCE_COLUMNS = { "activation_score_poc": "TV activation score — needed for compare_tv_trades.py score-delta diagnostics", } def _validate_column_coverage() -> None: """ Cross-check SCORE_COLUMNS against config.WEIGHT_COLS at startup. Prints a WARNING (never aborts) if any WEIGHT_COLS entry is not covered by either SCORE_COLUMNS.values() or GATE_COLUMNS. Catches the class of bug where a new signal is added to config.py but the CSV-column→weight mapping here is not updated. Known intentional gap: i_w_stoch_peaking → CSV column 'stoch_peak_norm' lives in GATE_COLUMNS (not SCORE_COLUMNS) because it is a condition gate, not a score weight. """ try: _root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, _root) from config import WEIGHT_COLS except Exception as e: print(f"[WARN] Could not import config.WEIGHT_COLS for coverage check: {e}") return covered = set(SCORE_COLUMNS.values()) | _GATE_WEIGHT_PARAMS missing = [w for w in WEIGHT_COLS if w not in covered] extra = [v for v in SCORE_COLUMNS.values() if v not in WEIGHT_COLS] if missing: print() print("!! WARNING: The following config.WEIGHT_COLS entries are NOT mapped in") print(" SCORE_COLUMNS (or GATE_COLUMNS). This tool will not check their CSV columns.") print(" Add them to SCORE_COLUMNS or _GATE_WEIGHT_PARAMS to silence this warning.") for w in sorted(missing): print(f" {w}") print() if extra: print() print("!! WARNING: The following SCORE_COLUMNS values are NOT in config.WEIGHT_COLS.") print(" They may be stale entries. Remove or add to WEIGHT_COLS.") for w in sorted(extra): print(f" {w}") print() def load_params(params_path: str) -> dict: if not params_path: return {} try: if params_path.endswith('.csv'): return pd.read_csv(params_path).iloc[0].to_dict() if params_path.endswith('.json'): with open(params_path) as f: return json.load(f) except Exception as e: print(f"[WARN] Could not load params: {e}") return {} def main(): parser = argparse.ArgumentParser( description='Diagnose missing indicator columns in a TradingView data CSV', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument('file', help='Path to data CSV (e.g. data/COINBASE_BTCUSD-4H.csv)') parser.add_argument('--params', default=None, help='Winner params CSV or JSON — highlights non-zero missing weights') args = parser.parse_args() _validate_column_coverage() try: df = pd.read_csv(args.file, nrows=5) df.columns = df.columns.str.lower().str.strip() except Exception as e: print(f'Error reading {args.file}: {e}') sys.exit(1) csv_cols = set(df.columns) params = load_params(args.params) exit_code = 0 print(f'\nDiagnosing {args.file} ({len(csv_cols)} columns total)\n') # ── Score columns ────────────────────────────────────────────────────────── missing_locked = {} # col → weight_param (weight=0 → no impact) missing_nonzero = {} # col → (weight_param, weight_value) → score difference! present_count = 0 for col, weight_param in SCORE_COLUMNS.items(): if col in csv_cols: present_count += 1 else: weight_val = float(params.get(weight_param, 0.0)) if params else None if weight_val is not None and weight_val != 0.0: missing_nonzero[col] = (weight_param, weight_val) else: missing_locked[col] = weight_param print(f'Score columns {present_count}/{len(SCORE_COLUMNS)} present') if missing_nonzero: print(f'\n !! MISSING with non-zero weight ({len(missing_nonzero)}) — will cause score differences:') for col, (weight_param, wv) in missing_nonzero.items(): print(f' {col:<32} ({weight_param} = {wv:+.0f})') print() print(' Python reads these as 0.0 but TV computes actual values.') print(' Fix: paste latest Pine to TV, re-export the data CSV.') exit_code = 1 if missing_locked: status = 'locked=0 or unknown' if params else 'no params provided — assuming weight=0' print(f'\n Missing but locked=0 ({len(missing_locked)}) — no score impact [{status}]:') for col, weight_param in missing_locked.items(): print(f' {col:<32} ({weight_param})') if not missing_nonzero and not missing_locked: print(' All score columns present ✓') elif not missing_nonzero: print('\n All non-zero weight columns present ✓') # ── Gate columns ─────────────────────────────────────────────────────────── print(f'\nCondition gate columns (critical — must be present regardless of weight)') gate_ok = True for col, desc in GATE_COLUMNS.items(): if col in csv_cols: print(f' {col:<32} PRESENT ✓') else: print(f' {col:<32} MISSING ✗') print(f' {desc}') gate_ok = False exit_code = 1 if not gate_ok: print() print(' !! HIGH SEVERITY: gate column missing — Python entry/exit logic diverges from TV.') # ── Reference columns ───────────────────────────────────────────────────── print(f'\nReference columns (informational)') for col, desc in REFERENCE_COLUMNS.items(): if col in csv_cols: print(f' {col:<32} PRESENT ✓') else: print(f' {col:<32} missing ({desc})') # ── Final verdict ───────────────────────────────────────────────────────── print() if not gate_ok: print('=== RESULT: HIGH SEVERITY — gate column missing (see above) ===') elif missing_nonzero: print('=== RESULT: SCORE DIFFERENCE — non-zero weight column(s) missing ===') print(' Python scores will diverge from TV. Re-export data CSV to fix.') elif missing_locked: print('=== RESULT: OK — missing columns are all locked=0 (no score impact) ===') else: print('=== RESULT: ALL CLEAR — all columns present ===') sys.exit(exit_code) if __name__ == '__main__': main()