ek-ui 0.2.21

//! T5 界面线 CTP 实连全流程(集成测试战役;`--features ctp-native` + `#[ignore]`)。
//!
//! 形态 = **无头的界面控制**:与真实窗口同一 UiShared/UiCmd/ask 链路
//! (set_ctp_form = 手工输入位,push_cmd = 钮点击,eTimer 请辞 = 插件出队),
//! 只是没有像素窗口。账号从**命令行环境变量**手工注入(测试者手敲,
//! 插件与界面都不读配置文件 —— 零配置口径)。
//!
//! 运行(手工输入示例):
//! ```text
//! $env:EK_CTP_USER='278153'; $env:EK_CTP_PASSWORD='...'; $env:EK_CTP_BROKER='9999'
//! $env:EK_CTP_APP_ID='simnow_client_test'; $env:EK_CTP_AUTH_CODE='0000000000000000'
//! $env:EK_CTP_TD='tcp://182.254.243.31:40001'; $env:EK_CTP_MD='tcp://182.254.243.31:40011'
//! $env:EK_CTP_SYMBOLS='sn2709'; $env:EK_CTP_PRICE='378550'(行情板空时的价格锚)
//! cargo test --features ctp-native --test ui_ctp_live -- --ignored --nocapture
//! ```
//!
//! 步骤表(12 步,判据见每步注释)与 docs/ek-integration-test-plan.md T5 对齐。

#![cfg(feature = "ctp-native")]

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

use ek_core::main_engine::MainEngine;
use ek_data::constant::{Direction, Exchange, Offset, OrderType};
use ek_data::order_request::OrderRequest;
use ek_event::{EVENT_TIMER, Event};
use ek_station::StationConfig;

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

fn env(key: &str) -> String {
    std::env::var(key).unwrap_or_else(|_| panic!("手工输入缺失: 请先设 ${key}(见文件头示例)"))
}

fn env_opt(key: &str) -> String {
    std::env::var(key).unwrap_or_default()
}

fn put_timer(engine: &Arc<MainEngine>) {
    engine
        .event_engine()
        .put(Event::new(EVENT_TIMER.to_string()))
        .expect("timer put");
}

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

/// 失败诊断:把日志环全量打出来(判挂点在哪一环)。
fn dump_logs(shared: &UiShared, tag: &str) {
    let (lines, _) = shared.logs_snapshot();
    println!("── {tag} 日志环({} 条)──", lines.len());
    for l in lines {
        println!("  {l}");
    }
}

fn log_has(shared: &UiShared, needle: &str) -> bool {
    shared.logs_snapshot().0.iter().any(|l| l.contains(needle))
}

fn logs_since(shared: &UiShared, mark: usize) -> Vec<String> {
    let (lines, _) = shared.logs_snapshot();
    lines.iter().skip(mark).cloned().collect()
}

/// T5 全流:手工配置 → 启动链 → 合约 → 订阅 → 行情上屏 → 发单(可成交/挂单)
/// → 撤单 → 拒单 → 资金 → 持仓 → 错密码 → 改密重连。
#[test]
#[ignore = "CTP 实连(SimNow);需环境变量手工注入账号 + 可达前置"]
fn t5_ui_ctp_live_full_flow() {
    let user = env("EK_CTP_USER");
    let password = env("EK_CTP_PASSWORD");
    let broker = env_opt("EK_CTP_BROKER");
    let broker = if broker.is_empty() {
        "9999"
    } else {
        broker.as_str()
    };
    let app_id = env_opt("EK_CTP_APP_ID");
    let app_id = if app_id.is_empty() {
        "simnow_client_test"
    } else {
        app_id.as_str()
    };
    let auth_code = env_opt("EK_CTP_AUTH_CODE");
    let td_addr = env("EK_CTP_TD");
    let md_addr = env("EK_CTP_MD");
    let symbols: Vec<String> = env_opt("EK_CTP_SYMBOLS")
        .split(',')
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
        .collect();
    let symbols = if symbols.is_empty() {
        vec!["sn2709".to_string()]
    } else {
        symbols
    };
    let main_symbol = symbols[0].clone();

    let engine = Arc::new(MainEngine::new(None));
    let shared = Arc::new(UiShared::new());
    // ui 先上(观察序 = 装配序,启动链日志一条不落)。
    let cfg = StationConfig {
        plugins: vec![
            "ui".to_string(),
            "dm".to_string(),
            "oms".to_string(),
            "ctp".to_string(),
        ],
    };
    boot_start(&engine, &shared, &cfg);

    // ── 步 1 手工输入 + 连接(连接 CTP 钮)────────────────────────
    shared.set_ctp_form(CtpForm {
        broker_id: broker.to_string(),
        td_addr: td_addr.clone(),
        td_user: user.clone(),
        td_password: password.clone(),
        td_auth_code: auth_code.clone(),
        td_app_id: app_id.to_string(),
        md_addr: md_addr.clone(),
        ..CtpForm::default()
    });
    shared.push_cmd(UiCmd::CtpConfigure {
        gateway: "ctp".to_string(),
        form: shared.ctp_form(),
    });
    put_timer(&engine);
    assert!(
        wait_until("ctp.configure 受理", 10, || log_has(
            &shared,
            "ctp.configure 受理"
        )),
        "步1 FAIL: configure 未受理"
    );
    println!("[T5-1] configure 受理 PASS");

    // ── 步 2 启动链六日志(TD 先于 MD 语义)──────────────────────
    for needle in [
        "交易服务器连接成功",
        "交易服务器授权验证成功",
        "交易服务器登录成功",
        "结算信息确认成功",
        "行情服务器连接成功",
        "行情服务器登录成功",
    ] {
        if !wait_until(needle, 45, || log_has(&shared, needle)) {
            dump_logs(&shared, "步2");
            panic!("步2 FAIL: 启动链缺日志 {needle}");
        }
    }
    println!("[T5-2] 启动链六日志 PASS");

    // ── 步 3 合约查询(夜盘开场 jam 容忍:120s)──────────────────
    assert!(
        wait_until("合约信息查询成功", 120, || log_has(
            &shared,
            "合约信息查询成功"
        )),
        "步3 FAIL: 合约查询未收尾(环境侧 jam 亦在此显形,换时段复跑)"
    );
    println!("[T5-3] 合约查询收尾 PASS");

    // ── 步 4 订阅(订阅行情 钮)──────────────────────────────────
    shared.push_cmd(UiCmd::CtpSubscribe {
        gateway: "ctp".to_string(),
        symbols: symbols.clone(),
    });
    put_timer(&engine);
    assert!(
        wait_until("ctp.subscribe 受理", 10, || log_has(
            &shared,
            "ctp.subscribe"
        )),
        "步4 FAIL: 订阅未受理"
    );
    println!("[T5-4] 订阅受理 PASS");

    // ── 步 5 行情上屏(dm 转 eBar/eBarUpdate → 行情板计数)───────
    // 7x24 套快照推送稀疏(实测 90s 内可零推送):软判据,零帧记环境侧,
    // 价格锚回落手工输入位 EK_CTP_PRICE。
    let market_ok = wait_until("行情板出帧", 180, || {
        let (frames, updates, _) = shared.market().counts();
        frames > 0 || updates > 0
    });
    let (frames, updates, _) = shared.market().counts();
    println!(
        "[T5-5] 行情上屏 {}(frames={frames} updates={updates};零推送 = 7x24 快照稀疏,环境侧记档)",
        if market_ok { "PASS" } else { "OBSERVE" }
    );

    // 取主标的最新收盘(下单价格锚;行情板空则回落手工输入 EK_CTP_PRICE)。
    let close = shared
        .market()
        .cells()
        .iter()
        .find(|c| c.symbol == main_symbol)
        .and_then(|c| c.bars.last().map(|b| b.close_price))
        .unwrap_or(0.0);
    let anchor = if close > 0.0 {
        close
    } else {
        env_opt("EK_CTP_PRICE").parse::<f64>().unwrap_or(0.0)
    };
    println!("[T5] 价格锚 {main_symbol} close={close} 采用={anchor}");
    assert!(
        anchor > 0.0,
        "步5 FAIL: 行情板与 EK_CTP_PRICE 双空,无价格锚"
    );

    // ── 步 6 发单(可成交价;票据网关位 = ctp)───────────────────
    // 卖开挂锚价会悬在卖一之上不成;下压 1% 吃穿买一(仍在全天涨跌停内)。
    // 网格钉(多账号面 0.2.15:oms 装配的 `ek-risk:contract` 预检按合约
    // price_tick 判网格,不在格内 = 发送前整体拒)→ 价格一律贴格取整。
    let tick = shared
        .contracts(&main_symbol)
        .iter()
        .find(|c| c.symbol == main_symbol)
        .map(|c| c.price_tick)
        .filter(|t| *t > 0.0)
        .unwrap_or(1.0);
    let snap_grid = |p: f64| (p / tick).round() * tick;
    println!("[T5] {main_symbol} price_tick={tick}(网格贴格由合约表取,不猜)");
    let fill_price = snap_grid(anchor * 0.99);
    let before_ids: Vec<String> = shared
        .trade_snap()
        .orders
        .iter()
        .map(|o| o.orderid.clone())
        .collect();
    shared.set_order_form(ek_ui::state::OrderForm {
        exchange: "SHFE".to_string(),
        symbol: main_symbol.clone(),
        direction: Direction::SHORT,
        offset: Offset::OPEN,
        order_type: OrderType::LIMIT,
        price: format!("{fill_price}"),
        volume: "1".to_string(),
        gateway: "ctp".to_string(),
        follow_price: false,
        auto_split: true,
    });
    shared.push_cmd(UiCmd::SendOrder {
        req: OrderRequest {
            symbol: main_symbol.clone(),
            exchange: Exchange::SHFE,
            direction: Direction::SHORT,
            order_type: OrderType::LIMIT,
            volume: 1.0,
            price: fill_price,
            offset: Offset::OPEN,
            ..OrderRequest::default()
        },
        gateway: "ctp".to_string(),
        mode: ek_oms::LockMode::Auto,
    });
    put_timer(&engine);
    // 本轮新单(柜台流回放会把历史委托灌进账本 → 判据必须排除旧行)。
    if !wait_until("本轮 ctp 委托上屏", 15, || {
        shared
            .trade_snap()
            .orders
            .iter()
            .any(|o| o.gateway_name == "ctp" && !before_ids.contains(&o.orderid))
    }) {
        dump_logs(&shared, "步6");
        panic!("步6 FAIL: ctp 委托未入 oms(本轮无新行;发单可能被闸拒)");
    }
    let new_order = shared
        .trade_snap()
        .orders
        .iter()
        .find(|o| o.gateway_name == "ctp" && !before_ids.contains(&o.orderid))
        .cloned()
        .expect("本轮新委托行");
    assert_ne!(
        new_order.status,
        ek_data::constant::Status::REJECTED,
        "步6 FAIL: 发单被拒(status_msg={})",
        new_order.status_msg
    );
    let filled = wait_until("ctp 成交回报", 30, || {
        shared
            .trade_snap()
            .trades
            .iter()
            .any(|t| t.gateway_name == "ctp" && t.orderid == new_order.orderid)
    });
    println!(
        "[T5-6] 发单 PASS({} @{} status={:?};成交={} ;无成交 = 快照市不活跃,环境侧记档)",
        new_order.orderid, fill_price, new_order.status, filled
    );

    // ── 步 7 挂单 + 撤单(悬价在卖一之上)────────────────────────
    // 挂价口径:锚价 ×1.05 贴格(合约 price_tick 网格内 + 涨跌停带内;
    // 早先取 0.5× 在活跃市会被柜台判废单,撤单面无单可撤)。
    let rest_price = snap_grid(fill_price * 1.05).max(tick);
    shared.push_cmd(UiCmd::SendOrder {
        req: OrderRequest {
            symbol: main_symbol.clone(),
            exchange: Exchange::SHFE,
            direction: Direction::LONG,
            order_type: OrderType::LIMIT,
            volume: 1.0,
            price: rest_price,
            offset: Offset::OPEN,
            ..OrderRequest::default()
        },
        gateway: "ctp".to_string(),
        mode: ek_oms::LockMode::Auto,
    });
    put_timer(&engine);
    let is_rest = |o: &ek_data::order::OrderData| {
        o.gateway_name == "ctp"
            && (o.price - rest_price).abs() < f64::EPSILON
            && !before_ids.contains(&o.orderid)
    };
    let resting = wait_until("挂单上屏", 15, || {
        shared.trade_snap().orders.iter().any(is_rest)
    });
    if !resting {
        dump_logs(&shared, "步7");
        let rows: Vec<String> = shared
            .trade_snap()
            .orders
            .iter()
            .map(|o| format!("{}@{} {:?}", o.orderid, o.price, o.status))
            .collect();
        panic!("步7 FAIL: 挂单未上屏(账本行:{rows:?})");
    }
    let rest_orderid = shared
        .trade_snap()
        .orders
        .iter()
        .find(|o| is_rest(o))
        .map(|o| o.orderid.clone())
        .expect("resting orderid");
    shared.push_cmd(UiCmd::CancelOrder {
        orderid: rest_orderid.clone(),
        gateway: "ctp".to_string(),
    });
    put_timer(&engine);
    if !wait_until("挂单转 CANCELLED", 20, || {
        shared
            .trade_snap()
            .orders
            .iter()
            .any(|o| o.orderid == rest_orderid && o.status == ek_data::constant::Status::CANCELLED)
    }) {
        dump_logs(&shared, "步7");
        let row = shared
            .trade_snap()
            .orders
            .iter()
            .find(|o| o.orderid == rest_orderid)
            .map(|o| format!("{:?} msg={}", o.status, o.status_msg))
            .unwrap_or_else(|| "行不见".to_string());
        panic!("步7 FAIL: 撤单未达 CANCELLED(挂单 {rest_orderid} @{rest_price} 终态 = {row})");
    }
    println!("[T5-7] 挂单+撤单 PASS({rest_orderid} @{rest_price})");

    // ── 步 8 拒单(不存在合约)───────────────────────────────────
    // 口径变更钉(多账号面 0.2.15:oms 装配 ek-risk ContractGate 打头四闸):
    // 未知合约现在在**发送前**就被闸整体拒(`contract:missing`),不再到柜台
    // 拿 ErrorID 16 —— 旧「交易委托失败」柜台日志面由此退役(CTP 错误回报面
    // 仍由 ek-ctp 自有测试钉)。判据改两条:闸拒日志在场 + 账本不预记账。
    let mark = shared.logs_snapshot().0.len();
    let rows_before = shared.trade_snap().orders.len();
    let _ = mark; // 日志环已满容量(合约状态洪水 500+ 行)→ 索引差不可靠,判据改内容钉
    shared.push_cmd(UiCmd::SendOrder {
        req: OrderRequest {
            symbol: "NOPE9999".to_string(),
            exchange: Exchange::SHFE,
            direction: Direction::LONG,
            order_type: OrderType::LIMIT,
            volume: 1.0,
            price: 100.0,
            offset: Offset::OPEN,
            ..OrderRequest::default()
        },
        gateway: "ctp".to_string(),
        mode: ek_oms::LockMode::Auto,
    });
    put_timer(&engine);
    let hit = |needle: &str| {
        shared
            .logs_snapshot()
            .0
            .iter()
            .any(|l| l.contains("NOPE9999") && l.contains(needle))
    };
    if !wait_until("闸拒日志", 20, || {
        hit("contract:missing") || hit("交易委托失败")
    }) {
        dump_logs(&shared, "步8");
        panic!("步8 FAIL: 未知合约发单既无闸拒日志也无柜台拒单日志");
    }
    let gate_rejected = hit("contract:missing");
    assert_eq!(
        shared.trade_snap().orders.len(),
        rows_before,
        "步8 FAIL: 闸拒必须零腿发出、不预记账"
    );
    println!(
        "[T5-8] 拒单探针 PASS({})",
        if gate_rejected {
            "闸拒 contract:missing(发送前整体拒)"
        } else {
            "柜台拒单日志(闸未挡)"
        }
    );

    // ── 步 9 资金(eTimer 轮转 → eAccount → oms → 监控页帧)──────
    assert!(
        wait_until("资金帧上屏", 40, || !shared
            .trade_snap()
            .accounts
            .is_empty()),
        "步9 FAIL: 资金未入账"
    );
    let acct = &shared.trade_snap().accounts[0];
    println!(
        "[T5-9] 资金 PASS(accountid={} balance={:.2})",
        acct.accountid, acct.balance
    );

    // ── 步 10 持仓(ePositionDone 收尾)─────────────────────────
    assert!(
        wait_until("持仓帧上屏", 60, || shared
            .trade_snap()
            .positions
            .iter()
            .any(|p| p.gateway_name == "ctp")),
        "步10 FAIL: 持仓未入账"
    );
    println!("[T5-10] 持仓 PASS");

    // ── 步 11 错密码 configure(登录失败显形)───────────────────
    let mark = shared.logs_snapshot().0.len();
    shared.push_cmd(UiCmd::CtpConfigure {
        gateway: "ctp".to_string(),
        form: CtpForm {
            td_password: "wrong-password-xyz".to_string(),
            ..shared.ctp_form()
        },
    });
    put_timer(&engine);
    let rejected = wait_until("错密码登录失败日志", 40, || {
        logs_since(&shared, mark)
            .iter()
            .any(|l| l.contains("交易服务器登录失败") || l.contains("登录失败"))
    });
    println!(
        "[T5-11] 错密码面 {}(已连接态重 Init 的登录拒 = 端口显形;若会话复用不重登 = 记观察项)",
        if rejected { "PASS" } else { "OBSERVE" }
    );

    // ── 步 12 改密再 configure(守卫复位 → 再登录成功)──────────
    let mark = shared.logs_snapshot().0.len();
    shared.push_cmd(UiCmd::CtpConfigure {
        gateway: "ctp".to_string(),
        form: CtpForm {
            td_password: password.clone(),
            ..shared.ctp_form()
        },
    });
    put_timer(&engine);
    let back = wait_until("改密后登录成功日志", 40, || {
        logs_since(&shared, mark)
            .iter()
            .any(|l| l.contains("交易服务器登录成功"))
    });
    println!(
        "[T5-12] 改密重连 {}(同会话不重发登录 = 环境侧记档)",
        if back { "PASS" } else { "OBSERVE" }
    );

    // ── 步 13 管线平仓(0.1.1 BD-UI-30:步 6 的 SHORT 若成交,经
    //    oms.submit_order 平掉 = CLOSETODAY 拆腿实连证据;无仓 = OBSERVE)──
    let has_short = shared.trade_snap().positions.iter().any(|p| {
        p.gateway_name == "ctp"
            && p.symbol == main_symbol
            && p.direction == Direction::SHORT
            && p.volume > 0.0
    });
    if !has_short {
        println!("[T5-13] 平仓 OBSERVE(步6 未成交无仓可平,7x24 快照市记档)");
    } else {
        let trades_before = shared
            .trade_snap()
            .trades
            .iter()
            .filter(|t| t.gateway_name == "ctp" && t.symbol == main_symbol)
            .count();
        let mark = shared.logs_snapshot().0.len();
        let close_price = (anchor * 1.01).ceil();
        shared.push_cmd(UiCmd::SendOrder {
            req: OrderRequest {
                symbol: main_symbol.clone(),
                exchange: Exchange::SHFE,
                direction: Direction::LONG,
                order_type: OrderType::LIMIT,
                volume: 1.0,
                price: close_price,
                offset: Offset::CLOSE,
                ..OrderRequest::default()
            },
            gateway: "ctp".to_string(),
            mode: ek_oms::LockMode::Auto,
        });
        put_timer(&engine);
        // 管线日志钉(submit … parent) + 成交增量钉(CLOSETODAY 腿吃穿买一)。
        let pipeline_log = wait_until("管线平仓日志", 20, || {
            logs_since(&shared, mark)
                .iter()
                .any(|l| l.contains("[ops] submit"))
        });
        let closed = wait_until("平仓成交回报", 30, || {
            shared
                .trade_snap()
                .trades
                .iter()
                .filter(|t| t.gateway_name == "ctp" && t.symbol == main_symbol)
                .count()
                > trades_before
        });
        assert!(pipeline_log, "步13 FAIL: 平仓未经 oms 管线(日志缺失)");
        println!(
            "[T5-13] 管线平仓 {}(成交={}(CLOSETODAY 实连);无成交 = 快照市不活跃,环境侧记档)",
            if closed { "PASS" } else { "OBSERVE" },
            closed
        );
    }

    // 收场:优雅 close(与窗口关闭同路)。
    shared.request_stop();
    assert!(
        wait_until("观察线程收场", 15, || log_has(
            &shared,
            "窗口关闭,close_all"
        )),
        "收场 FAIL: 观察线程未动"
    );
    println!("[T5] 收场 PASS(全链 13 步完成)");
}

fn boot_start(engine: &Arc<MainEngine>, shared: &Arc<UiShared>, cfg: &StationConfig) {
    ek_station::boot(engine, &ui_kinds(shared, true), cfg).expect("boot");
    engine.start().expect("start");
    // 观察线程(runtime 同款:stop 位 → close_all)。
    ek_ui::runtime::spawn_host_watcher(engine, shared);
}