TradeBot/crypto_spot_bot/learning.py

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,
        "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))