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().append(true).create(true).open(&path) {
100 use std::io::Write as _;
101 let _ = writeln!(f, "{} | {}", entry.at, entry.text);
102 }
103 }
104
105 pub fn push(&mut self, line: impl Into<String>) {
107 let at = chrono::Utc::now()
108 .with_timezone(&beijing_tz())
109 .format("%H:%M:%S")
110 .to_string();
111 let entry = LogEntry {
112 at,
113 text: line.into(),
114 };
115 if self.spill_dir.is_some() {
116 self.spill_write(&entry);
117 }
118 if self.cap == 0 {
119 self.spilled += 1;
120 return;
121 }
122 if self.lines.len() == self.cap {
123 self.lines.pop_front();
124 self.spilled += 1;
125 }
126 self.lines.push_back(entry);
127 }
128
129 pub fn len(&self) -> usize {
131 self.lines.len()
132 }
133
134 pub fn is_empty(&self) -> bool {
136 self.lines.is_empty()
137 }
138
139 pub fn spilled(&self) -> u64 {
141 self.spilled
142 }
143
144 pub fn snapshot(&self) -> Vec<String> {
146 self.lines.iter().map(|e| e.text.clone()).collect()
147 }
148
149 pub fn entries(&self) -> Vec<LogEntry> {
151 self.lines.iter().cloned().collect()
152 }
153
154 pub fn clear(&mut self) {
156 self.lines.clear();
157 }
158}
159
160#[derive(Debug, Clone, Default)]
162pub struct BusSnap {
163 pub rows: Vec<PlugRow>,
165 pub services: Vec<(String, Vec<String>)>,
167}
168
169#[derive(Debug, Clone, PartialEq)]
171pub enum UiCmd {
172 Start { factory: String, name: String },
175 Remove { name: String },
177 Reload {
179 interval: i64,
180 symbol: String,
181 bars: Vec<BarData>,
182 },
183 SendOrder {
189 req: ek_data::order_request::OrderRequest,
190 gateway: String,
191 mode: ek_oms::LockMode,
192 },
193 CancelOrder { orderid: String, gateway: String },
197 CtpConfigure { gateway: String, form: CtpForm },
201 Connect { gateway: String, form: CtpForm },
206 CtpSubscribe {
209 gateway: String,
210 symbols: Vec<String>,
211 },
212 Disconnect { gateway: String },
216 CfgList,
218 CfgDelete { ns: String },
220 CfgResave { ns: String },
223}
224
225pub const KLINE_CELL_CAP: usize = 4096;
227
228pub const MAX_ACCOUNTS: usize = 64;
233
234pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
239
240pub const ACCOUNT_STALE_SECS: i64 = 30;
246
247pub use ek_ctp::driver::CHAIN_STATE_PREFIX;
253
254pub const LIST_POLL_TICKS: u32 = 4;
260
261pub const STATUS_ASKS_PER_TICK: usize = 2;
266
267pub const TRADE_CAP: usize = 5000;
272
273pub const PENDING_WARN_FLOOR: u64 = 4_096;
275pub const PENDING_CRIT_FLOOR: u64 = 32_768;
277
278#[must_use]
286pub fn pending_thresholds(soft_cap: u64) -> (u64, u64) {
287 let warn = (soft_cap / 16).max(PENDING_WARN_FLOOR);
288 let crit = (soft_cap / 4).max(PENDING_CRIT_FLOOR);
289 (warn.min(soft_cap.max(1)), crit.min(soft_cap.max(1)))
290}
291
292#[derive(Debug, Clone, Copy, PartialEq, Eq)]
294pub enum ConfirmStep {
295 Armed,
296 Fired,
297 Switched,
298}
299
300#[must_use]
303pub fn is_quarantined(state: &str) -> bool {
304 state.contains("Quarantin")
305}
306
307#[derive(Debug, Clone)]
311pub struct KlineFetch {
312 pub symbol: String,
313 pub interval_secs: i64,
314}
315
316#[derive(Debug, Default)]
318pub struct MarketStore {
319 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
320 frames: AtomicU64,
322 updates: AtomicU64,
324 reloads: AtomicU64,
326}
327
328#[derive(Debug, Clone, PartialEq)]
330pub struct CellView {
331 pub symbol: String,
332 pub interval: i64,
333 pub bars: Vec<BarData>,
334}
335
336impl MarketStore {
337 pub fn on_frame(&self, frame: &BarsFrame) {
339 let mut cells = self.cells.lock().unwrap();
340 for b in &frame.bars {
341 let key = (b.symbol.clone(), frame.interval);
342 let cell = cells.entry(key).or_default();
343 match cell.last() {
344 Some(last) if b.datetime < last.datetime => continue,
345 Some(last) if b.datetime == last.datetime => {
346 let slot = cell.last_mut().expect("len>1");
347 *slot = b.clone();
348 }
349 _ => {
350 cell.push(b.clone());
351 if cell.len() > KLINE_CELL_CAP {
352 cell.remove(0);
353 }
354 }
355 }
356 }
357 self.frames.fetch_add(1, Ordering::AcqRel);
358 }
359
360 pub fn on_update(&self, b: &BarData) {
362 let mut cells = self.cells.lock().unwrap();
363 let iv = b
364 .interval
365 .unwrap_or(ek_data::constant::Interval::MINUTE)
366 .seconds();
367 let key = (b.symbol.clone(), iv);
368 let cell = cells.entry(key).or_default();
369 match cell.last() {
370 Some(last) if b.datetime == last.datetime => {
371 let slot = cell.last_mut().expect("len>1");
372 *slot = b.clone();
373 }
374 Some(last) if b.datetime < last.datetime => {}
375 _ => {
376 cell.push(b.clone());
377 if cell.len() > KLINE_CELL_CAP {
378 cell.remove(0);
379 }
380 }
381 }
382 self.updates.fetch_add(1, Ordering::AcqRel);
383 }
384
385 pub fn note_history(&self, symbol: &str, iv_secs: i64, bars: &[BarData]) {
391 let mut cells = self.cells.lock().unwrap();
392 let cell = cells.entry((symbol.to_string(), iv_secs)).or_default();
393 cell.extend(bars.iter().cloned());
394 cell.sort_unstable_by_key(|b| b.datetime);
395 cell.dedup_by(|a, b| a.datetime == b.datetime);
396 if cell.len() > KLINE_CELL_CAP {
397 let excess = cell.len() - KLINE_CELL_CAP;
398 cell.drain(0..excess);
399 }
400 self.frames.fetch_add(1, Ordering::AcqRel);
401 }
402
403 pub fn note_reload(&self) {
405 self.reloads.fetch_add(1, Ordering::AcqRel);
406 }
407
408 pub fn cells(&self) -> Vec<CellView> {
410 self.cells
411 .lock()
412 .unwrap()
413 .iter()
414 .map(|((sym, iv), bars)| CellView {
415 symbol: sym.clone(),
416 interval: *iv,
417 bars: bars.clone(),
418 })
419 .collect()
420 }
421
422 pub fn find_cell(&self, sym: &str) -> Option<CellView> {
427 self.cells()
428 .into_iter()
429 .find(|c| c.symbol == sym || c.symbol.ends_with(&format!(".{sym}")))
430 }
431
432 pub fn counts(&self) -> (u64, u64, u64) {
434 (
435 self.frames.load(Ordering::Acquire),
436 self.updates.load(Ordering::Acquire),
437 self.reloads.load(Ordering::Acquire),
438 )
439 }
440}
441
442#[derive(Debug, Default, Clone, PartialEq)]
445pub struct ReloadForm {
446 pub interval: String,
447 pub symbol: String,
448 pub csv: String,
449}
450
451pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
458 let g = raw.trim();
459 if !g.is_empty() {
460 if g.chars().any(|c| c.is_whitespace()) {
461 return Err(format!("网关解析: 含空白:{raw:?}"));
462 }
463 return Ok(g.to_string());
464 }
465 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
466 Some(f) => Ok(f.to_string()),
467 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
468 }
469}
470
471#[derive(Debug, Clone, PartialEq)]
474pub struct CtpForm {
475 pub broker_id: String,
476 pub td_addr: String,
477 pub td_user: String,
478 pub td_password: String,
479 pub td_auth_code: String,
480 pub td_app_id: String,
481 pub md_addr: String,
482 pub md_user: String,
483 pub md_password: String,
484 pub symbols: String,
486 pub accept_futures: bool,
489 pub accept_options: bool,
490}
491
492impl Default for CtpForm {
495 fn default() -> Self {
496 Self {
497 broker_id: String::new(),
498 td_addr: String::new(),
499 td_user: String::new(),
500 td_password: String::new(),
501 td_auth_code: String::new(),
502 td_app_id: String::new(),
503 md_addr: String::new(),
504 md_user: String::new(),
505 md_password: String::new(),
506 symbols: String::new(),
507 accept_futures: true,
508 accept_options: false,
509 }
510 }
511}
512
513impl CtpForm {
514 #[must_use]
518 pub fn product_classes(&self) -> Vec<String> {
519 let mut codes = Vec::new();
520 if self.accept_futures {
521 codes.push("1".to_string());
522 }
523 if self.accept_options {
524 codes.push("2".to_string());
525 }
526 codes
527 }
528
529 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
531 ek_ctp::plugin::ConfigureReq {
532 broker_id: self.broker_id.trim().to_string(),
533 td_addr: self.td_addr.trim().to_string(),
534 td_user: self.td_user.trim().to_string(),
535 td_password: self.td_password.clone().into(),
536 td_auth_code: self.td_auth_code.trim().to_string(),
537 td_app_id: self.td_app_id.trim().to_string(),
538 md_addr: self.md_addr.trim().to_string(),
539 md_user: self.md_user.trim().to_string(),
540 md_password: self.md_password.clone().into(),
541 product_classes: self.product_classes(),
542 }
543 }
544
545 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
547 let list: Vec<String> = self
548 .symbols
549 .split([',', ',', ' '])
550 .map(|s| s.trim().to_string())
551 .filter(|s| !s.is_empty())
552 .collect();
553 if list.is_empty() {
554 return Err("订阅解析: 标的位空".to_string());
555 }
556 Ok(list)
557 }
558
559 pub fn validate(&self) -> Result<(), String> {
568 fn front(name: &str, v: &str) -> Result<(), String> {
569 let v = v.trim();
570 let Some(rest) = v.strip_prefix("tcp://") else {
571 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
572 };
573 let Some((host, port)) = rest.rsplit_once(':') else {
574 return Err(format!("{name} 缺端口:{name} {:?}", v));
575 };
576 if host.trim().is_empty() {
577 return Err(format!("{name} 主机位空:{}", v));
578 }
579 let port: u32 = port
580 .parse()
581 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
582 if port == 0 || port > 65535 {
583 return Err(format!("{name} 端口越界(1..=65535):{}", v));
584 }
585 Ok(())
586 }
587 if self.broker_id.trim().is_empty() {
588 return Err("broker 位空".to_string());
589 }
590 if self.td_user.trim().is_empty() {
591 return Err("交易账号位空".to_string());
592 }
593 front("td 前置", &self.td_addr)?;
594 if !self.md_addr.trim().is_empty() {
595 front("md 前置", &self.md_addr)?;
596 }
597 if self.td_app_id.trim().is_empty() {
598 return Err("AppID 位空(CTP 登录必需)".to_string());
599 }
600 if self.td_auth_code.trim().is_empty() {
601 return Err("AuthCode 位空(CTP 登录必需)".to_string());
602 }
603 Ok(())
604 }
605}
606
607#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
614pub enum AccountStage {
615 #[default]
617 NotAssembled,
618 Assembling,
620 Assembled,
622 AssembleFailed,
624}
625
626impl AccountStage {
627 #[must_use]
629 pub fn label(self) -> &'static str {
630 match self {
631 AccountStage::NotAssembled => "未装配",
632 AccountStage::Assembling => "装配中…",
633 AccountStage::Assembled => "已装配·登录态未知",
636 AccountStage::AssembleFailed => "装配失败",
637 }
638 }
639}
640
641#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
646pub enum LoginState {
647 #[default]
652 Unknown,
653 Online,
655 LoginFailed,
657 FrontDown,
659 QueryStarved,
661 Stopped,
663}
664
665impl LoginState {
666 #[must_use]
668 pub fn label(self) -> &'static str {
669 match self {
670 LoginState::Unknown => "登录态未知",
671 LoginState::Online => "在线",
672 LoginState::LoginFailed => "登录失败",
673 LoginState::FrontDown => "掉线",
674 LoginState::QueryStarved => "查询断供",
675 LoginState::Stopped => "已断开",
676 }
677 }
678
679 #[must_use]
681 pub fn is_alarm(self) -> bool {
682 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
683 }
684}
685
686fn count_note(label: &str, n: u32) -> Option<String> {
688 (n > 0).then(|| format!("{label} {n} 次"))
689}
690
691fn brief_counter_reason(full: &str) -> Option<String> {
698 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
699 let body = tail.trim().trim_start_matches("CTP:").trim();
700 if body.is_empty() {
701 return None;
702 }
703 let code = full
704 .rfind("代码:")
705 .map(|i| {
706 let rest = &full[i + "代码:".len()..];
707 let end = rest.find(',').unwrap_or(rest.len());
708 rest[..end].trim().to_string()
709 })
710 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
711 Some(match code {
712 Some(c) => format!("{body}({c})"),
713 None => body.to_string(),
714 })
715}
716
717#[derive(Debug, Clone, PartialEq, Default)]
723pub struct AccountRow {
724 pub gateway: String,
726 pub alias: String,
728 pub broker_id: String,
729 pub account: String,
730 pub md_source: bool,
732 pub form: CtpForm,
734 pub restored: bool,
740 pub auto_connect: bool,
745 pub password_ref: bool,
750 pub stage: AccountStage,
752 pub fail_note: Option<String>,
754 pub effective_classes: Vec<String>,
759 pub connected: bool,
762 pub login: LoginState,
766 pub status_note: Option<String>,
769 pub login_brief: Option<String>,
776 pub chain_note: Option<String>,
784}
785
786impl AccountRow {
787 pub fn derive(form: &CtpForm) -> Result<Self, String> {
789 Self::derive_with(form, &[])
790 }
791
792 pub fn derive_with(
802 form: &CtpForm,
803 taken: &[(String, String, String, String)],
804 ) -> Result<Self, String> {
805 let broker = form.broker_id.trim();
806 let account = form.td_user.trim();
807 let gateway = crate::appconfig::unique_gw_name(
808 taken,
809 broker,
810 account,
811 form.td_addr.trim(),
812 )?;
813 Ok(Self {
814 gateway,
815 alias: String::new(),
816 broker_id: broker.to_string(),
817 account: account.to_string(),
818 md_source: false,
819 form: form.clone(),
820 stage: AccountStage::NotAssembled,
821 fail_note: None,
822 effective_classes: Vec::new(),
823 connected: false,
824 login: LoginState::Unknown,
825 status_note: None,
826 login_brief: None,
827 chain_note: None,
828 restored: false,
831 auto_connect: false,
832 password_ref: false,
833 })
834 }
835
836 #[must_use]
842 pub fn status_label(&self) -> String {
843 if self.stage == AccountStage::Assembled {
844 match self.login {
845 LoginState::Unknown => "已装配·登录态未知".to_string(),
846 other => format!("已装配·{}", other.label()),
847 }
848 } else {
849 self.stage.label().to_string()
850 }
851 }
852
853 pub fn apply_status(
866 &mut self,
867 st: &ek_ctp::plugin::GatewayStatus,
868 now_ms: i64,
869 reject_limit: u32,
870 stale_secs: i64,
871 ) {
872 let td_side = !self.form.td_addr.trim().is_empty();
873 let (login, note) = if td_side {
874 if st.td_login_failed && !st.td_logged_in {
875 (
876 LoginState::LoginFailed,
877 st.td_last_error
878 .clone()
879 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
880 )
881 } else if st.td_logged_in {
882 if st.td_query_reject_streak >= reject_limit {
883 (
884 LoginState::QueryStarved,
885 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
886 )
887 } else if let Some(last) = st.td_last_account_ms {
888 let idle_ms = now_ms.saturating_sub(last);
889 if idle_ms > stale_secs.saturating_mul(1_000) {
890 (
891 LoginState::QueryStarved,
892 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
893 )
894 } else {
895 (LoginState::Online, count_note("掉线", st.td_disconnects))
896 }
897 } else {
898 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
905 match since_login_ms {
906 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
907 LoginState::QueryStarved,
908 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
909 ),
910 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
911 }
912 }
913 } else if self.login == LoginState::Stopped {
914 let extra = if st.td_disconnects > 0 {
915 format!("(此前掉线 {} 次)", st.td_disconnects)
916 } else {
917 String::new()
918 };
919 (
920 LoginState::Stopped,
921 Some(format!("会话已停{extra};『连接』可恢复")),
922 )
923 } else if st.td_disconnects > 0 {
924 (
925 LoginState::FrontDown,
926 Some(format!("掉线 {} 次", st.td_disconnects)),
927 )
928 } else {
929 (LoginState::Unknown, st.td_last_error.clone())
930 }
931 } else if st.md_logged_in {
932 (
933 LoginState::Online,
934 count_note("掉线", st.md_disconnects)
935 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
936 )
937 } else if self.login == LoginState::Stopped {
938 (
939 LoginState::Stopped,
940 Some("会话已停;『连接』可恢复".to_string()),
941 )
942 } else if st.md_disconnects > 0 {
943 (
944 LoginState::FrontDown,
945 Some(format!("掉线 {} 次", st.md_disconnects)),
946 )
947 } else {
948 (LoginState::Unknown, st.md_last_error.clone())
949 };
950 let note = if td_side && st.td_instrument_rejects > 0 {
955 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
956 match note {
957 Some(base) if base.contains("合约查询被拒") => Some(base),
959 Some(base) => Some(format!("{base};{add}")),
960 None => Some(add),
961 }
962 } else {
963 note
964 };
965 self.login = login;
966 self.status_note = note;
967 self.login_brief = if login.is_alarm() {
970 st.td_last_error
971 .as_deref()
972 .or(st.md_last_error.as_deref())
973 .and_then(brief_counter_reason)
974 } else {
975 None
976 };
977 }
978
979 #[must_use]
981 pub fn display_name(&self) -> String {
982 let a = self.alias.trim();
983 if a.is_empty() {
984 self.account.clone()
985 } else {
986 a.to_string()
987 }
988 }
989}
990
991#[derive(Debug, Clone, PartialEq, Default)]
994pub struct TradeSnap {
995 pub orders: Vec<ek_data::order::OrderData>,
996 pub trades: Vec<ek_data::trade::TradeData>,
997 pub positions: Vec<ek_data::position::PositionData>,
998 pub accounts: Vec<ek_data::account::AccountData>,
1000}
1001
1002pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
1005 let interval: i64 = form
1006 .interval
1007 .trim()
1008 .parse()
1009 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
1010 let symbol = form.symbol.trim().to_string();
1011 if symbol.is_empty() {
1012 return Err("回灌解析: symbol 空".to_string());
1013 }
1014 if form.csv.trim().is_empty() {
1015 return Err("回灌解析: csv 空".to_string());
1016 }
1017 let mut bars = Vec::new();
1018 for (i, raw) in form.csv.lines().enumerate() {
1019 let line = raw.trim();
1020 if line.is_empty() || line.starts_with('#') {
1021 continue;
1022 }
1023 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
1024 if parts.len() != 6 {
1025 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
1026 }
1027 let num = |s: &str| -> Result<f64, String> {
1028 s.parse::<f64>()
1029 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
1030 };
1031 let dt_s: i64 = parts[0]
1032 .parse()
1033 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
1034 bars.push(BarData {
1035 symbol: symbol.clone(),
1036 datetime: dt_s * 1_000_000_000,
1037 interval: Some(
1038 ek_dm::wire::interval_of_seconds(interval)
1039 .unwrap_or(ek_data::constant::Interval::MINUTE),
1040 ),
1041 open_price: num(parts[1])?,
1042 high_price: num(parts[2])?,
1043 low_price: num(parts[3])?,
1044 close_price: num(parts[4])?,
1045 volume: num(parts[5])?,
1046 ..BarData::default()
1047 });
1048 }
1049 if bars.is_empty() {
1050 return Err("回灌解析: 无有效行".to_string());
1051 }
1052 Ok((interval, symbol, bars))
1053}
1054
1055#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1060pub enum TopLeftTab {
1061 #[default]
1062 Orders,
1063 Active,
1064}
1065
1066impl TopLeftTab {
1067 pub fn label(self) -> &'static str {
1068 match self {
1069 Self::Orders => "委托",
1070 Self::Active => "未成交",
1071 }
1072 }
1073}
1074
1075#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1077pub enum BottomRightTab {
1078 #[default]
1079 Account,
1080 Logs,
1081}
1082
1083impl BottomRightTab {
1084 pub fn label(self) -> &'static str {
1085 match self {
1086 Self::Account => "资金",
1087 Self::Logs => "日志",
1088 }
1089 }
1090}
1091
1092#[derive(Debug, Clone, Copy, PartialEq)]
1095pub struct DataSplit {
1096 pub col: f32,
1098 pub row: f32,
1100}
1101
1102impl Default for DataSplit {
1103 fn default() -> Self {
1104 Self {
1105 col: crate::splitters::DATA_SPLIT_DEFAULT,
1106 row: crate::splitters::DATA_SPLIT_DEFAULT,
1107 }
1108 }
1109}
1110
1111pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
1114 ek_data::constant::Exchange::SHFE,
1115 ek_data::constant::Exchange::DCE,
1116 ek_data::constant::Exchange::CZCE,
1117 ek_data::constant::Exchange::CFFEX,
1118 ek_data::constant::Exchange::INE,
1119 ek_data::constant::Exchange::GFEX,
1120];
1121
1122#[derive(Debug, Default)]
1132pub struct TickStore {
1133 ticks: BTreeMap<String, ek_data::tick::TickData>,
1134}
1135
1136impl TickStore {
1137 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1140 self.ticks
1141 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1142 }
1143
1144 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1146 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1147 v.sort_by(|a, b| {
1148 (a.exchange.as_str().to_string(), a.symbol.clone())
1149 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1150 });
1151 v
1152 }
1153
1154 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1156 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1157 for t in self.rows() {
1158 let ex = t.exchange.as_str().to_string();
1159 match out.last_mut() {
1160 Some((e, v)) if *e == ex => v.push(t),
1161 _ => out.push((ex, vec![t])),
1162 }
1163 }
1164 out
1165 }
1166
1167 pub fn len(&self) -> usize {
1168 self.ticks.len()
1169 }
1170
1171 pub fn is_empty(&self) -> bool {
1172 self.ticks.is_empty()
1173 }
1174}
1175
1176#[derive(Debug, Clone, PartialEq)]
1178pub struct OrderForm {
1179 pub exchange: String,
1181 pub symbol: String,
1183 pub direction: ek_data::constant::Direction,
1184 pub offset: ek_data::constant::Offset,
1185 pub order_type: ek_data::constant::OrderType,
1186 pub price: String,
1188 pub volume: String,
1189 pub gateway: String,
1191 pub follow_price: bool,
1193 pub auto_split: bool,
1196}
1197
1198impl Default for OrderForm {
1199 fn default() -> Self {
1200 Self {
1201 exchange: "SHFE".to_string(),
1202 symbol: String::new(),
1203 direction: ek_data::constant::Direction::LONG,
1204 offset: ek_data::constant::Offset::OPEN,
1205 order_type: ek_data::constant::OrderType::LIMIT,
1206 price: String::new(),
1207 volume: "1".to_string(),
1208 gateway: String::new(),
1209 follow_price: true,
1210 auto_split: true,
1211 }
1212 }
1213}
1214
1215impl OrderForm {
1216 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1218 use std::str::FromStr;
1219 let symbol = self.symbol.trim();
1220 if symbol.is_empty() {
1221 return Err("下单: 代码为空".to_string());
1222 }
1223 if symbol.contains('.') {
1224 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1225 }
1226 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1227 .map_err(|e| format!("下单: {e}"))?;
1228 let price: f64 = self
1229 .price
1230 .trim()
1231 .parse()
1232 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1233 let volume: f64 = self
1234 .volume
1235 .trim()
1236 .parse()
1237 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1238 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1239 return Err("下单: 限价单价格须为正".to_string());
1240 }
1241 if volume <= 0.0 {
1242 return Err("下单: 数量须为正".to_string());
1243 }
1244 Ok(ek_data::order_request::OrderRequest {
1245 symbol: symbol.to_string(),
1246 exchange,
1247 direction: self.direction,
1248 order_type: self.order_type,
1249 volume,
1250 price,
1251 offset: self.offset,
1252 ..ek_data::order_request::OrderRequest::default()
1253 })
1254 }
1255}
1256
1257#[derive(Debug, Default)]
1263pub struct ContractStore {
1264 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1265}
1266
1267impl ContractStore {
1268 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1269 self.contracts
1270 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1271 }
1272
1273 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1275 let p = pattern.trim();
1276 self.contracts
1277 .values()
1278 .filter(|c| {
1279 p.is_empty()
1280 || c.symbol.contains(p)
1281 || c.exchange.as_str().eq_ignore_ascii_case(p)
1282 || c.name.contains(p)
1283 })
1284 .cloned()
1285 .collect()
1286 }
1287
1288 pub fn len(&self) -> usize {
1289 self.contracts.len()
1290 }
1291
1292 pub fn is_empty(&self) -> bool {
1293 self.contracts.is_empty()
1294 }
1295}
1296
1297#[derive(Debug, Default)]
1299pub struct UiShared {
1300 logs: Mutex<LogRing>,
1301 snap: Mutex<Arc<BusSnap>>,
1302 market: MarketStore,
1303 ctp_form: Mutex<CtpForm>,
1304 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1307 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1308 trade_ids: Mutex<HashSet<String>>,
1316 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1317 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1318 trade_seq: AtomicU64,
1320 trade_dropped: AtomicU64,
1322 trade_dirty: AtomicBool,
1324 trade_snap: Mutex<Arc<TradeSnap>>,
1326 tick_store: Mutex<TickStore>,
1327 order_form: Mutex<OrderForm>,
1328 data_split: Mutex<DataSplit>,
1331 selected_exchange: Mutex<String>,
1333 topleft_tab: Mutex<TopLeftTab>,
1335 bottomright_tab: Mutex<BottomRightTab>,
1337 ops_window: AtomicBool,
1338 accounts_window: AtomicBool,
1341 contracts_window: AtomicBool,
1342 contract_store: Mutex<ContractStore>,
1343 contract_query: Mutex<String>,
1344 contract_selected: Mutex<Vec<String>>,
1345 kline_windows: Mutex<Vec<String>>,
1346 kline_fetches: Mutex<Vec<KlineFetch>>,
1348 kline_fetched: Mutex<HashSet<(String, i64)>>,
1350 rev: AtomicU64,
1351 stop: AtomicBool,
1353 cmds: Mutex<VecDeque<UiCmd>>,
1355 pending_confirm: Mutex<Option<String>>,
1357 accounts: Mutex<Vec<AccountRow>>,
1359 account_edit_window: AtomicBool,
1361 editing_account: Mutex<Option<String>>,
1363 edit_alias: Mutex<String>,
1365 edit_error: Mutex<Option<String>>,
1367 md_source: Mutex<String>,
1369 subscriptions: Mutex<Vec<String>>,
1371 row_sel: Mutex<BTreeMap<String, String>>,
1374 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1378 status_cursor: AtomicUsize,
1384 status_seed: Mutex<Vec<String>>,
1387 list_tick: AtomicU32,
1389 list_due: AtomicBool,
1393 seed_due: AtomicBool,
1397 pipeline_due: AtomicBool,
1399 cfg_due: AtomicBool,
1401 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1404 cfg_delete_arm: Mutex<Option<String>>,
1406 theme: AtomicU8,
1410 theme_persist_due: AtomicBool,
1413 auto_connect_persist_due: AtomicBool,
1417 alias_persist_due: AtomicBool,
1421 archive_purge: Mutex<Vec<String>>,
1425 bad_payload: Mutex<BTreeMap<String, u64>>,
1427 pipeline_lines: Mutex<Vec<String>>,
1430 status_thresholds: Mutex<Option<(u32, i64)>>,
1433 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1435 toasts: crate::toast::ToastsCell,
1438 alerted: Mutex<std::collections::BTreeSet<String>>,
1440 exit_confirm_open: AtomicBool,
1442 exit_confirmed: AtomicBool,
1444 assembly_locked: AtomicBool,
1447}
1448
1449impl UiShared {
1450 pub fn new() -> Self {
1452 Self::default()
1453 }
1454
1455 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1459 self.tick_store.lock().unwrap().put(t);
1460 self.bump();
1461 }
1462
1463 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1465 self.tick_store.lock().unwrap().rows()
1466 }
1467
1468 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1470 self.tick_store.lock().unwrap().groups()
1471 }
1472
1473 pub fn tick_count(&self) -> usize {
1475 self.tick_store.lock().unwrap().len()
1476 }
1477
1478 pub fn order_form(&self) -> OrderForm {
1480 self.order_form.lock().unwrap().clone()
1481 }
1482
1483 pub fn set_order_form(&self, f: OrderForm) {
1485 *self.order_form.lock().unwrap() = f;
1486 self.bump();
1487 }
1488
1489 pub fn selected_exchange(&self) -> String {
1493 let s = self.selected_exchange.lock().unwrap().clone();
1494 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1495 s
1496 } else {
1497 MARKET_ORDER[0].as_str().to_string()
1498 }
1499 }
1500
1501 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1503 *self.selected_exchange.lock().unwrap() = s.into();
1504 self.bump();
1505 }
1506
1507 pub fn topleft_tab(&self) -> TopLeftTab {
1509 *self.topleft_tab.lock().unwrap()
1510 }
1511
1512 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1514 *self.topleft_tab.lock().unwrap() = t;
1515 self.bump();
1516 }
1517
1518 pub fn bottomright_tab(&self) -> BottomRightTab {
1520 *self.bottomright_tab.lock().unwrap()
1521 }
1522
1523 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1525 *self.bottomright_tab.lock().unwrap() = t;
1526 self.bump();
1527 }
1528
1529 pub fn data_split(&self) -> DataSplit {
1531 *self.data_split.lock().unwrap()
1532 }
1533
1534 pub fn set_data_split(&self, s: DataSplit) {
1537 let s = DataSplit {
1538 col: crate::splitters::clamp_frac(s.col),
1539 row: crate::splitters::clamp_frac(s.row),
1540 };
1541 let mut g = self.data_split.lock().unwrap();
1542 if g.col != s.col || g.row != s.row {
1543 *g = s;
1544 self.bump();
1545 }
1546 }
1547
1548 pub fn reset_layout(&self) {
1552 *self.data_split.lock().unwrap() = DataSplit::default();
1553 *self.selected_exchange.lock().unwrap() = String::new();
1554 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1555 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1556 self.bump();
1557 }
1558
1559 pub fn ops_window_open(&self) -> bool {
1563 self.ops_window.load(Ordering::Acquire)
1564 }
1565
1566 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1569 self.table_sorts
1570 .lock()
1571 .unwrap()
1572 .get(table)
1573 .copied()
1574 .flatten()
1575 }
1576
1577 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1579 let mut m = self.table_sorts.lock().unwrap();
1580 let asc = match m.get(table).copied().flatten() {
1581 Some((c, a)) if c == col => !a,
1582 _ => true,
1583 };
1584 m.insert(table.to_string(), Some((col, asc)));
1585 }
1586
1587 pub fn set_ops_window_open(&self, v: bool) {
1588 self.ops_window.store(v, Ordering::Release);
1589 self.bump();
1590 }
1591
1592 pub fn accounts_window_open(&self) -> bool {
1596 self.accounts_window.load(Ordering::Acquire)
1597 }
1598
1599 pub fn set_accounts_window_open(&self, v: bool) {
1601 self.accounts_window.store(v, Ordering::Release);
1602 self.bump();
1603 }
1604
1605 pub fn account_edit_open(&self) -> bool {
1607 self.account_edit_window.load(Ordering::Acquire)
1608 }
1609
1610 pub fn editing_account(&self) -> Option<String> {
1612 self.editing_account.lock().unwrap().clone()
1613 }
1614
1615 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1619 match gateway {
1620 Some(gw) => {
1621 let Some(row) = self.account(gw) else {
1622 return false;
1623 };
1624 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1625 *self.edit_alias.lock().unwrap() = row.alias.clone();
1626 self.set_ctp_form(row.form.clone());
1627 }
1628 None => {
1629 *self.editing_account.lock().unwrap() = None;
1630 *self.edit_alias.lock().unwrap() = String::new();
1631 let draft = self.draft_from_existing();
1632 self.set_ctp_form(draft);
1633 }
1634 }
1635 *self.edit_error.lock().unwrap() = None;
1636 self.account_edit_window.store(true, Ordering::Release);
1637 self.bump();
1638 true
1639 }
1640
1641 pub fn end_account_edit(&self) {
1643 *self.editing_account.lock().unwrap() = None;
1644 *self.edit_alias.lock().unwrap() = String::new();
1645 *self.edit_error.lock().unwrap() = None;
1646 self.account_edit_window.store(false, Ordering::Release);
1647 let draft = self.draft_from_existing();
1648 self.set_ctp_form(draft);
1649 self.bump();
1650 }
1651
1652 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1658 if self.editing_account().as_deref() == Some(gateway) {
1659 self.end_account_edit();
1660 }
1661 }
1662
1663 pub fn edit_alias(&self) -> String {
1665 self.edit_alias.lock().unwrap().clone()
1666 }
1667
1668 pub fn set_edit_alias(&self, v: impl Into<String>) {
1670 *self.edit_alias.lock().unwrap() = v.into();
1671 self.bump();
1672 }
1673
1674 pub fn edit_error(&self) -> Option<String> {
1676 self.edit_error.lock().unwrap().clone()
1677 }
1678
1679 pub fn set_edit_error(&self, e: Option<String>) {
1681 *self.edit_error.lock().unwrap() = e;
1682 self.bump();
1683 }
1684
1685 pub fn account_edit_dirty(&self) -> bool {
1687 let editing = self.editing_account();
1688 let draft = self.ctp_form();
1689 match editing {
1690 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1691 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1692 }
1693 }
1694
1695 pub fn accounts(&self) -> Vec<AccountRow> {
1697 self.accounts.lock().unwrap().clone()
1698 }
1699
1700 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1702 self.accounts
1703 .lock()
1704 .unwrap()
1705 .iter()
1706 .find(|a| a.gateway == gateway)
1707 .cloned()
1708 }
1709
1710 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1716 let mut acc = self.accounts.lock().unwrap();
1717 if acc.iter().any(|a| a.gateway == row.gateway) {
1718 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1719 }
1720 let alias = row.alias.trim();
1721 if !alias.is_empty()
1722 && acc
1723 .iter()
1724 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1725 {
1726 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1727 }
1728 if acc.len() >= MAX_ACCOUNTS {
1729 return Err(format!(
1730 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1731 ));
1732 }
1733 let has_alias = !row.alias.trim().is_empty();
1734 acc.push(row);
1735 drop(acc);
1736 if has_alias {
1737 self.alias_persist_due.store(true, Ordering::Release);
1739 }
1740 self.bump();
1741 Ok(())
1742 }
1743
1744 #[must_use]
1746 pub fn can_add_account(&self) -> bool {
1747 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1748 }
1749
1750 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1756 let alias = alias.trim().to_string();
1757 let mut acc = self.accounts.lock().unwrap();
1758 if !alias.is_empty()
1759 && acc
1760 .iter()
1761 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1762 {
1763 return false;
1764 }
1765 match acc.iter_mut().find(|a| a.gateway == gateway) {
1766 Some(a) => {
1767 a.alias = alias;
1768 drop(acc);
1769 self.alias_persist_due.store(true, Ordering::Release);
1770 self.bump();
1771 true
1772 }
1773 None => false,
1774 }
1775 }
1776
1777 pub fn note_connect_sent(&self, gateway: &str) {
1782 let mut acc = self.accounts.lock().unwrap();
1783 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1784 a.stage = AccountStage::Assembling;
1785 a.fail_note = None;
1786 a.login = LoginState::Unknown;
1787 a.status_note = None;
1788 a.login_brief = None;
1789 a.chain_note = None;
1791 drop(acc);
1792 self.bump();
1793 }
1794 }
1795
1796 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1800 let mut acc = self.accounts.lock().unwrap();
1801 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1802 match result {
1803 Ok(()) => {
1804 a.stage = AccountStage::Assembled;
1805 a.fail_note = None;
1806 }
1807 Err(e) => {
1808 a.stage = AccountStage::AssembleFailed;
1809 a.fail_note = Some(e);
1810 }
1811 }
1812 drop(acc);
1813 self.bump();
1814 }
1815 }
1816
1817 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1821 let mut acc = self.accounts.lock().unwrap();
1822 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1823 a.effective_classes = classes.to_vec();
1824 drop(acc);
1825 self.bump();
1826 }
1827 }
1828
1829 pub fn note_chain(&self, gateway: &str, note: &str) {
1834 let mut acc = self.accounts.lock().unwrap();
1835 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1836 if a.chain_note.as_deref() == Some(note) {
1837 return; }
1839 a.chain_note = Some(note.to_string());
1840 drop(acc);
1841 self.bump();
1842 }
1843 }
1844
1845 #[must_use]
1849 pub fn status_thresholds(&self) -> (u32, i64) {
1850 self.status_thresholds
1851 .lock()
1852 .unwrap()
1853 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1854 }
1855
1856 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1858 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1859 self.bump();
1860 }
1861
1862 pub fn list_poll_due(&self) -> bool {
1867 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1868 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1869 }
1870
1871 pub fn request_list_refresh(&self) {
1873 self.list_due.store(true, Ordering::Release);
1874 }
1875
1876 pub fn request_trade_reseed(&self) {
1878 self.seed_due.store(true, Ordering::Release);
1879 }
1880
1881 pub fn take_trade_reseed(&self) -> bool {
1883 self.seed_due.swap(false, Ordering::AcqRel)
1884 }
1885
1886 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1891 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1895 for gw in present {
1896 if !uniq.iter().any(|u| u == gw) {
1897 uniq.push(gw.clone());
1898 }
1899 }
1900 let n = uniq.len();
1901 if n == 0 {
1902 return Vec::new();
1903 }
1904 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1905 let take = STATUS_ASKS_PER_TICK.min(n);
1906 let mut out = Vec::with_capacity(take);
1907 for i in 0..take {
1908 out.push(uniq[(start + i) % n].clone());
1909 }
1910 out
1911 }
1912
1913 pub fn request_status_seed(&self, gateway: &str) {
1915 let mut q = self.status_seed.lock().unwrap();
1916 if !q.iter().any(|g| g == gateway) {
1917 q.push(gateway.to_string());
1918 }
1919 }
1920
1921 pub fn take_status_seeds(&self) -> Vec<String> {
1923 let mut q = self.status_seed.lock().unwrap();
1924 std::mem::take(&mut *q)
1925 }
1926
1927 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
1937 let gw = ev.gateway.as_str();
1938 let text = ev.describe();
1939 self.log_line(format!("[INFO][{gw}] {text}"));
1940 let note = text
1941 .strip_prefix(CHAIN_STATE_PREFIX)
1942 .map(|s| s.trim().to_string())
1943 .unwrap_or_else(|| text.clone());
1944 self.note_chain(gw, ¬e);
1945 self.note_gateway_status(gw, &ev.status);
1946 let alarm_kind = matches!(
1947 ev.change,
1948 ek_ctp::GatewayChange::TdDisconnected
1949 | ek_ctp::GatewayChange::TdAuthFailed
1950 | ek_ctp::GatewayChange::TdLoginFailed
1951 | ek_ctp::GatewayChange::ContractQueryStopped
1952 | ek_ctp::GatewayChange::QueryRejected
1953 | ek_ctp::GatewayChange::MdDisconnected
1954 | ek_ctp::GatewayChange::MdLoginFailed
1955 | ek_ctp::GatewayChange::SubscribeRejected
1956 );
1957 if alarm_kind {
1958 let session = if ev.change.kind().starts_with("md-") {
1960 ev.status.md_disconnects
1961 } else {
1962 ev.status.td_disconnects
1963 };
1964 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
1965 if self.alert_once(key) {
1966 self.push_alert(
1967 crate::toast::AlertLevel::Warning,
1968 format!("账户 {gw}:{}", note.trim()),
1969 );
1970 }
1971 }
1972 }
1973
1974 pub fn request_pipeline_refresh(&self) {
1976 self.pipeline_due.store(true, Ordering::Release);
1977 }
1978 pub fn take_pipeline_due(&self) -> bool {
1980 self.pipeline_due.swap(false, Ordering::AcqRel)
1981 }
1982
1983 pub fn request_cfg_refresh(&self) {
1985 self.cfg_due.store(true, Ordering::Release);
1986 }
1987
1988 pub fn take_cfg_due(&self) -> bool {
1990 self.cfg_due.swap(false, Ordering::AcqRel)
1991 }
1992
1993 pub fn theme(&self) -> crate::tokens::Theme {
1995 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
1996 }
1997
1998 pub fn set_theme(&self, theme: crate::tokens::Theme) {
2001 self.theme.store(theme.as_u8(), Ordering::Release);
2002 self.bump();
2003 }
2004
2005 pub fn toggle_theme(&self) -> crate::tokens::Theme {
2007 let next = self.theme().flipped();
2008 self.theme.store(next.as_u8(), Ordering::Release);
2009 self.theme_persist_due.store(true, Ordering::Release);
2010 self.bump();
2011 next
2012 }
2013
2014 pub fn take_theme_persist_due(&self) -> bool {
2016 self.theme_persist_due.swap(false, Ordering::AcqRel)
2017 }
2018
2019 pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
2023 {
2024 let mut acc = self.accounts.lock().unwrap();
2025 match acc.iter_mut().find(|a| a.gateway == gateway) {
2026 Some(a) => {
2027 if a.auto_connect != on {
2028 a.auto_connect = on;
2029 self.auto_connect_persist_due.store(true, Ordering::Release);
2030 self.bump();
2031 }
2032 }
2033 None => return Err(format!("账户 {gateway} 不在清册")),
2034 }
2035 }
2036 Ok(())
2037 }
2038
2039 pub fn take_auto_connect_persist_due(&self) -> bool {
2041 self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
2042 }
2043
2044 pub fn take_alias_persist_due(&self) -> bool {
2046 self.alias_persist_due.swap(false, Ordering::AcqRel)
2047 }
2048
2049 pub fn take_archive_purges(&self) -> Vec<String> {
2051 let mut q = self.archive_purge.lock().unwrap();
2052 std::mem::take(&mut *q)
2053 }
2054
2055 pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
2070 let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
2072 let mut aliases: std::collections::BTreeMap<String, String> = Default::default();
2073 match store.load(crate::plugin::THEME_NS) {
2074 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
2075 for (k, _f) in map.iter() {
2076 if let Some(gw) = k.strip_prefix("auto_connect.") {
2077 flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
2078 }
2079 if let Some(gw) = k.strip_prefix("alias.")
2080 && let Some(v) = map.get_plain(k)
2081 {
2082 let v = v.trim();
2083 if !v.is_empty() {
2084 aliases.insert(gw.to_string(), v.to_string());
2085 }
2086 }
2087 }
2088 }
2089 Ok(_) => {} Err(_) => {} }
2092 let mut restored: Vec<String> = Vec::new();
2093 let ns_list = match store.list_namespaces() {
2094 Ok(v) => v,
2095 Err(_) => {
2096 let msg = "[ui] 档案清点失败:存储根不可读(具体见运行日志)".to_string();
2097 self.log_line(msg.clone());
2098 return restored;
2099 }
2100 };
2101 for ns in ns_list {
2102 let Some(gw) = ns.strip_prefix("ctp.") else {
2103 continue;
2104 };
2105 if self.account(gw).is_some() {
2106 self.log_line(format!(
2107 "[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
2108 ));
2109 continue;
2110 }
2111 let map = match store.load(&ns) {
2112 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
2113 Ok(_) => {
2114 self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
2115 continue;
2116 }
2117 Err(e) => {
2118 self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
2119 continue;
2120 }
2121 };
2122 let password_ref = map
2124 .get_secret_key("td_password")
2125 .is_some_and(|k| !k.is_empty());
2126 let mut form = CtpForm {
2127 broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
2128 td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
2129 td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
2130 td_password: String::new(),
2131 td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
2132 td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
2133 md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
2134 md_user: String::new(),
2135 md_password: String::new(),
2136 symbols: String::new(),
2137 accept_futures: false,
2138 accept_options: false,
2139 };
2140 for c in map
2143 .get_plain("product_classes")
2144 .unwrap_or_default()
2145 .split(',')
2146 {
2147 if c.trim() == "1" {
2148 form.accept_futures = true;
2149 }
2150 }
2151 let Ok(mut row) = AccountRow::derive(&form) else {
2152 self.log_line(format!(
2153 "[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
2154 ));
2155 continue;
2156 };
2157 row.gateway = gw.to_string();
2160 row.restored = true;
2161 row.auto_connect = flags.get(gw).copied().unwrap_or(false);
2162 row.password_ref = password_ref;
2163 if let Some(a) = aliases.get(gw) {
2166 let taken = self
2167 .accounts
2168 .lock()
2169 .unwrap()
2170 .iter()
2171 .any(|x| x.alias.trim().eq_ignore_ascii_case(a));
2172 if taken {
2173 self.log_line(format!(
2174 "[ui] 档案别名 {a:?} 已被在册行占用,{gw} 不带标签复原(账户面板改别名即可)"
2175 ));
2176 } else {
2177 row.alias = a.clone();
2178 }
2179 }
2180 restored.push(row.gateway.clone());
2181 self.add_account(row).ok();
2182 }
2183 if !restored.is_empty() {
2184 self.log_line(format!(
2185 "[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
2186 restored.len()
2187 ));
2188 if self.md_source.lock().unwrap().is_empty() {
2191 let first = self
2192 .accounts
2193 .lock()
2194 .unwrap()
2195 .iter()
2196 .find(|a| a.auto_connect)
2197 .map(|a| a.gateway.clone());
2198 if let Some(gw) = first
2199 && self.set_md_source(&gw).is_ok()
2200 {
2201 self.log_line(format!(
2202 "[账户] 行情源(默认=第一个勾选自动连接的)= {gw}"
2203 ));
2204 }
2205 }
2206 self.bump();
2207 }
2208 restored
2209 }
2210
2211 pub fn restore_saved_accounts(&self) -> Vec<String> {
2213 match ek_cfg::Store::eco() {
2214 Ok(store) => self.restore_saved_accounts_from(&store),
2215 Err(_) => {
2216 self.log_line(
2217 "[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
2218 );
2219 Vec::new()
2220 }
2221 }
2222 }
2223
2224 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
2226 *self.cfg_briefs.lock().unwrap() = Some(b);
2227 self.bump();
2228 }
2229 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
2231 self.cfg_briefs.lock().unwrap().clone()
2232 }
2233
2234 pub fn cfg_delete_armed(&self) -> Option<String> {
2236 self.cfg_delete_arm.lock().unwrap().clone()
2237 }
2238 pub fn arm_cfg_delete(&self, ns: Option<String>) {
2239 *self.cfg_delete_arm.lock().unwrap() = ns;
2240 self.bump();
2241 }
2242
2243 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
2245 *self.pipeline_lines.lock().unwrap() = lines;
2246 self.bump();
2247 }
2248
2249 pub fn pipeline_lines(&self) -> Vec<String> {
2251 self.pipeline_lines.lock().unwrap().clone()
2252 }
2253
2254 pub fn note_bad_payload(&self, line: &str) {
2260 let mut book = self.bad_payload.lock().unwrap();
2261 let n = book.entry(line.to_string()).or_insert(0);
2262 *n += 1;
2263 if *n == 1 {
2264 self.log_line(format!(
2265 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
2266 ));
2267 }
2268 }
2269
2270 #[must_use]
2272 pub fn bad_payload_count(&self, line: &str) -> u64 {
2273 self.bad_payload
2274 .lock()
2275 .unwrap()
2276 .get(line)
2277 .copied()
2278 .unwrap_or(0)
2279 }
2280
2281 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
2295 for r in rows {
2296 let (warn, crit) = pending_thresholds(r.soft_cap);
2297 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
2298 self.push_alert(
2299 crate::toast::AlertLevel::Warning,
2300 format!(
2301 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
2302 r.name, r.pending, r.peak, r.soft_cap
2303 ),
2304 );
2305 }
2306 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
2307 self.push_alert(
2308 crate::toast::AlertLevel::Error,
2309 format!(
2310 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
2311 r.name, r.pending, r.soft_cap
2312 ),
2313 );
2314 }
2315 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
2316 self.push_alert(
2317 crate::toast::AlertLevel::Error,
2318 format!(
2319 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
2320 r.name, r.soft_cap, r.spilled, r.peak
2321 ),
2322 );
2323 }
2324 }
2325 }
2326
2327 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
2335 let (limit, stale) = self.status_thresholds();
2336 let now = ek_ctp::map::now_epoch_ms();
2337 let mut alarm: Option<(LoginState, String)> = None;
2340 let mut found = false;
2341 {
2342 let mut acc = self.accounts.lock().unwrap();
2343 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2344 found = true;
2345 a.apply_status(st, now, limit, stale);
2346 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
2347 alarm = Some((
2348 a.login,
2349 a.status_note
2350 .clone()
2351 .unwrap_or_else(|| a.login.label().to_string()),
2352 ));
2353 }
2354 }
2355 }
2356 if found {
2357 self.bump();
2358 }
2359 if let Some((login, note)) = alarm
2360 && self.alert_once(format!("{gateway}|{}", login.label()))
2361 {
2362 self.push_alert(
2363 crate::toast::AlertLevel::Error,
2364 format!("账户 {gateway}:{note}"),
2365 );
2366 }
2367 }
2368
2369 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2373 let mut acc = self.accounts.lock().unwrap();
2374 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2375 a.login = LoginState::Unknown;
2376 a.status_note = Some(format!("状态口问不到:{why}"));
2377 a.login_brief = None;
2378 drop(acc);
2379 self.bump();
2380 }
2381 }
2382
2383 pub fn mark_account_stopped(&self, gateway: &str) {
2385 let mut acc = self.accounts.lock().unwrap();
2386 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2387 a.login = LoginState::Stopped;
2388 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2389 a.login_brief = None;
2390 drop(acc);
2391 self.bump();
2392 }
2393 }
2394
2395 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2398 let source = self.md_source();
2399 let mut f = form;
2400 if source.as_str() != gateway {
2401 f.md_addr.clear();
2402 f.md_user.clear();
2403 f.md_password.clear();
2404 }
2405 self.update_account_form(gateway, f.clone());
2406 f
2407 }
2408
2409 #[must_use]
2413 pub fn draft_from_existing(&self) -> CtpForm {
2414 let acc = self.accounts.lock().unwrap();
2415 let mut draft = CtpForm::default();
2416 if let Some(a) = acc.last() {
2417 draft.broker_id = a.form.broker_id.clone();
2418 draft.td_auth_code = a.form.td_auth_code.clone();
2419 draft.td_app_id = a.form.td_app_id.clone();
2420 draft.accept_futures = a.form.accept_futures;
2421 draft.accept_options = a.form.accept_options;
2422 }
2423 draft
2424 }
2425
2426 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2428 let mut acc = self.accounts.lock().unwrap();
2429 match acc.iter_mut().find(|a| a.gateway == gateway) {
2430 Some(a) => {
2431 a.form = form;
2432 drop(acc);
2433 self.bump();
2434 true
2435 }
2436 None => false,
2437 }
2438 }
2439
2440 pub fn remove_account(&self, gateway: &str) -> bool {
2442 let mut acc = self.accounts.lock().unwrap();
2443 let before = acc.len();
2444 acc.retain(|a| a.gateway != gateway);
2445 let removed = acc.len() != before;
2446 if removed {
2447 let mut md = self.md_source.lock().unwrap();
2449 if md.as_str() == gateway {
2450 md.clear();
2451 }
2452 drop(acc);
2453 self.alias_persist_due.store(true, Ordering::Release);
2455 self.archive_purge.lock().unwrap().push(gateway.to_string());
2457 self.bump();
2458 }
2459 removed
2460 }
2461
2462 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2471 let mut newly = Vec::new();
2472 {
2473 let mut acc = self.accounts.lock().unwrap();
2474 for a in acc.iter_mut() {
2475 let up = rows.iter().any(|r| r.name == a.gateway);
2476 if up && !a.connected {
2477 newly.push(a.gateway.clone());
2478 }
2479 a.connected = up;
2480 if up {
2481 a.stage = AccountStage::Assembled;
2482 a.fail_note = None;
2483 } else {
2484 a.chain_note = None;
2487 if a.stage == AccountStage::Assembled {
2488 a.stage = AccountStage::NotAssembled;
2489 a.login = LoginState::Unknown;
2490 a.status_note = None;
2491 a.login_brief = None;
2492 }
2493 }
2494 }
2495 }
2496 if !newly.is_empty() {
2497 self.bump();
2498 }
2499 newly
2500 }
2501
2502 pub fn md_source(&self) -> String {
2504 self.md_source.lock().unwrap().clone()
2505 }
2506
2507 pub fn accounts_locked(&self) -> bool {
2509 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2510 }
2511
2512 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2514 {
2515 let acc = self.accounts.lock().unwrap();
2516 if acc.iter().any(|a| a.connected) {
2517 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2518 }
2519 if !acc.iter().any(|a| a.gateway == gateway) {
2520 return Err(format!("账户 {gateway} 不在清册"));
2521 }
2522 }
2523 let mut acc = self.accounts.lock().unwrap();
2524 for a in acc.iter_mut() {
2525 a.md_source = a.gateway == gateway;
2526 }
2527 drop(acc);
2528 let mut md = self.md_source.lock().unwrap();
2529 *md = gateway.to_string();
2530 self.bump();
2531 Ok(())
2532 }
2533
2534 pub fn subscriptions(&self) -> Vec<String> {
2536 self.subscriptions.lock().unwrap().clone()
2537 }
2538
2539 pub fn note_subscriptions(&self, symbols: &[String]) {
2541 {
2542 let mut subs = self.subscriptions.lock().unwrap();
2543 for s in symbols {
2544 let s = s.trim();
2545 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2546 subs.push(s.to_string());
2547 }
2548 }
2549 }
2550 self.bump();
2551 }
2552
2553 pub fn default_gateway(&self) -> Option<String> {
2560 let chosen = {
2561 let f = self.order_form.lock().unwrap();
2562 f.gateway.trim().to_string()
2563 };
2564 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2565 return Some(chosen);
2566 }
2567 let acc = self.accounts.lock().unwrap();
2568 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2569 }
2570
2571 pub fn kline_symbols(&self) -> Vec<String> {
2573 self.kline_windows.lock().unwrap().clone()
2574 }
2575
2576 pub fn open_kline(&self, symbol: impl Into<String>) {
2578 let s = symbol.into();
2579 let mut v = self.kline_windows.lock().unwrap();
2580 if !v.contains(&s) {
2581 v.push(s.clone());
2582 }
2583 drop(v);
2584 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2585 let fq = if s.contains('.') {
2588 s.clone()
2589 } else {
2590 format!("{}.{}", self.selected_exchange(), s)
2591 };
2592 const KLINE_INTERVAL_SECS: i64 = 60;
2593 {
2594 let mut fetched = self.kline_fetched.lock().unwrap();
2595 if fetched.contains(&(fq.clone(), KLINE_INTERVAL_SECS)) {
2596 return;
2597 }
2598 fetched.insert((fq.clone(), KLINE_INTERVAL_SECS));
2599 }
2600 self.kline_fetches.lock().unwrap().push(KlineFetch {
2601 symbol: fq,
2602 interval_secs: KLINE_INTERVAL_SECS,
2603 });
2604 self.log_line(format!("[kline] 入队历史补数 {s}(1min×300,tq 桥)"));
2605 }
2606
2607 pub fn take_kline_fetches(&self) -> Vec<KlineFetch> {
2609 std::mem::take(&mut self.kline_fetches.lock().unwrap())
2610 }
2611
2612 pub fn kline_fetch_pending(&self) -> bool {
2614 !self.kline_fetches.lock().unwrap().is_empty()
2615 }
2616
2617 pub fn close_kline(&self, symbol: &str) {
2619 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2620 self.bump();
2621 }
2622
2623 pub fn contracts_window_open(&self) -> bool {
2627 self.contracts_window.load(Ordering::Acquire)
2628 }
2629
2630 pub fn set_contracts_window_open(&self, v: bool) {
2632 self.contracts_window.store(v, Ordering::Release);
2633 self.bump();
2634 }
2635
2636 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2643 self.contract_store.lock().unwrap().put(c);
2644 if !c.name.is_empty() {
2645 self.backfill_names(&c.symbol, &c.name);
2646 }
2647 self.bump();
2648 }
2649
2650 fn backfill_names(&self, symbol: &str, name: &str) {
2655 let mut touched = false;
2656 {
2657 let mut rows = self.trade_orders.lock().unwrap();
2658 for (_, (_, o)) in rows.iter_mut() {
2659 if o.symbol == symbol && o.name.is_empty() {
2660 o.name = name.to_string();
2661 touched = true;
2662 }
2663 }
2664 }
2665 {
2666 let mut rows = self.trade_trades.lock().unwrap();
2667 for (_, t) in rows.iter_mut() {
2668 if t.symbol == symbol && t.name.is_empty() {
2669 t.name = name.to_string();
2670 touched = true;
2671 }
2672 }
2673 }
2674 {
2675 let mut rows = self.trade_positions.lock().unwrap();
2676 for (_, (_, p)) in rows.iter_mut() {
2677 if p.symbol == symbol && p.name.is_empty() {
2678 p.name = name.to_string();
2679 touched = true;
2680 }
2681 }
2682 }
2683 if touched {
2684 self.trade_dirty.store(true, Ordering::Release);
2686 }
2687 }
2688
2689 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2691 self.contract_store.lock().unwrap().query(pattern)
2692 }
2693
2694 pub fn contract_count(&self) -> usize {
2696 self.contract_store.lock().unwrap().len()
2697 }
2698
2699 fn contract_name_of(&self, symbol: &str) -> Option<String> {
2701 self.contract_store
2702 .lock()
2703 .unwrap()
2704 .query(symbol)
2705 .into_iter()
2706 .find(|c| c.symbol == symbol && !c.name.is_empty())
2707 .map(|c| c.name)
2708 }
2709
2710 pub fn contract_query(&self) -> String {
2712 self.contract_query.lock().unwrap().clone()
2713 }
2714
2715 pub fn set_contract_query(&self, q: impl Into<String>) {
2717 *self.contract_query.lock().unwrap() = q.into();
2718 self.bump();
2719 }
2720
2721 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2723 let mut v = self.contract_selected.lock().unwrap();
2724 match v.iter().position(|s| s == en_symbol) {
2725 Some(i) => {
2726 v.remove(i);
2727 }
2728 None => v.push(en_symbol.to_string()),
2729 }
2730 self.bump();
2731 }
2732
2733 pub fn selected_contracts(&self) -> Vec<String> {
2735 self.contract_selected.lock().unwrap().clone()
2736 }
2737
2738 #[must_use]
2744 pub fn cancel_reject_note(msg: &str) -> String {
2745 let head = msg.split(' ').next().unwrap_or(msg);
2746 let note = if head.starts_with("unknown:order") {
2747 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
2748 } else if head.starts_with("mismatch:gateway") {
2749 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
2750 } else if head.starts_with("terminal:status") {
2751 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
2752 } else if head.starts_with("limit:cancel-guard") {
2753 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
2754 } else {
2755 ""
2756 };
2757 if note.is_empty() {
2758 format!("撤单被拒:{msg}")
2760 } else {
2761 format!("撤单被拒:{note}(机器码:{msg})")
2762 }
2763 }
2764
2765 pub fn log_line(&self, line: impl Into<String>) {
2767 self.logs.lock().unwrap().push(line);
2768 self.bump();
2769 }
2770
2771 pub fn clear_logs(&self) {
2773 self.logs.lock().unwrap().clear();
2774 self.bump();
2775 }
2776
2777 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2779 let r = self.logs.lock().unwrap();
2780 (r.snapshot(), r.spilled())
2781 }
2782
2783 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2785 let r = self.logs.lock().unwrap();
2786 (r.entries(), r.spilled())
2787 }
2788
2789 pub fn logs_len(&self) -> usize {
2791 self.logs.lock().unwrap().len()
2792 }
2793
2794 pub fn set_snapshot(&self, snap: BusSnap) {
2796 *self.snap.lock().unwrap() = Arc::new(snap);
2797 self.bump();
2798 }
2799
2800 pub fn snapshot(&self) -> Arc<BusSnap> {
2802 Arc::clone(&self.snap.lock().unwrap())
2803 }
2804
2805 pub fn rev(&self) -> u64 {
2807 self.rev.load(Ordering::Acquire)
2808 }
2809
2810 pub fn bump(&self) {
2812 self.rev.fetch_add(1, Ordering::AcqRel);
2813 }
2814
2815 pub fn request_stop(&self) {
2817 self.stop.store(true, Ordering::Release);
2818 }
2819
2820 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2823 self.pending_alerts
2824 .lock()
2825 .unwrap_or_else(std::sync::PoisonError::into_inner)
2826 .push((level, text.into()));
2827 }
2828
2829 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2831 let mut q = self
2832 .pending_alerts
2833 .lock()
2834 .unwrap_or_else(std::sync::PoisonError::into_inner);
2835 std::mem::take(&mut *q)
2836 }
2837
2838 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2840 self.toasts.lock()
2841 }
2842
2843 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2845 let key = key.into();
2846 self.alerted
2847 .lock()
2848 .unwrap_or_else(std::sync::PoisonError::into_inner)
2849 .insert(key)
2850 }
2851
2852 #[must_use]
2854 pub fn pending_alert_count(&self) -> usize {
2855 self.pending_alerts
2856 .lock()
2857 .unwrap_or_else(std::sync::PoisonError::into_inner)
2858 .len()
2859 }
2860
2861 #[must_use]
2863 pub fn alerted_count(&self) -> usize {
2864 self.alerted
2865 .lock()
2866 .unwrap_or_else(std::sync::PoisonError::into_inner)
2867 .len()
2868 }
2869
2870 pub fn exit_confirm_open(&self) -> bool {
2872 self.exit_confirm_open
2873 .load(std::sync::atomic::Ordering::Relaxed)
2874 }
2875
2876 pub fn open_exit_confirm(&self) {
2877 self.exit_confirm_open
2878 .store(true, std::sync::atomic::Ordering::Relaxed);
2879 }
2880
2881 pub fn close_exit_confirm(&self) {
2882 self.exit_confirm_open
2883 .store(false, std::sync::atomic::Ordering::Relaxed);
2884 }
2885
2886 pub fn exit_confirmed(&self) -> bool {
2887 self.exit_confirmed
2888 .load(std::sync::atomic::Ordering::Relaxed)
2889 }
2890
2891 pub fn confirm_exit(&self) {
2892 self.exit_confirmed
2893 .store(true, std::sync::atomic::Ordering::Relaxed);
2894 }
2895
2896 pub fn is_stop(&self) -> bool {
2897 self.stop.load(Ordering::Acquire)
2898 }
2899
2900 #[must_use]
2902 pub fn assembly_locked(&self) -> bool {
2903 self.assembly_locked.load(Ordering::Relaxed)
2904 }
2905
2906 pub fn set_assembly_locked(&self, on: bool) {
2908 self.assembly_locked.store(on, Ordering::Relaxed);
2909 }
2910
2911 pub fn push_cmd(&self, cmd: UiCmd) {
2915 self.cmds.lock().unwrap().push_back(cmd);
2916 self.bump();
2917 }
2918
2919 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2921 self.cmds.lock().unwrap().drain(..).collect()
2922 }
2923
2924 #[must_use]
2926 pub fn cmd_pending(&self) -> bool {
2927 !self.cmds.lock().unwrap().is_empty()
2928 }
2929
2930 pub fn pending_confirm(&self) -> Option<String> {
2932 self.pending_confirm.lock().unwrap().clone()
2933 }
2934
2935 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2937 let mut g = self.pending_confirm.lock().unwrap();
2938 match g.take() {
2939 Some(p) if p == name => ConfirmStep::Fired,
2940 Some(_) => {
2941 *g = Some(name.to_string());
2942 ConfirmStep::Switched
2943 }
2944 None => {
2945 *g = Some(name.to_string());
2946 ConfirmStep::Armed
2947 }
2948 }
2949 }
2950
2951 pub fn confirm_cancel(&self) {
2953 *self.pending_confirm.lock().unwrap() = None;
2954 self.bump();
2955 }
2956
2957 pub fn market(&self) -> &MarketStore {
2961 &self.market
2962 }
2963
2964 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2973 let named = |symbol: &str, name: String| -> String {
2976 if name.is_empty() {
2977 self.contract_name_of(symbol).unwrap_or(name)
2978 } else {
2979 name
2980 }
2981 };
2982 {
2983 let mut book = self.trade_orders.lock().unwrap();
2984 book.clear();
2985 for mut o in ts.orders.into_iter().rev() {
2986 o.name = named(&o.symbol, o.name);
2987 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2988 book.insert(o.en_orderid(), (seq, o));
2989 }
2990 }
2991 {
2992 let mut book = self.trade_trades.lock().unwrap();
2993 book.clear();
2994 let mut ids = self.trade_ids.lock().unwrap();
2995 ids.clear();
2996 for mut t in ts.trades.into_iter().rev() {
2997 t.name = named(&t.symbol, t.name);
2998 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2999 ids.insert(t.en_tradeid());
3000 book.push_front((seq, t));
3001 }
3002 }
3003 {
3004 let mut book = self.trade_positions.lock().unwrap();
3005 book.clear();
3006 for mut p in ts.positions.into_iter().rev() {
3007 p.name = named(&p.symbol, p.name);
3008 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3009 book.insert(p.en_positionid(), (seq, p));
3010 }
3011 }
3012 {
3013 let mut book = self.trade_funds.lock().unwrap();
3014 book.clear();
3015 for a in ts.accounts.into_iter().rev() {
3016 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3017 book.insert(a.en_accountid(), (seq, a));
3018 }
3019 }
3020 self.trade_dirty.store(true, Ordering::Release);
3021 self.bump();
3022 }
3023
3024 pub fn note_order(&self, o: &ek_data::order::OrderData) {
3032 let key = o.en_orderid();
3033 {
3034 let mut book = self.trade_orders.lock().unwrap();
3035 match book.get_mut(&key) {
3036 Some((_, cur)) => {
3037 cur.name.clone_from(&o.name);
3038 cur.traded = o.traded;
3039 cur.traded_price = o.traded_price;
3040 cur.status = o.status;
3041 cur.status_msg.clone_from(&o.status_msg);
3042 cur.datetime = o.datetime;
3043 }
3044 None => {
3045 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3046 book.insert(key, (seq, o.clone()));
3047 }
3048 }
3049 }
3050 self.trade_dirty.store(true, Ordering::Release);
3051 self.bump();
3052 }
3053
3054 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
3062 let id = t.en_tradeid();
3063 if self.trade_ids.lock().unwrap().contains(&id) {
3064 return;
3065 }
3066 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3067 let dropped = {
3068 let mut book = self.trade_trades.lock().unwrap();
3069 book.push_front((seq, t.clone()));
3070 self.trade_ids.lock().unwrap().insert(id);
3071 let over = book.len().saturating_sub(TRADE_CAP);
3072 for _ in 0..over {
3073 if let Some((_, old)) = book.pop_back() {
3075 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
3076 }
3077 }
3078 over
3079 };
3080 if dropped > 0 {
3081 let total = self
3082 .trade_dropped
3083 .fetch_add(dropped as u64, Ordering::AcqRel)
3084 + dropped as u64;
3085 if total == dropped as u64 {
3087 self.push_alert(
3088 crate::toast::AlertLevel::Warning,
3089 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
3090 );
3091 }
3092 }
3093 self.trade_dirty.store(true, Ordering::Release);
3094 self.bump();
3095 }
3096
3097 pub fn note_position(&self, p: &ek_data::position::PositionData) {
3102 let key = p.en_positionid();
3103 {
3104 let mut book = self.trade_positions.lock().unwrap();
3105 match book.get_mut(&key) {
3106 Some((_, cur)) => {
3107 cur.volume = p.volume;
3108 cur.yd_volume = p.yd_volume;
3109 cur.can_use_volume = p.can_use_volume;
3110 cur.frozen = p.frozen;
3111 cur.on_road_volume = p.on_road_volume;
3112 cur.market_value = p.market_value;
3113 cur.avg_price = p.avg_price;
3114 cur.price = p.price;
3115 cur.last_price = p.last_price;
3116 cur.pnl = p.pnl;
3117 cur.profit_rate = p.profit_rate;
3118 cur.name.clone_from(&p.name);
3119 }
3120 None => {
3121 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3122 book.insert(key, (seq, p.clone()));
3123 }
3124 }
3125 }
3126 self.trade_dirty.store(true, Ordering::Release);
3127 self.bump();
3128 }
3129
3130 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
3132 let key = a.en_accountid();
3133 {
3134 let mut book = self.trade_funds.lock().unwrap();
3135 match book.get_mut(&key) {
3136 Some((_, cur)) => {
3137 cur.balance = a.balance;
3138 cur.frozen = a.frozen;
3139 }
3140 None => {
3141 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3142 book.insert(key, (seq, a.clone()));
3143 }
3144 }
3145 }
3146 self.trade_dirty.store(true, Ordering::Release);
3147 self.bump();
3148 }
3149
3150 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
3152 let mut removed = 0usize;
3153 {
3154 let mut book = self.trade_orders.lock().unwrap();
3155 let before = book.len();
3156 book.retain(|_, (_, o)| o.gateway_name != gateway);
3157 removed += before - book.len();
3158 }
3159 {
3160 let mut book = self.trade_trades.lock().unwrap();
3161 let before = book.len();
3162 book.retain(|(_, t)| t.gateway_name != gateway);
3163 removed += before - book.len();
3164 let prefix = format!("{gateway}.");
3167 self.trade_ids
3168 .lock()
3169 .unwrap()
3170 .retain(|id| !id.starts_with(&prefix));
3171 }
3172 {
3173 let mut book = self.trade_positions.lock().unwrap();
3174 let before = book.len();
3175 book.retain(|_, (_, p)| p.gateway_name != gateway);
3176 removed += before - book.len();
3177 }
3178 {
3179 let mut book = self.trade_funds.lock().unwrap();
3180 let before = book.len();
3181 book.retain(|_, (_, a)| a.gateway_name != gateway);
3182 removed += before - book.len();
3183 }
3184 if removed > 0 {
3185 self.trade_dirty.store(true, Ordering::Release);
3186 self.bump();
3187 }
3188 removed
3189 }
3190
3191 #[must_use]
3193 pub fn trade_dropped(&self) -> u64 {
3194 self.trade_dropped.load(Ordering::Acquire)
3195 }
3196
3197 pub fn trade_snap(&self) -> Arc<TradeSnap> {
3202 let mut cache = self.trade_snap.lock().unwrap();
3203 if self.trade_dirty.swap(false, Ordering::AcqRel) {
3204 *cache = Arc::new(self.build_trade_snap());
3205 }
3206 Arc::clone(&cache)
3207 }
3208
3209 fn build_trade_snap(&self) -> TradeSnap {
3212 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
3213 where
3214 T: Clone,
3215 {
3216 let mut v: Vec<(u64, &String, &T)> =
3217 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
3218 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
3219 v.into_iter().map(|(_, _, v)| v.clone()).collect()
3220 }
3221 TradeSnap {
3222 orders: by_seq(&self.trade_orders.lock().unwrap()),
3223 trades: self
3224 .trade_trades
3225 .lock()
3226 .unwrap()
3227 .iter()
3228 .map(|(_, t)| t.clone())
3229 .collect(),
3230 positions: by_seq(&self.trade_positions.lock().unwrap()),
3231 accounts: by_seq(&self.trade_funds.lock().unwrap()),
3232 }
3233 }
3234
3235 pub fn ctp_form(&self) -> CtpForm {
3239 self.ctp_form.lock().unwrap().clone()
3240 }
3241
3242 pub fn set_ctp_form(&self, f: CtpForm) {
3244 *self.ctp_form.lock().unwrap() = f;
3245 self.bump();
3246 }
3247
3248 pub fn row_selected(&self, table: &str, key: &str) -> bool {
3252 self.row_sel
3253 .lock()
3254 .unwrap()
3255 .get(table)
3256 .is_some_and(|k| k == key)
3257 }
3258
3259 pub fn select_row(&self, table: &str, key: &str) {
3261 self.row_sel
3262 .lock()
3263 .unwrap()
3264 .insert(table.to_string(), key.to_string());
3265 }
3266}