engine/data_source/
simulated.rs

1use super::DataSource;
2use crate::types::{FillEvent, MarketState, Side};
3use market_model::process::{
4    StochasticProcess, bates::BatesProcess, gbm::GeometricBrownianMotion, heston::HestonProcess,
5};
6use ndarray::Array1;
7use rand::prelude::*;
8use rand::rng;
9use rand::rngs::StdRng;
10use std::collections::HashMap;
11
12pub trait MarketProcess: StochasticProcess {
13    fn get_price(&self, state: &Self::State) -> f64;
14    fn get_volatility(&self, _state: &Self::State) -> Option<f64> {
15        None
16    }
17    fn get_drift(&self, _state: &Self::State) -> Option<f64> {
18        None
19    }
20    fn get_parameters(&self) -> Option<HashMap<String, f64>> {
21        None
22    }
23    fn add_price_impact(&self, state: &mut Self::State, impact: f64);
24}
25
26impl MarketProcess for GeometricBrownianMotion {
27    fn get_price(&self, state: &f64) -> f64 {
28        *state
29    }
30
31    fn get_volatility(&self, _state: &f64) -> Option<f64> {
32        Some(self.sigma)
33    }
34
35    fn get_drift(&self, _state: &f64) -> Option<f64> {
36        Some(self.mu)
37    }
38
39    fn get_parameters(&self) -> Option<HashMap<String, f64>> {
40        let mut p = HashMap::new();
41        p.insert("mu".to_string(), self.mu);
42        p.insert("sigma".to_string(), self.sigma);
43        Some(p)
44    }
45
46    fn add_price_impact(&self, state: &mut f64, impact: f64) {
47        *state += impact;
48    }
49}
50
51impl MarketProcess for HestonProcess {
52    fn get_price(&self, state: &Array1<f64>) -> f64 {
53        state[0]
54    }
55
56    fn get_volatility(&self, state: &Array1<f64>) -> Option<f64> {
57        Some(state[1].max(0.0).sqrt())
58    }
59
60    fn get_drift(&self, _state: &Array1<f64>) -> Option<f64> {
61        Some(self.mu)
62    }
63
64    fn get_parameters(&self) -> Option<HashMap<String, f64>> {
65        let mut p = HashMap::new();
66        p.insert("mu".to_string(), self.mu);
67        p.insert("kappa".to_string(), self.kappa);
68        p.insert("theta".to_string(), self.theta);
69        p.insert("sigma".to_string(), self.sigma);
70        p.insert("rho".to_string(), self.rho);
71        Some(p)
72    }
73
74    fn add_price_impact(&self, state: &mut Array1<f64>, impact: f64) {
75        state[0] += impact;
76    }
77}
78
79impl MarketProcess for BatesProcess {
80    fn get_price(&self, state: &Array1<f64>) -> f64 {
81        state[0]
82    }
83
84    fn get_volatility(&self, state: &Array1<f64>) -> Option<f64> {
85        Some(state[1].max(0.0).sqrt())
86    }
87
88    fn get_drift(&self, _state: &Array1<f64>) -> Option<f64> {
89        Some(self.mu())
90    }
91
92    fn get_parameters(&self) -> Option<HashMap<String, f64>> {
93        let mut p = HashMap::new();
94        p.insert("mu".to_string(), self.mu());
95        p.insert("kappa".to_string(), self.kappa());
96        p.insert("theta".to_string(), self.theta());
97        p.insert("sigma".to_string(), self.sigma());
98        p.insert("rho".to_string(), self.rho());
99        p.insert("lambda".to_string(), self.lambda);
100        p.insert("mu_jump".to_string(), self.mu_jump);
101        p.insert("sigma_jump".to_string(), self.sigma_jump);
102        Some(p)
103    }
104
105    fn add_price_impact(&self, state: &mut Array1<f64>, impact: f64) {
106        state[0] += impact;
107    }
108}
109
110pub struct SimulatedDataSource<P: MarketProcess> {
111    process: P,
112    current_state: P::State,
113    current_time: f64,
114    dt: f64,
115    spread: f64,
116    max_steps: usize,
117    current_step: usize,
118    rng: StdRng,
119    impact_factor: f64,
120    #[allow(clippy::type_complexity)]
121    precalculated_data: Option<Vec<(f64, Option<f64>, Option<f64>)>>,
122    cached_parameters: Option<HashMap<String, f64>>,
123}
124
125impl<P: MarketProcess> SimulatedDataSource<P>
126where
127    P::State: Clone,
128{
129    pub fn new(
130        process: P,
131        initial_state: P::State,
132        dt: f64,
133        spread: f64,
134        max_steps: usize,
135        impact_factor: f64,
136    ) -> Self {
137        let mut rng = StdRng::from_rng(&mut rng());
138        let cached_parameters = process.get_parameters();
139        let precalculated_data = if impact_factor.abs() < f64::EPSILON {
140            // If no market impact, we can pre-calculate the entire price trajectory
141            // This is faster as it avoids repeated function calls and allows for potential vectorization
142            let mut data = Vec::with_capacity(max_steps);
143            let mut state = initial_state.clone();
144            let mut time = 0.0;
145
146            for _ in 0..max_steps {
147                let dw = process.generate_increment(&mut rng, dt);
148                state = process.step(&state, time, dt, &dw, &mut rng);
149                time += dt;
150                data.push((
151                    process.get_price(&state),
152                    process.get_volatility(&state),
153                    process.get_drift(&state),
154                ));
155            }
156            Some(data)
157        } else {
158            None
159        };
160
161        Self {
162            process,
163            current_state: initial_state,
164            current_time: 0.0,
165            dt,
166            spread,
167            max_steps,
168            current_step: 0,
169            rng,
170            impact_factor,
171            precalculated_data,
172            cached_parameters,
173        }
174    }
175
176    /// Seed this data source with an explicit RNG (for reproducible, Send-safe copies).
177    pub fn with_rng(mut self, rng: StdRng) -> Self {
178        self.rng = rng;
179        self
180    }
181}
182
183impl<P: MarketProcess> DataSource for SimulatedDataSource<P> {
184    fn next_quote(&mut self) -> Option<MarketState> {
185        if self.current_step >= self.max_steps {
186            return None;
187        }
188
189        // Get price and volatility
190        let (price, vol, drift) = if let Some(data) = &self.precalculated_data {
191            let (p, v, d) = data[self.current_step];
192            self.current_step += 1;
193            self.current_time += self.dt;
194            (p, v, d)
195        } else {
196            // Calculate on the fly (needed for market impact)
197            self.current_step += 1;
198
199            // Generate Brownian increment
200            let dw = self.process.generate_increment(&mut self.rng, self.dt);
201
202            // Step the process
203            self.current_state = self.process.step(
204                &self.current_state,
205                self.current_time,
206                self.dt,
207                &dw,
208                &mut self.rng,
209            );
210
211            let drift = self.process.get_drift(&self.current_state);
212            let vol = self.process.get_volatility(&self.current_state);
213            let price = self.process.get_price(&self.current_state);
214
215            self.current_time += self.dt;
216            (price, vol, drift)
217        };
218
219        // Timestamp is elapsed time in seconds (normalized): current_step * dt.
220        // Strategies use obs.timestamp as a plain number to compute tau = T - elapsed.
221        let timestamp = self.current_step as f64 * self.dt;
222
223        Some(MarketState {
224            timestamp,
225            best_bid: price - self.spread / 2.0,
226            best_ask: price + self.spread / 2.0,
227            last_price: Some(price),
228            true_volatility: vol,
229            true_drift: drift,
230            parameters: self.cached_parameters.clone(),
231        })
232    }
233
234    fn apply_impact(&mut self, fills: &[FillEvent]) {
235        if self.impact_factor.abs() < f64::EPSILON {
236            return;
237        }
238
239        let net_quantity: f64 = fills
240            .iter()
241            .map(|f| match f.side {
242                Side::Buy => f.quantity,
243                Side::Sell => -f.quantity,
244            })
245            .sum();
246
247        // Adjust the current price state of the simulation
248        // This will affect the starting point for the next stochastic step
249        let impact = self.impact_factor * net_quantity;
250        self.process
251            .add_price_impact(&mut self.current_state, impact);
252    }
253}
engine/data_source/simulated.rs

engine/data_source/
simulated.rs
