ek_ui/

sim.rs

//! S5 SimGatewayPlugin —— 与 ek-ctp 四口 ask 契约同构的**纸面网关**。
//!
//! 四口(`sim.subscribe` / `sim.send_order` / `sim.cancel_order` /
//! `sim.query_rate`):前三口直接复用 ek-ctp 的 Req/Reply 类型(同构 =
//! 同一类型面,接线换 sim 即插真网关);第四口 `query_rate` 为 sim 自定
//! 流控查询(ek-ctp 0.2.0 无此口;CTP 流控语义:1s 窗内已用/上限/余量)。
//!
//! 撮合(纸面):由 `eBar.`(BarsFrame)/`eTick.`(TickData)回放驱动 ——
//! - 限价单:买(bar.low ≤ 委托价) / 卖(bar.high ≥ 委托价)成交于委托价;
//!   市价单成交于 bar.close / tick.last;
//! - **手数排队**:每根 bar 的 volume 是本拍成交预算,挂单按到达序(FIFO)
//!   吃预算,先挂先成交,吃满即排队到下一拍;
//! - 成交产出:eOrder.(状态单)+ eTrade.(成交笔)+ ePosition.(方向位行)
//!   → ek-oms 被动入账 → 监控页 ask 全量查询上屏。
//!
//! 纪律:锁内撮合造快照,锁外 put(不持锁发事件,与 DmPlugin 同形)。

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
use std::time::{Duration, Instant};

use ek_core::context::PluginContext;
use ek_core::envelope::{Envelope, ReplyErr, e_req_type};
use ek_core::plugin::Plugin;
use ek_core::service::ServiceDesc;
use ek_ctp::plugin::{
    CancelOrderReply, CancelOrderReq, SendOrderReply, SendOrderReq, SubscribeReply, SubscribeReq,
};
use ek_data::bar::BarData;
use ek_data::constant::{Direction, Offset, OrderType, Status};
use ek_data::event::{EVENT_BAR, EVENT_ORDER, EVENT_POSITION, EVENT_TICK, EVENT_TRADE};
use ek_data::order::OrderData;
use ek_data::order_request::OrderRequest;
use ek_data::position::PositionData;
use ek_data::tick::TickData;
use ek_data::trade::TradeData;
use ek_dm::wire::BarsFrame;
use ek_event::Event;

/// 网关注册名(工厂名与注册名一致,BD-U5)。
pub const SIM_NAME: &str = "sim";
/// 流控上限(CTP 缺省口径的纸面镜像)。
pub const RATE_LIMIT: u32 = 6;

/// `sim.query_rate`(第四口;sim 自定,其余三口与 ek-ctp 同型)。
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct QueryRateReq;

/// 流控回包:1s 窗内已用 / 上限 / 余量。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QueryRateReply {
    pub used: u32,
    pub limit: u32,
    pub remaining: u32,
}

/// 一次成交的三件套产出(订单状态单 + 成交笔 + 方向位行)。
#[derive(Debug, Clone)]
pub struct FillOut {
    pub order: OrderData,
    pub trade: TradeData,
    pub position: PositionData,
}

/// 1s 流控窗。
#[derive(Debug)]
struct RateWindow {
    used: u32,
    started: Instant,
}

impl RateWindow {
    fn new() -> Self {
        Self {
            used: 0,
            started: Instant::now(),
        }
    }
    fn roll(&mut self) {
        if self.started.elapsed() >= Duration::from_secs(1) {
            self.used = 0;
            self.started = Instant::now();
        }
    }
    fn hit(&mut self) -> bool {
        self.roll();
        if self.used >= RATE_LIMIT {
            // 超限拒绝(T7 注错面:CTP 流控语义的纸面镜像,超限请求显形被拒)。
            return false;
        }
        self.used += 1;
        true
    }
    fn snapshot(&mut self) -> QueryRateReply {
        self.roll();
        QueryRateReply {
            used: self.used,
            limit: RATE_LIMIT,
            remaining: RATE_LIMIT.saturating_sub(self.used),
        }
    }
}

/// 纸面撮合器(纯逻辑零总线;金样本直断此层)。
#[derive(Debug, Default)]
pub struct SimMatcher {
    /// 在挂订单(到达序 = FIFO;终态即摘除)。
    pub resting: Vec<OrderData>,
    /// 已成交笔号序列。
    pub trade_seq: u64,
    /// 委托引用计数(订单号尾段)。
    pub order_ref: u64,
    /// 方向位行(key = symbol × 方向原文;Direction 无 Ord → 键用 as_str)。
    pub positions: BTreeMap<(String, String), PositionData>,
    /// 上次 bar 收口时刻(key = symbol;严格递增守卫 —— 重复/重放帧不重复撮合)。
    pub last_bar_dt: BTreeMap<String, ek_data::time::DtNs>,
    /// 本段已发生的 tick 成交量(key = symbol;bar 收口时从 bar 预算扣回 ——
    /// 防 tick 与 bar 对同一笔市场量双重计价)。
    pub tick_filled: BTreeMap<String, f64>,
}

impl SimMatcher {
    /// 下单校验 + 入挂(拒单口径与 ek-ctp 对齐:开平 NET / 类型不支持 / 量非正)。
    pub fn place(&mut self, req: &OrderRequest) -> Result<OrderData, String> {
        if req.offset == Offset::NONE {
            return Err("请选择开平方向".to_string());
        }
        if !matches!(req.order_type, OrderType::LIMIT | OrderType::MARKET) {
            return Err(format!(
                "当前接口不支持该类型的委托{}",
                req.order_type.as_str()
            ));
        }
        if req.volume <= 0.0 {
            return Err("委托量必须为正".to_string());
        }
        self.order_ref += 1;
        let orderid = format!("SIM_0_{}", self.order_ref);
        let od = req.create_order_data(&orderid, SIM_NAME);
        self.resting.push(od.clone());
        Ok(od)
    }

    /// 撤单(接受本地号或 `sim.` 全号;命中即 CANCELLED 摘挂)。
    pub fn cancel(&mut self, orderid: &str) -> Option<OrderData> {
        let local = orderid.strip_prefix("sim.").unwrap_or(orderid);
        let idx = self.resting.iter().position(|o| o.orderid == local)?;
        let mut od = self.resting.remove(idx);
        od.status = Status::CANCELLED;
        Some(od)
    }

    /// eBar. 驱动:逐 bar 撮合(预算 = bar.volume − 本段 tick 已成;FIFO)。
    ///
    /// 单调守卫:bar.datetime 必须严格大于该 symbol 上次收口时刻(重复帧/
    /// 重放帧直接忽略,不重复撮合);预算扣回本段 tick 成交量(tick 与 bar
    /// 不双重计价同一笔市场量)。
    pub fn on_bar(&mut self, bar: &BarData) -> Vec<FillOut> {
        if self
            .last_bar_dt
            .get(&bar.symbol)
            .is_some_and(|&last| bar.datetime <= last)
        {
            return Vec::new();
        }
        self.last_bar_dt.insert(bar.symbol.clone(), bar.datetime);
        let pending = self.tick_filled.remove(&bar.symbol).unwrap_or(0.0);
        let mut budget = (bar.volume - pending).max(0.0);
        let mut outs = Vec::new();
        let mut i = 0;
        while i < self.resting.len() && budget > 0.0 {
            let (fillable, price) = {
                let o = &self.resting[i];
                if o.symbol != bar.symbol {
                    (false, 0.0)
                } else {
                    match o.direction {
                        Some(Direction::LONG) => match o.order_type {
                            OrderType::MARKET => (true, bar.close_price),
                            _ => (bar.low_price <= o.price, o.price),
                        },
                        Some(Direction::SHORT) => match o.order_type {
                            OrderType::MARKET => (true, bar.close_price),
                            _ => (bar.high_price >= o.price, o.price),
                        },
                        _ => (false, 0.0),
                    }
                }
            };
            if fillable {
                if let Some(out) = self.fill_at(i, price, &mut budget, bar.datetime) {
                    outs.push(out);
                }
            }
            // 全成交即摘(索引不前移);其余前移。
            if self
                .resting
                .get(i)
                .is_some_and(|o| o.status == Status::ALLTRADED)
            {
                self.resting.remove(i);
            } else {
                i += 1;
            }
        }
        outs
    }

    /// eTick. 驱动:单 tick 撮合(预算 = last_volume;FIFO)。
    /// tick 成交量记入本段账(bar 收口时扣回,见 on_bar)。
    pub fn on_tick(&mut self, t: &TickData) -> Vec<FillOut> {
        let mut budget = t.last_volume.max(0.0);
        let mut outs = Vec::new();
        let mut i = 0;
        while i < self.resting.len() && budget > 0.0 {
            let (fillable, price) = {
                let o = &self.resting[i];
                if o.symbol != t.symbol {
                    (false, 0.0)
                } else {
                    match o.direction {
                        Some(Direction::LONG) => match o.order_type {
                            OrderType::MARKET => (true, t.last_price),
                            _ => (t.last_price <= o.price, o.price),
                        },
                        Some(Direction::SHORT) => match o.order_type {
                            OrderType::MARKET => (true, t.last_price),
                            _ => (t.last_price >= o.price, o.price),
                        },
                        _ => (false, 0.0),
                    }
                }
            };
            if fillable {
                if let Some(out) = self.fill_at(i, price, &mut budget, t.datetime) {
                    outs.push(out);
                }
            }
            if self
                .resting
                .get(i)
                .is_some_and(|o| o.status == Status::ALLTRADED)
            {
                self.resting.remove(i);
            } else {
                i += 1;
            }
        }
        // tick 来源成交挂账(bar 预算扣回用;bar 来源不挂)。
        let tick_vol: f64 = outs.iter().map(|o| o.trade.volume).sum();
        if tick_vol > 0.0 {
            *self.tick_filled.entry(t.symbol.clone()).or_insert(0.0) += tick_vol;
        }
        outs
    }

    fn fill_at(
        &mut self,
        idx: usize,
        price: f64,
        budget: &mut f64,
        dt: ek_data::time::DtNs,
    ) -> Option<FillOut> {
        let remaining = self.resting[idx].volume - self.resting[idx].traded;
        let vol = remaining.min(*budget);
        if vol <= 0.0 {
            return None;
        }
        *budget -= vol;
        self.trade_seq += 1;
        let tradeid = format!("T{}", self.trade_seq);
        {
            let o = &mut self.resting[idx];
            o.traded += vol;
            o.traded_price = price;
            o.datetime = Some(dt);
            o.status = if o.traded >= o.volume {
                Status::ALLTRADED
            } else {
                Status::PARTTRADED
            };
        }
        let order = self.resting[idx].clone();
        let trade = TradeData {
            gateway_name: SIM_NAME.to_string(),
            symbol: order.symbol.clone(),
            exchange: order.exchange,
            orderid: order.orderid.clone(),
            tradeid,
            direction: order.direction,
            offset: order.offset,
            price,
            volume: vol,
            datetime: Some(dt),
            traded_amount: price * vol,
            ..TradeData::default()
        };
        let position = self.bump_position(&trade);
        Some(FillOut {
            order,
            trade,
            position,
        })
    }

    fn bump_position(&mut self, t: &TradeData) -> PositionData {
        let dir = t.direction.unwrap_or(Direction::NET);
        let row = self
            .positions
            .entry((t.symbol.clone(), dir.as_str().to_string()))
            .or_insert_with(|| PositionData {
                gateway_name: SIM_NAME.to_string(),
                symbol: t.symbol.clone(),
                exchange: t.exchange,
                direction: dir,
                ..PositionData::default()
            });
        let before = row.volume;
        row.volume += t.volume;
        if row.volume > 0.0 {
            row.avg_price = (row.avg_price * before + t.price * t.volume) / row.volume;
        }
        row.can_use_volume = row.volume;
        row.last_price = t.price;
        row.clone()
    }
}

/// 纸面网关插件(四口 ask + eBar./eTick. 撮合驱动)。
pub struct SimGatewayPlugin {
    inner: Mutex<SimMatcher>,
    subscribed: Mutex<Vec<String>>,
    rate: Mutex<RateWindow>,
    ctx: OnceLock<PluginContext>,
}

impl SimGatewayPlugin {
    /// 缺省构造(注册名 "sim")。
    #[must_use]
    pub fn new() -> Self {
        Self {
            inner: Mutex::new(SimMatcher::default()),
            subscribed: Mutex::new(Vec::new()),
            rate: Mutex::new(RateWindow::new()),
            ctx: OnceLock::new(),
        }
    }

    fn lock(&self) -> MutexGuard<'_, SimMatcher> {
        self.inner.lock().unwrap_or_else(|e| e.into_inner())
    }

    /// 成交三件套上总线(锁外;纪律同 DmPlugin)。
    fn emit_fills(&self, outs: &[FillOut]) {
        let Some(ctx) = self.ctx.get() else { return };
        for o in outs {
            let _ = ctx.put(Event::with_data(
                EVENT_ORDER.to_string(),
                Arc::new(o.order.clone()),
            ));
            let _ = ctx.put(Event::with_data(
                EVENT_TRADE.to_string(),
                Arc::new(o.trade.clone()),
            ));
            let _ = ctx.put(Event::with_data(
                EVENT_POSITION.to_string(),
                Arc::new(o.position.clone()),
            ));
        }
    }

    fn emit_order(&self, od: &OrderData) {
        let Some(ctx) = self.ctx.get() else { return };
        let _ = ctx.put(Event::with_data(
            EVENT_ORDER.to_string(),
            Arc::new(od.clone()),
        ));
    }

    /// eReq.sim:四口 arm 分派(与 ek-dm/ek-ctp 同形)。
    fn on_request(&self, event: &Event) {
        let Some(env) = event.data.downcast_ref::<Envelope>() else {
            return;
        };
        macro_rules! arm {
            ($ty:ty, $res:ty, $body:expr) => {
                if let Some(req) = env.payload.downcast_ref::<$ty>() {
                    let res: $res = $body(req);
                    let _ = env.reply.send(Box::new(Ok::<$res, ReplyErr>(res)));
                    return;
                }
            };
        }
        arm!(SubscribeReq, SubscribeReply, |r: &SubscribeReq| {
            if !self.rate.lock().unwrap().hit() {
                return SubscribeReply {
                    ok: false,
                    msg: format!("流控被拒:1s 窗内请求已达上限 {RATE_LIMIT}"),
                };
            }
            let mut sub = self.subscribed.lock().unwrap();
            for s in &r.symbols {
                if !sub.contains(s) {
                    sub.push(s.clone());
                }
            }
            SubscribeReply {
                ok: true,
                msg: format!("已登记 {} 个合约(纸面)", sub.len()),
            }
        });
        arm!(SendOrderReq, SendOrderReply, |r: &SendOrderReq| {
            if !self.rate.lock().unwrap().hit() {
                return SendOrderReply {
                    ok: false,
                    en_orderid: String::new(),
                    msg: format!("流控被拒:1s 窗内请求已达上限 {RATE_LIMIT}"),
                };
            }
            // 锁内造单,锁外上总线(纪律同 on_mail 撮合面;match 暂存守卫会跨臂存活,故显式开块)。
            let placed = { self.lock().place(&r.req) };
            match placed {
                Ok(od) => {
                    self.emit_order(&od);
                    SendOrderReply {
                        ok: true,
                        en_orderid: format!("{SIM_NAME}.{}", od.orderid),
                        msg: String::new(),
                    }
                }
                Err(msg) => SendOrderReply {
                    ok: false,
                    en_orderid: String::new(),
                    msg,
                },
            }
        });
        arm!(CancelOrderReq, CancelOrderReply, |r: &CancelOrderReq| {
            if !self.rate.lock().unwrap().hit() {
                return CancelOrderReply {
                    ok: false,
                    msg: format!("流控被拒:1s 窗内请求已达上限 {RATE_LIMIT}"),
                };
            }
            let cancelled = { self.lock().cancel(&r.req.orderid) };
            match cancelled {
                Some(od) => {
                    self.emit_order(&od);
                    CancelOrderReply {
                        ok: true,
                        msg: format!("已撤 {}", od.orderid),
                    }
                }
                None => CancelOrderReply {
                    ok: false,
                    msg: "委托不存在/已终态".to_string(),
                },
            }
        });
        arm!(QueryRateReq, QueryRateReply, |_r: &QueryRateReq| {
            self.rate.lock().unwrap().snapshot()
        });
    }
}

impl Default for SimGatewayPlugin {
    fn default() -> Self {
        Self::new()
    }
}

impl Plugin for SimGatewayPlugin {
    fn name(&self) -> &str {
        SIM_NAME
    }

    fn deps(&self) -> &[&str] {
        &[] // 纯总线契约;dm/oms 先后皆可(与 UiPlugin 同理)
    }

    fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
        vec![
            std::borrow::Cow::Borrowed(EVENT_BAR),
            std::borrow::Cow::Borrowed(EVENT_TICK),
            std::borrow::Cow::Owned(e_req_type(SIM_NAME)),
        ]
    }

    fn mailbox_cap(&self) -> usize {
        2048
    }

    fn version(&self) -> std::borrow::Cow<'static, str> {
        std::borrow::Cow::Borrowed(env!("CARGO_PKG_VERSION"))
    }

    fn services(&self) -> Vec<ServiceDesc> {
        vec![
            ServiceDesc::describe::<SubscribeReq, SubscribeReply>(
                format!("{SIM_NAME}.subscribe"),
                "登记订阅(纸面:仅登记不外发)",
            ),
            ServiceDesc::describe::<SendOrderReq, SendOrderReply>(
                format!("{SIM_NAME}.send_order"),
                "下单(SUBMITTING 单上总线;撮合由 eBar./eTick. 驱动)",
            ),
            ServiceDesc::describe::<CancelOrderReq, CancelOrderReply>(
                format!("{SIM_NAME}.cancel_order"),
                "撤单(CANCELLED 单上总线)",
            ),
            ServiceDesc::describe::<QueryRateReq, QueryRateReply>(
                format!("{SIM_NAME}.query_rate"),
                "流控查询(1s 窗内已用/上限/余量)",
            ),
        ]
    }

    fn on_init(&self, ctx: &PluginContext) {
        let _ = self.ctx.set(ctx.clone());
    }

    fn on_mail(&self, event: &Event) {
        let ty = event.event_type.as_str();
        if ty == EVENT_BAR {
            if let Some(f) = event.data.downcast_ref::<BarsFrame>() {
                let outs: Vec<FillOut> = {
                    let mut m = self.lock();
                    f.bars.iter().flat_map(|b| m.on_bar(b)).collect()
                };
                self.emit_fills(&outs);
            }
        } else if ty == EVENT_TICK {
            if let Some(t) = event.data.downcast_ref::<TickData>() {
                let outs = {
                    let mut m = self.lock();
                    m.on_tick(t)
                };
                self.emit_fills(&outs);
            }
        } else if ty == e_req_type(SIM_NAME) {
            self.on_request(event);
        }
    }
}