You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""Slippage stress test for any ORB strategy.
|
|
|
|
Runs the same backtest at increasing slippage levels to find the break-even
|
|
slippage point. Useful for validating that a strategy's edge survives realistic
|
|
transaction costs.
|
|
|
|
Usage:
|
|
python -m apps.intraday_bt.scripts.run_slippage_ladder \\
|
|
--config configs/intraday/strategies/orb_pullback_v1.yaml \\
|
|
--days 200 \\
|
|
--bps 5,10,15,25
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
from apps.intraday_bt.composite import _build_sleeve_config
|
|
from apps.intraday_bt.run import load_config, run as run_sleeve
|
|
|
|
|
|
async def _main(args: argparse.Namespace) -> None:
|
|
bps_values = [float(x.strip()) for x in args.bps.split(",")]
|
|
base_config = load_config(args.config)
|
|
days = args.days
|
|
capital = args.capital
|
|
name = Path(args.config).stem
|
|
|
|
print(f"\nSlippage ladder: {name} ({days}d, ${capital:,.0f})")
|
|
print(f"{'BPS':>6} {'Return':>9} {'Sharpe':>7} {'MaxDD':>8} {'WR':>7} {'Trades':>7}")
|
|
print("-" * 52)
|
|
|
|
baseline_return = None
|
|
for bps in bps_values:
|
|
sleeve = _build_sleeve_config(base_config, capital, days)
|
|
# Override slippage_bps
|
|
params = sleeve.orb_strategy
|
|
if params is not None:
|
|
params_dict = params.model_dump()
|
|
params_dict["slippage_bps"] = bps
|
|
from libs.intraday.domain import ORBStrategyParams
|
|
sleeve_config_dict = sleeve.model_dump()
|
|
sleeve_config_dict["orb_strategy"] = params_dict
|
|
from libs.intraday.domain import IntradayConfig
|
|
sleeve = IntradayConfig(**sleeve_config_dict)
|
|
|
|
_, metrics, _, _ = await run_sleeve(sleeve)
|
|
ret = (metrics.total_return_pct or 0) * 100
|
|
sharpe = metrics.sharpe_ratio or 0.0
|
|
dd = (metrics.max_drawdown_pct or 0) * 100
|
|
wr = (metrics.win_rate or 0) * 100
|
|
trades = metrics.total_trades or 0
|
|
|
|
if baseline_return is None:
|
|
baseline_return = ret
|
|
|
|
degradation = ret - baseline_return
|
|
deg_str = f"({degradation:+.1f}pp)" if bps != bps_values[0] else " "
|
|
|
|
print(f"{bps:>6.0f} {ret:>+8.2f}% {sharpe:>7.3f} {dd:>+7.2f}% {wr:>6.1f}% {trades:>7} {deg_str}")
|
|
|
|
# Find break-even (last bps where return > 0)
|
|
print()
|
|
print(f" Run with --bps ladder to find break-even slippage.")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Slippage ladder stress test")
|
|
parser.add_argument("--config", required=True)
|
|
parser.add_argument("--days", type=int, default=200)
|
|
parser.add_argument("--capital", type=float, default=10_000.0)
|
|
parser.add_argument("--bps", default="5,10,15,25",
|
|
help="Comma-separated slippage bps values (default: 5,10,15,25)")
|
|
args = parser.parse_args()
|
|
asyncio.run(_main(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|