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

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-27 17:51:15 +00:00
parent ae887a3d98
commit bebfa63213
3 changed files with 252 additions and 12 deletions
+17 -7
View File
@@ -26,6 +26,14 @@ def sync(verbose: bool = True) -> dict:
fills = [f for r in runs for f in r.fills] fills = [f for r in runs for f in r.fills]
n_f = T.upsert_fills(con, 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]) 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) n_r = T.rebuild_roundtrips(con)
if runs: if runs:
T.set_meta(con, "ultimo_giro", runs[-1].ts_utc) 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"] 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 # 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_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 >= ?", n_db_dal = con.execute("SELECT COUNT(*) c FROM fills WHERE ts_utc >= ?",
(T.CUTOFF_TS_VERO,)).fetchone()["c"] (T.CUTOFF_TS_VERO,)).fetchone()["c"]
@@ -58,9 +66,11 @@ def sync(verbose: bool = True) -> dict:
if verbose: if verbose:
print(f" fills: +{n_f} (totale {tot}) | equity: +{n_e} (totale {tot_e}) | round-trip {n_r}") 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() con.close()
return dict(nuovi_fill=n_f, nuovi_equity=n_e, roundtrip=n_r, totale_fill=tot, 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: def reconcile() -> None:
+88 -5
View File
@@ -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 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), 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 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 FONTI, in ordine di autorita' — e nessuna delle tre e' completa da sola
---------------------------------------------------------------------- ----------------------------------------------------------------------
@@ -43,7 +46,7 @@ import json
import re import re
import sqlite3 import sqlite3
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, timezone from datetime import datetime, timedelta, timezone
from pathlib import Path from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[2] PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -348,6 +351,86 @@ def upsert_fills(con: sqlite3.Connection, fills: list[Fill]) -> int:
return n 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: def upsert_equity(con: sqlite3.Connection, punti: list[tuple[str, float, str]]) -> int:
n = 0 n = 0
for ts, v, src in punti: for ts, v, src in punti:
+147
View File
@@ -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