Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Offline and live diagnostics for adaptive bot learning quality. | |
| This script helps validate whether the model is learning directionally useful signals. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import math | |
| import os | |
| from collections import defaultdict | |
| from typing import Any, Dict, List, Optional, Tuple | |
| from urllib.error import URLError | |
| from urllib.request import urlopen | |
| import joblib | |
| from telegram_notifier import TelegramNotifier | |
| def _safe_float(v: Any, default: float = 0.0) -> float: | |
| try: | |
| return float(v) | |
| except Exception: | |
| return default | |
| def _mean(values: List[float]) -> Optional[float]: | |
| return (sum(values) / len(values)) if values else None | |
| def _trade_p_long(trade: Dict[str, Any], open_probs: Dict[Tuple[str, int], float]) -> float: | |
| if "entry_p_long" in trade and trade["entry_p_long"] is not None: | |
| return float(max(0.0, min(1.0, _safe_float(trade["entry_p_long"], 0.5)))) | |
| key = (str(trade.get("coin", "")), int(_safe_float(trade.get("entry_ts", 0), 0))) | |
| return float(max(0.0, min(1.0, open_probs.get(key, 0.5)))) | |
| def _summarize_basic(trades: List[Dict[str, Any]]) -> Dict[str, float]: | |
| n = len(trades) | |
| wins = sum(1 for t in trades if _safe_float(t.get("pnl_usd", 0.0), 0.0) > 0) | |
| pnl = sum(_safe_float(t.get("pnl_usd", 0.0), 0.0) for t in trades) | |
| return { | |
| "trades": float(n), | |
| "wins": float(wins), | |
| "win_rate": (wins / n) if n else 0.0, | |
| "total_pnl_usd": pnl, | |
| "expectancy_usd": (pnl / n) if n else 0.0, | |
| } | |
| def _directional_sanity(trades: List[Dict[str, Any]], open_probs: Dict[Tuple[str, int], float]) -> Dict[str, Optional[float]]: | |
| longs = [t for t in trades if str(t.get("side", "")).lower() == "long"] | |
| shorts = [t for t in trades if str(t.get("side", "")).lower() == "short"] | |
| long_win = [_trade_p_long(t, open_probs) for t in longs if _safe_float(t.get("pnl_usd", 0.0), 0.0) > 0] | |
| long_lose = [_trade_p_long(t, open_probs) for t in longs if _safe_float(t.get("pnl_usd", 0.0), 0.0) <= 0] | |
| short_win = [_trade_p_long(t, open_probs) for t in shorts if _safe_float(t.get("pnl_usd", 0.0), 0.0) > 0] | |
| short_lose = [_trade_p_long(t, open_probs) for t in shorts if _safe_float(t.get("pnl_usd", 0.0), 0.0) <= 0] | |
| # For a healthy long-probability model: | |
| # - long winners should have higher p_long than long losers | |
| # - short winners should have lower p_long than short losers | |
| return { | |
| "long_winners_avg_p_long": _mean(long_win), | |
| "long_losers_avg_p_long": _mean(long_lose), | |
| "short_winners_avg_p_long": _mean(short_win), | |
| "short_losers_avg_p_long": _mean(short_lose), | |
| } | |
| def _direction_aware_brier(trades: List[Dict[str, Any]], open_probs: Dict[Tuple[str, int], float]) -> Optional[float]: | |
| scores: List[float] = [] | |
| for t in trades: | |
| side = str(t.get("side", "")).lower() | |
| if side not in {"long", "short"}: | |
| continue | |
| p_long = _trade_p_long(t, open_probs) | |
| won = 1.0 if _safe_float(t.get("pnl_usd", 0.0), 0.0) > 0 else 0.0 | |
| # Target for long probability: | |
| # - long trade: y=1 if win | |
| # - short trade: y=1 if short loses (long would've won) | |
| y_long = won if side == "long" else (1.0 - won) | |
| scores.append((p_long - y_long) ** 2) | |
| return _mean(scores) | |
| def _calibration_table( | |
| trades: List[Dict[str, Any]], open_probs: Dict[Tuple[str, int], float], side: str | |
| ) -> List[Dict[str, Any]]: | |
| bins: Dict[int, List[Dict[str, Any]]] = defaultdict(list) | |
| for t in trades: | |
| t_side = str(t.get("side", "")).lower() | |
| if t_side != side: | |
| continue | |
| p = _trade_p_long(t, open_probs) | |
| decile = min(9, max(0, int(math.floor(p * 10.0)))) | |
| bins[decile].append(t) | |
| rows: List[Dict[str, Any]] = [] | |
| for decile in range(10): | |
| arr = bins.get(decile, []) | |
| if not arr: | |
| continue | |
| n = len(arr) | |
| wins = sum(1 for t in arr if _safe_float(t.get("pnl_usd", 0.0), 0.0) > 0) | |
| avg_p = _mean([_trade_p_long(t, open_probs) for t in arr]) | |
| rows.append( | |
| { | |
| "decile": decile, | |
| "range": f"{decile / 10:.1f}-{(decile + 1) / 10:.1f}", | |
| "trades": n, | |
| "avg_p_long": avg_p, | |
| "win_rate": (wins / n) if n else None, | |
| } | |
| ) | |
| return rows | |
| def _open_prob_lookup(events: List[Dict[str, Any]]) -> Dict[Tuple[str, int], float]: | |
| out: Dict[Tuple[str, int], float] = {} | |
| for e in events: | |
| if e.get("event") != "open": | |
| continue | |
| coin = str(e.get("coin", "")) | |
| entry_ts = int(_safe_float(e.get("entry_ts", 0), 0)) | |
| p_long = float(max(0.0, min(1.0, _safe_float(e.get("p_long", 0.5), 0.5)))) | |
| out[(coin, entry_ts)] = p_long | |
| return out | |
| def _fetch_status(url: str) -> Optional[Dict[str, Any]]: | |
| try: | |
| with urlopen(url, timeout=3) as response: | |
| if response.status != 200: | |
| return None | |
| body = response.read().decode("utf-8") | |
| return json.loads(body) | |
| except (URLError, TimeoutError, json.JSONDecodeError, ValueError): | |
| return None | |
| def build_report(payload: Dict[str, Any], status_payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: | |
| closed = payload.get("closed_trades", []) or [] | |
| events = payload.get("trade_events", []) or [] | |
| open_probs = _open_prob_lookup(events) | |
| by_coin: Dict[str, List[Dict[str, Any]]] = defaultdict(list) | |
| by_side: Dict[str, List[Dict[str, Any]]] = defaultdict(list) | |
| for t in closed: | |
| by_coin[str(t.get("coin", "unknown"))].append(t) | |
| by_side[str(t.get("side", "unknown")).lower()].append(t) | |
| per_coin = {coin: _summarize_basic(trades) for coin, trades in sorted(by_coin.items())} | |
| per_side = {side: _summarize_basic(trades) for side, trades in sorted(by_side.items())} | |
| report: Dict[str, Any] = { | |
| "state_samples": { | |
| "closed_trades": len(closed), | |
| "trade_events": len(events), | |
| }, | |
| "overall": _summarize_basic(closed), | |
| "per_coin": per_coin, | |
| "per_side": per_side, | |
| "directional_sanity": _directional_sanity(closed, open_probs), | |
| "brier_direction_aware": _direction_aware_brier(closed, open_probs), | |
| "calibration": { | |
| "long": _calibration_table(closed, open_probs, "long"), | |
| "short": _calibration_table(closed, open_probs, "short"), | |
| }, | |
| "runtime_status": status_payload, | |
| } | |
| # High-level warnings to speed up triage. | |
| warnings: List[str] = [] | |
| ds = report["directional_sanity"] | |
| lw = ds.get("long_winners_avg_p_long") | |
| ll = ds.get("long_losers_avg_p_long") | |
| sw = ds.get("short_winners_avg_p_long") | |
| sl = ds.get("short_losers_avg_p_long") | |
| if lw is not None and ll is not None and lw <= ll: | |
| warnings.append("Long directional signal is inverted or weak: long winners do not have higher p_long than long losers.") | |
| if sw is not None and sl is not None and sw >= sl: | |
| warnings.append("Short directional signal is inverted or weak: short winners do not have lower p_long than short losers.") | |
| overall_wr = report["overall"].get("win_rate", 0.0) | |
| overall_exp = report["overall"].get("expectancy_usd", 0.0) | |
| if len(closed) >= 50 and overall_wr < 0.45: | |
| warnings.append("Win rate below 45% after 50+ trades; model may be mis-specified or overtrading.") | |
| if len(closed) >= 50 and overall_exp < 0: | |
| warnings.append("Negative expectancy after 50+ trades; consider reducing exploration and revisiting labels.") | |
| report["warnings"] = warnings | |
| return report | |
| def print_report(report: Dict[str, Any]) -> None: | |
| print("=== Adaptive Bot ML Diagnostics ===") | |
| samples = report["state_samples"] | |
| print(f"Closed trades: {samples['closed_trades']}") | |
| print(f"Trade events: {samples['trade_events']}") | |
| overall = report["overall"] | |
| print( | |
| "Overall: " | |
| f"n={int(overall['trades'])}, " | |
| f"win_rate={overall['win_rate'] * 100:.2f}%, " | |
| f"total_pnl={overall['total_pnl_usd']:.2f}, " | |
| f"expectancy={overall['expectancy_usd']:.4f}" | |
| ) | |
| print("\nBy side:") | |
| for side, s in report["per_side"].items(): | |
| print( | |
| f" {side}: n={int(s['trades'])}, win_rate={s['win_rate'] * 100:.2f}%, " | |
| f"total_pnl={s['total_pnl_usd']:.2f}, expectancy={s['expectancy_usd']:.4f}" | |
| ) | |
| print("\nBy coin:") | |
| for coin, s in report["per_coin"].items(): | |
| print( | |
| f" {coin}: n={int(s['trades'])}, win_rate={s['win_rate'] * 100:.2f}%, " | |
| f"total_pnl={s['total_pnl_usd']:.2f}, expectancy={s['expectancy_usd']:.4f}" | |
| ) | |
| ds = report["directional_sanity"] | |
| print("\nDirectional sanity:") | |
| print( | |
| " Long winners avg p_long: " | |
| f"{ds['long_winners_avg_p_long'] if ds['long_winners_avg_p_long'] is not None else 'n/a'}" | |
| ) | |
| print( | |
| " Long losers avg p_long: " | |
| f"{ds['long_losers_avg_p_long'] if ds['long_losers_avg_p_long'] is not None else 'n/a'}" | |
| ) | |
| print( | |
| " Short winners avg p_long: " | |
| f"{ds['short_winners_avg_p_long'] if ds['short_winners_avg_p_long'] is not None else 'n/a'}" | |
| ) | |
| print( | |
| " Short losers avg p_long: " | |
| f"{ds['short_losers_avg_p_long'] if ds['short_losers_avg_p_long'] is not None else 'n/a'}" | |
| ) | |
| brier = report.get("brier_direction_aware") | |
| print(f"\nDirection-aware Brier (lower is better): {brier if brier is not None else 'n/a'}") | |
| warnings = report.get("warnings", []) | |
| if warnings: | |
| print("\nWarnings:") | |
| for w in warnings: | |
| print(f" - {w}") | |
| runtime = report.get("runtime_status") | |
| if runtime: | |
| print("\nLive status snapshot:") | |
| metrics = runtime.get("metrics", {}) | |
| ml_progress = runtime.get("ml_progress", {}) | |
| print( | |
| " runtime trades=" | |
| f"{int(_safe_float(metrics.get('trades', 0), 0))}, " | |
| f"win_rate={_safe_float(metrics.get('win_rate', 0.0), 0.0) * 100:.2f}%, " | |
| f"pf={_safe_float(metrics.get('profit_factor', 0.0), 0.0):.2f}" | |
| ) | |
| if ml_progress: | |
| print( | |
| " runtime brier_50=" | |
| f"{ml_progress.get('brier_rolling_50')}, " | |
| f"expectancy_30={ml_progress.get('expectancy_usd_30')}, " | |
| f"max_dd={ml_progress.get('max_drawdown_pct')}" | |
| ) | |
| def _telegram_brief(report: Dict[str, Any], title: str) -> str: | |
| overall = report.get("overall", {}) | |
| ds = report.get("directional_sanity", {}) | |
| warnings = report.get("warnings", []) | |
| msg = ( | |
| f"{title}\n" | |
| f"Trades: {int(_safe_float(overall.get('trades', 0), 0))}\n" | |
| f"Win rate: {_safe_float(overall.get('win_rate', 0.0), 0.0) * 100.0:.1f}%\n" | |
| f"Expectancy: {_safe_float(overall.get('expectancy_usd', 0.0), 0.0):+.4f} USD\n" | |
| f"Brier: {report.get('brier_direction_aware')}\n" | |
| f"Long win/lose p_long: {ds.get('long_winners_avg_p_long')} / {ds.get('long_losers_avg_p_long')}\n" | |
| f"Short win/lose p_long: {ds.get('short_winners_avg_p_long')} / {ds.get('short_losers_avg_p_long')}" | |
| ) | |
| if warnings: | |
| msg += "\nWarnings:\n- " + "\n- ".join(str(w) for w in warnings[:3]) | |
| return msg | |
| def parse_args() -> argparse.Namespace: | |
| p = argparse.ArgumentParser(description="Diagnose adaptive bot ML learning quality") | |
| p.add_argument("--state-file", default="learning_state.joblib", help="Path to persisted bot state file") | |
| p.add_argument( | |
| "--status-url", | |
| default="http://127.0.0.1:7860/status/full", | |
| help="Optional bot status endpoint for runtime snapshot", | |
| ) | |
| p.add_argument("--skip-status", action="store_true", help="Skip HTTP status fetch") | |
| p.add_argument("--json", action="store_true", help="Print full report as JSON") | |
| p.add_argument("--send-telegram", action="store_true", help="Send a compact diagnostics report via Telegram") | |
| p.add_argument("--telegram-title", default="Adaptive Bot Diagnostics", help="Telegram report title") | |
| return p.parse_args() | |
| def main() -> int: | |
| args = parse_args() | |
| if not os.path.exists(args.state_file): | |
| print(f"State file not found: {args.state_file}") | |
| return 1 | |
| payload = joblib.load(args.state_file) | |
| status_payload = None if args.skip_status else _fetch_status(args.status_url) | |
| report = build_report(payload, status_payload=status_payload) | |
| if args.json: | |
| print(json.dumps(report, indent=2)) | |
| else: | |
| print_report(report) | |
| if args.send_telegram: | |
| notifier = TelegramNotifier(disable_on_error=True) | |
| notifier.send_info(_telegram_brief(report, args.telegram_title)) | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |