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 restored: bool,
654 pub auto_connect: bool,
659 pub password_ref: bool,
664 pub stage: AccountStage,
666 pub fail_note: Option<String>,
668 pub effective_classes: Vec<String>,
673 pub connected: bool,
676 pub login: LoginState,
680 pub status_note: Option<String>,
683 pub login_brief: Option<String>,
690 pub chain_note: Option<String>,
698}
699
700impl AccountRow {
701 pub fn derive(form: &CtpForm) -> Result<Self, String> {
706 let broker = form.broker_id.trim();
707 let account = form.td_user.trim();
708 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
709 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
710 })?;
711 Ok(Self {
712 gateway,
713 alias: String::new(),
714 broker_id: broker.to_string(),
715 account: account.to_string(),
716 md_source: false,
717 form: form.clone(),
718 stage: AccountStage::NotAssembled,
719 fail_note: None,
720 effective_classes: Vec::new(),
721 connected: false,
722 login: LoginState::Unknown,
723 status_note: None,
724 login_brief: None,
725 chain_note: None,
726 restored: false,
729 auto_connect: false,
730 password_ref: false,
731 })
732 }
733
734 #[must_use]
740 pub fn status_label(&self) -> String {
741 if self.stage == AccountStage::Assembled {
742 match self.login {
743 LoginState::Unknown => "已装配·登录态未知".to_string(),
744 other => format!("已装配·{}", other.label()),
745 }
746 } else {
747 self.stage.label().to_string()
748 }
749 }
750
751 pub fn apply_status(
764 &mut self,
765 st: &ek_ctp::plugin::GatewayStatus,
766 now_ms: i64,
767 reject_limit: u32,
768 stale_secs: i64,
769 ) {
770 let td_side = !self.form.td_addr.trim().is_empty();
771 let (login, note) = if td_side {
772 if st.td_login_failed && !st.td_logged_in {
773 (
774 LoginState::LoginFailed,
775 st.td_last_error
776 .clone()
777 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
778 )
779 } else if st.td_logged_in {
780 if st.td_query_reject_streak >= reject_limit {
781 (
782 LoginState::QueryStarved,
783 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
784 )
785 } else if let Some(last) = st.td_last_account_ms {
786 let idle_ms = now_ms.saturating_sub(last);
787 if idle_ms > stale_secs.saturating_mul(1_000) {
788 (
789 LoginState::QueryStarved,
790 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
791 )
792 } else {
793 (LoginState::Online, count_note("掉线", st.td_disconnects))
794 }
795 } else {
796 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
803 match since_login_ms {
804 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
805 LoginState::QueryStarved,
806 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
807 ),
808 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
809 }
810 }
811 } else if self.login == LoginState::Stopped {
812 let extra = if st.td_disconnects > 0 {
813 format!("(此前掉线 {} 次)", st.td_disconnects)
814 } else {
815 String::new()
816 };
817 (
818 LoginState::Stopped,
819 Some(format!("会话已停{extra};『连接』可恢复")),
820 )
821 } else if st.td_disconnects > 0 {
822 (
823 LoginState::FrontDown,
824 Some(format!("掉线 {} 次", st.td_disconnects)),
825 )
826 } else {
827 (LoginState::Unknown, st.td_last_error.clone())
828 }
829 } else if st.md_logged_in {
830 (
831 LoginState::Online,
832 count_note("掉线", st.md_disconnects)
833 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
834 )
835 } else if self.login == LoginState::Stopped {
836 (
837 LoginState::Stopped,
838 Some("会话已停;『连接』可恢复".to_string()),
839 )
840 } else if st.md_disconnects > 0 {
841 (
842 LoginState::FrontDown,
843 Some(format!("掉线 {} 次", st.md_disconnects)),
844 )
845 } else {
846 (LoginState::Unknown, st.md_last_error.clone())
847 };
848 let note = if td_side && st.td_instrument_rejects > 0 {
853 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
854 match note {
855 Some(base) if base.contains("合约查询被拒") => Some(base),
857 Some(base) => Some(format!("{base};{add}")),
858 None => Some(add),
859 }
860 } else {
861 note
862 };
863 self.login = login;
864 self.status_note = note;
865 self.login_brief = if login.is_alarm() {
868 st.td_last_error
869 .as_deref()
870 .or(st.md_last_error.as_deref())
871 .and_then(brief_counter_reason)
872 } else {
873 None
874 };
875 }
876
877 #[must_use]
879 pub fn display_name(&self) -> String {
880 let a = self.alias.trim();
881 if a.is_empty() {
882 self.account.clone()
883 } else {
884 a.to_string()
885 }
886 }
887}
888
889#[derive(Debug, Clone, PartialEq, Default)]
892pub struct TradeSnap {
893 pub orders: Vec<ek_data::order::OrderData>,
894 pub trades: Vec<ek_data::trade::TradeData>,
895 pub positions: Vec<ek_data::position::PositionData>,
896 pub accounts: Vec<ek_data::account::AccountData>,
898}
899
900pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
903 let interval: i64 = form
904 .interval
905 .trim()
906 .parse()
907 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
908 let symbol = form.symbol.trim().to_string();
909 if symbol.is_empty() {
910 return Err("回灌解析: symbol 空".to_string());
911 }
912 if form.csv.trim().is_empty() {
913 return Err("回灌解析: csv 空".to_string());
914 }
915 let mut bars = Vec::new();
916 for (i, raw) in form.csv.lines().enumerate() {
917 let line = raw.trim();
918 if line.is_empty() || line.starts_with('#') {
919 continue;
920 }
921 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
922 if parts.len() != 6 {
923 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
924 }
925 let num = |s: &str| -> Result<f64, String> {
926 s.parse::<f64>()
927 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
928 };
929 let dt_s: i64 = parts[0]
930 .parse()
931 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
932 bars.push(BarData {
933 symbol: symbol.clone(),
934 datetime: dt_s * 1_000_000_000,
935 interval: Some(
936 ek_dm::wire::interval_of_seconds(interval)
937 .unwrap_or(ek_data::constant::Interval::MINUTE),
938 ),
939 open_price: num(parts[1])?,
940 high_price: num(parts[2])?,
941 low_price: num(parts[3])?,
942 close_price: num(parts[4])?,
943 volume: num(parts[5])?,
944 ..BarData::default()
945 });
946 }
947 if bars.is_empty() {
948 return Err("回灌解析: 无有效行".to_string());
949 }
950 Ok((interval, symbol, bars))
951}
952
953#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
958pub enum TopLeftTab {
959 #[default]
960 Orders,
961 Active,
962}
963
964impl TopLeftTab {
965 pub fn label(self) -> &'static str {
966 match self {
967 Self::Orders => "委托",
968 Self::Active => "未成交",
969 }
970 }
971}
972
973#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
975pub enum BottomRightTab {
976 #[default]
977 Account,
978 Logs,
979}
980
981impl BottomRightTab {
982 pub fn label(self) -> &'static str {
983 match self {
984 Self::Account => "资金",
985 Self::Logs => "日志",
986 }
987 }
988}
989
990#[derive(Debug, Clone, Copy, PartialEq)]
993pub struct DataSplit {
994 pub col: f32,
996 pub row: f32,
998}
999
1000impl Default for DataSplit {
1001 fn default() -> Self {
1002 Self {
1003 col: crate::splitters::DATA_SPLIT_DEFAULT,
1004 row: crate::splitters::DATA_SPLIT_DEFAULT,
1005 }
1006 }
1007}
1008
1009pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
1012 ek_data::constant::Exchange::SHFE,
1013 ek_data::constant::Exchange::DCE,
1014 ek_data::constant::Exchange::CZCE,
1015 ek_data::constant::Exchange::CFFEX,
1016 ek_data::constant::Exchange::INE,
1017 ek_data::constant::Exchange::GFEX,
1018];
1019
1020#[derive(Debug, Default)]
1030pub struct TickStore {
1031 ticks: BTreeMap<String, ek_data::tick::TickData>,
1032}
1033
1034impl TickStore {
1035 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1038 self.ticks
1039 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1040 }
1041
1042 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1044 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1045 v.sort_by(|a, b| {
1046 (a.exchange.as_str().to_string(), a.symbol.clone())
1047 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1048 });
1049 v
1050 }
1051
1052 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1054 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1055 for t in self.rows() {
1056 let ex = t.exchange.as_str().to_string();
1057 match out.last_mut() {
1058 Some((e, v)) if *e == ex => v.push(t),
1059 _ => out.push((ex, vec![t])),
1060 }
1061 }
1062 out
1063 }
1064
1065 pub fn len(&self) -> usize {
1066 self.ticks.len()
1067 }
1068
1069 pub fn is_empty(&self) -> bool {
1070 self.ticks.is_empty()
1071 }
1072}
1073
1074#[derive(Debug, Clone, PartialEq)]
1076pub struct OrderForm {
1077 pub exchange: String,
1079 pub symbol: String,
1081 pub direction: ek_data::constant::Direction,
1082 pub offset: ek_data::constant::Offset,
1083 pub order_type: ek_data::constant::OrderType,
1084 pub price: String,
1086 pub volume: String,
1087 pub gateway: String,
1089 pub follow_price: bool,
1091 pub auto_split: bool,
1094}
1095
1096impl Default for OrderForm {
1097 fn default() -> Self {
1098 Self {
1099 exchange: "SHFE".to_string(),
1100 symbol: String::new(),
1101 direction: ek_data::constant::Direction::LONG,
1102 offset: ek_data::constant::Offset::OPEN,
1103 order_type: ek_data::constant::OrderType::LIMIT,
1104 price: String::new(),
1105 volume: "1".to_string(),
1106 gateway: String::new(),
1107 follow_price: true,
1108 auto_split: true,
1109 }
1110 }
1111}
1112
1113impl OrderForm {
1114 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1116 use std::str::FromStr;
1117 let symbol = self.symbol.trim();
1118 if symbol.is_empty() {
1119 return Err("下单: 代码为空".to_string());
1120 }
1121 if symbol.contains('.') {
1122 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1123 }
1124 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1125 .map_err(|e| format!("下单: {e}"))?;
1126 let price: f64 = self
1127 .price
1128 .trim()
1129 .parse()
1130 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1131 let volume: f64 = self
1132 .volume
1133 .trim()
1134 .parse()
1135 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1136 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1137 return Err("下单: 限价单价格须为正".to_string());
1138 }
1139 if volume <= 0.0 {
1140 return Err("下单: 数量须为正".to_string());
1141 }
1142 Ok(ek_data::order_request::OrderRequest {
1143 symbol: symbol.to_string(),
1144 exchange,
1145 direction: self.direction,
1146 order_type: self.order_type,
1147 volume,
1148 price,
1149 offset: self.offset,
1150 ..ek_data::order_request::OrderRequest::default()
1151 })
1152 }
1153}
1154
1155#[derive(Debug, Default)]
1161pub struct ContractStore {
1162 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1163}
1164
1165impl ContractStore {
1166 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1167 self.contracts
1168 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1169 }
1170
1171 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1173 let p = pattern.trim();
1174 self.contracts
1175 .values()
1176 .filter(|c| {
1177 p.is_empty()
1178 || c.symbol.contains(p)
1179 || c.exchange.as_str().eq_ignore_ascii_case(p)
1180 || c.name.contains(p)
1181 })
1182 .cloned()
1183 .collect()
1184 }
1185
1186 pub fn len(&self) -> usize {
1187 self.contracts.len()
1188 }
1189
1190 pub fn is_empty(&self) -> bool {
1191 self.contracts.is_empty()
1192 }
1193}
1194
1195#[derive(Debug, Default)]
1197pub struct UiShared {
1198 logs: Mutex<LogRing>,
1199 snap: Mutex<Arc<BusSnap>>,
1200 market: MarketStore,
1201 ctp_form: Mutex<CtpForm>,
1202 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1205 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1206 trade_ids: Mutex<HashSet<String>>,
1214 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1215 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1216 trade_seq: AtomicU64,
1218 trade_dropped: AtomicU64,
1220 trade_dirty: AtomicBool,
1222 trade_snap: Mutex<Arc<TradeSnap>>,
1224 tick_store: Mutex<TickStore>,
1225 order_form: Mutex<OrderForm>,
1226 data_split: Mutex<DataSplit>,
1229 selected_exchange: Mutex<String>,
1231 topleft_tab: Mutex<TopLeftTab>,
1233 bottomright_tab: Mutex<BottomRightTab>,
1235 ops_window: AtomicBool,
1236 accounts_window: AtomicBool,
1239 contracts_window: AtomicBool,
1240 contract_store: Mutex<ContractStore>,
1241 contract_query: Mutex<String>,
1242 contract_selected: Mutex<Vec<String>>,
1243 kline_windows: Mutex<Vec<String>>,
1244 rev: AtomicU64,
1245 stop: AtomicBool,
1247 cmds: Mutex<VecDeque<UiCmd>>,
1249 pending_confirm: Mutex<Option<String>>,
1251 accounts: Mutex<Vec<AccountRow>>,
1253 account_edit_window: AtomicBool,
1255 editing_account: Mutex<Option<String>>,
1257 edit_alias: Mutex<String>,
1259 edit_error: Mutex<Option<String>>,
1261 md_source: Mutex<String>,
1263 subscriptions: Mutex<Vec<String>>,
1265 row_sel: Mutex<BTreeMap<String, String>>,
1268 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1272 status_cursor: AtomicUsize,
1278 status_seed: Mutex<Vec<String>>,
1281 list_tick: AtomicU32,
1283 list_due: AtomicBool,
1287 seed_due: AtomicBool,
1291 pipeline_due: AtomicBool,
1293 cfg_due: AtomicBool,
1295 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1298 cfg_delete_arm: Mutex<Option<String>>,
1300 theme: AtomicU8,
1304 theme_persist_due: AtomicBool,
1307 auto_connect_persist_due: AtomicBool,
1311 bad_payload: Mutex<BTreeMap<String, u64>>,
1313 pipeline_lines: Mutex<Vec<String>>,
1316 status_thresholds: Mutex<Option<(u32, i64)>>,
1319 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1321 toasts: crate::toast::ToastsCell,
1324 alerted: Mutex<std::collections::BTreeSet<String>>,
1326 exit_confirm_open: AtomicBool,
1328 exit_confirmed: AtomicBool,
1330 pluginwin_open: AtomicBool,
1332 bundle_open: AtomicBool,
1334 assembly_locked: AtomicBool,
1337}
1338
1339impl UiShared {
1340 pub fn new() -> Self {
1342 Self::default()
1343 }
1344
1345 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1349 self.tick_store.lock().unwrap().put(t);
1350 self.bump();
1351 }
1352
1353 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1355 self.tick_store.lock().unwrap().rows()
1356 }
1357
1358 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1360 self.tick_store.lock().unwrap().groups()
1361 }
1362
1363 pub fn tick_count(&self) -> usize {
1365 self.tick_store.lock().unwrap().len()
1366 }
1367
1368 pub fn order_form(&self) -> OrderForm {
1370 self.order_form.lock().unwrap().clone()
1371 }
1372
1373 pub fn set_order_form(&self, f: OrderForm) {
1375 *self.order_form.lock().unwrap() = f;
1376 self.bump();
1377 }
1378
1379 pub fn selected_exchange(&self) -> String {
1383 let s = self.selected_exchange.lock().unwrap().clone();
1384 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1385 s
1386 } else {
1387 MARKET_ORDER[0].as_str().to_string()
1388 }
1389 }
1390
1391 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1393 *self.selected_exchange.lock().unwrap() = s.into();
1394 self.bump();
1395 }
1396
1397 pub fn topleft_tab(&self) -> TopLeftTab {
1399 *self.topleft_tab.lock().unwrap()
1400 }
1401
1402 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1404 *self.topleft_tab.lock().unwrap() = t;
1405 self.bump();
1406 }
1407
1408 pub fn bottomright_tab(&self) -> BottomRightTab {
1410 *self.bottomright_tab.lock().unwrap()
1411 }
1412
1413 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1415 *self.bottomright_tab.lock().unwrap() = t;
1416 self.bump();
1417 }
1418
1419 pub fn data_split(&self) -> DataSplit {
1421 *self.data_split.lock().unwrap()
1422 }
1423
1424 pub fn set_data_split(&self, s: DataSplit) {
1427 let s = DataSplit {
1428 col: crate::splitters::clamp_frac(s.col),
1429 row: crate::splitters::clamp_frac(s.row),
1430 };
1431 let mut g = self.data_split.lock().unwrap();
1432 if g.col != s.col || g.row != s.row {
1433 *g = s;
1434 self.bump();
1435 }
1436 }
1437
1438 pub fn reset_layout(&self) {
1442 *self.data_split.lock().unwrap() = DataSplit::default();
1443 *self.selected_exchange.lock().unwrap() = String::new();
1444 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1445 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1446 self.bump();
1447 }
1448
1449 pub fn ops_window_open(&self) -> bool {
1453 self.ops_window.load(Ordering::Acquire)
1454 }
1455
1456 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1459 self.table_sorts
1460 .lock()
1461 .unwrap()
1462 .get(table)
1463 .copied()
1464 .flatten()
1465 }
1466
1467 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1469 let mut m = self.table_sorts.lock().unwrap();
1470 let asc = match m.get(table).copied().flatten() {
1471 Some((c, a)) if c == col => !a,
1472 _ => true,
1473 };
1474 m.insert(table.to_string(), Some((col, asc)));
1475 }
1476
1477 pub fn set_ops_window_open(&self, v: bool) {
1478 self.ops_window.store(v, Ordering::Release);
1479 self.bump();
1480 }
1481
1482 pub fn accounts_window_open(&self) -> bool {
1486 self.accounts_window.load(Ordering::Acquire)
1487 }
1488
1489 pub fn set_accounts_window_open(&self, v: bool) {
1491 self.accounts_window.store(v, Ordering::Release);
1492 self.bump();
1493 }
1494
1495 pub fn account_edit_open(&self) -> bool {
1497 self.account_edit_window.load(Ordering::Acquire)
1498 }
1499
1500 pub fn editing_account(&self) -> Option<String> {
1502 self.editing_account.lock().unwrap().clone()
1503 }
1504
1505 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1509 match gateway {
1510 Some(gw) => {
1511 let Some(row) = self.account(gw) else {
1512 return false;
1513 };
1514 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1515 *self.edit_alias.lock().unwrap() = row.alias.clone();
1516 self.set_ctp_form(row.form.clone());
1517 }
1518 None => {
1519 *self.editing_account.lock().unwrap() = None;
1520 *self.edit_alias.lock().unwrap() = String::new();
1521 let draft = self.draft_from_existing();
1522 self.set_ctp_form(draft);
1523 }
1524 }
1525 *self.edit_error.lock().unwrap() = None;
1526 self.account_edit_window.store(true, Ordering::Release);
1527 self.bump();
1528 true
1529 }
1530
1531 pub fn end_account_edit(&self) {
1533 *self.editing_account.lock().unwrap() = None;
1534 *self.edit_alias.lock().unwrap() = String::new();
1535 *self.edit_error.lock().unwrap() = None;
1536 self.account_edit_window.store(false, Ordering::Release);
1537 let draft = self.draft_from_existing();
1538 self.set_ctp_form(draft);
1539 self.bump();
1540 }
1541
1542 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1548 if self.editing_account().as_deref() == Some(gateway) {
1549 self.end_account_edit();
1550 }
1551 }
1552
1553 pub fn edit_alias(&self) -> String {
1555 self.edit_alias.lock().unwrap().clone()
1556 }
1557
1558 pub fn set_edit_alias(&self, v: impl Into<String>) {
1560 *self.edit_alias.lock().unwrap() = v.into();
1561 self.bump();
1562 }
1563
1564 pub fn edit_error(&self) -> Option<String> {
1566 self.edit_error.lock().unwrap().clone()
1567 }
1568
1569 pub fn set_edit_error(&self, e: Option<String>) {
1571 *self.edit_error.lock().unwrap() = e;
1572 self.bump();
1573 }
1574
1575 pub fn account_edit_dirty(&self) -> bool {
1577 let editing = self.editing_account();
1578 let draft = self.ctp_form();
1579 match editing {
1580 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1581 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1582 }
1583 }
1584
1585 pub fn accounts(&self) -> Vec<AccountRow> {
1587 self.accounts.lock().unwrap().clone()
1588 }
1589
1590 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1592 self.accounts
1593 .lock()
1594 .unwrap()
1595 .iter()
1596 .find(|a| a.gateway == gateway)
1597 .cloned()
1598 }
1599
1600 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1606 let mut acc = self.accounts.lock().unwrap();
1607 if acc.iter().any(|a| a.gateway == row.gateway) {
1608 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1609 }
1610 let alias = row.alias.trim();
1611 if !alias.is_empty()
1612 && acc
1613 .iter()
1614 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1615 {
1616 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1617 }
1618 if acc.len() >= MAX_ACCOUNTS {
1619 return Err(format!(
1620 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1621 ));
1622 }
1623 acc.push(row);
1624 drop(acc);
1625 self.bump();
1626 Ok(())
1627 }
1628
1629 #[must_use]
1631 pub fn can_add_account(&self) -> bool {
1632 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1633 }
1634
1635 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1641 let alias = alias.trim().to_string();
1642 let mut acc = self.accounts.lock().unwrap();
1643 if !alias.is_empty()
1644 && acc
1645 .iter()
1646 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1647 {
1648 return false;
1649 }
1650 match acc.iter_mut().find(|a| a.gateway == gateway) {
1651 Some(a) => {
1652 a.alias = alias;
1653 drop(acc);
1654 self.bump();
1655 true
1656 }
1657 None => false,
1658 }
1659 }
1660
1661 pub fn note_connect_sent(&self, gateway: &str) {
1666 let mut acc = self.accounts.lock().unwrap();
1667 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1668 a.stage = AccountStage::Assembling;
1669 a.fail_note = None;
1670 a.login = LoginState::Unknown;
1671 a.status_note = None;
1672 a.login_brief = None;
1673 a.chain_note = None;
1675 drop(acc);
1676 self.bump();
1677 }
1678 }
1679
1680 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1684 let mut acc = self.accounts.lock().unwrap();
1685 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1686 match result {
1687 Ok(()) => {
1688 a.stage = AccountStage::Assembled;
1689 a.fail_note = None;
1690 }
1691 Err(e) => {
1692 a.stage = AccountStage::AssembleFailed;
1693 a.fail_note = Some(e);
1694 }
1695 }
1696 drop(acc);
1697 self.bump();
1698 }
1699 }
1700
1701 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1705 let mut acc = self.accounts.lock().unwrap();
1706 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1707 a.effective_classes = classes.to_vec();
1708 drop(acc);
1709 self.bump();
1710 }
1711 }
1712
1713 pub fn note_chain(&self, gateway: &str, note: &str) {
1718 let mut acc = self.accounts.lock().unwrap();
1719 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1720 if a.chain_note.as_deref() == Some(note) {
1721 return; }
1723 a.chain_note = Some(note.to_string());
1724 drop(acc);
1725 self.bump();
1726 }
1727 }
1728
1729 #[must_use]
1733 pub fn status_thresholds(&self) -> (u32, i64) {
1734 self.status_thresholds
1735 .lock()
1736 .unwrap()
1737 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1738 }
1739
1740 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1742 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1743 self.bump();
1744 }
1745
1746 pub fn list_poll_due(&self) -> bool {
1751 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1752 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1753 }
1754
1755 pub fn request_list_refresh(&self) {
1757 self.list_due.store(true, Ordering::Release);
1758 }
1759
1760 pub fn request_trade_reseed(&self) {
1762 self.seed_due.store(true, Ordering::Release);
1763 }
1764
1765 pub fn take_trade_reseed(&self) -> bool {
1767 self.seed_due.swap(false, Ordering::AcqRel)
1768 }
1769
1770 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1775 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1779 for gw in present {
1780 if !uniq.iter().any(|u| u == gw) {
1781 uniq.push(gw.clone());
1782 }
1783 }
1784 let n = uniq.len();
1785 if n == 0 {
1786 return Vec::new();
1787 }
1788 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1789 let take = STATUS_ASKS_PER_TICK.min(n);
1790 let mut out = Vec::with_capacity(take);
1791 for i in 0..take {
1792 out.push(uniq[(start + i) % n].clone());
1793 }
1794 out
1795 }
1796
1797 pub fn request_status_seed(&self, gateway: &str) {
1799 let mut q = self.status_seed.lock().unwrap();
1800 if !q.iter().any(|g| g == gateway) {
1801 q.push(gateway.to_string());
1802 }
1803 }
1804
1805 pub fn take_status_seeds(&self) -> Vec<String> {
1807 let mut q = self.status_seed.lock().unwrap();
1808 std::mem::take(&mut *q)
1809 }
1810
1811 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
1821 let gw = ev.gateway.as_str();
1822 let text = ev.describe();
1823 self.log_line(format!("[INFO][{gw}] {text}"));
1824 let note = text
1825 .strip_prefix(CHAIN_STATE_PREFIX)
1826 .map(|s| s.trim().to_string())
1827 .unwrap_or_else(|| text.clone());
1828 self.note_chain(gw, ¬e);
1829 self.note_gateway_status(gw, &ev.status);
1830 let alarm_kind = matches!(
1831 ev.change,
1832 ek_ctp::GatewayChange::TdDisconnected
1833 | ek_ctp::GatewayChange::TdAuthFailed
1834 | ek_ctp::GatewayChange::TdLoginFailed
1835 | ek_ctp::GatewayChange::ContractQueryStopped
1836 | ek_ctp::GatewayChange::QueryRejected
1837 | ek_ctp::GatewayChange::MdDisconnected
1838 | ek_ctp::GatewayChange::MdLoginFailed
1839 | ek_ctp::GatewayChange::SubscribeRejected
1840 );
1841 if alarm_kind {
1842 let session = if ev.change.kind().starts_with("md-") {
1844 ev.status.md_disconnects
1845 } else {
1846 ev.status.td_disconnects
1847 };
1848 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
1849 if self.alert_once(key) {
1850 self.push_alert(
1851 crate::toast::AlertLevel::Warning,
1852 format!("账户 {gw}:{}", note.trim()),
1853 );
1854 }
1855 }
1856 }
1857
1858 pub fn request_pipeline_refresh(&self) {
1860 self.pipeline_due.store(true, Ordering::Release);
1861 }
1862 pub fn take_pipeline_due(&self) -> bool {
1864 self.pipeline_due.swap(false, Ordering::AcqRel)
1865 }
1866
1867 pub fn request_cfg_refresh(&self) {
1869 self.cfg_due.store(true, Ordering::Release);
1870 }
1871
1872 pub fn take_cfg_due(&self) -> bool {
1874 self.cfg_due.swap(false, Ordering::AcqRel)
1875 }
1876
1877 pub fn theme(&self) -> crate::tokens::Theme {
1879 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
1880 }
1881
1882 pub fn set_theme(&self, theme: crate::tokens::Theme) {
1885 self.theme.store(theme.as_u8(), Ordering::Release);
1886 self.bump();
1887 }
1888
1889 pub fn toggle_theme(&self) -> crate::tokens::Theme {
1891 let next = self.theme().flipped();
1892 self.theme.store(next.as_u8(), Ordering::Release);
1893 self.theme_persist_due.store(true, Ordering::Release);
1894 self.bump();
1895 next
1896 }
1897
1898 pub fn take_theme_persist_due(&self) -> bool {
1900 self.theme_persist_due.swap(false, Ordering::AcqRel)
1901 }
1902
1903 pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
1907 {
1908 let mut acc = self.accounts.lock().unwrap();
1909 match acc.iter_mut().find(|a| a.gateway == gateway) {
1910 Some(a) => {
1911 if a.auto_connect != on {
1912 a.auto_connect = on;
1913 self.auto_connect_persist_due.store(true, Ordering::Release);
1914 self.bump();
1915 }
1916 }
1917 None => return Err(format!("账户 {gateway} 不在清册")),
1918 }
1919 }
1920 Ok(())
1921 }
1922
1923 pub fn take_auto_connect_persist_due(&self) -> bool {
1925 self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
1926 }
1927
1928 pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
1943 let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
1945 match store.load(crate::plugin::THEME_NS) {
1946 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
1947 for (k, _f) in map.iter() {
1948 if let Some(gw) = k.strip_prefix("auto_connect.") {
1949 flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
1950 }
1951 }
1952 }
1953 Ok(_) => {} Err(_) => {} }
1956 let mut restored: Vec<String> = Vec::new();
1957 let ns_list = match store.list_namespaces() {
1958 Ok(v) => v,
1959 Err(_) => {
1960 let msg = "[ui] 档案清点失败:存储根不可读(具体见运行日志)".to_string();
1961 self.log_line(msg.clone());
1962 return restored;
1963 }
1964 };
1965 for ns in ns_list {
1966 let Some(gw) = ns.strip_prefix("ctp.") else {
1967 continue;
1968 };
1969 if self.account(gw).is_some() {
1970 self.log_line(format!(
1971 "[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
1972 ));
1973 continue;
1974 }
1975 let map = match store.load(&ns) {
1976 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
1977 Ok(_) => {
1978 self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
1979 continue;
1980 }
1981 Err(e) => {
1982 self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
1983 continue;
1984 }
1985 };
1986 let password_ref = map
1988 .get_secret_key("td_password")
1989 .is_some_and(|k| !k.is_empty());
1990 let mut form = CtpForm {
1991 broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
1992 td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
1993 td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
1994 td_password: String::new(),
1995 td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
1996 td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
1997 md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
1998 md_user: String::new(),
1999 md_password: String::new(),
2000 symbols: String::new(),
2001 accept_futures: false,
2002 accept_options: false,
2003 };
2004 for c in map
2007 .get_plain("product_classes")
2008 .unwrap_or_default()
2009 .split(',')
2010 {
2011 if c.trim() == "1" {
2012 form.accept_futures = true;
2013 }
2014 }
2015 let Ok(mut row) = AccountRow::derive(&form) else {
2016 self.log_line(format!(
2017 "[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
2018 ));
2019 continue;
2020 };
2021 row.restored = true;
2022 row.auto_connect = flags.get(gw).copied().unwrap_or(false);
2023 row.password_ref = password_ref;
2024 restored.push(row.gateway.clone());
2025 self.add_account(row).ok();
2026 }
2027 if !restored.is_empty() {
2028 self.log_line(format!(
2029 "[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
2030 restored.len()
2031 ));
2032 self.bump();
2033 }
2034 restored
2035 }
2036
2037 pub fn restore_saved_accounts(&self) -> Vec<String> {
2039 match ek_cfg::Store::eco() {
2040 Ok(store) => self.restore_saved_accounts_from(&store),
2041 Err(_) => {
2042 self.log_line(
2043 "[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
2044 );
2045 Vec::new()
2046 }
2047 }
2048 }
2049
2050 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
2052 *self.cfg_briefs.lock().unwrap() = Some(b);
2053 self.bump();
2054 }
2055 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
2057 self.cfg_briefs.lock().unwrap().clone()
2058 }
2059
2060 pub fn cfg_delete_armed(&self) -> Option<String> {
2062 self.cfg_delete_arm.lock().unwrap().clone()
2063 }
2064 pub fn arm_cfg_delete(&self, ns: Option<String>) {
2065 *self.cfg_delete_arm.lock().unwrap() = ns;
2066 self.bump();
2067 }
2068
2069 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
2071 *self.pipeline_lines.lock().unwrap() = lines;
2072 self.bump();
2073 }
2074
2075 pub fn pipeline_lines(&self) -> Vec<String> {
2077 self.pipeline_lines.lock().unwrap().clone()
2078 }
2079
2080 pub fn note_bad_payload(&self, line: &str) {
2086 let mut book = self.bad_payload.lock().unwrap();
2087 let n = book.entry(line.to_string()).or_insert(0);
2088 *n += 1;
2089 if *n == 1 {
2090 self.log_line(format!(
2091 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
2092 ));
2093 }
2094 }
2095
2096 #[must_use]
2098 pub fn bad_payload_count(&self, line: &str) -> u64 {
2099 self.bad_payload
2100 .lock()
2101 .unwrap()
2102 .get(line)
2103 .copied()
2104 .unwrap_or(0)
2105 }
2106
2107 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
2121 for r in rows {
2122 let (warn, crit) = pending_thresholds(r.soft_cap);
2123 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
2124 self.push_alert(
2125 crate::toast::AlertLevel::Warning,
2126 format!(
2127 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
2128 r.name, r.pending, r.peak, r.soft_cap
2129 ),
2130 );
2131 }
2132 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
2133 self.push_alert(
2134 crate::toast::AlertLevel::Error,
2135 format!(
2136 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
2137 r.name, r.pending, r.soft_cap
2138 ),
2139 );
2140 }
2141 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
2142 self.push_alert(
2143 crate::toast::AlertLevel::Error,
2144 format!(
2145 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
2146 r.name, r.soft_cap, r.spilled, r.peak
2147 ),
2148 );
2149 }
2150 }
2151 }
2152
2153 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
2161 let (limit, stale) = self.status_thresholds();
2162 let now = ek_ctp::map::now_epoch_ms();
2163 let mut alarm: Option<(LoginState, String)> = None;
2166 let mut found = false;
2167 {
2168 let mut acc = self.accounts.lock().unwrap();
2169 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2170 found = true;
2171 a.apply_status(st, now, limit, stale);
2172 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
2173 alarm = Some((
2174 a.login,
2175 a.status_note
2176 .clone()
2177 .unwrap_or_else(|| a.login.label().to_string()),
2178 ));
2179 }
2180 }
2181 }
2182 if found {
2183 self.bump();
2184 }
2185 if let Some((login, note)) = alarm
2186 && self.alert_once(format!("{gateway}|{}", login.label()))
2187 {
2188 self.push_alert(
2189 crate::toast::AlertLevel::Error,
2190 format!("账户 {gateway}:{note}"),
2191 );
2192 }
2193 }
2194
2195 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2199 let mut acc = self.accounts.lock().unwrap();
2200 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2201 a.login = LoginState::Unknown;
2202 a.status_note = Some(format!("状态口问不到:{why}"));
2203 a.login_brief = None;
2204 drop(acc);
2205 self.bump();
2206 }
2207 }
2208
2209 pub fn mark_account_stopped(&self, gateway: &str) {
2211 let mut acc = self.accounts.lock().unwrap();
2212 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2213 a.login = LoginState::Stopped;
2214 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2215 a.login_brief = None;
2216 drop(acc);
2217 self.bump();
2218 }
2219 }
2220
2221 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2224 let source = self.md_source();
2225 let mut f = form;
2226 if source.as_str() != gateway {
2227 f.md_addr.clear();
2228 f.md_user.clear();
2229 f.md_password.clear();
2230 }
2231 self.update_account_form(gateway, f.clone());
2232 f
2233 }
2234
2235 #[must_use]
2239 pub fn draft_from_existing(&self) -> CtpForm {
2240 let acc = self.accounts.lock().unwrap();
2241 let mut draft = CtpForm::default();
2242 if let Some(a) = acc.last() {
2243 draft.broker_id = a.form.broker_id.clone();
2244 draft.td_auth_code = a.form.td_auth_code.clone();
2245 draft.td_app_id = a.form.td_app_id.clone();
2246 draft.accept_futures = a.form.accept_futures;
2247 draft.accept_options = a.form.accept_options;
2248 }
2249 draft
2250 }
2251
2252 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2254 let mut acc = self.accounts.lock().unwrap();
2255 match acc.iter_mut().find(|a| a.gateway == gateway) {
2256 Some(a) => {
2257 a.form = form;
2258 drop(acc);
2259 self.bump();
2260 true
2261 }
2262 None => false,
2263 }
2264 }
2265
2266 pub fn remove_account(&self, gateway: &str) -> bool {
2268 let mut acc = self.accounts.lock().unwrap();
2269 let before = acc.len();
2270 acc.retain(|a| a.gateway != gateway);
2271 let removed = acc.len() != before;
2272 if removed {
2273 let mut md = self.md_source.lock().unwrap();
2275 if md.as_str() == gateway {
2276 md.clear();
2277 }
2278 drop(acc);
2279 self.bump();
2280 }
2281 removed
2282 }
2283
2284 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2293 let mut newly = Vec::new();
2294 {
2295 let mut acc = self.accounts.lock().unwrap();
2296 for a in acc.iter_mut() {
2297 let up = rows.iter().any(|r| r.name == a.gateway);
2298 if up && !a.connected {
2299 newly.push(a.gateway.clone());
2300 }
2301 a.connected = up;
2302 if up {
2303 a.stage = AccountStage::Assembled;
2304 a.fail_note = None;
2305 } else {
2306 a.chain_note = None;
2309 if a.stage == AccountStage::Assembled {
2310 a.stage = AccountStage::NotAssembled;
2311 a.login = LoginState::Unknown;
2312 a.status_note = None;
2313 a.login_brief = None;
2314 }
2315 }
2316 }
2317 }
2318 if !newly.is_empty() {
2319 self.bump();
2320 }
2321 newly
2322 }
2323
2324 pub fn md_source(&self) -> String {
2326 self.md_source.lock().unwrap().clone()
2327 }
2328
2329 pub fn accounts_locked(&self) -> bool {
2331 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2332 }
2333
2334 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2336 {
2337 let acc = self.accounts.lock().unwrap();
2338 if acc.iter().any(|a| a.connected) {
2339 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2340 }
2341 if !acc.iter().any(|a| a.gateway == gateway) {
2342 return Err(format!("账户 {gateway} 不在清册"));
2343 }
2344 }
2345 let mut acc = self.accounts.lock().unwrap();
2346 for a in acc.iter_mut() {
2347 a.md_source = a.gateway == gateway;
2348 }
2349 drop(acc);
2350 let mut md = self.md_source.lock().unwrap();
2351 *md = gateway.to_string();
2352 self.bump();
2353 Ok(())
2354 }
2355
2356 pub fn subscriptions(&self) -> Vec<String> {
2358 self.subscriptions.lock().unwrap().clone()
2359 }
2360
2361 pub fn note_subscriptions(&self, symbols: &[String]) {
2363 {
2364 let mut subs = self.subscriptions.lock().unwrap();
2365 for s in symbols {
2366 let s = s.trim();
2367 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2368 subs.push(s.to_string());
2369 }
2370 }
2371 }
2372 self.bump();
2373 }
2374
2375 pub fn default_gateway(&self) -> Option<String> {
2382 let chosen = {
2383 let f = self.order_form.lock().unwrap();
2384 f.gateway.trim().to_string()
2385 };
2386 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2387 return Some(chosen);
2388 }
2389 let acc = self.accounts.lock().unwrap();
2390 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2391 }
2392
2393 pub fn kline_symbols(&self) -> Vec<String> {
2395 self.kline_windows.lock().unwrap().clone()
2396 }
2397
2398 pub fn open_kline(&self, symbol: impl Into<String>) {
2400 let s = symbol.into();
2401 let mut v = self.kline_windows.lock().unwrap();
2402 if !v.contains(&s) {
2403 v.push(s.clone());
2404 }
2405 drop(v);
2406 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2407 }
2408
2409 pub fn close_kline(&self, symbol: &str) {
2411 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2412 self.bump();
2413 }
2414
2415 pub fn contracts_window_open(&self) -> bool {
2419 self.contracts_window.load(Ordering::Acquire)
2420 }
2421
2422 pub fn set_contracts_window_open(&self, v: bool) {
2424 self.contracts_window.store(v, Ordering::Release);
2425 self.bump();
2426 }
2427
2428 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2435 self.contract_store.lock().unwrap().put(c);
2436 if !c.name.is_empty() {
2437 self.backfill_names(&c.symbol, &c.name);
2438 }
2439 self.bump();
2440 }
2441
2442 fn backfill_names(&self, symbol: &str, name: &str) {
2447 let mut touched = false;
2448 {
2449 let mut rows = self.trade_orders.lock().unwrap();
2450 for (_, (_, o)) in rows.iter_mut() {
2451 if o.symbol == symbol && o.name.is_empty() {
2452 o.name = name.to_string();
2453 touched = true;
2454 }
2455 }
2456 }
2457 {
2458 let mut rows = self.trade_trades.lock().unwrap();
2459 for (_, t) in rows.iter_mut() {
2460 if t.symbol == symbol && t.name.is_empty() {
2461 t.name = name.to_string();
2462 touched = true;
2463 }
2464 }
2465 }
2466 {
2467 let mut rows = self.trade_positions.lock().unwrap();
2468 for (_, (_, p)) in rows.iter_mut() {
2469 if p.symbol == symbol && p.name.is_empty() {
2470 p.name = name.to_string();
2471 touched = true;
2472 }
2473 }
2474 }
2475 if touched {
2476 self.trade_dirty.store(true, Ordering::Release);
2478 }
2479 }
2480
2481 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2483 self.contract_store.lock().unwrap().query(pattern)
2484 }
2485
2486 pub fn contract_count(&self) -> usize {
2488 self.contract_store.lock().unwrap().len()
2489 }
2490
2491 fn contract_name_of(&self, symbol: &str) -> Option<String> {
2493 self.contract_store
2494 .lock()
2495 .unwrap()
2496 .query(symbol)
2497 .into_iter()
2498 .find(|c| c.symbol == symbol && !c.name.is_empty())
2499 .map(|c| c.name)
2500 }
2501
2502 pub fn contract_query(&self) -> String {
2504 self.contract_query.lock().unwrap().clone()
2505 }
2506
2507 pub fn set_contract_query(&self, q: impl Into<String>) {
2509 *self.contract_query.lock().unwrap() = q.into();
2510 self.bump();
2511 }
2512
2513 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2515 let mut v = self.contract_selected.lock().unwrap();
2516 match v.iter().position(|s| s == en_symbol) {
2517 Some(i) => {
2518 v.remove(i);
2519 }
2520 None => v.push(en_symbol.to_string()),
2521 }
2522 self.bump();
2523 }
2524
2525 pub fn selected_contracts(&self) -> Vec<String> {
2527 self.contract_selected.lock().unwrap().clone()
2528 }
2529
2530 #[must_use]
2536 pub fn cancel_reject_note(msg: &str) -> String {
2537 let head = msg.split(' ').next().unwrap_or(msg);
2538 let note = if head.starts_with("unknown:order") {
2539 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
2540 } else if head.starts_with("mismatch:gateway") {
2541 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
2542 } else if head.starts_with("terminal:status") {
2543 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
2544 } else if head.starts_with("limit:cancel-guard") {
2545 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
2546 } else {
2547 ""
2548 };
2549 if note.is_empty() {
2550 format!("撤单被拒:{msg}")
2552 } else {
2553 format!("撤单被拒:{note}(机器码:{msg})")
2554 }
2555 }
2556
2557 pub fn log_line(&self, line: impl Into<String>) {
2559 self.logs.lock().unwrap().push(line);
2560 self.bump();
2561 }
2562
2563 pub fn clear_logs(&self) {
2565 self.logs.lock().unwrap().clear();
2566 self.bump();
2567 }
2568
2569 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2571 let r = self.logs.lock().unwrap();
2572 (r.snapshot(), r.spilled())
2573 }
2574
2575 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2577 let r = self.logs.lock().unwrap();
2578 (r.entries(), r.spilled())
2579 }
2580
2581 pub fn logs_len(&self) -> usize {
2583 self.logs.lock().unwrap().len()
2584 }
2585
2586 pub fn set_snapshot(&self, snap: BusSnap) {
2588 *self.snap.lock().unwrap() = Arc::new(snap);
2589 self.bump();
2590 }
2591
2592 pub fn snapshot(&self) -> Arc<BusSnap> {
2594 Arc::clone(&self.snap.lock().unwrap())
2595 }
2596
2597 pub fn rev(&self) -> u64 {
2599 self.rev.load(Ordering::Acquire)
2600 }
2601
2602 pub fn bump(&self) {
2604 self.rev.fetch_add(1, Ordering::AcqRel);
2605 }
2606
2607 pub fn request_stop(&self) {
2609 self.stop.store(true, Ordering::Release);
2610 }
2611
2612 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2615 self.pending_alerts
2616 .lock()
2617 .unwrap_or_else(std::sync::PoisonError::into_inner)
2618 .push((level, text.into()));
2619 }
2620
2621 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2623 let mut q = self
2624 .pending_alerts
2625 .lock()
2626 .unwrap_or_else(std::sync::PoisonError::into_inner);
2627 std::mem::take(&mut *q)
2628 }
2629
2630 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2632 self.toasts.lock()
2633 }
2634
2635 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2637 let key = key.into();
2638 self.alerted
2639 .lock()
2640 .unwrap_or_else(std::sync::PoisonError::into_inner)
2641 .insert(key)
2642 }
2643
2644 #[must_use]
2646 pub fn pending_alert_count(&self) -> usize {
2647 self.pending_alerts
2648 .lock()
2649 .unwrap_or_else(std::sync::PoisonError::into_inner)
2650 .len()
2651 }
2652
2653 #[must_use]
2655 pub fn alerted_count(&self) -> usize {
2656 self.alerted
2657 .lock()
2658 .unwrap_or_else(std::sync::PoisonError::into_inner)
2659 .len()
2660 }
2661
2662 pub fn exit_confirm_open(&self) -> bool {
2664 self.exit_confirm_open
2665 .load(std::sync::atomic::Ordering::Relaxed)
2666 }
2667
2668 pub fn open_exit_confirm(&self) {
2669 self.exit_confirm_open
2670 .store(true, std::sync::atomic::Ordering::Relaxed);
2671 }
2672
2673 pub fn close_exit_confirm(&self) {
2674 self.exit_confirm_open
2675 .store(false, std::sync::atomic::Ordering::Relaxed);
2676 }
2677
2678 pub fn exit_confirmed(&self) -> bool {
2679 self.exit_confirmed
2680 .load(std::sync::atomic::Ordering::Relaxed)
2681 }
2682
2683 pub fn confirm_exit(&self) {
2684 self.exit_confirmed
2685 .store(true, std::sync::atomic::Ordering::Relaxed);
2686 }
2687
2688 pub fn is_stop(&self) -> bool {
2689 self.stop.load(Ordering::Acquire)
2690 }
2691
2692 #[must_use]
2694 pub fn pluginwin_open(&self) -> bool {
2695 self.pluginwin_open.load(Ordering::Relaxed)
2696 }
2697
2698 pub fn set_pluginwin_open(&self, on: bool) {
2700 self.pluginwin_open.store(on, Ordering::Relaxed);
2701 }
2702
2703 #[must_use]
2705 pub fn bundle_open(&self) -> bool {
2706 self.bundle_open.load(Ordering::Relaxed)
2707 }
2708
2709 pub fn set_bundle_open(&self, on: bool) {
2711 self.bundle_open.store(on, Ordering::Relaxed);
2712 }
2713
2714 #[must_use]
2716 pub fn assembly_locked(&self) -> bool {
2717 self.assembly_locked.load(Ordering::Relaxed)
2718 }
2719
2720 pub fn set_assembly_locked(&self, on: bool) {
2722 self.assembly_locked.store(on, Ordering::Relaxed);
2723 }
2724
2725 pub fn push_cmd(&self, cmd: UiCmd) {
2729 self.cmds.lock().unwrap().push_back(cmd);
2730 self.bump();
2731 }
2732
2733 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2735 self.cmds.lock().unwrap().drain(..).collect()
2736 }
2737
2738 #[must_use]
2740 pub fn cmd_pending(&self) -> bool {
2741 !self.cmds.lock().unwrap().is_empty()
2742 }
2743
2744 pub fn pending_confirm(&self) -> Option<String> {
2746 self.pending_confirm.lock().unwrap().clone()
2747 }
2748
2749 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2751 let mut g = self.pending_confirm.lock().unwrap();
2752 match g.take() {
2753 Some(p) if p == name => ConfirmStep::Fired,
2754 Some(_) => {
2755 *g = Some(name.to_string());
2756 ConfirmStep::Switched
2757 }
2758 None => {
2759 *g = Some(name.to_string());
2760 ConfirmStep::Armed
2761 }
2762 }
2763 }
2764
2765 pub fn confirm_cancel(&self) {
2767 *self.pending_confirm.lock().unwrap() = None;
2768 self.bump();
2769 }
2770
2771 pub fn market(&self) -> &MarketStore {
2775 &self.market
2776 }
2777
2778 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2787 let named = |symbol: &str, name: String| -> String {
2790 if name.is_empty() {
2791 self.contract_name_of(symbol).unwrap_or(name)
2792 } else {
2793 name
2794 }
2795 };
2796 {
2797 let mut book = self.trade_orders.lock().unwrap();
2798 book.clear();
2799 for mut o in ts.orders.into_iter().rev() {
2800 o.name = named(&o.symbol, o.name);
2801 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2802 book.insert(o.en_orderid(), (seq, o));
2803 }
2804 }
2805 {
2806 let mut book = self.trade_trades.lock().unwrap();
2807 book.clear();
2808 let mut ids = self.trade_ids.lock().unwrap();
2809 ids.clear();
2810 for mut t in ts.trades.into_iter().rev() {
2811 t.name = named(&t.symbol, t.name);
2812 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2813 ids.insert(t.en_tradeid());
2814 book.push_front((seq, t));
2815 }
2816 }
2817 {
2818 let mut book = self.trade_positions.lock().unwrap();
2819 book.clear();
2820 for mut p in ts.positions.into_iter().rev() {
2821 p.name = named(&p.symbol, p.name);
2822 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2823 book.insert(p.en_positionid(), (seq, p));
2824 }
2825 }
2826 {
2827 let mut book = self.trade_funds.lock().unwrap();
2828 book.clear();
2829 for a in ts.accounts.into_iter().rev() {
2830 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2831 book.insert(a.en_accountid(), (seq, a));
2832 }
2833 }
2834 self.trade_dirty.store(true, Ordering::Release);
2835 self.bump();
2836 }
2837
2838 pub fn note_order(&self, o: &ek_data::order::OrderData) {
2846 let key = o.en_orderid();
2847 {
2848 let mut book = self.trade_orders.lock().unwrap();
2849 match book.get_mut(&key) {
2850 Some((_, cur)) => {
2851 cur.name.clone_from(&o.name);
2852 cur.traded = o.traded;
2853 cur.traded_price = o.traded_price;
2854 cur.status = o.status;
2855 cur.status_msg.clone_from(&o.status_msg);
2856 cur.datetime = o.datetime;
2857 }
2858 None => {
2859 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2860 book.insert(key, (seq, o.clone()));
2861 }
2862 }
2863 }
2864 self.trade_dirty.store(true, Ordering::Release);
2865 self.bump();
2866 }
2867
2868 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
2876 let id = t.en_tradeid();
2877 if self.trade_ids.lock().unwrap().contains(&id) {
2878 return;
2879 }
2880 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2881 let dropped = {
2882 let mut book = self.trade_trades.lock().unwrap();
2883 book.push_front((seq, t.clone()));
2884 self.trade_ids.lock().unwrap().insert(id);
2885 let over = book.len().saturating_sub(TRADE_CAP);
2886 for _ in 0..over {
2887 if let Some((_, old)) = book.pop_back() {
2889 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
2890 }
2891 }
2892 over
2893 };
2894 if dropped > 0 {
2895 let total = self
2896 .trade_dropped
2897 .fetch_add(dropped as u64, Ordering::AcqRel)
2898 + dropped as u64;
2899 if total == dropped as u64 {
2901 self.push_alert(
2902 crate::toast::AlertLevel::Warning,
2903 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
2904 );
2905 }
2906 }
2907 self.trade_dirty.store(true, Ordering::Release);
2908 self.bump();
2909 }
2910
2911 pub fn note_position(&self, p: &ek_data::position::PositionData) {
2916 let key = p.en_positionid();
2917 {
2918 let mut book = self.trade_positions.lock().unwrap();
2919 match book.get_mut(&key) {
2920 Some((_, cur)) => {
2921 cur.volume = p.volume;
2922 cur.yd_volume = p.yd_volume;
2923 cur.can_use_volume = p.can_use_volume;
2924 cur.frozen = p.frozen;
2925 cur.on_road_volume = p.on_road_volume;
2926 cur.market_value = p.market_value;
2927 cur.avg_price = p.avg_price;
2928 cur.price = p.price;
2929 cur.last_price = p.last_price;
2930 cur.pnl = p.pnl;
2931 cur.profit_rate = p.profit_rate;
2932 cur.name.clone_from(&p.name);
2933 }
2934 None => {
2935 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2936 book.insert(key, (seq, p.clone()));
2937 }
2938 }
2939 }
2940 self.trade_dirty.store(true, Ordering::Release);
2941 self.bump();
2942 }
2943
2944 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
2946 let key = a.en_accountid();
2947 {
2948 let mut book = self.trade_funds.lock().unwrap();
2949 match book.get_mut(&key) {
2950 Some((_, cur)) => {
2951 cur.balance = a.balance;
2952 cur.frozen = a.frozen;
2953 }
2954 None => {
2955 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2956 book.insert(key, (seq, a.clone()));
2957 }
2958 }
2959 }
2960 self.trade_dirty.store(true, Ordering::Release);
2961 self.bump();
2962 }
2963
2964 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
2966 let mut removed = 0usize;
2967 {
2968 let mut book = self.trade_orders.lock().unwrap();
2969 let before = book.len();
2970 book.retain(|_, (_, o)| o.gateway_name != gateway);
2971 removed += before - book.len();
2972 }
2973 {
2974 let mut book = self.trade_trades.lock().unwrap();
2975 let before = book.len();
2976 book.retain(|(_, t)| t.gateway_name != gateway);
2977 removed += before - book.len();
2978 let prefix = format!("{gateway}.");
2981 self.trade_ids
2982 .lock()
2983 .unwrap()
2984 .retain(|id| !id.starts_with(&prefix));
2985 }
2986 {
2987 let mut book = self.trade_positions.lock().unwrap();
2988 let before = book.len();
2989 book.retain(|_, (_, p)| p.gateway_name != gateway);
2990 removed += before - book.len();
2991 }
2992 {
2993 let mut book = self.trade_funds.lock().unwrap();
2994 let before = book.len();
2995 book.retain(|_, (_, a)| a.gateway_name != gateway);
2996 removed += before - book.len();
2997 }
2998 if removed > 0 {
2999 self.trade_dirty.store(true, Ordering::Release);
3000 self.bump();
3001 }
3002 removed
3003 }
3004
3005 #[must_use]
3007 pub fn trade_dropped(&self) -> u64 {
3008 self.trade_dropped.load(Ordering::Acquire)
3009 }
3010
3011 pub fn trade_snap(&self) -> Arc<TradeSnap> {
3016 let mut cache = self.trade_snap.lock().unwrap();
3017 if self.trade_dirty.swap(false, Ordering::AcqRel) {
3018 *cache = Arc::new(self.build_trade_snap());
3019 }
3020 Arc::clone(&cache)
3021 }
3022
3023 fn build_trade_snap(&self) -> TradeSnap {
3026 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
3027 where
3028 T: Clone,
3029 {
3030 let mut v: Vec<(u64, &String, &T)> =
3031 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
3032 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
3033 v.into_iter().map(|(_, _, v)| v.clone()).collect()
3034 }
3035 TradeSnap {
3036 orders: by_seq(&self.trade_orders.lock().unwrap()),
3037 trades: self
3038 .trade_trades
3039 .lock()
3040 .unwrap()
3041 .iter()
3042 .map(|(_, t)| t.clone())
3043 .collect(),
3044 positions: by_seq(&self.trade_positions.lock().unwrap()),
3045 accounts: by_seq(&self.trade_funds.lock().unwrap()),
3046 }
3047 }
3048
3049 pub fn ctp_form(&self) -> CtpForm {
3053 self.ctp_form.lock().unwrap().clone()
3054 }
3055
3056 pub fn set_ctp_form(&self, f: CtpForm) {
3058 *self.ctp_form.lock().unwrap() = f;
3059 self.bump();
3060 }
3061
3062 pub fn row_selected(&self, table: &str, key: &str) -> bool {
3066 self.row_sel
3067 .lock()
3068 .unwrap()
3069 .get(table)
3070 .is_some_and(|k| k == key)
3071 }
3072
3073 pub fn select_row(&self, table: &str, key: &str) {
3075 self.row_sel
3076 .lock()
3077 .unwrap()
3078 .insert(table.to_string(), key.to_string());
3079 }
3080}