nautilus_common/python/
response.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
5//  You may not use this file except in compliance with the License.
6//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
7//  Unless required by applicable law or agreed to in writing, software
8//  distributed under the License is distributed on an "AS IS" BASIS,
9//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10//  See the License for the specific language governing permissions and
11//  limitations under the License.
12// -------------------------------------------------------------------------------------------------
13
14//! Historical response conversion and Python callback delivery.
15
16use std::fmt::Debug;
17
18use nautilus_core::{
19    UUID4,
20    python::{params::params_to_pydict, to_pyruntime_err, to_pytype_err, to_pyvalue_err},
21};
22use nautilus_model::{
23    data::CustomData,
24    identifiers::ClientId,
25    python::{data::PyNautilusDataType, instruments::instrument_any_to_pyobject},
26};
27use pyo3::{
28    IntoPyObjectExt,
29    prelude::*,
30    sync::PyOnceLock,
31    types::{PyDict, PyList},
32};
33use ustr::Ustr;
34
35use crate::{
36    actor::{DataActor, DataActorNative, ResponseScope, registry::try_get_actor_unchecked},
37    messages::data::{DataResponse, DataResponseEnvelope},
38    msgbus::TypedHandler,
39    python::logging::format_exception,
40};
41
42/// Historical response metadata with a lazily converted Python payload.
43#[pyclass(name = "DataResponse", module = "nautilus_trader.common", frozen)]
44#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.common")]
45pub struct PyDataResponse {
46    envelope: DataResponseEnvelope,
47    data: PyOnceLock<Py<PyAny>>,
48}
49
50impl Debug for PyDataResponse {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        f.debug_struct(stringify!(PyDataResponse))
53            .field("envelope", &self.envelope)
54            .finish_non_exhaustive()
55    }
56}
57
58#[pymethods]
59#[pyo3_stub_gen::derive::gen_stub_pymethods]
60impl PyDataResponse {
61    #[getter]
62    fn correlation_id(&self) -> String {
63        self.envelope.correlation_id().to_string()
64    }
65
66    #[getter]
67    fn completed(&self) -> bool {
68        self.envelope.completed()
69    }
70
71    #[getter]
72    fn sub_correlation_id(&self) -> Option<String> {
73        self.envelope.sub_correlation_id().map(|id| id.to_string())
74    }
75
76    #[getter]
77    fn failure_reason(&self) -> Option<&str> {
78        self.envelope.failure_reason()
79    }
80
81    #[getter]
82    fn kind(&self) -> Option<PyNautilusDataType> {
83        self.envelope
84            .response()
85            .map(|r| PyNautilusDataType::new(r.data_kind()))
86    }
87
88    #[getter]
89    fn client_id(&self) -> Option<ClientId> {
90        self.envelope.response().map(|r| *r.client_id())
91    }
92
93    #[getter]
94    fn params(&self, py: Python<'_>) -> PyResult<Option<Py<PyDict>>> {
95        self.envelope
96            .response()
97            .and_then(DataResponse::params)
98            .map(|params| params_to_pydict(py, params))
99            .transpose()
100    }
101
102    #[getter]
103    fn start(&self) -> Option<u64> {
104        self.envelope
105            .response()
106            .and_then(DataResponse::start)
107            .map(|v| v.as_u64())
108    }
109
110    #[getter]
111    fn end(&self) -> Option<u64> {
112        self.envelope
113            .response()
114            .and_then(DataResponse::end)
115            .map(|v| v.as_u64())
116    }
117
118    #[getter]
119    fn ts_init(&self) -> Option<u64> {
120        self.envelope.response().map(|r| r.ts_init().as_u64())
121    }
122
123    #[getter]
124    fn data(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
125        self.data
126            .get_or_try_init(py, || response_data(py, self.envelope.response()))
127            .map(|value| value.clone_ref(py))
128    }
129}
130
131/// Builds a response handler that releases actor access before invoking Python hooks.
132#[doc(hidden)]
133pub fn response_handler<T>(
134    actor_id: Ustr,
135    instance: fn(&T) -> PyResult<Option<Py<PyAny>>>,
136    callback: Option<TypedHandler<DataResponseEnvelope>>,
137) -> TypedHandler<DataResponseEnvelope>
138where
139    T: DataActor + DataActorNative + Debug + 'static,
140{
141    TypedHandler::from(move |envelope: &DataResponseEnvelope| {
142        let Some((mut scope, instance)) = prepare_response::<T>(actor_id, instance, envelope)
143        else {
144            return;
145        };
146
147        Python::attach(|py| {
148            let wrapper = PyDataResponse {
149                envelope: envelope.clone(),
150                data: PyOnceLock::new(),
151            };
152
153            if let Some(instance) = &instance
154                && let Err(e) = dispatch_payload(py, instance, &wrapper)
155            {
156                log::error!(
157                    "Python historical data hook failed: {}",
158                    format_exception(&e)
159                );
160            }
161
162            let current = try_get_actor_unchecked::<T>(&actor_id)
163                .is_some_and(|actor| actor.core().complete_request(envelope, &scope));
164
165            if current {
166                dispatch_completion(py, instance.as_ref(), wrapper);
167                dispatch_callback::<T>(actor_id, &scope, envelope, callback.as_ref());
168            }
169
170            scope.leave();
171            drain_groups::<T>(py, actor_id, instance.as_ref(), &scope);
172        });
173    })
174}
175
176/// Builds a Python historical response handler with an optional terminal callback.
177///
178/// # Errors
179///
180/// Returns an error if `callback` is not callable.
181#[doc(hidden)]
182pub fn request_handler<T>(
183    py: Python<'_>,
184    actor_id: Ustr,
185    instance: fn(&T) -> PyResult<Option<Py<PyAny>>>,
186    callback: Option<Py<PyAny>>,
187) -> PyResult<TypedHandler<DataResponseEnvelope>>
188where
189    T: DataActor + DataActorNative + Debug + 'static,
190{
191    let callback = completion_handler(py, callback)?;
192    Ok(response_handler(actor_id, instance, callback))
193}
194
195fn drain_groups<T>(
196    py: Python<'_>,
197    actor_id: Ustr,
198    instance: Option<&Py<PyAny>>,
199    origin: &ResponseScope,
200) where
201    T: DataActor + DataActorNative + Debug + 'static,
202{
203    loop {
204        let prepared = try_get_actor_unchecked::<T>(&actor_id).and_then(|actor| {
205            if !actor.core().is_response_current(origin) {
206                return None;
207            }
208
209            let (envelope, callback) = actor.core().next_group_response()?;
210            let scope = actor.core().begin_response(envelope.correlation_id())?;
211            actor
212                .core()
213                .complete_request(&envelope, &scope)
214                .then_some((envelope, scope, callback))
215        });
216
217        let Some((envelope, scope, callback)) = prepared else {
218            return;
219        };
220
221        let wrapper = PyDataResponse {
222            envelope: envelope.clone(),
223            data: PyOnceLock::new(),
224        };
225
226        dispatch_completion(py, instance, wrapper);
227        dispatch_callback::<T>(actor_id, &scope, &envelope, callback.as_ref());
228    }
229}
230
231/// Extracts a request ID accepted by Python joins and pending inspection.
232///
233/// # Errors
234///
235/// Returns an error if the value is neither a UUID nor a valid UUID string.
236#[doc(hidden)]
237pub fn extract_request_id(value: &Bound<'_, PyAny>) -> PyResult<UUID4> {
238    if let Ok(id) = value.extract::<UUID4>() {
239        return Ok(id);
240    }
241
242    value.extract::<String>()?.parse().map_err(to_pyvalue_err)
243}
244
245/// Registers a Python completion join and drains ready notifications outside actor access.
246///
247/// # Errors
248///
249/// Returns an error if the actor is unavailable, its Python instance cannot be resolved,
250/// the callback is not callable, or the group IDs fail native validation.
251#[doc(hidden)]
252pub fn request_join<T>(
253    py: Python<'_>,
254    actor_id: Ustr,
255    instance: fn(&T) -> PyResult<Option<Py<PyAny>>>,
256    request_ids: Vec<UUID4>,
257    request_id: Option<UUID4>,
258    callback: Option<Py<PyAny>>,
259) -> PyResult<String>
260where
261    T: DataActor + DataActorNative + Debug + 'static,
262{
263    let callback = completion_handler(py, callback)?;
264
265    let (id, mut scope, instance) = {
266        let actor = try_get_actor_unchecked::<T>(&actor_id)
267            .ok_or_else(|| to_pyruntime_err("Actor is not registered"))?;
268        let instance = instance(&actor)?;
269        let id = actor
270            .core()
271            .register_request_group(request_ids, request_id, callback)
272            .map_err(to_pyvalue_err)?;
273        let scope = actor
274            .core()
275            .begin_response(&id)
276            .expect("newly registered group is pending");
277        (id, scope, instance)
278    };
279
280    scope.leave();
281    drain_groups::<T>(py, actor_id, instance.as_ref(), &scope);
282    Ok(id.to_string())
283}
284
285fn completion_handler(
286    py: Python<'_>,
287    callback: Option<Py<PyAny>>,
288) -> PyResult<Option<TypedHandler<DataResponseEnvelope>>> {
289    let Some(callback) = callback else {
290        return Ok(None);
291    };
292
293    if !callback.bind(py).is_callable() {
294        return Err(to_pytype_err("callback must be callable"));
295    }
296
297    Ok(Some(TypedHandler::from(
298        move |envelope: &DataResponseEnvelope| {
299            Python::attach(|py| {
300                if let Err(e) = callback.call1(py, (envelope.correlation_id().to_string(),)) {
301                    log::error!(
302                        "Python historical request callback failed: {}",
303                        format_exception(&e)
304                    );
305                }
306            });
307        },
308    )))
309}
310
311fn dispatch_callback<T>(
312    actor_id: Ustr,
313    scope: &ResponseScope,
314    envelope: &DataResponseEnvelope,
315    callback: Option<&TypedHandler<DataResponseEnvelope>>,
316) where
317    T: DataActor + DataActorNative + Debug + 'static,
318{
319    let Some(callback) = callback else {
320        return;
321    };
322
323    if envelope.completed()
324        && try_get_actor_unchecked::<T>(&actor_id)
325            .is_some_and(|actor| actor.core().is_response_current(scope))
326    {
327        callback.0.handle(envelope);
328    }
329}
330
331fn dispatch_completion(py: Python<'_>, instance: Option<&Py<PyAny>>, wrapper: PyDataResponse) {
332    let Some(instance) = instance else {
333        return;
334    };
335
336    let result = Py::new(py, wrapper)
337        .and_then(|response| instance.call_method1(py, "on_data_response", (response,)));
338
339    if let Err(e) = result {
340        log::error!("Python on_data_response failed: {}", format_exception(&e));
341    }
342}
343
344fn prepare_response<T>(
345    actor_id: Ustr,
346    instance: fn(&T) -> PyResult<Option<Py<PyAny>>>,
347    envelope: &DataResponseEnvelope,
348) -> Option<(ResponseScope, Option<Py<PyAny>>)>
349where
350    T: DataActor + DataActorNative + Debug + 'static,
351{
352    let Some(actor) = try_get_actor_unchecked::<T>(&actor_id) else {
353        log::error!("Actor {actor_id} not found for response handling");
354        return None;
355    };
356
357    let scope = actor.core().begin_response(envelope.correlation_id())?;
358    let result = match envelope.response() {
359        Some(DataResponse::Quotes(r)) => actor.core().handle_indicators_for_quotes(&r.data),
360        Some(DataResponse::Trades(r)) => actor.core().handle_indicators_for_trades(&r.data),
361        Some(DataResponse::Bars(r)) => actor.core().handle_indicators_for_bars(&r.data),
362        _ => Ok(()),
363    };
364
365    if let Err(e) = result {
366        log::error!("Historical indicator update failed: {e}");
367    }
368
369    match instance(&actor) {
370        Ok(instance) => Some((scope, instance)),
371        Err(e) => {
372            log::error!(
373                "Python response instance lookup failed: {}",
374                format_exception(&e)
375            );
376            Some((scope, None))
377        }
378    }
379}
380
381fn dispatch_payload(
382    py: Python<'_>,
383    instance: &Py<PyAny>,
384    wrapper: &PyDataResponse,
385) -> PyResult<()> {
386    let Some(response) = wrapper.envelope.response() else {
387        return Ok(());
388    };
389
390    match response {
391        DataResponse::Instruments(r) => {
392            for instrument in &r.data {
393                let value = instrument_any_to_pyobject(py, instrument.clone())?;
394                instance.call_method1(py, "on_instrument", (value,))?;
395            }
396
397            return Ok(());
398        }
399        DataResponse::OptionChainReferencePrice(_) => return Ok(()),
400        _ => {}
401    }
402
403    let hook = match response {
404        DataResponse::Data(_) => "on_historical_data",
405        DataResponse::Instrument(_) => "on_instrument",
406        DataResponse::Book(_) => "on_book",
407        DataResponse::BookDeltas(_) => "on_historical_book_deltas",
408        DataResponse::BookDepth(_) => "on_historical_book_depth",
409        DataResponse::Quotes(_) => "on_historical_quotes",
410        DataResponse::Trades(_) => "on_historical_trades",
411        DataResponse::FundingRates(_) => "on_historical_funding_rates",
412        DataResponse::Bars(_) => "on_historical_bars",
413        DataResponse::Instruments(_) | DataResponse::OptionChainReferencePrice(_) => unreachable!(),
414    };
415
416    let data = wrapper.data(py)?;
417    instance.call_method1(py, hook, (data,))?;
418    Ok(())
419}
420
421fn response_data(py: Python<'_>, response: Option<&DataResponse>) -> PyResult<Py<PyAny>> {
422    let Some(response) = response else {
423        return Ok(py.None());
424    };
425
426    match response {
427        DataResponse::Data(r) => {
428            if let Some(data) = r.data.downcast_ref::<CustomData>() {
429                data.clone().into_py_any(py)
430            } else if let Some(data) = r.data.downcast_ref::<Vec<CustomData>>() {
431                data.clone().into_py_any(py)
432            } else {
433                Err(to_pytype_err("Unsupported historical custom data payload"))
434            }
435        }
436        DataResponse::Instrument(r) => instrument_any_to_pyobject(py, r.data.clone()),
437        DataResponse::Instruments(r) => {
438            let values = r
439                .data
440                .iter()
441                .map(|value| instrument_any_to_pyobject(py, value.clone()))
442                .collect::<PyResult<Vec<_>>>()?;
443            Ok(PyList::new(py, values)?.into_any().unbind())
444        }
445        DataResponse::Book(r) => r.data.clone().into_py_any(py),
446        DataResponse::BookDeltas(r) => r.data.to_vec().into_py_any(py),
447        DataResponse::BookDepth(r) => r.data.to_vec().into_py_any(py),
448        DataResponse::Quotes(r) => r.data.to_vec().into_py_any(py),
449        DataResponse::Trades(r) => r.data.to_vec().into_py_any(py),
450        DataResponse::FundingRates(r) => r.data.to_vec().into_py_any(py),
451        DataResponse::Bars(r) => r.data.to_vec().into_py_any(py),
452        DataResponse::OptionChainReferencePrice(r) => r.price.into_py_any(py),
453    }
454}
455
456#[cfg(test)]
457mod tests {
458    use nautilus_core::{UUID4, UnixNanos};
459    use nautilus_model::{
460        data::{BatchView, DataType, NautilusDataType, QuoteTick},
461        identifiers::ClientId,
462    };
463    use rstest::rstest;
464
465    use super::*;
466    use crate::messages::data::{CustomDataResponse, QuotesResponse};
467
468    #[rstest]
469    fn test_response_metadata_is_lazy_and_terminal_can_omit_payload() {
470        Python::initialize();
471        Python::attach(|py| {
472            let root = UUID4::new();
473
474            let payload = CustomDataResponse::new(
475                root,
476                ClientId::new("HISTORY"),
477                None,
478                DataType::new("Opaque", None, None),
479                42_u8,
480                Some(UnixNanos::from(11)),
481                Some(UnixNanos::from(23)),
482                UnixNanos::from(37),
483                None,
484            );
485
486            let wrapper = PyDataResponse {
487                envelope: DataResponseEnvelope::new(DataResponse::Data(payload), false, None),
488                data: PyOnceLock::new(),
489            };
490
491            assert_eq!(wrapper.correlation_id(), root.to_string());
492            assert_eq!(
493                wrapper.kind().unwrap().inner(),
494                NautilusDataType::Custom {
495                    type_name: "Opaque".to_owned()
496                }
497            );
498            assert_eq!(wrapper.client_id(), Some(ClientId::new("HISTORY")));
499            assert_eq!(
500                (wrapper.start(), wrapper.end(), wrapper.ts_init()),
501                (Some(11), Some(23), Some(37))
502            );
503            assert!(!wrapper.completed());
504            assert!(wrapper.data.get(py).is_none());
505            assert!(wrapper.data(py).is_err());
506            assert!(wrapper.data.get(py).is_none());
507
508            let terminal = PyDataResponse {
509                envelope: DataResponseEnvelope::terminal(root, Some("source failed".to_owned())),
510                data: PyOnceLock::new(),
511            };
512
513            assert!(terminal.data(py).unwrap().is_none(py));
514            assert!(terminal.completed());
515            assert_eq!(terminal.failure_reason(), Some("source failed"));
516            assert!(terminal.kind().is_none());
517            assert!(terminal.client_id().is_none());
518        });
519    }
520
521    #[rstest]
522    fn test_response_data_is_an_ordinary_cached_python_list() {
523        Python::initialize();
524        Python::attach(|py| {
525            let payload = CustomDataResponse::new(
526                UUID4::new(),
527                ClientId::new("HISTORY"),
528                None,
529                DataType::new("Custom", None, None),
530                Vec::<CustomData>::new(),
531                None,
532                None,
533                UnixNanos::from(37),
534                None,
535            );
536
537            let wrapper = PyDataResponse {
538                envelope: DataResponseEnvelope::new(DataResponse::Data(payload), true, None),
539                data: PyOnceLock::new(),
540            };
541
542            let first = wrapper.data(py).unwrap();
543            let second = wrapper.data(py).unwrap();
544            assert!(first.is(&second));
545            assert_eq!(first.bind(py).cast::<PyList>().unwrap().len(), 0);
546        });
547    }
548
549    #[rstest]
550    #[case::visible(1..2)]
551    #[case::empty(1..1)]
552    fn test_response_data_list_contains_only_visible_rows(#[case] range: std::ops::Range<usize>) {
553        Python::initialize();
554        Python::attach(|py| {
555            let rows = [11, 23, 37].map(|ts| QuoteTick {
556                ts_init: UnixNanos::from(ts),
557                ..Default::default()
558            });
559
560            let expected = rows[range.clone()].to_vec();
561
562            let payload = QuotesResponse::new(
563                UUID4::new(),
564                ClientId::new("HISTORY"),
565                rows[1].instrument_id,
566                BatchView::new(std::sync::Arc::new(rows.to_vec()), range),
567                None,
568                None,
569                UnixNanos::from(41),
570                None,
571            );
572
573            let wrapper = PyDataResponse {
574                envelope: DataResponseEnvelope::new(DataResponse::Quotes(payload), true, None),
575                data: PyOnceLock::new(),
576            };
577
578            let first = wrapper.data(py).unwrap();
579            let second = wrapper.data(py).unwrap();
580            let list = first.bind(py).cast::<PyList>().unwrap();
581
582            assert!(first.is(&second));
583            assert_eq!(list.extract::<Vec<QuoteTick>>().unwrap(), expected);
584        });
585    }
586}
nautilus_common/python/response.rs

nautilus_common/python/
response.rs
