Warrior_EA/research/run_screen.py

"""
Multi-symbol screen of the RECOVERED live configuration.

The parameters are not searched. They are the ones recovered from the surviving
tester configs (z_*.ini / m*.ini) of the lost 2026-09-13 build:

    Period=H4  DipEntry=1 (z-score)  DipZ=1.5  DipExitMA=20
    DipTrendMA=0  DipMaxBars=10  Direction=1 (long only)  StopMode=3

Testing a recovered configuration instead of fitting a new one is the single
biggest defence against the curve-fit result this project keeps producing:
there is no free parameter left to tune, so an OOS number means something.

Every candidate is measured against a LONG-ONLY RANDOM ENTRY with matched
holding period. On a drifting index that is the only honest control.
"""

from __future__ import annotations

import sys
import numpy as np

sys.path.insert(0, __file__.rsplit("\\", 1)[0] if "\\" in __file__ else ".")

import backtest as bt  # noqa: E402

RISK = 0.01
STOP_ATR = 3.0
PROP_DD = 0.05          # 5% prop limit


def zscore_entries(d, n=20, z_th=-1.5, trend_ma=0):
    c = d["c"]
    m = bt.sma(c, n)
    s = bt.rolling_std(c, n)
    z = (c - m) / np.where(s > 0, s, np.nan)
    e = z <= z_th
    if trend_ma and trend_ma > 0:
        e &= c > bt.sma(c, trend_ma)
    return np.nan_to_num(e, nan=0).astype(bool), bt.sma(c, n)


def rsi_entries(d, n=2, th=10.0):
    return bt.rsi(d["c"], n) <= th, bt.sma(d["c"], 20)


def breakout_entries(d, n=20):
    c, h = d["c"], d["h"]
    hh = np.full(len(c), np.nan)
    for i in range(n, len(c)):
        hh[i] = h[i - n:i].max()
    return (c > hh), bt.sma(c, 20)


def random_entries(d, rate, seed=0):
    rng = np.random.default_rng(seed)
    return rng.random(len(d["c"])) < rate, bt.sma(d["c"], 20)


def evaluate(d, entries, exit_ma, side=1, max_bars=10, lo=None, hi=None):
    ts = d["ts"]
    mask = np.ones(len(ts), bool)
    if lo is not None:
        mask &= ts >= np.datetime64(lo)
    if hi is not None:
        mask &= ts < np.datetime64(hi)
    e = entries & mask
    tr = bt.simulate(d, e, side=side, exit_ma=exit_ma, max_bars=max_bars, stop_atr=STOP_ATR)
    tr = bt.add_r(tr, d, stop_atr=STOP_ATR)
    sub = ts[mask]
    if len(sub) < 10 or not tr:
        return None
    return bt.metrics(tr, sub, risk_frac=RISK, stop_atr=STOP_ATR), tr


def row(tag, m):
    if m is None:
        return f"{tag:<22}      no trades"
    flags = []
    if m["per_month"] < 2.0:
        flags.append("cadence")
    if m["maxdd"] > PROP_DD:
        flags.append("maxDD")
    if not (m["ret_dd"] >= 2.0):
        flags.append("ret/DD")
    verdict = "PASS" if not flags else "fail:" + ",".join(flags)
    return (f"{tag:<22}{m['n']:>6}{m['per_month']:>7.1f}{m['exp_bp']:>9.1f}"
            f"{m['hit']:>7.2f}{m['total']:>9.1%}{m['maxdd']:>8.1%}"
            f"{m['ret_dd']:>8.2f}  {verdict}")


HDR = (f"{'candidate':<22}{'n':>6}{'/mo':>7}{'exp_bp':>9}{'hit':>7}"
       f"{'total':>9}{'maxDD':>8}{'ret/DD':>8}  screen")

SYMS = ["SP500", "NAS100", "US30", "DAX40", "EURUSD", "USDJPY", "XAUUSD"]
SPLIT = "2024-01-01"

if __name__ == "__main__":
    period = sys.argv[1] if len(sys.argv) > 1 else "PERIOD_H4"
    print(f"=== RECOVERED CONFIG, {period}, long only, z<=-1.5, exit SMA20, "
          f"max 10 bars, stop 3 ATR, risk {RISK:.0%} ===")
    print(f"=== IS: start..{SPLIT}   OOS: {SPLIT}..end   (prop limit {PROP_DD:.0%}) ===\n")
    for s in SYMS:
        try:
            d = bt.load(s, period)
        except Exception as ex:  # noqa: BLE001
            print(f"{s}: load failed {ex}")
            continue
        e, xma = zscore_entries(d, 20, -1.5, 0)
        print(f"--- {s}  ({d['ts'][0]} .. {d['ts'][-1]}, {len(d['c'])} bars, "
              f"median spread {np.median(d['cost'] / d['c']) * 1e4:.2f} bp) ---")
        print(HDR)
        for tag, lo, hi in (("dip-z IS", None, SPLIT), ("dip-z OOS", SPLIT, None),
                            ("dip-z FULL", None, None)):
            r = evaluate(d, e, xma, 1, 10, lo, hi)
            print(row(tag, r[0] if r else None))
        # control: long-only random entry, matched trade count
        full = evaluate(d, e, xma, 1, 10, None, None)
        if full:
            rate = full[0]["n"] / len(d["c"])
            accs = []
            for sd in range(8):
                re_, rma = random_entries(d, rate, seed=sd)
                rr = evaluate(d, re_, rma, 1, 10, None, None)
                if rr:
                    accs.append(rr[0])
            if accs:
                print(f"{'random-long control':<22}{np.mean([a['n'] for a in accs]):>6.0f}"
                      f"{np.mean([a['per_month'] for a in accs]):>7.1f}"
                      f"{np.mean([a['exp_bp'] for a in accs]):>9.1f}"
                      f"{np.mean([a['hit'] for a in accs]):>7.2f}"
                      f"{np.mean([a['total'] for a in accs]):>9.1%}"
                      f"{np.mean([a['maxdd'] for a in accs]):>8.1%}"
                      f"{np.nanmean([a['ret_dd'] for a in accs]):>8.2f}  (8 seeds)")
                print(f"{'  edge over control':<22}"
                      f"{'':>6}{'':>7}{full[0]['exp_bp'] - np.mean([a['exp_bp'] for a in accs]):>+9.1f} bp")
        print()