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

"""Eigener Lauf mit echten Daten: Der Tagesverlust zaehlt nach einem Neustart weiter ab dem Tagesstart.

Koordinator-Pruefcode, nicht Teil des Repos. Echte IEX-Kerzen eines Tages (JSON aus dem lesenden
Abruf der Abschlusspruefung). Aus den Rohkerzen wird vorgerechnet, bei welcher Kerze die Grenze von
0,2 Prozent gegen den Tagesstart reisst und bei welcher gegen den Neustartkurs; dann laufen zwei
echte PaperSitzungen nacheinander (vor und nach dem Neustart) mit gemeinsamer Tagesstart-Datei,
dazu die Gegenprobe ohne Datei. Kein Netz: Meldungen gehen an einen mitschreibenden Transport.
Aufruf (WSL, aus dem Projektordner): python neustart_echtlauf.py <kerzen.json>
"""

import json
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 FesteKerzenquelle
from trading.paper.kerzen import Kerze
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.status import lies_tagesstart
from trading.paper.uhr import ProbeUhr
from trading.registry.registry import Registry
from trading.risk.limits import lade_limits

STUECK = 20
ZIEL_KAPITAL = 12_676.20  # so klein, dass 20 Stueck die echte Grenze von 0,2 Prozent erreichen


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"])
        return Antwort()


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


def lade(pfad):
    roh = json.loads(Path(pfad).read_text(encoding="utf-8"))
    return [
        Kerze(zeit=datetime.fromisoformat(e["zeit"]), open=e["open"], high=e["high"],
              low=e["low"], close=e["close"], volume=e["volume"])
        for e in roh
    ]


def lauf(ordner, kerzen, *, beginn, position, cash, tagesstart, config, limits, ks_config):
    ordner.mkdir(parents=True, exist_ok=True)
    quelle = FesteKerzenquelle(kerzen)
    uhr = ProbeUhr(beginn)
    broker = ProbeBroker(kerzenquelle=quelle, uhr=uhr, instrument_id=config.instrument_id,
                         startkapital=cash, position=position)
    registry = Registry(db_pfad=ordner / "registry.db", artefakt_wurzel=ordner / "artifacts")
    registry.setze_killswitch_zustand("angehalten", ausloeser="neustart")
    protokoll = Protokoll()
    transport = Transport()
    sitzung = PaperSitzung(
        broker=broker, kerzenquelle=quelle, agent=Flach(), agent_beschreibung={"quelle": "echtlauf"},
        registry=registry, limits=limits, killswitch_config=ks_config, config=config, uhr=uhr,
        protokoll=protokoll, protokoll_pfad="echtlauf.jsonl", status_pfad=ordner / "status.json",
        modus="probe", soll_abbrechen=lambda: False,
        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),
        tagesstart_pfad=tagesstart,
    )
    code = sitzung.laufe()
    return code, protokoll.zeilen, transport.aufrufe, broker, registry


def kerze_beim_grenzereignis(zeilen, index_nach_zeit):
    """Index der juengsten Kerze in dem Takt, in dem die Zeile grenzwert steht, oder None."""
    for stelle, zeile in enumerate(zeilen):
        if zeile.get("was") == "grenzwert":
            takt = next(z for z in zeilen[stelle:] if z.get("art") == "takt")
            return index_nach_zeit[takt["letzte_kerze"]], takt["schritt"]
    return None, None


def erster_riss(schluesse, *, ab, cash, position, bezug, anteil):
    for c in range(ab, len(schluesse)):
        if (cash + position * schluesse[c] - bezug) / bezug < -anteil:
            return c
    return None


def main():
    kerzen = lade(sys.argv[1])
    schluesse = [k.close for k in kerzen]
    index_nach_zeit = {k.zeit.isoformat(): i for i, k in enumerate(kerzen)}
    config = lade_paper_config("config/paper.toml")
    limits = enge_limits(lade_limits("config/risk.toml"), config)
    ks_config = lade_killswitch_config("config/killswitch.toml")
    anteil = limits.max_tagesverlust_anteil
    tag = kerzen[0].zeit.date().isoformat()
    print(f"{tag}: {len(kerzen)} Kerzen, Schluss Kerze 0 {schluesse[0]:.2f}, "
          f"Tief {min(schluesse):.2f}, Hoch {max(schluesse):.2f}, Grenze {anteil}")

    befunde = 0
    for seite, position in (("long", STUECK), ("short", -STUECK)):
        cash = ZIEL_KAPITAL - position * schluesse[0]
        tagesstart_wert = cash + position * schluesse[0]
        riss_tag = erster_riss(schluesse, ab=0, cash=cash, position=position,
                               bezug=tagesstart_wert, anteil=anteil)
        if riss_tag is None or riss_tag < 3:
            print(f"[{seite}] der Tag reisst die Grenze gegen den Tagesstart nicht (oder sofort): uebersprungen")
            continue
        # Neustart so, dass die Sitzung davor die Kerze des Risses nicht mehr sieht: Sie laeuft
        # ueber kerzen[:r] und sieht die Kerzen 0 bis r-2. Die Sitzung danach beginnt in der
        # Minute der Kerze r und sieht zuerst die Kerze r-1.
        r = riss_tag
        bezug_neu = cash + position * schluesse[r - 1]
        riss_neu = erster_riss(schluesse, ab=r - 1, cash=cash, position=position,
                               bezug=bezug_neu, anteil=anteil)
        print(f"[{seite}] Tagesstart {tagesstart_wert:.2f}; vorgerechnet: Riss gegen den Tagesstart "
              f"bei Kerze {riss_tag}, gegen den Neustartkurs ({bezug_neu:.2f}) bei Kerze {riss_neu}")

        with tempfile.TemporaryDirectory(prefix="neustart-echtlauf-") as wurzel:
            wurzel = Path(wurzel)
            datei = wurzel / "paper" / "tagesstart.json"
            gemeinsam = dict(position=position, cash=cash, config=config, limits=limits,
                             ks_config=ks_config)

            code1, zeilen1, _, _, reg1 = lauf(
                wurzel / "vorher", kerzen[:r], beginn=kerzen[0].zeit - timedelta(minutes=1),
                tagesstart=datei, **gemeinsam)
            riss1, _ = kerze_beim_grenzereignis(zeilen1, index_nach_zeit)
            gemerkt = lies_tagesstart(datei, handelstag=tag)
            print(f"[{seite}] vor dem Neustart: Exit {code1}, Killswitch "
                  f"{reg1.killswitch_zustand().zustand}, Riss {riss1}, gemerkt {gemerkt}")

            code2, zeilen2, titel2, broker2, reg2 = lauf(
                wurzel / "nachher", kerzen, beginn=kerzen[r].zeit - timedelta(minutes=1),
                tagesstart=datei, **gemeinsam)
            riss2, schritt2 = kerze_beim_grenzereignis(zeilen2, index_nach_zeit)
            takte2 = [z for z in zeilen2 if z.get("art") == "takt"]
            # Vor dem Ende des with-Blocks lesen: Danach ist das Wegwerf-Register weg.
            zustand2 = reg2.killswitch_zustand().zustand
            position2 = sum(p.menge for p in broker2.positionen())
            print(f"[{seite}] nach dem Neustart mit Datei: Startkapital {takte2[0]['startkapital']}, "
                  f"Riss bei Kerze {riss2} (Schritt {schritt2}), Exit {code2}, Killswitch "
                  f"{zustand2}, Position {position2}, Meldungen {titel2}")

            code3, zeilen3, _, broker3, reg3 = lauf(
                wurzel / "ohne", kerzen, beginn=kerzen[r].zeit - timedelta(minutes=1),
                tagesstart=None, **gemeinsam)
            riss3, _ = kerze_beim_grenzereignis(zeilen3, index_nach_zeit)
            takte3 = [z for z in zeilen3 if z.get("art") == "takt"]
            print(f"[{seite}] nach dem Neustart ohne Datei: Startkapital {takte3[0]['startkapital']}, "
                  f"Riss bei Kerze {riss3}, Exit {code3}, Killswitch {reg3.killswitch_zustand().zustand}")

        ok = (
            riss1 is None and code1 == 4
            and gemerkt is not None and abs(gemerkt - tagesstart_wert) < 1e-9
            and takte2[0]["startkapital"] == gemerkt
            and riss2 == riss_tag
            and zustand2 == "gesperrt"
            and position2 == 0
            and titel2 == ["Paper-Runner wartet", "Grenze verletzt", "Paper-Runner beendet"]
            and abs(takte3[0]["startkapital"] - bezug_neu) < 1e-9
            and riss3 == riss_neu
        )
        print(f"[{seite}] {'OK' if ok else 'BEFUND'}")
        befunde += 0 if ok else 1

    print(f"{befunde} Befund(e).")
    return 2 if befunde else 0


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