Ursache der P(break)-Live-Degradation gefunden: Merkmals-Diskrepanz
backtest_pbreak_touches.py (M30-Level wie live, Touch-Zahl als 5. Merkmal) plus Rekonstruktion der Live-Merkmale aus candles_m1. Beweiskette: 1) Der Backtest reproduziert die 0,71 (Fit H1 -> AUC oos H2 0,715). Modell und Pipeline sind in Ordnung. 2) Populations-Hypothese widerlegt: live ist ausgerechnet die 1. Beruehrung (die trainierte Situation) mit AUC 0,445 die SCHLECHTESTE Gruppe, die Chop-Faelle (5.+) mit 0,577 die beste. Der Versatz "Modell 23-25 %, real 40-42 %" ist in JEDER Teilmenge gleich gross. 3) Merkmale rekonstruiert (n=2275): mom3 live 0,149 vs Training 1,4523 (-1,303), mom6 0,181 vs 1,6565 (-1,476). Daraus z-Versatz -1,05 -> aus 39,4 % Basisrate werden ~19 %, beobachtet 23 %. Erklaert praktisch die gesamte Fehlkalibrierung. MECHANISMUS: Training fixiert das Level beim Entry (>=0,3xATR entfernt) und wartet auf den ersten Touch -> mom3 ~1,45xATR = ein Schub. Live waehlt _draw_levels das naechstgelegene Level jede Sekunde NEU -> der Kurs steht oft neben einem Level, das gerade erst zum naechsten wurde, ohne Anlauf -> mom3 ~0,15. Gleicher Code, andere Situation. Folge: das Modell liegt live immer unter der Schwelle 0,55 -> das Gate war seit Inbetriebnahme faktisch nie aktiv (91 % aller Beruehrungen geschlossen = der pauschale S/R-Close, der 2x verworfen wurde). Zwei Reparaturwege (beide ungemessen): (a) Ziel-Level beim Oeffnen fixieren statt laufend neu waehlen, oder (b) Modell auf live-spiegelnder Stichprobe neu trainieren. TOUCH-ZAHL (User-Idee): im Backtest AUC 0,715 -> 0,725, Gewicht +0,176 fuer die 2.-4. Beruehrung (Vorzeichen bestaetigt), aber R-Ertrag nicht robust (H1 +131, H2 -43). Im Trainings-Datensatz 76 % Erst-Beruehrungen und KEINE 5+-Faelle - das live beobachtete umgekehrte U ist dort nicht abbildbar. Kleine Verbesserung, nicht der Hebel. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Axel Hocks
co-authored by
Claude Opus 5
Claude Opus 5
parent
92995b08d7
commit
b137967a68
@@ -0,0 +1,284 @@
|
||||
#!/usr/bin/env python3
|
||||
"""P(break) mit TOUCH-ZAHL als 5. Merkmal — trägt sie über 2 echte Halbjahre?
|
||||
|
||||
User-Idee 2026-07-31: „berücksichtige, ob der Kurs das erste, zweite oder dritte Mal
|
||||
in einer Zeitspanne den S/R trifft — je öfter, desto wahrscheinlicher der Durchbruch."
|
||||
|
||||
AUF LIVE-DATEN BEREITS BESTÄTIGT (`analyze_pbreak_live.py`, 2275 Vorhersagen, beide
|
||||
Hälften): Bruchrate nach n-ter Berührung im 30-min-Fenster
|
||||
1. 38,3 / 41,7 % 3. 43,2 / 45,1 %
|
||||
2. 40,8 / 45,6 % 4. 45,1 / 47,4 % ← Maximum, +6/+4 Pp über Basis
|
||||
5.+ 37,8 / 41,8 % ← FÄLLT ZURÜCK unter die Basis (größte Gruppe, n≈1400)
|
||||
Also ein **umgekehrtes U**: 2–4 Berührungen machen das Level mürbe, ab 5 in 30 min
|
||||
ist es eine Range-Begrenzung und HÄLT. Das aktuelle Modell sieht davon nichts (sagt
|
||||
über alle Berührungszahlen konstant ~23 % voraus).
|
||||
|
||||
⚠ Die Live-Bestätigung ist schwach (beide „Hälften" nur 4 Tage auseinander, Buckets
|
||||
n=82–139) — DESHALB dieser Test über 80k M5-Bars / 2 echte Halbjahre.
|
||||
⚠ `backtest_srbreak.py` fand die Touch-Zahl früher „nicht robust" — allerdings ohne
|
||||
Fenster-Definition und ohne die 5+-Umkehr.
|
||||
|
||||
WAS HIER ANDERS IST ALS IN `backtest_srclose_prob.py`:
|
||||
* **M30-Level** (wie live seit 2026-07-30), nicht die alten M5-Pivots.
|
||||
* Direkter Vergleich **4 Merkmale (Live-Modell) vs. 5 Merkmale (+Touch-Zahl)** —
|
||||
ohne diesen Kontrast wäre nicht unterscheidbar, ob eine bessere AUC vom neuen
|
||||
Merkmal kommt oder von der Neuanpassung.
|
||||
* Urteil: das 5-Merkmal-Modell muss in BEIDEN Hälften besser sein (AUC UND R-Ertrag).
|
||||
|
||||
Die Touch-Zahl wird als **umgekehrtes U** kodiert (zwei Dummies statt eines linearen
|
||||
Terms): `t_mid` = 1 bei 2–4 Berührungen, `t_many` = 1 ab 5. Ein linearer Term würde
|
||||
die gemessene Umkehr systematisch verfehlen.
|
||||
"""
|
||||
import sys
|
||||
from collections import deque
|
||||
|
||||
import numpy as np
|
||||
import MetaTrader5 as mt5
|
||||
|
||||
from core.analysis import calc_trend_angle
|
||||
from core.wave_rec import (_EMA_FAST, _EMA_SLOW, _N_BARS, _ANGLE_LR, _ANGLE_DEAD,
|
||||
_REVERSAL_STRETCH, _STRETCH_MAX)
|
||||
|
||||
_MAXH = 200; _ATRMIN = 0.12; _SL_ATR = 2.0; _TRAILON = 0.3; _MULT = 1.5; _BE = 1.3
|
||||
_LOCK_START = 3.5; _LOCK_SCALE = 0.6; _LOCK_MIN = 1.2; _TP_INIT = 3.5
|
||||
_PIV_K = 3; _BRK_W = 12; _BRK_ATR = 0.5
|
||||
_M30 = 6; _PIV_LOOKBACK = 50 # M30-Level wie live
|
||||
_ZONE = 0.15 # „am Level" = |Kurs−Level| <= X×ATR (wie live)
|
||||
_TWIN = 6 # Touch-Fenster in M5-Bars (6 = 30 min, wie live)
|
||||
|
||||
|
||||
def _ema_series(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 build_levels(H, L):
|
||||
"""M30-Pivots k=3, kausal (Pivot erst 3 M30-Bars später bekannt) — wie live."""
|
||||
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 touch_count(H, L, level, atr, jt, win):
|
||||
"""Wievielte Berührung dieses Levels innerhalb der letzten `win` Bars?
|
||||
|
||||
Gezählt werden EPISODEN (Eintritte in die Zone |Kurs−Level| <= _ZONE×ATR nach
|
||||
einem Bar außerhalb) — nicht Bars, sonst zählt ein langes Verweilen als viele
|
||||
Berührungen. Die aktuelle Berührung bei jt zählt mit, Ergebnis also >= 1.
|
||||
"""
|
||||
z = _ZONE * atr
|
||||
n = 0; inside_prev = False
|
||||
for j in range(max(0, jt - win), jt + 1):
|
||||
inside = (L[j] <= level + z) and (H[j] >= level - z)
|
||||
if inside and not inside_prev:
|
||||
n += 1
|
||||
inside_prev = inside
|
||||
return max(1, n)
|
||||
|
||||
|
||||
def sim_run(entry, d, atr, H, L, C, j0):
|
||||
sl = entry - d * _SL_ATR * atr; tp = entry + d * _TP_INIT * atr
|
||||
hw = entry; rank = 0
|
||||
end = min(j0 + _MAXH, len(C) - 1)
|
||||
for j in range(j0, end + 1):
|
||||
hi, lo = H[j], L[j]
|
||||
if (lo <= sl) if d > 0 else (hi >= sl): return (sl - entry) * d / atr
|
||||
if (hi >= tp) if d > 0 else (lo <= tp): return (tp - entry) * d / atr
|
||||
hw = max(hw, hi) if d > 0 else min(hw, lo)
|
||||
profit = (hw - entry) * d
|
||||
ph = 0 if profit < _TRAILON * atr else (1 if profit < _LOCK_START * atr else 2)
|
||||
if ph < rank: ph = rank
|
||||
rank = ph
|
||||
if ph == 1:
|
||||
cand = hw - d * _MULT * atr
|
||||
if profit >= _BE * atr: cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||||
sl = max(sl, cand) if d > 0 else min(sl, cand)
|
||||
elif ph == 2:
|
||||
tm = max(_LOCK_MIN, _MULT * _LOCK_SCALE); cand = hw - d * tm * atr
|
||||
cand = max(cand, entry) if d > 0 else min(cand, entry)
|
||||
sl = max(sl, cand) if d > 0 else min(sl, cand)
|
||||
return (C[end] - entry) * d / atr
|
||||
|
||||
|
||||
def collect(a, b, H, L, C, EF, ES, AT, SP, LV):
|
||||
TH = _REVERSAL_STRETCH; out = []
|
||||
for i in range(max(a, _N_BARS, 320), min(b, len(C) - _MAXH - 1)):
|
||||
atr = AT[i]
|
||||
if not atr or atr <= 0: continue
|
||||
atr = max(atr, _ATRMIN); es = ES[i]; ef = EF[i]
|
||||
stretch = (C[i] - es) / atr
|
||||
ang = calc_trend_angle(C[i - _ANGLE_LR - 2:i], _ANGLE_LR); ad = ang - 90.0
|
||||
d = 0
|
||||
if stretch <= -TH and ad >= _ANGLE_DEAD: d = 1
|
||||
elif stretch >= TH and ad <= -_ANGLE_DEAD: d = -1
|
||||
elif abs(stretch) < _STRETCH_MAX: d = 1 if ef > es else -1 if ef < es else 0
|
||||
if not d: continue
|
||||
entry = C[i]; cost = (SP[i] if SP[i] > 0 else 0.0225) / atr
|
||||
R_run = sim_run(entry, d, atr, H, L, C, i + 1) - cost
|
||||
rec = {"R_run": R_run, "touched": False}
|
||||
lv = LV[min(i // _M30, len(LV) - 1)]
|
||||
level = None
|
||||
if lv:
|
||||
if d > 0:
|
||||
cands = [p for p in lv if p > entry + 0.3 * atr]
|
||||
level = min(cands) if cands else None
|
||||
else:
|
||||
cands = [p for p in lv if p < entry - 0.3 * atr]
|
||||
level = max(cands) if cands else None
|
||||
if level is not None:
|
||||
jt = None
|
||||
for j in range(i + 1, min(i + _MAXH, len(C) - _BRK_W)):
|
||||
if ((H[j] >= level) if d > 0 else (L[j] <= level)): jt = j; break
|
||||
if ((entry - L[j]) if d > 0 else (H[j] - entry)) >= 2.0 * atr: break
|
||||
if jt is not None:
|
||||
up = level + d * _BRK_ATR * atr; dn = level - d * _BRK_ATR * atr
|
||||
brk = False
|
||||
for j in range(jt, min(jt + _BRK_W, len(H))):
|
||||
if (H[j] >= up) if d > 0 else (L[j] <= up): brk = True; break
|
||||
if (L[j] <= dn) if d > 0 else (H[j] >= dn): brk = False; break
|
||||
tc = touch_count(H, L, level, atr, jt, _TWIN)
|
||||
base4 = [(C[jt] - C[max(0, jt - 6)]) * d / atr,
|
||||
(C[jt] - C[max(0, jt - 3)]) * d / atr,
|
||||
1.0 if (EF[jt] - ES[jt]) * d > 0 else 0.0,
|
||||
abs(level - entry) / atr]
|
||||
rec.update(touched=True, brk=1.0 if brk else 0.0, tc=tc,
|
||||
R_close=(level - entry) * d / atr - cost,
|
||||
feat4=base4,
|
||||
# umgekehrtes U: 2–4 Berührungen vs. 5+
|
||||
feat5=base4 + [1.0 if 2 <= tc <= 4 else 0.0,
|
||||
1.0 if tc >= 5 else 0.0])
|
||||
out.append(rec)
|
||||
return out
|
||||
|
||||
|
||||
def fit_logreg(X, y, iters=3000, lr=0.3):
|
||||
n, m = X.shape
|
||||
Xb = np.hstack([np.ones((n, 1)), X]); w = np.zeros(m + 1)
|
||||
for _ in range(iters):
|
||||
p = 1 / (1 + np.exp(-(Xb @ w)))
|
||||
w -= lr * (Xb.T @ (p - y)) / n
|
||||
return w
|
||||
|
||||
|
||||
def auc(ps, ys):
|
||||
order = np.argsort(ps); ys = np.asarray(ys)[order]
|
||||
pos = ys.sum(); neg = len(ys) - pos
|
||||
if not pos or not neg: return 0.5
|
||||
ranks = np.arange(1, len(ys) + 1)
|
||||
return float((ranks[ys == 1].sum() - pos * (pos + 1) / 2) / (pos * neg))
|
||||
|
||||
|
||||
def main():
|
||||
global _TWIN
|
||||
n = int(sys.argv[1]) if len(sys.argv) > 1 else 80000
|
||||
if len(sys.argv) > 2:
|
||||
_TWIN = int(sys.argv[2])
|
||||
mt5.initialize(); sym = None
|
||||
for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD"):
|
||||
if mt5.symbol_info(c): sym = c; break
|
||||
si = mt5.symbol_info(sym); point = si.point
|
||||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, n)
|
||||
mt5.shutdown()
|
||||
H = [float(b["high"]) for b in bars]; L = [float(b["low"]) for b in bars]
|
||||
C = [float(b["close"]) for b in bars]; SP = [float(b["spread"]) * point for b in bars]
|
||||
EF = _ema_series(C, _EMA_FAST); ES = _ema_series(C, _EMA_SLOW); AT = _atr_series(H, L, C)
|
||||
LV = build_levels(H, L)
|
||||
mid = len(C) // 2
|
||||
ev1 = collect(_N_BARS, mid, H, L, C, EF, ES, AT, SP, LV)
|
||||
ev2 = collect(mid, len(C), H, L, C, EF, ES, AT, SP, LV)
|
||||
t1 = [e for e in ev1 if e["touched"]]; t2 = [e for e in ev2 if e["touched"]]
|
||||
|
||||
print("=" * 98)
|
||||
print(f" P(break) + TOUCH-ZAHL — {sym} M5 ({len(C)} Bars), M30-Level wie live")
|
||||
print(f" Touch-Fenster {_TWIN} Bars = {_TWIN*5} min · Zone {_ZONE}×ATR")
|
||||
print(f" Fit auf H1 ({len(t1)} Touches), geprüft auf H2 ({len(t2)} Touches)")
|
||||
print("=" * 98)
|
||||
|
||||
y1 = np.array([e["brk"] for e in t1]); y2 = np.array([e["brk"] for e in t2])
|
||||
print(f"\n Basisrate Durchbruch: H1 {100*y1.mean():.1f} % H2 {100*y2.mean():.1f} %")
|
||||
|
||||
# ── Rohbefund: Bruchrate je Berührungszahl, BEIDE Hälften
|
||||
print(f"\n Rohbefund — Bruchrate je Berührungszahl (User-Hypothese):")
|
||||
print(f" {'Berührung':<12}{'H1 n':>7}{'H1 Bruch':>11}{'H2 n':>7}{'H2 Bruch':>11}")
|
||||
for b in (1, 2, 3, 4, 5):
|
||||
lab = f"{b}." if b < 5 else "5.+"
|
||||
a1 = [e["brk"] for e in t1 if (e["tc"] == b if b < 5 else e["tc"] >= 5)]
|
||||
a2 = [e["brk"] for e in t2 if (e["tc"] == b if b < 5 else e["tc"] >= 5)]
|
||||
if len(a1) < 20 or len(a2) < 20:
|
||||
print(f" {lab:<12}{len(a1):>7}{'-':>11}{len(a2):>7}{'-':>11}"); continue
|
||||
print(f" {lab:<12}{len(a1):>7}{100*np.mean(a1):>10.1f}%"
|
||||
f"{len(a2):>7}{100*np.mean(a2):>10.1f}%")
|
||||
|
||||
# ── Modellvergleich 4 vs 5 Merkmale
|
||||
results = {}
|
||||
for tag, key in (("4 Merkmale (Live-Modell)", "feat4"),
|
||||
("5 Merkmale (+Touch-Zahl)", "feat5")):
|
||||
X1 = np.array([e[key] for e in t1]); X2 = np.array([e[key] for e in t2])
|
||||
mu = X1.mean(0); sd = X1.std(0) + 1e-9
|
||||
w = fit_logreg((X1 - mu) / sd, y1)
|
||||
p2 = 1 / (1 + np.exp(-(np.hstack([np.ones((len(X2), 1)), (X2 - mu) / sd]) @ w)))
|
||||
p1 = 1 / (1 + np.exp(-(np.hstack([np.ones((len(X1), 1)), (X1 - mu) / sd]) @ w)))
|
||||
results[key] = (mu, sd, w, p1, p2)
|
||||
print(f"\n {tag}")
|
||||
print(f" AUC in-sample H1 {auc(p1, y1):.3f} out-of-sample H2 {auc(p2, y2):.3f}")
|
||||
if key == "feat5":
|
||||
print(f" Gewichte Touch: 2–4 Ber. {w[5]:+.3f} · 5+ Ber. {w[6]:+.3f}")
|
||||
print(f" Kalibrierung H2: {'Bucket':<12}{'n':>6}{'vorherg.':>10}{'real':>8}")
|
||||
for lo, hi in ((0, .2), (.2, .35), (.35, .5), (.5, .65), (.65, 1.01)):
|
||||
sel = [(p, y) for p, y in zip(p2, y2) if lo <= p < hi]
|
||||
if len(sel) < 40: continue
|
||||
print(f" {lo:.2f}–{hi:<7.2f}{len(sel):>6}"
|
||||
f"{100*np.mean([p for p, _ in sel]):>9.1f}%"
|
||||
f"{100*np.mean([y for _, y in sel]):>7.1f}%")
|
||||
|
||||
# ── R-Ertrag der Close-Regel, beide Modelle, beide Hälften
|
||||
print(f"\n{'='*98}\n R-ERTRAG der Close-Regel (Echtkosten, Baseline = immer laufen lassen)\n{'='*98}")
|
||||
for lbl, ev, ts, idx in (("H1", ev1, t1, 3), ("H2", ev2, t2, 4)):
|
||||
base = sum(e["R_run"] for e in ev)
|
||||
untouched = sum(e["R_run"] for e in ev if not e["touched"])
|
||||
print(f"\n {lbl}: {len(ev)} Wellen · {len(ts)} Touches · Baseline ΣR={base:+.0f}")
|
||||
print(f" {'Schwelle':<10}{'4 Merkmale':>14}{'5 Merkmale':>14}{'Differenz':>12}")
|
||||
for X in (0.35, 0.45, 0.55, 0.60):
|
||||
row = []
|
||||
for key in ("feat4", "feat5"):
|
||||
mu, sd, w, p1, p2 = results[key]
|
||||
ps = p1 if lbl == "H1" else p2
|
||||
tot = sum((e["R_close"] if p < X else e["R_run"])
|
||||
for e, p in zip(ts, ps)) + untouched
|
||||
row.append(tot)
|
||||
print(f" P≥{X:.2f} {row[0]:>+14.0f}{row[1]:>+14.0f}{row[1]-row[0]:>+12.0f}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user