futu_backend/

quote_sub.rs

// 行情订阅管理
//
// FTAPI SubType → 后端 SubscribeBit 映射
// CMD 6211: 发送订阅请求到后端 (set-state semantics, server 覆盖式)
// CMD 6212: 接收行情推送
//
// **v1.4.106 codex 1131 F1+F2** 重构:
// - submit_global_desired_set 公共 fn 发"全集"CMD6211，并返回 QotSubError。
//   backend reject / decode 错 / timeout 都 loud 返回，caller 据此决定
//   ack-then-commit (F1)；旧 silent legacy wrapper 已在 v1.4.109 删除。
// - backend 覆盖式
//   ("每次请求，客户端都需要提供当前需要订阅的所有股票的所有订阅位")
//   unsub 通过发**新的更小集**实现 — 与 sub 同 entry 复用 (F2)
// - SubscribeRequestParams 携带 session / orderbook_detail / broker_detail /
//   rehab_type / first_push 等 backend req fields (F6)

use std::collections::{BTreeMap, BTreeSet};

use prost::Message;

use futu_core::error::{FutuError, Result};

use crate::conn::BackendConn;
use crate::proto_internal::ft_cmd_stock_quote_sub;
use crate::proto_internal::ft_cmd_stock_quote_sub_data;
use crate::proto_internal::ft_cmd_tick;

mod sub_bits;

pub use sub_bits::{
    SubBitOptions, SubscribeBitInfo, sub_type_to_bit_infos_with_options, sub_type_to_bits,
    sub_type_to_bits_with_options,
};

/// 后端订阅命令 ID
pub const CMD_QOT_SUB: u16 = 6211;
/// 后端推送命令 ID
pub const CMD_QOT_PUSH: u16 = 6212;
/// 逐笔主动拉取命令 ID.
///
/// Ref: C++ `NNBiz_Qot_PullQot.cpp:327-365`
/// `SendPullTickerListReq(... NN_ProtoCmd_Qot_Pull_Ticker ...)`.
pub const CMD_QOT_PULL_TICKER: u16 = 6128;

/// C++ `NN_QuoteTickerKey_FetchLatest ((u64_t)-1)`.
const TICKER_FETCH_LATEST_KEY: u64 = u64::MAX;
/// C++ `NN_QuotePullTicker_LatestTime ((u32_t)-1)`.
const TICKER_LATEST_DATE_TIME_S: u32 = u32::MAX;

mod nn_quote_session {
    pub const RTH: i32 = 0;
    pub const ETH: i32 = 1;
    pub const ALL: i32 = 2;
}

mod tick_period_type {
    // Ref: C++ `FTCmdTick.proto` TickPeriodType.
    pub const NORMAL: u32 = 0;
    pub const BEFORE: u32 = 1;
    pub const AFTER: u32 = 2;
    pub const OVERNIGHT: u32 = 4;
}

/// FTAPI SubType 枚举值
pub mod sub_type {
    pub const BASIC: i32 = 1;
    pub const ORDER_BOOK: i32 = 2;
    pub const TICKER: i32 = 4;
    pub const RT: i32 = 5;
    pub const KL_DAY: i32 = 6;
    pub const KL_5MIN: i32 = 7;
    pub const KL_15MIN: i32 = 8;
    pub const KL_30MIN: i32 = 9;
    pub const KL_60MIN: i32 = 10;
    pub const KL_1MIN: i32 = 11;
    pub const KL_WEEK: i32 = 12;
    pub const KL_MONTH: i32 = 13;
    pub const BROKER: i32 = 14;
    pub const KL_QUARTER: i32 = 15;
    pub const KL_YEAR: i32 = 16;
    pub const KL_3MIN: i32 = 17;
    pub const ORDER_BOOK_ODD: i32 = 22;
}

fn common_session_to_nn(session: i32) -> i32 {
    match session {
        // FTAPI Common.Session: 1=RTH, 2=ETH, 3=ALL.
        2 => nn_quote_session::ETH,
        3 => nn_quote_session::ALL,
        _ => nn_quote_session::RTH,
    }
}

fn ticker_periods_for_nn_session(nn_session: i32) -> Vec<u32> {
    match nn_session {
        nn_quote_session::ALL => vec![
            tick_period_type::NORMAL,
            tick_period_type::BEFORE,
            tick_period_type::AFTER,
            tick_period_type::OVERNIGHT,
        ],
        nn_quote_session::ETH => vec![
            tick_period_type::NORMAL,
            tick_period_type::BEFORE,
            tick_period_type::AFTER,
        ],
        _ => vec![tick_period_type::NORMAL],
    }
}

/// 拉取最新逐笔，用于对齐 C++ Qot_Sub 成功后“订阅逐笔要提前拉一根”。
///
/// Ref:
/// - `APIServer_Qot_Sub.cpp:265-280`: subscribe Ticker 后调用
///   `PullNewestTickerList_Lot(..., 1, false, SessionToNN(enSession))`.
/// - `NNBiz_Qot_PullQot.cpp:126-129`: 非 US 强制 RTH，再从 latest key 拉取。
/// - `NNBiz_Qot_PullQot.cpp:327-365`: CMD6128 body + reserved[0]=head market,
///   reserved[1]=NN_QuoteExType_SECURITY(0).
pub async fn pull_latest_ticker(
    backend: &BackendConn,
    stock_id: u64,
    nn_mkt_type: u8,
    common_session: i32,
    pull_count: u32,
    broker_id: Option<i32>,
) -> Result<ft_cmd_tick::TickRsp> {
    if stock_id == 0 || pull_count == 0 {
        return Err(FutuError::Codec(format!(
            "PullLatestTicker: invalid stock_id={stock_id} pull_count={pull_count}"
        )));
    }

    // C++ `PullNewestTickerList`: only US keeps ETH/ALL; other markets collapse
    // to RTH because backend does not distinguish pre/after sessions there.
    let nn_session = if nn_mkt_type == ftapi_market_to_quote_mkt(11) {
        common_session_to_nn(common_session)
    } else {
        nn_quote_session::RTH
    };

    let req = ft_cmd_tick::TickReq {
        security_id: Some(stock_id),
        date_time_s: Some(TICKER_LATEST_DATE_TIME_S),
        begin_tick_key: Some(TICKER_FETCH_LATEST_KEY),
        tick_count: Some(pull_count),
        tick_period_type: None,
        tick_period_type_ex: ticker_periods_for_nn_session(nn_session),
        req_auth: None,
        end_tick_key: None,
        date_time_s_v2: None,
        // v1.4.110 codex Phase 3 Slice 6a: caller-provided broker_id (crypto-only).
        // 对齐 C++ NNBiz_Qot_PullQot.cpp:344-349 `pbReq.set_broker_id(...)`.
        broker_id,
    };

    let mut reserved = [0u8; 10];
    reserved[0] = nn_mkt_type;
    // reserved[1] = 0 == NN_QuoteExType_SECURITY.

    let frame = backend
        .request_with_reserved(CMD_QOT_PULL_TICKER, req.encode_to_vec(), reserved)
        .await?;
    let rsp: ft_cmd_tick::TickRsp = Message::decode(frame.body.as_ref())?;
    let result = rsp.result.unwrap_or(-1);
    if result != 0 {
        return Err(FutuError::ServerError {
            ret_type: result,
            msg: format!("CMD6128 PullLatestTicker result={result}"),
        });
    }
    Ok(rsp)
}

/// 后端 SubscribeBit 值
pub mod sbit {
    pub const PRICE: u32 = 0;
    pub const STOCK_STATE: u32 = 1;
    pub const STOCK_TYPE_SPECIFIC: u32 = 2;
    pub const ORDER_BOOK: u32 = 3;
    pub const ORDER_BOOK_DETAIL: u32 = 4; // v1.4.106 codex 1131 F6: detail 走独立 bit
    pub const DEAL_STATISTICS: u32 = 5;
    pub const HK_BROKER_QUEUE: u32 = 9;
    pub const HK_BROKER_DETAIL: u32 = 10; // v1.4.106 codex 1131 F6: HK broker detail
    pub const US_PREMARKET_AFTERHOURS: u32 = 13;
    pub const US_LV2_ORDER: u32 = 17;
    pub const TIME_SHARING: u32 = 20;
    pub const KLINE_1MIN: u32 = 21;
    pub const KLINE_3MIN: u32 = 22;
    pub const KLINE_5MIN: u32 = 23;
    pub const KLINE_15MIN: u32 = 24;
    pub const KLINE_30MIN: u32 = 25;
    pub const KLINE_60MIN: u32 = 26;
    pub const KLINE_DAY: u32 = 27;
    pub const KLINE_WEEK: u32 = 28;
    pub const KLINE_MONTH: u32 = 29;
    pub const KLINE_QUARTER: u32 = 30;
    pub const KLINE_YEAR: u32 = 31;
    pub const TICK: u32 = 35;
    pub const MEGER_LV2_ORDER: u32 = 39;
}

/// **v1.4.106 codex 1131 F1+F2**: 单个 (security, sub_types_with_opts) 集合.
/// 让 caller 一次表达 (stock_id, ftapi_market, [(sub_type, SubBitOptions)]).
///
/// **v1.4.110 Phase 2 Slice 4**: 从 3-tuple 升级为 struct, 加 `broker_id` 字段
/// 让 CMD6211 `SecuritySubscribe.broker_id` (FTCmdStockQuoteSubData.proto:269,
/// 对齐 C++ `MktQotSub.cpp:454-463 SecuritySubscribe::set_broker_id`) 真正写
/// 出去. broker_id = `None` 走 no-broker 路径 (普通股); `Some(NonZeroU32)`
/// = crypto multi-broker 隔离.
///
/// 当前 (Phase 2) 所有 caller 都传 `broker_id = None` (Phase 3 才在 SubHandler
/// 入口接 `securityFirm` resolve 写 broker, 见 codex 调研 12:13 增量).
#[derive(Debug, Clone)]
pub struct SecurityWithOpts {
    pub stock_id: u64,
    pub ftapi_market: i32,
    pub sub_types_with_opts: Vec<(i32, SubBitOptions)>,
    /// **v1.4.110 Phase 2 Slice 4**: broker-aware 订阅 (C++ `StockKey(stockID, brokerID)`,
    /// 见 `qot_stock_key.rs` Phase 1 doc).
    /// - `None`: C++ `m_hasBroker = false`, CMD6211 不写 broker_id field
    /// - `Some(N)`: C++ `m_hasBroker = true`, CMD6211 写 broker_id = N
    pub broker_id: Option<std::num::NonZeroU32>,
}

impl SecurityWithOpts {
    /// 不带 broker 的快捷构造 (普通股 / 非 crypto).
    pub fn new(
        stock_id: u64,
        ftapi_market: i32,
        sub_types_with_opts: Vec<(i32, SubBitOptions)>,
    ) -> Self {
        Self {
            stock_id,
            ftapi_market,
            sub_types_with_opts,
            // v1.4.110 codex audit P3 #10: 默认 no-broker. 普通股 / non-crypto
            // 路径走这里; crypto multi-broker 用 `with_broker(...)` 构造.
            broker_id: None,
        }
    }

    /// 带 broker 的快捷构造 (crypto multi-broker). `broker_id = 0` 自动降级 None.
    pub fn with_broker(
        stock_id: u64,
        ftapi_market: i32,
        sub_types_with_opts: Vec<(i32, SubBitOptions)>,
        broker_id: u32,
    ) -> Self {
        Self {
            stock_id,
            ftapi_market,
            sub_types_with_opts,
            broker_id: std::num::NonZeroU32::new(broker_id),
        }
    }
}

/// Explicit backend market group for an empty CMD6211 set-state request.
///
/// Empty requests are used when the last desired item in a market bucket is
/// removed. They still must carry the original backend route market in protocol
/// header `reserved[0]`; routeMarketID=0 is rejected by backend.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct EmptyDesiredMarket {
    pub mkt_type: u8,
    pub is_depth: bool,
}

pub fn empty_desired_market_for_sub(
    ftapi_market: i32,
    sub_type: i32,
) -> Option<EmptyDesiredMarket> {
    let mkt_type = ftapi_market_to_quote_mkt(ftapi_market);
    if mkt_type == 0 {
        return None;
    }
    Some(EmptyDesiredMarket {
        mkt_type,
        is_depth: is_depth_sub_type(sub_type),
    })
}

/// **v1.4.106 codex 1131 F1**: `submit_global_desired_set` 错误类型.
/// 让 caller (SubHandler) 区分 backend reject vs decode err vs timeout →
/// ack-then-commit pattern (F1 P1).
#[derive(Debug)]
pub enum QotSubError {
    /// backend 返了 SubscribeSetRsp.result != 0 (reject), 携带 result 数 + warning.
    BackendRejected { result: i32, warning: i32 },
    /// 响应 decode 失败 — 该批次状态未知, 不写 local state.
    DecodeFailed(String),
    /// 网络 / TCP 错误 (timeout / 连接断). 透传 inner.
    Transport(FutuError),
    /// **v1.4.106 codex 0631 F3 [P2]**: caller 传了一个 backend 不识别的
    /// `ftapi_market` (`ftapi_market_to_quote_mkt → 0`). 防御性 fail loud:
    /// 不发任何 CMD6211, 整批 reject. caller 应早 validate 后再调.
    /// 携带 offending list 让 caller 报清晰错给用户.
    UnsupportedMarket { offending: Vec<i32> },
    /// **v1.4.106 codex 0631 F3 [P2]**: 多 market 分批发送时, 部分市场
    /// backend 失败 (BackendRejected / DecodeFailed) 但其它 OK — 该批次
    /// 是部分应用 (split state). caller 不能当全成功, 需明示用户
    /// "succeeded markets 已生效, failed 需重发".
    PartialMarketFailure { succeeded: Vec<u8>, failed: Vec<u8> },
}

impl std::fmt::Display for QotSubError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            QotSubError::BackendRejected { result, warning } => {
                write!(
                    f,
                    "backend rejected CMD6211: result={result} warning={warning}"
                )
            }
            QotSubError::DecodeFailed(s) => write!(f, "CMD6211 response decode failed: {s}"),
            QotSubError::Transport(e) => write!(f, "CMD6211 transport error: {e}"),
            QotSubError::UnsupportedMarket { offending } => write!(
                f,
                "CMD6211 unsupported ftapi_market(s): {offending:?} \
                 (ftapi_market_to_quote_mkt returned 0). Caller must validate \
                 ftapi_market before submit_global_desired_set."
            ),
            QotSubError::PartialMarketFailure { succeeded, failed } => write!(
                f,
                "CMD6211 partial-market failure: succeeded={succeeded:?} \
                 failed={failed:?}. State is split: succeeded markets are \
                 applied, failed markets need re-submit."
            ),
        }
    }
}

impl std::error::Error for QotSubError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            QotSubError::Transport(e) => Some(e),
            _ => None,
        }
    }
}

impl From<FutuError> for QotSubError {
    fn from(e: FutuError) -> Self {
        QotSubError::Transport(e)
    }
}

/// **v1.4.110 Phase 2 Slice 4**: per-security 输入 — `(stock_id, broker_id, sub_types_with_opts)`
/// 3-tuple. broker_id = `None` 走 no-broker 路径 (普通股, Phase 2 默认),
/// `Some(NonZeroU32)` 走 broker-aware 路径 (crypto multi-broker).
///
/// 抽出 type alias 防 clippy `type_complexity` warn.
pub type SecuritySubscribeInput = (u64, Option<std::num::NonZeroU32>, Vec<(i32, SubBitOptions)>);

/// **v1.4.106 codex 1131 F6**: 构建带 SubBitOptions 的订阅请求 — 真传 session
/// / detail / extended_time 给 backend.
///
/// **v1.4.110 Phase 2 Slice 4**: 升 3-tuple `(stock_id, broker_id, sub_types)`
/// 让 SecuritySubscribe.broker_id 真写出去 (对齐 C++ `MktQotSub.cpp:454-463`):
/// ```cpp
/// SecuritySubscribe::set_security_id(stSecKey.nStockID);
/// if (stSecKey.HasBroker()) {
///     SecuritySubscribe::set_broker_id(stSecKey.GetBrokerID());
/// }
/// ```
/// - `broker_id = None` ⟺ C++ `m_hasBroker = false`, 不调 `set_broker_id`
/// - `broker_id = Some(N)` ⟺ C++ `m_hasBroker = true`, 写 `broker_id = N`
pub fn build_subscribe_req_with_options(
    securities: &[SecuritySubscribeInput],
) -> ft_cmd_stock_quote_sub::SubscribeSetReq {
    build_subscribe_req_with_options_inner(securities, None)
}

pub fn build_keep_subscribe_req_with_options(
    securities: &[SecuritySubscribeInput],
) -> ft_cmd_stock_quote_sub::SubscribeSetReq {
    // Ref: C++ `MktQotSub.cpp:532-535` stores the last normal SubscribeSetReq
    // and sets `timer_sub=1` only for periodic keep-sub replay.
    build_subscribe_req_with_options_inner(securities, Some(1))
}

fn build_subscribe_req_with_options_inner(
    securities: &[SecuritySubscribeInput],
    timer_sub: Option<i32>,
) -> ft_cmd_stock_quote_sub::SubscribeSetReq {
    let mut security_list = Vec::new();

    for (stock_id, broker_id, sub_types_with_opts) in securities {
        let mut bit_info_list = Vec::new();
        for (st, opts) in sub_types_with_opts {
            for info in sub_type_to_bit_infos_with_options(*st, *opts) {
                bit_info_list.push(ft_cmd_stock_quote_sub_data::BitInfo {
                    bit: Some(info.bit),
                    prob: Some(info.prob),
                    prob2: info.prob2,
                    prob2_v2: info.prob2_v2,
                });
            }
        }
        security_list.push(ft_cmd_stock_quote_sub_data::SecuritySubscribe {
            security_id: Some(*stock_id),
            bit_info_list,
            // v1.4.110 Phase 2 Slice 4: broker-aware CMD6211 wire.
            // Some(NZ) → 写 i32 (crypto multi-broker); None → 不写 (普通股).
            broker_id: broker_id.map(|nz| nz.get() as i32),
        });
    }

    ft_cmd_stock_quote_sub::SubscribeSetReq {
        security_list,
        reserved: None,
        timer_sub,
    }
}

/// FTAPI market → 后端 NN_QuoteMktType 映射
///
/// 0 = unknown (caller should reject loud per F7).
pub fn ftapi_market_to_quote_mkt(market: i32) -> u8 {
    match market {
        1 => 1,  // HK → NN_QuoteMktType_HK
        11 => 2, // US → NN_QuoteMktType_US
        21 => 3, // SH → NN_QuoteMktType_SH
        22 => 4, // SZ → NN_QuoteMktType_SZ
        // Internal subscribe-market markers derived from static security info.
        // Public QotMarket maps HK futures back to HK_Security (C++
        // APIServer_Inner_API.cpp::Market_NNToAPI), but CMD6211 reserved[0]
        // still needs the backend NN_QuoteMktType. These values are not
        // client-facing QotMarket values; SubHandler derives them from
        // CachedSecurityInfo.sec_type/mkt_id before entering this function.
        5 => 5,   // HK futures legacy → NN_QuoteMktType_FUT_HK
        6 => 6,   // HK futures current → NN_QuoteMktType_FUT_HK_NEW
        9 => 9,   // HK_OPTIONS → NN_QuoteMktType_HK_OPTIONS
        13 => 13, // SG futures → NN_QuoteMktType_SG_FUTURE
        31 => 15, // SG stock → NN_QuoteMktType_SG_SECURITY
        15 => 7,  // US_OPTIONS → NN_QuoteMktType_US_OPTIONS
        14 => 8,  // US_FUTURE → NN_QuoteMktType_US_FUT
        16 => 16, // JP futures → NN_QuoteMktType_JP_FUTURE
        41 => 25, // JP stock → NN_QuoteMktType_JP_SECURITY
        61 => 27, // MY stock → NN_QuoteMktType_MY_SECURITY
        23 => 10, // SH_KCB → NN_QuoteMktType_SH_KCB
        // v1.4.110 codex Phase 3 Slice 6b: QotMarket_CC_Security=91 (Crypto)
        // → NN_QuoteMktType_CRYPTO=17. C++ `NNBase_Define_Enum.h:517`.
        91 => 17, // Crypto → NN_QuoteMktType_CRYPTO
        // F7 [P2]: AU=51 / CA=71 / FX=81 当前无明确 NN_QuoteMktType 对应。
        // SubHandler 会按 F7 loud reject ftapi_market_to_quote_mkt==0 的 case.
        _ => 0,
    }
}

/// "深度"类 SubType.
///
/// 仅用于旧的 empty desired-set bucket 标记和 trace；非空 CMD6211 不能按
/// normal/depth 拆分。CMD6211 是 backend 覆盖式 set-state，请求必须携带同一
/// market 的完整订阅 bit 集合；C++ `MktQotSub.cpp::Timer_SendSubReq` 也是每个
/// market 发一条 `SubscribeSetReq`，在同一个 `SecuritySubscribe` 内追加全部 bit。
fn is_depth_sub_type(sub_type: i32) -> bool {
    matches!(
        sub_type,
        sub_type::ORDER_BOOK | sub_type::ORDER_BOOK_ODD | sub_type::BROKER
    )
}

/// **v1.4.106 codex 1131 F1+F2**: 给 caller (SubHandler) 的 set-state 接口.
///
/// **语义**: 发送整组 desired (stock_id, ftapi_market, sub_type_with_opts) →
/// backend 返 SubscribeSetRsp. backend "覆盖式" — 所有不在新集合中的旧订阅
/// 自动取消 (per FTCmdStockQuoteSub.proto 设计 doc:
/// "server会覆盖此客户端之前的订阅，并主动推送一次新增加的股票订阅位数据").
///
/// **F1 ack-then-commit**: caller 必须仅在 Ok 后才写 SubscriptionManager state.
/// Err 时 → 不写 state, 返用户 ret_type=-1 + 错误原因.
///
/// **F2 unsub via fresh set**: 退订通过传"new desired set 不含 removed key"
/// 实现, 不调单独 unsub backend cmd. SubHandler 计算
/// `current global - removed` 后调本 fn.
///
/// **max_sub_count**: 响应中 backend 下发的 quota — caller 应调
/// `SubscriptionManager::set_total_quota_from_backend(max_sub_count as u32)`
/// 同步真值 (F5).
///
/// **return 值**: backend 下发的 max_sub_count (caller 据此 update SubscriptionManager
/// 总配额 — F5 P2 dynamic quota).
pub async fn submit_global_desired_set(
    backend: &BackendConn,
    securities: &[SecurityWithOpts],
) -> std::result::Result<i32, QotSubError> {
    submit_global_desired_set_inner(backend, securities, false).await
}

/// Periodically replay the current CMD6211 desired set with `timer_sub=1`.
///
/// Ref: C++ `MktQotSub.cpp:543-580` `Timer_KeepSubStatus()` sends the cached
/// `m_pbLastSendReq`; `MktQotSub.cpp:532-535` sets `timer_sub=1` for that replay.
pub async fn submit_keep_subscribe_desired_set(
    backend: &BackendConn,
    securities: &[SecurityWithOpts],
) -> std::result::Result<i32, QotSubError> {
    submit_global_desired_set_inner(backend, securities, true).await
}

async fn submit_global_desired_set_inner(
    backend: &BackendConn,
    securities: &[SecurityWithOpts],
    is_timer_sub: bool,
) -> std::result::Result<i32, QotSubError> {
    if securities.is_empty() {
        // Empty desired-set submit is ambiguous without the original backend
        // route market. Call `submit_empty_desired_set_for_markets` with the
        // market bucket(s) being cleared instead.
        return Err(QotSubError::UnsupportedMarket { offending: vec![0] });
    }

    // **v1.4.106 codex 0631 F3 [P2]**: 入口 validate 全部 ftapi_market 已知.
    // 任一未知 → Err(UnsupportedMarket{offending}), 不发任何 CMD6211 请求.
    //
    // 老代码 (v1.4.106 codex 1131 F7) 是 silent skip + warn —— 全部未知时返
    // Ok(()) 但 0 backend 请求 = silent success (CLAUDE.md 反模式 D
    // silent-success). 修法: validate first, fail loud, caller 早 reject.
    let offending: Vec<i32> = securities
        .iter()
        .filter_map(|sec| {
            if ftapi_market_to_quote_mkt(sec.ftapi_market) == 0 {
                Some(sec.ftapi_market)
            } else {
                None
            }
        })
        .collect();
    if !offending.is_empty() {
        let mut dedup: Vec<i32> = offending;
        dedup.sort_unstable();
        dedup.dedup();
        tracing::warn!(
            offending = ?dedup,
            "v1.4.106 audit 0631 F3: submit_global_desired_set rejected — unsupported ftapi_market(s)"
        );
        return Err(QotSubError::UnsupportedMarket { offending: dedup });
    }

    // 按后端市场类型分组（C++ 每个市场独立发送订阅请求）。
    //
    // 不再按 normal/depth 拆分: CMD6211 是覆盖式 set-state；同一 market 拆两条
    // 请求会让后发请求覆盖先发请求，表现为 QUOTE 或 ORDER_BOOK 只能有一边持续
    // push。C++ 对齐点: `MktQotSub.cpp::Timer_SendSubReq` 为一个 market 构造
    // 单个 `SubscribeSetReq`，同一个 security 内聚合全部 `SubBit`。
    //
    // v1.4.110 Phase 2 Slice 4: MktGroup 内每项升 `(stock_id, broker_id, sub_types_with_opts)`
    // 3-tuple, 让 build_subscribe_req_with_options 能写出
    // SecuritySubscribe.broker_id.
    type MktGroup = Vec<SecuritySubscribeInput>;
    let mut by_market: BTreeMap<u8, MktGroup> = BTreeMap::new();
    for sec in securities {
        let mkt = ftapi_market_to_quote_mkt(sec.ftapi_market);
        // F3 已 validate, 不可能是 0 — 此处 debug_assert.
        debug_assert!(mkt != 0, "F3 validate 应已拒未知 market");
        by_market.entry(mkt).or_default().push((
            sec.stock_id,
            sec.broker_id,
            sec.sub_types_with_opts.clone(),
        ));
    }

    // **v1.4.106 codex 0631 F3 [P2]**: 多 market 分批发送, 部分失败 → 不能当
    // 全成功. 收集 succeeded / failed market 列表, 返 PartialMarketFailure.
    //
    // 老代码 .await? 短路: 第一个失败市场退出, max_sub_count=0 — caller 看不到
    // 哪些市场已成功 (split state). 改为收集所有结果, 部分失败 loud Err.
    let mut max_sub_count = 0i32;
    let mut succeeded_markets: Vec<u8> = Vec::new();
    let mut failed_markets: Vec<u8> = Vec::new();
    let mut first_transport_err: Option<FutuError> = None;
    for (mkt_type, secs) in &by_market {
        let contains_depth_type = secs
            .iter()
            .any(|(_, _, sub_types)| sub_types.iter().any(|(st, _)| is_depth_sub_type(*st)));
        match submit_subscribe_with_market(
            backend,
            secs,
            *mkt_type,
            contains_depth_type,
            false,
            is_timer_sub,
        )
        .await
        {
            Ok(count) => {
                if count > max_sub_count {
                    max_sub_count = count;
                }
                if !succeeded_markets.contains(mkt_type) {
                    succeeded_markets.push(*mkt_type);
                }
            }
            Err(QotSubError::Transport(e)) => {
                // Transport error: 网络断 / TCP 错 — 整批中断, 透传不分批.
                // 这种情况下 partial 没意义 (后续市场也会同样 Transport 错).
                first_transport_err = Some(e);
                break;
            }
            Err(_) => {
                if !failed_markets.contains(mkt_type) {
                    failed_markets.push(*mkt_type);
                }
            }
        }
    }
    if let Some(e) = first_transport_err {
        return Err(QotSubError::Transport(e));
    }
    if !failed_markets.is_empty() {
        succeeded_markets.sort_unstable();
        failed_markets.sort_unstable();
        tracing::warn!(
            succeeded = ?succeeded_markets,
            failed = ?failed_markets,
            "v1.4.106 audit 0631 F3: submit_global_desired_set partial failure"
        );
        return Err(QotSubError::PartialMarketFailure {
            succeeded: succeeded_markets,
            failed: failed_markets,
        });
    }

    Ok(max_sub_count)
}

pub async fn submit_empty_desired_set_for_markets(
    backend: &BackendConn,
    markets: &[EmptyDesiredMarket],
) -> std::result::Result<i32, QotSubError> {
    if markets.is_empty() {
        return Err(QotSubError::UnsupportedMarket { offending: vec![0] });
    }

    let groups: Vec<EmptyDesiredMarket> = BTreeSet::from_iter(markets.iter().copied())
        .into_iter()
        .collect();
    if groups.iter().any(|g| g.mkt_type == 0) {
        return Err(QotSubError::UnsupportedMarket { offending: vec![0] });
    }

    let mut max_sub_count = 0i32;
    let mut succeeded_markets: Vec<u8> = Vec::new();
    let mut failed_markets: Vec<u8> = Vec::new();
    let mut first_transport_err: Option<FutuError> = None;

    for group in groups {
        match submit_subscribe_with_market(
            backend,
            &[],
            group.mkt_type,
            group.is_depth,
            true,
            false,
        )
        .await
        {
            Ok(count) => {
                max_sub_count = max_sub_count.max(count);
                if !succeeded_markets.contains(&group.mkt_type) {
                    succeeded_markets.push(group.mkt_type);
                }
            }
            Err(QotSubError::Transport(e)) => {
                first_transport_err = Some(e);
                break;
            }
            Err(_) => {
                if !failed_markets.contains(&group.mkt_type) {
                    failed_markets.push(group.mkt_type);
                }
            }
        }
    }

    if let Some(e) = first_transport_err {
        return Err(QotSubError::Transport(e));
    }
    if !failed_markets.is_empty() {
        succeeded_markets.sort_unstable();
        succeeded_markets.dedup();
        failed_markets.sort_unstable();
        failed_markets.dedup();
        return Err(QotSubError::PartialMarketFailure {
            succeeded: succeeded_markets,
            failed: failed_markets,
        });
    }

    Ok(max_sub_count)
}

/// 发送单个 (mkt_type, is_depth) 的 CMD6211 请求, 返 max_sub_count.
///
/// v1.4.110 Phase 2 Slice 4: secs 升 3-tuple `(stock_id, broker_id, sub_types)`
/// 让 build_subscribe_req_with_options 能写出 SecuritySubscribe.broker_id.
async fn submit_subscribe_with_market(
    backend: &BackendConn,
    secs: &[SecuritySubscribeInput],
    mkt_type: u8,
    is_depth: bool,
    is_unsub_all: bool,
    is_timer_sub: bool,
) -> std::result::Result<i32, QotSubError> {
    let req = if is_unsub_all {
        // 全退场景 — security_list 空, reserved=1 让 body 非零 (Windows backend 兼容).
        ft_cmd_stock_quote_sub::SubscribeSetReq {
            security_list: vec![],
            reserved: Some(1),
            timer_sub: None,
        }
    } else if is_timer_sub {
        build_keep_subscribe_req_with_options(secs)
    } else {
        build_subscribe_req_with_options(secs)
    };
    let body = req.encode_to_vec();

    let mut reserved = [0u8; 10];
    reserved[0] = mkt_type;
    // v1.4.110 Phase 2 Slice 4: request_bits 是 trace 用, 不必含 broker_id;
    // SecuritySubscribe.broker_id 在 build_subscribe_req_with_options 内部写.
    let request_bits: Vec<(u64, Vec<(u32, i64)>)> = secs
        .iter()
        .map(|(stock_id, _broker_id, sub_types)| {
            let bits = sub_types
                .iter()
                .flat_map(|(sub_type, opts)| sub_type_to_bits_with_options(*sub_type, *opts))
                .collect();
            (*stock_id, bits)
        })
        .collect();

    tracing::info!(
        mkt_type,
        is_depth,
        is_unsub_all,
        is_timer_sub,
        count = secs.len(),
        body_len = body.len(),
        request_bits = ?request_bits,
        "v1.4.106 audit 1131 F1: sending CMD6211 subscribe (set-state)"
    );

    let resp = backend
        .request_with_reserved(CMD_QOT_SUB, body, reserved)
        .await
        .map_err(QotSubError::Transport)?;

    let parsed: ft_cmd_stock_quote_sub::SubscribeSetRsp = Message::decode(resp.body.as_ref())
        .map_err(|e| QotSubError::DecodeFailed(format!("{e}")))?;

    let result = parsed.result.unwrap_or(-1);
    let warning = parsed.warning_code.unwrap_or(0);
    let max_sub_count = parsed.max_sub_count.unwrap_or(0);

    if result != 0 {
        // **F1 P1**: backend reject → 让 caller 知道 (Err), 不 silent-warn.
        tracing::warn!(
            mkt_type,
            is_depth,
            is_timer_sub,
            result,
            warning,
            request_bits = ?request_bits,
            "v1.4.106 audit 1131 F1: CMD6211 backend rejected"
        );
        return Err(QotSubError::BackendRejected { result, warning });
    }

    tracing::info!(
        mkt_type,
        is_depth,
        is_timer_sub,
        max_sub_count,
        "v1.4.106 audit 1131 F1: CMD6211 ok"
    );
    Ok(max_sub_count)
}

#[cfg(test)]
mod tests;