""" Stats Monitor - legge trade da Bybit (closed-pnl) e calcola KPI con rilevamento deterioramento. KPI calcolati: - Win Rate rolling (ultimi 20 trade) - PnL cumulativo - PnL rolling - Avg Win vs Avg Loss ratio - Max Drawdown rolling - Consecutive losses - Expectancy per trade - Slope PnL (trend su 7gg) Output: - stats.json per la dashboard Flask - stats_monitor.log con tutti i calcoli - STATS_CRITICAL.flag se deterioramento grave - STATS_WARNING.flag se deterioramento moderato """ import csv import json import os import sys import re import time import hmac import hashlib 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") STATS_JSON = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_listener\stats.json") LOG_FILE = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_listener\logs\stats_monitor.log") FLAG_DIR = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_listener\flags") WEBHOOK_PY = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_receiver.py") BYBIT_BASE = "https://api-demo.bybit.com" def load_api_keys(): 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 # fallback dal file .env env_file = Path(r"G:\AI TRADING ENGINE\API_KEY_BYBIT.env") if env_file.exists(): for line in env_file.read_text(encoding="utf-8").splitlines(): if "=" in line and not line.strip().startswith("#"): k, v = line.split("=", 1) if k.strip() == "BYBIT_DEMO_API_KEY": api_key = v.strip() elif k.strip() == "BYBIT_DEMO_SECRET_KEY": api_secret = v.strip() if api_key and api_secret: return api_key, api_secret return "", "" def bybit_signed_request(method, path, params, api_key, api_secret): ts = str(int(time.time() * 1000)) recv_window = "5000" sorted_items = sorted(params.items()) param_str = "&".join([f"{k}={v}" for k, v in sorted_items]) sign_payload = ts + api_key + recv_window + param_str signature = hmac.new(api_secret.encode("utf-8"), sign_payload.encode("utf-8"), hashlib.sha256).hexdigest() headers = { "X-BAPI-API-KEY": api_key, "X-BAPI-SIGN": signature, "X-BAPI-TIMESTAMP": ts, "X-BAPI-RECV-WINDOW": recv_window, } if method == "GET": url = BYBIT_BASE + path + "?" + param_str req = urllib.request.Request(url, headers=headers, method="GET") else: url = BYBIT_BASE + path data = json.dumps(params).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers, method=method) r = urllib.request.urlopen(req, timeout=15) return json.loads(r.read().decode("utf-8")) def fetch_trades_from_bybit(api_key, api_secret, days=180): """Scarica trade chiusi da Bybit (closed-pnl) per categoria linear.""" end_ms = int(time.time() * 1000) start_ms = end_ms - days * 24 * 3600 * 1000 all_rows = [] cursor = None page = 0 while True: page += 1 params = {"category": "linear", "startTime": start_ms, "endTime": end_ms, "limit": 200} if cursor: params["cursor"] = cursor try: r = bybit_signed_request("GET", "/v5/position/closed-pnl", params, api_key, api_secret) except Exception as e: log_event("err fetch bybit: " + 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) return all_rows def parse_bybit_trades(rows): """Normalizza righe Bybit in dict trade.""" trades = [] for r in rows: try: symbol = r.get("symbol") side_raw = r.get("side", "").lower() side = "Buy" if side_raw 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_pct = float(r.get("orderPnlPct", 0) or 0) pnl_usd = float(r.get("closedPnl", 0) or 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 trades.append({ "timestamp": datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).astimezone().isoformat(), "symbol": symbol, "side": side, "qty": qty, "entry_price": entry_price, "exit_price": exit_price, "pnl_pct": pnl_pct, "pnl_usd": pnl_usd, "ts_dt": datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).astimezone(), "is_win": pnl_pct > 0, "pnl_pct_f": pnl_pct, "entry_price_f": entry_price, "exit_price_f": exit_price, "qty_f": qty, }) except (ValueError, TypeError): continue return trades # Soglie di alert THRESH_WR_ROLLING = 30.0 # % THRESH_MAX_DD = 10.0 # % THRESH_CONSEC_LOSSES = 5 THRESH_EDGE_RATIO = 1.0 THRESH_SLOPE_DAYS = 7 # giorni di slope negativa # Backtest Charter atteso (180gg SOL+BTC+ETH) per confronto BACKTEST_EXPECTED = { "SOLUSDT": {"pnl_pct": 28.13, "wr_pct": 62.5}, "BTCUSDT": {"pnl_pct": 15.72, "wr_pct": 59.1}, "ETHUSDT": {"pnl_pct": 12.97, "wr_pct": 45.0}, } def log_event(msg): ts = datetime.now(timezone.utc).astimezone().isoformat() line = "[" + ts + "] " + msg print(line, flush=True) try: LOG_FILE.parent.mkdir(parents=True, exist_ok=True) with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: pass def set_flag(name): try: FLAG_DIR.mkdir(parents=True, exist_ok=True) f = FLAG_DIR / name f.write_text(datetime.now(timezone.utc).astimezone().isoformat(), encoding="utf-8") except Exception: pass def clear_flag(name): try: f = FLAG_DIR / name if f.exists(): f.unlink() except Exception: pass def read_trades(): """Ritorna lista di dict con trade chiusi (entry + close accoppiati).""" if not TRADES_CSV.exists(): return [] rows = [] with open(TRADES_CSV, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for r in reader: if r.get("reason") == "close" and r.get("pnl_pct") and r.get("entry_price"): try: r["ts_dt"] = datetime.fromisoformat(r["timestamp"]) r["pnl_pct_f"] = float(r["pnl_pct"]) r["pnl_usd_f"] = float(r["pnl_usd"]) if r["pnl_usd"] else 0.0 r["entry_price_f"] = float(r["entry_price"]) r["exit_price_f"] = float(r["exit_price"]) if r["exit_price"] else 0.0 r["qty_f"] = float(r["qty"]) r["is_win"] = r["pnl_pct_f"] > 0 rows.append(r) except (ValueError, TypeError): continue return rows def compute_kpis(trades): """Calcola KPI globali e rolling.""" if not trades: return { "n_trades": 0, "wins": 0, "losses": 0, "wr_pct": 0.0, "pnl_pct_total": 0.0, "avg_win_pct": 0.0, "avg_loss_pct": 0.0, "edge_ratio": 0.0, "max_dd_pct": 0.0, "consec_losses": 0, "expectancy_pct": 0.0, "slope_7d": 0.0, "rolling_20_wr": 0.0, "rolling_20_pnl": 0.0, "by_symbol": {}, "by_strategy": {}, "alerts": [], "day_pnl_pct": 0.0, "day_pnl_usd": 0.0, "day_pnl_date": None, "n_days_trading": 0, "first_trade_date": None, "last_trade_date": None, "n_long": 0, "n_short": 0, "pct_long": 0.0, "pct_short": 0.0, "long_wins": 0, "long_losses": 0, "short_wins": 0, "short_losses": 0, "long_win_pct": 0.0, "long_loss_pct": 0.0, "short_win_pct": 0.0, "short_loss_pct": 0.0, "daily_data": [], "generated_at": datetime.now(timezone.utc).astimezone().isoformat() } n = len(trades) wins = sum(1 for t in trades if t["is_win"]) losses = n - wins wr = wins / n * 100 pnl_total = sum(t["pnl_pct_f"] for t in trades) win_pnls = [t["pnl_pct_f"] for t in trades if t["is_win"]] loss_pnls = [t["pnl_pct_f"] for t in trades if not t["is_win"]] avg_w = sum(win_pnls) / max(len(win_pnls), 1) avg_l = sum(loss_pnls) / max(len(loss_pnls), 1) edge_ratio = abs(avg_w / avg_l) if avg_l != 0 else 999.0 expectancy = pnl_total / n # per side (LONG=Buy, SHORT=Sell) long_trades = [t for t in trades if t.get("side", "").lower() == "buy"] short_trades = [t for t in trades if t.get("side", "").lower() == "sell"] n_long = len(long_trades) n_short = len(short_trades) pct_long = (n_long / n * 100) if n > 0 else 0 pct_short = (n_short / n * 100) if n > 0 else 0 long_wins = sum(1 for t in long_trades if t["is_win"]) long_losses = n_long - long_wins short_wins = sum(1 for t in short_trades if t["is_win"]) short_losses = n_short - short_wins long_win_pct = (long_wins / n_long * 100) if n_long > 0 else 0 long_loss_pct = (long_losses / n_long * 100) if n_long > 0 else 0 short_win_pct = (short_wins / n_short * 100) if n_short > 0 else 0 short_loss_pct = (short_losses / n_short * 100) if n_short > 0 else 0 # max drawdown cum = 0 peak = 0 max_dd = 0 for t in trades: cum += t["pnl_pct_f"] peak = max(peak, cum) dd = peak - cum max_dd = max(max_dd, dd) # consecutive losses consec = 0 max_consec = 0 for t in trades: if not t["is_win"]: consec += 1 max_consec = max(max_consec, consec) else: consec = 0 # slope 7gg (Pnl daily) by_day = {} for t in trades: d = t["ts_dt"].date() by_day[d] = by_day.get(d, 0) + t["pnl_pct_f"] days_sorted = sorted(by_day.keys()) slope = 0.0 if len(days_sorted) >= 2: # regressione lineare semplice n_d = len(days_sorted) xs = list(range(n_d)) ys = [by_day[d] for d in days_sorted] mean_x = sum(xs) / n_d mean_y = sum(ys) / n_d num = sum((xs[i] - mean_x) * (ys[i] - mean_y) for i in range(n_d)) den = sum((xs[i] - mean_x) ** 2 for i in range(n_d)) or 1 slope = num / den if den != 0 else 0 # rolling window ultimi 20 last_20 = trades[-20:] if n >= 20 else trades rolling_wr = sum(1 for t in last_20 if t["is_win"]) / len(last_20) * 100 rolling_pnl = sum(t["pnl_pct_f"] for t in last_20) # per symbol by_sym = {} for t in trades: s = t["symbol"] if s not in by_sym: by_sym[s] = {"n": 0, "wins": 0, "losses": 0, "pnl": 0.0} by_sym[s]["n"] += 1 by_sym[s]["pnl"] += t["pnl_pct_f"] if t["is_win"]: by_sym[s]["wins"] += 1 else: by_sym[s]["losses"] += 1 for s in by_sym: ns = by_sym[s]["n"] by_sym[s]["wr"] = by_sym[s]["wins"] / ns * 100 if ns else 0 # daily aggregation per grafici temporali (globale + per simbolo per sparkline) daily_data = {} # date_str -> {pnl, wins, losses, n} daily_by_sym = {} # symbol -> date_str -> {pnl, n} for t in trades: d = t["ts_dt"].date().isoformat() if d not in daily_data: daily_data[d] = {"pnl": 0.0, "wins": 0, "losses": 0, "n": 0} daily_data[d]["pnl"] += t["pnl_pct_f"] daily_data[d]["n"] += 1 if t["is_win"]: daily_data[d]["wins"] += 1 else: daily_data[d]["losses"] += 1 # per simbolo s = t["symbol"] if s not in daily_by_sym: daily_by_sym[s] = {} if d not in daily_by_sym[s]: daily_by_sym[s][d] = {"pnl": 0.0, "n": 0} daily_by_sym[s][d]["pnl"] += t["pnl_pct_f"] daily_by_sym[s][d]["n"] += 1 # costruisci by_sym_daily (lista ordinata per ogni simbolo con PnL cumulativo) by_sym_daily = {} for s, by_d in daily_by_sym.items(): cum = 0.0 lst = [] for d in sorted(by_d.keys()): cum += by_d[d]["pnl"] lst.append({"date": d, "pnl_pct": round(by_d[d]["pnl"], 3), "cum_pnl_pct": round(cum, 3), "n": by_d[d]["n"]}) by_sym_daily[s] = lst # per strategia (VPTR3 vs RETTANGOLO) by_strat = {} by_strat_daily = {} daily_by_strat = {} for t in trades: st = t.get("strategy", "vptr3") or "vptr3" # default vptr3 se mancante if st not in by_strat: by_strat[st] = {"n": 0, "wins": 0, "losses": 0, "pnl": 0.0} by_strat[st]["n"] += 1 by_strat[st]["pnl"] += t["pnl_pct_f"] if t["is_win"]: by_strat[st]["wins"] += 1 else: by_strat[st]["losses"] += 1 for st in by_strat: ns = by_strat[st]["n"] by_strat[st]["wr"] = by_strat[st]["wins"] / ns * 100 if ns else 0 # daily per strategia for t in trades: st = t.get("strategy", "vptr3") or "vptr3" d = t["ts_dt"].date().isoformat() if st not in daily_by_strat: daily_by_strat[st] = {} if d not in daily_by_strat[st]: daily_by_strat[st][d] = {"pnl": 0.0, "n": 0} daily_by_strat[st][d]["pnl"] += t["pnl_pct_f"] daily_by_strat[st][d]["n"] += 1 for st, by_d in daily_by_strat.items(): cum = 0.0 lst = [] for d in sorted(by_d.keys()): cum += by_d[d]["pnl"] lst.append({"date": d, "pnl_pct": round(by_d[d]["pnl"], 3), "cum_pnl_pct": round(cum, 3), "n": by_d[d]["n"]}) by_strat_daily[st] = lst # converti in lista ordinata + WR per giorno daily_list = [] cum_pnl = 0 for d in sorted(daily_data.keys()): data = daily_data[d] wr_day = (data["wins"] / data["n"] * 100) if data["n"] > 0 else 0 cum_pnl += data["pnl"] daily_list.append({ "date": d, "pnl_pct": round(data["pnl"], 3), "cum_pnl_pct": round(cum_pnl, 3), "wr_pct": round(wr_day, 1), "n_trades": data["n"], "wins": data["wins"], "losses": data["losses"] }) # WR rolling a finestra mobile rolling_window = 10 for i, d in enumerate(daily_list): win = daily_list[max(0, i - rolling_window + 1):i + 1] total_n = sum(x["n_trades"] for x in win) total_wins = sum(x["wins"] for x in win) d["wr_rolling"] = round(total_wins / total_n * 100, 1) if total_n > 0 else 0 # n_days_trading = "da quando hai iniziato al giorno attuale" (include giorni vuoti senza trade) # Chiarimento semantico: distinguo "Giorni Attivi" (con almeno 1 trade) da "Da Inizio" (calendario) n_days_active = len(daily_list) first_trade_date = daily_list[0]["date"] if daily_list else None last_trade_date = daily_list[-1]["date"] if daily_list else None if first_trade_date: first_dt = datetime.strptime(first_trade_date, "%Y-%m-%d") today_dt = datetime.now() n_days_trading = (today_dt - first_dt).days + 1 # inclusivo di oggi else: n_days_trading = 0 # day P&L (ultimo giorno disponibile) + conversione in USDT con capitale fittizio 2000 CAPITAL_FITTUZIO_USDT = 2000.0 if daily_list: last_day = daily_list[-1] day_pnl_pct = last_day["pnl_pct"] day_pnl_date = last_day["date"] else: day_pnl_pct = 0.0 day_pnl_date = None day_pnl_usd = round(day_pnl_pct * CAPITAL_FITTUZIO_USDT / 100, 2) # alerts alerts = [] if n >= 20 and rolling_wr < THRESH_WR_ROLLING: alerts.append({"level": "WARNING", "msg": "WR rolling < 30% (" + format(rolling_wr, ".1f") + "%)"}) if max_consec >= THRESH_CONSEC_LOSSES: alerts.append({"level": "WARNING", "msg": str(max_consec) + " loss consecutive"}) if max_dd > THRESH_MAX_DD: alerts.append({"level": "CRITICAL", "msg": "Max DD > 10% (" + format(max_dd, ".1f") + "%)"}) if avg_l != 0 and edge_ratio < THRESH_EDGE_RATIO: alerts.append({"level": "CRITICAL", "msg": "Edge ratio < 1.0 (" + format(edge_ratio, ".2f") + ")"}) if len(days_sorted) >= THRESH_SLOPE_DAYS and slope < 0: alerts.append({"level": "WARNING", "msg": "Slope PnL negativa per " + str(len(days_sorted)) + "gg"}) return { "n_trades": n, "wins": wins, "losses": losses, "wr_pct": wr, "pnl_pct_total": pnl_total, "avg_win_pct": avg_w, "avg_loss_pct": avg_l, "edge_ratio": edge_ratio, "max_dd_pct": max_dd, "consec_losses": max_consec, "expectancy_pct": expectancy, "slope_7d": slope, "rolling_20_wr": rolling_wr, "rolling_20_pnl": rolling_pnl, "by_symbol": by_sym, "by_symbol_daily": by_sym_daily, "alerts": alerts, "by_strategy": by_strat, "by_strategy_daily": by_strat_daily, "day_pnl_pct": day_pnl_pct, "day_pnl_usd": day_pnl_usd, "day_pnl_date": day_pnl_date, "n_days_trading": n_days_trading, "n_days_active": n_days_active, "first_trade_date": first_trade_date, "last_trade_date": last_trade_date, "n_long": n_long, "n_short": n_short, "pct_long": round(pct_long, 1), "pct_short": round(pct_short, 1), "long_wins": long_wins, "long_losses": long_losses, "short_wins": short_wins, "short_losses": short_losses, "long_win_pct": round(long_win_pct, 1), "long_loss_pct": round(long_loss_pct, 1), "short_win_pct": round(short_win_pct, 1), "short_loss_pct": round(short_loss_pct, 1), "daily_data": daily_list, "generated_at": datetime.now(timezone.utc).astimezone().isoformat() } def generate_demo_trades(n=20): """Genera trade fake realistici per popolare la dashboard in demo mode.""" import random from datetime import timedelta random.seed(42) # riproducibile symbols = ["SOLUSDT", "BTCUSDT", "ETHUSDT", "WIFUSDT"] base_prices = {"SOLUSDT": 75, "BTCUSDT": 62500, "ETHUSDT": 1790, "WIFUSDT": 0.15} trades = [] now = datetime.now(timezone.utc) for i in range(n): sym = random.choice(symbols) bp = base_prices[sym] side = random.choice(["Buy", "Sell"]) # WR circa 50% (un po' sotto per simulare edge debole) is_win = random.random() < 0.5 if is_win: pnl_pct = random.uniform(1.5, 6.0) # win else: pnl_pct = random.uniform(-3.0, -0.8) # loss qty = round(random.uniform(1, 5), 2) entry = bp * random.uniform(0.95, 1.05) exit_ = entry * (1 + pnl_pct/100) ts = now - timedelta(days=random.uniform(0, 30), hours=random.uniform(0, 23)) trades.append({ "timestamp": ts.isoformat(), "symbol": sym, "side": side, "qty": qty, "entry_price": entry, "exit_price": exit_, "pnl_pct": pnl_pct, "pnl_usd": pnl_pct * entry * qty / 100, "ts_dt": ts, "is_win": pnl_pct > 0, "pnl_pct_f": pnl_pct, "entry_price_f": entry, "exit_price_f": exit_, "qty_f": qty, }) trades.sort(key=lambda x: x["ts_dt"]) return trades def read_trades_csv(): """Legge trade da CSV locale (popolato da webhook o da downloader).""" if not TRADES_CSV.exists(): return [] rows = [] with open(TRADES_CSV, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for r in reader: if r.get("reason", "").startswith("close") and r.get("pnl_pct") and r.get("entry_price"): try: r["ts_dt"] = datetime.fromisoformat(r["timestamp"]) r["pnl_pct_f"] = float(r["pnl_pct"]) r["pnl_usd_f"] = float(r["pnl_usd"]) if r["pnl_usd"] else 0.0 r["entry_price_f"] = float(r["entry_price"]) r["exit_price_f"] = float(r["exit_price"]) if r["exit_price"] else 0.0 r["qty_f"] = float(r["qty"]) r["is_win"] = r["pnl_pct_f"] > 0 rows.append(r) except (ValueError, TypeError): continue return rows def main(): log_event("=== STATS MONITOR RUN ===") # fonte primaria: CSV locale (popolato da webhook + downloader) trades = read_trades_csv() log_event("trade letti da CSV: " + str(len(trades))) # DEMO MODE: se non ci sono trade, genera dati fake per popolare dashboard if not trades: log_event("!!! DEMO MODE: genero 20 trade fake per popolare la dashboard") trades = generate_demo_trades(20) kpis = compute_kpis(trades) # scrivi stats.json per dashboard try: STATS_JSON.parent.mkdir(parents=True, exist_ok=True) with open(STATS_JSON, "w", encoding="utf-8") as f: json.dump(kpis, f, indent=2, default=str) except Exception as e: log_event("err scrittura stats.json: " + str(e)) # alert if any(a["level"] == "CRITICAL" for a in kpis["alerts"]): set_flag("STATS_CRITICAL.flag") log_event("!!! CRITICAL alert: " + "; ".join(a["msg"] for a in kpis["alerts"] if a["level"] == "CRITICAL")) else: clear_flag("STATS_CRITICAL.flag") if any(a["level"] == "WARNING" for a in kpis["alerts"]): set_flag("STATS_WARNING.flag") log_event("!! WARNING alert: " + "; ".join(a["msg"] for a in kpis["alerts"] if a["level"] == "WARNING")) else: clear_flag("STATS_WARNING.flag") log_event("KPI: trades=" + str(kpis["n_trades"]) + " WR=" + format(kpis["wr_pct"], ".1f") + "% PnL=" + format(kpis["pnl_pct_total"], ".2f") + "% DD=" + format(kpis["max_dd_pct"], ".1f") + "%") if not kpis["alerts"]: log_event("OK: nessun deterioramento rilevato") log_event("=== FINE RUN ===\n") if __name__ == "__main__": main()