diff --git a/CLAUDE.md b/CLAUDE.md
index eeb74bd..d0ff9c2 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -69,7 +69,7 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz
 
 ## Rules for making changes (read before editing signal logic)
 
-1. **Never change a tuned value without re-validation.** Acceptance rule: positive total R in BOTH train and validation halves of a ≥12-month window, PF ≥ 1.3, at realistic cost (XAU $0.30 round trip), trailing exits, next-open fills. Marginal edges must additionally pass the 2024-06→2025-03 out-of-sample window — within-year splits alone have approved configs that later failed OOS.
+1. **Never change a tuned value without re-validation.** Acceptance rule: positive total R in BOTH train and validation halves of a ≥12-month window, PF ≥ 1.3, at realistic cost (XAU $0.65 round trip headline; retain $0.30 for TD-era cross-book comparability), trailing exits, next-open fills. `round_trip_cost` stays $0.30 because it also drives the live TP spread gate; changing it is a separate rule-1 event. Marginal edges must additionally pass the 2024-06→2025-03 out-of-sample window — within-year splits alone have approved configs that later failed OOS.
 2. **Per-timeframe overrides live in the pair factory**, never in global defaults — the same knob has opposite effects on different TFs (proven repeatedly: `tp_floor_rr`, hour-9 block, trail params). Document validation evidence in the factory docstring.
 3. **Pair isolation is a hard invariant.** XAU gates in the aggregator are display-name-gated so a future pair port cannot inherit gold-calibrated logic by accident.
 4. **Live == backtest parity.** If you add a live gate, add it to the replay path in the same commit; all time filters evaluate bar-close time in both paths.
@@ -81,12 +81,15 @@ Signal currency everywhere is `SignalStrength` (−2..+2). DataFrames are UTC tz
 
 ## 2026-08-04 QA audit — owner decisions ledger
 
-- **Weekend calendar: FIXED (owner-approved 2026-08-04).** All three `_is_forex_weekend` copies are NY-anchored (Fri 17:00 → Sun 18:00 ET; during US DST that is Fri 21:00 UTC — measured on 5yr of Oanda bars). Kills the weekly false-"STALE DATA/feed lagging" hours. Book impact, fully accounted: M15 byte-identical; M5 12mo dropped exactly ONE trade — a phantom whose signal fired on the final summer-Friday bar (close 21:00 UTC) and "filled" at Sunday's reopen 49 h later (−1.08R, live-impossible). M5 12mo reference at HEAD: 555 trades, +98.64R PF≈1.50 @ $0.65 (was 556/+97.56/1.48). A Module 21 test sweeps all three copies hourly across both DST transitions.
+- **Weekend calendar: FIXED (owner-approved 2026-08-04).** All three `_is_forex_weekend` copies are NY-anchored (Fri 17:00 → Sun 18:00 ET; during US DST that is Fri 21:00 UTC — measured on 5yr of Oanda bars). Kills the weekly false-"STALE DATA/feed lagging" hours. Book impact at the audit HEAD: M15 byte-identical; M5 12mo dropped exactly ONE trade — a phantom whose signal fired on the final summer-Friday bar (close 21:00 UTC) and "filled" at Sunday's reopen 49 h later (−1.08R, live-impossible), leaving 555 trades / +98.64R / PF≈1.50 @ $0.65 (was 556/+97.56/1.48). The current-HEAD canonical reference is recorded below. A Module 21 test sweeps all three copies hourly across both DST transitions.
 - **Spread in exit-trigger detection (rule 6): owner DECLINED 2026-08-04.** Backtest stop triggers stay on raw mid High/Low; the measured optimism (~4–6% of total R: −3.55R @ $0.30, −4.24R @ $0.65 on the M15 12mo book) is accepted and books stay comparable. Re-open only as a conscious re-pin campaign.
 - **`FB2_EMA_LO`: HARDENED 2026-08-04 (delegated).** Research override for the EMA200-proximity band, default = tuned 0.5; an unparseable value now degrades to the default with a warning instead of crashing every scan. Documented here — it is a knob that exists.
 - **H4 seeding divergence: parked.** Live H4 fetches only 380 bars vs the backtest's unbounded history — ~2% of sampled H4 bars flip an indicator signal (composite Δ up to 12.3). M5/M15/H1 clean at their real fetch depths. Fold into any future H4 campaign; H4 remains "unproven for auto-execution".
 
-## Current state notes (2026-07-30)
+## Current state notes (2026-08-14)
 
-- Oanda v20 is the live feed (`data_providers.active: "oanda"`), but the pinned backtest books are still Twelve-Data-pinned and the backtest cost assumption is $0.30 while measured real spreads are ~$0.65–0.72 — re-pin books on Oanda bars before treating feed-switch comparisons as valid (see Oanda/TD divergence study in memory).
+- Oanda v20 is the live feed (`data_providers.active: "oanda"`). The 2026-08-04 re-pin made the 27 `backtest/oanda_*.csv` books the baseline; all TD-era books are superseded. Current-HEAD M5 canonical book: `backtest/oanda_m5_12mo_065_nopb_20260814.csv`, +106.82R PF 1.51 at $0.65 (both halves positive; PF 1.46 / 1.55). M15 reference: +73.93R PF 1.53. See `backtest/research/repin_verification_2026-08-14.md`.
+- Cost reporting is dual: $0.65 is the measured-Oanda realistic headline ($0.65-$0.72 round trip), while $0.30 remains for cross-book comparability and as the live TP spread-gate input. Raising `round_trip_cost` from $0.30 would move the TP floor from about $9 to $19.50 and requires full rule-1 revalidation.
+- Twelve Data is signal-incompatible with Oanda (Phase 1: about 19% M5 bar-direction disagreement and a systematic ~$1 high bias; see `backtest/research/feed_divergence_2026-08-14.md`). On an Oanda candle outage, halt auto-execution rather than switching to TD candles.
+- Tiingo is streaming-only (`data_providers.streaming: "tiingo"`), not a designated candle fallback. Oanda-Tiingo signal parity is documented in `backtest/research/signal_parity_2026-08-14.md`; Phase 4's two-week live shadow scan remains a manual operational step.
 - Human-facing docs in `docs/` (`user_guide.html`, `beginner_guide.html` + PDF builds) must be kept in sync when signal behavior changes.
diff --git a/config.py b/config.py
index c62b369..335026b 100755
--- a/config.py
+++ b/config.py
@@ -31,8 +31,10 @@ class SignalStrength(Enum):
 XAUUSD = "XAU/USD"
 
 # price_precision: decimals for SL/TP rounding and display.
-# default_spread: round-trip cost per trade in price units, charged by
-# the backtester when --spread is not given.
+# default_spread: $0.30 round-trip cost retained for TD-era cross-book
+# comparability and used when the backtester receives no --spread override.
+# Oanda's measured round trip is $0.65-$0.72; research reports headline it
+# separately with --spread 0.65.
 PAIR_INFO = {
     XAUUSD: {"display": "XAU/USD", "pip_value": 0.10,
              "price_precision": 2, "default_spread": 0.30},
@@ -110,7 +112,10 @@ class SignalConfig:
     tp_tightening_factor: float = 1.0
 
     # ── Trade geometry ────────────────────────────────────────────
-    # Round-trip transaction cost estimate in price units (XAU ~$0.30).
+    # Deliberately $0.30 for TD-era cross-book comparability. Oanda's measured
+    # round trip is $0.65-$0.72, reported separately in re-pin campaigns.
+    # This is also a LIVE gate input: increasing it to $0.65 raises the TP1
+    # floor from about $9 to $19.50, so it requires rule-1 revalidation.
     round_trip_cost: float = 0.30
     # Spread gate: reject setups whose TP1 distance is below
     # round_trip_cost × this multiple (caps cost at ~3% of target).
diff --git a/docs/user_guide.html b/docs/user_guide.html
index 8cb4e74..ddc7078 100644
--- a/docs/user_guide.html
+++ b/docs/user_guide.html
@@ -84,16 +84,16 @@
 </ol>
 
 <h2>4 · Choosing your timeframe</h2>
-<p>Twelve-month backtest (Jul 2025 – Jul 2026, real spreads, entry at next candle open):</p>
+<p>Twelve-month Oanda-candle backtest (Aug 2025 – Aug 2026, $0.65 round-trip cost, entry at next candle open):</p>
 <table>
 <tr><th></th><th>M5</th><th>M15</th><th>H1</th></tr>
-<tr><td>Trades / week</td><td>~10</td><td>~9.5</td><td>~3</td></tr>
-<tr><td>Total result (12 mo)</td><td>+135.6R</td><td>+67.2R</td><td>+49.9R</td></tr>
-<tr><td>Win rate / profit factor</td><td>59.7% / 1.81</td><td>55.8% / 1.40</td><td>59.4% / 1.94</td></tr>
-<tr><td>Typical (median) hold</td><td>2.4 h</td><td>6 h</td><td>12.5 h</td></tr>
-<tr><td>90% of trades close within</td><td>16 h</td><td>40 h</td><td>3.7 days</td></tr>
-<tr><td>Typical stop distance</td><td>~$19</td><td>~$29</td><td>~$44</td></tr>
-<tr><td>Worst drawdown</td><td>11.2R</td><td>9.4R</td><td>9.5R</td></tr>
+<tr><td>Trades / week</td><td>~11.3</td><td>~7.6</td><td>~2.7</td></tr>
+<tr><td>Total result (12 mo)</td><td>+106.8R</td><td>+73.9R</td><td>+24.0R</td></tr>
+<tr><td>Win rate / profit factor</td><td>52.9% / 1.51</td><td>62.2% / 1.53</td><td>60.9% / 1.55</td></tr>
+<tr><td>Typical (median) hold</td><td>2.4 h</td><td>7 h</td><td>15.5 h</td></tr>
+<tr><td>90% of trades close within</td><td>10.5 h</td><td>29.8 h</td><td>3.2 days</td></tr>
+<tr><td>Typical stop distance</td><td>~$21</td><td>~$32</td><td>~$52</td></tr>
+<tr><td>Worst drawdown</td><td>10.7R</td><td>10.8R</td><td>8.0R</td></tr>
 </table>
 <ul>
 <li><b>M5 — intraday, execution-critical.</b> Highest total edge, but stops are tight: a couple of dollars of slippage or a late entry eats a meaningful share of every trade. Suitable only if you can act within a minute or two of the alert (or automate). Trades open and close within a session; you sleep flat.</li>
@@ -104,7 +104,7 @@
 
 <h2>5 · What to expect (and what not to)</h2>
 <ul>
-<li><b>You will lose ~40% of trades.</b> A 56–60% win rate means losing streaks of 5+ are routine over a year. The system's edge is the average across hundreds of trades, not any single one.</li>
+<li><b>You will lose roughly 38–47% of trades.</b> A 53–62% win rate means losing streaks of 5+ are routine over a year. The system's edge is the average across hundreds of trades, not any single one.</li>
 <li><b>Stops hit fast, winners take time.</b> Losing trades die in roughly half the time winners take. An early stop-out is the system doing its job.</li>
 <li><b>Drawdowns of ~10R are normal.</b> At 1% risk per trade, expect to be ~10% down from a peak at some point in a normal year. This is why risking more than 1% per trade is how accounts die.</li>
 <li><b>Backtest ≠ live.</b> Results above charge real spreads and realistic fills, but cannot model your personal latency, broker slippage, or requotes. Assume live results will be somewhat worse — especially on M5.</li>
diff --git a/providers/tiingo.py b/providers/tiingo.py
index 797fc6f..8337090 100644
--- a/providers/tiingo.py
+++ b/providers/tiingo.py
@@ -3,6 +3,8 @@ ForexBot2 — Tiingo data provider
 =================================
 OHLCV candles from Tiingo's FX endpoint
 (`/tiingo/fx/{ticker}/prices`, resampled mid-price bars, no volume).
+Tiingo is streaming-only in production: its REST candles remain available
+for comparison research, but it is not a designated candle fallback.
 All shared behavior (cache, validation, boundary guarantees, range
 pagination) lives in `BaseDataProvider`; this module only maps
 symbols/resample frequencies, sizes the request window, and parses
diff --git a/scripts/compare_oanda_tiingo.py b/scripts/compare_oanda_tiingo.py
new file mode 100644
index 0000000..b9a9ba1
--- /dev/null
+++ b/scripts/compare_oanda_tiingo.py
@@ -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())
diff --git a/scripts/feed_divergence_study.py b/scripts/feed_divergence_study.py
new file mode 100644
index 0000000..0488ef8
--- /dev/null
+++ b/scripts/feed_divergence_study.py
@@ -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()
diff --git a/scripts/shadow_scan.py b/scripts/shadow_scan.py
new file mode 100644
index 0000000..f031a4c
--- /dev/null
+++ b/scripts/shadow_scan.py
@@ -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()
diff --git a/scripts/signal_parity_study.py b/scripts/signal_parity_study.py
new file mode 100644
index 0000000..c7dd7da
--- /dev/null
+++ b/scripts/signal_parity_study.py
@@ -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()
diff --git a/signal_log.py b/signal_log.py
index f1fff9d..72115dc 100644
--- a/signal_log.py
+++ b/signal_log.py
@@ -60,20 +60,34 @@ CREATE TABLE IF NOT EXISTS signals (
     news_description TEXT,
     news_summary TEXT,
     events_json TEXT,
-    headlines_json TEXT
+    headlines_json TEXT,
+    panel_verdict TEXT,
+    panel_confidence REAL,
+    panel_rationale TEXT,
+    panel_takes_json TEXT
 );
 CREATE INDEX IF NOT EXISTS idx_signals_sym_tf_ts
     ON signals (symbol, timeframe, logged_at);
 """
 
+# Additive migration: older DBs created before the panel columns existed
+# get them added here. Idempotent — skips columns already present.
+_PANEL_COLS = (
+    ("panel_verdict", "TEXT"),
+    ("panel_confidence", "REAL"),
+    ("panel_rationale", "TEXT"),
+    ("panel_takes_json", "TEXT"),
+)
+
 _INSERT = """
 INSERT INTO signals (
     logged_at, source, symbol, timeframe, signal, score, confidence,
     quality, action, price, spread, market_regime, volatility_regime,
     mtf_json, setup_json, reasons_json, warnings_json,
     event_risk, event_description, news_sentiment, news_description,
-    news_summary, events_json, headlines_json
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    news_summary, events_json, headlines_json,
+    panel_verdict, panel_confidence, panel_rationale, panel_takes_json
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 """
 
 
@@ -207,6 +221,8 @@ def _report_row(
     events_json, headlines_json = _context_news_json(context)
 
     signal = getattr(report, "final_signal", None)
+    panel_takes = getattr(report, "panel_takes", None) or []
+    panel_takes_json = _dumps(panel_takes) if panel_takes else None
     return (
         datetime.now(timezone.utc).isoformat(),
         source,
@@ -232,6 +248,10 @@ def _report_row(
         getattr(report, "news_summary", "") or "",
         events_json,
         headlines_json,
+        getattr(report, "panel_verdict", None),
+        _finite(getattr(report, "panel_confidence", None)),
+        getattr(report, "panel_rationale", None),
+        panel_takes_json,
     )
 
 
@@ -251,11 +271,21 @@ class SignalLogger:
         self._lock = threading.Lock()
         with self._lock:
             self._conn.executescript(_SCHEMA)
+            self._migrate_panel_columns()
 
     @property
     def path(self) -> str:
         return self._path
 
+    def _migrate_panel_columns(self) -> None:
+        """Add panel_* columns to pre-existing tables (idempotent)."""
+        existing = {r[1] for r in self._conn.execute("PRAGMA table_info(signals)")}
+        for name, col_type in _PANEL_COLS:
+            if name not in existing:
+                self._conn.execute(
+                    f"ALTER TABLE signals ADD COLUMN {name} {col_type}")
+        self._conn.commit()
+
     def log(
         self,
         report: Any,