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 = 8;
231
232pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
237
238pub const ACCOUNT_STALE_SECS: i64 = 30;
244
245pub use ek_ctp::driver::CHAIN_STATE_PREFIX;
251
252pub const LIST_POLL_TICKS: u32 = 4;
258
259pub const STATUS_ASKS_PER_TICK: usize = 2;
264
265pub const TRADE_CAP: usize = 5000;
270
271pub const PENDING_WARN_FLOOR: u64 = 4_096;
273pub const PENDING_CRIT_FLOOR: u64 = 32_768;
275
276#[must_use]
284pub fn pending_thresholds(soft_cap: u64) -> (u64, u64) {
285 let warn = (soft_cap / 16).max(PENDING_WARN_FLOOR);
286 let crit = (soft_cap / 4).max(PENDING_CRIT_FLOOR);
287 (warn.min(soft_cap.max(1)), crit.min(soft_cap.max(1)))
288}
289
290#[derive(Debug, Clone, Copy, PartialEq, Eq)]
292pub enum ConfirmStep {
293 Armed,
294 Fired,
295 Switched,
296}
297
298#[must_use]
301pub fn is_quarantined(state: &str) -> bool {
302 state.contains("Quarantin")
303}
304
305#[derive(Debug, Clone)]
309pub struct KlineFetch {
310 pub symbol: String,
311 pub interval_secs: i64,
312}
313
314#[derive(Debug, Default)]
316pub struct MarketStore {
317 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
318 frames: AtomicU64,
320 updates: AtomicU64,
322 reloads: AtomicU64,
324}
325
326#[derive(Debug, Clone, PartialEq)]
328pub struct CellView {
329 pub symbol: String,
330 pub interval: i64,
331 pub bars: Vec<BarData>,
332}
333
334impl MarketStore {
335 pub fn on_frame(&self, frame: &BarsFrame) {
337 let mut cells = self.cells.lock().unwrap();
338 for b in &frame.bars {
339 let key = (b.symbol.clone(), frame.interval);
340 let cell = cells.entry(key).or_default();
341 match cell.last() {
342 Some(last) if b.datetime < last.datetime => continue,
343 Some(last) if b.datetime == last.datetime => {
344 let slot = cell.last_mut().expect("len>1");
345 *slot = b.clone();
346 }
347 _ => {
348 cell.push(b.clone());
349 if cell.len() > KLINE_CELL_CAP {
350 cell.remove(0);
351 }
352 }
353 }
354 }
355 self.frames.fetch_add(1, Ordering::AcqRel);
356 }
357
358 pub fn on_update(&self, b: &BarData) {
360 let mut cells = self.cells.lock().unwrap();
361 let iv = b
362 .interval
363 .unwrap_or(ek_data::constant::Interval::MINUTE)
364 .seconds();
365 let key = (b.symbol.clone(), iv);
366 let cell = cells.entry(key).or_default();
367 match cell.last() {
368 Some(last) if b.datetime == last.datetime => {
369 let slot = cell.last_mut().expect("len>1");
370 *slot = b.clone();
371 }
372 Some(last) if b.datetime < last.datetime => {}
373 _ => {
374 cell.push(b.clone());
375 if cell.len() > KLINE_CELL_CAP {
376 cell.remove(0);
377 }
378 }
379 }
380 self.updates.fetch_add(1, Ordering::AcqRel);
381 }
382
383 pub fn note_history(&self, symbol: &str, iv_secs: i64, bars: &[BarData]) {
389 let mut cells = self.cells.lock().unwrap();
390 let cell = cells.entry((symbol.to_string(), iv_secs)).or_default();
391 cell.extend(bars.iter().cloned());
392 cell.sort_unstable_by_key(|b| b.datetime);
393 cell.dedup_by(|a, b| a.datetime == b.datetime);
394 if cell.len() > KLINE_CELL_CAP {
395 let excess = cell.len() - KLINE_CELL_CAP;
396 cell.drain(0..excess);
397 }
398 self.frames.fetch_add(1, Ordering::AcqRel);
399 }
400
401 pub fn note_reload(&self) {
403 self.reloads.fetch_add(1, Ordering::AcqRel);
404 }
405
406 pub fn cells(&self) -> Vec<CellView> {
408 self.cells
409 .lock()
410 .unwrap()
411 .iter()
412 .map(|((sym, iv), bars)| CellView {
413 symbol: sym.clone(),
414 interval: *iv,
415 bars: bars.clone(),
416 })
417 .collect()
418 }
419
420 pub fn find_cell(&self, sym: &str) -> Option<CellView> {
425 self.cells()
426 .into_iter()
427 .find(|c| c.symbol == sym || c.symbol.ends_with(&format!(".{sym}")))
428 }
429
430 pub fn counts(&self) -> (u64, u64, u64) {
432 (
433 self.frames.load(Ordering::Acquire),
434 self.updates.load(Ordering::Acquire),
435 self.reloads.load(Ordering::Acquire),
436 )
437 }
438}
439
440#[derive(Debug, Default, Clone, PartialEq)]
443pub struct ReloadForm {
444 pub interval: String,
445 pub symbol: String,
446 pub csv: String,
447}
448
449pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
456 let g = raw.trim();
457 if !g.is_empty() {
458 if g.chars().any(|c| c.is_whitespace()) {
459 return Err(format!("网关解析: 含空白:{raw:?}"));
460 }
461 return Ok(g.to_string());
462 }
463 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
464 Some(f) => Ok(f.to_string()),
465 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
466 }
467}
468
469#[derive(Debug, Clone, PartialEq)]
472pub struct CtpForm {
473 pub broker_id: String,
474 pub td_addr: String,
475 pub td_user: String,
476 pub td_password: String,
477 pub td_auth_code: String,
478 pub td_app_id: String,
479 pub md_addr: String,
480 pub md_user: String,
481 pub md_password: String,
482 pub symbols: String,
484 pub accept_futures: bool,
487 pub accept_options: bool,
488}
489
490impl Default for CtpForm {
493 fn default() -> Self {
494 Self {
495 broker_id: String::new(),
496 td_addr: String::new(),
497 td_user: String::new(),
498 td_password: String::new(),
499 td_auth_code: String::new(),
500 td_app_id: String::new(),
501 md_addr: String::new(),
502 md_user: String::new(),
503 md_password: String::new(),
504 symbols: String::new(),
505 accept_futures: true,
506 accept_options: false,
507 }
508 }
509}
510
511impl CtpForm {
512 #[must_use]
516 pub fn product_classes(&self) -> Vec<String> {
517 let mut codes = Vec::new();
518 if self.accept_futures {
519 codes.push("1".to_string());
520 }
521 if self.accept_options {
522 codes.push("2".to_string());
523 }
524 codes
525 }
526
527 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
529 ek_ctp::plugin::ConfigureReq {
530 broker_id: self.broker_id.trim().to_string(),
531 td_addr: self.td_addr.trim().to_string(),
532 td_user: self.td_user.trim().to_string(),
533 td_password: self.td_password.clone(),
534 td_auth_code: self.td_auth_code.trim().to_string(),
535 td_app_id: self.td_app_id.trim().to_string(),
536 md_addr: self.md_addr.trim().to_string(),
537 md_user: self.md_user.trim().to_string(),
538 md_password: self.md_password.clone(),
539 product_classes: self.product_classes(),
540 }
541 }
542
543 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
545 let list: Vec<String> = self
546 .symbols
547 .split([',', ',', ' '])
548 .map(|s| s.trim().to_string())
549 .filter(|s| !s.is_empty())
550 .collect();
551 if list.is_empty() {
552 return Err("订阅解析: 标的位空".to_string());
553 }
554 Ok(list)
555 }
556
557 pub fn validate(&self) -> Result<(), String> {
566 fn front(name: &str, v: &str) -> Result<(), String> {
567 let v = v.trim();
568 let Some(rest) = v.strip_prefix("tcp://") else {
569 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
570 };
571 let Some((host, port)) = rest.rsplit_once(':') else {
572 return Err(format!("{name} 缺端口:{name} {:?}", v));
573 };
574 if host.trim().is_empty() {
575 return Err(format!("{name} 主机位空:{}", v));
576 }
577 let port: u32 = port
578 .parse()
579 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
580 if port == 0 || port > 65535 {
581 return Err(format!("{name} 端口越界(1..=65535):{}", v));
582 }
583 Ok(())
584 }
585 if self.broker_id.trim().is_empty() {
586 return Err("broker 位空".to_string());
587 }
588 if self.td_user.trim().is_empty() {
589 return Err("交易账号位空".to_string());
590 }
591 front("td 前置", &self.td_addr)?;
592 if !self.md_addr.trim().is_empty() {
593 front("md 前置", &self.md_addr)?;
594 }
595 if self.td_app_id.trim().is_empty() {
596 return Err("AppID 位空(CTP 登录必需)".to_string());
597 }
598 if self.td_auth_code.trim().is_empty() {
599 return Err("AuthCode 位空(CTP 登录必需)".to_string());
600 }
601 Ok(())
602 }
603}
604
605#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
612pub enum AccountStage {
613 #[default]
615 NotAssembled,
616 Assembling,
618 Assembled,
620 AssembleFailed,
622}
623
624impl AccountStage {
625 #[must_use]
627 pub fn label(self) -> &'static str {
628 match self {
629 AccountStage::NotAssembled => "未装配",
630 AccountStage::Assembling => "装配中…",
631 AccountStage::Assembled => "已装配·登录态未知",
634 AccountStage::AssembleFailed => "装配失败",
635 }
636 }
637}
638
639#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
644pub enum LoginState {
645 #[default]
650 Unknown,
651 Online,
653 LoginFailed,
655 FrontDown,
657 QueryStarved,
659 Stopped,
661}
662
663impl LoginState {
664 #[must_use]
666 pub fn label(self) -> &'static str {
667 match self {
668 LoginState::Unknown => "登录态未知",
669 LoginState::Online => "在线",
670 LoginState::LoginFailed => "登录失败",
671 LoginState::FrontDown => "掉线",
672 LoginState::QueryStarved => "查询断供",
673 LoginState::Stopped => "已断开",
674 }
675 }
676
677 #[must_use]
679 pub fn is_alarm(self) -> bool {
680 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
681 }
682}
683
684fn count_note(label: &str, n: u32) -> Option<String> {
686 (n > 0).then(|| format!("{label} {n} 次"))
687}
688
689fn brief_counter_reason(full: &str) -> Option<String> {
696 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
697 let body = tail.trim().trim_start_matches("CTP:").trim();
698 if body.is_empty() {
699 return None;
700 }
701 let code = full
702 .rfind("代码:")
703 .map(|i| {
704 let rest = &full[i + "代码:".len()..];
705 let end = rest.find(',').unwrap_or(rest.len());
706 rest[..end].trim().to_string()
707 })
708 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
709 Some(match code {
710 Some(c) => format!("{body}({c})"),
711 None => body.to_string(),
712 })
713}
714
715#[derive(Debug, Clone, PartialEq, Default)]
721pub struct AccountRow {
722 pub gateway: String,
724 pub alias: String,
726 pub broker_id: String,
727 pub account: String,
728 pub md_source: bool,
730 pub form: CtpForm,
732 pub restored: bool,
738 pub auto_connect: bool,
743 pub password_ref: bool,
748 pub stage: AccountStage,
750 pub fail_note: Option<String>,
752 pub effective_classes: Vec<String>,
757 pub connected: bool,
760 pub login: LoginState,
764 pub status_note: Option<String>,
767 pub login_brief: Option<String>,
774 pub chain_note: Option<String>,
782}
783
784impl AccountRow {
785 pub fn derive(form: &CtpForm) -> Result<Self, String> {
790 let broker = form.broker_id.trim();
791 let account = form.td_user.trim();
792 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
793 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
794 })?;
795 Ok(Self {
796 gateway,
797 alias: String::new(),
798 broker_id: broker.to_string(),
799 account: account.to_string(),
800 md_source: false,
801 form: form.clone(),
802 stage: AccountStage::NotAssembled,
803 fail_note: None,
804 effective_classes: Vec::new(),
805 connected: false,
806 login: LoginState::Unknown,
807 status_note: None,
808 login_brief: None,
809 chain_note: None,
810 restored: false,
813 auto_connect: false,
814 password_ref: false,
815 })
816 }
817
818 #[must_use]
824 pub fn status_label(&self) -> String {
825 if self.stage == AccountStage::Assembled {
826 match self.login {
827 LoginState::Unknown => "已装配·登录态未知".to_string(),
828 other => format!("已装配·{}", other.label()),
829 }
830 } else {
831 self.stage.label().to_string()
832 }
833 }
834
835 pub fn apply_status(
848 &mut self,
849 st: &ek_ctp::plugin::GatewayStatus,
850 now_ms: i64,
851 reject_limit: u32,
852 stale_secs: i64,
853 ) {
854 let td_side = !self.form.td_addr.trim().is_empty();
855 let (login, note) = if td_side {
856 if st.td_login_failed && !st.td_logged_in {
857 (
858 LoginState::LoginFailed,
859 st.td_last_error
860 .clone()
861 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
862 )
863 } else if st.td_logged_in {
864 if st.td_query_reject_streak >= reject_limit {
865 (
866 LoginState::QueryStarved,
867 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
868 )
869 } else if let Some(last) = st.td_last_account_ms {
870 let idle_ms = now_ms.saturating_sub(last);
871 if idle_ms > stale_secs.saturating_mul(1_000) {
872 (
873 LoginState::QueryStarved,
874 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
875 )
876 } else {
877 (LoginState::Online, count_note("掉线", st.td_disconnects))
878 }
879 } else {
880 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
887 match since_login_ms {
888 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
889 LoginState::QueryStarved,
890 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
891 ),
892 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
893 }
894 }
895 } else if self.login == LoginState::Stopped {
896 let extra = if st.td_disconnects > 0 {
897 format!("(此前掉线 {} 次)", st.td_disconnects)
898 } else {
899 String::new()
900 };
901 (
902 LoginState::Stopped,
903 Some(format!("会话已停{extra};『连接』可恢复")),
904 )
905 } else if st.td_disconnects > 0 {
906 (
907 LoginState::FrontDown,
908 Some(format!("掉线 {} 次", st.td_disconnects)),
909 )
910 } else {
911 (LoginState::Unknown, st.td_last_error.clone())
912 }
913 } else if st.md_logged_in {
914 (
915 LoginState::Online,
916 count_note("掉线", st.md_disconnects)
917 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
918 )
919 } else if self.login == LoginState::Stopped {
920 (
921 LoginState::Stopped,
922 Some("会话已停;『连接』可恢复".to_string()),
923 )
924 } else if st.md_disconnects > 0 {
925 (
926 LoginState::FrontDown,
927 Some(format!("掉线 {} 次", st.md_disconnects)),
928 )
929 } else {
930 (LoginState::Unknown, st.md_last_error.clone())
931 };
932 let note = if td_side && st.td_instrument_rejects > 0 {
937 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
938 match note {
939 Some(base) if base.contains("合约查询被拒") => Some(base),
941 Some(base) => Some(format!("{base};{add}")),
942 None => Some(add),
943 }
944 } else {
945 note
946 };
947 self.login = login;
948 self.status_note = note;
949 self.login_brief = if login.is_alarm() {
952 st.td_last_error
953 .as_deref()
954 .or(st.md_last_error.as_deref())
955 .and_then(brief_counter_reason)
956 } else {
957 None
958 };
959 }
960
961 #[must_use]
963 pub fn display_name(&self) -> String {
964 let a = self.alias.trim();
965 if a.is_empty() {
966 self.account.clone()
967 } else {
968 a.to_string()
969 }
970 }
971}
972
973#[derive(Debug, Clone, PartialEq, Default)]
976pub struct TradeSnap {
977 pub orders: Vec<ek_data::order::OrderData>,
978 pub trades: Vec<ek_data::trade::TradeData>,
979 pub positions: Vec<ek_data::position::PositionData>,
980 pub accounts: Vec<ek_data::account::AccountData>,
982}
983
984pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
987 let interval: i64 = form
988 .interval
989 .trim()
990 .parse()
991 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
992 let symbol = form.symbol.trim().to_string();
993 if symbol.is_empty() {
994 return Err("回灌解析: symbol 空".to_string());
995 }
996 if form.csv.trim().is_empty() {
997 return Err("回灌解析: csv 空".to_string());
998 }
999 let mut bars = Vec::new();
1000 for (i, raw) in form.csv.lines().enumerate() {
1001 let line = raw.trim();
1002 if line.is_empty() || line.starts_with('#') {
1003 continue;
1004 }
1005 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
1006 if parts.len() != 6 {
1007 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
1008 }
1009 let num = |s: &str| -> Result<f64, String> {
1010 s.parse::<f64>()
1011 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
1012 };
1013 let dt_s: i64 = parts[0]
1014 .parse()
1015 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
1016 bars.push(BarData {
1017 symbol: symbol.clone(),
1018 datetime: dt_s * 1_000_000_000,
1019 interval: Some(
1020 ek_dm::wire::interval_of_seconds(interval)
1021 .unwrap_or(ek_data::constant::Interval::MINUTE),
1022 ),
1023 open_price: num(parts[1])?,
1024 high_price: num(parts[2])?,
1025 low_price: num(parts[3])?,
1026 close_price: num(parts[4])?,
1027 volume: num(parts[5])?,
1028 ..BarData::default()
1029 });
1030 }
1031 if bars.is_empty() {
1032 return Err("回灌解析: 无有效行".to_string());
1033 }
1034 Ok((interval, symbol, bars))
1035}
1036
1037#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1042pub enum TopLeftTab {
1043 #[default]
1044 Orders,
1045 Active,
1046}
1047
1048impl TopLeftTab {
1049 pub fn label(self) -> &'static str {
1050 match self {
1051 Self::Orders => "委托",
1052 Self::Active => "未成交",
1053 }
1054 }
1055}
1056
1057#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1059pub enum BottomRightTab {
1060 #[default]
1061 Account,
1062 Logs,
1063}
1064
1065impl BottomRightTab {
1066 pub fn label(self) -> &'static str {
1067 match self {
1068 Self::Account => "资金",
1069 Self::Logs => "日志",
1070 }
1071 }
1072}
1073
1074#[derive(Debug, Clone, Copy, PartialEq)]
1077pub struct DataSplit {
1078 pub col: f32,
1080 pub row: f32,
1082}
1083
1084impl Default for DataSplit {
1085 fn default() -> Self {
1086 Self {
1087 col: crate::splitters::DATA_SPLIT_DEFAULT,
1088 row: crate::splitters::DATA_SPLIT_DEFAULT,
1089 }
1090 }
1091}
1092
1093pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
1096 ek_data::constant::Exchange::SHFE,
1097 ek_data::constant::Exchange::DCE,
1098 ek_data::constant::Exchange::CZCE,
1099 ek_data::constant::Exchange::CFFEX,
1100 ek_data::constant::Exchange::INE,
1101 ek_data::constant::Exchange::GFEX,
1102];
1103
1104#[derive(Debug, Default)]
1114pub struct TickStore {
1115 ticks: BTreeMap<String, ek_data::tick::TickData>,
1116}
1117
1118impl TickStore {
1119 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1122 self.ticks
1123 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1124 }
1125
1126 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1128 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1129 v.sort_by(|a, b| {
1130 (a.exchange.as_str().to_string(), a.symbol.clone())
1131 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1132 });
1133 v
1134 }
1135
1136 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1138 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1139 for t in self.rows() {
1140 let ex = t.exchange.as_str().to_string();
1141 match out.last_mut() {
1142 Some((e, v)) if *e == ex => v.push(t),
1143 _ => out.push((ex, vec![t])),
1144 }
1145 }
1146 out
1147 }
1148
1149 pub fn len(&self) -> usize {
1150 self.ticks.len()
1151 }
1152
1153 pub fn is_empty(&self) -> bool {
1154 self.ticks.is_empty()
1155 }
1156}
1157
1158#[derive(Debug, Clone, PartialEq)]
1160pub struct OrderForm {
1161 pub exchange: String,
1163 pub symbol: String,
1165 pub direction: ek_data::constant::Direction,
1166 pub offset: ek_data::constant::Offset,
1167 pub order_type: ek_data::constant::OrderType,
1168 pub price: String,
1170 pub volume: String,
1171 pub gateway: String,
1173 pub follow_price: bool,
1175 pub auto_split: bool,
1178}
1179
1180impl Default for OrderForm {
1181 fn default() -> Self {
1182 Self {
1183 exchange: "SHFE".to_string(),
1184 symbol: String::new(),
1185 direction: ek_data::constant::Direction::LONG,
1186 offset: ek_data::constant::Offset::OPEN,
1187 order_type: ek_data::constant::OrderType::LIMIT,
1188 price: String::new(),
1189 volume: "1".to_string(),
1190 gateway: String::new(),
1191 follow_price: true,
1192 auto_split: true,
1193 }
1194 }
1195}
1196
1197impl OrderForm {
1198 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1200 use std::str::FromStr;
1201 let symbol = self.symbol.trim();
1202 if symbol.is_empty() {
1203 return Err("下单: 代码为空".to_string());
1204 }
1205 if symbol.contains('.') {
1206 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1207 }
1208 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1209 .map_err(|e| format!("下单: {e}"))?;
1210 let price: f64 = self
1211 .price
1212 .trim()
1213 .parse()
1214 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1215 let volume: f64 = self
1216 .volume
1217 .trim()
1218 .parse()
1219 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1220 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1221 return Err("下单: 限价单价格须为正".to_string());
1222 }
1223 if volume <= 0.0 {
1224 return Err("下单: 数量须为正".to_string());
1225 }
1226 Ok(ek_data::order_request::OrderRequest {
1227 symbol: symbol.to_string(),
1228 exchange,
1229 direction: self.direction,
1230 order_type: self.order_type,
1231 volume,
1232 price,
1233 offset: self.offset,
1234 ..ek_data::order_request::OrderRequest::default()
1235 })
1236 }
1237}
1238
1239#[derive(Debug, Default)]
1245pub struct ContractStore {
1246 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1247}
1248
1249impl ContractStore {
1250 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1251 self.contracts
1252 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1253 }
1254
1255 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1257 let p = pattern.trim();
1258 self.contracts
1259 .values()
1260 .filter(|c| {
1261 p.is_empty()
1262 || c.symbol.contains(p)
1263 || c.exchange.as_str().eq_ignore_ascii_case(p)
1264 || c.name.contains(p)
1265 })
1266 .cloned()
1267 .collect()
1268 }
1269
1270 pub fn len(&self) -> usize {
1271 self.contracts.len()
1272 }
1273
1274 pub fn is_empty(&self) -> bool {
1275 self.contracts.is_empty()
1276 }
1277}
1278
1279#[derive(Debug, Default)]
1281pub struct UiShared {
1282 logs: Mutex<LogRing>,
1283 snap: Mutex<Arc<BusSnap>>,
1284 market: MarketStore,
1285 ctp_form: Mutex<CtpForm>,
1286 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1289 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1290 trade_ids: Mutex<HashSet<String>>,
1298 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1299 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1300 trade_seq: AtomicU64,
1302 trade_dropped: AtomicU64,
1304 trade_dirty: AtomicBool,
1306 trade_snap: Mutex<Arc<TradeSnap>>,
1308 tick_store: Mutex<TickStore>,
1309 order_form: Mutex<OrderForm>,
1310 data_split: Mutex<DataSplit>,
1313 selected_exchange: Mutex<String>,
1315 topleft_tab: Mutex<TopLeftTab>,
1317 bottomright_tab: Mutex<BottomRightTab>,
1319 ops_window: AtomicBool,
1320 accounts_window: AtomicBool,
1323 contracts_window: AtomicBool,
1324 contract_store: Mutex<ContractStore>,
1325 contract_query: Mutex<String>,
1326 contract_selected: Mutex<Vec<String>>,
1327 kline_windows: Mutex<Vec<String>>,
1328 kline_fetches: Mutex<Vec<KlineFetch>>,
1330 kline_fetched: Mutex<HashSet<(String, i64)>>,
1332 rev: AtomicU64,
1333 stop: AtomicBool,
1335 cmds: Mutex<VecDeque<UiCmd>>,
1337 pending_confirm: Mutex<Option<String>>,
1339 accounts: Mutex<Vec<AccountRow>>,
1341 account_edit_window: AtomicBool,
1343 editing_account: Mutex<Option<String>>,
1345 edit_alias: Mutex<String>,
1347 edit_error: Mutex<Option<String>>,
1349 md_source: Mutex<String>,
1351 subscriptions: Mutex<Vec<String>>,
1353 row_sel: Mutex<BTreeMap<String, String>>,
1356 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1360 status_cursor: AtomicUsize,
1366 status_seed: Mutex<Vec<String>>,
1369 list_tick: AtomicU32,
1371 list_due: AtomicBool,
1375 seed_due: AtomicBool,
1379 pipeline_due: AtomicBool,
1381 cfg_due: AtomicBool,
1383 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1386 cfg_delete_arm: Mutex<Option<String>>,
1388 theme: AtomicU8,
1392 theme_persist_due: AtomicBool,
1395 auto_connect_persist_due: AtomicBool,
1399 bad_payload: Mutex<BTreeMap<String, u64>>,
1401 pipeline_lines: Mutex<Vec<String>>,
1404 status_thresholds: Mutex<Option<(u32, i64)>>,
1407 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1409 toasts: crate::toast::ToastsCell,
1412 alerted: Mutex<std::collections::BTreeSet<String>>,
1414 exit_confirm_open: AtomicBool,
1416 exit_confirmed: AtomicBool,
1418 pluginwin_open: AtomicBool,
1420 bundle_open: AtomicBool,
1422 assembly_locked: AtomicBool,
1425}
1426
1427impl UiShared {
1428 pub fn new() -> Self {
1430 Self::default()
1431 }
1432
1433 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1437 self.tick_store.lock().unwrap().put(t);
1438 self.bump();
1439 }
1440
1441 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1443 self.tick_store.lock().unwrap().rows()
1444 }
1445
1446 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1448 self.tick_store.lock().unwrap().groups()
1449 }
1450
1451 pub fn tick_count(&self) -> usize {
1453 self.tick_store.lock().unwrap().len()
1454 }
1455
1456 pub fn order_form(&self) -> OrderForm {
1458 self.order_form.lock().unwrap().clone()
1459 }
1460
1461 pub fn set_order_form(&self, f: OrderForm) {
1463 *self.order_form.lock().unwrap() = f;
1464 self.bump();
1465 }
1466
1467 pub fn selected_exchange(&self) -> String {
1471 let s = self.selected_exchange.lock().unwrap().clone();
1472 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1473 s
1474 } else {
1475 MARKET_ORDER[0].as_str().to_string()
1476 }
1477 }
1478
1479 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1481 *self.selected_exchange.lock().unwrap() = s.into();
1482 self.bump();
1483 }
1484
1485 pub fn topleft_tab(&self) -> TopLeftTab {
1487 *self.topleft_tab.lock().unwrap()
1488 }
1489
1490 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1492 *self.topleft_tab.lock().unwrap() = t;
1493 self.bump();
1494 }
1495
1496 pub fn bottomright_tab(&self) -> BottomRightTab {
1498 *self.bottomright_tab.lock().unwrap()
1499 }
1500
1501 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1503 *self.bottomright_tab.lock().unwrap() = t;
1504 self.bump();
1505 }
1506
1507 pub fn data_split(&self) -> DataSplit {
1509 *self.data_split.lock().unwrap()
1510 }
1511
1512 pub fn set_data_split(&self, s: DataSplit) {
1515 let s = DataSplit {
1516 col: crate::splitters::clamp_frac(s.col),
1517 row: crate::splitters::clamp_frac(s.row),
1518 };
1519 let mut g = self.data_split.lock().unwrap();
1520 if g.col != s.col || g.row != s.row {
1521 *g = s;
1522 self.bump();
1523 }
1524 }
1525
1526 pub fn reset_layout(&self) {
1530 *self.data_split.lock().unwrap() = DataSplit::default();
1531 *self.selected_exchange.lock().unwrap() = String::new();
1532 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1533 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1534 self.bump();
1535 }
1536
1537 pub fn ops_window_open(&self) -> bool {
1541 self.ops_window.load(Ordering::Acquire)
1542 }
1543
1544 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1547 self.table_sorts
1548 .lock()
1549 .unwrap()
1550 .get(table)
1551 .copied()
1552 .flatten()
1553 }
1554
1555 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1557 let mut m = self.table_sorts.lock().unwrap();
1558 let asc = match m.get(table).copied().flatten() {
1559 Some((c, a)) if c == col => !a,
1560 _ => true,
1561 };
1562 m.insert(table.to_string(), Some((col, asc)));
1563 }
1564
1565 pub fn set_ops_window_open(&self, v: bool) {
1566 self.ops_window.store(v, Ordering::Release);
1567 self.bump();
1568 }
1569
1570 pub fn accounts_window_open(&self) -> bool {
1574 self.accounts_window.load(Ordering::Acquire)
1575 }
1576
1577 pub fn set_accounts_window_open(&self, v: bool) {
1579 self.accounts_window.store(v, Ordering::Release);
1580 self.bump();
1581 }
1582
1583 pub fn account_edit_open(&self) -> bool {
1585 self.account_edit_window.load(Ordering::Acquire)
1586 }
1587
1588 pub fn editing_account(&self) -> Option<String> {
1590 self.editing_account.lock().unwrap().clone()
1591 }
1592
1593 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1597 match gateway {
1598 Some(gw) => {
1599 let Some(row) = self.account(gw) else {
1600 return false;
1601 };
1602 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1603 *self.edit_alias.lock().unwrap() = row.alias.clone();
1604 self.set_ctp_form(row.form.clone());
1605 }
1606 None => {
1607 *self.editing_account.lock().unwrap() = None;
1608 *self.edit_alias.lock().unwrap() = String::new();
1609 let draft = self.draft_from_existing();
1610 self.set_ctp_form(draft);
1611 }
1612 }
1613 *self.edit_error.lock().unwrap() = None;
1614 self.account_edit_window.store(true, Ordering::Release);
1615 self.bump();
1616 true
1617 }
1618
1619 pub fn end_account_edit(&self) {
1621 *self.editing_account.lock().unwrap() = None;
1622 *self.edit_alias.lock().unwrap() = String::new();
1623 *self.edit_error.lock().unwrap() = None;
1624 self.account_edit_window.store(false, Ordering::Release);
1625 let draft = self.draft_from_existing();
1626 self.set_ctp_form(draft);
1627 self.bump();
1628 }
1629
1630 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1636 if self.editing_account().as_deref() == Some(gateway) {
1637 self.end_account_edit();
1638 }
1639 }
1640
1641 pub fn edit_alias(&self) -> String {
1643 self.edit_alias.lock().unwrap().clone()
1644 }
1645
1646 pub fn set_edit_alias(&self, v: impl Into<String>) {
1648 *self.edit_alias.lock().unwrap() = v.into();
1649 self.bump();
1650 }
1651
1652 pub fn edit_error(&self) -> Option<String> {
1654 self.edit_error.lock().unwrap().clone()
1655 }
1656
1657 pub fn set_edit_error(&self, e: Option<String>) {
1659 *self.edit_error.lock().unwrap() = e;
1660 self.bump();
1661 }
1662
1663 pub fn account_edit_dirty(&self) -> bool {
1665 let editing = self.editing_account();
1666 let draft = self.ctp_form();
1667 match editing {
1668 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1669 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1670 }
1671 }
1672
1673 pub fn accounts(&self) -> Vec<AccountRow> {
1675 self.accounts.lock().unwrap().clone()
1676 }
1677
1678 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1680 self.accounts
1681 .lock()
1682 .unwrap()
1683 .iter()
1684 .find(|a| a.gateway == gateway)
1685 .cloned()
1686 }
1687
1688 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1694 let mut acc = self.accounts.lock().unwrap();
1695 if acc.iter().any(|a| a.gateway == row.gateway) {
1696 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1697 }
1698 let alias = row.alias.trim();
1699 if !alias.is_empty()
1700 && acc
1701 .iter()
1702 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1703 {
1704 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1705 }
1706 if acc.len() >= MAX_ACCOUNTS {
1707 return Err(format!(
1708 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1709 ));
1710 }
1711 acc.push(row);
1712 drop(acc);
1713 self.bump();
1714 Ok(())
1715 }
1716
1717 #[must_use]
1719 pub fn can_add_account(&self) -> bool {
1720 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1721 }
1722
1723 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1729 let alias = alias.trim().to_string();
1730 let mut acc = self.accounts.lock().unwrap();
1731 if !alias.is_empty()
1732 && acc
1733 .iter()
1734 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1735 {
1736 return false;
1737 }
1738 match acc.iter_mut().find(|a| a.gateway == gateway) {
1739 Some(a) => {
1740 a.alias = alias;
1741 drop(acc);
1742 self.bump();
1743 true
1744 }
1745 None => false,
1746 }
1747 }
1748
1749 pub fn note_connect_sent(&self, gateway: &str) {
1754 let mut acc = self.accounts.lock().unwrap();
1755 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1756 a.stage = AccountStage::Assembling;
1757 a.fail_note = None;
1758 a.login = LoginState::Unknown;
1759 a.status_note = None;
1760 a.login_brief = None;
1761 a.chain_note = None;
1763 drop(acc);
1764 self.bump();
1765 }
1766 }
1767
1768 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1772 let mut acc = self.accounts.lock().unwrap();
1773 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1774 match result {
1775 Ok(()) => {
1776 a.stage = AccountStage::Assembled;
1777 a.fail_note = None;
1778 }
1779 Err(e) => {
1780 a.stage = AccountStage::AssembleFailed;
1781 a.fail_note = Some(e);
1782 }
1783 }
1784 drop(acc);
1785 self.bump();
1786 }
1787 }
1788
1789 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1793 let mut acc = self.accounts.lock().unwrap();
1794 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1795 a.effective_classes = classes.to_vec();
1796 drop(acc);
1797 self.bump();
1798 }
1799 }
1800
1801 pub fn note_chain(&self, gateway: &str, note: &str) {
1806 let mut acc = self.accounts.lock().unwrap();
1807 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1808 if a.chain_note.as_deref() == Some(note) {
1809 return; }
1811 a.chain_note = Some(note.to_string());
1812 drop(acc);
1813 self.bump();
1814 }
1815 }
1816
1817 #[must_use]
1821 pub fn status_thresholds(&self) -> (u32, i64) {
1822 self.status_thresholds
1823 .lock()
1824 .unwrap()
1825 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1826 }
1827
1828 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1830 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1831 self.bump();
1832 }
1833
1834 pub fn list_poll_due(&self) -> bool {
1839 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1840 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1841 }
1842
1843 pub fn request_list_refresh(&self) {
1845 self.list_due.store(true, Ordering::Release);
1846 }
1847
1848 pub fn request_trade_reseed(&self) {
1850 self.seed_due.store(true, Ordering::Release);
1851 }
1852
1853 pub fn take_trade_reseed(&self) -> bool {
1855 self.seed_due.swap(false, Ordering::AcqRel)
1856 }
1857
1858 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1863 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1867 for gw in present {
1868 if !uniq.iter().any(|u| u == gw) {
1869 uniq.push(gw.clone());
1870 }
1871 }
1872 let n = uniq.len();
1873 if n == 0 {
1874 return Vec::new();
1875 }
1876 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1877 let take = STATUS_ASKS_PER_TICK.min(n);
1878 let mut out = Vec::with_capacity(take);
1879 for i in 0..take {
1880 out.push(uniq[(start + i) % n].clone());
1881 }
1882 out
1883 }
1884
1885 pub fn request_status_seed(&self, gateway: &str) {
1887 let mut q = self.status_seed.lock().unwrap();
1888 if !q.iter().any(|g| g == gateway) {
1889 q.push(gateway.to_string());
1890 }
1891 }
1892
1893 pub fn take_status_seeds(&self) -> Vec<String> {
1895 let mut q = self.status_seed.lock().unwrap();
1896 std::mem::take(&mut *q)
1897 }
1898
1899 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
1909 let gw = ev.gateway.as_str();
1910 let text = ev.describe();
1911 self.log_line(format!("[INFO][{gw}] {text}"));
1912 let note = text
1913 .strip_prefix(CHAIN_STATE_PREFIX)
1914 .map(|s| s.trim().to_string())
1915 .unwrap_or_else(|| text.clone());
1916 self.note_chain(gw, ¬e);
1917 self.note_gateway_status(gw, &ev.status);
1918 let alarm_kind = matches!(
1919 ev.change,
1920 ek_ctp::GatewayChange::TdDisconnected
1921 | ek_ctp::GatewayChange::TdAuthFailed
1922 | ek_ctp::GatewayChange::TdLoginFailed
1923 | ek_ctp::GatewayChange::ContractQueryStopped
1924 | ek_ctp::GatewayChange::QueryRejected
1925 | ek_ctp::GatewayChange::MdDisconnected
1926 | ek_ctp::GatewayChange::MdLoginFailed
1927 | ek_ctp::GatewayChange::SubscribeRejected
1928 );
1929 if alarm_kind {
1930 let session = if ev.change.kind().starts_with("md-") {
1932 ev.status.md_disconnects
1933 } else {
1934 ev.status.td_disconnects
1935 };
1936 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
1937 if self.alert_once(key) {
1938 self.push_alert(
1939 crate::toast::AlertLevel::Warning,
1940 format!("账户 {gw}:{}", note.trim()),
1941 );
1942 }
1943 }
1944 }
1945
1946 pub fn request_pipeline_refresh(&self) {
1948 self.pipeline_due.store(true, Ordering::Release);
1949 }
1950 pub fn take_pipeline_due(&self) -> bool {
1952 self.pipeline_due.swap(false, Ordering::AcqRel)
1953 }
1954
1955 pub fn request_cfg_refresh(&self) {
1957 self.cfg_due.store(true, Ordering::Release);
1958 }
1959
1960 pub fn take_cfg_due(&self) -> bool {
1962 self.cfg_due.swap(false, Ordering::AcqRel)
1963 }
1964
1965 pub fn theme(&self) -> crate::tokens::Theme {
1967 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
1968 }
1969
1970 pub fn set_theme(&self, theme: crate::tokens::Theme) {
1973 self.theme.store(theme.as_u8(), Ordering::Release);
1974 self.bump();
1975 }
1976
1977 pub fn toggle_theme(&self) -> crate::tokens::Theme {
1979 let next = self.theme().flipped();
1980 self.theme.store(next.as_u8(), Ordering::Release);
1981 self.theme_persist_due.store(true, Ordering::Release);
1982 self.bump();
1983 next
1984 }
1985
1986 pub fn take_theme_persist_due(&self) -> bool {
1988 self.theme_persist_due.swap(false, Ordering::AcqRel)
1989 }
1990
1991 pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
1995 {
1996 let mut acc = self.accounts.lock().unwrap();
1997 match acc.iter_mut().find(|a| a.gateway == gateway) {
1998 Some(a) => {
1999 if a.auto_connect != on {
2000 a.auto_connect = on;
2001 self.auto_connect_persist_due.store(true, Ordering::Release);
2002 self.bump();
2003 }
2004 }
2005 None => return Err(format!("账户 {gateway} 不在清册")),
2006 }
2007 }
2008 Ok(())
2009 }
2010
2011 pub fn take_auto_connect_persist_due(&self) -> bool {
2013 self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
2014 }
2015
2016 pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
2031 let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
2033 match store.load(crate::plugin::THEME_NS) {
2034 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
2035 for (k, _f) in map.iter() {
2036 if let Some(gw) = k.strip_prefix("auto_connect.") {
2037 flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
2038 }
2039 }
2040 }
2041 Ok(_) => {} Err(_) => {} }
2044 let mut restored: Vec<String> = Vec::new();
2045 let ns_list = match store.list_namespaces() {
2046 Ok(v) => v,
2047 Err(_) => {
2048 let msg = "[ui] 档案清点失败:存储根不可读(具体见运行日志)".to_string();
2049 self.log_line(msg.clone());
2050 return restored;
2051 }
2052 };
2053 for ns in ns_list {
2054 let Some(gw) = ns.strip_prefix("ctp.") else {
2055 continue;
2056 };
2057 if self.account(gw).is_some() {
2058 self.log_line(format!(
2059 "[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
2060 ));
2061 continue;
2062 }
2063 let map = match store.load(&ns) {
2064 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
2065 Ok(_) => {
2066 self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
2067 continue;
2068 }
2069 Err(e) => {
2070 self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
2071 continue;
2072 }
2073 };
2074 let password_ref = map
2076 .get_secret_key("td_password")
2077 .is_some_and(|k| !k.is_empty());
2078 let mut form = CtpForm {
2079 broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
2080 td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
2081 td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
2082 td_password: String::new(),
2083 td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
2084 td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
2085 md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
2086 md_user: String::new(),
2087 md_password: String::new(),
2088 symbols: String::new(),
2089 accept_futures: false,
2090 accept_options: false,
2091 };
2092 for c in map
2095 .get_plain("product_classes")
2096 .unwrap_or_default()
2097 .split(',')
2098 {
2099 if c.trim() == "1" {
2100 form.accept_futures = true;
2101 }
2102 }
2103 let Ok(mut row) = AccountRow::derive(&form) else {
2104 self.log_line(format!(
2105 "[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
2106 ));
2107 continue;
2108 };
2109 row.restored = true;
2110 row.auto_connect = flags.get(gw).copied().unwrap_or(false);
2111 row.password_ref = password_ref;
2112 restored.push(row.gateway.clone());
2113 self.add_account(row).ok();
2114 }
2115 if !restored.is_empty() {
2116 self.log_line(format!(
2117 "[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
2118 restored.len()
2119 ));
2120 self.bump();
2121 }
2122 restored
2123 }
2124
2125 pub fn restore_saved_accounts(&self) -> Vec<String> {
2127 match ek_cfg::Store::eco() {
2128 Ok(store) => self.restore_saved_accounts_from(&store),
2129 Err(_) => {
2130 self.log_line(
2131 "[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
2132 );
2133 Vec::new()
2134 }
2135 }
2136 }
2137
2138 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
2140 *self.cfg_briefs.lock().unwrap() = Some(b);
2141 self.bump();
2142 }
2143 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
2145 self.cfg_briefs.lock().unwrap().clone()
2146 }
2147
2148 pub fn cfg_delete_armed(&self) -> Option<String> {
2150 self.cfg_delete_arm.lock().unwrap().clone()
2151 }
2152 pub fn arm_cfg_delete(&self, ns: Option<String>) {
2153 *self.cfg_delete_arm.lock().unwrap() = ns;
2154 self.bump();
2155 }
2156
2157 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
2159 *self.pipeline_lines.lock().unwrap() = lines;
2160 self.bump();
2161 }
2162
2163 pub fn pipeline_lines(&self) -> Vec<String> {
2165 self.pipeline_lines.lock().unwrap().clone()
2166 }
2167
2168 pub fn note_bad_payload(&self, line: &str) {
2174 let mut book = self.bad_payload.lock().unwrap();
2175 let n = book.entry(line.to_string()).or_insert(0);
2176 *n += 1;
2177 if *n == 1 {
2178 self.log_line(format!(
2179 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
2180 ));
2181 }
2182 }
2183
2184 #[must_use]
2186 pub fn bad_payload_count(&self, line: &str) -> u64 {
2187 self.bad_payload
2188 .lock()
2189 .unwrap()
2190 .get(line)
2191 .copied()
2192 .unwrap_or(0)
2193 }
2194
2195 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
2209 for r in rows {
2210 let (warn, crit) = pending_thresholds(r.soft_cap);
2211 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
2212 self.push_alert(
2213 crate::toast::AlertLevel::Warning,
2214 format!(
2215 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
2216 r.name, r.pending, r.peak, r.soft_cap
2217 ),
2218 );
2219 }
2220 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
2221 self.push_alert(
2222 crate::toast::AlertLevel::Error,
2223 format!(
2224 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
2225 r.name, r.pending, r.soft_cap
2226 ),
2227 );
2228 }
2229 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
2230 self.push_alert(
2231 crate::toast::AlertLevel::Error,
2232 format!(
2233 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
2234 r.name, r.soft_cap, r.spilled, r.peak
2235 ),
2236 );
2237 }
2238 }
2239 }
2240
2241 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
2249 let (limit, stale) = self.status_thresholds();
2250 let now = ek_ctp::map::now_epoch_ms();
2251 let mut alarm: Option<(LoginState, String)> = None;
2254 let mut found = false;
2255 {
2256 let mut acc = self.accounts.lock().unwrap();
2257 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2258 found = true;
2259 a.apply_status(st, now, limit, stale);
2260 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
2261 alarm = Some((
2262 a.login,
2263 a.status_note
2264 .clone()
2265 .unwrap_or_else(|| a.login.label().to_string()),
2266 ));
2267 }
2268 }
2269 }
2270 if found {
2271 self.bump();
2272 }
2273 if let Some((login, note)) = alarm
2274 && self.alert_once(format!("{gateway}|{}", login.label()))
2275 {
2276 self.push_alert(
2277 crate::toast::AlertLevel::Error,
2278 format!("账户 {gateway}:{note}"),
2279 );
2280 }
2281 }
2282
2283 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2287 let mut acc = self.accounts.lock().unwrap();
2288 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2289 a.login = LoginState::Unknown;
2290 a.status_note = Some(format!("状态口问不到:{why}"));
2291 a.login_brief = None;
2292 drop(acc);
2293 self.bump();
2294 }
2295 }
2296
2297 pub fn mark_account_stopped(&self, gateway: &str) {
2299 let mut acc = self.accounts.lock().unwrap();
2300 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2301 a.login = LoginState::Stopped;
2302 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2303 a.login_brief = None;
2304 drop(acc);
2305 self.bump();
2306 }
2307 }
2308
2309 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2312 let source = self.md_source();
2313 let mut f = form;
2314 if source.as_str() != gateway {
2315 f.md_addr.clear();
2316 f.md_user.clear();
2317 f.md_password.clear();
2318 }
2319 self.update_account_form(gateway, f.clone());
2320 f
2321 }
2322
2323 #[must_use]
2327 pub fn draft_from_existing(&self) -> CtpForm {
2328 let acc = self.accounts.lock().unwrap();
2329 let mut draft = CtpForm::default();
2330 if let Some(a) = acc.last() {
2331 draft.broker_id = a.form.broker_id.clone();
2332 draft.td_auth_code = a.form.td_auth_code.clone();
2333 draft.td_app_id = a.form.td_app_id.clone();
2334 draft.accept_futures = a.form.accept_futures;
2335 draft.accept_options = a.form.accept_options;
2336 }
2337 draft
2338 }
2339
2340 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2342 let mut acc = self.accounts.lock().unwrap();
2343 match acc.iter_mut().find(|a| a.gateway == gateway) {
2344 Some(a) => {
2345 a.form = form;
2346 drop(acc);
2347 self.bump();
2348 true
2349 }
2350 None => false,
2351 }
2352 }
2353
2354 pub fn remove_account(&self, gateway: &str) -> bool {
2356 let mut acc = self.accounts.lock().unwrap();
2357 let before = acc.len();
2358 acc.retain(|a| a.gateway != gateway);
2359 let removed = acc.len() != before;
2360 if removed {
2361 let mut md = self.md_source.lock().unwrap();
2363 if md.as_str() == gateway {
2364 md.clear();
2365 }
2366 drop(acc);
2367 self.bump();
2368 }
2369 removed
2370 }
2371
2372 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2381 let mut newly = Vec::new();
2382 {
2383 let mut acc = self.accounts.lock().unwrap();
2384 for a in acc.iter_mut() {
2385 let up = rows.iter().any(|r| r.name == a.gateway);
2386 if up && !a.connected {
2387 newly.push(a.gateway.clone());
2388 }
2389 a.connected = up;
2390 if up {
2391 a.stage = AccountStage::Assembled;
2392 a.fail_note = None;
2393 } else {
2394 a.chain_note = None;
2397 if a.stage == AccountStage::Assembled {
2398 a.stage = AccountStage::NotAssembled;
2399 a.login = LoginState::Unknown;
2400 a.status_note = None;
2401 a.login_brief = None;
2402 }
2403 }
2404 }
2405 }
2406 if !newly.is_empty() {
2407 self.bump();
2408 }
2409 newly
2410 }
2411
2412 pub fn md_source(&self) -> String {
2414 self.md_source.lock().unwrap().clone()
2415 }
2416
2417 pub fn accounts_locked(&self) -> bool {
2419 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2420 }
2421
2422 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2424 {
2425 let acc = self.accounts.lock().unwrap();
2426 if acc.iter().any(|a| a.connected) {
2427 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2428 }
2429 if !acc.iter().any(|a| a.gateway == gateway) {
2430 return Err(format!("账户 {gateway} 不在清册"));
2431 }
2432 }
2433 let mut acc = self.accounts.lock().unwrap();
2434 for a in acc.iter_mut() {
2435 a.md_source = a.gateway == gateway;
2436 }
2437 drop(acc);
2438 let mut md = self.md_source.lock().unwrap();
2439 *md = gateway.to_string();
2440 self.bump();
2441 Ok(())
2442 }
2443
2444 pub fn subscriptions(&self) -> Vec<String> {
2446 self.subscriptions.lock().unwrap().clone()
2447 }
2448
2449 pub fn note_subscriptions(&self, symbols: &[String]) {
2451 {
2452 let mut subs = self.subscriptions.lock().unwrap();
2453 for s in symbols {
2454 let s = s.trim();
2455 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2456 subs.push(s.to_string());
2457 }
2458 }
2459 }
2460 self.bump();
2461 }
2462
2463 pub fn default_gateway(&self) -> Option<String> {
2470 let chosen = {
2471 let f = self.order_form.lock().unwrap();
2472 f.gateway.trim().to_string()
2473 };
2474 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2475 return Some(chosen);
2476 }
2477 let acc = self.accounts.lock().unwrap();
2478 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2479 }
2480
2481 pub fn kline_symbols(&self) -> Vec<String> {
2483 self.kline_windows.lock().unwrap().clone()
2484 }
2485
2486 pub fn open_kline(&self, symbol: impl Into<String>) {
2488 let s = symbol.into();
2489 let mut v = self.kline_windows.lock().unwrap();
2490 if !v.contains(&s) {
2491 v.push(s.clone());
2492 }
2493 drop(v);
2494 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2495 let fq = if s.contains('.') {
2498 s.clone()
2499 } else {
2500 format!("{}.{}", self.selected_exchange(), s)
2501 };
2502 const KLINE_INTERVAL_SECS: i64 = 60;
2503 {
2504 let mut fetched = self.kline_fetched.lock().unwrap();
2505 if fetched.contains(&(fq.clone(), KLINE_INTERVAL_SECS)) {
2506 return;
2507 }
2508 fetched.insert((fq.clone(), KLINE_INTERVAL_SECS));
2509 }
2510 self.kline_fetches.lock().unwrap().push(KlineFetch {
2511 symbol: fq,
2512 interval_secs: KLINE_INTERVAL_SECS,
2513 });
2514 self.log_line(format!("[kline] 入队历史补数 {s}(1min×300,tq 桥)"));
2515 }
2516
2517 pub fn take_kline_fetches(&self) -> Vec<KlineFetch> {
2519 std::mem::take(&mut self.kline_fetches.lock().unwrap())
2520 }
2521
2522 pub fn kline_fetch_pending(&self) -> bool {
2524 !self.kline_fetches.lock().unwrap().is_empty()
2525 }
2526
2527 pub fn close_kline(&self, symbol: &str) {
2529 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2530 self.bump();
2531 }
2532
2533 pub fn contracts_window_open(&self) -> bool {
2537 self.contracts_window.load(Ordering::Acquire)
2538 }
2539
2540 pub fn set_contracts_window_open(&self, v: bool) {
2542 self.contracts_window.store(v, Ordering::Release);
2543 self.bump();
2544 }
2545
2546 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2553 self.contract_store.lock().unwrap().put(c);
2554 if !c.name.is_empty() {
2555 self.backfill_names(&c.symbol, &c.name);
2556 }
2557 self.bump();
2558 }
2559
2560 fn backfill_names(&self, symbol: &str, name: &str) {
2565 let mut touched = false;
2566 {
2567 let mut rows = self.trade_orders.lock().unwrap();
2568 for (_, (_, o)) in rows.iter_mut() {
2569 if o.symbol == symbol && o.name.is_empty() {
2570 o.name = name.to_string();
2571 touched = true;
2572 }
2573 }
2574 }
2575 {
2576 let mut rows = self.trade_trades.lock().unwrap();
2577 for (_, t) in rows.iter_mut() {
2578 if t.symbol == symbol && t.name.is_empty() {
2579 t.name = name.to_string();
2580 touched = true;
2581 }
2582 }
2583 }
2584 {
2585 let mut rows = self.trade_positions.lock().unwrap();
2586 for (_, (_, p)) in rows.iter_mut() {
2587 if p.symbol == symbol && p.name.is_empty() {
2588 p.name = name.to_string();
2589 touched = true;
2590 }
2591 }
2592 }
2593 if touched {
2594 self.trade_dirty.store(true, Ordering::Release);
2596 }
2597 }
2598
2599 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2601 self.contract_store.lock().unwrap().query(pattern)
2602 }
2603
2604 pub fn contract_count(&self) -> usize {
2606 self.contract_store.lock().unwrap().len()
2607 }
2608
2609 fn contract_name_of(&self, symbol: &str) -> Option<String> {
2611 self.contract_store
2612 .lock()
2613 .unwrap()
2614 .query(symbol)
2615 .into_iter()
2616 .find(|c| c.symbol == symbol && !c.name.is_empty())
2617 .map(|c| c.name)
2618 }
2619
2620 pub fn contract_query(&self) -> String {
2622 self.contract_query.lock().unwrap().clone()
2623 }
2624
2625 pub fn set_contract_query(&self, q: impl Into<String>) {
2627 *self.contract_query.lock().unwrap() = q.into();
2628 self.bump();
2629 }
2630
2631 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2633 let mut v = self.contract_selected.lock().unwrap();
2634 match v.iter().position(|s| s == en_symbol) {
2635 Some(i) => {
2636 v.remove(i);
2637 }
2638 None => v.push(en_symbol.to_string()),
2639 }
2640 self.bump();
2641 }
2642
2643 pub fn selected_contracts(&self) -> Vec<String> {
2645 self.contract_selected.lock().unwrap().clone()
2646 }
2647
2648 #[must_use]
2654 pub fn cancel_reject_note(msg: &str) -> String {
2655 let head = msg.split(' ').next().unwrap_or(msg);
2656 let note = if head.starts_with("unknown:order") {
2657 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
2658 } else if head.starts_with("mismatch:gateway") {
2659 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
2660 } else if head.starts_with("terminal:status") {
2661 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
2662 } else if head.starts_with("limit:cancel-guard") {
2663 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
2664 } else {
2665 ""
2666 };
2667 if note.is_empty() {
2668 format!("撤单被拒:{msg}")
2670 } else {
2671 format!("撤单被拒:{note}(机器码:{msg})")
2672 }
2673 }
2674
2675 pub fn log_line(&self, line: impl Into<String>) {
2677 self.logs.lock().unwrap().push(line);
2678 self.bump();
2679 }
2680
2681 pub fn clear_logs(&self) {
2683 self.logs.lock().unwrap().clear();
2684 self.bump();
2685 }
2686
2687 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2689 let r = self.logs.lock().unwrap();
2690 (r.snapshot(), r.spilled())
2691 }
2692
2693 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2695 let r = self.logs.lock().unwrap();
2696 (r.entries(), r.spilled())
2697 }
2698
2699 pub fn logs_len(&self) -> usize {
2701 self.logs.lock().unwrap().len()
2702 }
2703
2704 pub fn set_snapshot(&self, snap: BusSnap) {
2706 *self.snap.lock().unwrap() = Arc::new(snap);
2707 self.bump();
2708 }
2709
2710 pub fn snapshot(&self) -> Arc<BusSnap> {
2712 Arc::clone(&self.snap.lock().unwrap())
2713 }
2714
2715 pub fn rev(&self) -> u64 {
2717 self.rev.load(Ordering::Acquire)
2718 }
2719
2720 pub fn bump(&self) {
2722 self.rev.fetch_add(1, Ordering::AcqRel);
2723 }
2724
2725 pub fn request_stop(&self) {
2727 self.stop.store(true, Ordering::Release);
2728 }
2729
2730 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2733 self.pending_alerts
2734 .lock()
2735 .unwrap_or_else(std::sync::PoisonError::into_inner)
2736 .push((level, text.into()));
2737 }
2738
2739 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2741 let mut q = self
2742 .pending_alerts
2743 .lock()
2744 .unwrap_or_else(std::sync::PoisonError::into_inner);
2745 std::mem::take(&mut *q)
2746 }
2747
2748 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2750 self.toasts.lock()
2751 }
2752
2753 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2755 let key = key.into();
2756 self.alerted
2757 .lock()
2758 .unwrap_or_else(std::sync::PoisonError::into_inner)
2759 .insert(key)
2760 }
2761
2762 #[must_use]
2764 pub fn pending_alert_count(&self) -> usize {
2765 self.pending_alerts
2766 .lock()
2767 .unwrap_or_else(std::sync::PoisonError::into_inner)
2768 .len()
2769 }
2770
2771 #[must_use]
2773 pub fn alerted_count(&self) -> usize {
2774 self.alerted
2775 .lock()
2776 .unwrap_or_else(std::sync::PoisonError::into_inner)
2777 .len()
2778 }
2779
2780 pub fn exit_confirm_open(&self) -> bool {
2782 self.exit_confirm_open
2783 .load(std::sync::atomic::Ordering::Relaxed)
2784 }
2785
2786 pub fn open_exit_confirm(&self) {
2787 self.exit_confirm_open
2788 .store(true, std::sync::atomic::Ordering::Relaxed);
2789 }
2790
2791 pub fn close_exit_confirm(&self) {
2792 self.exit_confirm_open
2793 .store(false, std::sync::atomic::Ordering::Relaxed);
2794 }
2795
2796 pub fn exit_confirmed(&self) -> bool {
2797 self.exit_confirmed
2798 .load(std::sync::atomic::Ordering::Relaxed)
2799 }
2800
2801 pub fn confirm_exit(&self) {
2802 self.exit_confirmed
2803 .store(true, std::sync::atomic::Ordering::Relaxed);
2804 }
2805
2806 pub fn is_stop(&self) -> bool {
2807 self.stop.load(Ordering::Acquire)
2808 }
2809
2810 #[must_use]
2812 pub fn pluginwin_open(&self) -> bool {
2813 self.pluginwin_open.load(Ordering::Relaxed)
2814 }
2815
2816 pub fn set_pluginwin_open(&self, on: bool) {
2818 self.pluginwin_open.store(on, Ordering::Relaxed);
2819 }
2820
2821 #[must_use]
2823 pub fn bundle_open(&self) -> bool {
2824 self.bundle_open.load(Ordering::Relaxed)
2825 }
2826
2827 pub fn set_bundle_open(&self, on: bool) {
2829 self.bundle_open.store(on, Ordering::Relaxed);
2830 }
2831
2832 #[must_use]
2834 pub fn assembly_locked(&self) -> bool {
2835 self.assembly_locked.load(Ordering::Relaxed)
2836 }
2837
2838 pub fn set_assembly_locked(&self, on: bool) {
2840 self.assembly_locked.store(on, Ordering::Relaxed);
2841 }
2842
2843 pub fn push_cmd(&self, cmd: UiCmd) {
2847 self.cmds.lock().unwrap().push_back(cmd);
2848 self.bump();
2849 }
2850
2851 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2853 self.cmds.lock().unwrap().drain(..).collect()
2854 }
2855
2856 #[must_use]
2858 pub fn cmd_pending(&self) -> bool {
2859 !self.cmds.lock().unwrap().is_empty()
2860 }
2861
2862 pub fn pending_confirm(&self) -> Option<String> {
2864 self.pending_confirm.lock().unwrap().clone()
2865 }
2866
2867 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2869 let mut g = self.pending_confirm.lock().unwrap();
2870 match g.take() {
2871 Some(p) if p == name => ConfirmStep::Fired,
2872 Some(_) => {
2873 *g = Some(name.to_string());
2874 ConfirmStep::Switched
2875 }
2876 None => {
2877 *g = Some(name.to_string());
2878 ConfirmStep::Armed
2879 }
2880 }
2881 }
2882
2883 pub fn confirm_cancel(&self) {
2885 *self.pending_confirm.lock().unwrap() = None;
2886 self.bump();
2887 }
2888
2889 pub fn market(&self) -> &MarketStore {
2893 &self.market
2894 }
2895
2896 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2905 let named = |symbol: &str, name: String| -> String {
2908 if name.is_empty() {
2909 self.contract_name_of(symbol).unwrap_or(name)
2910 } else {
2911 name
2912 }
2913 };
2914 {
2915 let mut book = self.trade_orders.lock().unwrap();
2916 book.clear();
2917 for mut o in ts.orders.into_iter().rev() {
2918 o.name = named(&o.symbol, o.name);
2919 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2920 book.insert(o.en_orderid(), (seq, o));
2921 }
2922 }
2923 {
2924 let mut book = self.trade_trades.lock().unwrap();
2925 book.clear();
2926 let mut ids = self.trade_ids.lock().unwrap();
2927 ids.clear();
2928 for mut t in ts.trades.into_iter().rev() {
2929 t.name = named(&t.symbol, t.name);
2930 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2931 ids.insert(t.en_tradeid());
2932 book.push_front((seq, t));
2933 }
2934 }
2935 {
2936 let mut book = self.trade_positions.lock().unwrap();
2937 book.clear();
2938 for mut p in ts.positions.into_iter().rev() {
2939 p.name = named(&p.symbol, p.name);
2940 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2941 book.insert(p.en_positionid(), (seq, p));
2942 }
2943 }
2944 {
2945 let mut book = self.trade_funds.lock().unwrap();
2946 book.clear();
2947 for a in ts.accounts.into_iter().rev() {
2948 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2949 book.insert(a.en_accountid(), (seq, a));
2950 }
2951 }
2952 self.trade_dirty.store(true, Ordering::Release);
2953 self.bump();
2954 }
2955
2956 pub fn note_order(&self, o: &ek_data::order::OrderData) {
2964 let key = o.en_orderid();
2965 {
2966 let mut book = self.trade_orders.lock().unwrap();
2967 match book.get_mut(&key) {
2968 Some((_, cur)) => {
2969 cur.name.clone_from(&o.name);
2970 cur.traded = o.traded;
2971 cur.traded_price = o.traded_price;
2972 cur.status = o.status;
2973 cur.status_msg.clone_from(&o.status_msg);
2974 cur.datetime = o.datetime;
2975 }
2976 None => {
2977 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2978 book.insert(key, (seq, o.clone()));
2979 }
2980 }
2981 }
2982 self.trade_dirty.store(true, Ordering::Release);
2983 self.bump();
2984 }
2985
2986 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
2994 let id = t.en_tradeid();
2995 if self.trade_ids.lock().unwrap().contains(&id) {
2996 return;
2997 }
2998 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2999 let dropped = {
3000 let mut book = self.trade_trades.lock().unwrap();
3001 book.push_front((seq, t.clone()));
3002 self.trade_ids.lock().unwrap().insert(id);
3003 let over = book.len().saturating_sub(TRADE_CAP);
3004 for _ in 0..over {
3005 if let Some((_, old)) = book.pop_back() {
3007 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
3008 }
3009 }
3010 over
3011 };
3012 if dropped > 0 {
3013 let total = self
3014 .trade_dropped
3015 .fetch_add(dropped as u64, Ordering::AcqRel)
3016 + dropped as u64;
3017 if total == dropped as u64 {
3019 self.push_alert(
3020 crate::toast::AlertLevel::Warning,
3021 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
3022 );
3023 }
3024 }
3025 self.trade_dirty.store(true, Ordering::Release);
3026 self.bump();
3027 }
3028
3029 pub fn note_position(&self, p: &ek_data::position::PositionData) {
3034 let key = p.en_positionid();
3035 {
3036 let mut book = self.trade_positions.lock().unwrap();
3037 match book.get_mut(&key) {
3038 Some((_, cur)) => {
3039 cur.volume = p.volume;
3040 cur.yd_volume = p.yd_volume;
3041 cur.can_use_volume = p.can_use_volume;
3042 cur.frozen = p.frozen;
3043 cur.on_road_volume = p.on_road_volume;
3044 cur.market_value = p.market_value;
3045 cur.avg_price = p.avg_price;
3046 cur.price = p.price;
3047 cur.last_price = p.last_price;
3048 cur.pnl = p.pnl;
3049 cur.profit_rate = p.profit_rate;
3050 cur.name.clone_from(&p.name);
3051 }
3052 None => {
3053 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3054 book.insert(key, (seq, p.clone()));
3055 }
3056 }
3057 }
3058 self.trade_dirty.store(true, Ordering::Release);
3059 self.bump();
3060 }
3061
3062 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
3064 let key = a.en_accountid();
3065 {
3066 let mut book = self.trade_funds.lock().unwrap();
3067 match book.get_mut(&key) {
3068 Some((_, cur)) => {
3069 cur.balance = a.balance;
3070 cur.frozen = a.frozen;
3071 }
3072 None => {
3073 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3074 book.insert(key, (seq, a.clone()));
3075 }
3076 }
3077 }
3078 self.trade_dirty.store(true, Ordering::Release);
3079 self.bump();
3080 }
3081
3082 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
3084 let mut removed = 0usize;
3085 {
3086 let mut book = self.trade_orders.lock().unwrap();
3087 let before = book.len();
3088 book.retain(|_, (_, o)| o.gateway_name != gateway);
3089 removed += before - book.len();
3090 }
3091 {
3092 let mut book = self.trade_trades.lock().unwrap();
3093 let before = book.len();
3094 book.retain(|(_, t)| t.gateway_name != gateway);
3095 removed += before - book.len();
3096 let prefix = format!("{gateway}.");
3099 self.trade_ids
3100 .lock()
3101 .unwrap()
3102 .retain(|id| !id.starts_with(&prefix));
3103 }
3104 {
3105 let mut book = self.trade_positions.lock().unwrap();
3106 let before = book.len();
3107 book.retain(|_, (_, p)| p.gateway_name != gateway);
3108 removed += before - book.len();
3109 }
3110 {
3111 let mut book = self.trade_funds.lock().unwrap();
3112 let before = book.len();
3113 book.retain(|_, (_, a)| a.gateway_name != gateway);
3114 removed += before - book.len();
3115 }
3116 if removed > 0 {
3117 self.trade_dirty.store(true, Ordering::Release);
3118 self.bump();
3119 }
3120 removed
3121 }
3122
3123 #[must_use]
3125 pub fn trade_dropped(&self) -> u64 {
3126 self.trade_dropped.load(Ordering::Acquire)
3127 }
3128
3129 pub fn trade_snap(&self) -> Arc<TradeSnap> {
3134 let mut cache = self.trade_snap.lock().unwrap();
3135 if self.trade_dirty.swap(false, Ordering::AcqRel) {
3136 *cache = Arc::new(self.build_trade_snap());
3137 }
3138 Arc::clone(&cache)
3139 }
3140
3141 fn build_trade_snap(&self) -> TradeSnap {
3144 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
3145 where
3146 T: Clone,
3147 {
3148 let mut v: Vec<(u64, &String, &T)> =
3149 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
3150 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
3151 v.into_iter().map(|(_, _, v)| v.clone()).collect()
3152 }
3153 TradeSnap {
3154 orders: by_seq(&self.trade_orders.lock().unwrap()),
3155 trades: self
3156 .trade_trades
3157 .lock()
3158 .unwrap()
3159 .iter()
3160 .map(|(_, t)| t.clone())
3161 .collect(),
3162 positions: by_seq(&self.trade_positions.lock().unwrap()),
3163 accounts: by_seq(&self.trade_funds.lock().unwrap()),
3164 }
3165 }
3166
3167 pub fn ctp_form(&self) -> CtpForm {
3171 self.ctp_form.lock().unwrap().clone()
3172 }
3173
3174 pub fn set_ctp_form(&self, f: CtpForm) {
3176 *self.ctp_form.lock().unwrap() = f;
3177 self.bump();
3178 }
3179
3180 pub fn row_selected(&self, table: &str, key: &str) -> bool {
3184 self.row_sel
3185 .lock()
3186 .unwrap()
3187 .get(table)
3188 .is_some_and(|k| k == key)
3189 }
3190
3191 pub fn select_row(&self, table: &str, key: &str) {
3193 self.row_sel
3194 .lock()
3195 .unwrap()
3196 .insert(table.to_string(), key.to_string());
3197 }
3198}