Barra — Live Stream Streaming Pipeline, Analytics & Predictive ML

Side Project • Production

Individual Role: Author, Data Pipeline & Machine Learning Systems Architect

Barra is a production-grade live-stream streaming ingestion engine, real-time analytics web dashboard, and predictive machine learning platform. Built in Python 3.11+, it captures real-time WebSocket binary Protobuf events across concurrent rooms, persisting records to SQLite/PostgreSQL with integrated ARIMAX revenue forecasting, contributor churn models, and user lifetime value (LTV) prediction.

GitHub: opedium/Barra

Applied AI: Auto-ARIMAX Forecasting & Whale Churn ML

Protocol: WebSocket + Binary Protobuf Deserialization

Live Server Dashboard GitHub

TECHNOLOGY STACK:

Python 3.11+ Auto-ARIMAX & Time-Series ML Logistic Regression / LTV Modeling Flask Web Server (SSE Real-Time) Google Protocol Buffers (Protobuf) PostgreSQL & SQLite WebSocket Client Node.js Signature Engine Linux Systemd Daemon

Real-Time Ingestion & ML Pipeline Flow

Live Stream Gateway

WebSocket Protobuf Unpack

13-Type Message Dispatcher

Auto-ARIMAX / Churn ML

Flask SSE Live Dashboard

System Architecture & Key Engineering Features

1. Predictive Machine Learning & Time-Series Analytics (predictor.py)

Architected statistical modeling and predictive machine learning algorithms to convert raw streaming telemetry into actionable commercial intelligence.

Auto-ARIMAX Forecasting: Developed time-series forecasting modules (fit_arimax, auto_arimax) predicting revenue trajectories, viewer traffic curves, and interaction spikes with exogenous live event regressors.
Whale Contributor Modeling & Churn Risk: Trained logistic regression classifiers (predict_whale_logistic, predict_whale_churn) to identify top-tier spenders and forecast disengagement risks before churn occurs.
User LTV & Next Spend Predictions: Built lifetime value algorithms (predict_user_ltv, predict_user_next_spend) and session clustering (cluster_sessions) to evaluate broadcast performance tiers.

def auto_arimax(series: pd.Series, exog: pd.DataFrame = None, max_p: int = 3, max_q: int = 3):
    # Grid search optimal ARIMA(p,d,q) order via AIC minimization
    best_aic, best_model = float('inf'), None
    for p in range(max_p + 1):
        for q in range(max_q + 1):
            model = fit_arimax(series, order=(p, 1, q), exog=exog)
            if model and model.aic < best_aic:
                best_aic, best_model = model.aic, model
    return best_model

2. High-Concurrency Streaming Ingestion & Binary Protobuf Parsing

Engineered the core socket streaming pipeline capable of ingesting high-throughput binary frames with sub-millisecond dispatch.

Asynchronous WebSocket consumer extracting gzip-compressed binary payload frames directly from live stream gateways.
Two-stage deserialization pipeline decoding envelope frames (PushFrame / Response) and dispatching across 13 distinct message types (danmaku, gift combos, user entries, follower upgrades, lucky bags, live room metrics).
Multi-room concurrent monitoring engine running simultaneous background capture loops with dynamic room switching and status panels.

3. Multi-Tier Watchdog Daemon & Pseudo-Alive Detection

Engineered advanced fault-tolerance mechanisms to eliminate silent stream dropouts during multi-hour unattended monitoring.

Independent daemon watchdog tracking frame timestamps; automatically initiates exponential backoff reconnects if network drops occur.
Pseudo-Alive Detection: Business-level heartbeat watchdog that identifies stagnant connection states (where TCP remains connected but zero danmaku events arrive), automatically cycling network sockets.
Clean TCP/SSL socket teardown preventing orphan sockets or memory growth during long-running server deployments.

4. Full-Stack Analytics Dashboard & Server-Sent Events (SSE)

Designed and implemented a Flask web application with PostgreSQL persistence providing visual analytics for live streaming operations.

Interactive analytics views: Live session overview, top gift contributor leaderboards, user interaction profiles, and cross-session performance comparisons.
Server-Sent Events (SSE) integration (set_sse_callback) delivering zero-latency updates to active dashboard clients without polling.
Dual output pipeline supporting live database ingestion alongside streaming JSONL logs and CSV export (UTF-8 BOM for instant Excel compatibility).

5. Node.js Signature Generation & Production Automation

Bridged cross-runtime security tooling and production service automation.

Integrated Node.js signature generation subprocess (signer.py) to dynamically generate anti-bot signature hashes required for stream handshakes.
Engineered Linux systemd service units (douyin-monitor.service) ensuring 24/7 background persistence and automated recovery upon server reboots.