futu_core/

delay_stats.rs

//! C++ `INNData_ProtoDelay`-style local delay statistics.
//!
//! The request/reply statistic needs to observe both public API dispatch and
//! backend request spans.  Public API dispatch is wired from `futu-server`,
//! while backend spans are wired in `BackendConn::request_with_reserved_timeout`.
//! Keeping this store in `futu-core` avoids a dependency from `futu-server` back
//! to the whole backend crate while preserving one process-wide stats store.

use std::collections::HashMap;
use std::future::Future;
use std::sync::Arc;
use std::sync::LazyLock;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use parking_lot::Mutex;

mod qot_push;
mod response;

use qot_push::QotPushCounter;
use response::response_ret_type_is_success;

#[derive(Debug, Clone, PartialEq)]
pub struct ReqReplyStatisticsSnapshot {
    pub proto_id: u32,
    pub count: u32,
    pub total_cost_avg_ms: f32,
    pub open_d_cost_avg_ms: f32,
    pub net_delay_avg_ms: f32,
    pub is_local_reply: bool,
}

#[derive(Debug, Clone, PartialEq)]
pub struct QotPushStatisticsSnapshot {
    pub qot_push_type: i32,
    pub item_list: Vec<DelayStatisticsItemSnapshot>,
    pub delay_avg_ms: f32,
    pub count: i32,
}

#[derive(Debug, Clone, PartialEq)]
pub struct PlaceOrderStatisticsSnapshot {
    pub order_id: String,
    pub total_cost_ms: f32,
    pub open_d_cost_ms: f32,
    pub net_delay_ms: f32,
    pub update_cost_ms: f32,
}

#[derive(Debug, Clone, PartialEq)]
pub struct DelayStatisticsItemSnapshot {
    pub begin: i32,
    pub end: i32,
    pub count: i32,
    pub proportion: f32,
    pub cumulative_ratio: f32,
}

pub const QOT_PUSH_TYPE_PRICE: i32 = 1;
pub const QOT_PUSH_TYPE_TICKER: i32 = 2;
pub const QOT_PUSH_TYPE_ORDER_BOOK: i32 = 3;
pub const QOT_PUSH_TYPE_BROKER: i32 = 4;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QotPushDelayRecordOutcome {
    Recorded,
    SkippedByCountGate,
    SkippedByNotCalDelay,
}

#[derive(Debug, Clone, Copy)]
pub struct QotPushDelaySample {
    pub count_delay: bool,
    pub not_cal_delay: Option<i32>,
    pub qot_push_type: i32,
    pub server_recv_from_exchange_time_ms: Option<i64>,
    pub server_send_to_client_time_ms: Option<i64>,
    pub f3c_recv_at: SystemTime,
    pub api_response_at: SystemTime,
}

const QOT_PUSH_STAGE_SR2SS: i32 = 1;
const QOT_PUSH_STAGE_SS2CR: i32 = 2;
const QOT_PUSH_STAGE_CR2CS: i32 = 3;
const QOT_PUSH_STAGE_SS2CS: i32 = 4;
const QOT_PUSH_STAGE_SR2CS: i32 = 5;

#[derive(Default)]
struct ReqReplyCounter {
    count: u32,
    total_cost_avg_ms: f64,
    open_d_cost_avg_ms: f64,
    net_delay_avg_ms: f64,
    is_local_reply: bool,
}

impl ReqReplyCounter {
    fn add(&mut self, total: Duration, open_d: Duration, net_delay: Duration, local_reply: bool) {
        let count = self.count as f64;
        self.total_cost_avg_ms =
            ((self.total_cost_avg_ms * count) + duration_ms(total)) / (count + 1.0);
        self.open_d_cost_avg_ms =
            ((self.open_d_cost_avg_ms * count) + duration_ms(open_d)) / (count + 1.0);
        self.net_delay_avg_ms =
            ((self.net_delay_avg_ms * count) + duration_ms(net_delay)) / (count + 1.0);
        self.count += 1;
        if local_reply {
            self.is_local_reply = true;
        }
    }

    fn snapshot(&self, proto_id: u32) -> ReqReplyStatisticsSnapshot {
        ReqReplyStatisticsSnapshot {
            proto_id,
            count: self.count,
            total_cost_avg_ms: self.total_cost_avg_ms as f32,
            open_d_cost_avg_ms: self.open_d_cost_avg_ms as f32,
            net_delay_avg_ms: self.net_delay_avg_ms as f32,
            is_local_reply: self.is_local_reply,
        }
    }
}

#[derive(Default)]
struct DelayStatisticsStore {
    req_reply_counts: Mutex<HashMap<u32, ReqReplyCounter>>,
    qot_push_counts: Mutex<HashMap<i32, QotPushCounter>>,
    place_order_req_details: Mutex<Vec<PlaceOrderReqDetail>>,
    place_order_push_once: Mutex<Vec<PlaceOrderPushOnce>>,
    time_adjustment: Mutex<TimeAdjustment>,
}

impl DelayStatisticsStore {
    fn set_time_adjustment(&self, s2c_time_diff_us: i64, net_delay_us: i64) {
        *self.time_adjustment.lock() = TimeAdjustment {
            s2c_time_diff_us,
            net_delay_us,
        };
    }

    fn time_adjustment(&self) -> TimeAdjustment {
        *self.time_adjustment.lock()
    }

    fn record_req_reply(&self, ctx: &ApiRequestDelayContext) {
        let total_cost = ctx.begin.elapsed();
        let api_end_at = SystemTime::now();
        let backend_spans = ctx.backend_spans.lock();
        let is_local_reply = backend_spans.is_empty();
        let backend_cost = backend_union_duration(&backend_spans);
        let open_d_cost = total_cost.saturating_sub(backend_cost);

        self.req_reply_counts
            .lock()
            .entry(ctx.proto_id)
            .or_default()
            .add(total_cost, open_d_cost, Duration::ZERO, is_local_reply);

        self.record_place_order_req_detail(ctx, &backend_spans, api_end_at);
    }

    fn snapshot_req_reply(&self) -> Vec<ReqReplyStatisticsSnapshot> {
        self.snapshot_req_reply_with(|| {})
    }

    fn snapshot_req_reply_with(&self, hook: impl FnOnce()) -> Vec<ReqReplyStatisticsSnapshot> {
        let counts = self.req_reply_counts.lock();
        hook();
        let mut items: Vec<_> = counts
            .iter()
            .map(|(&proto_id, counter)| counter.snapshot(proto_id))
            .collect();
        items.sort_by_key(|item| item.proto_id);
        items
    }

    #[cfg(test)]
    fn record_req_reply_sample_for_test(&self, proto_id: u32, total: Duration) {
        self.req_reply_counts
            .lock()
            .entry(proto_id)
            .or_default()
            .add(total, total, Duration::ZERO, true);
    }

    #[cfg(test)]
    fn snapshot_req_reply_with_test_hook(
        &self,
        hook: impl FnOnce(),
    ) -> Vec<ReqReplyStatisticsSnapshot> {
        self.snapshot_req_reply_with(hook)
    }

    fn record_qot_push_count(
        &self,
        qot_push_type: i32,
        server_recv_from_exchange_time_ms: Option<i64>,
        server_send_to_client_time_ms: Option<i64>,
        f3c_recv_at: SystemTime,
        api_response_at: SystemTime,
    ) {
        let Some(f3c_recv_us) = system_time_us(f3c_recv_at) else {
            return;
        };
        let Some(api_response_us) = system_time_us(api_response_at) else {
            return;
        };

        let TimeAdjustment {
            s2c_time_diff_us,
            net_delay_us,
        } = *self.time_adjustment.lock();

        let mut server_recv_us = server_recv_from_exchange_time_ms.unwrap_or(0) * 1000;
        let mut server_send_us = server_send_to_client_time_ms.unwrap_or(0) * 1000;
        if server_recv_us != 0 {
            server_recv_us -= s2c_time_diff_us;
        }
        if server_send_us != 0 {
            server_send_us -= s2c_time_diff_us;
        }

        // Ref: FutuOpenD/Src/NNDataCenter/Other/NNData_ProtoDelay.cpp:179-202.
        // C++ corrects impossible clock skew (server send later than client
        // receive) by preserving OpenD processing cost and shifting CR/CS by
        // the measured one-way network delay.
        let mut f3c_recv_us = f3c_recv_us;
        let mut api_response_us = api_response_us;
        if server_send_us > f3c_recv_us {
            let open_d_cost_us = api_response_us.saturating_sub(f3c_recv_us);
            f3c_recv_us = server_send_us + net_delay_us;
            api_response_us = f3c_recv_us + open_d_cost_us;
        }

        let mut counts = self.qot_push_counts.lock();
        let counter = counts.entry(qot_push_type).or_default();
        counter.add(QOT_PUSH_STAGE_SR2SS, server_recv_us, server_send_us);
        counter.add(QOT_PUSH_STAGE_SS2CR, server_send_us, f3c_recv_us);
        counter.add(QOT_PUSH_STAGE_CR2CS, f3c_recv_us, api_response_us);
        counter.add(QOT_PUSH_STAGE_SS2CS, server_send_us, api_response_us);
        counter.add(QOT_PUSH_STAGE_SR2CS, server_recv_us, api_response_us);
        counter.total_count = counter.total_count.saturating_add(1);
    }

    fn record_qot_push_count_if_sampled(
        &self,
        sample: QotPushDelaySample,
    ) -> QotPushDelayRecordOutcome {
        if !sample.count_delay {
            return QotPushDelayRecordOutcome::SkippedByCountGate;
        }
        if !qot_push_should_sample_delay(sample.not_cal_delay) {
            return QotPushDelayRecordOutcome::SkippedByNotCalDelay;
        }

        self.record_qot_push_count(
            sample.qot_push_type,
            sample.server_recv_from_exchange_time_ms,
            sample.server_send_to_client_time_ms,
            sample.f3c_recv_at,
            sample.api_response_at,
        );
        QotPushDelayRecordOutcome::Recorded
    }

    fn snapshot_qot_push(
        &self,
        qot_push_stage: i32,
        segment_list: &[i32],
    ) -> Vec<QotPushStatisticsSnapshot> {
        if qot_push_stage == 0 || segment_list.len() < 2 {
            return Vec::new();
        }

        let counts = self.qot_push_counts.lock();
        let mut items = Vec::new();
        for qot_push_type in [
            QOT_PUSH_TYPE_PRICE,
            QOT_PUSH_TYPE_TICKER,
            QOT_PUSH_TYPE_ORDER_BOOK,
            QOT_PUSH_TYPE_BROKER,
        ] {
            let Some(counter) = counts.get(&qot_push_type) else {
                continue;
            };
            let Some(stage) = counter.stages.get(&qot_push_stage) else {
                continue;
            };
            if stage.total == 0 {
                continue;
            }

            let mut cumulative_count = 0i32;
            let total = stage.total as f32;
            let item_list = segment_list
                .windows(2)
                .map(|window| {
                    let begin = window[0];
                    let end = window[1];
                    let count = stage.range_count(begin, end) as i32;
                    cumulative_count += count;
                    DelayStatisticsItemSnapshot {
                        begin,
                        end,
                        count,
                        proportion: count as f32 / total * 100.0,
                        cumulative_ratio: cumulative_count as f32 / total * 100.0,
                    }
                })
                .collect();

            items.push(QotPushStatisticsSnapshot {
                qot_push_type,
                item_list,
                delay_avg_ms: stage.cost_avg_ms,
                count: stage.total as i32,
            });
        }
        items
    }

    fn record_place_order_req_detail(
        &self,
        ctx: &ApiRequestDelayContext,
        backend_spans: &[BackendSpan],
        api_end_at: SystemTime,
    ) {
        if ctx.proto_id != crate::proto_id::TRD_PLACE_ORDER {
            return;
        }
        let Some(user_data) = ctx.place_order_user_data.lock().clone() else {
            return;
        };
        let Some(api_begin_us) = system_time_us(ctx.begin_at) else {
            return;
        };
        let Some(api_end_us) = system_time_us(api_end_at) else {
            return;
        };
        if api_end_us < api_begin_us {
            return;
        }
        let first_span = backend_spans.first();
        let first_backend_begin_us = first_span.and_then(|span| system_time_us(span.begin_at));
        let first_backend_end_us = first_span.and_then(|span| system_time_us(span.end_at));

        self.place_order_req_details
            .lock()
            .push(PlaceOrderReqDetail {
                order_id: user_data.order_id,
                trd_env: user_data.trd_env,
                _market: user_data.market,
                api_begin_us,
                api_end_us,
                first_backend_begin_us,
                first_backend_end_us,
                net_delay_us: ctx.time_adjustment.net_delay_us,
            });
    }

    fn set_place_order_user_data(&self, user_data: PlaceOrderUserData) {
        let _ = CURRENT_API_REQUEST.try_with(|ctx| {
            *ctx.place_order_user_data.lock() = Some(user_data);
        });
    }

    fn record_place_order_update_push(
        &self,
        order_id: String,
        trd_env: i32,
        market: i32,
        order_status: i32,
        api_response_at: SystemTime,
    ) {
        let Some(api_response_us) = system_time_us(api_response_at) else {
            return;
        };
        self.place_order_push_once.lock().push(PlaceOrderPushOnce {
            order_id,
            _trd_env: trd_env,
            _market: market,
            order_status,
            api_response_us,
        });
    }

    fn snapshot_place_order(&self) -> Vec<PlaceOrderStatisticsSnapshot> {
        let mut earliest_push_by_order: HashMap<String, i64> = HashMap::new();
        for push in self.place_order_push_once.lock().iter() {
            if !place_order_update_status_counts(push.order_status) {
                continue;
            }
            earliest_push_by_order
                .entry(push.order_id.clone())
                .and_modify(|existing| *existing = (*existing).min(push.api_response_us))
                .or_insert(push.api_response_us);
        }

        self.place_order_req_details
            .lock()
            .iter()
            .filter(|detail| detail.trd_env != 0)
            .filter_map(|detail| {
                let total_cost_us = detail.api_end_us.checked_sub(detail.api_begin_us)?;
                let open_d_cost_us =
                    match (detail.first_backend_begin_us, detail.first_backend_end_us) {
                        (Some(begin), Some(end)) => begin
                            .saturating_sub(detail.api_begin_us)
                            .saturating_add(detail.api_end_us.saturating_sub(end)),
                        _ => 0,
                    };
                let update_cost_us = earliest_push_by_order
                    .get(&detail.order_id)
                    .map(|push_us| push_us.saturating_sub(detail.api_end_us))
                    .unwrap_or(0);

                Some(PlaceOrderStatisticsSnapshot {
                    order_id: detail.order_id.clone(),
                    total_cost_ms: us_to_ms(total_cost_us),
                    open_d_cost_ms: us_to_ms(open_d_cost_us),
                    net_delay_ms: us_to_ms(detail.net_delay_us.max(0)),
                    update_cost_ms: us_to_ms(update_cost_us),
                })
            })
            .collect()
    }
}

#[derive(Debug, Clone, Copy, Default)]
struct TimeAdjustment {
    s2c_time_diff_us: i64,
    net_delay_us: i64,
}

#[derive(Debug, Clone)]
struct PlaceOrderUserData {
    order_id: String,
    trd_env: i32,
    market: i32,
}

#[derive(Debug, Clone)]
struct PlaceOrderReqDetail {
    order_id: String,
    trd_env: i32,
    _market: i32,
    api_begin_us: i64,
    api_end_us: i64,
    first_backend_begin_us: Option<i64>,
    first_backend_end_us: Option<i64>,
    net_delay_us: i64,
}

#[derive(Debug, Clone)]
struct PlaceOrderPushOnce {
    order_id: String,
    _trd_env: i32,
    _market: i32,
    order_status: i32,
    api_response_us: i64,
}

#[derive(Debug)]
struct ApiRequestDelayContext {
    proto_id: u32,
    begin: Instant,
    begin_at: SystemTime,
    backend_spans: Mutex<Vec<BackendSpan>>,
    place_order_user_data: Mutex<Option<PlaceOrderUserData>>,
    time_adjustment: TimeAdjustment,
}

impl ApiRequestDelayContext {
    fn new(proto_id: u32) -> Self {
        Self {
            proto_id,
            begin: Instant::now(),
            begin_at: SystemTime::now(),
            backend_spans: Mutex::new(Vec::new()),
            place_order_user_data: Mutex::new(None),
            time_adjustment: DELAY_STATS.time_adjustment(),
        }
    }

    fn record_backend_span(
        &self,
        begin: Instant,
        end: Instant,
        begin_at: SystemTime,
        end_at: SystemTime,
    ) {
        if end >= begin {
            self.backend_spans.lock().push(BackendSpan {
                begin,
                end,
                begin_at,
                end_at,
            });
        }
    }
}

#[derive(Debug, Clone, Copy)]
struct BackendSpan {
    begin: Instant,
    end: Instant,
    begin_at: SystemTime,
    end_at: SystemTime,
}

static DELAY_STATS: LazyLock<DelayStatisticsStore> = LazyLock::new(DelayStatisticsStore::default);

tokio::task_local! {
    static CURRENT_API_REQUEST: Arc<ApiRequestDelayContext>;
}

/// Run a public API handler inside the current request/reply delay context.
///
/// C++ starts recording at `APIServerCS_Conn.cpp:279`
/// (`ReqReply_APIReqBegin`) and records successful responses in
/// `APIServer_Inner_API.h` via `ReqReply_APIReqEnd(..., bSuc)`.
pub async fn with_api_request<F, Fut>(
    _conn_id: u64,
    _serial_no: u32,
    proto_id: u32,
    future: F,
) -> Option<Vec<u8>>
where
    F: FnOnce() -> Fut,
    Fut: Future<Output = Option<Vec<u8>>>,
{
    let ctx = Arc::new(ApiRequestDelayContext::new(proto_id));
    let response = CURRENT_API_REQUEST.scope(ctx.clone(), future()).await;

    if response
        .as_deref()
        .is_some_and(response_ret_type_is_success)
    {
        DELAY_STATS.record_req_reply(&ctx);
    }

    response
}

/// Record a backend request span for the API request currently executing on
/// this Tokio task.  Calls outside public API dispatch are intentionally no-op,
/// matching C++'s explicit `ReqReply_RelatedAPISvrReq` linking model.
pub async fn trace_backend_request<Fut, T>(_cmd_id: u16, future: Fut) -> T
where
    Fut: Future<Output = T>,
{
    let begin = Instant::now();
    let begin_at = SystemTime::now();
    let output = future.await;
    let end_at = SystemTime::now();
    let end = Instant::now();

    let _ =
        CURRENT_API_REQUEST.try_with(|ctx| ctx.record_backend_span(begin, end, begin_at, end_at));
    output
}

pub fn snapshot_req_reply_statistics() -> Vec<ReqReplyStatisticsSnapshot> {
    DELAY_STATS.snapshot_req_reply()
}

/// Decide whether a QOT push should contribute to delay statistics.
///
/// Ref: FutuOpenD/Src/APIServer/Business/Quote/QotRealTimeData.cpp:2202-2210
/// and the sibling push paths for BrokerQueue / OrderBook / US LV2 / Crypto
/// LV2: C++ samples delay unless `not_cal_delay` is present and non-zero.
#[must_use]
pub fn qot_push_should_sample_delay(not_cal_delay: Option<i32>) -> bool {
    match not_cal_delay {
        Some(flag) => flag == 0,
        None => true,
    }
}

/// Select the server recv time used for QOT push delay statistics.
///
/// Ref: FutuOpenD/Src/APIServer/Business/Quote/QotRealTimeData.cpp:855-860,
/// 1518-1544, 1635-1661, and 1942-1972.  C++ starts from `0` and updates
/// `stPushTime.nSvrRecvTime_ms` only when a candidate recv time is greater
/// than the current value, so missing/zero timestamps contribute no sample.
#[must_use]
pub fn qot_push_max_server_recv_time_ms<I>(times: I) -> Option<i64>
where
    I: IntoIterator<Item = Option<i64>>,
{
    times
        .into_iter()
        .flatten()
        .filter(|time_ms| *time_ms > 0)
        .max()
}

pub fn record_qot_push_count(
    qot_push_type: i32,
    server_recv_from_exchange_time_ms: Option<i64>,
    server_send_to_client_time_ms: Option<i64>,
    f3c_recv_at: SystemTime,
    api_response_at: SystemTime,
) {
    DELAY_STATS.record_qot_push_count(
        qot_push_type,
        server_recv_from_exchange_time_ms,
        server_send_to_client_time_ms,
        f3c_recv_at,
        api_response_at,
    );
}

pub fn record_qot_push_count_if_sampled(sample: QotPushDelaySample) -> QotPushDelayRecordOutcome {
    DELAY_STATS.record_qot_push_count_if_sampled(sample)
}

/// Attach successful `Trd_PlaceOrder` user data to the current API request.
/// Calls outside `with_api_request` are intentionally no-op.
///
/// Ref: `APIServer_Trd_PlaceOrder.cpp:864-873` serializes `trdEnv`,
/// `trdMarket`, and backend `order.szOrderID` into `ReqReply_SetUserData`.
pub fn record_place_order_request(order_id: &str, trd_env: i32, market: i32) {
    let order_id = order_id.trim();
    if order_id.is_empty() {
        return;
    }
    DELAY_STATS.set_place_order_user_data(PlaceOrderUserData {
        order_id: order_id.to_string(),
        trd_env,
        market,
    });
}

/// Record one `Trd_UpdateOrder` push candidate for PlaceOrder update-cost
/// statistics.
///
/// Ref: `APIServer_Trd_UpdateOrder.cpp:36-38` records `Push_Once_Add` after
/// sending the FTAPI push.  `api_response_at` should therefore be captured
/// after the local push dispatch attempt.
pub fn record_place_order_update_push(
    order_id: &str,
    trd_env: i32,
    market: i32,
    order_status: i32,
    api_response_at: SystemTime,
) {
    let order_id = order_id.trim();
    if order_id.is_empty() {
        return;
    }
    DELAY_STATS.record_place_order_update_push(
        order_id.to_string(),
        trd_env,
        market,
        order_status,
        api_response_at,
    );
}

/// Update the server-to-client clock correction used by QOT push delay
/// statistics.
///
/// Ref: `FutuOpenD/Src/NNProtoCenter/Other/NNBiz_SvrTime.cpp:41-78` calls
/// `INNData_ProtoDelay::SetS2CTimeDiffAndNetDelay(nDiffTime_ms * 1000,
/// nNetDelay_ms * 1000)`. `DelayCalibrationRuntime` supplies both values from
/// the precise Platform server-time anchor and independent ICMP RTT/2. Before
/// its first successful probe, startup may install a clock-only value with
/// zero delay; probe failures preserve the last complete sample.
pub fn set_time_adjustment(s2c_time_diff_us: i64, net_delay_us: i64) {
    DELAY_STATS.set_time_adjustment(s2c_time_diff_us, net_delay_us);
}

pub fn snapshot_qot_push_statistics(
    qot_push_stage: i32,
    segment_list: &[i32],
) -> Vec<QotPushStatisticsSnapshot> {
    DELAY_STATS.snapshot_qot_push(qot_push_stage, segment_list)
}

pub fn snapshot_place_order_statistics() -> Vec<PlaceOrderStatisticsSnapshot> {
    DELAY_STATS.snapshot_place_order()
}

fn backend_union_duration(spans: &[BackendSpan]) -> Duration {
    if spans.is_empty() {
        return Duration::ZERO;
    }

    let mut spans = spans.to_vec();
    spans.sort_by_key(|span| span.begin);

    let mut total = Duration::ZERO;
    let mut cur_begin = spans[0].begin;
    let mut cur_end = spans[0].end;

    for span in spans.into_iter().skip(1) {
        if span.begin <= cur_end {
            if span.end > cur_end {
                cur_end = span.end;
            }
        } else {
            total += cur_end.duration_since(cur_begin);
            cur_begin = span.begin;
            cur_end = span.end;
        }
    }

    total + cur_end.duration_since(cur_begin)
}

fn duration_ms(duration: Duration) -> f64 {
    duration.as_secs_f64() * 1000.0
}

fn us_to_ms(us: i64) -> f32 {
    us as f32 / 1000.0
}

fn system_time_us(time: SystemTime) -> Option<i64> {
    let duration = time.duration_since(UNIX_EPOCH).ok()?;
    i64::try_from(duration.as_micros()).ok()
}

fn place_order_update_status_counts(order_status: i32) -> bool {
    // Ref: APIServer_GetDelayStatistics.cpp:323-324.
    // NN_OrderStatus_Submitted / Filled_All / Cancelled_Part / Filled_Part
    // map to FTAPI values 5 / 11 / 14 / 10.
    matches!(order_status, 5 | 11 | 14 | 10)
}

#[cfg(test)]
mod tests;