AH-Oil-Trader/tools/check_slots.py

"""Statische Prüfung: Attribut-Reihenfolge in `__init__` + Slot-Trennung.

⚠⚠ WOZU — beide Fehlerklassen haben in ZWEI TAGEN je einmal zugeschlagen, und
KEINE bestehende Stufe hat sie gefangen:

  (A) ATTRIBUT VOR SEINER ZUWEISUNG BENUTZT.
      `self.trader_brk.history = self.history` stand in `__init__` VOR
      `self.trader_brk = TradeManager(...)`. Ergebnis: AttributeError im
      Konstruktor → der Serverstart brach KOMPLETT ab, und der Nutzer konnte
      zwei Minuten lang keine Position schliessen ("Netzwerkfehler" am
      Close-Knopf). py_compile, ruff F821, ty und pytest liefen alle GRÜN —
      syntaktisch ist es korrekt, und kein Test instanziiert die Engine (sie
      braucht MT5). Gefunden hat es erst `deploy.py` Schritt 2 — also NACH dem
      Kill des laufenden Servers.
      Derselbe Fehler am Vortag mit `_margin_brk` vor `_load_runtime_state()`:
      dort blieb er sogar still (der Wert wurde nur ueberschrieben).

  (B) SLOT-VERWECHSLUNG. Seit dem Zwei-Slot-Umbau (19.08.) gibt es
      `self.trader` (manuell + M15) und `self.trader_brk` (Breakout). Mehrere
      Pfade griffen nach dem falschen: `engine.close()` schloss nur Slot 1 (eine
      BRK-Position war NICHT schliessbar), `squeeze_reverse` schloss die
      manuelle statt der BRK-Position, `open_long` routete den Market-Fallback
      auf Slot 1 ("Position bereits offen!").

Beide Prüfungen sind STATISCH (AST) — kein MT5, keine Laufzeit, Millisekunden.

Rückgabe: Liste von Befunden (leer = sauber). Wird von `tools/check_nfalle.py`
als Stufe D eingebunden und läuft damit im pre-commit-Hook mit.

Aufruf einzeln:  python tools/check_slots.py
"""
from __future__ import annotations

import ast
import pathlib
import sys

WURZEL = pathlib.Path(__file__).resolve().parent.parent

# Funktionen, die zum BRK-Pfad gehören und deshalb NICHT auf `self.trader`
# zugreifen dürfen. ⚠ Bewusst eine WEISSE Liste: eine schwarze Liste („alles
# ausser…") würde bei jeder neuen Funktion stillschweigend durchlassen.
BRK_PFADE = {
    "_check_auto_squeeze",
    "_manage_squeeze_pending",
    "_check_pending_fill",
}
# Zugriffe, die den SLOT-ZUSTAND betreffen. `pending_orders`, `cancel_pending`
# und `place_stop` sind magic-basierte BROKER-Abfragen ohne Slot-Zustand und
# deshalb ausdruecklich erlaubt (geprüft 2026-08-20).
SLOT_METHODEN = {"close", "snapshot", "open_long", "open_short",
                 "set_open_context", "refresh"}


def _init_reihenfolge(baum: ast.AST, quelle: str) -> list:
    """(A) `self.X.…` benutzt, bevor `self.X = …` zugewiesen wurde."""
    befunde = []
    for kls in [n for n in ast.walk(baum) if isinstance(n, ast.ClassDef)]:
        init = next((f for f in kls.body
                     if isinstance(f, ast.FunctionDef) and f.name == "__init__"), None)
        if not init:
            continue
        gesetzt: set[str] = set()
        for knoten in ast.walk(init):
            # Zuweisungen in Reihenfolge einsammeln … aber ast.walk ist NICHT
            # zeilentreu. Deshalb unten über die Zeilennummer entschieden.
            pass
        # Zeile → Name, an der `self.X` zugewiesen wird (erste Zuweisung zählt)
        erste: dict[str, int] = {}
        for knoten in ast.walk(init):
            if isinstance(knoten, ast.Assign):
                for ziel in knoten.targets:
                    for t in ([ziel] if not isinstance(ziel, ast.Tuple) else ziel.elts):
                        if (isinstance(t, ast.Attribute)
                                and isinstance(t.value, ast.Name)
                                and t.value.id == "self"):
                            erste.setdefault(t.attr, knoten.lineno)
        # Benutzungen: `self.X.irgendwas` — X muss vorher zugewiesen sein
        for knoten in ast.walk(init):
            if (isinstance(knoten, ast.Attribute)
                    and isinstance(knoten.value, ast.Attribute)
                    and isinstance(knoten.value.value, ast.Name)
                    and knoten.value.value.id == "self"):
                name = knoten.value.attr
                if name not in erste:
                    continue              # kommt von ausserhalb → nicht pruefbar
                if knoten.lineno < erste[name]:
                    befunde.append(
                        f"{kls.name}.__init__:{knoten.lineno}: `self.{name}` "
                        f"benutzt, aber erst in Zeile {erste[name]} zugewiesen "
                        f"→ AttributeError beim Start")
    return sorted(set(befunde))


def _slot_trennung(baum: ast.AST) -> list:
    """(B) BRK-Pfade duerfen den Slot-Zustand von `self.trader` nicht anfassen."""
    befunde = []
    for fn in [n for n in ast.walk(baum) if isinstance(n, ast.FunctionDef)]:
        if fn.name not in BRK_PFADE:
            continue
        for knoten in ast.walk(fn):
            if not isinstance(knoten, ast.Attribute):
                continue
            v = knoten.value
            if (isinstance(v, ast.Attribute) and v.attr == "trader"
                    and isinstance(v.value, ast.Name) and v.value.id == "self"
                    and knoten.attr in SLOT_METHODEN):
                befunde.append(
                    f"{fn.name}:{knoten.lineno}: `self.trader.{knoten.attr}` im "
                    f"BRK-Pfad — muss `self.trader_brk` sein")
    return sorted(set(befunde))


def _close_beide(baum: ast.AST) -> list:
    """`engine.close()` muss BEIDE Slots schliessen (sonst ist eine BRK-Position
    nicht schliessbar — die gefaehrlichste Variante der Luecke)."""
    for kls in [n for n in ast.walk(baum) if isinstance(n, ast.ClassDef)]:
        if kls.name != "TradingEngine":
            continue
        fn = next((f for f in kls.body
                   if isinstance(f, ast.FunctionDef) and f.name == "close"), None)
        if fn and "trader_brk" not in ast.unparse(fn):
            return ["TradingEngine.close: erwaehnt `trader_brk` NICHT — eine "
                    "BRK-Position waere per Dashboard nicht schliessbar"]
    return []


def _sperrliste_vollstaendig(baum: ast.AST) -> list:
    """(D) Die Adoptions-Sperre muss ALLE DREI Quellen eines BRK-Tickets kennen.

    ⚠⚠ Sie hatte am 2026-08-20 eine Luecke, die bei JEDEM Neustart aufriss:
    `trader_brk.ticket` und `_pending_tickets` leben nur im Speicher, nach einem
    Neustart sind beide leer. Das Ticket steht dann allein im Merker
    `_brk_restore` — und weil `trader.refresh()` im `_pos_loop` DREI Zeilen vor
    `_rebind_brk()` laeuft, griff Slot 1 die BRK-Position, bevor sie
    zurueckgebunden werden konnte. Folge: ihr G/V stand links statt rechts, und
    „G/V BRK" zeigte „—". Belegt in `tests/test_slot_gv.py`.
    """
    for kls in [n for n in ast.walk(baum) if isinstance(n, ast.ClassDef)]:
        if kls.name != "TradingEngine":
            continue
        init = next((f for f in kls.body if isinstance(f, ast.FunctionDef)
                     and f.name == "__init__"), None)
        if not init:
            continue
        for kn in ast.walk(init):
            if (isinstance(kn, ast.Assign)
                    and any(isinstance(t, ast.Attribute) and t.attr == "fremde_tickets"
                            for t in kn.targets)):
                q = ast.unparse(kn)
                fehlt = [n for n in ("trader_brk", "_pending_tickets", "_brk_restore")
                         if n not in q]
                if fehlt:
                    return ["fremde_tickets: Sperrliste erwaehnt {} NICHT — eine "
                            "BRK-Position kann von Slot 1 adoptiert werden "
                            "(G/V stuende dann im falschen Feld)"
                            .format(", ".join(fehlt))]
                return []
    return []


def _brk_bekommt_trailing(baum: ast.AST) -> list:
    """(E) Jeder Pfad, der eine BRK-Position EROEFFNET, muss `trail_brk` anfassen.

    ⚠⚠ DER GEFAEHRLICHSTE FUND DER SLOT-PRUEFUNG (2026-08-20): `trail_brk.toggle()`
    wurde an genau ZWEI Stellen gerufen — beim Rueckbinden nach einem Neustart
    (`_rebind_brk`) und ueber den TRAIL-Knopf (`toggle_trail`). Der FRISCHE
    Pending-Fill kam in keiner davon vor, und das ist der haeufigste Weg, auf dem
    eine BRK-Position entsteht (7 von 8 Trades seit dem Pending-Umbau). Der
    Market-Fallback (`_open`) war noch schlechter: er schaltete `self.trail` ein,
    also das Trailing der MANUELLEN Position — die BRK-Position blieb ungeschuetzt
    UND dem Nutzer wurde auf seinem eigenen Trade das Trailing wieder angemacht.
    Eine nur mit dem Broker-SL laufende BRK-Position ist genau die Variante, die
    `backtest_brk_slonly.py` verworfen hat (alle zehn KI enthalten die Null,
    Trefferquote faellt 40 -> 21 %).
    ⚠ Von aussen unsichtbar: die Position sieht in jeder Anzeige normal aus, ihr
    fehlt nur der Schutz. Deshalb diese statische Regel.
    """
    OEFFNER = {"_check_pending_fill": "der Pending-Fill",
               "_open": "der Market-Fallback"}
    befunde = []
    for fn in [n for n in ast.walk(baum) if isinstance(n, ast.FunctionDef)]:
        if fn.name not in OEFFNER:
            continue
        if "trail_brk" not in ast.unparse(fn):
            befunde.append(
                f"{fn.name}: erwaehnt `trail_brk` NICHT — {OEFFNER[fn.name]} "
                f"wuerde eine BRK-Position OHNE Trailing eroeffnen "
                f"(nur Broker-SL = gemessen durchgefallen)")
    return sorted(befunde)


def pruefe() -> list:
    p = WURZEL / "core" / "engine.py"
    quelle = p.read_text(encoding="utf-8")
    baum = ast.parse(quelle)
    return (_init_reihenfolge(baum, quelle) + _slot_trennung(baum)
            + _close_beide(baum) + _sperrliste_vollstaendig(baum)
            + _brk_bekommt_trailing(baum))


if __name__ == "__main__":
    b = pruefe()
    for z in b:
        print("  FEHLER  " + z)
    print("  keine Slot-/Reihenfolge-Fehler." if not b
          else f"  {len(b)} Befund(e)")
    sys.exit(1 if b else 0)