Metadata-Version: 2.4
Name: binflow
Version: 0.1.1
Summary: Persistent WebSocket engine for Binance market data (Spot, USD-M, COIN-M)
Author: nand0san
License-Expression: MIT
Project-URL: Homepage, https://github.com/nand0san/binflow
Project-URL: Repository, https://github.com/nand0san/binflow
Project-URL: Issues, https://github.com/nand0san/binflow/issues
Keywords: binance,websocket,market-data,trading,cryptocurrency,streaming
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Financial and Insurance Industry
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Office/Business :: Financial :: Investment
Classifier: Topic :: Software Development :: Libraries
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: websockets>=13.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Dynamic: license-file

# binflow

Motor de WebSockets persistente para consumo de datos de mercado en tiempo real desde Binance.

Soporta **Spot**, **USD-M Futures** y **COIN-M Futures**. Componente de infraestructura reutilizable que mantiene un cache de mercado vivo en memoria, con reconexion automatica, observabilidad integrada y API sincrona simple.

## Filosofia

binflow es una libreria de **infraestructura de datos de mercado**. Su responsabilidad es servir datos fiables, continuos y disponibles.

**Lo que SI hace binflow:**

- Mantener conexiones WebSocket vivas y reconectar automaticamente
- Garantizar continuidad del dato (sin gaps en trade IDs)
- Servir datos normalizados, listos para consumir
- Escalar a cientos de simbolos con eficiencia (combined streams)
- Order books correctos e integros en memoria

**Lo que NO hace binflow (responsabilidad del consumidor):**

- Persistir datos en base de datos (TimescaleDB, Parquet, etc.)
- Callbacks o event-driven processing
- Metricas y monitoring (Prometheus, Grafana)
- Logica de trading o senales

Otra libreria o servicio debe consumir de binflow via polling (`get_recent_trades`, `get_order_book`) y encargarse del resto.

## Caracteristicas

- **Multi-mercado** -- Spot, USD-M Futures y COIN-M Futures con un solo parametro
- **Streaming en tiempo real** de trades y order book desde Binance
- **Streams combinados** -- multiplexa cientos de simbolos en pocas conexiones WebSocket
- **Order books profundos** -- hasta 200 niveles por lado via diff depth stream
- **Cache en memoria** thread-safe con acceso O(1)
- **Reconexion automatica** con backoff exponencial y jitter
- **API sincrona** -- se usa desde codigo Python normal, sin `async`/`await`
- **Observabilidad** -- estado de streams, metricas, deteccion de degradacion
- **Logging persistente** -- rotacion por tamano, archivo de errores separado, limites de disco
- **Shutdown limpio** -- sin tareas pendientes ni warnings
- **Context manager** -- soporte `with` para manejo automatico del ciclo de vida
- **Probado bajo carga** -- 728 simbolos simultaneos, 60s, cero gaps en trade IDs

## Instalacion

```bash
pip install -e .
```

O con dependencias de desarrollo:

```bash
pip install -e ".[dev]"
```

### Requisitos

- Python >= 3.11
- `websockets >= 13.0`

## Inicio rapido

```python
import time
from market_streaming import MarketStreamManager, StreamConfig

config = StreamConfig(
    max_recent_trades=500,
    default_order_book_depth=10,
)

with MarketStreamManager(config) as manager:
    manager.add_trade_stream("BTCUSDT")
    manager.add_order_book_stream("BTCUSDT", depth=10)

    time.sleep(5)  # esperar a que lleguen datos

    # Ultimo trade
    trade = manager.get_last_trade("BTCUSDT")
    if trade:
        print(f"{trade.price} x {trade.quantity}")

    # Order book
    book = manager.get_order_book("BTCUSDT")
    if book:
        print(f"Mejor bid: {book.bids[0].price}")
        print(f"Mejor ask: {book.asks[0].price}")

    # Salud del sistema
    print(manager.is_healthy())
```

### Streams masivos (cientos de simbolos)

Para suscribirse a muchos simbolos a la vez, usar `add_trade_streams()` o `add_order_book_streams()`. Estos metodos multiplexan automaticamente los simbolos en conexiones combinadas (hasta 200 por WebSocket), evitando el rate limit de conexiones de Binance:

```python
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT", ...]  # cientos de simbolos

with MarketStreamManager(config) as manager:
    # Registra todos los simbolos en pocas conexiones combinadas
    manager.add_trade_streams(symbols)

    time.sleep(10)

    for symbol in symbols:
        trade = manager.get_last_trade(symbol)
        if trade:
            print(f"{symbol}: {trade.price}")
```

### Order books profundos

Para acumular mas de 20 niveles, configurar `default_order_book_depth` > 20. El sistema usara automaticamente el diff depth stream de Binance:

```python
config = StreamConfig(default_order_book_depth=200)

with MarketStreamManager(config) as manager:
    manager.add_order_book_streams(["BTCUSDT", "ETHUSDT"], depth=200)
    time.sleep(60)  # acumular niveles

    book = manager.get_order_book("BTCUSDT")
    print(f"Niveles bid: {len(book.bids)}, ask: {len(book.asks)}")
```

> **Nota**: profundidades de 5, 10 y 20 usan el Partial Book Depth Stream (snapshots completos). Cualquier otro valor usa el Diff Depth Stream (actualizaciones incrementales), que acumula niveles con el tiempo.

### Futuros USD-M

```python
from market_streaming import MarketStreamManager, StreamConfig, Market

config = StreamConfig(market=Market.FUTURES_USD)

with MarketStreamManager(config) as manager:
    manager.add_trade_stream("BTCUSDT")
    manager.add_order_book_stream("BTCUSDT")
    # ...
```

### Futuros COIN-M

```python
from market_streaming import StreamConfig, Market

config = StreamConfig(market=Market.FUTURES_COIN)
# Los pares de COIN-M usan formato distinto (ej: BTCUSD_PERP)
```

## API publica

### MarketStreamManager

Fachada principal del sistema. Gestiona streams, expone datos del cache y metricas.

#### Ciclo de vida

| Metodo | Descripcion |
|--------|-------------|
| `start()` | Arranca el motor en un thread dedicado |
| `stop()` | Detiene todos los streams y libera recursos |

Soporta context manager (`with MarketStreamManager(config) as mgr:`).

#### Registro de streams individuales

| Metodo | Descripcion |
|--------|-------------|
| `add_trade_stream(symbol)` | Suscribe al stream de trades de un simbolo |
| `add_order_book_stream(symbol, depth=20)` | Suscribe al stream de profundidad de un simbolo |
| `remove_stream(stream_id)` | Detiene y elimina un stream |

Cada metodo devuelve un `stream_id` (ej: `trade_BTCUSDT`, `depth_BTCUSDT`).

#### Registro de streams masivos (combinados)

| Metodo | Descripcion |
|--------|-------------|
| `add_trade_streams(symbols, batch_size=200)` | Registra trades para multiples simbolos en conexiones combinadas |
| `add_order_book_streams(symbols, depth=None, batch_size=200)` | Registra order books para multiples simbolos en conexiones combinadas |

Estos metodos crean `CombinedStreamWorker`s que multiplexan hasta `batch_size` simbolos por conexion WebSocket usando el endpoint `/stream?streams=...` de Binance. Devuelven una lista de worker IDs.

#### Consulta de datos

| Metodo | Retorno |
|--------|---------|
| `get_last_trade(symbol)` | `Trade` o `None` |
| `get_recent_trades(symbol, limit=100)` | `list[Trade]` |
| `get_order_book(symbol)` | `OrderBookSnapshot` o `None` |

Los simbolos son case-insensitive: `"btcusdt"` y `"BTCUSDT"` son equivalentes.

#### Observabilidad

| Metodo | Descripcion |
|--------|-------------|
| `get_stream_status(stream_id)` | Estado de un stream individual |
| `get_all_streams_status()` | Estado de todos los streams |
| `get_health()` | Diagnostico de salud global (`SystemHealth`) |
| `get_stats()` | Estadisticas resumidas (dict) |
| `is_healthy()` | `True` si todos los streams estan sanos |
| `print_summary()` | Imprime resumen por logging |

### Modelos de datos

#### Trade

```python
@dataclass(frozen=True)
class Trade:
    symbol: str          # par de trading (ej: "BTCUSDT")
    trade_id: int        # ID unico del trade en Binance
    price: float         # precio de ejecucion
    quantity: float      # cantidad ejecutada
    quote_quantity: float # cantidad en moneda cotizada
    buyer_maker: bool    # True si el comprador es el maker (trade = venta)
    timestamp_ms: int    # timestamp del trade en milisegundos
    price_str: str       # precio como cadena original (precision completa)
    quantity_str: str    # cantidad como cadena original
    quote_quantity_str: str
    is_best_match: bool  # solo Spot
    event_time_ms: int   # timestamp del evento WebSocket
```

Incluye `to_rest_format()` que devuelve un dict compatible con GET /api/v3/trades.

#### OrderBookSnapshot

```python
@dataclass
class OrderBookSnapshot:
    symbol: str
    bids: list[OrderBookLevel]   # ordenados precio descendente (mejor bid primero)
    asks: list[OrderBookLevel]   # ordenados precio ascendente (mejor ask primero)
    last_update_id: int          # ID de la ultima actualizacion aplicada
    timestamp: float             # time.time() de la ultima actualizacion
    event_time_ms: int
    transaction_time_ms: int
```

#### OrderBookLevel

```python
@dataclass(frozen=True)
class OrderBookLevel:
    price: float
    quantity: float
    price_str: str       # cadena original de Binance
    quantity_str: str
```

Incluye `to_rest_format()` que devuelve `[price_str, quantity_str]` compatible con la REST API.

#### StreamState

Estados posibles de un stream:

| Estado | Significado |
|--------|-------------|
| `starting` | Iniciando conexion |
| `running` | Conectado y recibiendo datos |
| `reconnecting` | Reconectando tras un error |
| `stale` | Conectado pero sin datos recientes |
| `stopped` | Detenido de forma limpia |
| `failed` | Error fatal, no reconectable |

#### SystemHealth

```python
@dataclass
class SystemHealth:
    is_healthy: bool             # True si todos los streams estan sanos
    total_streams: int
    running_streams: int
    stale_streams: int
    reconnecting_streams: int
    stopped_streams: int
    uptime_s: float              # segundos desde el arranque
    total_messages: int          # total de mensajes procesados
    stream_details: dict[str, StreamStatus]
```

## Configuracion

Toda la configuracion se centraliza en `StreamConfig`:

```python
from market_streaming import StreamConfig, Market

config = StreamConfig(
    # Mercado (default: Spot)
    market=Market.SPOT,       # SPOT | FUTURES_USD | FUTURES_COIN
    # base_url="wss://...",   # si se pasa, tiene prioridad sobre market

    # Conexion
    ping_interval_s=20.0,
    ping_timeout_s=10.0,

    # Cache
    max_recent_trades=1000,
    default_order_book_depth=20,   # >20 usa diff depth stream

    # Reconexion
    reconnect_base_delay_s=1.0,
    reconnect_max_delay_s=60.0,
    reconnect_jitter_factor=0.2,
    max_reconnect_attempts=0,  # 0 = sin limite

    # Health
    stale_threshold_s=30.0,

    # Logging
    log_level=logging.INFO,
    log_dir="logs",
    log_max_bytes=10 * 1024 * 1024,  # 10 MB por archivo
    log_backup_count=5,               # max 5 archivos rotados
)
```

### Mercados soportados

| Market | URL WebSocket | Ejemplo de par |
|--------|---------------|----------------|
| `Market.SPOT` | `wss://stream.binance.com:9443/ws` | `BTCUSDT` |
| `Market.FUTURES_USD` | `wss://fstream.binance.com/ws` | `BTCUSDT` |
| `Market.FUTURES_COIN` | `wss://dstream.binance.com/ws` | `BTCUSD_PERP` |

Todos los campos tienen valores por defecto razonables. `StreamConfig()` sin argumentos conecta a Spot.

## Logging

El sistema genera dos archivos de log con rotacion automatica:

| Archivo | Nivel | Uso |
|---------|-------|-----|
| `binflow.log` | DEBUG+ | Analisis completo, debug, tracing |
| `binflow.error.log` | WARNING+ | Deteccion rapida de fallos e interrupciones |

### Rotacion y limites de disco

- Cada archivo rota al alcanzar `log_max_bytes` (default: 10 MB)
- Se mantienen como maximo `log_backup_count` archivos rotados (default: 5)
- Espacio maximo en disco: ~120 MB (2 archivos x 6 versiones x 10 MB)
- Los archivos rotados se nombran `binflow.log.1`, `binflow.log.2`, etc.

### Formato de archivo

```
2026-03-06 20:15:32 | INFO     | market_streaming.worker:_connect_and_consume:178 | Stream trade_BTCUSDT: conexion establecida
2026-03-06 20:15:33 | WARNING  | market_streaming.worker:_run:123 | Error en stream depth_BTCUSDT (intento 1): [Errno 111] Connection refused
```

### Configuracion de logging

```python
import logging
from market_streaming import StreamConfig

# Produccion: solo errores en consola, todo en disco
config = StreamConfig(
    log_level=logging.ERROR,
    log_dir="logs",
    log_max_bytes=10 * 1024 * 1024,
    log_backup_count=5,
)

# Desarrollo: verbose en consola, sin disco
config = StreamConfig(
    log_level=logging.DEBUG,
    log_dir="",  # desactiva archivos en disco
)

# Directorio personalizado
config = StreamConfig(log_dir="/var/log/binflow")
```

## Arquitectura

```
MarketStreamManager           API publica sincrona (fachada)
    |
    +-- threading.Thread       Thread dedicado con event loop asyncio
    |       |
    |       +-- WebSocketStreamWorker (trade_BTCUSDT)     1 simbolo / conexion
    |       +-- WebSocketStreamWorker (depth_BTCUSDT)     1 simbolo / conexion
    |       +-- CombinedStreamWorker (combined_trades_0)  hasta 200 simbolos / conexion
    |       +-- CombinedStreamWorker (combined_depth_0)   hasta 200 simbolos / conexion
    |       +-- ...
    |
    +-- MarketDataCache        Cache en memoria (threading.Lock)
    |
    +-- HealthMonitor          Evaluacion de salud
    |
    +-- Logging                RotatingFileHandler (general + errores)
```

### Workers

| Worker | Uso | Conexiones |
|--------|-----|-----------|
| `WebSocketStreamWorker` | 1 simbolo por conexion (`add_trade_stream`, `add_order_book_stream`) | 1 WS por stream |
| `CombinedStreamWorker` | Hasta 200 simbolos por conexion (`add_trade_streams`, `add_order_book_streams`) | 1 WS por batch |

Los workers combinados usan el endpoint `/stream?streams=sym1@trade/sym2@trade/...` de Binance y desenvuelven el formato `{stream, data}` de los mensajes.

### Modelo de concurrencia

El sistema usa un **thread dedicado con event loop asyncio**:

- **API publica sincrona**: el codigo cliente llama `manager.start()`, `manager.get_last_trade()`, etc. sin `await`. Esto permite usar el motor desde cualquier contexto Python.
- **Event loop interno**: un thread daemon ejecuta el loop asyncio donde corren todos los workers WebSocket de forma concurrente.
- **Cache thread-safe**: `threading.Lock` protege las escrituras (desde el thread asyncio) y las lecturas (desde el thread principal). El contention es despreciable porque las operaciones son O(1).

### Reconexion

Cada worker gestiona su propia reconexion:

1. Error de conexion o desconexion detectada
2. Backoff exponencial: `min(base * 2^n, max_delay)`
3. Jitter aleatorio de +-20% sobre el delay calculado
4. Contador de reconexiones incrementado
5. Si se supera `max_reconnect_attempts` (y no es 0), el stream pasa a `failed`
6. Errores fatales (ej: stream invalido) no reconectan

### Depth streams

| Profundidad | Stream | Tipo de mensaje |
|-------------|--------|-----------------|
| 5, 10, 20 | `symbol@depth{N}` (Partial Book Depth) | Snapshots completos cada ~1s |
| Cualquier otro valor | `symbol@depth` (Diff Depth) | Actualizaciones incrementales |

El diff depth stream acumula niveles con el tiempo. Para un order book completo desde el primer momento, usar profundidades de 5, 10 o 20.

### Cache en memoria

| Dato | Estructura | Acceso |
|------|-----------|--------|
| Ultimo trade por simbolo | `dict[symbol, Trade]` | O(1) |
| Trades recientes por simbolo | `dict[symbol, deque(maxlen=N)]` | O(1) amortizado |
| Order book por simbolo | `dict[symbol, OrderBookSnapshot]` | O(1) |

El order book soporta tanto snapshots completos como actualizaciones incrementales con eliminacion de niveles con cantidad cero.

## Estructura del proyecto

```
market_streaming/
    __init__.py          Re-exports de la API publica
    config.py            StreamConfig, Market, MARKET_URLS
    models.py            Trade, OrderBookSnapshot, StreamStatus, SystemHealth
    exceptions.py        Excepciones del dominio
    cache.py             MarketDataCache
    worker.py            WebSocketStreamWorker, CombinedStreamWorker
    health.py            HealthMonitor
    manager.py           MarketStreamManager
    logging_config.py    Setup de logging con rotacion
tests/
    test_cache.py        18 tests - trades, order book, continuidad
    test_models.py       16 tests - inmutabilidad, defaults, config, mercados
    test_health.py        8 tests - deteccion stale, salud global
    test_worker.py       12 tests - URLs por mercado, parsing mensajes, backoff
    test_manager.py      13 tests - lifecycle, streams, observabilidad
    test_logging.py      13 tests - handlers, rotacion, archivos, teardown
    test_load.py          8 tests - carga: 728 symbols trades, 60s
    test_load_orderbook.py 13 tests - carga: 728 symbols order books profundos
examples/
    inspect_trade.py     Estructura del Trade: campos, tipos, precision, formato REST
    inspect_order_book.py  OrderBookSnapshot: estructura, niveles, formato REST
    inspect_config.py    StreamConfig: campos, Market enum, URLs (sin conexion)
    inspect_health.py    SystemHealth, StreamStatus, StreamState
    live_trades.py       Feed de trades en tiempo real
    live_order_book.py   Order book visual con barras de volumen y spread
    multi_market.py      Tabla comparativa multi-simbolo en tiempo real
notebooks/
    streaming_demo.ipynb Demo completa: stream, inspeccion, acumulacion, verificacion
```

## Ejemplos

Todos los ejemplos se ejecutan con `python examples/<nombre>.py`.

### Inspeccion de datos (sin conexion)

```bash
python examples/inspect_trade.py        # estructura Trade
python examples/inspect_order_book.py   # estructura OrderBookSnapshot
python examples/inspect_config.py       # StreamConfig, Market, URLs
python examples/inspect_health.py       # SystemHealth, StreamStatus
```

### Streams en vivo

```bash
python examples/live_trades.py          # feed de trades BTCUSDT en tiempo real
python examples/live_order_book.py      # order book visual con spread
python examples/multi_market.py         # tabla comparativa multi-simbolo
```

### Notebook

```bash
jupyter notebook notebooks/streaming_demo.ipynb
```

Demo interactiva: arranque, inspeccion, acumulacion de 2 minutos con verificacion de continuidad, estadisticas y shutdown limpio.

## Tests

```bash
# Ejecutar todos los tests unitarios (83 tests, <1s)
pytest tests/ --ignore=tests/test_load.py --ignore=tests/test_load_orderbook.py -v

# Tests de carga contra Binance real (~70s cada uno)
pytest tests/test_load.py -v -s              # 728 symbols, trades, 60s
pytest tests/test_load_orderbook.py -v -s    # 728 symbols, depth 200, 60s

# Todos los tests (102 tests)
pytest tests/ -v -s

# Un test especifico
pytest tests/test_cache.py::TestTradeCache::test_trade_ids_continuous -v
```

### Pruebas de carga

Los tests de carga (`test_load.py`, `test_load_orderbook.py`) conectan a Binance real y verifican:

- **Continuidad**: trade IDs consecutivos sin huecos
- **Unicidad**: cero duplicados
- **Orden**: IDs estrictamente crecientes
- **Timestamps**: no decrecientes
- **Profundidad**: order books de hasta 200 niveles por lado
- **Integridad estructural**: bids descendentes, asks ascendentes, spread positivo
- **Precision**: cadenas originales de Binance preservadas
- **Estabilidad**: todos los workers running, cero reconexiones

## Limitaciones

- No implementa el flujo completo de sincronizacion del order book de Binance (REST snapshot + WebSocket diff). Para profundidades > 20, el diff depth stream acumula niveles incrementalmente sin snapshot REST inicial.
- No persiste datos en disco. Si el proceso muere, se pierde el cache.
- Las metricas son in-process. No hay exportacion a Prometheus, StatsD u otros sistemas externos.
- No implementa autenticacion para streams privados (user data streams).

## Licencia

MIT
