import json
import math
import os
import random
import sys
import mpmath
import QuantLib as ql
mpmath.mp.dps = 50
random.seed(42)
SCALE = 10**12
OUTPUT_DIR = os.path.join(os.path.dirname(__file__), '..', 'benchmark')
def quantlib_barrier_exact_T(S, K, H, r, sigma, T, barrier_type, option_type):
try:
today = ql.Date(1, 1, 2025)
ql.Settings.instance().evaluationDate = today
days = max(1, round(T * 365))
maturity = today + ql.Period(int(days), ql.Days)
dc = ql.Actual365Fixed()
actual_T = dc.yearFraction(today, maturity)
spot_handle = ql.QuoteHandle(ql.SimpleQuote(S))
vol_handle = ql.BlackVolTermStructureHandle(
ql.BlackConstantVol(today, ql.NullCalendar(),
ql.QuoteHandle(ql.SimpleQuote(sigma)), dc)
)
rate_handle = ql.YieldTermStructureHandle(
ql.FlatForward(today, ql.QuoteHandle(ql.SimpleQuote(r)), dc)
)
div_handle = ql.YieldTermStructureHandle(
ql.FlatForward(today, 0.0, dc)
)
process = ql.BlackScholesMertonProcess(
spot_handle, div_handle, rate_handle, vol_handle
)
payoff = ql.PlainVanillaPayoff(option_type, K)
exercise = ql.EuropeanExercise(maturity)
barrier_opt = ql.BarrierOption(barrier_type, H, 0.0, payoff, exercise)
barrier_opt.setPricingEngine(ql.AnalyticBarrierEngine(process))
barrier_price = barrier_opt.NPV()
vanilla = ql.EuropeanOption(payoff, exercise)
vanilla.setPricingEngine(ql.AnalyticEuropeanEngine(process))
vanilla_price = vanilla.NPV()
return barrier_price, vanilla_price, actual_T
except Exception:
return None
def mpmath_barrier_down_out_call(S, K, H, r, sigma, T):
S, K, H, r, sigma, T = [mpmath.mpf(x) for x in [S, K, H, r, sigma, T]]
if T <= 0 or sigma <= 0 or S <= 0 or K <= 0 or H <= 0:
return None
if S <= H:
return 0.0, 0.0
if K < H:
return None
def bs_call(spot, strike):
vol_sqrt_t = sigma * mpmath.sqrt(T)
if vol_sqrt_t < mpmath.mpf('1e-30'):
disc = mpmath.exp(-r * T)
return max(spot - strike * disc, 0)
d1 = (mpmath.log(spot / strike) + (r + sigma**2 / 2) * T) / vol_sqrt_t
d2 = d1 - vol_sqrt_t
disc = mpmath.exp(-r * T)
return spot * mpmath.ncdf(d1) - strike * disc * mpmath.ncdf(d2)
alpha = 2 * (r - sigma**2 / 2) / sigma**2
power = (H / S) ** alpha
vanilla = bs_call(S, K)
reflected = bs_call(H**2 / S, K)
return float(vanilla - power * reflected), float(vanilla)
def generate_param_grid():
spots = [50, 80, 100, 120, 150, 200]
sk_ratios = [0.80, 0.90, 0.95, 1.00, 1.05, 1.10, 1.20]
sigmas = [0.10, 0.15, 0.20, 0.25, 0.30, 0.40, 0.50, 0.60]
times = [0.05, 0.10, 0.25, 0.50, 1.00, 2.00]
rates = [0.00, 0.02, 0.05, 0.08, 0.10]
down_hs = [0.70, 0.75, 0.80, 0.85, 0.90, 0.95]
up_hs = [1.05, 1.10, 1.15, 1.20, 1.30, 1.50]
return spots, sk_ratios, sigmas, times, rates, down_hs, up_hs
def generate_vectors():
spots, sk_ratios, sigmas, times, rates, down_hs, up_hs = generate_param_grid()
barrier_configs = [
("down_out_call", ql.Barrier.DownOut, ql.Option.Call, down_hs, True),
("down_in_call", ql.Barrier.DownIn, ql.Option.Call, down_hs, True),
("down_out_put", ql.Barrier.DownOut, ql.Option.Put, down_hs, True),
("down_in_put", ql.Barrier.DownIn, ql.Option.Put, down_hs, True),
("up_out_call", ql.Barrier.UpOut, ql.Option.Call, up_hs, False),
("up_in_call", ql.Barrier.UpIn, ql.Option.Call, up_hs, False),
("up_out_put", ql.Barrier.UpOut, ql.Option.Put, up_hs, False),
("up_in_put", ql.Barrier.UpIn, ql.Option.Put, up_hs, False),
]
all_vectors = []
skipped = 0
total = 0
for name, barrier_type, option_type, hs_ratios, is_down in barrier_configs:
config_vectors = []
is_call = (option_type == ql.Option.Call)
for S in spots:
for sk in sk_ratios:
K = S * sk
for hs in hs_ratios:
H = S * hs
if is_down and S <= H:
continue
if not is_down and S >= H:
continue
if is_call and not is_down and K >= H:
continue
if not is_call and is_down and K <= H:
continue
for sigma in sigmas:
for T in times:
for r in rates:
total += 1
result = quantlib_barrier_exact_T(
S, K, H, r, sigma, T,
barrier_type, option_type
)
if result is None:
skipped += 1
continue
barrier_price, vanilla_price, actual_T = result
s_fp = int(round(S * SCALE))
k_fp = int(round(K * SCALE))
h_fp = int(round(H * SCALE))
r_fp = int(round(r * SCALE))
sigma_fp = int(round(sigma * SCALE))
t_fp = int(round(actual_T * SCALE))
vec = {
"s": str(s_fp),
"k": str(k_fp),
"h": str(h_fp),
"r": str(r_fp),
"sigma": str(sigma_fp),
"t": str(t_fp),
"is_call": is_call,
"barrier_type": name,
"barrier_price": barrier_price,
"vanilla_price": vanilla_price,
"barrier_price_fp": str(int(round(
barrier_price * SCALE))),
"vanilla_price_fp": str(int(round(
vanilla_price * SCALE))),
}
config_vectors.append(vec)
print(f" {name}: {len(config_vectors)} vectors")
all_vectors.extend(config_vectors)
print(f"\nTotal: {len(all_vectors)} vectors "
f"({skipped} skipped out of {total} attempted)")
print("\nCross-validating down-out calls (K>=H) against mpmath 50-digit...")
doc = [v for v in all_vectors
if v['barrier_type'] == 'down_out_call'
and int(v['k']) >= int(v['h'])]
sample = random.sample(doc, min(1000, len(doc)))
max_ulp = 0.0
cross_ok = 0
for v in sample:
S = int(v['s']) / SCALE
K = int(v['k']) / SCALE
H = int(v['h']) / SCALE
r_val = int(v['r']) / SCALE
sigma_val = int(v['sigma']) / SCALE
T = int(v['t']) / SCALE
mp_result = mpmath_barrier_down_out_call(S, K, H, r_val, sigma_val, T)
if mp_result is None:
continue
mp_barrier, _ = mp_result
ql_barrier = v['barrier_price']
ulp = abs(ql_barrier - mp_barrier) * SCALE
max_ulp = max(max_ulp, ulp)
cross_ok += 1
print(f" {cross_ok} vectors, max diff: {max_ulp:.1f} ULP")
from collections import defaultdict
by_params = defaultdict(dict)
for v in all_vectors:
key = (v['s'], v['k'], v['h'], v['r'], v['sigma'], v['t'], v['is_call'])
by_params[key][v['barrier_type']] = v
conservation_checked = 0
max_conservation_err = 0
for key, types in by_params.items():
for in_name, out_name in [
('down_in_call', 'down_out_call'),
('down_in_put', 'down_out_put'),
('up_in_call', 'up_out_call'),
('up_in_put', 'up_out_put'),
]:
if in_name in types and out_name in types:
diff = abs(types[in_name]['barrier_price'] +
types[out_name]['barrier_price'] -
types[out_name]['vanilla_price'])
conservation_checked += 1
max_conservation_err = max(max_conservation_err, diff)
print(f"\nIn/out conservation (QuantLib): {conservation_checked} pairs")
print(f" Max error: ${max_conservation_err:.2e}")
return all_vectors
def main():
print(f"Generating barrier option test vectors")
print(f"QuantLib {ql.__version__} (AnalyticBarrierEngine)")
print(f"mpmath at {mpmath.mp.dps} digits for cross-validation\n")
vectors = generate_vectors()
path = os.path.join(OUTPUT_DIR, 'barrier_vectors.json')
with open(path, 'w') as f:
json.dump({
"meta": {
"source": f"QuantLib {ql.__version__} AnalyticBarrierEngine",
"cross_validation": f"mpmath {mpmath.mp.dps} digits",
"n_vectors": len(vectors),
"scale": SCALE,
"no_dividends": True,
"no_rebate": True,
"T_note": "actual_T from QuantLib Actual365Fixed day count",
},
"vectors": vectors,
}, f, indent=None)
size_mb = os.path.getsize(path) / 1e6
print(f"\nWrote {path} ({len(vectors)} vectors, {size_mb:.1f} MB)")
if __name__ == '__main__':
main()