nautilus_coinapi/csv/
ohlcv.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//! `T-OHLCV_ACTIVE_CONSOLIDATED` files.
17
18use std::path::Path;
19
20use ahash::AHashMap;
21use nautilus_core::UnixNanos;
22use nautilus_model::{
23    data::{Bar, BarType},
24    enums::AggregationSource,
25    identifiers::InstrumentId,
26    instruments::Instrument,
27    types::fixed::FIXED_PRECISION,
28};
29use rust_decimal::{Decimal, RoundingStrategy};
30use ustr::Ustr;
31
32use super::rows::{Precision, Rows, date_time, number, text};
33use crate::{
34    http::{models::Symbol, parse::parse_instruments},
35    websocket::parse::{period_id_to_bar_spec, price_at, quantity_at},
36};
37
38/// Loads [`Bar`]s from exchange-wide `T-OHLCV_ACTIVE_CONSOLIDATED` files.
39///
40/// Each row names its symbol by the numeric `symbol_id`, the REST `symbol_id_int`, so the loader
41/// holds the instrument each one maps to and skips rows of unmapped symbols. Infers each
42/// instrument's price and size precisions from its values. Captured `sx_sum` values can carry
43/// numeric tails, so the loader rounds each volume half to even at the maximum precision.
44/// It rejects negative volumes and positive volumes rounded to zero.
45#[derive(Debug, Clone, Default)]
46#[cfg_attr(
47    feature = "python",
48    pyo3::pyclass(module = "nautilus_trader.adapters.coinapi", from_py_object)
49)]
50#[cfg_attr(
51    feature = "python",
52    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.coinapi")
53)]
54pub struct CoinApiOhlcvLoader {
55    instrument_ids: AHashMap<u64, InstrumentId>,
56}
57
58impl CoinApiOhlcvLoader {
59    /// Creates a loader mapping each numeric `symbol_id` to an instrument.
60    #[must_use]
61    pub fn new(instrument_ids: impl IntoIterator<Item = (u64, InstrumentId)>) -> Self {
62        Self {
63            instrument_ids: instrument_ids.into_iter().collect(),
64        }
65    }
66
67    /// Creates a loader for the instruments that CoinAPI symbol metadata maps to.
68    ///
69    /// Skips symbols without a `symbol_id_int` or a native instrument.
70    #[must_use]
71    pub fn from_symbols(symbols: &[Symbol]) -> Self {
72        let (instruments, _) = parse_instruments(symbols, UnixNanos::default());
73        let symbol_ids: AHashMap<Ustr, u64> = symbols
74            .iter()
75            .filter_map(|symbol| Some((symbol.symbol_id, symbol.symbol_id_int?)))
76            .collect();
77
78        Self::new(instruments.iter().filter_map(|loaded| {
79            let symbol_id = *symbol_ids.get(&loaded.symbol_id)?;
80            Some((symbol_id, loaded.instrument.id()))
81        }))
82    }
83
84    /// Loads the bars of mapped symbols from the file at `path`.
85    ///
86    /// The bar period comes from `period_id`, such as `1MIN`, or from the file's `TP-` name when
87    /// `period_id` is `None`.
88    ///
89    /// # Errors
90    ///
91    /// Returns an error if the period is unknown, the file cannot be read, a column is missing, or
92    /// a mapped row is invalid.
93    pub fn load_bars(
94        &self,
95        path: impl AsRef<Path>,
96        period_id: Option<&str>,
97        limit: Option<usize>,
98    ) -> anyhow::Result<Vec<Bar>> {
99        let path = path.as_ref();
100
101        let period_id = match period_id {
102            Some(period_id) => period_id,
103            None => file_period_id(path).ok_or_else(|| {
104                anyhow::anyhow!("no period given and '{}' is not a TP- file", path.display())
105            })?,
106        };
107
108        let spec = period_id_to_bar_spec(Ustr::from(period_id))
109            .ok_or_else(|| anyhow::anyhow!("unsupported period {period_id}"))?;
110
111        let mut rows = Rows::open(path)?;
112        let symbol_id = rows.column("symbol_id")?;
113        let time_period_end = rows.column("time_period_end")?;
114        let px_open = rows.column("px_open")?;
115        let px_high = rows.column("px_high")?;
116        let px_low = rows.column("px_low")?;
117        let px_close = rows.column("px_close")?;
118        let sx_sum = rows.column("sx_sum")?;
119
120        let mut bars = Vec::new();
121        let mut precisions: AHashMap<InstrumentId, (Precision, Precision)> = AHashMap::new();
122
123        while limit.is_none_or(|limit| bars.len() < limit) {
124            let Some(record) = rows.next()? else {
125                break;
126            };
127
128            let Some(&instrument_id) = text(record, symbol_id)?
129                .parse::<u64>()
130                .ok()
131                .and_then(|id| self.instrument_ids.get(&id))
132            else {
133                continue;
134            };
135
136            let open = number(record, px_open, "px_open")?;
137            let high = number(record, px_high, "px_high")?;
138            let low = number(record, px_low, "px_low")?;
139            let close = number(record, px_close, "px_close")?;
140            let source_volume = number(record, sx_sum, "sx_sum")?;
141            anyhow::ensure!(
142                source_volume >= Decimal::ZERO,
143                "sx_sum must be non-negative: {source_volume}"
144            );
145            let volume = source_volume.round_dp_with_strategy(
146                u32::from(FIXED_PRECISION),
147                RoundingStrategy::MidpointNearestEven,
148            );
149            anyhow::ensure!(
150                source_volume == Decimal::ZERO || volume > Decimal::ZERO,
151                "sx_sum {source_volume} rounds to zero at precision {FIXED_PRECISION}"
152            );
153
154            let (price, size) = precisions
155                .entry(instrument_id)
156                .or_insert((Precision::new(None), Precision::new(None)));
157
158            for value in [open, high, low, close] {
159                price.fit(value);
160            }
161
162            size.fit(volume);
163
164            let ts_event = date_time(text(record, time_period_end)?, "time_period_end")?;
165            bars.push(Bar::new_checked(
166                BarType::new(instrument_id, spec, AggregationSource::External),
167                price_at(open, price.value, "px_open")?,
168                price_at(high, price.value, "px_high")?,
169                price_at(low, price.value, "px_low")?,
170                price_at(close, price.value, "px_close")?,
171                quantity_at(volume, size.value, "sx_sum")?,
172                ts_event,
173                ts_event,
174            )?);
175        }
176
177        // Bars converted before a finer value keep their exact raw value at the final precision
178        for bar in &mut bars {
179            let (price, size) = precisions[&bar.instrument_id()];
180            bar.open.precision = price.value;
181            bar.high.precision = price.value;
182            bar.low.precision = price.value;
183            bar.close.precision = price.value;
184            bar.volume.precision = size.value;
185        }
186
187        Ok(bars)
188    }
189}
190
191fn file_period_id(path: &Path) -> Option<&str> {
192    let name = path.file_name()?.to_str()?;
193    let stem = name
194        .strip_suffix(".csv.gz")
195        .or_else(|| name.strip_suffix(".csv"))?;
196    stem.strip_prefix("TP-")
197}
nautilus_coinapi/csv/ohlcv.rs

nautilus_coinapi/csv/
ohlcv.rs
