Actors

A data actor receives requested and subscribed data, handles system events, and manages component state. In Python, extend the DataActor class; in Rust, implement the DataActor trait. A strategy adds order‑management capabilities.

Capabilities:

• Market and custom data subscriptions and requests.
• Custom data and signal publishing.
• Event, timer, and alert handling.
• Cache access.
• Structured logging.

Basic Python example

Actors support configuration through a pattern similar to strategies.

from nautilus_trader.common import DataActor
from nautilus_trader.config import DataActorConfig
from nautilus_trader.model import Bar
from nautilus_trader.model import BarType


class MyActorConfig(DataActorConfig):
    def __init__(self, bar_type: BarType, **_kwargs) -> None:
        self.bar_type = bar_type


class MyActor(DataActor):
    def __init__(self, config: MyActorConfig) -> None:
        super().__init__(config)

        # Keep runtime state on the actor
        self.count_of_processed_bars: int = 0

    def on_start(self) -> None:
        # Subscribe to bars matching the configured bar type
        self.subscribe_bars(self.config.bar_type)

    def on_bar(self, bar: Bar) -> None:
        self.count_of_processed_bars += 1

Actor configuration and IDs

Data actors can receive a DataActorConfig subclass. The base config accepts an optional actor_id. If supplied, the actor registers with that ID; otherwise a Python actor registers under its class name. Give each instance an explicit actor_id when running more than one instance of the same actor, because a duplicate ID is rejected at registration (a RuntimeError in Python).

Treat configuration as construction data for the actor. Read user‑supplied settings through self.config, and keep runtime state on the actor itself.

:::info Rust implementation Rust actors store runtime identity and state in DataActorCore. Read the runtime ID through actor_id() rather than expecting a generated ID to be written back into DataActorConfig. A Rust actor without a configured actor_id registers as DataActor whatever its type, so give each Rust actor an explicit actor_id.

Rust authors implement DataActor and use the facade methods on self. DataActorNative is native‑only access for runtime wiring and borrowed core state. Import it only for same‑binary performance paths or internal runtime wiring. :::

Lifecycle

Actors move through the main stable states shown below:

Main flow only: transitional states and less common valid edges are omitted. For actions with a lifecycle handler, the actor reaches the destination state only after that handler succeeds.

Override these methods to hook into lifecycle events:

Timers and alerts

Actors have access to a clock for scheduling:

from datetime import timedelta

from nautilus_trader.common import TimeEvent


def on_start(self) -> None:
    self._schedule_clock_events()


def on_resume(self) -> None:
    self._schedule_clock_events()


def on_stop(self) -> None:
    self._cancel_clock_events()


def on_degrade(self) -> None:
    self._cancel_clock_events()


def _cancel_clock_events(self) -> None:
    self.clock.cancel_timer("my_actor.timer")
    self.clock.cancel_timer("my_actor.alert")


def _schedule_clock_events(self) -> None:
    # Set a recurring timer with a callback that fires every 5 seconds
    self.clock.set_timer(
        "my_actor.timer",
        timedelta(seconds=5),
        callback=self._on_timer,
    )

    # Set a one-time alert with a callback
    self.clock.set_time_alert(
        "my_actor.alert",
        self.clock.utc_now() + timedelta(minutes=1),
        callback=self._on_alert,
    )


def _on_timer(self, event: TimeEvent) -> None:
    self.log.info("Timer fired!")


def _on_alert(self, event: TimeEvent) -> None:
    self.log.info("Alert triggered!")

Pass a callback to direct TimeEvent objects to your own method. Without one, the actor runtime connects the clock's registered default handler to on_time_event(). Use explicit callbacks when components share a clock. Timer names also share the clock's namespace, so use names unique to the component; registering the same name replaces the existing timer.

System access

Actors have access to core system components:

The Python DataActor and Strategy APIs do not expose self.msgbus. Use custom data for structured payloads and signals for lightweight values.

Queue pressure state

Actors can subscribe to runner queue pressure state changes:

use nautilus_common::{
    actor::DataActor,
    messages::system::QueueStateChanged,
};

impl DataActor for MyActor {
    fn on_start(&mut self) -> anyhow::Result<()> {
        self.subscribe_queue_state(Some(50));
        Ok(())
    }

    fn on_stop(&mut self) -> anyhow::Result<()> {
        self.unsubscribe_queue_state();
        Ok(())
    }

    fn on_queue_state(&mut self, event: &QueueStateChanged) -> anyhow::Result<()> {
        log::warn!(
            "Queue {:?} changed {:?} to {:?} at depth {}",
            event.channel,
            event.condition,
            event.state,
            event.queue_depth,
        );
        Ok(())
    }
}

The optional priority controls delivery order among matching subscribers. Higher values run first. Subscribing again does not change an existing priority; unsubscribe before subscribing with a new priority.

QueueStateChanged includes the trader ID, runner channel, queue condition, condition state, queue depth, mean dispatch time, event ID, and timestamps. Delivery uses the typed in‑process message bus and has no external wire representation. See Queue pressure monitoring for the trigger and clear semantics.

Socket transport state

Actors can subscribe to socket state changes from live adapters that report them:

use nautilus_common::{
    actor::DataActor,
    messages::system::SocketStateChanged,
};

impl DataActor for MyActor {
    fn on_start(&mut self) -> anyhow::Result<()> {
        self.subscribe_socket_state(Some(50));
        Ok(())
    }

    fn on_stop(&mut self) -> anyhow::Result<()> {
        self.unsubscribe_socket_state();
        Ok(())
    }

    fn on_socket_state(&mut self, event: &SocketStateChanged) -> anyhow::Result<()> {
        log::info!(
            "Socket {} for {} changed to {:?}",
            event.endpoint,
            event.client_id,
            event.state,
        );
        Ok(())
    }
}

The optional priority controls delivery order among matching subscribers. Higher values run first. Subscribing again does not change an existing priority; unsubscribe before subscribing with a new priority.

SocketStateChanged includes the trader ID, client ID, optional venue, stable endpoint label, transport state, event ID, and timestamps. SocketState.CONNECTED reports transport availability, not authentication, subscription replay, or adapter readiness. SocketState.DISCONNECTED reports the loss of an active transport. The endpoint is a non‑secret logical label, not a raw connection URL.

Delivery uses the typed in‑process message bus and has no external wire representation. See Socket transport state for supported adapters and the precise connection edge semantics.

Reconnect a socket endpoint

Live actors and strategies can request recovery of one endpoint without restarting its data or execution client. Pass the client_id and the endpoint label reported by SocketStateChanged:

use nautilus_common::actor::DataActor;
use nautilus_model::identifiers::ClientId;

impl MyActor {
    fn recover_market_socket(&self) -> anyhow::Result<()> {
        self.reconnect_socket(
            ClientId::from("POLYMARKET"),
            "polymarket-market-streams",
        )?;
        Ok(())
    }
}

This API is fire‑and‑observe. A successful return means the command passed local validation and was queued. It does not acknowledge that the kernel accepted the request or that recovery completed. Subscribe with subscribe_socket_state() and inspect SocketStateChanged events for the same client and endpoint. An accepted request reports SocketState.DISCONNECTED as the transport enters reconnect mode, followed by SocketState.CONNECTED after transport recovery.

The kernel logs unknown clients, unsupported clients, unknown or ambiguous endpoints, duplicate requests, disconnecting transports, and closed transports. These rejections do not emit a socket state change or affect another endpoint. Invalid endpoint labels and unavailable or closed runner channels fail synchronously. Endpoint labels accept only ASCII letters, digits, ., -, and _; pass a logical label rather than a raw URL.

Data handling and callbacks

The system dispatches request responses separately from subscribed updates. The operation determines which callback handles the data.

Request responses and subscriptions

The system distinguishes between two data flows:

1. Request responses:

   • Obtained through methods like request_bars(), request_quotes(), etc.
   • Processed through type‑specific batch handlers such as on_historical_bars() and on_historical_quotes().
   • Custom data uses on_historical_data() once per response. A scalar CustomData arrives as that object, while a batch arrives as one list, including an empty list.
   • Used for initial data loading and historical analysis.

2. Subscribed data:

   • Obtained through methods like subscribe_bars(), subscribe_quotes(), etc.
   • Processed through specific handlers like on_bar(), on_quote(), etc.
   • Used for ongoing event processing.

Callback handlers

Common data operations map to these handlers:

Request and subscription example

This example shows both request and subscription handling:

from collections.abc import Sequence

from nautilus_trader.common import DataActor
from nautilus_trader.config import DataActorConfig
from nautilus_trader.model import Bar
from nautilus_trader.model import BarType


class MyActorConfig(DataActorConfig):
    def __init__(self, bar_type: BarType, **_kwargs) -> None:
        self.bar_type = bar_type


class MyActor(DataActor):
    def __init__(self, config: MyActorConfig) -> None:
        super().__init__(config)

    def on_start(self) -> None:
        # Limit the historical response, which is handled by on_historical_bars()
        self.request_bars(
            bar_type=self.config.bar_type,
            limit=100,
        )

        # Deliver subscribed updates to on_bar()
        self.subscribe_bars(self.config.bar_type)

    def on_historical_bars(self, bars: Sequence[Bar]) -> None:
        for bar in bars:
            self.log.info(f"Received historical bar: {bar}")

    def on_bar(self, bar: Bar) -> None:
        self.log.info(f"Received subscribed bar: {bar}")

Separate request and subscription handlers let an actor distinguish bootstrap data from ongoing updates. Use historical bars to initialize indicators or baseline state, and apply different validation or logging to response batches and individual subscribed updates.

Order event handling

Data actors do not manage orders or define order event callbacks. Handle order events in a Strategy through its specific order callbacks or on_order_event(). Use custom data or signals to pass derived values to a data actor when another component needs them. See Strategies: order management for the callback list.

Related guides

• Strategies: Strategies extend actors with order‑management capabilities.
• Data: Data types and subscriptions available to actors.
• Message Bus: The messaging system actors use for communication.