Adriano/Cerbero-Bite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(analysis): audit_option_chain — coverage, quote stats, bid>ask, IV null

Browse Source 
Implementa la funzione dichiarata in __all__ ma mancante. Helper _pct
usa statistics.quantiles(method="inclusive") con fallback per len<=1.
Niente check su book_depth_top3: per design è NULL sugli snapshot
(popolato solo da entry_cycle per gli strike candidati al picker).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-05-16 19:32:52 +00:00
parent bb2ca425a7
commit 06ce2c0eb2
2 changed files with 279 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cerbero_bite/analysis/data_audit.py
+101
View File
@@ -220,3 +220,104 @@ def audit_market_snapshots(
max_fetch_ok_zero_streak=max_streak,
null_rate_by_column=null_rates,
)
def _pct(values: list[int], q: int) -> int:
"""Integer percentile via `statistics.quantiles` (inclusive).
Returns 0 on empty input; the single value on len==1. ``q`` is in 1..100.
"""
if not values:
return 0
if len(values) == 1:
return values[0]
sorted_vals = sorted(values)
if q >= 100:
return sorted_vals[-1]
cuts = statistics.quantiles(sorted_vals, n=100, method="inclusive")
return int(round(cuts[q - 1]))
def audit_option_chain(
conn: sqlite3.Connection,
*,
asset: str,
since: datetime,
until: datetime,
) -> ChainAuditReport:
"""Compute the option_chain_snapshots audit report in [since, until)."""
expected = _expected_ticks(since, until)
snap_counts = conn.execute(
"SELECT timestamp, COUNT(*) AS n FROM option_chain_snapshots "
"WHERE asset = ? AND timestamp >= ? AND timestamp < ? "
"GROUP BY timestamp ORDER BY timestamp",
(asset, since.isoformat(), until.isoformat()),
).fetchall()
actual = len(snap_counts)
coverage = (
(Decimal(actual) / Decimal(expected) * Decimal("100")).quantize(
Decimal("0.01")
)
if expected > 0
else Decimal("0")
)
quotes_per_snap = [r["n"] for r in snap_counts]
median_q = _pct(quotes_per_snap, 50)
p10_q = _pct(quotes_per_snap, 10)
p90_q = _pct(quotes_per_snap, 90)
quote_rows = conn.execute(
"SELECT bid, ask, iv FROM option_chain_snapshots "
"WHERE asset = ? AND timestamp >= ? AND timestamp < ?",
(asset, since.isoformat(), until.isoformat()),
).fetchall()
total_quotes = len(quote_rows)
if total_quotes == 0:
return ChainAuditReport(
asset=asset,
since=since,
until=until,
expected_snapshots=expected,
actual_snapshots=actual,
coverage_pct=coverage,
quotes_per_snap_median=median_q,
quotes_per_snap_p10=p10_q,
quotes_per_snap_p90=p90_q,
)
bid_gt_ask = 0
iv_null = 0
for r in quote_rows:
bid_s, ask_s, iv_s = r["bid"], r["ask"], r["iv"]
if bid_s is not None and ask_s is not None:
try:
if Decimal(bid_s) > Decimal(ask_s):
bid_gt_ask += 1
except (ValueError, ArithmeticError):
pass
if iv_s is None:
iv_null += 1
else:
try:
Decimal(iv_s)
except (ValueError, ArithmeticError):
iv_null += 1
iv_null_pct = (
Decimal(iv_null) / Decimal(total_quotes) * Decimal("100")
).quantize(Decimal("0.01"))
return ChainAuditReport(
asset=asset,
since=since,
until=until,
expected_snapshots=expected,
actual_snapshots=actual,
coverage_pct=coverage,
quotes_per_snap_median=median_q,
quotes_per_snap_p10=p10_q,
quotes_per_snap_p90=p90_q,
bid_gt_ask_count=bid_gt_ask,
iv_null_count=iv_null,
iv_null_pct=iv_null_pct,
)
tests/unit/test_data_audit.py
+178
View File
@@ -8,13 +8,16 @@ from decimal import Decimal
from pathlib import Path
from cerbero_bite.analysis.data_audit import (
ChainAuditReport,
MarketAuditReport,
audit_market_snapshots,
audit_option_chain,
)
from cerbero_bite.analysis.data_audit import ( # noqa: PLC2701
_detect_gaps,
_expected_ticks,
_max_zero_streak,
_pct,
)
from cerbero_bite.state import connect, run_migrations, transaction
@@ -213,3 +216,178 @@ def test_audit_market_null_rate_per_column(tmp_path: Path) -> None:
conn.close()
assert report.null_rate_by_column["dealer_net_gamma"] == Decimal("0.25")
assert report.null_rate_by_column["spot"] == Decimal("0")
def test_pct_empty_returns_zero() -> None:
assert _pct([], 50) == 0
def test_pct_single_value_returns_that_value() -> None:
assert _pct([42], 50) == 42
assert _pct([42], 10) == 42
assert _pct([42], 90) == 42
def test_pct_median_odd_count() -> None:
assert _pct([10, 20, 30, 40, 50], 50) == 30
def test_pct_p10_p90_on_100_values() -> None:
# inclusive method on [1..100]:
# p10 interpolates to 10.9 → 11; p90 interpolates to 90.1 → 90.
values = list(range(1, 101))
assert _pct(values, 10) == 11
assert _pct(values, 90) == 90
def _seed_chain_row(
conn: sqlite3.Connection,
*,
asset: str,
ts: datetime,
instrument: str,
strike: str = "3000",
expiry: datetime | None = None,
option_type: str = "C",
bid: str | None = "0.01",
ask: str | None = "0.02",
iv: str | None = "0.7",
) -> None:
if expiry is None:
expiry = datetime(2026, 6, 12, 8, 0, tzinfo=UTC)
conn.execute(
"INSERT INTO option_chain_snapshots(timestamp, asset, instrument_name, "
"strike, expiry, option_type, bid, ask, iv) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(
ts.isoformat(),
asset,
instrument,
strike,
expiry.isoformat(),
option_type,
bid,
ask,
iv,
),
)
def test_audit_chain_full_coverage(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
with transaction(conn):
for minute in (0, 15, 30, 45):
ts = since.replace(minute=minute)
for i in range(10):
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument=f"ETH-EXP-{i}-C",
strike=str(3000 + i * 50),
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert isinstance(report, ChainAuditReport)
assert report.asset == "ETH"
assert report.expected_snapshots == 4
assert report.actual_snapshots == 4
assert report.coverage_pct == Decimal("100")
assert report.quotes_per_snap_median == 10
assert report.quotes_per_snap_p10 == 10
assert report.quotes_per_snap_p90 == 10
assert report.bid_gt_ask_count == 0
assert report.iv_null_count == 0
assert report.iv_null_pct == Decimal("0")
def test_audit_chain_detects_bid_gt_ask_and_iv_null(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 12, 30, tzinfo=UTC)
try:
with transaction(conn):
ts = since.replace(minute=0)
# Crossed BBO: bid > ask
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-A",
bid="0.10",
ask="0.05",
)
# Missing IV
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-B",
strike="3050",
iv=None,
)
# Normal row (control)
_seed_chain_row(
conn,
asset="ETH",
ts=ts,
instrument="ETH-C",
strike="3100",
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.bid_gt_ask_count == 1
assert report.iv_null_count == 1
# 1 of 3 quotes → 33.33%
assert report.iv_null_pct == Decimal("33.33")
def test_audit_chain_missing_snapshots(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
with transaction(conn):
# Only 2 of the expected 4 ticks have data
for minute in (0, 15):
_seed_chain_row(
conn,
asset="ETH",
ts=since.replace(minute=minute),
instrument=f"ETH-{minute}",
)
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.expected_snapshots == 4
assert report.actual_snapshots == 2
assert report.coverage_pct == Decimal("50")
def test_audit_chain_empty_window_returns_zero_coverage(tmp_path: Path) -> None:
conn = _make_conn(tmp_path)
since = datetime(2026, 5, 12, 12, 0, tzinfo=UTC)
until = datetime(2026, 5, 12, 13, 0, tzinfo=UTC)
try:
report = audit_option_chain(
conn, asset="ETH", since=since, until=until
)
finally:
conn.close()
assert report.expected_snapshots == 4
assert report.actual_snapshots == 0
assert report.coverage_pct == Decimal("0")
assert report.quotes_per_snap_median == 0
assert report.bid_gt_ask_count == 0
assert report.iv_null_count == 0