nautilus_testkit/
cache.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//! Stateful cache database test double.
17
18use std::sync::Arc;
19
20use ahash::AHashMap;
21use bytes::Bytes;
22use indexmap::IndexMap;
23use nautilus_common::{
24    cache::database::{CacheDatabaseAdapter, CacheMap},
25    signal::Signal,
26};
27use nautilus_core::UnixNanos;
28use nautilus_model::{
29    accounts::AccountAny,
30    data::{
31        Bar, CustomData, DataType, FundingRateUpdate, InstrumentClose, QuoteTick, TradeTick,
32        greeks::{GreeksData, YieldCurve},
33    },
34    events::{OrderEventAny, OrderSnapshot, position::snapshot::PositionSnapshot},
35    identifiers::{
36        AccountId, ActorId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId,
37        VenueOrderId,
38    },
39    instruments::{InstrumentAny, SyntheticInstrument},
40    orderbook::OrderBook,
41    orders::OrderAny,
42    position::Position,
43    types::{Currency, Money},
44};
45use parking_lot::Mutex;
46use ustr::Ustr;
47
48#[expect(
49    clippy::struct_excessive_bools,
50    reason = "independent switches cover lifecycle persistence failure modes"
51)]
52#[derive(Debug, Default)]
53struct TestCacheDatabaseState {
54    actors: AHashMap<ActorId, AHashMap<String, Bytes>>,
55    strategies: AHashMap<StrategyId, AHashMap<String, Bytes>>,
56    instrument_closes: AHashMap<InstrumentId, InstrumentClose>,
57    events: Vec<String>,
58    fail_load_actor: bool,
59    fail_load_strategy: bool,
60    fail_update_actor: bool,
61    fail_update_strategy: bool,
62    fail_update_position: bool,
63    fail_update_position_id: Option<PositionId>,
64    updated_positions: Vec<Position>,
65}
66
67/// Shared control and observation handle for [`TestCacheDatabase`].
68#[derive(Clone, Debug, Default)]
69pub struct TestCacheDatabaseControl {
70    state: Arc<Mutex<TestCacheDatabaseState>>,
71}
72
73impl TestCacheDatabaseControl {
74    /// Creates an adapter and its shared control handle.
75    #[must_use]
76    pub fn create() -> (TestCacheDatabase, Self) {
77        let control = Self::default();
78        (
79            TestCacheDatabase {
80                control: control.clone(),
81            },
82            control,
83        )
84    }
85
86    /// Records an event in the shared lifecycle log.
87    pub fn record(&self, event: impl Into<String>) {
88        self.state.lock().events.push(event.into());
89    }
90
91    /// Returns the recorded lifecycle events.
92    #[must_use]
93    pub fn events(&self) -> Vec<String> {
94        self.state.lock().events.clone()
95    }
96
97    /// Seeds actor state for a later load.
98    pub fn set_actor_state(&self, actor_id: ActorId, state: &IndexMap<String, Vec<u8>>) {
99        self.state
100            .lock()
101            .actors
102            .insert(actor_id, encode_state(state));
103    }
104
105    /// Seeds strategy state for a later load.
106    pub fn set_strategy_state(&self, strategy_id: StrategyId, state: &IndexMap<String, Vec<u8>>) {
107        self.state
108            .lock()
109            .strategies
110            .insert(strategy_id, encode_state(state));
111    }
112
113    /// Returns persisted actor state.
114    #[must_use]
115    pub fn actor_state(&self, actor_id: &ActorId) -> Option<IndexMap<String, Vec<u8>>> {
116        self.state
117            .lock()
118            .actors
119            .get(actor_id)
120            .cloned()
121            .map(decode_state)
122    }
123
124    /// Returns persisted strategy state.
125    #[must_use]
126    pub fn strategy_state(&self, strategy_id: &StrategyId) -> Option<IndexMap<String, Vec<u8>>> {
127        self.state
128            .lock()
129            .strategies
130            .get(strategy_id)
131            .cloned()
132            .map(decode_state)
133    }
134
135    /// Configures actor loads to fail.
136    pub fn set_fail_load_actor(&self, fail: bool) {
137        self.state.lock().fail_load_actor = fail;
138    }
139
140    /// Configures strategy loads to fail.
141    pub fn set_fail_load_strategy(&self, fail: bool) {
142        self.state.lock().fail_load_strategy = fail;
143    }
144
145    /// Configures actor updates to fail.
146    pub fn set_fail_update_actor(&self, fail: bool) {
147        self.state.lock().fail_update_actor = fail;
148    }
149
150    /// Configures strategy updates to fail.
151    pub fn set_fail_update_strategy(&self, fail: bool) {
152        self.state.lock().fail_update_strategy = fail;
153    }
154
155    /// Configures position updates to fail.
156    pub fn set_fail_update_position(&self, fail: bool) {
157        self.state.lock().fail_update_position = fail;
158    }
159
160    /// Configures updates of a single position to fail.
161    pub fn set_fail_update_position_id(&self, position_id: Option<PositionId>) {
162        self.state.lock().fail_update_position_id = position_id;
163    }
164
165    /// Returns the positions persisted by successful updates, in update order.
166    #[must_use]
167    pub fn updated_positions(&self) -> Vec<Position> {
168        self.state.lock().updated_positions.clone()
169    }
170}
171
172/// Stateful cache database adapter for lifecycle tests.
173#[derive(Debug)]
174pub struct TestCacheDatabase {
175    control: TestCacheDatabaseControl,
176}
177
178#[async_trait::async_trait]
179impl CacheDatabaseAdapter for TestCacheDatabase {
180    fn close(&mut self) -> anyhow::Result<()> {
181        self.control.record("database.close");
182        Ok(())
183    }
184
185    fn flush(&mut self) -> anyhow::Result<()> {
186        Ok(())
187    }
188
189    async fn load_all(&self) -> anyhow::Result<CacheMap> {
190        Ok(CacheMap {
191            instrument_closes: self.control.state.lock().instrument_closes.clone(),
192            ..Default::default()
193        })
194    }
195
196    fn load(&self) -> anyhow::Result<AHashMap<String, Bytes>> {
197        Ok(AHashMap::new())
198    }
199
200    async fn load_currencies(&self) -> anyhow::Result<AHashMap<Ustr, Currency>> {
201        Ok(AHashMap::new())
202    }
203
204    async fn load_instruments(&self) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>> {
205        Ok(AHashMap::new())
206    }
207
208    async fn load_instrument_closes(
209        &self,
210    ) -> anyhow::Result<AHashMap<InstrumentId, InstrumentClose>> {
211        Ok(self.control.state.lock().instrument_closes.clone())
212    }
213
214    async fn load_synthetics(&self) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>> {
215        Ok(AHashMap::new())
216    }
217
218    async fn load_accounts(&self) -> anyhow::Result<AHashMap<AccountId, AccountAny>> {
219        Ok(AHashMap::new())
220    }
221
222    async fn load_orders(&self) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>> {
223        Ok(AHashMap::new())
224    }
225
226    async fn load_positions(&self) -> anyhow::Result<AHashMap<PositionId, Position>> {
227        Ok(AHashMap::new())
228    }
229
230    fn load_index_order_position(&self) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
231        Ok(AHashMap::new())
232    }
233
234    fn load_index_order_client(&self) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
235        Ok(AHashMap::new())
236    }
237
238    async fn load_currency(&self, _code: &Ustr) -> anyhow::Result<Option<Currency>> {
239        Ok(None)
240    }
241
242    async fn load_instrument(
243        &self,
244        _instrument_id: &InstrumentId,
245    ) -> anyhow::Result<Option<InstrumentAny>> {
246        Ok(None)
247    }
248
249    async fn load_synthetic(
250        &self,
251        _instrument_id: &InstrumentId,
252    ) -> anyhow::Result<Option<SyntheticInstrument>> {
253        Ok(None)
254    }
255
256    async fn load_account(&self, _account_id: &AccountId) -> anyhow::Result<Option<AccountAny>> {
257        Ok(None)
258    }
259
260    async fn load_order(
261        &self,
262        _client_order_id: &ClientOrderId,
263    ) -> anyhow::Result<Option<OrderAny>> {
264        Ok(None)
265    }
266
267    async fn load_position(&self, _position_id: &PositionId) -> anyhow::Result<Option<Position>> {
268        Ok(None)
269    }
270
271    fn load_actor(&self, actor_id: &ActorId) -> anyhow::Result<AHashMap<String, Bytes>> {
272        self.control.record(format!("actor.load:{actor_id}"));
273        let state = self.control.state.lock();
274        if state.fail_load_actor {
275            anyhow::bail!("test actor load failure");
276        }
277        Ok(state.actors.get(actor_id).cloned().unwrap_or_default())
278    }
279
280    fn load_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<AHashMap<String, Bytes>> {
281        self.control.record(format!("strategy.load:{strategy_id}"));
282        let state = self.control.state.lock();
283        if state.fail_load_strategy {
284            anyhow::bail!("test strategy load failure");
285        }
286        Ok(state
287            .strategies
288            .get(strategy_id)
289            .cloned()
290            .unwrap_or_default())
291    }
292
293    fn load_signals(&self, _name: &str) -> anyhow::Result<Vec<Signal>> {
294        Ok(Vec::new())
295    }
296
297    fn load_custom_data(&self, _data_type: &DataType) -> anyhow::Result<Vec<CustomData>> {
298        Ok(Vec::new())
299    }
300
301    fn load_order_snapshot(
302        &self,
303        _client_order_id: &ClientOrderId,
304    ) -> anyhow::Result<Option<OrderSnapshot>> {
305        Ok(None)
306    }
307
308    fn load_position_snapshot(
309        &self,
310        _position_id: &PositionId,
311    ) -> anyhow::Result<Option<PositionSnapshot>> {
312        Ok(None)
313    }
314
315    fn load_quotes(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<QuoteTick>> {
316        Ok(Vec::new())
317    }
318
319    fn load_trades(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<TradeTick>> {
320        Ok(Vec::new())
321    }
322
323    fn load_funding_rates(
324        &self,
325        _instrument_id: &InstrumentId,
326    ) -> anyhow::Result<Vec<FundingRateUpdate>> {
327        Ok(Vec::new())
328    }
329
330    fn load_bars(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<Bar>> {
331        Ok(Vec::new())
332    }
333
334    fn add(&self, _key: String, _value: Bytes) -> anyhow::Result<()> {
335        Ok(())
336    }
337
338    fn add_currency(&self, _currency: &Currency) -> anyhow::Result<()> {
339        Ok(())
340    }
341
342    fn add_instrument(&self, _instrument: &InstrumentAny) -> anyhow::Result<()> {
343        Ok(())
344    }
345
346    fn add_instrument_close(&self, close: &InstrumentClose) -> anyhow::Result<()> {
347        let mut state = self.control.state.lock();
348        state.instrument_closes.insert(close.instrument_id, *close);
349        Ok(())
350    }
351
352    fn add_synthetic(&self, _synthetic: &SyntheticInstrument) -> anyhow::Result<()> {
353        Ok(())
354    }
355
356    fn add_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
357        Ok(())
358    }
359
360    fn add_order(&self, _order: &OrderAny, _client_id: Option<ClientId>) -> anyhow::Result<()> {
361        Ok(())
362    }
363
364    fn add_order_snapshot(&self, _snapshot: &OrderSnapshot) -> anyhow::Result<()> {
365        Ok(())
366    }
367
368    fn add_position(&self, _position: &Position) -> anyhow::Result<()> {
369        Ok(())
370    }
371
372    fn add_position_snapshot(&self, _snapshot: &PositionSnapshot) -> anyhow::Result<()> {
373        Ok(())
374    }
375
376    fn add_order_book(&self, _order_book: &OrderBook) -> anyhow::Result<()> {
377        Ok(())
378    }
379
380    fn add_signal(&self, _signal: &Signal) -> anyhow::Result<()> {
381        Ok(())
382    }
383
384    fn add_custom_data(&self, _data: &CustomData) -> anyhow::Result<()> {
385        Ok(())
386    }
387
388    fn add_quote(&self, _quote: &QuoteTick) -> anyhow::Result<()> {
389        Ok(())
390    }
391
392    fn add_trade(&self, _trade: &TradeTick) -> anyhow::Result<()> {
393        Ok(())
394    }
395
396    fn add_funding_rate(&self, _funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
397        Ok(())
398    }
399
400    fn add_bar(&self, _bar: &Bar) -> anyhow::Result<()> {
401        Ok(())
402    }
403
404    fn add_greeks(&self, _greeks: &GreeksData) -> anyhow::Result<()> {
405        Ok(())
406    }
407
408    fn add_yield_curve(&self, _yield_curve: &YieldCurve) -> anyhow::Result<()> {
409        Ok(())
410    }
411
412    fn delete_actor(&self, _actor_id: &ActorId) -> anyhow::Result<()> {
413        Ok(())
414    }
415
416    fn delete_strategy(&self, _component_id: &StrategyId) -> anyhow::Result<()> {
417        Ok(())
418    }
419
420    fn delete_order(&self, _client_order_id: &ClientOrderId) -> anyhow::Result<()> {
421        Ok(())
422    }
423
424    fn delete_position(&self, _position_id: &PositionId) -> anyhow::Result<()> {
425        Ok(())
426    }
427
428    fn delete_account_event(&self, _account_id: &AccountId, _event_id: &str) -> anyhow::Result<()> {
429        Ok(())
430    }
431
432    fn index_venue_order_id(
433        &self,
434        _client_order_id: ClientOrderId,
435        _venue_order_id: VenueOrderId,
436    ) -> anyhow::Result<()> {
437        Ok(())
438    }
439
440    fn index_order_position(
441        &self,
442        _client_order_id: ClientOrderId,
443        _position_id: PositionId,
444    ) -> anyhow::Result<()> {
445        Ok(())
446    }
447
448    fn update_actor(
449        &self,
450        actor_id: &ActorId,
451        actor_state: &AHashMap<String, Bytes>,
452    ) -> anyhow::Result<()> {
453        self.control.record(format!("actor.update:{actor_id}"));
454        let mut state = self.control.state.lock();
455        if state.fail_update_actor {
456            anyhow::bail!("test actor update failure");
457        }
458        state.actors.insert(*actor_id, actor_state.clone());
459        Ok(())
460    }
461
462    fn update_strategy(
463        &self,
464        strategy_id: &StrategyId,
465        strategy_state: &AHashMap<String, Bytes>,
466    ) -> anyhow::Result<()> {
467        self.control
468            .record(format!("strategy.update:{strategy_id}"));
469        let mut state = self.control.state.lock();
470        if state.fail_update_strategy {
471            anyhow::bail!("test strategy update failure");
472        }
473        state
474            .strategies
475            .insert(*strategy_id, strategy_state.clone());
476        Ok(())
477    }
478
479    fn update_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
480        Ok(())
481    }
482
483    fn update_order(&self, _order_event: &OrderEventAny) -> anyhow::Result<()> {
484        Ok(())
485    }
486
487    fn update_position(&self, position: &Position) -> anyhow::Result<()> {
488        let mut state = self.control.state.lock();
489        if state.fail_update_position || state.fail_update_position_id == Some(position.id) {
490            anyhow::bail!("test position update failure");
491        }
492        state.updated_positions.push(position.clone());
493        Ok(())
494    }
495
496    fn snapshot_order_state(&self, _order: &OrderAny) -> anyhow::Result<()> {
497        Ok(())
498    }
499
500    fn snapshot_position_state(
501        &self,
502        _position: &Position,
503        _ts_snapshot: UnixNanos,
504        _unrealized_pnl: Option<Money>,
505    ) -> anyhow::Result<()> {
506        Ok(())
507    }
508
509    fn heartbeat(&self, _timestamp: UnixNanos) -> anyhow::Result<()> {
510        Ok(())
511    }
512}
513
514fn decode_state(state: AHashMap<String, Bytes>) -> IndexMap<String, Vec<u8>> {
515    state
516        .into_iter()
517        .map(|(key, value)| (key, value.to_vec()))
518        .collect()
519}
520
521fn encode_state(state: &IndexMap<String, Vec<u8>>) -> AHashMap<String, Bytes> {
522    state
523        .iter()
524        .map(|(key, value)| (key.clone(), Bytes::copy_from_slice(value)))
525        .collect()
526}
527
528#[cfg(test)]
529mod tests {
530    use nautilus_model::{data::InstrumentClose, enums::InstrumentCloseType, types::Price};
531    use rstest::rstest;
532
533    use super::*;
534
535    #[rstest]
536    #[tokio::test]
537    async fn test_instrument_close_persistence_stores_latest_value() {
538        let instrument_id = InstrumentId::from("BINARY-1.POLYMARKET");
539        let first = InstrumentClose::new(
540            instrument_id,
541            Price::from("1.00000"),
542            InstrumentCloseType::ContractExpired,
543            UnixNanos::from(10),
544            UnixNanos::from(11),
545        );
546        let replacement = InstrumentClose::new(
547            instrument_id,
548            Price::from("0.00000"),
549            InstrumentCloseType::EndOfSession,
550            UnixNanos::from(20),
551            UnixNanos::from(21),
552        );
553        let (database, _) = TestCacheDatabaseControl::create();
554
555        database.add_instrument_close(&first).unwrap();
556        database.add_instrument_close(&replacement).unwrap();
557        let loaded = database.load_instrument_closes().await.unwrap();
558        let loaded_all = database.load_all().await.unwrap();
559
560        assert_eq!(loaded, AHashMap::from([(instrument_id, replacement)]));
561        assert_eq!(loaded_all.instrument_closes, loaded);
562    }
563}
nautilus_testkit/cache.rs

nautilus_testkit/
cache.rs
