| """
|
| 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)
|
ff46cfd57e