InvestflexInvestflexSDK
Streaming

Streaming

O SDK expõe dois streams de longa duração:

  • Stream de snapshots — atualizações de algos enviadas pela engine.
  • Stream de market data — atualizações de book e trades por papel, além dos feeds brutos do mdhub da B3 (trd, xtrd, mbp, mbo, …) retransmitidos em /mdh; veja Feeds mdhub.

Ambos são async iterators com reconexão automática, replay de assinaturas e um sentinela StreamReconnected após cada reconexão bem-sucedida, para que o consumidor possa re-inicializar seu estado.

Stream de snapshots

Assine todos os seus algos rodando (ou um subconjunto filtrado) e reaja conforme a engine envia atualizações.

from investflex.ws.stream import StreamReconnected
from investflex.models.snapshot import StrategySnapshot

async with AsyncInvestflexClient.from_env() as client:
    async with client.algos.stream() as stream:
        await stream.subscribe_all()
        async for evt in stream:
            if isinstance(evt, StreamReconnected):
                # A conexão caiu e voltou. Reinicialize se você
                # mantém um cache local.
                continue
            if isinstance(evt, StrategySnapshot):
                print(evt.clord_id, evt.status.name,
                      evt.output.cum_qty_perc, "%")

import { StreamReconnected } from "@investflex/sdk";

const client = InvestflexClient.fromEnv();

const stream = await client.algos.stream().open();
await stream.subscribeAll();
for await (const evt of stream) {
  if (evt instanceof StreamReconnected) {
    // A conexão caiu e voltou. Ressemeie se você
    // mantém um cache local.
    continue;
  }
  console.log(evt.clOrdId, evt.status, evt.output.cumQtyPerc, "%");
}
await stream.close();

Se você prefere um dict gerenciado em vez de consumir eventos na mão, veja Snapshots → Live view.

Backpressure

Os construtores de stream aceitam um modo backpressure:

  • "block" (padrão) — consumidores lentos seguram o produtor.
  • "drop_oldest" — limita a fila descartando eventos antigos.
client.algos.stream(queue_size=2048, backpressure="drop_oldest")

client.algos.stream({ queueSize: 2048, backpressure: "drop_oldest" });

Stream de market data

O stream de market data é por canal: você assina um canal book ou trades para um símbolo de cada vez. Assinaturas são contadas por referência, então assinar o mesmo par (canal, símbolo) duas vezes gera apenas uma assinatura upstream.

from investflex.models.marketdata import Book, Trade

async with AsyncInvestflexClient.from_env() as client:
    async with client.marketdata.stream() as stream:
        await stream.subscribe_book("PETR4")
        await stream.subscribe_trades("PETR4")
        async for evt in stream:
            channel = evt.get("channel")
            data = evt.get("data") or {}
            if channel == "book":
                book = Book.model_validate(data)
                if book.bids and book.asks:
                    print("toque", book.bids[0].price, "/", book.asks[0].price)
            elif channel == "trades":
                trade = Trade.model_validate(data)
                print("trade", trade.price, "x", trade.qty)

const client = InvestflexClient.fromEnv();

const stream = await client.marketdata.stream().open();
await stream.subscribeBook("PETR4");
await stream.subscribeTrades("PETR4");
for await (const evt of stream) {
  const data = evt.data ?? {};
  if (evt.channel === "book") {
    const book = data as { bids?: { price: number }[]; asks?: { price: number }[] };
    if (book.bids?.length && book.asks?.length) {
      console.log("touch", book.bids[0]!.price, "/", book.asks[0]!.price);
    }
  } else if (evt.channel === "trades") {
    const trade = data as { price?: number | null; qty?: number | null };
    console.log("trade", trade.price, "x", trade.qty);
  }
}
await stream.close();

Para conveniência, MarketDataService.parse(stream) faz a classificação + model_validate e devolve Book, Trade ou Definition tipados:

from investflex.services.marketdata import MarketDataService
from investflex.models.marketdata import Book, Trade

async with AsyncInvestflexClient.from_env() as client:
    async with client.marketdata.stream() as stream:
        await stream.subscribe_book("PETR4")
        await stream.subscribe_trades("PETR4")
        async for evt in MarketDataService.parse(stream):
            if isinstance(evt, Book):
                print("book", len(evt.bids), "/", len(evt.asks))
            elif isinstance(evt, Trade):
                print("trade", evt.price)

import { MarketDataService, type Book, type Trade } from "@investflex/sdk";

const client = InvestflexClient.fromEnv();

const stream = await client.marketdata.stream().open();
await stream.subscribeBook("PETR4");
await stream.subscribeTrades("PETR4");
for await (const evt of MarketDataService.parse(stream)) {
  const book = evt as Partial<Book>;
  const trade = evt as Partial<Trade>;
  if (book.bids && book.asks) {
    console.log("book", book.bids.length, "/", book.asks.length);
  } else if (trade.price != null) {
    console.log("trade", trade.price);
  }
}
await stream.close();

Book

Book é a visão MBP agregada de um papel:

Campo Tipo Observações
symbol str
bids list de BookLevel Ordenado do melhor ao pior.
asks list de BookLevel Ordenado do melhor ao pior.
timestamp datetime | None

BookLevel: price, qty, orders.

Minibooks de opções agrupados

O canal de minibook de opções faz uma única assinatura para todos os contratos de uma combinação ativo-objeto/vencimento. A data aceita YYYY-MM-DD ou YYYYMMDD. Os frames usam channel == "option_books"; os metadados de rota ficam ao lado de data, e campos ausentes não devem apagar o último book/trade. Faça o merge por símbolo.

await stream.subscribe_option_books("VALE3", "2026-09-04")
# ... consuma frames option_books ...
await stream.unsubscribe_option_books("VALE3", "2026-09-04")

await stream.subscribeOptionBooks("VALE3", "2026-09-04");
stream.on("option_books", (frame) => {
  console.log(frame.underlying, frame.maturityDate, frame.data);
});
await stream.unsubscribeOptionBooks("VALE3", "2026-09-04");

O destino é /topic/minibook/{market}/{underlying}/{securityType}/{YYYYMMDD}; securityType usa OPT por padrão. Após reaplicar as assinaturas, o stream TypeScript emite reconnect com o canal afetado para permitir um novo seed REST.

Trades detalhados

Por padrão os frames de trade trazem symbol, price, qty, trade_id e timestamp. Passe detailed=True na assinatura e cada evento de trade daquele papel passa a carregar também os campos extras do venue:

await stream.subscribe_trades("PETR4", detailed=True)

await stream.subscribeTrades("PETR4", { detailed: true });
Campo Tipo Notas
buyer_id / seller_id str | None Códigos das corretoras.
vwap float | None VWAP do pregão.
trd_vol int | None Quantidade acumulada negociada.
fin_vol float | None Volume financeiro acumulado.
var float | None Variação sobre o preço de referência.
band_max_price / band_min_price float | None Bandas de leilão.
quantity_band_limit int | None Banda de quantidade do leilão.
open_int int | None Contratos em aberto.
open_px / close_px / high_px / low_px float | None Preços do pregão.
trade_nbr int | None Número sequencial do negócio.

A opção vale por papel e persiste enquanto a assinatura existir: o stream tem uma única fila de eventos, então basta um assinante pedir detalhe para que todos os consumidores daquele papel vejam os campos extras. Ela é limpa quando o último assinante do papel cancela a assinatura — reassine com detailed=True se ainda os quiser.

Cancelando assinatura

await stream.unsubscribe(channel="trades", symbol="PETR4")

await stream.unsubscribe({ channel: "trades", symbol: "PETR4" });

Unsubscribes também são contados por referência; a assinatura upstream permanece enquanto outro consumidor a mantiver.

Feeds mdhub

O mesmo objeto de stream também transporta os feeds mdhub da B3 vindos do broadcaster /mdh: negócios, books por ordem e agregados, resumos de execução, estatísticas, fases de negociação e mais. Cada feed tem um método com o seu nome de wire (subscribe_trd, subscribe_mbp, subscribe_mbo, …; subscribeTrd, subscribeMbp, subscribeMbo, … em TypeScript). Os eventos chegam com o nome do feed em channel e o payload da bolsa repassado sem alterações:

await stream.subscribe_xtrd("PETR4")
await stream.subscribe_mbp("PETR4")
async for evt in stream:
    if evt["channel"] == "xtrd":
        print("trade", evt["data"]["price"], "vwap", evt["data"]["vwapPrice"])

await stream.subscribeXtrd("PETR4");
await stream.subscribeMbp("PETR4");
stream.on("xtrd", (frame) => console.log("trade", frame.data.price, "vwap", frame.data.vwapPrice));

A lista completa de feeds, o formato dos eventos, os modelos e as peculiaridades do wire estão especificados em Feeds mdhub.

Semântica de reconexão

Ambos os streams reconectam automaticamente em falhas transientes de rede, com backoff exponencial. Após uma reconexão bem-sucedida:

  • Assinaturas pré-existentes são reaplicadas no upstream.
  • Um StreamReconnected(attempt=N) é emitido no fluxo de eventos para você re-inicializar estado local (ex.: listando os algos novamente, ou limpando uma escada de livro).

Se você quer essa re-inicialização tratada automaticamente, o helper live view faz isso de forma transparente.

Limpeza

Streams são async context managers — entrar no async with conecta, sair derruba o WebSocket. Não gerencie a conexão na mão.