flowex

Real-time multi-exchange crypto market data for Python — powered by Go via cgo shared library. WebSocket streaming, order book metrics, technical indicators, and per-symbol actor workers for Binance, Bybit, and Bitget.

No subprocess, no gRPC, no serialization overhead on the hot path. The Go runtime lives inside the Python process. Every symbol gets its own goroutine. get_snapshot() hits the atomic store directly.

Install

pip install flowex

Quick start

import time
from flowex import Manager

mgr = Manager("binance")
mgr.subscribe(["BTCUSDT", "ETHUSDT", "SOLUSDT"])
time.sleep(2)

snap = mgr.get_snapshot("BTCUSDT")
print(snap.depth.spread_bps)
print(snap.depth.mid_price)
print(snap.candles[-1].close)

# Technical indicators (RSI, MACD, Bollinger Bands, etc.)
if snap.indicators:
    print(snap.indicators.rsi_14)
    print(snap.indicators.macd_line)
    print(snap.indicators.bb_upper)

mgr.close()

Or use as a context manager:

with Manager("binance") as mgr:
    mgr.subscribe(["BTCUSDT"])
    time.sleep(2)
    snap = mgr.get_snapshot("BTCUSDT")

Supported exchanges

• Binance — futures WebSocket (candles, depth, aggTrade)
• Bybit — linear futures WebSocket
• Bitget — USDT futures WebSocket

API

Manager(exchange)

Create a manager for an exchange. Initializes the Go runtime and WebSocket connections.

mgr = Manager("binance")  # or "bybit", "bitget"

mgr.subscribe(symbols)

Subscribe to all streams (candle, depth, trade) for one or more symbols.

mgr.subscribe("BTCUSDT")
mgr.subscribe(["BTCUSDT", "ETHUSDT", "SOLUSDT"])

mgr.subscribe_candle(symbol) / subscribe_depth(symbol) / subscribe_trade(symbol)

Subscribe to individual stream types for granular control.

mgr.subscribe_candle("BTCUSDT")   # candles only
mgr.subscribe_depth("ETHUSDT")    # depth only
mgr.subscribe_trade("SOLUSDT")    # trades only

mgr.get_snapshot(symbol) -> Snapshot | None

Get the latest point-in-time snapshot for a symbol.

snap = mgr.get_snapshot("BTCUSDT")
snap.timestamp      # Unix ms
snap.depth          # DepthMetrics
snap.candles        # list[CandleHLCV]
snap.trades         # list[NormalizedTrade]
snap.indicators     # TechnicalIndicators | None

mgr.get_all_snapshots() -> dict[str, Snapshot]

Get snapshots for all subscribed symbols in one call.

mgr.get_status() -> dict

Get manager status info (subscribed symbols, connection state).

mgr.get_depth_history(symbol, count=0) -> list[DepthMetrics]

Get recent depth metrics from the ring buffer. count=0 returns the full buffer (default 100 entries).

history = mgr.get_depth_history("BTCUSDT", count=10)
for m in history:
    print(m.mid_price, m.spread_bps)

mgr.get_depth_by_time_range(symbol, start_ms, end_ms) -> list[DepthMetrics]

Get depth metrics within a Unix-millisecond time window.

import time
now = int(time.time() * 1000)
metrics = mgr.get_depth_by_time_range("BTCUSDT", now - 30_000, now)

mgr.unsubscribe(symbol)

Remove all streams for a symbol.

mgr.unsubscribe_stream(symbol, stream)

Remove a specific stream type: "candle", "depth", or "trade".

mgr.unsubscribe_stream("BTCUSDT", "depth")

mgr.close()

Shutdown all managers and WebSocket connections.

Data models

TechnicalIndicators

Computed from candle history (needs ~20+ bars):

DepthMetrics

Order book metrics computed from raw depth data:

Plus 70+ more fields: liquidity at 5/10/20/50 levels, volumes, imbalance deltas, largest walls, slippage at 8 USD sizes, velocity, momentum, z-scores, and depth quality metrics.

CandleHLCV

candle.ts       # Unix ms
candle.open
candle.high
candle.low
candle.close
candle.volume

NormalizedTrade

trade.timestamp  # Unix ms
trade.price
trade.size
trade.side       # "buy" or "sell"
trade.symbol

Architecture

Python process
|
+-- ctypes loads libflowex.so  <--  Go runtime starts inside Python
|                                    |
|                                    +-- Manager (binance)
|                                    |     +-- Worker: BTCUSDT -> atomic snapshot
|                                    |     +-- Worker: ETHUSDT -> atomic snapshot
|                                    |     +-- Worker: SOLUSDT -> atomic snapshot
|                                    |
|                                    +-- Manager (bybit)
|                                          +-- Worker: BTCUSDT -> atomic snapshot
|
+-- mgr.subscribe(["BTCUSDT"])     ->  FlowexSubscribeBatch()
+-- mgr.get_snapshot("BTCUSDT")    ->  FlowexGetSnapshot() -> atomic.Load -> JSON -> Python
+-- mgr.get_all_snapshots()        ->  FlowexGetSnapshots() -> one call, all symbols

Build from source

Requires Go 1.22+ and Python 3.10+.

# macOS (Apple Silicon)
make build-mac

# Linux (amd64)
make build-linux

# All platforms (requires cross-compilers)
make build-all

License

MIT