Skip to content

Honest Early-Warning Auditor

scpn_phase_orchestrator.evaluation scores whether an early-warning detector has real skill, not just a high detection rate. A detector that alarms on 90 % of pre-transition events is worthless if it also alarms on 90 % of transition-free nulls; the honest question is whether it alarms on events more often than its own false-alarm rate explains. This subsystem answers that question for any detector — the SCPN suite, an AR(1)/Kendall-τ trend baseline, or a black-box deep classifier — because it reads only the per-segment score each one emits.

Why this exists: the early-warning-signals literature has repeatedly found that most indicators sit at chance once a matched false-alarm rate and a null model are imposed (Boettiger & Hastings 2012 asked for exactly this discipline; O'Brien & Clements 2023 confirmed the null result on real data). A packaged, detector-neutral harness that enforces that discipline — and seals the verdict — did not exist. That harness is the asset here, above any one detector.

What an audit reports

audit_detector takes two arrays of per-segment scores — one on genuine pre-transition event segments, one on transition-free null segments — and returns a DetectorAudit:

  • matched_threshold — the alarm threshold calibrated from the null scores so at most a target fraction of nulls alarm. It is placed just above an order statistic of the nulls, never on a fixed grid, so a detector whose nulls need a high gate is not silently clipped. -inf means the target permitted every null to alarm (the gate is fully open).
  • achieved_false_alarm — the false-alarm rate the threshold actually held, reported alongside the target so a detector held below (or, in a degenerate corpus, unable to hold) the target is visible rather than assumed.
  • detection_rate — the fraction of event segments that alarmed at that threshold.
  • p_value (the headline) — the one-sided label-permutation p-value: under the exchangeability null that events and nulls are interchangeable, the fraction of random relabellings whose event-slot alarm count reaches the observed count, add-one corrected so it is never zero. A small p-value means the events alarmed more than the matched false alarm on this corpus.

A convention flag beats_chance (p_value < alpha) is offered for convenience; the p-value itself is the honest quantity and is always reported. Higher score means more evidence of a transition — orient a falling statistic by negating it.

Convention alpha defaults to 0.05. audit_scoring_detector is the convenience wrapper for a detector still expressed as a callable on a raw window: it applies the scoring function to each event and null series, then defers to audit_detector.

A worked audit distinguishes skill from no skill

bench/honest_auditor_worked_example.py audits two detectors on one synthetic corpus of AR(1) event windows (rising autocorrelation — the textbook critical-slowing-down signature) and white-noise nulls. Both arms are zero-mean with the same marginal variance, so the window mean is genuinely uninformative and serves as an honest negative control:

$ python -m bench.honest_auditor_worked_example
lag1-autocorrelation   target_fa=0.10 achieved_fa=0.100 detect=1.000 p=9.999e-05 beats_chance=True  hash=d97f55721c09
window-mean-control    target_fa=0.10 achieved_fa=0.050 detect=0.050 p=0.6215    beats_chance=False hash=52242831ed23

The lag-1 autocorrelation detector — a real critical-slowing signal — beats chance decisively; the window-mean control lands at chance. Both verdicts are read from scores alone, so a competitor's classifier would be judged on identical footing.

A head-to-head between two real detectors

bench/auditor_detector_head_to_head.py goes further: it audits the real SCPN modal envelope-growth detector (modal_growth_score) against a published competitor — the Dakos et al. 2008 AR(1)/Kendall-τ rising-autocorrelation trend — on two synthetic-but-honest regimes (a growing oscillatory mode, and a monotone rising-autocorrelation slowdown). With clearly skilful detectors the permutation p-value saturates (both beat chance), so the auditor's detection rate at the matched false alarm is the discriminator, and it is regime-dependent:

$ python -m bench.auditor_detector_head_to_head
oscillatory  scpn-modal-growth    achieved_fa=0.100 detect=1.000 p=9.999e-05 beats_chance=True
oscillatory  ar1-kendall-tau      achieved_fa=0.100 detect=0.775 p=9.999e-05 beats_chance=True
monotone     scpn-modal-growth    achieved_fa=0.100 detect=0.550 p=9.999e-05 beats_chance=True
monotone     ar1-kendall-tau      achieved_fa=0.100 detect=0.775 p=9.999e-05 beats_chance=True

The envelope-growth detector leads on the oscillatory regime, the AR(1) competitor leads on the monotone one — the eigenvalue-regime-map finding, adjudicated without bias by one matched-false-alarm + permutation test. Real field data would replace the synthetic corpora without changing the auditor. This is the integration proof: the productised auditor plugs into actual detector code, not just toy scorers.

Sealing an audit

seal_detector_audit binds a verdict to its corpus provenance — an identifier and a caller-supplied capture timestamp — under a SHA-256 over its canonical JSON, reusing the same hashing path as the assurance bundle. Recomputing the hash from the recorded fields detects any later edit, so a published audit verdict cannot be quietly altered. A fully open (-inf) threshold is serialised to the string "-inf" so the record stays strict JSON. The sealed record carries an explicit disclaimer: an audit measures skill on the supplied corpus only and is not a certification of field performance.

Auditing from the command line

spo audit-detector runs the same audit without writing Python, so a detector's skill can be checked from a scores file. The file is a JSON object with event_scores and null_scores arrays of per-segment scores (higher means more evidence of a transition) and an optional detector_name:

spo audit-detector scores.json \
  --target-false-alarm 0.10 \
  --corpus-id grid-2026 \
  --captured-at 2026-07-07T15:00:00+02:00

Without --corpus-id/--captured-at the command prints the bare verdict; supply both (they must be given together) to seal it into a hash-addressed record. --output also writes the JSON to a file. Score entries must be finite numbers — a missing key, an empty list, or a non-numeric or non-finite entry is an error, never a silently dropped score. The command reads a local file and prints JSON; it never actuates, signs, or reaches the network.

Skill primitives

For callers composing their own harness, the detector-agnostic primitives are public: calibrate_score_threshold (matched-false-alarm calibration), matched_false_alarm_rate, permutation_significance_from_alarms (the exchangeability test), and surrogate_rank_pvalue (the single-statistic counterpart against a surrogate ensemble).

auditor

Audit any early-warning detector's event-vs-null skill honestly.

:func:audit_detector takes only two arrays of per-segment scores — one on genuine pre-transition event segments, one on transition-free null segments — and returns a :class:DetectorAudit: the threshold that holds the null false-alarm rate at a target, the rate it actually achieved, how many events alarmed at it, and the label-permutation p-value that says whether the events alarm more than that matched rate by chance. Because it reads scores, not the detector's internals, it audits the SCPN suite, an AR(1)/Kendall-τ baseline, and a black-box deep classifier on the same footing.

:func:audit_scoring_detector is the convenience wrapper for a detector still expressed as a callable: it applies a scoring function to each raw event and null series to obtain the two score arrays, then defers to :func:audit_detector.

The verdict carries a beats_chance boolean only as a convenience at a caller chosen alpha; the honest quantity is the p-value itself, always reported. An audit is a statement about the supplied corpus and detector, not a certification of field performance — seal it with :mod:~scpn_phase_orchestrator.evaluation.record to make the corpus, scores, and verdict tamper-evident.

Classes

DetectorAudit dataclass 

DetectorAudit(
    detector_name: str,
    target_false_alarm: float,
    matched_threshold: float,
    achieved_false_alarm: float,
    n_events: int,
    n_events_alarmed: int,
    detection_rate: float,
    n_nulls: int,
    significance: PermutationSignificance,
    alpha: float,
    beats_chance: bool,
)

The honest event-vs-null skill of one detector at a matched false alarm.

Attributes

detector_name : str A label for the audited detector, carried into the sealed record. target_false_alarm : float The false-alarm rate the threshold was calibrated to hold at or below. matched_threshold : float The calibrated alarm threshold; -inf means the gate is fully open because the target permitted every null to alarm. achieved_false_alarm : float The false-alarm rate the threshold actually held on the null scores. n_events : int Number of event segments audited. n_events_alarmed : int Number of event segments that alarmed at matched_threshold. detection_rate : float Fraction of event segments that alarmed (n_events_alarmed / n_events). n_nulls : int Number of null segments audited. significance : PermutationSignificance The label-permutation test of the event alarm count against the null. alpha : float The significance level at which beats_chance was decided. beats_chance : bool Whether significance.p_value < alpha — a convenience, not the finding; the p-value is the honest quantity.

Attributes
p_value property 
p_value: float

The permutation p-value — the audit's headline number.

Methods:
to_record 
to_record() -> dict[str, object]

Return a JSON-safe mapping of the audit verdict.

The matched_threshold is emitted as the string "-inf" when the gate is fully open, so the record stays strict JSON (a hash of it rejects non-finite numbers).

Returns

dict[str, object] The detector label, target and achieved false alarm, matched threshold, detection counts, nested permutation significance, and the beats_chance decision.

Source code in src/scpn_phase_orchestrator/evaluation/auditor.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def to_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the audit verdict.

    The ``matched_threshold`` is emitted as the string ``"-inf"`` when the
    gate is fully open, so the record stays strict JSON (a hash of it rejects
    non-finite numbers).

    Returns
    -------
    dict[str, object]
        The detector label, target and achieved false alarm, matched
        threshold, detection counts, nested permutation significance, and the
        ``beats_chance`` decision.
    """
    threshold: object = self.matched_threshold
    if self.matched_threshold == float("-inf"):
        threshold = "-inf"
    return {
        "detector_name": self.detector_name,
        "target_false_alarm": self.target_false_alarm,
        "matched_threshold": threshold,
        "achieved_false_alarm": self.achieved_false_alarm,
        "n_events": self.n_events,
        "n_events_alarmed": self.n_events_alarmed,
        "detection_rate": self.detection_rate,
        "n_nulls": self.n_nulls,
        "significance": self.significance.to_record(),
        "alpha": self.alpha,
        "beats_chance": self.beats_chance,
    }

Functions:

audit_detector 

audit_detector(
    *,
    event_scores: Sequence[float],
    null_scores: Sequence[float],
    detector_name: str = "detector",
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> DetectorAudit

Audit a detector's event-vs-null skill at a matched false-alarm rate.

Calibrate an alarm threshold on the null scores to hold the false-alarm rate at target_false_alarm, count how many event scores alarm at it, and test that count against the exchangeability null with a label permutation. Higher score means more evidence of a transition; orient a falling statistic by negating it before calling.

Parameters

event_scores : sequence of float One per-segment score on each genuine pre-transition event segment. null_scores : sequence of float One per-segment score on each transition-free null segment. detector_name : str A label for the audited detector. target_false_alarm : float The false-alarm rate to hold the threshold at or below, in [0, 1]. n_permutations : int Random relabellings drawn for the permutation p-value. seed : int Seed of the permutation resampling. alpha : float Significance level for the convenience beats_chance flag, in [0, 1].

Returns

DetectorAudit The calibrated threshold, achieved false alarm, detection rate, and the permutation significance of the event alarm count.

Raises

ValueError If event_scores is empty or alpha is not in [0, 1]. (Empty null_scores and an out-of-range target_false_alarm are rejected by :func:~scpn_phase_orchestrator.evaluation.skill.calibrate_score_threshold.)

Source code in src/scpn_phase_orchestrator/evaluation/auditor.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def audit_detector(
    *,
    event_scores: Sequence[float],
    null_scores: Sequence[float],
    detector_name: str = "detector",
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> DetectorAudit:
    """Audit a detector's event-vs-null skill at a matched false-alarm rate.

    Calibrate an alarm threshold on the null scores to hold the false-alarm rate
    at ``target_false_alarm``, count how many event scores alarm at it, and test
    that count against the exchangeability null with a label permutation. Higher
    score means more evidence of a transition; orient a falling statistic by
    negating it before calling.

    Parameters
    ----------
    event_scores : sequence of float
        One per-segment score on each genuine pre-transition event segment.
    null_scores : sequence of float
        One per-segment score on each transition-free null segment.
    detector_name : str
        A label for the audited detector.
    target_false_alarm : float
        The false-alarm rate to hold the threshold at or below, in ``[0, 1]``.
    n_permutations : int
        Random relabellings drawn for the permutation p-value.
    seed : int
        Seed of the permutation resampling.
    alpha : float
        Significance level for the convenience ``beats_chance`` flag, in ``[0, 1]``.

    Returns
    -------
    DetectorAudit
        The calibrated threshold, achieved false alarm, detection rate, and the
        permutation significance of the event alarm count.

    Raises
    ------
    ValueError
        If ``event_scores`` is empty or ``alpha`` is not in ``[0, 1]``. (Empty
        ``null_scores`` and an out-of-range ``target_false_alarm`` are rejected by
        :func:`~scpn_phase_orchestrator.evaluation.skill.calibrate_score_threshold`.)
    """
    if len(event_scores) == 0:
        raise ValueError("event_scores must not be empty")
    if not 0.0 <= alpha <= 1.0:
        raise ValueError(f"alpha must be in [0, 1], got {alpha}")
    threshold = calibrate_score_threshold(null_scores, target_fa=target_false_alarm)
    achieved = matched_false_alarm_rate(null_scores, threshold)
    event_alarms = [float(score) >= threshold for score in event_scores]
    null_alarms = [float(score) >= threshold for score in null_scores]
    significance = permutation_significance_from_alarms(
        event_alarms, null_alarms, n_permutations=n_permutations, seed=seed
    )
    n_events = len(event_alarms)
    n_alarmed = int(sum(event_alarms))
    return DetectorAudit(
        detector_name=detector_name,
        target_false_alarm=target_false_alarm,
        matched_threshold=threshold,
        achieved_false_alarm=achieved,
        n_events=n_events,
        n_events_alarmed=n_alarmed,
        detection_rate=n_alarmed / n_events,
        n_nulls=len(null_alarms),
        significance=significance,
        alpha=alpha,
        beats_chance=significance.p_value < alpha,
    )

audit_scoring_detector 

audit_scoring_detector(
    *,
    score: Callable[[_Segment], float],
    event_series: Sequence[_Segment],
    null_series: Sequence[_Segment],
    detector_name: str = "detector",
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> DetectorAudit

Audit a detector expressed as a per-series scoring callable.

Apply score to each raw event and null series to obtain the two score arrays, then defer to :func:audit_detector. A convenience for a detector that still lives as a function of a raw window rather than a precomputed score; the honest evaluation is identical.

Parameters

score : callable Maps one raw series to a single per-segment score (higher = more evidence of a transition). event_series : sequence of sequence of float The raw pre-transition event segments. null_series : sequence of sequence of float The raw transition-free null segments. detector_name, target_false_alarm, n_permutations, seed, alpha : Forwarded to :func:audit_detector.

Returns

DetectorAudit The audit of the scored segments.

Raises

ValueError If event_series or null_series is empty.

Source code in src/scpn_phase_orchestrator/evaluation/auditor.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def audit_scoring_detector(
    *,
    score: Callable[[_Segment], float],
    event_series: Sequence[_Segment],
    null_series: Sequence[_Segment],
    detector_name: str = "detector",
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> DetectorAudit:
    """Audit a detector expressed as a per-series scoring callable.

    Apply ``score`` to each raw event and null series to obtain the two score
    arrays, then defer to :func:`audit_detector`. A convenience for a detector
    that still lives as a function of a raw window rather than a precomputed
    score; the honest evaluation is identical.

    Parameters
    ----------
    score : callable
        Maps one raw series to a single per-segment score (higher = more evidence
        of a transition).
    event_series : sequence of sequence of float
        The raw pre-transition event segments.
    null_series : sequence of sequence of float
        The raw transition-free null segments.
    detector_name, target_false_alarm, n_permutations, seed, alpha :
        Forwarded to :func:`audit_detector`.

    Returns
    -------
    DetectorAudit
        The audit of the scored segments.

    Raises
    ------
    ValueError
        If ``event_series`` or ``null_series`` is empty.
    """
    if len(event_series) == 0:
        raise ValueError("event_series must not be empty")
    if len(null_series) == 0:
        raise ValueError("null_series must not be empty")
    event_scores = [float(score(series)) for series in event_series]
    null_scores = [float(score(series)) for series in null_series]
    return audit_detector(
        event_scores=event_scores,
        null_scores=null_scores,
        detector_name=detector_name,
        target_false_alarm=target_false_alarm,
        n_permutations=n_permutations,
        seed=seed,
        alpha=alpha,
    )

skill

Detector-agnostic primitives for scoring early-warning skill honestly.

An early-warning detector is only useful if it fires on genuine pre-transition segments more often than its own false-alarm rate on transition-free nulls. A raw detection rate hides that: a detector that alarms on 90 % of events but also on 90 % of quiet nulls has learnt nothing. These primitives make the honest comparison, on bounded per-segment scores alone, so they judge any detector — the SCPN suite, an AR(1)/Kendall-τ trend baseline, or a black-box deep classifier — by the single number each emits per segment.

Two operations compose into an audit:

  • :func:calibrate_score_threshold sets the alarm threshold from the null scores so at most a target fraction of transition-free segments alarm — the matched false-alarm rate. The threshold is placed just above an order statistic of the nulls, never on a fixed grid, so a detector whose nulls need a high gate is not silently clipped.
  • :func:permutation_significance_from_alarms then asks whether the event segments alarm more than that matched rate by chance, under the exchangeability null that events and nulls are interchangeable. :func:surrogate_rank_pvalue is the same one-sided add-one-corrected rank test for a single observed statistic against a surrogate ensemble.

The primitives are pure and free of any SCPN detector or observable type; the suite-specific glue that turns SCPN observables into per-segment scores lives in the benchmark harness, not here.

References

  • Boettiger & Hastings 2012, J. R. Soc. Interface 9, 2527 — early-warning signals need a null model and a quantified false-positive rate.
  • Scheffer et al. 2009, Nature 461, 53 — generic early-warning signals.

Classes

PermutationSignificance dataclass 

PermutationSignificance(
    observed_alarms: int,
    n_events: int,
    pooled_alarm_rate: float,
    expected_alarms: float,
    p_value: float,
    n_permutations: int,
    seed: int,
)

Whether an event alarm count beats a matched-false-alarm null by chance.

A threshold calibrated to a matched false alarm makes some fraction of the transition-free nulls alarm by construction. The open question a raw event alarm count leaves is whether the events alarm more often than that — or whether the count is what a random relabelling of events and nulls would give. This holds the answer as a label-permutation (exchangeability) test.

Attributes

observed_alarms : int Number of event segments that alarmed at the calibrated threshold. n_events : int Number of event segments tested. pooled_alarm_rate : float Fraction of the pooled event-and-null segments that alarmed. expected_alarms : float Alarm count expected under the null (n_events × pooled_alarm_rate). p_value : float One-sided permutation p-value: the fraction of random relabellings whose event-slot alarm count reached observed_alarms, with an add-one correction so it is never zero. n_permutations : int Number of random relabellings drawn. seed : int Seed of the resampling, so the p-value is reproducible.

Methods:
to_record 
to_record() -> dict[str, object]

Return a JSON-safe mapping of the significance test.

Returns

dict[str, object] The observed and expected alarm counts, the pooled alarm rate, the permutation p-value, and the resampling parameters.

Source code in src/scpn_phase_orchestrator/evaluation/skill.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def to_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the significance test.

    Returns
    -------
    dict[str, object]
        The observed and expected alarm counts, the pooled alarm rate, the
        permutation p-value, and the resampling parameters.
    """
    return {
        "observed_alarms": self.observed_alarms,
        "n_events": self.n_events,
        "pooled_alarm_rate": self.pooled_alarm_rate,
        "expected_alarms": self.expected_alarms,
        "p_value": self.p_value,
        "n_permutations": self.n_permutations,
        "seed": self.seed,
    }

Functions:

calibrate_score_threshold 

calibrate_score_threshold(
    null_scores: Sequence[float],
    *,
    target_fa: float = DEFAULT_TARGET_FALSE_ALARM,
) -> float

Return the tightest score threshold holding the null false-alarm rate at target.

The matched-false-alarm calibrator for a bounded per-segment score — any statistic already on its own scale (a Kendall-τ, a growth rate, a classifier logit). The null scores are sorted and the alarm threshold placed just above the floor(target_fa · n)-th largest, so at most that fraction of nulls has score ≥ threshold. A convention of higher score = more evidence of a transition is assumed; orient a falling statistic by negating it first.

Parameters

null_scores : sequence of float The per-segment score of each transition-free null trial. target_fa : float Target false-alarm rate the detector is held at or below.

Returns

float The matched-false-alarm score threshold; -inf (the gate fully open) when the budget permits every null to alarm.

Raises

ValueError If null_scores is empty or target_fa is not in [0, 1].

Source code in src/scpn_phase_orchestrator/evaluation/skill.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def calibrate_score_threshold(
    null_scores: Sequence[float], *, target_fa: float = DEFAULT_TARGET_FALSE_ALARM
) -> float:
    """Return the tightest score threshold holding the null false-alarm rate at target.

    The matched-false-alarm calibrator for a bounded per-segment score — any
    statistic already on its own scale (a Kendall-τ, a growth rate, a classifier
    logit). The null scores are sorted and the alarm threshold placed just above
    the ``floor(target_fa · n)``-th largest, so at most that fraction of nulls has
    ``score ≥ threshold``. A convention of *higher score = more evidence of a
    transition* is assumed; orient a falling statistic by negating it first.

    Parameters
    ----------
    null_scores : sequence of float
        The per-segment score of each transition-free null trial.
    target_fa : float
        Target false-alarm rate the detector is held at or below.

    Returns
    -------
    float
        The matched-false-alarm score threshold; ``-inf`` (the gate fully open)
        when the budget permits every null to alarm.

    Raises
    ------
    ValueError
        If ``null_scores`` is empty or ``target_fa`` is not in ``[0, 1]``.
    """
    if len(null_scores) == 0:
        raise ValueError("null_scores must not be empty")
    if not 0.0 <= target_fa <= 1.0:
        raise ValueError(f"target_fa must be in [0, 1], got {target_fa}")
    scores = sorted((float(score) for score in null_scores), reverse=True)
    allowed = int(np.floor(target_fa * len(scores)))
    if allowed >= len(scores):
        return float(-np.inf)
    return float(np.nextafter(scores[allowed], np.inf))

matched_false_alarm_rate 

matched_false_alarm_rate(
    null_scores: Sequence[float], threshold: float
) -> float

Return the fraction of null scores that alarm at threshold.

The false-alarm rate a threshold actually achieves on the nulls — reported alongside the target so a detector held below target (or, in a degenerate corpus, unable to be) is visible rather than assumed.

Parameters

null_scores : sequence of float The per-segment score of each transition-free null trial. threshold : float The alarm threshold; a null alarms when score ≥ threshold.

Returns

float The fraction of nulls with score ≥ threshold.

Raises

ValueError If null_scores is empty.

Source code in src/scpn_phase_orchestrator/evaluation/skill.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def matched_false_alarm_rate(null_scores: Sequence[float], threshold: float) -> float:
    """Return the fraction of null scores that alarm at ``threshold``.

    The false-alarm rate a threshold actually achieves on the nulls — reported
    alongside the target so a detector held below target (or, in a degenerate
    corpus, unable to be) is visible rather than assumed.

    Parameters
    ----------
    null_scores : sequence of float
        The per-segment score of each transition-free null trial.
    threshold : float
        The alarm threshold; a null alarms when ``score ≥ threshold``.

    Returns
    -------
    float
        The fraction of nulls with ``score ≥ threshold``.

    Raises
    ------
    ValueError
        If ``null_scores`` is empty.
    """
    if len(null_scores) == 0:
        raise ValueError("null_scores must not be empty")
    alarms = sum(float(score) >= threshold for score in null_scores)
    return alarms / len(null_scores)

permutation_significance_from_alarms 

permutation_significance_from_alarms(
    event_alarms: Sequence[bool],
    null_alarms: Sequence[bool],
    *,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
) -> PermutationSignificance

Test whether an event alarm count beats a matched-false-alarm null.

Needs only the binary alarm outcome of each event segment and each null trial, so it scores any detector once its per-segment score is thresholded to a matched false alarm. Under the null that event segments are exchangeable with nulls, drawing n_permutations random event-sized subsets of the pooled outcomes builds the null distribution of the alarm count; the one-sided p-value is the fraction reaching the observed count, with an add-one correction.

Parameters

event_alarms : sequence of bool Whether each event segment alarmed at its calibrated threshold. null_alarms : sequence of bool Whether each null trial alarmed at that threshold. n_permutations : int Number of random relabellings to draw; must be a positive integer. seed : int Seed of the resampling, so the p-value is reproducible.

Returns

PermutationSignificance The observed and expected alarm counts and the permutation p-value.

Raises

ValueError If either alarm set is empty or n_permutations is not positive.

Source code in src/scpn_phase_orchestrator/evaluation/skill.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def permutation_significance_from_alarms(
    event_alarms: Sequence[bool],
    null_alarms: Sequence[bool],
    *,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
) -> PermutationSignificance:
    """Test whether an event alarm count beats a matched-false-alarm null.

    Needs only the binary alarm outcome of each event segment and each null trial,
    so it scores any detector once its per-segment score is thresholded to a
    matched false alarm. Under the null that event segments are exchangeable with
    nulls, drawing ``n_permutations`` random event-sized subsets of the pooled
    outcomes builds the null distribution of the alarm count; the one-sided
    p-value is the fraction reaching the observed count, with an add-one
    correction.

    Parameters
    ----------
    event_alarms : sequence of bool
        Whether each event segment alarmed at its calibrated threshold.
    null_alarms : sequence of bool
        Whether each null trial alarmed at that threshold.
    n_permutations : int
        Number of random relabellings to draw; must be a positive integer.
    seed : int
        Seed of the resampling, so the p-value is reproducible.

    Returns
    -------
    PermutationSignificance
        The observed and expected alarm counts and the permutation p-value.

    Raises
    ------
    ValueError
        If either alarm set is empty or ``n_permutations`` is not positive.
    """
    if len(event_alarms) == 0:
        raise ValueError("event_alarms must not be empty")
    if len(null_alarms) == 0:
        raise ValueError("null_alarms must not be empty")
    draws = _positive_int(n_permutations, "n_permutations")
    events = [bool(alarm) for alarm in event_alarms]
    nulls = [bool(alarm) for alarm in null_alarms]
    observed = int(sum(events))
    n_events = len(events)
    pool = np.array(events + nulls, dtype=bool)
    total = int(pool.shape[0])
    rng = np.random.default_rng(seed)
    reached = 0
    for _ in range(draws):
        subset = rng.permutation(total)[:n_events]
        if int(pool[subset].sum()) >= observed:
            reached += 1
    p_value = (1 + reached) / (draws + 1)
    pooled_rate = float(pool.mean())
    return PermutationSignificance(
        observed_alarms=observed,
        n_events=n_events,
        pooled_alarm_rate=pooled_rate,
        expected_alarms=n_events * pooled_rate,
        p_value=p_value,
        n_permutations=draws,
        seed=int(seed),
    )

surrogate_rank_pvalue 

surrogate_rank_pvalue(
    observed: float, surrogates: Sequence[float]
) -> float

Return the one-sided surrogate-rank p-value with the add-one correction.

The single-statistic counterpart to :func:permutation_significance_from_alarms: given one observed statistic and an ensemble of surrogate statistics drawn under the null (e.g. label-shuffled or phase-randomised), the p-value is the fraction of surrogates reaching the observed value, add-one corrected so it is never zero. A convention of higher = stronger is assumed.

Parameters

observed : float The observed statistic. surrogates : sequence of float The null-ensemble surrogate statistics.

Returns

float (1 + #{surrogate ≥ observed}) / (1 + n_surrogates) — never zero.

Raises

ValueError If surrogates is empty.

Source code in src/scpn_phase_orchestrator/evaluation/skill.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def surrogate_rank_pvalue(observed: float, surrogates: Sequence[float]) -> float:
    """Return the one-sided surrogate-rank p-value with the add-one correction.

    The single-statistic counterpart to :func:`permutation_significance_from_alarms`:
    given one observed statistic and an ensemble of surrogate statistics drawn
    under the null (e.g. label-shuffled or phase-randomised), the p-value is the
    fraction of surrogates reaching the observed value, add-one corrected so it is
    never zero. A convention of *higher = stronger* is assumed.

    Parameters
    ----------
    observed : float
        The observed statistic.
    surrogates : sequence of float
        The null-ensemble surrogate statistics.

    Returns
    -------
    float
        ``(1 + #{surrogate ≥ observed}) / (1 + n_surrogates)`` — never zero.

    Raises
    ------
    ValueError
        If ``surrogates`` is empty.
    """
    if len(surrogates) == 0:
        raise ValueError("surrogates must not be empty")
    reached = int(sum(1 for score in surrogates if float(score) >= observed))
    return (1 + reached) / (1 + len(surrogates))

benjamini_hochberg 

benjamini_hochberg(
    p_values: Sequence[float],
) -> list[float]

Return Benjamini–Hochberg FDR-adjusted p-values, preserving input order.

When several detectors are tested on the same corpus, the smallest raw p-value across the family is optimistic — some detector beats chance by luck. The Benjamini–Hochberg step-up procedure controls the false discovery rate: the k-th smallest of n p-values is scaled by n / k, the resulting sequence is made monotone non-decreasing from the largest downward, and every value is clamped to [0, 1]. Reporting the adjusted value beside the raw one stops a leaderboard from turning a family of raw p-values into a cherry-picked "winner".

Parameters

p_values : sequence of float The raw p-values of the family being corrected together, each in [0, 1]. Order is preserved in the result.

Returns

list of float The FDR-adjusted p-values in the same order as p_values.

Raises

ValueError If p_values is empty or any value lies outside [0, 1].

Source code in src/scpn_phase_orchestrator/evaluation/skill.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def benjamini_hochberg(p_values: Sequence[float]) -> list[float]:
    """Return Benjamini–Hochberg FDR-adjusted p-values, preserving input order.

    When several detectors are tested on the same corpus, the smallest raw
    p-value across the family is optimistic — some detector beats chance by luck.
    The Benjamini–Hochberg step-up procedure controls the *false discovery rate*:
    the ``k``-th smallest of ``n`` p-values is scaled by ``n / k``, the resulting
    sequence is made monotone non-decreasing from the largest downward, and every
    value is clamped to ``[0, 1]``. Reporting the adjusted value beside the raw one
    stops a leaderboard from turning a family of raw p-values into a cherry-picked
    "winner".

    Parameters
    ----------
    p_values : sequence of float
        The raw p-values of the family being corrected together, each in
        ``[0, 1]``. Order is preserved in the result.

    Returns
    -------
    list of float
        The FDR-adjusted p-values in the same order as ``p_values``.

    Raises
    ------
    ValueError
        If ``p_values`` is empty or any value lies outside ``[0, 1]``.
    """
    if len(p_values) == 0:
        raise ValueError("p_values must not be empty")
    values = [float(p) for p in p_values]
    for value in values:
        if not 0.0 <= value <= 1.0:
            raise ValueError(f"p-values must be in [0, 1], got {value}")
    n = len(values)
    order = sorted(range(n), key=lambda index: values[index])
    adjusted = [0.0] * n
    running_min = 1.0
    for rank in range(n, 0, -1):
        index = order[rank - 1]
        running_min = min(running_min, values[index] * n / rank)
        adjusted[index] = min(1.0, running_min)
    return adjusted

record

Seal a detector audit into a tamper-evident, hash-addressed record.

:func:seal_detector_audit binds a :class:~scpn_phase_orchestrator.evaluation.auditor.DetectorAudit to the provenance of the corpus it was measured on — an identifier and a caller-supplied capture timestamp — and stamps the whole with a SHA-256 over its canonical JSON. Recomputing the hash from the recorded fields detects any later edit, so a published audit verdict cannot be quietly altered. The record makes no claim beyond the supplied corpus and detector; the disclaimer says so.

The bare SHA-256 seal is tamper-evident but not tamper-resistant: anyone who can edit the record can recompute the hash. Pass a signing key (or call :meth:AuditRecord.sign) to add an HMAC-SHA256 signature over the content hash, reusing the same key discovery as the audit log (:mod:~scpn_phase_orchestrator.runtime.audit_signing); a verifier without the secret then cannot forge an edit. Signing is additive — an unsigned record's serialisation and content hash are unchanged, so records sealed before signing existed still verify.

Classes

AuditRecord dataclass 

AuditRecord(
    corpus_id: str,
    captured_at: str,
    audit: dict[str, object],
    framework: str = AUDIT_FRAMEWORK,
    disclaimer: str = AUDIT_DISCLAIMER,
    signature: str | None = None,
    signing_key_id: str | None = None,
)

A hash-sealed early-warning detector audit bound to its corpus provenance.

Attributes

corpus_id : str Provenance identifier of the event-and-null corpus the audit ran on. captured_at : str Caller-supplied timestamp of the audit, echoed for provenance. audit : dict[str, object] The JSON-safe audit verdict (:meth:DetectorAudit.to_record). framework : str The methodology label (:data:AUDIT_FRAMEWORK). disclaimer : str The honest-scope disclaimer (:data:AUDIT_DISCLAIMER). signature : str | None HMAC-SHA256 over :attr:content_hash, or None for an unsigned record. It sits outside the hashed payload, so signing leaves the content hash and the serialisation of an unsigned record unchanged. signing_key_id : str | None Identifier of the key that produced :attr:signature (:func:~scpn_phase_orchestrator.runtime.audit_signing.key_id_for_secret), or None when unsigned. content_hash : str SHA-256 over the canonical payload, computed at construction.

Methods:
__post_init__ 
__post_init__() -> None

Compute the content hash from the canonical audit payload.

Source code in src/scpn_phase_orchestrator/evaluation/record.py 
107
108
109
110
111
def __post_init__(self) -> None:
    """Compute the content hash from the canonical audit payload."""
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )
to_record 
to_record() -> dict[str, object]

Return the canonical payload plus the computed content_hash.

Returns

dict[str, object] The corpus provenance, the audit verdict, the framework and disclaimer, and the content_hash sealed over them.

Source code in src/scpn_phase_orchestrator/evaluation/record.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def to_record(self) -> dict[str, object]:
    """Return the canonical payload plus the computed ``content_hash``.

    Returns
    -------
    dict[str, object]
        The corpus provenance, the audit verdict, the framework and
        disclaimer, and the ``content_hash`` sealed over them.
    """
    record = self._canonical_payload()
    record["content_hash"] = self.content_hash
    if self.signature is not None:
        record["signature"] = self.signature
        record["signing_key_id"] = self.signing_key_id
        record["signature_algorithm"] = SIGNATURE_ALGORITHM
    return record
verify 
verify() -> bool

Return whether the stored hash still matches the recorded fields.

Returns

bool True when recomputing the content hash from the current fields reproduces the stored content_hash; False after any tampering.

Source code in src/scpn_phase_orchestrator/evaluation/record.py 
140
141
142
143
144
145
146
147
148
149
def verify(self) -> bool:
    """Return whether the stored hash still matches the recorded fields.

    Returns
    -------
    bool
        ``True`` when recomputing the content hash from the current fields
        reproduces the stored ``content_hash``; ``False`` after any tampering.
    """
    return self.content_hash == canonical_record_hash(self._canonical_payload())
sign 
sign(key: str) -> AuditRecord

Return a copy carrying an HMAC-SHA256 signature over the content hash.

The signature covers :attr:content_hash — itself a hash of the whole payload — so it authenticates the record without changing what the content hash is taken over. A verifier holding the secret (or a keyring including it) can then confirm the record was sealed by a key holder, not merely left internally consistent by whoever last edited it.

Parameters

key : str The HMAC signing-key material; must be non-empty.

Returns

AuditRecord An otherwise-identical record with :attr:signature and :attr:signing_key_id populated.

Raises

ValueError If key is empty.

Source code in src/scpn_phase_orchestrator/evaluation/record.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def sign(self, key: str) -> AuditRecord:
    """Return a copy carrying an HMAC-SHA256 signature over the content hash.

    The signature covers :attr:`content_hash` — itself a hash of the whole
    payload — so it authenticates the record without changing what the content
    hash is taken over. A verifier holding the secret (or a keyring including
    it) can then confirm the record was sealed by a key holder, not merely
    left internally consistent by whoever last edited it.

    Parameters
    ----------
    key : str
        The HMAC signing-key material; must be non-empty.

    Returns
    -------
    AuditRecord
        An otherwise-identical record with :attr:`signature` and
        :attr:`signing_key_id` populated.

    Raises
    ------
    ValueError
        If ``key`` is empty.
    """
    key_id = key_id_for_secret(key)
    signature = hmac.new(
        key.encode(), self.content_hash.encode(), hashlib.sha256
    ).hexdigest()
    return AuditRecord(
        corpus_id=self.corpus_id,
        captured_at=self.captured_at,
        audit=self.audit,
        framework=self.framework,
        disclaimer=self.disclaimer,
        signature=signature,
        signing_key_id=key_id,
    )
verify_signature 
verify_signature(keys: dict[str, str]) -> bool

Return whether the signature verifies against a known key.

Parameters

keys : dict[str, str] Candidate verification keys by key id, e.g. from :func:~scpn_phase_orchestrator.runtime.audit_signing.audit_verification_keys.

Returns

bool True only when the record is signed, its key id is present in keys, and the HMAC over :attr:content_hash matches (constant-time comparison). An unsigned record, an unknown key id, or a mismatch all return False.

Source code in src/scpn_phase_orchestrator/evaluation/record.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def verify_signature(self, keys: dict[str, str]) -> bool:
    """Return whether the signature verifies against a known key.

    Parameters
    ----------
    keys : dict[str, str]
        Candidate verification keys by key id, e.g. from
        :func:`~scpn_phase_orchestrator.runtime.audit_signing.audit_verification_keys`.

    Returns
    -------
    bool
        ``True`` only when the record is signed, its key id is present in
        ``keys``, and the HMAC over :attr:`content_hash` matches (constant-time
        comparison). An unsigned record, an unknown key id, or a mismatch all
        return ``False``.
    """
    if self.signature is None or self.signing_key_id is None:
        return False
    key = keys.get(self.signing_key_id)
    if key is None:
        return False
    expected = hmac.new(
        key.encode(), self.content_hash.encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(self.signature, expected)

Functions:

seal_detector_audit 

seal_detector_audit(
    audit: DetectorAudit,
    *,
    corpus_id: str,
    captured_at: str,
    key: str | None = None,
) -> AuditRecord

Seal a detector audit into a hash-addressed record, optionally signed.

Parameters

audit : DetectorAudit The audit verdict to seal. corpus_id : str Provenance identifier of the corpus the audit ran on; must be non-empty. captured_at : str Caller-supplied timestamp of the audit; must be non-empty. key : str | None An HMAC signing key. When given, the sealed record is signed (:meth:AuditRecord.sign); when None the record is left unsigned with its bare content-hash seal.

Returns

AuditRecord The sealed, hash-addressed record — signed when key was supplied.

Raises

ValueError If corpus_id or captured_at is empty, or key is an empty string.

Source code in src/scpn_phase_orchestrator/evaluation/record.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def seal_detector_audit(
    audit: DetectorAudit,
    *,
    corpus_id: str,
    captured_at: str,
    key: str | None = None,
) -> AuditRecord:
    """Seal a detector audit into a hash-addressed record, optionally signed.

    Parameters
    ----------
    audit : DetectorAudit
        The audit verdict to seal.
    corpus_id : str
        Provenance identifier of the corpus the audit ran on; must be non-empty.
    captured_at : str
        Caller-supplied timestamp of the audit; must be non-empty.
    key : str | None
        An HMAC signing key. When given, the sealed record is signed
        (:meth:`AuditRecord.sign`); when ``None`` the record is left unsigned with
        its bare content-hash seal.

    Returns
    -------
    AuditRecord
        The sealed, hash-addressed record — signed when ``key`` was supplied.

    Raises
    ------
    ValueError
        If ``corpus_id`` or ``captured_at`` is empty, or ``key`` is an empty string.
    """
    if not corpus_id:
        raise ValueError("corpus_id must not be empty")
    if not captured_at:
        raise ValueError("captured_at must not be empty")
    record = AuditRecord(
        corpus_id=corpus_id,
        captured_at=captured_at,
        audit=audit.to_record(),
    )
    if key is None:
        return record
    return record.sign(key)

Cross-domain meta-analysis

detector_meta_analysis reads the committed detector-evidence aggregates under examples/real_data/*/, normalises each detector's performance into a common table, ranks detectors per domain and overall, and renders a Markdown ranking report with a refinement backlog.

detector_meta_analysis

Cross-domain meta-analysis of committed detector-evidence aggregates.

This module reads aggregate JSONs produced by the honest-audit and early-warning lead-time pipelines, normalises each detector's performance into a common table, ranks detectors per domain and overall, and emits a ranked refinement backlog that can be written to docs/studies/.

Supported schema families:

  • Honest-audit aggregates — top-level detector summary objects that contain mean_detection_rate, geometric_mean_p_value and fraction_beats_chance (e.g. cap_multichannel_aggregate.json, synthetic_honest_audit_demo.json).
  • Early-warning lead-time results — top-level permutation_significance mapping detectors to observed_led, n_transitions and p_value (e.g. EEG/cardiac/climate/grid early_warning_leadtime_*_results.json).

Other JSON artefacts are discovered but reported as unsupported rather than raising an error, so the tool stays safe to run as the corpus evolves.

Classes

EvidenceRow dataclass 

EvidenceRow(
    domain: str,
    detector: str,
    detection_rate: float,
    p_value: float,
    beats_chance: bool,
    source_file: str,
)

Normalised detector-performance row used for ranking.

Functions:

discover_aggregate_jsons 

discover_aggregate_jsons(root: Path) -> list[Path]

Return every aggregate JSON found directly under root/*.

A file is treated as an aggregate if it is a direct child of a domain directory and either matches a known committed-aggregate filename or ends with one of the recognised aggregate suffixes.

Parameters

root : Path Directory whose immediate subdirectories are domain folders; each is scanned one level deep for aggregate JSON files.

Returns

list of Path Sorted paths of the discovered aggregate JSONs. Empty when root is not a directory or holds no matching files.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def discover_aggregate_jsons(root: Path) -> list[Path]:
    """Return every aggregate JSON found directly under ``root/*``.

    A file is treated as an aggregate if it is a direct child of a domain
    directory and either matches a known committed-aggregate filename or ends
    with one of the recognised aggregate suffixes.

    Parameters
    ----------
    root : Path
        Directory whose immediate subdirectories are domain folders; each is
        scanned one level deep for aggregate JSON files.

    Returns
    -------
    list of Path
        Sorted paths of the discovered aggregate JSONs. Empty when ``root`` is
        not a directory or holds no matching files.
    """
    paths: list[Path] = []
    if not root.is_dir():
        return paths
    for domain_dir in sorted(root.iterdir()):
        if not domain_dir.is_dir():
            continue
        for candidate in domain_dir.iterdir():
            if not candidate.is_file() or candidate.suffix != ".json":
                continue
            if candidate.name in KNOWN_AGGREGATE_NAMES or candidate.name.endswith(
                AGGREGATE_SUFFIXES
            ):
                paths.append(candidate)
    return sorted(paths)

extract_evidence 

extract_evidence(path: Path) -> list[EvidenceRow]

Parse a single aggregate JSON into normalised evidence rows.

The schema family is inferred from the payload: a per_recording key selects the honest-audit extractor, a permutation_significance key selects the early-warning lead-time extractor.

Parameters

path : Path Aggregate JSON file to parse. Its parent directory name becomes the domain label.

Returns

list of EvidenceRow Normalised rows, one per detector. Empty when the payload matches no supported schema.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def extract_evidence(path: Path) -> list[EvidenceRow]:
    """Parse a single aggregate JSON into normalised evidence rows.

    The schema family is inferred from the payload: a ``per_recording`` key
    selects the honest-audit extractor, a ``permutation_significance`` key
    selects the early-warning lead-time extractor.

    Parameters
    ----------
    path : Path
        Aggregate JSON file to parse. Its parent directory name becomes the
        domain label.

    Returns
    -------
    list of EvidenceRow
        Normalised rows, one per detector. Empty when the payload matches no
        supported schema.
    """
    data = json.loads(path.read_text(encoding="utf-8"))
    domain = path.parent.name

    if "per_recording" in data:
        honest = _extract_honest_audit(path, data, domain)
        if honest:
            return honest

    if "permutation_significance" in data:
        return _extract_leadtime(path, data, domain)

    return []

rank_per_domain 

rank_per_domain(
    rows: list[EvidenceRow],
) -> dict[str, list[tuple[int, EvidenceRow]]]

Return detectors ranked within each domain.

Ranking uses competition ranking (1, 2, 2, 4, …) on detection rate descending, with p-value ascending as a tie-breaker and detector name as a final stable tie-breaker.

Parameters

rows : list of EvidenceRow Normalised detector rows spanning all domains.

Returns

dict Mapping of domain name to a list of (rank, EvidenceRow) pairs, ordered by rank within that domain.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def rank_per_domain(
    rows: list[EvidenceRow],
) -> dict[str, list[tuple[int, EvidenceRow]]]:
    """Return detectors ranked within each domain.

    Ranking uses competition ranking (1, 2, 2, 4, …) on detection rate
    descending, with p-value ascending as a tie-breaker and detector name as a
    final stable tie-breaker.

    Parameters
    ----------
    rows : list of EvidenceRow
        Normalised detector rows spanning all domains.

    Returns
    -------
    dict
        Mapping of domain name to a list of ``(rank, EvidenceRow)`` pairs,
        ordered by rank within that domain.
    """
    by_domain: dict[str, list[EvidenceRow]] = {}
    for row in rows:
        by_domain.setdefault(row.domain, []).append(row)

    rankings: dict[str, list[tuple[int, EvidenceRow]]] = {}
    for domain, drows in sorted(by_domain.items()):
        sorted_rows = sorted(
            drows,
            key=lambda r: (-r.detection_rate, r.p_value, r.detector),
        )
        ranked: list[tuple[int, EvidenceRow]] = []
        current_rank = 0
        previous_key: tuple[float, float] | None = None
        for position, row in enumerate(sorted_rows, start=1):
            key = (-row.detection_rate, row.p_value)
            if key != previous_key:
                current_rank = position
                previous_key = key
            ranked.append((current_rank, row))
        rankings[domain] = ranked
    return rankings

rank_overall 

rank_overall(
    rankings: dict[str, list[tuple[int, EvidenceRow]]],
) -> list[dict[str, Any]]

Aggregate per-domain ranks into an overall detector ranking.

The overall ordering prioritises lower mean rank, then more outright domain wins, then broader cross-domain presence, and finally alphabetical detector name for stability.

Parameters

rankings : dict Per-domain rankings as returned by :func:rank_per_domain.

Returns

list of dict One entry per detector with keys detector, mean_rank, domains_present, wins and domain_ranks, sorted best-first.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def rank_overall(
    rankings: dict[str, list[tuple[int, EvidenceRow]]],
) -> list[dict[str, Any]]:
    """Aggregate per-domain ranks into an overall detector ranking.

    The overall ordering prioritises lower mean rank, then more outright
    domain wins, then broader cross-domain presence, and finally alphabetical
    detector name for stability.

    Parameters
    ----------
    rankings : dict
        Per-domain rankings as returned by :func:`rank_per_domain`.

    Returns
    -------
    list of dict
        One entry per detector with keys ``detector``, ``mean_rank``,
        ``domains_present``, ``wins`` and ``domain_ranks``, sorted best-first.
    """
    detector_ranks: dict[str, list[tuple[str, int]]] = {}
    detector_wins: dict[str, int] = {}

    for domain, ranked in rankings.items():
        for rank, row in ranked:
            detector_ranks.setdefault(row.detector, []).append((domain, rank))
            if rank == 1:
                detector_wins[row.detector] = detector_wins.get(row.detector, 0) + 1

    overall: list[dict[str, Any]] = []
    for detector, ranks in detector_ranks.items():
        overall.append(
            {
                "detector": detector,
                "mean_rank": statistics.mean(rank for _, rank in ranks),
                "domains_present": len(ranks),
                "wins": detector_wins.get(detector, 0),
                "domain_ranks": ranks,
            }
        )

    overall.sort(
        key=lambda x: (
            x["mean_rank"],
            -x["wins"],
            -x["domains_present"],
            x["detector"],
        )
    )
    return overall

benjamini_hochberg_by_domain 

benjamini_hochberg_by_domain(
    rows: list[EvidenceRow],
) -> dict[tuple[str, str], float]

Return the FDR-adjusted p-value of each detector within its domain family.

The multiplicity correction is applied per domain: every detector scored on a domain forms one family, so a raw p-value that only looks significant because several detectors were tried is deflated. The raw p-value is still reported beside it; this is the honest companion, not a replacement.

Parameters

rows : list of EvidenceRow Normalised detector rows spanning all domains.

Returns

dict Mapping of (domain, detector) to its Benjamini–Hochberg-adjusted p-value, corrected within the domain.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def benjamini_hochberg_by_domain(
    rows: list[EvidenceRow],
) -> dict[tuple[str, str], float]:
    """Return the FDR-adjusted p-value of each detector within its domain family.

    The multiplicity correction is applied *per domain*: every detector scored on
    a domain forms one family, so a raw p-value that only looks significant because
    several detectors were tried is deflated. The raw p-value is still reported
    beside it; this is the honest companion, not a replacement.

    Parameters
    ----------
    rows : list of EvidenceRow
        Normalised detector rows spanning all domains.

    Returns
    -------
    dict
        Mapping of ``(domain, detector)`` to its Benjamini–Hochberg-adjusted
        p-value, corrected within the domain.
    """
    by_domain: dict[str, list[EvidenceRow]] = {}
    for row in rows:
        by_domain.setdefault(row.domain, []).append(row)
    adjusted: dict[tuple[str, str], float] = {}
    for domain, domain_rows in by_domain.items():
        family = benjamini_hochberg([row.p_value for row in domain_rows])
        for row, adjusted_p in zip(domain_rows, family, strict=True):
            adjusted[(domain, row.detector)] = adjusted_p
    return adjusted

build_report 

build_report(
    rows: list[EvidenceRow],
    rankings: dict[str, list[tuple[int, EvidenceRow]]],
    overall: list[dict[str, Any]],
    source_paths: list[Path],
    unsupported_paths: list[Path],
) -> str

Build the Markdown cross-domain detector ranking report.

Parameters

rows : list of EvidenceRow Normalised evidence rows, used to label each domain's source schema. rankings : dict Per-domain rankings from :func:rank_per_domain. overall : list of dict Overall detector ranking from :func:rank_overall. source_paths : list of Path Every aggregate JSON that was discovered. unsupported_paths : list of Path Discovered files whose schema was not recognised.

Returns

str The full report body rendered as Markdown.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
def build_report(
    rows: list[EvidenceRow],
    rankings: dict[str, list[tuple[int, EvidenceRow]]],
    overall: list[dict[str, Any]],
    source_paths: list[Path],
    unsupported_paths: list[Path],
) -> str:
    """Build the Markdown cross-domain detector ranking report.

    Parameters
    ----------
    rows : list of EvidenceRow
        Normalised evidence rows, used to label each domain's source schema.
    rankings : dict
        Per-domain rankings from :func:`rank_per_domain`.
    overall : list of dict
        Overall detector ranking from :func:`rank_overall`.
    source_paths : list of Path
        Every aggregate JSON that was discovered.
    unsupported_paths : list of Path
        Discovered files whose schema was not recognised.

    Returns
    -------
    str
        The full report body rendered as Markdown.
    """
    now = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC")
    lines: list[str] = [
        "# Cross-Domain Detector Meta-Analysis Report",
        "",
        "**Generated:** " + now,
        "",
        "This report is produced automatically from the committed detector-"
        "evidence aggregates under ``examples/real_data/*/``. It normalises "
        "each detector's performance, ranks detectors within every domain, "
        "and derives a ranked backlog of refinement candidates.",
        "",
    ]

    lines.extend(
        [
            "## Data sources",
            "",
            "| Domain | Source file | Schema |",
            "| --- | --- | --- |",
        ]
    )
    schema_names: dict[str, str] = {}
    for row in rows:
        schema_names[row.domain] = (
            "honest-audit aggregate"
            if "_demo" in row.source_file or "_aggregate" in row.source_file
            else "early-warning lead-time"
        )
    for path in source_paths:
        domain = path.parent.name
        schema = schema_names.get(domain, "unsupported / unknown")
        lines.append(f"| {domain} | `{path.name}` | {schema} |")
    lines.append("")

    if unsupported_paths:
        lines.extend(
            [
                "### Unsupported artefacts",
                "",
                "The following files were discovered but do not match a known "
                "aggregate schema, so they were not included in the ranking:",
                "",
            ]
        )
        for path in unsupported_paths:
            lines.append(f"* `{path.parent.name}/{path.name}`")
        lines.append("")

    adjusted_p = benjamini_hochberg_by_domain(rows)
    lines.extend(
        [
            "## Per-domain rankings",
            "",
            "`BH-adj p` is the Benjamini–Hochberg false-discovery-rate-adjusted "
            "p-value, corrected across the detectors compared on that domain, so a "
            "raw p-value that only looks significant because several detectors were "
            "tried is not read as a discovery. The raw `p-value` is kept beside it.",
            "",
        ]
    )
    for domain in sorted(rankings):
        lines.extend(
            [
                f"### {domain}",
                "",
                "| Rank | Detector | Detection rate | p-value | BH-adj p "
                "| Beats chance |",
                "| --- | --- | --- | --- | --- | --- |",
            ]
        )
        for rank, row in rankings[domain]:
            rate_str = _format_fraction(row.detection_rate)
            p_str = _format_p(row.p_value)
            adj_str = _format_p(adjusted_p[(domain, row.detector)])
            lines.append(
                f"| {rank} | `{row.detector}` | {rate_str} | {p_str} | {adj_str} | "
                f"{row.beats_chance} |"
            )
        lines.append("")

    lines.extend(
        [
            "## Cross-domain overall ranking",
            "",
            "| Rank | Detector | Mean rank | Domains present | Wins | Domain wins |",
            "| --- | --- | --- | --- | --- | --- |",
        ]
    )
    overall_rank = 0
    previous_sort_key: tuple[float, int, int] | None = None
    for position, entry in enumerate(overall, start=1):
        sort_key = (entry["mean_rank"], -entry["wins"], -entry["domains_present"])
        if sort_key != previous_sort_key:
            overall_rank = position
            previous_sort_key = sort_key
        win_list = (
            ", ".join(
                f"{domain} ({rank})"
                for domain, rank in entry["domain_ranks"]
                if rank == 1
            )
            or "—"
        )
        lines.append(
            f"| {overall_rank} | `{entry['detector']}` | "
            f"{entry['mean_rank']:.2f} | {entry['domains_present']} | "
            f"{entry['wins']} | {win_list} |"
        )
    lines.append("")

    lines.extend(["## Cross-domain patterns", ""])
    multi_domain = [e for e in overall if e["domains_present"] > 1]
    if multi_domain:
        lines.append(
            "Detectors that appear in more than one domain, sorted by mean rank:"
        )
        lines.append("")
        for entry in multi_domain:
            rank_summary = ", ".join(
                f"{domain} ({rank})" for domain, rank in entry["domain_ranks"]
            )
            lines.append(
                f"* **`{entry['detector']}`** — mean rank "
                f"{entry['mean_rank']:.2f}, present in "
                f"{entry['domains_present']} domain(s), wins "
                f"{entry['wins']}: {rank_summary}."
            )
    else:
        lines.append("No detector currently appears in more than one domain.")
    lines.append("")

    lines.extend(["## Ranked refinement backlog", ""])
    backlog = _build_backlog(overall, rankings)
    for idx, item in enumerate(backlog, start=1):
        lines.append(f"{idx}. {item}")
    lines.append("")

    lines.extend(
        [
            "## Notes",
            "",
            "* Detection rate for early-warning aggregates is approximated by "
            "``observed_led / n_transitions`` — the fraction of transitions "
            "for which the detector produced a statistically meaningful lead.",
            "* A detector is marked as *beating chance* when its reported "
            "p-value is below 0.05; honest-audit aggregates additionally "
            "report the committed ``fraction_beats_chance`` value.",
            "* The CAP multichannel finding that **SNR-weighted Kuramoto did "
            "not improve** over the simple mean-R Kuramoto detector is "
            "carried forward explicitly; further investment in that exact "
            "spatial-R feature is not supported by the current evidence.",
            "",
        ]
    )

    return "\n".join(lines)

run_analysis 

run_analysis(
    root: Path,
) -> tuple[
    list[EvidenceRow],
    dict[str, list[tuple[int, EvidenceRow]]],
    list[dict[str, Any]],
    list[Path],
    list[Path],
]

Discover, extract, rank and return all meta-analysis artefacts.

Parameters

root : Path Directory containing the per-domain aggregate JSON subdirectories.

Returns

tuple (rows, rankings, overall, source_paths, unsupported) — the normalised rows, per-domain rankings, overall ranking, every discovered source path, and the subset whose schema was unsupported.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
def run_analysis(
    root: Path,
) -> tuple[
    list[EvidenceRow],
    dict[str, list[tuple[int, EvidenceRow]]],
    list[dict[str, Any]],
    list[Path],
    list[Path],
]:
    """Discover, extract, rank and return all meta-analysis artefacts.

    Parameters
    ----------
    root : Path
        Directory containing the per-domain aggregate JSON subdirectories.

    Returns
    -------
    tuple
        ``(rows, rankings, overall, source_paths, unsupported)`` — the
        normalised rows, per-domain rankings, overall ranking, every discovered
        source path, and the subset whose schema was unsupported.
    """
    source_paths = discover_aggregate_jsons(root)
    rows: list[EvidenceRow] = []
    unsupported: list[Path] = []
    for path in source_paths:
        extracted = extract_evidence(path)
        if extracted:
            rows.extend(extracted)
        else:
            unsupported.append(path)
    rankings = rank_per_domain(rows)
    overall = rank_overall(rankings)
    return rows, rankings, overall, source_paths, unsupported

generate_report 

generate_report(root: Path) -> str

Run the full meta-analysis and return the Markdown report body.

Parameters

root : Path Directory containing the per-domain aggregate JSON subdirectories.

Returns

str The rendered Markdown report.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
def generate_report(root: Path) -> str:
    """Run the full meta-analysis and return the Markdown report body.

    Parameters
    ----------
    root : Path
        Directory containing the per-domain aggregate JSON subdirectories.

    Returns
    -------
    str
        The rendered Markdown report.
    """
    rows, rankings, overall, source_paths, unsupported = run_analysis(root)
    return build_report(rows, rankings, overall, source_paths, unsupported)

main 

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

Command-line entry point for the cross-domain meta-analysis tool.

Parameters

argv : list of str or None Command-line arguments excluding the program name. When None, the arguments are read from :data:sys.argv.

Returns

int Process exit code: 0 on success, 1 when the root directory does not exist.

Source code in src/scpn_phase_orchestrator/evaluation/detector_meta_analysis.py 
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
def main(argv: list[str] | None = None) -> int:
    """Command-line entry point for the cross-domain meta-analysis tool.

    Parameters
    ----------
    argv : list of str or None
        Command-line arguments excluding the program name. When ``None``, the
        arguments are read from :data:`sys.argv`.

    Returns
    -------
    int
        Process exit code: ``0`` on success, ``1`` when the root directory does
        not exist.
    """
    parser = argparse.ArgumentParser(
        description="Cross-domain detector ranking and refinement backlog."
    )
    parser.add_argument(
        "--root",
        type=Path,
        default=Path("examples/real_data"),
        help="Directory containing domain subdirectories of aggregate JSONs.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("docs/studies/detector_ranking_report.md"),
        help="Path where the Markdown report will be written.",
    )
    args = parser.parse_args(argv)

    if not args.root.is_dir():
        print(
            f"ERROR: root directory not found: {args.root}",
            file=__import__("sys").stderr,
        )
        return 1

    report = generate_report(args.root)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(report, encoding="utf-8")
    print(f"Wrote {args.output}")
    return 0

Cross-domain transfer audit

cross_domain_transfer asks the harder question the meta-analysis cannot: does a detector trained on one domain carry real skill when applied to another? audit_cross_domain_transfer reuses audit_detector for three arms at the same matched false alarm — the transferred detector on the target, a target-trained detector on the target (the detectability ceiling), and a shuffled-source floor — and returns one of transfer_positive, transfer_null, or transfer_negative. A transfer that beats its own null but not the scrambled floor is reported transfer_null, never upgraded; a target detectable within-domain whose transfer arm shows no skill is a decisive transfer_negative — the honest verdict the recorded CHB-MIT cross-subject null (AUC ≈ 0.50) demands.

cross_domain_transfer

Audit whether a detector trained on one domain transfers to another, honestly.

A detector that scores well on a new domain is easy to over-read: the apparent skill may be the new domain's own structure (any competent detector would find it), or it may be a scoring-pipeline artefact rather than genuine transfer of what the source domain taught. Reporting a bare cross-domain detection rate as "transfer" is exactly the overclaim this module refuses to make — the recorded CHB-MIT cross-subject result (AUC ≈ 0.50, no transfer) is what an honest harness must be able to return.

:func:audit_cross_domain_transfer therefore judges transfer against two orthogonal null models, reusing the detector-agnostic :func:~scpn_phase_orchestrator.evaluation.auditor.audit_detector for every arm so the calibration is identical across them:

  • Transfer arm — the source-trained detector's scores on the target domain's event and null segments. Its label-permutation p-value asks whether it separates the target's events from its nulls more than the matched false alarm explains.
  • Within-domain arm — a detector trained on the target scoring the same target segments. This is the ceiling: how detectable the target is at all. If the target is undetectable even within-domain, a low transfer number cannot be held against transfer — the question is untestable, not answered.
  • Shuffled-source floor — an ensemble of controls in which the source signal is scrambled before scoring the target. Their skill margins form the null a genuine transfer must rise above, ranked by the one-sided :func:~scpn_phase_orchestrator.evaluation.skill.surrogate_rank_pvalue.

The verdict (:data:TRANSFER_POSITIVE, :data:TRANSFER_NULL, :data:TRANSFER_NEGATIVE) is decided by :func:classify_transfer_verdict from three booleans — transfer beats its own null, transfer rises above the shuffled floor, and the target is detectable within-domain. A transfer that beats its own null but not the scrambled floor is reported transfer_null (indistinguishable from a pipeline artefact), never upgraded. Crucially, a target that is detectable within-domain yet whose transfer arm does not beat its own null is a decisive transfer_negative — the honest failure the CHB-MIT case demands.

Notes

The shuffled-source floor gate can only be significant at alpha when the control ensemble is large enough that 1 / (1 + n_controls) < alpha — the add-one-corrected rank p-value floors there. With too few controls the best a transfer can earn is transfer_null, never transfer_positive: the harness fails toward caution, never toward an unsupported positive claim.

Classes

ScorePair dataclass 

ScorePair(
    event_scores: tuple[float, ...],
    null_scores: tuple[float, ...],
)

A detector's per-segment scores on one domain's event and null segments.

Attributes

event_scores : tuple[float, ...] One score per genuine pre-transition event segment (higher = more evidence of a transition). null_scores : tuple[float, ...] One score per transition-free null segment.

Any finite-float sequence is accepted at construction and normalised to a tuple, so the pair is hashable and its serialisation is deterministic.

CrossDomainTransferAudit dataclass 

CrossDomainTransferAudit(
    source_domain: str,
    target_domain: str,
    verdict: str,
    transfer: DetectorAudit,
    within_domain: DetectorAudit,
    transfer_margin: float,
    within_domain_margin: float,
    floor_margin_mean: float,
    floor_pvalue: float,
    n_shuffled_controls: int,
    alpha: float,
)

The honest verdict on whether a source detector transfers to a target.

Attributes

source_domain : str Label of the domain the transferred detector was trained on. target_domain : str Label of the domain it is being transferred to. verdict : str One of :data:TRANSFER_POSITIVE, :data:TRANSFER_NULL, :data:TRANSFER_NEGATIVE. transfer : DetectorAudit The audit of the source detector on the target's segments. within_domain : DetectorAudit The audit of a target-trained detector on the target's segments — the detectability ceiling. transfer_margin : float The transfer arm's detection rate above its achieved false alarm. within_domain_margin : float The within-domain arm's detection rate above its achieved false alarm. floor_margin_mean : float Mean skill margin of the shuffled-source control ensemble. floor_pvalue : float One-sided rank p-value of transfer_margin against the shuffled floor. n_shuffled_controls : int Number of shuffled-source controls in the floor ensemble. alpha : float Significance level at which each gate was decided.

Attributes
p_value property 
p_value: float

The transfer arm's permutation p-value — the audit's headline number.

transferred property 
transferred: bool

Whether the verdict is a positive transfer — a convenience flag.

Methods:
to_record 
to_record() -> dict[str, object]

Return a JSON-safe mapping of the transfer verdict.

Returns

dict[str, object] The source and target labels, the verdict, both nested audit records, the transfer and within-domain skill margins, the shuffled-source floor summary, and the significance level.

Source code in src/scpn_phase_orchestrator/evaluation/cross_domain_transfer.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def to_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the transfer verdict.

    Returns
    -------
    dict[str, object]
        The source and target labels, the verdict, both nested audit records,
        the transfer and within-domain skill margins, the shuffled-source floor
        summary, and the significance level.
    """
    return {
        "schema": "scpn_cross_domain_transfer_audit_v1",
        "source_domain": self.source_domain,
        "target_domain": self.target_domain,
        "verdict": self.verdict,
        "transfer": self.transfer.to_record(),
        "within_domain": self.within_domain.to_record(),
        "transfer_margin": self.transfer_margin,
        "within_domain_margin": self.within_domain_margin,
        "floor_margin_mean": self.floor_margin_mean,
        "floor_pvalue": self.floor_pvalue,
        "n_shuffled_controls": self.n_shuffled_controls,
        "alpha": self.alpha,
    }

Functions:

classify_transfer_verdict 

classify_transfer_verdict(
    *,
    transfer_beats_own_null: bool,
    transfer_above_floor: bool,
    within_domain_detectable: bool,
) -> str

Classify a cross-domain transfer verdict from three honesty gates.

Parameters

transfer_beats_own_null : bool Whether the transfer arm's label-permutation p-value is below alpha. transfer_above_floor : bool Whether the transfer skill margin rises above the shuffled-source floor. within_domain_detectable : bool Whether a target-trained detector beats its own null on the target — i.e. the target is detectable at all.

Returns

str :data:TRANSFER_POSITIVE when transfer beats its own null and rises above the floor; :data:TRANSFER_NULL when it beats its own null but not the floor (a possible pipeline artefact) or the target is undetectable within-domain (untestable); :data:TRANSFER_NEGATIVE when the target is detectable within-domain yet transfer does not beat its own null.

Source code in src/scpn_phase_orchestrator/evaluation/cross_domain_transfer.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def classify_transfer_verdict(
    *,
    transfer_beats_own_null: bool,
    transfer_above_floor: bool,
    within_domain_detectable: bool,
) -> str:
    """Classify a cross-domain transfer verdict from three honesty gates.

    Parameters
    ----------
    transfer_beats_own_null : bool
        Whether the transfer arm's label-permutation p-value is below ``alpha``.
    transfer_above_floor : bool
        Whether the transfer skill margin rises above the shuffled-source floor.
    within_domain_detectable : bool
        Whether a target-trained detector beats its own null on the target — i.e.
        the target is detectable at all.

    Returns
    -------
    str
        :data:`TRANSFER_POSITIVE` when transfer beats its own null and rises above
        the floor; :data:`TRANSFER_NULL` when it beats its own null but not the
        floor (a possible pipeline artefact) or the target is undetectable
        within-domain (untestable); :data:`TRANSFER_NEGATIVE` when the target is
        detectable within-domain yet transfer does not beat its own null.
    """
    if transfer_beats_own_null and transfer_above_floor:
        return TRANSFER_POSITIVE
    if transfer_beats_own_null:
        # Beats its own exchangeability null but not the scrambled-source floor:
        # the apparent skill is not distinguishable from a pipeline artefact.
        return TRANSFER_NULL
    if not within_domain_detectable:
        # Neither transfer nor a within-domain detector separates the target's
        # events from its nulls — transfer is untestable, not refuted.
        return TRANSFER_NULL
    # The target is detectable within-domain, yet the transferred detector does
    # not beat its own null — a genuine, decisive failure to transfer.
    return TRANSFER_NEGATIVE

audit_cross_domain_transfer 

audit_cross_domain_transfer(
    *,
    transfer: ScorePair,
    within_domain: ScorePair,
    shuffled_source: Sequence[ScorePair],
    source_domain: str = "source",
    target_domain: str = "target",
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> CrossDomainTransferAudit

Audit whether a source-trained detector transfers to a target domain.

Every arm is scored through :func:~scpn_phase_orchestrator.evaluation.auditor.audit_detector at the same matched false alarm, so the transfer, within-domain, and shuffled-source controls are calibrated identically. The verdict is decided by :func:classify_transfer_verdict from the transfer arm's own-null p-value, the rank p-value of the transfer margin against the shuffled-source floor, and the within-domain arm's detectability.

Parameters

transfer : ScorePair The source-trained detector's scores on the target's event and null segments. within_domain : ScorePair A target-trained detector's scores on the same target segments — the detectability ceiling. shuffled_source : sequence of ScorePair Controls in which the source signal is scrambled before scoring the target; their skill margins form the floor a genuine transfer must beat. At least one is required, but 1 / (1 + len) < alpha is needed for a positive verdict to be reachable. source_domain, target_domain : str Labels carried into the verdict record. target_false_alarm : float The false-alarm rate every arm's threshold is calibrated to hold. n_permutations : int Random relabellings drawn for each arm's label-permutation p-value. seed : int Seed of the permutation resampling, so the verdict is reproducible. alpha : float Significance level at which each gate is decided.

Returns

CrossDomainTransferAudit The transfer verdict with both audit arms and the shuffled-source floor.

Raises

ValueError If shuffled_source is empty.

Source code in src/scpn_phase_orchestrator/evaluation/cross_domain_transfer.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def audit_cross_domain_transfer(
    *,
    transfer: ScorePair,
    within_domain: ScorePair,
    shuffled_source: Sequence[ScorePair],
    source_domain: str = "source",
    target_domain: str = "target",
    target_false_alarm: float = DEFAULT_TARGET_FALSE_ALARM,
    n_permutations: int = DEFAULT_PERMUTATIONS,
    seed: int = DEFAULT_PERMUTATION_SEED,
    alpha: float = DEFAULT_ALPHA,
) -> CrossDomainTransferAudit:
    """Audit whether a source-trained detector transfers to a target domain.

    Every arm is scored through
    :func:`~scpn_phase_orchestrator.evaluation.auditor.audit_detector` at the same
    matched false alarm, so the transfer, within-domain, and shuffled-source
    controls are calibrated identically. The verdict is decided by
    :func:`classify_transfer_verdict` from the transfer arm's own-null p-value, the
    rank p-value of the transfer margin against the shuffled-source floor, and the
    within-domain arm's detectability.

    Parameters
    ----------
    transfer : ScorePair
        The source-trained detector's scores on the target's event and null
        segments.
    within_domain : ScorePair
        A target-trained detector's scores on the same target segments — the
        detectability ceiling.
    shuffled_source : sequence of ScorePair
        Controls in which the source signal is scrambled before scoring the target;
        their skill margins form the floor a genuine transfer must beat. At least
        one is required, but ``1 / (1 + len) < alpha`` is needed for a positive
        verdict to be reachable.
    source_domain, target_domain : str
        Labels carried into the verdict record.
    target_false_alarm : float
        The false-alarm rate every arm's threshold is calibrated to hold.
    n_permutations : int
        Random relabellings drawn for each arm's label-permutation p-value.
    seed : int
        Seed of the permutation resampling, so the verdict is reproducible.
    alpha : float
        Significance level at which each gate is decided.

    Returns
    -------
    CrossDomainTransferAudit
        The transfer verdict with both audit arms and the shuffled-source floor.

    Raises
    ------
    ValueError
        If ``shuffled_source`` is empty.
    """
    if len(shuffled_source) == 0:
        raise ValueError("shuffled_source must provide at least one control")
    transfer_audit = audit_detector(
        event_scores=transfer.event_scores,
        null_scores=transfer.null_scores,
        detector_name=f"{source_domain}->{target_domain}",
        target_false_alarm=target_false_alarm,
        n_permutations=n_permutations,
        seed=seed,
        alpha=alpha,
    )
    within_audit = audit_detector(
        event_scores=within_domain.event_scores,
        null_scores=within_domain.null_scores,
        detector_name=f"{target_domain}(within)",
        target_false_alarm=target_false_alarm,
        n_permutations=n_permutations,
        seed=seed,
        alpha=alpha,
    )
    floor_margins = [
        _skill_margin(
            audit_detector(
                event_scores=control.event_scores,
                null_scores=control.null_scores,
                detector_name=f"{source_domain}(shuffled)",
                target_false_alarm=target_false_alarm,
                n_permutations=n_permutations,
                seed=seed,
                alpha=alpha,
            )
        )
        for control in shuffled_source
    ]
    transfer_margin = _skill_margin(transfer_audit)
    within_margin = _skill_margin(within_audit)
    floor_pvalue = surrogate_rank_pvalue(transfer_margin, floor_margins)
    verdict = classify_transfer_verdict(
        transfer_beats_own_null=transfer_audit.p_value < alpha,
        transfer_above_floor=floor_pvalue < alpha,
        within_domain_detectable=within_audit.p_value < alpha,
    )
    return CrossDomainTransferAudit(
        source_domain=source_domain,
        target_domain=target_domain,
        verdict=verdict,
        transfer=transfer_audit,
        within_domain=within_audit,
        transfer_margin=transfer_margin,
        within_domain_margin=within_margin,
        floor_margin_mean=sum(floor_margins) / len(floor_margins),
        floor_pvalue=floor_pvalue,
        n_shuffled_controls=len(floor_margins),
        alpha=alpha,
    )