engine/data_source/
historical.rs

1use super::DataSource;
2use crate::types::{FillEvent, MarketState, Side};
3use parquet::file::reader::{FileReader, SerializedFileReader};
4use parquet::record::RowAccessor;
5use std::fs::File;
6use std::path::Path;
7
8pub struct ParquetDataSource {
9    reader: SerializedFileReader<File>,
10    current_row: usize,
11    total_rows: usize,
12    cumulative_impact: f64,
13    impact_factor: f64,
14}
15
16impl ParquetDataSource {
17    pub fn new<P: AsRef<Path>>(path: P, impact_factor: f64) -> Self {
18        let file = File::open(path).expect("Failed to open parquet file");
19        let reader = SerializedFileReader::new(file).expect("Failed to create parquet reader");
20        let total_rows = reader.metadata().file_metadata().num_rows() as usize;
21
22        Self {
23            reader,
24            current_row: 0,
25            total_rows,
26            cumulative_impact: 0.0,
27            impact_factor,
28        }
29    }
30}
31
32impl DataSource for ParquetDataSource {
33    fn next_quote(&mut self) -> Option<MarketState> {
34        if self.current_row >= self.total_rows {
35            return None;
36        }
37
38        // This is a simplified implementation.
39        let row_result = self.reader.get_row_iter(None).ok()?.nth(self.current_row)?;
40        let row = row_result.ok()?;
41        self.current_row += 1;
42
43        // Assume columns: timestamp (i64 micros), bid (double), ask (double)
44
45        let timestamp_val = row.get_long(0).ok()?; // Assuming col 0 is timestamp (micros)
46        let bid = row.get_double(1).ok()?; // Assuming col 1 is bid
47        let ask = row.get_double(2).ok()?; // Assuming col 2 is ask
48
49        let timestamp = timestamp_val as f64 / 1_000_000.0; // microseconds -> seconds
50
51        // Apply cumulative impact to the historical price
52        Some(MarketState {
53            timestamp,
54            best_bid: bid + self.cumulative_impact,
55            best_ask: ask + self.cumulative_impact,
56            last_price: None,
57            true_volatility: None,
58            true_drift: None,
59            parameters: None,
60        })
61    }
62
63    fn apply_impact(&mut self, fills: &[FillEvent]) {
64        if self.impact_factor.abs() < f64::EPSILON {
65            return;
66        }
67
68        let net_quantity: f64 = fills
69            .iter()
70            .map(|f| match f.side {
71                Side::Buy => f.quantity,
72                Side::Sell => -f.quantity,
73            })
74            .sum();
75
76        // Linear permanent impact: Price moves by lambda * quantity
77        self.cumulative_impact += self.impact_factor * net_quantity;
78    }
79}
engine/data_source/historical.rs

engine/data_source/
historical.rs
