Source: https://databento.com/docs/api-live/client


## Client


### Live

To access Databento's live API, first create an instance of the
`Live` client. The entire API for a streaming session is exposed
through instance methods of the client.

Databento's `Live` client is built with Python's [asyncio](https://docs.python.org/3/library/asyncio.html) module and can be
easily integrated into asynchronous applications.

Note that the API key can be passed as a parameter, which is
[not recommended for production applications](https://databento.com/docs/portal/api-keys.md#securing-your-api-keys).
Instead, you can leave out this parameter to pass your API key via the `DATABENTO_API_KEY` environment variable:

#### API method

```python
class Live(
    key: str | None = None,
    gateway: str | None = None,
    port: int | None = None,
    ts_out: bool = False,
    heartbeat_interval_s: int | None = None,
    reconnect_policy: ReconnectPolicy | str = ReconnectPolicy.NONE,
    slow_reader_behavior: SlowReaderBehavior | str | None = None,
    compression: Compression | str = Compression.NONE,
    loop: asyncio.AbstractEventLoop | None = None,
)
```

#### Example usage

```python
import databento as db

# Pass as parameter
client = db.Live(key="$YOUR_API_KEY")

# Or, pass as `DATABENTO_API_KEY` environment variable
client = db.Live()
```
#### Parameters

| Parameter | Type | Required | Description |
| `key` | str | optional | 32-character API key. Found on your [API keys]($PLATFORM_HOSTNAME/portal/keys) page. If `None` then `DATABENTO_API_KEY` environment variable is used. |
| `gateway` | str | optional | The url of the remote gateway to connect to. This is for advanced use. |
| `port` | int | optional | The port to connect to the remote gateway on. This is for advanced use. |
| `ts_out` | bool, default False | optional | Whether the gateway should append the send timestamp (`ts_out`) to each record. |
| `heartbeat_interval_s` | int, default None | optional | The interval in seconds at which the gateway will send [heartbeat records](https://databento.com/docs/api-live/basics/system-messages.md) if no other data records are sent. By default heartbeats will be sent at the gateway's default interval. Must be between 5 and 1800 seconds. |
| `reconnect_policy` | ReconnectPolicy or str, default None | optional | The reconnection policy for automatically resuming a live session when an unexpected disconnection occurs. |
| `slow_reader_behavior` | SlowReaderBehavior or str, default None | optional | How the gateway should handle a connection unable to keep up with realtime. |
| `compression` | Compression or str, default None | optional | Sets the [compression](https://databento.com/docs/api-live/basics/compression.md) mode for the read stream. Uncompressed by default. |
| `loop` | asyncio.AbstractEventLoop, default None | optional | The event loop to run the client connection in. If `None`, a shared event loop running on a background thread will be used. |


### Live.add\_callback

Add a callback to the live client. This callback must take a single record argument.
Refer to the [What's a schema](https://databento.com/docs/schemas/whats-a-schema.md) article for documentation on the fields contained with each record type.

A callback will receive error messages from the gateway with the `ErrorMsg` record type.
This indicates a problem occurred with the session.
Exceptions raised in a callback can be handled explicitly by specifying an `exception_callback`.

The callbacks are executed in the order they are added.

> **Info**
>
> It is recommended that callback functions be non-blocking to ensure the networking event loop remains running.

#### API method

```python
Live.add_callback(
    record_callback: Callable[[DBNRecord], None],
    exception_callback: Callable[[Exception], None] | None = None,
) -> None
```

#### Parameters

| Parameter | Type | Required | Description |
| `record_callback` | Callable[[DBNRecord], None] | required | A callback to dispatch records to. The callback must take a single record argument. If no exception callback is provided, any exceptions encountered will be logged and raised as warnings for visibility. |
| `exception_callback` | Callable[[Exception], None] | optional | A callback to dispatch exceptions to that are raised when executing `record_callback`. The callback must take a single exception argument. |

#### Example usage

```python
import databento as db

# Create a callback to handle DBN records
def user_callback(record: db.DBNRecord) -> None:
    print(f"callback run for {record}")

# Create a callback to handle exceptions from `user_callback`
def error_handler(exception: Exception) -> None:
    print(f"an error occurred {exception}")

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Add the user_callback to the client
client.add_callback(
    record_callback=user_callback,
    exception_callback=error_handler,  # optional error handler
)
```


### Live.add\_stream

Add an output stream to the live client. The client will write binary DBN records to the stream.
This stream must be writable in bytes mode, or a path to a file that is writable.

A stream will receive error messages from the gateway with the `ErrMsg` record type.
This indicates a problem occurred with the session.
Exceptions raised in a callback can be handled explicitly by specifying an `exception_callback`.

The writes are performed in the order the streams were added.

> **See also**
>
> DBN records are optimized for stream-like processing.
>
> While pandas DataFrames are not well-suited for this due to their column-oriented format,
> they can still be used by first streaming DBN data to a file, then converting to a DataFrame with DBNStore.from_file().to_df().
> See [this example](https://databento.com/docs/examples/basics-live/live-stream-to-file.md) for more information.

#### API method

```python
Live.add_stream(
    stream: IO[bytes] | PathLike[str] | str,
    exception_callback: Callable[[Exception], None] | None = None,
) -> None
```

#### Parameters

| Parameter | Type | Required | Description |
| `stream` | IO[bytes] or PathLike[str] or str | required | An IO stream to write records to, or a writable file path. The stream must support writing bytes. |
| `exception_callback` | Callable[[Exception], None] | optional | An error handling callback to process exceptions that are raised when writing to the stream. The callback must take a single exception argument. If no exception callback is provided, any exceptions encountered will be logged and raised as warnings for visibility. |

#### Example usage

```python
import databento as db

# Open a file for writing
output = open("output.dbn", mode="wb")

# Create a callback to handle exceptions when writing to `output`
def error_handler(exception: Exception) -> None:
    print(f"an error occurred {exception}")

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Add the output file stream to the client
client.add_stream(
    stream=output,
    exception_callback=error_handler,  # optional error handler
)
```


### Live.add\_reconnect\_callback

Add a reconnect callback to the live client. This callback must take two arguments.
When a reconnection policy is set on the `Live` client, this callback will be executed when a reconnection occurs with two arguments:
- The last received record's `ts_event` timestamp, or `start` timestamp from the session `Metadata` if no records were received yet.
- The `start` timestamp from the reconnected session's `Metadata`.

These values can be used to record any gaps in the received data due to a disconnection.
Exceptions raised in a callback can be handled explicitly by specifying an `exception_callback`.

The callbacks are executed in the order they are added.

> **Info**
> It is recommended that callback functions be non-blocking to ensure the networking event loop remains running.

#### API method

```python
Live.add_reconnect_callback(
    reconnect_callback: Callable[[pd.Timestamp, pd.Timestamp], None]
    exception_callback: Callable[[Exception], None] | None = None,
) -> None
```

#### Parameters

| Parameter | Type | Required | Description |
| `reconnect_callback` | Callable[[pd.Timestamp, pd.Timestamp], None] | required | A callback for handling reconnections. The callback must take two arguments. |
| `exception_callback` | Callable[[Exception], None] | optional | A callback to dispatch exceptions to that are raised when executing `reconnect_callback`. The callback must take a single exception argument. |

#### Example usage

```python
import databento as db

# Create a callback to handle reconnections
def reconnect_callback(start, end) -> None:
    print(f"reconnection gap from {start} to {end}")

# Create a callback to handle exceptions from `reconnect_callback`
def error_handler(exception: Exception) -> None:
    print(f"an error occurred {exception}")

# Create a live client with a reconnect policy
client = db.Live(
    key="$YOUR_API_KEY",
    reconnect_policy="reconnect",
)

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Add the reconnect callback to the client
client.add_reconnect_callback(
    reconnect_callback=reconnect_callback,
    exception_callback=error_handler,  # optional error handler
)
```


### Live.block\_for\_close

Block until the session closes or a timeout is reached.
A session will close when the remote gateway disconnects or after [Live.stop](https://databento.com/docs/api-live/client/stop.md) or [Live.terminate](https://databento.com/docs/api-live/client/terminate.md) are called.
In the event the connection is closed unexpectedly, a `BentoError` will be raised.

If a `timeout` is specified, [Live.terminate](https://databento.com/docs/api-live/client/terminate.md) will be called when the timeout is reached.

When this method unblocks, the session is guaranteed to be closed.

#### API method

```python
Live.block_for_close(
    timeout: float | None = None,
) -> None
```

#### Parameters

| Parameter | Type | Required | Description |
| `timeout` | float | optional | A duration in seconds to wait for the session to close. If `None`, wait forever. |

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Start the session
client.start()

# Block for the session to close
client.block_for_close()
```


### Live.start

Start the session.

A client can only be started once, and after a successful connection is made by calling [Live.subscribe](https://databento.com/docs/api-live/client/subscribe.md).

A client can be gracefully stopped by calling [Live.stop](https://databento.com/docs/api-live/client/stop.md) or forcefully using [Live.terminate](https://databento.com/docs/api-live/client/terminate.md)

When iterating records with [Live.\_\_iter\_\_](https://databento.com/docs/api-live/client/iter.md) or [Live.\_\_aiter\_\_](https://databento.com/docs/api-live/client/aiter.md) it is not necessary to call [Live.start](https://databento.com/docs/api-live/client/start.md) and doing so before iterating will raise a `ValueError`.

> **Info**
>
> A session cannot be started more than once.

#### API method

```python
Live.start() -> None
```

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Print records as they arrive
client.add_callback(print)

# Start the streaming session
client.start()

# Block until disconnection
client.block_for_close()
```



### Live.stop

Stop the session and finish processing received records.

A client can only be stopped after a successful connection is made by calling [Live.subscribe](https://databento.com/docs/api-live/client/subscribe.md).

This method does not block waiting for the connection to close.
If this is desired, use [Live.block\_for\_close](https://databento.com/docs/api-live/client/block-for-close.md) or [Live.wait\_for\_close](https://databento.com/docs/api-live/client/wait-for-close.md) to wait for the client to disconnect.

The connection will eventually close after calling this method.
Once the connection has closed, the `Live` client can be reused but the session state is not preserved.

#### API method

```python
Live.stop() -> None
```

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Start the streaming session
client.start()

# Stop the streaming session
client.stop()
```


### Live.subscribe

Add a new subscription to the session.
All subscriptions must be for the same `dataset`.
The returned integer is the `subscription_id` which will match the id used by the gateway when acknowledging the subscription.

Supports multiple subscriptions for different schemas, which enables rich data streams containing mixed record types.

Specify an optional `start` time for [intraday replay](https://databento.com/docs/api-live/basics/intraday-replay.md) with subscriptions made before starting the session.

Please note there is no unsubscribe method.
Subscriptions end when the client disconnects or by calling [Live.stop](https://databento.com/docs/api-live/client/stop.md) or [Live.terminate](https://databento.com/docs/api-live/client/terminate.md)

#### API method

```python
Live.subscribe(
    dataset: Dataset | str,
    schema: Schema | str,
    symbols: Iterable[str | int] | str | int = "ALL_SYMBOLS",
    stype_in: SType | str = "raw_symbol",
    start: pd.Timestamp | datetime | date | str | int | None = None,
    snapshot: bool = False,
) -> int
```

#### Parameters

| Parameter | Type | Required | Description |
| `dataset` | Dataset or str | required | The [dataset](https://databento.com/docs/api-historical/basics/datasets.md) code (string identifier). Must be one of the values from [list\_datasets](https://databento.com/docs/api-historical/metadata/metadata-list-datasets.md). |
| `schema` | Schema or str | required | The data record [schema](https://databento.com/docs/schemas/whats-a-schema.md). Must be one of the values from [list\_schemas](https://databento.com/docs/api-historical/metadata/metadata-list-schemas.md). |
| `symbols` | Iterable[str \ | int] or str or int | optional | The product symbols to subscribe to. If `'ALL_SYMBOLS'` then will select **all** symbols. |
| `stype_in` | SType or str, default 'raw_symbol' | optional | The [symbology type](https://databento.com/docs/api-historical/basics/symbology.md) of input `symbols`. Must be one of 'raw_symbol', 'instrument_id', 'parent', or 'continuous'. |
| `start` | pd.Timestamp, datetime, date, str or int | optional | The inclusive start of subscription replay. Filters on `ts_event` except for the CBBO and BBO schemas, which filter on `ts_recv`. Takes [pd.Timestamp](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.Timestamp.html), [Python datetime](https://docs.python.org/3/library/datetime.html#datetime-objects), [Python date](https://docs.python.org/3/library/datetime.html#date-objects), [ISO 8601](https://www.iso.org/iso-8601-date-and-time-format.html) string, or [UNIX timestamp](https://en.wikipedia.org/wiki/Unix_time) in nanoseconds. Assumes UTC as timezone unless otherwise specified. Must be within the last 24 hours. Pass `0` to request all available data. Cannot be specified after the [session](https://databento.com/docs/api-live/basics/sessions.md) is [started](https://databento.com/docs/api-live/client/start.md). |
| `snapshot` | bool, default False | optional | Request subscription with [snapshot](https://databento.com/docs/conventions/mbo-snapshot.md). Only supported with `mbo` schema. Defaults to `False`. Conflicts with the `start` parameter. |

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)
```


### Live.terminate

Terminate the session and stop processing records immediately.

A client can only be terminated after a successful connection is made by calling [Live.subscribe](https://databento.com/docs/api-live/client/subscribe.md).

Unlike [Live.stop](https://databento.com/docs/api-live/client/stop.md), the session will end immediately and received records will no longer be processed.

Once terminated, the `Live` client can be reused but the session state is not preserved.

#### API method

```python
Live.terminate() -> None
```

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Start the streaming session
client.start()

# Terminate the streaming session
client.terminate()
```


### Live.wait\_for\_close

Wait until the session closes or a timeout is reached.
A session will close when the remote gateway disconnects, or after [Live.stop](https://databento.com/docs/api-live/client/terminate.md) or [Live.terminate](https://databento.com/docs/api-live/client/terminate.md) are called.
In the event the connection is closed unexpectedly, a `BentoError` will be raised.

If a `timeout` is specified, [Live.terminate](https://databento.com/docs/api-live/client/terminate.md) will be called when the timeout is reached.

When this method unblocks, the session is guaranteed to be closed.

#### API method

```python
Live.wait_for_close(
    timeout: float | None = None,
) -> None
```

> **Info**
>
> This method is a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutines) and must be used with an [await](https://docs.python.org/3/reference/expressions.html#await) expression.

#### Parameters

| Parameter | Type | Required | Description |
| `timeout` | float | optional | A duration in seconds to wait for the session to close. If `None`, wait forever. |

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to the trades schema for all ES futures
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Start the session
client.start()

# Wait for the session to end
await client.wait_for_close()
```


### Live.\_\_aiter\_\_
Using [async for](https://docs.python.org/3/reference/compound_stmts.html#the-async-for-statement); records will be yielded as they arrive.
Iteration will stop when the connection is closed and all records are processed.
This is best for integration into asynchronous applications.

Asynchronous iteration is only supported inside a [coroutine](https://docs.python.org/3/library/asyncio-task.html#coroutines).

Asynchronous iteration will automatically call [Live.start](https://databento.com/docs/api-live/client/start.md) if the session is connected but it has not been started.
Starting iteration after the session has started will cause a `ValueError`.

Asynchronous iteration will automatically call [Live.stop](https://databento.com/docs/api-live/client/stop.md) if the iterator is destroyed, such as when an async for loop is escaped with an exception or `break` statement.
To prevent this behavior, an iterator can be created explicitly using [aiter()](https://docs.python.org/3/library/functions.html#aiter).

#### API method

```python
Live.__aiter__() -> AsyncIterator[DBNRecord]
```

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to a data stream
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Start streaming and asynchronously iterate the records
async for record in client:
    print(record)
```


### Live.\_\_iter\_\_
Using [for](https://docs.python.org/3/reference/compound_stmts.html#the-for-statement); records will be yielded as they arrive.
Iteration will stop when the connection is closed and all records are processed.
This is best for integration into simple synchronous applications.

Iteration will automatically call [Live.start](https://databento.com/docs/api-live/client/start.md) if the session is connected but it has not been started.
Starting iteration after the session has started will cause a `ValueError`.

Iteration will automatically call [Live.stop](https://databento.com/docs/api-live/client/stop.md) if the iterator is destroyed, such as when a for loop is escaped with an exception or `break` statement.
To prevent this behavior, an iterator can be created explicitly using [iter()](https://docs.python.org/3/library/functions.html#iter).

#### API method

```python
Live.__iter__() -> Iterator[DBNRecord]
```

#### Example usage

```python
import databento as db

# Create a live client
client = db.Live(key="$YOUR_API_KEY")

# Subscribe to a data stream
client.subscribe(
    dataset="GLBX.MDP3",
    schema="trades",
    stype_in="parent",
    symbols="ES.FUT",
)

# Start streaming and synchronously iterate the records
for record in client:
    print(record)
```