1use std::sync::OnceLock;
21use std::time::Duration;
22
23use ek_core::context::PluginContext;
24use ek_core::plugin::Plugin;
25use ek_core::service::{PlugNotice, ServiceDesc};
26use ek_core::{EVENT_PLUG_DOWN, EVENT_PLUG_UP};
27use ek_ctp::plugin::{
28 ConfigureReply, ConfigureReq, DisconnectReply, DisconnectReq, GatewayStatus, StatusReq,
29 SubscribeReply as CtpSubscribeReply, SubscribeReq as CtpSubscribeReq,
30};
31use ek_data::account::AccountData;
32use ek_data::event::{EVENT_BAR, EVENT_CONTRACT, EVENT_LOG, EVENT_TICK};
33use ek_data::log::LogData;
34use ek_data::order::OrderData;
35use ek_data::position::PositionData;
36use ek_data::trade::TradeData;
37use ek_dm::plugin::{EVENT_BAR_UPDATE, ReloadReq};
38use ek_event::EVENT_TIMER;
39use ek_oms::plugin::{
40 GetAllAccountsReq, GetAllOrdersReq, GetAllPositionsReq, GetAllTradesReq, OmsCancelOrderReq,
41 OmsCancelReply, PipelineStatusReq, SubmitOrderReq,
42};
43use ek_oms::{SubmitReply, SubmitStatus};
44use ek_station::wire::{
45 ListReq, PlugRow, RemoveReq, RemoveRes, SVC_LIST, SVC_REMOVE, SVC_START, StartReq, StartRes,
46};
47
48use ek_cfg::wire::{
56 DeleteReq as CfgDeleteReq, DeleteRes as CfgDeleteRes, ListReq as CfgListReq,
57 ListRes as CfgListRes,
58};
59
60fn cfg_svc_ids() -> (String, String, String) {
63 let descs = ek_cfg::wire::cfg_services();
64 let id = |i: usize| -> String {
65 descs
66 .get(i)
67 .map_or_else(|| format!("cfg.svc[{i}]"), |d| d.id.clone())
68 };
69 (id(0), id(1), id(2))
70}
71
72pub(crate) fn ui_cfg_store() -> ek_cfg::Store {
79 match ek_cfg::Store::eco() {
80 Ok(s) => s,
81 Err(e) => {
82 eprintln!("ek-cfg:运维修复根 eco 失败({e}),回落仓内 ./ekcfg-local(显形口径)");
83 ek_cfg::Store::with_root(std::path::PathBuf::from("ekcfg-local"))
84 }
85 }
86}
87
88pub const THEME_NS: &str = "ui";
96pub const THEME_KEY: &str = "theme";
98
99#[must_use]
102pub fn load_theme() -> crate::tokens::Theme {
103 match ui_cfg_store().load(THEME_NS) {
104 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map
105 .get_plain(THEME_KEY)
106 .map(crate::tokens::Theme::from_str)
107 .unwrap_or_default(),
108 Ok(ek_cfg::LoadOutcome::Fresh) => crate::tokens::Theme::default(),
109 Err(e) => {
110 eprintln!("ek-cfg:读主题档失败({e}),按缺省深色起界面");
111 crate::tokens::Theme::default()
112 }
113 }
114}
115
116fn build_theme_map(
122 store: &ek_cfg::Store,
123 theme: crate::tokens::Theme,
124) -> Result<ek_core::SectionMap, String> {
125 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
126 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
127 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
128 };
129 map = map.set(THEME_KEY, ek_core::Field::plain(theme.as_str()));
130 Ok(map)
131}
132
133fn build_auto_connect_map(
141 store: &ek_cfg::Store,
142 flags: &[(String, bool)],
143) -> Result<ek_core::SectionMap, String> {
144 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
145 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
146 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
147 };
148 let stale: Vec<String> = map
149 .iter()
150 .filter(|(k, _)| k.starts_with("auto_connect."))
151 .map(|(k, _)| (*k).to_string())
152 .collect();
153 for k in stale {
154 map.remove(&k);
155 }
156 for (gw, on) in flags {
157 map = map.set(
158 format!("auto_connect.{gw}"),
159 ek_core::Field::plain(if *on { "1" } else { "0" }),
160 );
161 }
162 Ok(map)
163}
164
165pub fn save_auto_connect_flags_into(
168 store: &ek_cfg::Store,
169 flags: &[(String, bool)],
170) -> Result<(), String> {
171 let map = build_auto_connect_map(store, flags)?;
172 ek_cfg::cfg_save_at(store.root(), THEME_NS, &map).map_err(|e| e.to_string())
173}
174
175fn build_aliases_map(
180 store: &ek_cfg::Store,
181 aliases: &[(String, String)],
182) -> Result<ek_core::SectionMap, String> {
183 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
184 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
185 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
186 };
187 let stale: Vec<String> = map
188 .iter()
189 .filter(|(k, _)| k.starts_with("alias."))
190 .map(|(k, _)| (*k).to_string())
191 .collect();
192 for k in stale {
193 map.remove(&k);
194 }
195 for (gw, alias) in aliases {
196 let alias = alias.trim();
197 if alias.is_empty() {
198 continue;
199 }
200 map = map.set(format!("alias.{gw}"), ek_core::Field::plain(alias));
201 }
202 Ok(map)
203}
204
205pub fn save_account_aliases_into(
207 store: &ek_cfg::Store,
208 aliases: &[(String, String)],
209) -> Result<(), String> {
210 let map = build_aliases_map(store, aliases)?;
211 ek_cfg::cfg_save_at(store.root(), THEME_NS, &map).map_err(|e| e.to_string())
212}
213
214use crate::state::{BusSnap, UiShared};
215
216pub const UI_NAME: &str = "ui";
218
219pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000;
228
229const W_ORDER: &str = ek_oms::wire::EVENT_ORDER;
232const W_TRADE: &str = ek_oms::wire::EVENT_TRADE;
233const W_POSITION: &str = ek_oms::wire::EVENT_POSITION;
234const W_ACCOUNT: &str = ek_oms::wire::EVENT_ACCOUNT;
235
236pub struct UiPlugin {
238 shared: std::sync::Arc<UiShared>,
239 ctx: OnceLock<PluginContext>,
240 cmd_stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
242}
243
244impl UiPlugin {
245 pub fn new(shared: std::sync::Arc<UiShared>) -> Self {
247 Self {
248 shared,
249 ctx: OnceLock::new(),
250 cmd_stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
251 }
252 }
253
254 #[must_use]
257 pub fn ctx_handle(&self) -> Option<PluginContext> {
258 self.ctx.get().cloned()
259 }
260
261 fn refresh(&self) {
264 let Some(ctx) = self.ctx.get() else {
265 return; };
267 if self.shared.list_poll_due() {
272 self.poll_list(ctx);
273 }
274 self.seed_status(ctx);
275 if self.shared.take_pipeline_due() {
276 self.poll_pipeline(ctx);
277 }
278 if self.shared.take_cfg_due() {
279 self.poll_cfg(ctx);
282 }
283 self.flush_persist();
284 }
285
286 fn cfg_bus_save(&self, ns: &str, map: &ek_core::SectionMap) -> Result<(), String> {
290 if let Some(ctx) = self.ctx.get() {
291 match ctx.ask_service::<ek_cfg::wire::SaveReq, ek_cfg::wire::SaveRes>(
292 "cfg.save",
293 ek_cfg::wire::SaveReq {
294 ns: ns.to_string(),
295 map: map.clone(),
296 },
297 Duration::from_millis(1500),
298 ) {
299 Ok(r) if r.ok => return Ok(()),
300 Ok(r) => return Err(r.msg),
301 Err(ek_core::error::AskError::UnknownService { .. }) => {
302 self.shared
303 .log_line(format!("[cfg] cfg 缺席,{ns} 直写托底(软集中负形)"));
304 }
305 Err(e) => return Err(format!("{e:?}")),
306 }
307 }
308 ui_cfg_store().save(ns, map).map_err(|e| e.to_string())
309 }
310
311 fn cfg_bus_delete(&self, ns: &str) -> Result<(), String> {
314 if let Some(ctx) = self.ctx.get() {
315 match ctx.ask_service::<ek_cfg::wire::DeleteReq, ek_cfg::wire::DeleteRes>(
316 "cfg.delete",
317 ek_cfg::wire::DeleteReq {
318 ns: ns.to_string(),
319 key: String::new(),
320 },
321 Duration::from_millis(1500),
322 ) {
323 Ok(r) if r.ok => return Ok(()),
324 Ok(r) => return Err(r.msg),
325 Err(ek_core::error::AskError::UnknownService { .. }) => {
326 self.shared
327 .log_line(format!("[cfg] cfg 缺席,{ns} 直清托底(软集中负形)"));
328 }
329 Err(e) => return Err(format!("{e:?}")),
330 }
331 }
332 crate::appconfig::purge_ns_direct(&ui_cfg_store(), ns)
333 }
334
335 fn flush_persist(&self) {
341 if self.shared.take_theme_persist_due() {
342 let theme = self.shared.theme();
345 let wr = build_theme_map(&ui_cfg_store(), theme)
346 .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
347 match wr {
348 Ok(()) => self
349 .shared
350 .log_line(format!("[ui] 主题已存(ui.theme={})", theme.as_str())),
351 Err(e) => self
352 .shared
353 .log_line(format!("[ui] 主题落盘失败:{e}(本次切换只在本次运行内生效)")),
354 }
355 }
356 let purges = self.shared.take_archive_purges();
357 for gw in purges {
358 match self.cfg_bus_delete(&format!("ctp.{gw}")) {
363 Ok(()) => self.shared.log_line(format!(
364 "[账户] 档案段 ctp.{gw} 已抹(删除已落盘:下次启动清点不再复活)"
365 )),
366 Err(e) => self.shared.log_line(format!(
367 "[账户] 档案 ctp.{gw} 抹除失败:{e}(本次删除只在今次运行生效,重启会复活)"
368 )),
369 }
370 }
371 if self.shared.take_auto_connect_persist_due() {
372 let flags: Vec<(String, bool)> = self
375 .shared
376 .accounts()
377 .iter()
378 .map(|a| (a.gateway.clone(), a.auto_connect))
379 .collect();
380 let wr = build_auto_connect_map(&ui_cfg_store(), &flags)
382 .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
383 match wr {
384 Ok(()) => {
385 let on: Vec<&str> = flags
386 .iter()
387 .filter(|(_, on)| *on)
388 .map(|(gw, _)| gw.as_str())
389 .collect();
390 if on.is_empty() {
391 self.shared
392 .log_line("[ui] 自动连接勾选已存(ui 段;当前全部关)".to_string());
393 } else {
394 self.shared.log_line(format!(
395 "[ui] 自动连接勾选已存(ui 段;开:{};守护/桌面启动清点按它起连)",
396 on.join(", ")
397 ));
398 }
399 }
400 Err(e) => self.shared.log_line(format!(
401 "[ui] 自动连接勾选落盘失败:{e}(本轮勾选只在本次运行内生效)"
402 )),
403 }
404 }
405 if self.shared.take_alias_persist_due() {
406 let aliases: Vec<(String, String)> = self
408 .shared
409 .accounts()
410 .iter()
411 .map(|a| (a.gateway.clone(), a.alias.clone()))
412 .collect();
413 let wr = build_aliases_map(&ui_cfg_store(), &aliases)
415 .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
416 match wr {
417 Ok(()) => {
418 let named: Vec<String> = aliases
419 .iter()
420 .filter(|(_, al)| !al.trim().is_empty())
421 .map(|(gw, al)| format!("{al}({gw})"))
422 .collect();
423 if named.is_empty() {
424 self.shared
425 .log_line("[ui] 账户别名已存(ui 段;当前全部无别名)".to_string());
426 } else {
427 self.shared.log_line(format!(
428 "[ui] 账户别名已存(ui 段;{};启动清点按它贴标签)",
429 named.join(", ")
430 ));
431 }
432 }
433 Err(e) => self.shared.log_line(format!(
434 "[ui] 账户别名落盘失败:{e}(本次改名只在本次运行内生效)"
435 )),
436 }
437 }
438 }
439
440 fn poll_cfg(&self, ctx: &PluginContext) {
442 let (_, svc_list, _) = cfg_svc_ids();
443 match ctx.ask_service::<CfgListReq, CfgListRes>(
444 svc_list.as_str(),
445 CfgListReq,
446 Duration::from_secs(2),
447 ) {
448 Ok(res) => {
449 self.shared.set_cfg_briefs(res.entries);
450 if !res.msg.is_empty() {
451 self.shared
452 .log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
453 }
454 }
455 Err(e) => self
456 .shared
457 .log_line(format!("[cfg] 段落目录失败(cfg 缺席或超时): {e:?}")),
458 }
459 }
460
461 fn poll_list(&self, ctx: &PluginContext) {
467 match ctx.ask_service::<ListReq, Vec<PlugRow>>(SVC_LIST, ListReq, Duration::from_secs(2)) {
468 Ok(rows) => {
469 let mut services = Vec::with_capacity(rows.len());
470 for r in &rows {
471 let ids = ctx
472 .services_of(&r.name)
473 .map(|v| v.iter().map(|s| s.id.clone()).collect::<Vec<_>>())
474 .unwrap_or_default();
475 services.push((r.name.clone(), ids));
476 }
477 self.shared.sync_accounts_connected(&rows);
480 self.shared.note_plugin_spill(&rows);
483 self.shared.set_snapshot(BusSnap { rows, services });
484 }
485 Err(e) => self.shared.log_line(format!("[ui] svctl.list 失败: {e}")),
486 }
487 }
488
489 fn seed_status(&self, ctx: &PluginContext) {
496 for gw in self.shared.take_status_seeds() {
497 let svc = format!("{gw}.{}", ek_ctp::SVC_STATUS);
498 match ctx.ask_service::<StatusReq, GatewayStatus>(
499 &svc,
500 StatusReq,
501 Duration::from_secs(2),
502 ) {
503 Ok(st) => self.shared.note_gateway_status(&gw, &st),
504 Err(e) => {
505 self.shared.note_status_unreachable(&gw, &format!("{e}"));
509 }
510 }
511 }
512 }
513
514 fn poll_pipeline(&self, ctx: &PluginContext) {
519 match ctx.ask_service::<PipelineStatusReq, Vec<String>>(
520 ek_oms::SVC_PIPELINE_STATUS,
521 PipelineStatusReq,
522 Duration::from_secs(2),
523 ) {
524 Ok(lines) => self.shared.set_pipeline_lines(lines),
525 Err(e) => {
526 if !self.shared.pipeline_lines().is_empty() {
530 self.shared
531 .log_line(format!("[ui] oms.pipeline_status 失手: {e:?}"));
532 }
533 self.shared.set_pipeline_lines(Vec::new());
534 }
535 }
536 }
537
538 fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {
548 let orders = ctx.ask_service::<GetAllOrdersReq, Vec<OrderData>>(
549 "oms.get_all_orders",
550 GetAllOrdersReq,
551 Duration::from_secs(2),
552 );
553 let trades = ctx.ask_service::<GetAllTradesReq, Vec<TradeData>>(
554 "oms.get_all_trades",
555 GetAllTradesReq,
556 Duration::from_secs(2),
557 );
558 let positions = ctx.ask_service::<GetAllPositionsReq, Vec<PositionData>>(
559 "oms.get_all_positions",
560 GetAllPositionsReq,
561 Duration::from_secs(2),
562 );
563 let funds = ctx.ask_service::<GetAllAccountsReq, Vec<AccountData>>(
564 "oms.get_all_accounts",
565 GetAllAccountsReq,
566 Duration::from_secs(2),
567 );
568 let mut n = 0usize;
569 if let Ok(rows) = &orders {
570 for o in rows {
571 shared.note_order(o);
572 }
573 n += rows.len();
574 }
575 if let Ok(rows) = &trades {
576 for t in rows {
577 shared.note_trade(t);
578 }
579 n += rows.len();
580 }
581 if let Ok(rows) = &positions {
582 for p in rows {
583 shared.note_position(p);
584 }
585 n += rows.len();
586 }
587 if let Ok(rows) = &funds {
588 for a in rows {
589 shared.note_funds(a);
590 }
591 n += rows.len();
592 }
593 if n > 0 {
595 let bad = usize::from(orders.is_err())
596 + usize::from(trades.is_err())
597 + usize::from(positions.is_err())
598 + usize::from(funds.is_err());
599 shared.log_line(format!(
600 "[ui] 启动播种 {n} 行(委托 {} / 成交 {} / 持仓 {} / 资金 {};失手 {} 口)",
601 orders.as_ref().map_or(0, |v| v.len()),
602 trades.as_ref().map_or(0, |v| v.len()),
603 positions.as_ref().map_or(0, |v| v.len()),
604 funds.as_ref().map_or(0, |v| v.len()),
605 bad
606 ));
607 }
608 }
609
610 fn spawn_cmd_thread(&self) {
618 let Some(ctx) = self.ctx.get().cloned() else {
619 return; };
621 let shared = std::sync::Arc::clone(&self.shared);
622 let stop = std::sync::Arc::clone(&self.cmd_stop);
623 if std::thread::Builder::new()
624 .name("ui-cmd".to_string())
625 .spawn(move || {
626 loop {
627 if stop.load(std::sync::atomic::Ordering::Acquire) {
628 break;
629 }
630 if shared.cmd_pending() {
632 Self::run_cmds(&shared, &ctx);
633 continue;
634 }
635 if shared.take_trade_reseed() {
637 Self::reconcile_trade(&shared, &ctx);
638 continue;
639 }
640 if shared.kline_fetch_pending() {
642 Self::run_kline_fetches(&shared, &ctx);
643 continue;
644 }
645 std::thread::sleep(Duration::from_millis(20));
646 }
647 })
648 .is_err()
649 {
650 self.shared
652 .log_line("[ui] 命令线程起不来:操作指令会滞留在队列里");
653 }
654 }
655
656 fn run_kline_fetches(shared: &UiShared, ctx: &PluginContext) {
660 for f in shared.take_kline_fetches() {
661 let Some(gw) = ctx
662 .plugins()
663 .iter()
664 .map(|p| p.name.clone())
665 .find(|n| n.starts_with("tq"))
666 else {
667 shared.log_line(format!(
668 "[kline] {} tq 网关不在线,历史补数丢弃(重开 K线窗再试)",
669 f.symbol
670 ));
671 continue;
672 };
673 let req = ek_tq::KlinesReq {
674 symbol: f.symbol.clone(),
675 raw: false,
676 interval_secs: f.interval_secs,
677 range: ek_tq::RangeSpec::Length {
678 count: 300,
679 end_ns: None,
680 },
681 max_rows: Some(300),
682 };
683 match ctx.ask::<ek_tq::KlinesReq, ek_tq::KlinesReply>(&gw, req, Duration::from_secs(45))
684 {
685 Ok(reply) => {
686 if let Some(err) = &reply.error {
687 shared.log_line(format!(
688 "[kline] {}/{}s 历史显形: {}",
689 f.symbol, f.interval_secs, err.message
690 ));
691 continue;
692 }
693 let n = reply.rows.len();
694 shared
695 .market()
696 .note_history(&f.symbol, f.interval_secs, &reply.rows);
697 shared.bump();
698 shared.log_line(format!(
699 "[kline] {}/{}s 历史补数 +{n} 根(mode={} truncated={})",
700 f.symbol,
701 f.interval_secs,
702 reply.mode.as_str(),
703 reply.truncated
704 ));
705 }
706 Err(e) => shared.log_line(format!(
707 "[kline] {}/{}s 历史 ask 失手: {e:?}",
708 f.symbol, f.interval_secs
709 )),
710 }
711 }
712 }
713
714 fn run_cmds(shared: &UiShared, ctx: &PluginContext) {
716 for cmd in shared.drain_cmds() {
717 match cmd {
718 crate::state::UiCmd::Start { factory, name } => {
719 match ctx.ask_service::<StartReq, StartRes>(
720 SVC_START,
721 StartReq {
722 factory: factory.clone(),
723 name: name.clone(),
724 },
725 Duration::from_secs(3),
726 ) {
727 Ok(StartRes::Started { name }) => {
728 shared.log_line(format!("[ops] svctl.start {factory} → {name} 上线"));
729 }
730 Ok(StartRes::Rejected { reason }) => {
731 shared.log_line(format!("[ops] start {factory} 被拒: {reason}"));
732 }
733 Err(e) => {
734 shared.log_line(format!("[ops] start {factory} 失败: {e:?}"));
735 }
736 }
737 shared.request_list_refresh();
739 shared.remount_modules();
742 }
743 crate::state::UiCmd::Remove { name } => {
744 if name == UI_NAME {
752 shared.log_line(
753 "[ops] 拒绝移除 ui:观测面自身不可自摘(摘掉后没有下一拍执行任何指令)"
754 .to_string(),
755 );
756 continue;
757 }
758 match ctx.ask_service::<RemoveReq, RemoveRes>(
759 SVC_REMOVE,
760 RemoveReq { name: name.clone() },
761 Duration::from_secs(3),
762 ) {
763 Ok(RemoveRes::Removed { name }) => {
764 shared.log_line(format!("[ops] svctl.remove {name} 完成"));
765 }
766 Ok(RemoveRes::NoSuchName { name }) => {
767 shared.log_line(format!("[ops] remove {name} 不存在/不可摘"));
768 }
769 Ok(RemoveRes::Rejected { name, reason }) => {
773 shared.log_line(format!("[ops] remove {name} 被拒: {reason}"));
774 }
775 Err(e) => {
776 shared.log_line(format!("[ops] remove {name} 失败: {e:?}"));
777 }
778 }
779 shared.request_list_refresh();
780 shared.remount_modules();
782 }
783 crate::state::UiCmd::Reload {
784 interval,
785 symbol,
786 bars,
787 } => {
788 match ctx.ask_service::<ReloadReq, usize>(
790 "dm.reload",
791 ReloadReq {
792 interval,
793 symbol: symbol.clone(),
794 bars: bars.clone(),
795 size: Some(bars.len()),
796 push: true,
797 },
798 Duration::from_secs(5),
799 ) {
800 Ok(n) => {
801 shared.market().note_reload();
802 shared.log_line(format!(
803 "[ops] dm.reload {symbol}@{interval}s 回灌 {n} 条"
804 ));
805 }
806 Err(e) => {
807 shared.log_line(format!("[ops] dm.reload {symbol} 失败: {e:?}"));
808 }
809 }
810 }
811 crate::state::UiCmd::SendOrder { req, gateway, mode } => {
812 match ctx.ask_service::<SubmitOrderReq, SubmitReply>(
816 "oms.submit_order",
817 SubmitOrderReq {
818 req: req.clone(),
819 gateway_name: gateway.clone(),
820 mode,
821 },
822 Duration::from_secs(10),
823 ) {
824 Ok(reply) => {
825 let tag = match reply.status {
826 SubmitStatus::Sent => "SENT",
827 SubmitStatus::Partial => "PARTIAL",
828 SubmitStatus::Rejected => "REJECTED",
829 };
830 for leg in &reply.legs {
831 match (leg.ok, &leg.en_orderid, &leg.reason) {
832 (true, Some(id), _) => shared.log_line(format!(
833 "[ops] submit {} {} {}x{} → {id} (parent {})",
834 leg.req.symbol,
835 leg.req.direction.as_str(),
836 leg.req.offset.as_str(),
837 leg.req.volume,
838 reply.parent_id
839 )),
840 _ => shared.log_line(format!(
841 "[ops] submit {} {} {}x{} 被拒: {} (parent {})",
842 leg.req.symbol,
843 leg.req.direction.as_str(),
844 leg.req.offset.as_str(),
845 leg.req.volume,
846 leg.reason.as_deref().unwrap_or("?"),
847 reply.parent_id
848 )),
849 }
850 }
851 if reply.legs.is_empty() {
852 shared.log_line(format!(
853 "[ops] submit {} {}x{} {tag}:可平不足零腿 (parent {})",
854 req.symbol,
855 req.offset.as_str(),
856 req.volume,
857 reply.parent_id
858 ));
859 }
860 }
861 Err(e) => {
862 shared.log_line(format!("[ops] oms.submit_order 失败: {e:?}"));
863 }
864 }
865 }
866 crate::state::UiCmd::CancelOrder { orderid, gateway } => {
867 match ctx.ask_service::<OmsCancelOrderReq, OmsCancelReply>(
871 "oms.cancel_order",
872 OmsCancelOrderReq {
873 en_orderid: format!("{gateway}.{orderid}"),
874 gateway_name: gateway.clone(),
875 },
876 Duration::from_secs(10),
877 ) {
878 Ok(OmsCancelReply { ok: true, msg }) => {
879 shared.log_line(format!("[ops] {msg}"));
880 }
881 Ok(OmsCancelReply { ok: false, msg }) => {
882 shared
884 .log_line(format!("[ops] {}", UiShared::cancel_reject_note(&msg)));
885 }
886 Err(e) => {
887 shared.log_line(format!("[ops] oms.cancel_order 失败: {e:?}"));
888 }
889 }
890 }
891 crate::state::UiCmd::CtpConfigure { gateway, form } => {
892 let svc = format!("{gateway}.configure");
895 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
896 svc.as_str(),
897 form.to_configure_req(),
898 Duration::from_secs(5),
899 ) {
900 Ok(ConfigureReply { ok: true, msg, .. }) => {
904 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
905 }
906 Ok(ConfigureReply { ok: false, msg, .. }) => {
907 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
908 }
909 Err(e) => {
910 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
911 }
912 }
913 }
914 crate::state::UiCmd::Connect { gateway, form } => {
915 shared.note_connect_sent(&gateway);
920 let form = shared.prepare_connect_form(&gateway, form);
923 let started = match ctx.ask_service::<StartReq, StartRes>(
924 SVC_START,
925 StartReq {
926 factory: "ctp".to_string(),
927 name: gateway.clone(),
928 },
929 Duration::from_secs(3),
930 ) {
931 Ok(StartRes::Started { name }) => {
932 shared.log_line(format!("[ops] svctl.start ctp → {name} 上线"));
933 true
934 }
935 Ok(StartRes::Rejected { reason }) => {
936 if shared.account(&gateway).is_some_and(|a| a.connected) {
939 shared
940 .log_line(format!("[ops] {gateway} 已在场,重投凭据: {reason}"));
941 true
942 } else {
943 shared.log_line(format!("[ops] {gateway} 装配被拒: {reason}"));
944 shared.note_connect_result(&gateway, Err(reason));
945 continue;
946 }
947 }
948 Err(e) => {
949 shared.log_line(format!("[ops] {gateway} 装配失败: {e:?}"));
950 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
951 continue;
952 }
953 };
954 if !started {
955 continue;
956 }
957 let svc = format!("{gateway}.configure");
958 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
959 svc.as_str(),
960 form.to_configure_req(),
961 Duration::from_secs(5),
962 ) {
963 Ok(ConfigureReply {
967 ok: true,
968 msg,
969 product_classes,
970 }) => {
971 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
972 shared.note_effective_product_classes(&gateway, &product_classes);
973 shared.note_connect_result(&gateway, Ok(()));
974 }
975 Ok(ConfigureReply { ok: false, msg, .. }) => {
976 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
977 shared.note_connect_result(&gateway, Err(msg));
978 }
979 Err(e) => {
980 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
981 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
982 }
983 }
984 }
985 crate::state::UiCmd::CtpSubscribe { gateway, symbols } => {
986 let svc = format!("{gateway}.subscribe");
990 match ctx.ask_service::<CtpSubscribeReq, CtpSubscribeReply>(
991 svc.as_str(),
992 CtpSubscribeReq {
993 symbols: symbols.clone(),
994 },
995 Duration::from_secs(5),
996 ) {
997 Ok(CtpSubscribeReply { ok: true, msg }) => {
998 shared.note_subscriptions(&symbols);
999 shared.log_line(format!(
1000 "[ops] {svc} {} 受理:{}",
1001 symbols.join(","),
1002 msg
1003 ));
1004 }
1005 Ok(CtpSubscribeReply { ok: false, msg }) => {
1006 shared.log_line(format!(
1007 "[ops] {svc} {} 被拒:{}",
1008 symbols.join(","),
1009 msg
1010 ));
1011 }
1012 Err(e) => {
1013 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
1014 }
1015 }
1016 }
1017 crate::state::UiCmd::Disconnect { gateway } => {
1018 let svc = format!("{gateway}.disconnect");
1022 match ctx.ask_service::<DisconnectReq, DisconnectReply>(
1023 svc.as_str(),
1024 DisconnectReq,
1025 Duration::from_secs(5),
1026 ) {
1027 Ok(DisconnectReply { ok: true, msg }) => {
1028 shared.mark_account_stopped(&gateway);
1029 shared.log_line(format!("[账户] 断开 {gateway}:{msg}(实例仍在场)"));
1030 }
1031 Ok(DisconnectReply { ok: false, msg }) => {
1032 shared.log_line(format!("[账户] 断开 {gateway} 未执行:{msg}"));
1033 }
1034 Err(e) => {
1035 shared.log_line(format!(
1036 "[账户] 断开 {gateway} 失败:{e:?}(网关不在场或超时)",
1037 ));
1038 }
1039 }
1040 }
1041 crate::state::UiCmd::CfgList => {
1042 let (_, svc_list, _) = cfg_svc_ids();
1044 match ctx.ask_service::<CfgListReq, CfgListRes>(
1045 svc_list.as_str(),
1046 CfgListReq,
1047 Duration::from_secs(2),
1048 ) {
1049 Ok(res) => {
1050 shared.set_cfg_briefs(res.entries);
1051 if !res.msg.is_empty() {
1052 shared.log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
1053 }
1054 }
1055 Err(e) => shared.log_line(format!("[cfg] 段落目录失败: {e:?}")),
1056 }
1057 }
1058 crate::state::UiCmd::CfgDelete { ns } => {
1059 let (svc_save, svc_list, svc_delete) = cfg_svc_ids();
1061 let _ = (svc_save, svc_list);
1062 match ctx.ask_service::<CfgDeleteReq, CfgDeleteRes>(
1063 svc_delete.as_str(),
1064 CfgDeleteReq {
1065 ns: ns.clone(),
1066 key: String::new(),
1067 },
1068 Duration::from_secs(2),
1069 ) {
1070 Ok(res) if res.ok => {
1071 shared.log_line(format!(
1072 "[cfg] 清段 {ns}:键 {} / 秘条 {}",
1073 res.removed_keys, res.removed_secrets
1074 ));
1075 shared.request_cfg_refresh();
1076 }
1077 Ok(res) => shared.log_line(format!("[cfg] 清段 {ns} 未执行:{}", res.msg)),
1078 Err(e) => shared.log_line(format!("[cfg] 清段 {ns} 失败: {e:?}")),
1079 }
1080 shared.arm_cfg_delete(None);
1081 }
1082 crate::state::UiCmd::CfgResave { ns } => {
1083 let store = ui_cfg_store();
1088 let loaded = store
1089 .load(&ns)
1090 .map_err(|e| e.to_string())
1091 .and_then(|o| match o {
1092 ek_cfg::LoadOutcome::Fresh => Err(format!("{ns} 段不存在,不持久空形")),
1093 map @ ek_cfg::LoadOutcome::Loaded { .. } => Ok(map),
1094 });
1095 match loaded {
1096 Ok(ek_cfg::LoadOutcome::Fresh) => {
1097 unreachable!("Fresh 已被上面的 and_then 挡走");
1098 }
1099 Ok(ek_cfg::LoadOutcome::Loaded { map, warnings }) => {
1100 if !warnings.is_empty() {
1101 shared.log_line(format!("[cfg] {ns} 重读警告:{warnings:?}"));
1102 }
1103 let wr = match ctx
1106 .ask_service::<ek_cfg::wire::SaveReq, ek_cfg::wire::SaveRes>(
1107 "cfg.save",
1108 ek_cfg::wire::SaveReq {
1109 ns: ns.clone(),
1110 map: map.clone(),
1111 },
1112 Duration::from_millis(1500),
1113 ) {
1114 Ok(r) if r.ok => Ok(()),
1115 Ok(r) => Err(r.msg),
1116 Err(ek_core::error::AskError::UnknownService { .. }) => {
1117 shared.log_line(format!(
1118 "[cfg] cfg 缺席,{ns} 直写托底(软集中负形)"
1119 ));
1120 ek_cfg::cfg_save_at(store.root(), &ns, &map)
1121 .map_err(|e| e.to_string())
1122 }
1123 Err(e) => Err(format!("{e:?}")),
1124 };
1125 match wr {
1126 Ok(()) => {
1127 shared.log_line(format!("[cfg] {ns} 已重存(rev {})", map.rev))
1128 }
1129 Err(e) => shared.log_line(format!("[cfg] {ns} 重存失败:{e}")),
1130 }
1131 }
1132 Err(msg) => shared.log_line(format!("[cfg] {msg}")),
1133 }
1134 }
1135 crate::state::UiCmd::StratAdd {
1136 class_name,
1137 name,
1138 params,
1139 } => {
1140 let r = ask_strategy::<
1141 ek_strategy::contract::AddReq,
1142 ek_strategy::contract::Reply,
1143 >(
1144 ctx,
1145 "strategy.add",
1146 ek_strategy::contract::AddReq {
1147 class_name,
1148 name: name.clone(),
1149 params,
1150 },
1151 );
1152 let (ok, msg) = match &r {
1154 Ok(rep) => (rep.ok, rep.msg.clone()),
1155 Err(e) => (false, e.clone()),
1156 };
1157 shared.set_strategy_feedback(crate::state::StratFeedback {
1158 op: "add",
1159 name: name.clone(),
1160 ok,
1161 msg,
1162 });
1163 report_strategy(shared, "添加", r);
1164 }
1165 crate::state::UiCmd::StratInit { name } => {
1166 let r = ask_strategy::<
1167 ek_strategy::contract::InitReq,
1168 ek_strategy::contract::Reply,
1169 >(
1170 ctx,
1171 "strategy.init",
1172 ek_strategy::contract::InitReq { name },
1173 );
1174 report_strategy(shared, "初始化", r);
1175 }
1176 crate::state::UiCmd::StratStart { name } => {
1177 let r = ask_strategy::<
1178 ek_strategy::contract::StartReq,
1179 ek_strategy::contract::Reply,
1180 >(
1181 ctx,
1182 "strategy.start",
1183 ek_strategy::contract::StartReq { name },
1184 );
1185 report_strategy(shared, "启动", r);
1186 }
1187 crate::state::UiCmd::StratPause { name } => {
1188 let r = ask_strategy::<
1189 ek_strategy::contract::PauseReq,
1190 ek_strategy::contract::Reply,
1191 >(
1192 ctx,
1193 "strategy.pause",
1194 ek_strategy::contract::PauseReq { name },
1195 );
1196 report_strategy(shared, "暂停", r);
1197 }
1198 crate::state::UiCmd::StratResume { name } => {
1199 let r = ask_strategy::<
1200 ek_strategy::contract::ResumeReq,
1201 ek_strategy::contract::Reply,
1202 >(
1203 ctx,
1204 "strategy.resume",
1205 ek_strategy::contract::ResumeReq { name },
1206 );
1207 report_strategy(shared, "恢复", r);
1208 }
1209 crate::state::UiCmd::StratStop { name } => {
1210 let r = ask_strategy::<
1211 ek_strategy::contract::StopReq,
1212 ek_strategy::contract::Reply,
1213 >(
1214 ctx,
1215 "strategy.stop",
1216 ek_strategy::contract::StopReq { name },
1217 );
1218 report_strategy(shared, "停止", r);
1219 }
1220 crate::state::UiCmd::StratEdit { name, params } => {
1221 let r = ask_strategy::<
1222 ek_strategy::contract::EditReq,
1223 ek_strategy::contract::Reply,
1224 >(
1225 ctx,
1226 "strategy.edit",
1227 ek_strategy::contract::EditReq {
1228 name: name.clone(),
1229 params,
1230 },
1231 );
1232 let (ok, msg) = match &r {
1233 Ok(rep) => (rep.ok, rep.msg.clone()),
1234 Err(e) => (false, e.clone()),
1235 };
1236 shared.set_strategy_feedback(crate::state::StratFeedback {
1237 op: "edit",
1238 name: name.clone(),
1239 ok,
1240 msg,
1241 });
1242 report_strategy(shared, "改参数", r);
1243 }
1244 crate::state::UiCmd::StratRemove { name } => {
1245 let r = ask_strategy::<
1246 ek_strategy::contract::RemoveReq,
1247 ek_strategy::contract::Reply,
1248 >(
1249 ctx,
1250 "strategy.remove",
1251 ek_strategy::contract::RemoveReq { name },
1252 );
1253 report_strategy(shared, "移除", r);
1254 }
1255 crate::state::UiCmd::StratStartAll => {
1256 let r = ask_strategy::<(), ek_strategy::contract::Reply>(
1257 ctx,
1258 "strategy.start_all",
1259 (),
1260 );
1261 report_strategy(shared, "全部启动", r);
1262 }
1263 crate::state::UiCmd::StratStopAll => {
1264 let r = ask_strategy::<(), ek_strategy::contract::Reply>(
1265 ctx,
1266 "strategy.stop_all",
1267 (),
1268 );
1269 report_strategy(shared, "全部停止", r);
1270 }
1271 crate::state::UiCmd::StratResync => {
1272 let r = ask_strategy::<(), ek_strategy::contract::Reply>(
1273 ctx,
1274 "strategy.resync",
1275 (),
1276 );
1277 match r {
1278 Ok(rep) if rep.ok => shared.log_line("[strategy] 已重推全量快照"),
1279 Ok(rep) => shared.log_line(format!("[strategy] {}", rep.msg)),
1280 Err(e) => shared.log_line(format!("[strategy] 重推失败:{e}")),
1281 }
1282 }
1283 crate::state::UiCmd::StratSettings { name } => {
1284 match ask_strategy::<
1285 ek_strategy::contract::NameReq,
1286 ek_strategy::contract::SettingsRes,
1287 >(
1288 ctx,
1289 "strategy.get_settings",
1290 ek_strategy::contract::NameReq { name: name.clone() },
1291 ) {
1292 Ok(res) => shared.set_strategy_settings(&name, res.params),
1293 Err(e) => shared.log_line(format!("[strategy] 取参数失败:{e}")),
1294 }
1295 }
1296 crate::state::UiCmd::StratPrice { symbol } => {
1297 match ask_strategy::<
1298 ek_strategy::contract::PriceSymbolReq,
1299 ek_strategy::contract::PriceRes,
1300 >(
1301 ctx,
1302 "strategy.get_price",
1303 ek_strategy::contract::PriceSymbolReq {
1304 symbol: symbol.clone(),
1305 },
1306 ) {
1307 Ok(res) => {
1308 if let Some(p) = res.price {
1309 shared.set_strategy_price(&symbol, p);
1310 }
1311 }
1312 Err(e) => shared.log_line(format!("[strategy] 取最新价失败:{e}")),
1313 }
1314 }
1315 crate::state::UiCmd::StratMetaSync => {
1316 if strategy_meta_due(shared) {
1318 STRAT_META_ASKS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1319 match ask_strategy::<(), Vec<ek_strategy::strategy::StrategyMeta>>(
1320 ctx,
1321 "strategy.meta",
1322 (),
1323 ) {
1324 Ok(metas) => shared.set_strategy_metas(metas),
1325 Err(e) => shared.log_line(format!("[strategy] 取策略类失败:{e}")),
1326 }
1327 }
1328 }
1329 crate::state::UiCmd::StratSetEngine {
1330 trade_gateway,
1331 md_gateway,
1332 } => {
1333 let res = ask_strategy::<
1334 ek_strategy::contract::SetEngineReq,
1335 ek_strategy::contract::Reply,
1336 >(
1337 ctx,
1338 "strategy.set_engine",
1339 ek_strategy::contract::SetEngineReq {
1340 trade_gateway: trade_gateway.clone(),
1341 md_gateway: md_gateway.clone(),
1342 },
1343 );
1344 let (ok, msg) = match res {
1345 Ok(r) => (r.ok, r.msg),
1346 Err(e) => (false, format!("网关设置失败:{e}")),
1347 };
1348 if !ok {
1349 shared.log_line(format!("[strategy] {msg}"));
1350 }
1351 shared.set_strategy_feedback(crate::state::StratFeedback {
1352 op: "engine",
1353 name: String::new(),
1354 ok,
1355 msg,
1356 });
1357 }
1358 }
1359 }
1360 }
1361}
1362
1363#[doc(hidden)]
1365#[must_use]
1366pub fn strategy_meta_due(shared: &UiShared) -> bool {
1367 shared.strategy_metas().is_empty()
1368}
1369
1370#[doc(hidden)]
1372pub static STRAT_META_ASKS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
1373
1374fn ask_strategy<Req, Res>(
1376 ctx: &PluginContext,
1377 service: &'static str,
1378 req: Req,
1379) -> Result<Res, String>
1380where
1381 Req: std::any::Any + Send + Sync,
1382 Res: std::any::Any + Send + Sync,
1383{
1384 ctx.ask_service::<Req, Res>(service, req, Duration::from_millis(2000))
1385 .map_err(|e| match e {
1386 ek_core::error::AskError::UnknownService { .. } => "策略插件不在场".to_string(),
1387 other => other.to_string(),
1388 })
1389}
1390
1391fn report_strategy(shared: &UiShared, verb: &str, r: Result<ek_strategy::contract::Reply, String>) {
1393 match r {
1394 Ok(rep) if rep.ok => {
1395 if !rep.msg.is_empty() {
1396 shared.log_line(format!("[strategy] {verb}:{}", rep.msg));
1397 }
1398 }
1399 Ok(rep) => {
1400 shared.log_line(format!("[strategy] {verb}失败:{}", rep.msg));
1401 shared.set_strategy_feedback(crate::state::StratFeedback {
1404 op: "toast",
1405 name: String::new(),
1406 ok: false,
1407 msg: format!("{verb}失败:{}", rep.msg),
1408 });
1409 }
1410 Err(e) => {
1411 shared.log_line(format!("[strategy] {verb}失败:{e}"));
1412 shared.set_strategy_feedback(crate::state::StratFeedback {
1413 op: "toast",
1414 name: String::new(),
1415 ok: false,
1416 msg: format!("{verb}失败:{e}"),
1417 });
1418 }
1419 }
1420}
1421
1422impl Plugin for UiPlugin {
1423 fn name(&self) -> &str {
1424 UI_NAME
1425 }
1426
1427 fn deps(&self) -> &[&str] {
1428 &["cfg"]
1431 }
1432
1433 fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
1434 vec![
1435 std::borrow::Cow::Borrowed(EVENT_LOG),
1436 std::borrow::Cow::Borrowed(EVENT_PLUG_UP),
1437 std::borrow::Cow::Borrowed(EVENT_PLUG_DOWN),
1438 std::borrow::Cow::Borrowed(EVENT_BAR),
1439 std::borrow::Cow::Borrowed(EVENT_BAR_UPDATE),
1440 std::borrow::Cow::Borrowed(EVENT_TICK),
1441 std::borrow::Cow::Borrowed(EVENT_CONTRACT),
1442 std::borrow::Cow::Borrowed(ek_ctp::EVENT_CONTRACT_FINISH),
1443 std::borrow::Cow::Borrowed(W_ORDER),
1447 std::borrow::Cow::Borrowed(W_TRADE),
1448 std::borrow::Cow::Borrowed(W_POSITION),
1449 std::borrow::Cow::Borrowed(W_ACCOUNT),
1450 std::borrow::Cow::Borrowed(EVENT_TIMER),
1451 std::borrow::Cow::Borrowed(ek_ctp::EVENT_STATUS_PREFIX),
1453 std::borrow::Cow::Borrowed(ek_strategy::contract::EVENT_STRATEGY),
1456 std::borrow::Cow::Borrowed(ek_strategy::contract::EVENT_STRATEGY_META),
1457 ]
1458 }
1459
1460 fn mailbox_soft_cap(&self) -> u64 {
1461 UI_MAILBOX_SOFT_CAP
1462 }
1463
1464 fn version(&self) -> std::borrow::Cow<'static, str> {
1465 std::borrow::Cow::Borrowed(env!("CARGO_PKG_VERSION"))
1466 }
1467
1468 fn services(&self) -> Vec<ServiceDesc> {
1469 Vec::new() }
1471
1472 fn on_init(&self, ctx: &PluginContext) {
1473 let _ = self.ctx.set(ctx.clone());
1474 }
1475
1476 fn on_start(&self) {
1477 self.shared.log_line("[ui] 界面插件上线");
1478 self.spawn_cmd_thread();
1482 self.shared.request_trade_reseed();
1485 self.shared.request_pipeline_refresh();
1486 let theme = load_theme();
1488 self.shared.set_theme(theme);
1489 if theme == crate::tokens::Theme::Light {
1490 self.shared.log_line("[ui] 主题:浅色(读盘 ui.theme)");
1491 }
1492 }
1493
1494 fn on_stop(&self) {
1495 self.shared.log_line("[ui] 界面插件下线");
1496 self.cmd_stop
1498 .store(true, std::sync::atomic::Ordering::Release);
1499 self.flush_persist();
1503 }
1504
1505 fn on_mail(&self, event: &ek_event::Event) {
1506 let ty = event.event_type.as_str();
1507 if ty == ek_strategy::contract::EVENT_STRATEGY {
1509 if let Some(ev) = event
1510 .data
1511 .downcast_ref::<ek_strategy::contract::StrategyEvent>()
1512 {
1513 self.shared.apply_strategy_event(ev);
1514 } else {
1515 self.shared
1516 .log_line("[eStrategy] 载荷类型异常(非 StrategyEvent)");
1517 }
1518 return;
1519 }
1520 if ty == ek_strategy::contract::EVENT_STRATEGY_META {
1521 if let Some(metas) = event
1522 .data
1523 .downcast_ref::<Vec<ek_strategy::strategy::StrategyMeta>>()
1524 {
1525 self.shared.set_strategy_metas(metas.clone());
1526 } else {
1527 self.shared
1528 .log_line("[eStrategyMeta] 载荷类型异常(非 Vec<StrategyMeta>)");
1529 }
1530 return;
1531 }
1532 if ty == EVENT_LOG {
1533 if let Some(d) = event.data.downcast_ref::<LogData>() {
1534 let tag = if d.is_warning_or_above() {
1535 "WARN"
1536 } else {
1537 "INFO"
1538 };
1539 self.shared
1540 .log_line(format!("[{tag}][{}] {}", d.gateway_name, d.msg));
1541 if let Some(note) = d.msg.strip_prefix(crate::state::CHAIN_STATE_PREFIX) {
1547 self.shared.note_chain(&d.gateway_name, note);
1548 }
1549 } else {
1550 self.shared.log_line("[eLog] 载荷类型异常(非 LogData)");
1551 }
1552 return;
1553 }
1554 if ty == ek_ctp::EVENT_STATUS_PREFIX {
1555 if let Some(ev) = event.data.downcast_ref::<ek_ctp::GatewayStatusEvent>() {
1557 self.shared.note_status_event(ev);
1558 } else {
1559 self.shared
1560 .log_line("[eStatus] 载荷类型异常(非 GatewayStatusEvent)");
1561 }
1562 return;
1563 }
1564 if ty == EVENT_PLUG_UP || ty == EVENT_PLUG_DOWN {
1565 if let Some(n) = event.data.downcast_ref::<PlugNotice>() {
1566 let verb = if ty == EVENT_PLUG_UP {
1567 "上线"
1568 } else {
1569 "下线"
1570 };
1571 self.shared
1572 .log_line(format!("[host] 插件{verb} {} v{}", n.name, n.version));
1573 self.shared.request_list_refresh();
1575 if ty == EVENT_PLUG_UP {
1576 if n.name == ek_strategy::PLUGIN_NAME {
1579 self.shared.push_cmd(crate::state::UiCmd::StratResync);
1580 }
1581 if n.name == ek_oms::NAME {
1583 self.shared.request_pipeline_refresh();
1584 }
1585 let is_account = self.shared.accounts().iter().any(|a| a.gateway == n.name);
1590 if n.name == ek_oms::NAME || is_account {
1591 self.shared.request_trade_reseed();
1592 }
1593 if is_account {
1596 self.shared.request_status_seed(&n.name);
1597 }
1598 let md = self.shared.md_source();
1601 let subs = self.shared.subscriptions();
1602 if !md.is_empty() && n.name == md && !subs.is_empty() {
1603 self.shared.push_cmd(crate::state::UiCmd::CtpSubscribe {
1604 gateway: n.name.clone(),
1605 symbols: subs.clone(),
1606 });
1607 self.shared.log_line(format!(
1608 "[ops] 行情源 {} 上线,重放订阅 {} 项",
1609 n.name,
1610 subs.len()
1611 ));
1612 }
1613 } else {
1614 let removed = self.shared.clear_gateway_rows(&n.name);
1618 if removed > 0 {
1619 self.shared.log_line(format!(
1620 "[ui] 插件下线 {}:作废 {} 行交易数据",
1621 n.name, removed
1622 ));
1623 }
1624 if n.name == ek_oms::NAME {
1629 self.shared.request_pipeline_refresh();
1630 }
1631 }
1632 }
1633 self.shared.bump();
1634 return;
1635 }
1636 if ty == EVENT_BAR {
1637 if let Some(f) = event.data.downcast_ref::<ek_dm::wire::BarsFrame>() {
1638 self.shared.market().on_frame(f);
1639 }
1640 self.shared.bump();
1641 return;
1642 }
1643 if ty == EVENT_BAR_UPDATE {
1644 if let Some(b) = event.data.downcast_ref::<ek_data::bar::BarData>() {
1645 self.shared.market().on_update(b);
1646 }
1647 self.shared.bump();
1648 return;
1649 }
1650 if ty == EVENT_TICK {
1651 if let Some(t) = event.data.downcast_ref::<ek_data::tick::TickData>() {
1652 self.shared.update_tick(t);
1653 }
1654 return;
1655 }
1656 if ty == EVENT_CONTRACT {
1657 if let Some(v) = event
1660 .data
1661 .downcast_ref::<Vec<ek_data::contract::ContractData>>()
1662 {
1663 for c in v {
1664 self.shared.update_contract(c);
1665 }
1666 } else if let Some(c) = event.data.downcast_ref::<ek_data::contract::ContractData>() {
1667 self.shared.update_contract(c);
1668 }
1669 return;
1670 }
1671 if ty == ek_ctp::EVENT_CONTRACT_FINISH {
1672 let gw = event
1673 .data
1674 .downcast_ref::<String>()
1675 .cloned()
1676 .unwrap_or_else(|| "?".to_string());
1677 self.shared
1678 .log_line(format!("[ctp] 合约流完成 {gw}(查询合约窗可勾选订阅)"));
1679 return;
1680 }
1681 if ty == W_ORDER {
1684 match event.data.downcast_ref::<OrderData>() {
1685 Some(o) => self.shared.note_order(o),
1686 None => self.shared.note_bad_payload(W_ORDER),
1687 }
1688 return;
1689 }
1690 if ty == W_TRADE {
1691 match event.data.downcast_ref::<TradeData>() {
1692 Some(t) => self.shared.note_trade(t),
1693 None => self.shared.note_bad_payload(W_TRADE),
1694 }
1695 return;
1696 }
1697 if ty == W_POSITION {
1698 match event.data.downcast_ref::<PositionData>() {
1699 Some(p) => self.shared.note_position(p),
1700 None => self.shared.note_bad_payload(W_POSITION),
1701 }
1702 return;
1703 }
1704 if ty == W_ACCOUNT {
1705 match event.data.downcast_ref::<AccountData>() {
1706 Some(a) => self.shared.note_funds(a),
1707 None => self.shared.note_bad_payload(W_ACCOUNT),
1708 }
1709 return;
1710 }
1711 if ty == EVENT_TIMER {
1712 self.refresh();
1713 }
1714 }
1715}