feat(replay): R0 snapshot recorder — archive per-cycle decisions into signals
The signals and markets tables existed since Phase 2/5 but never had a writer; the replay engine (phase plan line 2.1) needs a per-(market, cycle) archive of what the strategy saw and decided. This wires them up: - signals: one row per evaluated market per cycle, now carrying INPUTS (news_sentiment, feat_*_lo, volume_24h, days_to_resolution) plus the existing outputs (probs, edges, gates, skip_reason). skip_reason is granular: unsupported/no_signals/prior_extreme/family/edge_net/ confidence/reentry_guard. news_budget_skipped distinguishes "GNews not asked" (5-query budget) from "no news". - ext_snapshots: one row per cycle with the ExternalSignals snapshot; signals rows join on cycle_ts. - markets: metadata upserted each cycle (replay rebuilds Market from it). - Retention: prune > SIGNALS_RETENTION_DAYS (default 90) once a day. - SIGNAL_RECORDER_ENABLED (default true) gates all DB writes; every write is try/except — the recorder can never break trading. Strategy changes are purely additive (record accumulation at each exit path of evaluate()); no weights, thresholds, gates or sizing touched, per the freeze in the current phase plan. Tests: 10 new deterministic tests (85 total passing). Schema migration dry-run validated against prod postgres inside a rolled-back transaction. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
117d2b33b2
commit
919fe1617a
+43
@@ -43,6 +43,14 @@ PAPER_BANKROLL = float(os.getenv("PAPER_BANKROLL", "10000"))
|
||||
# position per 10 minutes.
|
||||
RESOLUTION_CHECK_INTERVAL = 10
|
||||
|
||||
# Replay R0: persist per-(market, cycle) decision records + the ExternalSignals
|
||||
# snapshot each cycle, so the replay engine can re-run past decisions. The
|
||||
# recorder must never break trading — every write is wrapped in try/except.
|
||||
SIGNAL_RECORDER_ENABLED = os.getenv("SIGNAL_RECORDER_ENABLED", "true").lower() == "true"
|
||||
SIGNALS_RETENTION_DAYS = int(os.getenv("SIGNALS_RETENTION_DAYS", "90"))
|
||||
# Prune the archive roughly once a day at the 60s cycle cadence.
|
||||
SIGNALS_PRUNE_INTERVAL_CYCLES = 1440
|
||||
|
||||
|
||||
async def check_resolutions(
|
||||
poly: PolymarketClient,
|
||||
@@ -122,6 +130,16 @@ async def run_trading_loop(
|
||||
# 2. Get external signals
|
||||
ext_data = await external.get_all_signals()
|
||||
|
||||
# 2b. Replay R0: archive this cycle's inputs (ext snapshot + market
|
||||
# metadata). cycle_ts groups all signals rows of this cycle.
|
||||
cycle_ts = datetime.now(timezone.utc)
|
||||
if SIGNAL_RECORDER_ENABLED:
|
||||
try:
|
||||
await db.save_ext_snapshot(cycle_ts, ext_data)
|
||||
await db.upsert_markets(markets)
|
||||
except Exception as exc:
|
||||
log.warning("Signal recorder (inputs) failed: %s", exc)
|
||||
|
||||
# 3. Build occupied_families from the current open portfolio positions.
|
||||
# This prevents re-entering a family where we already hold a position.
|
||||
# We also pull from DB to survive pod restarts.
|
||||
@@ -176,6 +194,7 @@ async def run_trading_loop(
|
||||
|
||||
reentry_guard_count = 0
|
||||
cycle_trades = 0
|
||||
traded_market_ids: set[str] = set()
|
||||
for market in markets:
|
||||
if market.id in inverted_guard:
|
||||
log.info(
|
||||
@@ -183,6 +202,7 @@ async def run_trading_loop(
|
||||
market.id, market.question[:60],
|
||||
)
|
||||
reentry_guard_count += 1
|
||||
strategy.record_skip(market, "reentry_guard")
|
||||
continue
|
||||
|
||||
# evaluate() returns None for all skips — reasons are logged internally
|
||||
@@ -214,6 +234,7 @@ async def run_trading_loop(
|
||||
# Block this family for the rest of the cycle (Phase 2)
|
||||
occupied_families.add(signal.family_key)
|
||||
cycle_trades += 1
|
||||
traded_market_ids.add(market.id)
|
||||
# Mark manifold audit record as used in this trade
|
||||
if signal.mfld_audit_id:
|
||||
try:
|
||||
@@ -221,6 +242,28 @@ async def run_trading_loop(
|
||||
except Exception as exc:
|
||||
log.warning("Failed to mark manifold audit used: %s", exc)
|
||||
|
||||
# 7b. Replay R0: flush this cycle's decision records to the archive.
|
||||
# acted_on marks records whose signal actually became a trade
|
||||
# (evaluate() can emit a signal that risk sizing later rejects).
|
||||
records = strategy.drain_cycle_records()
|
||||
if SIGNAL_RECORDER_ENABLED and records:
|
||||
for rec in records:
|
||||
if rec["market_id"] in traded_market_ids:
|
||||
rec["acted_on"] = True
|
||||
try:
|
||||
await db.save_signal_records(cycle_ts, records)
|
||||
except Exception as exc:
|
||||
log.warning("Signal recorder (records) failed: %s", exc)
|
||||
if cycle_count % SIGNALS_PRUNE_INTERVAL_CYCLES == 1:
|
||||
try:
|
||||
pruned = await db.prune_signal_records(SIGNALS_RETENTION_DAYS)
|
||||
log.info(
|
||||
"Signal archive pruned: %d rows older than %d days removed",
|
||||
pruned, SIGNALS_RETENTION_DAYS,
|
||||
)
|
||||
except Exception as exc:
|
||||
log.warning("Signal archive prune failed: %s", exc)
|
||||
|
||||
# 8. [CYCLE SUMMARY] — one block per cycle, stable format for grep/compare
|
||||
stats = strategy.get_cycle_stats()
|
||||
legacy_incomplete_count = await db.get_legacy_incomplete_count()
|
||||
|
||||
Reference in New Issue
Block a user