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 parking_lot::RwLock;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use tokio::sync::{Notify, watch};

mod kline;
mod order_book_merge;
mod rt;
mod spread;
mod waiters;
pub use rt::{RtAverageMode, RtPullFlightGuard, RtPullGeneration, RtPushApplyOutcome};

pub use kline::{CachedKLine, KlineDims};
pub use order_book_merge::merge_multiple_order_book_caches;
pub use spread::CachedSpreadBand;
use spread::spread_value_raw_from_bands;
pub use waiters::{
    basic_qot_wait_key, odd_lot_order_book_cache_key, odd_lot_order_book_wait_key,
    order_book_wait_key, ticker_wait_key,
};

// Ref: FutuOpenD/Src/NNBase/NNBase_Define_Enum.h `NN_AppLanguage`
// and FutuOpenD/Src/NNProtoCenter/Quote/NNBiz_Qot_TenBuySellBroker.cpp::GetBrokerName.
const APP_LANGUAGE_ZH: i32 = 0;
const APP_LANGUAGE_HK: i32 = 1;
const APP_LANGUAGE_EN: i32 = 2;
const APP_LANGUAGE_JA: i32 = 5;

/// 股票行情缓存 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 (external reviewer 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>,
}

/// 摆盘缓存 (对齐 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>,
    /// C++ keeps an explicit `m_setUSLv2OrderPushed` marker and `GetOrderBook`
    /// waits for it before serving US/Jp/Crypto Lv2 orderbooks. A non-empty
    /// Lv1 fallback cache is not enough.
    pub accepted_lv2: bool,
}

/// 摆盘单层
#[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 hp_volume: f64,
    /// C++ `QotRealTimeData::AddTickerData` / `NNBiz_Qot_PullQot::OnReply`
    /// retain the event-contract price and direction beside the legacy ticker
    /// price. The legacy public `Qot_Common.Ticker` has no matching fields, so
    /// this metadata remains typed cache state for event-contract consumers.
    pub no_price: Option<f64>,
    pub event_contract_direction: Option<i32>,
    pub backend_strategy: Option<u32>,
    pub backend_order_type: Option<u64>,
    pub backend_hp_turnover: Option<f64>,
    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 is_blank: bool,
    pub price: f64,
    pub last_close_price: f64,
    pub avg_price: f64,
    pub volume: i64,
    pub hp_volume: f64,
    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 = `sec:r{rehab}:k{type}:s{aggregate}` where aggregate is
    /// typed RTH/ETH/ALL, not a raw backend RequestSection.
    pub klines: DashMap<String, Vec<CachedKLine>>,
    /// C++ mutates every RTH/ETH/ALL KLine aggregate under one cache lock.
    /// All production KLine readers and writers take this owner so a pull
    /// replacement cannot interleave with a multi-aggregate push update.
    kline_data_lock: RwLock<()>,
    /// 摆盘缓存
    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>>,
    rt_pull_in_flight: DashMap<String, watch::Sender<bool>>,
    rt_data_publish_lock: RwLock<()>,
    rt_pull_generations: DashMap<String, RtPullGeneration>,
    /// 经纪队列缓存
    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>,
    /// C++ `INNData_Qot_Spread`: spread table code → price bands.
    ///
    /// Filled from CMD6503 at QOT handler registration time and refreshed every
    /// 8h, matching `NNBiz_Qot_Spread::SetTimerUpdateSpreadInfo`. Read paths
    /// are synchronous and lock-free enough for push hot paths.
    pub spread_tables: DashMap<i32, Vec<CachedSpreadBand>>,
    /// 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>>,
    cold_cache_waiter_entries: AtomicUsize,
}

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

    /// Replace the whole spread-table cache after a successful CMD6503 pull.
    ///
    /// C++ `INNData_Qot_Spread::SetSpreadInfo` installs a flattened full table
    /// snapshot. We use a code-keyed map but preserve the same replacement
    /// semantics so stale removed codes do not linger across refreshes.
    pub fn replace_spread_tables<I>(&self, tables: I)
    where
        I: IntoIterator<Item = (i32, Vec<CachedSpreadBand>)>,
    {
        self.spread_tables.clear();
        for (code, bands) in tables {
            if code != 0 && !bands.is_empty() {
                self.spread_tables.insert(code, bands);
            }
        }
    }

    /// Return the raw 1e9 fixed-point spread value for a security price.
    ///
    /// Mirrors the spread-band selection in
    /// `NNBiz_Qot_Spread::GetStockSpreadPriceForTade` with `bUp=true` and
    /// `enTrdMarket=Unknown`, which is what quote snapshot / BasicQot push use.
    /// Missing table returns 0, matching C++ cache miss falling through with
    /// initial `nPriceSpread=0` at the API projection layer.
    pub fn spread_value_raw(&self, spread_code: u32, price_raw: i64) -> i64 {
        let Some(bands) = self.spread_tables.get(&(spread_code as i32)) else {
            return 0;
        };
        spread_value_raw_from_bands(&bands, price_raw)
    }

    /// Event Contract contract-list projection uses the first configured band
    /// as the contract tick size, independent of a current quote price.
    ///
    /// Ref: frozen C++ 10.9.6918
    /// `APIServer_Qot_GetEventContract.cpp`, `GetSpreadInfo(...)[0]`.
    pub fn first_spread_value_raw(&self, spread_code: u32) -> i64 {
        self.spread_tables
            .get(&(spread_code as i32))
            .and_then(|bands| bands.first().map(|band| band.value))
            .unwrap_or(0)
    }

    /// Project `BasicQot.priceSpread` / `SnapshotBasicData.priceSpread`.
    pub fn price_spread_for_raw_price(&self, spread_code: u32, price_raw: i64) -> f64 {
        self.spread_value_raw(spread_code, price_raw) as f64 / 1_000_000_000.0
    }

    /// Project `priceSpread` from a floating-point API price.
    pub fn price_spread_for_price(&self, spread_code: u32, price: f64) -> f64 {
        if spread_code == 0 || price <= 0.0 {
            return 0.0;
        }
        self.price_spread_for_raw_price(spread_code, (price * 1_000_000_000.0) as i64)
    }

    /// 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.get_broker_name_for_app_lang(broker_id, APP_LANGUAGE_ZH)
    }

    /// latest C++ `GetBrokerName` 按 App 语言选择券商简称，并在缺省时
    /// fallback 到 en/tc/sc。Rust 当前 CMD 18008 cache 只保存简称三语字段；
    /// 因此这里对齐 C++ 的 abbreviation-first 分支，完整名 fallback 等
    /// cache 结构扩展后再自然接入。
    pub fn get_broker_name_for_app_lang(&self, broker_id: i64, app_lang: i32) -> Option<String> {
        self.broker_dict.get(&broker_id).and_then(|info| {
            let candidates: [&str; 4] = match app_lang {
                APP_LANGUAGE_EN | APP_LANGUAGE_JA => [
                    &info.name_en,
                    &info.name_en,
                    &info.name_tc,
                    &info.name_zh_cn,
                ],
                APP_LANGUAGE_HK => [
                    &info.name_tc,
                    &info.name_en,
                    &info.name_tc,
                    &info.name_zh_cn,
                ],
                APP_LANGUAGE_ZH => [
                    &info.name_zh_cn,
                    &info.name_en,
                    &info.name_tc,
                    &info.name_zh_cn,
                ],
                _ => [
                    &info.name_zh_cn,
                    &info.name_en,
                    &info.name_tc,
                    &info.name_zh_cn,
                ],
            };
            candidates
                .into_iter()
                .find(|name| !name.is_empty())
                .map(ToOwned::to_owned)
        })
    }

    /// 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_basic_qot_cold_cache_waiters(key);
    }

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

    pub fn get_basic_qot_by_cache_key(&self, cache_key: &str) -> Option<CachedBasicQot> {
        self.get_basic_qot(cache_key)
    }

    pub fn basic_qot_last_close_by_cache_key(&self, cache_key: &str) -> Option<f64> {
        self.basic_qot
            .get(cache_key)
            .map(|quote| quote.last_close_price)
    }

    /// **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_basic_qot_cold_cache_waiters(&cache_key);
    }

    /// **v1.4.110 Phase 2 Slice 5**: 获取基本报价 (broker-aware).
    pub fn get_basic_qot_broker(&self, key: &QotSecurityKey) -> Option<CachedBasicQot> {
        self.get_basic_qot_by_cache_key(key.cache_key_cow().as_ref())
    }

    /// 更新摆盘
    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_order_book_cold_cache_waiters(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_order_book_cold_cache_waiters(&cache_key);
    }

    /// Update odd-lot order book cache (MY/SG only in C++ 10.7).
    pub fn update_odd_lot_order_book_broker(&self, key: &QotSecurityKey, ob: CachedOrderBook) {
        let cache_key = key.cache_key();
        let odd_key = odd_lot_order_book_cache_key(&cache_key);
        self.order_books.insert(odd_key.clone(), ob);
        self.notify_odd_lot_order_book_cold_cache_waiters(&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())
    }

    /// Get odd-lot order book cache (MY/SG only in C++ 10.7).
    pub fn get_odd_lot_order_book_broker(&self, key: &QotSecurityKey) -> Option<CachedOrderBook> {
        let cache_key = key.cache_key();
        self.order_books
            .get(&odd_lot_order_book_cache_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();
        upsert_tickers_by_sequence(&mut entry, new_tickers);
        // C++ `APIServer_Qot_Ticker` wakes waiters when ticker data arrives
        // through `NN_OMEvent_Qot_Update_Ticker`; Rust uses the cache append
        // point as the equivalent notify source for active-pull and push writes.
        self.notify_ticker_cold_cache_waiters(key);
    }

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

    /// **v1.4.110 Phase 2 Slice 5**: 获取逐笔 (broker-aware).
    pub fn get_tickers_broker(&self, key: &QotSecurityKey) -> Option<Vec<CachedTicker>> {
        self.get_tickers_by_cache_key(key.cache_key_cow().as_ref())
    }

    pub fn get_tickers_by_cache_key(&self, cache_key: &str) -> Option<Vec<CachedTicker>> {
        self.tickers.get(cache_key).map(|tickers| tickers.clone())
    }

    pub fn ticker_count_by_cache_key(&self, cache_key: &str) -> usize {
        self.tickers
            .get(cache_key)
            .map_or(0, |tickers| tickers.len())
    }

    pub fn recent_tickers_by_cache_key(
        &self,
        cache_key: &str,
        count: usize,
    ) -> Option<Vec<CachedTicker>> {
        self.tickers.get(cache_key).map(|tickers| {
            let start = tickers.len().saturating_sub(count);
            tickers[start..].to_vec()
        })
    }

    /// 更新经纪队列
    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}:");
        {
            let _owner = self.kline_data_lock.write();
            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 全部桶.
        let _publish = self.rt_data_publish_lock.write();
        self.rt_data.retain(|k, _| !k.starts_with(&prefix));
        self.rt_pull_generations.remove(key);
    }

    /// **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}:");
        {
            let _owner = self.kline_data_lock.write();
            self.klines.retain(|k, _| !k.starts_with(&prefix));
        }
        let _publish = self.rt_data_publish_lock.write();
        self.rt_data.retain(|k, _| !k.starts_with(&prefix));
        self.rt_pull_generations.remove(&cache_key);
    }

    /// Clear only the C++ `ReSub()` realtime cache families for one backend
    /// quote-market bucket.
    ///
    /// Ref: `QotRealTimeData.cpp:613-671` clears Basic, OrderBook (including
    /// odd-lot), Broker and Ticker before a non-CMD6304 replay. KLine and RT
    /// are deliberately retained. Market membership is derived from each
    /// cache key's public FTAPI market and the canonical FTAPI -> backend QOT
    /// mapping; string prefixes are not used as market identity.
    pub fn clear_realtime_quote_market(&self, quote_market_type: u8) {
        if quote_market_type == 0 {
            return;
        }
        self.clear_realtime_quotes_where(|key| {
            cache_key_matches_quote_market(key, quote_market_type)
        });
    }

    /// Clear the C++ `ReSub()` realtime cache families selected by an
    /// owner-supplied market predicate.
    ///
    /// The gateway supplies static-security-aware ownership for ordinary
    /// market replay. Cache-only callers may continue using
    /// `clear_realtime_quote_market`, whose public-market mapping remains
    /// correct for crypto exchange-ready replay.
    pub fn clear_realtime_quotes_where(&self, belongs_to_target: impl Fn(&str) -> bool) {
        self.basic_qot.retain(|key, _| !belongs_to_target(key));
        self.order_books.retain(|key, _| {
            let base = key.strip_suffix(":orderbook_odd").unwrap_or(key);
            !belongs_to_target(base)
        });
        self.tickers.retain(|key, _| !belongs_to_target(key));
        self.brokers.retain(|key, _| !belongs_to_target(key));
    }

    /// Clear all C++ `ReSubAll()` realtime cache families.
    ///
    /// C++ constructs every supported `MktQotSub` bucket up front and calls
    /// `ReSub()` on every bucket after reconnect. Each ordinary `ReSub()`
    /// invokes `QotRealTimeData::OnClearRealTimeData` before rebuilding the
    /// backend desired set. KLine, RT and reference/configuration caches are
    /// deliberately retained.
    pub fn clear_all_realtime_quotes(&self) {
        self.basic_qot.clear();
        self.order_books.clear();
        self.tickers.clear();
        self.brokers.clear();
    }
}

/// C++ `AddTickerList` sorts/deduplicates an incoming batch by ticker key and
/// replaces an existing row when the same key is observed again. Keep that
/// identity rule in the single cache write boundary shared by pull and push.
fn upsert_tickers_by_sequence(entry: &mut Vec<CachedTicker>, mut incoming: Vec<CachedTicker>) {
    incoming.sort_by_key(|ticker| ticker.sequence);
    incoming.dedup_by_key(|ticker| ticker.sequence);

    // Older cache generations may already contain duplicates. Normalize them
    // before applying the new authoritative rows so an upgrade repairs the
    // bucket on its first write.
    let mut existing = std::mem::take(entry);
    existing.sort_by_key(|ticker| ticker.sequence);
    existing.dedup_by_key(|ticker| ticker.sequence);

    let mut existing = existing.into_iter().peekable();
    let mut incoming = incoming.into_iter().peekable();
    let mut merged = Vec::with_capacity(existing.len() + incoming.len());
    while let (Some(old), Some(new)) = (existing.peek(), incoming.peek()) {
        match old.sequence.cmp(&new.sequence) {
            std::cmp::Ordering::Less => {
                if let Some(ticker) = existing.next() {
                    merged.push(ticker);
                }
            }
            std::cmp::Ordering::Greater => {
                if let Some(ticker) = incoming.next() {
                    merged.push(ticker);
                }
            }
            std::cmp::Ordering::Equal => {
                drop(existing.next());
                if let Some(ticker) = incoming.next() {
                    merged.push(ticker);
                }
            }
        }
    }
    merged.extend(existing);
    merged.extend(incoming);

    const TICKER_CACHE_MAX_ITEMS: usize = 1000;
    if merged.len() > TICKER_CACHE_MAX_ITEMS {
        let drain_count = merged.len() - TICKER_CACHE_MAX_ITEMS;
        merged.drain(..drain_count);
    }
    *entry = merged;
}

fn cache_key_matches_quote_market(cache_key: &str, quote_market_type: u8) -> bool {
    QotSecurityKey::parse_cache_key(cache_key)
        .and_then(|(public, _)| {
            QotSecurityKey::parse_public_sec_key(&public).map(|parsed| parsed.market)
        })
        .map(futu_core::qot_subscription::ftapi_market_to_quote_mkt)
        == Some(quote_market_type)
}

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

#[cfg(test)]
mod merge_tests;

#[cfg(test)]
mod tests;