futucli/cmd/
indicator.rs

1use std::time::Duration;
2
3use anyhow::{Result, anyhow, bail};
4use prost::Message as _;
5
6use crate::common::connect_gateway;
7use crate::output::OutputFormat;
8
9pub async fn calculate(
10    gateway: &str,
11    c2s_json: &str,
12    timeout_secs: u64,
13    output: OutputFormat,
14) -> Result<()> {
15    if timeout_secs == 0 {
16        bail!("indicator-calc --timeout-secs must be positive");
17    }
18    let mut value: serde_json::Value = serde_json::from_str(c2s_json)
19        .map_err(|error| anyhow!("indicator-calc c2s json: {error}"))?;
20    let spec = futu_surface_spec::lookup_endpoint_by_cli_subcommand("indicator-calc")
21        .ok_or_else(|| anyhow!("indicator-calc surface spec missing"))?;
22    futu_surface_spec::validate_and_normalize(spec, &mut value)
23        .map_err(|error| anyhow!("indicator-calc c2s json: {error}"))?;
24    let c2s: futu_proto::qot_request_indicator_calc::C2s = serde_json::from_value(value)
25        .map_err(|error| anyhow!("indicator-calc c2s json: {error}"))?;
26
27    // One connection owns both the 3260 ACK and its paired 3261 push. Do not
28    // route this command through REST/MCP/generic gRPC or reconnect between
29    // the two frames.
30    let (client, mut push_rx) = connect_gateway(gateway, "futucli-indicator-calc").await?;
31    let ack_frame = client
32        .request(
33            futu_core::proto_id::QOT_REQUEST_INDICATOR_CALC,
34            futu_proto::qot_request_indicator_calc::Request { c2s }.encode_to_vec(),
35        )
36        .await?;
37    let ack = futu_proto::qot_request_indicator_calc::Response::decode(ack_frame.body.as_ref())
38        .map_err(|error| anyhow!("decode indicator-calc ACK: {error}"))?;
39    let calc_id = ack
40        .s2c
41        .as_ref()
42        .map(|s2c| s2c.calc_id.as_str())
43        .unwrap_or("");
44    if calc_id.is_empty() {
45        bail!("indicator-calc ACK omitted calcId");
46    }
47    if ack.ret_type != 0 {
48        bail!(
49            "indicator-calc calc_id={calc_id} ret_type={} msg={:?} err_code={:?}",
50            ack.ret_type,
51            ack.ret_msg,
52            ack.err_code
53        );
54    }
55
56    let push = tokio::time::timeout(Duration::from_secs(timeout_secs), async {
57        loop {
58            let message = push_rx
59                .recv()
60                .await
61                .ok_or_else(|| anyhow!("indicator-calc push channel closed"))?;
62            if message.proto_id != futu_core::proto_id::QOT_PUSH_INDICATOR_CALC {
63                continue;
64            }
65            let decoded =
66                futu_proto::qot_push_indicator_calc::Response::decode(message.body.as_ref())
67                    .map_err(|error| anyhow!("decode indicator-calc 3261: {error}"))?;
68            if decoded.s2c.as_ref().map(|s2c| s2c.calc_id.as_str()) == Some(calc_id) {
69                return Ok::<_, anyhow::Error>(decoded);
70            }
71        }
72    })
73    .await
74    .map_err(|_| anyhow!("indicator-calc calc_id={calc_id} timed out after {timeout_secs}s"))??;
75    if push.ret_type != 0 {
76        bail!(
77            "indicator-calc calc_id={calc_id} push ret_type={} msg={:?} err_code={:?}",
78            push.ret_type,
79            push.ret_msg,
80            push.err_code
81        );
82    }
83    match output {
84        OutputFormat::Jsonl => println!("{}", serde_json::to_string(&push)?),
85        OutputFormat::Table | OutputFormat::Json | OutputFormat::Markdown => {
86            println!("{}", serde_json::to_string_pretty(&push)?)
87        }
88    }
89    Ok(())
90}
futucli/cmd/indicator.rs

futucli/cmd/
indicator.rs
