Adriano Dal Pastro
1279ea605a
Il feed 1h certificato si ferma alle 00:00: per le letture successive `asof` dava la stessa barra a t0 e t1, mercato "fermo" = 0, e qualunque calo >=10% del giorno sarebbe stato un "movimento" scorporato come prelievo (giornale dal 25/08, report da oggi). Con feed assente tutto era "ambiguo" e il report stampava e1/e0-1 sotto l'etichetta TWR. - journal.movimenti_capitale: stato "mercato non misurabile" (feed fermo/assente/barra mancante) => ambiguo con la ragione; leva_tetto = min(frac x n x scala, LEVA_LORDA_MAX) (P1). - journal.rendimento_twr: salto non classificabile => twr E trading None con motivo. - journal.pnl_giorno CHIAMA rendimento_twr (prima rifaceva cum - certi: "entrambi la chiamano" era falso); la pagina stampa il TWR; analista qualifica il cumulato "di cui versati". - trades_db --report: la classe di ogni salto con la sua misura (mercato max a leva piena). - book_execute docstring: banda dello stop -33,5%/-26,5% (era invertita), "non costa ordini" -> micro-ordini di ri-taglia (22/48, C2), latenza <=1h solo con feed fresca, conteggi di r0823 al posto di "raddoppia" (2 contro 1 e' rotolante vs pavimento sulla riga 4h; la peggiore misurata e' 24h). - test_book_cadenza: LTF_MIN importato da skyhook; tutte le righe attive della crontab (una sola); tre stati dello skip (col progetto ma senza cron_book => ROSSO); slot di release testato con venue_probe (il :07 come controllo positivo); parser regolare, passo > 0. - r0823_sl_anchor: guardia sul :47 (era sul vecchio :07), prosa al passato; cron_chain.sh idem. - CLAUDE.md: §5.7 frase invertita corretta, §5.14 limite vero (non l'artefatto della fixture), §2 ora della LETTURA (13:47Z), nuovo debito §5.15 (USDE fuori dal bound di mercato). - diari: tempi "scritto" corretti coi commit; sezione "Seconda tornata". Test +11 (858 verdi). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XDJsH3iDSaBns3ccpPBwiu
662 lines
33 KiB
Python
662 lines
33 KiB
Python
"""Il giornale di bordo: tre stati invece di due, e la nota dell'operatore che non si perde.
|
|
|
|
Il rischio di un giornale automatico non e' sbagliare un numero: e' scrivere `0` dove il dato
|
|
non c'era, e leggerlo fra sei mesi come una misura.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.live import journal as J # noqa: E402
|
|
from src.live import tradesdb as T # noqa: E402
|
|
|
|
|
|
@pytest.fixture()
|
|
def db(tmp_path):
|
|
con = T.connect(tmp_path / "t.db")
|
|
yield con
|
|
con.close()
|
|
|
|
|
|
# ------------------------------------------------- i giri attesi non sono sempre 24
|
|
|
|
def test_giorno_in_corso_non_e_un_giorno_bucato(db, monkeypatch):
|
|
"""24 giri attesi su un giorno non finito = allarme a ogni esecuzione, cioe' rumore."""
|
|
oggi = datetime.now(timezone.utc).date()
|
|
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=datetime.now(timezone.utc).hour + 1,
|
|
feed_skh_min=0))
|
|
s = J.salute(db, oggi)
|
|
assert s["giorno_in_corso"] is True
|
|
assert s["atteso_giri"] == datetime.now(timezone.utc).hour + 1
|
|
assert s["giri_mancanti"] == 0
|
|
|
|
|
|
def test_giorno_chiuso_si_giudica_su_24(db, monkeypatch):
|
|
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=20, feed_skh_min=0))
|
|
s = J.salute(db, date(2026, 8, 21))
|
|
assert s["giorno_in_corso"] is False
|
|
assert s["atteso_giri"] == 24 and s["giri_mancanti"] == 4
|
|
|
|
|
|
def test_controllo_positivo_un_giorno_davvero_bucato_viene_segnalato(db, monkeypatch):
|
|
monkeypatch.setattr(J, "stato_libro", lambda c, g: dict(giri=6, feed_skh_min=None))
|
|
assert J.salute(db, date(2026, 8, 21))["giri_mancanti"] == 18
|
|
|
|
|
|
# ------------------------------------------------- dato assente != zero
|
|
|
|
def test_dvol_assente_e_None_con_ragione_mai_zero(db, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path) # nessun file DVOL qui
|
|
(tmp_path / "data" / "raw").mkdir(parents=True)
|
|
m = J.metriche_mercato(date(2026, 8, 21))
|
|
for a in ("BTC", "ETH"):
|
|
if "errore" in m[a]:
|
|
continue
|
|
assert m[a]["dvol"] is None
|
|
assert m[a]["dvol_nota"] == "file DVOL assente" # la ragione, non un valore
|
|
|
|
|
|
def test_giorno_senza_giri_e_un_errore_dichiarato_non_un_libro_vuoto(db, monkeypatch):
|
|
monkeypatch.setattr(T, "CRON_LOG", Path("/tmp/non-esiste-affatto.log"))
|
|
lb = J.stato_libro(db, date(2026, 8, 21))
|
|
assert "errore" in lb and lb["giri"] == 0
|
|
assert "equity" not in lb # non inventa uno zero
|
|
|
|
|
|
def test_pnl_senza_letture_di_equity_non_fabbrica_un_delta(db):
|
|
p = J.pnl_giorno(db, date(2026, 8, 21))
|
|
assert p["delta_equity"] is None and p["equity_fine"] is None
|
|
assert p["realizzato_netto"] == 0.0 and p["roundtrip_chiusi"] == 0 # 0 trade e' un fatto
|
|
|
|
|
|
# ------------------------------------------------- la nota dell'operatore
|
|
|
|
def test_la_nota_sopravvive_a_un_ricalcolo(db):
|
|
g = date(2026, 8, 21)
|
|
voce = dict(giorno=g.isoformat(), ts_scritto=T.ora(), mercato={}, libro={}, pnl={}, salute={})
|
|
J.salva(db, voce, nota="rally post-CPI, non ho toccato niente")
|
|
J.salva(db, voce) # ricalcolo automatico
|
|
n = db.execute("SELECT nota FROM journal WHERE giorno=?", (g.isoformat(),)).fetchone()["nota"]
|
|
assert n == "rally post-CPI, non ho toccato niente"
|
|
|
|
|
|
def test_la_nota_si_puo_sostituire_esplicitamente(db):
|
|
g = date(2026, 8, 21)
|
|
voce = dict(giorno=g.isoformat(), ts_scritto=T.ora(), mercato={}, libro={}, pnl={}, salute={})
|
|
J.salva(db, voce, nota="prima")
|
|
J.salva(db, voce, nota="seconda")
|
|
n = db.execute("SELECT nota FROM journal WHERE giorno=?", (g.isoformat(),)).fetchone()["nota"]
|
|
assert n == "seconda"
|
|
|
|
|
|
# ------------------------------------------------- il markdown non mente
|
|
|
|
def test_il_markdown_dice_n_d_e_non_zero_quando_manca_il_dato(db):
|
|
voce = dict(giorno="2026-08-21", ts_scritto=T.ora(),
|
|
mercato={"BTC": {"errore": "feed non leggibile: FileNotFoundError"},
|
|
"ETH": {"errore": "feed non leggibile: FileNotFoundError"}},
|
|
libro={"errore": "nessun giro di book_execute quel giorno", "giri": 0},
|
|
pnl=J.pnl_giorno(db, date(2026, 8, 21)),
|
|
salute=dict(giri_book=0, feed_skh_min=None, atteso_giri=24,
|
|
giorno_in_corso=False, giri_mancanti=24))
|
|
J.salva(db, voce)
|
|
md = J.rendi_markdown(db, voce)
|
|
assert "n/d" in md and "24 mancanti" in md
|
|
assert "non misurata" in md # eta' feed sconosciuta
|
|
assert "$+0.00 di equity" not in md # niente zeri inventati
|
|
|
|
|
|
def test_il_markdown_riporta_il_segnale_che_il_libro_leggeva(db):
|
|
voce = dict(giorno="2026-08-21", ts_scritto=T.ora(), mercato={},
|
|
libro=dict(giri=24, ultimo_giro="2026-08-21T23:07:01+00:00", equity=644.0,
|
|
barra="2026-08-21", feed_skh_min=0, nozionale_lordo=196.0,
|
|
leva_lorda=0.304,
|
|
asset={"BTC": dict(tp_frac=0.176, skh_sign=1, skh_entry=75280.0,
|
|
target=123.0, posizione=126.0, azione="HOLD (a target)")}),
|
|
pnl=J.pnl_giorno(db, date(2026, 8, 21)),
|
|
salute=dict(giri_book=24, feed_skh_min=0, atteso_giri=24,
|
|
giorno_in_corso=False, giri_mancanti=0))
|
|
J.salva(db, voce)
|
|
md = J.rendi_markdown(db, voce)
|
|
assert "+0.176" in md and "LONG @ 75,280.0" in md and "0.30x" in md
|
|
|
|
|
|
def test_lo_stato_tsmom_usa_gli_orizzonti_di_produzione():
|
|
"""Il giornale deve riportare CIO' CHE IL LIBRO LEGGE, non un trend a caso."""
|
|
from src.strategies import trend_portfolio as tp
|
|
src = Path(tp.__file__).read_text()
|
|
for n in J.ORIZZONTI:
|
|
assert str(n) in src, f"orizzonte {n} non compare in trend_portfolio.py"
|
|
|
|
|
|
# =============================================================================================
|
|
# LETTURA RAGIONATA — ogni regola ha un test che la ACCENDE e uno che la tiene SPENTA
|
|
# =============================================================================================
|
|
# Una regola che si accende sempre non sta leggendo niente, e una che non si accende mai e'
|
|
# indistinguibile da una rotta. Servono entrambe le direzioni.
|
|
|
|
def _voce(mercato=None, libro=None, pnl=None, salute=None, giorno="2026-08-21"):
|
|
return dict(giorno=giorno, ts_scritto=T.ora(), mercato=mercato or {},
|
|
libro=libro or {"errore": "nessun giro", "giri": 0}, pnl=pnl or {},
|
|
salute=salute or dict(giri_book=24, feed_skh_min=0, atteso_giri=24,
|
|
giorno_in_corso=False, giri_mancanti=0))
|
|
|
|
|
|
def _libro(**asset):
|
|
a = {k: dict(dict(tp_frac=0.0, skh_sign=0, skh_entry=None, target=0.0, posizione=0.0,
|
|
azione="HOLD (a target)"), **v) for k, v in asset.items()}
|
|
lordo = sum(abs(v["posizione"]) for v in a.values())
|
|
return dict(giri=24, ultimo_giro="2026-08-21T23:07:01+00:00", equity=600.0, barra="2026-08-21",
|
|
feed_skh_min=0, nozionale_lordo=lordo, leva_lorda=lordo / 600.0, asset=a)
|
|
|
|
|
|
def _pnl(**kw):
|
|
base = dict(equity_inizio=600.0, equity_fine=600.0, delta_equity=0.0, realizzato_lordo=0.0,
|
|
realizzato_netto=0.0, roundtrip_chiusi=0, fill=0, fee=0.0,
|
|
cumulato_da_arming=0.0, letture_equity=24)
|
|
return {**base, **kw}
|
|
|
|
|
|
def _ids(righe):
|
|
return [r["id"] for r in righe]
|
|
|
|
|
|
def test_libro_flat_dice_anche_PERCHE(db):
|
|
v = _voce(libro=_libro(BTC={}, ETH={}), pnl=_pnl())
|
|
r = J.lettura(db, v)
|
|
assert "stato" in _ids(r) and "perche_flat" in _ids(r)
|
|
testo = next(x["testo"] for x in r if x["id"] == "perche_flat")
|
|
assert "TP01 a zero" in testo and "SKH01 senza breakout" in testo
|
|
|
|
|
|
def test_libro_a_mercato_non_spiega_un_flat_che_non_c_e(db):
|
|
v = _voce(libro=_libro(BTC=dict(tp_frac=0.15, skh_sign=1, posizione=116.0)), pnl=_pnl())
|
|
r = J.lettura(db, v)
|
|
assert "perche_flat" not in _ids(r)
|
|
assert "$+116" in next(x["testo"] for x in r if x["id"] == "stato")
|
|
|
|
|
|
def test_disaccordo_si_accende_a_2_su_3_e_tace_a_3_su_3(db):
|
|
lb = _libro(BTC=dict(tp_frac=0.15, skh_sign=1, posizione=116.0))
|
|
misto = {"BTC": dict(chiusura=77000.0, tsmom_su=(2, 3), tsmom={})}
|
|
pieno = {"BTC": dict(chiusura=77000.0, tsmom_su=(3, 3), tsmom={})}
|
|
assert "disaccordo" in _ids(J.lettura(db, _voce(mercato=misto, libro=lb, pnl=_pnl())))
|
|
assert "disaccordo" not in _ids(J.lettura(db, _voce(mercato=pieno, libro=lb, pnl=_pnl())))
|
|
|
|
|
|
def test_incoerenza_trend_su_ma_libro_fuori(db):
|
|
lb = _libro(BTC=dict(tp_frac=0.0, skh_sign=0, posizione=0.0))
|
|
m = {"BTC": dict(chiusura=77000.0, tsmom_su=(3, 3), tsmom={})}
|
|
r = J.lettura(db, _voce(mercato=m, libro=lb, pnl=_pnl()))
|
|
riga = next(x for x in r if x["id"] == "incoerenza")
|
|
assert riga["livello"] == "attenzione"
|
|
|
|
|
|
def test_il_tetto_di_leva_e_DERIVATO_dal_config_non_riscritto(db, monkeypatch, tmp_path):
|
|
"""Se qualcuno cambia `max_notional_per_asset_frac`, la lettura deve cambiare da sola."""
|
|
(tmp_path / "config").mkdir()
|
|
(tmp_path / "config" / "live.json").write_text(json.dumps({"max_notional_per_asset_frac": 0.25}))
|
|
monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path)
|
|
lb = _libro(BTC=dict(posizione=150.0), ETH=dict(posizione=150.0))
|
|
r = J.lettura(db, _voce(libro=lb, pnl=_pnl()))
|
|
testo = next(x["testo"] for x in r if x["id"] == "leva")
|
|
assert "0.50x" in testo and "0.25" in testo # 0.25 x 2 asset, NON 1.00x
|
|
|
|
|
|
def test_config_illeggibile_lo_dice_invece_di_inventare_un_tetto(db, monkeypatch, tmp_path):
|
|
monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path) # nessun config qui
|
|
r = J.lettura(db, _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl()))
|
|
riga = next(x for x in r if x["id"] == "leva")
|
|
assert "non verificabile" in riga["testo"] and riga["livello"] == "attenzione"
|
|
|
|
|
|
def test_soglia_IV_rank_derivata_dal_sleeve(db):
|
|
"""La soglia del gate viene da VRP_CFG, e l'IV-rank e' quello ESPANDENTE del sleeve."""
|
|
from src.portfolio.sleeves import VRP_CFG
|
|
lb = _libro(BTC=dict(posizione=100.0))
|
|
m = {"BTC": dict(chiusura=77000.0, dvol=42.0, rv30_annua=43.0, dvol_rank_exp=0.18)}
|
|
testo = next(x["testo"] for x in J.lettura(db, _voce(mercato=m, libro=lb, pnl=_pnl()))
|
|
if x["id"] == "vol")
|
|
assert f"{VRP_CFG['gate_ivr']:.2f}" in testo and "sotto la soglia" in testo
|
|
m["BTC"]["dvol_rank_exp"] = 0.90
|
|
testo = next(x["testo"] for x in J.lettura(db, _voce(mercato=m, libro=lb, pnl=_pnl()))
|
|
if x["id"] == "vol")
|
|
assert "**sopra**" in testo
|
|
|
|
|
|
def test_iv_rank_espandente_e_diverso_dal_rank_a_un_anno():
|
|
"""I due rank rispondono a domande diverse: il 2026-08-23 danno 0.18 contro 0.52.
|
|
|
|
Etichettare l'uno col nome dell'altro ribalta il verdetto sul gate di VRP01.
|
|
"""
|
|
m = J.metriche_mercato(date.today())
|
|
for a in ("BTC", "ETH"):
|
|
d = m.get(a) or {}
|
|
if d.get("dvol_rank_exp") is None or d.get("dvol_rank_1a") is None:
|
|
continue
|
|
assert d["dvol_rank_exp"] != d["dvol_rank_1a"]
|
|
|
|
|
|
def test_regola_fee_quando_il_costo_supera_il_movimento(db):
|
|
lb = _libro(BTC=dict(posizione=100.0))
|
|
r = J.lettura(db, _voce(libro=lb, pnl=_pnl(delta_equity=0.01, fill=2, fee=0.05)))
|
|
assert "fee" in _ids(r)
|
|
r = J.lettura(db, _voce(libro=lb, pnl=_pnl(delta_equity=5.00, fill=2, fee=0.05)))
|
|
assert "fee" not in _ids(r) and "pnl" in _ids(r)
|
|
|
|
|
|
def test_evento_si_accende_oltre_2_deviazioni_e_tace_sotto(db):
|
|
lb = _libro(BTC=dict(posizione=100.0))
|
|
# RV30 36.5% annua -> sd giornaliera ~1.91%
|
|
grosso = {"BTC": dict(chiusura=77000.0, ret_1g=-6.0, rv30_annua=36.5)}
|
|
normale = {"BTC": dict(chiusura=77000.0, ret_1g=-1.0, rv30_annua=36.5)}
|
|
assert "evento" in _ids(J.lettura(db, _voce(mercato=grosso, libro=lb, pnl=_pnl())))
|
|
assert "evento" not in _ids(J.lettura(db, _voce(mercato=normale, libro=lb, pnl=_pnl())))
|
|
|
|
|
|
def test_le_regole_tacciono_sul_non_misurato(db):
|
|
"""Mercato assente: nessuna riga su vol o eventi. Il silenzio e' il comportamento giusto."""
|
|
r = J.lettura(db, _voce(mercato={"BTC": {"errore": "feed KO"}},
|
|
libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(delta_equity=None)))
|
|
assert "vol" not in _ids(r) and "evento" not in _ids(r) and "pnl" not in _ids(r)
|
|
|
|
|
|
def test_giri_mancanti_e_feed_vecchio_sono_attenzioni(db):
|
|
s = dict(giri_book=6, feed_skh_min=685, atteso_giri=24, giorno_in_corso=False, giri_mancanti=18)
|
|
r = J.lettura(db, _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(), salute=s))
|
|
liv = {x["id"]: x["livello"] for x in r}
|
|
assert liv.get("giri") == "attenzione" and liv.get("feed") == "attenzione"
|
|
|
|
|
|
def test_senza_giri_la_lettura_dice_NON_MISURATO_e_si_ferma(db):
|
|
r = J.lettura(db, _voce())
|
|
assert _ids(r) == ["stato"] and "non e' stato misurato" in r[0]["testo"]
|
|
|
|
|
|
def test_il_bucket_di_convinzione_non_inventa_i_due_terzi():
|
|
"""`tsmom_blend` media tre np.sign(): dopo il clip long-flat la direzione sta in {0,1/3,1}."""
|
|
assert J._bucket_convinzione(0.0) == "flat"
|
|
assert J._bucket_convinzione(0.15) == "parziale"
|
|
assert J._bucket_convinzione(0.90) == "piena"
|
|
|
|
|
|
def test_concentrazione_tace_su_un_cumulato_irrilevante(db):
|
|
"""Vera ma inutile: su pochi dollari la concentrazione e' aritmetica, non informazione."""
|
|
for i in range(60):
|
|
db.execute("INSERT INTO equity VALUES (?,?,?)",
|
|
(f"2026-06-{(i % 28) + 1:02d}T{i % 24:02d}:00:00+00:00", 600.0 + i * 0.03, "t"))
|
|
db.commit()
|
|
v = _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(), giorno="2026-06-28")
|
|
assert "concentrazione" not in _ids(J.lettura(db, v)) # ~$1.8 di cumulato
|
|
|
|
|
|
def test_concentrazione_si_accende_quando_la_taglia_conta(db):
|
|
for i in range(60):
|
|
v = 600.0 if i < 50 else 600.0 + (i - 49) * 4.0 # salto nella coda
|
|
db.execute("INSERT INTO equity VALUES (?,?,?)",
|
|
(f"2026-06-{(i % 28) + 1:02d}T{i % 24:02d}:00:00+00:00", v, "t"))
|
|
db.commit()
|
|
v = _voce(libro=_libro(BTC=dict(posizione=100.0)), pnl=_pnl(), giorno="2026-06-28")
|
|
assert "concentrazione" in _ids(J.lettura(db, v))
|
|
|
|
|
|
# ------------------------------------------- il giorno IN CORSO non si presenta come chiuso
|
|
#
|
|
# Una pagina di giorno in corso ha TUTTE le sezioni di una chiusa: passa ogni controllo di
|
|
# completezza e freschezza, e si legge come una giornata. Fino al 2026-08-25 lo diceva solo
|
|
# la §Salute — quinta sezione su otto — mentre titolo, P&L e Lettura parlavano al passato di
|
|
# una frazione di giornata. Le due prove sono accoppiate: una accende la marcatura, l'altra
|
|
# verifica che su un giorno chiuso resti SPENTA (una marcatura sempre accesa non si legge).
|
|
|
|
def _voce_salute(con, in_corso: bool, giri: int):
|
|
return dict(giorno="2026-08-25", ts_scritto=T.ora(), mercato={},
|
|
libro=dict(giri=giri, ultimo_giro="2026-08-25T07:07:01+00:00", equity=642.0,
|
|
barra="2026-08-25", feed_skh_min=0, nozionale_lordo=200.0,
|
|
leva_lorda=0.31, asset={}),
|
|
pnl=J.pnl_giorno(con, date(2026, 8, 25)),
|
|
salute=dict(giri_book=giri, feed_skh_min=0, atteso_giri=giri if in_corso else 24,
|
|
giorno_in_corso=in_corso, giri_mancanti=0))
|
|
|
|
|
|
def test_una_pagina_di_giorno_in_corso_si_dichiara_parziale_nel_titolo(db):
|
|
v = _voce_salute(db, in_corso=True, giri=8)
|
|
J.salva(db, v)
|
|
md = J.rendi_markdown(db, v)
|
|
titolo = md.splitlines()[0]
|
|
assert "PARZIALE" in titolo and "giorno in corso" in titolo # nel posto che si legge
|
|
testata = "\n".join(md.splitlines()[:6])
|
|
assert "NON e' chiusa" in testata # e non solo nel titolo
|
|
assert "8 giri su 24" in testata # con la copertura dichiarata
|
|
assert "non si aggiorna da sola" in testata # e col fatto che e' congelata
|
|
|
|
|
|
def test_una_pagina_di_giorno_chiuso_non_si_dichiara_parziale(db):
|
|
v = _voce_salute(db, in_corso=False, giri=24)
|
|
J.salva(db, v)
|
|
md = J.rendi_markdown(db, v)
|
|
assert "PARZIALE" not in md and "non si aggiorna da sola" not in md
|
|
assert md.splitlines()[0] == "# Giornale di bordo — 2026-08-25"
|
|
|
|
|
|
def test_il_cron_non_apre_piu_la_voce_del_giorno_in_corso():
|
|
"""Guardia sul SORGENTE: `journal.py` senza `--giorno` scrive OGGI, e alle 00:30 UTC oggi
|
|
e' mezz'ora. Chi la rimettesse congelerebbe di nuovo una pagina che sembra una giornata.
|
|
Il test non vieta di rigenerarla spesso: vieta di scriverla UNA volta e lasciarla li'.
|
|
"""
|
|
cron = (ROOT / "scripts" / "cron_daily.sh").read_text()
|
|
chiamate = [r.strip() for r in cron.splitlines()
|
|
if "scripts/live/journal.py" in r and not r.strip().startswith("#")]
|
|
assert chiamate, "il giornale non gira piu' nel cron: e' un'altra rottura, non questa"
|
|
for r in chiamate:
|
|
assert "--giorno" in r, f"apre un giorno implicito (= oggi, parziale): {r}"
|
|
|
|
|
|
# ------------------------------------------------- movimenti di capitale nel P&L
|
|
# Il difetto (2026-08-25): la voce dichiarava «giorno: +$1.414,57» quando $1.399 erano un
|
|
# deposito USDC — il P&L era un delta di equity e ogni versamento ci finiva dentro come
|
|
# profitto, per sempre (il cumulato dall'arming non si sarebbe mai piu' corretto).
|
|
|
|
def _px_piatto(livello=100.0):
|
|
ix = pd.date_range("2026-08-20", "2026-08-27", freq="1h", tz="UTC")
|
|
return pd.Series(livello, index=ix)
|
|
|
|
|
|
def _serie_equity(con, punti):
|
|
for ts, eq in punti:
|
|
con.execute("INSERT INTO equity (ts_utc, equity, src) VALUES (?,?,?)", (ts, eq, "test"))
|
|
con.commit()
|
|
|
|
|
|
def test_un_versamento_viene_rilevato_e_scorporato(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T09:47:01+00:00", 667.0),
|
|
("2026-08-25T10:47:01+00:00", 667.5),
|
|
("2026-08-25T11:47:01+00:00", 2066.5),
|
|
("2026-08-25T12:47:01+00:00", 2064.0)])
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
ev = p["movimenti_giorno"]
|
|
assert len(ev) == 1 and ev[0]["classe"] == "movimento"
|
|
assert p["mov_giorno"] == pytest.approx(1399.0)
|
|
# il delta resta il FATTO, il trading e' il netto
|
|
assert p["delta_equity"] == pytest.approx(2064.0 - 667.0)
|
|
assert p["trading_giorno"] == pytest.approx(p["delta_equity"] - 1399.0)
|
|
assert p["trading_da_arming"] == pytest.approx(p["cumulato_da_arming"] - 1399.0)
|
|
|
|
|
|
def test_il_markdown_scorpora_e_lo_dice(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 667.0),
|
|
("2026-08-25T11:47:01+00:00", 2066.0)])
|
|
voce = _voce(pnl=J.pnl_giorno(db, date(2026, 8, 25)), giorno="2026-08-25")
|
|
md = J.rendi_markdown(db, voce)
|
|
assert "movimenti di capitale" in md and "trading" in md
|
|
righe = J.lettura(db, voce)
|
|
assert "movimento" in _ids(righe)
|
|
|
|
|
|
def test_un_crash_a_tutta_leva_resta_AMBIGUO_e_non_diventa_prelievo(db, monkeypatch):
|
|
# controllo positivo al contrario (M15): il mercato fa -15% nell'ora, il libro al tetto
|
|
# di leva (1.0x) puo' spiegare il -14% di equity -> NON e' un prelievo, e' dichiarato.
|
|
ix = pd.date_range("2026-08-20", "2026-08-27", freq="1h", tz="UTC")
|
|
px = pd.Series(100.0, index=ix)
|
|
px[ix >= "2026-08-25T11:00:00+00:00"] = 85.0
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: px)
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0),
|
|
("2026-08-25T11:47:01+00:00", 516.0)])
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
assert p["movimenti_giorno"][0]["classe"] == "ambiguo"
|
|
assert p["mov_giorno"] == 0.0 # niente scorporo
|
|
assert p["trading_giorno"] == pytest.approx(p["delta_equity"])
|
|
righe = J.lettura(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
amb = [r for r in righe if r["id"] == "movimento_ambiguo"]
|
|
assert amb and amb[0]["livello"] == "attenzione"
|
|
|
|
|
|
def test_sotto_soglia_niente_eventi_e_trading_uguale_al_delta(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0),
|
|
("2026-08-25T11:47:01+00:00", 612.0)]) # +2%: sotto EQUITY_JUMP_ALERT
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
assert p["movimenti_giorno"] == [] and p["mov_giorno"] == 0.0
|
|
assert p["trading_giorno"] == pytest.approx(p["delta_equity"])
|
|
md = J.rendi_markdown(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
assert "movimenti di capitale" not in md
|
|
|
|
|
|
def test_la_soglia_e_DERIVATA_dal_book_non_riscritta(db, monkeypatch):
|
|
# P1: se il rilevatore del book cambia soglia, il giornale la segue senza essere toccato.
|
|
from src.live import book as B
|
|
monkeypatch.setattr(B, "EQUITY_JUMP_ALERT", 0.50)
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0),
|
|
("2026-08-25T11:47:01+00:00", 780.0)]) # +30%: sotto la soglia alzata
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
assert p["movimenti_giorno"] == []
|
|
|
|
|
|
def test_mercato_non_misurabile_e_ambiguo_dichiarato_non_movimento(db, monkeypatch):
|
|
# P5: "non vedo" non e' "va tutto bene" — senza feed il salto NON si attribuisce.
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: None)
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 667.0),
|
|
("2026-08-25T11:47:01+00:00", 2066.0)])
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
e = p["movimenti_giorno"][0]
|
|
assert e["classe"] == "ambiguo" and e["mercato_max_pct"] is None
|
|
righe = J.lettura(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
assert any(r["id"] == "movimento_ambiguo" and "non e' misurabile" in r["testo"] for r in righe)
|
|
|
|
|
|
def test_la_concentrazione_non_conta_i_versamenti_come_pnl(db, monkeypatch):
|
|
# 30 giorni piatti a 600, poi un versamento da 1400: senza scorporo la regola direbbe
|
|
# "tutto il P&L viene dagli ultimi 7 giorni" su un P&L che non esiste.
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
punti = [(f"2026-07-{g:02d}T12:00:00+00:00", 600.0) for g in range(1, 32)]
|
|
punti += [(f"2026-08-{g:02d}T12:00:00+00:00", 600.0) for g in range(1, 25)]
|
|
punti += [("2026-08-25T11:47:01+00:00", 2000.0)]
|
|
_serie_equity(db, punti)
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
righe = J.lettura(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
conc = [r for r in righe if r["id"] == "concentrazione"]
|
|
assert conc == [] # il cumulato di TRADING e' 0: la regola deve tacere
|
|
|
|
|
|
# =============================================================================================
|
|
# DECISIONI PARCHEGGIATE SU UNA SOGLIA DI CAPITALE
|
|
# =============================================================================================
|
|
# Il 2026-08-28 la decisione «GTAA01: tenere o bloccare» e' stata rinviata a $15k di book, che e'
|
|
# la soglia sotto la quale lo sleeve non e' deployabile. Una decisione parcheggiata su un numero
|
|
# che nessuno sorveglia e' parcheggiata per sempre (N9): questi test blindano la sorveglianza.
|
|
|
|
def test_sotto_soglia_non_dice_niente():
|
|
assert J.soglie_superate(2_068.0) == []
|
|
|
|
|
|
def test_la_soglia_gtaa_e_15k_e_scatta_quando_l_equity_arriva():
|
|
soglie = [s for s, _ in J.SOGLIE_CAPITALE]
|
|
assert 15_000.0 in soglie, "la soglia di GTAA01 non e' piu' dichiarata"
|
|
assert J.soglie_superate(14_999.99) == []
|
|
assert [s for s, _ in J.soglie_superate(15_000.0)] == [15_000.0]
|
|
|
|
|
|
def test_le_soglie_si_accumulano_e_restano_ordinate():
|
|
"""A $20k si devono presentare ENTRAMBE: la seconda non cancella la prima."""
|
|
assert [s for s, _ in J.soglie_superate(20_000.0)] == [15_000.0, 20_000.0]
|
|
|
|
|
|
def test_equity_non_leggibile_non_e_una_soglia_superata():
|
|
"""P5: «non vedo» non e' «va tutto bene» — e non e' nemmeno «la soglia e' arrivata»."""
|
|
assert J.soglie_superate(None) == []
|
|
|
|
|
|
def test_ogni_soglia_dice_COSA_decidere_non_solo_che_e_arrivata():
|
|
"""P4: un'allerta risponde a due domande. Una riga che dice solo «sei a $15k» non serve."""
|
|
for _, testo in J.SOGLIE_CAPITALE:
|
|
assert len(testo) > 80 and "**" in testo, "la soglia non dice quale decisione sblocca"
|
|
|
|
|
|
# ------------------------------------------------- TWR: il rendimento che non e' un bonifico
|
|
# Debito #14 (2026-09-01): `trades_db --report` stampava `e1/e0-1` sulla serie grezza, «+243%»
|
|
# per il 96,3% un versamento. La riparazione (`movimenti_capitale`) viveva qui e non aveva
|
|
# attraversato il confine fra i due lettori. `rendimento_twr` e' la funzione che ENTRAMBI chiamano.
|
|
|
|
def test_twr_spezza_sul_versamento_e_il_grezzo_no(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
# +5% di trading: SOTTO la soglia del rilevatore (0,10) — un +10% con mercato piatto
|
|
# sarebbe esso stesso un "movimento", ed e' il limite dichiarato (D5) del classificatore.
|
|
_serie_equity(db, [("2026-08-24T10:47:01+00:00", 600.0),
|
|
("2026-08-25T10:47:01+00:00", 630.0), # +5% di trading
|
|
("2026-08-25T11:47:01+00:00", 2030.0), # +1400: versamento
|
|
("2026-08-26T10:47:01+00:00", 2009.7)]) # -1% di trading
|
|
r = J.rendimento_twr(db, "2026-08-26T23:59:59+00:00")
|
|
assert r["twr"] == pytest.approx(1.05 * 0.99 - 1.0) # +3,95%
|
|
assert r["certi"] == pytest.approx(1400.0) and r["ambigui"] == 0.0
|
|
assert [round(s["ret"], 4) for s in r["segmenti"]] == [0.05, -0.01]
|
|
assert r["trading"] == pytest.approx(2009.7 - 600.0 - 1400.0) # +9,7
|
|
grezzo = 2009.7 / 600.0 - 1.0 # +235%: il numero sbagliato
|
|
assert abs(r["twr"] - grezzo) > 2.0
|
|
|
|
|
|
def test_twr_senza_movimenti_e_il_rendimento_semplice(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-24T10:47:01+00:00", 600.0),
|
|
("2026-08-25T10:47:01+00:00", 630.0)])
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert r["twr"] == pytest.approx(0.05) and len(r["segmenti"]) == 1
|
|
assert r["trading"] == pytest.approx(30.0) and r["eventi"] == []
|
|
|
|
|
|
def test_twr_un_ambiguo_non_spezza_e_resta_dichiarato(db, monkeypatch):
|
|
# stesso caso del test sul crash a tutta leva: -14% con il mercato a -15% -> ambiguo
|
|
ix = pd.date_range("2026-08-20", "2026-08-27", freq="1h", tz="UTC")
|
|
px = pd.Series(100.0, index=ix)
|
|
px[ix >= "2026-08-25T11:00:00+00:00"] = 85.0
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: px)
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0),
|
|
("2026-08-25T11:47:01+00:00", 516.0)])
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert r["twr"] == pytest.approx(-0.14) and len(r["segmenti"]) == 1
|
|
assert r["certi"] == 0.0 and r["ambigui"] == pytest.approx(-84.0)
|
|
assert r["trading"] == pytest.approx(-84.0) # niente scorporo
|
|
|
|
|
|
def test_twr_serie_vuota_e_None_con_motivo_non_zero(db):
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert r["twr"] is None and r["motivo"] and r["trading"] is None
|
|
|
|
|
|
def test_twr_riproduce_il_numero_del_diario_01_09(db, monkeypatch):
|
|
# M23: prima di pubblicare un numero nuovo, far riprodurre alla macchina quello vecchio.
|
|
# Diario 2026-09-01-stato-trades: 598,06 -> 667,49 | +1.399,39 | 2.066,88 -> 2.051,84 = +10,80%
|
|
ix = pd.date_range("2026-06-20", "2026-09-02", freq="1h", tz="UTC")
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: pd.Series(100.0, index=ix))
|
|
# una lettura intermedia: nella serie vera ci sono ~1.500 letture orarie fra le due date,
|
|
# e nessun salto di trading fra due consecutive tocca la soglia del rilevatore
|
|
_serie_equity(db, [("2026-06-23T22:00:00+00:00", 598.06),
|
|
("2026-07-20T10:47:01+00:00", 630.00),
|
|
("2026-08-25T10:47:01+00:00", 667.49),
|
|
("2026-08-25T11:47:01+00:00", 2066.88),
|
|
("2026-09-01T16:47:01+00:00", 2051.84)])
|
|
r = J.rendimento_twr(db, "2026-09-01T23:59:59+00:00")
|
|
assert round(100 * r["twr"], 2) == 10.80
|
|
assert r["certi"] == pytest.approx(1399.39)
|
|
|
|
|
|
def test_twr_a_base_zero_e_None_anche_sul_trading(db, monkeypatch):
|
|
# revisione 02/09: a base zero `movimenti_capitale` non vede il salto 0 -> X del primo
|
|
# versamento, quindi `trading = e1-e0-certi` varrebbe l'intero conto. Tre stati, non due.
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-24T09:47:01+00:00", 0.0),
|
|
("2026-08-24T10:47:01+00:00", 598.06),
|
|
("2026-08-25T10:47:01+00:00", 610.0)])
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert r["twr"] is None and r["trading"] is None and "base" in r["motivo"]
|
|
|
|
|
|
def test_twr_due_movimenti_consecutivi_non_producono_segmenti_a_lunghezza_zero(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-24T10:47:01+00:00", 600.0),
|
|
("2026-08-24T11:47:01+00:00", 2000.0), # movimento nel PRIMO intervallo
|
|
("2026-08-24T12:47:01+00:00", 3400.0), # movimento consecutivo
|
|
("2026-08-25T10:47:01+00:00", 3434.0)]) # +1% di trading
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert len(r["segmenti"]) == 1 and r["segmenti"][0]["ret"] == pytest.approx(0.01)
|
|
assert r["twr"] == pytest.approx(0.01)
|
|
assert all(s["da"] != s["a"] for s in r["segmenti"])
|
|
assert r["certi"] == pytest.approx(1400.0 + 1400.0)
|
|
|
|
|
|
def test_twr_senza_tempo_a_mercato_e_zero_non_None(db, monkeypatch):
|
|
# due letture, la seconda e' un versamento: nessun segmento, rendimento 0, dichiarato numero
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-24T10:47:01+00:00", 600.0),
|
|
("2026-08-24T11:47:01+00:00", 2000.0)])
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert r["twr"] == 0.0 and r["segmenti"] == [] and r["trading"] == pytest.approx(0.0)
|
|
|
|
|
|
# ------------------------------------------------- mercato NON misurabile (revisione 02/09)
|
|
# Il feed 1h certificato si ferma alle 00:00 (rebuild alle 00:30): per il resto del giorno
|
|
# `asof` restituiva la STESSA barra per le due letture, max_mkt 0, e qualunque calo >=10%
|
|
# diventava un "movimento" — un crash stampato come prelievo. E con il feed assente tutto
|
|
# era "ambiguo", certi=0, e il TWR tornava a essere e1/e0-1 sotto l'etichetta sbagliata.
|
|
|
|
def test_un_salto_oltre_la_fine_del_feed_e_ambiguo_con_la_ragione(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto()) # copre fino al 27/08
|
|
_serie_equity(db, [("2026-08-28T10:47:01+00:00", 2000.0),
|
|
("2026-08-28T11:47:01+00:00", 1760.0)]) # -12% dopo la fine
|
|
p = J.pnl_giorno(db, date(2026, 8, 28))
|
|
ev = p["movimenti_giorno"]
|
|
assert ev[0]["classe"] == "ambiguo" and ev[0]["mercato_max_pct"] is None
|
|
assert "fermo a 2026-08-27" in ev[0]["mercato_nota"]
|
|
assert p["mov_giorno"] == 0.0 # NON scorporato
|
|
assert p["trading_da_arming"] is None and p["twr_da_arming"] is None
|
|
assert "non classificabile" in p["twr_motivo"]
|
|
|
|
|
|
def test_con_il_feed_assente_il_twr_e_None_non_il_grezzo(db, monkeypatch):
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: None)
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 667.0),
|
|
("2026-08-25T11:47:01+00:00", 2066.0)])
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert r["twr"] is None and r["trading"] is None
|
|
assert "non leggibile" in r["motivo"]
|
|
assert r["eventi"][0]["classe"] == "ambiguo"
|
|
|
|
|
|
def test_il_giornale_e_il_report_usano_la_stessa_funzione(db, monkeypatch):
|
|
# P1, revisione 02/09: pnl_giorno non rifa' piu' `cum - certi`; prende trading e TWR da
|
|
# rendimento_twr, e la pagina lo stampa accanto al cumulato
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-24T10:47:01+00:00", 600.0),
|
|
("2026-08-25T10:47:01+00:00", 630.0),
|
|
("2026-08-25T11:47:01+00:00", 2030.0),
|
|
("2026-08-25T12:47:01+00:00", 2009.7)])
|
|
p = J.pnl_giorno(db, date(2026, 8, 25))
|
|
r = J.rendimento_twr(db, "2026-08-25T23:59:59+00:00")
|
|
assert p["trading_da_arming"] == pytest.approx(r["trading"])
|
|
assert p["twr_da_arming"] == pytest.approx(r["twr"]) == pytest.approx(1.05 * 0.99 - 1)
|
|
md = J.rendi_markdown(db, _voce(pnl=p, giorno="2026-08-25"))
|
|
assert "TWR **+3.95%**" in md
|
|
|
|
|
|
def test_la_leva_del_classificatore_include_la_scala_e_il_tetto_di_codice(db, monkeypatch, tmp_path):
|
|
# P1: la chiave di scala del 01/09 entra nel bound da sola; e il fallback e' il tetto di CODICE
|
|
from src.live import book as B
|
|
cfg = tmp_path / "live.json"
|
|
cfg.write_text(json.dumps({"max_notional_per_asset_frac": 0.5, "book_scale_k": 1.25}))
|
|
monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path); (tmp_path / "config").mkdir()
|
|
cfg.rename(tmp_path / "config" / "live.json")
|
|
monkeypatch.setattr(J, "_px_1h", lambda a: _px_piatto())
|
|
_serie_equity(db, [("2026-08-25T10:47:01+00:00", 600.0), ("2026-08-25T11:47:01+00:00", 2000.0)])
|
|
mv = J.movimenti_capitale(db, "2026-08-25T23:59:59+00:00")
|
|
assert mv["leva_tetto"] == pytest.approx(min(0.5 * 2 * 1.25, B.LEVA_LORDA_MAX))
|