146 lines
6.1 KiB
Python
146 lines
6.1 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
|
|
underlying_value = scenario.initial_portfolio.underlying_units * day.close
|
|
net_portfolio_value = underlying_value + option_market_value + cash_balance
|
|
ltv_unhedged = scenario.initial_portfolio.loan_amount / underlying_value
|
|
ltv_hedged = scenario.initial_portfolio.loan_amount / net_portfolio_value
|
|
daily_points.append(
|
|
BacktestDailyPoint(
|
|
date=day.date,
|
|
spot_close=day.close,
|
|
underlying_value=underlying_value,
|
|
option_market_value=option_market_value,
|
|
premium_cashflow=premium_cashflow,
|
|
realized_option_cashflow=realized_option_cashflow,
|
|
net_portfolio_value=net_portfolio_value,
|
|
loan_amount=scenario.initial_portfolio.loan_amount,
|
|
ltv_unhedged=ltv_unhedged,
|
|
ltv_hedged=ltv_hedged,
|
|
margin_call_unhedged=ltv_unhedged >= scenario.initial_portfolio.margin_call_ltv,
|
|
margin_call_hedged=ltv_hedged >= 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)
|