887565be74
- Fix return type annotation for get_default_premium_for_product - Add type narrowing for Weight|Money union using _as_money helper - Add isinstance checks before float() calls for object types - Add type guard for Decimal.exponent comparison - Use _unit_typed and _currency_typed properties for type narrowing - Cast option_type to OptionType Literal after validation - Fix provider type hierarchy in backtesting services - Add types-requests to dev dependencies - Remove '|| true' from CI type-check job All 36 mypy errors resolved across 15 files.
428 lines
17 KiB
Python
428 lines
17 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from decimal import Decimal, InvalidOperation
|
||
from typing import Any, Mapping
|
||
|
||
from app.domain.backtesting_math import PricePerAsset
|
||
from app.domain.conversions import is_gld_mode
|
||
from app.domain.instruments import gld_ounces_per_share, instrument_metadata
|
||
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 _as_money(value: Weight | Money) -> Money:
|
||
"""Narrow Weight | Money to Money after multiplication."""
|
||
if isinstance(value, Money):
|
||
return value
|
||
raise TypeError(f"Expected Money, got {type(value).__name__}")
|
||
|
||
|
||
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 _safe_quote_price(value: object) -> float:
|
||
try:
|
||
if isinstance(value, (int, float)):
|
||
parsed = float(value)
|
||
elif isinstance(value, str):
|
||
parsed = float(value.strip())
|
||
else:
|
||
return 0.0
|
||
except (TypeError, ValueError):
|
||
return 0.0
|
||
if parsed <= 0:
|
||
return 0.0
|
||
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)
|
||
) or Decimal("0")
|
||
|
||
|
||
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_collateral_spot_from_quote(
|
||
quote: Mapping[str, object],
|
||
*,
|
||
fallback_symbol: str | None = None,
|
||
) -> tuple[float, str, str] | None:
|
||
quote_price = _safe_quote_price(quote.get("price"))
|
||
quote_source = str(quote.get("source", "unknown"))
|
||
quote_updated_at = str(quote.get("updated_at", ""))
|
||
quote_symbol = str(quote.get("symbol", fallback_symbol or "")).strip().upper()
|
||
quote_unit = str(quote.get("quote_unit", "")).strip().lower()
|
||
|
||
if quote_price <= 0 or not quote_symbol or quote_unit != "share":
|
||
return None
|
||
|
||
try:
|
||
metadata = instrument_metadata(quote_symbol)
|
||
except ValueError:
|
||
return None
|
||
|
||
converted_spot = metadata.price_per_weight_from_asset_price(
|
||
PricePerAsset(amount=decimal_from_float(quote_price), currency=BaseCurrency.USD, symbol=quote_symbol),
|
||
per_unit=WeightUnit.OUNCE_TROY,
|
||
)
|
||
return _decimal_to_float(converted_spot.amount), quote_source, quote_updated_at
|
||
|
||
|
||
def resolve_portfolio_spot_from_quote(
|
||
config: PortfolioConfig,
|
||
quote: Mapping[str, object],
|
||
*,
|
||
fallback_symbol: str | None = None,
|
||
) -> tuple[float, str, str]:
|
||
"""Resolve spot price from quote based on display mode.
|
||
|
||
In GLD display mode: return GLD share price directly (no conversion)
|
||
In XAU display mode: convert GLD share price to oz-equivalent using expense-adjusted backing
|
||
|
||
Args:
|
||
config: Portfolio configuration with display_mode setting
|
||
quote: Quote data from data service
|
||
fallback_symbol: Fallback symbol if quote lacks symbol
|
||
|
||
Returns:
|
||
Tuple of (spot_price, source, updated_at)
|
||
"""
|
||
display_mode = getattr(config, "display_mode", "XAU")
|
||
|
||
# Try to resolve from quote first
|
||
resolved = resolve_collateral_spot_from_quote(quote, fallback_symbol=fallback_symbol)
|
||
|
||
if resolved is None:
|
||
# No valid quote, use configured entry price
|
||
configured_price = float(config.entry_price or 0.0)
|
||
return configured_price, "configured_entry_price", ""
|
||
|
||
spot_price, source, updated_at = resolved
|
||
|
||
# In GLD mode, return share price directly (no conversion to oz)
|
||
if is_gld_mode(display_mode):
|
||
# For GLD mode, we want the share price, not the converted oz price
|
||
quote_price = _safe_quote_price(quote.get("price"))
|
||
if quote_price > 0:
|
||
return quote_price, source, updated_at
|
||
|
||
# XAU mode: use the converted oz-equivalent price (already done in resolve_collateral_spot_from_quote)
|
||
return spot_price, source, updated_at
|
||
|
||
|
||
def portfolio_snapshot_from_config(
|
||
config: PortfolioConfig | None = None,
|
||
*,
|
||
runtime_spot_price: float | None = None,
|
||
) -> dict[str, float | str]:
|
||
"""Build portfolio snapshot with display-mode-aware calculations.
|
||
|
||
In GLD mode:
|
||
- gold_units: shares (not oz)
|
||
- spot_price: GLD share price
|
||
- gold_value: shares × share_price
|
||
|
||
In XAU mode:
|
||
- gold_units: oz
|
||
- spot_price: USD/oz
|
||
- gold_value: oz × oz_price
|
||
"""
|
||
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)
|
||
display_mode = "XAU"
|
||
else:
|
||
display_mode = getattr(config, "display_mode", "XAU")
|
||
resolved_spot = runtime_spot_price if runtime_spot_price is not None else float(config.entry_price or 0.0)
|
||
|
||
if is_gld_mode(display_mode):
|
||
# GLD mode: work with shares directly
|
||
# Use positions if available, otherwise fall back to legacy gold_ounces as shares
|
||
if config.positions:
|
||
# Sum GLD position quantities in shares
|
||
total_shares = Decimal("0")
|
||
for pos in config.positions:
|
||
if pos.underlying == "GLD" and pos.unit == "shares":
|
||
total_shares += pos.quantity
|
||
elif pos.underlying == "GLD" and pos.unit == "oz":
|
||
# Convert oz to shares using current backing
|
||
backing = gld_ounces_per_share(date.today())
|
||
total_shares += pos.quantity / backing
|
||
else:
|
||
# Non-GLD positions: treat as oz and convert to shares
|
||
backing = gld_ounces_per_share(date.today())
|
||
total_shares += pos.quantity / backing
|
||
|
||
gold_weight = Weight(amount=total_shares, unit=WeightUnit.OUNCE_TROY) # Store shares in weight for now
|
||
else:
|
||
# Legacy: treat gold_ounces as oz, convert to shares
|
||
backing = gld_ounces_per_share(date.today())
|
||
shares = Decimal(str(config.gold_ounces or 0.0)) / backing
|
||
gold_weight = Weight(amount=shares, unit=WeightUnit.OUNCE_TROY)
|
||
|
||
spot = PricePerWeight(
|
||
amount=decimal_from_float(resolved_spot), currency=BaseCurrency.USD, per_unit=WeightUnit.OUNCE_TROY
|
||
)
|
||
else:
|
||
# XAU mode: work with oz
|
||
gold_weight = _gold_weight(float(config.gold_ounces or 0.0))
|
||
spot = _spot_price(resolved_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))
|
||
hedge_budget = Money(amount=decimal_from_float(float(config.monthly_budget)), currency=BaseCurrency.USD)
|
||
|
||
gold_value = _as_money(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),
|
||
"display_mode": display_mode,
|
||
}
|
||
|
||
|
||
def build_alert_context(
|
||
config: PortfolioConfig,
|
||
*,
|
||
spot_price: float,
|
||
source: str,
|
||
updated_at: str,
|
||
) -> dict[str, float | str]:
|
||
"""Build alert context with display-mode-aware calculations."""
|
||
display_mode = getattr(config, "display_mode", "XAU")
|
||
|
||
if is_gld_mode(display_mode):
|
||
# GLD mode: work with shares
|
||
backing = gld_ounces_per_share(date.today())
|
||
shares = Decimal(str(config.gold_ounces or 0.0)) / backing
|
||
gold_weight = Weight(amount=shares, unit=WeightUnit.OUNCE_TROY)
|
||
else:
|
||
# XAU mode: work with oz
|
||
gold_weight = _gold_weight(float(config.gold_ounces or 0.0))
|
||
|
||
live_spot = _spot_price(spot_price)
|
||
gold_value = _as_money(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,
|
||
"display_mode": display_mode,
|
||
}
|
||
|
||
|
||
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)
|
||
cost = _strategy_decimal(strategy.get("estimated_cost")) or _DECIMAL_ZERO
|
||
|
||
scenario_prices = [spot * _pct_factor(pct) for pct in range(-25, 30, 5)]
|
||
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 = _as_money(gold_weight * PricePerWeight(
|
||
amount=scenario_price,
|
||
currency=BaseCurrency.USD,
|
||
per_unit=WeightUnit.OUNCE_TROY,
|
||
))
|
||
current_gold_value = _as_money(gold_weight * current_spot)
|
||
unhedged_equity = scenario_gold_value - loan_amount
|
||
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)
|
||
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),
|
||
}
|