futu_rest/routes/qot/

mod.rs

//! 行情 REST API 路由
//!
//! 所有行情相关接口通过 proto_request 适配到现有 handler。

use axum::extract::Json;
use axum::http::StatusCode;
use bytes::Bytes;
use prost::Message;
use serde_json::Value;

use futu_codec::header::ProtoFmtType;
use futu_proto::qot_get_basic_qot;
use futu_proto::qot_get_broker;
use futu_proto::qot_get_capital_distribution;
use futu_proto::qot_get_capital_flow;
use futu_proto::qot_get_code_change;
use futu_proto::qot_get_company_executive_background;
use futu_proto::qot_get_company_executives;
use futu_proto::qot_get_company_operational_efficiency;
use futu_proto::qot_get_company_profile;
use futu_proto::qot_get_corporate_actions_buybacks;
use futu_proto::qot_get_corporate_actions_dividends;
use futu_proto::qot_get_corporate_actions_stock_splits;
use futu_proto::qot_get_daily_short_volume;
use futu_proto::qot_get_financials_earnings_price_history;
use futu_proto::qot_get_financials_earnings_price_move;
use futu_proto::qot_get_financials_revenue_breakdown;
use futu_proto::qot_get_financials_statements;
use futu_proto::qot_get_future_info;
use futu_proto::qot_get_holding_change_list;
use futu_proto::qot_get_insider_holder_list;
use futu_proto::qot_get_insider_trade_list;
use futu_proto::qot_get_ipo_list;
use futu_proto::qot_get_kl;
use futu_proto::qot_get_market_state;
use futu_proto::qot_get_option_chain;
use futu_proto::qot_get_option_exercise_probability;
use futu_proto::qot_get_option_expiration_date;
use futu_proto::qot_get_option_quote;
use futu_proto::qot_get_option_strategy;
use futu_proto::qot_get_option_strategy_analysis;
use futu_proto::qot_get_option_strategy_spread;
use futu_proto::qot_get_option_volatility;
use futu_proto::qot_get_order_book;
use futu_proto::qot_get_owner_plate;
use futu_proto::qot_get_plate_security;
use futu_proto::qot_get_plate_set;
use futu_proto::qot_get_price_reminder;
use futu_proto::qot_get_reference;
use futu_proto::qot_get_research_analyst_consensus;
use futu_proto::qot_get_research_morningstar_report;
use futu_proto::qot_get_research_rating_summary;
use futu_proto::qot_get_rt;
use futu_proto::qot_get_security_snapshot;
use futu_proto::qot_get_shareholders_holder_detail;
use futu_proto::qot_get_shareholders_holding_changes;
use futu_proto::qot_get_shareholders_institutional;
use futu_proto::qot_get_shareholders_overview;
use futu_proto::qot_get_short_interest;
use futu_proto::qot_get_static_info;
use futu_proto::qot_get_sub_info;
use futu_proto::qot_get_suspend;
use futu_proto::qot_get_ticker;
use futu_proto::qot_get_top_ten_buy_sell_brokers;
use futu_proto::qot_get_user_security;
use futu_proto::qot_get_valuation_detail;
use futu_proto::qot_get_valuation_plate_stock_list;
use futu_proto::qot_get_warrant;
use futu_proto::qot_modify_user_security;
use futu_proto::qot_option_screen;
use futu_proto::qot_request_history_kl;
use futu_proto::qot_request_history_kl_quota;
use futu_proto::qot_request_rehab;
use futu_proto::qot_request_trade_date;
use futu_proto::qot_set_price_reminder;
use futu_proto::qot_stock_filter;
use futu_proto::qot_stock_screen;
use futu_proto::qot_sub;
use futu_proto::qot_warrant_screen;
use futu_proto::skill_wrap_api;
use futu_proto::used_quota;
use futu_server::conn::IncomingRequest;

use crate::adapter::{self, JsonRequestMode, RestState, decode_json_request};

type ApiResult = Result<Json<Value>, (StatusCode, Json<Value>)>;
type RawApiResult = Result<adapter::RawJson, (StatusCode, Json<Value>)>;

/// v1.4.90 P0-B: REST 全 endpoint 共享 conn_id（替代之前每次请求 `next_conn_id()`
/// 自增 → 永久泄漏 sub-quota 直到耗尽 4000 上限）。
///
/// **背景**：v1.4.81 一直到 v1.4.89 之前，REST `proto_request` 每次都调
/// `state.next_conn_id()` 拿新 virtual conn_id（10_000_000 自增）。`SubscriptionManager`
/// 按 conn_id 记账：subscribe 把 (security, sub_type) 挂在 conn_id 下、quota
/// +=1；unsubscribe 从同一 conn_id 删除、quota -=1。但 REST `subscribe` 用 conn_id=X
/// → backend 订阅 ✓，下一次 REST `unsubscribe` 用 conn_id=Y → 找不到任何挂载 →
/// 静默 no-op，quota 永不释放。日积月累 4000 quota 耗尽，整个 daemon 不能再
/// 订阅任何东西。
///
/// **修法**：所有 REST sub-related endpoint（`subscribe` / `unsubscribe` /
/// `get_sub_info` / `query_subscription`）锁定到 `REST_SHARED_CONN`. lifecycle
/// 由 daemon 进程管，不随 REST 请求生灭。
///
/// **REST 视角合理性**：REST 是 stateless API，调用方不持续持有 daemon TCP
/// 连接，"连接 ID" 在 REST 层面没物理意义。把所有 REST 流量当作"REST gateway"
/// 这一个虚拟客户端的多次调用，对应一个固定 conn_id 是最干净的语义。v1.4.106
/// 起 quote / kline / orderbook / ticker / rt handler 也会检查
/// `SubscriptionManager::is_qot_subscribed(conn_id, ...)`，所以这些订阅门禁
/// 读路径也必须锁定到同一 `REST_SHARED_CONN`，否则 REST subscribe 后下一次
/// REST read 看不到同一个 conn 的订阅。
///
/// **取值选择**：`0xFFFF_FFFE_u64`（4_294_967_294）。`next_conn_id` 起点
/// 10_000_000，要 ~4.28B 次调用才碰到此值，daemon 早已重启。`0xFFFF_FFFF_u64`
/// 留给未来可能的 sentinel（如"all REST"广播）。
///
/// 对齐 MCP 路径行为：MCP 用单 `FutuClient` 复用底层 TCP，所有 MCP 请求自然
/// 共享同一个 daemon 分配的 conn_id，不存在该 bug。REST 现在显式实现同语义。
pub const REST_SHARED_CONN: u64 = 0xFFFF_FFFE;

/// v1.4.90 P0-B: 用 REST_SHARED_CONN 而非 `state.next_conn_id()` 派发请求。
///
/// 复刻 `adapter::proto_request_with_idempotency` 的 JSON normalize → encode →
/// dispatch → decode → JSON 流程，唯一差别：dispatch 时 `conn_id =
/// REST_SHARED_CONN`. 用于 sub-related endpoint 防 quota 泄漏（见
/// `REST_SHARED_CONN` 注释）。
///
/// 不复用 adapter::proto_request 是因为该函数硬编码 `state.next_conn_id()`，
/// 改 adapter 会越权（v1.4.90 多 agent 并行约定 agent C 改 adapter，不交叉）。
///
/// **codex 0522 F3 v1.4.106 (option B)**: 接 `Option<&CallerContext>` 让
/// gateway handler 知道 REST caller key 身份, 即使 conn_id 是
/// `REST_SHARED_CONN` 全局共享 (per-key 订阅配额 / cleanup / 审计). QOT
/// 行情 path 不直接受 `allowed_acc_ids` 约束, 但 `caller_key_id` 让未来
/// per-key 订阅 owner 模型 (e.g. QotSubscriptionState owner) 可识别 caller
/// 而不是看作 "REST 全局".
async fn proto_request_shared_conn<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    ctx: Option<&crate::caller_context::CallerContext>,
) -> ApiResult
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    // 1. JSON → protobuf 请求（与 adapter::proto_request_with_idempotency 同源）。
    // QOT shared-conn 不做 TRD header expansion, 但空 body 同样必须经过
    // EndpointSpec validation, 防止 required field 被 `Req::default()` 静默绕过。
    let req_msg: Req = decode_json_request(proto_id, json_body, JsonRequestMode::QotSharedConn)?;

    // 2. encode
    let body = Bytes::from(req_msg.encode_to_vec());

    // 3. dispatch with REST_SHARED_CONN（区别点）
    // codex 0522 F3 v1.4.106: 同时填 caller_key_id (per-call snapshot) 让
    // gateway handler 识别 REST caller 身份. allowed_acc_ids 仍 None
    // (QOT 行情不直接 acc-bound), 真填走 ctx.caller_allowed_acc_ids_arc.
    let incoming = IncomingRequest::builder(
        REST_SHARED_CONN,
        proto_id,
        state.next_serial(),
        ProtoFmtType::Protobuf,
        body,
    )
    .with_caller_scope(
        ctx.and_then(|c| c.caller_allowed_acc_ids_arc()),
        ctx.and_then(|c| c.caller_key_id()),
    )
    .build();
    let resp_bytes = state
        .router
        .dispatch(REST_SHARED_CONN, &incoming)
        .await
        .ok_or_else(|| {
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({
                    "error": "handler returned no response"
                })),
            )
        })?;

    // 4. decode
    let rsp_msg = Rsp::decode(Bytes::from(resp_bytes)).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": format!("failed to decode response: {e}")
            })),
        )
    })?;

    // 5. serialize JSON
    let mut json_rsp = serde_json::to_value(&rsp_msg).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": format!("failed to serialize response: {e}")
            })),
        )
    })?;

    // 6. err_code 前缀（与通用 adapter path 同源，避免 shared-conn REST
    //    订阅路径和普通 REST 路径的错误契约漂移）
    adapter::maybe_wrap_err_code_prefix(&mut json_rsp);

    Ok(Json(json_rsp))
}

async fn proto_request_shared_conn_raw<Req, Rsp>(
    state: &RestState,
    proto_id: u32,
    json_body: Option<Value>,
    ctx: Option<&crate::caller_context::CallerContext>,
) -> RawApiResult
where
    Req: Message + Default + serde::de::DeserializeOwned,
    Rsp: Message + Default + serde::Serialize,
{
    let req_msg: Req = decode_json_request(proto_id, json_body, JsonRequestMode::QotSharedConn)?;
    let body = Bytes::from(req_msg.encode_to_vec());

    let incoming = IncomingRequest::builder(
        REST_SHARED_CONN,
        proto_id,
        state.next_serial(),
        ProtoFmtType::Protobuf,
        body,
    )
    .with_caller_scope(
        ctx.and_then(|c| c.caller_allowed_acc_ids_arc()),
        ctx.and_then(|c| c.caller_key_id()),
    )
    .build();
    let resp_bytes = state
        .router
        .dispatch(REST_SHARED_CONN, &incoming)
        .await
        .ok_or_else(|| {
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({
                    "error": "handler returned no response"
                })),
            )
        })?;

    let rsp_msg = Rsp::decode(Bytes::from(resp_bytes)).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({
                "error": format!("failed to decode response: {e}")
            })),
        )
    })?;

    adapter::raw_json_from_proto_response(&rsp_msg)
}

#[cfg(test)]
fn map_surface_spec_error(
    spec: &'static futu_surface_spec::EndpointSpec,
    err: futu_surface_spec::DispatchError,
) -> (StatusCode, Json<Value>) {
    let proto_id = spec
        .proto_id()
        .map(|id| id.to_string())
        .unwrap_or_else(|| "daemon-local".to_string());
    let ret_msg = format!(
        "{} (endpoint: {}, proto_id: {})",
        err, spec.canonical_name, proto_id
    );
    let mut body = validation_error_body(ret_msg);
    let machine_error_field = spec.runtime.error.machine_error_field;
    if let Some(obj) = body.as_object_mut() {
        obj.insert(
            machine_error_field.to_string(),
            serde_json::json!({
                "kind": "validation_error",
                "message": err.to_string(),
                "endpoint": spec.canonical_name,
                "proto_id": proto_id,
            }),
        );
    }
    (StatusCode::BAD_REQUEST, Json(body))
}

#[cfg(test)]
fn validation_error_body(message: impl Into<String>) -> Value {
    let message = message.into();
    serde_json::json!({
        "ret_type": -1,
        "ret_msg": message,
        "error": message,
    })
}

/// POST /api/subscribe — 订阅/退订行情
///
/// **v1.4.90 P0-B fix**: 用 `REST_SHARED_CONN` 替代 `state.next_conn_id()`,
/// 杜绝 quota 永久泄漏（详见 `REST_SHARED_CONN` 注释）.
///
/// **v1.4.104 external reviewer S-005 (P1) fix**: REST 层显式检查 `is_sub_or_un_sub` 字段
/// 在 raw JSON body 是否 present. proto bool 字段 missing 时 prost 默认为
/// false → handler 走 unsub 路径; 但 unsub 路径对 invalid ticker silent
/// success (handler 注释 line 187-189: unsub by-design 不做 backend 解析).
/// agent 调用方默认只传 `symbols`, 不写 boolean → silent 没订阅.
///
/// 修法: REST adapter 入口先检查 `is_sub_or_un_sub` 在 body 里 (snake-case
/// 归一化后) 是否 present. 不 present → 400 提示用户必须显式传字段.
///
/// **v1.4.104 codex round 2 F5 (P2) fix**: REST 全部 caller 共用一个虚拟连接
/// `REST_SHARED_CONN=0xFFFFFFFE` (v1.4.90 P0-B 防 quota 泄漏的设计). 因此
/// `is_unsub_all=true` 调用会**清掉所有 REST callers** 的 QOT 订阅 (因为大家
/// 共享同一 conn_id 的 subscription bucket). 对于多 REST client 部署:
///
///   - 当前合约: `is_unsub_all=true` 在 REST 显式 reject, 仅返 400 解释
///     "REST is_unsub_all 是 process-wide 操作, 跨 caller 影响, 默认禁用".
///   - 可用替代: 显式列 `security_list` + `sub_type_list` 做单 symbol 退订;
///     或改用 MCP / gRPC / WS 这类 per-conn surface 执行 unsub_all.
///   - REST 目前没有 process-wide opt-in query, 也没有 admin clear endpoint;
///     后续若要支持 REST per-key unsub_all, 必须先补独立状态模型与公开契约。
///
/// 当前实装: 选项 A 保守 (拒+提示), 防止 caller A 意外清掉 caller B 的 subs.
/// MCP / gRPC / WS 各 caller 有自己 conn_id, 不受影响.
// Split: 1408 行 → 5 子文件 (contiguous fn 段)
mod misc;
mod quotes;
mod reference;
mod snapshot;
mod subscribe;

#[cfg(test)]
mod tests;

#[cfg(test)]
use misc::inject_default_is_req_all_conn;
#[cfg(test)]
use quotes::{annotate_quote_cache_miss, orderbook_loud_unsub_hint};
#[cfg(test)]
use snapshot::{
    augment_snapshot_with_exchange_code, augment_static_info_with_exchange_code,
    check_static_info_input,
};
#[cfg(test)]
use subscribe::body_has_sub_or_unsub_flag;

pub use misc::{
    get_risk_free_rate, get_spread_table, get_ticker_statistic, get_ticker_statistic_detail,
    list_plates, query_subscription, unsubscribe,
};
pub use quotes::{get_basic_qot, get_broker, get_kl, get_order_book, get_rt, get_ticker};
pub use reference::{
    get_capital_distribution, get_capital_flow, get_code_change, get_company_executive_background,
    get_company_executives, get_company_operational_efficiency, get_company_profile,
    get_corporate_actions_buybacks, get_corporate_actions_dividends,
    get_corporate_actions_stock_splits, get_daily_short_volume, get_derivative_unusual,
    get_financial_unusual, get_financials_earnings_price_history,
    get_financials_earnings_price_move, get_financials_revenue_breakdown,
    get_financials_statements, get_future_info, get_holding_change, get_insider_holder_list,
    get_insider_trade_list, get_ipo_list, get_market_state, get_option_chain,
    get_option_exercise_probability, get_option_expiration_date, get_option_quote,
    get_option_strategy, get_option_strategy_analysis, get_option_strategy_spread,
    get_option_volatility, get_owner_plate, get_plate_security, get_plate_set, get_price_reminder,
    get_reference, get_research_analyst_consensus, get_research_morningstar_report,
    get_research_rating_summary, get_shareholders_holder_detail, get_shareholders_holding_changes,
    get_shareholders_institutional, get_shareholders_overview, get_short_interest, get_suspend,
    get_technical_unusual, get_top_ten_buy_sell_brokers, get_used_quota, get_user_security,
    get_valuation_detail, get_valuation_plate_stock_list, get_warrant, modify_user_security,
    option_screen, request_history_kl, request_history_kl_quota, request_rehab,
    request_trading_days, set_price_reminder, stock_filter, stock_screen, warrant_screen,
};
pub use snapshot::{get_snapshot, get_static_info};
pub use subscribe::{get_sub_info, subscribe};