from __future__ import annotations
from dataclasses import dataclass, field
from typing import Literal, Tuple, List
import yaml
from cann.engine.calibration_config import CalibrationConfig
from cann.engine.training_config import TrainingConfig
ScaleType = Literal["linear", "log"]
[docs]
@dataclass
class ParameterSpec:
"""
Specification for a single model parameter (e.g., `rho`, `kappa`).
Fields:
``name``, ``min_value``, ``max_value``, optional ``default``, optional ``description``,
and ``scale`` ("linear" or "log").
"""
name: str
min_value: float
max_value: float
default: float | None = None
scale: ScaleType = "linear"
description: str | None = None
[docs]
def validate_value(self, value: float) -> None:
"""
Validate that the given value is within the specified bounds.
Args:
value (float): The value to validate.
Raises:
ValueError: If the value is out of bounds [min_value, max_value].
"""
if not (self.min_value <= value <= self.max_value):
raise ValueError(
f"Value {value} for parameter '{self.name}' is out of bounds [{self.min_value}, {self.max_value}]"
)
[docs]
@dataclass
class FeatureSpec:
"""
Describes a single market input feature (e.g. `moneyness`, `time_to_maturity`).
Fields:
``name``, optional ``description``, and optional ``domain`` (min, max).
"""
name: str
description: str | None = None
domain: Tuple[float, float] | None = None
[docs]
@dataclass
class OutputSpec:
"""
Describes a single model output (e.g., `implied_volatility`, `option_price`).
Fields:
``name``, optional ``description``, and optional ``domain`` (min, max).
"""
name: str
description: str | None = None
domain: Tuple[float, float] | None = None
[docs]
@dataclass
class ModelConfig:
"""Configuration for a pricing model family.
This object describes the parameter vector theta and
the market input feature vector x used by the surrogate.
Key fields include ``parameters`` (with bounds), ``market_inputs`` (feature layout),
``outputs``, and optional default configs (``train_defaults`` and
``calibration_defaults``).
"""
name: str
parameters: List[ParameterSpec] = field(default_factory=list)
market_inputs: List[FeatureSpec] = field(default_factory=list)
outputs: List[OutputSpec] = field(default_factory=list)
train_defaults: TrainingConfig | None = None
calibration_defaults: CalibrationConfig | None = None
# --- Convenience helpers ---
@property
def num_parameters(self) -> int:
"""Number of model parameters."""
return len(self.parameters)
@property
def num_market_inputs(self) -> int:
"""Number of market input features."""
return len(self.market_inputs)
@property
def num_outputs(self) -> int:
"""Number of model outputs."""
return len(self.outputs)
[docs]
def bounds_array(self) -> List[Tuple[float, float]]:
"""Get list of (min, max) bounds for all parameters in order."""
return [(p.min_value, p.max_value) for p in self.parameters]
[docs]
@classmethod
def from_dict(cls, cfg: dict) -> ModelConfig:
"""Create ModelConfig from a plain dictionary.
The dictionary format is expected to match the ``*.yaml`` model spec files
bundled in the project artifacts.
"""
params = [
ParameterSpec(
name=p["name"],
min_value=p["min"],
max_value=p["max"],
default=p.get("default"),
scale=p.get("scale", "linear"),
description=p.get("description"),
)
for p in cfg.get("parameters", [])
]
market_inputs = [
FeatureSpec(
name=f["name"],
description=f.get("description"),
domain=tuple(f["domain"]) if "domain" in f else None,
)
for f in cfg.get("market_inputs", [])
]
outputs = [
OutputSpec(
name=o["name"],
description=o.get("description"),
domain=tuple(o["domain"]) if "domain" in o else None,
)
for o in cfg.get("outputs", [])
]
train_defaults = (
TrainingConfig.from_dict(cfg["train_defaults"])
if "train_defaults" in cfg
else None
)
calibration_defaults = (
CalibrationConfig.from_dict(cfg["calibration_defaults"])
if "calibration_defaults" in cfg
else None
)
return cls(
name=cfg["name"],
parameters=params,
market_inputs=market_inputs,
outputs=outputs,
train_defaults=train_defaults,
calibration_defaults=calibration_defaults,
)
[docs]
@classmethod
def from_yaml(cls, path: str) -> ModelConfig:
"""Load ModelConfig from a YAML file."""
with open(path, "r", encoding="utf-8") as f:
cfg = yaml.safe_load(f)
return cls.from_dict(cfg)
def __str__(self) -> str:
param_strs = [
f" - {p.name}: [{p.min_value}, {p.max_value}] (default={p.default}, scale={p.scale})"
for p in self.parameters
]
feature_strs = [f" - {f.name}: domain={f.domain}" for f in self.market_inputs]
output_strs = [f" - {o.name}: domain={o.domain}" for o in self.outputs]
return (
f"ModelConfig(name={self.name},\n"
f" Parameters:\n" + "\n".join(param_strs) + "\n"
f" Market Inputs:\n" + "\n".join(feature_strs) + "\n"
f" Outputs:\n" + "\n".join(output_strs) + "\n)"
)
[docs]
def describe(self) -> None:
"""Print a human-readable description of the model configuration. Which includes descriptions."""
print(f"Model: {self.name}")
print("Parameters:")
for p in self.parameters:
desc = f" - {p.name} (scale={p.scale}): [{p.min_value}, {p.max_value}], default={p.default}"
if p.description:
desc += f"\n Description: {p.description}"
print(desc)
print("Market Inputs:")
for f in self.market_inputs:
desc = f" - {f.name}: domain={f.domain}"
if f.description:
desc += f"\n Description: {f.description}"
print(desc)
print("Outputs:")
for o in self.outputs:
desc = f" - {o.name}: domain={o.domain}"
if o.description:
desc += f"\n Description: {o.description}"
print(desc)