Real-time data

Subscriptions, ticks, and current-value boards.

A Subscription is an async iterator of sparse ticks, an Arrow batch stream through arrow(), and a polled board through latest(). Subscriptions to the same security share one Bloomberg feed per engine.

WARNING

Live subscriptions need real-time entitlements for every security and field. Without them Bloomberg may send delayed data or reject fields; xbbg reports both instead of hiding them.

Open a subscription

Call	Service	Notes
subscribe(tickers, fields, options) or asubscribe(...)	//blp/mktdata	Top-level and blp helpers on the engine set by configure(); accept a string or an array
engine.subscribe(tickers, fields, options)	//blp/mktdata	The core market-data subscription
engine.stream(tickers, fields, options)	//blp/mktdata	Same as engine.subscribe
engine.vwap, mktbar, depth, chains	//blp/mktvwap, //blp/mktbar, //blp/mktdepthdata, //blp/mktlist	See Other services
engine.subscribeWithOptions(service, ...)	Any	Positional arguments; subscription options are passed through unchanged
engine.subscribeAuction, streamAuction, subscribeEtfInav	//blp/mktdata	Resolve venues or iNAV tickers first; see Recipes

All of them resolve to a Subscription once the subscription is open.

import { asubscribe } from '@xbbg/core';

const sub = await asubscribe(['AAPL US Equity', 'MSFT US Equity'], ['LAST_PRICE', 'BID', 'ASK']);
let received = 0;
try {
  for await (const tick of sub) {
    if (tick.has('LAST_PRICE')) {
      console.log(tick.topic, tick.timestampUs, tick.f64('LAST_PRICE'));
    }
    received += 1;
    if (received === 100) break; // leaving the loop unsubscribes
  }
} finally {
  await sub.unsubscribe();
}

Read ticks

Iterating a Subscription yields one Tick per update. Ticks are sparse deltas: each one carries only the fields present in that update.

Member	Returns
topic	The consumer label of the security (its alias, or the topic as subscribed)
timestampUs	Update timestamp in microseconds since the epoch
has(field)	true when the update carries the field, including an explicit clear
get(field)	The value, null for an explicit clear, or undefined when the field is absent
f64(field), i64(field), str(field)	The value as number, bigint, or string, keeping null and undefined
toObject()	{ timestampUs, topic, ...fields } with absent fields omitted

import { connect } from '@xbbg/core';

const engine = await connect();
const sub = await engine.subscribe(['ES1 Index'], ['BID', 'ASK']);
const bidField = sub.field('BID'); // a reusable FieldHandle

try {
  for await (const tick of sub) {
    const bid = tick.get(bidField);
    if (bid === undefined) continue; // BID is not in this update
    console.log(tick.topic, bid === null ? 'BID cleared' : bid);
  }
} finally {
  await sub.unsubscribe();
}

get() returns booleans, numbers (32-bit integers and floats), bigint (64-bit integers, and times and timestamps in microseconds), Date (dates, at UTC midnight), or strings, according to the field's type.

Read Arrow batches

sub.arrow() returns an ArrowSubscription over the same subscription. Each read returns one Apache Arrow Table built directly from native buffers, without an IPC round trip, holding every update taken in one native crossing.

import { connect } from '@xbbg/core';

const engine = await connect();
const sub = await engine.subscribe(['ES1 Index', 'NQ1 Index'], ['BID', 'ASK', 'LAST_PRICE'], {
  flushThreshold: 256, // take up to 256 queued updates per read
});
const batches = sub.arrow();

try {
  for await (const table of batches) {
    console.log(table.numRows, table.schema.fields.map((field) => field.name));
  }
} finally {
  await batches.unsubscribe();
}

• Columns are timestamp, topic, the fields, then a non-null binary __xbbg_present column. Bit i of __xbbg_present (least-significant bit first) marks schema field i + 2: a set bit with a null value is an explicit clear, an unset bit means unchanged. Read each table against its own schema, because the layout can change between batches. The schema metadata key xbbg.subscription_presence describes the encoding.
• Pick one read style per subscription: scalar ticks, Arrow tables through sub.arrow(), or delivery envelopes through sub.deliveries(). The first read fixes the mode; reading another way rejects with a TypeError. All views share one serialized source, not independent consumers.
• flushThreshold (default: the engine's subscriptionFlushThreshold, 16) is the maximum number of already-queued updates combined into one Arrow table or delivery. A read never waits for a batch to fill: a quiet stream still yields one-row tables, while a burst drains in fewer, larger batches. Scalar iteration still yields one Tick per update.
• Unsupported column types fail with a column-level error instead of switching transport.

Options

StreamOptions applies to subscribe, asubscribe, stream, and the service helpers.

Option	Default	Meaning
options	none	Bloomberg subscription options such as 'interval=5'; leading & is removed
conflate	false	Ask Bloomberg to conflate quote updates (trades still stream as received). //blp/mktdata only; cannot be combined with an interval option
allFields	false	Deliver every top-level scalar field Bloomberg sends, not only the requested fields plus MKTDATA_EVENT_TYPE and MKTDATA_EVENT_SUBTYPE
aliases	none	{ [upstreamTopic]: label }; labels appear in ticks, tickers, status keys, and remove()
onDelayed	'warn'	Delayed-data policy: 'warn', 'raise', or 'ignore'
onFieldError	'warn'	Rejected-field policy: 'warn', 'raise', or 'ignore'
isolated	false	true gives this consumer its own feeds and session instead of sharing
rows	true	false keeps only a current-value image for latest(); no rows are queued
zeroAsNull	[]	Fields whose numeric zero (including -0) becomes null in latest() and state deliveries
flushThreshold	Engine subscriptionFlushThreshold (16)	Maximum already-queued updates combined into one read batch
streamCapacity	Engine subscriptionStreamCapacity	Per-consumer buffer size
overflowPolicy	Engine overflowPolicy	'drop_newest', 'block', or 'drop_oldest' when the buffer is full; 'drop_oldest' requires sub.deliveries()
delivery	'deltas'	'state' streams conflated current images through sub.deliveries(); see Gap-aware deliveries
fields	none	Field list for mktbar, depth, chains, subscribeAuction, and subscribeEtfInav, which take no positional fields argument

In the default filtered mode, rows that contain none of the consumer's requested fields are skipped, so a consumer's output does not depend on other consumers of the same feed.

Shared feeds and dynamic membership

Within one Engine, //blp/mktdata consumers share a Bloomberg feed when the service, the trimmed topic, and the normalized subscription options match; aliases do not matter.

• The feed requests the union of its consumers' fields. The union grows when a consumer joins or calls addFields(), with a single re-subscribe, and does not shrink until the feed closes.
• A row consumer that joins a live feed first receives a synthetic SUMMARY/INITPAINT row built from the known values. When the union grows, consumers still waiting for an image get the full repaint, and existing readers receive only values that changed.
• isolated: true opts out of sharing. Other services are never shared.
• Overflow or failure in one consumer's queue affects only that consumer, and unsubscribing one consumer leaves its siblings running.
• engine.subscriptionFeeds() lists the current feeds (FeedInfo: service, topic, options, union fields, consumer count, delayed flag, state, isolation, and field errors). It contains no authentication or connection details.

import { connect } from '@xbbg/core';

const engine = await connect();
const sub = await engine.subscribe(['IBM UN Equity'], ['BID'], {
  aliases: { 'IBM UN Equity': 'IBM US Equity' },
});

try {
  await sub.addFields(['ASK']); // grows this consumer and the shared feed
  await sub.add(['MSFT UW Equity'], { 'MSFT UW Equity': 'MSFT US Equity' });
  console.log(sub.tickers, sub.fields); // labels and the current field projection
  console.log(engine.subscriptionFeeds());
  await sub.remove(['IBM US Equity']); // remove by label
} finally {
  await sub.unsubscribe();
}

add(tickers, aliases?) takes upstream topics plus optional labels; remove(tickers) takes labels.

Current-value boards

latest({ backend }) returns the last known value of every field, one row per security, as an Arrow table or JSON rows. It is synchronous, sends no Bloomberg request, and does not consume ticks, so it works on any subscription. With rows: false, nothing is queued, so the board cannot overflow.

import { setTimeout as sleep } from 'node:timers/promises';
import { connect } from '@xbbg/core';

const engine = await connect();
const board = await engine.subscribe(['IBM US Equity', 'MSFT US Equity'], ['BID', 'ASK', 'LAST_PRICE'], {
  rows: false,
  zeroAsNull: ['BID', 'ASK'],
});

try {
  for (let i = 0; i < 10; i += 1) {
    await sleep(1_000);
    console.log(board.latest({ backend: 'json' }));
  }
} finally {
  await board.unsubscribe();
}

• Columns are topic (the label), nullable UTC last_update, live, nullable delayed, then the data fields. Metadata fields are excluded; unknown values and explicit clears are null.
• Field types come from Bloomberg field metadata when it is available before the first value.
• Reading rows from a rows: false board rejects with BlpValidationError and leaves the board open.
• latest() throws once the subscription has failed, closed, or been unsubscribed. Removing every topic from an open subscription returns an empty table.

Delayed data and rejected fields

Option	'warn' (default)	'raise'	'ignore'
onDelayed	One BlpDelayedDataWarning per delayed topic	Fails only the delayed topic; healthy topics keep streaming	Records the delayed flag only
onFieldError	One BlpFieldWarning per rejected field	Fails only topics with a rejected explicit field, recording the field and category in the failure reason	Records fieldErrors without a warning

Warnings go through Node's process.emitWarning(message, { type, code, detail }):

Event	Warning type	Code
Delayed topic	BlpDelayedDataWarning	XBBG_DELAYED_STREAM
Rejected explicit field	BlpFieldWarning	XBBG_FIELD_EXCEPTION

import { connect } from '@xbbg/core';

process.on('warning', (warning) => {
  if (warning.name === 'BlpDelayedDataWarning' || warning.name === 'BlpFieldWarning') {
    console.warn(`${warning.name}: ${warning.message}`);
  }
});

const engine = await connect();
const sub = await engine.subscribe(['VOD LN Equity'], ['LAST_PRICE', 'BID'], {
  onDelayed: 'warn',
  onFieldError: 'raise',
});

try {
  for await (const tick of sub) {
    console.log(tick.toObject());
    break;
  }
  console.log(sub.topicStates, sub.fieldErrors);
} finally {
  await sub.unsubscribe();
}

Warnings are emitted when the subscription opens and when it is read, closed, or changed with add, remove, addFields, or latest. Emitting a warning does not clear the events history or fieldErrors. Request-level delayed checks are separate; see Output Formats.

Diagnostics

Getter	Contents
status	One combined snapshot: events, failures, failedTickers, topicStates, fieldErrors, session, services, and admin
events	Bounded history of SubscriptionEvent (atUs, category, level, messageType, topic, detail)
failures, failedTickers	Failed topics with reason and kind, and their labels
topicStates	Per label: feedTopic, delayed, state, streamsActive, and change times
fieldErrors	Label to field to error category
sessionStatus	Session state, disconnectCount, reconnectCount
serviceStatus	Per service: up and change time
adminStatus	Slow-consumer warnings and data-loss counts with timestamps
stats	messagesReceived, droppedBatches, batchesSent, slowConsumer
isActive, tickers, fields	Whether the subscription is open, its labels, and its current field projection

import { connect } from '@xbbg/core';

const engine = await connect();
const sub = await engine.subscribe(['AAPL US Equity'], ['LAST_PRICE']);
try {
  const { failures, topicStates, session, admin } = sub.status;
  console.log(failures, topicStates, session.reconnectCount, admin.dataLossCount, sub.stats);
} finally {
  await sub.unsubscribe();
}

Gap-aware deliveries

sub.deliveries(options?) returns a DeliveryStream of SubscriptionDelivery envelopes instead of bare ticks or tables. Each envelope has kind ('deltas' or 'state'), data (an Arrow table by default; { backend: 'json' } or { backend: 'polars' } changes only the payload), and an optional gapBefore. Two options need these envelope reads, and plain scalar and Arrow reads reject them:

• delivery: 'state' conflates updates into current images. Each read returns full rows for the topics that changed, taken fairly from a deduplicated queue, with topic, last_update, live, delayed, valid, and the selected fields; upsert them by topic. valid tracks source continuity: an unknown or invalidated image has valid: false and null fields, and recovery sends a fresh valid image. State mode rejects rows: false, streamCapacity, and overflowPolicy, and never reports a gap.
• overflowPolicy: 'drop_oldest' keeps sparse deltas but evicts the oldest pending updates when the buffer is full. The next envelope's gapBefore ({ cause: 'local_overflow', droppedUpdates }) holds the exact number of evicted updates as a bigint. A delta-derived image is discontinuous after a gap.

import { connect } from '@xbbg/core';

const engine = await connect();

// On-change board: full current images for the topics that changed
const board = await engine.subscribe(['IBM US Equity', 'MSFT US Equity'], ['BID', 'ASK', 'LAST_PRICE'], {
  delivery: 'state',
});
try {
  for await (const delivery of board.deliveries({ backend: 'json' })) {
    console.log(delivery.kind, delivery.data); // rows: topic, last_update, live, delayed, valid, fields
    break;
  }
} finally {
  await board.unsubscribe();
}

// Newest-pending deltas with exact local eviction counts
const deltas = await engine.subscribe(['ES1 Index'], ['BID', 'ASK'], { overflowPolicy: 'drop_oldest' });
try {
  for await (const delivery of deltas.deliveries()) {
    if (delivery.gapBefore !== undefined) {
      console.warn(`${delivery.gapBefore.droppedUpdates} updates dropped before this batch`);
    }
    console.log(delivery.data.numRows);
  }
} finally {
  await deltas.unsubscribe();
}

• sub.delivery reports the subscription's mode.
• A failed backend conversion leaves its batch unread, so another delivery view can retry it. await sub.deliveries().unsubscribe(true) returns the batches still pending; a plain unsubscribe() discards them.

Data loss

Delta subscriptions fail closed instead of silently skipping data. A Bloomberg DATALOSS event, a full buffer under 'drop_newest', or a 'block' forwarding overflow or timeout ends the subscription with BlpSubscriptionDataLossError (code: 'DATALOSS', plus topic and detail; topic is '*' for session-wide loss). Updates already queued are delivered first. Under 'drop_oldest', local overflow becomes a gapBefore count instead, but Bloomberg DATALOSS still ends the stream. Resubscribe to get a fresh image:

import { BlpSubscriptionDataLossError, connect } from '@xbbg/core';

const engine = await connect();

for (;;) {
  const sub = await engine.subscribe(['ES1 Index'], ['LAST_PRICE']);
  try {
    for await (const tick of sub) {
      console.log(tick.topic, tick.get('LAST_PRICE'));
    }
    break; // the subscription ended without data loss
  } catch (error) {
    if (!(error instanceof BlpSubscriptionDataLossError)) throw error;
    console.warn(`Data loss on ${error.topic ?? '*'}; resubscribing`);
  } finally {
    await sub.unsubscribe();
  }
}

Boards opened with rows: false and delivery: 'state' subscriptions recover on their own: after a Bloomberg DATALOSS, the shared feed resubscribes and live turns false at once. A board's image is cleared and rebuilt when the recovery paint arrives, so fields missing from the recovery are null; a state subscription delivers invalid images (valid: false, null fields) before the valid recovery images. The next ordinary update sets live again, and events records DataLoss and FeedRecovered. Session termination ends every subscription without automatic recovery, while ordinary connection-down events and failures of only some topics do not end the subscription.

Other services

import { connect } from '@xbbg/core';

const engine = await connect();

// //blp/mktvwap: explicit VWAP topics and the VWAP field, with an optional window
const vwap = await engine.vwap(['//blp/mktvwap/ticker/IBM US Equity'], ['VWAP'], {
  options: ['VWAP_START_TIME=09:30', 'VWAP_END_TIME=16:00'],
});

// //blp/mktbar: one explicit bar topic, LAST_PRICE, and the bar size in minutes
const bars = await engine.mktbar('//blp/mktbar/ticker/ES1 Index', {
  fields: ['LAST_PRICE'],
  options: ['bar_size=5'],
});

// B-PIPE only: //blp/mktdepthdata and //blp/mktlist
const depth = await engine.depth('AAPL US Equity');
const chain = await engine.chains('AAPL US Equity', { options: ['chainType=OPTIONS'] });

// Positional form for any service; options pass through unchanged
const intervalized = await engine.subscribeWithOptions(
  '//blp/mktdata',
  ['AAPL US Equity'],
  ['LAST_PRICE'],
  ['interval=5'],
);

await Promise.all([vwap, bars, depth, chain, intervalized].map((sub) => sub.unsubscribe()));

These helpers pass topics, fields, and options to Bloomberg as given. Unlike the Python helpers, they do not normalize //blp/mktvwap or //blp/mktbar topics, so supply the topic forms, fields, and options each service expects. Feeds on services other than //blp/mktdata are never shared.

Cleanup

• unsubscribe() closes the subscription and discards unread updates. unsubscribe(true) instead resolves to the updates still buffered (Tick[], or Table[] from the Arrow view after Arrow reads).
• Leaving a for await loop early, or calling return(), unsubscribes without draining.
• next({ signal }) accepts an AbortSignal. Aborting closes the whole subscription, not only that read, and rejects after cleanup with the abort reason.
• Closing is shared across the scalar, Arrow, and delivery views, and concurrent closes share one cleanup. If cleanup fails after a read error, the rejection is an AggregateError carrying both.
• Unsubscribe before engine.signalShutdown(); shutdown ends the engine's remaining subscriptions.

import { connect } from '@xbbg/core';

const engine = await connect();
const sub = await engine.subscribe(['AAPL US Equity'], ['LAST_PRICE']);
const signal = AbortSignal.timeout(30_000);

try {
  for (;;) {
    const result = await sub.next({ signal });
    if (result.done) break;
    console.log(result.value.toObject());
  }
} catch (error) {
  if (!signal.aborted) throw error;
} finally {
  await sub.unsubscribe();
}

Related pages

Configuration

Engine-wide stream capacity, overflow policy, and subscription session limits.

Recipes

Auction and iNAV subscriptions with venue and relationship preflight.

Python streaming

The same shared-feed engine from Python.

LangGraph Tools

Bounded snapshot tools for agents.