diff --git a/app/models/__init__.py b/app/models/__init__.py index ddd913c..97df178 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -3,6 +3,7 @@ from .option import Greeks, OptionContract, OptionMoneyness from .portfolio import LombardPortfolio from .strategy import HedgingStrategy, ScenarioResult, StrategyType +from .strategy_template import EntryPolicy, RollPolicy, StrategyTemplate, TemplateLeg __all__ = [ "Greeks", @@ -12,4 +13,8 @@ __all__ = [ "OptionMoneyness", "ScenarioResult", "StrategyType", + "StrategyTemplate", + "TemplateLeg", + "RollPolicy", + "EntryPolicy", ] diff --git a/app/models/strategy_template.py b/app/models/strategy_template.py new file mode 100644 index 0000000..aec7e83 --- /dev/null +++ b/app/models/strategy_template.py @@ -0,0 +1,309 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Literal + +StrategyTemplateKind = Literal["protective_put", "laddered_put"] +StrategyTemplateStatus = Literal["draft", "active", "archived"] +ContractMode = Literal["continuous_units", "listed_contracts"] +LegSide = Literal["long", "short"] +LegOptionType = Literal["put", "call"] +StrikeRuleType = Literal["spot_pct"] +QuantityRule = Literal["target_coverage_pct"] +RollPolicyType = Literal["hold_to_expiry", "roll_n_days_before_expiry"] +EntryTiming = Literal["scenario_start_close"] + + +@dataclass(frozen=True) +class StrikeRule: + rule_type: StrikeRuleType + value: float + + def __post_init__(self) -> None: + if self.rule_type != "spot_pct": + raise ValueError("unsupported strike rule") + if self.value <= 0: + raise ValueError("strike rule value must be positive") + + def to_dict(self) -> dict[str, Any]: + return {"rule_type": self.rule_type, "value": self.value} + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> StrikeRule: + return cls(rule_type=payload["rule_type"], value=float(payload["value"])) + + +@dataclass(frozen=True) +class TemplateLeg: + leg_id: str + side: LegSide + option_type: LegOptionType + allocation_weight: float + strike_rule: StrikeRule + target_expiry_days: int + quantity_rule: QuantityRule + target_coverage_pct: float = 1.0 + + def __post_init__(self) -> None: + if self.side not in {"long", "short"}: + raise ValueError("unsupported leg side") + if self.option_type not in {"put", "call"}: + raise ValueError("unsupported option type") + if self.allocation_weight <= 0: + raise ValueError("allocation_weight must be positive") + if self.target_expiry_days <= 0: + raise ValueError("target_expiry_days must be positive") + if self.quantity_rule != "target_coverage_pct": + raise ValueError("unsupported quantity rule") + if self.target_coverage_pct <= 0: + raise ValueError("target_coverage_pct must be positive") + + def to_dict(self) -> dict[str, Any]: + return { + "leg_id": self.leg_id, + "side": self.side, + "option_type": self.option_type, + "allocation_weight": self.allocation_weight, + "strike_rule": self.strike_rule.to_dict(), + "target_expiry_days": self.target_expiry_days, + "quantity_rule": self.quantity_rule, + "target_coverage_pct": self.target_coverage_pct, + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> TemplateLeg: + return cls( + leg_id=payload["leg_id"], + side=payload["side"], + option_type=payload["option_type"], + allocation_weight=float(payload["allocation_weight"]), + strike_rule=StrikeRule.from_dict(payload["strike_rule"]), + target_expiry_days=int(payload["target_expiry_days"]), + quantity_rule=payload.get("quantity_rule", "target_coverage_pct"), + target_coverage_pct=float(payload.get("target_coverage_pct", 1.0)), + ) + + +@dataclass(frozen=True) +class RollPolicy: + policy_type: RollPolicyType + days_before_expiry: int | None = None + rebalance_on_new_deposit: bool = False + + def __post_init__(self) -> None: + if self.policy_type not in {"hold_to_expiry", "roll_n_days_before_expiry"}: + raise ValueError("unsupported roll policy") + if self.policy_type == "roll_n_days_before_expiry" and (self.days_before_expiry or 0) <= 0: + raise ValueError("days_before_expiry is required for rolling policies") + + def to_dict(self) -> dict[str, Any]: + return { + "policy_type": self.policy_type, + "days_before_expiry": self.days_before_expiry, + "rebalance_on_new_deposit": self.rebalance_on_new_deposit, + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> RollPolicy: + return cls( + policy_type=payload["policy_type"], + days_before_expiry=payload.get("days_before_expiry"), + rebalance_on_new_deposit=bool(payload.get("rebalance_on_new_deposit", False)), + ) + + +@dataclass(frozen=True) +class EntryPolicy: + entry_timing: EntryTiming + stagger_days: int | None = None + + def __post_init__(self) -> None: + if self.entry_timing != "scenario_start_close": + raise ValueError("unsupported entry timing") + if self.stagger_days is not None and self.stagger_days < 0: + raise ValueError("stagger_days must be non-negative") + + def to_dict(self) -> dict[str, Any]: + return {"entry_timing": self.entry_timing, "stagger_days": self.stagger_days} + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> EntryPolicy: + return cls(entry_timing=payload["entry_timing"], stagger_days=payload.get("stagger_days")) + + +@dataclass(frozen=True) +class StrategyTemplate: + template_id: str + slug: str + display_name: str + description: str + template_kind: StrategyTemplateKind + status: StrategyTemplateStatus + version: int + underlying_symbol: str + contract_mode: ContractMode + legs: tuple[TemplateLeg, ...] + roll_policy: RollPolicy + entry_policy: EntryPolicy + tags: tuple[str, ...] = field(default_factory=tuple) + created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + def __post_init__(self) -> None: + if self.template_kind not in {"protective_put", "laddered_put"}: + raise ValueError("unsupported template_kind") + if self.status not in {"draft", "active", "archived"}: + raise ValueError("unsupported template status") + if self.contract_mode not in {"continuous_units", "listed_contracts"}: + raise ValueError("unsupported contract mode") + if not self.slug: + raise ValueError("slug is required") + if not self.display_name: + raise ValueError("display_name is required") + if self.version <= 0: + raise ValueError("version must be positive") + if not self.legs: + raise ValueError("at least one template leg is required") + if self.template_kind in {"protective_put", "laddered_put"}: + if any(leg.side != "long" or leg.option_type != "put" for leg in self.legs): + raise ValueError("put templates support only long put legs") + total_weight = sum(leg.allocation_weight for leg in self.legs) + if abs(total_weight - 1.0) > 1e-9: + raise ValueError("weights must sum to 1.0") + expiry_days = {leg.target_expiry_days for leg in self.legs} + if len(expiry_days) != 1: + raise ValueError("all template legs must share target_expiry_days in MVP") + + @property + def target_expiry_days(self) -> int: + return self.legs[0].target_expiry_days + + def to_dict(self) -> dict[str, Any]: + return { + "template_id": self.template_id, + "slug": self.slug, + "display_name": self.display_name, + "description": self.description, + "template_kind": self.template_kind, + "status": self.status, + "version": self.version, + "underlying_symbol": self.underlying_symbol, + "contract_mode": self.contract_mode, + "legs": [leg.to_dict() for leg in self.legs], + "roll_policy": self.roll_policy.to_dict(), + "entry_policy": self.entry_policy.to_dict(), + "tags": list(self.tags), + "created_at": self.created_at.isoformat(), + "updated_at": self.updated_at.isoformat(), + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> StrategyTemplate: + return cls( + template_id=payload["template_id"], + slug=payload["slug"], + display_name=payload["display_name"], + description=payload.get("description", ""), + template_kind=payload["template_kind"], + status=payload.get("status", "active"), + version=int(payload.get("version", 1)), + underlying_symbol=payload.get("underlying_symbol", "GLD"), + contract_mode=payload.get("contract_mode", "continuous_units"), + legs=tuple(TemplateLeg.from_dict(leg) for leg in payload["legs"]), + roll_policy=RollPolicy.from_dict(payload["roll_policy"]), + entry_policy=EntryPolicy.from_dict(payload["entry_policy"]), + tags=tuple(payload.get("tags", [])), + created_at=datetime.fromisoformat(payload["created_at"]), + updated_at=datetime.fromisoformat(payload["updated_at"]), + ) + + @classmethod + def protective_put( + cls, + *, + template_id: str, + slug: str, + display_name: str, + description: str, + strike_pct: float, + target_expiry_days: int, + underlying_symbol: str = "GLD", + tags: tuple[str, ...] = (), + ) -> StrategyTemplate: + now = datetime.now(timezone.utc) + return cls( + template_id=template_id, + slug=slug, + display_name=display_name, + description=description, + template_kind="protective_put", + status="active", + version=1, + underlying_symbol=underlying_symbol, + contract_mode="continuous_units", + legs=( + TemplateLeg( + leg_id=f"{template_id}-leg-1", + side="long", + option_type="put", + allocation_weight=1.0, + strike_rule=StrikeRule(rule_type="spot_pct", value=strike_pct), + target_expiry_days=target_expiry_days, + quantity_rule="target_coverage_pct", + target_coverage_pct=1.0, + ), + ), + roll_policy=RollPolicy(policy_type="hold_to_expiry"), + entry_policy=EntryPolicy(entry_timing="scenario_start_close"), + tags=tags, + created_at=now, + updated_at=now, + ) + + @classmethod + def laddered_put( + cls, + *, + template_id: str, + slug: str, + display_name: str, + description: str, + strike_pcts: tuple[float, ...], + weights: tuple[float, ...], + target_expiry_days: int, + underlying_symbol: str = "GLD", + tags: tuple[str, ...] = (), + ) -> StrategyTemplate: + if len(strike_pcts) != len(weights): + raise ValueError("strike_pcts and weights must have the same length") + now = datetime.now(timezone.utc) + return cls( + template_id=template_id, + slug=slug, + display_name=display_name, + description=description, + template_kind="laddered_put", + status="active", + version=1, + underlying_symbol=underlying_symbol, + contract_mode="continuous_units", + legs=tuple( + TemplateLeg( + leg_id=f"{template_id}-leg-{index}", + side="long", + option_type="put", + allocation_weight=weight, + strike_rule=StrikeRule(rule_type="spot_pct", value=strike_pct), + target_expiry_days=target_expiry_days, + quantity_rule="target_coverage_pct", + target_coverage_pct=1.0, + ) + for index, (strike_pct, weight) in enumerate(zip(strike_pcts, weights, strict=True), start=1) + ), + roll_policy=RollPolicy(policy_type="hold_to_expiry"), + entry_policy=EntryPolicy(entry_timing="scenario_start_close"), + tags=tags, + created_at=now, + updated_at=now, + ) diff --git a/app/pages/common.py b/app/pages/common.py index deccfd4..8dc272b 100644 --- a/app/pages/common.py +++ b/app/pages/common.py @@ -6,6 +6,8 @@ from typing import Any from nicegui import ui +from app.services.strategy_templates import StrategyTemplateService + NAV_ITEMS: list[tuple[str, str, str]] = [ ("overview", "/", "Overview"), ("hedge", "/hedge", "Hedge Analysis"), @@ -38,33 +40,7 @@ def portfolio_snapshot() -> dict[str, float]: def strategy_catalog() -> list[dict[str, Any]]: - return [ - { - "name": "protective_put", - "label": "Protective Put", - "description": "Full downside protection below the hedge strike with uncapped upside.", - "estimated_cost": 6.25, - "max_drawdown_floor": 210.0, - "coverage": "High", - }, - { - "name": "collar", - "label": "Collar", - "description": "Lower premium by financing puts with covered call upside caps.", - "estimated_cost": 2.10, - "max_drawdown_floor": 208.0, - "upside_cap": 228.0, - "coverage": "Balanced", - }, - { - "name": "laddered_puts", - "label": "Laddered Puts", - "description": "Multiple maturities and strikes reduce roll concentration and smooth protection.", - "estimated_cost": 4.45, - "max_drawdown_floor": 205.0, - "coverage": "Layered", - }, - ] + return StrategyTemplateService().catalog_items() def quick_recommendations(portfolio: dict[str, Any] | None = None) -> list[dict[str, str]]: @@ -73,7 +49,7 @@ def quick_recommendations(portfolio: dict[str, Any] | None = None) -> list[dict[ return [ { "title": "Balanced hedge favored", - "summary": "A collar keeps the current LTV comfortably below the margin threshold while limiting upfront spend.", + "summary": "A 95% protective put balances margin-call protection with a lower upfront hedge cost.", "tone": "positive", }, { diff --git a/app/pages/hedge.py b/app/pages/hedge.py index 46de2c3..e2dd8c1 100644 --- a/app/pages/hedge.py +++ b/app/pages/hedge.py @@ -75,7 +75,7 @@ def _waterfall_options(metrics: dict) -> dict: def hedge_page() -> None: strategies = strategy_catalog() strategy_map = {strategy["label"]: strategy["name"] for strategy in strategies} - selected = {"strategy": strategies[0]["name"], "scenario_pct": 0} + selected = {"strategy": strategies[0]["name"], "label": strategies[0]["label"], "scenario_pct": 0} with dashboard_page( "Hedge Analysis", @@ -87,9 +87,7 @@ def hedge_page() -> None: "w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900" ): ui.label("Strategy Controls").classes("text-lg font-semibold text-slate-900 dark:text-slate-100") - selector = ui.select(strategy_map, value=selected["strategy"], label="Strategy selector").classes( - "w-full" - ) + selector = ui.select(strategy_map, value=selected["label"], label="Strategy selector").classes("w-full") slider_value = ui.label("Scenario move: +0%").classes("text-sm text-slate-500 dark:text-slate-400") slider = ui.slider(min=-25, max=25, value=0, step=5).classes("w-full") ui.label(f"Current spot reference: ${demo_spot_price():,.2f}").classes( @@ -134,13 +132,16 @@ def hedge_page() -> None: ui.label(value).classes("text-2xl font-bold text-slate-900 dark:text-slate-100") ui.label(strategy["description"]).classes("text-sm text-slate-600 dark:text-slate-300") - cost_chart.options = _cost_benefit_options(metrics) + cost_chart.options.clear() + cost_chart.options.update(_cost_benefit_options(metrics)) cost_chart.update() - waterfall_chart.options = _waterfall_options(metrics) + waterfall_chart.options.clear() + waterfall_chart.options.update(_waterfall_options(metrics)) waterfall_chart.update() def refresh_from_selector(event) -> None: - selected["strategy"] = event.value + selected["label"] = str(event.value) + selected["strategy"] = strategy_map[selected["label"]] render_summary() def refresh_from_slider(event) -> None: diff --git a/app/services/strategy_templates.py b/app/services/strategy_templates.py new file mode 100644 index 0000000..a99c8d4 --- /dev/null +++ b/app/services/strategy_templates.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from app.models.strategy_template import StrategyTemplate +from app.strategies.base import BaseStrategy, StrategyConfig +from app.strategies.laddered_put import LadderedPutStrategy, LadderSpec +from app.strategies.protective_put import ProtectivePutSpec, ProtectivePutStrategy + +DEFAULT_TEMPLATE_FILE = Path(__file__).resolve().parents[2] / "config" / "strategy_templates.json" + + +def default_strategy_templates() -> list[StrategyTemplate]: + return [ + StrategyTemplate.protective_put( + template_id="protective-put-atm-12m-v1", + slug="protective-put-atm-12m", + display_name="Protective Put ATM", + description="Full downside protection using a 12-month at-the-money put.", + strike_pct=1.0, + target_expiry_days=365, + tags=("system", "protective_put", "conservative"), + ), + StrategyTemplate.protective_put( + template_id="protective-put-95pct-12m-v1", + slug="protective-put-95pct-12m", + display_name="Protective Put 95%", + description="Lower-cost 12-month protective put using a 95% spot strike.", + strike_pct=0.95, + target_expiry_days=365, + tags=("system", "protective_put", "balanced"), + ), + StrategyTemplate.protective_put( + template_id="protective-put-90pct-12m-v1", + slug="protective-put-90pct-12m", + display_name="Protective Put 90%", + description="Cost-sensitive 12-month protective put using a 90% spot strike.", + strike_pct=0.90, + target_expiry_days=365, + tags=("system", "protective_put", "cost_sensitive"), + ), + StrategyTemplate.laddered_put( + template_id="ladder-50-50-atm-95pct-12m-v1", + slug="ladder-50-50-atm-95pct-12m", + display_name="Laddered Puts 50/50 ATM + 95%", + description="Split hedge evenly across ATM and 95% strike 12-month puts.", + strike_pcts=(1.0, 0.95), + weights=(0.5, 0.5), + target_expiry_days=365, + tags=("system", "laddered_put", "balanced"), + ), + StrategyTemplate.laddered_put( + template_id="ladder-33-33-33-atm-95pct-90pct-12m-v1", + slug="ladder-33-33-33-atm-95pct-90pct-12m", + display_name="Laddered Puts 33/33/33 ATM + 95% + 90%", + description="Three-layer 12-month put ladder across ATM, 95%, and 90% strikes.", + strike_pcts=(1.0, 0.95, 0.90), + weights=(1 / 3, 1 / 3, 1 / 3), + target_expiry_days=365, + tags=("system", "laddered_put", "cost_sensitive"), + ), + ] + + +class FileStrategyTemplateRepository: + def __init__(self, path: str | Path = DEFAULT_TEMPLATE_FILE) -> None: + self.path = Path(path) + + def list_templates(self) -> list[StrategyTemplate]: + self._ensure_seeded() + payload = json.loads(self.path.read_text()) + return [StrategyTemplate.from_dict(item) for item in payload.get("templates", [])] + + def get_by_slug(self, slug: str) -> StrategyTemplate | None: + return next((template for template in self.list_templates() if template.slug == slug), None) + + def save_all(self, templates: list[StrategyTemplate]) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + payload = {"templates": [template.to_dict() for template in templates]} + self.path.write_text(json.dumps(payload, indent=2) + "\n") + + def _ensure_seeded(self) -> None: + if self.path.exists(): + return + self.save_all(default_strategy_templates()) + + +class StrategyTemplateService: + def __init__(self, repository: FileStrategyTemplateRepository | None = None) -> None: + self.repository = repository or FileStrategyTemplateRepository() + + def list_active_templates(self, underlying_symbol: str = "GLD") -> list[StrategyTemplate]: + symbol = underlying_symbol.upper() + return [ + template + for template in self.repository.list_templates() + if template.status == "active" and template.underlying_symbol.upper() in {symbol, "*"} + ] + + def get_template(self, slug: str) -> StrategyTemplate: + template = self.repository.get_by_slug(slug) + if template is None: + raise KeyError(f"Unknown strategy template: {slug}") + return template + + def build_strategy(self, config: StrategyConfig, slug: str) -> BaseStrategy: + return self.build_strategy_from_template(config, self.get_template(slug)) + + def build_strategy_from_template(self, config: StrategyConfig, template: StrategyTemplate) -> BaseStrategy: + months = max(1, round(template.target_expiry_days / 30.4167)) + if template.template_kind == "protective_put": + leg = template.legs[0] + return ProtectivePutStrategy( + config, + ProtectivePutSpec( + label=self._protective_label(leg.strike_rule.value), + strike_pct=leg.strike_rule.value, + months=months, + ), + ) + if template.template_kind == "laddered_put": + return LadderedPutStrategy( + config, + LadderSpec( + label=self._ladder_label(template), + weights=tuple(leg.allocation_weight for leg in template.legs), + strike_pcts=tuple(leg.strike_rule.value for leg in template.legs), + months=months, + ), + ) + raise ValueError(f"Unsupported template kind: {template.template_kind}") + + def catalog_items(self) -> list[dict[str, Any]]: + ui_defaults = { + "protective_put_atm": {"estimated_cost": 6.25, "max_drawdown_floor": 210.0, "coverage": "High"}, + "protective_put_otm_95": {"estimated_cost": 4.95, "max_drawdown_floor": 205.0, "coverage": "Balanced"}, + "protective_put_otm_90": {"estimated_cost": 3.7, "max_drawdown_floor": 194.0, "coverage": "Cost-efficient"}, + "laddered_put_50_50_atm_otm95": { + "estimated_cost": 4.45, + "max_drawdown_floor": 205.0, + "coverage": "Layered", + }, + "laddered_put_33_33_33_atm_otm95_otm90": { + "estimated_cost": 3.85, + "max_drawdown_floor": 200.0, + "coverage": "Layered", + }, + } + items: list[dict[str, Any]] = [] + for template in self.list_active_templates(): + strategy_name = self.strategy_name(template) + items.append( + { + "name": strategy_name, + "template_slug": template.slug, + "label": template.display_name, + "description": template.description, + **ui_defaults.get(strategy_name, {}), + } + ) + return items + + def strategy_name(self, template: StrategyTemplate) -> str: + strategy = self.build_strategy_from_template( + StrategyConfig(portfolio=self._stub_portfolio(), spot_price=1.0, volatility=0.16, risk_free_rate=0.045), + template, + ) + return strategy.name + + @staticmethod + def _protective_label(strike_pct: float) -> str: + if abs(strike_pct - 1.0) < 1e-9: + return "ATM" + return f"OTM_{int(round(strike_pct * 100))}" + + def _ladder_label(self, template: StrategyTemplate) -> str: + weight_labels = "_".join(str(int(round(leg.allocation_weight * 100))) for leg in template.legs) + strike_labels = "_".join(self._strike_label(leg.strike_rule.value) for leg in template.legs) + return f"{weight_labels}_{strike_labels}" + + @staticmethod + def _strike_label(strike_pct: float) -> str: + if abs(strike_pct - 1.0) < 1e-9: + return "ATM" + return f"OTM{int(round(strike_pct * 100))}" + + @staticmethod + def _stub_portfolio(): + from app.models.portfolio import LombardPortfolio + + return LombardPortfolio( + gold_ounces=1.0, + gold_price_per_ounce=1.0, + loan_amount=0.5, + initial_ltv=0.5, + margin_call_ltv=0.75, + ) diff --git a/app/strategies/__init__.py b/app/strategies/__init__.py index dcf7a42..a2e2a04 100644 --- a/app/strategies/__init__.py +++ b/app/strategies/__init__.py @@ -1,5 +1,4 @@ from .base import BaseStrategy, StrategyConfig -from .engine import StrategySelectionEngine from .laddered_put import LadderedPutStrategy, LadderSpec from .lease import LeaseAnalysisSpec, LeaseStrategy from .protective_put import ProtectivePutSpec, ProtectivePutStrategy @@ -15,3 +14,11 @@ __all__ = [ "LeaseStrategy", "StrategySelectionEngine", ] + + +def __getattr__(name: str): + if name == "StrategySelectionEngine": + from .engine import StrategySelectionEngine + + return StrategySelectionEngine + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/app/strategies/engine.py b/app/strategies/engine.py index 5a57ba6..58a8896 100644 --- a/app/strategies/engine.py +++ b/app/strategies/engine.py @@ -9,10 +9,9 @@ from app.core.pricing.black_scholes import ( DEFAULT_VOLATILITY, ) from app.models.portfolio import LombardPortfolio +from app.services.strategy_templates import StrategyTemplateService from app.strategies.base import BaseStrategy, StrategyConfig -from app.strategies.laddered_put import LadderedPutStrategy, LadderSpec from app.strategies.lease import LeaseStrategy -from app.strategies.protective_put import ProtectivePutSpec, ProtectivePutStrategy RiskProfile = Literal["conservative", "balanced", "cost_sensitive"] @@ -34,6 +33,7 @@ class StrategySelectionEngine: spot_price: float = RESEARCH_GLD_SPOT volatility: float = RESEARCH_VOLATILITY risk_free_rate: float = RESEARCH_RISK_FREE_RATE + template_service: StrategyTemplateService | None = None def _config(self) -> StrategyConfig: portfolio = LombardPortfolio( @@ -52,30 +52,12 @@ class StrategySelectionEngine: def _strategies(self) -> list[BaseStrategy]: config = self._config() - return [ - ProtectivePutStrategy(config, ProtectivePutSpec(label="ATM", strike_pct=1.0, months=12)), - ProtectivePutStrategy(config, ProtectivePutSpec(label="OTM_95", strike_pct=0.95, months=12)), - ProtectivePutStrategy(config, ProtectivePutSpec(label="OTM_90", strike_pct=0.90, months=12)), - LadderedPutStrategy( - config, - LadderSpec( - label="50_50_ATM_OTM95", - weights=(0.5, 0.5), - strike_pcts=(1.0, 0.95), - months=12, - ), - ), - LadderedPutStrategy( - config, - LadderSpec( - label="33_33_33_ATM_OTM95_OTM90", - weights=(1 / 3, 1 / 3, 1 / 3), - strike_pcts=(1.0, 0.95, 0.90), - months=12, - ), - ), - LeaseStrategy(config), + template_service = self.template_service or StrategyTemplateService() + template_strategies = [ + template_service.build_strategy_from_template(config, template) + for template in template_service.list_active_templates("GLD") ] + return [*template_strategies, LeaseStrategy(config)] def compare_all_strategies(self) -> list[dict]: comparisons: list[dict] = [] @@ -149,6 +131,7 @@ class StrategySelectionEngine: spot_price=self.spot_price, volatility=volatility, risk_free_rate=self.risk_free_rate, + template_service=self.template_service, ) recommendation = engine.recommend("balanced") results["volatility"].append( @@ -169,6 +152,7 @@ class StrategySelectionEngine: spot_price=spot_price, volatility=DEFAULT_VOLATILITY, risk_free_rate=DEFAULT_RISK_FREE_RATE, + template_service=self.template_service, ) recommendation = engine.recommend("balanced") results["spot_price"].append( diff --git a/config/strategy_templates.json b/config/strategy_templates.json new file mode 100644 index 0000000..564cf84 --- /dev/null +++ b/config/strategy_templates.json @@ -0,0 +1,253 @@ +{ + "templates": [ + { + "template_id": "protective-put-atm-12m-v1", + "slug": "protective-put-atm-12m", + "display_name": "Protective Put ATM", + "description": "Full downside protection using a 12-month at-the-money put.", + "template_kind": "protective_put", + "status": "active", + "version": 1, + "underlying_symbol": "GLD", + "contract_mode": "continuous_units", + "legs": [ + { + "leg_id": "protective-put-atm-12m-v1-leg-1", + "side": "long", + "option_type": "put", + "allocation_weight": 1.0, + "strike_rule": { + "rule_type": "spot_pct", + "value": 1.0 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + } + ], + "roll_policy": { + "policy_type": "hold_to_expiry", + "days_before_expiry": null, + "rebalance_on_new_deposit": false + }, + "entry_policy": { + "entry_timing": "scenario_start_close", + "stagger_days": null + }, + "tags": [ + "system", + "protective_put", + "conservative" + ], + "created_at": "2026-03-24T00:00:00+00:00", + "updated_at": "2026-03-24T00:00:00+00:00" + }, + { + "template_id": "protective-put-95pct-12m-v1", + "slug": "protective-put-95pct-12m", + "display_name": "Protective Put 95%", + "description": "Lower-cost 12-month protective put using a 95% spot strike.", + "template_kind": "protective_put", + "status": "active", + "version": 1, + "underlying_symbol": "GLD", + "contract_mode": "continuous_units", + "legs": [ + { + "leg_id": "protective-put-95pct-12m-v1-leg-1", + "side": "long", + "option_type": "put", + "allocation_weight": 1.0, + "strike_rule": { + "rule_type": "spot_pct", + "value": 0.95 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + } + ], + "roll_policy": { + "policy_type": "hold_to_expiry", + "days_before_expiry": null, + "rebalance_on_new_deposit": false + }, + "entry_policy": { + "entry_timing": "scenario_start_close", + "stagger_days": null + }, + "tags": [ + "system", + "protective_put", + "balanced" + ], + "created_at": "2026-03-24T00:00:00+00:00", + "updated_at": "2026-03-24T00:00:00+00:00" + }, + { + "template_id": "protective-put-90pct-12m-v1", + "slug": "protective-put-90pct-12m", + "display_name": "Protective Put 90%", + "description": "Cost-sensitive 12-month protective put using a 90% spot strike.", + "template_kind": "protective_put", + "status": "active", + "version": 1, + "underlying_symbol": "GLD", + "contract_mode": "continuous_units", + "legs": [ + { + "leg_id": "protective-put-90pct-12m-v1-leg-1", + "side": "long", + "option_type": "put", + "allocation_weight": 1.0, + "strike_rule": { + "rule_type": "spot_pct", + "value": 0.9 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + } + ], + "roll_policy": { + "policy_type": "hold_to_expiry", + "days_before_expiry": null, + "rebalance_on_new_deposit": false + }, + "entry_policy": { + "entry_timing": "scenario_start_close", + "stagger_days": null + }, + "tags": [ + "system", + "protective_put", + "cost_sensitive" + ], + "created_at": "2026-03-24T00:00:00+00:00", + "updated_at": "2026-03-24T00:00:00+00:00" + }, + { + "template_id": "ladder-50-50-atm-95pct-12m-v1", + "slug": "ladder-50-50-atm-95pct-12m", + "display_name": "Laddered Puts 50/50 ATM + 95%", + "description": "Split hedge evenly across ATM and 95% strike 12-month puts.", + "template_kind": "laddered_put", + "status": "active", + "version": 1, + "underlying_symbol": "GLD", + "contract_mode": "continuous_units", + "legs": [ + { + "leg_id": "ladder-50-50-atm-95pct-12m-v1-leg-1", + "side": "long", + "option_type": "put", + "allocation_weight": 0.5, + "strike_rule": { + "rule_type": "spot_pct", + "value": 1.0 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + }, + { + "leg_id": "ladder-50-50-atm-95pct-12m-v1-leg-2", + "side": "long", + "option_type": "put", + "allocation_weight": 0.5, + "strike_rule": { + "rule_type": "spot_pct", + "value": 0.95 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + } + ], + "roll_policy": { + "policy_type": "hold_to_expiry", + "days_before_expiry": null, + "rebalance_on_new_deposit": false + }, + "entry_policy": { + "entry_timing": "scenario_start_close", + "stagger_days": null + }, + "tags": [ + "system", + "laddered_put", + "balanced" + ], + "created_at": "2026-03-24T00:00:00+00:00", + "updated_at": "2026-03-24T00:00:00+00:00" + }, + { + "template_id": "ladder-33-33-33-atm-95pct-90pct-12m-v1", + "slug": "ladder-33-33-33-atm-95pct-90pct-12m", + "display_name": "Laddered Puts 33/33/33 ATM + 95% + 90%", + "description": "Three-layer 12-month put ladder across ATM, 95%, and 90% strikes.", + "template_kind": "laddered_put", + "status": "active", + "version": 1, + "underlying_symbol": "GLD", + "contract_mode": "continuous_units", + "legs": [ + { + "leg_id": "ladder-33-33-33-atm-95pct-90pct-12m-v1-leg-1", + "side": "long", + "option_type": "put", + "allocation_weight": 0.3333333333333333, + "strike_rule": { + "rule_type": "spot_pct", + "value": 1.0 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + }, + { + "leg_id": "ladder-33-33-33-atm-95pct-90pct-12m-v1-leg-2", + "side": "long", + "option_type": "put", + "allocation_weight": 0.3333333333333333, + "strike_rule": { + "rule_type": "spot_pct", + "value": 0.95 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + }, + { + "leg_id": "ladder-33-33-33-atm-95pct-90pct-12m-v1-leg-3", + "side": "long", + "option_type": "put", + "allocation_weight": 0.3333333333333333, + "strike_rule": { + "rule_type": "spot_pct", + "value": 0.9 + }, + "target_expiry_days": 365, + "quantity_rule": "target_coverage_pct", + "target_coverage_pct": 1.0 + } + ], + "roll_policy": { + "policy_type": "hold_to_expiry", + "days_before_expiry": null, + "rebalance_on_new_deposit": false + }, + "entry_policy": { + "entry_timing": "scenario_start_close", + "stagger_days": null + }, + "tags": [ + "system", + "laddered_put", + "cost_sensitive" + ], + "created_at": "2026-03-24T00:00:00+00:00", + "updated_at": "2026-03-24T00:00:00+00:00" + } + ] +} diff --git a/tests/test_e2e_playwright.py b/tests/test_e2e_playwright.py index 051accc..945f2ca 100644 --- a/tests/test_e2e_playwright.py +++ b/tests/test_e2e_playwright.py @@ -39,4 +39,14 @@ def test_homepage_and_options_page_render() -> None: assert "Server error" not in settings_text page.screenshot(path=str(ARTIFACTS / "settings.png"), full_page=True) + page.goto(f"{BASE_URL}/hedge", wait_until="domcontentloaded", timeout=30000) + expect(page.locator("text=Hedge Analysis").first).to_be_visible(timeout=15000) + expect(page.locator("text=Strategy selector").first).to_be_visible(timeout=15000) + expect(page.locator("text=Strategy Controls").first).to_be_visible(timeout=15000) + hedge_text = page.locator("body").inner_text(timeout=15000) + assert "Scenario Summary" in hedge_text + assert "RuntimeError" not in hedge_text + assert "Server error" not in hedge_text + page.screenshot(path=str(ARTIFACTS / "hedge.png"), full_page=True) + browser.close() diff --git a/tests/test_strategy_templates.py b/tests/test_strategy_templates.py new file mode 100644 index 0000000..e06b4d8 --- /dev/null +++ b/tests/test_strategy_templates.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +import app.core.pricing.black_scholes as black_scholes +from app.models.strategy_template import EntryPolicy, RollPolicy, StrategyTemplate, StrikeRule, TemplateLeg +from app.services.strategy_templates import FileStrategyTemplateRepository, StrategyTemplateService +from app.strategies.engine import StrategySelectionEngine + + +def _force_analytic_pricing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(black_scholes, "ql", None) + + +def test_strategy_template_rejects_invalid_weight_sum() -> None: + template = StrategyTemplate.protective_put( + template_id="bad-template", + slug="bad-template", + display_name="Bad Template", + description="Invalid weights should fail validation.", + strike_pct=1.0, + target_expiry_days=365, + ) + + with pytest.raises(ValueError, match="weights must sum to 1.0"): + StrategyTemplate( + 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=( + TemplateLeg( + leg_id="leg-1", + side="long", + option_type="put", + allocation_weight=0.6, + strike_rule=template.legs[0].strike_rule, + target_expiry_days=365, + quantity_rule="target_coverage_pct", + target_coverage_pct=1.0, + ), + TemplateLeg( + leg_id="leg-2", + side="long", + option_type="put", + allocation_weight=0.3, + strike_rule=template.legs[0].strike_rule, + target_expiry_days=365, + quantity_rule="target_coverage_pct", + target_coverage_pct=1.0, + ), + ), + roll_policy=template.roll_policy, + entry_policy=template.entry_policy, + tags=template.tags, + created_at=template.created_at, + updated_at=template.updated_at, + ) + + +def test_file_strategy_template_repository_seeds_default_templates(tmp_path: Path) -> None: + repository = FileStrategyTemplateRepository(tmp_path / "strategy_templates.json") + + templates = repository.list_templates() + + assert [template.slug for template in templates] == [ + "protective-put-atm-12m", + "protective-put-95pct-12m", + "protective-put-90pct-12m", + "ladder-50-50-atm-95pct-12m", + "ladder-33-33-33-atm-95pct-90pct-12m", + ] + assert (tmp_path / "strategy_templates.json").exists() is True + + +def test_strategy_template_rejects_collar_kind_for_mvp() -> None: + with pytest.raises(ValueError, match="unsupported template_kind"): + StrategyTemplate( + template_id="collar-template", + slug="collar-template", + display_name="Collar Template", + description="Collars are not supported by the template service in the MVP.", + template_kind="collar", # type: ignore[arg-type] + status="active", + version=1, + underlying_symbol="GLD", + contract_mode="continuous_units", + legs=( + TemplateLeg( + leg_id="collar-put-leg", + side="long", + option_type="put", + allocation_weight=1.0, + strike_rule=StrikeRule(rule_type="spot_pct", value=0.95), + target_expiry_days=365, + quantity_rule="target_coverage_pct", + target_coverage_pct=1.0, + ), + ), + roll_policy=RollPolicy(policy_type="hold_to_expiry"), + entry_policy=EntryPolicy(entry_timing="scenario_start_close"), + ) + + +def test_strategy_selection_engine_uses_named_templates(monkeypatch: pytest.MonkeyPatch) -> None: + _force_analytic_pricing(monkeypatch) + service = StrategyTemplateService() + engine = StrategySelectionEngine(template_service=service) + + strategies = engine.compare_all_strategies() + + assert [item["name"] for item in strategies[:5]] == [ + "protective_put_atm", + "protective_put_otm_95", + "protective_put_otm_90", + "laddered_put_50_50_atm_otm95", + "laddered_put_33_33_33_atm_otm95_otm90", + ] + + +def test_strategy_template_service_catalog_reads_named_templates() -> None: + catalog = StrategyTemplateService().catalog_items() + + assert [item["label"] for item in catalog] == [ + "Protective Put ATM", + "Protective Put 95%", + "Protective Put 90%", + "Laddered Puts 50/50 ATM + 95%", + "Laddered Puts 33/33/33 ATM + 95% + 90%", + ] + assert [item["name"] for item in catalog] == [ + "protective_put_atm", + "protective_put_otm_95", + "protective_put_otm_90", + "laddered_put_50_50_atm_otm95", + "laddered_put_33_33_33_atm_otm95_otm90", + ]