chemavxandClaude Fable 5
919fe1617a
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>
956 lines
48 KiB
Python
956 lines
48 KiB
Python
"""Database layer using asyncpg for PostgreSQL."""
|
||
import logging
|
||
import os
|
||
from typing import Optional
|
||
import asyncpg
|
||
|
||
from bot.data.manifold import MANIFOLD_MATCHER_VERSION
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
class Database:
|
||
def __init__(self) -> None:
|
||
self._url = os.getenv("DATABASE_URL", "postgresql://bot:bot@localhost:5432/polymarket")
|
||
self._pool: Optional[asyncpg.Pool] = None
|
||
|
||
async def connect(self) -> None:
|
||
self._pool = await asyncpg.create_pool(self._url)
|
||
log.info("Database connected")
|
||
|
||
async def disconnect(self) -> None:
|
||
if self._pool:
|
||
await self._pool.close()
|
||
|
||
async def run_migrations(self) -> None:
|
||
schema_path = os.path.join(os.path.dirname(__file__), "schema.sql")
|
||
with open(schema_path) as f:
|
||
schema = f.read()
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute(schema)
|
||
log.info("Migrations applied")
|
||
|
||
async def save_trade(self, trade) -> None:
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute("""
|
||
INSERT INTO trades (
|
||
id, market_id, question, direction, size_usdc,
|
||
entry_price, shares, fee_usdc, net_cost, timestamp, reasoning, paper,
|
||
edge_gross, edge_net, prior_prob, final_prob,
|
||
mid_price, spread_estimate, commission, family_key,
|
||
feat_fg_lo, feat_mom_lo, feat_news_lo, feat_mfld_lo, feat_btc_dom_lo,
|
||
mfld_market_id, mfld_market_title, mfld_market_url,
|
||
mfld_prob_raw, mfld_prob_final, mfld_inverted,
|
||
mfld_match_score, mfld_match_reason, mfld_match_status
|
||
) VALUES (
|
||
$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,
|
||
$13,$14,$15,$16,$17,$18,$19,$20,
|
||
$21,$22,$23,$24,$25,
|
||
$26,$27,$28,$29,$30,$31,$32,$33,$34
|
||
)
|
||
ON CONFLICT (id) DO NOTHING
|
||
""",
|
||
trade.id, trade.market_id, trade.question, trade.direction,
|
||
trade.size_usdc, trade.entry_price, trade.shares, trade.fee_usdc,
|
||
trade.net_cost, trade.timestamp, trade.reasoning, trade.paper,
|
||
# Phase 1 fields
|
||
trade.edge_gross, trade.edge_net, trade.prior_prob, trade.final_prob,
|
||
trade.mid_price, trade.spread_estimate, trade.commission, trade.family_key,
|
||
# Phase 6 feature log-odds
|
||
trade.feat_fg_lo, trade.feat_mom_lo, trade.feat_news_lo,
|
||
trade.feat_mfld_lo, trade.feat_btc_dom_lo,
|
||
# Manifold audit fields
|
||
trade.mfld_market_id, trade.mfld_market_title, trade.mfld_market_url,
|
||
trade.mfld_prob_raw, trade.mfld_prob_final, trade.mfld_inverted,
|
||
trade.mfld_match_score, trade.mfld_match_reason, trade.mfld_match_status,
|
||
)
|
||
|
||
async def save_daily_metrics(self, metrics: dict) -> None:
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute("""
|
||
INSERT INTO metrics_daily (
|
||
timestamp, total_trades, total_deployed, total_fees,
|
||
unrealized_pnl_est, realized_pnl, total_pnl,
|
||
win_rate, avg_edge, sharpe_ratio, calibration_score, paper_mode,
|
||
open_count, closed_count, resolved_count
|
||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
|
||
""",
|
||
metrics["timestamp"],
|
||
metrics["total_trades"],
|
||
metrics["total_deployed"],
|
||
metrics["total_fees"],
|
||
metrics["unrealized_pnl_est"],
|
||
metrics["realized_pnl"],
|
||
metrics["total_pnl"],
|
||
metrics["win_rate"],
|
||
metrics["avg_edge"],
|
||
metrics["sharpe_ratio"],
|
||
metrics["calibration_score"],
|
||
metrics["paper_mode"],
|
||
metrics["open_count"],
|
||
metrics["closed_count"],
|
||
metrics["resolved_count"],
|
||
)
|
||
|
||
async def get_open_positions(self) -> dict[str, float]:
|
||
"""Return {market_id: total_net_cost} for all open (not closed) trades in DB."""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch(
|
||
"SELECT market_id, SUM(net_cost) AS total "
|
||
"FROM trades WHERE closed_at IS NULL GROUP BY market_id"
|
||
)
|
||
return {r["market_id"]: float(r["total"]) for r in rows}
|
||
|
||
async def get_open_position_data(self) -> tuple[dict[str, float], float]:
|
||
"""Return (positions_by_size_usdc, total_net_cost) for all open trades.
|
||
|
||
positions_by_size_usdc — {market_id: size_usdc} mirrors what live trading
|
||
stores in portfolio.positions (no fee included).
|
||
total_net_cost — SUM(net_cost) across all open trades, used to
|
||
reconstruct cash = bankroll − total_net_cost.
|
||
|
||
Together these let initialize() replicate the exact same accounting model
|
||
that execute() uses at runtime, eliminating the phantom exposure overage
|
||
caused by the old net_cost-in-positions approach.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch(
|
||
"SELECT market_id, SUM(size_usdc) AS sz, SUM(net_cost) AS nc "
|
||
"FROM trades WHERE closed_at IS NULL GROUP BY market_id"
|
||
)
|
||
positions = {r["market_id"]: float(r["sz"]) for r in rows}
|
||
total_net_cost = sum(float(r["nc"]) for r in rows)
|
||
return positions, total_net_cost
|
||
|
||
async def get_open_families(self) -> set[str]:
|
||
"""Return the set of family_key values from all open positions.
|
||
|
||
Used at startup to rebuild occupied_families from DB state so the
|
||
family-deduplication logic survives pod restarts.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch(
|
||
"SELECT DISTINCT family_key FROM trades "
|
||
"WHERE family_key IS NOT NULL AND closed_at IS NULL"
|
||
)
|
||
return {r["family_key"] for r in rows if r["family_key"]}
|
||
|
||
async def get_open_position_details(self) -> list[dict]:
|
||
"""Return one row per open position with family_key and direction.
|
||
|
||
Used at startup to detect positions that share a family_key (same
|
||
underlying event), which indicates a contradictory paper trade entered
|
||
before the general-election family fix was deployed.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch("""
|
||
SELECT DISTINCT ON (market_id)
|
||
market_id, question, direction, edge_net, family_key, timestamp
|
||
FROM trades
|
||
WHERE paper = TRUE AND closed_at IS NULL
|
||
ORDER BY market_id, timestamp DESC
|
||
""")
|
||
return [dict(r) for r in rows]
|
||
|
||
async def get_open_trades_for_market(self, market_id: str) -> list[dict]:
|
||
"""Return direction, shares and net_cost for each open trade in a market.
|
||
|
||
Used by PaperExecutor.close_position() to compute the settlement
|
||
payout per direction (BUY_NO pays out when resolution = 0.0).
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch(
|
||
"SELECT direction, shares, net_cost FROM trades "
|
||
"WHERE market_id = $1 AND closed_at IS NULL",
|
||
market_id,
|
||
)
|
||
return [dict(r) for r in rows]
|
||
|
||
async def close_paper_position(
|
||
self, market_id: str, reason: str = "", resolution: Optional[float] = None
|
||
) -> None:
|
||
"""Mark a paper position as closed.
|
||
|
||
resolution: 1.0 if YES resolved, 0.0 if NO resolved, None if unknown
|
||
(legacy closes, inversion fixes). When resolution is provided, close_pnl
|
||
is computed in SQL per row as payout − net_cost — NET of fee, the single
|
||
PnL definition shared with PaperExecutor.close_position() (logs/Telegram):
|
||
BUY_YES: resolution * shares − net_cost
|
||
BUY_NO: (1 − resolution) * shares − net_cost
|
||
paper.py aggregates payout − net_cost over these same open rows, so
|
||
SUM(close_pnl) per market equals the pnl it reports exactly. The
|
||
aggregate is intentionally NOT passed in as a parameter: writing it to
|
||
every row would double-count markets with more than one open trade.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
# $3 is cast on every use: Postgres cannot infer the parameter type
|
||
# from a bare "$3 IS NOT NULL" and fails the prepare with
|
||
# AmbiguousParameterError otherwise.
|
||
await conn.execute("""
|
||
UPDATE trades
|
||
SET closed_at = NOW(),
|
||
close_reason = $2,
|
||
resolution = $3::double precision,
|
||
close_pnl = CASE
|
||
WHEN $3::double precision IS NOT NULL AND direction = 'BUY_YES'
|
||
THEN ($3::double precision * shares) - net_cost
|
||
WHEN $3::double precision IS NOT NULL AND direction = 'BUY_NO'
|
||
THEN ((1.0 - $3::double precision) * shares) - net_cost
|
||
ELSE NULL
|
||
END
|
||
WHERE market_id = $1 AND closed_at IS NULL
|
||
""", market_id, reason, resolution)
|
||
|
||
async def update_family_key(self, market_id: str, new_key: str) -> None:
|
||
"""Persist a corrected family_key for all open trades of a market."""
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute(
|
||
"UPDATE trades SET family_key = $2 WHERE market_id = $1 AND closed_at IS NULL",
|
||
market_id, new_key,
|
||
)
|
||
|
||
async def get_legacy_incomplete_count(self) -> int:
|
||
"""Return count of open trades with NULL edge_net (legacy data without signal values)."""
|
||
async with self._pool.acquire() as conn:
|
||
row = await conn.fetchrow(
|
||
"SELECT COUNT(*) FROM trades WHERE closed_at IS NULL AND edge_net IS NULL"
|
||
)
|
||
return int(row[0])
|
||
|
||
async def get_recently_closed_inverted(self, hours: int = 24) -> set[str]:
|
||
"""Return market_ids closed for inversion bug within the last N hours.
|
||
|
||
Used as a reentry guard: prevents re-entering a market that was just
|
||
closed because the signal direction was inverted.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch("""
|
||
SELECT DISTINCT market_id FROM trades
|
||
WHERE closed_at > NOW() - ($1 || ' hours')::interval
|
||
AND close_reason ILIKE '%inversion bug%'
|
||
""", str(hours))
|
||
return {r["market_id"] for r in rows}
|
||
|
||
async def compute_metrics_from_db(self) -> dict:
|
||
"""Compute all trading metrics directly from the trades table.
|
||
|
||
This is the single source of truth for MetricsTracker — no in-memory
|
||
state required. Safe to call after pod restarts: always reflects the
|
||
full DB history.
|
||
|
||
Returns a dict with keys:
|
||
total_trades, open_count, closed_count, resolved_count,
|
||
total_deployed, total_fees,
|
||
unrealized_pnl_est — estimated, open trades with edge_net
|
||
realized_pnl — exact, closed trades with resolution
|
||
wins_realized — closed trades where close_pnl > 0
|
||
calibration_score — Brier-based (1 − MSE), null if resolved < 10
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
row = await conn.fetchrow("""
|
||
SELECT
|
||
COUNT(*) AS total_trades,
|
||
COUNT(*) FILTER (WHERE closed_at IS NULL) AS open_count,
|
||
COUNT(*) FILTER (WHERE closed_at IS NOT NULL) AS closed_count,
|
||
-- excluded_from_metrics trades are omitted from resolved_count,
|
||
-- realized_pnl, wins_realized, and calibration_score.
|
||
-- resolved_count does NOT require final_prob: legacy trades
|
||
-- without signal data still count as resolved. Calibration
|
||
-- below keeps the final_prob requirement (it needs the
|
||
-- estimated probability to score).
|
||
COUNT(*) FILTER (WHERE resolution IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)) AS resolved_count,
|
||
|
||
COALESCE(SUM(net_cost)
|
||
FILTER (WHERE closed_at IS NULL), 0) AS total_deployed,
|
||
COALESCE(SUM(fee_usdc), 0) AS total_fees,
|
||
|
||
-- Estimated unrealized PnL: open trades with known edge.
|
||
-- Formula: edge_net × net_cost − fee_usdc.
|
||
-- Trades with NULL edge_net (legacy data) are excluded.
|
||
COALESCE(SUM(edge_net * net_cost - fee_usdc)
|
||
FILTER (WHERE closed_at IS NULL
|
||
AND edge_net IS NOT NULL), 0) AS unrealized_pnl_est,
|
||
|
||
-- Realized PnL: admin-excluded trades omitted (close_pnl=0 by convention
|
||
-- but excluded explicitly so they don't skew the aggregate).
|
||
COALESCE(SUM(close_pnl)
|
||
FILTER (WHERE closed_at IS NOT NULL
|
||
AND close_pnl IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)), 0) AS realized_pnl,
|
||
|
||
COUNT(*) FILTER (WHERE closed_at IS NOT NULL
|
||
AND close_pnl IS NOT NULL
|
||
AND close_pnl > 0
|
||
AND (excluded_from_metrics IS NOT TRUE)) AS wins_realized,
|
||
|
||
-- Calibration (Brier score transformed to higher-is-better):
|
||
-- 1 − AVG((final_prob − resolution)²) on resolved trades.
|
||
-- final_prob is the model's estimated YES probability at entry.
|
||
-- resolution is 1.0 (YES won) or 0.0 (NO won).
|
||
-- Perfect calibration → 1.0 | Random → ~0.75 | Worst → 0.0
|
||
-- Returns NULL if fewer than 10 resolved trades with final_prob.
|
||
-- Admin-excluded trades omitted from both threshold and average.
|
||
CASE
|
||
WHEN COUNT(*) FILTER (WHERE resolution IS NOT NULL
|
||
AND final_prob IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)) >= 10
|
||
THEN 1.0 - AVG((final_prob - resolution) * (final_prob - resolution))
|
||
FILTER (WHERE resolution IS NOT NULL
|
||
AND final_prob IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE))
|
||
ELSE NULL
|
||
END AS calibration_score
|
||
|
||
FROM trades
|
||
""")
|
||
return dict(row)
|
||
|
||
async def get_recent_trades(self, limit: int = 100, status: Optional[str] = None) -> list[dict]:
|
||
"""Return trades ordered by timestamp DESC.
|
||
|
||
status: None (all) | "open" (closed_at IS NULL) | "closed" (closed_at IS NOT NULL)
|
||
Each row includes a computed "status" field ("open" or "closed").
|
||
"""
|
||
if status == "open":
|
||
where = "WHERE closed_at IS NULL"
|
||
elif status == "closed":
|
||
where = "WHERE closed_at IS NOT NULL"
|
||
else:
|
||
where = ""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch(
|
||
f"SELECT * FROM trades {where} ORDER BY timestamp DESC LIMIT $1", limit
|
||
)
|
||
result = []
|
||
for r in rows:
|
||
d = dict(r)
|
||
d["status"] = "closed" if d.get("closed_at") else "open"
|
||
result.append(d)
|
||
return result
|
||
|
||
async def get_metrics_history(self, days: int = 42) -> list[dict]:
|
||
"""Return the closing snapshot of each UTC day, newest day first.
|
||
|
||
metrics_daily receives one snapshot per trading cycle (~1/min), so a
|
||
plain LIMIT over raw rows would cover minutes, not days. DISTINCT ON
|
||
collapses each calendar day to its last snapshot, making `days` bound
|
||
actual days. history[0] remains the most recent snapshot overall.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch(
|
||
"""
|
||
SELECT DISTINCT ON (timestamp::date) *
|
||
FROM metrics_daily
|
||
ORDER BY timestamp::date DESC, timestamp DESC
|
||
LIMIT $1
|
||
""", days
|
||
)
|
||
return [dict(r) for r in rows]
|
||
|
||
async def get_daily_pnl_closes(self) -> list[float]:
|
||
"""Return the closing total_pnl of every observed UTC day, oldest first.
|
||
|
||
One value per calendar day with at least one metrics_daily snapshot
|
||
(the day's last snapshot, same collapse rule as get_metrics_history).
|
||
This is the input series for the Sharpe ratio: len() = days observed,
|
||
consecutive deltas = daily PnL changes.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch(
|
||
"""
|
||
SELECT DISTINCT ON (timestamp::date) total_pnl
|
||
FROM metrics_daily
|
||
ORDER BY timestamp::date ASC, timestamp DESC
|
||
"""
|
||
)
|
||
return [float(r["total_pnl"] or 0) for r in rows]
|
||
|
||
async def backfill_feature_columns(self) -> int:
|
||
"""Back-populate feat_*_lo for trades created before Phase 6.
|
||
|
||
Parses the reasoning string (format: 'fg=+0.0600 mom=... news=... mfld=...').
|
||
fg / mom raw values are multiplied by 2 to convert to log-odds.
|
||
news / mfld are already in log-odds (no scaling).
|
||
feat_btc_dom_lo cannot be recovered from the old reasoning string and
|
||
remains NULL for legacy trades.
|
||
|
||
Returns the number of rows updated.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
result = await conn.execute("""
|
||
UPDATE trades
|
||
SET
|
||
feat_fg_lo = ((regexp_match(reasoning, 'fg=([^ |]+)'))[1])::DOUBLE PRECISION * 2,
|
||
feat_mom_lo = ((regexp_match(reasoning, 'mom=([^ |]+)'))[1])::DOUBLE PRECISION * 2,
|
||
feat_news_lo = ((regexp_match(reasoning, 'news=([^ |]+)'))[1])::DOUBLE PRECISION,
|
||
feat_mfld_lo = ((regexp_match(reasoning, 'mfld=([^ |]+)'))[1])::DOUBLE PRECISION,
|
||
feat_btc_dom_lo = NULL
|
||
WHERE feat_fg_lo IS NULL
|
||
AND reasoning IS NOT NULL
|
||
AND reasoning LIKE '%fg=%'
|
||
AND reasoning NOT LIKE '%fg_lo=%'
|
||
""")
|
||
updated = int(result.split()[-1]) if result else 0
|
||
if updated:
|
||
log.info("backfill_feature_columns: updated %d trade(s)", updated)
|
||
return updated
|
||
|
||
async def get_legacy_incomplete_trades(self) -> list[dict]:
|
||
"""Return trades with NULL edge_net — pre-Phase-1 data with no signal quality info."""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch("""
|
||
SELECT id, market_id, question, direction, net_cost, entry_price,
|
||
timestamp, reasoning, closed_at, close_reason, family_key,
|
||
feat_fg_lo, feat_mom_lo, feat_news_lo, feat_mfld_lo, feat_btc_dom_lo
|
||
FROM trades
|
||
WHERE edge_net IS NULL
|
||
ORDER BY timestamp DESC
|
||
""")
|
||
return [dict(r) for r in rows]
|
||
|
||
async def compute_feature_metrics_from_db(self) -> dict:
|
||
"""Per-feature performance metrics, all in log-odds space.
|
||
|
||
For each feature (fg, mom, news, mfld, btc_dom) returns:
|
||
unit — always "log_odds"
|
||
materiality_threshold — |lo| threshold for "material" classification
|
||
triggered_count — trades where |feat_lo| > 0.0001
|
||
material_count — trades where |feat_lo| >= materiality_threshold
|
||
avg_contribution_lo — mean signed lo value (triggered trades)
|
||
avg_abs_contribution_lo — mean absolute lo value (triggered trades)
|
||
avg_edge_net_when_material — mean edge_net for material trades
|
||
unrealized_pnl_est — sum edge_net*net_cost−fee for triggered open trades
|
||
realized_pnl — sum close_pnl for triggered resolved trades
|
||
resolved_count — closed trades with known outcome (triggered)
|
||
win_rate — NULL if resolved_count < 5
|
||
net_positive_count — triggered trades where feat_lo > 0
|
||
net_negative_count — triggered trades where feat_lo < 0
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch("""
|
||
WITH feature_values AS (
|
||
SELECT 'fg' AS feature,
|
||
0.05::DOUBLE PRECISION AS mat_thresh,
|
||
feat_fg_lo AS fval,
|
||
edge_net, net_cost, fee_usdc, closed_at, close_pnl
|
||
FROM trades WHERE feat_fg_lo IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)
|
||
UNION ALL
|
||
SELECT 'mom', 0.05, feat_mom_lo,
|
||
edge_net, net_cost, fee_usdc, closed_at, close_pnl
|
||
FROM trades WHERE feat_mom_lo IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)
|
||
UNION ALL
|
||
SELECT 'news', 0.10, feat_news_lo,
|
||
edge_net, net_cost, fee_usdc, closed_at, close_pnl
|
||
FROM trades WHERE feat_news_lo IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)
|
||
UNION ALL
|
||
SELECT 'mfld', 0.10, feat_mfld_lo,
|
||
edge_net, net_cost, fee_usdc, closed_at, close_pnl
|
||
FROM trades WHERE feat_mfld_lo IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)
|
||
UNION ALL
|
||
SELECT 'btc_dom', 0.05, feat_btc_dom_lo,
|
||
edge_net, net_cost, fee_usdc, closed_at, close_pnl
|
||
FROM trades WHERE feat_btc_dom_lo IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)
|
||
)
|
||
SELECT
|
||
feature,
|
||
mat_thresh AS materiality_threshold,
|
||
COUNT(*) FILTER (WHERE ABS(fval) > 0.0001) AS triggered_count,
|
||
COUNT(*) FILTER (WHERE ABS(fval) >= mat_thresh) AS material_count,
|
||
AVG(fval) FILTER (WHERE ABS(fval) > 0.0001) AS avg_contribution_lo,
|
||
AVG(ABS(fval)) FILTER (WHERE ABS(fval) > 0.0001) AS avg_abs_contribution_lo,
|
||
AVG(edge_net) FILTER (WHERE ABS(fval) >= mat_thresh
|
||
AND edge_net IS NOT NULL) AS avg_edge_net_when_material,
|
||
COALESCE(SUM(edge_net * net_cost - fee_usdc)
|
||
FILTER (WHERE ABS(fval) > 0.0001
|
||
AND closed_at IS NULL
|
||
AND edge_net IS NOT NULL), 0) AS unrealized_pnl_est,
|
||
COALESCE(SUM(close_pnl)
|
||
FILTER (WHERE ABS(fval) > 0.0001
|
||
AND close_pnl IS NOT NULL), 0) AS realized_pnl,
|
||
COUNT(*) FILTER (WHERE ABS(fval) > 0.0001
|
||
AND close_pnl IS NOT NULL
|
||
AND close_pnl > 0) AS wins_realized,
|
||
COUNT(*) FILTER (WHERE ABS(fval) > 0.0001
|
||
AND close_pnl IS NOT NULL) AS resolved_count,
|
||
COUNT(*) FILTER (WHERE fval > 0.0001) AS net_positive_count,
|
||
COUNT(*) FILTER (WHERE fval < -0.0001) AS net_negative_count
|
||
FROM feature_values
|
||
GROUP BY feature, mat_thresh
|
||
ORDER BY feature
|
||
""")
|
||
|
||
result: dict[str, dict] = {}
|
||
for r in rows:
|
||
d = dict(r)
|
||
feature = d["feature"]
|
||
resolved = int(d.get("resolved_count") or 0)
|
||
wins = int(d.get("wins_realized") or 0)
|
||
result[feature] = {
|
||
"unit": "log_odds",
|
||
"materiality_threshold": float(d["materiality_threshold"]),
|
||
"triggered_count": int(d.get("triggered_count") or 0),
|
||
"material_count": int(d.get("material_count") or 0),
|
||
"avg_contribution_lo": _f(d.get("avg_contribution_lo")),
|
||
"avg_abs_contribution_lo": _f(d.get("avg_abs_contribution_lo")),
|
||
"avg_edge_net_when_material": _f(d.get("avg_edge_net_when_material")),
|
||
"unrealized_pnl_est": float(d.get("unrealized_pnl_est") or 0),
|
||
"realized_pnl": float(d.get("realized_pnl") or 0),
|
||
"resolved_count": resolved,
|
||
"win_rate": (wins / resolved) if resolved >= 5 else None,
|
||
"net_positive_count": int(d.get("net_positive_count") or 0),
|
||
"net_negative_count": int(d.get("net_negative_count") or 0),
|
||
}
|
||
return result
|
||
|
||
|
||
async def compute_attribution_from_db(self) -> dict:
|
||
"""Alpha attribution grouped by dominant signal feature.
|
||
|
||
For each Phase 6 trade, the dominant feature is the feat_*_lo with the
|
||
largest absolute value (> 0.0001). Trades are then aggregated per group.
|
||
|
||
Returns {feature_name: {trade_count, avg_edge_net, unrealized_pnl_est,
|
||
realized_pnl, resolved_count, win_rate}}.
|
||
"none" group collects trades where all features are below threshold.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch("""
|
||
WITH dominant_per_trade AS (
|
||
SELECT
|
||
edge_net, net_cost, fee_usdc, closed_at, close_pnl,
|
||
(
|
||
SELECT key
|
||
FROM (VALUES
|
||
('fg', ABS(COALESCE(feat_fg_lo, 0))),
|
||
('mom', ABS(COALESCE(feat_mom_lo, 0))),
|
||
('news', ABS(COALESCE(feat_news_lo, 0))),
|
||
('mfld', ABS(COALESCE(feat_mfld_lo, 0))),
|
||
('btc_dom', ABS(COALESCE(feat_btc_dom_lo, 0)))
|
||
) AS t(key, val)
|
||
WHERE val > 0.0001
|
||
ORDER BY val DESC
|
||
LIMIT 1
|
||
) AS dominant
|
||
FROM trades
|
||
WHERE feat_fg_lo IS NOT NULL
|
||
AND (excluded_from_metrics IS NOT TRUE)
|
||
)
|
||
SELECT
|
||
COALESCE(dominant, 'none') AS dominant_feature,
|
||
COUNT(*) AS trade_count,
|
||
AVG(edge_net) AS avg_edge_net,
|
||
COALESCE(SUM(edge_net * net_cost - fee_usdc)
|
||
FILTER (WHERE closed_at IS NULL
|
||
AND edge_net IS NOT NULL), 0) AS unrealized_pnl_est,
|
||
COALESCE(SUM(close_pnl)
|
||
FILTER (WHERE close_pnl IS NOT NULL), 0) AS realized_pnl,
|
||
COUNT(*) FILTER (WHERE close_pnl IS NOT NULL) AS resolved_count,
|
||
COUNT(*) FILTER (WHERE close_pnl IS NOT NULL AND close_pnl > 0) AS wins
|
||
FROM dominant_per_trade
|
||
GROUP BY dominant_feature
|
||
ORDER BY trade_count DESC
|
||
""")
|
||
|
||
result: dict[str, dict] = {}
|
||
for r in rows:
|
||
d = dict(r)
|
||
feature = d["dominant_feature"]
|
||
resolved = int(d.get("resolved_count") or 0)
|
||
wins = int(d.get("wins") or 0)
|
||
result[feature] = {
|
||
"trade_count": int(d["trade_count"]),
|
||
"avg_edge_net": _f(d.get("avg_edge_net")),
|
||
"unrealized_pnl_est": float(d.get("unrealized_pnl_est") or 0),
|
||
"realized_pnl": float(d.get("realized_pnl") or 0),
|
||
"resolved_count": resolved,
|
||
"win_rate": (wins / resolved) if resolved >= 5 else None,
|
||
}
|
||
return result
|
||
|
||
|
||
async def save_manifold_audit(
|
||
self,
|
||
audit_id: str,
|
||
poly_market_id: str,
|
||
poly_question: str,
|
||
search_query: str,
|
||
mfld_market_id: Optional[str],
|
||
mfld_market_title: Optional[str],
|
||
mfld_market_url: Optional[str],
|
||
prob_raw: Optional[float],
|
||
prob_final: Optional[float],
|
||
inverted: bool,
|
||
match_score: Optional[float],
|
||
match_reason: Optional[str],
|
||
match_status: str,
|
||
poly_outcome_type: Optional[str] = None,
|
||
mfld_outcome_type: Optional[str] = None,
|
||
matcher_version: Optional[str] = None,
|
||
) -> None:
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute("""
|
||
INSERT INTO manifold_match_audit (
|
||
id, poly_market_id, poly_question, search_query,
|
||
mfld_market_id, mfld_market_title, mfld_market_url,
|
||
prob_raw, prob_final, inverted,
|
||
match_score, match_reason, match_status, used_in_trade,
|
||
poly_outcome_type, mfld_outcome_type, matcher_version
|
||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,FALSE,$14,$15,$16)
|
||
""",
|
||
audit_id, poly_market_id, poly_question, search_query,
|
||
mfld_market_id, mfld_market_title, mfld_market_url,
|
||
prob_raw, prob_final, inverted,
|
||
match_score, match_reason, match_status,
|
||
poly_outcome_type, mfld_outcome_type, matcher_version,
|
||
)
|
||
|
||
async def get_manifold_cooldown(self, poly_market_id: str) -> Optional[dict]:
|
||
"""Return the cooldown row for a market, or None if no cooldown is recorded.
|
||
|
||
The caller decides whether the cooldown is still active by comparing
|
||
now() against retry_after (see bayesian.evaluate()).
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
row = await conn.fetchrow(
|
||
"SELECT poly_market_id, last_evaluated_at, last_status, "
|
||
"retry_after, cooldown_reason FROM manifold_eval_cooldown "
|
||
"WHERE poly_market_id = $1",
|
||
poly_market_id,
|
||
)
|
||
return dict(row) if row else None
|
||
|
||
async def upsert_manifold_cooldown(
|
||
self,
|
||
poly_market_id: str,
|
||
last_status: str,
|
||
retry_after,
|
||
cooldown_reason: Optional[str] = None,
|
||
) -> None:
|
||
"""Insert or refresh the cooldown for a market after a real evaluation.
|
||
|
||
last_evaluated_at is stamped server-side with now(); retry_after is the
|
||
caller-computed earliest re-evaluation time.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute("""
|
||
INSERT INTO manifold_eval_cooldown (
|
||
poly_market_id, last_evaluated_at, last_status,
|
||
retry_after, cooldown_reason
|
||
) VALUES ($1, now(), $2, $3, $4)
|
||
ON CONFLICT (poly_market_id) DO UPDATE SET
|
||
last_evaluated_at = now(),
|
||
last_status = EXCLUDED.last_status,
|
||
retry_after = EXCLUDED.retry_after,
|
||
cooldown_reason = EXCLUDED.cooldown_reason
|
||
""", poly_market_id, last_status, retry_after, cooldown_reason)
|
||
|
||
# ── Replay R0: snapshot recorder ─────────────────────────────────────────
|
||
|
||
async def save_ext_snapshot(self, cycle_ts, ext) -> None:
|
||
"""Persist the ExternalSignals snapshot for one cycle (Replay R0)."""
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute("""
|
||
INSERT INTO ext_snapshots (
|
||
cycle_ts, btc_price, btc_change_24h, eth_price, eth_change_24h,
|
||
btc_dominance, fear_greed_index, fear_greed_label,
|
||
total_market_cap_change, valid
|
||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
|
||
ON CONFLICT (cycle_ts) DO NOTHING
|
||
""",
|
||
cycle_ts, ext.btc_price, ext.btc_change_24h,
|
||
ext.eth_price, ext.eth_change_24h, ext.btc_dominance,
|
||
ext.fear_greed_index, ext.fear_greed_label,
|
||
ext.total_market_cap_change, ext.valid,
|
||
)
|
||
|
||
async def upsert_markets(self, markets: list) -> None:
|
||
"""Refresh market metadata (Replay R0) — replay rebuilds Market from here."""
|
||
rows = [
|
||
(m.id, m.condition_id, m.question, m.category, m.end_date, m.active)
|
||
for m in markets
|
||
]
|
||
async with self._pool.acquire() as conn:
|
||
await conn.executemany("""
|
||
INSERT INTO markets (id, condition_id, question, category, end_date, active, last_seen)
|
||
VALUES ($1,$2,$3,$4,$5,$6, now())
|
||
ON CONFLICT (id) DO UPDATE SET
|
||
condition_id = EXCLUDED.condition_id,
|
||
question = EXCLUDED.question,
|
||
category = EXCLUDED.category,
|
||
end_date = EXCLUDED.end_date,
|
||
active = EXCLUDED.active,
|
||
last_seen = now()
|
||
""", rows)
|
||
|
||
async def save_signal_records(self, cycle_ts, records: list[dict]) -> None:
|
||
"""Batch-insert one cycle's decision records into signals (Replay R0)."""
|
||
if not records:
|
||
return
|
||
rows = [
|
||
(
|
||
r["market_id"], cycle_ts, cycle_ts,
|
||
r["polymarket_price"], r["category"], r["volume_24h"],
|
||
r["skip_reason"], r["family_key"],
|
||
r["prior_prob"], r["estimated_prob"], r["raw_final_prob"],
|
||
r["edge_gross"], r["edge_net"], r["regime_min_edge"],
|
||
r["days_to_resolution"], r["confidence"], r["direction"],
|
||
r["passed_gross"], r["passed_net"],
|
||
r["news_sentiment"], r["news_budget_skipped"],
|
||
r["guardrail_applied"], r["guardrail_changed_decision"],
|
||
r["feat_fg_lo"], r["feat_mom_lo"], r["feat_news_lo"],
|
||
r["feat_mfld_lo"], r["feat_btc_dom_lo"],
|
||
r["edge_gross"], # legacy `edge` column mirrors edge_gross
|
||
r["acted_on"],
|
||
)
|
||
for r in records
|
||
]
|
||
async with self._pool.acquire() as conn:
|
||
await conn.executemany("""
|
||
INSERT INTO signals (
|
||
market_id, timestamp, cycle_ts,
|
||
polymarket_price, category, volume_24h,
|
||
skip_reason, family_key,
|
||
prior_prob, estimated_prob, raw_final_prob,
|
||
edge_gross, edge_net, regime_min_edge,
|
||
days_to_resolution, confidence, direction,
|
||
passed_gross, passed_net,
|
||
news_sentiment, news_budget_skipped,
|
||
guardrail_applied, guardrail_changed_decision,
|
||
feat_fg_lo, feat_mom_lo, feat_news_lo,
|
||
feat_mfld_lo, feat_btc_dom_lo,
|
||
edge, acted_on
|
||
) VALUES (
|
||
$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,
|
||
$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30
|
||
)
|
||
""", rows)
|
||
|
||
async def prune_signal_records(self, retention_days: int) -> int:
|
||
"""Delete archive rows older than retention_days; returns rows deleted."""
|
||
async with self._pool.acquire() as conn:
|
||
result = await conn.execute(
|
||
"DELETE FROM signals WHERE timestamp < now() - ($1 || ' days')::interval",
|
||
str(retention_days),
|
||
)
|
||
await conn.execute(
|
||
"DELETE FROM ext_snapshots WHERE cycle_ts < now() - ($1 || ' days')::interval",
|
||
str(retention_days),
|
||
)
|
||
try:
|
||
return int(result.split()[-1])
|
||
except (ValueError, IndexError):
|
||
return 0
|
||
|
||
async def mark_manifold_audit_used(self, audit_id: str) -> None:
|
||
async with self._pool.acquire() as conn:
|
||
await conn.execute(
|
||
"UPDATE manifold_match_audit SET used_in_trade = TRUE WHERE id = $1",
|
||
audit_id,
|
||
)
|
||
|
||
async def get_manifold_matches(self, limit: int = 50) -> dict:
|
||
"""Manifold match audit, with summary split by matcher version.
|
||
|
||
The summary separates the current matcher (MANIFOLD_MATCHER_VERSION) from
|
||
all-time totals and from legacy pre-outcome-guard records, whose accepted
|
||
matches would now be rejected by the outcome-compatibility guard and so
|
||
must not be conflated with current-version stats.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
current = await conn.fetchrow("""
|
||
SELECT
|
||
COUNT(*) FILTER (WHERE match_status = 'accepted') AS total_accepted,
|
||
COUNT(*) FILTER (WHERE match_status = 'rejected') AS total_rejected,
|
||
COUNT(*) FILTER (WHERE match_status = 'no_results') AS total_no_results,
|
||
AVG(match_score) FILTER (WHERE match_status = 'accepted') AS avg_match_score,
|
||
COUNT(*) FILTER (WHERE used_in_trade = TRUE) AS used_in_trade
|
||
FROM manifold_match_audit
|
||
WHERE matcher_version = $1
|
||
""", MANIFOLD_MATCHER_VERSION)
|
||
all_time = await conn.fetchrow("""
|
||
SELECT
|
||
COUNT(*) FILTER (WHERE match_status = 'accepted') AS total_accepted,
|
||
COUNT(*) FILTER (WHERE match_status = 'rejected') AS total_rejected,
|
||
COUNT(*) FILTER (WHERE match_status = 'no_results') AS total_no_results
|
||
FROM manifold_match_audit
|
||
""")
|
||
legacy = await conn.fetchrow("""
|
||
SELECT COUNT(*) AS accepted_without_outcome_type
|
||
FROM manifold_match_audit
|
||
WHERE matcher_version = 'legacy_pre_outcome_guard'
|
||
AND match_status = 'accepted'
|
||
""")
|
||
unique_markets = await conn.fetchrow("""
|
||
SELECT
|
||
COUNT(DISTINCT poly_market_id) AS evaluated,
|
||
COUNT(DISTINCT poly_market_id) FILTER (
|
||
WHERE match_status = 'accepted'
|
||
AND matcher_version = $1
|
||
) AS accepted
|
||
FROM manifold_match_audit
|
||
""", MANIFOLD_MATCHER_VERSION)
|
||
mfld_dominated = await conn.fetchrow("""
|
||
SELECT COUNT(*) AS cnt FROM trades
|
||
WHERE (excluded_from_metrics IS NOT TRUE)
|
||
AND mfld_match_status = 'accepted'
|
||
AND feat_mfld_lo IS NOT NULL
|
||
AND ABS(feat_mfld_lo) > 0.0001
|
||
AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_fg_lo, 0))
|
||
AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_mom_lo, 0))
|
||
AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_news_lo, 0))
|
||
AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_btc_dom_lo, 0))
|
||
""")
|
||
rows = await conn.fetch(
|
||
"SELECT * FROM manifold_match_audit ORDER BY timestamp DESC LIMIT $1",
|
||
limit,
|
||
)
|
||
return {
|
||
"summary": {
|
||
"current_version": {
|
||
"version": MANIFOLD_MATCHER_VERSION,
|
||
"total_accepted": int(current["total_accepted"] or 0),
|
||
"total_rejected": int(current["total_rejected"] or 0),
|
||
"total_no_results": int(current["total_no_results"] or 0),
|
||
"avg_match_score": _f(current["avg_match_score"]),
|
||
"used_in_trade": int(current["used_in_trade"] or 0),
|
||
},
|
||
"all_time": {
|
||
"total_accepted": int(all_time["total_accepted"] or 0),
|
||
"total_rejected": int(all_time["total_rejected"] or 0),
|
||
"total_no_results": int(all_time["total_no_results"] or 0),
|
||
},
|
||
"legacy": {
|
||
"accepted_without_outcome_type":
|
||
int(legacy["accepted_without_outcome_type"] or 0),
|
||
},
|
||
"trades_dominated_by_mfld": int(mfld_dominated["cnt"] or 0),
|
||
"unique_markets": {
|
||
"evaluated": int(unique_markets["evaluated"] or 0),
|
||
"accepted": int(unique_markets["accepted"] or 0),
|
||
"coverage_rate": (
|
||
float(unique_markets["accepted"]) / float(unique_markets["evaluated"])
|
||
if unique_markets["evaluated"] else None
|
||
),
|
||
},
|
||
},
|
||
"recent_matches": [dict(r) for r in rows],
|
||
}
|
||
|
||
async def get_manifold_coverage_by_category(self) -> dict:
|
||
"""Manifold coverage by semantic market category, counted by UNIQUE market.
|
||
|
||
Base table is manifold_match_audit filtered to the current matcher
|
||
(v3_outcome_guard). Each poly_market_id is collapsed to one row first, so
|
||
a market is counted once regardless of how many audit attempts or trades it
|
||
has — this measures coverage, not retry volume.
|
||
|
||
Category is inferred from the market's trade family_key when available, else
|
||
from poly_question (LEFT JOIN: audited markets that never produced a trade
|
||
are kept). Buckets (accepted/rejected/no_results) are not mutually exclusive
|
||
at the market level — a market that was no_results then later rejected counts
|
||
in both — matching COUNT(DISTINCT CASE WHEN status=...) semantics.
|
||
"""
|
||
async with self._pool.acquire() as conn:
|
||
rows = await conn.fetch("""
|
||
WITH audit AS (
|
||
SELECT
|
||
poly_market_id,
|
||
MAX(poly_question) AS poly_question,
|
||
bool_or(match_status = 'accepted') AS has_accepted,
|
||
bool_or(match_status = 'rejected') AS has_rejected,
|
||
bool_or(match_status = 'no_results') AS has_no_results
|
||
FROM manifold_match_audit
|
||
WHERE matcher_version = 'v3_outcome_guard'
|
||
GROUP BY poly_market_id
|
||
),
|
||
fam AS (
|
||
SELECT market_id, MAX(family_key) AS family_key
|
||
FROM trades
|
||
GROUP BY market_id
|
||
),
|
||
categorized AS (
|
||
SELECT
|
||
a.has_accepted, a.has_rejected, a.has_no_results,
|
||
CASE
|
||
WHEN f.family_key ILIKE '%gubernatorial%' THEN 'gubernatorial'
|
||
WHEN f.family_key ILIKE '%mayoral%' THEN 'mayoral'
|
||
WHEN f.family_key ILIKE '%senate%' THEN 'senate'
|
||
WHEN f.family_key ILIKE '%republican%' THEN 'primary-republican'
|
||
WHEN f.family_key ILIKE '%democrat%' THEN 'primary-democrat'
|
||
WHEN f.family_key ILIKE '%openai%'
|
||
OR f.family_key ILIKE '%nvidia%'
|
||
OR f.family_key ILIKE '%anthropic%' THEN 'big-tech'
|
||
-- family_key NULL or unmatched → infer from question
|
||
WHEN a.poly_question ILIKE '%governor%'
|
||
OR a.poly_question ILIKE '%gubernatorial%' THEN 'gubernatorial'
|
||
WHEN a.poly_question ILIKE '%mayor%'
|
||
OR a.poly_question ILIKE '%mayoral%' THEN 'mayoral'
|
||
WHEN a.poly_question ILIKE '%senate%' THEN 'senate'
|
||
WHEN a.poly_question ILIKE '%republican primary%' THEN 'primary-republican'
|
||
WHEN a.poly_question ILIKE '%democratic primary%'
|
||
OR a.poly_question ILIKE '%democrat primary%' THEN 'primary-democrat'
|
||
WHEN a.poly_question ILIKE '%openai%'
|
||
OR a.poly_question ILIKE '%nvidia%'
|
||
OR a.poly_question ILIKE '%anthropic%' THEN 'big-tech'
|
||
WHEN a.poly_question ILIKE '%russia%'
|
||
OR a.poly_question ILIKE '%ukraine%'
|
||
OR a.poly_question ILIKE '%israel%'
|
||
OR a.poly_question ILIKE '%ceasefire%'
|
||
OR a.poly_question ILIKE '%military%' THEN 'geopolitics'
|
||
ELSE 'other'
|
||
END AS category
|
||
FROM audit a
|
||
LEFT JOIN fam f ON f.market_id = a.poly_market_id
|
||
)
|
||
SELECT
|
||
category,
|
||
COUNT(*) AS unique_evaluated,
|
||
COUNT(*) FILTER (WHERE has_accepted) AS unique_accepted,
|
||
COUNT(*) FILTER (WHERE has_rejected) AS unique_rejected,
|
||
COUNT(*) FILTER (WHERE has_no_results) AS unique_no_results
|
||
FROM categorized
|
||
GROUP BY category
|
||
ORDER BY unique_evaluated DESC
|
||
""")
|
||
|
||
coverage_by_category = []
|
||
total_evaluated = 0
|
||
total_accepted = 0
|
||
categories_with_coverage = 0
|
||
for r in rows:
|
||
evaluated = int(r["unique_evaluated"] or 0)
|
||
accepted = int(r["unique_accepted"] or 0)
|
||
total_evaluated += evaluated
|
||
total_accepted += accepted
|
||
if accepted > 0:
|
||
categories_with_coverage += 1
|
||
coverage_by_category.append({
|
||
"category": r["category"],
|
||
"unique_evaluated": evaluated,
|
||
"unique_accepted": accepted,
|
||
"unique_rejected": int(r["unique_rejected"] or 0),
|
||
"unique_no_results": int(r["unique_no_results"] or 0),
|
||
"coverage_rate": (accepted / evaluated) if evaluated else None,
|
||
})
|
||
|
||
return {
|
||
"coverage_by_category": coverage_by_category,
|
||
"summary": {
|
||
"total_unique_evaluated": total_evaluated,
|
||
"total_unique_accepted": total_accepted,
|
||
"overall_coverage_rate": (total_accepted / total_evaluated) if total_evaluated else None,
|
||
"categories_with_coverage": categories_with_coverage,
|
||
},
|
||
}
|
||
|
||
|
||
def _f(v) -> Optional[float]:
|
||
"""None-safe float cast for asyncpg Decimal/None values."""
|
||
return float(v) if v is not None else None
|