160 lines
8.1 KiB
Python
160 lines
8.1 KiB
Python
"""
|
|
Saxo-first pricing for Portfolio positions — options legs are priced off this Cockpit's own
|
|
accumulated Saxo option-chain history (services.option_chain, services.vol_surface) whenever
|
|
the position's underlying has a saxo_option_symbol link in the Watchlist (Config ->
|
|
Instruments Watchlist -> "Option"), the SAME real market data Options Lab and Strategy
|
|
Builder already use. Mirrors services.strategy_engine.entry_price()'s own two-tier pattern:
|
|
an exact Saxo bid/ask quote for the listed contract if one happens to exist ("saxo_quote"),
|
|
else the real Saxo-fitted vol smile (services.vol_surface.Surface) priced through
|
|
Black-Scholes ("saxo_surface") — both grounded in real Saxo data, unlike the previous
|
|
unconditional fallback to yfinance's historical realized vol as a stand-in for implied vol
|
|
("yfinance_bs"), which is what silently produced a materially different premium than Saxo's
|
|
real chain (27.2% yfinance-historical vs Saxo's real ~32% chain IV on the ^NDX example that
|
|
prompted this).
|
|
|
|
An exact "saxo_quote" match is rare in practice: positions carry a nominal expiry_date/
|
|
expiry_days the AI or user chose freely, not necessarily a real listed Saxo expiry — so
|
|
most legs land on "saxo_surface" (real Saxo-implied vol, interpolated to the requested
|
|
strike/tenor) rather than a literal listed-contract quote. That's still a real improvement
|
|
over yfinance historical vol, and every priced leg carries its `source` so the Portfolio UI
|
|
can say plainly which basis was used instead of always labeling everything "Black-Scholes"
|
|
regardless of where the inputs actually came from.
|
|
"""
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
from services.options_pricer import black_scholes
|
|
|
|
|
|
def resolve_saxo_chain(underlying: str, target_days: int) -> Tuple[Optional[Dict[str, Any]], Optional[Any]]:
|
|
"""Returns (chain_slice, Surface) for this underlying's linked Saxo option chain, or
|
|
(None, None) if it isn't linked, or the chain can't be built right now (Saxo down, no
|
|
snapshot yet, entitlement gap, etc.) — callers fall back to yfinance pricing in that case."""
|
|
from services.database import get_saxo_option_symbol_for_ticker
|
|
saxo_symbol = get_saxo_option_symbol_for_ticker(underlying)
|
|
if not saxo_symbol:
|
|
return None, None
|
|
try:
|
|
from services.option_chain import get_chain_slice
|
|
from services.vol_surface import Surface
|
|
chain = get_chain_slice(saxo_symbol, target_days=max(target_days, 1))
|
|
if not chain.get("spot"):
|
|
return None, None
|
|
surface = Surface(chain["spot"], chain["expiries"])
|
|
return chain, surface
|
|
except Exception:
|
|
return None, None
|
|
|
|
|
|
def price_leg(
|
|
strike: float, option_type: str, days_to_expiry: float, r: float,
|
|
chain: Optional[Dict[str, Any]], surface: Optional[Any], expiry_date: Optional[str],
|
|
fallback_spot: float, fallback_sigma: float,
|
|
) -> Dict[str, Any]:
|
|
"""One leg's {price, spot, sigma, source}."""
|
|
T = max(days_to_expiry, 1) / 365
|
|
if chain is not None and surface is not None:
|
|
quote = None
|
|
if expiry_date:
|
|
from services.option_chain import find_quote
|
|
quote = find_quote(chain, expiry_date, strike, option_type)
|
|
if quote and quote.get("bid", 0) > 0 and quote.get("ask", 0) > 0:
|
|
return {"price": quote["mid"], "spot": chain["spot"], "sigma": quote["iv"], "source": "saxo_quote"}
|
|
sigma = surface.iv_at(strike, max(days_to_expiry, 1))
|
|
price = black_scholes(chain["spot"], strike, T, r, sigma, option_type)["price"]
|
|
return {"price": price, "spot": chain["spot"], "sigma": sigma, "source": "saxo_surface"}
|
|
price = black_scholes(fallback_spot, strike, T, r, fallback_sigma, option_type)["price"]
|
|
return {"price": price, "spot": fallback_spot, "sigma": fallback_sigma, "source": "yfinance_bs"}
|
|
|
|
|
|
SOURCE_LABELS = {
|
|
"saxo_quote": "Cotation Saxo réelle",
|
|
"saxo_surface": "Surface de vol Saxo (réelle)",
|
|
"yfinance_bs": "Black-Scholes (vol historique yfinance)",
|
|
}
|
|
|
|
|
|
def _intrinsic(S: float, K: float, option_type: str) -> float:
|
|
return max(0.0, S - K) if option_type == "call" else max(0.0, K - S)
|
|
|
|
|
|
def compute_payoff(pos: Dict[str, Any], n_points: int = 61, range_pct: float = 0.25) -> Dict[str, Any]:
|
|
"""P&L vs. underlying price across a ±range_pct band around the current spot — two
|
|
curves: "at_expiry" (pure intrinsic value, no vol at all — the textbook payoff diagram)
|
|
and "today" (Black-Scholes reprice at each hypothetical spot, holding each leg's
|
|
CURRENT implied vol fixed — from the real Saxo surface when linked, so the time-value
|
|
bulge/skew asymmetry actually reflects Saxo's real market vol instead of a flat
|
|
textbook number). Both curves net out entry cost and entry fees, so y=0 is genuine
|
|
breakeven, matching what the Position card's PnL already shows at the current spot."""
|
|
from datetime import date, datetime
|
|
from services.data_fetcher import get_quote
|
|
|
|
underlying = pos["underlying"]
|
|
legs = pos.get("legs", [])
|
|
if not legs:
|
|
return {"spot_range": [], "at_expiry": [], "today": [], "current_spot": None,
|
|
"entry_spot": pos.get("entry_underlying_price"), "strikes": [], "pricing_source": None}
|
|
|
|
expiry_date = pos.get("expiry_date") or ""
|
|
if expiry_date:
|
|
try:
|
|
exp = datetime.strptime(expiry_date[:10], "%Y-%m-%d").date()
|
|
days_remaining = max(0, (exp - date.today()).days)
|
|
except ValueError:
|
|
days_remaining = 0
|
|
else:
|
|
entry = datetime.strptime(pos["entry_date"][:10], "%Y-%m-%d").date()
|
|
days_remaining = max(0, pos.get("expiry_days", 90) - (date.today() - entry).days)
|
|
|
|
r = 0.05
|
|
chain, surface = resolve_saxo_chain(underlying, target_days=max(days_remaining, 1))
|
|
fallback_spot = pos.get("entry_underlying_price") or 100.0
|
|
fallback_sigma = 0.20
|
|
if chain is None:
|
|
q = get_quote(underlying)
|
|
fallback_spot = (q.get("price") if q else None) or fallback_spot
|
|
from services.data_fetcher import compute_historical_iv
|
|
fallback_sigma = compute_historical_iv(underlying)
|
|
S = chain["spot"] if chain else fallback_spot
|
|
|
|
resolved_legs = []
|
|
for leg in legs:
|
|
K = leg.get("strike") or S
|
|
opt_type = leg.get("option_type", "call")
|
|
entry_premium = leg.get("premium_paid")
|
|
if entry_premium is None:
|
|
entry_premium = price_leg(K, opt_type, pos.get("expiry_days", 90), r, chain, surface,
|
|
expiry_date, fallback_spot, fallback_sigma)["price"]
|
|
priced_now = price_leg(K, opt_type, days_remaining, r, chain, surface, expiry_date, fallback_spot, fallback_sigma)
|
|
resolved_legs.append({
|
|
"strike": K, "option_type": opt_type, "qty": leg.get("quantity", 1),
|
|
"sign": 1 if leg.get("position", "long") == "long" else -1,
|
|
"entry_premium": entry_premium, "sigma": priced_now["sigma"],
|
|
})
|
|
|
|
ib_entry = pos.get("ib_fees_entry", 0)
|
|
lo, hi = S * (1 - range_pct), S * (1 + range_pct)
|
|
spot_range = [lo + (hi - lo) * i / (n_points - 1) for i in range(n_points)]
|
|
T_remaining = days_remaining / 365
|
|
|
|
at_expiry, today = [], []
|
|
for Sx in spot_range:
|
|
pnl_exp = -ib_entry
|
|
pnl_today = -ib_entry
|
|
for leg in resolved_legs:
|
|
pnl_exp += leg["sign"] * leg["qty"] * 100 * (_intrinsic(Sx, leg["strike"], leg["option_type"]) - leg["entry_premium"])
|
|
bs_price = (black_scholes(Sx, leg["strike"], T_remaining, r, leg["sigma"], leg["option_type"])["price"]
|
|
if T_remaining > 0 else _intrinsic(Sx, leg["strike"], leg["option_type"]))
|
|
pnl_today += leg["sign"] * leg["qty"] * 100 * (bs_price - leg["entry_premium"])
|
|
at_expiry.append(round(pnl_exp, 2))
|
|
today.append(round(pnl_today, 2))
|
|
|
|
return {
|
|
"spot_range": [round(s, 4) for s in spot_range],
|
|
"at_expiry": at_expiry,
|
|
"today": today,
|
|
"current_spot": round(S, 4),
|
|
"entry_spot": pos.get("entry_underlying_price"),
|
|
"strikes": sorted({leg["strike"] for leg in resolved_legs}),
|
|
"pricing_source": "saxo" if chain else "yfinance_bs",
|
|
}
|