347 lines
14 KiB
Python
347 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
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
|
|
|
|
CONFIG_TEMPLATE_FILE = Path(__file__).resolve().parents[2] / "config" / "strategy_templates.json"
|
|
DATA_TEMPLATE_FILE = Path("data/strategy_templates.json")
|
|
_SLUGIFY_RE = re.compile(r"[^a-z0-9]+")
|
|
|
|
|
|
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 = DATA_TEMPLATE_FILE,
|
|
*,
|
|
seed_path: str | Path | None = CONFIG_TEMPLATE_FILE,
|
|
) -> None:
|
|
self.path = Path(path)
|
|
self.seed_path = Path(seed_path) if seed_path is not None else None
|
|
|
|
def list_templates(self) -> list[StrategyTemplate]:
|
|
self._ensure_store()
|
|
defaults = self._seed_templates()
|
|
payload = json.loads(self.path.read_text())
|
|
customs = [StrategyTemplate.from_dict(item) for item in payload.get("templates", [])]
|
|
merged: dict[str, StrategyTemplate] = {template.slug: template for template in defaults}
|
|
for template in customs:
|
|
merged[template.slug] = template
|
|
return list(merged.values())
|
|
|
|
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)
|
|
default_slugs = {template.slug for template in self._seed_templates()}
|
|
payload = {
|
|
"templates": [
|
|
template.to_dict()
|
|
for template in templates
|
|
if template.slug not in default_slugs or "system" not in template.tags
|
|
]
|
|
}
|
|
self.path.write_text(json.dumps(payload, indent=2) + "\n")
|
|
|
|
def _ensure_store(self) -> None:
|
|
if self.path.exists():
|
|
return
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.path.write_text(json.dumps({"templates": []}, indent=2) + "\n")
|
|
|
|
def _seed_templates(self) -> list[StrategyTemplate]:
|
|
if self.seed_path is not None and self.seed_path.exists():
|
|
payload = json.loads(self.seed_path.read_text())
|
|
return [StrategyTemplate.from_dict(item) for item in payload.get("templates", [])]
|
|
return 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 create_custom_template(
|
|
self,
|
|
*,
|
|
display_name: str,
|
|
template_kind: str,
|
|
target_expiry_days: int,
|
|
strike_pcts: tuple[float, ...],
|
|
weights: tuple[float, ...] | None = None,
|
|
underlying_symbol: str = "GLD",
|
|
) -> StrategyTemplate:
|
|
name = display_name.strip()
|
|
if not name:
|
|
raise ValueError("Template name is required")
|
|
if target_expiry_days <= 0:
|
|
raise ValueError("Expiration days must be positive")
|
|
if not strike_pcts:
|
|
raise ValueError("At least one strike is required")
|
|
if any(strike_pct <= 0 for strike_pct in strike_pcts):
|
|
raise ValueError("Strike percentages must be positive")
|
|
|
|
templates = self.repository.list_templates()
|
|
normalized_name = name.casefold()
|
|
if any(template.display_name.casefold() == normalized_name for template in templates):
|
|
raise ValueError("Template name already exists")
|
|
|
|
slug = self._slugify(name)
|
|
if any(template.slug == slug for template in templates):
|
|
raise ValueError("Template slug already exists; choose a different name")
|
|
|
|
template_id = f"custom-{uuid4()}"
|
|
if template_kind == "protective_put":
|
|
if len(strike_pcts) != 1:
|
|
raise ValueError("Protective put builder expects exactly one strike")
|
|
template = StrategyTemplate.protective_put(
|
|
template_id=template_id,
|
|
slug=slug,
|
|
display_name=name,
|
|
description=f"Custom {target_expiry_days}-day protective put at {strike_pcts[0] * 100:.0f}% strike.",
|
|
strike_pct=strike_pcts[0],
|
|
target_expiry_days=target_expiry_days,
|
|
underlying_symbol=underlying_symbol,
|
|
tags=("custom", "protective_put"),
|
|
)
|
|
elif template_kind == "laddered_put":
|
|
if len(strike_pcts) < 2:
|
|
raise ValueError("Laddered put builder expects at least two strikes")
|
|
resolved_weights = weights or self._equal_weights(len(strike_pcts))
|
|
if len(resolved_weights) != len(strike_pcts):
|
|
raise ValueError("Weights must match the number of strikes")
|
|
template = StrategyTemplate.laddered_put(
|
|
template_id=template_id,
|
|
slug=slug,
|
|
display_name=name,
|
|
description=(
|
|
f"Custom {target_expiry_days}-day put ladder at "
|
|
+ ", ".join(f"{strike_pct * 100:.0f}%" for strike_pct in strike_pcts)
|
|
+ " strikes."
|
|
),
|
|
strike_pcts=strike_pcts,
|
|
weights=resolved_weights,
|
|
target_expiry_days=target_expiry_days,
|
|
underlying_symbol=underlying_symbol,
|
|
tags=("custom", "laddered_put"),
|
|
)
|
|
else:
|
|
raise ValueError(f"Unsupported strategy type: {template_kind}")
|
|
|
|
templates.append(template)
|
|
self.repository.save_all(templates)
|
|
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, "coverage": "High"},
|
|
"protective_put_otm_95": {"estimated_cost": 4.95, "coverage": "Balanced"},
|
|
"protective_put_otm_90": {"estimated_cost": 3.7, "coverage": "Cost-efficient"},
|
|
"laddered_put_50_50_atm_otm95": {
|
|
"estimated_cost": 4.45,
|
|
"coverage": "Layered",
|
|
},
|
|
"laddered_put_33_33_33_atm_otm95_otm90": {
|
|
"estimated_cost": 3.85,
|
|
"coverage": "Layered",
|
|
},
|
|
}
|
|
items: list[dict[str, Any]] = []
|
|
for template in self.list_active_templates():
|
|
strategy_name = self.strategy_name(template)
|
|
downside_put_legs = [
|
|
{
|
|
"allocation_weight": leg.allocation_weight,
|
|
"strike_pct": leg.strike_rule.value,
|
|
}
|
|
for leg in template.legs
|
|
if leg.side == "long" and leg.option_type == "put"
|
|
]
|
|
defaults = ui_defaults.get(strategy_name, {}) if "system" in template.tags else {}
|
|
items.append(
|
|
{
|
|
"name": strategy_name,
|
|
"template_slug": template.slug,
|
|
"label": template.display_name,
|
|
"description": template.description,
|
|
"downside_put_legs": downside_put_legs,
|
|
"estimated_cost": defaults.get("estimated_cost", self._estimated_cost(template)),
|
|
"coverage": defaults.get("coverage", self._coverage_label(template)),
|
|
}
|
|
)
|
|
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 _slugify(display_name: str) -> str:
|
|
slug = _SLUGIFY_RE.sub("-", display_name.strip().lower()).strip("-")
|
|
if not slug:
|
|
raise ValueError("Template name must contain letters or numbers")
|
|
return slug
|
|
|
|
@staticmethod
|
|
def _equal_weights(count: int) -> tuple[float, ...]:
|
|
if count <= 0:
|
|
raise ValueError("count must be positive")
|
|
base = round(1.0 / count, 10)
|
|
weights = [base for _ in range(count)]
|
|
weights[-1] = 1.0 - sum(weights[:-1])
|
|
return tuple(weights)
|
|
|
|
@staticmethod
|
|
def _estimated_cost(template: StrategyTemplate) -> float:
|
|
weighted_cost = sum(
|
|
leg.allocation_weight * max(1.1, 6.25 - ((1.0 - leg.strike_rule.value) * 25.5)) for leg in template.legs
|
|
)
|
|
expiry_factor = max(0.45, (template.target_expiry_days / 365) ** 0.5)
|
|
weighted_cost *= expiry_factor
|
|
if len(template.legs) > 1:
|
|
weighted_cost *= 0.8
|
|
return round(weighted_cost, 2)
|
|
|
|
@staticmethod
|
|
def _coverage_label(template: StrategyTemplate) -> str:
|
|
if len(template.legs) > 1:
|
|
return "Layered"
|
|
strike_pct = template.legs[0].strike_rule.value
|
|
if strike_pct >= 0.99:
|
|
return "High"
|
|
if strike_pct >= 0.95:
|
|
return "Balanced"
|
|
return "Cost-efficient"
|
|
|
|
@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,
|
|
)
|