ml_diagnostics.py

#!/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())