signal_log: persist agent panel verdict + per-persona takes to SQLite
Add panel_verdict, panel_confidence, panel_rationale, panel_takes_json columns to the signals table (additive, idempotent ALTER TABLE migration for existing DBs). _report_row now reads the panel fields attach_panel writes onto the SignalReport, so LLM panel verdicts stop evaporating on process exit. Also documents the Oanda re-pin (round_trip_cost stays $0.30 for cross-book comparability and the live TP gate; $0.65 is the measured headline) and adds the feed-divergence / parity / shadow-scan study tooling referenced by CLAUDE.md.
This commit is contained in:
Harwinder Sekhon
@@ -0,0 +1,194 @@
|
||||
#!/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_<field>, b_<field> 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()
|
||||
Reference in New Issue
Block a user