Tobias Zimmermann
150 lines
5.3 KiB
Python
150 lines
5.3 KiB
Python
"""Kernstrategie: Signal-Score aus EMA-Cross + RSI, gewichtet durch antrainierbare Gewichte.
|
|
|
|
Die Gewichte werden beim Training (Parameter-Optimierung) so angepasst, dass die
|
|
Fitness (Return/Sharpe/Drawdown) gestiegen wird. Dadurch 'lernt' der Bot aus den
|
|
Simulations-Ergebnissen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from .config import StrategyConfig
|
|
from .indicators import atr, crossover, crossunder, ema, rsi
|
|
|
|
|
|
@dataclass
|
|
class Signal:
|
|
action: int # +1 long-ein, -1 long-aus, +2 short-ein (fakultativ), 0 halten
|
|
score: float = 0.0
|
|
price: float = 0.0
|
|
stop: float = 0.0
|
|
reason: str = ""
|
|
|
|
|
|
def default_weights() -> Dict[str, float]:
|
|
return {"ema_cross": 1.0, "rsi_long": 0.8, "rsi_exit": 0.6}
|
|
|
|
|
|
class Strategy:
|
|
def __init__(self, cfg: StrategyConfig, weights: Optional[Dict[str, float]] = None):
|
|
self.cfg = cfg
|
|
self.weights = {**default_weights(), **(weights or {})}
|
|
|
|
def parameters(self) -> Dict[str, float]:
|
|
"""Alle anpassbaren Parameter (für den Optimierer)."""
|
|
return {
|
|
"fast_period": self.cfg.fast_period,
|
|
"slow_period": self.cfg.slow_period,
|
|
"signal_period": self.cfg.signal_period,
|
|
"rsi_period": self.cfg.rsi_period,
|
|
"rsi_overbought": self.cfg.rsi_overbought,
|
|
"rsi_oversold": self.cfg.rsi_oversold,
|
|
"atr_stop_mult": self.cfg.atr_stop_mult,
|
|
**{f"w_{k}": v for k, v in self.weights.items()},
|
|
}
|
|
|
|
def set_parameter(self, key: str, value: float) -> None:
|
|
if key.startswith("w_"):
|
|
self.weights[key[2:]] = max(0.0, value)
|
|
return
|
|
if hasattr(self.cfg, key):
|
|
v = int(round(float(value))) if isinstance(getattr(self.cfg, key), int) and key != "rsi_overbought" and key != "rsi_oversold" and key != "atr_stop_mult" else float(value)
|
|
if key.endswith("_period"):
|
|
v = max(2, int(round(float(value))))
|
|
setattr(self.cfg, key, v)
|
|
|
|
def prepare(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Berechnet alle Indikatoren auf dem Candles-Frame. Liefert neuen Frame."""
|
|
out = df.copy()
|
|
c = self.cfg
|
|
out["ema_fast"] = ema(out["close"], c.fast_period)
|
|
out["ema_slow"] = ema(out["close"], c.slow_period)
|
|
out["macd"] = out["ema_fast"] - out["ema_slow"]
|
|
out["signal"] = ema(out["macd"], c.signal_period)
|
|
out["rsi"] = rsi(out["close"], c.rsi_period)
|
|
out["atr"] = atr(out, c.atr_period)
|
|
out["cross_up"] = crossover(out["macd"], out["signal"])
|
|
out["cross_dn"] = crossunder(out["macd"], out["signal"])
|
|
return out
|
|
|
|
def score_row(self, row: pd.Series) -> float:
|
|
"""Gewichteter Score: > 0 Kauf, < 0 Verkauf/Ausstieg."""
|
|
w = self.weights
|
|
s = 0.0
|
|
if row["cross_up"]:
|
|
s += w["ema_cross"]
|
|
if row["cross_dn"]:
|
|
s -= w["ema_cross"]
|
|
rsi = row["rsi"]
|
|
if rsi <= self.cfg.rsi_oversold:
|
|
s += w["rsi_long"]
|
|
if rsi >= self.cfg.rsi_overbought:
|
|
s -= w["rsi_exit"]
|
|
return s
|
|
|
|
def decide(self, prep: pd.DataFrame) -> list[Signal]:
|
|
"""Erzeugt pro Candle ein Signal (für Backtest) bzw. das letzte (Live)."""
|
|
signals: list[Signal] = []
|
|
for i in range(len(prep)):
|
|
row = prep.iloc[i]
|
|
if i < max(self.cfg.slow_period, self.cfg.signal_period):
|
|
signals.append(Signal(0, 0.0, float(row["close"]), 0.0, "warmup"))
|
|
continue
|
|
score = self.score_row(row)
|
|
price = float(row["close"])
|
|
stop = price - self.cfg.atr_stop_mult * row["atr"] if self.cfg.allow_long else price
|
|
if score > 0 and self.cfg.allow_long:
|
|
action = 1
|
|
elif score < 0:
|
|
action = -1
|
|
else:
|
|
action = 0
|
|
reasons = []
|
|
if row["cross_up"]:
|
|
reasons.append("macd_cross_up")
|
|
if row["rsi"] <= self.cfg.rsi_oversold:
|
|
reasons.append("rsi_oversold")
|
|
if score < 0:
|
|
if row["cross_dn"]:
|
|
reasons.append("macd_cross_down")
|
|
if row["rsi"] >= self.cfg.rsi_overbought:
|
|
reasons.append("rsi_overbought")
|
|
signals.append(
|
|
Signal(
|
|
action,
|
|
score,
|
|
price,
|
|
float(stop) if action == 1 else 0.0,
|
|
",".join(reasons) or "neutral",
|
|
)
|
|
)
|
|
return signals
|
|
|
|
def last_signal(self, prep: pd.DataFrame) -> Signal:
|
|
return self.decide(prep)[-1]
|
|
|
|
|
|
def fitness(
|
|
returns: np.ndarray,
|
|
final_equity: float,
|
|
initial: float,
|
|
max_drawdown: float,
|
|
w_return: float = 0.6,
|
|
w_sharpe: float = 0.3,
|
|
w_dd: float = 0.1,
|
|
) -> float:
|
|
"""Fitness-Bewertung für den Optimierer (maximieren)."""
|
|
if len(returns) == 0:
|
|
return -1.0
|
|
total_return = (final_equity / initial) - 1.0
|
|
std = float(np.std(returns))
|
|
sharpe = (float(np.mean(returns)) / std * np.sqrt(len(returns))) if std > 0 else 0.0
|
|
dd_penalty = max_drawdown # 0.3 -> 0.3 Abzug
|
|
score = w_return * total_return + w_sharpe * (sharpe / 10.0) - w_dd * dd_penalty
|
|
return float(score)
|