bu5hm4nn/vault-dash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(EXEC-001): add hedge strategy builder

Browse Source
This commit is contained in:
Bu5hm4nn
2026-03-27 22:33:20 +01:00
parent 554a41a060
commit 4620234967
9 changed files with 429 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/pages/common.py
View File

@@ -110,14 +110,15 @@ def option_chain() -> list[dict[str, Any]]:
def strategy_metrics(
strategy_name: str,
strategy_key: str,
scenario_pct: int,
*,
portfolio: dict[str, Any] | None = None,
) -> dict[str, Any]:
catalog = strategy_catalog()
strategy = next(
(item for item in strategy_catalog() if item["name"] == strategy_name),
strategy_catalog()[0],
(item for item in catalog if item.get("template_slug") == strategy_key or item.get("name") == strategy_key),
catalog[0],
)
portfolio = portfolio or portfolio_snapshot()
return strategy_metrics_from_snapshot(strategy, scenario_pct, portfolio)

124
app/pages/hedge.py
View File

@@ -12,10 +12,10 @@ from app.pages.common import (
demo_spot_price,
portfolio_snapshot,
split_page_panes,
strategy_catalog,
strategy_metrics,
)
from app.services.runtime import get_data_service
from app.services.strategy_templates import StrategyTemplateService
logger = logging.getLogger(__name__)
@@ -99,12 +99,18 @@ async def _resolve_hedge_spot(workspace_id: str | None = None) -> tuple[dict[str
async def _render_hedge_page(workspace_id: str | None = None) -> None:
repo = get_workspace_repository()
config = repo.load_portfolio_config(workspace_id) if workspace_id else None
portfolio, quote_source, quote_updated_at = await _resolve_hedge_spot(workspace_id)
strategies = strategy_catalog()
strategy_map = {strategy["label"]: strategy["name"] for strategy in strategies}
selected = {"strategy": strategies[0]["name"], "label": strategies[0]["label"], "scenario_pct": 0}
template_service = StrategyTemplateService()
strategies_state = {"items": template_service.catalog_items()}
def strategy_map() -> dict[str, str]:
return {strategy["label"]: strategy["template_slug"] for strategy in strategies_state["items"]}
selected = {
"strategy": strategies_state["items"][0]["template_slug"],
"label": strategies_state["items"][0]["label"],
"scenario_pct": 0,
}
if quote_source == "configured_entry_price":
spot_label = f"Current spot reference: ${portfolio['spot_price']:,.2f}/oz (configured entry price)"
@@ -130,7 +136,9 @@ async def _render_hedge_page(workspace_id: str | None = None) -> 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["label"], label="Strategy selector").classes("w-full")
selector = ui.select(
list(strategy_map().keys()), 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(spot_label).classes("text-sm text-slate-500 dark:text-slate-400")
@@ -144,6 +152,45 @@ async def _render_hedge_page(workspace_id: str | None = None) -> None:
"text-xs text-slate-500 dark:text-slate-400"
)
with (
ui.card()
.classes(
"w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900"
)
.props("data-testid=strategy-builder-card")
):
ui.label("Strategy Builder").classes("text-lg font-semibold text-slate-900 dark:text-slate-100")
ui.label(
"Save a custom protective put or equal-weight two-leg ladder for reuse across hedge, backtests, and event comparison."
).classes("text-sm text-slate-500 dark:text-slate-400")
builder_type_options = {"Protective put": "protective_put", "Two-leg ladder": "laddered_put"}
builder_name = ui.input("Template name", placeholder="Crash Guard 97").classes("w-full")
builder_type = ui.select(
list(builder_type_options.keys()),
value="Protective put",
label="Strategy type",
).classes("w-full")
builder_expiry_days = ui.number("Expiration days", value=365, min=30, step=30).classes("w-full")
builder_primary_strike = ui.number(
"Primary strike (% of spot)",
value=100,
min=1,
max=150,
step=1,
).classes("w-full")
builder_secondary_strike = ui.number(
"Secondary strike (% of spot)",
value=95,
min=1,
max=150,
step=1,
).classes("w-full")
ui.label("Two-leg ladders currently save with equal 50/50 weights.").classes(
"text-xs text-slate-500 dark:text-slate-400"
)
builder_status = ui.label("").classes("text-sm text-slate-600 dark:text-slate-300")
save_template_button = ui.button("Save template").props("color=primary outline")
summary = ui.card().classes(
"w-full rounded-2xl border border-slate-200 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-900"
)
@@ -161,12 +208,32 @@ async def _render_hedge_page(workspace_id: str | None = None) -> None:
"h-96 w-full rounded-2xl border border-slate-200 bg-white p-4 shadow-sm dark:border-slate-800 dark:bg-slate-900"
)
syncing_controls = {"value": False}
def refresh_available_strategies() -> None:
strategies_state["items"] = template_service.catalog_items()
options = strategy_map()
syncing_controls["value"] = True
try:
selector.options = list(options.keys())
if selected["label"] not in options:
first_label = next(iter(options))
selected["label"] = first_label
selected["strategy"] = options[first_label]
selector.value = first_label
selector.update()
finally:
syncing_controls["value"] = False
def render_summary() -> None:
metrics = strategy_metrics(selected["strategy"], selected["scenario_pct"], portfolio=portfolio)
strategy = metrics["strategy"]
summary.clear()
with summary:
ui.label("Scenario Summary").classes("text-lg font-semibold text-slate-900 dark:text-slate-100")
ui.label(f"Selected template: {strategy['label']}").classes(
"text-sm text-slate-500 dark:text-slate-400"
)
ui.label(strategy["description"]).classes("text-sm text-slate-600 dark:text-slate-300")
with ui.grid(columns=1).classes("w-full gap-4 sm:grid-cols-2 lg:grid-cols-1 xl:grid-cols-2"):
cards = [
@@ -190,7 +257,7 @@ async def _render_hedge_page(workspace_id: str | None = None) -> None:
with ui.grid(columns=2).classes("w-full gap-4 max-md:grid-cols-1"):
result_cards = [
("Scenario spot", f"${metrics['scenario_price']:,.2f}"),
("Hedge cost", f"${strategy['estimated_cost']:,.2f}/oz"),
("Hedge cost", f"${float(strategy.get('estimated_cost', 0.0)):,.2f}/oz"),
("Unhedged equity", f"${metrics['unhedged_equity']:,.0f}"),
("Hedged equity", f"${metrics['hedged_equity']:,.0f}"),
("Net hedge benefit", f"${metrics['hedged_equity'] - metrics['unhedged_equity']:,.0f}"),
@@ -211,8 +278,10 @@ async def _render_hedge_page(workspace_id: str | None = None) -> None:
waterfall_chart.update()
def refresh_from_selector(event) -> None:
if syncing_controls["value"]:
return
selected["label"] = str(event.value)
selected["strategy"] = strategy_map[selected["label"]]
selected["strategy"] = strategy_map()[selected["label"]]
render_summary()
def refresh_from_slider(event) -> None:
@@ -221,6 +290,43 @@ async def _render_hedge_page(workspace_id: str | None = None) -> None:
slider_value.set_text(f"Scenario move: {sign}{selected['scenario_pct']}%")
render_summary()
def save_template() -> None:
builder_status.set_text("")
try:
builder_kind = builder_type_options[str(builder_type.value)]
strikes = (float(builder_primary_strike.value or 0.0) / 100.0,)
weights: tuple[float, ...] | None = None
if builder_kind == "laddered_put":
strikes = (
float(builder_primary_strike.value or 0.0) / 100.0,
float(builder_secondary_strike.value or 0.0) / 100.0,
)
weights = (0.5, 0.5)
template = template_service.create_custom_template(
display_name=str(builder_name.value or ""),
template_kind=builder_kind,
target_expiry_days=int(builder_expiry_days.value or 0),
strike_pcts=strikes,
weights=weights,
)
except (ValueError, KeyError) as exc:
builder_status.set_text(str(exc))
return
refresh_available_strategies()
selected["label"] = template.display_name
selected["strategy"] = template.slug
syncing_controls["value"] = True
try:
selector.value = template.display_name
selector.update()
finally:
syncing_controls["value"] = False
builder_status.set_text(
f"Saved template {template.display_name}. Reusable on hedge, backtests, and event comparison."
)
render_summary()
selector.on_value_change(refresh_from_selector)
slider.on_value_change(refresh_from_slider)
save_template_button.on_click(lambda: save_template())
render_summary()

156
app/services/strategy_templates.py
View File

@@ -1,15 +1,19 @@
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
DEFAULT_TEMPLATE_FILE = Path(__file__).resolve().parents[2] / "config" / "strategy_templates.json"
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]:
@@ -65,26 +69,51 @@ def default_strategy_templates() -> list[StrategyTemplate]:
class FileStrategyTemplateRepository:
def __init__(self, path: str | Path = DEFAULT_TEMPLATE_FILE) -> None:
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_seeded()
self._ensure_store()
defaults = self._seed_templates()
payload = json.loads(self.path.read_text())
return [StrategyTemplate.from_dict(item) for item in payload.get("templates", [])]
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)
payload = {"templates": [template.to_dict() for template in templates]}
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_seeded(self) -> None:
def _ensure_store(self) -> None:
if self.path.exists():
return
self.save_all(default_strategy_templates())
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:
@@ -105,6 +134,77 @@ class StrategyTemplateService:
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))
@@ -157,6 +257,7 @@ class StrategyTemplateService:
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,
@@ -164,7 +265,8 @@ class StrategyTemplateService:
"label": template.display_name,
"description": template.description,
"downside_put_legs": downside_put_legs,
**ui_defaults.get(strategy_name, {}),
"estimated_cost": defaults.get("estimated_cost", self._estimated_cost(template)),
"coverage": defaults.get("coverage", self._coverage_label(template)),
}
)
return items
@@ -176,6 +278,44 @@ class StrategyTemplateService:
)
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: