futu_gateway/handlers/

qot.rs

// 行情业务处理器 — 全量实现（无占位）

use std::sync::Arc;

use async_trait::async_trait;

use futu_cache::login_cache::LoginCache;
use futu_cache::qot_cache::{self, QotCache};
use futu_cache::static_data::StaticDataCache;
use futu_core::proto_id;
use futu_server::conn::IncomingRequest;
use futu_server::router::{RequestHandler, RequestRouter};
use futu_server::subscription::SubscriptionManager;

use crate::bridge::GatewayBridge;

#[allow(dead_code)]
fn hex_preview(data: &[u8], max: usize) -> String {
    let n = data.len().min(max);
    data[..n]
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect::<Vec<_>>()
        .join(" ")
}

/// 注册所有行情处理器
pub fn register_handlers(router: &Arc<RequestRouter>, bridge: &GatewayBridge) {
    let cache = Arc::clone(&bridge.qot_cache);
    let subs = Arc::clone(&bridge.subscriptions);
    let static_cache = Arc::clone(&bridge.static_cache);

    // GetGlobalState
    router.register(
        proto_id::GET_GLOBAL_STATE,
        Arc::new(GetGlobalStateHandler {
            login_cache: Arc::clone(&bridge.login_cache),
            backend: bridge.backend.clone(),
        }),
    );

    // 连接管理
    router.register(
        proto_id::QOT_SUB,
        Arc::new(SubHandler {
            subscriptions: subs.clone(),
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_REG_QOT_PUSH,
        Arc::new(RegQotPushHandler {
            subscriptions: subs.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_SUB_INFO,
        Arc::new(GetSubInfoHandler {
            subscriptions: subs.clone(),
        }),
    );

    // 实时行情
    router.register(
        proto_id::QOT_GET_BASIC_QOT,
        Arc::new(GetBasicQotHandler {
            cache: cache.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_KL,
        Arc::new(GetKLHandler {
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_ORDER_BOOK,
        Arc::new(GetOrderBookHandler {
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_BROKER,
        Arc::new(GetBrokerHandler {
            cache: cache.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_TICKER,
        Arc::new(GetTickerHandler {
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_RT,
        Arc::new(GetRTHandler {
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_BROKER,
        Arc::new(GetBrokerHandler {
            cache: cache.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    // NOTE: QOT_GET_ORDER_DETAIL proto removed; handler not registered.

    // 静态/快照
    router.register(
        proto_id::QOT_GET_STATIC_INFO,
        Arc::new(GetStaticInfoHandler {
            cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_SECURITY_SNAPSHOT,
        Arc::new(GetSecuritySnapshotHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );

    // 历史/市场数据
    router.register(
        proto_id::QOT_GET_HISTORY_KL,
        Arc::new(GetHistoryKLHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_REQUEST_HISTORY_KL,
        Arc::new(RequestHistoryKLHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
            kl_quota_counter: bridge.kl_quota_counter.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_HISTORY_KL_POINTS,
        Arc::new(GetHistoryKLPointsHandler),
    );
    router.register(
        proto_id::QOT_GET_TRADE_DATE,
        Arc::new(GetTradeDateHandler {
            backend: bridge.backend.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_SUSPEND,
        Arc::new(GetSuspendHandler {
            suspend_cache: bridge.suspend_cache.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_REHAB,
        Arc::new(GetRehabHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_PLATE_SET,
        Arc::new(GetPlateSetHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_PLATE_SECURITY,
        Arc::new(GetPlateSecurityHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_OWNER_PLATE,
        Arc::new(GetOwnerPlateHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_REFERENCE,
        Arc::new(GetReferenceHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_OPTION_CHAIN,
        Arc::new(GetOptionChainHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_HOLDING_CHANGE_LIST,
        Arc::new(GetHoldingChangeListHandler),
    );

    // ===== 新版 API handler 注册 =====
    router.register(
        proto_id::QOT_GET_WARRANT,
        Arc::new(GetWarrantHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_CAPITAL_FLOW,
        Arc::new(GetCapitalFlowHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_CAPITAL_DISTRIBUTION,
        Arc::new(GetCapitalDistributionHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_USER_SECURITY,
        Arc::new(GetUserSecurityHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
            app_lang: bridge.app_lang,
        }),
    );
    router.register(
        proto_id::QOT_MODIFY_USER_SECURITY,
        Arc::new(ModifyUserSecurityHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_STOCK_FILTER,
        Arc::new(StockFilterHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_CODE_CHANGE,
        Arc::new(GetCodeChangeHandler {
            code_change_cache: bridge.code_change_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_IPO_LIST,
        Arc::new(GetIpoListHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_FUTURE_INFO,
        Arc::new(GetFutureInfoHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_REQUEST_TRADE_DATE,
        Arc::new(RequestTradeDateHandler {
            backend: bridge.backend.clone(),
        }),
    );
    router.register(
        proto_id::QOT_SET_PRICE_REMINDER,
        Arc::new(SetPriceReminderHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_PRICE_REMINDER,
        Arc::new(GetPriceReminderHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_USER_SECURITY_GROUP,
        Arc::new(GetUserSecurityGroupHandler {
            backend: bridge.backend.clone(),
            app_lang: bridge.app_lang,
        }),
    );
    router.register(
        proto_id::QOT_GET_MARKET_STATE,
        Arc::new(GetMarketStateHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_GET_OPTION_EXPIRATION_DATE,
        Arc::new(GetOptionExpirationDateHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );
    router.register(
        proto_id::QOT_REQUEST_HISTORY_KL_QUOTA,
        Arc::new(RequestHistoryKLQuotaHandler),
    );
    router.register(
        proto_id::QOT_REQUEST_REHAB,
        Arc::new(RequestRehabHandler {
            backend: bridge.backend.clone(),
            static_cache: static_cache.clone(),
        }),
    );

    // GetUsedQuota (1010)
    router.register(
        proto_id::GET_USED_QUOTA,
        Arc::new(GetUsedQuotaHandler {
            subscriptions: subs.clone(),
            kl_quota_counter: bridge.kl_quota_counter.clone(),
        }),
    );

    tracing::debug!("quote handlers registered (all implemented)");
}

// ===== Sub =====
struct SubHandler {
    subscriptions: Arc<SubscriptionManager>,
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for SubHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_sub::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        // 收集需要向后端订阅的 (stock_id, ftapi_market, sub_types)
        let mut backend_subs: Vec<(u64, i32, Vec<i32>)> = Vec::new();

        for sec in &c2s.security_list {
            let sec_key = format!("{}_{}", sec.market, sec.code);
            for &sub_type in &c2s.sub_type_list {
                if c2s.is_sub_or_un_sub {
                    self.subscriptions
                        .subscribe_qot(conn_id, &sec_key, sub_type);
                } else {
                    self.subscriptions
                        .unsubscribe_qot(conn_id, &sec_key, sub_type);
                }
            }

            // 查找 stock_id 用于后端订阅
            if c2s.is_sub_or_un_sub {
                if let Some(info) = self.static_cache.get_security_info(&sec_key) {
                    if info.stock_id > 0 {
                        backend_subs.push((info.stock_id, sec.market, c2s.sub_type_list.clone()));
                    }
                }
            }
        }

        if c2s.is_reg_or_un_reg_push.unwrap_or(false) {
            self.subscriptions.subscribe_notify(conn_id);
        }

        // 转发到后端 CMD 6211
        if let Some(backend) = super::load_backend(&self.backend) {
            if !backend_subs.is_empty() {
                if let Err(e) =
                    futu_backend::quote_sub::subscribe_to_backend(&backend, &backend_subs).await
                {
                    tracing::warn!(error = %e, "backend subscribe failed");
                }
            }
        }

        let resp = futu_proto::qot_sub::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_sub::S2c {}),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== RegQotPush =====
struct RegQotPushHandler {
    subscriptions: Arc<SubscriptionManager>,
}

#[async_trait]
impl RequestHandler for RegQotPushHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_reg_qot_push::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;
        for sec in &c2s.security_list {
            let sec_key = format!("{}_{}", sec.market, sec.code);
            for &sub_type in &c2s.sub_type_list {
                if c2s.is_reg_or_un_reg {
                    self.subscriptions
                        .subscribe_qot(conn_id, &sec_key, sub_type);
                } else {
                    self.subscriptions
                        .unsubscribe_qot(conn_id, &sec_key, sub_type);
                }
            }
        }
        let resp = futu_proto::qot_reg_qot_push::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_reg_qot_push::S2c {}),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetSubInfo =====
struct GetSubInfoHandler {
    subscriptions: Arc<SubscriptionManager>,
}

#[async_trait]
impl RequestHandler for GetSubInfoHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_sub_info::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let is_req_all = req.c2s.is_req_all_conn.unwrap_or(false);

        let used = self.subscriptions.get_total_used_quota();
        let remain = futu_server::subscription::TOTAL_QUOTA.saturating_sub(used);

        // 构建 ConnSubInfo 列表
        let conn_ids: Vec<u64> = if is_req_all {
            self.subscriptions
                .get_all_qot_conn_ids()
                .into_iter()
                .collect()
        } else {
            vec![conn_id]
        };

        let conn_sub_info_list: Vec<futu_proto::qot_common::ConnSubInfo> = conn_ids
            .iter()
            .map(|&cid| {
                let subs = self.subscriptions.get_conn_qot_subs(cid);
                let sub_info_list: Vec<futu_proto::qot_common::SubInfo> = subs
                    .iter()
                    .map(|(&sub_type, sec_keys)| {
                        let security_list: Vec<futu_proto::qot_common::Security> = sec_keys
                            .iter()
                            .filter_map(|sk| {
                                let parts: Vec<&str> = sk.splitn(2, '_').collect();
                                if parts.len() == 2 {
                                    Some(futu_proto::qot_common::Security {
                                        market: parts[0].parse().unwrap_or(0),
                                        code: parts[1].to_string(),
                                    })
                                } else {
                                    None
                                }
                            })
                            .collect();
                        futu_proto::qot_common::SubInfo {
                            sub_type,
                            security_list,
                        }
                    })
                    .collect();
                let conn_used = self.subscriptions.get_conn_used_quota(cid);
                futu_proto::qot_common::ConnSubInfo {
                    sub_info_list,
                    used_quota: conn_used as i32,
                    is_own_conn_data: cid == conn_id,
                }
            })
            .collect();

        let resp = futu_proto::qot_get_sub_info::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_sub_info::S2c {
                conn_sub_info_list,
                total_used_quota: used as i32,
                remain_quota: remain as i32,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetBasicQot =====
struct GetBasicQotHandler {
    cache: Arc<QotCache>,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetBasicQotHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_basic_qot::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let mut basic_qot_list = Vec::new();
        for sec in &req.c2s.security_list {
            let key = qot_cache::make_key(sec.market, &sec.code);
            if let Some(c) = self.cache.get_basic_qot(&key) {
                // 从 static_cache 补全 name 和 listing_date
                let static_info = self.static_cache.get_security_info(&key);
                let name = static_info.as_ref().map(|s| s.name.clone());
                let list_time = static_info
                    .as_ref()
                    .map(|s| s.list_time.clone())
                    .unwrap_or_default();

                basic_qot_list.push(futu_proto::qot_common::BasicQot {
                    security: sec.clone(),
                    name,
                    is_suspended: c.is_suspended,
                    list_time,
                    price_spread: 0.0,
                    update_time: c.update_time.clone(),
                    high_price: c.high_price,
                    open_price: c.open_price,
                    low_price: c.low_price,
                    cur_price: c.cur_price,
                    last_close_price: c.last_close_price,
                    volume: c.volume,
                    turnover: c.turnover,
                    turnover_rate: c.turnover_rate,
                    amplitude: c.amplitude,
                    dark_status: None,
                    option_ex_data: None,
                    list_timestamp: None,
                    update_timestamp: Some(c.update_timestamp),
                    pre_market: None,
                    after_market: None,
                    sec_status: None,
                    future_ex_data: None,
                    warrant_ex_data: None,
                    overnight: None,
                });
            }
        }
        let resp = futu_proto::qot_get_basic_qot::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_basic_qot::S2c { basic_qot_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetKL =====
struct GetKLHandler {
    cache: Arc<QotCache>,
}

#[async_trait]
impl RequestHandler for GetKLHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_kl::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;
        let key = qot_cache::make_key(c2s.security.market, &c2s.security.code);
        let kl_list = self
            .cache
            .get_klines(&key, c2s.kl_type)
            .unwrap_or_default()
            .into_iter()
            .take(c2s.req_num as usize)
            .map(|kl| futu_proto::qot_common::KLine {
                time: kl.time,
                is_blank: false,
                high_price: Some(kl.high_price),
                open_price: Some(kl.open_price),
                low_price: Some(kl.low_price),
                close_price: Some(kl.close_price),
                last_close_price: None,
                volume: Some(kl.volume),
                turnover: Some(kl.turnover),
                turnover_rate: None,
                pe: None,
                change_rate: None,
                timestamp: None,
            })
            .collect();
        let resp = futu_proto::qot_get_kl::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_kl::S2c {
                security: c2s.security.clone(),
                name: None,
                kl_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetOrderBook =====
struct GetOrderBookHandler {
    cache: Arc<QotCache>,
}

#[async_trait]
impl RequestHandler for GetOrderBookHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_order_book::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;
        let key = qot_cache::make_key(c2s.security.market, &c2s.security.code);
        let ob = self.cache.order_books.get(&key);
        let n = c2s.num as usize;
        let (ask_list, bid_list) = match ob.as_deref() {
            Some(ob) => (
                ob.ask_list
                    .iter()
                    .take(n)
                    .map(|a| futu_proto::qot_common::OrderBook {
                        price: a.price,
                        volume: a.volume,
                        oreder_count: a.order_count,
                        detail_list: vec![],
                    })
                    .collect(),
                ob.bid_list
                    .iter()
                    .take(n)
                    .map(|b| futu_proto::qot_common::OrderBook {
                        price: b.price,
                        volume: b.volume,
                        oreder_count: b.order_count,
                        detail_list: vec![],
                    })
                    .collect(),
            ),
            None => (vec![], vec![]),
        };
        let resp = futu_proto::qot_get_order_book::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_order_book::S2c {
                security: c2s.security.clone(),
                name: None,
                order_book_ask_list: ask_list,
                order_book_bid_list: bid_list,
                svr_recv_time_bid: None,
                svr_recv_time_bid_timestamp: None,
                svr_recv_time_ask: None,
                svr_recv_time_ask_timestamp: None,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetTicker =====
struct GetTickerHandler {
    cache: Arc<QotCache>,
}

#[async_trait]
impl RequestHandler for GetTickerHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_ticker::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;
        let key = qot_cache::make_key(c2s.security.market, &c2s.security.code);
        let n = c2s.max_ret_num as usize;
        let tickers = self.cache.tickers.get(&key);
        let ticker_list: Vec<futu_proto::qot_common::Ticker> = tickers
            .as_deref()
            .map(|t| {
                t.iter()
                    .rev()
                    .take(n)
                    .rev()
                    .map(|t| futu_proto::qot_common::Ticker {
                        time: t.time.clone(),
                        sequence: 0,
                        dir: t.dir,
                        price: t.price,
                        volume: t.volume,
                        turnover: 0.0,
                        recv_time: None,
                        r#type: None,
                        type_sign: None,
                        push_data_type: None,
                        timestamp: None,
                    })
                    .collect()
            })
            .unwrap_or_default();
        let resp = futu_proto::qot_get_ticker::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_ticker::S2c {
                security: c2s.security.clone(),
                name: None,
                ticker_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetRT =====
struct GetRTHandler {
    cache: Arc<QotCache>,
}

#[async_trait]
impl RequestHandler for GetRTHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_rt::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let key = qot_cache::make_key(req.c2s.security.market, &req.c2s.security.code);
        let rt_list = self
            .cache
            .rt_data
            .get(&key)
            .map(|v| {
                v.iter()
                    .map(|rt| futu_proto::qot_common::TimeShare {
                        time: rt.time.clone(),
                        minute: rt.minute,
                        is_blank: false,
                        price: Some(rt.price),
                        last_close_price: if rt.last_close_price > 0.0 {
                            Some(rt.last_close_price)
                        } else {
                            None
                        },
                        avg_price: if rt.avg_price > 0.0 {
                            Some(rt.avg_price)
                        } else {
                            None
                        },
                        volume: Some(rt.volume),
                        turnover: Some(rt.turnover),
                        timestamp: Some(rt.timestamp),
                    })
                    .collect()
            })
            .unwrap_or_default();
        let resp = futu_proto::qot_get_rt::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_rt::S2c {
                security: req.c2s.security.clone(),
                name: None,
                rt_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetStaticInfo =====
struct GetStaticInfoHandler {
    cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetStaticInfoHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_static_info::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let mut static_info_list = Vec::new();

        if !req.c2s.security_list.is_empty() {
            // 模式 1: 指定股票列表查询
            for sec in &req.c2s.security_list {
                let key = qot_cache::make_key(sec.market, &sec.code);
                if let Some(info) = self.cache.get_security_info(&key) {
                    static_info_list.push(make_static_info(sec.clone(), &info));
                }
            }
        } else {
            // 模式 2/3: 按市场 (+ 类型) 查询
            let filter_market = req.c2s.market;
            let filter_sec_type = req.c2s.sec_type;
            for entry in self.cache.securities.iter() {
                let info = entry.value();
                // 对齐 C++ SQL: WHERE no_search=0 (不返回不可搜索的股票)
                if info.no_search {
                    continue;
                }
                if let Some(m) = filter_market {
                    if info.market != m {
                        continue;
                    }
                }
                if let Some(st) = filter_sec_type {
                    if info.sec_type != st {
                        continue;
                    }
                }
                let sec = futu_proto::qot_common::Security {
                    market: info.market,
                    code: info.code.clone(),
                };
                static_info_list.push(make_static_info(sec, info));
            }
        }

        let resp = futu_proto::qot_get_static_info::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_static_info::S2c { static_info_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

fn make_static_info(
    security: futu_proto::qot_common::Security,
    info: &futu_cache::static_data::CachedSecurityInfo,
) -> futu_proto::qot_common::SecurityStaticInfo {
    futu_proto::qot_common::SecurityStaticInfo {
        basic: futu_proto::qot_common::SecurityStaticBasic {
            security,
            id: info.stock_id as i64,
            lot_size: info.lot_size,
            sec_type: info.sec_type,
            name: info.name.clone(),
            list_time: info.list_time.clone(),
            delisting: Some(info.delisting),
            list_timestamp: None,
            exch_type: if info.exch_type != 0 {
                Some(info.exch_type)
            } else {
                None
            },
        },
        warrant_ex_data: None,
        option_ex_data: None,
        future_ex_data: None,
    }
}

// ===== GetSecuritySnapshot (CMD 6824) =====
struct GetSecuritySnapshotHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

/// CMD 6824
const CMD_PULL_SUB_DATA: u16 = 6824;

/// 从 BitQuote 数据解析快照信息
#[derive(Default)]
struct SnapshotData {
    // SBIT_PRICE
    cur_price: i64,
    last_close_price: i64,
    timestamp: i64,
    // SBIT_STOCK_STATE
    suspend_flag: bool,
    sec_status: i32,
    // SBIT_DEAL_STATISTICS
    open_price: i64,
    high_price: i64,
    low_price: i64,
    volume: i64,
    turnover: i64,
    turnover_ratio: i64,
    amplitude: i64,
    avg_price: i64,
    bid_ask_ratio: i64,
    volume_ratio: i64,
    // KCB after market
    kcb_after_volume: i64,
    kcb_after_turnover: i64,
    kcb_has_after: bool,
    // SBIT_ORDER_BOOK_SIMPLE
    bid_price: i64,
    ask_price: i64,
    bid_vol: i64,
    ask_vol: i64,
    // SBIT_HISTORY_HIGHLOW_PRICE
    history_highest_price: i64,
    history_lowest_price: i64,
    week52_highest_price: i64,
    week52_lowest_price: i64,
    // SBIT_HISTORY_CLOSE_PRICE
    close_price_5min: i64,
    // SBIT_FINANCIAL_INDICATOR (Eqty only)
    total_shares: i64,
    total_market_cap: i64,
    outstanding_shares: i64,
    pe_lyr: i64,
    pb_ratio: i64,
    pe_ttm: i64,
    eps_lyr: i64,
    dividend_ttm: i64,
    dividend_ratio_ttm: i64,
    dividend_lfy: i64,
    dividend_lfy_ratio: i64,
    // SBIT_STATIC_FINANCIAL_INDICATOR (Eqty only)
    net_asset: i64,
    net_asset_pershare: i64,
    net_profit: i64,
    ey_ratio: i32,
    // SBIT_STOCK_TYPE_SPECIFIC
    stock_specific: Option<StockSpecificData>,
    // SBIT_US_PREMARKET_AFTERHOURS_DETAIL
    pre_market: Option<PreAfterMarketItem>,
    after_market: Option<PreAfterMarketItem>,
    overnight: Option<PreAfterMarketItem>,
}

struct PreAfterMarketItem {
    price: i64,
    high_price: i64,
    low_price: i64,
    volume: i64,
    turnover: i64,
    change_val: i64,
    change_ratio: i64,
    amplitude: i64,
}

enum StockSpecificData {
    Warrant(futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data::HkWarrantCbbc),
    Option(futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data::Option),
    Index(futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data::Index),
    Plate(futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data::Plate),
    Future(futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data::Future),
    Trust(futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data::Trust),
}

/// 价格转换: raw / 10^precision
fn dp(raw: i64, precision: u32) -> f64 {
    raw as f64 / 10f64.powi(precision as i32)
}

/// 解析一个 SecurityQuote 的所有 BitQuote
fn parse_security_quote(
    sq: &futu_backend::proto_internal::ft_cmd_stock_quote_sub_data::SecurityQuote,
    sec_type: i32,
    is_us: bool,
) -> SnapshotData {
    use futu_backend::proto_internal::ft_cmd_stock_quote_coverage_data as cov;
    use prost::Message;
    let mut s = SnapshotData::default();
    for bq in &sq.bit_qta_list {
        let bit = bq.bit.unwrap_or(999);
        let data = &bq.data.as_deref().unwrap_or(&[]);
        match bit {
            0 => {
                // SBIT_PRICE
                if let Ok(p) = cov::Price::decode(*data) {
                    s.cur_price = p.price_nominal.unwrap_or(0);
                    s.last_close_price = p.price_last_close.unwrap_or(0);
                    let exch_time_ms = p.exchange_data_time_ms.unwrap_or(0);
                    s.timestamp = exch_time_ms / 1000;
                }
            }
            1 => {
                // SBIT_STOCK_STATE
                if let Ok(st) = cov::StockState::decode(*data) {
                    let state_type = st.state_type.unwrap_or(0);
                    s.suspend_flag = state_type == 8; // STOCK_STATE_SUSPENDED
                    s.sec_status = stock_state_to_api_status(state_type);
                }
            }
            2 => {
                // SBIT_STOCK_TYPE_SPECIFIC
                if let Ok(sp) = cov::StockTypeSpecific::decode(*data) {
                    match sec_type {
                        5 => {
                            if let Some(w) = sp.hk_warrant_cbbc {
                                s.stock_specific = Some(StockSpecificData::Warrant(w));
                            }
                        }
                        8 => {
                            if let Some(o) = sp.option {
                                s.stock_specific = Some(StockSpecificData::Option(o));
                            }
                        }
                        6 => {
                            if let Some(i) = sp.index {
                                s.stock_specific = Some(StockSpecificData::Index(i));
                            }
                        }
                        7 => {
                            if let Some(p) = sp.plate {
                                s.stock_specific = Some(StockSpecificData::Plate(p));
                            }
                        }
                        10 => {
                            if let Some(f) = sp.future {
                                s.stock_specific = Some(StockSpecificData::Future(f));
                            }
                        }
                        4 => {
                            if let Some(t) = sp.trust {
                                s.stock_specific = Some(StockSpecificData::Trust(t));
                            }
                        }
                        _ => {}
                    }
                }
            }
            4 => {
                // SBIT_ORDER_BOOK_SIMPLE
                if let Ok(ob) = cov::OrderBookSimple::decode(*data) {
                    s.bid_price = ob.price_bid.unwrap_or(0);
                    s.ask_price = ob.price_ask.unwrap_or(0);
                    s.bid_vol = ob.volume_bid.unwrap_or(0);
                    s.ask_vol = ob.volume_ask.unwrap_or(0);
                }
            }
            5 => {
                // SBIT_DEAL_STATISTICS
                if let Ok(ds) = cov::DealStatistics::decode(*data) {
                    s.open_price = ds.price_open.unwrap_or(0);
                    s.high_price = ds.price_highest.unwrap_or(0);
                    s.low_price = ds.price_lowest.unwrap_or(0);
                    s.volume = ds.volume.unwrap_or(0);
                    s.turnover = ds.turnover.unwrap_or(0);
                    s.turnover_ratio = ds.ratio_turnover.unwrap_or(0);
                    s.amplitude = ds.amplitude_price.unwrap_or(0);
                    s.avg_price = ds.price_average.unwrap_or(0);
                    s.bid_ask_ratio = ds.ratio_bid_ask.unwrap_or(0);
                    s.volume_ratio = ds.ratio_volume.unwrap_or(0);
                    // KCB after market data
                    if let Some(kcb) = &ds.kcb_stock_static {
                        s.kcb_has_after = true;
                        s.kcb_after_volume = kcb.volume.unwrap_or(0);
                        s.kcb_after_turnover = kcb.turnover.unwrap_or(0);
                    }
                }
            }
            6 => {
                // SBIT_HISTORY_HIGHLOW_PRICE
                if let Ok(hh) = cov::HistoryHighLowPrice::decode(*data) {
                    s.history_highest_price = hh.price_highest_history.unwrap_or(0);
                    s.history_lowest_price = hh.price_lowest_history.unwrap_or(0);
                    s.week52_highest_price = hh.price_highest_52week.unwrap_or(0);
                    s.week52_lowest_price = hh.price_lowest_52week.unwrap_or(0);
                }
            }
            7 => {
                // SBIT_HISTORY_CLOSE_PRICE
                if let Ok(hc) = cov::HistoryClosePrice::decode(*data) {
                    s.close_price_5min = hc.price_close_5min.unwrap_or(0);
                }
            }
            8 => {
                // SBIT_FINANCIAL_INDICATOR (Eqty only)
                if sec_type == 3 {
                    // NN_QuoteSecurityType_Eqty
                    if let Ok(fi) = cov::FinacialIndicator::decode(*data) {
                        s.total_shares = fi.total_shares.unwrap_or(0);
                        s.total_market_cap = fi.total_market_cap.unwrap_or(0);
                        s.outstanding_shares = fi.outstanding_shares.unwrap_or(0);
                        s.pe_lyr = fi.pe_lyr.unwrap_or(0);
                        s.pb_ratio = fi.pb_ratio.unwrap_or(0);
                        s.pe_ttm = fi.pe_ttm.unwrap_or(0);
                        s.eps_lyr = fi.eps_lyr.unwrap_or(0);
                        s.dividend_ttm = fi.dividend.unwrap_or(0);
                        s.dividend_ratio_ttm = fi.dividend_ratio.unwrap_or(0);
                        s.dividend_lfy = fi.dividend_lfy.unwrap_or(0);
                        s.dividend_lfy_ratio = fi.dividend_lfy_ratio.unwrap_or(0);
                    }
                }
            }
            18 => {
                // SBIT_STATIC_FINANCIAL_INDICATOR (Eqty only)
                if sec_type == 3 {
                    if let Ok(sfi) = cov::StaticFinancialIndicator::decode(*data) {
                        s.net_asset = sfi.net_asset.unwrap_or(0);
                        s.net_asset_pershare = sfi.net_asset_pershare.unwrap_or(0);
                        s.net_profit = sfi.net_profit_lyr.unwrap_or(0);
                        s.ey_ratio = sfi.ey_ratio.unwrap_or(0);
                    }
                }
            }
            13 => {
                // SBIT_US_PREMARKET_AFTERHOURS_DETAIL (US only)
                if is_us {
                    if let Ok(pa) = cov::UsPreMarketAfterHoursDetail::decode(*data) {
                        if let Some(pm) = &pa.pre_market {
                            s.pre_market = Some(PreAfterMarketItem {
                                price: pm.price.unwrap_or(0),
                                high_price: pm.price_highest.unwrap_or(0),
                                low_price: pm.price_lowest.unwrap_or(0),
                                volume: pm.volume.unwrap_or(0),
                                turnover: pm.turnover.unwrap_or(0),
                                change_val: pm.price_change.unwrap_or(0),
                                change_ratio: pm.ratio_price_change.unwrap_or(0),
                                amplitude: pm.amplitude_price.unwrap_or(0),
                            });
                        }
                        if let Some(ah) = &pa.after_hours {
                            s.after_market = Some(PreAfterMarketItem {
                                price: ah.price.unwrap_or(0),
                                high_price: ah.price_highest.unwrap_or(0),
                                low_price: ah.price_lowest.unwrap_or(0),
                                volume: ah.volume.unwrap_or(0),
                                turnover: ah.turnover.unwrap_or(0),
                                change_val: ah.price_change.unwrap_or(0),
                                change_ratio: ah.ratio_price_change.unwrap_or(0),
                                amplitude: ah.amplitude_price.unwrap_or(0),
                            });
                        }
                        if let Some(on) = &pa.overnight {
                            s.overnight = Some(PreAfterMarketItem {
                                price: on.price.unwrap_or(0),
                                high_price: on.price_highest.unwrap_or(0),
                                low_price: on.price_lowest.unwrap_or(0),
                                volume: on.volume.unwrap_or(0),
                                turnover: on.turnover.unwrap_or(0),
                                change_val: on.price_change.unwrap_or(0),
                                change_ratio: on.ratio_price_change.unwrap_or(0),
                                amplitude: on.amplitude_price.unwrap_or(0),
                            });
                        }
                    }
                }
            }
            _ => {}
        }
    }
    s
}

/// StockStateType → FTAPI Qot_Common::SecurityStatus
fn stock_state_to_api_status(state_type: i32) -> i32 {
    match state_type {
        0 => 0, // Normal → Unknown (C++ maps to NN_QotSecStatus_None=0 for Normal, but API maps differently)
        1 => 2, // PendingListing → Listing
        2 => 3, // IPO_Purchasing_CN → Purchasing
        3 => 4, // IPO_Purchasing_Other → Subscribing
        4 => 5, // DarkTradePending → BeforeDrakTradeOpening
        5 => 6, // DarkTradeHappening → DrakTrading
        6 => 7, // DarkTradeEnd → DrakTradeEnd
        7 => 8, // ToBeOpen → ToBeOpen
        8 => 9, // Suspended → Suspended
        9 => 10, // Called
        10 => 11, // ExpiredLastTradingDate
        11 => 12, // Expired
        12 => 13, // Delisted
        13 => 14, // ChangeToTemporaryCode
        14 => 15, // TemporaryCodeEndTrading
        15 => 16, // ChangedPlateEndTrading
        16 => 17, // ChangedCodeEndTrading
        17 => 18, // RecoverableCircuitBreaker
        18 => 19, // UnRecoverableCircuitBreaker
        19 => 20, // AfterCombination
        20 => 21, // AfterTransaction
        _ => 0,
    }
}

/// PreAfterMarketItem → Qot_Common::PreAfterMarketData
fn build_pre_after_market(item: &PreAfterMarketItem) -> futu_proto::qot_common::PreAfterMarketData {
    futu_proto::qot_common::PreAfterMarketData {
        price: Some(dp(item.price, 9)),
        high_price: Some(dp(item.high_price, 9)),
        low_price: Some(dp(item.low_price, 9)),
        volume: Some(item.volume),
        turnover: Some(dp(item.turnover, 3)),
        change_val: Some(dp(item.change_val, 9)),
        change_rate: Some(dp(item.change_ratio, 3)),
        amplitude: Some(dp(item.amplitude, 4)),
    }
}

#[async_trait]
impl RequestHandler for GetSecuritySnapshotHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_security_snapshot::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Build security info: stock_id, sec_type, qot_market, backend_mkt for each security
        struct SecEntry {
            sec: futu_proto::qot_common::Security,
            stock_id: u64,
            sec_type: i32,
            qot_market: i32,
            name: String,
            lot_size: i32,
            list_time: String,
            backend_mkt: u8, // QuoteMktType
        }

        let mut entries = Vec::new();
        for sec in &req.c2s.security_list {
            let key = qot_cache::make_key(sec.market, &sec.code);
            let info = match self.static_cache.get_security_info(&key) {
                Some(i) if i.stock_id > 0 => i,
                _ => continue,
            };
            let backend_mkt = match futu_backend::stock_list::qot_market_to_backend(sec.market) {
                Some(m) => m as u8,
                None => continue,
            };
            entries.push(SecEntry {
                sec: sec.clone(),
                stock_id: info.stock_id,
                sec_type: info.sec_type,
                qot_market: sec.market,
                name: info.name.clone(),
                lot_size: info.lot_size,
                list_time: info.list_time.clone(),
                backend_mkt,
            });
        }

        // Group by backend market
        let mut market_groups: std::collections::HashMap<u8, Vec<usize>> =
            std::collections::HashMap::new();
        for (idx, entry) in entries.iter().enumerate() {
            market_groups
                .entry(entry.backend_mkt)
                .or_default()
                .push(idx);
        }

        // Send CMD 6824 for each market group, collect responses
        let mut stock_snapshots: std::collections::HashMap<u64, SnapshotData> =
            std::collections::HashMap::new();

        for (mkt, indices) in &market_groups {
            // Build FetchQuoteReq
            let mut security_list = Vec::new();
            for &idx in indices {
                let entry = &entries[idx];
                let mut bit_info_list = Vec::new();
                // 10 subscription bits (same as C++ GetSnapshotSubBit)
                for &bit in &[0u32, 1, 2, 5, 4, 8, 18, 7, 6, 13] {
                    let prob = if bit == 6 { Some(2i64) } else { None }; // SBIT_HISTORY_HIGHLOW_PRICE: prob=Forward
                    bit_info_list.push(
                        futu_backend::proto_internal::ft_cmd_stock_quote_sub_data::BitInfo {
                            bit: Some(bit),
                            prob,
                            prob2: None,
                        },
                    );
                }
                security_list.push(
                    futu_backend::proto_internal::ft_cmd_stock_quote_sub_data::SecuritySubscribe {
                        security_id: Some(entry.stock_id),
                        bit_info_list,
                    },
                );
            }

            let fetch_req = futu_backend::proto_internal::ft_cmd_stock_quote_fetch::FetchQuoteReq {
                security_list,
                reserved: Some(0),
            };

            let reserved = futu_backend::stock_list::make_quote_reserved(
                match mkt {
                    1 => futu_backend::stock_list::QuoteMktType::HK,
                    2 => futu_backend::stock_list::QuoteMktType::US,
                    3 => futu_backend::stock_list::QuoteMktType::SH,
                    4 => futu_backend::stock_list::QuoteMktType::SZ,
                    5 => futu_backend::stock_list::QuoteMktType::HKFuture,
                    _ => futu_backend::stock_list::QuoteMktType::HK,
                },
                0, // ex_type = SECURITY
            );

            let body = prost::Message::encode_to_vec(&fetch_req);
            match backend
                .request_with_reserved(CMD_PULL_SUB_DATA, body, reserved)
                .await
            {
                Ok(resp) => {
                    let fetch_rsp: futu_backend::proto_internal::ft_cmd_stock_quote_fetch::FetchQuoteRsp =
                        match prost::Message::decode(resp.body.as_ref()) {
                            Ok(r) => r,
                            Err(e) => {
                                tracing::warn!(conn_id, mkt, error = %e, "CMD6824 decode failed");
                                continue;
                            }
                        };
                    for sq in &fetch_rsp.security_qta_list {
                        let stock_id = sq.security_id.unwrap_or(0);
                        // Find sec_type and is_us for this stock
                        let (sec_type, is_us) = indices
                            .iter()
                            .find_map(|&i| {
                                let e = &entries[i];
                                if e.stock_id == stock_id {
                                    Some((e.sec_type, e.qot_market == 11))
                                } else {
                                    None
                                }
                            })
                            .unwrap_or((0, false));
                        let sd = parse_security_quote(sq, sec_type, is_us);
                        stock_snapshots.insert(stock_id, sd);
                    }
                }
                Err(e) => {
                    tracing::warn!(conn_id, mkt, error = %e, "CMD6824 request failed");
                }
            }
        }

        // Build response snapshots in request order
        let mut snapshot_list = Vec::new();
        for entry in &entries {
            let sd = stock_snapshots.get(&entry.stock_id);
            let basic = futu_proto::qot_get_security_snapshot::SnapshotBasicData {
                security: entry.sec.clone(),
                r#type: entry.sec_type,
                is_suspend: sd.map(|s| s.suspend_flag).unwrap_or(false),
                list_time: if entry.sec_type != 8 && entry.sec_type != 10 {
                    entry.list_time.clone()
                } else {
                    String::new()
                },
                lot_size: entry.lot_size,
                price_spread: 0.0,
                update_time: String::new(),
                high_price: sd.map(|s| dp(s.high_price, 9)).unwrap_or(0.0),
                open_price: sd.map(|s| dp(s.open_price, 9)).unwrap_or(0.0),
                low_price: sd.map(|s| dp(s.low_price, 9)).unwrap_or(0.0),
                last_close_price: sd.map(|s| dp(s.last_close_price, 9)).unwrap_or(0.0),
                cur_price: sd.map(|s| dp(s.cur_price, 9)).unwrap_or(0.0),
                volume: sd.map(|s| s.volume).unwrap_or(0),
                turnover: sd.map(|s| dp(s.turnover, 3)).unwrap_or(0.0),
                turnover_rate: sd.map(|s| dp(s.turnover_ratio, 3)).unwrap_or(0.0),
                name: Some(entry.name.clone()),
                list_timestamp: None,
                update_timestamp: sd.and_then(|s| {
                    if s.timestamp > 0 {
                        Some(s.timestamp as f64)
                    } else {
                        None
                    }
                }),
                ask_price: sd.map(|s| dp(s.ask_price, 9)),
                bid_price: sd.map(|s| dp(s.bid_price, 9)),
                ask_vol: sd.map(|s| s.ask_vol),
                bid_vol: sd.map(|s| s.bid_vol),
                enable_margin: None,
                mortgage_ratio: None,
                long_margin_initial_ratio: None,
                enable_short_sell: None,
                short_sell_rate: None,
                short_available_volume: None,
                short_margin_initial_ratio: None,
                amplitude: sd.map(|s| dp(s.amplitude, 3)),
                avg_price: sd.map(|s| dp(s.avg_price, 9)),
                bid_ask_ratio: sd.map(|s| dp(s.bid_ask_ratio, 3)),
                volume_ratio: sd.map(|s| dp(s.volume_ratio, 3)),
                highest52_weeks_price: sd.map(|s| dp(s.week52_highest_price, 9)),
                lowest52_weeks_price: sd.map(|s| dp(s.week52_lowest_price, 9)),
                highest_history_price: sd.map(|s| dp(s.history_highest_price, 9)),
                lowest_history_price: sd.map(|s| dp(s.history_lowest_price, 9)),
                sec_status: sd.map(|s| s.sec_status),
                close_price5_minute: sd.map(|s| dp(s.close_price_5min, 9)),
                pre_market: sd.and_then(|s| s.pre_market.as_ref().map(build_pre_after_market)),
                after_market: {
                    // KCB stocks: after_market from DealStatistics; US stocks: from SBIT_13
                    if let Some(s) = sd {
                        if s.kcb_has_after {
                            Some(futu_proto::qot_common::PreAfterMarketData {
                                price: None,
                                high_price: None,
                                low_price: None,
                                volume: Some(s.kcb_after_volume),
                                turnover: Some(dp(s.kcb_after_turnover, 3)),
                                change_val: None,
                                change_rate: None,
                                amplitude: None,
                            })
                        } else {
                            s.after_market.as_ref().map(build_pre_after_market)
                        }
                    } else {
                        None
                    }
                },
                overnight: sd.and_then(|s| s.overnight.as_ref().map(build_pre_after_market)),
            };

            // Build ex_data based on sec_type
            let mut equity_ex = None;
            let mut warrant_ex = None;
            let mut option_ex = None;
            let mut index_ex = None;
            let mut plate_ex = None;
            let mut future_ex = None;
            let mut trust_ex = None;

            if let Some(s) = sd {
                match entry.sec_type {
                    3 => {
                        // Eqty
                        let cur_price_f = dp(s.cur_price, 9);
                        equity_ex = Some(
                            futu_proto::qot_get_security_snapshot::EquitySnapshotExData {
                                issued_shares: s.total_shares,
                                issued_market_val: dp(s.total_market_cap, 3),
                                net_asset: dp(s.net_asset, 3),
                                net_profit: dp(s.net_profit, 3),
                                earnings_pershare: dp(s.eps_lyr, 9),
                                outstanding_shares: s.outstanding_shares,
                                outstanding_market_val: s.outstanding_shares as f64 * cur_price_f,
                                net_asset_pershare: dp(s.net_asset_pershare, 9),
                                ey_rate: dp(s.ey_ratio as i64, 3),
                                pe_ttm_rate: dp(s.pe_ttm, 3),
                                pb_rate: dp(s.pb_ratio, 3),
                                pe_rate: dp(s.pe_lyr, 3),
                                dividend_ttm: Some(dp(s.dividend_ttm, 9)),
                                dividend_ratio_ttm: Some(dp(s.dividend_ratio_ttm, 2)), // C++: precision=2
                                dividend_lfy: Some(dp(s.dividend_lfy, 9)),
                                dividend_lfy_ratio: Some(dp(s.dividend_lfy_ratio, 3)), // C++: precision=3
                            },
                        );
                    }
                    5 => {
                        // Warrant
                        if let Some(StockSpecificData::Warrant(ref w)) = s.stock_specific {
                            warrant_ex = Some(
                                futu_proto::qot_get_security_snapshot::WarrantSnapshotExData {
                                    conversion_price: Some(dp(w.price_entitlement.unwrap_or(0), 9)),
                                    warrant_type: w.r#type.unwrap_or(0),
                                    strike_price: dp(w.price_strike.unwrap_or(0), 9),
                                    maturity_time: String::new(),
                                    end_trade_time: String::new(),
                                    owner: futu_proto::qot_common::Security {
                                        market: entry.qot_market,
                                        code: String::new(),
                                    },
                                    recovery_price: dp(w.price_call.unwrap_or(0), 9),
                                    street_volumn: w.volume_street.unwrap_or(0),
                                    issue_volumn: w.issued_shares.unwrap_or(0),
                                    street_rate: dp(w.ratio_street.unwrap_or(0), 5),
                                    delta: dp(w.delta.unwrap_or(0), 3),
                                    implied_volatility: dp(w.implied_volatility.unwrap_or(0), 3),
                                    premium: dp(w.premium.unwrap_or(0), 5),
                                    maturity_timestamp: w.expiry_date_time_s.map(|t| t as f64),
                                    end_trade_timestamp: w
                                        .last_trading_date_time_s
                                        .map(|t| t as f64),
                                    leverage: Some(dp(w.leverage.unwrap_or(0), 3)),
                                    ipop: Some(dp(w.ratio_itm_otm.unwrap_or(0), 5)),
                                    break_even_point: Some(dp(
                                        w.price_break_even_point.unwrap_or(0),
                                        9,
                                    )),
                                    conversion_rate: dp(w.ratio_entitlement.unwrap_or(0), 3),
                                    price_recovery_ratio: Some(dp(
                                        w.ratio_price_call.unwrap_or(0),
                                        5,
                                    )),
                                    score: Some(dp(w.score_faxing.unwrap_or(0) as i64, 0)),
                                    upper_strike_price: Some(dp(w.upper_price.unwrap_or(0), 9)),
                                    lower_strike_price: Some(dp(w.lower_price.unwrap_or(0), 9)),
                                    in_line_price_status: w.in_or_out.map(|v| {
                                        if v == 0 {
                                            1
                                        } else {
                                            2
                                        }
                                    }),
                                    issuer_code: None,
                                },
                            );
                        }
                    }
                    8 => {
                        // Option/Drvt
                        if let Some(StockSpecificData::Option(ref o)) = s.stock_specific {
                            let greek = o.greek.as_ref();
                            option_ex = Some(
                                futu_proto::qot_get_security_snapshot::OptionSnapshotExData {
                                    r#type: 0,
                                    owner: futu_proto::qot_common::Security {
                                        market: entry.qot_market,
                                        code: String::new(),
                                    },
                                    strike_time: String::new(),
                                    strike_price: dp(o.price_strike.unwrap_or(0), 9),
                                    contract_size: o.contract_size.unwrap_or(0) as i32,
                                    open_interest: o.open_interest.unwrap_or(0) as i32,
                                    implied_volatility: dp(o.implied_volatility.unwrap_or(0), 5),
                                    premium: dp(o.premium.unwrap_or(0), 5),
                                    delta: greek
                                        .map(|g| dp(g.hp_delta.unwrap_or(0), 9))
                                        .unwrap_or(0.0),
                                    gamma: greek
                                        .map(|g| dp(g.hp_gamma.unwrap_or(0), 9))
                                        .unwrap_or(0.0),
                                    vega: greek
                                        .map(|g| dp(g.hp_vega.unwrap_or(0), 9))
                                        .unwrap_or(0.0),
                                    theta: greek
                                        .map(|g| dp(g.hp_theta.unwrap_or(0), 9))
                                        .unwrap_or(0.0),
                                    rho: greek.map(|g| dp(g.hp_rho.unwrap_or(0), 9)).unwrap_or(0.0),
                                    strike_timestamp: None,
                                    index_option_type: None,
                                    net_open_interest: Some(o.open_interest_net.unwrap_or(0) as i32),
                                    expiry_date_distance: Some(
                                        o.distance_due_date.unwrap_or(0) as i32
                                    ),
                                    contract_nominal_value: Some(dp(
                                        o.contract_nominal_ammount.unwrap_or(0) as i64,
                                        9,
                                    )),
                                    owner_lot_multiplier: Some(
                                        o.positive_number_of_hand.unwrap_or(0) as f64,
                                    ),
                                    option_area_type: Some(o.option_type.unwrap_or(0) + 1),
                                    contract_multiplier: o.hp_multiplier.map(|m| dp(m as i64, 9)),
                                    contract_size_float: o
                                        .hp_contract_size
                                        .map(|c| dp(c as i64, 9)),
                                },
                            );
                        }
                    }
                    6 => {
                        // Index
                        if let Some(StockSpecificData::Index(ref i)) = s.stock_specific {
                            index_ex =
                                Some(futu_proto::qot_get_security_snapshot::IndexSnapshotExData {
                                    raise_count: i.raise_count.unwrap_or(0) as i32,
                                    fall_count: i.fall_count.unwrap_or(0) as i32,
                                    equal_count: i.equal_count.unwrap_or(0) as i32,
                                });
                        }
                    }
                    7 => {
                        // Plate
                        if let Some(StockSpecificData::Plate(ref p)) = s.stock_specific {
                            plate_ex =
                                Some(futu_proto::qot_get_security_snapshot::PlateSnapshotExData {
                                    raise_count: p.raise_count.unwrap_or(0) as i32,
                                    fall_count: p.fall_count.unwrap_or(0) as i32,
                                    equal_count: p.equal_count.unwrap_or(0) as i32,
                                });
                        }
                    }
                    10 => {
                        // Future
                        if let Some(StockSpecificData::Future(ref f)) = s.stock_specific {
                            future_ex = Some(
                                futu_proto::qot_get_security_snapshot::FutureSnapshotExData {
                                    last_settle_price: dp(f.last_settlement_price.unwrap_or(0), 9),
                                    position: f.hold_vol.unwrap_or(0) as i32,
                                    position_change: f.diff_hold_vol.unwrap_or(0) as i32,
                                    last_trade_time: String::new(),
                                    last_trade_timestamp: None,
                                    is_main_contract: f.main_flag.unwrap_or(0) == 1,
                                },
                            );
                        }
                    }
                    4 => {
                        // Trust
                        if let Some(StockSpecificData::Trust(ref t)) = s.stock_specific {
                            trust_ex =
                                Some(futu_proto::qot_get_security_snapshot::TrustSnapshotExData {
                                    dividend_yield: dp(t.dividend_yield.unwrap_or(0), 5),
                                    aum: t.aum.unwrap_or(0) as f64,
                                    outstanding_units: t.outstanding_units.unwrap_or(0) as i64,
                                    net_asset_value: dp(t.net_asset_value.unwrap_or(0) as i64, 9),
                                    premium: dp(t.premium.unwrap_or(0), 5),
                                    asset_class: t.asset_class.unwrap_or(0) as i32,
                                });
                        }
                    }
                    _ => {}
                }
            }

            snapshot_list.push(futu_proto::qot_get_security_snapshot::Snapshot {
                basic,
                equity_ex_data: equity_ex,
                warrant_ex_data: warrant_ex,
                option_ex_data: option_ex,
                index_ex_data: index_ex,
                plate_ex_data: plate_ex,
                future_ex_data: future_ex,
                trust_ex_data: trust_ex,
            });
        }

        tracing::debug!(
            conn_id,
            count = snapshot_list.len(),
            "GetSecuritySnapshot via CMD6824"
        );

        let resp = futu_proto::qot_get_security_snapshot::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_security_snapshot::S2c { snapshot_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetGlobalState =====
struct GetGlobalStateHandler {
    login_cache: Arc<LoginCache>,
    backend: crate::bridge::SharedBackend,
}

#[async_trait]
impl RequestHandler for GetGlobalStateHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let _req: futu_proto::get_global_state::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;

        let login_state = self.login_cache.get_login_state();
        let is_logged_in = login_state
            .as_ref()
            .map(|s| s.is_logged_in)
            .unwrap_or(false);
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs() as i64;

        // 从后端拉取真实市场状态（对应 C++ INNData_Qot_EventNotice::GetMarketStateInfo）
        use futu_backend::stock_list::{pull_single_market_status, QuoteMktType};
        let mut market_hk: i32 = 0; // QotMarketState_None
        let mut market_us: i32 = 0;
        let mut market_sh: i32 = 0;
        let mut market_sz: i32 = 0;
        let mut market_hk_future: i32 = 0;
        let mut market_us_future: Option<i32> = None;
        let mut market_sg_future: Option<i32> = None;
        let mut market_jp_future: Option<i32> = None;

        if let Some(backend) = super::load_backend(&self.backend) {
            let queries = [
                (QuoteMktType::HK, "HK"),
                (QuoteMktType::US, "US"),
                (QuoteMktType::SH, "SH"),
                (QuoteMktType::SZ, "SZ"),
                (QuoteMktType::HKFuture, "HKFuture"),
                (QuoteMktType::USFuture, "USFuture"),
                (QuoteMktType::SGFuture, "SGFuture"),
                (QuoteMktType::JPFuture, "JPFuture"),
            ];
            for (mkt, name) in &queries {
                match pull_single_market_status(&backend, *mkt).await {
                    Ok(statuses) if !statuses.is_empty() => {
                        let state = statuses[0].status as i32;
                        match *name {
                            "HK" => market_hk = state,
                            "US" => market_us = state,
                            "SH" => market_sh = state,
                            "SZ" => market_sz = state,
                            "HKFuture" => market_hk_future = state,
                            "USFuture" => market_us_future = Some(state),
                            "SGFuture" => market_sg_future = Some(state),
                            "JPFuture" => market_jp_future = Some(state),
                            _ => {}
                        }
                    }
                    Ok(_) => {
                        tracing::debug!(market = name, "no market status data");
                    }
                    Err(e) => {
                        tracing::debug!(market = name, error = %e, "market status query failed");
                    }
                }
            }
        }

        let resp = futu_proto::get_global_state::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::get_global_state::S2c {
                market_hk,
                market_us,
                market_sh,
                market_sz,
                market_hk_future,
                market_us_future,
                market_sg_future,
                market_jp_future,
                qot_logined: is_logged_in,
                trd_logined: is_logged_in,
                server_ver: 1002,
                server_build_no: 1,
                time: now,
                local_time: Some(now as f64),
                program_status: None,
                qot_svr_ip_addr: login_state.as_ref().map(|s| s.server_addr.clone()),
                trd_svr_ip_addr: login_state.as_ref().map(|s| s.server_addr.clone()),
                conn_id: Some(conn_id),
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== RequestHistoryKL =====
struct RequestHistoryKLHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
    kl_quota_counter: Arc<std::sync::atomic::AtomicU32>,
}

/// FTAPI KLType → backend KlineType 映射
/// 两者的枚举值完全一致，直接传递
fn ftapi_kl_type_to_backend(kl_type: i32) -> Option<u32> {
    match kl_type {
        1 => Some(1),   // 1Min
        2 => Some(2),   // Day
        3 => Some(3),   // Week
        4 => Some(4),   // Month
        5 => Some(5),   // Year
        6 => Some(6),   // 5Min
        7 => Some(7),   // 15Min
        8 => Some(8),   // 30Min
        9 => Some(9),   // 60Min
        10 => Some(10), // 3Min
        11 => Some(11), // Quarter
        _ => None,
    }
}

/// 将日期字符串转换为 UTC 时间戳，考虑市场时区。
/// C++ 使用 `APITimeStrToTimeStamp_Qot` + `GetTimeZoneByAPIQotMkt`。
/// 例如 "2024-01-02" 在 HK 市场 = 2024-01-02 00:00:00 +08:00 = 2024-01-01T16:00:00Z
fn date_str_to_timestamp(s: &str, market: i32) -> Option<u64> {
    // Support both "yyyy-MM-dd" and "yyyy-MM-dd HH:MM:SS"
    let date_part = s.split(' ').next().unwrap_or(s);
    let parts: Vec<&str> = date_part.split('-').collect();
    if parts.len() != 3 {
        return None;
    }
    let year: i32 = parts[0].parse().ok()?;
    let month: u32 = parts[1].parse().ok()?;
    let day: u32 = parts[2].parse().ok()?;

    if !(1..=12).contains(&month) || !(1..=31).contains(&day) {
        return None;
    }

    let mut total_days: i64 = 0;
    for y in 1970..year {
        total_days += if is_leap_year(y) { 366 } else { 365 };
    }
    let month_days = [
        31,
        if is_leap_year(year) { 29 } else { 28 },
        31,
        30,
        31,
        30,
        31,
        31,
        30,
        31,
        30,
        31,
    ];
    for &md in month_days.iter().take(month as usize - 1) {
        total_days += md as i64;
    }
    total_days += (day as i64) - 1;

    if total_days < 0 {
        return None;
    }
    let utc_secs = total_days * 86400;

    // 根据市场获取时区偏移（C++: GetTimeZoneByAPIQotMkt）
    let tz_offset_hours: i64 = market_timezone_offset(market);
    let ts = utc_secs - tz_offset_hours * 3600;
    if ts < 0 {
        return None;
    }
    Some(ts as u64)
}

/// 获取市场时区偏移（小时）。C++ GetTimeZoneByAPIQotMkt。
fn market_timezone_offset(market: i32) -> i64 {
    match market {
        // QotMarket_HK_Security = 1, QotMarket_HK_Future = 2
        1 | 2 => 8,
        // QotMarket_US_Security = 11
        11 => -5,
        // QotMarket_CNSH_Security = 21, QotMarket_CNSZ_Security = 22
        21 | 22 => 8,
        // QotMarket_SG_Security = 31
        31 => 8,
        // QotMarket_JP_Security = 41
        41 => 9,
        // QotMarket_AU_Security = 51
        51 => 10,
        // 默认东八区
        _ => 8,
    }
}

fn is_leap_year(y: i32) -> bool {
    (y % 4 == 0 && y % 100 != 0) || y % 400 == 0
}

/// Unix 时间戳 → "yyyy-MM-dd HH:mm:ss" 字符串（东八区）
fn timestamp_to_datetime_str(ts: u64) -> String {
    // 所有行情时间戳需要转换为东八区（HK/SH/SZ 市场所在时区）
    let (year, month, day, hour, minute, second) = timestamp_to_components(ts + 8 * 3600);
    format!(
        "{:04}-{:02}-{:02} {:02}:{:02}:{:02}",
        year, month, day, hour, minute, second
    )
}

fn timestamp_to_date_str(ts: u64) -> String {
    let (year, month, day, _, _, _) = timestamp_to_components(ts + 8 * 3600);
    format!("{:04}-{:02}-{:02}", year, month, day)
}

/// 解析 "YYYY-MM-DD" 日期字符串为 Unix 时间戳（东八区 0:00:00）
fn parse_date_to_timestamp(s: &str) -> Option<i64> {
    let parts: Vec<&str> = s.split('-').collect();
    if parts.len() != 3 {
        return None;
    }
    let y: i64 = parts[0].parse().ok()?;
    let m: i64 = parts[1].parse().ok()?;
    let d: i64 = parts[2].parse().ok()?;
    if !(1..=12).contains(&m) || !(1..=31).contains(&d) {
        return None;
    }
    // 简化的日期→天数算法 (Rata Die)
    let (y2, m2) = if m <= 2 { (y - 1, m + 9) } else { (y, m - 3) };
    let days = 365 * y2 + y2 / 4 - y2 / 100 + y2 / 400 + (m2 * 306 + 5) / 10 + d - 1 - 719468;
    Some(days * 86400 - 8 * 3600) // 东八区 0:00:00 → UTC
}

/// 后端 ExpirationType → FTAPI ExpirationCycle 映射
/// C++ ExpirationCycle_NNToAPI
fn backend_expiration_to_api(backend_type: u32) -> i32 {
    match backend_type {
        0 => 2,   // Month → ExpirationCycle_Month
        1 => 1,   // Week → ExpirationCycle_Week
        2 => 3,   // MonthEnd → ExpirationCycle_MonthEnd
        3 => 4,   // Quarter → ExpirationCycle_Quarter
        11 => 11, // Week_Mon → ExpirationCycle_WeekMon
        12 => 12, // Week_Tue → ExpirationCycle_WeekTue
        13 => 13, // Week_Wed → ExpirationCycle_WeekWed
        14 => 14, // Week_Thu → ExpirationCycle_WeekThu
        15 => 15, // Week_Fri → ExpirationCycle_WeekFri
        _ => 0,   // Unknown
    }
}

/// Unix 时间戳（UTC）→ (year, month, day, hour, minute, second)
fn timestamp_to_components(ts: u64) -> (i32, u32, u32, u32, u32, u32) {
    let secs = ts as i64;
    let second = ((secs % 60) + 60) as u32 % 60;
    let minute = (((secs / 60) % 60) + 60) as u32 % 60;
    let hour = (((secs / 3600) % 24) + 24) as u32 % 24;

    let mut days = secs / 86400;
    let mut year = 1970_i32;

    loop {
        let days_in_year: i64 = if is_leap_year(year) { 366 } else { 365 };
        if days < days_in_year {
            break;
        }
        days -= days_in_year;
        year += 1;
    }

    let month_days = [
        31,
        if is_leap_year(year) { 29 } else { 28 },
        31,
        30,
        31,
        30,
        31,
        31,
        30,
        31,
        30,
        31,
    ];
    let mut month = 1_u32;
    for &md in &month_days {
        if days < md as i64 {
            break;
        }
        days -= md as i64;
        month += 1;
    }
    let day = days as u32 + 1;

    (year, month, day, hour, minute, second)
}

/// 将 backend KlineItem 转换为 FTAPI KLine
fn kline_item_to_ftapi(
    item: &futu_backend::proto_internal::ft_cmd_kline::KlineItem,
) -> futu_proto::qot_common::KLine {
    let close_raw = item.close_price.unwrap_or(0);
    let open_raw = item.open_price.unwrap_or(0);
    let high_raw = item.highest_price.unwrap_or(0);
    let low_raw = item.lowest_price.unwrap_or(0);
    let last_close_raw = item.last_close_price.unwrap_or(0);

    let divisor = 1_000_000_000.0_f64;

    let close_price = close_raw as f64 / divisor;
    let open_price = open_raw as f64 / divisor;
    let high_price = high_raw as f64 / divisor;
    let low_price = low_raw as f64 / divisor;
    let last_close_price = last_close_raw as f64 / divisor;

    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 volume = item.volume.unwrap_or(0) as i64;
    let timestamp = item.time.unwrap_or(0) as f64;

    // change_rate = (close - last_close) / last_close * 100
    let change_rate = if last_close_raw != 0 {
        (close_raw - last_close_raw) as f64 / last_close_raw as f64 * 100.0
    } else {
        0.0
    };

    // C++ 用 has_close_price() 判断是否空白K线，而非检查值是否为0
    let is_blank = item.close_price.is_none();

    // Format time as string from timestamp (seconds), using the timezone from the item
    let ts = item.time.unwrap_or(0);
    let tz_offset = item.time_zone.unwrap_or(0) as i64;
    let time_str = format_timestamp_with_tz(ts, tz_offset);

    if is_blank {
        // C++ KL_NNToAPI: blank entry only has time/is_blank/timestamp
        return futu_proto::qot_common::KLine {
            time: time_str,
            is_blank: true,
            high_price: None,
            open_price: None,
            low_price: None,
            close_price: None,
            last_close_price: None,
            volume: None,
            turnover: None,
            turnover_rate: None,
            pe: None,
            change_rate: None,
            timestamp: Some(timestamp),
        };
    }

    futu_proto::qot_common::KLine {
        time: time_str,
        is_blank: false,
        high_price: Some(high_price),
        open_price: Some(open_price),
        low_price: Some(low_price),
        close_price: Some(close_price),
        last_close_price: Some(last_close_price),
        volume: Some(volume),
        turnover: Some(turnover),
        turnover_rate: Some(turnover_rate),
        pe: Some(pe),
        change_rate: Some(change_rate),
        timestamp: Some(timestamp),
    }
}

/// 格式化 Unix 时间戳为 "yyyy-MM-dd HH:mm:ss"，使用指定时区偏移（小时）
fn format_timestamp_with_tz(ts: u64, tz_hours: i64) -> String {
    let local_ts = ts as i64 + tz_hours * 3600;
    format_timestamp(local_ts as u64)
}

/// 格式化 Unix 时间戳为 "yyyy-MM-dd HH:mm:ss" (UTC)
fn format_timestamp(ts: u64) -> String {
    let secs = ts as i64;
    let days = secs.div_euclid(86400);
    let day_secs = secs.rem_euclid(86400);
    let hours = day_secs / 3600;
    let minutes = (day_secs % 3600) / 60;
    let seconds = day_secs % 60;

    // Convert days since epoch to date
    let mut remaining_days = days;
    let mut year = 1970_i32;

    loop {
        let days_in_year: i64 = if is_leap_year(year) { 366 } else { 365 };
        if remaining_days < days_in_year {
            break;
        }
        remaining_days -= days_in_year;
        year += 1;
    }

    let month_days = [
        31,
        if is_leap_year(year) { 29 } else { 28 },
        31,
        30,
        31,
        30,
        31,
        31,
        30,
        31,
        30,
        31,
    ];
    let mut month = 0_usize;
    while month < 12 && remaining_days >= month_days[month] as i64 {
        remaining_days -= month_days[month] as i64;
        month += 1;
    }
    let day = remaining_days + 1;

    format!(
        "{:04}-{:02}-{:02} {:02}:{:02}:{:02}",
        year,
        month + 1,
        day,
        hours,
        minutes,
        seconds
    )
}

#[async_trait]
impl RequestHandler for RequestHistoryKLHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_request_history_kl::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "RequestHistoryKL: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Look up stock_id from static data cache
        let sec_key = format!("{}_{}", c2s.security.market, c2s.security.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(conn_id, sec_key, "RequestHistoryKL: stock_id not found");
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        // Convert FTAPI KLType → backend KlineType
        let backend_kl_type = match ftapi_kl_type_to_backend(c2s.kl_type) {
            Some(v) => v,
            None => {
                tracing::warn!(
                    conn_id,
                    kl_type = c2s.kl_type,
                    "RequestHistoryKL: invalid kl_type"
                );
                return Some(super::make_error_response(-1, "invalid kl_type"));
            }
        };

        // RehabType → ExrightType (same values: 0=None, 1=Forward, 2=Backward)
        let exright_type = c2s.rehab_type as u32;

        // Convert begin_time / end_time from date string to timestamp (with market timezone)
        let qot_market = c2s.security.market;
        let begin_ts = date_str_to_timestamp(&c2s.begin_time, qot_market).unwrap_or(0);
        // end_time: use u64::MAX-like sentinel if empty, or parse
        let end_ts = if c2s.end_time.is_empty() {
            u64::MAX
        } else {
            // Add 86400-1 to include the end day fully
            date_str_to_timestamp(&c2s.end_time, qot_market)
                .map(|t| t + 86399)
                .unwrap_or(u64::MAX)
        };

        // C++ always uses data_range_type=1 (BEGIN_TIME + END_TIME)
        // max_ack_kl_num is applied locally after receiving data
        let max_kl_num = c2s.max_ack_kl_num.unwrap_or(0) as usize;

        // Build KlineReq
        let kline_req = futu_backend::proto_internal::ft_cmd_kline::KlineReq {
            security_id: Some(stock_id),
            kline_type: Some(backend_kl_type),
            exright_type: Some(exright_type),
            data_set_type: Some(0),   // DATA_SET_TYPE_KLINE
            data_range_type: Some(1), // always BEGIN_TIME + END_TIME
            begin_time: Some(begin_ts),
            end_time: Some(end_ts),
            item_count: None,
            end_time_offset: None,
        };

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

        tracing::debug!(
            conn_id,
            stock_id,
            kl_type = backend_kl_type,
            exright = exright_type,
            body_len = body.len(),
            "sending CMD6161 KlineReq"
        );

        // Send CMD 6161 (unencrypted)
        let resp_frame = match backend.request(6161, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6161 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        // Parse KlineRsp (raw protobuf, no OMBinSrz prefix)
        let kline_rsp: futu_backend::proto_internal::ft_cmd_kline::KlineRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, body_len = resp_frame.body.len(), "CMD6161 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        let result = kline_rsp.result.unwrap_or(-1);
        if result != 0 {
            tracing::warn!(conn_id, result, "CMD6161 returned error");
            return Some(super::make_error_response(
                -1,
                "backend kline request failed",
            ));
        }

        // Convert KlineItem list → FTAPI KLine list
        // Apply max_ack_kl_num locally (C++ SetMaxPointCount + MergeAllChildItem)
        let mut kl_list: Vec<futu_proto::qot_common::KLine> = kline_rsp
            .kline_item_list
            .iter()
            .map(kline_item_to_ftapi)
            .collect();
        if max_kl_num > 0 && kl_list.len() > max_kl_num {
            kl_list.truncate(max_kl_num);
        }

        tracing::debug!(
            conn_id,
            count = kl_list.len(),
            "RequestHistoryKL returning klines"
        );

        // 获取股票名称 (C++: GetStockName)
        let stock_name = self
            .static_cache
            .get_security_info(&sec_key)
            .map(|info| info.name.clone());

        // 递增 KL 请求计数（用于 GetUsedQuota）
        self.kl_quota_counter
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);

        let resp = futu_proto::qot_request_history_kl::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_request_history_kl::S2c {
                security: c2s.security.clone(),
                name: stock_name,
                kl_list,
                next_req_key: None,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== RequestTradeDate (CMD 6733) =====
struct RequestTradeDateHandler {
    backend: crate::bridge::SharedBackend,
}

/// FTAPI QotMarket → backend market_id
fn ftapi_market_to_backend(market: i32) -> Option<u32> {
    match market {
        1 => Some(1),   // HK
        11 => Some(10), // US
        21 => Some(30), // CNSH
        22 => Some(31), // CNSZ
        _ => None,
    }
}

/// Parse "yyyy-MM-dd" → YYYYMMDD u32 (e.g. "2024-01-15" → 20240115)
fn date_str_to_yyyymmdd(s: &str) -> Option<u32> {
    // 兼容 "yyyy-MM-dd" 和 "yyyy-MM-dd HH:mm:ss" 格式
    let date_part = s.split(' ').next().unwrap_or(s);
    let parts: Vec<&str> = date_part.split('-').collect();
    if parts.len() != 3 {
        return None;
    }
    let year: u32 = parts[0].parse().ok()?;
    let month: u32 = parts[1].parse().ok()?;
    let day: u32 = parts[2].parse().ok()?;
    if !(1..=12).contains(&month) || !(1..=31).contains(&day) {
        return None;
    }
    Some(year * 10000 + month * 100 + day)
}

/// YYYYMMDD u32 → "yyyy-MM-dd" string
fn yyyymmdd_to_date_str(v: u32) -> String {
    let year = v / 10000;
    let month = (v % 10000) / 100;
    let day = v % 100;
    format!("{:04}-{:02}-{:02}", year, month, day)
}

/// Backend DayInfo.trading_type → FTAPI TradeDateType
/// NOT_CLOSE(4)→0, CLOSE_WHOLE(0)→1, CLOSE_AM(1)→2, CLOSE_PM(2)→3
fn backend_trading_type_to_ftapi(trading_type: u32) -> i32 {
    match trading_type {
        4 => 0, // NOT_CLOSE → 全天交易
        0 => 1, // CLOSE_WHOLE → 全天休市
        1 => 2, // CLOSE_AM → 上午休市
        2 => 3, // CLOSE_PM → 下午休市
        _ => 0, // default to 全天交易
    }
}

#[async_trait]
impl RequestHandler for RequestTradeDateHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_request_trade_date::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "RequestTradeDate: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        let backend_market = match ftapi_market_to_backend(c2s.market) {
            Some(m) => m,
            None => {
                tracing::warn!(
                    conn_id,
                    market = c2s.market,
                    "RequestTradeDate: unsupported market"
                );
                return Some(super::make_error_response(-1, "unsupported market"));
            }
        };

        let begin_key = match date_str_to_yyyymmdd(&c2s.begin_time) {
            Some(v) => v,
            None => {
                tracing::warn!(conn_id, begin = %c2s.begin_time, "RequestTradeDate: invalid begin_time");
                return Some(super::make_error_response(-1, "invalid begin_time"));
            }
        };

        let end_key = match date_str_to_yyyymmdd(&c2s.end_time) {
            Some(v) => v,
            None => {
                tracing::warn!(conn_id, end = %c2s.end_time, "RequestTradeDate: invalid end_time");
                return Some(super::make_error_response(-1, "invalid end_time"));
            }
        };

        let range_req = futu_backend::proto_internal::market_trading_day::RangeTradingDayReq {
            begin_date_key: Some(begin_key),
            end_date_key: Some(end_key),
            market_id: Some(backend_market),
        };

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

        tracing::debug!(
            conn_id,
            backend_market,
            begin_key,
            end_key,
            "sending CMD6733 RangeTradingDayReq"
        );

        let resp_frame = match backend.request(6733, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6733 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let range_rsp: futu_backend::proto_internal::market_trading_day::RangeTradingDayRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6733 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        let code = range_rsp.code.unwrap_or(-1);
        if code != 0 {
            tracing::warn!(conn_id, code, "CMD6733 returned error");
            return Some(super::make_error_response(
                -1,
                "backend trade date request failed",
            ));
        }

        // Convert DayInfo list → FTAPI TradeDate list
        let trade_date_list: Vec<futu_proto::qot_request_trade_date::TradeDate> = range_rsp
            .day_infos
            .iter()
            .map(|day| {
                let time_date = day.time_date.unwrap_or(0);
                let time_str = yyyymmdd_to_date_str(time_date);
                let trading_type = day.trading_type.unwrap_or(4); // default NOT_CLOSE
                let trade_date_type = backend_trading_type_to_ftapi(trading_type);
                // timestamp from date_key (seconds since epoch)
                let ts = day.date_key.unwrap_or(0) as f64;

                futu_proto::qot_request_trade_date::TradeDate {
                    time: time_str,
                    timestamp: Some(ts),
                    trade_date_type: Some(trade_date_type),
                }
            })
            .collect();

        tracing::debug!(
            conn_id,
            count = trade_date_list.len(),
            "RequestTradeDate returning trade dates"
        );

        let resp = futu_proto::qot_request_trade_date::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_request_trade_date::S2c { trade_date_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetRehab (CMD 6811) =====
struct GetRehabHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetRehabHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_request_rehab::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetRehab: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Look up stock_id from static data cache
        let sec_key = format!("{}_{}", c2s.security.market, c2s.security.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(conn_id, sec_key, "GetRehab: stock_id not found");
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        let exright_req = futu_backend::proto_internal::ft_cmd6811::ExrightFactorReq {
            stock_id: Some(stock_id),
            sequence: Some(0),
        };

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

        tracing::debug!(conn_id, stock_id, "sending CMD6811 ExrightFactorReq");

        let resp_frame = match backend.request(6811, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6811 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let exright_rsp: futu_backend::proto_internal::ft_cmd6811::ExrightFactorRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6811 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        let result = exright_rsp.result.unwrap_or(-1);
        // result 0 = has updates, 1 = no updates needed; both are success
        if result != 0 && result != 1 {
            tracing::warn!(conn_id, result, "CMD6811 returned error");
            return Some(super::make_error_response(
                -1,
                "backend rehab request failed",
            ));
        }

        // Convert ExrightFactorItem → FTAPI Rehab
        let rehab_list: Vec<futu_proto::qot_common::Rehab> = exright_rsp
            .exs
            .iter()
            .map(|item| {
                let ex_ts = item.ex_time.unwrap_or(0);
                let time_str = timestamp_to_date_str(ex_ts as u64);

                // companyActFlag: bitmask of which optional fields are present
                let mut flag: i64 = 0;
                if item.split_base.is_some() {
                    flag |= 1;
                }
                if item.join_base.is_some() {
                    flag |= 2;
                }
                if item.bonus_base.is_some() {
                    flag |= 4;
                }
                if item.transfer_base.is_some() {
                    flag |= 8;
                }
                if item.allot_base.is_some() {
                    flag |= 16;
                }
                if item.add_base.is_some() {
                    flag |= 32;
                }
                if item.dividend_amount.is_some() {
                    flag |= 64;
                }
                if item.sp_dividend_amount.is_some() {
                    flag |= 128;
                }

                let divisor = 100_000.0;

                futu_proto::qot_common::Rehab {
                    time: time_str,
                    company_act_flag: flag,
                    fwd_factor_a: item.origin_fwd_a.unwrap_or(100_000.0) / divisor,
                    fwd_factor_b: item.origin_fwd_b.unwrap_or(0.0) / divisor,
                    bwd_factor_a: item.origin_bwd_a.unwrap_or(100_000.0) / divisor,
                    bwd_factor_b: item.origin_bwd_b.unwrap_or(0.0) / divisor,
                    split_base: item.split_base.map(|v| v as i32),
                    split_ert: item.split_ert.map(|v| v as i32),
                    join_base: item.join_base.map(|v| v as i32),
                    join_ert: item.join_ert.map(|v| v as i32),
                    bonus_base: item.bonus_base.map(|v| v as i32),
                    bonus_ert: item.bonus_ert.map(|v| v as i32),
                    transfer_base: item.transfer_base.map(|v| v as i32),
                    transfer_ert: item.transfer_ert.map(|v| v as i32),
                    allot_base: item.allot_base.map(|v| v as i32),
                    allot_ert: item.allot_ert.map(|v| v as i32),
                    allot_price: item.allot_price.map(|v| v as f64 / divisor),
                    add_base: item.add_base.map(|v| v as i32),
                    add_ert: item.add_ert.map(|v| v as i32),
                    add_price: item.add_price.map(|v| v as f64 / divisor),
                    dividend: item.dividend_amount.map(|v| v as f64 / divisor),
                    sp_dividend: item.sp_dividend_amount.map(|v| v as f64 / divisor),
                    timestamp: Some(ex_ts as f64),
                    spin_off_base: item.spin_off_base.map(|v| v as f64),
                    spin_off_ert: item.spin_off_ert.map(|v| v as f64),
                }
            })
            .collect();

        tracing::debug!(
            conn_id,
            count = rehab_list.len(),
            "GetRehab returning rehab items"
        );

        let resp = futu_proto::qot_request_rehab::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_request_rehab::S2c { rehab_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetPlateSet (CMD 6600) =====
struct GetPlateSetHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

/// Map FTAPI market + plateSetType → backend plate_set_id
fn map_plate_set_id(market: i32, plate_set_type: i32) -> Option<u64> {
    match (market, plate_set_type) {
        // HK
        (1, 0) => Some(9700009), // All
        (1, 1) => Some(9700000), // Industry
        (1, 2) => Some(9700008), // Region → Other
        (1, 3) => Some(9700001), // Concept
        (1, 4) => Some(9700008), // Other
        // US
        (11, 0) => Some(9700309), // All
        (11, 1) => Some(9700300), // Industry
        (11, 2) => Some(9700308), // Region → Other
        (11, 3) => Some(9700301), // Concept
        (11, 4) => Some(9700308), // Other
        // CN (Shanghai + Shenzhen share same plate sets)
        (21 | 22, 0) => Some(9700609), // All
        (21 | 22, 1) => Some(9700600), // Industry
        (21 | 22, 2) => Some(9700602), // Region
        (21 | 22, 3) => Some(9700601), // Concept
        (21 | 22, 4) => Some(9700608), // Other
        _ => None,
    }
}

#[async_trait]
impl RequestHandler for GetPlateSetHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_plate_set::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetPlateSet: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        let plate_set_id = match map_plate_set_id(c2s.market, c2s.plate_set_type) {
            Some(id) => id,
            None => {
                tracing::warn!(
                    conn_id,
                    market = c2s.market,
                    plate_set_type = c2s.plate_set_type,
                    "GetPlateSet: unsupported market/plateSetType"
                );
                return Some(super::make_error_response(
                    -1,
                    "unsupported market or plateSetType",
                ));
            }
        };

        // Send CMD 6600 PlateListIDs_Req to get plate IDs in this set
        // C++ uses PullPlateOrSetID(nPlateSetID, SortMode_Code=100, SortDir_Asc=1, 0, 99999)
        let plate_req = futu_backend::proto_internal::ft_cmd_plate::PlateListIDsReq {
            plate_id: plate_set_id,
            sort_type: 1, // ascending
            sort_id: 100, // sort by code
            data_from: Some(0),
            data_max_count: Some(99999),
            check_code: Some(0),
        };
        let body = prost::Message::encode_to_vec(&plate_req);

        tracing::debug!(
            conn_id,
            plate_set_id,
            "sending CMD6600 PlateListIDsReq for plate set"
        );

        let resp_frame = match backend.request(6600, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6600 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let plate_rsp: futu_backend::proto_internal::ft_cmd_plate::PlateListIDsRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6600 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if plate_rsp.result != 0 {
            tracing::warn!(
                conn_id,
                result = plate_rsp.result,
                "CMD6600 returned error for plate set"
            );
            return Some(super::make_error_response(
                -1,
                "backend plate set request failed",
            ));
        }

        // Convert plate IDs to FTAPI PlateInfo
        // C++ does: SearchSecByID(plateID) → szCode, szNameZhCn
        // We use id_to_key → get_security_info to do the same
        let plate_info_list: Vec<futu_proto::qot_common::PlateInfo> = plate_rsp
            .arry_items
            .iter()
            .filter_map(|&pid| {
                // Look up plate code and name from static cache via stock_id
                let key = self.static_cache.id_to_key.get(&pid);
                if key.is_none() {
                    tracing::warn!(
                        conn_id,
                        plate_id = pid,
                        "plate_id not found in id_to_key cache"
                    );
                    return None;
                }
                let key = key.unwrap();
                let info = self.static_cache.get_security_info(&key)?;

                Some(futu_proto::qot_common::PlateInfo {
                    plate: futu_proto::qot_common::Security {
                        market: info.market,
                        code: info.code.clone(),
                    },
                    name: info.name.clone(),
                    plate_type: None,
                })
            })
            .collect();

        tracing::debug!(
            conn_id,
            count = plate_info_list.len(),
            "GetPlateSet returning plates"
        );

        let resp = futu_proto::qot_get_plate_set::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_plate_set::S2c { plate_info_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetPlateSecurity (CMD 6600) =====
struct GetPlateSecurityHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetPlateSecurityHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_plate_security::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetPlateSecurity: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // plate.code is a symbolic code like "LIST1001" / "GangGuTong" / "MotherBoard",
        // NOT a numeric plate_id. 对齐 C++ OpenD (APIServer_Inner_API.cpp::GetStockIDInternal):
        // 查 StaticDataCache 从 (market, code) → stock_id（板块在缓存里是以 security 形式存在）。
        let key = futu_cache::qot_cache::make_key(c2s.plate.market, &c2s.plate.code);
        let plate_id: u64 = match self.static_cache.get_security_info(&key) {
            Some(info) => info.stock_id,
            None => {
                tracing::warn!(
                    conn_id,
                    market = c2s.plate.market,
                    code = %c2s.plate.code,
                    "GetPlateSecurity: plate not found in static cache"
                );
                return Some(super::make_error_response(-1, "invalid plate code"));
            }
        };

        // Map sort field: default ascending by code (sort_type=0, sort_id=0)
        let sort_type = if c2s.ascend.unwrap_or(true) { 0 } else { 1 };
        let sort_id = c2s.sort_field.unwrap_or(0);

        // Send CMD 6600 PlateListIDsReq to get stock IDs in this plate
        let plate_req = futu_backend::proto_internal::ft_cmd_plate::PlateListIDsReq {
            plate_id,
            sort_type,
            sort_id,
            data_from: None,
            data_max_count: None,
            check_code: None,
        };
        let body = prost::Message::encode_to_vec(&plate_req);

        tracing::debug!(conn_id, plate_id, "sending CMD6600 PlateListIDsReq");

        let resp_frame = match backend.request(6600, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6600 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let plate_rsp: futu_backend::proto_internal::ft_cmd_plate::PlateListIDsRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6600 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if plate_rsp.result != 0 {
            tracing::warn!(conn_id, result = plate_rsp.result, "CMD6600 returned error");
            return Some(super::make_error_response(
                -1,
                "backend plate request failed",
            ));
        }

        // Look up each stock_id in StaticDataCache to build SecurityStaticInfo
        let static_info_list: Vec<futu_proto::qot_common::SecurityStaticInfo> = plate_rsp
            .arry_items
            .iter()
            .filter_map(|&stock_id| {
                let key = self.static_cache.id_to_key.get(&stock_id)?;
                let info = self.static_cache.get_security_info(key.value())?;
                Some(futu_proto::qot_common::SecurityStaticInfo {
                    basic: futu_proto::qot_common::SecurityStaticBasic {
                        security: futu_proto::qot_common::Security {
                            market: info.market,
                            code: info.code.clone(),
                        },
                        id: info.stock_id as i64,
                        lot_size: info.lot_size,
                        sec_type: info.sec_type,
                        name: info.name.clone(),
                        list_time: info.list_time.clone(),
                        delisting: None,
                        list_timestamp: None,
                        exch_type: None,
                    },
                    warrant_ex_data: None,
                    option_ex_data: None,
                    future_ex_data: None,
                })
            })
            .collect();

        tracing::debug!(
            conn_id,
            count = static_info_list.len(),
            "GetPlateSecurity returning securities"
        );

        let resp = futu_proto::qot_get_plate_security::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_plate_security::S2c { static_info_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetOwnerPlate (CMD 6608) =====
struct GetOwnerPlateHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetOwnerPlateHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_owner_plate::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetOwnerPlate: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Look up stock_ids from cache
        let mut stock_ids: Vec<u64> = Vec::new();
        let mut sec_info_map: Vec<(futu_proto::qot_common::Security, u64)> = Vec::new();

        for sec in &c2s.security_list {
            let sec_key = format!("{}_{}", sec.market, sec.code);
            if let Some(info) = self.static_cache.get_security_info(&sec_key) {
                if info.stock_id > 0 {
                    stock_ids.push(info.stock_id);
                    sec_info_map.push((sec.clone(), info.stock_id));
                }
            }
        }

        if stock_ids.is_empty() {
            // Return empty result
            let resp = futu_proto::qot_get_owner_plate::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::qot_get_owner_plate::S2c {
                    owner_plate_list: vec![],
                }),
            };
            return Some(prost::Message::encode_to_vec(&resp));
        }

        // Send CMD 6608 OwnerPlateReq
        let owner_req = futu_backend::proto_internal::ft_cmd_plate::OwnerPlateReq {
            stock_id_list: stock_ids,
            owner_plate_type: 0, // OPT_ALL
            need_plate_quote: None,
        };
        let body = prost::Message::encode_to_vec(&owner_req);

        tracing::debug!(
            conn_id,
            count = sec_info_map.len(),
            "sending CMD6608 OwnerPlateReq"
        );

        let resp_frame = match backend.request(6608, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6608 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let owner_rsp: futu_backend::proto_internal::ft_cmd_plate::OwnerPlateRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6608 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if owner_rsp.result != 0 {
            tracing::warn!(conn_id, result = owner_rsp.result, "CMD6608 returned error");
            return Some(super::make_error_response(
                -1,
                "backend owner plate request failed",
            ));
        }

        // Convert backend response to FTAPI format
        let owner_plate_list: Vec<futu_proto::qot_get_owner_plate::SecurityOwnerPlate> = owner_rsp
            .stock_info_list
            .iter()
            .filter_map(|stock_info| {
                // Find the original security for this stock_id
                let (sec, _) = sec_info_map
                    .iter()
                    .find(|(_, sid)| *sid == stock_info.stock_id)?;
                let sec_key = format!("{}_{}", sec.market, sec.code);
                let cached_info = self.static_cache.get_security_info(&sec_key);

                let plate_info_list: Vec<futu_proto::qot_common::PlateInfo> = stock_info
                    .plate_info_list
                    .iter()
                    .filter_map(|pi| {
                        // Look up plate code and name from static cache via stock_id
                        let key = self.static_cache.id_to_key.get(&pi.plate_id)?;
                        let plate_info = self.static_cache.get_security_info(&key)?;

                        Some(futu_proto::qot_common::PlateInfo {
                            plate: futu_proto::qot_common::Security {
                                market: plate_info.market,
                                code: plate_info.code.clone(),
                            },
                            name: plate_info.name.clone(),
                            plate_type: pi.owner_set_id.map(|set_id| {
                                // C++ GetPlateSetType: map owner_set_id → PlateSetType
                                // Trade → Industry(1), Region(2), Concept(3), else → Other(4)
                                match set_id {
                                    9700000 | 9700300 | 9700600 => 1, // Industry
                                    9700602 => 2,                     // Region (CN only)
                                    9700001 | 9700301 | 9700601 => 3, // Concept
                                    _ => 4,                           // Other (includes All)
                                }
                            }),
                        })
                    })
                    .collect();

                Some(futu_proto::qot_get_owner_plate::SecurityOwnerPlate {
                    security: sec.clone(),
                    name: cached_info.map(|i| i.name),
                    plate_info_list,
                })
            })
            .collect();

        tracing::debug!(
            conn_id,
            count = owner_plate_list.len(),
            "GetOwnerPlate returning results"
        );

        let resp = futu_proto::qot_get_owner_plate::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_owner_plate::S2c { owner_plate_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetMarketState =====
struct GetMarketStateHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetMarketStateHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_market_state::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetMarketState: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Collect unique markets from the request
        let mut markets_needed: std::collections::HashSet<i32> = std::collections::HashSet::new();
        for sec in &req.c2s.security_list {
            markets_needed.insert(sec.market);
        }

        // Fetch market status from backend for each unique market
        // Backend returns MarketStatus with market_id (which is the internal market ID)
        // and status (which maps directly to QotMarketState)
        let mut market_status_map: std::collections::HashMap<i32, u32> =
            std::collections::HashMap::new();
        for &qot_market in &markets_needed {
            if let Some(mkt) = futu_backend::stock_list::qot_market_to_backend(qot_market) {
                match futu_backend::stock_list::pull_single_market_status(&backend, mkt).await {
                    Ok(statuses) => {
                        // Use the first status entry as the market state
                        // (backend may return multiple sub-markets)
                        if let Some(first) = statuses.first() {
                            market_status_map.insert(qot_market, first.status);
                            tracing::debug!(
                                conn_id,
                                qot_market,
                                status = first.status,
                                text = %first.status_text,
                                "GetMarketState: backend status"
                            );
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            conn_id,
                            qot_market,
                            error = %e,
                            "GetMarketState: backend request failed, using default"
                        );
                    }
                }
            }
        }

        let market_info_list: Vec<futu_proto::qot_get_market_state::MarketInfo> = req
            .c2s
            .security_list
            .iter()
            .map(|sec| {
                let sec_key = format!("{}_{}", sec.market, sec.code);
                let name = self
                    .static_cache
                    .get_security_info(&sec_key)
                    .map(|info| info.name)
                    .unwrap_or_default();

                // Use real status from backend, fallback to QotMarketState_None (0)
                let market_state = market_status_map.get(&sec.market).copied().unwrap_or(0) as i32;

                futu_proto::qot_get_market_state::MarketInfo {
                    security: sec.clone(),
                    name,
                    market_state,
                }
            })
            .collect();

        tracing::debug!(conn_id, count = market_info_list.len(), "GetMarketState");

        let resp = futu_proto::qot_get_market_state::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_market_state::S2c { market_info_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetOptionExpirationDate (CMD 6311) =====
struct GetOptionExpirationDateHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetOptionExpirationDateHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_option_expiration_date::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetOptionExpirationDate: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Resolve owner security to stock_id
        let sec_key = format!("{}_{}", c2s.owner.market, c2s.owner.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(
                    conn_id,
                    sec_key,
                    "GetOptionExpirationDate: stock_id not found"
                );
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        // Send CMD 6311 StrikeDateReq
        let strike_req = futu_backend::proto_internal::ftcmd_option_chain::StrikeDateReq {
            stock_id: Some(stock_id),
        };
        let body = prost::Message::encode_to_vec(&strike_req);

        tracing::debug!(conn_id, stock_id, "sending CMD6311 StrikeDateReq");

        let resp_frame = match backend.request(6311, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6311 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let strike_rsp: futu_backend::proto_internal::ftcmd_option_chain::StrikeDateRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6311 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if strike_rsp.ret.unwrap_or(1) != 0 {
            tracing::warn!(conn_id, ret = ?strike_rsp.ret, "CMD6311 returned error");
            return Some(super::make_error_response(
                -1,
                "backend strike date request failed",
            ));
        }

        // Convert strike_date_list (unix timestamps) to FTAPI OptionExpirationDate
        let now_ts = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);

        // Prefer strike_dates (structured, new version), fallback to legacy fields
        let date_list: Vec<futu_proto::qot_get_option_expiration_date::OptionExpirationDate> =
            if !strike_rsp.strike_dates.is_empty() {
                strike_rsp
                    .strike_dates
                    .iter()
                    .map(|item| {
                        let ts = item.strike_date.unwrap_or(0) as u64;
                        let strike_time_str = timestamp_to_date_str(ts);
                        let left_day = item.left_day.unwrap_or(0);
                        let cycle = item.expiration.map(backend_expiration_to_api);

                        futu_proto::qot_get_option_expiration_date::OptionExpirationDate {
                            strike_time: Some(strike_time_str),
                            strike_timestamp: Some(ts as f64),
                            option_expiry_date_distance: left_day,
                            cycle,
                        }
                    })
                    .collect::<Vec<_>>()
            } else {
                strike_rsp
                    .strike_date_list
                    .iter()
                    .enumerate()
                    .map(|(i, &ts)| {
                        let strike_time_str = timestamp_to_date_str(ts as u64);
                        let left_day = strike_rsp
                            .left_day
                            .get(i)
                            .copied()
                            .unwrap_or((((ts as i64) - now_ts as i64) / 86400) as i32);
                        let cycle = strike_rsp
                            .expiration_cycle
                            .get(i)
                            .map(|&e| backend_expiration_to_api(e));

                        futu_proto::qot_get_option_expiration_date::OptionExpirationDate {
                            strike_time: Some(strike_time_str),
                            strike_timestamp: Some(ts as f64),
                            option_expiry_date_distance: left_day,
                            cycle,
                        }
                    })
                    .collect()
            };

        tracing::debug!(
            conn_id,
            count = date_list.len(),
            "GetOptionExpirationDate returning"
        );

        let resp = futu_proto::qot_get_option_expiration_date::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_option_expiration_date::S2c { date_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== RequestHistoryKLQuota =====
struct RequestHistoryKLQuotaHandler;

#[async_trait]
impl RequestHandler for RequestHistoryKLQuotaHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        // Parse request to validate format
        let _req: futu_proto::qot_request_history_kl_quota::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;

        tracing::debug!(conn_id, "RequestHistoryKLQuota: returning unlimited quota");

        // Return high quota since backend forwarding doesn't have the same limits
        let resp = futu_proto::qot_request_history_kl_quota::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_request_history_kl_quota::S2c {
                used_quota: 0,
                remain_quota: 10000,
                detail_list: Vec::new(),
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== RequestRehab (CMD 6811, same as GetRehab) =====
struct RequestRehabHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for RequestRehabHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_request_rehab::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "RequestRehab: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        let sec_key = format!("{}_{}", c2s.security.market, c2s.security.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(conn_id, sec_key, "RequestRehab: stock_id not found");
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        let exright_req = futu_backend::proto_internal::ft_cmd6811::ExrightFactorReq {
            stock_id: Some(stock_id),
            sequence: Some(0),
        };
        let body = prost::Message::encode_to_vec(&exright_req);

        tracing::debug!(conn_id, stock_id, "RequestRehab: sending CMD6811");

        let resp_frame = match backend.request(6811, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "RequestRehab CMD6811 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let exright_rsp: futu_backend::proto_internal::ft_cmd6811::ExrightFactorRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "RequestRehab CMD6811 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        let result = exright_rsp.result.unwrap_or(-1);
        if result != 0 && result != 1 {
            tracing::warn!(conn_id, result, "RequestRehab CMD6811 returned error");
            return Some(super::make_error_response(
                -1,
                "backend rehab request failed",
            ));
        }

        let rehab_list: Vec<futu_proto::qot_common::Rehab> = exright_rsp
            .exs
            .iter()
            .map(|item| {
                let ex_ts = item.ex_time.unwrap_or(0);
                let time_str = timestamp_to_date_str(ex_ts as u64);

                let mut flag: i64 = 0;
                if item.split_base.is_some() {
                    flag |= 1;
                }
                if item.join_base.is_some() {
                    flag |= 2;
                }
                if item.bonus_base.is_some() {
                    flag |= 4;
                }
                if item.transfer_base.is_some() {
                    flag |= 8;
                }
                if item.allot_base.is_some() {
                    flag |= 16;
                }
                if item.add_base.is_some() {
                    flag |= 32;
                }
                if item.dividend_amount.is_some() {
                    flag |= 64;
                }
                if item.sp_dividend_amount.is_some() {
                    flag |= 128;
                }

                let divisor = 100_000.0;

                futu_proto::qot_common::Rehab {
                    time: time_str,
                    company_act_flag: flag,
                    fwd_factor_a: item.origin_fwd_a.unwrap_or(100_000.0) / divisor,
                    fwd_factor_b: item.origin_fwd_b.unwrap_or(0.0) / divisor,
                    bwd_factor_a: item.origin_bwd_a.unwrap_or(100_000.0) / divisor,
                    bwd_factor_b: item.origin_bwd_b.unwrap_or(0.0) / divisor,
                    split_base: item.split_base.map(|v| v as i32),
                    split_ert: item.split_ert.map(|v| v as i32),
                    join_base: item.join_base.map(|v| v as i32),
                    join_ert: item.join_ert.map(|v| v as i32),
                    bonus_base: item.bonus_base.map(|v| v as i32),
                    bonus_ert: item.bonus_ert.map(|v| v as i32),
                    transfer_base: item.transfer_base.map(|v| v as i32),
                    transfer_ert: item.transfer_ert.map(|v| v as i32),
                    allot_base: item.allot_base.map(|v| v as i32),
                    allot_ert: item.allot_ert.map(|v| v as i32),
                    allot_price: item.allot_price.map(|v| v as f64 / divisor),
                    add_base: item.add_base.map(|v| v as i32),
                    add_ert: item.add_ert.map(|v| v as i32),
                    add_price: item.add_price.map(|v| v as f64 / divisor),
                    dividend: item.dividend_amount.map(|v| v as f64 / divisor),
                    sp_dividend: item.sp_dividend_amount.map(|v| v as f64 / divisor),
                    timestamp: Some(ex_ts as f64),
                    spin_off_base: item.spin_off_base.map(|v| v as f64),
                    spin_off_ert: item.spin_off_ert.map(|v| v as f64),
                }
            })
            .collect();

        tracing::debug!(conn_id, count = rehab_list.len(), "RequestRehab returning");

        let resp = futu_proto::qot_request_rehab::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_request_rehab::S2c { rehab_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetWarrant (CMD 6513 → 后端转发) =====
struct GetWarrantHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetWarrantHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_warrant::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetWarrant: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Resolve owner stock_id if present
        let owner_stock_id = if let Some(ref owner) = c2s.owner {
            let sec_key = format!("{}_{}", owner.market, owner.code);
            match self.static_cache.get_security_info(&sec_key) {
                Some(info) if info.stock_id > 0 => Some(info.stock_id),
                _ => None,
            }
        } else {
            None
        };

        // Build backend WarrantListReq with essential fields
        let backend_req = futu_backend::proto_internal::ftcmd6513::WarrantListReq {
            only_count: None,
            issuer_id: None,
            stock_owner: owner_stock_id,
            arry_warrant_type: c2s.type_list.clone(),
            cur_min: None,
            cur_max: None,
            street_min: None,
            street_max: None,
            vol_min: c2s.vol_min,
            vol_max: c2s.vol_max,
            maturity_date_min: None,
            maturity_date_max: None,
            strick_min: None,
            strick_max: None,
            conversion_min: None,
            conversion_max: None,
            ipop_min: None,
            ipop_max: None,
            premium_min: None,
            premium_max: None,
            recovery_min: None,
            recovery_max: None,
            implied_min: None,
            implied_max: None,
            leverage_ratio_min: None,
            leverage_ratio_max: None,
            lang_id: None,
            price_recovery_ratio_min: None,
            price_recovery_ratio_max: None,
            delta_min: None,
            delta_max: None,
            sort_col: c2s.sort_field,
            sort_ascend: if c2s.ascend { 1 } else { 0 },
            data_from: Some(c2s.begin),
            data_max_count: Some(c2s.num),
            status_filter: c2s.status,
            multiple_issuers: c2s.issuer_list.clone(),
            ipo_period: c2s.ipo_period,
            buy_vol_min: None,
            buy_vol_max: None,
            sell_vol_min: None,
            sell_vol_max: None,
            effective_leverage_min: None,
            effective_leverage_max: None,
            filter_no_trade_status: None,
            is_bmp: None,
            maturity_date_screens: Vec::new(),
            leverage_ratio_screens: Vec::new(),
            status_filter_screens: Vec::new(),
            market: None,
        };
        let body = prost::Message::encode_to_vec(&backend_req);

        tracing::debug!(conn_id, "sending CMD6513 WarrantListReq");

        let resp_frame = match backend.request(6513, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6513 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let backend_rsp: futu_backend::proto_internal::ftcmd6513::WarrantListRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6513 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if backend_rsp.result != 0 {
            tracing::warn!(
                conn_id,
                result = backend_rsp.result,
                "CMD6513 returned error"
            );
            return Some(super::make_error_response(
                -1,
                "backend warrant request failed",
            ));
        }

        let all_count = backend_rsp.all_count as i32;
        let last_page = backend_rsp.if_last_page != 0;

        // Map each backend WarrantItem to FTAPI WarrantData
        let warrant_data_list: Vec<futu_proto::qot_get_warrant::WarrantData> = backend_rsp
            .arry_items
            .iter()
            .filter_map(|item| {
                // Resolve stock Security from stock_id
                let stock_id = item.stock_id?;
                let stock_key = self.static_cache.id_to_key.get(&stock_id)?;
                let stock_info = self.static_cache.get_security_info(stock_key.value())?;
                let stock = futu_proto::qot_common::Security {
                    market: stock_info.market,
                    code: stock_info.code.clone(),
                };

                // Resolve owner Security from stock_owner
                let owner_id = item.stock_owner?;
                let owner_key = self.static_cache.id_to_key.get(&owner_id)?;
                let owner_info = self.static_cache.get_security_info(owner_key.value())?;
                let owner = futu_proto::qot_common::Security {
                    market: owner_info.market,
                    code: owner_info.code.clone(),
                };

                let warrant_type = item.warrant_type.unwrap_or(0);

                // Issuer: map backend IssuerAgency enum to API Issuer enum
                let issuer = map_issuer_backend_to_api(item.issuer_id.unwrap_or(0));

                // Status: map backend WarrantStatus to API WarrantStatus
                let status = map_warrant_status_backend_to_api(item.status.unwrap_or(0));

                // Prices: backend values are *1000, divide to get real values
                let cur_price = item.current_price.unwrap_or(0) as f64 / 1000.0;
                let last_close_price = item.lastclose_price.unwrap_or(0) as f64 / 1000.0;
                let high_price = item.high_price.unwrap_or(0) as f64 / 1000.0;
                let low_price = item.low_price.unwrap_or(0) as f64 / 1000.0;
                let strike_price = item.strick_price.unwrap_or(0) as f64 / 1000.0;
                let bid_price = item.buy_price.unwrap_or(0) as f64 / 1000.0;
                let ask_price = item.sell_price.unwrap_or(0) as f64 / 1000.0;
                let recovery_price_val = item.recovery_price.unwrap_or(0) as f64 / 1000.0;
                let break_even_point = item.break_even_point.unwrap_or(0) as f64 / 1000.0;

                // Computed fields (matching C++ logic)
                let price_change_val = cur_price - last_close_price;
                let change_rate = if last_close_price.abs() > 1e-6 {
                    price_change_val / last_close_price * 100.0
                } else {
                    0.0
                };

                // Ratio/percentage fields: backend *1000, divide
                let conversion_ratio = item.conversion_ratio.unwrap_or(0) as f64 / 1000.0;
                let conversion_price = conversion_ratio * cur_price;
                let street_rate = item.street_rate.unwrap_or(0) as f64 / 1000.0;
                let premium = item.premium.unwrap_or(0) as f64 / 1000.0;
                let leverage = item.leverage.unwrap_or(0) as f64 / 1000.0;
                let effective_leverage = item.effective_leverage.unwrap_or(0) as f64 / 1000.0;
                let ipop = item.ipop.unwrap_or(0) as f64 / 1000.0;
                let amplitude = item.amplitude.unwrap_or(0) as f64 / 1000.0;
                let turnover = item.turnover.unwrap_or(0) as f64 / 1000.0;
                let score = item.fx_score.unwrap_or(0) as f64 / 1000.0;

                // Volumes (no scaling)
                let volume = item.volume.unwrap_or(0) as i64;
                let bid_vol = item.buy_vol.unwrap_or(0) as i64;
                let ask_vol = item.sell_vol.unwrap_or(0) as i64;
                let street_vol = item.street_vol.unwrap_or(0) as i64;
                let issue_size = item.issue_size.unwrap_or(0) as i64;
                let lot_size = item.lot_size.unwrap_or(0) as i32;

                // Timestamps -> date strings
                let maturity_time = item.maturity_date.map(format_timestamp).unwrap_or_default();
                let maturity_timestamp = item.maturity_date.map(|ts| ts as f64);
                let list_time = item.ipo_time.map(format_timestamp).unwrap_or_default();
                let list_timestamp = item.ipo_time.map(|ts| ts as f64);
                let last_trade_time = item
                    .last_trade_date
                    .map(format_timestamp)
                    .unwrap_or_default();
                let last_trade_timestamp = item.last_trade_date.map(|ts| ts as f64);

                // Name from static cache
                let name = stock_info.name.clone();

                // Type-specific optional fields (matching C++ GetAPIStockWarrant)
                let (delta, implied_volatility) = if warrant_type == 1 || warrant_type == 2 {
                    // Buy/Sell
                    (
                        Some(item.delta.unwrap_or(0) as f64 / 1000.0),
                        Some(item.implied_volatility.unwrap_or(0) as f64 / 1000.0),
                    )
                } else {
                    (None, None)
                };

                let (recovery_price, price_recovery_ratio) =
                    if warrant_type == 3 || warrant_type == 4 {
                        // Bull/Bear
                        (
                            Some(recovery_price_val),
                            Some(item.price_recovery_ratio.unwrap_or(0) as f64 / 1000.0),
                        )
                    } else {
                        (None, None)
                    };

                // Upper/lower strike price: backend *10^9
                let (upper_strike_price, lower_strike_price, in_line_price_status) =
                    if warrant_type == 5 {
                        // InLine
                        let upper = item.upper_strike_price.unwrap_or(0) as f64 / 1_000_000_000.0;
                        let lower = item.lower_strike_price.unwrap_or(0) as f64 / 1_000_000_000.0;
                        // iw_price_status: 0=WithIn(1), 1=Outside(2)
                        let price_status = if item.iw_price_status.unwrap_or(0) == 0 {
                            1_i32
                        } else {
                            2_i32
                        };
                        (Some(upper), Some(lower), Some(price_status))
                    } else {
                        (None, None, None)
                    };

                Some(futu_proto::qot_get_warrant::WarrantData {
                    stock,
                    owner,
                    r#type: warrant_type,
                    issuer,
                    maturity_time,
                    maturity_timestamp,
                    list_time,
                    list_timestamp,
                    last_trade_time,
                    last_trade_timestamp,
                    recovery_price,
                    conversion_ratio,
                    lot_size,
                    strike_price,
                    last_close_price,
                    name,
                    cur_price,
                    price_change_val,
                    change_rate,
                    status,
                    bid_price,
                    ask_price,
                    bid_vol,
                    ask_vol,
                    volume,
                    turnover,
                    score,
                    premium,
                    break_even_point,
                    leverage,
                    ipop,
                    price_recovery_ratio,
                    conversion_price,
                    street_rate,
                    street_vol,
                    amplitude,
                    issue_size,
                    high_price,
                    low_price,
                    implied_volatility,
                    delta,
                    effective_leverage,
                    upper_strike_price,
                    lower_strike_price,
                    in_line_price_status,
                })
            })
            .collect();

        tracing::debug!(
            conn_id,
            all_count,
            last_page,
            mapped_count = warrant_data_list.len(),
            "GetWarrant: mapped backend warrant items"
        );

        let resp = futu_proto::qot_get_warrant::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_warrant::S2c {
                last_page,
                all_count,
                warrant_data_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

/// Map backend IssuerAgency enum value to FTAPI Issuer enum value.
/// Backend uses FTCMD6513/20334 IssuerAgency, API uses Qot_Common::Issuer.
fn map_issuer_backend_to_api(backend_issuer: u32) -> i32 {
    match backend_issuer {
        1 => 19,  // BC -> Issuer_BC
        2 => 12,  // BI -> Issuer_BI
        3 => 2,   // BP -> Issuer_BP
        4 => 3,   // CS -> Issuer_CS
        5 => 4,   // CT -> Issuer_CT
        6 => 13,  // DB -> Issuer_DB
        7 => 14,  // DC -> Issuer_DC
        8 => 5,   // EA -> Issuer_EA
        9 => 6,   // GS -> Issuer_GS
        10 => 7,  // HS -> Issuer_HS
        11 => 8,  // JP -> Issuer_JP
        12 => 22, // KC -> Issuer_KC
        13 => 9,  // MB -> Issuer_MB
        // 14 = MC, no API mapping
        15 => 15, // ML -> Issuer_ML
        16 => 16, // NM -> Issuer_NM
        17 => 17, // RB -> Issuer_RB
        18 => 18, // RS -> Issuer_RS
        19 => 10, // SC -> Issuer_SC
        20 => 1,  // SG -> Issuer_SG
        21 => 11, // UB -> Issuer_UB
        22 => 20, // HT -> Issuer_HT
        23 => 21, // VT -> Issuer_VT
        24 => 23, // MS -> Issuer_MS
        25 => 24, // GJ -> Issuer_GJ
        26 => 25, // XZ -> Issuer_XZ
        27 => 26, // HU -> Issuer_HU
        28 => 27, // KS -> Issuer_KS
        29 => 28, // CI -> Issuer_CI
        _ => 0,   // Issuer_Unknow
    }
}

/// Map backend WarrantStatus enum value to FTAPI WarrantStatus enum value.
fn map_warrant_status_backend_to_api(backend_status: i32) -> i32 {
    match backend_status {
        0 => 1, // status_normal -> WarrantStatus_Normal
        1 => 2, // status_common (suspend) -> WarrantStatus_Suspend
        2 => 3, // status_stop -> WarrantStatus_StopTrade
        3 => 4, // status_listing -> WarrantStatus_PendingListing
        _ => 0, // WarrantStatus_Unknow
    }
}

// ===== GetCapitalFlow (CMD 6694 → 后端转发) =====
struct GetCapitalFlowHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetCapitalFlowHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_capital_flow::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetCapitalFlow: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Resolve stock_id
        let sec_key = format!("{}_{}", c2s.security.market, c2s.security.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(conn_id, sec_key, "GetCapitalFlow: stock_id not found");
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        // 判断 period_type: INTRADAY(1) 或未设置 → CMD 6694, 其他 → CMD 6695
        let period_type = c2s.period_type.unwrap_or(1); // 默认 INTRADAY

        let (flow_item_list, last_valid_time, last_valid_ts) = if period_type <= 1 {
            // 实时资金流向: CMD 6694
            let backend_req = futu_backend::proto_internal::cash_flow_cs::RealFlowTrendReq {
                stock_id: Some(stock_id),
                req_section: Some(0), // REQUEST_SECTION_NORMAL (C++ 对齐)
            };
            let body = prost::Message::encode_to_vec(&backend_req);

            tracing::debug!(conn_id, stock_id, "sending CMD6694 RealFlowTrendReq");

            let resp_frame = match backend.request(6694, body).await {
                Ok(f) => f,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6694 request failed");
                    return Some(super::make_error_response(-1, "backend request failed"));
                }
            };

            let backend_rsp: futu_backend::proto_internal::cash_flow_cs::RealFlowTrendRsp =
                match prost::Message::decode(resp_frame.body.as_ref()) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::error!(conn_id, error = %e, "CMD6694 decode failed");
                        return Some(super::make_error_response(
                            -1,
                            "backend response decode failed",
                        ));
                    }
                };

            if let Some(err) = backend_rsp.error_code {
                if err != 0 {
                    tracing::warn!(conn_id, err, "CMD6694 returned error");
                    return Some(super::make_error_response(
                        -1,
                        "backend capital flow request failed",
                    ));
                }
            }

            let items: Vec<futu_proto::qot_get_capital_flow::CapitalFlowItem> = backend_rsp
                .section_list
                .iter()
                .flat_map(|section| {
                    section.point_list.iter().filter_map(|pt| {
                        // C++ 跳过 bIsEmpty 的条目 (is_empty != 0 表示空点)
                        if pt.is_empty.unwrap_or(0) != 0 {
                            return None;
                        }
                        Some(futu_proto::qot_get_capital_flow::CapitalFlowItem {
                            in_flow: pt.total_net_in.unwrap_or(0) as f64 / 1000.0,
                            time: pt.time.map(timestamp_to_datetime_str),
                            timestamp: pt.time.map(|t| t as f64),
                            main_in_flow: None, // C++: 实时数据不设 main_in_flow
                            super_in_flow: pt.super_net_in.map(|v| v as f64 / 1000.0),
                            big_in_flow: pt.big_net_in.map(|v| v as f64 / 1000.0),
                            mid_in_flow: pt.mid_net_in.map(|v| v as f64 / 1000.0),
                            sml_in_flow: pt.sml_net_in.map(|v| v as f64 / 1000.0),
                        })
                    })
                })
                .collect();

            let last_valid_ts = backend_rsp.update_time.map(|t| t as f64);
            let last_valid_time = backend_rsp.update_time.map(timestamp_to_datetime_str);
            (items, last_valid_time, last_valid_ts)
        } else {
            // 历史资金流向: CMD 6695
            // FTAPI PeriodType → 后端 PeriodType 映射
            let nn_period = match period_type {
                2 => 1u32, // DAY
                3 => 2,    // WEEK
                4 => 3,    // MONTH
                _ => 1,    // 默认 DAY
            };

            // 解析时间范围
            let begin_time = c2s.begin_time.as_deref().and_then(parse_date_to_timestamp);
            let end_time = c2s.end_time.as_deref().and_then(parse_date_to_timestamp);

            // 如果时间无效，返回空数据 (C++ 行为)
            if let (Some(bt), Some(et)) = (begin_time, end_time) {
                if bt > et {
                    let resp = futu_proto::qot_get_capital_flow::Response {
                        ret_type: 0,
                        ret_msg: None,
                        err_code: None,
                        s2c: Some(futu_proto::qot_get_capital_flow::S2c {
                            flow_item_list: vec![],
                            last_valid_time: None,
                            last_valid_timestamp: None,
                        }),
                    };
                    return Some(prost::Message::encode_to_vec(&resp));
                }
            }

            let backend_req = futu_backend::proto_internal::cash_flow_cs::HistoryCashFlowReq {
                stock_id: Some(stock_id),
                req_period_type: Some(nn_period),
                base_time: end_time.map(|t| t as u64),
                count: Some(200), // C++ MaxReqNum = 200
                need_detail: Some(0),
                analysis_type: Some(0),
            };
            let body = prost::Message::encode_to_vec(&backend_req);

            tracing::debug!(
                conn_id,
                stock_id,
                nn_period,
                "sending CMD6695 HistoryCashFlowReq"
            );

            let resp_frame = match backend.request(6695, body).await {
                Ok(f) => f,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6695 request failed");
                    return Some(super::make_error_response(-1, "backend request failed"));
                }
            };

            let backend_rsp: futu_backend::proto_internal::cash_flow_cs::HistoryCashFlowRsp =
                match prost::Message::decode(resp_frame.body.as_ref()) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::error!(conn_id, error = %e, "CMD6695 decode failed");
                        return Some(super::make_error_response(
                            -1,
                            "backend response decode failed",
                        ));
                    }
                };

            if let Some(err) = backend_rsp.error_code {
                if err != 0 {
                    tracing::warn!(conn_id, err, "CMD6695 returned error");
                    return Some(super::make_error_response(
                        -1,
                        "backend history capital flow request failed",
                    ));
                }
            }

            let items: Vec<futu_proto::qot_get_capital_flow::CapitalFlowItem> = backend_rsp
                .point_list
                .iter()
                .filter_map(|pt| {
                    if pt.is_empty.unwrap_or(0) != 0 {
                        return None;
                    }
                    Some(futu_proto::qot_get_capital_flow::CapitalFlowItem {
                        in_flow: pt.total_net_in.unwrap_or(0) as f64 / 1000.0,
                        time: pt.time.map(timestamp_to_datetime_str),
                        timestamp: pt.time.map(|t| t as f64),
                        main_in_flow: pt.main_net_in.map(|v| v as f64 / 1000.0),
                        super_in_flow: pt.super_net_in.map(|v| v as f64 / 1000.0),
                        big_in_flow: pt.big_net_in.map(|v| v as f64 / 1000.0),
                        mid_in_flow: pt.mid_net_in.map(|v| v as f64 / 1000.0),
                        sml_in_flow: pt.sml_net_in.map(|v| v as f64 / 1000.0),
                    })
                })
                .collect();

            (items, None, None::<f64>)
        };

        tracing::debug!(
            conn_id,
            count = flow_item_list.len(),
            "GetCapitalFlow: returning flow items"
        );

        let resp = futu_proto::qot_get_capital_flow::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_capital_flow::S2c {
                flow_item_list,
                last_valid_time,
                last_valid_timestamp: last_valid_ts,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetCapitalDistribution (CMD 6693 → 后端转发) =====
struct GetCapitalDistributionHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetCapitalDistributionHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_capital_distribution::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetCapitalDistribution: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Resolve stock_id
        let sec_key = format!("{}_{}", c2s.security.market, c2s.security.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(
                    conn_id,
                    sec_key,
                    "GetCapitalDistribution: stock_id not found"
                );
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        // Build backend RealDistributionReq (CMD 6693)
        let backend_req = futu_backend::proto_internal::cash_flow_cs::RealDistributionReq {
            stock_id: Some(stock_id),
        };
        let body = prost::Message::encode_to_vec(&backend_req);

        tracing::debug!(conn_id, stock_id, "sending CMD6693 RealDistributionReq");

        let resp_frame = match backend.request(6693, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6693 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let backend_rsp: futu_backend::proto_internal::cash_flow_cs::RealDistributionRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6693 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if let Some(err) = backend_rsp.error_code {
            if err != 0 {
                tracing::warn!(conn_id, err, "CMD6693 returned error");
                return Some(super::make_error_response(
                    -1,
                    "backend capital distribution request failed",
                ));
            }
        }

        // Convert backend DistributionItem list → FTAPI S2c fields
        // Backend OrderType: 1=Small, 2=Middle, 3=Big, 4=Supper
        // Backend values are in 0.001 units
        let mut in_super = 0.0_f64;
        let mut in_big = 0.0_f64;
        let mut in_mid = 0.0_f64;
        let mut in_small = 0.0_f64;
        let mut out_super = 0.0_f64;
        let mut out_big = 0.0_f64;
        let mut out_mid = 0.0_f64;
        let mut out_small = 0.0_f64;

        for item in &backend_rsp.items {
            let flow_in = item.r#in.unwrap_or(0) as f64 / 1000.0;
            let flow_out = item.out.unwrap_or(0) as f64 / 1000.0;
            match item.order_type.unwrap_or(0) {
                1 => {
                    in_small = flow_in;
                    out_small = flow_out;
                }
                2 => {
                    in_mid = flow_in;
                    out_mid = flow_out;
                }
                3 => {
                    in_big = flow_in;
                    out_big = flow_out;
                }
                4 => {
                    in_super = flow_in;
                    out_super = flow_out;
                }
                _ => {}
            }
        }

        let update_timestamp = backend_rsp.update_time.map(|t| t as f64);

        tracing::debug!(
            conn_id,
            items = backend_rsp.items.len(),
            "GetCapitalDistribution: returning distribution"
        );

        let resp = futu_proto::qot_get_capital_distribution::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_capital_distribution::S2c {
                capital_in_super: Some(in_super),
                capital_in_big: in_big,
                capital_in_mid: in_mid,
                capital_in_small: in_small,
                capital_out_super: Some(out_super),
                capital_out_big: out_big,
                capital_out_mid: out_mid,
                capital_out_small: out_small,
                update_time: backend_rsp.update_time.map(timestamp_to_datetime_str),
                update_timestamp,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetHistoryKL (CMD 6161, same as RequestHistoryKL) =====
struct GetHistoryKLHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetHistoryKLHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        // GetHistoryKL uses the same request format as RequestHistoryKL
        let req: futu_proto::qot_request_history_kl::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetHistoryKL: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        let sec_key = format!("{}_{}", c2s.security.market, c2s.security.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(conn_id, sec_key, "GetHistoryKL: stock_id not found");
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        let backend_kl_type = match ftapi_kl_type_to_backend(c2s.kl_type) {
            Some(v) => v,
            None => {
                tracing::warn!(
                    conn_id,
                    kl_type = c2s.kl_type,
                    "GetHistoryKL: invalid kl_type"
                );
                return Some(super::make_error_response(-1, "invalid kl_type"));
            }
        };

        let exright_type = c2s.rehab_type as u32;
        let qot_market = c2s.security.market;
        let begin_ts = date_str_to_timestamp(&c2s.begin_time, qot_market).unwrap_or(0);
        let end_ts = if c2s.end_time.is_empty() {
            u64::MAX
        } else {
            date_str_to_timestamp(&c2s.end_time, qot_market)
                .map(|t| t + 86399)
                .unwrap_or(u64::MAX)
        };

        // C++ always uses data_range_type=1 (BEGIN_TIME + END_TIME)
        let max_kl_num = c2s.max_ack_kl_num.unwrap_or(0) as usize;

        let kline_req = futu_backend::proto_internal::ft_cmd_kline::KlineReq {
            security_id: Some(stock_id),
            kline_type: Some(backend_kl_type),
            exright_type: Some(exright_type),
            data_set_type: Some(0),
            data_range_type: Some(1), // always BEGIN_TIME + END_TIME
            begin_time: Some(begin_ts),
            end_time: Some(end_ts),
            item_count: None,
            end_time_offset: None,
        };

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

        tracing::debug!(
            conn_id,
            stock_id,
            kl_type = backend_kl_type,
            exright = exright_type,
            "GetHistoryKL: sending CMD6161 KlineReq"
        );

        let resp_frame = match backend.request(6161, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "GetHistoryKL: CMD6161 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let kline_rsp: futu_backend::proto_internal::ft_cmd_kline::KlineRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "GetHistoryKL: CMD6161 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        let result = kline_rsp.result.unwrap_or(-1);
        if result != 0 {
            tracing::warn!(conn_id, result, "GetHistoryKL: CMD6161 returned error");
            return Some(super::make_error_response(
                -1,
                "backend kline request failed",
            ));
        }

        let mut kl_list: Vec<futu_proto::qot_common::KLine> = kline_rsp
            .kline_item_list
            .iter()
            .map(kline_item_to_ftapi)
            .collect();
        if max_kl_num > 0 && kl_list.len() > max_kl_num {
            kl_list.truncate(max_kl_num);
        }

        tracing::debug!(
            conn_id,
            count = kl_list.len(),
            "GetHistoryKL returning klines"
        );

        // GetHistoryKL uses the same response format as RequestHistoryKL
        let resp = futu_proto::qot_request_history_kl::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_request_history_kl::S2c {
                security: c2s.security.clone(),
                name: None,
                kl_list,
                next_req_key: None,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetHistoryKLPoints (已在 v5.21 分支删除) =====
// C++ 源码中 FTAPI_ProtoID_Qot_GetHistoryKLPoints (3101) 已被注释掉:
// //#define FTAPI_ProtoID_Qot_GetHistoryKLPoints 3101 /**< 获取多只股票历史单点K线（v5.21分支删除相关功能） */
struct GetHistoryKLPointsHandler;

#[async_trait]
impl RequestHandler for GetHistoryKLPointsHandler {
    async fn handle(&self, conn_id: u64, _request: &IncomingRequest) -> Option<Vec<u8>> {
        // 此接口已在 C++ v5.21 分支中被删除，proto 也已移除，不再支持。
        // 不尝试解析请求体，直接返回错误。
        tracing::debug!(conn_id, "GetHistoryKLPoints: feature removed in v5.21");
        Some(super::make_error_response(
            -1,
            "GetHistoryKLPoints has been removed since v5.21. Please use RequestHistoryKL instead.",
        ))
    }
}

// ===== GetTradeDate (CMD 6733, same logic as RequestTradeDate) =====
struct GetTradeDateHandler {
    backend: crate::bridge::SharedBackend,
}

#[async_trait]
impl RequestHandler for GetTradeDateHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        // GetTradeDate uses the same request format as RequestTradeDate
        let req: futu_proto::qot_request_trade_date::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetTradeDate: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        let backend_market = match ftapi_market_to_backend(c2s.market) {
            Some(m) => m,
            None => {
                tracing::warn!(
                    conn_id,
                    market = c2s.market,
                    "GetTradeDate: unsupported market"
                );
                return Some(super::make_error_response(-1, "unsupported market"));
            }
        };

        let begin_key = match date_str_to_yyyymmdd(&c2s.begin_time) {
            Some(v) => v,
            None => {
                tracing::warn!(conn_id, begin = %c2s.begin_time, "GetTradeDate: invalid begin_time");
                return Some(super::make_error_response(-1, "invalid begin_time"));
            }
        };

        let end_key = match date_str_to_yyyymmdd(&c2s.end_time) {
            Some(v) => v,
            None => {
                tracing::warn!(conn_id, end = %c2s.end_time, "GetTradeDate: invalid end_time");
                return Some(super::make_error_response(-1, "invalid end_time"));
            }
        };

        let range_req = futu_backend::proto_internal::market_trading_day::RangeTradingDayReq {
            begin_date_key: Some(begin_key),
            end_date_key: Some(end_key),
            market_id: Some(backend_market),
        };

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

        tracing::debug!(
            conn_id,
            backend_market,
            begin_key,
            end_key,
            "GetTradeDate: sending CMD6733 RangeTradingDayReq"
        );

        let resp_frame = match backend.request(6733, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "GetTradeDate: CMD6733 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let range_rsp: futu_backend::proto_internal::market_trading_day::RangeTradingDayRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "GetTradeDate: CMD6733 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        let code = range_rsp.code.unwrap_or(-1);
        if code != 0 {
            tracing::warn!(conn_id, code, "GetTradeDate: CMD6733 returned error");
            return Some(super::make_error_response(
                -1,
                "backend trade date request failed",
            ));
        }

        let trade_date_list: Vec<futu_proto::qot_request_trade_date::TradeDate> = range_rsp
            .day_infos
            .iter()
            .map(|day| {
                let time_date = day.time_date.unwrap_or(0);
                let time_str = yyyymmdd_to_date_str(time_date);
                let trading_type = day.trading_type.unwrap_or(4);
                let trade_date_type = backend_trading_type_to_ftapi(trading_type);
                let ts = day.date_key.unwrap_or(0) as f64;

                futu_proto::qot_request_trade_date::TradeDate {
                    time: time_str,
                    timestamp: Some(ts),
                    trade_date_type: Some(trade_date_type),
                }
            })
            .collect();

        tracing::debug!(
            conn_id,
            count = trade_date_list.len(),
            "GetTradeDate returning trade dates"
        );

        let resp = futu_proto::qot_request_trade_date::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_request_trade_date::S2c { trade_date_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetSuspend =====
// C++ 对应: APIServer_Qot_Suspend::OnClientReq_Suspend
// 停牌数据来源是 HTTP 下载的 zip 文件（从 cosgz.myqcloud.com 下载），
// 解压后解析 StockSuspend.proto 格式的二进制 .dat 文件，缓存在本地内存中。
// 查询时对每只股票用 lower_bound/upper_bound 在已排序的时间戳数组中筛选。
struct GetSuspendHandler {
    suspend_cache: futu_backend::suspend_data::SuspendCache,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetSuspendHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_suspend::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;

        let c2s = &req.c2s;

        // C++ 流程: APIServer_Qot_Suspend::OnClientReq_Suspend
        // 1. 解析 security_list → stock_id
        // 2. 解析 begin_time / end_time → timestamp
        // 3. 对每只股票调用 GetSuspendData(stockID, startTime, endTime)
        //    内部使用 lower_bound/upper_bound 在已排序的时间戳数组中筛选
        // 4. 将时间戳转为 API 时间字符串

        let security_suspend_list: Vec<futu_proto::qot_get_suspend::SecuritySuspend> = c2s
            .security_list
            .iter()
            .map(|sec| {
                // Look up stock_id from static cache
                let sec_key = format!("{}_{}", sec.market, sec.code);
                let stock_id = self
                    .static_cache
                    .get_security_info(&sec_key)
                    .map(|info| info.stock_id)
                    .unwrap_or(0);

                // Parse begin_time/end_time to timestamps
                let begin_ts = parse_api_time_str(&c2s.begin_time, sec.market).unwrap_or(0);
                let end_ts = parse_api_time_str(&c2s.end_time, sec.market).unwrap_or(u64::MAX);

                // Query suspend data using binary search (matching C++ lower_bound/upper_bound)
                let suspend_list = if stock_id > 0 {
                    if let Some(timestamps) = self.suspend_cache.get(&stock_id) {
                        // C++ 使用 lower_bound(begin) + upper_bound(end) 在已排序数组上筛选
                        let start_idx = timestamps.partition_point(|&t| t < begin_ts);
                        let end_idx = timestamps.partition_point(|&t| t <= end_ts);
                        timestamps[start_idx..end_idx]
                            .iter()
                            .map(|&ts| futu_proto::qot_get_suspend::Suspend {
                                time: timestamp_to_datetime_str(ts),
                                timestamp: Some(ts as f64),
                            })
                            .collect()
                    } else {
                        Vec::new()
                    }
                } else {
                    Vec::new()
                };

                futu_proto::qot_get_suspend::SecuritySuspend {
                    security: sec.clone(),
                    suspend_list,
                }
            })
            .collect();

        tracing::debug!(conn_id, count = security_suspend_list.len(), "GetSuspend");

        let resp = futu_proto::qot_get_suspend::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_suspend::S2c {
                security_suspend_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

/// 解析 API 时间字符串 "yyyy-MM-dd" 或 "yyyy-MM-dd HH:mm:ss" 为 Unix 时间戳
fn parse_api_time_str(s: &str, market: i32) -> Option<u64> {
    // 先尝试 "yyyy-MM-dd HH:mm:ss" 格式
    if s.len() >= 19 {
        let date_part = &s[..10];
        let time_part = &s[11..19];
        if let Some(day_ts) = date_str_to_timestamp(date_part, market) {
            let time_parts: Vec<&str> = time_part.split(':').collect();
            if time_parts.len() == 3 {
                let h: u64 = time_parts[0].parse().ok()?;
                let m: u64 = time_parts[1].parse().ok()?;
                let sec: u64 = time_parts[2].parse().ok()?;
                return Some(day_ts + h * 3600 + m * 60 + sec);
            }
        }
    }
    // 回退到 "yyyy-MM-dd" 格式
    date_str_to_timestamp(s, market)
}

// ===== GetBroker (cache-based) =====
struct GetBrokerHandler {
    cache: Arc<QotCache>,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetBrokerHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_broker::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let sec = &req.c2s.security;
        let sec_key = qot_cache::make_key(sec.market, &sec.code);

        let name = self
            .static_cache
            .get_security_info(&sec_key)
            .map(|info| info.name);

        let cached = self.cache.get_broker(&sec_key);
        let (broker_ask_list, broker_bid_list) = match cached {
            Some(b) => {
                let to_proto =
                    |items: &[qot_cache::CachedBrokerItem]| -> Vec<futu_proto::qot_common::Broker> {
                        items
                            .iter()
                            .map(|item| futu_proto::qot_common::Broker {
                                id: item.id,
                                name: item.name.clone(),
                                pos: item.pos,
                                order_id: None,
                                volume: None,
                            })
                            .collect()
                    };
                (to_proto(&b.ask_list), to_proto(&b.bid_list))
            }
            None => (Vec::new(), Vec::new()),
        };

        let resp = futu_proto::qot_get_broker::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_broker::S2c {
                security: sec.clone(),
                name,
                broker_ask_list,
                broker_bid_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetOptionChain (CMD 6311 strike dates + CMD 6312 chain) =====
struct GetOptionChainHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetOptionChainHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_option_chain::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetOptionChain: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Resolve owner security to stock_id
        let sec_key = format!("{}_{}", c2s.owner.market, c2s.owner.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                tracing::warn!(conn_id, sec_key, "GetOptionChain: stock_id not found");
                return Some(super::make_error_response(
                    -1,
                    "security not found in cache",
                ));
            }
        };

        // Step 1: Get strike dates via CMD 6311
        let strike_req = futu_backend::proto_internal::ftcmd_option_chain::StrikeDateReq {
            stock_id: Some(stock_id),
        };
        let body = prost::Message::encode_to_vec(&strike_req);

        tracing::debug!(conn_id, stock_id, "GetOptionChain: sending CMD6311");

        let resp_frame = match backend.request(6311, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "GetOptionChain: CMD6311 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let strike_rsp: futu_backend::proto_internal::ftcmd_option_chain::StrikeDateRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "GetOptionChain: CMD6311 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if strike_rsp.ret.unwrap_or(1) != 0 {
            tracing::warn!(conn_id, ret = ?strike_rsp.ret, "GetOptionChain: CMD6311 error");
            return Some(super::make_error_response(
                -1,
                "backend strike date request failed",
            ));
        }

        // Filter strike dates by begin_time / end_time
        let qot_market = c2s.owner.market;
        let begin_ts = date_str_to_timestamp(&c2s.begin_time, qot_market).unwrap_or(0);
        let end_ts = if c2s.end_time.is_empty() {
            u64::MAX
        } else {
            date_str_to_timestamp(&c2s.end_time, qot_market)
                .map(|t| t + 86399)
                .unwrap_or(u64::MAX)
        };

        let filtered_dates: Vec<u32> = strike_rsp
            .strike_date_list
            .iter()
            .copied()
            .filter(|&ts| {
                let ts64 = ts as u64;
                ts64 >= begin_ts && ts64 <= end_ts
            })
            .collect();

        // Step 2: For each strike date, fetch option chain via CMD 6312
        let option_type = c2s.r#type.unwrap_or(0) as u32; // 0=all, 1=call, 2=put

        let mut option_chain_list: Vec<futu_proto::qot_get_option_chain::OptionChain> = Vec::new();

        for &strike_date in &filtered_dates {
            let chain_req = futu_backend::proto_internal::ftcmd_option_chain::OptionChainReq {
                stock_id: Some(stock_id),
                strike_date: Some(strike_date),
                option_type: if option_type > 0 {
                    Some(option_type)
                } else {
                    Some(0)
                },
                sort_type: Some(2), // ASC
                sort_id: Some(1),   // by strike price
                from: Some(0),
                count: Some(200),
            };
            let body = prost::Message::encode_to_vec(&chain_req);

            let resp_frame = match backend.request(6312, body).await {
                Ok(f) => f,
                Err(e) => {
                    tracing::warn!(
                        conn_id,
                        error = %e,
                        strike_date,
                        "GetOptionChain: CMD6312 request failed"
                    );
                    continue;
                }
            };

            let chain_rsp: futu_backend::proto_internal::ftcmd_option_chain::OptionChainRsp =
                match prost::Message::decode(resp_frame.body.as_ref()) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::warn!(conn_id, error = %e, "GetOptionChain: CMD6312 decode failed");
                        continue;
                    }
                };

            if chain_rsp.ret.unwrap_or(1) != 0 {
                continue;
            }

            let strike_time_str = timestamp_to_date_str(strike_date as u64);

            let options: Vec<futu_proto::qot_get_option_chain::OptionItem> = chain_rsp
                .option_chain
                .iter()
                .map(|item| {
                    let call_info = item.call_option.as_ref().map(|opt| {
                        backend_option_to_static_info(opt, c2s.owner.market, strike_date)
                    });
                    let put_info = item.put_option.as_ref().map(|opt| {
                        backend_option_to_static_info(opt, c2s.owner.market, strike_date)
                    });
                    futu_proto::qot_get_option_chain::OptionItem {
                        call: call_info,
                        put: put_info,
                    }
                })
                .collect();

            option_chain_list.push(futu_proto::qot_get_option_chain::OptionChain {
                strike_time: strike_time_str,
                option: options,
                strike_timestamp: Some(strike_date as f64),
            });
        }

        tracing::debug!(
            conn_id,
            chain_count = option_chain_list.len(),
            "GetOptionChain returning"
        );

        let resp = futu_proto::qot_get_option_chain::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_option_chain::S2c {
                option_chain: option_chain_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

/// Convert backend Option to FTAPI SecurityStaticInfo for option chain
fn backend_option_to_static_info(
    opt: &futu_backend::proto_internal::ftcmd_option_chain::Option,
    market: i32,
    strike_date: u32,
) -> futu_proto::qot_common::SecurityStaticInfo {
    let code = opt
        .option_string_code
        .clone()
        .unwrap_or_else(|| opt.option_id.unwrap_or(0).to_string());

    let strike_price_raw = opt.hp_strike_price.unwrap_or(0);
    let strike_price = strike_price_raw as f64 / 1_000_000_000.0;

    let option_type = opt.option_type.unwrap_or(0);
    // FTAPI OptionType: 1=Call, 2=Put
    let ftapi_option_type = match option_type {
        1 => 1,
        2 => 2,
        _ => 0,
    };

    let name = opt.option_name.clone().unwrap_or_default();
    let strike_time_str = timestamp_to_date_str(strike_date as u64);

    futu_proto::qot_common::SecurityStaticInfo {
        basic: futu_proto::qot_common::SecurityStaticBasic {
            security: futu_proto::qot_common::Security {
                market,
                code: code.clone(),
            },
            id: opt.option_id.unwrap_or(0) as i64,
            lot_size: opt.contract_share_size.unwrap_or(100) as i32,
            sec_type: 8, // SecurityType_Drvt (derivative/option)
            name: name.clone(),
            list_time: String::new(),
            delisting: opt.delisting_flag.map(|f| f != 0),
            list_timestamp: None,
            exch_type: None,
        },
        warrant_ex_data: None,
        option_ex_data: Some(futu_proto::qot_common::OptionStaticExData {
            r#type: ftapi_option_type,
            owner: futu_proto::qot_common::Security {
                market,
                code: String::new(),
            },
            strike_time: strike_time_str,
            strike_price,
            suspend: opt.suspend_flag.unwrap_or(0) != 0,
            market: opt.market.clone().unwrap_or_default(),
            index_option_type: None,
            strike_timestamp: Some(strike_date as f64),
            expiration_cycle: None,
            option_standard_type: None,
            option_settlement_mode: None,
        }),
        future_ex_data: None,
    }
}

// ===== GetReference (关联股票查询) =====
// C++ 流程 (APIServer_Qot_StockList::OnClientReq_Reference):
// - ReferenceType_Warrant(1): 本地 SecList 搜索该正股对应的窝轮列表 (INNBiz_Qot_SecList::SearchWarrantByOwnerID)
//   使用 StaticDataCache.owner_to_warrants 反向映射实现
// - ReferenceType_Future(2): 通过后端 CMD 6701 (PullFutureRelated) 拉取相关期货合约
//   C++ 中仅当输入为期货类型且 code 包含 "main" 时才发请求，否则返回空
struct GetReferenceHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetReferenceHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_reference::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;

        let ref_type = req.c2s.reference_type;
        let security = &req.c2s.security;
        let key = futu_cache::qot_cache::make_key(security.market, &security.code);

        tracing::debug!(
            conn_id,
            ref_type,
            code = %security.code,
            market = security.market,
            "GetReference"
        );

        // ReferenceType: 0=Unknow, 1=Warrant, 2=Future
        match ref_type {
            1 => {
                // Warrant: 从 static_cache 的 owner_to_warrants 映射中搜索
                let stock_info = self.static_cache.get_security_info(&key);
                let stock_id = stock_info.map(|i| i.stock_id).unwrap_or(0);

                let warrant_ids = if stock_id > 0 {
                    self.static_cache.search_warrants_by_owner(stock_id)
                } else {
                    Vec::new()
                };

                let static_info_list: Vec<futu_proto::qot_common::SecurityStaticInfo> = warrant_ids
                    .iter()
                    .filter_map(|&wid| {
                        let wkey = self.static_cache.id_to_key.get(&wid)?;
                        let winfo = self.static_cache.get_security_info(wkey.value())?;
                        // 过滤掉 market=0 (Unknown) 的窝轮，与 C++ 行为一致
                        if winfo.market == 0 {
                            return None;
                        }
                        Some(make_static_info(
                            futu_proto::qot_common::Security {
                                market: winfo.market,
                                code: winfo.code.clone(),
                            },
                            &winfo,
                        ))
                    })
                    .collect();

                tracing::debug!(
                    conn_id,
                    count = static_info_list.len(),
                    "GetReference(Warrant) returning securities"
                );

                let resp = futu_proto::qot_get_reference::Response {
                    ret_type: 0,
                    ret_msg: None,
                    err_code: None,
                    s2c: Some(futu_proto::qot_get_reference::S2c { static_info_list }),
                };
                Some(prost::Message::encode_to_vec(&resp))
            }
            2 => {
                // Future: 通过后端 CMD 6701 拉取相关期货合约
                // C++ 中仅当 sec_type 为 Future 且 code 包含 "main" 时才发请求
                let stock_info = self.static_cache.get_security_info(&key);
                let (stock_id, sec_type) = stock_info
                    .map(|i| (i.stock_id, i.sec_type))
                    .unwrap_or((0, 0));

                // sec_type 6 = SecurityType_Future (C++ NN_QuoteSecurityType_Future)
                // 非期货类型或非主连合约，直接返回空 (与 C++ 行为一致)
                if sec_type != 6 || !security.code.contains("main") {
                    tracing::debug!(
                        conn_id,
                        sec_type,
                        code = %security.code,
                        "GetReference(Future): not a main future contract, returning empty"
                    );
                    let resp = futu_proto::qot_get_reference::Response {
                        ret_type: 0,
                        ret_msg: None,
                        err_code: None,
                        s2c: Some(futu_proto::qot_get_reference::S2c {
                            static_info_list: Vec::new(),
                        }),
                    };
                    return Some(prost::Message::encode_to_vec(&resp));
                }

                let backend = match super::load_backend(&self.backend) {
                    Some(b) => b,
                    None => {
                        tracing::warn!(conn_id, "GetReference(Future): no backend connection");
                        return Some(super::make_error_response(-1, "Network interruption"));
                    }
                };

                // 发送 CMD 6701 到后端
                let backend_req =
                    futu_backend::proto_internal::ft_cmd_hp_plate::PlateUsFutrueRelatedListReq {
                        future_id: Some(stock_id),
                    };
                let body = prost::Message::encode_to_vec(&backend_req);

                let frame = match backend.request(6701, body).await {
                    Ok(f) => f,
                    Err(e) => {
                        tracing::error!(conn_id, error = %e, "CMD6701 request failed");
                        return Some(super::make_error_response(-1, "pull future related failed"));
                    }
                };

                let rsp: futu_backend::proto_internal::ft_cmd_hp_plate::PlateUsFutrueRelatedListRsp =
                    match prost::Message::decode(frame.body.as_ref()) {
                        Ok(r) => r,
                        Err(e) => {
                            tracing::error!(conn_id, error = %e, "CMD6701 decode failed");
                            return Some(super::make_error_response(
                                -1,
                                "decode future related response failed",
                            ));
                        }
                    };

                if rsp.result != 0 {
                    tracing::warn!(conn_id, result = rsp.result, "CMD6701 returned error");
                    return Some(super::make_error_response(
                        -1,
                        "backend future related query failed",
                    ));
                }

                // 提取 security_id 列表，查找 static_cache 构建 SecurityStaticInfo
                let static_info_list: Vec<futu_proto::qot_common::SecurityStaticInfo> = rsp
                    .security_qta_list
                    .iter()
                    .filter_map(|sq| {
                        let sid = sq.security_id?;
                        let skey = self.static_cache.id_to_key.get(&sid)?;
                        let sinfo = self.static_cache.get_security_info(skey.value())?;
                        Some(make_static_info(
                            futu_proto::qot_common::Security {
                                market: sinfo.market,
                                code: sinfo.code.clone(),
                            },
                            &sinfo,
                        ))
                    })
                    .collect();

                tracing::debug!(
                    conn_id,
                    count = static_info_list.len(),
                    "GetReference(Future) returning securities"
                );

                let resp = futu_proto::qot_get_reference::Response {
                    ret_type: 0,
                    ret_msg: None,
                    err_code: None,
                    s2c: Some(futu_proto::qot_get_reference::S2c { static_info_list }),
                };
                Some(prost::Message::encode_to_vec(&resp))
            }
            _ => {
                // ReferenceType_Unknow or invalid
                Some(super::make_error_response(-1, "unsupported reference type"))
            }
        }
    }
}

// ===== GetHoldingChangeList (已废弃, C++ 返回 NNData_StaticText_HoldingChangeNoMoreSupport) =====
struct GetHoldingChangeListHandler;

#[async_trait]
impl RequestHandler for GetHoldingChangeListHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let _req: futu_proto::qot_get_holding_change_list::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        // C++ 源码中此接口已被废弃，直接返回错误信息，不再查询后端数据。
        // 参见: APIServer_Qot_HoldingChange.cpp → strErrDesc = StcText(NNData_StaticText_HoldingChangeNoMoreSupport)
        tracing::debug!(
            conn_id,
            "GetHoldingChangeList: interface deprecated since 2020-12-21"
        );
        Some(super::make_error_response(
            -1,
            "Due to the reasons of the upstream data provider, the interface for Get Major Shareholders' Shareholding Changes (protocol ID:3208) will be abandoned after 2020-12-21",
        ))
    }
}

// ===== GetUserSecurity (CMD 5121 拉分组 + CMD 5120 拉股票列表) =====
struct GetUserSecurityHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
    app_lang: i32,
}

#[async_trait]
impl RequestHandler for GetUserSecurityHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_user_security::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let group_name = &req.c2s.group_name;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetUserSecurity: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

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

        // Step 1: 拉取分组信息 (CMD 5121)
        let group_req = futu_backend::proto_internal::wch_lst::GetGroupListReq {
            user_id: Some(user_id),
        };
        let group_body = prost::Message::encode_to_vec(&group_req);
        tracing::debug!(conn_id, %group_name, "GetUserSecurity: sending CMD5121");

        let group_frame = match backend.request(5121, group_body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD5121 request failed");
                return Some(super::make_error_response(-1, "pull group info failed"));
            }
        };

        // CMD5121 响应的 SRPC field 5 包含 repeated GroupInfo（不是 GetGroupListResp），
        // 需要使用专用解码函数
        let (group_rsp, group_langs) = super::decode_cmd5121_groups(group_frame.body.as_ref());

        if group_rsp.result_code.unwrap_or(-1) != 0 {
            tracing::warn!(conn_id, ret = ?group_rsp.result_code, "CMD5121 returned error");
            return Some(super::make_error_response(-1, "pull group info failed"));
        }

        // Step 2: 通过 groupName 找到 groupID
        // 优先匹配 multi_lang_name（按客户端语言），再匹配 group_name，最后用硬编码映射
        let app_lang = self.app_lang;
        let group_id = group_rsp
            .group_list
            .iter()
            .enumerate()
            .find(|(i, g)| {
                // 先检查 multi_lang_name
                if let Some(langs) = group_langs.get(*i) {
                    if langs
                        .iter()
                        .any(|ml| ml.language_id == app_lang && ml.name == group_name.as_str())
                    {
                        return true;
                    }
                }
                // 再检查 group_name
                g.group_name.as_deref() == Some(group_name.as_str())
            })
            .map(|(_, g)| g.group_id.unwrap_or(0))
            .or_else(|| system_group_name_to_id(group_name))
            .unwrap_or(0);
        if group_id == 0 {
            tracing::warn!(conn_id, %group_name, "GetUserSecurity: unknown group name");
            return Some(super::make_error_response(
                -1,
                "unknown user security group",
            ));
        }

        // 不支持的系统分组: 基金宝(890), 外汇(891), 持仓(896)
        if group_id == 890 || group_id == 891 || group_id == 896 {
            return Some(super::make_error_response(
                -1,
                "unsupported user security group",
            ));
        }

        // Step 3: 拉取该分组的股票列表 (CMD 5120)
        let stock_req = futu_backend::proto_internal::wch_lst::GetStockListReq {
            user_id: Some(user_id),
            group_id: Some(group_id),
            is_case_top: None,
        };
        let stock_body = prost::Message::encode_to_vec(&stock_req);
        tracing::info!(conn_id, group_id, "GetUserSecurity: sending CMD5120");

        let stock_frame = match backend.request(5120, stock_body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD5120 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        // CMD5120 响应可能被 SRPC 封装
        let body = stock_frame.body.as_ref();

        let stock_rsp = super::decode_srpc_or_direct::<
            futu_backend::proto_internal::wch_lst::GetStockListResp,
        >(body, |r| {
            r.result_code == Some(0) && r.stock_list.len() == r.stock_count.unwrap_or(0) as usize
        });

        tracing::info!(conn_id, result_code = ?stock_rsp.result_code,
            stock_list_len = stock_rsp.stock_list.len(),
            "CMD5120 response");

        if stock_rsp.result_code.unwrap_or(0) != 0 && stock_rsp.stock_list.is_empty() {
            return Some(super::make_error_response(-1, "pull stock list failed"));
        }

        // Step 4: stock_id -> SecurityStaticInfo
        let mut miss_id_count = 0u32;
        let mut miss_info_count = 0u32;
        let static_info_list: Vec<futu_proto::qot_common::SecurityStaticInfo> = stock_rsp
            .stock_list
            .iter()
            .filter_map(|si| {
                let stock_id = si.stock_id?;
                let key = match self.static_cache.id_to_key.get(&stock_id) {
                    Some(k) => k,
                    None => {
                        miss_id_count += 1;
                        if miss_id_count <= 3 {
                            tracing::debug!(stock_id, "GetUserSecurity: stock_id not in id_to_key");
                        }
                        return None;
                    }
                };
                let info = match self.static_cache.get_security_info(key.value()) {
                    Some(i) => i,
                    None => {
                        miss_info_count += 1;
                        return None;
                    }
                };
                let security = futu_proto::qot_common::Security {
                    market: info.market,
                    code: info.code.clone(),
                };
                Some(make_static_info(security, &info))
            })
            .collect();

        tracing::info!(
            conn_id,
            total = stock_rsp.stock_list.len(),
            found = static_info_list.len(),
            miss_id = miss_id_count,
            miss_info = miss_info_count,
            "GetUserSecurity result"
        );

        let resp = futu_proto::qot_get_user_security::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_user_security::S2c { static_info_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== ModifyUserSecurity (CMD 6682 修改自选股) =====
struct ModifyUserSecurityHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for ModifyUserSecurityHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_modify_user_security::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;
        let group_name = &c2s.group_name;
        let op = c2s.op;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "ModifyUserSecurity: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // 空股票列表时直接返回成功 (与 C++ 一致)
        if c2s.security_list.is_empty() {
            let resp = futu_proto::qot_modify_user_security::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::qot_modify_user_security::S2c {}),
            };
            return Some(prost::Message::encode_to_vec(&resp));
        }

        // Resolve stock IDs
        let mut stock_ids = Vec::new();
        for sec in &c2s.security_list {
            let sec_key = format!("{}_{}", sec.market, sec.code);
            if let Some(info) = self.static_cache.get_security_info(&sec_key) {
                if info.stock_id > 0 {
                    stock_ids.push(info.stock_id);
                }
            }
        }

        if stock_ids.is_empty() {
            tracing::warn!(conn_id, "ModifyUserSecurity: no valid stock_ids found");
            return Some(super::make_error_response(-1, "unknown stock"));
        }

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

        // 拉取分组信息 (CMD 5121)
        let group_req = futu_backend::proto_internal::wch_lst::GetGroupListReq {
            user_id: Some(user_id),
        };
        let group_body = prost::Message::encode_to_vec(&group_req);

        let group_frame = match backend.request(5121, group_body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD5121 request failed");
                return Some(super::make_error_response(-1, "pull group info failed"));
            }
        };

        // CMD5121 响应的 SRPC field 5 包含 repeated GroupInfo
        let (group_rsp, _group_langs) = super::decode_cmd5121_groups(group_frame.body.as_ref());

        if group_rsp.result_code.unwrap_or(-1) != 0 {
            return Some(super::make_error_response(-1, "pull group info failed"));
        }

        let group_id = match group_rsp
            .group_list
            .iter()
            .find(|g| g.group_name.as_deref() == Some(group_name.as_str()))
        {
            Some(g) => g.group_id.unwrap_or(0),
            None => {
                return Some(super::make_error_response(
                    -1,
                    "unknown user security group",
                ));
            }
        };

        // 系统分组不可修改
        let is_system = (group_id > 0 && group_id < 900) || group_id == 1000;
        if is_system {
            return Some(super::make_error_response(
                -1,
                "system group cannot be modified",
            ));
        }

        let action = match op {
            1 => 1u32,
            2 => 2u32,
            3 => 2u32, // MoveOut -> Del from this group only
            _ => {
                return Some(super::make_error_response(-1, "unknown modify operation"));
            }
        };

        // Add: 先添加到"全部"分组(1000)
        if op == 1 && group_id != 1000 {
            let all_req = futu_backend::proto_internal::wch_lst::SetStockReq {
                action: Some(1),
                group_id: Some(1000),
                stock_info: stock_ids
                    .iter()
                    .map(|&id| futu_backend::proto_internal::wch_lst::StockInfo {
                        stock_id: Some(id),
                        is_top: None,
                    })
                    .collect(),
            };
            let _ = backend
                .request(6682, prost::Message::encode_to_vec(&all_req))
                .await;
        }

        // 发送目标分组操作 (CMD 6682)
        let set_req = futu_backend::proto_internal::wch_lst::SetStockReq {
            action: Some(action),
            group_id: Some(group_id),
            stock_info: stock_ids
                .iter()
                .map(|&id| futu_backend::proto_internal::wch_lst::StockInfo {
                    stock_id: Some(id),
                    is_top: None,
                })
                .collect(),
        };

        tracing::debug!(
            conn_id,
            group_id,
            op,
            count = stock_ids.len(),
            "ModifyUserSecurity: sending CMD6682"
        );

        let set_frame = match backend
            .request(6682, prost::Message::encode_to_vec(&set_req))
            .await
        {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6682 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let set_rsp: futu_backend::proto_internal::wch_lst::SetStockResp =
            match prost::Message::decode(set_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6682 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if set_rsp.result_code.unwrap_or(-1) != 0 {
            let msg = if set_rsp.result_code == Some(3) {
                "user security total number exceeded"
            } else {
                "modify user security failed"
            };
            return Some(super::make_error_response(-1, msg));
        }

        // Del: 从其他所有非系统分组 + "全部"分组也删除 (与 C++ 一致)
        if op == 2 {
            let mut other_groups = vec![1000u32];
            for g in &group_rsp.group_list {
                let gid = g.group_id.unwrap_or(0);
                let g_is_system = (gid > 0 && gid < 900) || gid == 1000;
                if !g_is_system && gid != group_id {
                    other_groups.push(gid);
                }
            }
            for gid in other_groups {
                let del_req = futu_backend::proto_internal::wch_lst::SetStockReq {
                    action: Some(2),
                    group_id: Some(gid),
                    stock_info: stock_ids
                        .iter()
                        .map(|&id| futu_backend::proto_internal::wch_lst::StockInfo {
                            stock_id: Some(id),
                            is_top: None,
                        })
                        .collect(),
                };
                let _ = backend
                    .request(6682, prost::Message::encode_to_vec(&del_req))
                    .await;
            }
        }

        let resp = futu_proto::qot_modify_user_security::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_modify_user_security::S2c {}),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== StockFilter (条件选股, 后端 CMD 9010 Screen + CMD 9011 Retrieve) =====
// C++ 流程 (APIServer_Qot_StockFilter + NNBiz_Qot_StockFilter):
// 1. 将 FTAPI Qot_StockFilter 请求中的过滤条件转换为后端 FTCmdStockScreener::ScreenRequest
//    - 涉及 SimpleProperty, CumulativeProperty, FinancialProperty, PatternProperty,
//      CustomIndicator 等数百个字段的枚举映射
// 2. 发送 CMD 9010 (ScreenRequest) 到后端，获取符合条件的 stock_id 列表 (ScreenResponse)
// 3. 再发送 CMD 9011 (RetrieveRequest) 获取这些股票的详细数据 (RetrieveResponse)
// 4. 将后端返回的数据转换回 FTAPI StockData 格式
struct StockFilterHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for StockFilterHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_stock_filter::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "StockFilter: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // 参数校验 (与 C++ 一致)
        if c2s.begin < 0 {
            return Some(super::make_error_response(-1, "begin must be >= 0"));
        }
        if c2s.num < 0 {
            return Some(super::make_error_response(-1, "num must be >= 0"));
        }
        if c2s.num > 200 {
            return Some(super::make_error_response(-1, "num exceeds limit 200"));
        }

        let nn_market = stock_filter_market_api_to_nn(c2s.market);
        if nn_market == 0 {
            return Some(super::make_error_response(-1, "unsupported market"));
        }

        tracing::debug!(
            conn_id,
            market = c2s.market,
            begin = c2s.begin,
            num = c2s.num,
            base_filters = c2s.base_filter_list.len(),
            accumulate_filters = c2s.accumulate_filter_list.len(),
            financial_filters = c2s.financial_filter_list.len(),
            pattern_filters = c2s.pattern_filter_list.len(),
            custom_filters = c2s.custom_indicator_filter_list.len(),
            "StockFilter: building CMD 9010 ScreenRequest"
        );

        // ===== Step 1: 构建 CMD 9010 ScreenRequest =====
        let mut screen_req = futu_backend::proto_internal::ft_cmd_stock_screener::ScreenRequest {
            market: nn_market,
            ..Default::default()
        };

        // 板块过滤
        if let Some(plate) = &c2s.plate {
            if !plate.code.is_empty() {
                if let Some(plate_id) = stock_filter_security_to_id(&self.static_cache, plate) {
                    screen_req.categories.push(plate_id);
                }
            }
        }

        // 构建简单属性查询 (base_filter_list)
        stock_filter_build_base_queries(c2s, &mut screen_req);

        // 构建累积属性查询 (accumulate_filter_list)
        stock_filter_build_accumulate_queries(c2s, &mut screen_req);

        // 构建财务属性查询 (financial_filter_list)
        stock_filter_build_financial_queries(c2s, nn_market, &mut screen_req);

        // 构建形态指标查询 (pattern_filter_list)
        stock_filter_build_pattern_queries(c2s, &mut screen_req);

        // 构建自定义指标查询 (custom_indicator_filter_list)
        stock_filter_build_custom_queries(c2s, &mut screen_req);

        // 构建排序条件
        stock_filter_build_sort(c2s, &mut screen_req);

        // ===== Step 2: 发送 CMD 9010 =====
        let screen_body = prost::Message::encode_to_vec(&screen_req);
        tracing::debug!(
            conn_id,
            body_len = screen_body.len(),
            "sending CMD 9010 ScreenRequest"
        );

        let screen_frame = match backend.request(9010, screen_body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD 9010 request failed");
                return Some(super::make_error_response(
                    -1,
                    "backend screen request failed",
                ));
            }
        };

        let screen_rsp: futu_backend::proto_internal::ft_cmd_stock_screener::ScreenResponse =
            match prost::Message::decode(screen_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD 9010 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend screen response decode failed",
                    ));
                }
            };

        if screen_rsp.result_code != 0 {
            tracing::warn!(
                conn_id,
                result = screen_rsp.result_code,
                "CMD 9010 returned error"
            );
            return Some(super::make_error_response(
                -1,
                "backend screen request failed",
            ));
        }

        let all_stock_ids = &screen_rsp.stock_ids;
        let all_count = all_stock_ids.len() as i32;

        // 分页: 与 C++ 一致
        let real_begin = 0.max(c2s.begin.min(all_count)) as usize;
        let real_count = 0.max(c2s.num.min(all_count - real_begin as i32)) as usize;
        let last_page = (real_begin + real_count) >= all_count as usize;

        let paged_ids: Vec<u64> = if real_count > 0 {
            all_stock_ids[real_begin..real_begin + real_count].to_vec()
        } else {
            Vec::new()
        };

        if paged_ids.is_empty() {
            let resp = futu_proto::qot_stock_filter::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::qot_stock_filter::S2c {
                    last_page: true,
                    all_count,
                    data_list: Vec::new(),
                }),
            };
            return Some(prost::Message::encode_to_vec(&resp));
        }

        // ===== Step 3: 构建 CMD 9011 RetrieveRequest =====
        let mut retrieve_req =
            futu_backend::proto_internal::ft_cmd_stock_screener::RetrieveRequest {
                stock_ids: paged_ids.clone(),
                ..Default::default()
            };

        // 默认返回 code + name
        retrieve_req.basic_properties.push(
            futu_backend::proto_internal::ft_cmd_stock_screener::PropertyBasic {
                name: Some(1101), // PROPERTY_BASIC_INFO_CODE
            },
        );
        retrieve_req.basic_properties.push(
            futu_backend::proto_internal::ft_cmd_stock_screener::PropertyBasic {
                name: Some(1102), // PROPERTY_BASIC_INFO_NAME
            },
        );

        // 根据筛选条件添加 retrieve 属性 (与 C++ StructInnerToPb 对应)
        stock_filter_build_retrieve_properties(c2s, &mut retrieve_req);

        // ===== Step 4: 发送 CMD 9011 =====
        let retrieve_body = prost::Message::encode_to_vec(&retrieve_req);
        tracing::debug!(
            conn_id,
            body_len = retrieve_body.len(),
            stock_count = paged_ids.len(),
            "sending CMD 9011 RetrieveRequest"
        );

        let retrieve_frame = match backend.request(9011, retrieve_body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD 9011 request failed");
                return Some(super::make_error_response(
                    -1,
                    "backend retrieve request failed",
                ));
            }
        };

        let retrieve_rsp: futu_backend::proto_internal::ft_cmd_stock_screener::RetrieveResponse =
            match prost::Message::decode(retrieve_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD 9011 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend retrieve response decode failed",
                    ));
                }
            };

        if retrieve_rsp.result_code != 0 {
            tracing::warn!(
                conn_id,
                result = retrieve_rsp.result_code,
                "CMD 9011 returned error"
            );
            return Some(super::make_error_response(
                -1,
                "backend retrieve request failed",
            ));
        }

        // ===== Step 5: 将 RetrieveResponse 转换为 FTAPI StockData =====
        let data_list: Vec<futu_proto::qot_stock_filter::StockData> = retrieve_rsp
            .items
            .iter()
            .filter_map(|item| self.stock_filter_item_to_stock_data(item))
            .collect();

        tracing::debug!(
            conn_id,
            all_count,
            returned = data_list.len(),
            last_page,
            "StockFilter returning data"
        );

        let resp = futu_proto::qot_stock_filter::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_stock_filter::S2c {
                last_page,
                all_count,
                data_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

impl StockFilterHandler {
    /// 将 RetrieveResponse::Item 转换为 FTAPI StockData
    fn stock_filter_item_to_stock_data(
        &self,
        item: &futu_backend::proto_internal::ft_cmd_stock_screener::retrieve_response::Item,
    ) -> Option<futu_proto::qot_stock_filter::StockData> {
        let stock_id = item.stock_id;

        // 解析 security (market + code) 从 static_cache
        let security = self.resolve_security_from_stock_id(stock_id)?;

        // 从 basic_property_results 获取 name
        let name = item
            .basic_property_results
            .iter()
            .find(|r| r.property.name == Some(1102)) // PROPERTY_BASIC_INFO_NAME
            .map(|r| r.value.clone())
            .unwrap_or_default();

        // 转换简单属性数据 (simple → BaseData)
        let mut base_data_list = Vec::new();

        for r in &item.simple_property_results {
            let prop = &r.property;
            let nn_name = prop.name.unwrap_or(0);
            let api_field = stock_filter_simple_nn_to_api(nn_name);
            if api_field != 0 {
                let scale = stock_filter_base_scaling_ratio(nn_name);
                let value = if scale > 0.0 {
                    r.value as f64 / scale
                } else {
                    r.value as f64
                };
                base_data_list.push(futu_proto::qot_stock_filter::BaseData {
                    field_name: api_field,
                    value,
                });
            }
        }

        // 转换累积属性数据
        let accumulate_data_list: Vec<futu_proto::qot_stock_filter::AccumulateData> = item
            .cumulative_property_results
            .iter()
            .filter_map(|r| {
                let prop = &r.property;
                let nn_name = prop.name.unwrap_or(0);
                let api_field = stock_filter_accumulate_nn_to_api(nn_name);
                if api_field == 0 {
                    return None;
                }
                let scale = stock_filter_accumulate_scaling_ratio(nn_name);
                let value = if scale > 0.0 {
                    r.value as f64 / scale
                } else {
                    r.value as f64
                };
                let days = prop.days.unwrap_or(1) as i32;
                Some(futu_proto::qot_stock_filter::AccumulateData {
                    field_name: api_field,
                    value,
                    days,
                })
            })
            .collect();

        // 转换财务属性数据
        // 后端的 financial_property_results 中，"基础量价"字段映射为 BaseData，其余为 FinancialData
        let mut financial_data_list = Vec::new();
        for r in &item.financial_property_results {
            let prop = &r.property;
            let nn_name = prop.name.unwrap_or(0);
            let scale = stock_filter_financial_scaling_ratio(nn_name);
            let value = if scale > 0.0 {
                r.value as f64 / scale
            } else {
                r.value as f64
            };

            // 检查是否是 "基础量价" 字段 (映射为 BaseData)
            if let Some(api_base) = stock_filter_financial_to_base_api(nn_name) {
                base_data_list.push(futu_proto::qot_stock_filter::BaseData {
                    field_name: api_base,
                    value,
                });
            } else {
                let api_field = stock_filter_financial_nn_to_api(nn_name);
                if api_field != 0 {
                    let quarter = stock_filter_financial_quarter_nn_to_api(prop.term.unwrap_or(0));
                    financial_data_list.push(futu_proto::qot_stock_filter::FinancialData {
                        field_name: api_field,
                        value,
                        quarter,
                    });
                }
            }
        }

        // 转换自定义指标数据
        let custom_indicator_data_list: Vec<futu_proto::qot_stock_filter::CustomIndicatorData> =
            item.indicator_property_results
                .iter()
                .filter_map(|r| {
                    let prop = &r.property;
                    let nn_indicator = prop.name.unwrap_or(0);
                    let api_field = stock_filter_indicator_nn_to_custom_api(nn_indicator);
                    if api_field == 0 {
                        return None;
                    }
                    let scale = stock_filter_custom_scaling_ratio(nn_indicator);
                    let value = if scale > 0.0 {
                        r.value as f64 / scale
                    } else {
                        r.value as f64
                    };
                    let kl_type = stock_filter_period_nn_to_api(prop.period.unwrap_or(0));
                    let field_para_list: Vec<i32> =
                        prop.indicator_params.iter().map(|&p| p as i32).collect();
                    Some(futu_proto::qot_stock_filter::CustomIndicatorData {
                        field_name: api_field,
                        value,
                        kl_type,
                        field_para_list,
                    })
                })
                .collect();

        Some(futu_proto::qot_stock_filter::StockData {
            security,
            name,
            base_data_list,
            accumulate_data_list,
            financial_data_list,
            custom_indicator_data_list,
        })
    }

    fn resolve_security_from_stock_id(
        &self,
        stock_id: u64,
    ) -> Option<futu_proto::qot_common::Security> {
        let key = self.static_cache.id_to_key.get(&stock_id)?;
        let parts: Vec<&str> = key.split('_').collect();
        if parts.len() != 2 {
            return None;
        }
        let market: i32 = parts[0].parse().ok()?;
        let code = parts[1].to_string();
        Some(futu_proto::qot_common::Security { market, code })
    }
}

// ===== StockFilter 辅助函数: FTAPI market → 后端 FTCmdStockScreener::Market =====

fn stock_filter_market_api_to_nn(api_market: i32) -> i32 {
    match api_market {
        1 | 12 => 1,  // HK_Security / HK_Future → MKT_HK
        11 => 2,      // US_Security → MKT_US
        21 | 22 => 3, // CNSH_Security / CNSZ_Security → MKT_CN
        _ => 0,       // MKT_UNKNOWN
    }
}

fn stock_filter_security_to_id(
    cache: &StaticDataCache,
    sec: &futu_proto::qot_common::Security,
) -> Option<u64> {
    let key = format!("{}_{}", sec.market, sec.code);
    cache.id_to_key.iter().find_map(|entry| {
        if *entry.value() == key {
            Some(*entry.key())
        } else {
            None
        }
    })
}

// ===== 简单属性字段映射 (FTAPI StockField → 后端 PropertyNameSimple) =====

fn stock_filter_base_api_to_nn_simple(api_field: i32) -> i32 {
    match api_field {
        3 => 2201,  // CurPrice → PROPERTY_SIMPLE_QUOTE_PRICE
        4 => 2209,  // CurPriceToHighest52WeeksRatio
        5 => 2210,  // CurPriceToLowest52WeeksRatio
        6 => 2211,  // HighPriceToHighest52WeeksRatio
        7 => 2212,  // LowPriceToLowest52WeeksRatio
        8 => 2217,  // VolumeRatio
        9 => 2218,  // BidAskRatio
        10 => 2219, // LotPrice
        11 => 2301, // MarketVal
        12 => 2302, // PeAnnual
        13 => 2303, // PeTTM
        14 => 2304, // PbRate
        15 => 2213, // ChangeRate5min
        16 => 2214, // ChangeRateBeginYear
        _ => 0,
    }
}

/// "基础量价" — FTAPI StockField 中映射到后端 financial 的字段
fn stock_filter_base_api_to_nn_financial(api_field: i32) -> i32 {
    match api_field {
        17 => 4904, // PSTTM → PROPERTY_FINANCIAL_PS_TTM
        18 => 4905, // PCFTTM → PROPERTY_FINANCIAL_PRICE_CASH_FLOW_RATIO_TTM
        19 => 4901, // TotalShare → PROPERTY_FINANCIAL_TOTAL_SHARE
        20 => 4902, // FloatShare → PROPERTY_FINANCIAL_FLOAT_SHARE
        21 => 4903, // FloatMarketVal → PROPERTY_FINANCIAL_FLOAT_MARKET_CAP
        _ => 0,
    }
}

fn stock_filter_is_base_price_field(api_field: i32) -> bool {
    matches!(api_field, 17..=21)
}

fn stock_filter_simple_nn_to_api(nn_name: i32) -> i32 {
    match nn_name {
        2201 => 3,  // QUOTE_PRICE → CurPrice
        2209 => 4,  // CurPriceToHighest52WeeksRatio
        2210 => 5,  // CurPriceToLowest52WeeksRatio
        2211 => 6,  // HighPriceToHighest52WeeksRatio
        2212 => 7,  // LowPriceToLowest52WeeksRatio
        2217 => 8,  // VolumeRatio
        2218 => 9,  // BidAskRatio
        2219 => 10, // LotPrice
        2301 => 11, // MarketVal
        2302 => 12, // PeAnnual
        2303 => 13, // PeTTM
        2304 => 14, // PbRate
        2213 => 15, // ChangeRate5min
        2214 => 16, // ChangeRateBeginYear
        _ => 0,
    }
}

fn stock_filter_base_scaling_ratio(nn_name: i32) -> f64 {
    match nn_name {
        2201 | 2209 | 2210 | 2211 | 2212 | 2213 | 2214 | 2219 | 2301 => 1000.0,
        2217 | 2218 | 2302 | 2303 | 2304 => 100_000.0,
        _ => 1.0,
    }
}

// ===== 累积属性字段映射 =====

fn stock_filter_accumulate_api_to_nn(api_field: i32) -> i32 {
    match api_field {
        1 => 3102, // ChangeRate → CUMULATIVE_QUOTE_PRICE_CHANGE_PERCENTAGE
        2 => 3103, // Amplitude
        3 => 3104, // Volume
        4 => 3105, // Turnover
        5 => 3106, // TurnoverRate
        _ => 0,
    }
}

fn stock_filter_accumulate_nn_to_api(nn_name: i32) -> i32 {
    match nn_name {
        3102 => 1, // ChangeRate
        3103 => 2, // Amplitude
        3104 => 3, // Volume
        3105 => 4, // Turnover
        3106 => 5, // TurnoverRate
        _ => 0,
    }
}

fn stock_filter_accumulate_scaling_ratio(nn_name: i32) -> f64 {
    match nn_name {
        3102 | 3103 | 3105 | 3106 => 1000.0,
        3104 => 1.0,
        _ => 1.0,
    }
}

// ===== 财务属性字段映射 =====

fn stock_filter_financial_api_to_nn(api_field: i32) -> i32 {
    match api_field {
        1 => 4101,  // NetProfit
        2 => 4102,  // NetProfitGrowth
        3 => 4105,  // SumOfBusiness → Revenue
        4 => 4106,  // SumOfBusinessGrowth → RevenueGrowth
        5 => 4107,  // NetProfitRate
        6 => 4108,  // GrossProfitRate
        7 => 4109,  // DebtAssetsRate
        8 => 4110,  // ReturnOnEquityRate
        9 => 4202,  // ROIC
        10 => 4209, // ROATTM
        11 => 4206, // EBITTTM
        12 => 4201, // EBITDA
        13 => 4210, // OperatingMarginTTM
        14 => 4203, // EBITMargin
        15 => 4204, // EBITDAMargin
        16 => 4205, // FinancialCostRate
        17 => 4207, // OperatingProfitTTM
        18 => 4208, // ShareholderNetProfitTTM
        19 => 4211, // NetProfitCashCoverTTM
        20 => 4301, // CurrentRatio
        21 => 4302, // QuickRatio
        22 => 4402, // CurrentAssetRatio
        23 => 4403, // CurrentDebtRatio
        24 => 4404, // EquityMultiplier
        25 => 4405, // PropertyRatio
        26 => 4401, // CashAndCashEquivalents
        27 => 4502, // TotalAssetTurnover
        28 => 4503, // FixedAssetTurnover
        29 => 4504, // InventoryTurnover
        30 => 4505, // OperatingCashFlowTTM
        31 => 4501, // AccountsReceivable
        32 => 4601, // EBITGrowthRate
        33 => 4602, // OperatingProfitGrowthRate
        34 => 4603, // TotalAssetsGrowthRate
        35 => 4604, // ProfitToShareholdersGrowthRate
        36 => 4605, // ProfitBeforeTaxGrowthRate
        37 => 4606, // EPSGrowthRate
        38 => 4607, // ROEGrowthRate
        39 => 4608, // ROICGrowthRate
        40 => 4609, // NOCFGrowthRate
        41 => 4610, // NOCFPerShareGrowthRate
        42 => 4701, // OperatingRevenueCashCover
        43 => 4702, // OperatingProfitToTotalProfit
        44 => 4801, // BasicEPS
        45 => 4802, // DilutedEPS
        46 => 4803, // NOCFPerShare
        _ => 0,
    }
}

fn stock_filter_financial_nn_to_api(nn_name: i32) -> i32 {
    match nn_name {
        4101 => 1,
        4102 => 2,
        4105 => 3,
        4106 => 4,
        4107 => 5,
        4108 => 6,
        4109 => 7,
        4110 => 8,
        4202 => 9,
        4209 => 10,
        4206 => 11,
        4201 => 12,
        4210 => 13,
        4203 => 14,
        4204 => 15,
        4205 => 16,
        4207 => 17,
        4208 => 18,
        4211 => 19,
        4301 => 20,
        4302 => 21,
        4402 => 22,
        4403 => 23,
        4404 => 24,
        4405 => 25,
        4401 => 26,
        4502 => 27,
        4503 => 28,
        4504 => 29,
        4505 => 30,
        4501 => 31,
        4601 => 32,
        4602 => 33,
        4603 => 34,
        4604 => 35,
        4605 => 36,
        4606 => 37,
        4607 => 38,
        4608 => 39,
        4609 => 40,
        4610 => 41,
        4701 => 42,
        4702 => 43,
        4801 => 44,
        4802 => 45,
        4803 => 46,
        _ => 0,
    }
}

/// 后端 PropertyNameFinancial 中 "基础量价" 字段 → FTAPI StockField
/// 这些后端字段是 financial，但 FTAPI 中归类为 BaseData
fn stock_filter_financial_to_base_api(nn_name: i32) -> Option<i32> {
    match nn_name {
        4901 => Some(19), // TotalShare
        4902 => Some(20), // FloatShare
        4903 => Some(21), // FloatMarketVal
        4904 => Some(17), // PSTTM
        4905 => Some(18), // PCFTTM
        _ => None,
    }
}

fn stock_filter_financial_scaling_ratio(nn_name: i32) -> f64 {
    match nn_name {
        4404 => 100_000.0, // EquityMultiplier
        _ => 1000.0,       // 绝大多数财务字段使用 1000
    }
}

// ===== 财务季度映射 =====

fn stock_filter_financial_quarter_api_to_nn(api_quarter: i32) -> i32 {
    match api_quarter {
        1 => 100, // Annual → TERM_ANNUAL
        2 => 1,   // FirstQuarter → TERM_Q1
        3 => 6,   // Interim → TERM_Q6
        4 => 9,   // ThirdQuarter → TERM_Q9
        5 => 10,  // MostRecentQuarter → TERM_Q_LATEST
        _ => 0,
    }
}

fn stock_filter_financial_quarter_nn_to_api(nn_term: i32) -> i32 {
    match nn_term {
        100 => 1, // Annual
        1 => 2,   // Q1
        6 => 3,   // Interim
        9 => 4,   // ThirdQuarter
        10 => 5,  // MostRecentQuarter
        _ => 0,
    }
}

fn stock_filter_is_annual_only(api_field: i32) -> bool {
    matches!(api_field, 10 | 11 | 13 | 17 | 18 | 19 | 30)
}

// ===== 形态指标映射 =====

fn stock_filter_pattern_api_to_nn(api_field: i32) -> i32 {
    match api_field {
        1 => 1,   // MAAlignmentLong
        2 => 2,   // MAAlignmentShort
        3 => 3,   // EMAAlignmentLong
        4 => 4,   // EMAAlignmentShort
        5 => 31,  // RSIGoldCrossLow
        6 => 32,  // RSIDeathCrossHigh
        7 => 33,  // RSITopDivergence
        8 => 34,  // RSIBottomDivergence
        9 => 11,  // KDJGoldCrossLow
        10 => 12, // KDJDeathCrossHigh
        11 => 13, // KDJTopDivergence
        12 => 14, // KDJBottomDivergence
        13 => 21, // MACDGoldCrossLow
        14 => 22, // MACDDeathCrossHigh
        15 => 23, // MACDTopDivergence
        16 => 24, // MACDBottomDivergence
        17 => 41, // BOLLBreakUpper
        18 => 42, // BOLLBreakLower
        19 => 43, // BOLLCrossMiddleUp
        20 => 44, // BOLLCrossMiddleDown
        _ => 0,
    }
}

// ===== 自定义指标映射 =====

fn stock_filter_custom_api_to_nn(api_field: i32) -> i32 {
    match api_field {
        1 => 1,   // Price → INDICATOR_PRICE
        2 => 11,  // MA5
        3 => 12,  // MA10
        4 => 13,  // MA20
        5 => 14,  // MA30
        6 => 15,  // MA60
        7 => 16,  // MA120
        8 => 17,  // MA250
        9 => 52,  // RSI → INDICATOR_RSI (动态)
        10 => 21, // EMA5
        11 => 22, // EMA10
        12 => 23, // EMA20
        13 => 24, // EMA30
        14 => 25, // EMA60
        15 => 26, // EMA120
        16 => 27, // EMA250
        17 => 0,  // Value (特殊处理)
        30 => 18, // MA (动态)
        40 => 28, // EMA (动态)
        50 => 34, // KDJ_K (动态)
        51 => 35, // KDJ_D
        52 => 36, // KDJ_J
        60 => 44, // MACD_DIFF (动态)
        61 => 45, // MACD_DEA
        62 => 46, // MACD_MACD
        70 => 64, // BOLL_UPPER (动态)
        71 => 65, // BOLL_MIDDLER
        72 => 66, // BOLL_LOWER
        _ => 0,
    }
}

fn stock_filter_indicator_nn_to_custom_api(nn_indicator: i32) -> i32 {
    match nn_indicator {
        1 => 1,        // PRICE
        11 => 2,       // MA5
        12 => 3,       // MA10
        13 => 4,       // MA20
        14 => 5,       // MA30
        15 => 6,       // MA60
        16 => 7,       // MA120
        17 => 8,       // MA250
        18 => 30,      // MA (动态)
        21 => 10,      // EMA5
        22 => 11,      // EMA10
        23 => 12,      // EMA20
        24 => 13,      // EMA30
        25 => 14,      // EMA60
        26 => 15,      // EMA120
        27 => 16,      // EMA250
        28 => 40,      // EMA (动态)
        31 | 34 => 50, // KDJ_K
        32 | 35 => 51, // KDJ_D
        33 | 36 => 52, // KDJ_J
        41 | 44 => 60, // MACD_DIFF
        42 | 45 => 61, // MACD_DEA
        43 | 46 => 62, // MACD_MACD
        51 | 52 => 9,  // RSI
        61 | 64 => 70, // BOLL_UPPER
        62 | 65 => 71, // BOLL_MIDDLER
        63 | 66 => 72, // BOLL_LOWER
        _ => 0,
    }
}

fn stock_filter_custom_scaling_ratio(nn_indicator: i32) -> f64 {
    match nn_indicator {
        0 => 1.0,
        _ => 1000.0,
    }
}

// ===== 相对位置 / KL周期 / 排序方向映射 =====

fn stock_filter_relative_position_api_to_nn(api_pos: i32) -> i32 {
    match api_pos {
        1 => 1, // More → POSITION_OVER
        2 => 2, // Less → POSITION_BELOW
        3 => 3, // CrossUp
        4 => 4, // CrossDown
        _ => 0,
    }
}

fn stock_filter_kl_type_api_to_nn(api_kl: i32) -> i32 {
    match api_kl {
        6 => 5,   // KLType_60Min → PERIOD_HOUR_1
        8 => 11,  // KLType_Day → PERIOD_DAY_1
        9 => 21,  // KLType_Week → PERIOD_WEEK_1
        10 => 31, // KLType_Month → PERIOD_MONTH_1
        _ => 0,
    }
}

fn stock_filter_period_nn_to_api(nn_period: i32) -> i32 {
    match nn_period {
        5 => 6,   // KLType_60Min
        11 => 8,  // KLType_Day
        21 => 9,  // KLType_Week
        31 => 10, // KLType_Month
        _ => 0,
    }
}

fn stock_filter_sort_dir_api_to_nn(api_dir: i32) -> i32 {
    match api_dir {
        1 => 1, // Ascend
        2 => 2, // Descend
        _ => 0,
    }
}

// ===== ScreenRequest 构建函数 =====

fn stock_filter_make_boundary(
    value: f64,
    scale: f64,
) -> futu_backend::proto_internal::ft_cmd_stock_screener::Bounary {
    let scaled = if scale > 0.0 {
        (value * scale) as i64
    } else {
        value as i64
    };
    futu_backend::proto_internal::ft_cmd_stock_screener::Bounary {
        value: scaled,
        includes: true,
    }
}

fn stock_filter_build_base_queries(
    c2s: &futu_proto::qot_stock_filter::C2s,
    screen_req: &mut futu_backend::proto_internal::ft_cmd_stock_screener::ScreenRequest,
) {
    for f in &c2s.base_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let api_field = f.field_name;
        if api_field == 1 || api_field == 2 {
            continue;
        }

        if stock_filter_is_base_price_field(api_field) {
            let nn_fin = stock_filter_base_api_to_nn_financial(api_field);
            if nn_fin == 0 {
                continue;
            }
            let scale = stock_filter_financial_scaling_ratio(nn_fin);
            let mut query =
                futu_backend::proto_internal::ft_cmd_stock_screener::QueryPropertyFinancial {
                    property:
                        futu_backend::proto_internal::ft_cmd_stock_screener::PropertyFinancial {
                            name: Some(nn_fin),
                            term: None,
                            duration: None,
                            year: None,
                            period_average: None,
                        },
                    ..Default::default()
                };
            if let Some(min) = f.filter_min {
                query.lower = Some(stock_filter_make_boundary(min, scale));
            }
            if let Some(max) = f.filter_max {
                query.upper = Some(stock_filter_make_boundary(max, scale));
            }
            screen_req.financial_property_queries.push(query);
        } else {
            let nn_simple = stock_filter_base_api_to_nn_simple(api_field);
            if nn_simple == 0 {
                continue;
            }
            let scale = stock_filter_base_scaling_ratio(nn_simple);
            let mut query =
                futu_backend::proto_internal::ft_cmd_stock_screener::QueryPropertySimple {
                    property: futu_backend::proto_internal::ft_cmd_stock_screener::PropertySimple {
                        name: Some(nn_simple),
                    },
                    ..Default::default()
                };
            if let Some(min) = f.filter_min {
                query.lower = Some(stock_filter_make_boundary(min, scale));
            }
            if let Some(max) = f.filter_max {
                query.upper = Some(stock_filter_make_boundary(max, scale));
            }
            screen_req.simple_property_queries.push(query);
        }
    }
}

fn stock_filter_build_accumulate_queries(
    c2s: &futu_proto::qot_stock_filter::C2s,
    screen_req: &mut futu_backend::proto_internal::ft_cmd_stock_screener::ScreenRequest,
) {
    for f in &c2s.accumulate_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let nn_field = stock_filter_accumulate_api_to_nn(f.field_name);
        if nn_field == 0 {
            continue;
        }
        let scale = stock_filter_accumulate_scaling_ratio(nn_field);
        let mut query =
            futu_backend::proto_internal::ft_cmd_stock_screener::QueryPropertyCumulative {
                property: futu_backend::proto_internal::ft_cmd_stock_screener::PropertyCumulative {
                    name: Some(nn_field),
                    days: Some(f.days as u32),
                    period_average: None,
                },
                ..Default::default()
            };
        if let Some(min) = f.filter_min {
            query.lower = Some(stock_filter_make_boundary(min, scale));
        }
        if let Some(max) = f.filter_max {
            query.upper = Some(stock_filter_make_boundary(max, scale));
        }
        screen_req.cumulative_property_queries.push(query);
    }
}

fn stock_filter_build_financial_queries(
    c2s: &futu_proto::qot_stock_filter::C2s,
    nn_market: i32,
    screen_req: &mut futu_backend::proto_internal::ft_cmd_stock_screener::ScreenRequest,
) {
    for f in &c2s.financial_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let nn_field = stock_filter_financial_api_to_nn(f.field_name);
        if nn_field == 0 {
            continue;
        }
        let scale = stock_filter_financial_scaling_ratio(nn_field);

        let nn_term = if stock_filter_is_annual_only(f.field_name) {
            100
        } else {
            let q = stock_filter_financial_quarter_api_to_nn(f.quarter);
            if q == 0 {
                continue;
            }
            if (nn_market == 1 || nn_market == 3) && q == 10 {
                continue;
            }
            if nn_market == 2 && matches!(q, 1 | 6 | 9) {
                continue;
            }
            q
        };

        let mut query =
            futu_backend::proto_internal::ft_cmd_stock_screener::QueryPropertyFinancial {
                property: futu_backend::proto_internal::ft_cmd_stock_screener::PropertyFinancial {
                    name: Some(nn_field),
                    term: Some(nn_term),
                    duration: None,
                    year: None,
                    period_average: None,
                },
                ..Default::default()
            };
        if let Some(min) = f.filter_min {
            query.lower = Some(stock_filter_make_boundary(min, scale));
        }
        if let Some(max) = f.filter_max {
            query.upper = Some(stock_filter_make_boundary(max, scale));
        }
        screen_req.financial_property_queries.push(query);
    }
}

fn stock_filter_build_pattern_queries(
    c2s: &futu_proto::qot_stock_filter::C2s,
    screen_req: &mut futu_backend::proto_internal::ft_cmd_stock_screener::ScreenRequest,
) {
    for f in &c2s.pattern_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let nn_pattern = stock_filter_pattern_api_to_nn(f.field_name);
        if nn_pattern == 0 {
            continue;
        }
        let nn_period = stock_filter_kl_type_api_to_nn(f.kl_type);
        let mut query =
            futu_backend::proto_internal::ft_cmd_stock_screener::QueryIndicatorPattern {
                pattern: nn_pattern,
                period: nn_period,
                continuous_period: None,
            };
        if let Some(cp) = f.consecutive_period {
            if cp > 0 {
                query.continuous_period = Some(cp);
            }
        }
        screen_req.indicator_pattern_queries.push(query);
    }
}

fn stock_filter_build_custom_queries(
    c2s: &futu_proto::qot_stock_filter::C2s,
    screen_req: &mut futu_backend::proto_internal::ft_cmd_stock_screener::ScreenRequest,
) {
    for f in &c2s.custom_indicator_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let nn_first = stock_filter_custom_api_to_nn(f.first_field_name);
        if nn_first == 0 {
            continue;
        }
        let nn_second_raw = f.second_field_name;
        let nn_second = stock_filter_custom_api_to_nn(nn_second_raw);
        let nn_position = stock_filter_relative_position_api_to_nn(f.relative_position);
        let nn_period = stock_filter_kl_type_api_to_nn(f.kl_type);
        let scale = stock_filter_custom_scaling_ratio(nn_first);

        let mut query =
            futu_backend::proto_internal::ft_cmd_stock_screener::QueryIndicatorPositional {
                position: nn_position,
                period: nn_period,
                first_indicator: nn_first,
                second_indicator: None,
                second_value: None,
                first_indicator_params: f.first_field_para_list.iter().map(|&v| v as i64).collect(),
                second_indicator_params: f
                    .second_field_para_list
                    .iter()
                    .map(|&v| v as i64)
                    .collect(),
                continuous_period: None,
            };

        if nn_second_raw != 17 && nn_second_raw != 0 && nn_second != 0 {
            query.second_indicator = Some(nn_second);
        }
        if nn_second_raw == 17 {
            if let Some(val) = f.field_value {
                query.second_value = Some((val * scale) as i64);
            }
        }
        if let Some(cp) = f.consecutive_period {
            if cp > 0 {
                query.continuous_period = Some(cp);
            }
        }
        screen_req.indicator_positional_queries.push(query);
    }
}

fn stock_filter_build_sort(
    c2s: &futu_proto::qot_stock_filter::C2s,
    screen_req: &mut futu_backend::proto_internal::ft_cmd_stock_screener::ScreenRequest,
) {
    // base_filter_list
    for f in &c2s.base_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let dir = f.sort_dir.unwrap_or(0);
        let nn_dir = stock_filter_sort_dir_api_to_nn(dir);
        if nn_dir == 0 {
            continue;
        }
        let api_field = f.field_name;
        if api_field == 1 || api_field == 2 {
            continue;
        }
        let mut sort = futu_backend::proto_internal::ft_cmd_stock_screener::Sort {
            direction: nn_dir,
            ..Default::default()
        };
        if stock_filter_is_base_price_field(api_field) {
            sort.financial_property = Some(
                futu_backend::proto_internal::ft_cmd_stock_screener::PropertyFinancial {
                    name: Some(stock_filter_base_api_to_nn_financial(api_field)),
                    term: None,
                    duration: None,
                    year: None,
                    period_average: None,
                },
            );
        } else {
            sort.simple_property = Some(
                futu_backend::proto_internal::ft_cmd_stock_screener::PropertySimple {
                    name: Some(stock_filter_base_api_to_nn_simple(api_field)),
                },
            );
        }
        screen_req.sort = Some(sort);
        return;
    }

    // accumulate_filter_list
    for f in &c2s.accumulate_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let dir = f.sort_dir.unwrap_or(0);
        let nn_dir = stock_filter_sort_dir_api_to_nn(dir);
        if nn_dir == 0 {
            continue;
        }
        let nn_field = stock_filter_accumulate_api_to_nn(f.field_name);
        if nn_field == 0 {
            continue;
        }
        screen_req.sort = Some(futu_backend::proto_internal::ft_cmd_stock_screener::Sort {
            direction: nn_dir,
            cumulative_property: Some(
                futu_backend::proto_internal::ft_cmd_stock_screener::PropertyCumulative {
                    name: Some(nn_field),
                    days: Some(f.days as u32),
                    period_average: None,
                },
            ),
            ..Default::default()
        });
        return;
    }

    // financial_filter_list
    for f in &c2s.financial_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let dir = f.sort_dir.unwrap_or(0);
        let nn_dir = stock_filter_sort_dir_api_to_nn(dir);
        if nn_dir == 0 {
            continue;
        }
        let nn_field = stock_filter_financial_api_to_nn(f.field_name);
        if nn_field == 0 {
            continue;
        }
        let nn_term = if stock_filter_is_annual_only(f.field_name) {
            100
        } else {
            stock_filter_financial_quarter_api_to_nn(f.quarter)
        };
        screen_req.sort = Some(futu_backend::proto_internal::ft_cmd_stock_screener::Sort {
            direction: nn_dir,
            financial_property: Some(
                futu_backend::proto_internal::ft_cmd_stock_screener::PropertyFinancial {
                    name: Some(nn_field),
                    term: Some(nn_term),
                    duration: None,
                    year: None,
                    period_average: None,
                },
            ),
            ..Default::default()
        });
        return;
    }
}

// ===== RetrieveRequest 属性构建 =====

fn stock_filter_build_retrieve_properties(
    c2s: &futu_proto::qot_stock_filter::C2s,
    retrieve_req: &mut futu_backend::proto_internal::ft_cmd_stock_screener::RetrieveRequest,
) {
    for f in &c2s.base_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let api_field = f.field_name;
        if api_field == 1 || api_field == 2 {
            continue;
        }
        if stock_filter_is_base_price_field(api_field) {
            let nn_fin = stock_filter_base_api_to_nn_financial(api_field);
            if nn_fin != 0 {
                retrieve_req.financial_properties.push(
                    futu_backend::proto_internal::ft_cmd_stock_screener::PropertyFinancial {
                        name: Some(nn_fin),
                        term: None,
                        duration: None,
                        year: None,
                        period_average: None,
                    },
                );
            }
        } else {
            let nn_simple = stock_filter_base_api_to_nn_simple(api_field);
            if nn_simple != 0 {
                retrieve_req.simple_properties.push(
                    futu_backend::proto_internal::ft_cmd_stock_screener::PropertySimple {
                        name: Some(nn_simple),
                    },
                );
            }
        }
    }

    for f in &c2s.accumulate_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let nn_field = stock_filter_accumulate_api_to_nn(f.field_name);
        if nn_field != 0 {
            retrieve_req.cumulative_properties.push(
                futu_backend::proto_internal::ft_cmd_stock_screener::PropertyCumulative {
                    name: Some(nn_field),
                    days: Some(f.days as u32),
                    period_average: None,
                },
            );
        }
    }

    for f in &c2s.financial_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        let nn_field = stock_filter_financial_api_to_nn(f.field_name);
        if nn_field != 0 {
            let nn_term = if stock_filter_is_annual_only(f.field_name) {
                100
            } else {
                stock_filter_financial_quarter_api_to_nn(f.quarter)
            };
            retrieve_req.financial_properties.push(
                futu_backend::proto_internal::ft_cmd_stock_screener::PropertyFinancial {
                    name: Some(nn_field),
                    term: Some(nn_term),
                    duration: None,
                    year: None,
                    period_average: None,
                },
            );
        }
    }

    // custom_indicator_filter_list → indicator_properties (去重)
    let mut seen_indicators: std::collections::HashSet<(i32, i32, Vec<i32>)> =
        std::collections::HashSet::new();
    for f in &c2s.custom_indicator_filter_list {
        if f.is_no_filter.unwrap_or(true) {
            continue;
        }
        if f.consecutive_period.unwrap_or(0) > 1 {
            continue;
        }
        let nn_period = stock_filter_kl_type_api_to_nn(f.kl_type);

        let nn_first = stock_filter_custom_api_to_nn(f.first_field_name);
        if nn_first != 0 {
            let params: Vec<i32> = f.first_field_para_list.clone();
            let key = (nn_first, nn_period, params.clone());
            if seen_indicators.insert(key) {
                retrieve_req.indicator_properties.push(
                    futu_backend::proto_internal::ft_cmd_stock_screener::PropertyIndicator {
                        name: Some(nn_first),
                        period: Some(nn_period),
                        indicator_params: params.iter().map(|&v| v as i64).collect(),
                    },
                );
            }
        }

        let nn_second_raw = f.second_field_name;
        if nn_second_raw != 17 && nn_second_raw != 0 {
            let nn_second = stock_filter_custom_api_to_nn(nn_second_raw);
            if nn_second != 0 {
                let params: Vec<i32> = f.second_field_para_list.clone();
                let key = (nn_second, nn_period, params.clone());
                if seen_indicators.insert(key) {
                    retrieve_req.indicator_properties.push(
                        futu_backend::proto_internal::ft_cmd_stock_screener::PropertyIndicator {
                            name: Some(nn_second),
                            period: Some(nn_period),
                            indicator_params: params.iter().map(|&v| v as i64).collect(),
                        },
                    );
                }
            }
        }
    }
}

// ===== GetCodeChange (HTTP 数据源 → 本地缓存 + 条件过滤) =====
//
// C++ 流程 (APIServer_Qot_CodeChange::OnClientReq_CodeChange):
// 1. 数据源: 通过 HTTP 从腾讯云 CDN 下载 JSON 数据:
//    - CodeRelation: 创业板转主板 (CodeDefine=202 → GemToMain)
//    - CodeTemp: 临时代码 (EventType 1-7 → Unpaid/ChangeLot/Split/Joint/JointSplit/SplitJoint/Other)
// 2. 查询: 从本地缓存获取全部 CodeChangeInfo, 按 C2S 条件过滤:
//    - security_list: 按股票代码筛选
//    - type_list: 按 CodeChangeType 筛选
//    - time_filter_list: 按 Public/Effective/End 时间范围筛选
struct GetCodeChangeHandler {
    code_change_cache: futu_backend::code_change::CodeChangeCache,
}

/// C++ IsFitCond: 检查 CodeChangeInfo 是否匹配 C2S 过滤条件
fn code_change_fits_condition(
    info: &futu_backend::code_change::CodeChangeInfo,
    c2s: &futu_proto::qot_get_code_change::C2s,
) -> bool {
    // C++: If_OMWarn_Return(stNN.enType == CodeChangeType_Unkonw, false)
    if info.change_type == futu_backend::code_change::CodeChangeType::Unknown {
        return false;
    }

    // 股票筛选: C++ 遍历 securitylist, 如果任意一个不匹配就 return false
    for sec in &c2s.security_list {
        if sec.market != info.qot_market || sec.code != info.sec_code {
            return false;
        }
    }

    // 类型筛选: C++ 遍历 typelist, 如果任意一个不匹配就 return false
    for &type_val in &c2s.type_list {
        if type_val != info.change_type as i32 {
            return false;
        }
    }

    // 时间筛选
    for tf in &c2s.time_filter_list {
        let has_begin = tf.begin_time.is_some();
        let has_end = tf.end_time.is_some();
        if !has_begin && !has_end {
            continue;
        }

        // 根据 TimeFilterType 选择对应的时间字段
        let time_val = match tf.r#type {
            1 => info.public_time,    // TimeFilterType_Public
            2 => info.effective_time, // TimeFilterType_Effective
            3 => info.end_time,       // TimeFilterType_End
            _ => continue,
        };
        if time_val == 0 {
            continue;
        }

        // CodeChange 的时间过滤默认使用东八区 (market=1 HK)
        let filter_market = c2s.security_list.first().map(|s| s.market).unwrap_or(1);
        if let Some(ref begin_str) = tf.begin_time {
            let begin_ts = parse_api_time_str(begin_str, filter_market).unwrap_or(0);
            if begin_ts > 0 && time_val < begin_ts {
                return false;
            }
        }

        if let Some(ref end_str) = tf.end_time {
            // C++: nEndTime = GetTimeCache(...) + M_OneDaySecs - 1
            // 即结束时间包含整天 (加上 86400-1 秒)
            let end_ts = parse_api_time_str(end_str, filter_market).unwrap_or(0) + 86400 - 1;
            if end_ts > 0 && time_val > end_ts {
                return false;
            }
        }
    }

    true
}

// parse_api_time_str 定义在 line ~3469，返回 Option<u64>

/// CodeChangeInfo → Qot_GetCodeChange::CodeChangeInfo proto
///
/// C++ CodeChangeInfo_NNToAPI
fn code_change_info_to_proto(
    info: &futu_backend::code_change::CodeChangeInfo,
) -> futu_proto::qot_get_code_change::CodeChangeInfo {
    let security = futu_proto::qot_common::Security {
        market: info.qot_market,
        code: info.sec_code.clone(),
    };

    let related_security = futu_proto::qot_common::Security {
        market: info.qot_market,
        code: info.relate_sec_code.clone(),
    };

    // C++: bReplyTimestamp / bReplyTime — 同时返回时间戳和时间字符串
    let public_time = if info.public_time > 0 {
        Some(format_timestamp(info.public_time))
    } else {
        None
    };
    let public_timestamp = if info.public_time > 0 {
        Some(info.public_time as f64)
    } else {
        None
    };

    let effective_time = if info.effective_time > 0 {
        Some(format_timestamp(info.effective_time))
    } else {
        None
    };
    let effective_timestamp = if info.effective_time > 0 {
        Some(info.effective_time as f64)
    } else {
        None
    };

    // C++: GemToMain 类型不设置 endTime/endTimestamp
    let (end_time, end_timestamp) = if info.change_type
        != futu_backend::code_change::CodeChangeType::GemToMain
        && info.end_time > 0
    {
        (
            Some(format_timestamp(info.end_time)),
            Some(info.end_time as f64),
        )
    } else {
        (None, None)
    };

    futu_proto::qot_get_code_change::CodeChangeInfo {
        r#type: info.change_type as i32,
        security,
        related_security,
        public_time,
        public_timestamp,
        effective_time,
        effective_timestamp,
        end_time,
        end_timestamp,
    }
}

#[async_trait]
impl RequestHandler for GetCodeChangeHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_code_change::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let cache = self.code_change_cache.read();
        let mut code_change_list = Vec::new();

        for info in cache.iter() {
            if code_change_fits_condition(info, c2s) {
                code_change_list.push(code_change_info_to_proto(info));
            }
        }

        tracing::debug!(
            conn_id,
            total_cached = cache.len(),
            matched = code_change_list.len(),
            "GetCodeChange"
        );

        let resp = futu_proto::qot_get_code_change::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_code_change::S2c { code_change_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetIpoList (CMD 6956 → 后端转发) =====
struct GetIpoListHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetIpoListHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_ipo_list::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetIpoList: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Build backend IpoListReq (CMD 6956)
        // 对齐 C++: HK 需要发两个请求 (HK_APPLYING=6 + HK_WAITING_IPO=7)
        //           CN 需要发四个请求 (NEW_NOTICE=1 + APPLYING=2 + APPLY_RESULT=3 + WAITING_IPO=4)
        //           US 只需一个请求 (ALL=5)
        let request_types: Vec<i32> = match c2s.market {
            1 => vec![6, 7],             // HK: HK_APPLYING, HK_WAITING_IPO
            2 => vec![5],                // US: ALL
            21 | 22 => vec![1, 2, 3, 4], // CN: NEW_NOTICE, APPLYING, APPLY_RESULT, WAITING_IPO
            _ => vec![5],                // Default: ALL
        };

        // 合并多个请求的响应
        let mut merged_hk_list = Vec::new();
        let mut merged_us_list = Vec::new();
        let mut merged_cn_list = Vec::new();

        for req_type in &request_types {
            let backend_req =
                futu_backend::proto_internal::ft_cmd_ipo_calender6955_6959::IpoListReq {
                    market: Some(c2s.market),
                    request_type: Some(*req_type),
                    language: Some(1), // 简体中文
                };
            let body = prost::Message::encode_to_vec(&backend_req);

            tracing::debug!(
                conn_id,
                market = c2s.market,
                request_type = req_type,
                "sending CMD6956 IpoListReq"
            );

            let resp_frame = match backend.request(6956, body).await {
                Ok(f) => f,
                Err(e) => {
                    tracing::warn!(conn_id, error = %e, request_type = req_type, "CMD6956 request failed");
                    continue;
                }
            };

            let rsp: futu_backend::proto_internal::ft_cmd_ipo_calender6955_6959::IpoListRsp =
                match prost::Message::decode(resp_frame.body.as_ref()) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::warn!(conn_id, error = %e, request_type = req_type, "CMD6956 decode failed");
                        continue;
                    }
                };

            merged_hk_list.extend(rsp.hk_list);
            merged_us_list.extend(rsp.us_list);
            merged_cn_list.extend(rsp.cn_list);
        }

        let backend_rsp = futu_backend::proto_internal::ft_cmd_ipo_calender6955_6959::IpoListRsp {
            market: None,
            hk_list: merged_hk_list,
            us_list: merged_us_list,
            cn_list: merged_cn_list,
            request_type: None,
        };

        let ipo_count =
            backend_rsp.cn_list.len() + backend_rsp.hk_list.len() + backend_rsp.us_list.len();

        tracing::debug!(
            conn_id,
            cn = backend_rsp.cn_list.len(),
            hk = backend_rsp.hk_list.len(),
            us = backend_rsp.us_list.len(),
            "GetIpoList: received {} IPO items from backend",
            ipo_count
        );

        let mut ipo_list = Vec::with_capacity(ipo_count);

        // Map HK IPO items
        for item in &backend_rsp.hk_list {
            let stock_id = match item.stock_id {
                Some(id) if id > 0 => id,
                _ => continue,
            };
            let security = match self.resolve_security(stock_id) {
                Some(s) => s,
                None => {
                    tracing::warn!(
                        stock_id,
                        name = ?item.stock_name,
                        "GetIpoList: HK stock_id not found in cache, skipping"
                    );
                    continue;
                }
            };
            let name = item.stock_name.clone().unwrap_or_default();

            // 上市日期
            let (list_time, list_timestamp) = ipo_timestamp_fields(item.ipo_timestamp);

            let basic = futu_proto::qot_get_ipo_list::BasicIpoData {
                security,
                name,
                list_time,
                list_timestamp,
            };

            // 发售价区间: ipo_price_low/ipo_price_high 保留两位精度（即123代表1.23）
            let ipo_price_min = item.ipo_price_low.map(|v| v as f64 / 100.0).unwrap_or(0.0);
            let ipo_price_max = item.ipo_price_high.map(|v| v as f64 / 100.0).unwrap_or(0.0);
            // 上市价 (string → f64)
            let list_price = item
                .ipo_price
                .as_deref()
                .and_then(|s| s.parse::<f64>().ok())
                .unwrap_or(0.0);
            // 每手股数 (string → i32)
            let lot_size = item
                .lot_size
                .as_deref()
                .and_then(|s| s.parse::<i32>().ok())
                .unwrap_or(0);
            // 入场费: entrance_fee_num 2位精度（即123434表示1234.34）
            let entrance_price = item
                .entrance_fee_num
                .map(|v| v as f64 / 100.0)
                .unwrap_or(0.0);
            // 是否为认购状态: eipo_flag != 0 且 apply_countdown_secs > 0
            let is_subscribe_status =
                item.eipo_flag.unwrap_or(0) != 0 && item.apply_countdown_secs.unwrap_or(0) > 0;

            // 截止认购时间
            let (apply_end_time, apply_end_timestamp) =
                ipo_timestamp_fields(item.apply_end_timestamp);

            let hk_ex = futu_proto::qot_get_ipo_list::HkIpoExData {
                ipo_price_min,
                ipo_price_max,
                list_price,
                lot_size,
                entrance_price,
                is_subscribe_status,
                apply_end_time,
                apply_end_timestamp,
            };

            ipo_list.push(futu_proto::qot_get_ipo_list::IpoData {
                basic,
                cn_ex_data: None,
                hk_ex_data: Some(hk_ex),
                us_ex_data: None,
            });
        }

        // Map US IPO items
        for item in &backend_rsp.us_list {
            let stock_id = match item.stock_id {
                Some(id) if id > 0 => id,
                _ => continue,
            };
            let security = match self.resolve_security(stock_id) {
                Some(s) => s,
                None => continue,
            };
            let name = item.stock_name.clone().unwrap_or_default();

            // 预计上市日
            let (list_time, list_timestamp) = ipo_timestamp_fields_u64(item.ipo_date_timestamp);

            let basic = futu_proto::qot_get_ipo_list::BasicIpoData {
                security,
                name,
                list_time,
                list_timestamp,
            };

            // 发行价区间: ipo_price_low/ipo_price_high 保留两位精度（即1234表示12.34）
            let ipo_price_min = item.ipo_price_low.map(|v| v as f64 / 100.0).unwrap_or(0.0);
            let ipo_price_max = item.ipo_price_high.map(|v| v as f64 / 100.0).unwrap_or(0.0);
            // 发行量
            let issue_size = item.shares_num.unwrap_or(0);

            let us_ex = futu_proto::qot_get_ipo_list::UsIpoExData {
                ipo_price_min,
                ipo_price_max,
                issue_size,
            };

            ipo_list.push(futu_proto::qot_get_ipo_list::IpoData {
                basic,
                cn_ex_data: None,
                hk_ex_data: None,
                us_ex_data: Some(us_ex),
            });
        }

        // Map CN IPO items
        for item in &backend_rsp.cn_list {
            let stock_id = match item.stock_id {
                Some(id) if id > 0 => id,
                _ => continue,
            };
            let security = match self.resolve_security(stock_id) {
                Some(s) => s,
                None => continue,
            };
            let name = item.stock_name.clone().unwrap_or_default();

            // 上市日期
            let (list_time, list_timestamp) = ipo_timestamp_fields_u64(item.ipo_date_timestamp);

            let basic = futu_proto::qot_get_ipo_list::BasicIpoData {
                security,
                name,
                list_time,
                list_timestamp,
            };

            let apply_code = item.apply_code.clone().unwrap_or_default();
            // 发行价 (string → f64)
            let ipo_price = item
                .ipo_price
                .as_deref()
                .and_then(|s| s.parse::<f64>().ok())
                .unwrap_or(0.0);
            let is_estimate_ipo_price = item.is_ipo_price_preview.unwrap_or(0) != 0;
            // 申购上限
            let apply_upper_limit = item.apply_limit_num.unwrap_or(0);
            // 发行市盈率: pe_num 两位精度（即2012表示20.12）
            let issue_pe_rate = item.pe_num.map(|v| v as f64 / 100.0).unwrap_or(0.0);
            // 中签率 (string → f64)
            let winning_ratio = item
                .lucky_ratio
                .as_deref()
                .and_then(|s| s.parse::<f64>().ok())
                .unwrap_or(0.0);
            let is_estimate_winning_ratio = item.is_lucky_ratio_preview.unwrap_or(0) != 0;
            // 是否已公布中签号
            let is_has_won = item.lucky_ok.unwrap_or(0) != 0;

            // 申购日期
            let (apply_time, apply_timestamp) = ipo_timestamp_fields_u64(item.apply_date_timestamp);
            // 公布中签日
            let (winning_time, winning_timestamp) =
                ipo_timestamp_fields_u64(item.lucky_date_timestamp);

            // issue_size / online_issue_size / apply_limit_market_value / industry_pe_rate:
            // 后端列表接口没有单独提供这些字段的数值格式，使用 0 作为默认值
            // （与C++行为一致：C++从二进制流解析，默认为0）
            let cn_ex = futu_proto::qot_get_ipo_list::CnIpoExData {
                apply_code,
                issue_size: 0,
                online_issue_size: 0,
                apply_upper_limit,
                apply_limit_market_value: 0,
                is_estimate_ipo_price,
                ipo_price,
                industry_pe_rate: 0.0,
                is_estimate_winning_ratio,
                winning_ratio,
                issue_pe_rate,
                apply_time,
                apply_timestamp,
                winning_time,
                winning_timestamp,
                is_has_won,
                winning_num_data: Vec::new(),
            };

            ipo_list.push(futu_proto::qot_get_ipo_list::IpoData {
                basic,
                cn_ex_data: Some(cn_ex),
                hk_ex_data: None,
                us_ex_data: None,
            });
        }

        tracing::debug!(
            conn_id,
            mapped = ipo_list.len(),
            "GetIpoList: mapped {} IPO items to FTAPI",
            ipo_list.len()
        );

        let resp = futu_proto::qot_get_ipo_list::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_ipo_list::S2c { ipo_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

impl GetIpoListHandler {
    /// 通过 stock_id 从 static_cache 解析出 Security（market + code）
    fn resolve_security(&self, stock_id: u64) -> Option<futu_proto::qot_common::Security> {
        let key = self.static_cache.id_to_key.get(&stock_id)?;
        let parts: Vec<&str> = key.split('_').collect();
        if parts.len() != 2 {
            return None;
        }
        let market: i32 = parts[0].parse().ok()?;
        let code = parts[1].to_string();
        Some(futu_proto::qot_common::Security { market, code })
    }
}

/// 将 Option<u32> 时间戳转换为日期字符串和 f64 时间戳（用于 FTAPI 响应）
/// 如果时间戳为 None 或 0，返回 (None, None)
fn ipo_timestamp_fields(ts: Option<u32>) -> (Option<String>, Option<f64>) {
    match ts {
        Some(v) if v > 0 => {
            let time_str = timestamp_to_date_str(v as u64);
            (Some(time_str), Some(v as f64))
        }
        _ => (None, None),
    }
}

/// 将 Option<u64> 时间戳转换为日期字符串和 f64 时间戳
fn ipo_timestamp_fields_u64(ts: Option<u64>) -> (Option<String>, Option<f64>) {
    match ts {
        Some(v) if v > 0 => {
            let time_str = timestamp_to_date_str(v);
            (Some(time_str), Some(v as f64))
        }
        _ => (None, None),
    }
}

// ===== GetFutureInfo (CMD 6337 → 后端转发) =====
struct GetFutureInfoHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetFutureInfoHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_future_info::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetFutureInfo: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Resolve stock_ids from security_list
        let mut stock_ids = Vec::new();
        for sec in &c2s.security_list {
            let sec_key = format!("{}_{}", sec.market, sec.code);
            if let Some(info) = self.static_cache.get_security_info(&sec_key) {
                if info.stock_id > 0 {
                    stock_ids.push(info.stock_id);
                }
            }
        }

        if stock_ids.is_empty() {
            tracing::warn!(conn_id, "GetFutureInfo: no valid stock_ids found");
            let resp = futu_proto::qot_get_future_info::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::qot_get_future_info::S2c {
                    future_info_list: Vec::new(),
                }),
            };
            return Some(prost::Message::encode_to_vec(&resp));
        }

        // Build backend FutureDetailInfoListReq (CMD 6337)
        let backend_req =
            futu_backend::proto_internal::ft_cmd_us_future_info::FutureDetailInfoListReq {
                security_id_list: stock_ids.clone(),
            };
        let body = prost::Message::encode_to_vec(&backend_req);

        tracing::debug!(
            conn_id,
            count = stock_ids.len(),
            "sending CMD6337 FutureDetailInfoListReq"
        );

        let resp_frame = match backend.request(6337, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6337 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let backend_rsp: futu_backend::proto_internal::ft_cmd_us_future_info::FutureDetailInfoListRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6337 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if backend_rsp.ret_code.unwrap_or(-1) != 0 {
            tracing::warn!(conn_id, ret = ?backend_rsp.ret_code, "CMD6337 returned error");
            return Some(super::make_error_response(
                -1,
                "backend future info request failed",
            ));
        }

        // Convert backend FutureDetailInfo → FTAPI FutureInfo
        let future_info_list: Vec<futu_proto::qot_get_future_info::FutureInfo> = backend_rsp
            .future_detail_info_list
            .iter()
            .filter_map(|detail| {
                // Resolve security_id back to market/code
                let security_id = detail.security_id?;
                let key = self.static_cache.id_to_key.get(&security_id)?;
                let parts: Vec<&str> = key.split('_').collect();
                if parts.len() != 2 {
                    return None;
                }
                let market: i32 = parts[0].parse().ok()?;
                let code = parts[1].to_string();
                let security = futu_proto::qot_common::Security { market, code };

                // Convert last_trade_date from timestamp
                let last_trade_time = detail
                    .last_trade_date
                    .map(|ts| timestamp_to_date_str(ts as u64))
                    .unwrap_or_default();

                let last_trade_timestamp = detail.last_trade_date.map(|ts| ts as f64);

                // Trade time conversion
                let trade_time: Vec<futu_proto::qot_get_future_info::TradeTime> = detail
                    .trade_time_detail_list
                    .iter()
                    .map(|tt| futu_proto::qot_get_future_info::TradeTime {
                        begin: tt.begin.map(|v| v as f64),
                        end: tt.end.map(|v| v as f64),
                    })
                    .collect();

                Some(futu_proto::qot_get_future_info::FutureInfo {
                    name: detail.name.clone().unwrap_or_default(),
                    security,
                    last_trade_time,
                    last_trade_timestamp,
                    owner: None,
                    owner_other: detail.show_variety.clone().unwrap_or_default(),
                    exchange: detail.exchange.clone().unwrap_or_default(),
                    contract_type: detail.category.clone().unwrap_or_default(),
                    contract_size: detail.scale.unwrap_or(0) as f64,
                    contract_size_unit: detail.scale_unit.clone().unwrap_or_default(),
                    quote_currency: detail.currency.clone().unwrap_or_default(),
                    min_var: detail.minimum_variation_value.unwrap_or(0.0),
                    min_var_unit: detail.minimum_variation_unit.clone().unwrap_or_default(),
                    quote_unit: detail.price_quotation.clone(),
                    trade_time,
                    time_zone: detail.time_zone.clone().unwrap_or_default(),
                    exchange_format_url: detail.specification_url.clone().unwrap_or_default(),
                    origin: None,
                })
            })
            .collect();

        tracing::debug!(
            conn_id,
            count = future_info_list.len(),
            "GetFutureInfo: returning future info"
        );

        let resp = futu_proto::qot_get_future_info::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_future_info::S2c { future_info_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== SetPriceReminder (CMD 6809 设置到价提醒) =====
struct SetPriceReminderHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for SetPriceReminderHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_set_price_reminder::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "SetPriceReminder: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // Resolve stock_id
        let sec = &c2s.security;
        let sec_key = format!("{}_{}", sec.market, sec.code);
        let stock_id = match self.static_cache.get_security_info(&sec_key) {
            Some(info) if info.stock_id > 0 => info.stock_id,
            _ => {
                return Some(super::make_error_response(-1, "unknown stock"));
            }
        };

        // 映射 API op -> 后端 oper_type: 1=Add, 2=Del, 3=Modify
        // API: 1=Add, 2=Del, 3=Enable, 4=Disable, 5=Modify, 6=DelAll
        let api_op = c2s.op;
        let is_del_all = api_op == 6;

        let mut backend_req = futu_backend::proto_internal::ft_cmd_price_warn::PriceWarnSetNewReq {
            sid: Some(stock_id),
            delete_flag: None,
            attr: Vec::new(),
        };

        if is_del_all {
            backend_req.delete_flag = Some(true);
        } else {
            let oper_type: u32 = match api_op {
                1 => 1,     // Add
                2 => 2,     // Del
                3..=5 => 3, // Enable/Disable/Modify -> Modify
                _ => {
                    return Some(super::make_error_response(-1, "unknown price reminder op"));
                }
            };

            let key_val = c2s.key.filter(|_| api_op != 1); // Add 不需要 key
            let enable = match api_op {
                3 => Some(true),
                4 => Some(false),
                _ => None,
            };

            // 映射 API PriceReminderType -> 后端 WarnType
            let warn_type = c2s.r#type.and_then(api_reminder_type_to_warn_type);

            // 映射 API PriceReminderFreq -> 后端 FreqType
            let freq_type = c2s.freq.and_then(api_freq_to_backend_freq);

            // value -> warn_param (value * 1000, 四舍五入)
            let warn_param = c2s.value.map(|v| (v * 1000.0).round() as i64);

            let note = c2s.note.clone();

            // 提醒时段
            let notify_time_periods: Vec<u32> = c2s
                .reminder_session_list
                .iter()
                .filter_map(|&s| api_market_status_to_notify_period(s))
                .collect();

            let now_ts = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs();

            let attr = futu_backend::proto_internal::ft_cmd_price_warn::ItemAttr {
                key: key_val.map(|k| k as u64),
                warn_type,
                warn_param,
                note,
                enable,
                update_time: Some(now_ts),
                freq_type,
                oper_type: Some(oper_type),
                stock_id: None,
                fine_warn_param: None,
                notify_time_periods,
            };
            backend_req.attr.push(attr);
        }

        let body = prost::Message::encode_to_vec(&backend_req);
        tracing::debug!(
            conn_id,
            stock_id,
            api_op,
            "SetPriceReminder: sending CMD6809"
        );

        let resp_frame = match backend.request(6809, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6809 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let backend_rsp: futu_backend::proto_internal::ft_cmd_price_warn::PriceWarnSetNewRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6809 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if backend_rsp.result.unwrap_or(-1) != 0 {
            let err_text = backend_rsp.err_text.unwrap_or_default();
            let msg = if err_text.is_empty() {
                "set price reminder failed".to_string()
            } else {
                err_text
            };
            return Some(super::make_error_response(-1, &msg));
        }

        // 对于 Add 操作, 后端返回 warn_items 中包含新 key
        // server_seq 可以作为 key 返回 (与 C++ 行为类似, C++ 从 warn_items 中匹配)
        let key = backend_rsp.server_seq.unwrap_or(0) as i64;

        let resp = futu_proto::qot_set_price_reminder::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_set_price_reminder::S2c { key }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

/// API PriceReminderType -> 后端 WarnType
fn api_reminder_type_to_warn_type(api_type: i32) -> Option<u32> {
    match api_type {
        1 => Some(4),   // PriceUp -> UP_PRICE
        2 => Some(8),   // PriceDown -> DOWN_PRICE
        3 => Some(1),   // ChangeRateUp -> DAY_UP_RATIO
        4 => Some(2),   // ChangeRateDown -> DAY_DOWN_RATIO
        5 => Some(9),   // 5MinChangeRateUp -> FIVE_MIN_UP_RATIO
        6 => Some(10),  // 5MinChangeRateDown -> FIVE_MIN_DOWN_RATIO
        7 => Some(11),  // VolumeUp -> UP_VOLUME
        8 => Some(12),  // TurnoverUp -> UP_TURNOVER
        9 => Some(13),  // TurnoverRateUp -> UP_TOR
        10 => Some(14), // BidPriceUp -> UP_BUY_PRICE_1
        11 => Some(15), // AskPriceDown -> DOWN_SELL_PRICE_1
        12 => Some(16), // BidVolUp -> UP_BUY_VOLUME_1
        13 => Some(17), // AskVolUp -> DOWN_SELL_VOLUME_1
        14 => Some(19), // 3MinChangeRateUp -> THREE_MIN_UP_RATIO
        15 => Some(20), // 3MinChangeRateDown -> THREE_MIN_DOWN_RATIO
        _ => None,
    }
}

/// 后端 WarnType -> API PriceReminderType
fn warn_type_to_api_reminder_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 -> 5MinChangeRateUp
        10 => 6,  // FIVE_MIN_DOWN_RATIO -> 5MinChangeRateDown
        11 => 7,  // UP_VOLUME -> VolumeUp
        12 => 8,  // UP_TURNOVER -> TurnoverUp
        13 => 9,  // UP_TOR -> TurnoverRateUp
        14 => 10, // UP_BUY_PRICE_1 -> BidPriceUp
        15 => 11, // DOWN_SELL_PRICE_1 -> AskPriceDown
        16 => 12, // UP_BUY_VOLUME_1 -> BidVolUp
        17 => 13, // DOWN_SELL_VOLUME_1 -> AskVolUp
        19 => 14, // THREE_MIN_UP_RATIO -> 3MinChangeRateUp
        20 => 15, // THREE_MIN_DOWN_RATIO -> 3MinChangeRateDown
        _ => 0,   // Unknown
    }
}

/// API PriceReminderFreq -> 后端 FreqType
fn api_freq_to_backend_freq(api_freq: i32) -> Option<u32> {
    match api_freq {
        1 => Some(1), // Always -> FREQ_TYPE_ALWAYS
        2 => Some(2), // OnceADay -> FREQ_TYPE_ONCE_FOR_ONE_DAY
        3 => Some(4), // OnlyOnce -> FREQ_TYPE_ONLY_ONCE
        _ => None,
    }
}

/// 后端 FreqType -> API PriceReminderFreq
fn backend_freq_to_api_freq(freq: u32) -> i32 {
    match freq {
        1 => 1, // FREQ_TYPE_ALWAYS -> Always
        2 => 2, // FREQ_TYPE_ONCE_FOR_ONE_DAY -> OnceADay
        4 => 3, // FREQ_TYPE_ONLY_ONCE -> OnlyOnce
        _ => 0, // Unknown
    }
}

/// API PriceReminderMarketStatus -> 后端 NotifyTimePeriod
fn api_market_status_to_notify_period(status: i32) -> Option<u32> {
    match status {
        1 => Some(1), // Open -> NORMAL
        2 => Some(2), // USPre -> US_BEFORE
        3 => Some(3), // USAfter -> US_AFTER
        4 => Some(4), // USOverNight -> US_OVERNIGHT
        _ => None,
    }
}

/// 后端 NotifyTimePeriod -> API PriceReminderMarketStatus
fn notify_period_to_api_market_status(period: u32) -> i32 {
    match period {
        1 => 1, // NORMAL -> Open
        2 => 2, // US_BEFORE -> USPre
        3 => 3, // US_AFTER -> USAfter
        4 => 4, // US_OVERNIGHT -> USOverNight
        _ => 0, // Unknown
    }
}

/// API QotMarket -> 后端 PriceReminder MarketType
fn api_qot_market_to_price_reminder_market(qot_market: i32) -> u32 {
    match qot_market {
        1 | 6 => 1,    // HK_Security | HK_Future -> HK(1)
        11 => 10,      // US_Security -> US(10)
        21 | 22 => 30, // CNSH | CNSZ -> HS(30)
        _ => 0,
    }
}

// ===== GetPriceReminder (CMD 6808 拉取到价提醒) =====
struct GetPriceReminderHandler {
    backend: crate::bridge::SharedBackend,
    static_cache: Arc<StaticDataCache>,
}

#[async_trait]
impl RequestHandler for GetPriceReminderHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_price_reminder::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetPriceReminder: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // security 和 market 二选一, security 优先
        let (stock_id_opt, market_opt) = if let Some(sec) = &c2s.security {
            let sec_key = format!("{}_{}", sec.market, sec.code);
            match self.static_cache.get_security_info(&sec_key) {
                Some(info) if info.stock_id > 0 => (Some(info.stock_id), None),
                _ => {
                    return Some(super::make_error_response(-1, "unknown stock"));
                }
            }
        } else if let Some(market) = c2s.market {
            let backend_market = api_qot_market_to_price_reminder_market(market);
            if backend_market == 0 {
                return Some(super::make_error_response(-1, "unsupported market"));
            }
            (None, Some(backend_market))
        } else {
            return Some(super::make_error_response(
                -1,
                "must specify security or market",
            ));
        };

        // Build backend PriceWarnGetNew_Req (CMD 6808)
        let backend_req = futu_backend::proto_internal::ft_cmd_price_warn::PriceWarnGetNewReq {
            stock_id: stock_id_opt,
            market: market_opt,
        };
        let body = prost::Message::encode_to_vec(&backend_req);

        tracing::debug!(
            conn_id,
            stock_id = ?stock_id_opt,
            market = ?market_opt,
            "GetPriceReminder: sending CMD6808"
        );

        let resp_frame = match backend.request(6808, body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD6808 request failed");
                return Some(super::make_error_response(-1, "backend request failed"));
            }
        };

        let backend_rsp: futu_backend::proto_internal::ft_cmd_price_warn::PriceWarnGetNewRsp =
            match prost::Message::decode(resp_frame.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::error!(conn_id, error = %e, "CMD6808 decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "backend response decode failed",
                    ));
                }
            };

        if backend_rsp.result.unwrap_or(-1) != 0 {
            let err_text = backend_rsp.err_text.unwrap_or_default();
            let msg = if err_text.is_empty() {
                "get price reminder failed".to_string()
            } else {
                err_text
            };
            return Some(super::make_error_response(-1, &msg));
        }

        // 转换 WarnItem -> PriceReminder
        let price_reminder_list: Vec<futu_proto::qot_get_price_reminder::PriceReminder> =
            backend_rsp
                .warn_items
                .iter()
                .filter_map(|warn_item| {
                    let sid = warn_item.stock_id?;
                    // Resolve stock_id -> Security
                    let key_ref = self.static_cache.id_to_key.get(&sid)?;
                    let info = self.static_cache.get_security_info(key_ref.value())?;
                    let security = futu_proto::qot_common::Security {
                        market: info.market,
                        code: info.code.clone(),
                    };

                    let item_list: Vec<futu_proto::qot_get_price_reminder::PriceReminderItem> =
                        warn_item
                            .attr
                            .iter()
                            .filter_map(|attr| {
                                let wt = attr.warn_type?;
                                // 过滤公告提醒类型 (INFO_WARN = 18)
                                if wt == 18 {
                                    return None;
                                }
                                let api_type = warn_type_to_api_reminder_type(wt);
                                let value = attr.warn_param.unwrap_or(0) as f64 / 1000.0;
                                let note = attr.note.clone().unwrap_or_default();
                                let freq = backend_freq_to_api_freq(attr.freq_type.unwrap_or(0));
                                let is_enable = attr.enable.unwrap_or(false);
                                let key = attr.key.unwrap_or(0) as i64;

                                let reminder_session_list: Vec<i32> = attr
                                    .notify_time_periods
                                    .iter()
                                    .map(|&p| notify_period_to_api_market_status(p))
                                    .collect();

                                Some(futu_proto::qot_get_price_reminder::PriceReminderItem {
                                    key,
                                    r#type: api_type,
                                    value,
                                    note,
                                    freq,
                                    is_enable,
                                    reminder_session_list,
                                })
                            })
                            .collect();

                    Some(futu_proto::qot_get_price_reminder::PriceReminder {
                        security,
                        name: Some(info.name.clone()),
                        item_list,
                    })
                })
                .collect();

        tracing::debug!(
            conn_id,
            count = price_reminder_list.len(),
            "GetPriceReminder: returning {} reminders",
            price_reminder_list.len()
        );

        let resp = futu_proto::qot_get_price_reminder::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_price_reminder::S2c {
                price_reminder_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetUserSecurityGroup (CMD 5121 拉取分组列表) =====
// C++ 流程: 拉取分组信息后, 按 groupType 过滤(Custom/System/All),
// 跳过基金宝(890), 外汇(891), 持仓(896)
struct GetUserSecurityGroupHandler {
    backend: crate::bridge::SharedBackend,
    app_lang: i32,
}

#[async_trait]
impl RequestHandler for GetUserSecurityGroupHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::qot_get_user_security_group::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let group_type = req.c2s.group_type;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                tracing::warn!(conn_id, "GetUserSecurityGroup: no backend connection");
                return Some(super::make_error_response(-1, "no backend connection"));
            }
        };

        // 校验 groupType: 1=Custom, 2=System, 3=All
        if !(1..=3).contains(&group_type) {
            return Some(super::make_error_response(-1, "unsupported group type"));
        }

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

        let group_req = futu_backend::proto_internal::wch_lst::GetGroupListReq {
            user_id: Some(user_id),
        };
        let group_body = prost::Message::encode_to_vec(&group_req);
        tracing::info!(conn_id, group_type, "GetUserSecurityGroup: sending CMD5121");

        let group_frame = match backend.request(5121, group_body).await {
            Ok(f) => f,
            Err(e) => {
                tracing::error!(conn_id, error = %e, "CMD5121 request failed");
                return Some(super::make_error_response(-1, "pull group info failed"));
            }
        };

        // CMD5121 响应的 SRPC field 5 包含 repeated GroupInfo（不是 GetGroupListResp），
        // 需要使用专用解码函数
        let (group_rsp, group_langs) = super::decode_cmd5121_groups(group_frame.body.as_ref());

        if group_rsp.result_code.unwrap_or(-1) != 0 {
            return Some(super::make_error_response(-1, "pull group info failed"));
        }

        // 过滤并转换分组
        let app_lang = self.app_lang;
        let mut group_list = Vec::new();
        for (i, g) in group_rsp.group_list.iter().enumerate() {
            let gid = g.group_id.unwrap_or(0);
            let is_system_group = (gid > 0 && gid < 900) || gid == 1000;

            // 系统分组名称获取优先级（与 C++ 一致）：
            // 1. 从 multi_lang_name 中按客户端语言匹配
            // 2. group_name 字段
            // 3. 硬编码映射表 fallback
            let mut name = String::new();
            if is_system_group {
                if let Some(langs) = group_langs.get(i) {
                    if let Some(ml) = langs.iter().find(|m| m.language_id == app_lang) {
                        name = ml.name.clone();
                    }
                }
            }
            if name.is_empty() {
                name = g.group_name.clone().unwrap_or_default();
            }
            if is_system_group && name.is_empty() {
                if let Some(localized) = system_group_id_to_name(gid, app_lang as u8) {
                    name = localized.to_string();
                }
            }

            // 跳过: 基金宝(890), 外汇(891), 持仓(896)
            if gid == 890 || gid == 891 || gid == 896 {
                continue;
            }

            let include = match group_type {
                1 => !is_system_group, // Custom
                2 => is_system_group,  // System
                3 => true,             // All
                _ => false,
            };

            if include {
                let api_group_type = if is_system_group { 2 } else { 1 };
                group_list.push(futu_proto::qot_get_user_security_group::GroupData {
                    group_name: name,
                    group_type: api_group_type,
                });
            }
        }

        tracing::debug!(
            conn_id,
            count = group_list.len(),
            "GetUserSecurityGroup: returning {} groups",
            group_list.len()
        );

        let resp = futu_proto::qot_get_user_security_group::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::qot_get_user_security_group::S2c { group_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

/// 系统分组名称到 group_id 的映射。
/// 支持简体中文、繁体中文和英文名。
/// C++ 中系统分组名称来自 multi_lang_name，按客户端语言匹配。
fn system_group_name_to_id(name: &str) -> Option<u32> {
    match name {
        "全部" | "ALL" | "All" => Some(1000),
        "沪深" | "滬深" | "A股" | "CN" => Some(899),
        "港股" | "HK" => Some(898),
        "美股" | "US" => Some(897),
        "持仓" | "持倉" | "Position" => Some(896),
        "美股期权" | "美股期權" | "US Options" => Some(895),
        "特别关注" | "特別關注" | "Focus" => Some(894),
        "港股期权" | "港股期權" | "HK Options" => Some(893),
        "期货" | "期貨" | "Futures" => Some(892),
        "外汇" | "外匯" | "Forex" => Some(891),
        "基金宝" | "基金寶" | "Fund" => Some(890),
        "期权" | "期權" | "Options" => Some(889),
        "债券" | "債券" | "Bond" => Some(888),
        "新加坡" | "SG" => Some(887),
        "指数" | "指數" | "Index" => Some(886),
        "数字货币" | "數字貨幣" | "Crypto" => Some(885),
        _ => None,
    }
}

/// 系统分组 ID 到本地化名称的映射。
/// lang: 0=简体中文, 1=繁体中文, 2=English
fn system_group_id_to_name(id: u32, lang: u8) -> Option<&'static str> {
    // 未知语言 fallback 到简体中文
    let lang = if lang > 2 { 0 } else { lang };
    match (id, lang) {
        // 简体中文
        (1000, 0) => Some("全部"),
        (899, 0) => Some("沪深"),
        (898, 0) => Some("港股"),
        (897, 0) => Some("美股"),
        (896, 0) => Some("持仓"),
        (895, 0) => Some("美股期权"),
        (894, 0) => Some("特别关注"),
        (893, 0) => Some("港股期权"),
        (892, 0) => Some("期货"),
        (891, 0) => Some("外汇"),
        (890, 0) => Some("基金宝"),
        (889, 0) => Some("期权"),
        (888, 0) => Some("债券"),
        (887, 0) => Some("新加坡"),
        (886, 0) => Some("指数"),
        (885, 0) => Some("数字货币"),
        // 繁体中文
        (1000, 1) => Some("全部"),
        (899, 1) => Some("滬深"),
        (898, 1) => Some("港股"),
        (897, 1) => Some("美股"),
        (896, 1) => Some("持倉"),
        (895, 1) => Some("美股期權"),
        (894, 1) => Some("特別關注"),
        (893, 1) => Some("港股期權"),
        (892, 1) => Some("期貨"),
        (891, 1) => Some("外匯"),
        (890, 1) => Some("基金寶"),
        (889, 1) => Some("期權"),
        (888, 1) => Some("債券"),
        (887, 1) => Some("新加坡"),
        (886, 1) => Some("指數"),
        (885, 1) => Some("數字貨幣"),
        // English
        (1000, 2) => Some("All"),
        (899, 2) => Some("CN"),
        (898, 2) => Some("HK"),
        (897, 2) => Some("US"),
        (896, 2) => Some("Position"),
        (895, 2) => Some("US Options"),
        (894, 2) => Some("Focus"),
        (893, 2) => Some("HK Options"),
        (892, 2) => Some("Futures"),
        (891, 2) => Some("Forex"),
        (890, 2) => Some("Fund"),
        (889, 2) => Some("Options"),
        (888, 2) => Some("Bond"),
        (887, 2) => Some("SG"),
        (886, 2) => Some("Index"),
        (885, 2) => Some("Crypto"),
        _ => None,
    }
}

// ===== GetUsedQuota (1010) =====
struct GetUsedQuotaHandler {
    subscriptions: Arc<SubscriptionManager>,
    kl_quota_counter: Arc<std::sync::atomic::AtomicU32>,
}

#[async_trait]
impl RequestHandler for GetUsedQuotaHandler {
    async fn handle(&self, conn_id: u64, _req: &IncomingRequest) -> Option<Vec<u8>> {
        let used_sub_quota = self.subscriptions.get_total_used_quota() as i32;
        let used_kl_quota = self
            .kl_quota_counter
            .load(std::sync::atomic::Ordering::Relaxed) as i32;

        tracing::debug!(conn_id, used_sub_quota, used_kl_quota, "GetUsedQuota");

        let resp = futu_proto::used_quota::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::used_quota::S2c {
                used_sub_quota: Some(used_sub_quota),
                used_k_line_quota: Some(used_kl_quota),
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}