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.
250 lines
10 KiB
Python
250 lines
10 KiB
Python
"""Portfolio contribution analysis for multi-engine ORB strategies.
|
|
|
|
Runs two strategy sleeves independently, then computes portfolio-level
|
|
metrics that measure how well the new engine diversifies V23:
|
|
- Daily PnL correlation
|
|
- Trade overlap (same ticker+date)
|
|
- Combined equity curve and drawdown
|
|
- Worst-20% day relief (how much the new engine helps on V23's bad days)
|
|
|
|
Usage:
|
|
python -m apps.intraday_bt.portfolio_report \\
|
|
--base configs/intraday/strategies/orb_gainers_v23.yaml \\
|
|
--new configs/intraday/strategies/orb_pullback_v1.yaml \\
|
|
--days 600
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import math
|
|
import statistics
|
|
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
|
|
|
|
|
|
# ── Stats helpers ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def _pearson_corr(xs: list[float], ys: list[float]) -> float:
|
|
n = len(xs)
|
|
if n < 2:
|
|
return float("nan")
|
|
mx, my = statistics.mean(xs), statistics.mean(ys)
|
|
cov = sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / (n - 1)
|
|
sx = statistics.stdev(xs)
|
|
sy = statistics.stdev(ys)
|
|
if sx == 0 or sy == 0:
|
|
return float("nan")
|
|
return cov / (sx * sy)
|
|
|
|
|
|
def _max_drawdown(equity_curve: list[float]) -> float:
|
|
peak = equity_curve[0]
|
|
max_dd = 0.0
|
|
for eq in equity_curve:
|
|
if eq > peak:
|
|
peak = eq
|
|
dd = (eq - peak) / peak
|
|
if dd < max_dd:
|
|
max_dd = dd
|
|
return max_dd
|
|
|
|
|
|
def _build_equity_curve(dates: list[str], pnl_by_date: dict[str, float], initial: float) -> list[float]:
|
|
curve = [initial]
|
|
eq = initial
|
|
for d in dates:
|
|
eq += pnl_by_date.get(d, 0.0)
|
|
curve.append(eq)
|
|
return curve
|
|
|
|
|
|
# ── Analysis ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def analyze(
|
|
base_day_results: list,
|
|
new_day_results: list,
|
|
base_name: str,
|
|
new_name: str,
|
|
base_capital: float,
|
|
new_capital: float,
|
|
) -> None:
|
|
# Build daily PnL dicts
|
|
base_pnl: dict[str, float] = {}
|
|
new_pnl: dict[str, float] = {}
|
|
base_trades_by_date: dict[str, list] = {}
|
|
new_trades_by_date: dict[str, list] = {}
|
|
|
|
for dr in base_day_results:
|
|
base_pnl[dr.date] = dr.daily_pnl
|
|
base_trades_by_date[dr.date] = dr.trades or []
|
|
for dr in new_day_results:
|
|
new_pnl[dr.date] = dr.daily_pnl
|
|
new_trades_by_date[dr.date] = dr.trades or []
|
|
|
|
all_dates = sorted(set(base_pnl.keys()) | set(new_pnl.keys()))
|
|
if not all_dates:
|
|
print("No data to analyze.")
|
|
return
|
|
|
|
# Daily PnL vectors (only dates where BOTH sleeves traded or had any activity)
|
|
common_dates = sorted(set(base_pnl.keys()) & set(new_pnl.keys()))
|
|
|
|
# Correlation
|
|
if common_dates:
|
|
xs = [base_pnl[d] for d in common_dates]
|
|
ys = [new_pnl[d] for d in common_dates]
|
|
corr = _pearson_corr(xs, ys)
|
|
else:
|
|
corr = float("nan")
|
|
|
|
# Trade overlap: same (ticker, date) pairs
|
|
base_pairs: set[tuple[str, str]] = set()
|
|
new_pairs: set[tuple[str, str]] = set()
|
|
for date, trades in base_trades_by_date.items():
|
|
for t in trades:
|
|
base_pairs.add((t.ticker, date))
|
|
for date, trades in new_trades_by_date.items():
|
|
for t in trades:
|
|
new_pairs.add((t.ticker, date))
|
|
overlap_count = len(base_pairs & new_pairs)
|
|
overlap_pct = overlap_count / min(len(base_pairs), len(new_pairs)) if min(len(base_pairs), len(new_pairs)) > 0 else 0.0
|
|
|
|
# Combined equity curve and DD
|
|
total_capital = base_capital + new_capital
|
|
combined_pnl_by_date = {d: base_pnl.get(d, 0.0) + new_pnl.get(d, 0.0) for d in all_dates}
|
|
base_curve = _build_equity_curve(all_dates, base_pnl, base_capital)
|
|
new_curve = _build_equity_curve(all_dates, new_pnl, new_capital)
|
|
combined_curve = _build_equity_curve(all_dates, combined_pnl_by_date, total_capital)
|
|
|
|
base_dd = _max_drawdown(base_curve)
|
|
new_dd = _max_drawdown(new_curve)
|
|
combined_dd = _max_drawdown(combined_curve)
|
|
|
|
base_return = (base_curve[-1] - base_capital) / base_capital
|
|
new_return = (new_curve[-1] - new_capital) / new_capital
|
|
combined_return = (combined_curve[-1] - total_capital) / total_capital
|
|
|
|
# Worst-20% day relief
|
|
base_trading_dates = [d for d in all_dates if d in base_pnl]
|
|
if base_trading_dates:
|
|
sorted_by_pnl = sorted(base_trading_dates, key=lambda d: base_pnl[d])
|
|
n_worst = max(1, len(sorted_by_pnl) // 5)
|
|
worst_dates = sorted_by_pnl[:n_worst]
|
|
base_worst_sum = sum(base_pnl[d] for d in worst_dates)
|
|
combined_worst_sum = sum(base_pnl.get(d, 0.0) + new_pnl.get(d, 0.0) for d in worst_dates)
|
|
relief_pct = (combined_worst_sum - base_worst_sum) / base_capital if base_capital > 0 else 0.0
|
|
else:
|
|
worst_dates = []
|
|
base_worst_sum = combined_worst_sum = relief_pct = 0.0
|
|
|
|
# Sharpe
|
|
def _sharpe(curve: list[float], capital: float) -> float:
|
|
rets = [(curve[i + 1] - curve[i]) / curve[i] for i in range(len(curve) - 1) if curve[i] > 0]
|
|
if len(rets) < 2:
|
|
return 0.0
|
|
return statistics.mean(rets) / statistics.stdev(rets) * math.sqrt(252)
|
|
|
|
# Trade counts
|
|
base_trade_count = sum(len(v) for v in base_trades_by_date.values())
|
|
new_trade_count = sum(len(v) for v in new_trades_by_date.values())
|
|
base_wr = (
|
|
sum(1 for v in base_trades_by_date.values() for t in v if t.pnl > 0) / base_trade_count
|
|
if base_trade_count > 0 else 0.0
|
|
)
|
|
new_wr = (
|
|
sum(1 for v in new_trades_by_date.values() for t in v if t.pnl > 0) / new_trade_count
|
|
if new_trade_count > 0 else 0.0
|
|
)
|
|
|
|
# Print report
|
|
print("\n" + "=" * 65)
|
|
print(" PORTFOLIO CONTRIBUTION REPORT")
|
|
print("=" * 65)
|
|
print(f" Period: {all_dates[0]} → {all_dates[-1]} ({len(all_dates)} calendar days)")
|
|
print()
|
|
print(f" {'Metric':<35} {'Base':>10} {'New':>10} {'Combined':>10}")
|
|
print(f" {'-'*35} {'-'*10} {'-'*10} {'-'*10}")
|
|
print(f" {'Capital':<35} {'${:,.0f}'.format(base_capital):>10} {'${:,.0f}'.format(new_capital):>10} {'${:,.0f}'.format(total_capital):>10}")
|
|
print(f" {'Total return':<35} {base_return*100:>+9.2f}% {new_return*100:>+9.2f}% {combined_return*100:>+9.2f}%")
|
|
print(f" {'Sharpe':<35} {_sharpe(base_curve, base_capital):>10.3f} {_sharpe(new_curve, new_capital):>10.3f} {_sharpe(combined_curve, total_capital):>10.3f}")
|
|
print(f" {'Max DD':<35} {base_dd*100:>+9.2f}% {new_dd*100:>+9.2f}% {combined_dd*100:>+9.2f}%")
|
|
print(f" {'Trades':<35} {base_trade_count:>10} {new_trade_count:>10} {base_trade_count+new_trade_count:>10}")
|
|
print(f" {'Win rate':<35} {base_wr*100:>9.1f}% {new_wr*100:>9.1f}%")
|
|
print()
|
|
print(f" Daily PnL correlation (common dates: {len(common_dates)}): {corr:+.4f}")
|
|
print(f" {' → Gate: corr ≤ 0.30':<45} {'PASS' if corr <= 0.30 else 'FAIL':>5}")
|
|
print()
|
|
print(f" Trade overlap: {overlap_count} shared (ticker, date) pairs")
|
|
print(f" = {overlap_pct*100:.1f}% of min(|base|, |new|) trade set")
|
|
print(f" {' → Gate: overlap ≤ 20%':<45} {'PASS' if overlap_pct <= 0.20 else 'FAIL':>5}")
|
|
print()
|
|
print(f" DD improvement (combined vs base): {(combined_dd - base_dd)*100:+.2f}pp")
|
|
print(f" {' → Gate: DD improvement ≥ 5pp':<45} {'PASS' if combined_dd - base_dd <= -0.05 else 'FAIL':>5}")
|
|
print()
|
|
print(f" Worst-20% day relief ({len(worst_dates)} days):")
|
|
print(f" Base worst sum: ${base_worst_sum:+.2f}")
|
|
print(f" Combined worst sum: ${combined_worst_sum:+.2f}")
|
|
print(f" Relief: {relief_pct*100:+.2f}% of base capital")
|
|
print(f" {' → Gate: combined > base on worst days':<45} {'PASS' if combined_worst_sum > base_worst_sum else 'FAIL':>5}")
|
|
print()
|
|
print(" G1 standalone gates (new engine):")
|
|
print(f" trades ≥ 50: {new_trade_count:4d} {'PASS' if new_trade_count >= 50 else 'FAIL'}")
|
|
print(f" WR ≥ 50%: {new_wr*100:5.1f}% {'PASS' if new_wr >= 0.50 else 'FAIL'}")
|
|
print(f" return ≥ 0%: {new_return*100:+5.2f}% {'PASS' if new_return >= 0 else 'FAIL'}")
|
|
print(f" max_dd ≥ -20%: {new_dd*100:+5.2f}% {'PASS' if new_dd >= -0.20 else 'FAIL'}")
|
|
print("=" * 65)
|
|
|
|
|
|
# ── Entry point ───────────────────────────────────────────────────────────
|
|
|
|
|
|
async def _main(args: argparse.Namespace) -> None:
|
|
days = args.days
|
|
total_cap = args.total_capital
|
|
base_weight = args.base_weight
|
|
new_weight = 1.0 - base_weight
|
|
|
|
base_config = load_config(args.base)
|
|
new_config = load_config(args.new)
|
|
|
|
base_capital = total_cap * base_weight
|
|
new_capital = total_cap * new_weight
|
|
|
|
print(f"\nRunning base sleeve: {Path(args.base).stem} (capital ${base_capital:,.0f})")
|
|
base_sleeve = _build_sleeve_config(base_config, base_capital, days)
|
|
base_day_results, base_metrics, _, _ = await run_sleeve(base_sleeve)
|
|
|
|
print(f"\nRunning new sleeve: {Path(args.new).stem} (capital ${new_capital:,.0f})")
|
|
new_sleeve = _build_sleeve_config(new_config, new_capital, days)
|
|
new_day_results, new_metrics, _, _ = await run_sleeve(new_sleeve)
|
|
|
|
analyze(
|
|
base_day_results=base_day_results,
|
|
new_day_results=new_day_results,
|
|
base_name=Path(args.base).stem,
|
|
new_name=Path(args.new).stem,
|
|
base_capital=base_capital,
|
|
new_capital=new_capital,
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Portfolio contribution report for multi-engine ORB")
|
|
parser.add_argument("--base", required=True, help="Base strategy YAML (e.g. V23)")
|
|
parser.add_argument("--new", required=True, help="New engine YAML (e.g. orb_pullback_v1)")
|
|
parser.add_argument("--days", type=int, default=600)
|
|
parser.add_argument("--total-capital", type=float, default=10_000.0, dest="total_capital")
|
|
parser.add_argument("--base-weight", type=float, default=0.60, dest="base_weight",
|
|
help="Fraction of capital allocated to base sleeve (default 0.60)")
|
|
args = parser.parse_args()
|
|
asyncio.run(_main(args))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|