From f81cc63dc7e44e72bba051a8e6bbcd520ffacf1a Mon Sep 17 00:00:00 2001 From: Axel Hocks Date: Fri, 31 Jul 2026 08:30:55 +0200 Subject: [PATCH] analyze_reversal.py: P(Trendumkehr) gemessen - Muenzwurf, kein Modul gebaut User-Frage "koennen wir die Trendumkehr-Wahrscheinlichkeit berechnen?". Aufgebaut exakt wie das erfolgreiche P(break)-Modell: mechanisch scharfes Barriere-Rennen (zuerst -X x ATR GEGEN oder +X x ATR MIT dem Trend?), logistische Regression auf 8 kausalen Merkmalen, Fit auf H1, AUC + Kalibrierung gemessen auf H2. 80k M5-Bars, ~18,3k/18,5k Stichproben. Muenzwurf in JEDER von 5 Ereignis-Definitionen (AUC oos 0,499-0,509, Basisrate ~50 %). Definitionen bewusst variiert (0,5/1,0/1,5/2,0 x ATR, Fenster 60-360 min) - ein Null-Ergebnis ist nur belastbar, wenn es nicht an EINER willkuerlichen Schwelle haengt. Diagnostisch wichtigste Zahl: die IN-SAMPLE-AUC von nur 0,52-0,53. Das Modell erklaert nicht einmal seine eigenen Trainingsdaten - das ist NICHT Overfitting (da waere in-sample 0,70+ und oos braeche ein), sondern das Fehlen jeder Information. Einzelmerkmale allein: alle 0,489-0,515. Kern-Lehre: P(break) fragt nicht "wohin geht der Kurs", sondern "haelt diese konkrete Struktur?" - bedingt auf ein reales Objekt. "Trendumkehr" ist die UNBEDINGTE Richtungsfrage ohne Struktur zum Festmachen. Genau die Klasse, an der das Projekt 18x gescheitert ist. Die kalibrierte umkehr-nahe Groesse existiert bereits: stop_approach (P(break) auf dem Gegen-/Stop-Level, AUC 0,68/0,72). Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 29 +++++ analyze_reversal.py | 298 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 327 insertions(+) create mode 100644 analyze_reversal.py diff --git a/CLAUDE.md b/CLAUDE.md index 37fdefb..44b8a7c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1426,6 +1426,35 @@ dort bereits nachvalidiert, ØR +0,305.) **Eine** Oberfläche: Continuation = breakeven-negativ (−0,011/−0,005, Momentum-Klasse). Damit ist auch die Struktur-Idee empirisch durch — die Kachel bleibt Ist-Zustand-Anzeige (wie TF-Ampel/Squeeze), steuert nichts. +- **P(Trendumkehr) = GEMESSEN, NICHT BERECHENBAR (`analyze_reversal.py`, 2026-07-31, + User-Frage „können wir die Trendumkehr-Wahrscheinlichkeit berechnen?"):** Aufgebaut + exakt wie das erfolgreiche P(break)-Modell — mechanisch scharfes **Barriere-Rennen** + (erreicht der Kurs zuerst −X×ATR GEGEN oder +X×ATR MIT dem Trend?), logistische + Regression auf 8 kausalen Merkmalen (Überdehnung · Regressionswinkel · mom3/mom6 · + Trendstärke · Trendalter · Abstand zum M30-Gegenlevel · Vola-Regime), Fit auf H1, + **AUC + Kalibrierung gemessen auf H2**. 80k M5-Bars, ~18,3k/18,5k Stichproben. + **Ergebnis: Münzwurf in JEDER von 5 Ereignis-Definitionen.** + | Barriere / Fenster | Basisrate | AUC oos | AUC **in-sample** | + |---|---|---|---| + | 0,5×ATR / 60 min | 54,0 % | 0,509 | 0,521 | + | 1,0×ATR / 120 min | 50,8 % | 0,506 | 0,523 | + | 1,0×ATR / 360 min | 51,0 % | 0,507 | 0,523 | + | 1,5×ATR / 180 min | 49,7 % | 0,499 | 0,530 | + | 2,0×ATR / 240 min | 47,8 % | 0,501 | 0,524 | + ⚠ **Die diagnostisch wichtigste Zahl ist die IN-SAMPLE-AUC (0,52–0,53):** das Modell + erklärt nicht einmal seine eigenen Trainingsdaten. Das ist NICHT Overfitting (da wäre + in-sample 0,70+ und oos bräche ein) — es ist schlicht **keine Information vorhanden**. + Einzelmerkmale allein: alle zwischen AUC 0,489 und 0,515. Basisrate ~50 % → der Trend + hat auf diesen Skalen **keine Persistenz**. Kleines kohärentes Muster: kurz (0,5×ATR/ + 60 min) leichte Mean-Reversion (54 %), lang (2,0×ATR/240 min) leichte Trendpersistenz + (47,8 %) — aber 2–4 Pp und **unbedingt** (das WANN ist nicht vorhersagbar). + ⚠ **Warum P(break) funktioniert und das hier nicht — die Kern-Lehre:** P(break) fragt + nicht „wohin geht der Kurs", sondern **„hält diese konkrete Struktur?"** — bedingt auf + ein reales Objekt (Level mit Orders dahinter). „Trendumkehr" ist die UNBEDINGTE + Richtungsfrage, überall im Chart gestellt, ohne Struktur zum Festmachen. Genau die + Klasse, an der das Projekt 18× gescheitert ist. **Die kalibrierte umkehr-nahe Größe + existiert bereits: `stop_approach`** (P(break) auf dem Gegen-/Stop-Level, AUC 0,68/ + 0,72) — strukturgebunden statt frei schwebend. Kein Modul gebaut. - **Wahrscheinlichkeits-Kegel „Erwartete Spanne" (`core/cone.py` + Kachel `#card-cone` + MT5 `CN;…`, 2026-07-31): STREUUNG, NICHT RICHTUNG — und out-of-sample kalibriert.** Der User wollte ein „Kursvorhersagemodul, das mögliche Verläufe in MT5 einzeichnet". diff --git a/analyze_reversal.py b/analyze_reversal.py new file mode 100644 index 0000000..a460281 --- /dev/null +++ b/analyze_reversal.py @@ -0,0 +1,298 @@ +#!/usr/bin/env python3 +"""P(Trendumkehr) — kalibriertes Modell nach dem Vorbild von P(break). + +User-Frage 2026-07-31: „können wir die Trendumkehr-Wahrscheinlichkeit berechnen?" + +WARUM DIESER AUFBAU: Das einzige belegt-prädiktive Modell im Projekt ist P(break) +(AUC 0,71 out-of-sample, Kalibrierung trifft). Es gelang, weil das Ereignis +**mechanisch scharf** definiert war — ein Barriere-Rennen (Level ± 0,5×ATR binnen +12 Bars), nicht „der Kurs bricht aus". Genau daran scheitern Umkehr-Modelle sonst. + +EREIGNIS-DEFINITION (dieselbe Bauart): + Trendrichtung d = sign(EMA12 − EMA50) mit Totband. + Ab Bar i läuft ein Rennen über die nächsten _W Bars: + erreicht der Kurs zuerst −_REV×ATR GEGEN d → UMKEHR (y=1) + erreicht er zuerst +_REV×ATR MIT d → Fortsetzung (y=0) + keins von beidem binnen _W → Fortsetzung (y=0, wie im + P(break)-Training beim Timeout) + Symmetrisch → die Basisrate sagt direkt, ob der Trend überhaupt persistiert. + +MERKMALE (alle kausal aus Bars <= i, ATR-normiert): + stretch Überdehnung in Trendrichtung (C − EMA50)/ATR × d + angle Regressionswinkel × d (dreht der Winkel schon?) + mom3/mom6 kurzes/mittleres Momentum × d + strength Trendstärke |EMA12−EMA50|/ATR + age Bars seit dem letzten EMA-Kreuz (gedeckelt/log-skaliert) + gap Abstand zum nächsten M30-Gegenlevel in ATR + volr ATR relativ zum eigenen Mittel (Vola-Regime) + +METHODE: Logistische Regression (Gradientenabstieg auf standardisierten Merkmalen), +Fit auf H1, **AUC + Kalibrierung gemessen auf H2**. Ein Modell wird nur dann +empfohlen, wenn es out-of-sample trennt UND kalibriert ist — sonst ist es eine +weitere unbelegte Zahl (die Lehre aus `analyze_verdict_calibration.py`, wo conf_pct +in H2 invertiert war). + +⚠ Erwartungsmanagement: Antizyklisch am Extrem ist im Projekt 6× als Münzwurf +gemessen, die Bounce-Anzeige hat ~50 % Richtungstreffer. Ein AUC nahe 0,5 wäre also +die BESTÄTIGUNG des bisherigen Befunds, kein Rückschlag. +""" +import sys +from bisect import bisect_left +from math import exp, log + +import MetaTrader5 as mt5 + +_W = 24 # Renn-Fenster in M5-Bars (2 h) +_REV = 1.0 # Barriere in ×ATR (symmetrisch) +_ATRMIN = 0.06 +_DEAD = 0.15 # Totband für die Trendrichtung (wie _TREND_DEADBAND) +_ANGLE_LR = 20 +_M30 = 6 +_PIV_K = 3 +_PIV_LOOKBACK = 50 +_FEATS = ("stretch", "angle", "mom3", "mom6", "strength", "age", "gap", "volr") + + +def _ema(v, p): + k = 2.0 / (p + 1); o = []; e = v[0] + for i, x in enumerate(v): + e = x if i == 0 else x * k + e * (1 - k) + o.append(e) + return o + + +def _atr_series(H, L, C, p=14): + t = [0.0] + for i in range(1, len(C)): + t.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1]))) + out = [] + for i in range(len(C)): + if not i: + out.append(None); continue + w = t[max(1, i - p + 1):i + 1] + out.append(sum(w) / max(1, len(w))) + return out + + +def _slope(ys): + """Steigung einer LS-Regression über ys (x = 0..n-1).""" + n = len(ys) + if n < 3: + return 0.0 + mx = (n - 1) / 2.0 + my = sum(ys) / n + num = sum((i - mx) * (y - my) for i, y in enumerate(ys)) + den = sum((i - mx) ** 2 for i in range(n)) + return num / den if den else 0.0 + + +def build_levels(H, L): + from collections import deque + nm = len(H) // _M30 + mh = [max(H[i * _M30:(i + 1) * _M30]) for i in range(nm)] + ml = [min(L[i * _M30:(i + 1) * _M30]) for i in range(nm)] + born = [[] for _ in range(nm)] + for p in range(_PIV_K, nm - _PIV_K): + if mh[p] == max(mh[p - _PIV_K:p + _PIV_K + 1]) and p + _PIV_K < nm: + born[p + _PIV_K].append(mh[p]) + if ml[p] == min(ml[p - _PIV_K:p + _PIV_K + 1]) and p + _PIV_K < nm: + born[p + _PIV_K].append(ml[p]) + lv = []; win = deque(); cur = [] + for m in range(nm): + win.append(born[m]); cur.extend(born[m]) + while len(win) > _PIV_LOOKBACK: + for x in win.popleft(): + try: cur.remove(x) + except ValueError: pass + lv.append(sorted(cur)) + return lv + + +def samples(O, H, L, C, AT, EF, ES, LV, a, b, step=2): + """→ [(feature-dict, y)] — kausal: Merkmale aus <=i, Ausgang aus >i.""" + out = [] + atr_mean = None + age = 0; last_d = 0 + # `age` muss lückenlos mitlaufen → über den ganzen Bereich iterieren, + # aber nur jeden `step`-ten Bar als Stichprobe nehmen. + for i in range(max(a, _ANGLE_LR + 60), min(b, len(C) - _W - 1)): + atr = AT[i] + if not atr: + continue + atr = max(atr, _ATRMIN) + diff = EF[i] - ES[i] + d = 1 if diff > _DEAD * atr else -1 if diff < -_DEAD * atr else 0 + if d != last_d and d != 0: + age = 0; last_d = d + else: + age += 1 + if d == 0: + continue # kein Trend → keine „Umkehr" definierbar + if (i - a) % step: + continue + # ── Ausgang: Barriere-Rennen + up = C[i] + _REV * atr + dn = C[i] - _REV * atr + hit_rev = hit_con = None + for j in range(i + 1, min(i + 1 + _W, len(C))): + if d > 0: + if L[j] <= dn: hit_rev = j; break + if H[j] >= up: hit_con = j; break + else: + if H[j] >= up: hit_rev = j; break + if L[j] <= dn: hit_con = j; break + y = 1 if hit_rev is not None else 0 + # ── Merkmale + w = C[max(0, i - _ANGLE_LR):i + 1] + ang = _slope(w) / atr * _ANGLE_LR # normierte Steigung + atr_w = [x for x in AT[max(1, i - 200):i + 1] if x] + atr_mean = (sum(atr_w) / len(atr_w)) if atr_w else atr + lv = LV[min(i // _M30, len(LV) - 1)] + gap = 3.0 + if lv: + k = bisect_left(lv, C[i]) + # Gegenlevel = das, auf das der Trend zuläuft + tgt = None + if d > 0 and k < len(lv): + tgt = lv[k] + elif d < 0 and k > 0: + tgt = lv[k - 1] + if tgt is not None: + gap = min(3.0, abs(tgt - C[i]) / atr) + f = { + "stretch": (C[i] - ES[i]) / atr * d, + "angle": ang * d, + "mom3": (C[i] - C[i - 3]) / atr * d, + "mom6": (C[i] - C[i - 6]) / atr * d, + "strength": abs(diff) / atr, + "age": log(1 + min(age, 300)), + "gap": gap, + "volr": atr / atr_mean if atr_mean else 1.0, + } + out.append((f, y)) + return out + + +def fit(rows, iters=400, lr=0.35): + """Logistische Regression auf standardisierten Merkmalen → (mu, sd, w).""" + n = len(rows) + mu = {}; sd = {} + for k in _FEATS: + xs = [r[0][k] for r in rows] + m = sum(xs) / n + v = sum((x - m) ** 2 for x in xs) / n + mu[k] = m; sd[k] = (v ** 0.5) or 1.0 + X = [[(r[0][k] - mu[k]) / sd[k] for k in _FEATS] for r in rows] + Y = [r[1] for r in rows] + w = [0.0] * (len(_FEATS) + 1) # w[0] = Bias + for _ in range(iters): + g = [0.0] * len(w) + for xi, yi in zip(X, Y): + z = w[0] + sum(wk * xk for wk, xk in zip(w[1:], xi)) + p = 1.0 / (1.0 + exp(-max(-30, min(30, z)))) + e = p - yi + g[0] += e + for t in range(len(xi)): + g[t + 1] += e * xi[t] + for t in range(len(w)): + w[t] -= lr * g[t] / n + return mu, sd, w + + +def predict(f, mu, sd, w): + z = w[0] + sum(w[t + 1] * ((f[k] - mu[k]) / sd[k]) for t, k in enumerate(_FEATS)) + return 1.0 / (1.0 + exp(-max(-30, min(30, z)))) + + +def auc(ps, ys): + pairs = sorted(zip(ps, ys)) + pos = sum(ys); neg = len(ys) - pos + if not pos or not neg: + return 0.5 + rank = 0.0; i = 0 + while i < len(pairs): + j = i + while j + 1 < len(pairs) and pairs[j + 1][0] == pairs[i][0]: + j += 1 + r = (i + j) / 2.0 + 1 + for k in range(i, j + 1): + if pairs[k][1] == 1: + rank += r + i = j + 1 + return (rank - pos * (pos + 1) / 2) / (pos * neg) + + +def main(): + global _REV, _W + n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000 + # Ereignis-Definition variabel — ein Null-Ergebnis ist nur dann belastbar, wenn + # es nicht an EINER willkürlichen Schwelle hängt. + if len(sys.argv) > 2: + _REV = float(sys.argv[2]) + if len(sys.argv) > 3: + _W = int(sys.argv[3]) + mt5.initialize(); sym = None + for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"): + if mt5.symbol_info(c): + sym = c; break + bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n) + mt5.shutdown() + O = [float(b["open"]) for b in bars]; H = [float(b["high"]) for b in bars] + L = [float(b["low"]) for b in bars]; C = [float(b["close"]) for b in bars] + AT = _atr_series(H, L, C); EF = _ema(C, 12); ES = _ema(C, 50) + LV = build_levels(H, L) + N = len(C); mid = N // 2 + + print("=" * 96) + print(f" P(TRENDUMKEHR) — {sym} M5 ({N} Bars)") + print(f" Ereignis: erreicht der Kurs zuerst −{_REV}×ATR GEGEN den Trend (Umkehr)") + print(f" oder +{_REV}×ATR MIT dem Trend, binnen {_W} Bars = {_W*5} min?") + print(" Fit auf H1, AUC + Kalibrierung gemessen auf H2 (out-of-sample).") + print("=" * 96) + + tr = samples(O, H, L, C, AT, EF, ES, LV, 100, mid) + te = samples(O, H, L, C, AT, EF, ES, LV, mid, N) + if not tr or not te: + print("zu wenig Daten"); return + br_tr = sum(y for _, y in tr) / len(tr) + br_te = sum(y for _, y in te) / len(te) + print(f"\n Stichprobe H1 n={len(tr)} Basisrate Umkehr {100*br_tr:.1f}%") + print(f" H2 n={len(te)} Basisrate Umkehr {100*br_te:.1f}%") + print(" (50 % = der Trend persistiert gar nicht; <50 % = Trendfolge trägt)") + + mu, sd, w = fit(tr) + print("\n Gewichte (standardisiert, + = erhöht P(Umkehr)):") + print(f" {'bias':<10}{w[0]:+.3f}") + for t, k in enumerate(_FEATS): + print(f" {k:<10}{w[t+1]:+.3f}") + + ps = [predict(f, mu, sd, w) for f, _ in te] + ys = [y for _, y in te] + a = auc(ps, ys) + print(f"\n AUC out-of-sample (H2): {a:.3f} " + f"(0,50 = Münzwurf · P(break) erreichte 0,71)") + + print("\n Kalibrierung H2 — trifft die vorhergesagte Rate die echte?") + print(f" {'Bucket':<14}{'n':>7}{'vorhergesagt':>14}{'real':>9}") + edges = [0.0, 0.35, 0.45, 0.50, 0.55, 0.65, 1.01] + for lo, hi in zip(edges, edges[1:]): + sel = [(p, y) for p, y in zip(ps, ys) if lo <= p < hi] + if len(sel) < 50: + continue + pm = sum(p for p, _ in sel) / len(sel) + rm = sum(y for _, y in sel) / len(sel) + print(f" {lo:.2f}–{hi:<9.2f}{len(sel):>7}{100*pm:>13.1f}%{100*rm:>8.1f}%") + + # In-sample-Kontrolle: trennt es wenigstens auf den Trainingsdaten? + ps_tr = [predict(f, mu, sd, w) for f, _ in tr] + print(f"\n (Kontrolle) AUC in-sample H1: {auc(ps_tr, [y for _, y in tr]):.3f}") + + print("\n Einzelmerkmale — AUC allein (H2), zeigt wer überhaupt trennt:") + for k in _FEATS: + xs = [f[k] for f, _ in te] + ak = auc(xs, ys) + print(f" {k:<10}{ak:.3f}" + (" <-- traegt" if abs(ak - 0.5) >= 0.03 else "")) + + +if __name__ == "__main__": + main()