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
72fn 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(store: &ek_cfg::Store, theme: crate::tokens::Theme) -> Result<ek_core::SectionMap, String> {
122 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
123 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
124 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
125 };
126 map = map.set(THEME_KEY, ek_core::Field::plain(theme.as_str()));
127 Ok(map)
128}
129
130fn build_auto_connect_map(
138 store: &ek_cfg::Store,
139 flags: &[(String, bool)],
140) -> Result<ek_core::SectionMap, String> {
141 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
142 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
143 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
144 };
145 let stale: Vec<String> = map
146 .iter()
147 .filter(|(k, _)| k.starts_with("auto_connect."))
148 .map(|(k, _)| (*k).to_string())
149 .collect();
150 for k in stale {
151 map.remove(&k);
152 }
153 for (gw, on) in flags {
154 map = map.set(
155 format!("auto_connect.{gw}"),
156 ek_core::Field::plain(if *on { "1" } else { "0" }),
157 );
158 }
159 Ok(map)
160}
161
162pub fn save_auto_connect_flags_into(
165 store: &ek_cfg::Store,
166 flags: &[(String, bool)],
167) -> Result<(), String> {
168 let map = build_auto_connect_map(store, flags)?;
169 ek_cfg::cfg_save_at(store.root(), THEME_NS, &map).map_err(|e| e.to_string())
170}
171
172fn build_aliases_map(
177 store: &ek_cfg::Store,
178 aliases: &[(String, String)],
179) -> Result<ek_core::SectionMap, String> {
180 let mut map = match store.load(THEME_NS).map_err(|e| e.to_string())? {
181 ek_cfg::LoadOutcome::Loaded { map, .. } => map,
182 ek_cfg::LoadOutcome::Fresh => ek_core::SectionMap::new(1),
183 };
184 let stale: Vec<String> = map
185 .iter()
186 .filter(|(k, _)| k.starts_with("alias."))
187 .map(|(k, _)| (*k).to_string())
188 .collect();
189 for k in stale {
190 map.remove(&k);
191 }
192 for (gw, alias) in aliases {
193 let alias = alias.trim();
194 if alias.is_empty() {
195 continue;
196 }
197 map = map.set(format!("alias.{gw}"), ek_core::Field::plain(alias));
198 }
199 Ok(map)
200}
201
202pub fn save_account_aliases_into(
204 store: &ek_cfg::Store,
205 aliases: &[(String, String)],
206) -> Result<(), String> {
207 let map = build_aliases_map(store, aliases)?;
208 ek_cfg::cfg_save_at(store.root(), THEME_NS, &map).map_err(|e| e.to_string())
209}
210
211use crate::state::{BusSnap, UiShared};
212
213pub const UI_NAME: &str = "ui";
215
216pub const UI_MAILBOX_SOFT_CAP: u64 = 100_000;
225
226const W_ORDER: &str = ek_oms::wire::EVENT_ORDER;
229const W_TRADE: &str = ek_oms::wire::EVENT_TRADE;
230const W_POSITION: &str = ek_oms::wire::EVENT_POSITION;
231const W_ACCOUNT: &str = ek_oms::wire::EVENT_ACCOUNT;
232
233pub struct UiPlugin {
235 shared: std::sync::Arc<UiShared>,
236 ctx: OnceLock<PluginContext>,
237 cmd_stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
239}
240
241impl UiPlugin {
242 pub fn new(shared: std::sync::Arc<UiShared>) -> Self {
244 Self {
245 shared,
246 ctx: OnceLock::new(),
247 cmd_stop: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
248 }
249 }
250
251 fn refresh(&self) {
254 let Some(ctx) = self.ctx.get() else {
255 return; };
257 if self.shared.list_poll_due() {
262 self.poll_list(ctx);
263 }
264 self.seed_status(ctx);
265 if self.shared.take_pipeline_due() {
266 self.poll_pipeline(ctx);
267 }
268 if self.shared.take_cfg_due() {
269 self.poll_cfg(ctx);
272 }
273 self.flush_persist();
274 }
275
276 fn cfg_bus_save(&self, ns: &str, map: &ek_core::SectionMap) -> Result<(), String> {
280 if let Some(ctx) = self.ctx.get() {
281 match ctx.ask_service::<ek_cfg::wire::SaveReq, ek_cfg::wire::SaveRes>(
282 "cfg.save",
283 ek_cfg::wire::SaveReq {
284 ns: ns.to_string(),
285 map: map.clone(),
286 },
287 Duration::from_millis(1500),
288 ) {
289 Ok(r) if r.ok => return Ok(()),
290 Ok(r) => return Err(r.msg),
291 Err(ek_core::error::AskError::UnknownService { .. }) => {
292 self.shared
293 .log_line(format!("[cfg] cfg 缺席,{ns} 直写托底(软集中负形)"));
294 }
295 Err(e) => return Err(format!("{e:?}")),
296 }
297 }
298 ui_cfg_store().save(ns, map).map_err(|e| e.to_string())
299 }
300
301 fn cfg_bus_delete(&self, ns: &str) -> Result<(), String> {
304 if let Some(ctx) = self.ctx.get() {
305 match ctx.ask_service::<ek_cfg::wire::DeleteReq, ek_cfg::wire::DeleteRes>(
306 "cfg.delete",
307 ek_cfg::wire::DeleteReq {
308 ns: ns.to_string(),
309 key: String::new(),
310 },
311 Duration::from_millis(1500),
312 ) {
313 Ok(r) if r.ok => return Ok(()),
314 Ok(r) => return Err(r.msg),
315 Err(ek_core::error::AskError::UnknownService { .. }) => {
316 self.shared
317 .log_line(format!("[cfg] cfg 缺席,{ns} 直清托底(软集中负形)"));
318 }
319 Err(e) => return Err(format!("{e:?}")),
320 }
321 }
322 crate::appconfig::purge_ns_direct(&ui_cfg_store(), ns)
323 }
324
325 fn flush_persist(&self) {
331 if self.shared.take_theme_persist_due() {
332 let theme = self.shared.theme();
335 let wr = build_theme_map(&ui_cfg_store(), theme)
336 .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
337 match wr {
338 Ok(()) => self
339 .shared
340 .log_line(format!("[ui] 主题已存(ui.theme={})", theme.as_str())),
341 Err(e) => self
342 .shared
343 .log_line(format!("[ui] 主题落盘失败:{e}(本次切换只在本次运行内生效)")),
344 }
345 }
346 let purges = self.shared.take_archive_purges();
347 for gw in purges {
348 match self.cfg_bus_delete(&format!("ctp.{gw}")) {
353 Ok(()) => self.shared.log_line(format!(
354 "[账户] 档案段 ctp.{gw} 已抹(删除已落盘:下次启动清点不再复活)"
355 )),
356 Err(e) => self.shared.log_line(format!(
357 "[账户] 档案 ctp.{gw} 抹除失败:{e}(本次删除只在今次运行生效,重启会复活)"
358 )),
359 }
360 }
361 if self.shared.take_auto_connect_persist_due() {
362 let flags: Vec<(String, bool)> = self
365 .shared
366 .accounts()
367 .iter()
368 .map(|a| (a.gateway.clone(), a.auto_connect))
369 .collect();
370 let wr = build_auto_connect_map(&ui_cfg_store(), &flags)
372 .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
373 match wr {
374 Ok(()) => {
375 let on: Vec<&str> = flags
376 .iter()
377 .filter(|(_, on)| *on)
378 .map(|(gw, _)| gw.as_str())
379 .collect();
380 if on.is_empty() {
381 self.shared
382 .log_line("[ui] 自动连接勾选已存(ui 段;当前全部关)".to_string());
383 } else {
384 self.shared.log_line(format!(
385 "[ui] 自动连接勾选已存(ui 段;开:{};守护/桌面启动清点按它起连)",
386 on.join(", ")
387 ));
388 }
389 }
390 Err(e) => self.shared.log_line(format!(
391 "[ui] 自动连接勾选落盘失败:{e}(本轮勾选只在本次运行内生效)"
392 )),
393 }
394 }
395 if self.shared.take_alias_persist_due() {
396 let aliases: Vec<(String, String)> = self
398 .shared
399 .accounts()
400 .iter()
401 .map(|a| (a.gateway.clone(), a.alias.clone()))
402 .collect();
403 let wr = build_aliases_map(&ui_cfg_store(), &aliases)
405 .and_then(|map| self.cfg_bus_save(THEME_NS, &map));
406 match wr {
407 Ok(()) => {
408 let named: Vec<String> = aliases
409 .iter()
410 .filter(|(_, al)| !al.trim().is_empty())
411 .map(|(gw, al)| format!("{al}({gw})"))
412 .collect();
413 if named.is_empty() {
414 self.shared
415 .log_line("[ui] 账户别名已存(ui 段;当前全部无别名)".to_string());
416 } else {
417 self.shared.log_line(format!("[ui] 账户别名已存(ui 段;{};启动清点按它贴标签)", named.join(", ")));
418 }
419 }
420 Err(e) => self.shared.log_line(format!(
421 "[ui] 账户别名落盘失败:{e}(本次改名只在本次运行内生效)"
422 )),
423 }
424 }
425 }
426
427 fn poll_cfg(&self, ctx: &PluginContext) {
429 let (_, svc_list, _) = cfg_svc_ids();
430 match ctx.ask_service::<CfgListReq, CfgListRes>(
431 svc_list.as_str(),
432 CfgListReq,
433 Duration::from_secs(2),
434 ) {
435 Ok(res) => {
436 self.shared.set_cfg_briefs(res.entries);
437 if !res.msg.is_empty() {
438 self.shared
439 .log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
440 }
441 }
442 Err(e) => self
443 .shared
444 .log_line(format!("[cfg] 段落目录失败(cfg 缺席或超时): {e:?}")),
445 }
446 }
447
448 fn poll_list(&self, ctx: &PluginContext) {
454 match ctx.ask_service::<ListReq, Vec<PlugRow>>(SVC_LIST, ListReq, Duration::from_secs(2)) {
455 Ok(rows) => {
456 let mut services = Vec::with_capacity(rows.len());
457 for r in &rows {
458 let ids = ctx
459 .services_of(&r.name)
460 .map(|v| v.iter().map(|s| s.id.clone()).collect::<Vec<_>>())
461 .unwrap_or_default();
462 services.push((r.name.clone(), ids));
463 }
464 self.shared.sync_accounts_connected(&rows);
467 self.shared.note_plugin_spill(&rows);
470 self.shared.set_snapshot(BusSnap { rows, services });
471 }
472 Err(e) => self.shared.log_line(format!("[ui] svctl.list 失败: {e}")),
473 }
474 }
475
476 fn seed_status(&self, ctx: &PluginContext) {
483 for gw in self.shared.take_status_seeds() {
484 let svc = format!("{gw}.{}", ek_ctp::SVC_STATUS);
485 match ctx.ask_service::<StatusReq, GatewayStatus>(
486 &svc,
487 StatusReq,
488 Duration::from_secs(2),
489 ) {
490 Ok(st) => self.shared.note_gateway_status(&gw, &st),
491 Err(e) => {
492 self.shared.note_status_unreachable(&gw, &format!("{e}"));
496 }
497 }
498 }
499 }
500
501 fn poll_pipeline(&self, ctx: &PluginContext) {
506 match ctx.ask_service::<PipelineStatusReq, Vec<String>>(
507 ek_oms::SVC_PIPELINE_STATUS,
508 PipelineStatusReq,
509 Duration::from_secs(2),
510 ) {
511 Ok(lines) => self.shared.set_pipeline_lines(lines),
512 Err(e) => {
513 if !self.shared.pipeline_lines().is_empty() {
517 self.shared
518 .log_line(format!("[ui] oms.pipeline_status 失手: {e:?}"));
519 }
520 self.shared.set_pipeline_lines(Vec::new());
521 }
522 }
523 }
524
525 fn reconcile_trade(shared: &UiShared, ctx: &PluginContext) {
535 let orders = ctx.ask_service::<GetAllOrdersReq, Vec<OrderData>>(
536 "oms.get_all_orders",
537 GetAllOrdersReq,
538 Duration::from_secs(2),
539 );
540 let trades = ctx.ask_service::<GetAllTradesReq, Vec<TradeData>>(
541 "oms.get_all_trades",
542 GetAllTradesReq,
543 Duration::from_secs(2),
544 );
545 let positions = ctx.ask_service::<GetAllPositionsReq, Vec<PositionData>>(
546 "oms.get_all_positions",
547 GetAllPositionsReq,
548 Duration::from_secs(2),
549 );
550 let funds = ctx.ask_service::<GetAllAccountsReq, Vec<AccountData>>(
551 "oms.get_all_accounts",
552 GetAllAccountsReq,
553 Duration::from_secs(2),
554 );
555 let mut n = 0usize;
556 if let Ok(rows) = &orders {
557 for o in rows {
558 shared.note_order(o);
559 }
560 n += rows.len();
561 }
562 if let Ok(rows) = &trades {
563 for t in rows {
564 shared.note_trade(t);
565 }
566 n += rows.len();
567 }
568 if let Ok(rows) = &positions {
569 for p in rows {
570 shared.note_position(p);
571 }
572 n += rows.len();
573 }
574 if let Ok(rows) = &funds {
575 for a in rows {
576 shared.note_funds(a);
577 }
578 n += rows.len();
579 }
580 if n > 0 {
582 let bad = usize::from(orders.is_err())
583 + usize::from(trades.is_err())
584 + usize::from(positions.is_err())
585 + usize::from(funds.is_err());
586 shared.log_line(format!(
587 "[ui] 启动播种 {n} 行(委托 {} / 成交 {} / 持仓 {} / 资金 {};失手 {} 口)",
588 orders.as_ref().map_or(0, |v| v.len()),
589 trades.as_ref().map_or(0, |v| v.len()),
590 positions.as_ref().map_or(0, |v| v.len()),
591 funds.as_ref().map_or(0, |v| v.len()),
592 bad
593 ));
594 }
595 }
596
597 fn spawn_cmd_thread(&self) {
605 let Some(ctx) = self.ctx.get().cloned() else {
606 return; };
608 let shared = std::sync::Arc::clone(&self.shared);
609 let stop = std::sync::Arc::clone(&self.cmd_stop);
610 if std::thread::Builder::new()
611 .name("ui-cmd".to_string())
612 .spawn(move || {
613 loop {
614 if stop.load(std::sync::atomic::Ordering::Acquire) {
615 break;
616 }
617 if shared.cmd_pending() {
619 Self::run_cmds(&shared, &ctx);
620 continue;
621 }
622 if shared.take_trade_reseed() {
624 Self::reconcile_trade(&shared, &ctx);
625 continue;
626 }
627 if shared.kline_fetch_pending() {
629 Self::run_kline_fetches(&shared, &ctx);
630 continue;
631 }
632 std::thread::sleep(Duration::from_millis(20));
633 }
634 })
635 .is_err()
636 {
637 self.shared
639 .log_line("[ui] 命令线程起不来:操作指令会滞留在队列里");
640 }
641 }
642
643 fn run_kline_fetches(shared: &UiShared, ctx: &PluginContext) {
647 for f in shared.take_kline_fetches() {
648 let Some(gw) = ctx
649 .plugins()
650 .iter()
651 .map(|p| p.name.clone())
652 .find(|n| n.starts_with("tq"))
653 else {
654 shared.log_line(format!(
655 "[kline] {} tq 网关不在线,历史补数丢弃(重开 K线窗再试)",
656 f.symbol
657 ));
658 continue;
659 };
660 let req = ek_tq::KlinesReq {
661 symbol: f.symbol.clone(),
662 raw: false,
663 interval_secs: f.interval_secs,
664 range: ek_tq::RangeSpec::Length { count: 300, end_ns: None },
665 max_rows: Some(300),
666 };
667 match ctx.ask::<ek_tq::KlinesReq, ek_tq::KlinesReply>(
668 &gw,
669 req,
670 Duration::from_secs(45),
671 ) {
672 Ok(reply) => {
673 if let Some(err) = &reply.error {
674 shared.log_line(format!(
675 "[kline] {}/{}s 历史显形: {}",
676 f.symbol, f.interval_secs, err.message
677 ));
678 continue;
679 }
680 let n = reply.rows.len();
681 shared.market().note_history(&f.symbol, f.interval_secs, &reply.rows);
682 shared.bump();
683 shared.log_line(format!(
684 "[kline] {}/{}s 历史补数 +{n} 根(mode={} truncated={})",
685 f.symbol,
686 f.interval_secs,
687 reply.mode.as_str(),
688 reply.truncated
689 ));
690 }
691 Err(e) => shared.log_line(format!(
692 "[kline] {}/{}s 历史 ask 失手: {e:?}",
693 f.symbol, f.interval_secs
694 )),
695 }
696 }
697 }
698
699 fn run_cmds(shared: &UiShared, ctx: &PluginContext) { for cmd in shared.drain_cmds() {
701 match cmd {
702 crate::state::UiCmd::Start { factory, name } => {
703 match ctx.ask_service::<StartReq, StartRes>(
704 SVC_START,
705 StartReq {
706 factory: factory.clone(),
707 name: name.clone(),
708 },
709 Duration::from_secs(3),
710 ) {
711 Ok(StartRes::Started { name }) => {
712 shared.log_line(format!("[ops] svctl.start {factory} → {name} 上线"));
713 }
714 Ok(StartRes::Rejected { reason }) => {
715 shared.log_line(format!("[ops] start {factory} 被拒: {reason}"));
716 }
717 Err(e) => {
718 shared.log_line(format!("[ops] start {factory} 失败: {e:?}"));
719 }
720 }
721 shared.request_list_refresh();
723 }
724 crate::state::UiCmd::Remove { name } => {
725 if name == UI_NAME {
733 shared.log_line(
734 "[ops] 拒绝移除 ui:观测面自身不可自摘(摘掉后没有下一拍执行任何指令)"
735 .to_string(),
736 );
737 continue;
738 }
739 match ctx.ask_service::<RemoveReq, RemoveRes>(
740 SVC_REMOVE,
741 RemoveReq { name: name.clone() },
742 Duration::from_secs(3),
743 ) {
744 Ok(RemoveRes::Removed { name }) => {
745 shared.log_line(format!("[ops] svctl.remove {name} 完成"));
746 }
747 Ok(RemoveRes::NoSuchName { name }) => {
748 shared.log_line(format!("[ops] remove {name} 不存在/不可摘"));
749 }
750 Ok(RemoveRes::Rejected { name, reason }) => {
754 shared.log_line(format!("[ops] remove {name} 被拒: {reason}"));
755 }
756 Err(e) => {
757 shared.log_line(format!("[ops] remove {name} 失败: {e:?}"));
758 }
759 }
760 shared.request_list_refresh();
761 }
762 crate::state::UiCmd::Reload {
763 interval,
764 symbol,
765 bars,
766 } => {
767 match ctx.ask_service::<ReloadReq, usize>(
769 "dm.reload",
770 ReloadReq {
771 interval,
772 symbol: symbol.clone(),
773 bars: bars.clone(),
774 size: Some(bars.len()),
775 push: true,
776 },
777 Duration::from_secs(5),
778 ) {
779 Ok(n) => {
780 shared.market().note_reload();
781 shared.log_line(format!(
782 "[ops] dm.reload {symbol}@{interval}s 回灌 {n} 条"
783 ));
784 }
785 Err(e) => {
786 shared.log_line(format!("[ops] dm.reload {symbol} 失败: {e:?}"));
787 }
788 }
789 }
790 crate::state::UiCmd::SendOrder { req, gateway, mode } => {
791 match ctx.ask_service::<SubmitOrderReq, SubmitReply>(
795 "oms.submit_order",
796 SubmitOrderReq {
797 req: req.clone(),
798 gateway_name: gateway.clone(),
799 mode,
800 },
801 Duration::from_secs(10),
802 ) {
803 Ok(reply) => {
804 let tag = match reply.status {
805 SubmitStatus::Sent => "SENT",
806 SubmitStatus::Partial => "PARTIAL",
807 SubmitStatus::Rejected => "REJECTED",
808 };
809 for leg in &reply.legs {
810 match (leg.ok, &leg.en_orderid, &leg.reason) {
811 (true, Some(id), _) => shared.log_line(format!(
812 "[ops] submit {} {} {}x{} → {id} (parent {})",
813 leg.req.symbol,
814 leg.req.direction.as_str(),
815 leg.req.offset.as_str(),
816 leg.req.volume,
817 reply.parent_id
818 )),
819 _ => shared.log_line(format!(
820 "[ops] submit {} {} {}x{} 被拒: {} (parent {})",
821 leg.req.symbol,
822 leg.req.direction.as_str(),
823 leg.req.offset.as_str(),
824 leg.req.volume,
825 leg.reason.as_deref().unwrap_or("?"),
826 reply.parent_id
827 )),
828 }
829 }
830 if reply.legs.is_empty() {
831 shared.log_line(format!(
832 "[ops] submit {} {}x{} {tag}:可平不足零腿 (parent {})",
833 req.symbol,
834 req.offset.as_str(),
835 req.volume,
836 reply.parent_id
837 ));
838 }
839 }
840 Err(e) => {
841 shared.log_line(format!("[ops] oms.submit_order 失败: {e:?}"));
842 }
843 }
844 }
845 crate::state::UiCmd::CancelOrder { orderid, gateway } => {
846 match ctx.ask_service::<OmsCancelOrderReq, OmsCancelReply>(
850 "oms.cancel_order",
851 OmsCancelOrderReq {
852 en_orderid: format!("{gateway}.{orderid}"),
853 gateway_name: gateway.clone(),
854 },
855 Duration::from_secs(10),
856 ) {
857 Ok(OmsCancelReply { ok: true, msg }) => {
858 shared.log_line(format!("[ops] {msg}"));
859 }
860 Ok(OmsCancelReply { ok: false, msg }) => {
861 shared
863 .log_line(format!("[ops] {}", UiShared::cancel_reject_note(&msg)));
864 }
865 Err(e) => {
866 shared.log_line(format!("[ops] oms.cancel_order 失败: {e:?}"));
867 }
868 }
869 }
870 crate::state::UiCmd::CtpConfigure { gateway, form } => {
871 let svc = format!("{gateway}.configure");
874 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
875 svc.as_str(),
876 form.to_configure_req(),
877 Duration::from_secs(5),
878 ) {
879 Ok(ConfigureReply { ok: true, msg, .. }) => {
883 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
884 }
885 Ok(ConfigureReply { ok: false, msg, .. }) => {
886 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
887 }
888 Err(e) => {
889 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
890 }
891 }
892 }
893 crate::state::UiCmd::Connect { gateway, form } => {
894 shared.note_connect_sent(&gateway);
899 let form = shared.prepare_connect_form(&gateway, form);
902 let started = match ctx.ask_service::<StartReq, StartRes>(
903 SVC_START,
904 StartReq {
905 factory: "ctp".to_string(),
906 name: gateway.clone(),
907 },
908 Duration::from_secs(3),
909 ) {
910 Ok(StartRes::Started { name }) => {
911 shared.log_line(format!("[ops] svctl.start ctp → {name} 上线"));
912 true
913 }
914 Ok(StartRes::Rejected { reason }) => {
915 if shared.account(&gateway).is_some_and(|a| a.connected) {
918 shared
919 .log_line(format!("[ops] {gateway} 已在场,重投凭据: {reason}"));
920 true
921 } else {
922 shared.log_line(format!("[ops] {gateway} 装配被拒: {reason}"));
923 shared.note_connect_result(&gateway, Err(reason));
924 continue;
925 }
926 }
927 Err(e) => {
928 shared.log_line(format!("[ops] {gateway} 装配失败: {e:?}"));
929 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
930 continue;
931 }
932 };
933 if !started {
934 continue;
935 }
936 let svc = format!("{gateway}.configure");
937 match ctx.ask_service::<ConfigureReq, ConfigureReply>(
938 svc.as_str(),
939 form.to_configure_req(),
940 Duration::from_secs(5),
941 ) {
942 Ok(ConfigureReply {
946 ok: true,
947 msg,
948 product_classes,
949 }) => {
950 shared.log_line(format!("[ops] {svc} 受理:{msg}"));
951 shared.note_effective_product_classes(&gateway, &product_classes);
952 shared.note_connect_result(&gateway, Ok(()));
953 }
954 Ok(ConfigureReply { ok: false, msg, .. }) => {
955 shared.log_line(format!("[ops] {svc} 被拒:{msg}"));
956 shared.note_connect_result(&gateway, Err(msg));
957 }
958 Err(e) => {
959 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
960 shared.note_connect_result(&gateway, Err(format!("{e:?}")));
961 }
962 }
963 }
964 crate::state::UiCmd::CtpSubscribe { gateway, symbols } => {
965 let svc = format!("{gateway}.subscribe");
969 match ctx.ask_service::<CtpSubscribeReq, CtpSubscribeReply>(
970 svc.as_str(),
971 CtpSubscribeReq {
972 symbols: symbols.clone(),
973 },
974 Duration::from_secs(5),
975 ) {
976 Ok(CtpSubscribeReply { ok: true, msg }) => {
977 shared.note_subscriptions(&symbols);
978 shared.log_line(format!(
979 "[ops] {svc} {} 受理:{}",
980 symbols.join(","),
981 msg
982 ));
983 }
984 Ok(CtpSubscribeReply { ok: false, msg }) => {
985 shared.log_line(format!(
986 "[ops] {svc} {} 被拒:{}",
987 symbols.join(","),
988 msg
989 ));
990 }
991 Err(e) => {
992 shared.log_line(format!("[ops] {svc} 失败: {e:?}"));
993 }
994 }
995 }
996 crate::state::UiCmd::Disconnect { gateway } => {
997 let svc = format!("{gateway}.disconnect");
1001 match ctx.ask_service::<DisconnectReq, DisconnectReply>(
1002 svc.as_str(),
1003 DisconnectReq,
1004 Duration::from_secs(5),
1005 ) {
1006 Ok(DisconnectReply { ok: true, msg }) => {
1007 shared.mark_account_stopped(&gateway);
1008 shared.log_line(format!("[账户] 断开 {gateway}:{msg}(实例仍在场)"));
1009 }
1010 Ok(DisconnectReply { ok: false, msg }) => {
1011 shared.log_line(format!("[账户] 断开 {gateway} 未执行:{msg}"));
1012 }
1013 Err(e) => {
1014 shared.log_line(format!(
1015 "[账户] 断开 {gateway} 失败:{e:?}(网关不在场或超时)",
1016 ));
1017 }
1018 }
1019 }
1020 crate::state::UiCmd::CfgList => {
1021 let (_, svc_list, _) = cfg_svc_ids();
1023 match ctx.ask_service::<CfgListReq, CfgListRes>(
1024 svc_list.as_str(),
1025 CfgListReq,
1026 Duration::from_secs(2),
1027 ) {
1028 Ok(res) => {
1029 shared.set_cfg_briefs(res.entries);
1030 if !res.msg.is_empty() {
1031 shared.log_line(format!("[cfg] 目录:{msg}", msg = res.msg));
1032 }
1033 }
1034 Err(e) => shared.log_line(format!("[cfg] 段落目录失败: {e:?}")),
1035 }
1036 }
1037 crate::state::UiCmd::CfgDelete { ns } => {
1038 let (svc_save, svc_list, svc_delete) = cfg_svc_ids();
1040 let _ = (svc_save, svc_list);
1041 match ctx.ask_service::<CfgDeleteReq, CfgDeleteRes>(
1042 svc_delete.as_str(),
1043 CfgDeleteReq {
1044 ns: ns.clone(),
1045 key: String::new(),
1046 },
1047 Duration::from_secs(2),
1048 ) {
1049 Ok(res) if res.ok => {
1050 shared.log_line(format!(
1051 "[cfg] 清段 {ns}:键 {} / 秘条 {}",
1052 res.removed_keys, res.removed_secrets
1053 ));
1054 shared.request_cfg_refresh();
1055 }
1056 Ok(res) => shared.log_line(format!("[cfg] 清段 {ns} 未执行:{}", res.msg)),
1057 Err(e) => shared.log_line(format!("[cfg] 清段 {ns} 失败: {e:?}")),
1058 }
1059 shared.arm_cfg_delete(None);
1060 }
1061 crate::state::UiCmd::CfgResave { ns } => {
1062 let store = ui_cfg_store();
1067 let loaded = store
1068 .load(&ns)
1069 .map_err(|e| e.to_string())
1070 .and_then(|o| match o {
1071 ek_cfg::LoadOutcome::Fresh => Err(format!("{ns} 段不存在,不持久空形")),
1072 map @ ek_cfg::LoadOutcome::Loaded { .. } => Ok(map),
1073 });
1074 match loaded {
1075 Ok(ek_cfg::LoadOutcome::Fresh) => {
1076 unreachable!("Fresh 已被上面的 and_then 挡走");
1077 }
1078 Ok(ek_cfg::LoadOutcome::Loaded { map, warnings }) => {
1079 if !warnings.is_empty() {
1080 shared.log_line(format!("[cfg] {ns} 重读警告:{warnings:?}"));
1081 }
1082 let wr = match ctx.ask_service::<
1085 ek_cfg::wire::SaveReq,
1086 ek_cfg::wire::SaveRes,
1087 >(
1088 "cfg.save",
1089 ek_cfg::wire::SaveReq {
1090 ns: ns.clone(),
1091 map: map.clone(),
1092 },
1093 Duration::from_millis(1500),
1094 ) {
1095 Ok(r) if r.ok => Ok(()),
1096 Ok(r) => Err(r.msg),
1097 Err(ek_core::error::AskError::UnknownService { .. }) => {
1098 shared.log_line(format!(
1099 "[cfg] cfg 缺席,{ns} 直写托底(软集中负形)"
1100 ));
1101 ek_cfg::cfg_save_at(store.root(), &ns, &map).map_err(|e| e.to_string())
1102 }
1103 Err(e) => Err(format!("{e:?}")),
1104 };
1105 match wr {
1106 Ok(()) => {
1107 shared.log_line(format!("[cfg] {ns} 已重存(rev {})", map.rev))
1108 }
1109 Err(e) => shared.log_line(format!("[cfg] {ns} 重存失败:{e}")),
1110 }
1111 }
1112 Err(msg) => shared.log_line(format!("[cfg] {msg}")),
1113 }
1114 }
1115 }
1116 }
1117 }
1118}
1119
1120impl Plugin for UiPlugin {
1121 fn name(&self) -> &str {
1122 UI_NAME
1123 }
1124
1125 fn deps(&self) -> &[&str] {
1126 &["cfg"]
1129 }
1130
1131 fn routes(&self) -> Vec<std::borrow::Cow<'static, str>> {
1132 vec![
1133 std::borrow::Cow::Borrowed(EVENT_LOG),
1134 std::borrow::Cow::Borrowed(EVENT_PLUG_UP),
1135 std::borrow::Cow::Borrowed(EVENT_PLUG_DOWN),
1136 std::borrow::Cow::Borrowed(EVENT_BAR),
1137 std::borrow::Cow::Borrowed(EVENT_BAR_UPDATE),
1138 std::borrow::Cow::Borrowed(EVENT_TICK),
1139 std::borrow::Cow::Borrowed(EVENT_CONTRACT),
1140 std::borrow::Cow::Borrowed(ek_ctp::EVENT_CONTRACT_FINISH),
1141 std::borrow::Cow::Borrowed(W_ORDER),
1145 std::borrow::Cow::Borrowed(W_TRADE),
1146 std::borrow::Cow::Borrowed(W_POSITION),
1147 std::borrow::Cow::Borrowed(W_ACCOUNT),
1148 std::borrow::Cow::Borrowed(EVENT_TIMER),
1149 std::borrow::Cow::Borrowed(ek_ctp::EVENT_STATUS_PREFIX),
1151 ]
1152 }
1153
1154 fn mailbox_soft_cap(&self) -> u64 {
1155 UI_MAILBOX_SOFT_CAP
1156 }
1157
1158 fn version(&self) -> std::borrow::Cow<'static, str> {
1159 std::borrow::Cow::Borrowed(env!("CARGO_PKG_VERSION"))
1160 }
1161
1162 fn services(&self) -> Vec<ServiceDesc> {
1163 Vec::new() }
1165
1166 fn on_init(&self, ctx: &PluginContext) {
1167 let _ = self.ctx.set(ctx.clone());
1168 }
1169
1170 fn on_start(&self) {
1171 self.shared.log_line("[ui] 界面插件上线");
1172 self.spawn_cmd_thread();
1176 self.shared.request_trade_reseed();
1179 self.shared.request_pipeline_refresh();
1180 let theme = load_theme();
1182 self.shared.set_theme(theme);
1183 if theme == crate::tokens::Theme::Light {
1184 self.shared.log_line("[ui] 主题:浅色(读盘 ui.theme)");
1185 }
1186 }
1187
1188 fn on_stop(&self) {
1189 self.shared.log_line("[ui] 界面插件下线");
1190 self.cmd_stop
1192 .store(true, std::sync::atomic::Ordering::Release);
1193 self.flush_persist();
1197 }
1198
1199 fn on_mail(&self, event: &ek_event::Event) {
1200 let ty = event.event_type.as_str();
1201 if ty == EVENT_LOG {
1202 if let Some(d) = event.data.downcast_ref::<LogData>() {
1203 let tag = if d.is_warning_or_above() {
1204 "WARN"
1205 } else {
1206 "INFO"
1207 };
1208 self.shared
1209 .log_line(format!("[{tag}][{}] {}", d.gateway_name, d.msg));
1210 if let Some(note) = d.msg.strip_prefix(crate::state::CHAIN_STATE_PREFIX) {
1216 self.shared.note_chain(&d.gateway_name, note);
1217 }
1218 } else {
1219 self.shared.log_line("[eLog] 载荷类型异常(非 LogData)");
1220 }
1221 return;
1222 }
1223 if ty == ek_ctp::EVENT_STATUS_PREFIX {
1224 if let Some(ev) = event.data.downcast_ref::<ek_ctp::GatewayStatusEvent>() {
1226 self.shared.note_status_event(ev);
1227 } else {
1228 self.shared
1229 .log_line("[eStatus] 载荷类型异常(非 GatewayStatusEvent)");
1230 }
1231 return;
1232 }
1233 if ty == EVENT_PLUG_UP || ty == EVENT_PLUG_DOWN {
1234 if let Some(n) = event.data.downcast_ref::<PlugNotice>() {
1235 let verb = if ty == EVENT_PLUG_UP {
1236 "上线"
1237 } else {
1238 "下线"
1239 };
1240 self.shared
1241 .log_line(format!("[host] 插件{verb} {} v{}", n.name, n.version));
1242 self.shared.request_list_refresh();
1244 if ty == EVENT_PLUG_UP {
1245 if n.name == ek_oms::NAME {
1247 self.shared.request_pipeline_refresh();
1248 }
1249 let is_account = self.shared.accounts().iter().any(|a| a.gateway == n.name);
1254 if n.name == ek_oms::NAME || is_account {
1255 self.shared.request_trade_reseed();
1256 }
1257 if is_account {
1260 self.shared.request_status_seed(&n.name);
1261 }
1262 let md = self.shared.md_source();
1265 let subs = self.shared.subscriptions();
1266 if !md.is_empty() && n.name == md && !subs.is_empty() {
1267 self.shared.push_cmd(crate::state::UiCmd::CtpSubscribe {
1268 gateway: n.name.clone(),
1269 symbols: subs.clone(),
1270 });
1271 self.shared.log_line(format!(
1272 "[ops] 行情源 {} 上线,重放订阅 {} 项",
1273 n.name,
1274 subs.len()
1275 ));
1276 }
1277 } else {
1278 let removed = self.shared.clear_gateway_rows(&n.name);
1282 if removed > 0 {
1283 self.shared.log_line(format!(
1284 "[ui] 插件下线 {}:作废 {} 行交易数据",
1285 n.name, removed
1286 ));
1287 }
1288 if n.name == ek_oms::NAME {
1293 self.shared.request_pipeline_refresh();
1294 }
1295 }
1296 }
1297 self.shared.bump();
1298 return;
1299 }
1300 if ty == EVENT_BAR {
1301 if let Some(f) = event.data.downcast_ref::<ek_dm::wire::BarsFrame>() {
1302 self.shared.market().on_frame(f);
1303 }
1304 self.shared.bump();
1305 return;
1306 }
1307 if ty == EVENT_BAR_UPDATE {
1308 if let Some(b) = event.data.downcast_ref::<ek_data::bar::BarData>() {
1309 self.shared.market().on_update(b);
1310 }
1311 self.shared.bump();
1312 return;
1313 }
1314 if ty == EVENT_TICK {
1315 if let Some(t) = event.data.downcast_ref::<ek_data::tick::TickData>() {
1316 self.shared.update_tick(t);
1317 }
1318 return;
1319 }
1320 if ty == EVENT_CONTRACT {
1321 if let Some(v) = event
1324 .data
1325 .downcast_ref::<Vec<ek_data::contract::ContractData>>()
1326 {
1327 for c in v {
1328 self.shared.update_contract(c);
1329 }
1330 } else if let Some(c) = event.data.downcast_ref::<ek_data::contract::ContractData>() {
1331 self.shared.update_contract(c);
1332 }
1333 return;
1334 }
1335 if ty == ek_ctp::EVENT_CONTRACT_FINISH {
1336 let gw = event
1337 .data
1338 .downcast_ref::<String>()
1339 .cloned()
1340 .unwrap_or_else(|| "?".to_string());
1341 self.shared
1342 .log_line(format!("[ctp] 合约流完成 {gw}(查询合约窗可勾选订阅)"));
1343 return;
1344 }
1345 if ty == W_ORDER {
1348 match event.data.downcast_ref::<OrderData>() {
1349 Some(o) => self.shared.note_order(o),
1350 None => self.shared.note_bad_payload(W_ORDER),
1351 }
1352 return;
1353 }
1354 if ty == W_TRADE {
1355 match event.data.downcast_ref::<TradeData>() {
1356 Some(t) => self.shared.note_trade(t),
1357 None => self.shared.note_bad_payload(W_TRADE),
1358 }
1359 return;
1360 }
1361 if ty == W_POSITION {
1362 match event.data.downcast_ref::<PositionData>() {
1363 Some(p) => self.shared.note_position(p),
1364 None => self.shared.note_bad_payload(W_POSITION),
1365 }
1366 return;
1367 }
1368 if ty == W_ACCOUNT {
1369 match event.data.downcast_ref::<AccountData>() {
1370 Some(a) => self.shared.note_funds(a),
1371 None => self.shared.note_bad_payload(W_ACCOUNT),
1372 }
1373 return;
1374 }
1375 if ty == EVENT_TIMER {
1376 self.refresh();
1377 }
1378 }
1379}