ek-ui 0.2.2

//! S5 验收:Sim 撮合金样本(成交/持仓/手数排队)+ 全链闭环
//! (eTick→DM→eBar→Sim 成交→eOrder/eTrade→OMS→监控页)+ 撤单 + 流控查询。

use std::sync::Arc;
use std::time::Duration;

use ek_core::main_engine::MainEngine;
use ek_data::bar::BarData;
use ek_data::constant::{Direction, Exchange, Offset, OrderType, Status};
use ek_data::event::{EVENT_CONTRACT, EVENT_TICK};
use ek_data::order_request::OrderRequest;
use ek_data::tick::TickData;
use ek_event::{EVENT_TIMER, Event};
use ek_oms::LockMode;
use ek_station::{StationConfig, boot};

use ek_ui::runtime::ui_kinds;
use ek_ui::sim::SimMatcher;
use ek_ui::state::{UiCmd, UiShared};

const BASE: i64 = 36_000;
const NANOS: i64 = 1_000_000_000;

fn bar(dt_s: i64, o: f64, h: f64, l: f64, c: f64, v: f64) -> BarData {
    BarData {
        symbol: "rb2605".to_string(),
        exchange: Exchange::SHFE,
        datetime: dt_s * NANOS,
        interval: Some(ek_data::constant::Interval::MINUTE),
        volume: v,
        open_price: o,
        high_price: h,
        low_price: l,
        close_price: c,
        ..BarData::default()
    }
}

fn req(price: f64, volume: f64) -> OrderRequest {
    OrderRequest {
        symbol: "rb2605".to_string(),
        exchange: Exchange::SHFE,
        direction: Direction::LONG,
        order_type: OrderType::LIMIT,
        volume,
        price,
        offset: Offset::OPEN,
        ..OrderRequest::default()
    }
}

/// 撮合金样本:FIFO 手数排队 + 预算耗尽排队 + 双向 + 市价 + 撤单 + 拒单口径。
#[test]
fn u0_matcher_golden() {
    let mut m = SimMatcher::default();
    let o1 = m.place(&req(100.0, 10.0)).unwrap();
    let o2 = m.place(&req(100.0, 8.0)).unwrap();
    let o3 = m.place(&req(100.0, 5.0)).unwrap();
    assert_eq!(o1.orderid, "SIM_0_1");
    assert_eq!(o2.orderid, "SIM_0_2");
    assert_eq!(o3.orderid, "SIM_0_3");
    assert_eq!(o1.status, Status::SUBMITTING);

    // bar1 预算 18:先挂先吃 —— o1 全成(10),o2 全成(8),预算尽;o3 排队。
    let outs = m.on_bar(&bar(BASE, 100.0, 103.0, 99.5, 102.0, 18.0));
    assert_eq!(outs.len(), 2);
    assert_eq!(outs[0].trade.tradeid, "T1");
    assert_eq!(outs[0].trade.volume, 10.0);
    assert_eq!(outs[0].trade.price, 100.0, "限价成交于委托价");
    assert_eq!(outs[0].order.status, Status::ALLTRADED);
    assert_eq!(outs[1].trade.tradeid, "T2");
    assert_eq!(outs[1].trade.volume, 8.0);
    assert_eq!(m.resting.len(), 1);
    assert_eq!(m.resting[0].orderid, "SIM_0_3");
    assert_eq!(m.resting[0].traded, 0.0);
    let pos = m
        .positions
        .get(&("rb2605".to_string(), "LONG".to_string()))
        .unwrap();
    assert_eq!(pos.volume, 18.0);
    assert_eq!(pos.avg_price, 100.0);

    // bar2 预算 5(low 99 ≤ 100):o3 排队到下一拍全成。
    let outs = m.on_bar(&bar(BASE + 60, 102.0, 104.0, 99.0, 103.0, 5.0));
    assert_eq!(outs.len(), 1);
    assert_eq!(outs[0].trade.tradeid, "T3");
    assert_eq!(outs[0].order.status, Status::ALLTRADED);
    assert!(m.resting.is_empty());
    assert_eq!(
        m.positions
            .get(&("rb2605".to_string(), "LONG".to_string()))
            .unwrap()
            .volume,
        23.0
    );

    // 卖向:SHORT 105,bar high 103 不成;high 106 成于 105。
    let mut sreq = req(105.0, 4.0);
    sreq.direction = Direction::SHORT;
    m.place(&sreq).unwrap();
    assert!(
        m.on_bar(&bar(BASE + 120, 103.0, 103.0, 102.0, 103.0, 9.0))
            .is_empty()
    );
    let outs = m.on_bar(&bar(BASE + 180, 103.0, 106.0, 103.0, 105.5, 9.0));
    assert_eq!(outs.len(), 1);
    assert_eq!(outs[0].trade.price, 105.0);
    assert_eq!(outs[0].trade.tradeid, "T4");
    assert_eq!(
        m.positions
            .get(&("rb2605".to_string(), "SHORT".to_string()))
            .unwrap()
            .volume,
        4.0
    );

    // 市价单:成交于 bar.close。
    let mut mreq = req(0.0, 2.0);
    mreq.order_type = OrderType::MARKET;
    m.place(&mreq).unwrap();
    let outs = m.on_bar(&bar(BASE + 240, 105.0, 106.0, 104.0, 105.5, 2.0));
    assert_eq!(outs[0].trade.price, 105.5);

    // 撤单:在挂即 CANCELLED;未知号 None。
    let mut far = req(50.0, 1.0);
    far.direction = Direction::SHORT;
    m.place(&far).unwrap();
    let cancelled = m.cancel("SIM_0_6").unwrap();
    assert_eq!(cancelled.status, Status::CANCELLED);
    assert!(m.cancel("SIM_0_99").is_none());

    // 拒单口径(与 ek-ctp 对齐):开平 NONE / 类型不支持 / 量非正。
    let mut bad = req(100.0, 1.0);
    bad.offset = Offset::NONE;
    assert_eq!(m.place(&bad).unwrap_err(), "请选择开平方向");
    let mut bad = req(100.0, 1.0);
    bad.order_type = OrderType::FAK;
    assert!(m.place(&bad).unwrap_err().contains("不支持"));
    assert!(m.place(&req(100.0, 0.0)).unwrap_err().contains("为正"));
}

fn put_evt(engine: &Arc<MainEngine>, e: Event) {
    engine.event_engine().put(e).expect("event put");
}

fn put_timer(engine: &Arc<MainEngine>) {
    put_evt(engine, Event::new(EVENT_TIMER.to_string()));
}

fn wait_until(win: &str, secs: u64, mut f: impl FnMut() -> bool) {
    for _ in 0..(secs * 20) {
        if f() {
            return;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("timeout: {win}");
}

/// 全链闭环:eTick→(dm)→eBar→(sim)→eOrder/eTrade/ePosition→(oms)→监控页帧。
/// tick 全带 last_volume=0(tick 面零预算)→ 成交只由 bar 预算驱动(金样本清晰)。
#[test]
fn u1_full_chain_closed_loop() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec![
            "dm".to_string(),
            "oms".to_string(),
            "sim".to_string(),
            "ui".to_string(),
        ],
    };
    boot(&engine, &ui_kinds(&shared, true), &cfg).unwrap();
    engine.start().unwrap();

    // 两笔买 102.5(委托价高于 bar low → bar 收口时成交)。0.1.1:走 oms 管线。
    shared.push_cmd(UiCmd::SendOrder {
        req: req(102.5, 10.0),
        gateway: "sim".to_string(),
        mode: LockMode::Auto,
    });
    shared.push_cmd(UiCmd::SendOrder {
        req: req(102.5, 8.0),
        gateway: "sim".to_string(),
        mode: LockMode::Auto,
    });
    put_timer(&engine); // 辞行 → sim 受理(SUBMITTING 单已上总线入 oms)

    wait_until("submitting orders in oms", 6, || {
        shared.trade_snap().orders.len() == 2
    });
    assert!(
        shared
            .trade_snap()
            .orders
            .iter()
            .all(|o| o.status == Status::SUBMITTING)
    );

    // 金样本 tick 流(与 ui_market m0 同轴;last_volume 全 0 → tick 零预算)。
    let ticks = [
        (BASE - 55, 101.0, 10.0),
        (BASE - 30, 103.0, 15.0),
        (BASE - 10, 102.0, 18.0),
        (BASE + 10, 101.0, 25.0),
    ];
    for (dt, last, vol) in ticks {
        put_evt(
            &engine,
            Event::with_data(
                EVENT_TICK.to_string(),
                Arc::new(TickData {
                    gateway_name: "sim".to_string(),
                    symbol: "rb2605".to_string(),
                    exchange: Exchange::SHFE,
                    datetime: dt * NANOS,
                    volume: vol,
                    last_price: last,
                    last_volume: 0.0,
                    ..TickData::default()
                }),
            ),
        );
    }

    // bar1 收口(low 102 ≤ 102.5)→ sim 成交 → oms 入账;eTimer 拍上屏。
    put_timer(&engine);
    wait_until("fills in trade snap", 6, || {
        let ts = shared.trade_snap();
        ts.trades.len() == 2 && ts.orders.iter().all(|o| o.status == Status::ALLTRADED)
    });

    let ts = shared.trade_snap();
    // 委托读数逐位。
    let mut orders = ts.orders.clone();
    orders.sort_by(|a, b| a.orderid.cmp(&b.orderid));
    assert_eq!(orders[0].traded, 10.0);
    assert_eq!(orders[0].traded_price, 102.5);
    assert_eq!(orders[1].traded, 8.0);
    // 成交读数逐位(FIFO 笔号)。
    assert_eq!(ts.trades[0].tradeid, "T1");
    assert_eq!(ts.trades[0].volume, 10.0);
    assert_eq!(ts.trades[0].price, 102.5);
    assert_eq!(ts.trades[1].tradeid, "T2");
    assert_eq!(ts.trades[1].volume, 8.0);
    // 持仓读数。
    assert_eq!(ts.positions.len(), 1);
    assert_eq!(ts.positions[0].direction, Direction::LONG);
    assert_eq!(ts.positions[0].volume, 18.0);
    assert_eq!(ts.positions[0].avg_price, 102.5);

    // 下单回执日志在场(ops 面可断言;0.1.1 = oms 管线 parent 聚合形)。
    assert!(
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("submit rb2605 LONG OPENx10 → sim.SIM_0_1 (parent P1)")),
        "send ack log missing"
    );
}

/// 撤单闭环:不可成交委托 → 监控页见 SUBMITTING → 撤 → CANCELLED。
#[test]
fn u2_cancel_loop() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec![
            "dm".to_string(),
            "sim".to_string(),
            "oms".to_string(),
            "ui".to_string(),
        ],
    };
    boot(&engine, &ui_kinds(&shared, true), &cfg).unwrap();
    engine.start().unwrap();

    shared.push_cmd(UiCmd::SendOrder {
        req: req(50.0, 3.0),
        gateway: "sim".to_string(),
        mode: LockMode::Auto,
    }); // 远低于市价,bar 不成
    put_timer(&engine);
    wait_until("order visible", 6, || shared.trade_snap().orders.len() == 1);

    shared.push_cmd(UiCmd::CancelOrder {
        orderid: "SIM_0_1".to_string(),
        gateway: "sim".to_string(),
    });
    put_timer(&engine);
    wait_until("cancelled in snap", 6, || {
        shared
            .trade_snap()
            .orders
            .iter()
            .any(|o| o.status == Status::CANCELLED)
    });
    assert!(
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("[ops] 已撤 SIM_0_1")),
        "cancel ack log missing"
    );
}

/// 流控查询:cmd → ask sim.query_rate → 回执进日志(计数面可断言)。
#[test]
fn u3_query_rate_loop() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        // 0.1.1:下单走 oms 管线,u3 也需 oms 在场(薄下单者无旁路口)。
        plugins: vec!["sim".to_string(), "oms".to_string(), "ui".to_string()],
    };
    boot(&engine, &ui_kinds(&shared, true), &cfg).unwrap();
    engine.start().unwrap();

    shared.push_cmd(UiCmd::SendOrder {
        req: req(100.0, 1.0),
        gateway: "sim".to_string(),
        mode: LockMode::Auto,
    }); // 计一笔流控(查询口不计数)
    shared.push_cmd(UiCmd::QueryRate);
    put_timer(&engine);
    wait_until("rate log", 6, || {
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("sim.query_rate used=1/6 余5"))
    });
}

/// T7 注错:1s 窗内第 7 笔请求被流控拒绝(6 收 1 拒,超限显形不进 oms)。
#[test]
fn u7_rate_limit_seventh_rejected() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec!["sim".to_string(), "oms".to_string(), "ui".to_string()],
    };
    boot(&engine, &ui_kinds(&shared, true), &cfg).unwrap();
    engine.start().unwrap();

    // 7 笔入列,一拍请辞(同一 1s 窗)。oms 地板 20/s 放行,sim 网关 6/s 拒第 7。
    for _ in 0..7 {
        shared.push_cmd(UiCmd::SendOrder {
            req: req(50.0, 1.0), // 远价不成交,只数流控
            gateway: "sim".to_string(),
            mode: LockMode::Auto,
        });
    }
    put_timer(&engine);
    wait_until("流控被拒日志", 8, || {
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("被拒: send:流控被拒:1s 窗内请求已达上限 6"))
    });
    assert!(
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("流控被拒:1s 窗内请求已达上限 6")),
        "第 7 笔必须被流控拒绝"
    );
    // 6 收 1 拒:oms 只见 6 笔委托(再拍 eTimer 让 ui 插件刷新交易帧)。
    for _ in 0..8 {
        put_timer(&engine);
        std::thread::sleep(Duration::from_millis(250));
        if shared.trade_snap().orders.len() == 6 {
            break;
        }
    }
    let n = shared.trade_snap().orders.len();
    assert_eq!(n, 6, "被拒笔不得进 oms");
    assert_eq!(
        shared
            .logs_snapshot()
            .0
            .iter()
            .filter(|l| l.contains("流控被拒"))
            .count(),
        1,
        "恰一笔被拒日志"
    );
}

/// 双重计价防护:tick 成交挂账,bar 收口预算扣回 —— 同一笔市场量只吃一次。
#[test]
fn u5_tick_bar_budget_single_count() {
    let mut m = SimMatcher::default();
    m.place(&req(100.0, 10.0)).unwrap();
    // tick 先吃 6 手(last_volume=6,last 100 ≤ 委托价)。
    let t = TickData {
        gateway_name: "sim".to_string(),
        symbol: "rb2605".to_string(),
        exchange: Exchange::SHFE,
        datetime: (BASE - 30) * NANOS,
        volume: 10.0,
        last_price: 100.0,
        last_volume: 6.0,
        ..TickData::default()
    };
    let outs = m.on_tick(&t);
    assert_eq!(outs.len(), 1);
    assert_eq!(outs[0].trade.volume, 6.0);
    assert_eq!(outs[0].order.status, Status::PARTTRADED);
    assert_eq!(m.tick_filled.get("rb2605"), Some(&6.0));

    // bar 收口 volume=10:预算 = 10 − 6 = 4 → 只补 4,总量恰 10(不双重)。
    let outs = m.on_bar(&bar(BASE, 100.0, 101.0, 99.5, 100.5, 10.0));
    assert_eq!(outs.len(), 1);
    assert_eq!(outs[0].trade.volume, 4.0);
    assert_eq!(outs[0].order.status, Status::ALLTRADED);
    assert!(m.resting.is_empty());
    assert_eq!(
        m.positions
            .get(&("rb2605".to_string(), "LONG".to_string()))
            .unwrap()
            .volume,
        10.0
    );
    assert!(!m.tick_filled.contains_key("rb2605"), "收口即清账");
}

/// 重复帧守卫:同 symbol 的 bar.datetime 非严格递增 → 忽略,不重复撮合。
#[test]
fn u6_bar_monotonic_guard() {
    let mut m = SimMatcher::default();
    m.place(&req(100.0, 10.0)).unwrap();
    let b = bar(BASE, 100.0, 103.0, 99.5, 102.0, 18.0);
    assert_eq!(m.on_bar(&b).len(), 1);
    // 同刻重放帧 + 旧时刻帧:全部忽略(成交不复活)。
    assert!(m.on_bar(&b).is_empty());
    assert!(
        m.on_bar(&bar(BASE - 60, 100.0, 103.0, 99.5, 102.0, 18.0))
            .is_empty()
    );
    assert_eq!(m.trade_seq, 1, "重复帧不得加笔");
}

// 票据解析面已退役(v4 下单面板 = OrderForm::to_request,断言迁 ui_state r7)。

/// u8:eTick. 裸键上屏 → tick_store(v4 行情带唯一源;覆盖 + 跨所分组 + 排序)。
#[test]
fn u8_tick_store_from_bus() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec![
            "dm".to_string(),
            "oms".to_string(),
            "sim".to_string(),
            "ui".to_string(),
        ],
    };
    boot(&engine, &ui_kinds(&shared, true), &cfg).unwrap();
    engine.start().unwrap();

    let tick = |ex: Exchange, symbol: &str, dt: i64, last: f64| {
        Event::with_data(
            EVENT_TICK.to_string(),
            Arc::new(TickData {
                gateway_name: "sim".to_string(),
                symbol: symbol.to_string(),
                exchange: ex,
                datetime: dt * NANOS,
                last_price: last,
                ..TickData::default()
            }),
        )
    };
    put_evt(&engine, tick(Exchange::SHFE, "rb2605", BASE, 101.0));
    put_evt(&engine, tick(Exchange::DCE, "a2709", BASE, 3500.0));
    put_evt(&engine, tick(Exchange::SHFE, "rb2605", BASE + 1, 102.0)); // 覆盖

    wait_until("ticks in store", 6, || shared.tick_count() == 2);
    let rows = shared.tick_rows();
    assert_eq!(rows[0].symbol, "a2709", "DCE < SHFE 排序");
    assert_eq!(rows[1].last_price, 102.0, "覆盖生效");
    let groups: Vec<String> = shared
        .tick_groups()
        .iter()
        .map(|(e, _)| e.clone())
        .collect();
    assert_eq!(groups, vec!["DCE".to_string(), "SHFE".to_string()]);
}

/// u9:eContract. 裸键上屏 → 合约库(v4 查询合约窗数据源;覆盖去重)。
#[test]
fn u9_contract_store_from_bus() {
    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    let cfg = StationConfig {
        plugins: vec![
            "dm".to_string(),
            "oms".to_string(),
            "sim".to_string(),
            "ui".to_string(),
        ],
    };
    boot(&engine, &ui_kinds(&shared, true), &cfg).unwrap();
    engine.start().unwrap();

    let contract = |ex: Exchange, symbol: &str, name: &str| {
        Event::with_data(
            EVENT_CONTRACT.to_string(),
            Arc::new(ek_data::contract::ContractData {
                gateway_name: "ctp".to_string(),
                symbol: symbol.to_string(),
                exchange: ex,
                name: name.to_string(),
                asset: ek_data::constant::Asset::default(),
                instrument_type: None,
                size: 10.0,
                price_tick: 1.0,
                product_id: None,
                max_limit_order_volume: None,
                min_limit_order_volume: None,
                max_market_order_volume: None,
                min_market_order_volume: None,
                delivery_year: None,
                delivery_month: None,
                open_date: None,
                pre_close: None,
                up_stop_price: None,
                down_stop_price: None,
                float_volume: None,
                total_volume: None,
                instrument_status: None,
                is_trading: None,
            }),
        )
    };
    put_evt(&engine, contract(Exchange::SHFE, "rb2605", "螺纹钢2605"));
    put_evt(&engine, contract(Exchange::DCE, "a2709", "黄玉米2709"));
    put_evt(&engine, contract(Exchange::SHFE, "rb2605", "螺纹钢2605")); // 覆盖

    wait_until("contracts in store", 6, || shared.contract_count() == 2);
    assert_eq!(shared.contracts("rb").len(), 1);
    assert_eq!(shared.contracts("玉米").len(), 1);
}