1use std::collections::{BTreeMap, HashSet, VecDeque};
7use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, AtomicUsize, Ordering};
8use std::sync::{Arc, Mutex};
9
10use ek_data::bar::BarData;
11use ek_dm::wire::BarsFrame;
12use ek_station::wire::PlugRow;
13
14pub const RING_CAP: usize = 4096;
19
20pub(crate) fn beijing_tz() -> chrono::FixedOffset {
25 chrono::FixedOffset::east_opt(8 * 3600).expect("+08 offset is always valid")
26}
27
28#[derive(Debug, Clone)]
32pub struct LogEntry {
33 pub at: String,
35 pub text: String,
37}
38
39#[derive(Debug)]
43pub struct LogRing {
44 lines: VecDeque<LogEntry>,
45 cap: usize,
46 spilled: u64,
47 spill_dir: Option<std::path::PathBuf>,
48}
49
50pub(crate) fn default_spill_dir() -> std::path::PathBuf {
52 if let Ok(d) = std::env::var("EK_UI_LOG_DIR")
53 && !d.trim().is_empty()
54 {
55 return std::path::PathBuf::from(d);
56 }
57 std::env::var("APPDATA")
58 .map(|r| std::path::PathBuf::from(r).join("ek").join("logs"))
59 .unwrap_or_else(|_| std::path::PathBuf::from(".ek-logs"))
60}
61
62impl Default for LogRing {
63 fn default() -> Self {
64 Self::new_spilling(RING_CAP, default_spill_dir())
65 }
66}
67
68impl LogRing {
69 pub fn new(cap: usize) -> Self {
71 Self {
72 lines: VecDeque::with_capacity(cap),
73 cap,
74 spilled: 0,
75 spill_dir: None,
76 }
77 }
78
79 pub fn new_spilling(cap: usize, dir: std::path::PathBuf) -> Self {
83 Self {
84 lines: VecDeque::with_capacity(cap),
85 cap,
86 spilled: 0,
87 spill_dir: Some(dir),
88 }
89 }
90
91 fn spill_write(&self, entry: &LogEntry) {
93 let Some(dir) = self.spill_dir.as_ref() else {
94 return;
95 };
96 let day = chrono::Local::now().format("%Y%m%d");
97 let path = dir.join(format!("ekapp-{day}.log"));
98 let _ = std::fs::create_dir_all(dir);
99 if let Ok(mut f) = std::fs::OpenOptions::new().append(true).create(true).open(&path) {
100 use std::io::Write as _;
101 let _ = writeln!(f, "{} | {}", entry.at, entry.text);
102 }
103 }
104
105 pub fn push(&mut self, line: impl Into<String>) {
107 let at = chrono::Utc::now()
108 .with_timezone(&beijing_tz())
109 .format("%H:%M:%S")
110 .to_string();
111 let entry = LogEntry {
112 at,
113 text: line.into(),
114 };
115 if self.spill_dir.is_some() {
116 self.spill_write(&entry);
117 }
118 if self.cap == 0 {
119 self.spilled += 1;
120 return;
121 }
122 if self.lines.len() == self.cap {
123 self.lines.pop_front();
124 self.spilled += 1;
125 }
126 self.lines.push_back(entry);
127 }
128
129 pub fn len(&self) -> usize {
131 self.lines.len()
132 }
133
134 pub fn is_empty(&self) -> bool {
136 self.lines.is_empty()
137 }
138
139 pub fn spilled(&self) -> u64 {
141 self.spilled
142 }
143
144 pub fn snapshot(&self) -> Vec<String> {
146 self.lines.iter().map(|e| e.text.clone()).collect()
147 }
148
149 pub fn entries(&self) -> Vec<LogEntry> {
151 self.lines.iter().cloned().collect()
152 }
153
154 pub fn clear(&mut self) {
156 self.lines.clear();
157 }
158}
159
160#[derive(Debug, Clone, Default)]
162pub struct BusSnap {
163 pub rows: Vec<PlugRow>,
165 pub services: Vec<(String, Vec<String>)>,
167}
168
169#[derive(Debug, Clone, PartialEq)]
171pub enum UiCmd {
172 Start { factory: String, name: String },
175 Remove { name: String },
177 Reload {
179 interval: i64,
180 symbol: String,
181 bars: Vec<BarData>,
182 },
183 SendOrder {
189 req: ek_data::order_request::OrderRequest,
190 gateway: String,
191 mode: ek_oms::LockMode,
192 },
193 CancelOrder { orderid: String, gateway: String },
197 CtpConfigure { gateway: String, form: CtpForm },
201 Connect { gateway: String, form: CtpForm },
206 CtpSubscribe {
209 gateway: String,
210 symbols: Vec<String>,
211 },
212 Disconnect { gateway: String },
216 CfgList,
218 CfgDelete { ns: String },
220 CfgResave { ns: String },
223}
224
225pub const KLINE_CELL_CAP: usize = 4096;
227
228pub const MAX_ACCOUNTS: usize = 64;
233
234pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
239
240pub const ACCOUNT_STALE_SECS: i64 = 30;
246
247pub use ek_ctp::driver::CHAIN_STATE_PREFIX;
253
254pub const LIST_POLL_TICKS: u32 = 4;
260
261pub const STATUS_ASKS_PER_TICK: usize = 2;
266
267pub const TRADE_CAP: usize = 5000;
272
273pub const PENDING_WARN_FLOOR: u64 = 4_096;
275pub const PENDING_CRIT_FLOOR: u64 = 32_768;
277
278#[must_use]
286pub fn pending_thresholds(soft_cap: u64) -> (u64, u64) {
287 let warn = (soft_cap / 16).max(PENDING_WARN_FLOOR);
288 let crit = (soft_cap / 4).max(PENDING_CRIT_FLOOR);
289 (warn.min(soft_cap.max(1)), crit.min(soft_cap.max(1)))
290}
291
292#[derive(Debug, Clone, Copy, PartialEq, Eq)]
294pub enum ConfirmStep {
295 Armed,
296 Fired,
297 Switched,
298}
299
300#[must_use]
303pub fn is_quarantined(state: &str) -> bool {
304 state.contains("Quarantin")
305}
306
307#[derive(Debug, Clone)]
311pub struct KlineFetch {
312 pub symbol: String,
313 pub interval_secs: i64,
314}
315
316#[derive(Debug, Default)]
318pub struct MarketStore {
319 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
320 frames: AtomicU64,
322 updates: AtomicU64,
324 reloads: AtomicU64,
326}
327
328#[derive(Debug, Clone, PartialEq)]
330pub struct CellView {
331 pub symbol: String,
332 pub interval: i64,
333 pub bars: Vec<BarData>,
334}
335
336impl MarketStore {
337 pub fn on_frame(&self, frame: &BarsFrame) {
339 let mut cells = self.cells.lock().unwrap();
340 for b in &frame.bars {
341 let key = (b.symbol.clone(), frame.interval);
342 let cell = cells.entry(key).or_default();
343 match cell.last() {
344 Some(last) if b.datetime < last.datetime => continue,
345 Some(last) if b.datetime == last.datetime => {
346 let slot = cell.last_mut().expect("len>1");
347 *slot = b.clone();
348 }
349 _ => {
350 cell.push(b.clone());
351 if cell.len() > KLINE_CELL_CAP {
352 cell.remove(0);
353 }
354 }
355 }
356 }
357 self.frames.fetch_add(1, Ordering::AcqRel);
358 }
359
360 pub fn on_update(&self, b: &BarData) {
362 let mut cells = self.cells.lock().unwrap();
363 let iv = b
364 .interval
365 .unwrap_or(ek_data::constant::Interval::MINUTE)
366 .seconds();
367 let key = (b.symbol.clone(), iv);
368 let cell = cells.entry(key).or_default();
369 match cell.last() {
370 Some(last) if b.datetime == last.datetime => {
371 let slot = cell.last_mut().expect("len>1");
372 *slot = b.clone();
373 }
374 Some(last) if b.datetime < last.datetime => {}
375 _ => {
376 cell.push(b.clone());
377 if cell.len() > KLINE_CELL_CAP {
378 cell.remove(0);
379 }
380 }
381 }
382 self.updates.fetch_add(1, Ordering::AcqRel);
383 }
384
385 pub fn note_history(&self, symbol: &str, iv_secs: i64, bars: &[BarData]) {
391 let mut cells = self.cells.lock().unwrap();
392 let cell = cells.entry((symbol.to_string(), iv_secs)).or_default();
393 cell.extend(bars.iter().cloned());
394 cell.sort_unstable_by_key(|b| b.datetime);
395 cell.dedup_by(|a, b| a.datetime == b.datetime);
396 if cell.len() > KLINE_CELL_CAP {
397 let excess = cell.len() - KLINE_CELL_CAP;
398 cell.drain(0..excess);
399 }
400 self.frames.fetch_add(1, Ordering::AcqRel);
401 }
402
403 pub fn note_reload(&self) {
405 self.reloads.fetch_add(1, Ordering::AcqRel);
406 }
407
408 pub fn cells(&self) -> Vec<CellView> {
410 self.cells
411 .lock()
412 .unwrap()
413 .iter()
414 .map(|((sym, iv), bars)| CellView {
415 symbol: sym.clone(),
416 interval: *iv,
417 bars: bars.clone(),
418 })
419 .collect()
420 }
421
422 pub fn find_cell(&self, sym: &str) -> Option<CellView> {
427 self.cells()
428 .into_iter()
429 .find(|c| c.symbol == sym || c.symbol.ends_with(&format!(".{sym}")))
430 }
431
432 pub fn counts(&self) -> (u64, u64, u64) {
434 (
435 self.frames.load(Ordering::Acquire),
436 self.updates.load(Ordering::Acquire),
437 self.reloads.load(Ordering::Acquire),
438 )
439 }
440}
441
442#[derive(Debug, Default, Clone, PartialEq)]
445pub struct ReloadForm {
446 pub interval: String,
447 pub symbol: String,
448 pub csv: String,
449}
450
451pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
458 let g = raw.trim();
459 if !g.is_empty() {
460 if g.chars().any(|c| c.is_whitespace()) {
461 return Err(format!("网关解析: 含空白:{raw:?}"));
462 }
463 return Ok(g.to_string());
464 }
465 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
466 Some(f) => Ok(f.to_string()),
467 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
468 }
469}
470
471#[derive(Debug, Clone, PartialEq)]
474pub struct CtpForm {
475 pub broker_id: String,
476 pub td_addr: String,
477 pub td_user: String,
478 pub td_password: String,
479 pub td_auth_code: String,
480 pub td_app_id: String,
481 pub md_addr: String,
482 pub md_user: String,
483 pub md_password: String,
484 pub symbols: String,
486 pub accept_futures: bool,
489 pub accept_options: bool,
490}
491
492impl Default for CtpForm {
495 fn default() -> Self {
496 Self {
497 broker_id: String::new(),
498 td_addr: String::new(),
499 td_user: String::new(),
500 td_password: String::new(),
501 td_auth_code: String::new(),
502 td_app_id: String::new(),
503 md_addr: String::new(),
504 md_user: String::new(),
505 md_password: String::new(),
506 symbols: String::new(),
507 accept_futures: true,
508 accept_options: false,
509 }
510 }
511}
512
513impl CtpForm {
514 #[must_use]
518 pub fn product_classes(&self) -> Vec<String> {
519 let mut codes = Vec::new();
520 if self.accept_futures {
521 codes.push("1".to_string());
522 }
523 if self.accept_options {
524 codes.push("2".to_string());
525 }
526 codes
527 }
528
529 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
531 ek_ctp::plugin::ConfigureReq {
532 broker_id: self.broker_id.trim().to_string(),
533 td_addr: self.td_addr.trim().to_string(),
534 td_user: self.td_user.trim().to_string(),
535 td_password: self.td_password.clone().into(),
536 td_auth_code: self.td_auth_code.trim().to_string(),
537 td_app_id: self.td_app_id.trim().to_string(),
538 md_addr: self.md_addr.trim().to_string(),
539 md_user: self.md_user.trim().to_string(),
540 md_password: self.md_password.clone().into(),
541 product_classes: self.product_classes(),
542 }
543 }
544
545 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
547 let list: Vec<String> = self
548 .symbols
549 .split([',', ',', ' '])
550 .map(|s| s.trim().to_string())
551 .filter(|s| !s.is_empty())
552 .collect();
553 if list.is_empty() {
554 return Err("订阅解析: 标的位空".to_string());
555 }
556 Ok(list)
557 }
558
559 pub fn validate(&self) -> Result<(), String> {
568 fn front(name: &str, v: &str) -> Result<(), String> {
569 let v = v.trim();
570 let Some(rest) = v.strip_prefix("tcp://") else {
571 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
572 };
573 let Some((host, port)) = rest.rsplit_once(':') else {
574 return Err(format!("{name} 缺端口:{name} {:?}", v));
575 };
576 if host.trim().is_empty() {
577 return Err(format!("{name} 主机位空:{}", v));
578 }
579 let port: u32 = port
580 .parse()
581 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
582 if port == 0 || port > 65535 {
583 return Err(format!("{name} 端口越界(1..=65535):{}", v));
584 }
585 Ok(())
586 }
587 if self.broker_id.trim().is_empty() {
588 return Err("broker 位空".to_string());
589 }
590 if self.td_user.trim().is_empty() {
591 return Err("交易账号位空".to_string());
592 }
593 front("td 前置", &self.td_addr)?;
594 if !self.md_addr.trim().is_empty() {
595 front("md 前置", &self.md_addr)?;
596 }
597 if self.td_app_id.trim().is_empty() {
598 return Err("AppID 位空(CTP 登录必需)".to_string());
599 }
600 if self.td_auth_code.trim().is_empty() {
601 return Err("AuthCode 位空(CTP 登录必需)".to_string());
602 }
603 Ok(())
604 }
605}
606
607#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
614pub enum AccountStage {
615 #[default]
617 NotAssembled,
618 Assembling,
620 Assembled,
622 AssembleFailed,
624}
625
626impl AccountStage {
627 #[must_use]
629 pub fn label(self) -> &'static str {
630 match self {
631 AccountStage::NotAssembled => "未装配",
632 AccountStage::Assembling => "装配中…",
633 AccountStage::Assembled => "已装配·登录态未知",
636 AccountStage::AssembleFailed => "装配失败",
637 }
638 }
639}
640
641#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
646pub enum LoginState {
647 #[default]
652 Unknown,
653 Online,
655 LoginFailed,
657 FrontDown,
659 QueryStarved,
661 Stopped,
663}
664
665impl LoginState {
666 #[must_use]
668 pub fn label(self) -> &'static str {
669 match self {
670 LoginState::Unknown => "登录态未知",
671 LoginState::Online => "在线",
672 LoginState::LoginFailed => "登录失败",
673 LoginState::FrontDown => "掉线",
674 LoginState::QueryStarved => "查询断供",
675 LoginState::Stopped => "已断开",
676 }
677 }
678
679 #[must_use]
681 pub fn is_alarm(self) -> bool {
682 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
683 }
684}
685
686fn count_note(label: &str, n: u32) -> Option<String> {
688 (n > 0).then(|| format!("{label} {n} 次"))
689}
690
691fn brief_counter_reason(full: &str) -> Option<String> {
698 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
699 let body = tail.trim().trim_start_matches("CTP:").trim();
700 if body.is_empty() {
701 return None;
702 }
703 let code = full
704 .rfind("代码:")
705 .map(|i| {
706 let rest = &full[i + "代码:".len()..];
707 let end = rest.find(',').unwrap_or(rest.len());
708 rest[..end].trim().to_string()
709 })
710 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
711 Some(match code {
712 Some(c) => format!("{body}({c})"),
713 None => body.to_string(),
714 })
715}
716
717#[derive(Debug, Clone, PartialEq, Default)]
723pub struct AccountRow {
724 pub gateway: String,
726 pub alias: String,
728 pub broker_id: String,
729 pub account: String,
730 pub md_source: bool,
732 pub form: CtpForm,
734 pub restored: bool,
740 pub auto_connect: bool,
745 pub password_ref: bool,
750 pub stage: AccountStage,
752 pub fail_note: Option<String>,
754 pub effective_classes: Vec<String>,
759 pub connected: bool,
762 pub login: LoginState,
766 pub status_note: Option<String>,
769 pub login_brief: Option<String>,
776 pub chain_note: Option<String>,
784}
785
786impl AccountRow {
787 pub fn derive(form: &CtpForm) -> Result<Self, String> {
789 Self::derive_with(form, &[])
790 }
791
792 pub fn derive_with(
802 form: &CtpForm,
803 taken: &[(String, String, String, String)],
804 ) -> Result<Self, String> {
805 let broker = form.broker_id.trim();
806 let account = form.td_user.trim();
807 let gateway = crate::appconfig::unique_gw_name(
808 taken,
809 broker,
810 account,
811 form.td_addr.trim(),
812 )?;
813 Ok(Self {
814 gateway,
815 alias: String::new(),
816 broker_id: broker.to_string(),
817 account: account.to_string(),
818 md_source: false,
819 form: form.clone(),
820 stage: AccountStage::NotAssembled,
821 fail_note: None,
822 effective_classes: Vec::new(),
823 connected: false,
824 login: LoginState::Unknown,
825 status_note: None,
826 login_brief: None,
827 chain_note: None,
828 restored: false,
831 auto_connect: false,
832 password_ref: false,
833 })
834 }
835
836 #[must_use]
842 pub fn status_label(&self) -> String {
843 if self.stage == AccountStage::Assembled {
844 match self.login {
845 LoginState::Unknown => "已装配·登录态未知".to_string(),
846 other => format!("已装配·{}", other.label()),
847 }
848 } else {
849 self.stage.label().to_string()
850 }
851 }
852
853 pub fn apply_status(
866 &mut self,
867 st: &ek_ctp::plugin::GatewayStatus,
868 now_ms: i64,
869 reject_limit: u32,
870 stale_secs: i64,
871 ) {
872 let td_side = !self.form.td_addr.trim().is_empty();
873 let (login, note) = if td_side {
874 if st.td_login_failed && !st.td_logged_in {
875 (
876 LoginState::LoginFailed,
877 st.td_last_error
878 .clone()
879 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
880 )
881 } else if st.td_logged_in {
882 if st.td_query_reject_streak >= reject_limit {
883 (
884 LoginState::QueryStarved,
885 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
886 )
887 } else if let Some(last) = st.td_last_account_ms {
888 let idle_ms = now_ms.saturating_sub(last);
889 if idle_ms > stale_secs.saturating_mul(1_000) {
890 (
891 LoginState::QueryStarved,
892 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
893 )
894 } else {
895 (LoginState::Online, count_note("掉线", st.td_disconnects))
896 }
897 } else {
898 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
905 match since_login_ms {
906 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
907 LoginState::QueryStarved,
908 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
909 ),
910 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
911 }
912 }
913 } else if self.login == LoginState::Stopped {
914 let extra = if st.td_disconnects > 0 {
915 format!("(此前掉线 {} 次)", st.td_disconnects)
916 } else {
917 String::new()
918 };
919 (
920 LoginState::Stopped,
921 Some(format!("会话已停{extra};『连接』可恢复")),
922 )
923 } else if st.td_disconnects > 0 {
924 (
925 LoginState::FrontDown,
926 Some(format!("掉线 {} 次", st.td_disconnects)),
927 )
928 } else {
929 (LoginState::Unknown, st.td_last_error.clone())
930 }
931 } else if st.md_logged_in {
932 (
933 LoginState::Online,
934 count_note("掉线", st.md_disconnects)
935 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
936 )
937 } else if self.login == LoginState::Stopped {
938 (
939 LoginState::Stopped,
940 Some("会话已停;『连接』可恢复".to_string()),
941 )
942 } else if st.md_disconnects > 0 {
943 (
944 LoginState::FrontDown,
945 Some(format!("掉线 {} 次", st.md_disconnects)),
946 )
947 } else {
948 (LoginState::Unknown, st.md_last_error.clone())
949 };
950 let note = if td_side && st.td_instrument_rejects > 0 {
955 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
956 match note {
957 Some(base) if base.contains("合约查询被拒") => Some(base),
959 Some(base) => Some(format!("{base};{add}")),
960 None => Some(add),
961 }
962 } else {
963 note
964 };
965 self.login = login;
966 self.status_note = note;
967 self.login_brief = if login.is_alarm() {
970 st.td_last_error
971 .as_deref()
972 .or(st.md_last_error.as_deref())
973 .and_then(brief_counter_reason)
974 } else {
975 None
976 };
977 }
978
979 #[must_use]
981 pub fn display_name(&self) -> String {
982 let a = self.alias.trim();
983 if a.is_empty() {
984 self.account.clone()
985 } else {
986 a.to_string()
987 }
988 }
989}
990
991#[derive(Debug, Clone, PartialEq, Default)]
994pub struct TradeSnap {
995 pub orders: Vec<ek_data::order::OrderData>,
996 pub trades: Vec<ek_data::trade::TradeData>,
997 pub positions: Vec<ek_data::position::PositionData>,
998 pub accounts: Vec<ek_data::account::AccountData>,
1000}
1001
1002pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
1005 let interval: i64 = form
1006 .interval
1007 .trim()
1008 .parse()
1009 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
1010 let symbol = form.symbol.trim().to_string();
1011 if symbol.is_empty() {
1012 return Err("回灌解析: symbol 空".to_string());
1013 }
1014 if form.csv.trim().is_empty() {
1015 return Err("回灌解析: csv 空".to_string());
1016 }
1017 let mut bars = Vec::new();
1018 for (i, raw) in form.csv.lines().enumerate() {
1019 let line = raw.trim();
1020 if line.is_empty() || line.starts_with('#') {
1021 continue;
1022 }
1023 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
1024 if parts.len() != 6 {
1025 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
1026 }
1027 let num = |s: &str| -> Result<f64, String> {
1028 s.parse::<f64>()
1029 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
1030 };
1031 let dt_s: i64 = parts[0]
1032 .parse()
1033 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
1034 bars.push(BarData {
1035 symbol: symbol.clone(),
1036 datetime: dt_s * 1_000_000_000,
1037 interval: Some(
1038 ek_dm::wire::interval_of_seconds(interval)
1039 .unwrap_or(ek_data::constant::Interval::MINUTE),
1040 ),
1041 open_price: num(parts[1])?,
1042 high_price: num(parts[2])?,
1043 low_price: num(parts[3])?,
1044 close_price: num(parts[4])?,
1045 volume: num(parts[5])?,
1046 gateway_name: String::new(),
1049 exchange: ek_data::constant::Exchange::UNKNOWN,
1051 turnover: 0.0,
1055 open_interest: 0.0,
1056 pct_change: 0.0,
1057 });
1058 }
1059 if bars.is_empty() {
1060 return Err("回灌解析: 无有效行".to_string());
1061 }
1062 Ok((interval, symbol, bars))
1063}
1064
1065#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1070pub enum TopLeftTab {
1071 #[default]
1072 Orders,
1073 Active,
1074}
1075
1076impl TopLeftTab {
1077 pub fn label(self) -> &'static str {
1078 match self {
1079 Self::Orders => "委托",
1080 Self::Active => "未成交",
1081 }
1082 }
1083}
1084
1085#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1087pub enum BottomRightTab {
1088 #[default]
1089 Account,
1090 Logs,
1091}
1092
1093impl BottomRightTab {
1094 pub fn label(self) -> &'static str {
1095 match self {
1096 Self::Account => "资金",
1097 Self::Logs => "日志",
1098 }
1099 }
1100}
1101
1102#[derive(Debug, Clone, Copy, PartialEq)]
1105pub struct DataSplit {
1106 pub col: f32,
1108 pub row: f32,
1110}
1111
1112impl Default for DataSplit {
1113 fn default() -> Self {
1114 Self {
1115 col: crate::splitters::DATA_SPLIT_DEFAULT,
1116 row: crate::splitters::DATA_SPLIT_DEFAULT,
1117 }
1118 }
1119}
1120
1121pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
1124 ek_data::constant::Exchange::SHFE,
1125 ek_data::constant::Exchange::DCE,
1126 ek_data::constant::Exchange::CZCE,
1127 ek_data::constant::Exchange::CFFEX,
1128 ek_data::constant::Exchange::INE,
1129 ek_data::constant::Exchange::GFEX,
1130];
1131
1132#[derive(Debug, Default)]
1142pub struct TickStore {
1143 ticks: BTreeMap<String, ek_data::tick::TickData>,
1144}
1145
1146impl TickStore {
1147 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1150 self.ticks
1151 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1152 }
1153
1154 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1156 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1157 v.sort_by(|a, b| {
1158 (a.exchange.as_str().to_string(), a.symbol.clone())
1159 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1160 });
1161 v
1162 }
1163
1164 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1166 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1167 for t in self.rows() {
1168 let ex = t.exchange.as_str().to_string();
1169 match out.last_mut() {
1170 Some((e, v)) if *e == ex => v.push(t),
1171 _ => out.push((ex, vec![t])),
1172 }
1173 }
1174 out
1175 }
1176
1177 pub fn len(&self) -> usize {
1178 self.ticks.len()
1179 }
1180
1181 pub fn is_empty(&self) -> bool {
1182 self.ticks.is_empty()
1183 }
1184}
1185
1186#[derive(Debug, Clone, PartialEq)]
1188pub struct OrderForm {
1189 pub exchange: String,
1191 pub symbol: String,
1193 pub direction: ek_data::constant::Direction,
1194 pub offset: ek_data::constant::Offset,
1195 pub order_type: ek_data::constant::OrderType,
1196 pub price: String,
1198 pub volume: String,
1199 pub gateway: String,
1201 pub follow_price: bool,
1203 pub auto_split: bool,
1206}
1207
1208impl Default for OrderForm {
1209 fn default() -> Self {
1210 Self {
1211 exchange: "SHFE".to_string(),
1212 symbol: String::new(),
1213 direction: ek_data::constant::Direction::LONG,
1214 offset: ek_data::constant::Offset::OPEN,
1215 order_type: ek_data::constant::OrderType::LIMIT,
1216 price: String::new(),
1217 volume: "1".to_string(),
1218 gateway: String::new(),
1219 follow_price: true,
1220 auto_split: true,
1221 }
1222 }
1223}
1224
1225impl OrderForm {
1226 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1228 use std::str::FromStr;
1229 let symbol = self.symbol.trim();
1230 if symbol.is_empty() {
1231 return Err("下单: 代码为空".to_string());
1232 }
1233 if symbol.contains('.') {
1234 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1235 }
1236 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1237 .map_err(|e| format!("下单: {e}"))?;
1238 let price: f64 = self
1239 .price
1240 .trim()
1241 .parse()
1242 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1243 let volume: f64 = self
1244 .volume
1245 .trim()
1246 .parse()
1247 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1248 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1249 return Err("下单: 限价单价格须为正".to_string());
1250 }
1251 if volume <= 0.0 {
1252 return Err("下单: 数量须为正".to_string());
1253 }
1254 Ok(ek_data::order_request::OrderRequest {
1255 symbol: symbol.to_string(),
1256 exchange,
1257 direction: self.direction,
1258 order_type: self.order_type,
1259 volume,
1260 price,
1261 offset: self.offset,
1262 ..ek_data::order_request::OrderRequest::default()
1263 })
1264 }
1265}
1266
1267#[derive(Debug, Default)]
1273pub struct ContractStore {
1274 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1275}
1276
1277impl ContractStore {
1278 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1279 self.contracts
1280 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1281 }
1282
1283 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1285 let p = pattern.trim();
1286 self.contracts
1287 .values()
1288 .filter(|c| {
1289 p.is_empty()
1290 || c.symbol.contains(p)
1291 || c.exchange.as_str().eq_ignore_ascii_case(p)
1292 || c.name.contains(p)
1293 })
1294 .cloned()
1295 .collect()
1296 }
1297
1298 pub fn len(&self) -> usize {
1299 self.contracts.len()
1300 }
1301
1302 pub fn is_empty(&self) -> bool {
1303 self.contracts.is_empty()
1304 }
1305}
1306
1307#[derive(Debug, Default)]
1309pub struct UiShared {
1310 logs: Mutex<LogRing>,
1311 snap: Mutex<Arc<BusSnap>>,
1312 market: MarketStore,
1313 ctp_form: Mutex<CtpForm>,
1314 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1317 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1318 trade_ids: Mutex<HashSet<String>>,
1326 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1327 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1328 trade_seq: AtomicU64,
1330 trade_dropped: AtomicU64,
1332 trade_dirty: AtomicBool,
1334 trade_snap: Mutex<Arc<TradeSnap>>,
1336 tick_store: Mutex<TickStore>,
1337 order_form: Mutex<OrderForm>,
1338 data_split: Mutex<DataSplit>,
1341 selected_exchange: Mutex<String>,
1343 topleft_tab: Mutex<TopLeftTab>,
1345 bottomright_tab: Mutex<BottomRightTab>,
1347 ops_window: AtomicBool,
1348 accounts_window: AtomicBool,
1351 contracts_window: AtomicBool,
1352 contract_store: Mutex<ContractStore>,
1353 contract_query: Mutex<String>,
1354 contract_selected: Mutex<Vec<String>>,
1355 kline_windows: Mutex<Vec<String>>,
1356 kline_fetches: Mutex<Vec<KlineFetch>>,
1358 kline_fetched: Mutex<HashSet<(String, i64)>>,
1360 rev: AtomicU64,
1361 stop: AtomicBool,
1363 cmds: Mutex<VecDeque<UiCmd>>,
1365 pending_confirm: Mutex<Option<String>>,
1367 accounts: Mutex<Vec<AccountRow>>,
1369 account_edit_window: AtomicBool,
1371 editing_account: Mutex<Option<String>>,
1373 edit_alias: Mutex<String>,
1375 edit_error: Mutex<Option<String>>,
1377 md_source: Mutex<String>,
1379 subscriptions: Mutex<Vec<String>>,
1381 row_sel: Mutex<BTreeMap<String, String>>,
1384 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1388 status_cursor: AtomicUsize,
1394 status_seed: Mutex<Vec<String>>,
1397 list_tick: AtomicU32,
1399 list_due: AtomicBool,
1403 seed_due: AtomicBool,
1407 pipeline_due: AtomicBool,
1409 cfg_due: AtomicBool,
1411 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1414 cfg_delete_arm: Mutex<Option<String>>,
1416 theme: AtomicU8,
1420 theme_persist_due: AtomicBool,
1423 auto_connect_persist_due: AtomicBool,
1427 alias_persist_due: AtomicBool,
1431 archive_purge: Mutex<Vec<String>>,
1435 bad_payload: Mutex<BTreeMap<String, u64>>,
1437 pipeline_lines: Mutex<Vec<String>>,
1440 status_thresholds: Mutex<Option<(u32, i64)>>,
1443 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1445 toasts: crate::toast::ToastsCell,
1448 alerted: Mutex<std::collections::BTreeSet<String>>,
1450 exit_confirm_open: AtomicBool,
1452 exit_confirmed: AtomicBool,
1454 assembly_locked: AtomicBool,
1457}
1458
1459impl UiShared {
1460 pub fn new() -> Self {
1462 Self::default()
1463 }
1464
1465 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1469 self.tick_store.lock().unwrap().put(t);
1470 self.bump();
1471 }
1472
1473 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1475 self.tick_store.lock().unwrap().rows()
1476 }
1477
1478 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1480 self.tick_store.lock().unwrap().groups()
1481 }
1482
1483 pub fn tick_count(&self) -> usize {
1485 self.tick_store.lock().unwrap().len()
1486 }
1487
1488 pub fn order_form(&self) -> OrderForm {
1490 self.order_form.lock().unwrap().clone()
1491 }
1492
1493 pub fn set_order_form(&self, f: OrderForm) {
1495 *self.order_form.lock().unwrap() = f;
1496 self.bump();
1497 }
1498
1499 pub fn selected_exchange(&self) -> String {
1503 let s = self.selected_exchange.lock().unwrap().clone();
1504 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1505 s
1506 } else {
1507 MARKET_ORDER[0].as_str().to_string()
1508 }
1509 }
1510
1511 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1513 *self.selected_exchange.lock().unwrap() = s.into();
1514 self.bump();
1515 }
1516
1517 pub fn topleft_tab(&self) -> TopLeftTab {
1519 *self.topleft_tab.lock().unwrap()
1520 }
1521
1522 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1524 *self.topleft_tab.lock().unwrap() = t;
1525 self.bump();
1526 }
1527
1528 pub fn bottomright_tab(&self) -> BottomRightTab {
1530 *self.bottomright_tab.lock().unwrap()
1531 }
1532
1533 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1535 *self.bottomright_tab.lock().unwrap() = t;
1536 self.bump();
1537 }
1538
1539 pub fn data_split(&self) -> DataSplit {
1541 *self.data_split.lock().unwrap()
1542 }
1543
1544 pub fn set_data_split(&self, s: DataSplit) {
1547 let s = DataSplit {
1548 col: crate::splitters::clamp_frac(s.col),
1549 row: crate::splitters::clamp_frac(s.row),
1550 };
1551 let mut g = self.data_split.lock().unwrap();
1552 if g.col != s.col || g.row != s.row {
1553 *g = s;
1554 self.bump();
1555 }
1556 }
1557
1558 pub fn reset_layout(&self) {
1562 *self.data_split.lock().unwrap() = DataSplit::default();
1563 *self.selected_exchange.lock().unwrap() = String::new();
1564 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1565 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1566 self.bump();
1567 }
1568
1569 pub fn ops_window_open(&self) -> bool {
1573 self.ops_window.load(Ordering::Acquire)
1574 }
1575
1576 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1579 self.table_sorts
1580 .lock()
1581 .unwrap()
1582 .get(table)
1583 .copied()
1584 .flatten()
1585 }
1586
1587 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1589 let mut m = self.table_sorts.lock().unwrap();
1590 let asc = match m.get(table).copied().flatten() {
1591 Some((c, a)) if c == col => !a,
1592 _ => true,
1593 };
1594 m.insert(table.to_string(), Some((col, asc)));
1595 }
1596
1597 pub fn set_ops_window_open(&self, v: bool) {
1598 self.ops_window.store(v, Ordering::Release);
1599 self.bump();
1600 }
1601
1602 pub fn accounts_window_open(&self) -> bool {
1606 self.accounts_window.load(Ordering::Acquire)
1607 }
1608
1609 pub fn set_accounts_window_open(&self, v: bool) {
1611 self.accounts_window.store(v, Ordering::Release);
1612 self.bump();
1613 }
1614
1615 pub fn account_edit_open(&self) -> bool {
1617 self.account_edit_window.load(Ordering::Acquire)
1618 }
1619
1620 pub fn editing_account(&self) -> Option<String> {
1622 self.editing_account.lock().unwrap().clone()
1623 }
1624
1625 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1629 match gateway {
1630 Some(gw) => {
1631 let Some(row) = self.account(gw) else {
1632 return false;
1633 };
1634 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1635 *self.edit_alias.lock().unwrap() = row.alias.clone();
1636 self.set_ctp_form(row.form.clone());
1637 }
1638 None => {
1639 *self.editing_account.lock().unwrap() = None;
1640 *self.edit_alias.lock().unwrap() = String::new();
1641 let draft = self.draft_from_existing();
1642 self.set_ctp_form(draft);
1643 }
1644 }
1645 *self.edit_error.lock().unwrap() = None;
1646 self.account_edit_window.store(true, Ordering::Release);
1647 self.bump();
1648 true
1649 }
1650
1651 pub fn end_account_edit(&self) {
1653 *self.editing_account.lock().unwrap() = None;
1654 *self.edit_alias.lock().unwrap() = String::new();
1655 *self.edit_error.lock().unwrap() = None;
1656 self.account_edit_window.store(false, Ordering::Release);
1657 let draft = self.draft_from_existing();
1658 self.set_ctp_form(draft);
1659 self.bump();
1660 }
1661
1662 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1668 if self.editing_account().as_deref() == Some(gateway) {
1669 self.end_account_edit();
1670 }
1671 }
1672
1673 pub fn edit_alias(&self) -> String {
1675 self.edit_alias.lock().unwrap().clone()
1676 }
1677
1678 pub fn set_edit_alias(&self, v: impl Into<String>) {
1680 *self.edit_alias.lock().unwrap() = v.into();
1681 self.bump();
1682 }
1683
1684 pub fn edit_error(&self) -> Option<String> {
1686 self.edit_error.lock().unwrap().clone()
1687 }
1688
1689 pub fn set_edit_error(&self, e: Option<String>) {
1691 *self.edit_error.lock().unwrap() = e;
1692 self.bump();
1693 }
1694
1695 pub fn account_edit_dirty(&self) -> bool {
1697 let editing = self.editing_account();
1698 let draft = self.ctp_form();
1699 match editing {
1700 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1701 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1702 }
1703 }
1704
1705 pub fn accounts(&self) -> Vec<AccountRow> {
1707 self.accounts.lock().unwrap().clone()
1708 }
1709
1710 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1712 self.accounts
1713 .lock()
1714 .unwrap()
1715 .iter()
1716 .find(|a| a.gateway == gateway)
1717 .cloned()
1718 }
1719
1720 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1726 let mut acc = self.accounts.lock().unwrap();
1727 if acc.iter().any(|a| a.gateway == row.gateway) {
1728 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1729 }
1730 let alias = row.alias.trim();
1731 if !alias.is_empty()
1732 && acc
1733 .iter()
1734 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1735 {
1736 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1737 }
1738 if acc.len() >= MAX_ACCOUNTS {
1739 return Err(format!(
1740 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1741 ));
1742 }
1743 let has_alias = !row.alias.trim().is_empty();
1744 acc.push(row);
1745 drop(acc);
1746 if has_alias {
1747 self.alias_persist_due.store(true, Ordering::Release);
1749 }
1750 self.bump();
1751 Ok(())
1752 }
1753
1754 #[must_use]
1756 pub fn can_add_account(&self) -> bool {
1757 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1758 }
1759
1760 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1766 let alias = alias.trim().to_string();
1767 let mut acc = self.accounts.lock().unwrap();
1768 if !alias.is_empty()
1769 && acc
1770 .iter()
1771 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1772 {
1773 return false;
1774 }
1775 match acc.iter_mut().find(|a| a.gateway == gateway) {
1776 Some(a) => {
1777 a.alias = alias;
1778 drop(acc);
1779 self.alias_persist_due.store(true, Ordering::Release);
1780 self.bump();
1781 true
1782 }
1783 None => false,
1784 }
1785 }
1786
1787 pub fn note_connect_sent(&self, gateway: &str) {
1792 let mut acc = self.accounts.lock().unwrap();
1793 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1794 a.stage = AccountStage::Assembling;
1795 a.fail_note = None;
1796 a.login = LoginState::Unknown;
1797 a.status_note = None;
1798 a.login_brief = None;
1799 a.chain_note = None;
1801 drop(acc);
1802 self.bump();
1803 }
1804 }
1805
1806 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1810 let mut acc = self.accounts.lock().unwrap();
1811 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1812 match result {
1813 Ok(()) => {
1814 a.stage = AccountStage::Assembled;
1815 a.fail_note = None;
1816 }
1817 Err(e) => {
1818 a.stage = AccountStage::AssembleFailed;
1819 a.fail_note = Some(e);
1820 }
1821 }
1822 drop(acc);
1823 self.bump();
1824 }
1825 }
1826
1827 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1831 let mut acc = self.accounts.lock().unwrap();
1832 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1833 a.effective_classes = classes.to_vec();
1834 drop(acc);
1835 self.bump();
1836 }
1837 }
1838
1839 pub fn note_chain(&self, gateway: &str, note: &str) {
1844 let mut acc = self.accounts.lock().unwrap();
1845 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1846 if a.chain_note.as_deref() == Some(note) {
1847 return; }
1849 a.chain_note = Some(note.to_string());
1850 drop(acc);
1851 self.bump();
1852 }
1853 }
1854
1855 #[must_use]
1859 pub fn status_thresholds(&self) -> (u32, i64) {
1860 self.status_thresholds
1861 .lock()
1862 .unwrap()
1863 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1864 }
1865
1866 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1868 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1869 self.bump();
1870 }
1871
1872 pub fn list_poll_due(&self) -> bool {
1877 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1878 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1879 }
1880
1881 pub fn request_list_refresh(&self) {
1883 self.list_due.store(true, Ordering::Release);
1884 }
1885
1886 pub fn request_trade_reseed(&self) {
1888 self.seed_due.store(true, Ordering::Release);
1889 }
1890
1891 pub fn take_trade_reseed(&self) -> bool {
1893 self.seed_due.swap(false, Ordering::AcqRel)
1894 }
1895
1896 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1901 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1905 for gw in present {
1906 if !uniq.iter().any(|u| u == gw) {
1907 uniq.push(gw.clone());
1908 }
1909 }
1910 let n = uniq.len();
1911 if n == 0 {
1912 return Vec::new();
1913 }
1914 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1915 let take = STATUS_ASKS_PER_TICK.min(n);
1916 let mut out = Vec::with_capacity(take);
1917 for i in 0..take {
1918 out.push(uniq[(start + i) % n].clone());
1919 }
1920 out
1921 }
1922
1923 pub fn request_status_seed(&self, gateway: &str) {
1925 let mut q = self.status_seed.lock().unwrap();
1926 if !q.iter().any(|g| g == gateway) {
1927 q.push(gateway.to_string());
1928 }
1929 }
1930
1931 pub fn take_status_seeds(&self) -> Vec<String> {
1933 let mut q = self.status_seed.lock().unwrap();
1934 std::mem::take(&mut *q)
1935 }
1936
1937 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
1947 let gw = ev.gateway.as_str();
1948 let text = ev.describe();
1949 self.log_line(format!("[INFO][{gw}] {text}"));
1950 let note = text
1951 .strip_prefix(CHAIN_STATE_PREFIX)
1952 .map(|s| s.trim().to_string())
1953 .unwrap_or_else(|| text.clone());
1954 self.note_chain(gw, ¬e);
1955 self.note_gateway_status(gw, &ev.status);
1956 let alarm_kind = matches!(
1957 ev.change,
1958 ek_ctp::GatewayChange::TdDisconnected
1959 | ek_ctp::GatewayChange::TdAuthFailed
1960 | ek_ctp::GatewayChange::TdLoginFailed
1961 | ek_ctp::GatewayChange::ContractQueryStopped
1962 | ek_ctp::GatewayChange::QueryRejected
1963 | ek_ctp::GatewayChange::MdDisconnected
1964 | ek_ctp::GatewayChange::MdLoginFailed
1965 | ek_ctp::GatewayChange::SubscribeRejected
1966 );
1967 if alarm_kind {
1968 let session = if ev.change.kind().starts_with("md-") {
1970 ev.status.md_disconnects
1971 } else {
1972 ev.status.td_disconnects
1973 };
1974 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
1975 if self.alert_once(key) {
1976 self.push_alert(
1977 crate::toast::AlertLevel::Warning,
1978 format!("账户 {gw}:{}", note.trim()),
1979 );
1980 }
1981 }
1982 }
1983
1984 pub fn request_pipeline_refresh(&self) {
1986 self.pipeline_due.store(true, Ordering::Release);
1987 }
1988 pub fn take_pipeline_due(&self) -> bool {
1990 self.pipeline_due.swap(false, Ordering::AcqRel)
1991 }
1992
1993 pub fn request_cfg_refresh(&self) {
1995 self.cfg_due.store(true, Ordering::Release);
1996 }
1997
1998 pub fn take_cfg_due(&self) -> bool {
2000 self.cfg_due.swap(false, Ordering::AcqRel)
2001 }
2002
2003 pub fn theme(&self) -> crate::tokens::Theme {
2005 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
2006 }
2007
2008 pub fn set_theme(&self, theme: crate::tokens::Theme) {
2011 self.theme.store(theme.as_u8(), Ordering::Release);
2012 self.bump();
2013 }
2014
2015 pub fn toggle_theme(&self) -> crate::tokens::Theme {
2017 let next = self.theme().flipped();
2018 self.theme.store(next.as_u8(), Ordering::Release);
2019 self.theme_persist_due.store(true, Ordering::Release);
2020 self.bump();
2021 next
2022 }
2023
2024 pub fn take_theme_persist_due(&self) -> bool {
2026 self.theme_persist_due.swap(false, Ordering::AcqRel)
2027 }
2028
2029 pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
2033 {
2034 let mut acc = self.accounts.lock().unwrap();
2035 match acc.iter_mut().find(|a| a.gateway == gateway) {
2036 Some(a) => {
2037 if a.auto_connect != on {
2038 a.auto_connect = on;
2039 self.auto_connect_persist_due.store(true, Ordering::Release);
2040 self.bump();
2041 }
2042 }
2043 None => return Err(format!("账户 {gateway} 不在清册")),
2044 }
2045 }
2046 Ok(())
2047 }
2048
2049 pub fn take_auto_connect_persist_due(&self) -> bool {
2051 self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
2052 }
2053
2054 pub fn take_alias_persist_due(&self) -> bool {
2056 self.alias_persist_due.swap(false, Ordering::AcqRel)
2057 }
2058
2059 pub fn take_archive_purges(&self) -> Vec<String> {
2061 let mut q = self.archive_purge.lock().unwrap();
2062 std::mem::take(&mut *q)
2063 }
2064
2065 pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
2080 let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
2082 let mut aliases: std::collections::BTreeMap<String, String> = Default::default();
2083 match store.load(crate::plugin::THEME_NS) {
2084 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
2085 for (k, _f) in map.iter() {
2086 if let Some(gw) = k.strip_prefix("auto_connect.") {
2087 flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
2088 }
2089 if let Some(gw) = k.strip_prefix("alias.")
2090 && let Some(v) = map.get_plain(k)
2091 {
2092 let v = v.trim();
2093 if !v.is_empty() {
2094 aliases.insert(gw.to_string(), v.to_string());
2095 }
2096 }
2097 }
2098 }
2099 Ok(_) => {} Err(_) => {} }
2102 let mut restored: Vec<String> = Vec::new();
2103 let ns_list = match store.list_namespaces() {
2104 Ok(v) => v,
2105 Err(_) => {
2106 let msg = "[ui] 档案清点失败:存储根不可读(具体见运行日志)".to_string();
2107 self.log_line(msg.clone());
2108 return restored;
2109 }
2110 };
2111 for ns in ns_list {
2112 let Some(gw) = ns.strip_prefix("ctp.") else {
2113 continue;
2114 };
2115 if self.account(gw).is_some() {
2116 self.log_line(format!(
2117 "[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
2118 ));
2119 continue;
2120 }
2121 let map = match store.load(&ns) {
2122 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
2123 Ok(_) => {
2124 self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
2125 continue;
2126 }
2127 Err(e) => {
2128 self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
2129 continue;
2130 }
2131 };
2132 let password_ref = map
2134 .get_secret_key("td_password")
2135 .is_some_and(|k| !k.is_empty());
2136 let mut form = CtpForm {
2137 broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
2138 td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
2139 td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
2140 td_password: String::new(),
2141 td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
2142 td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
2143 md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
2144 md_user: String::new(),
2145 md_password: String::new(),
2146 symbols: String::new(),
2147 accept_futures: false,
2148 accept_options: false,
2149 };
2150 for c in map
2153 .get_plain("product_classes")
2154 .unwrap_or_default()
2155 .split(',')
2156 {
2157 if c.trim() == "1" {
2158 form.accept_futures = true;
2159 }
2160 }
2161 let Ok(mut row) = AccountRow::derive(&form) else {
2162 self.log_line(format!(
2163 "[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
2164 ));
2165 continue;
2166 };
2167 row.gateway = gw.to_string();
2170 row.restored = true;
2171 row.auto_connect = flags.get(gw).copied().unwrap_or(false);
2172 row.password_ref = password_ref;
2173 if let Some(a) = aliases.get(gw) {
2176 let taken = self
2177 .accounts
2178 .lock()
2179 .unwrap()
2180 .iter()
2181 .any(|x| x.alias.trim().eq_ignore_ascii_case(a));
2182 if taken {
2183 self.log_line(format!(
2184 "[ui] 档案别名 {a:?} 已被在册行占用,{gw} 不带标签复原(账户面板改别名即可)"
2185 ));
2186 } else {
2187 row.alias = a.clone();
2188 }
2189 }
2190 restored.push(row.gateway.clone());
2191 self.add_account(row).ok();
2192 }
2193 if !restored.is_empty() {
2194 self.log_line(format!(
2195 "[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
2196 restored.len()
2197 ));
2198 if self.md_source.lock().unwrap().is_empty() {
2201 let first = self
2202 .accounts
2203 .lock()
2204 .unwrap()
2205 .iter()
2206 .find(|a| a.auto_connect)
2207 .map(|a| a.gateway.clone());
2208 if let Some(gw) = first
2209 && self.set_md_source(&gw).is_ok()
2210 {
2211 self.log_line(format!(
2212 "[账户] 行情源(默认=第一个勾选自动连接的)= {gw}"
2213 ));
2214 }
2215 }
2216 self.bump();
2217 }
2218 restored
2219 }
2220
2221 pub fn restore_saved_accounts(&self) -> Vec<String> {
2223 match ek_cfg::Store::eco() {
2224 Ok(store) => self.restore_saved_accounts_from(&store),
2225 Err(_) => {
2226 self.log_line(
2227 "[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
2228 );
2229 Vec::new()
2230 }
2231 }
2232 }
2233
2234 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
2236 *self.cfg_briefs.lock().unwrap() = Some(b);
2237 self.bump();
2238 }
2239 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
2241 self.cfg_briefs.lock().unwrap().clone()
2242 }
2243
2244 pub fn cfg_delete_armed(&self) -> Option<String> {
2246 self.cfg_delete_arm.lock().unwrap().clone()
2247 }
2248 pub fn arm_cfg_delete(&self, ns: Option<String>) {
2249 *self.cfg_delete_arm.lock().unwrap() = ns;
2250 self.bump();
2251 }
2252
2253 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
2255 *self.pipeline_lines.lock().unwrap() = lines;
2256 self.bump();
2257 }
2258
2259 pub fn pipeline_lines(&self) -> Vec<String> {
2261 self.pipeline_lines.lock().unwrap().clone()
2262 }
2263
2264 pub fn note_bad_payload(&self, line: &str) {
2270 let mut book = self.bad_payload.lock().unwrap();
2271 let n = book.entry(line.to_string()).or_insert(0);
2272 *n += 1;
2273 if *n == 1 {
2274 self.log_line(format!(
2275 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
2276 ));
2277 }
2278 }
2279
2280 #[must_use]
2282 pub fn bad_payload_count(&self, line: &str) -> u64 {
2283 self.bad_payload
2284 .lock()
2285 .unwrap()
2286 .get(line)
2287 .copied()
2288 .unwrap_or(0)
2289 }
2290
2291 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
2305 for r in rows {
2306 let (warn, crit) = pending_thresholds(r.soft_cap);
2307 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
2308 self.push_alert(
2309 crate::toast::AlertLevel::Warning,
2310 format!(
2311 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
2312 r.name, r.pending, r.peak, r.soft_cap
2313 ),
2314 );
2315 }
2316 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
2317 self.push_alert(
2318 crate::toast::AlertLevel::Error,
2319 format!(
2320 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
2321 r.name, r.pending, r.soft_cap
2322 ),
2323 );
2324 }
2325 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
2326 self.push_alert(
2327 crate::toast::AlertLevel::Error,
2328 format!(
2329 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
2330 r.name, r.soft_cap, r.spilled, r.peak
2331 ),
2332 );
2333 }
2334 }
2335 }
2336
2337 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
2345 let (limit, stale) = self.status_thresholds();
2346 let now = ek_ctp::map::now_epoch_ms();
2347 let mut alarm: Option<(LoginState, String)> = None;
2350 let mut found = false;
2351 {
2352 let mut acc = self.accounts.lock().unwrap();
2353 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2354 found = true;
2355 a.apply_status(st, now, limit, stale);
2356 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
2357 alarm = Some((
2358 a.login,
2359 a.status_note
2360 .clone()
2361 .unwrap_or_else(|| a.login.label().to_string()),
2362 ));
2363 }
2364 }
2365 }
2366 if found {
2367 self.bump();
2368 }
2369 if let Some((login, note)) = alarm
2370 && self.alert_once(format!("{gateway}|{}", login.label()))
2371 {
2372 self.push_alert(
2373 crate::toast::AlertLevel::Error,
2374 format!("账户 {gateway}:{note}"),
2375 );
2376 }
2377 }
2378
2379 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2383 let mut acc = self.accounts.lock().unwrap();
2384 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2385 a.login = LoginState::Unknown;
2386 a.status_note = Some(format!("状态口问不到:{why}"));
2387 a.login_brief = None;
2388 drop(acc);
2389 self.bump();
2390 }
2391 }
2392
2393 pub fn mark_account_stopped(&self, gateway: &str) {
2395 let mut acc = self.accounts.lock().unwrap();
2396 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2397 a.login = LoginState::Stopped;
2398 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2399 a.login_brief = None;
2400 drop(acc);
2401 self.bump();
2402 }
2403 }
2404
2405 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2408 let source = self.md_source();
2409 let mut f = form;
2410 if source.as_str() != gateway {
2411 f.md_addr.clear();
2412 f.md_user.clear();
2413 f.md_password.clear();
2414 }
2415 self.update_account_form(gateway, f.clone());
2416 f
2417 }
2418
2419 #[must_use]
2423 pub fn draft_from_existing(&self) -> CtpForm {
2424 let acc = self.accounts.lock().unwrap();
2425 let mut draft = CtpForm::default();
2426 if let Some(a) = acc.last() {
2427 draft.broker_id = a.form.broker_id.clone();
2428 draft.td_auth_code = a.form.td_auth_code.clone();
2429 draft.td_app_id = a.form.td_app_id.clone();
2430 draft.accept_futures = a.form.accept_futures;
2431 draft.accept_options = a.form.accept_options;
2432 }
2433 draft
2434 }
2435
2436 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2438 let mut acc = self.accounts.lock().unwrap();
2439 match acc.iter_mut().find(|a| a.gateway == gateway) {
2440 Some(a) => {
2441 a.form = form;
2442 drop(acc);
2443 self.bump();
2444 true
2445 }
2446 None => false,
2447 }
2448 }
2449
2450 pub fn remove_account(&self, gateway: &str) -> bool {
2452 let mut acc = self.accounts.lock().unwrap();
2453 let before = acc.len();
2454 acc.retain(|a| a.gateway != gateway);
2455 let removed = acc.len() != before;
2456 if removed {
2457 let mut md = self.md_source.lock().unwrap();
2459 if md.as_str() == gateway {
2460 md.clear();
2461 }
2462 drop(acc);
2463 self.alias_persist_due.store(true, Ordering::Release);
2465 self.archive_purge.lock().unwrap().push(gateway.to_string());
2467 self.bump();
2468 }
2469 removed
2470 }
2471
2472 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2481 let mut newly = Vec::new();
2482 {
2483 let mut acc = self.accounts.lock().unwrap();
2484 for a in acc.iter_mut() {
2485 let up = rows.iter().any(|r| r.name == a.gateway);
2486 if up && !a.connected {
2487 newly.push(a.gateway.clone());
2488 }
2489 a.connected = up;
2490 if up {
2491 a.stage = AccountStage::Assembled;
2492 a.fail_note = None;
2493 } else {
2494 a.chain_note = None;
2497 if a.stage == AccountStage::Assembled {
2498 a.stage = AccountStage::NotAssembled;
2499 a.login = LoginState::Unknown;
2500 a.status_note = None;
2501 a.login_brief = None;
2502 }
2503 }
2504 }
2505 }
2506 if !newly.is_empty() {
2507 self.bump();
2508 }
2509 newly
2510 }
2511
2512 pub fn md_source(&self) -> String {
2514 self.md_source.lock().unwrap().clone()
2515 }
2516
2517 pub fn accounts_locked(&self) -> bool {
2519 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2520 }
2521
2522 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2524 {
2525 let acc = self.accounts.lock().unwrap();
2526 if acc.iter().any(|a| a.connected) {
2527 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2528 }
2529 if !acc.iter().any(|a| a.gateway == gateway) {
2530 return Err(format!("账户 {gateway} 不在清册"));
2531 }
2532 }
2533 let mut acc = self.accounts.lock().unwrap();
2534 for a in acc.iter_mut() {
2535 a.md_source = a.gateway == gateway;
2536 }
2537 drop(acc);
2538 let mut md = self.md_source.lock().unwrap();
2539 *md = gateway.to_string();
2540 self.bump();
2541 Ok(())
2542 }
2543
2544 pub fn subscriptions(&self) -> Vec<String> {
2546 self.subscriptions.lock().unwrap().clone()
2547 }
2548
2549 pub fn note_subscriptions(&self, symbols: &[String]) {
2551 {
2552 let mut subs = self.subscriptions.lock().unwrap();
2553 for s in symbols {
2554 let s = s.trim();
2555 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2556 subs.push(s.to_string());
2557 }
2558 }
2559 }
2560 self.bump();
2561 }
2562
2563 pub fn default_gateway(&self) -> Option<String> {
2570 let chosen = {
2571 let f = self.order_form.lock().unwrap();
2572 f.gateway.trim().to_string()
2573 };
2574 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2575 return Some(chosen);
2576 }
2577 let acc = self.accounts.lock().unwrap();
2578 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2579 }
2580
2581 pub fn kline_symbols(&self) -> Vec<String> {
2583 self.kline_windows.lock().unwrap().clone()
2584 }
2585
2586 pub fn open_kline(&self, symbol: impl Into<String>) {
2588 let s = symbol.into();
2589 let mut v = self.kline_windows.lock().unwrap();
2590 if !v.contains(&s) {
2591 v.push(s.clone());
2592 }
2593 drop(v);
2594 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2595 let fq = if s.contains('.') {
2598 s.clone()
2599 } else {
2600 format!("{}.{}", self.selected_exchange(), s)
2601 };
2602 const KLINE_INTERVAL_SECS: i64 = 60;
2603 {
2604 let mut fetched = self.kline_fetched.lock().unwrap();
2605 if fetched.contains(&(fq.clone(), KLINE_INTERVAL_SECS)) {
2606 return;
2607 }
2608 fetched.insert((fq.clone(), KLINE_INTERVAL_SECS));
2609 }
2610 self.kline_fetches.lock().unwrap().push(KlineFetch {
2611 symbol: fq,
2612 interval_secs: KLINE_INTERVAL_SECS,
2613 });
2614 self.log_line(format!("[kline] 入队历史补数 {s}(1min×300,tq 桥)"));
2615 }
2616
2617 pub fn take_kline_fetches(&self) -> Vec<KlineFetch> {
2619 std::mem::take(&mut self.kline_fetches.lock().unwrap())
2620 }
2621
2622 pub fn kline_fetch_pending(&self) -> bool {
2624 !self.kline_fetches.lock().unwrap().is_empty()
2625 }
2626
2627 pub fn close_kline(&self, symbol: &str) {
2629 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2630 self.bump();
2631 }
2632
2633 pub fn contracts_window_open(&self) -> bool {
2637 self.contracts_window.load(Ordering::Acquire)
2638 }
2639
2640 pub fn set_contracts_window_open(&self, v: bool) {
2642 self.contracts_window.store(v, Ordering::Release);
2643 self.bump();
2644 }
2645
2646 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2653 self.contract_store.lock().unwrap().put(c);
2654 if !c.name.is_empty() {
2655 self.backfill_names(&c.symbol, &c.name);
2656 }
2657 self.bump();
2658 }
2659
2660 fn backfill_names(&self, symbol: &str, name: &str) {
2665 let mut touched = false;
2666 {
2667 let mut rows = self.trade_orders.lock().unwrap();
2668 for (_, (_, o)) in rows.iter_mut() {
2669 if o.symbol == symbol && o.name.is_empty() {
2670 o.name = name.to_string();
2671 touched = true;
2672 }
2673 }
2674 }
2675 {
2676 let mut rows = self.trade_trades.lock().unwrap();
2677 for (_, t) in rows.iter_mut() {
2678 if t.symbol == symbol && t.name.is_empty() {
2679 t.name = name.to_string();
2680 touched = true;
2681 }
2682 }
2683 }
2684 {
2685 let mut rows = self.trade_positions.lock().unwrap();
2686 for (_, (_, p)) in rows.iter_mut() {
2687 if p.symbol == symbol && p.name.is_empty() {
2688 p.name = name.to_string();
2689 touched = true;
2690 }
2691 }
2692 }
2693 if touched {
2694 self.trade_dirty.store(true, Ordering::Release);
2696 }
2697 }
2698
2699 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2701 self.contract_store.lock().unwrap().query(pattern)
2702 }
2703
2704 pub fn contract_count(&self) -> usize {
2706 self.contract_store.lock().unwrap().len()
2707 }
2708
2709 fn contract_name_of(&self, symbol: &str) -> Option<String> {
2711 self.contract_store
2712 .lock()
2713 .unwrap()
2714 .query(symbol)
2715 .into_iter()
2716 .find(|c| c.symbol == symbol && !c.name.is_empty())
2717 .map(|c| c.name)
2718 }
2719
2720 pub fn contract_query(&self) -> String {
2722 self.contract_query.lock().unwrap().clone()
2723 }
2724
2725 pub fn set_contract_query(&self, q: impl Into<String>) {
2727 *self.contract_query.lock().unwrap() = q.into();
2728 self.bump();
2729 }
2730
2731 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2733 let mut v = self.contract_selected.lock().unwrap();
2734 match v.iter().position(|s| s == en_symbol) {
2735 Some(i) => {
2736 v.remove(i);
2737 }
2738 None => v.push(en_symbol.to_string()),
2739 }
2740 self.bump();
2741 }
2742
2743 pub fn selected_contracts(&self) -> Vec<String> {
2745 self.contract_selected.lock().unwrap().clone()
2746 }
2747
2748 #[must_use]
2754 pub fn cancel_reject_note(msg: &str) -> String {
2755 let head = msg.split(' ').next().unwrap_or(msg);
2756 let note = if head.starts_with("unknown:order") {
2757 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
2758 } else if head.starts_with("mismatch:gateway") {
2759 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
2760 } else if head.starts_with("terminal:status") {
2761 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
2762 } else if head.starts_with("limit:cancel-guard") {
2763 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
2764 } else {
2765 ""
2766 };
2767 if note.is_empty() {
2768 format!("撤单被拒:{msg}")
2770 } else {
2771 format!("撤单被拒:{note}(机器码:{msg})")
2772 }
2773 }
2774
2775 pub fn log_line(&self, line: impl Into<String>) {
2777 self.logs.lock().unwrap().push(line);
2778 self.bump();
2779 }
2780
2781 pub fn clear_logs(&self) {
2783 self.logs.lock().unwrap().clear();
2784 self.bump();
2785 }
2786
2787 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2789 let r = self.logs.lock().unwrap();
2790 (r.snapshot(), r.spilled())
2791 }
2792
2793 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2795 let r = self.logs.lock().unwrap();
2796 (r.entries(), r.spilled())
2797 }
2798
2799 pub fn logs_len(&self) -> usize {
2801 self.logs.lock().unwrap().len()
2802 }
2803
2804 pub fn set_snapshot(&self, snap: BusSnap) {
2806 *self.snap.lock().unwrap() = Arc::new(snap);
2807 self.bump();
2808 }
2809
2810 pub fn snapshot(&self) -> Arc<BusSnap> {
2812 Arc::clone(&self.snap.lock().unwrap())
2813 }
2814
2815 pub fn rev(&self) -> u64 {
2817 self.rev.load(Ordering::Acquire)
2818 }
2819
2820 pub fn bump(&self) {
2822 self.rev.fetch_add(1, Ordering::AcqRel);
2823 }
2824
2825 pub fn request_stop(&self) {
2827 self.stop.store(true, Ordering::Release);
2828 }
2829
2830 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2833 self.pending_alerts
2834 .lock()
2835 .unwrap_or_else(std::sync::PoisonError::into_inner)
2836 .push((level, text.into()));
2837 }
2838
2839 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2841 let mut q = self
2842 .pending_alerts
2843 .lock()
2844 .unwrap_or_else(std::sync::PoisonError::into_inner);
2845 std::mem::take(&mut *q)
2846 }
2847
2848 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2850 self.toasts.lock()
2851 }
2852
2853 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2855 let key = key.into();
2856 self.alerted
2857 .lock()
2858 .unwrap_or_else(std::sync::PoisonError::into_inner)
2859 .insert(key)
2860 }
2861
2862 #[must_use]
2864 pub fn pending_alert_count(&self) -> usize {
2865 self.pending_alerts
2866 .lock()
2867 .unwrap_or_else(std::sync::PoisonError::into_inner)
2868 .len()
2869 }
2870
2871 #[must_use]
2873 pub fn alerted_count(&self) -> usize {
2874 self.alerted
2875 .lock()
2876 .unwrap_or_else(std::sync::PoisonError::into_inner)
2877 .len()
2878 }
2879
2880 pub fn exit_confirm_open(&self) -> bool {
2882 self.exit_confirm_open
2883 .load(std::sync::atomic::Ordering::Relaxed)
2884 }
2885
2886 pub fn open_exit_confirm(&self) {
2887 self.exit_confirm_open
2888 .store(true, std::sync::atomic::Ordering::Relaxed);
2889 }
2890
2891 pub fn close_exit_confirm(&self) {
2892 self.exit_confirm_open
2893 .store(false, std::sync::atomic::Ordering::Relaxed);
2894 }
2895
2896 pub fn exit_confirmed(&self) -> bool {
2897 self.exit_confirmed
2898 .load(std::sync::atomic::Ordering::Relaxed)
2899 }
2900
2901 pub fn confirm_exit(&self) {
2902 self.exit_confirmed
2903 .store(true, std::sync::atomic::Ordering::Relaxed);
2904 }
2905
2906 pub fn is_stop(&self) -> bool {
2907 self.stop.load(Ordering::Acquire)
2908 }
2909
2910 #[must_use]
2912 pub fn assembly_locked(&self) -> bool {
2913 self.assembly_locked.load(Ordering::Relaxed)
2914 }
2915
2916 pub fn set_assembly_locked(&self, on: bool) {
2918 self.assembly_locked.store(on, Ordering::Relaxed);
2919 }
2920
2921 pub fn push_cmd(&self, cmd: UiCmd) {
2925 self.cmds.lock().unwrap().push_back(cmd);
2926 self.bump();
2927 }
2928
2929 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2931 self.cmds.lock().unwrap().drain(..).collect()
2932 }
2933
2934 #[must_use]
2936 pub fn cmd_pending(&self) -> bool {
2937 !self.cmds.lock().unwrap().is_empty()
2938 }
2939
2940 pub fn pending_confirm(&self) -> Option<String> {
2942 self.pending_confirm.lock().unwrap().clone()
2943 }
2944
2945 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2947 let mut g = self.pending_confirm.lock().unwrap();
2948 match g.take() {
2949 Some(p) if p == name => ConfirmStep::Fired,
2950 Some(_) => {
2951 *g = Some(name.to_string());
2952 ConfirmStep::Switched
2953 }
2954 None => {
2955 *g = Some(name.to_string());
2956 ConfirmStep::Armed
2957 }
2958 }
2959 }
2960
2961 pub fn confirm_cancel(&self) {
2963 *self.pending_confirm.lock().unwrap() = None;
2964 self.bump();
2965 }
2966
2967 pub fn market(&self) -> &MarketStore {
2971 &self.market
2972 }
2973
2974 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2983 let named = |symbol: &str, name: String| -> String {
2986 if name.is_empty() {
2987 self.contract_name_of(symbol).unwrap_or(name)
2988 } else {
2989 name
2990 }
2991 };
2992 {
2993 let mut book = self.trade_orders.lock().unwrap();
2994 book.clear();
2995 for mut o in ts.orders.into_iter().rev() {
2996 o.name = named(&o.symbol, o.name);
2997 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2998 book.insert(o.en_orderid(), (seq, o));
2999 }
3000 }
3001 {
3002 let mut book = self.trade_trades.lock().unwrap();
3003 book.clear();
3004 let mut ids = self.trade_ids.lock().unwrap();
3005 ids.clear();
3006 for mut t in ts.trades.into_iter().rev() {
3007 t.name = named(&t.symbol, t.name);
3008 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3009 ids.insert(t.en_tradeid());
3010 book.push_front((seq, t));
3011 }
3012 }
3013 {
3014 let mut book = self.trade_positions.lock().unwrap();
3015 book.clear();
3016 for mut p in ts.positions.into_iter().rev() {
3017 p.name = named(&p.symbol, p.name);
3018 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3019 book.insert(p.en_positionid(), (seq, p));
3020 }
3021 }
3022 {
3023 let mut book = self.trade_funds.lock().unwrap();
3024 book.clear();
3025 for a in ts.accounts.into_iter().rev() {
3026 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3027 book.insert(a.en_accountid(), (seq, a));
3028 }
3029 }
3030 self.trade_dirty.store(true, Ordering::Release);
3031 self.bump();
3032 }
3033
3034 pub fn note_order(&self, o: &ek_data::order::OrderData) {
3042 let key = o.en_orderid();
3043 {
3044 let mut book = self.trade_orders.lock().unwrap();
3045 match book.get_mut(&key) {
3046 Some((_, cur)) => {
3047 cur.name.clone_from(&o.name);
3048 cur.traded = o.traded;
3049 cur.traded_price = o.traded_price;
3050 cur.status = o.status;
3051 cur.status_msg.clone_from(&o.status_msg);
3052 cur.datetime = o.datetime;
3053 }
3054 None => {
3055 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3056 book.insert(key, (seq, o.clone()));
3057 }
3058 }
3059 }
3060 self.trade_dirty.store(true, Ordering::Release);
3061 self.bump();
3062 }
3063
3064 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
3072 let id = t.en_tradeid();
3073 if self.trade_ids.lock().unwrap().contains(&id) {
3074 return;
3075 }
3076 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3077 let dropped = {
3078 let mut book = self.trade_trades.lock().unwrap();
3079 book.push_front((seq, t.clone()));
3080 self.trade_ids.lock().unwrap().insert(id);
3081 let over = book.len().saturating_sub(TRADE_CAP);
3082 for _ in 0..over {
3083 if let Some((_, old)) = book.pop_back() {
3085 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
3086 }
3087 }
3088 over
3089 };
3090 if dropped > 0 {
3091 let total = self
3092 .trade_dropped
3093 .fetch_add(dropped as u64, Ordering::AcqRel)
3094 + dropped as u64;
3095 if total == dropped as u64 {
3097 self.push_alert(
3098 crate::toast::AlertLevel::Warning,
3099 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
3100 );
3101 }
3102 }
3103 self.trade_dirty.store(true, Ordering::Release);
3104 self.bump();
3105 }
3106
3107 pub fn note_position(&self, p: &ek_data::position::PositionData) {
3112 let key = p.en_positionid();
3113 {
3114 let mut book = self.trade_positions.lock().unwrap();
3115 match book.get_mut(&key) {
3116 Some((_, cur)) => {
3117 cur.volume = p.volume;
3118 cur.yd_volume = p.yd_volume;
3119 cur.can_use_volume = p.can_use_volume;
3120 cur.frozen = p.frozen;
3121 cur.on_road_volume = p.on_road_volume;
3122 cur.market_value = p.market_value;
3123 cur.avg_price = p.avg_price;
3124 cur.price = p.price;
3125 cur.last_price = p.last_price;
3126 cur.pnl = p.pnl;
3127 cur.profit_rate = p.profit_rate;
3128 cur.name.clone_from(&p.name);
3129 }
3130 None => {
3131 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3132 book.insert(key, (seq, p.clone()));
3133 }
3134 }
3135 }
3136 self.trade_dirty.store(true, Ordering::Release);
3137 self.bump();
3138 }
3139
3140 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
3142 let key = a.en_accountid();
3143 {
3144 let mut book = self.trade_funds.lock().unwrap();
3145 match book.get_mut(&key) {
3146 Some((_, cur)) => {
3147 cur.balance = a.balance;
3148 cur.frozen = a.frozen;
3149 }
3150 None => {
3151 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3152 book.insert(key, (seq, a.clone()));
3153 }
3154 }
3155 }
3156 self.trade_dirty.store(true, Ordering::Release);
3157 self.bump();
3158 }
3159
3160 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
3162 let mut removed = 0usize;
3163 {
3164 let mut book = self.trade_orders.lock().unwrap();
3165 let before = book.len();
3166 book.retain(|_, (_, o)| o.gateway_name != gateway);
3167 removed += before - book.len();
3168 }
3169 {
3170 let mut book = self.trade_trades.lock().unwrap();
3171 let before = book.len();
3172 book.retain(|(_, t)| t.gateway_name != gateway);
3173 removed += before - book.len();
3174 let prefix = format!("{gateway}.");
3177 self.trade_ids
3178 .lock()
3179 .unwrap()
3180 .retain(|id| !id.starts_with(&prefix));
3181 }
3182 {
3183 let mut book = self.trade_positions.lock().unwrap();
3184 let before = book.len();
3185 book.retain(|_, (_, p)| p.gateway_name != gateway);
3186 removed += before - book.len();
3187 }
3188 {
3189 let mut book = self.trade_funds.lock().unwrap();
3190 let before = book.len();
3191 book.retain(|_, (_, a)| a.gateway_name != gateway);
3192 removed += before - book.len();
3193 }
3194 if removed > 0 {
3195 self.trade_dirty.store(true, Ordering::Release);
3196 self.bump();
3197 }
3198 removed
3199 }
3200
3201 #[must_use]
3203 pub fn trade_dropped(&self) -> u64 {
3204 self.trade_dropped.load(Ordering::Acquire)
3205 }
3206
3207 pub fn trade_snap(&self) -> Arc<TradeSnap> {
3212 let mut cache = self.trade_snap.lock().unwrap();
3213 if self.trade_dirty.swap(false, Ordering::AcqRel) {
3214 *cache = Arc::new(self.build_trade_snap());
3215 }
3216 Arc::clone(&cache)
3217 }
3218
3219 fn build_trade_snap(&self) -> TradeSnap {
3222 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
3223 where
3224 T: Clone,
3225 {
3226 let mut v: Vec<(u64, &String, &T)> =
3227 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
3228 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
3229 v.into_iter().map(|(_, _, v)| v.clone()).collect()
3230 }
3231 TradeSnap {
3232 orders: by_seq(&self.trade_orders.lock().unwrap()),
3233 trades: self
3234 .trade_trades
3235 .lock()
3236 .unwrap()
3237 .iter()
3238 .map(|(_, t)| t.clone())
3239 .collect(),
3240 positions: by_seq(&self.trade_positions.lock().unwrap()),
3241 accounts: by_seq(&self.trade_funds.lock().unwrap()),
3242 }
3243 }
3244
3245 pub fn ctp_form(&self) -> CtpForm {
3249 self.ctp_form.lock().unwrap().clone()
3250 }
3251
3252 pub fn set_ctp_form(&self, f: CtpForm) {
3254 *self.ctp_form.lock().unwrap() = f;
3255 self.bump();
3256 }
3257
3258 pub fn row_selected(&self, table: &str, key: &str) -> bool {
3262 self.row_sel
3263 .lock()
3264 .unwrap()
3265 .get(table)
3266 .is_some_and(|k| k == key)
3267 }
3268
3269 pub fn select_row(&self, table: &str, key: &str) {
3271 self.row_sel
3272 .lock()
3273 .unwrap()
3274 .insert(table.to_string(), key.to_string());
3275 }
3276}