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 = 512;
16
17pub(crate) fn beijing_tz() -> chrono::FixedOffset {
22 chrono::FixedOffset::east_opt(8 * 3600).expect("+08 offset is always valid")
23}
24
25#[derive(Debug, Clone)]
29pub struct LogEntry {
30 pub at: String,
32 pub text: String,
34}
35
36#[derive(Debug)]
38pub struct LogRing {
39 lines: VecDeque<LogEntry>,
40 cap: usize,
41 spilled: u64,
42}
43
44impl Default for LogRing {
45 fn default() -> Self {
46 Self::new(RING_CAP)
47 }
48}
49
50impl LogRing {
51 pub fn new(cap: usize) -> Self {
53 Self {
54 lines: VecDeque::with_capacity(cap),
55 cap,
56 spilled: 0,
57 }
58 }
59
60 pub fn push(&mut self, line: impl Into<String>) {
62 if self.cap == 0 {
63 self.spilled += 1;
64 return;
65 }
66 if self.lines.len() == self.cap {
67 self.lines.pop_front();
68 self.spilled += 1;
69 }
70 let at = chrono::Utc::now()
71 .with_timezone(&beijing_tz())
72 .format("%H:%M:%S")
73 .to_string();
74 self.lines.push_back(LogEntry {
75 at,
76 text: line.into(),
77 });
78 }
79
80 pub fn len(&self) -> usize {
82 self.lines.len()
83 }
84
85 pub fn is_empty(&self) -> bool {
87 self.lines.is_empty()
88 }
89
90 pub fn spilled(&self) -> u64 {
92 self.spilled
93 }
94
95 pub fn snapshot(&self) -> Vec<String> {
97 self.lines.iter().map(|e| e.text.clone()).collect()
98 }
99
100 pub fn entries(&self) -> Vec<LogEntry> {
102 self.lines.iter().cloned().collect()
103 }
104
105 pub fn clear(&mut self) {
107 self.lines.clear();
108 }
109}
110
111#[derive(Debug, Clone, Default)]
113pub struct BusSnap {
114 pub rows: Vec<PlugRow>,
116 pub services: Vec<(String, Vec<String>)>,
118}
119
120#[derive(Debug, Clone, PartialEq)]
122pub enum UiCmd {
123 Start { factory: String, name: String },
126 Remove { name: String },
128 Reload {
130 interval: i64,
131 symbol: String,
132 bars: Vec<BarData>,
133 },
134 SendOrder {
140 req: ek_data::order_request::OrderRequest,
141 gateway: String,
142 mode: ek_oms::LockMode,
143 },
144 CancelOrder { orderid: String, gateway: String },
148 CtpConfigure { gateway: String, form: CtpForm },
152 Connect { gateway: String, form: CtpForm },
157 CtpSubscribe {
160 gateway: String,
161 symbols: Vec<String>,
162 },
163 Disconnect { gateway: String },
167 CfgList,
169 CfgDelete { ns: String },
171 CfgResave { ns: String },
174}
175
176pub const KLINE_CELL_CAP: usize = 4096;
178
179pub const MAX_ACCOUNTS: usize = 8;
182
183pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
188
189pub const ACCOUNT_STALE_SECS: i64 = 30;
195
196pub use ek_ctp::driver::CHAIN_STATE_PREFIX;
202
203pub const LIST_POLL_TICKS: u32 = 4;
209
210pub const STATUS_ASKS_PER_TICK: usize = 2;
215
216pub const TRADE_CAP: usize = 5000;
221
222pub const PENDING_WARN_FLOOR: u64 = 4_096;
224pub const PENDING_CRIT_FLOOR: u64 = 32_768;
226
227#[must_use]
235pub fn pending_thresholds(soft_cap: u64) -> (u64, u64) {
236 let warn = (soft_cap / 16).max(PENDING_WARN_FLOOR);
237 let crit = (soft_cap / 4).max(PENDING_CRIT_FLOOR);
238 (warn.min(soft_cap.max(1)), crit.min(soft_cap.max(1)))
239}
240
241#[derive(Debug, Clone, Copy, PartialEq, Eq)]
243pub enum ConfirmStep {
244 Armed,
245 Fired,
246 Switched,
247}
248
249#[must_use]
252pub fn is_quarantined(state: &str) -> bool {
253 state.contains("Quarantin")
254}
255
256#[derive(Debug, Default)]
260pub struct MarketStore {
261 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
262 frames: AtomicU64,
264 updates: AtomicU64,
266 reloads: AtomicU64,
268}
269
270#[derive(Debug, Clone, PartialEq)]
272pub struct CellView {
273 pub symbol: String,
274 pub interval: i64,
275 pub bars: Vec<BarData>,
276}
277
278impl MarketStore {
279 pub fn on_frame(&self, frame: &BarsFrame) {
281 let mut cells = self.cells.lock().unwrap();
282 for b in &frame.bars {
283 let key = (b.symbol.clone(), frame.interval);
284 let cell = cells.entry(key).or_default();
285 match cell.last() {
286 Some(last) if b.datetime < last.datetime => continue,
287 Some(last) if b.datetime == last.datetime => {
288 let slot = cell.last_mut().expect("len>1");
289 *slot = b.clone();
290 }
291 _ => {
292 cell.push(b.clone());
293 if cell.len() > KLINE_CELL_CAP {
294 cell.remove(0);
295 }
296 }
297 }
298 }
299 self.frames.fetch_add(1, Ordering::AcqRel);
300 }
301
302 pub fn on_update(&self, b: &BarData) {
304 let mut cells = self.cells.lock().unwrap();
305 let iv = b
306 .interval
307 .unwrap_or(ek_data::constant::Interval::MINUTE)
308 .seconds();
309 let key = (b.symbol.clone(), iv);
310 let cell = cells.entry(key).or_default();
311 match cell.last() {
312 Some(last) if b.datetime == last.datetime => {
313 let slot = cell.last_mut().expect("len>1");
314 *slot = b.clone();
315 }
316 Some(last) if b.datetime < last.datetime => {}
317 _ => {
318 cell.push(b.clone());
319 if cell.len() > KLINE_CELL_CAP {
320 cell.remove(0);
321 }
322 }
323 }
324 self.updates.fetch_add(1, Ordering::AcqRel);
325 }
326
327 pub fn note_reload(&self) {
329 self.reloads.fetch_add(1, Ordering::AcqRel);
330 }
331
332 pub fn cells(&self) -> Vec<CellView> {
334 self.cells
335 .lock()
336 .unwrap()
337 .iter()
338 .map(|((sym, iv), bars)| CellView {
339 symbol: sym.clone(),
340 interval: *iv,
341 bars: bars.clone(),
342 })
343 .collect()
344 }
345
346 pub fn counts(&self) -> (u64, u64, u64) {
348 (
349 self.frames.load(Ordering::Acquire),
350 self.updates.load(Ordering::Acquire),
351 self.reloads.load(Ordering::Acquire),
352 )
353 }
354}
355
356#[derive(Debug, Default, Clone, PartialEq)]
359pub struct ReloadForm {
360 pub interval: String,
361 pub symbol: String,
362 pub csv: String,
363}
364
365pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
372 let g = raw.trim();
373 if !g.is_empty() {
374 if g.chars().any(|c| c.is_whitespace()) {
375 return Err(format!("网关解析: 含空白:{raw:?}"));
376 }
377 return Ok(g.to_string());
378 }
379 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
380 Some(f) => Ok(f.to_string()),
381 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
382 }
383}
384
385#[derive(Debug, Clone, PartialEq)]
388pub struct CtpForm {
389 pub broker_id: String,
390 pub td_addr: String,
391 pub td_user: String,
392 pub td_password: String,
393 pub td_auth_code: String,
394 pub td_app_id: String,
395 pub md_addr: String,
396 pub md_user: String,
397 pub md_password: String,
398 pub symbols: String,
400 pub accept_futures: bool,
403 pub accept_options: bool,
404}
405
406impl Default for CtpForm {
409 fn default() -> Self {
410 Self {
411 broker_id: String::new(),
412 td_addr: String::new(),
413 td_user: String::new(),
414 td_password: String::new(),
415 td_auth_code: String::new(),
416 td_app_id: String::new(),
417 md_addr: String::new(),
418 md_user: String::new(),
419 md_password: String::new(),
420 symbols: String::new(),
421 accept_futures: true,
422 accept_options: false,
423 }
424 }
425}
426
427impl CtpForm {
428 #[must_use]
432 pub fn product_classes(&self) -> Vec<String> {
433 let mut codes = Vec::new();
434 if self.accept_futures {
435 codes.push("1".to_string());
436 }
437 if self.accept_options {
438 codes.push("2".to_string());
439 }
440 codes
441 }
442
443 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
445 ek_ctp::plugin::ConfigureReq {
446 broker_id: self.broker_id.trim().to_string(),
447 td_addr: self.td_addr.trim().to_string(),
448 td_user: self.td_user.trim().to_string(),
449 td_password: self.td_password.clone(),
450 td_auth_code: self.td_auth_code.trim().to_string(),
451 td_app_id: self.td_app_id.trim().to_string(),
452 md_addr: self.md_addr.trim().to_string(),
453 md_user: self.md_user.trim().to_string(),
454 md_password: self.md_password.clone(),
455 product_classes: self.product_classes(),
456 }
457 }
458
459 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
461 let list: Vec<String> = self
462 .symbols
463 .split([',', ',', ' '])
464 .map(|s| s.trim().to_string())
465 .filter(|s| !s.is_empty())
466 .collect();
467 if list.is_empty() {
468 return Err("订阅解析: 标的位空".to_string());
469 }
470 Ok(list)
471 }
472
473 pub fn validate(&self) -> Result<(), String> {
482 fn front(name: &str, v: &str) -> Result<(), String> {
483 let v = v.trim();
484 let Some(rest) = v.strip_prefix("tcp://") else {
485 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
486 };
487 let Some((host, port)) = rest.rsplit_once(':') else {
488 return Err(format!("{name} 缺端口:{name} {:?}", v));
489 };
490 if host.trim().is_empty() {
491 return Err(format!("{name} 主机位空:{}", v));
492 }
493 let port: u32 = port
494 .parse()
495 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
496 if port == 0 || port > 65535 {
497 return Err(format!("{name} 端口越界(1..=65535):{}", v));
498 }
499 Ok(())
500 }
501 if self.broker_id.trim().is_empty() {
502 return Err("broker 位空".to_string());
503 }
504 if self.td_user.trim().is_empty() {
505 return Err("交易账号位空".to_string());
506 }
507 front("td 前置", &self.td_addr)?;
508 if !self.md_addr.trim().is_empty() {
509 front("md 前置", &self.md_addr)?;
510 }
511 if self.td_app_id.trim().is_empty() {
512 return Err("AppID 位空(CTP 登录必需)".to_string());
513 }
514 if self.td_auth_code.trim().is_empty() {
515 return Err("AuthCode 位空(CTP 登录必需)".to_string());
516 }
517 Ok(())
518 }
519}
520
521#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
528pub enum AccountStage {
529 #[default]
531 NotAssembled,
532 Assembling,
534 Assembled,
536 AssembleFailed,
538}
539
540impl AccountStage {
541 #[must_use]
543 pub fn label(self) -> &'static str {
544 match self {
545 AccountStage::NotAssembled => "未装配",
546 AccountStage::Assembling => "装配中…",
547 AccountStage::Assembled => "已装配·登录态未知",
550 AccountStage::AssembleFailed => "装配失败",
551 }
552 }
553}
554
555#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
560pub enum LoginState {
561 #[default]
566 Unknown,
567 Online,
569 LoginFailed,
571 FrontDown,
573 QueryStarved,
575 Stopped,
577}
578
579impl LoginState {
580 #[must_use]
582 pub fn label(self) -> &'static str {
583 match self {
584 LoginState::Unknown => "登录态未知",
585 LoginState::Online => "在线",
586 LoginState::LoginFailed => "登录失败",
587 LoginState::FrontDown => "掉线",
588 LoginState::QueryStarved => "查询断供",
589 LoginState::Stopped => "已断开",
590 }
591 }
592
593 #[must_use]
595 pub fn is_alarm(self) -> bool {
596 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
597 }
598}
599
600fn count_note(label: &str, n: u32) -> Option<String> {
602 (n > 0).then(|| format!("{label} {n} 次"))
603}
604
605fn brief_counter_reason(full: &str) -> Option<String> {
612 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
613 let body = tail.trim().trim_start_matches("CTP:").trim();
614 if body.is_empty() {
615 return None;
616 }
617 let code = full
618 .rfind("代码:")
619 .map(|i| {
620 let rest = &full[i + "代码:".len()..];
621 let end = rest.find(',').unwrap_or(rest.len());
622 rest[..end].trim().to_string()
623 })
624 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
625 Some(match code {
626 Some(c) => format!("{body}({c})"),
627 None => body.to_string(),
628 })
629}
630
631#[derive(Debug, Clone, PartialEq, Default)]
637pub struct AccountRow {
638 pub gateway: String,
640 pub alias: String,
642 pub broker_id: String,
643 pub account: String,
644 pub md_source: bool,
646 pub form: CtpForm,
648 pub stage: AccountStage,
650 pub fail_note: Option<String>,
652 pub effective_classes: Vec<String>,
657 pub connected: bool,
660 pub login: LoginState,
664 pub status_note: Option<String>,
667 pub login_brief: Option<String>,
674 pub chain_note: Option<String>,
682}
683
684impl AccountRow {
685 pub fn derive(form: &CtpForm) -> Result<Self, String> {
690 let broker = form.broker_id.trim();
691 let account = form.td_user.trim();
692 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
693 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
694 })?;
695 Ok(Self {
696 gateway,
697 alias: String::new(),
698 broker_id: broker.to_string(),
699 account: account.to_string(),
700 md_source: false,
701 form: form.clone(),
702 stage: AccountStage::NotAssembled,
703 fail_note: None,
704 effective_classes: Vec::new(),
705 connected: false,
706 login: LoginState::Unknown,
707 status_note: None,
708 login_brief: None,
709 chain_note: None,
710 })
711 }
712
713 #[must_use]
719 pub fn status_label(&self) -> String {
720 if self.stage == AccountStage::Assembled {
721 match self.login {
722 LoginState::Unknown => "已装配·登录态未知".to_string(),
723 other => format!("已装配·{}", other.label()),
724 }
725 } else {
726 self.stage.label().to_string()
727 }
728 }
729
730 pub fn apply_status(
743 &mut self,
744 st: &ek_ctp::plugin::GatewayStatus,
745 now_ms: i64,
746 reject_limit: u32,
747 stale_secs: i64,
748 ) {
749 let td_side = !self.form.td_addr.trim().is_empty();
750 let (login, note) = if td_side {
751 if st.td_login_failed && !st.td_logged_in {
752 (
753 LoginState::LoginFailed,
754 st.td_last_error
755 .clone()
756 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
757 )
758 } else if st.td_logged_in {
759 if st.td_query_reject_streak >= reject_limit {
760 (
761 LoginState::QueryStarved,
762 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
763 )
764 } else if let Some(last) = st.td_last_account_ms {
765 let idle_ms = now_ms.saturating_sub(last);
766 if idle_ms > stale_secs.saturating_mul(1_000) {
767 (
768 LoginState::QueryStarved,
769 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
770 )
771 } else {
772 (LoginState::Online, count_note("掉线", st.td_disconnects))
773 }
774 } else {
775 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
782 match since_login_ms {
783 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
784 LoginState::QueryStarved,
785 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
786 ),
787 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
788 }
789 }
790 } else if self.login == LoginState::Stopped {
791 let extra = if st.td_disconnects > 0 {
792 format!("(此前掉线 {} 次)", st.td_disconnects)
793 } else {
794 String::new()
795 };
796 (
797 LoginState::Stopped,
798 Some(format!("会话已停{extra};『连接』可恢复")),
799 )
800 } else if st.td_disconnects > 0 {
801 (
802 LoginState::FrontDown,
803 Some(format!("掉线 {} 次", st.td_disconnects)),
804 )
805 } else {
806 (LoginState::Unknown, st.td_last_error.clone())
807 }
808 } else if st.md_logged_in {
809 (
810 LoginState::Online,
811 count_note("掉线", st.md_disconnects)
812 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
813 )
814 } else if self.login == LoginState::Stopped {
815 (
816 LoginState::Stopped,
817 Some("会话已停;『连接』可恢复".to_string()),
818 )
819 } else if st.md_disconnects > 0 {
820 (
821 LoginState::FrontDown,
822 Some(format!("掉线 {} 次", st.md_disconnects)),
823 )
824 } else {
825 (LoginState::Unknown, st.md_last_error.clone())
826 };
827 let note = if td_side && st.td_instrument_rejects > 0 {
832 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
833 match note {
834 Some(base) if base.contains("合约查询被拒") => Some(base),
836 Some(base) => Some(format!("{base};{add}")),
837 None => Some(add),
838 }
839 } else {
840 note
841 };
842 self.login = login;
843 self.status_note = note;
844 self.login_brief = if login.is_alarm() {
847 st.td_last_error
848 .as_deref()
849 .or(st.md_last_error.as_deref())
850 .and_then(brief_counter_reason)
851 } else {
852 None
853 };
854 }
855
856 #[must_use]
858 pub fn display_name(&self) -> String {
859 let a = self.alias.trim();
860 if a.is_empty() {
861 self.account.clone()
862 } else {
863 a.to_string()
864 }
865 }
866}
867
868#[derive(Debug, Clone, PartialEq, Default)]
871pub struct TradeSnap {
872 pub orders: Vec<ek_data::order::OrderData>,
873 pub trades: Vec<ek_data::trade::TradeData>,
874 pub positions: Vec<ek_data::position::PositionData>,
875 pub accounts: Vec<ek_data::account::AccountData>,
877}
878
879pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
882 let interval: i64 = form
883 .interval
884 .trim()
885 .parse()
886 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
887 let symbol = form.symbol.trim().to_string();
888 if symbol.is_empty() {
889 return Err("回灌解析: symbol 空".to_string());
890 }
891 if form.csv.trim().is_empty() {
892 return Err("回灌解析: csv 空".to_string());
893 }
894 let mut bars = Vec::new();
895 for (i, raw) in form.csv.lines().enumerate() {
896 let line = raw.trim();
897 if line.is_empty() || line.starts_with('#') {
898 continue;
899 }
900 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
901 if parts.len() != 6 {
902 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
903 }
904 let num = |s: &str| -> Result<f64, String> {
905 s.parse::<f64>()
906 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
907 };
908 let dt_s: i64 = parts[0]
909 .parse()
910 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
911 bars.push(BarData {
912 symbol: symbol.clone(),
913 datetime: dt_s * 1_000_000_000,
914 interval: Some(
915 ek_dm::wire::interval_of_seconds(interval)
916 .unwrap_or(ek_data::constant::Interval::MINUTE),
917 ),
918 open_price: num(parts[1])?,
919 high_price: num(parts[2])?,
920 low_price: num(parts[3])?,
921 close_price: num(parts[4])?,
922 volume: num(parts[5])?,
923 ..BarData::default()
924 });
925 }
926 if bars.is_empty() {
927 return Err("回灌解析: 无有效行".to_string());
928 }
929 Ok((interval, symbol, bars))
930}
931
932#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
937pub enum TopLeftTab {
938 #[default]
939 Orders,
940 Active,
941}
942
943impl TopLeftTab {
944 pub fn label(self) -> &'static str {
945 match self {
946 Self::Orders => "委托",
947 Self::Active => "未成交",
948 }
949 }
950}
951
952#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
954pub enum BottomRightTab {
955 #[default]
956 Account,
957 Logs,
958}
959
960impl BottomRightTab {
961 pub fn label(self) -> &'static str {
962 match self {
963 Self::Account => "资金",
964 Self::Logs => "日志",
965 }
966 }
967}
968
969#[derive(Debug, Clone, Copy, PartialEq)]
972pub struct DataSplit {
973 pub col: f32,
975 pub row: f32,
977}
978
979impl Default for DataSplit {
980 fn default() -> Self {
981 Self {
982 col: crate::splitters::DATA_SPLIT_DEFAULT,
983 row: crate::splitters::DATA_SPLIT_DEFAULT,
984 }
985 }
986}
987
988pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
991 ek_data::constant::Exchange::SHFE,
992 ek_data::constant::Exchange::DCE,
993 ek_data::constant::Exchange::CZCE,
994 ek_data::constant::Exchange::CFFEX,
995 ek_data::constant::Exchange::INE,
996 ek_data::constant::Exchange::GFEX,
997];
998
999#[derive(Debug, Default)]
1009pub struct TickStore {
1010 ticks: BTreeMap<String, ek_data::tick::TickData>,
1011}
1012
1013impl TickStore {
1014 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1017 self.ticks
1018 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1019 }
1020
1021 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1023 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1024 v.sort_by(|a, b| {
1025 (a.exchange.as_str().to_string(), a.symbol.clone())
1026 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1027 });
1028 v
1029 }
1030
1031 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1033 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1034 for t in self.rows() {
1035 let ex = t.exchange.as_str().to_string();
1036 match out.last_mut() {
1037 Some((e, v)) if *e == ex => v.push(t),
1038 _ => out.push((ex, vec![t])),
1039 }
1040 }
1041 out
1042 }
1043
1044 pub fn len(&self) -> usize {
1045 self.ticks.len()
1046 }
1047
1048 pub fn is_empty(&self) -> bool {
1049 self.ticks.is_empty()
1050 }
1051}
1052
1053#[derive(Debug, Clone, PartialEq)]
1055pub struct OrderForm {
1056 pub exchange: String,
1058 pub symbol: String,
1060 pub direction: ek_data::constant::Direction,
1061 pub offset: ek_data::constant::Offset,
1062 pub order_type: ek_data::constant::OrderType,
1063 pub price: String,
1065 pub volume: String,
1066 pub gateway: String,
1068 pub follow_price: bool,
1070 pub auto_split: bool,
1073}
1074
1075impl Default for OrderForm {
1076 fn default() -> Self {
1077 Self {
1078 exchange: "SHFE".to_string(),
1079 symbol: String::new(),
1080 direction: ek_data::constant::Direction::LONG,
1081 offset: ek_data::constant::Offset::OPEN,
1082 order_type: ek_data::constant::OrderType::LIMIT,
1083 price: String::new(),
1084 volume: "1".to_string(),
1085 gateway: String::new(),
1086 follow_price: true,
1087 auto_split: true,
1088 }
1089 }
1090}
1091
1092impl OrderForm {
1093 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1095 use std::str::FromStr;
1096 let symbol = self.symbol.trim();
1097 if symbol.is_empty() {
1098 return Err("下单: 代码为空".to_string());
1099 }
1100 if symbol.contains('.') {
1101 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1102 }
1103 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1104 .map_err(|e| format!("下单: {e}"))?;
1105 let price: f64 = self
1106 .price
1107 .trim()
1108 .parse()
1109 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1110 let volume: f64 = self
1111 .volume
1112 .trim()
1113 .parse()
1114 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1115 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1116 return Err("下单: 限价单价格须为正".to_string());
1117 }
1118 if volume <= 0.0 {
1119 return Err("下单: 数量须为正".to_string());
1120 }
1121 Ok(ek_data::order_request::OrderRequest {
1122 symbol: symbol.to_string(),
1123 exchange,
1124 direction: self.direction,
1125 order_type: self.order_type,
1126 volume,
1127 price,
1128 offset: self.offset,
1129 ..ek_data::order_request::OrderRequest::default()
1130 })
1131 }
1132}
1133
1134#[derive(Debug, Default)]
1140pub struct ContractStore {
1141 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1142}
1143
1144impl ContractStore {
1145 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1146 self.contracts
1147 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1148 }
1149
1150 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1152 let p = pattern.trim();
1153 self.contracts
1154 .values()
1155 .filter(|c| {
1156 p.is_empty()
1157 || c.symbol.contains(p)
1158 || c.exchange.as_str().eq_ignore_ascii_case(p)
1159 || c.name.contains(p)
1160 })
1161 .cloned()
1162 .collect()
1163 }
1164
1165 pub fn len(&self) -> usize {
1166 self.contracts.len()
1167 }
1168
1169 pub fn is_empty(&self) -> bool {
1170 self.contracts.is_empty()
1171 }
1172}
1173
1174#[derive(Debug, Default)]
1176pub struct UiShared {
1177 logs: Mutex<LogRing>,
1178 snap: Mutex<Arc<BusSnap>>,
1179 market: MarketStore,
1180 ctp_form: Mutex<CtpForm>,
1181 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1184 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1185 trade_ids: Mutex<HashSet<String>>,
1193 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1194 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1195 trade_seq: AtomicU64,
1197 trade_dropped: AtomicU64,
1199 trade_dirty: AtomicBool,
1201 trade_snap: Mutex<Arc<TradeSnap>>,
1203 tick_store: Mutex<TickStore>,
1204 order_form: Mutex<OrderForm>,
1205 data_split: Mutex<DataSplit>,
1208 selected_exchange: Mutex<String>,
1210 topleft_tab: Mutex<TopLeftTab>,
1212 bottomright_tab: Mutex<BottomRightTab>,
1214 ops_window: AtomicBool,
1215 accounts_window: AtomicBool,
1218 contracts_window: AtomicBool,
1219 contract_store: Mutex<ContractStore>,
1220 contract_query: Mutex<String>,
1221 contract_selected: Mutex<Vec<String>>,
1222 kline_windows: Mutex<Vec<String>>,
1223 rev: AtomicU64,
1224 stop: AtomicBool,
1226 cmds: Mutex<VecDeque<UiCmd>>,
1228 pending_confirm: Mutex<Option<String>>,
1230 accounts: Mutex<Vec<AccountRow>>,
1232 account_edit_window: AtomicBool,
1234 editing_account: Mutex<Option<String>>,
1236 edit_alias: Mutex<String>,
1238 edit_error: Mutex<Option<String>>,
1240 md_source: Mutex<String>,
1242 subscriptions: Mutex<Vec<String>>,
1244 row_sel: Mutex<BTreeMap<String, String>>,
1247 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1251 status_cursor: AtomicUsize,
1257 status_seed: Mutex<Vec<String>>,
1260 list_tick: AtomicU32,
1262 list_due: AtomicBool,
1266 seed_due: AtomicBool,
1270 pipeline_due: AtomicBool,
1272 cfg_due: AtomicBool,
1274 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1277 cfg_delete_arm: Mutex<Option<String>>,
1279 theme: AtomicU8,
1283 theme_persist_due: AtomicBool,
1286 bad_payload: Mutex<BTreeMap<String, u64>>,
1288 pipeline_lines: Mutex<Vec<String>>,
1291 status_thresholds: Mutex<Option<(u32, i64)>>,
1294 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1296 toasts: crate::toast::ToastsCell,
1299 alerted: Mutex<std::collections::BTreeSet<String>>,
1301 exit_confirm_open: AtomicBool,
1303 exit_confirmed: AtomicBool,
1305}
1306
1307impl UiShared {
1308 pub fn new() -> Self {
1310 Self::default()
1311 }
1312
1313 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1317 self.tick_store.lock().unwrap().put(t);
1318 self.bump();
1319 }
1320
1321 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1323 self.tick_store.lock().unwrap().rows()
1324 }
1325
1326 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1328 self.tick_store.lock().unwrap().groups()
1329 }
1330
1331 pub fn tick_count(&self) -> usize {
1333 self.tick_store.lock().unwrap().len()
1334 }
1335
1336 pub fn order_form(&self) -> OrderForm {
1338 self.order_form.lock().unwrap().clone()
1339 }
1340
1341 pub fn set_order_form(&self, f: OrderForm) {
1343 *self.order_form.lock().unwrap() = f;
1344 self.bump();
1345 }
1346
1347 pub fn selected_exchange(&self) -> String {
1351 let s = self.selected_exchange.lock().unwrap().clone();
1352 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1353 s
1354 } else {
1355 MARKET_ORDER[0].as_str().to_string()
1356 }
1357 }
1358
1359 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1361 *self.selected_exchange.lock().unwrap() = s.into();
1362 self.bump();
1363 }
1364
1365 pub fn topleft_tab(&self) -> TopLeftTab {
1367 *self.topleft_tab.lock().unwrap()
1368 }
1369
1370 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1372 *self.topleft_tab.lock().unwrap() = t;
1373 self.bump();
1374 }
1375
1376 pub fn bottomright_tab(&self) -> BottomRightTab {
1378 *self.bottomright_tab.lock().unwrap()
1379 }
1380
1381 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1383 *self.bottomright_tab.lock().unwrap() = t;
1384 self.bump();
1385 }
1386
1387 pub fn data_split(&self) -> DataSplit {
1389 *self.data_split.lock().unwrap()
1390 }
1391
1392 pub fn set_data_split(&self, s: DataSplit) {
1395 let s = DataSplit {
1396 col: crate::splitters::clamp_frac(s.col),
1397 row: crate::splitters::clamp_frac(s.row),
1398 };
1399 let mut g = self.data_split.lock().unwrap();
1400 if g.col != s.col || g.row != s.row {
1401 *g = s;
1402 self.bump();
1403 }
1404 }
1405
1406 pub fn reset_layout(&self) {
1410 *self.data_split.lock().unwrap() = DataSplit::default();
1411 *self.selected_exchange.lock().unwrap() = String::new();
1412 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1413 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1414 self.bump();
1415 }
1416
1417 pub fn ops_window_open(&self) -> bool {
1421 self.ops_window.load(Ordering::Acquire)
1422 }
1423
1424 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1427 self.table_sorts
1428 .lock()
1429 .unwrap()
1430 .get(table)
1431 .copied()
1432 .flatten()
1433 }
1434
1435 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1437 let mut m = self.table_sorts.lock().unwrap();
1438 let asc = match m.get(table).copied().flatten() {
1439 Some((c, a)) if c == col => !a,
1440 _ => true,
1441 };
1442 m.insert(table.to_string(), Some((col, asc)));
1443 }
1444
1445 pub fn set_ops_window_open(&self, v: bool) {
1446 self.ops_window.store(v, Ordering::Release);
1447 self.bump();
1448 }
1449
1450 pub fn accounts_window_open(&self) -> bool {
1454 self.accounts_window.load(Ordering::Acquire)
1455 }
1456
1457 pub fn set_accounts_window_open(&self, v: bool) {
1459 self.accounts_window.store(v, Ordering::Release);
1460 self.bump();
1461 }
1462
1463 pub fn account_edit_open(&self) -> bool {
1465 self.account_edit_window.load(Ordering::Acquire)
1466 }
1467
1468 pub fn editing_account(&self) -> Option<String> {
1470 self.editing_account.lock().unwrap().clone()
1471 }
1472
1473 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1477 match gateway {
1478 Some(gw) => {
1479 let Some(row) = self.account(gw) else {
1480 return false;
1481 };
1482 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1483 *self.edit_alias.lock().unwrap() = row.alias.clone();
1484 self.set_ctp_form(row.form.clone());
1485 }
1486 None => {
1487 *self.editing_account.lock().unwrap() = None;
1488 *self.edit_alias.lock().unwrap() = String::new();
1489 let draft = self.draft_from_existing();
1490 self.set_ctp_form(draft);
1491 }
1492 }
1493 *self.edit_error.lock().unwrap() = None;
1494 self.account_edit_window.store(true, Ordering::Release);
1495 self.bump();
1496 true
1497 }
1498
1499 pub fn end_account_edit(&self) {
1501 *self.editing_account.lock().unwrap() = None;
1502 *self.edit_alias.lock().unwrap() = String::new();
1503 *self.edit_error.lock().unwrap() = None;
1504 self.account_edit_window.store(false, Ordering::Release);
1505 let draft = self.draft_from_existing();
1506 self.set_ctp_form(draft);
1507 self.bump();
1508 }
1509
1510 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1516 if self.editing_account().as_deref() == Some(gateway) {
1517 self.end_account_edit();
1518 }
1519 }
1520
1521 pub fn edit_alias(&self) -> String {
1523 self.edit_alias.lock().unwrap().clone()
1524 }
1525
1526 pub fn set_edit_alias(&self, v: impl Into<String>) {
1528 *self.edit_alias.lock().unwrap() = v.into();
1529 self.bump();
1530 }
1531
1532 pub fn edit_error(&self) -> Option<String> {
1534 self.edit_error.lock().unwrap().clone()
1535 }
1536
1537 pub fn set_edit_error(&self, e: Option<String>) {
1539 *self.edit_error.lock().unwrap() = e;
1540 self.bump();
1541 }
1542
1543 pub fn account_edit_dirty(&self) -> bool {
1545 let editing = self.editing_account();
1546 let draft = self.ctp_form();
1547 match editing {
1548 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1549 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1550 }
1551 }
1552
1553 pub fn accounts(&self) -> Vec<AccountRow> {
1555 self.accounts.lock().unwrap().clone()
1556 }
1557
1558 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1560 self.accounts
1561 .lock()
1562 .unwrap()
1563 .iter()
1564 .find(|a| a.gateway == gateway)
1565 .cloned()
1566 }
1567
1568 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1574 let mut acc = self.accounts.lock().unwrap();
1575 if acc.iter().any(|a| a.gateway == row.gateway) {
1576 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1577 }
1578 let alias = row.alias.trim();
1579 if !alias.is_empty()
1580 && acc
1581 .iter()
1582 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1583 {
1584 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1585 }
1586 if acc.len() >= MAX_ACCOUNTS {
1587 return Err(format!(
1588 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1589 ));
1590 }
1591 acc.push(row);
1592 drop(acc);
1593 self.bump();
1594 Ok(())
1595 }
1596
1597 #[must_use]
1599 pub fn can_add_account(&self) -> bool {
1600 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1601 }
1602
1603 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1609 let alias = alias.trim().to_string();
1610 let mut acc = self.accounts.lock().unwrap();
1611 if !alias.is_empty()
1612 && acc
1613 .iter()
1614 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1615 {
1616 return false;
1617 }
1618 match acc.iter_mut().find(|a| a.gateway == gateway) {
1619 Some(a) => {
1620 a.alias = alias;
1621 drop(acc);
1622 self.bump();
1623 true
1624 }
1625 None => false,
1626 }
1627 }
1628
1629 pub fn note_connect_sent(&self, gateway: &str) {
1634 let mut acc = self.accounts.lock().unwrap();
1635 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1636 a.stage = AccountStage::Assembling;
1637 a.fail_note = None;
1638 a.login = LoginState::Unknown;
1639 a.status_note = None;
1640 a.login_brief = None;
1641 a.chain_note = None;
1643 drop(acc);
1644 self.bump();
1645 }
1646 }
1647
1648 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1652 let mut acc = self.accounts.lock().unwrap();
1653 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1654 match result {
1655 Ok(()) => {
1656 a.stage = AccountStage::Assembled;
1657 a.fail_note = None;
1658 }
1659 Err(e) => {
1660 a.stage = AccountStage::AssembleFailed;
1661 a.fail_note = Some(e);
1662 }
1663 }
1664 drop(acc);
1665 self.bump();
1666 }
1667 }
1668
1669 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1673 let mut acc = self.accounts.lock().unwrap();
1674 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1675 a.effective_classes = classes.to_vec();
1676 drop(acc);
1677 self.bump();
1678 }
1679 }
1680
1681 pub fn note_chain(&self, gateway: &str, note: &str) {
1686 let mut acc = self.accounts.lock().unwrap();
1687 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1688 if a.chain_note.as_deref() == Some(note) {
1689 return; }
1691 a.chain_note = Some(note.to_string());
1692 drop(acc);
1693 self.bump();
1694 }
1695 }
1696
1697 #[must_use]
1701 pub fn status_thresholds(&self) -> (u32, i64) {
1702 self.status_thresholds
1703 .lock()
1704 .unwrap()
1705 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1706 }
1707
1708 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1710 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1711 self.bump();
1712 }
1713
1714 pub fn list_poll_due(&self) -> bool {
1719 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1720 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1721 }
1722
1723 pub fn request_list_refresh(&self) {
1725 self.list_due.store(true, Ordering::Release);
1726 }
1727
1728 pub fn request_trade_reseed(&self) {
1730 self.seed_due.store(true, Ordering::Release);
1731 }
1732
1733 pub fn take_trade_reseed(&self) -> bool {
1735 self.seed_due.swap(false, Ordering::AcqRel)
1736 }
1737
1738 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1743 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1747 for gw in present {
1748 if !uniq.iter().any(|u| u == gw) {
1749 uniq.push(gw.clone());
1750 }
1751 }
1752 let n = uniq.len();
1753 if n == 0 {
1754 return Vec::new();
1755 }
1756 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1757 let take = STATUS_ASKS_PER_TICK.min(n);
1758 let mut out = Vec::with_capacity(take);
1759 for i in 0..take {
1760 out.push(uniq[(start + i) % n].clone());
1761 }
1762 out
1763 }
1764
1765 pub fn request_status_seed(&self, gateway: &str) {
1767 let mut q = self.status_seed.lock().unwrap();
1768 if !q.iter().any(|g| g == gateway) {
1769 q.push(gateway.to_string());
1770 }
1771 }
1772
1773 pub fn take_status_seeds(&self) -> Vec<String> {
1775 let mut q = self.status_seed.lock().unwrap();
1776 std::mem::take(&mut *q)
1777 }
1778
1779 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
1789 let gw = ev.gateway.as_str();
1790 let text = ev.describe();
1791 self.log_line(format!("[INFO][{gw}] {text}"));
1792 let note = text
1793 .strip_prefix(CHAIN_STATE_PREFIX)
1794 .map(|s| s.trim().to_string())
1795 .unwrap_or_else(|| text.clone());
1796 self.note_chain(gw, ¬e);
1797 self.note_gateway_status(gw, &ev.status);
1798 let alarm_kind = matches!(
1799 ev.change,
1800 ek_ctp::GatewayChange::TdDisconnected
1801 | ek_ctp::GatewayChange::TdAuthFailed
1802 | ek_ctp::GatewayChange::TdLoginFailed
1803 | ek_ctp::GatewayChange::ContractQueryStopped
1804 | ek_ctp::GatewayChange::QueryRejected
1805 | ek_ctp::GatewayChange::MdDisconnected
1806 | ek_ctp::GatewayChange::MdLoginFailed
1807 | ek_ctp::GatewayChange::SubscribeRejected
1808 );
1809 if alarm_kind {
1810 let session = if ev.change.kind().starts_with("md-") {
1812 ev.status.md_disconnects
1813 } else {
1814 ev.status.td_disconnects
1815 };
1816 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
1817 if self.alert_once(key) {
1818 self.push_alert(
1819 crate::toast::AlertLevel::Warning,
1820 format!("账户 {gw}:{}", note.trim()),
1821 );
1822 }
1823 }
1824 }
1825
1826 pub fn request_pipeline_refresh(&self) {
1828 self.pipeline_due.store(true, Ordering::Release);
1829 }
1830 pub fn take_pipeline_due(&self) -> bool {
1832 self.pipeline_due.swap(false, Ordering::AcqRel)
1833 }
1834
1835 pub fn request_cfg_refresh(&self) {
1837 self.cfg_due.store(true, Ordering::Release);
1838 }
1839
1840 pub fn take_cfg_due(&self) -> bool {
1842 self.cfg_due.swap(false, Ordering::AcqRel)
1843 }
1844
1845 pub fn theme(&self) -> crate::tokens::Theme {
1847 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
1848 }
1849
1850 pub fn set_theme(&self, theme: crate::tokens::Theme) {
1853 self.theme.store(theme.as_u8(), Ordering::Release);
1854 self.bump();
1855 }
1856
1857 pub fn toggle_theme(&self) -> crate::tokens::Theme {
1859 let next = self.theme().flipped();
1860 self.theme.store(next.as_u8(), Ordering::Release);
1861 self.theme_persist_due.store(true, Ordering::Release);
1862 self.bump();
1863 next
1864 }
1865
1866 pub fn take_theme_persist_due(&self) -> bool {
1868 self.theme_persist_due.swap(false, Ordering::AcqRel)
1869 }
1870
1871 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
1873 *self.cfg_briefs.lock().unwrap() = Some(b);
1874 self.bump();
1875 }
1876 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
1878 self.cfg_briefs.lock().unwrap().clone()
1879 }
1880
1881 pub fn cfg_delete_armed(&self) -> Option<String> {
1883 self.cfg_delete_arm.lock().unwrap().clone()
1884 }
1885 pub fn arm_cfg_delete(&self, ns: Option<String>) {
1886 *self.cfg_delete_arm.lock().unwrap() = ns;
1887 self.bump();
1888 }
1889
1890 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
1892 *self.pipeline_lines.lock().unwrap() = lines;
1893 self.bump();
1894 }
1895
1896 pub fn pipeline_lines(&self) -> Vec<String> {
1898 self.pipeline_lines.lock().unwrap().clone()
1899 }
1900
1901 pub fn note_bad_payload(&self, line: &str) {
1907 let mut book = self.bad_payload.lock().unwrap();
1908 let n = book.entry(line.to_string()).or_insert(0);
1909 *n += 1;
1910 if *n == 1 {
1911 self.log_line(format!(
1912 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
1913 ));
1914 }
1915 }
1916
1917 #[must_use]
1919 pub fn bad_payload_count(&self, line: &str) -> u64 {
1920 self.bad_payload
1921 .lock()
1922 .unwrap()
1923 .get(line)
1924 .copied()
1925 .unwrap_or(0)
1926 }
1927
1928 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
1942 for r in rows {
1943 let (warn, crit) = pending_thresholds(r.soft_cap);
1944 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
1945 self.push_alert(
1946 crate::toast::AlertLevel::Warning,
1947 format!(
1948 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
1949 r.name, r.pending, r.peak, r.soft_cap
1950 ),
1951 );
1952 }
1953 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
1954 self.push_alert(
1955 crate::toast::AlertLevel::Error,
1956 format!(
1957 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
1958 r.name, r.pending, r.soft_cap
1959 ),
1960 );
1961 }
1962 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
1963 self.push_alert(
1964 crate::toast::AlertLevel::Error,
1965 format!(
1966 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
1967 r.name, r.soft_cap, r.spilled, r.peak
1968 ),
1969 );
1970 }
1971 }
1972 }
1973
1974 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
1982 let (limit, stale) = self.status_thresholds();
1983 let now = ek_ctp::map::now_epoch_ms();
1984 let mut alarm: Option<(LoginState, String)> = None;
1987 let mut found = false;
1988 {
1989 let mut acc = self.accounts.lock().unwrap();
1990 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1991 found = true;
1992 a.apply_status(st, now, limit, stale);
1993 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
1994 alarm = Some((
1995 a.login,
1996 a.status_note
1997 .clone()
1998 .unwrap_or_else(|| a.login.label().to_string()),
1999 ));
2000 }
2001 }
2002 }
2003 if found {
2004 self.bump();
2005 }
2006 if let Some((login, note)) = alarm
2007 && self.alert_once(format!("{gateway}|{}", login.label()))
2008 {
2009 self.push_alert(
2010 crate::toast::AlertLevel::Error,
2011 format!("账户 {gateway}:{note}"),
2012 );
2013 }
2014 }
2015
2016 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2020 let mut acc = self.accounts.lock().unwrap();
2021 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2022 a.login = LoginState::Unknown;
2023 a.status_note = Some(format!("状态口问不到:{why}"));
2024 a.login_brief = None;
2025 drop(acc);
2026 self.bump();
2027 }
2028 }
2029
2030 pub fn mark_account_stopped(&self, gateway: &str) {
2032 let mut acc = self.accounts.lock().unwrap();
2033 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2034 a.login = LoginState::Stopped;
2035 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2036 a.login_brief = None;
2037 drop(acc);
2038 self.bump();
2039 }
2040 }
2041
2042 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2045 let source = self.md_source();
2046 let mut f = form;
2047 if source.as_str() != gateway {
2048 f.md_addr.clear();
2049 f.md_user.clear();
2050 f.md_password.clear();
2051 }
2052 self.update_account_form(gateway, f.clone());
2053 f
2054 }
2055
2056 #[must_use]
2060 pub fn draft_from_existing(&self) -> CtpForm {
2061 let acc = self.accounts.lock().unwrap();
2062 let mut draft = CtpForm::default();
2063 if let Some(a) = acc.last() {
2064 draft.broker_id = a.form.broker_id.clone();
2065 draft.td_auth_code = a.form.td_auth_code.clone();
2066 draft.td_app_id = a.form.td_app_id.clone();
2067 draft.accept_futures = a.form.accept_futures;
2068 draft.accept_options = a.form.accept_options;
2069 }
2070 draft
2071 }
2072
2073 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2075 let mut acc = self.accounts.lock().unwrap();
2076 match acc.iter_mut().find(|a| a.gateway == gateway) {
2077 Some(a) => {
2078 a.form = form;
2079 drop(acc);
2080 self.bump();
2081 true
2082 }
2083 None => false,
2084 }
2085 }
2086
2087 pub fn remove_account(&self, gateway: &str) -> bool {
2089 let mut acc = self.accounts.lock().unwrap();
2090 let before = acc.len();
2091 acc.retain(|a| a.gateway != gateway);
2092 let removed = acc.len() != before;
2093 if removed {
2094 let mut md = self.md_source.lock().unwrap();
2096 if md.as_str() == gateway {
2097 md.clear();
2098 }
2099 drop(acc);
2100 self.bump();
2101 }
2102 removed
2103 }
2104
2105 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2114 let mut newly = Vec::new();
2115 {
2116 let mut acc = self.accounts.lock().unwrap();
2117 for a in acc.iter_mut() {
2118 let up = rows.iter().any(|r| r.name == a.gateway);
2119 if up && !a.connected {
2120 newly.push(a.gateway.clone());
2121 }
2122 a.connected = up;
2123 if up {
2124 a.stage = AccountStage::Assembled;
2125 a.fail_note = None;
2126 } else {
2127 a.chain_note = None;
2130 if a.stage == AccountStage::Assembled {
2131 a.stage = AccountStage::NotAssembled;
2132 a.login = LoginState::Unknown;
2133 a.status_note = None;
2134 a.login_brief = None;
2135 }
2136 }
2137 }
2138 }
2139 if !newly.is_empty() {
2140 self.bump();
2141 }
2142 newly
2143 }
2144
2145 pub fn md_source(&self) -> String {
2147 self.md_source.lock().unwrap().clone()
2148 }
2149
2150 pub fn accounts_locked(&self) -> bool {
2152 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2153 }
2154
2155 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2157 {
2158 let acc = self.accounts.lock().unwrap();
2159 if acc.iter().any(|a| a.connected) {
2160 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2161 }
2162 if !acc.iter().any(|a| a.gateway == gateway) {
2163 return Err(format!("账户 {gateway} 不在清册"));
2164 }
2165 }
2166 let mut acc = self.accounts.lock().unwrap();
2167 for a in acc.iter_mut() {
2168 a.md_source = a.gateway == gateway;
2169 }
2170 drop(acc);
2171 let mut md = self.md_source.lock().unwrap();
2172 *md = gateway.to_string();
2173 self.bump();
2174 Ok(())
2175 }
2176
2177 pub fn subscriptions(&self) -> Vec<String> {
2179 self.subscriptions.lock().unwrap().clone()
2180 }
2181
2182 pub fn note_subscriptions(&self, symbols: &[String]) {
2184 {
2185 let mut subs = self.subscriptions.lock().unwrap();
2186 for s in symbols {
2187 let s = s.trim();
2188 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2189 subs.push(s.to_string());
2190 }
2191 }
2192 }
2193 self.bump();
2194 }
2195
2196 pub fn default_gateway(&self) -> Option<String> {
2203 let chosen = {
2204 let f = self.order_form.lock().unwrap();
2205 f.gateway.trim().to_string()
2206 };
2207 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2208 return Some(chosen);
2209 }
2210 let acc = self.accounts.lock().unwrap();
2211 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2212 }
2213
2214 pub fn kline_symbols(&self) -> Vec<String> {
2216 self.kline_windows.lock().unwrap().clone()
2217 }
2218
2219 pub fn open_kline(&self, symbol: impl Into<String>) {
2221 let s = symbol.into();
2222 let mut v = self.kline_windows.lock().unwrap();
2223 if !v.contains(&s) {
2224 v.push(s.clone());
2225 }
2226 drop(v);
2227 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2228 }
2229
2230 pub fn close_kline(&self, symbol: &str) {
2232 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2233 self.bump();
2234 }
2235
2236 pub fn contracts_window_open(&self) -> bool {
2240 self.contracts_window.load(Ordering::Acquire)
2241 }
2242
2243 pub fn set_contracts_window_open(&self, v: bool) {
2245 self.contracts_window.store(v, Ordering::Release);
2246 self.bump();
2247 }
2248
2249 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2256 self.contract_store.lock().unwrap().put(c);
2257 if !c.name.is_empty() {
2258 self.backfill_names(&c.symbol, &c.name);
2259 }
2260 self.bump();
2261 }
2262
2263 fn backfill_names(&self, symbol: &str, name: &str) {
2268 let mut touched = false;
2269 {
2270 let mut rows = self.trade_orders.lock().unwrap();
2271 for (_, (_, o)) in rows.iter_mut() {
2272 if o.symbol == symbol && o.name.is_empty() {
2273 o.name = name.to_string();
2274 touched = true;
2275 }
2276 }
2277 }
2278 {
2279 let mut rows = self.trade_trades.lock().unwrap();
2280 for (_, t) in rows.iter_mut() {
2281 if t.symbol == symbol && t.name.is_empty() {
2282 t.name = name.to_string();
2283 touched = true;
2284 }
2285 }
2286 }
2287 {
2288 let mut rows = self.trade_positions.lock().unwrap();
2289 for (_, (_, p)) in rows.iter_mut() {
2290 if p.symbol == symbol && p.name.is_empty() {
2291 p.name = name.to_string();
2292 touched = true;
2293 }
2294 }
2295 }
2296 if touched {
2297 self.trade_dirty.store(true, Ordering::Release);
2299 }
2300 }
2301
2302 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2304 self.contract_store.lock().unwrap().query(pattern)
2305 }
2306
2307 pub fn contract_count(&self) -> usize {
2309 self.contract_store.lock().unwrap().len()
2310 }
2311
2312 fn contract_name_of(&self, symbol: &str) -> Option<String> {
2314 self.contract_store
2315 .lock()
2316 .unwrap()
2317 .query(symbol)
2318 .into_iter()
2319 .find(|c| c.symbol == symbol && !c.name.is_empty())
2320 .map(|c| c.name)
2321 }
2322
2323 pub fn contract_query(&self) -> String {
2325 self.contract_query.lock().unwrap().clone()
2326 }
2327
2328 pub fn set_contract_query(&self, q: impl Into<String>) {
2330 *self.contract_query.lock().unwrap() = q.into();
2331 self.bump();
2332 }
2333
2334 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2336 let mut v = self.contract_selected.lock().unwrap();
2337 match v.iter().position(|s| s == en_symbol) {
2338 Some(i) => {
2339 v.remove(i);
2340 }
2341 None => v.push(en_symbol.to_string()),
2342 }
2343 self.bump();
2344 }
2345
2346 pub fn selected_contracts(&self) -> Vec<String> {
2348 self.contract_selected.lock().unwrap().clone()
2349 }
2350
2351 #[must_use]
2357 pub fn cancel_reject_note(msg: &str) -> String {
2358 let head = msg.split(' ').next().unwrap_or(msg);
2359 let note = if head.starts_with("unknown:order") {
2360 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
2361 } else if head.starts_with("mismatch:gateway") {
2362 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
2363 } else if head.starts_with("terminal:status") {
2364 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
2365 } else if head.starts_with("limit:cancel-guard") {
2366 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
2367 } else {
2368 ""
2369 };
2370 if note.is_empty() {
2371 format!("撤单被拒:{msg}")
2373 } else {
2374 format!("撤单被拒:{note}(机器码:{msg})")
2375 }
2376 }
2377
2378 pub fn log_line(&self, line: impl Into<String>) {
2380 self.logs.lock().unwrap().push(line);
2381 self.bump();
2382 }
2383
2384 pub fn clear_logs(&self) {
2386 self.logs.lock().unwrap().clear();
2387 self.bump();
2388 }
2389
2390 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2392 let r = self.logs.lock().unwrap();
2393 (r.snapshot(), r.spilled())
2394 }
2395
2396 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2398 let r = self.logs.lock().unwrap();
2399 (r.entries(), r.spilled())
2400 }
2401
2402 pub fn logs_len(&self) -> usize {
2404 self.logs.lock().unwrap().len()
2405 }
2406
2407 pub fn set_snapshot(&self, snap: BusSnap) {
2409 *self.snap.lock().unwrap() = Arc::new(snap);
2410 self.bump();
2411 }
2412
2413 pub fn snapshot(&self) -> Arc<BusSnap> {
2415 Arc::clone(&self.snap.lock().unwrap())
2416 }
2417
2418 pub fn rev(&self) -> u64 {
2420 self.rev.load(Ordering::Acquire)
2421 }
2422
2423 pub fn bump(&self) {
2425 self.rev.fetch_add(1, Ordering::AcqRel);
2426 }
2427
2428 pub fn request_stop(&self) {
2430 self.stop.store(true, Ordering::Release);
2431 }
2432
2433 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2436 self.pending_alerts
2437 .lock()
2438 .unwrap_or_else(std::sync::PoisonError::into_inner)
2439 .push((level, text.into()));
2440 }
2441
2442 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2444 let mut q = self
2445 .pending_alerts
2446 .lock()
2447 .unwrap_or_else(std::sync::PoisonError::into_inner);
2448 std::mem::take(&mut *q)
2449 }
2450
2451 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2453 self.toasts.lock()
2454 }
2455
2456 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2458 let key = key.into();
2459 self.alerted
2460 .lock()
2461 .unwrap_or_else(std::sync::PoisonError::into_inner)
2462 .insert(key)
2463 }
2464
2465 #[must_use]
2467 pub fn pending_alert_count(&self) -> usize {
2468 self.pending_alerts
2469 .lock()
2470 .unwrap_or_else(std::sync::PoisonError::into_inner)
2471 .len()
2472 }
2473
2474 #[must_use]
2476 pub fn alerted_count(&self) -> usize {
2477 self.alerted
2478 .lock()
2479 .unwrap_or_else(std::sync::PoisonError::into_inner)
2480 .len()
2481 }
2482
2483 pub fn exit_confirm_open(&self) -> bool {
2485 self.exit_confirm_open
2486 .load(std::sync::atomic::Ordering::Relaxed)
2487 }
2488
2489 pub fn open_exit_confirm(&self) {
2490 self.exit_confirm_open
2491 .store(true, std::sync::atomic::Ordering::Relaxed);
2492 }
2493
2494 pub fn close_exit_confirm(&self) {
2495 self.exit_confirm_open
2496 .store(false, std::sync::atomic::Ordering::Relaxed);
2497 }
2498
2499 pub fn exit_confirmed(&self) -> bool {
2500 self.exit_confirmed
2501 .load(std::sync::atomic::Ordering::Relaxed)
2502 }
2503
2504 pub fn confirm_exit(&self) {
2505 self.exit_confirmed
2506 .store(true, std::sync::atomic::Ordering::Relaxed);
2507 }
2508
2509 pub fn is_stop(&self) -> bool {
2510 self.stop.load(Ordering::Acquire)
2511 }
2512
2513 pub fn push_cmd(&self, cmd: UiCmd) {
2517 self.cmds.lock().unwrap().push_back(cmd);
2518 self.bump();
2519 }
2520
2521 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2523 self.cmds.lock().unwrap().drain(..).collect()
2524 }
2525
2526 #[must_use]
2528 pub fn cmd_pending(&self) -> bool {
2529 !self.cmds.lock().unwrap().is_empty()
2530 }
2531
2532 pub fn pending_confirm(&self) -> Option<String> {
2534 self.pending_confirm.lock().unwrap().clone()
2535 }
2536
2537 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2539 let mut g = self.pending_confirm.lock().unwrap();
2540 match g.take() {
2541 Some(p) if p == name => ConfirmStep::Fired,
2542 Some(_) => {
2543 *g = Some(name.to_string());
2544 ConfirmStep::Switched
2545 }
2546 None => {
2547 *g = Some(name.to_string());
2548 ConfirmStep::Armed
2549 }
2550 }
2551 }
2552
2553 pub fn confirm_cancel(&self) {
2555 *self.pending_confirm.lock().unwrap() = None;
2556 self.bump();
2557 }
2558
2559 pub fn market(&self) -> &MarketStore {
2563 &self.market
2564 }
2565
2566 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2575 let named = |symbol: &str, name: String| -> String {
2578 if name.is_empty() {
2579 self.contract_name_of(symbol).unwrap_or(name)
2580 } else {
2581 name
2582 }
2583 };
2584 {
2585 let mut book = self.trade_orders.lock().unwrap();
2586 book.clear();
2587 for mut o in ts.orders.into_iter().rev() {
2588 o.name = named(&o.symbol, o.name);
2589 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2590 book.insert(o.en_orderid(), (seq, o));
2591 }
2592 }
2593 {
2594 let mut book = self.trade_trades.lock().unwrap();
2595 book.clear();
2596 let mut ids = self.trade_ids.lock().unwrap();
2597 ids.clear();
2598 for mut t in ts.trades.into_iter().rev() {
2599 t.name = named(&t.symbol, t.name);
2600 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2601 ids.insert(t.en_tradeid());
2602 book.push_front((seq, t));
2603 }
2604 }
2605 {
2606 let mut book = self.trade_positions.lock().unwrap();
2607 book.clear();
2608 for mut p in ts.positions.into_iter().rev() {
2609 p.name = named(&p.symbol, p.name);
2610 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2611 book.insert(p.en_positionid(), (seq, p));
2612 }
2613 }
2614 {
2615 let mut book = self.trade_funds.lock().unwrap();
2616 book.clear();
2617 for a in ts.accounts.into_iter().rev() {
2618 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2619 book.insert(a.en_accountid(), (seq, a));
2620 }
2621 }
2622 self.trade_dirty.store(true, Ordering::Release);
2623 self.bump();
2624 }
2625
2626 pub fn note_order(&self, o: &ek_data::order::OrderData) {
2634 let key = o.en_orderid();
2635 {
2636 let mut book = self.trade_orders.lock().unwrap();
2637 match book.get_mut(&key) {
2638 Some((_, cur)) => {
2639 cur.name.clone_from(&o.name);
2640 cur.traded = o.traded;
2641 cur.traded_price = o.traded_price;
2642 cur.status = o.status;
2643 cur.status_msg.clone_from(&o.status_msg);
2644 cur.datetime = o.datetime;
2645 }
2646 None => {
2647 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2648 book.insert(key, (seq, o.clone()));
2649 }
2650 }
2651 }
2652 self.trade_dirty.store(true, Ordering::Release);
2653 self.bump();
2654 }
2655
2656 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
2664 let id = t.en_tradeid();
2665 if self.trade_ids.lock().unwrap().contains(&id) {
2666 return;
2667 }
2668 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2669 let dropped = {
2670 let mut book = self.trade_trades.lock().unwrap();
2671 book.push_front((seq, t.clone()));
2672 self.trade_ids.lock().unwrap().insert(id);
2673 let over = book.len().saturating_sub(TRADE_CAP);
2674 for _ in 0..over {
2675 if let Some((_, old)) = book.pop_back() {
2677 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
2678 }
2679 }
2680 over
2681 };
2682 if dropped > 0 {
2683 let total = self
2684 .trade_dropped
2685 .fetch_add(dropped as u64, Ordering::AcqRel)
2686 + dropped as u64;
2687 if total == dropped as u64 {
2689 self.push_alert(
2690 crate::toast::AlertLevel::Warning,
2691 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
2692 );
2693 }
2694 }
2695 self.trade_dirty.store(true, Ordering::Release);
2696 self.bump();
2697 }
2698
2699 pub fn note_position(&self, p: &ek_data::position::PositionData) {
2704 let key = p.en_positionid();
2705 {
2706 let mut book = self.trade_positions.lock().unwrap();
2707 match book.get_mut(&key) {
2708 Some((_, cur)) => {
2709 cur.volume = p.volume;
2710 cur.yd_volume = p.yd_volume;
2711 cur.can_use_volume = p.can_use_volume;
2712 cur.frozen = p.frozen;
2713 cur.on_road_volume = p.on_road_volume;
2714 cur.market_value = p.market_value;
2715 cur.avg_price = p.avg_price;
2716 cur.price = p.price;
2717 cur.last_price = p.last_price;
2718 cur.pnl = p.pnl;
2719 cur.profit_rate = p.profit_rate;
2720 cur.name.clone_from(&p.name);
2721 }
2722 None => {
2723 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2724 book.insert(key, (seq, p.clone()));
2725 }
2726 }
2727 }
2728 self.trade_dirty.store(true, Ordering::Release);
2729 self.bump();
2730 }
2731
2732 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
2734 let key = a.en_accountid();
2735 {
2736 let mut book = self.trade_funds.lock().unwrap();
2737 match book.get_mut(&key) {
2738 Some((_, cur)) => {
2739 cur.balance = a.balance;
2740 cur.frozen = a.frozen;
2741 }
2742 None => {
2743 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2744 book.insert(key, (seq, a.clone()));
2745 }
2746 }
2747 }
2748 self.trade_dirty.store(true, Ordering::Release);
2749 self.bump();
2750 }
2751
2752 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
2754 let mut removed = 0usize;
2755 {
2756 let mut book = self.trade_orders.lock().unwrap();
2757 let before = book.len();
2758 book.retain(|_, (_, o)| o.gateway_name != gateway);
2759 removed += before - book.len();
2760 }
2761 {
2762 let mut book = self.trade_trades.lock().unwrap();
2763 let before = book.len();
2764 book.retain(|(_, t)| t.gateway_name != gateway);
2765 removed += before - book.len();
2766 let prefix = format!("{gateway}.");
2769 self.trade_ids
2770 .lock()
2771 .unwrap()
2772 .retain(|id| !id.starts_with(&prefix));
2773 }
2774 {
2775 let mut book = self.trade_positions.lock().unwrap();
2776 let before = book.len();
2777 book.retain(|_, (_, p)| p.gateway_name != gateway);
2778 removed += before - book.len();
2779 }
2780 {
2781 let mut book = self.trade_funds.lock().unwrap();
2782 let before = book.len();
2783 book.retain(|_, (_, a)| a.gateway_name != gateway);
2784 removed += before - book.len();
2785 }
2786 if removed > 0 {
2787 self.trade_dirty.store(true, Ordering::Release);
2788 self.bump();
2789 }
2790 removed
2791 }
2792
2793 #[must_use]
2795 pub fn trade_dropped(&self) -> u64 {
2796 self.trade_dropped.load(Ordering::Acquire)
2797 }
2798
2799 pub fn trade_snap(&self) -> Arc<TradeSnap> {
2804 let mut cache = self.trade_snap.lock().unwrap();
2805 if self.trade_dirty.swap(false, Ordering::AcqRel) {
2806 *cache = Arc::new(self.build_trade_snap());
2807 }
2808 Arc::clone(&cache)
2809 }
2810
2811 fn build_trade_snap(&self) -> TradeSnap {
2814 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
2815 where
2816 T: Clone,
2817 {
2818 let mut v: Vec<(u64, &String, &T)> =
2819 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
2820 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
2821 v.into_iter().map(|(_, _, v)| v.clone()).collect()
2822 }
2823 TradeSnap {
2824 orders: by_seq(&self.trade_orders.lock().unwrap()),
2825 trades: self
2826 .trade_trades
2827 .lock()
2828 .unwrap()
2829 .iter()
2830 .map(|(_, t)| t.clone())
2831 .collect(),
2832 positions: by_seq(&self.trade_positions.lock().unwrap()),
2833 accounts: by_seq(&self.trade_funds.lock().unwrap()),
2834 }
2835 }
2836
2837 pub fn ctp_form(&self) -> CtpForm {
2841 self.ctp_form.lock().unwrap().clone()
2842 }
2843
2844 pub fn set_ctp_form(&self, f: CtpForm) {
2846 *self.ctp_form.lock().unwrap() = f;
2847 self.bump();
2848 }
2849
2850 pub fn row_selected(&self, table: &str, key: &str) -> bool {
2854 self.row_sel
2855 .lock()
2856 .unwrap()
2857 .get(table)
2858 .is_some_and(|k| k == key)
2859 }
2860
2861 pub fn select_row(&self, table: &str, key: &str) {
2863 self.row_sel
2864 .lock()
2865 .unwrap()
2866 .insert(table.to_string(), key.to_string());
2867 }
2868}