Skip to content

Adapters

Concrete implementations of the ports. Optional dependencies are imported lazily inside the methods that need them.

Inbound

Facade

wayfault.adapters.inbound.api

Thin Python facade over the WrongWayRiskService.

estimate_wwr 

estimate_wwr(exposure: ExposureSource, credit: CreditCurveSource, model: DependenceModel, discount: ndarray | None = None, pfe_quantile: float = 0.95, sink: ResultSink | None = None) -> WWRResult

Estimate Wrong-Way Risk for one counterparty.

Parameters:

Name Type Description Default
exposure ExposureSource

The outbound-port collaborators.

 required
credit ExposureSource

The outbound-port collaborators.

 required
model ExposureSource

The outbound-port collaborators.

 required
discount ndarray | None

Optional discount factors on the grid.

 None
pfe_quantile float

Quantile for the PFE profile.

 0.95
sink ResultSink | None

Optional result sink; if given, the result is also written to it.

 None

Returns:

Type Description
WWRResult

The full result object.
Source code in src/wayfault/adapters/inbound/api.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def estimate_wwr(
    exposure: ExposureSource,
    credit: CreditCurveSource,
    model: DependenceModel,
    discount: np.ndarray | None = None,
    pfe_quantile: float = 0.95,
    sink: ResultSink | None = None,
) -> WWRResult:
    """Estimate Wrong-Way Risk for one counterparty.

    Parameters
    ----------
    exposure, credit, model:
        The outbound-port collaborators.
    discount:
        Optional discount factors on the grid.
    pfe_quantile:
        Quantile for the PFE profile.
    sink:
        Optional result sink; if given, the result is also written to it.

    Returns
    -------
    WWRResult
        The full result object.
    """
    request = WWRRequest(
        exposure=exposure,
        credit=credit,
        model=model,
        discount=discount,
        pfe_quantile=pfe_quantile,
    )
    result = WrongWayRiskService().estimate(request)
    if sink is not None:
        sink.write(result)
    return result

CLI

wayfault.adapters.inbound.cli

argparse CLI: python -m wayfault (stdlib only).

main 

main(argv: Sequence[str] | None = None) -> int

CLI entry point. Returns a process exit code.

Source code in src/wayfault/adapters/inbound/cli.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def main(argv: Sequence[str] | None = None) -> int:
    """CLI entry point. Returns a process exit code."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    if args.command == "estimate":
        model = _build_model(args.model, args.b, args.rho, args.theta)
        sink = JsonReportWriter(args.out) if args.out else None
        result = estimate_wwr(
            exposure=CsvExposureSource(args.exposure),
            credit=CsvCreditCurveSource(args.credit, recovery=args.recovery),
            model=model,
            pfe_quantile=args.pfe_quantile,
            sink=sink,
        )
        json.dump(result.to_dict(), sys.stdout, indent=2)
        sys.stdout.write("\n")
        return 0
    return 1

Outbound — Exposure sources

wayfault.adapters.outbound.exposure_inmemory

In-memory exposure source (no extras).

InMemoryExposureSource 

InMemoryExposureSource(values: ndarray, tenors: list[float] | ndarray)

Wraps an in-memory array (or EE profile) as an :class:ExposureSource.

Source code in src/wayfault/adapters/outbound/exposure_inmemory.py 
14
15
16
def __init__(self, values: np.ndarray, tenors: list[float] | np.ndarray) -> None:
    self._grid = TenorGrid(np.asarray(tenors, dtype=float))
    self._cube = ExposureCube(self._grid, np.asarray(values, dtype=float))

from_ee_profile classmethod 

from_ee_profile(profile: EEProfile) -> InMemoryExposureSource

Build a degenerate source from a precomputed EE profile.

Source code in src/wayfault/adapters/outbound/exposure_inmemory.py 
18
19
20
21
22
@classmethod
def from_ee_profile(cls, profile: EEProfile) -> InMemoryExposureSource:
    """Build a degenerate source from a precomputed EE profile."""
    cube = ExposureCube.from_ee_profile(profile)
    return cls(cube.values, profile.grid.times)

load 

load() -> ExposureCube

Return the wrapped exposure cube.

Source code in src/wayfault/adapters/outbound/exposure_inmemory.py 
24
25
26
def load(self) -> ExposureCube:
    """Return the wrapped exposure cube."""
    return self._cube

wayfault.adapters.outbound.exposure_csv

CSV/Parquet exposure source ([io] extra, lazy import).

CsvExposureSource 

CsvExposureSource(path: str, fmt: str = 'csv')

Loads an exposure cube from CSV or Parquet via pandas/pyarrow.

The file's column headers are interpreted as tenor year-fractions and each row as one Monte-Carlo scenario.

Source code in src/wayfault/adapters/outbound/exposure_csv.py 
19
20
21
def __init__(self, path: str, fmt: str = "csv") -> None:
    self._path = path
    self._fmt = fmt

load 

load() -> ExposureCube

Read the file lazily and build an :class:ExposureCube.

Source code in src/wayfault/adapters/outbound/exposure_csv.py 
23
24
25
26
27
28
29
30
31
32
33
def load(self) -> ExposureCube:
    """Read the file lazily and build an :class:`ExposureCube`."""
    pd = require("pandas", "io")
    if self._fmt == "parquet":
        require("pyarrow", "io")
        frame = pd.read_parquet(self._path)
    else:
        frame = pd.read_csv(self._path)
    tenors = np.asarray([float(c) for c in frame.columns], dtype=float)
    grid = TenorGrid(tenors)
    return ExposureCube(grid, frame.to_numpy(dtype=float))

Outbound — Credit sources

wayfault.adapters.outbound.credit_flat

Flat-hazard credit-curve source (no extras).

FlatHazardCreditCurveSource 

FlatHazardCreditCurveSource(hazard: float, recovery: float = 0.4)

Supplies a flat-hazard :class:CreditCurve.

Source code in src/wayfault/adapters/outbound/credit_flat.py 
11
12
def __init__(self, hazard: float, recovery: float = 0.4) -> None:
    self._curve = CreditCurve.flat(hazard=hazard, recovery=recovery)

load 

load() -> CreditCurve

Return the flat credit curve.

Source code in src/wayfault/adapters/outbound/credit_flat.py 
14
15
16
def load(self) -> CreditCurve:
    """Return the flat credit curve."""
    return self._curve

wayfault.adapters.outbound.credit_piecewise

Piecewise-constant-hazard credit-curve source (no extras).

PiecewiseHazardCreditCurveSource 

PiecewiseHazardCreditCurveSource(knots: list[float], hazards: list[float], recovery: float = 0.4)

Supplies a piecewise-constant-hazard :class:CreditCurve.

Source code in src/wayfault/adapters/outbound/credit_piecewise.py 
11
12
13
14
def __init__(
    self, knots: list[float], hazards: list[float], recovery: float = 0.4
) -> None:
    self._curve = CreditCurve.piecewise(knots=knots, hazards=hazards, recovery=recovery)

load 

load() -> CreditCurve

Return the piecewise credit curve.

Source code in src/wayfault/adapters/outbound/credit_piecewise.py 
16
17
18
def load(self) -> CreditCurve:
    """Return the piecewise credit curve."""
    return self._curve

wayfault.adapters.outbound.credit_csv

CSV credit-curve source ([io] extra, lazy import).

Expects a CSV with columns knot and hazard (one row per segment). The recovery rate is supplied at construction.

CsvCreditCurveSource 

CsvCreditCurveSource(path: str, recovery: float = 0.4)

Loads a piecewise-constant-hazard curve from CSV via pandas.

Source code in src/wayfault/adapters/outbound/credit_csv.py 
16
17
18
def __init__(self, path: str, recovery: float = 0.4) -> None:
    self._path = path
    self._recovery = recovery

load 

load() -> CreditCurve

Read the curve lazily and build a :class:CreditCurve.

Source code in src/wayfault/adapters/outbound/credit_csv.py 
20
21
22
23
24
25
26
def load(self) -> CreditCurve:
    """Read the curve lazily and build a :class:`CreditCurve`."""
    pd = require("pandas", "io")
    frame = pd.read_csv(self._path)
    knots = [float(x) for x in frame["knot"]]
    hazards = [float(x) for x in frame["hazard"]]
    return CreditCurve.piecewise(knots=knots, hazards=hazards, recovery=self._recovery)

Outbound — Dependence models

wayfault.adapters.outbound.dependence_independent

Independent dependence model (FR-WWR-021).

IndependentModel

Conditional EE equals unconditional EE (no dependence).

calibrate_to_curve 

calibrate_to_curve(curve: CreditCurve, grid: TenorGrid) -> None

No-op: the independent model has nothing to calibrate.

Source code in src/wayfault/adapters/outbound/dependence_independent.py 
15
16
def calibrate_to_curve(self, curve: CreditCurve, grid: TenorGrid) -> None:
    """No-op: the independent model has nothing to calibrate."""

conditional_ee 

conditional_ee(cube: ExposureCube, curve: CreditCurve, grid: TenorGrid) -> EEProfile

Return the unconditional EPE profile unchanged.

Source code in src/wayfault/adapters/outbound/dependence_independent.py 
18
19
20
21
22
23
def conditional_ee(
    self, cube: ExposureCube, curve: CreditCurve, grid: TenorGrid
) -> EEProfile:
    """Return the unconditional EPE profile unchanged."""
    values = np.maximum(cube.values, 0.0).mean(axis=0)
    return EEProfile(grid, values)

dependence_param 

dependence_param() -> float

The independent model has a zero dependence parameter.

Source code in src/wayfault/adapters/outbound/dependence_independent.py 
25
26
27
def dependence_param(self) -> float:
    """The independent model has a zero dependence parameter."""
    return 0.0

params 

params() -> dict[str, float]

Model metadata.

Source code in src/wayfault/adapters/outbound/dependence_independent.py 
29
30
31
def params(self) -> dict[str, float]:
    """Model metadata."""
    return {}

wayfault.adapters.outbound.dependence_hullwhite

Hull-White stochastic-hazard dependence model (FR-WWR-022).

Canonical WWR formulation following Hull & White (2012):

.. math::

\lambda(t) = \exp(a(t) + b\, V(t))

The intensity offset a(t) is solved per tenor so that, integrated over the exposure distribution, the model reproduces the input curve's marginal PDs (arbitrage consistency). b is the wrong-way-risk knob: b > 0 is wrong-way, b < 0 right-way, b = 0 independence.

HullWhiteHazardModel 

HullWhiteHazardModel(b: float = 0.0)

Stochastic-hazard WWR model with intensity exp(a(t) + b*V(t)).

Parameters:

Name Type Description Default
b float

The wrong-way-risk coupling. b > 0 increases default likelihood in high-exposure scenarios (wrong-way); b < 0 is right-way.

 0.0
Source code in src/wayfault/adapters/outbound/dependence_hullwhite.py 
35
36
37
def __init__(self, b: float = 0.0) -> None:
    self.b = float(b)
    self._a: np.ndarray | None = None

calibrate_to_curve 

calibrate_to_curve(curve: CreditCurve, grid: TenorGrid) -> None

Record the target marginal PDs; a(t) is solved at re-weighting.

The per-tenor offset cancels in the normalised conditional expectation, so calibration only needs the target marginal PDs, which are recovered from the curve on demand.

Source code in src/wayfault/adapters/outbound/dependence_hullwhite.py 
39
40
41
42
43
44
45
46
47
48
def calibrate_to_curve(self, curve: CreditCurve, grid: TenorGrid) -> None:
    """Record the target marginal PDs; ``a(t)`` is solved at re-weighting.

    The per-tenor offset cancels in the normalised conditional expectation,
    so calibration only needs the target marginal PDs, which are recovered
    from the curve on demand.
    """
    # Stored implicitly via the curve at conditional_ee time; nothing to do
    # eagerly besides validating that a curve/grid were provided.
    _ = curve.marginal_pd(grid)

conditional_ee 

conditional_ee(cube: ExposureCube, curve: CreditCurve, grid: TenorGrid) -> EEProfile

Conditional EE by re-weighting scenarios toward their default risk.

Fully vectorised: every tenor column is re-weighted in a single numpy expression via the shared column-wise softmax core.

Source code in src/wayfault/adapters/outbound/dependence_hullwhite.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def conditional_ee(
    self, cube: ExposureCube, curve: CreditCurve, grid: TenorGrid
) -> EEProfile:
    """Conditional EE by re-weighting scenarios toward their default risk.

    Fully vectorised: every tenor column is re-weighted in a single numpy
    expression via the shared column-wise softmax core.
    """
    v = cube.values
    weights = _reweight.softmax_columns(self.b * v)
    out = _reweight.conditional_ee_columns(v, weights)

    # Offsets a(t) implied by reproducing the marginal PD in expectation
    # (kept for diagnostics; they cancel in the normalised expectation).
    pd_target = curve.marginal_pd(grid)
    shifted = self.b * (v - v.max(axis=0, keepdims=True))
    mean_exp = np.mean(np.exp(shifted), axis=0)
    self._a = np.log(np.maximum(pd_target, 1e-300)) - (self.b * v).max(axis=0) - np.log(
        mean_exp
    )
    return EEProfile(grid, out)

implied_marginal_pd 

implied_marginal_pd(cube: ExposureCube, curve: CreditCurve, grid: TenorGrid) -> np.ndarray

Model-implied marginal PDs integrated over the exposure distribution.

By construction this reproduces curve.marginal_pd(grid) (criterion 4): the per-scenario conditional PD is the target PD scaled by the normalised exposure weight, whose scenario mean is the target.

Source code in src/wayfault/adapters/outbound/dependence_hullwhite.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def implied_marginal_pd(
    self, cube: ExposureCube, curve: CreditCurve, grid: TenorGrid
) -> np.ndarray:
    """Model-implied marginal PDs integrated over the exposure distribution.

    By construction this reproduces ``curve.marginal_pd(grid)`` (criterion
    4): the per-scenario conditional PD is the target PD scaled by the
    normalised exposure weight, whose scenario mean is the target.
    """
    pd_target = curve.marginal_pd(grid)
    w = _reweight.softmax_columns(self.b * cube.values)
    # q_i(s) = pd_target_i * w_{s,i} * S has column mean exactly pd_target_i.
    implied: np.ndarray = np.mean(pd_target[None, :] * w * cube.n_scenarios, axis=0)
    return implied

dependence_param 

dependence_param() -> float

Return the WWR coupling b.

Source code in src/wayfault/adapters/outbound/dependence_hullwhite.py 
87
88
89
def dependence_param(self) -> float:
    """Return the WWR coupling ``b``."""
    return self.b

params 

params() -> dict[str, float]

Model metadata.

Source code in src/wayfault/adapters/outbound/dependence_hullwhite.py 
91
92
93
def params(self) -> dict[str, float]:
    """Model metadata."""
    return {"b": self.b}

wayfault.adapters.outbound.dependence_copula

Gaussian-copula dependence model (FR-WWR-023).

Couples a credit latent factor to a portfolio/market factor with correlation rho via a one-factor Gaussian copula. The conditional default probability given the (rank-normalised) portfolio value re-weights scenarios to produce the conditional EE. rho > 0 is wrong-way, rho < 0 right-way.

Fully vectorised across tenors via the shared column-wise re-weighting core.

GaussianCopulaModel 

GaussianCopulaModel(rho: float = 0.0)

One-factor Gaussian-copula WWR model with correlation rho.

Source code in src/wayfault/adapters/outbound/dependence_copula.py 
24
25
26
27
def __init__(self, rho: float = 0.0) -> None:
    if not (-1.0 < rho < 1.0):
        raise ValueError("rho must be in (-1, 1).")
    self.rho = float(rho)

calibrate_to_curve 

calibrate_to_curve(curve: CreditCurve, grid: TenorGrid) -> None

No eager state: marginal PDs are read from the curve on demand.

Source code in src/wayfault/adapters/outbound/dependence_copula.py 
29
30
31
def calibrate_to_curve(self, curve: CreditCurve, grid: TenorGrid) -> None:
    """No eager state: marginal PDs are read from the curve on demand."""
    _ = curve.marginal_pd(grid)

conditional_ee 

conditional_ee(cube: ExposureCube, curve: CreditCurve, grid: TenorGrid) -> EEProfile

Conditional EE via Gaussian-copula default re-weighting (vectorised).

Source code in src/wayfault/adapters/outbound/dependence_copula.py 
33
34
35
36
37
38
39
40
41
42
43
def conditional_ee(
    self, cube: ExposureCube, curve: CreditCurve, grid: TenorGrid
) -> EEProfile:
    """Conditional EE via Gaussian-copula default re-weighting (vectorised)."""
    pd = curve.marginal_pd(grid)
    denom = math.sqrt(max(1.0 - self.rho * self.rho, 1e-12))
    y = _normals.norm_ppf(_reweight.rank_uniform_columns(cube.values))
    thresh = _normals.norm_ppf(pd)[None, :]
    cond_pd = _normals.norm_cdf((thresh + self.rho * y) / denom)
    out = _reweight.conditional_ee_columns(cube.values, cond_pd)
    return EEProfile(grid, out)

dependence_param 

dependence_param() -> float

Return the copula correlation rho.

Source code in src/wayfault/adapters/outbound/dependence_copula.py 
45
46
47
def dependence_param(self) -> float:
    """Return the copula correlation ``rho``."""
    return self.rho

params 

params() -> dict[str, float]

Model metadata.

Source code in src/wayfault/adapters/outbound/dependence_copula.py 
49
50
51
def params(self) -> dict[str, float]:
    """Model metadata."""
    return {"rho": self.rho}

wayfault.adapters.outbound.dependence_archimedean

Advanced Archimedean-copula dependence models (numpy-only).

These couple a credit margin to the portfolio/market margin through an Archimedean copula, using the copula's closed-form h-function (conditional CDF) as the per-scenario default re-weight. Unlike the Gaussian copula they capture asymmetric tail dependence, which is the realistic shape of wrong-way risk: defaults cluster precisely in the high-exposure tail.

  • :class:ClaytonCopulaModel — lower-tail dependence; theta > 0 is wrong-way (and theta -> 0 recovers independence). Pure powers, no special functions.
  • :class:FrankCopulaModel — symmetric, no tail dependence; the sign of theta flips the direction (theta > 0 wrong-way, theta < 0 right-way). Uses only exp.

Both are fully vectorised across tenors via the shared re-weighting core. The market margin is oriented so that high exposure maps to the lower tail (v = 1 - empirical_grade(V)), making a positive theta wrong-way.

ClaytonCopulaModel 

ClaytonCopulaModel(theta: float = 1.0)

Clayton-copula WWR model with lower-tail dependence.

Parameters:

Name Type Description Default
theta float

Dependence strength theta > 0. Larger theta means stronger clustering of defaults with high exposure (more wrong-way). As theta -> 0 the model converges to independence.

 1.0
Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
49
50
51
52
def __init__(self, theta: float = 1.0) -> None:
    if theta <= 0.0:
        raise ValueError("Clayton theta must be > 0.")
    self.theta = float(theta)

calibrate_to_curve 

calibrate_to_curve(curve: CreditCurve, grid: TenorGrid) -> None

No eager state; marginal PDs are read on demand.

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
54
55
56
def calibrate_to_curve(self, curve: CreditCurve, grid: TenorGrid) -> None:
    """No eager state; marginal PDs are read on demand."""
    _ = curve.marginal_pd(grid)

conditional_ee 

conditional_ee(cube: ExposureCube, curve: CreditCurve, grid: TenorGrid) -> EEProfile

Conditional EE via the Clayton h-function re-weighting (vectorised).

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
58
59
60
61
62
63
64
65
66
67
68
69
def conditional_ee(
    self, cube: ExposureCube, curve: CreditCurve, grid: TenorGrid
) -> EEProfile:
    """Conditional EE via the Clayton h-function re-weighting (vectorised)."""
    th = self.theta
    u = np.clip(curve.marginal_pd(grid), _EPS, 1.0 - _EPS)[None, :]
    v = _market_grade(cube.values)
    # h(u|v) = v^{-(theta+1)} (u^{-theta} + v^{-theta} - 1)^{-(theta+1)/theta}
    inner = np.maximum(u ** (-th) + v ** (-th) - 1.0, _EPS)
    weights = v ** (-(th + 1.0)) * inner ** (-(th + 1.0) / th)
    out = _reweight.conditional_ee_columns(cube.values, weights)
    return EEProfile(grid, out)

dependence_param 

dependence_param() -> float

Positive parameter ⇒ wrong-way; magnitude is the Clayton theta.

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
71
72
73
def dependence_param(self) -> float:
    """Positive parameter ⇒ wrong-way; magnitude is the Clayton ``theta``."""
    return self.theta

params 

params() -> dict[str, float]

Model metadata.

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
75
76
77
def params(self) -> dict[str, float]:
    """Model metadata."""
    return {"theta": self.theta}

FrankCopulaModel 

FrankCopulaModel(theta: float = 2.0)

Frank-copula WWR model (symmetric, sign-controlled direction).

Parameters:

Name Type Description Default
theta float

Non-zero dependence parameter. theta > 0 is wrong-way, theta < 0 right-way; theta -> 0 is independence.

 2.0
Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
90
91
92
93
def __init__(self, theta: float = 2.0) -> None:
    if abs(theta) < _EPS:
        raise ValueError("Frank theta must be non-zero (use IndependentModel for 0).")
    self.theta = float(theta)

calibrate_to_curve 

calibrate_to_curve(curve: CreditCurve, grid: TenorGrid) -> None

No eager state; marginal PDs are read on demand.

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
95
96
97
def calibrate_to_curve(self, curve: CreditCurve, grid: TenorGrid) -> None:
    """No eager state; marginal PDs are read on demand."""
    _ = curve.marginal_pd(grid)

conditional_ee 

conditional_ee(cube: ExposureCube, curve: CreditCurve, grid: TenorGrid) -> EEProfile

Conditional EE via the Frank h-function re-weighting (vectorised).

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def conditional_ee(
    self, cube: ExposureCube, curve: CreditCurve, grid: TenorGrid
) -> EEProfile:
    """Conditional EE via the Frank h-function re-weighting (vectorised)."""
    th = self.theta
    u = np.clip(curve.marginal_pd(grid), _EPS, 1.0 - _EPS)[None, :]
    v = _market_grade(cube.values)
    eu = np.expm1(-th * u)  # e^{-theta u} - 1
    ev = np.expm1(-th * v)
    e1 = np.expm1(-th)  # e^{-theta} - 1
    # h(u|v) = e^{-th v}(e^{-th u}-1) / [(e^{-th}-1) + (e^{-th u}-1)(e^{-th v}-1)]
    num = np.exp(-th * v) * eu
    den = e1 + eu * ev
    weights = np.abs(num / np.where(np.abs(den) < _EPS, _EPS, den))
    out = _reweight.conditional_ee_columns(cube.values, weights)
    return EEProfile(grid, out)

dependence_param 

dependence_param() -> float

Signed parameter: > 0 wrong-way, < 0 right-way.

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
116
117
118
def dependence_param(self) -> float:
    """Signed parameter: ``> 0`` wrong-way, ``< 0`` right-way."""
    return self.theta

params 

params() -> dict[str, float]

Model metadata.

Source code in src/wayfault/adapters/outbound/dependence_archimedean.py 
120
121
122
def params(self) -> dict[str, float]:
    """Model metadata."""
    return {"theta": self.theta}

Outbound — Calibrators

wayfault.adapters.outbound.calibrator_regression

Numpy-only regression calibrator (FR-WWR-041).

RegressionCalibrator

Estimate the Hull-White b by least-squares regression.

The Hull-White intensity is lambda = exp(a + b*V), so log lambda is linear in the portfolio value V with slope b. The credit_factor samples are interpreted as realised hazard rates; their logs are regressed on the portfolio value to recover b (and the intercept a).

fit 

fit(portfolio_value: ndarray, credit_factor: ndarray) -> dict[str, float]

Return {"a": intercept, "b": slope} from an OLS fit.

Source code in src/wayfault/adapters/outbound/calibrator_regression.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def fit(
    self, portfolio_value: np.ndarray, credit_factor: np.ndarray
) -> dict[str, float]:
    """Return ``{"a": intercept, "b": slope}`` from an OLS fit."""
    v = np.asarray(portfolio_value, dtype=float).ravel()
    c = np.asarray(credit_factor, dtype=float).ravel()
    if v.size != c.size:
        raise ValidationError("portfolio_value and credit_factor must be equal length.")
    if v.size < 2:
        raise ValidationError("Need at least two samples to calibrate.")
    if np.any(c <= 0.0):
        raise ValidationError("credit_factor (hazard) samples must be positive.")
    log_hazard = np.log(c)
    design = np.column_stack([np.ones_like(v), v])
    coef, *_ = np.linalg.lstsq(design, log_hazard, rcond=None)
    return {"a": float(coef[0]), "b": float(coef[1])}

wayfault.adapters.outbound.calibrator_sklearn

scikit-learn covariate-hazard calibrator ([ml] extra, lazy import).

SklearnSurvivalCalibrator 

SklearnSurvivalCalibrator(**estimator_kwargs: object)

Fit a covariate-driven (log-)hazard model and reduce it to b.

Uses a gradient-boosting regressor on log(credit_factor) against the portfolio value, then linearises the fitted response to a single Hull-White b slope that a :class:DependenceModel can consume. Raises :class:MissingDependencyError if scikit-learn is absent.

Source code in src/wayfault/adapters/outbound/calibrator_sklearn.py 
20
21
def __init__(self, **estimator_kwargs: object) -> None:
    self._kwargs = estimator_kwargs

fit 

fit(portfolio_value: ndarray, credit_factor: ndarray) -> dict[str, float]

Fit the survival surrogate and return {"b": slope}.

Source code in src/wayfault/adapters/outbound/calibrator_sklearn.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def fit(
    self, portfolio_value: np.ndarray, credit_factor: np.ndarray
) -> dict[str, float]:
    """Fit the survival surrogate and return ``{"b": slope}``."""
    ensemble = require("sklearn.ensemble", "ml")
    v = np.asarray(portfolio_value, dtype=float).ravel()
    c = np.asarray(credit_factor, dtype=float).ravel()
    if v.size != c.size:
        raise ValidationError("portfolio_value and credit_factor must be equal length.")
    if np.any(c <= 0.0):
        raise ValidationError("credit_factor (hazard) samples must be positive.")
    log_hazard = np.log(c)
    model = ensemble.GradientBoostingRegressor(**self._kwargs)
    model.fit(v.reshape(-1, 1), log_hazard)
    # Linearise: finite-difference slope of the fitted response over the
    # observed support, reduced to a single Hull-White coupling.
    lo, hi = float(np.min(v)), float(np.max(v))
    if hi - lo < 1e-12:
        return {"b": 0.0}
    grid = np.linspace(lo, hi, 50).reshape(-1, 1)
    pred = model.predict(grid)
    slope = float(np.polyfit(grid.ravel(), pred, 1)[0])
    return {"b": slope}

Outbound — Sinks

wayfault.adapters.outbound.sinks

Result sinks: in-memory dict and JSON report writer (stdlib json).

DictResultSink 

DictResultSink()

Captures the latest result as a plain dictionary in memory.

Source code in src/wayfault/adapters/outbound/sinks.py 
13
14
def __init__(self) -> None:
    self.result: dict[str, object] | None = None

write 

write(result: WWRResult) -> None

Store result.to_dict() on the sink.

Source code in src/wayfault/adapters/outbound/sinks.py 
16
17
18
def write(self, result: WWRResult) -> None:
    """Store ``result.to_dict()`` on the sink."""
    self.result = result.to_dict()

JsonReportWriter 

JsonReportWriter(path: str, indent: int = 2)

Writes the result to a JSON file using the stdlib json module.

Source code in src/wayfault/adapters/outbound/sinks.py 
24
25
26
def __init__(self, path: str, indent: int = 2) -> None:
    self._path = path
    self._indent = indent

write 

write(result: WWRResult) -> None

Serialise result to the configured path.

Source code in src/wayfault/adapters/outbound/sinks.py 
28
29
30
31
def write(self, result: WWRResult) -> None:
    """Serialise ``result`` to the configured path."""
    with open(self._path, "w", encoding="utf-8") as fh:
        json.dump(result.to_dict(), fh, indent=self._indent)

Outbound — Visualization

The plotting API is documented on the dedicated Visualization page.