venue_watch: BLIND ha una severity propria, il titolo non chiede piu' il prelievo (#31)
Con un asset BLIND e nessun ALERT, severity restava None e il titolo del
runner cadeva sul default piu' forte ("PRIMO PASSO: prelievo di prova") per
un episodio che e' "non vedo", non "stress misurato". run_once ora assegna
severity="blind" quando c'e' un BLIND senza ALERT (ALERT resta piu' forte, P9);
_titolo distingue il caso con "venue_watch CIECO — referenze non leggibili,
nessun giudizio sul venue". La ripetizione oraria di BLIND non e' toccata.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -499,3 +499,84 @@ def test_senza_sender_il_report_lo_DICE(monkeypatch, tmp_path):
|
||||
"""'non tentato' non e' 'inviato': il silenzio non deve leggersi come un invio riuscito."""
|
||||
sp, rep = _porta_ad_alert(monkeypatch, tmp_path, None)
|
||||
assert "non tentato" in rep["invio"]
|
||||
|
||||
|
||||
# =============================================================================================
|
||||
# issue #31 — BLIND senza ALERT lasciava severity=None: il titolo del runner cadeva sul default
|
||||
# piu' forte ("PRIMO PASSO: prelievo di prova") anche per un episodio che e' "non vedo", non
|
||||
# "stress misurato". BLIND e' lo stato MORBIDO (docstring del modulo).
|
||||
# =============================================================================================
|
||||
|
||||
def test_severity_blind_quando_un_asset_e_cieco_senza_alert(monkeypatch, tmp_path):
|
||||
sp = tmp_path / "vw.json"
|
||||
|
||||
def fake():
|
||||
return {"ts": 1, "platform_locked": False, "locked_raw": "false", "maintenance": False,
|
||||
"assets": {a: dict(deribit=1.0, n_refs=1, ref_spread_bps=None, bps=None)
|
||||
for a in VW.ASSETS},
|
||||
"deribit_errors": []}
|
||||
monkeypatch.setattr(VW, "observe", fake)
|
||||
|
||||
rep = None
|
||||
for _ in range(VW.BLIND_ALERT_HOURS):
|
||||
rep = VW.run_once(state_path=sp, sender=lambda r: (True, "ok"))
|
||||
assert any(l == "BLIND" for l in rep["levels"].values())
|
||||
assert not any(l == "ALERT" for l in rep["levels"].values())
|
||||
assert rep["severity"] == "blind"
|
||||
|
||||
|
||||
def test_severity_alert_prevale_su_blind(monkeypatch, tmp_path):
|
||||
"""Se un asset e' cieco ma l'altro e' in ALERT vero, la gravita' massima vince (P9).
|
||||
|
||||
BTC resta senza referenze per tutto il giro (arriva a BLIND all'ora `BLIND_ALERT_HOURS`);
|
||||
ETH parte harmless e passa a un vero streak sopra soglia SOLO negli ultimi `PERSIST_HOURS`
|
||||
round, cosi' che il round in cui ETH scatta ALERT sia lo stesso in cui BTC e' gia' BLIND."""
|
||||
sp = tmp_path / "vw.json"
|
||||
letture = {"n": 0}
|
||||
soglia_eth = VW.BLIND_ALERT_HOURS - VW.PERSIST_HOURS # da qui in poi ETH costruisce lo streak
|
||||
|
||||
def fake():
|
||||
letture["n"] += 1
|
||||
eth_bps = VW.THRESHOLD_BPS * 3 if letture["n"] > soglia_eth else 1.0
|
||||
return {"ts": letture["n"], "platform_locked": False, "locked_raw": "false",
|
||||
"maintenance": False,
|
||||
"assets": {"BTC": dict(deribit=1.0, n_refs=1, ref_spread_bps=None, bps=None),
|
||||
"ETH": dict(deribit=1.0, n_refs=3, ref_spread_bps=1.0, bps=eth_bps)},
|
||||
"deribit_errors": []}
|
||||
monkeypatch.setattr(VW, "observe", fake)
|
||||
|
||||
rep = None
|
||||
for _ in range(VW.BLIND_ALERT_HOURS):
|
||||
rep = VW.run_once(state_path=sp, sender=lambda r: (True, "ok"))
|
||||
assert rep["levels"]["BTC"] == "BLIND"
|
||||
assert rep["levels"]["ETH"] == "ALERT"
|
||||
assert rep["severity"] == "alert"
|
||||
|
||||
|
||||
def _carica_script_venue_watch():
|
||||
"""scripts/live/venue_watch.py (il runner) — via importlib, per non collidere col nome
|
||||
del modulo `src.live.venue_watch` gia' importato sopra come `VW`."""
|
||||
import importlib.util
|
||||
sp = importlib.util.spec_from_file_location(
|
||||
"venue_watch_script", ROOT / "scripts" / "live" / "venue_watch.py")
|
||||
m = importlib.util.module_from_spec(sp)
|
||||
sp.loader.exec_module(m)
|
||||
return m
|
||||
|
||||
|
||||
def test_titolo_blind_non_dice_prelievo():
|
||||
SVW = _carica_script_venue_watch()
|
||||
t = SVW._titolo({"severity": "blind"})
|
||||
assert "prelievo" not in t.lower()
|
||||
assert "CIECO" in t
|
||||
|
||||
|
||||
def test_titolo_alert_dice_ancora_prelievo():
|
||||
SVW = _carica_script_venue_watch()
|
||||
assert "prelievo" in SVW._titolo({"severity": "alert"}).lower()
|
||||
assert "prelievo" in SVW._titolo({"severity": None}).lower() # default: nessuna gravita' letta
|
||||
|
||||
|
||||
def test_titolo_warn_resta_quello_di_prima():
|
||||
SVW = _carica_script_venue_watch()
|
||||
assert "Nessuna azione" in SVW._titolo({"severity": "warn"})
|
||||
|
||||
Reference in New Issue
Block a user