feat(state): Repository.iv_rv_history + dvol_lookback per gate adaptive

Due nuovi metodi che leggono market_snapshots filtrando NULL e
fetch_ok=0. iv_rv_history limita a max_days; dvol_lookback trova
il tick più vicino a un istante con tolerance configurabile.

Tests: ordered ASC, asset filter, NULL skip, fetch_ok=0 skip,
lookback closest, gap returns None.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-05-08 22:53:19 +00:00
parent d36cdff609
commit 395191ea13
2 changed files with 197 additions and 1 deletions
+59 -1
View File
@@ -13,7 +13,7 @@ Decimals are stored as TEXT to preserve precision (see
from __future__ import annotations
import sqlite3
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Any
from uuid import UUID
@@ -408,6 +408,64 @@ class Repository:
).fetchall()
return [_row_to_market_snapshot(r) for r in rows]
def iv_rv_history(
self,
conn: sqlite3.Connection,
*,
asset: str,
max_days: int,
) -> list[Decimal]:
"""Lista IV-RV ordinata ASC sull'intervallo `[now-max_days, now]`.
Esclude righe con ``fetch_ok=0`` o ``iv_minus_rv IS NULL``.
Usata dal validator quando il gate adattivo è abilitato.
"""
rows = conn.execute(
"SELECT iv_minus_rv FROM market_snapshots "
"WHERE asset = ? "
" AND fetch_ok = 1 "
" AND iv_minus_rv IS NOT NULL "
" AND timestamp >= datetime('now', ?) "
"ORDER BY timestamp ASC",
(asset, f"-{int(max_days)} days"),
).fetchall()
return [Decimal(str(r["iv_minus_rv"])) for r in rows]
def dvol_lookback(
self,
conn: sqlite3.Connection,
*,
asset: str,
reference: datetime,
tolerance_minutes: int = 15,
) -> Decimal | None:
"""DVOL al tick più vicino a `reference`, entro ±tolerance_minutes.
Ritorna ``None`` se non esiste un tick valido (``fetch_ok=1``,
``dvol IS NOT NULL``) entro la tolerance. Usato dal Vol-of-Vol
guard per stimare DVOL N ore fa.
"""
ref_lo = (reference - timedelta(minutes=tolerance_minutes)).astimezone(UTC)
ref_hi = (reference + timedelta(minutes=tolerance_minutes)).astimezone(UTC)
row = conn.execute(
"SELECT dvol, timestamp FROM market_snapshots "
"WHERE asset = ? "
" AND fetch_ok = 1 "
" AND dvol IS NOT NULL "
" AND timestamp >= ? "
" AND timestamp <= ? "
"ORDER BY ABS(julianday(timestamp) - julianday(?)) ASC LIMIT 1",
(
asset,
_enc_dt(ref_lo),
_enc_dt(ref_hi),
_enc_dt(reference),
),
).fetchone()
if row is None:
return None
return Decimal(str(row["dvol"]))
# ------------------------------------------------------------------
# option_chain_snapshots
# ------------------------------------------------------------------