Skip to content

Domain

Pure value objects and functions — stdlib + numpy only.

Tenors

wayfault.domain.tenors

The :class:TenorGrid value object.

TenorGrid dataclass 

TenorGrid(times: ndarray)

A strictly increasing grid of positive year-fractions t_1 < ... < t_n.

Parameters:

Name Type Description Default
times ndarray

One-dimensional array of strictly increasing, strictly positive year-fractions.

 required

Raises:

Type Description
ValidationError

If the grid is empty, not one-dimensional, contains non-positive values, or is not strictly increasing.

n property 

n: int

Number of tenors on the grid.

intervals property 

intervals: ndarray

Interval lengths :math:\Delta t_i.

The first interval is measured from 0 to t_1; subsequent intervals are consecutive differences. Length n.

midpoints property 

midpoints: ndarray

Midpoints of each interval (from the previous node). Length n.

from_list classmethod 

from_list(times: list[float]) -> TenorGrid

Build a grid from a plain Python list of year-fractions.

Source code in src/wayfault/domain/tenors.py 
43
44
45
46
@classmethod
def from_list(cls, times: list[float]) -> TenorGrid:
    """Build a grid from a plain Python list of year-fractions."""
    return cls(np.asarray(times, dtype=float))

Exposure

wayfault.domain.exposure

Exposure value objects: :class:ExposureCube and :class:EEProfile.

EEProfile dataclass 

EEProfile(grid: TenorGrid, values: ndarray)

An expected-exposure curve over a tenor grid.

Parameters:

Name Type Description Default
grid TenorGrid

The tenor grid the profile is defined on.

 required
values ndarray

Expected positive exposure E[V(t)+] at each tenor. Length must equal grid.n. Values must be non-negative.

 required

ExposureCube dataclass 

ExposureCube(grid: TenorGrid, values: ndarray)

A Monte-Carlo exposure cube of netting-set mark-to-market values.

Parameters:

Name Type Description Default
grid TenorGrid

The tenor grid the columns are defined on.

 required
values ndarray

Array of shape (n_scenarios, n_tenors) of mark-to-market values (can be positive or negative).

 required

n_scenarios property 

n_scenarios: int

Number of Monte-Carlo scenarios (rows).

n_tenors property 

n_tenors: int

Number of tenors (columns).

tenor_slice 

tenor_slice(i: int) -> np.ndarray

Return the scenario vector at tenor index i.

Source code in src/wayfault/domain/exposure.py 
76
77
78
def tenor_slice(self, i: int) -> np.ndarray:
    """Return the scenario vector at tenor index ``i``."""
    return self.values[:, i]

from_ee_profile classmethod 

from_ee_profile(profile: EEProfile) -> ExposureCube

Build a degenerate single-scenario cube from an :class:EEProfile.

This lets users with only a precomputed EE profile still run the independent metrics. The single row reproduces the profile exactly for the positive-exposure expectation.

Source code in src/wayfault/domain/exposure.py 
80
81
82
83
84
85
86
87
88
@classmethod
def from_ee_profile(cls, profile: EEProfile) -> ExposureCube:
    """Build a degenerate single-scenario cube from an :class:`EEProfile`.

    This lets users with only a precomputed EE profile still run the
    independent metrics. The single row reproduces the profile exactly for
    the positive-exposure expectation.
    """
    return cls(profile.grid, profile.values.reshape(1, -1))

Credit

wayfault.domain.credit

Credit term-structure value objects.

Provides :class:CreditCurve with interchangeable hazard / survival / marginal PD representations, and a :class:RecoveryRate value object. Two concrete hazard shapes are supported: flat and piecewise-constant.

RecoveryRate dataclass 

RecoveryRate(value: float)

A recovery rate R in [0, 1).

loss_given_default property 

loss_given_default: float

Loss-given-default 1 - R.

CreditCurve dataclass 

CreditCurve(knots: ndarray, hazards: ndarray, recovery: RecoveryRate)

A piecewise-constant hazard credit curve.

A flat curve is the degenerate single-segment case. Internally the curve is represented by piecewise-constant hazard rates over the segments defined by knots.

Parameters:

Name Type Description Default
knots ndarray

Strictly increasing positive segment end-times. Hazard hazards[k] applies on (knots[k-1], knots[k]] (with knots[-1] = 0). Beyond the last knot the final hazard is held flat.

 required
hazards ndarray

Non-negative piecewise-constant hazard rates, same length as knots.

 required
recovery RecoveryRate

The :class:RecoveryRate.

 required

flat classmethod 

flat(hazard: float, recovery: float = 0.4) -> CreditCurve

Construct a flat-hazard curve.

Source code in src/wayfault/domain/credit.py 
74
75
76
77
78
79
80
81
@classmethod
def flat(cls, hazard: float, recovery: float = 0.4) -> CreditCurve:
    """Construct a flat-hazard curve."""
    return cls(
        knots=np.array([100.0]),
        hazards=np.array([float(hazard)]),
        recovery=RecoveryRate(recovery),
    )

piecewise classmethod 

piecewise(knots: list[float], hazards: list[float], recovery: float = 0.4) -> CreditCurve

Construct a piecewise-constant hazard curve.

Source code in src/wayfault/domain/credit.py 
83
84
85
86
87
88
89
90
91
92
@classmethod
def piecewise(
    cls, knots: list[float], hazards: list[float], recovery: float = 0.4
) -> CreditCurve:
    """Construct a piecewise-constant hazard curve."""
    return cls(
        knots=np.asarray(knots, dtype=float),
        hazards=np.asarray(hazards, dtype=float),
        recovery=RecoveryRate(recovery),
    )

hazard 

hazard(t: ndarray) -> np.ndarray

Instantaneous hazard lambda(t) at each time in t.

Source code in src/wayfault/domain/credit.py 
94
95
96
97
98
99
def hazard(self, t: np.ndarray) -> np.ndarray:
    """Instantaneous hazard ``lambda(t)`` at each time in ``t``."""
    t = np.asarray(t, dtype=float)
    idx = np.searchsorted(self.knots, t, side="left")
    idx = np.clip(idx, 0, self.hazards.size - 1)
    return self.hazards[idx]

cumulative_hazard 

cumulative_hazard(t: ndarray) -> np.ndarray

Integrated hazard :math:\int_0^t \lambda(u)\,du at each time.

Source code in src/wayfault/domain/credit.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def cumulative_hazard(self, t: np.ndarray) -> np.ndarray:
    r"""Integrated hazard :math:`\int_0^t \lambda(u)\,du` at each time."""
    t = np.asarray(t, dtype=float)
    seg_start = np.concatenate(([0.0], self.knots[:-1]))
    seg_len = self.knots - seg_start
    cum_at_knot = np.concatenate(([0.0], np.cumsum(self.hazards * seg_len)))

    out = np.empty_like(t)
    for j, tj in np.ndenumerate(t):
        k = int(np.searchsorted(self.knots, tj, side="left"))
        k = min(k, self.hazards.size - 1)
        base = cum_at_knot[k]
        out[j] = base + self.hazards[k] * (tj - seg_start[k])
    return out

survival 

survival(grid: TenorGrid) -> np.ndarray

Survival probabilities S(t_i) on the grid.

Source code in src/wayfault/domain/credit.py 
116
117
118
119
def survival(self, grid: TenorGrid) -> np.ndarray:
    """Survival probabilities ``S(t_i)`` on the grid."""
    surv: np.ndarray = np.exp(-self.cumulative_hazard(grid.times))
    return surv

marginal_pd 

marginal_pd(grid: TenorGrid) -> np.ndarray

Marginal default probabilities PD(t_{i-1}, t_i) on the grid.

Returns the probability of default in each interval, length grid.n, with the first interval measured from time 0.

Source code in src/wayfault/domain/credit.py 
121
122
123
124
125
126
127
128
129
130
def marginal_pd(self, grid: TenorGrid) -> np.ndarray:
    """Marginal default probabilities ``PD(t_{i-1}, t_i)`` on the grid.

    Returns the probability of default in each interval, length ``grid.n``,
    with the first interval measured from time ``0``.
    """
    s = self.survival(grid)
    s_prev = np.concatenate(([1.0], s[:-1]))
    pd: np.ndarray = s_prev - s
    return pd

Portfolio

wayfault.domain.portfolio

Counterparty / netting-set aggregation.

NettingSet dataclass 

NettingSet(name: str, cube: ExposureCube)

A named netting set holding an exposure cube.

Counterparty dataclass 

Counterparty(name: str, netting_sets: tuple[NettingSet, ...])

A counterparty aggregating one or more netting sets.

grid property 

grid: TenorGrid

The shared tenor grid.

aggregate 

aggregate() -> ExposureCube

Aggregate netting-set cubes to a single counterparty-level cube.

Netting applies within a set; across sets values are summed scenario-by-scenario (no further netting benefit across sets).

Source code in src/wayfault/domain/portfolio.py 
45
46
47
48
49
50
51
52
53
54
def aggregate(self) -> ExposureCube:
    """Aggregate netting-set cubes to a single counterparty-level cube.

    Netting applies *within* a set; across sets values are summed
    scenario-by-scenario (no further netting benefit across sets).
    """
    total = np.zeros_like(self.netting_sets[0].cube.values)
    for ns in self.netting_sets:
        total = total + ns.cube.values
    return ExposureCube(self.grid, total)

Metrics

wayfault.domain.metrics

Baseline exposure metrics: EPE, ENE, PFE, EEPE.

epe 

epe(cube: ExposureCube) -> EEProfile

Expected positive exposure EPE(t) = E[V(t)+] per tenor.

Source code in src/wayfault/domain/metrics.py 
11
12
13
14
def epe(cube: ExposureCube) -> EEProfile:
    """Expected positive exposure ``EPE(t) = E[V(t)+]`` per tenor."""
    values = np.maximum(cube.values, 0.0).mean(axis=0)
    return EEProfile(cube.grid, values)

ene 

ene(cube: ExposureCube) -> EEProfile

Expected negative exposure ENE(t) = E[(-V(t))+] per tenor.

Source code in src/wayfault/domain/metrics.py 
17
18
19
20
def ene(cube: ExposureCube) -> EEProfile:
    """Expected negative exposure ``ENE(t) = E[(-V(t))+]`` per tenor."""
    values = np.maximum(-cube.values, 0.0).mean(axis=0)
    return EEProfile(cube.grid, values)

pfe 

pfe(cube: ExposureCube, q: float = 0.95) -> np.ndarray

Potential future exposure: the q-quantile of V(t)+ per tenor.

Source code in src/wayfault/domain/metrics.py 
23
24
25
26
27
28
29
def pfe(cube: ExposureCube, q: float = 0.95) -> np.ndarray:
    """Potential future exposure: the ``q``-quantile of ``V(t)+`` per tenor."""
    if not (0.0 < q < 1.0):
        raise ValidationError("PFE quantile q must be in (0, 1).")
    positive = np.maximum(cube.values, 0.0)
    quantiles: np.ndarray = np.quantile(positive, q, axis=0)
    return quantiles

eepe 

eepe(cube: ExposureCube) -> float

Effective expected positive exposure (time-weighted running-max EE).

EEE(t) is the running maximum of EPE up to t; EEPE is the time-weighted average of EEE over the grid (Basel definition), using the grid interval lengths as weights.

Source code in src/wayfault/domain/metrics.py 
32
33
34
35
36
37
38
39
40
41
42
43
def eepe(cube: ExposureCube) -> float:
    """Effective expected positive exposure (time-weighted running-max EE).

    EEE(t) is the running maximum of EPE up to ``t``; EEPE is the
    time-weighted average of EEE over the grid (Basel definition), using the
    grid interval lengths as weights.
    """
    ee = epe(cube).values
    eee = np.maximum.accumulate(ee)
    dt = cube.grid.intervals
    horizon = float(cube.grid.times[-1])
    return float(np.sum(eee * dt) / horizon)

CVA

wayfault.domain.cva

Unilateral CVA integral (independent and conditional).

discounted_cva 

discounted_cva(ee: EEProfile, curve: CreditCurve, grid: TenorGrid, discount: ndarray | None = None) -> float

Unilateral CVA over the grid.

Computes

.. math::

CVA = (1 - R) \sum_i DF(t_i)\, EE(t_i)\, PD(t_{i-1}, t_i)

Parameters:

Name Type Description Default
ee EEProfile

The (independent or conditional) expected-exposure profile.

 required
curve CreditCurve

The counterparty credit curve (supplies marginal PDs and recovery).

 required
grid TenorGrid

The tenor grid.

 required
discount ndarray | None

Optional discount factors on the grid (length grid.n). Defaults to 1.0 at every tenor.

 None

Returns:

Type Description
float

The CVA value.
Source code in src/wayfault/domain/cva.py 
13
14
15
16
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
54
def discounted_cva(
    ee: EEProfile,
    curve: CreditCurve,
    grid: TenorGrid,
    discount: np.ndarray | None = None,
) -> float:
    r"""Unilateral CVA over the grid.

    Computes

    .. math::

        CVA = (1 - R) \sum_i DF(t_i)\, EE(t_i)\, PD(t_{i-1}, t_i)

    Parameters
    ----------
    ee:
        The (independent or conditional) expected-exposure profile.
    curve:
        The counterparty credit curve (supplies marginal PDs and recovery).
    grid:
        The tenor grid.
    discount:
        Optional discount factors on the grid (length ``grid.n``). Defaults to
        ``1.0`` at every tenor.

    Returns
    -------
    float
        The CVA value.
    """
    if ee.grid != grid:
        raise ValidationError("EEProfile grid must match the supplied grid.")
    if discount is None:
        df = np.ones(grid.n)
    else:
        df = np.asarray(discount, dtype=float)
        if df.shape != (grid.n,):
            raise ValidationError("Discount factors must have length grid.n.")
    pd = curve.marginal_pd(grid)
    lgd = curve.recovery.loss_given_default
    return float(lgd * np.sum(df * ee.values * pd))

Wrong-Way Risk

wayfault.domain.wwr

Wrong-Way / Right-Way Risk: alpha multiplier, classification, diagnostics.

WWRClass

Bases: StrEnum

Classification of the dependence direction.

Diagnostics dataclass 

Diagnostics(uplift_pct: float, ee_ratio: ndarray, ee_hazard_corr: float)

Diagnostic summary of the WWR adjustment.

alpha_multiplier 

alpha_multiplier(wwr_cva: float, baseline_cva: float) -> float

Empirical alpha multiplier alpha = WWR-CVA / independent-CVA.

Returns 1.0 if the baseline CVA is (near) zero, since there is no exposure to amplify.

Source code in src/wayfault/domain/wwr.py 
21
22
23
24
25
26
27
28
29
def alpha_multiplier(wwr_cva: float, baseline_cva: float) -> float:
    """Empirical alpha multiplier ``alpha = WWR-CVA / independent-CVA``.

    Returns ``1.0`` if the baseline CVA is (near) zero, since there is no
    exposure to amplify.
    """
    if abs(baseline_cva) < 1e-300:
        return 1.0
    return wwr_cva / baseline_cva

ead 

ead(alpha: float, eepe: float) -> float

Exposure-at-default view EAD = alpha * EEPE.

Source code in src/wayfault/domain/wwr.py 
32
33
34
def ead(alpha: float, eepe: float) -> float:
    """Exposure-at-default view ``EAD = alpha * EEPE``."""
    return alpha * eepe

classify 

classify(dependence_param: float, alpha: float, tol: float = 1e-09) -> WWRClass

Label the dependence from the parameter sign and CVA uplift.

A positive parameter together with alpha > 1 is wrong-way; a negative parameter with alpha < 1 is right-way; otherwise neutral.

Source code in src/wayfault/domain/wwr.py 
37
38
39
40
41
42
43
44
45
46
47
def classify(dependence_param: float, alpha: float, tol: float = 1e-9) -> WWRClass:
    """Label the dependence from the parameter sign and CVA uplift.

    A positive parameter together with ``alpha > 1`` is wrong-way; a negative
    parameter with ``alpha < 1`` is right-way; otherwise neutral.
    """
    if dependence_param > tol and alpha > 1.0 + tol:
        return WWRClass.WRONG_WAY
    if dependence_param < -tol and alpha < 1.0 - tol:
        return WWRClass.RIGHT_WAY
    return WWRClass.NEUTRAL

diagnostics 

diagnostics(unconditional: EEProfile, conditional: EEProfile, hazard: ndarray, baseline_cva: float, wwr_cva: float) -> Diagnostics

Compute per-tenor diagnostics for the WWR adjustment.

Parameters:

Name Type Description Default
unconditional EEProfile

Unconditional EPE profile.

 required
conditional EEProfile

Conditional-on-default EE profile.

 required
hazard ndarray

Per-tenor hazard rates (length grid.n).

 required
baseline_cva float

The two CVA figures, for the uplift percentage.

 required
wwr_cva float

The two CVA figures, for the uplift percentage.

 required
Source code in src/wayfault/domain/wwr.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def diagnostics(
    unconditional: EEProfile,
    conditional: EEProfile,
    hazard: np.ndarray,
    baseline_cva: float,
    wwr_cva: float,
) -> Diagnostics:
    """Compute per-tenor diagnostics for the WWR adjustment.

    Parameters
    ----------
    unconditional:
        Unconditional EPE profile.
    conditional:
        Conditional-on-default EE profile.
    hazard:
        Per-tenor hazard rates (length ``grid.n``).
    baseline_cva, wwr_cva:
        The two CVA figures, for the uplift percentage.
    """
    uplift = 0.0 if abs(baseline_cva) < 1e-300 else (wwr_cva - baseline_cva) / baseline_cva * 100.0
    with np.errstate(divide="ignore", invalid="ignore"):
        ratio = np.where(unconditional.values > 0.0, conditional.values / unconditional.values, 1.0)
    ee = unconditional.values
    if ee.size > 1 and np.std(ee) > 0 and np.std(hazard) > 0:
        corr = float(np.corrcoef(ee, hazard)[0, 1])
    else:
        corr = 0.0
    return Diagnostics(uplift_pct=uplift, ee_ratio=ratio, ee_hazard_corr=corr)

Errors

wayfault.domain.errors

Domain exceptions for :mod:wayfault.

These exceptions are part of the pure domain layer and depend only on the standard library.

WayfaultError

Bases: Exception

Base class for all :mod:wayfault errors.

ValidationError

Bases: WayfaultError

Raised when a value object fails its construction-time invariants.

MissingDependencyError 

MissingDependencyError(package: str, extra: str)

Bases: WayfaultError

Raised when an optional adapter is used without its extra installed.

Parameters:

Name Type Description Default
package str

The importable package that was missing (e.g. "pandas").

 required
extra str

The :mod:wayfault extras group that provides it (e.g. "io").

 required
Source code in src/wayfault/domain/errors.py 
29
30
31
32
33
34
35
def __init__(self, package: str, extra: str) -> None:
    self.package = package
    self.extra = extra
    super().__init__(
        f"Optional dependency {package!r} is required for this feature. "
        f"Install it with: pip install 'wayfault[{extra}]'"
    )