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    /// Clears persisted data according to the backing's scope.
147    ///
148    /// Redis clears the entire logical database, including other traders and user keys.
149    /// PostgreSQL removes only this trader's rows and saved component state.
150    ///
151    /// # Errors
152    ///
153    /// Returns an error if removing the persisted data fails.
154    fn flush(&mut self) -> anyhow::Result<()>;
155
156    /// Loads all cached data into memory.
157    ///
158    /// # Errors
159    ///
160    /// Returns an error if loading data from the database fails.
161    async fn load_all(&self) -> anyhow::Result<CacheMap>;
162
163    /// Loads raw key-value data from the database.
164    ///
165    /// # Errors
166    ///
167    /// Returns an error if the load operation fails.
168    fn load(&self) -> anyhow::Result<AHashMap<String, Bytes>>;
169
170    /// Loads all currencies from the cache.
171    ///
172    /// # Errors
173    ///
174    /// Returns an error if loading currencies fails.
175    async fn load_currencies(&self) -> anyhow::Result<AHashMap<Ustr, Currency>>;
176
177    /// Loads all instruments from the cache.
178    ///
179    /// # Errors
180    ///
181    /// Returns an error if loading instruments fails.
182    async fn load_instruments(&self) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>>;
183
184    /// Loads all instrument closes from the cache.
185    ///
186    /// # Errors
187    ///
188    /// Returns an error if loading instrument closes fails.
189    async fn load_instrument_closes(
190        &self,
191    ) -> anyhow::Result<AHashMap<InstrumentId, InstrumentClose>>;
192
193    /// Loads all synthetic instruments from the cache.
194    ///
195    /// # Errors
196    ///
197    /// Returns an error if loading synthetic instruments fails.
198    async fn load_synthetics(&self) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>>;
199
200    /// Loads all accounts from the cache.
201    ///
202    /// # Errors
203    ///
204    /// Returns an error if loading accounts fails.
205    async fn load_accounts(&self) -> anyhow::Result<AHashMap<AccountId, AccountAny>>;
206
207    /// Loads all orders from the cache.
208    ///
209    /// # Errors
210    ///
211    /// Returns an error if loading orders fails.
212    async fn load_orders(&self) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>>;
213
214    /// Loads all positions from the cache.
215    ///
216    /// # Errors
217    ///
218    /// Returns an error if loading positions fails.
219    async fn load_positions(&self) -> anyhow::Result<AHashMap<PositionId, Position>>;
220
221    /// Loads all [`GreeksData`] from the cache.
222    ///
223    /// # Errors
224    ///
225    /// Returns an error if loading greeks data fails or the adapter does not support it.
226    async fn load_greeks(&self) -> anyhow::Result<AHashMap<InstrumentId, GreeksData>> {
227        anyhow::bail!("Loading greeks for cache adapter not supported")
228    }
229
230    /// Loads all [`YieldCurve`] from the cache.
231    ///
232    /// # Errors
233    ///
234    /// Returns an error if loading yield curve data fails or the adapter does not support it.
235    async fn load_yield_curves(&self) -> anyhow::Result<AHashMap<String, YieldCurve>> {
236        anyhow::bail!("Loading yield curves for cache adapter not supported")
237    }
238
239    /// Loads mapping from order IDs to position IDs.
240    ///
241    /// # Errors
242    ///
243    /// Returns an error if loading the index order-position mapping fails.
244    fn load_index_order_position(&self) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>>;
245
246    /// Loads mapping from order IDs to client IDs.
247    ///
248    /// # Errors
249    ///
250    /// Returns an error if loading the index order-client mapping fails.
251    fn load_index_order_client(&self) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>>;
252
253    /// Loads a single currency by code.
254    ///
255    /// # Errors
256    ///
257    /// Returns an error if loading a single currency fails.
258    async fn load_currency(&self, code: &Ustr) -> anyhow::Result<Option<Currency>>;
259
260    /// Loads a single instrument by ID.
261    ///
262    /// # Errors
263    ///
264    /// Returns an error if loading a single instrument fails.
265    async fn load_instrument(
266        &self,
267        instrument_id: &InstrumentId,
268    ) -> anyhow::Result<Option<InstrumentAny>>;
269
270    /// Loads a single synthetic instrument by ID.
271    ///
272    /// # Errors
273    ///
274    /// Returns an error if loading a single synthetic instrument fails.
275    async fn load_synthetic(
276        &self,
277        instrument_id: &InstrumentId,
278    ) -> anyhow::Result<Option<SyntheticInstrument>>;
279
280    /// Loads an instrument for a synchronous cache caller.
281    ///
282    /// # Errors
283    ///
284    /// Returns an error if loading fails or synchronous reads are unsupported.
285    fn load_instrument_sync(
286        &self,
287        _instrument_id: &InstrumentId,
288    ) -> anyhow::Result<Option<InstrumentAny>> {
289        anyhow::bail!("Synchronous instrument loading is not supported by this cache database")
290    }
291
292    /// Loads a synthetic instrument for a synchronous cache caller.
293    ///
294    /// # Errors
295    ///
296    /// Returns an error if loading fails or synchronous reads are unsupported.
297    fn load_synthetic_sync(
298        &self,
299        _instrument_id: &InstrumentId,
300    ) -> anyhow::Result<Option<SyntheticInstrument>> {
301        anyhow::bail!("Synchronous synthetic loading is not supported by this cache database")
302    }
303
304    /// Loads a single account by ID.
305    ///
306    /// # Errors
307    ///
308    /// Returns an error if loading a single account fails.
309    async fn load_account(&self, account_id: &AccountId) -> anyhow::Result<Option<AccountAny>>;
310
311    /// Loads a single order by client order ID.
312    ///
313    /// # Errors
314    ///
315    /// Returns an error if loading a single order fails.
316    async fn load_order(&self, client_order_id: &ClientOrderId)
317    -> anyhow::Result<Option<OrderAny>>;
318
319    /// Loads a single position by position ID.
320    ///
321    /// # Errors
322    ///
323    /// Returns an error if loading a single position fails.
324    async fn load_position(&self, position_id: &PositionId) -> anyhow::Result<Option<Position>>;
325
326    /// Loads actor state by actor ID.
327    ///
328    /// # Errors
329    ///
330    /// Returns an error if loading actor state fails.
331    fn load_actor(&self, actor_id: &ActorId) -> anyhow::Result<AHashMap<String, Bytes>>;
332
333    /// Loads strategy state by strategy ID.
334    ///
335    /// # Errors
336    ///
337    /// Returns an error if loading strategy state fails.
338    fn load_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<AHashMap<String, Bytes>>;
339
340    /// Loads signals by name.
341    ///
342    /// # Errors
343    ///
344    /// Returns an error if loading signals fails.
345    fn load_signals(&self, name: &str) -> anyhow::Result<Vec<Signal>>;
346
347    /// Loads custom data by data type.
348    ///
349    /// # Errors
350    ///
351    /// Returns an error if loading custom data fails.
352    fn load_custom_data(&self, data_type: &DataType) -> anyhow::Result<Vec<CustomData>>;
353
354    /// Loads an order snapshot by client order ID.
355    ///
356    /// # Errors
357    ///
358    /// Returns an error if loading the order snapshot fails.
359    fn load_order_snapshot(
360        &self,
361        client_order_id: &ClientOrderId,
362    ) -> anyhow::Result<Option<OrderSnapshot>>;
363
364    /// Loads a position snapshot by position ID.
365    ///
366    /// # Errors
367    ///
368    /// Returns an error if loading the position snapshot fails.
369    fn load_position_snapshot(
370        &self,
371        position_id: &PositionId,
372    ) -> anyhow::Result<Option<PositionSnapshot>>;
373
374    /// Loads quote ticks by instrument ID.
375    ///
376    /// # Errors
377    ///
378    /// Returns an error if loading quotes fails.
379    fn load_quotes(&self, instrument_id: &InstrumentId) -> anyhow::Result<Vec<QuoteTick>>;
380
381    /// Loads trade ticks by instrument ID.
382    ///
383    /// # Errors
384    ///
385    /// Returns an error if loading trades fails.
386    fn load_trades(&self, instrument_id: &InstrumentId) -> anyhow::Result<Vec<TradeTick>>;
387
388    /// Loads funding rate updates by instrument ID.
389    ///
390    /// # Errors
391    ///
392    /// Returns an error if loading funding rates fails.
393    fn load_funding_rates(
394        &self,
395        instrument_id: &InstrumentId,
396    ) -> anyhow::Result<Vec<FundingRateUpdate>>;
397
398    /// Loads bars by instrument ID.
399    ///
400    /// # Errors
401    ///
402    /// Returns an error if loading bars fails.
403    fn load_bars(&self, instrument_id: &InstrumentId) -> anyhow::Result<Vec<Bar>>;
404
405    /// Adds a generic key-value pair to the cache.
406    ///
407    /// # Errors
408    ///
409    /// Returns an error if adding a generic key/value fails.
410    fn add(&self, key: String, value: Bytes) -> anyhow::Result<()>;
411
412    /// Adds a currency to the cache.
413    ///
414    /// # Errors
415    ///
416    /// Returns an error if adding a currency fails.
417    fn add_currency(&self, currency: &Currency) -> anyhow::Result<()>;
418
419    /// Adds an instrument to the cache.
420    ///
421    /// # Errors
422    ///
423    /// Returns an error if adding an instrument fails.
424    fn add_instrument(&self, instrument: &InstrumentAny) -> anyhow::Result<()>;
425
426    /// Persists a token instrument, checking legacy replacement before committing it.
427    ///
428    /// # Errors
429    ///
430    /// Returns an error if integrity or migration checks fail, or the backend lacks DeFi support.
431    #[cfg(feature = "defi")]
432    fn add_defi_instrument(
433        &self,
434        _instrument: &InstrumentAny,
435        _migration: crate::defi::cache::InstrumentMigration,
436    ) -> anyhow::Result<()> {
437        anyhow::bail!("DeFi instrument persistence is unsupported by this backend")
438    }
439
440    /// Adds an instrument close to the cache, replacing any existing value for the instrument.
441    /// Implementations must queue persistence without waiting for the database operation to
442    /// complete.
443    ///
444    /// # Errors
445    ///
446    /// Returns an error if the instrument close cannot be queued for persistence.
447    fn add_instrument_close(&self, close: &InstrumentClose) -> anyhow::Result<()>;
448
449    /// Adds a synthetic instrument to the cache.
450    ///
451    /// # Errors
452    ///
453    /// Returns an error if adding a synthetic instrument fails.
454    fn add_synthetic(&self, synthetic: &SyntheticInstrument) -> anyhow::Result<()>;
455
456    /// Adds an account to the cache.
457    ///
458    /// # Errors
459    ///
460    /// Returns an error if adding an account fails.
461    fn add_account(&self, account: &AccountAny) -> anyhow::Result<()>;
462
463    /// Adds an order to the cache.
464    ///
465    /// # Errors
466    ///
467    /// Returns an error if adding an order fails.
468    fn add_order(&self, order: &OrderAny, client_id: Option<ClientId>) -> anyhow::Result<()>;
469
470    /// Adds an order snapshot to the cache.
471    ///
472    /// # Errors
473    ///
474    /// Returns an error if adding an order snapshot fails.
475    fn add_order_snapshot(&self, snapshot: &OrderSnapshot) -> anyhow::Result<()>;
476
477    /// Adds a position to the cache.
478    ///
479    /// # Errors
480    ///
481    /// Returns an error if adding a position fails.
482    fn add_position(&self, position: &Position) -> anyhow::Result<()>;
483
484    /// Adds a position snapshot to the cache.
485    ///
486    /// # Errors
487    ///
488    /// Returns an error if adding a position snapshot fails.
489    fn add_position_snapshot(&self, snapshot: &PositionSnapshot) -> anyhow::Result<()>;
490
491    /// Adds an order book to the cache.
492    ///
493    /// # Errors
494    ///
495    /// Returns an error if adding an order book fails.
496    fn add_order_book(&self, order_book: &OrderBook) -> anyhow::Result<()>;
497
498    /// Adds a signal to the cache.
499    ///
500    /// # Errors
501    ///
502    /// Returns an error if adding a signal fails.
503    fn add_signal(&self, signal: &Signal) -> anyhow::Result<()>;
504
505    /// Adds custom data to the cache.
506    ///
507    /// # Errors
508    ///
509    /// Returns an error if adding custom data fails.
510    fn add_custom_data(&self, data: &CustomData) -> anyhow::Result<()>;
511
512    /// Adds a quote tick to the cache.
513    ///
514    /// # Errors
515    ///
516    /// Returns an error if adding a quote tick fails.
517    fn add_quote(&self, quote: &QuoteTick) -> anyhow::Result<()>;
518
519    /// Adds a trade tick to the cache.
520    ///
521    /// # Errors
522    ///
523    /// Returns an error if adding a trade tick fails.
524    fn add_trade(&self, trade: &TradeTick) -> anyhow::Result<()>;
525
526    /// Adds a funding rate update to the cache.
527    ///
528    /// # Errors
529    ///
530    /// Returns an error if adding a funding rate update fails.
531    fn add_funding_rate(&self, funding_rate: &FundingRateUpdate) -> anyhow::Result<()>;
532
533    /// Adds a bar to the cache.
534    ///
535    /// # Errors
536    ///
537    /// Returns an error if adding a bar fails.
538    fn add_bar(&self, bar: &Bar) -> anyhow::Result<()>;
539
540    /// Adds greeks data to the cache.
541    ///
542    /// # Errors
543    ///
544    /// Returns an error if adding greeks data fails or the adapter does not support it.
545    fn add_greeks(&self, _greeks: &GreeksData) -> anyhow::Result<()> {
546        anyhow::bail!("Saving greeks for cache adapter not supported")
547    }
548
549    /// Adds yield curve data to the cache.
550    ///
551    /// # Errors
552    ///
553    /// Returns an error if adding yield curve data fails or the adapter does not support it.
554    fn add_yield_curve(&self, _yield_curve: &YieldCurve) -> anyhow::Result<()> {
555        anyhow::bail!("Saving yield curves for cache adapter not supported")
556    }
557
558    /// Deletes actor state from the cache.
559    ///
560    /// # Errors
561    ///
562    /// Returns an error if deleting actor state fails.
563    fn delete_actor(&self, actor_id: &ActorId) -> anyhow::Result<()>;
564
565    /// Deletes strategy state from the cache.
566    ///
567    /// # Errors
568    ///
569    /// Returns an error if deleting strategy state fails.
570    fn delete_strategy(&self, component_id: &StrategyId) -> anyhow::Result<()>;
571
572    /// Deletes an order from the cache.
573    ///
574    /// # Errors
575    ///
576    /// Returns an error if deleting an order fails.
577    fn delete_order(&self, client_order_id: &ClientOrderId) -> anyhow::Result<()>;
578
579    /// Deletes a position from the cache.
580    ///
581    /// # Errors
582    ///
583    /// Returns an error if deleting a position fails.
584    fn delete_position(&self, position_id: &PositionId) -> anyhow::Result<()>;
585
586    /// Requests deletion of an account event from the database.
587    ///
588    /// Redis and PostgreSQL retain the full account history for balance and margin replay;
589    /// their implementations log the request and return success without deleting it.
590    ///
591    /// # Errors
592    ///
593    /// Returns an error if deleting account events fails.
594    fn delete_account_event(&self, account_id: &AccountId, event_id: &str) -> anyhow::Result<()>;
595
596    /// Indexes a venue order ID with its client order ID.
597    ///
598    /// # Errors
599    ///
600    /// Returns an error if indexing venue order ID fails.
601    fn index_venue_order_id(
602        &self,
603        client_order_id: ClientOrderId,
604        venue_order_id: VenueOrderId,
605    ) -> anyhow::Result<()>;
606
607    /// Indexes an order-position mapping.
608    ///
609    /// # Errors
610    ///
611    /// Returns an error if indexing order-position mapping fails.
612    fn index_order_position(
613        &self,
614        client_order_id: ClientOrderId,
615        position_id: PositionId,
616    ) -> anyhow::Result<()>;
617
618    /// Indexes order-client mappings as one batch operation.
619    ///
620    /// # Errors
621    ///
622    /// Returns an error if batch order-client indexing is unsupported or cannot be enqueued.
623    fn index_order_clients(&self, claims: &[(ClientOrderId, ClientId)]) -> anyhow::Result<()> {
624        if claims.is_empty() {
625            return Ok(());
626        }
627
628        anyhow::bail!("Batch order-client indexing is not supported by this cache database")
629    }
630
631    /// Updates actor state in the cache.
632    ///
633    /// # Errors
634    ///
635    /// Returns an error if updating actor state fails.
636    fn update_actor(
637        &self,
638        actor_id: &ActorId,
639        state: &AHashMap<String, Bytes>,
640    ) -> anyhow::Result<()>;
641
642    /// Updates strategy state in the cache.
643    ///
644    /// # Errors
645    ///
646    /// Returns an error if updating strategy state fails.
647    fn update_strategy(
648        &self,
649        strategy_id: &StrategyId,
650        state: &AHashMap<String, Bytes>,
651    ) -> anyhow::Result<()>;
652
653    /// Updates an account in the cache.
654    ///
655    /// # Errors
656    ///
657    /// Returns an error if updating an account fails.
658    fn update_account(&self, account: &AccountAny) -> anyhow::Result<()>;
659
660    /// Updates an order in the cache with an order event.
661    ///
662    /// # Errors
663    ///
664    /// Returns an error if updating an order fails.
665    fn update_order(&self, order_event: &OrderEventAny) -> anyhow::Result<()>;
666
667    /// Updates a position in the cache.
668    ///
669    /// # Errors
670    ///
671    /// Returns an error if updating a position fails.
672    fn update_position(&self, position: &Position) -> anyhow::Result<()>;
673
674    /// Creates a snapshot of order state.
675    ///
676    /// # Errors
677    ///
678    /// Returns an error if snapshotting order state fails.
679    fn snapshot_order_state(&self, order: &OrderAny) -> anyhow::Result<()>;
680
681    /// Creates a snapshot of position state.
682    ///
683    /// # Errors
684    ///
685    /// Returns an error if snapshotting position state fails.
686    fn snapshot_position_state(
687        &self,
688        position: &Position,
689        ts_snapshot: UnixNanos,
690        unrealized_pnl: Option<Money>,
691    ) -> anyhow::Result<()>;
692
693    /// Records a heartbeat timestamp.
694    ///
695    /// # Errors
696    ///
697    /// Returns an error if heartbeat recording fails.
698    fn heartbeat(&self, timestamp: UnixNanos) -> anyhow::Result<()>;
699}
nautilus_common/cache/database.rs

nautilus_common/cache/
database.rs
