PythagorasGoal/tests/test_collect_chain.py

"""Test del collettore catena opzioni (successore di cerbero-bite).

Il collettore esiste per NON ripetere tre difetti misurati su bite il 2026-07-30. I test
sorvegliano esattamente quei tre, perche' sono difetti SILENZIOSI: nessuno di loro fa fallire un
giro, tutti e tre corrompono una serie irrecuperabile.

  1. una chiamata fallita non deve diventare una riga che sembra un dato;
  2. "book vuoto" e "chiamata fallita" non devono collassare sullo stesso valore;
  3. il pacing deve esserci davvero (bite si auto-saturava il rate limit per-IP).

Nessun test tocca la rete.
"""
from __future__ import annotations

import sys
import time
from datetime import UTC, datetime
from pathlib import Path

import pandas as pd
import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts" / "live"))

from collect_chain import Budget, _depth_top3, snapshot_row, write  # noqa: E402

TS = datetime(2026, 7, 30, 20, 0, tzinfo=UTC)
INST = {
    "instrument_name": "ETH-31JUL26-1750-P", "base_currency": "ETH",
    "strike": 1750.0, "option_type": "put",
    "expiration_timestamp": int(pd.Timestamp("2026-07-31T08:00:00Z").timestamp() * 1000),
}


def _ob(bid=0.016, ask=0.0175, bids=((0.016, 10.0),), asks=((0.0175, 12.0),)):
    return {
        "best_bid_price": bid, "best_ask_price": ask, "mark_iv": 52.3,
        "greeks": {"delta": -0.28, "gamma": 0.001, "theta": -1.2, "vega": 0.9},
        "open_interest": 500.0, "stats": {"volume": 31.0},
        "bids": [list(b) for b in bids], "asks": [list(a) for a in asks],
        "underlying_price": 1923.4, "index_price": 1922.7,
    }


# ------------------------------------------------- 1. una chiamata fallita non e' un dato

def test_chiamata_fallita_marcata_error_e_senza_quota():
    r = snapshot_row(INST, None, TS)
    assert r["quote_status"] == "error"
    assert r["bid"] is None and r["ask"] is None and r["iv"] is None
    assert r["instrument_name"] == INST["instrument_name"], (
        "la riga si scrive lo stesso: cio' che cambia e' che dichiara di non avere il dato"
    )


def test_profondita_e_NULL_su_errore_mai_zero():
    """Il difetto di bite: depth 0 su chiamata fallita = indistinguibile da 'book vuoto'.
    Qui su errore la profondita' non esiste, e un book davvero vuoto vale 0."""
    assert snapshot_row(INST, None, TS)["book_depth_top3"] is None
    vuoto = snapshot_row(INST, _ob(bid=0.0, ask=0.0, bids=(), asks=()), TS)
    assert vuoto["book_depth_top3"] == 0.0
    assert vuoto["quote_status"] == "no_quote"


def test_book_vuoto_e_chiamata_fallita_sono_stati_DIVERSI():
    """Controllo positivo dell'idea stessa del collettore: se questi due collassassero, la
    diagnostica del 29/07 (50% di quote perse per rate-limit) sarebbe di nuovo impossibile."""
    fallita = snapshot_row(INST, None, TS)["quote_status"]
    vuoto = snapshot_row(INST, _ob(bid=0.0, ask=0.0, bids=(), asks=()), TS)["quote_status"]
    assert fallita == "error" and vuoto == "no_quote" and fallita != vuoto


def test_un_solo_lato_quotato_e_comunque_ok():
    """Un book con solo il bid e' un mercato reale (illiquido), non un guasto."""
    r = snapshot_row(INST, _ob(ask=0.0, asks=()), TS)
    assert r["quote_status"] == "ok"
    assert r["bid"] is not None and r["ask"] is None
    assert r["mid"] is None, "senza un lato il mid non esiste: non si inventa"


def test_campi_valorizzati_quando_la_quota_c_e():
    r = snapshot_row(INST, _ob(), TS)
    assert r["quote_status"] == "ok"
    assert r["mid"] == pytest.approx((0.016 + 0.0175) / 2)
    assert r["delta"] == pytest.approx(-0.28) and r["iv"] == pytest.approx(52.3)
    assert r["book_depth_top3"] == pytest.approx(22.0)
    assert r["underlying_price"] == pytest.approx(1923.4)
    assert r["option_type"] == "P"


def test_depth_top3_somma_solo_i_primi_tre_livelli():
    assert _depth_top3([[1, 5], [2, 5], [3, 5], [4, 100]]) == pytest.approx(15.0)
    assert _depth_top3([]) == 0.0
    assert _depth_top3(None) is None


# ------------------------------------------------- 3. il pacing deve esistere

def test_il_budget_impone_davvero_una_cadenza():
    """Bite non aveva pacing: ~44 chiamate/s per 26s. Qui la cadenza e' una proprieta' misurabile."""
    b = Budget(rps=20.0)
    t0 = time.monotonic()
    for _ in range(6):
        b.wait()
    dur = time.monotonic() - t0
    assert dur >= 5 / 20.0 * 0.8, f"6 chiamate a 20/s non possono durare {dur:.3f}s"


def test_il_budget_conta_i_rate_limit_e_gli_errori():
    b = Budget(rps=100.0)
    b.note_error("boom")
    b.rate_limited += 2
    assert b.errors == 1 and b.rate_limited == 2 and b.err_samples == ["boom"]


# ------------------------------------------------- scrittura

def test_write_deduplica_e_tiene_l_ultima_osservazione(tmp_path, monkeypatch):
    import collect_chain as CC
    monkeypatch.setattr(CC, "STORE", tmp_path)
    a = pd.DataFrame([snapshot_row(INST, _ob(bid=0.010), TS)])
    b = pd.DataFrame([snapshot_row(INST, _ob(bid=0.020), TS)])   # stesso (ts, strumento)
    CC.write(a, TS)
    p = CC.write(b, TS)
    out = pd.read_parquet(p)
    assert len(out) == 1, "stesso istante e stesso strumento = una riga sola"
    assert out.iloc[0]["bid"] == pytest.approx(0.020), "vince l'osservazione piu' recente"


def test_write_separa_i_giorni(tmp_path, monkeypatch):
    import collect_chain as CC
    monkeypatch.setattr(CC, "STORE", tmp_path)
    CC.write(pd.DataFrame([snapshot_row(INST, _ob(), TS)]), TS)
    dopo = TS.replace(day=31)
    CC.write(pd.DataFrame([snapshot_row(INST, _ob(), dopo)]), dopo)
    assert sorted(p.name for p in tmp_path.glob("*.parquet")) == \
        ["2026-07-30.parquet", "2026-07-31.parquet"]