TradeBot/crypto_spot_bot/analytics.py

from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Any

from crypto_spot_bot.config import Settings
from crypto_spot_bot.storage import Storage


def analytics_snapshot(settings: Settings, storage: Storage) -> dict[str, Any]:
    closed = _closed_trades_with_diagnostics(storage, settings.learning_lookback_trades)
    return {
        "pnl": pnl_attribution(closed),
        "probability_calibration": probability_calibration(closed),
        "drift": drift_snapshot(settings, closed, storage.recent_signals(240)),
        "risk_guard": risk_guard_snapshot(settings, closed, storage.latest_equity()),
    }


def pnl_attribution(closed_trades: list[dict[str, Any]]) -> dict[str, Any]:
    total = _trade_stats(closed_trades)
    by_symbol = _group_stats(closed_trades, lambda trade: str(trade.get("symbol", "")))
    by_exit = _group_stats(closed_trades, lambda trade: _exit_category(str(trade.get("reason", ""))))
    by_model = _group_stats(closed_trades, lambda trade: _entry_model(trade))
    recent = [_trade_summary(trade) for trade in closed_trades[:20]]
    return {
        "total": total,
        "by_symbol": by_symbol,
        "by_exit": by_exit,
        "by_model": by_model,
        "recent": recent,
    }


def probability_calibration(closed_trades: list[dict[str, Any]]) -> dict[str, Any]:
    buckets: dict[str, list[dict[str, Any]]] = {}
    for trade in closed_trades:
        probability = _entry_probability(trade)
        if probability is None:
            continue
        low = max(0.0, min(0.95, int(probability * 20) / 20))
        high = low + 0.05
        key = f"{low:.2f}-{high:.2f}"
        buckets.setdefault(key, []).append(trade)
    rows = []
    for key in sorted(buckets):
        trades = buckets[key]
        stats = _trade_stats(trades)
        predicted = [_entry_probability(trade) for trade in trades]
        avg_probability = sum(value for value in predicted if value is not None) / len(trades)
        rows.append(
            {
                "bucket": key,
                "trades": len(trades),
                "avg_probability": round(avg_probability, 4),
                "actual_win_rate": stats["win_rate"],
                "calibration_error": round(stats["win_rate"] - avg_probability, 4),
                "net_pnl": stats["net_pnl"],
                "avg_net_percent": stats["avg_net_percent"],
            }
        )
    status = "insufficient"
    if sum(row["trades"] for row in rows) >= 12:
        avg_abs_error = sum(abs(row["calibration_error"]) * row["trades"] for row in rows) / sum(row["trades"] for row in rows)
        status = "ok" if avg_abs_error <= 0.12 else "warn"
    else:
        avg_abs_error = None
    return {
        "status": status,
        "buckets": rows,
        "samples": sum(row["trades"] for row in rows),
        "avg_abs_error": round(avg_abs_error, 4) if avg_abs_error is not None else None,
    }


def drift_snapshot(settings: Settings, closed_trades: list[dict[str, Any]], signals: list[dict[str, Any]]) -> dict[str, Any]:
    window = max(4, settings.risk_recent_trade_window)
    recent = closed_trades[:window]
    previous = closed_trades[window : window * 2]
    recent_stats = _trade_stats(recent)
    previous_stats = _trade_stats(previous)
    failed_checks = _failed_check_counts(signals)
    status = "insufficient"
    warnings: list[str] = []
    if recent_stats["trades"] >= max(4, min(window, 8)):
        status = "ok"
        if recent_stats["profit_factor"] < settings.risk_min_recent_profit_factor:
            warnings.append("recent_profit_factor_below_min")
        if recent_stats["avg_net_percent"] <= 0:
            warnings.append("recent_expectancy_non_positive")
        if _consecutive_losses(closed_trades) >= settings.risk_max_consecutive_losses:
            warnings.append("consecutive_losses")
        if warnings:
            status = "warn"
    return {
        "status": status,
        "warnings": warnings,
        "recent": recent_stats,
        "previous": previous_stats,
        "failed_checks": failed_checks,
    }


def risk_guard_snapshot(
    settings: Settings,
    closed_trades: list[dict[str, Any]],
    latest_equity: dict[str, Any] | None,
) -> dict[str, Any]:
    if not settings.risk_guard_enabled:
        return {
            "enabled": False,
            "block_new_entries": False,
            "position_size_multiplier": 1.0,
            "symbol_guard_enabled": settings.risk_symbol_guard_enabled,
            "blocked_symbols": [],
            "symbols": [],
            "reasons": [],
        }
    active_trades = _active_universe_trades(settings, closed_trades)
    reasons: list[str] = []
    degraded_reasons: list[str] = []
    consecutive_losses = _consecutive_losses(active_trades)
    if consecutive_losses >= settings.risk_max_consecutive_losses:
        degraded_reasons.append("consecutive_losses")
    today_pnl = _today_pnl(active_trades)
    if today_pnl <= -abs(settings.max_daily_drawdown_usdt):
        reasons.append("daily_loss_limit")
    window = max(4, settings.risk_recent_trade_window)
    recent_stats = _trade_stats(active_trades[:window])
    if recent_stats["trades"] >= max(4, min(window, 8)):
        if recent_stats["profit_factor"] < settings.risk_min_recent_profit_factor:
            degraded_reasons.append("recent_profit_factor_below_min")
        if recent_stats["avg_net_percent"] <= 0:
            degraded_reasons.append("recent_expectancy_non_positive")
    latest_drawdown = float((latest_equity or {}).get("drawdown", 0.0) or 0.0)
    if latest_drawdown >= abs(settings.max_daily_drawdown_usdt):
        reasons.append("equity_drawdown_limit")
    symbol_stats = _symbol_guard_stats(settings, active_trades)
    blocked_symbols = sorted(row["symbol"] for row in symbol_stats if row["block_new_entries"])
    block = bool(reasons)
    all_reasons = reasons + degraded_reasons
    multiplier = 0.0 if block else (settings.risk_reduce_multiplier if degraded_reasons else 1.0)
    return {
        "enabled": True,
        "block_new_entries": block,
        "position_size_multiplier": round(max(0.0, min(1.0, multiplier)), 4),
        "reasons": all_reasons,
        "global_reasons": reasons,
        "degraded_reasons": degraded_reasons,
        "symbol_guard_enabled": settings.risk_symbol_guard_enabled,
        "blocked_symbols": blocked_symbols,
        "symbols": symbol_stats,
        "consecutive_losses": consecutive_losses,
        "today_pnl": round(today_pnl, 6),
        "recent": recent_stats,
    }


def _closed_trades_with_diagnostics(storage: Storage, limit: int) -> list[dict[str, Any]]:
    rows = storage.closed_trades(limit)
    for row in rows:
        row["entry_diagnostics"] = _json_or_default(row.get("entry_diagnostics_json"), {})
    return rows


def _trade_stats(trades: list[dict[str, Any]]) -> dict[str, Any]:
    values = [float(trade.get("net_pnl", 0.0) or 0.0) for trade in trades]
    wins = [value for value in values if value > 0]
    losses = [value for value in values if value < 0]
    gross_profit = sum(wins)
    gross_loss = abs(sum(losses))
    profit_factor = gross_profit / gross_loss if gross_loss > 0 else (999.0 if gross_profit > 0 else 0.0)
    percents = [_trade_net_percent(trade) for trade in trades]
    return {
        "trades": len(trades),
        "wins": len(wins),
        "losses": len(losses),
        "win_rate": round(len(wins) / len(trades), 4) if trades else 0.0,
        "net_pnl": round(sum(values), 6),
        "fees": round(sum(float(trade.get("fee_usdt", 0.0) or 0.0) for trade in trades), 6),
        "avg_net_pnl": round(sum(values) / len(trades), 6) if trades else 0.0,
        "avg_net_percent": round(sum(percents) / len(percents), 4) if percents else 0.0,
        "profit_factor": round(profit_factor, 4),
        "best": round(max(values), 6) if values else 0.0,
        "worst": round(min(values), 6) if values else 0.0,
    }


def _group_stats(trades: list[dict[str, Any]], key_fn) -> list[dict[str, Any]]:
    groups: dict[str, list[dict[str, Any]]] = {}
    for trade in trades:
        key = key_fn(trade) or "unknown"
        groups.setdefault(key, []).append(trade)
    rows = [{"key": key, **_trade_stats(items)} for key, items in groups.items()]
    return sorted(rows, key=lambda row: (row["net_pnl"], row["trades"]), reverse=True)


def _active_universe_trades(settings: Settings, trades: list[dict[str, Any]]) -> list[dict[str, Any]]:
    symbols = {symbol.upper() for symbol in settings.symbols}
    if not symbols:
        return trades
    return [trade for trade in trades if str(trade.get("symbol", "")).upper() in symbols]


def _symbol_guard_stats(settings: Settings, trades: list[dict[str, Any]]) -> list[dict[str, Any]]:
    expectancy_min_samples = max(6, min(settings.risk_recent_trade_window, 10))
    loss_streak_min_samples = max(3, settings.risk_max_consecutive_losses)
    symbol_guard_enabled = settings.risk_symbol_guard_enabled
    rows: list[dict[str, Any]] = []
    for symbol in settings.symbols:
        symbol_trades = [trade for trade in trades if str(trade.get("symbol", "")).upper() == symbol.upper()]
        recent = symbol_trades[: settings.risk_recent_trade_window]
        stats = _trade_stats(recent)
        losses = _consecutive_losses(recent)
        reasons: list[str] = []
        if symbol_guard_enabled:
            if stats["trades"] >= expectancy_min_samples:
                if stats["profit_factor"] < settings.risk_min_recent_profit_factor and stats["avg_net_percent"] <= 0:
                    reasons.append("symbol_expectancy_negative")
            if stats["trades"] >= loss_streak_min_samples:
                if losses >= settings.risk_max_consecutive_losses:
                    reasons.append("symbol_consecutive_losses")
        rows.append(
            {
                "symbol": symbol.upper(),
                "block_new_entries": bool(reasons) if symbol_guard_enabled else False,
                "reasons": reasons,
                "symbol_guard_enabled": symbol_guard_enabled,
                "consecutive_losses": losses,
                **stats,
            }
        )
    return rows


def _trade_summary(trade: dict[str, Any]) -> dict[str, Any]:
    diagnostics = trade.get("entry_diagnostics") if isinstance(trade.get("entry_diagnostics"), dict) else {}
    forecast = diagnostics.get("forecast") if isinstance(diagnostics.get("forecast"), dict) else {}
    return {
        "id": trade.get("id"),
        "symbol": trade.get("symbol"),
        "net_pnl": round(float(trade.get("net_pnl", 0.0) or 0.0), 6),
        "net_percent": _trade_net_percent(trade),
        "fee_usdt": round(float(trade.get("fee_usdt", 0.0) or 0.0), 6),
        "entry_price": trade.get("entry_price"),
        "exit_price": trade.get("exit_price"),
        "closed_at": trade.get("closed_at"),
        "exit_category": _exit_category(str(trade.get("reason", ""))),
        "entry_probability": forecast.get("probability_up"),
        "entry_expected_percent": forecast.get("expected_return_percent"),
        "entry_confidence": trade.get("entry_confidence"),
        "model": forecast.get("model"),
    }


def _trade_net_percent(trade: dict[str, Any]) -> float:
    qty = float(trade.get("qty", 0.0) or 0.0)
    entry_price = float(trade.get("entry_price", 0.0) or 0.0)
    notional = qty * entry_price
    if notional <= 0:
        return 0.0
    return round(float(trade.get("net_pnl", 0.0) or 0.0) / notional * 100, 4)


def _exit_category(reason: str) -> str:
    text = reason.lower()
    if "stop" in text:
        return "stop_loss"
    if "trailing" in text:
        return "trailing_stop"
    if "negative" in text or "turned" in text:
        return "forecast_negative"
    if "weak" in text or "enough edge" in text:
        return "forecast_weak"
    if "take" in text:
        return "take_profit"
    return "other"


def _entry_model(trade: dict[str, Any]) -> str:
    diagnostics = trade.get("entry_diagnostics") if isinstance(trade.get("entry_diagnostics"), dict) else {}
    forecast = diagnostics.get("forecast") if isinstance(diagnostics.get("forecast"), dict) else {}
    return str(forecast.get("model") or "unknown")


def _entry_probability(trade: dict[str, Any]) -> float | None:
    diagnostics = trade.get("entry_diagnostics") if isinstance(trade.get("entry_diagnostics"), dict) else {}
    forecast = diagnostics.get("forecast") if isinstance(diagnostics.get("forecast"), dict) else {}
    value = forecast.get("probability_up")
    try:
        probability = float(value)
    except (TypeError, ValueError):
        return None
    return max(0.0, min(1.0, probability))


def _failed_check_counts(signals: list[dict[str, Any]]) -> dict[str, int]:
    counts: dict[str, int] = {}
    for signal in signals:
        diagnostics = _json_or_default(signal.get("diagnostics_json"), {})
        checks = diagnostics.get("checks") if isinstance(diagnostics, dict) else {}
        if not isinstance(checks, dict):
            continue
        for key, ok in checks.items():
            if not ok:
                counts[str(key)] = counts.get(str(key), 0) + 1
    return dict(sorted(counts.items(), key=lambda item: item[1], reverse=True))


def _consecutive_losses(closed_trades: list[dict[str, Any]]) -> int:
    count = 0
    for trade in closed_trades:
        if float(trade.get("net_pnl", 0.0) or 0.0) < 0:
            count += 1
        else:
            break
    return count


def _today_pnl(closed_trades: list[dict[str, Any]]) -> float:
    today = datetime.now(timezone.utc).date()
    total = 0.0
    for trade in closed_trades:
        closed_at = _parse_datetime(trade.get("closed_at"))
        if closed_at and closed_at.date() == today:
            total += float(trade.get("net_pnl", 0.0) or 0.0)
    return total


def _json_or_default(value: Any, default: Any) -> Any:
    if isinstance(value, (dict, list)):
        return value
    if not isinstance(value, str):
        return default
    try:
        return json.loads(value)
    except json.JSONDecodeError:
        return default


def _parse_datetime(value: Any) -> datetime | None:
    if not isinstance(value, str) or not value:
        return None
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)