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.
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 for more information.
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. |
1 2 3 4
Live.add_stream(
stream: IO[bytes] | PathLike[str] | str,
exception_callback: Callable[[Exception], None] | None = None,
) -> None 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
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
)