239 lines
9.5 KiB
Python
239 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from app.models.backtest import BacktestPortfolioState, BacktestScenario, ProviderRef, TemplateRef
|
|
from app.models.strategy_template import EntryPolicy, RollPolicy, StrategyTemplate
|
|
from app.services.backtesting.historical_provider import (
|
|
DailyClosePoint,
|
|
SyntheticHistoricalProvider,
|
|
YFinanceHistoricalPriceSource,
|
|
)
|
|
from app.services.backtesting.service import BacktestService
|
|
from app.services.strategy_templates import StrategyTemplateService
|
|
|
|
|
|
class FakeHistorySource:
|
|
def __init__(self, rows: list[DailyClosePoint]) -> None:
|
|
self.rows = rows
|
|
self.calls: list[tuple[str, date, date]] = []
|
|
|
|
def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]:
|
|
self.calls.append((symbol, start_date, end_date))
|
|
return list(self.rows)
|
|
|
|
|
|
FIXTURE_HISTORY = [
|
|
DailyClosePoint(date=date(2024, 1, 8), close=85.0),
|
|
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),
|
|
]
|
|
|
|
|
|
def test_synthetic_historical_provider_sorts_and_filters_daily_closes() -> None:
|
|
source = FakeHistorySource(FIXTURE_HISTORY)
|
|
provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.01)
|
|
|
|
series = provider.load_history(symbol="GLD", start_date=date(2024, 1, 3), end_date=date(2024, 1, 5))
|
|
|
|
assert [(point.date.isoformat(), point.close) for point in series] == [
|
|
("2024-01-03", 96.0),
|
|
("2024-01-04", 92.0),
|
|
("2024-01-05", 88.0),
|
|
]
|
|
assert source.calls == [("GLD", date(2024, 1, 3), date(2024, 1, 5))]
|
|
|
|
|
|
def _build_scenario(
|
|
*,
|
|
provider_ref: ProviderRef | None = None,
|
|
symbol: str = "GLD",
|
|
entry_spot: float = 100.0,
|
|
) -> BacktestScenario:
|
|
return BacktestScenario(
|
|
scenario_id="gld-selloff-1",
|
|
display_name="GLD selloff",
|
|
symbol=symbol,
|
|
start_date=date(2024, 1, 2),
|
|
end_date=date(2024, 1, 8),
|
|
initial_portfolio=BacktestPortfolioState(
|
|
currency="USD",
|
|
underlying_units=1000.0,
|
|
entry_spot=entry_spot,
|
|
loan_amount=68_000.0,
|
|
margin_call_ltv=0.75,
|
|
),
|
|
template_refs=(TemplateRef(slug="protective-put-atm-12m", version=1),),
|
|
provider_ref=provider_ref or ProviderRef(provider_id="synthetic_v1", pricing_mode="synthetic_bs_mid"),
|
|
)
|
|
|
|
|
|
def test_backtest_service_runs_template_backtest_with_daily_points() -> None:
|
|
source = FakeHistorySource(FIXTURE_HISTORY)
|
|
provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0)
|
|
service = BacktestService(provider=provider, template_service=StrategyTemplateService())
|
|
|
|
result = service.run_scenario(_build_scenario())
|
|
|
|
assert result.scenario_id == "gld-selloff-1"
|
|
assert len(result.template_results) == 1
|
|
|
|
template_result = result.template_results[0]
|
|
summary = template_result.summary_metrics
|
|
|
|
assert summary.start_value == 100_000.0
|
|
assert summary.end_value_unhedged == 85_000.0
|
|
assert summary.end_value_hedged_net > summary.end_value_unhedged
|
|
assert summary.total_hedge_cost > 0.0
|
|
assert summary.max_ltv_hedged < summary.max_ltv_unhedged
|
|
assert summary.margin_threshold_breached_unhedged is True
|
|
assert summary.margin_threshold_breached_hedged is False
|
|
|
|
assert [point.date.isoformat() for point in template_result.daily_path] == [
|
|
"2024-01-02",
|
|
"2024-01-03",
|
|
"2024-01-04",
|
|
"2024-01-05",
|
|
"2024-01-08",
|
|
]
|
|
assert template_result.daily_path[0].premium_cashflow < 0.0
|
|
assert template_result.daily_path[-1].margin_call_unhedged is True
|
|
assert template_result.daily_path[-1].margin_call_hedged is False
|
|
|
|
|
|
def test_backtest_keeps_long_dated_option_open_past_scenario_end() -> None:
|
|
source = FakeHistorySource(FIXTURE_HISTORY)
|
|
provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0)
|
|
service = BacktestService(provider=provider, template_service=StrategyTemplateService())
|
|
|
|
result = service.run_scenario(_build_scenario())
|
|
|
|
template_result = result.template_results[0]
|
|
final_day = template_result.daily_path[-1]
|
|
|
|
assert final_day.date == date(2024, 1, 8)
|
|
assert final_day.realized_option_cashflow == 0.0
|
|
assert final_day.option_market_value > 0.0
|
|
assert final_day.active_position_ids == ("protective-put-atm-12m-position-1",)
|
|
assert template_result.summary_metrics.total_option_payoff_realized == 0.0
|
|
|
|
|
|
def test_backtest_rejects_unsupported_provider_pricing_combination() -> None:
|
|
source = FakeHistorySource(FIXTURE_HISTORY)
|
|
provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0)
|
|
service = BacktestService(provider=provider, template_service=StrategyTemplateService())
|
|
scenario = _build_scenario(provider_ref=ProviderRef(provider_id="daily_snapshots_v1", pricing_mode="snapshot_mid"))
|
|
|
|
with pytest.raises(ValueError, match="Unsupported provider/pricing combination"):
|
|
service.run_scenario(scenario)
|
|
|
|
|
|
def test_backtest_rejects_scenario_start_close_when_history_starts_after_scenario_start() -> None:
|
|
delayed_history = [row for row in FIXTURE_HISTORY if row.date >= date(2024, 1, 3)]
|
|
source = FakeHistorySource(delayed_history)
|
|
provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0)
|
|
service = BacktestService(provider=provider, template_service=StrategyTemplateService())
|
|
|
|
with pytest.raises(ValueError, match="Scenario start_date must match the first available historical price point"):
|
|
service.run_scenario(_build_scenario())
|
|
|
|
|
|
def test_backtest_rejects_mismatched_entry_spot_for_scenario_start_close() -> None:
|
|
source = FakeHistorySource(FIXTURE_HISTORY)
|
|
provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0)
|
|
service = BacktestService(provider=provider, template_service=StrategyTemplateService())
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="initial_portfolio.entry_spot must match the first historical close used for entry",
|
|
):
|
|
service.run_scenario(_build_scenario(entry_spot=100.02))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "message"),
|
|
[
|
|
("contract_mode", "listed_contracts", "Unsupported contract_mode"),
|
|
(
|
|
"roll_policy",
|
|
RollPolicy(policy_type="roll_n_days_before_expiry", days_before_expiry=5),
|
|
"Unsupported roll_policy",
|
|
),
|
|
("entry_policy", EntryPolicy(entry_timing="scenario_start_close", stagger_days=1), None),
|
|
],
|
|
)
|
|
def test_backtest_rejects_unsupported_template_behaviors(field: str, value: object, message: str | None) -> None:
|
|
source = FakeHistorySource(FIXTURE_HISTORY)
|
|
provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0)
|
|
template_service = StrategyTemplateService()
|
|
service = BacktestService(provider=provider, template_service=template_service)
|
|
template = template_service.get_template("protective-put-atm-12m")
|
|
|
|
template_kwargs = {
|
|
"template_id": template.template_id,
|
|
"slug": template.slug,
|
|
"display_name": template.display_name,
|
|
"description": template.description,
|
|
"template_kind": template.template_kind,
|
|
"status": template.status,
|
|
"version": template.version,
|
|
"underlying_symbol": template.underlying_symbol,
|
|
"contract_mode": template.contract_mode,
|
|
"legs": template.legs,
|
|
"roll_policy": template.roll_policy,
|
|
"entry_policy": template.entry_policy,
|
|
"tags": template.tags,
|
|
"created_at": template.created_at,
|
|
"updated_at": template.updated_at,
|
|
}
|
|
template_kwargs[field] = value
|
|
unsupported_template = StrategyTemplate(**template_kwargs)
|
|
|
|
if field == "entry_policy":
|
|
unsupported_template = StrategyTemplate(
|
|
**{
|
|
**template_kwargs,
|
|
"entry_policy": EntryPolicy(entry_timing="scenario_start_close", stagger_days=1),
|
|
}
|
|
)
|
|
with pytest.raises(ValueError, match="Unsupported entry_policy configuration"):
|
|
service._validate_template_for_mvp(unsupported_template)
|
|
return
|
|
|
|
with pytest.raises(ValueError, match=message or "Unsupported"):
|
|
service._validate_template_for_mvp(unsupported_template)
|
|
|
|
|
|
def test_yfinance_price_source_treats_end_date_inclusively(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[tuple[str, str, str]] = []
|
|
|
|
class FakeTicker:
|
|
def __init__(self, symbol: str) -> None:
|
|
self.symbol = symbol
|
|
|
|
def history(self, start: str, end: str, interval: str):
|
|
calls.append((start, end, interval))
|
|
return pd.DataFrame(
|
|
{"Close": [100.0, 101.0]},
|
|
index=pd.to_datetime(["2024-01-04", "2024-01-05"]),
|
|
)
|
|
|
|
class FakeYFinance:
|
|
Ticker = FakeTicker
|
|
|
|
monkeypatch.setattr("app.services.backtesting.historical_provider.yf", FakeYFinance())
|
|
|
|
source = YFinanceHistoricalPriceSource()
|
|
rows = source.load_daily_closes(symbol="GLD", start_date=date(2024, 1, 4), end_date=date(2024, 1, 5))
|
|
|
|
assert calls == [("2024-01-04", "2024-01-06", "1d")]
|
|
assert [(row.date.isoformat(), row.close) for row in rows] == [
|
|
("2024-01-04", 100.0),
|
|
("2024-01-05", 101.0),
|
|
]
|