""" v2_stats_dashboard.py — Stats Dashboard 2° account (port 5511, Basic Auth) NO template engine: tutto render server-side in Python (niente {{}} / {% %} nel template). """ import os import sys import json import math import sqlite3 import hmac import secrets import string import html import hashlib from email.parser import BytesParser from email.policy import default as email_policy from datetime import datetime, timezone from pathlib import Path from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import threading DATA_REQUEST_LOCK = threading.RLock() from urllib.parse import urlparse, parse_qs from collections import defaultdict from zoneinfo import ZoneInfo # === CONFIG === ROOT = Path("/opt/charter-live/live_deploy_v2") DB_PATH = ROOT / "logs" / "webhook_queue.db" STATS_POLICY_PATH = ROOT / "v2_stats_trade_policy.json" REVIEW_QUEUE_PATH = ROOT / "v2_classification_review_queue.json" REVIEW_EVIDENCE_DIR = ROOT / "v2_classification_review_evidence" CERTIFICATION_AUDIT_PATH = ROOT / "v2_classification_certification_audit.jsonl" STATS_POLICY_BACKUP_DIR = ROOT / "v2_stats_policy_backups" MAX_REVIEW_BODY_BYTES = 5 * 1024 * 1024 MAX_EVIDENCE_BYTES = 4 * 1024 * 1024 PORT = int(os.environ.get("V2_STATS_PORT", "5511")) LISTEN_HOST = "0.0.0.0" LOG_DIR = ROOT / "logs" LOG_FILE = LOG_DIR / "v2_stats_dashboard.log" # === AUTH (caricata da .v2_stats_creds via systemd EnvironmentFile) === def _load_auth_from_envfile(): creds_file = ROOT / ".v2_stats_creds" if creds_file.exists(): try: for line in creds_file.read_text().splitlines(): line = line.strip() if line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k = k.strip() v = v.strip().strip('"').strip("'") if k == "V2_STATS_USER" and not os.environ.get("V2_STATS_USER"): os.environ["V2_STATS_USER"] = v elif k == "V2_STATS_PASS" and not os.environ.get("V2_STATS_PASS"): os.environ["V2_STATS_PASS"] = v except Exception: pass _load_auth_from_envfile() AUTH_USER = os.environ.get("V2_STATS_USER", "mattia_v2_stats") AUTH_PASS = os.environ.get("V2_STATS_PASS", "") REVIEW_ACTIONS = frozenset({ "classify", "exclude_test", "exclude_operational", "needs_review", }) REVIEW_TIMEZONE = ZoneInfo("Europe/Rome") def _parse_bybit_ts(ts): """Bybit V5 closed-pnl restituisce createdAt/updatedTime in MILLISECONDI come stringa (es. "1785856148609"). Accetta anche ISO 8601 ("2026-08-05T20:39:31Z") o None/"". Ritorna datetime (UTC) oppure None se non parsabile. """ if ts is None: return None s = str(ts).strip() if not s: return None # Millisecondi: tutti digits e lunghezza >= 10 if s.isdigit() and len(s) >= 10: try: ms = int(s) return datetime.fromtimestamp(ms / 1000.0, tz=timezone.utc) except Exception: return None # ISO 8601 fallback try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except Exception: return None def _fmt_bybit_ts(ts, fmt="%Y-%m-%d %H:%M:%S"): """Ritorna una stringa formattata (UTC) per ts Bybit (ms o ISO). None se non parsabile.""" dt = _parse_bybit_ts(ts) if dt is None: return "" return dt.strftime(fmt) def _fmt_review_ts(ts): dt = _parse_bybit_ts(ts) if dt is None: return "--" return dt.astimezone(REVIEW_TIMEZONE).strftime("%d/%m/%Y %H:%M") def log(msg): ts = datetime.now(timezone.utc).astimezone().isoformat() line = f"[{ts}] {msg}" print(line, flush=True) try: LOG_DIR.mkdir(parents=True, exist_ok=True) with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(line + "\n") except Exception: pass def _load_review_queue(): if not REVIEW_QUEUE_PATH.exists(): return [] try: data = json.loads(REVIEW_QUEUE_PATH.read_text(encoding="utf-8-sig")) return data if isinstance(data, list) else [] except Exception as exc: log(f"review queue read err: {exc}") return [] def _save_review_queue(items): """Scrittura atomica: la coda non modifica mai la policy certificata.""" tmp = REVIEW_QUEUE_PATH.with_name( f".{REVIEW_QUEUE_PATH.name}.{secrets.token_hex(6)}.tmp" ) try: tmp.write_text( json.dumps(items, ensure_ascii=False, indent=2), encoding="utf-8", ) os.chmod(tmp, 0o600) os.replace(tmp, REVIEW_QUEUE_PATH) finally: if tmp.exists(): tmp.unlink(missing_ok=True) def _load_raw_stats_trade_policy(): if not STATS_POLICY_PATH.exists(): return {} data = json.loads(STATS_POLICY_PATH.read_text(encoding="utf-8-sig")) if not isinstance(data, dict): raise ValueError("Policy statistiche non valida") return data def _save_raw_stats_trade_policy(data): """Backup e sostituzione atomica. Non modifica DB webhook o ordini.""" STATS_POLICY_BACKUP_DIR.mkdir(parents=True, exist_ok=True) os.chmod(STATS_POLICY_BACKUP_DIR, 0o700) if STATS_POLICY_PATH.exists(): stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S_%f") backup = STATS_POLICY_BACKUP_DIR / f"v2_stats_trade_policy_{stamp}.json" backup.write_bytes(STATS_POLICY_PATH.read_bytes()) os.chmod(backup, 0o600) tmp = STATS_POLICY_PATH.with_name( f".{STATS_POLICY_PATH.name}.{secrets.token_hex(6)}.tmp" ) try: tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") os.chmod(tmp, 0o600) os.replace(tmp, STATS_POLICY_PATH) finally: tmp.unlink(missing_ok=True) def _append_certification_audit(event): event = dict(event) event.setdefault("timestamp", datetime.now(timezone.utc).isoformat()) fd = os.open( str(CERTIFICATION_AUDIT_PATH), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600, ) with os.fdopen(fd, "a", encoding="utf-8") as handle: handle.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n") def _parse_review_form(content_type, body): if content_type.startswith("application/x-www-form-urlencoded"): raw = body.decode("utf-8", errors="strict") values = parse_qs(raw, keep_blank_values=True) return {key: value[-1] if value else "" for key, value in values.items()}, None if not content_type.startswith("multipart/form-data"): raise ValueError("Formato richiesta non supportato") message = BytesParser(policy=email_policy).parsebytes( b"Content-Type: " + content_type.encode("ascii", errors="strict") + b"\r\nMIME-Version: 1.0\r\n\r\n" + body ) if not message.is_multipart(): raise ValueError("Upload multipart non valido") form = {} attachment = None for part in message.iter_parts(): name = part.get_param("name", header="content-disposition") if not name: continue payload = part.get_payload(decode=True) or b"" filename = part.get_filename() if filename: if name != "evidence_file" or attachment is not None: raise ValueError("Allegato non valido") attachment = { "original_name": str(filename), "content_type": str(part.get_content_type() or ""), "data": payload, } else: charset = part.get_content_charset() or "utf-8" form[str(name)] = payload.decode(charset, errors="strict") return form, attachment def _validate_review_attachment(attachment): if not attachment or not attachment.get("data"): return None data = attachment["data"] if len(data) > MAX_EVIDENCE_BYTES: raise ValueError("Screenshot troppo grande: massimo 4 MB") if data.startswith(b"\x89PNG\r\n\x1a\n"): detected_type, extension = "image/png", ".png" elif data.startswith(b"\xff\xd8\xff"): detected_type, extension = "image/jpeg", ".jpg" elif len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP": detected_type, extension = "image/webp", ".webp" else: raise ValueError("Formato screenshot non valido: usare PNG, JPG o WEBP") declared_type = str(attachment.get("content_type") or "").lower() if declared_type not in (detected_type, "application/octet-stream"): raise ValueError("Il tipo dichiarato dello screenshot non corrisponde al file") return { "original_name": Path(str(attachment.get("original_name") or "screenshot")).name[:180], "content_type": detected_type, "extension": extension, "size": len(data), "sha256": hashlib.sha256(data).hexdigest(), "data": data, } def _save_review_attachment(proposal_id, attachment): REVIEW_EVIDENCE_DIR.mkdir(parents=True, exist_ok=True) os.chmod(REVIEW_EVIDENCE_DIR, 0o700) stored_name = f"{proposal_id}{attachment['extension']}" target = REVIEW_EVIDENCE_DIR / stored_name fd = os.open(str(target), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(fd, "wb") as handle: handle.write(attachment["data"]) except Exception: target.unlink(missing_ok=True) raise metadata = {key: value for key, value in attachment.items() if key != "data"} metadata["stored_name"] = stored_name return metadata, target def _rejected_order_ids(): return { str(item.get("order_id") or "").strip() for item in _load_review_queue() if item.get("status") == "pending" and item.get("action") == "certification_rejected" and str(item.get("order_id") or "").strip() } def _strategy_history_for_symbol(data, symbol): counts = defaultdict(int) for row in data.get("asset_trade_details", {}).get(symbol, []): strategy = _canonical_strategy(row.get("strategy")) if strategy in VALID_CLOSED_TRADE_STRATEGIES: counts[strategy] += 1 return dict(sorted(counts.items(), key=lambda item: (-item[1], item[0]))) def _archived_review_order_ids(): """Current cases cleared by the operator; new order IDs remain visible.""" return { str(item.get("order_id") or "").strip() for item in _load_review_queue() if item.get("status") == "archived" and item.get("action") == "clear_review" } def _certification_candidates(data): """Genera suggerimenti, mai attribuzioni automatiche.""" rejected = _rejected_order_ids() archived = _archived_review_order_ids() candidates = [] for row in data.get("unclassified_trades", []): order_id = str(row.get("order_id") or "").strip() if not order_id or order_id in rejected or order_id in archived: continue item = dict(row) symbol = str(item.get("symbol") or "").upper() history = _strategy_history_for_symbol(data, symbol) configured = _canonical_strategy(get_strategy_for_symbol(symbol)) proposed = "" confidence = "LOW" reasons = [] if history: proposed = next(iter(history)) total = sum(history.values()) top_count = history[proposed] if len(history) == 1: confidence = "HIGH" if configured == proposed else "MEDIUM" elif top_count > total / 2: confidence = "MEDIUM" else: confidence = "LOW" reasons.append("storico asset: " + ", ".join( f"{name.upper()}={count}" for name, count in history.items() )) if configured in VALID_CLOSED_TRADE_STRATEGIES: reasons.append(f"config attuale: {configured.upper()}") if not proposed: proposed = configured confidence = "MEDIUM" elif configured != proposed: confidence = "LOW" reasons.append("conflitto storico/config") if not proposed: reasons.append("nessuna prova storica o configurazione univoca") item.update({ "anomaly_type": "PENDING_CERTIFICATION", "proposed_strategy": proposed, "proposal_confidence": confidence, "proposal_evidence": "; ".join(reasons), "proposal_alternatives": list(history), }) candidates.append(item) return candidates def _submitted_review_order_ids(): """Decisions already submitted stay in the proposal queue, not new cases.""" latest = {} for item in _load_review_queue(): order_id = str(item.get("order_id") or "").strip() if order_id and item.get("action") in REVIEW_ACTIONS: latest[order_id] = item return { order_id for order_id, item in latest.items() if item.get("action") in {"classify", "exclude_test", "exclude_operational"} and item.get("status") in {"applied", "certified", "resolved"} } def _review_candidates(data): """Solo casi rifiutati: i nuovi trade restano nella coda certificazioni.""" rejected = _rejected_order_ids() archived = _archived_review_order_ids() submitted = _submitted_review_order_ids() candidates = [] for row in data.get("unclassified_trades", []): order_id = str(row.get("order_id") or "").strip() if order_id and order_id in rejected and order_id not in archived and order_id not in submitted: item = dict(row) item["anomaly_type"] = "PROPOSTA_RIFIUTATA" candidates.append(item) return candidates def _current_unclassified(data, order_id): order_id = str(order_id or "").strip() for row in data.get("unclassified_trades", []): if str(row.get("order_id") or "").strip() == order_id: return row raise ValueError("Trade non presente tra le classificazioni sospese") def _resolve_queue_for_order(queue, order_id, status): now = datetime.now(timezone.utc).isoformat() for item in queue: if item.get("status") == "pending" and item.get("order_id") == order_id: item["status"] = status item["resolved_at"] = now return queue def _certify_trade(form, data): order_id = str(form.get("order_id") or "").strip() strategy = _canonical_strategy(form.get("strategy")) if strategy not in VALID_CLOSED_TRADE_STRATEGIES: raise ValueError("Strategia obbligatoria o non valida") trade = _current_unclassified(data, order_id) proposal = next( (row for row in _certification_candidates(data) if row.get("order_id") == order_id), None, ) if proposal is None: raise ValueError("Caso gia rifiutato o non piu certificabile") policy = _load_raw_stats_trade_policy() mapping = policy.setdefault("strategy_by_closed_order_id", {}) sources = policy.setdefault("classification_source_by_closed_order_id", {}) existing = _canonical_strategy(mapping.get(order_id)) if existing and existing != strategy: raise ValueError("Order ID gia certificato con strategia diversa") mapping[order_id] = strategy sources[order_id] = "operator_certified_from_dashboard" _save_raw_stats_trade_policy(policy) queue = _resolve_queue_for_order(_load_review_queue(), order_id, "certified") _save_review_queue(queue) _append_certification_audit({ "event": "CERTIFIED", "operator": AUTH_USER, "order_id": order_id, "symbol": trade.get("symbol", ""), "proposed_strategy": proposal.get("proposed_strategy", ""), "chosen_strategy": strategy, "confidence": proposal.get("proposal_confidence", ""), "evidence": proposal.get("proposal_evidence", ""), }) log(f"trade certified order={order_id} strategy={strategy}") def _reject_certification(form, data): order_id = str(form.get("order_id") or "").strip() trade = _current_unclassified(data, order_id) if order_id in _rejected_order_ids(): raise ValueError("Trade gia presente nelle anomalie") proposal = next( (row for row in _certification_candidates(data) if row.get("order_id") == order_id), None, ) or {} queue = _load_review_queue() queue.append({ "proposal_id": f"reject-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}-{secrets.token_hex(4)}", "created_at": datetime.now(timezone.utc).isoformat(), "status": "pending", "action": "certification_rejected", "strategy": "", "reason": "PROPOSTA_RIFIUTATA_DA_OPERATORE", "evidence": proposal.get("proposal_evidence", ""), "notes": "Da approfondire nella gestione anomalie", "order_id": order_id, "trade_snapshot": dict(trade), "proposal_snapshot": { "strategy": proposal.get("proposed_strategy", ""), "confidence": proposal.get("proposal_confidence", ""), "evidence": proposal.get("proposal_evidence", ""), }, }) _save_review_queue(queue) _append_certification_audit({ "event": "REJECTED_TO_ANOMALY", "operator": AUTH_USER, "order_id": order_id, "symbol": trade.get("symbol", ""), "proposed_strategy": proposal.get("proposed_strategy", ""), "confidence": proposal.get("proposal_confidence", ""), }) log(f"trade rejected to anomaly order={order_id}") def _apply_review_decision(proposal): """Apply the operator's explicit decision to this exact closed order only.""" action = proposal.get("action") if action not in {"classify", "exclude_test", "exclude_operational"}: return False oid = str(proposal.get("order_id") or "").strip() if not oid or not proposal.get("reason") or not (proposal.get("evidence") or proposal.get("evidence_attachment")): raise ValueError("Decisione priva di ID, motivo o prova") policy = _load_raw_stats_trade_policy() if action == "classify": strategy = _canonical_strategy(proposal.get("strategy")) if strategy not in VALID_CLOSED_TRADE_STRATEGIES: raise ValueError("Strategia obbligatoria o non valida") existing = _canonical_strategy(policy.get("strategy_by_closed_order_id", {}).get(oid)) if existing and existing != strategy: raise ValueError("Order ID gia certificato con strategia diversa") if oid in policy.get("excluded_closed_order_ids", []) or oid in policy.get("operational_excluded_closed_order_ids", {}): raise ValueError("Order ID gia escluso: decisioni in conflitto") policy.setdefault("strategy_by_closed_order_id", {})[oid] = strategy policy.setdefault("classification_source_by_closed_order_id", {})[oid] = "review_proposal:" + proposal["proposal_id"] else: if oid in policy.get("strategy_by_closed_order_id", {}): raise ValueError("Order ID gia certificato: decisioni in conflitto") if action == "exclude_test": ids = policy.setdefault("excluded_closed_order_ids", []) if oid not in ids: ids.append(oid) policy.setdefault("exclusion_reasons", {})[oid] = proposal["reason"] else: policy.setdefault("operational_excluded_closed_order_ids", {})[oid] = { "reason": proposal["reason"], "proposal_id": proposal["proposal_id"], "approved_by": AUTH_USER, "approved_at": datetime.now(timezone.utc).isoformat()} _save_raw_stats_trade_policy(policy) _append_certification_audit({"event": "REVIEW_APPLIED", "operator": AUTH_USER, "order_id": oid, "action": action, "chosen_strategy": proposal.get("strategy", ""), "proposal_id": proposal["proposal_id"], "reason": proposal["reason"], "evidence": proposal.get("evidence", "")}) return True def _create_review_proposal(form, data, attachment=None): order_id = str(form.get("order_id") or "").strip() action = str(form.get("action") or "").strip().lower() strategy = _canonical_strategy(form.get("strategy")) reason = str(form.get("reason") or "").strip() evidence = str(form.get("evidence") or "").strip() notes = str(form.get("notes") or "").strip() candidates = {str(row.get("order_id")): row for row in _review_candidates(data)} trade = candidates.get(order_id) if trade is None: raise ValueError("Caso non presente tra le anomalie correnti") if action not in REVIEW_ACTIONS: raise ValueError("Azione non valida") if action == "classify" and strategy not in VALID_CLOSED_TRADE_STRATEGIES: raise ValueError("Strategia obbligatoria o non valida") if action != "classify": strategy = "" if not reason: raise ValueError("Motivo obbligatorio") validated_attachment = _validate_review_attachment(attachment) if not evidence and validated_attachment is None: raise ValueError("Inserire una prova testuale oppure allegare uno screenshot") queue = _load_review_queue() if any( item.get("status") == "pending" and item.get("order_id") == order_id and item.get("action") == action for item in queue ): raise ValueError("Esiste gia una proposta pendente identica per questo trade") proposal_id = f"review-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}-{secrets.token_hex(4)}" proposal = { "proposal_id": proposal_id, "created_at": datetime.now(timezone.utc).isoformat(), "status": "pending", "action": action, "strategy": strategy, "reason": reason, "evidence": evidence, "notes": notes, "order_id": order_id, "trade_snapshot": { "anomaly_type": trade.get("anomaly_type", ""), "symbol": trade.get("symbol", ""), "side": trade.get("side", ""), "closed_size": trade.get("closed_size", trade.get("qty", 0)), "avg_entry_price": trade.get("avg_entry_price", trade.get("entry_price", 0)), "entry_time": trade.get("entry_time", ""), "created_at": trade.get("created_at", ""), "updated_at": trade.get("updated_at", ""), "pnl_net": trade.get("pnl_net", trade.get("pnl", 0)), "classification_source": trade.get("classification_source", ""), "operational_reason": trade.get("operational_reason", ""), }, } attachment_path = None try: if validated_attachment is not None: metadata, attachment_path = _save_review_attachment(proposal_id, validated_attachment) proposal["evidence_attachment"] = metadata proposal["status"] = "applying" if action != "needs_review" else "pending" queue.append(proposal) # Persist intent first, preserving recovery evidence if policy writing fails. _save_review_queue(queue) if _apply_review_decision(proposal): proposal["status"] = "pending" _resolve_queue_for_order(queue, order_id, "applied") _save_review_queue(queue) except Exception: if proposal in queue: proposal["status"] = "failed" _save_review_queue(queue) if attachment_path is not None: attachment_path.unlink(missing_ok=True) raise log( f"review proposal saved id={proposal['proposal_id']} order={order_id} " f"action={action} strategy={strategy or '-'}" ) return proposal def render_review_dashboard(data, message="", error="", main_url="/"): esc = lambda value: html.escape(str(value if value is not None else ""), quote=True) safe_main_url = esc(main_url if str(main_url).startswith("/") else "/") certifications = _certification_candidates(data) anomalies = _review_candidates(data) queue = _load_review_queue() pending_queue = [item for item in queue if item.get("status") == "pending"] def strategy_options(selected=""): return "".join( f"" for strategy in sorted(VALID_CLOSED_TRADE_STRATEGIES) ) certification_rows = "".join( "{closed}{symbol}{side}{qty}" "{pnl}{confidence}" "{evidence}
" "" "" "
" "
" "
".format( base=safe_main_url, closed=esc(_fmt_review_ts(row.get("updated_at"))), symbol=esc(row.get("symbol", "")), side=esc(row.get("side", "")), qty=esc(row.get("closed_size", row.get("qty", 0))), pnl=esc(_fmt_usdt(row.get("pnl_net", row.get("pnl", 0)))) + " USDT", confidence=esc(row.get("proposal_confidence", "LOW")), confidence_class=esc(str(row.get("proposal_confidence", "LOW")).lower()), evidence=esc(row.get("proposal_evidence", "")), order=esc(row.get("order_id", "")), options=strategy_options(row.get("proposed_strategy", "")), ) for row in certifications ) or "Nessun trade da certificare." candidate_options = "".join( "".format( order=esc(row.get("order_id", "")), kind=esc(row.get("anomaly_type", "")), symbol=esc(row.get("symbol", "")), side=esc(row.get("side", "")), pnl=esc(_fmt_usdt(row.get("pnl_net", row.get("pnl", 0)))), ) for row in anomalies ) anomaly_rows = "".join( "{kind}{symbol}{side}{qty}" "{entry}{opened}{closed}" "{order}{pnl}".format( kind=esc(row.get("anomaly_type", "")), symbol=esc(row.get("symbol", "")), side=esc(row.get("side", "")), qty=esc(row.get("closed_size", row.get("qty", 0))), entry=esc(row.get("avg_entry_price", row.get("entry_price", 0))), opened=esc(_fmt_review_ts(row.get("entry_time") or row.get("created_at"))), closed=esc(_fmt_review_ts(row.get("updated_at"))), order=esc(row.get("order_id", "")), pnl=esc(_fmt_usdt(row.get("pnl_net", row.get("pnl", 0)))), ) for row in anomalies ) or "Nessun caso anomalo corrente." queue_rows = "".join( "{created}{status}{action}{strategy}" "{symbol}{order}{reason}{evidence}".format( created=esc(item.get("created_at", "")), status=esc(item.get("status", "")), action=esc(item.get("action", "")), strategy=esc(item.get("strategy", "")), symbol=esc(item.get("trade_snapshot", {}).get("symbol", "")), order=esc(item.get("order_id", "")), reason=esc(item.get("reason", "")), evidence=( esc(item.get("evidence", "")) + ( "
Apri screenshot".format( name=esc(item.get("evidence_attachment", {}).get("stored_name", "")) ) if item.get("evidence_attachment", {}).get("stored_name") else "" ) ), ) for item in reversed(pending_queue) ) or "Nessuna proposta registrata." message_html = f"
{esc(message)}
" if message else "" error_html = f"
{esc(error)}
" if error else "" disabled = " disabled" if not anomalies else "" return f""" V2 Certificazioni e anomalie

V2 Certificazioni e anomalie

Torna alla dashboard principale

Il sistema propone l'abbinamento. Certifica e Registra decisione applicano la scelta ai dati; No sposta il caso nelle anomalie. Richiede verifica lascia il caso aperto. Nessun effetto su ordini o posizioni.

{message_html}{error_html}

Da certificare ({len(certifications)})

{certification_rows}
ChiusuraAssetSideQtyNetConfidenzaProva propostaDecisione

Anomalie reali ({len(anomalies)})

{anomaly_rows}
TipoAssetSideQtyEntryApertura (Roma)Chiusura (Roma)Order IDNet

Risoluzione manuale anomalia

Proposte da risolvere ({len(pending_queue)})

Esporta audit JSON
{queue_rows}
CreataStatoAzioneStrategiaAssetOrder IDMotivoProva
""" # === DATA LOADERS === def get_bybit_positions(): try: sys.path.insert(0, str(ROOT)) from bybit_demo_client import BybitDemoClient c = BybitDemoClient() return c.fetch_positions() except Exception as e: log(f"bybit positions err: {e}") return [] def get_bybit_balance(): try: sys.path.insert(0, str(ROOT)) from bybit_demo_client import BybitDemoClient c = BybitDemoClient() b = c._request("GET", "/v5/account/wallet-balance", {"accountType": "UNIFIED"}, signed=True) coin = b["result"]["list"][0]["coin"][0] return float(coin["walletBalance"]) except Exception as e: log(f"bybit balance err: {e}") return 0.0 # Lista dei trade/posizioni di MIO TEST Pine-style da escludere dalla dashboard MY_TEST_FILTER = { "symbols_qty": { # (symbol, qty) -> True se e' MIO test ("VIRTUALUSDT", 1500.0): True, # VIRTUAL placeholder 1500 (mio test Pine-style) ("NEARUSDT", 599.9): True, # NEAR mio test ma_trailing }, "symbols": { "VIRTUALUSDT", # symbol sospetto placeholder price=1.0 }, } def _canonical_strategy(raw): value = str(raw or "").strip().lower() while value.endswith("_close"): value = value[:-6] aliases = { "vptr_v3": "vptr3", "rettangolo_v2": "rettangolo", "rettangolo_tv_simple": "rettangolo_simple", "rett_simple": "rettangolo_simple", "rettangolo semplice": "rettangolo_simple", "rsi swing breakout": "rsi_swing_breakout", "ma trailing": "ma_trailing", "range": "range_fib", "range fib": "range_fib", "adx_dmi_bb_vol": "adx", "supertrend extensions": "supertrend_bosw", "supertrend bosw": "supertrend_bosw", "manual": "manuale", } return aliases.get(value, value) VALID_CLOSED_TRADE_STRATEGIES = frozenset({ "vptr3", "rettangolo", "rettangolo_simple", "ma_trailing", "rsi_swing_breakout", "sqw", "range_fib", "adx", "supertrend_bosw", "manuale", }) UNCLASSIFIED_STRATEGY = "unclassified" def _load_stats_trade_policy(): """Regole storiche per orderId. Non modifica il DB operativo webhook.""" empty = { "excluded_closed_order_ids": set(), "operational_excluded_closed_order_ids": {}, "excluded_trade_fingerprints": [], "operational_excluded_trade_fingerprints": [], "strategy_by_closed_order_id": {}, "strategy_by_trade_fingerprint": [], "blocked_assets": {}, } if not STATS_POLICY_PATH.exists(): return empty try: data = json.loads(STATS_POLICY_PATH.read_text(encoding="utf-8-sig")) excluded = { str(value).strip() for value in data.get("excluded_closed_order_ids", []) if str(value).strip() } mapping = { str(order_id).strip(): _canonical_strategy(strategy) for order_id, strategy in data.get("strategy_by_closed_order_id", {}).items() if str(order_id).strip() and _canonical_strategy(strategy) } operational_by_order_id = {} raw_operational_ids = data.get("operational_excluded_closed_order_ids", {}) if isinstance(raw_operational_ids, dict): for order_id, rule in raw_operational_ids.items(): order_id = str(order_id or "").strip() if not order_id: continue if isinstance(rule, dict): operational_by_order_id[order_id] = dict(rule) else: operational_by_order_id[order_id] = {"reason": str(rule or "").strip()} blocked_assets = {} for symbol, rule in data.get("blocked_assets", {}).items(): symbol = str(symbol or "").strip().upper() if not symbol or not isinstance(rule, dict): continue effective_from = str(rule.get("effective_from") or "").strip() if _parse_bybit_ts(effective_from) is None: log(f"blocked asset policy ignored: invalid effective_from for {symbol}") continue blocked_assets[symbol] = { "effective_from": effective_from, "reason": str(rule.get("reason") or "blocked_asset").strip(), "scope": "new_entries", } return { "excluded_closed_order_ids": excluded, "operational_excluded_closed_order_ids": operational_by_order_id, "excluded_trade_fingerprints": [ item for item in data.get("excluded_trade_fingerprints", []) if isinstance(item, dict) ], "operational_excluded_trade_fingerprints": [ item for item in data.get("operational_excluded_trade_fingerprints", []) if isinstance(item, dict) ], "strategy_by_closed_order_id": mapping, "strategy_by_trade_fingerprint": [ item for item in data.get("strategy_by_trade_fingerprint", []) if isinstance(item, dict) and _canonical_strategy(item.get("strategy")) in VALID_CLOSED_TRADE_STRATEGIES ], "blocked_assets": blocked_assets, } except Exception as exc: log(f"stats trade policy err: {exc}") return empty def _fingerprint_float_equal(actual, expected, explicit_tolerance=None): try: actual = float(actual) expected = float(expected) except (TypeError, ValueError): return False if explicit_tolerance not in (None, ""): try: tolerance = abs(float(explicit_tolerance)) except (TypeError, ValueError): return False else: tolerance = max(1e-8, abs(expected) * 1e-6) return abs(actual - expected) <= tolerance def _matches_trade_fingerprint(closed_item, fingerprint): """Confronta l'intero trade. Non consente mai una regola basata sul solo asset.""" required = ("symbol", "side", "closed_size", "avg_entry_price") if any(fingerprint.get(key) in (None, "") for key in required): return False if str(closed_item.get("symbol") or "").strip().upper() != str( fingerprint["symbol"] ).strip().upper(): return False if str(closed_item.get("side") or "").strip().upper() != str( fingerprint["side"] ).strip().upper(): return False if not _fingerprint_float_equal( closed_item.get("closedSize"), fingerprint["closed_size"] ): return False if not _fingerprint_float_equal( closed_item.get("avgEntryPrice"), fingerprint["avg_entry_price"], fingerprint.get("avg_entry_price_tolerance"), ): return False if fingerprint.get("avg_exit_price") not in (None, "") and not _fingerprint_float_equal( closed_item.get("avgExitPrice"), fingerprint["avg_exit_price"] ): return False if fingerprint.get("closed_pnl") not in (None, "") and not _fingerprint_float_equal( closed_item.get("closedPnl"), fingerprint["closed_pnl"], fingerprint.get("closed_pnl_tolerance"), ): return False if fingerprint.get("closed_at_from") or fingerprint.get("closed_at_to"): try: closed_ms = int(float(closed_item.get("updatedTime") or closed_item.get("createdAt") or 0)) if not closed_ms: return False if fingerprint.get("closed_at_from"): start_ms = int( datetime.fromisoformat( str(fingerprint["closed_at_from"]).replace("Z", "+00:00") ).timestamp() * 1000 ) if closed_ms < start_ms: return False if fingerprint.get("closed_at_to"): end_ms = int( datetime.fromisoformat( str(fingerprint["closed_at_to"]).replace("Z", "+00:00") ).timestamp() * 1000 ) if closed_ms > end_ms: return False except (TypeError, ValueError, OverflowError): return False return True def _matching_fingerprints(closed_item, fingerprints): return [item for item in fingerprints if _matches_trade_fingerprint(closed_item, item)] def _strategy_from_order_db(order_id): """Usa esclusivamente la strategia registrata sullo stesso orderId.""" if not order_id or not DB_PATH.exists(): return "" conn = None try: conn = sqlite3.connect(DB_PATH) row = conn.execute( "SELECT strategy FROM orders WHERE order_id = ? ORDER BY id DESC LIMIT 1", (str(order_id),), ).fetchone() return _canonical_strategy(row[0]) if row and row[0] else "" except Exception as exc: log(f"strategy order lookup err orderId={order_id}: {exc}") return "" finally: if conn is not None: conn.close() def _strict_closed_trade_strategy(order_id, certified_mapping): """Classifica solo con prova univoca per orderId, mai in base al simbolo.""" order_id = str(order_id or "").strip() if not order_id: return UNCLASSIFIED_STRATEGY, "missing_order_id" certified = _canonical_strategy(certified_mapping.get(order_id)) if certified in VALID_CLOSED_TRADE_STRATEGIES: return certified, "certified_order_id" recorded = _canonical_strategy(_strategy_from_order_db(order_id)) if recorded in VALID_CLOSED_TRADE_STRATEGIES: return recorded, "exact_order_id_db" if certified or recorded: return UNCLASSIFIED_STRATEGY, "invalid_strategy_for_order_id" return UNCLASSIFIED_STRATEGY, "order_id_not_registered" def _unique_entry_lifecycle_evidence(closed_item): """Trova una sola entrata usando l'intera firma del trade, mai il solo asset.""" close_order_id = str(closed_item.get("orderId") or "").strip() if close_order_id: exact_conn = None try: exact_conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) exact_conn.row_factory = sqlite3.Row table_exists = exact_conn.execute( "SELECT 1 FROM sqlite_master WHERE type='table' " "AND name='closed_trade_entry_evidence'" ).fetchone() exact_row = exact_conn.execute( "SELECT close_order_id, entry_order_id, symbol, open_side, qty, " "entry_price, strategy, opened_at, source " "FROM closed_trade_entry_evidence WHERE close_order_id = ?", (close_order_id,), ).fetchone() if table_exists else None except Exception as exc: log(f"exact entry evidence lookup err orderId={close_order_id}: {exc}") return None, "exact_entry_evidence_db_error" finally: if exact_conn is not None: exact_conn.close() if exact_row is not None: symbol = str(closed_item.get("symbol") or "").strip().upper() close_side = str(closed_item.get("side") or "").strip().upper() expected_open_side = "BUY" if close_side == "SELL" else "SELL" if close_side == "BUY" else "" try: closed_qty = abs(float(closed_item.get("closedSize") or 0)) closed_entry = float(closed_item.get("avgEntryPrice") or 0) stored_qty = abs(float(exact_row["qty"] or 0)) stored_entry = float(exact_row["entry_price"] or 0) except (TypeError, ValueError): return None, "exact_entry_evidence_invalid_numeric" opened_at = _parse_bybit_ts(exact_row["opened_at"]) strategy = _canonical_strategy(exact_row["strategy"]) valid = ( str(exact_row["symbol"] or "").strip().upper() == symbol and str(exact_row["open_side"] or "").strip().upper() == expected_open_side and abs(stored_qty - closed_qty) <= max(1e-8, closed_qty * 1e-6) and abs(stored_entry - closed_entry) <= max(1e-10, abs(closed_entry) * 5e-5) and opened_at is not None and strategy in VALID_CLOSED_TRADE_STRATEGIES ) if not valid: return None, "exact_entry_evidence_signature_mismatch" return { "entry_order_id": str(exact_row["entry_order_id"] or "").strip(), "request_id": "", "entry_time": opened_at, "entry_time_raw": str(exact_row["opened_at"] or ""), "strategy": strategy, }, "exact_closed_order_entry_evidence" if not DB_PATH.exists(): return None, "entry_db_missing" symbol = str(closed_item.get("symbol") or "").strip().upper() close_side = str(closed_item.get("side") or "").strip().upper() open_side = "BUY" if close_side == "SELL" else "SELL" if close_side == "BUY" else "" close_dt = _parse_bybit_ts( closed_item.get("updatedTime") or closed_item.get("createdAt") ) try: qty = abs(float(closed_item.get("closedSize") or 0)) entry_price = float(closed_item.get("avgEntryPrice") or 0) except (TypeError, ValueError): return None, "entry_signature_invalid" if not symbol or not open_side or close_dt is None or qty <= 0 or entry_price <= 0: return None, "entry_signature_incomplete" conn = None try: conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT order_id, request_id, symbol, side, qty, price, strategy, created_at " "FROM orders WHERE UPPER(symbol) = ? AND UPPER(side) = ? ORDER BY id DESC", (symbol, open_side), ).fetchall() except Exception as exc: log(f"entry lifecycle lookup err symbol={symbol}: {exc}") return None, "entry_db_error" finally: if conn is not None: conn.close() qty_tolerance = max(1e-8, qty * 1e-6) price_tolerance = max(1e-10, abs(entry_price) * 5e-5) candidates = [] for row in rows: raw_strategy = str(row["strategy"] or "").strip() order_id = str(row["order_id"] or "").strip() if not order_id or order_id in { "no_position", "anti_dup_skip", "render_blocked", "render_short_blocked" }: continue if raw_strategy.lower().endswith("_close"): continue entry_dt = _parse_bybit_ts(row["created_at"]) if entry_dt is None or entry_dt > close_dt: continue try: row_qty = abs(float(row["qty"] or 0)) row_price = float(row["price"] or 0) except (TypeError, ValueError): continue if abs(row_qty - qty) > qty_tolerance: continue if abs(row_price - entry_price) > price_tolerance: continue strategy = _canonical_strategy(raw_strategy) if strategy not in VALID_CLOSED_TRADE_STRATEGIES: continue candidates.append({ "entry_order_id": order_id, "request_id": str(row["request_id"] or "").strip(), "entry_time": entry_dt, "entry_time_raw": str(row["created_at"] or ""), "strategy": strategy, }) if len(candidates) != 1: return None, f"entry_lifecycle_candidates_{len(candidates)}" return candidates[0], "unique_entry_lifecycle" def _classify_closed_item(closed_item, policy): """Classifica il trade e applica il blocco all'ora di entrata, non di chiusura.""" close_order_id = str(closed_item.get("orderId") or "").strip() exact_operational = policy.get("operational_excluded_closed_order_ids", {}).get( close_order_id ) if exact_operational is not None: reason = str( exact_operational.get("reason") or "APPROVED_OPERATIONAL_EXCLUSION" ).strip() return { "strategy": UNCLASSIFIED_STRATEGY, "source": "operational_exclusion_closed_order_id", "performance_excluded": True, "operational_reason": reason, "entry_evidence": None, } operational_matches = _matching_fingerprints( closed_item, policy.get("operational_excluded_trade_fingerprints", []) ) if operational_matches: reason = str( operational_matches[0].get("reason") or "UNAUTHORIZED_ORDER_WITHOUT_TV_ALERT" ).strip() return { "strategy": UNCLASSIFIED_STRATEGY, "source": "operational_exclusion_fingerprint", "performance_excluded": True, "operational_reason": reason, "entry_evidence": None, } exact_strategy, exact_source = _strict_closed_trade_strategy( close_order_id, policy["strategy_by_closed_order_id"], ) entry_evidence, entry_source = _unique_entry_lifecycle_evidence(closed_item) symbol = str(closed_item.get("symbol") or "").strip().upper() block_rule = policy.get("blocked_assets", {}).get(symbol) if block_rule: blocked_from = _parse_bybit_ts(block_rule["effective_from"]) close_dt = _parse_bybit_ts( closed_item.get("updatedTime") or closed_item.get("createdAt") ) # Una chiusura precedente al blocco prova gia' che anche l'entrata era # precedente. La prova lifecycle rafforzata serve solo da quel momento in poi. if blocked_from is not None and (close_dt is None or close_dt >= blocked_from): if entry_evidence is None: return { "strategy": UNCLASSIFIED_STRATEGY, "source": f"blocked_asset_entry_unverified:{entry_source}", "performance_excluded": False, "operational_reason": "", "entry_evidence": None, } if entry_evidence["entry_time"] >= blocked_from: return { "strategy": UNCLASSIFIED_STRATEGY, "source": "blocked_entry_after_effective_from", "performance_excluded": True, "operational_reason": "BLOCKED_ASSET_ORDER_LEAK", "entry_evidence": entry_evidence, } fingerprint_matches = _matching_fingerprints( closed_item, policy.get("strategy_by_trade_fingerprint", []) ) fingerprint_strategies = { _canonical_strategy(item.get("strategy")) for item in fingerprint_matches } if exact_strategy in VALID_CLOSED_TRADE_STRATEGIES: strategy, source = exact_strategy, exact_source elif len(fingerprint_strategies) == 1: strategy = fingerprint_strategies.pop() source = str( fingerprint_matches[0].get("source") or "certified_trade_fingerprint" ).strip() elif len(fingerprint_strategies) > 1: strategy, source = UNCLASSIFIED_STRATEGY, "conflicting_trade_fingerprints" elif entry_evidence is not None: strategy, source = entry_evidence["strategy"], entry_source else: strategy, source = UNCLASSIFIED_STRATEGY, f"{exact_source}:{entry_source}" return { "strategy": strategy, "source": source, "performance_excluded": False, "operational_reason": "", "entry_evidence": entry_evidence, } # Data di partenza: 1° agosto 2026 SINCE_DATE_MS = int(__import__("datetime").datetime(2026, 8, 1, 0, 0, 0, tzinfo=__import__("datetime").timezone.utc).timestamp() * 1000) def is_mio_test_trade(trade): """True se il trade chiuso Bybit corrisponde a un MIO test Pine-style.""" sym = trade.get("symbol", "") qty = float(trade.get("closedSize", 0) or 0) if (sym, qty) in MY_TEST_FILTER["symbols_qty"] and (sym, qty) in MY_TEST_FILTER["symbols_qty"]: return True avg_entry = float(trade.get("avgEntryPrice", 0) or 0) # VIRTUAL placeholder Pine: avgEntry ~ 1.0 (placeholder Pine) o avgEntry 0.5711 (eseguito Bybit mio test) if sym == "VIRTUALUSDT" and abs(avg_entry - 1.0) < 1e-6 and abs(qty - 1500.0) < 0.01: return True if sym == "NEARUSDT" and abs(qty - 599.9) < 0.01: return True if sym == "BTCUSDT" and abs(qty - 0.023) < 1e-6: return True return False def is_mio_test_position(pos): """True se la posizione aperta e' un MIO test Pine-style. FIX 2026-08-10 (Mattia): rimosso BTCUSDT 0.023 dal filtro perche' era una posizione reale del 2° account, non un test. Tenuto NEARUSDT 599.9 e VIRTUALUSDT 1500 (quei SI sono test).""" sym = pos.get("symbol", "") try: size = float(pos.get("size", 0) or 0) except (TypeError, ValueError): return False if sym == "NEARUSDT" and abs(size - 599.9) < 0.01: return True if sym == "VIRTUALUSDT" and abs(size - 1500.0) < 0.01: return True return False def get_closed_trades_bybit(): """Legge TUTTI i trade chiusi da Bybit API (closed-pnl endpoint), filtra dal 1° agosto, esclude MIEI test Pine-style. FIX 08/08 (Mattia): NO cursor-based paging perche' bybit_demo_client._request calcola firma sul cursor RAW ma `requests.get` ri-encoda il `%3A` -> err 10004. Uso finestre di 1 giorno: ~5 trade/giorno, no rischio limite 50. FIX 12/08 (Mattia, REVISIONE FINALE #3): closed_trade.strategy = CLOSED_TRADES_STRATEGY_OVERRIDE (hardcoded, esplicito) per i simboli in cui il CSV NON riflette la SOURCE Pine TV al momento: - WIF: CSV dice rsi_swing ma Pine WIF = RETTANGOLO (Python/SL engine, non Pine RSI swing) - VIRTUAL: VPTR3 enabled=false ma Pine VIRTUAL = VPTR3 (aperti quando era attivo) - AEO: CSV dice rsi_swing ma Pine AEO = RSI swing SOLO con SL/TP (per Mattia lo conteggia RSI swing) Per TUTTI gli altri simboli: get_strategy_for_symbol (CSV con priorita').""" out = [] policy = _load_stats_trade_policy() excluded_order_ids = policy["excluded_closed_order_ids"] excluded_fingerprints = policy.get("excluded_trade_fingerprints", []) strategy_by_order_id = policy["strategy_by_closed_order_id"] try: sys.path.insert(0, str(ROOT)) from bybit_demo_client import BybitDemoClient c = BybitDemoClient() # Endpoint Bybit V5: GET /v5/position/closed-pnl now_ms = int(__import__("datetime").datetime.now(__import__("datetime").timezone.utc).timestamp() * 1000) # FIX 08/08: finestre 1gg (no cursor, no err 10004 doppio encoding) WINDOW_MS = 1 * 24 * 60 * 60 * 1000 current_start = SINCE_DATE_MS while current_start < now_ms: window_end = min(current_start + WINDOW_MS, now_ms) params = { "category": "linear", "startTime": current_start, "endTime": window_end, "limit": 50, } r = c._request("GET", "/v5/position/closed-pnl", params, signed=True) items = (r.get("result") or {}).get("list", []) or [] for it in items: sym_raw = it.get("symbol", "") close_order_id = str(it.get("orderId") or "").strip() if close_order_id in excluded_order_ids or _matching_fingerprints( it, excluded_fingerprints ): continue classification = _classify_closed_item(it, policy) strat = classification["strategy"] attribution_source = classification["source"] entry_evidence = classification.get("entry_evidence") or {} qty_t = float(it.get("closedSize", 0) or 0) entry_t = float(it.get("avgEntryPrice", 0) or 0) exit_t = float(it.get("avgExitPrice", 0) or 0) net_pnl = float(it.get("closedPnl", 0) or 0) # Fee: Bybit V5 fornisce openFee + closeFee (somma = totalFee) open_fee = float(it.get("openFee", 0) or 0) close_fee = float(it.get("closeFee", 0) or 0) # Se Bybit non restituisce fee separate, prova totalFee total_fee = float(it.get("totalFee", 0) or 0) or (open_fee + close_fee) # Gross PnL = Net PnL + Total Fee (fee detratte dal lordo) gross_pnl = net_pnl + total_fee # PnL % basato su nozionale entry notional_entry = abs(entry_t * qty_t) pnl_pct_gross = (gross_pnl / notional_entry * 100) if notional_entry > 0 else 0 pnl_pct_net = (net_pnl / notional_entry * 100) if notional_entry > 0 else 0 out.append({ "symbol": sym_raw, "side": it.get("side", ""), "qty": qty_t, "entry_price": entry_t, "exit_price": exit_t, "pnl": net_pnl, "pnl_net": net_pnl, "pnl_gross": gross_pnl, "fee": total_fee, "pnl_pct_gross": pnl_pct_gross, "pnl_pct_net": pnl_pct_net, "strategy": strat, "classification_status": ( "classified" if strat != UNCLASSIFIED_STRATEGY else "unclassified" ), "classification_source": attribution_source, "performance_excluded": classification["performance_excluded"], "operational_reason": classification["operational_reason"], "entry_order_id": entry_evidence.get("entry_order_id", ""), "entry_time": entry_evidence.get("entry_time_raw", ""), "order_id": close_order_id, "created_at": it.get("createdAt", ""), "updated_at": it.get("updatedTime", ""), }) current_start = window_end + 1 return out except Exception as e: log(f"bybit closed trades err: {e}") return out def get_open_positions_bybit_no_test(): """Legge posizioni aperte da Bybit, ESCLUDE MIEI test Pine-style.""" raw = get_bybit_positions() return [p for p in raw if not is_mio_test_position(p)] def get_orders(limit=2000, since_date="2026-08-01"): """Carica ordini dal DB, filtrati per data >= since_date (default 1° agosto 2026).""" out = [] if not DB_PATH.exists(): return out try: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute( "SELECT order_id, symbol, side, qty, price, notional, strategy, created_at " "FROM orders WHERE created_at >= ? ORDER BY id DESC LIMIT ?", (since_date, limit) ) for row in cur.fetchall(): out.append(dict(row)) conn.close() except Exception as e: log(f"orders load err: {e}") return out def get_queue_stats(): if not DB_PATH.exists(): return {"pending": 0, "completed": 0, "failed": 0, "total": 0} try: conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute("SELECT status, COUNT(*) FROM queue GROUP BY status") stats = {"pending": 0, "completed": 0, "failed": 0, "total": 0} for status, count in cur.fetchall(): stats[status] = count stats["total"] += count conn.close() return stats except Exception as e: log(f"queue stats err: {e}") return {"pending": 0, "completed": 0, "failed": 0, "total": 0} MY_TEST_ORDER_IDS = { "31a9c2fd-540e-45a5-b44f-fe2f8fdb4495", # VIRTUAL 1500 buy placeholder (mio test) "4cbb1304-8020-42bd-bcad-8e5d0f19caed", # NEAR 599.9 buy (mio test) "5c53a9ef-dcab-469e-bbd1-3057fe2746d9", # BTC 0.023 buy (mio test) "8a53e7b1-c7d4-4e5a-b67b-4e4fbf4c0977", # VIRTUAL 2496 buy (mio test) "2d2e625b-d3b2-4d87-931a-9b378cf8db93", # VIRTUAL 2679 buy (mio test) "514f5553-6aa3-4148-8e1a-82961c47310b", # VIRTUAL 2712 buy (mio test) "b426dc94-b72f-427b-b2f2-7e021f5da985", # VIRTUAL 2496 buy (mio test) "4c8a4478-377e-4b8c-b371-44c64090e26d", # VIRTUAL buy qty=0 (mio test) "f79ab621-dc17-488e-950e-e984fc0d9e43", # VIRTUAL sell qty=0 (mio test) "2cd6e92c-b82b-4814-808e-e23579f1a533", # VIRTUAL sell Max exit (chiude mio test) "e45f561b-c4d8-4045-a2fe-581140fadac1", # VIRTUAL sell Max exit (chiude mio test 1500) "c60d098a-1ec5-4101-9e77-966f67f4f393", # VIRTUAL sell Pine reale (Max exit di VIRTUAL 2632 Pine reale) "2d2e625b-d3b2-4d87-931a-9b378cf8db93", "79969acc-3e4b-4101-91bc-69eede1ef6aa", # ZEC rettangolo (non mio) "44e4a3ad-23ec-4b7f-b6f8-35778858e5bf", # BTC rettangolo (non mio) "7ed5cc82-3a99-4978-8ae3-c295e82f4609", # BTC rettangolo (non mio) } def is_mio_test(o): """True se l'ordine e' un MIO test Pine-style (placeholder price=1.0, NEAR 599.9, o order_id in MY_TEST_ORDER_IDS).""" oid = str(o.get("order_id", "")) if oid in MY_TEST_ORDER_IDS: return True sym = o.get("symbol", "") try: qty = float(o.get("qty", 0) or 0) price = float(o.get("price", 0) or 0) except (TypeError, ValueError): return False # VIRTUAL Pine placeholder (price=1.0, qty variabile) if sym == "VIRTUALUSDT" and abs(price - 1.0) < 1e-9: return True # NEAR mio test (qty=600 placeholder Pine 2.5, oppure 599.9) if sym == "NEARUSDT" and qty >= 500: return True # BTC mio test (qty=0.023) if sym == "BTCUSDT" and abs(qty - 0.023) < 1e-6: return True return False def compute_trade_pnls(orders): """Matcha buy+sell FIFO per symbol. Esclude trade che coinvolgono MIEI test Pine-style (placeholder price=1.0 o order_id in MY_TEST_ORDER_IDS).""" by_symbol = defaultdict(list) for o in sorted(orders, key=lambda x: x.get("created_at", "")): sym = o.get("symbol", "") if not sym: continue oid = o.get("order_id", "") if oid in ("no_position", "anti_dup_skip", "", None): continue if o.get("qty", 0) <= 0: continue # Filtra MIEI test if is_mio_test(o): continue by_symbol[sym].append(o) trades = [] for sym, ords in by_symbol.items(): open_buys = [] for o in ords: side = o.get("side", "").lower() qty = float(o.get("qty", 0) or 0) price = float(o.get("price", 0) or 0) strat = o.get("strategy", "") ts = o.get("created_at", "") if side == "buy": placeholder = abs(price - 1.0) < 1e-9 open_buys.append({"qty": qty, "price": price, "strategy": strat, "ts": ts, "placeholder": placeholder}) elif side == "sell" and open_buys: qty_to_close = qty while qty_to_close > 0 and open_buys: buy = open_buys[0] matched = min(qty_to_close, buy["qty"]) pnl = (price - buy["price"]) * matched pnl_pct = ((price - buy["price"]) / buy["price"]) * 100 if buy["price"] > 0 else 0 # Warning se PnL% estremo (>30%) o buy era placeholder Pine warn = buy.get("placeholder", False) or abs(pnl_pct) > 30 trades.append({ "symbol": sym, "side": "long", "qty": matched, "entry_price": buy["price"], "exit_price": price, "pnl": pnl, "pnl_pct": pnl_pct, "strategy": buy["strategy"], "entry_ts": buy["ts"], "exit_ts": ts, "warn": warn, }) qty_to_close -= matched buy["qty"] -= matched if buy["qty"] <= 1e-9: open_buys.pop(0) return trades def compute_kpis_from_bybit(closed_trades, positions, balance): """Calcola KPI da trade chiusi Bybit (gia filtrati per data e test) + posizioni aperte (gia filtrate).""" n = len(closed_trades) if n == 0: return { "n_trades": 0, "wins": 0, "losses": 0, "winrate": 0.0, "total_pnl": 0.0, "total_pnl_net": 0.0, "total_pnl_gross": 0.0, "total_fees": 0.0, "total_notional": 0.0, "total_losses": 0.0, "pnl_realized": 0.0, "avg_pnl": 0.0, "avg_pnl_gross": 0.0, "best_trade": 0.0, "worst_trade": 0.0, "open_positions": len(positions), "unrealized_pnl": 0.0, "balance": balance, "total_pnl_with_unrealized": 0.0, "total_pnl_gross_with_unrealized": 0.0, "trading_days": 0, "profit_factor": 0.0, "first_trade_date": "", "sharpe_ratio": 0.0, "sortino_ratio": 0.0, } wins = [t for t in closed_trades if t.get("pnl_net", t.get("pnl", 0)) > 0] losses = [t for t in closed_trades if t.get("pnl_net", t.get("pnl", 0)) <= 0] total_pnl_net = sum(t.get("pnl_net", t.get("pnl", 0)) for t in closed_trades) total_pnl_gross = sum(t.get("pnl_gross", 0) for t in closed_trades) total_fees = sum(t.get("fee", 0) for t in closed_trades) # FIX 07/08 (Mattia 13:51): nozionale Bybit totale (avgEntryPrice * closedSize) di tutti i trade chiusi total_notional = sum(abs(float(t.get("entry_price", 0) or 0) * float(t.get("qty", 0) or 0)) for t in closed_trades) # FIX 07/08 (Mattia): somma TUTTE le singole loss chiuse (in valore assoluto, positivo). # Serve per la formula: P&L REALIZED = GROSS - FEE - LOSS total_losses = abs(sum(t.get("pnl_net", t.get("pnl", 0)) for t in losses)) unrealized = 0.0 for p in positions: try: unrealized += float(p.get("unrealisedPnl", 0) or 0) except Exception: pass # P&L REALIZED = GROSS - FEE (formula Mattia 07/08 v3 - LOSS escluso dalla formula Net) pnl_realized = total_pnl_gross - total_fees # = chiusi netti (senza togliere LOSS) # === Trading days: dal PRIMO trade chiuso a oggi (UTC) === first_trade_dt = None for t in closed_trades: ts = _parse_bybit_ts(t.get("updated_at") or t.get("created_at")) if ts is not None: if first_trade_dt is None or ts < first_trade_dt: first_trade_dt = ts if first_trade_dt is not None: # Converti now a UTC from datetime import datetime as _dt, timezone as _tz now_utc = _dt.now(_tz.utc) first_trade_date = first_trade_dt.astimezone(_tz.utc).strftime("%Y-%m-%d") delta = now_utc - first_trade_dt # Conta giorni inclusivo (se primo trade oggi, = 1 giorno; se 5gg fa, = 5) trading_days = max(1, delta.days + 1) else: trading_days = 0 first_trade_date = "" # === Profit Factor: somma(win PnL) / |somma(loss PnL)| === # Convenzione: PF > 1.5 buono, > 2.0 ottimo, < 1.0 in perdita. total_win_pnl = sum(t.get("pnl_net", t.get("pnl", 0)) for t in wins) total_loss_pnl = sum(t.get("pnl_net", t.get("pnl", 0)) for t in losses) # <= 0 if total_loss_pnl < 0: profit_factor = total_win_pnl / abs(total_loss_pnl) elif total_win_pnl > 0: # Solo vincite, nessuna perdita: PF "infinito" → mostriamo 99.99 profit_factor = 99.99 else: profit_factor = 0.0 # === Sharpe & Sortino globali RIMOSSI 09/08 (Mattia): meglio per strategia === sharpe_ratio = 0.0 sortino_ratio = 0.0 return { "n_trades": n, "wins": len(wins), "losses": len(losses), "winrate": (len(wins) / n * 100) if n else 0.0, "total_pnl": total_pnl_net, # alias per compatibilita "total_pnl_net": total_pnl_net, # NET (dopo fee) = chiusi netti "total_pnl_gross": total_pnl_gross, # GROSS (lordo chiusi, prima delle fee) "total_fees": total_fees, # TOTALE FEE "total_notional": total_notional, # FIX 07/08 13:51 Mattia: nozionale Bybit totale "total_losses": total_losses, # FIX 07/08 Mattia: somma singole loss chiuse (valore assoluto) "pnl_realized": pnl_realized, # FIX 07/08 Mattia: P&L REALIZED = GROSS - FEE - LOSS "avg_pnl": total_pnl_net / n if n else 0.0, "avg_pnl_gross": total_pnl_gross / n if n else 0.0, "best_trade": max(t.get("pnl_net", t.get("pnl", 0)) for t in closed_trades) if closed_trades else 0.0, "worst_trade": min(t.get("pnl_net", t.get("pnl", 0)) for t in closed_trades) if closed_trades else 0.0, "open_positions": len(positions), "unrealized_pnl": unrealized, "balance": balance, "total_pnl_with_unrealized": total_pnl_net + unrealized, "total_pnl_gross_with_unrealized": total_pnl_gross + unrealized, "trading_days": trading_days, "first_trade_date": first_trade_date, "profit_factor": profit_factor, "sharpe_ratio": sharpe_ratio, "sortino_ratio": sortino_ratio, } def compute_equity_curve_from_bybit(closed_trades, balance): """Calcola equity curve da trade chiusi Bybit (gia filtrati). Ritorna anche max_dd_usdt e max_dd_pct (FIX 07/08 Mattia: drawdown real come Dash 5511). FIX 12/08 (Mattia): aggiunto cum_pnl (PnL cumulativo progressivo) per grafico equity. """ if not closed_trades: return {"dates": [], "equity": [], "cum_pnl": [], "trades_count": 0, "max_dd_usdt": 0.0, "max_dd_pct": 0.0} closed_trades.sort(key=lambda t: t.get("updated_at") or t.get("created_at") or "") initial_equity = max(balance - sum(t["pnl"] for t in closed_trades), 0) dates, equity, cum_pnl, cum = [], [], [], 0 for t in closed_trades: cum += t["pnl"] ts = t.get("updated_at") or t.get("created_at") or "" dates.append(ts) equity.append(initial_equity + cum) cum_pnl.append(cum) # FIX 07/08 (Mattia): max drawdown real (come Dash 5511) peak = equity[0] if equity else 0 max_dd_usdt = 0.0 max_dd_pct = 0.0 for v in equity: if v > peak: peak = v dd = peak - v if dd > max_dd_usdt: max_dd_usdt = dd max_dd_pct = (dd / peak * 100) if peak > 0 else 0 return { "dates": dates, "equity": equity, "cum_pnl": cum_pnl, "trades_count": len(closed_trades), "max_dd_usdt": max_dd_usdt, "max_dd_pct": max_dd_pct, } def compute_by_strategy_from_bybit(closed_trades): by_strat = defaultdict(lambda: {"n": 0, "wins": 0, "pnl": 0.0, "pnl_gross": 0.0, "fees": 0.0, "pnls": []}) for t in closed_trades: s = t.get("strategy", UNCLASSIFIED_STRATEGY) if s == UNCLASSIFIED_STRATEGY: continue pnl_n = t.get("pnl_net", t.get("pnl", 0)) by_strat[s]["n"] += 1 if pnl_n > 0: by_strat[s]["wins"] += 1 by_strat[s]["pnl"] += pnl_n by_strat[s]["pnl_gross"] += t.get("pnl_gross", 0) by_strat[s]["fees"] += t.get("fee", 0) by_strat[s]["pnls"].append(pnl_n) out = [] for s, d in by_strat.items(): wr = (d["wins"] / d["n"] * 100) if d["n"] else 0 # Sharpe/Sortino per strategia (09/08 Mattia): per-trade, rf=0 n = d["n"] pnls = d["pnls"] if n >= 2: mean = sum(pnls) / n var = sum((x - mean) ** 2 for x in pnls) / (n - 1) std = math.sqrt(var) if var > 0 else 0.0 sharpe = mean / std if std > 0 else 0.0 ddsq = sum(min(0.0, x) ** 2 for x in pnls) dd = math.sqrt(ddsq / n) if ddsq > 0 else 0.0 sortino = mean / dd if dd > 0 else 0.0 else: sharpe = sortino = 0.0 out.append({"strategy": s, "n": n, "wins": d["wins"], "winrate": wr, "pnl": d["pnl"], "pnl_gross": d["pnl_gross"], "fees": d["fees"], "sharpe": sharpe, "sortino": sortino}) out.sort(key=lambda x: x["pnl"], reverse=True) return out def compute_by_asset_from_bybit(closed_trades): by_asset = defaultdict(lambda: {"n": 0, "wins": 0, "pnl": 0.0, "pnl_gross": 0.0, "fees": 0.0, "volume": 0.0}) for t in closed_trades: a = t.get("symbol", "") if not a: continue pnl_n = t.get("pnl_net", t.get("pnl", 0)) by_asset[a]["n"] += 1 if pnl_n > 0: by_asset[a]["wins"] += 1 by_asset[a]["pnl"] += pnl_n by_asset[a]["pnl_gross"] += t.get("pnl_gross", 0) by_asset[a]["fees"] += t.get("fee", 0) by_asset[a]["volume"] += t.get("qty", 0) * t.get("entry_price", 0) out = [] for a, d in by_asset.items(): wr = (d["wins"] / d["n"] * 100) if d["n"] else 0 out.append({"symbol": a, "n": d["n"], "wins": d["wins"], "winrate": wr, "pnl": d["pnl"], "pnl_gross": d["pnl_gross"], "fees": d["fees"], "volume": d["volume"]}) out.sort(key=lambda x: x["pnl"], reverse=True) return out def compute_asset_trade_details(closed_trades): """Movimenti certificati che compongono il saldo di ogni asset.""" details = defaultdict(list) for trade in closed_trades: symbol = str(trade.get("symbol") or "").upper() if not symbol: continue close_side = str(trade.get("side") or "") direction = "LONG" if close_side.lower() == "sell" else "SHORT" if close_side.lower() == "buy" else "-" opened = _parse_bybit_ts(trade.get("entry_time")) closed = _parse_bybit_ts(trade.get("updated_at") or trade.get("created_at")) details[symbol].append({ "opened_at": opened.astimezone().strftime("%Y-%m-%d %H:%M") if opened else "-", "closed_at": closed.astimezone().strftime("%Y-%m-%d %H:%M") if closed else "-", "strategy": str(trade.get("strategy") or "").upper(), "direction": direction, "qty": _safe_float(trade.get("qty")), "entry": _safe_float(trade.get("entry_price")), "exit": _safe_float(trade.get("exit_price")), "gross": _safe_float(trade.get("pnl_gross")), "fee": _safe_float(trade.get("fee")), "net": _safe_float(trade.get("pnl_net", trade.get("pnl", 0))), "order_id": str(trade.get("order_id") or ""), }) for rows in details.values(): rows.sort(key=lambda row: row["closed_at"], reverse=True) return dict(details) def compute_strategy_trade_details(closed_trades): """Movimenti certificati che compongono il saldo di ogni strategia.""" details = defaultdict(list) for trade in closed_trades: strategy = str(trade.get("strategy") or UNCLASSIFIED_STRATEGY) if strategy == UNCLASSIFIED_STRATEGY: continue close_side = str(trade.get("side") or "") direction = "LONG" if close_side.lower() == "sell" else "SHORT" if close_side.lower() == "buy" else "-" opened = _parse_bybit_ts(trade.get("entry_time")) closed = _parse_bybit_ts(trade.get("updated_at") or trade.get("created_at")) details[strategy].append({ "opened_at": opened.astimezone().strftime("%Y-%m-%d %H:%M") if opened else "-", "closed_at": closed.astimezone().strftime("%Y-%m-%d %H:%M") if closed else "-", "symbol": str(trade.get("symbol") or "").upper(), "direction": direction, "qty": _safe_float(trade.get("qty")), "entry": _safe_float(trade.get("entry_price")), "exit": _safe_float(trade.get("exit_price")), "gross": _safe_float(trade.get("pnl_gross")), "fee": _safe_float(trade.get("fee")), "net": _safe_float(trade.get("pnl_net", trade.get("pnl", 0))), "order_id": str(trade.get("order_id") or ""), }) for rows in details.values(): rows.sort(key=lambda row: row["closed_at"], reverse=True) return dict(details) def compute_by_asset_strategy_from_bybit(closed_trades): """FIX 11/08 (Mattia): aggrega per (symbol, strategy) per evitare abbinamenti errati tipo AEO RSI swing classificato come VPTR3, oppure WIF RSI swing classificato come RETTANGOLO. Ogni entry: {symbol, strategy, n, wins, winrate, pnl, pnl_gross, fees}.""" by_as = defaultdict(lambda: {"n": 0, "wins": 0, "pnl": 0.0, "pnl_gross": 0.0, "fees": 0.0}) for t in closed_trades: a = t.get("symbol", "") s = t.get("strategy", UNCLASSIFIED_STRATEGY) if not a or s == UNCLASSIFIED_STRATEGY: continue pnl_n = t.get("pnl_net", t.get("pnl", 0)) key = (a, s) by_as[key]["n"] += 1 if pnl_n > 0: by_as[key]["wins"] += 1 by_as[key]["pnl"] += pnl_n by_as[key]["pnl_gross"] += t.get("pnl_gross", 0) by_as[key]["fees"] += t.get("fee", 0) out = [] for (a, s), d in by_as.items(): wr = (d["wins"] / d["n"] * 100) if d["n"] else 0 out.append({"symbol": a, "strategy": s, "n": d["n"], "wins": d["wins"], "winrate": wr, "pnl": d["pnl"], "pnl_gross": d["pnl_gross"], "fees": d["fees"]}) out.sort(key=lambda x: (x["pnl"], x["n"]), reverse=True) return out def compute_heatmap_from_bybit(closed_trades): """Heatmap WIN/LOSS per giorno x ora. Ritorna matrix_w (vittorie) e matrix_l (loss). FIX 07/08: prima la heatmap mostrava solo il conteggio totale senza distinguere W da L. Adesso separa W (verde) e L (rosso) cosi' VIRTUAL che ha 1 loss -25.47 a Ven 02 appare in rosso, non in verde come tutti gli altri. """ matrix_w = [[0] * 24 for _ in range(7)] matrix_l = [[0] * 24 for _ in range(7)] matrix_gross = [[0.0] * 24 for _ in range(7)] matrix_fees = [[0.0] * 24 for _ in range(7)] matrix_net = [[0.0] * 24 for _ in range(7)] days = ["Lun", "Mar", "Mer", "Gio", "Ven", "Sab", "Dom"] for t in closed_trades: dt = _parse_bybit_ts(t.get("updated_at") or t.get("created_at")) if dt is None: continue # Converti in local time per heatmap (Europa) local_dt = dt.astimezone() pnl = _safe_float(t.get("pnl_net", t.get("pnl", 0))) d_idx = local_dt.weekday() hour = local_dt.hour matrix_gross[d_idx][hour] += _safe_float(t.get("pnl_gross", pnl)) matrix_fees[d_idx][hour] += _safe_float(t.get("fee", 0)) matrix_net[d_idx][hour] += pnl if pnl > 0: matrix_w[d_idx][hour] += 1 elif pnl < 0: matrix_l[d_idx][hour] += 1 else: # pnl == 0: conta come win (no loss) matrix_w[d_idx][hour] += 1 return { "days": days, "hours": list(range(24)), "matrix_w": matrix_w, "matrix_l": matrix_l, "matrix_gross": matrix_gross, "matrix_fees": matrix_fees, "matrix_net": matrix_net, } def compute_trades_by_day(closed_trades): """Raggruppa trade chiusi per giorno, calcola Gross/Fee/Net per ogni giorno.""" by_day = defaultdict(lambda: {"n": 0, "wins": 0, "pnl_gross": 0.0, "fees": 0.0, "pnl_net": 0.0, "symbols": []}) for t in closed_trades: dt = _parse_bybit_ts(t.get("updated_at") or t.get("created_at")) if dt is None: continue day_key = dt.astimezone().strftime("%Y-%m-%d") d = by_day[day_key] d["n"] += 1 pnl_n = t.get("pnl_net", t.get("pnl", 0)) if pnl_n > 0: d["wins"] += 1 d["pnl_gross"] += t.get("pnl_gross", 0) d["fees"] += t.get("fee", 0) d["pnl_net"] += pnl_n sym = t.get("symbol", "") if sym and sym not in d["symbols"]: d["symbols"].append(sym) out = [] for day_key in sorted(by_day.keys(), reverse=True): # piu recenti prima d = by_day[day_key] out.append({ "date": day_key, "n": d["n"], "wins": d["wins"], "winrate": (d["wins"] / d["n"] * 100) if d["n"] else 0, "pnl_gross": d["pnl_gross"], "fees": d["fees"], "pnl_net": d["pnl_net"], "symbols": ", ".join(d["symbols"]), }) return out def compute_entries_by_day(orders, since_date="2026-08-01"): """FIX 07/08 v3: conta le APERTURE (entry Buy) per giorno, NON le posizioni ancora aperte. Un ordine Buy = un'apertura, anche se poi chiusa in piu' tranche. Esclude ordini 'no_pos' e 'anti_dup_skip' (qty=0 o price=0) E MIEI test (is_mio_test). FIX 07/08 v4 Mattia 14:40: escludi NEARUSDT 600 (test Pine) e popola con 0 i giorni dal since_date a oggi che non hanno entry (trasparenza). """ from datetime import datetime, timezone, timedelta by_day = defaultdict(list) for o in orders: if is_mio_test(o): continue sym = o.get("symbol", "") side = (o.get("side", "") or "").lower() try: qty = float(o.get("qty", 0) or 0) price = float(o.get("price", 0) or 0) except (TypeError, ValueError): continue if qty <= 0 or price <= 0: continue if side != "buy": continue ts = o.get("created_at", "") if not ts: continue day = ts[:10] # YYYY-MM-DD by_day[day].append({"symbol": sym, "qty": qty, "price": price, "ts": ts}) # 0-padding: inserisci giorni dal since_date a oggi che mancano try: start = datetime.fromisoformat(since_date).date() except Exception: start = datetime(2026, 8, 1).date() today = datetime.now(timezone.utc).date() cur = start while cur <= today: d = cur.isoformat() if d not in by_day: by_day[d] = [] # giorno vuoto cur += timedelta(days=1) out = [] for day in sorted(by_day.keys(), reverse=True): entries = by_day[day] symbols_str = ", ".join([f"{e['symbol']} ({e['qty']:,.0f}@{e['price']:.4f})" for e in entries]) if entries else "—" out.append({ "date": day, "n": len(entries), "symbols": symbols_str, }) return out def compute_open_positions_by_entry_day(bybit_positions, db_orders, since_date="2026-08-01"): """DEPRECATED dopo v3: contava le posizioni ANCORA APERTE per giorno di entry. Usava FIFO su DB orders, ma non vedeva le chiusure automatiche di Bybit TP/SL (es. WIF 10765 chiusa da TP ma DB diceva ancora aperto). Sostituita da compute_entries_by_day (aperture totali per giorno) che è piu' utile per Mattia. Mantenuta per retrocompatibilita' se Mattia la vuole di nuovo. """ buys_by_symbol = defaultdict(list) for o in db_orders: sym = o.get("symbol", "") side = (o.get("side", "") or "").lower() try: qty = float(o.get("qty", 0) or 0) price = float(o.get("price", 0) or 0) except (TypeError, ValueError): continue if qty <= 0 or price <= 0: continue if side == "buy": buys_by_symbol[sym].append({ "qty": qty, "price": price, "ts": o.get("created_at", ""), }) by_day = defaultdict(lambda: {"n": 0, "symbols": []}) for pos in bybit_positions: sym = pos.get("symbol", "") size = _safe_float(pos.get("size")) avg_price = _safe_float(pos.get("avgPrice")) if not sym or size <= 0: continue buys = buys_by_symbol.get(sym, []) if not buys: day = "N/D" else: last_buy = max(buys, key=lambda x: x.get("ts", "")) day = (last_buy.get("ts", "") or "")[:10] if not day: day = "N/D" by_day[day]["n"] += 1 by_day[day]["symbols"].append(f"{sym} ({size:,.0f}@{avg_price:.4f})") out = [] for day in sorted(by_day.keys(), reverse=True): d = by_day[day] out.append({ "date": day, "n": d["n"], "symbols": ", ".join(d["symbols"]), }) return out def compute_top_bad_from_bybit(closed_trades, top_n=5): """Top N best (asset con pnl > 0, ordinati per pnl desc) + Top N worst (asset che hanno ALMENO 1 singola loss, ordinati per worst_single_loss asc). FIX 07/08: prima il 'bad' richiedeva len(by_asset) > top_n, quindi con <= 5 asset il worst restava vuoto anche se c'erano loser. FIX 07/08 (2): il 'best' filtra solo asset con pnl > 0. FIX 07/08 (3): il 'worst' ora include asset che hanno almeno 1 singola loss, indipendentemente dal pnl aggregato. Cosi' VIRTUAL (pnl aggregato +18.47) compare perche' ha una loss singola da -25.47. Ordinato per worst_single_loss ascendente. """ by_asset = compute_by_asset_from_bybit(closed_trades) # Top: SOLO asset in profitto, ordinati per pnl desc winners = [a for a in by_asset if a.get("pnl", 0) > 0] winners.sort(key=lambda x: x.get("pnl", 0), reverse=True) top = winners[:top_n] # Worst: asset con ALMENO 1 singola loss, ordinati per worst_single_loss ascendente # Calcola worst_single_loss per symbol asset_worst_loss = {} for t in closed_trades: sym = t.get("symbol", "") pnl = t.get("pnl_net", t.get("pnl", 0)) if pnl < 0: if sym not in asset_worst_loss or pnl < asset_worst_loss[sym]: asset_worst_loss[sym] = pnl worst_list = [] for a in by_asset: if a.get("symbol") in asset_worst_loss: a2 = dict(a) a2["worst_single_loss"] = asset_worst_loss[a["symbol"]] worst_list.append(a2) worst_list.sort(key=lambda x: x.get("worst_single_loss", 0)) bad = worst_list[:top_n] return {"top": top, "bad": bad} def compute_top_single_losses(closed_trades, top_n=5): """Top N peggiori trade SINGOLI (per pnl_net ascendente). FIX 07/08: serve perche' VIRTUAL aggregato e' +18.47 ma ha una singola loss -25.47 che non comparirebbe mai in 'by_asset' worst. Mostra le singole chiusure peggiori. """ losers = [t for t in closed_trades if t.get("pnl_net", t.get("pnl", 0)) < 0] losers.sort(key=lambda x: x.get("pnl_net", x.get("pnl", 0))) # ascendente = peggiore prima out = [] for t in losers[:top_n]: dt = _parse_bybit_ts(t.get("updated_at") or t.get("created_at")) out.append({ "symbol": t.get("symbol", ""), "side": t.get("side", ""), "qty": t.get("qty", 0), "entry_price": t.get("entry_price", 0), "exit_price": t.get("exit_price", 0), "pnl_net": t.get("pnl_net", t.get("pnl", 0)), "pnl_gross": t.get("pnl_gross", 0), "fee": t.get("fee", 0), "date_fmt": dt.strftime("%Y-%m-%d %H:%M") if dt else "", }) return out def get_strategy_for_symbol(symbol): """Replica la logica di sltp_engine.get_strategy_for_symbol per le 4 strategie v2. FIX 11/08 (Mattia): scorre TUTTI i CSV e ritorna lo strategy con priorita' piu' alta. Priorita': rsi_swing_breakout > rettangolo_simple > rettangolo > ma_trailing > vptr3. Necessario perche' WIFUSDT e' sia in rettangolo_assets.csv (strategy=rettangolo) che in RETTANGOLO_SIMPLE_ASSETS.csv (strategy=rsi_swing_breakout), e la versione specifica RSI swing deve avere priorita' sulla generica rettangolo.""" sym = symbol.upper() vptr3_csv = ROOT / "VPTR_V3_ASSETS.csv" rett_csv = ROOT / "rettangolo_assets.csv" rett_simple_csv = ROOT / "RETTANGOLO_SIMPLE_ASSETS.csv" ma_csv = ROOT / "MA_TRAILING_ASSETS.csv" import csv as csvmod # Priorita' strategie: piu' alto = piu' specifico PRIORITY = { "rsi_swing_breakout": 50, "rettangolo_simple": 40, "rettangolo": 30, "ma_trailing": 20, "vptr3": 10, } candidates = [] # lista di (priority, strategy) for csv_path in (vptr3_csv, rett_csv, rett_simple_csv, ma_csv): if not csv_path.exists(): continue try: with open(csv_path, "r", encoding="utf-8") as f: reader = csvmod.DictReader(f) for row in reader: if not row.get("symbol"): continue if row["symbol"].strip().upper() == sym: if str(row.get("enabled", "true")).lower() == "true": s = row.get("strategy", "").strip().lower() if s == "vptr_v3": s = "vptr3" prio = PRIORITY.get(s, 0) candidates.append((prio, s)) except Exception: pass if not candidates: return "rettangolo" # default legacy # Prendi il match con priorita' piu' alta candidates.sort(key=lambda x: x[0], reverse=True) return candidates[0][1] def get_position_strategy_override(): """FIX 11/08 (Mattia): per le POSIZIONI LIVE, legge un mapping esplicito symbol -> strategy per gestire il mismatch tra CSV e SOURCE Pine. Caso tipico: WIF ha chart Pine TV = RSI swing MA le posizioni live sono state aperte da Python (SL/TP engine) con strategy=rettangolo. Il CSV dice rsi_swing ma la SOURCE reale e' Python/rettangolo. Stessa cosa per AERO (VPTR3 senza SL/TP). File: live_deploy_v2/position_strategy_override.json Formato: {"WIFUSDT": "rettangolo", "AEROUSDT": "vptr3"} Se il symbol non e' nel file, fallback a get_strategy_for_symbol (CSV con priorita').""" import json path = ROOT / "position_strategy_override.json" if not path.exists(): return {} try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) return {k.upper(): v.lower() for k, v in data.items() if isinstance(v, str)} except Exception: return {} def _fmt_usdt(v, decimals=2): try: return f"{float(v):,.{decimals}f}" except Exception: return "0.00" def _fmt_pct(v, decimals=1): try: return f"{float(v):.{decimals}f}" except Exception: return "0.0" def _safe_float(v, default=0.0): try: return float(v) if v is not None else default except Exception: return default def _percentile(sorted_values, percentile): """Percentile lineare deterministico, senza dipendenze esterne.""" if not sorted_values: return 0.0 if len(sorted_values) == 1: return float(sorted_values[0]) position = (len(sorted_values) - 1) * float(percentile) lower = int(math.floor(position)) upper = int(math.ceil(position)) if lower == upper: return float(sorted_values[lower]) weight = position - lower return float(sorted_values[lower]) * (1.0 - weight) + float(sorted_values[upper]) * weight def _fmt_duration(seconds): try: total = max(0, int(round(float(seconds)))) except (TypeError, ValueError): return "—" days, remainder = divmod(total, 86400) hours, remainder = divmod(remainder, 3600) minutes, secs = divmod(remainder, 60) if days: return f"{days}g {hours}h {minutes}m" if hours: return f"{hours}h {minutes}m" if minutes: return f"{minutes}m {secs}s" return f"{secs}s" def compute_trade_duration_stats(closed_trades): """Durata apertura->chiusura per tutti i trade classificati inclusi nei KPI.""" samples = [] by_strategy = defaultdict(list) by_strategy_asset = defaultdict(lambda: {"all": [], "winners": [], "losers": []}) winners = [] losers = [] missing = 0 for trade in closed_trades: opened = _parse_bybit_ts(trade.get("entry_time")) closed = _parse_bybit_ts(trade.get("updated_at") or trade.get("created_at")) if opened is None or closed is None or closed < opened: missing += 1 continue duration = (closed - opened).total_seconds() samples.append(duration) strategy = str(trade.get("strategy") or "unknown") symbol = str(trade.get("symbol") or "unknown").upper() by_strategy[strategy].append(duration) asset_bucket = by_strategy_asset[(strategy, symbol)] asset_bucket["all"].append(duration) pnl = _safe_float(trade.get("pnl_net", trade.get("pnl", 0))) if pnl > 0: winners.append(duration) asset_bucket["winners"].append(duration) elif pnl < 0: losers.append(duration) asset_bucket["losers"].append(duration) ordered = sorted(samples) def summary(values): values = sorted(values) if not values: return {"count": 0, "avg_seconds": 0.0, "median_seconds": 0.0} return { "count": len(values), "avg_seconds": sum(values) / len(values), "median_seconds": _percentile(values, 0.50), } strategies = [] for strategy, values in sorted(by_strategy.items()): item = summary(values) item["strategy"] = strategy strategies.append(item) strategy_assets = [] for (strategy, symbol), buckets in sorted(by_strategy_asset.items()): values = sorted(buckets["all"]) win_values = buckets["winners"] loss_values = buckets["losers"] strategy_assets.append({ "strategy": strategy, "symbol": symbol, "count": len(values), "avg_seconds": sum(values) / len(values), "median_seconds": _percentile(values, 0.50), "min_seconds": values[0], "max_seconds": values[-1], "p25_seconds": _percentile(values, 0.25), "p75_seconds": _percentile(values, 0.75), "p90_seconds": _percentile(values, 0.90), "winner_avg_seconds": sum(win_values) / len(win_values) if win_values else None, "loser_avg_seconds": sum(loss_values) / len(loss_values) if loss_values else None, }) return { "total_count": len(closed_trades), "count": len(ordered), "missing": missing, "avg_seconds": sum(ordered) / len(ordered) if ordered else 0.0, "median_seconds": _percentile(ordered, 0.50), "min_seconds": ordered[0] if ordered else 0.0, "max_seconds": ordered[-1] if ordered else 0.0, "p25_seconds": _percentile(ordered, 0.25), "p75_seconds": _percentile(ordered, 0.75), "p90_seconds": _percentile(ordered, 0.90), "winners": summary(winners), "losers": summary(losers), "by_strategy": strategies, "by_strategy_asset": strategy_assets, } # === HTML RENDER (server-side, niente template engine) === def render_dashboard(data): kpis = data["kpis"] equity = data["equity"] by_strategy = data["by_strategy"] by_asset = data["by_asset"] asset_trade_details = data.get("asset_trade_details", {}) strategy_trade_details = data.get("strategy_trade_details", {}) duration_stats = data.get("trade_duration_stats", {}) top = data["top_bad"]["top"] bad = data["top_bad"]["bad"] heatmap = data["heatmap"] positions = data["positions"] recent = data["recent_orders"] generated_at = data["generated_at"] unclassified = data.get("unclassified_trades", []) operational_exclusions = data.get("operational_exclusions", []) certification_count = len(_certification_candidates(data)) anomaly_count = len(_review_candidates(data)) if unclassified: unclassified_rows = [] for trade in unclassified: pnl = trade.get("pnl_net", trade.get("pnl", 0)) pnl_class = "good" if pnl > 0 else "bad" if pnl < 0 else "" unclassified_rows.append( f"{trade.get('updated_at', '')}" f"{trade.get('symbol', '')}" f"{trade.get('order_id', '')}" f"{trade.get('classification_source', '')}" f"{_fmt_usdt(pnl)} USDT" ) classification_alert_html = ( "
CLASSIFICAZIONE BLOCCATA: " f"{len(unclassified)} trade senza prova univoca. Sono inclusi nel PnL totale, " "ma esclusi dalle statistiche per strategia." "" "" + "".join(unclassified_rows) + "
ChiusuraAssetOrder IDMotivoNet
" ) else: classification_alert_html = ( "
CLASSIFICAZIONE CERTIFICATA: " "tutti i trade inclusi hanno una strategia provata.
" ) if operational_exclusions: exclusion_rows = [] for trade in operational_exclusions: pnl = trade.get("pnl_net", trade.get("pnl", 0)) exclusion_rows.append( f"{trade.get('entry_time', '')}" f"{trade.get('updated_at', '')}" f"{trade.get('symbol', '')}" f"{trade.get('entry_order_id', '')}" f"{trade.get('order_id', '')}" f"{trade.get('operational_reason', '')}" f"{_fmt_usdt(pnl)} USDT" ) operational_exclusions_html = ( "
TRADE OPERATIVI ESCLUSI: " f"{len(operational_exclusions)} ingresso successivo al blocco asset. " "Visibile per audit, escluso da KPI, equity e classifiche." "" "" + "".join(exclusion_rows) + "
EntrataChiusuraAssetEntry order IDClose order IDMotivoNet
" ) else: operational_exclusions_html = "" # I dettagli operativi sono gestiti nella mini-dashboard dedicata. classification_alert_html = "" operational_exclusions_html = "" # KPI pnl_class = "good" if kpis["total_pnl_net"] >= 0 else "bad" pgross_class = "good" if kpis["total_pnl_gross"] >= 0 else "bad" fees_class = "warn" if kpis["total_fees"] > 0 else "" upnl_class = "good" if kpis["unrealized_pnl"] >= 0 else "bad" tot_class = "good" if kpis["total_pnl_with_unrealized"] >= 0 else "bad" wr_class = "good" if kpis["winrate"] >= 50 else ("bad" if kpis["winrate"] < 40 else "") avg_class = "good" if kpis["avg_pnl"] >= 0 else "bad" # Profit Factor: >2 ottimo, >1.5 buono, <1 in perdita pf = kpis.get("profit_factor", 0.0) if pf >= 2.0: pf_class = "good" elif pf >= 1.5: pf_class = "good" elif pf >= 1.0: pf_class = "" else: pf_class = "bad" pf_display = f"{pf:.2f}" if pf < 99 else "∞" first_date = kpis.get("first_trade_date", "") trading_days = kpis.get("trading_days", 0) days_label = f"Trading days (dal {first_date})" if first_date else "Trading days" # === Sharpe & Sortino GLOBALI RIMOSSI 09/08 (Mattia): ora per strategia === sharpe_v = 0.0 sortino_v = 0.0 sharpe_class = "" sortino_class = "" sharpe_display = "" sortino_display = "" # FIX 07/08 (Mattia): composizione del Net P&L # GROSS = lordo chiusi - FEE - LOSS (somma singole loss) = P&L REALIZED # P&L REALIZED + UNREALIZED = NET P&L (= PnL Totale) gross_base = kpis["total_pnl_gross"] # GROSS base (lordo chiusi) - non usato per display fees = kpis["total_fees"] fees_class = "warn" losses = kpis["total_losses"] # LOSS = somma singole loss (valore assoluto) losses_class = "bad" pnl_realized = kpis["pnl_realized"] # P&L REALIZED = GROSS - FEE - LOSS pnl_realized_class = "good" if pnl_realized >= 0 else "bad" # FIX 07/08 (Mattia v4): GROSS mostrato = P&L Realized + LOSS + FEE (formula inversa) gross = pnl_realized + losses + fees gross_class = "good" if gross >= 0 else "bad" # FIX 07/08 (Mattia): Win to Loss ratio (W/L) = wins / losses n_wins = kpis.get("wins", 0) n_losses_count = kpis.get("losses", 0) if n_losses_count > 0: wtl = n_wins / n_losses_count wtl_display = f"{wtl:.2f}" if wtl >= 2.0: wtl_class = "good" elif wtl >= 1.0: wtl_class = "" else: wtl_class = "bad" else: wtl_display = "∞" # nessuna loss wtl_class = "good" unrealized = kpis["unrealized_pnl"] unrealized_class = "good" if unrealized >= 0 else "bad" # NET P&L = P&L REALIZED + UNREALIZED net_pnl = pnl_realized + unrealized net_class = "good" if net_pnl >= 0 else "bad" # === FIX 07/08 (Mattia 13:51): 4 box DI FIANCO a "Posizioni aperte" === trading_days = kpis.get("trading_days", 0) or 0 td = max(trading_days, 1) # Box 1: Trade al giorno trades_per_day = round(kpis.get("n_trades", 0) / td, 2) if kpis.get("n_trades", 0) else 0 # Box 2: Nozionale Bybit medio/giorno notional_per_day = kpis.get("total_notional", 0) / td # Box 4: Day Profit in valore assoluto (Net P&L / N° giorni) day_profit_abs = net_pnl / td day_profit_class = "good" if day_profit_abs >= 0 else "bad" # Box 3: rendimento medio giornaliero sul capitale iniziale fisso. DAY_PROFIT_CAPITALE = 4000.0 day_profit_pct_4000 = (day_profit_abs / DAY_PROFIT_CAPITALE) * 100 day_profit_pct_class = "good" if day_profit_pct_4000 >= 0 else "bad" roi_pct_4000 = (net_pnl / DAY_PROFIT_CAPITALE) * 100 roi_pct_class = "good" if roi_pct_4000 >= 0 else "bad" kpi_html = f"""
Saldo capitale (4.000 + Net P&L)
{_fmt_usdt(DAY_PROFIT_CAPITALE + net_pnl)} USDT
ROI su capitale iniziale (4.000 USDT)
{roi_pct_4000:.2f}%
Trade chiusi
{kpis['n_trades']}
{days_label}
{trading_days}
Win rate
{_fmt_pct(kpis['winrate'])}%
{kpis['wins']}W / {kpis['losses']}L
Win to Loss
{wtl_display}
Profit Factor
{pf_display}
Gross PnL (chiusi)
{_fmt_usdt(gross)} USDT
Fee totali
-{_fmt_usdt(fees)} USDT
Loss chiuse
{_fmt_usdt(losses)} USDT
P&L Realized (Gross-Loss-Fee)
{_fmt_usdt(pnl_realized)} USDT
Unrealized PnL
{_fmt_usdt(unrealized)} USDT
Net P&L (Realized-Unrealized)
{_fmt_usdt(net_pnl)} USDT
Avg trade
{_fmt_usdt(kpis['avg_pnl'])} USDT
Best trade
{_fmt_usdt(kpis['best_trade'])} USDT
Posizioni aperte
{kpis['open_positions']}
Trade al giorno
{trades_per_day}
Nozionale Bybit/giorno
{_fmt_usdt(notional_per_day)} USDT
Day Profit %
{day_profit_pct_4000:.2f}%/gg
Day Profit
{_fmt_usdt(day_profit_abs)} USDT/gg
Drawdown real
{_fmt_usdt(equity["max_dd_usdt"])} USDT
{equity["max_dd_pct"]:.2f}%
""" duration_strategy_rows = [] duration_asset_dialogs = [] duration_esc = lambda value: __import__("html").escape(str(value if value is not None else ""), quote=True) for row in duration_stats.get("by_strategy", []): strategy = str(row["strategy"]) strategy_safe = "".join(ch for ch in strategy if ch.isalnum()) dialog_id = f"duration-asset-detail-{strategy_safe}" duration_strategy_rows.append( f"" f"{row['count']}{_fmt_duration(row['avg_seconds'])}" f"{_fmt_duration(row['median_seconds'])}" ) asset_rows = [] assets = [ item for item in duration_stats.get("by_strategy_asset", []) if str(item.get("strategy") or "") == strategy ] assets.sort(key=lambda item: item.get("avg_seconds", 0), reverse=True) for asset in assets: winner_avg = asset.get("winner_avg_seconds") loser_avg = asset.get("loser_avg_seconds") asset_rows.append( f"{duration_esc(asset['symbol'])}{asset['count']}" f"{_fmt_duration(asset['avg_seconds'])}{_fmt_duration(asset['median_seconds'])}" f"{_fmt_duration(asset['min_seconds'])}{_fmt_duration(asset['max_seconds'])}" f"{_fmt_duration(asset['p25_seconds'])}{_fmt_duration(asset['p75_seconds'])}" f"{_fmt_duration(asset['p90_seconds'])}" f"{_fmt_duration(winner_avg) if winner_avg is not None else '—'}" f"{_fmt_duration(loser_avg) if loser_avg is not None else '—'}" ) duration_asset_dialogs.append( f"" f"
DURATA PER ASSET

{duration_esc(strategy.upper())}

" f"
" f"
{row['count']} trade" f"Media {_fmt_duration(row['avg_seconds'])}" f"Mediana {_fmt_duration(row['median_seconds'])}
" "
" "" "" + "".join(asset_rows) + "
AssetTradeMediaMedianaMinMaxP25P75P90Media vincentiMedia perdenti
" ) duration_html = f"""
Trade totali{duration_stats.get('total_count', duration_stats.get('count', 0) + duration_stats.get('missing', 0))}Durata disponibile: {duration_stats.get('count', 0)} · Mancanti: {duration_stats.get('missing', 0)}
Durata media{_fmt_duration(duration_stats.get('avg_seconds', 0))}
Mediana{_fmt_duration(duration_stats.get('median_seconds', 0))}
Min / Max{_fmt_duration(duration_stats.get('min_seconds', 0))}{_fmt_duration(duration_stats.get('max_seconds', 0))}
P25 / P75{_fmt_duration(duration_stats.get('p25_seconds', 0))}{_fmt_duration(duration_stats.get('p75_seconds', 0))}
P90{_fmt_duration(duration_stats.get('p90_seconds', 0))}
Vincenti · media{_fmt_duration(duration_stats.get('winners', {}).get('avg_seconds', 0))}{duration_stats.get('winners', {}).get('count', 0)} trade
Perdenti · media{_fmt_duration(duration_stats.get('losers', {}).get('avg_seconds', 0))}{duration_stats.get('losers', {}).get('count', 0)} trade
{"".join(duration_strategy_rows)}
StrategiaTradeMediaMediana
Come leggere P25, P75 e P90

P25, P75 e P90 sono percentili della durata:

Esempio: P25 = 30 minuti, P75 = 4 ore, P90 = 10 ore.

Significa che:

La fascia P25–P75 rappresenta la durata del 50% centrale dei trade ed è meno influenzata da operazioni eccezionalmente lunghe rispetto alla media.

Il P90 è utile per stabilire quando un trade sta diventando insolitamente lungo. Non significa però che debba essere chiuso automaticamente: indica solamente che ha già superato la durata del 90% dei trade storici.

Attenzione ai campioni piccoli: se un asset ha uno o due trade, i percentili sono poco affidabili. Con un solo trade, P25, mediana, P75 e P90 coincidono tutti con la durata di quell’unica operazione.

{"".join(duration_asset_dialogs)} """ # Equity Curve reale: balance iniziale ricostruito + Net PnL progressivo. if equity["trades_count"] > 0: # SVG inline (no plotly): polyline semplice eq_pts = equity["equity"] eq_dates = equity["dates"] cum_pnl_pts = eq_pts if len(cum_pnl_pts) >= 2: # Range Y adattato all'equity per rendere leggibili variazioni e drawdown. min_v, max_v = min(cum_pnl_pts), max(cum_pnl_pts) rng = max(max_v - min_v, 0.01) # Normalizza in 0-100 per SVG w, h = 800, 200 pad = 10 # Un punto per ogni trade chiuso. points = [] for i, v in enumerate(cum_pnl_pts): x = pad + (i / (len(cum_pnl_pts) - 1)) * (w - 2 * pad) y = h - pad - ((v - min_v) / rng) * (h - 2 * pad) points.append(f"{x:.1f},{y:.1f}") svg_line = " ".join(points) cum_pnl_svg_line = "" # non piu' usato last_v = cum_pnl_pts[-1] first_v = cum_pnl_pts[0] last_change = last_v - first_v change_pct = (last_change / abs(first_v) * 100) if first_v != 0 else 0 color = "#3fb950" if last_change >= 0 else "#f85149" # Genera marker hover per ogni trade (tooltip nativi browser via ) hover_points = "" n_eq = len(cum_pnl_pts) for i_p, (pt_xy, v_p) in enumerate(zip(points, cum_pnl_pts)): cx_p, cy_p = pt_xy.split(",") dt_raw = eq_dates[i_p] if i_p < len(eq_dates) else "" # Converti unix seconds in data leggibile (formato italiano) dt_label = f"#{i_p+1}" if dt_raw: try: ts_val = float(dt_raw) if ts_val > 1e12: # millisecondi ts_val = ts_val / 1000.0 from datetime import datetime as _dt, timezone as _tz dt_obj = _dt.fromtimestamp(ts_val, tz=_tz.utc) dt_label = dt_obj.strftime("%d/%m %H:%M") except Exception: dt_label = dt_raw[:16] if dt_raw else f"#{i_p+1}" delta_v = v_p - cum_pnl_pts[0] delta_pct = (delta_v / abs(cum_pnl_pts[0]) * 100) if cum_pnl_pts[0] != 0 else 0 # Marker visible (raggio 4) + tooltip custom JS al hover (no delay browser) dt_js = "'" + dt_label.replace("\\", "\\\\").replace("'", "\\'") + "'" hover_points += ( f'<circle class="eq-pt" cx="{cx_p}" cy="{cy_p}" r="4" fill="#fff" fill-opacity="0.95" stroke="{color}" stroke-width="1.5" ' f'style="cursor:pointer" ' f'onmouseover="eqShowTip(event, {i_p+1}, {dt_js}, {v_p:.2f}, {delta_v:.2f}, {delta_pct:.2f})" ' f'onmousemove="eqMoveTip(event)" ' f'onmouseout="eqHideTip()">' f'<title>Trade #{i_p+1} | {dt_label} | PnL cum: {v_p:.2f} USDT' f'\n ' ) # Label Y (5 tacche: min, q1, mid, q3, max) y_labels = [] for frac in [0.0, 0.25, 0.5, 0.75, 1.0]: v_label = min_v + frac * rng y_pos = h - pad - frac * (h - 2 * pad) y_labels.append((v_label, y_pos)) y_label_svg = "" grid_h_svg = "" for v_label, y_pos in y_labels: grid_h_svg += f'\n ' y_label_svg += f'{v_label:.2f}\n ' # Label X (data primo, 1/3, 2/3, ultimo) n = len(eq_pts) x_idx = [0, n//3, 2*n//3, n-1] if n >= 4 else list(range(n)) x_label_svg = "" grid_v_svg = "" for idx in x_idx: x = pad + (idx / max(n - 1, 1)) * (w - 2 * pad) dt_raw = eq_dates[idx] if idx < len(eq_dates) else "" # Converti unix seconds/ms in data leggibile dt_label = "" if dt_raw: try: ts_v = float(dt_raw) if ts_v > 1e12: ts_v = ts_v / 1000.0 from datetime import datetime as _dt2, timezone as _tz2 dt_label = _dt2.fromtimestamp(ts_v, tz=_tz2.utc).strftime("%d/%m %H:%M") except Exception: dt_label = str(dt_raw)[:10] grid_v_svg += f'\n ' anchor = "start" if idx == 0 else ("end" if idx == n-1 else "middle") tx = x + 2 if anchor == "start" else (x - 2 if anchor == "end" else x) x_label_svg += f'{dt_label}\n ' # Fill area sotto la curva (gradiente) fill_pts = svg_line + f" {points[-1].split(chr(44))[0]},{h-pad} {pad},{h-pad}" # Linea zero (asse orizzontale a y=0 USDT) come riferimento zero_y = h - pad - ((0 - min_v) / rng) * (h - 2 * pad) if rng > 0 else h - pad equity_html = f'''
{grid_h_svg}{grid_v_svg} {hover_points} {y_label_svg}{x_label_svg}
Start: {cum_pnl_pts[0]:+.2f} USDT End: {last_v:+.2f} USDT ({change_pct:+.2f}%)
{len(cum_pnl_pts)} trade chiusi Equity
''' else: # 1 solo trade v = eq_pts[0] equity_html = f'
1 trade chiuso: equity = {v:.2f} USDT
' equity_json = json.dumps(equity) else: equity_html = '
Nessun trade chiuso dal 1° agosto. Aspetto chiusure Pine (Max exit bars).
' equity_json = json.dumps({"dates": [], "equity": [], "cum_pnl": [], "max_dd_usdt": 0.0, "max_dd_pct": 0.0}) # Per strategia (tabella aggregata) if by_strategy: rows = [] strategy_dialogs = [] strategy_esc = lambda value: __import__("html").escape(str(value if value is not None else ""), quote=True) for s in by_strategy: cls = "good" if s["pnl"] >= 0 else "bad" strategy = str(s["strategy"]) strategy_safe = "".join(ch for ch in strategy if ch.isalnum()) dialog_id = f"strategy-detail-{strategy_safe}" rows.append( f"" f"{s['n']} ({s['wins']}W){_fmt_pct(s['winrate'])}%" f"{_fmt_usdt(s['pnl_gross'])}-{_fmt_usdt(s['fees'])}" f"{_fmt_usdt(s['pnl'])} USDT" ) trade_rows = [] for trade in strategy_trade_details.get(strategy, []): net_cls = "good" if trade["net"] >= 0 else "bad" trade_rows.append( f"{strategy_esc(trade['opened_at'])}{strategy_esc(trade['closed_at'])}" f"{strategy_esc(trade['symbol'])}{strategy_esc(trade['direction'])}" f"{trade['qty']:,.4f}{trade['entry']:,.6f}{trade['exit']:,.6f}" f"{_fmt_usdt(trade['gross'])}-{_fmt_usdt(trade['fee'])}" f"{_fmt_usdt(trade['net'])}{strategy_esc(trade['order_id'])}" ) detail_table = ( "
" "" + "".join(trade_rows) + "
AperturaChiusuraAssetDirezioneQtyEntryExitGrossFeeNetClose Order ID
" ) strategy_dialogs.append( f"
" f"
DETTAGLIO STRATEGIA

{strategy_esc(strategy)}

" f"
" f"
{s['n']} trade{s['wins']}W / {s['n']-s['wins']}L" f"WR {_fmt_pct(s['winrate'])}%Gross {_fmt_usdt(s['pnl_gross'])} USDT" f"Fee -{_fmt_usdt(s['fees'])} USDTNet {_fmt_usdt(s['pnl'])} USDT
" f"{detail_table}" ) total_n = sum(s["n"] for s in by_strategy) total_wins = sum(s["wins"] for s in by_strategy) total_wr = (total_wins / total_n * 100) if total_n else 0.0 total_gross = sum(s["pnl_gross"] for s in by_strategy) total_fees = sum(s["fees"] for s in by_strategy) total_net = sum(s["pnl"] for s in by_strategy) gross_cls = "good" if total_gross >= 0 else "bad" net_cls = "good" if total_net >= 0 else "bad" rows.append( "" f"TOTALE{total_n} ({total_wins}W)" f"{_fmt_pct(total_wr)}%" f"{_fmt_usdt(total_gross)}" f"-{_fmt_usdt(total_fees)}" f"{_fmt_usdt(total_net)} USDT" ) strategy_html = ( "" + "".join(rows) + "
StrategiaTradesWRGrossFeeNet
" + "".join(strategy_dialogs) ) else: strategy_html = '
Nessun trade chiuso ancora. Aspetto chiusure Pine (Max exit bars).
' # Sharpe / Sortino per strategia (09/08 Mattia): box per ogni strategia con entrambi i valori if by_strategy: boxes = [] for s in by_strategy: n = s["n"] sh = s.get("sharpe", 0.0) so = s.get("sortino", 0.0) # Nome strategia "pretty" name_raw = s["strategy"] name_disp = name_raw.replace("_", " ").title() if name_raw != "unknown" else "Unknown" if n >= 2: sh_cls = "good" if sh >= 1.0 else ("bad" if sh < 0 else "") so_cls = "good" if so >= 1.5 else ("bad" if so < 0 else "") sh_disp = f"{sh:.3f}" so_disp = f"{so:.3f}" title = (f"Strategia: {name_raw}\nTrade: {n} ({s['wins']}W/{n - s['wins']}L)\n" f"WR: {s['winrate']:.1f}%\nNet: {s['pnl']:.2f} USDT\n" f"Sharpe: mean/std (rf=0) - >=1 buono, <0 in perdita\n" f"Sortino: mean/downside_dev (rf=0) - >=1.5 buono, <0 in perdita") else: sh_cls = so_cls = "" sh_disp = so_disp = "n/a" title = f"Strategia: {name_raw} ({n} trade: servono almeno 2 trade per calcolare Sharpe/Sortino)" boxes.append( f'
' f'
{name_disp} ({n}t)
' f'
Sh: {sh_disp}
' f'
So: {so_disp}
' f'
' ) strategy_sharpe_html = f'
{"".join(boxes)}
' else: strategy_sharpe_html = '
Nessun trade chiuso ancora. Servono almeno 2 trade per strategia per calcolare Sharpe/Sortino.
' # Per asset if by_asset: rows = [] asset_dialogs = [] asset_esc = lambda value: __import__("html").escape(str(value if value is not None else ""), quote=True) for a in by_asset: cls = "good" if a["pnl"] >= 0 else "bad" symbol = str(a["symbol"]).upper() symbol_safe = "".join(ch for ch in symbol if ch.isalnum()) dialog_id = f"asset-detail-{symbol_safe}" rows.append( f"" f"{a['n']} ({a['wins']}W){_fmt_pct(a['winrate'])}%" f"{_fmt_usdt(a['pnl_gross'])}-{_fmt_usdt(a['fees'])}" f"{_fmt_usdt(a['pnl'])} USDT" ) trade_rows = [] for trade in asset_trade_details.get(symbol, []): net_cls = "good" if trade["net"] >= 0 else "bad" trade_rows.append( f"{asset_esc(trade['opened_at'])}{asset_esc(trade['closed_at'])}" f"{asset_esc(trade['strategy'])}{asset_esc(trade['direction'])}" f"{trade['qty']:,.4f}{trade['entry']:,.6f}{trade['exit']:,.6f}" f"{_fmt_usdt(trade['gross'])}-{_fmt_usdt(trade['fee'])}" f"{_fmt_usdt(trade['net'])}{asset_esc(trade['order_id'])}" ) detail_table = ( "
" "" + "".join(trade_rows) + "
AperturaChiusuraStrategiaDirezioneQtyEntryExitGrossFeeNetClose Order ID
" ) asset_dialogs.append( f"
" f"
DETTAGLIO ASSET

{asset_esc(symbol)}

" f"
" f"
{a['n']} trade{a['wins']}W / {a['n']-a['wins']}L" f"WR {_fmt_pct(a['winrate'])}%Gross {_fmt_usdt(a['pnl_gross'])} USDT" f"Fee -{_fmt_usdt(a['fees'])} USDTNet {_fmt_usdt(a['pnl'])} USDT
" f"{detail_table}" ) asset_html = ( "" + "".join(rows) + "
SymbolTradesWRGrossFeeNet
" + "".join(asset_dialogs) ) else: asset_html = '
Nessun trade chiuso ancora
' # Top 5 / Bad 5 def _top_table(items, cls_color): """FIX 07/08: colore PnL DINAMICO basato sul segno (verde se >0, rosso se <0). cls_color indica sezione ('best' o 'worst'). Per Worst, se l'item ha 'worst_single_loss', mostra quella al posto del pnl aggregato (perche' un asset puo' avere pnl aggregato positivo ma loss singola grossa). """ if not items: return '
Nessun trade chiuso ancora
' rows = [] for a in items: p = a.get("pnl", 0) # Per Worst: se worst_single_loss presente, mostriamo quella wsl = a.get("worst_single_loss") if cls_color == "bad" and wsl is not None: # Worst: mostra worst single loss (sempre rosso) + pnl aggregato in tooltip wsl_cls = "bad" title = f"title=\"Worst single: {_fmt_usdt(wsl)} USDT | PnL aggregato: {_fmt_usdt(p)} USDT\"" pnl_cell = f'{_fmt_usdt(wsl)} USDT' else: # Best o worst senza worst_single_loss row_cls = "good" if p > 0 else ("bad" if p < 0 else "") pnl_cell = f'{_fmt_usdt(p)} USDT' rows.append(f"{a['symbol']}{a['n']}{_fmt_pct(a['winrate'])}%{pnl_cell}") return "" + "".join(rows) + "
SymbolTradesWRPnL
" top_html = _top_table(top, "good") bad_html = _top_table(bad, "bad") # Top 5 Single Losses (singoli trade peggiori) - FIX 07/08 single_losses = data.get("single_losses", []) if single_losses: rows = [] for t in single_losses: rows.append(f'{t.get("date_fmt", "")}{t.get("symbol", "")}{t.get("side", "")}{t.get("qty", 0):,.2f}{t.get("entry_price", 0):,.5f}{t.get("exit_price", 0):,.5f}{t.get("pnl_net", 0):.2f} USDT-{t.get("fee", 0):.2f}') single_losses_html = "" + "".join(rows) + "
DataSymbolSideQtyEntryExitNet PnLFee
" else: single_losses_html = '
Nessuna singola loss. Vai tranquillo.
' # APERTURE (entry Buy) per Giorno - FIX 07/08 v3 open_by_day = data.get("open_by_day", []) if open_by_day: rows = [] total_n = 0 for d in open_by_day: total_n += d["n"] rows.append(f'{d["date"]}{d["n"]}{d["symbols"]}') # Riga totale rows.append(f'TOTALE{total_n}aperture totali (entry Buy) dal 1° agosto') open_by_day_html = "" + "".join(rows) + "
Data EntryN. ApertureSymbols (qty@price)
" else: open_by_day_html = '
Nessuna apertura registrata.
' # Heatmap (HTML puro, no Plotly CDN) - FIX 07/08: WIN/LOSS separati matrix_w = heatmap.get("matrix_w", [[0]*24 for _ in range(7)]) matrix_l = heatmap.get("matrix_l", [[0]*24 for _ in range(7)]) matrix_gross = heatmap.get("matrix_gross", [[0.0]*24 for _ in range(7)]) matrix_fees = heatmap.get("matrix_fees", [[0.0]*24 for _ in range(7)]) matrix_net = heatmap.get("matrix_net", [[0.0]*24 for _ in range(7)]) matrix_tot = [[matrix_w[d][h] + matrix_l[d][h] for h in range(24)] for d in range(7)] heatmap_stats_html = '
Nessun dato statistico giorno/ora.
' if any(sum(row) > 0 for row in matrix_tot): # max_count basato sul totale per la scala di intensita' max_count = max(max(row) for row in matrix_tot) if matrix_tot else 1 max_count = max(max_count, 1) heatmap_rows = [] # Header (ore) cells = ["Giorno / Ora"] for h in heatmap["hours"]: cells.append(f"{h:02d}:00") heatmap_rows.append("" + "".join(cells) + "") for d_idx, day in enumerate(heatmap["days"]): cells = [f"{day}"] for h in range(24): w = matrix_w[d_idx][h] l = matrix_l[d_idx][h] if w == 0 and l == 0: cells.append('-') else: # Cella divisa: meta' sinistra W (verde), meta' destra L (rosso) # Intensita' in base al totale nella cella intensity_w = w / max_count if max_count > 0 else 0 intensity_l = l / max_count if max_count > 0 else 0 # W: da scuro (#0d1117) a verde (#3fb950) rw = int(13 + (63 - 13) * intensity_w) gw = int(17 + (185 - 17) * intensity_w) bw = int(23 + (80 - 23) * intensity_w) color_w = f"rgb({rw},{gw},{bw})" # L: da scuro (#0d1117) a rosso (#f85149) rl = int(13 + (248 - 13) * intensity_l) gl = int(17 + (81 - 17) * intensity_l) bl = int(23 + (73 - 23) * intensity_l) color_l = f"rgb({rl},{gl},{bl})" cell_title = f"{day} {h:02d}:00 - Win {w}, Loss {l}" # Solo W: cella verde piena con numero W if l == 0: cells.append(f'W {w}') # Solo L: cella rossa piena con numero L elif w == 0: cells.append(f'L {l}') # Miste: gradiente orizzontale W|L else: bg = f"linear-gradient(to right, {color_w} 0%, {color_w} 50%, {color_l} 50%, {color_l} 100%)" cells.append(f'W {w}L {l}') heatmap_rows.append("" + "".join(cells) + "") # Legenda legenda = ('
' 'W = trade positivi' 'L = trade negativi' 'Scorrere orizzontalmente per tutte le 24 ore
') heatmap_html = "
" + "".join(heatmap_rows) + "
" + legenda def _period_row(label, n, wins, losses, gross, fees, net): wr = (wins / n * 100) if n else 0.0 avg = (net / n) if n else 0.0 cls = "good" if net >= 0 else "bad" return (f"{label}{n}{wins}W / {losses}L" f"{wr:.1f}%{_fmt_usdt(gross)} USDT" f"-{_fmt_usdt(fees)} USDT" f"{_fmt_usdt(net)} USDT" f"{_fmt_usdt(avg)} USDT") day_stats = [] for d_idx, day in enumerate(heatmap["days"]): n = sum(matrix_tot[d_idx]) if n: day_stats.append({ "day_index": d_idx, "label": day, "n": n, "wins": sum(matrix_w[d_idx]), "losses": sum(matrix_l[d_idx]), "gross": sum(matrix_gross[d_idx]), "fees": sum(matrix_fees[d_idx]), "net": sum(matrix_net[d_idx]), }) hour_stats = [] for h in range(24): n = sum(matrix_tot[d][h] for d in range(7)) if n: hour_stats.append({ "label": f"{h:02d}:00", "n": n, "wins": sum(matrix_w[d][h] for d in range(7)), "losses": sum(matrix_l[d][h] for d in range(7)), "gross": sum(matrix_gross[d][h] for d in range(7)), "fees": sum(matrix_fees[d][h] for d in range(7)), "net": sum(matrix_net[d][h] for d in range(7)), }) best_day = max(day_stats, key=lambda x: x["net"]) worst_day = min(day_stats, key=lambda x: x["net"]) best_hour = max(hour_stats, key=lambda x: x["net"]) worst_hour = min(hour_stats, key=lambda x: x["net"]) summary = ( '
' f'
GIORNO MIGLIORE{best_day["label"]}{_fmt_usdt(best_day["net"])} USDT
' f'
GIORNO PEGGIORE{worst_day["label"]}{_fmt_usdt(worst_day["net"])} USDT
' f'
ORA MIGLIORE{best_hour["label"]}{_fmt_usdt(best_hour["net"])} USDT
' f'
ORA PEGGIORE{worst_hour["label"]}{_fmt_usdt(worst_hour["net"])} USDT
' '
' ) header = "PeriodoTradeW/LWRGrossFeeNetMedia/trade" weekday_names = ["Lunedì", "Martedì", "Mercoledì", "Giovedì", "Venerdì", "Sabato", "Domenica"] weekday_esc = lambda value: __import__("html").escape(str(value if value is not None else ""), quote=True) weekday_trade_details = defaultdict(list) for strategy, trades in strategy_trade_details.items(): for trade in trades: try: closed_local = datetime.strptime(trade["closed_at"], "%Y-%m-%d %H:%M") except (KeyError, TypeError, ValueError): continue weekday_trade_details[closed_local.weekday()].append({**trade, "strategy": strategy}) for trades in weekday_trade_details.values(): trades.sort(key=lambda trade: trade["closed_at"], reverse=True) def _weekday_period_row(row): d_idx = row["day_index"] label = weekday_esc(weekday_names[d_idx]) wr = (row["wins"] / row["n"] * 100) if row["n"] else 0.0 avg = (row["net"] / row["n"]) if row["n"] else 0.0 cls = "good" if row["net"] >= 0 else "bad" return ( f"" f"{row['n']}{row['wins']}W / {row['losses']}L" f"{wr:.1f}%{_fmt_usdt(row['gross'])} USDT" f"-{_fmt_usdt(row['fees'])} USDT" f"{_fmt_usdt(row['net'])} USDT" f"{_fmt_usdt(avg)} USDT" ) weekday_dialogs = [] for row in day_stats: d_idx = row["day_index"] cls = "good" if row["net"] >= 0 else "bad" trade_rows = [] for trade in weekday_trade_details.get(d_idx, []): net_cls = "good" if trade["net"] >= 0 else "bad" trade_rows.append( f"{weekday_esc(trade['opened_at'])}" f"{weekday_esc(trade['closed_at'])}{weekday_esc(trade['symbol'])}" f"{weekday_esc(str(trade['strategy']).upper())}{weekday_esc(trade['direction'])}" f"{trade['qty']:,.4f}{trade['entry']:,.6f}{trade['exit']:,.6f}" f"{_fmt_usdt(trade['gross'])}-{_fmt_usdt(trade['fee'])}" f"{_fmt_usdt(trade['net'])}" f"{weekday_esc(trade['order_id'])}" ) weekday_dialogs.append( f"" f"
TRADE CHIUSI NEL GIORNO" f"

{weekday_esc(weekday_names[d_idx])}

" f"
" f"
{row['n']} trade" f"{row['wins']}W / {row['losses']}LWR {(row['wins'] / row['n'] * 100):.1f}%" f"Gross {_fmt_usdt(row['gross'])} USDTFee -{_fmt_usdt(row['fees'])} USDT" f"Net {_fmt_usdt(row['net'])} USDT
" "
" "" "" + "".join(trade_rows) + "
AperturaChiusuraAssetStrategiaDirezioneQtyEntryExitGrossFeeNetClose Order ID
" ) day_rows = "".join(_weekday_period_row(row) for row in day_stats) hour_rows = "".join(_period_row(**row) for row in hour_stats) heatmap_stats_html = ( summary + '
' '

Risultati per giorno della settimana

' + header + day_rows + '
' '

Risultati per ora di chiusura

' + header + hour_rows + '
' '
' + "".join(weekday_dialogs) ) else: heatmap_html = '
Nessun dato heatmap. Aspetto chiusure Pine.
' # Trade per Giorno _trades_by_day = data.get("trades_by_day", []) if _trades_by_day: rows = [] for d in _trades_by_day: cls = "good" if d["pnl_net"] >= 0 else "bad" rows.append(f"{d['date']}{d['n']} ({d['wins']}W){d['winrate']:.1f}%{_fmt_usdt(d['pnl_gross'])}-{_fmt_usdt(d['fees'])}{_fmt_usdt(d['pnl_net'])} USDT{d['symbols']}") # Totale tot_gross = sum(d['pnl_gross'] for d in _trades_by_day) tot_fees = sum(d['fees'] for d in _trades_by_day) tot_net = sum(d['pnl_net'] for d in _trades_by_day) tot_n = sum(d['n'] for d in _trades_by_day) tot_wins = sum(d['wins'] for d in _trades_by_day) tot_wr = (tot_wins / tot_n * 100) if tot_n else 0 tot_cls = "good" if tot_net >= 0 else "bad" rows.append(f"TOTALE{tot_n} ({tot_wins}W){tot_wr:.1f}%{_fmt_usdt(tot_gross)}-{_fmt_usdt(tot_fees)}{_fmt_usdt(tot_net)} USDT") trades_by_day_html = "" + "".join(rows) + "
DataTradesWRGrossFeeNetSimboli
" else: trades_by_day_html = '
Nessun trade chiuso dal 1° agosto.
' # FIX 12/08 (Mattia): grafico dual axis Daily PnL (barre) + Cumulativo (linea). # Asse Y sinistro: PnL giornaliero (barre). Asse Y destro: PnL cumulativo (linea). daily_chart_html = '
Nessun dato giornaliero per il grafico.
' if _trades_by_day and len(_trades_by_day) >= 1: days_sorted = sorted(_trades_by_day, key=lambda d: d["date"]) n_days = len(days_sorted) daily_pnl = [d["pnl_net"] for d in days_sorted] cum_pnl = [] running = 0.0 for v in daily_pnl: running += v cum_pnl.append(running) # Range Y sinistro (daily PnL) d_max = max(max(daily_pnl), 0) d_min = min(min(daily_pnl), 0) d_rng = max(d_max - d_min, 0.01) d_top = d_max + d_rng * 0.1 d_bot = d_min - d_rng * 0.1 d_rng_total = d_top - d_bot # Range Y destro (cum PnL) c_max = max(max(cum_pnl), 0) c_min = min(min(cum_pnl), 0) c_rng = max(c_max - c_min, 0.01) c_top = c_max + c_rng * 0.1 c_bot = c_min - c_rng * 0.1 c_rng_total = c_top - c_bot w, h = 900, 260 pad_l, pad_r, pad_t, pad_b = 60, 60, 10, 35 plot_w = w - pad_l - pad_r plot_h = h - pad_t - pad_b x_positions = [] if n_days > 1: for i in range(n_days): x_positions.append(pad_l + (i / (n_days - 1)) * plot_w) else: x_positions.append(pad_l + plot_w / 2) bar_w = max(plot_w / max(n_days, 1) * 0.6, 2) bars_svg = "" for i, (xv, dv) in enumerate(zip(x_positions, daily_pnl)): y_top = pad_t + (1 - (max(dv, 0) - d_bot) / d_rng_total) * plot_h y_bot = pad_t + (1 - (min(dv, 0) - d_bot) / d_rng_total) * plot_h y_zero = pad_t + (1 - (0 - d_bot) / d_rng_total) * plot_h if dv >= 0: y1 = y_zero y2 = y_top color = "#3fb950" else: y1 = y_bot y2 = y_zero color = "#f85149" # Mattia: NON piu' barre ma linee. Disegno come linee con marker. # Per ora lascio il fallback commentato (era per barre). # La linea daily e' generata sotto in daily_line pass cum_pts = [] for i, (xv, cv) in enumerate(zip(x_positions, cum_pnl)): y = pad_t + (1 - (cv - c_bot) / c_rng_total) * plot_h cum_pts.append(f"{xv:.1f},{y:.1f}") cum_line = " ".join(cum_pts) # FIX 12/08 (Mattia): linee invece di barre. Daily PnL = polyline colorata per segno daily_pts = [] for i, (xv, dv) in enumerate(zip(x_positions, daily_pnl)): y = pad_t + (1 - (dv - d_bot) / d_rng_total) * plot_h daily_pts.append(f"{xv:.1f},{y:.1f}") daily_line = " ".join(daily_pts) y_labels_left = [] for frac in [0.0, 0.25, 0.5, 0.75, 1.0]: v_label = d_bot + frac * d_rng_total y_pos = pad_t + (1 - frac) * plot_h y_labels_left.append((v_label, y_pos)) y_left_svg = "" grid_h_svg = "" for v_label, y_pos in y_labels_left: grid_h_svg += f'\n ' y_left_svg += f'{v_label:+.2f}\n ' y_labels_right = [] for frac in [0.0, 0.25, 0.5, 0.75, 1.0]: v_label = c_bot + frac * c_rng_total y_pos = pad_t + (1 - frac) * plot_h y_labels_right.append((v_label, y_pos)) y_right_svg = "" for v_label, y_pos in y_labels_right: y_right_svg += f'{v_label:+.2f}\n ' x_label_svg = "" if n_days > 10: step = max(1, n_days // 8) x_idx = list(range(0, n_days, step)) else: x_idx = list(range(n_days)) for idx in x_idx: if idx < n_days: x = x_positions[idx] x_label_svg += f'{days_sorted[idx]["date"][5:]}\n ' last_cum_pt = cum_pts[-1].split(",") if cum_pts else ["0", "0"] daily_chart_html = f'''
{grid_h_svg} {y_left_svg}{y_right_svg}{x_label_svg}
{n_days} giorni di trading Daily PnL (asse sx) Cumulativo (asse dx): {cum_pnl[-1]:+.2f} USDT
''' # Posizioni aperte if positions: rows = [] for p in positions: cls = "good" if p["pnl"] >= 0 else "bad" rows.append(f"{p['symbol']}{p['strategy']}{p['opened_at_fmt']}{p['side']}{p['size_fmt']}{p['entry_fmt']}{p['mark_fmt']}{p['sl_fmt']}{p['pnl_fmt']} USDT{p['lev']}x") positions_html = "" + "".join(rows) + "
SymbolStrategiaApertura (Roma)SideSizeEntryMarkSLPnLLev
" else: positions_html = '
Nessuna posizione aperta
' # Ordini recenti (trade chiusi Bybit, no miei test, con Gross/Fee/Net per riga) if recent: rows = [] for o in recent: warn = " ⚠️placeholder" if o.get("placeholder") else "" pnl_gross = o.get("pnl_gross", 0) fee = o.get("fee", 0) pnl_net = o.get("pnl_net", o.get("pnl", 0)) pnl_net_class = "good" if pnl_net > 0 else "bad" if pnl_net < 0 else "" rows.append(f"{o['created_at_fmt']}{o['symbol']}{o['side']}{o['qty_fmt']}{o['price_fmt']}{warn}{o['notional_fmt']}{o['strategy']}{_fmt_usdt(pnl_gross)}-{_fmt_usdt(fee)}{_fmt_usdt(pnl_net)}") if rows: orders_html = f"" + "".join(rows) + "
QuandoSymbolSideQtyPriceNotionalStrategyGrossFeeNet
Trade chiusi Bybit (no miei test). {len(rows)} trade mostrati.
" else: orders_html = '
Nessun trade chiuso reale (solo miei test)
' else: orders_html = '
Nessun trade chiuso
' # HTML finale html = f""" v2 Stats Dashboard — 2° Account

📊 v2 Stats Dashboard — 2° Account (U05gSwYlB5) Da certificare {certification_count} | Anomalie {anomaly_count}

Auto-refresh 30s • Generato: {generated_at} • /json per API
{classification_alert_html} {operational_exclusions_html}
{kpi_html}

🎯 Sharpe / Sortino per Strategia (per-trade, rf=0)

{strategy_sharpe_html}

📈 Equity Curve

{equity_html}

📋 Posizioni Aperte (live Bybit)

{positions_html}

⏱ Durata statistica di tutti i trade

{duration_html}

📊 Per Strategia

{strategy_html}

💎 Per Asset

{asset_html}

🔥 Heatmap Giorno × Ora (chiusure trade)

{heatmap_html}

Risultati numerici per giorno e ora

{heatmap_stats_html}

📅 Trade per Giorno

{trades_by_day_html}

📊 Daily PnL (barre) + Cumulativo (linea)

{daily_chart_html}

📜 Ordini Recenti (ultimi 20)

{orders_html}
v2_stats_dashboard.py — 2° account Bybit (U05gSwYlB5)
""" return html # === DATA ASSEMBLY === def assemble_data(): """Dati SOLO da Bybit (no webhook DB orders). Trade chiusi + posizioni aperte, filtrati dal 1° agosto ed escludendo MIEI test Pine-style.""" all_closed_trades = get_closed_trades_bybit() operational_exclusions = [ trade for trade in all_closed_trades if trade.get("performance_excluded") ] performance_candidates = [ trade for trade in all_closed_trades if not trade.get("performance_excluded") ] unclassified_trades = [ trade for trade in performance_candidates if trade.get("strategy") == UNCLASSIFIED_STRATEGY ] # Gate di certificazione: un trade sospeso non influenza alcun KPI finche' # l'operatore non approva un abbinamento esatto per close order ID. closed_trades = [ trade for trade in performance_candidates if trade.get("strategy") != UNCLASSIFIED_STRATEGY ] positions = get_open_positions_bybit_no_test() balance = get_bybit_balance() queue_stats = get_queue_stats() # Carica anche ordini dal DB per la statistica 'posizioni aperte per giorno di entry' db_orders = get_orders(limit=2000, since_date="2026-08-01") kpis = compute_kpis_from_bybit(closed_trades, positions, balance) equity = compute_equity_curve_from_bybit(closed_trades, balance) by_strategy = compute_by_strategy_from_bybit(closed_trades) by_asset = compute_by_asset_from_bybit(closed_trades) asset_trade_details = compute_asset_trade_details(closed_trades) strategy_trade_details = compute_strategy_trade_details(closed_trades) trade_duration_stats = compute_trade_duration_stats(closed_trades) by_asset_strategy = compute_by_asset_strategy_from_bybit(closed_trades) top_bad = compute_top_bad_from_bybit(closed_trades, top_n=5) heatmap = compute_heatmap_from_bybit(closed_trades) trades_by_day = compute_trades_by_day(closed_trades) single_losses = compute_top_single_losses(closed_trades, top_n=5) open_by_day = compute_entries_by_day(db_orders) positions_fmt = [] for p in positions: size = _safe_float(p.get("size")) entry = _safe_float(p.get("avgPrice")) mark = _safe_float(p.get("markPrice")) sl = _safe_float(p.get("stopLoss")) pnl = _safe_float(p.get("unrealisedPnl")) lev = _safe_float(p.get("leverage"), 1) sym = p.get("symbol", "") # FIX 11/08 (Mattia): per posizioni live, prima controlla override esplicito # (SOURCE Pine vs Python). Fallback a get_strategy_for_symbol (CSV). pos_override = get_position_strategy_override() if sym.upper() in pos_override: strat = pos_override[sym.upper()] else: strat = get_strategy_for_symbol(sym) strat_labels = { "vptr3": "VPTR3", "rettangolo": "RETTANGOLO", "rettangolo_simple": "RETT_SIMPLE", "ma_trailing": "MA_TRAILING", "supertrend_bosw": "SUPERTREND_BOSW", } strat_label = strat_labels.get(strat, strat.upper()) positions_fmt.append({ "symbol": sym, "strategy": strat_label, "opened_at_fmt": _fmt_review_ts(p.get("createdTime")), "side": p.get("side", ""), "size": size, "size_fmt": f"{size:,.4f}", "entry": entry, "entry_fmt": f"{entry:,.5f}", "mark": mark, "mark_fmt": f"{mark:,.5f}", "sl": sl, "sl_fmt": f"{sl:,.5f}" if sl > 0 else "—", "pnl": pnl, "pnl_fmt": f"{pnl:,.2f}", "lev": f"{lev:.0f}" if lev > 0 else "—", }) # Trade chiusi recenti (gia' filtrati per data e test) - max 20 # FIX 2026-08-10 (Mattia): ordina per updated_at DESC, altrimenti closed_trades[:20] prende # i primi che Bybit restituisce (casuali), non i piu' recenti. recent = [] closed_trades_sorted_recent = sorted( closed_trades, key=lambda t: t.get("updated_at") or t.get("created_at") or "", reverse=True, ) for t in closed_trades_sorted_recent[:20]: _ts = _parse_bybit_ts(t.get("updated_at") or t.get("created_at")) recent.append({ "created_at_fmt": _ts.strftime("%Y-%m-%d %H:%M") if _ts else "", "symbol": t.get("symbol", ""), "side": t.get("side", ""), "qty_fmt": f"{t.get('qty', 0):,.4f}", "price_fmt": f"{t.get('entry_price', 0):,.5f}", "notional_fmt": f"{t.get('qty', 0) * t.get('entry_price', 0):,.2f}", "strategy": t.get("strategy", ""), "pnl_gross": t.get("pnl_gross", 0), "fee": t.get("fee", 0), "pnl_net": t.get("pnl_net", t.get("pnl", 0)), "pnl": t.get("pnl_net", t.get("pnl", 0)), # alias "placeholder": False, "is_real": True, "is_test": False, }) return { "kpis": kpis, "equity": equity, "by_strategy": by_strategy, "by_asset": by_asset, "asset_trade_details": asset_trade_details, "strategy_trade_details": strategy_trade_details, "trade_duration_stats": trade_duration_stats, "by_asset_strategy": by_asset_strategy, "top_bad": top_bad, "heatmap": heatmap, "trades_by_day": trades_by_day, "single_losses": single_losses, "open_by_day": open_by_day, "positions": positions_fmt, "recent_orders": recent, "unclassified_trades": unclassified_trades, "operational_exclusions": operational_exclusions, "classification_integrity": { "status": "OK" if not unclassified_trades else "WARNING", "classified": len(closed_trades), "unclassified": len(unclassified_trades), "pending_net_excluded": sum( _safe_float(trade.get("pnl_net", trade.get("pnl", 0))) for trade in unclassified_trades ), "operational_excluded": len(operational_exclusions), "rule": "exact_order_id_or_unique_entry_lifecycle", }, "queue_stats": queue_stats, "generated_at": datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S"), } # === HTTP HANDLER === class StatsHandler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): pass def _check_auth(self): # Direct access requested for V2 dashboard and anomaly management. if os.environ.get("V2_STATS_AUTH_ENABLED", "0").strip().lower() not in {"1", "true", "yes"} or not AUTH_PASS: return True auth = self.headers.get("Authorization", "") if not auth.startswith("Basic "): return False try: import base64 decoded = base64.b64decode(auth[6:]).decode("utf-8", errors="ignore") if ":" not in decoded: return False u, p = decoded.split(":", 1) return hmac.compare_digest(u, AUTH_USER) and hmac.compare_digest(p, AUTH_PASS) except Exception: return False def _send_unauthorized(self): self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="v2-stats"') self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"401 Unauthorized") def _send_json(self, code, data, attachment_name=None): body = json.dumps(data, default=str, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") if attachment_name: self.send_header("Content-Disposition", f'attachment; filename="{attachment_name}"') self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(body) def _send_html(self, code, body): body_bytes = body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body_bytes))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body_bytes) def _send_evidence(self, path): marker = "/review/evidence/" stored_name = path.split(marker, 1)[1] if marker in path else "" if not stored_name or stored_name != Path(stored_name).name: return self._send_json(404, {"error": "evidence_not_found"}) target = REVIEW_EVIDENCE_DIR / stored_name if not target.is_file(): return self._send_json(404, {"error": "evidence_not_found"}) suffix = target.suffix.lower() content_type = {".png": "image/png", ".jpg": "image/jpeg", ".webp": "image/webp"}.get(suffix) if content_type is None: return self._send_json(404, {"error": "evidence_not_found"}) body = target.read_bytes() self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "private, no-store") self.send_header("Content-Disposition", "inline") self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(body) def do_GET(self): if not self._check_auth(): return self._send_unauthorized() if urlparse(self.path).path.rstrip("/") == "/healthz": return self._send_json(200, {"status":"ok", "service":"v2-stats-dashboard"}) # Preserve serialized report/review access, while health stays independent. with DATA_REQUEST_LOCK: return self._do_GET_data() def _do_GET_data(self): if not self._check_auth(): return self._send_unauthorized() path = urlparse(self.path).path.rstrip("/") if "/review/evidence/" in path: return self._send_evidence(path) is_review = path == "/review" or path.endswith("/review") is_review_export = path == "/review/export.json" or path.endswith("/review/export.json") try: data = assemble_data() except Exception as e: import traceback return self._send_html(500, f"

500

{traceback.format_exc()}
") if is_review_export: return self._send_json(200, { "generated_at": datetime.now(timezone.utc).isoformat(), "service": "v2-stats-dashboard", "policy_effect": "none_pending_review", "proposals": _load_review_queue(), }, attachment_name="v2_classification_review_queue.json") elif is_review: query = parse_qs(urlparse(self.path).query) if query.get("certified") == ["1"]: message = "Trade certificato. Statistiche aggiornate." elif query.get("rejected") == ["1"]: message = "Proposta rifiutata e spostata nelle anomalie." elif query.get("created") == ["1"]: message = "Proposta registrata nella coda di revisione." else: message = "" main_url = "/v2-monthly/" if self.headers.get("X-Dashboard-Base") == "/v2-monthly/" else "/v2-stats/" return self._send_html(200, render_review_dashboard(data, message=message, main_url=main_url)) elif path in ("", "/", "/index") or path.endswith("/v2-stats"): return self._send_html(200, render_dashboard(data)) elif path == "/json": data["review_counts"] = {"certifications": len(_certification_candidates(data)), "anomalies": len(_review_candidates(data))} return self._send_json(200, data) elif path == "/healthz": return self._send_json(200, {"status": "ok", "service": "v2-stats-dashboard"}) else: self.send_response(404) self.end_headers() self.wfile.write(b"404") def do_POST(self): with DATA_REQUEST_LOCK: return self._do_POST_data() def _do_POST_data(self): if not self._check_auth(): return self._send_unauthorized() path = urlparse(self.path).path.rstrip("/") action = "" for suffix, name in ( ("/review/proposals", "proposal"), ("/review/certify", "certify"), ("/review/reject", "reject"), ): if path == suffix or path.endswith(suffix): action = name break if not action: return self._send_json(405, {"error": "method_not_allowed"}) origin = self.headers.get("Origin", "").strip() host = self.headers.get("Host", "").strip() if origin and urlparse(origin).netloc != host: return self._send_json(403, {"error": "origin_not_allowed"}) try: length = int(self.headers.get("Content-Length", "0")) except ValueError: return self._send_json(400, {"error": "invalid_content_length"}) if length <= 0 or length > MAX_REVIEW_BODY_BYTES: return self._send_json(413, {"error": "invalid_payload_size"}) try: body = self.rfile.read(length) form, attachment = _parse_review_form(self.headers.get("Content-Type", ""), body) data = assemble_data() if action == "certify": if attachment is not None: raise ValueError("Allegato non previsto nella certificazione rapida") _certify_trade(form, data) elif action == "reject": if attachment is not None: raise ValueError("Allegato non previsto nel rifiuto rapido") _reject_certification(form, data) else: _create_review_proposal(form, data, attachment=attachment) except ValueError as exc: try: data = assemble_data() review_path = path.rsplit("/", 1)[0] main_url = "/v2-monthly/" if self.headers.get("X-Dashboard-Base") == "/v2-monthly/" else "/v2-stats/" return self._send_html(400, render_review_dashboard(data, error=str(exc), main_url=main_url)) except Exception: return self._send_json(400, {"error": str(exc)}) except Exception as exc: log(f"review proposal err: {exc}") return self._send_json(500, {"error": "review_save_failed"}) self.send_response(303) result_flag = "certified=1" if action == "certify" else "rejected=1" if action == "reject" else "created=1" main_url = "/v2-monthly/" if self.headers.get("X-Dashboard-Base") == "/v2-monthly/" else "/v2-stats/" self.send_header("Location", main_url if action != "reject" else main_url + "review?" + result_flag) self.send_header("Cache-Control", "no-store") self.end_headers() def main(): log(f"v2 stats dashboard avviato su {LISTEN_HOST}:{PORT}") log(f"DB: {DB_PATH}") log(f"Auth: {'abilitata' if AUTH_PASS else 'disabilitata (no password settata)'}") httpd = ThreadingHTTPServer((LISTEN_HOST, PORT), StatsHandler) try: httpd.serve_forever() except KeyboardInterrupt: pass finally: httpd.server_close() if __name__ == "__main__": main()