1use std::collections::{BTreeMap, HashSet, VecDeque};
7use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, AtomicUsize, Ordering};
8use std::sync::{Arc, Mutex};
9
10use ek_data::bar::BarData;
11use ek_dm::wire::BarsFrame;
12use ek_station::wire::PlugRow;
13
14pub const RING_CAP: usize = 4096;
19
20pub(crate) fn beijing_tz() -> chrono::FixedOffset {
25 chrono::FixedOffset::east_opt(8 * 3600).expect("+08 offset is always valid")
26}
27
28#[derive(Debug, Clone)]
32pub struct LogEntry {
33 pub at: String,
35 pub text: String,
37}
38
39#[derive(Debug)]
43pub struct LogRing {
44 lines: VecDeque<LogEntry>,
45 cap: usize,
46 spilled: u64,
47 spill_dir: Option<std::path::PathBuf>,
48}
49
50pub(crate) fn default_spill_dir() -> std::path::PathBuf {
52 if let Ok(d) = std::env::var("EK_UI_LOG_DIR")
53 && !d.trim().is_empty()
54 {
55 return std::path::PathBuf::from(d);
56 }
57 std::env::var("APPDATA")
58 .map(|r| std::path::PathBuf::from(r).join("ek").join("logs"))
59 .unwrap_or_else(|_| std::path::PathBuf::from(".ek-logs"))
60}
61
62impl Default for LogRing {
63 fn default() -> Self {
64 Self::new_spilling(RING_CAP, default_spill_dir())
65 }
66}
67
68impl LogRing {
69 pub fn new(cap: usize) -> Self {
71 Self {
72 lines: VecDeque::with_capacity(cap),
73 cap,
74 spilled: 0,
75 spill_dir: None,
76 }
77 }
78
79 pub fn new_spilling(cap: usize, dir: std::path::PathBuf) -> Self {
83 Self {
84 lines: VecDeque::with_capacity(cap),
85 cap,
86 spilled: 0,
87 spill_dir: Some(dir),
88 }
89 }
90
91 fn spill_write(&self, entry: &LogEntry) {
93 let Some(dir) = self.spill_dir.as_ref() else {
94 return;
95 };
96 let day = chrono::Local::now().format("%Y%m%d");
97 let path = dir.join(format!("ekapp-{day}.log"));
98 let _ = std::fs::create_dir_all(dir);
99 if let Ok(mut f) = std::fs::OpenOptions::new()
100 .append(true)
101 .create(true)
102 .open(&path)
103 {
104 use std::io::Write as _;
105 let _ = writeln!(f, "{} | {}", entry.at, entry.text);
106 }
107 }
108
109 pub fn push(&mut self, line: impl Into<String>) {
111 let at = chrono::Utc::now()
112 .with_timezone(&beijing_tz())
113 .format("%H:%M:%S")
114 .to_string();
115 let entry = LogEntry {
116 at,
117 text: line.into(),
118 };
119 if self.spill_dir.is_some() {
120 self.spill_write(&entry);
121 }
122 if self.cap == 0 {
123 self.spilled += 1;
124 return;
125 }
126 if self.lines.len() == self.cap {
127 self.lines.pop_front();
128 self.spilled += 1;
129 }
130 self.lines.push_back(entry);
131 }
132
133 pub fn len(&self) -> usize {
135 self.lines.len()
136 }
137
138 pub fn is_empty(&self) -> bool {
140 self.lines.is_empty()
141 }
142
143 pub fn spilled(&self) -> u64 {
145 self.spilled
146 }
147
148 pub fn snapshot(&self) -> Vec<String> {
150 self.lines.iter().map(|e| e.text.clone()).collect()
151 }
152
153 pub fn entries(&self) -> Vec<LogEntry> {
155 self.lines.iter().cloned().collect()
156 }
157
158 pub fn clear(&mut self) {
160 self.lines.clear();
161 }
162}
163
164#[derive(Debug, Clone, Default)]
166pub struct BusSnap {
167 pub rows: Vec<PlugRow>,
169 pub services: Vec<(String, Vec<String>)>,
171}
172
173#[derive(Debug, Clone, PartialEq)]
175pub enum UiCmd {
176 Start { factory: String, name: String },
179 Remove { name: String },
181 Reload {
183 interval: i64,
184 symbol: String,
185 bars: Vec<BarData>,
186 },
187 SendOrder {
193 req: ek_data::order_request::OrderRequest,
194 gateway: String,
195 mode: ek_oms::LockMode,
196 },
197 CancelOrder { orderid: String, gateway: String },
201 CtpConfigure { gateway: String, form: CtpForm },
205 Connect { gateway: String, form: CtpForm },
210 CtpSubscribe {
213 gateway: String,
214 symbols: Vec<String>,
215 },
216 Disconnect { gateway: String },
220 CfgList,
222 CfgDelete { ns: String },
224 CfgResave { ns: String },
227
228 StratAdd {
231 class_name: String,
233 name: String,
235 params: Vec<(String, String)>,
237 },
238 StratInit { name: String },
240 StratStart { name: String },
242 StratPause { name: String },
244 StratResume { name: String },
246 StratStop { name: String },
248 StratEdit {
250 name: String,
252 params: Vec<(String, String)>,
254 },
255 StratRemove { name: String },
257 StratStartAll,
259 StratStopAll,
261 StratResync,
263 StratSettings { name: String },
265 StratPrice { symbol: String },
267 StratMetaSync,
269 StratSetEngine {
271 trade_gateway: String,
273 md_gateway: String,
275 },
276}
277
278#[derive(Debug, Clone, PartialEq, Eq)]
283pub struct StratFeedback {
284 pub op: &'static str,
286 pub name: String,
288 pub ok: bool,
290 pub msg: String,
292}
293
294pub const KLINE_CELL_CAP: usize = 4096;
296
297pub const MAX_ACCOUNTS: usize = 64;
302
303pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
308
309pub const ACCOUNT_STALE_SECS: i64 = 30;
315
316pub use ek_ctp::driver::CHAIN_STATE_PREFIX;
322
323pub const LIST_POLL_TICKS: u32 = 4;
329
330pub const STATUS_ASKS_PER_TICK: usize = 2;
335
336pub const TRADE_CAP: usize = 5000;
341
342pub const PENDING_WARN_FLOOR: u64 = 4_096;
344pub const PENDING_CRIT_FLOOR: u64 = 32_768;
346
347#[must_use]
355pub fn pending_thresholds(soft_cap: u64) -> (u64, u64) {
356 let warn = (soft_cap / 16).max(PENDING_WARN_FLOOR);
357 let crit = (soft_cap / 4).max(PENDING_CRIT_FLOOR);
358 (warn.min(soft_cap.max(1)), crit.min(soft_cap.max(1)))
359}
360
361#[derive(Debug, Clone, Copy, PartialEq, Eq)]
363pub enum ConfirmStep {
364 Armed,
365 Fired,
366 Switched,
367}
368
369#[must_use]
372pub fn is_quarantined(state: &str) -> bool {
373 state.contains("Quarantin")
374}
375
376#[derive(Debug, Clone)]
380pub struct KlineFetch {
381 pub symbol: String,
382 pub interval_secs: i64,
383}
384
385#[derive(Debug, Default)]
387pub struct MarketStore {
388 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
389 frames: AtomicU64,
391 updates: AtomicU64,
393 reloads: AtomicU64,
395}
396
397#[derive(Debug, Clone, PartialEq)]
399pub struct CellView {
400 pub symbol: String,
401 pub interval: i64,
402 pub bars: Vec<BarData>,
403}
404
405impl MarketStore {
406 pub fn on_frame(&self, frame: &BarsFrame) {
408 let mut cells = self.cells.lock().unwrap();
409 for b in &frame.bars {
410 let key = (b.symbol.clone(), frame.interval);
411 let cell = cells.entry(key).or_default();
412 match cell.last() {
413 Some(last) if b.datetime < last.datetime => continue,
414 Some(last) if b.datetime == last.datetime => {
415 let slot = cell.last_mut().expect("len>1");
416 *slot = b.clone();
417 }
418 _ => {
419 cell.push(b.clone());
420 if cell.len() > KLINE_CELL_CAP {
421 cell.remove(0);
422 }
423 }
424 }
425 }
426 self.frames.fetch_add(1, Ordering::AcqRel);
427 }
428
429 pub fn on_update(&self, b: &BarData) {
431 let mut cells = self.cells.lock().unwrap();
432 let iv = b
433 .interval
434 .unwrap_or(ek_data::constant::Interval::MINUTE)
435 .seconds();
436 let key = (b.symbol.clone(), iv);
437 let cell = cells.entry(key).or_default();
438 match cell.last() {
439 Some(last) if b.datetime == last.datetime => {
440 let slot = cell.last_mut().expect("len>1");
441 *slot = b.clone();
442 }
443 Some(last) if b.datetime < last.datetime => {}
444 _ => {
445 cell.push(b.clone());
446 if cell.len() > KLINE_CELL_CAP {
447 cell.remove(0);
448 }
449 }
450 }
451 self.updates.fetch_add(1, Ordering::AcqRel);
452 }
453
454 pub fn note_history(&self, symbol: &str, iv_secs: i64, bars: &[BarData]) {
460 let mut cells = self.cells.lock().unwrap();
461 let cell = cells.entry((symbol.to_string(), iv_secs)).or_default();
462 cell.extend(bars.iter().cloned());
463 cell.sort_unstable_by_key(|b| b.datetime);
464 cell.dedup_by(|a, b| a.datetime == b.datetime);
465 if cell.len() > KLINE_CELL_CAP {
466 let excess = cell.len() - KLINE_CELL_CAP;
467 cell.drain(0..excess);
468 }
469 self.frames.fetch_add(1, Ordering::AcqRel);
470 }
471
472 pub fn note_reload(&self) {
474 self.reloads.fetch_add(1, Ordering::AcqRel);
475 }
476
477 pub fn cells(&self) -> Vec<CellView> {
479 self.cells
480 .lock()
481 .unwrap()
482 .iter()
483 .map(|((sym, iv), bars)| CellView {
484 symbol: sym.clone(),
485 interval: *iv,
486 bars: bars.clone(),
487 })
488 .collect()
489 }
490
491 pub fn find_cell(&self, sym: &str) -> Option<CellView> {
496 self.cells()
497 .into_iter()
498 .find(|c| c.symbol == sym || c.symbol.ends_with(&format!(".{sym}")))
499 }
500
501 pub fn counts(&self) -> (u64, u64, u64) {
503 (
504 self.frames.load(Ordering::Acquire),
505 self.updates.load(Ordering::Acquire),
506 self.reloads.load(Ordering::Acquire),
507 )
508 }
509}
510
511#[derive(Debug, Default, Clone, PartialEq)]
514pub struct ReloadForm {
515 pub interval: String,
516 pub symbol: String,
517 pub csv: String,
518}
519
520pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
527 let g = raw.trim();
528 if !g.is_empty() {
529 if g.chars().any(|c| c.is_whitespace()) {
530 return Err(format!("网关解析: 含空白:{raw:?}"));
531 }
532 return Ok(g.to_string());
533 }
534 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
535 Some(f) => Ok(f.to_string()),
536 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
537 }
538}
539
540#[derive(Debug, Clone, PartialEq)]
543pub struct CtpForm {
544 pub broker_id: String,
545 pub td_addr: String,
546 pub td_user: String,
547 pub td_password: String,
548 pub td_auth_code: String,
549 pub td_app_id: String,
550 pub md_addr: String,
551 pub md_user: String,
552 pub md_password: String,
553 pub symbols: String,
555 pub accept_futures: bool,
558 pub accept_options: bool,
559}
560
561impl Default for CtpForm {
564 fn default() -> Self {
565 Self {
566 broker_id: String::new(),
567 td_addr: String::new(),
568 td_user: String::new(),
569 td_password: String::new(),
570 td_auth_code: String::new(),
571 td_app_id: String::new(),
572 md_addr: String::new(),
573 md_user: String::new(),
574 md_password: String::new(),
575 symbols: String::new(),
576 accept_futures: true,
577 accept_options: false,
578 }
579 }
580}
581
582impl CtpForm {
583 #[must_use]
587 pub fn product_classes(&self) -> Vec<String> {
588 let mut codes = Vec::new();
589 if self.accept_futures {
590 codes.push("1".to_string());
591 }
592 if self.accept_options {
593 codes.push("2".to_string());
594 }
595 codes
596 }
597
598 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
600 ek_ctp::plugin::ConfigureReq {
601 broker_id: self.broker_id.trim().to_string(),
602 td_addr: self.td_addr.trim().to_string(),
603 td_user: self.td_user.trim().to_string(),
604 td_password: self.td_password.clone().into(),
605 td_auth_code: self.td_auth_code.trim().to_string(),
606 td_app_id: self.td_app_id.trim().to_string(),
607 md_addr: self.md_addr.trim().to_string(),
608 md_user: self.md_user.trim().to_string(),
609 md_password: self.md_password.clone().into(),
610 product_classes: self.product_classes(),
611 }
612 }
613
614 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
616 let list: Vec<String> = self
617 .symbols
618 .split([',', ',', ' '])
619 .map(|s| s.trim().to_string())
620 .filter(|s| !s.is_empty())
621 .collect();
622 if list.is_empty() {
623 return Err("订阅解析: 标的位空".to_string());
624 }
625 Ok(list)
626 }
627
628 pub fn validate(&self) -> Result<(), String> {
637 fn front(name: &str, v: &str) -> Result<(), String> {
638 let v = v.trim();
639 let Some(rest) = v.strip_prefix("tcp://") else {
640 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
641 };
642 let Some((host, port)) = rest.rsplit_once(':') else {
643 return Err(format!("{name} 缺端口:{name} {:?}", v));
644 };
645 if host.trim().is_empty() {
646 return Err(format!("{name} 主机位空:{}", v));
647 }
648 let port: u32 = port
649 .parse()
650 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
651 if port == 0 || port > 65535 {
652 return Err(format!("{name} 端口越界(1..=65535):{}", v));
653 }
654 Ok(())
655 }
656 if self.broker_id.trim().is_empty() {
657 return Err("broker 位空".to_string());
658 }
659 if self.td_user.trim().is_empty() {
660 return Err("交易账号位空".to_string());
661 }
662 front("td 前置", &self.td_addr)?;
663 if !self.md_addr.trim().is_empty() {
664 front("md 前置", &self.md_addr)?;
665 }
666 if self.td_app_id.trim().is_empty() {
667 return Err("AppID 位空(CTP 登录必需)".to_string());
668 }
669 if self.td_auth_code.trim().is_empty() {
670 return Err("AuthCode 位空(CTP 登录必需)".to_string());
671 }
672 Ok(())
673 }
674}
675
676#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
683pub enum AccountStage {
684 #[default]
686 NotAssembled,
687 Assembling,
689 Assembled,
691 AssembleFailed,
693}
694
695impl AccountStage {
696 #[must_use]
698 pub fn label(self) -> &'static str {
699 match self {
700 AccountStage::NotAssembled => "未装配",
701 AccountStage::Assembling => "装配中…",
702 AccountStage::Assembled => "已装配·登录态未知",
705 AccountStage::AssembleFailed => "装配失败",
706 }
707 }
708}
709
710#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
715pub enum LoginState {
716 #[default]
721 Unknown,
722 Online,
724 LoginFailed,
726 FrontDown,
728 QueryStarved,
730 Stopped,
732}
733
734impl LoginState {
735 #[must_use]
737 pub fn label(self) -> &'static str {
738 match self {
739 LoginState::Unknown => "登录态未知",
740 LoginState::Online => "在线",
741 LoginState::LoginFailed => "登录失败",
742 LoginState::FrontDown => "掉线",
743 LoginState::QueryStarved => "查询断供",
744 LoginState::Stopped => "已断开",
745 }
746 }
747
748 #[must_use]
750 pub fn is_alarm(self) -> bool {
751 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
752 }
753}
754
755fn count_note(label: &str, n: u32) -> Option<String> {
757 (n > 0).then(|| format!("{label} {n} 次"))
758}
759
760fn brief_counter_reason(full: &str) -> Option<String> {
767 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
768 let body = tail.trim().trim_start_matches("CTP:").trim();
769 if body.is_empty() {
770 return None;
771 }
772 let code = full
773 .rfind("代码:")
774 .map(|i| {
775 let rest = &full[i + "代码:".len()..];
776 let end = rest.find(',').unwrap_or(rest.len());
777 rest[..end].trim().to_string()
778 })
779 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
780 Some(match code {
781 Some(c) => format!("{body}({c})"),
782 None => body.to_string(),
783 })
784}
785
786#[derive(Debug, Clone, PartialEq, Default)]
792pub struct AccountRow {
793 pub gateway: String,
795 pub alias: String,
797 pub broker_id: String,
798 pub account: String,
799 pub md_source: bool,
801 pub form: CtpForm,
803 pub restored: bool,
809 pub auto_connect: bool,
814 pub password_ref: bool,
819 pub stage: AccountStage,
821 pub fail_note: Option<String>,
823 pub effective_classes: Vec<String>,
828 pub td_anomaly: u32,
835 pub md_anomaly: u32,
837 pub connected: bool,
840 pub login: LoginState,
844 pub status_note: Option<String>,
847 pub login_brief: Option<String>,
854 pub chain_note: Option<String>,
862}
863
864impl AccountRow {
865 pub fn derive(form: &CtpForm) -> Result<Self, String> {
867 Self::derive_with(form, &[])
868 }
869
870 pub fn derive_with(
880 form: &CtpForm,
881 taken: &[(String, String, String, String)],
882 ) -> Result<Self, String> {
883 let broker = form.broker_id.trim();
884 let account = form.td_user.trim();
885 let gateway =
886 crate::appconfig::unique_gw_name(taken, broker, account, form.td_addr.trim())?;
887 Ok(Self {
888 gateway,
889 alias: String::new(),
890 broker_id: broker.to_string(),
891 account: account.to_string(),
892 md_source: false,
893 form: form.clone(),
894 stage: AccountStage::NotAssembled,
895 fail_note: None,
896 effective_classes: Vec::new(),
897 td_anomaly: 0,
898 md_anomaly: 0,
899 connected: false,
900 login: LoginState::Unknown,
901 status_note: None,
902 login_brief: None,
903 chain_note: None,
904 restored: false,
907 auto_connect: false,
908 password_ref: false,
909 })
910 }
911
912 #[must_use]
918 pub fn status_label(&self) -> String {
919 if self.stage == AccountStage::Assembled {
920 match self.login {
921 LoginState::Unknown => "已装配·登录态未知".to_string(),
922 other => format!("已装配·{}", other.label()),
923 }
924 } else {
925 self.stage.label().to_string()
926 }
927 }
928
929 pub fn apply_status(
942 &mut self,
943 st: &ek_ctp::plugin::GatewayStatus,
944 now_ms: i64,
945 reject_limit: u32,
946 stale_secs: i64,
947 ) {
948 self.td_anomaly = st.td_anomalies.total;
950 self.md_anomaly = st.md_anomalies.total;
951 let td_side = !self.form.td_addr.trim().is_empty();
952 let (login, note) = if td_side {
953 if st.td_login_failed && !st.td_logged_in {
954 (
955 LoginState::LoginFailed,
956 st.td_last_error
957 .clone()
958 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
959 )
960 } else if st.td_logged_in {
961 if st.td_query_reject_streak >= reject_limit {
962 (
963 LoginState::QueryStarved,
964 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
965 )
966 } else if let Some(last) = st.td_last_account_ms {
967 let idle_ms = now_ms.saturating_sub(last);
968 if idle_ms > stale_secs.saturating_mul(1_000) {
969 (
970 LoginState::QueryStarved,
971 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
972 )
973 } else {
974 (LoginState::Online, count_note("掉线", st.td_disconnects))
975 }
976 } else {
977 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
984 match since_login_ms {
985 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
986 LoginState::QueryStarved,
987 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
988 ),
989 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
990 }
991 }
992 } else if self.login == LoginState::Stopped {
993 let extra = if st.td_disconnects > 0 {
994 format!("(此前掉线 {} 次)", st.td_disconnects)
995 } else {
996 String::new()
997 };
998 (
999 LoginState::Stopped,
1000 Some(format!("会话已停{extra};『连接』可恢复")),
1001 )
1002 } else if st.td_disconnects > 0 {
1003 (
1004 LoginState::FrontDown,
1005 Some(format!("掉线 {} 次", st.td_disconnects)),
1006 )
1007 } else {
1008 (LoginState::Unknown, st.td_last_error.clone())
1009 }
1010 } else if st.md_logged_in {
1011 (
1012 LoginState::Online,
1013 count_note("掉线", st.md_disconnects)
1014 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
1015 )
1016 } else if self.login == LoginState::Stopped {
1017 (
1018 LoginState::Stopped,
1019 Some("会话已停;『连接』可恢复".to_string()),
1020 )
1021 } else if st.md_disconnects > 0 {
1022 (
1023 LoginState::FrontDown,
1024 Some(format!("掉线 {} 次", st.md_disconnects)),
1025 )
1026 } else {
1027 (LoginState::Unknown, st.md_last_error.clone())
1028 };
1029 let note = if td_side && st.td_instrument_rejects > 0 {
1034 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
1035 match note {
1036 Some(base) if base.contains("合约查询被拒") => Some(base),
1038 Some(base) => Some(format!("{base};{add}")),
1039 None => Some(add),
1040 }
1041 } else {
1042 note
1043 };
1044 self.login = login;
1045 self.status_note = note;
1046 self.login_brief = if login.is_alarm() {
1049 st.td_last_error
1050 .as_deref()
1051 .or(st.md_last_error.as_deref())
1052 .and_then(brief_counter_reason)
1053 } else {
1054 None
1055 };
1056 }
1057
1058 #[must_use]
1060 pub fn display_name(&self) -> String {
1061 let a = self.alias.trim();
1062 if a.is_empty() {
1063 self.account.clone()
1064 } else {
1065 a.to_string()
1066 }
1067 }
1068}
1069
1070#[derive(Debug, Clone, PartialEq, Default)]
1073pub struct TradeSnap {
1074 pub orders: Vec<ek_data::order::OrderData>,
1075 pub trades: Vec<ek_data::trade::TradeData>,
1076 pub positions: Vec<ek_data::position::PositionData>,
1077 pub accounts: Vec<ek_data::account::AccountData>,
1079}
1080
1081pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
1084 let interval: i64 = form
1085 .interval
1086 .trim()
1087 .parse()
1088 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
1089 let symbol = form.symbol.trim().to_string();
1090 if symbol.is_empty() {
1091 return Err("回灌解析: symbol 空".to_string());
1092 }
1093 if form.csv.trim().is_empty() {
1094 return Err("回灌解析: csv 空".to_string());
1095 }
1096 let mut bars = Vec::new();
1097 for (i, raw) in form.csv.lines().enumerate() {
1098 let line = raw.trim();
1099 if line.is_empty() || line.starts_with('#') {
1100 continue;
1101 }
1102 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
1103 if parts.len() != 6 {
1104 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
1105 }
1106 let num = |s: &str| -> Result<f64, String> {
1107 s.parse::<f64>()
1108 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
1109 };
1110 let dt_s: i64 = parts[0]
1111 .parse()
1112 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
1113 bars.push(BarData {
1114 symbol: symbol.clone(),
1115 datetime: dt_s * 1_000_000_000,
1116 interval: Some(
1117 ek_dm::wire::interval_of_seconds(interval)
1118 .unwrap_or(ek_data::constant::Interval::MINUTE),
1119 ),
1120 open_price: num(parts[1])?,
1121 high_price: num(parts[2])?,
1122 low_price: num(parts[3])?,
1123 close_price: num(parts[4])?,
1124 volume: num(parts[5])?,
1125 gateway_name: String::new(),
1128 exchange: ek_data::constant::Exchange::UNKNOWN,
1130 turnover: 0.0,
1134 open_interest: 0.0,
1135 pct_change: 0.0,
1136 });
1137 }
1138 if bars.is_empty() {
1139 return Err("回灌解析: 无有效行".to_string());
1140 }
1141 Ok((interval, symbol, bars))
1142}
1143
1144#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1149pub enum TopLeftTab {
1150 #[default]
1151 Orders,
1152 Active,
1153}
1154
1155impl TopLeftTab {
1156 pub fn label(self) -> &'static str {
1157 match self {
1158 Self::Orders => "委托",
1159 Self::Active => "未成交",
1160 }
1161 }
1162}
1163
1164#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1166pub enum BottomRightTab {
1167 #[default]
1168 Account,
1169 Logs,
1170}
1171
1172impl BottomRightTab {
1173 pub fn label(self) -> &'static str {
1174 match self {
1175 Self::Account => "资金",
1176 Self::Logs => "日志",
1177 }
1178 }
1179}
1180
1181#[derive(Debug, Clone, Copy, PartialEq)]
1184pub struct DataSplit {
1185 pub col: f32,
1187 pub row: f32,
1189}
1190
1191impl Default for DataSplit {
1192 fn default() -> Self {
1193 Self {
1194 col: crate::splitters::DATA_SPLIT_DEFAULT,
1195 row: crate::splitters::DATA_SPLIT_DEFAULT,
1196 }
1197 }
1198}
1199
1200pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
1203 ek_data::constant::Exchange::SHFE,
1204 ek_data::constant::Exchange::DCE,
1205 ek_data::constant::Exchange::CZCE,
1206 ek_data::constant::Exchange::CFFEX,
1207 ek_data::constant::Exchange::INE,
1208 ek_data::constant::Exchange::GFEX,
1209];
1210
1211#[derive(Debug, Default)]
1221pub struct TickStore {
1222 ticks: BTreeMap<String, ek_data::tick::TickData>,
1223}
1224
1225impl TickStore {
1226 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1229 self.ticks
1230 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1231 }
1232
1233 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1235 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1236 v.sort_by(|a, b| {
1237 (a.exchange.as_str().to_string(), a.symbol.clone())
1238 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1239 });
1240 v
1241 }
1242
1243 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1245 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1246 for t in self.rows() {
1247 let ex = t.exchange.as_str().to_string();
1248 match out.last_mut() {
1249 Some((e, v)) if *e == ex => v.push(t),
1250 _ => out.push((ex, vec![t])),
1251 }
1252 }
1253 out
1254 }
1255
1256 pub fn len(&self) -> usize {
1257 self.ticks.len()
1258 }
1259
1260 pub fn is_empty(&self) -> bool {
1261 self.ticks.is_empty()
1262 }
1263}
1264
1265#[derive(Debug, Clone, PartialEq)]
1267pub struct OrderForm {
1268 pub exchange: String,
1270 pub symbol: String,
1272 pub direction: ek_data::constant::Direction,
1273 pub offset: ek_data::constant::Offset,
1274 pub order_type: ek_data::constant::OrderType,
1275 pub price: String,
1277 pub volume: String,
1278 pub gateway: String,
1280 pub follow_price: bool,
1282 pub auto_split: bool,
1285}
1286
1287impl Default for OrderForm {
1288 fn default() -> Self {
1289 Self {
1290 exchange: "SHFE".to_string(),
1291 symbol: String::new(),
1292 direction: ek_data::constant::Direction::LONG,
1293 offset: ek_data::constant::Offset::OPEN,
1294 order_type: ek_data::constant::OrderType::LIMIT,
1295 price: String::new(),
1296 volume: "1".to_string(),
1297 gateway: String::new(),
1298 follow_price: true,
1299 auto_split: true,
1300 }
1301 }
1302}
1303
1304impl OrderForm {
1305 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1307 use std::str::FromStr;
1308 let symbol = self.symbol.trim();
1309 if symbol.is_empty() {
1310 return Err("下单: 代码为空".to_string());
1311 }
1312 if symbol.contains('.') {
1313 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1314 }
1315 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1316 .map_err(|e| format!("下单: {e}"))?;
1317 let price: f64 = self
1318 .price
1319 .trim()
1320 .parse()
1321 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1322 let volume: f64 = self
1323 .volume
1324 .trim()
1325 .parse()
1326 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1327 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1328 return Err("下单: 限价单价格须为正".to_string());
1329 }
1330 if volume <= 0.0 {
1331 return Err("下单: 数量须为正".to_string());
1332 }
1333 Ok(ek_data::order_request::OrderRequest {
1334 symbol: symbol.to_string(),
1335 exchange,
1336 direction: self.direction,
1337 order_type: self.order_type,
1338 volume,
1339 price,
1340 offset: self.offset,
1341 ..ek_data::order_request::OrderRequest::default()
1342 })
1343 }
1344}
1345
1346#[derive(Debug, Default)]
1352pub struct ContractStore {
1353 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1354}
1355
1356impl ContractStore {
1357 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1358 self.contracts
1359 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1360 }
1361
1362 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1364 let p = pattern.trim();
1365 self.contracts
1366 .values()
1367 .filter(|c| {
1368 p.is_empty()
1369 || c.symbol.contains(p)
1370 || c.exchange.as_str().eq_ignore_ascii_case(p)
1371 || c.name.contains(p)
1372 })
1373 .cloned()
1374 .collect()
1375 }
1376
1377 pub fn len(&self) -> usize {
1378 self.contracts.len()
1379 }
1380
1381 pub fn is_empty(&self) -> bool {
1382 self.contracts.is_empty()
1383 }
1384}
1385
1386#[derive(Debug, Default)]
1388pub struct UiShared {
1389 setup_phase: std::sync::atomic::AtomicBool,
1392 store_err: Mutex<Option<String>>,
1394 window_visible: Mutex<HashSet<String>>,
1396 logs: Mutex<LogRing>,
1397 snap: Mutex<Arc<BusSnap>>,
1398 market: MarketStore,
1399 ctp_form: Mutex<CtpForm>,
1400 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1403 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1404 trade_ids: Mutex<HashSet<String>>,
1412 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1413 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1414 trade_seq: AtomicU64,
1416 trade_dropped: AtomicU64,
1418 trade_dirty: AtomicBool,
1420 trade_snap: Mutex<Arc<TradeSnap>>,
1422 tick_store: Mutex<TickStore>,
1423 order_form: Mutex<OrderForm>,
1424 data_split: Mutex<DataSplit>,
1427 selected_exchange: Mutex<String>,
1429 topleft_tab: Mutex<TopLeftTab>,
1431 bottomright_tab: Mutex<BottomRightTab>,
1433 ops_window: AtomicBool,
1434 contracts_window: AtomicBool,
1439 contract_store: Mutex<ContractStore>,
1440 contract_query: Mutex<String>,
1441 contract_selected: Mutex<Vec<String>>,
1442 kline_windows: Mutex<Vec<String>>,
1443 kline_fetches: Mutex<Vec<KlineFetch>>,
1445 kline_fetched: Mutex<HashSet<(String, i64)>>,
1447 rev: AtomicU64,
1448 stop: AtomicBool,
1450 cmds: Mutex<VecDeque<UiCmd>>,
1452 pending_confirm: Mutex<Option<String>>,
1454 accounts: Mutex<Vec<AccountRow>>,
1456 account_edit_window: AtomicBool,
1458 editing_account: Mutex<Option<String>>,
1460 edit_alias: Mutex<String>,
1462 edit_error: Mutex<Option<String>>,
1464 md_source: Mutex<String>,
1466 subscriptions: Mutex<Vec<String>>,
1468 row_sel: Mutex<BTreeMap<String, String>>,
1471 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1475 status_cursor: AtomicUsize,
1481 status_seed: Mutex<Vec<String>>,
1484 list_tick: AtomicU32,
1486 list_due: AtomicBool,
1490 seed_due: AtomicBool,
1494 pipeline_due: AtomicBool,
1496 cfg_due: AtomicBool,
1498 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1501 cfg_delete_arm: Mutex<Option<String>>,
1503 theme: AtomicU8,
1507 theme_persist_due: AtomicBool,
1510 auto_connect_persist_due: AtomicBool,
1514 alias_persist_due: AtomicBool,
1518 archive_purge: Mutex<Vec<String>>,
1522 bad_payload: Mutex<BTreeMap<String, u64>>,
1524 pipeline_lines: Mutex<Vec<String>>,
1527 status_thresholds: Mutex<Option<(u32, i64)>>,
1530 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1532 toasts: crate::toast::ToastsCell,
1535 alerted: Mutex<std::collections::BTreeSet<String>>,
1537 exit_confirm_open: AtomicBool,
1539 exit_confirmed: AtomicBool,
1541 assembly_locked: AtomicBool,
1544 engine: Mutex<std::sync::Weak<ek_core::main_engine::MainEngine>>,
1548 modules: Mutex<crate::mount::Registry>,
1550 strategies: Mutex<BTreeMap<String, ek_strategy::contract::StrategyRow>>,
1554 strategy_metas: Mutex<Vec<ek_strategy::strategy::StrategyMeta>>,
1556 strategy_settings: Mutex<BTreeMap<String, Vec<(String, String)>>>,
1558 strategy_prices: Mutex<BTreeMap<String, f64>>,
1560 strategy_feedback: Mutex<std::collections::VecDeque<StratFeedback>>,
1562}
1563
1564impl UiShared {
1565 pub fn install_engine(&self, engine: &std::sync::Arc<ek_core::main_engine::MainEngine>) {
1569 *self.engine.lock().unwrap() = std::sync::Arc::downgrade(engine);
1570 }
1571
1572 #[must_use]
1574 pub fn engine_arc(&self) -> Option<std::sync::Arc<ek_core::main_engine::MainEngine>> {
1575 self.engine.lock().unwrap().upgrade()
1576 }
1577
1578 pub fn modules(&self) -> std::sync::MutexGuard<'_, crate::mount::Registry> {
1580 self.modules.lock().unwrap()
1581 }
1582
1583 pub fn apply_strategy_event(&self, ev: &ek_strategy::contract::StrategyEvent) {
1592 use ek_strategy::contract::StrategyEvent as Ev;
1593 {
1594 let mut map = self.strategies.lock().unwrap();
1595 match ev {
1596 Ev::Upsert(row) => {
1597 map.insert(row.name.clone(), (**row).clone());
1598 }
1599 Ev::Removed { name } => {
1600 map.remove(name);
1601 }
1602 Ev::Snapshot(rows) => {
1603 map.clear();
1604 for r in rows {
1605 map.insert(r.name.clone(), r.clone());
1606 }
1607 }
1608 }
1609 }
1610 self.bump();
1611 }
1612
1613 pub fn set_strategy_metas(&self, metas: Vec<ek_strategy::strategy::StrategyMeta>) {
1615 *self.strategy_metas.lock().unwrap() = metas;
1616 self.bump();
1617 }
1618
1619 #[must_use]
1621 pub fn strategy_metas(&self) -> Vec<ek_strategy::strategy::StrategyMeta> {
1622 self.strategy_metas.lock().unwrap().clone()
1623 }
1624
1625 #[must_use]
1627 pub fn strategy_rows(&self) -> Vec<ek_strategy::contract::StrategyRow> {
1628 self.strategies.lock().unwrap().values().cloned().collect()
1629 }
1630
1631 #[must_use]
1633 pub fn strategy_row(&self, name: &str) -> Option<ek_strategy::contract::StrategyRow> {
1634 self.strategies.lock().unwrap().get(name).cloned()
1635 }
1636
1637 #[must_use]
1639 pub fn strategy_count(&self) -> usize {
1640 self.strategies.lock().unwrap().len()
1641 }
1642
1643 pub fn set_strategy_settings(&self, name: &str, params: Vec<(String, String)>) {
1645 self.strategy_settings
1646 .lock()
1647 .unwrap()
1648 .insert(name.to_string(), params);
1649 self.bump();
1650 }
1651
1652 pub fn take_strategy_settings(&self, name: &str) -> Option<Vec<(String, String)>> {
1654 self.strategy_settings.lock().unwrap().remove(name)
1655 }
1656
1657 pub fn set_strategy_price(&self, symbol: &str, price: f64) {
1659 self.strategy_prices
1660 .lock()
1661 .unwrap()
1662 .insert(symbol.to_string(), price);
1663 self.bump();
1664 }
1665
1666 #[must_use]
1668 pub fn strategy_price(&self, symbol: &str) -> Option<f64> {
1669 self.strategy_prices.lock().unwrap().get(symbol).copied()
1670 }
1671
1672 pub const STRAT_FEEDBACK_CAP: usize = 32;
1677
1678 pub fn set_strategy_feedback(&self, fb: StratFeedback) {
1680 {
1681 let mut q = self.strategy_feedback.lock().unwrap();
1682 if q.len() >= Self::STRAT_FEEDBACK_CAP {
1683 q.pop_front();
1684 }
1685 q.push_back(fb);
1686 }
1687 self.bump();
1688 }
1689
1690 pub fn push_strategy_feedback_front(&self, fb: StratFeedback) {
1692 self.strategy_feedback.lock().unwrap().push_front(fb);
1693 self.bump();
1694 }
1695
1696 pub fn take_strategy_feedback(&self) -> Option<StratFeedback> {
1698 self.strategy_feedback.lock().unwrap().pop_front()
1699 }
1700
1701 pub fn set_store_err(&self, msg: Option<String>) {
1703 *self.store_err.lock().unwrap() = msg;
1704 self.bump();
1705 }
1706
1707 #[must_use]
1709 pub fn store_err(&self) -> Option<String> {
1710 self.store_err.lock().unwrap().clone()
1711 }
1712
1713 pub fn begin_window_frame(&self) {
1717 self.window_visible.lock().unwrap().clear();
1718 }
1719 pub fn note_window_visible(&self, id: &str) {
1721 self.window_visible.lock().unwrap().insert(id.to_string());
1722 }
1723 #[must_use]
1725 pub fn window_was_visible(&self, id: &str) -> bool {
1726 self.window_visible.lock().unwrap().contains(id)
1727 }
1728 pub fn set_setup_phase(&self, on: bool) {
1730 self.setup_phase
1731 .store(on, std::sync::atomic::Ordering::Release);
1732 self.bump();
1733 }
1734
1735 #[must_use]
1737 pub fn in_setup(&self) -> bool {
1738 self.setup_phase.load(std::sync::atomic::Ordering::Acquire)
1739 }
1740
1741 pub fn remount_modules(&self) {
1747 let groups = crate::mount::scan_groups(self);
1748 let mut reg = self.modules.lock().unwrap();
1749 reg.sync_groups(&groups);
1750 }
1751
1752 pub fn new() -> Self {
1754 Self {
1755 window_visible: Mutex::new(HashSet::new()),
1757 modules: Mutex::new(crate::mount::Registry::with_always()),
1758 ..Self::default()
1759 }
1760 }
1761
1762 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1766 self.tick_store.lock().unwrap().put(t);
1767 self.bump();
1768 }
1769
1770 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1772 self.tick_store.lock().unwrap().rows()
1773 }
1774
1775 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1777 self.tick_store.lock().unwrap().groups()
1778 }
1779
1780 pub fn tick_count(&self) -> usize {
1782 self.tick_store.lock().unwrap().len()
1783 }
1784
1785 pub fn order_form(&self) -> OrderForm {
1787 self.order_form.lock().unwrap().clone()
1788 }
1789
1790 pub fn set_order_form(&self, f: OrderForm) {
1792 *self.order_form.lock().unwrap() = f;
1793 self.bump();
1794 }
1795
1796 pub fn selected_exchange(&self) -> String {
1800 let s = self.selected_exchange.lock().unwrap().clone();
1801 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1802 s
1803 } else {
1804 MARKET_ORDER[0].as_str().to_string()
1805 }
1806 }
1807
1808 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1810 *self.selected_exchange.lock().unwrap() = s.into();
1811 self.bump();
1812 }
1813
1814 pub fn topleft_tab(&self) -> TopLeftTab {
1816 *self.topleft_tab.lock().unwrap()
1817 }
1818
1819 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1821 *self.topleft_tab.lock().unwrap() = t;
1822 self.bump();
1823 }
1824
1825 pub fn bottomright_tab(&self) -> BottomRightTab {
1827 *self.bottomright_tab.lock().unwrap()
1828 }
1829
1830 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1832 *self.bottomright_tab.lock().unwrap() = t;
1833 self.bump();
1834 }
1835
1836 pub fn data_split(&self) -> DataSplit {
1838 *self.data_split.lock().unwrap()
1839 }
1840
1841 pub fn set_data_split(&self, s: DataSplit) {
1844 let s = DataSplit {
1845 col: crate::splitters::clamp_frac(s.col),
1846 row: crate::splitters::clamp_frac(s.row),
1847 };
1848 let mut g = self.data_split.lock().unwrap();
1849 if g.col != s.col || g.row != s.row {
1850 *g = s;
1851 self.bump();
1852 }
1853 }
1854
1855 pub fn reset_layout(&self) {
1859 *self.data_split.lock().unwrap() = DataSplit::default();
1860 *self.selected_exchange.lock().unwrap() = String::new();
1861 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1862 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1863 self.bump();
1864 }
1865
1866 pub fn ops_window_open(&self) -> bool {
1870 self.ops_window.load(Ordering::Acquire)
1871 }
1872
1873 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1876 self.table_sorts
1877 .lock()
1878 .unwrap()
1879 .get(table)
1880 .copied()
1881 .flatten()
1882 }
1883
1884 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1886 let mut m = self.table_sorts.lock().unwrap();
1887 let asc = match m.get(table).copied().flatten() {
1888 Some((c, a)) if c == col => !a,
1889 _ => true,
1890 };
1891 m.insert(table.to_string(), Some((col, asc)));
1892 }
1893
1894 pub fn set_ops_window_open(&self, v: bool) {
1895 self.ops_window.store(v, Ordering::Release);
1896 self.bump();
1897 }
1898
1899 pub fn accounts_window_open(&self) -> bool {
1906 self.modules().is_open("ctp")
1907 }
1908
1909 pub fn set_accounts_window_open(&self, v: bool) {
1911 self.modules().set_open("ctp", v);
1912 self.bump();
1913 }
1914
1915 pub fn account_edit_open(&self) -> bool {
1917 self.account_edit_window.load(Ordering::Acquire)
1918 }
1919
1920 pub fn editing_account(&self) -> Option<String> {
1922 self.editing_account.lock().unwrap().clone()
1923 }
1924
1925 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1929 match gateway {
1930 Some(gw) => {
1931 let Some(row) = self.account(gw) else {
1932 return false;
1933 };
1934 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1935 *self.edit_alias.lock().unwrap() = row.alias.clone();
1936 self.set_ctp_form(row.form.clone());
1937 }
1938 None => {
1939 *self.editing_account.lock().unwrap() = None;
1940 *self.edit_alias.lock().unwrap() = String::new();
1941 let draft = self.draft_from_existing();
1942 self.set_ctp_form(draft);
1943 }
1944 }
1945 *self.edit_error.lock().unwrap() = None;
1946 self.account_edit_window.store(true, Ordering::Release);
1947 self.bump();
1948 true
1949 }
1950
1951 pub fn end_account_edit(&self) {
1953 *self.editing_account.lock().unwrap() = None;
1954 *self.edit_alias.lock().unwrap() = String::new();
1955 *self.edit_error.lock().unwrap() = None;
1956 self.account_edit_window.store(false, Ordering::Release);
1957 let draft = self.draft_from_existing();
1958 self.set_ctp_form(draft);
1959 self.bump();
1960 }
1961
1962 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1968 if self.editing_account().as_deref() == Some(gateway) {
1969 self.end_account_edit();
1970 }
1971 }
1972
1973 pub fn edit_alias(&self) -> String {
1975 self.edit_alias.lock().unwrap().clone()
1976 }
1977
1978 pub fn set_edit_alias(&self, v: impl Into<String>) {
1980 *self.edit_alias.lock().unwrap() = v.into();
1981 self.bump();
1982 }
1983
1984 pub fn edit_error(&self) -> Option<String> {
1986 self.edit_error.lock().unwrap().clone()
1987 }
1988
1989 pub fn set_edit_error(&self, e: Option<String>) {
1991 *self.edit_error.lock().unwrap() = e;
1992 self.bump();
1993 }
1994
1995 pub fn account_edit_dirty(&self) -> bool {
1997 let editing = self.editing_account();
1998 let draft = self.ctp_form();
1999 match editing {
2000 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
2001 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
2002 }
2003 }
2004
2005 pub fn accounts(&self) -> Vec<AccountRow> {
2007 self.accounts.lock().unwrap().clone()
2008 }
2009
2010 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
2012 self.accounts
2013 .lock()
2014 .unwrap()
2015 .iter()
2016 .find(|a| a.gateway == gateway)
2017 .cloned()
2018 }
2019
2020 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
2026 let mut acc = self.accounts.lock().unwrap();
2027 if acc.iter().any(|a| a.gateway == row.gateway) {
2028 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
2029 }
2030 let alias = row.alias.trim();
2031 if !alias.is_empty()
2032 && acc
2033 .iter()
2034 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
2035 {
2036 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
2037 }
2038 if acc.len() >= MAX_ACCOUNTS {
2039 return Err(format!(
2040 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
2041 ));
2042 }
2043 let has_alias = !row.alias.trim().is_empty();
2044 acc.push(row);
2045 drop(acc);
2046 if has_alias {
2047 self.alias_persist_due.store(true, Ordering::Release);
2049 }
2050 self.bump();
2051 Ok(())
2052 }
2053
2054 #[must_use]
2056 pub fn can_add_account(&self) -> bool {
2057 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
2058 }
2059
2060 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
2066 let alias = alias.trim().to_string();
2067 let mut acc = self.accounts.lock().unwrap();
2068 if !alias.is_empty()
2069 && acc
2070 .iter()
2071 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
2072 {
2073 return false;
2074 }
2075 match acc.iter_mut().find(|a| a.gateway == gateway) {
2076 Some(a) => {
2077 a.alias = alias;
2078 drop(acc);
2079 self.alias_persist_due.store(true, Ordering::Release);
2080 self.bump();
2081 true
2082 }
2083 None => false,
2084 }
2085 }
2086
2087 pub fn note_connect_sent(&self, gateway: &str) {
2092 let mut acc = self.accounts.lock().unwrap();
2093 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2094 a.stage = AccountStage::Assembling;
2095 a.fail_note = None;
2096 a.login = LoginState::Unknown;
2097 a.status_note = None;
2098 a.login_brief = None;
2099 a.chain_note = None;
2101 drop(acc);
2102 self.bump();
2103 }
2104 }
2105
2106 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
2110 let mut acc = self.accounts.lock().unwrap();
2111 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2112 match result {
2113 Ok(()) => {
2114 a.stage = AccountStage::Assembled;
2115 a.fail_note = None;
2116 }
2117 Err(e) => {
2118 a.stage = AccountStage::AssembleFailed;
2119 a.fail_note = Some(e);
2120 }
2121 }
2122 drop(acc);
2123 self.bump();
2124 }
2125 }
2126
2127 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
2131 let mut acc = self.accounts.lock().unwrap();
2132 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2133 a.effective_classes = classes.to_vec();
2134 drop(acc);
2135 self.bump();
2136 }
2137 }
2138
2139 pub fn note_chain(&self, gateway: &str, note: &str) {
2144 let mut acc = self.accounts.lock().unwrap();
2145 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2146 if a.chain_note.as_deref() == Some(note) {
2147 return; }
2149 a.chain_note = Some(note.to_string());
2150 drop(acc);
2151 self.bump();
2152 }
2153 }
2154
2155 #[must_use]
2159 pub fn status_thresholds(&self) -> (u32, i64) {
2160 self.status_thresholds
2161 .lock()
2162 .unwrap()
2163 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
2164 }
2165
2166 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
2168 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
2169 self.bump();
2170 }
2171
2172 pub fn list_poll_due(&self) -> bool {
2177 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
2178 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
2179 }
2180
2181 pub fn request_list_refresh(&self) {
2183 self.list_due.store(true, Ordering::Release);
2184 }
2185
2186 pub fn request_trade_reseed(&self) {
2188 self.seed_due.store(true, Ordering::Release);
2189 }
2190
2191 pub fn take_trade_reseed(&self) -> bool {
2193 self.seed_due.swap(false, Ordering::AcqRel)
2194 }
2195
2196 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
2201 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
2205 for gw in present {
2206 if !uniq.iter().any(|u| u == gw) {
2207 uniq.push(gw.clone());
2208 }
2209 }
2210 let n = uniq.len();
2211 if n == 0 {
2212 return Vec::new();
2213 }
2214 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
2215 let take = STATUS_ASKS_PER_TICK.min(n);
2216 let mut out = Vec::with_capacity(take);
2217 for i in 0..take {
2218 out.push(uniq[(start + i) % n].clone());
2219 }
2220 out
2221 }
2222
2223 pub fn request_status_seed(&self, gateway: &str) {
2225 let mut q = self.status_seed.lock().unwrap();
2226 if !q.iter().any(|g| g == gateway) {
2227 q.push(gateway.to_string());
2228 }
2229 }
2230
2231 pub fn take_status_seeds(&self) -> Vec<String> {
2233 let mut q = self.status_seed.lock().unwrap();
2234 std::mem::take(&mut *q)
2235 }
2236
2237 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
2247 let gw = ev.gateway.as_str();
2248 let text = ev.describe();
2249 self.log_line(format!("[INFO][{gw}] {text}"));
2250 let note = text
2251 .strip_prefix(CHAIN_STATE_PREFIX)
2252 .map(|s| s.trim().to_string())
2253 .unwrap_or_else(|| text.clone());
2254 self.note_chain(gw, ¬e);
2255 self.note_gateway_status(gw, &ev.status);
2256 let alarm_kind = matches!(
2257 ev.change,
2258 ek_ctp::GatewayChange::TdDisconnected
2259 | ek_ctp::GatewayChange::TdAuthFailed
2260 | ek_ctp::GatewayChange::TdLoginFailed
2261 | ek_ctp::GatewayChange::ContractQueryStopped
2262 | ek_ctp::GatewayChange::QueryRejected
2263 | ek_ctp::GatewayChange::MdDisconnected
2264 | ek_ctp::GatewayChange::MdLoginFailed
2265 | ek_ctp::GatewayChange::SubscribeRejected
2266 );
2267 if alarm_kind {
2268 let session = if ev.change.kind().starts_with("md-") {
2270 ev.status.md_disconnects
2271 } else {
2272 ev.status.td_disconnects
2273 };
2274 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
2275 if self.alert_once(key) {
2276 self.push_alert(
2277 crate::toast::AlertLevel::Warning,
2278 format!("账户 {gw}:{}", note.trim()),
2279 );
2280 }
2281 }
2282 }
2283
2284 pub fn request_pipeline_refresh(&self) {
2286 self.pipeline_due.store(true, Ordering::Release);
2287 }
2288 pub fn take_pipeline_due(&self) -> bool {
2290 self.pipeline_due.swap(false, Ordering::AcqRel)
2291 }
2292
2293 pub fn request_cfg_refresh(&self) {
2295 self.cfg_due.store(true, Ordering::Release);
2296 }
2297
2298 pub fn take_cfg_due(&self) -> bool {
2300 self.cfg_due.swap(false, Ordering::AcqRel)
2301 }
2302
2303 pub fn theme(&self) -> crate::tokens::Theme {
2305 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
2306 }
2307
2308 pub fn set_theme(&self, theme: crate::tokens::Theme) {
2311 self.theme.store(theme.as_u8(), Ordering::Release);
2312 self.bump();
2313 }
2314
2315 pub fn toggle_theme(&self) -> crate::tokens::Theme {
2317 let next = self.theme().flipped();
2318 self.theme.store(next.as_u8(), Ordering::Release);
2319 self.theme_persist_due.store(true, Ordering::Release);
2320 self.bump();
2321 next
2322 }
2323
2324 pub fn take_theme_persist_due(&self) -> bool {
2326 self.theme_persist_due.swap(false, Ordering::AcqRel)
2327 }
2328
2329 pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
2333 {
2334 let mut acc = self.accounts.lock().unwrap();
2335 match acc.iter_mut().find(|a| a.gateway == gateway) {
2336 Some(a) => {
2337 if a.auto_connect != on {
2338 a.auto_connect = on;
2339 self.auto_connect_persist_due.store(true, Ordering::Release);
2340 self.bump();
2341 }
2342 }
2343 None => return Err(format!("账户 {gateway} 不在清册")),
2344 }
2345 }
2346 Ok(())
2347 }
2348
2349 pub fn take_auto_connect_persist_due(&self) -> bool {
2351 self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
2352 }
2353
2354 pub fn take_alias_persist_due(&self) -> bool {
2356 self.alias_persist_due.swap(false, Ordering::AcqRel)
2357 }
2358
2359 pub fn take_archive_purges(&self) -> Vec<String> {
2361 let mut q = self.archive_purge.lock().unwrap();
2362 std::mem::take(&mut *q)
2363 }
2364
2365 pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
2380 let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
2382 let mut aliases: std::collections::BTreeMap<String, String> = Default::default();
2383 match store.load(crate::plugin::THEME_NS) {
2384 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
2385 for (k, _f) in map.iter() {
2386 if let Some(gw) = k.strip_prefix("auto_connect.") {
2387 flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
2388 }
2389 if let Some(gw) = k.strip_prefix("alias.")
2390 && let Some(v) = map.get_plain(k)
2391 {
2392 let v = v.trim();
2393 if !v.is_empty() {
2394 aliases.insert(gw.to_string(), v.to_string());
2395 }
2396 }
2397 }
2398 }
2399 Ok(_) => {} Err(_) => {} }
2402 let mut restored: Vec<String> = Vec::new();
2403 let ns_list = match store.list_namespaces() {
2404 Ok(v) => v,
2405 Err(e) => {
2406 let msg = format!(
2409 "档案清点失败:{e} —— 账户表会是空的,但**档案没丢**。\
2410 多半是安全软件拦了本程序写 %APPDATA%\\ek\\cfg(放行后重启即恢复)"
2411 );
2412 self.log_line(format!("[ui] {msg}"));
2413 self.set_store_err(Some(msg));
2414 return restored;
2415 }
2416 };
2417 self.set_store_err(None);
2419 for ns in ns_list {
2420 let Some(gw) = ns.strip_prefix("ctp.") else {
2421 continue;
2422 };
2423 if self.account(gw).is_some() {
2424 self.log_line(format!(
2425 "[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
2426 ));
2427 continue;
2428 }
2429 let map = match store.load(&ns) {
2430 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
2431 Ok(_) => {
2432 self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
2433 continue;
2434 }
2435 Err(e) => {
2436 self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
2437 continue;
2438 }
2439 };
2440 let password_ref = map
2442 .get_secret_key("td_password")
2443 .is_some_and(|k| !k.is_empty());
2444 let mut form = CtpForm {
2445 broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
2446 td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
2447 td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
2448 td_password: String::new(),
2449 td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
2450 td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
2451 md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
2452 md_user: String::new(),
2453 md_password: String::new(),
2454 symbols: String::new(),
2455 accept_futures: false,
2456 accept_options: false,
2457 };
2458 for c in map
2461 .get_plain("product_classes")
2462 .unwrap_or_default()
2463 .split(',')
2464 {
2465 if c.trim() == "1" {
2466 form.accept_futures = true;
2467 }
2468 }
2469 let Ok(mut row) = AccountRow::derive(&form) else {
2470 self.log_line(format!(
2471 "[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
2472 ));
2473 continue;
2474 };
2475 row.gateway = gw.to_string();
2478 row.restored = true;
2479 row.auto_connect = flags.get(gw).copied().unwrap_or(false);
2480 row.password_ref = password_ref;
2481 if let Some(a) = aliases.get(gw) {
2484 let taken = self
2485 .accounts
2486 .lock()
2487 .unwrap()
2488 .iter()
2489 .any(|x| x.alias.trim().eq_ignore_ascii_case(a));
2490 if taken {
2491 self.log_line(format!(
2492 "[ui] 档案别名 {a:?} 已被在册行占用,{gw} 不带标签复原(账户面板改别名即可)"
2493 ));
2494 } else {
2495 row.alias = a.clone();
2496 }
2497 }
2498 restored.push(row.gateway.clone());
2499 self.add_account(row).ok();
2500 }
2501 if !restored.is_empty() {
2502 self.log_line(format!(
2503 "[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
2504 restored.len()
2505 ));
2506 if self.md_source.lock().unwrap().is_empty() {
2509 let first = self
2510 .accounts
2511 .lock()
2512 .unwrap()
2513 .iter()
2514 .find(|a| a.auto_connect)
2515 .map(|a| a.gateway.clone());
2516 if let Some(gw) = first
2517 && self.set_md_source(&gw).is_ok()
2518 {
2519 self.log_line(format!("[账户] 行情源(默认=第一个勾选自动连接的)= {gw}"));
2520 }
2521 }
2522 self.bump();
2523 }
2524 restored
2525 }
2526
2527 pub fn restore_saved_accounts(&self) -> Vec<String> {
2529 match ek_cfg::Store::eco() {
2530 Ok(store) => self.restore_saved_accounts_from(&store),
2531 Err(_) => {
2532 self.log_line(
2533 "[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
2534 );
2535 Vec::new()
2536 }
2537 }
2538 }
2539
2540 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
2542 *self.cfg_briefs.lock().unwrap() = Some(b);
2543 self.bump();
2544 }
2545 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
2547 self.cfg_briefs.lock().unwrap().clone()
2548 }
2549
2550 pub fn cfg_delete_armed(&self) -> Option<String> {
2552 self.cfg_delete_arm.lock().unwrap().clone()
2553 }
2554 pub fn arm_cfg_delete(&self, ns: Option<String>) {
2555 *self.cfg_delete_arm.lock().unwrap() = ns;
2556 self.bump();
2557 }
2558
2559 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
2561 *self.pipeline_lines.lock().unwrap() = lines;
2562 self.bump();
2563 }
2564
2565 pub fn pipeline_lines(&self) -> Vec<String> {
2567 self.pipeline_lines.lock().unwrap().clone()
2568 }
2569
2570 pub fn note_bad_payload(&self, line: &str) {
2576 let mut book = self.bad_payload.lock().unwrap();
2577 let n = book.entry(line.to_string()).or_insert(0);
2578 *n += 1;
2579 if *n == 1 {
2580 self.log_line(format!(
2581 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
2582 ));
2583 }
2584 }
2585
2586 #[must_use]
2588 pub fn bad_payload_count(&self, line: &str) -> u64 {
2589 self.bad_payload
2590 .lock()
2591 .unwrap()
2592 .get(line)
2593 .copied()
2594 .unwrap_or(0)
2595 }
2596
2597 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
2611 for r in rows {
2612 let (warn, crit) = pending_thresholds(r.soft_cap);
2613 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
2614 self.push_alert(
2615 crate::toast::AlertLevel::Warning,
2616 format!(
2617 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
2618 r.name, r.pending, r.peak, r.soft_cap
2619 ),
2620 );
2621 }
2622 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
2623 self.push_alert(
2624 crate::toast::AlertLevel::Error,
2625 format!(
2626 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
2627 r.name, r.pending, r.soft_cap
2628 ),
2629 );
2630 }
2631 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
2632 self.push_alert(
2633 crate::toast::AlertLevel::Error,
2634 format!(
2635 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
2636 r.name, r.soft_cap, r.spilled, r.peak
2637 ),
2638 );
2639 }
2640 }
2641 }
2642
2643 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
2651 let (limit, stale) = self.status_thresholds();
2652 let now = ek_ctp::map::now_epoch_ms();
2653 let mut alarm: Option<(LoginState, String)> = None;
2656 let mut found = false;
2657 {
2658 let mut acc = self.accounts.lock().unwrap();
2659 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2660 found = true;
2661 a.apply_status(st, now, limit, stale);
2662 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
2663 alarm = Some((
2664 a.login,
2665 a.status_note
2666 .clone()
2667 .unwrap_or_else(|| a.login.label().to_string()),
2668 ));
2669 }
2670 }
2671 }
2672 if found {
2673 self.bump();
2674 }
2675 if let Some((login, note)) = alarm
2676 && self.alert_once(format!("{gateway}|{}", login.label()))
2677 {
2678 self.push_alert(
2679 crate::toast::AlertLevel::Error,
2680 format!("账户 {gateway}:{note}"),
2681 );
2682 }
2683 }
2684
2685 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2689 let mut acc = self.accounts.lock().unwrap();
2690 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2691 a.login = LoginState::Unknown;
2692 a.status_note = Some(format!("状态口问不到:{why}"));
2693 a.login_brief = None;
2694 drop(acc);
2695 self.bump();
2696 }
2697 }
2698
2699 pub fn mark_account_stopped(&self, gateway: &str) {
2701 let mut acc = self.accounts.lock().unwrap();
2702 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2703 a.login = LoginState::Stopped;
2704 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2705 a.login_brief = None;
2706 drop(acc);
2707 self.bump();
2708 }
2709 }
2710
2711 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2714 let source = self.md_source();
2715 let mut f = form;
2716 if source.as_str() != gateway {
2717 f.md_addr.clear();
2718 f.md_user.clear();
2719 f.md_password.clear();
2720 }
2721 self.update_account_form(gateway, f.clone());
2722 f
2723 }
2724
2725 #[must_use]
2729 pub fn draft_from_existing(&self) -> CtpForm {
2730 let acc = self.accounts.lock().unwrap();
2731 let mut draft = CtpForm::default();
2732 if let Some(a) = acc.last() {
2733 draft.broker_id = a.form.broker_id.clone();
2734 draft.td_auth_code = a.form.td_auth_code.clone();
2735 draft.td_app_id = a.form.td_app_id.clone();
2736 draft.accept_futures = a.form.accept_futures;
2737 draft.accept_options = a.form.accept_options;
2738 }
2739 draft
2740 }
2741
2742 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2744 let mut acc = self.accounts.lock().unwrap();
2745 match acc.iter_mut().find(|a| a.gateway == gateway) {
2746 Some(a) => {
2747 a.form = form;
2748 drop(acc);
2749 self.bump();
2750 true
2751 }
2752 None => false,
2753 }
2754 }
2755
2756 pub fn remove_account(&self, gateway: &str) -> bool {
2758 let mut acc = self.accounts.lock().unwrap();
2759 let before = acc.len();
2760 acc.retain(|a| a.gateway != gateway);
2761 let removed = acc.len() != before;
2762 if removed {
2763 let mut md = self.md_source.lock().unwrap();
2765 if md.as_str() == gateway {
2766 md.clear();
2767 }
2768 drop(acc);
2769 self.alias_persist_due.store(true, Ordering::Release);
2771 self.archive_purge.lock().unwrap().push(gateway.to_string());
2773 self.bump();
2774 }
2775 removed
2776 }
2777
2778 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2787 let mut newly = Vec::new();
2788 {
2789 let mut acc = self.accounts.lock().unwrap();
2790 for a in acc.iter_mut() {
2791 let up = rows.iter().any(|r| r.name == a.gateway);
2792 if up && !a.connected {
2793 newly.push(a.gateway.clone());
2794 }
2795 a.connected = up;
2796 if up {
2797 a.stage = AccountStage::Assembled;
2798 a.fail_note = None;
2799 } else {
2800 a.chain_note = None;
2803 if a.stage == AccountStage::Assembled {
2804 a.stage = AccountStage::NotAssembled;
2805 a.login = LoginState::Unknown;
2806 a.status_note = None;
2807 a.login_brief = None;
2808 }
2809 }
2810 }
2811 }
2812 if !newly.is_empty() {
2813 self.bump();
2814 }
2815 newly
2816 }
2817
2818 pub fn md_source(&self) -> String {
2820 self.md_source.lock().unwrap().clone()
2821 }
2822
2823 pub fn accounts_locked(&self) -> bool {
2825 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2826 }
2827
2828 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2830 {
2831 let acc = self.accounts.lock().unwrap();
2832 if acc.iter().any(|a| a.connected) {
2833 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2834 }
2835 if !acc.iter().any(|a| a.gateway == gateway) {
2836 return Err(format!("账户 {gateway} 不在清册"));
2837 }
2838 }
2839 let mut acc = self.accounts.lock().unwrap();
2840 for a in acc.iter_mut() {
2841 a.md_source = a.gateway == gateway;
2842 }
2843 drop(acc);
2844 let mut md = self.md_source.lock().unwrap();
2845 *md = gateway.to_string();
2846 self.bump();
2847 Ok(())
2848 }
2849
2850 pub fn subscriptions(&self) -> Vec<String> {
2852 self.subscriptions.lock().unwrap().clone()
2853 }
2854
2855 pub fn note_subscriptions(&self, symbols: &[String]) {
2857 {
2858 let mut subs = self.subscriptions.lock().unwrap();
2859 for s in symbols {
2860 let s = s.trim();
2861 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2862 subs.push(s.to_string());
2863 }
2864 }
2865 }
2866 self.bump();
2867 }
2868
2869 pub fn default_gateway(&self) -> Option<String> {
2876 let chosen = {
2877 let f = self.order_form.lock().unwrap();
2878 f.gateway.trim().to_string()
2879 };
2880 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2881 return Some(chosen);
2882 }
2883 let acc = self.accounts.lock().unwrap();
2884 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2885 }
2886
2887 pub fn kline_symbols(&self) -> Vec<String> {
2889 self.kline_windows.lock().unwrap().clone()
2890 }
2891
2892 pub fn open_kline(&self, symbol: impl Into<String>) {
2894 let s = symbol.into();
2895 let mut v = self.kline_windows.lock().unwrap();
2896 if !v.contains(&s) {
2897 v.push(s.clone());
2898 }
2899 drop(v);
2900 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2901 let fq = if s.contains('.') {
2904 s.clone()
2905 } else {
2906 format!("{}.{}", self.selected_exchange(), s)
2907 };
2908 const KLINE_INTERVAL_SECS: i64 = 60;
2909 {
2910 let mut fetched = self.kline_fetched.lock().unwrap();
2911 if fetched.contains(&(fq.clone(), KLINE_INTERVAL_SECS)) {
2912 return;
2913 }
2914 fetched.insert((fq.clone(), KLINE_INTERVAL_SECS));
2915 }
2916 self.kline_fetches.lock().unwrap().push(KlineFetch {
2917 symbol: fq,
2918 interval_secs: KLINE_INTERVAL_SECS,
2919 });
2920 self.log_line(format!("[kline] 入队历史补数 {s}(1min×300,tq 桥)"));
2921 }
2922
2923 pub fn take_kline_fetches(&self) -> Vec<KlineFetch> {
2925 std::mem::take(&mut self.kline_fetches.lock().unwrap())
2926 }
2927
2928 pub fn kline_fetch_pending(&self) -> bool {
2930 !self.kline_fetches.lock().unwrap().is_empty()
2931 }
2932
2933 pub fn close_kline(&self, symbol: &str) {
2935 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2936 self.bump();
2937 }
2938
2939 pub fn contracts_window_open(&self) -> bool {
2943 self.contracts_window.load(Ordering::Acquire)
2944 }
2945
2946 pub fn set_contracts_window_open(&self, v: bool) {
2948 self.contracts_window.store(v, Ordering::Release);
2949 self.bump();
2950 }
2951
2952 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2959 self.contract_store.lock().unwrap().put(c);
2960 if !c.name.is_empty() {
2961 self.backfill_names(&c.symbol, &c.name);
2962 }
2963 self.bump();
2964 }
2965
2966 fn backfill_names(&self, symbol: &str, name: &str) {
2971 let mut touched = false;
2972 {
2973 let mut rows = self.trade_orders.lock().unwrap();
2974 for (_, (_, o)) in rows.iter_mut() {
2975 if o.symbol == symbol && o.name.is_empty() {
2976 o.name = name.to_string();
2977 touched = true;
2978 }
2979 }
2980 }
2981 {
2982 let mut rows = self.trade_trades.lock().unwrap();
2983 for (_, t) in rows.iter_mut() {
2984 if t.symbol == symbol && t.name.is_empty() {
2985 t.name = name.to_string();
2986 touched = true;
2987 }
2988 }
2989 }
2990 {
2991 let mut rows = self.trade_positions.lock().unwrap();
2992 for (_, (_, p)) in rows.iter_mut() {
2993 if p.symbol == symbol && p.name.is_empty() {
2994 p.name = name.to_string();
2995 touched = true;
2996 }
2997 }
2998 }
2999 if touched {
3000 self.trade_dirty.store(true, Ordering::Release);
3002 }
3003 }
3004
3005 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
3007 self.contract_store.lock().unwrap().query(pattern)
3008 }
3009
3010 pub fn contract_count(&self) -> usize {
3012 self.contract_store.lock().unwrap().len()
3013 }
3014
3015 fn contract_name_of(&self, symbol: &str) -> Option<String> {
3017 self.contract_store
3018 .lock()
3019 .unwrap()
3020 .query(symbol)
3021 .into_iter()
3022 .find(|c| c.symbol == symbol && !c.name.is_empty())
3023 .map(|c| c.name)
3024 }
3025
3026 pub fn contract_query(&self) -> String {
3028 self.contract_query.lock().unwrap().clone()
3029 }
3030
3031 pub fn set_contract_query(&self, q: impl Into<String>) {
3033 *self.contract_query.lock().unwrap() = q.into();
3034 self.bump();
3035 }
3036
3037 pub fn toggle_contract_selection(&self, en_symbol: &str) {
3039 let mut v = self.contract_selected.lock().unwrap();
3040 match v.iter().position(|s| s == en_symbol) {
3041 Some(i) => {
3042 v.remove(i);
3043 }
3044 None => v.push(en_symbol.to_string()),
3045 }
3046 self.bump();
3047 }
3048
3049 pub fn selected_contracts(&self) -> Vec<String> {
3051 self.contract_selected.lock().unwrap().clone()
3052 }
3053
3054 #[must_use]
3060 pub fn cancel_reject_note(msg: &str) -> String {
3061 let head = msg.split(' ').next().unwrap_or(msg);
3062 let note = if head.starts_with("unknown:order") {
3063 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
3064 } else if head.starts_with("mismatch:gateway") {
3065 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
3066 } else if head.starts_with("terminal:status") {
3067 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
3068 } else if head.starts_with("limit:cancel-guard") {
3069 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
3070 } else {
3071 ""
3072 };
3073 if note.is_empty() {
3074 format!("撤单被拒:{msg}")
3076 } else {
3077 format!("撤单被拒:{note}(机器码:{msg})")
3078 }
3079 }
3080
3081 pub fn log_line(&self, line: impl Into<String>) {
3083 self.logs.lock().unwrap().push(line);
3084 self.bump();
3085 }
3086
3087 pub fn clear_logs(&self) {
3089 self.logs.lock().unwrap().clear();
3090 self.bump();
3091 }
3092
3093 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
3095 let r = self.logs.lock().unwrap();
3096 (r.snapshot(), r.spilled())
3097 }
3098
3099 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
3101 let r = self.logs.lock().unwrap();
3102 (r.entries(), r.spilled())
3103 }
3104
3105 pub fn logs_len(&self) -> usize {
3107 self.logs.lock().unwrap().len()
3108 }
3109
3110 pub fn set_snapshot(&self, snap: BusSnap) {
3112 *self.snap.lock().unwrap() = Arc::new(snap);
3113 self.bump();
3114 }
3115
3116 pub fn snapshot(&self) -> Arc<BusSnap> {
3118 Arc::clone(&self.snap.lock().unwrap())
3119 }
3120
3121 pub fn rev(&self) -> u64 {
3123 self.rev.load(Ordering::Acquire)
3124 }
3125
3126 pub fn bump(&self) {
3128 self.rev.fetch_add(1, Ordering::AcqRel);
3129 }
3130
3131 pub fn request_stop(&self) {
3133 self.stop.store(true, Ordering::Release);
3134 }
3135
3136 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
3139 self.pending_alerts
3140 .lock()
3141 .unwrap_or_else(std::sync::PoisonError::into_inner)
3142 .push((level, text.into()));
3143 }
3144
3145 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
3147 let mut q = self
3148 .pending_alerts
3149 .lock()
3150 .unwrap_or_else(std::sync::PoisonError::into_inner);
3151 std::mem::take(&mut *q)
3152 }
3153
3154 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
3156 self.toasts.lock()
3157 }
3158
3159 pub fn alert_once(&self, key: impl Into<String>) -> bool {
3161 let key = key.into();
3162 self.alerted
3163 .lock()
3164 .unwrap_or_else(std::sync::PoisonError::into_inner)
3165 .insert(key)
3166 }
3167
3168 #[must_use]
3170 pub fn pending_alert_count(&self) -> usize {
3171 self.pending_alerts
3172 .lock()
3173 .unwrap_or_else(std::sync::PoisonError::into_inner)
3174 .len()
3175 }
3176
3177 #[must_use]
3179 pub fn alerted_count(&self) -> usize {
3180 self.alerted
3181 .lock()
3182 .unwrap_or_else(std::sync::PoisonError::into_inner)
3183 .len()
3184 }
3185
3186 pub fn exit_confirm_open(&self) -> bool {
3188 self.exit_confirm_open
3189 .load(std::sync::atomic::Ordering::Relaxed)
3190 }
3191
3192 pub fn open_exit_confirm(&self) {
3193 self.exit_confirm_open
3194 .store(true, std::sync::atomic::Ordering::Relaxed);
3195 }
3196
3197 pub fn close_exit_confirm(&self) {
3198 self.exit_confirm_open
3199 .store(false, std::sync::atomic::Ordering::Relaxed);
3200 }
3201
3202 pub fn exit_confirmed(&self) -> bool {
3203 self.exit_confirmed
3204 .load(std::sync::atomic::Ordering::Relaxed)
3205 }
3206
3207 pub fn confirm_exit(&self) {
3208 self.exit_confirmed
3209 .store(true, std::sync::atomic::Ordering::Relaxed);
3210 }
3211
3212 pub fn is_stop(&self) -> bool {
3213 self.stop.load(Ordering::Acquire)
3214 }
3215
3216 #[must_use]
3218 pub fn assembly_locked(&self) -> bool {
3219 self.assembly_locked.load(Ordering::Relaxed)
3220 }
3221
3222 pub fn set_assembly_locked(&self, on: bool) {
3224 self.assembly_locked.store(on, Ordering::Relaxed);
3225 }
3226
3227 pub fn push_cmd(&self, cmd: UiCmd) {
3231 self.cmds.lock().unwrap().push_back(cmd);
3232 self.bump();
3233 }
3234
3235 pub fn drain_cmds(&self) -> Vec<UiCmd> {
3237 self.cmds.lock().unwrap().drain(..).collect()
3238 }
3239
3240 #[must_use]
3242 pub fn cmd_pending(&self) -> bool {
3243 !self.cmds.lock().unwrap().is_empty()
3244 }
3245
3246 pub fn pending_confirm(&self) -> Option<String> {
3248 self.pending_confirm.lock().unwrap().clone()
3249 }
3250
3251 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
3253 let mut g = self.pending_confirm.lock().unwrap();
3254 match g.take() {
3255 Some(p) if p == name => ConfirmStep::Fired,
3256 Some(_) => {
3257 *g = Some(name.to_string());
3258 ConfirmStep::Switched
3259 }
3260 None => {
3261 *g = Some(name.to_string());
3262 ConfirmStep::Armed
3263 }
3264 }
3265 }
3266
3267 pub fn confirm_cancel(&self) {
3269 *self.pending_confirm.lock().unwrap() = None;
3270 self.bump();
3271 }
3272
3273 pub fn market(&self) -> &MarketStore {
3277 &self.market
3278 }
3279
3280 pub fn seed_trade_snap(&self, ts: TradeSnap) {
3289 let named = |symbol: &str, name: String| -> String {
3292 if name.is_empty() {
3293 self.contract_name_of(symbol).unwrap_or(name)
3294 } else {
3295 name
3296 }
3297 };
3298 {
3299 let mut book = self.trade_orders.lock().unwrap();
3300 book.clear();
3301 for mut o in ts.orders.into_iter().rev() {
3302 o.name = named(&o.symbol, o.name);
3303 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3304 book.insert(o.en_orderid(), (seq, o));
3305 }
3306 }
3307 {
3308 let mut book = self.trade_trades.lock().unwrap();
3309 book.clear();
3310 let mut ids = self.trade_ids.lock().unwrap();
3311 ids.clear();
3312 for mut t in ts.trades.into_iter().rev() {
3313 t.name = named(&t.symbol, t.name);
3314 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3315 ids.insert(t.en_tradeid());
3316 book.push_front((seq, t));
3317 }
3318 }
3319 {
3320 let mut book = self.trade_positions.lock().unwrap();
3321 book.clear();
3322 for mut p in ts.positions.into_iter().rev() {
3323 p.name = named(&p.symbol, p.name);
3324 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3325 book.insert(p.en_positionid(), (seq, p));
3326 }
3327 }
3328 {
3329 let mut book = self.trade_funds.lock().unwrap();
3330 book.clear();
3331 for a in ts.accounts.into_iter().rev() {
3332 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3333 book.insert(a.en_accountid(), (seq, a));
3334 }
3335 }
3336 self.trade_dirty.store(true, Ordering::Release);
3337 self.bump();
3338 }
3339
3340 pub fn note_order(&self, o: &ek_data::order::OrderData) {
3348 let key = o.en_orderid();
3349 {
3350 let mut book = self.trade_orders.lock().unwrap();
3351 match book.get_mut(&key) {
3352 Some((_, cur)) => {
3353 cur.name.clone_from(&o.name);
3354 cur.traded = o.traded;
3355 cur.traded_price = o.traded_price;
3356 cur.status = o.status;
3357 cur.status_msg.clone_from(&o.status_msg);
3358 cur.datetime = o.datetime;
3359 }
3360 None => {
3361 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3362 book.insert(key, (seq, o.clone()));
3363 }
3364 }
3365 }
3366 self.trade_dirty.store(true, Ordering::Release);
3367 self.bump();
3368 }
3369
3370 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
3378 let id = t.en_tradeid();
3379 if self.trade_ids.lock().unwrap().contains(&id) {
3380 return;
3381 }
3382 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3383 let dropped = {
3384 let mut book = self.trade_trades.lock().unwrap();
3385 book.push_front((seq, t.clone()));
3386 self.trade_ids.lock().unwrap().insert(id);
3387 let over = book.len().saturating_sub(TRADE_CAP);
3388 for _ in 0..over {
3389 if let Some((_, old)) = book.pop_back() {
3391 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
3392 }
3393 }
3394 over
3395 };
3396 if dropped > 0 {
3397 let total = self
3398 .trade_dropped
3399 .fetch_add(dropped as u64, Ordering::AcqRel)
3400 + dropped as u64;
3401 if total == dropped as u64 {
3403 self.push_alert(
3404 crate::toast::AlertLevel::Warning,
3405 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
3406 );
3407 }
3408 }
3409 self.trade_dirty.store(true, Ordering::Release);
3410 self.bump();
3411 }
3412
3413 pub fn note_position(&self, p: &ek_data::position::PositionData) {
3418 let key = p.en_positionid();
3419 {
3420 let mut book = self.trade_positions.lock().unwrap();
3421 match book.get_mut(&key) {
3422 Some((_, cur)) => {
3423 cur.volume = p.volume;
3424 cur.yd_volume = p.yd_volume;
3425 cur.can_use_volume = p.can_use_volume;
3426 cur.frozen = p.frozen;
3427 cur.on_road_volume = p.on_road_volume;
3428 cur.market_value = p.market_value;
3429 cur.avg_price = p.avg_price;
3430 cur.price = p.price;
3431 cur.last_price = p.last_price;
3432 cur.pnl = p.pnl;
3433 cur.profit_rate = p.profit_rate;
3434 cur.name.clone_from(&p.name);
3435 }
3436 None => {
3437 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3438 book.insert(key, (seq, p.clone()));
3439 }
3440 }
3441 }
3442 self.trade_dirty.store(true, Ordering::Release);
3443 self.bump();
3444 }
3445
3446 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
3448 let key = a.en_accountid();
3449 {
3450 let mut book = self.trade_funds.lock().unwrap();
3451 match book.get_mut(&key) {
3452 Some((_, cur)) => {
3453 cur.balance = a.balance;
3454 cur.frozen = a.frozen;
3455 }
3456 None => {
3457 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3458 book.insert(key, (seq, a.clone()));
3459 }
3460 }
3461 }
3462 self.trade_dirty.store(true, Ordering::Release);
3463 self.bump();
3464 }
3465
3466 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
3468 let mut removed = 0usize;
3469 {
3470 let mut book = self.trade_orders.lock().unwrap();
3471 let before = book.len();
3472 book.retain(|_, (_, o)| o.gateway_name != gateway);
3473 removed += before - book.len();
3474 }
3475 {
3476 let mut book = self.trade_trades.lock().unwrap();
3477 let before = book.len();
3478 book.retain(|(_, t)| t.gateway_name != gateway);
3479 removed += before - book.len();
3480 let prefix = format!("{gateway}.");
3483 self.trade_ids
3484 .lock()
3485 .unwrap()
3486 .retain(|id| !id.starts_with(&prefix));
3487 }
3488 {
3489 let mut book = self.trade_positions.lock().unwrap();
3490 let before = book.len();
3491 book.retain(|_, (_, p)| p.gateway_name != gateway);
3492 removed += before - book.len();
3493 }
3494 {
3495 let mut book = self.trade_funds.lock().unwrap();
3496 let before = book.len();
3497 book.retain(|_, (_, a)| a.gateway_name != gateway);
3498 removed += before - book.len();
3499 }
3500 if removed > 0 {
3501 self.trade_dirty.store(true, Ordering::Release);
3502 self.bump();
3503 }
3504 removed
3505 }
3506
3507 #[must_use]
3509 pub fn trade_dropped(&self) -> u64 {
3510 self.trade_dropped.load(Ordering::Acquire)
3511 }
3512
3513 pub fn trade_snap(&self) -> Arc<TradeSnap> {
3518 let mut cache = self.trade_snap.lock().unwrap();
3519 if self.trade_dirty.swap(false, Ordering::AcqRel) {
3520 *cache = Arc::new(self.build_trade_snap());
3521 }
3522 Arc::clone(&cache)
3523 }
3524
3525 fn build_trade_snap(&self) -> TradeSnap {
3528 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
3529 where
3530 T: Clone,
3531 {
3532 let mut v: Vec<(u64, &String, &T)> =
3533 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
3534 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
3535 v.into_iter().map(|(_, _, v)| v.clone()).collect()
3536 }
3537 TradeSnap {
3538 orders: by_seq(&self.trade_orders.lock().unwrap()),
3539 trades: self
3540 .trade_trades
3541 .lock()
3542 .unwrap()
3543 .iter()
3544 .map(|(_, t)| t.clone())
3545 .collect(),
3546 positions: by_seq(&self.trade_positions.lock().unwrap()),
3547 accounts: by_seq(&self.trade_funds.lock().unwrap()),
3548 }
3549 }
3550
3551 pub fn ctp_form(&self) -> CtpForm {
3555 self.ctp_form.lock().unwrap().clone()
3556 }
3557
3558 pub fn set_ctp_form(&self, f: CtpForm) {
3560 *self.ctp_form.lock().unwrap() = f;
3561 self.bump();
3562 }
3563
3564 pub fn row_selected(&self, table: &str, key: &str) -> bool {
3568 self.row_sel
3569 .lock()
3570 .unwrap()
3571 .get(table)
3572 .is_some_and(|k| k == key)
3573 }
3574
3575 pub fn select_row(&self, table: &str, key: &str) {
3577 self.row_sel
3578 .lock()
3579 .unwrap()
3580 .insert(table.to_string(), key.to_string());
3581 }
3582}