#!/usr/bin/env python3 """ analyze_signal_log.py — signal_log performance evaluator ========================================================= READ-ONLY analysis of logs/signals.db: how are the logged live signals and the advisory LLM panel actually performing? Three lenses: 1. TRADE REPLAY — every logged row carrying a trade_setup is deduped (telegram + dashboard log the same scan; a persisting signal re-fires every bar) into unique signal EVENTS, then replayed against the RAW Oanda price path with the backtester's own resolution code (backtest.gpu_backtest.simulate_trades): entry at the next bar open after the scan boundary, SL/TP distances preserved, SL-first ties, gap fills at bar open, trail ratchet at bar close using the trail params stamped on the setup. Dual cost reporting per repo convention: $0.65 round trip headline, $0.30 derived analytically. 2. PANEL SCORECARD — for every row with a panel verdict, the counterfactual: what did/would the underlying trade do? OPPOSE on a loser = R saved; OPPOSE on a winner = R cost. 3. DIRECTIONAL HIT-RATE — non-setup BUY/SELL rows scored by forward close-to-close return at +1h/+4h/+24h vs signal direction. 4. SURFACE PARITY (--parity) — telegram vs dashboard rows on the same (timeframe, signal bar): signal / grade / setup / action / gate config fingerprint must agree; a divergence means the two processes ran different configs (2026-08-19: the dashboard applied a different session filter for a day). Also reports rows-per-bar per source (the over-scanning symptom). 5. STORED-CANDLE REPLAY (--stored-candles) — the same event replay, but resolved on the closed-candle streams stored in the DB itself (no network, no parquet): identical model to advisory_outcome.py, plus the "opportunity" (first event per tf+direction cluster) and one-position-per-timeframe views. 6. NEWS SENTIMENT (--news, with --stored-candles) — is the logged news_sentiment score a leading or a trailing signal? Hourly mean sentiment vs past/forward price moves at 1/4/8/24 h (split-half on the forward number), the daily open->close read, what it said at the window's high and low, and resolved-event R bucketed by direction x news bucket x macro-persona stance. Warning-only feature: nothing here gates a signal. The DB is opened read-only (mode=ro — WAL-visible; the older immutable=1 URI silently missed the newest rows). Bar data for the default replay comes from the active provider via the backtester's fetch_history (parquet-cached). Sample-size caveat: the log started 2026-08-13, so results are directional, not statistically significant. Usage: python analyze_signal_log.py # full report python analyze_signal_log.py --csv out.csv # + per-event CSV python analyze_signal_log.py --no-fetch # cache only python analyze_signal_log.py --parity # surface parity only python analyze_signal_log.py --stored-candles # DB-only replay python analyze_signal_log.py --stored-candles --news # + news lens """ import argparse import json import math import os import sqlite3 import sys from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional, Sequence, Tuple import pandas as pd sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import Timeframe # noqa: E402 # Dual-cost convention (CLAUDE.md): $0.65 measured-Oanda headline, # $0.30 retained for cross-book comparability. COST_HEADLINE = 0.65 COST_COMPARE = 0.30 DEFAULT_DB = os.path.join("logs", "signals.db") MAX_BARS_DEFAULT = 300 # matches the backtester's trailing default CLUSTER_GAP_DEFAULT_MIN = 60 # same setup re-logged within this = one event # ── Data model ──────────────────────────────────────────────────────── @dataclass class LogRow: """One signals.db row, decoded.""" id: int logged_at: datetime source: str timeframe: str signal: str score: Optional[float] quality: str price: Optional[float] market_regime: str setup: Optional[dict] panel_verdict: str panel_confidence: Optional[float] action: str = "" bar_ts: Optional[str] = None # primary-TF signal bar label gate_reason: str = "" actionable: Optional[int] = None config_fingerprint: str = "" advisory_event_id: Optional[int] = None advisory_status: str = "" advisory_r: Optional[float] = None scan_kind: str = "" news_sentiment: Optional[float] = None event_risk: str = "" @dataclass class SignalEvent: """A deduped actionable signal (cluster of identical re-logs).""" rows: List[LogRow] representative: LogRow # FIRST occurrence — when a trader acts panel_verdict: str = "" panel_confidence: Optional[float] = None # Filled by replay: exit_reason: str = "" exit_time: Optional[pd.Timestamp] = None bars_held: int = 0 r_headline: Optional[float] = None # R at $0.65 round trip r_compare: Optional[float] = None # R at $0.30 round trip mfe_r: Optional[float] = None mae_r: Optional[float] = None entry_fill: Optional[float] = None # actual next-open fill # ── Loading (immutable — never writes to the DB) ───────────────────── _COLS = ("id, logged_at, source, timeframe, signal, score, quality," " price, market_regime, setup_json, panel_verdict," " panel_confidence, action") # Columns added by the 2026-09 observability work; absent on older DBs. _OPTIONAL_COLS = ("gate_reason", "actionable", "config_fingerprint", "advisory_event_id", "advisory_outcome_status", "advisory_r_multiple", "scan_kind", "news_sentiment", "event_risk") # The aggregator's own "News:" reason threshold — the band outside which a # logged sentiment score is called bullish/bearish for gold. NEWS_BULL = 0.1 NEWS_BEAR = -0.1 NEWS_LAGS_H = (1, 4, 8, 24) def open_readonly(db_path: str) -> sqlite3.Connection: """Read-only connection that still sees WAL-resident rows. ``immutable=1`` (the original choice) ignores the -wal file, so the rows the live loggers wrote since the last checkpoint were invisible. ``mode=ro`` needs the -shm/-wal files to be readable, which they are while a writer exists; fall back to immutable only if that fails.""" path = os.path.abspath(db_path) try: con = sqlite3.connect(f"file:{path}?mode=ro", uri=True) con.execute("SELECT 1 FROM sqlite_master LIMIT 1").fetchall() return con except sqlite3.OperationalError: print("WARNING: mode=ro open failed — falling back to immutable=1 " "(newest WAL rows may be missing)") return sqlite3.connect(f"file:{path}?immutable=1", uri=True) def load_rows(db_path: str) -> List[LogRow]: """Read every signals row (read-only, WAL-visible).""" con = open_readonly(db_path) try: have = {r[1] for r in con.execute("PRAGMA table_info(signals)")} opt = [c for c in _OPTIONAL_COLS if c in have] cols = _COLS + ("".join(", " + c for c in opt)) bar_expr = (", json_extract(indicators_json, '$.\"' || timeframe" " || '\".timestamp')" if "indicators_json" in have else ", NULL") out: List[LogRow] = [] for r in con.execute(f"SELECT {cols}{bar_expr} FROM signals" " ORDER BY id"): setup = json.loads(r[9]) if r[9] else None extra = dict(zip(opt, r[13:13 + len(opt)])) out.append(LogRow( id=r[0], logged_at=datetime.fromisoformat(r[1]), source=r[2], timeframe=r[3], signal=r[4], score=r[5], quality=r[6] or "", price=r[7], market_regime=r[8] or "", setup=setup, panel_verdict=r[10] or "", panel_confidence=r[11], action=r[12] or "", bar_ts=r[-1], gate_reason=extra.get("gate_reason") or "", actionable=extra.get("actionable"), config_fingerprint=extra.get("config_fingerprint") or "", advisory_event_id=extra.get("advisory_event_id"), advisory_status=extra.get("advisory_outcome_status") or "", advisory_r=extra.get("advisory_r_multiple"), scan_kind=extra.get("scan_kind") or "", news_sentiment=(float(extra["news_sentiment"]) if extra.get("news_sentiment") is not None else None), event_risk=extra.get("event_risk") or "", )) return out finally: con.close() # ── Dedup / clustering ──────────────────────────────────────────────── def _setup_key(row: LogRow) -> Tuple: """Delegates to signal_log.setup_key — the SAME identity the logger uses for advisory-event dedup, so both tools cluster identically.""" from signal_log import setup_key return (row.timeframe, setup_key(row.setup, row.timeframe)) def cluster_setup_rows( rows: Sequence[LogRow], gap: timedelta = timedelta(minutes=CLUSTER_GAP_DEFAULT_MIN), ) -> List[SignalEvent]: """Collapse re-logs of the same setup into one SignalEvent. Two layers of duplication exist in the log: - telegram AND dashboard both log the same scan (seconds apart); - a persisting signal re-fires every bar with an identical setup (live, the one-position rule means it is taken at most once). Rows with an identical (tf, direction, entry, SL, strategy) key and consecutive-log gaps <= `gap` are ONE event; the representative is the FIRST occurrence (when a trader/daemon would have acted). """ events: List[SignalEvent] = [] last_by_key: Dict[Tuple, SignalEvent] = {} for row in sorted((r for r in rows if r.setup), key=lambda r: r.id): key = _setup_key(row) ev = last_by_key.get(key) if ev is not None and row.logged_at - ev.rows[-1].logged_at <= gap: ev.rows.append(row) else: ev = SignalEvent(rows=[row], representative=row) events.append(ev) last_by_key[key] = ev # Any panel verdict on any member row attaches to the event. if row.panel_verdict and not ev.panel_verdict: ev.panel_verdict = row.panel_verdict ev.panel_confidence = row.panel_confidence return events def _row_index(events: Sequence[SignalEvent]) -> Dict[int, SignalEvent]: """row id -> the SignalEvent it belongs to (every row of an event shares that event's replayed outcome).""" return {r.id: ev for ev in events for r in ev.rows} # ── Trade construction + replay (backtester realism, reused code) ──── def entry_bar_open(ts: datetime, tf: Timeframe) -> pd.Timestamp: """The bar opening at the scan boundary just before/at `ts`. Live scans run at boundary+buffer, so flooring the log time to the TF boundary gives the entry bar — identical to the backtester's "first raw bar strictly after the signal bar". """ secs = {Timeframe.M5: 300, Timeframe.M15: 900, Timeframe.M30: 1800, Timeframe.H1: 3600, Timeframe.H4: 14400, Timeframe.D1: 86400}[tf] epoch = int(ts.timestamp()) return pd.Timestamp(epoch - epoch % secs, unit="s", tz="UTC") def build_trade(spec: LogRow, raw_df: pd.DataFrame): """LogRow with setup -> gpu_backtest.Trade at the next-open fill. SL/TP distances from the logged setup are preserved relative to the actual fill, exactly like the backtester's replay loop. Returns (trade, risk_distance) or (None, 0) when no bar exists at/after the entry boundary (signal too recent). """ from backtest.gpu_backtest import Trade s = spec.setup tf = Timeframe(spec.timeframe) want = entry_bar_open(spec.logged_at, tf) pos = int(raw_df.index.searchsorted(want)) if pos >= len(raw_df): return None, 0.0 entry_time = raw_df.index[pos] entry_price = float(raw_df["Open"].iloc[pos]) direction = s["direction"] log_entry = float(s["entry"]) sl_dist = abs(log_entry - float(s["stop_loss"])) tp1_dist = abs(float(s["take_profit_1"]) - log_entry) tp2_dist = abs(float(s["take_profit_2"]) - log_entry) if sl_dist <= 0: return None, 0.0 sign = 1.0 if direction == "BUY" else -1.0 risk_pips = float(s.get("risk_pips") or 0) pip_value = sl_dist / risk_pips if risk_pips > 0 else 0.1 trade = Trade( pair="XAU/USD", entry_time=entry_time, entry_price=entry_price, sl=entry_price - sign * sl_dist, tp1=entry_price + sign * tp1_dist, tp2=entry_price + sign * tp2_dist, direction=direction, pip_value=pip_value, risk_pips=risk_pips, reward_pips=float(s.get("reward_pips") or 0), rr_planned=float(s.get("risk_reward") or 0), quality=spec.quality, final_score=float(spec.score or 0), final_confidence=0.0, market_regime=spec.market_regime, strategy_tag=s.get("strategy") or "GENERIC", trail_arm_r=float(s.get("trail_arm_r") or 0), trail_dist_r=float(s.get("trail_dist_r") or 0), ) return trade, sl_dist def replay_events( events: List[SignalEvent], raw_by_tf: Dict[Timeframe, pd.DataFrame], max_bars: int = MAX_BARS_DEFAULT, spread: float = COST_HEADLINE, ) -> None: """Resolve every event on the raw price path and stamp R multiples. Reuses backtest.gpu_backtest.simulate_trades verbatim: raw-path resolution, SL-first ties, gap fills at open, close-only trail ratchet with the per-setup trail params as per-trade overrides. The $0.30 book is derived analytically (R is linear in cost). """ from backtest.gpu_backtest import simulate_trades trades, owners, risks = [], [], [] for ev in events: tf = Timeframe(ev.representative.timeframe) raw = raw_by_tf.get(tf) if raw is None or raw.empty: ev.exit_reason = "NO_DATA" continue trade, risk = build_trade(ev.representative, raw) if trade is None: ev.exit_reason = "PENDING" # no bar at/after the boundary continue trades.append(trade) owners.append(ev) risks.append(risk) # simulate per-TF so each trade's window indexes the right frame. # Run-level trail params are placeholders: every trailing trade # carries its own stamped overrides, which win inside # _simulate_trailing. Static trades (arm=0) resolve SL/TP1. # Group by (TF, exit mode): a static trade inside a trailing call # would wrongly inherit the run-level arm. groups: Dict[Tuple[Timeframe, bool], list] = {} for t, ev in zip(trades, owners): key = (Timeframe(ev.representative.timeframe), t.trail_arm_r > 0) groups.setdefault(key, []).append(t) for (gtf, is_trailing), gtrades in groups.items(): simulate_trades(gtrades, raw_by_tf[gtf], max_bars=max_bars, spread=spread, trail_arm_r=0.6 if is_trailing else 0.0, trail_dist_r=0.75) for trade, ev, risk in zip(trades, owners, risks): ev.entry_fill = trade.entry_price ev.exit_time = trade.exit_time ev.bars_held = trade.bars_held ev.mfe_r = trade.mfe_r ev.mae_r = trade.mae_r if trade.exit_reason in ("NO_DATA", ""): ev.exit_reason = "NO_DATA" continue # Data ran out before max_bars → position still open, marked to # market; distinct from a true 300-bar TIMEOUT. if trade.exit_reason == "TIMEOUT" and trade.bars_held < max_bars: ev.exit_reason = "OPEN" else: ev.exit_reason = trade.exit_reason ev.r_headline = trade.r_multiple ev.r_compare = trade.r_multiple + (spread - COST_COMPARE) / risk # ── Panel scorecard ─────────────────────────────────────────────────── def panel_scorecard( rows: Sequence[LogRow], events: Sequence[SignalEvent], m5: Optional[pd.DataFrame], ) -> List[dict]: """Counterfactual per panel verdict: did it save or cost R? Verdicted rows with a setup join their SignalEvent outcome; rows without a setup are scored on the +4h forward return instead. """ by_row = _row_index(events) out = [] for row in rows: if not row.panel_verdict: continue entry = {"id": row.id, "logged_at": row.logged_at, "timeframe": row.timeframe, "signal": row.signal, "quality": row.quality, "verdict": row.panel_verdict, "confidence": row.panel_confidence, "basis": "", "r_headline": None, "note": ""} ev = by_row.get(row.id) if ev is not None and ev.r_headline is not None: entry["basis"] = "trade_replay" entry["r_headline"] = ev.r_headline entry["note"] = f"exit={ev.exit_reason}" elif m5 is not None and row.price: fwd = _forward_return(m5, row.logged_at, row.price, hours=4) if fwd is not None: sign = 1 if row.signal in ("BUY", "STRONG_BUY") else -1 entry["basis"] = "fwd_4h_return" entry["r_headline"] = None entry["note"] = f"4h move {sign * fwd:+.2f}$ in dir" out.append(entry) return out # ── Directional hit-rate (non-setup signals) ────────────────────────── def _forward_return( df: pd.DataFrame, ts: datetime, price: float, hours: float ) -> Optional[float]: """Close-to-close move from `price` to the last close at ts+hours.""" target = pd.Timestamp(ts) + pd.Timedelta(hours=hours) closes = df["Close"] # The horizon must have ELAPSED. Without this the searchsorted below # clamps to the last available bar, so every row logged within # `hours` of the end of the data reported the latest close as its # "forward" move — a near-zero reading for a trade that has had no # time to develop, blended into the hit-rate/mean-move aggregate on # every run. Not-yet-known is None, the same as insufficient history. if len(df.index) == 0 or df.index[-1] < target: return None pos = int(df.index.searchsorted(target, side="right")) - 1 if pos < 0: return None return float(closes.iloc[pos]) - price def directional_hit_rate( rows: Sequence[LogRow], m5: pd.DataFrame, horizons_h: Sequence[float] = (1, 4, 24), ) -> List[dict]: """Forward-return direction check for non-setup BUY/SELL rows. Cross-source dupes collapse on (tf, signal, 5-minute bucket). Returns per-horizon aggregates plus the per-row detail under 'rows'. """ seen = set() detail = [] for row in sorted(rows, key=lambda r: r.id): if row.setup or row.signal not in ( "BUY", "STRONG_BUY", "SELL", "STRONG_SELL"): continue if row.price is None: continue bucket = int(row.logged_at.timestamp()) // 300 key = (row.timeframe, row.signal, bucket) if key in seen: continue seen.add(key) sign = 1 if row.signal in ("BUY", "STRONG_BUY") else -1 rec = {"id": row.id, "logged_at": row.logged_at, "timeframe": row.timeframe, "signal": row.signal, "quality": row.quality, "regime": row.market_regime} for h in horizons_h: fwd = _forward_return(m5, row.logged_at, row.price, h) rec[f"fwd_{h}h"] = None if fwd is None else sign * fwd detail.append(rec) agg = [] for h in horizons_h: k = f"fwd_{h}h" vals = [d[k] for d in detail if d[k] is not None] wins = [v for v in vals if v > 0] agg.append({"horizon_h": h, "n": len(vals), "hit_rate": (len(wins) / len(vals)) if vals else None, "mean_move": (sum(vals) / len(vals)) if vals else None}) return {"aggregate": agg, "rows": detail} # ── Data fetching (backtester fetch_history: provider + parquet) ───── def fetch_raw_frames( tfs: Sequence[Timeframe], start: datetime, end: datetime, no_fetch: bool = False, ) -> Dict[Timeframe, pd.DataFrame]: """RAW (uncleaned) bars per TF — trade resolution must see spikes.""" from backtest.gpu_backtest import _cache_path, fetch_history from pair_signals.xauusd import make_config from providers import make_data_provider # Factory config (never bare PlatformConfig) — fetch_range and # validate_and_clean read lookback/refresh knobs off it. provider = make_data_provider(make_config(Timeframe.M15)) out = {} for tf in tfs: if no_fetch and not _cache_path( "XAU/USD", tf, start, end, provider.name).exists(): raise SystemExit( f"--no-fetch: no cached {tf.value} bars for " f"{start.date()}→{end.date()}; run without --no-fetch once") df = fetch_history(provider, "XAU/USD", tf, start, end, use_cache=True, clean=False) out[tf] = df return out # ── Surface parity ──────────────────────────────────────────────────── def parity_report(rows: Sequence[LogRow]) -> dict: """Compare every surface's row on the same (timeframe, signal bar). Groups rows by (tf, bar label). For groups with 2+ sources, the first row per source is compared on signal / quality / setup presence / action / config fingerprint. Returns coverage, rows-per-bar per source (over-scanning symptom), and the divergence list.""" groups: Dict[Tuple[str, str], List[LogRow]] = {} for r in rows: if not r.bar_ts or r.scan_kind == "blocked": continue groups.setdefault((r.timeframe, str(r.bar_ts)), []).append(r) per_bar: Dict[Tuple[str, str], List[int]] = {} divergences = [] both = 0 for (tf, bar), members in groups.items(): by_src: Dict[str, LogRow] = {} counts: Dict[str, int] = {} for m in sorted(members, key=lambda x: x.id): counts[m.source] = counts.get(m.source, 0) + 1 by_src.setdefault(m.source, m) for src, n in counts.items(): per_bar.setdefault((src, tf), []).append(n) if len(by_src) < 2: continue both += 1 srcs = sorted(by_src) a = by_src[srcs[0]] for src in srcs[1:]: b = by_src[src] diffs = [] if a.signal != b.signal: diffs.append(f"signal {a.signal}/{b.signal}") if a.quality != b.quality: diffs.append(f"quality {a.quality}/{b.quality}") if (a.setup is None) != (b.setup is None): diffs.append("setup presence") if a.action != b.action: diffs.append("action") if (a.config_fingerprint and b.config_fingerprint and a.config_fingerprint != b.config_fingerprint): diffs.append("config fingerprint") if diffs: divergences.append({ "timeframe": tf, "bar": bar, "sources": (srcs[0], src), "ids": (a.id, b.id), "diffs": diffs, "a_action": a.action, "b_action": b.action, }) rows_per_bar = { f"{src}/{tf}": sum(v) / len(v) for (src, tf), v in per_bar.items() if v } return {"bars_with_2_sources": both, "n_groups": len(groups), "rows_per_bar": rows_per_bar, "divergences": divergences} def print_parity(rep: dict) -> None: print("\n── Surface parity (same tf + signal bar) " + "─" * 30) print(f" bars seen by 2+ sources: {rep['bars_with_2_sources']} " f"of {rep['n_groups']}") print(" rows per bar (>1.05 = the same bar was scanned more than once):") for k, v in sorted(rep["rows_per_bar"].items()): flag = " <-- over-scanning" if v > 1.05 else "" print(f" {k:20} {v:.2f}{flag}") divs = rep["divergences"] print(f" divergences: {len(divs)}") by_day: Dict[str, int] = {} for d in divs: by_day[d["bar"][:10]] = by_day.get(d["bar"][:10], 0) + 1 for day, n in sorted(by_day.items()): print(f" {day}: {n}") for d in divs[:25]: print(f" {d['timeframe']:5} {d['bar']} ids {d['ids']} " f"{'; '.join(d['diffs'])} | {d['a_action'][:32]!r} vs " f"{d['b_action'][:32]!r}") # ── Stored-candle replay (no network) ──────────────────────────────── _TF_SECS = {"1min": 60, "5min": 300, "15min": 900, "30min": 1800, "1h": 3600, "4h": 14400, "1d": 86400, "1day": 86400} def stored_streams(db_path: str) -> Dict[str, List[dict]]: """Union of every closed bar in candles_json per timeframe.""" con = open_readonly(db_path) bars: Dict[str, Dict[str, dict]] = {} try: for (blob,) in con.execute( "SELECT candles_json FROM signals WHERE candles_json" " IS NOT NULL ORDER BY id"): try: snap = json.loads(blob) except (TypeError, ValueError): continue for tf, arr in (snap or {}).items(): store = bars.setdefault(tf, {}) for b in arr or []: ts = b.get("timestamp") if ts: store.setdefault(str(ts), b) finally: con.close() return {tf: [v[k] for k in sorted(v)] for tf, v in bars.items()} def replay_stored(events: List[SignalEvent], streams: Dict[str, List[dict]], max_bars: int = MAX_BARS_DEFAULT) -> None: """Resolve events with advisory_outcome's model on the stored bars. Same next-open reanchoring, intrabar peak, close-only ratchet, SL- first, 300-bar timeout and dual cost as the DB's own outcome column — extended to EVERY event, not only panel rows.""" from advisory_outcome import advance_outcome, initial_outcome for ev in events: rep_row = ev.representative bars = streams.get(rep_row.timeframe) or [] if not bars or not rep_row.bar_ts: ev.exit_reason = "NO_DATA" continue indicators = {rep_row.timeframe: {"timestamp": rep_row.bar_ts}} status, state = initial_outcome(rep_row.setup, indicators, rep_row.timeframe) if status != "PENDING": ev.exit_reason = "NO_DATA" continue state["policy"]["max_hold_bars"] = max_bars state = advance_outcome(state, bars) prog = state.get("progress") or {} ev.entry_fill = prog.get("entry_price") ev.bars_held = int(prog.get("bars_held") or 0) ev.mfe_r = prog.get("mfe_r") ev.mae_r = prog.get("mae_r") res = state.get("result") if state.get("status") == "RESOLVED" and res: ev.exit_reason = res["exit_reason"] ev.exit_time = pd.Timestamp(res["exit_at"]) ev.r_headline = res["net_r_065"] ev.r_compare = res["net_r_030"] elif state.get("status") == "PENDING": ev.exit_reason = "OPEN" if ev.entry_fill else "PENDING" else: ev.exit_reason = "UNRESOLVABLE" def opportunity_view(events: Sequence[SignalEvent], gap_min: float = 60) -> List[SignalEvent]: """First event per (tf, direction) cluster — consecutive same-direction events on a timeframe within ``gap_min`` are one opportunity.""" out: List[SignalEvent] = [] last: Dict[Tuple[str, str], datetime] = {} for ev in sorted(events, key=lambda e: e.representative.logged_at): r = ev.representative key = (r.timeframe, r.setup["direction"]) prev = last.get(key) if prev is None or r.logged_at - prev > timedelta(minutes=gap_min): out.append(ev) last[key] = r.logged_at return out def one_position_view(events: Sequence[SignalEvent], cooldown_bars: int = 5) -> List[SignalEvent]: """Daemon-style book: one open position per timeframe, next signal only after exit + cooldown (wall-clock bars).""" out: List[SignalEvent] = [] busy: Dict[str, datetime] = {} for ev in sorted(events, key=lambda e: e.representative.logged_at): if ev.r_headline is None or ev.exit_time is None: continue r = ev.representative until = busy.get(r.timeframe) if until is not None and r.logged_at <= until: continue out.append(ev) secs = _TF_SECS.get(r.timeframe, 900) * cooldown_bars busy[r.timeframe] = (ev.exit_time.to_pydatetime() + timedelta(seconds=secs)) return out def _book_line(label: str, evs: Sequence[SignalEvent]) -> str: closed = [e for e in evs if e.r_headline is not None] if not closed: return f" {label:28} n=0" tot = sum(e.r_headline for e in closed) wins = [e for e in closed if e.r_headline > 0] gp = sum(e.r_headline for e in wins) gl = -sum(e.r_headline for e in closed if e.r_headline < 0) pf = gp / gl if gl > 0 else math.inf return (f" {label:28} n={len(closed):3d} wins {len(wins):3d} " f"({100 * len(wins) / len(closed):3.0f}%) " f"total {tot:+7.2f}R @0.65 PF {pf:.2f}") def print_stored_books(events: Sequence[SignalEvent]) -> None: print("\n── Stored-candle replay books ($0.65 rt) " + "─" * 30) print(_book_line("all unique events", events)) print(_book_line("opportunities (60-min)", opportunity_view(events))) print(_book_line("one-position-per-TF", one_position_view(events))) by_dir: Dict[str, List[SignalEvent]] = {} for e in events: by_dir.setdefault(e.representative.setup["direction"], []).append(e) for d, evs in sorted(by_dir.items()): print(_book_line(f" direction {d}", evs)) by_tf: Dict[str, List[SignalEvent]] = {} for e in events: by_tf.setdefault(e.representative.timeframe, []).append(e) for tf, evs in sorted(by_tf.items()): print(_book_line(f" timeframe {tf}", evs)) # ── Reporting ───────────────────────────────────────────────────────── def _fmt_r(r: Optional[float]) -> str: return " n/a " if r is None else f"{r:+.2f}R" def persona_scorecard(events: Sequence[SignalEvent], db_path: str) -> List[dict]: """Per persona × stance: n, mean R and win rate of the underlying event (needs panel_takes_json — read straight from the DB).""" by_row = _row_index(events) con = open_readonly(db_path) agg: Dict[Tuple[str, str], List[float]] = {} seen_events: set = set() try: for rid, takes in con.execute( "SELECT id, panel_takes_json FROM signals" " WHERE panel_takes_json IS NOT NULL"): ev = by_row.get(rid) if ev is None or ev.r_headline is None: continue # One vote per EVENT, not per row. A scan re-logged inside the # cluster window carries its own panel takes, and counting # both rows appended the SAME event R twice — inflating n and # the persona means (news_outcome_buckets already iterates # unique events; this is the same rule). ekey = ev.rows[0].id if ev.rows else rid if ekey in seen_events: continue seen_events.add(ekey) try: for t in json.loads(takes): agg.setdefault((t.get("persona", "?"), t.get("stance", "?")), []).append( ev.r_headline) except (TypeError, ValueError): continue finally: con.close() return [{"persona": p, "stance": s, "n": len(v), "mean_r": sum(v) / len(v), "win_rate": sum(1 for x in v if x > 0) / len(v)} for (p, s), v in sorted(agg.items())] # ── News sentiment lead/lag ─────────────────────────────────────────── def _hourly_sentiment(rows: Sequence[LogRow]) -> pd.Series: """Mean logged news_sentiment per UTC hour (priced rows only).""" buckets: Dict[pd.Timestamp, List[float]] = {} for r in rows: if not r.price or r.news_sentiment is None: continue ts = pd.Timestamp(r.logged_at) ts = ts.tz_localize("UTC") if ts.tzinfo is None else ts.tz_convert("UTC") buckets.setdefault(ts.floor("h"), []).append(float(r.news_sentiment)) if not buckets: return pd.Series(dtype=float) return pd.Series({k: sum(v) / len(v) for k, v in buckets.items()}).sort_index() def _stream_frame(streams: Dict[str, List[dict]], tf: str) -> pd.DataFrame: """Stored bars for one timeframe as an OHLC frame indexed by UTC ts.""" bars = streams.get(tf) or [] if not bars: return pd.DataFrame() df = pd.DataFrame(bars) if "timestamp" not in df.columns or "close" not in df.columns: return pd.DataFrame() df.index = pd.to_datetime(df["timestamp"], utc=True) keep = [c for c in ("open", "high", "low", "close") if c in df.columns] return df[keep].astype(float).sort_index() def _corr(a: pd.Series, b: pd.Series) -> Optional[float]: """Pearson correlation of the overlapping, finite pairs (None if NaN).""" try: c = float(a.corr(b)) except (TypeError, ValueError): return None return None if math.isnan(c) else c def news_lead_lag(rows: Sequence[LogRow], streams: Dict[str, List[dict]], tf: str = "1h") -> dict: """Does the logged sentiment LEAD price, or only follow it? For each lag k (hours) the hourly mean sentiment is correlated with the PAST move (open_t - open_{t-k}) and the FORWARD move (close_{t+k-1} - open_t) — the scan in hour H sees the bar that closed at H, so the forward window starts at bar H's open. A high past-correlation with a ~zero forward-correlation means the score is a trailing return in disguise. Split-half on the forward number guards against reading one regime as an edge. """ sent = _hourly_sentiment(rows) df = _stream_frame(streams, tf) if sent.empty or df.empty: return {"n_hours": 0} joined = df.join(pd.DataFrame({"sent": sent}), how="inner").dropna( subset=["sent"]) n = len(joined) if n < 8: return {"n_hours": n} s = joined["sent"] half = n // 2 first, second = joined.index[:half], joined.index[half:] corr_past, corr_fwd, corr_split = {}, {}, {} for k in NEWS_LAGS_H: past = (df["open"] - df["open"].shift(k)).reindex(joined.index) fwd = (df["close"].shift(-(k - 1)) - df["open"]).reindex(joined.index) corr_past[k] = _corr(s, past) corr_fwd[k] = _corr(s, fwd) corr_split[k] = (_corr(s.loc[first], fwd.loc[first]), _corr(s.loc[second], fwd.loc[second])) day_sent = sent.groupby(sent.index.floor("D")).mean() day = df.resample("1D").agg({"open": "first", "close": "last"}).dropna() move = day["close"] - day["open"] # .corr aligns on the index; shifting over the BAR index keeps # "previous/next" meaning the previous/next day that actually traded. daily = {"same": _corr(day_sent, move), "prev": _corr(day_sent, move.shift(1)), "next": _corr(day_sent, move.shift(-1))} return {"n_hours": n, "corr_past": corr_past, "corr_fwd": corr_fwd, "corr_fwd_split": corr_split, "daily": daily} def news_extremes(rows: Sequence[LogRow], streams: Dict[str, List[dict]], tf: str = "1h") -> dict: """What the sentiment said at the window's high and low. Plus how long AFTER the top the first bearish hour arrived (and vice versa at the bottom) — a turn called late is a trailing indicator. """ df = _stream_frame(streams, tf) if df.empty or "high" not in df.columns or "low" not in df.columns: return {} sent = _hourly_sentiment(rows) hi_ts = df["high"].idxmax() lo_ts = df["low"].idxmin() hi_px = float(df["high"].loc[hi_ts]) lo_px = float(df["low"].loc[lo_ts]) def _at(ts): v = sent.get(ts) return None if v is None or (isinstance(v, float) and math.isnan(v)) \ else float(v) def _first(from_ts, ref_px, sign): for ts, val in sent.items(): if ts < from_ts: continue if (val <= NEWS_BEAR) if sign < 0 else (val >= NEWS_BULL): close = df["close"].asof(ts) if close is None or (isinstance(close, float) and math.isnan(close)): continue return {"lag_h": (ts - from_ts).total_seconds() / 3600.0, "price_move": float(close) - ref_px} return None return { "high": {"ts": hi_ts.isoformat(), "price": hi_px, "sentiment": _at(hi_ts)}, "low": {"ts": lo_ts.isoformat(), "price": lo_px, "sentiment": _at(lo_ts)}, "bear_after_high": _first(hi_ts, hi_px, -1), "bull_after_low": _first(lo_ts, lo_px, +1), } def news_outcome_buckets(events: Sequence[SignalEvent], db_path: str) -> List[dict]: """Resolved events bucketed by (direction, news bucket, macro stance). The macro persona is the only voice the 2026-09 scorecard found informative, so it rides along as a third axis here. """ by_row = _row_index(events) macro: Dict[int, str] = {} con = open_readonly(db_path) try: for rid, takes in con.execute( "SELECT id, panel_takes_json FROM signals" " WHERE panel_takes_json IS NOT NULL ORDER BY id"): ev = by_row.get(rid) if ev is None or ev.representative.id in macro: continue try: parsed = json.loads(takes) except (TypeError, ValueError): continue stance = "none" for t in parsed or []: if isinstance(t, dict) and t.get("persona") == "macro": stance = t.get("stance") or "none" break macro[ev.representative.id] = stance finally: con.close() agg: Dict[Tuple[str, str, str], List[float]] = {} for ev in events: if ev.r_headline is None: continue rep = ev.representative s = rep.news_sentiment if s is None: bucket = "neutral" elif s >= NEWS_BULL: bucket = "bull" elif s <= NEWS_BEAR: bucket = "bear" else: bucket = "neutral" direction = (rep.setup or {}).get("direction") or "?" agg.setdefault((direction, bucket, macro.get(rep.id, "none")), []).append(ev.r_headline) return [{"direction": d, "bucket": b, "macro": m, "n": len(v), "sum_r": sum(v), "mean_r": sum(v) / len(v), "win_rate": sum(1 for x in v if x > 0) / len(v)} for (d, b, m), v in sorted(agg.items())] def _fmt_c(c: Optional[float], nd: int = 3) -> str: return " n/a " if c is None else f"{c:+.{nd}f}" def print_news(lead_lag: dict, extremes: dict, buckets: Sequence[dict], tf: str = "1h") -> None: print("\n── News sentiment (warning-only; hourly mean vs stored " + tf + " bars) " + "─" * 10) n = lead_lag.get("n_hours", 0) if "corr_past" not in lead_lag: print(f" only {n} joined hour(s) of logged sentiment — " "need 8+ for a correlation read") else: print(f" {n} hours with a logged sentiment joined to a bar") for k in NEWS_LAGS_H: a, b = lead_lag["corr_fwd_split"].get(k, (None, None)) print(f" past {k:2d}h {_fmt_c(lead_lag['corr_past'].get(k))}" f" fwd {k:2d}h {_fmt_c(lead_lag['corr_fwd'].get(k))}" f" (halves {_fmt_c(a, 2)} / {_fmt_c(b, 2)})") d = lead_lag.get("daily") or {} print(f" daily open->close: same day {_fmt_c(d.get('same'))}" f" previous {_fmt_c(d.get('prev'))}" f" next {_fmt_c(d.get('next'))}") for side, key, tag in (("high", "bear_after_high", "bearish"), ("low", "bull_after_low", "bullish")): info = (extremes or {}).get(side) if not info: continue sv = info.get("sentiment") head = (f" {side:4} {info['ts'][:16]}Z ${info['price']:.1f} " f"sentiment {_fmt_c(sv, 2)}") nxt = (extremes or {}).get(key) if nxt: print(f"{head} → first {tag} hour +{nxt['lag_h']:.0f} h later, " f"price {nxt['price_move']:+.1f}") else: print(f"{head} → no {tag} hour logged after it") if buckets: print(f" outcome buckets ({sum(b['n'] for b in buckets)} resolved " "events):") print(f" {'dir':5} {'news':7} {'macro':8} {'n':>3} {'total':>8} " f"{'mean':>8} {'win':>5}") for b in buckets: print(f" {b['direction']:5} {b['bucket']:7} {b['macro']:8} " f"{b['n']:3d} {_fmt_r(b['sum_r']):>8} " f"{_fmt_r(b['mean_r']):>8} " f"{100 * b['win_rate']:4.0f}%") else: print(" no resolved events to bucket yet") print(" Caveat: sentiment ≈ trailing 24h return; buckets inherit the " "trend confound.") def print_report( rows: List[LogRow], events: List[SignalEvent], panel: List[dict], hit: dict, ) -> None: print("=" * 72) print("SIGNAL LOG PERFORMANCE REPORT") print("=" * 72) if rows: print(f"Window: {rows[0].logged_at:%Y-%m-%d %H:%M} → " f"{rows[-1].logged_at:%Y-%m-%d %H:%M} UTC " f"({(rows[-1].logged_at - rows[0].logged_at).days + 1} days)") by_src: Dict[str, int] = {} by_tf: Dict[str, int] = {} for r in rows: by_src[r.source] = by_src.get(r.source, 0) + 1 by_tf[r.timeframe] = by_tf.get(r.timeframe, 0) + 1 print(f"Rows: {len(rows)} by source {by_src} by tf {by_tf}") n_setup = sum(1 for r in rows if r.setup) print(f"Rows with trade setup: {n_setup} → " f"{len(events)} unique signal events after dedup") print("\n── Trade replay (next-open fill, raw path, trail per setup, " f"cost ${COST_HEADLINE:.2f} rt) " + "─" * 8) hdr = (f"{'first logged':16} {'tf':5} {'dir':11} {'q':2} {'strategy':8}" f" {'fill':>9} {'exit':7} {'bars':>4} {'R@0.65':>8} {'R@0.30':>8}" f" {'MFE':>6} {'MAE':>6} panel") print(hdr) print("-" * len(hdr)) for ev in events: rep = ev.representative print(f"{rep.logged_at:%m-%d %H:%M} " f"{rep.timeframe:5} {rep.signal:11} {rep.quality:2} " f"{(rep.setup.get('strategy') or ''):8} " f"{ev.entry_fill if ev.entry_fill else float('nan'):9.2f} " f"{ev.exit_reason:7} {ev.bars_held:4} " f"{_fmt_r(ev.r_headline):>8} {_fmt_r(ev.r_compare):>8} " f"{(f'{ev.mfe_r:.2f}' if ev.mfe_r is not None else 'n/a'):>6} " f"{(f'{ev.mae_r:.2f}' if ev.mae_r is not None else 'n/a'):>6} " f"{ev.panel_verdict or ''}") closed = [e for e in events if e.r_headline is not None] if closed: tot = sum(e.r_headline for e in closed) tot30 = sum(e.r_compare for e in closed) wins = [e for e in closed if e.r_headline > 0] gp = sum(e.r_headline for e in wins) gl = -sum(e.r_headline for e in closed if e.r_headline < 0) pf = gp / gl if gl > 0 else math.inf print(f"\nClosed: {len(closed)} wins: {len(wins)} " f"({100 * len(wins) / len(closed):.0f}%) " f"total {tot:+.2f}R @0.65 / {tot30:+.2f}R @0.30 PF {pf:.2f}") by_q: Dict[str, List[SignalEvent]] = {} for e in closed: by_q.setdefault(e.representative.quality, []).append(e) for q, grp in sorted(by_q.items()): print(f" grade {q}: n={len(grp)} " f"total {sum(e.r_headline for e in grp):+.2f}R") n_open = sum(1 for e in events if e.exit_reason == "OPEN") n_pend = sum(1 for e in events if e.exit_reason in ("PENDING", "NO_DATA")) if n_open or n_pend: print(f" ({n_open} still open / mark-to-market, " f"{n_pend} pending — insufficient bars yet)") print("\n── Advisory panel scorecard " + "─" * 42) if not panel: print(" no panel verdicts logged yet") for p in panel: r = p["r_headline"] if r is None: impact = p["note"] or "no outcome yet" elif p["verdict"] == "OPPOSE": impact = f"saved {-r:+.2f}R" if r < 0 else f"cost {r:+.2f}R" elif p["verdict"] == "AGREE": impact = f"confirmed {r:+.2f}R" else: impact = f"passed, trade {r:+.2f}R" print(f" #{p['id']:<5} {p['logged_at']:%m-%d %H:%M} " f"{p['timeframe']:5} {p['signal']:11} {p['quality']:2} " f"{p['verdict']:7} @{p['confidence'] or 0:3.0f}% " f"[{p['basis'] or 'pending'}] {impact}") print("\n── Directional hit-rate (non-setup signals, M5 closes) " + "─" * 12) for a in hit["aggregate"]: if a["n"] == 0: print(f" +{a['horizon_h']}h: no data") continue print(f" +{a['horizon_h']}h: n={a['n']} " f"hit-rate {100 * a['hit_rate']:.0f}% " f"mean move in dir {a['mean_move']:+.2f}$") print("\nCaveat: log started 2026-08-13 — small sample, directional " "read only.") def export_csv(events: Sequence[SignalEvent], path: str) -> None: recs = [] for ev in events: rep = ev.representative s = rep.setup recs.append({ "row_ids": ",".join(str(r.id) for r in ev.rows), "first_logged": rep.logged_at.isoformat(), "timeframe": rep.timeframe, "signal": rep.signal, "quality": rep.quality, "score": rep.score, "strategy": s.get("strategy"), "logged_entry": s.get("entry"), "fill": ev.entry_fill, "sl": s.get("stop_loss"), "tp1": s.get("take_profit_1"), "trail_arm_r": s.get("trail_arm_r"), "trail_dist_r": s.get("trail_dist_r"), "exit_reason": ev.exit_reason, "exit_time": ev.exit_time.isoformat() if ev.exit_time is not None else "", "bars_held": ev.bars_held, "r_at_0.65": ev.r_headline, "r_at_0.30": ev.r_compare, "mfe_r": ev.mfe_r, "mae_r": ev.mae_r, "panel_verdict": ev.panel_verdict, "panel_confidence": ev.panel_confidence, }) pd.DataFrame(recs).to_csv(path, index=False) print(f"\nWrote {len(recs)} events to {path}") # ── Scan outcomes (model-free forward returns; no bar data needed) ──── def scan_outcome_rows(db_path: str) -> List[dict]: """One labelled scan per (timeframe, bar_ts) — the LOWEST signal id. Several surfaces log the same bar, so the raw table is duplicated; SQLite takes the bare columns from the row min() selected.""" con = open_readonly(db_path) try: if not con.execute( "SELECT 1 FROM sqlite_master WHERE type='table'" " AND name='scan_outcomes'").fetchone(): return [] return [ {"id": r[0], "timeframe": r[1], "signal": r[3] or "", "quality": r[4] or "-", "watch_level": r[5], "fwd_4h_atr": r[6], "burst_1h": r[7], "complete": r[8]} for r in con.execute( "SELECT MIN(o.signal_id), o.timeframe, o.bar_ts, s.signal," " s.quality, s.watch_level, o.fwd_4h_atr, o.burst_1h," " o.complete FROM scan_outcomes o" " JOIN signals s ON s.id = o.signal_id" " GROUP BY o.timeframe, o.bar_ts") ] finally: con.close() def _scan_stats(rows: Sequence[dict], directional: bool) -> Optional[dict]: """n / mean / median forward move and the hit rate for one bucket. 'Hit' = the 4 h move agreed with the signal; for the NEUTRAL baseline it degrades to the share of UP moves.""" vals = sorted(r["fwd_4h_atr"] for r in rows if r["fwd_4h_atr"] is not None) if not vals: return None hits = 0 for row in rows: move = row["fwd_4h_atr"] if move is None: continue sign = -1 if (directional and "SELL" in row["signal"]) else 1 hits += int(sign * move > 0) mid = len(vals) // 2 return {"n": len(vals), "mean": sum(vals) / len(vals), "median": vals[mid] if len(vals) % 2 else (vals[mid - 1] + vals[mid]) / 2, "hit_rate": hits / len(vals)} def print_scan_outcomes(rows: Sequence[dict]) -> None: """Forward-move report: no replay, no model — just what happened.""" if not rows: print("No scan_outcomes rows yet " "(python scripts/label_scan_outcomes.py)") return complete = [r for r in rows if r["complete"]] print(f"\n{'═' * 66}\n SCAN OUTCOMES — 4 h forward move in ATRs " f"({len(complete)} complete of {len(rows)} labelled bars)") print(f"{'═' * 66}") directional = [r for r in complete if r["signal"] not in ("", "NEUTRAL")] neutral = [r for r in complete if r["signal"] == "NEUTRAL"] for title, groups, keyfn, is_dir in ( ("by timeframe x signal", directional, lambda r: (r["timeframe"], r["signal"]), True), ("by timeframe x grade", directional, lambda r: (r["timeframe"], r["quality"]), True), ("NEUTRAL baseline", neutral, lambda r: (r["timeframe"], "NEUTRAL"), False)): print(f"\n── {title} " + "─" * (52 - len(title))) buckets: Dict[Tuple[str, str], List[dict]] = {} for row in groups: buckets.setdefault(keyfn(row), []).append(row) for key in sorted(buckets): stat = _scan_stats(buckets[key], is_dir) if stat is None: continue print(f" {key[0]:6} {key[1]:12} n={stat['n']:5d} " f"mean {stat['mean']:+.3f} med {stat['median']:+.3f} " f"{'hit' if is_dir else 'up '} " f"{100 * stat['hit_rate']:.0f}%") print("\n── burst (>=3x ATR bar) within 1 h, by watch level " + "─" * 15) levels: Dict[str, List[int]] = {} for row in complete: if row["burst_1h"] is None: continue levels.setdefault(row["watch_level"] or "NULL", []).append( int(row["burst_1h"])) for level in sorted(levels): hits = levels[level] print(f" {level:6} n={len(hits):5d} burst " f"{100 * sum(hits) / len(hits):.1f}%") # ── main ────────────────────────────────────────────────────────────── def main(argv: Optional[Sequence[str]] = None) -> int: ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--db", default=DEFAULT_DB) ap.add_argument("--csv", default=None, help="per-event CSV export") ap.add_argument("--max-bars", type=int, default=MAX_BARS_DEFAULT) ap.add_argument("--cluster-gap-min", type=float, default=CLUSTER_GAP_DEFAULT_MIN) ap.add_argument("--no-fetch", action="store_true", help="parquet cache only, never hit the network") ap.add_argument("--parity", action="store_true", help="surface parity report only (no bar data needed)") ap.add_argument("--stored-candles", action="store_true", help="replay on the candle streams stored in the DB " "(no network) instead of provider history") ap.add_argument("--news", action="store_true", help="news-sentiment lead/lag + outcome buckets " "(requires --stored-candles)") ap.add_argument("--scan-outcomes", action="store_true", help="labelled forward moves per scan (no bar data " "needed); see scripts/label_scan_outcomes.py") args = ap.parse_args(argv) if args.scan_outcomes: print_scan_outcomes(scan_outcome_rows(args.db)) return 0 if args.news and not args.stored_candles: print("--news requires --stored-candles") return 2 rows = load_rows(args.db) if not rows: print(f"No rows in {args.db}") return 0 if args.parity: print_parity(parity_report(rows)) return 0 events = cluster_setup_rows( rows, gap=timedelta(minutes=args.cluster_gap_min)) if args.stored_candles: streams = stored_streams(args.db) replay_stored(events, streams, max_bars=args.max_bars) panel = panel_scorecard(rows, events, None) m5 = streams.get(Timeframe.M5.value) or [] m5_df = pd.DataFrame(m5) if not m5_df.empty: m5_df.index = pd.to_datetime(m5_df["timestamp"], utc=True) m5_df = m5_df.rename(columns={"close": "Close"}) hit = directional_hit_rate(rows, m5_df) if not m5_df.empty else \ {"aggregate": [], "rows": []} print_report(rows, events, panel, hit) print_stored_books(events) print("\n── Persona scorecard (stance × event outcome) " + "─" * 22) for p in persona_scorecard(events, args.db): print(f" {p['persona']:11} {p['stance']:8} n={p['n']:3d} " f"mean {p['mean_r']:+.3f}R win {100 * p['win_rate']:.0f}%") if args.news: print_news(news_lead_lag(rows, streams), news_extremes(rows, streams), news_outcome_buckets(events, args.db)) print_parity(parity_report(rows)) if args.csv: export_csv(events, args.csv) return 0 tfs = {Timeframe(r.timeframe) for r in rows if r.setup} tfs.add(Timeframe.M5) # hit-rate + panel fallback frame start = (rows[0].logged_at - timedelta(days=1)).replace( hour=0, minute=0, second=0, microsecond=0) end = datetime.now(timezone.utc) raw = fetch_raw_frames(sorted(tfs, key=lambda t: t.value), start, end, no_fetch=args.no_fetch) replay_events(events, raw, max_bars=args.max_bars) panel = panel_scorecard(rows, events, raw.get(Timeframe.M5)) hit = directional_hit_rate(rows, raw[Timeframe.M5]) print_report(rows, events, panel, hit) print_parity(parity_report(rows)) if args.csv: export_csv(events, args.csv) return 0 if __name__ == "__main__": sys.exit(main())