ek-data 0.2.8

//! flows_position_account —— F6 持仓流 + F7 账户流。
//!
//! 键形(原作 gateway.py:130-144):持仓 = 裸键 + `ePosition.<en_symbol>` 双投;
//! 账户 = 裸键 + `eAccount.<en_accountid>` 双投(逐账户)。

mod common;

use common::{Capture, wait_until};
use ek_data::{AccountData, Direction, EVENT_ACCOUNT, EVENT_POSITION, Exchange, PositionData};
use ek_event::Event;
use std::sync::Arc;
use std::time::Duration;

fn fx_pos(symbol: &str, volume: f64, pnl: f64) -> PositionData {
    PositionData {
        gateway_name: "xt".into(),
        symbol: symbol.into(),
        exchange: Exchange::CFFEX,
        direction: Direction::LONG,
        volume,
        frozen: 1.0,
        price: 3520.2,
        pnl,
        yd_volume: 1.0,
        ..Default::default()
    }
}

/// F6-1 持仓流:双标的组合键互不串台;en_positionid 键形回环。
#[test]
fn f6_position_flow_isolated() {
    let eng = ek_event::EventEngine::new();
    let long_cap = Capture::<PositionData>::new();
    let short_cap = Capture::<PositionData>::new();
    long_cap.register_on(&eng, format!("{EVENT_POSITION}CFFEX.IF2506"));
    short_cap.register_on(&eng, format!("{EVENT_POSITION}CFFEX.IF2509"));
    eng.start();

    let mut long = fx_pos("IF2506", 2.0, 420.0);
    long.direction = Direction::LONG;
    let mut short = fx_pos("IF2509", 1.0, -90.0);
    short.direction = Direction::SHORT;
    // 派生 ID:gateway.en_symbol.线值(线值决定 3:英文)
    assert_eq!(long.en_positionid(), "xt.CFFEX.IF2506.LONG");
    assert_eq!(short.en_positionid(), "xt.CFFEX.IF2509.SHORT");

    eng.put(Event::with_data(
        format!("{EVENT_POSITION}CFFEX.IF2506"),
        Arc::new(long.clone()),
    ))
    .unwrap();
    eng.put(Event::with_data(
        format!("{EVENT_POSITION}CFFEX.IF2509"),
        Arc::new(short.clone()),
    ))
    .unwrap();

    assert!(wait_until(Duration::from_secs(2), || long_cap.len() == 1
        && short_cap.len() == 1));
    eng.stop();
    let l = long_cap.last().unwrap();
    assert_eq!(l.en_symbol(), "CFFEX.IF2506");
    assert_eq!(l.volume, 2.0);
    assert_eq!(l.frozen, 1.0);
    assert_eq!(l.pnl, 420.0);
    assert_eq!(l.volume - l.frozen, 1.0, "可用 = volume - frozen(读侧口径)");
    assert_eq!(short_cap.last().unwrap().pnl, -90.0);
}

/// F6-2 持仓增量时间轴:同一键上两帧递推,捕获端按序看到 2 手 → 4 手。
#[test]
fn f6_position_update_timeline() {
    let eng = ek_event::EventEngine::new();
    let cap = Capture::<PositionData>::new();
    cap.register_on(&eng, format!("{EVENT_POSITION}CFFEX.IF2506"));
    eng.start();

    let p0 = fx_pos("IF2506", 2.0, 420.0);
    let p1 = fx_pos("IF2506", 4.0, 1260.0);
    eng.put(Event::with_data(
        format!("{EVENT_POSITION}CFFEX.IF2506"),
        Arc::new(p0),
    ))
    .unwrap();
    eng.put(Event::with_data(
        format!("{EVENT_POSITION}CFFEX.IF2506"),
        Arc::new(p1),
    ))
    .unwrap();

    assert!(wait_until(Duration::from_secs(2), || cap.len() == 2));
    eng.stop();
    let timeline = cap.all();
    assert_eq!(timeline[0].volume, 2.0);
    assert_eq!(timeline[1].volume, 4.0);
    assert!(timeline[1].pnl > timeline[0].pnl, "盈亏沿时间轴递增");
}

/// F7 账户流:双账户键互不串台;available = balance - frozen 读侧口径。
#[test]
fn f7_account_flow_isolated() {
    let eng = ek_event::EventEngine::new();
    let demo = Capture::<AccountData>::new();
    let hedge = Capture::<AccountData>::new();
    demo.register_on(&eng, format!("{EVENT_ACCOUNT}xt.demo"));
    hedge.register_on(&eng, format!("{EVENT_ACCOUNT}xt.hedge"));
    eng.start();

    let a0 = AccountData {
        gateway_name: "xt".into(),
        accountid: "demo".into(),
        balance: 100_000.0,
        frozen: 300.0,
    };
    let a1 = AccountData {
        gateway_name: "xt".into(),
        accountid: "hedge".into(),
        balance: 50_000.0,
        frozen: 1_200.5,
    };
    assert_eq!(a0.en_accountid(), "xt.demo");
    assert_eq!(a1.en_accountid(), "xt.hedge");
    assert_eq!(a0.available(), 99_700.0);
    assert_eq!(a1.available(), 48_799.5);

    eng.put(Event::with_data(
        format!("{EVENT_ACCOUNT}xt.demo"),
        Arc::new(a0),
    ))
    .unwrap();
    eng.put(Event::with_data(
        format!("{EVENT_ACCOUNT}xt.hedge"),
        Arc::new(a1),
    ))
    .unwrap();

    assert!(wait_until(Duration::from_secs(2), || demo.len() == 1
        && hedge.len() == 1));
    eng.stop();
    assert_eq!(demo.last().unwrap().balance, 100_000.0);
    assert_eq!(hedge.last().unwrap().frozen, 1_200.5);
}