working cva computation using quantlib

python/cvatesting/.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+    "python-envs.defaultEnvManager": "ms-python.python:conda",
+    "python-envs.defaultPackageManager": "ms-python.python:conda",
+    "python-envs.pythonProjects": []
+}

python/cvatesting/driver.py

@@ -0,0 +1,83 @@
+
+import numpy as np
+import Quantlib as ql
+
+from hullwhite import (build_calibrated_hw,simulate_hw_state,swap_npv_from_state)
+from instruments import (make_vanilla_swap,make_swaption_helpers)
+
+today = ql.Date(15, 1, 2025)
+ql.Settings.instance().evaluationDate = today
+calendar = ql.TARGET()
+day_count = ql.Actual365Fixed()
+
+
+def expected_exposure(
+    swap,
+    hw,
+    curve_handle,
+    today,
+    time_grid,
+    x_paths
+):
+    ee = np.zeros(len(time_grid))
+
+    for j, t in enumerate(time_grid):
+        values = []
+
+        for i in range(len(x_paths)):
+            v = swap_npv_from_state(
+                swap, hw, curve_handle, t, x_paths[i, j], today
+            )
+            values.append(max(v, 0.0))
+
+        ee[j] = np.mean(values)
+
+    return ee
+
+def main():
+    # Time grid
+    time_grid = np.linspace(0, 10, 121)
+
+    flat_rate = 0.02
+    curve = ql.FlatForward(today, flat_rate, day_count)
+    curve_handle = ql.YieldTermStructureHandle(curve)
+
+    #creat and calibrate HW
+    a_init = 0.05
+    sigma_init = 0.01
+    hw = ql.HullWhite(curve_handle, a_init, sigma_init)
+
+    index = ql.Euribor6M(curve_handle)
+    helpers=make_swaption_helpers(curve_handle,index,hw)
+    optimizer = ql.LevenbergMarquardt()
+    end_criteria = ql.EndCriteria(1000, 500, 1e-8, 1e-8, 1e-8)
+    hw.calibrate(helpers, optimizer, end_criteria)
+
+    a, sigma = hw.params()
+    print(f"Calibrated a     = {a:.4f}")
+    print(f"Calibrated sigma = {sigma:.4f}")
+
+
+
+    # Simulate model
+    x_paths = simulate_hw_state(
+        hw, curve_handle, time_grid, n_paths=10000
+    )
+
+    # Swap
+    swap = make_receiver_swap(
+        today, curve_handle, maturity_years=10, fixed_rate=0.025
+    )
+
+    # Exposure
+    ee = expected_exposure(
+        swap, hw, curve_handle, today, time_grid, x_paths
+    )
+
+
+
+    return 0
+
+
+if __name__ =='__main__':
+    main()

python/cvatesting/hullwhite.py

@@ -0,0 +1,66 @@
+import QuantLib as ql
+
+
+def build_calibrated_hw(
+    curve_handle,
+    swaption_helpers,
+    a_init=0.05,
+    sigma_init=0.01
+):
+    hw = ql.HullWhite(curve_handle, a_init, sigma_init)
+
+    optimizer = ql.LevenbergMarquardt()
+    end_criteria = ql.EndCriteria(1000, 500, 1e-8, 1e-8, 1e-8)
+
+    hw.calibrate(swaption_helpers, optimizer, end_criteria)
+
+    return hw
+
+import numpy as np
+
+
+def simulate_hw_state(
+    hw: ql.HullWhite,
+    curve_handle,
+    time_grid,
+    n_paths,
+    seed=42
+):
+    a, sigma = hw.params()
+
+    dt = np.diff(time_grid)
+    n_steps = len(time_grid)
+
+    np.random.seed(seed)
+    x = np.zeros((n_paths, n_steps))
+
+    for i in range(1, n_steps):
+        decay = np.exp(-a * dt[i-1])
+        vol = sigma * np.sqrt((1 - np.exp(-2 * a * dt[i-1])) / (2 * a))
+        z = np.random.normal(size=n_paths)
+
+        x[:, i] = x[:, i-1] * decay + vol * z
+
+    return x
+
+
+def swap_npv_from_state(
+    swap,
+    hw,
+    curve_handle,
+    t,
+    x_t,
+    today
+):
+    ql.Settings.instance().evaluationDate = today + int(t * 365)
+
+    hw.setState(x_t)
+
+    engine = ql.DiscountingSwapEngine(
+        curve_handle,
+        False,
+        hw
+    )
+
+    swap.setPricingEngine(engine)
+    return swap.NPV()

python/cvatesting/instruments.py

@@ -0,0 +1,165 @@
+import QuantLib as ql
+
+
+# ============================================================
+# Global defaults (override via parameters if needed)
+# ============================================================
+
+CALENDAR = ql.TARGET()
+BUSINESS_CONVENTION = ql.ModifiedFollowing
+DATE_GEN = ql.DateGeneration.Forward
+
+FIXED_DAYCOUNT = ql.Thirty360(ql.Thirty360.European)
+FLOAT_DAYCOUNT = ql.Actual360()
+
+
+# ============================================================
+# Yield curve factory
+# ============================================================
+
+def make_flat_curve(
+    evaluation_date: ql.Date,
+    rate: float,
+    day_count: ql.DayCounter = ql.Actual365Fixed()
+) -> ql.YieldTermStructureHandle:
+    """
+    Flat yield curve factory.
+    """
+    ql.Settings.instance().evaluationDate = evaluation_date
+    curve = ql.FlatForward(evaluation_date, rate, day_count)
+    return ql.YieldTermStructureHandle(curve)
+
+
+# ============================================================
+# Index factory
+# ============================================================
+
+def make_euribor_6m(
+    curve_handle: ql.YieldTermStructureHandle
+) -> ql.IborIndex:
+    """
+    Euribor 6M index factory.
+    """
+    return ql.Euribor6M(curve_handle)
+
+
+# ============================================================
+# Schedule factory
+# ============================================================
+
+def make_schedule(
+    start: ql.Date,
+    maturity: ql.Date,
+    tenor: ql.Period,
+    calendar: ql.Calendar = CALENDAR
+) -> ql.Schedule:
+    """
+    Generic schedule factory.
+    """
+    return ql.Schedule(
+        start,
+        maturity,
+        tenor,
+        calendar,
+        BUSINESS_CONVENTION,
+        BUSINESS_CONVENTION,
+        DATE_GEN,
+        False
+    )
+
+
+# ============================================================
+# Swap factory
+# ============================================================
+
+def make_vanilla_swap(
+    evaluation_date: ql.Date,
+    curve_handle: ql.YieldTermStructureHandle,
+    notional: float,
+    fixed_rate: float,
+    maturity_years: int,
+    pay_fixed: bool = False,
+    fixed_leg_freq: ql.Period = ql.Annual,
+    float_leg_freq: ql.Period = ql.Semiannual
+) -> ql.VanillaSwap:    
+    """
+    Vanilla fixed-for-float IRS factory.
+
+    pay_fixed = True  -> payer swap
+    pay_fixed = False -> receiver swap
+    """
+
+    ql.Settings.instance().evaluationDate = evaluation_date
+
+    index = make_euribor_6m(curve_handle)
+
+    start = CALENDAR.advance(evaluation_date, 2, ql.Days)
+    maturity = CALENDAR.advance(start, maturity_years, ql.Years)
+
+    fixed_schedule = make_schedule(
+        start, maturity, ql.Period(fixed_leg_freq)
+    )
+    float_schedule = make_schedule(
+        start, maturity, ql.Period(float_leg_freq)
+    )
+
+    swap_type = (
+        ql.VanillaSwap.Payer if pay_fixed else ql.VanillaSwap.Receiver
+    )
+
+    swap = ql.VanillaSwap(
+        swap_type,
+        notional,
+        fixed_schedule,
+        fixed_rate,
+        FIXED_DAYCOUNT,
+        float_schedule,
+        index,
+        0.0,
+        FLOAT_DAYCOUNT
+    )
+
+    swap.setPricingEngine(
+        ql.DiscountingSwapEngine(curve_handle)
+    )
+
+    return swap
+
+
+# ============================================================
+# Swaption helper factory (for HW calibration)
+# ============================================================
+
+def make_swaption_helpers(
+    swaption_data: list,
+    curve_handle: ql.YieldTermStructureHandle,
+    index: ql.IborIndex,
+    model: ql.HullWhite
+) -> list:
+    """
+    Create ATM swaption helpers.
+
+    swaption_data = [(expiry, tenor, vol), ...]
+    """
+
+    helpers = []
+
+    for expiry, tenor, vol in swaption_data:
+        helper = ql.SwaptionHelper(
+            expiry,
+            tenor,
+            ql.QuoteHandle(ql.SimpleQuote(vol)),
+            index,
+            index.tenor(),
+            index.dayCounter(),
+            index.dayCounter(),
+            curve_handle
+        )
+
+        helper.setPricingEngine(
+            ql.JamshidianSwaptionEngine(model)
+        )
+
+        helpers.append(helper)
+
+    return helpers

python/cvatesting/main.py

@@ -0,0 +1,109 @@
+import QuantLib as ql
+import numpy as np
+import matplotlib.pyplot as plt
+
+# --- UNIT 1: Instrument Setup ---
+def create_30y_swap(today, rate_quote):
+    ql.Settings.instance().evaluationDate = today
+    
+    # We use a Relinkable handle to swap curves per path/time-step
+    yield_handle = ql.RelinkableYieldTermStructureHandle()
+    yield_handle.linkTo(ql.FlatForward(today, rate_quote, ql.Actual365Fixed()))
+    
+    calendar = ql.TARGET()
+    settle_date = calendar.advance(today, 2, ql.Days)
+    maturity_date = calendar.advance(settle_date, 30, ql.Years)
+    
+    index = ql.Euribor6M(yield_handle)
+    
+    fixed_schedule = ql.Schedule(settle_date, maturity_date, ql.Period(ql.Annual), 
+                                 calendar, ql.ModifiedFollowing, ql.ModifiedFollowing, 
+                                 ql.DateGeneration.Forward, False)
+    float_schedule = ql.Schedule(settle_date, maturity_date, ql.Period(ql.Semiannual), 
+                                 calendar, ql.ModifiedFollowing, ql.ModifiedFollowing, 
+                                 ql.DateGeneration.Forward, False)
+    
+    swap = ql.VanillaSwap(ql.VanillaSwap.Payer, 1e6, fixed_schedule, rate_quote, 
+                          ql.Thirty360(ql.Thirty360.BondBasis), float_schedule, 
+                          index, 0.0, ql.Actual360())
+    
+    # Pre-calculate all fixing dates needed for the life of the swap
+    fixing_dates = [index.fixingDate(d) for d in float_schedule]
+    
+    return swap, yield_handle, index, fixing_dates
+
+# --- UNIT 2: The Simulation Loop ---
+def run_simulation(n_paths=50):
+    today = ql.Date(27, 1, 2026)
+    swap, yield_handle, index, fixing_dates = create_30y_swap(today, 0.03)
+    
+    # HW Parameters: a=mean reversion, sigma=vol
+    model = ql.HullWhite(yield_handle, 0.01, 0.01)
+    process = ql.HullWhiteProcess(yield_handle, 0.01, 0.01)
+    
+    times = np.arange(0, 31, 1.0) # Annual buckets
+    grid = ql.TimeGrid(times)
+    
+    rng = ql.GaussianRandomSequenceGenerator(ql.UniformRandomSequenceGenerator(
+        len(grid)-1, ql.UniformRandomGenerator()))
+    seq = ql.GaussianMultiPathGenerator(process, grid, rng, False)
+    
+    npv_matrix = np.zeros((n_paths, len(times)))
+
+    for i in range(n_paths):
+        path = seq.next().value()[0]
+        # 1. Clear previous path's fixings to avoid data pollution
+        ql.IndexManager.instance().clearHistories()
+        
+        for j, t in enumerate(times):
+            if t >= 30: continue
+            
+            eval_date = ql.TARGET().advance(today, int(t), ql.Years)
+            ql.Settings.instance().evaluationDate = eval_date
+            
+            # 2. Update the curve with simulated short rate rt
+            rt = path[j]
+            # Use pillars up to 35Y to avoid extrapolation crashes
+            tenors = [0, 1, 2, 5, 10, 20, 30, 35]
+            dates = [ql.TARGET().advance(eval_date, y, ql.Years) for y in tenors]
+            discounts = [model.discountBond(t, t + y, rt) for y in tenors]
+            
+            sim_curve = ql.DiscountCurve(dates, discounts, ql.Actual365Fixed())
+            sim_curve.enableExtrapolation()
+            yield_handle.linkTo(sim_curve)
+            
+            # 3. MANUAL FIXING INJECTION
+            # For every fixing date that has passed or is 'today'
+            for fd in fixing_dates:
+                if fd <= eval_date:
+                    # Direct manual calculation to avoid index.fixing() error
+                    # We calculate the forward rate for the 6M period starting at fd
+                    start_date = fd
+                    end_date = ql.TARGET().advance(start_date, 6, ql.Months)
+                    
+                    # Manual Euribor rate formula: (P1/P2 - 1) / dt
+                    p1 = sim_curve.discount(start_date)
+                    p2 = sim_curve.discount(end_date)
+                    dt = ql.Actual360().yearFraction(start_date, end_date)
+                    fwd_rate = (p1 / p2 - 1.0) / dt
+                    
+                    index.addFixing(fd, fwd_rate, True) # Force overwrite
+            
+            # 4. Valuation
+            swap.setPricingEngine(ql.DiscountingSwapEngine(yield_handle))
+            npv_matrix[i, j] = max(swap.NPV(), 0)
+            
+        ql.Settings.instance().evaluationDate = today
+
+    return times, npv_matrix
+
+# --- UNIT 3: Visualization ---
+times, npv_matrix = run_simulation(n_paths=100)
+ee = np.mean(npv_matrix, axis=0)
+
+plt.figure(figsize=(10, 5))
+plt.plot(times, ee, lw=2, label="Expected Exposure (EE)")
+plt.fill_between(times, ee, alpha=0.3)
+plt.title("30Y Swap EE Profile - Hull White")
+plt.xlabel("Years"); plt.ylabel("Exposure"); plt.grid(True); plt.legend()
+plt.show()