Application¶

Service¶

wayfault.application.service ¶

The :class:WrongWayRiskService orchestration.

WrongWayRiskService ¶

Orchestrates the WWR estimation through the outbound ports.

The service contains no I/O and no numpy-free leakage of external concerns: everything outside the domain is reached through the request's ports.

estimate ¶

estimate(request: WWRRequest) -> WWRResult

Run the full WWR pipeline and return a :class:WWRResult.

Source code in src/wayfault/application/service.py

def estimate(self, request: WWRRequest) -> WWRResult:
    """Run the full WWR pipeline and return a :class:`WWRResult`."""
    cube = request.exposure.load()
    curve = request.credit.load()
    grid = cube.grid

    model = request.model
    model.calibrate_to_curve(curve, grid)

    # Baseline (independent) metrics.
    epe_profile = metrics.epe(cube)
    pfe_profile = metrics.pfe(cube, request.pfe_quantile)
    eepe_value = metrics.eepe(cube)
    baseline_cva = cva.discounted_cva(epe_profile, curve, grid, request.discount)

    # Conditional exposure and WWR-CVA.
    conditional = model.conditional_ee(cube, curve, grid)
    wwr_cva = cva.discounted_cva(conditional, curve, grid, request.discount)

    alpha = wwr.alpha_multiplier(wwr_cva, baseline_cva)
    params = _model_params(model)
    dep_param = _dependence_param(model)
    classification = wwr.classify(dep_param, alpha)
    ead_value = wwr.ead(alpha, eepe_value)

    diag = wwr.diagnostics(
        unconditional=epe_profile,
        conditional=conditional,
        hazard=curve.hazard(grid.times),
        baseline_cva=baseline_cva,
        wwr_cva=wwr_cva,
    )

    return WWRResult(
        baseline_cva=baseline_cva,
        wwr_cva=wwr_cva,
        alpha=alpha,
        classification=classification,
        tenors=grid.times,
        epe=epe_profile.values,
        conditional_ee=conditional.values,
        pfe=pfe_profile,
        eepe=eepe_value,
        ead=ead_value,
        uplift_pct=diag.uplift_pct,
        ee_ratio=diag.ee_ratio,
        ee_hazard_corr=diag.ee_hazard_corr,
        model=type(model).__name__,
        params=params,
    )

DTOs¶

wayfault.application.dto ¶

Data-transfer objects for the WWR use case.

WWRRequest `dataclass` ¶

WWRRequest(exposure: ExposureSource, credit: CreditCurveSource, model: DependenceModel, discount: ndarray | None = None, pfe_quantile: float = 0.95)

Inputs for a WWR estimation.

Parameters:

Name	Type	Description	Default
`exposure`	`ExposureSource`	Port supplying the exposure cube.	required
`credit`	`CreditCurveSource`	Port supplying the credit curve.	required
`model`	`DependenceModel`	The dependence model to apply.	required
`discount`	`ndarray \| None`	Optional discount factors on the grid (defaults to 1.0).	`None`
`pfe_quantile`	`float`	Quantile used for the PFE profile.	`0.95`

WWRResult `dataclass` ¶

WWRResult(baseline_cva: float, wwr_cva: float, alpha: float, classification: WWRClass, tenors: ndarray, epe: ndarray, conditional_ee: ndarray, pfe: ndarray, eepe: float, ead: float, uplift_pct: float, ee_ratio: ndarray, ee_hazard_corr: float, model: str, params: dict[str, float] = dict())

Outputs of a WWR estimation.

Attributes:

Name	Type	Description
`baseline_cva, wwr_cva, alpha`		The independent CVA, the WWR-adjusted CVA, and their ratio.
`classification`	`WWRClass`	WWR / RWR / NEUTRAL label.
`tenors`	`ndarray`	The tenor grid year-fractions (x-axis for the per-tenor profiles).
`epe, conditional_ee, pfe`		Per-tenor profiles (numpy arrays).
`eepe`	`float`	Effective EPE scalar.
`ead`	`float`	`alpha * eepe`.
`uplift_pct, ee_ratio, ee_hazard_corr`		Diagnostics.
`model, params`		Metadata about the dependence model used.

to_dict ¶

to_dict() -> dict[str, Any]

Return a JSON-serialisable dictionary representation.

Source code in src/wayfault/application/dto.py

def to_dict(self) -> dict[str, Any]:
    """Return a JSON-serialisable dictionary representation."""
    return {
        "baseline_cva": self.baseline_cva,
        "wwr_cva": self.wwr_cva,
        "alpha": self.alpha,
        "classification": self.classification.value,
        "tenors": self.tenors.tolist(),
        "epe": self.epe.tolist(),
        "conditional_ee": self.conditional_ee.tolist(),
        "pfe": self.pfe.tolist(),
        "eepe": self.eepe,
        "ead": self.ead,
        "uplift_pct": self.uplift_pct,
        "ee_ratio": self.ee_ratio.tolist(),
        "ee_hazard_corr": self.ee_hazard_corr,
        "model": self.model,
        "params": self.params,
    }

Inverse solvers¶

wayfault.application.inverse ¶

Inverse WWR solvers (prescriptive analytics).

Where the rest of the library answers "given this dependence, what is the WWR?", this module inverts the question:

:func:calibrate_to_alpha / :func:calibrate_to_cva — find the dependence parameter that reproduces a target alpha multiplier or WWR-CVA (e.g. a desk/regulator target, or a historically observed CVA).
:func:find_breakpoint — find the dependence level at which a chosen metric (alpha, WWR-CVA, EAD) crosses a threshold — a reverse-stress / capital breach diagnostic: "how much wrong-way correlation until we breach?".

All solvers root-find on the monotone metric-vs-parameter curve with a robust bracketed bisection (deterministic, numpy-only). The caller supplies a model_factory mapping a scalar parameter to a dependence model, so the same solver works for the Hull-White b, the Gaussian rho, or a copula theta.

SolveResult `dataclass` ¶

SolveResult(param: float, achieved: float, target: float, iterations: int, converged: bool, result: WWRResult)

Outcome of an inverse solve.

Attributes:

Name	Type	Description
`param`	`float`	The solved dependence parameter.
`achieved`	`float`	The metric value attained at `param`.
`target`	`float`	The requested target / threshold.
`iterations`	`int`	Number of bisection steps taken.
`converged`	`bool`	Whether the solver reached `tol` within `max_iter`.
`result`	`WWRResult`	The full :class:`WWRResult` evaluated at `param`.