Harwinder Sekhon
ad14ea4fed
Agent panel v2: personas return p_win anchored to the book base rates with a code-computed decision context; verdict = equal-weight mean (verdict_source "mean"); only the macro persona runs live (agent_panel.personas ["macro"]); prompt version and exact payloads logged; offline replay harness scripts/panel_replay.py; base-rate table agent_panel_base_rates.json (scripts/build_panel_base_rates.py). Backtester `strategy` column now carries the generator's own stamp (label-only; five reference books re-pinned, all 62 other columns identical). News corpus (after evaluating the owner's ML-NEWS scikit model against DeepSeek: golden set 39/78 vs 78/78, out-of-sample agreement 64% against a 90% self-agreement ceiling; verdict: fallback only, not a replacement): - NewsSentiment.xau_score + archive column xau_score keep DeepSeek's gold score for every headline (930 of the first 1,546 DeepSeek rows had none); an existing newsdata.db gains the column in place on first open - headlines naming no tracked currency are scored and archived (currencies_json []) but never returned to the platform - _apply_llm_sentiment sends every new headline in its cycle, 25 per call; a failed call ends the cycle's sending Dashboard publisher never writes a database: main() calls app_config.set_db_writes_for_process(False), honoured by signal_log_enabled() and news_archive_enabled() (owner decision: the bot is the single writer). Test suite: the hermetic app config now points both writers at a suite-private temp dir — the RSS fixture tests were writing fixture headlines into the real logs/newsdata.db — with a Module 32 test pinning it. 529 tests.
515 lines
25 KiB
Python
515 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline replay of the agent panel on logged setups, scored against the
|
|
realised outcome — so a prompt or payload version is judged on the SAME
|
|
trades as its predecessor before it is enabled live.
|
|
|
|
venv/bin/python scripts/panel_replay.py --db logs/signals.db --dry-run
|
|
venv/bin/python scripts/panel_replay.py --db logs/signals.db --max-calls 160 --out replay.csv
|
|
|
|
What it does
|
|
------------
|
|
1. Loads every resolved setup event (analyze_signal_log's dedup: one
|
|
event per identical setup, first occurrence) whose advisory outcome is
|
|
RESOLVED; the outcome is `advisory_r_multiple > 0` (net R at $0.65).
|
|
2. Rebuilds what each persona would see. Rows logged by agent_panel v2
|
|
carry the exact payloads (panel_payloads_json); older rows are
|
|
reconstructed from the stored indicators_json / setup_json / mtf_json /
|
|
penalty_tags_json / reasons_json / warnings_json, the interned
|
|
headlines and events (signals_with_news), and candles from the kept
|
|
Oanda cache windows so the price path is what the scan saw. Calendar
|
|
events are evaluated AS OF the signal bar (released / minutes_until),
|
|
not as of today.
|
|
3. Runs the CURRENT prompts (agent_panel.PANEL_PROMPT_VERSION) through the
|
|
real DeepSeek client on each payload, validates the replies with the
|
|
live code path, aggregates with agent_panel.mean_verdict.
|
|
4. Scores three predictors on the same events: the new run, the logged
|
|
v1 takes (stance x confidence -> pseudo-probability, when present) and
|
|
the base-rate-only predictor (the class win rate for every trade):
|
|
Brier score, log-loss, calibration by probability band, discrimination
|
|
(mean R and win rate by band), constant-voter rate per persona, and the
|
|
counterfactual books (all / skip p <= 40 / only p >= 60).
|
|
|
|
Spend: 4 calls per event against the shared DeepSeek daily cap (500);
|
|
`--max-calls` is a hard stop. `--dry-run` builds and writes the payloads
|
|
without a single call. Read-only on the database.
|
|
|
|
Caveat printed with every summary: a few dozen outcomes from one regime
|
|
can reject a pathological prompt (constant voting, narrative anchoring);
|
|
they cannot prove accuracy. The live scorecard at 100 and 300 outcomes
|
|
(analyze_signal_log.py --stored-candles) is the real test.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import glob
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
|
|
|
import pandas as pd
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if ROOT not in sys.path:
|
|
sys.path.insert(0, ROOT)
|
|
|
|
import agent_panel as ap # noqa: E402
|
|
from analyze_signal_log import (cluster_setup_rows, load_rows, # noqa: E402
|
|
open_readonly)
|
|
from config import SignalStrength, Timeframe # noqa: E402
|
|
from news_engine import (EconomicEvent, FundamentalContext, # noqa: E402
|
|
NewsSentiment)
|
|
from signal_aggregator import SignalReport, TradeSetup # noqa: E402
|
|
from signal_log import read_news # noqa: E402
|
|
|
|
PERSONAS = ap.ALL_PERSONAS # research default: score every persona
|
|
CACHE_DIR = os.path.join(ROOT, "backtest", "cache")
|
|
_TF = {tf.value: tf for tf in Timeframe}
|
|
|
|
|
|
# ── Reconstruction ────────────────────────────────────────────────────
|
|
|
|
class _AsOfEvent(EconomicEvent):
|
|
"""EconomicEvent whose `is_upcoming` / `minutes_until` are evaluated
|
|
relative to the signal bar, not the wall clock."""
|
|
def __init__(self, as_of: datetime, **kw):
|
|
super().__init__(**kw)
|
|
self._as_of = as_of
|
|
|
|
@property
|
|
def is_upcoming(self) -> bool: # type: ignore[override]
|
|
return self.time > self._as_of
|
|
|
|
@property
|
|
def minutes_until(self) -> float: # type: ignore[override]
|
|
return (self.time - self._as_of).total_seconds() / 60.0
|
|
|
|
|
|
def _dt(s: Optional[str]) -> Optional[datetime]:
|
|
if not s:
|
|
return None
|
|
try:
|
|
d = datetime.fromisoformat(str(s).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
return d if d.tzinfo else d.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def _loads(s: Any, default: Any) -> Any:
|
|
if s in (None, "", "null"):
|
|
return default
|
|
try:
|
|
return json.loads(s)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _analysis(tf: str, blob: Dict[str, Any]) -> SimpleNamespace:
|
|
inds = {}
|
|
for name, d in (blob.get("indicators") or {}).items():
|
|
try:
|
|
sig = SignalStrength[d.get("signal", "NEUTRAL")]
|
|
except KeyError:
|
|
sig = SignalStrength.NEUTRAL
|
|
inds[name] = SimpleNamespace(
|
|
signal=sig, confidence=float(d.get("confidence") or 0.0),
|
|
weight=float(d.get("weight") or 0.0), value=d.get("value"),
|
|
description=d.get("description") or "")
|
|
return SimpleNamespace(timeframe=tf, timestamp=blob.get("timestamp"),
|
|
key_levels=dict(blob.get("key_levels") or {}),
|
|
indicators=inds)
|
|
|
|
|
|
def _context_from_row(events_json: Optional[str], headlines_json: Optional[str],
|
|
as_of: datetime) -> FundamentalContext:
|
|
ctx = FundamentalContext()
|
|
ev = _loads(events_json, {}) or {}
|
|
for key in ("events_today", "events_upcoming_1h"):
|
|
out = []
|
|
for e in ev.get(key) or []:
|
|
t = _dt(e.get("time"))
|
|
if t is None:
|
|
continue
|
|
out.append(_AsOfEvent(as_of, time=t, currency=e.get("currency") or "",
|
|
impact=e.get("impact") or "", event=e.get("event") or "",
|
|
forecast=e.get("forecast"), previous=e.get("previous"),
|
|
actual=e.get("actual")))
|
|
setattr(ctx, key, out)
|
|
ctx.news_sentiment = dict(ev.get("sentiment_by_currency") or {})
|
|
ctx.calendar_available = bool(ev.get("calendar_available", False))
|
|
for h in _loads(headlines_json, []) or []:
|
|
t = _dt(h.get("timestamp")) or as_of
|
|
ctx.news_headlines.append(NewsSentiment(
|
|
headline=h.get("headline") or "", source=h.get("source") or "",
|
|
timestamp=t, currencies_mentioned=list(h.get("currencies") or []),
|
|
sentiment_score=float(h.get("score") or 0.0),
|
|
keywords=list(h.get("keywords") or []),
|
|
sentiment_by_currency={k: float(v) for k, v in
|
|
(h.get("score_by_currency") or {}).items()
|
|
if v is not None},
|
|
score_source=h.get("score_source") or "keyword"))
|
|
return ctx
|
|
|
|
|
|
_frame_cache: Dict[str, pd.DataFrame] = {}
|
|
_have_cols: set = set()
|
|
|
|
|
|
def _col(name: str) -> str:
|
|
"""Column reference that tolerates a database logged before the
|
|
column existed (a read-only copy is never migrated)."""
|
|
return name if name in _have_cols else f"NULL AS {name}"
|
|
|
|
|
|
def _cached_frame(tf: str) -> Optional[pd.DataFrame]:
|
|
"""Newest kept Oanda cache window for `tf` (raw bars, UTC index)."""
|
|
if tf in _frame_cache:
|
|
return _frame_cache[tf]
|
|
files = sorted(glob.glob(os.path.join(CACHE_DIR, f"oanda_XAU_USD_{tf}_*.parquet")),
|
|
key=lambda p: p.rsplit("_", 1)[-1])
|
|
df = None
|
|
if files:
|
|
try:
|
|
df = pd.read_parquet(files[-1])
|
|
if df.index.tz is None:
|
|
df.index = df.index.tz_localize("UTC")
|
|
except Exception:
|
|
df = None
|
|
_frame_cache[tf] = df
|
|
return df
|
|
|
|
|
|
def _frames_as_of(tfs: Sequence[str], bar_ts: datetime) -> Dict[Timeframe, pd.DataFrame]:
|
|
out: Dict[Timeframe, pd.DataFrame] = {}
|
|
for tf in tfs:
|
|
df = _cached_frame(tf)
|
|
tfe = _TF.get(tf)
|
|
if df is None or tfe is None:
|
|
continue
|
|
sl = df[df.index <= pd.Timestamp(bar_ts)]
|
|
if len(sl) >= 3:
|
|
out[tfe] = sl
|
|
return out
|
|
|
|
|
|
def build_report(con, rowid: int) -> Tuple[Optional[SignalReport], Optional[FundamentalContext],
|
|
Dict[Timeframe, pd.DataFrame], Optional[Dict[str, Any]]]:
|
|
"""(report, context, frames, stored_payloads) for one logged row."""
|
|
row = con.execute(
|
|
"SELECT symbol, timeframe, " + _col("bar_ts") + ", logged_at, signal, score,"
|
|
" confidence, quality, price, market_regime, volatility_regime, mtf_json,"
|
|
" setup_json, reasons_json, warnings_json, event_risk, event_description,"
|
|
" news_sentiment, indicators_json, " + _col("penalty_tags_json") + ", "
|
|
+ _col("exhaustion_count") + ", " + _col("penalty_total") + ", "
|
|
+ _col("momentum_fading") + ", " + _col("panel_payloads_json")
|
|
+ " FROM signals WHERE id=?", (rowid,)).fetchone()
|
|
if row is None:
|
|
return None, None, {}, None
|
|
(symbol, tf, bar_ts, logged_at, signal, score, conf, quality, price, regime, vol,
|
|
mtf_json, setup_json, reasons_json, warnings_json, event_risk, event_desc, news_sent,
|
|
ind_json, pen_json, exh, pen_total, mom, payloads_json) = row
|
|
inds = _loads(ind_json, {}) or {}
|
|
bar_label = bar_ts or (inds.get(tf) or {}).get("timestamp")
|
|
as_of_bar = _dt(bar_label) or _dt(logged_at)
|
|
if as_of_bar is None:
|
|
return None, None, {}, None
|
|
secs = ap._TF_SECONDS.get(tf, 900)
|
|
as_of = as_of_bar + timedelta(seconds=secs) # the bar closed
|
|
|
|
report = SignalReport(symbol=symbol, display_name="XAU/USD",
|
|
timestamp=str(bar_label or logged_at))
|
|
try:
|
|
report.final_signal = SignalStrength[signal]
|
|
except KeyError:
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.final_score = float(score or 0.0)
|
|
report.final_confidence = float(conf or 0.0)
|
|
report.signal_quality = quality or ""
|
|
report.current_price = float(price or 0.0)
|
|
report.market_regime = regime or ""
|
|
report.volatility_regime = vol or ""
|
|
mtf = _loads(mtf_json, {}) or {}
|
|
report.mtf_agreement = bool(mtf.get("agreement", False))
|
|
for k, v in (mtf.get("signals") or {}).items():
|
|
try:
|
|
report.mtf_signals[k] = SignalStrength[v]
|
|
except KeyError:
|
|
report.mtf_signals[k] = SignalStrength.NEUTRAL
|
|
report.mtf_scores = {k: float(v or 0.0) for k, v in (mtf.get("scores") or {}).items()}
|
|
reasons = _loads(reasons_json, {}) or {}
|
|
report.bullish_reasons = list(reasons.get("bullish") or [])
|
|
report.bearish_reasons = list(reasons.get("bearish") or [])
|
|
report.warnings = list(_loads(warnings_json, []) or [])
|
|
report.event_risk = event_risk or "clear"
|
|
report.event_description = event_desc or ""
|
|
report.news_sentiment = float(news_sent or 0.0)
|
|
report.exhaustion_count = int(exh or 0)
|
|
report.penalty_tags = {str(k): float(v or 0.0)
|
|
for k, v in (_loads(pen_json, {}) or {}).items()}
|
|
report.penalty_total = float(pen_total or 0.0)
|
|
report.momentum_fading = bool(mom)
|
|
setup = _loads(setup_json, None)
|
|
if setup:
|
|
report.trade_setup = TradeSetup(
|
|
pair=symbol, display_name="XAU/USD", direction=setup.get("direction"),
|
|
entry_price=float(setup.get("entry") or 0.0),
|
|
stop_loss=float(setup.get("stop_loss") or 0.0),
|
|
take_profit_1=float(setup.get("take_profit_1") or 0.0),
|
|
take_profit_2=float(setup.get("take_profit_2") or 0.0),
|
|
risk_reward=float(setup.get("risk_reward") or 0.0),
|
|
risk_pips=float(setup.get("risk_pips") or 0.0),
|
|
reward_pips=float(setup.get("reward_pips") or 0.0),
|
|
pip_value=0.10, strategy=setup.get("strategy") or "GENERIC",
|
|
setup_pattern=setup.get("setup_pattern"),
|
|
round_trip_cost=setup.get("round_trip_cost"),
|
|
trail_arm_r=setup.get("trail_arm_r"), trail_dist_r=setup.get("trail_dist_r"))
|
|
if tf in inds:
|
|
report.primary_analysis = _analysis(tf, inds[tf])
|
|
report.confirmation_analyses = [_analysis(k, v) for k, v in inds.items() if k != tf]
|
|
events_json, headlines_json = read_news(con, rowid)
|
|
context = _context_from_row(events_json, headlines_json, as_of)
|
|
frames = _frames_as_of([tf] + [k for k in inds if k != tf], as_of_bar)
|
|
return report, context, frames, _loads(payloads_json, None)
|
|
|
|
|
|
def build_jobs(report: SignalReport, context: FundamentalContext,
|
|
frames: Dict[Timeframe, pd.DataFrame], stored: Optional[Dict[str, Any]],
|
|
settings: Dict[str, Any]) -> Tuple[Tuple[str, str, Dict[str, Any]], ...]:
|
|
"""Persona jobs on the CURRENT prompts: stored payloads when the row
|
|
has them, else rebuilt through the live payload builders."""
|
|
systems = {"macro": ap._MACRO_SYSTEM, "technical": ap._TECH_SYSTEM,
|
|
"risk": ap._RISK_SYSTEM, "contrarian": ap._CONTRARIAN_SYSTEM}
|
|
if stored and all(p in stored for p in PERSONAS):
|
|
return tuple((p, systems[p], stored[p]) for p in PERSONAS)
|
|
candles: Dict[str, Any] = {}
|
|
analyses = [report.primary_analysis] + list(report.confirmation_analyses)
|
|
counts = [int(settings.get("candle_bars_primary", ap.DEFAULT_CANDLE_BARS_PRIMARY))] + \
|
|
[int(settings.get("candle_bars_confirm", ap.DEFAULT_CANDLE_BARS_CONFIRM))] * \
|
|
len(report.confirmation_analyses)
|
|
for analysis, n in zip(analyses, counts):
|
|
if analysis is None:
|
|
continue
|
|
atr = ap._finite(getattr((analysis.indicators or {}).get("atr"), "value", None))
|
|
df = frames.get(_TF.get(analysis.timeframe))
|
|
if atr and atr > 0 and df is not None and len(df) >= 2:
|
|
candles[analysis.timeframe] = ap._normalize_candles(df, atr, n)
|
|
ctx = ap._price_context(report, frames, context)
|
|
return ap._panel_jobs(report, context, candles or None, ctx, PERSONAS)
|
|
|
|
|
|
# ── Scoring ───────────────────────────────────────────────────────────
|
|
|
|
def _brier(ps: Sequence[float], ys: Sequence[int]) -> float:
|
|
return sum((p / 100.0 - y) ** 2 for p, y in zip(ps, ys)) / len(ps)
|
|
|
|
|
|
def _logloss(ps: Sequence[float], ys: Sequence[int]) -> float:
|
|
eps = 1e-3
|
|
tot = 0.0
|
|
for p, y in zip(ps, ys):
|
|
q = min(1 - eps, max(eps, p / 100.0))
|
|
tot += -(y * math.log(q) + (1 - y) * math.log(1 - q))
|
|
return tot / len(ps)
|
|
|
|
|
|
def score_predictor(name: str, ps: Sequence[Optional[float]], rs: Sequence[float]) -> Dict[str, Any]:
|
|
pairs = [(p, r) for p, r in zip(ps, rs) if p is not None]
|
|
if not pairs:
|
|
return {"predictor": name, "n": 0}
|
|
P = [p for p, _ in pairs]; R = [r for _, r in pairs]; Y = [1 if r > 0 else 0 for r in R]
|
|
bands = {}
|
|
for label, lo, hi in (("<=40", -1, ap.P_OPPOSE), ("40-60", ap.P_OPPOSE, ap.P_SUPPORT),
|
|
(">=60", ap.P_SUPPORT, 101)):
|
|
sel = [(p, r) for p, r in pairs if (lo < p <= hi if label == "<=40" else
|
|
lo < p < hi if label == "40-60" else p >= lo)]
|
|
if sel:
|
|
bands[label] = {"n": len(sel), "mean_p": round(sum(p for p, _ in sel) / len(sel), 1),
|
|
"win_rate": round(100 * sum(1 for _, r in sel if r > 0) / len(sel), 1),
|
|
"mean_r": round(sum(r for _, r in sel) / len(sel), 3),
|
|
"sum_r": round(sum(r for _, r in sel), 2)}
|
|
else:
|
|
bands[label] = {"n": 0}
|
|
stances = [ap.derive_stance(p) for p in P]
|
|
modal = max(set(stances), key=stances.count)
|
|
return {
|
|
"predictor": name, "n": len(P),
|
|
"brier": round(_brier(P, Y), 4), "logloss": round(_logloss(P, Y), 4),
|
|
"mean_p": round(sum(P) / len(P), 1), "win_rate": round(100 * sum(Y) / len(Y), 1),
|
|
"bands": bands, "modal_stance": modal,
|
|
"constant_voter_rate": round(100 * stances.count(modal) / len(stances), 1),
|
|
"sum_r_all": round(sum(R), 2),
|
|
"sum_r_skip_oppose": round(sum(r for p, r in pairs if p > ap.P_OPPOSE), 2),
|
|
"sum_r_support_only": round(sum(r for p, r in pairs if p >= ap.P_SUPPORT), 2),
|
|
}
|
|
|
|
|
|
def print_summary(results: List[Dict[str, Any]]) -> None:
|
|
print("\n== Predictor scorecard (outcome = advisory R > 0 at $0.65) ==")
|
|
print(f"{'predictor':22s} {'n':>3s} {'Brier':>6s} {'logloss':>7s} {'mean p':>6s} {'win%':>5s} "
|
|
f"{'modal':>8s} {'const%':>6s} {'R all':>7s} {'R skip<=40':>10s} {'R >=60 only':>11s}")
|
|
for r in results:
|
|
if r.get("n", 0) == 0:
|
|
print(f"{r['predictor']:22s} 0"); continue
|
|
print(f"{r['predictor']:22s} {r['n']:3d} {r['brier']:6.3f} {r['logloss']:7.3f} {r['mean_p']:6.1f} "
|
|
f"{r['win_rate']:5.1f} {r['modal_stance']:>8s} {r['constant_voter_rate']:6.1f} "
|
|
f"{r['sum_r_all']:+7.2f} {r['sum_r_skip_oppose']:+10.2f} {r['sum_r_support_only']:+11.2f}")
|
|
print("\n calibration bands (n / mean p / observed win% / mean R):")
|
|
for r in results:
|
|
if r.get("n", 0) == 0:
|
|
continue
|
|
cells = " ".join(f"{k}: {v['n']}/{v.get('mean_p', '-')}/{v.get('win_rate', '-')}/{v.get('mean_r', '-')}"
|
|
if v["n"] else f"{k}: 0" for k, v in r["bands"].items())
|
|
print(f" {r['predictor']:22s} {cells}")
|
|
print("\nCaveat: a few dozen outcomes from one regime can reject a pathological prompt "
|
|
"(constant voting, narrative anchoring); they cannot prove accuracy.")
|
|
|
|
|
|
# ── Main ──────────────────────────────────────────────────────────────
|
|
|
|
def main(argv: Optional[Sequence[str]] = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument("--db", default=os.path.join(ROOT, "logs", "signals.db"))
|
|
parser.add_argument("--limit", type=int, default=0, help="max events (0 = all)")
|
|
parser.add_argument("--max-calls", type=int, default=200,
|
|
help="hard stop on LLM requests (4 per event)")
|
|
parser.add_argument("--dry-run", action="store_true",
|
|
help="build payloads only; no LLM call")
|
|
parser.add_argument("--out", default=None, help="per-event CSV (and .payloads.json on --dry-run)")
|
|
parser.add_argument("--include-pending", action="store_true",
|
|
help="also build payloads for unresolved setups (dry-run only)")
|
|
parser.add_argument("--personas", default=",".join(ap.ALL_PERSONAS),
|
|
help="comma-separated personas to run (default: all four, so every "
|
|
"voice is scored; the live default is agent_panel.personas = macro)")
|
|
args = parser.parse_args(argv)
|
|
global PERSONAS
|
|
PERSONAS = ap.normalize_personas([p for p in args.personas.split(",") if p.strip()])
|
|
|
|
rows = load_rows(args.db)
|
|
events = cluster_setup_rows(rows)
|
|
con = open_readonly(args.db)
|
|
global _have_cols
|
|
_have_cols = {r[1] for r in con.execute("PRAGMA table_info(signals)")}
|
|
settings = {"candle_bars_primary": ap.DEFAULT_CANDLE_BARS_PRIMARY,
|
|
"candle_bars_confirm": ap.DEFAULT_CANDLE_BARS_CONFIRM}
|
|
picked = []
|
|
for ev in events:
|
|
res = next((r for r in ev.rows if r.advisory_status == "RESOLVED" and r.advisory_r is not None), None)
|
|
if res is None and not (args.dry_run and args.include_pending):
|
|
continue
|
|
picked.append((ev, res))
|
|
if args.limit:
|
|
picked = picked[:args.limit]
|
|
print(f"events: {len(events)} total, {len(picked)} selected "
|
|
f"({'dry run' if args.dry_run else 'live calls, max ' + str(args.max_calls)}), "
|
|
f"prompt version {ap.PANEL_PROMPT_VERSION}, personas {','.join(PERSONAS)}")
|
|
|
|
client = None if args.dry_run else ap.get_llm_client()
|
|
calls = 0
|
|
per_event: List[Dict[str, Any]] = []
|
|
payload_dump: Dict[str, Any] = {}
|
|
for ev, res in picked:
|
|
rep = ev.representative
|
|
# Prefer a member row that already carries the v2 payloads.
|
|
rowid = rep.id
|
|
for r in ev.rows:
|
|
has = (con.execute("SELECT panel_payloads_json IS NOT NULL FROM signals"
|
|
" WHERE id=?", (r.id,)).fetchone()
|
|
if "panel_payloads_json" in _have_cols else None)
|
|
if has and has[0]:
|
|
rowid = r.id
|
|
break
|
|
report, context, frames, stored = build_report(con, rowid)
|
|
if report is None or report.trade_setup is None:
|
|
continue
|
|
jobs = build_jobs(report, context, frames, stored, settings)
|
|
ctx = next((payload.get("context") for _, _, payload in jobs if payload.get("context")), {}) or {}
|
|
base_p = (ctx.get("base_rate") or {}).get("win_rate")
|
|
# logged v1 takes (any member row) -> pseudo-probabilities
|
|
v1 = {}
|
|
for r in ev.rows:
|
|
takes = con.execute("SELECT panel_takes_json, " + _col("panel_prompt_version")
|
|
+ " FROM signals WHERE id=?", (r.id,)).fetchone()
|
|
if takes and takes[0]:
|
|
for t in _loads(takes[0], []) or []:
|
|
p = t.get("p_win")
|
|
if p is None and t.get("stance") in ap._VALID_STANCES:
|
|
p = ap.pseudo_probability(t["stance"], float(t.get("confidence") or 0))
|
|
if p is not None:
|
|
v1[t.get("persona")] = float(p)
|
|
v1["_version"] = takes[1] or "v1-logged"
|
|
break
|
|
entry: Dict[str, Any] = {
|
|
"event_row": rep.id, "bar": report.timestamp, "tf": rep.timeframe,
|
|
"direction": report.trade_setup.direction, "grade": rep.quality,
|
|
"strategy": report.trade_setup.strategy,
|
|
"R": res.advisory_r if res else None, "stored_payloads": bool(stored),
|
|
"base_p": base_p, "prior_move_4h_atr": ctx.get("prior_move_4h_atr"),
|
|
"move_from_start_atr": ctx.get("move_from_24h_start_atr"),
|
|
"big_candle_last3": ctx.get("big_candle_last3"),
|
|
}
|
|
for p in PERSONAS:
|
|
entry[f"v1_{p}"] = v1.get(p)
|
|
v1_ps = [v1[p] for p in PERSONAS if p in v1]
|
|
entry["v1_mean"] = round(sum(v1_ps) / len(v1_ps), 1) if len(v1_ps) >= min(ap.MEAN_QUORUM, len(v1_ps) or 1) and v1_ps else None
|
|
if args.dry_run:
|
|
payload_dump[str(rep.id)] = {name: payload for name, _, payload in jobs}
|
|
else:
|
|
takes = []
|
|
for name, system, payload in jobs:
|
|
if calls >= args.max_calls:
|
|
break
|
|
calls += 1
|
|
reply = client.chat_json(system, payload)
|
|
take = ap._validate_take(name, reply)
|
|
if take is not None:
|
|
takes.append(take)
|
|
entry[f"new_{name}"] = take.p_win
|
|
entry[f"new_{name}_for"] = take.reason_for
|
|
entry[f"new_{name}_against"] = take.reason_against
|
|
v = ap.mean_verdict(takes, quorum=min(ap.MEAN_QUORUM, len(jobs)))
|
|
entry["new_mean"] = v.probability if v else None
|
|
entry["new_verdict"] = v.verdict if v else None
|
|
print(f" #{rep.id:<5d} {report.timestamp[:16]} {rep.timeframe:5s} "
|
|
f"{report.trade_setup.direction:4s} {rep.quality:2s} R={entry['R']:+.2f} "
|
|
f"base={base_p} new_mean={entry['new_mean']} "
|
|
f"({', '.join(f'{t.persona[:4]} {t.p_win:.0f}' for t in takes)}) "
|
|
f"v1_mean={entry['v1_mean']}")
|
|
per_event.append(entry)
|
|
if not args.dry_run and calls >= args.max_calls:
|
|
print(f" stopped at --max-calls {args.max_calls}")
|
|
break
|
|
con.close()
|
|
|
|
if args.dry_run:
|
|
print(f"built payloads for {len(per_event)} events")
|
|
if args.out:
|
|
with open(args.out.rsplit('.', 1)[0] + ".payloads.json", "w") as fh:
|
|
json.dump(payload_dump, fh, indent=1, sort_keys=True)
|
|
print(f"wrote {args.out.rsplit('.', 1)[0]}.payloads.json")
|
|
|
|
scored = [e for e in per_event if e["R"] is not None]
|
|
if scored:
|
|
rs = [e["R"] for e in scored]
|
|
results = [score_predictor("base-rate only",
|
|
[e["base_p"] if e["base_p"] is not None else None for e in scored], rs),
|
|
score_predictor("v1 logged (mean)", [e["v1_mean"] for e in scored], rs)]
|
|
for p in PERSONAS:
|
|
results.append(score_predictor(f"v1 logged {p}", [e.get(f"v1_{p}") for e in scored], rs))
|
|
if not args.dry_run:
|
|
results.append(score_predictor(f"NEW {ap.PANEL_PROMPT_VERSION} (mean)",
|
|
[e.get("new_mean") for e in scored], rs))
|
|
for p in PERSONAS:
|
|
results.append(score_predictor(f"NEW {p}", [e.get(f"new_{p}") for e in scored], rs))
|
|
print_summary(results)
|
|
if args.out and per_event:
|
|
pd.DataFrame(per_event).to_csv(args.out, index=False)
|
|
print(f"wrote {args.out} ({len(per_event)} events)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|