a8e710f790
- Extend DailyClosePoint to include low, high, open (optional) - Update Databento source to extract OHLC data from ohlcv-1d schema - Update YFinance source to extract Low, High, Open from history - Modify backtest engine to use worst-case (low) price for margin call detection This ensures margin calls are evaluated at the day's worst price, not just the closing price, providing more realistic risk assessment.
161 lines
6.9 KiB
Python
161 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from app.models.backtest import (
|
|
BacktestDailyPoint,
|
|
BacktestScenario,
|
|
BacktestSummaryMetrics,
|
|
TemplateBacktestResult,
|
|
)
|
|
from app.models.strategy_template import StrategyTemplate
|
|
from app.services.backtesting.historical_provider import (
|
|
BacktestHistoricalProvider,
|
|
DailyClosePoint,
|
|
HistoricalOptionPosition,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class OpenSyntheticPosition(HistoricalOptionPosition):
|
|
pass
|
|
|
|
|
|
class SyntheticBacktestEngine:
|
|
def __init__(self, provider: BacktestHistoricalProvider) -> None:
|
|
self.provider = provider
|
|
|
|
def run_template(
|
|
self,
|
|
scenario: BacktestScenario,
|
|
template: StrategyTemplate,
|
|
history: list[DailyClosePoint],
|
|
) -> TemplateBacktestResult:
|
|
start_day = history[0]
|
|
cash_balance = scenario.initial_portfolio.cash_balance
|
|
total_hedge_cost = 0.0
|
|
total_option_payoff_realized = 0.0
|
|
warnings: list[str] = []
|
|
open_positions = self._open_positions(scenario, template, history, start_day)
|
|
opening_cost = sum(position.entry_price * position.quantity for position in open_positions)
|
|
cash_balance -= opening_cost
|
|
total_hedge_cost += opening_cost
|
|
|
|
daily_points: list[BacktestDailyPoint] = []
|
|
for day in history:
|
|
premium_cashflow = -opening_cost if day.date == start_day.date else 0.0
|
|
realized_option_cashflow = 0.0
|
|
option_market_value = 0.0
|
|
active_position_ids: list[str] = []
|
|
remaining_positions: list[HistoricalOptionPosition] = []
|
|
|
|
for position in open_positions:
|
|
valuation = self.provider.mark_position(
|
|
position,
|
|
symbol=scenario.symbol,
|
|
as_of_date=day.date,
|
|
spot=day.close,
|
|
)
|
|
self._append_warning(warnings, valuation.warning)
|
|
if not valuation.is_active:
|
|
cash_balance += valuation.realized_cashflow
|
|
realized_option_cashflow += valuation.realized_cashflow
|
|
total_option_payoff_realized += valuation.realized_cashflow
|
|
continue
|
|
|
|
option_market_value += valuation.mark * position.quantity
|
|
active_position_ids.append(position.position_id)
|
|
remaining_positions.append(position)
|
|
|
|
open_positions = remaining_positions
|
|
|
|
# Use closing price for portfolio value calculations
|
|
underlying_value_close = scenario.initial_portfolio.underlying_units * day.close
|
|
net_portfolio_value_close = underlying_value_close + option_market_value + cash_balance
|
|
|
|
# Use day's low for margin call evaluation (worst case during the day)
|
|
# If low is not available, fall back to close
|
|
worst_price = day.low if day.low is not None else day.close
|
|
underlying_value_worst = scenario.initial_portfolio.underlying_units * worst_price
|
|
net_portfolio_value_worst = underlying_value_worst + option_market_value + cash_balance
|
|
|
|
# LTVs for display (end-of-day at close)
|
|
ltv_unhedged = scenario.initial_portfolio.loan_amount / underlying_value_close
|
|
ltv_hedged = scenario.initial_portfolio.loan_amount / net_portfolio_value_close
|
|
|
|
# Margin calls use worst-case (low price) scenario
|
|
ltv_unhedged_worst = scenario.initial_portfolio.loan_amount / underlying_value_worst
|
|
ltv_hedged_worst = scenario.initial_portfolio.loan_amount / net_portfolio_value_worst
|
|
|
|
daily_points.append(
|
|
BacktestDailyPoint(
|
|
date=day.date,
|
|
spot_close=day.close,
|
|
underlying_value=underlying_value_close,
|
|
option_market_value=option_market_value,
|
|
premium_cashflow=premium_cashflow,
|
|
realized_option_cashflow=realized_option_cashflow,
|
|
net_portfolio_value=net_portfolio_value_close,
|
|
loan_amount=scenario.initial_portfolio.loan_amount,
|
|
ltv_unhedged=ltv_unhedged,
|
|
ltv_hedged=ltv_hedged,
|
|
margin_call_unhedged=ltv_unhedged_worst >= scenario.initial_portfolio.margin_call_ltv,
|
|
margin_call_hedged=ltv_hedged_worst >= scenario.initial_portfolio.margin_call_ltv,
|
|
active_position_ids=tuple(active_position_ids),
|
|
)
|
|
)
|
|
|
|
margin_call_days_unhedged = sum(1 for point in daily_points if point.margin_call_unhedged)
|
|
margin_call_days_hedged = sum(1 for point in daily_points if point.margin_call_hedged)
|
|
summary = BacktestSummaryMetrics(
|
|
start_value=scenario.initial_portfolio.start_value,
|
|
end_value_unhedged=daily_points[-1].underlying_value,
|
|
end_value_hedged_net=daily_points[-1].net_portfolio_value,
|
|
total_hedge_cost=total_hedge_cost,
|
|
total_option_payoff_realized=total_option_payoff_realized,
|
|
max_ltv_unhedged=max(point.ltv_unhedged for point in daily_points),
|
|
max_ltv_hedged=max(point.ltv_hedged for point in daily_points),
|
|
margin_call_days_unhedged=margin_call_days_unhedged,
|
|
margin_call_days_hedged=margin_call_days_hedged,
|
|
margin_threshold_breached_unhedged=margin_call_days_unhedged > 0,
|
|
margin_threshold_breached_hedged=margin_call_days_hedged > 0,
|
|
)
|
|
return TemplateBacktestResult(
|
|
template_slug=template.slug,
|
|
template_id=template.template_id,
|
|
template_version=template.version,
|
|
template_name=template.display_name,
|
|
summary_metrics=summary,
|
|
daily_path=tuple(daily_points),
|
|
warnings=tuple(warnings),
|
|
)
|
|
|
|
def _open_positions(
|
|
self,
|
|
scenario: BacktestScenario,
|
|
template: StrategyTemplate,
|
|
history: list[DailyClosePoint],
|
|
start_day: DailyClosePoint,
|
|
) -> list[HistoricalOptionPosition]:
|
|
positions: list[HistoricalOptionPosition] = []
|
|
for index, leg in enumerate(template.legs, start=1):
|
|
positions.append(
|
|
self.provider.open_position(
|
|
symbol=scenario.symbol,
|
|
leg=leg,
|
|
position_id=f"{template.slug}-position-{index}",
|
|
quantity=(
|
|
scenario.initial_portfolio.underlying_units * leg.allocation_weight * leg.target_coverage_pct
|
|
),
|
|
as_of_date=start_day.date,
|
|
spot=start_day.close,
|
|
trading_days=history,
|
|
)
|
|
)
|
|
return positions
|
|
|
|
@staticmethod
|
|
def _append_warning(warnings: list[str], warning: str | None) -> None:
|
|
if warning and warning not in warnings:
|
|
warnings.append(warning)
|