213 lines
8.9 KiB
Python
213 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
|
|
from app.models.backtest import BacktestScenario, EventComparisonReport
|
|
from app.services.backtesting.comparison import EventComparisonService
|
|
from app.services.backtesting.historical_provider import DailyClosePoint, SyntheticHistoricalProvider
|
|
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, ...]
|
|
|
|
|
|
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.")
|
|
try:
|
|
scenario = self.comparison_service.preview_scenario_from_inputs(
|
|
preset_slug=preset_slug,
|
|
template_slugs=template_slugs,
|
|
underlying_units=underlying_units,
|
|
loan_amount=loan_amount,
|
|
margin_call_ltv=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(underlying_units, preview[0].close, loan_amount)
|
|
raise
|
|
_validate_initial_collateral(underlying_units, scenario.initial_portfolio.entry_spot, 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.")
|
|
if underlying_units <= 0:
|
|
raise ValueError("Underlying units must be positive")
|
|
if loan_amount < 0:
|
|
raise ValueError("Loan amount must be non-negative")
|
|
if not 0 < margin_call_ltv < 1:
|
|
raise ValueError("Margin call LTV must be between 0 and 1")
|
|
|
|
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=underlying_units,
|
|
loan_amount=loan_amount,
|
|
margin_call_ltv=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(underlying_units, preview_history[0].close, loan_amount)
|
|
raise
|
|
_validate_initial_collateral(underlying_units, preview.initial_portfolio.entry_spot, loan_amount)
|
|
return self.comparison_service.compare_event_from_inputs(
|
|
preset_slug=preset.slug,
|
|
template_slugs=template_slugs,
|
|
underlying_units=underlying_units,
|
|
loan_amount=loan_amount,
|
|
margin_call_ltv=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))
|