from fastapi import APIRouter from pydantic import BaseModel from typing import Optional, List import yfinance as yf import numpy as np from services.options_pricer import black_scholes from services.backtest_strategies import STRATEGIES, build_legs, synthetic_expiry router = APIRouter(prefix="/api/backtest", tags=["backtest"]) @router.get("/symbols") def backtest_symbols(): """Underlyings actually tracked via Saxo (Config → Instruments Watchlist, linked to a real Saxo options chain) — same yfinance-compatible `ticker` field used everywhere else in the app, so pricing here still runs off yfinance's long history, but the choices on screen match what's genuinely tradeable rather than an arbitrary ETF list.""" from services.database import get_instruments_watchlist return [ {"ticker": w["ticker"], "name": w["name"]} for w in get_instruments_watchlist() if w.get("saxo_option_symbol") ] @router.get("/strategies") def backtest_strategies(): return [{"key": k, "label": label, "n_legs": n} for k, label, n in STRATEGIES] class BacktestRequest(BaseModel): symbol: str start_date: str end_date: str strategy: str strike_offset_pct: float = 0.05 # e.g. 5% OTM — used by the 6 direct (non-template) strategies expiry_days: int = 90 capital: float = 1000.0 def _settle_leg(leg: dict, near_days: int, S_settle: float, sigma: float, r: float) -> float: """Value one leg at the near expiry: intrinsic if it expires there too (the common case), else a fresh Black-Scholes price for its remaining time (calendar/diagonal's far leg — closed alongside the near leg rather than held to its own later expiry, the standard way these are actually managed).""" remaining_days = leg["days_to_expiry"] - near_days if remaining_days <= 0: if leg["option_type"] == "call": return max(0.0, S_settle - leg["strike"]) return max(0.0, leg["strike"] - S_settle) T = remaining_days / 365 return float(black_scholes(S_settle, leg["strike"], T, r, sigma, leg["option_type"])["price"]) @router.post("/run") def run_backtest(req: BacktestRequest): try: ticker = yf.Ticker(req.symbol) hist = ticker.history(start=req.start_date, end=req.end_date, interval="1d") if hist.empty or len(hist) < 20: return {"error": "Insufficient data for the period"} hist = hist.reset_index() returns = np.log(hist["Close"] / hist["Close"].shift(1)).dropna() far_days = req.expiry_days * 2 # calendar/diagonal's far leg, closed alongside the near leg trades = [] equity = [req.capital] capital = req.capital r = 0.05 step = max(1, req.expiry_days // 3) for i in range(0, len(hist) - req.expiry_days, step): row = hist.iloc[i] S = float(row["Close"]) date_str = str(row["Date"])[:10] sigma_window = returns.iloc[max(0, i - 30):i] if len(sigma_window) < 5: continue sigma = float(sigma_window.std() * np.sqrt(252)) if sigma < 0.01: sigma = 0.20 near_expiry = synthetic_expiry(date_str, req.expiry_days, S) far_expiry = synthetic_expiry(date_str, far_days, S) if req.strategy in ("calendar_spread", "diagonal_spread") else None legs = build_legs(req.strategy, S, req.strike_offset_pct, near_expiry, far_expiry) if not legs: continue entry_premiums = [] for leg in legs: T = leg["days_to_expiry"] / 365 premium = float(black_scholes(S, leg["strike"], T, r, sigma, leg["option_type"])["price"]) entry_premiums.append(premium) signed_qty = [(1 if leg["position"] == "long" else -1) * leg["quantity"] for leg in legs] net_premium = sum(sq * p for sq, p in zip(signed_qty, entry_premiums)) # >0 debit, <0 credit risk_basis = max(abs(net_premium), 0.05 * S) contracts = max(1, int((capital * 0.1) / (risk_basis * 100))) cost = net_premium * contracts * 100 expiry_idx = min(i + req.expiry_days, len(hist) - 1) S_expiry = float(hist.iloc[expiry_idx]["Close"]) date_expiry = str(hist.iloc[expiry_idx]["Date"])[:10] exit_values = [_settle_leg(leg, req.expiry_days, S_expiry, sigma, r) for leg in legs] exit_signed_value = sum(sq * v for sq, v in zip(signed_qty, exit_values)) pnl = (exit_signed_value - net_premium) * contracts * 100 capital += pnl equity.append(round(capital, 2)) trades.append({ "entry_date": date_str, "exit_date": date_expiry, "strategy": req.strategy, "S_entry": round(S, 2), "S_expiry": round(S_expiry, 2), "legs": [ {"strike": round(leg["strike"], 2), "option_type": leg["option_type"], "position": leg["position"], "quantity": leg["quantity"], "days_to_expiry": leg["days_to_expiry"]} for leg in legs ], "net_premium": round(net_premium, 4), "contracts": contracts, "cost": round(cost, 2), "pnl": round(pnl, 2), "capital": round(capital, 2), }) if not trades: return {"error": "No trades generated"} wins = [t for t in trades if t["pnl"] > 0] losses = [t for t in trades if t["pnl"] <= 0] total_pnl = sum(t["pnl"] for t in trades) gross_profit = sum(t["pnl"] for t in wins) if wins else 0 gross_loss = abs(sum(t["pnl"] for t in losses)) if losses else 1 eq = np.array(equity) peak = np.maximum.accumulate(eq) drawdown = (eq - peak) / peak max_dd = float(drawdown.min()) * 100 equity_curve = [{"index": i, "capital": v} for i, v in enumerate(equity)] return { "symbol": req.symbol, "strategy": req.strategy, "period": f"{req.start_date} → {req.end_date}", "total_trades": len(trades), "wins": len(wins), "losses": len(losses), "win_rate": round(len(wins) / len(trades) * 100, 1) if trades else 0, "total_pnl": round(total_pnl, 2), "total_return_pct": round((capital - req.capital) / req.capital * 100, 2), "max_drawdown_pct": round(max_dd, 2), "profit_factor": round(gross_profit / gross_loss, 2) if gross_loss else 0, "final_capital": round(capital, 2), "equity_curve": equity_curve, "trades": trades[-20:], } except Exception as e: return {"error": str(e)}