claude-regelwerk/.vault-tools/trading-plattform-pruefung/pause_echtlauf.py

"""Eigener Lauf mit echten Daten: Tagesverlust-Schutz in der Pause auf den IEX-Kerzen eines echten Tages.

Koordinator-Pruefcode, nicht Teil des Repos. Er rechnet aus den Rohkerzen vor, in welchem Takt die
Grenze reissen muss, und haelt das gegen das Protokoll der echten PaperSitzung. Der einzige
Netzzugriff holt lesend die Kerzen; Meldungen gehen an einen mitschreibenden Transport, nie ins Netz.
Aufruf (aus dem Projektordner, in WSL): python pause_echtlauf.py JJJJ-MM-TT
"""

import sys
import tempfile
from datetime import datetime
from datetime import timedelta
from pathlib import Path

sys.path.insert(0, str(Path.cwd()))

from trading.killswitch.killswitch import lade_killswitch_config
from trading.paper.config import enge_limits
from trading.paper.config import lade_paper_config
from trading.paper.kerzen import HANDELSZONE
from trading.paper.kerzen import AlpacaKerzenquelle
from trading.paper.kerzen import FesteKerzenquelle
from trading.paper.kerzen import handelsende
from trading.paper.melder import Melder
from trading.paper.melder import MelderConfig
from trading.paper.probe import ProbeBroker
from trading.paper.sitzung import PaperSitzung
from trading.paper.uhr import ProbeUhr
from trading.registry.registry import Registry
from trading.risk.limits import lade_limits
from dataclasses import replace


class Protokoll:
    def __init__(self):
        self.zeilen = []

    def schreibe(self, eintrag):
        self.zeilen.append(dict(eintrag))


class Antwort:
    status_code = 200


class Transport:
    def __init__(self):
        self.aufrufe = []

    def __call__(self, url, *, data=None, headers=None, timeout=None):
        self.aufrufe.append((headers["Title"], headers["Priority"], data.decode("utf-8")))
        return Antwort()


class Flach:
    def predict(self, beobachtung, deterministic=False):
        return 2, None


def lauf(kerzen, *, config, limits, killswitch_config, zustand, position):
    ordner = Path(tempfile.mkdtemp(prefix="pause-echtlauf-"))
    quelle = FesteKerzenquelle(kerzen)
    uhr = ProbeUhr(kerzen[0].zeit - timedelta(minutes=1))
    broker = ProbeBroker(
        kerzenquelle=quelle, uhr=uhr, instrument_id=config.instrument_id, position=position
    )
    registry = Registry(db_pfad=ordner / "registry.db", artefakt_wurzel=ordner / "artifacts")
    registry.setze_killswitch_zustand(zustand, ausloeser="echtlauf")
    protokoll = Protokoll()
    transport = Transport()
    melder = Melder(
        config=MelderConfig(
            aktiv=True,
            ntfy_url="https://ntfy.sh",
            thema="echtlauf-geht-nie-ins-netz",
            zeitlimit_s=3.0,
            stille_nach_s=180.0,
        ),
        token="",
        transport=transport,
    )
    sitzung = PaperSitzung(
        broker=broker,
        kerzenquelle=quelle,
        agent=Flach(),
        agent_beschreibung={"quelle": "echtlauf"},
        registry=registry,
        limits=limits,
        killswitch_config=killswitch_config,
        config=config,
        uhr=uhr,
        protokoll=protokoll,
        protokoll_pfad="echtlauf.jsonl",
        status_pfad=ordner / "status.json",
        modus="probe",
        soll_abbrechen=lambda: False,
        melder=melder,
    )
    code = sitzung.laufe()
    return code, protokoll.zeilen, transport.aufrufe, broker, registry


def main():
    datum = sys.argv[1]
    mitternacht = datetime.strptime(datum, "%Y-%m-%d").replace(tzinfo=HANDELSZONE)
    config = lade_paper_config("config/paper.toml")
    killswitch_config = lade_killswitch_config("config/killswitch.toml")
    quelle = AlpacaKerzenquelle.from_secrets(None, feed=config.feed, timeout=10.0)
    kerzen = list(quelle.kerzen_des_tages(config.instrument_id, handelsende(mitternacht)))
    print(f"{datum}: {len(kerzen)} IEX-Kerzen in {config.instrument_id}")
    schluesse = [k.close for k in kerzen]
    print(f"Schluss erste Kerze {schluesse[0]:.2f}, Tief {min(schluesse):.2f}, Hoch {max(schluesse):.2f}")

    stueck = 20
    cash = 100_000.0
    befunde = 0
    for seite, position in (("long", stueck), ("short", -stueck)):
        # Eigene Rechnung aus den Rohkerzen: Takt k (schritt k) sieht die Kerze k-1 als juengste
        # abgeschlossene. Startkapital ist der Kontowert im ersten Takt.
        start = cash + position * schluesse[0]
        schlechtester = min(cash + position * s for s in schluesse) - start
        if schlechtester >= 0:
            print(f"[{seite}] der Tag bringt dieser Seite keinen Verlust gegen den ersten Takt; uebersprungen")
            continue
        # Die Grenze liegt bei der Haelfte des groessten Verlusts dieses Tages: Der Tag reisst sie
        # wirklich, und nicht im ersten Takt.
        anteil = abs(schlechtester) / start / 2.0
        erwartet = None
        for k, s in enumerate(schluesse, start=1):
            if (cash + position * s - start) / start < -anteil:
                erwartet = k
                break
        limits = replace(enge_limits(lade_limits("config/risk.toml"), config), max_tagesverlust_anteil=anteil)
        code, zeilen, meldungen, broker, registry = lauf(
            kerzen, config=config, limits=limits, killswitch_config=killswitch_config,
            zustand="angehalten", position=position,
        )
        takte = [z for z in zeilen if z.get("art") == "takt"]
        grenz = [z for z in zeilen if z.get("was") == "grenzwert"]
        # Der Takt, in dem die Zeile grenzwert steht: die Taktzeile danach traegt seinen Schritt.
        stelle = zeilen.index(grenz[0]) if grenz else None
        schritt = next((z["schritt"] for z in zeilen[stelle:] if z.get("art") == "takt"), None) if grenz else None
        titel = [m[0] for m in meldungen]
        print(f"[{seite}] Startkapital {start:.2f}, Grenze {anteil:.6f} (= {anteil * start:.2f} USD), "
              f"erwarteter Takt {erwartet}, gemessen {schritt}")
        print(f"[{seite}] Zustand {registry.killswitch_zustand().zustand}, Position danach "
              f"{sum(p.menge for p in broker.positionen())}, Exit {code}, Meldungen {titel}")
        print(f"[{seite}] schliesse_alle: {broker.aufrufe.count('schliesse_alle')}, Takte {len(takte)}, "
              f"grenzwert-Zeilen {len(grenz)}")
        ok = (
            schritt == erwartet
            and registry.killswitch_zustand().zustand == "gesperrt"
            and broker.aufrufe.count("schliesse_alle") == 1
            and len(grenz) == 1
            and titel == ["Paper-Runner wartet", "Grenze verletzt", "Paper-Runner beendet"]
            and code == 0
        )
        print(f"[{seite}] {'OK' if ok else 'BEFUND'}")
        befunde += 0 if ok else 1

    # Gegenprobe: dieselben Kerzen, angehalten, ohne Position -- genau eine Kontoabfrage, nichts geschlossen.
    limits = enge_limits(lade_limits("config/risk.toml"), config)
    code, zeilen, meldungen, broker, registry = lauf(
        kerzen, config=config, limits=limits, killswitch_config=killswitch_config,
        zustand="angehalten", position=0,
    )
    titel = [m[0] for m in meldungen]
    ok = (
        broker.aufrufe.count("konto") == 1
        and "schliesse_alle" not in broker.aufrufe
        and titel == ["Paper-Runner wartet", "Paper-Runner beendet"]
        and code == 0
    )
    print(f"[ohne Position] konto {broker.aufrufe.count('konto')}, Meldungen {titel}, Exit {code}: "
          f"{'OK' if ok else 'BEFUND'}")
    befunde += 0 if ok else 1

    # Gegenprobe 2: aktiv mit dem echten Haltemuster aendert sich nicht -- das prueft die Probe des Skripts.
    print(f"{befunde} Befund(e).")
    return 2 if befunde else 0


if __name__ == "__main__":
    sys.exit(main())