#!/usr/bin/env python3 """ Phase 2 — signal-level replay parity (data source change test) =============================================================== Replays the REAL decision core (`SignalAggregator.analyze_from_dataframes`, news off, use_news=False) over the same window for two providers and compares the *decisions* bar-by-bar: direction, composite/final score, quality grade, gate/setup presence, strategy tag, and entry/SL/TP levels. This turns the Phase-1 price-diff stats into flip rates. Feeds are read from the Phase-1 parquet cache (scripts/feed_divergence_study.py writes them); pass --no-cache-fetch only if you want to fetch fresh via fetch_range first. Slicing matches gpu_backtest.py exactly: decision at close of bar `pos`, each TF sliced to bars fully closed by that decision time, primary at the shared grid, so the two feeds are compared on identical timestamps. Usage: venv/bin/python scripts/signal_parity_study.py --a oanda --b tiingo \ --tf M15 --stride 8 --days 365 """ import argparse import sys import time from pathlib import Path import numpy as np import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from config import Timeframe, XAUUSD from pair_signals.xauusd import make_config from signal_aggregator import SignalAggregator REPO = Path(__file__).resolve().parent.parent CACHE_DIR = REPO / "backtest" / "cache" RESEARCH_DIR = REPO / "backtest" / "research" BAR_SECONDS = { Timeframe.M1: 60, Timeframe.M5: 300, Timeframe.M15: 900, Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, Timeframe.D1: 86400, Timeframe.W1: 604800, } _TF_LOOKUP = {t.value: t for t in Timeframe} _TF_LOOKUP.update({"M1": Timeframe.M1, "M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30, "H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1, "W1": Timeframe.W1}) _CONFIRMS = { Timeframe.M5: [Timeframe.M15, Timeframe.H1], Timeframe.M15: [Timeframe.M5, Timeframe.H1], Timeframe.H1: [Timeframe.M15, Timeframe.H4], Timeframe.H4: [Timeframe.H1, Timeframe.D1], } def load_cached(name: str, tf: Timeframe, start: pd.Timestamp, end: pd.Timestamp) -> pd.DataFrame: cp = CACHE_DIR / f"{name}_XAU_USD_{tf.value}_{start:%Y%m%d}_{end:%Y%m%d}.parquet" if not cp.exists(): sys.exit(f"missing cache {cp}; run scripts/feed_divergence_study.py " f"--a oanda --b {name} --tfs {tf.value} --days ... first") return pd.read_parquet(cp) def tag_of(report) -> str: first_bull = report.bullish_reasons[0] if report.bullish_reasons else "" first_bear = report.bearish_reasons[0] if report.bearish_reasons else "" if "XAU PULLBACK" in first_bull or "XAU PULLBACK" in first_bear: return "XAU_PB" if "XAU CHART PATTERN" in first_bull or "XAU CHART PATTERN" in first_bear: return "XAU_CP" if "XAU STRATEGY" in first_bull: return "XAU" return "GENERIC" def decide(agg, dfs, tss, stride, warmup) -> list: """Replay decisions over the shared grid, returning one row per bar.""" primary = dfs[agg.config.primary_timeframe] dur = pd.Timedelta(seconds=BAR_SECONDS[agg.config.primary_timeframe]) # positions of sampled bars in the primary frame primary_pos = primary.index.get_indexer(tss) rows = [] for i in range(0, len(tss), stride): ts = tss[i] pos = int(primary_pos[i]) if pos < warmup: continue slices = {} for tf, full in dfs.items(): cutoff = ts # decision_time = ts + dur; bars closed by ts cut = int(full.index.searchsorted(cutoff, side="right")) if cut <= 0: continue slices[tf] = full.iloc[:cut] if agg.config.primary_timeframe not in slices: continue r = agg.analyze_from_dataframes(XAUUSD, slices, use_news=False) ts_setup = r.trade_setup rows.append({ "ts": ts, "direction": r.final_signal.value, "final_score": r.final_score, "primary_score": r.primary_score, "grade": r.signal_quality, "has_setup": ts_setup is not None, "tag": tag_of(r) if ts_setup else "", "entry": ts_setup.entry_price if ts_setup else np.nan, "sl": ts_setup.stop_loss if ts_setup else np.nan, "tp1": ts_setup.take_profit_1 if ts_setup else np.nan, "tp2": ts_setup.take_profit_2 if ts_setup else np.nan, }) return rows def main(): ap = argparse.ArgumentParser() ap.add_argument("--a", required=True) ap.add_argument("--b", required=True) ap.add_argument("--tf", default="M15") ap.add_argument("--stride", type=int, default=1, help="sample every Nth shared bar (default 1)") ap.add_argument("--days", type=int, default=365) ap.add_argument("--end", default=None) ap.add_argument("--warmup", type=int, default=300) args = ap.parse_args() tf = _TF_LOOKUP.get(args.tf) if tf is None: sys.exit(f"unknown tf {args.tf}") confirms = _CONFIRMS[tf] tfs = [tf] + confirms end = pd.Timestamp(args.end, tz="UTC") if args.end \ else pd.Timestamp.now(tz="UTC") start = end - pd.Timedelta(days=args.days) cfg = make_config(tf, confirms) agg = SignalAggregator(cfg) a_dfs, b_dfs = {}, {} for t in tfs: a_dfs[t] = load_cached(args.a, t, start, end) b_dfs[t] = load_cached(args.b, t, start, end) # shared grid = primary timestamps present in both feeds shared = a_dfs[tf].index.intersection(b_dfs[tf].index) t0 = time.time() print(f"replaying {len(shared)} shared {tf.value} bars " f"(stride {args.stride} → ~{len(shared)//args.stride}) on {args.a}...") a_rows = decide(agg, a_dfs, shared, args.stride, args.warmup) print(f" {args.a}: {len(a_rows)} decisions in {time.time()-t0:.1f}s") t0 = time.time() b_rows = decide(agg, b_dfs, shared, args.stride, args.warmup) print(f" {args.b}: {len(b_rows)} decisions in {time.time()-t0:.1f}s") A = pd.DataFrame(a_rows).set_index("ts") B = pd.DataFrame(b_rows).set_index("ts") common = A.index.intersection(B.index) A, B = A.loc[common], B.loc[common] n = len(common) print(f"\n{'=' * 78}") print(f"SIGNAL-LEVEL PARITY — {tf.value} — {args.a} vs {args.b} " f"({n:,} bars)") print("=" * 78) def pct(x): return f"{100 * x:.2f}%" same_dir = (A["direction"] == B["direction"]) same_grade = (A["grade"] == B["grade"]) same_setup = (A["has_setup"] == B["has_setup"]) same_all = same_dir & same_grade & same_setup print(f"\nIDENTITY") print(f" same direction : {pct(same_dir.mean())}") print(f" same grade : {pct(same_grade.mean())}") print(f" same setup presence : {pct(same_setup.mean())}") print(f" same dir+grade+setup : {pct(same_all.mean())}") both_sig = (A["direction"] != 0) & (B["direction"] != 0) flips = both_sig & (A["direction"] != B["direction"]) opp = flips & (A["direction"] == -B["direction"]) print(f"\nDIRECTION") print(f" bars both actionable : {both_sig.sum():,} " f"({pct(both_sig.mean())})") print(f" direction flips : {flips.sum():,} " f"({pct(flips.sum()/max(both_sig.sum(),1))} of actionable)") print(f" opposite-sign flips : {opp.sum():,}") grade_a = set(A.loc[A["has_setup"], "grade"]) print(f"\nGRADE (of {args.a} setups, grades seen: {sorted(grade_a)})") changed = A.loc[both_sig, "grade"] != B.loc[both_sig, "grade"] print(f" grade changed on actionable bars: " f"{changed.sum():,} ({pct(changed.mean())})") # score deltas on shared bars ds = (A["final_score"] - B["final_score"]).abs() dp = (A["primary_score"] - B["primary_score"]).abs() print(f"\nSCORE Δ ({args.a} − {args.b})") print(f" final_score mean {ds.mean():6.2f} p95 {ds.quantile(.95):6.2f} " f"max {ds.max():6.2f}") print(f" primary_score mean {dp.mean():6.2f} p95 {dp.quantile(.95):6.2f} " f"max {dp.max():6.2f}") # setup price diffs on bars where BOTH generated a setup both_setup = A["has_setup"] & B["has_setup"] if both_setup.any(): print(f"\nSETUP LEVELS (both-setup bars: {both_setup.sum():,})") for col, label in (("entry", "entry"), ("sl", "SL"), ("tp1", "TP1"), ("tp2", "TP2")): d = (A.loc[both_setup, col] - B.loc[both_setup, col]).abs() print(f" {label:<6} mean ${d.mean():6.3f} p95 ${d.quantile(.95):7.3f} " f"max ${d.max():7.2f}") # tag disagreement on both-setup bars if both_setup.any(): same_tag = (A.loc[both_setup, "tag"] == B.loc[both_setup, "tag"]) print(f" same strategy tag : {pct(same_tag.mean())}") # where do flips cluster (hour)? flip_hours = A.loc[flips].index.hour.value_counts().sort_index() if len(flip_hours): print(f"\nFLIPS BY UTC HOUR") for h, c in flip_hours.items(): print(f" {h:02d}: {c}") out = { "tf": tf.value, "a": args.a, "b": args.b, "bars": n, "same_dir": same_dir.mean(), "same_grade": same_grade.mean(), "same_setup": same_setup.mean(), "same_all": same_all.mean(), "flip_rate_actionable": flips.sum() / max(both_sig.sum(), 1), "mean_abs_final_score_delta": ds.mean(), "mean_abs_primary_score_delta": dp.mean(), } print(f"\n{out}") if __name__ == "__main__": main()