1
2
3
4
5
6
7
8
9
10
11

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,
)

1
2
3
4
5
6
7

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()

1
2
3
4

Live.add_callback(
    record_callback: Callable[[DBNRecord], None],
    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
28

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
)

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
)

1
2
3
4

Live.add_reconnect_callback(
    reconnect_callback: Callable[[pd.Timestamp, pd.Timestamp], None]
    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
28
29
30
31

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
)

1
2
3

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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()

1

Live.start() -> None

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

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()

1

Live.stop() -> None

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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()

1
2
3
4
5
6
7
8

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

1
2
3
4
5
6
7
8
9
10
11
12

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",
)

1

Live.terminate() -> None

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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()

1
2
3

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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()

1

Live.__aiter__() -> AsyncIterator[DBNRecord]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

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)

1

Live.__iter__() -> Iterator[DBNRecord]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

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)