futucli/cmd/

sub.rs

//! `futucli sub` — 订阅推送并持续打印，Ctrl-C 停止
//!
//! v1.4.98 external reviewer BUG-005 fix (P2, 2026-04-27): `-o jsonl` 之前忽略, 仍打印
//! `[time] basic HK.00700 price=...` 文本. 修后 `-o jsonl` 每条 push 一行
//! 合法 JSON object (脚本 / agent 用).

use anyhow::Result;
use async_trait::async_trait;
use futu_qot::push::{QuoteHandler, QuotePushDispatcher};
use futu_qot::sub::SubscribeOptions;
use futu_qot::types::{BasicQot, KLine, OrderBookData, Security};
use futu_surface_spec::input::parse_subscription_session_id;

use crate::common::{connect_gateway, format_symbol, parse_sub_types, parse_symbol};
use crate::output::OutputFormat;

/// PrintHandler — text format (default & --output table).
struct PrintHandler;

/// JsonlHandler — emit one JSON object per push event (--output jsonl/json).
struct JsonlHandler;

#[async_trait]
impl QuoteHandler for PrintHandler {
    async fn on_basic_qot_update(&self, qot_list: Vec<BasicQot>) {
        for q in qot_list {
            let sym = format_symbol(&q.security);
            let change = q.cur_price - q.last_close_price;
            let pct = if q.last_close_price != 0.0 {
                change / q.last_close_price * 100.0
            } else {
                0.0
            };
            let sign = if change >= 0.0 { "+" } else { "" };
            println!(
                "[{}] basic {sym:<12} price={:.3} change={sign}{change:.3} ({sign}{pct:.2}%) vol={}",
                q.update_time, q.cur_price, q.volume
            );
        }
    }

    async fn on_kl_update(&self, security: Security, kl_list: Vec<KLine>) {
        let sym = format_symbol(&security);
        for k in kl_list {
            println!(
                "[{}] kl    {sym:<12} O={:.3} H={:.3} L={:.3} C={:.3} V={}",
                k.time, k.open_price, k.high_price, k.low_price, k.close_price, k.volume
            );
        }
    }

    async fn on_order_book_update(&self, data: OrderBookData) {
        let sym = format_symbol(&data.security);
        let top_bid = data.bid_list.first();
        let top_ask = data.ask_list.first();
        match (top_bid, top_ask) {
            (Some(b), Some(a)) => println!(
                "              ob    {sym:<12} bid={:.3}x{}  ask={:.3}x{}  spread={:.3}",
                b.price,
                b.volume,
                a.price,
                a.volume,
                a.price - b.price
            ),
            _ => println!("              ob    {sym:<12} (empty)"),
        }
    }

    async fn on_ticker_update(
        &self,
        security: Security,
        ticker_list: Vec<futu_qot::ticker::Ticker>,
    ) {
        let sym = format_symbol(&security);
        for t in ticker_list {
            println!(
                "              tick  {sym:<12} px={:.3} vol={} dir={}",
                t.price, t.volume, t.dir
            );
        }
    }

    async fn on_rt_update(&self, security: Security, rt_list: Vec<futu_qot::rt::TimeShare>) {
        let sym = format_symbol(&security);
        for r in rt_list {
            println!(
                "[{}] rt    {sym:<12} px={:.3} avg={:.3} vol={}",
                r.time, r.price, r.avg_price, r.volume
            );
        }
    }
}

#[async_trait]
impl QuoteHandler for JsonlHandler {
    async fn on_basic_qot_update(&self, qot_list: Vec<BasicQot>) {
        for q in qot_list {
            let line = serde_json::json!({
                "kind": "basic",
                "symbol": format_symbol(&q.security),
                "update_time": q.update_time,
                "cur_price": q.cur_price,
                "last_close": q.last_close_price,
                "volume": q.volume,
                "turnover": q.turnover,
                "open_price": q.open_price,
                "high_price": q.high_price,
                "low_price": q.low_price,
            });
            println!("{}", line);
        }
    }
    async fn on_kl_update(&self, security: Security, kl_list: Vec<KLine>) {
        for k in kl_list {
            let line = serde_json::json!({
                "kind": "kline",
                "symbol": format_symbol(&security),
                "time": k.time,
                "open": k.open_price,
                "high": k.high_price,
                "low": k.low_price,
                "close": k.close_price,
                "volume": k.volume,
            });
            println!("{}", line);
        }
    }
    async fn on_order_book_update(&self, data: OrderBookData) {
        let line = serde_json::json!({
            "kind": "orderbook",
            "symbol": format_symbol(&data.security),
            "bid_top": data.bid_list.first().map(|b| serde_json::json!({"price": b.price, "volume": b.volume})),
            "ask_top": data.ask_list.first().map(|a| serde_json::json!({"price": a.price, "volume": a.volume})),
        });
        println!("{}", line);
    }
    async fn on_ticker_update(
        &self,
        security: Security,
        ticker_list: Vec<futu_qot::ticker::Ticker>,
    ) {
        for t in ticker_list {
            let line = serde_json::json!({
                "kind": "ticker",
                "symbol": format_symbol(&security),
                "price": t.price,
                "volume": t.volume,
                "dir": t.dir,
            });
            println!("{}", line);
        }
    }
    async fn on_rt_update(&self, security: Security, rt_list: Vec<futu_qot::rt::TimeShare>) {
        for r in rt_list {
            let line = serde_json::json!({
                "kind": "rt",
                "symbol": format_symbol(&security),
                "time": r.time,
                "price": r.price,
                "avg_price": r.avg_price,
                "volume": r.volume,
            });
            println!("{}", line);
        }
    }
}

pub async fn run(
    gateway: &str,
    symbols: &[String],
    types_csv: &str,
    extended_time: bool,
    session: Option<&str>,
    orderbook_detail: bool,
    format: OutputFormat,
) -> Result<()> {
    let secs: Vec<_> = symbols
        .iter()
        .map(|s| parse_symbol(s))
        .collect::<Result<_>>()?;
    let sub_types = parse_sub_types(types_csv)?;
    let session = parse_qot_sub_session(session)?;

    let (client, mut push_rx) = connect_gateway(gateway, "futucli-sub").await?;

    // 订阅
    futu_qot::sub::subscribe_with_options(
        &client,
        &secs,
        &sub_types,
        SubscribeOptions {
            is_reg_push: true,
            is_first_push: true,
            extended_time: extended_time.then_some(true),
            session,
            orderbook_detail: orderbook_detail.then_some(true),
        },
    )
    .await?;
    eprintln!(
        "✓ subscribed: symbols={:?}, types={types_csv}, extended_time={extended_time}, session={session:?}, orderbook_detail={orderbook_detail}  (Ctrl-C to stop)",
        symbols,
    );

    // v1.4.98 external reviewer BUG-005: jsonl/json → JsonlHandler 输出每条 push 一行
    // JSON object; 否则 PrintHandler 文本.
    let use_jsonl = matches!(format, OutputFormat::Json | OutputFormat::Jsonl);

    loop {
        tokio::select! {
            maybe = push_rx.recv() => {
                let Some(msg) = maybe else {
                    eprintln!("⚠️  push channel closed");
                    break;
                };
                let dispatch_err = if use_jsonl {
                    QuotePushDispatcher::dispatch(&JsonlHandler, msg.proto_id, &msg.body).await
                } else {
                    QuotePushDispatcher::dispatch(&PrintHandler, msg.proto_id, &msg.body).await
                };
                if let Err(e) = dispatch_err {
                    eprintln!("push dispatch error: {e}");
                }
            }
            _ = tokio::signal::ctrl_c() => {
                eprintln!("\n⏹  stopping (ctrl-c)");
                // 退出路径保持 best-effort, 但失败必须可见；否则用户会误以为
                // server-side subscription 已清理。
                if let Err(err) = futu_qot::sub::unsubscribe(&client, &secs, &sub_types).await {
                    eprintln!("⚠️  unsubscribe during shutdown failed: {err}");
                }
                break;
            }
        }
    }
    Ok(())
}

fn parse_qot_sub_session(value: Option<&str>) -> Result<Option<i32>> {
    parse_subscription_session_id(value)
        .map_err(|error| anyhow::anyhow!("invalid --session: {error}"))
}

#[cfg(test)]
mod tests;