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)