feat(EXEC-001A): add named strategy templates

This commit is contained in:
Bu5hm4nn
2026-03-24 12:27:39 +01:00
parent 78a01d9fc5
commit 2161e10626
10 changed files with 949 additions and 61 deletions

View File

@@ -3,6 +3,7 @@
from .option import Greeks, OptionContract, OptionMoneyness from .option import Greeks, OptionContract, OptionMoneyness
from .portfolio import LombardPortfolio from .portfolio import LombardPortfolio
from .strategy import HedgingStrategy, ScenarioResult, StrategyType from .strategy import HedgingStrategy, ScenarioResult, StrategyType
from .strategy_template import EntryPolicy, RollPolicy, StrategyTemplate, TemplateLeg
__all__ = [ __all__ = [
"Greeks", "Greeks",
@@ -12,4 +13,8 @@ __all__ = [
"OptionMoneyness", "OptionMoneyness",
"ScenarioResult", "ScenarioResult",
"StrategyType", "StrategyType",
"StrategyTemplate",
"TemplateLeg",
"RollPolicy",
"EntryPolicy",
] ]

View File

@@ -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,
)

View File

@@ -6,6 +6,8 @@ from typing import Any
from nicegui import ui from nicegui import ui
from app.services.strategy_templates import StrategyTemplateService
NAV_ITEMS: list[tuple[str, str, str]] = [ NAV_ITEMS: list[tuple[str, str, str]] = [
("overview", "/", "Overview"), ("overview", "/", "Overview"),
("hedge", "/hedge", "Hedge Analysis"), ("hedge", "/hedge", "Hedge Analysis"),
@@ -38,33 +40,7 @@ def portfolio_snapshot() -> dict[str, float]:
def strategy_catalog() -> list[dict[str, Any]]: def strategy_catalog() -> list[dict[str, Any]]:
return [ return StrategyTemplateService().catalog_items()
{
"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",
},
]
def quick_recommendations(portfolio: dict[str, Any] | None = None) -> list[dict[str, str]]: 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 [ return [
{ {
"title": "Balanced hedge favored", "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", "tone": "positive",
}, },
{ {

View File

@@ -75,7 +75,7 @@ def _waterfall_options(metrics: dict) -> dict:
def hedge_page() -> None: def hedge_page() -> None:
strategies = strategy_catalog() strategies = strategy_catalog()
strategy_map = {strategy["label"]: strategy["name"] for strategy in strategies} 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( with dashboard_page(
"Hedge Analysis", "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" "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") 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( selector = ui.select(strategy_map, value=selected["label"], label="Strategy selector").classes("w-full")
"w-full"
)
slider_value = ui.label("Scenario move: +0%").classes("text-sm text-slate-500 dark:text-slate-400") 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") 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( 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(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") 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() cost_chart.update()
waterfall_chart.options = _waterfall_options(metrics) waterfall_chart.options.clear()
waterfall_chart.options.update(_waterfall_options(metrics))
waterfall_chart.update() waterfall_chart.update()
def refresh_from_selector(event) -> None: def refresh_from_selector(event) -> None:
selected["strategy"] = event.value selected["label"] = str(event.value)
selected["strategy"] = strategy_map[selected["label"]]
render_summary() render_summary()
def refresh_from_slider(event) -> None: def refresh_from_slider(event) -> None:

View File

@@ -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,
)

View File

@@ -1,5 +1,4 @@
from .base import BaseStrategy, StrategyConfig from .base import BaseStrategy, StrategyConfig
from .engine import StrategySelectionEngine
from .laddered_put import LadderedPutStrategy, LadderSpec from .laddered_put import LadderedPutStrategy, LadderSpec
from .lease import LeaseAnalysisSpec, LeaseStrategy from .lease import LeaseAnalysisSpec, LeaseStrategy
from .protective_put import ProtectivePutSpec, ProtectivePutStrategy from .protective_put import ProtectivePutSpec, ProtectivePutStrategy
@@ -15,3 +14,11 @@ __all__ = [
"LeaseStrategy", "LeaseStrategy",
"StrategySelectionEngine", "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}")

View File

@@ -9,10 +9,9 @@ from app.core.pricing.black_scholes import (
DEFAULT_VOLATILITY, DEFAULT_VOLATILITY,
) )
from app.models.portfolio import LombardPortfolio from app.models.portfolio import LombardPortfolio
from app.services.strategy_templates import StrategyTemplateService
from app.strategies.base import BaseStrategy, StrategyConfig from app.strategies.base import BaseStrategy, StrategyConfig
from app.strategies.laddered_put import LadderedPutStrategy, LadderSpec
from app.strategies.lease import LeaseStrategy from app.strategies.lease import LeaseStrategy
from app.strategies.protective_put import ProtectivePutSpec, ProtectivePutStrategy
RiskProfile = Literal["conservative", "balanced", "cost_sensitive"] RiskProfile = Literal["conservative", "balanced", "cost_sensitive"]
@@ -34,6 +33,7 @@ class StrategySelectionEngine:
spot_price: float = RESEARCH_GLD_SPOT spot_price: float = RESEARCH_GLD_SPOT
volatility: float = RESEARCH_VOLATILITY volatility: float = RESEARCH_VOLATILITY
risk_free_rate: float = RESEARCH_RISK_FREE_RATE risk_free_rate: float = RESEARCH_RISK_FREE_RATE
template_service: StrategyTemplateService | None = None
def _config(self) -> StrategyConfig: def _config(self) -> StrategyConfig:
portfolio = LombardPortfolio( portfolio = LombardPortfolio(
@@ -52,30 +52,12 @@ class StrategySelectionEngine:
def _strategies(self) -> list[BaseStrategy]: def _strategies(self) -> list[BaseStrategy]:
config = self._config() config = self._config()
return [ template_service = self.template_service or StrategyTemplateService()
ProtectivePutStrategy(config, ProtectivePutSpec(label="ATM", strike_pct=1.0, months=12)), template_strategies = [
ProtectivePutStrategy(config, ProtectivePutSpec(label="OTM_95", strike_pct=0.95, months=12)), template_service.build_strategy_from_template(config, template)
ProtectivePutStrategy(config, ProtectivePutSpec(label="OTM_90", strike_pct=0.90, months=12)), for template in template_service.list_active_templates("GLD")
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),
] ]
return [*template_strategies, LeaseStrategy(config)]
def compare_all_strategies(self) -> list[dict]: def compare_all_strategies(self) -> list[dict]:
comparisons: list[dict] = [] comparisons: list[dict] = []
@@ -149,6 +131,7 @@ class StrategySelectionEngine:
spot_price=self.spot_price, spot_price=self.spot_price,
volatility=volatility, volatility=volatility,
risk_free_rate=self.risk_free_rate, risk_free_rate=self.risk_free_rate,
template_service=self.template_service,
) )
recommendation = engine.recommend("balanced") recommendation = engine.recommend("balanced")
results["volatility"].append( results["volatility"].append(
@@ -169,6 +152,7 @@ class StrategySelectionEngine:
spot_price=spot_price, spot_price=spot_price,
volatility=DEFAULT_VOLATILITY, volatility=DEFAULT_VOLATILITY,
risk_free_rate=DEFAULT_RISK_FREE_RATE, risk_free_rate=DEFAULT_RISK_FREE_RATE,
template_service=self.template_service,
) )
recommendation = engine.recommend("balanced") recommendation = engine.recommend("balanced")
results["spot_price"].append( results["spot_price"].append(

View File

@@ -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"
}
]
}

View File

@@ -39,4 +39,14 @@ def test_homepage_and_options_page_render() -> None:
assert "Server error" not in settings_text assert "Server error" not in settings_text
page.screenshot(path=str(ARTIFACTS / "settings.png"), full_page=True) 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() browser.close()

View File

@@ -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",
]