310 lines
12 KiB
Python
310 lines
12 KiB
Python
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,
|
|
)
|