futu_cache/event_contract/
runtime.rs

1use super::*;
2
3impl EventContractCache {
4    #[must_use]
5    pub fn mvc_rule(&self, mvc: &str) -> Option<EventContractMvcRule> {
6        self.mvc_rules.load().get(mvc).cloned()
7    }
8
9    /// Publishes one complete MVC rule only after the exact CMD21328 body has
10    /// decoded and passed its business result.
11    ///
12    /// Ref: frozen C++ aec0f6cda1
13    /// `NNBiz_Qot_EventContractMVC.cpp:65-116`: absent ret_code is success,
14    /// missing limits become zero, zero event ids are skipped, and the
15    /// deprecated yes-only list is ignored.
16    pub fn apply_mvc_response(
17        &self,
18        mvc: &str,
19        body: &[u8],
20    ) -> Result<(), EventContractCacheError> {
21        if mvc.is_empty() {
22            return Err(EventContractCacheError::Validation(
23                "event contract MVC cache key is empty",
24            ));
25        }
26        let response = GetMvcInfoResponse::decode(body)
27            .map_err(|error| EventContractCacheError::Decode(error.to_string()))?;
28        if response.ret_code.unwrap_or(0) != 0 {
29            return Err(EventContractCacheError::Validation(
30                "event contract MVC backend ret_code is not success",
31            ));
32        }
33        let events = response
34            .mvc_event_info
35            .into_iter()
36            .filter_map(|event| {
37                let event_id = event.event_id.filter(|event_id| *event_id != 0)?;
38                Some(EventContractMvcEventRule {
39                    event_id,
40                    yes_only: event.yes_only.unwrap_or(false),
41                    min_contract_size: event.min_contract_size.unwrap_or(0),
42                    max_contract_size: event.max_contract_size.unwrap_or(0),
43                })
44            })
45            .collect();
46        let rule = EventContractMvcRule {
47            min_contract_size: response.min_contract_size.unwrap_or(0),
48            max_contract_size: response.max_contract_size.unwrap_or(0),
49            events,
50        };
51        let _writer = self.mvc_writer.lock();
52        let mut next = (**self.mvc_rules.load()).clone();
53        next.insert(mvc.to_owned(), rule);
54        self.mvc_rules.store(Arc::new(next));
55        Ok(())
56    }
57
58    /// Saves the RFQ identity required by the later combo-order consumer.
59    ///
60    /// The value deliberately contains only local write time, original-order
61    /// `(stock_id, pred_side)` legs, and combo origin. Price, expiry, account,
62    /// and market are not part of the C++ owner.
63    ///
64    /// Ref: frozen C++ aec0f6cda1
65    /// `NNData_EventContractRfqQuote.h:34-43` and
66    /// `NNData_EventContractRfqQuote.cpp:29-41`.
67    pub fn save_rfq_quote_at(
68        &self,
69        quote_id: &str,
70        legs: &[(u64, i32)],
71        combo_exchange: &str,
72        combo_origin_symbol: &str,
73        written_at: Instant,
74    ) {
75        if quote_id.is_empty() {
76            return;
77        }
78        self.purge_rfq_quotes_at(written_at);
79        let _writer = self.rfq_quote_writer.lock();
80        let mut next = (**self.rfq_quotes.load()).clone();
81        next.insert(
82            quote_id.to_owned(),
83            EventContractRfqQuoteEntry {
84                written_at,
85                legs: legs.to_vec(),
86                combo_exchange: combo_exchange.to_owned(),
87                combo_origin_symbol: combo_origin_symbol.to_owned(),
88            },
89        );
90        self.rfq_quotes.store(Arc::new(next));
91    }
92
93    pub fn save_rfq_quote(
94        &self,
95        quote_id: &str,
96        legs: &[(u64, i32)],
97        combo_exchange: &str,
98        combo_origin_symbol: &str,
99    ) {
100        self.save_rfq_quote_at(
101            quote_id,
102            legs,
103            combo_exchange,
104            combo_origin_symbol,
105            Instant::now(),
106        );
107    }
108
109    #[must_use]
110    pub fn rfq_quote_state(&self, quote_id: &str) -> Option<EventContractRfqQuoteState> {
111        self.purge_rfq_quotes_at(Instant::now());
112        self.rfq_quotes.load().get(quote_id).map(|entry| {
113            (
114                entry.written_at,
115                entry.legs.clone(),
116                entry.combo_exchange.clone(),
117                entry.combo_origin_symbol.clone(),
118            )
119        })
120    }
121
122    #[must_use]
123    pub fn is_rfq_quote_valid_at(&self, quote_id: &str, now: Instant) -> bool {
124        self.purge_rfq_quotes_at(now);
125        self.rfq_quotes.load().get(quote_id).is_some_and(|entry| {
126            // `Instant` is monotonic in production. A synthetic earlier
127            // instant (tests or a platform anomaly) is treated as zero age,
128            // matching the C++ `(now - stamp) <= validMs` intent that a
129            // negative age has not exceeded the 5000ms validity window.
130            now.checked_duration_since(entry.written_at)
131                .unwrap_or_default()
132                <= RFQ_QUOTE_VALIDITY
133        })
134    }
135
136    #[must_use]
137    pub fn rfq_quote_matches(&self, quote_id: &str, legs: &[(u64, i32)]) -> bool {
138        self.purge_rfq_quotes_at(Instant::now());
139        let quotes = self.rfq_quotes.load();
140        let Some(entry) = quotes.get(quote_id) else {
141            return false;
142        };
143        if entry.legs.len() != legs.len() {
144            return false;
145        }
146        let mut used = vec![false; legs.len()];
147        entry.legs.iter().all(|saved| {
148            let Some(index) = legs.iter().enumerate().find_map(|(index, candidate)| {
149                (!used[index] && candidate == saved).then_some(index)
150            }) else {
151                return false;
152            };
153            used[index] = true;
154            true
155        })
156    }
157
158    #[must_use]
159    pub fn rfq_quote_combo_origin(&self, quote_id: &str) -> Option<(String, String)> {
160        self.purge_rfq_quotes_at(Instant::now());
161        self.rfq_quotes.load().get(quote_id).and_then(|entry| {
162            (!entry.combo_exchange.is_empty() && !entry.combo_origin_symbol.is_empty()).then(|| {
163                (
164                    entry.combo_exchange.clone(),
165                    entry.combo_origin_symbol.clone(),
166                )
167            })
168        })
169    }
170
171    /// Removes entries whose local age is strictly greater than 60 seconds.
172    ///
173    /// Ref: frozen C++ aec0f6cda1
174    /// `NNData_EventContractRfqQuote.cpp:3-5,93-115`. C++ owns one cache-level
175    /// timer rather than one timer per quote. Rust performs the same single
176    /// owner cleanup opportunistically on every save/read: under activity the
177    /// map stays bounded, while an idle map cannot grow; every observable read
178    /// first removes the same `age > 60000ms` entries.
179    pub fn purge_rfq_quotes_at(&self, now: Instant) -> usize {
180        let _writer = self.rfq_quote_writer.lock();
181        let current = self.rfq_quotes.load();
182        let mut next = (**current).clone();
183        let before = next.len();
184        next.retain(|_, entry| {
185            now.checked_duration_since(entry.written_at)
186                .unwrap_or_default()
187                <= RFQ_QUOTE_STALE_AFTER
188        });
189        let removed = before - next.len();
190        if removed != 0 {
191            self.rfq_quotes.store(Arc::new(next));
192        }
193        removed
194    }
195
196    /// Builds transient subscription facts from the EventContract metadata
197    /// owner without copying the row into `StaticDataCache`.
198    ///
199    /// Ref: frozen C++ aec0f6cda1
200    /// `INNBiz_Qot_EventContract.h:40-44` and
201    /// `MktQotSubInstance.cpp:334-380`: EventContract identities use public
202    /// market 101, backend market 35, and quote security type 17.
203    #[must_use]
204    pub fn subscription_security_info(
205        &self,
206        public_sec_key: &str,
207    ) -> Option<crate::static_data::CachedSecurityInfo> {
208        let (market, code) = public_sec_key.split_once('_')?;
209        if market.parse::<i32>().ok()? != 101 || code.is_empty() {
210            return None;
211        }
212        let generation = self.metadata_snapshot();
213        let stock_id = *generation.security_id_by_code.get(code)?;
214        let metadata = generation.securities_by_id.get(&stock_id)?;
215        Some(crate::static_data::CachedSecurityInfo {
216            stock_id,
217            market: 101,
218            mkt_id: 35,
219            code: metadata.futu_symbol.clone(),
220            name: metadata.name.clone(),
221            lot_size: 1,
222            sec_type: 17,
223            spread_table_code: metadata.spread_table_code,
224            sub_instrument_type_v2: metadata.sub_instrument_type_v2,
225            ..Default::default()
226        })
227    }
228
229    #[must_use]
230    pub fn subscription_security_info_by_stock_id(
231        &self,
232        stock_id: u64,
233    ) -> Option<crate::static_data::CachedSecurityInfo> {
234        let generation = self.metadata_snapshot();
235        let metadata = generation.securities_by_id.get(&stock_id)?;
236        self.subscription_security_info(&format!("101_{}", metadata.futu_symbol))
237    }
238
239    /// Publishes one complete YES/NO order-book generation.
240    ///
241    /// The four quadrants move together so readers cannot combine YES from one
242    /// backend frame with NO from another. A delayed frame is rejected rather
243    /// than overwriting newer data.
244    #[must_use]
245    pub fn publish_order_book(
246        &self,
247        stock_id: u64,
248        generation: u64,
249        book: EventContractOrderBook,
250    ) -> bool {
251        if stock_id == 0 {
252            return false;
253        }
254        match self.order_books.entry(stock_id) {
255            dashmap::mapref::entry::Entry::Occupied(mut entry) => {
256                if generation <= entry.get().generation {
257                    return false;
258                }
259                entry.insert(Arc::new(EventContractOrderBookGeneration {
260                    generation,
261                    book,
262                    yes_generation: Some(generation),
263                    no_generation: Some(generation),
264                }));
265            }
266            dashmap::mapref::entry::Entry::Vacant(entry) => {
267                entry.insert(Arc::new(EventContractOrderBookGeneration {
268                    generation,
269                    book,
270                    yes_generation: Some(generation),
271                    no_generation: Some(generation),
272                }));
273            }
274        }
275        true
276    }
277
278    /// Merges one backend EventContract side into the four-quadrant public
279    /// book. YES and NO arrive as separate CMD6212 bit=3/prob=1|2 frames, so
280    /// each side owns an independent generation watermark.
281    ///
282    /// Ref: frozen C++ aec0f6cda1
283    /// `APIServer_Qot_EventContractPush.cpp`: the public 3450 body is emitted
284    /// only after both YES and NO `OrderBookYesNo` caches are ready.
285    #[must_use]
286    pub fn merge_order_book_side(
287        &self,
288        stock_id: u64,
289        generation: u64,
290        direction: i32,
291        bids: Vec<EventContractOrderBookLevel>,
292        asks: Vec<EventContractOrderBookLevel>,
293    ) -> bool {
294        if stock_id == 0 || !matches!(direction, 1 | 2) {
295            return false;
296        }
297        match self.order_books.entry(stock_id) {
298            dashmap::mapref::entry::Entry::Occupied(mut entry) => {
299                let current = entry.get();
300                let side_generation = if direction == 1 {
301                    current.yes_generation
302                } else {
303                    current.no_generation
304                };
305                if side_generation.is_some_and(|current| generation < current) {
306                    return false;
307                }
308                let mut next = (**current).clone();
309                next.generation = next.generation.max(generation);
310                if direction == 1 {
311                    next.book.yes_bids = bids;
312                    next.book.yes_asks = asks;
313                    next.yes_generation = Some(generation);
314                } else {
315                    next.book.no_bids = bids;
316                    next.book.no_asks = asks;
317                    next.no_generation = Some(generation);
318                }
319                entry.insert(Arc::new(next));
320            }
321            dashmap::mapref::entry::Entry::Vacant(entry) => {
322                let mut next = EventContractOrderBookGeneration {
323                    generation,
324                    book: EventContractOrderBook::default(),
325                    yes_generation: None,
326                    no_generation: None,
327                };
328                if direction == 1 {
329                    next.book.yes_bids = bids;
330                    next.book.yes_asks = asks;
331                    next.yes_generation = Some(generation);
332                } else {
333                    next.book.no_bids = bids;
334                    next.book.no_asks = asks;
335                    next.no_generation = Some(generation);
336                }
337                entry.insert(Arc::new(next));
338            }
339        }
340        true
341    }
342
343    #[must_use]
344    pub fn order_book_snapshot(
345        &self,
346        stock_id: u64,
347    ) -> Option<Arc<EventContractOrderBookGeneration>> {
348        self.order_books.get(&stock_id).map(|entry| entry.clone())
349    }
350}
futu_cache/event_contract/runtime.rs

futu_cache/event_contract/
runtime.rs
