futucli/cmd/

proto_json.rs

//! Proto-JSON passthrough helpers for newly added official endpoints.
//!
//! These commands intentionally accept generated `C2S` JSON instead of a large
//! hand-written flag surface. That keeps v10.7 combo-option CLI coverage tied to
//! the authoritative proto shape while the typed ergonomic CLI can evolve later.

use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicU32, Ordering};

use anyhow::{Result, anyhow, bail};
use futu_core::diagnostic_text::proto_json_numeric_qot_market_hint;
use prost::Message;
use serde::Serialize;
use serde::de::DeserializeOwned;

use crate::common::connect_gateway;
use crate::output::OutputFormat;

static PACKET_SERIAL: AtomicU32 = AtomicU32::new(1);

fn parse_c2s<T>(label: &str, json: &str) -> Result<T>
where
    T: DeserializeOwned,
{
    let mut value: serde_json::Value =
        serde_json::from_str(json).map_err(|err| anyhow!("{label} c2s json: {err}"))?;
    validate_proto_json_contract(label, &mut value)?;
    deserialize_c2s_value(label, value)
}

fn validate_proto_json_contract(label: &str, value: &mut serde_json::Value) -> Result<()> {
    let Some(spec) = futu_surface_spec::lookup_endpoint_by_cli_subcommand(label) else {
        return Ok(());
    };
    futu_surface_spec::validate_and_normalize(spec, value)
        .map_err(|err| anyhow!("{label} c2s json: {err}"))
}

fn parse_combo_max_c2s_json(json: &str) -> Result<futu_proto::trd_get_combo_max_trd_qtys::C2s> {
    let c2s: futu_proto::trd_get_combo_max_trd_qtys::C2s =
        parse_combo_c2s_json("combo-max-trd-qtys", json)?;
    ensure_write_trd_market("combo-max-trd-qtys", c2s.header.trd_market)?;
    Ok(c2s)
}

fn parse_place_combo_c2s_json(json: &str) -> Result<futu_proto::trd_place_combo_order::C2s> {
    let c2s: futu_proto::trd_place_combo_order::C2s = parse_combo_c2s_json("combo-order", json)?;
    ensure_write_trd_market("combo-order", c2s.header.trd_market)?;
    Ok(c2s)
}

fn parse_combo_c2s_json<T>(label: &str, json: &str) -> Result<T>
where
    T: DeserializeOwned,
{
    parse_c2s_with_required_paths(
        label,
        json,
        &[
            (["header", "trd_env"].as_slice(), "header.trd_env"),
            (["header", "acc_id"].as_slice(), "header.acc_id"),
            (["header", "trd_market"].as_slice(), "header.trd_market"),
            (["combo_legs"].as_slice(), "combo_legs"),
            (["qty"].as_slice(), "qty"),
            (["order_type"].as_slice(), "order_type"),
        ],
        Some(validate_combo_trade_contract),
    )
}

fn parse_c2s_with_required_paths<T>(
    label: &str,
    json: &str,
    required_paths: &[(&[&str], &'static str)],
    extra_validator: Option<fn(&str, &serde_json::Value) -> Result<()>>,
) -> Result<T>
where
    T: DeserializeOwned,
{
    let value: serde_json::Value =
        serde_json::from_str(json).map_err(|err| anyhow!("{label} c2s json: {err}"))?;
    for (path, name) in required_paths {
        if json_path(&value, path).is_none() {
            bail!("{label} c2s json missing required field {name}");
        }
    }
    if let Some(validate) = extra_validator {
        validate(label, &value)?;
    }
    deserialize_c2s_value(label, value)
}

fn deserialize_c2s_value<T>(label: &str, value: serde_json::Value) -> Result<T>
where
    T: DeserializeOwned,
{
    serde_json::from_value(value.clone()).map_err(|err| {
        anyhow!(
            "{label} c2s json: {err}{}",
            proto_json_deserialize_hint(&value)
        )
    })
}

fn proto_json_deserialize_hint(value: &serde_json::Value) -> String {
    if has_string_market_field(value) {
        format!(
            "; hint: {}",
            proto_json_numeric_qot_market_hint().text_with_code()
        )
    } else {
        String::new()
    }
}

fn has_string_market_field(value: &serde_json::Value) -> bool {
    match value {
        serde_json::Value::Object(map) => map.iter().any(|(key, nested)| {
            (key == "market" && nested.is_string()) || has_string_market_field(nested)
        }),
        serde_json::Value::Array(items) => items.iter().any(has_string_market_field),
        _ => false,
    }
}

fn json_path<'a>(value: &'a serde_json::Value, path: &[&str]) -> Option<&'a serde_json::Value> {
    let mut current = value;
    for segment in path {
        current = current.get(*segment)?;
    }
    Some(current)
}

fn validate_combo_trade_contract(label: &str, value: &serde_json::Value) -> Result<()> {
    match json_path(value, &["header", "acc_id"]).and_then(serde_json::Value::as_u64) {
        Some(acc_id) if acc_id > 0 => {}
        _ => bail!("{label} c2s json header.acc_id must be a positive integer"),
    }

    let legs = json_path(value, &["combo_legs"])
        .and_then(serde_json::Value::as_array)
        .ok_or_else(|| anyhow!("{label} c2s json combo_legs must be an array"))?;
    if legs.len() < 2 {
        bail!("{label} c2s json combo_legs must contain at least two legs");
    }

    match json_path(value, &["qty"]).and_then(serde_json::Value::as_f64) {
        Some(qty) if qty > 0.0 => {}
        _ => bail!("{label} c2s json qty must be positive"),
    }

    let order_type = json_path(value, &["order_type"])
        .and_then(serde_json::Value::as_i64)
        .ok_or_else(|| anyhow!("{label} c2s json order_type must be an integer"))?;
    if i32::try_from(order_type).is_err() {
        bail!("{label} c2s json order_type={order_type} is out of range");
    }

    Ok(())
}

fn print_proto_response<T>(format: OutputFormat, response: &T) -> Result<()>
where
    T: Serialize,
{
    match format {
        OutputFormat::Table | OutputFormat::Json | OutputFormat::Markdown => {
            println!("{}", serde_json::to_string_pretty(response)?);
        }
        OutputFormat::Jsonl => {
            println!("{}", serde_json::to_string(response)?);
        }
    }
    Ok(())
}

fn ensure_success(
    label: &str,
    ret_type: i32,
    ret_msg: Option<&str>,
    err_code: Option<i32>,
) -> Result<()> {
    if ret_type == 0 {
        return Ok(());
    }
    bail!("{label} ret_type={ret_type} msg={ret_msg:?} err_code={err_code:?}")
}

async fn send_proto<Req, Resp>(
    gateway: &str,
    client_id: &str,
    proto_id: u32,
    request: Req,
) -> Result<Resp>
where
    Req: Message,
    Resp: Message + Default,
{
    let (client, _rx) = connect_gateway(gateway, client_id).await?;
    let frame = client.request(proto_id, request.encode_to_vec()).await?;
    Resp::decode(frame.body.as_ref()).map_err(|err| anyhow!("decode response: {err}"))
}

macro_rules! qot_proto_json_command {
    ($fn_name:ident, $label:literal, $proto_id:expr, $module:ident) => {
        pub async fn $fn_name(gateway: &str, c2s_json: &str, output: OutputFormat) -> Result<()> {
            let c2s = parse_c2s::<futu_proto::$module::C2s>($label, c2s_json)?;
            let request = futu_proto::$module::Request { c2s };
            let response: futu_proto::$module::Response =
                send_proto(gateway, concat!("futucli-", $label), $proto_id, request).await?;
            ensure_success(
                $label,
                response.ret_type,
                response.ret_msg.as_deref(),
                response.err_code,
            )?;
            print_proto_response(output, &response)
        }
    };
}

qot_proto_json_command!(
    run_option_quote,
    "option-quote",
    futu_core::proto_id::QOT_GET_OPTION_QUOTE,
    qot_get_option_quote
);
qot_proto_json_command!(
    run_option_strategy,
    "option-strategy",
    futu_core::proto_id::QOT_GET_OPTION_STRATEGY,
    qot_get_option_strategy
);
qot_proto_json_command!(
    run_option_strategy_analysis,
    "option-strategy-analysis",
    futu_core::proto_id::QOT_GET_OPTION_STRATEGY_ANALYSIS,
    qot_get_option_strategy_analysis
);
qot_proto_json_command!(
    run_option_strategy_spread,
    "option-strategy-spread",
    futu_core::proto_id::QOT_GET_OPTION_STRATEGY_SPREAD,
    qot_get_option_strategy_spread
);
qot_proto_json_command!(
    run_earnings_calendar,
    "earnings-calendar",
    futu_core::proto_id::QOT_GET_EARNINGS_CALENDAR,
    qot_get_earnings_calendar
);
qot_proto_json_command!(
    run_macro_indicator_list,
    "macro-indicator-list",
    futu_core::proto_id::QOT_GET_MACRO_INDICATOR_LIST,
    qot_get_macro_indicator_list
);
qot_proto_json_command!(
    run_macro_indicator_history,
    "macro-indicator-history",
    futu_core::proto_id::QOT_GET_MACRO_INDICATOR_HISTORY,
    qot_get_macro_indicator_history
);
qot_proto_json_command!(
    run_fed_watch_target_rate,
    "fed-watch-target-rate",
    futu_core::proto_id::QOT_GET_FED_WATCH_TARGET_RATE,
    qot_get_fed_watch_target_rate
);
qot_proto_json_command!(
    run_fed_watch_dot_plot,
    "fed-watch-dot-plot",
    futu_core::proto_id::QOT_GET_FED_WATCH_DOT_PLOT,
    qot_get_fed_watch_dot_plot
);
qot_proto_json_command!(
    run_earnings_beat_rank,
    "earnings-beat-rank",
    futu_core::proto_id::QOT_GET_EARNINGS_BEAT_RANK,
    qot_get_earnings_beat_rank
);
qot_proto_json_command!(
    run_dividend_rank,
    "dividend-rank",
    futu_core::proto_id::QOT_GET_DIVIDEND_RANK,
    qot_get_dividend_rank
);
qot_proto_json_command!(
    run_dividend_calendar,
    "dividend-calendar",
    futu_core::proto_id::QOT_GET_DIVIDEND_CALENDAR,
    qot_get_dividend_calendar
);
qot_proto_json_command!(
    run_economic_calendar,
    "economic-calendar",
    futu_core::proto_id::QOT_GET_ECONOMIC_CALENDAR,
    qot_get_economic_calendar
);
qot_proto_json_command!(
    run_us_pre_market_rank,
    "us-pre-market-rank",
    futu_core::proto_id::QOT_GET_US_PRE_MARKET_RANK,
    qot_get_us_pre_market_rank
);
qot_proto_json_command!(
    run_us_after_hours_rank,
    "us-after-hours-rank",
    futu_core::proto_id::QOT_GET_US_AFTER_HOURS_RANK,
    qot_get_us_after_hours_rank
);
qot_proto_json_command!(
    run_us_overnight_rank,
    "us-overnight-rank",
    futu_core::proto_id::QOT_GET_US_OVERNIGHT_RANK,
    qot_get_us_overnight_rank
);
qot_proto_json_command!(
    run_top_movers_rank,
    "top-movers-rank",
    futu_core::proto_id::QOT_GET_TOP_MOVERS_RANK,
    qot_get_top_movers_rank
);
qot_proto_json_command!(
    run_hot_list,
    "hot-list",
    futu_core::proto_id::QOT_GET_HOT_LIST,
    qot_get_hot_list
);
qot_proto_json_command!(
    run_short_selling_rank,
    "short-selling-rank",
    futu_core::proto_id::QOT_GET_SHORT_SELLING_RANK,
    qot_get_short_selling_rank
);
qot_proto_json_command!(
    run_period_change_rank,
    "period-change-rank",
    futu_core::proto_id::QOT_GET_PERIOD_CHANGE_RANK,
    qot_get_period_change_rank
);
qot_proto_json_command!(
    run_high_dividend_soe_rank,
    "high-dividend-soe-rank",
    futu_core::proto_id::QOT_GET_HIGH_DIVIDEND_SOE_RANK,
    qot_get_high_dividend_soe_rank
);
qot_proto_json_command!(
    run_institution_list,
    "institution-list",
    futu_core::proto_id::QOT_GET_INSTITUTION_LIST,
    qot_get_institution_list
);
qot_proto_json_command!(
    run_institution_profile,
    "institution-profile",
    futu_core::proto_id::QOT_GET_INSTITUTION_PROFILE,
    qot_get_institution_profile
);
qot_proto_json_command!(
    run_institution_distribution,
    "institution-distribution",
    futu_core::proto_id::QOT_GET_INSTITUTION_DISTRIBUTION,
    qot_get_institution_distribution
);
qot_proto_json_command!(
    run_institution_holding_change,
    "institution-holding-change",
    futu_core::proto_id::QOT_GET_INSTITUTION_HOLDING_CHANGE,
    qot_get_institution_holding_change
);
qot_proto_json_command!(
    run_institution_holding_list,
    "institution-holding-list",
    futu_core::proto_id::QOT_GET_INSTITUTION_HOLDING_LIST,
    qot_get_institution_holding_list
);
qot_proto_json_command!(
    run_ark_fund_holding,
    "ark-fund-holding",
    futu_core::proto_id::QOT_GET_ARK_FUND_HOLDING,
    qot_get_ark_fund_holding
);
qot_proto_json_command!(
    run_ark_stock_dynamic,
    "ark-stock-dynamic",
    futu_core::proto_id::QOT_GET_ARK_STOCK_DYNAMIC,
    qot_get_ark_stock_dynamic
);
qot_proto_json_command!(
    run_ark_active_transaction,
    "ark-active-transaction",
    futu_core::proto_id::QOT_GET_ARK_ACTIVE_TRANSACTION,
    qot_get_ark_active_transaction
);
qot_proto_json_command!(
    run_rating_change,
    "rating-change",
    futu_core::proto_id::QOT_GET_RATING_CHANGE,
    qot_get_rating_change
);
qot_proto_json_command!(
    run_industrial_chain_list,
    "industrial-chain-list",
    futu_core::proto_id::QOT_GET_INDUSTRIAL_CHAIN_LIST,
    qot_get_industrial_chain_list
);
qot_proto_json_command!(
    run_industrial_chain_detail,
    "industrial-chain-detail",
    futu_core::proto_id::QOT_GET_INDUSTRIAL_CHAIN_DETAIL,
    qot_get_industrial_chain_detail
);
qot_proto_json_command!(
    run_industrial_chain_by_plate,
    "industrial-chain-by-plate",
    futu_core::proto_id::QOT_GET_INDUSTRIAL_CHAIN_BY_PLATE,
    qot_get_industrial_chain_by_plate
);
qot_proto_json_command!(
    run_industrial_plate_info,
    "industrial-plate-info",
    futu_core::proto_id::QOT_GET_INDUSTRIAL_PLATE_INFO,
    qot_get_industrial_plate_info
);
qot_proto_json_command!(
    run_industrial_plate_stock,
    "industrial-plate-stock",
    futu_core::proto_id::QOT_GET_INDUSTRIAL_PLATE_STOCK,
    qot_get_industrial_plate_stock
);
qot_proto_json_command!(
    run_heat_map_data,
    "heat-map-data",
    futu_core::proto_id::QOT_GET_HEAT_MAP_DATA,
    qot_get_heat_map_data
);
qot_proto_json_command!(
    run_rise_fall_distribution,
    "rise-fall-distribution",
    futu_core::proto_id::QOT_GET_RISE_FALL_DISTRIBUTION,
    qot_get_rise_fall_distribution
);

pub async fn run_combo_max_trd_qtys(
    gateway: &str,
    c2s_json: &str,
    output: OutputFormat,
) -> Result<()> {
    let c2s = parse_combo_max_c2s_json(c2s_json)?;
    let request = futu_proto::trd_get_combo_max_trd_qtys::Request { c2s };
    let response: futu_proto::trd_get_combo_max_trd_qtys::Response = send_proto(
        gateway,
        "futucli-combo-max-trd-qtys",
        futu_core::proto_id::TRD_GET_COMBO_MAX_TRD_QTYS,
        request,
    )
    .await?;
    ensure_success(
        "combo-max-trd-qtys",
        response.ret_type,
        response.ret_msg.as_deref(),
        response.err_code,
    )?;
    print_proto_response(output, &response)
}

pub async fn run_place_combo_order(
    gateway: &str,
    c2s_json: &str,
    confirm: bool,
    idempotency_key: Option<String>,
    output: OutputFormat,
) -> Result<()> {
    let mut c2s = parse_place_combo_c2s_json(c2s_json)?;

    ensure_combo_order_confirmed(c2s.header.trd_env, confirm)?;

    let (client, _rx) = connect_gateway(gateway, "futucli-combo-order").await?;
    c2s.packet_id = match idempotency_key.as_deref() {
        Some(key) => packet_id_for_idempotency_key(key),
        None => {
            let conn_id = client
                .conn_id()
                .ok_or_else(|| anyhow!("combo-order missing InitConnect conn_id"))?;
            next_packet_id(conn_id)
        }
    };

    let request = futu_proto::trd_place_combo_order::Request { c2s };
    let frame = client
        .request(
            futu_core::proto_id::TRD_PLACE_COMBO_ORDER,
            request.encode_to_vec(),
        )
        .await?;
    let response = futu_proto::trd_place_combo_order::Response::decode(frame.body.as_ref())
        .map_err(|err| anyhow!("decode combo-order response: {err}"))?;
    ensure_success(
        "combo-order",
        response.ret_type,
        response.ret_msg.as_deref(),
        response.err_code,
    )?;
    print_proto_response(output, &response)
}

fn next_packet_id(conn_id: u64) -> futu_proto::common::PacketId {
    let serial_no = PACKET_SERIAL.fetch_add(1, Ordering::Relaxed);
    futu_proto::common::PacketId { conn_id, serial_no }
}

fn ensure_combo_order_confirmed(trd_env: i32, confirm: bool) -> Result<()> {
    if trd_env == 1 && !confirm {
        bail!("combo-order real env requires --confirm");
    }
    Ok(())
}

fn ensure_write_trd_market(label: &str, trd_market: i32) -> Result<()> {
    if let Some(market) = futu_trd::market::trd_market_from_i32(trd_market)
        && let Some(fund_label) = futu_trd::market::canonical_fund_trd_market_label(market)
    {
        bail!(
            "{label} header.trd_market={trd_market} ({fund_label}) is view-only; \
             use a write-capable main market for combo trade paths"
        );
    }
    if futu_trd::market::trd_market_label(trd_market).is_none() {
        bail!("{label} unsupported header.trd_market={trd_market}");
    }
    Ok(())
}

// Same design as `futu-trd::order`: deterministic key-derived PacketID lets
// the daemon replay guard identify an explicit retry without turning every
// identical-looking combo order into an accidental duplicate.
fn packet_id_for_idempotency_key(key: &str) -> futu_proto::common::PacketId {
    let mut hasher = DefaultHasher::new();
    key.hash(&mut hasher);
    futu_proto::common::PacketId {
        conn_id: hasher.finish(),
        serial_no: 0,
    }
}

#[cfg(test)]
mod tests;