From bfb6c71be35c1c96e11291ef7102600b72679dd1 Mon Sep 17 00:00:00 2001 From: Bu5hm4nn Date: Wed, 25 Mar 2026 19:27:26 +0100 Subject: [PATCH] fix(pricing): correct relative hedge payoff calculations --- app/components/strategy_panel.py | 41 ++++++--- app/domain/portfolio_math.py | 138 ++++++++++++++++++++++++----- app/pages/hedge.py | 12 ++- app/services/strategy_templates.py | 17 ++-- tests/test_hedge_metrics.py | 38 ++++++-- tests/test_portfolio_math.py | 45 +++++++++- 6 files changed, 241 insertions(+), 50 deletions(-) diff --git a/app/components/strategy_panel.py b/app/components/strategy_panel.py index 5e0209a..e1cb165 100644 --- a/app/components/strategy_panel.py +++ b/app/components/strategy_panel.py @@ -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: {name} ${cost:,.2f} - {self._fmt_optional_money(floor)} - {self._fmt_optional_money(cap)} + {self._format_floor(strategy)} + {self._format_cap(strategy)} ${scenario:,.2f} """) @@ -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: diff --git a/app/domain/portfolio_math.py b/app/domain/portfolio_math.py index 0e489bd..c830b1f 100644 --- a/app/domain/portfolio_math.py +++ b/app/domain/portfolio_math.py @@ -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) diff --git a/app/pages/hedge.py b/app/pages/hedge.py index 84e1cef..7db304a 100644 --- a/app/pages/hedge.py +++ b/app/pages/hedge.py @@ -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}"}, }, ], } diff --git a/app/services/strategy_templates.py b/app/services/strategy_templates.py index a99c8d4..4a11b6c 100644 --- a/app/services/strategy_templates.py +++ b/app/services/strategy_templates.py @@ -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, {}), } ) diff --git a/tests/test_hedge_metrics.py b/tests/test_hedge_metrics.py index 64db089..488e5f6 100644 --- a/tests/test_hedge_metrics.py +++ b/tests/test_hedge_metrics.py @@ -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) diff --git a/tests/test_portfolio_math.py b/tests/test_portfolio_math.py index e6f7017..9fd9b02 100644 --- a/tests/test_portfolio_math.py +++ b/tests/test_portfolio_math.py @@ -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