futu_cache/

trd_cache.rs

// 交易数据缓存

mod types;

#[cfg(test)]
mod regression_tests;

pub use types::*;

use dashmap::DashMap;
use futu_core::account_locator;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

/// 交易数据缓存
pub struct TrdCache {
    /// C++ `NNData_Trd_AccList::m_mapUserAccList` equivalent.
    ///
    /// This is the authoritative internal account index used by request
    /// validation, broker routing, and funds/positions/order queries. It may
    /// contain universal parent accounts that are intentionally not exposed by
    /// public `Trd_GetAccList`.
    pub accounts: DashMap<AccKey, CachedTrdAcc>,
    /// C++ `NNData_Trd_AccList::m_mapIDRelation` equivalent:
    /// `universal_or_self_acc_id -> public sub account ids`.
    ///
    /// `Trd_GetAccList` uses this relation via `get_accounts()` to expose the
    /// same public projection as C++ `GetAllSubAccList`, while `lookup_account`
    /// and direct `accounts.get()` still see the full internal map.
    pub account_relations: DashMap<AccKey, Vec<AccKey>>,
    /// Public account ids derived from `account_relations`.
    pub public_account_ids: DashMap<AccKey, ()>,
    public_projection_ready: AtomicBool,
    /// 资金: `FundsCacheKey { acc_id, asset_category, currency }` → funds.
    /// **v1.4.106 Finding A**: 之前 `DashMap<AccKey, CachedFunds>` 一 acc 一 snapshot,
    /// Universal/Futures 多币种场景被覆盖 — 改 currency-aware key 对齐 C++
    /// `m_mapAccFund: NN_AssetKey -> NN_TrdCurrency -> Ndt_Trd_AccFund`.
    pub funds: DashMap<FundsCacheKey, CachedFunds>,
    /// 持仓: `PositionsCacheKey { acc_id, asset_category }` → Vec<position>.
    /// Category 0 preserves the legacy single-bucket path; JP margin and JP
    /// derivative requests use scoped categories to avoid cross-bucket leakage.
    pub positions: DashMap<PositionsCacheKey, Vec<CachedPosition>>,
    /// 当日订单: acc_id → Vec<order>
    pub orders: DashMap<AccKey, Vec<CachedOrder>>,
    /// 交易 cipher: acc_id → cipher bytes (解锁后获得)
    pub ciphers: DashMap<AccKey, Vec<u8>>,
    /// v1.4.48 #1: 订单 broker 映射（order_id_ex → broker_id_used）
    ///
    /// 起源：v1.4.47 P0.1 修了 PlaceOrder 按 `sec_market` 选 broker，但 ModifyOrder /
    /// CancelOrder 仍按 `account.security_firm` 选 broker，导致"在 broker 1007 (US)
    /// 下的单，cancel 去 broker 1019 (CA) 拒" 的 cross-broker 故障。
    ///
    /// 修法：PlaceOrder 成功后把 `(order_id_ex, broker_id_used)` 缓存到这里。
    /// ModifyOrder / CancelOrder 拿到 `c2s.order_id_ex` 后先查 broker_id；
    /// 命中 → 路由到同 broker；未命中 → fallback account.firm 路由。
    ///
    /// 注：cipher 按 sub-account `acc_id` 存储（`ciphers` map）。对照 C++
    /// `NNData_Trd_AccList::m_mapAccCipher`：不同 broker 的账户天然有不同
    /// `nAccID`，存储已隔离（v1.4.49 清理了 v1.4.48 `cipher_brokers` workaround，
    /// 该字段在 v1.4.48 #11 routing 对齐 C++ 后成 dead code）。
    pub order_brokers: DashMap<String, u32>,

    /// v1.4.73 A2 BUG-008 fix: per-account cipher state version counter。
    ///
    /// 外部 tester (v1.4.71) AI 报告 5 步 repro：
    /// ```text
    /// Step 1: unlock pwd       → cache EXECUTED (idem_key=unlock-xxx)
    /// Step 2: 同 body          → cache HIT (正常幂等)
    /// Step 3: EMPTY {} LOCK    → v1.4.39 cipher 清
    /// Step 4: 同 body          → cache HIT 返 stale 成功! (真 bug)
    /// Step 5: place-order      → -401 "交易未解锁"
    /// ```
    ///
    /// v1.4.72 Option C（空 body 不写 cache）只防 step 3 污染，未修 step 4 stale。
    ///
    /// Option A 真修：unlock `idem_key` 构造时纳入**当前 cipher_state_version**，
    /// lock 清 cipher 时 `fetch_add(1, SeqCst)` → version 递增 → step 4 同 body
    /// 得 idem_key **不同**（version=0 → version=1）→ cache miss → 真执行 unlock
    /// 或 backend 校验失败返清晰错误。
    ///
    /// 为啥 SeqCst：unlock_trade handler 可能并发，确保 version 递增对所有
    /// 后续 idem_key 构造 visible（`ciphers.remove()` + `fetch_add()` 顺序严格）。
    ///
    /// 注：version 不持久化 —— daemon restart 重新从 0 开始，等效于"新 cache"，
    /// 之前的 idem entries 也被 cache TTL 清光，零冲突。
    pub cipher_state_versions: DashMap<AccKey, Arc<std::sync::atomic::AtomicU64>>,

    /// v1.4.106 codex 0226 F1+F2: pending OrderConfirm context per
    /// `(acc_id, ftapi_order_id)`.
    ///
    /// PlaceOrder ack 响应里若 `OrderNewRsp.action.type == ORDER_CONFIRM=5` 且
    /// `action.order_confirm.is_some()`, daemon **必须** capture
    /// `CltActionOrderConfirm` 字段, 用于后续 `Trd_ReconfirmOrder` 处理时构造
    /// backend `OrderConfirmReq` (cmd 4728).
    ///
    /// **生命周期**:
    /// - PlaceOrder ack 路径: capture 后 `insert(key, ctx)`
    /// - ReconfirmOrder handler: lookup → 构造 backend req → 收到 `OrderConfirmRsp`
    ///   `result==0` 后 `remove(key)` (一次性消费, 防止重复 confirm)
    /// - TTL: 5min (`ORDER_CONFIRM_CONTEXT_TTL_MS`), `now - inserted_at_ms` 检查;
    ///   stale entry handler 拒绝 + GC 清理
    /// - daemon restart 全清 (内存 cache, backend 重新发 PlaceOrder 即可获新 context)
    ///
    /// 详见 `OrderConfirmContext` doc.
    pub pending_order_confirms: DashMap<OrderConfirmKey, OrderConfirmContext>,
}

impl TrdCache {
    pub fn get_cipher(&self, acc_id: u64) -> Option<Vec<u8>> {
        self.ciphers.get(&acc_id).map(|v| v.clone())
    }

    pub fn set_cipher(&self, acc_id: u64, cipher: Vec<u8>) {
        self.ciphers.insert(acc_id, cipher);
    }

    /// v1.4.73 A2 BUG-008 fix: 读当前账户的 cipher state version（用于 unlock idem_key）。
    ///
    /// 首次访问 acc_id 会初始化为 0。后续每次 lock 清 cipher 会 `fetch_add(1)`。
    /// `idem_key` 构造时把这个 version 纳入 hash → cipher 清后 version 递增 →
    /// 同 body 的 idem_key 不同 → cache miss → 真执行 unlock（或 backend 真校验）。
    pub fn get_cipher_state_version(&self, acc_id: u64) -> u64 {
        let entry = self
            .cipher_state_versions
            .entry(acc_id)
            .or_insert_with(|| Arc::new(AtomicU64::new(0)));
        entry.load(Ordering::SeqCst)
    }

    /// v1.4.73 A2 BUG-008 fix: lock 清 cipher 时调，递增 version → 让下次 unlock
    /// 同 body 得 cache miss。
    ///
    /// 返回 new version（递增后值），便于调用方 log。
    #[must_use]
    pub fn bump_cipher_state_version(&self, acc_id: u64) -> u64 {
        let entry = self
            .cipher_state_versions
            .entry(acc_id)
            .or_insert_with(|| Arc::new(AtomicU64::new(0)));
        entry.fetch_add(1, Ordering::SeqCst) + 1
    }

    /// v1.4.106 codex 0226 F1+F2: PlaceOrder 解析到 `OrderNewRsp.action.order_confirm`
    /// 时调用, 保存上下文用于后续 `Trd_ReconfirmOrder` 构造 backend `OrderConfirmReq`.
    ///
    /// `now_ms` 由 caller 传入 (便于单测注入固定时钟); 真实路径用
    /// `SystemTime::now()`.
    pub fn store_pending_order_confirm(
        &self,
        acc_id: u64,
        ftapi_order_id: u64,
        mut ctx: OrderConfirmContext,
        now_ms: u64,
    ) {
        ctx.inserted_at_ms = now_ms;
        let key = OrderConfirmKey::new(acc_id, ftapi_order_id);
        self.pending_order_confirms.insert(key, ctx);
    }

    /// v1.4.106 codex 0226 F1+F2: ReconfirmOrder handler 入口 lookup, 取出
    /// `(acc_id, ftapi_order_id)` 对应 OrderConfirmContext.
    ///
    /// 返 `None`: cache miss (PlaceOrder 没存 / TTL 过期 / 已被消费). caller 必须
    /// 早 reject loud, **不**允许 silent fallback (避免反模式 D / silent-success).
    ///
    /// `now_ms` 检查 TTL: `now - ctx.inserted_at_ms > ORDER_CONFIRM_CONTEXT_TTL_MS`
    /// 视为 stale → return None + remove (proactive GC).
    pub fn get_pending_order_confirm(
        &self,
        acc_id: u64,
        ftapi_order_id: u64,
        now_ms: u64,
    ) -> Option<OrderConfirmContext> {
        let key = OrderConfirmKey::new(acc_id, ftapi_order_id);
        let ctx = self.pending_order_confirms.get(&key)?.value().clone();
        if now_ms.saturating_sub(ctx.inserted_at_ms) > ORDER_CONFIRM_CONTEXT_TTL_MS {
            // Stale → proactive GC
            self.pending_order_confirms.remove(&key);
            return None;
        }
        Some(ctx)
    }

    /// v1.4.106 codex 0226 F1+F2: ReconfirmOrder backend 成功 (`OrderConfirmRsp.result==0`)
    /// 后调用, 从 cache 删除 (一次性消费, 防重复 confirm).
    ///
    /// 返 `true` 表示真有删除发生; `false` = 已被其他路径消费 / 过期 GC.
    pub fn remove_pending_order_confirm(&self, acc_id: u64, ftapi_order_id: u64) -> bool {
        let key = OrderConfirmKey::new(acc_id, ftapi_order_id);
        self.pending_order_confirms.remove(&key).is_some()
    }

    /// v1.4.106 codex 0226 F1+F2: GC stale OrderConfirmContext entries.
    ///
    /// 用于定时清理 (push dispatcher 收到 ORDER 类 push 时顺便扫一次), 防止
    /// stale ctx 累积. 返回清理掉的条目数.
    pub fn purge_stale_order_confirms(&self, now_ms: u64) -> usize {
        let mut purged = Vec::new();
        for entry in self.pending_order_confirms.iter() {
            if now_ms.saturating_sub(entry.value().inserted_at_ms) > ORDER_CONFIRM_CONTEXT_TTL_MS {
                purged.push(*entry.key());
            }
        }
        let n = purged.len();
        for key in purged {
            self.pending_order_confirms.remove(&key);
        }
        n
    }

    /// v1.4.106 codex 0554 F1 [P1]: 原子性清空所有 cipher + 同步 bump 各账户的
    /// `cipher_state_version`。
    ///
    /// 起源：`/api/admin/reload` 之前的实现是
    /// `bridge.caches.trd_cache.ciphers.clear()` 直接动 `DashMap`，但 **没** bump
    /// `cipher_state_version`。这与 v1.4.73 A2 BUG-008 修复的语义不一致：
    /// lock-trade 路径里 `ciphers.remove()` 之后必跟 `bump_cipher_state_version()`，
    /// 防止旧 idempotency cache entry（unlock idem_key 含 cipher_state_version
    /// hash）在 cipher 被清后仍命中返 stale "cached success"，导致
    /// step 4 / step 5 silent regression。
    ///
    /// admin/reload 漏 bump 的具体后果：
    /// - reload 清光 ciphers
    /// - 客户端再调 unlock-trade 同 body → idem_key 命中（cipher_state_version
    ///   未变）→ 返 stale 成功 → cipher cache 仍空 → place-order `-401` 解锁失败
    ///
    /// 本 helper 把两步打包，**禁止外部直接 `cache.ciphers.clear()`**（那条
    /// 路径 silent skip bump，复活 BUG-008）。所有清 cipher 的 control-plane
    /// 路径（reload / admin / 未来若加更多）必须走本 helper。
    ///
    /// 返回 `(cleared_count, bumped_versions)`：
    /// - `cleared_count`：清掉的 cipher 数（即 reload 前已解锁账户数）
    /// - `bumped_versions`：每个被清 acc_id 的 (acc_id, new_version) 列表，
    ///   便于 log + 客户端调试 idem_key 失效原因
    ///
    /// 与 lock-trade 路径的 bump 行为一致：仅对**实际清掉 cipher** 的 acc_id
    /// 递增 version；从未解锁的账户 cipher_state_version 保持不变。
    ///
    /// 并发：`DashMap::iter()` 期间其他线程的 `ciphers.remove()` /
    /// `ciphers.insert()` 可能 race，但本 helper 用 `remove(&key)` 逐个清，
    /// 拿到 `Some(_)` 才 bump，保证 `version` 单调递增 + 与 `ciphers` 实际
    /// 状态一致。`SeqCst` 保证 bump 对所有后续 `get_cipher_state_version()`
    /// 立即可见。
    #[must_use]
    pub fn clear_all_ciphers_and_bump_versions(&self) -> (usize, Vec<(u64, u64)>) {
        // 先收集 acc_ids（避免持 DashMap iter guard 时 mutate map → deadlock）
        let acc_ids: Vec<u64> = self.ciphers.iter().map(|e| *e.key()).collect();
        let mut cleared_count = 0usize;
        let mut bumped_versions: Vec<(u64, u64)> = Vec::with_capacity(acc_ids.len());
        for acc_id in acc_ids {
            if self.ciphers.remove(&acc_id).is_some() {
                cleared_count += 1;
                let new_ver = self.bump_cipher_state_version(acc_id);
                bumped_versions.push((acc_id, new_ver));
            }
        }
        (cleared_count, bumped_versions)
    }

    pub fn new() -> Self {
        Self {
            accounts: DashMap::new(),
            account_relations: DashMap::new(),
            public_account_ids: DashMap::new(),
            public_projection_ready: AtomicBool::new(false),
            funds: DashMap::new(),
            positions: DashMap::new(),
            orders: DashMap::new(),
            ciphers: DashMap::new(),
            order_brokers: DashMap::new(),
            cipher_state_versions: DashMap::new(),
            // v1.4.106 codex 0226 F1+F2: pending OrderConfirm context cache
            pending_order_confirms: DashMap::new(),
        }
    }

    pub fn set_accounts(&self, accounts: Vec<CachedTrdAcc>) {
        let relations = accounts
            .iter()
            .map(|acc| (acc.acc_id, vec![acc.acc_id]))
            .collect();
        self.set_accounts_with_relations(accounts, relations);
    }

    /// Atomically replace the internal account map and the public projection.
    ///
    /// `relations` mirrors C++ `m_mapIDRelation`: standalone accounts map to
    /// themselves, while universal parents map to their public sub accounts.
    /// This lets `GetAccList` expose only C++ `GetAllSubAccList` output without
    /// losing hidden parent accounts needed by `GetAccItem`-style request paths.
    pub fn set_accounts_with_relations(
        &self,
        accounts: Vec<CachedTrdAcc>,
        relations: Vec<(AccKey, Vec<AccKey>)>,
    ) {
        self.accounts.clear();
        self.account_relations.clear();
        self.public_account_ids.clear();
        for (idx, mut acc) in accounts.into_iter().enumerate() {
            acc.order_index = idx;
            self.accounts.insert(acc.acc_id, acc);
        }
        for (parent_id, sub_ids) in relations {
            for sub_id in &sub_ids {
                self.public_account_ids.insert(*sub_id, ());
            }
            self.account_relations.insert(parent_id, sub_ids);
        }
        self.public_projection_ready.store(true, Ordering::SeqCst);
    }

    #[must_use]
    pub fn get_accounts(&self) -> Vec<CachedTrdAcc> {
        if self.public_projection_ready.load(Ordering::SeqCst) {
            self.public_account_ids
                .iter()
                .filter_map(|e| self.accounts.get(e.key()).map(|acc| acc.value().clone()))
                .collect()
        } else {
            // Backward-compatible test path: many existing tests insert directly
            // into `cache.accounts`. Until production calls set_accounts*, expose
            // all entries, matching the old single-map behavior.
            self.accounts.iter().map(|e| e.value().clone()).collect()
        }
    }

    /// v1.4.106 codex 0932 F2 [P1]: 单 acc_id O(1) 查询 (DashMap key 直查).
    ///
    /// 用途: push_builder 构造 Trd_UpdateOrder / Trd_UpdateOrderFill header
    /// 之前 resolve `trd_env` + `trd_market`. 对齐 C++
    /// `INNData_Trd_AllAccList::GetAccEnv(nAccID)` / `GetAccMkt(nAccID)`.
    ///
    /// 返 `None` = cache miss (账户不在交易 cache 中). caller **必须 loud
    /// return** 不 fallback (sentinel 0 让 client filter reject =
    /// silent-success 反模式).
    #[must_use]
    pub fn lookup_account(&self, acc_id: u64) -> Option<CachedTrdAcc> {
        self.accounts.get(&acc_id).map(|e| e.value().clone())
    }

    /// v1.4.103 (B10): card_num → acc_id resolution helper.
    ///
    /// 接受输入:
    /// - **16 位完整 card_num** (`"1001100100800000"`): 完全匹配 `card_num` 字段.
    /// - **4 位末尾 suffix** (`"7680"`): 匹配 `card_num` 末 4 位 (App 显示格式).
    ///
    /// 返 `Vec<u64>` (matching acc_ids):
    /// - 0 个 → cache 中无 match (caller 决定 warn / abort);
    /// - 1 个 → unique resolution;
    /// - >= 2 个 → ambiguous (caller 必须 reject + log 候选, 不能 silent 接受).
    ///
    /// **空字符串 / 非纯数字 / 长度非 4 / 非 16** → 返 empty Vec (不 panic).
    /// 这是为了让 caller 输入校验 + resolution 双责权: 调用方应该已经校验过格式.
    #[must_use]
    pub fn find_acc_ids_by_card_num(&self, input: &str) -> Vec<u64> {
        // v1.4.103 codex F2.3 (P2): 同时匹配 `card_num` 和 `uni_card_num`
        // (综合账户卡号). 用户故事 B10 描述 App 显示的`保证金综合账户(7680)`末
        // 4 位 — 综合账户的卡号通常 in `uni_card_num`, 普通账户在 `card_num`.
        // 单独只看 `card_num` 会让综合账户用户写 `--allowed-card-nums 7680`
        // 时所有 resolve 都失败 → fail-closed sentinel reject (虽然安全, 但
        // UX 失效, 用户必须 fall back 用 acc_id). 双匹配后 fail-closed
        // sentinel 只在真没账户 match 时触发.
        let accounts = self.get_accounts();
        account_locator::match_card_num_in_records(&accounts, input, None).unwrap_or_default()
    }

    /// **v1.4.106 Finding A** (legacy compat): 不带 currency 维度的 update.
    /// 用 `FundsCacheKey::legacy(acc_id)` 作 key. 适用于:
    /// - 现有 caller 还没改 signature 的 (背景: backend push 不一定知 currency)
    /// - SingleCurrency / sim / Crypto / Forex 账户 (本来就单币种)
    ///
    /// **新 caller 应优先用 [`Self::update_funds_per_currency`]** 显式标
    /// currency 维度, 让 Universal/Futures 账户能存独立 snapshot per currency.
    pub fn update_funds(&self, acc_id: u64, funds: CachedFunds) {
        self.funds.insert(FundsCacheKey::legacy(acc_id), funds);
    }

    /// **v1.4.106 Finding A** (preferred for Universal/Futures): 带 currency
    /// 维度的 update. backend push 时若知 funds 的实际 currency (从 `f.currency`
    /// 字段或 push context 派生), 应该用这个 helper 让多币种 snapshot 不互相覆盖.
    ///
    /// 对齐 C++ `INNData_Trd_Acc::SetAccFund(stKey, enCurrency, ...)`.
    pub fn update_funds_per_currency(
        &self,
        acc_id: u64,
        currency: Option<i32>,
        funds: CachedFunds,
    ) {
        let key = match currency {
            Some(c) => FundsCacheKey::per_currency(acc_id, c),
            None => FundsCacheKey::legacy(acc_id),
        };
        self.funds.insert(key, funds);
    }

    /// Currency + asset-category aware funds update.
    ///
    /// JP derivative accounts use `asset_category` as part of the C++ asset key.
    /// Non-JP/legacy callers should pass `asset_category=0`, which preserves the
    /// existing legacy/per-currency key shape.
    pub fn update_funds_scoped(
        &self,
        acc_id: u64,
        asset_category: i32,
        currency: Option<i32>,
        funds: CachedFunds,
    ) {
        let key = if asset_category != 0 {
            FundsCacheKey::full(acc_id, asset_category, currency)
        } else {
            match currency {
                Some(c) => FundsCacheKey::per_currency(acc_id, c),
                None => FundsCacheKey::legacy(acc_id),
            }
        };
        self.funds.insert(key, funds);
    }

    /// Update the requested funds bucket and also mirror the returned backend
    /// currency bucket when it is known.
    ///
    /// C++ stores `Ndt_Trd_AccFund` under `accFund.enCurrency`
    /// (`INNData_Trd_Acc.cpp::SetAccFund`). A Rust caller may request CMD3020
    /// with `currency=None` because the daemon derived the backend default, but
    /// REST/CLI later read the same account through an explicit effective
    /// currency bucket. Mirroring prevents an older per-currency snapshot from
    /// masking a fresher default refresh.
    pub fn update_funds_scoped_with_returned_currency(
        &self,
        acc_id: u64,
        asset_category: i32,
        requested_currency: Option<i32>,
        funds: CachedFunds,
    ) {
        let returned_currency = funds.currency;
        self.update_funds_scoped(acc_id, asset_category, requested_currency, funds.clone());

        if let Some(returned_currency) = returned_currency
            && requested_currency != Some(returned_currency)
        {
            self.update_funds_scoped(acc_id, asset_category, Some(returned_currency), funds);
        }
    }

    /// **v1.4.106 Finding A**: cache lookup with C++-equivalent fallback.
    ///
    /// 对齐 C++ `INNData_Trd_Acc::GetAccFund(stKey, enCurrency, pAccFund)`:
    /// 先试 requested currency, 找不到则 fallback 到 latest/first available
    /// currency, **返 false** (caller 应看 boolean 决定是否 trust).
    ///
    /// 输入 `currency`:
    /// - `Some(c)`: Universal/Futures 路径, 优先 match per-currency snapshot
    /// - `None`: SingleCurrency 路径, 直接 match `legacy(acc_id)` snapshot
    ///
    /// 输出 `(funds, currency_match)`:
    /// - `(Some(funds), true)`: 精确命中 requested currency snapshot
    /// - `(Some(funds), false)`: 命中 fallback (legacy 或不同 currency 的
    ///   snapshot — caller 应**不要 silent trust**, 至少 log warn 或 surface
    ///   currency mismatch)
    /// - `(None, _)`: 完全 cache miss
    #[must_use]
    pub fn get_funds(&self, acc_id: u64, currency: Option<i32>) -> (Option<CachedFunds>, bool) {
        self.get_funds_scoped(acc_id, 0, currency)
    }

    /// Funds lookup using the same `(acc_id, asset_category, currency)` dimensions
    /// as [`Self::update_funds_scoped`].
    ///
    /// For `asset_category != 0` we require an exact scoped hit. Falling back to a
    /// legacy or another asset-category snapshot would mix JP derivative asset
    /// buckets and silently return the wrong funds.
    #[must_use]
    pub fn get_funds_scoped(
        &self,
        acc_id: u64,
        asset_category: i32,
        currency: Option<i32>,
    ) -> (Option<CachedFunds>, bool) {
        // Step 1: 精确 match
        let exact_key = if asset_category != 0 {
            FundsCacheKey::full(acc_id, asset_category, currency)
        } else {
            match currency {
                Some(c) => FundsCacheKey::per_currency(acc_id, c),
                None => FundsCacheKey::legacy(acc_id),
            }
        };
        if let Some(f) = self.funds.get(&exact_key) {
            return (Some(f.value().clone()), true);
        }
        if asset_category != 0 {
            return (None, false);
        }
        // Step 2: fallback to legacy(acc_id) — backend 不带 currency context
        // push 时落进 legacy key
        if currency.is_some()
            && let Some(f) = self.funds.get(&FundsCacheKey::legacy(acc_id))
        {
            return (Some(f.value().clone()), false);
        }
        // Step 3: fallback to ANY snapshot for this acc_id (latest available
        // currency, 等价于 C++ "first available")
        for entry in self.funds.iter() {
            if entry.key().acc_id == acc_id {
                return (Some(entry.value().clone()), false);
            }
        }
        (None, false)
    }

    pub fn update_positions(&self, acc_id: u64, positions: Vec<CachedPosition>) {
        self.update_positions_scoped(acc_id, 0, positions);
    }

    pub fn update_positions_scoped(
        &self,
        acc_id: u64,
        asset_category: i32,
        positions: Vec<CachedPosition>,
    ) {
        let key = if asset_category != 0 {
            PositionsCacheKey::scoped(acc_id, asset_category)
        } else {
            PositionsCacheKey::legacy(acc_id)
        };
        self.positions.insert(key, positions);
    }

    #[must_use]
    pub fn get_positions_scoped(
        &self,
        acc_id: u64,
        asset_category: i32,
    ) -> Option<Vec<CachedPosition>> {
        let key = if asset_category != 0 {
            PositionsCacheKey::scoped(acc_id, asset_category)
        } else {
            PositionsCacheKey::legacy(acc_id)
        };
        self.positions.get(&key).map(|p| p.value().clone())
    }

    #[must_use]
    pub fn has_positions_scoped(&self, acc_id: u64, asset_category: i32) -> bool {
        let key = if asset_category != 0 {
            PositionsCacheKey::scoped(acc_id, asset_category)
        } else {
            PositionsCacheKey::legacy(acc_id)
        };
        self.positions.contains_key(&key)
    }

    pub fn update_orders(&self, acc_id: u64, orders: Vec<CachedOrder>) {
        self.orders.insert(acc_id, orders);
    }

    /// 更新单个订单（推送场景）
    pub fn upsert_order(&self, acc_id: u64, order: CachedOrder) {
        let mut entry = self.orders.entry(acc_id).or_default();
        if let Some(existing) = entry.iter_mut().find(|o| o.order_id == order.order_id) {
            *existing = order;
        } else {
            entry.push(order);
        }
    }

    /// v1.4.106 codex 0219 Finding 1: resolve cached order context for trade-write
    /// (modify / cancel) handlers.
    ///
    /// 对齐 C++ `APIServer_Trd_ModifyOrder.cpp:251-256` + `:270-271`:
    /// - 优先用 client 传的 `orderIDEx` (= backend `szOrderID`).
    /// - 否则用 `(acc_id, order_id_hash)` 从 cache 找原 order, 取它的
    ///   `szOrderID` + `version` + `exchange*` 字段构造 backend req.
    ///
    /// **fail-closed 语义**: cache miss → 返 `Err`, caller 把错误透传到
    /// FTAPI client 让用户先刷新 `/api/orders` 或传 orderIDEx. 不允许 silent
    /// fall-through 到 `order_id.to_string()` (= 把 hash 当 backend id, 见
    /// pitfall #45 silent-success).
    ///
    /// **入参**:
    /// - `acc_id`: FTAPI `c2s.header.acc_id`.
    /// - `order_id`: FTAPI `c2s.order_id` (hash). `0` 视为 caller 没传, 仅靠
    ///   `order_id_ex` 路径生效.
    /// - `order_id_ex`: FTAPI `c2s.order_id_ex` (= backend szOrderID, 优先).
    ///
    /// **返回**:
    /// - `Ok(snap)`: 命中 cache, 字段已 populated.
    /// - `Err(ResolveOrderError::CacheMiss)`: cache 没存这个 (acc_id, order_id),
    ///   caller 应返清晰提示 "先刷新 /api/orders 或传 orderIDEx".
    /// - `Err(ResolveOrderError::MissingBackendId)`: cache 命中但 backend_order_id
    ///   字段空 (= cache entry 来自老版本, 没存 szOrderID), caller 应返同样提示.
    /// - `Err(ResolveOrderError::InvalidInput)`: 同时缺 order_id 和 order_id_ex.
    pub fn find_order_for_trade_write(
        &self,
        acc_id: u64,
        order_id: u64,
        order_id_ex: Option<&str>,
    ) -> Result<CachedOrderSnapshot, ResolveOrderError> {
        // Case 1: orderIDEx 传了 — 按 order_id_ex 在 cache 找完整 order
        // (匹配 backend `szOrderID`).
        //
        // **v1.4.106 codex 0920 F3 (P1) fail-closed 修**: cache miss 时**不再**
        // 返 default snapshot (= 把 ex 作 backend_id, 其他字段 0). 之前的 fallback
        // 让 modify/cancel handler 用 default `order_version=0` / 空 `exchange*` /
        // 空 `security_type` 发 backend, 后续 backend 拒错 / 路由错. 用户看 daemon
        // 接受了请求实则 silent fail.
        //
        // **新语义**: cache miss + 用户传 ex → `Err(CacheMiss)` 让 handler 透传
        // 给 FTAPI client, 用户应先调 `/api/orders` 刷新或确保 daemon 没 restart
        // 过. 对齐 #45 silent-success anti-pattern (snapshot 不变量必须严格).
        let trimmed_ex = order_id_ex.map(str::trim).filter(|s| !s.is_empty());
        if let Some(ex) = trimmed_ex {
            if let Some(orders) = self.orders.get(&acc_id) {
                if let Some(order) = orders
                    .iter()
                    .find(|o| !o.backend_order_id.is_empty() && o.backend_order_id == ex)
                {
                    return Ok(CachedOrderSnapshot::from_order(order));
                }
                // 未找到完整 order, 也接受老 entry (order_id_ex == ex 但 backend_order_id 空):
                // 把 ex 作 backend_order_id 用 (用户显式传, 信任 caller).
                if let Some(order) = orders.iter().find(|o| o.order_id_ex == ex) {
                    let mut snap = CachedOrderSnapshot::from_order(order);
                    if snap.backend_order_id.is_empty() {
                        snap.backend_order_id = ex.to_string();
                    }
                    return Ok(snap);
                }
            }
            // v1.4.106 codex 0920 F3 (P1): cache miss + 用户传 ex → fail closed.
            // 之前 silent fallback 到 default snapshot (= 0 / "" 字段 + ex 作
            // backend_id), 让 handler 发不完整 backend req. 现在统一 reject,
            // 让用户先刷新 cache.
            return Err(ResolveOrderError::CacheMiss);
        }

        // Case 2: 仅 order_id (hash) — 必须 cache 命中才能查到 backend_order_id.
        if order_id == 0 {
            return Err(ResolveOrderError::InvalidInput);
        }
        let orders = self
            .orders
            .get(&acc_id)
            .ok_or(ResolveOrderError::CacheMiss)?;
        let order = orders
            .iter()
            .find(|o| o.order_id == order_id)
            .ok_or(ResolveOrderError::CacheMiss)?;
        if order.backend_order_id.is_empty() {
            return Err(ResolveOrderError::MissingBackendId);
        }
        Ok(CachedOrderSnapshot::from_order(order))
    }

    /// v1.4.90 S BUG-e4da-009: stub TTL（30s）。
    ///
    /// stub 插入超过此 TTL 且 backend 仍不返该 `order_id` → 视为 backend 永久
    /// 拒单（never accepted into authoritative list）→ evict。
    pub const STUB_TTL_MS: u64 = 30_000;

    /// v1.4.90 S BUG-e4da-009: 当前 unix epoch ms。
    ///
    /// 抽出 helper 是为了 unit test 能用 mock 时间（不直接调）。
    fn now_ms() -> u64 {
        use std::time::{SystemTime, UNIX_EPOCH};
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0)
    }

    /// v1.4.90 S BUG-e4da-009 cache saga 真修：merge backend 权威列表，**保留** stub.
    ///
    /// 历史坑（跨 v1.4.73 → v1.4.89 7 版未真修）：
    /// ```text
    /// 17:36:44.204092 place_order.rs:427  v1.4.82 A2 stub upsert (order_id=X)
    /// 17:36:44.204126 place_order.rs:451  PlaceOrder success
    /// 17:36:44.204138 futu_audit:511      v1.4.38 idempotency: cached
    /// 17:36:44.226531 place_order.rs:488  v1.4.73 A1 orders refreshed count=0  ← 22.4ms 清零
    /// ```
    ///
    /// 根因：v1.4.73 A1 spawn refresh 直接 `orders.insert(acc_id, backend_list)`
    /// **整覆盖**，22ms 内把 v1.4.82 A2 刚 upsert 的 stub 抹掉。client 0ms 查
    /// `/api/orders` 命中 stub OK，但 22ms 后再查就 count=0 —— "**单子消失**" 假象。
    ///
    /// 修法（async-safe）：refresh 不再 `insert` 整覆盖，而是 **merge**：
    /// - backend 返的每个 order: upsert（同 `order_id` 命中 stub → 覆盖且
    ///   `is_stub=false`，不在 → push）
    /// - cache 里 backend 没返的 stub orders（`is_stub=true`）：
    ///   - `now_ms - stub_inserted_at_ms < STUB_TTL_MS` (30s) → **保留**
    ///   - 否则 → evict（backend 永久拒单兜底）
    /// - cache 里 backend 没返的非 stub orders（`is_stub=false`）：
    ///   全清空（backend 是权威，老的非 stub 该被替换）
    ///
    /// 并发语义：用 DashMap entry api 取写锁，整 merge 在锁内完成 → 多个
    /// `merge_preserving_stubs` 调用串行化（顺序与到达顺序一致）。
    /// `upsert_order` 与 `merge_preserving_stubs` 之间也通过同一 entry lock
    /// 排它，不会丢失 stub 插入与 merge 之间的并发更新。
    pub fn merge_preserving_stubs(&self, acc_id: u64, backend_orders: Vec<CachedOrder>) {
        self.merge_preserving_stubs_with_now(acc_id, backend_orders, Self::now_ms());
    }

    /// v1.4.90 S BUG-e4da-009: `merge_preserving_stubs` 的可注入时间版（test 用）。
    ///
    /// 业务代码只调 `merge_preserving_stubs`；本 fn 暴露便于 unit test 模拟
    /// "stub 已超 TTL" / "stub 仍 fresh" 两种边界。
    pub fn merge_preserving_stubs_with_now(
        &self,
        acc_id: u64,
        backend_orders: Vec<CachedOrder>,
        now_ms: u64,
    ) {
        let mut entry = self.orders.entry(acc_id).or_default();

        // 收集既有 cache 里的 stub orders（按 order_id 索引，便于 merge 后判断）
        let existing_stubs: Vec<CachedOrder> =
            entry.iter().filter(|o| o.is_stub).cloned().collect();

        // 重置 entry，按 backend list 重建（每个 backend order 必然 is_stub=false）
        let mut new_orders: Vec<CachedOrder> = backend_orders
            .into_iter()
            .map(|mut o| {
                // backend 是权威，强制 is_stub=false（防 caller 不慎传 stub）
                o.is_stub = false;
                o.stub_inserted_at_ms = 0;
                // v1.4.105 BUG-v1.4.104-001: backend 列表 = broker confirmed 的权威订单,
                // 强制 is_pending_broker_confirm=false (防 caller 不慎传 pending).
                o.is_pending_broker_confirm = false;
                o
            })
            .collect();

        // 把 backend 没返的 stub 按 TTL 保留（已 merge 进 backend list 的不重复）
        let backend_ids: std::collections::HashSet<u64> =
            new_orders.iter().map(|o| o.order_id).collect();
        for stub in existing_stubs {
            if backend_ids.contains(&stub.order_id) {
                // backend 已 ack → 不保留 stub，由 backend 版本胜出
                continue;
            }
            let age = now_ms.saturating_sub(stub.stub_inserted_at_ms);
            if age < Self::STUB_TTL_MS {
                new_orders.push(stub);
            }
            // else: 老 stub 超 TTL，evict（不 push）
        }

        *entry = new_orders;
    }

    /// v1.4.105 BUG-v1.4.104-001 (P0): broker async confirm 到达后清 pending 标志.
    ///
    /// 当 push notice_type=4/5/8/100 (ORDER_UPDATE / ORDER_LIST_UPDATE /
    /// TRADE_STATISTIC / ORDER_NTF) 到达对应 acc_id 时, 调本 fn 把所有
    /// `is_pending_broker_confirm=true` 的 order 翻成 `false`.
    ///
    /// 设计选择: 不按 order_id 精确匹配清 — push notice 通常不带具体 order_id,
    /// 只表 "本 acc 有 order 状态变化". 简化处理: acc 内任何 ORDER 类 push 到
    /// 即视为 broker 已开始处理本 acc 的 stub orders.
    /// 后续 query_orders refresh 会通过 `merge_preserving_stubs` 把 enriched
    /// 版本写入, 替换 stub.
    ///
    /// 返被清的 order 数 (caller 用于 audit log).
    pub fn clear_pending_confirm_for_acc(&self, acc_id: u64) -> usize {
        let mut cleared = 0;
        if let Some(mut entry) = self.orders.get_mut(&acc_id) {
            for o in entry.iter_mut() {
                if o.is_pending_broker_confirm {
                    o.is_pending_broker_confirm = false;
                    cleared += 1;
                }
            }
        }
        cleared
    }

    /// v1.4.106 codex 0226 F4 (P2): selective clear pending confirm by order_ids.
    ///
    /// `clear_pending_confirm_for_acc` 是 acc-level 全清, 但 ORDER push notify
    /// 在 backend 实际带具体 `order_ids` 时(notice_type=4 ORDER_UPDATE 通常
    /// 带), daemon 应**只**清对应订单的 pending flag, 而不是把同账户其他还没
    /// confirm 的 stub 一并误清.
    ///
    /// 触发场景 (`bridge/dispatcher.rs:251-268`):
    /// - notice_type=4/5/9 + 非空 `order_ids` (backend 真带 → 按订单清)
    /// - notice_type=4/5/9 + 空 `order_ids` → fall back to `clear_pending_confirm_for_acc`
    ///
    /// match 逻辑: `o.order_id_ex` (alphanumeric backend szOrderID) 与
    /// `order_ids` 任一相等. 不 match `o.order_id` (FTAPI u64 hash) 因为
    /// backend push 带的 `order_ids` 是 backend 原生 string id.
    ///
    /// 返被清的 order 数 (caller 用于 audit log).
    pub fn clear_pending_confirm_for_orders(&self, acc_id: u64, order_ids: &[String]) -> usize {
        if order_ids.is_empty() {
            return 0;
        }
        let mut cleared = 0;
        if let Some(mut entry) = self.orders.get_mut(&acc_id) {
            for o in entry.iter_mut() {
                if o.is_pending_broker_confirm && order_ids.iter().any(|id| id == &o.order_id_ex) {
                    o.is_pending_broker_confirm = false;
                    cleared += 1;
                }
            }
        }
        cleared
    }

    /// v1.4.105 BUG-v1.4.104-001 (P0): cleanup task 删超时未 confirm 的 pending stub.
    ///
    /// 触发: PlaceOrder spawn 一个 30s 延迟 task, 到点检查 (acc_id, order_id_ex)
    /// 对应的 stub 是否仍 `is_stub=true && is_pending_broker_confirm=true`.
    /// 若是 → 删 stub + warn (push channel 断 / broker 拒单未 push 的兜底).
    ///
    /// **不**简单调 STUB_TTL_MS evict — 那个是 query_orders merge 时的逻辑,
    /// 这里是主动 GC pending stub. 两者互补.
    ///
    /// 返 (purged: bool, reason: 描述), caller 写 audit log.
    pub fn purge_pending_stub_if_still_pending(
        &self,
        acc_id: u64,
        order_id: u64,
    ) -> Option<String> {
        if let Some(mut entry) = self.orders.get_mut(&acc_id) {
            let before = entry.len();
            let mut purged_code = None;
            entry.retain(|o| {
                let should_purge =
                    o.order_id == order_id && o.is_stub && o.is_pending_broker_confirm;
                if should_purge {
                    purged_code = Some(o.code.clone());
                }
                !should_purge
            });
            let after = entry.len();
            if before != after {
                return Some(purged_code.unwrap_or_default());
            }
        }
        None
    }

    /// v1.4.83 §9 F6: 扫全 cache 查 orphan orders.
    ///
    /// **Orphan 定义**: `order_status ∈ {0, 1, 2, 4}` (未达到 Submitted=5
    /// 之前的 in-flight stub) **且** `create_timestamp.is_some()` **且**
    /// `now_secs - create_timestamp > threshold_secs`.
    ///
    /// 含义对应 C++ proto OrderStatus enum (Trd_Common.proto:108):
    /// - 0 = Unsubmitted (未提交) — 极端情况, daemon stub 修后不应该出现 (v1.4.103 P0 hotfix)
    /// - 1 = WaitingSubmit (等待提交) — 条件单 stub 初值, 等触发
    /// - 2 = Submitting (提交中) — 普通单 stub 初值 (v1.4.103 起)
    /// - 4 = TimeOut (处理超时) — 后端回 timeout, 状态未知
    ///
    /// **为什么需要**: v1.4.82 A2 PlaceOrder 成功后直接 upsert stub order
    /// 让 `/api/orders` 立刻可见 (BUG-60b0-002 fix). 后续 push notice_type=
    /// 4/5/8 / re-fetch 把 status 推到 5 (Submitted) / 10/11 (Filled).
    /// 若 push 通道断流 (§9 CMD3020 chain broken), stub 卡住 5min+ = orphan.
    ///
    /// **v1.4.103 P0 (BUG-WUZONG-001)**: stub status 从 0 (proto 定义为
    /// Unsubmitted "未提交", 触发客户端 retry 多下单) 改成 1/2 (WaitingSubmit/
    /// Submitting, 对齐 C++ NNProto_Trd_OrderOp.cpp:483-510). orphan 检测同步
    /// 扩展到 {0, 1, 2, 4} 全 in-flight 状态 — 老 daemon 留下来 status=0 的
    /// 卡死 stub 也能被检测到.
    ///
    /// 返 `Vec<OrphanOrder>`; caller 决定 log 级别 + metric bump.
    #[must_use]
    pub fn scan_orphan_orders(&self, now_secs: f64, threshold_secs: f64) -> Vec<OrphanOrder> {
        let mut orphans = Vec::new();
        for entry in self.orders.iter() {
            let acc_id = *entry.key();
            for order in entry.value().iter() {
                // v1.4.103 codex F2.4 (P2) round 2: 仅扫**真 stub** orders
                // (is_stub == true). 非 stub 的 backend 权威 orders 即使 status
                // 是 0/1/2/4 (in-flight) 也不算 orphan — 它们是 backend 主动
                // 持续推送的真实状态, daemon 不应认为是卡住. 之前 v1.4.103
                // P0 hotfix 加 status filter {0,1,2,4} 但漏 is_stub guard,
                // 导致 backend 长时 Submitting/TimeOut 真单 (5min+) 也被 log
                // 当 orphan / push 通道断流候选 → metrics 噪音.
                if !order.is_stub {
                    continue;
                }
                // v1.4.103 P0: in-flight 状态 (未到 Submitted=5 之前的过渡值)
                // 都视为 stub 卡住候选 — 0 (Unsubmitted) / 1 (WaitingSubmit) /
                // 2 (Submitting) / 4 (TimeOut).
                if !matches!(order.order_status, 0 | 1 | 2 | 4) {
                    continue;
                }
                // **v1.4.106 codex 0920 F7 (P2)**: PlaceOrder stub `create_timestamp=None`
                // (backend 还没 push 真 timestamp). 只用 `create_timestamp.is_some()`
                // 时所有 stub 都被 scanner skip → orphan 检测对最关键场景 (stub
                // 卡死) 失效.
                //
                // 修法: stub (`is_stub=true`) 优先用 `stub_inserted_at_ms`
                // 推算 age (转 secs); 非 stub 用 `create_timestamp` (与原行为
                // 一致, 但本 scanner 已加 `is_stub` guard, 不会 reach 此分支).
                //
                // `stub_inserted_at_ms` 在 PlaceOrder ack 后 upsert 时写, 所以
                // 真 stub 必有非零值. = 0 视为没初始化 (老 cache entry 兼容)
                // → skip.
                let age_secs = if order.is_stub {
                    if order.stub_inserted_at_ms == 0 {
                        // 老 cache entry 未携带 stub_inserted_at_ms (e.g. 升级前
                        // 持久化数据反序列化), skip 而非误报.
                        continue;
                    }
                    let stub_inserted_secs = (order.stub_inserted_at_ms as f64) / 1000.0;
                    let now_unix_secs = now_secs;
                    now_unix_secs - stub_inserted_secs
                } else {
                    // 非 stub: 走 create_timestamp 老路径 (本 scanner 已加
                    // is_stub guard, 此分支不应 reach 但保留兼容).
                    let Some(create_ts) = order.create_timestamp else {
                        continue;
                    };
                    now_secs - create_ts
                };
                if age_secs > threshold_secs {
                    orphans.push(OrphanOrder {
                        acc_id,
                        order_id: order.order_id,
                        order_id_ex: order.order_id_ex.clone(),
                        code: order.code.clone(),
                        age_secs,
                    });
                }
            }
        }
        orphans
    }
}

/// v1.4.83 §9 F6: orphan order 结构化报告.
#[derive(Debug, Clone)]
pub struct OrphanOrder {
    pub acc_id: u64,
    pub order_id: u64,
    pub order_id_ex: String,
    pub code: String,
    /// 距离 create_timestamp 的秒数
    pub age_secs: f64,
}

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