nexustrader.core.nautilius_core

This module provides the core infrastructure components used throughout NexusTrader: the logger, message bus, and clock. Since version 0.3.7 these are pure-Python implementations and no longer depend on nautilus-trader. Since version 0.3.21 the logging backend is loguru, a pure-Python logger with zero build requirements and built-in time-based rotation.

Logger

Logger is the structured logger used throughout NexusTrader. Access it from a strategy via self.log:

self.log.info("Order placed")
self.log.warning("Unusual spread detected")
self.log.error("Connection lost")

# Optional color hint (ignored if the terminal does not support ANSI):
from nexustrader.constants import LogColor
self.log.info("Golden Cross signal!", color=LogColor.GREEN)

Log levels (lowest → highest): TRACE, DEBUG, INFO, WARNING, ERROR.

Note

Since 0.3.21, TRACE is natively supported (loguru level 5). logger.trace() emits at true TRACE severity.

Configure the minimum level via LogConfig:

from nexustrader.config import Config, LogConfig

config = Config(
    ...,
    log_config=LogConfig(level_stdout="DEBUG"),
)

MessageBus

MessageBus is the in-process pub/sub and point-to-point routing bus. It connects internal components (connectors, EMS, OMS, strategies) without polling overhead.

Two communication patterns:

• Topic (fan-out) — subscribe(topic, handler) / publish(topic, msg): one publisher, many subscribers.

• Endpoint (point-to-point) — register(endpoint, handler) / send(endpoint, msg): exactly one registered handler per endpoint.

LiveClock

LiveClock exposes the current wall-clock time and drives repeating timers.

Time accessors

clock.timestamp()     # float  — seconds since epoch
clock.timestamp_ms()  # int    — milliseconds since epoch
clock.timestamp_ns()  # int    — nanoseconds since epoch
clock.utc_now()       # datetime (UTC)

Timers

from datetime import datetime, timedelta, timezone
from nexustrader.core.nautilius_core import TimeEvent

def on_tick(event: TimeEvent) -> None:
    print(event.ts_event, event.ts_init)  # nanoseconds

clock.set_timer(
    name="my_timer",
    interval=timedelta(seconds=1),
    start_time=datetime.now(tz=timezone.utc) + timedelta(seconds=1),
    stop_time=None,       # runs indefinitely
    callback=on_tick,
)

clock.cancel_timer("my_timer")

TimeEvent

Emitted by LiveClock when a timer fires.

@dataclass
class TimeEvent:
    name: str     # timer name
    ts_event: int # scheduled fire time (nanoseconds, UTC epoch)
    ts_init: int  # actual call time   (nanoseconds, UTC epoch)