From 7c2729485c8e04e994d23af64c8831ed8fdb9763 Mon Sep 17 00:00:00 2001 From: Bu5hm4nn Date: Tue, 24 Mar 2026 21:57:40 +0100 Subject: [PATCH] feat(CORE-001B): migrate overview and hedge math to unit types --- app/domain/__init__.py | 8 ++ app/domain/portfolio_math.py | 173 +++++++++++++++++++++++++++++++++++ app/pages/common.py | 78 +--------------- app/services/alerts.py | 23 ++--- tests/test_e2e_playwright.py | 4 + tests/test_portfolio_math.py | 67 ++++++++++++++ 6 files changed, 262 insertions(+), 91 deletions(-) create mode 100644 app/domain/portfolio_math.py create mode 100644 tests/test_portfolio_math.py diff --git a/app/domain/__init__.py b/app/domain/__init__.py index 1e7450a..2fec095 100644 --- a/app/domain/__init__.py +++ b/app/domain/__init__.py @@ -1,3 +1,8 @@ +from app.domain.portfolio_math import ( + build_alert_context, + portfolio_snapshot_from_config, + strategy_metrics_from_snapshot, +) from app.domain.units import ( BaseCurrency, Money, @@ -16,4 +21,7 @@ __all__ = [ "PricePerWeight", "to_decimal", "decimal_from_float", + "portfolio_snapshot_from_config", + "build_alert_context", + "strategy_metrics_from_snapshot", ] diff --git a/app/domain/portfolio_math.py b/app/domain/portfolio_math.py new file mode 100644 index 0000000..49d34cf --- /dev/null +++ b/app/domain/portfolio_math.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +from decimal import Decimal +from typing import Any + +from app.domain.units import BaseCurrency, Money, PricePerWeight, Weight, WeightUnit, decimal_from_float +from app.models.portfolio import PortfolioConfig + +_DEFAULT_CASH_BUFFER = 18_500.0 +_DECIMAL_ZERO = Decimal("0") +_DECIMAL_ONE = Decimal("1") +_DECIMAL_HUNDRED = Decimal("100") + + +def _decimal_ratio(numerator: Decimal, denominator: Decimal) -> Decimal: + if denominator == 0: + return _DECIMAL_ZERO + return numerator / denominator + + +def _pct_factor(pct: int) -> Decimal: + return _DECIMAL_ONE + (Decimal(pct) / _DECIMAL_HUNDRED) + + +def _money_to_float(value: Money) -> float: + return float(value.amount) + + +def _decimal_to_float(value: Decimal) -> float: + return float(value) + + +def _spot_price(spot_price: float) -> PricePerWeight: + return PricePerWeight( + amount=decimal_from_float(spot_price), + currency=BaseCurrency.USD, + per_unit=WeightUnit.OUNCE_TROY, + ) + + +def _gold_weight(gold_ounces: float) -> Weight: + return Weight(amount=decimal_from_float(gold_ounces), unit=WeightUnit.OUNCE_TROY) + + +def portfolio_snapshot_from_config(config: PortfolioConfig | None = None) -> dict[str, float]: + if config is None: + gold_weight = Weight(amount=Decimal("1000"), unit=WeightUnit.OUNCE_TROY) + spot = PricePerWeight(amount=Decimal("215"), currency=BaseCurrency.USD, per_unit=WeightUnit.OUNCE_TROY) + loan_amount = Money(amount=Decimal("145000"), currency=BaseCurrency.USD) + margin_call_ltv = Decimal("0.75") + hedge_budget = Money(amount=Decimal("8000"), currency=BaseCurrency.USD) + else: + gold_weight = _gold_weight(float(config.gold_ounces or 0.0)) + spot = _spot_price(float(config.entry_price or 0.0)) + loan_amount = Money(amount=decimal_from_float(float(config.loan_amount)), currency=BaseCurrency.USD) + margin_call_ltv = decimal_from_float(float(config.margin_threshold)) + hedge_budget = Money(amount=decimal_from_float(float(config.monthly_budget)), currency=BaseCurrency.USD) + + gold_value = gold_weight * spot + net_equity = gold_value - loan_amount + ltv_ratio = _decimal_ratio(loan_amount.amount, gold_value.amount) + margin_call_price = loan_amount.amount / (margin_call_ltv * gold_weight.amount) + + return { + "gold_value": _money_to_float(gold_value), + "loan_amount": _money_to_float(loan_amount), + "ltv_ratio": _decimal_to_float(ltv_ratio), + "net_equity": _money_to_float(net_equity), + "spot_price": _decimal_to_float(spot.amount), + "gold_units": _decimal_to_float(gold_weight.amount), + "margin_call_ltv": _decimal_to_float(margin_call_ltv), + "margin_call_price": _decimal_to_float(margin_call_price), + "cash_buffer": _DEFAULT_CASH_BUFFER, + "hedge_budget": _money_to_float(hedge_budget), + } + + +def build_alert_context( + config: PortfolioConfig, + *, + spot_price: float, + source: str, + updated_at: str, +) -> dict[str, float | str]: + gold_weight = _gold_weight(float(config.gold_ounces or 0.0)) + live_spot = _spot_price(spot_price) + gold_value = gold_weight * live_spot + loan_amount = Money(amount=decimal_from_float(float(config.loan_amount)), currency=BaseCurrency.USD) + margin_call_ltv = decimal_from_float(float(config.margin_threshold)) + margin_call_price = ( + loan_amount.amount / (margin_call_ltv * gold_weight.amount) if gold_weight.amount > 0 else _DECIMAL_ZERO + ) + + return { + "spot_price": _decimal_to_float(live_spot.amount), + "gold_units": _decimal_to_float(gold_weight.amount), + "gold_value": _money_to_float(gold_value), + "loan_amount": _money_to_float(loan_amount), + "ltv_ratio": _decimal_to_float(_decimal_ratio(loan_amount.amount, gold_value.amount)), + "net_equity": _money_to_float(gold_value - loan_amount), + "margin_call_ltv": _decimal_to_float(margin_call_ltv), + "margin_call_price": _decimal_to_float(margin_call_price), + "quote_source": source, + "quote_updated_at": updated_at, + } + + +def strategy_metrics_from_snapshot( + strategy: dict[str, Any], scenario_pct: int, snapshot: dict[str, Any] +) -> dict[str, Any]: + spot = decimal_from_float(float(snapshot["spot_price"])) + gold_weight = _gold_weight(float(snapshot["gold_units"])) + 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"])) + + 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)) + + scenario_price = spot * _pct_factor(scenario_pct) + scenario_gold_value = gold_weight * PricePerWeight( + amount=scenario_price, + currency=BaseCurrency.USD, + per_unit=WeightUnit.OUNCE_TROY, + ) + 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) + + 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) + hedge_cost_cash = Money(amount=gold_weight.amount * cost, currency=BaseCurrency.USD) + hedged_equity = unhedged_equity + option_payoff_cash + capped_upside_cash - hedge_cost_cash + + waterfall_steps = [ + ("Base equity", round(_money_to_float(base_equity), 2)), + ("Spot move", round(_money_to_float(scenario_gold_value - current_gold_value), 2)), + ("Option payoff", round(_money_to_float(option_payoff_cash), 2)), + ("Call cap", round(_money_to_float(capped_upside_cash), 2)), + ("Hedge cost", round(_money_to_float(-hedge_cost_cash), 2)), + ("Net equity", round(_money_to_float(hedged_equity), 2)), + ] + + return { + "strategy": strategy, + "scenario_pct": scenario_pct, + "scenario_price": round(float(scenario_price), 2), + "scenario_series": [ + {"price": round(float(price), 2), "benefit": benefit} + for price, benefit in zip(scenario_prices, benefits, strict=True) + ], + "waterfall_steps": waterfall_steps, + "unhedged_equity": round(_money_to_float(unhedged_equity), 2), + "hedged_equity": round(_money_to_float(hedged_equity), 2), + } diff --git a/app/pages/common.py b/app/pages/common.py index cc387f7..3ac93ec 100644 --- a/app/pages/common.py +++ b/app/pages/common.py @@ -6,6 +6,7 @@ from typing import Any from nicegui import ui +from app.domain.portfolio_math import portfolio_snapshot_from_config, strategy_metrics_from_snapshot from app.models.portfolio import PortfolioConfig from app.services.strategy_templates import StrategyTemplateService @@ -37,33 +38,7 @@ def demo_spot_price() -> float: def portfolio_snapshot(config: PortfolioConfig | None = None) -> dict[str, float]: - if config is None: - gold_units = 1_000.0 - spot = demo_spot_price() - gold_value = gold_units * spot - loan_amount = 145_000.0 - margin_call_ltv = 0.75 - hedge_budget = 8_000.0 - else: - gold_units = float(config.gold_ounces or 0.0) - spot = float(config.entry_price or 0.0) - gold_value = float(config.gold_value or gold_units * spot) - loan_amount = float(config.loan_amount) - margin_call_ltv = float(config.margin_threshold) - hedge_budget = float(config.monthly_budget) - - return { - "gold_value": gold_value, - "loan_amount": loan_amount, - "ltv_ratio": loan_amount / gold_value, - "net_equity": gold_value - loan_amount, - "spot_price": spot, - "gold_units": gold_units, - "margin_call_ltv": margin_call_ltv, - "margin_call_price": loan_amount / (margin_call_ltv * gold_units), - "cash_buffer": 18_500.0, - "hedge_budget": hedge_budget, - } + return portfolio_snapshot_from_config(config) def strategy_catalog() -> list[dict[str, Any]]: @@ -145,54 +120,7 @@ def strategy_metrics( strategy_catalog()[0], ) portfolio = portfolio or portfolio_snapshot() - spot = float(portfolio["spot_price"]) - underlying_units = portfolio["gold_value"] / spot - loan_amount = float(portfolio["loan_amount"]) - base_equity = float(portfolio["net_equity"]) - floor = float(strategy.get("max_drawdown_floor", spot * 0.95)) - cap = strategy.get("upside_cap") - cost = float(strategy["estimated_cost"]) - - scenario_prices = [round(spot * (1 + pct / 100), 2) for pct in range(-25, 30, 5)] - benefits: list[float] = [] - for price in scenario_prices: - payoff = max(floor - price, 0.0) - if isinstance(cap, (int, float)) and price > float(cap): - payoff -= price - float(cap) - benefits.append(round(payoff - cost, 2)) - - scenario_price = round(spot * (1 + scenario_pct / 100), 2) - unhedged_equity = scenario_price * underlying_units - loan_amount - scenario_payoff_per_unit = max(floor - scenario_price, 0.0) - capped_upside_per_unit = 0.0 - if isinstance(cap, (int, float)) and scenario_price > float(cap): - capped_upside_per_unit = -(scenario_price - float(cap)) - - option_payoff_cash = scenario_payoff_per_unit * underlying_units - capped_upside_cash = capped_upside_per_unit * underlying_units - hedge_cost_cash = cost * underlying_units - hedged_equity = unhedged_equity + option_payoff_cash + capped_upside_cash - hedge_cost_cash - - waterfall_steps = [ - ("Base equity", round(base_equity, 2)), - ("Spot move", round((scenario_price - spot) * underlying_units, 2)), - ("Option payoff", round(option_payoff_cash, 2)), - ("Call cap", round(capped_upside_cash, 2)), - ("Hedge cost", round(-hedge_cost_cash, 2)), - ("Net equity", round(hedged_equity, 2)), - ] - - return { - "strategy": strategy, - "scenario_pct": scenario_pct, - "scenario_price": scenario_price, - "scenario_series": [ - {"price": price, "benefit": benefit} for price, benefit in zip(scenario_prices, benefits, strict=True) - ], - "waterfall_steps": waterfall_steps, - "unhedged_equity": round(unhedged_equity, 2), - "hedged_equity": round(hedged_equity, 2), - } + return strategy_metrics_from_snapshot(strategy, scenario_pct, portfolio) @contextmanager diff --git a/app/services/alerts.py b/app/services/alerts.py index 394adbf..8091d09 100644 --- a/app/services/alerts.py +++ b/app/services/alerts.py @@ -4,6 +4,7 @@ from __future__ import annotations from typing import Mapping +from app.domain.portfolio_math import build_alert_context from app.models.alerts import AlertEvent, AlertHistoryRepository, AlertStatus from app.models.portfolio import PortfolioConfig @@ -15,22 +16,12 @@ def build_portfolio_alert_context( source: str, updated_at: str, ) -> dict[str, float | str]: - gold_units = float(config.gold_ounces or 0.0) - live_gold_value = gold_units * spot_price - loan_amount = float(config.loan_amount) - margin_call_ltv = float(config.margin_threshold) - return { - "spot_price": float(spot_price), - "gold_units": gold_units, - "gold_value": live_gold_value, - "loan_amount": loan_amount, - "ltv_ratio": loan_amount / live_gold_value if live_gold_value > 0 else 0.0, - "net_equity": live_gold_value - loan_amount, - "margin_call_ltv": margin_call_ltv, - "margin_call_price": loan_amount / (margin_call_ltv * gold_units) if gold_units > 0 else 0.0, - "quote_source": source, - "quote_updated_at": updated_at, - } + return build_alert_context( + config, + spot_price=spot_price, + source=source, + updated_at=updated_at, + ) class AlertService: diff --git a/tests/test_e2e_playwright.py b/tests/test_e2e_playwright.py index 1750e59..f692da0 100644 --- a/tests/test_e2e_playwright.py +++ b/tests/test_e2e_playwright.py @@ -193,8 +193,12 @@ def test_homepage_and_options_page_render() -> None: assert "$222,000" in hedge_text assert "80.0%" in hedge_text assert "$12,345" in hedge_text + assert "Scenario spot" in hedge_text + assert "$3,520.00" in hedge_text assert "Unhedged equity" in hedge_text + assert "$552,400" in hedge_text assert "Hedged equity" in hedge_text + assert "$551,025" in hedge_text page.screenshot(path=str(ARTIFACTS / "hedge.png"), full_page=True) browser.close() diff --git a/tests/test_portfolio_math.py b/tests/test_portfolio_math.py new file mode 100644 index 0000000..755b0ad --- /dev/null +++ b/tests/test_portfolio_math.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from app.domain.portfolio_math import ( + build_alert_context, + portfolio_snapshot_from_config, + strategy_metrics_from_snapshot, +) +from app.models.portfolio import PortfolioConfig +from app.pages.common import strategy_catalog + + +def test_portfolio_snapshot_from_config_preserves_weight_price_and_margin_values() -> None: + config = PortfolioConfig(entry_price=4400.0, gold_ounces=220.0, entry_basis_mode="weight", loan_amount=145000.0) + + snapshot = portfolio_snapshot_from_config(config) + + assert snapshot["spot_price"] == 4400.0 + assert snapshot["gold_units"] == 220.0 + assert snapshot["gold_value"] == 968000.0 + assert snapshot["net_equity"] == 823000.0 + assert round(snapshot["margin_call_price"], 2) == 878.79 + + +def test_build_alert_context_uses_unit_safe_gold_value_calculation() -> None: + config = PortfolioConfig(entry_price=4400.0, gold_ounces=220.0, entry_basis_mode="weight", loan_amount=145000.0) + + context = build_alert_context( + config, + spot_price=4400.0, + source="configured_entry_price", + updated_at="", + ) + + assert context["gold_units"] == 220.0 + assert context["gold_value"] == 968000.0 + assert context["net_equity"] == 823000.0 + assert round(float(context["margin_call_price"]), 2) == 878.79 + + +def test_strategy_metrics_from_snapshot_preserves_minus_20pct_protective_put_example() -> None: + strategy = next(item for item in strategy_catalog() if item["name"] == "protective_put_atm") + snapshot = { + "gold_value": 215000.0, + "loan_amount": 145000.0, + "net_equity": 70000.0, + "spot_price": 215.0, + "gold_units": 1000.0, + "margin_call_ltv": 0.75, + "margin_call_price": 193.33, + "cash_buffer": 18500.0, + "hedge_budget": 8000.0, + "ltv_ratio": 145000.0 / 215000.0, + } + + metrics = strategy_metrics_from_snapshot(strategy, -20, snapshot) + + assert metrics["scenario_price"] == 172.0 + assert metrics["unhedged_equity"] == 27000.0 + assert metrics["hedged_equity"] == 58750.0 + assert metrics["waterfall_steps"] == [ + ("Base equity", 70000.0), + ("Spot move", -43000.0), + ("Option payoff", 38000.0), + ("Call cap", 0.0), + ("Hedge cost", -6250.0), + ("Net equity", 58750.0), + ]