futu_cache/

qot_cache.rs

// 行情数据缓存
//
// 对应 C++ NNDataCenter 中的 INNData_Qot_SecQot / INNData_Qot_KLRT 等
// 使用 DashMap 实现并发安全的内存缓存
//
// ## v1.4.110 Phase 2 Slice 5: broker-aware overloads
//
// 加 broker-aware overload (`*_broker` 后缀) 让 crypto multi-broker push 写
// 入独立 cache key (e.g. `"91_BTCUSDT@b1007"` vs `"91_BTCUSDT@b1008"`).
//
// 老 API 保留 — broker_id=None 时 `QotSecurityKey::cache_key()` 退化到原
// `"market_code"` 形式, 与升级前行为完全等价. Phase 3 才会替换 reader caller
// 改走 `*_broker` 版本 (handler `GetBasicQot` 等).

use dashmap::DashMap;
use futu_core::qot_stock_key::QotSecurityKey;
use std::sync::Arc;
use tokio::sync::Notify;

/// v1.4.110 codex Phase 3 Slice 6c: cold-cache wait key for BasicQot.
///
/// 输入 cache_key (legacy `"market_code"` 或 broker-aware
/// `"market_code@b1007"`), 输出 `"<cache_key>:basic"` 形式 wait key.
/// 与 OrderBook wait key 分桶, 避免互相错唤醒.
#[inline]
pub fn basic_qot_wait_key(cache_key: &str) -> String {
    format!("{cache_key}:basic")
}

/// v1.4.110 codex Phase 3 Slice 6c: cold-cache wait key for OrderBook.
#[inline]
pub fn order_book_wait_key(cache_key: &str) -> String {
    format!("{cache_key}:orderbook")
}

/// 股票行情缓存 key: "market_code" (如 "1_00700") 或 broker-aware "market_code@b1007".
///
/// **v1.4.110 Phase 2 Slice 5**: cache key encoding 仍是 String (Phase 5 不
/// 引入新 hash domain), broker-aware 后缀由 `QotSecurityKey::cache_key()` 编码:
/// - no_broker: `"91_BTCUSDT"` (与升级前等价)
/// - broker-aware: `"91_BTCUSDT@b1007"` (Phase 3 之后启用)
pub type SecurityKey = String;

/// 生成缓存 key
pub fn make_key(market: i32, code: &str) -> SecurityKey {
    format!("{market}_{code}")
}

/// 基本报价缓存
#[derive(Debug, Clone)]
pub struct CachedBasicQot {
    pub cur_price: f64,
    pub open_price: f64,
    pub high_price: f64,
    pub low_price: f64,
    pub last_close_price: f64,
    pub volume: i64,
    pub turnover: f64,
    pub turnover_rate: f64,
    pub amplitude: f64,
    pub is_suspended: bool,
    pub update_time: String,
    pub update_timestamp: f64,
    /// v1.4.72 BUG-006 L3 (eli v1.4.69 P1): US 夜盘 OHLCV 数据。
    /// backend 推送（CMD 6212 Qot_UpdateBasicQot）的 BasicQot.overnight (field 25)
    /// 在夜盘时段会填充，regular hours 为 None。push_parser 提取并缓存，让
    /// 下游 subscribe push + snapshot query 都能看到实时夜盘数据。
    pub overnight: Option<CachedPreAfterMarketData>,
    /// v1.4.106 codex 1140 F4 (P2): US 盘前 OHLCV 数据.
    /// SBIT_US_PREMARKET_AFTERHOURS_DETAIL 推送时由 push_parser 解析填充, US
    /// 盘前时段会有, regular hours / non-US → None. 下游 read 透传给 ftapi
    /// `BasicQot.pre_market` (proto Qot_Common.proto:671). audit Finding 4.
    pub pre_market: Option<CachedPreAfterMarketData>,
    /// v1.4.106 codex 1140 F4 (P2): US 盘后 OHLCV 数据.
    /// 同上, 但取 SBIT_US_PREMARKET_AFTERHOURS_DETAIL 的 after_hours 字段.
    /// 下游 read 透传给 ftapi `BasicQot.after_market`. audit Finding 4.
    pub after_market: Option<CachedPreAfterMarketData>,
}

/// v1.4.72 BUG-006 L3: 美股夜盘 OHLCV 数据（对齐 proto `Qot_Common::PreAfterMarketData`）
///
/// 同一 struct 在 pre_market / after_market / overnight 三个字段都复用。
#[derive(Debug, Clone, Default)]
pub struct CachedPreAfterMarketData {
    pub price: Option<f64>,
    pub high_price: Option<f64>,
    pub low_price: Option<f64>,
    pub volume: Option<i64>,
    pub turnover: Option<f64>,
    pub change_val: Option<f64>,
    pub change_rate: Option<f64>,
    pub amplitude: Option<f64>,
}

/// K 线缓存
#[derive(Debug, Clone)]
pub struct CachedKLine {
    pub time: String,
    pub open_price: f64,
    pub high_price: f64,
    pub low_price: f64,
    pub close_price: f64,
    pub volume: i64,
    pub turnover: f64,
}

/// 摆盘缓存 (对齐 C++ Qot_UpdateOrderBook::S2C)
#[derive(Debug, Clone, Default)]
pub struct CachedOrderBook {
    pub ask_list: Vec<CachedOrderBookLevel>,
    pub bid_list: Vec<CachedOrderBookLevel>,
    pub svr_recv_time_bid: Option<String>,
    pub svr_recv_time_bid_timestamp: Option<f64>,
    pub svr_recv_time_ask: Option<String>,
    pub svr_recv_time_ask_timestamp: Option<f64>,
}

/// 摆盘单层
#[derive(Debug, Clone)]
pub struct CachedOrderBookLevel {
    pub price: f64,
    pub volume: i64,
    pub order_count: i32,
    /// v1.4.106 codex 1140 F7 (P2 audit Finding 7): SF 行情订单明细列表.
    /// backend OrderBookItem.orders (重复 OrderInfo: order_id + order_size).
    /// 仅 HK SF 行情 + prob=BIT_PROB_ORDER_BOOK_ALL_WITH_ID 时 backend 才返;
    /// 普通行情 → 空 vec. 下游 ftapi `Qot_Common.OrderBook.detailList` 透传.
    pub detail_list: Vec<CachedOrderBookDetail>,
    /// v1.4.110 codex audit Round4 R4-4: 高精度委托数量 (crypto 适用).
    ///
    /// crypto 盘口的 `volume` 是放大整数 (`size × 10^order_size_precision`),
    /// i64 无法表示小数量; `hp_volume = volume / 10^precision` 是真实小数量.
    /// 普通行情 `volume` 已是精确整数 → `None` (对齐 C++ `has_hpvolume()==false`
    /// 时 fallback `volume`). 下游 emit 到 ftapi `Qot_Common.OrderBook.hpVolume`.
    ///
    /// 对齐 C++ `QotRealTimeData.cpp` `pOrderBookItem->set_hpvolume(...)` +
    /// merge `gear.dVolume += has_hpvolume() ? hpvolume() : volume()` —— merge
    /// 累加的是 de-scale 后的真实量, 故按 level 存 (不同交易所 precision 可能不同).
    pub hp_volume: Option<f64>,
}

/// v1.4.106 codex 1140 F7 (P2): 摆盘订单明细 (HK SF).
/// 对齐 ftapi `Qot_Common.OrderBookDetail` (proto field orderID + volume).
#[derive(Debug, Clone)]
pub struct CachedOrderBookDetail {
    pub order_id: i64,
    pub volume: i64,
}

/// 逐笔成交缓存 (对齐 C++ Qot_UpdateTicker::S2C)
///
/// v1.4.106 codex 1140 F5: 加 `type_sign` 字段 (audit Finding 5),
/// 对齐 ftapi `Qot_Common.Ticker.typeSign` (proto field 9). 之前 cache 缺
/// 此字段, push event 与 read response 都没法透传 type_sign.
#[derive(Debug, Clone)]
pub struct CachedTicker {
    pub time: String, // HH:MM:SS 时间字符串 (从 exchange_data_time_ms 派生, 按 market 时区)
    pub sequence: i64, // tick_key, 用于去重
    pub dir: i32,     // 1=Bid/卖盘, 2=Ask/买盘, 3=Neutral
    pub price: f64,
    pub volume: i64,
    pub turnover: f64,            // price × volume
    pub recv_time: Option<f64>,   // server_recv_from_exchange_time_ms (秒)
    pub ticker_type: Option<i32>, // 逐笔类型 (TickItemType: BUY=1/SELL=2/NEUTRAL=3)
    /// v1.4.106 codex 1140 F5: 逐笔类型符号 (audit Finding 5).
    /// 来自 TickItem.trade_type (一个英文字母的 ASCII 码), backend 推送 +
    /// FTAPI Ticker.typeSign 对外暴露给 UI.
    pub type_sign: Option<i32>,
    pub push_data_type: Option<i32>,
    pub timestamp: Option<f64>,
}

/// 分时数据点
#[derive(Debug, Clone)]
pub struct CachedTimeShare {
    pub time: String,
    pub minute: i32,
    pub price: f64,
    pub last_close_price: f64,
    pub avg_price: f64,
    pub volume: i64,
    pub turnover: f64,
    pub timestamp: f64,
}

/// 经纪队列缓存 (对齐 C++ Qot_UpdateBroker::S2C)
#[derive(Debug, Clone, Default)]
pub struct CachedBroker {
    pub bid_list: Vec<CachedBrokerItem>,
    pub ask_list: Vec<CachedBrokerItem>,
}

/// 经纪队列单项
#[derive(Debug, Clone)]
pub struct CachedBrokerItem {
    pub id: i64,
    pub name: String,
    pub pos: i32,
    /// v1.4.106 codex 1140 F7 (P2 audit Finding 7): HK SF 行情订单 ID.
    /// 对齐 ftapi `Qot_Common.Broker.orderID` (proto field 4 optional). 仅
    /// HK SF 时 backend HKBrokerQueue.order_id_list 含值, 普通行情 → None.
    pub order_id: Option<i64>,
    /// v1.4.106 codex 1140 F7 (P2): HK SF 订单股数. 对齐 ftapi
    /// `Qot_Common.Broker.volume` (proto field 5 optional).
    pub volume: Option<i64>,
}

/// v1.4.106 codex 1140 F7 (P2 audit Finding 7): 券商配置表 (broker_id → 名称).
/// 由 CMD 18008 (NN_ProtoCmd_Qot_Pull_BrokerInfo) 拉取并解析后填充.
/// 替代旧 `format!("Broker#{bid}")` 占位符进入公开 API 的反模式.
#[derive(Debug, Clone)]
pub struct CachedBrokerInfo {
    /// 中文简称 (sc) — 用作主显示名 (与 C++ GetBrokerName 同语义)
    pub name_zh_cn: String,
    /// 英文简称 (en)
    pub name_en: String,
    /// 中文繁体简称 (tc)
    pub name_tc: String,
}

/// 行情缓存管理器
pub struct QotCache {
    /// 基本报价缓存
    pub basic_qot: DashMap<SecurityKey, CachedBasicQot>,
    /// US stock overnight-enabled state, keyed by backend stock_id.
    ///
    /// C++ stores this as `stockID -> bool` in `INNData_Qot_USStockOvernight`:
    /// - `NNData_Qot_USStockOvernight.cpp:21-35` (missing key => false)
    /// - `NNBiz_Qot_USStockState.cpp:180-190` writes `overnight_type == 1`
    /// - `APIServer_Qot_MarketState.cpp:238-244` reads it for 11 -> 37 projection
    pub us_stock_overnight: DashMap<u64, bool>,
    /// K 线缓存: key = "market_code:kl_type"
    pub klines: DashMap<String, Vec<CachedKLine>>,
    /// 摆盘缓存
    pub order_books: DashMap<SecurityKey, CachedOrderBook>,
    /// 逐笔缓存: 保留最近 N 条
    pub tickers: DashMap<SecurityKey, Vec<CachedTicker>>,
    /// 分时缓存
    /// v1.4.106 codex 1140 F6 (P2 audit Finding 6): RT cache key 加 session
    /// 维度. 之前 `DashMap<SecurityKey, ...>` 把 RTH/ETH/PRE/AFTER 全部混到
    /// 同一桶, 客户端订阅 RTH 也能读到 PRE 数据. 现在 key 是
    /// "sec_key:s{session}" (RequestSection 0=NORMAL/1=FULL/2=PREMARKET/
    /// 3=AFTERHOURS), 隔离不同 session.
    pub rt_data: DashMap<String, Vec<CachedTimeShare>>,
    /// 经纪队列缓存
    pub brokers: DashMap<SecurityKey, CachedBroker>,
    /// v1.4.106 codex 1140 F7 (P2 audit Finding 7): 券商 ID → 信息映射.
    /// 由 CMD 18008 拉取后填充, 用于 push parser 从 broker_id 查真名 (替代
    /// `Broker#{bid}` 占位符).
    pub broker_dict: DashMap<i64, CachedBrokerInfo>,
    /// v1.4.110 codex Phase 3 Slice 6c: cold-cache wait waiters.
    ///
    /// key = `"<cache_key>:<wait_kind>"` (e.g. `"91_BTCUSDT@b1007:basic"` /
    /// `"1_00700:orderbook"`). value = shared `Arc<Notify>` 让 handler 阻塞等
    /// push parser 写 cache 后唤醒.
    ///
    /// 对齐 C++ `APIServer_Qot_StockBasic.cpp:226-320` `WaitForReady` —
    /// 已订阅但 cache 未就绪时 handler 主动 Pull_SubData + 等 push 写 cache.
    ///
    /// 设计 trade-off:
    /// - 用 `DashMap<String, Arc<Notify>>` 而非 `RwLock<HashMap>`: 高并发读
    ///   写不锁全表
    /// - key 编码 wait_kind 防 basic / orderbook 共用同一 Notify 互相错唤醒
    /// - update path 调 `notify_waiters` (broadcast 给所有 awaiter) 然后从
    ///   map 中 remove (Arc 被 awaiter 持有, 自然释放)
    pub cold_cache_waiters: DashMap<String, Arc<Notify>>,
}

impl QotCache {
    pub fn new() -> Self {
        Self {
            basic_qot: DashMap::new(),
            us_stock_overnight: DashMap::new(),
            klines: DashMap::new(),
            order_books: DashMap::new(),
            tickers: DashMap::new(),
            rt_data: DashMap::new(),
            brokers: DashMap::new(),
            // v1.4.106 codex 1140 F7: broker dict 由 CMD 18008 拉取后填充.
            broker_dict: DashMap::new(),
            // v1.4.110 codex Phase 3 Slice 6c: cold-cache wait waiters.
            cold_cache_waiters: DashMap::new(),
        }
    }

    /// v1.4.110 codex Phase 3 Slice 6c: 注册 cold-cache wait waiter.
    ///
    /// 返已存在或新建的 `Arc<Notify>`. handler 调:
    /// 1. `register_cold_cache_waiter("91_BTCUSDT@b1007:basic")` 获 Notify
    /// 2. 主动发 Pull_SubData CMD6824
    /// 3. `tokio::time::timeout(Duration::from_secs(3), notify.notified())` 等
    /// 4. 再 `get_basic_qot_broker(&key)` 读 cache (可能仍 None — 真 timeout)
    ///
    /// `wait_kind` 推荐: `"basic"` / `"orderbook"`. 不混 sub_type 数字防误唤.
    pub fn register_cold_cache_waiter(&self, wait_key: &str) -> Arc<Notify> {
        self.cold_cache_waiters
            .entry(wait_key.to_string())
            .or_insert_with(|| Arc::new(Notify::new()))
            .clone()
    }

    /// v1.4.110 codex Phase 3 Slice 6c: 唤醒指定 cold-cache wait waiter.
    ///
    /// push parser update path 调 (`update_basic_qot` / `update_order_book` /
    /// `_broker` 变种). 没 waiter → no-op. 有 waiter → `notify_waiters()`
    /// broadcast 给所有 awaiter, 然后 remove (Arc 仍被 awaiter 持有, 自然释放).
    pub fn notify_cold_cache_waiters(&self, wait_key: &str) {
        if let Some((_, n)) = self.cold_cache_waiters.remove(wait_key) {
            n.notify_waiters();
        }
    }

    /// v1.4.110 codex audit Round3 #22: cold-cache wait timeout 后清 idle waiter.
    ///
    /// `wait_for_basic_cache` / `wait_for_order_book_cache` 3s timeout 仍 cache
    /// miss 时调. 若 push 始终没来, `notify_cold_cache_waiters` 不会触发, entry
    /// 会一直留在 `cold_cache_waiters` map (虽 bounded by distinct wait_key 数,
    /// 仍是慢速 leak).
    ///
    /// **只删 caller 自己注册的那个 entry, 且无其他并发 awaiter 时才删**:
    /// `remove_if` closure 在 entry lock 下原子检查两条:
    /// 1. `Arc::ptr_eq(stored, caller_notify)` — stored 必须就是 caller 当初
    ///    `register_cold_cache_waiter` 拿到的同一 Arc. 防 race: caller timeout
    ///    后到本调用之间, 若 push 触发 `notify_cold_cache_waiters` 删了旧 entry,
    ///    另一个 `wait_for_*` 又 register 建了**新** entry (不同 Arc), `ptr_eq`
    ///    false → 不误删别人的新 entry.
    /// 2. `Arc::strong_count(stored) <= 2` — DashMap stored Arc 1 + caller
    ///    持有的 `caller_notify` 1. `> 2` 表示有其他 `wait_for_*` 仍 await 同
    ///    entry → 保留让它们能被 notify 唤醒.
    ///
    /// caller 约定: 必须把 `register_cold_cache_waiter` 返回的 `Arc<Notify>`
    /// 原样传进来 (caller 全程持有未 drop).
    pub fn cleanup_cold_cache_waiter_if_idle(&self, wait_key: &str, caller_notify: &Arc<Notify>) {
        self.cold_cache_waiters.remove_if(wait_key, |_, stored| {
            Arc::ptr_eq(stored, caller_notify) && Arc::strong_count(stored) <= 2
        });
    }

    /// Update C++-style US overnight stock state (`stockID -> bool`).
    ///
    /// Ref: `NNData_Qot_USStockOvernight.cpp:21-35` and
    /// `NNBiz_Qot_USStockState.cpp:180-190`.
    pub fn set_us_stock_overnight_state(&self, stock_id: u64, is_overnight: bool) {
        if stock_id == 0 {
            return;
        }
        self.us_stock_overnight.insert(stock_id, is_overnight);
    }

    /// Query whether a US stock is currently in overnight trading.
    ///
    /// C++ cache miss returns false (`NNData_Qot_USStockOvernight.cpp:29-34`).
    pub fn is_us_stock_overnight(&self, stock_id: u64) -> bool {
        self.us_stock_overnight
            .get(&stock_id)
            .map(|v| *v)
            .unwrap_or(false)
    }

    /// v1.4.106 codex 1140 F7 (P2): 查 broker_id → broker name (中文简称).
    /// cache miss → None, 调用方决定 fallback 策略 (push parser 用
    /// `Broker#{id}` 作 emergency fallback, 但同时 warn-log 提示 dict 未加载).
    pub fn get_broker_name(&self, broker_id: i64) -> Option<String> {
        self.broker_dict
            .get(&broker_id)
            .map(|info| info.name_zh_cn.clone())
    }

    /// v1.4.106 codex 1140 F7 (P2): 批量写入 broker dict (CMD 18008 解析后调).
    pub fn install_broker_dict(&self, entries: Vec<(i64, CachedBrokerInfo)>) {
        for (id, info) in entries {
            self.broker_dict.insert(id, info);
        }
    }

    /// 更新基本报价
    pub fn update_basic_qot(&self, key: &str, qot: CachedBasicQot) {
        self.basic_qot.insert(key.to_string(), qot);
        // v1.4.110 codex Phase 3 Slice 6c: cold-cache wait notify.
        self.notify_cold_cache_waiters(&basic_qot_wait_key(key));
    }

    /// 获取基本报价
    pub fn get_basic_qot(&self, key: &str) -> Option<CachedBasicQot> {
        self.basic_qot.get(key).map(|v| v.clone())
    }

    /// **v1.4.110 Phase 2 Slice 5**: 更新基本报价 (broker-aware).
    ///
    /// 用 `QotSecurityKey::cache_key()` 派生 String key. broker_id=None → 与
    /// `update_basic_qot(public_sec_key, ...)` 等价; broker_id=Some(N) → 写
    /// 独立 cache key `"market_code@b{N}"` (crypto multi-broker isolation).
    pub fn update_basic_qot_broker(&self, key: &QotSecurityKey, qot: CachedBasicQot) {
        let cache_key = key.cache_key();
        self.basic_qot.insert(cache_key.clone(), qot);
        // v1.4.110 codex Phase 3 Slice 6c: cold-cache wait notify.
        self.notify_cold_cache_waiters(&basic_qot_wait_key(&cache_key));
    }

    /// **v1.4.110 Phase 2 Slice 5**: 获取基本报价 (broker-aware).
    pub fn get_basic_qot_broker(&self, key: &QotSecurityKey) -> Option<CachedBasicQot> {
        self.basic_qot.get(&key.cache_key()).map(|v| v.clone())
    }

    /// 构造 RT 分时 cache key (v1.4.106 codex 1140 F6).
    ///
    /// 之前 key 仅 sec_key, RTH/ETH/PRE/AFTER/OVERNIGHT 混桶. 现在
    /// "sec_key:s{session}" 隔离. session 来自 push 解析的
    /// `TimeSharingPlans.section_type[0]` (RequestSection enum):
    /// 0=NORMAL/1=FULL/2=PREMARKET/3=AFTERHOURS/5=OVERNIGHT.
    /// GetRT handler 对 C++ 的 Session_ETH/Session_ALL 读取语义做动态拼接，
    /// 不依赖额外 aggregate cache 桶。
    pub fn make_rt_key(sec_key: &str, session: i32) -> String {
        format!("{sec_key}:s{session}")
    }

    /// **v1.4.110 Phase 2 Slice 5**: 构造 RT 分时 cache key (broker-aware).
    ///
    /// 用 `QotSecurityKey::cache_key()` 作 prefix. broker_id=None → 与
    /// `make_rt_key(public_sec_key, session)` 等价; broker_id=Some(N) → 写
    /// 独立 cache key `"market_code@b{N}:s{session}"`.
    pub fn make_rt_key_broker(key: &QotSecurityKey, session: i32) -> String {
        format!("{}:s{}", key.cache_key(), session)
    }

    /// **v1.4.110 Phase 2 Slice 5**: 更新 RT 分时 (broker-aware).
    pub fn update_rt_data_broker(
        &self,
        key: &QotSecurityKey,
        session: i32,
        rt_data: Vec<CachedTimeShare>,
    ) {
        let cache_key = Self::make_rt_key_broker(key, session);
        self.rt_data.insert(cache_key, rt_data);
    }

    /// **v1.4.110 Phase 2 Slice 5**: 获取 RT 分时 (broker-aware).
    pub fn get_rt_data_broker(
        &self,
        key: &QotSecurityKey,
        session: i32,
    ) -> Option<Vec<CachedTimeShare>> {
        let cache_key = Self::make_rt_key_broker(key, session);
        self.rt_data.get(&cache_key).map(|v| v.clone())
    }

    /// 构造 K 线 cache key (v1.4.106 codex 1140 F3 4-tuple).
    ///
    /// 之前 key 仅 `(sec_key, kl_type)` 2-tuple, 同股票同 KLType 但前复权 vs
    /// 后复权 / RTH vs ETH 数据互相覆盖. 对齐 C++ APIServer_Qot_KL.cpp:
    /// `GetNewestKLByCount(stock_id, enRehabType, enKLType, num, session, ...)`
    /// 用 4 维 key.
    ///
    /// - `rehab`: proto Qot_Common.RehabType (0=None, 1=Forward, 2=Backward),
    ///   对齐 backend `FTCmdKline.ExrightType`. 同一股票同一 kl_type 不同 rehab
    ///   走独立 cache, 不互相覆盖.
    /// - `kl_type`: proto Qot_Common.KLType (1=1Min, 2=Day, ..., 11=Quarter).
    /// - `session`: proto FTCmdKline.RequestSection (0=NORMAL, 1=FULL,
    ///   2=PREMARKET, 3=AFTERHOURS). RTH/ETH 隔离, push 来自 backend 的
    ///   `point.section_type[0]` 决定写入桶, read 由 client subscription
    ///   session 决定 (尚无, 默认 NORMAL).
    pub fn make_kline_key(sec_key: &str, rehab: i32, kl_type: i32, session: i32) -> String {
        format!("{sec_key}:r{rehab}:k{kl_type}:s{session}")
    }

    /// 更新 K 线 (v1.4.106 codex 1140 F3: 4-tuple key, rehab + session 隔离).
    pub fn update_klines(
        &self,
        sec_key: &str,
        rehab: i32,
        kl_type: i32,
        session: i32,
        klines: Vec<CachedKLine>,
    ) {
        let cache_key = Self::make_kline_key(sec_key, rehab, kl_type, session);
        self.klines.insert(cache_key, klines);
    }

    /// 获取 K 线 (v1.4.106 codex 1140 F3: 4-tuple key, rehab + session 隔离).
    pub fn get_klines(
        &self,
        sec_key: &str,
        rehab: i32,
        kl_type: i32,
        session: i32,
    ) -> Option<Vec<CachedKLine>> {
        let cache_key = Self::make_kline_key(sec_key, rehab, kl_type, session);
        self.klines.get(&cache_key).map(|v| v.clone())
    }

    /// **v1.4.110 Phase 2 Slice 5**: 更新 K 线 (broker-aware).
    ///
    /// 用 `QotSecurityKey::cache_key()` 作 prefix, broker_id=None 时退化到原行为.
    /// composite 维度仍是 4-tuple `(rehab, kl_type, session)`, broker_id 是第 5
    /// 维通过 `QotSecurityKey` 注入到 prefix.
    pub fn update_klines_broker(
        &self,
        key: &QotSecurityKey,
        rehab: i32,
        kl_type: i32,
        session: i32,
        klines: Vec<CachedKLine>,
    ) {
        let cache_key = Self::make_kline_key(&key.cache_key(), rehab, kl_type, session);
        self.klines.insert(cache_key, klines);
    }

    /// **v1.4.110 Phase 2 Slice 5**: 获取 K 线 (broker-aware).
    pub fn get_klines_broker(
        &self,
        key: &QotSecurityKey,
        rehab: i32,
        kl_type: i32,
        session: i32,
    ) -> Option<Vec<CachedKLine>> {
        let cache_key = Self::make_kline_key(&key.cache_key(), rehab, kl_type, session);
        self.klines.get(&cache_key).map(|v| v.clone())
    }

    /// 更新摆盘
    pub fn update_order_book(&self, key: &str, ob: CachedOrderBook) {
        self.order_books.insert(key.to_string(), ob);
        // v1.4.110 codex Phase 3 Slice 6c: cold-cache wait notify.
        self.notify_cold_cache_waiters(&order_book_wait_key(key));
    }

    /// **v1.4.110 Phase 2 Slice 5**: 更新摆盘 (broker-aware).
    pub fn update_order_book_broker(&self, key: &QotSecurityKey, ob: CachedOrderBook) {
        let cache_key = key.cache_key();
        self.order_books.insert(cache_key.clone(), ob);
        // v1.4.110 codex Phase 3 Slice 6c: cold-cache wait notify.
        self.notify_cold_cache_waiters(&order_book_wait_key(&cache_key));
    }

    /// **v1.4.110 Phase 2 Slice 5**: 获取摆盘 (broker-aware).
    pub fn get_order_book_broker(&self, key: &QotSecurityKey) -> Option<CachedOrderBook> {
        self.order_books.get(&key.cache_key()).map(|v| v.clone())
    }

    /// 追加逐笔（保留最近 1000 条）
    pub fn append_tickers(&self, key: &str, new_tickers: Vec<CachedTicker>) {
        let mut entry = self.tickers.entry(key.to_string()).or_default();
        entry.extend(new_tickers);
        if entry.len() > 1000 {
            let drain_count = entry.len() - 1000;
            entry.drain(..drain_count);
        }
    }

    /// **v1.4.110 Phase 2 Slice 5**: 追加逐笔 (broker-aware).
    pub fn append_tickers_broker(&self, key: &QotSecurityKey, new_tickers: Vec<CachedTicker>) {
        let mut entry = self.tickers.entry(key.cache_key()).or_default();
        entry.extend(new_tickers);
        if entry.len() > 1000 {
            let drain_count = entry.len() - 1000;
            entry.drain(..drain_count);
        }
    }

    /// **v1.4.110 Phase 2 Slice 5**: 获取逐笔 (broker-aware).
    pub fn get_tickers_broker(&self, key: &QotSecurityKey) -> Option<Vec<CachedTicker>> {
        self.tickers.get(&key.cache_key()).map(|v| v.clone())
    }

    /// 更新经纪队列
    pub fn update_broker(&self, key: &str, broker: CachedBroker) {
        self.brokers.insert(key.to_string(), broker);
    }

    /// 获取经纪队列
    pub fn get_broker(&self, key: &str) -> Option<CachedBroker> {
        self.brokers.get(key).map(|v| v.clone())
    }

    /// 清除指定股票的所有缓存
    pub fn clear_security(&self, key: &str) {
        self.basic_qot.remove(key);
        self.order_books.remove(key);
        self.tickers.remove(key);
        self.brokers.remove(key);
        // v1.4.106 codex 1140 F3: K 线 key 是 "sec_key:r{rehab}:k{kl_type}:s{session}"
        // 4-tuple, 仍是 sec_key prefix 起头, retain prefix match 仍正确清所有维度.
        let prefix = format!("{key}:");
        self.klines.retain(|k, _| !k.starts_with(&prefix));
        // v1.4.106 codex 1140 F6: rt_data key 也加 session 维度后变为
        // "sec_key:s{session}", 同样 prefix-match 清 RTH/ETH/ALL 全部桶.
        self.rt_data.retain(|k, _| !k.starts_with(&prefix));
    }

    /// **v1.4.110 Phase 2 Slice 5**: 清除指定股票的所有缓存 (broker-aware).
    ///
    /// 用 `QotSecurityKey::cache_key()` 派生 cache key 字符串. broker_id=None
    /// → 与 `clear_security(public_sec_key)` 等价; broker_id=Some(N) → 只清
    /// 该 broker 下的 cache (其他 broker 下同 stock_id 的 cache 保留).
    pub fn clear_security_broker(&self, key: &QotSecurityKey) {
        let cache_key = key.cache_key();
        self.basic_qot.remove(&cache_key);
        self.order_books.remove(&cache_key);
        self.tickers.remove(&cache_key);
        self.brokers.remove(&cache_key);
        let prefix = format!("{cache_key}:");
        self.klines.retain(|k, _| !k.starts_with(&prefix));
        self.rt_data.retain(|k, _| !k.starts_with(&prefix));
    }
}

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

// =============================================================================
// v1.4.110 codex QOT Phase 4 Slice 7: MergeMultipleOrderBookCaches pure fn.
//
// 对齐 C++ `QotRealTimeData::MergeMultipleOrderBookCaches` + `MergeOrderBookLists`
// (QotRealTimeData.cpp:1440-1591):
//
// 1. **Per-side price-key merge** — 每档按 `price * 1e9` 取整作 key, 同 key
//    累加 volume / order_count / detail_list (C++ MergedGear struct).
// 2. **Side-aware sort** — bid 降序 (高价先), ask 升序 (低价先).
// 3. **Top N truncate** — 仅当 `max_depth > 0` 时截断 (crypto LV2 用 40).
// 4. **Timestamp**: bid/ask 各自取多 source 的最大 timestamp.
// 5. **Skip price ≤ 0**: C++ "过滤空数据占位条目" (line 1466).
// =============================================================================

/// Per-side merge buffer (对齐 C++ `MergedGear`).
#[derive(Debug, Clone)]
struct MergedGear {
    price: f64,
    volume: i64,
    order_count: i32,
    detail_list: Vec<CachedOrderBookDetail>,
    /// v1.4.110 codex audit Round4 R4-4: 高精度量累加器. 对齐 C++
    /// `gear.dVolume += has_hpvolume() ? hpvolume() : volume()` —— 累加 de-scale
    /// 后的真实量 (有 hp_volume 用之, 否则 fallback `volume`).
    hp_volume: f64,
}

/// Per-side merge: take multiple lists, aggregate by price key, sort, optionally
/// truncate to `max_depth`. 不会修改输入.
///
/// `is_bid=true` → sort 降序; `false` → 升序.
fn merge_order_book_side(
    sources: &[&[CachedOrderBookLevel]],
    is_bid: bool,
    max_depth: usize,
) -> Vec<CachedOrderBookLevel> {
    if sources.is_empty() {
        return vec![];
    }
    // 价格精度放大到整数 (9 位小数, 对齐 C++ `PriceToKey`).
    fn price_to_key(p: f64) -> i64 {
        (p * 1e9 + 0.5) as i64
    }

    let mut merged: std::collections::HashMap<i64, MergedGear> = std::collections::HashMap::new();
    for list in sources {
        for level in list.iter() {
            // C++: 过滤空数据占位条目 (price <= 0).
            if level.price <= 0.0 {
                continue;
            }
            let key = price_to_key(level.price);
            let entry = merged.entry(key).or_insert(MergedGear {
                price: level.price,
                volume: 0,
                order_count: 0,
                detail_list: vec![],
                hp_volume: 0.0,
            });
            entry.volume = entry.volume.saturating_add(level.volume);
            entry.order_count = entry.order_count.saturating_add(level.order_count);
            entry.detail_list.extend(level.detail_list.iter().cloned());
            // v1.4.110 codex audit Round4 R4-4: 对齐 C++ merge gear.dVolume 累加 ——
            // 有 hp_volume 用真实小数量, 否则 fallback `volume`.
            entry.hp_volume += level.hp_volume.unwrap_or(level.volume as f64);
        }
    }

    let mut keys: Vec<i64> = merged.keys().copied().collect();
    if is_bid {
        // 买盘: 价格从高到低 (降序).
        keys.sort_by(|a, b| b.cmp(a));
    } else {
        // 卖盘: 价格从低到高 (升序).
        keys.sort();
    }

    let mut out: Vec<CachedOrderBookLevel> = keys
        .iter()
        .map(|k| {
            let g = &merged[k];
            CachedOrderBookLevel {
                price: g.price,
                volume: g.volume,
                order_count: g.order_count,
                detail_list: g.detail_list.clone(),
                // v1.4.110 codex audit Round4 R4-4: merge 输出 crypto 多交易所
                // 合单盘口, 始终带累加后的 hp_volume (对齐 C++ set_hpvolume).
                hp_volume: Some(g.hp_volume),
            }
        })
        .collect();

    // 深度截断: max_depth > 0 时生效.
    if max_depth > 0 && out.len() > max_depth {
        out.truncate(max_depth);
    }
    out
}

/// Merge N exchange-level orderbook caches into one broker-level cache.
///
/// 对齐 C++ `QotRealTimeData::MergeMultipleOrderBookCaches(vSources, pbMerged,
/// nMaxDepth=40)` (QotRealTimeData.cpp:1534-1591):
/// - 第一个 source 作为基础 (拷 svr_recv_time 等基本字段, 但 bid/ask list 重算)
/// - 按 price key 聚合 bid/ask, 同 price 累加 volume + order_count + detail
/// - bid 高价优先, ask 低价优先, 截断到 `max_depth`
/// - bid/ask timestamp 取多 source 最大值
///
/// `max_depth = 0` → 不截断 (C++ 行为, line 1579-1583 `if nMaxDepth > 0`).
/// `max_depth = 40` → crypto LV2 broker-level cache 用 (line 1014).
pub fn merge_multiple_order_book_caches(
    sources: &[CachedOrderBook],
    max_depth: usize,
) -> CachedOrderBook {
    if sources.is_empty() {
        return CachedOrderBook::default();
    }

    // v1.4.110 codex audit Round2 P2 #17: 收 slice-of-slice 而非 deep clone.
    // merge_order_book_side 只读 levels (price/volume/order_count/detail_list),
    // 不消费 Vec — 之前 `.bid_list.clone()` 每 source deep clone N×M levels
    // (5 exchange × 40 level = 200 clone per LV2 push), crypto LV2 push 1Hz+
    // 频率下浪费. 改 `.as_slice()` 只收引用, 零 Level clone.
    let bid_sources: Vec<&[CachedOrderBookLevel]> =
        sources.iter().map(|s| s.bid_list.as_slice()).collect();
    let ask_sources: Vec<&[CachedOrderBookLevel]> =
        sources.iter().map(|s| s.ask_list.as_slice()).collect();

    let bid_merged = merge_order_book_side(&bid_sources, true, max_depth);
    let ask_merged = merge_order_book_side(&ask_sources, false, max_depth);

    // Timestamp 取所有 source 最大 (对齐 C++ line 1572-1575).
    let max_bid_ts = sources
        .iter()
        .filter_map(|s| s.svr_recv_time_bid_timestamp)
        .fold(None::<f64>, |acc, t| {
            Some(acc.map(|a| a.max(t)).unwrap_or(t))
        });
    let max_ask_ts = sources
        .iter()
        .filter_map(|s| s.svr_recv_time_ask_timestamp)
        .fold(None::<f64>, |acc, t| {
            Some(acc.map(|a| a.max(t)).unwrap_or(t))
        });

    CachedOrderBook {
        bid_list: bid_merged,
        ask_list: ask_merged,
        svr_recv_time_bid: None,
        svr_recv_time_bid_timestamp: max_bid_ts,
        svr_recv_time_ask: None,
        svr_recv_time_ask_timestamp: max_ask_ts,
    }
}

#[cfg(test)]
mod merge_tests;

#[cfg(test)]
mod tests;