Temporal Smoothing

Temporal posterior smoothing can be run directly on probability observations exported by time decoding:

python -m neureptrace.temporal_smoothing \
  results/nod_animate_logistic_temporal_smoothing_all/observations/*_observations.csv \
  --out-observations results/nod_animate_logistic_temporal_smoothing_all/temporal_smoothing/smoothed_observations.csv \
  --out-metrics results/nod_animate_logistic_temporal_smoothing_all/temporal_smoothing/smoothed_metrics.csv \
  --fit-window 0.1 0.8

For benchmark comparisons, prefer neureptrace.benchmark --temporal-smoothing-dir. That runs smoothing on the exact held-out observations from the decoder run and aggregates raw plus smoothed metrics in one summary table. The companion provenance.csv records the raw and smoothed emission modes side by side with selected accuracy, log loss, and ECE, so calibration-only changes are easy to spot.

neureptrace.temporal_smoothing

metrics_from_probability_observations(observations, *, ece_bins=10)

Compute fold/time decoding metrics directly from probability observations.

This is useful after temporal smoothing because the updated posterior probabilities are the result object; averaging old per-fold CSV metrics would silently discard the temporal posterior.

Source code in src/neureptrace/temporal_smoothing.py

def metrics_from_probability_observations(observations: pd.DataFrame, *, ece_bins: int = 10) -> pd.DataFrame:
    """Compute fold/time decoding metrics directly from probability observations.

    This is useful after temporal smoothing because the updated posterior probabilities are the result object;
    averaging old per-fold CSV metrics would silently discard the temporal posterior.
    """

    prob_columns = probability_columns(observations)
    n_classes = len(prob_columns)
    label_values = _label_values(prob_columns)
    labels = _numeric_label_values(observations, label_values)
    working = observations.copy()
    working["true_label"] = labels
    for column in prob_columns:
        working[column] = pd.to_numeric(working[column], errors="coerce")
    if working[prob_columns].isna().any().any():
        raise ValueError("prob_class_* columns must be numeric and non-missing.")

    group_columns = [column for column in METRIC_GROUP_COLUMNS if column in working.columns]
    if "time" not in group_columns:
        raise ValueError("Probability observations must contain a time column.")

    rows: list[dict[str, object]] = []
    for key, group in _iter_groups(working, group_columns):
        row = _group_values(group_columns, key)
        probabilities = _normalize_probabilities(group[prob_columns].to_numpy(dtype=float))
        group_labels = group["true_label"].to_numpy(dtype=int)
        group_positions = _label_positions(group_labels, label_values)
        predictions = probabilities.argmax(axis=1)
        predicted_labels = np.asarray([label_values[position] for position in predictions], dtype=int)
        class_names = _class_names(group, prob_columns)
        row.update(
            {
                "accuracy": accuracy_score(group_labels, predicted_labels),
                "balanced_accuracy": balanced_accuracy_score(group_labels, predicted_labels),
                "top2_accuracy": _top_k_accuracy_from_label_values(probabilities, group_labels, label_values, k=2),
                "top3_accuracy": _top_k_accuracy_from_label_values(probabilities, group_labels, label_values, k=3),
                "log_loss": log_loss(group_labels, probabilities, labels=list(label_values)),
                "brier": brier_score_multiclass(probabilities, group_positions),
                "ece": expected_calibration_error(probabilities, group_positions, n_bins=ece_bins),
                "n_test": int(len(group)),
                "n_classes": int(n_classes),
                "class_names": "|".join(map(str, class_names)),
            }
        )
        for optional_column in METRIC_PROVENANCE_COLUMNS:
            if optional_column not in group.columns:
                continue
            values = group[optional_column].dropna().astype(str).unique()
            if len(values) == 1:
                row[optional_column] = values[0]
        rows.append(row)

    return pd.DataFrame(rows).sort_values(group_columns).reset_index(drop=True)

smooth_probability_observations(observation_csvs, *, fit_window=DEFAULT_FIT_WINDOW, stay_grid_size=200, mode='forward_backward', apply_window=None, emission_suffix=DEFAULT_EMISSION_SUFFIX, ece_bins=10, out_observations=None, out_metrics=None)

Replace decoder probabilities by sticky-switching temporal posteriors.

The sticky transition probability is fit without labels from held-out probability observations within fit_window for each decoder/emission group, then applied to every complete sequence in that group. forward_backward uses all times in a sequence, forward_only only uses observations up to each time, and poststimulus_forward_only applies forward filtering only inside apply_window while leaving probabilities outside that window unchanged. The returned observation table preserves the NeuRepTrace probability-observation schema but changes prob_class_* to temporally smoothed posterior probabilities and updates prediction columns.

Source code in src/neureptrace/temporal_smoothing.py

def smooth_probability_observations(
    observation_csvs: list[Path],
    *,
    fit_window: tuple[float, float] | None = DEFAULT_FIT_WINDOW,
    stay_grid_size: int = 200,
    mode: str = "forward_backward",
    apply_window: tuple[float, float] | None = None,
    emission_suffix: str = DEFAULT_EMISSION_SUFFIX,
    ece_bins: int = 10,
    out_observations: Path | None = None,
    out_metrics: Path | None = None,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Replace decoder probabilities by sticky-switching temporal posteriors.

    The sticky transition probability is fit without labels from held-out probability observations within
    ``fit_window`` for each decoder/emission group, then applied to every complete sequence in that group.
    ``forward_backward`` uses all times in a sequence, ``forward_only`` only uses observations up to each
    time, and ``poststimulus_forward_only`` applies forward filtering only inside ``apply_window`` while
    leaving probabilities outside that window unchanged.
    The returned observation table preserves the NeuRepTrace probability-observation schema but changes
    ``prob_class_*`` to temporally smoothed posterior probabilities and updates prediction columns.
    """

    if mode not in SMOOTHING_MODE_CHOICES:
        raise ValueError(f"Unknown temporal smoothing mode '{mode}'. Available modes: {', '.join(SMOOTHING_MODE_CHOICES)}.")
    if mode == "poststimulus_forward_only" and apply_window is None:
        apply_window = DEFAULT_POSTSTIMULUS_APPLY_WINDOW
    elif mode != "poststimulus_forward_only":
        apply_window = None
    smoothing_method = SMOOTHING_METHODS[mode]

    observations = read_probability_observations(observation_csvs).copy()
    observations["__input_order"] = np.arange(len(observations))
    prob_columns = probability_columns(observations)
    group_columns = _smoothing_group_columns(observations) or _model_group_columns(observations)
    smoothed_frames: list[pd.DataFrame] = []

    for _, decoder_frame in _iter_groups(observations, group_columns):
        fit_frame = _filter_time_window(decoder_frame, fit_window) if fit_window is not None else decoder_frame.copy()
        fit_sequences = _sequences_from_frame(fit_frame, prob_columns)
        fit = fit_sticky_switching_model(fit_sequences, stay_grid_size=stay_grid_size)
        stay_probability = float(fit["best_stay_probability"])
        class_names = _class_names(decoder_frame, prob_columns)
        key_columns = sequence_key_columns(decoder_frame)
        validate_unique_sequence_times(decoder_frame, key_columns)

        for _, sequence_frame in decoder_frame.sort_values([*key_columns, "time"]).groupby(key_columns, sort=True, dropna=False):
            probabilities = _normalize_probabilities(sequence_frame[prob_columns].to_numpy(dtype=float))
            if len(probabilities) < 2:
                posterior = probabilities
            else:
                posterior = _smooth_sequence_posteriors(
                    sequence_frame,
                    probabilities,
                    stay_probability=stay_probability,
                    mode=mode,
                    apply_window=apply_window,
                )
            smoothed_frames.append(
                _with_posterior_columns(
                    sequence_frame,
                    posterior,
                    prob_columns=prob_columns,
                    class_names=class_names,
                    stay_probability=stay_probability,
                    fit_window=fit_window,
                    apply_window=apply_window,
                    emission_suffix=emission_suffix,
                    smoothing_method=smoothing_method,
                )
            )

    if not smoothed_frames:
        raise ValueError("Need at least one sequence with two or more time points for temporal smoothing.")

    smoothed = pd.concat(smoothed_frames, ignore_index=True).sort_values("__input_order").drop(columns=["__input_order"]).reset_index(drop=True)
    metrics = metrics_from_probability_observations(smoothed, ece_bins=ece_bins)

    if out_observations is not None:
        out_observations.parent.mkdir(parents=True, exist_ok=True)
        smoothed.to_csv(out_observations, index=False)
    if out_metrics is not None:
        out_metrics.parent.mkdir(parents=True, exist_ok=True)
        metrics.to_csv(out_metrics, index=False)
    return smoothed, metrics