""" Download trade storici da Bybit (demo o mainnet) per popolare stats. Usa /v5/position/closed-pnl per ottenere trade chiusi. Output: trades.csv compatibile con stats_monitor. """ import os import sys import csv import hmac import hashlib import time import json import argparse import urllib.request import urllib.parse from datetime import datetime, timezone from pathlib import Path TRADES_CSV = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_listener\logs\trades.csv") ENV_FILE = Path(r"G:\AI TRADING ENGINE\API_KEY_BYBIT.env") WEBHOOK_PY = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_receiver.py") def load_api_keys(): """Carica chiavi da env o da file .env o dal webhook_receiver.py (hardcoded).""" api_key = os.environ.get("BYBIT_API_KEY", "") api_secret = os.environ.get("BYBIT_API_SECRET", "") if api_key and api_secret: return api_key, api_secret if ENV_FILE.exists(): with open(ENV_FILE, "r", encoding="utf-8") as f: for line in f: if "=" in line: k, v = line.strip().split("=", 1) if k == "BYBIT_DEMO_API_KEY": api_key = v elif k == "BYBIT_DEMO_SECRET_KEY": api_secret = v if api_key and api_secret: return api_key, api_secret # fallback: leggi da webhook_receiver.py (hardcoded li) if WEBHOOK_PY.exists(): import re content = WEBHOOK_PY.read_text(encoding="utf-8") m_key = re.search(r'BYBIT_API_KEY\s*=\s*"([^"]+)"', content) m_sec = re.search(r'BYBIT_API_SECRET\s*=\s*"([^"]+)"', content) if m_key and m_sec: api_key = m_key.group(1) api_secret = m_sec.group(1) return api_key, api_secret def bybit_request(api_key, api_secret, method, path, params=None, base="https://api-demo.bybit.com"): ts = str(int(time.time() * 1000)) recv_window = "5000" if params: sorted_items = sorted(params.items()) if method == "GET": param_str = "&".join([f"{k}={v}" for k, v in sorted_items]) url_path = path + ("?" + param_str if param_str else "") sign_payload = ts + api_key + recv_window + param_str else: param_str = json.dumps(params, separators=(", ", ": ")) url_path = path sign_payload = ts + api_key + recv_window + param_str signature = hmac.new(api_secret.encode(), sign_payload.encode(), hashlib.sha256).hexdigest() headers = { "X-BAPI-API-KEY": api_key, "X-BAPI-SIGN": signature, "X-BAPI-TIMESTAMP": ts, "X-BAPI-RECV-WINDOW": recv_window, "Content-Type": "application/json" } url = base + url_path else: headers = {"Content-Type": "application/json"} url = base + path if method == "GET": r = urllib.request.urlopen(urllib.request.Request(url, headers=headers, method="GET"), timeout=15) else: r = urllib.request.urlopen(urllib.request.Request(url, headers=headers, method=method), timeout=15) return json.loads(r.read().decode()) def download_closed_pnl(api_key, api_secret, days=180, base="https://api-demo.bybit.com"): """Scarica closed PnL per categoria linear ultimi N giorni. Bybit demo limita a 7gg per query, faccio chunking.""" end_ms = int(time.time() * 1000) start_ms = end_ms - days * 24 * 3600 * 1000 all_rows = [] # chunking: 6 giorni per query (margine di sicurezza rispetto al limite di 7) chunk_ms = 6 * 24 * 3600 * 1000 chunk_start = start_ms while chunk_start < end_ms: chunk_end = min(chunk_start + chunk_ms, end_ms) cursor = None page = 0 while True: page += 1 params = { "category": "linear", "startTime": chunk_start, "endTime": chunk_end, "limit": 200 } if cursor: params["cursor"] = cursor try: r = bybit_request(api_key, api_secret, "GET", "/v5/position/closed-pnl", params, base) except Exception as e: print(" err chunk: " + str(e)) break result = r.get("result", {}) rows = result.get("list", []) all_rows.extend(rows) cursor = result.get("nextPageCursor") if not cursor or len(rows) == 0: break if page > 50: break time.sleep(0.2) chunk_start = chunk_end time.sleep(0.1) return all_rows def parse_closed_pnl(rows, api_key, api_secret, base="https://api-demo.bybit.com"): """Converte righe closed-pnl in formato trade normalizzato. Ricalcola pnl_pct da entry/exit price (Bybit orderPnlPct รจ rapporto PnL/EntryValue, non variazione prezzo).""" trades = [] for r in rows: try: symbol = r.get("symbol") side = "Buy" if r.get("side", "").lower() in ("buy", "long") else "Sell" entry_price = float(r.get("avgEntryPrice", 0) or 0) exit_price = float(r.get("avgExitPrice", 0) or 0) qty = float(r.get("qty", 0) or r.get("size", 0) or 0) pnl_usd = float(r.get("closedPnl", 0) or 0) if entry_price > 0 and exit_price > 0: if side == "Buy": pnl_pct = (exit_price - entry_price) / entry_price * 100 else: pnl_pct = (entry_price - exit_price) / entry_price * 100 else: pnl_pct = 0 ts_ms = int(r.get("updatedTime", 0) or r.get("createdTime", 0) or 0) if not symbol or ts_ms == 0 or qty <= 0: continue ts_iso = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).astimezone().isoformat() trades.append({ "timestamp": ts_iso, "symbol": symbol, "side": side, "qty": qty, "entry_price": entry_price, "exit_price": exit_price, "pnl_pct": pnl_pct, "pnl_usd": pnl_usd, "reason": "close_downloaded" }) except (ValueError, TypeError) as e: print(" skip row: " + str(e)) continue return trades def append_to_csv(trades): """Aggiunge trade al CSV, evitando duplicati per timestamp+symbol.""" TRADES_CSV.parent.mkdir(parents=True, exist_ok=True) existing = set() if TRADES_CSV.exists(): with open(TRADES_CSV, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for r in reader: key = (r.get("timestamp", ""), r.get("symbol", ""), r.get("reason", "")) existing.add(key) file_exists = TRADES_CSV.exists() added = 0 with open(TRADES_CSV, "a", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=["timestamp","symbol","side","qty","entry_price","exit_price","pnl_pct","pnl_usd","reason"]) if not file_exists: writer.writeheader() for t in trades: key = (t["timestamp"], t["symbol"], t.get("reason", "")) if key in existing: continue writer.writerow(t) added += 1 return added def main(): parser = argparse.ArgumentParser() parser.add_argument("--days", type=int, default=180, help="giorni di storico") parser.add_argument("--mainnet", action="store_true", help="usa mainnet invece di demo") args = parser.parse_args() api_key, api_secret = load_api_keys() if not api_key or not api_secret: print("ERR: chiavi API mancanti. Setta BYBIT_API_KEY e BYBIT_API_SECRET o mettile in " + str(ENV_FILE)) sys.exit(1) base = "https://api.bybit.com" if args.mainnet else "https://api-demo.bybit.com" print(f"Scarico closed PnL da {base} ultimi {args.days}gg...") rows = download_closed_pnl(api_key, api_secret, args.days, base) print(f" ricevute {len(rows)} righe") if not rows: print("Nessun dato. Esci.") return trades = parse_closed_pnl(rows, api_key, api_secret, base) print(f" parsate {len(trades)} trades") added = append_to_csv(trades) print(f" aggiunte {added} nuove righe a {TRADES_CSV}") print(f" (duplicati saltati: {len(trades) - added})") print("\nFatto! Ora puoi rilanciare stats_monitor.py per aggiornare stats.json") if __name__ == "__main__": main()