200 lines
8.2 KiB
Python
200 lines
8.2 KiB
Python
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,
|
|
)
|