futu_cache/

static_data.rs

// 静态数据缓存：股票列表、经纪商、节假日、停牌

use dashmap::DashMap;
use std::collections::HashSet;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, RwLock};

mod readiness;
mod types;

pub use readiness::{StaticDataReadiness, StockListSyncStatus};
pub use types::{
    CachedPlateInfo, CachedSecurityInfo, CachedTradeDate, CryptoPairInfo, CryptoTradeConfig,
    OptionContractInfo, SecurityInfoSource,
};

/// 静态数据缓存
pub struct StaticDataCache {
    /// 股票静态信息: "market_code" → info.
    ///
    /// 三索引写入只能通过 `upsert_full_security_info` /
    /// `upsert_basic_security_info` / `delete_security_info`，避免 caller
    /// 绕过 `id_to_key` / `owner_to_warrants` 维护。
    securities: DashMap<String, Arc<CachedSecurityInfo>>,
    /// stock_id → "market_code" key (反向映射，用于推送时查找)
    ///
    id_to_key: DashMap<u64, String>,
    /// 期货主连/连续合约 push 路由别名: real/origin stock_id → main-link sec_key set.
    ///
    /// Backend 的实时 push 可能以真实月份合约 stock_id 下发，而客户端订阅的是
    /// `HSImain` / `NQmain` 这类主连 symbol。C++ 的 QotSubscribe 用 stock_id
    /// 级别的主连关系做回投；Rust 这里保留同样的数据驱动关系，来源仅限
    /// stock-list 下发的 `origin_id` / `zhuli_id` 字段，不按 code 字符串特判。
    future_main_link_aliases: DashMap<u64, HashSet<String>>,
    /// option stock_id -> contract metadata from CMD20106 `OptionResultInfo`.
    option_contracts: DashMap<u64, OptionContractInfo>,
    /// Crypto sec_key -> 货币对元数据 (`cc_origin` / `cc_destination`)。
    crypto_pairs: DashMap<String, CryptoPairInfo>,
    /// `(broker_id, symbol, exchange)` -> crypto 交易配置。
    crypto_trade_configs: DashMap<String, CryptoTradeConfig>,
    /// 交易日: "market:year-month" → Vec<TradeDate>
    pub trade_dates: DashMap<String, Vec<CachedTradeDate>>,
    /// 板块: "market:plate_type" → Vec<PlateInfo>
    pub plates: DashMap<String, Vec<CachedPlateInfo>>,
    /// 窝轮正股 owner_id → 该正股对应的所有窝轮 stock_id 集合
    ///
    /// **v1.4.106 codex 1148 F6**: value 从 `Vec<u64>` 改为 `HashSet<u64>` 防
    /// 重复 (SQLite reload + stock-list re-sync 同 warrant 多次 push 不会再重复).
    /// stock-list `delete_flag` 时**反向索引清理**: 旧 owner 下移除旧 warrant
    /// (`delete_security_info` 内部维护), update 时也维护。
    owner_to_warrants: RwLock<std::collections::HashMap<u64, HashSet<u64>>>,

    /// v1.4.89 P2-A: 需要 mkt_id refresh 的 cache key 集合.
    ///
    /// Callers `get_security_info_trigger_refresh` 在返 info 前检查
    /// `info.needs_mkt_id_refresh()`, 是则 mark key 到这里. 背景 worker
    /// (gateway bridge) 定期 `drain_stale_mkt_ids()` 批量 CMD 20106 refresh.
    ///
    /// 用 DashMap<String, ()> 替代 HashSet<String> 免 lock 竞争.
    pub stale_mkt_ids: DashMap<String, ()>,

    /// v1.4.89 P2-A: mkt_id refresh 统计计数, 用于 metrics 观察.
    ///
    /// - `mkt_id_refresh_marked_total`: 累积 mark stale 次数
    /// - `mkt_id_refresh_done_total`: 累积 backend CMD 20106 成功 refresh 次数
    /// - `mkt_id_refresh_failed_total`: 累积 refresh failure 次数
    pub mkt_id_refresh_marked_total: AtomicU64,
    pub mkt_id_refresh_done_total: AtomicU64,
    pub mkt_id_refresh_failed_total: AtomicU64,

    /// Stock-list SQLite bootstrap / backend sync readiness diagnostics.
    ///
    /// C++ OpenD opens and validates its SecList DB before APIServer becomes
    /// ready; Rust also needs a concrete readiness signal so static-data
    /// comparisons do not race the background CMD6746 worker on cold start.
    stock_list_first_sync_done: AtomicBool,
    stock_list_sync_started_total: AtomicU64,
    stock_list_sync_finished_total: AtomicU64,
    stock_list_sync_failed_total: AtomicU64,
    stock_list_sync_recoverable_retry_total: AtomicU64,
    stock_list_sync_zero_delta_total: AtomicU64,
    stock_list_converged: AtomicBool,
    stock_list_sync_last_version: AtomicU64,
    stock_list_sync_last_total_stocks: AtomicU64,
    stock_list_sync_last_cached_count: AtomicU64,
    stock_list_sync_last_finished_ms: AtomicU64,
}

impl StaticDataCache {
    pub fn new() -> Self {
        Self {
            securities: DashMap::new(),
            id_to_key: DashMap::new(),
            future_main_link_aliases: DashMap::new(),
            option_contracts: DashMap::new(),
            crypto_pairs: DashMap::new(),
            crypto_trade_configs: DashMap::new(),
            trade_dates: DashMap::new(),
            plates: DashMap::new(),
            owner_to_warrants: RwLock::new(std::collections::HashMap::new()),
            stale_mkt_ids: DashMap::new(),
            mkt_id_refresh_marked_total: AtomicU64::new(0),
            mkt_id_refresh_done_total: AtomicU64::new(0),
            mkt_id_refresh_failed_total: AtomicU64::new(0),
            stock_list_first_sync_done: AtomicBool::new(false),
            stock_list_sync_started_total: AtomicU64::new(0),
            stock_list_sync_finished_total: AtomicU64::new(0),
            stock_list_sync_failed_total: AtomicU64::new(0),
            stock_list_sync_recoverable_retry_total: AtomicU64::new(0),
            stock_list_sync_zero_delta_total: AtomicU64::new(0),
            stock_list_converged: AtomicBool::new(false),
            stock_list_sync_last_version: AtomicU64::new(0),
            stock_list_sync_last_total_stocks: AtomicU64::new(0),
            stock_list_sync_last_cached_count: AtomicU64::new(0),
            stock_list_sync_last_finished_ms: AtomicU64::new(0),
        }
    }

    pub fn record_stock_list_sync_started(&self) {
        self.stock_list_sync_started_total
            .fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_stock_list_sync_finished(
        &self,
        version: u64,
        total_stocks: u64,
        cached_count: u64,
        finished_ms: u64,
    ) {
        self.stock_list_first_sync_done
            .store(true, Ordering::Release);
        // `converged` is a sticky cold-start readiness signal: after the first
        // zero-delta pass proves catch-up reached backend head, later normal
        // tiny deltas must not move static queries back to startup loading.
        if total_stocks == 0 {
            self.stock_list_sync_zero_delta_total
                .fetch_add(1, Ordering::Relaxed);
            self.stock_list_converged.store(true, Ordering::Release);
        }
        self.stock_list_sync_finished_total
            .fetch_add(1, Ordering::Relaxed);
        self.stock_list_sync_last_version
            .store(version, Ordering::Relaxed);
        self.stock_list_sync_last_total_stocks
            .store(total_stocks, Ordering::Relaxed);
        self.stock_list_sync_last_cached_count
            .store(cached_count, Ordering::Relaxed);
        self.stock_list_sync_last_finished_ms
            .store(finished_ms, Ordering::Relaxed);
    }

    pub fn record_stock_list_sync_failed(&self) {
        self.stock_list_sync_failed_total
            .fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_stock_list_sync_recoverable_retry(&self) {
        self.stock_list_sync_recoverable_retry_total
            .fetch_add(1, Ordering::Relaxed);
    }

    pub fn stock_list_sync_status(&self) -> StockListSyncStatus {
        let finished_ms = self
            .stock_list_sync_last_finished_ms
            .load(Ordering::Relaxed);
        StockListSyncStatus {
            first_sync_done: self.stock_list_first_sync_done.load(Ordering::Acquire),
            started_total: self.stock_list_sync_started_total.load(Ordering::Relaxed),
            finished_total: self.stock_list_sync_finished_total.load(Ordering::Relaxed),
            failed_total: self.stock_list_sync_failed_total.load(Ordering::Relaxed),
            recoverable_retry_total: self
                .stock_list_sync_recoverable_retry_total
                .load(Ordering::Relaxed),
            zero_delta_total: self
                .stock_list_sync_zero_delta_total
                .load(Ordering::Relaxed),
            converged: self.stock_list_converged.load(Ordering::Acquire),
            last_version: self.stock_list_sync_last_version.load(Ordering::Relaxed),
            last_total_stocks: self
                .stock_list_sync_last_total_stocks
                .load(Ordering::Relaxed),
            last_cached_count: self
                .stock_list_sync_last_cached_count
                .load(Ordering::Relaxed),
            last_finished_ms: (finished_ms > 0).then_some(finished_ms),
        }
    }

    pub fn security_info_count(&self) -> usize {
        self.securities.len()
    }

    pub fn stock_list_readiness(&self) -> StaticDataReadiness {
        self.stock_list_sync_status()
            .readiness_for_security_count(self.security_info_count())
    }

    /// v1.4.89 P2-A: 取 cache info 同时机会性 mark stale (若 mkt_id=0).
    ///
    /// 返 Some(info) 如果 cache hit (无论是否 stale). 返 None 如果 miss.
    ///
    /// 调 `info.needs_mkt_id_refresh()` 判 stale → mark `stale_mkt_ids`,
    /// bump `mkt_id_refresh_marked_total` counter. 用 DashMap::insert 幂等
    /// (同 key 重入不 double mark 但会 bump counter — 可接受).
    ///
    /// 替代 `get_security_info` 的推荐路径; 老 method 保留作 lookup-only 接口.
    pub fn get_security_info_trigger_refresh(&self, key: &str) -> Option<CachedSecurityInfo> {
        let info = self.get_security_info(key)?;
        if info.needs_mkt_id_refresh() {
            self.mark_stale_mkt_id(key);
        }
        Some(info)
    }

    /// v1.4.89 P2-A: 显式 mark key 需要 mkt_id refresh.
    ///
    /// 幂等: 同 key 可重入. Counter `mkt_id_refresh_marked_total` 每次都 bump
    /// (用作 metrics 观察 heuristic fallback 触发频率).
    pub fn mark_stale_mkt_id(&self, key: &str) {
        self.stale_mkt_ids.insert(key.to_string(), ());
        self.mkt_id_refresh_marked_total
            .fetch_add(1, Ordering::Relaxed);
    }

    /// v1.4.89 P2-A: drain 所有 stale keys, 清空集合, 返 Vec (给 backend worker
    /// 批量 CMD 20106 refresh).
    ///
    /// 背景 worker 用法 (伪码):
    /// ```text
    /// loop {
    ///     sleep(Duration::from_secs(60)).await;
    ///     let stale = cache.drain_stale_mkt_ids();
    ///     if stale.is_empty() { continue; }
    ///     for chunk in stale.chunks(50) {
    ///         // CMD 20106 SecuritiesReq for chunk
    ///         // on success: cache.update_mkt_id(key, new_mkt_id)
    ///         //              + cache.record_mkt_id_refresh_done()
    ///         // on failure: cache.record_mkt_id_refresh_failed()
    ///     }
    /// }
    /// ```
    pub fn drain_stale_mkt_ids(&self) -> Vec<String> {
        let keys: Vec<String> = self.stale_mkt_ids.iter().map(|e| e.key().clone()).collect();
        for k in &keys {
            self.stale_mkt_ids.remove(k);
        }
        keys
    }

    /// v1.4.89 P2-A: 更新已 cache row 的 mkt_id (refresh success 时调).
    ///
    /// 只改 mkt_id 字段, 其他字段保留 (info 可能有 SQLite 里更精准的 lot_size /
    /// list_time 等). 若 key 不在 cache (已被 evict), no-op.
    pub fn update_mkt_id(&self, key: &str, new_mkt_id: u32) -> bool {
        if let Some(mut entry) = self.securities.get_mut(key) {
            Arc::make_mut(&mut entry).mkt_id = new_mkt_id;
            self.mkt_id_refresh_done_total
                .fetch_add(1, Ordering::Relaxed);
            true
        } else {
            false
        }
    }

    /// v1.4.89 P2-A: 记录 refresh failure (不改 cache row, 让下次 drain 重试).
    pub fn record_mkt_id_refresh_failed(&self) {
        self.mkt_id_refresh_failed_total
            .fetch_add(1, Ordering::Relaxed);
    }

    /// v1.4.89 P2-A: 当前 stale keys 数 (给 observability / debug).
    #[must_use]
    pub fn stale_mkt_ids_count(&self) -> usize {
        self.stale_mkt_ids.len()
    }

    /// v1.4.106 codex 1148 F9 (P3): 统一写入口 — 完整静态行 (`StockListFull` /
    /// `Bootstrap` source)。同步维护 `securities` + `id_to_key` + 若有 owner
    /// 还更新 `owner_to_warrants`。**自动 dedup**: 已有同 key 但 `warrnt_stock_owner`
    /// 变化时, 旧 owner 下移除该 warrant id, 新 owner 下添加。
    ///
    /// 替代生产代码里手动调 `securities.insert()` + `id_to_key.insert()` +
    /// `add_warrant_owner()` 三步骤的 pattern。
    ///
    /// **不允许** caller 把不完整的 source 标 `StockListFull`(若 `info.source ==
    /// OnDemandBasic`, 用 `upsert_basic_security_info` 而非本 fn)。
    pub fn upsert_full_security_info(&self, key: &str, info: CachedSecurityInfo) {
        debug_assert!(
            info.source.is_complete(),
            "upsert_full_security_info called with non-complete source ({:?})",
            info.source
        );
        self.upsert_with_owner_index_maintenance(key, info);
    }

    /// 写入 stock-list 下发的 crypto 货币对元数据。
    pub fn upsert_crypto_pair_info(&self, key: &str, pair: CryptoPairInfo) {
        if pair.cc_origin.is_empty() && pair.cc_destination.is_empty() {
            self.crypto_pairs.remove(key);
        } else {
            self.crypto_pairs.insert(key.to_string(), pair);
        }
    }

    /// Cache option contract metadata from CMD20106 `OptionResultInfo`.
    pub fn set_option_contract_info(&self, stock_id: u64, info: OptionContractInfo) {
        if stock_id == 0 {
            return;
        }
        self.option_contracts.insert(stock_id, info);
    }

    /// Read option contract metadata by option stock_id.
    pub fn get_option_contract_info_by_stock_id(
        &self,
        stock_id: u64,
    ) -> Option<OptionContractInfo> {
        self.option_contracts
            .get(&stock_id)
            .map(|entry| *entry.value())
    }

    /// 读取 crypto 货币对元数据。
    pub fn get_crypto_pair_info(&self, key: &str) -> Option<CryptoPairInfo> {
        self.crypto_pairs.get(key).map(|v| v.clone())
    }

    fn crypto_trade_config_key(broker_id: u32, symbol: &str, exchange: &str) -> String {
        format!(
            "{broker_id}:{}:{}",
            symbol.trim().to_ascii_uppercase(),
            exchange.trim().to_ascii_uppercase()
        )
    }

    /// 用 backend CMD20102 拉回的配置替换某个 broker 的 crypto trade config。
    pub fn set_crypto_trade_configs_for_broker(
        &self,
        broker_id: u32,
        configs: Vec<CryptoTradeConfig>,
    ) {
        let prefix = format!("{broker_id}:");
        self.crypto_trade_configs
            .retain(|key, _| !key.starts_with(&prefix));
        for config in configs {
            if config.symbol.trim().is_empty() || config.exchange.trim().is_empty() {
                continue;
            }
            let key = Self::crypto_trade_config_key(broker_id, &config.symbol, &config.exchange);
            self.crypto_trade_configs.insert(key, config);
        }
    }

    /// 查询某个 crypto symbol 的交易配置。
    pub fn get_crypto_trade_config(
        &self,
        broker_id: u32,
        symbol: &str,
        exchange: &str,
    ) -> Option<CryptoTradeConfig> {
        let key = Self::crypto_trade_config_key(broker_id, symbol, exchange);
        self.crypto_trade_configs.get(&key).map(|v| v.clone())
    }

    /// v1.4.106 codex 1148 F9 (P3): 统一写入口 — 部分静态行 (`OnDemandBasic`
    /// source)。同步维护 `securities` + `id_to_key`, **不动** `owner_to_warrants`
    /// (因为 OnDemandBasic 不含 `warrnt_stock_owner` 字段, value 必为 0)。
    ///
    /// `info.source` 必须是 `OnDemandBasic` (debug_assert)。
    pub fn upsert_basic_security_info(&self, key: &str, info: CachedSecurityInfo) {
        debug_assert!(
            !info.source.is_complete(),
            "upsert_basic_security_info called with complete source ({:?}); use upsert_full",
            info.source
        );
        debug_assert_eq!(
            info.warrnt_stock_owner, 0,
            "OnDemandBasic must have warrnt_stock_owner=0 (caller didn't query the field)"
        );
        // 不维护 owner_to_warrants (basic 没这个字段).
        // 但仍需检查既有完整行的 owner 是否会被 basic 错误覆盖.
        // 策略: 如果 key 已有 StockListFull / Bootstrap 行, 不让 basic 行覆盖 (full
        // 数据更完整). 这处理 case "subscribe on-demand 后, stock-list sync 来时
        // 应该 prevail; 反过来不行".
        let old_info = if let Some(existing) = self.securities.get(key) {
            if existing.is_complete() {
                tracing::debug!(
                    key,
                    "upsert_basic_security_info skipped: existing complete row prevails"
                );
                return;
            }
            Some(Arc::clone(existing.value()))
        } else {
            None
        };
        if let Some(old_info) = old_info {
            self.remove_future_main_link_aliases(key, &old_info);
        }
        self.securities
            .insert(key.to_string(), Arc::new(info.clone()));
        self.id_to_key.insert(info.stock_id, key.to_string());
        self.add_future_main_link_aliases(key, &info);
    }

    /// v1.4.106 codex 1148 F9 (P3): 删除 cache row + 同步清三个索引
    /// (`securities`, `id_to_key`, `owner_to_warrants`)。
    ///
    /// 用于 stock-list `delete_flag = true` 场景。
    /// 返 `true` 如果 row 存在并被删除, `false` 如果 stock_id 不在 `id_to_key`。
    pub fn delete_security_info(&self, stock_id: u64) -> bool {
        let Some((_, key)) = self.id_to_key.remove(&stock_id) else {
            return false;
        };
        self.option_contracts.remove(&stock_id);
        // 拿被删 row 的 owner (用于反向索引清理), 如果 key 已不在 securities, owner = 0
        let old_info = self.securities.remove(&key).map(|(_, info)| info);
        let old_owner = old_info.as_ref().map(|r| r.warrnt_stock_owner).unwrap_or(0);
        if let Some(old_info) = old_info.as_ref() {
            self.remove_future_main_link_aliases(&key, old_info);
        }
        self.crypto_pairs.remove(&key);
        // F6: stock-list delete 时把 warrant 从 old_owner 反向索引里清掉
        if old_owner != 0
            && let Ok(mut map) = self.owner_to_warrants.write()
            && let Some(set) = map.get_mut(&old_owner)
        {
            set.remove(&stock_id);
            if set.is_empty() {
                map.remove(&old_owner);
            }
        }
        // F6: 该 stock_id 自己也可能是某 owner — 清掉它作为 owner 的 entry
        if let Ok(mut map) = self.owner_to_warrants.write() {
            map.remove(&stock_id);
        }
        true
    }

    /// 内部 helper: F9 unified upsert with owner-index maintenance (F6).
    fn upsert_with_owner_index_maintenance(&self, key: &str, info: CachedSecurityInfo) {
        // 先看老 row 是否存在 + 旧 owner 是什么 (F6: 如果 owner 变了要清旧索引)
        let old_info = self.securities.get(key).map(|r| Arc::clone(r.value()));
        let old_owner = old_info.as_ref().map(|r| r.warrnt_stock_owner).unwrap_or(0);
        let new_owner = info.warrnt_stock_owner;

        if let Some(old) = old_info.as_ref() {
            self.remove_future_main_link_aliases(key, old);
        }

        // 写 securities + id_to_key
        let stock_id = info.stock_id;
        self.securities
            .insert(key.to_string(), Arc::new(info.clone()));
        self.id_to_key.insert(stock_id, key.to_string());
        self.add_future_main_link_aliases(key, &info);

        // F6: 维护反向索引
        if old_owner != new_owner {
            // owner 变了 (含 0→X / X→Y / X→0)
            if let Ok(mut map) = self.owner_to_warrants.write() {
                if old_owner != 0
                    && let Some(set) = map.get_mut(&old_owner)
                {
                    set.remove(&stock_id);
                    if set.is_empty() {
                        map.remove(&old_owner);
                    }
                }
                if new_owner != 0 {
                    map.entry(new_owner).or_default().insert(stock_id);
                }
            }
        } else if new_owner != 0 {
            // owner 未变, 但同一 owner 下重 add (idempotent due to HashSet).
            if let Ok(mut map) = self.owner_to_warrants.write() {
                map.entry(new_owner).or_default().insert(stock_id);
            }
        }
    }

    fn future_main_link_target_ids(info: &CachedSecurityInfo) -> Vec<u64> {
        let mut ids = Vec::with_capacity(2);
        for target in [info.future_origin_id, info.zhuli_id] {
            if target != 0 && target != info.stock_id && !ids.contains(&target) {
                ids.push(target);
            }
        }
        ids
    }

    fn add_future_main_link_aliases(&self, key: &str, info: &CachedSecurityInfo) {
        for target in Self::future_main_link_target_ids(info) {
            self.future_main_link_aliases
                .entry(target)
                .or_default()
                .insert(key.to_string());
        }
    }

    fn remove_future_main_link_aliases(&self, key: &str, info: &CachedSecurityInfo) {
        for target in Self::future_main_link_target_ids(info) {
            if let Some(mut aliases) = self.future_main_link_aliases.get_mut(&target) {
                aliases.remove(key);
                let empty = aliases.is_empty();
                drop(aliases);
                if empty {
                    self.future_main_link_aliases.remove(&target);
                }
            }
        }
    }

    /// 查询某个 backend push stock_id 对应的主连/连续合约 sec_key 别名。
    ///
    /// 只返回 stock-list 明确下发 `origin_id` / `zhuli_id` 关系的 key；不做
    /// `HSImain` 等字符串启发式。调用方通常先按 `id_to_key` 处理真实合约，
    /// 再把这里返回的 main-link key 一并投递。
    #[must_use]
    pub fn get_future_main_link_alias_keys(&self, stock_id: u64) -> Vec<String> {
        let Some(aliases) = self.future_main_link_aliases.get(&stock_id) else {
            return Vec::new();
        };
        let mut keys: Vec<String> = aliases.iter().cloned().collect();
        keys.sort();
        keys
    }

    /// 查询 backend push stock_id 的所有 quote 投递目标。
    ///
    /// 顺序保持为：真实 stock_id 对应 key（如有）优先，然后是 stock-list
    /// `origin_id` / `zhuli_id` 下发的主连/连续合约别名 key。调用方不再直接
    /// 读取 `id_to_key` 和 `future_main_link_aliases` 两个索引，避免 alias 逻辑
    /// 分散在 push parser 里。
    #[must_use]
    pub fn quote_push_targets_for_stock_id(
        &self,
        stock_id: u64,
    ) -> Vec<(String, Arc<CachedSecurityInfo>)> {
        let mut targets = Vec::new();

        if let Some(sec_key_ref) = self.id_to_key.get(&stock_id) {
            let sec_key = sec_key_ref.clone();
            drop(sec_key_ref);
            if let Some(info) = self.get_security_info_arc(&sec_key) {
                targets.push((sec_key, info));
            }
        }

        for alias_key in self.get_future_main_link_alias_keys(stock_id) {
            if targets.iter().any(|(key, _)| key == &alias_key) {
                continue;
            }
            if let Some(info) = self.get_security_info_arc(&alias_key) {
                targets.push((alias_key, info));
            }
        }

        targets
    }

    /// **v1.4.110 Phase 2 Slice 5**: broker-aware 推送投递目标查询.
    ///
    /// 对应 push parser 从 `SecurityQuote.broker_id` 重建 broker-aware
    /// `QotStockKey` 的路径 (对齐 C++ `NNBiz_Qot_PushQot.cpp:220-269`).
    ///
    /// 语义:
    /// - `broker_id = None` (C++ `m_hasBroker=false`): 只查 no-broker key,
    ///   返 `QotSecurityKey::no_broker(public_sec_key, stock_id)` 与
    ///   `quote_push_targets_for_stock_id` 等价
    /// - `broker_id = Some(N)` (C++ `m_hasBroker=true`): 沿 stock_id 反向
    ///   找到 public_sec_key, 再用 `QotSecurityKey::from_broker_id(...)`
    ///   构造 broker-aware key. 该 broker 下 cache 写入独立桶
    ///   `"market_code@b{N}"`, 不污染同 stock_id 其他 broker 的 cache.
    ///
    /// **Phase 2 默认**: backend 当前对普通股 push 仍不下发 broker_id (=None),
    /// 与升级前行为完全等价. crypto multi-broker push 会带 broker_id,
    /// Phase 3 reader caller (handler `GetBasicQot` 等) 替换走 `_broker`
    /// 版本后, broker-aware cache 才被消费.
    #[must_use]
    pub fn quote_push_targets_for_stock_key(
        &self,
        stock_id: u64,
        broker_id: Option<std::num::NonZeroU32>,
    ) -> Vec<(
        futu_core::qot_stock_key::QotSecurityKey,
        Arc<CachedSecurityInfo>,
    )> {
        // 先用 no-broker 路径查 stock_id → (public_sec_key, info) 列表
        // (id_to_key + future_main_link_aliases). broker_id 注入到返 key
        // 不改变 lookup 逻辑.
        let bare = self.quote_push_targets_for_stock_id(stock_id);
        bare.into_iter()
            .map(|(public_sec_key, info)| {
                let key = match broker_id {
                    Some(nz) => futu_core::qot_stock_key::QotSecurityKey::from_broker_id(
                        public_sec_key,
                        stock_id,
                        nz.get(),
                    ),
                    None => futu_core::qot_stock_key::QotSecurityKey::no_broker(
                        public_sec_key,
                        stock_id,
                    ),
                };
                (key, info)
            })
            .collect()
    }

    /// **deprecated**: 改用 `upsert_full_security_info` /
    /// `upsert_basic_security_info` 显式表达数据完整度。
    ///
    /// v1.4.111 codex legacy deep-dive follow-up: 保留此 public wrapper 兼容
    /// 既有 tests/bench/下游辅助代码，但不再直接写 `securities` 造成半索引行。
    /// `source.is_complete()` 走 full upsert, 否则走 basic upsert, 始终维护
    /// `id_to_key` / `owner_to_warrants` 与主表一致。
    ///
    /// Removal trigger: repo 内 tests/benches 全部迁移到显式 upsert/delete API，
    /// 且一个 minor release 内没有下游兼容反馈后删除。
    #[deprecated(
        since = "1.4.106",
        note = "use upsert_full_security_info / upsert_basic_security_info / delete_security_info"
    )]
    pub fn set_security_info(&self, key: &str, info: CachedSecurityInfo) {
        if info.source.is_complete() {
            self.upsert_full_security_info(key, info);
        } else {
            self.upsert_basic_security_info(key, info);
        }
    }

    pub fn get_security_info(&self, key: &str) -> Option<CachedSecurityInfo> {
        self.get_security_info_arc(key)
            .map(|info| info.as_ref().clone())
    }

    pub fn get_security_info_arc(&self, key: &str) -> Option<Arc<CachedSecurityInfo>> {
        self.securities.get(key).map(|v| Arc::clone(v.value()))
    }

    pub fn security_id_for_key(&self, key: &str) -> Option<u64> {
        self.get_security_info(key)
            .map(|info| info.stock_id)
            .filter(|stock_id| *stock_id > 0)
    }

    pub fn security_info_snapshot(&self) -> Vec<CachedSecurityInfo> {
        self.security_info_snapshot_matching(|_| true)
    }

    pub fn security_info_snapshot_matching(
        &self,
        mut predicate: impl FnMut(&CachedSecurityInfo) -> bool,
    ) -> Vec<CachedSecurityInfo> {
        self.securities
            .iter()
            .filter_map(|entry| {
                let info = entry.value();
                predicate(info.as_ref()).then(|| info.as_ref().clone())
            })
            .collect()
    }

    pub fn security_key_by_stock_id(&self, stock_id: u64) -> Option<String> {
        self.id_to_key.get(&stock_id).map(|key| key.value().clone())
    }

    /// 通过 stock_id 查找股票信息 (使用 id_to_key 反向映射)
    pub fn get_security_info_by_stock_id(&self, stock_id: u64) -> Option<CachedSecurityInfo> {
        let key = self.security_key_by_stock_id(stock_id)?;
        self.get_security_info(&key)
    }

    pub fn get_security_info_by_stock_id_trigger_refresh(
        &self,
        stock_id: u64,
    ) -> Option<CachedSecurityInfo> {
        let key = self.security_key_by_stock_id(stock_id)?;
        self.get_security_info_trigger_refresh(&key)
    }

    /// 添加窝轮→正股的映射关系
    ///
    /// **v1.4.106 codex 1148 F6**: HashSet 自动去重, 重复 add 同 (warrant, owner)
    /// 是 idempotent — SQLite reload + stock-list sync 不会重复入。
    pub fn add_warrant_owner(&self, warrant_stock_id: u64, owner_stock_id: u64) {
        if owner_stock_id == 0 {
            return;
        }
        if let Ok(mut map) = self.owner_to_warrants.write() {
            map.entry(owner_stock_id)
                .or_default()
                .insert(warrant_stock_id);
        }
    }

    /// 通过正股 ID 搜索该正股的所有窝轮
    ///
    /// **v1.4.106 codex 1148 F6**: 内部 HashSet, 返 Vec (call site backward
    /// compatible)。返序非确定 (HashSet 不保留 insertion order); call site 若
    /// 需要稳定序应自己 sort。
    ///
    /// v1.4.111 P2-1 Tier 3 audit comment: warrant lookup helper — empty Vec =
    /// "no warrants for this owner_stock_id" (legit "正股无窝轮"), 跟 C++
    /// `SearchWarrantsByOwner` empty 行为对齐. caller decide 怎么用 (display 0
    /// warrants 是合理). 非 silent-success risk (audit verified, essentials/
    /// 2026-05-27).
    #[must_use]
    pub fn search_warrants_by_owner(&self, owner_stock_id: u64) -> Vec<u64> {
        match self.owner_to_warrants.read() {
            Ok(map) => map
                .get(&owner_stock_id)
                .map(|set| set.iter().copied().collect())
                .unwrap_or_default(),
            _ => Vec::new(),
        }
    }
}

impl Default for StaticDataCache {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests;