Axel HocksandClaude Opus 4.8
75d28827e8
Headless FastAPI-Backend (server.py + core/engine.py) mit Mobile-PWA (web/), Strategie-/Backtest-Suite und Doku. Secrets, DB, Logs und Laufzeit-State sind via .gitignore ausgeschlossen; Config-Vorlage: oil_widget_config.ini.example. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
173 lines
8.2 KiB
Python
173 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""B4-Wochenreport (Strategie-Lebenszyklus, s. CLAUDE.md Track B).
|
||
Hält die LIVE-Zahlen (trades-DB) gegen die BACKTEST-Erwartung und flaggt die bekannten
|
||
Leckagen (Sizing-Asymmetrie, Gegen-Signal-Trades, Winner-Kappen, Regime-Verfall).
|
||
Aufruf: python weekly_review.py (letzte 6 Wochen + diese Woche im Detail)
|
||
Zeitspalten der DB sind LOKALE Epoch (Broker-Offset bereits korrigiert).
|
||
"""
|
||
import sqlite3, datetime as dt
|
||
DB = "oil_widget_history.db"
|
||
|
||
# ── Backtest-Erwartung (R-basiert, exit-sim; s. CLAUDE.md / backtest_*.py) ──
|
||
EXP_WR = 0.68 # Treffer ~68–71 % (Reversal/Trend, SL2+Trail)
|
||
EXP_PF = 1.30 # Profit-Faktor in R
|
||
EXP_VERH = 0.55 # Ø-Gewinn/Ø-Verlust in R (WR hoch → Wins kleiner, trotzdem PF>1)
|
||
|
||
def net(r): return (r["pnl"] or 0.0) + (r["commission"] or 0.0)
|
||
def d_of(r): return "LONG" if r["direction"] in ("BUY","LONG","buy") else "SHORT"
|
||
def align(r):
|
||
rs = (r["rec_signal"] or "WARTEN").upper()
|
||
if rs in ("LONG","SHORT"): return "mit" if rs == d_of(r) else "gegen"
|
||
return "ohne"
|
||
|
||
def agg(rows):
|
||
v = [net(r) for r in rows]
|
||
if not v: return None
|
||
n = len(v); wins = [x for x in v if x > 0]; loss = [x for x in v if x < 0]
|
||
w = len(wins); g = sum(wins); ls = -sum(loss)
|
||
aw = g / w if w else 0.0; al = ls / len(loss) if loss else 0.0
|
||
return dict(n=n, wr=w/n, pf=(g/ls if ls > 0 else 99.9), sum=sum(v),
|
||
aw=aw, al=al, verh=(aw/al if al > 0 else 99.9), worst=min(v))
|
||
|
||
def line(lbl, s):
|
||
if not s: return f" {lbl:<14} —"
|
||
return (f" {lbl:<14} n={s['n']:>3} WR={100*s['wr']:>3.0f}% PF={s['pf']:>4.2f} "
|
||
f"Verh={s['verh']:>4.2f} Øgew={s['aw']:>+5.2f} Øverl={s['al']:>+6.2f} "
|
||
f"Σ={s['sum']:>+8.2f} Worst={s['worst']:>+7.2f}")
|
||
|
||
def _squeeze_section(m5) -> list:
|
||
"""B4-Monitor für den Volatilitäts-Squeeze-Breakout: rechnet die Regel mit den
|
||
LIVE-Params auf den letzten ~30 Tagen M5 NEU (der Setup ist mechanisch → seine
|
||
Live-Performance = Neuberechnung auf frischen Bars, keine Trade-Attribution nötig)
|
||
und hält ØR/PF gegen die Backtest-Erwartung (+0,14…+0,23). m5 = {H,L,C,SP} oder None."""
|
||
if not m5 or len(m5.get("C") or []) < 300:
|
||
return ["", "SQUEEZE-BREAKOUT (B4): keine M5-Daten — übersprungen."]
|
||
try:
|
||
import backtest_breakout_squeeze as BQ
|
||
from core.wave_rec import _SQ_N, _SQ_MULT, _SQ_K
|
||
BQ._N = _SQ_N; BQ._K = _SQ_K # live-Params in den Scanner
|
||
H, L, C, SP = m5["H"], m5["L"], m5["C"], m5["SP"]
|
||
A = BQ._atr_series(H, L, C)
|
||
sq, _ = BQ.scan(H, L, C, A, SP, 1.0, 0, len(C), _SQ_MULT) # SP schon in Preis
|
||
out = ["", "-" * 92,
|
||
f"SQUEEZE-BREAKOUT (B4-Monitor · ~30 Tage M5 neu gerechnet · live-Params "
|
||
f"Box={_SQ_N}/k={_SQ_K}/≤{_SQ_MULT}×ATR):"]
|
||
n = len(sq)
|
||
if n == 0:
|
||
out.append(" keine Squeeze-Ausbrüche im Fenster — weiter beobachten.")
|
||
return out
|
||
w = sum(1 for x in sq if x > 0); s = sum(sq)
|
||
up = sum(x for x in sq if x > 0); dn = -sum(x for x in sq if x < 0)
|
||
pf = up / dn if dn > 0 else 99.9; oR = s / n
|
||
out.append(f" n={n} Treffer={100*w/n:.0f}% ØR={oR:+.3f} PF={pf:.2f} ΣR={s:+.0f}"
|
||
f" (Erwartung ØR +0,14…+0,23 · PF>1)")
|
||
if oR <= 0 or pf < 1.0:
|
||
out.append(" (!) ØR≤0/PF<1 im Fenster → Edge driftet? B5 prüfen (Regimewechsel). "
|
||
"⚠ kleine Stichprobe, nicht überreagieren.")
|
||
elif n < 6:
|
||
out.append(" ⚠ noch wenige Trades — nicht aussagekräftig, weiter beobachten.")
|
||
else:
|
||
out.append(" [OK] im Rahmen der Erwartung — Setup trägt live.")
|
||
return out
|
||
except Exception as e:
|
||
return ["", f"SQUEEZE-BREAKOUT (B4): Fehler ({e})"]
|
||
|
||
|
||
def _fetch_m5(days=30):
|
||
"""M5-Bars für den Squeeze-Monitor (nur STANDALONE — schließt MT5 nur, wenn WIR
|
||
es initialisiert haben; die Server-Instanz reicht die Bars direkt in build_report)."""
|
||
try:
|
||
import MetaTrader5 as mt5
|
||
pre = mt5.terminal_info() is not None # schon vom Server initialisiert?
|
||
if not pre and not mt5.initialize():
|
||
return None
|
||
try:
|
||
sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD")
|
||
if mt5.symbol_info(c)), None)
|
||
if not sym:
|
||
return None
|
||
bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, days * 288 + 30)
|
||
point = mt5.symbol_info(sym).point
|
||
finally:
|
||
if not pre:
|
||
mt5.shutdown()
|
||
if bars is None:
|
||
return None
|
||
return {"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]}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def build_report(db=DB, m5=None) -> str:
|
||
"""Baut den B4-Wochenreport als reinen Text (für Konsole, Mail-<pre>, Telegram).
|
||
m5 = vorab geholte M5-Bars für den Squeeze-Monitor (Server reicht sie unter
|
||
mt5_lock durch); None → Squeeze-Abschnitt versucht selbst zu laden (standalone)."""
|
||
if m5 is None:
|
||
m5 = _fetch_m5()
|
||
out = []; P = out.append
|
||
c = sqlite3.connect(db); c.row_factory = sqlite3.Row
|
||
rows = [r for r in c.execute(
|
||
"SELECT * FROM trades WHERE exit_time IS NOT NULL ORDER BY entry_time")]
|
||
now = dt.datetime.now()
|
||
def wkey(ts): return dt.datetime.fromtimestamp(ts).isocalendar()[:2] # (jahr,kw)
|
||
thisk = now.isocalendar()[:2]
|
||
|
||
P("=" * 92)
|
||
P(f" B4-WOCHENREPORT · {now:%d.%m.%Y %H:%M} · Live (EUR netto) vs Backtest-Erwartung")
|
||
P(f" Erwartung: WR~{100*EXP_WR:.0f}% · PF~{EXP_PF:.2f} · Verhältnis~{EXP_VERH:.2f} (R-basiert)")
|
||
P("=" * 92)
|
||
|
||
from collections import OrderedDict
|
||
byw = OrderedDict()
|
||
for r in rows: byw.setdefault(wkey(r["entry_time"]), []).append(r)
|
||
P("\nWochenverlauf (Verfall-Check):")
|
||
for (yr, kw), rs in list(byw.items())[-6:]:
|
||
tag = " <- diese Woche" if (yr, kw) == thisk else ""
|
||
P(line(f"KW{kw}/{yr%100}", agg(rs)) + tag)
|
||
|
||
tw = byw.get(thisk, [])
|
||
P(f"\nDiese Woche — nach Empfehlung (n={len(tw)}):")
|
||
for k in ("mit", "gegen", "ohne"):
|
||
P(line(k, agg([r for r in tw if align(r) == k])))
|
||
|
||
P("\nGesamt (alle Trades):")
|
||
P(line("alle", agg(rows)))
|
||
|
||
recent = [r for (yr, kw), rs in list(byw.items())[-4:] for r in rs]
|
||
s = agg(recent)
|
||
P("\n" + "-" * 92)
|
||
P(f"BEWERTUNG (letzte 4 Wochen, n={s['n'] if s else 0}):")
|
||
if s:
|
||
def mark(ok): return "[OK]" if ok else "[!]"
|
||
P(f" {mark(s['wr']>=EXP_WR-0.08)} Trefferquote {100*s['wr']:.0f}% (erwartet ~{100*EXP_WR:.0f}%)")
|
||
P(f" {mark(s['pf']>=1.0)} Profit-Faktor {s['pf']:.2f} (erwartet ~{EXP_PF:.2f}; <1 = verlustbringend)")
|
||
P(f" {mark(s['verh']>=EXP_VERH-0.10)} Verhältnis {s['verh']:.2f} (erwartet ~{EXP_VERH:.2f})")
|
||
flags = []
|
||
gg = agg([r for r in recent if align(r) == "gegen"])
|
||
share_gg = (gg["n"]/s["n"]) if gg else 0
|
||
if share_gg > 0.12:
|
||
flags.append(f"(!) {100*share_gg:.0f}% GEGEN-Signal-Trades (dort WR ~14 % — meiden!)")
|
||
if s["verh"] < EXP_VERH - 0.10:
|
||
flags.append("(!) Verhältnis unter Erwartung → Gewinner zu klein (Kappen) oder Verlierer zu groß (Sizing 90 %)")
|
||
if s["wr"] < EXP_WR - 0.10:
|
||
flags.append("(!) Treffer unter Erwartung → Regime-Verfall ODER zu viele Nicht-Setup-Trades")
|
||
if s["pf"] < 1.0:
|
||
flags.append("(!) PF<1 = Verlustphase → B5 prüfen: Regimewechsel? Sizing? Disziplin?")
|
||
if s["worst"] < -25:
|
||
flags.append(f"(!) Einzelverlust {s['worst']:+.0f} € — 90 %-Margin-Sizing (Memory: risk_pct=1.5)")
|
||
P("\n " + ("\n ".join(flags) if flags else "[OK] keine Leckage-Flags — im Rahmen der Erwartung."))
|
||
P("-" * 92)
|
||
P("Nächster Schritt bei [!]: B5 (behalten/verwerfen) — driftet der Live-Edge weg,")
|
||
P("erst Ursache trennen (Regime vs. Disziplin vs. Sizing), nicht am Signal drehen.")
|
||
for ln in _squeeze_section(m5):
|
||
P(ln)
|
||
return "\n".join(out)
|
||
|
||
def main():
|
||
print(build_report())
|
||
|
||
if __name__ == "__main__":
|
||
main()
|