Harwinder Sekhon
ae6a6658f6
- seed an advisory outcome on every setup row; same-setup re-logs within 60 min are DUPLICATE rows (advisory_event_id); resolver rebuilds bar streams across gaps - headlines/events interned into news_snapshots (DB 667 MB -> 262 MB); readers use signals_with_news / read_news and mode=ro (immutable=1 missed WAL rows) - post-gate truth columns: gate_reason, actionable, cleaned_warnings, scan_kind, config_fingerprint; SELL-hour gates fail closed - telegram: breach rescan reuses this bar's report; grade dropped from alert fingerprint (upgrade-only re-alert) - market_outlook: hour-bucket dedup (old rule skipped alternate hours), skip heartbeats - agent_panel: macro-anchored OPPOSE guard (opposed trades were +30R over 3 weeks) - analyze_signal_log --stored-candles/--parity; backfill + news migration scripts - tests: Module 25 (analyze_signal_log) + 27 (observability); 429 total Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FC4s6VCURFte4dub4xKxYL
952 lines
39 KiB
Python
952 lines
39 KiB
Python
#!/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.
|
||
|
||
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
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import math
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
from dataclasses import dataclass, field
|
||
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 = ""
|
||
|
||
|
||
@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")
|
||
|
||
|
||
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 "",
|
||
))
|
||
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
|
||
|
||
|
||
# ── 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: Dict[int, SignalEvent] = {}
|
||
for ev in events:
|
||
for r in ev.rows:
|
||
by_row[r.id] = ev
|
||
|
||
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"]
|
||
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(rows: Sequence[LogRow], 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: Dict[int, SignalEvent] = {}
|
||
for ev in events:
|
||
for r in ev.rows:
|
||
by_row[r.id] = ev
|
||
con = open_readonly(db_path)
|
||
agg: Dict[Tuple[str, str], List[float]] = {}
|
||
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
|
||
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())]
|
||
|
||
|
||
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}")
|
||
|
||
|
||
# ── 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")
|
||
args = ap.parse_args(argv)
|
||
|
||
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(rows, 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}%")
|
||
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())
|