sei sorveglianti registrano i giri e stanno in monitor_health (Fixes #16)
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>
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
"""registra_giro (src/live/giri.py, #16): il registro che i sorveglianti scrivono a ogni giro.
|
||||
|
||||
Il contratto e' quello di cc01_regime_watch, generalizzato: appende `{"ts": <ms>, **campi}` come
|
||||
riga jsonl e non solleva MAI — un registro che fallisce non deve buttare via una misura gia' fatta.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from src.live.giri import registra_giro # noqa: E402
|
||||
|
||||
|
||||
def test_scrive_una_riga_con_ts_in_millisecondi(tmp_path):
|
||||
p = tmp_path / "sub" / "runs.jsonl"
|
||||
prima = time.time() * 1000
|
||||
assert registra_giro(p, stato="OK", n=3) is True
|
||||
riga = json.loads(p.read_text().strip())
|
||||
assert abs(riga["ts"] - prima) < 5_000
|
||||
assert isinstance(riga["ts"], int)
|
||||
assert riga["stato"] == "OK" and riga["n"] == 3
|
||||
|
||||
|
||||
def test_crea_la_cartella_se_manca(tmp_path):
|
||||
p = tmp_path / "non-esiste-ancora" / "runs.jsonl"
|
||||
assert not p.parent.exists()
|
||||
registra_giro(p, x=1)
|
||||
assert p.exists()
|
||||
|
||||
|
||||
def test_appende_senza_troncare(tmp_path):
|
||||
p = tmp_path / "runs.jsonl"
|
||||
registra_giro(p, giro=1)
|
||||
registra_giro(p, giro=2)
|
||||
righe = [json.loads(r) for r in p.read_text().splitlines()]
|
||||
assert [r["giro"] for r in righe] == [1, 2]
|
||||
|
||||
|
||||
def test_default_str_copre_i_valori_non_serializzabili(tmp_path):
|
||||
"""json.dumps(default=str): un valore che non e' nativamente JSON (es. un Path, o un
|
||||
dataclass senza __dict__ diretto) non deve far fallire l'append — diventa la sua str()."""
|
||||
p = tmp_path / "runs.jsonl"
|
||||
assert registra_giro(p, percorso=Path("/tmp/x")) is True
|
||||
riga = json.loads(p.read_text().strip())
|
||||
assert riga["percorso"] == str(Path("/tmp/x"))
|
||||
|
||||
|
||||
def test_oserror_ritorna_false_senza_sollevare(tmp_path, capsys):
|
||||
"""Un percorso impossibile (qui: dentro un FILE, non una cartella) e' un OSError alla
|
||||
mkdir/scrittura. Il chiamante non deve vedere un'eccezione: la misura e' gia' fatta."""
|
||||
ostacolo = tmp_path / "non-e-una-cartella"
|
||||
ostacolo.write_text("sono un file")
|
||||
p = ostacolo / "runs.jsonl"
|
||||
assert registra_giro(p, stato="OK") is False
|
||||
err = capsys.readouterr().err
|
||||
assert "registra_giro" in err and str(p) in err
|
||||
|
||||
|
||||
def test_un_valore_non_serializzabile_diventa_stringa_non_solleva(tmp_path):
|
||||
p = tmp_path / "runs.jsonl"
|
||||
oggetto = object()
|
||||
assert registra_giro(p, oggetto=oggetto) is True
|
||||
riga = json.loads(p.read_text().strip())
|
||||
assert riga["oggetto"] == str(oggetto)
|
||||
@@ -0,0 +1,270 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user