ek-data 0.2.8

//! flows_stream —— F1 订阅链 + F2 行情/K线逐合约流。
//!
//! 覆盖:vt 符号解析 → SubscribeRequest → 字典键组合(组合键互不串台)
//! → TickData 经真总线往返 → reach_limit 风控读取 → BarData 流。
//! 键形与原作约定一致:网关发「前缀裸键 + 前缀+派生键」双投(f2_sect 本文件两端演示)。

mod common;

use common::{Capture, fixture, wait_until};
use ek_data::{
    EVENT_BAR, EVENT_TICK, Exchange, Interval, SubscribeRequest, TickData, parse_vt_symbol,
    round_to,
};
use ek_event::Event;
use std::sync::Arc;
use std::time::Duration;

/// F1 订阅链:vt 符号 → SubscribeRequest → en_symbol 回环(订阅键的实况形态)。
#[test]
fn f1_subscribe_chain() {
    let (sym, ex) = parse_vt_symbol("CFFEX.IF2506").unwrap();
    assert_eq!(sym, "IF2506");
    assert_eq!(ex, Exchange::CFFEX);
    let (sym, ex) = parse_vt_symbol("SZSE.000300.IDX").unwrap();
    assert_eq!(sym, "000300.IDX");
    assert_eq!(ex, Exchange::SZSE);

    let req = SubscribeRequest::from_vt_symbol("CFFEX.IF2506").unwrap();
    assert_eq!(req.symbol, "IF2506");
    assert_eq!(req.exchange, Exchange::CFFEX);
    assert_eq!(req.en_symbol(), "CFFEX.IF2506");

    // 非法 vt 符号在订阅入口被拒(不进任何流)
    assert!(SubscribeRequest::from_vt_symbol("IF2506").is_err());
    assert!(SubscribeRequest::from_vt_symbol("FAKE.X").is_err());
}

/// F2-1 行情逐合约流:组合键互不串台;裸键收全市场。
#[test]
fn f2_1_tick_composed_keys_isolated() {
    let eng = ek_event::EventEngine::new();
    let if_cap = Capture::<TickData>::new();
    let sse_cap = Capture::<TickData>::new();
    // routes()={"eTick."}(裸键,收全市场) + 组合键定向;与原作网关双投的两条腿对应
    let _bare = if_cap.register_on(&eng, EVENT_TICK);
    let _if_key = if_cap.register_on(&eng, format!("{EVENT_TICK}CFFEX.IF2506"));
    let _sse_key = sse_cap.register_on(&eng, format!("{EVENT_TICK}SSE.600000"));
    eng.start();

    let t_if = TickData {
        gateway_name: "xt".into(),
        symbol: "IF2506".into(),
        exchange: Exchange::CFFEX,
        datetime: fixture::t0(),
        last_price: 3520.2,
        ..Default::default()
    };
    // 网关双投第一腿:组合键;第二腿:裸键
    eng.put(Event::with_data(
        format!("{EVENT_TICK}CFFEX.IF2506"),
        Arc::new(t_if.clone()),
    ))
    .unwrap();
    eng.put(Event::with_data(EVENT_TICK, Arc::new(t_if.clone())))
        .unwrap();
    eng.put(Event::with_data(EVENT_TICK, Arc::new(t_if)))
        .unwrap();

    assert!(
        wait_until(Duration::from_secs(2), || if_cap.len() == 3),
        "组合键×1 + 裸键×2 = 3 次捕获"
    );
    eng.stop();
    let seen = if_cap.all();
    assert_eq!(seen[0].en_symbol(), "CFFEX.IF2506");
    assert_eq!(seen[0].last_price, 3520.2);
    assert_eq!(seen[0].datetime, fixture::t0());
    assert_eq!(sse_cap.len(), 0, "eTick.SSE.600000 不收 CFFEX 的 tick");
}

/// F2-2 风控读取:触沿判定(object.py:105/111 一致,bid/ask 观察)与价签对齐。
#[test]
fn f2_2_tick_limit_gate_and_price_tick() {
    let eng = ek_event::EventEngine::new();
    let cap = Capture::<TickData>::new();
    cap.register_on(&eng, format!("{EVENT_TICK}SSE.600000"));
    eng.start();

    let t = TickData {
        gateway_name: "sina".into(),
        symbol: "600000".into(),
        exchange: Exchange::SSE,
        datetime: fixture::t0(),
        last_price: 8.13,
        bid_price_1: 8.5,
        ask_price_1: 8.5,
        limit_up: 9.5,
        limit_down: 7.5,
        ..Default::default()
    };
    // bid+diff >= limit_up / ask-diff <= limit_down(浮点精确安全值 0.5 步进)
    assert!(t.reach_limit_up(1.0), "8.5+1.0 == 9.5 触上沿");
    assert!(!t.reach_limit_up(0.9), "9.4 < 9.5 未触");
    assert!(t.reach_limit_down(1.0), "8.5-1.0 == 7.5 触下沿");
    assert!(!t.reach_limit_down(0.9), "7.6 > 7.5 未触");
    assert!(t.reach_limit(1.0) && !t.reach_limit(0.9));
    // 原作怪癖(byte 级保留):缺省 tick(bid/ask/上下沿全 0)恒判触及
    assert!(TickData::default().reach_limit(0.0));

    eng.put(Event::with_data(
        format!("{EVENT_TICK}SSE.600000"),
        Arc::new(t),
    ))
    .unwrap();

    assert!(wait_until(Duration::from_secs(2), || cap.len() == 1));
    eng.stop();
    // 价签对齐:下游委托仍按「合约 price_tick × round_to」出价(闸门上下文)
    assert_eq!(round_to(8.139, 0.01), 8.14);
    assert_eq!(round_to(8.1349, 0.005), 8.135);
}

/// F2-3 K线逐合约流:BarData 往返 + Interval 载荷在流上不退化。
#[test]
fn f2_3_bar_flow() {
    let eng = ek_event::EventEngine::new();
    let cap = Capture::<(String, f64, f64, f64, f64, f64)>::new();
    cap.register_on(&eng, format!("{EVENT_BAR}SSE.600000"));
    eng.start();

    let b0 = ek_data::BarData {
        gateway_name: "sina".into(),
        symbol: "600000".into(),
        exchange: Exchange::SSE,
        datetime: fixture::t0() - 60_000_000_000,
        interval: Some(Interval::MINUTE),
        open_price: 8.10,
        high_price: 8.22,
        low_price: 8.08,
        close_price: 8.13,
        volume: 120_650.0,
        ..Default::default()
    };
    let b1 = ek_data::BarData {
        close_price: 8.15,
        datetime: fixture::t0(),
        ..b0.clone()
    };
    type BarTuple = (String, f64, f64, f64, f64, f64);
    eng.put(Event::with_data(
        format!("{EVENT_BAR}SSE.600000"),
        Arc::new((
            "600000".to_string(),
            b0.open_price,
            b0.high_price,
            b0.low_price,
            b0.close_price,
            b0.volume,
        ) as BarTuple),
    ))
    .unwrap();
    eng.put(Event::with_data(
        format!("{EVENT_BAR}SSE.600000"),
        Arc::new((
            "600000".to_string(),
            b1.open_price,
            b1.high_price,
            b1.low_price,
            b1.close_price,
            121_100.0_f64,
        ) as BarTuple),
    ))
    .unwrap();

    assert!(wait_until(Duration::from_secs(2), || cap.len() == 2));
    eng.stop();
    let all = cap.all();
    assert_eq!(all[0].1, 8.10, "b0 open");
    assert_eq!(
        all[1].4, 8.15,
        "b1 close(序位:0 标的/1 开/2 高/3 低/4 收/5 量)"
    );
    assert_eq!(b0.interval, Some(Interval::MINUTE));
    assert!(b0.datetime < b1.datetime, "时间轴单调递增");
}