"""Report "stato trades" — separa SEMPRE pool reale e paper-stats, e mostra sia realizzato che unrealized. Verita' aggregata = ledger PORT06; il dettaglio per-trade viene dai worker dir. Fonte unica della separazione (NON la memoria — qui ho gia' sbagliato una volta includendo XS01 nell'equity): data/portfolio_paper/ -> POOL REALE (nel ledger, muove l'equity/il conto) data/portfolio_paper_stats/ -> PAPER-STATS (solo statistica, MAI nel ledger) Uso: uv run python scripts/analysis/trades_status.py # ultime 24h uv run python scripts/analysis/trades_status.py --since 2026-06-16T14:00 """ import argparse import datetime as dt import glob import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) # project root -> import src.* DATA = Path("data") POOL_DIR = DATA / "portfolio_paper" STATS_DIR = DATA / "portfolio_paper_stats" LEDGER = DATA / "portfolios" / "PORT06" / "status.json" EQUITY = DATA / "portfolios" / "PORT06" / "equity.jsonl" def _closes(worker_dir: Path, since: dt.datetime): """CLOSE events dei worker in worker_dir dal momento `since`.""" out = [] for f in sorted(glob.glob(str(worker_dir / "*" / "trades.jsonl"))): name = Path(f).parent.name for line in open(f): try: d = json.loads(line) except json.JSONDecodeError: continue if d.get("event") != "CLOSE": continue try: t = dt.datetime.fromisoformat(d.get("ts", "")) except ValueError: continue if t < since: continue # PnL che conta: reale se eseguito (real_truth), altrimenti il sim/pnl real = d.get("real_pnl") real = real if real is not None else d.get("pnl", 0.0) out.append({ "ts": d.get("ts", "")[:19], "name": name, "reason": d.get("reason"), "held": d.get("bars_held"), "pnl": d.get("pnl"), "real": d.get("real_pnl"), "sim": d.get("sim_pnl"), "eff": real or 0.0, }) return sorted(out, key=lambda r: r["ts"]) def _open_positions(worker_dir: Path): """Posizioni aperte con i campi per il PnL non realizzato (entry REALE di esecuzione quando eseguito; il feed di decisione sim e' dislocato).""" out = [] for f in sorted(glob.glob(str(worker_dir / "*" / "status.json"))): d = json.load(open(f)) if not d.get("in_position"): continue wid = Path(f).parent.name is_pair = "entry_a" in d executed = bool(d.get("real_in_position")) row = {"name": wid, "pair": is_pair, "real": executed, "dir": d.get("direction", 0), "held": d.get("bars_held"), "max_bars": d.get("max_bars"), "tp": d.get("tp", 0.0), "unreal": None} if is_pair: a_, b_ = wid.split("__")[1].split("_") row.update({ "assets": [a_, b_], "leg_a": a_, "leg_b": b_, "entry_a": d.get("real_entry_a") if executed else d.get("entry_a"), "entry_b": d.get("real_entry_b") if executed else d.get("entry_b"), "notional_a": d.get("real_notional_a") or 0.0, "notional_b": d.get("real_notional_b") or 0.0, }) else: asset = wid.split("__")[1] entry = (d.get("real_entry_price") if executed else None) or d.get("entry_price") or 0.0 row.update({ "assets": [asset], "asset": asset, "entry": entry, "notional": d.get("real_entry_notional") or round(d.get("capital", 0.0) * 0.5 * 3, 1), }) out.append(row) return out def _marks(assets: set) -> dict: """Mark correnti USDC perp (best-effort). {} se Cerbero non risponde.""" if not assets: return {} try: from src.live.cerbero_client import CerberoClient insts = [f"{a}_USDC-PERPETUAL" for a in assets] data = CerberoClient().get_ticker_batch(insts) return {t["instrument_name"].split("_")[0].replace("-PERPETUAL", ""): t.get("mark_price") for t in data.get("tickers", [])} except Exception as e: print(f"[trades_status] mark fetch fallita ({e!r}) -> PnL aperti n/d", file=sys.stderr) return {} def _annotate_pnl(positions, marks): """PnL non realizzato live, convenzione identica alla dashboard (entry reale vs mark USDC).""" for a in positions: if a["pair"]: ma, mb = marks.get(a["leg_a"]), marks.get(a["leg_b"]) if ma and mb and a.get("entry_a") and a.get("entry_b"): s = 1 if a["dir"] > 0 else -1 # +1: long A / short B ga = a["notional_a"] * s * (ma - a["entry_a"]) / a["entry_a"] gb = a["notional_b"] * (-s) * (mb - a["entry_b"]) / a["entry_b"] a["unreal"] = ga + gb a["mark"], a["mark_b"] = ma, mb else: mk = marks.get(a["asset"]) if mk and a.get("entry"): sign = 1 if a["dir"] > 0 else -1 pct = sign * (mk - a["entry"]) / a["entry"] a["unreal"] = a["notional"] * pct a["unreal_pct"] = pct * 100 a["mark"] = mk return positions def _equity_at(since: dt.datetime): """Equity piu' vicino (e <=) a `since`, per il Δ della finestra.""" base = None if not EQUITY.exists(): return None for line in open(EQUITY): try: d = json.loads(line) t = dt.datetime.fromisoformat(d["ts"]) except (json.JSONDecodeError, KeyError, ValueError): continue if t <= since: base = d["equity"] else: break return base def main(): ap = argparse.ArgumentParser() ap.add_argument("--since", default=None, help="ISO (es. 2026-06-16T14:00). Default: 24h fa.") args = ap.parse_args() now = dt.datetime.now(dt.timezone.utc) if args.since: since = dt.datetime.fromisoformat(args.since) if since.tzinfo is None: since = since.replace(tzinfo=dt.timezone.utc) else: since = now - dt.timedelta(hours=24) led = json.load(open(LEDGER)) equity = led["equity"] cap = led["total_capital"] unreal = equity - cap eq_base = _equity_at(since) ver = Path("VERSION").read_text().strip() if Path("VERSION").exists() else "?" print(f"STATO TRADES — v{ver} — {now:%Y-%m-%d %H:%M} UTC (finestra dal {since:%Y-%m-%d %H:%M})\n") # ---- aggregato (ledger = verita') ---- print("LEDGER PORT06 (verita' aggregata)") print(f" equity = {equity:.2f}") print(f" capitale realizz.= {cap:.2f}") print(f" unrealized aperti= {unreal:+.2f} (equity - capitale; pos. aperte mark-to-market)") print(f" peak {led.get('peak'):.2f} maxDD {led.get('max_dd')}%") if eq_base is not None: print(f" Δ equity finestra= {equity - eq_base:+.2f} ({eq_base:.2f} -> {equity:.2f})") # ---- POOL REALE ---- pool = _closes(POOL_DIR, since) s_real = sum(r["eff"] for r in pool) wins = sum(1 for r in pool if r["eff"] > 0) print(f"\n=== POOL REALE (muove l'equity) — {len(pool)} chiusure, {wins}W/{len(pool)-wins}L ===") for r in pool: tag = "LOSS" if r["eff"] < 0 else "win " print(f" {r['ts']} {tag} {r['name']:36} {str(r['reason']):11} " f"real={r['real']} sim={r['sim']}") print(f" Σ REALIZZATO POOL = {s_real:+.2f}") op = _open_positions(POOL_DIR) if op: assets = set().union(*[set(p["assets"]) for p in op]) marks = _marks(assets) _annotate_pnl(op, marks) s_unreal = sum(p["unreal"] for p in op if p["unreal"] is not None) n_marked = sum(1 for p in op if p["unreal"] is not None) print(f"\n posizioni aperte ({len(op)}) — PnL non realizzato live (entry reale vs mark USDC):") for p in sorted(op, key=lambda x: (x["unreal"] is None, x.get("unreal") or 0)): d = {1: "LONG", -1: "SHORT"}.get(p["dir"], "?") if p["unreal"] is not None: pct = f"{p['unreal_pct']:+.2f}%" if not p["pair"] and "unreal_pct" in p else "" mark = f"mark={p['mark']}" if not p["pair"] else f"mk {p['mark']}/{p.get('mark_b')}" pnl = f"{p['unreal']:+.2f}" else: pct = mark = ""; pnl = "n/d (no mark)" ent = f"entry={p['entry']}" if not p["pair"] else f"a={p.get('entry_a')} b={p.get('entry_b')}" print(f" {p['name']:36} {d:5} h={p['held']}/{p['max_bars']} {ent} {mark} " f"PnL {pnl} {pct}") print(f" Σ UNREALIZED REALE = {s_unreal:+.2f} ({n_marked}/{len(op)} con mark live) " f"<- verita' del conto (entry reale vs mark USDC)") print(f" ledger unrealized = {unreal:+.2f} (feed sim-decisione testnet DISLOCATO); " f"lo scarto {s_unreal - unreal:+.2f} e' la dislocazione del feed, non soldi.") # ---- PAPER-STATS ---- stats = _closes(STATS_DIR, since) s_stats = sum(r["eff"] for r in stats) print(f"\n=== PAPER-STATS (solo statistica — NON tocca equity/conto) — {len(stats)} chiusure ===") for r in stats: print(f" {r['ts']} {r['name']:36} pnl={r['pnl']}") print(f" Σ PAPER-STATS = {s_stats:+.2f} <- ignorare per l'equity") # ---- riconciliazione ---- print("\nRICONCILIAZIONE") print(f" realizzato POOL {s_real:+.2f} + Δunrealized aperti ≈ Δ equity finestra") if eq_base is not None: residuo = (equity - eq_base) - s_real print(f" Δequity {equity - eq_base:+.2f} - realizzato {s_real:+.2f} = " f"{residuo:+.2f} (= Δunrealized + ribilanci/timing)") print(f" paper-stats {s_stats:+.2f} NON incluso (book in sola simulazione).") if __name__ == "__main__": main()