mne_rt.protocols.ZScoreProtocol#

class mne_rt.protocols.ZScoreProtocol(direction: str = 'up', warmup_windows: int = 20, smoothing: float = 0.0, min_std: float = 1e-06, zscore_threshold: float = 0.5)#

Bases: object

Z-score feedback protocol with rolling baseline normalisation.

Normalises each incoming NF value against the running mean and standard deviation accumulated since initialisation (or the last reset()). A reward is issued when the z-score magnitude exceeds zscore_threshold in the requested direction.

During the first warmup_windows calls to evaluate() the baseline statistics are accumulating; crossed is always False and magnitude is always 0.0 until warmup completes.

Parameters:

direction{“up”, “down”}: “up” -> reward when z-score > zscore_threshold. “down” -> reward when z-score < -zscore_threshold.
warmup_windowsint: Number of initial evaluations used solely to seed the baseline statistics before any reward can be issued. Default is 20.
smoothingfloat: EMA smoothing coefficient applied to the raw input before z-scoring. Must be in [0, 1). 0.0 disables smoothing. Applied as: smoothed = (1 - smoothing) * new + smoothing * prev. Default is 0.0.
min_stdfloat: Floor applied to the running standard deviation to prevent division by zero or near-zero blowup. Default is 1e-6.
zscore_thresholdfloat: Minimum absolute z-score required to issue a reward. Default is 0.5.

Raises:

ValueError: If any parameter is outside its valid range.

Notes

The running mean and variance are updated with Welford’s online algorithm, which is numerically stable and requires O(1) memory per update.

Examples

Reward upward alpha-power deviations beyond half a standard deviation:

proto = ZScoreProtocol(direction="up", zscore_threshold=0.5)
for value in nf_stream:
    crossed, magnitude = proto.evaluate(value)
    if crossed:
        send_reward(magnitude)

Added in version 1.0.0.

__init__(direction: str = 'up', warmup_windows: int = 20, smoothing: float = 0.0, min_std: float = 1e-06, zscore_threshold: float = 0.5) → None[source]#

Methods

`__init__`([direction, warmup_windows, ...])
`evaluate`(value)	Evaluate one NF value and return (crossed, magnitude).
`reset`()	Reset all state.

Attributes

`mean_`	Current running baseline mean.
`n_evaluated`	Total number of values evaluated since init or last reset.
`std_`	Current running baseline standard deviation.
`zscore`	Z-score computed during the most recent `evaluate()` call.

evaluate(value: float) → tuple[bool, float][source]#

Evaluate one NF value and return (crossed, magnitude).

Updates the running baseline with the (optionally smoothed) input, computes the z-score, and determines whether the reward criterion is met.

Parameters:

valuefloat: Current NF feature value.

Returns:

crossedbool: True if the z-score criterion is met in the target direction and warmup has completed.
magnitudefloat: Absolute value of the z-score when crossed is True; 0.0 otherwise.

Notes

During warmup (n_evaluated < warmup_windows) this method always returns (False, 0.0) while still accumulating baseline statistics.

property mean_: float#

Current running baseline mean.

Returns 0.0 before any evaluation.

property n_evaluated: int#: Total number of values evaluated since init or last reset.

reset() → None[source]#

Reset all state.

Clears the running baseline statistics, smoothed value, z-score, and evaluation counter. All constructor parameters are preserved.

property std_: float#

Current running baseline standard deviation.

Returns min_std before at least two evaluations.

property zscore: float#

Z-score computed during the most recent evaluate() call.

Returns 0.0 before any evaluation.