data_preprocess/
models.rs

1use chrono::NaiveDateTime;
2use serde::{Deserialize, Serialize};
3
4use crate::error::{DataError, Result};
5
6/// Query parameters for tick view/query commands.
7pub struct QueryOpts {
8    pub exchange: String,
9    pub symbol: String,
10    pub from: Option<NaiveDateTime>,
11    pub to: Option<NaiveDateTime>,
12    pub limit: usize,
13    pub tail: bool,
14    pub descending: bool,
15}
16
17/// Query parameters for bar view/query commands.
18pub struct BarQueryOpts {
19    pub exchange: String,
20    pub symbol: String,
21    pub timeframe: String,
22    pub from: Option<NaiveDateTime>,
23    pub to: Option<NaiveDateTime>,
24    pub limit: usize,
25    pub tail: bool,
26    pub descending: bool,
27}
28
29/// Supported bar timeframes.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
31pub enum Timeframe {
32    M1,
33    M3,
34    M5,
35    M15,
36    M30,
37    H1,
38    H4,
39    D1,
40    W1,
41    MN1,
42}
43
44impl Timeframe {
45    /// Parse a CLI or storage timeframe, preserving the canonical monthly `1M` label.
46    pub fn parse(s: &str) -> Result<Self> {
47        if s == "1M" {
48            return Ok(Self::MN1);
49        }
50
51        match s.to_ascii_lowercase().as_str() {
52            "1m" | "m1" => Ok(Self::M1),
53            "3m" | "m3" => Ok(Self::M3),
54            "5m" | "m5" => Ok(Self::M5),
55            "15m" | "m15" => Ok(Self::M15),
56            "30m" | "m30" => Ok(Self::M30),
57            "1h" | "h1" => Ok(Self::H1),
58            "4h" | "h4" => Ok(Self::H4),
59            "1d" | "d1" => Ok(Self::D1),
60            "1w" | "w1" => Ok(Self::W1),
61            "1mn" | "mn1" | "1m0" | "mn" => Ok(Self::MN1),
62            _ => Err(DataError::InvalidTimeframe(s.to_string())),
63        }
64    }
65
66    /// Length in seconds for timeframes that have a fixed duration.
67    ///
68    /// Monthly bars have no fixed length and return `None`, so a caller that needs deterministic bucket arithmetic must reject them. Weekly buckets are fixed-length but align to the Unix epoch week unless the caller supplies an alignment offset.
69    pub fn fixed_duration_seconds(&self) -> Option<i64> {
70        let seconds = match self {
71            Self::M1 => 60,
72            Self::M3 => 180,
73            Self::M5 => 300,
74            Self::M15 => 900,
75            Self::M30 => 1_800,
76            Self::H1 => 3_600,
77            Self::H4 => 14_400,
78            Self::D1 => 86_400,
79            Self::W1 => 604_800,
80            Self::MN1 => return None,
81        };
82        Some(seconds)
83    }
84
85    /// Canonical short label for storage: "1m", "3m", "5m", ...
86    pub fn as_str(&self) -> &'static str {
87        match self {
88            Self::M1 => "1m",
89            Self::M3 => "3m",
90            Self::M5 => "5m",
91            Self::M15 => "15m",
92            Self::M30 => "30m",
93            Self::H1 => "1h",
94            Self::H4 => "4h",
95            Self::D1 => "1d",
96            Self::W1 => "1w",
97            Self::MN1 => "1M",
98        }
99    }
100}
101
102impl std::fmt::Display for Timeframe {
103    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
104        f.write_str(self.as_str())
105    }
106}
107
108/// A single tick (bid/ask/last at a point in time).
109#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
110pub struct Tick {
111    pub exchange: String,
112    pub symbol: String,
113    pub ts: NaiveDateTime,
114    pub bid: Option<f64>,
115    pub ask: Option<f64>,
116    pub last: Option<f64>,
117    pub volume: Option<f64>,
118    pub flags: Option<i32>,
119}
120
121/// A single OHLCV bar.
122#[derive(Debug, Clone, Serialize, Deserialize)]
123pub struct Bar {
124    pub exchange: String,
125    pub symbol: String,
126    pub timeframe: Timeframe,
127    pub ts: NaiveDateTime,
128    pub open: f64,
129    pub high: f64,
130    pub low: f64,
131    pub close: f64,
132    pub tick_vol: i64,
133    pub volume: i64,
134    pub spread: i32,
135}
136
137/// Observed capabilities of one legacy tick CSV import.
138#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
139pub struct TickImportAudit {
140    pub parsed_rows: usize,
141    pub distinct_timestamps: usize,
142    pub simultaneous_rows: usize,
143    pub maximum_fractional_digits: u8,
144    pub provider_sequence_available: bool,
145    pub stable_import_ordinal_persisted: bool,
146    pub legacy_timestamp_dedup: bool,
147    pub exact_quote_path_capable: bool,
148}
149
150impl TickImportAudit {
151    pub fn parquet_precision_loss_possible(&self) -> bool {
152        self.maximum_fractional_digits > 6
153    }
154}
155
156/// Declared capability of a stored tick path.
157#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
158#[serde(rename_all = "snake_case")]
159pub enum TickPathCapability {
160    LegacyTimestampDeduplicated,
161    OrderedWithoutProviderSequence,
162    OrderedWithProviderSequence,
163}
164
165impl TickPathCapability {
166    pub const fn preserves_simultaneous_rows(self) -> bool {
167        !matches!(self, Self::LegacyTimestampDeduplicated)
168    }
169
170    pub const fn exact_quote_path_capable(self) -> bool {
171        !matches!(self, Self::LegacyTimestampDeduplicated)
172    }
173
174    pub const fn verifies_true_duplicates(self) -> bool {
175        matches!(self, Self::OrderedWithProviderSequence)
176    }
177}
178
179/// Stable imported quote with persisted source order and optional provider identity.
180#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
181pub struct StoredTick {
182    pub tick: Tick,
183    pub source_ordinal: u64,
184    pub source_identity: Option<String>,
185    pub provider_sequence: Option<u64>,
186}
187
188impl StoredTick {
189    pub fn validate(&self) -> Result<()> {
190        if self.source_identity.as_deref().is_some_and(str::is_empty)
191            || self.source_identity.is_some() != self.provider_sequence.is_some()
192        {
193            return Err(DataError::Other(
194                "provider sequence and nonempty source identity must be supplied together".into(),
195            ));
196        }
197        Ok(())
198    }
199}
200
201#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
202#[serde(rename_all = "snake_case")]
203pub enum StoredPriceBasis {
204    Bid,
205    Ask,
206    Mid,
207    Last,
208}
209
210#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
211#[serde(rename_all = "snake_case")]
212pub enum CountCapability {
213    KnownPositive,
214    Optional,
215    Unavailable,
216}
217
218/// Caller-owned physical-series contract. `verified = false` identifies a legacy assertion.
219#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
220pub struct SeriesDescriptor {
221    pub source_identity: String,
222    pub exchange: String,
223    pub symbol: String,
224    pub timeframe_seconds: u64,
225    pub price_basis: StoredPriceBasis,
226    pub alignment_offset_seconds: i64,
227    pub digits: u32,
228    pub point_size: f64,
229    pub count_capability: CountCapability,
230    pub verified: bool,
231}
232
233impl SeriesDescriptor {
234    pub fn validate(&self) -> Result<()> {
235        if self.source_identity.is_empty()
236            || self.exchange.is_empty()
237            || self.symbol.is_empty()
238            || self.timeframe_seconds == 0
239            || !self.point_size.is_finite()
240            || self.point_size <= 0.0
241            || self.digits > 18
242        {
243            return Err(DataError::Other("invalid series descriptor".into()));
244        }
245        Ok(())
246    }
247}
248
249/// Additive price-only bar representation with an explicit optional source count.
250#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
251pub struct PriceBar {
252    pub exchange: String,
253    pub symbol: String,
254    pub timeframe: Timeframe,
255    pub ts: NaiveDateTime,
256    pub available_at: NaiveDateTime,
257    pub open: f64,
258    pub high: f64,
259    pub low: f64,
260    pub close: f64,
261    pub tick_count: Option<u64>,
262    pub spread: Option<i32>,
263}
264
265impl PriceBar {
266    pub fn validate(&self) -> Result<()> {
267        if self.exchange.is_empty()
268            || self.symbol.is_empty()
269            || self.available_at < self.ts
270            || ![self.open, self.high, self.low, self.close]
271                .into_iter()
272                .all(f64::is_finite)
273            || self.open <= 0.0
274            || self.high < self.open.max(self.close)
275            || self.low > self.open.min(self.close)
276            || self.low <= 0.0
277            || self.tick_count == Some(0)
278        {
279            return Err(DataError::Other("invalid price-only bar".into()));
280        }
281        Ok(())
282    }
283}
284
285impl TryFrom<Bar> for PriceBar {
286    type Error = DataError;
287
288    fn try_from(value: Bar) -> Result<Self> {
289        let tick_count = u64::try_from(value.tick_vol)
290            .ok()
291            .filter(|count| *count > 0)
292            .ok_or_else(|| DataError::Other("legacy tick count must be positive".into()))?;
293        let duration = value.timeframe.fixed_duration_seconds().ok_or_else(|| {
294            DataError::Other("price-only conversion requires fixed timeframe".into())
295        })?;
296        let available_at = value
297            .ts
298            .checked_add_signed(chrono::Duration::seconds(duration))
299            .ok_or_else(|| DataError::Other("bar availability overflowed".into()))?;
300        let result = Self {
301            exchange: value.exchange,
302            symbol: value.symbol,
303            timeframe: value.timeframe,
304            ts: value.ts,
305            available_at,
306            open: value.open,
307            high: value.high,
308            low: value.low,
309            close: value.close,
310            tick_count: Some(tick_count),
311            spread: Some(value.spread),
312        };
313        result.validate()?;
314        Ok(result)
315    }
316}
317
318impl TryFrom<PriceBar> for Bar {
319    type Error = DataError;
320
321    fn try_from(value: PriceBar) -> Result<Self> {
322        value.validate()?;
323        let count = value
324            .tick_count
325            .ok_or_else(|| DataError::Other("legacy bar requires a known count".into()))?;
326        let tick_vol = i64::try_from(count)
327            .map_err(|_| DataError::Other("tick count exceeds legacy i64 range".into()))?;
328        Ok(Self {
329            exchange: value.exchange,
330            symbol: value.symbol,
331            timeframe: value.timeframe,
332            ts: value.ts,
333            open: value.open,
334            high: value.high,
335            low: value.low,
336            close: value.close,
337            tick_vol,
338            volume: 0,
339            spread: value.spread.unwrap_or(0),
340        })
341    }
342}
343
344/// Aggregate exactly one complete, ordered parent bucket. No incomplete tail is emitted.
345pub fn aggregate_price_bars(
346    children: &[PriceBar],
347    parent_timeframe: Timeframe,
348    descriptor: &SeriesDescriptor,
349) -> Result<PriceBar> {
350    descriptor.validate()?;
351    let child_seconds = descriptor.timeframe_seconds;
352    let parent_seconds = u64::try_from(
353        parent_timeframe
354            .fixed_duration_seconds()
355            .ok_or_else(|| DataError::Other("parent timeframe must be fixed".into()))?,
356    )
357    .map_err(|_| DataError::Other("invalid parent duration".into()))?;
358    if parent_seconds <= child_seconds || parent_seconds % child_seconds != 0 {
359        return Err(DataError::Other(
360            "parent duration must be a larger exact multiple of child duration".into(),
361        ));
362    }
363    let expected = usize::try_from(parent_seconds / child_seconds)
364        .map_err(|_| DataError::Other("parent child count overflowed".into()))?;
365    if children.len() != expected {
366        return Err(DataError::Other("parent bucket is incomplete".into()));
367    }
368    let first = &children[0];
369    let offset = descriptor.alignment_offset_seconds;
370    let parent_i64 = i64::try_from(parent_seconds)
371        .map_err(|_| DataError::Other("parent duration exceeds i64".into()))?;
372    if (first.ts.and_utc().timestamp() - offset).rem_euclid(parent_i64) != 0 {
373        return Err(DataError::Other("parent bucket is misaligned".into()));
374    }
375    let child_i64 = i64::try_from(child_seconds)
376        .map_err(|_| DataError::Other("child duration exceeds i64".into()))?;
377    for (index, child) in children.iter().enumerate() {
378        child.validate()?;
379        let expected_ts = first
380            .ts
381            .checked_add_signed(chrono::Duration::seconds(
382                child_i64
383                    .checked_mul(i64::try_from(index).unwrap())
384                    .ok_or_else(|| DataError::Other("child timestamp overflowed".into()))?,
385            ))
386            .ok_or_else(|| DataError::Other("child timestamp overflowed".into()))?;
387        if child.exchange != descriptor.exchange
388            || child.symbol != descriptor.symbol
389            || child.ts != expected_ts
390            || u64::try_from(child.timeframe.fixed_duration_seconds().unwrap_or_default()).ok()
391                != Some(child_seconds)
392        {
393            return Err(DataError::Other(
394                "child identity, duration, order or completeness mismatch".into(),
395            ));
396        }
397    }
398    let tick_count =
399        children
400            .iter()
401            .try_fold(Some(0_u64), |sum, child| match (sum, child.tick_count) {
402                (Some(sum), Some(count)) => sum
403                    .checked_add(count)
404                    .map(Some)
405                    .ok_or_else(|| DataError::Other("aggregated count overflowed".into())),
406                _ => Ok(None),
407            })?;
408    let result = PriceBar {
409        exchange: first.exchange.clone(),
410        symbol: first.symbol.clone(),
411        timeframe: parent_timeframe,
412        ts: first.ts,
413        available_at: children.iter().map(|bar| bar.available_at).max().unwrap(),
414        open: first.open,
415        high: children
416            .iter()
417            .map(|bar| bar.high)
418            .fold(f64::NEG_INFINITY, f64::max),
419        low: children
420            .iter()
421            .map(|bar| bar.low)
422            .fold(f64::INFINITY, f64::min),
423        close: children.last().unwrap().close,
424        tick_count,
425        spread: None,
426    };
427    result.validate()?;
428    Ok(result)
429}
430
431/// Summary row returned by stats queries.
432#[derive(Debug)]
433pub struct StatRow {
434    pub exchange: String,
435    pub symbol: String,
436    pub data_type: String,
437    pub count: u64,
438    pub ts_min: NaiveDateTime,
439    pub ts_max: NaiveDateTime,
440}
441
442/// Result of an import operation.
443#[derive(Debug)]
444pub struct ImportResult {
445    pub file: String,
446    pub exchange: String,
447    pub symbol: String,
448    pub rows_parsed: usize,
449    pub rows_inserted: usize,
450    pub rows_skipped: usize,
451    pub elapsed: std::time::Duration,
452}
data_preprocess/models.rs

data_preprocess/
models.rs
