nautilus_common/cache/
database.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//! Provides a `Cache` database backing.
17
18use std::fmt::Debug;
19
20use ahash::AHashMap;
21use bytes::Bytes;
22use nautilus_core::{UUID4, UnixNanos};
23use nautilus_model::{
24    accounts::AccountAny,
25    data::{
26        Bar, CustomData, DataType, FundingRateUpdate, InstrumentClose, QuoteTick, TradeTick,
27        greeks::{GreeksData, YieldCurve},
28    },
29    events::{OrderEventAny, OrderSnapshot, position::snapshot::PositionSnapshot},
30    identifiers::{
31        AccountId, ActorId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId,
32        TraderId, VenueOrderId,
33    },
34    instruments::{InstrumentAny, SyntheticInstrument},
35    orderbook::OrderBook,
36    orders::OrderAny,
37    position::Position,
38    types::{Currency, Money},
39};
40use ustr::Ustr;
41
42use super::config::CacheConfig;
43use crate::signal::Signal;
44
45/// Registers currencies loaded from a cache database, keeping the registry's definition.
46///
47/// A `Money` or `Currency` decoded afterwards resolves its code through the global registry, so a
48/// loaded currency must reach it before the dependent payloads are read. `Currency::register` does
49/// not overwrite, so a built-in constant or a currency an adapter registered first keeps its
50/// definition; the map takes that definition too, so the cache and the decoded payloads agree, and a
51/// stored record that differs from it is logged.
52///
53/// # Errors
54///
55/// Returns an error if `Currency::register` does; it does not at present, so the result only
56/// keeps callers forward-compatible.
57pub fn register_loaded_currencies(currencies: &mut AHashMap<Ustr, Currency>) -> anyhow::Result<()> {
58    for (code, currency) in currencies.iter_mut() {
59        Currency::register(*currency, false)?;
60
61        let Some(registered) = Currency::try_from_str(code) else {
62            continue;
63        };
64        let mut differences: Vec<String> = Vec::new();
65        if registered.precision != currency.precision {
66            differences.push(format!(
67                "precision {} vs registered {}",
68                currency.precision, registered.precision
69            ));
70        }
71
72        if registered.iso4217 != currency.iso4217 {
73            differences.push(format!(
74                "iso4217 {} vs registered {}",
75                currency.iso4217, registered.iso4217
76            ));
77        }
78
79        if registered.name != currency.name {
80            differences.push(format!(
81                "name {} vs registered {}",
82                currency.name, registered.name
83            ));
84        }
85
86        if registered.currency_type != currency.currency_type {
87            differences.push(format!(
88                "type {:?} vs registered {:?}",
89                currency.currency_type, registered.currency_type
90            ));
91        }
92
93        if !differences.is_empty() {
94            log::warn!(
95                "Stored currency {code} differs from the registered definition ({}); using the registered one",
96                differences.join(", "),
97            );
98            *currency = registered;
99        }
100    }
101
102    Ok(())
103}
104
105#[derive(Debug, Default)]
106pub struct CacheMap {
107    pub currencies: AHashMap<Ustr, Currency>,
108    pub instruments: AHashMap<InstrumentId, InstrumentAny>,
109    pub instrument_closes: AHashMap<InstrumentId, InstrumentClose>,
110    pub synthetics: AHashMap<InstrumentId, SyntheticInstrument>,
111    pub accounts: AHashMap<AccountId, AccountAny>,
112    pub orders: AHashMap<ClientOrderId, OrderAny>,
113    pub positions: AHashMap<PositionId, Position>,
114    pub greeks: AHashMap<InstrumentId, GreeksData>,
115    pub yield_curves: AHashMap<String, YieldCurve>,
116}
117
118/// Factory for constructing cache database adapters at runtime.
119///
120/// Implementations own the concrete database configuration and return the transport-neutral
121/// [`CacheDatabaseAdapter`] surface used by the cache.
122#[async_trait::async_trait]
123pub trait CacheDatabaseFactory: Debug + Send + Sync {
124    /// Creates a cache database adapter for the given cache runtime.
125    ///
126    /// # Errors
127    ///
128    /// Returns an error if adapter construction or connection setup fails.
129    async fn create(
130        &self,
131        trader_id: TraderId,
132        instance_id: UUID4,
133        config: CacheConfig,
134    ) -> anyhow::Result<Box<dyn CacheDatabaseAdapter>>;
135}
136
137#[async_trait::async_trait]
138pub trait CacheDatabaseAdapter {
139    /// Closes the cache database connection.
140    ///
141    /// # Errors
142    ///
143    /// Returns an error if the database fails to close properly.
144    fn close(&mut self) -> anyhow::Result<()>;
145
146    /// Flushes any pending changes to the database.
147    ///
148    /// # Errors
149    ///
150    /// Returns an error if flushing changes fails.
151    fn flush(&mut self) -> anyhow::Result<()>;
152
153    /// Loads all cached data into memory.
154    ///
155    /// # Errors
156    ///
157    /// Returns an error if loading data from the database fails.
158    async fn load_all(&self) -> anyhow::Result<CacheMap>;
159
160    /// Loads raw key-value data from the database.
161    ///
162    /// # Errors
163    ///
164    /// Returns an error if the load operation fails.
165    fn load(&self) -> anyhow::Result<AHashMap<String, Bytes>>;
166
167    /// Loads all currencies from the cache.
168    ///
169    /// # Errors
170    ///
171    /// Returns an error if loading currencies fails.
172    async fn load_currencies(&self) -> anyhow::Result<AHashMap<Ustr, Currency>>;
173
174    /// Loads all instruments from the cache.
175    ///
176    /// # Errors
177    ///
178    /// Returns an error if loading instruments fails.
179    async fn load_instruments(&self) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>>;
180
181    /// Loads all instrument closes from the cache.
182    ///
183    /// # Errors
184    ///
185    /// Returns an error if loading instrument closes fails.
186    async fn load_instrument_closes(
187        &self,
188    ) -> anyhow::Result<AHashMap<InstrumentId, InstrumentClose>>;
189
190    /// Loads all synthetic instruments from the cache.
191    ///
192    /// # Errors
193    ///
194    /// Returns an error if loading synthetic instruments fails.
195    async fn load_synthetics(&self) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>>;
196
197    /// Loads all accounts from the cache.
198    ///
199    /// # Errors
200    ///
201    /// Returns an error if loading accounts fails.
202    async fn load_accounts(&self) -> anyhow::Result<AHashMap<AccountId, AccountAny>>;
203
204    /// Loads all orders from the cache.
205    ///
206    /// # Errors
207    ///
208    /// Returns an error if loading orders fails.
209    async fn load_orders(&self) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>>;
210
211    /// Loads all positions from the cache.
212    ///
213    /// # Errors
214    ///
215    /// Returns an error if loading positions fails.
216    async fn load_positions(&self) -> anyhow::Result<AHashMap<PositionId, Position>>;
217
218    /// Loads all [`GreeksData`] from the cache.
219    ///
220    /// # Errors
221    ///
222    /// Returns an error if loading greeks data fails.
223    async fn load_greeks(&self) -> anyhow::Result<AHashMap<InstrumentId, GreeksData>> {
224        Ok(AHashMap::new())
225    }
226
227    /// Loads all [`YieldCurve`] from the cache.
228    ///
229    /// # Errors
230    ///
231    /// Returns an error if loading yield curve data fails.
232    async fn load_yield_curves(&self) -> anyhow::Result<AHashMap<String, YieldCurve>> {
233        Ok(AHashMap::new())
234    }
235
236    /// Loads mapping from order IDs to position IDs.
237    ///
238    /// # Errors
239    ///
240    /// Returns an error if loading the index order-position mapping fails.
241    fn load_index_order_position(&self) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>>;
242
243    /// Loads mapping from order IDs to client IDs.
244    ///
245    /// # Errors
246    ///
247    /// Returns an error if loading the index order-client mapping fails.
248    fn load_index_order_client(&self) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>>;
249
250    /// Loads a single currency by code.
251    ///
252    /// # Errors
253    ///
254    /// Returns an error if loading a single currency fails.
255    async fn load_currency(&self, code: &Ustr) -> anyhow::Result<Option<Currency>>;
256
257    /// Loads a single instrument by ID.
258    ///
259    /// # Errors
260    ///
261    /// Returns an error if loading a single instrument fails.
262    async fn load_instrument(
263        &self,
264        instrument_id: &InstrumentId,
265    ) -> anyhow::Result<Option<InstrumentAny>>;
266
267    /// Loads a single synthetic instrument by ID.
268    ///
269    /// # Errors
270    ///
271    /// Returns an error if loading a single synthetic instrument fails.
272    async fn load_synthetic(
273        &self,
274        instrument_id: &InstrumentId,
275    ) -> anyhow::Result<Option<SyntheticInstrument>>;
276
277    /// Loads a single account by ID.
278    ///
279    /// # Errors
280    ///
281    /// Returns an error if loading a single account fails.
282    async fn load_account(&self, account_id: &AccountId) -> anyhow::Result<Option<AccountAny>>;
283
284    /// Loads a single order by client order ID.
285    ///
286    /// # Errors
287    ///
288    /// Returns an error if loading a single order fails.
289    async fn load_order(&self, client_order_id: &ClientOrderId)
290    -> anyhow::Result<Option<OrderAny>>;
291
292    /// Loads a single position by position ID.
293    ///
294    /// # Errors
295    ///
296    /// Returns an error if loading a single position fails.
297    async fn load_position(&self, position_id: &PositionId) -> anyhow::Result<Option<Position>>;
298
299    /// Loads actor state by actor ID.
300    ///
301    /// # Errors
302    ///
303    /// Returns an error if loading actor state fails.
304    fn load_actor(&self, actor_id: &ActorId) -> anyhow::Result<AHashMap<String, Bytes>>;
305
306    /// Loads strategy state by strategy ID.
307    ///
308    /// # Errors
309    ///
310    /// Returns an error if loading strategy state fails.
311    fn load_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<AHashMap<String, Bytes>>;
312
313    /// Loads signals by name.
314    ///
315    /// # Errors
316    ///
317    /// Returns an error if loading signals fails.
318    fn load_signals(&self, name: &str) -> anyhow::Result<Vec<Signal>>;
319
320    /// Loads custom data by data type.
321    ///
322    /// # Errors
323    ///
324    /// Returns an error if loading custom data fails.
325    fn load_custom_data(&self, data_type: &DataType) -> anyhow::Result<Vec<CustomData>>;
326
327    /// Loads an order snapshot by client order ID.
328    ///
329    /// # Errors
330    ///
331    /// Returns an error if loading the order snapshot fails.
332    fn load_order_snapshot(
333        &self,
334        client_order_id: &ClientOrderId,
335    ) -> anyhow::Result<Option<OrderSnapshot>>;
336
337    /// Loads a position snapshot by position ID.
338    ///
339    /// # Errors
340    ///
341    /// Returns an error if loading the position snapshot fails.
342    fn load_position_snapshot(
343        &self,
344        position_id: &PositionId,
345    ) -> anyhow::Result<Option<PositionSnapshot>>;
346
347    /// Loads quote ticks by instrument ID.
348    ///
349    /// # Errors
350    ///
351    /// Returns an error if loading quotes fails.
352    fn load_quotes(&self, instrument_id: &InstrumentId) -> anyhow::Result<Vec<QuoteTick>>;
353
354    /// Loads trade ticks by instrument ID.
355    ///
356    /// # Errors
357    ///
358    /// Returns an error if loading trades fails.
359    fn load_trades(&self, instrument_id: &InstrumentId) -> anyhow::Result<Vec<TradeTick>>;
360
361    /// Loads funding rate updates by instrument ID.
362    ///
363    /// # Errors
364    ///
365    /// Returns an error if loading funding rates fails.
366    fn load_funding_rates(
367        &self,
368        instrument_id: &InstrumentId,
369    ) -> anyhow::Result<Vec<FundingRateUpdate>>;
370
371    /// Loads bars by instrument ID.
372    ///
373    /// # Errors
374    ///
375    /// Returns an error if loading bars fails.
376    fn load_bars(&self, instrument_id: &InstrumentId) -> anyhow::Result<Vec<Bar>>;
377
378    /// Adds a generic key-value pair to the cache.
379    ///
380    /// # Errors
381    ///
382    /// Returns an error if adding a generic key/value fails.
383    fn add(&self, key: String, value: Bytes) -> anyhow::Result<()>;
384
385    /// Adds a currency to the cache.
386    ///
387    /// # Errors
388    ///
389    /// Returns an error if adding a currency fails.
390    fn add_currency(&self, currency: &Currency) -> anyhow::Result<()>;
391
392    /// Adds an instrument to the cache.
393    ///
394    /// # Errors
395    ///
396    /// Returns an error if adding an instrument fails.
397    fn add_instrument(&self, instrument: &InstrumentAny) -> anyhow::Result<()>;
398
399    /// Adds an instrument close to the cache, replacing any existing value for the instrument.
400    /// Implementations must queue persistence without waiting for the database operation to
401    /// complete.
402    ///
403    /// # Errors
404    ///
405    /// Returns an error if the instrument close cannot be queued for persistence.
406    fn add_instrument_close(&self, close: &InstrumentClose) -> anyhow::Result<()>;
407
408    /// Adds a synthetic instrument to the cache.
409    ///
410    /// # Errors
411    ///
412    /// Returns an error if adding a synthetic instrument fails.
413    fn add_synthetic(&self, synthetic: &SyntheticInstrument) -> anyhow::Result<()>;
414
415    /// Adds an account to the cache.
416    ///
417    /// # Errors
418    ///
419    /// Returns an error if adding an account fails.
420    fn add_account(&self, account: &AccountAny) -> anyhow::Result<()>;
421
422    /// Adds an order to the cache.
423    ///
424    /// # Errors
425    ///
426    /// Returns an error if adding an order fails.
427    fn add_order(&self, order: &OrderAny, client_id: Option<ClientId>) -> anyhow::Result<()>;
428
429    /// Adds an order snapshot to the cache.
430    ///
431    /// # Errors
432    ///
433    /// Returns an error if adding an order snapshot fails.
434    fn add_order_snapshot(&self, snapshot: &OrderSnapshot) -> anyhow::Result<()>;
435
436    /// Adds a position to the cache.
437    ///
438    /// # Errors
439    ///
440    /// Returns an error if adding a position fails.
441    fn add_position(&self, position: &Position) -> anyhow::Result<()>;
442
443    /// Adds a position snapshot to the cache.
444    ///
445    /// # Errors
446    ///
447    /// Returns an error if adding a position snapshot fails.
448    fn add_position_snapshot(&self, snapshot: &PositionSnapshot) -> anyhow::Result<()>;
449
450    /// Adds an order book to the cache.
451    ///
452    /// # Errors
453    ///
454    /// Returns an error if adding an order book fails.
455    fn add_order_book(&self, order_book: &OrderBook) -> anyhow::Result<()>;
456
457    /// Adds a signal to the cache.
458    ///
459    /// # Errors
460    ///
461    /// Returns an error if adding a signal fails.
462    fn add_signal(&self, signal: &Signal) -> anyhow::Result<()>;
463
464    /// Adds custom data to the cache.
465    ///
466    /// # Errors
467    ///
468    /// Returns an error if adding custom data fails.
469    fn add_custom_data(&self, data: &CustomData) -> anyhow::Result<()>;
470
471    /// Adds a quote tick to the cache.
472    ///
473    /// # Errors
474    ///
475    /// Returns an error if adding a quote tick fails.
476    fn add_quote(&self, quote: &QuoteTick) -> anyhow::Result<()>;
477
478    /// Adds a trade tick to the cache.
479    ///
480    /// # Errors
481    ///
482    /// Returns an error if adding a trade tick fails.
483    fn add_trade(&self, trade: &TradeTick) -> anyhow::Result<()>;
484
485    /// Adds a funding rate update to the cache.
486    ///
487    /// # Errors
488    ///
489    /// Returns an error if adding a funding rate update fails.
490    fn add_funding_rate(&self, funding_rate: &FundingRateUpdate) -> anyhow::Result<()>;
491
492    /// Adds a bar to the cache.
493    ///
494    /// # Errors
495    ///
496    /// Returns an error if adding a bar fails.
497    fn add_bar(&self, bar: &Bar) -> anyhow::Result<()>;
498
499    /// Adds greeks data to the cache.
500    ///
501    /// # Errors
502    ///
503    /// Returns an error if adding greeks data fails.
504    fn add_greeks(&self, _greeks: &GreeksData) -> anyhow::Result<()> {
505        Ok(())
506    }
507
508    /// Adds yield curve data to the cache.
509    ///
510    /// # Errors
511    ///
512    /// Returns an error if adding yield curve data fails.
513    fn add_yield_curve(&self, _yield_curve: &YieldCurve) -> anyhow::Result<()> {
514        Ok(())
515    }
516
517    /// Deletes actor state from the cache.
518    ///
519    /// # Errors
520    ///
521    /// Returns an error if deleting actor state fails.
522    fn delete_actor(&self, actor_id: &ActorId) -> anyhow::Result<()>;
523
524    /// Deletes strategy state from the cache.
525    ///
526    /// # Errors
527    ///
528    /// Returns an error if deleting strategy state fails.
529    fn delete_strategy(&self, component_id: &StrategyId) -> anyhow::Result<()>;
530
531    /// Deletes an order from the cache.
532    ///
533    /// # Errors
534    ///
535    /// Returns an error if deleting an order fails.
536    fn delete_order(&self, client_order_id: &ClientOrderId) -> anyhow::Result<()>;
537
538    /// Deletes a position from the cache.
539    ///
540    /// # Errors
541    ///
542    /// Returns an error if deleting a position fails.
543    fn delete_position(&self, position_id: &PositionId) -> anyhow::Result<()>;
544
545    /// Deletes an account event from the cache.
546    ///
547    /// # Errors
548    ///
549    /// Returns an error if deleting account events fails.
550    fn delete_account_event(&self, account_id: &AccountId, event_id: &str) -> anyhow::Result<()>;
551
552    /// Indexes a venue order ID with its client order ID.
553    ///
554    /// # Errors
555    ///
556    /// Returns an error if indexing venue order ID fails.
557    fn index_venue_order_id(
558        &self,
559        client_order_id: ClientOrderId,
560        venue_order_id: VenueOrderId,
561    ) -> anyhow::Result<()>;
562
563    /// Indexes an order-position mapping.
564    ///
565    /// # Errors
566    ///
567    /// Returns an error if indexing order-position mapping fails.
568    fn index_order_position(
569        &self,
570        client_order_id: ClientOrderId,
571        position_id: PositionId,
572    ) -> anyhow::Result<()>;
573
574    /// Indexes order-client mappings as one batch operation.
575    ///
576    /// # Errors
577    ///
578    /// Returns an error if batch order-client indexing is unsupported or cannot be enqueued.
579    fn index_order_clients(&self, claims: &[(ClientOrderId, ClientId)]) -> anyhow::Result<()> {
580        if claims.is_empty() {
581            return Ok(());
582        }
583
584        anyhow::bail!("Batch order-client indexing is not supported by this cache database")
585    }
586
587    /// Updates actor state in the cache.
588    ///
589    /// # Errors
590    ///
591    /// Returns an error if updating actor state fails.
592    fn update_actor(
593        &self,
594        actor_id: &ActorId,
595        state: &AHashMap<String, Bytes>,
596    ) -> anyhow::Result<()>;
597
598    /// Updates strategy state in the cache.
599    ///
600    /// # Errors
601    ///
602    /// Returns an error if updating strategy state fails.
603    fn update_strategy(
604        &self,
605        strategy_id: &StrategyId,
606        state: &AHashMap<String, Bytes>,
607    ) -> anyhow::Result<()>;
608
609    /// Updates an account in the cache.
610    ///
611    /// # Errors
612    ///
613    /// Returns an error if updating an account fails.
614    fn update_account(&self, account: &AccountAny) -> anyhow::Result<()>;
615
616    /// Updates an order in the cache with an order event.
617    ///
618    /// # Errors
619    ///
620    /// Returns an error if updating an order fails.
621    fn update_order(&self, order_event: &OrderEventAny) -> anyhow::Result<()>;
622
623    /// Updates a position in the cache.
624    ///
625    /// # Errors
626    ///
627    /// Returns an error if updating a position fails.
628    fn update_position(&self, position: &Position) -> anyhow::Result<()>;
629
630    /// Creates a snapshot of order state.
631    ///
632    /// # Errors
633    ///
634    /// Returns an error if snapshotting order state fails.
635    fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()>;
636
637    /// Creates a snapshot of position state.
638    ///
639    /// # Errors
640    ///
641    /// Returns an error if snapshotting position state fails.
642    fn snapshot_position_state(
643        &self,
644        position: &Position,
645        ts_snapshot: UnixNanos,
646        unrealized_pnl: Option<Money>,
647    ) -> anyhow::Result<()>;
648
649    /// Records a heartbeat timestamp.
650    ///
651    /// # Errors
652    ///
653    /// Returns an error if heartbeat recording fails.
654    fn heartbeat(&self, timestamp: UnixNanos) -> anyhow::Result<()>;
655}
nautilus_common/cache/database.rs

nautilus_common/cache/
database.rs
