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

fix(pricing): correct relative hedge payoff calculations

Browse Source
This commit is contained in:
Bu5hm4nn
2026-03-25 19:27:26 +01:00
parent 5217304624
commit bfb6c71be3
6 changed files with 241 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
app/components/strategy_panel.py
View File

@@ -4,6 +4,12 @@ from typing import Any
from nicegui import ui
from app.domain.portfolio_math import (
strategy_benefit_per_unit,
strategy_protection_floor_bounds,
strategy_upside_cap_price,
)
class StrategyComparisonPanel:
"""Interactive strategy comparison with scenario slider and cost-benefit table."""
@@ -107,8 +113,6 @@ class StrategyComparisonPanel:
for strategy in self.strategies:
name = str(strategy.get("name", "strategy")).replace("_", " ").title()
cost = float(strategy.get("estimated_cost", 0.0))
floor = strategy.get("max_drawdown_floor", "—")
cap = strategy.get("upside_cap", "—")
scenario = self._scenario_benefit(strategy)
scenario_class = (
"text-emerald-600 dark:text-emerald-400" if scenario >= 0 else "text-rose-600 dark:text-rose-400"
@@ -117,8 +121,8 @@ class StrategyComparisonPanel:
<tr class=\"border-b border-slate-200 dark:border-slate-800\">
<td class=\"px-4 py-3 font-medium text-slate-900 dark:text-slate-100\">{name}</td>
<td class=\"px-4 py-3 text-slate-600 dark:text-slate-300\">${cost:,.2f}</td>
<td class=\"px-4 py-3 text-slate-600 dark:text-slate-300\">{self._fmt_optional_money(floor)}</td>
<td class=\"px-4 py-3 text-slate-600 dark:text-slate-300\">{self._fmt_optional_money(cap)}</td>
<td class=\"px-4 py-3 text-slate-600 dark:text-slate-300\">{self._format_floor(strategy)}</td>
<td class=\"px-4 py-3 text-slate-600 dark:text-slate-300\">{self._format_cap(strategy)}</td>
<td class=\"px-4 py-3 font-semibold {scenario_class}\">${scenario:,.2f}</td>
</tr>
""")
@@ -140,17 +144,26 @@ class StrategyComparisonPanel:
"""
def _scenario_benefit(self, strategy: dict[str, Any]) -> float:
scenario_spot = self._scenario_spot()
cost = float(strategy.get("estimated_cost", 0.0))
floor = strategy.get("max_drawdown_floor")
cap = strategy.get("upside_cap")
benefit = -cost
return strategy_benefit_per_unit(
strategy,
current_spot=self.current_spot,
scenario_spot=self._scenario_spot(),
)
if isinstance(floor, (int, float)) and scenario_spot < float(floor):
benefit += float(floor) - scenario_spot
if isinstance(cap, (int, float)) and scenario_spot > float(cap):
benefit -= scenario_spot - float(cap)
return benefit
def _format_floor(self, strategy: dict[str, Any]) -> str:
bounds = strategy_protection_floor_bounds(strategy, current_spot=self.current_spot)
if bounds is None:
return self._fmt_optional_money(strategy.get("max_drawdown_floor"))
low, high = bounds
if abs(high - low) < 1e-9:
return f"${high:,.2f}"
return f"${low:,.2f}–${high:,.2f}"
def _format_cap(self, strategy: dict[str, Any]) -> str:
cap = strategy_upside_cap_price(strategy, current_spot=self.current_spot)
if cap is None:
return self._fmt_optional_money(strategy.get("upside_cap"))
return f"${cap:,.2f}"
@staticmethod
def _fmt_optional_money(value: Any) -> str:

138
app/domain/portfolio_math.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from decimal import Decimal
from decimal import Decimal, InvalidOperation
from typing import Any, Mapping
from app.domain.backtesting_math import PricePerAsset
@@ -54,6 +54,111 @@ def _safe_quote_price(value: object) -> float:
return parsed
def _strategy_decimal(value: object) -> Decimal | None:
if value is None or isinstance(value, bool):
return None
if isinstance(value, Decimal):
return value if value.is_finite() else None
if isinstance(value, int):
return Decimal(value)
if isinstance(value, float):
return decimal_from_float(value)
if isinstance(value, str):
stripped = value.strip()
if not stripped:
return None
try:
parsed = Decimal(stripped)
except InvalidOperation:
return None
return parsed if parsed.is_finite() else None
return None
def _strategy_downside_put_legs(strategy: Mapping[str, Any], current_spot: Decimal) -> list[tuple[Decimal, Decimal]]:
raw_legs = strategy.get("downside_put_legs")
if isinstance(raw_legs, (list, tuple)):
parsed_legs: list[tuple[Decimal, Decimal]] = []
for leg in raw_legs:
if not isinstance(leg, Mapping):
continue
weight = _strategy_decimal(leg.get("allocation_weight", leg.get("weight")))
strike_pct = _strategy_decimal(leg.get("strike_pct"))
if weight is None or strike_pct is None or weight <= 0 or strike_pct <= 0:
continue
parsed_legs.append((weight, current_spot * strike_pct))
if parsed_legs:
return parsed_legs
protection_floor_pct = _strategy_decimal(strategy.get("protection_floor_pct"))
if protection_floor_pct is not None and protection_floor_pct > 0:
return [(_DECIMAL_ONE, current_spot * protection_floor_pct)]
absolute_floor = _strategy_decimal(strategy.get("max_drawdown_floor"))
if absolute_floor is not None and absolute_floor > 0:
return [(_DECIMAL_ONE, absolute_floor)]
return [(_DECIMAL_ONE, current_spot * Decimal("0.95"))]
def _strategy_upside_cap_decimal(strategy: Mapping[str, Any], current_spot: Decimal) -> Decimal | None:
upside_cap_pct = _strategy_decimal(strategy.get("upside_cap_pct"))
if upside_cap_pct is not None and upside_cap_pct > 0:
return current_spot * upside_cap_pct
absolute_cap = _strategy_decimal(strategy.get("upside_cap"))
if absolute_cap is not None and absolute_cap > 0:
return absolute_cap
return None
def _strategy_option_payoff_per_unit(
strategy: Mapping[str, Any], current_spot: Decimal, scenario_spot: Decimal
) -> Decimal:
return sum(
weight * max(strike_price - scenario_spot, _DECIMAL_ZERO)
for weight, strike_price in _strategy_downside_put_legs(strategy, current_spot)
)
def _strategy_upside_cap_effect_per_unit(
strategy: Mapping[str, Any], current_spot: Decimal, scenario_spot: Decimal
) -> Decimal:
cap = _strategy_upside_cap_decimal(strategy, current_spot)
if cap is None or scenario_spot <= cap:
return _DECIMAL_ZERO
return -(scenario_spot - cap)
def strategy_protection_floor_bounds(strategy: Mapping[str, Any], *, current_spot: float) -> tuple[float, float] | None:
current_spot_decimal = decimal_from_float(current_spot)
legs = _strategy_downside_put_legs(strategy, current_spot_decimal)
if not legs:
return None
floor_prices = [strike_price for _, strike_price in legs]
return _decimal_to_float(min(floor_prices)), _decimal_to_float(max(floor_prices))
def strategy_upside_cap_price(strategy: Mapping[str, Any], *, current_spot: float) -> float | None:
cap = _strategy_upside_cap_decimal(strategy, decimal_from_float(current_spot))
if cap is None:
return None
return _decimal_to_float(cap)
def strategy_benefit_per_unit(strategy: Mapping[str, Any], *, current_spot: float, scenario_spot: float) -> float:
current_spot_decimal = decimal_from_float(current_spot)
scenario_spot_decimal = decimal_from_float(scenario_spot)
cost = _strategy_decimal(strategy.get("estimated_cost")) or _DECIMAL_ZERO
benefit = (
_strategy_option_payoff_per_unit(strategy, current_spot_decimal, scenario_spot_decimal)
+ _strategy_upside_cap_effect_per_unit(strategy, current_spot_decimal, scenario_spot_decimal)
- cost
)
return round(float(benefit), 2)
def resolve_portfolio_spot_from_quote(
config: PortfolioConfig,
quote: Mapping[str, object],
@@ -158,24 +263,17 @@ def strategy_metrics_from_snapshot(
current_spot = PricePerWeight(amount=spot, currency=BaseCurrency.USD, per_unit=WeightUnit.OUNCE_TROY)
loan_amount = Money(amount=decimal_from_float(float(snapshot["loan_amount"])), currency=BaseCurrency.USD)
base_equity = Money(amount=decimal_from_float(float(snapshot["net_equity"])), currency=BaseCurrency.USD)
default_floor = spot * Decimal("0.95")
floor = (
decimal_from_float(float(strategy["max_drawdown_floor"])) if "max_drawdown_floor" in strategy else default_floor
)
cap_decimal = (
decimal_from_float(float(strategy["upside_cap"]))
if isinstance(strategy.get("upside_cap"), (int, float))
else None
)
cost = decimal_from_float(float(strategy["estimated_cost"]))
cost = _strategy_decimal(strategy.get("estimated_cost")) or _DECIMAL_ZERO
scenario_prices = [spot * _pct_factor(pct) for pct in range(-25, 30, 5)]
benefits: list[float] = []
for price in scenario_prices:
payoff = max(floor - price, _DECIMAL_ZERO)
if cap_decimal is not None and price > cap_decimal:
payoff -= price - cap_decimal
benefits.append(round(float(payoff - cost), 2))
benefits = [
strategy_benefit_per_unit(
strategy,
current_spot=_decimal_to_float(spot),
scenario_spot=_decimal_to_float(price),
)
for price in scenario_prices
]
scenario_price = spot * _pct_factor(scenario_pct)
scenario_gold_value = gold_weight * PricePerWeight(
@@ -185,10 +283,8 @@ def strategy_metrics_from_snapshot(
)
current_gold_value = gold_weight * current_spot
unhedged_equity = scenario_gold_value - loan_amount
scenario_payoff_per_unit = max(floor - scenario_price, _DECIMAL_ZERO)
capped_upside_per_unit = _DECIMAL_ZERO
if cap_decimal is not None and scenario_price > cap_decimal:
capped_upside_per_unit = -(scenario_price - cap_decimal)
scenario_payoff_per_unit = _strategy_option_payoff_per_unit(strategy, spot, scenario_price)
capped_upside_per_unit = _strategy_upside_cap_effect_per_unit(strategy, spot, scenario_price)
option_payoff_cash = Money(amount=gold_weight.amount * scenario_payoff_per_unit, currency=BaseCurrency.USD)
capped_upside_cash = Money(amount=gold_weight.amount * capped_upside_per_unit, currency=BaseCurrency.USD)

12
app/pages/hedge.py
View File

@@ -22,12 +22,13 @@ logger = logging.getLogger(__name__)
def _cost_benefit_options(metrics: dict) -> dict:
return {
"tooltip": {"trigger": "axis"},
"grid": {"left": 64, "right": 24, "top": 24, "bottom": 48},
"xAxis": {
"type": "category",
"data": [f"${point['price']:.0f}" for point in metrics["scenario_series"]],
"name": "GLD spot",
"name": "Collateral spot",
},
"yAxis": {"type": "value", "name": "Net benefit / oz"},
"yAxis": {"type": "value", "name": "Net hedge benefit / oz"},
"series": [
{
"type": "bar",
@@ -35,6 +36,11 @@ def _cost_benefit_options(metrics: dict) -> dict:
"itemStyle": {
"color": "#0ea5e9",
},
"markLine": {
"symbol": "none",
"lineStyle": {"color": "#94a3b8", "type": "dashed"},
"data": [{"yAxis": 0}],
},
}
],
}
@@ -49,12 +55,14 @@ def _waterfall_options(metrics: dict) -> dict:
return {
"tooltip": {"trigger": "axis", "axisPointer": {"type": "shadow"}},
"grid": {"left": 80, "right": 24, "top": 24, "bottom": 48},
"xAxis": {"type": "category", "data": [label for label, _ in steps]},
"yAxis": {"type": "value", "name": "USD"},
"series": [
{
"type": "bar",
"data": values,
"label": {"show": True, "position": "top", "formatter": "{c}"},
},
],
}

17
app/services/strategy_templates.py
View File

@@ -134,29 +134,36 @@ class StrategyTemplateService:
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"},
"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,
"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)
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"
]
items.append(
{
"name": strategy_name,
"template_slug": template.slug,
"label": template.display_name,
"description": template.description,
"downside_put_legs": downside_put_legs,
**ui_defaults.get(strategy_name, {}),
}
)

38
tests/test_hedge_metrics.py
View File

@@ -3,7 +3,7 @@ from __future__ import annotations
from app.domain.portfolio_math import resolve_portfolio_spot_from_quote
from app.models.portfolio import PortfolioConfig
from app.pages.common import strategy_metrics
from app.pages.hedge import _waterfall_options
from app.pages.hedge import _cost_benefit_options, _waterfall_options
def test_protective_put_atm_minus_20pct_improves_equity() -> None:
@@ -11,15 +11,15 @@ def test_protective_put_atm_minus_20pct_improves_equity() -> None:
assert metrics["scenario_price"] == 172.0
assert metrics["unhedged_equity"] == 27_000.0
assert metrics["hedged_equity"] == 58_750.0
assert metrics["hedged_equity"] == 63_750.0
assert metrics["hedged_equity"] > metrics["unhedged_equity"]
assert metrics["waterfall_steps"] == [
("Base equity", 70_000.0),
("Spot move", -43_000.0),
("Option payoff", 38_000.0),
("Option payoff", 43_000.0),
("Call cap", 0.0),
("Hedge cost", -6_250.0),
("Net equity", 58_750.0),
("Net equity", 63_750.0),
]
@@ -28,11 +28,39 @@ def test_hedge_waterfall_uses_zero_based_contribution_bars() -> None:
assert len(options["series"]) == 1
assert options["series"][0]["type"] == "bar"
assert options["series"][0]["label"]["show"] is True
values = options["series"][0]["data"]
assert values[2]["value"] == 38_000.0
assert values[2]["value"] == 43_000.0
assert values[2]["itemStyle"]["color"] == "#22c55e"
def test_cost_benefit_chart_shows_positive_downside_benefit_when_puts_are_in_the_money() -> None:
metrics = strategy_metrics(
"protective_put_atm",
-20,
portfolio={
**{
"gold_value": 968000.0,
"loan_amount": 145000.0,
"ltv_ratio": 145000.0 / 968000.0,
"net_equity": 823000.0,
"spot_price": 4400.0,
"gold_units": 220.0,
"margin_call_ltv": 0.75,
"margin_call_price": 878.79,
"cash_buffer": 18500.0,
"hedge_budget": 8000.0,
}
},
)
options = _cost_benefit_options(metrics)
assert options["xAxis"]["name"] == "Collateral spot"
assert options["yAxis"]["name"] == "Net hedge benefit / oz"
assert options["series"][0]["data"][0] > 0
assert options["series"][0]["data"][1] > 0
def test_hedge_quote_resolution_converts_gld_share_price_to_ozt_spot() -> None:
"""Hedge page should convert GLD share quotes to USD/ozt for display."""
config = PortfolioConfig(entry_price=4400.0, gold_ounces=220.0, entry_basis_mode="weight", loan_amount=145000.0)

45
tests/test_portfolio_math.py
View File

@@ -8,7 +8,9 @@ from app.domain.portfolio_math import (
build_alert_context,
portfolio_snapshot_from_config,
resolve_portfolio_spot_from_quote,
strategy_benefit_per_unit,
strategy_metrics_from_snapshot,
strategy_protection_floor_bounds,
)
from app.domain.units import BaseCurrency, WeightUnit
from app.models.portfolio import PortfolioConfig
@@ -81,14 +83,14 @@ def test_strategy_metrics_from_snapshot_preserves_minus_20pct_protective_put_exa
assert metrics["scenario_price"] == 172.0
assert metrics["unhedged_equity"] == 27000.0
assert metrics["hedged_equity"] == 58750.0
assert metrics["hedged_equity"] == 63750.0
assert metrics["waterfall_steps"] == [
("Base equity", 70000.0),
("Spot move", -43000.0),
("Option payoff", 38000.0),
("Option payoff", 43000.0),
("Call cap", 0.0),
("Hedge cost", -6250.0),
("Net equity", 58750.0),
("Net equity", 63750.0),
]
@@ -131,6 +133,43 @@ def test_resolve_portfolio_spot_from_quote_fails_closed_to_configured_price() ->
assert updated_at == ""
def test_strategy_metrics_from_snapshot_uses_relative_template_strikes_for_high_spot_portfolios() -> None:
from app.pages.common import strategy_catalog
strategy = next(item for item in strategy_catalog() if item["name"] == "protective_put_atm")
snapshot = portfolio_snapshot_from_config(
PortfolioConfig(entry_price=4400.0, gold_ounces=220.0, entry_basis_mode="weight", loan_amount=145000.0)
)
metrics = strategy_metrics_from_snapshot(strategy, -20, snapshot)
assert metrics["scenario_price"] == 3520.0
assert metrics["unhedged_equity"] == 629400.0
assert metrics["hedged_equity"] == 821625.0
assert metrics["waterfall_steps"] == [
("Base equity", 823000.0),
("Spot move", -193600.0),
("Option payoff", 193600.0),
("Call cap", 0.0),
("Hedge cost", -1375.0),
("Net equity", 821625.0),
]
assert metrics["scenario_series"][0]["benefit"] == 1093.75
assert metrics["scenario_series"][1]["benefit"] == 873.75
def test_strategy_benefit_and_floor_bounds_support_laddered_relative_put_legs() -> None:
from app.pages.common import strategy_catalog
strategy = next(item for item in strategy_catalog() if item["name"] == "laddered_put_50_50_atm_otm95")
floor_bounds = strategy_protection_floor_bounds(strategy, current_spot=4400.0)
benefit = strategy_benefit_per_unit(strategy, current_spot=4400.0, scenario_spot=3520.0)
assert floor_bounds == (4180.0, 4400.0)
assert benefit == 765.55
def test_resolve_portfolio_spot_from_quote_fails_closed_for_unsupported_symbol() -> None:
"""Unsupported symbols without instrument metadata should fail closed."""
from app.models.portfolio import PortfolioConfig