from __future__ import annotations from dataclasses import asdict, dataclass, field from typing import Any from crypto_spot_bot.config import Settings from crypto_spot_bot.storage import Storage @dataclass(slots=True) class LearningAdjustment: symbol: str pattern: str sample_size: int net_pnl: float win_rate: float confidence_adjustment: float reason: str adaptive_rules: dict[str, Any] = field(default_factory=dict) def as_dict(self) -> dict[str, Any]: return asdict(self) @dataclass(slots=True) class LearningState: enabled: bool sample_size: int net_pnl: float win_rate: float symbol_stats: dict[str, dict[str, Any]] = field(default_factory=dict) pattern_stats: dict[str, dict[str, Any]] = field(default_factory=dict) adaptive_rules: dict[str, Any] = field(default_factory=dict) def as_dict(self) -> dict[str, Any]: return asdict(self) class TradeLearner: def __init__(self, settings: Settings, storage: Storage): self.settings = settings self.storage = storage self.state = LearningState( enabled=settings.learning_enabled, sample_size=0, net_pnl=0.0, win_rate=0.0, adaptive_rules=_neutral_rules(settings, "мало закрытых сделок для изменения правил"), ) def refresh(self) -> LearningState: if not self.settings.learning_enabled: self.state = LearningState( enabled=False, sample_size=0, net_pnl=0.0, win_rate=0.0, adaptive_rules={"enabled": False, "reason": "обучение выключено"}, ) self.storage.set_runtime("learning_state", self.state.as_dict()) return self.state trades = self.storage.closed_trades(self.settings.learning_lookback_trades) total_net = sum(float(trade.get("net_pnl") or 0.0) for trade in trades) wins = sum(1 for trade in trades if float(trade.get("net_pnl") or 0.0) > 0) symbol_stats = _group_stats(trades, "symbol") pattern_stats = _group_stats(trades, "entry_pattern") reason_stats = _group_stats(trades, "reason") adaptive_rules = _build_adaptive_rules( trades=trades, settings=self.settings, total_net=total_net, wins=wins, symbol_stats=symbol_stats, pattern_stats=pattern_stats, reason_stats=reason_stats, ) self.state = LearningState( enabled=True, sample_size=len(trades), net_pnl=round(total_net, 6), win_rate=round(wins / len(trades), 4) if trades else 0.0, symbol_stats=symbol_stats, pattern_stats=pattern_stats, adaptive_rules=adaptive_rules, ) self.storage.set_runtime("learning_state", self.state.as_dict()) return self.state def adjustment_for(self, symbol: str, pattern: str) -> LearningAdjustment: if not self.settings.learning_enabled: return LearningAdjustment(symbol, pattern, 0, 0.0, 0.0, 0.0, "обучение выключено") state = self.state symbol_stat = state.symbol_stats.get(symbol, {}) pattern_stat = state.pattern_stats.get(pattern, {}) symbol_adj = self._stat_adjustment(symbol_stat) pattern_adj = self._stat_adjustment(pattern_stat) adjustment = _clamp( symbol_adj + pattern_adj, -self.settings.learning_max_adjustment, self.settings.learning_max_adjustment, ) samples = int(symbol_stat.get("sample_size", 0)) + int(pattern_stat.get("sample_size", 0)) net_pnl = float(symbol_stat.get("net_pnl", 0.0)) + float(pattern_stat.get("net_pnl", 0.0)) win_rate = _weighted_win_rate(symbol_stat, pattern_stat) if samples < self.settings.learning_min_samples: adjustment = 0.0 reason = "мало закрытых сделок для вывода" elif adjustment > 0: reason = "прошлые сделки по символу/шаблону были лучше среднего" elif adjustment < 0: reason = "прошлые сделки по символу/шаблону были убыточными" else: reason = "статистика нейтральна" return LearningAdjustment( symbol=symbol, pattern=pattern, sample_size=samples, net_pnl=round(net_pnl, 6), win_rate=round(win_rate, 4), confidence_adjustment=round(adjustment, 4), reason=reason, adaptive_rules=self.rules_for(symbol, pattern), ) def rules_for(self, symbol: str, pattern: str = "") -> dict[str, Any]: rules = dict(self.state.adaptive_rules or {}) symbol_adjustments = rules.get("symbol_threshold_adjustments") or {} pattern_adjustments = rules.get("pattern_threshold_adjustments") or {} symbol_multipliers = rules.get("symbol_position_multipliers") or {} pattern_multipliers = rules.get("pattern_position_multipliers") or {} symbol_adjustment = float(symbol_adjustments.get(symbol, 0.0) or 0.0) pattern_adjustment = float(pattern_adjustments.get(pattern, 0.0) or 0.0) base_adjustment = float(rules.get("entry_threshold_adjustment", 0.0) or 0.0) base_multiplier = float(rules.get("position_size_multiplier", 1.0) or 1.0) symbol_multiplier = float(symbol_multipliers.get(symbol, 1.0) or 1.0) pattern_multiplier = float(pattern_multipliers.get(pattern, 1.0) or 1.0) max_multiplier = max(1.0, self.settings.learning_max_position_multiplier) effective = _clamp( base_adjustment + symbol_adjustment + pattern_adjustment, -self.settings.learning_max_adjustment, self.settings.learning_max_adjustment, ) effective_multiplier = _clamp(base_multiplier * symbol_multiplier * pattern_multiplier, 0.25, max_multiplier) rules["symbol_entry_threshold_adjustment"] = round(symbol_adjustment, 4) rules["pattern_entry_threshold_adjustment"] = round(pattern_adjustment, 4) rules["effective_entry_threshold_adjustment"] = round(effective, 4) rules["symbol_position_size_multiplier"] = round(symbol_multiplier, 4) rules["pattern_position_size_multiplier"] = round(pattern_multiplier, 4) rules["effective_position_size_multiplier"] = round(effective_multiplier, 4) rules["symbol_blocked"] = symbol in set(rules.get("blocked_symbols") or []) rules["pattern_blocked"] = pattern in set(rules.get("blocked_patterns") or []) return rules def _stat_adjustment(self, stat: dict[str, Any]) -> float: sample_size = int(stat.get("sample_size", 0)) if sample_size < self.settings.learning_min_samples: return 0.0 net_pnl = float(stat.get("net_pnl", 0.0)) win_rate = float(stat.get("win_rate", 0.0)) avg_pnl = net_pnl / sample_size raw = (win_rate - 0.5) * 0.12 + avg_pnl * 0.05 return _clamp( raw, -self.settings.learning_max_adjustment / 2, self.settings.learning_max_adjustment / 2, ) def _group_stats(trades: list[dict[str, Any]], key: str) -> dict[str, dict[str, Any]]: buckets: dict[str, list[dict[str, Any]]] = {} for trade in trades: raw = str(trade.get(key) or "неизвестно").strip() or "неизвестно" buckets.setdefault(raw, []).append(trade) result: dict[str, dict[str, Any]] = {} for name, rows in buckets.items(): net = sum(float(row.get("net_pnl") or 0.0) for row in rows) wins = sum(1 for row in rows if float(row.get("net_pnl") or 0.0) > 0) losses = sum(1 for row in rows if float(row.get("net_pnl") or 0.0) < 0) result[name] = { "sample_size": len(rows), "net_pnl": round(net, 6), "win_count": wins, "loss_count": losses, "win_rate": round(wins / len(rows), 4) if rows else 0.0, "average_net_pnl": round(net / len(rows), 6) if rows else 0.0, } return result def _weighted_win_rate(*stats: dict[str, Any]) -> float: wins = 0.0 samples = 0.0 for stat in stats: sample_size = float(stat.get("sample_size", 0) or 0) wins += float(stat.get("win_rate", 0.0) or 0.0) * sample_size samples += sample_size return wins / samples if samples else 0.0 def _build_adaptive_rules( *, trades: list[dict[str, Any]], settings: Settings, total_net: float, wins: int, symbol_stats: dict[str, dict[str, Any]], pattern_stats: dict[str, dict[str, Any]], reason_stats: dict[str, dict[str, Any]], ) -> dict[str, Any]: sample_size = len(trades) rules = _neutral_rules(settings) rules["sample_size"] = sample_size rules["net_pnl"] = round(total_net, 6) rules["win_rate"] = round(wins / sample_size, 4) if sample_size else 0.0 rules["exit_reason_stats"] = reason_stats if sample_size < settings.learning_min_samples: rules["reasons"].append("мало закрытых сделок для изменения правил") return rules win_rate = wins / sample_size if sample_size else 0.0 if total_net < 0: rules["risk_mode"] = "selective" rules["trade_permission"] = "selective_growth" rules["reasons"].append( "общая статистика обучения убыточна: глобальная защита капитала выключена, блокируются только плохие пары/паттерны" ) elif win_rate >= 0.55: threshold = -min(settings.learning_max_adjustment / 2, (win_rate - 0.5) * 0.08) rules["entry_threshold_adjustment"] = round(threshold, 4) rules["risk_mode"] = "growth" rules["trade_permission"] = "positive_expectancy" rules["position_size_multiplier"] = _global_position_multiplier(total_net, win_rate, sample_size, settings) rules["reasons"].append("общая статистика обучения положительная: хорошие сетапы можно масштабировать по Kelly") for name, stat in symbol_stats.items(): adjustment = _threshold_adjustment_from_stat(stat, settings) if adjustment: rules["symbol_threshold_adjustments"][name] = adjustment multiplier = _position_multiplier_from_stat(stat, settings) if multiplier > 1.0: rules["symbol_position_multipliers"][name] = multiplier if _should_block_stat(stat, settings): rules["blocked_symbols"].append(name) for name, stat in pattern_stats.items(): adjustment = _threshold_adjustment_from_stat(stat, settings) if adjustment: rules["pattern_threshold_adjustments"][name] = adjustment multiplier = _position_multiplier_from_stat(stat, settings) if multiplier > 1.0: rules["pattern_position_multipliers"][name] = multiplier if _should_block_stat(stat, settings): rules["blocked_patterns"].append(name) for reason, stat in reason_stats.items(): if not _is_bad_stat(stat, settings): continue reason_text = reason.lower() if "ema50" in reason_text: rules["ema_exit_mode"] = "profit_only" rules["reasons"].append("выход по EMA50 убыточен: разрешен только при прибыли после издержек") if "rsi" in reason_text: rules["rsi_exit_mode"] = "profit_only" rules["reasons"].append("выход по RSI убыточен: разрешен только при прибыли после издержек") rules["blocked_symbols"] = sorted(set(rules["blocked_symbols"])) rules["blocked_patterns"] = sorted(set(rules["blocked_patterns"])) validation = _closed_trade_validation(trades, rules) rules["validation"] = validation if validation["status"] == "rejected": fallback = _neutral_rules(settings, "адаптивные правила не прошли проверку на закрытых сделках") fallback["validation"] = validation return fallback return rules def _neutral_rules(settings: Settings, reason: str | None = None) -> dict[str, Any]: rules: dict[str, Any] = { "enabled": settings.learning_enabled, "sample_size": 0, "net_pnl": 0.0, "win_rate": 0.0, "round_trip_cost_percent": round(_round_trip_cost_percent(settings), 4), "entry_threshold_adjustment": 0.0, "position_size_multiplier": 1.0, "risk_mode": "neutral", "trade_permission": "normal", "allow_new_entries": True, "reduce_exposure": False, "bad_market_entry_block": False, "target_total_exposure_usdt": settings.max_total_exposure_usdt, "target_symbol_exposure_usdt": settings.max_symbol_exposure_usdt, "current_total_exposure_usdt": 0.0, "over_target_exposure": False, "reduce_now": False, "min_hold_seconds": settings.min_hold_seconds, "min_exit_profit_percent": round(_round_trip_cost_percent(settings) + 0.05, 4), "ema_exit_mode": "normal", "rsi_exit_mode": "normal", "stop_loss_percent": settings.stop_loss_percent, "stop_loss_exit_enabled": settings.stop_loss_exit_enabled, "take_profit_percent": settings.take_profit_percent, "trailing_stop_percent": settings.trailing_stop_percent, "symbol_threshold_adjustments": {}, "pattern_threshold_adjustments": {}, "symbol_position_multipliers": {}, "pattern_position_multipliers": {}, "blocked_symbols": [], "blocked_patterns": [], "exit_reason_stats": {}, "validation": { "method": "closed_trade_counterfactual", "status": "not_enough_data", "sample_size": 0, "baseline_net_pnl": 0.0, "validated_net_pnl": 0.0, "skipped_trades": 0, "skipped_net_pnl": 0.0, "avoided_loss_usdt": 0.0, }, "reasons": [], } if reason: rules["reasons"].append(reason) return rules def _round_trip_cost_percent(settings: Settings) -> float: return (settings.taker_fee_rate * 2 + settings.slippage_rate * 2) * 100 def _closed_trade_validation(trades: list[dict[str, Any]], rules: dict[str, Any]) -> dict[str, Any]: baseline_net = sum(float(trade.get("net_pnl") or 0.0) for trade in trades) skipped_net = 0.0 kept_net = 0.0 skipped_count = 0 for trade in trades: net_pnl = float(trade.get("net_pnl") or 0.0) if _would_skip_closed_trade(trade, rules): skipped_net += net_pnl skipped_count += 1 else: kept_net += net_pnl avoided_loss = kept_net - baseline_net status = "accepted" if avoided_loss >= 0 else "rejected" return { "method": "closed_trade_counterfactual", "status": status, "sample_size": len(trades), "baseline_net_pnl": round(baseline_net, 6), "validated_net_pnl": round(kept_net, 6), "skipped_trades": skipped_count, "skipped_net_pnl": round(skipped_net, 6), "avoided_loss_usdt": round(avoided_loss, 6), } def _would_skip_closed_trade(trade: dict[str, Any], rules: dict[str, Any]) -> bool: reason = str(trade.get("reason") or "").lower() pattern = str(trade.get("entry_pattern") or "") symbol = str(trade.get("symbol") or "") net_pnl = float(trade.get("net_pnl") or 0.0) if net_pnl >= 0: return False if "ema50" in reason and rules.get("ema_exit_mode") == "profit_only": return True if "rsi" in reason and rules.get("rsi_exit_mode") == "profit_only": return True if symbol in set(rules.get("blocked_symbols") or []): return True if pattern in set(rules.get("blocked_patterns") or []): return True return False def _threshold_adjustment_from_stat(stat: dict[str, Any], settings: Settings) -> float: sample_size = int(stat.get("sample_size", 0) or 0) if sample_size < settings.learning_min_samples: return 0.0 net_pnl = float(stat.get("net_pnl", 0.0) or 0.0) win_rate = float(stat.get("win_rate", 0.0) or 0.0) avg_pnl = net_pnl / sample_size if sample_size else 0.0 if net_pnl < 0: raw = (0.5 - win_rate) * 0.08 + min(abs(avg_pnl), 0.08) * 0.4 return round(_clamp(raw, 0.0, settings.learning_max_adjustment / 2), 4) if win_rate >= 0.55 and avg_pnl > 0: raw = -((win_rate - 0.5) * 0.05 + min(avg_pnl, 0.08) * 0.2) return round(_clamp(raw, -settings.learning_max_adjustment / 2, 0.0), 4) return 0.0 def _position_multiplier_from_stat(stat: dict[str, Any], settings: Settings) -> float: sample_size = int(stat.get("sample_size", 0) or 0) if sample_size < settings.learning_min_samples: return 1.0 net_pnl = float(stat.get("net_pnl", 0.0) or 0.0) avg_pnl = float(stat.get("average_net_pnl", 0.0) or 0.0) win_rate = float(stat.get("win_rate", 0.0) or 0.0) if net_pnl <= 0 or avg_pnl <= 0 or win_rate < 0.50: return 1.0 cost_usdt = max(settings.min_position_usdt * _round_trip_cost_percent(settings) / 100, 0.01) edge_score = _clamp(avg_pnl / cost_usdt, 0.0, 1.0) win_score = _clamp((win_rate - 0.50) / 0.35, 0.0, 1.0) raw = 1.0 + edge_score * 0.25 + win_score * 0.35 return round(_clamp(raw, 1.0, max(1.0, settings.learning_max_position_multiplier)), 4) def _global_position_multiplier(total_net: float, win_rate: float, sample_size: int, settings: Settings) -> float: if sample_size < settings.learning_min_samples or total_net <= 0 or win_rate < 0.55: return 1.0 avg_pnl = total_net / max(sample_size, 1) cost_usdt = max(settings.min_position_usdt * _round_trip_cost_percent(settings) / 100, 0.01) edge_score = _clamp(avg_pnl / cost_usdt, 0.0, 1.0) win_score = _clamp((win_rate - 0.55) / 0.30, 0.0, 1.0) raw = 1.0 + edge_score * 0.15 + win_score * 0.20 return round(_clamp(raw, 1.0, max(1.0, settings.learning_max_position_multiplier)), 4) def _is_bad_stat(stat: dict[str, Any], settings: Settings) -> bool: sample_size = int(stat.get("sample_size", 0) or 0) net_pnl = float(stat.get("net_pnl", 0.0) or 0.0) win_rate = float(stat.get("win_rate", 0.0) or 0.0) return sample_size >= settings.learning_min_samples and net_pnl < 0 and win_rate <= 0.25 def _should_block_stat(stat: dict[str, Any], settings: Settings) -> bool: sample_size = int(stat.get("sample_size", 0) or 0) avg_pnl = float(stat.get("average_net_pnl", 0.0) or 0.0) win_rate = float(stat.get("win_rate", 0.0) or 0.0) return sample_size >= max(settings.learning_min_samples * 3, 9) and win_rate == 0 and avg_pnl <= -0.08 def _clamp(value: float, low: float, high: float) -> float: return max(low, min(high, value))