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 counts(&self) -> (u64, u64, u64) {
422 (
423 self.frames.load(Ordering::Acquire),
424 self.updates.load(Ordering::Acquire),
425 self.reloads.load(Ordering::Acquire),
426 )
427 }
428}
429
430#[derive(Debug, Default, Clone, PartialEq)]
433pub struct ReloadForm {
434 pub interval: String,
435 pub symbol: String,
436 pub csv: String,
437}
438
439pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
446 let g = raw.trim();
447 if !g.is_empty() {
448 if g.chars().any(|c| c.is_whitespace()) {
449 return Err(format!("网关解析: 含空白:{raw:?}"));
450 }
451 return Ok(g.to_string());
452 }
453 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
454 Some(f) => Ok(f.to_string()),
455 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
456 }
457}
458
459#[derive(Debug, Clone, PartialEq)]
462pub struct CtpForm {
463 pub broker_id: String,
464 pub td_addr: String,
465 pub td_user: String,
466 pub td_password: String,
467 pub td_auth_code: String,
468 pub td_app_id: String,
469 pub md_addr: String,
470 pub md_user: String,
471 pub md_password: String,
472 pub symbols: String,
474 pub accept_futures: bool,
477 pub accept_options: bool,
478}
479
480impl Default for CtpForm {
483 fn default() -> Self {
484 Self {
485 broker_id: String::new(),
486 td_addr: String::new(),
487 td_user: String::new(),
488 td_password: String::new(),
489 td_auth_code: String::new(),
490 td_app_id: String::new(),
491 md_addr: String::new(),
492 md_user: String::new(),
493 md_password: String::new(),
494 symbols: String::new(),
495 accept_futures: true,
496 accept_options: false,
497 }
498 }
499}
500
501impl CtpForm {
502 #[must_use]
506 pub fn product_classes(&self) -> Vec<String> {
507 let mut codes = Vec::new();
508 if self.accept_futures {
509 codes.push("1".to_string());
510 }
511 if self.accept_options {
512 codes.push("2".to_string());
513 }
514 codes
515 }
516
517 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
519 ek_ctp::plugin::ConfigureReq {
520 broker_id: self.broker_id.trim().to_string(),
521 td_addr: self.td_addr.trim().to_string(),
522 td_user: self.td_user.trim().to_string(),
523 td_password: self.td_password.clone(),
524 td_auth_code: self.td_auth_code.trim().to_string(),
525 td_app_id: self.td_app_id.trim().to_string(),
526 md_addr: self.md_addr.trim().to_string(),
527 md_user: self.md_user.trim().to_string(),
528 md_password: self.md_password.clone(),
529 product_classes: self.product_classes(),
530 }
531 }
532
533 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
535 let list: Vec<String> = self
536 .symbols
537 .split([',', ',', ' '])
538 .map(|s| s.trim().to_string())
539 .filter(|s| !s.is_empty())
540 .collect();
541 if list.is_empty() {
542 return Err("订阅解析: 标的位空".to_string());
543 }
544 Ok(list)
545 }
546
547 pub fn validate(&self) -> Result<(), String> {
556 fn front(name: &str, v: &str) -> Result<(), String> {
557 let v = v.trim();
558 let Some(rest) = v.strip_prefix("tcp://") else {
559 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
560 };
561 let Some((host, port)) = rest.rsplit_once(':') else {
562 return Err(format!("{name} 缺端口:{name} {:?}", v));
563 };
564 if host.trim().is_empty() {
565 return Err(format!("{name} 主机位空:{}", v));
566 }
567 let port: u32 = port
568 .parse()
569 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
570 if port == 0 || port > 65535 {
571 return Err(format!("{name} 端口越界(1..=65535):{}", v));
572 }
573 Ok(())
574 }
575 if self.broker_id.trim().is_empty() {
576 return Err("broker 位空".to_string());
577 }
578 if self.td_user.trim().is_empty() {
579 return Err("交易账号位空".to_string());
580 }
581 front("td 前置", &self.td_addr)?;
582 if !self.md_addr.trim().is_empty() {
583 front("md 前置", &self.md_addr)?;
584 }
585 if self.td_app_id.trim().is_empty() {
586 return Err("AppID 位空(CTP 登录必需)".to_string());
587 }
588 if self.td_auth_code.trim().is_empty() {
589 return Err("AuthCode 位空(CTP 登录必需)".to_string());
590 }
591 Ok(())
592 }
593}
594
595#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
602pub enum AccountStage {
603 #[default]
605 NotAssembled,
606 Assembling,
608 Assembled,
610 AssembleFailed,
612}
613
614impl AccountStage {
615 #[must_use]
617 pub fn label(self) -> &'static str {
618 match self {
619 AccountStage::NotAssembled => "未装配",
620 AccountStage::Assembling => "装配中…",
621 AccountStage::Assembled => "已装配·登录态未知",
624 AccountStage::AssembleFailed => "装配失败",
625 }
626 }
627}
628
629#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
634pub enum LoginState {
635 #[default]
640 Unknown,
641 Online,
643 LoginFailed,
645 FrontDown,
647 QueryStarved,
649 Stopped,
651}
652
653impl LoginState {
654 #[must_use]
656 pub fn label(self) -> &'static str {
657 match self {
658 LoginState::Unknown => "登录态未知",
659 LoginState::Online => "在线",
660 LoginState::LoginFailed => "登录失败",
661 LoginState::FrontDown => "掉线",
662 LoginState::QueryStarved => "查询断供",
663 LoginState::Stopped => "已断开",
664 }
665 }
666
667 #[must_use]
669 pub fn is_alarm(self) -> bool {
670 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
671 }
672}
673
674fn count_note(label: &str, n: u32) -> Option<String> {
676 (n > 0).then(|| format!("{label} {n} 次"))
677}
678
679fn brief_counter_reason(full: &str) -> Option<String> {
686 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
687 let body = tail.trim().trim_start_matches("CTP:").trim();
688 if body.is_empty() {
689 return None;
690 }
691 let code = full
692 .rfind("代码:")
693 .map(|i| {
694 let rest = &full[i + "代码:".len()..];
695 let end = rest.find(',').unwrap_or(rest.len());
696 rest[..end].trim().to_string()
697 })
698 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
699 Some(match code {
700 Some(c) => format!("{body}({c})"),
701 None => body.to_string(),
702 })
703}
704
705#[derive(Debug, Clone, PartialEq, Default)]
711pub struct AccountRow {
712 pub gateway: String,
714 pub alias: String,
716 pub broker_id: String,
717 pub account: String,
718 pub md_source: bool,
720 pub form: CtpForm,
722 pub restored: bool,
728 pub auto_connect: bool,
733 pub password_ref: bool,
738 pub stage: AccountStage,
740 pub fail_note: Option<String>,
742 pub effective_classes: Vec<String>,
747 pub connected: bool,
750 pub login: LoginState,
754 pub status_note: Option<String>,
757 pub login_brief: Option<String>,
764 pub chain_note: Option<String>,
772}
773
774impl AccountRow {
775 pub fn derive(form: &CtpForm) -> Result<Self, String> {
780 let broker = form.broker_id.trim();
781 let account = form.td_user.trim();
782 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
783 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
784 })?;
785 Ok(Self {
786 gateway,
787 alias: String::new(),
788 broker_id: broker.to_string(),
789 account: account.to_string(),
790 md_source: false,
791 form: form.clone(),
792 stage: AccountStage::NotAssembled,
793 fail_note: None,
794 effective_classes: Vec::new(),
795 connected: false,
796 login: LoginState::Unknown,
797 status_note: None,
798 login_brief: None,
799 chain_note: None,
800 restored: false,
803 auto_connect: false,
804 password_ref: false,
805 })
806 }
807
808 #[must_use]
814 pub fn status_label(&self) -> String {
815 if self.stage == AccountStage::Assembled {
816 match self.login {
817 LoginState::Unknown => "已装配·登录态未知".to_string(),
818 other => format!("已装配·{}", other.label()),
819 }
820 } else {
821 self.stage.label().to_string()
822 }
823 }
824
825 pub fn apply_status(
838 &mut self,
839 st: &ek_ctp::plugin::GatewayStatus,
840 now_ms: i64,
841 reject_limit: u32,
842 stale_secs: i64,
843 ) {
844 let td_side = !self.form.td_addr.trim().is_empty();
845 let (login, note) = if td_side {
846 if st.td_login_failed && !st.td_logged_in {
847 (
848 LoginState::LoginFailed,
849 st.td_last_error
850 .clone()
851 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
852 )
853 } else if st.td_logged_in {
854 if st.td_query_reject_streak >= reject_limit {
855 (
856 LoginState::QueryStarved,
857 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
858 )
859 } else if let Some(last) = st.td_last_account_ms {
860 let idle_ms = now_ms.saturating_sub(last);
861 if idle_ms > stale_secs.saturating_mul(1_000) {
862 (
863 LoginState::QueryStarved,
864 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
865 )
866 } else {
867 (LoginState::Online, count_note("掉线", st.td_disconnects))
868 }
869 } else {
870 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
877 match since_login_ms {
878 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
879 LoginState::QueryStarved,
880 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
881 ),
882 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
883 }
884 }
885 } else if self.login == LoginState::Stopped {
886 let extra = if st.td_disconnects > 0 {
887 format!("(此前掉线 {} 次)", st.td_disconnects)
888 } else {
889 String::new()
890 };
891 (
892 LoginState::Stopped,
893 Some(format!("会话已停{extra};『连接』可恢复")),
894 )
895 } else if st.td_disconnects > 0 {
896 (
897 LoginState::FrontDown,
898 Some(format!("掉线 {} 次", st.td_disconnects)),
899 )
900 } else {
901 (LoginState::Unknown, st.td_last_error.clone())
902 }
903 } else if st.md_logged_in {
904 (
905 LoginState::Online,
906 count_note("掉线", st.md_disconnects)
907 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
908 )
909 } else if self.login == LoginState::Stopped {
910 (
911 LoginState::Stopped,
912 Some("会话已停;『连接』可恢复".to_string()),
913 )
914 } else if st.md_disconnects > 0 {
915 (
916 LoginState::FrontDown,
917 Some(format!("掉线 {} 次", st.md_disconnects)),
918 )
919 } else {
920 (LoginState::Unknown, st.md_last_error.clone())
921 };
922 let note = if td_side && st.td_instrument_rejects > 0 {
927 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
928 match note {
929 Some(base) if base.contains("合约查询被拒") => Some(base),
931 Some(base) => Some(format!("{base};{add}")),
932 None => Some(add),
933 }
934 } else {
935 note
936 };
937 self.login = login;
938 self.status_note = note;
939 self.login_brief = if login.is_alarm() {
942 st.td_last_error
943 .as_deref()
944 .or(st.md_last_error.as_deref())
945 .and_then(brief_counter_reason)
946 } else {
947 None
948 };
949 }
950
951 #[must_use]
953 pub fn display_name(&self) -> String {
954 let a = self.alias.trim();
955 if a.is_empty() {
956 self.account.clone()
957 } else {
958 a.to_string()
959 }
960 }
961}
962
963#[derive(Debug, Clone, PartialEq, Default)]
966pub struct TradeSnap {
967 pub orders: Vec<ek_data::order::OrderData>,
968 pub trades: Vec<ek_data::trade::TradeData>,
969 pub positions: Vec<ek_data::position::PositionData>,
970 pub accounts: Vec<ek_data::account::AccountData>,
972}
973
974pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
977 let interval: i64 = form
978 .interval
979 .trim()
980 .parse()
981 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
982 let symbol = form.symbol.trim().to_string();
983 if symbol.is_empty() {
984 return Err("回灌解析: symbol 空".to_string());
985 }
986 if form.csv.trim().is_empty() {
987 return Err("回灌解析: csv 空".to_string());
988 }
989 let mut bars = Vec::new();
990 for (i, raw) in form.csv.lines().enumerate() {
991 let line = raw.trim();
992 if line.is_empty() || line.starts_with('#') {
993 continue;
994 }
995 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
996 if parts.len() != 6 {
997 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
998 }
999 let num = |s: &str| -> Result<f64, String> {
1000 s.parse::<f64>()
1001 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
1002 };
1003 let dt_s: i64 = parts[0]
1004 .parse()
1005 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
1006 bars.push(BarData {
1007 symbol: symbol.clone(),
1008 datetime: dt_s * 1_000_000_000,
1009 interval: Some(
1010 ek_dm::wire::interval_of_seconds(interval)
1011 .unwrap_or(ek_data::constant::Interval::MINUTE),
1012 ),
1013 open_price: num(parts[1])?,
1014 high_price: num(parts[2])?,
1015 low_price: num(parts[3])?,
1016 close_price: num(parts[4])?,
1017 volume: num(parts[5])?,
1018 ..BarData::default()
1019 });
1020 }
1021 if bars.is_empty() {
1022 return Err("回灌解析: 无有效行".to_string());
1023 }
1024 Ok((interval, symbol, bars))
1025}
1026
1027#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1032pub enum TopLeftTab {
1033 #[default]
1034 Orders,
1035 Active,
1036}
1037
1038impl TopLeftTab {
1039 pub fn label(self) -> &'static str {
1040 match self {
1041 Self::Orders => "委托",
1042 Self::Active => "未成交",
1043 }
1044 }
1045}
1046
1047#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1049pub enum BottomRightTab {
1050 #[default]
1051 Account,
1052 Logs,
1053}
1054
1055impl BottomRightTab {
1056 pub fn label(self) -> &'static str {
1057 match self {
1058 Self::Account => "资金",
1059 Self::Logs => "日志",
1060 }
1061 }
1062}
1063
1064#[derive(Debug, Clone, Copy, PartialEq)]
1067pub struct DataSplit {
1068 pub col: f32,
1070 pub row: f32,
1072}
1073
1074impl Default for DataSplit {
1075 fn default() -> Self {
1076 Self {
1077 col: crate::splitters::DATA_SPLIT_DEFAULT,
1078 row: crate::splitters::DATA_SPLIT_DEFAULT,
1079 }
1080 }
1081}
1082
1083pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
1086 ek_data::constant::Exchange::SHFE,
1087 ek_data::constant::Exchange::DCE,
1088 ek_data::constant::Exchange::CZCE,
1089 ek_data::constant::Exchange::CFFEX,
1090 ek_data::constant::Exchange::INE,
1091 ek_data::constant::Exchange::GFEX,
1092];
1093
1094#[derive(Debug, Default)]
1104pub struct TickStore {
1105 ticks: BTreeMap<String, ek_data::tick::TickData>,
1106}
1107
1108impl TickStore {
1109 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1112 self.ticks
1113 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1114 }
1115
1116 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1118 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1119 v.sort_by(|a, b| {
1120 (a.exchange.as_str().to_string(), a.symbol.clone())
1121 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1122 });
1123 v
1124 }
1125
1126 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1128 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1129 for t in self.rows() {
1130 let ex = t.exchange.as_str().to_string();
1131 match out.last_mut() {
1132 Some((e, v)) if *e == ex => v.push(t),
1133 _ => out.push((ex, vec![t])),
1134 }
1135 }
1136 out
1137 }
1138
1139 pub fn len(&self) -> usize {
1140 self.ticks.len()
1141 }
1142
1143 pub fn is_empty(&self) -> bool {
1144 self.ticks.is_empty()
1145 }
1146}
1147
1148#[derive(Debug, Clone, PartialEq)]
1150pub struct OrderForm {
1151 pub exchange: String,
1153 pub symbol: String,
1155 pub direction: ek_data::constant::Direction,
1156 pub offset: ek_data::constant::Offset,
1157 pub order_type: ek_data::constant::OrderType,
1158 pub price: String,
1160 pub volume: String,
1161 pub gateway: String,
1163 pub follow_price: bool,
1165 pub auto_split: bool,
1168}
1169
1170impl Default for OrderForm {
1171 fn default() -> Self {
1172 Self {
1173 exchange: "SHFE".to_string(),
1174 symbol: String::new(),
1175 direction: ek_data::constant::Direction::LONG,
1176 offset: ek_data::constant::Offset::OPEN,
1177 order_type: ek_data::constant::OrderType::LIMIT,
1178 price: String::new(),
1179 volume: "1".to_string(),
1180 gateway: String::new(),
1181 follow_price: true,
1182 auto_split: true,
1183 }
1184 }
1185}
1186
1187impl OrderForm {
1188 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1190 use std::str::FromStr;
1191 let symbol = self.symbol.trim();
1192 if symbol.is_empty() {
1193 return Err("下单: 代码为空".to_string());
1194 }
1195 if symbol.contains('.') {
1196 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1197 }
1198 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1199 .map_err(|e| format!("下单: {e}"))?;
1200 let price: f64 = self
1201 .price
1202 .trim()
1203 .parse()
1204 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1205 let volume: f64 = self
1206 .volume
1207 .trim()
1208 .parse()
1209 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1210 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1211 return Err("下单: 限价单价格须为正".to_string());
1212 }
1213 if volume <= 0.0 {
1214 return Err("下单: 数量须为正".to_string());
1215 }
1216 Ok(ek_data::order_request::OrderRequest {
1217 symbol: symbol.to_string(),
1218 exchange,
1219 direction: self.direction,
1220 order_type: self.order_type,
1221 volume,
1222 price,
1223 offset: self.offset,
1224 ..ek_data::order_request::OrderRequest::default()
1225 })
1226 }
1227}
1228
1229#[derive(Debug, Default)]
1235pub struct ContractStore {
1236 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1237}
1238
1239impl ContractStore {
1240 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1241 self.contracts
1242 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1243 }
1244
1245 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1247 let p = pattern.trim();
1248 self.contracts
1249 .values()
1250 .filter(|c| {
1251 p.is_empty()
1252 || c.symbol.contains(p)
1253 || c.exchange.as_str().eq_ignore_ascii_case(p)
1254 || c.name.contains(p)
1255 })
1256 .cloned()
1257 .collect()
1258 }
1259
1260 pub fn len(&self) -> usize {
1261 self.contracts.len()
1262 }
1263
1264 pub fn is_empty(&self) -> bool {
1265 self.contracts.is_empty()
1266 }
1267}
1268
1269#[derive(Debug, Default)]
1271pub struct UiShared {
1272 logs: Mutex<LogRing>,
1273 snap: Mutex<Arc<BusSnap>>,
1274 market: MarketStore,
1275 ctp_form: Mutex<CtpForm>,
1276 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1279 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1280 trade_ids: Mutex<HashSet<String>>,
1288 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1289 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1290 trade_seq: AtomicU64,
1292 trade_dropped: AtomicU64,
1294 trade_dirty: AtomicBool,
1296 trade_snap: Mutex<Arc<TradeSnap>>,
1298 tick_store: Mutex<TickStore>,
1299 order_form: Mutex<OrderForm>,
1300 data_split: Mutex<DataSplit>,
1303 selected_exchange: Mutex<String>,
1305 topleft_tab: Mutex<TopLeftTab>,
1307 bottomright_tab: Mutex<BottomRightTab>,
1309 ops_window: AtomicBool,
1310 accounts_window: AtomicBool,
1313 contracts_window: AtomicBool,
1314 contract_store: Mutex<ContractStore>,
1315 contract_query: Mutex<String>,
1316 contract_selected: Mutex<Vec<String>>,
1317 kline_windows: Mutex<Vec<String>>,
1318 kline_fetches: Mutex<Vec<KlineFetch>>,
1320 kline_fetched: Mutex<HashSet<(String, i64)>>,
1322 rev: AtomicU64,
1323 stop: AtomicBool,
1325 cmds: Mutex<VecDeque<UiCmd>>,
1327 pending_confirm: Mutex<Option<String>>,
1329 accounts: Mutex<Vec<AccountRow>>,
1331 account_edit_window: AtomicBool,
1333 editing_account: Mutex<Option<String>>,
1335 edit_alias: Mutex<String>,
1337 edit_error: Mutex<Option<String>>,
1339 md_source: Mutex<String>,
1341 subscriptions: Mutex<Vec<String>>,
1343 row_sel: Mutex<BTreeMap<String, String>>,
1346 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1350 status_cursor: AtomicUsize,
1356 status_seed: Mutex<Vec<String>>,
1359 list_tick: AtomicU32,
1361 list_due: AtomicBool,
1365 seed_due: AtomicBool,
1369 pipeline_due: AtomicBool,
1371 cfg_due: AtomicBool,
1373 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1376 cfg_delete_arm: Mutex<Option<String>>,
1378 theme: AtomicU8,
1382 theme_persist_due: AtomicBool,
1385 auto_connect_persist_due: AtomicBool,
1389 bad_payload: Mutex<BTreeMap<String, u64>>,
1391 pipeline_lines: Mutex<Vec<String>>,
1394 status_thresholds: Mutex<Option<(u32, i64)>>,
1397 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1399 toasts: crate::toast::ToastsCell,
1402 alerted: Mutex<std::collections::BTreeSet<String>>,
1404 exit_confirm_open: AtomicBool,
1406 exit_confirmed: AtomicBool,
1408 pluginwin_open: AtomicBool,
1410 bundle_open: AtomicBool,
1412 assembly_locked: AtomicBool,
1415}
1416
1417impl UiShared {
1418 pub fn new() -> Self {
1420 Self::default()
1421 }
1422
1423 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1427 self.tick_store.lock().unwrap().put(t);
1428 self.bump();
1429 }
1430
1431 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1433 self.tick_store.lock().unwrap().rows()
1434 }
1435
1436 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1438 self.tick_store.lock().unwrap().groups()
1439 }
1440
1441 pub fn tick_count(&self) -> usize {
1443 self.tick_store.lock().unwrap().len()
1444 }
1445
1446 pub fn order_form(&self) -> OrderForm {
1448 self.order_form.lock().unwrap().clone()
1449 }
1450
1451 pub fn set_order_form(&self, f: OrderForm) {
1453 *self.order_form.lock().unwrap() = f;
1454 self.bump();
1455 }
1456
1457 pub fn selected_exchange(&self) -> String {
1461 let s = self.selected_exchange.lock().unwrap().clone();
1462 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1463 s
1464 } else {
1465 MARKET_ORDER[0].as_str().to_string()
1466 }
1467 }
1468
1469 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1471 *self.selected_exchange.lock().unwrap() = s.into();
1472 self.bump();
1473 }
1474
1475 pub fn topleft_tab(&self) -> TopLeftTab {
1477 *self.topleft_tab.lock().unwrap()
1478 }
1479
1480 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1482 *self.topleft_tab.lock().unwrap() = t;
1483 self.bump();
1484 }
1485
1486 pub fn bottomright_tab(&self) -> BottomRightTab {
1488 *self.bottomright_tab.lock().unwrap()
1489 }
1490
1491 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1493 *self.bottomright_tab.lock().unwrap() = t;
1494 self.bump();
1495 }
1496
1497 pub fn data_split(&self) -> DataSplit {
1499 *self.data_split.lock().unwrap()
1500 }
1501
1502 pub fn set_data_split(&self, s: DataSplit) {
1505 let s = DataSplit {
1506 col: crate::splitters::clamp_frac(s.col),
1507 row: crate::splitters::clamp_frac(s.row),
1508 };
1509 let mut g = self.data_split.lock().unwrap();
1510 if g.col != s.col || g.row != s.row {
1511 *g = s;
1512 self.bump();
1513 }
1514 }
1515
1516 pub fn reset_layout(&self) {
1520 *self.data_split.lock().unwrap() = DataSplit::default();
1521 *self.selected_exchange.lock().unwrap() = String::new();
1522 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1523 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1524 self.bump();
1525 }
1526
1527 pub fn ops_window_open(&self) -> bool {
1531 self.ops_window.load(Ordering::Acquire)
1532 }
1533
1534 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1537 self.table_sorts
1538 .lock()
1539 .unwrap()
1540 .get(table)
1541 .copied()
1542 .flatten()
1543 }
1544
1545 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1547 let mut m = self.table_sorts.lock().unwrap();
1548 let asc = match m.get(table).copied().flatten() {
1549 Some((c, a)) if c == col => !a,
1550 _ => true,
1551 };
1552 m.insert(table.to_string(), Some((col, asc)));
1553 }
1554
1555 pub fn set_ops_window_open(&self, v: bool) {
1556 self.ops_window.store(v, Ordering::Release);
1557 self.bump();
1558 }
1559
1560 pub fn accounts_window_open(&self) -> bool {
1564 self.accounts_window.load(Ordering::Acquire)
1565 }
1566
1567 pub fn set_accounts_window_open(&self, v: bool) {
1569 self.accounts_window.store(v, Ordering::Release);
1570 self.bump();
1571 }
1572
1573 pub fn account_edit_open(&self) -> bool {
1575 self.account_edit_window.load(Ordering::Acquire)
1576 }
1577
1578 pub fn editing_account(&self) -> Option<String> {
1580 self.editing_account.lock().unwrap().clone()
1581 }
1582
1583 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1587 match gateway {
1588 Some(gw) => {
1589 let Some(row) = self.account(gw) else {
1590 return false;
1591 };
1592 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1593 *self.edit_alias.lock().unwrap() = row.alias.clone();
1594 self.set_ctp_form(row.form.clone());
1595 }
1596 None => {
1597 *self.editing_account.lock().unwrap() = None;
1598 *self.edit_alias.lock().unwrap() = String::new();
1599 let draft = self.draft_from_existing();
1600 self.set_ctp_form(draft);
1601 }
1602 }
1603 *self.edit_error.lock().unwrap() = None;
1604 self.account_edit_window.store(true, Ordering::Release);
1605 self.bump();
1606 true
1607 }
1608
1609 pub fn end_account_edit(&self) {
1611 *self.editing_account.lock().unwrap() = None;
1612 *self.edit_alias.lock().unwrap() = String::new();
1613 *self.edit_error.lock().unwrap() = None;
1614 self.account_edit_window.store(false, Ordering::Release);
1615 let draft = self.draft_from_existing();
1616 self.set_ctp_form(draft);
1617 self.bump();
1618 }
1619
1620 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1626 if self.editing_account().as_deref() == Some(gateway) {
1627 self.end_account_edit();
1628 }
1629 }
1630
1631 pub fn edit_alias(&self) -> String {
1633 self.edit_alias.lock().unwrap().clone()
1634 }
1635
1636 pub fn set_edit_alias(&self, v: impl Into<String>) {
1638 *self.edit_alias.lock().unwrap() = v.into();
1639 self.bump();
1640 }
1641
1642 pub fn edit_error(&self) -> Option<String> {
1644 self.edit_error.lock().unwrap().clone()
1645 }
1646
1647 pub fn set_edit_error(&self, e: Option<String>) {
1649 *self.edit_error.lock().unwrap() = e;
1650 self.bump();
1651 }
1652
1653 pub fn account_edit_dirty(&self) -> bool {
1655 let editing = self.editing_account();
1656 let draft = self.ctp_form();
1657 match editing {
1658 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1659 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1660 }
1661 }
1662
1663 pub fn accounts(&self) -> Vec<AccountRow> {
1665 self.accounts.lock().unwrap().clone()
1666 }
1667
1668 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1670 self.accounts
1671 .lock()
1672 .unwrap()
1673 .iter()
1674 .find(|a| a.gateway == gateway)
1675 .cloned()
1676 }
1677
1678 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1684 let mut acc = self.accounts.lock().unwrap();
1685 if acc.iter().any(|a| a.gateway == row.gateway) {
1686 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1687 }
1688 let alias = row.alias.trim();
1689 if !alias.is_empty()
1690 && acc
1691 .iter()
1692 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1693 {
1694 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1695 }
1696 if acc.len() >= MAX_ACCOUNTS {
1697 return Err(format!(
1698 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1699 ));
1700 }
1701 acc.push(row);
1702 drop(acc);
1703 self.bump();
1704 Ok(())
1705 }
1706
1707 #[must_use]
1709 pub fn can_add_account(&self) -> bool {
1710 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1711 }
1712
1713 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1719 let alias = alias.trim().to_string();
1720 let mut acc = self.accounts.lock().unwrap();
1721 if !alias.is_empty()
1722 && acc
1723 .iter()
1724 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1725 {
1726 return false;
1727 }
1728 match acc.iter_mut().find(|a| a.gateway == gateway) {
1729 Some(a) => {
1730 a.alias = alias;
1731 drop(acc);
1732 self.bump();
1733 true
1734 }
1735 None => false,
1736 }
1737 }
1738
1739 pub fn note_connect_sent(&self, gateway: &str) {
1744 let mut acc = self.accounts.lock().unwrap();
1745 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1746 a.stage = AccountStage::Assembling;
1747 a.fail_note = None;
1748 a.login = LoginState::Unknown;
1749 a.status_note = None;
1750 a.login_brief = None;
1751 a.chain_note = None;
1753 drop(acc);
1754 self.bump();
1755 }
1756 }
1757
1758 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1762 let mut acc = self.accounts.lock().unwrap();
1763 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1764 match result {
1765 Ok(()) => {
1766 a.stage = AccountStage::Assembled;
1767 a.fail_note = None;
1768 }
1769 Err(e) => {
1770 a.stage = AccountStage::AssembleFailed;
1771 a.fail_note = Some(e);
1772 }
1773 }
1774 drop(acc);
1775 self.bump();
1776 }
1777 }
1778
1779 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1783 let mut acc = self.accounts.lock().unwrap();
1784 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1785 a.effective_classes = classes.to_vec();
1786 drop(acc);
1787 self.bump();
1788 }
1789 }
1790
1791 pub fn note_chain(&self, gateway: &str, note: &str) {
1796 let mut acc = self.accounts.lock().unwrap();
1797 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1798 if a.chain_note.as_deref() == Some(note) {
1799 return; }
1801 a.chain_note = Some(note.to_string());
1802 drop(acc);
1803 self.bump();
1804 }
1805 }
1806
1807 #[must_use]
1811 pub fn status_thresholds(&self) -> (u32, i64) {
1812 self.status_thresholds
1813 .lock()
1814 .unwrap()
1815 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1816 }
1817
1818 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1820 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1821 self.bump();
1822 }
1823
1824 pub fn list_poll_due(&self) -> bool {
1829 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1830 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1831 }
1832
1833 pub fn request_list_refresh(&self) {
1835 self.list_due.store(true, Ordering::Release);
1836 }
1837
1838 pub fn request_trade_reseed(&self) {
1840 self.seed_due.store(true, Ordering::Release);
1841 }
1842
1843 pub fn take_trade_reseed(&self) -> bool {
1845 self.seed_due.swap(false, Ordering::AcqRel)
1846 }
1847
1848 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1853 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1857 for gw in present {
1858 if !uniq.iter().any(|u| u == gw) {
1859 uniq.push(gw.clone());
1860 }
1861 }
1862 let n = uniq.len();
1863 if n == 0 {
1864 return Vec::new();
1865 }
1866 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1867 let take = STATUS_ASKS_PER_TICK.min(n);
1868 let mut out = Vec::with_capacity(take);
1869 for i in 0..take {
1870 out.push(uniq[(start + i) % n].clone());
1871 }
1872 out
1873 }
1874
1875 pub fn request_status_seed(&self, gateway: &str) {
1877 let mut q = self.status_seed.lock().unwrap();
1878 if !q.iter().any(|g| g == gateway) {
1879 q.push(gateway.to_string());
1880 }
1881 }
1882
1883 pub fn take_status_seeds(&self) -> Vec<String> {
1885 let mut q = self.status_seed.lock().unwrap();
1886 std::mem::take(&mut *q)
1887 }
1888
1889 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
1899 let gw = ev.gateway.as_str();
1900 let text = ev.describe();
1901 self.log_line(format!("[INFO][{gw}] {text}"));
1902 let note = text
1903 .strip_prefix(CHAIN_STATE_PREFIX)
1904 .map(|s| s.trim().to_string())
1905 .unwrap_or_else(|| text.clone());
1906 self.note_chain(gw, ¬e);
1907 self.note_gateway_status(gw, &ev.status);
1908 let alarm_kind = matches!(
1909 ev.change,
1910 ek_ctp::GatewayChange::TdDisconnected
1911 | ek_ctp::GatewayChange::TdAuthFailed
1912 | ek_ctp::GatewayChange::TdLoginFailed
1913 | ek_ctp::GatewayChange::ContractQueryStopped
1914 | ek_ctp::GatewayChange::QueryRejected
1915 | ek_ctp::GatewayChange::MdDisconnected
1916 | ek_ctp::GatewayChange::MdLoginFailed
1917 | ek_ctp::GatewayChange::SubscribeRejected
1918 );
1919 if alarm_kind {
1920 let session = if ev.change.kind().starts_with("md-") {
1922 ev.status.md_disconnects
1923 } else {
1924 ev.status.td_disconnects
1925 };
1926 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
1927 if self.alert_once(key) {
1928 self.push_alert(
1929 crate::toast::AlertLevel::Warning,
1930 format!("账户 {gw}:{}", note.trim()),
1931 );
1932 }
1933 }
1934 }
1935
1936 pub fn request_pipeline_refresh(&self) {
1938 self.pipeline_due.store(true, Ordering::Release);
1939 }
1940 pub fn take_pipeline_due(&self) -> bool {
1942 self.pipeline_due.swap(false, Ordering::AcqRel)
1943 }
1944
1945 pub fn request_cfg_refresh(&self) {
1947 self.cfg_due.store(true, Ordering::Release);
1948 }
1949
1950 pub fn take_cfg_due(&self) -> bool {
1952 self.cfg_due.swap(false, Ordering::AcqRel)
1953 }
1954
1955 pub fn theme(&self) -> crate::tokens::Theme {
1957 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
1958 }
1959
1960 pub fn set_theme(&self, theme: crate::tokens::Theme) {
1963 self.theme.store(theme.as_u8(), Ordering::Release);
1964 self.bump();
1965 }
1966
1967 pub fn toggle_theme(&self) -> crate::tokens::Theme {
1969 let next = self.theme().flipped();
1970 self.theme.store(next.as_u8(), Ordering::Release);
1971 self.theme_persist_due.store(true, Ordering::Release);
1972 self.bump();
1973 next
1974 }
1975
1976 pub fn take_theme_persist_due(&self) -> bool {
1978 self.theme_persist_due.swap(false, Ordering::AcqRel)
1979 }
1980
1981 pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
1985 {
1986 let mut acc = self.accounts.lock().unwrap();
1987 match acc.iter_mut().find(|a| a.gateway == gateway) {
1988 Some(a) => {
1989 if a.auto_connect != on {
1990 a.auto_connect = on;
1991 self.auto_connect_persist_due.store(true, Ordering::Release);
1992 self.bump();
1993 }
1994 }
1995 None => return Err(format!("账户 {gateway} 不在清册")),
1996 }
1997 }
1998 Ok(())
1999 }
2000
2001 pub fn take_auto_connect_persist_due(&self) -> bool {
2003 self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
2004 }
2005
2006 pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
2021 let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
2023 match store.load(crate::plugin::THEME_NS) {
2024 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
2025 for (k, _f) in map.iter() {
2026 if let Some(gw) = k.strip_prefix("auto_connect.") {
2027 flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
2028 }
2029 }
2030 }
2031 Ok(_) => {} Err(_) => {} }
2034 let mut restored: Vec<String> = Vec::new();
2035 let ns_list = match store.list_namespaces() {
2036 Ok(v) => v,
2037 Err(_) => {
2038 let msg = "[ui] 档案清点失败:存储根不可读(具体见运行日志)".to_string();
2039 self.log_line(msg.clone());
2040 return restored;
2041 }
2042 };
2043 for ns in ns_list {
2044 let Some(gw) = ns.strip_prefix("ctp.") else {
2045 continue;
2046 };
2047 if self.account(gw).is_some() {
2048 self.log_line(format!(
2049 "[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
2050 ));
2051 continue;
2052 }
2053 let map = match store.load(&ns) {
2054 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
2055 Ok(_) => {
2056 self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
2057 continue;
2058 }
2059 Err(e) => {
2060 self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
2061 continue;
2062 }
2063 };
2064 let password_ref = map
2066 .get_secret_key("td_password")
2067 .is_some_and(|k| !k.is_empty());
2068 let mut form = CtpForm {
2069 broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
2070 td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
2071 td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
2072 td_password: String::new(),
2073 td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
2074 td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
2075 md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
2076 md_user: String::new(),
2077 md_password: String::new(),
2078 symbols: String::new(),
2079 accept_futures: false,
2080 accept_options: false,
2081 };
2082 for c in map
2085 .get_plain("product_classes")
2086 .unwrap_or_default()
2087 .split(',')
2088 {
2089 if c.trim() == "1" {
2090 form.accept_futures = true;
2091 }
2092 }
2093 let Ok(mut row) = AccountRow::derive(&form) else {
2094 self.log_line(format!(
2095 "[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
2096 ));
2097 continue;
2098 };
2099 row.restored = true;
2100 row.auto_connect = flags.get(gw).copied().unwrap_or(false);
2101 row.password_ref = password_ref;
2102 restored.push(row.gateway.clone());
2103 self.add_account(row).ok();
2104 }
2105 if !restored.is_empty() {
2106 self.log_line(format!(
2107 "[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
2108 restored.len()
2109 ));
2110 self.bump();
2111 }
2112 restored
2113 }
2114
2115 pub fn restore_saved_accounts(&self) -> Vec<String> {
2117 match ek_cfg::Store::eco() {
2118 Ok(store) => self.restore_saved_accounts_from(&store),
2119 Err(_) => {
2120 self.log_line(
2121 "[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
2122 );
2123 Vec::new()
2124 }
2125 }
2126 }
2127
2128 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
2130 *self.cfg_briefs.lock().unwrap() = Some(b);
2131 self.bump();
2132 }
2133 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
2135 self.cfg_briefs.lock().unwrap().clone()
2136 }
2137
2138 pub fn cfg_delete_armed(&self) -> Option<String> {
2140 self.cfg_delete_arm.lock().unwrap().clone()
2141 }
2142 pub fn arm_cfg_delete(&self, ns: Option<String>) {
2143 *self.cfg_delete_arm.lock().unwrap() = ns;
2144 self.bump();
2145 }
2146
2147 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
2149 *self.pipeline_lines.lock().unwrap() = lines;
2150 self.bump();
2151 }
2152
2153 pub fn pipeline_lines(&self) -> Vec<String> {
2155 self.pipeline_lines.lock().unwrap().clone()
2156 }
2157
2158 pub fn note_bad_payload(&self, line: &str) {
2164 let mut book = self.bad_payload.lock().unwrap();
2165 let n = book.entry(line.to_string()).or_insert(0);
2166 *n += 1;
2167 if *n == 1 {
2168 self.log_line(format!(
2169 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
2170 ));
2171 }
2172 }
2173
2174 #[must_use]
2176 pub fn bad_payload_count(&self, line: &str) -> u64 {
2177 self.bad_payload
2178 .lock()
2179 .unwrap()
2180 .get(line)
2181 .copied()
2182 .unwrap_or(0)
2183 }
2184
2185 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
2199 for r in rows {
2200 let (warn, crit) = pending_thresholds(r.soft_cap);
2201 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
2202 self.push_alert(
2203 crate::toast::AlertLevel::Warning,
2204 format!(
2205 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
2206 r.name, r.pending, r.peak, r.soft_cap
2207 ),
2208 );
2209 }
2210 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
2211 self.push_alert(
2212 crate::toast::AlertLevel::Error,
2213 format!(
2214 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
2215 r.name, r.pending, r.soft_cap
2216 ),
2217 );
2218 }
2219 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
2220 self.push_alert(
2221 crate::toast::AlertLevel::Error,
2222 format!(
2223 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
2224 r.name, r.soft_cap, r.spilled, r.peak
2225 ),
2226 );
2227 }
2228 }
2229 }
2230
2231 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
2239 let (limit, stale) = self.status_thresholds();
2240 let now = ek_ctp::map::now_epoch_ms();
2241 let mut alarm: Option<(LoginState, String)> = None;
2244 let mut found = false;
2245 {
2246 let mut acc = self.accounts.lock().unwrap();
2247 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2248 found = true;
2249 a.apply_status(st, now, limit, stale);
2250 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
2251 alarm = Some((
2252 a.login,
2253 a.status_note
2254 .clone()
2255 .unwrap_or_else(|| a.login.label().to_string()),
2256 ));
2257 }
2258 }
2259 }
2260 if found {
2261 self.bump();
2262 }
2263 if let Some((login, note)) = alarm
2264 && self.alert_once(format!("{gateway}|{}", login.label()))
2265 {
2266 self.push_alert(
2267 crate::toast::AlertLevel::Error,
2268 format!("账户 {gateway}:{note}"),
2269 );
2270 }
2271 }
2272
2273 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2277 let mut acc = self.accounts.lock().unwrap();
2278 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2279 a.login = LoginState::Unknown;
2280 a.status_note = Some(format!("状态口问不到:{why}"));
2281 a.login_brief = None;
2282 drop(acc);
2283 self.bump();
2284 }
2285 }
2286
2287 pub fn mark_account_stopped(&self, gateway: &str) {
2289 let mut acc = self.accounts.lock().unwrap();
2290 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2291 a.login = LoginState::Stopped;
2292 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2293 a.login_brief = None;
2294 drop(acc);
2295 self.bump();
2296 }
2297 }
2298
2299 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2302 let source = self.md_source();
2303 let mut f = form;
2304 if source.as_str() != gateway {
2305 f.md_addr.clear();
2306 f.md_user.clear();
2307 f.md_password.clear();
2308 }
2309 self.update_account_form(gateway, f.clone());
2310 f
2311 }
2312
2313 #[must_use]
2317 pub fn draft_from_existing(&self) -> CtpForm {
2318 let acc = self.accounts.lock().unwrap();
2319 let mut draft = CtpForm::default();
2320 if let Some(a) = acc.last() {
2321 draft.broker_id = a.form.broker_id.clone();
2322 draft.td_auth_code = a.form.td_auth_code.clone();
2323 draft.td_app_id = a.form.td_app_id.clone();
2324 draft.accept_futures = a.form.accept_futures;
2325 draft.accept_options = a.form.accept_options;
2326 }
2327 draft
2328 }
2329
2330 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2332 let mut acc = self.accounts.lock().unwrap();
2333 match acc.iter_mut().find(|a| a.gateway == gateway) {
2334 Some(a) => {
2335 a.form = form;
2336 drop(acc);
2337 self.bump();
2338 true
2339 }
2340 None => false,
2341 }
2342 }
2343
2344 pub fn remove_account(&self, gateway: &str) -> bool {
2346 let mut acc = self.accounts.lock().unwrap();
2347 let before = acc.len();
2348 acc.retain(|a| a.gateway != gateway);
2349 let removed = acc.len() != before;
2350 if removed {
2351 let mut md = self.md_source.lock().unwrap();
2353 if md.as_str() == gateway {
2354 md.clear();
2355 }
2356 drop(acc);
2357 self.bump();
2358 }
2359 removed
2360 }
2361
2362 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2371 let mut newly = Vec::new();
2372 {
2373 let mut acc = self.accounts.lock().unwrap();
2374 for a in acc.iter_mut() {
2375 let up = rows.iter().any(|r| r.name == a.gateway);
2376 if up && !a.connected {
2377 newly.push(a.gateway.clone());
2378 }
2379 a.connected = up;
2380 if up {
2381 a.stage = AccountStage::Assembled;
2382 a.fail_note = None;
2383 } else {
2384 a.chain_note = None;
2387 if a.stage == AccountStage::Assembled {
2388 a.stage = AccountStage::NotAssembled;
2389 a.login = LoginState::Unknown;
2390 a.status_note = None;
2391 a.login_brief = None;
2392 }
2393 }
2394 }
2395 }
2396 if !newly.is_empty() {
2397 self.bump();
2398 }
2399 newly
2400 }
2401
2402 pub fn md_source(&self) -> String {
2404 self.md_source.lock().unwrap().clone()
2405 }
2406
2407 pub fn accounts_locked(&self) -> bool {
2409 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2410 }
2411
2412 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2414 {
2415 let acc = self.accounts.lock().unwrap();
2416 if acc.iter().any(|a| a.connected) {
2417 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2418 }
2419 if !acc.iter().any(|a| a.gateway == gateway) {
2420 return Err(format!("账户 {gateway} 不在清册"));
2421 }
2422 }
2423 let mut acc = self.accounts.lock().unwrap();
2424 for a in acc.iter_mut() {
2425 a.md_source = a.gateway == gateway;
2426 }
2427 drop(acc);
2428 let mut md = self.md_source.lock().unwrap();
2429 *md = gateway.to_string();
2430 self.bump();
2431 Ok(())
2432 }
2433
2434 pub fn subscriptions(&self) -> Vec<String> {
2436 self.subscriptions.lock().unwrap().clone()
2437 }
2438
2439 pub fn note_subscriptions(&self, symbols: &[String]) {
2441 {
2442 let mut subs = self.subscriptions.lock().unwrap();
2443 for s in symbols {
2444 let s = s.trim();
2445 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2446 subs.push(s.to_string());
2447 }
2448 }
2449 }
2450 self.bump();
2451 }
2452
2453 pub fn default_gateway(&self) -> Option<String> {
2460 let chosen = {
2461 let f = self.order_form.lock().unwrap();
2462 f.gateway.trim().to_string()
2463 };
2464 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2465 return Some(chosen);
2466 }
2467 let acc = self.accounts.lock().unwrap();
2468 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2469 }
2470
2471 pub fn kline_symbols(&self) -> Vec<String> {
2473 self.kline_windows.lock().unwrap().clone()
2474 }
2475
2476 pub fn open_kline(&self, symbol: impl Into<String>) {
2478 let s = symbol.into();
2479 let mut v = self.kline_windows.lock().unwrap();
2480 if !v.contains(&s) {
2481 v.push(s.clone());
2482 }
2483 drop(v);
2484 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2485 let fq = if s.contains('.') {
2488 s.clone()
2489 } else {
2490 format!("{}.{}", self.selected_exchange(), s)
2491 };
2492 const KLINE_INTERVAL_SECS: i64 = 60;
2493 {
2494 let mut fetched = self.kline_fetched.lock().unwrap();
2495 if fetched.contains(&(fq.clone(), KLINE_INTERVAL_SECS)) {
2496 return;
2497 }
2498 fetched.insert((fq.clone(), KLINE_INTERVAL_SECS));
2499 }
2500 self.kline_fetches.lock().unwrap().push(KlineFetch {
2501 symbol: fq,
2502 interval_secs: KLINE_INTERVAL_SECS,
2503 });
2504 self.log_line(format!("[kline] 入队历史补数 {s}(1min×300,tq 桥)"));
2505 }
2506
2507 pub fn take_kline_fetches(&self) -> Vec<KlineFetch> {
2509 std::mem::take(&mut self.kline_fetches.lock().unwrap())
2510 }
2511
2512 pub fn kline_fetch_pending(&self) -> bool {
2514 !self.kline_fetches.lock().unwrap().is_empty()
2515 }
2516
2517 pub fn close_kline(&self, symbol: &str) {
2519 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2520 self.bump();
2521 }
2522
2523 pub fn contracts_window_open(&self) -> bool {
2527 self.contracts_window.load(Ordering::Acquire)
2528 }
2529
2530 pub fn set_contracts_window_open(&self, v: bool) {
2532 self.contracts_window.store(v, Ordering::Release);
2533 self.bump();
2534 }
2535
2536 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2543 self.contract_store.lock().unwrap().put(c);
2544 if !c.name.is_empty() {
2545 self.backfill_names(&c.symbol, &c.name);
2546 }
2547 self.bump();
2548 }
2549
2550 fn backfill_names(&self, symbol: &str, name: &str) {
2555 let mut touched = false;
2556 {
2557 let mut rows = self.trade_orders.lock().unwrap();
2558 for (_, (_, o)) in rows.iter_mut() {
2559 if o.symbol == symbol && o.name.is_empty() {
2560 o.name = name.to_string();
2561 touched = true;
2562 }
2563 }
2564 }
2565 {
2566 let mut rows = self.trade_trades.lock().unwrap();
2567 for (_, t) in rows.iter_mut() {
2568 if t.symbol == symbol && t.name.is_empty() {
2569 t.name = name.to_string();
2570 touched = true;
2571 }
2572 }
2573 }
2574 {
2575 let mut rows = self.trade_positions.lock().unwrap();
2576 for (_, (_, p)) in rows.iter_mut() {
2577 if p.symbol == symbol && p.name.is_empty() {
2578 p.name = name.to_string();
2579 touched = true;
2580 }
2581 }
2582 }
2583 if touched {
2584 self.trade_dirty.store(true, Ordering::Release);
2586 }
2587 }
2588
2589 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2591 self.contract_store.lock().unwrap().query(pattern)
2592 }
2593
2594 pub fn contract_count(&self) -> usize {
2596 self.contract_store.lock().unwrap().len()
2597 }
2598
2599 fn contract_name_of(&self, symbol: &str) -> Option<String> {
2601 self.contract_store
2602 .lock()
2603 .unwrap()
2604 .query(symbol)
2605 .into_iter()
2606 .find(|c| c.symbol == symbol && !c.name.is_empty())
2607 .map(|c| c.name)
2608 }
2609
2610 pub fn contract_query(&self) -> String {
2612 self.contract_query.lock().unwrap().clone()
2613 }
2614
2615 pub fn set_contract_query(&self, q: impl Into<String>) {
2617 *self.contract_query.lock().unwrap() = q.into();
2618 self.bump();
2619 }
2620
2621 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2623 let mut v = self.contract_selected.lock().unwrap();
2624 match v.iter().position(|s| s == en_symbol) {
2625 Some(i) => {
2626 v.remove(i);
2627 }
2628 None => v.push(en_symbol.to_string()),
2629 }
2630 self.bump();
2631 }
2632
2633 pub fn selected_contracts(&self) -> Vec<String> {
2635 self.contract_selected.lock().unwrap().clone()
2636 }
2637
2638 #[must_use]
2644 pub fn cancel_reject_note(msg: &str) -> String {
2645 let head = msg.split(' ').next().unwrap_or(msg);
2646 let note = if head.starts_with("unknown:order") {
2647 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
2648 } else if head.starts_with("mismatch:gateway") {
2649 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
2650 } else if head.starts_with("terminal:status") {
2651 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
2652 } else if head.starts_with("limit:cancel-guard") {
2653 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
2654 } else {
2655 ""
2656 };
2657 if note.is_empty() {
2658 format!("撤单被拒:{msg}")
2660 } else {
2661 format!("撤单被拒:{note}(机器码:{msg})")
2662 }
2663 }
2664
2665 pub fn log_line(&self, line: impl Into<String>) {
2667 self.logs.lock().unwrap().push(line);
2668 self.bump();
2669 }
2670
2671 pub fn clear_logs(&self) {
2673 self.logs.lock().unwrap().clear();
2674 self.bump();
2675 }
2676
2677 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2679 let r = self.logs.lock().unwrap();
2680 (r.snapshot(), r.spilled())
2681 }
2682
2683 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2685 let r = self.logs.lock().unwrap();
2686 (r.entries(), r.spilled())
2687 }
2688
2689 pub fn logs_len(&self) -> usize {
2691 self.logs.lock().unwrap().len()
2692 }
2693
2694 pub fn set_snapshot(&self, snap: BusSnap) {
2696 *self.snap.lock().unwrap() = Arc::new(snap);
2697 self.bump();
2698 }
2699
2700 pub fn snapshot(&self) -> Arc<BusSnap> {
2702 Arc::clone(&self.snap.lock().unwrap())
2703 }
2704
2705 pub fn rev(&self) -> u64 {
2707 self.rev.load(Ordering::Acquire)
2708 }
2709
2710 pub fn bump(&self) {
2712 self.rev.fetch_add(1, Ordering::AcqRel);
2713 }
2714
2715 pub fn request_stop(&self) {
2717 self.stop.store(true, Ordering::Release);
2718 }
2719
2720 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2723 self.pending_alerts
2724 .lock()
2725 .unwrap_or_else(std::sync::PoisonError::into_inner)
2726 .push((level, text.into()));
2727 }
2728
2729 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2731 let mut q = self
2732 .pending_alerts
2733 .lock()
2734 .unwrap_or_else(std::sync::PoisonError::into_inner);
2735 std::mem::take(&mut *q)
2736 }
2737
2738 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2740 self.toasts.lock()
2741 }
2742
2743 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2745 let key = key.into();
2746 self.alerted
2747 .lock()
2748 .unwrap_or_else(std::sync::PoisonError::into_inner)
2749 .insert(key)
2750 }
2751
2752 #[must_use]
2754 pub fn pending_alert_count(&self) -> usize {
2755 self.pending_alerts
2756 .lock()
2757 .unwrap_or_else(std::sync::PoisonError::into_inner)
2758 .len()
2759 }
2760
2761 #[must_use]
2763 pub fn alerted_count(&self) -> usize {
2764 self.alerted
2765 .lock()
2766 .unwrap_or_else(std::sync::PoisonError::into_inner)
2767 .len()
2768 }
2769
2770 pub fn exit_confirm_open(&self) -> bool {
2772 self.exit_confirm_open
2773 .load(std::sync::atomic::Ordering::Relaxed)
2774 }
2775
2776 pub fn open_exit_confirm(&self) {
2777 self.exit_confirm_open
2778 .store(true, std::sync::atomic::Ordering::Relaxed);
2779 }
2780
2781 pub fn close_exit_confirm(&self) {
2782 self.exit_confirm_open
2783 .store(false, std::sync::atomic::Ordering::Relaxed);
2784 }
2785
2786 pub fn exit_confirmed(&self) -> bool {
2787 self.exit_confirmed
2788 .load(std::sync::atomic::Ordering::Relaxed)
2789 }
2790
2791 pub fn confirm_exit(&self) {
2792 self.exit_confirmed
2793 .store(true, std::sync::atomic::Ordering::Relaxed);
2794 }
2795
2796 pub fn is_stop(&self) -> bool {
2797 self.stop.load(Ordering::Acquire)
2798 }
2799
2800 #[must_use]
2802 pub fn pluginwin_open(&self) -> bool {
2803 self.pluginwin_open.load(Ordering::Relaxed)
2804 }
2805
2806 pub fn set_pluginwin_open(&self, on: bool) {
2808 self.pluginwin_open.store(on, Ordering::Relaxed);
2809 }
2810
2811 #[must_use]
2813 pub fn bundle_open(&self) -> bool {
2814 self.bundle_open.load(Ordering::Relaxed)
2815 }
2816
2817 pub fn set_bundle_open(&self, on: bool) {
2819 self.bundle_open.store(on, Ordering::Relaxed);
2820 }
2821
2822 #[must_use]
2824 pub fn assembly_locked(&self) -> bool {
2825 self.assembly_locked.load(Ordering::Relaxed)
2826 }
2827
2828 pub fn set_assembly_locked(&self, on: bool) {
2830 self.assembly_locked.store(on, Ordering::Relaxed);
2831 }
2832
2833 pub fn push_cmd(&self, cmd: UiCmd) {
2837 self.cmds.lock().unwrap().push_back(cmd);
2838 self.bump();
2839 }
2840
2841 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2843 self.cmds.lock().unwrap().drain(..).collect()
2844 }
2845
2846 #[must_use]
2848 pub fn cmd_pending(&self) -> bool {
2849 !self.cmds.lock().unwrap().is_empty()
2850 }
2851
2852 pub fn pending_confirm(&self) -> Option<String> {
2854 self.pending_confirm.lock().unwrap().clone()
2855 }
2856
2857 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2859 let mut g = self.pending_confirm.lock().unwrap();
2860 match g.take() {
2861 Some(p) if p == name => ConfirmStep::Fired,
2862 Some(_) => {
2863 *g = Some(name.to_string());
2864 ConfirmStep::Switched
2865 }
2866 None => {
2867 *g = Some(name.to_string());
2868 ConfirmStep::Armed
2869 }
2870 }
2871 }
2872
2873 pub fn confirm_cancel(&self) {
2875 *self.pending_confirm.lock().unwrap() = None;
2876 self.bump();
2877 }
2878
2879 pub fn market(&self) -> &MarketStore {
2883 &self.market
2884 }
2885
2886 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2895 let named = |symbol: &str, name: String| -> String {
2898 if name.is_empty() {
2899 self.contract_name_of(symbol).unwrap_or(name)
2900 } else {
2901 name
2902 }
2903 };
2904 {
2905 let mut book = self.trade_orders.lock().unwrap();
2906 book.clear();
2907 for mut o in ts.orders.into_iter().rev() {
2908 o.name = named(&o.symbol, o.name);
2909 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2910 book.insert(o.en_orderid(), (seq, o));
2911 }
2912 }
2913 {
2914 let mut book = self.trade_trades.lock().unwrap();
2915 book.clear();
2916 let mut ids = self.trade_ids.lock().unwrap();
2917 ids.clear();
2918 for mut t in ts.trades.into_iter().rev() {
2919 t.name = named(&t.symbol, t.name);
2920 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2921 ids.insert(t.en_tradeid());
2922 book.push_front((seq, t));
2923 }
2924 }
2925 {
2926 let mut book = self.trade_positions.lock().unwrap();
2927 book.clear();
2928 for mut p in ts.positions.into_iter().rev() {
2929 p.name = named(&p.symbol, p.name);
2930 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2931 book.insert(p.en_positionid(), (seq, p));
2932 }
2933 }
2934 {
2935 let mut book = self.trade_funds.lock().unwrap();
2936 book.clear();
2937 for a in ts.accounts.into_iter().rev() {
2938 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2939 book.insert(a.en_accountid(), (seq, a));
2940 }
2941 }
2942 self.trade_dirty.store(true, Ordering::Release);
2943 self.bump();
2944 }
2945
2946 pub fn note_order(&self, o: &ek_data::order::OrderData) {
2954 let key = o.en_orderid();
2955 {
2956 let mut book = self.trade_orders.lock().unwrap();
2957 match book.get_mut(&key) {
2958 Some((_, cur)) => {
2959 cur.name.clone_from(&o.name);
2960 cur.traded = o.traded;
2961 cur.traded_price = o.traded_price;
2962 cur.status = o.status;
2963 cur.status_msg.clone_from(&o.status_msg);
2964 cur.datetime = o.datetime;
2965 }
2966 None => {
2967 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2968 book.insert(key, (seq, o.clone()));
2969 }
2970 }
2971 }
2972 self.trade_dirty.store(true, Ordering::Release);
2973 self.bump();
2974 }
2975
2976 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
2984 let id = t.en_tradeid();
2985 if self.trade_ids.lock().unwrap().contains(&id) {
2986 return;
2987 }
2988 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2989 let dropped = {
2990 let mut book = self.trade_trades.lock().unwrap();
2991 book.push_front((seq, t.clone()));
2992 self.trade_ids.lock().unwrap().insert(id);
2993 let over = book.len().saturating_sub(TRADE_CAP);
2994 for _ in 0..over {
2995 if let Some((_, old)) = book.pop_back() {
2997 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
2998 }
2999 }
3000 over
3001 };
3002 if dropped > 0 {
3003 let total = self
3004 .trade_dropped
3005 .fetch_add(dropped as u64, Ordering::AcqRel)
3006 + dropped as u64;
3007 if total == dropped as u64 {
3009 self.push_alert(
3010 crate::toast::AlertLevel::Warning,
3011 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
3012 );
3013 }
3014 }
3015 self.trade_dirty.store(true, Ordering::Release);
3016 self.bump();
3017 }
3018
3019 pub fn note_position(&self, p: &ek_data::position::PositionData) {
3024 let key = p.en_positionid();
3025 {
3026 let mut book = self.trade_positions.lock().unwrap();
3027 match book.get_mut(&key) {
3028 Some((_, cur)) => {
3029 cur.volume = p.volume;
3030 cur.yd_volume = p.yd_volume;
3031 cur.can_use_volume = p.can_use_volume;
3032 cur.frozen = p.frozen;
3033 cur.on_road_volume = p.on_road_volume;
3034 cur.market_value = p.market_value;
3035 cur.avg_price = p.avg_price;
3036 cur.price = p.price;
3037 cur.last_price = p.last_price;
3038 cur.pnl = p.pnl;
3039 cur.profit_rate = p.profit_rate;
3040 cur.name.clone_from(&p.name);
3041 }
3042 None => {
3043 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3044 book.insert(key, (seq, p.clone()));
3045 }
3046 }
3047 }
3048 self.trade_dirty.store(true, Ordering::Release);
3049 self.bump();
3050 }
3051
3052 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
3054 let key = a.en_accountid();
3055 {
3056 let mut book = self.trade_funds.lock().unwrap();
3057 match book.get_mut(&key) {
3058 Some((_, cur)) => {
3059 cur.balance = a.balance;
3060 cur.frozen = a.frozen;
3061 }
3062 None => {
3063 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3064 book.insert(key, (seq, a.clone()));
3065 }
3066 }
3067 }
3068 self.trade_dirty.store(true, Ordering::Release);
3069 self.bump();
3070 }
3071
3072 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
3074 let mut removed = 0usize;
3075 {
3076 let mut book = self.trade_orders.lock().unwrap();
3077 let before = book.len();
3078 book.retain(|_, (_, o)| o.gateway_name != gateway);
3079 removed += before - book.len();
3080 }
3081 {
3082 let mut book = self.trade_trades.lock().unwrap();
3083 let before = book.len();
3084 book.retain(|(_, t)| t.gateway_name != gateway);
3085 removed += before - book.len();
3086 let prefix = format!("{gateway}.");
3089 self.trade_ids
3090 .lock()
3091 .unwrap()
3092 .retain(|id| !id.starts_with(&prefix));
3093 }
3094 {
3095 let mut book = self.trade_positions.lock().unwrap();
3096 let before = book.len();
3097 book.retain(|_, (_, p)| p.gateway_name != gateway);
3098 removed += before - book.len();
3099 }
3100 {
3101 let mut book = self.trade_funds.lock().unwrap();
3102 let before = book.len();
3103 book.retain(|_, (_, a)| a.gateway_name != gateway);
3104 removed += before - book.len();
3105 }
3106 if removed > 0 {
3107 self.trade_dirty.store(true, Ordering::Release);
3108 self.bump();
3109 }
3110 removed
3111 }
3112
3113 #[must_use]
3115 pub fn trade_dropped(&self) -> u64 {
3116 self.trade_dropped.load(Ordering::Acquire)
3117 }
3118
3119 pub fn trade_snap(&self) -> Arc<TradeSnap> {
3124 let mut cache = self.trade_snap.lock().unwrap();
3125 if self.trade_dirty.swap(false, Ordering::AcqRel) {
3126 *cache = Arc::new(self.build_trade_snap());
3127 }
3128 Arc::clone(&cache)
3129 }
3130
3131 fn build_trade_snap(&self) -> TradeSnap {
3134 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
3135 where
3136 T: Clone,
3137 {
3138 let mut v: Vec<(u64, &String, &T)> =
3139 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
3140 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
3141 v.into_iter().map(|(_, _, v)| v.clone()).collect()
3142 }
3143 TradeSnap {
3144 orders: by_seq(&self.trade_orders.lock().unwrap()),
3145 trades: self
3146 .trade_trades
3147 .lock()
3148 .unwrap()
3149 .iter()
3150 .map(|(_, t)| t.clone())
3151 .collect(),
3152 positions: by_seq(&self.trade_positions.lock().unwrap()),
3153 accounts: by_seq(&self.trade_funds.lock().unwrap()),
3154 }
3155 }
3156
3157 pub fn ctp_form(&self) -> CtpForm {
3161 self.ctp_form.lock().unwrap().clone()
3162 }
3163
3164 pub fn set_ctp_form(&self, f: CtpForm) {
3166 *self.ctp_form.lock().unwrap() = f;
3167 self.bump();
3168 }
3169
3170 pub fn row_selected(&self, table: &str, key: &str) -> bool {
3174 self.row_sel
3175 .lock()
3176 .unwrap()
3177 .get(table)
3178 .is_some_and(|k| k == key)
3179 }
3180
3181 pub fn select_row(&self, table: &str, key: &str) {
3183 self.row_sel
3184 .lock()
3185 .unwrap()
3186 .insert(table.to_string(), key.to_string());
3187 }
3188}