AH-Oil-Trader/run_all_backtests.py

#!/usr/bin/env python3
"""Führt ALLE Backtest-/Analyse-Skripte aus und protokolliert das Ergebnis.

Zweck (User-Auftrag 2026-08-04): nach den Änderungen am Signal (Reversal aus,
`angle=`-Befund) prüfen, ob die Skripte noch laufen UND ob ihre Aussagen halten.

⚠ Die Aufruf-Signaturen sind NICHT einheitlich. Die meisten nehmen eine Bar-Zahl
als erstes Argument, einige aber einen Timeframe (`backtest_signal.py M5 3000`,
`backtest_htf_filter.py M5 M30 3000`). Ein Aufruf mit der falschen Signatur
quittiert mit `KeyError`/`ValueError` — das ist KEIN Defekt und darf nicht als
solcher gezählt werden. Deshalb probiert der Runner adaptiv mehrere Varianten
und meldet erst dann einen Fehler, wenn KEINE davon durchläuft.

Aufruf:  python run_all_backtests.py [bars] [--only muster]
"""
from __future__ import annotations

import pathlib
import subprocess
import sys
import time

PY = sys.executable
HERE = pathlib.Path(__file__).parent

# Skripte, die kein eigenständiger Lauf sind (Bibliothek/Helfer) oder extrem lange
# laufen und separat gefahren werden.
SKIP = {
    "run_all_backtests.py",
}


def varianten(n: int) -> list[list[str]]:
    """Aufruf-Varianten in der Reihenfolge, in der sie probiert werden."""
    return [
        [str(n)],                       # Standard: Bar-Zahl
        ["M5", str(n)],                 # TF + Bar-Zahl
        ["M5", "M30", str(n)],          # zwei TFs + Bar-Zahl
        [],                             # ohne Argumente (Defaults)
    ]


def main() -> None:
    n = 6000
    only = None
    args = sys.argv[1:]
    if args and args[0].isdigit():
        n = int(args[0]); args = args[1:]
    if len(args) >= 2 and args[0] == "--only":
        only = args[1]

    files = sorted(set(list(HERE.glob("backtest_*.py")) + list(HERE.glob("analyze_*.py"))))
    files = [f for f in files if f.name not in SKIP]
    if only:
        files = [f for f in files if only in f.name]

    print(f"{len(files)} Skripte · {n} Bars · Start {time.strftime('%H:%M:%S')}")
    print("=" * 96)
    ok, fail, slow = [], [], []
    for i, f in enumerate(files, 1):
        t0 = time.time()
        erfolg = None
        letzter_fehler = ""
        for v in varianten(n):
            try:
                r = subprocess.run([PY, "-X", "utf8", f.name] + v, cwd=HERE,
                                   capture_output=True, text=True, timeout=900,
                                   encoding="utf-8", errors="replace")
            except subprocess.TimeoutExpired:
                letzter_fehler = "TIMEOUT (>900 s)"
                continue
            if r.returncode == 0:
                erfolg = (v, r.stdout or "")
                break
            tail = [x for x in (r.stderr or "").splitlines() if x.strip()]
            letzter_fehler = tail[-1][:110] if tail else f"exit {r.returncode}"
        dt = time.time() - t0
        if erfolg:
            v, out = erfolg
            zeilen = [x for x in out.splitlines() if x.strip()]
            sig = " ".join(v) or "(ohne)"
            print(f"[{i:>2}/{len(files)}] OK      {f.name:<36} {dt:>6.1f}s  args={sig}")
            ok.append((f.name, sig, dt, zeilen[-1][:100] if zeilen else ""))
            if dt > 120:
                slow.append((f.name, dt))
        else:
            print(f"[{i:>2}/{len(files)}] FEHLER  {f.name:<36} {dt:>6.1f}s  {letzter_fehler}")
            fail.append((f.name, letzter_fehler))

    print("=" * 96)
    print(f"OK: {len(ok)}   FEHLER: {len(fail)}   (Ende {time.strftime('%H:%M:%S')})")
    if fail:
        print("\nFEHLGESCHLAGEN:")
        for name, err in fail:
            print(f"   {name:<36} {err}")
    if slow:
        print("\nLangsam (>120 s):")
        for name, dt in sorted(slow, key=lambda x: -x[1]):
            print(f"   {name:<36} {dt:.0f}s")


if __name__ == "__main__":
    main()