futu_cache/

trd_cache.rs

// 交易数据缓存

mod backend_merge;
mod cipher_exchange;
mod freshness;
mod jp_sub_account;
mod order_broker;
mod order_fill_upsert;
mod order_list;
mod order_relation_snapshot;
mod order_state;
mod order_upsert;
mod snapshots;
mod trade_write_lookup;
mod types;

#[cfg(test)]
mod regression_tests;

pub use cipher_exchange::{
    CipherExchangeAccount, CipherExchangeError, CipherExchangeLease, CipherExchangePublishReport,
};
pub use freshness::{FundsSnapshotLookup, PositionsSnapshotLookup, StampedTradeSnapshot};
pub use order_list::OrphanOrder;
pub use order_relation_snapshot::{
    OrderRelationSnapshotAvailability, OrderRelationSnapshotLookup, OrderRelationSourceChannel,
    OrderRelationSourceToken,
};
pub use types::*;

use arc_swap::ArcSwap;
use dashmap::DashMap;
use futu_core::account_locator;
use futu_domain_trade_account::{
    GlobalStateTradeLoginAccountFacts, required_global_state_trade_login_broker_like_cpp,
};
use std::collections::{BTreeSet, HashSet};
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 excluded from the public
    /// `Trd_GetAccList` projection.
    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, ()>,
    /// Immutable C++ `IsTradeConnLogin()` required-broker snapshot.
    ///
    /// This is derived directly from one `accounts + relations` submission,
    /// then published once after the account maps finish updating. Readers see
    /// either the previous complete set or the new complete set, never a
    /// clear/insert intermediate state.
    required_trade_login_broker_ids: ArcSwap<BTreeSet<u32>>,
    /// JP sub-account id -> public FTAPI `TrdSubAccType`.
    ///
    /// C++ `OrderData_NNToAPI` / `OrderFillData_NNToAPI` expose
    /// `nnOrder.enSubAccType` as `Order.jpAccType` / `OrderFill.jpAccType`.
    /// Rust backend order rows carry the account-protocol sub-account id, so
    /// the bridge stores this side index from CMD2282 account metadata.
    jp_sub_account_types: DashMap<(AccKey, u64), i32>,
    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, StampedTradeSnapshot<CachedFunds>>,
    /// 持仓: `PositionsCacheKey { acc_id, asset_category, currency }` → 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, StampedTradeSnapshot<Vec<CachedPosition>>>,
    /// 组合持仓视图: `PositionsCacheKey { acc_id, asset_category, currency=None }`
    /// → Vec<position>.
    ///
    /// C++ keeps ordinary and combo views in separate stores
    /// (`SetPositionList` vs `SetComboPositionList`) and `optionStrategyView`
    /// chooses which one to read. Keeping the caches separate prevents a combo
    /// refresh from overwriting ordinary position-list output.
    pub combo_positions: DashMap<PositionsCacheKey, StampedTradeSnapshot<Vec<CachedPosition>>>,
    snapshot_freshness: freshness::TradeSnapshotFreshnessStore,
    /// 当日订单: acc_id → Vec<order>
    pub orders: DashMap<AccKey, Vec<CachedOrder>>,
    /// Completeness state for the cache-only v1.8 order-relation projection.
    order_relation_snapshot_states:
        DashMap<AccKey, order_relation_snapshot::OrderRelationSnapshotState>,
    order_relation_next_sequence: AtomicU64,
    /// Serializes relation-relevant order writes with immutable graph reads.
    order_relation_snapshot_locks: DashMap<AccKey, Arc<parking_lot::Mutex<()>>>,
    /// 当日成交: acc_id → Vec<fill>.
    ///
    /// C++ `INNData_Trd_Deal` stores direct/query deal rows and suppresses
    /// unchanged status notifications through `NeedUpdateDeal`.
    pub order_fills: DashMap<AccKey, Vec<CachedOrderFill>>,
    /// 交易 cipher、单账户 revision 与连接 epoch freshness。
    ///
    /// 记录保留空值 tombstone，确保 unlock / lock / reload / CMD2902 publish
    /// 的 revision 在并发下单调递增，旧连接响应不能覆盖新 unlock。
    cipher_records: DashMap<AccKey, cipher_exchange::CipherRecord>,
    cipher_publication_lock: parking_lot::Mutex<()>,
    cipher_broker_generations: DashMap<u32, u64>,
    next_cipher_exchange_request_id: std::sync::atomic::AtomicU64,
    /// 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 路由。
    /// 订单快照更新后会按当前缓存订单 GC stale entry，避免 daemon 长跑时每单
    /// 一个 string key 永久累积。
    ///
    /// 注：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>,
    /// `order_brokers` 的账号归属索引：order_id_ex → acc_id。
    ///
    /// `order_brokers` 本身保持既有 `order_id_ex -> broker_id` 读取契约，供
    /// ModifyOrder / CancelOrder O(1) 路由。这个索引只用于按单个 acc_id 做
    /// stale broker mapping GC，避免每次订单刷新都扫描所有账户的订单快照。
    order_broker_accounts: DashMap<String, AccKey>,
    /// `order_broker_accounts` 的反向索引：acc_id → order_id_ex set。
    ///
    /// `update_orders` / `merge_preserving_stubs` 都是单账号刷新；用这个索引
    /// 可以只遍历该账号曾记录过的 broker mappings，避免在每次账号刷新时
    /// 扫描全量 `order_broker_accounts`。
    order_broker_ids_by_acc: DashMap<AccKey, HashSet<String>>,

    /// C++ `NNProto_Trd_OnPush::m_mapReqIDOrderID` equivalent:
    /// backend write `MsgHeader.req_id` -> backend `order_id`.
    ///
    /// C++ stores this after successful place/modify/cancel ACK and consumes it
    /// when `NOTICE_TYPE_ORDER_OP_RESULT` only carries `order_op_req_ids`. It is
    /// a best-effort helper for an extra order-detail refresh, not the primary
    /// order state source.
    pub order_op_req_orders: DashMap<OrderOpReqKey, String>,

    /// 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 new() -> Self {
        Self {
            accounts: DashMap::new(),
            account_relations: DashMap::new(),
            public_account_ids: DashMap::new(),
            required_trade_login_broker_ids: ArcSwap::from_pointee(BTreeSet::new()),
            jp_sub_account_types: DashMap::new(),
            public_projection_ready: AtomicBool::new(false),
            funds: DashMap::new(),
            positions: DashMap::new(),
            combo_positions: DashMap::new(),
            snapshot_freshness: freshness::TradeSnapshotFreshnessStore::new(),
            orders: DashMap::new(),
            order_relation_snapshot_states: DashMap::new(),
            order_relation_next_sequence: AtomicU64::new(1),
            order_relation_snapshot_locks: DashMap::new(),
            order_fills: DashMap::new(),
            cipher_records: DashMap::new(),
            cipher_publication_lock: parking_lot::Mutex::new(()),
            cipher_broker_generations: DashMap::new(),
            next_cipher_exchange_request_id: std::sync::atomic::AtomicU64::new(0),
            order_brokers: DashMap::new(),
            order_broker_accounts: DashMap::new(),
            order_broker_ids_by_acc: DashMap::new(),
            order_op_req_orders: 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>)>,
    ) {
        let relation_parent_ids = relations
            .iter()
            .map(|(parent_id, _)| *parent_id)
            .collect::<HashSet<_>>();
        let required_trade_login_broker_ids = accounts
            .iter()
            .filter_map(|account| {
                required_global_state_trade_login_broker_like_cpp(
                    GlobalStateTradeLoginAccountFacts {
                        is_relation_parent: relation_parent_ids.contains(&account.acc_id),
                        trd_env: account.trd_env,
                        security_firm: account.security_firm,
                        sort_key: account.sort_key,
                    },
                )
            })
            .collect::<BTreeSet<_>>();

        self.accounts.clear();
        self.account_relations.clear();
        self.public_account_ids.clear();
        self.jp_sub_account_types.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);
        self.required_trade_login_broker_ids
            .store(Arc::new(required_trade_login_broker_ids));
    }

    /// Return the complete immutable broker set required by C++
    /// `IsTradeConnLogin()` for the latest committed account snapshot.
    #[must_use]
    pub fn required_trade_login_broker_ids(&self) -> Arc<BTreeSet<u32>> {
        self.required_trade_login_broker_ids.load_full()
    }

    #[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()
        }
    }

    /// Resolve backend/mobile native account id for request bodies.
    ///
    /// `CachedTrdAcc::intra_acc_id` is the authoritative backend-native id
    /// from CMD2282. The public FTAPI `acc_id` low 32 bits are only the legacy
    /// fallback for old cache entries/tests that do not carry `intra_acc_id`.
    #[must_use]
    pub fn backend_native_account_id_or_public_low_bits(&self, acc_id: AccKey) -> u64 {
        self.accounts
            .get(&acc_id)
            .and_then(|acc| acc.intra_acc_id)
            .filter(|intra| *intra != 0)
            .unwrap_or(acc_id & 0xFFFF_FFFF)
    }

    /// Resolve backend-native account id, but require the account to exist in cache.
    #[must_use]
    pub fn backend_native_account_id_for_existing_account(&self, acc_id: AccKey) -> Option<u64> {
        self.accounts
            .contains_key(&acc_id)
            .then(|| self.backend_native_account_id_or_public_low_bits(acc_id))
    }

    /// 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 时触发.
        // v1.4.111 P2-1 Tier 3 audit comment: fail-closed by-design — empty Vec
        // 表示 "no card_num match", caller (e.g. allowed_card_nums whitelist
        // resolver) 把 empty 当 sentinel reject (v1.4.103 codex F2.3 P2 沉淀),
        // **不**是 silent accept. 非 silent-success risk (audit verified).
        let Ok(query) = account_locator::validate_card_num_query(input) else {
            return Vec::new();
        };
        let mut matches = Vec::new();
        if self.public_projection_ready.load(Ordering::SeqCst) {
            for public_id in self.public_account_ids.iter() {
                if let Some(acc) = self.accounts.get(public_id.key())
                    && account_locator::account_matches_card_num(acc.value(), query)
                {
                    matches.push(acc.value().acc_id);
                }
            }
        } else {
            for acc in self.accounts.iter() {
                if account_locator::account_matches_card_num(acc.value(), query) {
                    matches.push(acc.value().acc_id);
                }
            }
        }
        matches.sort_unstable();
        matches.dedup();
        matches
    }

    pub fn update_orders(&self, acc_id: u64, orders: Vec<CachedOrder>) {
        let relation_lock = self.order_relation_snapshot_lock(acc_id);
        let _relation_guard = relation_lock.lock();
        let availability = if orders
            .iter()
            .any(|order| order.is_stub || order.is_pending_broker_confirm)
        {
            OrderRelationSnapshotAvailability::Stale
        } else {
            OrderRelationSnapshotAvailability::Fresh
        };
        self.orders.insert(acc_id, orders);
        self.publish_order_relation_snapshot_state(acc_id, availability, None);
        self.prune_order_brokers_for_acc(acc_id);
    }
}

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