147 lines
6.1 KiB
Python
147 lines
6.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
|
|
from app.models.backtest import (
|
|
BacktestDailyPoint,
|
|
BacktestScenario,
|
|
BacktestSummaryMetrics,
|
|
TemplateBacktestResult,
|
|
)
|
|
from app.models.strategy_template import StrategyTemplate, TemplateLeg
|
|
from app.services.backtesting.historical_provider import DailyClosePoint, SyntheticHistoricalProvider
|
|
|
|
|
|
@dataclass
|
|
class OpenSyntheticPosition:
|
|
position_id: str
|
|
leg: TemplateLeg
|
|
strike: float
|
|
expiry: date
|
|
quantity: float
|
|
|
|
|
|
class SyntheticBacktestEngine:
|
|
def __init__(self, provider: SyntheticHistoricalProvider) -> 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
|
|
open_positions = self._open_positions(scenario, template, history, start_day)
|
|
opening_quotes = [self._mark_position(position, start_day) for position in open_positions]
|
|
opening_cost = sum(quote.mark * quote.quantity for quote in opening_quotes)
|
|
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[OpenSyntheticPosition] = []
|
|
|
|
for position in open_positions:
|
|
if day.date >= position.expiry:
|
|
intrinsic = self.provider.intrinsic_value(
|
|
option_type=position.leg.option_type,
|
|
spot=day.close,
|
|
strike=position.strike,
|
|
)
|
|
payoff = intrinsic * position.quantity
|
|
cash_balance += payoff
|
|
realized_option_cashflow += payoff
|
|
total_option_payoff_realized += payoff
|
|
continue
|
|
|
|
quote = self._mark_position(position, day)
|
|
option_market_value += quote.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),
|
|
)
|
|
)
|
|
|
|
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_threshold_breached_unhedged=any(point.margin_call_unhedged for point in daily_points),
|
|
margin_threshold_breached_hedged=any(point.margin_call_hedged for point in daily_points),
|
|
)
|
|
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),
|
|
)
|
|
|
|
def _open_positions(
|
|
self,
|
|
scenario: BacktestScenario,
|
|
template: StrategyTemplate,
|
|
history: list[DailyClosePoint],
|
|
start_day: DailyClosePoint,
|
|
) -> list[OpenSyntheticPosition]:
|
|
positions: list[OpenSyntheticPosition] = []
|
|
for index, leg in enumerate(template.legs, start=1):
|
|
expiry = self.provider.resolve_expiry(history, start_day.date, leg.target_expiry_days)
|
|
positions.append(
|
|
OpenSyntheticPosition(
|
|
position_id=f"{template.slug}-position-{index}",
|
|
leg=leg,
|
|
strike=start_day.close * leg.strike_rule.value,
|
|
expiry=expiry,
|
|
quantity=scenario.initial_portfolio.underlying_units
|
|
* leg.allocation_weight
|
|
* leg.target_coverage_pct,
|
|
)
|
|
)
|
|
return positions
|
|
|
|
def _mark_position(self, position: OpenSyntheticPosition, day: DailyClosePoint):
|
|
return self.provider.price_option(
|
|
position_id=position.position_id,
|
|
leg=position.leg,
|
|
spot=day.close,
|
|
strike=position.strike,
|
|
expiry=position.expiry,
|
|
quantity=position.quantity,
|
|
valuation_date=day.date,
|
|
)
|