#!/usr/bin/env python3 """ Phase 4 — live shadow scan (data source change test) ===================================================== Runs TWO data providers side-by-side at every candle close (M5/M15/H1) and logs how their signals diverge, so the feed switch can be validated under real fetch depths / caching / boundary timing for ≥2 weeks. Unlike the history studies, this uses the LIVE fetch path (fetch_ohlcv, forming-candle drop, freshness checks) for both providers, but calls `analyze_from_dataframes(use_news=False)` directly so the (provider- independent) news/DeepSeek budget is not burned on shadow scans. The weekend filter is applied here (skip the whole cycle), and the session/ day/hour gates run inside analyze_from_dataframes as in the backtest. No orders, no Telegram, no signal-log writes — output is a side-by-side CSV plus a rolling agreement tally. Usage: venv/bin/python scripts/shadow_scan.py --a oanda --b tiingo \ --tfs M5,M15,H1 --out logs/shadow_oanda_tiingo.csv --once # production (2+ weeks): nohup venv/bin/python scripts/shadow_scan.py --a oanda --b tiingo \ --out logs/shadow_oanda_tiingo.csv >> logs/shadow_scan.log 2>&1 & CSV columns (one row per TF per boundary): ts, tf, a_, b_ for direction/grade/setup/tag/entry/sl/tp1, a_price, b_price, a_spread, b_spread, a_warnings, b_warnings, dir_agree, grade_agree, setup_agree """ import argparse import csv import sys import time from datetime import datetime, timezone from pathlib import Path import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from config import PlatformConfig, Timeframe, XAUUSD from pair_signals.xauusd import make_config from providers import make_data_provider from providers.base import _is_forex_weekend from signal_aggregator import SignalAggregator, SignalReport _TF_LOOKUP = {"M5": Timeframe.M5, "M15": Timeframe.M15, "H1": Timeframe.H1, "H4": Timeframe.H4} _TF_MINUTES = {Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.H1: 60, Timeframe.H4: 240} _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]} FIELDS = ["direction", "grade", "setup", "tag", "entry", "sl", "tp1"] def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float: now = datetime.now(timezone.utc) elapsed = (now.minute % interval_minutes) * 60 + now.second \ + now.microsecond / 1e6 return interval_minutes * 60 - elapsed + buffer def due_timeframes(wake: datetime, tfs) -> list: minutes = wake.hour * 60 + wake.minute floored = minutes - (minutes % 5) return [tf for tf in tfs if floored % _TF_MINUTES[tf] == 0] def tag_of(report) -> str: fb = report.bullish_reasons[0] if report.bullish_reasons else "" fbe = report.bearish_reasons[0] if report.bearish_reasons else "" if "XAU PULLBACK" in fb or "XAU PULLBACK" in fbe: return "XAU_PB" if "XAU CHART PATTERN" in fb or "XAU CHART PATTERN" in fbe: return "XAU_CP" if "XAU STRATEGY" in fb: return "XAU" return "GENERIC" def scan(provider, symbol, tf) -> SignalReport: confirms = _CONFIRMS[tf] cfg = make_config(tf, confirms) agg = SignalAggregator(cfg) agg.data_provider = provider dfs = {} for t in [tf] + confirms: dfs[t] = provider.fetch_ohlcv(symbol, t) return agg.analyze_from_dataframes(symbol, dfs, use_news=False) def extract(provider, report) -> dict: s = report.trade_setup return { "direction": report.final_signal.value, "grade": report.signal_quality, "setup": 1 if s is not None else 0, "tag": tag_of(report) if s else "", "entry": round(s.entry_price, 3) if s else "", "sl": round(s.stop_loss, 3) if s else "", "tp1": round(s.take_profit_1, 3) if s else "", "price": round(report.current_price, 3), "spread": round(provider.get_spread(XAUUSD), 3), "warnings": " | ".join(report.warnings)[:200], } def main(): ap = argparse.ArgumentParser() ap.add_argument("--a", default="oanda") ap.add_argument("--b", default="tiingo") ap.add_argument("--tfs", default="M5,M15,H1") ap.add_argument("--out", default="logs/shadow_scan.csv") ap.add_argument("--once", action="store_true", help="one immediate scan cycle, then exit") args = ap.parse_args() tfs = [_TF_LOOKUP[s.strip()] for s in args.tfs.split(",")] pa = make_data_provider(PlatformConfig(), args.a) pb = make_data_provider(PlatformConfig(), args.b) out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) new_file = not out_path.exists() tally = {"cycles": 0, "dir": 0, "grade": 0, "setup": 0} with open(out_path, "a", newline="") as fh: w = csv.writer(fh) if new_file: w.writerow(["ts", "tf"] + [f"a_{x}" for x in FIELDS + ["price", "spread", "warnings"]] + [f"b_{x}" for x in FIELDS + ["price", "spread", "warnings"]] + ["dir_agree", "grade_agree", "setup_agree"]) while True: if args.once: wake = datetime.now(timezone.utc) else: wait = seconds_until_boundary(5, buffer=10) time.sleep(wait) wake = datetime.now(timezone.utc) if _is_forex_weekend(): if args.once: print("weekend — nothing to scan") return continue for tf in due_timeframes(wake, tfs): try: ra = scan(pa, XAUUSD, tf) except Exception as e: print(f"[{wake:%H:%M:%S}] {args.a} {tf.value} FAILED: {e}") continue try: rb = scan(pb, XAUUSD, tf) except Exception as e: print(f"[{wake:%H:%M:%S}] {args.b} {tf.value} FAILED: {e}") continue ea, eb = extract(pa, ra), extract(pb, rb) da = ea["direction"] == eb["direction"] ga = ea["grade"] == eb["grade"] sa = ea["setup"] == eb["setup"] tally["cycles"] += 1 tally["dir"] += da; tally["grade"] += ga; tally["setup"] += sa row = [wake.isoformat(), tf.value] row += [ea[x] for x in FIELDS + ["price", "spread", "warnings"]] row += [eb[x] for x in FIELDS + ["price", "spread", "warnings"]] row += [int(da), int(ga), int(sa)] w.writerow(row) fh.flush() print(f"[{wake:%Y-%m-%d %H:%M:%S}] {tf.value:<5} " f"{args.a}={ea['direction']:+d}/{ea['grade'] or '-':>2} " f"{args.b}={eb['direction']:+d}/{eb['grade'] or '-':>2} " f"dir={'OK' if da else 'FLIP'} " f"| Δprice {abs(ea['price']-eb['price']):.2f} " f"| agree dir {tally['dir']}/{tally['cycles']} " f"grade {tally['grade']}/{tally['cycles']} " f"setup {tally['setup']}/{tally['cycles']}") if args.once: print("shadow scan complete (--once)") return if __name__ == "__main__": main()