Adriano
1a7f670966
src/live/giri.registra_giro: una riga jsonl {ts ms, campi} dopo una misura completata,
mai un'eccezione. edge_watch, fee_watch, scale_watch, vrp_f_watch, venue_news (cron_daily)
e venue_watch (cron_book, orario) scrivono data/<dir>/runs.jsonl; niente riga senza misura.
monitor_health: sei MonitorSpec (30h i giornalieri, 3h venue_watch), open_labeled=False.
Stessa riparazione di cc01 (#10, #12), generalizzata. Primo giro notturno: NUOVO, non allerta.
Test: incrocio P1 percorso/RUNS, misura -> una riga, non-misura -> nessuna. Suite: 1098 passed.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
271 lines
11 KiB
Python
271 lines
11 KiB
Python
"""registra_giro nei sei sorveglianti generalizzati da cc01_regime_watch (2026-09-21, #16).
|
|
|
|
PERCHE' ESISTE. `cc01_regime_watch` e' morto 14 notti di fila (#10) senza che nessuno lo vedesse,
|
|
perche' non scriveva niente da sorvegliare (`monitor_health` guarda solo le serie che i monitor
|
|
scrivono). La riparazione di cc01 (`data/cc01_regime/runs.jsonl`, commit 31f5915,
|
|
tests/test_cc01_regime_watch.py) e' il modello: qui si verifica che sia stata applicata agli
|
|
altri sei sorveglianti giornalieri/orari, con lo stesso incrocio P1 con `monitor_health.MONITORS`
|
|
e lo stesso doppio ramo (misura -> una riga, non-misura -> nessuna riga).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
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 src.live.monitor_health import MONITORS # noqa: E402
|
|
|
|
import edge_watch as EW # noqa: E402
|
|
import fee_watch as FW # noqa: E402
|
|
import scale_watch as SW # noqa: E402
|
|
|
|
|
|
def _carica(nome: str):
|
|
sp = importlib.util.spec_from_file_location(nome, ROOT / "scripts" / "live" / f"{nome}.py")
|
|
m = importlib.util.module_from_spec(sp)
|
|
sp.loader.exec_module(m)
|
|
return m
|
|
|
|
|
|
VF = _carica("vrp_f_watch")
|
|
VN = _carica("venue_news")
|
|
VW = _carica("venue_watch")
|
|
|
|
|
|
def _righe(path: Path) -> list[dict]:
|
|
import json
|
|
if not path.exists():
|
|
return []
|
|
return [json.loads(r) for r in path.read_text().splitlines() if r.strip()]
|
|
|
|
|
|
def _spec(nome: str):
|
|
s = next((m for m in MONITORS if m.name == nome), None)
|
|
assert s is not None, f"{nome} non e' in monitor_health.MONITORS"
|
|
return s
|
|
|
|
|
|
# ===========================================================================
|
|
# P1 — il bersaglio della sorveglianza si deriva dal codice sorvegliato
|
|
# ===========================================================================
|
|
@pytest.mark.parametrize("nome,modulo,cadence_h,max_age_h", [
|
|
("edge_watch", EW, 24.0, 30.0),
|
|
("fee_watch", FW, 24.0, 30.0),
|
|
("scale_watch", SW, 24.0, 30.0),
|
|
("vrp_f_watch", VF, 24.0, 30.0),
|
|
("venue_news", VN, 24.0, 30.0),
|
|
("venue_watch", VW, 1.0, 3.0),
|
|
])
|
|
def test_registrato_in_monitor_health_sul_file_che_scrive(nome, modulo, cadence_h, max_age_h):
|
|
s = _spec(nome)
|
|
assert ROOT / "data" / s.directory / s.series == modulo.RUNS
|
|
assert s.open_labeled is False
|
|
assert s.cadence_h == cadence_h
|
|
assert s.max_age_h <= max_age_h
|
|
|
|
|
|
def test_venue_watch_ha_la_cadenza_oraria_del_cron_book():
|
|
"""⚠️ venue_watch gira OGNI ORA in cron_book.sh: max_age stretto, non i 30h dei giornalieri."""
|
|
assert _spec("venue_watch").cadence_h == 1.0 and _spec("venue_watch").max_age_h == 3.0
|
|
|
|
|
|
# ===========================================================================
|
|
# edge_watch — dopo run(): sharpe, n_alert
|
|
# ===========================================================================
|
|
def test_edge_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(EW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["edge_watch.py", "--quiet"])
|
|
monkeypatch.setattr(EW, "run", lambda: dict(sharpe=0.7, protection=[], consec_fail=0, alerts=[]))
|
|
assert EW.main() == 0
|
|
righe = _righe(runs)
|
|
assert len(righe) == 1
|
|
assert righe[0]["sharpe"] == 0.7 and righe[0]["n_alert"] == 0
|
|
assert isinstance(righe[0]["ts"], int)
|
|
|
|
|
|
def test_edge_watch_run_che_fallisce_non_scrive(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(EW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["edge_watch.py", "--quiet"])
|
|
|
|
def boom():
|
|
raise RuntimeError("dati di mercato non leggibili")
|
|
monkeypatch.setattr(EW, "run", boom)
|
|
with pytest.raises(RuntimeError):
|
|
EW.main()
|
|
assert not runs.exists()
|
|
|
|
|
|
# ===========================================================================
|
|
# fee_watch — dopo save_state(): level
|
|
# ===========================================================================
|
|
def test_fee_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(FW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["fee_watch.py", "--quiet"])
|
|
monkeypatch.setattr(FW, "run", lambda: dict(current={}, changes=[], worst_taker_bps=0.0,
|
|
level="OK", reason="tutto in regola", first_read=True))
|
|
monkeypatch.setattr(FW, "save_state", lambda cur: None)
|
|
assert FW.main() == 0
|
|
righe = _righe(runs)
|
|
assert len(righe) == 1 and righe[0]["level"] == "OK"
|
|
|
|
|
|
def test_fee_watch_run_che_fallisce_non_scrive(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(FW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["fee_watch.py", "--quiet"])
|
|
|
|
def boom():
|
|
raise RuntimeError("API Deribit irraggiungibile")
|
|
monkeypatch.setattr(FW, "run", boom)
|
|
with pytest.raises(RuntimeError):
|
|
FW.main()
|
|
assert not runs.exists()
|
|
|
|
|
|
# ===========================================================================
|
|
# scale_watch — dopo run_once(): stato
|
|
# ===========================================================================
|
|
_REP_SCALA = dict(stato="OK", scala_dichiarata=1.0, ladder="0.5/1/2", tetto=2.0,
|
|
domande={}, fallback={"stato": "OK", "quota": 0.0, "fallback_veri": 0,
|
|
"giri": 10, "soglia": 0.1, "paper": 0},
|
|
lettore={"stato": "OK", "motivo": "ok"}, invio="non tentato")
|
|
|
|
|
|
def test_scale_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(SW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["scale_watch.py", "--quiet", "--secco"])
|
|
monkeypatch.setattr(SW, "run_once", lambda sender=None: dict(_REP_SCALA))
|
|
SW.main()
|
|
righe = _righe(runs)
|
|
assert len(righe) == 1 and righe[0]["stato"] == "OK"
|
|
|
|
|
|
def test_scale_watch_run_once_che_fallisce_non_scrive(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(SW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["scale_watch.py", "--quiet", "--secco"])
|
|
|
|
def boom(sender=None):
|
|
raise RuntimeError("lettore di produzione irraggiungibile")
|
|
monkeypatch.setattr(SW, "run_once", boom)
|
|
with pytest.raises(RuntimeError):
|
|
SW.main()
|
|
assert not runs.exists()
|
|
|
|
|
|
# ===========================================================================
|
|
# vrp_f_watch — dopo _save_state(), SOLO se R non e' vuoto: pairs, ready
|
|
# ===========================================================================
|
|
def _frame_coppie(n: int = 45) -> pd.DataFrame:
|
|
rows = []
|
|
for i in range(n):
|
|
exp = pd.Timestamp("2026-01-01") + pd.Timedelta(days=i)
|
|
for name, v in (("canonico", 0.70), ("candidato", 0.80)):
|
|
rows.append(dict(asset="BTC", struct=name, exp=exp, ts=pd.Timestamp("2026-01-01"),
|
|
dte=7.0, f_short=1.0, f_long=2.0, f_net=v, f_net_mid=v,
|
|
quota_lunga=0.18))
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def test_vrp_f_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(VF, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["vrp_f_watch.py", "--quiet"])
|
|
frame = _frame_coppie()
|
|
monkeypatch.setattr(VF, "measure", lambda chain=None: frame)
|
|
monkeypatch.setattr(VF, "_save_state", lambda d: None)
|
|
monkeypatch.setattr(VF, "_load_state", lambda: {})
|
|
monkeypatch.setattr(VF, "notify", lambda *a, **k: True)
|
|
assert VF.main() == 0
|
|
righe = _righe(runs)
|
|
atteso = VF.assess(frame)
|
|
assert len(righe) == 1
|
|
assert righe[0]["pairs"] == atteso["pairs"] and righe[0]["ready"] == atteso["ready"]
|
|
|
|
|
|
def test_vrp_f_watch_senza_osservazioni_non_scrive(tmp_path, monkeypatch):
|
|
"""P5: R vuoto non e' un giro misurato — nessuna riga (a differenza di cc01, qui non
|
|
c'e' nemmeno un `return` anticipato: la guardia sta nella condizione prima di registrare)."""
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(VF, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["vrp_f_watch.py", "--quiet"])
|
|
monkeypatch.setattr(VF, "measure", lambda chain=None: pd.DataFrame())
|
|
monkeypatch.setattr(VF, "_save_state", lambda d: None)
|
|
monkeypatch.setattr(VF, "_load_state", lambda: {})
|
|
assert VF.main() == 0
|
|
assert not runs.exists()
|
|
|
|
|
|
# ===========================================================================
|
|
# venue_news — dopo la scrittura delle voci: voci, nuove. Feed illeggibile -> nessuna riga
|
|
# ===========================================================================
|
|
_FEED = """<rss><channel>
|
|
<item><title>Deribit Welcomes Orbit Markets</title>
|
|
<link>https://x/a</link><guid>https://x/?p=1</guid><pubDate>Fri, 10 Jul 2026 00:00:00 +0000</pubDate></item>
|
|
<item><title>Another Unrelated Announcement</title>
|
|
<link>https://x/b</link><guid>https://x/?p=2</guid><pubDate>Fri, 11 Jul 2026 00:00:00 +0000</pubDate></item>
|
|
</channel></rss>"""
|
|
|
|
|
|
def test_venue_news_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
state = tmp_path / "venue_news.jsonl"
|
|
monkeypatch.setattr(VN, "RUNS", runs)
|
|
monkeypatch.setattr(VN, "STATE", state)
|
|
monkeypatch.setattr(VN, "scarica", lambda *a, **k: _FEED)
|
|
monkeypatch.setattr(VN, "notify", lambda *a, **k: True)
|
|
monkeypatch.setattr(sys, "argv", ["venue_news.py", "--quiet"])
|
|
assert VN.main() == 0
|
|
righe = _righe(runs)
|
|
assert len(righe) == 1 and righe[0]["voci"] == 2 and righe[0]["nuove"] == 2
|
|
|
|
|
|
def test_venue_news_feed_illeggibile_non_scrive(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(VN, "RUNS", runs)
|
|
monkeypatch.setattr(VN, "STATE", tmp_path / "venue_news.jsonl")
|
|
monkeypatch.setattr(VN, "scarica", lambda *a, **k: None)
|
|
monkeypatch.setattr(VN, "notify", lambda *a, **k: True)
|
|
monkeypatch.setattr(sys, "argv", ["venue_news.py", "--quiet"])
|
|
assert VN.main() == 2
|
|
assert not runs.exists()
|
|
|
|
|
|
# ===========================================================================
|
|
# venue_watch — dopo run_once() e check_specs(): livelli (rep["levels"])
|
|
# ===========================================================================
|
|
def test_venue_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(VW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["venue_watch.py", "--quiet"])
|
|
monkeypatch.setattr(VW, "run_once", lambda sender=None: dict(
|
|
levels={"BTC": "OK", "ETH": "OK"}, alerts=[], platform_locked=False))
|
|
monkeypatch.setattr(VW, "check_specs", lambda: dict(divergenze=[]))
|
|
assert VW.main() == 0
|
|
righe = _righe(runs)
|
|
assert len(righe) == 1 and righe[0]["livelli"] == {"BTC": "OK", "ETH": "OK"}
|
|
|
|
|
|
def test_venue_watch_run_once_che_fallisce_non_scrive(tmp_path, monkeypatch):
|
|
runs = tmp_path / "runs.jsonl"
|
|
monkeypatch.setattr(VW, "RUNS", runs)
|
|
monkeypatch.setattr(sys, "argv", ["venue_watch.py", "--quiet"])
|
|
|
|
def boom(sender=None):
|
|
raise RuntimeError("osservazione del consenso fallita")
|
|
monkeypatch.setattr(VW, "run_once", boom)
|
|
with pytest.raises(RuntimeError):
|
|
VW.main()
|
|
assert not runs.exists()
|