What causes "Python stream processor failed: {e}" in nautechsystems/nautilus_trader?

Fix the exception reported in `{e}` — the message contains the full Python traceback.

Python stream processor failed: {e}

Error message

Python stream processor failed: {e}

What it means

This error is raised when the Python callback registered for the message bus stream processor raises an exception. The Rust closure `add_stream_processor_with_mapping` invokes the callback via PyO3; any `PyErr` it returns is converted with `anyhow!` into this message containing the Python traceback text. The stream processor mapping itself failed, so the mapped payload is not delivered.

Source

Thrown at crates/live/src/python/node.rs:927

    /// MessagePack typed payloads. Each callback receives an encoding-independent Python mapping
    /// that follows the concrete payload's serialized object shape and adds a `payload_type` field.
    /// Other encodings are skipped with a warning. External egress is suppressed while the callback
    /// runs, so synchronous publications remain local. Callback exceptions are logged, stop the
    /// remaining processors, and skip internal republishing.
    ///
    /// # Errors
    ///
    /// Returns an error if the node is no longer available for mutation.
    #[pyo3(name = "add_stream_processor")]
    fn py_add_stream_processor(&self, callback: Py<PyAny>) -> PyResult<()> {
        self.node_mut()?
            .add_stream_processor_with_mapping(move |_, mapping| {
                Python::attach(|py| {
                    let payload = json_value_to_py(py, mapping)?;
                    callback.call1(py, (payload,))?;
                    Ok(())
                })
                .map_err(|e: PyErr| anyhow::anyhow!("Python stream processor failed: {e}"))
            });
        Ok(())
    }

    /// Runs the live node on the caller's asyncio event loop.
    ///
    /// Takes the node and returns an awaitable that resolves once the node has stopped. The host
    /// owns the loop and its signal handling, so this installs no signal handlers. Stop the node
    /// through the handle from `handle()`; cancelling the awaiting task requests the same graceful
    /// shutdown, waits for it to finish, then re-raises the cancellation.
    ///
    /// Capture `cache`, `portfolio`, and `handle()` before calling this. They stay usable while the
    /// node runs, whereas the node itself is owned by the returned awaitable.
    ///
    /// # Limitations
    ///
    /// A node configured with a cache database backing is rejected. Those backings wait for their
    /// worker task by blocking the calling thread, which stalls the host loop rather than merely

View on GitHub (pinned to 18893faf8b)

Solutions

Fix the exception reported in `{e}` — the message contains the full Python traceback.
Ensure the callback accepts exactly one argument (the mapped payload dict) and is callable.
Test the callback standalone with a representative payload before attaching it to the stream.
Wrap the callback body in try/except if failures should be logged rather than aborting the stream registration.

Example fix

# before
def handler(mapping):
    return process(mapping['data'])  # KeyError if 'data' missing
# after
def handler(mapping):
    try:
        return process(mapping.get('data'))
    except Exception as e:
        logging.exception('stream handler failed: %s', e)

Defensive patterns

Strategy: try-catch

Validate before calling

# Validate the callback before registering
assert callable(cb) and cb.__code__.co_argcount == 1, "callback must take one payload arg"
try:
    cb({"test": 1})
except Exception as e:
    raise RuntimeError(f"stream callback not safe: {e}")

Try / catch

try:
    node.add_stream_processor(callback)
except Exception as e:
    if "Python stream processor failed" in str(e):
        logging.exception("stream callback raised: %s", e)
        # fix or replace the callback
    else:
        raise

Prevention

Give the callback exactly one parameter (the mapped payload dict)
Unit-test the callback with representative payloads before attaching it
Wrap callback bodies in try/except with logging for non-fatal failures
Don't mutate shared state from the callback without locks

When it happens

Trigger: Registering a stream-processing callback via the live node's Python API where the callback raises — e.g. `callback(payload)` throws a TypeError (wrong signature/arity), the handler references missing state, or JSON conversion of the mapping produces an object the callback mishandles.

Common situations: User callbacks with the wrong signature (expecting raw bytes instead of a dict); handlers raising on the first messages after config changes; exceptions in long-running signal-processing callbacks attached to the message bus.

Related errors

AI-assisted analysis of nautechsystems/nautilus_trader@18893faf8b (2026-09-08). Data as JSON: /api/errors/2b68d838f0063afa. Report an issue: GitHub.