Examples

Backward pass (online calibration)

Calibrate from a MATLAB .mat market surface

This mirrors the workflow in the repository notebook but keeps the example minimal. The example assumes your MAT file contains (at least) arrays T, K, S, r, and Vp.

import numpy as np
from scipy.io import loadmat

from cann.data.market_surface import MarketSurface
from cann.engine.calibration import calibrate_model
from cann.engine.calibration_config import CalibrationConfig
from cann.pretrained import PretrainedRoughHeston


trained_model = PretrainedRoughHeston()
model_config = trained_model.model_config

matfile = "rheston_market6par_v1_1.mat"
d = loadmat(matfile)

# Typical normalization used in this repo's scripts/notebooks.
KoS = d["K"].flatten() / (d["S"].flatten() + 1e-12)
target = d["Vp"].flatten() / (d["S"].flatten() + 1e-12)

# Optional: emphasize near-ATM points.
weight_atm = (
    model_config.calibration_defaults.weight_atm
    if model_config.calibration_defaults is not None
    else 1.0
)
weights = np.array([weight_atm if np.isclose(m, 1.0, atol=0.1) else 1.0 for m in KoS])

market_surface = MarketSurface(
    market_inputs=np.column_stack((d["T"].flatten(), KoS, d["r"].flatten())),
    observed=target,
    weights=weights,
    metadata={"source": matfile},
)

calib_cfg = model_config.calibration_defaults or CalibrationConfig()
result = calibrate_model(trained_model=trained_model, market_surface=market_surface, calibration_config=calib_cfg)

print("converged:", result.converged)
print("final_loss:", result.final_loss)
print("optimal_parameters:", result.optimal_parameters)

Validate fit with a prediction scatter plot

import matplotlib.pyplot as plt

calibrated = np.array(result.optimal_parameters).reshape(1, -1)
n = market_surface.market_inputs.shape[0]

preds = trained_model.surrogate.predict(
    parameters=np.tile(calibrated, (n, 1)),
    market_inputs=market_surface.market_inputs,
    batch_size=2048,
    device="cpu",
).squeeze()

y = np.asarray(market_surface.observed).squeeze()

plt.figure(figsize=(6, 6))
plt.plot(y, preds, "*", alpha=0.8)
plt.xlabel("Market")
plt.ylabel("Model")
plt.grid(True)
plt.show()

r2 = 1.0 - np.sum((y - preds) ** 2) / np.sum((y - np.mean(y)) ** 2)
print(f"R2: {r2:.6f}")

Forward pass (offline learning)

Train a surrogate from in-memory arrays

import numpy as np

from cann.config.model_config import ModelConfig
from cann.engine.training import train_surrogate
from cann.engine.training_config import TrainingConfig


 model_config = ModelConfig.from_yaml("cann/artifacts/heston_iv_v1/heston_iv.yaml")

parameters = np.random.randn(1000, model_config.num_parameters).astype(np.float32)
market_inputs = np.random.randn(1000, model_config.num_market_inputs).astype(np.float32)
observed = np.random.randn(1000).astype(np.float32)

training_cfg = model_config.train_defaults or TrainingConfig(epochs=200)

trained_model, test_dataset = train_surrogate(
    model_config=model_config,
    parameters=parameters,
    market_inputs=market_inputs,
    observed=observed,
    training_config=training_cfg,
    verbose=True,
)