feat(CORE-001B): migrate overview and hedge math to unit types
This commit is contained in:
Bu5hm4nn
@@ -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",
|
||||
]
|
||||
|
||||
173
app/domain/portfolio_math.py
Normal file
173
app/domain/portfolio_math.py
Normal file
@@ -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),
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user