#!/usr/bin/env python3 # QuantitativeTradingEngine.py """ Kapitel Handelsmaschine: Vollstaendige quantitative Handelsmaschine mit Walk-Forward-Backtest und OR-Optimierungsschicht. Eigenschaften: * Rebalancing-Termine sind garantiert Handelstage; die Zahl der tatsaechlich ausgefuehrten Umschichtungen wird ausgewiesen * Spaltenreihenfolge erzwungen * Einheiten konsistent (taegliche Groessen im Modell) * Szenario-Nebenbedingungen vektorisiert * Lookahead-Selbsttest eingebaut * Zusaetzliche Kennzahlen: Calmar, Trefferquote, Turnover, Kostenanteil """ import os import cvxpy as cp import numpy as np import pandas as pd import yfinance as yf from sklearn.covariance import LedoitWolf import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output") os.makedirs(OUTPUT_DIR, exist_ok=True) HANDELSTAGE = 252 class QuantitativeHandelsmaschine: """Walk-Forward-Backtest mit CVaR-Optimierung und Transaktionskosten.""" def __init__(self, universum, benchmark="SPY", jahre=5, startkapital=100_000.0, gebuehrensatz=0.0015, max_gewicht=0.20, alpha=0.95, risikoaversion=1.2, warmup=HANDELSTAGE): self.universum = list(universum) self.benchmark = benchmark self.jahre = jahre self.startkapital = startkapital self.gebuehrensatz = gebuehrensatz self.max_gewicht = max_gewicht self.alpha = alpha self.risikoaversion = risikoaversion self.warmup = warmup self.kurse = None self.benchmark_kurse = None self.renditen = None self.rebalancing_termine = None # -- 1. Datenpipeline ------------------------------------------------ def lade_daten(self): ende = pd.Timestamp.today().normalize() start = ende - pd.DateOffset(years=self.jahre) alle = self.universum + [self.benchmark] print(f"Lade {len(self.universum)} Titel + Benchmark ({self.benchmark}) " f"ueber {self.jahre} Jahre ...") roh = yf.download(alle, start=start, end=ende, auto_adjust=True, progress=False) if roh.empty: raise SystemExit("Download fehlgeschlagen (Netz/Ticker/Rate-Limit pruefen).") if isinstance(roh.columns, pd.MultiIndex): daten = roh["Close"][alle].dropna() # erzwingt eigene Spaltenreihenfolge else: daten = roh[["Close"]].dropna() daten.columns = alle assert list(daten.columns) == alle, "Spaltenreihenfolge weicht ab!" self.kurse = daten[self.universum] self.benchmark_kurse = daten[self.benchmark] self.renditen = self.kurse.pct_change().dropna() # --- erster HANDELStag je Monat, nicht Kalendermonatsanfang ----- monat = self.kurse.index.to_period("M") self.rebalancing_termine = self.kurse.index[~monat.duplicated()] assert self.rebalancing_termine.isin(self.kurse.index).all(), \ "Rebalancing-Termine sind keine Handelstage!" print(f" {len(self.kurse)} Handelstage, " f"{len(self.rebalancing_termine)} Rebalancing-Termine " f"(alle sind Handelstage)") # -- 2. Signal -------------------------------------------------------- def alpha_signal(self, stichtag, rueckblick=HANDELSTAGE): """ 12-1-Momentum: Kursentwicklung der letzten 12 Monate unter Auslassung des letzten Monats (kurzfristige Umkehr herausrechnen). Rueckgabe: TAEGLICHE Renditeerwartung. """ historie = self.kurse.loc[:stichtag] if len(historie) < rueckblick: return np.zeros(len(self.universum)) p_alt = historie.iloc[-rueckblick] p_neu = historie.iloc[-21] # vor ca. einem Monat momentum = (p_neu / p_alt - 1.0).values # Z-Score, damit das Signal skalenfrei wird streuung = momentum.std() score = (momentum - momentum.mean()) / (streuung + 1e-9) # In eine taegliche Renditeerwartung uebersetzen: # Basis 4 % p.a. plus 6 % p.a. je Standardabweichung Momentum. mu_jaehrlich = np.maximum(0.04 + 0.06 * score, 0.0) return mu_jaehrlich / HANDELSTAGE # -- 3./4. Risikomodell und Optimierung ------------------------------- def optimiere(self, stichtag, w_alt, rueckblick=HANDELSTAGE): historie = self.renditen.loc[:stichtag].iloc[-rueckblick:] if len(historie) < 100: return w_alt, False R = historie.values S, N = R.shape mu_taeglich = self.alpha_signal(stichtag, rueckblick) w = cp.Variable(N, nonneg=True) gamma = cp.Variable() u = cp.Variable(S, nonneg=True) cvar = gamma + (1.0 / (S * (1.0 - self.alpha))) * cp.sum(u) kosten = self.gebuehrensatz * cp.norm1(w - w_alt) ziel = cp.Maximize(mu_taeglich @ w - self.risikoaversion * cvar - kosten) bedingungen = [cp.sum(w) == 1, w <= self.max_gewicht, u >= -(R @ w) - gamma] # vektorisiert problem = cp.Problem(ziel, bedingungen) try: problem.solve(solver=cp.CLARABEL) except cp.error.SolverError: return w_alt, False if problem.status not in ("optimal", "optimal_inaccurate") or w.value is None: return w_alt, False gewichte = np.maximum(w.value, 0.0) return gewichte / gewichte.sum(), True # -- 5. Backtest ------------------------------------------------------- def backtest(self): print("Starte Walk-Forward-Backtest ...") N = len(self.universum) gewichte = np.ones(N) / N werte = [self.startkapital] datumsliste = [self.renditen.index[self.warmup]] gewichtshistorie = [] geplant = versucht = ausgefuehrt = 0 kosten_gesamt = 0.0 turnover_gesamt = 0.0 for t in range(self.warmup, len(self.renditen) - 1): heute = self.renditen.index[t] morgen = self.renditen.index[t + 1] ist_termin = heute in self.rebalancing_termine or t == self.warmup if ist_termin: geplant += 1 versucht += 1 neue_gewichte, erfolg = self.optimiere(heute, gewichte) if erfolg: ausgefuehrt += 1 turnover = float(np.abs(neue_gewichte - gewichte).sum()) gebuehr = werte[-1] * turnover * self.gebuehrensatz werte[-1] -= gebuehr kosten_gesamt += gebuehr turnover_gesamt += turnover gewichte = neue_gewichte gewichtshistorie.append((heute, gewichte.copy())) # Rendite von MORGEN auf die HEUTE festgelegten Gewichte anwenden tagesrenditen = self.renditen.iloc[t + 1].values portfoliorendite = float(gewichte @ tagesrenditen) werte.append(werte[-1] * (1.0 + portfoliorendite)) datumsliste.append(morgen) # Gewichte driften mit den Kursen weiter gedriftet = gewichte * (1.0 + tagesrenditen) gewichte = gedriftet / gedriftet.sum() print(f" Rebalancing: {geplant} geplant, {ausgefuehrt} ausgefuehrt " f"({ausgefuehrt/max(geplant,1)*100:.0f} %)") if ausgefuehrt < geplant: print(f" WARNUNG: {geplant - ausgefuehrt} Optimierungen sind " f"fehlgeschlagen - Ursache pruefen!") print(f" Kumulierter Turnover: {turnover_gesamt*100:.0f} % | " f"Transaktionskosten insgesamt: {kosten_gesamt:,.2f} EUR") ergebnis = pd.DataFrame({"Portfolio": werte}, index=datumsliste) benchmark = self.benchmark_kurse.loc[datumsliste] ergebnis["Benchmark"] = benchmark / benchmark.iloc[0] * self.startkapital return ergebnis, gewichtshistorie, kosten_gesamt # -- 6. Auswertung ----------------------------------------------------- @staticmethod def kennzahlen(reihe, risikofrei=0.02): renditen = reihe.pct_change().dropna() jahre = (reihe.index[-1] - reihe.index[0]).days / 365.25 cagr = (reihe.iloc[-1] / reihe.iloc[0]) ** (1 / jahre) - 1 vola = renditen.std() * np.sqrt(HANDELSTAGE) sharpe = (cagr - risikofrei) / vola if vola > 0 else np.nan drawdown = ((reihe - reihe.cummax()) / reihe.cummax()).min() calmar = cagr / abs(drawdown) if drawdown < 0 else np.nan trefferquote = float((renditen > 0).mean()) return {"CAGR": cagr, "Volatilitaet": vola, "Sharpe": sharpe, "Max Drawdown": drawdown, "Calmar": calmar, "Trefferquote": trefferquote, "Jahre": jahre} def bericht(self, ergebnis, kosten_gesamt): k_port = self.kennzahlen(ergebnis["Portfolio"]) k_bench = self.kennzahlen(ergebnis["Benchmark"]) print("\n" + "=" * 84) print(" LEISTUNGSREPORT DER QUANTITATIVEN ENGINE") print("=" * 84) print(f"Zeitraum: {ergebnis.index[0]:%Y-%m-%d} bis {ergebnis.index[-1]:%Y-%m-%d} " f"({k_port['Jahre']:.1f} Jahre) | Startkapital " f"{self.startkapital:,.0f} EUR\n") zeilen = [ ("Endkapital", f"{ergebnis['Portfolio'].iloc[-1]:,.0f} EUR", f"{ergebnis['Benchmark'].iloc[-1]:,.0f} EUR"), ("CAGR (Jahresrendite)", f"{k_port['CAGR']*100:6.2f} %", f"{k_bench['CAGR']*100:6.2f} %"), ("Volatilitaet p.a.", f"{k_port['Volatilitaet']*100:6.2f} %", f"{k_bench['Volatilitaet']*100:6.2f} %"), ("Sharpe Ratio (rf=2 %)", f"{k_port['Sharpe']:6.2f}", f"{k_bench['Sharpe']:6.2f}"), ("Maximum Drawdown", f"{k_port['Max Drawdown']*100:6.2f} %", f"{k_bench['Max Drawdown']*100:6.2f} %"), ("Calmar Ratio", f"{k_port['Calmar']:6.2f}", f"{k_bench['Calmar']:6.2f}"), ("Trefferquote (Tage)", f"{k_port['Trefferquote']*100:6.1f} %", f"{k_bench['Trefferquote']*100:6.1f} %"), ] print(pd.DataFrame(zeilen, columns=["Kennzahl", "OR-Strategie", "Benchmark"]) .to_string(index=False)) anteil = kosten_gesamt / self.startkapital print(f"\nTransaktionskosten: {kosten_gesamt:,.0f} EUR " f"= {anteil*100:.2f} % des Startkapitals " f"= {anteil/k_port['Jahre']*100:.2f} % p.a.") print("Ohne diese Kosten waere die ausgewiesene Rendite entsprechend hoeher -") print("genau deshalb gehoeren sie in den Backtest und nicht daneben.") print("=" * 84) return k_port, k_bench def zeichne(self, ergebnis): plt.figure(figsize=(11, 6)) plt.plot(ergebnis.index, ergebnis["Portfolio"], "b-", lw=2.2, label="OR-Handelsmaschine") plt.plot(ergebnis.index, ergebnis["Benchmark"], "k--", lw=1.5, alpha=0.75, label=f"Benchmark ({self.benchmark})") plt.title("Walk-Forward-Backtest: OR-Strategie gegen Benchmark", fontsize=12) plt.xlabel("Datum") plt.ylabel("Depotwert in EUR") plt.grid(True, linestyle=":", alpha=0.6) plt.legend(loc="upper left") plt.tight_layout() ziel = os.path.join(OUTPUT_DIR, "trading_engine_backtest.png") plt.savefig(ziel, dpi=150) print(f"Chart gespeichert unter '{ziel}'") def lookahead_selbsttest(maschine): """ Prueft, dass das Signal zum Zeitpunkt t NUR Daten bis t verwendet. Methode: kuenstlich die Zukunft veraendern und pruefen, ob sich das Signal von heute dadurch aendert. Tut es das, liegt Lookahead vor. """ stichtag = maschine.kurse.index[len(maschine.kurse) // 2] signal_original = maschine.alpha_signal(stichtag) original_kurse = maschine.kurse.copy() zukunft = maschine.kurse.index > stichtag maschine.kurse.loc[zukunft] *= 3.0 # Zukunft massiv veraendern signal_manipuliert = maschine.alpha_signal(stichtag) maschine.kurse = original_kurse # zuruecksetzen identisch = np.allclose(signal_original, signal_manipuliert) print(f"Lookahead-Selbsttest: Signal bleibt bei manipulierter Zukunft " f"{'UNVERAENDERT (gut)' if identisch else 'VERAENDERT - LOOKAHEAD-BIAS!'}") assert identisch, "Das Signal verwendet Zukunftsdaten!" if __name__ == "__main__": UNIVERSUM = ["AAPL", "MSFT", "NVDA", "GOOGL", "JNJ", "UNH", "PFE", "JPM", "BAC", "GS", "XOM", "CVX", "PG", "KO", "COST"] maschine = QuantitativeHandelsmaschine(UNIVERSUM, benchmark="SPY", jahre=5) maschine.lade_daten() lookahead_selbsttest(maschine) ergebnis, gewichte, kosten = maschine.backtest() maschine.bericht(ergebnis, kosten) maschine.zeichne(ergebnis)