tradesdb: fill cron_log arricchiti con ora, prezzo e pos_after dal jsonl (#40)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

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 —
+    # VERA per-fill (docstring di tradesdb.py). Un conteggio diverso in quella finestra segnala
-    # l'arricchimento dal jsonl e' TradingBook#11, non qui. Un conteggio diverso in quella
+    # che una delle due fonti sta perdendo fill — questo controllo resta indipendente
-    # finestra segnala che una delle due fonti sta perdendo fill. Notify UNA volta per episodio:
+    # dall'arricchimento sopra (che abbina riga per riga, non conta): un conteggio uguale non
-    # marcatore in `meta` (stessa regola di fee_watch #37 B9 — scritto solo se l'invio riesce,
+    # esclude un abbinamento ambiguo, e viceversa. Notify UNA volta per episodio: marcatore in
-    # azzerato quando i conteggi tornano uguali).
+    # `meta` (stessa regola di fee_watch #37 B9 — scritto solo se l'invio riesce, azzerato
-    righe_jsonl = T.parse_executions_jsonl(T.EXEC_JSONL.read_text()) if T.EXEC_JSONL.exists() else []
+    # 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:

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,
+campo `pos_after` (letto dal reconcile subito dopo il rebalance, #39 — None se quella lettura
-e non viene riscritta nel jsonl — vedi il commento in `_registra_fill`). E' quindi, dal 24/08 in
+solleva, o se il passo e' un `RebalanceParziale` a meta'). E' quindi, dal 24/08 in poi, la fonte
-poi, la fonte con l'ora PIU' precisa per-fill — ma resta usata solo da `reconcile()`: arricchire
+con l'ora PIU' precisa per-fill. Dal 2026-09-27 (#40) `sync()` la usa anche per ARRICCHIRE il
-`fills` con ts/prezzo/pos_after dal jsonl e' uno SVILUPPO non fatto qui (TradingBook#11).
+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:

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