Adriano Dal Pastro
f5d9409213
Aggiunto il quarto livello della pagina, tenuto separato dagli altri tre: numeri -> misurati dal feed e dal DB Lettura -> regole deterministiche, ognuna col suo id Analisi -> questo: prosa di un modello, che puo' sbagliare Nota -> l'operatore NON scrive dentro `nota`, che era la richiesta letterale: quel campo e' dell'operatore, ed e' cio' che a rileggere il giornale fra sei mesi permette di sapere chi ha scritto cosa. L'agente ha `analisi`, marcato col modello, con l'ora e con l'esito del controllo sui numeri. Gira via `claude -p` (verificato con env -i che risponda nell'ambiente nudo di cron), una chiamata al giorno sul giorno CHIUSO, tolte le tool. Tre guardie, una per ogni modo in cui una prosa generata rovina un registro: - NUMERO INVENTATO: numeri_non_supportati() estrae ogni cifra dall'analisi e verifica che compaia in cio' che il modello ha ricevuto. Oltre tre numeri liberi l'analisi e' RIFIUTATA e la pagina resta senza. E' un controllo debole per costruzione, e lo dichiara: prende l'invenzione, non il ragionamento sbagliato. - COMMENTO DI SE': senza_analisi() toglie dalla pagina la sezione dell'agente prima di dargliela. Al primo giro reale il modello aveva letto la propria uscita precedente e prodotto un paragrafo sull'avviso che si era preso il giorno prima — un ciclo di retroazione che in poche settimane avrebbe riempito il giornale di meta-commento, e che nessun controllo automatico puo' distinguere da prosa valida. - ANALISI DI IERI SPACCIATA PER OGGI: se il modello non risponde, la pagina resta VUOTA e il perche' viene registrato (stato + motivo). Il silenzio non diventa continuita'. Corretto anche un falso positivo mio: il tripwire validava sulla sola pagina mentre il prompt include anche il blocco storico, quindi bocciava un'equity vera. Una guardia piu' stretta del contratto produce allarmi che si impara a ignorare. Ogni guardia ha un test in entrambe le direzioni. 695 test passano. Strategia, pesi, config INVARIATI. Nessun ordine. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
373 lines
14 KiB
Python
373 lines
14 KiB
Python
"""Libro di bordo del book live: DB dei trade ALLINEATO COL TEMPO + serie di equity.
|
|
|
|
PERCHE' ESISTE
|
|
--------------
|
|
`data/live/book_executions.jsonl` registra i fill, ma il suo `ts_utc` e' la data della
|
|
**barra di segnale** (`book_execute.py`: `pd.Timestamp(r['last_data'])`), non l'ora del fill:
|
|
19 righe su 19 a `00:00:00`. L'ora vera sta solo in `logs/cron_book.log` — che e' **gitignored,
|
|
non nel backup e ruotabile**. Quindi oggi la cronologia reale del libro live vive in un file
|
|
che una rotazione cancella, e nessuno se ne accorgerebbe.
|
|
|
|
Questo modulo MATERIALIZZA quella cronologia in `data/live/trades.db` (sqlite), che sta dentro
|
|
il perimetro gia' coperto dal backup rotativo della VPS.
|
|
|
|
FONTI, in ordine di autorita' — e nessuna delle tre e' completa da sola
|
|
----------------------------------------------------------------------
|
|
1. `logs/cron_book.log` -> ORA VERA del giro + contesto (equity, tp_frac, skh, target, posizione)
|
|
e la riga di fill `-> BUY 0.0008 @ $78,094.2 fee 0.02187 (OK)`.
|
|
2. `data/live/book_executions.jsonl` -> gli stessi fill con piu' cifre, ma senza ora.
|
|
3. venue (`DeribitRead.trade_history`) -> autorevole su `order_id` e timestamp in ms,
|
|
**ma tronca**: al 2026-08-23 ritorna 1 trade su BTC e 0 su ETH.
|
|
|
|
Le tre si INCROCIANO, non si sovrascrivono: `reconcile()` riporta le divergenze e non ripara
|
|
niente da solo (una riparazione silenziosa fra due fonti che non concordano e' un'invenzione).
|
|
|
|
TRE STATI, non due: un fill puo' essere `ok` (le fonti concordano), `solo-log`, `solo-jsonl`.
|
|
"Non lo vedo" non e' "non c'e'".
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
DB_PATH = PROJECT_ROOT / "data" / "live" / "trades.db"
|
|
CRON_LOG = PROJECT_ROOT / "logs" / "cron_book.log"
|
|
EXEC_JSONL = PROJECT_ROOT / "data" / "live" / "book_executions.jsonl"
|
|
|
|
# Data di armamento dell'esecuzione reale (CLAUDE.md, 2026-06-20).
|
|
ARMING = "2026-06-20"
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS fills (
|
|
fill_id TEXT PRIMARY KEY,
|
|
ts_utc TEXT NOT NULL,
|
|
ts_source TEXT NOT NULL,
|
|
bar_ts TEXT,
|
|
asset TEXT NOT NULL,
|
|
side TEXT NOT NULL,
|
|
qty REAL NOT NULL,
|
|
price REAL NOT NULL,
|
|
fee REAL NOT NULL,
|
|
action TEXT,
|
|
net_target REAL,
|
|
pos_before REAL,
|
|
pos_after REAL,
|
|
tp_frac REAL,
|
|
skh_sign INTEGER,
|
|
skh_entry REAL,
|
|
equity REAL,
|
|
order_id TEXT,
|
|
verified INTEGER,
|
|
stato TEXT NOT NULL DEFAULT 'ok'
|
|
);
|
|
CREATE INDEX IF NOT EXISTS ix_fills_ts ON fills(ts_utc);
|
|
CREATE TABLE IF NOT EXISTS roundtrips (
|
|
rt_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
asset TEXT NOT NULL,
|
|
qty REAL NOT NULL,
|
|
ts_in TEXT NOT NULL,
|
|
px_in REAL NOT NULL,
|
|
ts_out TEXT NOT NULL,
|
|
px_out REAL NOT NULL,
|
|
ore_tenuta REAL,
|
|
pnl_lordo REAL NOT NULL,
|
|
fee_quota REAL NOT NULL,
|
|
pnl_netto REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS equity (
|
|
ts_utc TEXT PRIMARY KEY,
|
|
equity REAL NOT NULL,
|
|
src TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS journal (
|
|
giorno TEXT PRIMARY KEY,
|
|
ts_scritto TEXT NOT NULL,
|
|
mercato TEXT,
|
|
libro TEXT,
|
|
pnl TEXT,
|
|
salute TEXT,
|
|
nota TEXT NOT NULL DEFAULT ''
|
|
);
|
|
CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT);
|
|
"""
|
|
|
|
# Colonne aggiunte dopo la prima versione dello schema. sqlite non ha ADD COLUMN IF NOT EXISTS:
|
|
# si guarda il PRAGMA. `analisi` e' dell'AGENTE, `nota` resta dell'operatore — due campi, perche'
|
|
# a rileggere il giornale fra sei mesi si deve sapere chi ha scritto cosa.
|
|
COLONNE_TARDIVE = {
|
|
"journal": [("analisi", "TEXT NOT NULL DEFAULT ''"),
|
|
("analisi_modello", "TEXT"),
|
|
("analisi_ts", "TEXT"),
|
|
("analisi_stato", "TEXT"),
|
|
("analisi_motivi", "TEXT")],
|
|
}
|
|
|
|
|
|
def _migra(con) -> None:
|
|
for tabella, colonne in COLONNE_TARDIVE.items():
|
|
presenti = {r[1] for r in con.execute(f"PRAGMA table_info({tabella})")}
|
|
for nome, tipo in colonne:
|
|
if nome not in presenti:
|
|
con.execute(f"ALTER TABLE {tabella} ADD COLUMN {nome} {tipo}")
|
|
con.commit()
|
|
|
|
|
|
# =============================================================================================
|
|
# parsing — puro, nessun I/O: prende testo, ritorna dati
|
|
# =============================================================================================
|
|
|
|
@dataclass
|
|
class Fill:
|
|
ts_utc: str
|
|
asset: str
|
|
side: str
|
|
qty: float
|
|
price: float
|
|
fee: float
|
|
ts_source: str = "cron_log"
|
|
bar_ts: str | None = None
|
|
action: str | None = None
|
|
net_target: float | None = None
|
|
pos_before: float | None = None
|
|
pos_after: float | None = None
|
|
tp_frac: float | None = None
|
|
skh_sign: int | None = None
|
|
skh_entry: float | None = None
|
|
equity: float | None = None
|
|
order_id: str | None = None
|
|
verified: int = 1
|
|
|
|
@property
|
|
def fill_id(self) -> str:
|
|
raw = f"{self.ts_utc}|{self.asset}|{self.side}|{self.qty:.10f}|{self.price:.6f}"
|
|
return hashlib.sha1(raw.encode()).hexdigest()[:16]
|
|
|
|
|
|
@dataclass
|
|
class Run:
|
|
"""Un giro orario di `book_execute`."""
|
|
ts_utc: str
|
|
equity: float | None = None
|
|
last_bar: str | None = None
|
|
feed_min: int | None = None
|
|
fills: list[Fill] = field(default_factory=list)
|
|
stato_asset: dict = field(default_factory=dict)
|
|
|
|
|
|
_RE_HEAD = re.compile(r"^===== (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z) cron_book =====$")
|
|
_RE_EQ = re.compile(r"conto reale\s*:\s*\$([\d,\.]+)")
|
|
_RE_BAR = re.compile(r"ultima barra\s*:\s*(\S+)")
|
|
_RE_FEED = re.compile(r"feed SKH\s*:\s*\w+ \((\d+) min\)")
|
|
_RE_SIG = re.compile(
|
|
r"^\s*(BTC|ETH) TP ([+-][\d\.]+) · SKH ([+-]\d+)\(([^)]*)\) -> net \$([+-][\d,]+) "
|
|
r"\| pos \$([+-][\d,]+) -> (.+?)\s*$")
|
|
_RE_FILL = re.compile(
|
|
r"^\s*-> (BUY|SELL) ([\d\.]+) @ \$([\d,\.]+) fee ([\d\.]+) \((OK|NON VERIFICATO[^)]*)\)")
|
|
|
|
|
|
def _num(s: str) -> float:
|
|
return float(s.replace(",", "").replace("$", ""))
|
|
|
|
|
|
def parse_cron_log(text: str) -> list[Run]:
|
|
"""Estrae i giri da `cron_book.log`. L'ORA del blocco e' l'ora vera del fill."""
|
|
runs: list[Run] = []
|
|
cur: Run | None = None
|
|
ultimo_asset: str | None = None
|
|
for line in text.splitlines():
|
|
m = _RE_HEAD.match(line)
|
|
if m:
|
|
cur = Run(ts_utc=m.group(1).replace("Z", "+00:00"))
|
|
runs.append(cur)
|
|
ultimo_asset = None
|
|
continue
|
|
if cur is None:
|
|
continue
|
|
if (m := _RE_EQ.search(line)) and cur.equity is None:
|
|
cur.equity = _num(m.group(1)); continue
|
|
if (m := _RE_BAR.search(line)) and cur.last_bar is None:
|
|
cur.last_bar = m.group(1); continue
|
|
if (m := _RE_FEED.search(line)) and cur.feed_min is None:
|
|
cur.feed_min = int(m.group(1)); continue
|
|
if (m := _RE_SIG.match(line)):
|
|
asset = m.group(1)
|
|
entry = None
|
|
sk = m.group(4)
|
|
if "@" in sk:
|
|
try:
|
|
entry = float(sk.split("@", 1)[1])
|
|
except ValueError:
|
|
entry = None
|
|
cur.stato_asset[asset] = dict(
|
|
tp_frac=float(m.group(2)), skh_sign=int(m.group(3)), skh_entry=entry,
|
|
net=_num(m.group(5)), pos=_num(m.group(6)), azione=m.group(7).strip())
|
|
ultimo_asset = asset
|
|
continue
|
|
if (m := _RE_FILL.match(line)) and ultimo_asset:
|
|
st = cur.stato_asset.get(ultimo_asset, {})
|
|
cur.fills.append(Fill(
|
|
ts_utc=cur.ts_utc, asset=ultimo_asset, side=m.group(1).lower(),
|
|
qty=float(m.group(2)), price=_num(m.group(3)), fee=float(m.group(4)),
|
|
bar_ts=cur.last_bar, action=st.get("azione"), net_target=st.get("net"),
|
|
pos_before=st.get("pos"), tp_frac=st.get("tp_frac"),
|
|
skh_sign=st.get("skh_sign"), skh_entry=st.get("skh_entry"),
|
|
equity=cur.equity, verified=1 if m.group(5) == "OK" else 0))
|
|
return runs
|
|
|
|
|
|
def parse_executions_jsonl(text: str) -> list[dict]:
|
|
return [json.loads(l) for l in text.splitlines() if l.strip()]
|
|
|
|
|
|
def reconcile(fills_log: list[Fill], righe_jsonl: list[dict], tol_px: float = 0.51) -> dict:
|
|
"""Incrocia le due fonti su (data, asset, side, qty). NON ripara: riporta.
|
|
|
|
`tol_px`: i prezzi del log sono arrotondati a 1 decimale in stampa -> confronto con
|
|
tolleranza, e la divergenza sopra tolleranza si REGISTRA invece di essere assorbita.
|
|
"""
|
|
def chiave(g, a, s, q):
|
|
return (g, a, s, round(float(q), 8))
|
|
|
|
ix_log: dict = {}
|
|
for f in fills_log:
|
|
ix_log.setdefault(chiave(f.ts_utc[:10], f.asset, f.side, f.qty), []).append(f)
|
|
ix_js: dict = {}
|
|
for r in righe_jsonl:
|
|
ix_js.setdefault(chiave(r["ts_utc"][:10], r["asset"], r["side"], r["filled"]), []).append(r)
|
|
|
|
ok, solo_log, solo_js, px_diversi = [], [], [], []
|
|
for k, gl in ix_log.items():
|
|
gj = ix_js.get(k, [])
|
|
for i, f in enumerate(gl):
|
|
if i < len(gj):
|
|
r = gj[i]
|
|
if abs(float(r["price"]) - f.price) > tol_px:
|
|
px_diversi.append((f, r))
|
|
ok.append((f, r))
|
|
else:
|
|
solo_log.append(f)
|
|
for k, gj in ix_js.items():
|
|
extra = len(gj) - len(ix_log.get(k, []))
|
|
for r in gj[max(0, len(gj) - extra):] if extra > 0 else []:
|
|
solo_js.append(r)
|
|
return dict(ok=ok, solo_log=solo_log, solo_jsonl=solo_js, prezzi_divergenti=px_diversi)
|
|
|
|
|
|
def fifo_roundtrips(fills: list[Fill]) -> tuple[list[dict], dict]:
|
|
"""Round-trip chiusi per FIFO + lotti residui. Le fee si allocano PRO-QUOTA sulla qty."""
|
|
from collections import deque, defaultdict
|
|
lotti: dict[str, deque] = defaultdict(deque)
|
|
rts: list[dict] = []
|
|
for f in sorted(fills, key=lambda x: (x.ts_utc, x.asset)):
|
|
fee_u = f.fee / f.qty if f.qty else 0.0
|
|
if f.side == "buy":
|
|
lotti[f.asset].append([f.qty, f.price, f.ts_utc, fee_u])
|
|
else:
|
|
resto = f.qty
|
|
while resto > 1e-12 and lotti[f.asset]:
|
|
lq, lp, lts, lfee_u = lotti[f.asset][0]
|
|
usa = min(resto, lq)
|
|
lordo = usa * (f.price - lp)
|
|
quota_fee = usa * (lfee_u + fee_u)
|
|
ore = (datetime.fromisoformat(f.ts_utc) - datetime.fromisoformat(lts)).total_seconds() / 3600
|
|
rts.append(dict(asset=f.asset, qty=usa, ts_in=lts, px_in=lp, ts_out=f.ts_utc,
|
|
px_out=f.price, ore_tenuta=ore, pnl_lordo=lordo,
|
|
fee_quota=quota_fee, pnl_netto=lordo - quota_fee))
|
|
lq -= usa; resto -= usa
|
|
if lq <= 1e-12:
|
|
lotti[f.asset].popleft()
|
|
else:
|
|
lotti[f.asset][0][0] = lq
|
|
aperti = {a: [dict(qty=q, price=p, ts=t) for q, p, t, _ in dq] for a, dq in lotti.items() if dq}
|
|
return rts, aperti
|
|
|
|
|
|
# =============================================================================================
|
|
# persistenza
|
|
# =============================================================================================
|
|
|
|
def connect(path: Path | str = DB_PATH) -> sqlite3.Connection:
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
con = sqlite3.connect(path)
|
|
con.row_factory = sqlite3.Row
|
|
con.executescript(SCHEMA)
|
|
_migra(con)
|
|
return con
|
|
|
|
|
|
def upsert_fills(con: sqlite3.Connection, fills: list[Fill]) -> int:
|
|
n = 0
|
|
for f in fills:
|
|
cur = con.execute(
|
|
"""INSERT INTO fills (fill_id, ts_utc, ts_source, bar_ts, asset, side, qty, price, fee,
|
|
action, net_target, pos_before, pos_after, tp_frac, skh_sign,
|
|
skh_entry, equity, order_id, verified, stato)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(fill_id) DO NOTHING""",
|
|
(f.fill_id, f.ts_utc, f.ts_source, f.bar_ts, f.asset, f.side, f.qty, f.price, f.fee,
|
|
f.action, f.net_target, f.pos_before, f.pos_after, f.tp_frac, f.skh_sign,
|
|
f.skh_entry, f.equity, f.order_id, f.verified, "ok"))
|
|
n += cur.rowcount
|
|
con.commit()
|
|
return n
|
|
|
|
|
|
def upsert_equity(con: sqlite3.Connection, punti: list[tuple[str, float, str]]) -> int:
|
|
n = 0
|
|
for ts, v, src in punti:
|
|
cur = con.execute("INSERT INTO equity (ts_utc, equity, src) VALUES (?,?,?) "
|
|
"ON CONFLICT(ts_utc) DO NOTHING", (ts, v, src))
|
|
n += cur.rowcount
|
|
con.commit()
|
|
return n
|
|
|
|
|
|
def rebuild_roundtrips(con: sqlite3.Connection) -> int:
|
|
"""Ricalcola da zero: i round-trip sono DERIVATI, mai inseriti a mano."""
|
|
righe = con.execute("SELECT * FROM fills ORDER BY ts_utc, asset").fetchall()
|
|
fills = [Fill(ts_utc=r["ts_utc"], asset=r["asset"], side=r["side"], qty=r["qty"],
|
|
price=r["price"], fee=r["fee"]) for r in righe]
|
|
rts, _ = fifo_roundtrips(fills)
|
|
con.execute("DELETE FROM roundtrips")
|
|
con.executemany(
|
|
"""INSERT INTO roundtrips (asset, qty, ts_in, px_in, ts_out, px_out, ore_tenuta,
|
|
pnl_lordo, fee_quota, pnl_netto)
|
|
VALUES (:asset,:qty,:ts_in,:px_in,:ts_out,:px_out,:ore_tenuta,
|
|
:pnl_lordo,:fee_quota,:pnl_netto)""", rts)
|
|
con.commit()
|
|
return len(rts)
|
|
|
|
|
|
def set_meta(con: sqlite3.Connection, k: str, v: str) -> None:
|
|
con.execute("INSERT INTO meta (k,v) VALUES (?,?) ON CONFLICT(k) DO UPDATE SET v=excluded.v", (k, v))
|
|
con.commit()
|
|
|
|
|
|
def get_meta(con: sqlite3.Connection, k: str, default=None):
|
|
r = con.execute("SELECT v FROM meta WHERE k=?", (k,)).fetchone()
|
|
return r["v"] if r else default
|
|
|
|
|
|
def stato_aperto(con: sqlite3.Connection) -> dict:
|
|
righe = con.execute("SELECT * FROM fills ORDER BY ts_utc, asset").fetchall()
|
|
fills = [Fill(ts_utc=r["ts_utc"], asset=r["asset"], side=r["side"], qty=r["qty"],
|
|
price=r["price"], fee=r["fee"]) for r in righe]
|
|
_, aperti = fifo_roundtrips(fills)
|
|
out = {}
|
|
for a, lotti in aperti.items():
|
|
q = sum(x["qty"] for x in lotti)
|
|
costo = sum(x["qty"] * x["price"] for x in lotti)
|
|
out[a] = dict(qty=q, prezzo_medio=costo / q if q else 0.0, dal=min(x["ts"] for x in lotti))
|
|
return out
|
|
|
|
|
|
def ora() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|