""" Diagnose TV/Python Parity at the Component Level ================================================= Shows a per-component score breakdown for every bar in a specified date window, comparing Python's computed scores against the TV-exported activation_score_poc column (if present in the data CSV). Use this to identify *which* signal is causing a crossunder discrepancy. Usage ----- # Show component breakdown for bars around a specific date python tools/diagnose_tv_parity.py \\ --data data/COINBASE_BTCUSD-4H.csv \\ --params results/winners/optimization_winner_activation_scores_COINBASE_BTCUSD_4H.csv \\ --date 2020-03-12 # Show a 10-bar window centred on the date python tools/diagnose_tv_parity.py ... --date 2020-03-12 --window 10 # Show all entry/exit bars (crossunders) in the full data range python tools/diagnose_tv_parity.py ... --crossunders-only # Export breakdown to CSV for further analysis python tools/diagnose_tv_parity.py ... --date 2020-03-12 --export /tmp/breakdown.csv Column key in output table -------------------------- py_score : Python computed activation score (same scale as TV ±1000) tv_poc : TV-exported activation_score_poc (from data CSV, if present) delta : py_score - tv_poc (should be ~0; >10 suggests signal mismatch) [component] : per-component contribution = normalize(value) * weight (in score units) entry/exit : Python signal: E=entry crossunder, X=exit crossunder, .=none If delta is large on a specific bar, look at which component column is most different from what you'd expect — that component's normalisation or export is likely the source of the discrepancy. """ import argparse import importlib.util import json import os import sys from pathlib import Path import numpy as np import pandas as pd sys.path.append(str(Path(__file__).resolve().parent.parent)) from config import TRAIN_START # ── Param loader (shared with compare_tv_trades.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'} 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 {} # ── Strategy component extractor ────────────────────────────────────────────── def compute_components(data_path: str, params: dict) -> pd.DataFrame: """ Re-runs the strategy and returns a DataFrame with: - all standard columns from the data CSV - py_score : Python activation score - tv_poc : TV-exported activation_score_poc (if in CSV) - delta : py_score - tv_poc - per-component contribution columns (normalised_value * weight) - execute_entry / execute_exit """ strategy_path = Path(__file__).resolve().parent.parent / 'strategies' / 'strategy_activation_scores.py' spec = importlib.util.spec_from_file_location('strategy_module', strategy_path) mod = importlib.util.module_from_spec(spec) sys.modules['strategy_module'] = mod spec.loader.exec_module(mod) 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) # Run full strategy to get execute_entry/exit signals = mod.generate_signals(df, **params) # ── Re-extract components to annotate the output ────────────────────────── # All columns are pre-normalised to [-1, +1] by Pine and exported directly. # No normalization is performed here — Python reads the _norm columns as-is. def get_col(primary, fallback=None): if primary in df.columns: return df[primary].fillna(0).values if fallback and fallback in df.columns: return df[fallback].fillna(0).values return np.zeros(len(df)) # Library signals — Pine exports pre-normalised via plotchar stoch_n = get_col('stoch_norm') macd_n = get_col('macd_pred_norm') osc_n = get_col('osc_norm') macd_bull_n = get_col('macd_bullish_norm') m3_mom_n = get_col('m3_momentum_norm') m2_tiny_n = get_col('m2_tiny_norm') rsid_n = get_col('rsid_norm') stoch_div_n = get_col('stoch_div_norm') vwap_div_n = get_col('vwap_div_norm') stoch_peak_n = get_col('stoch_peak_norm') stoch_bot_n = get_col('stoch_bot_norm') m3_div_n = get_col('m3_div_norm') m2_div_n = get_col('m2_div_norm') m2_nooff_n = get_col('m2_nooff_norm') bearish_n = get_col('bearish_engulfing_score', 'db_bearish') bull_ham_n = get_col('bullish_hammer_score') bull_eng_n = get_col('bullish_engulfing_score') star_n = get_col('shooting_star_score') # Macro signals — already normalised, unchanged names btc_spx_corr = get_col('btc_spx_corr_30') dxy_roc_norm = get_col('dxy_roc_norm') vix_pr_inv = get_col('vix_pctrank_inv') btc_dom_sign = get_col('btc_dom_roc_sign') us10y_inv_sign = get_col('us10y_roc_inv_sign') spy_200ema = get_col('spy_above_200ema') gold_pctrank = get_col('gold_roc_pctrank') mvrv_z_val = get_col('mvrv_zscore_value') mvrv_z_cont = get_col('mvrv_zscore_cont') # pre-normalised by Pine: clip(zscore/50-1, -1, 1) nupl_norm = get_col('nupl_norm') fed_liq_sign = get_col('fed_net_liq_sign') gc_position = get_col('gc_position') us2y_inv_sign = get_col('us2y_roc_inv_sign') yield_curve_n = get_col('yield_curve_sign') qqq_spy_n = get_col('qqq_spy_roc_sign') # RSI divergence signals (added 2026-03-26) rsid_reg_bull_n = get_col('rsid_reg_bull_norm') rsid_reg_bear_n = get_col('rsid_reg_bear_norm') rsid_hid_bull_n = get_col('rsid_hid_bull_norm') rsid_hid_bear_n = get_col('rsid_hid_bear_norm') rsid_rt_bull_n = get_col('rsid_rt_bull_norm') rsid_rt_bear_n = get_col('rsid_rt_bear_norm') rsid_slow_bull_n = get_col('rsid_slow_bull_norm') rsid_slow_bear_n = get_col('rsid_slow_bear_norm') rsid_dpeak_n = get_col('rsid_delayed_peak_norm') rsid_ddip_n = get_col('rsid_delayed_dip_norm') # Apply i_div_window extension (matches Pine ta.barssince < i_div_window logic) div_window = int(params.get('i_div_window', 1)) if div_window > 1: def _extend(col): return pd.Series(col).rolling(window=div_window, min_periods=1).max().values rsid_reg_bull_n = _extend(rsid_reg_bull_n) rsid_reg_bear_n = _extend(rsid_reg_bear_n) rsid_hid_bull_n = _extend(rsid_hid_bull_n) rsid_hid_bear_n = _extend(rsid_hid_bear_n) rsid_rt_bull_n = _extend(rsid_rt_bull_n) rsid_rt_bear_n = _extend(rsid_rt_bear_n) rsid_slow_bull_n = _extend(rsid_slow_bull_n) rsid_slow_bear_n = _extend(rsid_slow_bear_n) rsid_dpeak_n = _extend(rsid_dpeak_n) rsid_ddip_n = _extend(rsid_ddip_n) # Derivatives / market structure signals (added 2026-03-28) oi_roc_n = get_col('oi_roc_norm') usdt_d_n = get_col('usdt_d_norm') basis_n = get_col('basis_norm') # Sentiment / sub-TF signals fear_greed_n = get_col('fear_greed_norm') btc_gold_n = get_col('btc_gold_norm') rsi_subtf_n = get_col('rsi_subtf_norm') w_stoch = float(params.get('i_w_stoch', 0)) w_macd_pred = float(params.get('i_w_macd_pred', 0)) w_osc = float(params.get('i_w_osc', 0)) w_macd_bull = float(params.get('i_w_macd_bullish', 0)) w_m3_mom = float(params.get('i_w_m3_momentum', 0)) w_m2_tiny = float(params.get('i_w_m2_tiny', 0)) w_rsid = float(params.get('i_w_rsid_osc', 0)) w_stoch_div = float(params.get('i_w_stoch_div_osc', 0)) w_vwap_div = float(params.get('i_w_vwap_div_osc', 0)) w_stoch_peak = float(params.get('i_w_stoch_peaking', 0)) w_stoch_bot = float(params.get('i_w_stoch_bottoming', 0)) w_m3_div = float(params.get('i_w_m3_div_osc', 0)) w_m2_div = float(params.get('i_w_m2_div_osc', 0)) w_m2_nooff = float(params.get('i_w_m2_div_osc_noOffset', 0)) w_bearish_eng = float(params.get('i_w_bearish_engulfing', 0)) w_bull_hammer = float(params.get('i_w_bullish_hammer', 0)) w_bull_eng = float(params.get('i_w_bullish_engulfing', 0)) w_star = float(params.get('i_w_shooting_star', 0)) w_btc_spx = float(params.get('i_w_btc_spx_corr', 0)) w_dxy = float(params.get('i_w_dxy', 0)) w_vix = float(params.get('i_w_vix', 0)) w_btc_dom = float(params.get('i_w_btc_dom', 0)) w_us10y = float(params.get('i_w_us10y', 0)) w_spy = float(params.get('i_w_spy', 0)) w_gold = float(params.get('i_w_gold', 0)) w_mvrv = float(params.get('i_w_mvrv', 0)) w_mvrv_cont = float(params.get('i_w_mvrv_cont', 0)) w_nupl = float(params.get('i_w_nupl', 0)) w_fed = float(params.get('i_w_fed_net_liq', 0)) w_gc = float(params.get('i_w_gc_position', 0)) w_us2y = float(params.get('i_w_us2y', 0)) w_yield_curve = float(params.get('i_w_yield_curve', 0)) w_qqq_spy = float(params.get('i_w_qqq_spy_ratio', 0)) w_rsid_reg_bull = float(params.get('i_w_rsid_reg_bull', 0)) w_rsid_reg_bear = float(params.get('i_w_rsid_reg_bear', 0)) w_rsid_hid_bull = float(params.get('i_w_rsid_hid_bull', 0)) w_rsid_hid_bear = float(params.get('i_w_rsid_hid_bear', 0)) w_rsid_rt_bull = float(params.get('i_w_rsid_rt_bull', 0)) w_rsid_rt_bear = float(params.get('i_w_rsid_rt_bear', 0)) w_rsid_slow_bull = float(params.get('i_w_rsid_slow_bull', 0)) w_rsid_slow_bear = float(params.get('i_w_rsid_slow_bear', 0)) w_rsid_dpeak = float(params.get('i_w_rsid_delayed_peak', 0)) w_rsid_ddip = float(params.get('i_w_rsid_delayed_dip', 0)) w_oi_roc = float(params.get('i_w_oi_roc', 0)) w_usdt_d = float(params.get('i_w_usdt_d', 0)) w_basis = float(params.get('i_w_basis', 0)) w_fear_greed = float(params.get('i_w_fear_greed', 0)) w_btc_gold = float(params.get('i_w_btc_gold', 0)) w_rsi_subtf = float(params.get('i_w_rsi_subtf', 0)) max_score = ( abs(w_stoch) + abs(w_macd_pred) + abs(w_osc) + abs(w_macd_bull) + abs(w_m3_mom) + abs(w_m2_tiny) + abs(w_rsid) + abs(w_stoch_div) + abs(w_vwap_div) + abs(w_stoch_peak) + abs(w_stoch_bot) + abs(w_m3_div) + abs(w_m2_div) + abs(w_m2_nooff) + abs(w_bearish_eng) + abs(w_bull_hammer) + abs(w_bull_eng) + abs(w_star) + abs(w_btc_spx) + abs(w_dxy) + abs(w_vix) + abs(w_btc_dom) + abs(w_us10y) + abs(w_spy) + abs(w_gold) + abs(w_mvrv) + abs(w_mvrv_cont) + abs(w_nupl) + abs(w_fed) + abs(w_gc) + abs(w_us2y) + abs(w_yield_curve) + abs(w_qqq_spy) + abs(w_rsid_reg_bull) + abs(w_rsid_reg_bear) + abs(w_rsid_hid_bull) + abs(w_rsid_hid_bear) + abs(w_rsid_rt_bull) + abs(w_rsid_rt_bear) + abs(w_rsid_slow_bull) + abs(w_rsid_slow_bear) + abs(w_rsid_dpeak) + abs(w_rsid_ddip) + abs(w_oi_roc) + abs(w_usdt_d) + abs(w_basis) + abs(w_fear_greed) + abs(w_btc_gold) + abs(w_rsi_subtf) ) scale = (1000.0 / max_score) if max_score > 0 else 1.0 def contrib(norm_vals, weight): """Component contribution in scaled score units.""" return norm_vals * weight * scale # Build output DataFrame out = pd.DataFrame() out['time'] = signals['time'] out['close'] = signals['close'] out['py_score'] = signals['activation_score'].round(2) if 'activation_score_poc' in signals.columns: out['tv_poc'] = signals['activation_score_poc'].round(2) out['delta'] = (out['py_score'] - out['tv_poc']).round(2) out['entry'] = signals['execute_entry'].map({True: 'E', False: '.'}) out['exit'] = signals['execute_exit'].map({True: 'X', False: '.'}) # Per-component contributions (only include if weight != 0) comps = [ ('stoch', stoch_n, w_stoch), ('macd_pred', macd_n, w_macd_pred), ('osc', osc_n, w_osc), ('macd_bull', macd_bull_n, w_macd_bull), ('m3_mom', m3_mom_n, w_m3_mom), ('m2_tiny', m2_tiny_n, w_m2_tiny), ('rsid', rsid_n, w_rsid), ('stoch_div', stoch_div_n, w_stoch_div), ('vwap_div', vwap_div_n, w_vwap_div), ('stoch_peak', stoch_peak_n, w_stoch_peak), ('stoch_bot', stoch_bot_n, w_stoch_bot), ('m3_div', m3_div_n, w_m3_div), ('m2_div', m2_div_n, w_m2_div), ('m2_nooff', m2_nooff_n, w_m2_nooff), ('bearish_eng', bearish_n, w_bearish_eng), ('bull_hammer', bull_ham_n, w_bull_hammer), ('bull_eng', bull_eng_n, w_bull_eng), ('shooting_star', star_n, w_star), ('btc_spx', btc_spx_corr, w_btc_spx), ('dxy', dxy_roc_norm, w_dxy), ('vix', vix_pr_inv, w_vix), ('btc_dom', btc_dom_sign, w_btc_dom), ('us10y', us10y_inv_sign, w_us10y), ('spy', spy_200ema, w_spy), ('gold', gold_pctrank, w_gold), ('mvrv', mvrv_z_val, w_mvrv), ('mvrv_cont', mvrv_z_cont, w_mvrv_cont), ('nupl', nupl_norm, w_nupl), ('fed_liq', fed_liq_sign, w_fed), ('gc_pos', gc_position, w_gc), ('us2y', us2y_inv_sign, w_us2y), ('yield_curve', yield_curve_n, w_yield_curve), ('qqq_spy', qqq_spy_n, w_qqq_spy), # RSI divergence (added 2026-03-26) ('rsid_reg_bull', rsid_reg_bull_n, w_rsid_reg_bull), ('rsid_reg_bear', rsid_reg_bear_n, w_rsid_reg_bear), ('rsid_hid_bull', rsid_hid_bull_n, w_rsid_hid_bull), ('rsid_hid_bear', rsid_hid_bear_n, w_rsid_hid_bear), ('rsid_rt_bull', rsid_rt_bull_n, w_rsid_rt_bull), ('rsid_rt_bear', rsid_rt_bear_n, w_rsid_rt_bear), ('rsid_slow_bull', rsid_slow_bull_n, w_rsid_slow_bull), ('rsid_slow_bear', rsid_slow_bear_n, w_rsid_slow_bear), ('rsid_dpeak', rsid_dpeak_n, w_rsid_dpeak), ('rsid_ddip', rsid_ddip_n, w_rsid_ddip), # Derivatives / market structure (added 2026-03-28) ('oi_roc', oi_roc_n, w_oi_roc), ('usdt_d', usdt_d_n, w_usdt_d), ('basis', basis_n, w_basis), # Sentiment / sub-TF ('fear_greed', fear_greed_n, w_fear_greed), ('btc_gold', btc_gold_n, w_btc_gold), ('rsi_subtf', rsi_subtf_n, w_rsi_subtf), ] for name, vals, weight in comps: if weight != 0: out[f'c_{name}'] = contrib(vals, weight).round(2) return out def main(): parser = argparse.ArgumentParser( description='Per-component score breakdown for TV/Python parity diagnosis', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument('--data', required=True, help='OHLCV+indicator data CSV') parser.add_argument('--params', default=None, help='Winner params CSV or JSON (auto-detected from data filename if omitted)') parser.add_argument('--date', default=None, help='Centre date for display window (YYYY-MM-DD)') parser.add_argument('--window', type=int, default=6, help='Bars to show before/after --date (default: 6)') parser.add_argument('--crossunders-only', action='store_true', help='Show only bars where Python fires entry or exit') parser.add_argument('--max-delta', type=float, default=None, help='Show only bars where |delta| >= this value (requires tv_poc column)') parser.add_argument('--export', default=None, help='Export full breakdown to CSV path') args = parser.parse_args() # Auto-detect params params_path = args.params if params_path is None: stem = Path(args.data).stem asset_tf = stem.replace('-', '_').upper() candidate = Path('results/winners') / f'optimization_winner_activation_scores_{asset_tf}.csv' if candidate.exists(): params_path = str(candidate) print(f'Auto-detected params: {params_path}') params = load_params(params_path) entry_threshold = float(params.get('i_long_entry_activation_threshold', 106.0)) exit_threshold = float(params.get('i_long_exit_activation_threshold', 140.5)) trail_pct = float(params.get('i_trailing_stop_threshold', 0.0)) print(f'Computing component breakdown for: {args.data}') print(f'Entry threshold: {entry_threshold} | Exit threshold: {exit_threshold} | Trail stop: {trail_pct}%') out = compute_components(args.data, params) if args.export: out.to_csv(args.export, index=False) print(f'Exported to: {args.export}') # ── Filter rows to display ───────────────────────────────────────────────── display = out.copy() if args.date: centre = pd.Timestamp(args.date) # Find bar index closest to centre idx_arr = out.index[out['time'].sub(centre).abs() == out['time'].sub(centre).abs().min()] if idx_arr.empty: print(f'No bars found near {args.date}') return centre_idx = idx_arr[0] lo = max(0, centre_idx - args.window) hi = min(len(out) - 1, centre_idx + args.window) display = out.iloc[lo:hi + 1] elif args.crossunders_only: display = out[(out['entry'] == 'E') | (out['exit'] == 'X')] elif args.max_delta is not None and 'delta' in out.columns: display = out[out['delta'].abs() >= args.max_delta] else: # Default: show last 20 bars if no filter given display = out.tail(20) print('[No --date filter given. Showing last 20 bars. Use --date YYYY-MM-DD or --crossunders-only.]\n') # ── Print table ──────────────────────────────────────────────────────────── comp_cols = [c for c in display.columns if c.startswith('c_')] # Score summary columns score_cols = ['time', 'close', 'py_score'] if 'tv_poc' in display.columns: score_cols += ['tv_poc', 'delta'] score_cols += ['entry', 'exit'] pd.set_option('display.width', 200) pd.set_option('display.max_columns', 60) pd.set_option('display.float_format', '{:.2f}'.format) print('\n── Score Summary ──────────────────────────────────────────────') print(display[score_cols].to_string(index=True)) if comp_cols: print('\n── Component Contributions (score units, non-zero weight only) ─') print(display[['time'] + comp_cols].to_string(index=True)) # Also print totals sanity check total_check = display[comp_cols].sum(axis=1) print(f'\n Component sum vs py_score for displayed rows:') for ridx, (row_idx, row) in enumerate(display.iterrows()): comp_sum = sum(row.get(c, 0) for c in comp_cols) py_s = row['py_score'] delta = comp_sum - py_s marker = ' ← MISMATCH' if abs(delta) > 1.0 else '' print(f' {str(row["time"])[:19]:>19} comp_sum={comp_sum:>8.2f} py_score={py_s:>8.2f} Δ={delta:>+6.2f}{marker}') print('\nDone.') if __name__ == '__main__': main()