vault-dash/app/core/calculations.py

from __future__ import annotations

from collections.abc import Iterable, Mapping
from datetime import date, datetime
from typing import cast

from app.core.pricing.black_scholes import (
    DEFAULT_RISK_FREE_RATE,
    DEFAULT_VOLATILITY,
    BlackScholesInputs,
    OptionType,
    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_typed: OptionType = cast(OptionType, option_type)
    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_typed,
                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,
    }