Warrior_EA/research/portfolio.py

"""
Portfolio test: the recovered dip-z config across the instruments that showed
an edge over the long-only control, on ONE equity curve.

WHY A PORTFOLIO IS THE ANSWER TO THE PROP LIMIT
-----------------------------------------------
Per symbol at 1% risk the strategy clears cadence and beats its control but
breaks the 5% drawdown limit. Drawdown scales ~linearly with risk per trade
while ret/DD does not, so the fix is sizing, not signal. Spreading the same
risk budget across several instruments then buys back some return, but ONLY to
the extent the drawdowns are not simultaneous -- and four equity indices are
highly correlated, so that benefit must be measured, never assumed.

HONESTY NOTE ON THE EQUITY CURVE: each trade's R is applied at its EXIT, in
chronological order. With concurrent positions this understates the true
intra-trade drawdown, because two open losers are not marked to market
together. The reported maxDD is therefore a FLOOR, not a ceiling -- treat the
5% test as necessary, not sufficient.
"""

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
from run_screen import zscore_entries  # noqa: E402

STOP_ATR = 3.0
PROP_DD = 0.05


def collect(symbols, period, lo=None, hi=None, max_bars=10):
    """All trades across symbols, chronological, with R attached."""
    out = []
    for s in symbols:
        d = bt.load(s, period)
        e, xma = zscore_entries(d, 20, -1.5, 0)
        ts = d["ts"]
        m = np.ones(len(ts), bool)
        if lo is not None:
            m &= ts >= np.datetime64(lo)
        if hi is not None:
            m &= ts < np.datetime64(hi)
        tr = bt.simulate(d, e & m, side=1, exit_ma=xma, max_bars=max_bars, stop_atr=STOP_ATR)
        tr = bt.add_r(tr, d, stop_atr=STOP_ATR)
        for t in tr:
            t["symbol"] = s
            t["exit_t"] = d["ts"][t["exit_i"]]
        out += tr
    out.sort(key=lambda t: t["exit_t"])
    return out


def curve(trades, risk, max_concurrent=None):
    """Equity curve; optionally refuse entries beyond `max_concurrent` open."""
    if max_concurrent is not None:
        kept, open_until = [], []
        for t in sorted(trades, key=lambda x: x["t"]):
            open_until = [u for u in open_until if u > t["t"]]
            if len(open_until) >= max_concurrent:
                continue
            open_until.append(t["exit_t"])
            kept.append(t)
        trades = sorted(kept, key=lambda x: x["exit_t"])
    eq = [1.0]
    for t in trades:
        eq.append(eq[-1] * (1.0 + risk * t["r"]))
    eq = np.array(eq)
    peak = np.maximum.accumulate(eq)
    dd = (peak - eq) / peak
    return trades, eq, dd.max()


def report(tag, trades, risk, max_concurrent=None):
    if not trades:
        print(f"{tag:<26} no trades")
        return None
    kept, eq, maxdd = curve(trades, risk, max_concurrent)
    t0 = min(t["t"] for t in kept)
    t1 = max(t["exit_t"] for t in kept)
    years = (t1 - t0) / np.timedelta64(365, "D")
    total = eq[-1] - 1.0
    cagr = eq[-1] ** (1 / max(years, 1e-9)) - 1.0
    permo = len(kept) / max(years * 12, 1e-9)
    rets = np.array([t["ret"] for t in kept])
    rmult = np.array([t["r"] for t in kept])
    ret_dd = total / maxdd if maxdd > 1e-9 else np.nan
    flags = []
    if permo < 2.0:
        flags.append("cadence")
    if maxdd > PROP_DD:
        flags.append("maxDD")
    if not (ret_dd >= 2.0):
        flags.append("ret/DD")
    print(f"{tag:<26}{len(kept):>6}{permo:>7.1f}{rets.mean()*1e4:>9.1f}"
          f"{rmult.mean():>8.3f}{total:>9.1%}{cagr:>8.1%}{maxdd:>8.1%}{ret_dd:>8.2f}"
          f"  {'PASS' if not flags else 'fail:' + ','.join(flags)}")
    return dict(n=len(kept), permo=permo, total=total, cagr=cagr, maxdd=maxdd, ret_dd=ret_dd)


HDR = (f"{'portfolio':<26}{'n':>6}{'/mo':>7}{'exp_bp':>9}{'meanR':>8}"
       f"{'total':>9}{'CAGR':>8}{'maxDD':>8}{'ret/DD':>8}  screen")

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

if __name__ == "__main__":
    period = "PERIOD_H4"
    print("=== 4-INDEX PORTFOLIO, recovered dip-z config, H4, long only ===")
    print("(equity applies each trade's R at exit; concurrent DD is understated)\n")

    full = collect(IDX, period)
    print(HDR)
    for risk in (0.010, 0.0075, 0.005, 0.0035, 0.0025):
        report(f"4 idx, risk {risk:.2%}", full, risk)
    print()
    print("--- risk 0.5%, capped concurrency (correlated indices draw down together) ---")
    print(HDR)
    for mc in (4, 3, 2, 1):
        report(f"4 idx, 0.50%, max {mc} open", full, 0.005, max_concurrent=mc)
    print()
    print("--- IS / OOS at the sizing that passes ---")
    print(HDR)
    report("4 idx IS  (..2024)", collect(IDX, period, None, SPLIT), 0.005)
    report("4 idx OOS (2024..)", collect(IDX, period, SPLIT, None), 0.005)
    print()
    print("--- all 7 symbols (incl. the three with no measured edge) ---")
    print(HDR)
    report("7 sym, risk 0.50%", collect(ALL7, period), 0.005)
    report("7 sym IS  (..2024)", collect(ALL7, period, None, SPLIT), 0.005)
    report("7 sym OOS (2024..)", collect(ALL7, period, SPLIT, None), 0.005)