nautilus_common/python/
logging.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use ahash::AHashMap;
17use log::LevelFilter;
18use nautilus_core::{UUID4, python::to_pyvalue_err};
19use nautilus_model::identifiers::TraderId;
20use pyo3::prelude::*;
21use ustr::Ustr;
22
23use crate::{
24    enums::{LogColor, LogLevel},
25    logging::{
26        self, headers,
27        logger::{self, LogGuard, LoggerConfig},
28        logging_clock_set_realtime_mode, logging_clock_set_static_mode,
29        logging_clock_set_static_time, logging_set_bypass, map_log_level_to_filter,
30        parse_level_filter_str,
31        writer::FileWriterConfig,
32    },
33};
34
35#[pymethods]
36impl LoggerConfig {
37    /// Creates a [`LoggerConfig`] from a spec string.
38    ///
39    /// # Errors
40    ///
41    /// Returns a Python exception if the spec string is invalid.
42    #[staticmethod]
43    #[pyo3(name = "from_spec")]
44    pub fn py_from_spec(spec: String) -> PyResult<Self> {
45        Self::from_spec(&spec).map_err(to_pyvalue_err)
46    }
47}
48
49#[pymethods]
50impl FileWriterConfig {
51    #[new]
52    #[pyo3(signature = (directory=None, file_name=None, file_format=None, file_rotate=None))]
53    #[must_use]
54    pub fn py_new(
55        directory: Option<String>,
56        file_name: Option<String>,
57        file_format: Option<String>,
58        file_rotate: Option<(u64, u32)>,
59    ) -> Self {
60        Self::new(directory, file_name, file_format, file_rotate)
61    }
62}
63
64/// Initialize tracing.
65///
66/// Tracing is meant to be used to trace/debug async Rust code. It can be
67/// configured to filter modules and write up to a specific level only using
68/// by passing a configuration using the `RUST_LOG` environment variable.
69///
70/// # Safety
71///
72/// Should only be called once during an applications run, ideally at the
73/// beginning of the run.
74///
75/// # Errors
76///
77/// Returns an error if tracing subscriber fails to initialize.
78#[pyfunction()]
79#[pyo3(name = "init_tracing")]
80pub fn py_init_tracing() -> PyResult<()> {
81    logging::init_tracing().map_err(to_pyvalue_err)
82}
83
84/// Initialize logging.
85///
86/// Logging should be used for Python and sync Rust logic which is most of
87/// the components in the [nautilus_trader](https://pypi.org/project/nautilus_trader) package.
88/// Logging can be configured to filter components and write up to a specific level only
89/// by passing a configuration using the `NAUTILUS_LOG` environment variable.
90///
91/// # Safety
92///
93/// Should only be called once during an applications run, ideally at the
94/// beginning of the run.
95/// Initializes logging via Python interface.
96///
97/// # Errors
98///
99/// Returns a Python exception if logger initialization fails.
100#[pyfunction]
101#[pyo3(name = "init_logging")]
102#[allow(clippy::too_many_arguments)]
103#[pyo3(signature = (trader_id, instance_id, level_stdout, level_file=None, component_levels=None, directory=None, file_name=None, file_format=None, file_rotate=None, is_colored=None, is_bypassed=None, print_config=None, log_components_only=None))]
104pub fn py_init_logging(
105    trader_id: TraderId,
106    instance_id: UUID4,
107    level_stdout: LogLevel,
108    level_file: Option<LogLevel>,
109    component_levels: Option<std::collections::HashMap<String, String>>,
110    directory: Option<String>,
111    file_name: Option<String>,
112    file_format: Option<String>,
113    file_rotate: Option<(u64, u32)>,
114    is_colored: Option<bool>,
115    is_bypassed: Option<bool>,
116    print_config: Option<bool>,
117    log_components_only: Option<bool>,
118) -> PyResult<LogGuard> {
119    let level_file = level_file.map_or(LevelFilter::Off, map_log_level_to_filter);
120
121    let component_levels = parse_component_levels(component_levels).map_err(to_pyvalue_err)?;
122
123    let config = LoggerConfig::new(
124        map_log_level_to_filter(level_stdout),
125        level_file,
126        component_levels,
127        log_components_only.unwrap_or(false),
128        is_colored.unwrap_or(true),
129        print_config.unwrap_or(false),
130    );
131
132    let file_config = FileWriterConfig::new(directory, file_name, file_format, file_rotate);
133
134    if is_bypassed.unwrap_or(false) {
135        logging_set_bypass();
136    }
137
138    logging::init_logging(trader_id, instance_id, config, file_config).map_err(to_pyvalue_err)
139}
140
141#[pyfunction()]
142#[pyo3(name = "logger_flush")]
143pub fn py_logger_flush() {
144    log::logger().flush();
145}
146
147fn parse_component_levels(
148    original_map: Option<std::collections::HashMap<String, String>>,
149) -> anyhow::Result<AHashMap<Ustr, LevelFilter>> {
150    match original_map {
151        Some(map) => {
152            let mut new_map = AHashMap::new();
153            for (key, value) in map {
154                let ustr_key = Ustr::from(&key);
155                let level = parse_level_filter_str(&value)?;
156                new_map.insert(ustr_key, level);
157            }
158            Ok(new_map)
159        }
160        None => Ok(AHashMap::new()),
161    }
162}
163
164/// Create a new log event.
165#[pyfunction]
166#[pyo3(name = "logger_log")]
167pub fn py_logger_log(level: LogLevel, color: LogColor, component: &str, message: &str) {
168    logger::log(level, color, Ustr::from(component), message);
169}
170
171/// Logs the standard Nautilus system header.
172#[pyfunction]
173#[pyo3(name = "log_header")]
174pub fn py_log_header(trader_id: TraderId, machine_id: &str, instance_id: UUID4, component: &str) {
175    headers::log_header(trader_id, machine_id, instance_id, Ustr::from(component));
176}
177
178/// Logs system information.
179#[pyfunction]
180#[pyo3(name = "log_sysinfo")]
181pub fn py_log_sysinfo(component: &str) {
182    headers::log_sysinfo(Ustr::from(component));
183}
184
185#[pyfunction]
186#[pyo3(name = "logging_clock_set_static_mode")]
187pub fn py_logging_clock_set_static_mode() {
188    logging_clock_set_static_mode();
189}
190
191#[pyfunction]
192#[pyo3(name = "logging_clock_set_realtime_mode")]
193pub fn py_logging_clock_set_realtime_mode() {
194    logging_clock_set_realtime_mode();
195}
196
197#[pyfunction]
198#[pyo3(name = "logging_clock_set_static_time")]
199pub fn py_logging_clock_set_static_time(time_ns: u64) {
200    logging_clock_set_static_time(time_ns);
201}
202
203/// A thin wrapper around the global Rust logger which exposes ergonomic
204/// logging helpers for Python code.
205///
206/// It mirrors the familiar Python `logging` interface while forwarding
207/// all records through the Nautilus logging infrastructure so that log levels
208/// and formatting remain consistent across Rust and Python.
209#[pyclass(
210    module = "nautilus_trader.core.nautilus_pyo3.common",
211    name = "Logger",
212    unsendable
213)]
214#[derive(Debug, Clone)]
215pub struct PyLogger {
216    name: Ustr,
217}
218
219impl PyLogger {
220    pub fn new(name: &str) -> Self {
221        Self {
222            name: Ustr::from(name),
223        }
224    }
225}
226
227#[pymethods]
228impl PyLogger {
229    /// Create a new `Logger` instance.
230    #[new]
231    #[pyo3(signature = (name="Python"))]
232    fn py_new(name: &str) -> Self {
233        Self::new(name)
234    }
235
236    /// The component identifier carried by this logger.
237    #[getter]
238    fn name(&self) -> &str {
239        &self.name
240    }
241
242    /// Emit a TRACE level record.
243    #[pyo3(name = "trace")]
244    fn py_trace(&self, message: &str, color: Option<LogColor>) {
245        self._log(LogLevel::Trace, color, message);
246    }
247
248    /// Emit a DEBUG level record.
249    #[pyo3(name = "debug")]
250    fn py_debug(&self, message: &str, color: Option<LogColor>) {
251        self._log(LogLevel::Debug, color, message);
252    }
253
254    /// Emit an INFO level record.
255    #[pyo3(name = "info")]
256    fn py_info(&self, message: &str, color: Option<LogColor>) {
257        self._log(LogLevel::Info, color, message);
258    }
259
260    /// Emit a WARNING level record.
261    #[pyo3(name = "warning")]
262    fn py_warning(&self, message: &str, color: Option<LogColor>) {
263        self._log(LogLevel::Warning, color, message);
264    }
265
266    /// Emit an ERROR level record.
267    #[pyo3(name = "error")]
268    fn py_error(&self, message: &str, color: Option<LogColor>) {
269        self._log(LogLevel::Error, color, message);
270    }
271
272    /// Emit an ERROR level record with the active Python exception info.
273    #[pyo3(name = "exception")]
274    #[pyo3(signature = (message="", color=None))]
275    fn py_exception(&self, py: Python, message: &str, color: Option<LogColor>) {
276        let mut full_msg = message.to_owned();
277
278        if pyo3::PyErr::occurred(py) {
279            let err = PyErr::fetch(py);
280            let err_str = err.to_string();
281            if full_msg.is_empty() {
282                full_msg = err_str;
283            } else {
284                full_msg = format!("{full_msg}: {err_str}");
285            }
286        }
287
288        self._log(LogLevel::Error, color, &full_msg);
289    }
290
291    /// Flush buffered log records.
292    #[pyo3(name = "flush")]
293    fn py_flush(&self) {
294        log::logger().flush();
295    }
296
297    fn _log(&self, level: LogLevel, color: Option<LogColor>, message: &str) {
298        let color = color.unwrap_or(LogColor::Normal);
299        logger::log(level, color, self.name, message);
300    }
301}
nautilus_common/python/logging.rs

nautilus_common/python/
logging.rs
