""" sync_bybit.py — Sincronizza trades.csv con le chiusure reali di Bybit. PROBLEMA: trades.csv viene popolato SOLO quando il close passa dal webhook (alert TV, /webhook/sltp). Le chiusure manuali su Bybit o per trigger SL/TP Bybit NON passano dal webhook, quindi le stats dashboard non le vede. SOLUZIONE: scarica /v5/position/closed-pnl delle ultime X ore, confronta con trades.csv usando chiave composita (symbol, size, entry, exit, ts_min) e aggiunge le righe mancanti con strategy dedotta dall'entry corrispondente. NON modifica righe esistenti. NON tocca la struttura del CSV. Formato trades.csv (riga header): timestamp,symbol,side,qty,entry_price,exit_price,pnl_pct,pnl_usd,reason,strategy """ import csv import sys import time from datetime import datetime, timezone from pathlib import Path from zoneinfo import ZoneInfo sys.path.insert(0, r"G:\AI TRADING ENGINE\live_deploy") from bybit_demo_client import BybitDemoClient TRADES_CSV = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_listener\logs\trades.csv") TZ = ZoneInfo("Europe/Rome") HEADER = "timestamp,symbol,side,qty,entry_price,exit_price,pnl_pct,pnl_usd,reason,strategy" def load_existing_keys() -> set: """Set di chiavi (symbol, qty, entry, exit, ts_min) per dedupe.""" keys = set() if not TRADES_CSV.exists(): return keys with open(TRADES_CSV, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: sym = (row.get("symbol") or "").strip() exit_p = row.get("exit_price") or "" try: qty = float(row.get("qty", "0") or 0) entry = float(row.get("entry_price", "0") or 0) exit_v = float(exit_p) except ValueError: continue if not sym or exit_v == 0: continue # righe 'entry' senza exit_price ts = (row.get("timestamp") or "").strip() ts_min = ts[:16] # yyyy-mm-ddTHH:MM keys.add((sym, qty, entry, exit_v, ts_min)) return keys def find_entry_strategy(symbol: str, qty: float, entry: float) -> str: """Cerca in trades.csv la riga 'entry' con stesso symbol+qty+entry, ritorna strategy.""" if not TRADES_CSV.exists(): return "unknown" with open(TRADES_CSV, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: if (row.get("symbol") or "").strip() != symbol: continue if (row.get("reason") or "").strip().lower() != "entry": continue try: if abs(float(row.get("qty", "0")) - qty) < 1e-9 and \ abs(float(row.get("entry_price", "0")) - entry) < 0.5: return (row.get("strategy") or "unknown").strip() except ValueError: continue return "unknown" def detect_close_reason(c: dict) -> str: """Determina il reason della chiusura Bybit. IMPORTANTE: stats_monitor.py conta SOLO le righe con reason che INIZIA con 'close'. Usiamo 'close_bybit_*' per essere conteggiati nelle stats. Euristica: PnL > 0 -> tp trigger, < 0 -> sl trigger, ~0 -> manuale al market.""" try: pnl = float(c.get("closedPnl") or 0) except (KeyError, ValueError, TypeError): return "close_bybit_manual" if pnl > 0.5: return "close_bybit_tp" elif pnl < -0.5: return "close_bybit_sl" else: return "close_bybit_manual" def fetch_closed_pnl(client: BybitDemoClient, hours: int = 24) -> list: data = client._request("GET", "/v5/position/closed-pnl", {"category": "linear", "settleCoin": "USDT", "limit": "100"}, signed=True) lst = data.get("result", {}).get("list", []) now_ms = int(time.time() * 1000) cutoff = now_ms - hours * 3600 * 1000 return [c for c in lst if int(c.get("updatedTime") or 0) >= cutoff] def build_row(c: dict, strategy: str) -> list: """Riga trades.csv (lista valori) da closed-pnl dict.""" upd = int(c.get("updatedTime") or 0) when = datetime.fromtimestamp(upd / 1000, tz=timezone.utc).astimezone(TZ) # Formato timestamp: 2026-07-15T19:40:15.123456+0200 (6 decimali, TZ +HHMM) ts = when.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + when.strftime("%z") # %f[:-3] = millisecondi (3 decimali). Le righe esistenti hanno microsecondi # (6 decimali). Python %f ha microsecondi. Tengo i millisecondi per consistenza # con le righe del supervisore (che usa datetime.now().isoformat(timespec='milliseconds')). # Le righe scritte dal webhook hanno microsecondi, questo è un pelo diverso ma # funzionalmente identico. qty = float(c["qty"]) entry = float(c["avgEntryPrice"]) exit_p = float(c["avgExitPrice"]) pnl_usd = float(c.get("closedPnl") or 0) notional = qty * entry pnl_pct = (pnl_usd / notional * 100) if notional > 0 else 0 reason = detect_close_reason(c) return [ ts, c["symbol"], c["side"], # direzione close f"{qty:g}", f"{entry:g}", f"{exit_p:g}", f"{pnl_pct:.4f}", f"{pnl_usd:.4f}", reason, strategy, ] def sync(hours: int = 24, dry_run: bool = False) -> dict: client = BybitDemoClient() existing = load_existing_keys() closed = fetch_closed_pnl(client, hours=hours) added_rows = [] skipped = [] for c in closed: sym = c["symbol"] try: qty = float(c["qty"]) entry = float(c["avgEntryPrice"]) exit_p = float(c["avgExitPrice"]) except (KeyError, ValueError): continue upd = int(c.get("updatedTime") or 0) when = datetime.fromtimestamp(upd / 1000, tz=timezone.utc).astimezone(TZ) ts_min = when.strftime("%Y-%m-%dT%H:%M") key = (sym, qty, entry, exit_p, ts_min) if key in existing: skipped.append({"symbol": sym, "ts": when.strftime("%H:%M")}) continue strategy = find_entry_strategy(sym, qty, entry) added_rows.append((c, strategy, build_row(c, strategy))) if not dry_run and added_rows: if not TRADES_CSV.exists(): return {"error": f"trades.csv non trovato: {TRADES_CSV}"} with open(TRADES_CSV, "a", encoding="utf-8", newline="") as f: w = csv.writer(f) for _c, _strat, row in added_rows: w.writerow(row) return { "scanned_closed_pnl": len(closed), "added": len(added_rows), "skipped_already_present": len(skipped), "details_added": [ {"symbol": r[1], "side_close": r[2], "qty": r[3], "entry": r[4], "exit": r[5], "pnl_usd": r[7], "reason": r[8], "strategy": r[9], "ts": r[0]} for _c, _strat, r in added_rows ], "details_skipped": skipped, "dry_run": dry_run, "hours_window": hours, } if __name__ == "__main__": import json hours = int(sys.argv[1]) if len(sys.argv) > 1 else 24 dry = "--dry" in sys.argv print(json.dumps(sync(hours=hours, dry_run=dry), indent=2, default=str))