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

"""Misst Luecken im IEX-Minutenfeed je Symbol und Handelstag. Nur lesend.

Koordinator-Werkzeug (2026-09-24 nachts, Paket M). Anlass: Der Gratis-Feed IEX liefert fuer weniger bei IEX
gehandelte Werte nicht jede Minute eine Kerze (am 2026-09-23 AMGN 287 von 390). Der Watchdog haelt an, wenn
der letzte Kurs aelter als kursstrom_max_alter_s (300 s) ist; diese Messung zeigt, wie lang die laengste
Minute ohne Kerze je Symbol ist, und wie oft eine Luecke ueber 5 Minuten reicht. Ausgegeben werden nur Zaehler.
Aufruf (WSL, TRADING_GEHEIMNISSE gesetzt): python iex_luecken.py TAG[,TAG...] SYMBOL [SYMBOL ...]
"""

import sys
import tomllib
from datetime import datetime
from pathlib import Path

import requests

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

from trading import geheimnisse  # noqa: E402

DATEN = "https://data.alpaca.markets"


def main() -> int:
    tage, symbole = sys.argv[1].split(","), sys.argv[2:]
    werte = tomllib.loads(geheimnisse.pfad().read_text(encoding="utf-8"))["alpaca"]
    s = requests.Session()
    s.trust_env = False
    s.headers.update({"APCA-API-KEY-ID": werte["key_id"], "APCA-API-SECRET-KEY": werte["secret_key"]})
    for tag in tage:
        p = {"symbols": ",".join(symbole), "timeframe": "1Min", "start": f"{tag}T13:30:00Z",
             "end": f"{tag}T20:00:00Z", "feed": "iex", "limit": 10000, "adjustment": "raw"}
        j = s.get(f"{DATEN}/v2/stocks/bars", params=p, timeout=30).json()
        assert not j.get("next_page_token"), "mehr als eine Seite"
        print(f"== {tag}")
        for sym in symbole:
            zeiten = [datetime.fromisoformat(b["t"].replace("Z", "+00:00")) for b in (j.get("bars") or {}).get(sym, [])]
            luecken = [int((b - a).total_seconds() // 60) - 1 for a, b in zip(zeiten, zeiten[1:])]
            ueber5 = sum(1 for x in luecken if x >= 5)
            print(f"{sym:5} Kerzen {len(zeiten):3}  laengste Luecke {max(luecken) if luecken else '-':>3} min  Luecken >=5 min: {ueber5}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())