Feeds mdhub (/mdh)

Especificação da interface do SDK para os feeds mdhub da B3 retransmitidos pelo kafka-mdh-marketdata-broadcaster (iflex-kafka) no endpoint /mdh. Esses feeds convivem com os canais de market data legados (subscribe_book, subscribe_trades, subscribe_candles, subscribe_option_books, subscribe_indicator) descritos em Streaming, sem substituí-los. As duas famílias compartilham o mesmo objeto KafkaMarketDataStream, a mesma fila de eventos, o mesmo loop de reconexão e a mesma contagem de referências.

Transporte

O broadcaster aceita apenas destinos no padrão /topic/mdh/([a-z0-9]+)/bvmf(?:/([A-Za-z0-9._:$/-]+))? com um feed que ele conheça. Um feed desconhecido encerra a sessão STOMP, levando junto todas as outras assinaturas /mdh; por isso o SDK valida o nome do feed no cliente e lança erro antes de enviar qualquer coisa.

Configuração

O broadcaster é um serviço separado do host Kafka WS legado. Por padrão ele usa o mesmo host, mas pode ser apontado para outro:

No momento em que este texto foi escrito, o broadcaster roda em https://dev-flexscan-kafkastream.flexscan.com.br, enquanto os canais legados são servidos por prd-flexscan-kafkastream; ajuste INVESTFLEX_MDH_WS_URL de acordo.

Feeds

Os 24 feeds publicados pelo broadcaster, com o método que assina cada um. Todo método aceita um scope opcional; omiti-lo assina o feed inteiro (todos os instrumentos). L3 não é um feed: o broadcaster não tem rota l3, então não existe subscribe_l3.

Os métodos nomeados são atalhos de uma linha para o par genérico:

A lista de feeds e o mapa feed → modelo são exportados como investflex.models.mdh.MDH_FEEDS / MDH_FEED_MODELS e MDH_FEEDS / MDH_FEED_SCHEMAS (TypeScript), com o tipo união MdhFeed e o guard isMdhFeed().

Eventos

Cada frame é emitido como um evento na fila compartilhada:

{
  "channel": "trd",
  "data": {
    "symbol": "PETR4",
    "timestamp": "2026-10-05T18:40:00.540Z",
    "securityId": "200003061028",
    "price": 56.18,
    "quantity": 100,
    "transactTime": 1791225600540415628,
    "...": "todos os outros campos do payload, sem alteração"
  }
}

• channel é o nome do feed — distinto dos nomes dos canais legados, então on("mbp", …) e evt["channel"] == "trd" funcionam. É o único lugar onde o nome do feed aparece num evento bruto; listeners e MarketDataService.parse() despacham por ele.
• data é o payload do broadcaster repassado sem alteração, no seu próprio camelCase, com dois acréscimos do SDK:
  • symbol — vindo do envelope (scope no caso de gphase).
  • timestamp — um datetime / Date com fuso, derivado do primeiro entre transactTime e marketDataEntryTimestamp (nanossegundos desde a epoch → precisão de ms). Books (mbp/cmbp) não têm nenhum dos dois e recebem None/null. receiveTimestampNanos nunca é usado: é um relógio monotônico de uptime, não uma epoch.
• Os modelos parseados carregam feed. MarketDataService.parse() descarta o envelope e vários modelos são compartilhados por mais de um feed (MdhPriceStatistic ← 7 feeds, MdhBook ← mbp/cmbp, MdhEnrichedTrade ← xtrd/cxtrd), então ele copia channel para o campo feed do modelo.

Timestamps em nanossegundos

transactTime, aggressorTime, marketDataEntryTimestamp e receiveTimestampNanos são inteiros de 19 dígitos. O int do Python os representa exatamente. Em JavaScript eles excedem Number.MAX_SAFE_INTEGER e o JSON.parse os arredondaria; por isso, no canal /mdh, o SDK coloca entre aspas todo literal inteiro com 16 ou mais dígitos no corpo STOMP bruto antes do parse: em TypeScript esses campos são strings ("1791225600540415628"), e mdhNanosToDate() aceita qualquer um dos dois. Esta é a única assimetria deliberada entre as linguagens nos modelos.

Uso

Iterando

from investflex.models.mdh import MdhBook, MdhEnrichedTrade
from investflex.services.marketdata import MarketDataService

async with AsyncInvestflexClient.from_env() as client:
    async with client.marketdata.stream() as stream:
        await stream.subscribe_xtrd("PETR4")
        await stream.subscribe_mbp("PETR4")
        await stream.subscribe_gphase("EQT")
        async for evt in MarketDataService.parse(stream):
            if isinstance(evt, MdhEnrichedTrade):
                print("trade", evt.price, "x", evt.quantity, "vwap", evt.vwap_price)
            elif isinstance(evt, MdhBook) and evt.bids and evt.asks:
                print("touch", evt.bids[0].price, "/", evt.asks[0].price)

import { MarketDataService, MdhBookSchema, MdhEnrichedTradeSchema } from "@investflex/sdk";

const client = InvestflexClient.fromEnv();

const stream = await client.marketdata.stream().open();
await stream.subscribeXtrd("PETR4");
await stream.subscribeMbp("PETR4");
await stream.subscribeGphase("EQT");
for await (const evt of stream) {
  if (evt.channel === "xtrd") {
    const trade = MdhEnrichedTradeSchema.parse(evt.data);
    console.log("trade", trade.price, "x", trade.quantity, "vwap", trade.vwapPrice);
  } else if (evt.channel === "mbp") {
    const book = MdhBookSchema.parse(evt.data);
    if (book.bids[0] && book.asks[0]) console.log("touch", book.bids[0].price, "/", book.asks[0].price);
  }
}
await stream.close();

MarketDataService.parse() mapeia cada feed para o seu modelo (MDH_FEED_MODELS[feed].model_validate(data) / MDH_FEED_SCHEMAS[feed].parse(data)); os canais legados mantêm o mapeamento existente e canais desconhecidos passam sem alteração.

Listeners

stream.on("mbo", (frame) => console.log(frame.channel, frame.data.action, frame.data.orderId));
stream.on("reconnect", (channel) => console.log("reconnected", channel));
await stream.subscribeMbo("WINV26");

O evento "reconnect" informa a conexão, então para qualquer feed mdhub ele reporta "mdh"; as assinaturas reenviadas cobrem todos os feeds dessa conexão.

Assinaturas do feed inteiro

await stream.subscribe_trd()          # negócios de todos os instrumentos
await stream.subscribe_gphase()       # todos os grupos de instrumentos
await stream.unsubscribe_mdh("trd")   # ou unsubscribe(channel="trd")

await stream.subscribeTrd();
await stream.subscribeGphase();
await stream.unsubscribeMdh("trd");

O mbp do feed inteiro é uma mangueira de incêndio (um snapshot completo do book a cada atualização; só o WINV26 gera ~1.100 msg/s). Prefira cmbp, ou scopes por papel, e atenção a queue_size/backpressure (veja Streaming → Backpressure).

Modelos

Os modelos Python ficam em investflex.models.mdh (pydantic, aliases camelCase, populate_by_name=True, campos desconhecidos preservados); os schemas TypeScript em @investflex/sdk (Mdh*Schema + tipos Mdh* inferidos, .passthrough()). Todos os campos são opcionais: a bolsa envia null explícito para tudo o que não sabe.

Comum a todo modelo parseado (MdhEvent): feed (preenchido por parse() a partir de channel), symbol, securityId, receiveTimestampNanos, packetSequenceNumber, reportSequence, timestamp. gphase tem scope em vez de symbol e não tem securityId/reportSequence.

Os campos de rótulo (aggressorSide, aggressorSource, action, side, reason) são a decodificação feita pelo próprio broadcaster do campo *Code correspondente:

Helpers: mdh_trade_date(days) / mdhTradeDate(days) converte o tradeDate em dias desde a epoch (20731 → 2026-10-05) numa data; mdh_nanos_to_datetime(ns) / mdhNanosToDate(ns) converte um timestamp em nanossegundos desde a epoch num datetime / Date.

Peculiaridades do wire

• Preços em notação científica. Preços/quantidades grandes podem chegar como 2.0938E+5. Isso é JSON válido e é parseado para o número correto; só fica estranho nos logs.
• tradeDate é em dias desde a epoch, não um inteiro YYYYMMDD. A exceção é o settle, cujos settlDate e tradeDate são strings ISO YYYY-MM-DD (ausentes em broadcasters antigos).
• null explícito. Campos opcionais são enviados como null, nunca omitidos. Os grupos de enriquecimento do xtrd são controlados por bits em enrichmentFlags (o bit 8, openInterest, fica desligado para ações).
• mbp é um snapshot, não um delta. Toda atualização traz as escadas bids/asks completas (bidDepth/askDepth níveis). Substitua, não mescle.
• def está silencioso no momento. A sincronização de definições do broadcaster está desativada na implantação atual; a assinatura funciona, mas nada chega.
• Sem L3. l3 não é um FeedType; o SDK o rejeita no cliente porque o servidor fecharia a sessão.
• receiveTimestampNanos não é hora de relógio. É o relógio monotônico de uptime do decodificador; use transactTime/timestamp.

Compatibilidade

Nada muda nos canais legados: subscribe_book, subscribe_trades, subscribe_candles, subscribe_option_books e subscribe_indicator mantêm seus destinos, endpoints, formatos normalizados e nomes de channel. A única ampliação de uma assinatura existente é unsubscribe, cujo symbol passou a ser opcional para que uma assinatura mdhub do feed inteiro possa ser liberada; para os canais legados, um symbol ausente não faz nada, como já acontecia com um canal desconhecido.