use std::sync::OnceLock;
use std::time::Duration;
use ek_core::context::PluginContext;
use ek_core::plugin::Plugin;
use ek_core::service::{PlugNotice, ServiceDesc};
use ek_core::{EVENT_PLUG_DOWN, EVENT_PLUG_UP};
use ek_ctp::plugin::{
ConfigureReply, ConfigureReq, DisconnectReply, DisconnectReq, GatewayStatus, StatusReq,
SubscribeReply as CtpSubscribeReply, SubscribeReq as CtpSubscribeReq,
};
use ek_data::account::AccountData;
use ek_data::event::{EVENT_BAR, EVENT_CONTRACT, EVENT_LOG, EVENT_TICK};
use ek_data::log::LogData;
use ek_data::order::OrderData;
use ek_data::position::PositionData;
use ek_data::trade::TradeData;
use ek_dm::plugin::{EVENT_BAR_UPDATE, ReloadReq};
use ek_event::EVENT_TIMER;
use ek_oms::plugin::{
GetAllAccountsReq, GetAllOrdersReq, GetAllPositionsReq, GetAllTradesReq, OmsCancelOrderReq,
OmsCancelReply, PipelineStatusReq, SubmitOrderReq,
};
use ek_oms::{SubmitReply, SubmitStatus};
use ek_station::wire::{
ListReq, PlugRow, RemoveReq, RemoveRes, SVC_LIST, SVC_REMOVE, SVC_START, StartReq, StartRes,
};
use ek_cfg::wire::{
DeleteReq as CfgDeleteReq, DeleteRes as CfgDeleteRes, ListReq as CfgListReq,
ListRes as CfgListRes,
};
fn cfg_svc_ids() -> (String, String, String) {
let descs = ek_cfg::wire::cfg_services();
let id = |i: usize| -> String {
descs
.get(i)
.map_or_else(|| format!("cfg.svc[{i}]"), |d| d.id.clone())
};
(id(0), id(1), id(2))
}
fn ui_cfg_store() -> ek_cfg::Store {
match ek_cfg::Store::eco() {
Ok(s) => s,
Err(e) => {
eprintln!("ek-cfg:运维修复根 eco 失败({e}),回落仓内 ./ekcfg-local(显形口径)");
ek_cfg::Store::with_root(std::path::PathBuf::from("ekcfg-local"))
}
}
}
pub const THEME_NS: &str = "ui";
pub const THEME_KEY: &str = "theme";
#[must_use]
pub fn load_theme() -> crate::tokens::Theme {
match ui_cfg_store().load(THEME_NS) {
Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map
.get_plain(THEME_KEY)
.map(crate::tokens::Theme::from_str)
.unwrap_or_default(),
Ok(ek_cfg::LoadOutcome::Fresh) => crate::tokens::Theme::default(),
Err(e) => {
eprintln!("ek-cfg:读主题档失败({e}),按缺省深色起界面");
crate::tokens::Theme::default()
}
}
}
fn save_theme(theme: crate::tokens::Theme) -> Result<(), String> {
let map = ek_core::SectionMap::new(1).set(THEME_KEY, ek_core::Field::plain(theme.as_str()));
ui_cfg_store()
.store(THEME_NS, &map)
.map_err(|e| e.to_string())
}
use crate::state::{BusSnap, UiShared};
pub const UI_NAME: &str = "ui";
pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000;
const W_ORDER: &str = ek_oms::wire::EVENT_ORDER;
const W_TRADE: &str = ek_oms::wire::EVENT_TRADE;
const W_POSITION: &str = ek_oms::wire::EVENT_POSITION;
const W_ACCOUNT: &str = ek_oms::wire::EVENT_ACCOUNT;
pub struct UiPlugin {
shared: std::sync::Arc<UiShared>,
ctx: OnceLock<PluginContext>,
cmd_stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
}
impl UiPlugin {
pub fn new(shared: std::sync::Arc<UiShared>) -> Self {
Self {
shared,
ctx: OnceLock::new(),
cmd_stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
}
}
fn refresh(&self) {
let Some(ctx) = self.ctx.get() else {
return; };
if self.shared.list_poll_due() {
self.poll_list(ctx);
}
self.seed_status(ctx);
if self.shared.take_pipeline_due() {
self.poll_pipeline(ctx);
}
if self.shared.take_cfg_due() {
self.poll_cfg(ctx);
}
if self.shared.take_theme_persist_due() {
let theme = self.shared.theme();
match save_theme(theme) {
Ok(()) => self
.shared
.log_line(format!("[ui] 主题已存(ui.theme={})", theme.as_str())),
Err(e) => self
.shared
.log_line(format!("[ui] 主题落盘失败:{e}(本次切换只在本次运行内生效)")),
}
}
}
fn poll_cfg(&self, ctx: &PluginContext) {
let (_, svc_list, _) = cfg_svc_ids();
match ctx.ask_service::<CfgListReq, CfgListRes>(
svc_list.as_str(),
CfgListReq,
Duration::from_secs(2),
) {
Ok(res) => {
self.shared.set_cfg_briefs(res.entries);
if !res.msg.is_empty() {
self.shared
.log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
}
}
Err(e) => self
.shared
.log_line(format!("[cfg] 段落目录失败(cfg 缺席或超时): {e:?}")),
}
}
fn poll_list(&self, ctx: &PluginContext) {
match ctx.ask_service::<ListReq, Vec<PlugRow>>(SVC_LIST, ListReq, Duration::from_secs(2)) {
Ok(rows) => {
let mut services = Vec::with_capacity(rows.len());
for r in &rows {
let ids = ctx
.services_of(&r.name)
.map(|v| v.iter().map(|s| s.id.clone()).collect::<Vec<_>>())
.unwrap_or_default();
services.push((r.name.clone(), ids));
}
self.shared.sync_accounts_connected(&rows);
self.shared.note_plugin_spill(&rows);
self.shared.set_snapshot(BusSnap { rows, services });
}
Err(e) => self.shared.log_line(format!("[ui] svctl.list 失败: {e}")),
}
}
fn seed_status(&self, ctx: &PluginContext) {
for gw in self.shared.take_status_seeds() {
let svc = format!("{gw}.{}", ek_ctp::SVC_STATUS);
match ctx.ask_service::<StatusReq, GatewayStatus>(
&svc,
StatusReq,
Duration::from_secs(2),
) {
Ok(st) => self.shared.note_gateway_status(&gw, &st),
Err(e) => {
self.shared.note_status_unreachable(&gw, &format!("{e}"));
}
}
}
}
fn poll_pipeline(&self, ctx: &PluginContext) {
match ctx.ask_service::<PipelineStatusReq, Vec<String>>(
ek_oms::SVC_PIPELINE_STATUS,
PipelineStatusReq,
Duration::from_secs(2),
) {
Ok(lines) => self.shared.set_pipeline_lines(lines),
Err(e) => {
if !self.shared.pipeline_lines().is_empty() {
self.shared
.log_line(format!("[ui] oms.pipeline_status 失手: {e:?}"));
}
self.shared.set_pipeline_lines(Vec::new());
}
}
}
fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {
let orders = ctx.ask_service::<GetAllOrdersReq, Vec<OrderData>>(
"oms.get_all_orders",
GetAllOrdersReq,
Duration::from_secs(2),
);
let trades = ctx.ask_service::<GetAllTradesReq, Vec<TradeData>>(
"oms.get_all_trades",
GetAllTradesReq,
Duration::from_secs(2),
);
let positions = ctx.ask_service::<GetAllPositionsReq, Vec<PositionData>>(
"oms.get_all_positions",
GetAllPositionsReq,
Duration::from_secs(2),
);
let funds = ctx.ask_service::<GetAllAccountsReq, Vec<AccountData>>(
"oms.get_all_accounts",
GetAllAccountsReq,
Duration::from_secs(2),
);
let mut n = 0usize;
if let Ok(rows) = &orders {
for o in rows {
shared.note_order(o);
}
n += rows.len();
}
if let Ok(rows) = &trades {
for t in rows {
shared.note_trade(t);
}
n += rows.len();
}
if let Ok(rows) = &positions {
for p in rows {
shared.note_position(p);
}
n += rows.len();
}
if let Ok(rows) = &funds {
for a in rows {
shared.note_funds(a);
}
n += rows.len();
}
if n > 0 {
let bad = usize::from(orders.is_err())
+ usize::from(trades.is_err())
+ usize::from(positions.is_err())
+ usize::from(funds.is_err());
shared.log_line(format!(
"[ui] 启动播种 {n} 行(委托 {} / 成交 {} / 持仓 {} / 资金 {};失手 {} 口)",
orders.as_ref().map_or(0, |v| v.len()),
trades.as_ref().map_or(0, |v| v.len()),
positions.as_ref().map_or(0, |v| v.len()),
funds.as_ref().map_or(0, |v| v.len()),
bad
));
}
}
fn spawn_cmd_thread(&self) {
let Some(ctx) = self.ctx.get().cloned() else {
return; };
let shared = std::sync::Arc::clone(&self.shared);
let stop = std::sync::Arc::clone(&self.cmd_stop);
if std::thread::Builder::new()
.name("ui-cmd".to_string())
.spawn(move || {
loop {
if stop.load(std::sync::atomic::Ordering::Acquire) {
break;
}
if shared.cmd_pending() {
Self::run_cmds(&shared, &ctx);
continue;
}
if shared.take_trade_reseed() {
Self::reconcile_trade(&shared, &ctx);
continue;
}
std::thread::sleep(Duration::from_millis(20));
}
})
.is_err()
{
self.shared
.log_line("[ui] 命令线程起不来:操作指令会滞留在队列里");
}
}
fn run_cmds(shared: &UiShared, ctx: &PluginContext) {
for cmd in shared.drain_cmds() {
match cmd {
crate::state::UiCmd::Start { factory, name } => {
match ctx.ask_service::<StartReq, StartRes>(
SVC_START,
StartReq {
factory: factory.clone(),
name: name.clone(),
},
Duration::from_secs(3),
) {
Ok(StartRes::Started { name }) => {
shared.log_line(format!("[ops] svctl.start {factory} → {name} 上线"));
}
Ok(StartRes::Rejected { reason }) => {
shared.log_line(format!("[ops] start {factory} 被拒: {reason}"));
}
Err(e) => {
shared.log_line(format!("[ops] start {factory} 失败: {e:?}"));
}
}
shared.request_list_refresh();
}
crate::state::UiCmd::Remove { name } => {
if name == UI_NAME {
shared.log_line(
"[ops] 拒绝移除 ui:观测面自身不可自摘(摘掉后没有下一拍执行任何指令)"
.to_string(),
);
continue;
}
match ctx.ask_service::<RemoveReq, RemoveRes>(
SVC_REMOVE,
RemoveReq { name: name.clone() },
Duration::from_secs(3),
) {
Ok(RemoveRes::Removed { name }) => {
shared.log_line(format!("[ops] svctl.remove {name} 完成"));
}
Ok(RemoveRes::NoSuchName { name }) => {
shared.log_line(format!("[ops] remove {name} 不存在/不可摘"));
}
Ok(RemoveRes::Rejected { name, reason }) => {
shared.log_line(format!("[ops] remove {name} 被拒: {reason}"));
}
Err(e) => {
shared.log_line(format!("[ops] remove {name} 失败: {e:?}"));
}
}
shared.request_list_refresh();
}
crate::state::UiCmd::Reload {
interval,
symbol,
bars,
} => {
match ctx.ask_service::<ReloadReq, usize>(
"dm.reload",
ReloadReq {
interval,
symbol: symbol.clone(),
bars: bars.clone(),
size: Some(bars.len()),
push: true,
},
Duration::from_secs(5),
) {
Ok(n) => {
shared.market().note_reload();
shared.log_line(format!(
"[ops] dm.reload {symbol}@{interval}s 回灌 {n} 条"
));
}
Err(e) => {
shared.log_line(format!("[ops] dm.reload {symbol} 失败: {e:?}"));
}
}
}
crate::state::UiCmd::SendOrder { req, gateway, mode } => {
match ctx.ask_service::<SubmitOrderReq, SubmitReply>(
"oms.submit_order",
SubmitOrderReq {
req: req.clone(),
gateway_name: gateway.clone(),
mode,
},
Duration::from_secs(10),
) {
Ok(reply) => {
let tag = match reply.status {
SubmitStatus::Sent => "SENT",
SubmitStatus::Partial => "PARTIAL",
SubmitStatus::Rejected => "REJECTED",
};
for leg in &reply.legs {
match (leg.ok, &leg.en_orderid, &leg.reason) {
(true, Some(id), _) => shared.log_line(format!(
"[ops] submit {} {} {}x{} → {id} (parent {})",
leg.req.symbol,
leg.req.direction.as_str(),
leg.req.offset.as_str(),
leg.req.volume,
reply.parent_id
)),
_ => shared.log_line(format!(
"[ops] submit {} {} {}x{} 被拒: {} (parent {})",
leg.req.symbol,
leg.req.direction.as_str(),
leg.req.offset.as_str(),
leg.req.volume,
leg.reason.as_deref().unwrap_or("?"),
reply.parent_id
)),
}
}
if reply.legs.is_empty() {
shared.log_line(format!(
"[ops] submit {} {}x{} {tag}:可平不足零腿 (parent {})",
req.symbol,
req.offset.as_str(),
req.volume,
reply.parent_id
));
}
}
Err(e) => {
shared.log_line(format!("[ops] oms.submit_order 失败: {e:?}"));
}
}
}
crate::state::UiCmd::CancelOrder { orderid, gateway } => {
match ctx.ask_service::<OmsCancelOrderReq, OmsCancelReply>(
"oms.cancel_order",
OmsCancelOrderReq {
en_orderid: format!("{gateway}.{orderid}"),
gateway_name: gateway.clone(),
},
Duration::from_secs(10),
) {
Ok(OmsCancelReply { ok: true, msg }) => {
shared.log_line(format!("[ops] {msg}"));
}
Ok(OmsCancelReply { ok: false, msg }) => {
shared
.log_line(format!("[ops] {}", UiShared::cancel_reject_note(&msg)));
}
Err(e) => {
shared.log_line(format!("[ops] oms.cancel_order 失败: {e:?}"));
}
}
}
crate::state::UiCmd::CtpConfigure { gateway, form } => {
let svc = format!("{gateway}.configure");
match ctx.ask_service::<ConfigureReq, ConfigureReply>(
svc.as_str(),
form.to_configure_req(),
Duration::from_secs(5),
) {
Ok(ConfigureReply { ok: true, msg, .. }) => {
shared.log_line(format!("[ops] {svc} 受理:{msg}"));
}
Ok(ConfigureReply { ok: false, msg, .. }) => {
shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
}
Err(e) => {
shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
}
}
}
crate::state::UiCmd::Connect { gateway, form } => {
shared.note_connect_sent(&gateway);
let form = shared.prepare_connect_form(&gateway, form);
let started = match ctx.ask_service::<StartReq, StartRes>(
SVC_START,
StartReq {
factory: "ctp".to_string(),
name: gateway.clone(),
},
Duration::from_secs(3),
) {
Ok(StartRes::Started { name }) => {
shared.log_line(format!("[ops] svctl.start ctp → {name} 上线"));
true
}
Ok(StartRes::Rejected { reason }) => {
if shared.account(&gateway).is_some_and(|a| a.connected) {
shared
.log_line(format!("[ops] {gateway} 已在场,重投凭据: {reason}"));
true
} else {
shared.log_line(format!("[ops] {gateway} 装配被拒: {reason}"));
shared.note_connect_result(&gateway, Err(reason));
continue;
}
}
Err(e) => {
shared.log_line(format!("[ops] {gateway} 装配失败: {e:?}"));
shared.note_connect_result(&gateway, Err(format!("{e:?}")));
continue;
}
};
if !started {
continue;
}
let svc = format!("{gateway}.configure");
match ctx.ask_service::<ConfigureReq, ConfigureReply>(
svc.as_str(),
form.to_configure_req(),
Duration::from_secs(5),
) {
Ok(ConfigureReply {
ok: true,
msg,
product_classes,
}) => {
shared.log_line(format!("[ops] {svc} 受理:{msg}"));
shared.note_effective_product_classes(&gateway, &product_classes);
shared.note_connect_result(&gateway, Ok(()));
}
Ok(ConfigureReply { ok: false, msg, .. }) => {
shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
shared.note_connect_result(&gateway, Err(msg));
}
Err(e) => {
shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
shared.note_connect_result(&gateway, Err(format!("{e:?}")));
}
}
}
crate::state::UiCmd::CtpSubscribe { gateway, symbols } => {
let svc = format!("{gateway}.subscribe");
match ctx.ask_service::<CtpSubscribeReq, CtpSubscribeReply>(
svc.as_str(),
CtpSubscribeReq {
symbols: symbols.clone(),
},
Duration::from_secs(5),
) {
Ok(CtpSubscribeReply { ok: true, msg }) => {
shared.note_subscriptions(&symbols);
shared.log_line(format!(
"[ops] {svc} {} 受理:{}",
symbols.join(","),
msg
));
}
Ok(CtpSubscribeReply { ok: false, msg }) => {
shared.log_line(format!(
"[ops] {svc} {} 被拒:{}",
symbols.join(","),
msg
));
}
Err(e) => {
shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
}
}
}
crate::state::UiCmd::Disconnect { gateway } => {
let svc = format!("{gateway}.disconnect");
match ctx.ask_service::<DisconnectReq, DisconnectReply>(
svc.as_str(),
DisconnectReq,
Duration::from_secs(5),
) {
Ok(DisconnectReply { ok: true, msg }) => {
shared.mark_account_stopped(&gateway);
shared.log_line(format!("[账户] 断开 {gateway}:{msg}(实例仍在场)"));
}
Ok(DisconnectReply { ok: false, msg }) => {
shared.log_line(format!("[账户] 断开 {gateway} 未执行:{msg}"));
}
Err(e) => {
shared.log_line(format!(
"[账户] 断开 {gateway} 失败:{e:?}(网关不在场或超时)",
));
}
}
}
crate::state::UiCmd::CfgList => {
let (_, svc_list, _) = cfg_svc_ids();
match ctx.ask_service::<CfgListReq, CfgListRes>(
svc_list.as_str(),
CfgListReq,
Duration::from_secs(2),
) {
Ok(res) => {
shared.set_cfg_briefs(res.entries);
if !res.msg.is_empty() {
shared.log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
}
}
Err(e) => shared.log_line(format!("[cfg] 段落目录失败: {e:?}")),
}
}
crate::state::UiCmd::CfgDelete { ns } => {
let (svc_save, svc_list, svc_delete) = cfg_svc_ids();
let _ = (svc_save, svc_list);
match ctx.ask_service::<CfgDeleteReq, CfgDeleteRes>(
svc_delete.as_str(),
CfgDeleteReq {
ns: ns.clone(),
key: String::new(),
},
Duration::from_secs(2),
) {
Ok(res) if res.ok => {
shared.log_line(format!(
"[cfg] 清段 {ns}:键 {} / 秘条 {}",
res.removed_keys, res.removed_secrets
));
shared.request_cfg_refresh();
}
Ok(res) => shared.log_line(format!("[cfg] 清段 {ns} 未执行:{}", res.msg)),
Err(e) => shared.log_line(format!("[cfg] 清段 {ns} 失败: {e:?}")),
}
shared.arm_cfg_delete(None);
}
crate::state::UiCmd::CfgResave { ns } => {
let store = ui_cfg_store();
let loaded = store
.load(&ns)
.map_err(|e| e.to_string())
.and_then(|o| match o {
ek_cfg::LoadOutcome::Fresh => Err(format!("{ns} 段不存在,不持久空形")),
map @ ek_cfg::LoadOutcome::Loaded { .. } => Ok(map),
});
match loaded {
Ok(ek_cfg::LoadOutcome::Fresh) => {
unreachable!("Fresh 已被上面的 and_then 挡走");
}
Ok(ek_cfg::LoadOutcome::Loaded {
map,
warnings,
backup_used,
}) => {
if !warnings.is_empty() {
shared.log_line(format!("[cfg] {ns} 重读警告:{warnings:?}"));
}
if backup_used {
shared.log_line(format!("[cfg] {ns} 走了 .bak 回落"));
}
match store.store(&ns, &map) {
Ok(()) => {
shared.log_line(format!("[cfg] {ns} 已重存(rev {})", map.rev))
}
Err(e) => shared.log_line(format!("[cfg] {ns} 重存失败:{e}")),
}
}
Err(msg) => shared.log_line(format!("[cfg] {msg}")),
}
}
}
}
}
}
impl Plugin for UiPlugin {
fn name(&self) -> &str {
UI_NAME
}
fn deps(&self) -> &[&str] {
&[] }
fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
vec![
std::borrow::Cow::Borrowed(EVENT_LOG),
std::borrow::Cow::Borrowed(EVENT_PLUG_UP),
std::borrow::Cow::Borrowed(EVENT_PLUG_DOWN),
std::borrow::Cow::Borrowed(EVENT_BAR),
std::borrow::Cow::Borrowed(EVENT_BAR_UPDATE),
std::borrow::Cow::Borrowed(EVENT_TICK),
std::borrow::Cow::Borrowed(EVENT_CONTRACT),
std::borrow::Cow::Borrowed(ek_ctp::EVENT_CONTRACT_FINISH),
std::borrow::Cow::Borrowed(W_ORDER),
std::borrow::Cow::Borrowed(W_TRADE),
std::borrow::Cow::Borrowed(W_POSITION),
std::borrow::Cow::Borrowed(W_ACCOUNT),
std::borrow::Cow::Borrowed(EVENT_TIMER),
std::borrow::Cow::Borrowed(ek_ctp::EVENT_STATUS_PREFIX),
]
}
fn mailbox_soft_cap(&self) -> u64 {
UI_MAILBOX_SOFT_CAP
}
fn version(&self) -> std::borrow::Cow<'static, str> {
std::borrow::Cow::Borrowed(env!("CARGO_PKG_VERSION"))
}
fn services(&self) -> Vec<ServiceDesc> {
Vec::new() }
fn on_init(&self, ctx: &PluginContext) {
let _ = self.ctx.set(ctx.clone());
}
fn on_start(&self) {
self.shared.log_line("[ui] 界面插件上线");
self.spawn_cmd_thread();
self.shared.request_trade_reseed();
self.shared.request_pipeline_refresh();
let theme = load_theme();
self.shared.set_theme(theme);
if theme == crate::tokens::Theme::Light {
self.shared.log_line("[ui] 主题:浅色(读盘 ui.theme)");
}
}
fn on_stop(&self) {
self.shared.log_line("[ui] 界面插件下线");
self.cmd_stop
.store(true, std::sync::atomic::Ordering::Release);
}
fn on_mail(&self, event: &ek_event::Event) {
let ty = event.event_type.as_str();
if ty == EVENT_LOG {
if let Some(d) = event.data.downcast_ref::<LogData>() {
let tag = if d.is_warning_or_above() {
"WARN"
} else {
"INFO"
};
self.shared
.log_line(format!("[{tag}][{}] {}", d.gateway_name, d.msg));
if let Some(note) = d.msg.strip_prefix(crate::state::CHAIN_STATE_PREFIX) {
self.shared.note_chain(&d.gateway_name, note);
}
} else {
self.shared.log_line("[eLog] 载荷类型异常(非 LogData)");
}
return;
}
if ty == ek_ctp::EVENT_STATUS_PREFIX {
if let Some(ev) = event.data.downcast_ref::<ek_ctp::GatewayStatusEvent>() {
self.shared.note_status_event(ev);
} else {
self.shared
.log_line("[eStatus] 载荷类型异常(非 GatewayStatusEvent)");
}
return;
}
if ty == EVENT_PLUG_UP || ty == EVENT_PLUG_DOWN {
if let Some(n) = event.data.downcast_ref::<PlugNotice>() {
let verb = if ty == EVENT_PLUG_UP {
"上线"
} else {
"下线"
};
self.shared
.log_line(format!("[host] 插件{verb} {} v{}", n.name, n.version));
self.shared.request_list_refresh();
if ty == EVENT_PLUG_UP {
if n.name == ek_oms::NAME {
self.shared.request_pipeline_refresh();
}
let is_account = self.shared.accounts().iter().any(|a| a.gateway == n.name);
if n.name == ek_oms::NAME || is_account {
self.shared.request_trade_reseed();
}
if is_account {
self.shared.request_status_seed(&n.name);
}
let md = self.shared.md_source();
let subs = self.shared.subscriptions();
if !md.is_empty() && n.name == md && !subs.is_empty() {
self.shared.push_cmd(crate::state::UiCmd::CtpSubscribe {
gateway: n.name.clone(),
symbols: subs.clone(),
});
self.shared.log_line(format!(
"[ops] 行情源 {} 上线,重放订阅 {} 项",
n.name,
subs.len()
));
}
} else {
let removed = self.shared.clear_gateway_rows(&n.name);
if removed > 0 {
self.shared.log_line(format!(
"[ui] 插件下线 {}:作废 {} 行交易数据",
n.name, removed
));
}
if n.name == ek_oms::NAME {
self.shared.request_pipeline_refresh();
}
}
}
self.shared.bump();
return;
}
if ty == EVENT_BAR {
if let Some(f) = event.data.downcast_ref::<ek_dm::wire::BarsFrame>() {
self.shared.market().on_frame(f);
}
self.shared.bump();
return;
}
if ty == EVENT_BAR_UPDATE {
if let Some(b) = event.data.downcast_ref::<ek_data::bar::BarData>() {
self.shared.market().on_update(b);
}
self.shared.bump();
return;
}
if ty == EVENT_TICK {
if let Some(t) = event.data.downcast_ref::<ek_data::tick::TickData>() {
self.shared.update_tick(t);
}
return;
}
if ty == EVENT_CONTRACT {
if let Some(v) = event
.data
.downcast_ref::<Vec<ek_data::contract::ContractData>>()
{
for c in v {
self.shared.update_contract(c);
}
} else if let Some(c) = event.data.downcast_ref::<ek_data::contract::ContractData>() {
self.shared.update_contract(c);
}
return;
}
if ty == ek_ctp::EVENT_CONTRACT_FINISH {
let gw = event
.data
.downcast_ref::<String>()
.cloned()
.unwrap_or_else(|| "?".to_string());
self.shared
.log_line(format!("[ctp] 合约流完成 {gw}(查询合约窗可勾选订阅)"));
return;
}
if ty == W_ORDER {
match event.data.downcast_ref::<OrderData>() {
Some(o) => self.shared.note_order(o),
None => self.shared.note_bad_payload(W_ORDER),
}
return;
}
if ty == W_TRADE {
match event.data.downcast_ref::<TradeData>() {
Some(t) => self.shared.note_trade(t),
None => self.shared.note_bad_payload(W_TRADE),
}
return;
}
if ty == W_POSITION {
match event.data.downcast_ref::<PositionData>() {
Some(p) => self.shared.note_position(p),
None => self.shared.note_bad_payload(W_POSITION),
}
return;
}
if ty == W_ACCOUNT {
match event.data.downcast_ref::<AccountData>() {
Some(a) => self.shared.note_funds(a),
None => self.shared.note_bad_payload(W_ACCOUNT),
}
return;
}
if ty == EVENT_TIMER {
self.refresh();
}
}
}