futu_gateway/handlers/

trd.rs

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

use std::sync::Arc;

use async_trait::async_trait;

use futu_cache::static_data::StaticDataCache;
use futu_cache::trd_cache::TrdCache;
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;

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

    // 缓存查询类
    router.register(
        proto_id::TRD_GET_ACC_LIST,
        Arc::new(GetAccListHandler {
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::TRD_GET_FUNDS,
        Arc::new(GetFundsHandler {
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::TRD_GET_POSITION_LIST,
        Arc::new(GetPositionListHandler {
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::TRD_GET_ORDER_LIST,
        Arc::new(GetOrderListHandler {
            cache: cache.clone(),
        }),
    );
    // 需要后端转发的交易操作
    let backend = bridge.backend.clone();

    router.register(
        proto_id::TRD_GET_ORDER_FILL_LIST,
        Arc::new(GetOrderFillListHandler {
            cache: cache.clone(),
            backend: backend.clone(),
        }),
    );

    // 订阅类
    router.register(
        proto_id::TRD_SUB_ACC_PUSH,
        Arc::new(SubAccPushHandler {
            subscriptions: subs,
        }),
    );

    // 后端交易命令 ID (C++ NNBase_Define_ProtoCmd.h)
    // 4701=PlaceOrder, 4702=ReplaceOrder, 4703=CancelOrder
    // 4704=QueryFund, 4705=QueryPosition, 4708=QueryOrderList
    // 4710=QueryDealList, 4711=HistoryOrder, 4712=HistoryDeal, 4713=MaxQty
    let fwd = |name: &'static str, cmd: Option<u16>| {
        Arc::new(BackendTrdForwardHandler {
            name,
            backend: backend.clone(),
            backend_cmd_id: cmd,
        })
    };

    router.register(
        proto_id::TRD_UNLOCK_TRADE,
        Arc::new(UnlockTradeHandler {
            backend: backend.clone(),
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::TRD_PLACE_ORDER,
        Arc::new(PlaceOrderHandler {
            backend: backend.clone(),
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::TRD_MODIFY_ORDER,
        Arc::new(ModifyOrderHandler {
            backend: backend.clone(),
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::TRD_GET_MAX_TRD_QTYS,
        fwd("GetMaxTrdQtys", Some(4713)),
    );
    router.register(proto_id::TRD_RECONFIRM_ORDER, fwd("ReconfirmOrder", None));
    router.register(
        proto_id::TRD_GET_HISTORY_ORDER_LIST,
        fwd("GetHistoryOrderList", Some(4711)),
    );
    router.register(
        proto_id::TRD_GET_HISTORY_ORDER_FILL_LIST,
        fwd("GetHistoryOrderFillList", Some(4712)),
    );

    // 新版交易 API — 自定义 handler（后端查询 + 字段映射）
    router.register(
        proto_id::TRD_GET_MARGIN_RATIO,
        Arc::new(GetMarginRatioHandler {
            backend: backend.clone(),
            cache: cache.clone(),
            static_cache: Arc::clone(&bridge.static_cache),
        }),
    );
    router.register(
        proto_id::TRD_GET_ORDER_FEE,
        Arc::new(GetOrderFeeHandler {
            backend: backend.clone(),
            cache: cache.clone(),
        }),
    );
    router.register(
        proto_id::TRD_FLOW_SUMMARY,
        Arc::new(FlowSummaryHandler {
            backend: backend.clone(),
            cache: cache.clone(),
        }),
    );

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

// ===== GetAccList =====
struct GetAccListHandler {
    cache: Arc<TrdCache>,
}

#[async_trait]
impl RequestHandler for GetAccListHandler {
    async fn handle(&self, _conn_id: u64, _request: &IncomingRequest) -> Option<Vec<u8>> {
        let mut accounts = self.cache.get_accounts();
        // C++ 顺序: active real → sim → disabled real
        // 每组内保持后端返回的自然顺序 (order_index)
        accounts.sort_by(|a, b| {
            let order = |acc: &futu_cache::trd_cache::CachedTrdAcc| -> (i32, u64) {
                let is_real = acc.trd_env == 1;
                let is_active = acc.acc_status == Some(0);
                let group = if is_real && is_active {
                    0 // active real → 排最前
                } else if !is_real {
                    1 // sim → 中间
                } else {
                    2 // disabled real → 排最后
                };
                (group, acc.sort_key)
            };
            order(a).cmp(&order(b))
        });
        let acc_list = accounts
            .iter()
            .map(|a| futu_proto::trd_common::TrdAcc {
                trd_env: a.trd_env,
                acc_id: a.acc_id,
                trd_market_auth_list: a.trd_market_auth_list.clone(),
                acc_type: a.acc_type,
                card_num: a.card_num.clone(),
                security_firm: a.security_firm,
                sim_acc_type: a.sim_acc_type,
                uni_card_num: a.uni_card_num.clone(),
                acc_status: a.acc_status,
                acc_role: a.acc_role,
                jp_acc_type: a.jp_acc_type.clone(),
            })
            .collect();
        let resp = futu_proto::trd_get_acc_list::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_get_acc_list::S2c { acc_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetFunds =====
struct GetFundsHandler {
    cache: Arc<TrdCache>,
}

#[async_trait]
impl RequestHandler for GetFundsHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::trd_get_funds::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let header = req.c2s.header;
        let cached_fund_keys: Vec<u64> = self.cache.funds.iter().map(|e| *e.key()).collect();
        tracing::info!(
            requested_acc_id = header.acc_id,
            cached_fund_keys = ?cached_fund_keys,
            "GetFunds lookup"
        );
        let funds = self
            .cache
            .funds
            .get(&header.acc_id)
            .map(|f| futu_proto::trd_common::Funds {
                power: f.power,
                total_assets: f.total_assets,
                cash: f.cash,
                market_val: f.market_val,
                frozen_cash: f.frozen_cash,
                debt_cash: f.debt_cash,
                avl_withdrawal_cash: f.avl_withdrawal_cash,
                currency: f.currency,
                available_funds: f.available_funds,
                unrealized_pl: f.unrealized_pl,
                realized_pl: f.realized_pl,
                risk_level: f.risk_level,
                initial_margin: f.initial_margin,
                maintenance_margin: f.maintenance_margin,
                cash_info_list: f
                    .cash_info_list
                    .iter()
                    .map(|c| futu_proto::trd_common::AccCashInfo {
                        currency: Some(c.currency),
                        cash: Some(c.cash),
                        available_balance: Some(c.available_balance),
                        net_cash_power: Some(c.net_cash_power),
                    })
                    .collect(),
                max_power_short: f.max_power_short,
                net_cash_power: f.net_cash_power,
                long_mv: f.long_mv,
                short_mv: f.short_mv,
                pending_asset: f.pending_asset,
                max_withdrawal: f.max_withdrawal,
                risk_status: f.risk_status,
                margin_call_margin: f.margin_call_margin,
                is_pdt: None,
                pdt_seq: None,
                beginning_dtbp: None,
                remaining_dtbp: None,
                dt_call_amount: None,
                dt_status: None,
                securities_assets: f.securities_assets,
                fund_assets: f.fund_assets,
                bond_assets: f.bond_assets,
                market_info_list: f
                    .market_info_list
                    .iter()
                    .map(|m| futu_proto::trd_common::AccMarketInfo {
                        trd_market: Some(m.trd_market),
                        assets: Some(m.assets),
                    })
                    .collect(),
            });
        let resp = futu_proto::trd_get_funds::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_get_funds::S2c { header, funds }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetPositionList =====
struct GetPositionListHandler {
    cache: Arc<TrdCache>,
}

#[async_trait]
impl RequestHandler for GetPositionListHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::trd_get_position_list::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let header = req.c2s.header;
        let cached_keys: Vec<u64> = self.cache.positions.iter().map(|e| *e.key()).collect();
        tracing::info!(
            requested_acc_id = header.acc_id,
            cached_position_keys = ?cached_keys,
            "GetPositionList lookup"
        );
        let positions = self
            .cache
            .positions
            .get(&header.acc_id)
            .map(|p| p.value().clone())
            .unwrap_or_default();
        let position_list = positions
            .iter()
            .map(|p| futu_proto::trd_common::Position {
                position_id: p.position_id,
                position_side: p.position_side,
                code: p.code.clone(),
                name: p.name.clone(),
                qty: p.qty,
                can_sell_qty: p.can_sell_qty,
                price: p.price,
                cost_price: Some(p.cost_price),
                val: p.val,
                pl_val: p.pl_val,
                pl_ratio: p.pl_ratio,
                sec_market: p.sec_market,
                td_pl_val: p.td_pl_val,
                td_trd_val: p.td_trd_val,
                td_buy_val: p.td_buy_val,
                td_buy_qty: p.td_buy_qty,
                td_sell_val: p.td_sell_val,
                td_sell_qty: p.td_sell_qty,
                unrealized_pl: p.unrealized_pl,
                realized_pl: p.realized_pl,
                currency: p.currency,
                trd_market: p.trd_market,
                diluted_cost_price: p.diluted_cost_price,
                average_cost_price: p.average_cost_price,
                average_pl_ratio: p.average_pl_ratio,
            })
            .collect();
        let resp = futu_proto::trd_get_position_list::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_get_position_list::S2c {
                header,
                position_list,
            }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetOrderList =====
struct GetOrderListHandler {
    cache: Arc<TrdCache>,
}

#[async_trait]
impl RequestHandler for GetOrderListHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::trd_get_order_list::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let header = req.c2s.header;
        let orders = self
            .cache
            .orders
            .get(&header.acc_id)
            .map(|o| o.value().clone())
            .unwrap_or_default();
        let order_list = orders
            .iter()
            .map(|o| futu_proto::trd_common::Order {
                trd_side: o.trd_side,
                order_type: o.order_type,
                order_status: o.order_status,
                order_id: o.order_id,
                order_id_ex: o.order_id_ex.clone(),
                code: o.code.clone(),
                name: o.name.clone(),
                qty: o.qty,
                price: Some(o.price),
                create_time: o.create_time.clone(),
                update_time: o.update_time.clone(),
                fill_qty: Some(o.fill_qty),
                fill_avg_price: Some(o.fill_avg_price),
                last_err_msg: o.last_err_msg.clone(),
                sec_market: o.sec_market,
                create_timestamp: o.create_timestamp,
                update_timestamp: o.update_timestamp,
                remark: o.remark.clone(),
                time_in_force: o.time_in_force,
                fill_outside_rth: o.fill_outside_rth,
                aux_price: o.aux_price,
                trail_type: o.trail_type,
                trail_value: o.trail_value,
                trail_spread: o.trail_spread,
                currency: o.currency,
                trd_market: o.trd_market,
                session: None,
                jp_acc_type: None,
            })
            .collect();
        let resp = futu_proto::trd_get_order_list::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_get_order_list::S2c { header, order_list }),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== GetOrderFillList =====
struct GetOrderFillListHandler {
    #[expect(dead_code)]
    cache: Arc<TrdCache>,
    backend: crate::bridge::SharedBackend,
}

/// 后端成交列表查询命令 ID (C++: NN_ProtoCmd_Trd_QueryDealList)
const CMD_ORDER_FILL_LIST: u16 = 4710;

/// 将后端 market 映射为 FTAPI TrdMarket
/// 后端: 1→HK, 2→US, 3→CN, 10→Futures, 15→SG
fn map_backend_market_to_trd_market(market: u32) -> i32 {
    match market {
        1 => 1,  // HK
        2 => 2,  // US
        3 => 3,  // CN
        10 => 5, // Futures
        15 => 6, // SG
        v => v as i32,
    }
}

#[async_trait]
impl RequestHandler for GetOrderFillListHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        use futu_backend::proto_internal::{odr_sys_cmn, order_sys_interface};

        let req: futu_proto::trd_get_order_fill_list::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let header = req.c2s.header;
        let acc_id = header.acc_id;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(
                    -1,
                    "GetOrderFillList: 后端未连接",
                ));
            }
        };

        // 构建后端 OrderFillListReq (CMD 4710)
        let backend_req = order_sys_interface::OrderFillListReq {
            msg_header: Some(odr_sys_cmn::MsgHeader {
                req_id: Some(format!("{acc_id}_{}", request.serial_no)),
                account_id: Some(acc_id),
                cipher: Some(vec![]), // C++ always sets cipher (empty if not unlocked)
                security_type: None,
                exchange_code: None,
                input_source: Some(6), // QUANTITATIVE_TRADING
                sub_account_id: None,
            }),
            page_flag: None,
            page_size: None,
            security_type: vec![0], // SECURITY_TYPE_COMMON (C++ 必须填)
            market: vec![],
        };

        tracing::info!(acc_id, "querying order fill list from backend");

        // 分页循环获取全部成交
        let mut all_fills: Vec<odr_sys_cmn::OrderFill> = Vec::new();
        let mut page_flag: Option<String> = None;

        loop {
            let mut paged_req = backend_req.clone();
            paged_req.page_flag = page_flag.clone();

            let resp = match backend
                .request(
                    CMD_ORDER_FILL_LIST,
                    prost::Message::encode_to_vec(&paged_req),
                )
                .await
            {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "GetOrderFillList backend request failed");
                    return Some(super::make_error_response(
                        -1,
                        &format!("GetOrderFillList: {e}"),
                    ));
                }
            };

            let backend_rsp: order_sys_interface::OrderFillListRsp =
                match super::decode_backend_proto(resp.body.as_ref()) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::warn!(error = %e, body_len = resp.body.len(),
                            "GetOrderFillList response decode failed");
                        return Some(super::make_error_response(
                            -1,
                            "GetOrderFillList: 响应解析失败",
                        ));
                    }
                };

            if backend_rsp.result.unwrap_or(-1) != 0 {
                let err_msg = backend_rsp
                    .err_msg
                    .unwrap_or_else(|| "unknown error".to_string());
                tracing::warn!(err_msg = %err_msg, "GetOrderFillList backend returned error");
                return Some(super::make_error_response(
                    -1,
                    &format!("GetOrderFillList: {err_msg}"),
                ));
            }

            all_fills.extend(backend_rsp.order_fills);

            if backend_rsp.completed.unwrap_or(true) {
                break;
            }
            match backend_rsp.page_flag {
                Some(ref pf) if !pf.is_empty() => {
                    page_flag = Some(pf.clone());
                }
                _ => break,
            }
        }

        tracing::info!(acc_id, count = all_fills.len(), "order fill list retrieved");

        // 转换后端 OrderFill → FTAPI Trd_Common::OrderFill
        let order_fill_list: Vec<futu_proto::trd_common::OrderFill> = all_fills
            .iter()
            .map(|f| {
                let pf = |s: &Option<String>| -> f64 {
                    s.as_ref()
                        .and_then(|v| v.parse::<f64>().ok())
                        .unwrap_or(0.0)
                };

                // fill_id: 后端 id 字符串 → hash 为 u64
                let fill_id_ex = f.id.clone().unwrap_or_default();
                let fill_id: u64 = fill_id_ex.parse().unwrap_or(0);

                // order_id: 后端 order_id 字符串 → hash 为 u64
                let order_id_ex = f.order_id.clone().unwrap_or_default();
                let order_id: u64 = order_id_ex.parse().unwrap_or(0);

                // create_time 微秒 → 秒 f64
                let create_timestamp = f.create_time.map(|t| t as f64 / 1_000_000.0);
                let update_timestamp = f.update_time.map(|t| t as f64 / 1_000_000.0);

                // counter_broker_id: 字符串 → i32
                let counter_broker_id = f
                    .counter_broker_id
                    .as_ref()
                    .and_then(|s| s.parse::<i32>().ok());

                // market → trd_market (后端 market 为 string 类型，先解析为 u32)
                let trd_market = f
                    .market
                    .as_ref()
                    .and_then(|s| s.parse::<u32>().ok())
                    .map(map_backend_market_to_trd_market);

                // 成交状态: 根据 is_cancelled / is_corrected 推断
                let status = if f.is_cancelled.unwrap_or(false) {
                    Some(2) // OrderFillStatus_Cancelled
                } else if f.is_corrected.unwrap_or(false) {
                    Some(3) // OrderFillStatus_Changed
                } else {
                    Some(1) // OrderFillStatus_OK
                };

                futu_proto::trd_common::OrderFill {
                    trd_side: f.side.unwrap_or(0) as i32,
                    fill_id,
                    fill_id_ex,
                    order_id: Some(order_id),
                    order_id_ex: Some(order_id_ex),
                    code: f
                        .symbol
                        .as_ref()
                        .map(|b| String::from_utf8_lossy(b).into_owned())
                        .unwrap_or_default(),
                    name: f
                        .stock_name
                        .as_ref()
                        .map(|b| String::from_utf8_lossy(b).into_owned())
                        .unwrap_or_default(),
                    qty: pf(&f.qty),
                    price: pf(&f.price),
                    create_time: String::new(), // 时间戳格式转换留空，使用 timestamp
                    counter_broker_id,
                    counter_broker_name: None,
                    sec_market: None,
                    create_timestamp,
                    update_timestamp,
                    status,
                    trd_market,
                    jp_acc_type: None,
                }
            })
            .collect();

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

// ===== SubAccPush =====
struct SubAccPushHandler {
    subscriptions: Arc<SubscriptionManager>,
}

#[async_trait]
impl RequestHandler for SubAccPushHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let req: futu_proto::trd_sub_acc_push::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        for &acc_id in &req.c2s.acc_id_list {
            self.subscriptions.subscribe_trd_acc(conn_id, acc_id);
        }
        let resp = futu_proto::trd_sub_acc_push::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_sub_acc_push::S2c {}),
        };
        Some(prost::Message::encode_to_vec(&resp))
    }
}

// ===== UnlockTrade =====
// C++ 发送 CMD 2900 (TradePswdAuthReq) 到后端进行解锁
// 后端验证密码后返回 cipher，标记账户为已解锁
struct UnlockTradeHandler {
    backend: crate::bridge::SharedBackend,
    cache: Arc<TrdCache>,
}

/// 后端交易密码验证命令 ID
const CMD_TRD_PWD_AUTH: u16 = 2900;

#[async_trait]
impl RequestHandler for UnlockTradeHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        use futu_backend::proto_internal::ft_cmd_trade_auth;

        let req: futu_proto::trd_unlock_trade::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let is_unlock = req.c2s.unlock;
        let pwd_md5 = req.c2s.pwd_md5.as_deref().unwrap_or("");

        tracing::info!(
            unlock = is_unlock,
            has_pwd = !pwd_md5.is_empty(),
            "trade unlock request"
        );

        // 如果是锁定操作或没有后端连接，直接返回成功
        if !is_unlock || pwd_md5.is_empty() {
            let resp = futu_proto::trd_unlock_trade::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::trd_unlock_trade::S2c {}),
            };
            return Some(prost::Message::encode_to_vec(&resp));
        }

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(-1, "UnlockTrade: 后端未连接"));
            }
        };

        // 构建 TradePswdAuthReq: 只对 ACTIVE 的真实账户解锁
        let accounts = self.cache.get_accounts();
        let user_id = backend.user_id.load(std::sync::atomic::Ordering::Relaxed) as u64;
        let mut pswds = Vec::new();
        for acc in &accounts {
            if acc.trd_env == 1 && acc.acc_status == Some(0) {
                // 只对 ACTIVE 真实账户解锁
                pswds.push(ft_cmd_trade_auth::AccountPswd {
                    account_id: acc.acc_id,
                    trade_pswd_hash: Some(pwd_md5.to_string()),
                });
            }
        }

        if pswds.is_empty() {
            let resp = futu_proto::trd_unlock_trade::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::trd_unlock_trade::S2c {}),
            };
            return Some(prost::Message::encode_to_vec(&resp));
        }

        let auth_req = ft_cmd_trade_auth::TradePswdAuthReq {
            uid: user_id,  // 用户 ID (C++ GetBrokerCustomID)
            flag: Some(1), // 1 = unlock
            pswds,
            bio_id_hash: None,
            bio_trade_sig: None,
        };

        tracing::info!(
            account_count = auth_req.pswds.len(),
            "sending CMD2900 TradePswdAuth to backend"
        );

        // 发送 CMD 2900 (加密命令)
        match backend
            .request(CMD_TRD_PWD_AUTH, prost::Message::encode_to_vec(&auth_req))
            .await
        {
            Ok(resp_frame) => {
                let auth_rsp: ft_cmd_trade_auth::TradePswdAuthRsp =
                    match prost::Message::decode(resp_frame.body.as_ref()) {
                        Ok(r) => r,
                        Err(e) => {
                            tracing::warn!(error = %e, "TradePswdAuth response decode failed");
                            return Some(super::make_error_response(-1, "解锁响应解析失败"));
                        }
                    };

                let result = auth_rsp.result_code;
                if result == 0 {
                    // 存储 cipher 到 TrdCache — PlaceOrder 需要
                    for cipher_item in &auth_rsp.ciphers {
                        if let Some(ref trade_cipher) = cipher_item.trade_cipher {
                            self.cache
                                .set_cipher(cipher_item.account_id, trade_cipher.clone());
                        }
                    }
                    let cipher_count = auth_rsp.ciphers.len();
                    tracing::info!(cipher_count, "trade unlock succeeded, ciphers stored");
                    let resp = futu_proto::trd_unlock_trade::Response {
                        ret_type: 0,
                        ret_msg: None,
                        err_code: None,
                        s2c: Some(futu_proto::trd_unlock_trade::S2c {}),
                    };
                    Some(prost::Message::encode_to_vec(&resp))
                } else {
                    let err_msg = auth_rsp.err_msg.unwrap_or_else(|| "密码错误".to_string());
                    tracing::warn!(result, err = %err_msg, "trade unlock failed");
                    Some(super::make_error_response(-1, &err_msg))
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, "TradePswdAuth request failed");
                Some(super::make_error_response(-1, &format!("UnlockTrade: {e}")))
            }
        }
    }
}

// ===== PlaceOrder — FTAPI → 后端 OrderNewReq 字段映射 =====
struct PlaceOrderHandler {
    backend: crate::bridge::SharedBackend,
    cache: Arc<TrdCache>,
}

/// 后端下单命令 ID
const CMD_PLACE_ORDER: u16 = 4701;

/// 将 FTAPI OrderType 映射为后端 OrderType:
/// FTAPI Normal(1) / AbsoluteLimit(5) → backend LIMIT(1)
/// FTAPI Market(2) → backend MARKET(3)
/// 其他值直接透传
fn map_order_type(ftapi_order_type: i32) -> u32 {
    match ftapi_order_type {
        1 | 5 => 1, // Normal / AbsoluteLimit → LIMIT
        2 => 3,     // Market → MARKET
        v => v as u32,
    }
}

/// 将 f64 格式化为字符串，去除多余的尾部零
fn format_qty(v: f64) -> String {
    // 如果是整数则不带小数点
    if v.fract() == 0.0 {
        format!("{}", v as i64)
    } else {
        // 移除尾部零
        let s = format!("{}", v);
        s.trim_end_matches('0').trim_end_matches('.').to_string()
    }
}

fn format_price(v: f64) -> String {
    format_qty(v)
}

#[async_trait]
impl RequestHandler for PlaceOrderHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        use futu_backend::proto_internal::{odr_sys_cmn, order_sys_interface};

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(-1, "PlaceOrder: 后端未连接"));
            }
        };

        // 1. 解析 FTAPI 请求
        let req: futu_proto::trd_place_order::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;
        let header = &c2s.header;
        let acc_id = header.acc_id;

        // 2. 构建后端 OrderNewReq (cipher 从 TrdCache 获取)
        let cipher = self.cache.get_cipher(acc_id);
        let backend_req = order_sys_interface::OrderNewReq {
            msg_header: Some(odr_sys_cmn::MsgHeader {
                req_id: Some(format!("{acc_id}_{}", request.serial_no)),
                account_id: Some(acc_id),
                cipher,
                security_type: Some(0), // SECURITY_TYPE_COMMON
                exchange_code: None,
                input_source: Some(6), // QUANTITATIVE_TRADING
                sub_account_id: None,
            }),
            order_type: Some(map_order_type(c2s.order_type)),
            order_side: Some(c2s.trd_side as u32),
            symbol: Some(c2s.code.clone()),
            qty: Some(format_qty(c2s.qty)),
            price: c2s.price.map(format_price),
            expire_time: None,
            text: c2s.remark.clone(),
            lot_type: None,
            trigger_info: None,
            algo_info: None,
        };

        tracing::info!(
            acc_id,
            symbol = %c2s.code,
            side = c2s.trd_side,
            order_type = c2s.order_type,
            qty = c2s.qty,
            "placing order via backend"
        );

        // 3. 发送到后端 CMD 4701 (加密命令)
        let resp = match backend
            .request(CMD_PLACE_ORDER, prost::Message::encode_to_vec(&backend_req))
            .await
        {
            Ok(r) => r,
            Err(e) => {
                tracing::warn!(error = %e, "PlaceOrder backend request failed");
                return Some(super::make_error_response(-1, &format!("PlaceOrder: {e}")));
            }
        };

        // 4. 解析后端响应
        let backend_rsp: order_sys_interface::OrderNewRsp =
            match prost::Message::decode(resp.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "PlaceOrder response decode failed");
                    return Some(super::make_error_response(-1, "PlaceOrder: 响应解析失败"));
                }
            };

        // 5. 检查后端返回结果
        if backend_rsp.result.unwrap_or(-1) != 0 {
            let err_msg = backend_rsp
                .err_msg
                .unwrap_or_else(|| "unknown error".to_string());
            tracing::warn!(err_msg = %err_msg, "PlaceOrder backend returned error");
            return Some(super::make_error_response(
                -1,
                &format!("PlaceOrder: {err_msg}"),
            ));
        }

        // 6. 构建 FTAPI 响应
        let order_id: u64 = backend_rsp
            .order_id
            .as_ref()
            .and_then(|s| s.parse().ok())
            .unwrap_or(0);

        let ftapi_resp = futu_proto::trd_place_order::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_place_order::S2c {
                header: *header,
                order_id: Some(order_id),
                order_id_ex: backend_rsp.order_id,
            }),
        };

        tracing::info!(order_id, "PlaceOrder success");
        Some(prost::Message::encode_to_vec(&ftapi_resp))
    }
}

// ===== ModifyOrder — FTAPI → 后端 OrderReplaceReq/OrderCancelReq 字段映射 =====
struct ModifyOrderHandler {
    backend: crate::bridge::SharedBackend,
    cache: Arc<TrdCache>,
}

/// 后端改单命令 ID
const CMD_MODIFY_ORDER: u16 = 4702;
/// 后端撤单命令 ID
const CMD_CANCEL_ORDER: u16 = 4703;

#[async_trait]
impl RequestHandler for ModifyOrderHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        use futu_backend::proto_internal::{odr_sys_cmn, order_sys_interface};

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(-1, "ModifyOrder: 后端未连接"));
            }
        };

        // 1. 解析 FTAPI 请求
        let req: futu_proto::trd_modify_order::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let c2s = &req.c2s;
        let header = &c2s.header;
        let acc_id = header.acc_id;
        let order_id = c2s.order_id;

        // ModifyOrderOp: 1=Normal(改单), 2=Cancel(撤单)
        let modify_op = c2s.modify_order_op;

        let cipher = self.cache.get_cipher(acc_id);
        let msg_header = odr_sys_cmn::MsgHeader {
            req_id: Some(format!("{acc_id}_{}", request.serial_no)),
            account_id: Some(acc_id),
            cipher,
            security_type: Some(0),
            exchange_code: None,
            input_source: Some(6),
            sub_account_id: None,
        };

        let order_id_str = if let Some(ref ex) = c2s.order_id_ex {
            ex.clone()
        } else {
            order_id.to_string()
        };

        tracing::info!(acc_id, order_id, modify_op, "modifying order via backend");

        if modify_op == 2 {
            // 撤单 → OrderCancelReq (CMD 4703)
            let cancel_all = c2s.for_all.unwrap_or(false);
            let cancel_req = order_sys_interface::OrderCancelReq {
                msg_header: Some(msg_header),
                order_id: Some(order_id_str),
                cancel_all: Some(cancel_all),
            };

            let resp = match backend
                .request(CMD_CANCEL_ORDER, prost::Message::encode_to_vec(&cancel_req))
                .await
            {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "CancelOrder backend request failed");
                    return Some(super::make_error_response(
                        -1,
                        &format!("ModifyOrder(Cancel): {e}"),
                    ));
                }
            };

            let backend_rsp: order_sys_interface::OrderCancelRsp =
                match prost::Message::decode(resp.body.as_ref()) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::warn!(error = %e, "CancelOrder response decode failed");
                        return Some(super::make_error_response(
                            -1,
                            "ModifyOrder(Cancel): 响应解析失败",
                        ));
                    }
                };

            if backend_rsp.result.unwrap_or(-1) != 0 {
                let err_msg = backend_rsp
                    .err_msg
                    .unwrap_or_else(|| "unknown error".to_string());
                return Some(super::make_error_response(
                    -1,
                    &format!("ModifyOrder(Cancel): {err_msg}"),
                ));
            }

            let ftapi_resp = futu_proto::trd_modify_order::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::trd_modify_order::S2c {
                    header: *header,
                    order_id,
                    order_id_ex: c2s.order_id_ex.clone(),
                }),
            };

            tracing::info!(order_id, "CancelOrder success");
            return Some(prost::Message::encode_to_vec(&ftapi_resp));
        }

        // 改单 (ModifyOrderOp_Normal=1) → OrderReplaceReq (CMD 4702)
        let replace_req = order_sys_interface::OrderReplaceReq {
            msg_header: Some(msg_header),
            order_id: Some(order_id_str),
            order_version: None,
            new_qty: c2s.qty.map(format_qty),
            new_price: c2s.price.map(format_price),
            trigger_info: None,
        };

        let resp = match backend
            .request(
                CMD_MODIFY_ORDER,
                prost::Message::encode_to_vec(&replace_req),
            )
            .await
        {
            Ok(r) => r,
            Err(e) => {
                tracing::warn!(error = %e, "ReplaceOrder backend request failed");
                return Some(super::make_error_response(-1, &format!("ModifyOrder: {e}")));
            }
        };

        let backend_rsp: order_sys_interface::OrderReplaceRsp =
            match prost::Message::decode(resp.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "ReplaceOrder response decode failed");
                    return Some(super::make_error_response(-1, "ModifyOrder: 响应解析失败"));
                }
            };

        if backend_rsp.result.unwrap_or(-1) != 0 {
            let err_msg = backend_rsp
                .err_msg
                .unwrap_or_else(|| "unknown error".to_string());
            return Some(super::make_error_response(
                -1,
                &format!("ModifyOrder: {err_msg}"),
            ));
        }

        let ftapi_resp = futu_proto::trd_modify_order::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_modify_order::S2c {
                header: *header,
                order_id,
                order_id_ex: c2s.order_id_ex.clone(),
            }),
        };

        tracing::info!(order_id, "ModifyOrder success");
        Some(prost::Message::encode_to_vec(&ftapi_resp))
    }
}

// ===== GetMarginRatio — CMD 2309 =====
struct GetMarginRatioHandler {
    backend: crate::bridge::SharedBackend,
    #[expect(dead_code)]
    cache: Arc<TrdCache>,
    #[expect(dead_code)]
    static_cache: Arc<StaticDataCache>,
}

/// 后端保证金率查询命令 ID
const CMD_MARGIN_RATIO: u16 = 2309;

/// 将 FTAPI QotMarket → 后端 exchange_code 字符串
fn qot_market_to_exchange(qot_market: i32) -> &'static str {
    match qot_market {
        1 => "SEHK", // QotMarket_HK_Security
        2 => "US",   // QotMarket_US_Security
        3 => "SSE",  // QotMarket_CNSH_Security
        4 => "SZSE", // QotMarket_CNSZ_Security
        6 => "SGX",  // QotMarket_SG_Security
        _ => "SEHK", // fallback
    }
}

/// 将后端 exchange_code → FTAPI QotMarket
fn exchange_to_qot_market(exchange: &str) -> i32 {
    match exchange {
        "SEHK" | "HKFE" => 1,
        "US" | "NYMEX" | "COMEX" | "CBOT" | "CME" | "CBOE" => 2,
        "SSE" => 3,
        "SZSE" => 4,
        "SGX" => 6,
        "OSE" => 23,
        "CN" => 3,
        _ => 0,
    }
}

/// 解析卖空池剩余量字符串（如 "1.5M", "3万"）→ f64
fn parse_short_pool_remain(s: &str) -> f64 {
    let mut number_str = String::new();
    for ch in s.chars() {
        if ch.is_ascii_digit() || ch == '.' {
            number_str.push(ch);
        }
    }
    let base: f64 = number_str.parse().unwrap_or(0.0);
    if s.contains('M') {
        base * 1_000_000.0
    } else if s.contains('K') {
        base * 1_000.0
    } else if s.contains('\u{4e07}') {
        // '万'
        base * 10_000.0
    } else {
        base
    }
}

#[async_trait]
impl RequestHandler for GetMarginRatioHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        use futu_backend::proto_internal::stock_ratio_info;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(-1, "GetMarginRatio: 后端未连接"));
            }
        };

        let req: futu_proto::trd_get_margin_ratio::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let header = req.c2s.header;
        let user_id = backend.user_id.load(std::sync::atomic::Ordering::Relaxed) as u64;

        // 构建后端 QueryMarginPropertyReq
        let mut stocks = Vec::new();
        for sec in &req.c2s.security_list {
            let exchange = qot_market_to_exchange(sec.market);

            stocks.push(stock_ratio_info::StockInfo {
                stock_id: sec.code.clone(),
                exchange_code: exchange.to_string(),
                uid: user_id,
            });
        }

        let backend_req = stock_ratio_info::QueryMarginPropertyReq { stocks, period: 0 };

        tracing::info!(
            acc_id = header.acc_id,
            stock_count = req.c2s.security_list.len(),
            "querying margin ratio from backend"
        );

        let resp = match backend
            .request(
                CMD_MARGIN_RATIO,
                prost::Message::encode_to_vec(&backend_req),
            )
            .await
        {
            Ok(r) => r,
            Err(e) => {
                tracing::warn!(error = %e, "GetMarginRatio backend request failed");
                return Some(super::make_error_response(
                    -1,
                    &format!("GetMarginRatio: {e}"),
                ));
            }
        };

        let backend_rsp: stock_ratio_info::QueryMarginPropertyRsp =
            match prost::Message::decode(resp.body.as_ref()) {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "GetMarginRatio response decode failed");
                    return Some(super::make_error_response(
                        -1,
                        "GetMarginRatio: 响应解析失败",
                    ));
                }
            };

        if backend_rsp.err_code != 0 {
            let err_msg = if backend_rsp.err_msg.is_empty() {
                "unknown error".to_string()
            } else {
                backend_rsp.err_msg.clone()
            };
            tracing::warn!(err_code = backend_rsp.err_code, err_msg = %err_msg, "GetMarginRatio backend error");
            return Some(super::make_error_response(
                -1,
                &format!("GetMarginRatio: {err_msg}"),
            ));
        }

        // 映射后端响应 → FTAPI MarginRatioInfo
        // C++ 将所有 ratio 值 * 100 (从小数转百分比)
        let margin_ratio_info_list: Vec<futu_proto::trd_get_margin_ratio::MarginRatioInfo> =
            backend_rsp
                .data
                .iter()
                .map(|p| {
                    let market = exchange_to_qot_market(&p.exchange_code);
                    futu_proto::trd_get_margin_ratio::MarginRatioInfo {
                        security: futu_proto::qot_common::Security {
                            market,
                            code: p.stock_id.clone(),
                        },
                        is_long_permit: Some(p.is_long_permit != 0),
                        is_short_permit: Some(p.is_short_permit != 0),
                        short_pool_remain: Some(parse_short_pool_remain(&p.short_pool_remain)),
                        short_fee_rate: {
                            let rate: f64 = p.short_fee_rate.parse().unwrap_or(0.0);
                            Some(rate)
                        },
                        alert_long_ratio: Some(p.alert_long_ratio * 100.0),
                        alert_short_ratio: Some(p.alert_short_ratio * 100.0),
                        im_long_ratio: Some(p.im_long_ratio * 100.0),
                        im_short_ratio: Some(p.im_short_ratio * 100.0),
                        mcm_long_ratio: Some(p.mcm_long_ratio * 100.0),
                        mcm_short_ratio: Some(p.mcm_short_ratio * 100.0),
                        mm_long_ratio: Some(p.mm_long_ratio * 100.0),
                        mm_short_ratio: Some(p.mm_short_ratio * 100.0),
                    }
                })
                .collect();

        tracing::info!(
            count = margin_ratio_info_list.len(),
            "GetMarginRatio success"
        );

        let ftapi_resp = futu_proto::trd_get_margin_ratio::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_get_margin_ratio::S2c {
                header,
                margin_ratio_info_list,
            }),
        };

        Some(prost::Message::encode_to_vec(&ftapi_resp))
    }
}

// ===== GetOrderFee — CMD 2273 =====
struct GetOrderFeeHandler {
    backend: crate::bridge::SharedBackend,
    #[expect(dead_code)]
    cache: Arc<TrdCache>,
}

/// 后端批量查询订单费用命令 ID
const CMD_BATCH_QUERY_ORDER_FEE: u16 = 2273;
/// 默认页大小 (与 C++ gs_knDefaultPageSize 一致)
const ORDER_FEE_DEFAULT_PAGE_SIZE: u32 = 400;
/// 结果上限 (与 C++ gs_knCountLimit 一致)
const ORDER_FEE_COUNT_LIMIT: usize = 400;
/// 历史开始日期 (与 C++ gs_uBeginTime 一致)
const ORDER_FEE_BEGIN_DATE: u32 = 20180101;

/// 获取明天的日期 (YYYYMMDD 格式)，与 C++ GetTomorrowDate 一致
fn get_tomorrow_date() -> u32 {
    use std::time::{Duration, SystemTime, UNIX_EPOCH};
    let tomorrow = SystemTime::now() + Duration::from_secs(86400);
    let secs = tomorrow
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();
    let days = secs / 86400;
    let (y, m, d) = days_to_ymd(days);
    (y * 10000 + m * 100 + d) as u32
}

/// 将天数 (自 1970-01-01) 转为 (year, month, day)
fn days_to_ymd(days: u64) -> (u64, u64, u64) {
    let z = days + 719468;
    let era = z / 146097;
    let doe = z - era * 146097;
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    (y, m, d)
}

#[async_trait]
impl RequestHandler for GetOrderFeeHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        use futu_backend::proto_internal::order_fee_query;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(-1, "GetOrderFee: 后端未连接"));
            }
        };

        let req: futu_proto::trd_get_order_fee::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let header = req.c2s.header;
        let acc_id = header.acc_id;

        // 如果没有订单ID, 直接返回空列表 (与 C++ 一致)
        if req.c2s.order_id_ex_list.is_empty() {
            let ftapi_resp = futu_proto::trd_get_order_fee::Response {
                ret_type: 0,
                ret_msg: None,
                err_code: None,
                s2c: Some(futu_proto::trd_get_order_fee::S2c {
                    header,
                    order_fee_list: vec![],
                }),
            };
            return Some(prost::Message::encode_to_vec(&ftapi_resp));
        }

        let end_date = get_tomorrow_date();

        tracing::info!(
            acc_id,
            order_count = req.c2s.order_id_ex_list.len(),
            "querying order fee from backend"
        );

        // 分页循环获取 (与 C++ 一致)
        let mut all_order_fees: Vec<order_fee_query::OrderFee> = Vec::new();
        let mut page_index: u32 = 0;

        loop {
            let backend_req = order_fee_query::BatchQueryOrderFeeReq {
                uniq_id: acc_id,
                begin_date: ORDER_FEE_BEGIN_DATE,
                end_date,
                exchange: vec![],
                order_id: req.c2s.order_id_ex_list.clone(),
                page_index,
                page_size: ORDER_FEE_DEFAULT_PAGE_SIZE,
            };

            let resp = match backend
                .request(
                    CMD_BATCH_QUERY_ORDER_FEE,
                    prost::Message::encode_to_vec(&backend_req),
                )
                .await
            {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "GetOrderFee backend request failed");
                    return Some(super::make_error_response(-1, &format!("GetOrderFee: {e}")));
                }
            };

            let backend_rsp: order_fee_query::BatchQueryOrderFeeRsp =
                match prost::Message::decode(resp.body.as_ref()) {
                    Ok(r) => r,
                    Err(e) => {
                        tracing::warn!(error = %e, "GetOrderFee response decode failed");
                        return Some(super::make_error_response(-1, "GetOrderFee: 响应解析失败"));
                    }
                };

            if backend_rsp.err_code != 0 {
                let err_msg = if backend_rsp.err_msg.is_empty() {
                    "unknown error".to_string()
                } else {
                    backend_rsp.err_msg.clone()
                };
                tracing::warn!(err_msg = %err_msg, "GetOrderFee backend error");
                return Some(super::make_error_response(
                    -1,
                    &format!("GetOrderFee: {err_msg}"),
                ));
            }

            all_order_fees.extend(backend_rsp.order_fee_list);

            let total = backend_rsp.total;
            let page_size = backend_rsp.page_size.max(ORDER_FEE_DEFAULT_PAGE_SIZE);
            let completed = (page_index + 1) * page_size >= total
                || all_order_fees.len() >= ORDER_FEE_COUNT_LIMIT;

            if completed {
                break;
            }
            page_index += 1;
        }

        all_order_fees.truncate(ORDER_FEE_COUNT_LIMIT);

        // 映射后端 OrderFee → FTAPI Trd_Common::OrderFee
        let order_fee_list: Vec<futu_proto::trd_common::OrderFee> = all_order_fees
            .iter()
            .map(|f| {
                let mut total: f64 = 0.0;
                let mut fee_items: Vec<futu_proto::trd_common::OrderFeeItem> = Vec::new();

                for group in &f.fee_group_list {
                    for item in &group.fee_list {
                        let value: f64 = item.val.parse().unwrap_or(0.0);
                        if !item.title.is_empty() {
                            fee_items.push(futu_proto::trd_common::OrderFeeItem {
                                title: Some(item.title.clone()),
                                value: Some(value),
                            });
                            total += value;
                        }
                    }
                }

                futu_proto::trd_common::OrderFee {
                    order_id_ex: f.order_id.clone(),
                    fee_amount: Some(total),
                    fee_list: fee_items,
                }
            })
            .collect();

        tracing::info!(count = order_fee_list.len(), "GetOrderFee success");

        let ftapi_resp = futu_proto::trd_get_order_fee::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_get_order_fee::S2c {
                header,
                order_fee_list,
            }),
        };

        Some(prost::Message::encode_to_vec(&ftapi_resp))
    }
}

// ===== FlowSummary — CMD 20963 =====
struct FlowSummaryHandler {
    backend: crate::bridge::SharedBackend,
    cache: Arc<TrdCache>,
}

/// 后端查询资金流水命令 ID
const CMD_FLOW_SUMMARY: u16 = 20963;
/// 流水分页大小 (与 C++ nMaxPageSize 一致)
const FLOW_SUMMARY_PAGE_SIZE: i32 = 2000;

/// 将日期 YYYYMMDD → "YYYY-MM-DD" 格式
fn format_date_str(date: i32) -> String {
    let y = date / 10000;
    let m = (date / 100) % 100;
    let d = date % 100;
    format!("{y:04}-{m:02}-{d:02}")
}

/// 解析 "YYYY-MM-DD" 日期 → YYYYMMDD 整数
fn parse_clearing_date(s: &str) -> Option<i32> {
    let parts: Vec<&str> = s.split('-').collect();
    if parts.len() == 3 {
        let y: i32 = parts[0].parse().ok()?;
        let m: i32 = parts[1].parse().ok()?;
        let d: i32 = parts[2].parse().ok()?;
        Some(y * 10000 + m * 100 + d)
    } else {
        None
    }
}

/// 币种字符串 → FTAPI 币种枚举值
fn currency_str_to_enum(ccy: &str) -> i32 {
    match ccy {
        "HKD" => 1,
        "USD" => 2,
        "CNH" | "CNY" | "RMB" => 3,
        "JPY" => 4,
        "SGD" => 5,
        "AUD" => 6,
        "CAD" => 7,
        "GBP" => 8,
        "MYR" => 9,
        _ => 0,
    }
}

/// 从 acc_id 提取内部 account_id (低 32 位)
fn extract_intra_acc_id(acc_id: u64) -> i64 {
    (acc_id & 0xFFFF_FFFF) as i64
}

/// 从 acc_id 提取 market_id (第 32~47 位)
fn extract_market_id(acc_id: u64) -> i32 {
    ((acc_id >> 32) & 0xFFFF) as i32
}

#[async_trait]
impl RequestHandler for FlowSummaryHandler {
    async fn handle(&self, _conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        use futu_backend::proto_internal::soa_std_data;

        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(-1, "FlowSummary: 后端未连接"));
            }
        };

        let req: futu_proto::trd_flow_summary::Request =
            prost::Message::decode(request.body.as_ref()).ok()?;
        let header = req.c2s.header;
        let acc_id = header.acc_id;

        let clearing_date = match parse_clearing_date(&req.c2s.clearing_date) {
            Some(d) => d,
            None => {
                return Some(super::make_error_response(
                    -1,
                    "FlowSummary: 日期格式错误，需 YYYY-MM-DD",
                ));
            }
        };

        let intra_acc_id = extract_intra_acc_id(acc_id);
        let market_id = extract_market_id(acc_id);
        let market = if market_id != 0 {
            market_id
        } else {
            self.cache
                .accounts
                .get(&acc_id)
                .and_then(|a| a.trd_market)
                .unwrap_or(1)
        };

        tracing::info!(
            acc_id,
            clearing_date,
            intra_acc_id,
            market,
            "querying flow summary from backend"
        );

        // 分页循环获取全部数据 (与 C++ 一致)
        let mut all_items: Vec<soa_std_data::FlowSummaryItem> = Vec::new();
        let mut start_id: u64 = 0;

        loop {
            let backend_req = soa_std_data::FlowSummaryReq {
                query_comm: Some(soa_std_data::QueryComm {
                    date: clearing_date,
                    market,
                    identifier: "std".to_string(),
                    start_id,
                    end_id: 0,
                    page_size: FLOW_SUMMARY_PAGE_SIZE,
                    cond_fields: vec![soa_std_data::CondField {
                        op: soa_std_data::Op::WhereIn as i32,
                        field_name: "account_id".to_string(),
                        string_list: vec![],
                        int32_list: vec![],
                        int64_list: vec![intra_acc_id],
                    }],
                    ret_fields: vec![],
                    persist: String::new(),
                }),
            };

            let resp = match backend
                .request(
                    CMD_FLOW_SUMMARY,
                    prost::Message::encode_to_vec(&backend_req),
                )
                .await
            {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "FlowSummary backend request failed");
                    if !all_items.is_empty() {
                        break;
                    }
                    return Some(super::make_error_response(-1, &format!("FlowSummary: {e}")));
                }
            };

            let backend_rsp: soa_std_data::FlowSummaryRsp =
                match prost::Message::decode(resp.body.as_ref()) {
                    Ok(r) => r,
                    Err(_) => {
                        // 空响应按成功处理 (C++: IllegalCallBack → 成功)
                        break;
                    }
                };

            let item_count = backend_rsp.items.len();

            let max_item_id = backend_rsp
                .items
                .iter()
                .filter_map(|item| if item.id > 0 { Some(item.id) } else { None })
                .max()
                .unwrap_or(0);

            all_items.extend(backend_rsp.items);

            let mut completed = true;
            if let Some(ref info) = backend_rsp.instance_info {
                if item_count as i32 >= FLOW_SUMMARY_PAGE_SIZE
                    && info.max_id > 0
                    && max_item_id != info.max_id
                {
                    completed = false;
                    start_id = max_item_id + 1;
                }
            }

            if completed {
                break;
            }
        }

        tracing::info!(
            total_items = all_items.len(),
            "FlowSummary raw items retrieved"
        );

        // 过滤 + 映射
        // C++ 过滤: trade_flag != 1 && (fAmount != 0 || fTotalFee != 0)
        let cash_flow_direction_filter = req.c2s.cash_flow_direction.unwrap_or(0);

        let flow_summary_info_list: Vec<futu_proto::trd_flow_summary::FlowSummaryInfo> = all_items
            .iter()
            .filter_map(|item| {
                let amount: f64 = item.net_amount.parse().unwrap_or(0.0);
                let total_fee: f64 = item.total_fee.parse().unwrap_or(0.0);
                let trade_flag = item.trade_flag;

                if trade_flag == 1 || (amount == 0.0 && total_fee == 0.0) {
                    return None;
                }

                let direction = if amount > 0.0 {
                    1 // TrdCashFlowDirection_In
                } else if amount < 0.0 {
                    2 // TrdCashFlowDirection_Out
                } else {
                    0
                };

                if cash_flow_direction_filter != 0 && direction != cash_flow_direction_filter {
                    return None;
                }

                let clearing_date_str = format_date_str(item.clearing_date);
                let settlement_date_str = format_date_str(item.settlement_date);
                let currency = currency_str_to_enum(&item.ccy);

                Some(futu_proto::trd_flow_summary::FlowSummaryInfo {
                    clearing_date: Some(clearing_date_str),
                    settlement_date: Some(settlement_date_str),
                    currency: Some(currency),
                    cash_flow_type: Some(item.comment.clone()),
                    cash_flow_direction: Some(direction),
                    cash_flow_amount: Some(amount),
                    cash_flow_remark: Some(item.comment.clone()),
                    cash_flow_id: Some(item.id),
                })
            })
            .collect();

        tracing::info!(count = flow_summary_info_list.len(), "FlowSummary success");

        let ftapi_resp = futu_proto::trd_flow_summary::Response {
            ret_type: 0,
            ret_msg: None,
            err_code: None,
            s2c: Some(futu_proto::trd_flow_summary::S2c {
                header,
                flow_summary_info_list,
            }),
        };

        Some(prost::Message::encode_to_vec(&ftapi_resp))
    }
}

// ===== 需要后端转发的交易操作处理器 =====
struct BackendTrdForwardHandler {
    name: &'static str,
    backend: crate::bridge::SharedBackend,
    /// FTAPI proto_id → 后端 cmd_id 映射
    backend_cmd_id: Option<u16>,
}

#[async_trait]
impl RequestHandler for BackendTrdForwardHandler {
    async fn handle(&self, conn_id: u64, request: &IncomingRequest) -> Option<Vec<u8>> {
        let backend = match super::load_backend(&self.backend) {
            Some(b) => b,
            None => {
                return Some(super::make_error_response(
                    -1,
                    &format!("{}: 后端未连接", self.name),
                ));
            }
        };

        let backend_cmd = match self.backend_cmd_id {
            Some(cmd) => cmd,
            None => {
                // 没有对应的后端命令，返回错误
                return Some(super::make_error_response(
                    -1,
                    &format!("{}: 暂不支持", self.name),
                ));
            }
        };

        tracing::info!(
            proto_id = request.proto_id,
            conn_id,
            handler = self.name,
            backend_cmd,
            "forwarding trade request to backend"
        );

        // 直接将 FTAPI 请求体转发到后端
        // 注意: 后端交易命令是加密的 (不在未加密列表中)
        match backend.request(backend_cmd, request.body.to_vec()).await {
            Ok(resp) => {
                tracing::debug!(
                    handler = self.name,
                    body_len = resp.body.len(),
                    "backend trade response"
                );
                // 直接返回后端响应给客户端
                // 注意: 后端响应格式可能和 FTAPI 不同，需要转换
                // 暂时先直接返回
                Some(resp.body.to_vec())
            }
            Err(e) => {
                tracing::warn!(
                    handler = self.name,
                    error = %e,
                    "backend trade request failed"
                );
                Some(super::make_error_response(
                    -1,
                    &format!("{}: {e}", self.name),
                ))
            }
        }
    }
}