321 lines
13 KiB
Python
321 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
|
|
from app.models.backtest import BacktestScenario, EventComparisonRanking, EventComparisonReport
|
|
from app.services.backtesting.comparison import EventComparisonService
|
|
from app.services.backtesting.historical_provider import DailyClosePoint, SyntheticHistoricalProvider
|
|
from app.services.backtesting.input_normalization import normalize_historical_scenario_inputs
|
|
from app.services.event_presets import EventPresetService
|
|
from app.services.strategy_templates import StrategyTemplateService
|
|
|
|
SUPPORTED_EVENT_COMPARISON_SYMBOL = "GLD"
|
|
|
|
|
|
def _validate_initial_collateral(underlying_units: float, entry_spot: float, loan_amount: float) -> None:
|
|
initial_collateral_value = underlying_units * entry_spot
|
|
if loan_amount >= initial_collateral_value:
|
|
raise ValueError(
|
|
"Historical scenario starts undercollateralized: "
|
|
f"loan ${loan_amount:,.0f} exceeds initial collateral ${initial_collateral_value:,.0f} "
|
|
f"at entry spot ${entry_spot:,.2f}. Reduce loan amount or increase underlying units."
|
|
)
|
|
|
|
|
|
DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY: tuple[DailyClosePoint, ...] = (
|
|
DailyClosePoint(date=date(2024, 1, 2), close=100.0),
|
|
DailyClosePoint(date=date(2024, 1, 3), close=96.0),
|
|
DailyClosePoint(date=date(2024, 1, 4), close=92.0),
|
|
DailyClosePoint(date=date(2024, 1, 5), close=88.0),
|
|
DailyClosePoint(date=date(2024, 1, 8), close=85.0),
|
|
)
|
|
|
|
FIXTURE_HISTORY_START = DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY[0].date
|
|
FIXTURE_HISTORY_END = DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY[-1].date
|
|
|
|
|
|
class EventComparisonFixtureHistoricalPriceSource:
|
|
def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]:
|
|
normalized_symbol = symbol.strip().upper()
|
|
if normalized_symbol != SUPPORTED_EVENT_COMPARISON_SYMBOL:
|
|
raise ValueError(
|
|
"BT-003A deterministic fixture data only supports GLD event-comparison presets on this page"
|
|
)
|
|
if start_date < FIXTURE_HISTORY_START or end_date > FIXTURE_HISTORY_END:
|
|
raise ValueError(
|
|
"BT-003A deterministic fixture data only supports the seeded 2024-01-02 through 2024-01-08 window"
|
|
)
|
|
return [
|
|
point for point in DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY if start_date <= point.date <= end_date
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventComparisonChartSeries:
|
|
name: str
|
|
values: tuple[float, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventComparisonChartModel:
|
|
dates: tuple[str, ...]
|
|
series: tuple[EventComparisonChartSeries, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventComparisonDrilldownRow:
|
|
date: str
|
|
spot_close: float
|
|
net_portfolio_value: float
|
|
option_market_value: float
|
|
realized_option_cashflow: float
|
|
ltv_unhedged: float
|
|
ltv_hedged: float
|
|
margin_call_hedged: bool
|
|
active_position_ids: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EventComparisonDrilldownModel:
|
|
rank: int
|
|
template_slug: str
|
|
template_name: str
|
|
survived_margin_call: bool
|
|
margin_call_days_hedged: int
|
|
total_option_payoff_realized: float
|
|
hedge_cost: float
|
|
final_equity: float
|
|
worst_ltv_hedged: float
|
|
worst_ltv_date: str | None
|
|
breach_dates: tuple[str, ...]
|
|
rows: tuple[EventComparisonDrilldownRow, ...]
|
|
|
|
|
|
class EventComparisonPageService:
|
|
def __init__(
|
|
self,
|
|
comparison_service: EventComparisonService | None = None,
|
|
event_preset_service: EventPresetService | None = None,
|
|
template_service: StrategyTemplateService | None = None,
|
|
) -> None:
|
|
self.event_preset_service = event_preset_service or EventPresetService()
|
|
self.template_service = template_service or StrategyTemplateService()
|
|
if comparison_service is None:
|
|
provider = SyntheticHistoricalProvider(source=EventComparisonFixtureHistoricalPriceSource())
|
|
comparison_service = EventComparisonService(
|
|
provider=provider,
|
|
event_preset_service=self.event_preset_service,
|
|
template_service=self.template_service,
|
|
)
|
|
self.comparison_service = comparison_service
|
|
|
|
def preset_options(self, symbol: str = SUPPORTED_EVENT_COMPARISON_SYMBOL) -> list[dict[str, object]]:
|
|
return [
|
|
{
|
|
"slug": preset.slug,
|
|
"label": preset.display_name,
|
|
"description": preset.description,
|
|
"default_template_slugs": list(preset.scenario_overrides.default_template_slugs),
|
|
}
|
|
for preset in self.event_preset_service.list_presets(symbol)
|
|
]
|
|
|
|
def template_options(self, symbol: str = SUPPORTED_EVENT_COMPARISON_SYMBOL) -> list[dict[str, object]]:
|
|
return [
|
|
{
|
|
"slug": template.slug,
|
|
"label": template.display_name,
|
|
"description": template.description,
|
|
}
|
|
for template in self.template_service.list_active_templates(symbol)
|
|
]
|
|
|
|
def default_template_selection(self, preset_slug: str) -> tuple[str, ...]:
|
|
preset = self.event_preset_service.get_preset(preset_slug)
|
|
return tuple(preset.scenario_overrides.default_template_slugs)
|
|
|
|
def preview_scenario(
|
|
self,
|
|
*,
|
|
preset_slug: str,
|
|
template_slugs: tuple[str, ...],
|
|
underlying_units: float,
|
|
loan_amount: float,
|
|
margin_call_ltv: float,
|
|
) -> BacktestScenario:
|
|
if not template_slugs:
|
|
raise ValueError("Select at least one strategy template.")
|
|
normalized_inputs = normalize_historical_scenario_inputs(
|
|
underlying_units=underlying_units,
|
|
loan_amount=loan_amount,
|
|
margin_call_ltv=margin_call_ltv,
|
|
)
|
|
try:
|
|
scenario = self.comparison_service.preview_scenario_from_inputs(
|
|
preset_slug=preset_slug,
|
|
template_slugs=template_slugs,
|
|
underlying_units=normalized_inputs.underlying_units,
|
|
loan_amount=normalized_inputs.loan_amount,
|
|
margin_call_ltv=normalized_inputs.margin_call_ltv,
|
|
)
|
|
except ValueError as exc:
|
|
if str(exc) == "loan_amount must be less than initial collateral value":
|
|
preset = self.event_preset_service.get_preset(preset_slug)
|
|
preview = self.comparison_service.provider.load_history(
|
|
preset.symbol.strip().upper(),
|
|
preset.window_start,
|
|
preset.window_end,
|
|
)
|
|
if preview:
|
|
_validate_initial_collateral(
|
|
normalized_inputs.underlying_units,
|
|
preview[0].close,
|
|
normalized_inputs.loan_amount,
|
|
)
|
|
raise
|
|
_validate_initial_collateral(
|
|
normalized_inputs.underlying_units,
|
|
scenario.initial_portfolio.entry_spot,
|
|
normalized_inputs.loan_amount,
|
|
)
|
|
return scenario
|
|
|
|
def run_read_only_comparison(
|
|
self,
|
|
*,
|
|
preset_slug: str,
|
|
template_slugs: tuple[str, ...],
|
|
underlying_units: float,
|
|
loan_amount: float,
|
|
margin_call_ltv: float,
|
|
) -> EventComparisonReport:
|
|
if not preset_slug:
|
|
raise ValueError("Preset selection is required")
|
|
if not template_slugs:
|
|
raise ValueError("Select at least one strategy template.")
|
|
|
|
normalized_inputs = normalize_historical_scenario_inputs(
|
|
underlying_units=underlying_units,
|
|
loan_amount=loan_amount,
|
|
margin_call_ltv=margin_call_ltv,
|
|
)
|
|
|
|
preset = self.event_preset_service.get_preset(preset_slug)
|
|
normalized_symbol = preset.symbol.strip().upper()
|
|
if normalized_symbol != SUPPORTED_EVENT_COMPARISON_SYMBOL:
|
|
raise ValueError("BT-003A event comparison is currently limited to GLD on this page")
|
|
|
|
try:
|
|
preview = self.comparison_service.preview_scenario_from_inputs(
|
|
preset_slug=preset.slug,
|
|
template_slugs=template_slugs,
|
|
underlying_units=normalized_inputs.underlying_units,
|
|
loan_amount=normalized_inputs.loan_amount,
|
|
margin_call_ltv=normalized_inputs.margin_call_ltv,
|
|
)
|
|
except ValueError as exc:
|
|
if str(exc) == "loan_amount must be less than initial collateral value":
|
|
preview_history = self.comparison_service.provider.load_history(
|
|
normalized_symbol,
|
|
preset.window_start,
|
|
preset.window_end,
|
|
)
|
|
if preview_history:
|
|
_validate_initial_collateral(
|
|
normalized_inputs.underlying_units,
|
|
preview_history[0].close,
|
|
normalized_inputs.loan_amount,
|
|
)
|
|
raise
|
|
_validate_initial_collateral(
|
|
normalized_inputs.underlying_units,
|
|
preview.initial_portfolio.entry_spot,
|
|
normalized_inputs.loan_amount,
|
|
)
|
|
return self.comparison_service.compare_event_from_inputs(
|
|
preset_slug=preset.slug,
|
|
template_slugs=template_slugs,
|
|
underlying_units=normalized_inputs.underlying_units,
|
|
loan_amount=normalized_inputs.loan_amount,
|
|
margin_call_ltv=normalized_inputs.margin_call_ltv,
|
|
)
|
|
|
|
@staticmethod
|
|
def chart_model(report: EventComparisonReport, max_ranked_series: int = 3) -> EventComparisonChartModel:
|
|
ranked = report.rankings[:max_ranked_series]
|
|
if not ranked:
|
|
return EventComparisonChartModel(dates=(), series=())
|
|
dates = tuple(point.date.isoformat() for point in ranked[0].result.daily_path)
|
|
series = [
|
|
EventComparisonChartSeries(
|
|
name="Unhedged collateral baseline",
|
|
values=tuple(round(point.underlying_value, 2) for point in ranked[0].result.daily_path),
|
|
)
|
|
]
|
|
for item in ranked:
|
|
series.append(
|
|
EventComparisonChartSeries(
|
|
name=item.template_name,
|
|
values=tuple(round(point.net_portfolio_value, 2) for point in item.result.daily_path),
|
|
)
|
|
)
|
|
return EventComparisonChartModel(dates=dates, series=tuple(series))
|
|
|
|
@staticmethod
|
|
def drilldown_model(
|
|
report: EventComparisonReport,
|
|
*,
|
|
template_slug: str | None = None,
|
|
) -> EventComparisonDrilldownModel:
|
|
ranking = EventComparisonPageService._select_ranking(report, template_slug=template_slug)
|
|
daily_path = ranking.result.daily_path
|
|
worst_ltv_point = max(daily_path, key=lambda point: point.ltv_hedged, default=None)
|
|
breach_dates = tuple(point.date.isoformat() for point in daily_path if point.margin_call_hedged)
|
|
return EventComparisonDrilldownModel(
|
|
rank=ranking.rank,
|
|
template_slug=ranking.template_slug,
|
|
template_name=ranking.template_name,
|
|
survived_margin_call=ranking.survived_margin_call,
|
|
margin_call_days_hedged=ranking.margin_call_days_hedged,
|
|
total_option_payoff_realized=ranking.result.summary_metrics.total_option_payoff_realized,
|
|
hedge_cost=ranking.hedge_cost,
|
|
final_equity=ranking.final_equity,
|
|
worst_ltv_hedged=ranking.max_ltv_hedged,
|
|
worst_ltv_date=worst_ltv_point.date.isoformat() if worst_ltv_point is not None else None,
|
|
breach_dates=breach_dates,
|
|
rows=tuple(
|
|
EventComparisonDrilldownRow(
|
|
date=point.date.isoformat(),
|
|
spot_close=point.spot_close,
|
|
net_portfolio_value=point.net_portfolio_value,
|
|
option_market_value=point.option_market_value,
|
|
realized_option_cashflow=point.realized_option_cashflow,
|
|
ltv_unhedged=point.ltv_unhedged,
|
|
ltv_hedged=point.ltv_hedged,
|
|
margin_call_hedged=point.margin_call_hedged,
|
|
active_position_ids=point.active_position_ids,
|
|
)
|
|
for point in daily_path
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def drilldown_options(report: EventComparisonReport) -> dict[str, str]:
|
|
return {ranking.template_slug: f"#{ranking.rank} — {ranking.template_name}" for ranking in report.rankings}
|
|
|
|
@staticmethod
|
|
def _select_ranking(
|
|
report: EventComparisonReport,
|
|
*,
|
|
template_slug: str | None = None,
|
|
) -> EventComparisonRanking:
|
|
if not report.rankings:
|
|
raise ValueError("Event comparison report has no ranked results")
|
|
if template_slug is None:
|
|
return report.rankings[0]
|
|
for ranking in report.rankings:
|
|
if ranking.template_slug == template_slug:
|
|
return ranking
|
|
raise ValueError(f"Unknown ranked template: {template_slug}")
|