nautilus_execution/python/
latency.rs1use nautilus_common::messages::execution::TradingCommand;
19use nautilus_core::{
20 DurationNanos, UnixNanos,
21 python::{clone_py_object, to_pynotimplemented_err, to_pytype_err},
22};
23use pyo3::{
24 IntoPyObjectExt,
25 prelude::*,
26 types::{PyBool, PyDict, PyTuple},
27};
28
29use crate::models::latency::{LatencyModel, LatencyModelAny, StaticLatencyModel};
30
31#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.execution")]
33#[pyclass(module = "nautilus_trader.execution", name = "LatencyModel", subclass)]
34#[derive(Debug)]
35pub struct PyLatencyModel;
36
37#[pyo3_stub_gen::derive::gen_stub_pymethods]
38#[pymethods]
39impl PyLatencyModel {
40 #[new]
41 #[gen_stub(override_return_type(type_repr = "typing.Self", imports = ("typing",)))]
42 #[pyo3(signature = (*_args, **_kwargs))]
43 fn py_new(_args: &Bound<'_, PyTuple>, _kwargs: Option<&Bound<'_, PyDict>>) -> Self {
44 Self
45 }
46
47 fn get_insert_latency(&self, _command: &Bound<'_, PyAny>, _ts_now: u64) -> PyResult<u64> {
49 Err(to_pynotimplemented_err(
50 "Method 'get_insert_latency' must be implemented in a subclass.",
51 ))
52 }
53
54 fn get_update_latency(&self, _command: &Bound<'_, PyAny>, _ts_now: u64) -> PyResult<u64> {
56 Err(to_pynotimplemented_err(
57 "Method 'get_update_latency' must be implemented in a subclass.",
58 ))
59 }
60
61 fn get_delete_latency(&self, _command: &Bound<'_, PyAny>, _ts_now: u64) -> PyResult<u64> {
63 Err(to_pynotimplemented_err(
64 "Method 'get_delete_latency' must be implemented in a subclass.",
65 ))
66 }
67}
68
69#[derive(Debug)]
71pub struct PythonLatencyModel {
72 obj: Py<PyAny>,
73}
74
75impl Clone for PythonLatencyModel {
76 fn clone(&self) -> Self {
77 Self {
78 obj: clone_py_object(&self.obj),
79 }
80 }
81}
82
83impl LatencyModel for PythonLatencyModel {
84 fn get_insert_latency(
85 &self,
86 command: &TradingCommand,
87 ts_now: UnixNanos,
88 ) -> anyhow::Result<DurationNanos> {
89 self.call_latency("get_insert_latency", command, ts_now)
90 }
91
92 fn get_update_latency(
93 &self,
94 command: &TradingCommand,
95 ts_now: UnixNanos,
96 ) -> anyhow::Result<DurationNanos> {
97 self.call_latency("get_update_latency", command, ts_now)
98 }
99
100 fn get_delete_latency(
101 &self,
102 command: &TradingCommand,
103 ts_now: UnixNanos,
104 ) -> anyhow::Result<DurationNanos> {
105 self.call_latency("get_delete_latency", command, ts_now)
106 }
107
108 fn get_base_latency(&self) -> DurationNanos {
109 DurationNanos::ZERO
110 }
111}
112
113impl PythonLatencyModel {
114 fn call_latency(
115 &self,
116 method: &str,
117 command: &TradingCommand,
118 ts_now: UnixNanos,
119 ) -> anyhow::Result<DurationNanos> {
120 Python::attach(|py| -> PyResult<DurationNanos> {
121 let command = match command {
122 TradingCommand::SubmitOrder(cmd) => cmd.clone().into_py_any(py)?,
123 TradingCommand::SubmitOrderList(cmd) => cmd.clone().into_py_any(py)?,
124 TradingCommand::ModifyOrder(cmd) => cmd.clone().into_py_any(py)?,
125 TradingCommand::ModifyOrders(cmd) => cmd.clone().into_py_any(py)?,
126 TradingCommand::CancelOrder(cmd) => cmd.clone().into_py_any(py)?,
127 TradingCommand::CancelOrders(cmd) => cmd.clone().into_py_any(py)?,
128 TradingCommand::CancelAllOrders(cmd) => cmd.clone().into_py_any(py)?,
129 TradingCommand::QueryOrder(cmd) => cmd.clone().into_py_any(py)?,
130 TradingCommand::QueryAccount(cmd) => cmd.clone().into_py_any(py)?,
131 };
132
133 let latency = self
134 .obj
135 .bind(py)
136 .call_method1(method, (command, ts_now.as_u64()))?;
137
138 if latency.is_instance_of::<PyBool>() {
139 return Err(to_pytype_err(
140 "Latency must be an integer nanosecond delay, not bool",
141 ));
142 }
143
144 let nanos = latency.extract::<u64>()?;
145 Ok(DurationNanos::new(nanos))
146 })
147 .map_err(|e| anyhow::anyhow!("Python LatencyModel.{method} failed: {e}"))
148 }
149}
150
151#[pymethods]
152#[pyo3_stub_gen::derive::gen_stub_pymethods]
153impl StaticLatencyModel {
154 #[new]
163 #[pyo3(signature = (
164 base_latency_nanos = 0,
165 insert_latency_nanos = 0,
166 update_latency_nanos = 0,
167 cancel_latency_nanos = 0,
168 ))]
169 fn py_new(
170 base_latency_nanos: u64,
171 insert_latency_nanos: u64,
172 update_latency_nanos: u64,
173 cancel_latency_nanos: u64,
174 ) -> Self {
175 Self::new(
176 DurationNanos::new(base_latency_nanos),
177 DurationNanos::new(insert_latency_nanos),
178 DurationNanos::new(update_latency_nanos),
179 DurationNanos::new(cancel_latency_nanos),
180 )
181 }
182
183 fn __repr__(&self) -> String {
184 format!("{self:?}")
185 }
186}
187
188pub fn pyobject_to_latency_model_any(obj: &Bound<'_, PyAny>) -> PyResult<LatencyModelAny> {
194 if let Ok(m) = obj.extract::<StaticLatencyModel>() {
195 return Ok(LatencyModelAny::Static(m));
196 }
197
198 if obj.is_exact_instance_of::<PyLatencyModel>() {
199 return Err(to_pytype_err(
200 "LatencyModel is a base class; use StaticLatencyModel or a LatencyModel subclass",
201 ));
202 }
203
204 if obj.is_instance_of::<PyLatencyModel>() {
205 return Ok(LatencyModelAny::Python(PythonLatencyModel {
206 obj: obj.clone().unbind(),
207 }));
208 }
209
210 let type_name = obj.get_type().name()?;
211 Err(to_pytype_err(format!(
212 "Cannot convert {type_name} to LatencyModel"
213 )))
214}
215
216pub fn latency_model_any_to_pyobject(
222 py: Python<'_>,
223 model: &LatencyModelAny,
224) -> PyResult<Py<PyAny>> {
225 match model {
226 LatencyModelAny::Static(model) => model.clone().into_py_any(py),
227 LatencyModelAny::Python(model) => Ok(model.obj.clone_ref(py)),
228 }
229}
230
231#[cfg(test)]
232mod tests {
233 use std::ffi::CString;
234
235 use nautilus_common::messages::execution::CancelAllOrders;
236 use nautilus_core::UUID4;
237 use nautilus_model::identifiers::{InstrumentId, StrategyId, TraderId};
238 use pyo3::ffi::c_str;
239 use rstest::rstest;
240
241 use super::*;
242
243 #[rstest]
244 fn test_python_latency_model_dispatches_context_and_preserves_instance() {
245 Python::initialize();
246 Python::attach(|py| {
247 let locals = PyDict::new(py);
248 locals
249 .set_item("LatencyModel", py.get_type::<PyLatencyModel>())
250 .unwrap();
251 py.run(
252 c_str!(
253 r#"
254class ReplayLatency(LatencyModel):
255 def __init__(self, offset):
256 self.offset = offset
257 self.calls = []
258 def get_insert_latency(self, command, ts_now):
259 self.calls.append((str(command.instrument_id), command.ts_init, ts_now, "insert"))
260 return self.offset + 11
261 def get_update_latency(self, command, ts_now):
262 self.calls.append((str(command.instrument_id), command.ts_init, ts_now, "update"))
263 return self.offset + 22
264 def get_delete_latency(self, command, ts_now):
265 self.calls.append((str(command.instrument_id), command.ts_init, ts_now, "delete"))
266 return self.offset + 33
267model = ReplayLatency(100)
268"#
269 ),
270 Some(&locals),
271 None,
272 )
273 .unwrap();
274 let obj = locals.get_item("model").unwrap().unwrap();
275 let model = pyobject_to_latency_model_any(&obj).unwrap();
276 let cloned = model.clone();
277 assert!(
278 latency_model_any_to_pyobject(py, &cloned)
279 .unwrap()
280 .bind(py)
281 .is(&obj)
282 );
283 let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
284 TraderId::from("TEST-001"),
285 None,
286 StrategyId::from("S-001"),
287 InstrumentId::from("AUD/USD.SIM"),
288 None,
289 UUID4::new(),
290 UnixNanos::new(17),
291 None,
292 None,
293 ));
294 let insert = model
295 .get_insert_latency(&command, UnixNanos::new(41))
296 .unwrap();
297 let update = cloned
298 .get_update_latency(&command, UnixNanos::new(52))
299 .unwrap();
300 let delete = model
301 .get_delete_latency(&command, UnixNanos::new(63))
302 .unwrap();
303 let calls: Vec<(String, u64, u64, String)> =
304 obj.getattr("calls").unwrap().extract().unwrap();
305 assert_eq!(
306 (insert, update, delete),
307 (
308 DurationNanos::new(111),
309 DurationNanos::new(122),
310 DurationNanos::new(133)
311 )
312 );
313 assert_eq!(model.get_base_latency(), DurationNanos::ZERO);
314 assert_eq!(
315 calls,
316 vec![
317 ("AUD/USD.SIM".to_string(), 17, 41, "insert".to_string()),
318 ("AUD/USD.SIM".to_string(), 17, 52, "update".to_string()),
319 ("AUD/USD.SIM".to_string(), 17, 63, "delete".to_string()),
320 ]
321 );
322 });
323 }
324
325 #[rstest]
326 #[case("-1")]
327 #[case("1.5")]
328 #[case("None")]
329 #[case("True")]
330 #[case("2 ** 64")]
331 fn test_python_latency_model_rejects_invalid_delay(#[case] value: &str) {
332 Python::initialize();
333 Python::attach(|py| {
334 let locals = PyDict::new(py);
335 locals
336 .set_item("LatencyModel", py.get_type::<PyLatencyModel>())
337 .unwrap();
338 let source = format!(
339 "type('InvalidLatency', (LatencyModel,), {{'get_delete_latency': lambda self, command, ts_now: {value}}})()"
340 );
341 let source = CString::new(source).unwrap();
342 let obj = py.eval(&source, Some(&locals), None).unwrap();
343 let model = pyobject_to_latency_model_any(&obj).unwrap();
344 let command = TradingCommand::CancelAllOrders(CancelAllOrders::new(
345 TraderId::from("TEST-001"),
346 None,
347 StrategyId::from("S-001"),
348 InstrumentId::from("AUD/USD.SIM"),
349 None,
350 UUID4::new(),
351 UnixNanos::new(17),
352 None,
353 None,
354 ));
355 let error = model
356 .get_delete_latency(&command, UnixNanos::new(31))
357 .unwrap_err();
358 assert!(
359 error
360 .to_string()
361 .starts_with("Python LatencyModel.get_delete_latency failed:")
362 );
363 });
364 }
365}