futu_cache/
market_event.rs

1use std::collections::BTreeSet;
2use std::sync::atomic::{AtomicU64, Ordering};
3
4use dashmap::DashMap;
5use futu_domain_qot_market_state::{
6    MarketEventUpdate, MarketStateSnapshot, MarketTradeDateSnapshot, TradeDateUpdatePlan,
7    merge_market_state_patch, plan_market_trade_date_update,
8};
9
10#[derive(Clone, Debug, Default, PartialEq, Eq)]
11pub struct MarketEventApplyOutcome {
12    pub market_state_updates: usize,
13    pub trade_date_updates: usize,
14    pub changed_trade_date_quote_markets: Vec<u8>,
15    pub changed_trade_date_market_ids: Vec<u32>,
16}
17
18#[derive(Clone, Copy, Debug, PartialEq, Eq)]
19pub struct MarketTradeDateToken {
20    pub market_id: u32,
21    pub trade_date: u32,
22    generation: u64,
23}
24
25#[derive(Default)]
26pub struct MarketEventCache {
27    market_states: DashMap<u32, MarketStateSnapshot>,
28    market_trade_dates: DashMap<u32, MarketTradeDateSnapshot>,
29    market_trade_date_generations: DashMap<u32, u64>,
30    next_trade_date_generation: AtomicU64,
31    trade_date_fence: parking_lot::RwLock<()>,
32}
33
34impl MarketEventCache {
35    #[must_use]
36    pub fn new() -> Self {
37        Self::default()
38    }
39
40    pub fn apply_batch(&self, updates: &[MarketEventUpdate]) -> MarketEventApplyOutcome {
41        let _fence = self.trade_date_fence.write();
42        let mut outcome = MarketEventApplyOutcome::default();
43        let mut changed_quote_markets = BTreeSet::new();
44        let mut changed_market_ids = BTreeSet::new();
45
46        for update in updates {
47            match update {
48                MarketEventUpdate::MarketState(patch) => {
49                    let existing = self
50                        .market_states
51                        .get(&patch.market_id)
52                        .map(|row| row.clone());
53                    let merged = merge_market_state_patch(existing.as_ref(), patch.clone());
54                    self.market_states.insert(merged.market_id, merged);
55                    outcome.market_state_updates += 1;
56                }
57                MarketEventUpdate::MarketTradeDate(update) => {
58                    let existing = self
59                        .market_trade_dates
60                        .get(&update.market_id)
61                        .map(|row| row.clone());
62                    if let TradeDateUpdatePlan::Changed(snapshot) =
63                        plan_market_trade_date_update(existing.as_ref(), update.clone())
64                    {
65                        // C++ stores unknown market ids but only inserts a
66                        // non-UNKNOWN type into the `ReSubMkt` set. Ref:
67                        // `NNBiz_Qot_EventNotice.cpp:286-299`.
68                        if snapshot.quote_market_type != 0 {
69                            changed_quote_markets.insert(snapshot.quote_market_type);
70                        }
71                        changed_market_ids.insert(snapshot.market_id);
72                        let generation = self
73                            .next_trade_date_generation
74                            .fetch_add(1, Ordering::AcqRel)
75                            .wrapping_add(1);
76                        self.market_trade_date_generations
77                            .insert(snapshot.market_id, generation);
78                        self.market_trade_dates.insert(snapshot.market_id, snapshot);
79                        outcome.trade_date_updates += 1;
80                    }
81                }
82            }
83        }
84
85        outcome.changed_trade_date_quote_markets = changed_quote_markets.into_iter().collect();
86        outcome.changed_trade_date_market_ids = changed_market_ids.into_iter().collect();
87        outcome
88    }
89
90    #[must_use]
91    pub fn market_state(&self, market_id: u32) -> Option<MarketStateSnapshot> {
92        self.market_states.get(&market_id).map(|row| row.clone())
93    }
94
95    #[must_use]
96    pub fn market_states(&self) -> Vec<MarketStateSnapshot> {
97        let mut rows: Vec<_> = self.market_states.iter().map(|row| row.clone()).collect();
98        rows.sort_by_key(|row| row.market_id);
99        rows
100    }
101
102    #[must_use]
103    pub fn market_trade_date(&self, market_id: u32) -> Option<MarketTradeDateSnapshot> {
104        self.market_trade_dates
105            .get(&market_id)
106            .map(|row| row.clone())
107    }
108
109    #[must_use]
110    pub fn market_trade_date_token(&self, market_id: u32) -> Option<MarketTradeDateToken> {
111        let _fence = self.trade_date_fence.read();
112        let snapshot = self.market_trade_dates.get(&market_id)?;
113        let generation = *self.market_trade_date_generations.get(&market_id)?;
114        Some(MarketTradeDateToken {
115            market_id,
116            trade_date: snapshot.trade_date,
117            generation,
118        })
119    }
120
121    pub fn with_trade_date_tokens_current<T>(
122        &self,
123        tokens: &[MarketTradeDateToken],
124        commit: impl FnOnce() -> T,
125    ) -> Option<T> {
126        let _fence = self.trade_date_fence.read();
127        let current = tokens.iter().all(|token| {
128            self.market_trade_dates
129                .get(&token.market_id)
130                .is_some_and(|snapshot| snapshot.trade_date == token.trade_date)
131                && self
132                    .market_trade_date_generations
133                    .get(&token.market_id)
134                    .is_some_and(|generation| *generation == token.generation)
135        });
136        current.then(commit)
137    }
138}
futu_cache/market_event.rs

futu_cache/
market_event.rs
