"""Trading-Engine: führt Long/Short über Candles aus und rechnet PnL. Wird sowohl für den Paper-/Simulations-Modus (live auf aktuellen Candles) als auch für die Backtests (Historie) genutzt. Im Simulations-Modus werden Käufe & Verkäufe nur simuliert (kein echtes Geld). """ from __future__ import annotations import json import logging from dataclasses import dataclass, asdict from typing import Dict, List, Optional import numpy as np import pandas as pd from .config import TradingConfig from .strategy import Strategy log = logging.getLogger("trademind.engine") @dataclass class Position: side: str # long entry_price: float size: float # base currency amount entry_time: str stop: float = 0.0 entry_cost: float = 0.0 @dataclass class Trade: side: str entry_price: float exit_price: float size: float entry_time: str exit_time: str pnl: float pnl_pct: float fees: float reason: str = "" @dataclass class Result: final_equity: float total_return_pct: float num_trades: int win_rate: float max_drawdown_pct: float avg_win: float avg_loss: float sharpe: float equity_curve: List[float] trades: List[Trade] def summary(self) -> Dict: return { "final_equity": round(self.final_equity, 2), "total_return_pct": round(self.total_return_pct, 3), "num_trades": self.num_trades, "win_rate": round(self.win_rate, 3), "max_drawdown_pct": round(self.max_drawdown_pct, 3), "avg_win": round(self.avg_win, 2), "avg_loss": round(self.avg_loss, 2), "sharpe": round(self.sharpe, 3), } class Engine: def __init__(self, trading: TradingConfig, strategy: Strategy): self.t = trading self.strategy = strategy # --- Rechenkerne --------------------------------------------------- def _position_size(self, equity: float, price: float) -> float: cash = equity * self.t.position_size_pct return cash / price if price > 0 else 0.0 def run(self, candles: pd.DataFrame) -> Result: """Führt die Strategie über die Candles aus (Simulierung).""" prep = self.strategy.prepare(candles) signals = self.strategy.decide(prep) equity = self.t.initial_balance cash = equity pos: Optional[Position] = None trades: List[Trade] = [] curve: List[float] = [] running_max = equity def equity_at(i: int, close: float) -> float: nonlocal pos if pos is None: return cash val = cash + pos.size * close return val for i in range(1, len(candles)): row = prep.iloc[i] close = float(row["close"]) sig = signals[i] time = str(row["time"]) # Stop-Loss-Check am Candle (intrabar low für Long) if pos is not None and pos.side == "long" and pos.stop > 0: if float(row["low"]) <= pos.stop: exit_price = min(close, pos.stop) cash = self._realize(pos, exit_price, time, "stop_loss", cash) trades.append(pos._trade) # type: ignore[attr-defined] pos = None if pos is None and sig.action == 1: size = self._position_size(equity_at(i, close), close) if size > 0: fee = size * close * self.t.fee_pct exit_px = close * (1 - self.t.slippage_pct) if size * close + fee <= cash: cash -= size * exit_px + fee pos = Position( side="long", entry_price=exit_px, size=size, entry_time=time, stop=sig.stop, entry_cost=fee, ) elif pos is not None and sig.action in (-1, 2): exit_px = close * (1 + self.t.slippage_pct) cash = self._realize(pos, exit_px, time, "signal_exit", cash) trades.append(pos._trade) # type: ignore[attr-defined] pos = None eq = equity_at(i, close) curve.append(eq) running_max = max(running_max, eq) # Ende: offene Position zu Schlusskurs schließen if pos is not None: last_close = float(candles["close"].iloc[-1]) last_time = str(candles["time"].iloc[-1]) cash = self._realize(pos, last_close, last_time, "end_of_data", cash) trades.append(pos._trade) # type: ignore[attr-defined] pos = None final = cash if curve: curve[-1] = final else: final = cash return self._summarize(final, curve, trades) def _realize(self, pos: Position, exit_price: float, time: str, reason: str, cash: float) -> float: """Schließt eine Position; legt das Ergebnis in pos._trade und gibt neues Cash zurück.""" exit_fee = pos.size * exit_price * self.t.fee_pct proceeds = pos.size * exit_price - exit_fee gross = pos.size * (exit_price - pos.entry_price) total_fees = exit_fee + pos.entry_cost pnl = gross - total_fees pnl_pct = (pnl / max(pos.size * pos.entry_price, 1e-9)) * 100 if pos.size else 0.0 pos._trade = Trade( # type: ignore[attr-defined] side=pos.side, entry_price=pos.entry_price, exit_price=exit_price, size=pos.size, entry_time=pos.entry_time, exit_time=time, pnl=pnl, pnl_pct=pnl_pct, fees=total_fees, reason=reason, ) return cash + proceeds def _summarize( self, final: float, curve: list, trades: List[Trade] ) -> Result: curve = curve or [self.t.initial_balance] arr = np.array(curve, dtype=float) peak = np.maximum.accumulate(arr) dd = (peak - arr) / np.where(peak > 0, peak, 1) max_dd = float(dd.max()) if len(dd) else 0.0 rets = arr[1:] / arr[:-1] - 1 if len(arr) > 1 else np.array([0.0]) std = float(np.std(rets)) sharpe = float(np.mean(rets) / std * np.sqrt(len(rets))) if std > 0 else 0.0 wins = [t.pnl for t in trades if t.pnl > 0] losses = [t.pnl for t in trades if t.pnl <= 0] win_rate = (len(wins) / len(trades)) if trades else 0.0 return Result( final_equity=final, total_return_pct=(final / self.t.initial_balance - 1) * 100, num_trades=len(trades), win_rate=win_rate, max_drawdown_pct=max_dd * 100, avg_win=float(np.mean(wins)) if wins else 0.0, avg_loss=float(np.mean(losses)) if losses else 0.0, sharpe=sharpe, equity_curve=[round(x, 2) for x in arr], trades=trades, ) def save_state(path: str, result: Result, params: Dict) -> None: import os os.makedirs(os.path.dirname(path) or ".", exist_ok=True) with open(path, "w", encoding="utf-8") as fh: json.dump( { "summary": result.summary(), "params": params, "trades": [asdict(t) for t in result.trades], }, fh, indent=2, )