feat(DATA-003): calculate live option greeks

app/core/calculations.py

@@ -1,7 +1,14 @@
 from __future__ import annotations
 
-from collections.abc import Iterable
+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
@@ -96,3 +103,77 @@ def portfolio_net_equity(
         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)
+
+    try:
+        strike = float(row.get("strike", 0.0))
+    except (TypeError, ValueError):
+        return dict(_ZERO_GREEKS)
+    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)
+
+    try:
+        implied_volatility = float(row.get("impliedVolatility", 0.0) or 0.0)
+    except (TypeError, ValueError):
+        implied_volatility = 0.0
+    volatility = implied_volatility if implied_volatility > 0 else DEFAULT_VOLATILITY
+
+    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,
+                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,
+    }