futu_backend/trade_query/orders/

query_fills.rs

//! trade_query/orders/query_fills — query_order_fills* + query_order_fills_inner + order_fill_proto
//! (v1.4.110 CC Batch J: 拆自 orders.rs L667-780 + 889-1072)

use futu_cache::trd_cache::TrdCache;
use futu_core::error::Result;

use super::super::common::hash_str_to_u64;
use crate::conn::BackendConn;
use crate::proto_internal::{odr_sys_cmn, order_sys_interface};

use super::types::{OrderFillInfo, QueryErrorMode};

use super::builders::*;
use super::helpers::*;
use super::status_helpers::*;

pub async fn query_order_fills(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
) -> Result<Vec<OrderFillInfo>> {
    query_order_fills_inner(backend, acc_id, trd_cache, QueryErrorMode::LenientCacheRead).await
}

/// v1.4.106 codex 0932 F3 [P2]: strict 版 query_order_fills for push-refresh path.
///
/// decode 失败 / `result != 0` → 返 `Err`, retry_with_exp_backoff 触发 retry.
/// 4 attempts 全失败 → caller 应 record F2 exhaust + log warn (F4 wired in
/// dispatcher, see codex 0932 F4). v1.4.106 codex 0955 F4: 同样接 `trd_cache`.
pub async fn query_order_fills_strict_for_push_refresh(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
) -> Result<Vec<OrderFillInfo>> {
    query_order_fills_inner(
        backend,
        acc_id,
        trd_cache,
        QueryErrorMode::StrictPushRefresh,
    )
    .await
}

pub fn order_fill_proto_to_info_with_market_and_jp_acc_type(
    f: &odr_sys_cmn::OrderFill,
    trd_market: Option<i32>,
    jp_acc_type: Option<i32>,
) -> OrderFillInfo {
    let (qty, price) = backend_order_fill_qty_price_for_ftapi(f);
    let fill_id_ex = f.id.clone().unwrap_or_default();
    // v1.4.106 codex 0226 F6 (parity with query_orders): backend fill_id
    // (= server-generated fill ID) 也是 alphanumeric 字符串. 用 hash
    // 派生 numeric id, 对齐 C++ 的 hash chain. empty 时返 0.
    let fill_id: u64 = if fill_id_ex.is_empty() {
        0
    } else {
        hash_str_to_u64(&fill_id_ex)
    };
    let order_id_ex = f.order_id.clone().unwrap_or_default();
    let order_id: u64 = if order_id_ex.is_empty() {
        0
    } else {
        hash_str_to_u64(&order_id_ex)
    };
    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);
    let counter_broker_id = f
        .counter_broker_id
        .as_ref()
        .and_then(|s| s.parse::<i32>().ok());
    // v1.4.106 codex 0955 F6: 用 shared helper, 与 active read path
    // (GetOrderFillListHandler) 一致 (0/1/2 对齐 C++ NN_DealStatus).
    let status = Some(backend_deal_status_to_ftapi(
        f.is_cancelled.unwrap_or(false),
        f.is_corrected.unwrap_or(false),
    ));

    OrderFillInfo {
        fill_id,
        fill_id_ex,
        order_id,
        order_id_ex,
        code: f.symbol.as_ref().cloned().unwrap_or_default(),
        name: f.stock_name.as_ref().cloned().unwrap_or_default(),
        trd_side: f.side.unwrap_or(0) as i32,
        qty,
        price,
        create_timestamp,
        update_timestamp,
        counter_broker_id,
        status,
        trd_market,
        jp_acc_type,
    }
}

/// Backend `odr_sys_cmn::OrderFill` → OpenD cache/API `OrderFillInfo`.
///
/// Ref: FutuOpenD/Src/NNProtoCenter/Trade/Deal/NNProto_Trd_DealReal.cpp:13-122.
/// C++ `UnPackDealItem` drops rows unless `id/side/market/symbol/exchange/
/// price/qty/create_time` are present, `price/qty` parse as doubles, and the
/// real-account market passes `IsValidTrdMarket(NN_TrdEnv_Real, market)`.
///
/// Simulated deal query is unsupported in C++ (`NNProto_Trd_DealSimulate.cpp:13-67`),
/// while `NOTICE_TYPE_ORDER_FILL_NTF` also calls `DealReal::UnPackDealList`, so
/// Rust uses the same real deal row gate for query-cache and direct-push paths.
pub fn try_order_fill_proto_to_info_like_cpp(f: &odr_sys_cmn::OrderFill) -> Option<OrderFillInfo> {
    try_order_fill_proto_to_info_with_jp_acc_type_like_cpp(f, None)
}

pub fn try_order_fill_proto_to_info_with_jp_acc_type_like_cpp(
    f: &odr_sys_cmn::OrderFill,
    jp_acc_type: Option<i32>,
) -> Option<OrderFillInfo> {
    f.id.as_ref()?;
    f.side?;
    let trd_market = backend_real_market_to_trd_market_like_cpp(f.market?);
    if !is_valid_real_trd_market_like_cpp(trd_market) {
        return None;
    }
    f.symbol.as_ref()?;
    f.exchange.as_ref()?;
    f.price.as_deref()?.parse::<f64>().ok()?;
    f.qty.as_deref()?.parse::<f64>().ok()?;
    f.create_time?;

    Some(order_fill_proto_to_info_with_market_and_jp_acc_type(
        f,
        Some(trd_market),
        jp_acc_type,
    ))
}

/// Match C++ `CheckRspHeaderAndGetSvrRet` response gating.
///
/// Ref: `FutuOpenD/Src/NNProtoCenter/Trade/_NNProto_Trd_Comm.h:20-30`.
pub async fn query_order_fill_info_strict_for_push_refresh(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    order_fill_ids: &[String],
    security_type: Option<u32>,
    exchange: Option<&str>,
) -> Result<Vec<OrderFillInfo>> {
    use prost::Message;

    let filtered_ids: Vec<String> = order_fill_ids
        .iter()
        .filter(|id| !id.is_empty())
        .cloned()
        .collect();
    if filtered_ids.is_empty() {
        return Ok(vec![]);
    }

    let (trd_env, _account_trd_market, _enabled_markets) =
        derive_acc_query_context(trd_cache, acc_id);
    let cmd_id = order_fill_info_cmd_for_env(trd_env);
    let req = build_order_fill_info_req_base(acc_id, security_type, exchange, &filtered_ids);
    let resp = backend.request(cmd_id, req.encode_to_vec()).await?;
    let parsed: order_sys_interface::OrderFillInfoRsp = Message::decode(resp.body.as_ref())?;

    if let Err(status_err) = backend_order_fill_info_status_like_cpp(
        parsed.result,
        parsed.msg_header.as_ref(),
        parsed.err_msg.as_deref(),
        acc_id,
        filtered_ids.len(),
        parsed.order_fills.len(),
    ) {
        return Err(futu_core::error::FutuError::ServerError {
            ret_type: status_err.result,
            msg: format!("{} (acc_id={acc_id})", status_err.message),
        });
    }

    Ok(parsed
        .order_fills
        .iter()
        .filter_map(|fill| {
            let jp_acc_type = fill
                .sub_account_id
                .and_then(|id| trd_cache.jp_acc_type_for_sub_account(acc_id, id));
            try_order_fill_proto_to_info_with_jp_acc_type_like_cpp(fill, jp_acc_type)
        })
        .collect())
}

pub async fn query_order_fills_inner(
    backend: &BackendConn,
    acc_id: u64,
    trd_cache: &TrdCache,
    error_mode: QueryErrorMode,
) -> Result<Vec<OrderFillInfo>> {
    use prost::Message;

    // v1.4.106 codex 0955 F4: req shape 对齐 C++ QueryDealList_Inner.
    let (trd_env, _account_trd_market, enabled_markets) =
        derive_acc_query_context(trd_cache, acc_id);
    let req = build_order_fill_list_req_base(acc_id, trd_env, &enabled_markets);
    let cmd_id = order_fill_list_cmd_for_env(trd_env);

    let mut all_fills = Vec::new();
    let mut page_flag: Option<String> = None;
    // v1.4.106 codex 0955 F5: 显式 track completion (parity with query_orders).
    const MAX_PAGES_FILLS: usize = 100;
    let mut fills_completed = false;

    for _page in 0..MAX_PAGES_FILLS {
        let mut paged_req = req.clone();
        paged_req.page_flag = page_flag.clone();

        let resp = match backend.request(cmd_id, paged_req.encode_to_vec()).await {
            Ok(r) => r,
            Err(e) => {
                tracing::debug!(acc_id, cmd_id, error = %e, "fill query failed");
                return Err(e);
            }
        };

        let parsed: order_sys_interface::OrderFillListRsp =
            match Message::decode(resp.body.as_ref()) {
                Ok(p) => p,
                Err(e) => {
                    // v1.4.106 codex 0932 F3 [P2]: strict 模式 → 返 Err 让上层 retry.
                    if error_mode.is_strict() {
                        tracing::warn!(
                            acc_id,
                            error = %e,
                            "v1.4.106 F3 strict: fill response decode failed → \
                             returning Err to trigger retry (push-refresh path)"
                        );
                        return Err(futu_core::error::FutuError::from(e));
                    }
                    tracing::debug!(acc_id, error = %e, "fill response decode failed (lenient)");
                    return Ok(vec![]);
                }
            };

        if let Err(status_err) = backend_order_fill_list_status_like_cpp(
            parsed.result,
            parsed.msg_header.as_ref(),
            parsed.err_msg.as_deref(),
            acc_id,
        ) {
            // v1.4.110: C++ CheckRspHeaderAndGetSvrRet also rejects missing or
            // mismatched msg_header.account_id. Treat structural response
            // issues as retryable even on lenient read paths so they cannot
            // masquerade as a legitimate empty fill snapshot.
            if error_mode.is_strict() || !status_err.is_backend_error {
                tracing::warn!(
                    acc_id,
                    result = status_err.result,
                    error = %status_err.message,
                    "v1.4.110: fill query status/header invalid -> returning Err \
                     to trigger retry (push-refresh path)"
                );
                return Err(futu_core::error::FutuError::ServerError {
                    ret_type: status_err.result,
                    msg: format!("{} (acc_id={acc_id})", status_err.message),
                });
            }
            return Ok(vec![]);
        }

        all_fills.extend(parsed.order_fills.iter().filter_map(|fill| {
            let jp_acc_type = fill
                .sub_account_id
                .and_then(|id| trd_cache.jp_acc_type_for_sub_account(acc_id, id));
            try_order_fill_proto_to_info_with_jp_acc_type_like_cpp(fill, jp_acc_type)
        }));

        // v1.4.106 codex 0955 F5: 与 query_orders 同 partial-pagination 防御.
        let parsed_completed = parsed.completed.unwrap_or(false);
        if parsed_completed {
            fills_completed = true;
            break;
        }
        match parsed.page_flag {
            Some(ref pf) if !pf.is_empty() => {
                page_flag = Some(pf.clone());
            }
            _ => {
                tracing::warn!(
                    acc_id,
                    partial_count = all_fills.len(),
                    "v1.4.106 audit 0955 F5: backend completed=false 但 page_flag \
                     缺失/空 — partial response, 不返 partial 列表防 caller 误用"
                );
                return Err(futu_core::error::FutuError::Codec(
                    "query_order_fills: partial response (completed=false + missing page_flag)"
                        .to_string(),
                ));
            }
        }
    }

    if !fills_completed {
        tracing::warn!(
            acc_id,
            max_pages = MAX_PAGES_FILLS,
            partial_count = all_fills.len(),
            "v1.4.106 audit 0955 F5: pagination 超 MAX_PAGES 仍未 completed — partial"
        );
        return Err(futu_core::error::FutuError::Codec(format!(
            "query_order_fills: pagination exceeded {MAX_PAGES_FILLS} pages without completed=true"
        )));
    }

    tracing::debug!(acc_id, count = all_fills.len(), "order fills queried");
    Ok(all_fills)
}