From bebfa63213fb820e94d19b9a852879449b18df1c Mon Sep 17 00:00:00 2001 From: "Adriano (goldrake)" Date: Sun, 27 Sep 2026 17:51:15 +0000 Subject: [PATCH] tradesdb: fill cron_log arricchiti con ora, prezzo e pos_after dal jsonl (#40) Co-Authored-By: Claude Sonnet 5 --- scripts/live/trades_db.py | 24 +++-- src/live/tradesdb.py | 93 +++++++++++++++++- tests/test_tradesdb_arricchisci.py | 147 +++++++++++++++++++++++++++++ 3 files changed, 252 insertions(+), 12 deletions(-) create mode 100644 tests/test_tradesdb_arricchisci.py diff --git a/scripts/live/trades_db.py b/scripts/live/trades_db.py index 1ba1b95..d2646f9 100644 --- a/scripts/live/trades_db.py +++ b/scripts/live/trades_db.py @@ -26,6 +26,14 @@ def sync(verbose: bool = True) -> dict: fills = [f for r in runs for f in r.fills] n_f = T.upsert_fills(con, fills) n_e = T.upsert_equity(con, [(r.ts_utc, r.equity, "cron_book") for r in runs if r.equity is not None]) + + # #40 (TradingBook#11): arricchimento PRIMA dei round-trip, cosi' il FIFO vede gia' i prezzi + # pieni e le ore vere delle righe abbinate al jsonl, non quelle (arrotondate, di inizio giro) + # di cron_book.log. `righe_jsonl` si legge una volta sola e serve anche al controllo di + # conteggio (#37 C11) qui sotto. + righe_jsonl = T.parse_executions_jsonl(T.EXEC_JSONL.read_text()) if T.EXEC_JSONL.exists() else [] + n_arricchiti, n_ambigui = T.arricchisci_da_jsonl(con, righe_jsonl) + n_r = T.rebuild_roundtrips(con) if runs: T.set_meta(con, "ultimo_giro", runs[-1].ts_utc) @@ -34,12 +42,12 @@ def sync(verbose: bool = True) -> dict: tot_e = con.execute("SELECT COUNT(*) c FROM equity").fetchone()["c"] # issue #37 (C11). Dal 2026-08-24T01:07Z (T.CUTOFF_TS_VERO) book_executions.jsonl porta l'ora - # VERA per-fill (docstring di tradesdb.py), ma `sync()` popola il DB SOLO da cron_book.log — - # l'arricchimento dal jsonl e' TradingBook#11, non qui. Un conteggio diverso in quella - # finestra segnala che una delle due fonti sta perdendo fill. Notify UNA volta per episodio: - # marcatore in `meta` (stessa regola di fee_watch #37 B9 — scritto solo se l'invio riesce, - # azzerato quando i conteggi tornano uguali). - righe_jsonl = T.parse_executions_jsonl(T.EXEC_JSONL.read_text()) if T.EXEC_JSONL.exists() else [] + # VERA per-fill (docstring di tradesdb.py). Un conteggio diverso in quella finestra segnala + # che una delle due fonti sta perdendo fill — questo controllo resta indipendente + # dall'arricchimento sopra (che abbina riga per riga, non conta): un conteggio uguale non + # esclude un abbinamento ambiguo, e viceversa. Notify UNA volta per episodio: marcatore in + # `meta` (stessa regola di fee_watch #37 B9 — scritto solo se l'invio riesce, azzerato + # quando i conteggi tornano uguali). n_jsonl_dal = sum(1 for r in righe_jsonl if str(r.get("ts_utc", "")) >= T.CUTOFF_TS_VERO) n_db_dal = con.execute("SELECT COUNT(*) c FROM fills WHERE ts_utc >= ?", (T.CUTOFF_TS_VERO,)).fetchone()["c"] @@ -58,9 +66,11 @@ def sync(verbose: bool = True) -> dict: if verbose: print(f" fills: +{n_f} (totale {tot}) | equity: +{n_e} (totale {tot_e}) | round-trip {n_r}") + print(f" arricchiti dal jsonl: {n_arricchiti} | ambigui: {n_ambigui}") con.close() return dict(nuovi_fill=n_f, nuovi_equity=n_e, roundtrip=n_r, totale_fill=tot, - jsonl_dal_24_08=n_jsonl_dal, db_dal_24_08=n_db_dal) + jsonl_dal_24_08=n_jsonl_dal, db_dal_24_08=n_db_dal, + arricchiti=n_arricchiti, ambigui=n_ambigui) def reconcile() -> None: diff --git a/src/live/tradesdb.py b/src/live/tradesdb.py index 17f56f0..a8c2ae4 100644 --- a/src/live/tradesdb.py +++ b/src/live/tradesdb.py @@ -13,10 +13,13 @@ il perimetro gia' coperto dal backup rotativo della VPS. 19 righe su 19 a `00:00:00`. Dal 2026-08-24 01:07Z `book_execute.py` (`_registra_fill`) scrive l'ora VERA del fill (`datetime.now(timezone.utc)` al momento dell'ordine, non piu' la barra), UNA per fill — con il prezzo a piena precisione (non arrotondato come nella stampa del log) e il -campo `pos_after` (oggi sempre `None` qui: la posizione post-rebalance si legge dopo, best-effort, -e non viene riscritta nel jsonl — vedi il commento in `_registra_fill`). E' quindi, dal 24/08 in -poi, la fonte con l'ora PIU' precisa per-fill — ma resta usata solo da `reconcile()`: arricchire -`fills` con ts/prezzo/pos_after dal jsonl e' uno SVILUPPO non fatto qui (TradingBook#11). +campo `pos_after` (letto dal reconcile subito dopo il rebalance, #39 — None se quella lettura +solleva, o se il passo e' un `RebalanceParziale` a meta'). E' quindi, dal 24/08 in poi, la fonte +con l'ora PIU' precisa per-fill. Dal 2026-09-27 (#40) `sync()` la usa anche per ARRICCHIRE il +DB: `arricchisci_da_jsonl()` gira dopo l'inserimento e prima dei round-trip, e le righe dal 24/08 +finiscono con `ts_utc`/`price` dal fill vero e `ts_source='jsonl'` invece che `'cron_log'` (e +`pos_after` quando il jsonl lo porta) — SOLO se l'abbinamento e' univoco, altrimenti resta com'e' +e conta come ambiguo. FONTI, in ordine di autorita' — e nessuna delle tre e' completa da sola ---------------------------------------------------------------------- @@ -43,7 +46,7 @@ import json import re import sqlite3 from dataclasses import dataclass, field -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] @@ -348,6 +351,86 @@ def upsert_fills(con: sqlite3.Connection, fills: list[Fill]) -> int: return n +def _ts_utc(s) -> datetime | None: + """Parsa un timestamp ISO in un datetime AWARE utc. None se illeggibile. + + Le 19 righe scritte prima del fix #35 A9 (barra di segnale, `pd.Timestamp` in stampa: es. + "2026-07-08 00:00:00", spazio invece di 'T', nessun offset) sono NAIVE — senza questa + normalizzazione, confrontarle con un ts_utc del DB (sempre aware, +00:00) solleva + `TypeError: can't compare offset-naive and offset-aware datetimes` (misurato sui dati veri, + 27/09/2026). Sono tutte precedenti a CUTOFF_TS_VERO: qui si limita a non farle esplodere, + non a farle abbinare — restano fuori dalla finestra di qualunque candidato dal 24/08.""" + if not s: + return None + try: + ts = datetime.fromisoformat(str(s)) + except ValueError: + return None + return ts.replace(tzinfo=timezone.utc) if ts.tzinfo is None else ts + + +def arricchisci_da_jsonl(con: sqlite3.Connection, righe_jsonl: list[dict]) -> tuple[int, int]: + """Arricchisce le righe `ts_source='cron_log'` con l'ora vera, il prezzo e `pos_after` letti + da `book_executions.jsonl` (#40, TradingBook#11: lo sviluppo lasciato non fatto dal + commento sopra). NON tocca `fill_id`: e' la chiave della riga, calcolata all'INSERT da + ts_utc/asset/side/qty/price ORIGINALI (cron_log) — qui si fa un UPDATE per `fill_id`, che + cambia solo le colonne, mai la chiave con cui la riga e' indicizzata. + + Candidate: `ts_source='cron_log'` e `ts_utc >= CUTOFF_TS_VERO` (prima di quella soglia il + jsonl portava la barra di segnale, non l'ora vera — non ha una controparte comparabile). + Abbinamento a una riga jsonl: stesso asset, stesso side, |qty - filled| < 1e-9, e ts jsonl in + [ts_utc DB, ts_utc DB + 15 minuti] (il DB ha l'ora di INIZIO GIRO, il jsonl l'ora del fill, + che viene dopo). Esattamente UN candidato -> UPDATE; zero o piu' di uno -> non si tocca (non + si indovina quale, o se ce n'e' uno) e conta come ambiguo. Un record jsonl assegnato non si + riusa: due fill identici nello stesso giro non finiscono abbinati alla stessa riga. + Idempotente: una riga gia' `ts_source='jsonl'` non e' piu' candidata al giro successivo. + + Ritorna (righe_aggiornate, righe_ambigue).""" + candidati = con.execute( + "SELECT fill_id, ts_utc, asset, side, qty FROM fills " + "WHERE ts_source='cron_log' AND ts_utc >= ? ORDER BY ts_utc, asset", + (CUTOFF_TS_VERO,)).fetchall() + usati: set[int] = set() # indici in righe_jsonl gia' assegnati a un fill_id + aggiornate = 0 + ambigue = 0 + for riga in candidati: + ts_db = _ts_utc(riga["ts_utc"]) + if ts_db is None: + ambigue += 1 # ts illeggibile: non e' un abbinamento, e' un dato rotto + continue + fine = ts_db + timedelta(minutes=15) + trovati = [] + for i, r in enumerate(righe_jsonl): + if i in usati: + continue + if r.get("asset") != riga["asset"] or r.get("side") != riga["side"]: + continue + filled = r.get("filled") + if filled is None or abs(float(filled) - riga["qty"]) >= 1e-9: + continue + ts_j = _ts_utc(r.get("ts_utc")) + if ts_j is None or not (ts_db <= ts_j <= fine): + continue + trovati.append(i) + if len(trovati) != 1: + ambigue += 1 + continue + i = trovati[0] + r = righe_jsonl[i] + usati.add(i) + pos_after = r.get("pos_after") + if pos_after is None: + con.execute("UPDATE fills SET ts_utc=?, ts_source='jsonl', price=? WHERE fill_id=?", + (r["ts_utc"], r["price"], riga["fill_id"])) + else: + con.execute( + "UPDATE fills SET ts_utc=?, ts_source='jsonl', price=?, pos_after=? WHERE fill_id=?", + (r["ts_utc"], r["price"], pos_after, riga["fill_id"])) + aggiornate += 1 + con.commit() + return aggiornate, ambigue + + def upsert_equity(con: sqlite3.Connection, punti: list[tuple[str, float, str]]) -> int: n = 0 for ts, v, src in punti: diff --git a/tests/test_tradesdb_arricchisci.py b/tests/test_tradesdb_arricchisci.py new file mode 100644 index 0000000..5044e16 --- /dev/null +++ b/tests/test_tradesdb_arricchisci.py @@ -0,0 +1,147 @@ +"""tradesdb.arricchisci_da_jsonl (#40, TradingBook#11): le righe `ts_source='cron_log'` dal +24/08 (`CUTOFF_TS_VERO`) prendono l'ora vera, il prezzo pieno e `pos_after` dal jsonl, SOLO +quando l'abbinamento (asset, side, qty, finestra di 15 minuti) e' univoco. `fill_id` e' la +chiave della riga: l'arricchimento aggiorna le colonne con un UPDATE per `fill_id`, mai la +chiave stessa. + +Nessun test tocca la rete o il libro vero: DB e righe jsonl sono finti, in tmp_path. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from src.live import tradesdb as T # noqa: E402 + +# ts di riferimento: subito dopo la soglia, cosi' i fill sono sempre candidati. +T0 = "2026-08-24T02:00:00+00:00" + + +@pytest.fixture() +def db(tmp_path): + con = T.connect(tmp_path / "t.db") + yield con + con.close() + + +def _f(ts, asset="BTC", side="buy", qty=0.005, price=70000.0, fee=0.02): + return T.Fill(ts_utc=ts, asset=asset, side=side, qty=qty, price=price, fee=fee) + + +def _riga_jsonl(ts, asset="BTC", side="buy", filled=0.005, price=70123.4, pos_after=None): + return dict(ts_utc=ts, asset=asset, side=side, filled=filled, price=price, pos_after=pos_after) + + +def test_abbinamento_unico_aggiorna_e_non_cambia_fill_id(db): + fill = _f(T0) + T.upsert_fills(db, [fill]) + fill_id_prima = fill.fill_id + + righe = [_riga_jsonl("2026-08-24T02:05:00+00:00", pos_after=555.0)] # +5 min, in finestra + aggiornate, ambigue = T.arricchisci_da_jsonl(db, righe) + assert (aggiornate, ambigue) == (1, 0) + + riga = db.execute("SELECT * FROM fills WHERE fill_id=?", (fill_id_prima,)).fetchone() + assert riga is not None, "il fill_id originale resta la chiave: l'UPDATE non lo cambia" + assert riga["ts_source"] == "jsonl" + assert riga["ts_utc"] == "2026-08-24T02:05:00+00:00" + assert riga["price"] == 70123.4 + assert riga["pos_after"] == 555.0 + # nessuna riga in piu' o in meno: l'arricchimento aggiorna, non inserisce. + assert db.execute("SELECT COUNT(*) c FROM fills").fetchone()["c"] == 1 + + +def test_pos_after_none_nel_jsonl_non_sovrascrive_quello_gia_nel_db(db): + """(#40, punto 1) `pos_after=None` nella riga jsonl non e' un valore: la colonna resta + quella che aveva (qui: None dal cron_log, ma la regola e' 'non toccare', non 'azzerare').""" + fill = _f(T0) + T.upsert_fills(db, [fill]) + + righe = [_riga_jsonl("2026-08-24T02:05:00+00:00", pos_after=None)] + aggiornate, ambigue = T.arricchisci_da_jsonl(db, righe) + assert (aggiornate, ambigue) == (1, 0) + + riga = db.execute("SELECT * FROM fills").fetchone() + assert riga["ts_source"] == "jsonl" and riga["price"] == 70123.4 + assert riga["pos_after"] is None + + +def test_due_jsonl_compatibili_e_ambiguo_non_si_tocca(db): + fill = _f(T0) + T.upsert_fills(db, [fill]) + + righe = [_riga_jsonl("2026-08-24T02:03:00+00:00", price=70100.0), + _riga_jsonl("2026-08-24T02:07:00+00:00", price=70200.0)] # entrambe in finestra + aggiornate, ambigue = T.arricchisci_da_jsonl(db, righe) + assert (aggiornate, ambigue) == (0, 1) + + riga = db.execute("SELECT * FROM fills").fetchone() + assert riga["ts_source"] == "cron_log" # non toccata + assert riga["ts_utc"] == T0 + assert riga["price"] == 70000.0 + + +def test_secondo_giro_zero_aggiornati_idempotente(db): + fill = _f(T0) + T.upsert_fills(db, [fill]) + righe = [_riga_jsonl("2026-08-24T02:05:00+00:00", pos_after=555.0)] + + assert T.arricchisci_da_jsonl(db, righe) == (1, 0) + # la riga e' gia' ts_source='jsonl': non e' piu' candidata, anche con le stesse righe jsonl. + assert T.arricchisci_da_jsonl(db, righe) == (0, 0) + assert db.execute("SELECT COUNT(*) c FROM fills WHERE ts_source='jsonl'").fetchone()["c"] == 1 + + +def test_riga_prima_del_cutoff_non_e_candidata(db): + ts_prima = "2026-08-10T00:00:00+00:00" # prima di CUTOFF_TS_VERO (24/08) + fill = _f(ts_prima) + T.upsert_fills(db, [fill]) + + righe = [_riga_jsonl("2026-08-10T00:05:00+00:00", pos_after=1.0)] # abbinamento perfetto + aggiornate, ambigue = T.arricchisci_da_jsonl(db, righe) + assert (aggiornate, ambigue) == (0, 0) # non e' ambigua: non e' nemmeno candidata + + riga = db.execute("SELECT * FROM fills").fetchone() + assert riga["ts_source"] == "cron_log" and riga["ts_utc"] == ts_prima and riga["price"] == 70000.0 + + +def test_fuori_dalla_finestra_dei_15_minuti_non_abbina(db): + fill = _f(T0) + T.upsert_fills(db, [fill]) + + righe = [_riga_jsonl("2026-08-24T02:16:00+00:00", pos_after=1.0)] # +16 min: fuori finestra + assert T.arricchisci_da_jsonl(db, righe) == (0, 1) + + riga = db.execute("SELECT * FROM fills").fetchone() + assert riga["ts_source"] == "cron_log" + + +def test_qty_diversa_non_abbina(db): + fill = _f(T0, qty=0.005) + T.upsert_fills(db, [fill]) + + righe = [_riga_jsonl("2026-08-24T02:05:00+00:00", filled=0.006, pos_after=1.0)] + assert T.arricchisci_da_jsonl(db, righe) == (0, 1) + + +def test_record_jsonl_usato_una_volta_non_si_riusa(db): + """Due fill IDENTICI nello stesso giro (stesso asset/side/qty, ts diversi) non devono + finire abbinati alla stessa riga jsonl: la prima riga jsonl la prende un solo fill_id.""" + f1 = _f("2026-08-24T02:00:00+00:00") + f2 = _f("2026-08-24T02:01:00+00:00") # ts diverso -> fill_id diverso, stesso asset/side/qty + T.upsert_fills(db, [f1, f2]) + assert db.execute("SELECT COUNT(*) c FROM fills").fetchone()["c"] == 2 + + righe = [_riga_jsonl("2026-08-24T02:05:00+00:00", pos_after=1.0)] # UNA sola riga jsonl + aggiornate, ambigue = T.arricchisci_da_jsonl(db, righe) + # un solo candidato jsonl per due righe DB candidate: la prima (ordinata per ts_utc) lo + # prende, la seconda resta senza candidati liberi -> ambigua (zero disponibili). + assert aggiornate == 1 + assert ambigue == 1 + presi = db.execute("SELECT COUNT(*) c FROM fills WHERE ts_source='jsonl'").fetchone()["c"] + assert presi == 1