00a68bc767
- FastAPI + NiceGUI web application - QuantLib-based Black-Scholes pricing with Greeks - Protective put, laddered, and LEAPS strategies - Real-time WebSocket updates - TradingView-style charts via Lightweight-Charts - Docker containerization - GitLab CI/CD pipeline for VPS deployment - VPN-only access configuration
211 lines
7.2 KiB
Python
211 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date, timedelta
|
|
import math
|
|
from typing import Any, Literal
|
|
|
|
try: # pragma: no cover - optional dependency
|
|
import QuantLib as ql
|
|
except ImportError: # pragma: no cover - optional dependency
|
|
ql = None
|
|
|
|
OptionType = Literal["call", "put"]
|
|
|
|
DEFAULT_GOLD_PRICE_PER_OUNCE: float = 4600.0
|
|
DEFAULT_GLD_PRICE: float = 460.0
|
|
DEFAULT_RISK_FREE_RATE: float = 0.045
|
|
DEFAULT_VOLATILITY: float = 0.16
|
|
DEFAULT_DIVIDEND_YIELD: float = 0.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BlackScholesInputs:
|
|
"""Inputs for European Black-Scholes pricing."""
|
|
|
|
spot: float = DEFAULT_GLD_PRICE
|
|
strike: float = DEFAULT_GLD_PRICE
|
|
time_to_expiry: float = 0.25
|
|
risk_free_rate: float = DEFAULT_RISK_FREE_RATE
|
|
volatility: float = DEFAULT_VOLATILITY
|
|
option_type: OptionType = "put"
|
|
dividend_yield: float = DEFAULT_DIVIDEND_YIELD
|
|
valuation_date: date | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PricingResult:
|
|
"""European option price and Greeks."""
|
|
|
|
price: float
|
|
delta: float
|
|
gamma: float
|
|
theta: float
|
|
vega: float
|
|
rho: float
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HedgingCost:
|
|
"""Annualized hedging cost summary."""
|
|
|
|
premium_paid: float
|
|
annual_cost_dollars: float
|
|
annual_cost_pct: float
|
|
|
|
|
|
def _validate_option_type(option_type: str) -> OptionType:
|
|
option = option_type.lower()
|
|
if option not in {"call", "put"}:
|
|
raise ValueError("option_type must be either 'call' or 'put'")
|
|
return option # type: ignore[return-value]
|
|
|
|
|
|
def _to_quantlib_option_type(option_type: OptionType) -> Any:
|
|
if ql is None:
|
|
raise RuntimeError("QuantLib is not installed")
|
|
return ql.Option.Call if option_type == "call" else ql.Option.Put
|
|
|
|
|
|
def _build_dates(time_to_expiry: float, valuation_date: date | None) -> tuple[Any, Any]:
|
|
if time_to_expiry <= 0.0:
|
|
raise ValueError("time_to_expiry must be positive")
|
|
if ql is None:
|
|
return (None, None)
|
|
|
|
valuation = valuation_date or date.today()
|
|
maturity = valuation + timedelta(days=max(1, round(time_to_expiry * 365)))
|
|
return (
|
|
ql.Date(valuation.day, valuation.month, valuation.year),
|
|
ql.Date(maturity.day, maturity.month, maturity.year),
|
|
)
|
|
|
|
|
|
def _norm_pdf(value: float) -> float:
|
|
return math.exp(-(value**2) / 2.0) / math.sqrt(2.0 * math.pi)
|
|
|
|
|
|
def _norm_cdf(value: float) -> float:
|
|
return 0.5 * (1.0 + math.erf(value / math.sqrt(2.0)))
|
|
|
|
|
|
def _analytic_black_scholes(params: BlackScholesInputs, option_type: OptionType) -> PricingResult:
|
|
if params.spot <= 0 or params.strike <= 0 or params.time_to_expiry <= 0 or params.volatility <= 0:
|
|
raise ValueError("spot, strike, time_to_expiry, and volatility must be positive")
|
|
|
|
t = params.time_to_expiry
|
|
sigma = params.volatility
|
|
sqrt_t = math.sqrt(t)
|
|
d1 = (
|
|
math.log(params.spot / params.strike)
|
|
+ (params.risk_free_rate - params.dividend_yield + 0.5 * sigma**2) * t
|
|
) / (sigma * sqrt_t)
|
|
d2 = d1 - sigma * sqrt_t
|
|
disc_r = math.exp(-params.risk_free_rate * t)
|
|
disc_q = math.exp(-params.dividend_yield * t)
|
|
pdf_d1 = _norm_pdf(d1)
|
|
|
|
if option_type == "call":
|
|
price = params.spot * disc_q * _norm_cdf(d1) - params.strike * disc_r * _norm_cdf(d2)
|
|
delta = disc_q * _norm_cdf(d1)
|
|
theta = (
|
|
-(params.spot * disc_q * pdf_d1 * sigma) / (2 * sqrt_t)
|
|
- params.risk_free_rate * params.strike * disc_r * _norm_cdf(d2)
|
|
+ params.dividend_yield * params.spot * disc_q * _norm_cdf(d1)
|
|
)
|
|
rho = params.strike * t * disc_r * _norm_cdf(d2)
|
|
else:
|
|
price = params.strike * disc_r * _norm_cdf(-d2) - params.spot * disc_q * _norm_cdf(-d1)
|
|
delta = disc_q * (_norm_cdf(d1) - 1.0)
|
|
theta = (
|
|
-(params.spot * disc_q * pdf_d1 * sigma) / (2 * sqrt_t)
|
|
+ params.risk_free_rate * params.strike * disc_r * _norm_cdf(-d2)
|
|
- params.dividend_yield * params.spot * disc_q * _norm_cdf(-d1)
|
|
)
|
|
rho = -params.strike * t * disc_r * _norm_cdf(-d2)
|
|
|
|
gamma = (disc_q * pdf_d1) / (params.spot * sigma * sqrt_t)
|
|
vega = params.spot * disc_q * pdf_d1 * sqrt_t
|
|
return PricingResult(price=float(price), delta=float(delta), gamma=float(gamma), theta=float(theta), vega=float(vega), rho=float(rho))
|
|
|
|
|
|
def black_scholes_price_and_greeks(params: BlackScholesInputs) -> PricingResult:
|
|
"""Price a European option with QuantLib when available, otherwise analytic BSM."""
|
|
|
|
option_type = _validate_option_type(params.option_type)
|
|
if ql is None:
|
|
return _analytic_black_scholes(params, option_type)
|
|
|
|
valuation_ql, maturity_ql = _build_dates(params.time_to_expiry, params.valuation_date)
|
|
ql.Settings.instance().evaluationDate = valuation_ql
|
|
|
|
day_count = ql.Actual365Fixed()
|
|
calendar = ql.NullCalendar()
|
|
|
|
spot_handle = ql.QuoteHandle(ql.SimpleQuote(params.spot))
|
|
dividend_curve = ql.YieldTermStructureHandle(
|
|
ql.FlatForward(valuation_ql, params.dividend_yield, day_count)
|
|
)
|
|
risk_free_curve = ql.YieldTermStructureHandle(
|
|
ql.FlatForward(valuation_ql, params.risk_free_rate, day_count)
|
|
)
|
|
volatility = ql.BlackVolTermStructureHandle(
|
|
ql.BlackConstantVol(valuation_ql, calendar, params.volatility, day_count)
|
|
)
|
|
|
|
process = ql.BlackScholesMertonProcess(spot_handle, dividend_curve, risk_free_curve, volatility)
|
|
payoff = ql.PlainVanillaPayoff(_to_quantlib_option_type(option_type), params.strike)
|
|
exercise = ql.EuropeanExercise(maturity_ql)
|
|
option = ql.VanillaOption(payoff, exercise)
|
|
option.setPricingEngine(ql.AnalyticEuropeanEngine(process))
|
|
|
|
return PricingResult(
|
|
price=float(option.NPV()),
|
|
delta=float(option.delta()),
|
|
gamma=float(option.gamma()),
|
|
theta=float(option.theta()),
|
|
vega=float(option.vega()),
|
|
rho=float(option.rho()),
|
|
)
|
|
|
|
|
|
def margin_call_threshold_price(
|
|
portfolio_value: float,
|
|
loan_amount: float,
|
|
current_price: float = DEFAULT_GLD_PRICE,
|
|
margin_call_ltv: float = 0.75,
|
|
) -> float:
|
|
"""Calculate the underlying price where a margin call is triggered."""
|
|
|
|
if portfolio_value <= 0 or loan_amount <= 0 or current_price <= 0:
|
|
raise ValueError("portfolio_value, loan_amount, and current_price must be positive")
|
|
if not 0 < margin_call_ltv < 1:
|
|
raise ValueError("margin_call_ltv must be between 0 and 1")
|
|
|
|
units = portfolio_value / current_price
|
|
return loan_amount / (margin_call_ltv * units)
|
|
|
|
|
|
def annual_hedging_cost(
|
|
premium_per_share: float,
|
|
shares_hedged: float,
|
|
portfolio_value: float,
|
|
hedge_term_years: float,
|
|
) -> HedgingCost:
|
|
"""Annualize the premium cost of a hedging program."""
|
|
|
|
if premium_per_share < 0 or shares_hedged <= 0 or portfolio_value <= 0 or hedge_term_years <= 0:
|
|
raise ValueError(
|
|
"premium_per_share must be non-negative and shares_hedged, portfolio_value, "
|
|
"and hedge_term_years must be positive"
|
|
)
|
|
|
|
premium_paid = premium_per_share * shares_hedged
|
|
annual_cost_dollars = premium_paid / hedge_term_years
|
|
annual_cost_pct = annual_cost_dollars / portfolio_value
|
|
return HedgingCost(
|
|
premium_paid=premium_paid,
|
|
annual_cost_dollars=annual_cost_dollars,
|
|
annual_cost_pct=annual_cost_pct,
|
|
)
|