155 lines
7.0 KiB
Python
155 lines
7.0 KiB
Python
"""Lock del report giornaliero Telegram (in cron dal 2026-07-25).
|
||
|
||
Gira in produzione dentro cron_daily.sh: se rompe, non deve rompere nulla d'altro e non deve
|
||
MAI inviare ordini. I test bloccano proprio questo: e' sola lettura, tollera i pezzi mancanti,
|
||
e non invia niente in dry-run.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from datetime import date
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
sys.path.insert(0, str(ROOT / "scripts" / "live"))
|
||
|
||
telegram_daily = pytest.importorskip("telegram_daily")
|
||
|
||
|
||
def test_build_non_solleva_e_ha_le_sezioni():
|
||
txt = telegram_daily.build()
|
||
assert isinstance(txt, str) and len(txt) > 50
|
||
for sezione in ("Conto", "Ordini", "Monitor in osservazione"):
|
||
assert sezione in txt, f"sezione mancante: {sezione}"
|
||
# (#35 C14) l'header sul trend e' condizionale: esattamente uno dei due, non piu' fisso.
|
||
assert ("Perche' non opera" in txt) != ("Stato trend" in txt)
|
||
|
||
|
||
def test_header_perche_non_opera_solo_se_il_libro_e_flat(monkeypatch):
|
||
"""(#35 C14) L'header fisso mentiva quando il libro NON e' fermo (posizioni o target != 0):
|
||
con flat=True resta "Perche' non opera"."""
|
||
monkeypatch.setattr(telegram_daily, "book_state", lambda: ([" equity $1.00"], True))
|
||
assert "Perche' non opera" in telegram_daily.build()
|
||
|
||
|
||
def test_header_neutro_se_il_libro_non_e_flat(monkeypatch):
|
||
monkeypatch.setattr(telegram_daily, "book_state",
|
||
lambda: ([" BTC: pos $300 → target $300"], False))
|
||
txt = telegram_daily.build()
|
||
assert "Stato trend" in txt and "Perche' non opera" not in txt
|
||
|
||
|
||
def test_header_neutro_se_il_libro_non_e_leggibile(monkeypatch):
|
||
"""Se book_state non sa dire se e' flat (eccezione -> None), niente header "non opera"
|
||
dichiarato senza saperlo davvero: si usa il neutro."""
|
||
monkeypatch.setattr(telegram_daily, "book_state",
|
||
lambda: ([" ⚠️ stato libro non leggibile: Errore"], None))
|
||
txt = telegram_daily.build()
|
||
assert "Stato trend" in txt and "Perche' non opera" not in txt
|
||
|
||
|
||
def test_nessuna_scrittura_ne_ordine():
|
||
"""Il modulo non deve importare l'esecutore ne' avere funzioni che inviano ordini."""
|
||
src = (ROOT / "scripts" / "live" / "telegram_daily.py").read_text()
|
||
for vietato in ("DeribitTrader", "book_execute", "--execute", "place_order", "buy(", "sell("):
|
||
assert vietato not in src, f"il report non deve poter operare: trovato {vietato!r}"
|
||
|
||
|
||
def test_tollera_i_log_mancanti(monkeypatch, tmp_path):
|
||
"""Se i log dei monitor non esistono, il report deve degradare, non esplodere."""
|
||
monkeypatch.setattr(telegram_daily, "ROOT", tmp_path)
|
||
righe = telegram_daily.monitors()
|
||
assert len(righe) == len(telegram_daily.GATES)
|
||
assert all(isinstance(r, str) and r.strip() for r in righe)
|
||
assert telegram_daily.last_trade().strip()
|
||
|
||
|
||
def test_gate_coerenti_con_i_pre_registrati():
|
||
"""Le date qui devono coincidere con i gate pre-registrati negli script di decisione."""
|
||
d = dict((n, x) for n, x, _, _ in telegram_daily.GATES)
|
||
assert str(d["STATARB-RESID"]) == "2026-09-27"
|
||
assert str(d["XSR01"]) == "2026-10-23"
|
||
|
||
|
||
def test_convenzione_trend_e_documentata():
|
||
"""La convenzione media-dei-segni (non frazione) e' la ragione per cui TP01 e' 0 con 1/3 su:
|
||
se sparisce dal codice, il report mentirebbe sul 'quanto manca'."""
|
||
assert "media dei SEGNI" in telegram_daily.trend_state.__doc__
|
||
|
||
|
||
def test_soglia_accensione_2_negativi_1_positivo():
|
||
"""2 negativi + 1 positivo: basta 1 flip -> soglia = il riferimento negativo PIU' BASSO."""
|
||
r = telegram_daily._soglia_accensione([-1, -1, 1], [100.0, 200.0])
|
||
assert r == 100.0
|
||
|
||
|
||
def test_soglia_accensione_3_negativi():
|
||
"""3 negativi: servono 2 flip -> soglia = il SECONDO riferimento negativo piu' basso, non il
|
||
minimo (il bug di #35 C3: superare solo il minimo lascia la media <= 0)."""
|
||
r = telegram_daily._soglia_accensione([-1, -1, -1], [100.0, 200.0, 300.0])
|
||
assert r == 200.0
|
||
|
||
|
||
def test_soglia_accensione_1_negativo_2_positivi_gia_acceso():
|
||
"""1 negativo + 2 positivi: la media e' gia' positiva (TP01 > 0) -> nessuna soglia da dare."""
|
||
assert telegram_daily._soglia_accensione([-1, 1, 1], [100.0]) is None
|
||
|
||
|
||
def test_soglia_accensione_nessun_negativo():
|
||
"""Tutti positivi: nessun riferimento negativo, nessuna soglia."""
|
||
assert telegram_daily._soglia_accensione([1, 1, 1], []) is None
|
||
|
||
|
||
def test_soglia_accensione_servono_tutti_i_flip():
|
||
"""2 negativi su 2 (nessun positivo): servono ENTRAMBI i flip -> k == len(neg_refs), soglia
|
||
= il piu' alto dei due (l'ultimo a bastare), non il minimo."""
|
||
r = telegram_daily._soglia_accensione([-1, -1], [200.0, 100.0])
|
||
assert r == 200.0
|
||
|
||
|
||
# ---------------------------------------------------------------- (#35 C14) gate: data, formato
|
||
|
||
|
||
def test_gate_date_legge_da_monitor_health():
|
||
"""La data non e' una seconda dichiarazione: viene da monitor_health.MONITORS (`gate`),
|
||
la stessa tabella che gli alert dei sorveglianti citano come "METTE A RISCHIO il gate"."""
|
||
from src.live.monitor_health import MONITORS
|
||
m = next(x for x in MONITORS if x.name == "paper_statarb")
|
||
assert m.gate[:10] == "2026-09-27"
|
||
assert telegram_daily._gate_date("paper_statarb", date(2030, 1, 1)) == date(2026, 9, 27)
|
||
m2 = next(x for x in MONITORS if x.name == "paper_xsr")
|
||
assert telegram_daily._gate_date("paper_xsr", date(2030, 1, 1)) == date.fromisoformat(m2.gate[:10])
|
||
|
||
|
||
def test_gate_date_usa_la_cablata_se_il_monitor_non_esiste():
|
||
"""Rete di sicurezza: se il nome non e' in MONITORS (o cambia), non si inventa una data."""
|
||
assert telegram_daily._gate_date("non_esiste_piu'", date(2030, 1, 1)) == date(2030, 1, 1)
|
||
|
||
|
||
def test_fmt_giorni_tre_casi_mai_il_doppio_segno():
|
||
"""(#35 C14) giorni>0 -> '-Ng'; 0 -> 'oggi'; <0 -> 'scaduto da Ng'. Mai '(−-1g)'."""
|
||
assert telegram_daily._fmt_giorni(5) == "-5g"
|
||
assert telegram_daily._fmt_giorni(0) == "oggi"
|
||
assert telegram_daily._fmt_giorni(-1) == "scaduto da 1g"
|
||
assert "-1g" not in telegram_daily._fmt_giorni(-1).replace("scaduto da 1g", "") # niente doppio segno
|
||
|
||
|
||
def test_monitors_formatta_i_tre_casi_del_giorno(monkeypatch, tmp_path):
|
||
"""I rami di monitors() (barra assente qui: file inesistenti in tmp_path) usano tutti
|
||
_fmt_giorni, non il vecchio '(−{giorni}g)' cablato: futuro, oggi e passato si distinguono."""
|
||
from datetime import timedelta
|
||
oggi = date.today()
|
||
monkeypatch.setattr(telegram_daily, "ROOT", tmp_path)
|
||
monkeypatch.setattr(telegram_daily, "GATES",
|
||
[("FUTURO", oggi + timedelta(days=10), "data/nonexiste1/r.jsonl", "k"),
|
||
("OGGI", oggi, "data/nonexiste2/r.jsonl", "k"),
|
||
("PASSATO", oggi - timedelta(days=3), "data/nonexiste3/r.jsonl", "k")])
|
||
righe = telegram_daily.monitors()
|
||
assert "(-10g)" in righe[0]
|
||
assert "(oggi)" in righe[1]
|
||
assert "(scaduto da 3g)" in righe[2]
|
||
assert "−-" not in "".join(righe) and "(−" not in "".join(righe) # mai il doppio segno
|