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,116 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Scratchpad — Oanda vs Tiingo candle parity check
|
||||
=================================================
|
||||
Fetches the last N closed XAU/USD candles from Oanda and Tiingo for
|
||||
M5/M15/H1 via the normal provider stack (make_data_provider →
|
||||
fetch_ohlcv: forming-candle drop + validate/clean included, i.e. exactly
|
||||
what the signal pipeline consumes) and reports how well they line up:
|
||||
|
||||
* last-bar timestamp per provider (feed-lag check)
|
||||
* timestamp mismatches (bars present in one feed only)
|
||||
* per-bar OHLC diff table for the common tail
|
||||
* max/mean |diff| per column, with the Close diff also in % of price
|
||||
|
||||
Both feeds are mid-price (Oanda mid of MBA candles; Tiingo resampled
|
||||
mid), so small diffs are expected — this measures how small.
|
||||
|
||||
Usage:
|
||||
venv/bin/python scripts/compare_oanda_tiingo.py [--n 10]
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
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 providers import make_data_provider
|
||||
|
||||
TIMEFRAMES = [Timeframe.M5, Timeframe.M15, Timeframe.H1]
|
||||
OHLC = ["Open", "High", "Low", "Close"]
|
||||
|
||||
|
||||
def fetch_tail(provider, tf, n):
|
||||
"""Last n closed bars from one provider, or None on failure."""
|
||||
try:
|
||||
df = provider.fetch_ohlcv(symbol=XAUUSD, timeframe=tf)
|
||||
except Exception as e: # scratchpad: report, don't crash
|
||||
print(f" !! {provider.name} {tf.value} fetch failed: {e}")
|
||||
return None
|
||||
return df.tail(n)
|
||||
|
||||
|
||||
def compare_tf(tf, oanda_df, tiingo_df, n):
|
||||
print(f"\n{'=' * 78}\n{tf.value} — last {n} closed bars\n{'=' * 78}")
|
||||
if oanda_df is None or tiingo_df is None:
|
||||
print(" skipped (a provider failed)")
|
||||
return
|
||||
|
||||
print(f" Oanda last bar: {oanda_df.index[-1]} "
|
||||
f"Tiingo last bar: {tiingo_df.index[-1]}")
|
||||
|
||||
common = oanda_df.index.intersection(tiingo_df.index)
|
||||
only_o = oanda_df.index.difference(tiingo_df.index)
|
||||
only_t = tiingo_df.index.difference(oanda_df.index)
|
||||
if len(only_o):
|
||||
print(f" bars only in Oanda : {[str(t) for t in only_o]}")
|
||||
if len(only_t):
|
||||
print(f" bars only in Tiingo: {[str(t) for t in only_t]}")
|
||||
if not len(common):
|
||||
print(" NO common timestamps — feeds are on different bar grids!")
|
||||
return
|
||||
|
||||
o = oanda_df.loc[common, OHLC]
|
||||
t = tiingo_df.loc[common, OHLC]
|
||||
diff = o - t
|
||||
|
||||
print(f"\n {len(common)}/{n} bars share timestamps "
|
||||
f"(window {common[0]} → {common[-1]})\n")
|
||||
|
||||
# Per-bar Close table — the number that matters most.
|
||||
hdr = f" {'bar close (UTC)':<20}{'Oanda C':>12}{'Tiingo C':>12}{'ΔC ($)':>10}"
|
||||
print(hdr + f"{'ΔO':>8}{'ΔH':>8}{'ΔL':>8}")
|
||||
for ts in common:
|
||||
row = (f" {str(ts):<20}{o.loc[ts, 'Close']:>12.2f}"
|
||||
f"{t.loc[ts, 'Close']:>12.2f}{diff.loc[ts, 'Close']:>10.3f}")
|
||||
row += (f"{diff.loc[ts, 'Open']:>8.3f}{diff.loc[ts, 'High']:>8.3f}"
|
||||
f"{diff.loc[ts, 'Low']:>8.3f}")
|
||||
print(row)
|
||||
|
||||
print("\n summary over common bars:")
|
||||
for col in OHLC:
|
||||
d = diff[col].abs()
|
||||
print(f" {col:<6} max |Δ| = {d.max():8.4f} mean |Δ| = {d.mean():8.4f}")
|
||||
px = o["Close"].iloc[-1]
|
||||
dc = diff["Close"].abs()
|
||||
print(f" Close Δ as % of price: max {dc.max() / px * 100:.4f}% "
|
||||
f"mean {dc.mean() / px * 100:.4f}%")
|
||||
exact = (diff[OHLC].abs() < 1e-9).all(axis=1).sum()
|
||||
print(f" bars with identical OHLC: {exact}/{len(common)}")
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--n", type=int, default=10,
|
||||
help="bars per timeframe to compare (default 10)")
|
||||
args = ap.parse_args()
|
||||
|
||||
cfg = PlatformConfig()
|
||||
oanda = make_data_provider(cfg, "oanda")
|
||||
tiingo = make_data_provider(cfg, "tiingo")
|
||||
|
||||
for tf in TIMEFRAMES:
|
||||
o_df = fetch_tail(oanda, tf, args.n)
|
||||
t_df = fetch_tail(tiingo, tf, args.n)
|
||||
compare_tf(tf, o_df, t_df, args.n)
|
||||
|
||||
print("\nDone. Note: Oanda=mid of MBA candles, Tiingo=resampled mid — "
|
||||
"sub-$1 diffs on gold are normal feed divergence, not a bug.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,259 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Phase 1 — historical candle divergence study (data source change test)
|
||||
=======================================================================
|
||||
Fetches the RAW closed-bar series for XAU/USD from two providers over a
|
||||
multi-month window via the normal provider stack (`fetch_range`, which
|
||||
paginates under the per-request bar cap and drops the forming candle),
|
||||
then quantifies how the feeds diverge:
|
||||
|
||||
* coverage: % shared timestamps, feed-only bars, where they cluster
|
||||
* OHLC |Δ| distribution (mean/median/p90/p95/p99/max, $ and bias)
|
||||
* |ΔClose| vs mean ATR14 (the scale the signal engine cares about)
|
||||
* threshold exceedance (how many bars differ by more than X $)
|
||||
* bar-direction agreement (sign of close−open)
|
||||
* close-return correlation on consecutive bars
|
||||
* per-hour and per-month drift
|
||||
|
||||
Both feeds are mid-price (Oanda mid of MBA candles; Tiingo resampled mid;
|
||||
Twelve Data its own mid). Frames are cached to parquet so re-runs are
|
||||
instant; pass --no-cache to force a fresh fetch.
|
||||
|
||||
Usage:
|
||||
venv/bin/python scripts/feed_divergence_study.py --a oanda --b tiingo \
|
||||
--tfs M15,H1,H4 --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 PlatformConfig, Timeframe, XAUUSD
|
||||
from providers import make_data_provider
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
CACHE_DIR = REPO / "backtest" / "cache"
|
||||
RESEARCH_DIR = REPO / "backtest" / "research"
|
||||
|
||||
_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,
|
||||
})
|
||||
|
||||
|
||||
def wilder_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
|
||||
hl = df["High"] - df["Low"]
|
||||
hc = (df["High"] - df["Close"].shift(1)).abs()
|
||||
lc = (df["Low"] - df["Close"].shift(1)).abs()
|
||||
tr = pd.concat([hl, hc, lc], axis=1).max(axis=1)
|
||||
return tr.ewm(alpha=1.0 / period, adjust=False).mean()
|
||||
|
||||
|
||||
def fmt_stats(name: str, s: pd.Series) -> str:
|
||||
a = s.abs()
|
||||
return (f" {name:<6} mean ${a.mean():.3f} med ${a.median():.3f} "
|
||||
f"p90 ${a.quantile(0.90):.3f} p95 ${a.quantile(0.95):.3f} "
|
||||
f"p99 ${a.quantile(0.99):.3f} max ${a.max():.2f} "
|
||||
f"bias {s.mean():+.4f}")
|
||||
|
||||
|
||||
def cache_path(name: str, tf: Timeframe, start: pd.Timestamp,
|
||||
end: pd.Timestamp) -> Path:
|
||||
return (CACHE_DIR /
|
||||
f"{name}_XAU_USD_{tf.value}_{start:%Y%m%d}_{end:%Y%m%d}.parquet")
|
||||
|
||||
|
||||
def load_or_fetch(provider, name: str, tf: Timeframe, start, end,
|
||||
no_cache: bool):
|
||||
cp = cache_path(name, tf, start, end)
|
||||
if not no_cache and cp.exists():
|
||||
df = pd.read_parquet(cp)
|
||||
print(f" [{name}] loaded {len(df):,} bars from cache "
|
||||
f"{cp.name}")
|
||||
return df
|
||||
t0 = time.time()
|
||||
print(f" [{name}] fetching {tf.value} {start.date()} → {end.date()} ...")
|
||||
df = provider.fetch_range(XAUUSD, tf, start, end)
|
||||
print(f" [{name}] fetched {len(df):,} bars in {time.time() - t0:.1f}s")
|
||||
df.to_parquet(cp)
|
||||
print(f" [{name}] cached → {cp.name}")
|
||||
return df
|
||||
|
||||
|
||||
def compare(tf: Timeframe, a_name: str, b_name: str, a: pd.DataFrame,
|
||||
b: pd.DataFrame):
|
||||
print("\n" + "=" * 80)
|
||||
print(f"{tf.value} — {a_name.upper()} vs {b_name.upper()}, XAU/USD")
|
||||
print("=" * 80)
|
||||
|
||||
common = a.index.intersection(b.index)
|
||||
a_only = a.index.difference(b.index)
|
||||
b_only = b.index.difference(a.index)
|
||||
|
||||
print(f"\nCOVERAGE")
|
||||
print(f" {a_name} bars : {len(a):>8,} head {a.index[0]} tail {a.index[-1]}")
|
||||
print(f" {b_name} bars : {len(b):>8,} head {b.index[0]} tail {b.index[-1]}")
|
||||
print(f" shared : {len(common):>8,} "
|
||||
f"({100 * len(common) / max(len(a.index.union(b.index)), 1):.2f}% of union)")
|
||||
print(f" {a_name}-only : {len(a_only):>8,} "
|
||||
f"({100 * len(a_only) / max(len(a), 1):.2f}% of {a_name})")
|
||||
print(f" {b_name}-only : {len(b_only):>8,} "
|
||||
f"({100 * len(b_only) / max(len(b), 1):.2f}% of {b_name})")
|
||||
|
||||
if len(common) == 0:
|
||||
print(" !! no shared timestamps — feeds on different bar grids")
|
||||
return
|
||||
|
||||
for label, idx in ((f"{a_name}-only", a_only),
|
||||
(f"{b_name}-only", b_only)):
|
||||
if not len(idx):
|
||||
continue
|
||||
by_date = pd.Series(1, index=idx).groupby(idx.date).sum()
|
||||
top = by_date.sort_values(ascending=False).head(6)
|
||||
print(f" {label} clusters on: "
|
||||
+ ", ".join(f"{d} ({n})" for d, n in top.items()))
|
||||
|
||||
x = a.loc[common]
|
||||
y = b.loc[common]
|
||||
|
||||
atr = wilder_atr(x).dropna()
|
||||
mean_price = x["Close"].mean()
|
||||
print(f"\nSCALE REFERENCES ({a_name} feed)")
|
||||
print(f" mean close : ${mean_price:,.2f}")
|
||||
print(f" mean bar range: ${(x['High'] - x['Low']).mean():.3f}")
|
||||
print(f" mean ATR14 : ${atr.mean():.3f} median ${atr.median():.3f}")
|
||||
|
||||
print(f"\nPER-BAR DIFFS ({a_name} − {b_name}, $)")
|
||||
diffs = {}
|
||||
for col in ("Open", "High", "Low", "Close"):
|
||||
d = x[col] - y[col]
|
||||
diffs[col] = d
|
||||
print(fmt_stats(col, d))
|
||||
|
||||
dclose = diffs["Close"].abs()
|
||||
print(f"\n |ΔClose| as % of mean ATR14 : "
|
||||
f"mean {100 * dclose.mean() / atr.mean():.1f}% "
|
||||
f"p95 {100 * dclose.quantile(0.95) / atr.mean():.1f}%")
|
||||
|
||||
print(f"\nTHRESHOLD EXCEEDANCE (|ΔClose|)")
|
||||
for thr in (0.10, 0.30, 0.65, 1.00, 2.00, 5.00):
|
||||
pct = 100 * (dclose > thr).mean()
|
||||
print(f" > ${thr:<5.2f}: {pct:6.2f}% of bars ({(dclose > thr).sum():,})")
|
||||
|
||||
# direction agreement
|
||||
a_dir = np.sign(x["Close"] - x["Open"])
|
||||
b_dir = np.sign(y["Close"] - y["Open"])
|
||||
both_nonflat = (a_dir != 0) & (b_dir != 0)
|
||||
dir_agree = 100 * (a_dir[both_nonflat] == b_dir[both_nonflat]).mean()
|
||||
print(f"\nAGREEMENT")
|
||||
print(f" bar direction (sign of close−open) agree: {dir_agree:.2f}%"
|
||||
f" (n={both_nonflat.sum():,})")
|
||||
|
||||
step = pd.Series(common, index=common).diff().median()
|
||||
a_ret = x["Close"].pct_change()
|
||||
b_ret = y["Close"].pct_change()
|
||||
step_ok = pd.Series(common, index=common).diff().le(step + pd.Timedelta(seconds=1))
|
||||
corr = a_ret[step_ok].corr(b_ret[step_ok])
|
||||
print(f" close-return correlation (consecutive bars): {corr:.5f}")
|
||||
|
||||
# by-hour
|
||||
hours = common.hour
|
||||
rows = []
|
||||
for h in range(24):
|
||||
m = hours == h
|
||||
if not m.any():
|
||||
continue
|
||||
rows.append({"hour_utc": h, "bars": int(m.sum()),
|
||||
"mean_abs_dclose": dclose[m].mean(),
|
||||
"p95_abs_dclose": dclose[m].quantile(0.95)})
|
||||
by_hour = pd.DataFrame(rows)
|
||||
print(f"\nBY UTC HOUR (mean |ΔClose|, $)")
|
||||
print(by_hour.to_string(index=False,
|
||||
float_format=lambda v: f"{v:.3f}"))
|
||||
|
||||
print(f"\nBY MONTH (mean |ΔClose|, $)")
|
||||
months = dclose.groupby(common.tz_localize(None).to_period("M")).mean()
|
||||
for p, v in months.items():
|
||||
print(f" {p}: {v:.3f}")
|
||||
|
||||
return {
|
||||
"tf": tf.value, "a": a_name, "b": b_name,
|
||||
"a_bars": len(a), "b_bars": len(b), "common": len(common),
|
||||
"a_only": len(a_only), "b_only": len(b_only),
|
||||
"dir_agree_pct": round(dir_agree, 3),
|
||||
"ret_corr": round(corr, 5),
|
||||
"mean_abs_dclose": dclose.mean(),
|
||||
"p95_abs_dclose": dclose.quantile(0.95),
|
||||
"max_abs_dclose": dclose.max(),
|
||||
"dclose_pct_atr_mean": 100 * dclose.mean() / atr.mean(),
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--a", required=True)
|
||||
ap.add_argument("--b", required=True)
|
||||
ap.add_argument("--tfs", default="M15,H1,H4",
|
||||
help="comma-separated timeframe values (default M15,H1,H4)")
|
||||
ap.add_argument("--days", type=int, default=365)
|
||||
ap.add_argument("--end", default=None, help="UTC end date (default now)")
|
||||
ap.add_argument("--no-cache", action="store_true")
|
||||
args = ap.parse_args()
|
||||
|
||||
tfs = []
|
||||
for s in args.tfs.split(","):
|
||||
s = s.strip()
|
||||
tf = _TF_LOOKUP.get(s)
|
||||
if tf is None:
|
||||
sys.exit(f"unknown timeframe '{s}'; use one of "
|
||||
f"{sorted(_TF_LOOKUP)}")
|
||||
tfs.append(tf)
|
||||
|
||||
end = pd.Timestamp(args.end, tz="UTC") if args.end \
|
||||
else pd.Timestamp.now(tz="UTC")
|
||||
start = end - pd.Timedelta(days=args.days)
|
||||
|
||||
cfg = PlatformConfig()
|
||||
pa = make_data_provider(cfg, args.a)
|
||||
pb = make_data_provider(cfg, args.b)
|
||||
|
||||
summary = []
|
||||
for tf in tfs:
|
||||
print(f"\n{'─' * 80}\nWINDOW {start.date()} → {end.date()} "
|
||||
f"({args.days} days), {tf.value}")
|
||||
try:
|
||||
a = load_or_fetch(pa, args.a, tf, start, end, args.no_cache)
|
||||
except Exception as e:
|
||||
print(f" !! {args.a} {tf.value} fetch failed: {e}")
|
||||
continue
|
||||
try:
|
||||
b = load_or_fetch(pb, args.b, tf, start, end, args.no_cache)
|
||||
except Exception as e:
|
||||
print(f" !! {args.b} {tf.value} fetch failed: {e}")
|
||||
continue
|
||||
r = compare(tf, args.a, args.b, a, b)
|
||||
if r:
|
||||
summary.append(r)
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
print("SUMMARY TABLE")
|
||||
print("=" * 80)
|
||||
sdf = pd.DataFrame(summary)
|
||||
print(sdf.to_string(index=False,
|
||||
float_format=lambda v: f"{v:.4f}"))
|
||||
out = RESEARCH_DIR / f"feed_divergence_{args.a}_vs_{args.b}.csv"
|
||||
if len(sdf):
|
||||
sdf.to_csv(out, index=False, float_format="%.6f")
|
||||
print(f"\nSummary written to {out}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -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()
|
||||
@@ -0,0 +1,246 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user