from __future__ import annotations from collections.abc import Iterable, Mapping from datetime import date, datetime from app.core.pricing.black_scholes import ( DEFAULT_RISK_FREE_RATE, DEFAULT_VOLATILITY, BlackScholesInputs, black_scholes_price_and_greeks, ) from app.models.option import OptionContract from app.models.portfolio import LombardPortfolio from app.models.strategy import HedgingStrategy def margin_call_price(gold_ounces: float, loan_amount: float, margin_call_ltv: float) -> float: """Calculate the gold price per ounce that triggers a margin call.""" if gold_ounces <= 0: raise ValueError("gold_ounces must be positive") if loan_amount < 0: raise ValueError("loan_amount must be non-negative") if not 0 < margin_call_ltv < 1: raise ValueError("margin_call_ltv must be between 0 and 1") return loan_amount / (margin_call_ltv * gold_ounces) def loan_to_value(loan_amount: float, collateral_value: float) -> float: """Calculate the loan-to-value ratio.""" if loan_amount < 0: raise ValueError("loan_amount must be non-negative") if collateral_value <= 0: raise ValueError("collateral_value must be positive") return loan_amount / collateral_value def ltv_scenarios(portfolio: LombardPortfolio, gold_prices: Iterable[float]) -> dict[float, float]: """Return LTV values for a collection of gold-price scenarios.""" scenarios: dict[float, float] = {} for price in gold_prices: if price <= 0: raise ValueError("scenario gold prices must be positive") scenarios[price] = portfolio.ltv_at_price(price) if not scenarios: raise ValueError("gold_prices must contain at least one scenario") return scenarios def option_payoff(contracts: Iterable[OptionContract], underlying_price: float, *, short: bool = False) -> float: """Aggregate expiry payoff across option contracts.""" if underlying_price <= 0: raise ValueError("underlying_price must be positive") payoff = sum(contract.payoff(underlying_price) for contract in contracts) return -payoff if short else payoff def strategy_payoff(strategy: HedgingStrategy, underlying_price: float) -> float: """Net option payoff before premium cost for a hedging strategy.""" return strategy.gross_payoff(underlying_price) def net_equity( gold_ounces: float, gold_price_per_ounce: float, loan_amount: float, hedge_cost: float = 0.0, option_payoff_value: float = 0.0, ) -> float: """Calculate net equity after debt and hedging effects. Formula: ``gold_value - loan_amount - hedge_cost + option_payoff`` """ if gold_ounces <= 0: raise ValueError("gold_ounces must be positive") if gold_price_per_ounce <= 0: raise ValueError("gold_price_per_ounce must be positive") if loan_amount < 0: raise ValueError("loan_amount must be non-negative") if hedge_cost < 0: raise ValueError("hedge_cost must be non-negative") gold_value = gold_ounces * gold_price_per_ounce return gold_value - loan_amount - hedge_cost + option_payoff_value def portfolio_net_equity( portfolio: LombardPortfolio, gold_price_per_ounce: float | None = None, strategy: HedgingStrategy | None = None, ) -> float: """Calculate scenario net equity for a portfolio with an optional hedge.""" scenario_price = portfolio.gold_price_per_ounce if gold_price_per_ounce is None else gold_price_per_ounce if scenario_price <= 0: raise ValueError("gold_price_per_ounce must be positive") payoff_value = strategy.gross_payoff(scenario_price) if strategy is not None else 0.0 hedge_cost = strategy.hedge_cost if strategy is not None else 0.0 return net_equity( gold_ounces=portfolio.gold_ounces, gold_price_per_ounce=scenario_price, loan_amount=portfolio.loan_amount, hedge_cost=hedge_cost, option_payoff_value=payoff_value, ) _ZERO_GREEKS = {"delta": 0.0, "gamma": 0.0, "theta": 0.0, "vega": 0.0, "rho": 0.0} def option_row_greeks( row: Mapping[str, object], underlying_price: float, *, risk_free_rate: float = DEFAULT_RISK_FREE_RATE, valuation_date: date | None = None, ) -> dict[str, float]: """Calculate Black-Scholes Greeks for an option-chain row. Prefers live implied volatility when available. If it is missing or invalid, a conservative default volatility is used. Invalid or expired rows return zero Greeks instead of raising. """ if underlying_price <= 0: return dict(_ZERO_GREEKS) strike_raw = row.get("strike", 0.0) strike = float(strike_raw) if isinstance(strike_raw, (int, float)) else 0.0 if strike <= 0: return dict(_ZERO_GREEKS) option_type = str(row.get("type", "")).lower() if option_type not in {"call", "put"}: return dict(_ZERO_GREEKS) expiry_raw = row.get("expiry") if not isinstance(expiry_raw, str) or not expiry_raw: return dict(_ZERO_GREEKS) try: expiry = datetime.fromisoformat(expiry_raw).date() except ValueError: return dict(_ZERO_GREEKS) valuation = valuation_date or date.today() days_to_expiry = (expiry - valuation).days if days_to_expiry <= 0: return dict(_ZERO_GREEKS) iv_raw = row.get("impliedVolatility", 0.0) or 0.0 implied_volatility = float(iv_raw) if isinstance(iv_raw, (int, float)) else 0.0 volatility = implied_volatility if implied_volatility > 0 else DEFAULT_VOLATILITY # option_type is validated to be in {"call", "put"} above, so it's safe to pass try: pricing = black_scholes_price_and_greeks( BlackScholesInputs( spot=underlying_price, strike=strike, time_to_expiry=days_to_expiry / 365.0, risk_free_rate=risk_free_rate, volatility=volatility, option_type=option_type, # type: ignore[arg-type] valuation_date=valuation, ) ) except ValueError: return dict(_ZERO_GREEKS) return { "delta": pricing.delta, "gamma": pricing.gamma, "theta": pricing.theta, "vega": pricing.vega, "rho": pricing.rho, }