1use dashmap::DashMap;
4use std::collections::{BTreeSet, HashSet};
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::{Arc, Mutex, RwLock};
7
8mod readiness;
9mod security_identity;
10mod sync_status;
11mod types;
12
13pub use readiness::{StaticDataReadiness, StockListSyncStatus};
14use sync_status::StockListSyncCounters;
15pub use types::{
16 CachedPlateInfo, CachedSecurityInfo, CachedTradeDate, CryptoPairInfo, CryptoTradeConfig,
17 OptionContractInfo, SecurityInfoSource,
18};
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum OnDemandSecurityPublishOutcome {
22 BasicPublished,
23 ZeroIdRepaired,
24 CompleteMktIdUpdated,
25 Rejected,
26}
27
28pub struct StaticDataCache {
30 securities: DashMap<String, Arc<CachedSecurityInfo>>,
36 securities_by_stock_id: DashMap<u64, Arc<CachedSecurityInfo>>,
42 id_to_key: DashMap<u64, String>,
45 security_request_aliases: DashMap<u64, HashSet<String>>,
55 public_stock_ids_by_key: DashMap<String, BTreeSet<u64>>,
63 future_main_link_aliases: DashMap<u64, HashSet<String>>,
70 option_contracts: DashMap<u64, OptionContractInfo>,
72 crypto_pairs: DashMap<String, CryptoPairInfo>,
74 crypto_pairs_by_stock_id: DashMap<u64, CryptoPairInfo>,
77 crypto_trade_configs: DashMap<String, CryptoTradeConfig>,
79 pub trade_dates: DashMap<String, Vec<CachedTradeDate>>,
81 pub plates: DashMap<String, Vec<CachedPlateInfo>>,
83 owner_to_warrants: RwLock<std::collections::HashMap<u64, HashSet<u64>>>,
90
91 zero_id_repair: Mutex<()>,
95
96 #[cfg(test)]
100 public_security_candidate_inspections: AtomicU64,
101
102 pub stale_mkt_ids: DashMap<String, ()>,
110
111 pub mkt_id_refresh_marked_total: AtomicU64,
117 pub mkt_id_refresh_done_total: AtomicU64,
118 pub mkt_id_refresh_failed_total: AtomicU64,
119
120 stock_list_sync: StockListSyncCounters,
126}
127
128impl StaticDataCache {
129 pub fn new() -> Self {
130 Self {
131 securities: DashMap::new(),
132 securities_by_stock_id: DashMap::new(),
133 id_to_key: DashMap::new(),
134 security_request_aliases: DashMap::new(),
135 public_stock_ids_by_key: DashMap::new(),
136 future_main_link_aliases: DashMap::new(),
137 option_contracts: DashMap::new(),
138 crypto_pairs: DashMap::new(),
139 crypto_pairs_by_stock_id: DashMap::new(),
140 crypto_trade_configs: DashMap::new(),
141 trade_dates: DashMap::new(),
142 plates: DashMap::new(),
143 owner_to_warrants: RwLock::new(std::collections::HashMap::new()),
144 zero_id_repair: Mutex::new(()),
145 #[cfg(test)]
146 public_security_candidate_inspections: AtomicU64::new(0),
147 stale_mkt_ids: DashMap::new(),
148 mkt_id_refresh_marked_total: AtomicU64::new(0),
149 mkt_id_refresh_done_total: AtomicU64::new(0),
150 mkt_id_refresh_failed_total: AtomicU64::new(0),
151 stock_list_sync: StockListSyncCounters::new(),
152 }
153 }
154
155 pub fn record_stock_list_sync_started(&self) {
156 self.stock_list_sync.record_started();
157 }
158
159 pub fn record_stock_list_sync_finished(
160 &self,
161 version: u64,
162 total_stocks: u64,
163 cached_count: u64,
164 finished_ms: u64,
165 ) {
166 self.stock_list_sync
167 .record_finished(version, total_stocks, cached_count, finished_ms);
168 }
169
170 pub fn record_stock_list_sync_failed(&self) {
171 self.stock_list_sync.record_failed();
172 }
173
174 pub fn record_stock_list_sync_recoverable_retry(&self) {
175 self.stock_list_sync.record_recoverable_retry();
176 }
177
178 pub fn stock_list_sync_status(&self) -> StockListSyncStatus {
179 self.stock_list_sync.status()
180 }
181
182 pub fn security_info_count(&self) -> usize {
183 let mut positive_ids = HashSet::new();
184 let mut zero_id_rows = 0usize;
185 for info in &self.securities {
186 if info.stock_id == 0 {
187 zero_id_rows += 1;
188 } else {
189 positive_ids.insert(info.stock_id);
190 }
191 }
192 positive_ids.len() + zero_id_rows
193 }
194
195 pub fn clear_stock_list_security_info(&self) -> usize {
202 let _identity_write = match self.zero_id_repair.lock() {
203 Ok(guard) => guard,
204 Err(_) => return 0,
205 };
206 let removed = self.security_info_count();
207 self.securities.clear();
208 self.securities_by_stock_id.clear();
209 self.id_to_key.clear();
210 self.security_request_aliases.clear();
211 self.public_stock_ids_by_key.clear();
212 self.future_main_link_aliases.clear();
213 self.option_contracts.clear();
214 self.crypto_pairs.clear();
215 self.crypto_pairs_by_stock_id.clear();
216 self.stale_mkt_ids.clear();
217 if let Ok(mut owners) = self.owner_to_warrants.write() {
218 owners.clear();
219 }
220 removed
221 }
222
223 pub fn stock_list_readiness(&self) -> StaticDataReadiness {
224 self.stock_list_sync_status()
225 .readiness_for_security_count(self.security_info_count())
226 }
227
228 pub fn get_security_info_trigger_refresh(&self, key: &str) -> Option<CachedSecurityInfo> {
238 let info = self.get_security_info(key)?;
239 if info.needs_mkt_id_refresh() {
240 self.mark_stale_mkt_id(key);
241 }
242 Some(info)
243 }
244
245 pub fn mark_stale_mkt_id(&self, key: &str) {
250 self.stale_mkt_ids.insert(key.to_string(), ());
251 self.mkt_id_refresh_marked_total
252 .fetch_add(1, Ordering::Relaxed);
253 }
254
255 pub fn drain_stale_mkt_ids(&self) -> Vec<String> {
273 let keys: Vec<String> = self.stale_mkt_ids.iter().map(|e| e.key().clone()).collect();
274 for k in &keys {
275 self.stale_mkt_ids.remove(k);
276 }
277 keys
278 }
279
280 pub fn update_mkt_id(&self, key: &str, new_mkt_id: u32) -> bool {
285 let _identity_write = match self.zero_id_repair.lock() {
286 Ok(guard) => guard,
287 Err(_) => return false,
288 };
289 self.update_mkt_id_locked(key, new_mkt_id)
290 }
291
292 fn update_mkt_id_locked(&self, key: &str, new_mkt_id: u32) -> bool {
293 let Some(public) = self
294 .securities
295 .get(key)
296 .map(|entry| Arc::clone(entry.value()))
297 else {
298 return false;
299 };
300
301 if public.stock_id == 0 {
302 if let Some(mut entry) = self.securities.get_mut(key) {
303 Arc::make_mut(&mut entry).mkt_id = new_mkt_id;
304 } else {
305 return false;
306 }
307 } else {
308 let Some(mut exact) = self.securities_by_stock_id.get_mut(&public.stock_id) else {
309 return false;
310 };
311 Arc::make_mut(&mut exact).mkt_id = new_mkt_id;
312 drop(exact);
313 self.refresh_public_keys_for_stock_id_locked(public.stock_id);
314 }
315
316 self.mkt_id_refresh_done_total
317 .fetch_add(1, Ordering::Relaxed);
318 true
319 }
320
321 pub fn record_mkt_id_refresh_failed(&self) {
323 self.mkt_id_refresh_failed_total
324 .fetch_add(1, Ordering::Relaxed);
325 }
326
327 #[must_use]
329 pub fn stale_mkt_ids_count(&self) -> usize {
330 self.stale_mkt_ids.len()
331 }
332
333 fn add_owner_relation(&self, info: &CachedSecurityInfo) {
334 if info.warrnt_stock_owner == 0 {
335 return;
336 }
337 if let Ok(mut map) = self.owner_to_warrants.write() {
338 map.entry(info.warrnt_stock_owner)
339 .or_default()
340 .insert(info.stock_id);
341 }
342 }
343
344 fn remove_owner_relation(&self, info: &CachedSecurityInfo) {
345 let owner = info.warrnt_stock_owner;
346 if owner == 0 {
347 return;
348 }
349 if info.stock_id == 0
350 && self
351 .securities
352 .iter()
353 .any(|candidate| candidate.stock_id == 0 && candidate.warrnt_stock_owner == owner)
354 {
355 return;
356 }
357 if let Ok(mut map) = self.owner_to_warrants.write()
358 && let Some(set) = map.get_mut(&owner)
359 {
360 set.remove(&info.stock_id);
361 if set.is_empty() {
362 map.remove(&owner);
363 }
364 }
365 }
366
367 fn remove_future_main_link_aliases_if_unreferenced(
368 &self,
369 key: &str,
370 removed: &CachedSecurityInfo,
371 ) {
372 for target in Self::future_main_link_target_ids(removed) {
373 let still_referenced = self.id_to_key.iter().any(|mapped| {
374 mapped.value() == key
375 && self
376 .securities_by_stock_id
377 .get(mapped.key())
378 .is_some_and(|info| {
379 Self::future_main_link_target_ids(&info).contains(&target)
380 })
381 });
382 if still_referenced {
383 continue;
384 }
385 if let Some(mut aliases) = self.future_main_link_aliases.get_mut(&target) {
386 aliases.remove(key);
387 let empty = aliases.is_empty();
388 drop(aliases);
389 if empty {
390 self.future_main_link_aliases.remove(&target);
391 }
392 }
393 }
394 }
395
396 fn future_main_link_target_ids(info: &CachedSecurityInfo) -> Vec<u64> {
397 let mut ids = Vec::with_capacity(2);
398 for target in [info.future_origin_id, info.zhuli_id] {
399 if target != 0 && target != info.stock_id && !ids.contains(&target) {
400 ids.push(target);
401 }
402 }
403 ids
404 }
405
406 fn add_future_main_link_aliases(&self, key: &str, info: &CachedSecurityInfo) {
407 for target in Self::future_main_link_target_ids(info) {
408 self.future_main_link_aliases
409 .entry(target)
410 .or_default()
411 .insert(key.to_string());
412 }
413 }
414
415 fn remove_future_main_link_aliases(&self, key: &str, info: &CachedSecurityInfo) {
416 for target in Self::future_main_link_target_ids(info) {
417 if let Some(mut aliases) = self.future_main_link_aliases.get_mut(&target) {
418 aliases.remove(key);
419 let empty = aliases.is_empty();
420 drop(aliases);
421 if empty {
422 self.future_main_link_aliases.remove(&target);
423 }
424 }
425 }
426 }
427
428 #[must_use]
434 pub fn get_future_main_link_alias_keys(&self, stock_id: u64) -> Vec<String> {
435 let Some(aliases) = self.future_main_link_aliases.get(&stock_id) else {
436 return Vec::new();
437 };
438 let mut keys: Vec<String> = aliases.iter().cloned().collect();
439 keys.sort();
440 keys
441 }
442
443 #[must_use]
450 pub fn quote_push_targets_for_stock_id(
451 &self,
452 stock_id: u64,
453 ) -> Vec<(String, Arc<CachedSecurityInfo>)> {
454 let mut targets = Vec::new();
455
456 if let Some(sec_key_ref) = self.id_to_key.get(&stock_id) {
457 let sec_key = sec_key_ref.clone();
458 drop(sec_key_ref);
459 if let Some(info) = self
460 .securities_by_stock_id
461 .get(&stock_id)
462 .map(|entry| Arc::clone(entry.value()))
463 {
464 targets.push((sec_key, info));
465 }
466 }
467
468 for alias_key in self.get_future_main_link_alias_keys(stock_id) {
469 if targets.iter().any(|(key, _)| key == &alias_key) {
470 continue;
471 }
472 if let Some(info) = self.get_security_info_arc(&alias_key) {
473 targets.push((alias_key, info));
474 }
475 }
476
477 targets
478 }
479
480 #[must_use]
499 pub fn quote_push_targets_for_stock_key(
500 &self,
501 stock_id: u64,
502 broker_id: Option<std::num::NonZeroU32>,
503 ) -> Vec<(
504 futu_core::qot_stock_key::QotSecurityKey,
505 Arc<CachedSecurityInfo>,
506 )> {
507 let bare = self.quote_push_targets_for_stock_id(stock_id);
511 bare.into_iter()
512 .map(|(public_sec_key, info)| {
513 let key = match broker_id {
514 Some(nz) => futu_core::qot_stock_key::QotSecurityKey::from_broker_id(
515 public_sec_key,
516 stock_id,
517 nz.get(),
518 ),
519 None => futu_core::qot_stock_key::QotSecurityKey::no_broker(
520 public_sec_key,
521 stock_id,
522 ),
523 };
524 (key, info)
525 })
526 .collect()
527 }
528
529 #[deprecated(
540 since = "1.4.106",
541 note = "use upsert_full_security_info / upsert_basic_security_info / delete_security_info"
542 )]
543 pub fn set_security_info(&self, key: &str, info: CachedSecurityInfo) {
544 if info.source.is_complete() {
545 self.upsert_full_security_info(key, info);
546 } else {
547 self.upsert_basic_security_info(key, info);
548 }
549 }
550
551 pub fn get_security_info(&self, key: &str) -> Option<CachedSecurityInfo> {
552 self.get_security_info_arc(key)
553 .map(|info| info.as_ref().clone())
554 }
555
556 pub fn get_security_info_arc(&self, key: &str) -> Option<Arc<CachedSecurityInfo>> {
557 self.securities.get(key).map(|v| Arc::clone(v.value()))
558 }
559
560 pub fn security_id_for_key(&self, key: &str) -> Option<u64> {
561 self.get_security_info(key)
562 .map(|info| info.stock_id)
563 .filter(|stock_id| *stock_id > 0)
564 }
565
566 pub fn security_info_snapshot_matching(
567 &self,
568 mut predicate: impl FnMut(&CachedSecurityInfo) -> bool,
569 ) -> Vec<CachedSecurityInfo> {
570 let mut seen_positive_ids = HashSet::new();
571 self.securities
572 .iter()
573 .filter_map(|entry| {
574 let info = entry.value();
575 if !predicate(info.as_ref())
576 || (info.stock_id > 0 && !seen_positive_ids.insert(info.stock_id))
577 {
578 return None;
579 }
580 Some(info.as_ref().clone())
581 })
582 .collect()
583 }
584
585 pub fn security_key_by_stock_id(&self, stock_id: u64) -> Option<String> {
586 self.id_to_key.get(&stock_id).map(|key| key.value().clone())
587 }
588
589 pub fn get_security_info_by_stock_id(&self, stock_id: u64) -> Option<CachedSecurityInfo> {
591 self.securities_by_stock_id
592 .get(&stock_id)
593 .map(|info| info.as_ref().clone())
594 }
595
596 pub fn get_security_info_by_stock_id_trigger_refresh(
597 &self,
598 stock_id: u64,
599 ) -> Option<CachedSecurityInfo> {
600 let info = self.get_security_info_by_stock_id(stock_id)?;
601 if info.needs_mkt_id_refresh()
602 && let Some(key) = self.security_key_by_stock_id(stock_id)
603 {
604 self.mark_stale_mkt_id(&key);
605 }
606 Some(info)
607 }
608
609 pub fn add_warrant_owner(&self, warrant_stock_id: u64, owner_stock_id: u64) {
614 if owner_stock_id == 0 {
615 return;
616 }
617 if let Ok(mut map) = self.owner_to_warrants.write() {
618 map.entry(owner_stock_id)
619 .or_default()
620 .insert(warrant_stock_id);
621 }
622 }
623
624 #[must_use]
636 pub fn search_warrants_by_owner(&self, owner_stock_id: u64) -> Vec<u64> {
637 match self.owner_to_warrants.read() {
638 Ok(map) => map
639 .get(&owner_stock_id)
640 .map(|set| set.iter().copied().collect())
641 .unwrap_or_default(),
642 _ => Vec::new(),
643 }
644 }
645}
646
647impl Default for StaticDataCache {
648 fn default() -> Self {
649 Self::new()
650 }
651}
652
653#[cfg(test)]
654mod tests;