AH-Oil-Trader/core/squeeze_scan.py

"""Box-Ausbruchs-Scan — die geteilte Mechanik hinter dem Squeeze-Setup.

⚠⚠ WARUM ES DIESES MODUL GIBT (2026-08-07). `core/engine.py` und
`weekly_review.py` importierten bis dahin **`backtest_breakout_squeeze`**, also
LIVE-Code, der von einem Backtest-Skript abhing. Zwei Folgen:
  1. Die Abhaengigkeitsrichtung war verkehrt. „Backtests sind nur Dokumentation"
     stimmte nicht — ein Fehler dort konnte den laufenden Bot treffen.
  2. Schlimmer: die Aufrufer haben **Modul-Globale des Backtests mutiert**
     (`BQ._N = ...; BQ._K = ...`), um Live-Parameter einzuschleusen. Das wirkt
     prozessweit und damit auf jeden anderen Nutzer desselben Moduls im selben
     Prozess.
Hier stehen die Parameter deshalb in der Signatur, nicht in Modul-Globalen.

⚠⚠ DER EXIT IST EIN PARAMETER, UND DAS IST ABSICHT. `exit_legacy` unten ist eine
BITGENAUE Kopie des Exits aus `backtest_breakout_squeeze.sim` — inklusive seiner
Abweichungen vom kanonischen Live-Exit (`core/exit_model.LIVE`):
    Trail 1,5 statt 1,0 · KEINE Lock-Phase · KEIN Time-Stop · KEIN TP ·
    max_hold 288 statt 200
Damit bleiben die dokumentierten Zahlen des Backtests UND das Verhalten des
B4-Monitors zunaechst unveraendert — der Umbau ist reine Struktur.
⚠ Dass der B4-Monitor damit gegen einen VERALTETEN Exit misst, ist ein separater,
offener Befund (s. CLAUDE.md). Ihn hier stillschweigend mitzuaendern haette die
Vergleichsgrundlage des Monitors verschoben, ohne dass es jemand merkt.
"""
from __future__ import annotations

from typing import Callable

# Vorgaben = der bisherige Stand aus `backtest_breakout_squeeze.py`, damit
# Aufrufer ohne explizite Angabe exakt dasselbe bekommen wie vorher.
BOX_N_STD = 12          # Box-Laenge in Bars
FENSTER_STD = 24        # Fenster nach der Box, in dem der Ausbruch fallen muss
COOLDOWN_STD = 12       # Vorsprung nach einem Trade (Ueberlappungsschutz)
K_STD = 0.1             # Ausbruch k×ATR ueber/unter die Box-Grenze
ATRMIN_STD = 0.12
MAXH_STD = 288          # Haltedauer-Obergrenze des Legacy-Exits


def atr_series(H: list, L: list, C: list, p: int = 14) -> list:
    """Gleitender ATR — bitgenau wie `backtest_breakout_squeeze._atr_series`."""
    tr = [0.0]
    for i in range(1, len(C)):
        tr.append(max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])))
    out = [None]
    for i in range(1, len(C)):
        seg = tr[max(1, i - p + 1):i + 1]
        out.append(sum(seg) / len(seg))
    return out


def exit_legacy(entry, d, atr, H, L, C, j0, *, sl_atr=2.0, trail=1.5,
                trail_on=0.3, be_on=1.3, max_hold=MAXH_STD) -> float:
    """⚠ BITGENAUE Kopie von `backtest_breakout_squeeze.sim` — bewusst inklusive
    ihrer Abweichungen vom kanonischen Exit (s. Modul-Docstring). Nicht fuer
    neuen Code verwenden; dort gehoert `core.exit_model.simulate` hin."""
    eff = entry - d * sl_atr * atr
    hw = entry
    end = min(j0 + max_hold, len(C) - 1)
    exit_px = C[end]
    for j in range(j0, end + 1):
        hi, lo = H[j], L[j]
        if (lo <= eff) if d > 0 else (hi >= eff):
            exit_px = eff
            break
        hw = max(hw, hi) if d > 0 else min(hw, lo)
        prof = (C[j] - entry) * d
        if prof >= trail_on * atr:
            cand = hw - d * trail * atr
            if prof >= be_on * atr:
                cand = max(cand, entry) if d > 0 else min(cand, entry)
            eff = max(eff, cand) if d > 0 else min(eff, cand)
    return (exit_px - entry) * d / atr


def scan(H, L, C, A, SP, point, lo_i, hi_i, squeeze_mult, *,
         box_n: int = BOX_N_STD, k: float = K_STD, atr_min: float = ATRMIN_STD,
         fenster: int = FENSTER_STD, cooldown: int = COOLDOWN_STD,
         max_hold: int = MAXH_STD,
         exit_fn: Callable | None = None,
         on_entry: Callable | None = None) -> tuple[list, list]:
    """Ein Durchlauf: sammelt R (netto) je Ausbruch.

    Rueckgabe `(Rs_sq, Rs_any)` — `Rs_any` = ALLE Box-Ausbrueche (die Kontrolle),
    `Rs_sq` = nur die aus einer KOMPRESSION (Box ≤ `squeeze_mult`×ATR). Der
    Vergleich der beiden ist es, der den Squeeze ueberhaupt validiert hat.
    `i` springt nach jedem Ausbruch um `cooldown` vor — fuer beide Listen gleich.

    ⚠ `point` wird nicht mehr benutzt (SP kommt bereits in Preis), bleibt aber in
    der Signatur, damit bestehende Aufrufe unveraendert funktionieren.

    ⚠ `on_entry(j, d, lvl, atr, komprimiert)` ist ADDITIV (2026-08-08): ein
    Rueckruf je Ausbruch, der auch sagt, ob er aus einer KOMPRESSION kam. Ohne
    ihn ist `exit_fn` der einzige Haken — und der wird bei JEDEM Box-Ausbruch
    gerufen, nicht nur bei den komprimierten. Wer darueber eine Entry-Liste
    sammelt, bekommt still die Kontrollgruppe statt des Squeeze (real passiert:
    519 statt 132 Einstiege). Verhalten ohne `on_entry` unveraendert.
    """
    ex = exit_fn or (lambda e, d, a, h, l, c, j: exit_legacy(
        e, d, a, h, l, c, j, max_hold=max_hold))

    def kosten(i, atr):
        return (SP[i] if SP[i] > 0 else 0.0225) / atr

    Rs_sq: list = []
    Rs_any: list = []
    i = max(lo_i, box_n + 15)
    while i < min(hi_i, len(C) - max_hold - 1):
        atr = A[i]
        if not atr or atr < atr_min:
            i += 1
            continue
        box_hi = max(H[i - box_n:i])
        box_lo = min(L[i - box_n:i])
        komprimiert = (box_hi - box_lo) <= squeeze_mult * atr
        treffer = None
        for j in range(i, min(i + fenster, len(C) - max_hold - 1)):
            oben = box_hi + k * atr
            unten = box_lo - k * atr
            if H[j] >= oben:
                treffer = (j, 1, oben)
                break
            if L[j] <= unten:
                treffer = (j, -1, unten)
                break
        if treffer is None:
            i += 1
            continue
        j, d, lvl = treffer
        if on_entry is not None:
            on_entry(j, d, lvl, atr, komprimiert)
        R = ex(lvl, d, atr, H, L, C, j + 1) - kosten(j, atr)
        Rs_any.append(R)
        if komprimiert:
            Rs_sq.append(R)
        i = j + cooldown
    return Rs_sq, Rs_any