โโโโโโโโโโโโโโโโโ โโโโโโ โโโโโโโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โโโโโโโโ โโโ โโโโโโโโ โโโ โโโโโโ โโโโโโโโโโโ โโโ โโโโโโโโ
โโโโโโโโ โโโ โโโโโโโโ โโโ โโโโโโ โโโโโโโโโโโ โโโ โโโโโโโโ
โโโโโโโโ โโโ โโโ โโโ โโโ โโโโโโโโโโโโโโโโโโโ โโโ โโโโโโโโ
โโโโโโโโ โโโ โโโ โโโ โโโ โโโโโโโโโโโโโโโ โโโ โโโ โโโโโโโโ
Every state change. Every reason. Every time.
๐ฆ Install ยท โก Quick Start ยท ๐ง ML/AI ยท ๐ข Enterprise ยท ๐ Docs
Traditional state management treats changes as black boxes โ values appear, disappear, and mutate with no record of why. Statebits changes that.
Every transition carries a reason, a confidence score, a timestamp, and full context. Your state becomes a story you can read, audit, query, and predict.
from statebits import Statebit
mood = Statebit(identity="user_mood", initial_value="neutral")
mood.transition("happy", "User received good news", confidence=0.90)
mood.transition("excited", "User won a prize", confidence=0.95)
print(mood.explain()) # Full causal history, human-readable| Traditional State | Statebits |
|---|---|
x = 42 โ no context |
x โ 42 with reason, confidence, timestamp |
| Debugging is guesswork | Full transition history to replay |
| Compliance is a nightmare | Built-in audit trail with integrity checks |
| ML feels bolted on | First-class LSTM, AutoML, anomaly detection |
| One machine, one state | Redis-backed distributed state out of the box |
pip install statebits # Core
pip install statebits[ml] # + LSTM, AutoML, anomaly detection, clustering
pip install statebits[enterprise] # + RBAC, multi-tenancy, audit logging
pip install statebits[all] # EverythingFull install options
pip install statebits[async_support] # Async/await support
pip install statebits[distributed] # Redis-backed distributed state
pip install statebits[streaming] # WebSocket real-time streaming
pip install statebits[graphql] # GraphQL API via FastAPI
pip install statebits[timeseries] # InfluxDB / TimescaleDB backends
pip install statebits[encryption] # AES-256 encrypted state
pip install statebits[visualization] # Charts and graph exportfrom statebits import Statebit, ConfidenceLevel
temperature = Statebit(identity="room_temp", initial_value=20.0)
temperature.transition(22.0, "Heater turned on", confidence=ConfidenceLevel.HIGH)
temperature.transition(25.0, "Target reached", confidence=ConfidenceLevel.VERY_HIGH)
temperature.transition(24.0, "Slight cooling", confidence=0.85)
# Query history
high_confidence = temperature.get_transitions_by_confidence(0.9)
stats = temperature.get_statistics()
# Rollback if needed
temperature.rollback(steps=1)# Non-destructive scenario testing
simulation = account.fork(reason="Testing withdrawal scenario")
simulation.transition(500, "Simulated withdrawal")
# Merge back only if simulation passed
account.merge(simulation, strategy='latest')def alert_on_high_load(statebit, transition):
if transition.to_value > 0.8:
send_alert(f"โ ๏ธ High CPU: {transition.to_value:.0%}")
cpu = Statebit(identity="cpu_load", initial_value=0.2)
cpu.add_observer(alert_on_high_load)|
๐ Transparent Transitions Every change tracked with reason, confidence, timestamp, tags, and context. ๐ฆ Collections & Batch Ops Group related statebits; apply transitions across all at once. ๐ฟ Fork & Merge Test scenarios on isolated forks; merge back when satisfied. โช Rollback Undo the last N transitions with a single call. |
๐ Freeze & Lock Prevent mutations temporarily or permanently. ๐ Observers React to transitions in real-time with custom callbacks. ๐พ Serialization Export to JSON, CSV, or Pickle; restore with full fidelity. ๐ Analysis Pattern detection, trend analysis, and statistical summaries. |
from statebits import AsyncStatebit
import asyncio
async def main():
sb = AsyncStatebit(identity="price", initial_value=100.0)
await sb.transition(105.0, "Market moved up", confidence=0.88)
asyncio.run(main())from statebits import DistributedStatebit
async def main():
sb = DistributedStatebit(
identity="global_config",
redis_url="redis://localhost:6379",
initial_value={"feature_flags": {}}
)
await sb.connect()
await sb.transition({"feature_flags": {"v2": True}}, "Feature flag enabled")
await sb.disconnect()from statebits import StatebitsWebSocketServer
server = StatebitsWebSocketServer(host="0.0.0.0", port=8765)
server.register_statebit(my_statebit)
await server.start() # Stream transitions to any WebSocket clientfrom statebits import create_graphql_router
from fastapi import FastAPI
app = FastAPI()
app.include_router(create_graphql_router(collection), prefix="/graphql")from statebits import EncryptedStatebit, EncryptionManager
encryption = EncryptionManager(key_derivation_password="secret")
sensitive = EncryptedStatebit(
identity="patient_record",
initial_value={"diagnosis": "..."},
encryption_manager=encryption
)No ML dependency required to get started.
NLExplainerworks out of the box. Heavier models needpip install statebits[ml].
from statebits import NLExplainer
explainer = NLExplainer()
print(explainer.summarize_history(sb)) # Human-readable summary
print(explainer.generate_explanation(sb)) # Full causal narrativefrom statebits import SequencePredictor
predictor = SequencePredictor(model_type="lstm")
predictor.train(sb, epochs=50)
prediction = predictor.predict(sb)
print(f"Next value: {prediction.predicted_value} (confidence: {prediction.confidence:.2f})")from statebits import AnomalyDetector
detector = AnomalyDetector(contamination=0.05)
detector.fit(sb)
result = detector.detect(sb, transition_index=-1)
if result.is_anomaly:
print(f"โ ๏ธ Anomaly score: {result.anomaly_score:.4f}")from statebits import AutoMLPredictor
automl = AutoMLPredictor()
results = automl.auto_select(sb) # Tests all available models
prediction = automl.predict(sb)
print(f"Best model: {results['best_model']}")
print(f"Predicted: {prediction.predicted_value}")from statebits import RLAgent
agent = RLAgent(state_space_size=20, action_space_size=5)
for episode in range(100):
state_idx = agent.get_state_index(sb)
action = agent.choose_action(state_idx)
reward = apply_action(sb, action)
agent.update(state_idx, action, reward, agent.get_state_index(sb))
optimal = agent.get_optimal_action(sb)Full ML feature list
| Feature | Class | Description |
|---|---|---|
| NL Explanations | NLExplainer |
No ML deps needed; summarizes history in plain English |
| Sequence Prediction | SequencePredictor |
LSTM and Transformer models |
| Anomaly Detection | AnomalyDetector |
Isolation Forest |
| Clustering | StateClusterer |
K-Means and DBSCAN |
| AutoML | AutoMLPredictor |
Automatically selects and tunes the best model |
| Explainable AI | ExplainableAI |
SHAP-based feature importance |
| Reinforcement Learning | RLAgent |
Q-learning for optimal transition policies |
# Multi-tenancy
from statebits import MultiTenantManager, TenantContext
manager = MultiTenantManager()
tenant = manager.create_tenant("acme_corp")
with TenantContext(tenant.tenant_id):
sb = Statebit(identity="revenue", initial_value=0)
# Role-Based Access Control
from statebits import AccessControlManager, Permission
ac = AccessControlManager()
ac.create_user("alice", roles=["admin"])
ac.protect(sb, owner="alice")
if ac.check_permission("alice", sb, Permission.WRITE):
sb.transition(1_000_000, "Q4 revenue", confidence=1.0)
# Audit Logging
from statebits import AuditLogger
logger = AuditLogger(storage="file")
logger.attach(sb)
report = logger.generate_compliance_report(sb.identity)
# Monitoring Dashboard
from statebits import MonitoringDashboard
dashboard = MonitoringDashboard(port=8080)
dashboard.register(sb)
await dashboard.start() # Web UI at http://localhost:8080| Domain | What Statebits solves |
|---|---|
| ๐ฆ Finance | Immutable audit trails for transactions, anomaly detection for fraud, AutoML for risk scoring |
| ๐ฅ Healthcare | Patient state monitoring with full context, compliance reporting, vitals anomaly detection |
| ๐ก๏ธ IoT | Distributed device state via Redis, LSTM sensor prediction, real-time WebSocket streaming |
| ๐ค ML Ops | Model training tracking, SHAP-based prediction explanation, experiment management |
| ๐ Supply Chain | End-to-end shipment tracking with complete transparency and delay prediction |
statebits analyze state.json # Statistical analysis
statebits visualize state.json --output graph.png # Export history graph
statebits validate state.json # Integrity check
statebits export state.json --format csv # Convert format
statebits compare state1.json state2.json # Diff two statebits| Operation | Benchmark |
|---|---|
| Single transition | < 1ms |
| Query over 1,000 transitions | < 10ms |
| ML model training | 5โ20 seconds |
| Scale | Millions of transitions |
Security: AES-256 encryption ยท SHA integrity checksums ยท Thread-safe ยท Secure key management
statebits/
โโโ core.py # Statebit, StateTransition, collections
โโโ advanced.py # Learning, Temporal, Persistent, Network, Scheduler
โโโ async_support.py # AsyncStatebit
โโโ distributed.py # Redis-backed DistributedStatebit
โโโ streaming.py # WebSocket server
โโโ graphql_api.py # GraphQL via Strawberry + FastAPI
โโโ timeseries.py # InfluxDB / TimescaleDB backends
โโโ encryption.py # AES-256 EncryptedStatebit
โโโ ml_integration.py # LSTM, AutoML, anomaly, clustering, RL, SHAP
โโโ utils.py # Serializer, Analyzer, Validator, Comparator
โโโ config.py # Global configuration
โโโ performance.py # Monitoring and profiling
34 working examples ยท 100+ test cases ยท 49+ major features
Tests span core, async, distributed, streaming, GraphQL, time-series, encryption, ML, and backward compatibility.
| ๐ Homepage | https://statebits.xyz |
| ๐ Documentation | https://statebits.xyz |
| ๐ฆ PyPI | https://pypi.org/project/statebits/ |
| ๐ Issues | https://github.com/semonkhan/Statebits/issues |
| ๐ฌ Repository | https://github.com/semonkhan/Statebits |
MIT ยฉ Semon Khan
Statebits โ because every change deserves an explanation.