futu_cache/

crypto_exchange_cache.rs

//! v1.4.110 codex QOT Phase 4 Slice 7: Crypto LV2 多交易所缓存.
//!
//! 对齐 C++ `APIServer/Business/Quote/QotRealTimeData.cpp` 三个 map +
//! `INNData_Qot_CryptoExchange::SetLv2RelatedExchange` 缓存写入:
//!
//! - **by_broker** (`m_mapCryptoStockBrokerExchange`): `(stock_id, broker_id) →
//!   Vec<CryptoExchangeInfo>` — 18012 response 写入, 表示该 broker 看该 stock
//!   有哪些 LV2 关联交易所. 通过 `GetLv2RelatedExchangeList(stKey)` 查询.
//! - **lv2_exchange_cache** (`m_mapLv2ExchangeCache`): `(stock_id, lv2_prob) →
//!   CachedOrderBook` — 单 exchange 的原始 LV2 摆盘缓存. 收到 push 时按
//!   `ExchangeCacheKey` 索引写入.
//! - **lv2_prob_to_brokers** (`m_mapLv2ProbToBrokers`): `(stock_id, lv2_prob)
//!   → HashSet<broker_id>` — 反向索引. 收到 exchange-level push 时, O(1) 找出
//!   所有受影响 broker, 然后 per-broker rebuild 调 `merge_multiple_order_book_caches`.
//!
//! ## 流程图
//!
//! ```text
//! GetOrderBookHandler: crypto LV2 第 1 次订阅
//!   → fetch_and_cache_exchanges (CMD18012, PT 过滤)
//!     → cache.set_lv2_related_exchange((stock_id, broker_id), Vec<info>)
//!       → 内部更新 by_broker + 重建 lv2_prob_to_brokers
//!     → resubscribe with prob2_v2 (level=60) for each lv2_prob
//!
//! Backend push exchange-level LV2 orderbook:
//!   → push_parser 识别 SBIT_US_LV2_ORDER (17) + sec_info.is_crypto + lv2_type
//!     → cache.set_lv2_exchange_cache((stock_id, lv2_prob), s2c)
//!     → cache.get_brokers_affected_by_lv2_prob((stock_id, lv2_prob)) → HashSet
//!       For each broker in HashSet:
//!         → cache.exchanges_for_broker((stock_id, broker_id)) → Vec
//!         → 收集所有 exchange caches → merge_multiple_order_book_caches(40)
//!         → qot_cache.update_order_book_broker(StockKey{stock_id, broker_id})
//!         → cold-cache wait notify (Slice 6c)
//! ```
//!
//! ## C++ 参照
//!
//! - `QotRealTimeData.cpp:918-1022` `ParseCryptoToExchangeCache` /
//!   `RebuildCryptoBrokerCache` / `MergeCryptoExchangesToBrokerCache`
//! - `QotRealTimeData.cpp:1024-1063` `UpdateLv2ProbToBrokersIndex`
//! - `NNBiz_Qot_CryptoExchange.cpp:131-141` `SetLv2RelatedExchange` 写完后
//!   触发 `IOMEvent::NotifyEvent(NN_OMEvent_Qot_CryptoExchange_IndexUpdate)` +
//!   `ReSubCryptoOrderBook`.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use dashmap::DashMap;

use crate::qot_cache::CachedOrderBook;

/// `(stock_id, lv2_prob)` — exchange-level cache key.
///
/// 对齐 C++ `ExchangeCacheKey(nStockID, nLv2Prob)` (QotRealTimeData.cpp).
pub type ExchangeCacheKey = (u64, i32);

/// `(stock_id, broker_id)` — broker-level exchange list key.
pub type BrokerExchangeKey = (u64, u32);

/// Daemon-side snapshot of one entry in CMD18012 `ExchangeInfo`.
///
/// 对齐 C++ `Ndt_Qot_CryptoExchangeInfo` (NNBiz_Qot_CryptoExchange.cpp:55-58).
///
/// 在 cache crate 单独定义 (而非引用 futu-backend) 是为了避免 futu-cache →
/// futu-backend 反向依赖 (C++ 也分 NNData / NNBiz / APIServer 三层, 类似关注点分离).
#[derive(Debug, Clone, PartialEq)]
pub struct CryptoExchangeInfo {
    /// 订阅位 prob — `USLV2OrderSubProb.us_lv2_order_type`.
    pub lv2_prob: i32,
    /// 交易所内部 name (e.g. "BINANCE", "OKX", "PT").
    pub exchange_name: String,
    /// 上市交易所标识 (FTAPI listed_exchange 字段).
    pub listed_exchange: String,
    /// 是否默认选中 (UI 显示用, daemon 一般不消费).
    pub is_pick: bool,
}

/// Immutable request-level view of one exact `(stock_id, broker_id)` fact.
///
/// The writer publishes version, exact vector, and aggregate as one DashMap value.
#[derive(Debug, Clone, PartialEq)]
pub struct CryptoExchangeFact {
    pub version: u64,
    pub exchanges: Arc<[CryptoExchangeInfo]>,
    pub positive_lv2_prob_sum: Option<i32>,
}

#[derive(Debug, Clone, Default)]
pub struct CryptoExchangeFactSnapshot {
    facts: HashMap<BrokerExchangeKey, CryptoExchangeFact>,
}

impl CryptoExchangeFactSnapshot {
    pub fn for_broker(&self, stock_id: u64, broker_id: u32) -> Option<&CryptoExchangeFact> {
        self.facts.get(&(stock_id, broker_id))
    }
}

/// Crypto LV2 多交易所缓存. 包装 3 个 DashMap + 反向索引重建逻辑.
///
/// **线程安全**: 内部全用 DashMap, 任意线程可同时 read/write.
/// **lifecycle**: 整个 daemon lifetime, 由 `GatewayBridge::crypto_exchange_cache`
/// 持有 Arc.
#[derive(Debug, Default)]
pub struct CryptoExchangeCache {
    /// `(stock_id, broker_id) → exchange list (from CMD18012 response)`.
    by_broker: DashMap<BrokerExchangeKey, CryptoExchangeFact>,

    /// `(stock_id, lv2_prob) → orderbook (exchange-level, raw from push)`.
    pub lv2_exchange_cache: DashMap<ExchangeCacheKey, CachedOrderBook>,

    /// `(stock_id, lv2_prob) → set<broker_id>` — 反向索引,
    /// O(1) 查询受 exchange-level push 影响的 broker 集.
    pub lv2_prob_to_brokers: DashMap<ExchangeCacheKey, HashSet<u32>>,

    /// v1.4.110 R6-5: 串行化所有"写 `by_broker` + 重建反向索引"的写路径
    /// (`set_lv2_related_exchange` / `clear_stock` / `clear_stock_broker`).
    ///
    /// 反向索引重建要遍历 `by_broker` 全表算 `new_index`; 两个并发 writer 各自
    /// 遍历会拿到不一致的 DashMap 快照, 一个 writer 的 `retain` 可能误删另一个
    /// writer 刚 `insert` 的 prob entry → 该 prob 反向索引永久丢失 → 之后 crypto
    /// LV2 push 对该 broker silent miss (坑 #45). R4-2 只修了"新 prob entry 不
    /// 经历空态", 没修这条跨线程 stale-retain (坑 #44: 同一反向索引并发 invariant
    /// 第 2 次复发 → 根治). 这把锁让 "insert by_broker + 重建索引" 成为真临界区.
    /// writer 都是低频 first-sub / unsub 路径, 锁零实际性能影响; reader
    /// (`get_brokers_affected_by_lv2_prob` 等) 不取锁, 仍走 DashMap lock-free 读.
    rebuild_lock: parking_lot::Mutex<()>,

    /// Monotonic fact version assigned inside `rebuild_lock`.
    next_fact_version: AtomicU64,
}

impl CryptoExchangeCache {
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// 设置 `(stock_id, broker_id)` 对应的 exchange 列表 (CMD18012 response).
    ///
    /// 副作用:
    /// 1. 更新 `by_broker[(stock_id, broker_id)] = exchanges`
    /// 2. 重建 `lv2_prob_to_brokers[(stock_id, *)]` 该 stock 的反向索引
    ///    (先清旧 entries 该 stock 的所有 prob → 再按本次 + 其他 broker 重建)
    ///
    /// 对齐 C++ `INNData_Qot_CryptoExchange::SetLv2RelatedExchange(stKey, vExchanges)` +
    /// `UpdateLv2ProbToBrokersIndex(nStockID)`.
    pub fn set_lv2_related_exchange(
        &self,
        stock_id: u64,
        broker_id: u32,
        exchanges: Vec<CryptoExchangeInfo>,
    ) -> u64 {
        // v1.4.110 R6-5: 整个 "写 by_broker + 重建反向索引" 在 rebuild_lock 内
        // 串行化 (见 `rebuild_lock` 字段注释).
        let _rebuild_guard = self.rebuild_lock.lock();
        let version = self.next_fact_version.fetch_add(1, Ordering::Relaxed) + 1;
        let exchanges: Arc<[CryptoExchangeInfo]> = Arc::from(exchanges.into_boxed_slice());
        let positive_lv2_prob_sum = positive_lv2_prob_sum(&exchanges);
        self.by_broker.insert(
            (stock_id, broker_id),
            CryptoExchangeFact {
                version,
                exchanges,
                positive_lv2_prob_sum,
            },
        );
        self.rebuild_reverse_index_for_stock(stock_id);
        tracing::debug!(
            stock_id,
            broker_id,
            "v1.4.110 audit Phase 4 Slice 7: set_lv2_related_exchange + rebuild reverse index"
        );
        version
    }

    /// 重建某 stock 的 `lv2_prob_to_brokers` 反向索引 (遍历 `by_broker` 当前
    /// 全部 `(stock_id, *)` entry 重算 prob → brokers 映射).
    ///
    /// **caller 必须持 `rebuild_lock`** —— 本 fn 遍历 `by_broker` 算 `new_index`,
    /// 多个并发 caller 不串行化会拿到不一致快照, 一个 caller 的 (c) `retain`
    /// 可能误删另一个 caller 刚 (b) `insert` 的 prob (见 R6-5 / 坑 #44).
    ///
    /// v1.4.110 codex audit Round4 R4-2: 不用 "retain 清旧 → for 重建" 两步法 ——
    /// 两步之间有 empty window: 并发 crypto LV2 push 在此间隙调
    /// `get_brokers_affected_by_lv2_prob` 会看到空索引 → affected_brokers 空 →
    /// push 被静默丢 (pitfall #45). 改 overwrite-then-remove:
    ///   (a) 本地算出该 stock 的新 prob → brokers 映射;
    ///   (b) 对每个新 prob `insert` 原子覆盖 —— 已存在的 prob entry 从不经历
    ///       空态, reader 永远看到 old-set / new-set 二选一;
    ///   (c) 再 `retain` 删掉本 stock 不在新映射里的旧 prob (这些 prob 已无
    ///       broker 订阅, push 本就该丢).
    fn rebuild_reverse_index_for_stock(&self, stock_id: u64) {
        let mut new_index: HashMap<i32, HashSet<u32>> = HashMap::new();
        for entry in self.by_broker.iter() {
            let (sid, bid) = entry.key();
            if *sid != stock_id {
                continue;
            }
            for info in entry.value().exchanges.iter() {
                new_index.entry(info.lv2_prob).or_default().insert(*bid);
            }
        }
        let new_probs: HashSet<i32> = new_index.keys().copied().collect();
        // (b) 原子覆盖每个新 prob.
        for (prob, brokers) in new_index.into_iter() {
            let cache_key: ExchangeCacheKey = (stock_id, prob);
            self.lv2_prob_to_brokers.insert(cache_key, brokers);
        }
        // (c) 删掉本 stock 不在新映射里的旧 prob.
        self.lv2_prob_to_brokers
            .retain(|key, _| key.0 != stock_id || new_probs.contains(&key.1));
    }

    /// 取某 broker 看某 stock 的 exchange 列表.
    pub fn exchanges_for_broker(
        &self,
        stock_id: u64,
        broker_id: u32,
    ) -> Option<Vec<CryptoExchangeInfo>> {
        self.by_broker
            .get(&(stock_id, broker_id))
            .map(|fact| fact.exchanges.to_vec())
    }

    /// Clone all exact broker facts while holding the existing writer
    /// serialization lock, so every consumer in one request sees one immutable
    /// view.
    pub fn fact_snapshot(&self) -> CryptoExchangeFactSnapshot {
        let _rebuild_guard = self.rebuild_lock.lock();
        let facts = self
            .by_broker
            .iter()
            .map(|entry| (*entry.key(), entry.value().clone()))
            .collect();
        CryptoExchangeFactSnapshot { facts }
    }

    /// 写 exchange-level cache (单个 exchange 的原始 LV2 摆盘).
    /// 对齐 C++ `m_mapLv2ExchangeCache[cacheKey] = pbOrderBook` (QotRealTimeData.cpp:938).
    pub fn set_lv2_exchange_cache(&self, stock_id: u64, lv2_prob: i32, orderbook: CachedOrderBook) {
        self.lv2_exchange_cache
            .insert((stock_id, lv2_prob), orderbook);
    }

    /// 取 exchange-level cache.
    pub fn get_lv2_exchange_cache(&self, stock_id: u64, lv2_prob: i32) -> Option<CachedOrderBook> {
        self.lv2_exchange_cache
            .get(&(stock_id, lv2_prob))
            .map(|v| v.clone())
    }

    /// 反向索引查询: `(stock_id, lv2_prob)` 影响哪些 broker.
    /// 对齐 C++ `setBrokerIDs = m_mapLv2ProbToBrokers[cacheKey]` (QotRealTimeData.cpp:941-944).
    ///
    /// v1.4.111 P2-1 Tier 3 audit comment: reverse lookup helper — empty HashSet =
    /// "no brokers affected by this (stock_id, lv2_prob)", caller iterate 空集合
    /// 自然跳过. 跟 C++ `setBrokerIDs` empty 行为对齐. 非 silent-success risk
    /// (audit verified, essentials/2026-05-27).
    pub fn get_brokers_affected_by_lv2_prob(&self, stock_id: u64, lv2_prob: i32) -> HashSet<u32> {
        self.lv2_prob_to_brokers
            .get(&(stock_id, lv2_prob))
            .map(|s| s.clone())
            .unwrap_or_default()
    }

    /// 给定 `(stock_id, broker_id)`, 收集该 broker 的所有 exchange 对应的
    /// exchange-level cache 列表 (用作 merge 输入).
    ///
    /// 对齐 C++ `MergeCryptoExchangesToBrokerCache` 内部逻辑 (line 987-1022).
    pub fn collect_exchange_caches_for_broker(
        &self,
        stock_id: u64,
        broker_id: u32,
    ) -> Vec<CachedOrderBook> {
        let infos = match self.exchanges_for_broker(stock_id, broker_id) {
            Some(v) => v,
            None => return vec![],
        };
        infos
            .iter()
            .filter_map(|info| self.get_lv2_exchange_cache(stock_id, info.lv2_prob))
            .collect()
    }

    /// 收集某 stock 的所有 exchange-level orderbook cache.
    ///
    /// C++ US stock TotalView/ARCA path does not use broker mappings: it writes
    /// `m_mapLv2ExchangeCache[(stock_id, lv2_type)]`, then
    /// `MergeExchangeCacheToMain` merges every exchange cache for that stock.
    /// Crypto still uses the broker-specific helper above.
    pub fn collect_exchange_caches_for_stock(&self, stock_id: u64) -> Vec<CachedOrderBook> {
        let mut entries: Vec<(i32, CachedOrderBook)> = self
            .lv2_exchange_cache
            .iter()
            .filter_map(|entry| {
                let (sid, lv2_prob) = *entry.key();
                (sid == stock_id).then(|| (lv2_prob, entry.value().clone()))
            })
            .collect();
        entries.sort_by_key(|(lv2_prob, _)| *lv2_prob);
        entries.into_iter().map(|(_, cache)| cache).collect()
    }

    /// 清除某 stock 的所有 cache (退订 / reset 时调).
    /// 对齐 C++ `ClearStockData(nStockID)`.
    pub fn clear_stock(&self, stock_id: u64) {
        // v1.4.110 R6-5: 与 set_lv2_related_exchange / clear_stock_broker 共用
        // rebuild_lock, 避免整 stock 清与反向索引重建交错.
        let _rebuild_guard = self.rebuild_lock.lock();
        self.by_broker.retain(|key, _| key.0 != stock_id);
        self.lv2_exchange_cache.retain(|key, _| key.0 != stock_id);
        self.lv2_prob_to_brokers.retain(|key, _| key.0 != stock_id);
    }

    /// v1.4.110 R6-8: 清除某 `(stock_id, broker_id)` 的 `by_broker` entry +
    /// 重建该 stock 反向索引. 用于 **部分 broker 退订** —— stock 还有别的 broker
    /// 在订 (`clear_stock` 整 stock 清不适用), 但本 broker 已全局无订阅, 其
    /// `by_broker[(stock, broker)]` 会 stale 滞留 → `by_broker` 慢漏累积死 entry.
    ///
    /// 移除本 broker 后该 stock 已无任何 broker → 连 `lv2_exchange_cache` 一并清
    /// (无 broker 引用的 exchange-level cache 也是 stale).
    pub fn clear_stock_broker(&self, stock_id: u64, broker_id: u32) {
        let _rebuild_guard = self.rebuild_lock.lock();
        self.by_broker.remove(&(stock_id, broker_id));
        self.rebuild_reverse_index_for_stock(stock_id);
        // 该 stock 已无任何 broker → exchange-level cache 也清.
        let stock_still_has_broker = self.by_broker.iter().any(|e| e.key().0 == stock_id);
        if !stock_still_has_broker {
            self.lv2_exchange_cache.retain(|key, _| key.0 != stock_id);
        }
    }

    /// v1.4.110 R6-8: 列出 `by_broker` 里某 stock 当前缓存的所有 broker_id.
    /// 退订路径用它枚举该 stock 的 broker, 逐个判是否已全局无订阅 → clear.
    pub fn brokers_for_stock(&self, stock_id: u64) -> Vec<u32> {
        self.by_broker
            .iter()
            .filter(|e| e.key().0 == stock_id)
            .map(|e| e.key().1)
            .collect()
    }

    /// v1.4.110 R6-2 + R6-4: 重建某 `(stock_id, broker_id)` 的 broker-level
    /// 摆盘缓存 —— 收集该 broker 所有 exchange 的 LV2 cache → merge 40 档 → 写
    /// `qot_cache.order_books[broker_cache_key]` → 唤醒 cold-cache waiter.
    ///
    /// 对齐 C++ `QotRealTimeData::RebuildCryptoBrokerCache` (line 960-985).
    /// `parse_crypto_lv2_order_book_to_push` (push 路径, 每个 affected broker 调
    /// 一次) 与 `maybe_fetch_crypto_exchanges` (CMD18012 完成后补 merge, R6-4)
    /// 共用本 fn.
    ///
    /// 返 `Some(merged)` 供 caller 构造 PushEvent; 该 broker 当前无任何 exchange
    /// cache (collect 空) → `None` (无数据可 merge, 不动 `qot_cache`).
    ///
    /// `broker_cache_key` 由 caller 传 (= `QotSecurityKey::cache_key()`,
    /// `"{public}@b{broker_id}"`) —— public sec_key 构造需要 market/code 上下文,
    /// 不在本 cache 持有.
    pub fn rebuild_broker_cache(
        &self,
        qot_cache: &crate::qot_cache::QotCache,
        stock_id: u64,
        broker_id: u32,
        broker_cache_key: &str,
    ) -> Option<CachedOrderBook> {
        let caches = self.collect_exchange_caches_for_broker(stock_id, broker_id);
        if caches.is_empty() {
            return None;
        }
        let merged = crate::qot_cache::merge_multiple_order_book_caches(&caches, 40);
        qot_cache
            .order_books
            .insert(broker_cache_key.to_string(), merged.clone());
        qot_cache.notify_order_book_cold_cache_waiters(broker_cache_key);
        Some(merged)
    }
}

fn positive_lv2_prob_sum(exchanges: &[CryptoExchangeInfo]) -> Option<i32> {
    let sum = exchanges
        .iter()
        .map(|info| i64::from(info.lv2_prob))
        .sum::<i64>();
    i32::try_from(sum).ok().filter(|sum| *sum > 0)
}

#[cfg(test)]
mod tests;