mne_rt.protocols.ThresholdProtocol#

class mne_rt.protocols.ThresholdProtocol(threshold: float = 0.0, direction: str = 'up', adaptive: bool = False, adapt_rate: float = 0.05, target_hit_rate: float = 0.7, smoothing: float = 0.0, history_len: int = 50)#

Bases: object

Threshold-based NF reward protocol with optional adaptive threshold.

Converts a continuous NF feature value into a binary reward signal. Optionally adapts the threshold over time to maintain a target success rate.

Parameters:

thresholdfloat: Initial decision threshold. Default is 0.0.
direction{“up”, “down”}: “up” -> reward when value > threshold (e.g., enhance alpha). “down” -> reward when value < threshold (e.g., suppress beta). Default is “up”.
adaptivebool: If True, slowly adjust threshold to keep hit_rate near target_hit_rate. Uses an exponential moving average of recent successes. Default is False.
adapt_ratefloat: Step size for threshold adaptation (in units of the NF signal’s running standard deviation). Larger = faster adaptation. Default is 0.05.
target_hit_ratefloat: Desired proportion of windows where the threshold is crossed. The adaptive mechanism pushes threshold toward this rate. Default is 0.70.
smoothingfloat: EMA smoothing factor for the input value before thresholding. 0.0 = no smoothing; 0.1 = light smoothing; 0.5 = heavy smoothing. Applied as: smoothed = (1 - smoothing) * new + smoothing * prev. Default is 0.0.
history_lenint: Number of recent evaluations used to estimate the running hit rate and running standard deviation (for adaptive scaling). Default is 50.

Raises:

ValueError: If any parameter is outside its valid range.

Examples

Basic usage — reward when alpha power exceeds a fixed threshold:

proto = ThresholdProtocol(threshold=0.5, direction="up")
crossed, magnitude = proto.evaluate(0.8)

Adaptive threshold that targets a 70 % success rate:

proto = ThresholdProtocol(
    threshold=0.0,
    direction="up",
    adaptive=True,
    target_hit_rate=0.70,
)
for value in nf_stream:
    crossed, magnitude = proto.evaluate(value)

__init__(threshold: float = 0.0, direction: str = 'up', adaptive: bool = False, adapt_rate: float = 0.05, target_hit_rate: float = 0.7, smoothing: float = 0.0, history_len: int = 50) → None[source]#

Methods

`__init__`([threshold, direction, adaptive, ...])
`evaluate`(value)	Evaluate one NF value and return (success, reward_magnitude).
`reset`()	Reset hit history and smoothed value, keep threshold.

Attributes

`hit_rate`	Fraction of recent windows that crossed the threshold (0–1).
`n_evaluated`	Total number of values evaluated since init or last reset.
`threshold`	Current threshold value.

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

Evaluate one NF value and return (success, reward_magnitude).

Parameters:

valuefloat: Current NF feature value.

Returns:

crossedbool: True if the threshold was crossed in the target direction.
magnitudefloat: Non-negative reward magnitude. 0 when not crossed; abs(smoothed - threshold) / (running_std + eps) when crossed.

Notes

EMA smoothing (if enabled) is applied first, then the smoothed value is compared against the current threshold. The hit/miss result is recorded in history, and (if adaptive is True) the threshold is updated before the return value is assembled.

magnitude is 0.0 when the threshold was not crossed; a positive float proportional to the distance from the threshold (normalised by the running standard deviation) when crossed.

property hit_rate: float#

Fraction of recent windows that crossed the threshold (0–1).

Returns 0.0 when no evaluations have been recorded yet.

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

reset() → None[source]#

Reset hit history and smoothed value, keep threshold.

Clears _history, _values_history, and _smoothed, and resets the evaluation counter to zero. The current threshold value is preserved.

property threshold: float#: Current threshold value.