mrv.validator – Regime diagnostics¶

mrv.validator – Regime diagnostics.

Validators¶

RepValidator: Representation Invariance (Paper 1)
ResValidator: Resolution Invariance (Paper 2)

Base class: BaseValidator – subclass to create custom tests.

class mrv.validator.BaseValidator(cfg=None, impact_fn=None)[source]¶

Bases: ABC

Base class for all validators.

Validators accept user-provided labels and compute invariance metrics. No model fitting is performed – users supply their own labels.

Parameters:

cfg (Optional[Dict[str, Any]]) – Configuration dict (for report paths, thresholds, etc.). Not required for core validation – only needed for report generation.
impact_fn (Optional[Callable[[ndarray, Series], float]]) – (labels: ndarray, prices: Series) -> float. When provided, computes a business impact matrix across specifications.

name: str = ''¶

abstractmethod validate(labels, **kwargs)[source]¶

Run the validation test on user-provided labels.

Return type:: Dict[str, Any]

class mrv.validator.RepValidator(cfg=None, impact_fn=None)[source]¶

Bases: BaseValidator

Representation Invariance validator.

Measures whether regime labels agree across different feature representations. Users provide their own labels – no model fitting is performed.

name: str = 'rep'¶

validate(labels, risk_proxy=None, prices=None)[source]¶

Run representation invariance test on user-provided labels.

Parameters:

labels (Dict[str, Dict[str, ndarray]]) – {asset_name: {spec_label: ndarray}}. Each asset must have >= 2 specifications. Each ndarray is a 1-D integer array of regime labels.
risk_proxy (Optional[Dict[str, ndarray]]) – {asset_name: ndarray}. A 1-D risk measure (e.g. rolling volatility) used for ordering consistency (Spearman). If not provided, ordering consistency is skipped.
prices (Optional[Dict[str, Series]]) – {asset_name: price_series}. Only used for business impact computation (if impact_fn was set).

Return type:

Dict[str, Any]

class mrv.validator.ResValidator(cfg=None, impact_fn=None)[source]¶

Bases: BaseValidator

Resolution Invariance validator.

Measures whether regime labels agree across time frequencies. Users provide their own labels – no model fitting is performed.

name: str = 'res'¶

validate(labels, event_window=None, calm_window=None)[source]¶

Run resolution invariance test on user-provided labels.

Parameters:

labels (Dict[str, Dict[str, Series]]) – {asset_name: {freq: pd.Series}}. Each Series must have a DatetimeIndex and contain integer labels. At least 2 frequencies per asset are required.
event_window (Optional[Tuple[str, str]]) – (start_date, end_date) for event-period analysis.
calm_window (Optional[Tuple[str, str]]) – (start_date, end_date) for calm-period analysis.

Return type:

Dict[str, Any]

mrv.validator.generate_report(json_path, template=None, cfg=None)[source]¶

Generate a PDF report from validator JSON + LaTeX template.

Works for both representation (Paper 1) and resolution (Paper 2) result JSON. The .tex file is always written; the PDF is produced only when pdflatex is available on PATH (otherwise None is returned).

Parameters:

json_path (str | Path)
template (Union[str, Path, None]) – template, or validator.report_template from cfg if it exists).
cfg (Optional[Dict[str, Any]])

Return type:

Optional[Path]

Sub-modules¶

Base validator¶

mrv.validator.base – Abstract base class for validators.

class mrv.validator.base.BaseValidator(cfg=None, impact_fn=None)[source]¶

Bases: ABC

Base class for all validators.

Validators accept user-provided labels and compute invariance metrics. No model fitting is performed – users supply their own labels.

Parameters:

cfg (Optional[Dict[str, Any]]) – Configuration dict (for report paths, thresholds, etc.). Not required for core validation – only needed for report generation.
impact_fn (Optional[Callable[[ndarray, Series], float]]) – (labels: ndarray, prices: Series) -> float. When provided, computes a business impact matrix across specifications.

name: str = ''¶

results: Dict[str, Any]¶

run_dir: Path | None¶

json_path: Path | None¶

abstractmethod validate(labels, **kwargs)[source]¶

Run the validation test on user-provided labels.

Return type:: Dict[str, Any]

Representation validator (Paper 1)¶

mrv.validator.rep – Representation Invariance validator.

Validates whether regime labels remain stable across different feature representations (Paper 1). Users supply pre-computed labels from their own models – this module only measures agreement.

Usage:

from mrv.validator.rep import RepValidator

v = RepValidator()
result = v.validate(labels={
    "SPY": {
        "vol+dd+var": labels_a,    # ndarray of regime labels
        "vol+var+cvar": labels_b,
        "skew+vol+var": labels_c,
    }
})

# With ordering consistency (requires a risk proxy per asset)
result = v.validate(
    labels={"SPY": {"rep_a": la, "rep_b": lb}},
    risk_proxy={"SPY": volatility_array},
)

class mrv.validator.rep.RepValidator(cfg=None, impact_fn=None)[source]¶

Bases: BaseValidator

Representation Invariance validator.

Measures whether regime labels agree across different feature representations. Users provide their own labels – no model fitting is performed.

name: str = 'rep'¶

validate(labels, risk_proxy=None, prices=None)[source]¶

Run representation invariance test on user-provided labels.

Parameters:

labels (Dict[str, Dict[str, ndarray]]) – {asset_name: {spec_label: ndarray}}. Each asset must have >= 2 specifications. Each ndarray is a 1-D integer array of regime labels.
risk_proxy (Optional[Dict[str, ndarray]]) – {asset_name: ndarray}. A 1-D risk measure (e.g. rolling volatility) used for ordering consistency (Spearman). If not provided, ordering consistency is skipped.
prices (Optional[Dict[str, Series]]) – {asset_name: price_series}. Only used for business impact computation (if impact_fn was set).

Return type:

Dict[str, Any]

Resolution validator (Paper 2)¶

mrv.validator.res – Resolution Invariance validator.

Validates whether regime labels agree across time frequencies (e.g. 5m / 15m / 1h / 1d). Users supply pre-computed labels at each frequency – this module only measures cross-frequency agreement.

Usage:

from mrv.validator.res import ResValidator

v = ResValidator()
result = v.validate(labels={
    "SPY": {
        "5m":  labels_5m,   # pd.Series with DatetimeIndex
        "15m": labels_15m,
        "1h":  labels_1h,
        "1d":  labels_1d,
    }
})

mrv.validator.res.align_labels_to_finest(labels_by_freq)[source]¶

Forward-fill each frequency’s labels onto the finest-resolution index.

The finest resolution is determined by the series with the most observations. All other series are forward-filled onto that index.

Return type:: Dict[str, Series]

mrv.validator.res.compute_ari_matrix(aligned, index_subset=None)[source]¶

Compute cross-frequency ARI matrix.

Return type:: DataFrame

mrv.validator.res.compute_all_metrics(aligned, index_subset=None)[source]¶

Compute ARI, AMI, and VI matrices in a single pass.

Return type:: Dict[str, DataFrame]

mrv.validator.res.mean_offdiag(mat)[source]¶

Mean of off-diagonal entries.

Return type:: Optional[float]

mrv.validator.res.permute_pvalue_mean_offdiag_ari(aligned, index_subset=None, n_perm=500, seed=42)[source]¶

Permutation test for mean off-diagonal ARI.

Return type:: Tuple[Optional[float], Optional[Tuple[float, float]]]

mrv.validator.res.subset_index_by_dates(index, start_date, end_date, tz='America/New_York')[source]¶

Select timestamps whose calendar date is in [start, end].

Uses ambiguous="NaT" to avoid AmbiguousTimeError on DST transitions (autumn clock-change produces duplicate hour that ambiguous="infer" cannot always resolve for sub-minute data). NaT entries are dropped before masking.

Return type:: DatetimeIndex

mrv.validator.res.compute_daily_outputs(aligned, rolling_days=7, tz='America/New_York')[source]¶

Build daily summaries and rolling ARI tables.

Return type:: Tuple[DataFrame, DataFrame, DataFrame, DataFrame]

mrv.validator.res.analyze_labels(asset_name, labels_by_freq, rolling_days=7, event_window=None, calm_window=None)[source]¶

Run cross-frequency analysis on pre-computed labels.

Parameters:

asset_name (str) – Name of the asset (for logging / output).
labels_by_freq (Dict[str, Series]) – {freq: pd.Series} – regime labels at each frequency. Each Series must have a DatetimeIndex.
rolling_days (int) – Window for rolling ARI summary.
event_window (Optional[Tuple[str, str]]) – If provided, compute ARI separately for event/calm periods.
calm_window (Optional[Tuple[str, str]]) – If provided, compute ARI separately for event/calm periods.

Return type:

Dict[str, Any]

class mrv.validator.res.ResValidator(cfg=None, impact_fn=None)[source]¶

Bases: BaseValidator

Resolution Invariance validator.

Measures whether regime labels agree across time frequencies. Users provide their own labels – no model fitting is performed.

name: str = 'res'¶

validate(labels, event_window=None, calm_window=None)[source]¶

Run resolution invariance test on user-provided labels.

Parameters:

labels (Dict[str, Dict[str, Series]]) – {asset_name: {freq: pd.Series}}. Each Series must have a DatetimeIndex and contain integer labels. At least 2 frequencies per asset are required.
event_window (Optional[Tuple[str, str]]) – (start_date, end_date) for event-period analysis.
calm_window (Optional[Tuple[str, str]]) – (start_date, end_date) for calm-period analysis.

Return type:

Dict[str, Any]

Metrics¶

mrv.validator.metrics – Label comparison metrics for regime diagnostics.

Standard statistical measures – not extensible by users.

mrv.validator.metrics.ari(labels_a, labels_b)[source]¶

Adjusted Rand Index. Range [-1,1]; 1=perfect, ~0=random.

Return type:: float

mrv.validator.metrics.ami(labels_a, labels_b)[source]¶

Adjusted Mutual Information. Range [0,1]; 1=perfect.

Return type:: float

mrv.validator.metrics.nmi(labels_a, labels_b)[source]¶

Normalized Mutual Information. Range [0,1]; 1=perfect.

Return type:: float

mrv.validator.metrics.ordering_consistency(labels_a, labels_b, features)[source]¶

Ordinal ordering consistency between two label sets.

Each representation’s states are ranked by mean risk (mean feature value). Each observation is then mapped to its state’s risk rank (0=lowest, K-1=highest). Returns Spearman correlation of these risk-rank sequences.

This measures whether the two representations agree on the relative risk ordering of observations, even if the exact partition boundaries differ.

Threshold: Spearman >= 0.85 indicates stable risk ordering.

Return type:: float

mrv.validator.metrics.variation_of_information(labels_a, labels_b)[source]¶

Variation of Information. Lower=more similar; 0=identical.

Return type:: float

Attribution¶

mrv.validator.attribution – Disagreement attribution & root cause analysis.

Three attribution methods: 1. Leave-one-out factor attribution (rep validator) 2. Frequency-pair decomposition (res validator) 3. Temporal hotspot detection (both validators)

mrv.validator.attribution.loo_factor_attribution(labels_dict, baseline_mean_ari)[source]¶

Leave-one-out factor attribution for representation invariance.

For each factor set i, remove it and recompute mean pairwise ARI from the remaining sets.

Returns:

{
    "baseline_mean_ari": 0.45,
    "scores": {"set_label": delta_ari, ...},
    "worst_contributor": "set_label",
    "summary": "..."
}

A positive delta means removing set i improves ARI → set i is a disagreement driver.

Return type:: Dict[str, Any]

mrv.validator.attribution.freq_pair_attribution(ari_matrix)[source]¶

Rank frequency pairs by pairwise ARI (ascending = worst first).

Returns a list of dicts with keys: freq_a, freq_b, ari, rank.

Return type:: List[Dict[str, Any]]

mrv.validator.attribution.temporal_attribution(labels_a, labels_b, window='1D', ari_threshold=0.3)[source]¶

Per-window ARI between two label sequences.

Groups timestamps by window (default: 1 calendar day), computes ARI per group, and flags hotspots where ARI < ari_threshold.

Returns DataFrame: window_start, n_obs, ari, is_hotspot.

Return type:: DataFrame

mrv.validator.attribution.generate_attribution_summary(attr_results, validator_type)[source]¶

Generate a plain-language attribution summary.

Return type:: str

Findings engine¶

mrv.validator.findings – Auto-generate SR 26-2 / OCC Bulletin 2026-13 findings from validation results. SR 11-7 (Federal Reserve, 2011) was superseded by SR 26-2 / OCC Bulletin 2026-13 on 2026-04-17; this module targets the new standard.

Severity levels: - Critical: overall_mean_ari < 0.1 - High: overall_mean_ari < ARI_THRESHOLD - Medium: any min_pairwise_ari < ARI_THRESHOLD - Low: any fallback triggered - Informational: all pass

class mrv.validator.findings.Finding(id, severity, title, description, evidence='', recommendation='', remediation_owner='', deadline='', management_response='')[source]¶

Bases: object

A single validation finding in SR 26-2 / OCC Bulletin 2026-13 format.

id: str¶

severity: str¶

title: str¶

description: str¶

evidence: str = ''¶

recommendation: str = ''¶

remediation_owner: str = ''¶

deadline: str = ''¶

management_response: str = ''¶

to_dict()[source]¶

Return type:: Dict[str, Any]

mrv.validator.findings.classify_severity(overall_mean_ari, min_pairwise_ari=None, any_fallback=False)[source]¶

Classify finding severity based on empirical rules.

Return type:: str

mrv.validator.findings.generate_findings(results, validator_type, overrides_path=None)[source]¶

Auto-generate findings from validation results.

Parameters:

results (Dict[str, Any]) – The assets dict from a validator run ({asset_name: result_dict}).
validator_type (str) – "rep" or "res".
overrides_path (Optional[Path]) – YAML file with user-provided overrides (owner, deadline, response).

Return type:

List[Finding]

mrv.validator.findings.overall_risk_rating(findings)[source]¶

Derive overall model risk rating from findings.

Return type:: str

mrv.validator.findings.findings_summary(findings)[source]¶

Count findings by severity.

Return type:: Dict[str, int]

Continuous monitoring¶

mrv.validator.monitor – Continuous monitoring with alerting.

Provides monitor() for incremental daily validation with persistent history tracking and configurable alerts (file + webhook).

Usage:

from mrv.validator.monitor import monitor

monitor("config.yaml", "rep", mode="init")          # baseline
monitor("config.yaml", "rep", mode="incremental")    # daily

mrv.validator.monitor.monitor(config=None, validator='rep', mode='incremental', cfg=None, impact_fn=None)[source]¶

Run monitoring cycle: validate -> update history -> check alerts.

Parameters:

config (Union[str, Path, None])
validator (str) – Only "rep" (representation invariance) is supported for continuous monitoring. Resolution invariance requires the caller to supply pre-fitted labels at each frequency; use validate_res() directly and build your own history loop.
mode (str) – "init" for first-run baseline, "incremental" for daily append.
cfg (Optional[Dict[str, Any]]) – Pre-loaded config dict. Overrides config when supplied.
impact_fn (callable, optional) – Optional business-impact callback passed through to the validator.

Raises:

ValueError – If validator is not "rep".

Return type:

Dict[str, Any]

Report generation¶

mrv.validator.report – Fill LaTeX template with JSON data and compile to PDF.

All text lives in the template. This module only: 1. Replaces {{KEY}} placeholders with data values 2. Evaluates %% IF_xxx / %% ELSE / %% ELIF_xxx / %% ENDIF conditionals 3. Expands %% BEGIN_ASSET / %% END_ASSET blocks per asset 4. Compiles .tex -> .pdf via pdflatex

Both representation (Paper 1) and resolution (Paper 2) validator JSON are supported. The renderer branches on data["test"]: the resolution case is rendered from overall_mean_ari (the Spearman / factor-set sections are omitted, since resolution invariance has no ordering layer).

mrv.validator.report.generate_report(json_path, template=None, cfg=None)[source]¶

Generate a PDF report from validator JSON + LaTeX template.

Parameters:

json_path (str | Path)
template (Union[str, Path, None]) – template, or validator.report_template from cfg if it exists).
cfg (Optional[Dict[str, Any]])

Return type:

Optional[Path]