@njit
def call_equiv_price_numba(is_call, price, df, fwd, strike):
if is_call:
return price
return price + df * (fwd - strike)
@njit
def iv_guess_cm_forward_numba(is_call, price, df, fwd, strike, tau):
a = df * fwd
b = df * strike
c = call_equiv_price_numba(is_call, price, df, fwd, strike)
core = c - 0.5 * (a - b)
disc = core * core - (a - b) * (a - b) / math.pi
if disc > 0.0 and (a + b) > 1e-12:
sigma = math.sqrt(2.0 * math.pi / max(tau, 1e-12)) * (core + math.sqrt(disc)) / (a + b)
else:
sigma = math.sqrt(2.0 * math.pi / max(tau, 1e-12)) * c / max(0.5 * (a + b), 1e-12)
if not np.isfinite(sigma):
sigma = 0.20
return min(max(sigma, 0.01), 4.0)
@njit
def iv_guess_transformed_lite_numba(price, low, df, fwd, strike, tau, log_fk):
time_value = max(price - low, 1e-14)
beta = time_value / max(df * math.sqrt(max(fwd * strike, 1e-14)), 1e-14)
x = abs(log_fk)
total_vol_guess = math.sqrt(2.0 * math.pi) * beta * (1.0 + 0.35 * x + 0.08 * x * x)
sigma = total_vol_guess / math.sqrt(max(tau, 1e-12))
return min(max(sigma, 0.01), 4.0)
@njit
def select_seed_lbr_lite_numba(is_call, price, low, high, guess, df, fwd, strike, tau, sqrt_tau, log_fk):
g0 = min(max(guess, 0.01), 4.0)
g1 = iv_guess_cm_forward_numba(is_call, price, df, fwd, strike, tau)
g2 = iv_guess_transformed_lite_numba(price, low, df, fwd, strike, tau, log_fk)
x = abs(log_fk)
if x < 0.08:
sigma0 = 0.5 * (g0 + g1)
elif x < 0.18:
sigma0 = 0.5 * (g1 + g2)
else:
sigma0 = g2
return min(max(sigma0, 0.01), 4.0)
@njit
def iv_one_lbr_lite_core_numba(is_call, price, low, high, guess, df, fwd, strike, tau, sqrt_tau, log_fk):
if not (np.isfinite(price) and np.isfinite(low) and np.isfinite(high)
and np.isfinite(df) and np.isfinite(fwd) and np.isfinite(strike) and np.isfinite(tau)):
return np.nan, 1, 0
if (price <= 0.0) or (df <= 0.0) or (fwd <= 0.0) or (strike <= 0.0) or (tau <= 0.0):
return np.nan, 1, 0
if (price < low - 1e-10) or (price > high + 1e-10):
return np.nan, 2, 0
lo = 1e-8
hi = 5.0
sigma = select_seed_lbr_lite_numba(is_call, price, low, high, guess, df, fwd, strike, tau, sqrt_tau, log_fk)
sigma = min(max(sigma, 0.01), 4.0)
n_iter = 0
diff = 0.0
for _ in range(4):
n_iter += 1
p = bsm_price_precomputed_numba(is_call, df, fwd, strike, tau, sqrt_tau, log_fk, sigma)
diff = p - price
if abs(diff) < 1e-8:
return sigma, 0, n_iter
if diff > 0.0:
hi = sigma
else:
lo = sigma
v = bsm_vega_precomputed_numba(df, fwd, tau, sqrt_tau, log_fk, sigma)
if (not np.isfinite(v)) or (v < 1e-10):
sigma = 0.5 * (lo + hi)
continue
volga = bsm_volga_precomputed_numba(df, fwd, tau, sqrt_tau, log_fk, sigma)
step = diff / v
denom = 1.0 - 0.5 * step * volga / max(v, 1e-12)
if np.isfinite(denom) and (abs(denom) > 0.35):
nxt = sigma - step / denom
else:
nxt = sigma - step
if (not np.isfinite(nxt)) or (nxt <= lo) or (nxt >= hi):
nxt = 0.5 * (lo + hi)
sigma = nxt
for _ in range(8):
n_iter += 1
p = bsm_price_precomputed_numba(is_call, df, fwd, strike, tau, sqrt_tau, log_fk, sigma)
diff = p - price
if abs(diff) < 1e-8:
return sigma, 0, n_iter
if diff > 0.0:
hi = sigma
else:
lo = sigma
v = bsm_vega_precomputed_numba(df, fwd, tau, sqrt_tau, log_fk, sigma)
if (not np.isfinite(v)) or (v < 1e-10):
sigma = 0.5 * (lo + hi)
else:
nxt = sigma - diff / v
if np.isfinite(nxt) and (lo < nxt < hi):
sigma = nxt
else:
sigma = 0.5 * (lo + hi)
for _ in range(28):
n_iter += 1
sigma = 0.5 * (lo + hi)
p = bsm_price_precomputed_numba(is_call, df, fwd, strike, tau, sqrt_tau, log_fk, sigma)
diff = p - price
if (abs(diff) < 1e-8) or (abs(diff) < 1e-6) or ((hi - lo) < 1e-7):
return sigma, 0, n_iter
if diff > 0.0:
hi = sigma
else:
lo = sigma
if (abs(diff) < 1e-6) or ((hi - lo) < 1e-7):
return sigma, 0, n_iter
return sigma, 3, n_iter
@njit
def iv_one_lbr_lite_numba(is_call, price, low, high, guess, df, fwd, strike, tau, sqrt_tau, log_fk):
sigma, status, _ = iv_one_lbr_lite_core_numba(is_call, price, low, high, guess,
df, fwd, strike, tau, sqrt_tau, log_fk)
return sigma, status
@njit(parallel=True)
def iv_array_lbr_lite_numba(is_call, price, low, high, guess, df, fwd, strike, tau, sqrt_tau, log_fk):
n = len(price)
sigma_out = np.empty(n, dtype=np.float64)
status_out = np.empty(n, dtype=np.int64)
for i in prange(n):
sigma_i, status_i = iv_one_lbr_lite_numba(is_call[i],price[i],low[i],high[i]
,guess[i],df[i],fwd[i],strike[i],
tau[i],sqrt_tau[i],log_fk[i])
sigma_out[i] = sigma_i
status_out[i] = status_i
return sigma_out, status_out
@njit(parallel=True)
def iv_array_lbr_lite_with_iters_numba(is_call, price, low, high, guess, df, fwd, strike, tau, sqrt_tau, log_fk):
n = len(price)
sigma_out = np.empty(n, dtype=np.float64)
status_out = np.empty(n, dtype=np.int64)
iter_out = np.empty(n, dtype=np.int64)
for i in prange(n):
sigma_i, status_i, iter_i = iv_one_lbr_lite_core_numba(is_call[i],price[i],low[i],high[i],
guess[i],df[i],fwd[i],strike[i],
tau[i],sqrt_tau[i],log_fk[i])
sigma_out[i] = sigma_i
status_out[i] = status_i
iter_out[i] = iter_i
return sigma_out, status_out, iter_out
def solve_iv_forward(cp, price, df, fwd, strike, tau, init_sigma=0.20):
tau = float(max(float(tau), 1e-12))
sqrt_tau = float(np.sqrt(tau))
log_fk = float(np.log(max(float(fwd), 1e-12) / max(float(strike), 1e-12)))
low, high = iv_bounds(cp, float(df), float(fwd), float(strike))
low = float(np.asarray(low).reshape(-1)[0])
high = float(np.asarray(high).reshape(-1)[0])
guess = float(np.clip(init_sigma, 0.03, 2.0))
sigma, status = iv_one_precomputed_numba(bool(cp_is_call(cp)),float(price),low,
high,guess,float(df),float(fwd),float(strike),
tau,sqrt_tau,log_fk)
reason = iv_status.get(int(status), "unknown")
return (float(sigma) if int(status) == 0 else np.nan), bool(int(status) == 0), reason
def estimate_forward_slice(df_slice, near_band=0.15):
if df_slice.empty:
return {"ok": False}
tau = float(np.nanmedian(df_slice["tau"]))
trade_date = pd.Timestamp(df_slice["trade_date"].iloc[0]).normalize()
df_tau = float(get_df(trade_date, tau))
strike = df_slice["strike"].to_numpy(dtype=float)
c_bid = df_slice["c_bid"].to_numpy(dtype=float)
c_ask = df_slice["c_ask"].to_numpy(dtype=float)
c_mid = df_slice["c_mid"].to_numpy(dtype=float)
p_bid = df_slice["p_bid"].to_numpy(dtype=float)
p_ask = df_slice["p_ask"].to_numpy(dtype=float)
p_mid = df_slice["p_mid"].to_numpy(dtype=float)
w = 1.0 / np.clip((c_ask - c_bid) + (p_ask - p_bid), 1e-6, None)
f_low = strike + (c_bid - p_ask) / df_tau
f_high = strike + (c_ask - p_bid) / df_tau
f_mid = strike + (c_mid - p_mid) / df_tau
prelim = weighted_median(f_mid, w)
lm = np.log(np.clip(strike, 1e-12, None) / np.clip(prelim, 1e-12, None))
core = np.abs(lm) <= near_band
if core.sum() < 3:
ord_idx = np.argsort(np.abs(lm))
core = np.zeros(len(lm), dtype=bool)
core[ord_idx[: min(6, len(lm))]] = True
f_hat = weighted_median(f_mid[core], w[core])
med = np.nanmedian(f_mid[core])
iqr = np.nanquantile(f_mid[core], 0.75) - np.nanquantile(f_mid[core], 0.25)
dispersion = float(iqr / max(abs(med), 1e-8))
feasibility = float(np.mean((f_hat >= f_low) & (f_hat <= f_high)))
f_low_ref = weighted_median(f_low[core], w[core])
f_high_ref = weighted_median(f_high[core], w[core])
if not np.isfinite(f_low_ref):
f_low_ref = float(np.nanmedian(f_low[core]))
if not np.isfinite(f_high_ref):
f_high_ref = float(np.nanmedian(f_high[core]))
return {
"ok": np.isfinite(f_hat),
"trade_date": trade_date,
"expiry_datetime": pd.Timestamp(df_slice["expiry_datetime"].iloc[0]),
"tau": tau,
"df": df_tau,
"r_short": get_r_short(trade_date),
"f_hat": float(f_hat),
"f_low_ref": float(f_low_ref),
"f_high_ref": float(f_high_ref),
"dispersion": dispersion,
"feasibility": feasibility,
"n_obs": int(len(df_slice)),
"n_core": int(core.sum())}
all_dates = np.sort(df_pairs["trade_date"].dropna().unique())
walk_trade_date = pd.Timestamp(all_dates[len(all_dates) // 2]).normalize()
day_slice = df_pairs[df_pairs["trade_date"] == walk_trade_date].copy()
day_slice["dte_days"] = day_slice["tau"] * 365.25
exp_rank = day_slice.groupby("expiry_datetime")["dte_days"].median().reset_index()
exp_rank["target_dist"] = (exp_rank["dte_days"] - 35.0).abs()
walk_expiry = exp_rank.sort_values("target_dist").iloc[0]["expiry_datetime"]
walk = day_slice[day_slice["expiry_datetime"] == walk_expiry].copy()
walk_est = estimate_forward_slice(walk, near_band=0.12)
walk_df = walk_est["df"]
walk_f_hat = walk_est["f_hat"]
walk["f_low"] = walk["strike"] + (walk["c_bid"] - walk["p_ask"]) / walk_df
walk["f_high"] = walk["strike"] + (walk["c_ask"] - walk["p_bid"]) / walk_df
walk["f_mid"] = walk["strike"] + (walk["c_mid"] - walk["p_mid"]) / walk_df
walk["lm_f"] = np.log(walk["strike"] / walk_f_hat)
fig, ax = plt.subplots(figsize=(8.2, 4.4))
ax.scatter(walk["strike"], walk["f_mid"], s=18, alpha=0.65, label="f_mid(k)")
ax.fill_between(walk["strike"], walk["f_low"], walk["f_high"], alpha=0.13, label="feasible interval")
ax.axhline(walk_f_hat, lw=2, label="f_hat")
ax.set_title(f"single-day parity forward extraction ({walk_trade_date.date()}, {pd.Timestamp(walk_expiry).date()})")
ax.set_xlabel("strike")
ax.set_ylabel("forward level")
ax.legend(loc="best")
plt.tight_layout()
plt.show()
walk_prices = walk["c_mid"].to_numpy(dtype=float)
walk_strikes = walk["strike"].to_numpy(dtype=float)
walk_tau = walk["tau"].to_numpy(dtype=float)
walk_df_arr = np.full(len(walk), float(walk_df))
walk_fwd_arr = np.full(len(walk), float(walk_f_hat))
walk_low, walk_high = iv_bounds("call", walk_df_arr, walk_fwd_arr, walk_strikes)
walk_guess = iv_guess_from_price(walk_prices, walk_low, walk_df_arr, walk_fwd_arr, walk_strikes, walk_tau)
walk_tau = np.clip(walk_tau, 1e-12, None)
walk_sqrt_tau = np.sqrt(walk_tau)
walk_log_fk = np.log(np.clip(walk_fwd_arr, 1e-12, None) / np.clip(walk_strikes, 1e-12, None))
_ = iv_array_numba(np.ones(min(len(walk), 1), dtype=np.bool_), walk_prices[:1], walk_low[:1], walk_high[:1], walk_guess[:1], walk_df_arr[:1], walk_fwd_arr[:1], walk_strikes[:1], walk_tau[:1], walk_sqrt_tau[:1], walk_log_fk[:1])
walk_iv, walk_status = iv_array_numba(
np.ones(len(walk), dtype=np.bool_), walk_prices, walk_low,
walk_high, walk_guess, walk_df_arr, walk_fwd_arr,
walk_strikes, walk_tau, walk_sqrt_tau, walk_log_fk)
df_walk_smile = pd.DataFrame({"lm_f": np.log(walk_strikes / walk_f_hat), "iv_mid": walk_iv, "status": walk_status})
df_walk_smile = df_walk_smile[df_walk_smile["status"] == 0].sort_values("lm_f")
fig, ax = plt.subplots(figsize=(8.2, 4.4))
ax.plot(df_walk_smile["lm_f"], df_walk_smile["iv_mid"], marker="o", lw=1.4, ms=3.5)
ax.set_title("single-day forward-based implied vol skew (mid call)")
ax.set_xlabel("log-moneyness")
ax.set_ylabel("implied vol")
plt.tight_layout()
plt.show()