nautilus_data/python/
option_chain_manager.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
16//! PyO3 wrapper for the option chain aggregation engine.
17//!
18//! [`PyOptionChainManager`] wraps [`OptionChainAggregator`] and [`AtmTracker`],
19//! exposing them to the Cython `DataEngine` without Rust msgbus, clock, or timer
20//! dependencies. The Cython engine owns the lifecycle: subscription routing,
21//! timer management, and msgbus publishing.
22
23use std::collections::HashMap;
24
25use nautilus_core::{
26    UnixNanos,
27    python::{correctness_error_to_pyvalue_err, to_pyvalue_err},
28};
29use nautilus_model::{
30    data::{
31        QuoteTick,
32        option_chain::{OptionChainSlice, OptionGreeks},
33    },
34    enums::OptionKind,
35    identifiers::{InstrumentId, OptionSeriesId},
36    python::data::option_chain::PyStrikeRange,
37    types::Price,
38};
39use pyo3::prelude::*;
40
41use crate::option_chains::{AtmTracker, OptionChainAggregator};
42
43fn parse_option_kind(value: u8) -> PyResult<OptionKind> {
44    match value {
45        0 => Ok(OptionKind::Call),
46        1 => Ok(OptionKind::Put),
47        _ => Err(to_pyvalue_err(format!(
48            "invalid `OptionKind` value, expected 0 (Call) or 1 (Put), received {value}"
49        ))),
50    }
51}
52
53/// Python-facing option chain manager that wraps [`OptionChainAggregator`] and
54/// [`AtmTracker`].
55///
56/// The Cython `DataEngine` creates one instance per subscribed option series.
57/// It feeds incoming market data (quotes, greeks) through the `handle_*`
58/// methods and periodically calls `snapshot()` to publish `OptionChainSlice`
59/// objects to the message bus.
60///
61/// ATM price is always derived from the exchange-provided forward price
62/// embedded in each option greeks/ticker update.
63#[pyclass(
64    name = "OptionChainManager",
65    module = "nautilus_trader.core.nautilus_pyo3.data"
66)]
67#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.data")]
68#[derive(Debug)]
69pub struct PyOptionChainManager {
70    aggregator: OptionChainAggregator,
71    series_id: OptionSeriesId,
72    raw_mode: bool,
73    bootstrapped: bool,
74}
75
76#[pymethods]
77#[pyo3_stub_gen::derive::gen_stub_pymethods]
78impl PyOptionChainManager {
79    /// Creates a new option chain manager.
80    #[new]
81    #[pyo3(signature = (series_id, strike_range, instruments, snapshot_interval_ms=None, initial_atm_price=None))]
82    fn py_new(
83        series_id: OptionSeriesId,
84        strike_range: PyStrikeRange,
85        instruments: HashMap<InstrumentId, (Price, u8)>,
86        snapshot_interval_ms: Option<u64>,
87        initial_atm_price: Option<Price>,
88    ) -> PyResult<Self> {
89        let rust_instruments: HashMap<InstrumentId, (Price, OptionKind)> = instruments
90            .into_iter()
91            .map(|(id, (strike, kind_u8))| {
92                parse_option_kind(kind_u8).map(|kind| (id, (strike, kind)))
93            })
94            .collect::<PyResult<_>>()?;
95
96        let mut tracker = AtmTracker::new();
97
98        // Derive precision from instrument strikes
99        if let Some((strike, _)) = rust_instruments.values().next() {
100            tracker.set_forward_precision(strike.precision);
101        }
102
103        if let Some(price) = initial_atm_price {
104            tracker.set_initial_price(price);
105        }
106
107        let aggregator =
108            OptionChainAggregator::new(series_id, strike_range.inner, tracker, rust_instruments);
109
110        let active_ids = aggregator.instrument_ids();
111        let all_ids = aggregator.all_instrument_ids();
112        let bootstrapped = !active_ids.is_empty() || all_ids.is_empty();
113        let raw_mode = snapshot_interval_ms.is_none();
114
115        Ok(Self {
116            aggregator,
117            series_id,
118            raw_mode,
119            bootstrapped,
120        })
121    }
122
123    /// Feeds a quote tick to the aggregator.
124    ///
125    /// Returns `True` if the manager just bootstrapped (first ATM price arrived
126    /// and the active instrument set was computed for the first time).
127    #[pyo3(name = "handle_quote")]
128    fn py_handle_quote(&mut self, quote: &Bound<'_, PyAny>) -> PyResult<bool> {
129        let tick = quote
130            .extract::<QuoteTick>()
131            .or_else(|_| QuoteTick::from_pyobject(quote))?;
132        self.aggregator.update_quote(&tick);
133
134        if !self.bootstrapped && self.aggregator.atm_tracker().atm_price().is_some() {
135            self.aggregator.recompute_active_set();
136            self.bootstrapped = true;
137            return Ok(true);
138        }
139        Ok(false)
140    }
141
142    /// Feeds option greeks to the aggregator.
143    ///
144    /// Returns `True` if the manager just bootstrapped (ATM price derived from
145    /// the greeks' `underlying_price`).
146    #[pyo3(name = "handle_greeks")]
147    fn py_handle_greeks(&mut self, greeks_obj: &Bound<'_, PyAny>) -> PyResult<bool> {
148        let greeks = greeks_obj
149            .extract::<OptionGreeks>()
150            .or_else(|_| OptionGreeks::from_pyobject(greeks_obj))?;
151
152        self.aggregator
153            .atm_tracker_mut()
154            .try_update_from_option_greeks(&greeks)
155            .map_err(correctness_error_to_pyvalue_err)?;
156
157        self.aggregator.update_greeks(&greeks);
158
159        if !self.bootstrapped && self.aggregator.atm_tracker().atm_price().is_some() {
160            self.aggregator.recompute_active_set();
161            self.bootstrapped = true;
162            return Ok(true);
163        }
164        Ok(false)
165    }
166
167    /// Creates a point-in-time snapshot from accumulated buffers.
168    ///
169    /// Returns `None` if no data has been accumulated yet (both buffers empty).
170    #[pyo3(name = "snapshot")]
171    fn py_snapshot(&self, ts_ns: u64) -> Option<OptionChainSlice> {
172        if self.aggregator.is_buffer_empty() {
173            return None;
174        }
175        Some(self.aggregator.snapshot(UnixNanos::from(ts_ns)))
176    }
177
178    /// Checks whether instruments should be rebalanced around the current ATM.
179    ///
180    /// Returns `None` when no rebalancing is needed.
181    /// Returns `(added_ids, removed_ids)` when the active set should change.
182    /// The caller is responsible for subscribing/unsubscribing instruments.
183    #[pyo3(name = "check_rebalance")]
184    fn py_check_rebalance(&mut self, ts_ns: u64) -> Option<(Vec<InstrumentId>, Vec<InstrumentId>)> {
185        let now = UnixNanos::from(ts_ns);
186        let action = self.aggregator.check_rebalance(now)?;
187        let add = action.add.clone();
188        let remove = action.remove.clone();
189        self.aggregator.apply_rebalance(&action, now);
190        Some((add, remove))
191    }
192
193    /// Returns the currently active instrument IDs (the subset being tracked).
194    #[pyo3(name = "active_instrument_ids")]
195    fn py_active_instrument_ids(&self) -> Vec<InstrumentId> {
196        self.aggregator.instrument_ids()
197    }
198
199    /// Returns all instrument IDs in the full catalog.
200    #[pyo3(name = "all_instrument_ids")]
201    fn py_all_instrument_ids(&self) -> Vec<InstrumentId> {
202        self.aggregator.all_instrument_ids()
203    }
204
205    /// Adds a newly discovered instrument to the series.
206    ///
207    /// Returns `True` if the instrument was newly inserted.
208    #[pyo3(name = "add_instrument")]
209    fn py_add_instrument(
210        &mut self,
211        instrument_id: InstrumentId,
212        strike: Price,
213        kind: u8,
214    ) -> PyResult<bool> {
215        let option_kind = parse_option_kind(kind)?;
216        Ok(self
217            .aggregator
218            .add_instrument(instrument_id, strike, option_kind))
219    }
220
221    /// Removes an instrument from the catalog.
222    ///
223    /// Returns `True` if the catalog is now empty.
224    #[pyo3(name = "remove_instrument")]
225    fn py_remove_instrument(&mut self, instrument_id: InstrumentId) -> bool {
226        let _ = self.aggregator.remove_instrument(&instrument_id);
227        self.aggregator.is_catalog_empty()
228    }
229
230    #[getter]
231    #[pyo3(name = "series_id")]
232    fn py_series_id(&self) -> OptionSeriesId {
233        self.series_id
234    }
235
236    #[getter]
237    #[pyo3(name = "bootstrapped")]
238    fn py_bootstrapped(&self) -> bool {
239        self.bootstrapped
240    }
241
242    #[getter]
243    #[pyo3(name = "raw_mode")]
244    fn py_raw_mode(&self) -> bool {
245        self.raw_mode
246    }
247
248    #[getter]
249    #[pyo3(name = "atm_price")]
250    fn py_atm_price(&self) -> Option<Price> {
251        self.aggregator.atm_tracker().atm_price()
252    }
253
254    fn __repr__(&self) -> String {
255        format!(
256            "OptionChainManager(series_id={}, bootstrapped={}, raw_mode={}, \
257             active={}/{})",
258            self.series_id,
259            self.bootstrapped,
260            self.raw_mode,
261            self.aggregator.instrument_ids().len(),
262            self.aggregator.all_instrument_ids().len(),
263        )
264    }
265}
nautilus_data/python/option_chain_manager.rs

nautilus_data/python/
option_chain_manager.rs
