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, Clone)]
260pub struct KlineFetch {
261 pub symbol: String,
262 pub interval_secs: i64,
263}
264
265#[derive(Debug, Default)]
267pub struct MarketStore {
268 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
269 frames: AtomicU64,
271 updates: AtomicU64,
273 reloads: AtomicU64,
275}
276
277#[derive(Debug, Clone, PartialEq)]
279pub struct CellView {
280 pub symbol: String,
281 pub interval: i64,
282 pub bars: Vec<BarData>,
283}
284
285impl MarketStore {
286 pub fn on_frame(&self, frame: &BarsFrame) {
288 let mut cells = self.cells.lock().unwrap();
289 for b in &frame.bars {
290 let key = (b.symbol.clone(), frame.interval);
291 let cell = cells.entry(key).or_default();
292 match cell.last() {
293 Some(last) if b.datetime < last.datetime => continue,
294 Some(last) if b.datetime == last.datetime => {
295 let slot = cell.last_mut().expect("len>1");
296 *slot = b.clone();
297 }
298 _ => {
299 cell.push(b.clone());
300 if cell.len() > KLINE_CELL_CAP {
301 cell.remove(0);
302 }
303 }
304 }
305 }
306 self.frames.fetch_add(1, Ordering::AcqRel);
307 }
308
309 pub fn on_update(&self, b: &BarData) {
311 let mut cells = self.cells.lock().unwrap();
312 let iv = b
313 .interval
314 .unwrap_or(ek_data::constant::Interval::MINUTE)
315 .seconds();
316 let key = (b.symbol.clone(), iv);
317 let cell = cells.entry(key).or_default();
318 match cell.last() {
319 Some(last) if b.datetime == last.datetime => {
320 let slot = cell.last_mut().expect("len>1");
321 *slot = b.clone();
322 }
323 Some(last) if b.datetime < last.datetime => {}
324 _ => {
325 cell.push(b.clone());
326 if cell.len() > KLINE_CELL_CAP {
327 cell.remove(0);
328 }
329 }
330 }
331 self.updates.fetch_add(1, Ordering::AcqRel);
332 }
333
334 pub fn note_history(&self, symbol: &str, iv_secs: i64, bars: &[BarData]) {
340 let mut cells = self.cells.lock().unwrap();
341 let cell = cells.entry((symbol.to_string(), iv_secs)).or_default();
342 cell.extend(bars.iter().cloned());
343 cell.sort_unstable_by_key(|b| b.datetime);
344 cell.dedup_by(|a, b| a.datetime == b.datetime);
345 if cell.len() > KLINE_CELL_CAP {
346 let excess = cell.len() - KLINE_CELL_CAP;
347 cell.drain(0..excess);
348 }
349 self.frames.fetch_add(1, Ordering::AcqRel);
350 }
351
352 pub fn note_reload(&self) {
354 self.reloads.fetch_add(1, Ordering::AcqRel);
355 }
356
357 pub fn cells(&self) -> Vec<CellView> {
359 self.cells
360 .lock()
361 .unwrap()
362 .iter()
363 .map(|((sym, iv), bars)| CellView {
364 symbol: sym.clone(),
365 interval: *iv,
366 bars: bars.clone(),
367 })
368 .collect()
369 }
370
371 pub fn counts(&self) -> (u64, u64, u64) {
373 (
374 self.frames.load(Ordering::Acquire),
375 self.updates.load(Ordering::Acquire),
376 self.reloads.load(Ordering::Acquire),
377 )
378 }
379}
380
381#[derive(Debug, Default, Clone, PartialEq)]
384pub struct ReloadForm {
385 pub interval: String,
386 pub symbol: String,
387 pub csv: String,
388}
389
390pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
397 let g = raw.trim();
398 if !g.is_empty() {
399 if g.chars().any(|c| c.is_whitespace()) {
400 return Err(format!("网关解析: 含空白:{raw:?}"));
401 }
402 return Ok(g.to_string());
403 }
404 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
405 Some(f) => Ok(f.to_string()),
406 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
407 }
408}
409
410#[derive(Debug, Clone, PartialEq)]
413pub struct CtpForm {
414 pub broker_id: String,
415 pub td_addr: String,
416 pub td_user: String,
417 pub td_password: String,
418 pub td_auth_code: String,
419 pub td_app_id: String,
420 pub md_addr: String,
421 pub md_user: String,
422 pub md_password: String,
423 pub symbols: String,
425 pub accept_futures: bool,
428 pub accept_options: bool,
429}
430
431impl Default for CtpForm {
434 fn default() -> Self {
435 Self {
436 broker_id: String::new(),
437 td_addr: String::new(),
438 td_user: String::new(),
439 td_password: String::new(),
440 td_auth_code: String::new(),
441 td_app_id: String::new(),
442 md_addr: String::new(),
443 md_user: String::new(),
444 md_password: String::new(),
445 symbols: String::new(),
446 accept_futures: true,
447 accept_options: false,
448 }
449 }
450}
451
452impl CtpForm {
453 #[must_use]
457 pub fn product_classes(&self) -> Vec<String> {
458 let mut codes = Vec::new();
459 if self.accept_futures {
460 codes.push("1".to_string());
461 }
462 if self.accept_options {
463 codes.push("2".to_string());
464 }
465 codes
466 }
467
468 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
470 ek_ctp::plugin::ConfigureReq {
471 broker_id: self.broker_id.trim().to_string(),
472 td_addr: self.td_addr.trim().to_string(),
473 td_user: self.td_user.trim().to_string(),
474 td_password: self.td_password.clone(),
475 td_auth_code: self.td_auth_code.trim().to_string(),
476 td_app_id: self.td_app_id.trim().to_string(),
477 md_addr: self.md_addr.trim().to_string(),
478 md_user: self.md_user.trim().to_string(),
479 md_password: self.md_password.clone(),
480 product_classes: self.product_classes(),
481 }
482 }
483
484 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
486 let list: Vec<String> = self
487 .symbols
488 .split([',', ',', ' '])
489 .map(|s| s.trim().to_string())
490 .filter(|s| !s.is_empty())
491 .collect();
492 if list.is_empty() {
493 return Err("订阅解析: 标的位空".to_string());
494 }
495 Ok(list)
496 }
497
498 pub fn validate(&self) -> Result<(), String> {
507 fn front(name: &str, v: &str) -> Result<(), String> {
508 let v = v.trim();
509 let Some(rest) = v.strip_prefix("tcp://") else {
510 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
511 };
512 let Some((host, port)) = rest.rsplit_once(':') else {
513 return Err(format!("{name} 缺端口:{name} {:?}", v));
514 };
515 if host.trim().is_empty() {
516 return Err(format!("{name} 主机位空:{}", v));
517 }
518 let port: u32 = port
519 .parse()
520 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
521 if port == 0 || port > 65535 {
522 return Err(format!("{name} 端口越界(1..=65535):{}", v));
523 }
524 Ok(())
525 }
526 if self.broker_id.trim().is_empty() {
527 return Err("broker 位空".to_string());
528 }
529 if self.td_user.trim().is_empty() {
530 return Err("交易账号位空".to_string());
531 }
532 front("td 前置", &self.td_addr)?;
533 if !self.md_addr.trim().is_empty() {
534 front("md 前置", &self.md_addr)?;
535 }
536 if self.td_app_id.trim().is_empty() {
537 return Err("AppID 位空(CTP 登录必需)".to_string());
538 }
539 if self.td_auth_code.trim().is_empty() {
540 return Err("AuthCode 位空(CTP 登录必需)".to_string());
541 }
542 Ok(())
543 }
544}
545
546#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
553pub enum AccountStage {
554 #[default]
556 NotAssembled,
557 Assembling,
559 Assembled,
561 AssembleFailed,
563}
564
565impl AccountStage {
566 #[must_use]
568 pub fn label(self) -> &'static str {
569 match self {
570 AccountStage::NotAssembled => "未装配",
571 AccountStage::Assembling => "装配中…",
572 AccountStage::Assembled => "已装配·登录态未知",
575 AccountStage::AssembleFailed => "装配失败",
576 }
577 }
578}
579
580#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
585pub enum LoginState {
586 #[default]
591 Unknown,
592 Online,
594 LoginFailed,
596 FrontDown,
598 QueryStarved,
600 Stopped,
602}
603
604impl LoginState {
605 #[must_use]
607 pub fn label(self) -> &'static str {
608 match self {
609 LoginState::Unknown => "登录态未知",
610 LoginState::Online => "在线",
611 LoginState::LoginFailed => "登录失败",
612 LoginState::FrontDown => "掉线",
613 LoginState::QueryStarved => "查询断供",
614 LoginState::Stopped => "已断开",
615 }
616 }
617
618 #[must_use]
620 pub fn is_alarm(self) -> bool {
621 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
622 }
623}
624
625fn count_note(label: &str, n: u32) -> Option<String> {
627 (n > 0).then(|| format!("{label} {n} 次"))
628}
629
630fn brief_counter_reason(full: &str) -> Option<String> {
637 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
638 let body = tail.trim().trim_start_matches("CTP:").trim();
639 if body.is_empty() {
640 return None;
641 }
642 let code = full
643 .rfind("代码:")
644 .map(|i| {
645 let rest = &full[i + "代码:".len()..];
646 let end = rest.find(',').unwrap_or(rest.len());
647 rest[..end].trim().to_string()
648 })
649 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
650 Some(match code {
651 Some(c) => format!("{body}({c})"),
652 None => body.to_string(),
653 })
654}
655
656#[derive(Debug, Clone, PartialEq, Default)]
662pub struct AccountRow {
663 pub gateway: String,
665 pub alias: String,
667 pub broker_id: String,
668 pub account: String,
669 pub md_source: bool,
671 pub form: CtpForm,
673 pub restored: bool,
679 pub auto_connect: bool,
684 pub password_ref: bool,
689 pub stage: AccountStage,
691 pub fail_note: Option<String>,
693 pub effective_classes: Vec<String>,
698 pub connected: bool,
701 pub login: LoginState,
705 pub status_note: Option<String>,
708 pub login_brief: Option<String>,
715 pub chain_note: Option<String>,
723}
724
725impl AccountRow {
726 pub fn derive(form: &CtpForm) -> Result<Self, String> {
731 let broker = form.broker_id.trim();
732 let account = form.td_user.trim();
733 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
734 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
735 })?;
736 Ok(Self {
737 gateway,
738 alias: String::new(),
739 broker_id: broker.to_string(),
740 account: account.to_string(),
741 md_source: false,
742 form: form.clone(),
743 stage: AccountStage::NotAssembled,
744 fail_note: None,
745 effective_classes: Vec::new(),
746 connected: false,
747 login: LoginState::Unknown,
748 status_note: None,
749 login_brief: None,
750 chain_note: None,
751 restored: false,
754 auto_connect: false,
755 password_ref: false,
756 })
757 }
758
759 #[must_use]
765 pub fn status_label(&self) -> String {
766 if self.stage == AccountStage::Assembled {
767 match self.login {
768 LoginState::Unknown => "已装配·登录态未知".to_string(),
769 other => format!("已装配·{}", other.label()),
770 }
771 } else {
772 self.stage.label().to_string()
773 }
774 }
775
776 pub fn apply_status(
789 &mut self,
790 st: &ek_ctp::plugin::GatewayStatus,
791 now_ms: i64,
792 reject_limit: u32,
793 stale_secs: i64,
794 ) {
795 let td_side = !self.form.td_addr.trim().is_empty();
796 let (login, note) = if td_side {
797 if st.td_login_failed && !st.td_logged_in {
798 (
799 LoginState::LoginFailed,
800 st.td_last_error
801 .clone()
802 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
803 )
804 } else if st.td_logged_in {
805 if st.td_query_reject_streak >= reject_limit {
806 (
807 LoginState::QueryStarved,
808 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
809 )
810 } else if let Some(last) = st.td_last_account_ms {
811 let idle_ms = now_ms.saturating_sub(last);
812 if idle_ms > stale_secs.saturating_mul(1_000) {
813 (
814 LoginState::QueryStarved,
815 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
816 )
817 } else {
818 (LoginState::Online, count_note("掉线", st.td_disconnects))
819 }
820 } else {
821 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
828 match since_login_ms {
829 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
830 LoginState::QueryStarved,
831 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
832 ),
833 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
834 }
835 }
836 } else if self.login == LoginState::Stopped {
837 let extra = if st.td_disconnects > 0 {
838 format!("(此前掉线 {} 次)", st.td_disconnects)
839 } else {
840 String::new()
841 };
842 (
843 LoginState::Stopped,
844 Some(format!("会话已停{extra};『连接』可恢复")),
845 )
846 } else if st.td_disconnects > 0 {
847 (
848 LoginState::FrontDown,
849 Some(format!("掉线 {} 次", st.td_disconnects)),
850 )
851 } else {
852 (LoginState::Unknown, st.td_last_error.clone())
853 }
854 } else if st.md_logged_in {
855 (
856 LoginState::Online,
857 count_note("掉线", st.md_disconnects)
858 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
859 )
860 } else if self.login == LoginState::Stopped {
861 (
862 LoginState::Stopped,
863 Some("会话已停;『连接』可恢复".to_string()),
864 )
865 } else if st.md_disconnects > 0 {
866 (
867 LoginState::FrontDown,
868 Some(format!("掉线 {} 次", st.md_disconnects)),
869 )
870 } else {
871 (LoginState::Unknown, st.md_last_error.clone())
872 };
873 let note = if td_side && st.td_instrument_rejects > 0 {
878 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
879 match note {
880 Some(base) if base.contains("合约查询被拒") => Some(base),
882 Some(base) => Some(format!("{base};{add}")),
883 None => Some(add),
884 }
885 } else {
886 note
887 };
888 self.login = login;
889 self.status_note = note;
890 self.login_brief = if login.is_alarm() {
893 st.td_last_error
894 .as_deref()
895 .or(st.md_last_error.as_deref())
896 .and_then(brief_counter_reason)
897 } else {
898 None
899 };
900 }
901
902 #[must_use]
904 pub fn display_name(&self) -> String {
905 let a = self.alias.trim();
906 if a.is_empty() {
907 self.account.clone()
908 } else {
909 a.to_string()
910 }
911 }
912}
913
914#[derive(Debug, Clone, PartialEq, Default)]
917pub struct TradeSnap {
918 pub orders: Vec<ek_data::order::OrderData>,
919 pub trades: Vec<ek_data::trade::TradeData>,
920 pub positions: Vec<ek_data::position::PositionData>,
921 pub accounts: Vec<ek_data::account::AccountData>,
923}
924
925pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
928 let interval: i64 = form
929 .interval
930 .trim()
931 .parse()
932 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
933 let symbol = form.symbol.trim().to_string();
934 if symbol.is_empty() {
935 return Err("回灌解析: symbol 空".to_string());
936 }
937 if form.csv.trim().is_empty() {
938 return Err("回灌解析: csv 空".to_string());
939 }
940 let mut bars = Vec::new();
941 for (i, raw) in form.csv.lines().enumerate() {
942 let line = raw.trim();
943 if line.is_empty() || line.starts_with('#') {
944 continue;
945 }
946 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
947 if parts.len() != 6 {
948 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
949 }
950 let num = |s: &str| -> Result<f64, String> {
951 s.parse::<f64>()
952 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
953 };
954 let dt_s: i64 = parts[0]
955 .parse()
956 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
957 bars.push(BarData {
958 symbol: symbol.clone(),
959 datetime: dt_s * 1_000_000_000,
960 interval: Some(
961 ek_dm::wire::interval_of_seconds(interval)
962 .unwrap_or(ek_data::constant::Interval::MINUTE),
963 ),
964 open_price: num(parts[1])?,
965 high_price: num(parts[2])?,
966 low_price: num(parts[3])?,
967 close_price: num(parts[4])?,
968 volume: num(parts[5])?,
969 ..BarData::default()
970 });
971 }
972 if bars.is_empty() {
973 return Err("回灌解析: 无有效行".to_string());
974 }
975 Ok((interval, symbol, bars))
976}
977
978#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
983pub enum TopLeftTab {
984 #[default]
985 Orders,
986 Active,
987}
988
989impl TopLeftTab {
990 pub fn label(self) -> &'static str {
991 match self {
992 Self::Orders => "委托",
993 Self::Active => "未成交",
994 }
995 }
996}
997
998#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1000pub enum BottomRightTab {
1001 #[default]
1002 Account,
1003 Logs,
1004}
1005
1006impl BottomRightTab {
1007 pub fn label(self) -> &'static str {
1008 match self {
1009 Self::Account => "资金",
1010 Self::Logs => "日志",
1011 }
1012 }
1013}
1014
1015#[derive(Debug, Clone, Copy, PartialEq)]
1018pub struct DataSplit {
1019 pub col: f32,
1021 pub row: f32,
1023}
1024
1025impl Default for DataSplit {
1026 fn default() -> Self {
1027 Self {
1028 col: crate::splitters::DATA_SPLIT_DEFAULT,
1029 row: crate::splitters::DATA_SPLIT_DEFAULT,
1030 }
1031 }
1032}
1033
1034pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
1037 ek_data::constant::Exchange::SHFE,
1038 ek_data::constant::Exchange::DCE,
1039 ek_data::constant::Exchange::CZCE,
1040 ek_data::constant::Exchange::CFFEX,
1041 ek_data::constant::Exchange::INE,
1042 ek_data::constant::Exchange::GFEX,
1043];
1044
1045#[derive(Debug, Default)]
1055pub struct TickStore {
1056 ticks: BTreeMap<String, ek_data::tick::TickData>,
1057}
1058
1059impl TickStore {
1060 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1063 self.ticks
1064 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1065 }
1066
1067 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1069 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1070 v.sort_by(|a, b| {
1071 (a.exchange.as_str().to_string(), a.symbol.clone())
1072 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1073 });
1074 v
1075 }
1076
1077 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1079 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1080 for t in self.rows() {
1081 let ex = t.exchange.as_str().to_string();
1082 match out.last_mut() {
1083 Some((e, v)) if *e == ex => v.push(t),
1084 _ => out.push((ex, vec![t])),
1085 }
1086 }
1087 out
1088 }
1089
1090 pub fn len(&self) -> usize {
1091 self.ticks.len()
1092 }
1093
1094 pub fn is_empty(&self) -> bool {
1095 self.ticks.is_empty()
1096 }
1097}
1098
1099#[derive(Debug, Clone, PartialEq)]
1101pub struct OrderForm {
1102 pub exchange: String,
1104 pub symbol: String,
1106 pub direction: ek_data::constant::Direction,
1107 pub offset: ek_data::constant::Offset,
1108 pub order_type: ek_data::constant::OrderType,
1109 pub price: String,
1111 pub volume: String,
1112 pub gateway: String,
1114 pub follow_price: bool,
1116 pub auto_split: bool,
1119}
1120
1121impl Default for OrderForm {
1122 fn default() -> Self {
1123 Self {
1124 exchange: "SHFE".to_string(),
1125 symbol: String::new(),
1126 direction: ek_data::constant::Direction::LONG,
1127 offset: ek_data::constant::Offset::OPEN,
1128 order_type: ek_data::constant::OrderType::LIMIT,
1129 price: String::new(),
1130 volume: "1".to_string(),
1131 gateway: String::new(),
1132 follow_price: true,
1133 auto_split: true,
1134 }
1135 }
1136}
1137
1138impl OrderForm {
1139 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1141 use std::str::FromStr;
1142 let symbol = self.symbol.trim();
1143 if symbol.is_empty() {
1144 return Err("下单: 代码为空".to_string());
1145 }
1146 if symbol.contains('.') {
1147 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1148 }
1149 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1150 .map_err(|e| format!("下单: {e}"))?;
1151 let price: f64 = self
1152 .price
1153 .trim()
1154 .parse()
1155 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1156 let volume: f64 = self
1157 .volume
1158 .trim()
1159 .parse()
1160 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1161 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1162 return Err("下单: 限价单价格须为正".to_string());
1163 }
1164 if volume <= 0.0 {
1165 return Err("下单: 数量须为正".to_string());
1166 }
1167 Ok(ek_data::order_request::OrderRequest {
1168 symbol: symbol.to_string(),
1169 exchange,
1170 direction: self.direction,
1171 order_type: self.order_type,
1172 volume,
1173 price,
1174 offset: self.offset,
1175 ..ek_data::order_request::OrderRequest::default()
1176 })
1177 }
1178}
1179
1180#[derive(Debug, Default)]
1186pub struct ContractStore {
1187 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1188}
1189
1190impl ContractStore {
1191 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1192 self.contracts
1193 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1194 }
1195
1196 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1198 let p = pattern.trim();
1199 self.contracts
1200 .values()
1201 .filter(|c| {
1202 p.is_empty()
1203 || c.symbol.contains(p)
1204 || c.exchange.as_str().eq_ignore_ascii_case(p)
1205 || c.name.contains(p)
1206 })
1207 .cloned()
1208 .collect()
1209 }
1210
1211 pub fn len(&self) -> usize {
1212 self.contracts.len()
1213 }
1214
1215 pub fn is_empty(&self) -> bool {
1216 self.contracts.is_empty()
1217 }
1218}
1219
1220#[derive(Debug, Default)]
1222pub struct UiShared {
1223 logs: Mutex<LogRing>,
1224 snap: Mutex<Arc<BusSnap>>,
1225 market: MarketStore,
1226 ctp_form: Mutex<CtpForm>,
1227 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1230 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1231 trade_ids: Mutex<HashSet<String>>,
1239 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1240 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1241 trade_seq: AtomicU64,
1243 trade_dropped: AtomicU64,
1245 trade_dirty: AtomicBool,
1247 trade_snap: Mutex<Arc<TradeSnap>>,
1249 tick_store: Mutex<TickStore>,
1250 order_form: Mutex<OrderForm>,
1251 data_split: Mutex<DataSplit>,
1254 selected_exchange: Mutex<String>,
1256 topleft_tab: Mutex<TopLeftTab>,
1258 bottomright_tab: Mutex<BottomRightTab>,
1260 ops_window: AtomicBool,
1261 accounts_window: AtomicBool,
1264 contracts_window: AtomicBool,
1265 contract_store: Mutex<ContractStore>,
1266 contract_query: Mutex<String>,
1267 contract_selected: Mutex<Vec<String>>,
1268 kline_windows: Mutex<Vec<String>>,
1269 kline_fetches: Mutex<Vec<KlineFetch>>,
1271 kline_fetched: Mutex<HashSet<(String, i64)>>,
1273 rev: AtomicU64,
1274 stop: AtomicBool,
1276 cmds: Mutex<VecDeque<UiCmd>>,
1278 pending_confirm: Mutex<Option<String>>,
1280 accounts: Mutex<Vec<AccountRow>>,
1282 account_edit_window: AtomicBool,
1284 editing_account: Mutex<Option<String>>,
1286 edit_alias: Mutex<String>,
1288 edit_error: Mutex<Option<String>>,
1290 md_source: Mutex<String>,
1292 subscriptions: Mutex<Vec<String>>,
1294 row_sel: Mutex<BTreeMap<String, String>>,
1297 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1301 status_cursor: AtomicUsize,
1307 status_seed: Mutex<Vec<String>>,
1310 list_tick: AtomicU32,
1312 list_due: AtomicBool,
1316 seed_due: AtomicBool,
1320 pipeline_due: AtomicBool,
1322 cfg_due: AtomicBool,
1324 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1327 cfg_delete_arm: Mutex<Option<String>>,
1329 theme: AtomicU8,
1333 theme_persist_due: AtomicBool,
1336 auto_connect_persist_due: AtomicBool,
1340 bad_payload: Mutex<BTreeMap<String, u64>>,
1342 pipeline_lines: Mutex<Vec<String>>,
1345 status_thresholds: Mutex<Option<(u32, i64)>>,
1348 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1350 toasts: crate::toast::ToastsCell,
1353 alerted: Mutex<std::collections::BTreeSet<String>>,
1355 exit_confirm_open: AtomicBool,
1357 exit_confirmed: AtomicBool,
1359 pluginwin_open: AtomicBool,
1361 bundle_open: AtomicBool,
1363 assembly_locked: AtomicBool,
1366}
1367
1368impl UiShared {
1369 pub fn new() -> Self {
1371 Self::default()
1372 }
1373
1374 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1378 self.tick_store.lock().unwrap().put(t);
1379 self.bump();
1380 }
1381
1382 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1384 self.tick_store.lock().unwrap().rows()
1385 }
1386
1387 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1389 self.tick_store.lock().unwrap().groups()
1390 }
1391
1392 pub fn tick_count(&self) -> usize {
1394 self.tick_store.lock().unwrap().len()
1395 }
1396
1397 pub fn order_form(&self) -> OrderForm {
1399 self.order_form.lock().unwrap().clone()
1400 }
1401
1402 pub fn set_order_form(&self, f: OrderForm) {
1404 *self.order_form.lock().unwrap() = f;
1405 self.bump();
1406 }
1407
1408 pub fn selected_exchange(&self) -> String {
1412 let s = self.selected_exchange.lock().unwrap().clone();
1413 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1414 s
1415 } else {
1416 MARKET_ORDER[0].as_str().to_string()
1417 }
1418 }
1419
1420 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1422 *self.selected_exchange.lock().unwrap() = s.into();
1423 self.bump();
1424 }
1425
1426 pub fn topleft_tab(&self) -> TopLeftTab {
1428 *self.topleft_tab.lock().unwrap()
1429 }
1430
1431 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1433 *self.topleft_tab.lock().unwrap() = t;
1434 self.bump();
1435 }
1436
1437 pub fn bottomright_tab(&self) -> BottomRightTab {
1439 *self.bottomright_tab.lock().unwrap()
1440 }
1441
1442 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1444 *self.bottomright_tab.lock().unwrap() = t;
1445 self.bump();
1446 }
1447
1448 pub fn data_split(&self) -> DataSplit {
1450 *self.data_split.lock().unwrap()
1451 }
1452
1453 pub fn set_data_split(&self, s: DataSplit) {
1456 let s = DataSplit {
1457 col: crate::splitters::clamp_frac(s.col),
1458 row: crate::splitters::clamp_frac(s.row),
1459 };
1460 let mut g = self.data_split.lock().unwrap();
1461 if g.col != s.col || g.row != s.row {
1462 *g = s;
1463 self.bump();
1464 }
1465 }
1466
1467 pub fn reset_layout(&self) {
1471 *self.data_split.lock().unwrap() = DataSplit::default();
1472 *self.selected_exchange.lock().unwrap() = String::new();
1473 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1474 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1475 self.bump();
1476 }
1477
1478 pub fn ops_window_open(&self) -> bool {
1482 self.ops_window.load(Ordering::Acquire)
1483 }
1484
1485 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1488 self.table_sorts
1489 .lock()
1490 .unwrap()
1491 .get(table)
1492 .copied()
1493 .flatten()
1494 }
1495
1496 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1498 let mut m = self.table_sorts.lock().unwrap();
1499 let asc = match m.get(table).copied().flatten() {
1500 Some((c, a)) if c == col => !a,
1501 _ => true,
1502 };
1503 m.insert(table.to_string(), Some((col, asc)));
1504 }
1505
1506 pub fn set_ops_window_open(&self, v: bool) {
1507 self.ops_window.store(v, Ordering::Release);
1508 self.bump();
1509 }
1510
1511 pub fn accounts_window_open(&self) -> bool {
1515 self.accounts_window.load(Ordering::Acquire)
1516 }
1517
1518 pub fn set_accounts_window_open(&self, v: bool) {
1520 self.accounts_window.store(v, Ordering::Release);
1521 self.bump();
1522 }
1523
1524 pub fn account_edit_open(&self) -> bool {
1526 self.account_edit_window.load(Ordering::Acquire)
1527 }
1528
1529 pub fn editing_account(&self) -> Option<String> {
1531 self.editing_account.lock().unwrap().clone()
1532 }
1533
1534 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1538 match gateway {
1539 Some(gw) => {
1540 let Some(row) = self.account(gw) else {
1541 return false;
1542 };
1543 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1544 *self.edit_alias.lock().unwrap() = row.alias.clone();
1545 self.set_ctp_form(row.form.clone());
1546 }
1547 None => {
1548 *self.editing_account.lock().unwrap() = None;
1549 *self.edit_alias.lock().unwrap() = String::new();
1550 let draft = self.draft_from_existing();
1551 self.set_ctp_form(draft);
1552 }
1553 }
1554 *self.edit_error.lock().unwrap() = None;
1555 self.account_edit_window.store(true, Ordering::Release);
1556 self.bump();
1557 true
1558 }
1559
1560 pub fn end_account_edit(&self) {
1562 *self.editing_account.lock().unwrap() = None;
1563 *self.edit_alias.lock().unwrap() = String::new();
1564 *self.edit_error.lock().unwrap() = None;
1565 self.account_edit_window.store(false, Ordering::Release);
1566 let draft = self.draft_from_existing();
1567 self.set_ctp_form(draft);
1568 self.bump();
1569 }
1570
1571 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1577 if self.editing_account().as_deref() == Some(gateway) {
1578 self.end_account_edit();
1579 }
1580 }
1581
1582 pub fn edit_alias(&self) -> String {
1584 self.edit_alias.lock().unwrap().clone()
1585 }
1586
1587 pub fn set_edit_alias(&self, v: impl Into<String>) {
1589 *self.edit_alias.lock().unwrap() = v.into();
1590 self.bump();
1591 }
1592
1593 pub fn edit_error(&self) -> Option<String> {
1595 self.edit_error.lock().unwrap().clone()
1596 }
1597
1598 pub fn set_edit_error(&self, e: Option<String>) {
1600 *self.edit_error.lock().unwrap() = e;
1601 self.bump();
1602 }
1603
1604 pub fn account_edit_dirty(&self) -> bool {
1606 let editing = self.editing_account();
1607 let draft = self.ctp_form();
1608 match editing {
1609 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1610 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1611 }
1612 }
1613
1614 pub fn accounts(&self) -> Vec<AccountRow> {
1616 self.accounts.lock().unwrap().clone()
1617 }
1618
1619 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1621 self.accounts
1622 .lock()
1623 .unwrap()
1624 .iter()
1625 .find(|a| a.gateway == gateway)
1626 .cloned()
1627 }
1628
1629 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1635 let mut acc = self.accounts.lock().unwrap();
1636 if acc.iter().any(|a| a.gateway == row.gateway) {
1637 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1638 }
1639 let alias = row.alias.trim();
1640 if !alias.is_empty()
1641 && acc
1642 .iter()
1643 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1644 {
1645 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1646 }
1647 if acc.len() >= MAX_ACCOUNTS {
1648 return Err(format!(
1649 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1650 ));
1651 }
1652 acc.push(row);
1653 drop(acc);
1654 self.bump();
1655 Ok(())
1656 }
1657
1658 #[must_use]
1660 pub fn can_add_account(&self) -> bool {
1661 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1662 }
1663
1664 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1670 let alias = alias.trim().to_string();
1671 let mut acc = self.accounts.lock().unwrap();
1672 if !alias.is_empty()
1673 && acc
1674 .iter()
1675 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1676 {
1677 return false;
1678 }
1679 match acc.iter_mut().find(|a| a.gateway == gateway) {
1680 Some(a) => {
1681 a.alias = alias;
1682 drop(acc);
1683 self.bump();
1684 true
1685 }
1686 None => false,
1687 }
1688 }
1689
1690 pub fn note_connect_sent(&self, gateway: &str) {
1695 let mut acc = self.accounts.lock().unwrap();
1696 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1697 a.stage = AccountStage::Assembling;
1698 a.fail_note = None;
1699 a.login = LoginState::Unknown;
1700 a.status_note = None;
1701 a.login_brief = None;
1702 a.chain_note = None;
1704 drop(acc);
1705 self.bump();
1706 }
1707 }
1708
1709 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1713 let mut acc = self.accounts.lock().unwrap();
1714 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1715 match result {
1716 Ok(()) => {
1717 a.stage = AccountStage::Assembled;
1718 a.fail_note = None;
1719 }
1720 Err(e) => {
1721 a.stage = AccountStage::AssembleFailed;
1722 a.fail_note = Some(e);
1723 }
1724 }
1725 drop(acc);
1726 self.bump();
1727 }
1728 }
1729
1730 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1734 let mut acc = self.accounts.lock().unwrap();
1735 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1736 a.effective_classes = classes.to_vec();
1737 drop(acc);
1738 self.bump();
1739 }
1740 }
1741
1742 pub fn note_chain(&self, gateway: &str, note: &str) {
1747 let mut acc = self.accounts.lock().unwrap();
1748 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1749 if a.chain_note.as_deref() == Some(note) {
1750 return; }
1752 a.chain_note = Some(note.to_string());
1753 drop(acc);
1754 self.bump();
1755 }
1756 }
1757
1758 #[must_use]
1762 pub fn status_thresholds(&self) -> (u32, i64) {
1763 self.status_thresholds
1764 .lock()
1765 .unwrap()
1766 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1767 }
1768
1769 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1771 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1772 self.bump();
1773 }
1774
1775 pub fn list_poll_due(&self) -> bool {
1780 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1781 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1782 }
1783
1784 pub fn request_list_refresh(&self) {
1786 self.list_due.store(true, Ordering::Release);
1787 }
1788
1789 pub fn request_trade_reseed(&self) {
1791 self.seed_due.store(true, Ordering::Release);
1792 }
1793
1794 pub fn take_trade_reseed(&self) -> bool {
1796 self.seed_due.swap(false, Ordering::AcqRel)
1797 }
1798
1799 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1804 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1808 for gw in present {
1809 if !uniq.iter().any(|u| u == gw) {
1810 uniq.push(gw.clone());
1811 }
1812 }
1813 let n = uniq.len();
1814 if n == 0 {
1815 return Vec::new();
1816 }
1817 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1818 let take = STATUS_ASKS_PER_TICK.min(n);
1819 let mut out = Vec::with_capacity(take);
1820 for i in 0..take {
1821 out.push(uniq[(start + i) % n].clone());
1822 }
1823 out
1824 }
1825
1826 pub fn request_status_seed(&self, gateway: &str) {
1828 let mut q = self.status_seed.lock().unwrap();
1829 if !q.iter().any(|g| g == gateway) {
1830 q.push(gateway.to_string());
1831 }
1832 }
1833
1834 pub fn take_status_seeds(&self) -> Vec<String> {
1836 let mut q = self.status_seed.lock().unwrap();
1837 std::mem::take(&mut *q)
1838 }
1839
1840 pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
1850 let gw = ev.gateway.as_str();
1851 let text = ev.describe();
1852 self.log_line(format!("[INFO][{gw}] {text}"));
1853 let note = text
1854 .strip_prefix(CHAIN_STATE_PREFIX)
1855 .map(|s| s.trim().to_string())
1856 .unwrap_or_else(|| text.clone());
1857 self.note_chain(gw, ¬e);
1858 self.note_gateway_status(gw, &ev.status);
1859 let alarm_kind = matches!(
1860 ev.change,
1861 ek_ctp::GatewayChange::TdDisconnected
1862 | ek_ctp::GatewayChange::TdAuthFailed
1863 | ek_ctp::GatewayChange::TdLoginFailed
1864 | ek_ctp::GatewayChange::ContractQueryStopped
1865 | ek_ctp::GatewayChange::QueryRejected
1866 | ek_ctp::GatewayChange::MdDisconnected
1867 | ek_ctp::GatewayChange::MdLoginFailed
1868 | ek_ctp::GatewayChange::SubscribeRejected
1869 );
1870 if alarm_kind {
1871 let session = if ev.change.kind().starts_with("md-") {
1873 ev.status.md_disconnects
1874 } else {
1875 ev.status.td_disconnects
1876 };
1877 let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
1878 if self.alert_once(key) {
1879 self.push_alert(
1880 crate::toast::AlertLevel::Warning,
1881 format!("账户 {gw}:{}", note.trim()),
1882 );
1883 }
1884 }
1885 }
1886
1887 pub fn request_pipeline_refresh(&self) {
1889 self.pipeline_due.store(true, Ordering::Release);
1890 }
1891 pub fn take_pipeline_due(&self) -> bool {
1893 self.pipeline_due.swap(false, Ordering::AcqRel)
1894 }
1895
1896 pub fn request_cfg_refresh(&self) {
1898 self.cfg_due.store(true, Ordering::Release);
1899 }
1900
1901 pub fn take_cfg_due(&self) -> bool {
1903 self.cfg_due.swap(false, Ordering::AcqRel)
1904 }
1905
1906 pub fn theme(&self) -> crate::tokens::Theme {
1908 crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
1909 }
1910
1911 pub fn set_theme(&self, theme: crate::tokens::Theme) {
1914 self.theme.store(theme.as_u8(), Ordering::Release);
1915 self.bump();
1916 }
1917
1918 pub fn toggle_theme(&self) -> crate::tokens::Theme {
1920 let next = self.theme().flipped();
1921 self.theme.store(next.as_u8(), Ordering::Release);
1922 self.theme_persist_due.store(true, Ordering::Release);
1923 self.bump();
1924 next
1925 }
1926
1927 pub fn take_theme_persist_due(&self) -> bool {
1929 self.theme_persist_due.swap(false, Ordering::AcqRel)
1930 }
1931
1932 pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
1936 {
1937 let mut acc = self.accounts.lock().unwrap();
1938 match acc.iter_mut().find(|a| a.gateway == gateway) {
1939 Some(a) => {
1940 if a.auto_connect != on {
1941 a.auto_connect = on;
1942 self.auto_connect_persist_due.store(true, Ordering::Release);
1943 self.bump();
1944 }
1945 }
1946 None => return Err(format!("账户 {gateway} 不在清册")),
1947 }
1948 }
1949 Ok(())
1950 }
1951
1952 pub fn take_auto_connect_persist_due(&self) -> bool {
1954 self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
1955 }
1956
1957 pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
1972 let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
1974 match store.load(crate::plugin::THEME_NS) {
1975 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
1976 for (k, _f) in map.iter() {
1977 if let Some(gw) = k.strip_prefix("auto_connect.") {
1978 flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
1979 }
1980 }
1981 }
1982 Ok(_) => {} Err(_) => {} }
1985 let mut restored: Vec<String> = Vec::new();
1986 let ns_list = match store.list_namespaces() {
1987 Ok(v) => v,
1988 Err(_) => {
1989 let msg = "[ui] 档案清点失败:存储根不可读(具体见运行日志)".to_string();
1990 self.log_line(msg.clone());
1991 return restored;
1992 }
1993 };
1994 for ns in ns_list {
1995 let Some(gw) = ns.strip_prefix("ctp.") else {
1996 continue;
1997 };
1998 if self.account(gw).is_some() {
1999 self.log_line(format!(
2000 "[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
2001 ));
2002 continue;
2003 }
2004 let map = match store.load(&ns) {
2005 Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
2006 Ok(_) => {
2007 self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
2008 continue;
2009 }
2010 Err(e) => {
2011 self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
2012 continue;
2013 }
2014 };
2015 let password_ref = map
2017 .get_secret_key("td_password")
2018 .is_some_and(|k| !k.is_empty());
2019 let mut form = CtpForm {
2020 broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
2021 td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
2022 td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
2023 td_password: String::new(),
2024 td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
2025 td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
2026 md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
2027 md_user: String::new(),
2028 md_password: String::new(),
2029 symbols: String::new(),
2030 accept_futures: false,
2031 accept_options: false,
2032 };
2033 for c in map
2036 .get_plain("product_classes")
2037 .unwrap_or_default()
2038 .split(',')
2039 {
2040 if c.trim() == "1" {
2041 form.accept_futures = true;
2042 }
2043 }
2044 let Ok(mut row) = AccountRow::derive(&form) else {
2045 self.log_line(format!(
2046 "[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
2047 ));
2048 continue;
2049 };
2050 row.restored = true;
2051 row.auto_connect = flags.get(gw).copied().unwrap_or(false);
2052 row.password_ref = password_ref;
2053 restored.push(row.gateway.clone());
2054 self.add_account(row).ok();
2055 }
2056 if !restored.is_empty() {
2057 self.log_line(format!(
2058 "[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
2059 restored.len()
2060 ));
2061 self.bump();
2062 }
2063 restored
2064 }
2065
2066 pub fn restore_saved_accounts(&self) -> Vec<String> {
2068 match ek_cfg::Store::eco() {
2069 Ok(store) => self.restore_saved_accounts_from(&store),
2070 Err(_) => {
2071 self.log_line(
2072 "[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
2073 );
2074 Vec::new()
2075 }
2076 }
2077 }
2078
2079 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
2081 *self.cfg_briefs.lock().unwrap() = Some(b);
2082 self.bump();
2083 }
2084 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
2086 self.cfg_briefs.lock().unwrap().clone()
2087 }
2088
2089 pub fn cfg_delete_armed(&self) -> Option<String> {
2091 self.cfg_delete_arm.lock().unwrap().clone()
2092 }
2093 pub fn arm_cfg_delete(&self, ns: Option<String>) {
2094 *self.cfg_delete_arm.lock().unwrap() = ns;
2095 self.bump();
2096 }
2097
2098 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
2100 *self.pipeline_lines.lock().unwrap() = lines;
2101 self.bump();
2102 }
2103
2104 pub fn pipeline_lines(&self) -> Vec<String> {
2106 self.pipeline_lines.lock().unwrap().clone()
2107 }
2108
2109 pub fn note_bad_payload(&self, line: &str) {
2115 let mut book = self.bad_payload.lock().unwrap();
2116 let n = book.entry(line.to_string()).or_insert(0);
2117 *n += 1;
2118 if *n == 1 {
2119 self.log_line(format!(
2120 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
2121 ));
2122 }
2123 }
2124
2125 #[must_use]
2127 pub fn bad_payload_count(&self, line: &str) -> u64 {
2128 self.bad_payload
2129 .lock()
2130 .unwrap()
2131 .get(line)
2132 .copied()
2133 .unwrap_or(0)
2134 }
2135
2136 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
2150 for r in rows {
2151 let (warn, crit) = pending_thresholds(r.soft_cap);
2152 if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
2153 self.push_alert(
2154 crate::toast::AlertLevel::Warning,
2155 format!(
2156 "插件 {}:邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
2157 r.name, r.pending, r.peak, r.soft_cap
2158 ),
2159 );
2160 }
2161 if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
2162 self.push_alert(
2163 crate::toast::AlertLevel::Error,
2164 format!(
2165 "插件 {}:邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
2166 r.name, r.pending, r.soft_cap
2167 ),
2168 );
2169 }
2170 if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
2171 self.push_alert(
2172 crate::toast::AlertLevel::Error,
2173 format!(
2174 "插件 {}:已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
2175 r.name, r.soft_cap, r.spilled, r.peak
2176 ),
2177 );
2178 }
2179 }
2180 }
2181
2182 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
2190 let (limit, stale) = self.status_thresholds();
2191 let now = ek_ctp::map::now_epoch_ms();
2192 let mut alarm: Option<(LoginState, String)> = None;
2195 let mut found = false;
2196 {
2197 let mut acc = self.accounts.lock().unwrap();
2198 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2199 found = true;
2200 a.apply_status(st, now, limit, stale);
2201 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
2202 alarm = Some((
2203 a.login,
2204 a.status_note
2205 .clone()
2206 .unwrap_or_else(|| a.login.label().to_string()),
2207 ));
2208 }
2209 }
2210 }
2211 if found {
2212 self.bump();
2213 }
2214 if let Some((login, note)) = alarm
2215 && self.alert_once(format!("{gateway}|{}", login.label()))
2216 {
2217 self.push_alert(
2218 crate::toast::AlertLevel::Error,
2219 format!("账户 {gateway}:{note}"),
2220 );
2221 }
2222 }
2223
2224 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
2228 let mut acc = self.accounts.lock().unwrap();
2229 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2230 a.login = LoginState::Unknown;
2231 a.status_note = Some(format!("状态口问不到:{why}"));
2232 a.login_brief = None;
2233 drop(acc);
2234 self.bump();
2235 }
2236 }
2237
2238 pub fn mark_account_stopped(&self, gateway: &str) {
2240 let mut acc = self.accounts.lock().unwrap();
2241 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
2242 a.login = LoginState::Stopped;
2243 a.status_note = Some("会话已停;『连接』可恢复".to_string());
2244 a.login_brief = None;
2245 drop(acc);
2246 self.bump();
2247 }
2248 }
2249
2250 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
2253 let source = self.md_source();
2254 let mut f = form;
2255 if source.as_str() != gateway {
2256 f.md_addr.clear();
2257 f.md_user.clear();
2258 f.md_password.clear();
2259 }
2260 self.update_account_form(gateway, f.clone());
2261 f
2262 }
2263
2264 #[must_use]
2268 pub fn draft_from_existing(&self) -> CtpForm {
2269 let acc = self.accounts.lock().unwrap();
2270 let mut draft = CtpForm::default();
2271 if let Some(a) = acc.last() {
2272 draft.broker_id = a.form.broker_id.clone();
2273 draft.td_auth_code = a.form.td_auth_code.clone();
2274 draft.td_app_id = a.form.td_app_id.clone();
2275 draft.accept_futures = a.form.accept_futures;
2276 draft.accept_options = a.form.accept_options;
2277 }
2278 draft
2279 }
2280
2281 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
2283 let mut acc = self.accounts.lock().unwrap();
2284 match acc.iter_mut().find(|a| a.gateway == gateway) {
2285 Some(a) => {
2286 a.form = form;
2287 drop(acc);
2288 self.bump();
2289 true
2290 }
2291 None => false,
2292 }
2293 }
2294
2295 pub fn remove_account(&self, gateway: &str) -> bool {
2297 let mut acc = self.accounts.lock().unwrap();
2298 let before = acc.len();
2299 acc.retain(|a| a.gateway != gateway);
2300 let removed = acc.len() != before;
2301 if removed {
2302 let mut md = self.md_source.lock().unwrap();
2304 if md.as_str() == gateway {
2305 md.clear();
2306 }
2307 drop(acc);
2308 self.bump();
2309 }
2310 removed
2311 }
2312
2313 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
2322 let mut newly = Vec::new();
2323 {
2324 let mut acc = self.accounts.lock().unwrap();
2325 for a in acc.iter_mut() {
2326 let up = rows.iter().any(|r| r.name == a.gateway);
2327 if up && !a.connected {
2328 newly.push(a.gateway.clone());
2329 }
2330 a.connected = up;
2331 if up {
2332 a.stage = AccountStage::Assembled;
2333 a.fail_note = None;
2334 } else {
2335 a.chain_note = None;
2338 if a.stage == AccountStage::Assembled {
2339 a.stage = AccountStage::NotAssembled;
2340 a.login = LoginState::Unknown;
2341 a.status_note = None;
2342 a.login_brief = None;
2343 }
2344 }
2345 }
2346 }
2347 if !newly.is_empty() {
2348 self.bump();
2349 }
2350 newly
2351 }
2352
2353 pub fn md_source(&self) -> String {
2355 self.md_source.lock().unwrap().clone()
2356 }
2357
2358 pub fn accounts_locked(&self) -> bool {
2360 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2361 }
2362
2363 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2365 {
2366 let acc = self.accounts.lock().unwrap();
2367 if acc.iter().any(|a| a.connected) {
2368 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2369 }
2370 if !acc.iter().any(|a| a.gateway == gateway) {
2371 return Err(format!("账户 {gateway} 不在清册"));
2372 }
2373 }
2374 let mut acc = self.accounts.lock().unwrap();
2375 for a in acc.iter_mut() {
2376 a.md_source = a.gateway == gateway;
2377 }
2378 drop(acc);
2379 let mut md = self.md_source.lock().unwrap();
2380 *md = gateway.to_string();
2381 self.bump();
2382 Ok(())
2383 }
2384
2385 pub fn subscriptions(&self) -> Vec<String> {
2387 self.subscriptions.lock().unwrap().clone()
2388 }
2389
2390 pub fn note_subscriptions(&self, symbols: &[String]) {
2392 {
2393 let mut subs = self.subscriptions.lock().unwrap();
2394 for s in symbols {
2395 let s = s.trim();
2396 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2397 subs.push(s.to_string());
2398 }
2399 }
2400 }
2401 self.bump();
2402 }
2403
2404 pub fn default_gateway(&self) -> Option<String> {
2411 let chosen = {
2412 let f = self.order_form.lock().unwrap();
2413 f.gateway.trim().to_string()
2414 };
2415 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2416 return Some(chosen);
2417 }
2418 let acc = self.accounts.lock().unwrap();
2419 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2420 }
2421
2422 pub fn kline_symbols(&self) -> Vec<String> {
2424 self.kline_windows.lock().unwrap().clone()
2425 }
2426
2427 pub fn open_kline(&self, symbol: impl Into<String>) {
2429 let s = symbol.into();
2430 let mut v = self.kline_windows.lock().unwrap();
2431 if !v.contains(&s) {
2432 v.push(s.clone());
2433 }
2434 drop(v);
2435 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2436 let fq = if s.contains('.') {
2439 s.clone()
2440 } else {
2441 format!("{}.{}", self.selected_exchange(), s)
2442 };
2443 const KLINE_INTERVAL_SECS: i64 = 60;
2444 {
2445 let mut fetched = self.kline_fetched.lock().unwrap();
2446 if fetched.contains(&(fq.clone(), KLINE_INTERVAL_SECS)) {
2447 return;
2448 }
2449 fetched.insert((fq.clone(), KLINE_INTERVAL_SECS));
2450 }
2451 self.kline_fetches.lock().unwrap().push(KlineFetch {
2452 symbol: fq,
2453 interval_secs: KLINE_INTERVAL_SECS,
2454 });
2455 self.log_line(format!("[kline] 入队历史补数 {s}(1min×300,tq 桥)"));
2456 }
2457
2458 pub fn take_kline_fetches(&self) -> Vec<KlineFetch> {
2460 std::mem::take(&mut self.kline_fetches.lock().unwrap())
2461 }
2462
2463 pub fn kline_fetch_pending(&self) -> bool {
2465 !self.kline_fetches.lock().unwrap().is_empty()
2466 }
2467
2468 pub fn close_kline(&self, symbol: &str) {
2470 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2471 self.bump();
2472 }
2473
2474 pub fn contracts_window_open(&self) -> bool {
2478 self.contracts_window.load(Ordering::Acquire)
2479 }
2480
2481 pub fn set_contracts_window_open(&self, v: bool) {
2483 self.contracts_window.store(v, Ordering::Release);
2484 self.bump();
2485 }
2486
2487 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2494 self.contract_store.lock().unwrap().put(c);
2495 if !c.name.is_empty() {
2496 self.backfill_names(&c.symbol, &c.name);
2497 }
2498 self.bump();
2499 }
2500
2501 fn backfill_names(&self, symbol: &str, name: &str) {
2506 let mut touched = false;
2507 {
2508 let mut rows = self.trade_orders.lock().unwrap();
2509 for (_, (_, o)) in rows.iter_mut() {
2510 if o.symbol == symbol && o.name.is_empty() {
2511 o.name = name.to_string();
2512 touched = true;
2513 }
2514 }
2515 }
2516 {
2517 let mut rows = self.trade_trades.lock().unwrap();
2518 for (_, t) in rows.iter_mut() {
2519 if t.symbol == symbol && t.name.is_empty() {
2520 t.name = name.to_string();
2521 touched = true;
2522 }
2523 }
2524 }
2525 {
2526 let mut rows = self.trade_positions.lock().unwrap();
2527 for (_, (_, p)) in rows.iter_mut() {
2528 if p.symbol == symbol && p.name.is_empty() {
2529 p.name = name.to_string();
2530 touched = true;
2531 }
2532 }
2533 }
2534 if touched {
2535 self.trade_dirty.store(true, Ordering::Release);
2537 }
2538 }
2539
2540 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2542 self.contract_store.lock().unwrap().query(pattern)
2543 }
2544
2545 pub fn contract_count(&self) -> usize {
2547 self.contract_store.lock().unwrap().len()
2548 }
2549
2550 fn contract_name_of(&self, symbol: &str) -> Option<String> {
2552 self.contract_store
2553 .lock()
2554 .unwrap()
2555 .query(symbol)
2556 .into_iter()
2557 .find(|c| c.symbol == symbol && !c.name.is_empty())
2558 .map(|c| c.name)
2559 }
2560
2561 pub fn contract_query(&self) -> String {
2563 self.contract_query.lock().unwrap().clone()
2564 }
2565
2566 pub fn set_contract_query(&self, q: impl Into<String>) {
2568 *self.contract_query.lock().unwrap() = q.into();
2569 self.bump();
2570 }
2571
2572 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2574 let mut v = self.contract_selected.lock().unwrap();
2575 match v.iter().position(|s| s == en_symbol) {
2576 Some(i) => {
2577 v.remove(i);
2578 }
2579 None => v.push(en_symbol.to_string()),
2580 }
2581 self.bump();
2582 }
2583
2584 pub fn selected_contracts(&self) -> Vec<String> {
2586 self.contract_selected.lock().unwrap().clone()
2587 }
2588
2589 #[must_use]
2595 pub fn cancel_reject_note(msg: &str) -> String {
2596 let head = msg.split(' ').next().unwrap_or(msg);
2597 let note = if head.starts_with("unknown:order") {
2598 "这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
2599 } else if head.starts_with("mismatch:gateway") {
2600 "该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
2601 } else if head.starts_with("terminal:status") {
2602 "这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
2603 } else if head.starts_with("limit:cancel-guard") {
2604 "撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
2605 } else {
2606 ""
2607 };
2608 if note.is_empty() {
2609 format!("撤单被拒:{msg}")
2611 } else {
2612 format!("撤单被拒:{note}(机器码:{msg})")
2613 }
2614 }
2615
2616 pub fn log_line(&self, line: impl Into<String>) {
2618 self.logs.lock().unwrap().push(line);
2619 self.bump();
2620 }
2621
2622 pub fn clear_logs(&self) {
2624 self.logs.lock().unwrap().clear();
2625 self.bump();
2626 }
2627
2628 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2630 let r = self.logs.lock().unwrap();
2631 (r.snapshot(), r.spilled())
2632 }
2633
2634 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2636 let r = self.logs.lock().unwrap();
2637 (r.entries(), r.spilled())
2638 }
2639
2640 pub fn logs_len(&self) -> usize {
2642 self.logs.lock().unwrap().len()
2643 }
2644
2645 pub fn set_snapshot(&self, snap: BusSnap) {
2647 *self.snap.lock().unwrap() = Arc::new(snap);
2648 self.bump();
2649 }
2650
2651 pub fn snapshot(&self) -> Arc<BusSnap> {
2653 Arc::clone(&self.snap.lock().unwrap())
2654 }
2655
2656 pub fn rev(&self) -> u64 {
2658 self.rev.load(Ordering::Acquire)
2659 }
2660
2661 pub fn bump(&self) {
2663 self.rev.fetch_add(1, Ordering::AcqRel);
2664 }
2665
2666 pub fn request_stop(&self) {
2668 self.stop.store(true, Ordering::Release);
2669 }
2670
2671 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2674 self.pending_alerts
2675 .lock()
2676 .unwrap_or_else(std::sync::PoisonError::into_inner)
2677 .push((level, text.into()));
2678 }
2679
2680 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2682 let mut q = self
2683 .pending_alerts
2684 .lock()
2685 .unwrap_or_else(std::sync::PoisonError::into_inner);
2686 std::mem::take(&mut *q)
2687 }
2688
2689 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2691 self.toasts.lock()
2692 }
2693
2694 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2696 let key = key.into();
2697 self.alerted
2698 .lock()
2699 .unwrap_or_else(std::sync::PoisonError::into_inner)
2700 .insert(key)
2701 }
2702
2703 #[must_use]
2705 pub fn pending_alert_count(&self) -> usize {
2706 self.pending_alerts
2707 .lock()
2708 .unwrap_or_else(std::sync::PoisonError::into_inner)
2709 .len()
2710 }
2711
2712 #[must_use]
2714 pub fn alerted_count(&self) -> usize {
2715 self.alerted
2716 .lock()
2717 .unwrap_or_else(std::sync::PoisonError::into_inner)
2718 .len()
2719 }
2720
2721 pub fn exit_confirm_open(&self) -> bool {
2723 self.exit_confirm_open
2724 .load(std::sync::atomic::Ordering::Relaxed)
2725 }
2726
2727 pub fn open_exit_confirm(&self) {
2728 self.exit_confirm_open
2729 .store(true, std::sync::atomic::Ordering::Relaxed);
2730 }
2731
2732 pub fn close_exit_confirm(&self) {
2733 self.exit_confirm_open
2734 .store(false, std::sync::atomic::Ordering::Relaxed);
2735 }
2736
2737 pub fn exit_confirmed(&self) -> bool {
2738 self.exit_confirmed
2739 .load(std::sync::atomic::Ordering::Relaxed)
2740 }
2741
2742 pub fn confirm_exit(&self) {
2743 self.exit_confirmed
2744 .store(true, std::sync::atomic::Ordering::Relaxed);
2745 }
2746
2747 pub fn is_stop(&self) -> bool {
2748 self.stop.load(Ordering::Acquire)
2749 }
2750
2751 #[must_use]
2753 pub fn pluginwin_open(&self) -> bool {
2754 self.pluginwin_open.load(Ordering::Relaxed)
2755 }
2756
2757 pub fn set_pluginwin_open(&self, on: bool) {
2759 self.pluginwin_open.store(on, Ordering::Relaxed);
2760 }
2761
2762 #[must_use]
2764 pub fn bundle_open(&self) -> bool {
2765 self.bundle_open.load(Ordering::Relaxed)
2766 }
2767
2768 pub fn set_bundle_open(&self, on: bool) {
2770 self.bundle_open.store(on, Ordering::Relaxed);
2771 }
2772
2773 #[must_use]
2775 pub fn assembly_locked(&self) -> bool {
2776 self.assembly_locked.load(Ordering::Relaxed)
2777 }
2778
2779 pub fn set_assembly_locked(&self, on: bool) {
2781 self.assembly_locked.store(on, Ordering::Relaxed);
2782 }
2783
2784 pub fn push_cmd(&self, cmd: UiCmd) {
2788 self.cmds.lock().unwrap().push_back(cmd);
2789 self.bump();
2790 }
2791
2792 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2794 self.cmds.lock().unwrap().drain(..).collect()
2795 }
2796
2797 #[must_use]
2799 pub fn cmd_pending(&self) -> bool {
2800 !self.cmds.lock().unwrap().is_empty()
2801 }
2802
2803 pub fn pending_confirm(&self) -> Option<String> {
2805 self.pending_confirm.lock().unwrap().clone()
2806 }
2807
2808 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2810 let mut g = self.pending_confirm.lock().unwrap();
2811 match g.take() {
2812 Some(p) if p == name => ConfirmStep::Fired,
2813 Some(_) => {
2814 *g = Some(name.to_string());
2815 ConfirmStep::Switched
2816 }
2817 None => {
2818 *g = Some(name.to_string());
2819 ConfirmStep::Armed
2820 }
2821 }
2822 }
2823
2824 pub fn confirm_cancel(&self) {
2826 *self.pending_confirm.lock().unwrap() = None;
2827 self.bump();
2828 }
2829
2830 pub fn market(&self) -> &MarketStore {
2834 &self.market
2835 }
2836
2837 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2846 let named = |symbol: &str, name: String| -> String {
2849 if name.is_empty() {
2850 self.contract_name_of(symbol).unwrap_or(name)
2851 } else {
2852 name
2853 }
2854 };
2855 {
2856 let mut book = self.trade_orders.lock().unwrap();
2857 book.clear();
2858 for mut o in ts.orders.into_iter().rev() {
2859 o.name = named(&o.symbol, o.name);
2860 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2861 book.insert(o.en_orderid(), (seq, o));
2862 }
2863 }
2864 {
2865 let mut book = self.trade_trades.lock().unwrap();
2866 book.clear();
2867 let mut ids = self.trade_ids.lock().unwrap();
2868 ids.clear();
2869 for mut t in ts.trades.into_iter().rev() {
2870 t.name = named(&t.symbol, t.name);
2871 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2872 ids.insert(t.en_tradeid());
2873 book.push_front((seq, t));
2874 }
2875 }
2876 {
2877 let mut book = self.trade_positions.lock().unwrap();
2878 book.clear();
2879 for mut p in ts.positions.into_iter().rev() {
2880 p.name = named(&p.symbol, p.name);
2881 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2882 book.insert(p.en_positionid(), (seq, p));
2883 }
2884 }
2885 {
2886 let mut book = self.trade_funds.lock().unwrap();
2887 book.clear();
2888 for a in ts.accounts.into_iter().rev() {
2889 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2890 book.insert(a.en_accountid(), (seq, a));
2891 }
2892 }
2893 self.trade_dirty.store(true, Ordering::Release);
2894 self.bump();
2895 }
2896
2897 pub fn note_order(&self, o: &ek_data::order::OrderData) {
2905 let key = o.en_orderid();
2906 {
2907 let mut book = self.trade_orders.lock().unwrap();
2908 match book.get_mut(&key) {
2909 Some((_, cur)) => {
2910 cur.name.clone_from(&o.name);
2911 cur.traded = o.traded;
2912 cur.traded_price = o.traded_price;
2913 cur.status = o.status;
2914 cur.status_msg.clone_from(&o.status_msg);
2915 cur.datetime = o.datetime;
2916 }
2917 None => {
2918 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2919 book.insert(key, (seq, o.clone()));
2920 }
2921 }
2922 }
2923 self.trade_dirty.store(true, Ordering::Release);
2924 self.bump();
2925 }
2926
2927 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
2935 let id = t.en_tradeid();
2936 if self.trade_ids.lock().unwrap().contains(&id) {
2937 return;
2938 }
2939 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2940 let dropped = {
2941 let mut book = self.trade_trades.lock().unwrap();
2942 book.push_front((seq, t.clone()));
2943 self.trade_ids.lock().unwrap().insert(id);
2944 let over = book.len().saturating_sub(TRADE_CAP);
2945 for _ in 0..over {
2946 if let Some((_, old)) = book.pop_back() {
2948 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
2949 }
2950 }
2951 over
2952 };
2953 if dropped > 0 {
2954 let total = self
2955 .trade_dropped
2956 .fetch_add(dropped as u64, Ordering::AcqRel)
2957 + dropped as u64;
2958 if total == dropped as u64 {
2960 self.push_alert(
2961 crate::toast::AlertLevel::Warning,
2962 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
2963 );
2964 }
2965 }
2966 self.trade_dirty.store(true, Ordering::Release);
2967 self.bump();
2968 }
2969
2970 pub fn note_position(&self, p: &ek_data::position::PositionData) {
2975 let key = p.en_positionid();
2976 {
2977 let mut book = self.trade_positions.lock().unwrap();
2978 match book.get_mut(&key) {
2979 Some((_, cur)) => {
2980 cur.volume = p.volume;
2981 cur.yd_volume = p.yd_volume;
2982 cur.can_use_volume = p.can_use_volume;
2983 cur.frozen = p.frozen;
2984 cur.on_road_volume = p.on_road_volume;
2985 cur.market_value = p.market_value;
2986 cur.avg_price = p.avg_price;
2987 cur.price = p.price;
2988 cur.last_price = p.last_price;
2989 cur.pnl = p.pnl;
2990 cur.profit_rate = p.profit_rate;
2991 cur.name.clone_from(&p.name);
2992 }
2993 None => {
2994 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2995 book.insert(key, (seq, p.clone()));
2996 }
2997 }
2998 }
2999 self.trade_dirty.store(true, Ordering::Release);
3000 self.bump();
3001 }
3002
3003 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
3005 let key = a.en_accountid();
3006 {
3007 let mut book = self.trade_funds.lock().unwrap();
3008 match book.get_mut(&key) {
3009 Some((_, cur)) => {
3010 cur.balance = a.balance;
3011 cur.frozen = a.frozen;
3012 }
3013 None => {
3014 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
3015 book.insert(key, (seq, a.clone()));
3016 }
3017 }
3018 }
3019 self.trade_dirty.store(true, Ordering::Release);
3020 self.bump();
3021 }
3022
3023 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
3025 let mut removed = 0usize;
3026 {
3027 let mut book = self.trade_orders.lock().unwrap();
3028 let before = book.len();
3029 book.retain(|_, (_, o)| o.gateway_name != gateway);
3030 removed += before - book.len();
3031 }
3032 {
3033 let mut book = self.trade_trades.lock().unwrap();
3034 let before = book.len();
3035 book.retain(|(_, t)| t.gateway_name != gateway);
3036 removed += before - book.len();
3037 let prefix = format!("{gateway}.");
3040 self.trade_ids
3041 .lock()
3042 .unwrap()
3043 .retain(|id| !id.starts_with(&prefix));
3044 }
3045 {
3046 let mut book = self.trade_positions.lock().unwrap();
3047 let before = book.len();
3048 book.retain(|_, (_, p)| p.gateway_name != gateway);
3049 removed += before - book.len();
3050 }
3051 {
3052 let mut book = self.trade_funds.lock().unwrap();
3053 let before = book.len();
3054 book.retain(|_, (_, a)| a.gateway_name != gateway);
3055 removed += before - book.len();
3056 }
3057 if removed > 0 {
3058 self.trade_dirty.store(true, Ordering::Release);
3059 self.bump();
3060 }
3061 removed
3062 }
3063
3064 #[must_use]
3066 pub fn trade_dropped(&self) -> u64 {
3067 self.trade_dropped.load(Ordering::Acquire)
3068 }
3069
3070 pub fn trade_snap(&self) -> Arc<TradeSnap> {
3075 let mut cache = self.trade_snap.lock().unwrap();
3076 if self.trade_dirty.swap(false, Ordering::AcqRel) {
3077 *cache = Arc::new(self.build_trade_snap());
3078 }
3079 Arc::clone(&cache)
3080 }
3081
3082 fn build_trade_snap(&self) -> TradeSnap {
3085 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
3086 where
3087 T: Clone,
3088 {
3089 let mut v: Vec<(u64, &String, &T)> =
3090 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
3091 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
3092 v.into_iter().map(|(_, _, v)| v.clone()).collect()
3093 }
3094 TradeSnap {
3095 orders: by_seq(&self.trade_orders.lock().unwrap()),
3096 trades: self
3097 .trade_trades
3098 .lock()
3099 .unwrap()
3100 .iter()
3101 .map(|(_, t)| t.clone())
3102 .collect(),
3103 positions: by_seq(&self.trade_positions.lock().unwrap()),
3104 accounts: by_seq(&self.trade_funds.lock().unwrap()),
3105 }
3106 }
3107
3108 pub fn ctp_form(&self) -> CtpForm {
3112 self.ctp_form.lock().unwrap().clone()
3113 }
3114
3115 pub fn set_ctp_form(&self, f: CtpForm) {
3117 *self.ctp_form.lock().unwrap() = f;
3118 self.bump();
3119 }
3120
3121 pub fn row_selected(&self, table: &str, key: &str) -> bool {
3125 self.row_sel
3126 .lock()
3127 .unwrap()
3128 .get(table)
3129 .is_some_and(|k| k == key)
3130 }
3131
3132 pub fn select_row(&self, table: &str, key: &str) {
3134 self.row_sel
3135 .lock()
3136 .unwrap()
3137 .insert(table.to_string(), key.to_string());
3138 }
3139}