Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(analysis): trades_status — PnL live per posizione aperta (entry reale vs mark USDC)

Browse Source 
Riusa la convenzione della dashboard (get_ticker_batch USDC perp, real_entry vs mark;
pairs a 2 gambe). Mostra unrealized per trade + Σ REALE, e segnala che il ledger
unrealized e' sul feed sim-decisione testnet DISLOCATO (lo scarto e' dislocazione,
non soldi). Fix sys.path: import src.* anche eseguito come script file.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-17 15:56:22 +00:00
parent 3a66907f43
commit a0842a61cc
1 changed files with 88 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/analysis/trades_status.py
+88 -10
View File
@@ -15,8 +15,11 @@ 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"
@@ -54,18 +57,77 @@ def _closes(worker_dir: Path, since: dt.datetime):
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 d.get("in_position"):
out.append({
"name": Path(f).parent.name, "dir": d.get("direction"),
"entry": d.get("entry_price"), "held": d.get("bars_held"),
"real": d.get("real_in_position"),
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
@@ -128,11 +190,27 @@ def main():
print(f" Σ REALIZZATO POOL = {s_real:+.2f}")
op = _open_positions(POOL_DIR)
if op:
print(f" posizioni aperte ({len(op)}):")
for p in op:
d = {1: "long", -1: "short"}.get(p["dir"], "?")
print(f" {p['name']:36} {d:5} entry={p['entry']} held={p['held']} "
f"real={'Y' if p['real'] else 'sim'}")
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)