futu_gateway/

bridge.rs

// 业务桥接层：直连富途后端的完整网关
//
// 启动流程：HTTP 认证 → TCP 连接后端 → 登录 → 获取 session_key → 拉取账户列表 → 启动 API 服务

use std::sync::atomic::AtomicU32;
use std::sync::Arc;

use arc_swap::ArcSwap;
use futu_backend::auth::{self, AuthConfig, AuthResult};
use futu_backend::conn::{BackendConn, PushCallback};
use futu_backend::login;
use futu_cache::login_cache::LoginCache;
use futu_cache::qot_cache::QotCache;
use futu_cache::qot_right::QotRightCache;
use futu_cache::static_data::StaticDataCache;
use futu_cache::trd_cache::{CachedTrdAcc, TrdCache};
use futu_core::error::{FutuError, Result};
use futu_server::listener::ApiServer;
use futu_server::metrics::GatewayMetrics;
use futu_server::push::PushDispatcher;
use futu_server::subscription::SubscriptionManager;

use crate::handlers;

/// 共享的后端连接引用，支持重连后原子替换。
/// 所有 handler 持有同一个 Arc，重连时通过 ArcSwap::store 替换内部值，
/// handler 下次 load() 即可获取新连接。
pub type SharedBackend = Arc<ArcSwap<Option<Arc<BackendConn>>>>;

/// listing_date (Unix timestamp 秒) → "YYYY-MM-DD" (东八区)
fn listing_date_to_str(ts: u32) -> String {
    // C++ 对 listing_date=0 也正常格式化为 "1970-01-01"（不跳过）
    let utc_secs = ts as i64 + 8 * 3600;
    let days = utc_secs.div_euclid(86400);
    let (y, m, d) = days_to_ymd(days as i32 + 719468);
    format!("{y:04}-{m:02}-{d:02}")
}

/// Unix 时间戳(秒) → "YYYY-MM-DD HH:MM:SS" (东八区)
fn timestamp_to_datetime_str(ts: f64) -> String {
    if ts <= 0.0 {
        return String::new();
    }
    let secs = ts as i64;
    // 东八区 offset = +8h
    let utc_secs = secs + 8 * 3600;
    let days = utc_secs.div_euclid(86400);
    let day_secs = utc_secs.rem_euclid(86400);
    let h = day_secs / 3600;
    let m = (day_secs % 3600) / 60;
    let s = day_secs % 60;
    // 从 Unix epoch (1970-01-01) 计算日期
    let (y, mo, d) = days_to_ymd(days as i32 + 719468); // 719468 = days from 0000-03-01 to 1970-01-01
    format!("{y:04}-{mo:02}-{d:02} {h:02}:{m:02}:{s:02}")
}

/// 内部: 从 civil day number 转 (year, month, day)
fn days_to_ymd(z: i32) -> (i32, u32, u32) {
    let era = z.div_euclid(146097);
    let doe = z.rem_euclid(146097) as u32;
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let y = yoe as i32 + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    (y, m, d)
}

/// 后端 market_code (NN_QuoteMktID) → FTAPI QotMarket 映射
/// 对齐 C++: NN_QuoteMktType_From_NN_QuoteMktID() → Market_NNToAPI()
fn market_code_to_qot_market(market_code: u32) -> i32 {
    match market_code {
        // HK market: mainboard, gemboard, nasdaq, extend → QotMarket_HK_Security
        1..=4 => 1,
        // HK futures (old/new) → QotMarket_HK_Security (QotMarket_HK_Future=2 已废弃)
        5 | 6 => 1,
        // HK options (stock option, index option) → QotMarket_HK_Security
        7 | 8 => 1,
        // US market: NYSE, NASDAQ, AMEX, PINK, etc.
        10..=29 => 11, // QotMarket_US_Security
        // CN A-shares
        30 | 32 | 33 | 34 | 36..=40 => 21, // SH, KCB → QotMarket_CNSH_Security
        31 | 35 => 22,                     // SZ, CYB → QotMarket_CNSZ_Security
        // US options
        41..=49 => 11, // QotMarket_US_Security
        // US futures: NYMEX, COMEX, CBOT, CME, CBOE
        60..=109 => 11, // QotMarket_US_Security
        // HK futures (new range)
        110..=119 => 1, // QotMarket_HK_Security
        // Forex
        120..=123 => 81, // QotMarket_FX_Security
        // SG futures
        160..=179 => 31, // QotMarket_SG_Security
        // JP futures
        185..=194 => 41, // QotMarket_JP_Security
        // HK index option
        570 => 1, // QotMarket_HK_Security
        // HK HSI index
        1000..=1049 => 1, // QotMarket_HK_Security
        // US new market codes
        1200..=1249 => 11, // QotMarket_US_Security
        _ => 0,            // Unknown
    }
}

/// 后端 market_code → FTAPI ExchType 映射
/// 参考 C++ Qot_Common.proto ExchType 枚举
fn market_code_to_exch_type(market_code: u32) -> i32 {
    match market_code {
        1 => 1,     // HK MainBoard → ExchType_HK_MainBoard
        2 => 2,     // HK GEM → ExchType_HK_GEMBoard
        3..=9 => 3, // HK other → ExchType_HK_HKEX
        10 => 4,    // US NYSE → ExchType_US_NYSE
        11 => 5,    // US Nasdaq → ExchType_US_Nasdaq
        12 => 7,    // US AMEX → ExchType_US_AMEX
        13 => 6,    // US OTC → ExchType_US_Pink
        30 => 14,   // CN SH → ExchType_CN_SH
        31 => 15,   // CN SZ → ExchType_CN_SZ
        41 => 8,    // US Option → ExchType_US_Option
        _ => 0,     // Unknown
    }
}

/// 网关配置
#[derive(Debug, Clone)]
pub struct GatewayConfig {
    /// 认证服务器 URL
    pub auth_server: String,
    /// 登录账号
    pub account: String,
    /// 密码（明文，内部做 MD5）
    pub password: String,
    /// 后端服务器区域 (gz/sh/hk)
    pub region: String,
    /// API 服务监听地址
    pub listen_addr: String,
    /// 设备 ID
    pub device_id: String,
    /// App 语言: 0=简体 1=繁体 2=英文 (对应 C++ NN_AppLanguage)
    pub app_lang: i32,
}

/// 推送事件 (从 push_callback 通过 channel 发送)
pub enum PushEvent {
    /// 行情推送 → 按订阅转发给客户端
    QuotePush {
        sec_key: String,
        sub_type: i32,
        proto_id: u32,
        body: Vec<u8>,
    },
    /// 广播推送 → 发送给所有已连接的客户端 (到价提醒等)
    BroadcastPush { proto_id: u32, body: Vec<u8> },
    /// 交易数据重查 (收到 _UPDATE 通知后触发)
    TradeReQuery {
        acc_id: u64,
        notice_type: u32,
        /// CMD 4716 中携带的 order_ids (notice_type=4/100 有效)
        order_ids: Vec<String>,
        /// CMD 4716 中携带的 order_fill_ids (notice_type=6/101 有效)
        order_fill_ids: Vec<String>,
    },
}

/// 网关桥接器
pub struct GatewayBridge {
    pub qot_cache: Arc<QotCache>,
    pub trd_cache: Arc<TrdCache>,
    pub static_cache: Arc<StaticDataCache>,
    pub login_cache: Arc<LoginCache>,
    pub subscriptions: Arc<SubscriptionManager>,
    pub backend: SharedBackend,
    pub auth_result: Option<AuthResult>,
    /// 推送事件 channel (push_callback → tokio 任务 → PushDispatcher)
    pub push_tx: Option<tokio::sync::mpsc::Sender<PushEvent>>,
    /// 停牌数据缓存 (stock_id → Vec<timestamp>)
    pub suspend_cache: futu_backend::suspend_data::SuspendCache,
    /// 代码变更数据缓存 (HTTP 下载的 CodeRelation + CodeTemp)
    pub code_change_cache: futu_backend::code_change::CodeChangeCache,
    /// App 语言: 0=简体 1=繁体 2=英文 (对应 C++ NN_AppLanguage)
    pub app_lang: i32,
    /// 历史 K 线请求次数计数器 (GetUsedQuota 返回)
    pub kl_quota_counter: Arc<AtomicU32>,
    /// 行情权限缓存 (CMD 6024 响应)
    pub qot_right_cache: Arc<QotRightCache>,
    /// 运行时监控指标
    pub metrics: Arc<GatewayMetrics>,
}

impl GatewayBridge {
    pub fn new() -> Self {
        Self {
            qot_cache: Arc::new(QotCache::new()),
            trd_cache: Arc::new(TrdCache::new()),
            static_cache: Arc::new(StaticDataCache::new()),
            login_cache: Arc::new(LoginCache::new()),
            subscriptions: Arc::new(SubscriptionManager::new()),
            backend: Arc::new(ArcSwap::new(Arc::new(None))),
            auth_result: None,
            push_tx: None,
            suspend_cache: Arc::new(dashmap::DashMap::new()),
            code_change_cache: futu_backend::code_change::new_cache(),
            app_lang: 0, // 默认简体中文，initialize() 时从 config 更新
            kl_quota_counter: Arc::new(AtomicU32::new(0)),
            qot_right_cache: Arc::new(QotRightCache::new()),
            metrics: Arc::new(GatewayMetrics::new()),
        }
    }

    /// 完整的启动流程: 认证 → 连接后端 → 登录 → 拉取数据
    ///
    /// `verify_cb`: 短信验证码回调（GUI 模式传入，CLI 模式传 None 使用 stdin）
    pub async fn initialize(
        &mut self,
        config: &GatewayConfig,
        verify_cb: Option<auth::VerifyCodeCallback>,
    ) -> Result<tokio::sync::mpsc::Receiver<PushEvent>> {
        self.app_lang = config.app_lang;

        // Step 1: HTTP 认证
        tracing::info!("step 1/4: HTTP authentication...");
        let auth_config = AuthConfig {
            auth_server: config.auth_server.clone(),
            account: config.account.clone(),
            password: config.password.clone(),
            device_id: config.device_id.clone(),
        };
        let auth_result = auth::authenticate_with_callback(&auth_config, verify_cb).await?;
        tracing::info!(user_id = auth_result.user_id, "authentication succeeded");

        // Step 2+3: TCP 连接后端 + 登录（支持重定向）
        let conn_points = auth::conn_points::get_by_region(&config.region);
        // C++ f3clogin 从每个区域随机选一个 IP 并发尝试
        // 我们按区域分组、每组随机选一个、顺序尝试
        let mut backend_addr = {
            use std::collections::HashMap;
            let all = auth::conn_points::ALL;
            let mut by_region: HashMap<&str, Vec<(&str, u16)>> = HashMap::new();
            for (ip, port, region) in all {
                by_region.entry(region).or_default().push((ip, *port));
            }
            // 从每个区域随机选一个
            let mut candidates: Vec<String> = by_region
                .values()
                .map(|ips| {
                    let idx = (std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .subsec_nanos() as usize)
                        % ips.len();
                    let (ip, port) = ips[idx];
                    format!("{ip}:{port}")
                })
                .collect();
            // 随机排列
            let seed = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .subsec_nanos();
            for i in (1..candidates.len()).rev() {
                let j = (seed as usize + i) % (i + 1);
                candidates.swap(i, j);
            }
            candidates
                .first()
                .cloned()
                .unwrap_or_else(|| format!("{}:{}", conn_points[0].0, conn_points[0].1))
        };

        // 推送 channel: push_callback (同步) → tokio 任务 (异步) → PushDispatcher
        let (push_tx, push_rx) = tokio::sync::mpsc::channel::<PushEvent>(4096);
        self.push_tx = Some(push_tx.clone());

        let push_cb: PushCallback = {
            let static_cache = Arc::clone(&self.static_cache);
            let qot_cache = Arc::clone(&self.qot_cache);
            let push_metrics = Arc::clone(&self.metrics);
            Arc::new(move |cmd_id, body| {
                push_metrics
                    .backend_pushes_received
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                match cmd_id {
                    6212 => {
                        // CMD 6212: 行情推送 → 解析 + 缓存 + 转发客户端
                        Self::handle_quote_push(&body, &static_cache, &qot_cache, &push_tx);
                    }
                    4716 => {
                        // CMD 4716: 交易通知 → 解析 + 触发重查
                        Self::handle_trade_notify(&body, &push_tx);
                    }
                    5300 => {
                        // CMD 5300: 消息中心推送（到价提醒等）
                        Self::handle_msg_center_push(&body, &static_cache, &push_tx);
                    }
                    6803 => {
                        // CMD 6803: 到价提醒设置变更通知（内部信号，不直接推送客户端）
                        tracing::debug!(
                            body_len = body.len(),
                            "price reminder settings changed notification"
                        );
                    }
                    _ => {
                        tracing::debug!(cmd_id, body_len = body.len(), "backend push");
                    }
                }
            })
        };

        let (mut backend, login_result) = {
            const MAX_REDIRECTS: usize = 3;
            let mut login_result = None;
            let mut final_backend = None;

            for attempt in 0..=MAX_REDIRECTS {
                tracing::info!(addr = %backend_addr, attempt, "connecting to backend...");
                let conn = BackendConn::connect(&backend_addr, push_cb.clone()).await?;

                tracing::info!("TCP login...");
                match login::tcp_login(
                    &conn,
                    &auth_result,
                    &auth_result.client_key,
                    true,
                    attempt as u32,
                )
                .await
                {
                    Ok(result) => {
                        conn.set_session_key(result.session_key);
                        conn.set_sec_data(result.sec_data);
                        conn.user_id
                            .store(result.user_id as u32, std::sync::atomic::Ordering::Relaxed);
                        login_result = Some(result);
                        final_backend = Some(conn);
                        break;
                    }
                    Err(FutuError::ServerError { ret_type: 1, msg }) => {
                        // 重定向: 从错误消息提取新地址
                        if let Some(new_addr) = msg.strip_prefix("redirect to ") {
                            tracing::info!(redirect_to = new_addr, "login redirect");
                            backend_addr = new_addr.to_string();
                            continue;
                        }
                        return Err(FutuError::ServerError { ret_type: 1, msg });
                    }
                    Err(e) => return Err(e),
                }
            }

            let backend = final_backend.ok_or(FutuError::Codec("too many redirects".into()))?;
            let login_result =
                login_result.expect("login_result must be Some after successful login");
            (backend, login_result)
        };

        tracing::info!(
            user_id = login_result.user_id,
            addr = %backend_addr,
            "login succeeded"
        );

        // Step 3b: 发送 CMD 1321 获取动态 IP 列表（对应 C++ UpdateConnIpAfterLogin）
        tracing::info!("step 3b: fetching ConnIP list (CMD 1321)...");
        let device_id_bytes = config.device_id.as_bytes().to_vec();
        let conn_ip_list = futu_backend::conn_ip::fetch_conn_ip_list(
            &backend,
            login_result.user_id,
            &device_id_bytes,
        )
        .await
        .unwrap_or_else(|e| {
            tracing::warn!(error = %e, "CMD1321 failed (non-fatal)");
            vec![]
        });

        // 如果动态 IP 列表中有推荐地址（REGION_VIRTUAL_RECOMMANDED），
        // 尝试重连到推荐的服务器（C++ 优先使用推荐地址）
        if !conn_ip_list.is_empty() {
            // 优先选推荐地址 (region >= 10001)，其次选第一个
            let best = conn_ip_list
                .iter()
                .find(|p| p.region >= 10001)
                .or_else(|| conn_ip_list.first());

            if let Some(best_point) = best {
                let new_addr = format!("{}:{}", best_point.ip, best_point.port);
                if new_addr != backend_addr {
                    tracing::info!(
                        old_addr = %backend_addr,
                        new_addr = %new_addr,
                        region = best_point.region,
                        "ConnIP: switching to recommended server"
                    );

                    // 重连到新地址
                    match BackendConn::connect(&new_addr, push_cb.clone()).await {
                        Ok(new_conn) => {
                            match login::tcp_login(
                                &new_conn,
                                &auth_result,
                                &auth_result.client_key,
                                true,
                                0,
                            )
                            .await
                            {
                                Ok(new_result) => {
                                    new_conn.set_session_key(new_result.session_key);
                                    new_conn.set_sec_data(new_result.sec_data);
                                    new_conn.user_id.store(
                                        new_result.user_id as u32,
                                        std::sync::atomic::Ordering::Relaxed,
                                    );
                                    backend_addr = new_addr;
                                    backend = new_conn;
                                    tracing::info!(
                                        addr = %backend_addr,
                                        "ConnIP: reconnected to recommended server"
                                    );
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        error = %e,
                                        addr = %new_addr,
                                        "ConnIP: reconnect login failed, keeping original"
                                    );
                                }
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                addr = %new_addr,
                                "ConnIP: connect failed, keeping original"
                            );
                        }
                    }
                }
            }
        }

        // 保存登录状态
        self.login_cache
            .set_login_state(futu_cache::login_cache::LoginState {
                user_id: login_result.user_id as u32,
                is_logged_in: true,
                login_account: config.account.clone(),
                region: config.region.clone(),
                server_addr: backend_addr.clone(),
            });

        let backend = Arc::new(backend);
        self.backend.store(Arc::new(Some(Arc::clone(&backend))));
        let auth_result_for_reconnect = auth_result.clone();
        self.auth_result = Some(auth_result);

        // Step 4: 拉取账户列表
        tracing::info!("step 4/7: fetching account list...");
        self.fetch_account_list(&backend).await?;

        // Step 4b: 后台查询资金/持仓 (不阻塞启动)
        {
            let q_backend = Arc::clone(&backend);
            let q_cache = Arc::clone(&self.trd_cache);
            tokio::spawn(async move {
                futu_backend::trade_query::init_trade_data(&q_backend, &q_cache).await;
            });
        }

        // Step 4c: 请求行情权限 (CMD 6024, 对齐 C++ RequestAPIQuota)
        {
            let qr_backend = Arc::clone(&backend);
            let qr_cache = Arc::clone(&self.qot_right_cache);
            tokio::spawn(async move {
                Self::request_qot_right(&qr_backend, &qr_cache).await;
            });
        }

        // Step 5: 注册市场事件推送
        tracing::info!("step 5/7: registering markets...");
        if let Err(e) = futu_backend::stock_list::register_markets(&backend).await {
            tracing::warn!(error = %e, "market registration failed (non-fatal)");
        }

        // Step 6: 拉取市场状态
        tracing::info!("step 6/7: pulling market status...");
        match futu_backend::stock_list::pull_market_status(&backend).await {
            Ok(statuses) => {
                for s in &statuses {
                    tracing::info!(
                        market_id = s.market_id,
                        status = s.status,
                        text = %s.status_text,
                        "market status"
                    );
                }
            }
            Err(e) => tracing::warn!(error = %e, "market status pull failed (non-fatal)"),
        }

        // Step 7: 后台定期同步股票列表 (C++ SecListDBUpdater 对齐)
        {
            let sync_backend = Arc::clone(&backend);
            let sync_cache = Arc::clone(&self.static_cache);
            tokio::spawn(async move {
                tracing::info!("stock list sync started (background)");

                // 从 SQLite 加载上次同步的数据（对齐 C++ SecListDBHelper）
                let stock_db = match crate::stock_db::StockDb::open() {
                    Ok(db) => {
                        let persisted_version = db.get_version();
                        if persisted_version > 0 {
                            // 从本地数据库恢复缓存
                            match db.load_all() {
                                Ok(items) => {
                                    let mut loaded = 0usize;
                                    for item in &items {
                                        let qot_market =
                                            market_code_to_qot_market(item.market_code);
                                        let key = format!("{qot_market}_{}", item.code);
                                        sync_cache.id_to_key.insert(item.stock_id, key.clone());
                                        sync_cache.securities.insert(
                                            key,
                                            futu_cache::static_data::CachedSecurityInfo {
                                                stock_id: item.stock_id,
                                                market: qot_market,
                                                code: item.code.clone(),
                                                name: item.name_sc.clone(),
                                                lot_size: item.lot_size as i32,
                                                sec_type: item.instrument_type as i32,
                                                list_time: listing_date_to_str(item.listing_date),
                                                warrnt_stock_owner: item.warrnt_stock_owner,
                                                delisting: item.delisting,
                                                exch_type: market_code_to_exch_type(
                                                    item.market_code,
                                                ),
                                                no_search: item.no_search,
                                            },
                                        );
                                        if item.warrnt_stock_owner != 0 {
                                            sync_cache.add_warrant_owner(
                                                item.stock_id,
                                                item.warrnt_stock_owner,
                                            );
                                        }
                                        loaded += 1;
                                    }
                                    tracing::info!(
                                        version = persisted_version,
                                        loaded,
                                        "restored stock list from SQLite"
                                    );
                                }
                                Err(e) => {
                                    tracing::warn!(error = %e, "failed to load stocks from SQLite")
                                }
                            }
                        }
                        Some(db)
                    }
                    Err(e) => {
                        tracing::warn!(error = %e, "failed to open stock SQLite db");
                        None
                    }
                };

                let persisted_version = stock_db.as_ref().map(|db| db.get_version()).unwrap_or(0);
                let version = std::sync::atomic::AtomicU64::new(persisted_version);

                // 首次同步 + 按 next_interval 定期刷新
                let mut retry_delay = 10u64; // 失败重试间隔（C++ UpdateFailRetryPause_S = 10）
                loop {
                    let sync_round = std::sync::atomic::AtomicUsize::new(0);
                    let count = std::sync::atomic::AtomicUsize::new(0);
                    let hk_eqty_count = std::sync::atomic::AtomicUsize::new(0);
                    let deleted_count = std::sync::atomic::AtomicUsize::new(0);
                    let unknown_market_count = std::sync::atomic::AtomicUsize::new(0);
                    let cache_ref = &sync_cache;
                    let db_ref = &stock_db;
                    // 开始 SQLite 事务
                    if let Some(db) = db_ref {
                        let _ = db.begin_batch();
                    }
                    let mut callback = |info: futu_backend::stock_list::StockInfo| {
                        if info.delete_flag {
                            deleted_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                            if let Some(old_key) = cache_ref.id_to_key.remove(&info.stock_id) {
                                cache_ref.securities.remove(&old_key.1);
                            }
                            // SQLite 也删除
                            if let Some(db) = db_ref {
                                let _ = db.delete(info.stock_id);
                            }
                            return;
                        }
                        let qot_market = market_code_to_qot_market(info.market_code);
                        if qot_market == 0 {
                            unknown_market_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                            if unknown_market_count.load(std::sync::atomic::Ordering::Relaxed) <= 5
                            {
                                tracing::warn!(
                                    market_code = info.market_code,
                                    code = %info.code,
                                    instrument_type = info.instrument_type,
                                    "unknown market_code"
                                );
                            }
                        }
                        let key = format!("{qot_market}_{}", info.code);
                        cache_ref.id_to_key.insert(info.stock_id, key.clone());
                        let sec_type = info.instrument_type as i32;
                        if qot_market == 1 && sec_type == 3 {
                            hk_eqty_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        }
                        cache_ref.securities.insert(
                            key,
                            futu_cache::static_data::CachedSecurityInfo {
                                stock_id: info.stock_id,
                                market: qot_market,
                                code: info.code.clone(),
                                name: info.name_sc.clone(),
                                lot_size: info.lot_size as i32,
                                sec_type,
                                list_time: listing_date_to_str(info.listing_date),
                                warrnt_stock_owner: info.warrnt_stock_owner,
                                delisting: info.delisting,
                                exch_type: market_code_to_exch_type(info.market_code),
                                no_search: info.no_search,
                            },
                        );
                        if info.warrnt_stock_owner != 0 {
                            cache_ref.add_warrant_owner(info.stock_id, info.warrnt_stock_owner);
                        }
                        // 写入 SQLite
                        if let Some(db) = db_ref {
                            let _ = db.upsert(&crate::stock_db::DbStockItem {
                                stock_id: info.stock_id,
                                code: info.code.clone(),
                                name_sc: info.name_sc.clone(),
                                market_code: info.market_code,
                                instrument_type: info.instrument_type,
                                lot_size: info.lot_size,
                                delisting: info.delisting,
                                warrnt_stock_owner: info.warrnt_stock_owner,
                                no_search: info.no_search,
                                listing_date: info.listing_date,
                            });
                        }
                        count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    };
                    match futu_backend::stock_list::sync_stock_list(
                        &sync_backend,
                        &version,
                        &mut callback,
                    )
                    .await
                    {
                        Ok(result) => {
                            let interval = result.next_interval_secs.max(30); // 至少 30 秒
                            tracing::info!(
                                total = result.total_stocks,
                                cached = count.load(std::sync::atomic::Ordering::Relaxed),
                                deleted = deleted_count.load(std::sync::atomic::Ordering::Relaxed),
                                hk_eqty = hk_eqty_count.load(std::sync::atomic::Ordering::Relaxed),
                                unknown_market =
                                    unknown_market_count.load(std::sync::atomic::Ordering::Relaxed),
                                next_interval = interval,
                                "stock list sync complete"
                            );
                            retry_delay = 10; // 成功后重置重试间隔
                                              // 提交 SQLite 事务 + 保存 version
                            if let Some(db) = &stock_db {
                                let _ = db.commit_batch();
                                let cur_ver = version.load(std::sync::atomic::Ordering::Relaxed);
                                let _ = db.set_version(cur_ver);
                                tracing::debug!(
                                    version = cur_ver,
                                    db_count = db.count(),
                                    "stock list version persisted to SQLite"
                                );
                            }
                            let round =
                                sync_round.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                            // 前几轮快速同步以加速数据收敛（对齐 C++ 长期运行的 SQLite 状态）
                            let sleep_secs = if round < 5 { 5 } else { interval as u64 };
                            tokio::time::sleep(std::time::Duration::from_secs(sleep_secs)).await;
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                retry_secs = retry_delay,
                                "stock list sync failed, retrying"
                            );
                            tokio::time::sleep(std::time::Duration::from_secs(retry_delay)).await;
                            retry_delay = (retry_delay * 2).min(300); // 指数退避，最大 5 分钟
                        }
                    }
                }
            });
        }

        // 后台异步加载停牌数据 (不阻塞启动)
        {
            let suspend_cache = Arc::clone(&self.suspend_cache);
            tokio::spawn(async move {
                tracing::info!("suspend data download started (background)");
                let loaded = futu_backend::suspend_data::load_suspend_data().await;
                // 将下载的数据合并到 bridge 的 suspend_cache
                for entry in loaded.iter() {
                    suspend_cache.insert(*entry.key(), entry.value().clone());
                }
                tracing::info!(
                    stocks = suspend_cache.len(),
                    "suspend data download complete"
                );
            });
        }

        // 后台异步加载代码变更数据 (不阻塞启动)
        {
            let code_change_cache = Arc::clone(&self.code_change_cache);
            tokio::spawn(async move {
                tracing::info!("code change data download started (background)");
                let loaded = futu_backend::code_change::load_code_change_data().await;
                let count = loaded.len();
                *code_change_cache.write() = loaded;
                tracing::info!(records = count, "code change data download complete");
            });
        }

        // 启动后端心跳 + 断连监控
        let heartbeat_interval =
            std::time::Duration::from_secs(login_result.keep_alive_interval as u64);
        tracing::info!(interval_secs = ?heartbeat_interval, "starting backend heartbeat");
        let heartbeat_handle =
            futu_backend::heartbeat::start_heartbeat(Arc::clone(&backend), heartbeat_interval);

        // 监控心跳任务——心跳退出后自动重连
        // 持有 SharedBackend 的克隆，重连成功后原子替换，所有 handler 自动获取新连接
        {
            let reconnect_config = config.clone();
            let reconnect_auth = auth_result_for_reconnect;
            let reconnect_trd_cache = Arc::clone(&self.trd_cache);
            let reconnect_static_cache = Arc::clone(&self.static_cache);
            let reconnect_subscriptions = Arc::clone(&self.subscriptions);
            let reconnect_metrics = Arc::clone(&self.metrics);
            let shared_backend = Arc::clone(&self.backend);
            let reconnect_push_cb = push_cb;
            tokio::spawn(async move {
                let mut current_heartbeat = heartbeat_handle;

                loop {
                    // 等待当前心跳退出
                    let _ = current_heartbeat.await;
                    tracing::error!("heartbeat exited — attempting reconnect...");

                    // 标记后端离线
                    reconnect_metrics
                        .backend_online
                        .store(0, std::sync::atomic::Ordering::Relaxed);

                    // 立即清空共享连接，防止 handler 继续使用已断开的连接
                    shared_backend.store(Arc::new(None));

                    // 指数退避重连（无限重试，与 C++ OnDisConnectRelpyAll 对齐）
                    let mut delay = 3u64;
                    loop {
                        tracing::info!(delay_secs = delay, "reconnecting to backend...");
                        tokio::time::sleep(std::time::Duration::from_secs(delay)).await;

                        // 重新连接 + 登录
                        let conn_points =
                            auth::conn_points::get_by_region(&reconnect_config.region);
                        let (addr, _port) = match conn_points.first() {
                            Some(p) => p,
                            None => {
                                reconnect_metrics
                                    .backend_reconnect_failures
                                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                delay = (delay * 2).min(60);
                                continue;
                            }
                        };
                        let backend_addr = format!("{addr}:{_port}");

                        let conn =
                            match BackendConn::connect(&backend_addr, reconnect_push_cb.clone())
                                .await
                            {
                                Ok(c) => c,
                                Err(e) => {
                                    reconnect_metrics
                                        .backend_reconnect_failures
                                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                    tracing::warn!(error = %e, "reconnect TCP failed");
                                    delay = (delay * 2).min(60);
                                    continue;
                                }
                            };

                        match login::tcp_login(
                            &conn,
                            &reconnect_auth,
                            &reconnect_auth.client_key,
                            true,
                            0,
                        )
                        .await
                        {
                            Ok(lr) => {
                                conn.set_session_key(lr.session_key);
                                conn.set_sec_data(lr.sec_data);
                                conn.user_id
                                    .store(lr.user_id as u32, std::sync::atomic::Ordering::Relaxed);
                                tracing::info!(user_id = lr.user_id, "reconnect login succeeded");

                                // 重新注册市场
                                let _ = futu_backend::stock_list::register_markets(&conn).await;

                                // 重新查询资金/持仓
                                let conn_arc = Arc::new(conn);
                                futu_backend::trade_query::init_trade_data(
                                    &conn_arc,
                                    &reconnect_trd_cache,
                                )
                                .await;

                                // 原子替换共享连接——所有 handler 立刻可见
                                shared_backend.store(Arc::new(Some(Arc::clone(&conn_arc))));
                                tracing::info!("shared backend connection updated");

                                // ===== 重新订阅行情 (reconnect hardening) =====
                                // 收集所有客户端的行情订阅，重新发送 CMD 6211
                                let resub_count = resubscribe_quotes(
                                    &conn_arc,
                                    &reconnect_subscriptions,
                                    &reconnect_static_cache,
                                )
                                .await;
                                if resub_count > 0 {
                                    reconnect_metrics.resubscribe_ops.fetch_add(
                                        resub_count as u64,
                                        std::sync::atomic::Ordering::Relaxed,
                                    );
                                    tracing::info!(
                                        securities = resub_count,
                                        "quote re-subscription complete"
                                    );
                                }

                                // 标记后端在线 + 更新指标
                                reconnect_metrics
                                    .backend_online
                                    .store(1, std::sync::atomic::Ordering::Relaxed);
                                reconnect_metrics
                                    .backend_reconnects
                                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                reconnect_metrics.last_reconnect_ms.store(
                                    std::time::SystemTime::now()
                                        .duration_since(std::time::UNIX_EPOCH)
                                        .unwrap_or_default()
                                        .as_millis() as u64,
                                    std::sync::atomic::Ordering::Relaxed,
                                );

                                // 重启心跳——保存 handle 用于下次循环
                                let hb_interval =
                                    std::time::Duration::from_secs(lr.keep_alive_interval as u64);
                                current_heartbeat = futu_backend::heartbeat::start_heartbeat(
                                    Arc::clone(&conn_arc),
                                    hb_interval,
                                );

                                tracing::info!("reconnect complete");
                                break; // 跳出重试循环，回到外层 loop 等待新心跳退出
                            }
                            Err(e) => {
                                reconnect_metrics
                                    .backend_reconnect_failures
                                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                tracing::warn!(error = %e, "reconnect login failed");
                                delay = (delay * 2).min(60);
                            }
                        }
                    }
                }
            });
        }

        tracing::info!("gateway initialization complete");
        Ok(push_rx)
    }

    /// 处理后端 CMD 6212 行情推送
    /// 解析 QuotePush → 更新 qot_cache → 通过 channel 发送 FTAPI 推送事件
    fn handle_quote_push(
        body: &[u8],
        static_cache: &futu_cache::static_data::StaticDataCache,
        qot_cache: &futu_cache::qot_cache::QotCache,
        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
    ) {
        use futu_backend::proto_internal::ft_cmd_stock_quote_sub;

        let push: ft_cmd_stock_quote_sub::QuotePush = match prost::Message::decode(body) {
            Ok(p) => p,
            Err(_) => {
                // Reply-type CMD6212 (首次订阅快照) 格式不同于标准 QuotePush:
                // 4字节前缀 + SecurityQuote-like protobuf (可能末尾截断)
                // 直接从中提取 BitQuote 数据并处理
                Self::handle_quote_push_reply(body, static_cache, qot_cache, push_tx);
                return;
            }
        };

        tracing::debug!(
            sec_count = push.security_qta_list.len(),
            "CMD6212 push parsed"
        );

        for sec_quote in &push.security_qta_list {
            let stock_id = match sec_quote.security_id {
                Some(id) => id,
                None => continue,
            };

            // stock_id → security key + info
            let sec_key = match static_cache.id_to_key.get(&stock_id) {
                Some(k) => k.clone(),
                None => {
                    tracing::debug!(stock_id, "CMD6212: stock_id not found in id_to_key cache");
                    continue;
                }
            };
            let sec_info = static_cache.get_security_info(&sec_key);

            for bit_quote in &sec_quote.bit_qta_list {
                let bit = bit_quote.bit.unwrap_or(u32::MAX);
                let data = match &bit_quote.data {
                    Some(d) => d.as_slice(),
                    None => continue,
                };

                tracing::debug!(
                    stock_id,
                    %sec_key,
                    bit,
                    data_len = data.len(),
                    "CMD6212 processing bit"
                );

                match bit {
                    0 => {
                        // SBIT_PRICE: 解析 + 缓存 + 构建 FTAPI push
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_price_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    1 => {
                        // SBIT_STOCK_STATE: 停牌状态 → 更新缓存 + 发推送
                        Self::parse_stock_state(data, &sec_key, qot_cache);
                        // C++ 每个 bit 处理完都推送当前累积状态
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::build_basic_qot_push_from_cache(&sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    3 => {
                        // SBIT_ORDER_BOOK: 买卖盘
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_order_book_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    5 => {
                        // SBIT_DEAL_STATISTICS: 成交量/额/OHLC → 更新缓存 + 发推送
                        Self::parse_deal_statistics(data, &sec_key, qot_cache);
                        // C++ 每个 bit 处理完都推送当前累积状态
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::build_basic_qot_push_from_cache(&sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    9 => {
                        // SBIT_HK_BROKER_QUEUE: 港股经纪队列
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_broker_queue_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    35 => {
                        // SBIT_TICK: 逐笔成交
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_ticker_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    20 => {
                        // SBIT_TIME_SHARING: 分时数据
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_rt_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    21..=31 => {
                        // SBIT_KLINE_*: K 线数据 (1min=21, 3min=22, ..., year=31)
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_kline_to_push(data, bit, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    _ => {}
                }
            }
        }
    }

    /// 处理 Reply-type CMD6212 帧
    ///
    /// 后端在首次订阅时发送 Reply-type (非 Push-type) 的 CMD6212 帧,
    /// 格式: 4字节 BE 前缀 + 1个 varint (变长) + QuotePush protobuf (可能末尾截断).
    /// 跳过前缀和 varint 后, 直接用 prost 解码为 QuotePush.
    fn handle_quote_push_reply(
        body: &[u8],
        static_cache: &futu_cache::static_data::StaticDataCache,
        qot_cache: &futu_cache::qot_cache::QotCache,
        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
    ) {
        use futu_backend::proto_internal::ft_cmd_stock_quote_sub;

        if body.len() < 8 {
            return;
        }

        // Reply 格式: 4字节 BE 前缀 + 1个 varint + QuotePush protobuf
        // 跳过 4 字节前缀, 然后读取并跳过一个 varint
        let data = &body[4..];
        let mut pos = 0;
        // 读取并跳过一个 varint
        while pos < data.len() {
            let b = data[pos];
            pos += 1;
            if b & 0x80 == 0 {
                break;
            }
        }

        if pos >= data.len() {
            tracing::debug!(
                body_len = body.len(),
                "CMD6212 reply: too short after prefix"
            );
            return;
        }

        let proto_data = &data[pos..];

        // 尝试解码为 QuotePush (末尾可能截断, 所以用 decode_length_delimited 的方式手动处理)
        let push: ft_cmd_stock_quote_sub::QuotePush = match prost::Message::decode(proto_data) {
            Ok(p) => p,
            Err(_) => {
                // 末尾截断时 prost 可能失败, 尝试只解析到可用长度
                // QuotePush = { repeated SecurityQuote security_qta_list = 1; }
                // 手动解析 field 1 (length-delimited) 条目
                match Self::parse_quote_push_truncated(proto_data) {
                    Some(p) => p,
                    None => {
                        tracing::debug!(
                            body_len = body.len(),
                            proto_len = proto_data.len(),
                            "CMD6212 reply: decode failed"
                        );
                        return;
                    }
                }
            }
        };

        tracing::debug!(
            sec_count = push.security_qta_list.len(),
            "CMD6212 reply parsed as QuotePush"
        );

        // 复用标准 QuotePush 处理逻辑
        for sec_quote in &push.security_qta_list {
            let stock_id = match sec_quote.security_id {
                Some(id) => id,
                None => continue,
            };

            let sec_key = match static_cache.id_to_key.get(&stock_id) {
                Some(k) => k.clone(),
                None => {
                    tracing::debug!(stock_id, "CMD6212 reply: stock_id not in cache");
                    continue;
                }
            };
            let sec_info = static_cache.get_security_info(&sec_key);

            for bit_quote in &sec_quote.bit_qta_list {
                let bit = bit_quote.bit.unwrap_or(u32::MAX);
                let data = match &bit_quote.data {
                    Some(d) => d.as_slice(),
                    None => continue,
                };

                tracing::debug!(
                    stock_id,
                    %sec_key,
                    bit,
                    data_len = data.len(),
                    "CMD6212 reply processing bit"
                );

                match bit {
                    0 => {
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_price_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    1 => {
                        Self::parse_stock_state(data, &sec_key, qot_cache);
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::build_basic_qot_push_from_cache(&sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    3 => {
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_order_book_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    5 => {
                        Self::parse_deal_statistics(data, &sec_key, qot_cache);
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::build_basic_qot_push_from_cache(&sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    9 => {
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_broker_queue_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    35 => {
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_ticker_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    20 => {
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_rt_to_push(data, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    21..=31 => {
                        if let Some(ref info) = sec_info {
                            if let Some(event) =
                                Self::parse_kline_to_push(data, bit, &sec_key, info, qot_cache)
                            {
                                let _ = push_tx.try_send(event);
                            }
                        }
                    }
                    _ => {
                        tracing::debug!(bit, "CMD6212 reply: unhandled bit");
                    }
                }
            }
        }
    }

    /// 解析可能截断的 QuotePush protobuf
    ///
    /// 当末尾截断时 prost::Message::decode 会失败,
    /// 这里手动解析 field 1 (repeated SecurityQuote) 并忽略截断的尾部.
    fn parse_quote_push_truncated(
        data: &[u8],
    ) -> Option<futu_backend::proto_internal::ft_cmd_stock_quote_sub::QuotePush> {
        use futu_backend::proto_internal::ft_cmd_stock_quote_sub;

        let mut entries = Vec::new();
        let mut pos = 0;

        while pos < data.len() {
            // 读取 tag
            let (tag, new_pos) = Self::read_varint(data, pos);
            if tag == 0 || new_pos >= data.len() {
                break;
            }
            pos = new_pos;
            let field_num = tag >> 3;
            let wire_type = tag & 7;

            if wire_type != 2 {
                break; // 只处理 length-delimited
            }

            let (length, new_pos) = Self::read_varint(data, pos);
            pos = new_pos;
            let length = length as usize;

            if pos + length > data.len() {
                // 截断 — 尝试解析已有部分
                if field_num == 1 {
                    let partial = &data[pos..];
                    if let Ok(sq) = prost::Message::decode(partial) {
                        entries.push(sq);
                    }
                }
                break;
            }

            if field_num == 1 {
                let content = &data[pos..pos + length];
                if let Ok(sq) = prost::Message::decode(content) {
                    entries.push(sq);
                }
            }
            pos += length;
        }

        if entries.is_empty() {
            return None;
        }

        Some(ft_cmd_stock_quote_sub::QuotePush {
            security_qta_list: entries,
        })
    }

    /// 从 protobuf 字节流读取 varint
    fn read_varint(data: &[u8], mut pos: usize) -> (u64, usize) {
        let mut result: u64 = 0;
        let mut shift = 0;
        while pos < data.len() {
            let b = data[pos];
            result |= ((b & 0x7f) as u64) << shift;
            pos += 1;
            if b & 0x80 == 0 {
                break;
            }
            shift += 7;
            if shift >= 64 {
                break;
            }
        }
        (result, pos)
    }

    /// 解析 SBIT_PRICE → 更新缓存 + 构建 Qot_UpdateBasicQot push event
    fn parse_price_to_push(
        data: &[u8],
        sec_key: &str,
        sec_info: &futu_cache::static_data::CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) -> Option<PushEvent> {
        use futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data;

        let price: ft_cmd_stock_quote_coverage_data::Price = prost::Message::decode(data).ok()?;

        let cur_price = price.price_nominal.unwrap_or(0) as f64 / 1_000_000_000.0;
        let last_close = price.price_last_close.unwrap_or(0) as f64 / 1_000_000_000.0;
        let update_time_ms = price.exchange_data_time_ms.unwrap_or(0);
        let update_ts = update_time_ms as f64 / 1000.0;

        if cur_price <= 0.0 {
            return None;
        }

        // 更新缓存 — 合并已有 OHLCV (来自 SBIT_DEAL_STATISTICS) + 停牌状态 (来自 SBIT_STOCK_STATE)
        let existing = qot_cache.get_basic_qot(sec_key);
        let (open, high, low, vol, tnov, tnov_rate, amp, suspended) = match &existing {
            Some(e) => (
                e.open_price,
                e.high_price,
                e.low_price,
                e.volume,
                e.turnover,
                e.turnover_rate,
                e.amplitude,
                e.is_suspended,
            ),
            None => (0.0, 0.0, 0.0, 0, 0.0, 0.0, 0.0, false),
        };
        let cached = futu_cache::qot_cache::CachedBasicQot {
            cur_price,
            last_close_price: last_close,
            open_price: open,
            high_price: high,
            low_price: low,
            volume: vol,
            turnover: tnov,
            turnover_rate: tnov_rate,
            amplitude: amp,
            is_suspended: suspended,
            update_time: timestamp_to_datetime_str(update_ts),
            update_timestamp: update_ts,
        };
        qot_cache.update_basic_qot(sec_key, cached.clone());

        // 构建 FTAPI Qot_UpdateBasicQot::Response — 包含完整 OHLCV
        let security = futu_proto::qot_common::Security {
            market: sec_info.market,
            code: sec_info.code.clone(),
        };
        let basic_qot = futu_proto::qot_common::BasicQot {
            security,
            name: Some(sec_info.name.clone()),
            is_suspended: cached.is_suspended,
            list_time: sec_info.list_time.clone(),
            price_spread: 0.0,
            update_time: cached.update_time.clone(),
            high_price: cached.high_price,
            open_price: cached.open_price,
            low_price: cached.low_price,
            cur_price,
            last_close_price: last_close,
            volume: cached.volume,
            turnover: cached.turnover,
            turnover_rate: cached.turnover_rate,
            amplitude: cached.amplitude,
            dark_status: None,
            option_ex_data: None,
            list_timestamp: None,
            update_timestamp: Some(update_ts),
            pre_market: None,
            after_market: None,
            overnight: None,
            future_ex_data: None,
            sec_status: None,
            warrant_ex_data: None,
        };

        let resp = futu_proto::qot_update_basic_qot::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_update_basic_qot::S2c {
                basic_qot_list: vec![basic_qot],
            }),
        };

        Some(PushEvent::QuotePush {
            sec_key: sec_key.to_string(),
            sub_type: 1, // SubType_Basic
            proto_id: futu_core::proto_id::QOT_UPDATE_BASIC_QOT,
            body: prost::Message::encode_to_vec(&resp),
        })
    }

    /// 处理后端 CMD 4716 交易通知
    /// C++ NoticeType: 1=CASH, 2=PSTN, 3=PSTN_LIST, 4=ORDER, 5=ORDER_LIST,
    /// 6=FILL, 7=FILL_LIST, 9=OP_RESULT, 11=ASSET, 12=QUOTE, 100=ORDER_NTF, 101=FILL_NTF
    fn handle_trade_notify(body: &[u8], push_tx: &tokio::sync::mpsc::Sender<PushEvent>) {
        use futu_backend::proto_internal::order_sys_interface;

        let notify: order_sys_interface::NotifyMsg = match prost::Message::decode(body) {
            Ok(n) => n,
            Err(e) => {
                tracing::debug!(error = %e, "CMD4716 decode failed");
                return;
            }
        };

        let notice_type = notify.notice_type.unwrap_or(0);
        let acc_id = notify
            .msg_header
            .as_ref()
            .and_then(|h| h.account_id)
            .unwrap_or(0);

        let order_ids = notify.order_ids.clone();
        let fill_ids = notify.order_fill_ids.clone();

        match notice_type {
            1 => {
                // NOTICE_TYPE_CASH_UPDATE → 重查资金
                tracing::info!(acc_id, "cash update → re-query funds");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: vec![],
                    order_fill_ids: vec![],
                });
            }
            2 => {
                // NOTICE_TYPE_PSTN_UPDATE → 重查持仓
                tracing::info!(acc_id, symbols = ?notify.symbols, "position update");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: vec![],
                    order_fill_ids: vec![],
                });
            }
            3 => {
                // NOTICE_TYPE_PSTN_LIST_UPDATE → 重查持仓列表
                tracing::info!(acc_id, "position list update");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: vec![],
                    order_fill_ids: vec![],
                });
            }
            4 => {
                // NOTICE_TYPE_ORDER_UPDATE → 重查订单 + 推送
                tracing::info!(acc_id, orders = ?order_ids, "order update");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: order_ids.clone(),
                    order_fill_ids: vec![],
                });
            }
            5 => {
                // NOTICE_TYPE_ORDER_LIST_UPDATE → 重查订单列表
                tracing::info!(acc_id, "order list update");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: vec![],
                    order_fill_ids: vec![],
                });
            }
            6 => {
                // NOTICE_TYPE_ORDER_FILL_UPDATE → 重查成交 + 推送
                tracing::info!(acc_id, fills = ?fill_ids, "fill update");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: vec![],
                    order_fill_ids: fill_ids.clone(),
                });
            }
            7 => {
                // NOTICE_TYPE_ORDER_FILL_LIST_UPDATE → 重查成交列表
                tracing::info!(acc_id, "fill list update");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: vec![],
                    order_fill_ids: vec![],
                });
            }
            9 => {
                // NOTICE_TYPE_ORDER_OP_RESULT → 重查订单
                tracing::info!(acc_id, req_ids = ?notify.order_op_req_ids, "order op result");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: order_ids.clone(),
                    order_fill_ids: vec![],
                });
            }
            11 | 12 => {
                // NOTICE_TYPE_ASSET_UPDATE / QUOTE_UPDATE → 重查资金
                tracing::info!(acc_id, notice_type, "asset/quote update");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type,
                    order_ids: vec![],
                    order_fill_ids: vec![],
                });
            }
            100 => {
                // NOTICE_TYPE_ORDER_NTF → 订单推送 + 重查
                tracing::info!(acc_id, orders = ?order_ids, "order notification");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type: 4,
                    order_ids: order_ids.clone(),
                    order_fill_ids: vec![],
                });
            }
            101 => {
                // NOTICE_TYPE_ORDER_FILL_NTF → 成交推送 + 重查
                tracing::info!(acc_id, fills = ?fill_ids, "fill notification");
                let _ = push_tx.try_send(PushEvent::TradeReQuery {
                    acc_id,
                    notice_type: 6,
                    order_ids: vec![],
                    order_fill_ids: fill_ids.clone(),
                });
            }
            _ => {
                tracing::debug!(acc_id, notice_type, "unhandled trade notification");
            }
        }
    }

    /// 处理后端 CMD 5300 消息中心推送
    ///
    /// C++ 流程: MsgCenterNotify → 按 msg_category 分类处理
    /// category=1 (MSG_CATEGORY_PRICEWARN): 解析 PriceWarnMsg → 转换为 Qot_UpdatePriceReminder 推送
    fn handle_msg_center_push(
        body: &[u8],
        static_cache: &Arc<StaticDataCache>,
        push_tx: &tokio::sync::mpsc::Sender<PushEvent>,
    ) {
        use futu_backend::proto_internal::msgcenter;
        use futu_backend::proto_internal::msgdefine;

        let notify: msgcenter::MsgCenterNotify = match prost::Message::decode(body) {
            Ok(n) => n,
            Err(e) => {
                tracing::debug!(error = %e, "CMD5300 MsgCenterNotify decode failed");
                return;
            }
        };

        let category = notify.msg_category.unwrap_or(0);
        tracing::debug!(
            user_id = notify.user_id.unwrap_or(0),
            category,
            "CMD5300 message center push"
        );

        // 目前只处理到价提醒 (category=1)
        if category != 1 {
            tracing::debug!(category, "CMD5300: unhandled msg category");
            return;
        }

        // 解析消息体
        let msg_item = match &notify.notify_msg {
            Some(item) => item,
            None => {
                tracing::debug!("CMD5300: no notify_msg");
                return;
            }
        };

        let msg_body = match &msg_item.body {
            Some(b) => b,
            None => {
                tracing::debug!("CMD5300: no msg body");
                return;
            }
        };

        let price_warn: msgdefine::PriceWarnMsg = match prost::Message::decode(msg_body.as_ref()) {
            Ok(m) => m,
            Err(e) => {
                tracing::debug!(error = %e, "CMD5300: PriceWarnMsg decode failed");
                return;
            }
        };

        let stock_id = price_warn.stock_id.unwrap_or(0);

        // 从 static_cache 中查找 stock_id 对应的 security
        let sec_info = static_cache.get_security_info_by_stock_id(stock_id);
        let (market, code) = match &sec_info {
            Some(info) => (info.market, info.code.clone()),
            None => {
                tracing::debug!(stock_id, "CMD5300: stock_id not found in cache");
                return;
            }
        };

        let stock_name = sec_info
            .as_ref()
            .map(|i| i.name.clone())
            .unwrap_or_default();

        // 价格: 后端 price * 1e3 → 真实价格
        let price = price_warn.price.unwrap_or(0) as f64 / 1000.0;
        // 涨跌幅: 后端 * 1e3 → 百分比
        let change_rate = price_warn.price_change_ratio.unwrap_or(0) as f64 / 1000.0;
        let market_status = price_warn.market_status.unwrap_or(0);

        // 消息内容（根据语言选择，默认简体）
        let content = price_warn.msg_content.clone().unwrap_or_default();
        let note = price_warn.note.clone().unwrap_or_default();

        // 解析 warn_detail (含市场类型相关的单位换算)
        let (key, reminder_type, set_value, cur_value) =
            if let Some(detail) = &price_warn.warn_detail {
                let warn_type = detail.r#type.unwrap_or(0);
                let api_type = Self::warn_type_to_api_type(warn_type);
                let sv_raw = detail.set_value.unwrap_or(0);
                let cv_raw = detail.present_value.unwrap_or(0);
                // 对齐 C++ GetPriceReminderCltValue: 按市场类型和提醒类型做单位换算
                let sv = Self::price_reminder_clt_value(sv_raw, warn_type, sec_info.as_ref());
                let cv = Self::price_reminder_clt_value(cv_raw, warn_type, sec_info.as_ref());
                (detail.key.unwrap_or(0) as i64, api_type, sv, cv)
            } else {
                (0i64, 0i32, 0.0f64, 0.0f64)
            };

        tracing::info!(
            stock_id,
            market,
            code = %code,
            price,
            change_rate,
            reminder_type,
            "price reminder push → 3019"
        );

        // 构建 Qot_UpdatePriceReminder::Response
        let s2c = futu_proto::qot_update_price_reminder::S2c {
            security: futu_proto::qot_common::Security { market, code },
            name: Some(stock_name),
            price,
            change_rate,
            market_status,
            content,
            note,
            key: Some(key),
            r#type: Some(reminder_type),
            set_value: Some(set_value),
            cur_value: Some(cur_value),
        };

        let resp = futu_proto::qot_update_price_reminder::Response {
            ret_type: 0, // RetType_Succeed
            ret_msg: None,
            err_code: None,
            s2c: Some(s2c),
        };

        let push_body = prost::Message::encode_to_vec(&resp);

        let _ = push_tx.try_send(PushEvent::BroadcastPush {
            proto_id: futu_core::proto_id::QOT_UPDATE_PRICE_REMINDER,
            body: push_body,
        });
    }

    /// 后端 WarnType → API PriceReminderType 映射
    fn warn_type_to_api_type(warn_type: u32) -> i32 {
        match warn_type {
            4 => 1,   // UP_PRICE -> PriceUp
            8 => 2,   // DOWN_PRICE -> PriceDown
            1 => 3,   // DAY_UP_RATIO -> ChangeRateUp
            2 => 4,   // DAY_DOWN_RATIO -> ChangeRateDown
            9 => 5,   // FIVE_MIN_UP_RATIO
            10 => 6,  // FIVE_MIN_DOWN_RATIO
            11 => 7,  // UP_VOLUME
            12 => 8,  // UP_TURNOVER
            13 => 9,  // UP_TOR
            14 => 10, // UP_BUY_PRICE_1
            15 => 11, // DOWN_SELL_PRICE_1
            16 => 12, // UP_BUY_VOLUME_1
            17 => 13, // DOWN_SELL_VOLUME_1
            19 => 14, // THREE_MIN_UP_RATIO
            20 => 15, // THREE_MIN_DOWN_RATIO
            _ => 0,
        }
    }

    /// 后端提醒值 → 客户端显示值的单位换算
    ///
    /// 对齐 C++ GetPriceReminderCltValue():
    /// - 基础: nValue / 1000.0
    /// - VolumeUp (11): 非期权/期货 × 10000; A股额外 × 100
    /// - TurnOverUp (12): × 10000
    /// - BuyVolumeUp (16) / SellVolumeUp (17): 非期权/期货 × 1000; A股额外 × 100
    fn price_reminder_clt_value(
        raw_value: i64,
        warn_type: u32,
        sec_info: Option<&futu_cache::static_data::CachedSecurityInfo>,
    ) -> f64 {
        let mut value = raw_value as f64 / 1000.0;

        // 判断市场类型
        let (is_option, is_future, is_cn) = if let Some(info) = sec_info {
            let is_opt = info.sec_type == 6 || info.sec_type == 8; // OPTION, INDEX_OPTION
            let is_fut = info.sec_type == 5; // FUTURE
            let is_cn_market = info.market == 21 || info.market == 22; // CNSH, CNSZ
            (is_opt, is_fut, is_cn_market)
        } else {
            (false, false, false)
        };

        match warn_type {
            11 => {
                // VolumeUp (成交量): 非期权/期货 ×10000; A股额外 ×100
                if !is_option && !is_future {
                    value *= 10000.0;
                }
                if is_cn {
                    value *= 100.0;
                }
            }
            12 => {
                // TurnOverUp (成交额): ×10000
                value *= 10000.0;
            }
            16 | 17 => {
                // BuyVolumeUp / SellVolumeUp: 非期权/期货 ×1000; A股额外 ×100
                if !is_option && !is_future {
                    value *= 1000.0;
                }
                if is_cn {
                    value *= 100.0;
                }
            }
            _ => {
                // 价格类型的提醒: 不需要额外换算，/ 1000.0 已足够
            }
        }

        value
    }

    /// 请求行情权限 (CMD 6024)
    /// 对齐 C++ RequestAPIQuota / RequestMkrQotRightWithPushedQotRight
    async fn request_qot_right(
        backend: &futu_backend::conn::BackendConn,
        qot_right_cache: &QotRightCache,
    ) {
        use futu_backend::proto_internal::qta_auth;

        tracing::info!("CMD6024: requesting qot right...");

        // C++ 发送空请求（不设置任何字段）
        let req = qta_auth::QtaAuth6024Req {
            req_hk_auth: None,
            req_us_auth: None,
            req_cn_auth: None,
        };

        let body = prost::Message::encode_to_vec(&req);

        // C++ CMD ID = 6032 (NN_ProtoCmd_Qot_Request_Right)
        // proto 文件名虽然叫 FTCmdQtaAuth6024，但实际 CMD ID 是 6032
        match backend.request(6032, body).await {
            Ok(frame) => {
                let body_bytes = frame.body.as_ref();

                // CMD 6032 响应是裸 protobuf (已验证不需要 SRPC 解包)
                let rsp: qta_auth::QtaAuth6024Rsp = crate::handlers::decode_srpc_or_direct(
                    body_bytes,
                    |r: &qta_auth::QtaAuth6024Rsp| {
                        // 只要有任何权限字段就认为有效
                        r.hk_qut_got_auth.is_some()
                            || r.us_qut_got_auth.is_some()
                            || r.open_api_auth.is_some()
                    },
                );

                // 提取 OpenAPI 额度
                let (sub_limit, kl_limit) = if let Some(api_auth) = &rsp.open_api_auth {
                    (api_auth.sub_limit, api_auth.history_k_line_limit)
                } else {
                    (None, None)
                };

                // 提取 US 期货细分权限
                let us_future_detail = rsp.us_future_auth.as_ref().map(|f| {
                    (
                        f.open_api_cme_auth.unwrap_or(0),
                        f.open_api_cbot_auth.unwrap_or(0),
                        f.open_api_nymex_auth.unwrap_or(0),
                        f.open_api_comex_auth.unwrap_or(0),
                    )
                });

                qot_right_cache.update_from_backend(
                    rsp.hk_qut_got_auth,
                    rsp.us_qut_got_auth,
                    rsp.cn_qut_got_auth,
                    rsp.shanghai_qut_auth,
                    rsp.shenzhen_qut_auth,
                    rsp.hk_option_auth,
                    rsp.hk_future_auth,
                    rsp.us_option_flag,
                    rsp.us_future_cme_cboe_auth,
                    us_future_detail,
                    rsp.sgx_future_auth,
                    rsp.jp_future_auth,
                    sub_limit,
                    kl_limit,
                );

                tracing::info!(
                    hk_got = rsp.hk_qut_got_auth.unwrap_or(0),
                    us_got = rsp.us_qut_got_auth.unwrap_or(0),
                    cn_got = rsp.cn_qut_got_auth.unwrap_or(0),
                    sub_limit = sub_limit.unwrap_or(0),
                    kl_limit = kl_limit.unwrap_or(0),
                    "CMD6024 qot right updated"
                );
            }
            Err(e) => {
                tracing::warn!(error = %e, "CMD6024 request failed (using defaults)");
            }
        }
    }

    /// 解析 SBIT_DEAL_STATISTICS(5) → 更新 BasicQot 缓存的成交量/额/OHLC
    fn parse_deal_statistics(
        data: &[u8],
        sec_key: &str,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) {
        use futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data;

        let stats: ft_cmd_stock_quote_coverage_data::DealStatistics =
            match prost::Message::decode(data) {
                Ok(s) => s,
                Err(_) => return,
            };

        // 价格放大 10^9, turnover 放大 10^3, turnover_rate 放大 10^5, amplitude 放大 10^5
        let open = stats.price_open.unwrap_or(0) as f64 / 1_000_000_000.0;
        let high = stats.price_highest.unwrap_or(0) as f64 / 1_000_000_000.0;
        let low = stats.price_lowest.unwrap_or(0) as f64 / 1_000_000_000.0;
        let volume = stats.volume.unwrap_or(0);
        let turnover = stats.turnover.unwrap_or(0) as f64 / 1_000.0;
        let turnover_rate = stats.ratio_turnover.unwrap_or(0) as f64 / 100_000.0;
        let amplitude = stats.amplitude_price.unwrap_or(0) as f64 / 100_000.0;

        // 合并更新：读取已有缓存，只更新非零字段
        if let Some(mut existing) = qot_cache.basic_qot.get_mut(sec_key) {
            if open > 0.0 {
                existing.open_price = open;
            }
            if high > 0.0 {
                existing.high_price = high;
            }
            if low > 0.0 {
                existing.low_price = low;
            }
            if volume > 0 {
                existing.volume = volume;
            }
            if turnover > 0.0 {
                existing.turnover = turnover;
            }
            if turnover_rate > 0.0 {
                existing.turnover_rate = turnover_rate;
            }
            if amplitude > 0.0 {
                existing.amplitude = amplitude;
            }
        } else if volume > 0 || open > 0.0 {
            // 还没有 price 数据，先创建
            qot_cache.update_basic_qot(
                sec_key,
                futu_cache::qot_cache::CachedBasicQot {
                    cur_price: 0.0,
                    last_close_price: 0.0,
                    open_price: open,
                    high_price: high,
                    low_price: low,
                    volume,
                    turnover,
                    turnover_rate,
                    amplitude,
                    is_suspended: false,
                    update_time: String::new(),
                    update_timestamp: 0.0,
                },
            );
        }
    }

    /// 解析 SBIT_STOCK_STATE(1) → 更新 BasicQot 停牌状态
    fn parse_stock_state(data: &[u8], sec_key: &str, qot_cache: &futu_cache::qot_cache::QotCache) {
        use futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data;

        let state: ft_cmd_stock_quote_coverage_data::StockState = match prost::Message::decode(data)
        {
            Ok(s) => s,
            Err(_) => return,
        };

        let state_type = state.state_type.unwrap_or(0);
        let is_suspended = state_type == 8; // STOCK_STATE_SUSPENDED = 8

        if let Some(mut existing) = qot_cache.basic_qot.get_mut(sec_key) {
            existing.is_suspended = is_suspended;
        }
    }

    /// 从缓存构建 Qot_UpdateBasicQot push event（供 bit 1/5 处理后触发推送）
    /// C++ 中每个 bit 处理完都会调用 UpdateRealTimeData 推送累积状态
    fn build_basic_qot_push_from_cache(
        sec_key: &str,
        sec_info: &futu_cache::static_data::CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) -> Option<PushEvent> {
        let cached = qot_cache.get_basic_qot(sec_key)?;

        let security = futu_proto::qot_common::Security {
            market: sec_info.market,
            code: sec_info.code.clone(),
        };
        let basic_qot = futu_proto::qot_common::BasicQot {
            security,
            name: Some(sec_info.name.clone()),
            is_suspended: cached.is_suspended,
            list_time: sec_info.list_time.clone(),
            price_spread: 0.0,
            update_time: cached.update_time.clone(),
            high_price: cached.high_price,
            open_price: cached.open_price,
            low_price: cached.low_price,
            cur_price: cached.cur_price,
            last_close_price: cached.last_close_price,
            volume: cached.volume,
            turnover: cached.turnover,
            turnover_rate: cached.turnover_rate,
            amplitude: cached.amplitude,
            dark_status: None,
            option_ex_data: None,
            list_timestamp: None,
            update_timestamp: Some(cached.update_timestamp),
            pre_market: None,
            after_market: None,
            overnight: None,
            future_ex_data: None,
            sec_status: None,
            warrant_ex_data: None,
        };

        let resp = futu_proto::qot_update_basic_qot::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_update_basic_qot::S2c {
                basic_qot_list: vec![basic_qot],
            }),
        };

        Some(PushEvent::QuotePush {
            sec_key: sec_key.to_string(),
            sub_type: 1, // SubType_Basic
            proto_id: futu_core::proto_id::QOT_UPDATE_BASIC_QOT,
            body: prost::Message::encode_to_vec(&resp),
        })
    }

    /// 解析 SBIT_ORDER_BOOK(3) → 更新缓存 + 构建 Qot_UpdateOrderBook push
    fn parse_order_book_to_push(
        data: &[u8],
        sec_key: &str,
        sec_info: &futu_cache::static_data::CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) -> Option<PushEvent> {
        use futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data;

        let ob: ft_cmd_stock_quote_coverage_data::OrderBook = prost::Message::decode(data).ok()?;

        let convert_levels =
            |items: &[ft_cmd_stock_quote_coverage_data::OrderBookItem]|
             -> Vec<futu_cache::qot_cache::CachedOrderBookLevel> {
                items
                    .iter()
                    .map(|item| futu_cache::qot_cache::CachedOrderBookLevel {
                        price: item.price.unwrap_or(0) as f64 / 1_000_000_000.0,
                        volume: item.volume.unwrap_or(0),
                        order_count: item.order_count.unwrap_or(0) as i32,
                    })
                    .collect()
            };

        let bid_list = convert_levels(&ob.bid);
        let ask_list = convert_levels(&ob.ask);

        let svr_recv_bid_ts = ob
            .server_recv_bid_from_exchange_time_ms
            .map(|t| t as f64 / 1000.0);
        let svr_recv_ask_ts = ob
            .server_recv_ask_from_exchange_time_ms
            .map(|t| t as f64 / 1000.0);

        // 更新缓存
        qot_cache.order_books.insert(
            sec_key.to_string(),
            futu_cache::qot_cache::CachedOrderBook {
                bid_list: bid_list.clone(),
                ask_list: ask_list.clone(),
                svr_recv_time_bid: None,
                svr_recv_time_bid_timestamp: svr_recv_bid_ts,
                svr_recv_time_ask: None,
                svr_recv_time_ask_timestamp: svr_recv_ask_ts,
            },
        );

        // 构建 FTAPI push
        let security = futu_proto::qot_common::Security {
            market: sec_info.market,
            code: sec_info.code.clone(),
        };
        let to_proto = |levels: &[futu_cache::qot_cache::CachedOrderBookLevel]|
         -> Vec<futu_proto::qot_common::OrderBook> {
            levels.iter().map(|l| futu_proto::qot_common::OrderBook {
                price: l.price,
                volume: l.volume,
                oreder_count: l.order_count,
                detail_list: vec![],
            }).collect()
        };

        let resp = futu_proto::qot_update_order_book::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_update_order_book::S2c {
                security,
                name: Some(sec_info.name.clone()),
                order_book_ask_list: to_proto(&ask_list),
                order_book_bid_list: to_proto(&bid_list),
                svr_recv_time_bid: None,
                svr_recv_time_bid_timestamp: svr_recv_bid_ts,
                svr_recv_time_ask: None,
                svr_recv_time_ask_timestamp: svr_recv_ask_ts,
            }),
        };

        Some(PushEvent::QuotePush {
            sec_key: sec_key.to_string(),
            sub_type: 2, // SubType_OrderBook
            proto_id: futu_core::proto_id::QOT_UPDATE_ORDER_BOOK,
            body: prost::Message::encode_to_vec(&resp),
        })
    }

    /// 解析 SBIT_HK_BROKER_QUEUE(9) → 更新缓存 + 构建 Qot_UpdateBroker push
    fn parse_broker_queue_to_push(
        data: &[u8],
        sec_key: &str,
        sec_info: &futu_cache::static_data::CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) -> Option<PushEvent> {
        use futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data;

        let bq: ft_cmd_stock_quote_coverage_data::HkBrokerQueue =
            prost::Message::decode(data).ok()?;

        // 转换内部 HKBrokerQueueItem → CachedBrokerItem
        // 每个 rank 中的每个 broker_id 各生成一条 CachedBrokerItem
        let convert_side =
            |items: &[ft_cmd_stock_quote_coverage_data::HkBrokerQueueItem]|
             -> Vec<futu_cache::qot_cache::CachedBrokerItem> {
                let mut result = Vec::new();
                for item in items {
                    let pos = item.rank_id.unwrap_or(0);
                    for &bid in &item.broker_id {
                        result.push(futu_cache::qot_cache::CachedBrokerItem {
                            id: bid as i64,
                            name: format!("Broker#{bid}"),
                            pos,
                        });
                    }
                }
                result
            };

        let bid_list = convert_side(&bq.bid);
        let ask_list = convert_side(&bq.ask);

        // 更新缓存
        qot_cache.update_broker(
            sec_key,
            futu_cache::qot_cache::CachedBroker {
                bid_list: bid_list.clone(),
                ask_list: ask_list.clone(),
            },
        );

        // 构建 FTAPI push
        let security = futu_proto::qot_common::Security {
            market: sec_info.market,
            code: sec_info.code.clone(),
        };
        let to_proto =
            |items: &[futu_cache::qot_cache::CachedBrokerItem]|
             -> Vec<futu_proto::qot_common::Broker> {
                items
                    .iter()
                    .map(|b| futu_proto::qot_common::Broker {
                        id: b.id,
                        name: b.name.clone(),
                        pos: b.pos,
                        order_id: None,
                        volume: None,
                    })
                    .collect()
            };

        let resp = futu_proto::qot_update_broker::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_update_broker::S2c {
                security,
                name: Some(sec_info.name.clone()),
                broker_ask_list: to_proto(&ask_list),
                broker_bid_list: to_proto(&bid_list),
            }),
        };

        Some(PushEvent::QuotePush {
            sec_key: sec_key.to_string(),
            sub_type: 14, // SubType_Broker
            proto_id: futu_core::proto_id::QOT_UPDATE_BROKER,
            body: prost::Message::encode_to_vec(&resp),
        })
    }

    /// 解析 SBIT_TICK(35) → 更新缓存 + 构建 Qot_UpdateTicker push
    fn parse_ticker_to_push(
        data: &[u8],
        sec_key: &str,
        sec_info: &futu_cache::static_data::CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) -> Option<PushEvent> {
        // Tick 在 FTCmdStockQuoteAccumulateData.proto 中定义
        use futu_backend::proto_internal::ft_cmd_stock_quote_accumulate_data;

        let tick: ft_cmd_stock_quote_accumulate_data::Tick = prost::Message::decode(data).ok()?;

        let mut ticker_list = Vec::new();
        let mut cached_tickers = Vec::new();

        for item in &tick.items {
            let price = item.price.unwrap_or(0) as f64 / 1_000_000_000.0;
            let volume = item.volume.unwrap_or(0) as i64;
            let timestamp_ms = item.exchange_data_time_ms.unwrap_or(0);
            let timestamp = timestamp_ms as f64 / 1000.0;
            let sequence = item.tick_key.unwrap_or(0) as i64;
            // C++ TickItemType: SELL=1→dir=1(Bid), BUY=2→dir=2(Ask), NEUTRAL=3→dir=3
            let dir = item.r#type.unwrap_or(0);
            let turnover = price * volume as f64;

            cached_tickers.push(futu_cache::qot_cache::CachedTicker {
                time: String::new(),
                sequence,
                dir,
                price,
                volume,
                turnover,
                recv_time: None,
                ticker_type: None,
                push_data_type: Some(1), // push
                timestamp: Some(timestamp),
            });

            ticker_list.push(futu_proto::qot_common::Ticker {
                time: String::new(),
                sequence,
                dir,
                price,
                volume,
                turnover,
                recv_time: None,
                r#type: None,
                type_sign: None,
                push_data_type: Some(1),
                timestamp: Some(timestamp),
            });
        }

        if ticker_list.is_empty() {
            return None;
        }

        // 缓存 — 累加 (C++ 是 AddTickerList，不是替换)
        let max_cached = 1000;
        let mut entry = qot_cache.tickers.entry(sec_key.to_string()).or_default();
        entry.extend(cached_tickers);
        if entry.len() > max_cached {
            let drain_count = entry.len() - max_cached;
            entry.drain(0..drain_count);
        }

        // 构建 FTAPI push
        let security = futu_proto::qot_common::Security {
            market: sec_info.market,
            code: sec_info.code.clone(),
        };
        let resp = futu_proto::qot_update_ticker::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_update_ticker::S2c {
                security,
                name: Some(sec_info.name.clone()),
                ticker_list,
            }),
        };

        Some(PushEvent::QuotePush {
            sec_key: sec_key.to_string(),
            sub_type: 4, // SubType_Ticker
            proto_id: futu_core::proto_id::QOT_UPDATE_TICKER,
            body: prost::Message::encode_to_vec(&resp),
        })
    }

    /// SBIT 值 → FTAPI KLType 枚举
    fn sbit_to_kl_type(bit: u32) -> i32 {
        match bit {
            21 => 1,  // KLType_1Min
            22 => 10, // KLType_3Min
            23 => 6,  // KLType_5Min
            24 => 7,  // KLType_15Min
            25 => 8,  // KLType_30Min
            26 => 9,  // KLType_60Min
            27 => 2,  // KLType_Day
            28 => 3,  // KLType_Week
            29 => 4,  // KLType_Month
            30 => 11, // KLType_Quarter
            31 => 5,  // KLType_Year
            _ => 0,
        }
    }

    /// 解析 SBIT_KLINE(21-31) → 构建 Qot_UpdateKL push
    fn parse_kline_to_push(
        data: &[u8],
        bit: u32,
        sec_key: &str,
        sec_info: &futu_cache::static_data::CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) -> Option<PushEvent> {
        use futu_backend::proto_internal::ft_cmd_stock_quote_accumulate_data;

        let kline: ft_cmd_stock_quote_accumulate_data::Kline = prost::Message::decode(data).ok()?;

        let mut kl_list = Vec::new();
        let mut cached_klines = Vec::new();
        let mut rehab_type = 0i32;

        for point in &kline.point_list {
            rehab_type = point.exright_type.unwrap_or(0) as i32;
            if let Some(ref item) = point.item {
                let close = item.close_price.unwrap_or(0) as f64 / 1_000_000_000.0;
                let open = item.open_price.unwrap_or(0) as f64 / 1_000_000_000.0;
                let high = item.highest_price.unwrap_or(0) as f64 / 1_000_000_000.0;
                let low = item.lowest_price.unwrap_or(0) as f64 / 1_000_000_000.0;
                let last_close = item.last_close_price.unwrap_or(0) as f64 / 1_000_000_000.0;
                let volume = item.volume.unwrap_or(0) as i64;
                let turnover = item.turnover.unwrap_or(0) as f64 / 1_000.0;
                let turnover_rate = item.turnover_rate.unwrap_or(0) as f64 / 100_000.0;
                let pe = item.pe.unwrap_or(0) as f64 / 1_000.0;
                let timestamp = item.time.unwrap_or(0) as f64;

                let is_blank = close <= 0.0 && open <= 0.0;

                kl_list.push(futu_proto::qot_common::KLine {
                    time: String::new(),
                    is_blank,
                    high_price: Some(high),
                    open_price: Some(open),
                    low_price: Some(low),
                    close_price: Some(close),
                    last_close_price: Some(last_close),
                    volume: Some(volume),
                    turnover: Some(turnover),
                    turnover_rate: Some(turnover_rate),
                    pe: Some(pe),
                    change_rate: if last_close > 0.0 {
                        Some((close - last_close) / last_close * 100.0)
                    } else {
                        None
                    },
                    timestamp: Some(timestamp),
                });

                if !is_blank {
                    cached_klines.push(futu_cache::qot_cache::CachedKLine {
                        time: String::new(),
                        open_price: open,
                        high_price: high,
                        low_price: low,
                        close_price: close,
                        volume,
                        turnover,
                    });
                }
            }
        }

        if kl_list.is_empty() {
            return None;
        }

        // 缓存 K 线
        let kl_type = Self::sbit_to_kl_type(bit);
        let cache_key = format!("{sec_key}:{kl_type}");
        let mut entry = qot_cache.klines.entry(cache_key).or_default();
        entry.extend(cached_klines);
        // 限制缓存大小
        if entry.len() > 2000 {
            let drain = entry.len() - 2000;
            entry.drain(0..drain);
        }

        let security = futu_proto::qot_common::Security {
            market: sec_info.market,
            code: sec_info.code.clone(),
        };
        let resp = futu_proto::qot_update_kl::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_update_kl::S2c {
                rehab_type,
                kl_type,
                security,
                name: Some(sec_info.name.clone()),
                kl_list,
            }),
        };

        // SubType 映射: KLType → SubType
        let sub_type = match kl_type {
            1 => 11,  // KL_1Min → SubType_KL_1Min
            2 => 6,   // KL_Day → SubType_KL_Day
            3 => 12,  // KL_Week
            4 => 13,  // KL_Month
            5 => 16,  // KL_Year
            6 => 7,   // KL_5Min
            7 => 8,   // KL_15Min
            8 => 9,   // KL_30Min
            9 => 10,  // KL_60Min
            10 => 17, // KL_3Min
            11 => 15, // KL_Quarter
            _ => 6,
        };

        Some(PushEvent::QuotePush {
            sec_key: sec_key.to_string(),
            sub_type,
            proto_id: futu_core::proto_id::QOT_UPDATE_KL,
            body: prost::Message::encode_to_vec(&resp),
        })
    }

    /// 解析 SBIT_TIME_SHARING(20) → 构建 Qot_UpdateRT push
    fn parse_rt_to_push(
        data: &[u8],
        sec_key: &str,
        sec_info: &futu_cache::static_data::CachedSecurityInfo,
        qot_cache: &futu_cache::qot_cache::QotCache,
    ) -> Option<PushEvent> {
        use futu_backend::proto_internal::ft_cmd_stock_quote_accumulate_data;

        let ts: ft_cmd_stock_quote_accumulate_data::TimeSharingPlans =
            prost::Message::decode(data).ok()?;

        let point = ts.point.as_ref()?;
        let price = point.close_price.unwrap_or(0) as f64 / 1_000_000_000.0;
        let volume = point.volume.unwrap_or(0) as i64;
        let turnover = point.turnover.unwrap_or(0) as f64 / 1_000.0;
        let timestamp = point.time.unwrap_or(0) as f64;

        if price <= 0.0 {
            return None;
        }

        // 缓存分时数据 — 累加模式
        let mut entry = qot_cache.rt_data.entry(sec_key.to_string()).or_default();
        entry.push(futu_cache::qot_cache::CachedTimeShare {
            time: String::new(),
            minute: 0,
            price,
            last_close_price: 0.0,
            avg_price: 0.0,
            volume,
            turnover,
            timestamp,
        });
        if entry.len() > 500 {
            let drain = entry.len() - 500;
            entry.drain(0..drain);
        }

        let security = futu_proto::qot_common::Security {
            market: sec_info.market,
            code: sec_info.code.clone(),
        };

        let rt = futu_proto::qot_common::TimeShare {
            time: String::new(),
            minute: 0,
            is_blank: false,
            price: Some(price),
            last_close_price: None,
            avg_price: None,
            volume: Some(volume),
            turnover: Some(turnover),
            timestamp: Some(timestamp),
        };

        let resp = futu_proto::qot_update_rt::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_update_rt::S2c {
                security,
                name: Some(sec_info.name.clone()),
                rt_list: vec![rt],
            }),
        };

        Some(PushEvent::QuotePush {
            sec_key: sec_key.to_string(),
            sub_type: 5, // SubType_RT
            proto_id: futu_core::proto_id::QOT_UPDATE_RT,
            body: prost::Message::encode_to_vec(&resp),
        })
    }

    /// C++ IsValidTrdMarket for Real env
    fn is_valid_real_trd_market(market: i32) -> bool {
        matches!(market, 1 | 2 | 4 | 5 | 6 | 8 | 15 | 111 | 112)
        // HK, US, HKCC, Futures, SG, AU, JP, MY, CA
    }

    /// broker_id → SecurityFirm 枚举 (C++ SecurityFirm_NNToAPI)
    fn broker_to_security_firm(broker_id: u32) -> i32 {
        match broker_id {
            1001 => 1, // NN_BrokerID_Futu → FutuSecurities
            1007 => 2, // NN_BrokerID_FutuUS (MooMoo) → FutuInc
            1008 => 3, // NN_BrokerID_FutuSG → FutuSG
            1009 => 4, // NN_BrokerID_FutuAU → FutuAU
            1019 => 5, // NN_BrokerID_FutuCA → FutuCA
            1017 => 6, // NN_BrokerID_FutuMY → FutuMY
            1012 => 7, // NN_BrokerID_FutuJP → FutuJP
            _ => 0,    // Unknown
        }
    }

    /// 从 Account proto 获取 acc_type (C++ GetAccTypeFromAccItem)
    fn get_acc_type(acc: &futu_backend::proto_internal::ft_usr_trd_acc::Account) -> Option<i32> {
        // Priority 1: IRA type (CA)
        if let Some(ira) = acc.ira_type {
            match ira {
                1 => return Some(3), // CA_TFSA → TrdAccType_TFSA
                2 => return Some(4), // CA_RRSP → TrdAccType_RRSP
                3 => return Some(5), // CA_SRRSP → TrdAccType_SRRSP
                _ => {}
            }
        }
        // Priority 2: Kouza type (JP)
        if let Some(kouza) = acc.kouza_type {
            match kouza {
                1 => return Some(1), // Cash → TrdAccType_Cash
                2 => return Some(2), // Margin → TrdAccType_Margin
                3 => return Some(6), // Derivative → TrdAccType_Derivatives
                _ => {}
            }
        }
        // Priority 3: General type (C++ does type() + 1 unconditionally, default CASH=0→1)
        Some(acc.r#type.unwrap_or(0) + 1) // CASH=0→1, MARGIN=1→2
    }

    /// 从 Account proto 获取 acc_role (C++ GetAccRoleFromAccItem)
    fn get_acc_role(acc: &futu_backend::proto_internal::ft_usr_trd_acc::Account) -> Option<i32> {
        // IPO check (MY broker)
        if acc.is_esop.unwrap_or(false) {
            return Some(3); // TrdAccRole_IPO
        }
        // Role mapping: ORDINARY=0→Normal(1), MASTER=1→Master(2)
        // C++ NN_AccRole: Ordinary=0, Master=1
        // FTAPI TrdAccRole: Unknown=0, Normal=1, Master=2, IPO=3
        // 从 UniversalAccount.role 或 Account 层面获取
        Some(1) // Default: Normal
    }

    /// 从 Account proto 提取 CachedTrdAcc
    fn account_to_cached(
        acc: &futu_backend::proto_internal::ft_usr_trd_acc::Account,
        uni_card_num: Option<&str>,
        uni_role: Option<u32>,
    ) -> Option<CachedTrdAcc> {
        let acc_id = acc.id?;
        let broker_id = acc.broker.as_ref().map(|b| b.id).unwrap_or(0);

        let acc_type = Self::get_acc_type(acc);
        let card_num = acc.card_number.as_ref().filter(|s| !s.is_empty()).cloned();
        let security_firm = Self::broker_to_security_firm(broker_id);
        // acc_status: state OPENED=1 → Active(0), 其他 → Disabled(1)
        let acc_status = acc.state.map(|s| if s == 1 { 0 } else { 1 });

        // acc_role: 从 UniversalAccount.role 或 Account 自身推断
        let acc_role = match uni_role {
            Some(1) => Some(2), // MASTER
            _ => Self::get_acc_role(acc),
        };

        // enable_market 数组 → trd_market_auth_list
        // C++ SubAccToAccItem: ConvS2C + IsValidTrdMarket + Fund 展开
        let mut trd_market_auth_list: Vec<i32> = Vec::new();
        let add_unique = |list: &mut Vec<i32>, val: i32| {
            if !list.contains(&val) {
                list.push(val);
            }
        };
        for &m in &acc.enable_market {
            // C++ NN_TrdMarket_ConvS2C for real env
            // 后端 enable_market 值 → 客户端 TrdMarket 值
            let market = match m as i32 {
                11 => 111, // 后端 11 → NN_TrdMarket_MY(111)
                12 => 112, // 后端 12 → NN_TrdMarket_CA(112)
                13 => 113, // 后端 13 → NN_TrdMarket_HK_Fund(113)
                14 => 114, // 后端 14 → NN_TrdMarket_Fund(114)
                15 => 15,  // 后端 15 → NN_TrdMarket_JP(15)
                23 => 123, // 后端 23 → NN_TrdMarket_US_Fund(123)
                24 => 124, // 后端 24 → NN_TrdMarket_SG_Fund(124)
                other => other,
            };

            if market == 114 {
                // NN_TrdMarket_Fund → 按券商展开
                match broker_id {
                    1001 => {
                        add_unique(&mut trd_market_auth_list, 113);
                        add_unique(&mut trd_market_auth_list, 123);
                    }
                    1008 => {
                        add_unique(&mut trd_market_auth_list, 124);
                        add_unique(&mut trd_market_auth_list, 123);
                    }
                    1017 => {
                        add_unique(&mut trd_market_auth_list, 125);
                        add_unique(&mut trd_market_auth_list, 123);
                    }
                    1012 => {
                        add_unique(&mut trd_market_auth_list, 126);
                        add_unique(&mut trd_market_auth_list, 123);
                    }
                    _ => {
                        add_unique(&mut trd_market_auth_list, 113);
                        add_unique(&mut trd_market_auth_list, 123);
                    }
                }
            } else if Self::is_valid_real_trd_market(market)
                || market == 113
                || market == 123
                || market == 124
                || market == 125
                || market == 126
            {
                // C++ IsValidTrdMarket for Real: HK(1), US(2), HKCC(4), Futures(5), SG(6), AU(8), JP(15), MY(111), CA(112)
                // 或 Fund 子市场
                add_unique(&mut trd_market_auth_list, market);
            }
        }

        // JP sub-account types (SubAccount.type)
        let jp_acc_type: Vec<i32> = acc
            .sub_accounts
            .iter()
            .filter_map(|sub| sub.r#type)
            .map(|t| t as i32)
            .collect();

        // C++ 排序 key: (BrokerID << 48) | (TrdMkt << 32) | IntraAccID
        let backend_market = acc.market.unwrap_or(0) as u64;
        let intra_acc_id = acc.acc_id.unwrap_or(0) as u64;
        let sort_key = ((broker_id as u64) << 48) | (backend_market << 32) | intra_acc_id;

        Some(CachedTrdAcc {
            acc_id,
            trd_env: 1, // TrdEnv_Real
            trd_market_auth_list,
            acc_type,
            card_num,
            security_firm: Some(security_firm),
            sim_acc_type: None,
            uni_card_num: uni_card_num.map(|s| s.to_string()),
            acc_status,
            acc_role,
            jp_acc_type,
            // 审计补全字段
            owner_uid: acc.owner_uid,
            opr_uid: acc.opr_uid,
            mixed_state: acc.mixed_state,
            ira_type: acc.ira_type.map(|t| t as i32),
            grant_state: acc.grant_state,
            kouza_type: acc.kouza_type.map(|k| k as i32),
            trd_market: acc.market.map(|m| m as i32),
            association_acc_id: None, // 后续从 fund_acc_list 填充
            acc_flag: None,           // 后续从 universal account 标记
            order_index: 0,           // set_accounts 会重新赋值
            sort_key,
        })
    }

    /// 从后端拉取真实账户列表 (CMD 2282 + CMD 2298)
    async fn fetch_real_accounts(&self, backend: &BackendConn) -> Result<Vec<CachedTrdAcc>> {
        use futu_backend::proto_internal::ft_cmd2282;
        use prost::Message;

        let mut all_accounts = Vec::new();

        // CMD 2282: 自有账户
        let req = ft_cmd2282::Request {
            page: Some(0),
            need_close: Some(true),
            roa_cid: None,
        };
        // CMD 2282 用裸 protobuf (SendTCPProto_ProtoBuf_WithAccID)
        let resp = backend.request(2282, req.encode_to_vec()).await?;
        let parsed: ft_cmd2282::Response = Message::decode(resp.body.as_ref())
            .map_err(|e| FutuError::Codec(format!("CMD2282 decode failed: {e}")))?;

        if parsed.result.unwrap_or(-1) != 0 {
            let err = parsed.err_text.as_deref().unwrap_or("unknown");
            return Err(FutuError::Codec(format!("CMD2282 error: {err}")));
        }

        // 从 universal_account_list 提取 (综合账户: 证券+期货子账户)
        for uni in &parsed.universal_account_list {
            let uni_card = uni.card_number.as_deref();
            let uni_role = uni.role;
            // 证券子账户
            if let Some(sec_acc) = &uni.sec_account {
                if let Some(cached) = Self::account_to_cached(sec_acc, uni_card, uni_role) {
                    all_accounts.push(cached);
                }
            }
            // 期货子账户
            if let Some(fut_acc) = &uni.futures_account {
                if let Some(cached) = Self::account_to_cached(fut_acc, uni_card, uni_role) {
                    all_accounts.push(cached);
                }
            }
            // 加密货币子账户
            if let Some(crypto_acc) = &uni.crypto_account {
                if let Some(cached) = Self::account_to_cached(crypto_acc, uni_card, uni_role) {
                    all_accounts.push(cached);
                }
            }
            // 外汇子账户
            if let Some(forex_acc) = &uni.forex_account {
                if let Some(cached) = Self::account_to_cached(forex_acc, uni_card, uni_role) {
                    all_accounts.push(cached);
                }
            }
        }

        // 从 account_list 提取 (单市场独立账户)
        for acc in &parsed.account_list {
            if let Some(cached) = Self::account_to_cached(acc, None, None) {
                all_accounts.push(cached);
            }
        }

        // 处理 fund_acc_list: 建立基金账户 ↔ 证券账户的关联
        for fund in &parsed.fund_acc_list {
            let fund_id = fund.fund_acc_id.as_ref().and_then(|f| f.id);
            let assoc_id = fund.association_acc_id.as_ref().and_then(|a| a.id);
            if let (Some(fid), Some(aid)) = (fund_id, assoc_id) {
                // 在已有账户中找到 fund_acc_id 匹配的，设置关联
                for acc in all_accounts.iter_mut() {
                    if acc.acc_id == fid {
                        acc.association_acc_id = Some(aid);
                    }
                }
            }
        }

        // 标记综合账户子账户的 acc_flag
        // uni_card_num 非空的就是综合账户下的子账户
        for acc in all_accounts.iter_mut() {
            if acc.uni_card_num.is_some() {
                acc.acc_flag = Some(1);
            }
        }

        tracing::info!(count = all_accounts.len(), "CMD2282 accounts fetched");

        // CMD 2298: 授权账户 — C++ 只对 Futu(1001) 和 FutuSG(1008) 查询
        // 检查是否有 Futu/FutuSG 账户来决定是否查询 2298
        let has_futu_or_sg = all_accounts.iter().any(|a| {
            matches!(a.security_firm, Some(1) | Some(2) | Some(3)) // FutuSec/FutuInc/FutuSG
        });
        if has_futu_or_sg {
            self.fetch_granted_accounts(backend, &mut all_accounts)
                .await;
        }

        tracing::info!(
            count = all_accounts.len(),
            "real accounts fetched (2282+2298)"
        );
        Ok(all_accounts)
    }

    /// CMD 2298: 拉取授权账户
    async fn fetch_granted_accounts(
        &self,
        backend: &BackendConn,
        accounts: &mut Vec<CachedTrdAcc>,
    ) {
        use futu_backend::proto_internal::ft_cmd2298;
        use prost::Message;

        let req = ft_cmd2298::Request { page: Some(0) };
        let resp = match backend.request(2298, req.encode_to_vec()).await {
            Ok(r) => r,
            Err(e) => {
                tracing::debug!(error = %e, "CMD2298 query skipped");
                return;
            }
        };

        let parsed: ft_cmd2298::Response = match Message::decode(resp.body.as_ref()) {
            Ok(p) => p,
            Err(e) => {
                tracing::debug!(error = %e, "CMD2298 decode failed");
                return;
            }
        };

        if parsed.result.unwrap_or(-1) != 0 {
            return;
        }

        let mut count = 0;
        for uni in &parsed.authorized_account_list {
            let uni_card = uni.card_number.as_deref();
            let uni_role = uni.role;
            if let Some(sec_acc) = &uni.sec_account {
                if let Some(cached) = Self::account_to_cached(sec_acc, uni_card, uni_role) {
                    accounts.push(cached);
                    count += 1;
                }
            }
            if let Some(fut_acc) = &uni.futures_account {
                if let Some(cached) = Self::account_to_cached(fut_acc, uni_card, uni_role) {
                    accounts.push(cached);
                    count += 1;
                }
            }
        }

        if count > 0 {
            tracing::info!(count, "CMD2298 granted accounts added");
        }
    }

    /// 从后端拉取模拟账户列表 (Proto 14800)
    async fn fetch_sim_accounts(&self, backend: &BackendConn) -> Result<Vec<CachedTrdAcc>> {
        use futu_backend::proto_internal::{
            sim_odr_sys_cmn, user_account_info_manager_service as uaims,
        };
        use prost::Message;

        let user_id = backend.user_id.load(std::sync::atomic::Ordering::Relaxed) as u64;

        let req = uaims::AccountListReq {
            msg_header: Some(sim_odr_sys_cmn::MsgHeader {
                req_id: Some(format!("{user_id}_sim")),
                account_id: Some(0),
                cipher: None,
                security_type: None,
                exchange_code: None,
                input_source: None,
                market: None,
            }),
            nn_uid: Some(user_id),
        };

        // Proto 14800 不用 OMBinSrz 包装，直接发 protobuf
        let resp = match backend.request(14800, req.encode_to_vec()).await {
            Ok(r) => r,
            Err(e) => {
                tracing::warn!(error = %e, "sim account query failed, skipping");
                return Ok(vec![]);
            }
        };

        tracing::info!(
            body_len = resp.body.len(),
            body_hex = ?&resp.body[..resp.body.len().min(64)],
            "proto14800 response"
        );
        // Proto 14800 不用 OMBinSrz，直接解析 protobuf
        let parsed: uaims::AccountListRsp = Message::decode(resp.body.as_ref()).map_err(|e| {
            FutuError::Codec(format!(
                "Proto14800 decode failed (body_len={}): {e}",
                resp.body.len()
            ))
        })?;

        if parsed.result.unwrap_or(-1) != 0 {
            let err_msg = parsed.err_msg.as_deref().unwrap_or("unknown");
            tracing::warn!(
                result = parsed.result,
                err = err_msg,
                "sim account query error"
            );
            return Ok(vec![]);
        }

        // 模拟账户: market_id 已是 TrdMarket 值
        // C++ 特殊映射: Sim_HK_Option → HK, Sim_US_Option/Margin → US
        let accounts: Vec<CachedTrdAcc> = parsed
            .accouts // proto 中的拼写错误 (accouts)
            .iter()
            .filter_map(|acc| {
                let acc_id = acc.account_id?;
                let raw_market = acc.market_id.unwrap_or(0) as i32;

                // 特殊市场映射 (C++ APIServer_Trd_GetAccList.cpp:138-150)
                // Sim_HK_Option(9) → HK(1), Sim_US_Option(16)/Sim_US_Margin → US(2)
                let trd_market = match raw_market {
                    9 => 1,  // HK option → TrdMarket_HK
                    16 => 2, // US option → TrdMarket_US
                    _ => raw_market,
                };

                // 从 market_id 推断 sim_acc_type 和 acc_type
                // C++ NNProto_Trd_SimAccList::UnPackAccList
                let (sim_acc_type, acc_type) = match raw_market {
                    9 | 16 => (Some(2), Some(2)), // 期权 → SimType_Option, AccType_Margin
                    _ if raw_market == 100 => (Some(3), Some(2)), // 期货 → SimType_Futures, AccType_Margin
                    _ if raw_market == 2 => (Some(1), Some(2)), // US 模拟 → SimType_Stock, AccType_Margin (卖空支持)
                    _ => (Some(1), Some(1)), // 其他股票 → SimType_Stock, AccType_Cash
                };

                // C++ sim: enBrokerID=NN_BrokerID_Unknown(0), nIntraAccID=0
                let sort_key = (raw_market as u64) << 32;

                Some(CachedTrdAcc {
                    acc_id,
                    trd_env: 0, // TrdEnv_Simulate
                    trd_market_auth_list: vec![trd_market],
                    acc_type,
                    card_num: None,
                    security_firm: Some(0), // SecurityFirm_Unknown
                    sim_acc_type,
                    uni_card_num: None,
                    acc_status: Some(0), // Active
                    acc_role: None,
                    jp_acc_type: vec![],
                    owner_uid: None,
                    opr_uid: None,
                    mixed_state: None,
                    ira_type: None,
                    grant_state: None,
                    kouza_type: None,
                    trd_market: Some(raw_market),
                    association_acc_id: None,
                    acc_flag: None,
                    order_index: 0, // set_accounts 会重新赋值
                    sort_key,
                })
            })
            .collect();

        tracing::info!(count = accounts.len(), "sim accounts fetched");
        Ok(accounts)
    }

    /// 拉取所有账户（真实 + 模拟）并填充缓存
    async fn fetch_account_list(&self, backend: &BackendConn) -> Result<()> {
        let mut all = self.fetch_real_accounts(backend).await?;
        let sim = self.fetch_sim_accounts(backend).await?;
        all.extend(sim);

        tracing::info!(
            total = all.len(),
            accounts = ?all.iter().map(|a| (a.acc_id, a.trd_env, &a.trd_market_auth_list)).collect::<Vec<_>>(),
            "all accounts cached"
        );
        self.trd_cache.set_accounts(all);
        Ok(())
    }

    /// 注册所有业务处理器
    pub fn register_handlers(&self, server: &ApiServer) {
        let router = server.router();
        handlers::qot::register_handlers(router, self);
        handlers::trd::register_handlers(router, self);
        handlers::sys::register_handlers(router, self);
        tracing::info!("all business handlers registered");
    }

    /// 创建推送分发器
    pub fn create_push_dispatcher(
        &self,
        server: &ApiServer,
        external_sinks: Vec<Arc<dyn futu_server::push::ExternalPushSink>>,
    ) -> PushDispatcher {
        let mut dispatcher = PushDispatcher::new(
            Arc::clone(server.connections()),
            Arc::clone(&self.subscriptions),
        )
        .with_metrics(Arc::clone(&self.metrics));
        for sink in external_sinks {
            dispatcher = dispatcher.with_external_sink(sink);
        }
        dispatcher
    }

    /// 启动推送分发任务: 消费 push_rx channel，处理行情推送和交易重查
    ///
    /// `external_sinks`: 外部推送接收器列表 (REST WebSocket, gRPC 等)
    pub fn start_push_dispatcher(
        &self,
        server: &ApiServer,
        mut push_rx: tokio::sync::mpsc::Receiver<PushEvent>,
        external_sinks: Vec<Arc<dyn futu_server::push::ExternalPushSink>>,
    ) {
        let dispatcher = self.create_push_dispatcher(server, external_sinks);
        let shared_backend = Arc::clone(&self.backend);
        let trd_cache = Arc::clone(&self.trd_cache);
        tokio::spawn(async move {
            tracing::info!("push dispatcher task started");
            while let Some(event) = push_rx.recv().await {
                match event {
                    PushEvent::QuotePush {
                        sec_key,
                        sub_type,
                        proto_id,
                        body,
                    } => {
                        dispatcher
                            .push_qot(&sec_key, sub_type, proto_id, body)
                            .await;
                    }
                    PushEvent::TradeReQuery {
                        acc_id,
                        notice_type,
                        order_ids,
                        order_fill_ids,
                    } => {
                        // 收到交易通知后重查对应数据 + 推送给订阅客户端
                        let be_opt = crate::handlers::load_backend(&shared_backend);
                        if let Some(ref be) = be_opt {
                            match notice_type {
                                1 | 11 | 12 => {
                                    // CASH_UPDATE / ASSET_UPDATE / QUOTE_UPDATE → 重查资金
                                    let _ = futu_backend::trade_query::query_account_info(
                                        be, acc_id, &trd_cache,
                                    )
                                    .await;
                                }
                                2 | 3 => {
                                    // PSTN_UPDATE / PSTN_LIST_UPDATE → 重查持仓+资金
                                    let _ = futu_backend::trade_query::query_account_info(
                                        be, acc_id, &trd_cache,
                                    )
                                    .await;
                                }
                                4 | 5 | 9 => {
                                    // ORDER_UPDATE / ORDER_LIST_UPDATE / ORDER_OP_RESULT
                                    // → 重查订单 + 资金 + 持仓，推送 Trd_UpdateOrder
                                    if let Ok(orders) = futu_backend::trade_query::query_orders(
                                        be, acc_id, &trd_cache,
                                    )
                                    .await
                                    {
                                        // 推送变动的订单 (如有 order_ids 则只推对应订单，否则推全部)
                                        for order in &orders {
                                            let should_push = order_ids.is_empty()
                                                || order_ids
                                                    .iter()
                                                    .any(|id| id == &order.order_id_ex);
                                            if should_push {
                                                let push_body =
                                                    Self::build_order_update_push(acc_id, order);
                                                dispatcher
                                                    .push_trd_acc(
                                                        acc_id,
                                                        futu_core::proto_id::TRD_UPDATE_ORDER,
                                                        push_body,
                                                    )
                                                    .await;
                                            }
                                        }
                                    }
                                    let _ = futu_backend::trade_query::query_account_info(
                                        be, acc_id, &trd_cache,
                                    )
                                    .await;
                                    // query_account_info already fetched positions too
                                }
                                6 | 7 => {
                                    // ORDER_FILL_UPDATE / ORDER_FILL_LIST_UPDATE
                                    // → 查成交列表 + 推送 Trd_UpdateOrderFill
                                    if let Ok(fills) =
                                        futu_backend::trade_query::query_order_fills(be, acc_id)
                                            .await
                                    {
                                        for fill in &fills {
                                            let should_push = order_fill_ids.is_empty()
                                                || order_fill_ids
                                                    .iter()
                                                    .any(|id| id == &fill.fill_id_ex);
                                            if should_push {
                                                let push_body = Self::build_order_fill_update_push(
                                                    acc_id, fill,
                                                );
                                                dispatcher
                                                    .push_trd_acc(
                                                        acc_id,
                                                        futu_core::proto_id::TRD_UPDATE_ORDER_FILL,
                                                        push_body,
                                                    )
                                                    .await;
                                            }
                                        }
                                    }
                                    // 成交变动通常伴随资金/持仓变化
                                    let _ = futu_backend::trade_query::query_account_info(
                                        be, acc_id, &trd_cache,
                                    )
                                    .await;
                                    // query_account_info already fetched positions too
                                }
                                _ => {}
                            }
                        }
                    }
                    PushEvent::BroadcastPush { proto_id, body } => {
                        // 广播推送: 发送给所有已注册推送通知的客户端
                        dispatcher.push_broadcast(proto_id, body).await;
                    }
                }
            }
            tracing::warn!("push dispatcher task ended");
        });
    }

    /// 构建 Trd_UpdateOrder 推送 body (C++ APIServer_Trd_UpdateOrder 对齐)
    fn build_order_update_push(acc_id: u64, order: &futu_cache::trd_cache::CachedOrder) -> Vec<u8> {
        let header = futu_proto::trd_common::TrdHeader {
            trd_env: 1,
            acc_id,
            trd_market: order.trd_market.unwrap_or(1),
            jp_acc_type: None,
        };
        let api_order = futu_proto::trd_common::Order {
            trd_side: order.trd_side,
            order_type: order.order_type,
            order_status: order.order_status,
            order_id: order.order_id,
            order_id_ex: order.order_id_ex.clone(),
            code: order.code.clone(),
            name: order.name.clone(),
            qty: order.qty,
            price: Some(order.price),
            create_time: order.create_time.clone(),
            update_time: order.update_time.clone(),
            fill_qty: Some(order.fill_qty),
            fill_avg_price: Some(order.fill_avg_price),
            last_err_msg: order.last_err_msg.clone(),
            sec_market: order.sec_market,
            create_timestamp: order.create_timestamp,
            update_timestamp: order.update_timestamp,
            remark: order.remark.clone(),
            time_in_force: order.time_in_force,
            fill_outside_rth: order.fill_outside_rth,
            aux_price: order.aux_price,
            trail_type: order.trail_type,
            trail_value: order.trail_value,
            trail_spread: order.trail_spread,
            currency: order.currency,
            trd_market: order.trd_market,
            session: None,
            jp_acc_type: None,
        };
        let resp = futu_proto::trd_update_order::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_update_order::S2c {
                header,
                order: api_order,
            }),
        };
        prost::Message::encode_to_vec(&resp)
    }

    /// 构建 Trd_UpdateOrderFill 推送 body (C++ APIServer_Trd_UpdateOrderFill 对齐)
    fn build_order_fill_update_push(
        acc_id: u64,
        fill: &futu_backend::trade_query::OrderFillInfo,
    ) -> Vec<u8> {
        let header = futu_proto::trd_common::TrdHeader {
            trd_env: 1,
            acc_id,
            trd_market: fill.trd_market.unwrap_or(1),
            jp_acc_type: None,
        };
        let api_fill = futu_proto::trd_common::OrderFill {
            trd_side: fill.trd_side,
            fill_id: fill.fill_id,
            fill_id_ex: fill.fill_id_ex.clone(),
            order_id: Some(fill.order_id),
            order_id_ex: Some(fill.order_id_ex.clone()),
            code: fill.code.clone(),
            name: fill.name.clone(),
            qty: fill.qty,
            price: fill.price,
            create_time: String::new(),
            counter_broker_id: fill.counter_broker_id,
            counter_broker_name: None,
            sec_market: None,
            create_timestamp: fill.create_timestamp,
            update_timestamp: fill.update_timestamp,
            status: fill.status,
            trd_market: fill.trd_market,
            jp_acc_type: None,
        };
        let resp = futu_proto::trd_update_order_fill::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_update_order_fill::S2c {
                header,
                order_fill: api_fill,
            }),
        };
        prost::Message::encode_to_vec(&resp)
    }
}

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

/// 重连后重新订阅行情
///
/// 遍历 SubscriptionManager 中所有客户端的行情订阅，
/// 聚合后发送 CMD 6211 到新的后端连接。
/// 返回重新订阅的证券数量。
async fn resubscribe_quotes(
    backend: &BackendConn,
    subscriptions: &SubscriptionManager,
    static_cache: &StaticDataCache,
) -> usize {
    use std::collections::{HashMap, HashSet};

    // 收集所有唯一的 (security_key, sub_type) 对
    let all_conn_ids = subscriptions.get_all_qot_conn_ids();
    let mut unique_subs: HashMap<String, HashSet<i32>> = HashMap::new();

    for conn_id in &all_conn_ids {
        let conn_subs = subscriptions.get_conn_qot_subs(*conn_id);
        for (sub_type, sec_keys) in conn_subs {
            for sec_key in sec_keys {
                unique_subs.entry(sec_key).or_default().insert(sub_type);
            }
        }
    }

    if unique_subs.is_empty() {
        tracing::debug!("no quote subscriptions to re-subscribe");
        return 0;
    }

    // 构建 (stock_id, ftapi_market, sub_types) 列表
    let mut backend_subs: Vec<(u64, i32, Vec<i32>)> = Vec::new();
    for (sec_key, sub_types) in &unique_subs {
        // sec_key 格式: "market_code"
        if let Some(info) = static_cache.get_security_info(sec_key) {
            if info.stock_id > 0 {
                let market = sec_key
                    .split('_')
                    .next()
                    .and_then(|s| s.parse::<i32>().ok())
                    .unwrap_or(0);
                backend_subs.push((info.stock_id, market, sub_types.iter().copied().collect()));
            }
        }
    }

    let count = backend_subs.len();
    if count == 0 {
        tracing::debug!("no valid stock_ids found for re-subscription");
        return 0;
    }

    tracing::info!(
        securities = count,
        connections = all_conn_ids.len(),
        "re-subscribing quotes after reconnect"
    );

    match futu_backend::quote_sub::subscribe_to_backend(backend, &backend_subs).await {
        Ok(()) => {
            tracing::info!(securities = count, "quote re-subscription succeeded");
        }
        Err(e) => {
            tracing::error!(error = %e, "quote re-subscription failed");
        }
    }

    count
}