Trademind_OpenCode/trademind/trainer.py

"""Antrainieren: evolutionäres Optimieren der Strategie-Parameter & Signalgewichte.

Der Bot wird über viele Simulationen/Backtests (auf Simulations-Daten) darauf
trainiert, seine Strategie-Parameter so anzupassen, dass die Fitness gestiegen
ist. Die 'trainings'-Fähigkeit kommt dadurch zustande, dass die Ergebnisse der
Simulationsläufe als Fitness-Signal (Return, Sharpe, Max Drawdown) genutzt werden.
"""

from __future__ import annotations

import logging
import random
from dataclasses import dataclass
from typing import Callable, Dict, List, Sequence

import numpy as np
import pandas as pd

from .config import StrategyConfig, TradingConfig, TrainingConfig
from .engine import Engine, Result
from .strategy import Strategy, fitness, default_weights

log = logging.getLogger("trademind.trainer")


# --- Parameter-Räume -----------------------------------------------------
def _param_bounds() -> Dict[str, tuple]:
    return {
        "fast_period": (5, 30),
        "slow_period": (20, 60),
        "signal_period": (5, 15),
        "rsi_period": (7, 21),
        "rsi_overbought": (60, 80),
        "rsi_oversold": (20, 40),
        "atr_stop_mult": (1.5, 4.0),
        "w_ema_cross": (0.2, 2.0),
        "w_rsi_long": (0.0, 1.5),
        "w_rsi_exit": (0.0, 1.5),
    }


def random_params(rng: random.Random) -> Dict[str, float]:
    b = _param_bounds()
    p = {k: rng.uniform(lo, hi) for k, (lo, hi) in b.items()}
    # slow muss immer > fast sein
    p["slow_period"] = max(int(p["slow_period"]), int(p["fast_period"]) + 5)
    return p


def build_strategy(base: StrategyConfig, p: Dict[str, float]) -> Strategy:
    sc = StrategyConfig(
        fast_period=int(round(p["fast_period"])),
        slow_period=int(round(p["slow_period"])),
        signal_period=int(round(p["signal_period"])),
        rsi_period=int(round(p["rsi_period"])),
        rsi_overbought=float(p["rsi_overbought"]),
        rsi_oversold=float(p["rsi_oversold"]),
        atr_period=base.atr_period,
        atr_stop_mult=float(p["atr_stop_mult"]),
        allow_long=True,
        allow_short=base.allow_short,
    )
    w = {
        "ema_cross": float(p["w_ema_cross"]),
        "rsi_long": float(p["w_rsi_long"]),
        "rsi_exit": float(p["w_rsi_exit"]),
    }
    return Strategy(sc, w)


def eval_params(
    p: Dict[str, float],
    candles: pd.DataFrame,
    trading: TradingConfig,
    base_strategy: Strategy,
    tcfg: TrainingConfig,
) -> float:
    st = build_strategy(base_strategy.cfg, p)
    eng = Engine(trading, st)
    try:
        res: Result = eng.run(candles)
    except Exception:  # pragma: no cover - defensive
        return -10.0
    return fitness(
        np.array(res.equity_curve[1:] or [1.0]),
        res.final_equity,
        trading.initial_balance,
        res.max_drawdown_pct / 100.0,
        tcfg.fitness_weight_return,
        tcfg.fitness_weight_sharpe,
        tcfg.fitness_weight_drawdown,
    )


def mutate(p: Dict[str, float], rate: float, rng: random.Random) -> Dict[str, float]:
    b = _param_bounds()
    out = dict(p)
    for k, (lo, hi) in b.items():
        if rng.random() < rate:
            width = (hi - lo) * 0.2
            out[k] = min(hi, max(lo, p[k] + rng.uniform(-width, width)))
    out["slow_period"] = max(int(out["slow_period"]), int(out["fast_period"]) + 5)
    return out


def crossover(a: Dict[str, float], b: Dict[str, float], rng: random.Random) -> Dict[str, float]:
    return {k: (a[k] if rng.random() < 0.5 else b[k]) for k in a}


@dataclass
class Individual:
    params: Dict[str, float]
    fitness: float = -1e9


class Trainer:
    def __init__(self, tcfg: TrainingConfig, training: Strategy):
        self.tcfg = tcfg
        self._weights = default_weights()

    def train(
        self,
        candles: pd.DataFrame,
        trading: TradingConfig,
        base_strategy: Strategy,
        progress: Callable[[int, float], None] | None = None,
    ) -> Dict[str, float]:
        """Liefert optimierte Parameter (inkl. Gewichte)."""
        rng = random.Random(self.tcfg.seed)
        pop = [Individual(random_params(rng)) for _ in range(self.tcfg.population)]

        best_params: Dict[str, float] = pop[0].params
        best_fit = -1e18

        for gen in range(self.tcfg.generations):
            for ind in pop:
                ind.fitness = eval_params(
                    ind.params, candles, trading, base_strategy, self.tcfg
                )
            ranked = sorted(pop, key=lambda x: x.fitness, reverse=True)
            if ranked[0].fitness > best_fit:
                best_fit = ranked[0].fitness
                best_params = ranked[0].params

            if progress:
                progress(gen + 1, ranked[0].fitness)

            # Elternteile (Elitismus) + Kinder
            elite = ranked[: max(2, self.tcfg.population // 5)]
            new_pop: List[Individual] = [Individual(dict(ind.params), ind.fitness) for ind in elite]
            while len(new_pop) < self.tcfg.population:
                pa, pb = rng.sample(elite, 2)
                child = crossover(pa.params, pb.params, rng)
                child = mutate(child, self.tcfg.mutation_rate, rng)
                new_pop.append(Individual(child))
            pop = new_pop

        log.info("Training abgeschlossen. Beste Fitness: %.4f", best_fit)
        return best_params


def apply_params(base: StrategyConfig, params: Dict[str, float]) -> Strategy:
    return build_strategy(base, params)