Python: NimbusSTS | Julia: Not currently available Mathematical Model: State-Space Model with Extended Kalman Filter (EKF)Bayesian STS is a stateful Bayesian classification model designed for non-stationary BCI data. It combines feature-based classification with latent state dynamics, allowing it to adapt to temporal drift, electrode changes, and long-session fatigue.
Available in Python SDK:
Python SDK: NimbusSTS class (sklearn-compatible with state management)
Julia SDK: Not currently available
This is the only model in the SDK that explicitly handles temporal dynamics and non-stationary distributions.
Bayesian STS extends beyond traditional static classifiers by modeling latent state evolution over time:
✅ Temporal state dynamics with Extended Kalman Filter
✅ Drift adaptation for non-stationary data
✅ State management API for explicit time propagation
✅ Online learning with delayed feedback support
✅ Cross-session transfer with state persistence
✅ Uncertainty quantification for predictions and states
✅ Fast inference (~20-30ms per trial)
from nimbus_bci import NimbusSTSimport numpy as np# Create and fit classifierclf = NimbusSTS(transition_cov=0.01, num_steps=100)clf.fit(X_train, y_train)# Stateful prediction with time propagationfor x_new in streaming_data: clf.propagate_state() # Advance time prediction = clf.predict(x_new.reshape(1, -1)) # After feedback arrives clf.partial_fit(x_new.reshape(1, -1), [true_label])# Inspect and save state for next sessionz_mean, z_cov = clf.get_latent_state()
Bayesian STS implements a state-space model with Extended Kalman Filter inference:Latent Dynamics:
z_t = A @ z_{t-1} + w_t, where w_t ~ N(0, Q)
Observation Model:
logits = W @ x_t + H @ z_t + bp(y=k | x_t, z_t) = softmax(logits)_k
Where:
z_t = latent state at time t (captures temporal patterns like class prior drift)
A = state transition matrix (default: identity for random walk)
Q = process noise covariance (controls drift speed)
W = feature weight matrix
H = state-to-logit projection matrix
x_t = observed features
Key Innovation: The latent state z captures temporal patterns that persist across samples, such as gradual shifts in class priors due to fatigue or electrode drift.
from nimbus_bci import NimbusSTSimport numpy as np# Generate sample data (in practice, use real EEG features)np.random.seed(42)n_samples, n_features = 100, 16X_train = np.random.randn(n_samples, n_features)y_train = np.random.randint(0, 4, n_samples)# Train with moderate drift trackingclf = NimbusSTS( transition_cov=0.05, # Moderate drift num_steps=100, verbose=True)clf.fit(X_train, y_train)# Standard prediction (each sample treated independently)predictions = clf.predict(X_test)probabilities = clf.predict_proba(X_test)print(f"Predictions: {predictions}")print(f"Probabilities shape: {probabilities.shape}")
Important: predict() and predict_proba() never mutate the state (sklearn API compatibility). For time-ordered evaluation, use propagate_state() explicitly.
For time-ordered streaming data, explicitly propagate state between samples:
Python
from nimbus_bci import NimbusSTSimport numpy as np# Train modelclf = NimbusSTS(transition_cov=0.05)clf.fit(X_train, y_train)# Streaming prediction with state propagationpredictions = []for x_t in X_stream: # Time-ordered samples # Advance state one step (EKF time update) clf.propagate_state(n_steps=1) # Predict using current state pred = clf.predict(x_t.reshape(1, -1))[0] predictions.append(pred) print(f"Prediction: {pred}")# Compare with vs without state propagationacc_with_state = np.mean(predictions == y_stream)acc_without_state = clf.score(X_stream, y_stream) # Treats samples independentlyprint(f"Accuracy with state propagation: {acc_with_state:.2%}")print(f"Accuracy without state propagation: {acc_without_state:.2%}")
The canonical BCI paradigm: predict → user acts → receive feedback → update
Python
from nimbus_bci import NimbusSTS# Initial training (calibration phase)clf = NimbusSTS(transition_cov=0.05, learning_rate=0.1)clf.fit(X_calibration, y_calibration)# Online session with delayed feedbackfor trial in online_session: # 1. Advance time (no measurement) clf.propagate_state() # 2. Predict using current state prediction = clf.predict(trial.features.reshape(1, -1))[0] # 3. User performs action based on prediction execute_action(prediction) # 4. Get feedback (true label) after user completes action true_label = wait_for_feedback() # 5. Update state with measurement (EKF update) clf.partial_fit(trial.features.reshape(1, -1), [true_label]) print(f"Trial: pred={prediction}, true={true_label}")
Why this matters: In real BCI, labels aren’t available at prediction time. The model must predict using only the prior, then update when feedback arrives.
from nimbus_bci import NimbusSTSimport pickle# Day 1: Train and save stateclf_day1 = NimbusSTS()clf_day1.fit(X_day1, y_day1)# Get final statez_final, P_final = clf_day1.get_latent_state()print(f"Final state mean: {z_final}")print(f"Final state uncertainty: {np.diag(P_final)}")# Save model and statewith open("model_day1.pkl", "wb") as f: pickle.dump({'model': clf_day1, 'state': (z_final, P_final)}, f)# Day 2: Transfer state with increased uncertaintywith open("model_day1.pkl", "rb") as f: saved = pickle.load(f)clf_day2 = saved['model']# Quick calibration (just a few trials)clf_day2.fit(X_day2_calib, y_day2_calib)# Transfer previous state (with decay)z_prior, P_prior = saved['state']clf_day2.set_latent_state( z_mean=z_prior * 0.5, # Decay mean toward 0 z_cov=P_prior * 2.0 # Increase uncertainty)# Use model with transferred statepredictions = clf_day2.predict(X_day2_test)
Use case: Cross-day transfer learning. Start with informed prior from previous session, but increase uncertainty to allow adaptation.
from nimbus_bci import NimbusSTSclf = NimbusSTS()clf.fit(X_train, y_train)# Run one sessionfor _ in range(100): clf.propagate_state() # ... predictions ...# Reset for new session (clean slate)clf.reset_state()# State is now back to initial valuesz_reset, P_reset = clf.get_latent_state()print(f"Reset state mean: {z_reset}") # Near zeroprint(f"Reset state cov diagonal: {np.diag(P_reset)}") # Identity
Pro Tip: Start with transition_cov=0.01 and num_steps=100. If you observe drift (accuracy degrades over time), increase transition_cov. If predictions are too noisy, decrease it.
Key Insight: NimbusSTS shines on non-stationary data where static models degrade over time. For short, stationary sessions, use NimbusLDA for faster inference.
✅ Handles Non-Stationarity: Explicitly models temporal drift
✅ Adaptive: Continuously learns from feedback
✅ Cross-Session Transfer: State persistence across days
✅ Uncertainty Quantification: For both predictions and states
✅ Delayed Feedback Support: Natural for BCI paradigms
✅ Production-Ready: Real-time capable with <30ms latency
✅ sklearn-Compatible: Works with pipelines and CV
❌ More Complex API: State management requires careful usage
❌ Slightly Slower: 5-10ms overhead vs static models
❌ Requires More Tuning: transition_cov is critical
❌ Not Ideal for Stationary Data: Use NimbusLDA if data is stable
❌ Memory: Maintains state history (minimal overhead)
Is your data non-stationary (drift over time)?├─ Yes → How long are your sessions?│ ├─ >30 min → NimbusSTS (essential)│ └─ <30 min → Try NimbusSTS, compare with static models└─ No → Do classes have different covariances? ├─ Yes → NimbusQDA └─ No → NimbusLDA (fastest)
Rule of thumb: If accuracy degrades >10% from start to end of session, use NimbusSTS.
