Live
The live subpackage groups all engine and client implementations for live trading.
Generally a common event loop is passed into each live engine to support the overarching design of a single efficient event loop, by default uvloop.
The LiveDataClient class is responsible for interfacing with a particular API which may be presented directly by a venue, or through a broker intermediary.
It could also be possible to write clients for specialized data providers.
class LiveDataClient
Bases: DataClient
The base class for all live data clients.
- Parameters:
- loop (asyncio.AbstractEventLoop) – The event loop for the client.
- client_id (ClientId) – The client ID.
- venue (Venue or
None
) – The client venue. If multi-venue then can beNone
. - msgbus (MessageBus) – The message bus for the client.
- cache (Cache) – The cache for the client.
- clock (LiveClock) – The clock for the client.
- config (NautilusConfig , optional) – The configuration for the instance.
WARNING
This class should not be used directly, but through a concrete subclass.
async run_after_delay(delay: float, coro: Coroutine) → None
Run the given coroutine after a delay.
- Parameters:
- delay (float) – The delay (seconds) before running the coroutine.
- coro (Coroutine) – The coroutine to run after the initial delay.
create_task(coro: ~collections.abc.Coroutine, log_msg: str | None = None, actions: ~collections.abc.Callable | None = None, success_msg: str | None = None, success_color: ~nautilus_trader.core.rust.common.LogColor = <LogColor.NORMAL: 0>) → Task
Run the given coroutine with error handling and optional callback actions when done.
- Parameters:
- coro (Coroutine) – The coroutine to run.
- log_msg (str , optional) – The log message for the task.
- actions (Callable , optional) – The actions callback to run when the coroutine is done.
- success_msg (str , optional) – The log message to write on actions success.
- success_color (LogColor, default
NORMAL
) – The log message color for actions success.
- Return type: asyncio.Task
connect() → None
Connect the client.
disconnect() → None
Disconnect the client.
subscribe(self, DataType data_type) → void
Subscribe to data for the given data type.
- Parameters: data_type (DataType) – The data type for the subscription.
unsubscribe(self, DataType data_type) → void
Unsubscribe from data for the given data type.
- Parameters: data_type (DataType) – The data type for the subscription.
request(self, DataType data_type, UUID4 correlation_id) → void
Request data for the given data type.
- Parameters:
degrade(self) → void
Degrade the component.
While executing on_degrade() any exception will be logged and reraised, then the component
will remain in a DEGRADING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
dispose(self) → void
Dispose of the component.
While executing on_dispose() any exception will be logged and reraised, then the component
will remain in a DISPOSING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
fault(self) → void
Fault the component.
Calling this method multiple times has the same effect as calling it once (it is idempotent). Once called, it cannot be reversed, and no other methods should be called on this instance.
While executing on_fault() any exception will be logged and reraised, then the component
will remain in a FAULTING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
classmethod fully_qualified_name(cls) → str
Return the fully qualified name for the components class.
- Return type: str
id
The components ID.
- Returns: ComponentId
is_connected
If the client is connected.
- Returns: bool
is_degraded
bool
Return whether the current component state is DEGRADED
.
- Return type: bool
- Type: Component.is_degraded
is_disposed
bool
Return whether the current component state is DISPOSED
.
- Return type: bool
- Type: Component.is_disposed
is_faulted
bool
Return whether the current component state is FAULTED
.
- Return type: bool
- Type: Component.is_faulted
is_initialized
bool
Return whether the component has been initialized (component.state >= INITIALIZED
).
- Return type: bool
- Type: Component.is_initialized
is_running
bool
Return whether the current component state is RUNNING
.
- Return type: bool
- Type: Component.is_running
is_stopped
bool
Return whether the current component state is STOPPED
.
- Return type: bool
- Type: Component.is_stopped
reset(self) → void
Reset the component.
All stateful fields are reset to their initial value.
While executing on_reset() any exception will be logged and reraised, then the component
will remain in a RESETTING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
resume(self) → void
Resume the component.
While executing on_resume() any exception will be logged and reraised, then the component
will remain in a RESUMING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
start(self) → void
Start the component.
While executing on_start() any exception will be logged and reraised, then the component
will remain in a STARTING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
state
ComponentState
Return the components current state.
- Return type: ComponentState
- Type: Component.state
stop(self) → void
Stop the component.
While executing on_stop() any exception will be logged and reraised, then the component
will remain in a STOPPING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
subscribed_custom_data(self) → list
Return the custom data types subscribed to.
- Return type: list[DataType]
trader_id
The trader ID associated with the component.
- Returns: TraderId
type
The components type.
- Returns: type
venue
The clients venue ID (if applicable).
- Returns:
Venue or
None
class LiveMarketDataClient
Bases: MarketDataClient
The base class for all live data clients.
- Parameters:
- loop (asyncio.AbstractEventLoop) – The event loop for the client.
- client_id (ClientId) – The client ID.
- venue (Venue or
None
) – The client venue. If multi-venue then can beNone
. - msgbus (MessageBus) – The message bus for the client.
- cache (Cache) – The cache for the client.
- clock (LiveClock) – The clock for the client.
- instrument_provider (InstrumentProvider) – The instrument provider for the client.
- config (NautilusConfig , optional) – The configuration for the instance.
WARNING
This class should not be used directly, but through a concrete subclass.
async run_after_delay(delay: float, coro: Coroutine) → None
Run the given coroutine after a delay.
- Parameters:
- delay (float) – The delay (seconds) before running the coroutine.
- coro (Coroutine) – The coroutine to run after the initial delay.
create_task(coro: ~collections.abc.Coroutine, log_msg: str | None = None, actions: ~collections.abc.Callable | None = None, success_msg: str | None = None, success_color: ~nautilus_trader.core.rust.common.LogColor = <LogColor.NORMAL: 0>) → Task
Run the given coroutine with error handling and optional callback actions when done.
- Parameters:
- coro (Coroutine) – The coroutine to run.
- log_msg (str , optional) – The log message for the task.
- actions (Callable , optional) – The actions callback to run when the coroutine is done.
- success_msg (str , optional) – The log message to write on actions success.
- success_color (LogColor, default
NORMAL
) – The log message color for actions success.
- Return type: asyncio.Task
connect() → None
Connect the client.
disconnect() → None
Disconnect the client.
subscribe(self, DataType data_type) → void
Subscribe to data for the given data type.
- Parameters: data_type (DataType) – The data type for the subscription.
subscribe_instruments(self) → void
Subscribe to all Instrument data.
subscribe_instrument(self, InstrumentId instrument_id) → void
Subscribe to the Instrument with the given instrument ID.
subscribe_order_book_deltas(self, InstrumentId instrument_id, BookType book_type, int depth=0, dict kwargs=None) → void
Subscribe to OrderBookDeltas data for the given instrument ID.
- Parameters:
- instrument_id (InstrumentId) – The order book instrument to subscribe to.
- book_type (BookType {
L1_MBP
,L2_MBP
,L3_MBO
}) – The order book type. - depth (int , optional , default None) – The maximum depth for the subscription.
- kwargs (dict , optional) – The keyword arguments for exchange specific parameters.
subscribe_order_book_snapshots(self, InstrumentId instrument_id, BookType book_type, int depth=0, dict kwargs=None) → void
Subscribe to OrderBook snapshots data for the given instrument ID.
- Parameters:
- instrument_id (InstrumentId) – The order book instrument to subscribe to.
- book_type (BookType {
L1_MBP
,L2_MBP
,L3_MBO
}) – The order book level. - depth (int , optional) – The maximum depth for the order book. A depth of 0 is maximum depth.
- kwargs (dict , optional) – The keyword arguments for exchange specific parameters.
subscribe_quote_ticks(self, InstrumentId instrument_id) → void
Subscribe to QuoteTick data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The tick instrument to subscribe to.
subscribe_trade_ticks(self, InstrumentId instrument_id) → void
Subscribe to TradeTick data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The tick instrument to subscribe to.
subscribe_bars(self, BarType bar_type) → void
Subscribe to Bar data for the given bar type.
- Parameters: bar_type (BarType) – The bar type to subscribe to.
subscribe_instrument_status(self, InstrumentId instrument_id) → void
Subscribe to InstrumentStatus data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The tick instrument to subscribe to.
subscribe_instrument_close(self, InstrumentId instrument_id) → void
Subscribe to InstrumentClose updates for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The tick instrument to subscribe to.
unsubscribe(self, DataType data_type) → void
Unsubscribe from data for the given data type.
- Parameters: data_type (DataType) – The data type for the subscription.
unsubscribe_instruments(self) → void
Unsubscribe from all Instrument data.
unsubscribe_instrument(self, InstrumentId instrument_id) → void
Unsubscribe from Instrument data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The instrument to unsubscribe from.
unsubscribe_order_book_deltas(self, InstrumentId instrument_id) → void
Unsubscribe from OrderBookDeltas data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The order book instrument to unsubscribe from.
unsubscribe_order_book_snapshots(self, InstrumentId instrument_id) → void
Unsubscribe from OrderBook snapshots data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The order book instrument to unsubscribe from.
unsubscribe_quote_ticks(self, InstrumentId instrument_id) → void
Unsubscribe from QuoteTick data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The tick instrument to unsubscribe from.
unsubscribe_trade_ticks(self, InstrumentId instrument_id) → void
Unsubscribe from TradeTick data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The tick instrument to unsubscribe from.
unsubscribe_bars(self, BarType bar_type) → void
Unsubscribe from Bar data for the given bar type.
- Parameters: bar_type (BarType) – The bar type to unsubscribe from.
unsubscribe_instrument_status(self, InstrumentId instrument_id) → void
Unsubscribe from InstrumentStatus data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The instrument status updates to unsubscribe from.
unsubscribe_instrument_close(self, InstrumentId instrument_id) → void
Unsubscribe from InstrumentClose data for the given instrument ID.
- Parameters: instrument_id (InstrumentId) – The tick instrument to unsubscribe from.
request(self, DataType data_type, UUID4 correlation_id) → void
Request data for the given data type.
- Parameters:
request_instrument(self, InstrumentId instrument_id, UUID4 correlation_id, datetime start=None, datetime end=None) → void
Request Instrument data for the given instrument ID.
- Parameters:
- instrument_id (InstrumentId) – The instrument ID for the request.
- correlation_id (UUID4) – The correlation ID for the request.
- start (datetime , optional) – The start datetime (UTC) of request time range (inclusive).
- end (datetime , optional) – The end datetime (UTC) of request time range. The inclusiveness depends on individual data client implementation.
request_instruments(self, Venue venue, UUID4 correlation_id, datetime start=None, datetime end=None) → void
Request all Instrument data for the given venue.
- Parameters:
- venue (Venue) – The venue for the request.
- correlation_id (UUID4) – The correlation ID for the request.
- start (datetime , optional) – The start datetime (UTC) of request time range (inclusive).
- end (datetime , optional) – The end datetime (UTC) of request time range. The inclusiveness depends on individual data client implementation.
request_quote_ticks(self, InstrumentId instrument_id, int limit, UUID4 correlation_id, datetime start=None, datetime end=None) → void
Request historical QuoteTick data.
- Parameters:
- instrument_id (InstrumentId) – The tick instrument ID for the request.
- limit (int) – The limit for the number of returned ticks.
- correlation_id (UUID4) – The correlation ID for the request.
- start (datetime , optional) – The start datetime (UTC) of request time range (inclusive).
- end (datetime , optional) – The end datetime (UTC) of request time range. The inclusiveness depends on individual data client implementation.
request_trade_ticks(self, InstrumentId instrument_id, int limit, UUID4 correlation_id, datetime start=None, datetime end=None) → void
Request historical TradeTick data.
- Parameters:
- instrument_id (InstrumentId) – The tick instrument ID for the request.
- limit (int) – The limit for the number of returned ticks.
- correlation_id (UUID4) – The correlation ID for the request.
- start (datetime , optional) – The start datetime (UTC) of request time range (inclusive).
- end (datetime , optional) – The end datetime (UTC) of request time range. The inclusiveness depends on individual data client implementation.
request_bars(self, BarType bar_type, int limit, UUID4 correlation_id, datetime start=None, datetime end=None) → void
Request historical Bar data.
- Parameters:
- bar_type (BarType) – The bar type for the request.
- limit (int) – The limit for the number of returned bars.
- correlation_id (UUID4) – The correlation ID for the request.
- start (datetime , optional) – The start datetime (UTC) of request time range (inclusive).
- end (datetime , optional) – The end datetime (UTC) of request time range. The inclusiveness depends on individual data client implementation.
request_order_book_snapshot(self, InstrumentId instrument_id, int limit, UUID4 correlation_id) → void
Request order book snapshot data.
- Parameters:
- instrument_id (InstrumentId) – The instrument ID for the order book snapshot request.
- limit (int) – The limit on the depth of the order book snapshot.
- correction_id (UUID4) – The correlation ID for the request.
degrade(self) → void
Degrade the component.
While executing on_degrade() any exception will be logged and reraised, then the component
will remain in a DEGRADING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
dispose(self) → void
Dispose of the component.
While executing on_dispose() any exception will be logged and reraised, then the component
will remain in a DISPOSING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
fault(self) → void
Fault the component.
Calling this method multiple times has the same effect as calling it once (it is idempotent). Once called, it cannot be reversed, and no other methods should be called on this instance.
While executing on_fault() any exception will be logged and reraised, then the component
will remain in a FAULTING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
classmethod fully_qualified_name(cls) → str
Return the fully qualified name for the components class.
- Return type: str
id
The components ID.
- Returns: ComponentId
is_connected
If the client is connected.
- Returns: bool
is_degraded
bool
Return whether the current component state is DEGRADED
.
- Return type: bool
- Type: Component.is_degraded
is_disposed
bool
Return whether the current component state is DISPOSED
.
- Return type: bool
- Type: Component.is_disposed
is_faulted
bool
Return whether the current component state is FAULTED
.
- Return type: bool
- Type: Component.is_faulted
is_initialized
bool
Return whether the component has been initialized (component.state >= INITIALIZED
).
- Return type: bool
- Type: Component.is_initialized
is_running
bool
Return whether the current component state is RUNNING
.
- Return type: bool
- Type: Component.is_running
is_stopped
bool
Return whether the current component state is STOPPED
.
- Return type: bool
- Type: Component.is_stopped
reset(self) → void
Reset the component.
All stateful fields are reset to their initial value.
While executing on_reset() any exception will be logged and reraised, then the component
will remain in a RESETTING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
resume(self) → void
Resume the component.
While executing on_resume() any exception will be logged and reraised, then the component
will remain in a RESUMING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
start(self) → void
Start the component.
While executing on_start() any exception will be logged and reraised, then the component
will remain in a STARTING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
state
ComponentState
Return the components current state.
- Return type: ComponentState
- Type: Component.state
stop(self) → void
Stop the component.
While executing on_stop() any exception will be logged and reraised, then the component
will remain in a STOPPING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
subscribed_bars(self) → list
Return the bar types subscribed to.
- Return type: list[BarType]
subscribed_custom_data(self) → list
Return the custom data types subscribed to.
- Return type: list[DataType]
subscribed_instrument_close(self) → list
Return the instrument closes subscribed to.
- Return type: list[InstrumentId]
subscribed_instrument_status(self) → list
Return the status update instruments subscribed to.
- Return type: list[InstrumentId]
subscribed_instruments(self) → list
Return the instruments subscribed to.
- Return type: list[InstrumentId]
subscribed_order_book_deltas(self) → list
Return the order book delta instruments subscribed to.
- Return type: list[InstrumentId]
subscribed_order_book_snapshots(self) → list
Return the order book snapshot instruments subscribed to.
- Return type: list[InstrumentId]
subscribed_quote_ticks(self) → list
Return the quote tick instruments subscribed to.
- Return type: list[InstrumentId]
subscribed_trade_ticks(self) → list
Return the trade tick instruments subscribed to.
- Return type: list[InstrumentId]
trader_id
The trader ID associated with the component.
- Returns: TraderId
type
The components type.
- Returns: type
venue
The clients venue ID (if applicable).
- Returns:
Venue or
None
class LiveDataEngine
Bases: DataEngine
Provides a high-performance asynchronous live data engine.
- Parameters:
- loop (asyncio.AbstractEventLoop) – The event loop for the engine.
- msgbus (MessageBus) – The message bus for the engine.
- cache (Cache) – The cache for the engine.
- clock (LiveClock) – The clock for the engine.
- config (LiveDataEngineConfig , optional) – The configuration for the instance.
- Raises: TypeError – If config is not of type LiveDataEngineConfig.
connect() → None
Connect the engine by calling connect on all registered clients.
disconnect() → None
Disconnect the engine by calling disconnect on all registered clients.
get_cmd_queue_task() → Task | None
Return the internal command queue task for the engine.
- Return type:
asyncio.Task or
None
get_req_queue_task() → Task | None
Return the internal request queue task for the engine.
- Return type:
asyncio.Task or
None
get_res_queue_task() → Task | None
Return the internal response queue task for the engine.
- Return type:
asyncio.Task or
None
get_data_queue_task() → Task | None
Return the internal data queue task for the engine.
- Return type:
asyncio.Task or
None
cmd_qsize() → int
Return the number of DataCommand objects buffered on the internal queue.
- Return type: int
req_qsize() → int
Return the number of DataRequest objects buffered on the internal queue.
- Return type: int
res_qsize() → int
Return the number of DataResponse objects buffered on the internal queue.
- Return type: int
data_qsize() → int
Return the number of Data objects buffered on the internal queue.
- Return type: int
kill() → None
Kill the engine by abruptly canceling the queue tasks and calling stop.
execute(command: DataCommand) → None
Execute the given data command.
If the internal queue is already full then will log a warning and block until queue size reduces.
- Parameters: command (DataCommand) – The command to execute.
WARNING
This method is not thread-safe and should only be called from the same thread the event loop is running on. Calling it from a different thread may lead to unexpected behavior.
request(request: DataRequest) → None
Handle the given request.
If the internal queue is already full then will log a warning and block until queue size reduces.
- Parameters: request (DataRequest) – The request to handle.
WARNING
This method is not thread-safe and should only be called from the same thread the event loop is running on. Calling it from a different thread may lead to unexpected behavior.
response(response: DataResponse) → None
Handle the given response.
If the internal queue is already full then will log a warning and block until queue size reduces.
- Parameters: response (DataResponse) – The response to handle.
WARNING
This method is not thread-safe and should only be called from the same thread the event loop is running on. Calling it from a different thread may lead to unexpected behavior.
process(data: Data) → None
Process the given data.
If the internal queue is already full then will log a warning and block until queue size reduces.
- Parameters: data (Data) – The data to process.
WARNING
This method is not thread-safe and should only be called from the same thread the event loop is running on. Calling it from a different thread may lead to unexpected behavior.
check_connected(self) → bool
Check all of the engines clients are connected.
- Returns: True if all clients connected, else False.
- Return type: bool
check_disconnected(self) → bool
Check all of the engines clients are disconnected.
- Returns: True if all clients disconnected, else False.
- Return type: bool
command_count
The total count of data commands received by the engine.
- Returns: int
data_count
The total count of data stream objects received by the engine.
- Returns: int
debug
If debug mode is active (will provide extra debug logging).
- Returns: bool
default_client
ClientId | None
Return the default data client registered with the engine.
- Return type:
ClientId or
None
- Type: DataEngine.default_client
degrade(self) → void
Degrade the component.
While executing on_degrade() any exception will be logged and reraised, then the component
will remain in a DEGRADING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.
deregister_client(self, DataClient client) → void
Deregister the given data client from the data engine.
- Parameters: client (DataClient) – The data client to deregister.
dispose(self) → void
Dispose of the component.
While executing on_dispose() any exception will be logged and reraised, then the component
will remain in a DISPOSING
state.
WARNING
Do not override.
If the component is not in a valid state from which to execute this method, then the component state will not change, and an error will be logged.