nautilus-common 0.67.0

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Python bindings for `DataActor` with complete command and event handler forwarding.

use std::{
    any::Any,
    cell::{RefCell, UnsafeCell},
    collections::HashMap,
    fmt::Debug,
    num::NonZeroUsize,
    ops::{Deref, DerefMut},
    rc::Rc,
};

use indexmap::IndexMap;
use jiff::Timestamp;
use nautilus_core::{
    from_pydict,
    nanos::UnixNanos,
    python::{to_pyruntime_err, to_pytype_err, to_pyvalue_err, upgrade_py_weakref},
};
#[cfg(feature = "defi")]
use nautilus_model::defi::{
    Block, Blockchain, Pool, PoolFeeCollect, PoolFlash, PoolLiquidityUpdate, PoolSwap,
};
use nautilus_model::{
    data::{
        Bar, BarType, CustomData, DataType, FundingRateUpdate, IndexPriceUpdate, InstrumentStatus,
        MarkPriceUpdate, OrderBookDelta, OrderBookDeltas, OrderBookDepth, QuoteTick, TradeTick,
        close::InstrumentClose,
        option_chain::{OptionChainSlice, OptionGreeks},
    },
    enums::BookType,
    identifiers::{
        ActorId, ClientId, ExecAlgorithmId, InstrumentId, OptionSeriesId, TraderId, Venue,
    },
    instruments::{InstrumentAny, SyntheticInstrument},
    orderbook::OrderBook,
    orders::{OrderAny, OrderList},
    python::{
        data::option_chain::PyStrikeRange, instruments::instrument_any_to_pyobject,
        orders::order_any_to_pyobject,
    },
};
use pyo3::{
    IntoPyObjectExt,
    prelude::*,
    types::{PyBytes, PyDict, PyList, PySet, PyWeakrefReference},
};
use ustr::Ustr;

use crate::{
    actor::{
        Actor, DataActor, DataActorNative,
        data_actor::{DataActorConfig, DataActorCore, ImportableActorConfig},
        registry::{try_get_actor_unchecked, with_actor_registry},
    },
    cache::Cache,
    clock::Clock,
    component::{Component, ComponentAccessError, with_component_registry},
    enums::ComponentState,
    logging::{CMD, RECV},
    messages::{
        data::{
            DataResponseEnvelope, IntoBarTypes, IntoInstrumentIds, ensure_routable_bar_types,
            ensure_routable_instrument_ids, ensure_subscription_routes, quote_routing,
        },
        execution::TradingCommand,
        system::{QueueStateChanged, SocketStateChanged},
    },
    msgbus::{self, ShareableMessageHandler, TypedHandler},
    python::{
        cache::PyCache,
        clock::PyClock,
        indicators::{registered_python_indicators, wrap_python_indicator},
        logging::{PyLogger, format_exception},
        response::PyDataResponse,
        wrappers::{get_python_message_bus, retain_python_wrapper},
    },
    runner::SystemChannel,
    signal::Signal,
    timer::{TimeEvent, TimeEventCallback},
};

#[pyo3::pymethods]
#[pyo3_stub_gen::derive::gen_stub_pymethods]
impl DataActorConfig {
    /// Common configuration for `DataActor` based components.
    #[new]
    #[pyo3(signature = (actor_id=None, log_events=true, log_commands=true, **_kwargs))]
    fn py_new(
        actor_id: Option<ActorId>,
        log_events: bool,
        log_commands: bool,
        _kwargs: Option<&Bound<'_, PyDict>>,
    ) -> Self {
        Self {
            actor_id,
            log_events,
            log_commands,
        }
    }

    #[getter]
    fn actor_id(&self) -> Option<ActorId> {
        self.actor_id
    }

    #[setter]
    fn set_actor_id(&mut self, actor_id: Option<ActorId>) {
        self.actor_id = actor_id;
    }

    #[getter]
    fn log_events(&self) -> bool {
        self.log_events
    }

    #[setter]
    fn set_log_events(&mut self, log_events: bool) {
        self.log_events = log_events;
    }

    #[getter]
    fn log_commands(&self) -> bool {
        self.log_commands
    }

    #[setter]
    fn set_log_commands(&mut self, log_commands: bool) {
        self.log_commands = log_commands;
    }
}

#[pyo3::pymethods]
#[pyo3_stub_gen::derive::gen_stub_pymethods]
impl ImportableActorConfig {
    /// Configuration for creating actors from importable paths.
    #[new]
    #[expect(clippy::needless_pass_by_value)]
    fn py_new(actor_path: String, config_path: String, config: Py<PyDict>) -> PyResult<Self> {
        let json_config = Python::attach(|py| -> PyResult<HashMap<String, serde_json::Value>> {
            let kwargs = PyDict::new(py);
            kwargs.set_item("default", py.eval(pyo3::ffi::c_str!("str"), None, None)?)?;
            let json_str: String = PyModule::import(py, "json")?
                .call_method("dumps", (config.bind(py),), Some(&kwargs))?
                .extract()?;

            let json_value: serde_json::Value =
                serde_json::from_str(&json_str).map_err(to_pyvalue_err)?;

            if let serde_json::Value::Object(map) = json_value {
                Ok(map.into_iter().collect())
            } else {
                Err(to_pyvalue_err("Config must be a dictionary"))
            }
        })?;

        Ok(Self {
            actor_path,
            config_path,
            config: json_config,
        })
    }

    #[getter]
    fn actor_path(&self) -> &String {
        &self.actor_path
    }

    #[getter]
    fn config_path(&self) -> &String {
        &self.config_path
    }

    #[getter]
    fn config(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
        // Convert HashMap<String, serde_json::Value> back to Python dict
        let py_dict = PyDict::new(py);

        for (key, value) in &self.config {
            // Convert serde_json::Value back to Python object via JSON
            let json_str = serde_json::to_string(value).map_err(to_pyvalue_err)?;
            let py_value = PyModule::import(py, "json")?.call_method("loads", (json_str,), None)?;
            py_dict.set_item(key, py_value)?;
        }
        Ok(py_dict.unbind())
    }
}

/// Inner state of `PyDataActor`, shared between Python wrapper and Rust registries.
///
/// This type holds the actual actor state and implements all the actor traits.
/// It is wrapped in `Rc<UnsafeCell<>>` to allow shared ownership between Python
/// and the global registries without copying.
pub struct PyDataActorInner {
    core: DataActorCore,
    py_self: Option<Py<PyWeakrefReference>>,
    config: Option<Py<PyAny>>,
    clock: PyClock,
    logger: PyLogger,
}

impl Debug for PyDataActorInner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(PyDataActorInner))
            .field("core", &self.core)
            .field(
                "py_self",
                &self.py_self.as_ref().map(|_| "<Py<PyWeakrefReference>>"),
            )
            .field("config", &self.config.as_ref().map(|_| "<Py<PyAny>>"))
            .field("clock", &self.clock)
            .field("logger", &self.logger)
            .finish()
    }
}

impl Deref for PyDataActorInner {
    type Target = DataActorCore;

    fn deref(&self) -> &Self::Target {
        &self.core
    }
}

impl DerefMut for PyDataActorInner {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.core
    }
}

impl DataActorNative for PyDataActorInner {
    fn core(&self) -> &DataActorCore {
        &self.core
    }

    fn core_mut(&mut self) -> &mut DataActorCore {
        &mut self.core
    }
}

#[expect(clippy::needless_pass_by_ref_mut)]
impl PyDataActorInner {
    fn execute_exec_algorithm_command(&mut self, command: &TradingCommand) -> anyhow::Result<()> {
        if self.core.config.log_commands {
            let id = self.core.actor_id;
            log::info!("{id} {RECV}{CMD} {command}");
        }

        if self.core.state() != ComponentState::Running {
            return Ok(());
        }

        match command {
            TradingCommand::SubmitOrder(cmd) => {
                let order = DataActor::cache(self).try_order(&cmd.client_order_id)?;
                self.dispatch_on_order(order).map_err(|e| {
                    anyhow::anyhow!("Python on_order failed:\n{}", format_exception(&e))
                })
            }
            TradingCommand::SubmitOrderList(cmd) => {
                let orders = self.orders_for_list(&cmd.order_list)?;
                self.dispatch_on_order_list(cmd.order_list.clone(), orders)
                    .map_err(|e| {
                        anyhow::anyhow!("Python on_order_list failed:\n{}", format_exception(&e))
                    })
            }
            _ => {
                log::warn!("Unhandled command type: {command}");
                Ok(())
            }
        }
    }

    fn orders_for_list(&self, order_list: &OrderList) -> anyhow::Result<Vec<OrderAny>> {
        let cache = DataActor::cache(self);
        let mut orders = Vec::with_capacity(order_list.client_order_ids.len());

        for client_order_id in &order_list.client_order_ids {
            orders.push(cache.try_order(client_order_id)?);
        }

        Ok(orders)
    }

    fn dispatch_on_order(&mut self, order: OrderAny) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| -> PyResult<()> {
                let py_order = order_any_to_pyobject(py, order)?;
                py_self.call_method1(py, "on_order", (py_order,))?;
                Ok(())
            })?;
        }
        Ok(())
    }

    fn dispatch_on_order_list(
        &mut self,
        order_list: OrderList,
        orders: Vec<OrderAny>,
    ) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| -> PyResult<()> {
                if py_self.bind(py).hasattr("on_order_list")? {
                    let py_order_list = order_list.into_py_any(py)?;
                    let py_orders = orders
                        .into_iter()
                        .map(|order| order_any_to_pyobject(py, order))
                        .collect::<PyResult<Vec<_>>>()?;
                    let py_orders = PyList::new(py, py_orders)?;

                    py_self.call_method1(py, "on_order_list", (py_order_list, py_orders))?;
                } else {
                    for order in orders {
                        let py_order = order_any_to_pyobject(py, order)?;
                        py_self.call_method1(py, "on_order", (py_order,))?;
                    }
                }
                Ok(())
            })?;
        }
        Ok(())
    }

    fn dispatch_on_start(&self) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method0(py, "on_start"))?;
        }
        Ok(())
    }

    fn dispatch_on_stop(&mut self) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method0(py, "on_stop"))?;
        }
        Ok(())
    }

    fn dispatch_on_resume(&mut self) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method0(py, "on_resume"))?;
        }
        Ok(())
    }

    fn dispatch_on_reset(&mut self) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method0(py, "on_reset"))?;
        }
        Ok(())
    }

    fn dispatch_on_dispose(&mut self) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method0(py, "on_dispose"))?;
        }
        Ok(())
    }

    fn dispatch_on_degrade(&mut self) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method0(py, "on_degrade"))?;
        }
        Ok(())
    }

    fn dispatch_on_fault(&mut self) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method0(py, "on_fault"))?;
        }
        Ok(())
    }

    fn dispatch_on_save(&self) -> PyResult<IndexMap<String, Vec<u8>>> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_state = py_self.call_method0(py, "on_save")?;
                let py_state: &Bound<'_, PyDict> = py_state.cast_bound::<PyDict>(py)?;
                pydict_to_state(py_state)
            })
        } else {
            Ok(IndexMap::new())
        }
    }

    fn dispatch_on_load(&mut self, state: &IndexMap<String, Vec<u8>>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| -> PyResult<()> {
                let py_state = state_to_pydict(py, state)?;
                py_self.call_method1(py, "on_load", (py_state,))?;
                Ok(())
            })?;
        }
        Ok(())
    }

    fn dispatch_on_time_event(&mut self, event: TimeEvent) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_time_event", (event.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_data(&mut self, data: Py<PyAny>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_data", (data,)))?;
        }
        Ok(())
    }

    fn dispatch_on_signal(&mut self, signal: &Signal) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_signal", (signal.clone().into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_queue_state(&mut self, event: &QueueStateChanged) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_queue_state", (event.clone().into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_socket_state(&mut self, event: &SocketStateChanged) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_socket_state", (event.clone().into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_instrument(&mut self, instrument: Py<PyAny>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_instrument", (instrument,)))?;
        }
        Ok(())
    }

    fn dispatch_on_quote(&mut self, quote: QuoteTick) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_quote", (quote.into_py_any(py)?,)))?;
        }
        Ok(())
    }

    fn dispatch_on_trade(&mut self, trade: TradeTick) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_trade", (trade.into_py_any(py)?,)))?;
        }
        Ok(())
    }

    fn dispatch_on_bar(&mut self, bar: Bar) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_bar", (bar.into_py_any(py)?,)))?;
        }
        Ok(())
    }

    fn dispatch_on_book_deltas(&mut self, deltas: OrderBookDeltas) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_book_deltas", (deltas.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_book_depth(&mut self, depth: &OrderBookDepth) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_book_depth", (depth.clone().into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_book(&mut self, book: &OrderBook) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_book", (book.clone().into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_mark_price(&mut self, mark_price: MarkPriceUpdate) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_mark_price", (mark_price.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_index_price(&mut self, index_price: IndexPriceUpdate) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_index_price", (index_price.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_funding_rate(&mut self, funding_rate: FundingRateUpdate) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_funding_rate", (funding_rate.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_instrument_status(&mut self, data: InstrumentStatus) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_instrument_status", (data.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_instrument_close(&mut self, update: InstrumentClose) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_instrument_close", (update.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_option_greeks(&mut self, greeks: OptionGreeks) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_option_greeks", (greeks.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_option_chain(&mut self, slice: OptionChainSlice) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_option_chain", (slice.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_data(&mut self, data: Py<PyAny>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_historical_data", (data,)))?;
        }
        Ok(())
    }

    fn dispatch_on_historical_book_deltas(&mut self, deltas: Vec<OrderBookDelta>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_deltas = deltas
                    .into_iter()
                    .map(|delta| delta.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_book_deltas", (py_deltas,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_book_depth(&mut self, depths: Vec<OrderBookDepth>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_depths = depths
                    .into_iter()
                    .map(|depth| depth.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_book_depth", (py_depths,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_quotes(&mut self, quotes: Vec<QuoteTick>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_quotes = quotes
                    .into_iter()
                    .map(|q| q.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_quotes", (py_quotes,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_trades(&mut self, trades: Vec<TradeTick>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_trades = trades
                    .into_iter()
                    .map(|t| t.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_trades", (py_trades,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_funding_rates(
        &mut self,
        funding_rates: Vec<FundingRateUpdate>,
    ) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_rates = funding_rates
                    .into_iter()
                    .map(|r| r.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_funding_rates", (py_rates,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_bars(&mut self, bars: Vec<Bar>) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_bars = bars
                    .into_iter()
                    .map(|b| b.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_bars", (py_bars,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_mark_prices(
        &mut self,
        mark_prices: Vec<MarkPriceUpdate>,
    ) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_prices = mark_prices
                    .into_iter()
                    .map(|p| p.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_mark_prices", (py_prices,))
            })?;
        }
        Ok(())
    }

    fn dispatch_on_historical_index_prices(
        &mut self,
        index_prices: Vec<IndexPriceUpdate>,
    ) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                let py_prices = index_prices
                    .into_iter()
                    .map(|p| p.into_py_any(py))
                    .collect::<PyResult<Vec<_>>>()?;
                py_self.call_method1(py, "on_historical_index_prices", (py_prices,))
            })?;
        }
        Ok(())
    }

    #[cfg(feature = "defi")]
    fn dispatch_on_block(&mut self, block: Block) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_block", (block.into_py_any(py)?,)))?;
        }
        Ok(())
    }

    #[cfg(feature = "defi")]
    fn dispatch_on_pool(&mut self, pool: Pool) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| py_self.call_method1(py, "on_pool", (pool.into_py_any(py)?,)))?;
        }
        Ok(())
    }

    #[cfg(feature = "defi")]
    fn dispatch_on_pool_swap(&mut self, swap: PoolSwap) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_pool_swap", (swap.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    #[cfg(feature = "defi")]
    fn dispatch_on_pool_liquidity_update(&mut self, update: PoolLiquidityUpdate) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_pool_liquidity_update", (update.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    #[cfg(feature = "defi")]
    fn dispatch_on_pool_fee_collect(&mut self, collect: PoolFeeCollect) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_pool_fee_collect", (collect.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    #[cfg(feature = "defi")]
    fn dispatch_on_pool_flash(&mut self, flash: PoolFlash) -> PyResult<()> {
        if let Some(py_self) = self.python_instance()? {
            Python::attach(|py| {
                py_self.call_method1(py, "on_pool_flash", (flash.into_py_any(py)?,))
            })?;
        }
        Ok(())
    }

    // The trader owns the wrapper for as long as the actor stays registered, so a collected
    // wrapper propagates as an error rather than a skipped callback.
    fn python_instance(&self) -> PyResult<Option<Py<PyAny>>> {
        upgrade_py_weakref(self.py_self.as_ref(), &self.core.actor_id)
    }
}

fn dict_to_params(
    py: Python<'_>,
    params: Option<Py<PyDict>>,
) -> PyResult<Option<nautilus_core::Params>> {
    match params {
        Some(dict) => from_pydict(py, &dict),
        None => Ok(None),
    }
}

fn state_to_pydict(py: Python<'_>, state: &IndexMap<String, Vec<u8>>) -> PyResult<Py<PyDict>> {
    let py_state = PyDict::new(py);
    for (key, value) in state {
        py_state.set_item(key, PyBytes::new(py, value))?;
    }
    Ok(py_state.unbind())
}

fn pydict_to_state(state: &Bound<'_, PyDict>) -> PyResult<IndexMap<String, Vec<u8>>> {
    let mut rust_state = IndexMap::with_capacity(state.len());
    for (key, value) in state.iter() {
        rust_state.insert(key.extract()?, value.extract()?);
    }
    Ok(rust_state)
}

/// Python-facing wrapper for `DataActor`.
///
/// This wrapper holds shared ownership of `PyDataActorInner` via `Rc<UnsafeCell<>>`.
/// Both Python (through this wrapper) and the global registries share the same
/// underlying actor instance, ensuring mutations are visible from both sides.
#[allow(non_camel_case_types)]
#[pyo3::pyclass(
    module = "nautilus_trader.common",
    name = "DataActor",
    unsendable,
    subclass,
    weakref
)]
#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")]
pub struct PyDataActor {
    inner: Rc<UnsafeCell<PyDataActorInner>>,
}

impl Debug for PyDataActor {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct(stringify!(PyDataActor))
            .field("inner", &self.inner())
            .finish()
    }
}

impl PyDataActor {
    /// Returns a reference to the inner actor state.
    ///
    /// # Safety
    ///
    /// This is safe for single-threaded use. The `UnsafeCell` allows interior
    /// mutability which is required for the registries to mutate the actor.
    #[inline]
    #[allow(unsafe_code)]
    pub(crate) fn inner(&self) -> &PyDataActorInner {
        unsafe { &*self.inner.get() }
    }

    /// Returns a mutable reference to the inner actor state.
    ///
    /// # Safety
    ///
    /// This is safe for single-threaded use. Callers must ensure no aliasing
    /// mutable references exist.
    #[inline]
    #[allow(unsafe_code, clippy::mut_from_ref)]
    pub(crate) fn inner_mut(&self) -> &mut PyDataActorInner {
        unsafe { &mut *self.inner.get() }
    }
}

impl Deref for PyDataActor {
    type Target = DataActorCore;

    fn deref(&self) -> &Self::Target {
        &self.inner().core
    }
}

impl DerefMut for PyDataActor {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.inner_mut().core
    }
}

impl PyDataActor {
    // Rust constructor for tests and direct Rust usage
    pub fn new(config: Option<DataActorConfig>) -> Self {
        let config = config.unwrap_or_default();
        let core = DataActorCore::new(config);
        let clock = PyClock::new_test(); // Temporary clock, will be updated on registration
        let logger = PyLogger::new(core.actor_id().as_str());

        let inner = PyDataActorInner {
            core,
            py_self: None,
            config: None,
            clock,
            logger,
        };

        Self {
            inner: Rc::new(UnsafeCell::new(inner)),
        }
    }

    /// Creates a new [`PyDataActor`] from a Python config object.
    ///
    /// Extracts a [`DataActorConfig`] when the object provides one, falling back to defaults
    /// otherwise, and retains the original object so `.config` returns what the caller passed.
    /// Subclass constructors share this so the base is built the same way everywhere.
    #[must_use]
    pub fn from_py_config(config: Option<Py<PyAny>>) -> Self {
        let actor_config = config
            .as_ref()
            .and_then(|obj| Python::attach(|py| obj.extract::<DataActorConfig>(py).ok()));
        let mut actor = Self::new(actor_config);
        actor.set_config(config);
        actor
    }

    /// Sets the Python instance reference for method dispatch.
    ///
    /// This enables the `PyDataActor` to forward method calls (like `on_start`, `on_stop`)
    /// to the original Python instance that contains this `PyDataActor`. This is essential
    /// for Python inheritance to work correctly, allowing Python subclasses to override
    /// `DataActor` methods and have them called by the Rust system.
    ///
    /// Only a weak reference is stored, so the caller keeps ownership of `py_obj`. The trader
    /// owns registered wrappers; an unregistered actor stays collectable.
    ///
    /// # Errors
    ///
    /// Returns an error if `py_obj` cannot be weakly referenced.
    pub fn set_python_instance(&mut self, py_obj: &Bound<'_, PyAny>) -> PyResult<()> {
        self.inner_mut().py_self = Some(PyWeakrefReference::new(py_obj)?.unbind());
        Ok(())
    }

    /// Stores the original Python config object passed at construction.
    ///
    /// Retained so the constructed instance exposes `.config` and instance-based registration can
    /// source the actor ID and logging flags from the same single config object.
    pub fn set_config(&mut self, config: Option<Py<PyAny>>) {
        self.inner_mut().config = config;
    }

    /// Updates the `actor_id` in both the core config and the `actor_id` field.
    ///
    /// This method is only exposed for the Python actor to assist with configuration and should
    /// **never** be called post registration. Calling this after registration will cause
    /// inconsistent state where the actor is registered under one ID but its internal `actor_id`
    /// field contains another, breaking message routing and lifecycle management.
    pub fn set_actor_id(&mut self, actor_id: ActorId) {
        let inner = self.inner_mut();
        inner.core.config.actor_id = Some(actor_id);
        inner.core.actor_id = actor_id;
        inner.logger = PyLogger::new(actor_id.as_str());
    }

    /// Updates the `log_events` setting in the core config.
    pub fn set_log_events(&mut self, log_events: bool) {
        self.inner_mut().core.config.log_events = log_events;
    }

    /// Updates the `log_commands` setting in the core config.
    pub fn set_log_commands(&mut self, log_commands: bool) {
        self.inner_mut().core.config.log_commands = log_commands;
    }

    /// Returns the memory address of this instance as a hexadecimal string.
    pub fn mem_address(&self) -> String {
        self.inner().core.mem_address()
    }

    /// Returns a value indicating whether the actor has been registered with a trader.
    pub fn is_registered(&self) -> bool {
        self.inner().core.is_registered()
    }

    /// Register the actor with a trader.
    ///
    /// # Errors
    ///
    /// Returns an error if the actor is already registered or if the registration process fails.
    pub fn register(
        &mut self,
        trader_id: TraderId,
        clock: Rc<RefCell<dyn Clock>>,
        cache: Rc<RefCell<Cache>>,
    ) -> anyhow::Result<()> {
        let inner = self.inner_mut();
        inner.core.register(trader_id, clock, cache)?;

        inner.clock = PyClock::from_rc(inner.core.clock_rc());

        // Register default time event handler for this actor
        let actor_id = inner.actor_id().inner();
        let callback = TimeEventCallback::from(move |event: TimeEvent| {
            if let Some(mut actor) = try_get_actor_unchecked::<PyDataActorInner>(&actor_id) {
                if let Err(e) = actor.on_time_event(&event) {
                    log::error!("Python time event handler failed for actor {actor_id}: {e}");
                }
            } else {
                log::error!("Actor {actor_id} not found for time event handling");
            }
        });

        inner.clock.inner_mut().register_default_handler(callback);

        inner.initialize()
    }

    /// Registers this actor in the global component, actor, and wrapper registries.
    ///
    /// Clones the internal `Rc` and inserts into both registries. This ensures
    /// Python and the registries share the exact same actor instance. The Python wrapper is
    /// retained as part of the same act, so a registered actor always has an owner for the
    /// wrapper its inner only weakly references.
    ///
    /// # Errors
    ///
    /// Returns an error if no Python wrapper is attached, or if the attached wrapper has already
    /// been collected. Nothing is registered in that case.
    pub fn register_in_global_registries(&self) -> PyResult<()> {
        let inner = self.inner();
        let component_id = inner.component_id();
        let actor_id = Actor::id(inner);

        let Some(wrapper) = inner.python_instance()? else {
            return Err(to_pyruntime_err(format!(
                "Cannot register actor {actor_id} without a Python wrapper, call `set_python_instance` first"
            )));
        };

        let inner_ref: Rc<UnsafeCell<PyDataActorInner>> = self.inner.clone();

        let component_trait_ref: Rc<UnsafeCell<dyn Component>> = inner_ref.clone();
        with_component_registry(|registry| {
            registry.insert(component_id.inner(), component_trait_ref);
        });

        let actor_trait_ref: Rc<UnsafeCell<dyn Actor>> = inner_ref;
        with_actor_registry(|registry| registry.insert(actor_id, actor_trait_ref));

        retain_python_wrapper(component_id, wrapper, inner.core.message_bus());

        Ok(())
    }
}

pub fn register_python_exec_algorithm_endpoint(exec_algorithm_id: ExecAlgorithmId) {
    let actor_id = exec_algorithm_id.inner();
    let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
    let handler = ShareableMessageHandler::from_typed(move |command: &TradingCommand| {
        if let Some(mut algo) = try_get_actor_unchecked::<PyDataActorInner>(&actor_id) {
            if let Err(e) = algo.execute_exec_algorithm_command(command) {
                log::error!("Error executing command on Python algorithm {actor_id}: {e}");
            }
        } else {
            log::error!("Python execution algorithm {actor_id} not found in registry");
        }
    });
    msgbus::register_any(endpoint.into(), handler);
}

impl DataActor for PyDataActorInner {
    fn response_handler(&self) -> TypedHandler<DataResponseEnvelope> {
        crate::python::response::response_handler(
            self.core().actor_id().inner(),
            Self::python_instance,
            None,
        )
    }

    fn on_start(&mut self) -> anyhow::Result<()> {
        self.dispatch_on_start()
            .map_err(|e| anyhow::anyhow!("Python on_start failed:\n{}", format_exception(&e)))
    }

    fn on_stop(&mut self) -> anyhow::Result<()> {
        self.dispatch_on_stop()
            .map_err(|e| anyhow::anyhow!("Python on_stop failed:\n{}", format_exception(&e)))
    }

    fn on_resume(&mut self) -> anyhow::Result<()> {
        self.dispatch_on_resume()
            .map_err(|e| anyhow::anyhow!("Python on_resume failed:\n{}", format_exception(&e)))
    }

    fn on_reset(&mut self) -> anyhow::Result<()> {
        self.dispatch_on_reset()
            .map_err(|e| anyhow::anyhow!("Python on_reset failed:\n{}", format_exception(&e)))
    }

    fn on_dispose(&mut self) -> anyhow::Result<()> {
        self.dispatch_on_dispose()
            .map_err(|e| anyhow::anyhow!("Python on_dispose failed:\n{}", format_exception(&e)))
    }

    fn on_degrade(&mut self) -> anyhow::Result<()> {
        self.dispatch_on_degrade()
            .map_err(|e| anyhow::anyhow!("Python on_degrade failed:\n{}", format_exception(&e)))
    }

    fn on_fault(&mut self) -> anyhow::Result<()> {
        self.dispatch_on_fault()
            .map_err(|e| anyhow::anyhow!("Python on_fault failed:\n{}", format_exception(&e)))
    }

    fn on_save(&self) -> anyhow::Result<IndexMap<String, Vec<u8>>> {
        self.dispatch_on_save()
            .map_err(|e| anyhow::anyhow!("Python on_save failed:\n{}", format_exception(&e)))
    }

    fn on_load(&mut self, state: IndexMap<String, Vec<u8>>) -> anyhow::Result<()> {
        self.dispatch_on_load(&state)
            .map_err(|e| anyhow::anyhow!("Python on_load failed:\n{}", format_exception(&e)))
    }

    fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
        self.dispatch_on_time_event(event.clone())
            .map_err(|e| anyhow::anyhow!("Python on_time_event failed:\n{}", format_exception(&e)))
    }

    #[allow(unused_variables)]
    fn on_data(&mut self, data: &CustomData) -> anyhow::Result<()> {
        Python::attach(|py| {
            let py_data: Py<PyAny> = Py::new(py, data.clone())?.into_any();
            self.dispatch_on_data(py_data)
                .map_err(|e| anyhow::anyhow!("Python on_data failed:\n{}", format_exception(&e)))
        })
    }

    fn on_signal(&mut self, signal: &Signal) -> anyhow::Result<()> {
        self.dispatch_on_signal(signal)
            .map_err(|e| anyhow::anyhow!("Python on_signal failed:\n{}", format_exception(&e)))
    }

    fn on_queue_state(&mut self, event: &QueueStateChanged) -> anyhow::Result<()> {
        self.dispatch_on_queue_state(event)
            .map_err(|e| anyhow::anyhow!("Python on_queue_state failed:\n{}", format_exception(&e)))
    }

    fn on_socket_state(&mut self, event: &SocketStateChanged) -> anyhow::Result<()> {
        self.dispatch_on_socket_state(event).map_err(|e| {
            anyhow::anyhow!("Python on_socket_state failed:\n{}", format_exception(&e))
        })
    }

    fn on_instrument(&mut self, instrument: &InstrumentAny) -> anyhow::Result<()> {
        Python::attach(|py| {
            let py_instrument = instrument_any_to_pyobject(py, instrument.clone())
                .map_err(|e| anyhow::anyhow!("Failed to convert InstrumentAny to Python: {e}"))?;
            self.dispatch_on_instrument(py_instrument).map_err(|e| {
                anyhow::anyhow!("Python on_instrument failed:\n{}", format_exception(&e))
            })
        })
    }

    fn on_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
        self.dispatch_on_quote(*quote)
            .map_err(|e| anyhow::anyhow!("Python on_quote failed:\n{}", format_exception(&e)))
    }

    fn on_trade(&mut self, tick: &TradeTick) -> anyhow::Result<()> {
        self.dispatch_on_trade(*tick)
            .map_err(|e| anyhow::anyhow!("Python on_trade failed:\n{}", format_exception(&e)))
    }

    fn on_bar(&mut self, bar: &Bar) -> anyhow::Result<()> {
        self.dispatch_on_bar(*bar)
            .map_err(|e| anyhow::anyhow!("Python on_bar failed:\n{}", format_exception(&e)))
    }

    fn on_book_deltas(&mut self, deltas: &OrderBookDeltas) -> anyhow::Result<()> {
        self.dispatch_on_book_deltas(deltas.clone())
            .map_err(|e| anyhow::anyhow!("Python on_book_deltas failed:\n{}", format_exception(&e)))
    }

    fn on_book_depth(&mut self, depth: &OrderBookDepth) -> anyhow::Result<()> {
        self.dispatch_on_book_depth(depth)
            .map_err(|e| anyhow::anyhow!("Python on_book_depth failed:\n{}", format_exception(&e)))
    }

    fn on_book(&mut self, order_book: &OrderBook) -> anyhow::Result<()> {
        self.dispatch_on_book(order_book)
            .map_err(|e| anyhow::anyhow!("Python on_book failed:\n{}", format_exception(&e)))
    }

    fn on_mark_price(&mut self, mark_price: &MarkPriceUpdate) -> anyhow::Result<()> {
        self.dispatch_on_mark_price(*mark_price)
            .map_err(|e| anyhow::anyhow!("Python on_mark_price failed:\n{}", format_exception(&e)))
    }

    fn on_index_price(&mut self, index_price: &IndexPriceUpdate) -> anyhow::Result<()> {
        self.dispatch_on_index_price(*index_price)
            .map_err(|e| anyhow::anyhow!("Python on_index_price failed:\n{}", format_exception(&e)))
    }

    fn on_funding_rate(&mut self, funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
        self.dispatch_on_funding_rate(*funding_rate).map_err(|e| {
            anyhow::anyhow!("Python on_funding_rate failed:\n{}", format_exception(&e))
        })
    }

    fn on_instrument_status(&mut self, data: &InstrumentStatus) -> anyhow::Result<()> {
        self.dispatch_on_instrument_status(*data).map_err(|e| {
            anyhow::anyhow!(
                "Python on_instrument_status failed:\n{}",
                format_exception(&e)
            )
        })
    }

    fn on_instrument_close(&mut self, update: &InstrumentClose) -> anyhow::Result<()> {
        self.dispatch_on_instrument_close(*update).map_err(|e| {
            anyhow::anyhow!(
                "Python on_instrument_close failed:\n{}",
                format_exception(&e)
            )
        })
    }

    fn on_option_greeks(&mut self, greeks: &OptionGreeks) -> anyhow::Result<()> {
        self.dispatch_on_option_greeks(*greeks).map_err(|e| {
            anyhow::anyhow!("Python on_option_greeks failed:\n{}", format_exception(&e))
        })
    }

    fn on_option_chain(&mut self, slice: &OptionChainSlice) -> anyhow::Result<()> {
        self.dispatch_on_option_chain(slice.clone()).map_err(|e| {
            anyhow::anyhow!("Python on_option_chain failed:\n{}", format_exception(&e))
        })
    }

    #[cfg(feature = "defi")]
    fn on_block(&mut self, block: &Block) -> anyhow::Result<()> {
        self.dispatch_on_block(block.clone())
            .map_err(|e| anyhow::anyhow!("Python on_block failed:\n{}", format_exception(&e)))
    }

    #[cfg(feature = "defi")]
    fn on_pool(&mut self, pool: &Pool) -> anyhow::Result<()> {
        self.dispatch_on_pool(pool.clone())
            .map_err(|e| anyhow::anyhow!("Python on_pool failed:\n{}", format_exception(&e)))
    }

    #[cfg(feature = "defi")]
    fn on_pool_swap(&mut self, swap: &PoolSwap) -> anyhow::Result<()> {
        self.dispatch_on_pool_swap(swap.clone())
            .map_err(|e| anyhow::anyhow!("Python on_pool_swap failed:\n{}", format_exception(&e)))
    }

    #[cfg(feature = "defi")]
    fn on_pool_liquidity_update(&mut self, update: &PoolLiquidityUpdate) -> anyhow::Result<()> {
        self.dispatch_on_pool_liquidity_update(update.clone())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_pool_liquidity_update failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    #[cfg(feature = "defi")]
    fn on_pool_fee_collect(&mut self, collect: &PoolFeeCollect) -> anyhow::Result<()> {
        self.dispatch_on_pool_fee_collect(collect.clone())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_pool_fee_collect failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    #[cfg(feature = "defi")]
    fn on_pool_flash(&mut self, flash: &PoolFlash) -> anyhow::Result<()> {
        self.dispatch_on_pool_flash(flash.clone())
            .map_err(|e| anyhow::anyhow!("Python on_pool_flash failed:\n{}", format_exception(&e)))
    }

    fn on_historical_data(&mut self, data: &dyn Any) -> anyhow::Result<()> {
        Python::attach(|py| {
            let py_data: Py<PyAny> = if let Some(custom_data) = data.downcast_ref::<CustomData>() {
                Py::new(py, custom_data.clone())?.into_any()
            } else if let Some(custom_data) = data.downcast_ref::<Vec<CustomData>>() {
                custom_data.clone().into_py_any(py)?
            } else {
                anyhow::bail!("Failed to convert historical data to Python: unsupported type");
            };

            self.dispatch_on_historical_data(py_data).map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_data failed:\n{}",
                    format_exception(&e)
                )
            })
        })
    }

    fn on_historical_book_deltas(&mut self, deltas: &[OrderBookDelta]) -> anyhow::Result<()> {
        self.dispatch_on_historical_book_deltas(deltas.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_book_deltas failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    fn on_historical_book_depth(&mut self, depths: &[OrderBookDepth]) -> anyhow::Result<()> {
        self.dispatch_on_historical_book_depth(depths.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_book_depth failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    fn on_historical_quotes(&mut self, quotes: &[QuoteTick]) -> anyhow::Result<()> {
        self.dispatch_on_historical_quotes(quotes.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_quotes failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    fn on_historical_trades(&mut self, trades: &[TradeTick]) -> anyhow::Result<()> {
        self.dispatch_on_historical_trades(trades.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_trades failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    fn on_historical_funding_rates(
        &mut self,
        funding_rates: &[FundingRateUpdate],
    ) -> anyhow::Result<()> {
        self.dispatch_on_historical_funding_rates(funding_rates.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_funding_rates failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    fn on_historical_bars(&mut self, bars: &[Bar]) -> anyhow::Result<()> {
        self.dispatch_on_historical_bars(bars.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_bars failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    fn on_historical_mark_prices(&mut self, mark_prices: &[MarkPriceUpdate]) -> anyhow::Result<()> {
        self.dispatch_on_historical_mark_prices(mark_prices.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_mark_prices failed:\n{}",
                    format_exception(&e)
                )
            })
    }

    fn on_historical_index_prices(
        &mut self,
        index_prices: &[IndexPriceUpdate],
    ) -> anyhow::Result<()> {
        self.dispatch_on_historical_index_prices(index_prices.to_vec())
            .map_err(|e| {
                anyhow::anyhow!(
                    "Python on_historical_index_prices failed:\n{}",
                    format_exception(&e)
                )
            })
    }
}

#[pymethods]
#[pyo3_stub_gen::derive::gen_stub_pymethods]
impl PyDataActor {
    /// Creates a new [`PyDataActor`] instance.
    ///
    /// Accepts `None` or any Python object. If the object is a [`DataActorConfig`]
    /// (or can be extracted as one via `from_py_object`), its values are used;
    /// otherwise the actor falls back to [`DataActorConfig::default()`].
    ///
    /// This permissive signature is required so that Python subclasses can pass a
    /// **custom** config dataclass to their `__init__`. The original object is retained
    /// here in `__new__`, which always receives the constructor arguments, so `.config`
    /// and registration see the config even when a subclass omits forwarding it to
    /// `super().__init__()`.
    #[new]
    #[pyo3(signature = (config=None))]
    fn py_new(config: Option<Py<PyAny>>) -> Self {
        Self::from_py_config(config)
    }

    #[pyo3(signature = (config=None))]
    fn __init__(slf: &Bound<'_, Self>, config: Option<Py<PyAny>>) -> PyResult<()> {
        {
            let mut borrowed = slf.borrow_mut();
            borrowed.set_python_instance(slf.as_any())?;
            // `__new__` retained the config; only a forwarded config overrides it
            if config.is_some() {
                borrowed.set_config(config);
            }
        }

        if !has_configured_actor_id(slf) {
            let py_type = slf.get_type();
            let type_name = py_type.name()?;
            let actor_id = ActorId::new_checked(type_name.to_str()?).map_err(to_pyvalue_err)?;
            slf.borrow_mut().set_actor_id(actor_id);
        }

        Ok(())
    }

    #[getter]
    #[pyo3(name = "clock")]
    fn py_clock(&self) -> PyResult<PyClock> {
        let inner = self.inner();
        if inner.core.is_registered() {
            Ok(inner.clock.clone())
        } else {
            Err(to_pyruntime_err(
                "Actor must be registered with a trader before accessing clock",
            ))
        }
    }

    #[getter]
    #[pyo3(name = "cache")]
    fn py_cache(&self) -> PyResult<PyCache> {
        let inner = self.inner();
        if inner.core.is_registered() {
            Ok(PyCache::from_rc(inner.core.cache_rc()))
        } else {
            Err(to_pyruntime_err(
                "Actor must be registered with a trader before accessing cache",
            ))
        }
    }

    #[getter]
    #[pyo3(name = "log")]
    fn py_log(&self) -> PyLogger {
        self.inner().logger.clone()
    }

    #[getter]
    #[pyo3(name = "actor_id")]
    fn py_actor_id(&self) -> ActorId {
        self.inner().core.actor_id
    }

    #[getter]
    #[pyo3(name = "config")]
    fn py_config(&self, py: Python<'_>) -> Option<Py<PyAny>> {
        self.inner()
            .config
            .as_ref()
            .map(|config| config.clone_ref(py))
    }

    #[getter]
    #[pyo3(name = "trader_id")]
    fn py_trader_id(&self) -> Option<TraderId> {
        self.inner().core.trader_id()
    }

    #[pyo3(name = "state")]
    fn py_state(&self) -> ComponentState {
        Component::state(self.inner())
    }

    #[pyo3(name = "is_ready")]
    fn py_is_ready(&self) -> bool {
        Component::is_ready(self.inner())
    }

    #[pyo3(name = "is_running")]
    fn py_is_running(&self) -> bool {
        Component::is_running(self.inner())
    }

    #[pyo3(name = "is_stopped")]
    fn py_is_stopped(&self) -> bool {
        Component::is_stopped(self.inner())
    }

    #[pyo3(name = "is_degraded")]
    fn py_is_degraded(&self) -> bool {
        Component::is_degraded(self.inner())
    }

    #[pyo3(name = "is_faulted")]
    fn py_is_faulted(&self) -> bool {
        Component::is_faulted(self.inner())
    }

    #[pyo3(name = "is_disposed")]
    fn py_is_disposed(&self) -> bool {
        Component::is_disposed(self.inner())
    }

    #[pyo3(name = "start")]
    fn py_start(&mut self) -> PyResult<()> {
        Component::start(self.inner_mut()).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "stop")]
    fn py_stop(&mut self) -> PyResult<()> {
        Component::stop(self.inner_mut()).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "save")]
    fn py_save(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
        let state = DataActor::on_save(self.inner()).map_err(to_pyruntime_err)?;
        state_to_pydict(py, &state)
    }

    #[pyo3(name = "load")]
    fn py_load(&mut self, state: &Bound<'_, PyDict>) -> PyResult<()> {
        let state = pydict_to_state(state)?;
        DataActor::on_load(self.inner_mut(), state).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "resume")]
    fn py_resume(&mut self) -> PyResult<()> {
        Component::resume(self.inner_mut()).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "reset")]
    fn py_reset(&mut self) -> PyResult<()> {
        Component::reset(self.inner_mut()).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "dispose")]
    fn py_dispose(&mut self) -> PyResult<()> {
        Component::dispose(self.inner_mut()).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "degrade")]
    fn py_degrade(&mut self) -> PyResult<()> {
        Component::degrade(self.inner_mut()).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "fault")]
    fn py_fault(&mut self) -> PyResult<()> {
        Component::fault(self.inner_mut()).map_err(to_pyruntime_err)
    }

    #[pyo3(name = "shutdown_system")]
    #[pyo3(signature = (reason=None))]
    fn py_shutdown_system(&self, reason: Option<String>) -> PyResult<()> {
        if !self.inner().core.is_registered() {
            return Err(to_pyruntime_err(
                "Actor must be registered with a trader before shutting down the system",
            ));
        }

        self.inner().shutdown_system(reason);
        Ok(())
    }

    #[pyo3(name = "publish_data")]
    fn py_publish_data(&self, data_type: &DataType, data: &CustomData) -> PyResult<()> {
        self.ensure_registered_for_data()?;
        self.inner().publish_data(data_type, data);
        Ok(())
    }

    #[pyo3(name = "publish_signal")]
    #[pyo3(signature = (name, value, ts_event=0))]
    #[expect(
        clippy::needless_pass_by_value,
        reason = "PyO3 accepts an owned PyAny handle for Python signal values"
    )]
    fn py_publish_signal(
        &self,
        py: Python<'_>,
        name: &str,
        value: Py<PyAny>,
        ts_event: u64,
    ) -> PyResult<()> {
        self.ensure_registered_for_data()?;
        let value_str: String = value.bind(py).str()?.extract()?;
        self.inner()
            .publish_signal(name, value_str, UnixNanos::from(ts_event));
        Ok(())
    }

    #[pyo3(name = "add_synthetic")]
    fn py_add_synthetic(&self, synthetic: SyntheticInstrument) -> PyResult<()> {
        self.ensure_registered_for_data()?;
        self.inner()
            .add_synthetic(synthetic)
            .map_err(to_pyvalue_err)
    }

    #[pyo3(name = "update_synthetic")]
    fn py_update_synthetic(&self, synthetic: SyntheticInstrument) -> PyResult<()> {
        self.ensure_registered_for_data()?;
        self.inner()
            .update_synthetic(synthetic)
            .map_err(to_pyvalue_err)
    }

    #[getter]
    #[pyo3(name = "registered_indicators")]
    fn py_registered_indicators(&self, py: Python<'_>) -> PyResult<Py<PyList>> {
        registered_python_indicators(py, self.inner().core.registered_indicators())
    }

    #[pyo3(name = "indicators_initialized")]
    fn py_indicators_initialized(&self, _py: Python<'_>) -> PyResult<bool> {
        self.inner()
            .core
            .indicators_initialized()
            .map_err(to_pyruntime_err)
    }

    #[pyo3(name = "register_indicator_for_quote_ticks")]
    fn py_register_indicator_for_quote_ticks(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        indicator: Py<PyAny>,
    ) {
        let indicator = wrap_python_indicator(py, indicator);
        self.inner_mut()
            .core
            .register_indicator_for_quote_ticks(instrument_id, indicator);
    }

    #[pyo3(name = "register_indicator_for_trade_ticks")]
    fn py_register_indicator_for_trade_ticks(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        indicator: Py<PyAny>,
    ) {
        let indicator = wrap_python_indicator(py, indicator);
        self.inner_mut()
            .core
            .register_indicator_for_trade_ticks(instrument_id, indicator);
    }

    #[pyo3(name = "register_indicator_for_bars")]
    fn py_register_indicator_for_bars(
        &mut self,
        py: Python<'_>,
        bar_type: BarType,
        indicator: Py<PyAny>,
    ) {
        let indicator = wrap_python_indicator(py, indicator);
        self.inner_mut()
            .core
            .register_indicator_for_bars(bar_type, indicator);
    }

    #[pyo3(name = "on_start")]
    fn py_on_start(_slf: &Bound<'_, Self>) {}

    #[pyo3(name = "on_stop")]
    fn py_on_stop(_slf: &Bound<'_, Self>) {}

    #[pyo3(name = "on_resume")]
    fn py_on_resume(_slf: &Bound<'_, Self>) {}

    #[pyo3(name = "on_reset")]
    fn py_on_reset(_slf: &Bound<'_, Self>) {}

    #[pyo3(name = "on_dispose")]
    fn py_on_dispose(_slf: &Bound<'_, Self>) {}

    #[pyo3(name = "on_degrade")]
    fn py_on_degrade(_slf: &Bound<'_, Self>) {}

    #[pyo3(name = "on_fault")]
    fn py_on_fault(_slf: &Bound<'_, Self>) {}

    #[pyo3(name = "on_save")]
    fn py_on_save(_slf: &Bound<'_, Self>, py: Python<'_>) -> Py<PyDict> {
        PyDict::new(py).unbind()
    }

    #[allow(unused_variables)]
    #[pyo3(name = "on_load")]
    fn py_on_load(_slf: &Bound<'_, Self>, state: &Bound<'_, PyDict>) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_time_event")]
    fn py_on_time_event(_slf: &Bound<'_, Self>, event: TimeEvent) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_data")]
    fn py_on_data(_slf: &Bound<'_, Self>, data: Py<PyAny>) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_signal")]
    fn py_on_signal(_slf: &Bound<'_, Self>, signal: &Signal) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_queue_state")]
    fn py_on_queue_state(_slf: &Bound<'_, Self>, event: QueueStateChanged) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_socket_state")]
    fn py_on_socket_state(_slf: &Bound<'_, Self>, event: SocketStateChanged) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_instrument")]
    fn py_on_instrument(_slf: &Bound<'_, Self>, instrument: Py<PyAny>) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_quote")]
    fn py_on_quote(_slf: &Bound<'_, Self>, quote: QuoteTick) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_trade")]
    fn py_on_trade(_slf: &Bound<'_, Self>, trade: TradeTick) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_bar")]
    fn py_on_bar(_slf: &Bound<'_, Self>, bar: Bar) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_book_deltas")]
    fn py_on_book_deltas(_slf: &Bound<'_, Self>, deltas: OrderBookDeltas) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_book_depth")]
    fn py_on_book_depth(_slf: &Bound<'_, Self>, depth: &OrderBookDepth) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_book")]
    fn py_on_book(_slf: &Bound<'_, Self>, book: &OrderBook) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_mark_price")]
    fn py_on_mark_price(_slf: &Bound<'_, Self>, mark_price: MarkPriceUpdate) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_index_price")]
    fn py_on_index_price(_slf: &Bound<'_, Self>, index_price: IndexPriceUpdate) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_funding_rate")]
    fn py_on_funding_rate(_slf: &Bound<'_, Self>, funding_rate: FundingRateUpdate) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_instrument_status")]
    fn py_on_instrument_status(_slf: &Bound<'_, Self>, status: InstrumentStatus) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_instrument_close")]
    fn py_on_instrument_close(_slf: &Bound<'_, Self>, close: InstrumentClose) {}

    #[allow(unused_variables)]
    #[pyo3(name = "on_option_greeks")]
    fn py_on_option_greeks(_slf: &Bound<'_, Self>, greeks: OptionGreeks) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_option_chain")]
    fn py_on_option_chain(_slf: &Bound<'_, Self>, slice: OptionChainSlice) {}

    #[pyo3(name = "subscribe_data")]
    #[pyo3(signature = (data_type, client_id=None, params=None))]
    fn py_subscribe_data(
        &mut self,
        py: Python<'_>,
        data_type: DataType,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_data(self.inner_mut(), data_type, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_signal")]
    #[pyo3(signature = (name="", priority=None))]
    fn py_subscribe_signal(
        slf: &Bound<'_, Self>,
        name: &str,
        priority: Option<u32>,
    ) -> PyResult<()> {
        let actor = borrow_actor_mut(slf, "subscribe_signal")?;
        actor.ensure_registered()?;
        DataActor::subscribe_signal(actor.inner_mut(), name, priority);
        Ok(())
    }

    #[pyo3(name = "subscribe_queue_state")]
    #[pyo3(signature = (channel=None, priority=None))]
    fn py_subscribe_queue_state(
        &mut self,
        channel: Option<SystemChannel>,
        priority: Option<u32>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        DataActor::subscribe_queue_state(self.inner_mut(), channel, priority);
        Ok(())
    }

    #[pyo3(name = "subscribe_socket_state")]
    #[pyo3(signature = (client_id=None, endpoint=None, priority=None))]
    fn py_subscribe_socket_state(
        &mut self,
        client_id: Option<ClientId>,
        endpoint: Option<&str>,
        priority: Option<u32>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        DataActor::subscribe_socket_state(self.inner_mut(), client_id, endpoint, priority);
        Ok(())
    }

    #[pyo3(name = "subscribe_instruments")]
    #[pyo3(signature = (venue, client_id=None, params=None))]
    fn py_subscribe_instruments(
        &mut self,
        py: Python<'_>,
        venue: Venue,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_instruments(self.inner_mut(), venue, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_instrument")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_instrument(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        ensure_subscription_routes(
            &instrument_id,
            client_id.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_instrument(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_book_deltas")]
    #[pyo3(signature = (instrument_id, book_type, depth=None, client_id=None, managed=false, params=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_subscribe_book_deltas(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        book_type: BookType,
        depth: Option<usize>,
        client_id: Option<ClientId>,
        managed: bool,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        ensure_subscription_routes(
            &instrument_id,
            client_id.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        let depth = depth.and_then(NonZeroUsize::new);
        DataActor::subscribe_book_deltas(
            self.inner_mut(),
            instrument_id,
            book_type,
            depth,
            client_id,
            managed,
            params,
        );
        Ok(())
    }

    #[expect(clippy::too_many_arguments)]
    #[pyo3(name = "subscribe_book_depth")]
    #[pyo3(signature = (instrument_id, book_type, depth=None, client_id=None, managed=false, params=None))]
    fn py_subscribe_book_depth(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        book_type: BookType,
        depth: Option<usize>,
        client_id: Option<ClientId>,
        managed: bool,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        ensure_subscription_routes(
            &instrument_id,
            client_id.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;

        let depth = depth
            .map(|value| {
                NonZeroUsize::new(value).ok_or_else(|| to_pyvalue_err("depth must be positive"))
            })
            .transpose()?;

        let params = dict_to_params(py, params)?;
        DataActor::subscribe_book_depth(
            self.inner_mut(),
            instrument_id,
            book_type,
            depth,
            client_id,
            managed,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "subscribe_book_at_interval")]
    #[pyo3(signature = (instrument_id, book_type, interval_ms, depth=None, client_id=None, params=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_subscribe_book_at_interval(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        book_type: BookType,
        interval_ms: usize,
        depth: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        let interval_ms = NonZeroUsize::new(interval_ms)
            .ok_or_else(|| to_pyvalue_err("interval_ms must be > 0"))?;

        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        ensure_subscription_routes(
            &instrument_id,
            client_id.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        let depth = depth.and_then(NonZeroUsize::new);
        DataActor::subscribe_book_at_interval(
            self.inner_mut(),
            instrument_id,
            book_type,
            depth,
            interval_ms,
            client_id,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "subscribe_quotes")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_quotes(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_ids = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_ids, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        quote_routing(
            &instrument_ids,
            client_id.as_ref(),
            params.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;
        DataActor::subscribe_quotes(self.inner_mut(), instrument_ids, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_trades")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_trades(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_ids = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_ids, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_trades(self.inner_mut(), instrument_ids, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_bars")]
    #[pyo3(signature = (bar_type, client_id=None, params=None))]
    fn py_subscribe_bars(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.BarType | typing.Sequence[model.BarType]"))]
        bar_type: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let bar_types = extract_bar_types(bar_type)?;
        ensure_routable_bar_types(&bar_types, client_id.is_some()).map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_bars(self.inner_mut(), bar_types, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_mark_prices")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_mark_prices(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_mark_prices(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_index_prices")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_index_prices(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_index_prices(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_funding_rates")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_funding_rates(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_funding_rates(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_option_greeks")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_option_greeks(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        ensure_subscription_routes(
            &instrument_id,
            client_id.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_option_greeks(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_instrument_status")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_instrument_status(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        ensure_subscription_routes(
            &instrument_id,
            client_id.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_instrument_status(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_instrument_close")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_instrument_close(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        ensure_subscription_routes(
            &instrument_id,
            client_id.as_ref(),
            &DataActorNative::cache_ref(self.inner()),
        )
        .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_instrument_close(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_option_chain")]
    #[pyo3(signature = (series_id, strike_range, snapshot_interval_ms=None, client_id=None, params=None))]
    fn py_subscribe_option_chain(
        &mut self,
        py: Python<'_>,
        series_id: OptionSeriesId,
        strike_range: PyStrikeRange,
        snapshot_interval_ms: Option<u64>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_option_chain(
            self.inner_mut(),
            series_id,
            strike_range.inner,
            snapshot_interval_ms,
            client_id,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "unsubscribe_data")]
    #[pyo3(signature = (data_type, client_id=None, params=None))]
    fn py_unsubscribe_data(
        &mut self,
        py: Python<'_>,
        data_type: DataType,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_data(self.inner_mut(), data_type, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_signal")]
    #[pyo3(signature = (name=""))]
    fn py_unsubscribe_signal(slf: &Bound<'_, Self>, name: &str) -> PyResult<()> {
        let actor = borrow_actor_mut(slf, "unsubscribe_signal")?;
        actor.ensure_registered()?;
        DataActor::unsubscribe_signal(actor.inner_mut(), name);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_queue_state")]
    #[pyo3(signature = (channel=None))]
    fn py_unsubscribe_queue_state(&mut self, channel: Option<SystemChannel>) -> PyResult<()> {
        self.ensure_registered()?;
        DataActor::unsubscribe_queue_state(self.inner_mut(), channel);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_socket_state")]
    #[pyo3(signature = (client_id=None, endpoint=None))]
    fn py_unsubscribe_socket_state(
        &mut self,
        client_id: Option<ClientId>,
        endpoint: Option<&str>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        DataActor::unsubscribe_socket_state(self.inner_mut(), client_id, endpoint);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_instruments")]
    #[pyo3(signature = (venue, client_id=None, params=None))]
    fn py_unsubscribe_instruments(
        &mut self,
        py: Python<'_>,
        venue: Venue,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_instruments(self.inner_mut(), venue, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_instrument")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_instrument(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_instrument(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_book_deltas")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_book_deltas(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_book_deltas(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_book_depth")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_book_depth(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_book_depth(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_book_at_interval")]
    #[pyo3(signature = (instrument_id, interval_ms, client_id=None, params=None))]
    fn py_unsubscribe_book_at_interval(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        interval_ms: usize,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        let interval_ms = NonZeroUsize::new(interval_ms)
            .ok_or_else(|| to_pyvalue_err("interval_ms must be > 0"))?;

        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_book_at_interval(
            self.inner_mut(),
            instrument_id,
            interval_ms,
            client_id,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "unsubscribe_quotes")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_quotes(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_ids = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_ids, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_quotes(self.inner_mut(), instrument_ids, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_trades")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_trades(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_ids = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_ids, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_trades(self.inner_mut(), instrument_ids, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_bars")]
    #[pyo3(signature = (bar_type, client_id=None, params=None))]
    fn py_unsubscribe_bars(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.BarType | typing.Sequence[model.BarType]"))]
        bar_type: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let bar_types = extract_bar_types(bar_type)?;
        ensure_routable_bar_types(&bar_types, client_id.is_some()).map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_bars(self.inner_mut(), bar_types, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_mark_prices")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_mark_prices(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_mark_prices(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_index_prices")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_index_prices(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_index_prices(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_funding_rates")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_funding_rates(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_funding_rates(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_option_greeks")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_option_greeks(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_option_greeks(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_instrument_status")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_instrument_status(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_instrument_status(
            self.inner_mut(),
            instrument_id,
            client_id,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "unsubscribe_instrument_close")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_instrument_close(
        &mut self,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let instrument_id = extract_instrument_ids(instrument_id)?;
        ensure_routable_instrument_ids(&instrument_id, client_id.is_some())
            .map_err(to_pyvalue_err)?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_instrument_close(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_option_chain")]
    #[pyo3(signature = (series_id, client_id=None))]
    fn py_unsubscribe_option_chain(
        &mut self,
        series_id: OptionSeriesId,
        client_id: Option<ClientId>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        DataActor::unsubscribe_option_chain(self.inner_mut(), series_id, client_id);
        Ok(())
    }

    #[pyo3(name = "request_data")]
    #[pyo3(signature = (data_type, client_id, start=None, end=None, limit=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_data(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        data_type: DataType,
        client_id: ClientId,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let limit = limit.and_then(NonZeroUsize::new);
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_data(
            actor.inner().core(),
            data_type,
            client_id,
            start,
            end,
            limit,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_instrument")]
    #[pyo3(signature = (instrument_id, start=None, end=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_instrument(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let instrument_id = extract_instrument_ids(instrument_id)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_instrument(
            actor.inner().core(),
            instrument_id,
            start,
            end,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_instruments")]
    #[pyo3(signature = (venue=None, start=None, end=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_instruments(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        venue: Option<Venue>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_instruments(
            actor.inner().core(),
            venue,
            start,
            end,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_book_snapshot")]
    #[pyo3(signature = (instrument_id, depth=None, client_id=None, params=None, callback=None))]
    fn py_request_book_snapshot(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        depth: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let instrument_id = extract_instrument_ids(instrument_id)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let depth = depth.and_then(NonZeroUsize::new);

        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_book_snapshot(
            actor.inner().core(),
            instrument_id,
            depth,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_book_deltas")]
    #[pyo3(signature = (instrument_id, start=None, end=None, limit=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_book_deltas(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let instrument_id = extract_instrument_ids(instrument_id)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let limit = limit.and_then(NonZeroUsize::new);
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_book_deltas(
            actor.inner().core(),
            instrument_id,
            start,
            end,
            limit,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_book_depth")]
    #[pyo3(signature = (instrument_id, start=None, end=None, limit=None, depth=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_book_depth(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
        depth: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let instrument_id = extract_instrument_ids(instrument_id)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let limit = limit.and_then(NonZeroUsize::new);
        let depth = depth.and_then(NonZeroUsize::new);
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_book_depth(
            actor.inner().core(),
            instrument_id,
            start,
            end,
            limit,
            depth,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_quotes")]
    #[pyo3(signature = (instrument_id, start=None, end=None, limit=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_quotes(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let instrument_id = extract_instrument_ids(instrument_id)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let limit = limit.and_then(NonZeroUsize::new);
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_quotes(
            actor.inner().core(),
            instrument_id,
            start,
            end,
            limit,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_trades")]
    #[pyo3(signature = (instrument_id, start=None, end=None, limit=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_trades(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let instrument_id = extract_instrument_ids(instrument_id)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let limit = limit.and_then(NonZeroUsize::new);
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_trades(
            actor.inner().core(),
            instrument_id,
            start,
            end,
            limit,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_funding_rates")]
    #[pyo3(signature = (instrument_id, start=None, end=None, limit=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_funding_rates(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(
            type_repr = "model.InstrumentId | typing.Sequence[model.InstrumentId]"
        ))]
        instrument_id: &Bound<'_, PyAny>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let instrument_id = extract_instrument_ids(instrument_id)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let limit = limit.and_then(NonZeroUsize::new);
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_funding_rates(
            actor.inner().core(),
            instrument_id,
            start,
            end,
            limit,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_bars")]
    #[pyo3(signature = (bar_type, start=None, end=None, limit=None, client_id=None, params=None, callback=None))]
    #[expect(clippy::too_many_arguments)]
    fn py_request_bars(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "model.BarType | typing.Sequence[model.BarType]"))]
        bar_type: &Bound<'_, PyAny>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<usize>,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
    ) -> PyResult<String> {
        let actor = Self {
            inner: Rc::clone(&slf.try_borrow()?.inner),
        };

        let bar_type = extract_bar_types(bar_type)?;
        actor.ensure_registered_for_data()?;
        let params = dict_to_params(py, params)?;
        let limit = limit.and_then(NonZeroUsize::new);
        let handler = crate::python::response::request_handler(
            py,
            actor.inner().core().actor_id().inner(),
            PyDataActorInner::python_instance,
            callback,
        )?;
        let request_id = DataActorCore::request_bars(
            actor.inner().core(),
            bar_type,
            start,
            end,
            limit,
            client_id,
            params,
            handler,
        )
        .map_err(to_pyvalue_err)?;
        Ok(request_id.to_string())
    }

    #[pyo3(name = "request_join", signature = (request_ids, callback=None, request_id=None))]
    fn py_request_join(
        slf: &Bound<'_, Self>,
        py: Python<'_>,
        #[gen_stub(override_type(type_repr = "typing.Sequence[str | core.UUID4]", imports = ("typing", "nautilus_trader.core")))]
        request_ids: Vec<Bound<'_, PyAny>>,
        #[gen_stub(override_type(type_repr = "typing.Callable[[str], None] | None"))]
        callback: Option<Py<PyAny>>,
        #[gen_stub(override_type(type_repr = "str | core.UUID4 | None", imports = ("nautilus_trader.core",)))]
        request_id: Option<&Bound<'_, PyAny>>,
    ) -> PyResult<String> {
        let actor_id = {
            let actor = slf.try_borrow()?;
            actor.ensure_registered_for_data()?;
            actor.inner().core().actor_id().inner()
        };

        let request_ids = request_ids
            .into_iter()
            .map(|id| crate::python::response::extract_request_id(&id))
            .collect::<PyResult<Vec<_>>>()?;
        let request_id = request_id
            .map(crate::python::response::extract_request_id)
            .transpose()?;
        crate::python::response::request_join::<PyDataActorInner>(
            py,
            actor_id,
            PyDataActorInner::python_instance,
            request_ids,
            request_id,
            callback,
        )
    }

    #[pyo3(name = "is_pending_request")]
    fn py_is_pending_request(
        &self,
        #[gen_stub(override_type(type_repr = "str | core.UUID4", imports = ("nautilus_trader.core",)))]
        request_id: &Bound<'_, PyAny>,
    ) -> PyResult<bool> {
        let id = crate::python::response::extract_request_id(request_id)?;
        Ok(self.inner().core().is_pending_request(&id))
    }

    #[pyo3(name = "has_pending_requests")]
    fn py_has_pending_requests(&self) -> bool {
        !self.inner().core().pending_requests().is_empty()
    }

    #[pyo3(name = "pending_requests")]
    #[gen_stub(override_return_type(type_repr = "set[str]"))]
    fn py_pending_requests(&self, py: Python<'_>) -> PyResult<Py<PySet>> {
        let ids = self.inner().core().pending_requests();
        PySet::new(py, ids.iter().map(ToString::to_string)).map(Bound::unbind)
    }

    /// Requests reconnect of one socket endpoint owned by `client_id`.
    #[pyo3(name = "reconnect_socket")]
    fn py_reconnect_socket(&self, client_id: ClientId, endpoint: &str) -> PyResult<()> {
        DataActor::reconnect_socket(self.inner(), client_id, endpoint).map_err(to_pyruntime_err)
    }

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_data_response")]
    fn py_on_data_response(_slf: &Bound<'_, Self>, response: Py<PyDataResponse>) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_data")]
    fn py_on_historical_data(_slf: &Bound<'_, Self>, data: Py<PyAny>) {
        // Default implementation - can be overridden in Python subclasses
    }

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_book_deltas")]
    fn py_on_historical_book_deltas(_slf: &Bound<'_, Self>, deltas: Vec<OrderBookDelta>) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_book_depth")]
    fn py_on_historical_book_depth(_slf: &Bound<'_, Self>, depths: Vec<OrderBookDepth>) {}

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_quotes")]
    fn py_on_historical_quotes(_slf: &Bound<'_, Self>, quotes: Vec<QuoteTick>) {
        // Default implementation - can be overridden in Python subclasses
    }

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_trades")]
    fn py_on_historical_trades(_slf: &Bound<'_, Self>, trades: Vec<TradeTick>) {
        // Default implementation - can be overridden in Python subclasses
    }

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_funding_rates")]
    fn py_on_historical_funding_rates(
        _slf: &Bound<'_, Self>,
        funding_rates: Vec<FundingRateUpdate>,
    ) {
        // Default implementation - can be overridden in Python subclasses
    }

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_bars")]
    fn py_on_historical_bars(_slf: &Bound<'_, Self>, bars: Vec<Bar>) {
        // Default implementation - can be overridden in Python subclasses
    }

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_mark_prices")]
    fn py_on_historical_mark_prices(_slf: &Bound<'_, Self>, mark_prices: Vec<MarkPriceUpdate>) {
        // Default implementation - can be overridden in Python subclasses
    }

    #[allow(unused_variables, clippy::needless_pass_by_value)]
    #[pyo3(name = "on_historical_index_prices")]
    fn py_on_historical_index_prices(_slf: &Bound<'_, Self>, index_prices: Vec<IndexPriceUpdate>) {
        // Default implementation - can be overridden in Python subclasses
    }
}

#[pyo3_stub_gen::derive::gen_stub_pymethods]
#[pyo3::pymethods]
impl PyDataActor {
    #[pyo3(name = "publish_message", signature = (topic, message))]
    fn py_publish_message(
        slf: &Bound<'_, Self>,
        topic: &str,
        #[gen_stub(override_type(type_repr = "object"))] message: Py<PyAny>,
    ) -> PyResult<()> {
        let messages = get_python_message_bus(slf.as_any())?;
        messages.publish_message(topic, message)
    }

    #[pyo3(name = "subscribe_topic")]
    #[pyo3(signature = (topic, handler, priority=0))]
    fn py_subscribe_topic(
        slf: &Bound<'_, Self>,
        topic: &str,
        #[gen_stub(override_type(type_repr = "collections.abc.Callable[[object], None]", imports = ("collections.abc",)))]
        handler: Py<PyAny>,
        priority: u32,
    ) -> PyResult<()> {
        let messages = get_python_message_bus(slf.as_any())?;
        messages.subscribe_topic(slf.py(), topic, handler, priority)
    }

    #[pyo3(name = "unsubscribe_topic", signature = (topic, handler))]
    fn py_unsubscribe_topic(
        slf: &Bound<'_, Self>,
        topic: &str,
        #[gen_stub(override_type(type_repr = "collections.abc.Callable[[object], None]", imports = ("collections.abc",)))]
        handler: &Bound<'_, PyAny>,
    ) -> PyResult<()> {
        let messages = get_python_message_bus(slf.as_any())?;
        messages.unsubscribe_topic(topic, handler)
    }
}

#[cfg(feature = "defi")]
#[pymethods]
#[pyo3_stub_gen::derive::gen_stub_pymethods]
impl PyDataActor {
    #[pyo3(name = "on_block")]
    #[allow(unused_variables, clippy::needless_pass_by_value)]
    fn py_on_block(_slf: &Bound<'_, Self>, block: Block) {}

    #[pyo3(name = "on_pool")]
    #[allow(unused_variables, clippy::needless_pass_by_value)]
    fn py_on_pool(_slf: &Bound<'_, Self>, pool: Pool) {}

    #[pyo3(name = "on_pool_swap")]
    #[allow(unused_variables, clippy::needless_pass_by_value)]
    fn py_on_pool_swap(_slf: &Bound<'_, Self>, swap: PoolSwap) {}

    #[pyo3(name = "on_pool_liquidity_update")]
    #[allow(unused_variables, clippy::needless_pass_by_value)]
    fn py_on_pool_liquidity_update(_slf: &Bound<'_, Self>, update: PoolLiquidityUpdate) {}

    #[pyo3(name = "on_pool_fee_collect")]
    #[allow(unused_variables, clippy::needless_pass_by_value)]
    fn py_on_pool_fee_collect(_slf: &Bound<'_, Self>, update: PoolFeeCollect) {}

    #[pyo3(name = "on_pool_flash")]
    #[allow(unused_variables, clippy::needless_pass_by_value)]
    fn py_on_pool_flash(_slf: &Bound<'_, Self>, flash: PoolFlash) {}

    #[pyo3(name = "subscribe_blocks")]
    #[pyo3(signature = (chain, client_id=None, params=None))]
    fn py_subscribe_blocks(
        &mut self,
        py: Python<'_>,
        chain: Blockchain,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_blocks(self.inner_mut(), chain, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_pool")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_pool(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_pool(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_pool_swaps")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_pool_swaps(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_pool_swaps(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_pool_liquidity_updates")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_pool_liquidity_updates(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_pool_liquidity_updates(
            self.inner_mut(),
            instrument_id,
            client_id,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "subscribe_pool_fee_collects")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_pool_fee_collects(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_pool_fee_collects(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "subscribe_pool_flash_events")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_subscribe_pool_flash_events(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::subscribe_pool_flash_events(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_blocks")]
    #[pyo3(signature = (chain, client_id=None, params=None))]
    fn py_unsubscribe_blocks(
        &mut self,
        py: Python<'_>,
        chain: Blockchain,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_blocks(self.inner_mut(), chain, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_pool")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_pool(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_pool(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_pool_swaps")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_pool_swaps(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_pool_swaps(self.inner_mut(), instrument_id, client_id, params);
        Ok(())
    }

    #[pyo3(name = "unsubscribe_pool_liquidity_updates")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_pool_liquidity_updates(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_pool_liquidity_updates(
            self.inner_mut(),
            instrument_id,
            client_id,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "unsubscribe_pool_fee_collects")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_pool_fee_collects(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_pool_fee_collects(
            self.inner_mut(),
            instrument_id,
            client_id,
            params,
        );
        Ok(())
    }

    #[pyo3(name = "unsubscribe_pool_flash_events")]
    #[pyo3(signature = (instrument_id, client_id=None, params=None))]
    fn py_unsubscribe_pool_flash_events(
        &mut self,
        py: Python<'_>,
        instrument_id: InstrumentId,
        client_id: Option<ClientId>,
        params: Option<Py<PyDict>>,
    ) -> PyResult<()> {
        self.ensure_registered()?;
        let params = dict_to_params(py, params)?;
        DataActor::unsubscribe_pool_flash_events(
            self.inner_mut(),
            instrument_id,
            client_id,
            params,
        );
        Ok(())
    }
}

impl PyDataActor {
    fn ensure_registered_for_data(&self) -> PyResult<()> {
        if self.inner().core.is_registered() {
            Ok(())
        } else {
            Err(to_pyruntime_err(
                "DataActor must be registered before publishing, managing synthetics, or requesting data",
            ))
        }
    }

    fn ensure_registered(&self) -> PyResult<()> {
        if self.inner().core.is_registered() {
            Ok(())
        } else {
            Err(to_pyruntime_err(
                "DataActor must be registered before managing subscriptions",
            ))
        }
    }
}

/// Prepares a Python actor for registration and returns its final actor ID.
///
/// Applies optional config overrides and stores a weak reference to the Python instance used for
/// method dispatch. Missing or unreadable attributes are ignored, and non-boolean `log_events` or
/// `log_commands` values leave the existing settings unchanged. When neither the config nor
/// `DataActor.__init__` supplies an ID, the runtime class name replaces the shared default so
/// subclasses that skip the base initializer still receive distinct IDs.
///
/// # Errors
///
/// Returns an error if the actor cannot be extracted, its configured `actor_id` is neither an
/// [`ActorId`] nor a valid ID string, its class-derived ID is invalid, or its Python instance
/// cannot be weakly referenced.
pub fn prepare_python_actor(
    actor_obj: &Bound<'_, PyAny>,
    config: Option<&Bound<'_, PyAny>>,
) -> anyhow::Result<ActorId> {
    let mut actor = actor_obj
        .extract::<PyRefMut<PyDataActor>>()
        .map_err(Into::<PyErr>::into)
        .map_err(|e| anyhow::anyhow!("Failed to extract PyDataActor: {e}"))?;

    if let Some(config) = config {
        if let Some(actor_id) = config
            .getattr("actor_id")
            .ok()
            .filter(|actor_id| !actor_id.is_none())
        {
            let actor_id = if let Ok(actor_id) = actor_id.extract::<ActorId>() {
                actor_id
            } else if let Ok(actor_id) = actor_id.extract::<String>() {
                ActorId::new_checked(&actor_id)?
            } else {
                anyhow::bail!("Invalid `actor_id` type");
            };
            actor.set_actor_id(actor_id);
        }

        if let Some(log_events) = extract_bool_config_attr(config, "log_events") {
            actor.set_log_events(log_events);
        }

        if let Some(log_commands) = extract_bool_config_attr(config, "log_commands") {
            actor.set_log_commands(log_commands);
        }
    }

    actor.set_python_instance(actor_obj)?;
    apply_class_derived_actor_id(&mut actor, actor_obj)?;
    Ok(actor.actor_id())
}

fn apply_class_derived_actor_id(
    actor: &mut PyRefMut<'_, PyDataActor>,
    actor_obj: &Bound<'_, PyAny>,
) -> PyResult<()> {
    if actor.inner().core.config.actor_id.is_some() {
        return Ok(());
    }

    let py_type = actor_obj.get_type();
    let type_name = py_type.name()?;
    let actor_id = ActorId::new_checked(type_name.to_str()?).map_err(to_pyvalue_err)?;
    actor.set_actor_id(actor_id);

    Ok(())
}

fn extract_bool_config_attr(config: &Bound<'_, PyAny>, attr: &str) -> Option<bool> {
    config
        .getattr(attr)
        .ok()
        .and_then(|value| value.extract::<bool>().ok())
}

fn borrow_actor_mut<'py>(
    actor: &Bound<'py, PyDataActor>,
    operation: &'static str,
) -> PyResult<PyRefMut<'py, PyDataActor>> {
    actor.try_borrow_mut().map_err(|_| {
        to_pyruntime_err(ComponentAccessError::WriteConflict {
            resource: "Python actor",
            operation,
        })
    })
}

/// Returns whether the config retained by the actor supplies an actor ID.
///
/// The config is read through Python rather than the extracted [`DataActorConfig`] so that a
/// custom subclass config which cannot be extracted still counts as configuring an ID. The actor
/// borrow is released before the attribute lookup, which can run user code.
fn has_configured_actor_id(slf: &Bound<'_, PyDataActor>) -> bool {
    let py = slf.py();
    let config = slf
        .borrow()
        .inner()
        .config
        .as_ref()
        .map(|config| config.clone_ref(py));

    config.is_some_and(|config| {
        config
            .bind(py)
            .getattr("actor_id")
            .is_ok_and(|actor_id| !actor_id.is_none())
    })
}

/// Extracts one instrument ID or a sequence into a normalized, non-empty list.
///
/// Duplicate IDs are removed in first-occurrence order.
///
/// # Errors
///
/// Returns an error if:
/// - The value is not an instrument ID or a sequence of instrument IDs.
/// - The sequence is empty.
pub fn extract_instrument_ids(value: &Bound<'_, PyAny>) -> PyResult<Vec<InstrumentId>> {
    let instrument_ids = if let Ok(instrument_id) = value.extract::<InstrumentId>() {
        vec![instrument_id]
    } else {
        value.extract::<Vec<InstrumentId>>().map_err(|_| {
            to_pytype_err("expected InstrumentId or a sequence of InstrumentId values")
        })?
    };

    instrument_ids.into_instrument_ids().map_err(to_pyvalue_err)
}

/// Extracts one bar type or a sequence into a normalized, non-empty list.
///
/// Duplicate bar types are removed in first-occurrence order.
///
/// # Errors
///
/// Returns an error if the sequence is empty or the value is not a bar type or sequence of bar types.
pub fn extract_bar_types(value: &Bound<'_, PyAny>) -> PyResult<Vec<BarType>> {
    let bar_types = if let Ok(bar_type) = value.extract::<BarType>() {
        vec![bar_type]
    } else {
        value
            .extract::<Vec<BarType>>()
            .map_err(|_| to_pytype_err("expected BarType or a sequence of BarType values"))?
    };

    bar_types.into_bar_types().map_err(to_pyvalue_err)
}

#[cfg(test)]
mod tests {
    use std::{cell::RefCell, collections::HashMap, rc::Rc, str::FromStr, sync::Arc};

    #[cfg(feature = "defi")]
    use alloy_primitives::{I256, U160, U256};
    use indexmap::IndexMap;
    use nautilus_core::{UUID4, UnixNanos, python::IntoPyObjectNautilusExt};
    #[cfg(feature = "defi")]
    use nautilus_model::defi::{
        AmmType, Block, Blockchain, Chain, Dex, DexType, Pool, PoolFeeCollect, PoolFlash,
        PoolIdentifier, PoolLiquidityUpdate, PoolLiquidityUpdateType, PoolSwap, Token,
    };
    use nautilus_model::{
        data::{
            Bar, BarType, CustomData, DataType, FundingRateUpdate, IndexPriceUpdate,
            InstrumentStatus, MarkPriceUpdate, OrderBookDelta, OrderBookDeltas, OrderBookDepth,
            QuoteTick, TradeTick,
            close::InstrumentClose,
            greeks::OptionGreekValues,
            option_chain::{OptionChainSlice, OptionGreeks},
            stubs::{stub_custom_data, stub_deltas, stub_depth10},
        },
        enums::{
            AggressorSide, BookType, GreeksConvention, InstrumentCloseType, MarketStatusAction,
        },
        identifiers::{
            ActorId, ClientId, ComponentId, InstrumentId, OptionSeriesId, TradeId, TraderId, Venue,
        },
        instruments::{CurrencyPair, InstrumentAny, stubs::audusd_sim},
        orderbook::OrderBook,
        types::{Price, Quantity},
    };
    use pyo3::{
        Bound, IntoPyObjectExt, Py, PyAny, PyRef, PyResult, Python,
        ffi::c_str,
        types::{
            PyAnyMethods, PyBytes, PyDict, PyList, PyModule, PyTuple, PyWeakrefMethods,
            PyWeakrefReference,
        },
    };
    use rstest::{fixture, rstest};
    use ustr::Ustr;

    use super::{PyDataActor, prepare_python_actor};
    use crate::{
        actor::{DataActor, data_actor::DataActorConfig, registry::actor_exists},
        cache::Cache,
        clock::VirtualClock,
        component::{Component, get_component},
        enums::ComponentState,
        live::runner::replace_system_command_sender,
        messages::{
            SystemCommand,
            data::{
                BarsResponse, CustomDataResponse, DataCommand, QuotesResponse, RequestCommand,
                TradesResponse,
            },
            system::{
                QueueCondition, QueueState, QueueStateChanged, SocketState, SocketStateChanged,
            },
        },
        msgbus::{
            self, MessageBus, MessagingSwitchboard, get_message_bus,
            stubs::get_typed_into_message_saving_handler,
            switchboard::{get_quotes_topic, get_trades_topic},
        },
        python::wrappers::get_python_wrapper,
        runner::{SyncDataCommandSender, SystemChannel, set_data_cmd_sender},
        signal::Signal,
        timer::TimeEvent,
    };

    #[fixture]
    fn clock() -> Rc<RefCell<VirtualClock>> {
        Rc::new(RefCell::new(VirtualClock::new()))
    }

    #[fixture]
    fn cache() -> Rc<RefCell<Cache>> {
        Rc::new(RefCell::new(Cache::new(None, None)))
    }

    #[fixture]
    fn trader_id() -> TraderId {
        TraderId::from("TRADER-001")
    }

    #[fixture]
    fn client_id() -> ClientId {
        ClientId::new("TestClient")
    }

    #[fixture]
    fn venue() -> Venue {
        Venue::from("SIM")
    }

    #[fixture]
    fn data_type() -> DataType {
        DataType::new("TestData", None, None)
    }

    #[fixture]
    fn bar_type(audusd_sim: CurrencyPair) -> BarType {
        BarType::from_str(&format!("{}-1-MINUTE-LAST-INTERNAL", audusd_sim.id)).unwrap()
    }

    fn create_unregistered_actor() -> PyDataActor {
        PyDataActor::new(None)
    }

    fn create_registered_actor(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) -> PyDataActor {
        // Set up sync data command sender for tests
        let sender = SyncDataCommandSender;
        set_data_cmd_sender(Arc::new(sender));

        let mut actor = PyDataActor::new(None);
        actor.register(trader_id, clock, cache).unwrap();
        actor
    }

    #[rstest]
    fn test_python_identifier_lists_require_registration_before_validation(
        #[values(
            "subscribe_quotes",
            "unsubscribe_quotes",
            "subscribe_trades",
            "unsubscribe_trades"
        )]
        method: &str,
        #[values("empty", "invalid", "mixed_venues")] input: &str,
    ) {
        Python::initialize();
        Python::attach(|py| {
            let actor = create_unregistered_actor();
            let actor = Py::new(py, actor).unwrap();
            let argument = match input {
                "empty" => PyList::empty(py).into_any(),
                "invalid" => PyList::new(py, [42]).unwrap().into_any(),
                "mixed_venues" => PyList::new(
                    py,
                    [
                        InstrumentId::from("BTC/USD.SIM"),
                        InstrumentId::from("ETH/USD.OTHER"),
                    ],
                )
                .unwrap()
                .into_any(),
                _ => unreachable!(),
            };

            let error = actor
                .bind(py)
                .call_method1(method, (&argument,))
                .unwrap_err();

            assert!(error.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
            assert_eq!(
                error.value(py).to_string(),
                "DataActor must be registered before managing subscriptions"
            );
        });
    }

    #[rstest]
    fn test_python_identifier_list_subscriptions(
        #[values(
            "instrument",
            "book_deltas",
            "book_depth",
            "book_at_interval",
            "quotes",
            "trades",
            "mark_prices",
            "index_prices",
            "funding_rates",
            "instrument_status",
            "instrument_close",
            "option_greeks"
        )]
        family: &str,
        #[values("single", "list", "tuple")] input: &str,
        #[values(false, true)] keyword: bool,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        let expected_variant = match family {
            "instrument" => "Instrument",
            "book_deltas" => "BookDeltas",
            "book_depth" => "BookDepth",
            "book_at_interval" => "BookSnapshots",
            "quotes" => "Quotes",
            "trades" => "Trades",
            "mark_prices" => "MarkPrices",
            "index_prices" => "IndexPrices",
            "funding_rates" => "FundingRates",
            "instrument_status" => "InstrumentStatus",
            "instrument_close" => "InstrumentClose",
            "option_greeks" => "OptionGreeks",
            _ => unreachable!(),
        };

        let first = InstrumentId::from("BTC/USD.SIM");
        let second = InstrumentId::from("ETH/USD.SIM");

        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = create_registered_actor(clock, cache, trader_id);
            let actor = Py::new(py, actor).unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );

            let (argument, expected_ids) = match input {
                "single" => (first.into_py_any(py).unwrap(), vec![first]),
                "list" => (
                    PyList::new(py, [second, first, second])
                        .unwrap()
                        .into_any()
                        .unbind(),
                    vec![second, first],
                ),
                "tuple" => (
                    PyTuple::new(py, [second, first, second])
                        .unwrap()
                        .into_any()
                        .unbind(),
                    vec![second, first],
                ),
                _ => unreachable!(),
            };

            let args = PyTuple::new(py, (!keyword).then_some(argument.bind(py))).unwrap();
            for action in ["subscribe", "unsubscribe"] {
                let kwargs = instrument_subscription_keywords(py, family, action == "subscribe");
                if keyword {
                    kwargs.set_item("instrument_id", argument.bind(py)).unwrap();
                }

                let method = format!("{action}_{family}");
                actor
                    .bind(py)
                    .call_method(method.as_str(), &args, Some(&kwargs))
                    .unwrap();
            }

            let commands = saver.get_messages();

            assert_eq!(commands.len(), 2);

            for (index, command) in commands.iter().enumerate() {
                let (subscribe, actual_ids, serialized) = match command {
                    DataCommand::Subscribe(command) => (
                        true,
                        command.instrument_ids().unwrap(),
                        serde_json::to_value(command).unwrap(),
                    ),
                    DataCommand::Unsubscribe(command) => (
                        false,
                        command.instrument_ids().unwrap(),
                        serde_json::to_value(command).unwrap(),
                    ),
                    _ => panic!("unexpected command"),
                };

                assert_eq!(
                    serialized.as_object().unwrap().keys().collect::<Vec<_>>(),
                    vec![expected_variant]
                );

                assert_eq!(subscribe, index == 0);

                assert_eq!(actual_ids, expected_ids);
            }
        });
    }

    #[rstest::rstest]
    fn test_python_subscription_lists_reject_invalid_routes_before_effects(
        #[values(
            "subscribe_quotes",
            "subscribe_instrument",
            "subscribe_book_deltas",
            "subscribe_book_depth",
            "subscribe_book_at_interval",
            "subscribe_option_greeks",
            "subscribe_instrument_status",
            "subscribe_instrument_close"
        )]
        method: &str,
        #[values(false, true)] external: bool,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            let client_id = ClientId::from("ROUTED");
            if external {
                cache.borrow_mut().add_external_data_client(client_id);
            }

            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let ids = PyList::new(
                py,
                [
                    InstrumentId::from("AUD/USD.SIM"),
                    InstrumentId::from("SYNTH.SYNTH"),
                ],
            )
            .unwrap();
            let kwargs = PyDict::new(py);
            kwargs.set_item("client_id", client_id).unwrap();
            if method.starts_with("subscribe_book") {
                kwargs.set_item("book_type", BookType::L2_MBP).unwrap();
            }

            if method == "subscribe_book_at_interval" {
                kwargs.set_item("interval_ms", 13).unwrap();
            }

            let result = actor.bind(py).call_method(method, (&ids,), Some(&kwargs));
            if external {
                result.unwrap();
            } else {
                let error = result.unwrap_err();
                assert!(error.is_instance_of::<pyo3::exceptions::PyValueError>(py));

                let expected = if method == "subscribe_quotes" {
                    "quote identifier lists cannot mix client, synthetic, and spread routes"
                } else {
                    "Cannot subscribe synthetic instruments for this data type"
                };

                assert_eq!(error.value(py).to_string(), expected);
                assert!(saver.get_messages().is_empty());
            }

            if method.starts_with("subscribe_book") {
                kwargs.del_item("book_type").unwrap();
            }

            let unsubscribe = method.replacen("subscribe", "unsubscribe", 1);
            actor
                .bind(py)
                .call_method(unsubscribe.as_str(), (&ids,), Some(&kwargs))
                .unwrap();
            assert_eq!(saver.get_messages().len(), if external { 2 } else { 0 });
        });
    }

    #[rstest]
    fn test_python_identifier_lists_reject_mixed_venues_before_effects(
        #[values(
            "subscribe_quotes",
            "unsubscribe_quotes",
            "subscribe_trades",
            "unsubscribe_trades"
        )]
        method: &str,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = create_registered_actor(clock, cache, trader_id);
            let actor = Py::new(py, actor).unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let first = InstrumentId::from("BTC/USD.SIM");
            let second = InstrumentId::from("ETH/USD.OTHER");
            let argument = PyList::new(py, [first, second]).unwrap();
            let quotes = method.ends_with("quotes");
            let subscribe = method.starts_with("subscribe_");
            if !subscribe {
                let kwargs = PyDict::new(py);
                kwargs
                    .set_item("client_id", ClientId::from("ORIGINAL"))
                    .unwrap();
                actor
                    .bind(py)
                    .call_method(
                        if quotes {
                            "subscribe_quotes"
                        } else {
                            "subscribe_trades"
                        },
                        (&argument,),
                        Some(&kwargs),
                    )
                    .unwrap();
            }

            let error = actor
                .bind(py)
                .call_method1(method, (&argument,))
                .unwrap_err();
            assert!(error.is_instance_of::<pyo3::exceptions::PyValueError>(py));
            assert_eq!(
                error.value(py).to_string(),
                "multiple instrument venues require an explicit client ID"
            );
            assert_eq!(saver.get_messages().len(), usize::from(!subscribe));

            for id in [first, second] {
                assert_eq!(
                    msgbus::exact_subscriber_count_quotes(get_quotes_topic(id)),
                    usize::from(!subscribe && quotes),
                );
                assert_eq!(
                    msgbus::exact_subscriber_count_trades(get_trades_topic(id)),
                    usize::from(!subscribe && !quotes),
                );
            }
        });
    }

    #[rstest]
    fn test_python_identifier_list_rejects_invalid_inputs_before_dispatch(
        #[values(
            "subscribe_quotes",
            "unsubscribe_quotes",
            "subscribe_trades",
            "unsubscribe_trades"
        )]
        method: &str,
        #[values(false, true)] empty: bool,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = create_registered_actor(clock, cache, trader_id);
            let actor = Py::new(py, actor).unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );

            let argument = if empty {
                PyList::empty(py).into_any().unbind()
            } else {
                let instrument_id = InstrumentId::from("BTC/USD.SIM").into_py_any(py).unwrap();
                PyList::new(py, [instrument_id, 42.into_py_any(py).unwrap()])
                    .unwrap()
                    .into_any()
                    .unbind()
            };

            let error = actor
                .bind(py)
                .call_method1(method, (argument.bind(py),))
                .unwrap_err();

            if empty {
                assert!(error.is_instance_of::<pyo3::exceptions::PyValueError>(py));
                assert_eq!(
                    error.value(py).to_string(),
                    "instrument IDs cannot be empty"
                );
            } else {
                assert!(error.is_instance_of::<pyo3::exceptions::PyTypeError>(py));
                assert_eq!(
                    error.value(py).to_string(),
                    "expected InstrumentId or a sequence of InstrumentId values"
                );
            }

            assert_eq!(saver.get_messages().len(), 0);
        });
    }

    #[rstest]
    fn test_new_actor_creation() {
        let actor = PyDataActor::new(None);
        assert!(actor.trader_id().is_none());
    }

    #[rstest]
    fn test_actor_retains_python_config_object() {
        pyo3::Python::initialize();
        Python::attach(|py| {
            let config = py
                .eval(
                    c_str!("type('_Cfg', (), {'actor_id': 'A-RETAIN-001'})()"),
                    None,
                    None,
                )
                .unwrap();

            let actor = py
                .get_type::<PyDataActor>()
                .as_any()
                .call1((config.clone(),))
                .unwrap();

            let retained = actor.getattr("config").unwrap();

            assert!(retained.is(&config));
        });
    }

    #[rstest]
    fn test_prepare_python_actor_applies_shared_registration_boundary() {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let namespace = PyDict::new(py);
            namespace
                .set_item("DataActor", py.get_type::<PyDataActor>())
                .unwrap();
            py.run(
                c_str!(
                    r#"
class PreparedActor(DataActor):
    def __init__(self, _config=None):
        pass
"#
                ),
                Some(&namespace),
                None,
            )
            .unwrap();
            let actor_class = namespace.get_item("PreparedActor").unwrap();

            let configured = py
                .eval(
                    c_str!(
                        "type('_Cfg', (), {'actor_id': 'CONFIGURED-001', 'log_events': False, 'log_commands': False})()"
                    ),
                    None,
                    None,
                )
                .unwrap();
            let configured_actor = actor_class.call1((configured.clone(),)).unwrap();
            let configured_id = prepare_python_actor(&configured_actor, Some(&configured)).unwrap();
            let configured_ref = configured_actor.extract::<PyRef<PyDataActor>>().unwrap();
            let configured_wrapper = configured_ref.inner().python_instance().unwrap().unwrap();

            let fallback_actor = actor_class.call0().unwrap();
            let fallback_id = prepare_python_actor(&fallback_actor, None).unwrap();
            let fallback_ref = fallback_actor.extract::<PyRef<PyDataActor>>().unwrap();

            let invalid_logging = py
                .eval(
                    c_str!(
                        "type('_Cfg', (), {'actor_id': None, 'log_events': object(), 'log_commands': object()})()"
                    ),
                    None,
                    None,
                )
                .unwrap();
            let invalid_logging_actor = actor_class.call1((invalid_logging.clone(),)).unwrap();
            let invalid_logging_id =
                prepare_python_actor(&invalid_logging_actor, Some(&invalid_logging)).unwrap();
            let invalid_logging_ref = invalid_logging_actor
                .extract::<PyRef<PyDataActor>>()
                .unwrap();

            let invalid = py
                .eval(
                    c_str!("type('_Cfg', (), {'actor_id': object()})()"),
                    None,
                    None,
                )
                .unwrap();
            let invalid_actor = actor_class.call1((invalid.clone(),)).unwrap();
            let error = prepare_python_actor(&invalid_actor, Some(&invalid)).unwrap_err();

            assert_eq!(configured_id, ActorId::from("CONFIGURED-001"));
            assert_eq!(configured_ref.config.actor_id, Some(configured_id));
            assert!(!configured_ref.config.log_events);
            assert!(!configured_ref.config.log_commands);
            assert!(configured_wrapper.bind(py).is(&configured_actor));
            assert_eq!(fallback_id, ActorId::from("PreparedActor"));
            assert_eq!(fallback_ref.config.actor_id, Some(fallback_id));
            assert_eq!(invalid_logging_id, ActorId::from("PreparedActor"));
            assert!(invalid_logging_ref.config.log_events);
            assert!(invalid_logging_ref.config.log_commands);
            assert_eq!(error.to_string(), "Invalid `actor_id` type");
        });
    }

    #[rstest]
    fn test_clock_access_before_registration_raises_error() {
        let actor = PyDataActor::new(None);

        // Accessing clock before registration should raise PyRuntimeError
        let result = actor.py_clock();
        assert!(result.is_err());

        let error = result.unwrap_err();
        pyo3::Python::initialize();

        pyo3::Python::attach(|py| {
            assert!(error.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
        });

        let error_msg = error.to_string();
        assert!(
            error_msg.contains("Actor must be registered with a trader before accessing clock")
        );
    }

    #[rstest]
    fn test_unregistered_actor_methods_work() {
        let actor = create_unregistered_actor();

        assert!(!actor.py_is_ready());
        assert!(!actor.py_is_running());
        assert!(!actor.py_is_stopped());
        assert!(!actor.py_is_disposed());
        assert!(!actor.py_is_degraded());
        assert!(!actor.py_is_faulted());

        // Verify unregistered state
        assert_eq!(actor.trader_id(), None);
    }

    #[rstest]
    fn test_registration_success(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        let mut actor = create_unregistered_actor();
        actor.register(trader_id, clock, cache).unwrap();
        assert!(actor.trader_id().is_some());
        assert_eq!(actor.trader_id().unwrap(), trader_id);
    }

    #[rstest]
    fn test_registered_actor_basic_properties(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        let actor = create_registered_actor(clock, cache, trader_id);

        assert_eq!(actor.state(), ComponentState::Ready);
        assert_eq!(actor.trader_id(), Some(TraderId::from("TRADER-001")));
        assert!(actor.py_is_ready());
        assert!(!actor.py_is_running());
        assert!(!actor.py_is_stopped());
        assert!(!actor.py_is_disposed());
        assert!(!actor.py_is_degraded());
        assert!(!actor.py_is_faulted());
    }

    #[rstest]
    fn test_basic_subscription_methods_compile(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        data_type: DataType,
        client_id: ClientId,
        audusd_sim: CurrencyPair,
    ) {
        let mut actor = create_registered_actor(clock, cache, trader_id);

        pyo3::Python::initialize();

        pyo3::Python::attach(|py| {
            assert!(
                actor
                    .py_subscribe_data(py, data_type.clone(), Some(client_id), None)
                    .is_ok()
            );
            assert!(
                actor
                    .py_subscribe_quotes(
                        py,
                        audusd_sim.id.into_py_any(py).unwrap().bind(py),
                        Some(client_id),
                        None
                    )
                    .is_ok()
            );
            assert!(
                actor
                    .py_unsubscribe_data(py, data_type, Some(client_id), None)
                    .is_ok()
            );
            assert!(
                actor
                    .py_unsubscribe_quotes(
                        py,
                        audusd_sim.id.into_py_any(py).unwrap().bind(py),
                        Some(client_id),
                        None
                    )
                    .is_ok()
            );
        });
    }

    #[rstest]
    fn test_shutdown_system_passes_through(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        let actor = create_registered_actor(clock, cache, trader_id);

        assert!(
            actor
                .py_shutdown_system(Some("Test shutdown".to_string()))
                .is_ok()
        );
        assert!(actor.py_shutdown_system(None).is_ok());
    }

    #[rstest]
    fn test_publish_data_delivers_to_any_subscriber(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::msgbus::{
            self, MessageBus, get_message_bus, switchboard::get_custom_topic,
            typed_handler::ShareableMessageHandler,
        };

        // Ensure clean msgbus for this test
        *get_message_bus().borrow_mut() = MessageBus::default();

        let actor = create_registered_actor(clock, cache, trader_id);
        let data = stub_custom_data(1, 42, None, None);
        let topic = get_custom_topic(&data.data_type);

        let received: Rc<RefCell<Vec<CustomData>>> = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();
        let handler = ShareableMessageHandler::from_typed(move |d: &CustomData| {
            received_clone.borrow_mut().push(d.clone());
        });
        msgbus::subscribe_any(topic.into(), handler, None);

        actor.py_publish_data(&data.data_type, &data).unwrap();

        let received = received.borrow();
        assert_eq!(received.len(), 1);
        assert_eq!(received[0].data_type, data.data_type);
    }

    #[rstest]
    fn test_publish_signal_delivers_to_customdata_subscriber(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::{
            msgbus::{
                self, MessageBus, Pattern, get_message_bus, typed_handler::ShareableMessageHandler,
            },
            signal::Signal,
        };

        *get_message_bus().borrow_mut() = MessageBus::default();

        let actor = create_registered_actor(clock, cache, trader_id);

        // Signals travel as `CustomData` on the bus so persistence and other
        // `CustomData`-aware subscribers pick them up. Downcast inside the handler.
        let received: Rc<RefCell<Vec<Signal>>> = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();
        let handler = ShareableMessageHandler::from_typed(move |data: &CustomData| {
            if let Some(sig) = data.data.as_any().downcast_ref::<Signal>() {
                received_clone.borrow_mut().push(sig.clone());
            }
        });
        let pattern: crate::msgbus::MStr<Pattern> = "data.Signal*".to_string().into();
        msgbus::subscribe_any(pattern, handler, None);

        pyo3::Python::initialize();

        Python::attach(|py| {
            let val1: Py<PyAny> = 1.0_f64.into_py_any_unwrap(py);
            let val2: Py<PyAny> = "HIGH".into_py_any_unwrap(py);
            actor.py_publish_signal(py, "example", val1, 0).unwrap();
            actor
                .py_publish_signal(py, "risk", val2, 1_700_000_000_000_000_000)
                .unwrap();
        });

        let received = received.borrow();
        assert_eq!(received.len(), 2);
        assert_eq!(received[0].name, "example");
        assert_eq!(received[0].value, "1.0");
        assert_eq!(received[1].name, "risk");
        assert_eq!(received[1].value, "HIGH");
        assert_eq!(
            received[1].ts_event,
            UnixNanos::from(1_700_000_000_000_000_000_u64)
        );
    }

    #[rstest]
    fn test_publish_signal_accepts_numeric_py_values(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::{
            msgbus::{
                self, MessageBus, Pattern, get_message_bus, typed_handler::ShareableMessageHandler,
            },
            signal::Signal,
        };

        *get_message_bus().borrow_mut() = MessageBus::default();

        let actor = create_registered_actor(clock, cache, trader_id);

        let received: Rc<RefCell<Vec<Signal>>> = Rc::new(RefCell::new(Vec::new()));
        let received_clone = received.clone();
        let handler = ShareableMessageHandler::from_typed(move |data: &CustomData| {
            if let Some(sig) = data.data.as_any().downcast_ref::<Signal>() {
                received_clone.borrow_mut().push(sig.clone());
            }
        });
        let pattern: crate::msgbus::MStr<Pattern> = "data.Signal*".to_string().into();
        msgbus::subscribe_any(pattern, handler, None);

        pyo3::Python::initialize();

        Python::attach(|py| {
            let int_value: Py<PyAny> = 42_i64.into_py_any_unwrap(py);
            let float_value: Py<PyAny> = 3.5_f64.into_py_any_unwrap(py);
            let bool_value: Py<PyAny> = true.into_py_any_unwrap(py);
            actor.py_publish_signal(py, "count", int_value, 0).unwrap();
            actor
                .py_publish_signal(py, "ratio", float_value, 0)
                .unwrap();
            actor
                .py_publish_signal(py, "active", bool_value, 0)
                .unwrap();
        });

        let received = received.borrow();
        assert_eq!(received.len(), 3);
        assert_eq!(received[0].value, "42");
        assert_eq!(received[1].value, "3.5");
        assert_eq!(received[2].value, "True");
    }

    #[rstest]
    fn test_subscribe_and_unsubscribe_signal_compile(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::msgbus::{MessageBus, get_message_bus};

        *get_message_bus().borrow_mut() = MessageBus::default();

        let actor = create_registered_actor(clock, cache, trader_id);
        Python::initialize();
        Python::attach(|py| {
            let actor = Bound::new(py, actor).unwrap();
            PyDataActor::py_subscribe_signal(&actor, "example", None).unwrap();
            PyDataActor::py_unsubscribe_signal(&actor, "example").unwrap();
            PyDataActor::py_subscribe_signal(&actor, "", None).unwrap();
            PyDataActor::py_unsubscribe_signal(&actor, "").unwrap();
        });
    }

    #[rstest]
    fn test_py_subscribe_signal_forwards_priority(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::msgbus::{MessageBus, get_message_bus, switchboard::get_signal_topic};

        *get_message_bus().borrow_mut() = MessageBus::default();

        let actor = create_registered_actor(clock, cache, trader_id);
        Python::initialize();
        Python::attach(|py| {
            let actor = Bound::new(py, actor).unwrap();
            PyDataActor::py_subscribe_signal(&actor, "trigger", Some(50)).unwrap();

            // The PyO3 binding must forward the priority to the bus unchanged.
            let topic = get_signal_topic("trigger");
            let subs = get_message_bus().borrow_mut().matching_subscriptions(topic);
            assert_eq!(subs.len(), 1);
            assert_eq!(subs[0].priority, 50);
        });
    }

    #[rstest]
    fn test_register_in_global_registries_retains_python_wrapper(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(Some(DataActorConfig {
                actor_id: Some(ActorId::from("RETAINED-ACTOR")),
                ..Default::default()
            }));
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();

            rust_actor.register_in_global_registries().unwrap();

            let retained = get_python_wrapper(ComponentId::from("RETAINED-ACTOR"))
                .expect("registering must retain the actor's Python wrapper");

            assert!(retained.bind(py).is(py_actor.bind(py)));
            assert!(get_component(&Ustr::from("RETAINED-ACTOR")).is_some());
            assert!(actor_exists(&Ustr::from("RETAINED-ACTOR")));
        });
    }

    #[rstest]
    fn test_register_in_global_registries_rejects_missing_python_wrapper(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|_py| {
            let mut rust_actor = PyDataActor::new(Some(DataActorConfig {
                actor_id: Some(ActorId::from("UNWRAPPED-ACTOR")),
                ..Default::default()
            }));
            rust_actor.register(trader_id, clock, cache).unwrap();

            let error = rust_actor
                .register_in_global_registries()
                .expect_err("registering without a Python wrapper must fail");

            assert!(error.to_string().contains("without a Python wrapper"));
            assert!(get_component(&Ustr::from("UNWRAPPED-ACTOR")).is_none());
            assert!(!actor_exists(&Ustr::from("UNWRAPPED-ACTOR")));
            assert!(get_python_wrapper(ComponentId::from("UNWRAPPED-ACTOR")).is_none());
        });
    }

    #[rstest]
    fn test_publish_data_dispatches_to_python_on_data(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::msgbus::{MessageBus, get_message_bus};

        *get_message_bus().borrow_mut() = MessageBus::default();

        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            rust_actor.register_in_global_registries().unwrap();
            rust_actor.py_start().unwrap();

            let data = stub_custom_data(1, 42, None, None);
            rust_actor
                .py_subscribe_data(py, data.data_type.clone(), None, None)
                .unwrap();

            rust_actor.py_publish_data(&data.data_type, &data).unwrap();
            rust_actor.py_publish_data(&data.data_type, &data).unwrap();

            assert!(python_method_was_called(&py_actor, py, "on_data"));
            assert_eq!(python_method_call_count(&py_actor, py, "on_data"), 2);
        });
    }

    #[rstest]
    fn test_publish_signal_dispatches_to_python_on_signal(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::msgbus::{MessageBus, get_message_bus};

        *get_message_bus().borrow_mut() = MessageBus::default();

        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            rust_actor.register_in_global_registries().unwrap();
            rust_actor.py_start().unwrap();

            let bound_actor = Bound::new(py, rust_actor).unwrap();
            PyDataActor::py_subscribe_signal(&bound_actor, "example", None).unwrap();
            let rust_actor = bound_actor.borrow_mut();
            let val1: Py<PyAny> = "1.5".into_py_any_unwrap(py);
            let val2: Py<PyAny> = 2.0_f64.into_py_any_unwrap(py);
            rust_actor
                .py_publish_signal(py, "example", val1, 0)
                .unwrap();
            rust_actor
                .py_publish_signal(py, "example", val2, 1_700_000_000_000_000_000)
                .unwrap();

            assert!(python_method_was_called(&py_actor, py, "on_signal"));
            assert_eq!(python_method_call_count(&py_actor, py, "on_signal"), 2);
        });
    }

    #[rstest]
    fn test_unsubscribe_signal_stops_python_dispatch(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::msgbus::{MessageBus, get_message_bus};

        *get_message_bus().borrow_mut() = MessageBus::default();

        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            rust_actor.register_in_global_registries().unwrap();
            rust_actor.py_start().unwrap();

            let bound_actor = Bound::new(py, rust_actor).unwrap();
            PyDataActor::py_subscribe_signal(&bound_actor, "example", None).unwrap();
            let rust_actor = bound_actor.borrow_mut();
            let val1: Py<PyAny> = "1".into_py_any_unwrap(py);
            let val2: Py<PyAny> = "2".into_py_any_unwrap(py);
            rust_actor
                .py_publish_signal(py, "example", val1, 0)
                .unwrap();

            drop(rust_actor);
            PyDataActor::py_unsubscribe_signal(&bound_actor, "example").unwrap();
            let rust_actor = bound_actor.borrow_mut();
            rust_actor
                .py_publish_signal(py, "example", val2, 0)
                .unwrap();

            assert_eq!(python_method_call_count(&py_actor, py, "on_signal"), 1);
        });
    }

    #[rstest]
    #[case(None)]
    #[case(Some(SystemChannel::ExecCommands))]
    fn test_queue_state_changed_subscription_dispatches_and_unsubscribes(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] channel: Option<SystemChannel>,
    ) {
        *get_message_bus().borrow_mut() = MessageBus::default();

        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            rust_actor.register_in_global_registries().unwrap();
            rust_actor.py_start().unwrap();
            rust_actor
                .py_subscribe_queue_state(channel, Some(50))
                .unwrap();

            let topic =
                MessagingSwitchboard::queue_state_changed_topic(SystemChannel::ExecCommands);
            let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
            assert_eq!(subscriptions.len(), 1);
            assert_eq!(subscriptions[0].priority, 50);
            let unrelated =
                MessagingSwitchboard::queue_state_changed_topic(SystemChannel::DataEvents);
            assert_eq!(
                get_message_bus()
                    .borrow_mut()
                    .matching_subscriptions(unrelated)
                    .len(),
                usize::from(channel.is_none())
            );

            let triggered = sample_queue_state_changed(QueueState::Triggered);
            msgbus::publish_any(topic, &triggered);

            let (received,) = py_actor
                .call_method1(py, "last_call_args", ("on_queue_state",))
                .unwrap()
                .extract::<(QueueStateChanged,)>(py)
                .unwrap();
            assert_eq!(received, triggered);

            rust_actor.py_unsubscribe_queue_state(channel).unwrap();
            let cleared = sample_queue_state_changed(QueueState::Cleared);
            msgbus::publish_any(topic, &cleared);

            assert_eq!(python_method_call_count(&py_actor, py, "on_queue_state"), 1);
        });
    }

    #[rstest]
    #[case(None, None)]
    #[case(Some(ClientId::from("BINANCE")), None)]
    #[case(None, Some("binance-futures-market-streams"))]
    #[case(
        Some(ClientId::from("BINANCE")),
        Some("binance-futures-market-streams")
    )]
    fn test_socket_state_changed_subscription_dispatches_and_unsubscribes(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] client_id: Option<ClientId>,
        #[case] endpoint: Option<&str>,
    ) {
        *get_message_bus().borrow_mut() = MessageBus::default();

        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            rust_actor.register_in_global_registries().unwrap();
            rust_actor.py_start().unwrap();
            rust_actor
                .py_subscribe_socket_state(client_id, endpoint, Some(50))
                .unwrap();

            let topic = MessagingSwitchboard::socket_state_changed_topic(
                ClientId::from("BINANCE"),
                "binance-futures-market-streams",
            );
            let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
            assert_eq!(subscriptions.len(), 1);
            assert_eq!(subscriptions[0].priority, 50);
            let unrelated =
                MessagingSwitchboard::socket_state_changed_topic(ClientId::from("BYBIT"), "orders");
            assert_eq!(
                get_message_bus()
                    .borrow_mut()
                    .matching_subscriptions(unrelated)
                    .len(),
                usize::from(client_id.is_none() && endpoint.is_none())
            );

            let connected = sample_socket_state_changed(SocketState::Connected);
            msgbus::publish_any(topic, &connected);

            let (received,) = py_actor
                .call_method1(py, "last_call_args", ("on_socket_state",))
                .unwrap()
                .extract::<(SocketStateChanged,)>(py)
                .unwrap();
            assert_eq!(received, connected);

            rust_actor
                .py_unsubscribe_socket_state(client_id, endpoint)
                .unwrap();
            let disconnected = sample_socket_state_changed(SocketState::Disconnected);
            msgbus::publish_any(topic, &disconnected);

            assert_eq!(
                python_method_call_count(&py_actor, py, "on_socket_state"),
                1
            );
        });
    }

    #[rstest]
    fn test_subscribe_signal_wildcard_dispatches_all_names_to_python(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use crate::msgbus::{MessageBus, get_message_bus};

        *get_message_bus().borrow_mut() = MessageBus::default();

        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            rust_actor.register_in_global_registries().unwrap();
            rust_actor.py_start().unwrap();

            let bound_actor = Bound::new(py, rust_actor).unwrap();
            PyDataActor::py_subscribe_signal(&bound_actor, "", None).unwrap();
            let rust_actor = bound_actor.borrow_mut();
            let val1: Py<PyAny> = "1".into_py_any_unwrap(py);
            let val2: Py<PyAny> = "2".into_py_any_unwrap(py);
            let val3: Py<PyAny> = "3".into_py_any_unwrap(py);
            rust_actor.py_publish_signal(py, "alpha", val1, 0).unwrap();
            rust_actor.py_publish_signal(py, "beta", val2, 0).unwrap();
            rust_actor.py_publish_signal(py, "gamma", val3, 0).unwrap();

            assert_eq!(python_method_call_count(&py_actor, py, "on_signal"), 3);
        });
    }

    #[rstest]
    fn test_signal_customdata_unwraps_to_python_signal(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        // Exercises the `Signal::to_pyobject` path: a `CustomData` wrapping a
        // `Signal` reaches Python `on_data`, and the PyO3 `.data` getter must
        // successfully unwrap the inner `Arc<dyn CustomDataTrait>` into a
        // Python `Signal`. Without `Signal::to_pyobject`, the getter raises
        // `TypeError` and this assertion fails.
        use crate::msgbus::{MessageBus, get_message_bus};

        *get_message_bus().borrow_mut() = MessageBus::default();

        pyo3::Python::initialize();

        Python::attach(|py| {
            let capture_code = c_str!(
                r#"
class CapturingActor:
    def __init__(self):
        self.captured = []

    def on_start(self): pass
    def on_stop(self): pass
    def on_resume(self): pass
    def on_reset(self): pass
    def on_dispose(self): pass
    def on_degrade(self): pass
    def on_fault(self): pass
    def on_signal(self, signal): pass

    def on_data(self, custom):
        # Exercise the CustomData.data getter: raises TypeError if the
        # inner payload cannot be converted back to a Python object.
        inner = custom.data
        self.captured.append((type(inner).__name__, inner.name, inner.value))
"#
            );
            py.run(capture_code, None, None).unwrap();
            let cls = py.eval(c_str!("CapturingActor"), None, None).unwrap();
            let py_actor: Py<PyAny> = cls.call0().unwrap().unbind();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            rust_actor.register_in_global_registries().unwrap();
            rust_actor.py_start().unwrap();

            // Subscribe as custom-data for the signal's advertised DataType
            // (`data.SignalExample`) so `on_data` fires with the wrapping CustomData.
            let data_type = DataType::new("SignalExample", None, None);
            rust_actor
                .py_subscribe_data(py, data_type, None, None)
                .unwrap();

            let val: Py<PyAny> = "1.5".into_py_any_unwrap(py);
            rust_actor.py_publish_signal(py, "example", val, 0).unwrap();

            let captured = py_actor
                .bind(py)
                .getattr("captured")
                .unwrap()
                .extract::<Vec<(String, String, String)>>()
                .unwrap();
            assert_eq!(captured.len(), 1);
            assert_eq!(captured[0].0, "Signal");
            assert_eq!(captured[0].1, "example");
            assert_eq!(captured[0].2, "1.5");
        });
    }

    #[rstest]
    fn test_add_and_update_synthetic_via_pyo3(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        use nautilus_model::{
            identifiers::{InstrumentId, Symbol},
            instruments::SyntheticInstrument,
        };

        let actor = create_registered_actor(clock, cache.clone(), trader_id);

        let comp1 = InstrumentId::from_str("BTC-USD.VENUE").unwrap();
        let comp2 = InstrumentId::from_str("ETH-USD.VENUE").unwrap();
        let formula = format!("({comp1} + {comp2}) / 2.0");
        let synthetic = SyntheticInstrument::builder()
            .symbol(Symbol::from("SYN"))
            .price_precision(2)
            .components(vec![comp1, comp2])
            .formula(&formula)
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();
        let synthetic_id = synthetic.id;

        actor.py_add_synthetic(synthetic.clone()).unwrap();
        assert!(cache.borrow().synthetic(&synthetic_id).is_some());

        // Adding again raises
        assert!(actor.py_add_synthetic(synthetic).is_err());

        let new_formula = format!("{comp1} + {comp2}");
        let updated = SyntheticInstrument::builder()
            .symbol(Symbol::from("SYN"))
            .price_precision(2)
            .components(vec![comp1, comp2])
            .formula(&new_formula)
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();
        actor.py_update_synthetic(updated).unwrap();
        assert_eq!(
            cache.borrow().synthetic(&synthetic_id).unwrap().formula,
            new_formula
        );

        // Updating a non-existent raises
        let missing = SyntheticInstrument::builder()
            .symbol(Symbol::from("GONE"))
            .price_precision(2)
            .components(vec![comp1, comp2])
            .formula(&formula)
            .ts_event(UnixNanos::default())
            .ts_init(UnixNanos::default())
            .build()
            .unwrap();
        assert!(actor.py_update_synthetic(missing).is_err());
    }

    #[rstest]
    fn test_book_at_interval_invalid_interval_ms(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        audusd_sim: CurrencyPair,
    ) {
        pyo3::Python::initialize();

        let mut actor = create_registered_actor(clock, cache, trader_id);

        pyo3::Python::attach(|py| {
            let result = actor.py_subscribe_book_at_interval(
                py,
                &pyo3::Py::new(py, audusd_sim.id)
                    .unwrap()
                    .into_bound(py)
                    .into_any(),
                BookType::L2_MBP,
                0,
                None,
                None,
                None,
            );
            assert!(result.is_err());
            assert_eq!(
                result.unwrap_err().to_string(),
                "ValueError: interval_ms must be > 0"
            );

            let result = actor.py_unsubscribe_book_at_interval(
                py,
                &pyo3::Py::new(py, audusd_sim.id)
                    .unwrap()
                    .into_bound(py)
                    .into_any(),
                0,
                None,
                None,
            );
            assert!(result.is_err());
            assert_eq!(
                result.unwrap_err().to_string(),
                "ValueError: interval_ms must be > 0"
            );
        });
    }

    #[rstest]
    #[case(None)]
    #[case(Some(25))]
    fn test_book_depth_subscription_methods_manage_handler(
        #[case] depth: Option<usize>,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        audusd_sim: CurrencyPair,
    ) {
        pyo3::Python::initialize();

        let mut actor = create_registered_actor(clock, cache, trader_id);

        Python::attach(|py| {
            actor
                .py_subscribe_book_depth(
                    py,
                    &pyo3::Py::new(py, audusd_sim.id)
                        .unwrap()
                        .into_bound(py)
                        .into_any(),
                    BookType::L2_MBP,
                    depth,
                    None,
                    false,
                    None,
                )
                .unwrap();
            assert_eq!(actor.inner().depth_handler_count(), 1);

            actor
                .py_unsubscribe_book_depth(
                    py,
                    &pyo3::Py::new(py, audusd_sim.id)
                        .unwrap()
                        .into_bound(py)
                        .into_any(),
                    None,
                    None,
                )
                .unwrap();
            assert_eq!(actor.inner().depth_handler_count(), 0);
        });
    }

    #[rstest]
    fn test_request_methods_signatures_exist() {
        let actor = create_unregistered_actor();
        assert!(actor.trader_id().is_none());
    }

    #[rstest]
    #[case("request_instrument", "Instrument")]
    #[case("request_book_snapshot", "BookSnapshot")]
    #[case("request_book_deltas", "BookDeltas")]
    #[case("request_book_depth", "BookDepth")]
    #[case("request_quotes", "Quotes")]
    #[case("request_trades", "Trades")]
    #[case("request_funding_rates", "FundingRates")]
    #[case("request_bars", "Bars")]
    fn test_python_request_identifier_shapes(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] method: &str,
        #[case] variant: &str,
        #[values("scalar", "list", "tuple")] shape: &str,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );

            let (first, second, key) = if variant == "Bars" {
                let first = BarType::from_str("AUD/USD.SIM-1-MINUTE-LAST-EXTERNAL").unwrap();
                let second = BarType::from_str("NZD/USD.SIM-5-MINUTE-LAST-EXTERNAL").unwrap();
                (
                    first.into_py_any(py).unwrap(),
                    second.into_py_any(py).unwrap(),
                    "bar_types",
                )
            } else {
                (
                    InstrumentId::from("AUD/USD.SIM").into_py_any(py).unwrap(),
                    InstrumentId::from("NZD/USD.SIM").into_py_any(py).unwrap(),
                    "instrument_ids",
                )
            };

            let argument = match shape {
                "scalar" => first.bind(py).clone(),
                "list" => PyList::new(py, [&first, &second, &first])
                    .unwrap()
                    .into_any(),
                "tuple" => PyTuple::new(py, [&first, &second, &first])
                    .unwrap()
                    .into_any(),
                _ => unreachable!(),
            };

            let result = actor.bind(py).call_method1(method, (&argument,));
            let commands = saver.get_messages();

            let root = result
                .unwrap()
                .extract::<String>()
                .unwrap()
                .parse::<UUID4>()
                .unwrap();
            assert_eq!(commands.len(), 1);

            let DataCommand::Request(command) = &commands[0] else {
                panic!("expected request");
            };

            assert_eq!(command.request_id(), &root);
            let wire = serde_json::to_value(command).unwrap();

            let count = if shape == "scalar" { 1 } else { 2 };

            let expected = if variant == "Bars" {
                serde_json::to_value(
                    [
                        "AUD/USD.SIM-1-MINUTE-LAST-EXTERNAL",
                        "NZD/USD.SIM-5-MINUTE-LAST-EXTERNAL",
                    ][..count]
                        .to_vec(),
                )
                .unwrap()
            } else {
                serde_json::to_value(["AUD/USD.SIM", "NZD/USD.SIM"][..count].to_vec()).unwrap()
            };

            assert_eq!(wire[variant][key], expected);
            assert!(actor.borrow(py).inner().core.is_pending_request(&root));
        });
    }

    #[rstest]
    fn test_python_request_rejects_invalid_identifiers_before_dispatch(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[values(
            "request_instrument",
            "request_book_snapshot",
            "request_book_deltas",
            "request_book_depth",
            "request_quotes",
            "request_trades",
            "request_funding_rates",
            "request_bars"
        )]
        method: &str,
        #[values(false, true)] empty: bool,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );

            let argument = if empty {
                PyList::empty(py)
            } else {
                PyList::new(py, [42]).unwrap()
            };

            let error = actor
                .bind(py)
                .call_method1(method, (argument,))
                .unwrap_err();

            let expected = match (method == "request_bars", empty) {
                (true, true) => "ValueError: bar types cannot be empty",
                (false, true) => "ValueError: instrument IDs cannot be empty",
                (true, false) => "TypeError: expected BarType or a sequence of BarType values",
                (false, false) => {
                    "TypeError: expected InstrumentId or a sequence of InstrumentId values"
                }
            };

            assert_eq!(error.to_string(), expected);
            assert!(saver.get_messages().is_empty());
        });
    }

    #[rstest]
    fn test_data_actor_trait_implementation(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        let actor = create_registered_actor(clock, cache, trader_id);
        let state = actor.state();
        assert_eq!(state, ComponentState::Ready);
    }

    #[rstest]
    fn test_python_reconnect_socket_enqueues_typed_command(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        let (system_tx, mut system_rx) = tokio::sync::mpsc::unbounded_channel();
        replace_system_command_sender(system_tx);
        let actor = create_registered_actor(clock, cache, trader_id);

        actor
            .py_reconnect_socket(ClientId::from("POLYMARKET"), "polymarket-market-streams")
            .expect("valid reconnect command");
        let command = system_rx
            .try_recv()
            .expect("reconnect command should be queued");
        let SystemCommand::ReconnectSocket(command) = command;

        assert_eq!(command.trader_id, trader_id);
        assert_eq!(command.client_id, ClientId::from("POLYMARKET"));
        assert_eq!(command.endpoint, "polymarket-market-streams");
        assert_eq!(command.ts_init, UnixNanos::default());
    }

    #[rstest]
    fn test_python_reconnect_socket_returns_errors_without_panicking() {
        let actor = create_unregistered_actor();

        assert!(
            actor
                .py_reconnect_socket(ClientId::from("POLYMARKET"), "polymarket-market-streams")
                .is_err()
        );
        assert!(
            actor
                .py_reconnect_socket(ClientId::from("POLYMARKET"), "wss://secret.example")
                .is_err()
        );
    }

    fn sample_instrument() -> CurrencyPair {
        audusd_sim()
    }

    fn sample_data() -> CustomData {
        stub_custom_data(1, 42, None, None)
    }

    fn sample_time_event() -> TimeEvent {
        TimeEvent::new(
            Ustr::from("test_timer"),
            UUID4::new(),
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_signal() -> Signal {
        Signal::new(
            Ustr::from("test_signal"),
            "1.0".to_string(),
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_queue_state_changed(state: QueueState) -> QueueStateChanged {
        QueueStateChanged::new(
            TraderId::from("TRADER-001"),
            SystemChannel::ExecCommands,
            QueueCondition::Backlogged,
            state,
            17,
            23,
            UUID4::from("00000000-0000-4000-8000-000000000001"),
            UnixNanos::from(1_700_000_000_000_000_001),
            UnixNanos::from(1_700_000_000_000_000_002),
        )
    }

    fn sample_socket_state_changed(state: SocketState) -> SocketStateChanged {
        SocketStateChanged::new(
            TraderId::from("TRADER-001"),
            ClientId::from("BINANCE"),
            Some(Venue::from("BINANCE")),
            Ustr::from("binance-futures-market-streams"),
            state,
            UUID4::from("00000000-0000-4000-8000-000000000001"),
            UnixNanos::from(1_700_000_000_000_000_001),
            UnixNanos::from(1_700_000_000_000_000_002),
        )
    }

    fn sample_quote() -> QuoteTick {
        let instrument = sample_instrument();
        QuoteTick::new(
            instrument.id,
            Price::from("1.00000"),
            Price::from("1.00001"),
            Quantity::from(100_000),
            Quantity::from(100_000),
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_trade() -> TradeTick {
        let instrument = sample_instrument();
        TradeTick::new(
            instrument.id,
            Price::from("1.00000"),
            Quantity::from(100_000),
            AggressorSide::Buy,
            TradeId::new("123456"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_bar() -> Bar {
        let instrument = sample_instrument();
        let bar_type =
            BarType::from_str(&format!("{}-1-MINUTE-LAST-INTERNAL", instrument.id)).unwrap();
        Bar::new(
            bar_type,
            Price::from("1.00000"),
            Price::from("1.00010"),
            Price::from("0.99990"),
            Price::from("1.00005"),
            Quantity::from(100_000),
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_book() -> OrderBook {
        OrderBook::new(sample_instrument().id, BookType::L2_MBP)
    }

    fn sample_book_deltas() -> OrderBookDeltas {
        let instrument = sample_instrument();
        let delta =
            OrderBookDelta::clear(instrument.id, 0, UnixNanos::default(), UnixNanos::default());
        OrderBookDeltas::new(instrument.id, vec![delta])
    }

    fn sample_book_depth() -> OrderBookDepth {
        stub_depth10()
    }

    fn sample_mark_price() -> MarkPriceUpdate {
        MarkPriceUpdate::new(
            sample_instrument().id,
            Price::from("1.00000"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_index_price() -> IndexPriceUpdate {
        IndexPriceUpdate::new(
            sample_instrument().id,
            Price::from("1.00000"),
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_funding_rate() -> FundingRateUpdate {
        FundingRateUpdate::new(
            sample_instrument().id,
            "0.0001".parse().unwrap(),
            None,
            None,
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_instrument_status() -> InstrumentStatus {
        InstrumentStatus::new(
            sample_instrument().id,
            MarketStatusAction::Trading,
            UnixNanos::default(),
            UnixNanos::default(),
            None,
            None,
            None,
            None,
            None,
        )
    }

    fn sample_instrument_close() -> InstrumentClose {
        InstrumentClose::new(
            sample_instrument().id,
            Price::from("1.00000"),
            InstrumentCloseType::EndOfSession,
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    fn sample_option_greeks() -> OptionGreeks {
        OptionGreeks {
            instrument_id: sample_instrument().id,
            convention: GreeksConvention::BlackScholes,
            greeks: OptionGreekValues {
                delta: 0.55,
                gamma: 0.03,
                vega: 0.12,
                theta: -0.05,
                rho: 0.01,
            },
            mark_iv: Some(0.25),
            bid_iv: None,
            ask_iv: None,
            underlying_price: None,
            open_interest: None,
            ts_event: UnixNanos::default(),
            ts_init: UnixNanos::default(),
        }
    }

    fn sample_option_chain() -> OptionChainSlice {
        OptionChainSlice {
            series_id: OptionSeriesId::new_derived(
                Venue::from("SIM"),
                Ustr::from("AUD"),
                Ustr::from("USD"),
                UnixNanos::from(1_711_036_800_000_000_000),
            ),
            atm_strike: None,
            calls: Default::default(),
            puts: Default::default(),
            ts_event: UnixNanos::default(),
            ts_init: UnixNanos::default(),
        }
    }

    #[cfg(feature = "defi")]
    fn sample_block() -> Block {
        Block::new(
            "0x1234567890abcdef".to_string(),
            "0xabcdef1234567890".to_string(),
            12345,
            "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0".into(),
            21000,
            20000,
            UnixNanos::default(),
            Some(Blockchain::Ethereum),
        )
    }

    #[cfg(feature = "defi")]
    fn sample_pool_components() -> (Arc<Chain>, Arc<Dex>, Pool) {
        let chain = Arc::new(Chain::new(Blockchain::Ethereum, 1));
        let dex = Arc::new(Dex::new(
            Chain::new(Blockchain::Ethereum, 1),
            DexType::UniswapV3,
            "0x1F98431c8aD98523631AE4a59f267346ea31F984",
            0,
            AmmType::CLAMM,
            "PoolCreated",
            "Swap",
            "Mint",
            "Burn",
            "Collect",
        ));
        let token0 = Token::new(
            chain.clone(),
            "0xa0b86a33e6441c8c06dd7b111a8c4e82e2b2a5e1"
                .parse()
                .unwrap(),
            "USDC".into(),
            "USD Coin".into(),
            6,
        );
        let token1 = Token::new(
            chain.clone(),
            "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"
                .parse()
                .unwrap(),
            "WETH".into(),
            "Wrapped Ether".into(),
            18,
        );
        let pool_address = "0x8ad599c3A0ff1De082011EFDDc58f1908eb6e6D8"
            .parse()
            .unwrap();
        let pool_identifier: PoolIdentifier = "0x8ad599c3A0ff1De082011EFDDc58f1908eb6e6D8"
            .parse()
            .unwrap();

        let pool = Pool::new(
            chain.clone(),
            dex.clone(),
            pool_address,
            pool_identifier,
            12345,
            token0,
            token1,
            Some(500),
            Some(10),
            UnixNanos::default(),
        );

        (chain, dex, pool)
    }

    #[cfg(feature = "defi")]
    fn sample_pool_swap() -> PoolSwap {
        let (chain, dex, pool) = sample_pool_components();
        PoolSwap::new(
            chain,
            dex,
            pool.instrument_id,
            pool.pool_identifier,
            12345,
            "0xabc123".to_string(),
            0,
            0,
            UnixNanos::default(),
            UnixNanos::default(),
            "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0"
                .parse()
                .unwrap(),
            "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0"
                .parse()
                .unwrap(),
            I256::from_str("1000000000000000000").unwrap(),
            I256::from_str("400000000000000").unwrap(),
            U160::from(59000000000000u128),
            1000000,
            100,
        )
    }

    #[cfg(feature = "defi")]
    fn sample_pool_liquidity_update() -> PoolLiquidityUpdate {
        let (chain, dex, pool) = sample_pool_components();
        PoolLiquidityUpdate::new(
            chain,
            dex,
            pool.instrument_id,
            pool.pool_identifier,
            PoolLiquidityUpdateType::Mint,
            12345,
            "0xabc123".to_string(),
            0,
            0,
            Some(
                "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0"
                    .parse()
                    .unwrap(),
            ),
            "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0"
                .parse()
                .unwrap(),
            1000,
            U256::from(1_000u64),
            U256::from(2_000u64),
            -10,
            10,
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    #[cfg(feature = "defi")]
    fn sample_pool_fee_collect() -> PoolFeeCollect {
        let (chain, dex, pool) = sample_pool_components();
        PoolFeeCollect::new(
            chain,
            dex,
            pool.instrument_id,
            pool.pool_identifier,
            12345,
            "0xabc123".to_string(),
            0,
            0,
            "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0"
                .parse()
                .unwrap(),
            100,
            200,
            -10,
            10,
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    #[cfg(feature = "defi")]
    fn sample_pool_flash() -> PoolFlash {
        let (chain, dex, pool) = sample_pool_components();
        PoolFlash::new(
            chain,
            dex,
            pool.instrument_id,
            pool.pool_identifier,
            12345,
            "0xabc123".to_string(),
            0,
            0,
            UnixNanos::default(),
            UnixNanos::default(),
            "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0"
                .parse()
                .unwrap(),
            "0x742E4422b21FB8B4dF463F28689AC98bD56c39e0"
                .parse()
                .unwrap(),
            U256::from(100u64),
            U256::from(200u64),
            U256::from(101u64),
            U256::from(201u64),
        )
    }

    #[rstest]
    #[case("request_data")]
    #[case("request_instrument")]
    #[case("request_instruments")]
    #[case("request_book_snapshot")]
    #[case("request_book_deltas")]
    #[case("request_book_depth")]
    #[case("request_quotes")]
    #[case("request_trades")]
    #[case("request_funding_rates")]
    #[case("request_bars")]
    fn test_python_request_completion_callback_and_pending(
        #[case] method: &str,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let module = PyModule::from_code(
                py,
                c_str!(
                    r#"
class CompletionActor:
    def __init__(self, native):
        self.native = native
        self.events = []
    def on_data_response(self, response):
        self.events.append(("response", response.correlation_id, response.completed))
        assert self.native.is_pending_request(response.correlation_id) == (not response.completed)
    def complete(self, request_id):
        assert not self.native.is_pending_request(request_id)
        assert self.native.pending_requests() == set()
        assert not self.native.has_pending_requests()
        self.events.append(("callback", request_id, True))
        self.next_request = self.native.request_instrument(instrument_id)
    def submit(self, method):
        if method == "request_data":
            args = (data_type, client_id)
        elif method == "request_instruments":
            args = ()
        elif method == "request_bars":
            args = (bar_type,)
        else:
            args = (instrument_id,)
        return getattr(self.native, method)(*args, callback=self.complete)
"#
                ),
                c_str!("completion_actor.py"),
                c_str!("completion_actor"),
            )
            .unwrap();
            module
                .setattr("instrument_id", sample_quote().instrument_id)
                .unwrap();
            module.setattr("bar_type", sample_bar().bar_type).unwrap();
            module
                .setattr("data_type", DataType::new("TestData", None, None))
                .unwrap();
            module
                .setattr("client_id", ClientId::new("HISTORY"))
                .unwrap();
            let tracker = module
                .getattr("CompletionActor")
                .unwrap()
                .call1((actor.clone_ref(py),))
                .unwrap();
            actor.borrow_mut(py).set_python_instance(&tracker).unwrap();
            actor.borrow(py).register_in_global_registries().unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let callback = tracker.getattr("complete").unwrap();
            tracker.setattr("complete", 42).unwrap();
            let error = tracker.call_method1("submit", (method,)).unwrap_err();
            assert!(error.is_instance_of::<pyo3::exceptions::PyTypeError>(py));
            assert!(saver.get_messages().is_empty());
            assert!(actor.borrow(py).inner().core.pending_requests().is_empty());
            tracker.setattr("complete", callback).unwrap();
            let root = tracker
                .call_method1("submit", (method,))
                .unwrap()
                .extract::<String>()
                .unwrap();
            let id: UUID4 = root.parse().unwrap();
            assert_eq!(
                actor
                    .bind(py)
                    .call_method0("pending_requests")
                    .unwrap()
                    .extract::<std::collections::HashSet<String>>()
                    .unwrap(),
                [root.clone()].into()
            );
            let terminal = crate::messages::data::DataResponseEnvelope::terminal(
                id,
                Some("empty result failure".to_owned()),
            );
            msgbus::send_response_envelope(&terminal);
            msgbus::send_response_envelope(&terminal);
            assert_eq!(
                tracker
                    .getattr("events")
                    .unwrap()
                    .extract::<Vec<(String, String, bool)>>()
                    .unwrap(),
                vec![
                    ("response".to_owned(), root.clone(), true),
                    ("callback".to_owned(), root, true)
                ]
            );
            let next = tracker
                .getattr("next_request")
                .unwrap()
                .extract::<String>()
                .unwrap();
            assert_eq!(
                actor
                    .bind(py)
                    .call_method0("pending_requests")
                    .unwrap()
                    .extract::<std::collections::HashSet<String>>()
                    .unwrap(),
                [next].into()
            );
            assert_eq!(saver.get_messages().len(), 2);
        });
    }

    #[rstest]
    #[case::success(false)]
    #[case::failure(true)]
    fn test_python_request_join_callback_order_and_late_join(
        #[case] failed: bool,
        #[values("none", "hook", "callback")] raises: &str,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let module = PyModule::from_code(
                py,
                c_str!(
                    r#"
class JoinActor:
    def __init__(self, native):
        self.native = native
        self.events = []
    def on_historical_quotes(self, quotes):
        self.events.append(("payload", len(quotes)))
    def on_data_response(self, response):
        self.events.append(("response", response.correlation_id))
        self.failure = response.failure_reason
        assert self.native.is_pending_request(response.correlation_id) == (not response.completed)
        if raises == "hook":
            raise RuntimeError("response hook failed")
    def complete(self, request_id):
        assert not self.native.is_pending_request(request_id)
        self.events.append(("callback", request_id))
        if raises == "callback":
            raise RuntimeError("completion callback failed")
    def late_complete(self, request_id):
        assert self.native.pending_requests() == set()
        self.events.append(("late", request_id))
        self.native.reset()
    def on_reset(self):
        pass
"#
                ),
                c_str!("join_actor.py"),
                c_str!("join_actor"),
            )
            .unwrap();
            module.setattr("raises", raises).unwrap();
            let tracker = module
                .getattr("JoinActor")
                .unwrap()
                .call1((actor.clone_ref(py),))
                .unwrap();
            actor.borrow_mut(py).set_python_instance(&tracker).unwrap();
            actor.borrow(py).register_in_global_registries().unwrap();
            let (handler, _) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let kwargs = PyDict::new(py);
            kwargs
                .set_item("callback", tracker.getattr("complete").unwrap())
                .unwrap();
            let quote = sample_quote();
            let root = actor
                .bind(py)
                .call_method("request_quotes", (quote.instrument_id,), Some(&kwargs))
                .unwrap()
                .extract::<String>()
                .unwrap();

            for (children, message) in [
                (
                    Vec::<String>::new(),
                    "Request group must not be empty".to_owned(),
                ),
                (
                    vec![root.clone(), root.clone()],
                    format!("Duplicate request ID in group: {root}"),
                ),
                (
                    vec![UUID4::new().to_string()],
                    "Unknown request ID:".to_owned(),
                ),
            ] {
                let error = actor
                    .bind(py)
                    .call_method("request_join", (children,), Some(&kwargs))
                    .unwrap_err();
                assert!(error.is_instance_of::<pyo3::exceptions::PyValueError>(py));
                assert!(error.to_string().contains(&message));
                assert_eq!(actor.borrow(py).inner().core.pending_requests().len(), 1);
            }

            let group = actor
                .bind(py)
                .call_method("request_join", (vec![root.clone()],), Some(&kwargs))
                .unwrap()
                .extract::<String>()
                .unwrap();
            let id: UUID4 = root.parse().unwrap();

            let payload = QuotesResponse::new(
                id,
                ClientId::new("HISTORY"),
                quote.instrument_id,
                vec![quote],
                None,
                None,
                UnixNanos::from(47),
                None,
            );

            let intermediate = crate::messages::data::DataResponseEnvelope::new(
                crate::messages::data::DataResponse::Quotes(payload),
                false,
                None,
            );
            msgbus::send_response_envelope(&intermediate);
            let failure = failed.then(|| "partial failure".to_owned());
            let terminal = crate::messages::data::DataResponseEnvelope::terminal(id, failure);
            msgbus::send_response_envelope(&terminal);
            msgbus::send_response_envelope(&terminal);
            let events = tracker
                .getattr("events")
                .unwrap()
                .repr()
                .unwrap()
                .to_string();
            assert_eq!(
                events,
                format!(
                    "[('payload', 1), ('response', '{root}'), ('response', '{root}'), ('callback', '{root}'), ('response', '{group}'), ('callback', '{group}')]"
                )
            );
            assert_eq!(
                tracker
                    .getattr("failure")
                    .unwrap()
                    .extract::<Option<String>>()
                    .unwrap(),
                failed.then(|| format!("Request {root}: partial failure"))
            );
            kwargs
                .set_item("callback", tracker.getattr("late_complete").unwrap())
                .unwrap();
            let late = UUID4::new().to_string();
            kwargs.set_item("request_id", &late).unwrap();
            let returned = actor
                .bind(py)
                .call_method("request_join", (vec![root, group],), Some(&kwargs))
                .unwrap()
                .extract::<String>()
                .unwrap();
            assert_eq!(returned, late);
            assert_eq!(
                tracker
                    .getattr("events")
                    .unwrap()
                    .get_item(-1)
                    .unwrap()
                    .extract::<(String, String)>()
                    .unwrap(),
                ("late".to_owned(), late)
            );
        });
    }

    #[rstest]
    #[case::reset("reset")]
    #[case::dispose("dispose")]
    #[case::fault("fault")]
    fn test_python_request_callbacks_release_on_cleanup(
        #[case] action: &str,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            let bus = py
                .get_type::<crate::python::msgbus::PyMessageBus>()
                .call1((trader_id,))
                .unwrap();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            actor
                .borrow_mut(py)
                .set_python_instance(actor.bind(py).as_any())
                .unwrap();
            actor.borrow(py).register_in_global_registries().unwrap();
            let (handler, _) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let module = PyModule::from_code(
                py,
                c_str!(
                    r#"
released = []
called = []
class Callback:
    def __init__(self, name):
        self.name = name
    def __call__(self, request_id):
        called.append(request_id)
    def __del__(self):
        bus.endpoints()
        released.append(self.name)
"#
                ),
                c_str!("cleanup_callbacks.py"),
                c_str!("cleanup_callbacks"),
            )
            .unwrap();
            module.setattr("bus", bus).unwrap();
            let callback = module
                .getattr("Callback")
                .unwrap()
                .call1(("request",))
                .unwrap();
            let kwargs = PyDict::new(py);
            kwargs.set_item("callback", &callback).unwrap();
            let root = actor
                .bind(py)
                .call_method(
                    "request_quotes",
                    (sample_quote().instrument_id,),
                    Some(&kwargs),
                )
                .unwrap()
                .extract::<String>()
                .unwrap();
            drop(callback);
            let callback = module
                .getattr("Callback")
                .unwrap()
                .call1(("group",))
                .unwrap();
            kwargs.set_item("callback", &callback).unwrap();
            let group = actor
                .bind(py)
                .call_method("request_join", (vec![root.clone()],), Some(&kwargs))
                .unwrap()
                .extract::<String>()
                .unwrap();
            drop(callback);
            drop(kwargs);

            if action == "fault" {
                actor.bind(py).call_method0("start").unwrap();
            }

            actor.bind(py).call_method0(action).unwrap();
            let root: UUID4 = root.parse().unwrap();
            msgbus::send_response_envelope(&crate::messages::data::DataResponseEnvelope::terminal(
                root, None,
            ));
            let mut released = module
                .getattr("released")
                .unwrap()
                .extract::<Vec<String>>()
                .unwrap();
            released.sort();
            assert_eq!(released, vec!["group", "request"]);
            assert_eq!(
                module
                    .getattr("called")
                    .unwrap()
                    .extract::<Vec<String>>()
                    .unwrap(),
                Vec::<String>::new()
            );
            assert!(
                !actor
                    .bind(py)
                    .call_method1("is_pending_request", (group,))
                    .unwrap()
                    .extract::<bool>()
                    .unwrap()
            );
            assert!(actor.borrow(py).inner().core.pending_requests().is_empty());
        });
    }

    #[rstest]
    #[case::payload("payload")]
    #[case::root_hook("root")]
    #[case::group_hook("group")]
    fn test_python_request_callback_reset_during_hook(
        #[case] reset_at: &str,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let module = PyModule::from_code(
                py,
                c_str!(
                    r#"
class ResetActor:
    def __init__(self, native, reset_at):
        self.native = native
        self.reset_at = reset_at
        self.events = []
    def on_reset(self):
        pass
    def on_historical_quotes(self, quotes):
        self.events.append("payload")
        if self.reset_at == "payload":
            self.native.reset()
    def on_data_response(self, response):
        kind = "root" if response.correlation_id == self.root else "group"
        self.events.append(kind)
        if self.reset_at == kind:
            self.native.reset()
    def complete(self, request_id):
        assert not self.native.is_pending_request(request_id)
        self.events.append("callback:" + request_id)
"#
                ),
                c_str!("reset_callback_actor.py"),
                c_str!("reset_callback_actor"),
            )
            .unwrap();
            let tracker = module
                .getattr("ResetActor")
                .unwrap()
                .call1((actor.clone_ref(py), reset_at))
                .unwrap();
            actor.borrow_mut(py).set_python_instance(&tracker).unwrap();
            actor.borrow(py).register_in_global_registries().unwrap();
            let (handler, _) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let kwargs = PyDict::new(py);
            kwargs
                .set_item("callback", tracker.getattr("complete").unwrap())
                .unwrap();
            let quote = sample_quote();
            let root = actor
                .bind(py)
                .call_method("request_quotes", (quote.instrument_id,), Some(&kwargs))
                .unwrap()
                .extract::<String>()
                .unwrap();
            tracker.setattr("root", &root).unwrap();
            actor
                .bind(py)
                .call_method("request_join", (vec![root.clone()],), Some(&kwargs))
                .unwrap();
            let id: UUID4 = root.parse().unwrap();

            let response = QuotesResponse::new(
                id,
                ClientId::new("HISTORY"),
                quote.instrument_id,
                vec![quote],
                None,
                None,
                UnixNanos::from(47),
                None,
            );
            msgbus::send_response(&id, &crate::messages::data::DataResponse::Quotes(response));

            let expected = match reset_at {
                "payload" => vec!["payload".to_owned()],
                "root" => vec!["payload".to_owned(), "root".to_owned()],
                "group" => vec![
                    "payload".to_owned(),
                    "root".to_owned(),
                    format!("callback:{root}"),
                    "group".to_owned(),
                ],
                _ => unreachable!(),
            };

            assert_eq!(
                tracker
                    .getattr("events")
                    .unwrap()
                    .extract::<Vec<String>>()
                    .unwrap(),
                expected
            );
            assert!(actor.borrow(py).inner().core.pending_requests().is_empty());
        });
    }

    #[rstest]
    fn test_python_response_hook_can_submit_next_request(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let module = PyModule::from_code(
                py,
                c_str!(
                    r#"
class ResponseActor:
    def __init__(self, native):
        self.native = native
        self.events = []
        self.next_request = None
    def on_historical_quotes(self, quotes):
        self.events.append("quotes")
        self.native.request_instrument(quotes[0].instrument_id)
    def on_data_response(self, response):
        self.events.append("complete")
        self.next_request = self.native.request_trades(response.data[0].instrument_id)
"#
                ),
                c_str!("response_actor.py"),
                c_str!("response_actor"),
            )
            .unwrap();
            let tracker = module
                .getattr("ResponseActor")
                .unwrap()
                .call1((actor.clone_ref(py),))
                .unwrap();
            actor.borrow_mut(py).set_python_instance(&tracker).unwrap();
            actor.borrow(py).register_in_global_registries().unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let quote = sample_quote();
            let root = actor
                .bind(py)
                .call_method1("request_quotes", (quote.instrument_id,))
                .unwrap()
                .extract::<String>()
                .unwrap();
            let root: UUID4 = root.parse().unwrap();

            let response = QuotesResponse::new(
                root,
                ClientId::new("HISTORY"),
                quote.instrument_id,
                vec![quote],
                None,
                None,
                UnixNanos::from(47),
                None,
            );
            msgbus::send_response(
                &root,
                &crate::messages::data::DataResponse::Quotes(response),
            );
            let commands = saver.get_messages();
            assert_eq!(commands.len(), 3);
            assert!(
                matches!(&commands[1], DataCommand::Request(crate::messages::data::RequestCommand::Instrument(r)) if r.instrument_ids == vec![quote.instrument_id])
            );

            let DataCommand::Request(crate::messages::data::RequestCommand::Trades(next)) =
                &commands[2]
            else {
                panic!("expected chained trade request");
            };

            assert_eq!(
                tracker
                    .getattr("events")
                    .unwrap()
                    .extract::<Vec<String>>()
                    .unwrap(),
                vec!["quotes", "complete"]
            );
            assert_eq!(
                tracker
                    .getattr("next_request")
                    .unwrap()
                    .extract::<String>()
                    .unwrap(),
                next.request_id.to_string()
            );
            assert!(!actor.borrow(py).inner().core.is_pending_request(&root));
            assert!(
                actor
                    .borrow(py)
                    .inner()
                    .core
                    .is_pending_request(&next.request_id)
            );
        });
    }

    #[rstest]
    #[case::complete("none")]
    #[case::payload_reset("payload")]
    #[case::child_reset("child")]
    #[case::group_reset("group")]
    #[case::child_error("error")]
    fn test_python_request_group_delivery_releases_actor_access(
        #[case] reset_at: &str,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            *get_message_bus().borrow_mut() = MessageBus::default();
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let module = PyModule::from_code(
                py,
                c_str!(
                    r#"
class ResponseActor:
    def __init__(self, native, reset_at):
        self.native = native
        self.reset_at = reset_at
        self.events = []
        self.next_request = None
    def on_reset(self):
        pass
    def on_historical_quotes(self, quotes):
        self.events.append("quotes")
        self.instrument_id = quotes[0].instrument_id

        if self.reset_at == "payload":
            self.native.reset()
    def on_data_response(self, response):
        assert response.completed
        assert response.failure_reason is None
        self.events.append(response.correlation_id)
        if response.correlation_id == self.root:

            if self.reset_at == "child":
                self.native.reset()
            elif self.reset_at == "error":
                raise RuntimeError("response hook failed")
        else:
            assert response.data is None

            if self.reset_at == "group":
                self.native.reset()
            else:
                self.next_request = self.native.request_trades(self.instrument_id)
"#
                ),
                c_str!("group_actor.py"),
                c_str!("group_actor"),
            )
            .unwrap();
            let tracker = module
                .getattr("ResponseActor")
                .unwrap()
                .call1((actor.clone_ref(py), reset_at))
                .unwrap();
            actor.borrow_mut(py).set_python_instance(&tracker).unwrap();
            actor.borrow(py).register_in_global_registries().unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );
            let quote = sample_quote();
            let root = actor
                .bind(py)
                .call_method1("request_quotes", (quote.instrument_id,))
                .unwrap()
                .extract::<String>()
                .unwrap();
            tracker.setattr("root", &root).unwrap();
            let root: UUID4 = root.parse().unwrap();
            let first = actor
                .borrow(py)
                .inner()
                .core
                .register_request_group(vec![root], None, None)
                .unwrap();
            let second = actor
                .borrow(py)
                .inner()
                .core
                .register_request_group(vec![root], None, None)
                .unwrap();

            let payload = QuotesResponse::new(
                root,
                ClientId::new("SOURCE"),
                quote.instrument_id,
                vec![quote],
                None,
                None,
                UnixNanos::from(47),
                None,
            );
            msgbus::send_response(&root, &crate::messages::data::DataResponse::Quotes(payload));
            let mut expected = vec!["quotes".to_owned()];
            if reset_at != "payload" {
                expected.push(root.to_string());
            }

            if !matches!(reset_at, "payload" | "child") {
                expected.push(first.to_string());
            }

            if matches!(reset_at, "none" | "error") {
                expected.push(second.to_string());
            }

            assert_eq!(
                tracker
                    .getattr("events")
                    .unwrap()
                    .extract::<Vec<String>>()
                    .unwrap(),
                expected
            );
            assert!(!actor.borrow(py).inner().core.is_pending_request(&root));
            assert!(!actor.borrow(py).inner().core.is_pending_request(&first));
            assert!(!actor.borrow(py).inner().core.is_pending_request(&second));
            let commands = saver.get_messages();

            if matches!(reset_at, "none" | "error") {
                let [
                    DataCommand::Request(RequestCommand::Quotes(_)),
                    DataCommand::Request(RequestCommand::Trades(first)),
                    DataCommand::Request(RequestCommand::Trades(second)),
                ] = commands.as_slice()
                else {
                    panic!("expected quotes and two chained trades requests")
                };

                assert_eq!(first.instrument_ids, vec![quote.instrument_id]);
                assert_eq!(second.instrument_ids, vec![quote.instrument_id]);
                assert_eq!(
                    tracker
                        .getattr("next_request")
                        .unwrap()
                        .extract::<String>()
                        .unwrap(),
                    second.request_id.to_string()
                );
            } else {
                assert_eq!(commands.len(), 1);
                assert!(tracker.getattr("next_request").unwrap().is_none());
            }
        });
    }

    const TRACKING_ACTOR_CODE: &std::ffi::CStr = c_str!(
        r#"
class TrackingActor:
    """A mock Python actor that tracks all method calls."""

    TRACKED_METHODS = {
        "on_start",
        "on_stop",
        "on_resume",
        "on_reset",
        "on_dispose",
        "on_degrade",
        "on_fault",
        "on_save",
        "on_load",
        "on_time_event",
        "on_data",
        "on_signal",
        "on_queue_state",
        "on_socket_state",
        "on_instrument",
        "on_quote",
        "on_trade",
        "on_bar",
        "on_book",
        "on_book_deltas",
        "on_book_depth",
        "on_mark_price",
        "on_index_price",
        "on_funding_rate",
        "on_instrument_status",
        "on_instrument_close",
        "on_option_greeks",
        "on_option_chain",
        "on_historical_data",
        "on_historical_book_deltas",
        "on_historical_book_depth",
        "on_historical_quotes",
        "on_historical_trades",
        "on_historical_funding_rates",
        "on_historical_bars",
        "on_historical_mark_prices",
        "on_historical_index_prices",
        "on_block",
        "on_pool",
        "on_pool_swap",
        "on_pool_liquidity_update",
        "on_pool_fee_collect",
        "on_pool_flash",
    }

    def __init__(self):
        self.calls = []
        self.raises = False

    def _record(self, method_name, *args):
        self.calls.append((method_name, args))
        if self.raises:
            raise RuntimeError("actor callback failure")

    def was_called(self, method_name):
        return any(call[0] == method_name for call in self.calls)

    def call_count(self, method_name):
        return sum(1 for call in self.calls if call[0] == method_name)

    def last_call_args(self, method_name):
        for called_method, args in reversed(self.calls):
            if called_method == method_name:
                return args
        raise AssertionError(f"{method_name} was not called")

    def last_loaded_state(self):

        for method_name, args in reversed(self.calls):
            if method_name == "on_load":
                return args[0]
        return None

    def on_save(self):
        self._record("on_save")
        return {"actor": b"saved"}

    def on_load(self, state):
        self._record("on_load", dict(state))

    def __getattr__(self, name):
        if name in self.TRACKED_METHODS:
            return lambda *args: self._record(name, *args)
        raise AttributeError(name)
"#
    );

    fn create_tracking_python_actor(py: Python<'_>) -> PyResult<Py<PyAny>> {
        py.run(TRACKING_ACTOR_CODE, None, None)?;
        let tracking_actor_class = py.eval(c_str!("TrackingActor"), None, None)?;
        let instance = tracking_actor_class.call0()?;
        Ok(instance.unbind())
    }

    fn python_method_was_called(py_actor: &Py<PyAny>, py: Python<'_>, method_name: &str) -> bool {
        py_actor
            .call_method1(py, "was_called", (method_name,))
            .and_then(|r| r.extract::<bool>(py))
            .unwrap_or(false)
    }

    fn python_method_call_count(py_actor: &Py<PyAny>, py: Python<'_>, method_name: &str) -> i32 {
        py_actor
            .call_method1(py, "call_count", (method_name,))
            .and_then(|r| r.extract::<i32>(py))
            .unwrap_or(0)
    }

    fn python_last_loaded_state(
        py_actor: &Py<PyAny>,
        py: Python<'_>,
    ) -> Option<HashMap<String, Vec<u8>>> {
        py_actor
            .call_method0(py, "last_loaded_state")
            .and_then(|result| result.extract::<Option<HashMap<String, Vec<u8>>>>(py))
            .unwrap_or(None)
    }

    const TRACKING_INDICATOR_CODE: &std::ffi::CStr = c_str!(
        r#"
class TrackingIndicator:
    def __init__(self, events=None):
        self.initialized = False
        self.calls = []
        self.events = events

    def handle_quote_tick(self, quote):
        self.calls.append("quote")
        if self.events is not None:
            self.events.append("indicator:quote")

    def handle_trade_tick(self, trade):
        self.calls.append("trade")
        if self.events is not None:
            self.events.append("indicator:trade")

    def handle_bar(self, bar):
        self.calls.append("bar")
        if self.events is not None:
            self.events.append("indicator:bar")

    def call_count(self, name):
        return self.calls.count(name)

class RaisingIndicator:
    initialized = True

    def handle_quote_tick(self, quote):
        raise RuntimeError("indicator failed")

class IndicatorEventActor:
    def __init__(self, events):
        self.events = events

    def on_start(self):
        pass

    def on_quote(self, quote):
        self.events.append("actor:quote")

    def on_trade(self, trade):
        self.events.append("actor:trade")

    def on_bar(self, bar):
        self.events.append("actor:bar")
"#
    );

    fn create_tracking_python_indicator(py: Python<'_>) -> PyResult<Py<PyAny>> {
        py.run(TRACKING_INDICATOR_CODE, None, None)?;
        let indicator_class = py.eval(c_str!("TrackingIndicator"), None, None)?;
        Ok(indicator_class.call0()?.unbind())
    }

    fn create_event_tracking_python_indicator(
        py: Python<'_>,
        events: &Bound<'_, PyList>,
    ) -> PyResult<Py<PyAny>> {
        py.run(TRACKING_INDICATOR_CODE, None, None)?;
        let indicator_class = py.eval(c_str!("TrackingIndicator"), None, None)?;
        Ok(indicator_class.call1((events,))?.unbind())
    }

    fn create_raising_python_indicator(py: Python<'_>) -> PyResult<Py<PyAny>> {
        py.run(TRACKING_INDICATOR_CODE, None, None)?;
        let indicator_class = py.eval(c_str!("RaisingIndicator"), None, None)?;
        Ok(indicator_class.call0()?.unbind())
    }

    fn create_indicator_event_actor(
        py: Python<'_>,
        events: &Bound<'_, PyList>,
    ) -> PyResult<Py<PyAny>> {
        py.run(TRACKING_INDICATOR_CODE, None, None)?;
        let actor_class = py.eval(c_str!("IndicatorEventActor"), None, None)?;
        Ok(actor_class.call1((events,))?.unbind())
    }

    fn python_indicator_call_count(
        indicator: &Py<PyAny>,
        py: Python<'_>,
        method_name: &str,
    ) -> i32 {
        indicator
            .call_method1(py, "call_count", (method_name,))
            .and_then(|r| r.extract::<i32>(py))
            .unwrap_or(0)
    }

    fn assert_python_dispatch<F>(
        py: Python<'_>,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        method_name: &str,
        invoke: F,
    ) -> Py<PyAny>
    where
        F: FnOnce(&mut PyDataActor) -> anyhow::Result<()>,
    {
        let py_actor = create_tracking_python_actor(py).unwrap();

        let mut rust_actor = PyDataActor::new(None);
        rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
        rust_actor.register(trader_id, clock, cache).unwrap();

        let result = invoke(&mut rust_actor);

        assert!(result.is_ok());
        assert!(python_method_was_called(&py_actor, py, method_name));
        assert_eq!(python_method_call_count(&py_actor, py, method_name), 1);

        py_actor
    }

    #[rstest]
    fn test_python_actor_callback_exception_preserves_traceback() {
        Python::initialize();
        Python::attach(|py| {
            let tracker = create_tracking_python_actor(py).unwrap();
            tracker.setattr(py, "raises", true).unwrap();
            let mut actor = PyDataActor::new(None);
            actor.set_python_instance(tracker.bind(py)).unwrap();
            let error = DataActor::on_bar(actor.inner_mut(), &sample_bar()).unwrap_err();
            let message = error.to_string();

            assert_eq!(python_method_call_count(&tracker, py, "on_bar"), 1);
            assert!(message.contains("Python on_bar failed:"));
            assert!(message.contains("in _record"));
            assert!(message.contains("RuntimeError: actor callback failure"));
        });
    }

    #[rstest]
    #[case("on_start")]
    #[case("on_stop")]
    #[case("on_resume")]
    #[case("on_reset")]
    #[case("on_dispose")]
    #[case("on_degrade")]
    #[case("on_fault")]
    fn test_python_dispatch_lifecycle_matrix(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] method_name: &str,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            assert_python_dispatch(py, clock, cache, trader_id, method_name, |rust_actor| {
                match method_name {
                    "on_start" => DataActor::on_start(rust_actor.inner_mut()),
                    "on_stop" => DataActor::on_stop(rust_actor.inner_mut()),
                    "on_resume" => DataActor::on_resume(rust_actor.inner_mut()),
                    "on_reset" => DataActor::on_reset(rust_actor.inner_mut()),
                    "on_dispose" => DataActor::on_dispose(rust_actor.inner_mut()),
                    "on_degrade" => DataActor::on_degrade(rust_actor.inner_mut()),
                    "on_fault" => DataActor::on_fault(rust_actor.inner_mut()),
                    _ => unreachable!("unhandled lifecycle case: {method_name}"),
                }
            });
        });
    }

    #[rstest]
    #[case("on_save")]
    #[case("on_load")]
    fn test_python_dispatch_persistence_matrix(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] method_name: &str,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            assert_python_dispatch(py, clock, cache, trader_id, method_name, |rust_actor| {
                match method_name {
                    "on_save" => {
                        let state = DataActor::on_save(rust_actor.inner()).unwrap();
                        assert_eq!(
                            state.get("actor").map(Vec::as_slice),
                            Some(b"saved".as_slice())
                        );
                        Ok(())
                    }
                    "on_load" => {
                        let mut state = IndexMap::new();
                        state.insert("actor".to_string(), b"loaded".to_vec());
                        DataActor::on_load(rust_actor.inner_mut(), state)
                    }
                    _ => unreachable!("unhandled persistence case: {method_name}"),
                }
            });
        });
    }

    #[rstest]
    fn test_python_persistence_methods_convert_state(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();

            let saved = rust_actor.py_save(py).unwrap();
            let saved_state = saved
                .bind(py)
                .extract::<HashMap<String, Vec<u8>>>()
                .unwrap();
            assert_eq!(
                saved_state.get("actor").map(Vec::as_slice),
                Some(&b"saved"[..])
            );

            let load_state = PyDict::new(py);
            load_state
                .set_item("actor", PyBytes::new(py, b"loaded-from-python"))
                .unwrap();

            rust_actor.py_load(&load_state).unwrap();

            let loaded_state = python_last_loaded_state(&py_actor, py).unwrap();
            assert_eq!(
                loaded_state.get("actor").map(Vec::as_slice),
                Some(&b"loaded-from-python"[..])
            );
        });
    }

    #[rstest]
    fn test_indicator_registration_exposes_readiness_and_registered_view(
        audusd_sim: CurrencyPair,
        bar_type: BarType,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let mut rust_actor = PyDataActor::new(None);
            let indicator = create_tracking_python_indicator(py).unwrap();

            assert_eq!(
                rust_actor
                    .py_registered_indicators(py)
                    .unwrap()
                    .bind(py)
                    .len()
                    .unwrap(),
                0
            );
            assert!(!rust_actor.py_indicators_initialized(py).unwrap());

            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                audusd_sim.id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_trade_ticks(
                py,
                audusd_sim.id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_bars(py, bar_type, indicator.clone_ref(py));

            let registered = rust_actor.py_registered_indicators(py).unwrap();
            let registered = registered.bind(py);

            assert_eq!(registered.len().unwrap(), 1);
            assert_eq!(
                registered.get_item(0).unwrap().as_ptr(),
                indicator.bind(py).as_ptr()
            );
            assert!(!rust_actor.py_indicators_initialized(py).unwrap());

            indicator.bind(py).setattr("initialized", true).unwrap();

            assert!(rust_actor.py_indicators_initialized(py).unwrap());
        });
    }

    #[rstest]
    fn test_registered_indicators_receive_quote_trade_and_bar_before_actor_callbacks(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let events = PyList::empty(py);
            let py_actor = create_indicator_event_actor(py, &events).unwrap();
            let indicator = create_event_tracking_python_indicator(py, &events).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            Component::start(rust_actor.inner_mut()).unwrap();

            let quote = sample_quote();
            let trade = sample_trade();
            let bar = sample_bar();
            let external_bar_type = BarType::from_str(&format!(
                "{}-1-MINUTE-LAST-EXTERNAL",
                bar.bar_type.instrument_id()
            ))
            .unwrap();

            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                quote.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_trade_ticks(
                py,
                trade.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_bars(
                py,
                external_bar_type,
                indicator.clone_ref(py),
            );

            DataActor::handle_quote(rust_actor.inner_mut(), &quote);
            DataActor::handle_trade(rust_actor.inner_mut(), &trade);
            DataActor::handle_bar(rust_actor.inner_mut(), &bar);

            let events = events.extract::<Vec<String>>().unwrap();

            assert_eq!(python_indicator_call_count(&indicator, py, "quote"), 1);
            assert_eq!(python_indicator_call_count(&indicator, py, "trade"), 1);
            assert_eq!(python_indicator_call_count(&indicator, py, "bar"), 1);
            assert_eq!(
                events,
                vec![
                    "indicator:quote",
                    "actor:quote",
                    "indicator:trade",
                    "actor:trade",
                    "indicator:bar",
                    "actor:bar",
                ]
            );
        });
    }

    #[rstest]
    fn test_registered_indicators_receive_live_data_when_actor_not_running(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let events = PyList::empty(py);
            let py_actor = create_indicator_event_actor(py, &events).unwrap();
            let indicator = create_event_tracking_python_indicator(py, &events).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();

            let quote = sample_quote();

            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                quote.instrument_id,
                indicator.clone_ref(py),
            );

            DataActor::handle_quote(rust_actor.inner_mut(), &quote);

            let events = events.extract::<Vec<String>>().unwrap();

            assert_eq!(python_indicator_call_count(&indicator, py, "quote"), 1);
            assert_eq!(events, vec!["indicator:quote"]);
        });
    }

    #[rstest]
    fn test_registered_indicators_receive_historical_quote_trade_and_bar_batches() {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let mut rust_actor = PyDataActor::new(None);
            let indicator = create_tracking_python_indicator(py).unwrap();
            let quote = sample_quote();
            let trade = sample_trade();
            let bar = sample_bar();
            let quotes = vec![quote];
            let trades = vec![trade];
            let bars = vec![bar];

            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                quote.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_trade_ticks(
                py,
                trade.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_bars(py, bar.bar_type, indicator.clone_ref(py));

            let client_id = ClientId::new("TEST");
            let quotes_response = QuotesResponse::new(
                UUID4::new(),
                client_id,
                quote.instrument_id,
                quotes,
                None,
                None,
                UnixNanos::default(),
                None,
            );
            let trades_response = TradesResponse::new(
                UUID4::new(),
                client_id,
                trade.instrument_id,
                trades,
                None,
                None,
                UnixNanos::default(),
                None,
            );
            let bars_response = BarsResponse::new(
                UUID4::new(),
                client_id,
                bar.bar_type,
                bars,
                None,
                None,
                UnixNanos::default(),
                None,
            );

            DataActor::handle_quotes_response(rust_actor.inner_mut(), &quotes_response);
            DataActor::handle_trades_response(rust_actor.inner_mut(), &trades_response);
            DataActor::handle_bars_response(rust_actor.inner_mut(), &bars_response);

            assert_eq!(python_indicator_call_count(&indicator, py, "quote"), 1);
            assert_eq!(python_indicator_call_count(&indicator, py, "trade"), 1);
            assert_eq!(python_indicator_call_count(&indicator, py, "bar"), 1);
        });
    }

    #[rstest]
    fn test_indicators_initialized_requires_all_registered_indicators(audusd_sim: CurrencyPair) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let mut rust_actor = PyDataActor::new(None);
            let first = create_tracking_python_indicator(py).unwrap();
            let second = create_tracking_python_indicator(py).unwrap();

            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                audusd_sim.id,
                first.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                audusd_sim.id,
                second.clone_ref(py),
            );

            first.bind(py).setattr("initialized", true).unwrap();

            assert!(!rust_actor.py_indicators_initialized(py).unwrap());

            second.bind(py).setattr("initialized", true).unwrap();

            assert!(rust_actor.py_indicators_initialized(py).unwrap());
        });
    }

    #[rstest]
    fn test_duplicate_indicator_registration_does_not_duplicate_callbacks(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let mut rust_actor = PyDataActor::new(None);
            let indicator = create_tracking_python_indicator(py).unwrap();
            let quote = sample_quote();
            let trade = sample_trade();
            let bar = sample_bar();
            rust_actor.register(trader_id, clock, cache).unwrap();
            Component::start(rust_actor.inner_mut()).unwrap();

            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                quote.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_quote_ticks(
                py,
                quote.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_trade_ticks(
                py,
                trade.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_trade_ticks(
                py,
                trade.instrument_id,
                indicator.clone_ref(py),
            );
            rust_actor.py_register_indicator_for_bars(py, bar.bar_type, indicator.clone_ref(py));
            rust_actor.py_register_indicator_for_bars(py, bar.bar_type, indicator.clone_ref(py));

            DataActor::handle_quote(rust_actor.inner_mut(), &quote);
            DataActor::handle_trade(rust_actor.inner_mut(), &trade);
            DataActor::handle_bar(rust_actor.inner_mut(), &bar);

            assert_eq!(
                rust_actor
                    .py_registered_indicators(py)
                    .unwrap()
                    .bind(py)
                    .len()
                    .unwrap(),
                1
            );
            assert_eq!(python_indicator_call_count(&indicator, py, "quote"), 1);
            assert_eq!(python_indicator_call_count(&indicator, py, "trade"), 1);
            assert_eq!(python_indicator_call_count(&indicator, py, "bar"), 1);
        });
    }

    #[rstest]
    fn test_indicator_error_prevents_actor_callback(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let events = PyList::empty(py);
            let py_actor = create_indicator_event_actor(py, &events).unwrap();
            let indicator = create_raising_python_indicator(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();
            Component::start(rust_actor.inner_mut()).unwrap();

            let quote = sample_quote();

            rust_actor.py_register_indicator_for_quote_ticks(py, quote.instrument_id, indicator);

            DataActor::handle_quote(rust_actor.inner_mut(), &quote);
            let events = events.extract::<Vec<String>>().unwrap();

            assert!(events.is_empty());
        });
    }

    #[rstest]
    #[case("on_time_event")]
    #[case("on_data")]
    #[case("on_signal")]
    #[case("on_queue_state")]
    #[case("on_socket_state")]
    #[case("on_instrument")]
    #[case("on_quote")]
    #[case("on_trade")]
    #[case("on_bar")]
    #[case("on_book")]
    #[case("on_book_deltas")]
    #[case("on_book_depth")]
    #[case("on_mark_price")]
    #[case("on_index_price")]
    #[case("on_funding_rate")]
    #[case("on_instrument_status")]
    #[case("on_instrument_close")]
    #[case("on_option_greeks")]
    #[case("on_option_chain")]
    fn test_python_dispatch_typed_callback_matrix(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] method_name: &str,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            assert_python_dispatch(py, clock, cache, trader_id, method_name, |rust_actor| {
                match method_name {
                    "on_time_event" => {
                        let event = sample_time_event();
                        rust_actor.inner_mut().on_time_event(&event)
                    }
                    "on_data" => {
                        let data = sample_data();
                        rust_actor.inner_mut().on_data(&data)
                    }
                    "on_signal" => {
                        let signal = sample_signal();
                        rust_actor.inner_mut().on_signal(&signal)
                    }
                    "on_queue_state" => {
                        let event = sample_queue_state_changed(QueueState::Triggered);
                        rust_actor.inner_mut().on_queue_state(&event)
                    }
                    "on_socket_state" => {
                        let event = sample_socket_state_changed(SocketState::Connected);
                        rust_actor.inner_mut().on_socket_state(&event)
                    }
                    "on_instrument" => {
                        let instrument = InstrumentAny::CurrencyPair(sample_instrument());
                        rust_actor.inner_mut().on_instrument(&instrument)
                    }
                    "on_quote" => {
                        let quote = sample_quote();
                        rust_actor.inner_mut().on_quote(&quote)
                    }
                    "on_trade" => {
                        let trade = sample_trade();
                        rust_actor.inner_mut().on_trade(&trade)
                    }
                    "on_bar" => {
                        let bar = sample_bar();
                        rust_actor.inner_mut().on_bar(&bar)
                    }
                    "on_book" => {
                        let book = sample_book();
                        rust_actor.inner_mut().on_book(&book)
                    }
                    "on_book_deltas" => {
                        let deltas = sample_book_deltas();
                        rust_actor.inner_mut().on_book_deltas(&deltas)
                    }
                    "on_book_depth" => {
                        let depth = sample_book_depth();
                        rust_actor.inner_mut().on_book_depth(&depth)
                    }
                    "on_mark_price" => {
                        let update = sample_mark_price();
                        rust_actor.inner_mut().on_mark_price(&update)
                    }
                    "on_index_price" => {
                        let update = sample_index_price();
                        rust_actor.inner_mut().on_index_price(&update)
                    }
                    "on_funding_rate" => {
                        let update = sample_funding_rate();
                        rust_actor.inner_mut().on_funding_rate(&update)
                    }
                    "on_instrument_status" => {
                        let status = sample_instrument_status();
                        rust_actor.inner_mut().on_instrument_status(&status)
                    }
                    "on_instrument_close" => {
                        let close = sample_instrument_close();
                        rust_actor.inner_mut().on_instrument_close(&close)
                    }
                    "on_option_greeks" => {
                        let greeks = sample_option_greeks();
                        rust_actor.inner_mut().on_option_greeks(&greeks)
                    }
                    "on_option_chain" => {
                        let chain = sample_option_chain();
                        rust_actor.inner_mut().on_option_chain(&chain)
                    }
                    _ => unreachable!("unhandled typed callback case: {method_name}"),
                }
            });
        });
    }

    #[rstest]
    #[case("on_historical_data")]
    #[case("on_historical_book_deltas")]
    #[case("on_historical_book_depth")]
    #[case("on_historical_quotes")]
    #[case("on_historical_trades")]
    #[case("on_historical_funding_rates")]
    #[case("on_historical_bars")]
    #[case("on_historical_mark_prices")]
    #[case("on_historical_index_prices")]
    fn test_python_dispatch_historical_callback_matrix(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] method_name: &str,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            assert_python_dispatch(py, clock, cache, trader_id, method_name, |rust_actor| {
                match method_name {
                    "on_historical_data" => {
                        let data = sample_data();
                        rust_actor.inner_mut().on_historical_data(&data)
                    }
                    "on_historical_book_deltas" => {
                        let deltas = sample_book_deltas().deltas;
                        rust_actor.inner_mut().on_historical_book_deltas(&deltas)
                    }
                    "on_historical_book_depth" => {
                        let depths = vec![sample_book_depth()];
                        rust_actor.inner_mut().on_historical_book_depth(&depths)
                    }
                    "on_historical_quotes" => {
                        let quotes = vec![sample_quote()];
                        rust_actor.inner_mut().on_historical_quotes(&quotes)
                    }
                    "on_historical_trades" => {
                        let trades = vec![sample_trade()];
                        rust_actor.inner_mut().on_historical_trades(&trades)
                    }
                    "on_historical_funding_rates" => {
                        let funding_rates = vec![sample_funding_rate()];
                        rust_actor
                            .inner_mut()
                            .on_historical_funding_rates(&funding_rates)
                    }
                    "on_historical_bars" => {
                        let bars = vec![sample_bar()];
                        rust_actor.inner_mut().on_historical_bars(&bars)
                    }
                    "on_historical_mark_prices" => {
                        let mark_prices = vec![sample_mark_price()];
                        rust_actor
                            .inner_mut()
                            .on_historical_mark_prices(&mark_prices)
                    }
                    "on_historical_index_prices" => {
                        let index_prices = vec![sample_index_price()];
                        rust_actor
                            .inner_mut()
                            .on_historical_index_prices(&index_prices)
                    }
                    _ => unreachable!("unhandled historical callback case: {method_name}"),
                }
            });
        });
    }

    #[rstest]
    fn test_python_dispatch_historical_book_deltas_preserves_batch(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let expected = stub_deltas().deltas;
            let py_actor = assert_python_dispatch(
                py,
                clock,
                cache,
                trader_id,
                "on_historical_book_deltas",
                |rust_actor| rust_actor.inner_mut().on_historical_book_deltas(&expected),
            );
            let actual = py_actor
                .call_method1(py, "last_call_args", ("on_historical_book_deltas",))
                .unwrap()
                .bind(py)
                .get_item(0)
                .unwrap()
                .extract::<Vec<OrderBookDelta>>()
                .unwrap();

            assert_eq!(actual, expected);
        });
    }

    #[rstest]
    fn test_python_dispatch_historical_book_depth_preserves_batch(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let first = stub_depth10();
            let mut second = first.clone();
            second.sequence = 17;
            second.ts_event = UnixNanos::from(18);
            second.ts_init = UnixNanos::from(19);
            let expected = vec![first, second];
            let py_actor = assert_python_dispatch(
                py,
                clock,
                cache,
                trader_id,
                "on_historical_book_depth",
                |rust_actor| rust_actor.inner_mut().on_historical_book_depth(&expected),
            );
            let actual = py_actor
                .call_method1(py, "last_call_args", ("on_historical_book_depth",))
                .unwrap()
                .bind(py)
                .get_item(0)
                .unwrap()
                .extract::<Vec<OrderBookDepth>>()
                .unwrap();

            assert_eq!(actual, expected);
        });
    }

    #[cfg(feature = "defi")]
    #[rstest]
    #[case("on_block")]
    #[case("on_pool")]
    #[case("on_pool_swap")]
    #[case("on_pool_liquidity_update")]
    #[case("on_pool_fee_collect")]
    #[case("on_pool_flash")]
    fn test_python_dispatch_defi_callback_matrix(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        #[case] method_name: &str,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            assert_python_dispatch(py, clock, cache, trader_id, method_name, |rust_actor| {
                match method_name {
                    "on_block" => {
                        let block = sample_block();
                        rust_actor.inner_mut().on_block(&block)
                    }
                    "on_pool" => {
                        let (_chain, _dex, pool) = sample_pool_components();
                        rust_actor.inner_mut().on_pool(&pool)
                    }
                    "on_pool_swap" => {
                        let swap = sample_pool_swap();
                        rust_actor.inner_mut().on_pool_swap(&swap)
                    }
                    "on_pool_liquidity_update" => {
                        let update = sample_pool_liquidity_update();
                        rust_actor.inner_mut().on_pool_liquidity_update(&update)
                    }
                    "on_pool_fee_collect" => {
                        let collect = sample_pool_fee_collect();
                        rust_actor.inner_mut().on_pool_fee_collect(&collect)
                    }
                    "on_pool_flash" => {
                        let flash = sample_pool_flash();
                        rust_actor.inner_mut().on_pool_flash(&flash)
                    }
                    _ => unreachable!("unhandled defi callback case: {method_name}"),
                }
            });
        });
    }

    #[rstest]
    fn test_python_dispatch_multiple_calls_tracked(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        audusd_sim: CurrencyPair,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();

            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();

            let quote = QuoteTick::new(
                audusd_sim.id,
                Price::from("1.00000"),
                Price::from("1.00001"),
                Quantity::from(100_000),
                Quantity::from(100_000),
                UnixNanos::default(),
                UnixNanos::default(),
            );

            rust_actor.inner_mut().on_quote(&quote).unwrap();
            rust_actor.inner_mut().on_quote(&quote).unwrap();
            rust_actor.inner_mut().on_quote(&quote).unwrap();

            assert_eq!(python_method_call_count(&py_actor, py, "on_quote"), 3);
        });
    }

    #[rstest]
    fn test_python_dispatch_historical_custom_data_preserves_payload_shape(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
        client_id: ClientId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|py| {
            let py_actor = create_tracking_python_actor(py).unwrap();
            let mut rust_actor = PyDataActor::new(None);
            rust_actor.set_python_instance(py_actor.bind(py)).unwrap();
            rust_actor.register(trader_id, clock, cache).unwrap();

            let data = vec![
                stub_custom_data(1, 42, None, None),
                stub_custom_data(2, 84, None, None),
            ];
            let scalar = stub_custom_data(3, 126, None, None);
            let scalar_response = CustomDataResponse::new(
                UUID4::new(),
                client_id,
                None,
                scalar.data_type.clone(),
                scalar.clone(),
                None,
                None,
                UnixNanos::default(),
                None,
            );

            DataActor::handle_data_response(rust_actor.inner_mut(), &scalar_response);

            let actual_scalar = py_actor
                .call_method1(py, "last_call_args", ("on_historical_data",))
                .unwrap()
                .bind(py)
                .get_item(0)
                .unwrap()
                .extract::<CustomData>()
                .unwrap();

            assert_eq!(
                python_method_call_count(&py_actor, py, "on_historical_data"),
                1
            );
            assert_eq!(actual_scalar, scalar);

            let empty_response = CustomDataResponse::new(
                UUID4::new(),
                client_id,
                None,
                data[0].data_type.clone(),
                Vec::<CustomData>::new(),
                None,
                None,
                UnixNanos::default(),
                None,
            );

            DataActor::handle_data_response(rust_actor.inner_mut(), &empty_response);

            let empty = py_actor
                .call_method1(py, "last_call_args", ("on_historical_data",))
                .unwrap()
                .bind(py)
                .get_item(0)
                .unwrap()
                .extract::<Vec<CustomData>>()
                .unwrap();

            assert_eq!(
                python_method_call_count(&py_actor, py, "on_historical_data"),
                2
            );
            assert!(empty.is_empty());

            let response = CustomDataResponse::new(
                UUID4::new(),
                client_id,
                None,
                data[0].data_type.clone(),
                data.clone(),
                None,
                None,
                UnixNanos::default(),
                None,
            );

            DataActor::handle_data_response(rust_actor.inner_mut(), &response);

            let actual = py_actor
                .call_method1(py, "last_call_args", ("on_historical_data",))
                .unwrap()
                .bind(py)
                .get_item(0)
                .unwrap()
                .extract::<Vec<CustomData>>()
                .unwrap();

            assert_eq!(
                python_method_call_count(&py_actor, py, "on_historical_data"),
                3
            );
            assert_eq!(actual, data);
        });
    }

    #[rstest]
    fn test_python_dispatch_no_call_when_py_self_not_set(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        Python::attach(|_py| {
            let mut rust_actor = PyDataActor::new(None);
            rust_actor.register(trader_id, clock, cache).unwrap();

            // When py_self is None, the dispatch returns Ok(()) without calling Python
            let result = DataActor::on_start(rust_actor.inner_mut());
            assert!(result.is_ok());
        });
    }

    #[rstest]
    fn test_python_on_historical_data_rejects_non_custom_data(
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        pyo3::Python::initialize();

        let mut rust_actor = PyDataActor::new(None);
        rust_actor.register(trader_id, clock, cache).unwrap();

        let non_custom: String = "not CustomData".to_string();
        let result = rust_actor.inner_mut().on_historical_data(&non_custom);

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("unsupported type"));
    }

    #[rstest]
    fn test_python_self_is_weak() {
        Python::initialize();

        Python::attach(|py| {
            let instance = py
                .get_type::<PyDataActor>()
                .call0()
                .expect("DataActor should construct");
            let weakref =
                PyWeakrefReference::new(&instance).expect("DataActor should be weak-referenceable");
            assert!(weakref.upgrade().is_some());

            drop(instance);

            // A strong `py_self` would form an untraceable Rust-Python cycle and keep this alive
            assert!(
                weakref.upgrade().is_none(),
                "an unregistered DataActor must be collected once its last Python owner is dropped",
            );
        });
    }

    #[rstest::rstest]
    #[case("on_start")]
    #[case("on_stop")]
    #[case("on_resume")]
    #[case("on_reset")]
    #[case("on_dispose")]
    #[case("on_degrade")]
    #[case("on_fault")]
    #[case("on_save")]
    #[case("on_load")]
    #[case("on_time_event")]
    #[case("on_data")]
    #[case("on_signal")]
    #[case("on_queue_state")]
    #[case("on_socket_state")]
    #[case("on_instrument")]
    #[case("on_quote")]
    #[case("on_trade")]
    #[case("on_bar")]
    #[case("on_book_deltas")]
    #[case("on_book_depth")]
    #[case("on_book")]
    #[case("on_mark_price")]
    #[case("on_index_price")]
    #[case("on_funding_rate")]
    #[case("on_instrument_status")]
    #[case("on_instrument_close")]
    #[case("on_option_greeks")]
    #[case("on_option_chain")]
    #[case("on_historical_data")]
    #[case("on_historical_book_deltas")]
    #[case("on_historical_book_depth")]
    #[case("on_historical_quotes")]
    #[case("on_historical_trades")]
    #[case("on_historical_funding_rates")]
    #[case("on_historical_bars")]
    #[case("on_historical_mark_prices")]
    #[case("on_historical_index_prices")]
    fn test_default_callback_during_exclusive_borrow(#[case] method: &str) {
        Python::initialize();
        Python::attach(|py| {
            let argument = match method {
                "on_start" | "on_stop" | "on_resume" | "on_reset" | "on_dispose" | "on_degrade"
                | "on_fault" | "on_save" => None,
                "on_load" => Some(PyDict::new(py).into_any().unbind()),
                "on_time_event" => Some(sample_time_event().into_py_any(py).unwrap()),
                "on_data" | "on_instrument" | "on_historical_data" => Some(py.None()),
                "on_signal" => Some(sample_signal().into_py_any(py).unwrap()),
                "on_queue_state" => Some(
                    sample_queue_state_changed(QueueState::Triggered)
                        .into_py_any(py)
                        .unwrap(),
                ),
                "on_socket_state" => Some(
                    sample_socket_state_changed(SocketState::Connected)
                        .into_py_any(py)
                        .unwrap(),
                ),
                "on_quote" => Some(sample_quote().into_py_any(py).unwrap()),
                "on_trade" => Some(sample_trade().into_py_any(py).unwrap()),
                "on_bar" => Some(sample_bar().into_py_any(py).unwrap()),
                "on_book_deltas" => Some(sample_book_deltas().into_py_any(py).unwrap()),
                "on_book_depth" => Some(sample_book_depth().into_py_any(py).unwrap()),
                "on_book" => Some(sample_book().into_py_any(py).unwrap()),
                "on_mark_price" => Some(sample_mark_price().into_py_any(py).unwrap()),
                "on_index_price" => Some(sample_index_price().into_py_any(py).unwrap()),
                "on_funding_rate" => Some(sample_funding_rate().into_py_any(py).unwrap()),
                "on_instrument_status" => Some(sample_instrument_status().into_py_any(py).unwrap()),
                "on_instrument_close" => Some(sample_instrument_close().into_py_any(py).unwrap()),
                "on_option_greeks" => Some(sample_option_greeks().into_py_any(py).unwrap()),
                "on_option_chain" => Some(sample_option_chain().into_py_any(py).unwrap()),
                "on_historical_book_deltas"
                | "on_historical_book_depth"
                | "on_historical_quotes"
                | "on_historical_trades"
                | "on_historical_funding_rates"
                | "on_historical_bars"
                | "on_historical_mark_prices"
                | "on_historical_index_prices" => Some(PyList::empty(py).into_any().unbind()),
                _ => unreachable!(),
            };

            let instance = py.get_type::<PyDataActor>().call0().unwrap();
            let _borrow = instance
                .extract::<pyo3::PyRefMut<'_, PyDataActor>>()
                .unwrap();

            let result = match argument {
                Some(argument) => instance.call_method1(method, (argument,)),
                None => instance.call_method0(method),
            }
            .unwrap();

            if method == "on_save" {
                assert_eq!(result.len().unwrap(), 0);
            } else {
                assert!(result.is_none());
            }
        });
    }

    #[cfg(feature = "defi")]
    #[rstest::rstest]
    #[case("on_block")]
    #[case("on_pool")]
    #[case("on_pool_swap")]
    #[case("on_pool_liquidity_update")]
    #[case("on_pool_fee_collect")]
    #[case("on_pool_flash")]
    fn test_default_defi_callback_during_exclusive_borrow(#[case] method: &str) {
        Python::initialize();
        Python::attach(|py| {
            let argument = match method {
                "on_block" => sample_block().into_py_any(py).unwrap(),
                "on_pool" => sample_pool_components().2.into_py_any(py).unwrap(),
                "on_pool_swap" => sample_pool_swap().into_py_any(py).unwrap(),
                "on_pool_liquidity_update" => {
                    sample_pool_liquidity_update().into_py_any(py).unwrap()
                }
                "on_pool_fee_collect" => sample_pool_fee_collect().into_py_any(py).unwrap(),
                "on_pool_flash" => sample_pool_flash().into_py_any(py).unwrap(),
                _ => unreachable!(),
            };

            let instance = py.get_type::<PyDataActor>().call0().unwrap();
            let _borrow = instance
                .extract::<pyo3::PyRefMut<'_, PyDataActor>>()
                .unwrap();
            assert!(
                instance
                    .call_method1(method, (argument,))
                    .unwrap()
                    .is_none()
            );
        });
    }

    #[rstest]
    fn test_python_instrument_lists_validate_before_dispatch(
        #[values(
            "instrument",
            "book_deltas",
            "book_depth",
            "book_at_interval",
            "mark_prices",
            "index_prices",
            "funding_rates",
            "instrument_status",
            "instrument_close",
            "option_greeks"
        )]
        family: &str,
        #[values(false, true)] subscribe: bool,
        #[values(false, true)] mixed_venues: bool,
        clock: Rc<RefCell<VirtualClock>>,
        cache: Rc<RefCell<Cache>>,
        trader_id: TraderId,
    ) {
        Python::initialize();
        Python::attach(|py| {
            let actor = Py::new(py, create_registered_actor(clock, cache, trader_id)).unwrap();
            let (handler, saver) = get_typed_into_message_saving_handler::<DataCommand>(None);
            msgbus::register_data_command_endpoint(
                MessagingSwitchboard::data_engine_queue_execute(),
                handler,
            );

            let ids = if mixed_venues {
                vec![
                    InstrumentId::from("BTC/USD.SIM"),
                    InstrumentId::from("ETH/USD.OTHER"),
                ]
            } else {
                vec![]
            };

            let argument = PyList::new(py, ids).unwrap();
            let kwargs = instrument_subscription_keywords(py, family, subscribe);
            let method = format!(
                "{}_{family}",
                if subscribe {
                    "subscribe"
                } else {
                    "unsubscribe"
                }
            );
            let error = actor
                .bind(py)
                .call_method(method.as_str(), (&argument,), Some(&kwargs))
                .unwrap_err();
            assert!(error.is_instance_of::<pyo3::exceptions::PyValueError>(py));
            assert_eq!(
                error.value(py).to_string(),
                if mixed_venues {
                    "multiple instrument venues require an explicit client ID"
                } else {
                    "instrument IDs cannot be empty"
                }
            );
            assert_eq!(saver.get_messages().len(), 0);
        });
    }

    fn instrument_subscription_keywords<'py>(
        py: Python<'py>,
        family: &str,
        subscribe: bool,
    ) -> Bound<'py, PyDict> {
        let kwargs = PyDict::new(py);
        if family.starts_with("book_") && subscribe {
            kwargs.set_item("book_type", BookType::L2_MBP).unwrap();
        }

        if family == "book_at_interval" {
            kwargs.set_item("interval_ms", 13).unwrap();
        }

        kwargs
    }
}