b7af2f1d9a
Erster Commit des Strangs "Optimierte Entscheidungsfindung mit Python" (Version 04). Die Historie der 71 Commits bis zur Trennung bleibt im uebergeordneten Repository OR_mit_Python liegen, das ab jetzt nur noch Version_03 (eingefroren) verwaltet und Version_04/ ignoriert. Bewusst kein "git subtree split": Der Pfad Version_04/ existiert erst seit der Verzeichnistrennung, ein Split braechte daher nur 7 der 41 einschlaegigen Commits - eine Teilhistorie, die vollstaendig aussieht und es nicht ist. Stand: 5 Teile, 23 Kapitel, 5 Anhaenge, 292 Abschnitte, 703 Querverweise, 325 Indexmarken, 73 Beispielprogramme, 32 SVGs, 4 Plotly-Figuren, 25 Notebooks, PDF mit 715 Seiten. Zusaetzlich in diesem Commit: * pyproject.toml mit Abhaengigkeitsgruppen finance, large-scale, api, figures, dev, empfehlungen. Die abgedruckte requirements.txt bleibt unveraendert daneben bestehen. ortools steht in der Grundausstattung, highspy erst in [large-scale] - so kann der HiGHS-Symbolkonflikt bei der schlanken Installation gar nicht erst auftreten. * Dabei zwei Funde: graphviz wird von erzeuge_architektur_diagramme.py importiert, fehlt aber in requirements.txt (jetzt in [figures]); pymoo steht in requirements.txt, wird aber von keinem Programm importiert, sondern nur im Kapitel Metaheuristiken empfohlen (jetzt in [empfehlungen]). * NEUER_TITEL.md nach Kritik_und_Verbesserungsvorschlaege/ verschoben - es ist die Vorlage des Titelblatts, kein Bestandteil des Werks. Die beiden Fundstellen in PROGRESS.md und erzeuge_titelseite.py nachgezogen. * PROGRESS.md nannte noch den Untertitel der ersten Fassung; auf den tatsaechlichen aus erzeuge_titelseite.py korrigiert. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
623 lines
29 KiB
Text
Generated
623 lines
29 KiB
Text
Generated
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"# Kapitel 21: Die vollständige quantitative Handelsmaschine\n",
|
|
"\n",
|
|
"Begleitnotebook zu *Optimierte Entscheidungsfindung mit Python*. Die Codezellen sind identisch mit den im Buch abgedruckten Programmen.\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Einmalig ausfuehren: installiert alle im Buch verwendeten Pakete.\n",
|
|
"# Lokal in einer virtuellen Umgebung genauso gueltig wie in Google Colab.\n",
|
|
"%pip install --quiet ortools highspy cvxpy scipy numpy pandas polars \\\n",
|
|
" scikit-learn matplotlib plotly pyomo linopy pymoo pydantic openpyxl"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Die Engine\n",
|
|
"\n",
|
|
"`QuantitativeTradingEngine.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# QuantitativeTradingEngine.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Handelsmaschine: Vollstaendige quantitative Handelsmaschine mit\n",
|
|
"Walk-Forward-Backtest und OR-Optimierungsschicht.\n",
|
|
"\n",
|
|
"Eigenschaften:\n",
|
|
" * Rebalancing-Termine sind garantiert Handelstage; die Zahl der\n",
|
|
" tatsaechlich ausgefuehrten Umschichtungen wird ausgewiesen\n",
|
|
" * Spaltenreihenfolge erzwungen\n",
|
|
" * Einheiten konsistent (taegliche Groessen im Modell)\n",
|
|
" * Szenario-Nebenbedingungen vektorisiert\n",
|
|
" * Lookahead-Selbsttest eingebaut\n",
|
|
" * Zusaetzliche Kennzahlen: Calmar, Trefferquote, Turnover, Kostenanteil\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"import os\n",
|
|
"\n",
|
|
"import cvxpy as cp\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"import yfinance as yf\n",
|
|
"from sklearn.covariance import LedoitWolf\n",
|
|
"import matplotlib\n",
|
|
"matplotlib.use(\"Agg\")\n",
|
|
"import matplotlib.pyplot as plt\n",
|
|
"\n",
|
|
"OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), \"output\")\n",
|
|
"os.makedirs(OUTPUT_DIR, exist_ok=True)\n",
|
|
"\n",
|
|
"HANDELSTAGE = 252\n",
|
|
"\n",
|
|
"\n",
|
|
"class QuantitativeHandelsmaschine:\n",
|
|
" \"\"\"Walk-Forward-Backtest mit CVaR-Optimierung und Transaktionskosten.\"\"\"\n",
|
|
"\n",
|
|
" def __init__(self, universum, benchmark=\"SPY\", jahre=5,\n",
|
|
" startkapital=100_000.0, gebuehrensatz=0.0015,\n",
|
|
" max_gewicht=0.20, alpha=0.95, risikoaversion=1.2,\n",
|
|
" warmup=HANDELSTAGE):\n",
|
|
" self.universum = list(universum)\n",
|
|
" self.benchmark = benchmark\n",
|
|
" self.jahre = jahre\n",
|
|
" self.startkapital = startkapital\n",
|
|
" self.gebuehrensatz = gebuehrensatz\n",
|
|
" self.max_gewicht = max_gewicht\n",
|
|
" self.alpha = alpha\n",
|
|
" self.risikoaversion = risikoaversion\n",
|
|
" self.warmup = warmup\n",
|
|
"\n",
|
|
" self.kurse = None\n",
|
|
" self.benchmark_kurse = None\n",
|
|
" self.renditen = None\n",
|
|
" self.rebalancing_termine = None\n",
|
|
"\n",
|
|
" # -- 1. Datenpipeline ------------------------------------------------\n",
|
|
" def lade_daten(self):\n",
|
|
" ende = pd.Timestamp.today().normalize()\n",
|
|
" start = ende - pd.DateOffset(years=self.jahre)\n",
|
|
" alle = self.universum + [self.benchmark]\n",
|
|
"\n",
|
|
" print(f\"Lade {len(self.universum)} Titel + Benchmark ({self.benchmark}) \"\n",
|
|
" f\"ueber {self.jahre} Jahre ...\")\n",
|
|
" roh = yf.download(alle, start=start, end=ende, auto_adjust=True, progress=False)\n",
|
|
" if roh.empty:\n",
|
|
" raise SystemExit(\"Download fehlgeschlagen (Netz/Ticker/Rate-Limit pruefen).\")\n",
|
|
"\n",
|
|
" if isinstance(roh.columns, pd.MultiIndex):\n",
|
|
" daten = roh[\"Close\"][alle].dropna() # erzwingt eigene Spaltenreihenfolge\n",
|
|
" else:\n",
|
|
" daten = roh[[\"Close\"]].dropna()\n",
|
|
" daten.columns = alle\n",
|
|
" assert list(daten.columns) == alle, \"Spaltenreihenfolge weicht ab!\"\n",
|
|
"\n",
|
|
" self.kurse = daten[self.universum]\n",
|
|
" self.benchmark_kurse = daten[self.benchmark]\n",
|
|
" self.renditen = self.kurse.pct_change().dropna()\n",
|
|
"\n",
|
|
" # --- erster HANDELStag je Monat, nicht Kalendermonatsanfang -----\n",
|
|
" monat = self.kurse.index.to_period(\"M\")\n",
|
|
" self.rebalancing_termine = self.kurse.index[~monat.duplicated()]\n",
|
|
" assert self.rebalancing_termine.isin(self.kurse.index).all(), \\\n",
|
|
" \"Rebalancing-Termine sind keine Handelstage!\"\n",
|
|
"\n",
|
|
" print(f\" {len(self.kurse)} Handelstage, \"\n",
|
|
" f\"{len(self.rebalancing_termine)} Rebalancing-Termine \"\n",
|
|
" f\"(alle sind Handelstage)\")\n",
|
|
"\n",
|
|
" # -- 2. Signal --------------------------------------------------------\n",
|
|
" def alpha_signal(self, stichtag, rueckblick=HANDELSTAGE):\n",
|
|
" \"\"\"\n",
|
|
" 12-1-Momentum: Kursentwicklung der letzten 12 Monate unter Auslassung\n",
|
|
" des letzten Monats (kurzfristige Umkehr herausrechnen).\n",
|
|
" Rueckgabe: TAEGLICHE Renditeerwartung.\n",
|
|
" \"\"\"\n",
|
|
" historie = self.kurse.loc[:stichtag]\n",
|
|
" if len(historie) < rueckblick:\n",
|
|
" return np.zeros(len(self.universum))\n",
|
|
"\n",
|
|
" p_alt = historie.iloc[-rueckblick]\n",
|
|
" p_neu = historie.iloc[-21] # vor ca. einem Monat\n",
|
|
" momentum = (p_neu / p_alt - 1.0).values\n",
|
|
"\n",
|
|
" # Z-Score, damit das Signal skalenfrei wird\n",
|
|
" streuung = momentum.std()\n",
|
|
" score = (momentum - momentum.mean()) / (streuung + 1e-9)\n",
|
|
"\n",
|
|
" # In eine taegliche Renditeerwartung uebersetzen:\n",
|
|
" # Basis 4 % p.a. plus 6 % p.a. je Standardabweichung Momentum.\n",
|
|
" mu_jaehrlich = np.maximum(0.04 + 0.06 * score, 0.0)\n",
|
|
" return mu_jaehrlich / HANDELSTAGE\n",
|
|
"\n",
|
|
" # -- 3./4. Risikomodell und Optimierung -------------------------------\n",
|
|
" def optimiere(self, stichtag, w_alt, rueckblick=HANDELSTAGE):\n",
|
|
" historie = self.renditen.loc[:stichtag].iloc[-rueckblick:]\n",
|
|
" if len(historie) < 100:\n",
|
|
" return w_alt, False\n",
|
|
"\n",
|
|
" R = historie.values\n",
|
|
" S, N = R.shape\n",
|
|
" mu_taeglich = self.alpha_signal(stichtag, rueckblick)\n",
|
|
"\n",
|
|
" w = cp.Variable(N, nonneg=True)\n",
|
|
" gamma = cp.Variable()\n",
|
|
" u = cp.Variable(S, nonneg=True)\n",
|
|
"\n",
|
|
" cvar = gamma + (1.0 / (S * (1.0 - self.alpha))) * cp.sum(u)\n",
|
|
" kosten = self.gebuehrensatz * cp.norm1(w - w_alt)\n",
|
|
" ziel = cp.Maximize(mu_taeglich @ w - self.risikoaversion * cvar - kosten)\n",
|
|
"\n",
|
|
" bedingungen = [cp.sum(w) == 1,\n",
|
|
" w <= self.max_gewicht,\n",
|
|
" u >= -(R @ w) - gamma] # vektorisiert\n",
|
|
" problem = cp.Problem(ziel, bedingungen)\n",
|
|
" try:\n",
|
|
" problem.solve(solver=cp.CLARABEL)\n",
|
|
" except cp.error.SolverError:\n",
|
|
" return w_alt, False\n",
|
|
"\n",
|
|
" if problem.status not in (\"optimal\", \"optimal_inaccurate\") or w.value is None:\n",
|
|
" return w_alt, False\n",
|
|
"\n",
|
|
" gewichte = np.maximum(w.value, 0.0)\n",
|
|
" return gewichte / gewichte.sum(), True\n",
|
|
"\n",
|
|
" # -- 5. Backtest -------------------------------------------------------\n",
|
|
" def backtest(self):\n",
|
|
" print(\"Starte Walk-Forward-Backtest ...\")\n",
|
|
" N = len(self.universum)\n",
|
|
" gewichte = np.ones(N) / N\n",
|
|
"\n",
|
|
" werte = [self.startkapital]\n",
|
|
" datumsliste = [self.renditen.index[self.warmup]]\n",
|
|
" gewichtshistorie = []\n",
|
|
" geplant = versucht = ausgefuehrt = 0\n",
|
|
" kosten_gesamt = 0.0\n",
|
|
" turnover_gesamt = 0.0\n",
|
|
"\n",
|
|
" for t in range(self.warmup, len(self.renditen) - 1):\n",
|
|
" heute = self.renditen.index[t]\n",
|
|
" morgen = self.renditen.index[t + 1]\n",
|
|
"\n",
|
|
" ist_termin = heute in self.rebalancing_termine or t == self.warmup\n",
|
|
" if ist_termin:\n",
|
|
" geplant += 1\n",
|
|
" versucht += 1\n",
|
|
" neue_gewichte, erfolg = self.optimiere(heute, gewichte)\n",
|
|
" if erfolg:\n",
|
|
" ausgefuehrt += 1\n",
|
|
" turnover = float(np.abs(neue_gewichte - gewichte).sum())\n",
|
|
" gebuehr = werte[-1] * turnover * self.gebuehrensatz\n",
|
|
" werte[-1] -= gebuehr\n",
|
|
" kosten_gesamt += gebuehr\n",
|
|
" turnover_gesamt += turnover\n",
|
|
" gewichte = neue_gewichte\n",
|
|
" gewichtshistorie.append((heute, gewichte.copy()))\n",
|
|
"\n",
|
|
" # Rendite von MORGEN auf die HEUTE festgelegten Gewichte anwenden\n",
|
|
" tagesrenditen = self.renditen.iloc[t + 1].values\n",
|
|
" portfoliorendite = float(gewichte @ tagesrenditen)\n",
|
|
" werte.append(werte[-1] * (1.0 + portfoliorendite))\n",
|
|
" datumsliste.append(morgen)\n",
|
|
"\n",
|
|
" # Gewichte driften mit den Kursen weiter\n",
|
|
" gedriftet = gewichte * (1.0 + tagesrenditen)\n",
|
|
" gewichte = gedriftet / gedriftet.sum()\n",
|
|
"\n",
|
|
" print(f\" Rebalancing: {geplant} geplant, {ausgefuehrt} ausgefuehrt \"\n",
|
|
" f\"({ausgefuehrt/max(geplant,1)*100:.0f} %)\")\n",
|
|
" if ausgefuehrt < geplant:\n",
|
|
" print(f\" WARNUNG: {geplant - ausgefuehrt} Optimierungen sind \"\n",
|
|
" f\"fehlgeschlagen - Ursache pruefen!\")\n",
|
|
" print(f\" Kumulierter Turnover: {turnover_gesamt*100:.0f} % | \"\n",
|
|
" f\"Transaktionskosten insgesamt: {kosten_gesamt:,.2f} EUR\")\n",
|
|
"\n",
|
|
" ergebnis = pd.DataFrame({\"Portfolio\": werte}, index=datumsliste)\n",
|
|
" benchmark = self.benchmark_kurse.loc[datumsliste]\n",
|
|
" ergebnis[\"Benchmark\"] = benchmark / benchmark.iloc[0] * self.startkapital\n",
|
|
" return ergebnis, gewichtshistorie, kosten_gesamt\n",
|
|
"\n",
|
|
" # -- 6. Auswertung -----------------------------------------------------\n",
|
|
" @staticmethod\n",
|
|
" def kennzahlen(reihe, risikofrei=0.02):\n",
|
|
" renditen = reihe.pct_change().dropna()\n",
|
|
" jahre = (reihe.index[-1] - reihe.index[0]).days / 365.25\n",
|
|
" cagr = (reihe.iloc[-1] / reihe.iloc[0]) ** (1 / jahre) - 1\n",
|
|
" vola = renditen.std() * np.sqrt(HANDELSTAGE)\n",
|
|
" sharpe = (cagr - risikofrei) / vola if vola > 0 else np.nan\n",
|
|
" drawdown = ((reihe - reihe.cummax()) / reihe.cummax()).min()\n",
|
|
" calmar = cagr / abs(drawdown) if drawdown < 0 else np.nan\n",
|
|
" trefferquote = float((renditen > 0).mean())\n",
|
|
" return {\"CAGR\": cagr, \"Volatilitaet\": vola, \"Sharpe\": sharpe,\n",
|
|
" \"Max Drawdown\": drawdown, \"Calmar\": calmar,\n",
|
|
" \"Trefferquote\": trefferquote, \"Jahre\": jahre}\n",
|
|
"\n",
|
|
" def bericht(self, ergebnis, kosten_gesamt):\n",
|
|
" k_port = self.kennzahlen(ergebnis[\"Portfolio\"])\n",
|
|
" k_bench = self.kennzahlen(ergebnis[\"Benchmark\"])\n",
|
|
"\n",
|
|
" print(\"\\n\" + \"=\" * 84)\n",
|
|
" print(\" LEISTUNGSREPORT DER QUANTITATIVEN ENGINE\")\n",
|
|
" print(\"=\" * 84)\n",
|
|
" print(f\"Zeitraum: {ergebnis.index[0]:%Y-%m-%d} bis {ergebnis.index[-1]:%Y-%m-%d} \"\n",
|
|
" f\"({k_port['Jahre']:.1f} Jahre) | Startkapital \"\n",
|
|
" f\"{self.startkapital:,.0f} EUR\\n\")\n",
|
|
"\n",
|
|
" zeilen = [\n",
|
|
" (\"Endkapital\", f\"{ergebnis['Portfolio'].iloc[-1]:,.0f} EUR\",\n",
|
|
" f\"{ergebnis['Benchmark'].iloc[-1]:,.0f} EUR\"),\n",
|
|
" (\"CAGR (Jahresrendite)\", f\"{k_port['CAGR']*100:6.2f} %\",\n",
|
|
" f\"{k_bench['CAGR']*100:6.2f} %\"),\n",
|
|
" (\"Volatilitaet p.a.\", f\"{k_port['Volatilitaet']*100:6.2f} %\",\n",
|
|
" f\"{k_bench['Volatilitaet']*100:6.2f} %\"),\n",
|
|
" (\"Sharpe Ratio (rf=2 %)\", f\"{k_port['Sharpe']:6.2f}\",\n",
|
|
" f\"{k_bench['Sharpe']:6.2f}\"),\n",
|
|
" (\"Maximum Drawdown\", f\"{k_port['Max Drawdown']*100:6.2f} %\",\n",
|
|
" f\"{k_bench['Max Drawdown']*100:6.2f} %\"),\n",
|
|
" (\"Calmar Ratio\", f\"{k_port['Calmar']:6.2f}\", f\"{k_bench['Calmar']:6.2f}\"),\n",
|
|
" (\"Trefferquote (Tage)\", f\"{k_port['Trefferquote']*100:6.1f} %\",\n",
|
|
" f\"{k_bench['Trefferquote']*100:6.1f} %\"),\n",
|
|
" ]\n",
|
|
" print(pd.DataFrame(zeilen, columns=[\"Kennzahl\", \"OR-Strategie\", \"Benchmark\"])\n",
|
|
" .to_string(index=False))\n",
|
|
"\n",
|
|
" anteil = kosten_gesamt / self.startkapital\n",
|
|
" print(f\"\\nTransaktionskosten: {kosten_gesamt:,.0f} EUR \"\n",
|
|
" f\"= {anteil*100:.2f} % des Startkapitals \"\n",
|
|
" f\"= {anteil/k_port['Jahre']*100:.2f} % p.a.\")\n",
|
|
" print(\"Ohne diese Kosten waere die ausgewiesene Rendite entsprechend hoeher -\")\n",
|
|
" print(\"genau deshalb gehoeren sie in den Backtest und nicht daneben.\")\n",
|
|
" print(\"=\" * 84)\n",
|
|
" return k_port, k_bench\n",
|
|
"\n",
|
|
" def zeichne(self, ergebnis):\n",
|
|
" plt.figure(figsize=(11, 6))\n",
|
|
" plt.plot(ergebnis.index, ergebnis[\"Portfolio\"], \"b-\", lw=2.2,\n",
|
|
" label=\"OR-Handelsmaschine\")\n",
|
|
" plt.plot(ergebnis.index, ergebnis[\"Benchmark\"], \"k--\", lw=1.5, alpha=0.75,\n",
|
|
" label=f\"Benchmark ({self.benchmark})\")\n",
|
|
" plt.title(\"Walk-Forward-Backtest: OR-Strategie gegen Benchmark\", fontsize=12)\n",
|
|
" plt.xlabel(\"Datum\")\n",
|
|
" plt.ylabel(\"Depotwert in EUR\")\n",
|
|
" plt.grid(True, linestyle=\":\", alpha=0.6)\n",
|
|
" plt.legend(loc=\"upper left\")\n",
|
|
" plt.tight_layout()\n",
|
|
" ziel = os.path.join(OUTPUT_DIR, \"trading_engine_backtest.png\")\n",
|
|
" plt.savefig(ziel, dpi=150)\n",
|
|
" print(f\"Chart gespeichert unter '{ziel}'\")\n",
|
|
"\n",
|
|
"\n",
|
|
"def lookahead_selbsttest(maschine):\n",
|
|
" \"\"\"\n",
|
|
" Prueft, dass das Signal zum Zeitpunkt t NUR Daten bis t verwendet.\n",
|
|
" Methode: kuenstlich die Zukunft veraendern und pruefen, ob sich das\n",
|
|
" Signal von heute dadurch aendert. Tut es das, liegt Lookahead vor.\n",
|
|
" \"\"\"\n",
|
|
" stichtag = maschine.kurse.index[len(maschine.kurse) // 2]\n",
|
|
" signal_original = maschine.alpha_signal(stichtag)\n",
|
|
"\n",
|
|
" original_kurse = maschine.kurse.copy()\n",
|
|
" zukunft = maschine.kurse.index > stichtag\n",
|
|
" maschine.kurse.loc[zukunft] *= 3.0 # Zukunft massiv veraendern\n",
|
|
" signal_manipuliert = maschine.alpha_signal(stichtag)\n",
|
|
" maschine.kurse = original_kurse # zuruecksetzen\n",
|
|
"\n",
|
|
" identisch = np.allclose(signal_original, signal_manipuliert)\n",
|
|
" print(f\"Lookahead-Selbsttest: Signal bleibt bei manipulierter Zukunft \"\n",
|
|
" f\"{'UNVERAENDERT (gut)' if identisch else 'VERAENDERT - LOOKAHEAD-BIAS!'}\")\n",
|
|
" assert identisch, \"Das Signal verwendet Zukunftsdaten!\"\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" UNIVERSUM = [\"AAPL\", \"MSFT\", \"NVDA\", \"GOOGL\",\n",
|
|
" \"JNJ\", \"UNH\", \"PFE\",\n",
|
|
" \"JPM\", \"BAC\", \"GS\",\n",
|
|
" \"XOM\", \"CVX\",\n",
|
|
" \"PG\", \"KO\", \"COST\"]\n",
|
|
"\n",
|
|
" maschine = QuantitativeHandelsmaschine(UNIVERSUM, benchmark=\"SPY\", jahre=5)\n",
|
|
" maschine.lade_daten()\n",
|
|
" lookahead_selbsttest(maschine)\n",
|
|
" ergebnis, gewichte, kosten = maschine.backtest()\n",
|
|
" maschine.bericht(ergebnis, kosten)\n",
|
|
" maschine.zeichne(ergebnis)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Die fünf Selbsttäuschungen des Backtestens\n",
|
|
"\n",
|
|
"`Backtest_Fallen.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# Backtest_Fallen.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Handelsmaschine: Wie leicht man sich selbst betruegt.\n",
|
|
"\n",
|
|
"Demonstriert an SIMULIERTEN Daten (also ohne jede echte Prognosekraft),\n",
|
|
"wie gross die scheinbare Ueberrendite durch typische Backtest-Fehler wird.\n",
|
|
"Weil die Daten reines Rauschen sind, ist JEDE positive Ueberrendite ein Artefakt.\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"import numpy as np\n",
|
|
"import pandas as pd\n",
|
|
"\n",
|
|
"\n",
|
|
"def erzeuge_zufallsmarkt(tage=1260, titel=15, seed=7):\n",
|
|
" \"\"\"Reines Rauschen: kein Titel hat echte Prognosekraft.\"\"\"\n",
|
|
" rng = np.random.default_rng(seed)\n",
|
|
" renditen = rng.normal(0.0003, 0.015, size=(tage, titel))\n",
|
|
" index = pd.bdate_range(\"2020-01-01\", periods=tage)\n",
|
|
" return pd.DataFrame(renditen, index=index,\n",
|
|
" columns=[f\"T{i:02d}\" for i in range(titel)])\n",
|
|
"\n",
|
|
"\n",
|
|
"def strategie(renditen, lookahead=False, survivorship=False,\n",
|
|
" ohne_kosten=False, gebuehr=0.0015):\n",
|
|
" \"\"\"Einfache Momentum-Strategie mit optional eingebauten Fehlern.\"\"\"\n",
|
|
" fenster = 60\n",
|
|
" kapital = [100.0]\n",
|
|
" gewichte = np.ones(renditen.shape[1]) / renditen.shape[1]\n",
|
|
"\n",
|
|
" daten = renditen\n",
|
|
" if survivorship:\n",
|
|
" # FEHLER: nur die im Nachhinein besten Titel ins Universum lassen\n",
|
|
" beste = renditen.sum().nlargest(renditen.shape[1] // 2).index\n",
|
|
" daten = renditen[beste]\n",
|
|
" gewichte = np.ones(len(beste)) / len(beste)\n",
|
|
"\n",
|
|
" for t in range(fenster, len(daten) - 1):\n",
|
|
" if t % 21 == 0: # monatlich\n",
|
|
" if lookahead:\n",
|
|
" # FEHLER: Signal nutzt die Rendite von MORGEN\n",
|
|
" signal = daten.iloc[t + 1].values\n",
|
|
" else:\n",
|
|
" signal = daten.iloc[t - fenster:t].mean().values\n",
|
|
"\n",
|
|
" neue = (signal > np.median(signal)).astype(float)\n",
|
|
" neue = neue / neue.sum() if neue.sum() > 0 else gewichte\n",
|
|
" if not ohne_kosten:\n",
|
|
" kapital[-1] *= (1 - np.abs(neue - gewichte).sum() * gebuehr)\n",
|
|
" gewichte = neue\n",
|
|
"\n",
|
|
" kapital.append(kapital[-1] * (1 + float(gewichte @ daten.iloc[t + 1].values)))\n",
|
|
" return np.array(kapital)\n",
|
|
"\n",
|
|
"\n",
|
|
"def cagr(verlauf, jahre):\n",
|
|
" return (verlauf[-1] / verlauf[0]) ** (1 / jahre) - 1\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" renditen = erzeuge_zufallsmarkt()\n",
|
|
" jahre = len(renditen) / 252\n",
|
|
" markt = 100 * (1 + renditen.mean(axis=1)).cumprod().values\n",
|
|
"\n",
|
|
" print(\"=\" * 84)\n",
|
|
" print(\" DIE FUENF SELBSTTAEUSCHUNGEN - GEMESSEN AN REINEM RAUSCHEN\")\n",
|
|
" print(\"=\" * 84)\n",
|
|
" print(\"Die Daten sind zufaellig erzeugt. Es gibt KEINE echte Prognosekraft.\")\n",
|
|
" print(\"Jede Ueberrendite unten ist daher ein reines Artefakt.\\n\")\n",
|
|
" print(f\"{'Variante':<44} {'CAGR':>9} {'ggue. Markt':>13}\")\n",
|
|
" print(\"-\" * 84)\n",
|
|
"\n",
|
|
" basis = cagr(markt, jahre)\n",
|
|
" print(f\"{'Markt (Gleichgewichtung, Referenz)':<44} {basis*100:>8.2f} % \"\n",
|
|
" f\"{0.0:>12.2f} %\")\n",
|
|
"\n",
|
|
" varianten = [\n",
|
|
" (\"Ehrlich: Signal aus Vergangenheit, mit Kosten\", {}),\n",
|
|
" (\"FEHLER 1: Lookahead (Signal kennt morgen)\", {\"lookahead\": True}),\n",
|
|
" (\"FEHLER 2: Survivorship (nur Gewinner im Universum)\", {\"survivorship\": True}),\n",
|
|
" (\"FEHLER 3: Transaktionskosten ignoriert\", {\"ohne_kosten\": True}),\n",
|
|
" (\"FEHLER 1+2+3 kombiniert\", {\"lookahead\": True, \"survivorship\": True,\n",
|
|
" \"ohne_kosten\": True}),\n",
|
|
" ]\n",
|
|
" for name, argumente in varianten:\n",
|
|
" verlauf = strategie(renditen, **argumente)\n",
|
|
" wert = cagr(verlauf, jahre)\n",
|
|
" print(f\"{name:<44} {wert*100:>8.2f} % {(wert-basis)*100:>+12.2f} %\")\n",
|
|
"\n",
|
|
" print(\"-\" * 84)\n",
|
|
" print(\"FEHLER 4 (Overfitting) und FEHLER 5 (Data Snooping) lassen sich so nicht\")\n",
|
|
" print(\"zeigen - sie entstehen erst durch WIEDERHOLTES Probieren. Faustregel:\")\n",
|
|
" print(\"Wer 20 Strategien testet, findet auch in reinem Rauschen eine mit\")\n",
|
|
" print(\"Signifikanz auf dem 5-%-Niveau. Genau das ist der Punkt.\")\n",
|
|
" print(\"=\" * 84)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Finde den Denkfehler\n",
|
|
"\n",
|
|
"`Data_Snooping.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# Data_Snooping.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Handelsmaschine: Die Selbsttaeuschung, die man einem Backtest nicht ansieht.\n",
|
|
"\n",
|
|
"Backtest_Fallen.py zeigt drei Fehler, die IM Backtest stecken: Lookahead,\n",
|
|
"Survivorship, vergessene Kosten. Alle drei kann man einem Programm ansehen,\n",
|
|
"wenn man es liest.\n",
|
|
"\n",
|
|
"Die beiden anderen - Overfitting und Data Snooping - entstehen nicht im\n",
|
|
"Backtest, sondern in der ARBEITSWEISE davor. Sie hinterlassen im Code keine\n",
|
|
"Spur. Ein Backtest, der zwanzigmal angepasst wurde, sieht am Ende genauso\n",
|
|
"sauber aus wie einer, der beim ersten Versuch funktioniert hat.\n",
|
|
"\n",
|
|
"Dieses Programm macht den Effekt messbar. Es testet Strategien, die\n",
|
|
"NACHWEISLICH keinerlei Prognosekraft haben (ihre taeglichen Ergebnisse sind\n",
|
|
"Zufallszahlen mit Erwartungswert exakt null), und zeigt, wie gut die jeweils\n",
|
|
"BESTE davon aussieht - allein als Funktion der Anzahl der Versuche.\n",
|
|
"\n",
|
|
"Benoetigt: numpy, scipy\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"from __future__ import annotations\n",
|
|
"\n",
|
|
"import numpy as np\n",
|
|
"from scipy import stats\n",
|
|
"\n",
|
|
"HANDELSTAGE = 1260 # fuenf Jahre\n",
|
|
"TAGESSCHWANKUNG = 0.010 # 1 % taeglich\n",
|
|
"WIEDERHOLUNGEN = 300 # damit die Tabelle nicht vom Zufall abhaengt\n",
|
|
"NIVEAU = 0.05\n",
|
|
"\n",
|
|
"RNG = np.random.default_rng(42)\n",
|
|
"\n",
|
|
"\n",
|
|
"def bewerte(ergebnisse: np.ndarray) -> tuple[np.ndarray, np.ndarray]:\n",
|
|
" \"\"\"Sharpe-Kennzahl und einseitiger p-Wert je Strategie.\n",
|
|
"\n",
|
|
" ergebnisse hat die Form (Strategien, Handelstage). Der p-Wert prueft die\n",
|
|
" Nullhypothese 'mittleres Tagesergebnis ist null' - also genau die Frage,\n",
|
|
" die ein Backtest beantworten soll.\n",
|
|
" \"\"\"\n",
|
|
" mittel = ergebnisse.mean(axis=1)\n",
|
|
" streuung = ergebnisse.std(axis=1, ddof=1)\n",
|
|
" sharpe = mittel / streuung * np.sqrt(252)\n",
|
|
" t_wert = mittel / (streuung / np.sqrt(ergebnisse.shape[1]))\n",
|
|
" return sharpe, stats.t.sf(t_wert, ergebnisse.shape[1] - 1)\n",
|
|
"\n",
|
|
"\n",
|
|
"def probiere(anzahl_strategien: int) -> tuple[float, float, float, float]:\n",
|
|
" \"\"\"Liefert (beste Sharpe, ihr p-Wert, Anteil scheinbar signifikanter,\n",
|
|
" Anteil, der auch die Bonferroni-Huerde nimmt) - gemittelt ueber\n",
|
|
" WIEDERHOLUNGEN unabhaengige Durchgaenge.\"\"\"\n",
|
|
" beste_sharpe, beste_p, anteil_signifikant, ueberlebt_bonferroni = [], [], [], []\n",
|
|
"\n",
|
|
" for _ in range(WIEDERHOLUNGEN):\n",
|
|
" # Der springende Punkt: Erwartungswert exakt 0. Keine dieser\n",
|
|
" # Strategien hat einen echten Vorteil - per Konstruktion.\n",
|
|
" ergebnisse = RNG.normal(0.0, TAGESSCHWANKUNG,\n",
|
|
" (anzahl_strategien, HANDELSTAGE))\n",
|
|
" sharpe, p_werte = bewerte(ergebnisse)\n",
|
|
"\n",
|
|
" beste = int(sharpe.argmax())\n",
|
|
" beste_sharpe.append(float(sharpe[beste]))\n",
|
|
" beste_p.append(float(p_werte[beste]))\n",
|
|
" anteil_signifikant.append(float((p_werte < NIVEAU).mean()))\n",
|
|
" # Bonferroni: Wer n Tests macht, muss die Huerde durch n teilen.\n",
|
|
" ueberlebt_bonferroni.append(\n",
|
|
" float((p_werte < NIVEAU / anzahl_strategien).mean()))\n",
|
|
"\n",
|
|
" return (float(np.mean(beste_sharpe)), float(np.mean(beste_p)),\n",
|
|
" float(np.mean(anteil_signifikant)),\n",
|
|
" float(np.mean(ueberlebt_bonferroni)))\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\" WIE GUT SIEHT DIE BESTE VON N STRATEGIEN AUS,\")\n",
|
|
" print(\" WENN KEINE EINZIGE ETWAS TAUGT?\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(f\"{HANDELSTAGE} Handelstage, taegliche Ergebnisse mit Erwartungswert \"\n",
|
|
" f\"EXAKT null.\")\n",
|
|
" print(f\"Jede Zeile ist der Mittelwert aus {WIEDERHOLUNGEN} unabhaengigen \"\n",
|
|
" f\"Durchgaengen.\\n\")\n",
|
|
"\n",
|
|
" print(f\"{'getestet':>9} {'beste Sharpe':>14} {'ihr p-Wert':>12} \"\n",
|
|
" f\"{'scheinbar sign.':>16}\")\n",
|
|
" print(\"-\" * 78)\n",
|
|
" ergebnisse = {}\n",
|
|
" for anzahl in (1, 10, 100, 1000):\n",
|
|
" sharpe, p_wert, anteil, _ = probiere(anzahl)\n",
|
|
" ergebnisse[anzahl] = (sharpe, p_wert, anteil)\n",
|
|
" print(f\"{anzahl:9d} {sharpe:14.2f} {p_wert:12.4f} {anteil * 100:15.1f} %\")\n",
|
|
" print(\"-\" * 78)\n",
|
|
"\n",
|
|
" sharpe_100, p_100, _ = ergebnisse[100]\n",
|
|
" sharpe_1000, p_1000, _ = ergebnisse[1000]\n",
|
|
"\n",
|
|
" print(\"\\nLesen Sie die dritte Spalte von unten nach oben:\")\n",
|
|
" print(f\" Wer {1000} Varianten durchprobiert, findet eine mit p = {p_1000:.4f} -\")\n",
|
|
" print(\" ein Wert, den jede Zeitschrift als 'hochsignifikant' druckt.\")\n",
|
|
" print(f\" Wer {100} probiert, findet p = {p_100:.4f} und eine Sharpe-Kennzahl\")\n",
|
|
" print(f\" von {sharpe_100:.2f} - damit geht man zum Vorgesetzten.\")\n",
|
|
" print(\" Und keine dieser Strategien hat auch nur den Hauch eines Vorteils.\")\n",
|
|
" print()\n",
|
|
" print(\"Die LETZTE Spalte bleibt dagegen konstant bei rund 5 %. Das ist kein\")\n",
|
|
" print(\"Widerspruch, sondern die Definition: Ein Test zum 5-%-Niveau irrt in\")\n",
|
|
" print(\"5 % der Faelle. Was sich aendert, ist nicht der Anteil, sondern die\")\n",
|
|
" print(\"ANZAHL - und weil man nur die beste Strategie praesentiert, sieht man\")\n",
|
|
" print(\"von allen Fehlversuchen genau einen: den erfolgreichsten.\")\n",
|
|
"\n",
|
|
" print(\"\\n\" + \"=\" * 78)\n",
|
|
" print(\" WAS MAN DAGEGEN TUN KANN\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\"1. ZAEHLEN. Notieren Sie, wie viele Varianten Sie ausprobiert haben -\")\n",
|
|
" print(\" auch die verworfenen, auch die 'nur mal schnell'. Ohne diese Zahl\")\n",
|
|
" print(\" ist kein p-Wert und keine Sharpe-Kennzahl interpretierbar.\")\n",
|
|
" print()\n",
|
|
" print(\"2. KORRIGIEREN. Die Bonferroni-Korrektur teilt die Huerde durch die\")\n",
|
|
" print(\" Zahl der Versuche:\")\n",
|
|
" for anzahl in (10, 100, 1000):\n",
|
|
" print(f\" bei {anzahl:5d} Versuchen: p < {NIVEAU / anzahl:.5f} statt \"\n",
|
|
" f\"p < {NIVEAU:.2f}\")\n",
|
|
" print(\" Sie ist konservativ, aber sie ist ehrlich. Fuer Handelsstrategien\")\n",
|
|
" print(\" gibt es verfeinerte Varianten (Deflated Sharpe Ratio nach Bailey\")\n",
|
|
" print(\" und Lopez de Prado), die dasselbe Prinzip verfolgen.\")\n",
|
|
" print()\n",
|
|
" print(\"3. ZURUECKHALTEN. Legen Sie einen Zeitraum beiseite, BEVOR Sie\")\n",
|
|
" print(\" anfangen, und ruehren Sie ihn bis zum Schluss nicht an. Er ist\")\n",
|
|
" print(\" der einzige Test, den Sie nicht durch Probieren verderben koennen -\")\n",
|
|
" print(\" und er ist genau einmal verwendbar.\")\n",
|
|
" print(\"=\" * 78)"
|
|
]
|
|
}
|
|
],
|
|
"metadata": {
|
|
"kernelspec": {
|
|
"display_name": "Python 3",
|
|
"language": "python",
|
|
"name": "python3"
|
|
},
|
|
"language_info": {
|
|
"name": "python",
|
|
"version": "3.11"
|
|
}
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 5
|
|
}
|