Harwinder Sekhon
59f69400a8
None of these touch the signal path. The 12-month M15 book (2024-03-31..2025-03-30, $0.65, cached Oanda bars) is byte-identical before and after (md5 c59d3e7e685c3c880b614640adfea74b, 375 trades), and signal/indicator logging to logs/signals.db is unchanged — verified end to end: per-TF indicators for all 3 TFs, 14 candles each, MTF and reasons all still persisted. agent_panel: the >=2-take quorum guards the CHAIR's synthesizer spend, but ran BEFORE the macro branch — so the shipped default (verdict_source="macro", which never calls the synthesizer) silently dropped a good macro verdict whenever the other three personas were slow. Quorum moved below the macro path. analyze_signal_log: _forward_return clamped to the last available bar when the horizon had not elapsed, reporting a near-zero "forward move" for every row logged within `hours` of the end of the data — on every run, blended into the hit-rate/mean-move aggregate. Now returns None. analyze_signal_log: persona_scorecard iterated raw rows through an index that maps EVERY member row of an event, so a re-logged scan appended the same event R twice. One vote per event, matching news_outcome_buckets. Measured inflation on the live DB: 1.03x. signal_log: dedup required a non-negative logged_at gap, but logged_at is stamped before the row wins the write lock (busy_timeout 5 s), so two processes logging one setup at a boundary can invert timestamp order against id order and seed a second event. Now abs(gap); same in backfill_seeds. Zero occurrences in the current 366 setup rows. advisory_outcome: the catch-all turned real evaluator faults into the same UNRESOLVABLE used for legitimate weekend gaps, with nothing in the log. Still fail-safe, now logged. llm_client: only the "cap" warn-once flag was ever reset, so in a daemon the second and every later network/http/parse outage logged at DEBUG only. Cleared on UTC-day roll and re-armed by a recovered call. dashboard.js: renderCards() redraws the selector for every timeframe but was gated on the SELECTED one changing, leaving the other buttons stale (up to 4 h on H4) while renderHero's rail already showed the new value. news_summary.php: atomic_write treated a short write (disk full) as success, publishing a truncated body the cache-first branch then served forever; and the daily cap was charged before the no-key check and before the stampede lock re-checked the cache, so a click burst could exhaust the 150/day budget with zero LLM calls. migrate_signal_log_news: logged_at is ISO with 'T' while SQLite's datetime() uses a space, so string-comparing them put the whole cutoff calendar day on the wrong side and it never pruned. NOTE: this now deletes one day more than before — it only runs when invoked by hand. Tests 414 -> 422 (new Module 28, one per fix). The existing "<2 valid takes" test was retargeted at the chair — the case it actually protects — rather than deleted, with new coverage for macro-alone. Not fixed, recorded for the owner: - Gate 6c's boost condition is textually identical to the C-grade definition, so every C in a "transitioning" regime is promoted to B before gate 6e reads it, making sell_min_quality's exclusion of "C" a no-op on M15/H4 (H1 and M5 are immune by other means). The backtester runs the same code, so every pinned book already includes this — changing it is a rule-1 revalidation event. - Safe-haven/geopolitical headlines score exactly 0.00 in the keyword fallback and are now dropped entirely by the ±0.1 neutral filter, contradicting the platform's own gold doctrine. Bites only when DeepSeek is capped/disabled/failing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ah5EB1goi5inQmSd37Ne6a
1240 lines
51 KiB
Python
1240 lines
51 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.
|
||
|
||
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, 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 = ""
|
||
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(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 = _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(rows: Sequence[LogRow],
|
||
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}")
|
||
|
||
|
||
# ── 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)")
|
||
args = ap.parse_args(argv)
|
||
|
||
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(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}%")
|
||
if args.news:
|
||
print_news(news_lead_lag(rows, streams),
|
||
news_extremes(rows, streams),
|
||
news_outcome_buckets(rows, 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())
|