1use std::collections::{BTreeMap, VecDeque};
7use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, 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}
168
169pub const KLINE_CELL_CAP: usize = 4096;
171
172pub const MAX_ACCOUNTS: usize = 8;
175
176pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
181
182pub const ACCOUNT_STALE_SECS: i64 = 30;
187
188pub const STATUS_POLL_TICKS: u32 = 2;
193
194pub const PIPELINE_POLL_TICKS: u32 = 4;
201
202#[derive(Debug, Clone, Copy, PartialEq, Eq)]
204pub enum ConfirmStep {
205 Armed,
206 Fired,
207 Switched,
208}
209
210#[must_use]
213pub fn is_quarantined(state: &str) -> bool {
214 state.contains("Quarantin")
215}
216
217#[derive(Debug, Default)]
221pub struct MarketStore {
222 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
223 frames: AtomicU64,
225 updates: AtomicU64,
227 reloads: AtomicU64,
229}
230
231#[derive(Debug, Clone, PartialEq)]
233pub struct CellView {
234 pub symbol: String,
235 pub interval: i64,
236 pub bars: Vec<BarData>,
237}
238
239impl MarketStore {
240 pub fn on_frame(&self, frame: &BarsFrame) {
242 let mut cells = self.cells.lock().unwrap();
243 for b in &frame.bars {
244 let key = (b.symbol.clone(), frame.interval);
245 let cell = cells.entry(key).or_default();
246 match cell.last() {
247 Some(last) if b.datetime < last.datetime => continue,
248 Some(last) if b.datetime == last.datetime => {
249 let slot = cell.last_mut().expect("len>1");
250 *slot = b.clone();
251 }
252 _ => {
253 cell.push(b.clone());
254 if cell.len() > KLINE_CELL_CAP {
255 cell.remove(0);
256 }
257 }
258 }
259 }
260 self.frames.fetch_add(1, Ordering::AcqRel);
261 }
262
263 pub fn on_update(&self, b: &BarData) {
265 let mut cells = self.cells.lock().unwrap();
266 let iv = b
267 .interval
268 .unwrap_or(ek_data::constant::Interval::MINUTE)
269 .seconds();
270 let key = (b.symbol.clone(), iv);
271 let cell = cells.entry(key).or_default();
272 match cell.last() {
273 Some(last) if b.datetime == last.datetime => {
274 let slot = cell.last_mut().expect("len>1");
275 *slot = b.clone();
276 }
277 Some(last) if b.datetime < last.datetime => {}
278 _ => {
279 cell.push(b.clone());
280 if cell.len() > KLINE_CELL_CAP {
281 cell.remove(0);
282 }
283 }
284 }
285 self.updates.fetch_add(1, Ordering::AcqRel);
286 }
287
288 pub fn note_reload(&self) {
290 self.reloads.fetch_add(1, Ordering::AcqRel);
291 }
292
293 pub fn cells(&self) -> Vec<CellView> {
295 self.cells
296 .lock()
297 .unwrap()
298 .iter()
299 .map(|((sym, iv), bars)| CellView {
300 symbol: sym.clone(),
301 interval: *iv,
302 bars: bars.clone(),
303 })
304 .collect()
305 }
306
307 pub fn counts(&self) -> (u64, u64, u64) {
309 (
310 self.frames.load(Ordering::Acquire),
311 self.updates.load(Ordering::Acquire),
312 self.reloads.load(Ordering::Acquire),
313 )
314 }
315}
316
317#[derive(Debug, Default, Clone, PartialEq)]
320pub struct ReloadForm {
321 pub interval: String,
322 pub symbol: String,
323 pub csv: String,
324}
325
326pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
333 let g = raw.trim();
334 if !g.is_empty() {
335 if g.chars().any(|c| c.is_whitespace()) {
336 return Err(format!("网关解析: 含空白:{raw:?}"));
337 }
338 return Ok(g.to_string());
339 }
340 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
341 Some(f) => Ok(f.to_string()),
342 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
343 }
344}
345
346#[derive(Debug, Clone, PartialEq)]
349pub struct CtpForm {
350 pub broker_id: String,
351 pub td_addr: String,
352 pub td_user: String,
353 pub td_password: String,
354 pub td_auth_code: String,
355 pub td_app_id: String,
356 pub md_addr: String,
357 pub md_user: String,
358 pub md_password: String,
359 pub symbols: String,
361 pub accept_futures: bool,
364 pub accept_options: bool,
365}
366
367impl Default for CtpForm {
370 fn default() -> Self {
371 Self {
372 broker_id: String::new(),
373 td_addr: String::new(),
374 td_user: String::new(),
375 td_password: String::new(),
376 td_auth_code: String::new(),
377 td_app_id: String::new(),
378 md_addr: String::new(),
379 md_user: String::new(),
380 md_password: String::new(),
381 symbols: String::new(),
382 accept_futures: true,
383 accept_options: false,
384 }
385 }
386}
387
388impl CtpForm {
389 #[must_use]
393 pub fn product_classes(&self) -> Vec<String> {
394 let mut codes = Vec::new();
395 if self.accept_futures {
396 codes.push("1".to_string());
397 }
398 if self.accept_options {
399 codes.push("2".to_string());
400 }
401 codes
402 }
403
404 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
406 ek_ctp::plugin::ConfigureReq {
407 broker_id: self.broker_id.trim().to_string(),
408 td_addr: self.td_addr.trim().to_string(),
409 td_user: self.td_user.trim().to_string(),
410 td_password: self.td_password.clone(),
411 td_auth_code: self.td_auth_code.trim().to_string(),
412 td_app_id: self.td_app_id.trim().to_string(),
413 md_addr: self.md_addr.trim().to_string(),
414 md_user: self.md_user.trim().to_string(),
415 md_password: self.md_password.clone(),
416 product_classes: self.product_classes(),
417 }
418 }
419
420 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
422 let list: Vec<String> = self
423 .symbols
424 .split([',', ',', ' '])
425 .map(|s| s.trim().to_string())
426 .filter(|s| !s.is_empty())
427 .collect();
428 if list.is_empty() {
429 return Err("订阅解析: 标的位空".to_string());
430 }
431 Ok(list)
432 }
433
434 pub fn validate(&self) -> Result<(), String> {
443 fn front(name: &str, v: &str) -> Result<(), String> {
444 let v = v.trim();
445 let Some(rest) = v.strip_prefix("tcp://") else {
446 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
447 };
448 let Some((host, port)) = rest.rsplit_once(':') else {
449 return Err(format!("{name} 缺端口:{name} {:?}", v));
450 };
451 if host.trim().is_empty() {
452 return Err(format!("{name} 主机位空:{}", v));
453 }
454 let port: u32 = port
455 .parse()
456 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
457 if port == 0 || port > 65535 {
458 return Err(format!("{name} 端口越界(1..=65535):{}", v));
459 }
460 Ok(())
461 }
462 if self.broker_id.trim().is_empty() {
463 return Err("broker 位空".to_string());
464 }
465 if self.td_user.trim().is_empty() {
466 return Err("交易账号位空".to_string());
467 }
468 front("td 前置", &self.td_addr)?;
469 if !self.md_addr.trim().is_empty() {
470 front("md 前置", &self.md_addr)?;
471 }
472 if self.td_app_id.trim().is_empty() {
473 return Err("AppID 位空(CTP 登录必需)".to_string());
474 }
475 if self.td_auth_code.trim().is_empty() {
476 return Err("AuthCode 位空(CTP 登录必需)".to_string());
477 }
478 Ok(())
479 }
480}
481
482#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
489pub enum AccountStage {
490 #[default]
492 NotAssembled,
493 Assembling,
495 Assembled,
497 AssembleFailed,
499}
500
501impl AccountStage {
502 #[must_use]
504 pub fn label(self) -> &'static str {
505 match self {
506 AccountStage::NotAssembled => "未装配",
507 AccountStage::Assembling => "装配中…",
508 AccountStage::Assembled => "已装配·登录态未知",
511 AccountStage::AssembleFailed => "装配失败",
512 }
513 }
514}
515
516#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
521pub enum LoginState {
522 #[default]
527 Unknown,
528 Online,
530 LoginFailed,
532 FrontDown,
534 QueryStarved,
536 Stopped,
538}
539
540impl LoginState {
541 #[must_use]
543 pub fn label(self) -> &'static str {
544 match self {
545 LoginState::Unknown => "登录态未知",
546 LoginState::Online => "在线",
547 LoginState::LoginFailed => "登录失败",
548 LoginState::FrontDown => "掉线",
549 LoginState::QueryStarved => "查询断供",
550 LoginState::Stopped => "已断开",
551 }
552 }
553
554 #[must_use]
556 pub fn is_alarm(self) -> bool {
557 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
558 }
559}
560
561fn count_note(label: &str, n: u32) -> Option<String> {
563 (n > 0).then(|| format!("{label} {n} 次"))
564}
565
566fn brief_counter_reason(full: &str) -> Option<String> {
573 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
574 let body = tail.trim().trim_start_matches("CTP:").trim();
575 if body.is_empty() {
576 return None;
577 }
578 let code = full
579 .rfind("代码:")
580 .map(|i| {
581 let rest = &full[i + "代码:".len()..];
582 let end = rest.find(',').unwrap_or(rest.len());
583 rest[..end].trim().to_string()
584 })
585 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
586 Some(match code {
587 Some(c) => format!("{body}({c})"),
588 None => body.to_string(),
589 })
590}
591
592#[derive(Debug, Clone, PartialEq, Default)]
598pub struct AccountRow {
599 pub gateway: String,
601 pub alias: String,
603 pub broker_id: String,
604 pub account: String,
605 pub md_source: bool,
607 pub form: CtpForm,
609 pub stage: AccountStage,
611 pub fail_note: Option<String>,
613 pub effective_classes: Vec<String>,
618 pub connected: bool,
621 pub login: LoginState,
625 pub status_note: Option<String>,
628 pub login_brief: Option<String>,
635}
636
637impl AccountRow {
638 pub fn derive(form: &CtpForm) -> Result<Self, String> {
643 let broker = form.broker_id.trim();
644 let account = form.td_user.trim();
645 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
646 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
647 })?;
648 Ok(Self {
649 gateway,
650 alias: String::new(),
651 broker_id: broker.to_string(),
652 account: account.to_string(),
653 md_source: false,
654 form: form.clone(),
655 stage: AccountStage::NotAssembled,
656 fail_note: None,
657 effective_classes: Vec::new(),
658 connected: false,
659 login: LoginState::Unknown,
660 status_note: None,
661 login_brief: None,
662 })
663 }
664
665 #[must_use]
671 pub fn status_label(&self) -> String {
672 if self.stage == AccountStage::Assembled {
673 match self.login {
674 LoginState::Unknown => "已装配·登录态未知".to_string(),
675 other => format!("已装配·{}", other.label()),
676 }
677 } else {
678 self.stage.label().to_string()
679 }
680 }
681
682 pub fn apply_status(
695 &mut self,
696 st: &ek_ctp::plugin::GatewayStatus,
697 now_ms: i64,
698 reject_limit: u32,
699 stale_secs: i64,
700 ) {
701 let td_side = !self.form.td_addr.trim().is_empty();
702 let (login, note) = if td_side {
703 if st.td_login_failed && !st.td_logged_in {
704 (
705 LoginState::LoginFailed,
706 st.td_last_error
707 .clone()
708 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
709 )
710 } else if st.td_logged_in {
711 if st.td_query_reject_streak >= reject_limit {
712 (
713 LoginState::QueryStarved,
714 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
715 )
716 } else if let Some(last) = st.td_last_account_ms {
717 let idle_ms = now_ms.saturating_sub(last);
718 if idle_ms > stale_secs.saturating_mul(1_000) {
719 (
720 LoginState::QueryStarved,
721 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
722 )
723 } else {
724 (LoginState::Online, count_note("掉线", st.td_disconnects))
725 }
726 } else {
727 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
734 match since_login_ms {
735 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
736 LoginState::QueryStarved,
737 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
738 ),
739 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
740 }
741 }
742 } else if self.login == LoginState::Stopped {
743 let extra = if st.td_disconnects > 0 {
744 format!("(此前掉线 {} 次)", st.td_disconnects)
745 } else {
746 String::new()
747 };
748 (
749 LoginState::Stopped,
750 Some(format!("会话已停{extra};『连接』可恢复")),
751 )
752 } else if st.td_disconnects > 0 {
753 (
754 LoginState::FrontDown,
755 Some(format!("掉线 {} 次", st.td_disconnects)),
756 )
757 } else {
758 (LoginState::Unknown, st.td_last_error.clone())
759 }
760 } else if st.md_logged_in {
761 (
762 LoginState::Online,
763 count_note("掉线", st.md_disconnects)
764 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
765 )
766 } else if self.login == LoginState::Stopped {
767 (
768 LoginState::Stopped,
769 Some("会话已停;『连接』可恢复".to_string()),
770 )
771 } else if st.md_disconnects > 0 {
772 (
773 LoginState::FrontDown,
774 Some(format!("掉线 {} 次", st.md_disconnects)),
775 )
776 } else {
777 (LoginState::Unknown, st.md_last_error.clone())
778 };
779 let note = if td_side && st.td_instrument_rejects > 0 {
784 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
785 match note {
786 Some(base) if base.contains("合约查询被拒") => Some(base),
788 Some(base) => Some(format!("{base};{add}")),
789 None => Some(add),
790 }
791 } else {
792 note
793 };
794 self.login = login;
795 self.status_note = note;
796 self.login_brief = if login.is_alarm() {
799 st.td_last_error
800 .as_deref()
801 .or(st.md_last_error.as_deref())
802 .and_then(brief_counter_reason)
803 } else {
804 None
805 };
806 }
807
808 #[must_use]
810 pub fn display_name(&self) -> String {
811 let a = self.alias.trim();
812 if a.is_empty() {
813 self.account.clone()
814 } else {
815 a.to_string()
816 }
817 }
818}
819
820#[derive(Debug, Clone, PartialEq, Default)]
822pub struct TradeSnap {
823 pub orders: Vec<ek_data::order::OrderData>,
824 pub trades: Vec<ek_data::trade::TradeData>,
825 pub positions: Vec<ek_data::position::PositionData>,
826 pub accounts: Vec<ek_data::account::AccountData>,
828}
829
830pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
833 let interval: i64 = form
834 .interval
835 .trim()
836 .parse()
837 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
838 let symbol = form.symbol.trim().to_string();
839 if symbol.is_empty() {
840 return Err("回灌解析: symbol 空".to_string());
841 }
842 if form.csv.trim().is_empty() {
843 return Err("回灌解析: csv 空".to_string());
844 }
845 let mut bars = Vec::new();
846 for (i, raw) in form.csv.lines().enumerate() {
847 let line = raw.trim();
848 if line.is_empty() || line.starts_with('#') {
849 continue;
850 }
851 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
852 if parts.len() != 6 {
853 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
854 }
855 let num = |s: &str| -> Result<f64, String> {
856 s.parse::<f64>()
857 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
858 };
859 let dt_s: i64 = parts[0]
860 .parse()
861 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
862 bars.push(BarData {
863 symbol: symbol.clone(),
864 datetime: dt_s * 1_000_000_000,
865 interval: Some(
866 ek_dm::wire::interval_of_seconds(interval)
867 .unwrap_or(ek_data::constant::Interval::MINUTE),
868 ),
869 open_price: num(parts[1])?,
870 high_price: num(parts[2])?,
871 low_price: num(parts[3])?,
872 close_price: num(parts[4])?,
873 volume: num(parts[5])?,
874 ..BarData::default()
875 });
876 }
877 if bars.is_empty() {
878 return Err("回灌解析: 无有效行".to_string());
879 }
880 Ok((interval, symbol, bars))
881}
882
883#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
888pub enum TopLeftTab {
889 #[default]
890 Orders,
891 Active,
892}
893
894impl TopLeftTab {
895 pub fn label(self) -> &'static str {
896 match self {
897 Self::Orders => "委托",
898 Self::Active => "未成交",
899 }
900 }
901}
902
903#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
905pub enum BottomRightTab {
906 #[default]
907 Account,
908 Logs,
909}
910
911impl BottomRightTab {
912 pub fn label(self) -> &'static str {
913 match self {
914 Self::Account => "资金",
915 Self::Logs => "日志",
916 }
917 }
918}
919
920#[derive(Debug, Clone, Copy, PartialEq)]
923pub struct DataSplit {
924 pub col: f32,
926 pub row: f32,
928}
929
930impl Default for DataSplit {
931 fn default() -> Self {
932 Self {
933 col: crate::splitters::DATA_SPLIT_DEFAULT,
934 row: crate::splitters::DATA_SPLIT_DEFAULT,
935 }
936 }
937}
938
939pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
942 ek_data::constant::Exchange::SHFE,
943 ek_data::constant::Exchange::DCE,
944 ek_data::constant::Exchange::CZCE,
945 ek_data::constant::Exchange::CFFEX,
946 ek_data::constant::Exchange::INE,
947 ek_data::constant::Exchange::GFEX,
948];
949
950#[derive(Debug, Default)]
960pub struct TickStore {
961 ticks: BTreeMap<String, ek_data::tick::TickData>,
962}
963
964impl TickStore {
965 pub fn put(&mut self, t: &ek_data::tick::TickData) {
968 self.ticks
969 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
970 }
971
972 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
974 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
975 v.sort_by(|a, b| {
976 (a.exchange.as_str().to_string(), a.symbol.clone())
977 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
978 });
979 v
980 }
981
982 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
984 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
985 for t in self.rows() {
986 let ex = t.exchange.as_str().to_string();
987 match out.last_mut() {
988 Some((e, v)) if *e == ex => v.push(t),
989 _ => out.push((ex, vec![t])),
990 }
991 }
992 out
993 }
994
995 pub fn len(&self) -> usize {
996 self.ticks.len()
997 }
998
999 pub fn is_empty(&self) -> bool {
1000 self.ticks.is_empty()
1001 }
1002}
1003
1004#[derive(Debug, Clone, PartialEq)]
1006pub struct OrderForm {
1007 pub exchange: String,
1009 pub symbol: String,
1011 pub direction: ek_data::constant::Direction,
1012 pub offset: ek_data::constant::Offset,
1013 pub order_type: ek_data::constant::OrderType,
1014 pub price: String,
1016 pub volume: String,
1017 pub gateway: String,
1019 pub follow_price: bool,
1021 pub auto_split: bool,
1024}
1025
1026impl Default for OrderForm {
1027 fn default() -> Self {
1028 Self {
1029 exchange: "SHFE".to_string(),
1030 symbol: String::new(),
1031 direction: ek_data::constant::Direction::LONG,
1032 offset: ek_data::constant::Offset::OPEN,
1033 order_type: ek_data::constant::OrderType::LIMIT,
1034 price: String::new(),
1035 volume: "1".to_string(),
1036 gateway: String::new(),
1037 follow_price: true,
1038 auto_split: true,
1039 }
1040 }
1041}
1042
1043impl OrderForm {
1044 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1046 use std::str::FromStr;
1047 let symbol = self.symbol.trim();
1048 if symbol.is_empty() {
1049 return Err("下单: 代码为空".to_string());
1050 }
1051 if symbol.contains('.') {
1052 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1053 }
1054 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1055 .map_err(|e| format!("下单: {e}"))?;
1056 let price: f64 = self
1057 .price
1058 .trim()
1059 .parse()
1060 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1061 let volume: f64 = self
1062 .volume
1063 .trim()
1064 .parse()
1065 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1066 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1067 return Err("下单: 限价单价格须为正".to_string());
1068 }
1069 if volume <= 0.0 {
1070 return Err("下单: 数量须为正".to_string());
1071 }
1072 Ok(ek_data::order_request::OrderRequest {
1073 symbol: symbol.to_string(),
1074 exchange,
1075 direction: self.direction,
1076 order_type: self.order_type,
1077 volume,
1078 price,
1079 offset: self.offset,
1080 ..ek_data::order_request::OrderRequest::default()
1081 })
1082 }
1083}
1084
1085#[derive(Debug, Default)]
1091pub struct ContractStore {
1092 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1093}
1094
1095impl ContractStore {
1096 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1097 self.contracts
1098 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1099 }
1100
1101 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1103 let p = pattern.trim();
1104 self.contracts
1105 .values()
1106 .filter(|c| {
1107 p.is_empty()
1108 || c.symbol.contains(p)
1109 || c.exchange.as_str().eq_ignore_ascii_case(p)
1110 || c.name.contains(p)
1111 })
1112 .cloned()
1113 .collect()
1114 }
1115
1116 pub fn len(&self) -> usize {
1117 self.contracts.len()
1118 }
1119
1120 pub fn is_empty(&self) -> bool {
1121 self.contracts.is_empty()
1122 }
1123}
1124
1125#[derive(Debug, Default)]
1127pub struct UiShared {
1128 logs: Mutex<LogRing>,
1129 snap: Mutex<Arc<BusSnap>>,
1130 market: MarketStore,
1131 ctp_form: Mutex<CtpForm>,
1132 trade_snap: Mutex<Arc<TradeSnap>>,
1133 tick_store: Mutex<TickStore>,
1134 order_form: Mutex<OrderForm>,
1135 data_split: Mutex<DataSplit>,
1138 selected_exchange: Mutex<String>,
1140 topleft_tab: Mutex<TopLeftTab>,
1142 bottomright_tab: Mutex<BottomRightTab>,
1144 ops_window: AtomicBool,
1145 accounts_window: AtomicBool,
1148 contracts_window: AtomicBool,
1149 contract_store: Mutex<ContractStore>,
1150 contract_query: Mutex<String>,
1151 contract_selected: Mutex<Vec<String>>,
1152 kline_windows: Mutex<Vec<String>>,
1153 rev: AtomicU64,
1154 stop: AtomicBool,
1156 cmds: Mutex<VecDeque<UiCmd>>,
1158 pending_confirm: Mutex<Option<String>>,
1160 accounts: Mutex<Vec<AccountRow>>,
1162 account_edit_window: AtomicBool,
1164 editing_account: Mutex<Option<String>>,
1166 edit_alias: Mutex<String>,
1168 edit_error: Mutex<Option<String>>,
1170 md_source: Mutex<String>,
1172 subscriptions: Mutex<Vec<String>>,
1174 row_sel: Mutex<BTreeMap<String, String>>,
1177 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1181 status_tick: AtomicU32,
1185 pipeline_tick: AtomicU32,
1187 pipeline_lines: Mutex<Vec<String>>,
1190 trade_stale: AtomicBool,
1192 status_thresholds: Mutex<Option<(u32, i64)>>,
1195 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1197 toasts: crate::toast::ToastsCell,
1200 alerted: Mutex<std::collections::BTreeSet<String>>,
1202 exit_confirm_open: AtomicBool,
1204 exit_confirmed: AtomicBool,
1206}
1207
1208impl UiShared {
1209 pub fn new() -> Self {
1211 Self::default()
1212 }
1213
1214 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1218 self.tick_store.lock().unwrap().put(t);
1219 self.bump();
1220 }
1221
1222 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1224 self.tick_store.lock().unwrap().rows()
1225 }
1226
1227 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1229 self.tick_store.lock().unwrap().groups()
1230 }
1231
1232 pub fn tick_count(&self) -> usize {
1234 self.tick_store.lock().unwrap().len()
1235 }
1236
1237 pub fn order_form(&self) -> OrderForm {
1239 self.order_form.lock().unwrap().clone()
1240 }
1241
1242 pub fn set_order_form(&self, f: OrderForm) {
1244 *self.order_form.lock().unwrap() = f;
1245 self.bump();
1246 }
1247
1248 pub fn selected_exchange(&self) -> String {
1252 let s = self.selected_exchange.lock().unwrap().clone();
1253 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1254 s
1255 } else {
1256 MARKET_ORDER[0].as_str().to_string()
1257 }
1258 }
1259
1260 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1262 *self.selected_exchange.lock().unwrap() = s.into();
1263 self.bump();
1264 }
1265
1266 pub fn topleft_tab(&self) -> TopLeftTab {
1268 *self.topleft_tab.lock().unwrap()
1269 }
1270
1271 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1273 *self.topleft_tab.lock().unwrap() = t;
1274 self.bump();
1275 }
1276
1277 pub fn bottomright_tab(&self) -> BottomRightTab {
1279 *self.bottomright_tab.lock().unwrap()
1280 }
1281
1282 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1284 *self.bottomright_tab.lock().unwrap() = t;
1285 self.bump();
1286 }
1287
1288 pub fn data_split(&self) -> DataSplit {
1290 *self.data_split.lock().unwrap()
1291 }
1292
1293 pub fn set_data_split(&self, s: DataSplit) {
1296 let s = DataSplit {
1297 col: crate::splitters::clamp_frac(s.col),
1298 row: crate::splitters::clamp_frac(s.row),
1299 };
1300 let mut g = self.data_split.lock().unwrap();
1301 if g.col != s.col || g.row != s.row {
1302 *g = s;
1303 self.bump();
1304 }
1305 }
1306
1307 pub fn reset_layout(&self) {
1311 *self.data_split.lock().unwrap() = DataSplit::default();
1312 *self.selected_exchange.lock().unwrap() = String::new();
1313 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1314 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1315 self.bump();
1316 }
1317
1318 pub fn ops_window_open(&self) -> bool {
1322 self.ops_window.load(Ordering::Acquire)
1323 }
1324
1325 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1328 self.table_sorts
1329 .lock()
1330 .unwrap()
1331 .get(table)
1332 .copied()
1333 .flatten()
1334 }
1335
1336 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1338 let mut m = self.table_sorts.lock().unwrap();
1339 let asc = match m.get(table).copied().flatten() {
1340 Some((c, a)) if c == col => !a,
1341 _ => true,
1342 };
1343 m.insert(table.to_string(), Some((col, asc)));
1344 }
1345
1346 pub fn set_ops_window_open(&self, v: bool) {
1347 self.ops_window.store(v, Ordering::Release);
1348 self.bump();
1349 }
1350
1351 pub fn accounts_window_open(&self) -> bool {
1355 self.accounts_window.load(Ordering::Acquire)
1356 }
1357
1358 pub fn set_accounts_window_open(&self, v: bool) {
1360 self.accounts_window.store(v, Ordering::Release);
1361 self.bump();
1362 }
1363
1364 pub fn account_edit_open(&self) -> bool {
1366 self.account_edit_window.load(Ordering::Acquire)
1367 }
1368
1369 pub fn editing_account(&self) -> Option<String> {
1371 self.editing_account.lock().unwrap().clone()
1372 }
1373
1374 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1378 match gateway {
1379 Some(gw) => {
1380 let Some(row) = self.account(gw) else {
1381 return false;
1382 };
1383 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1384 *self.edit_alias.lock().unwrap() = row.alias.clone();
1385 self.set_ctp_form(row.form.clone());
1386 }
1387 None => {
1388 *self.editing_account.lock().unwrap() = None;
1389 *self.edit_alias.lock().unwrap() = String::new();
1390 let draft = self.draft_from_existing();
1391 self.set_ctp_form(draft);
1392 }
1393 }
1394 *self.edit_error.lock().unwrap() = None;
1395 self.account_edit_window.store(true, Ordering::Release);
1396 self.bump();
1397 true
1398 }
1399
1400 pub fn end_account_edit(&self) {
1402 *self.editing_account.lock().unwrap() = None;
1403 *self.edit_alias.lock().unwrap() = String::new();
1404 *self.edit_error.lock().unwrap() = None;
1405 self.account_edit_window.store(false, Ordering::Release);
1406 let draft = self.draft_from_existing();
1407 self.set_ctp_form(draft);
1408 self.bump();
1409 }
1410
1411 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1417 if self.editing_account().as_deref() == Some(gateway) {
1418 self.end_account_edit();
1419 }
1420 }
1421
1422 pub fn edit_alias(&self) -> String {
1424 self.edit_alias.lock().unwrap().clone()
1425 }
1426
1427 pub fn set_edit_alias(&self, v: impl Into<String>) {
1429 *self.edit_alias.lock().unwrap() = v.into();
1430 self.bump();
1431 }
1432
1433 pub fn edit_error(&self) -> Option<String> {
1435 self.edit_error.lock().unwrap().clone()
1436 }
1437
1438 pub fn set_edit_error(&self, e: Option<String>) {
1440 *self.edit_error.lock().unwrap() = e;
1441 self.bump();
1442 }
1443
1444 pub fn account_edit_dirty(&self) -> bool {
1446 let editing = self.editing_account();
1447 let draft = self.ctp_form();
1448 match editing {
1449 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1450 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1451 }
1452 }
1453
1454 pub fn accounts(&self) -> Vec<AccountRow> {
1456 self.accounts.lock().unwrap().clone()
1457 }
1458
1459 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1461 self.accounts
1462 .lock()
1463 .unwrap()
1464 .iter()
1465 .find(|a| a.gateway == gateway)
1466 .cloned()
1467 }
1468
1469 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1475 let mut acc = self.accounts.lock().unwrap();
1476 if acc.iter().any(|a| a.gateway == row.gateway) {
1477 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1478 }
1479 let alias = row.alias.trim();
1480 if !alias.is_empty()
1481 && acc
1482 .iter()
1483 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1484 {
1485 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1486 }
1487 if acc.len() >= MAX_ACCOUNTS {
1488 return Err(format!(
1489 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1490 ));
1491 }
1492 acc.push(row);
1493 drop(acc);
1494 self.bump();
1495 Ok(())
1496 }
1497
1498 #[must_use]
1500 pub fn can_add_account(&self) -> bool {
1501 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1502 }
1503
1504 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1510 let alias = alias.trim().to_string();
1511 let mut acc = self.accounts.lock().unwrap();
1512 if !alias.is_empty()
1513 && acc
1514 .iter()
1515 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1516 {
1517 return false;
1518 }
1519 match acc.iter_mut().find(|a| a.gateway == gateway) {
1520 Some(a) => {
1521 a.alias = alias;
1522 drop(acc);
1523 self.bump();
1524 true
1525 }
1526 None => false,
1527 }
1528 }
1529
1530 pub fn note_connect_sent(&self, gateway: &str) {
1535 let mut acc = self.accounts.lock().unwrap();
1536 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1537 a.stage = AccountStage::Assembling;
1538 a.fail_note = None;
1539 a.login = LoginState::Unknown;
1540 a.status_note = None;
1541 a.login_brief = None;
1542 drop(acc);
1543 self.bump();
1544 }
1545 }
1546
1547 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1551 let mut acc = self.accounts.lock().unwrap();
1552 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1553 match result {
1554 Ok(()) => {
1555 a.stage = AccountStage::Assembled;
1556 a.fail_note = None;
1557 }
1558 Err(e) => {
1559 a.stage = AccountStage::AssembleFailed;
1560 a.fail_note = Some(e);
1561 }
1562 }
1563 drop(acc);
1564 self.bump();
1565 }
1566 }
1567
1568 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1572 let mut acc = self.accounts.lock().unwrap();
1573 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1574 a.effective_classes = classes.to_vec();
1575 drop(acc);
1576 self.bump();
1577 }
1578 }
1579
1580 #[must_use]
1584 pub fn status_thresholds(&self) -> (u32, i64) {
1585 self.status_thresholds
1586 .lock()
1587 .unwrap()
1588 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1589 }
1590
1591 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1593 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1594 self.bump();
1595 }
1596
1597 pub fn status_poll_due(&self) -> bool {
1599 let n = self.status_tick.fetch_add(1, Ordering::Relaxed) + 1;
1600 n.is_multiple_of(STATUS_POLL_TICKS)
1601 }
1602
1603 pub fn pipeline_poll_due(&self) -> bool {
1608 let n = self.pipeline_tick.fetch_add(1, Ordering::Relaxed) + 1;
1609 n.is_multiple_of(PIPELINE_POLL_TICKS)
1610 }
1611
1612 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
1614 *self.pipeline_lines.lock().unwrap() = lines;
1615 self.bump();
1616 }
1617
1618 pub fn pipeline_lines(&self) -> Vec<String> {
1620 self.pipeline_lines.lock().unwrap().clone()
1621 }
1622
1623 pub fn note_trade_frame_stale(&self, stale: bool) -> bool {
1634 let transitioned = self.trade_stale.swap(stale, Ordering::AcqRel) != stale;
1635 if transitioned && stale {
1636 self.push_alert(
1637 crate::toast::AlertLevel::Warning,
1638 "oms 四问有超时:交易监控显示的是上一帧(不是此刻没有委托)",
1639 );
1640 }
1641 transitioned
1642 }
1643
1644 pub fn trade_frame_stale(&self) -> bool {
1646 self.trade_stale.load(Ordering::Acquire)
1647 }
1648
1649 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
1658 for r in rows {
1659 if r.spilled == 0 {
1660 continue;
1661 }
1662 if self.alert_once(format!("spill:{}", r.name)) {
1663 self.push_alert(
1664 crate::toast::AlertLevel::Warning,
1665 format!(
1666 "插件 {}:邮箱溢出 {} 条(刷新停顿时行情会堆;溢出丢的是新到的事件)",
1667 r.name, r.spilled
1668 ),
1669 );
1670 }
1671 if r.spilled >= ek_core::plugin::MAILBOX_CAP as u64
1672 && self.alert_once(format!("spill-full:{}", r.name))
1673 {
1674 self.push_alert(
1675 crate::toast::AlertLevel::Error,
1676 format!(
1677 "插件 {}:溢出 {} 条 ≥ 邮箱容量 {}(丢的是一段时期的事件,不是零星几条)",
1678 r.name,
1679 r.spilled,
1680 ek_core::plugin::MAILBOX_CAP
1681 ),
1682 );
1683 }
1684 }
1685 }
1686
1687 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
1695 let (limit, stale) = self.status_thresholds();
1696 let now = ek_ctp::map::now_epoch_ms();
1697 let mut alarm: Option<(LoginState, String)> = None;
1700 let mut found = false;
1701 {
1702 let mut acc = self.accounts.lock().unwrap();
1703 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1704 found = true;
1705 a.apply_status(st, now, limit, stale);
1706 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
1707 alarm = Some((
1708 a.login,
1709 a.status_note
1710 .clone()
1711 .unwrap_or_else(|| a.login.label().to_string()),
1712 ));
1713 }
1714 }
1715 }
1716 if found {
1717 self.bump();
1718 }
1719 if let Some((login, note)) = alarm
1720 && self.alert_once(format!("{gateway}|{}", login.label()))
1721 {
1722 self.push_alert(
1723 crate::toast::AlertLevel::Error,
1724 format!("账户 {gateway}:{note}"),
1725 );
1726 }
1727 }
1728
1729 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
1733 let mut acc = self.accounts.lock().unwrap();
1734 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1735 a.login = LoginState::Unknown;
1736 a.status_note = Some(format!("状态口问不到:{why}"));
1737 a.login_brief = None;
1738 drop(acc);
1739 self.bump();
1740 }
1741 }
1742
1743 pub fn mark_account_stopped(&self, gateway: &str) {
1745 let mut acc = self.accounts.lock().unwrap();
1746 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1747 a.login = LoginState::Stopped;
1748 a.status_note = Some("会话已停;『连接』可恢复".to_string());
1749 a.login_brief = None;
1750 drop(acc);
1751 self.bump();
1752 }
1753 }
1754
1755 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
1758 let source = self.md_source();
1759 let mut f = form;
1760 if source.as_str() != gateway {
1761 f.md_addr.clear();
1762 f.md_user.clear();
1763 f.md_password.clear();
1764 }
1765 self.update_account_form(gateway, f.clone());
1766 f
1767 }
1768
1769 #[must_use]
1773 pub fn draft_from_existing(&self) -> CtpForm {
1774 let acc = self.accounts.lock().unwrap();
1775 let mut draft = CtpForm::default();
1776 if let Some(a) = acc.last() {
1777 draft.broker_id = a.form.broker_id.clone();
1778 draft.td_auth_code = a.form.td_auth_code.clone();
1779 draft.td_app_id = a.form.td_app_id.clone();
1780 draft.accept_futures = a.form.accept_futures;
1781 draft.accept_options = a.form.accept_options;
1782 }
1783 draft
1784 }
1785
1786 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
1788 let mut acc = self.accounts.lock().unwrap();
1789 match acc.iter_mut().find(|a| a.gateway == gateway) {
1790 Some(a) => {
1791 a.form = form;
1792 drop(acc);
1793 self.bump();
1794 true
1795 }
1796 None => false,
1797 }
1798 }
1799
1800 pub fn remove_account(&self, gateway: &str) -> bool {
1802 let mut acc = self.accounts.lock().unwrap();
1803 let before = acc.len();
1804 acc.retain(|a| a.gateway != gateway);
1805 let removed = acc.len() != before;
1806 if removed {
1807 let mut md = self.md_source.lock().unwrap();
1809 if md.as_str() == gateway {
1810 md.clear();
1811 }
1812 drop(acc);
1813 self.bump();
1814 }
1815 removed
1816 }
1817
1818 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
1827 let mut newly = Vec::new();
1828 {
1829 let mut acc = self.accounts.lock().unwrap();
1830 for a in acc.iter_mut() {
1831 let up = rows.iter().any(|r| r.name == a.gateway);
1832 if up && !a.connected {
1833 newly.push(a.gateway.clone());
1834 }
1835 a.connected = up;
1836 if up {
1837 a.stage = AccountStage::Assembled;
1838 a.fail_note = None;
1839 } else if a.stage == AccountStage::Assembled {
1840 a.stage = AccountStage::NotAssembled;
1841 a.login = LoginState::Unknown;
1842 a.status_note = None;
1843 a.login_brief = None;
1844 }
1845 }
1846 }
1847 if !newly.is_empty() {
1848 self.bump();
1849 }
1850 newly
1851 }
1852
1853 pub fn md_source(&self) -> String {
1855 self.md_source.lock().unwrap().clone()
1856 }
1857
1858 pub fn accounts_locked(&self) -> bool {
1860 self.accounts.lock().unwrap().iter().any(|a| a.connected)
1861 }
1862
1863 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
1865 {
1866 let acc = self.accounts.lock().unwrap();
1867 if acc.iter().any(|a| a.connected) {
1868 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
1869 }
1870 if !acc.iter().any(|a| a.gateway == gateway) {
1871 return Err(format!("账户 {gateway} 不在清册"));
1872 }
1873 }
1874 let mut acc = self.accounts.lock().unwrap();
1875 for a in acc.iter_mut() {
1876 a.md_source = a.gateway == gateway;
1877 }
1878 drop(acc);
1879 let mut md = self.md_source.lock().unwrap();
1880 *md = gateway.to_string();
1881 self.bump();
1882 Ok(())
1883 }
1884
1885 pub fn subscriptions(&self) -> Vec<String> {
1887 self.subscriptions.lock().unwrap().clone()
1888 }
1889
1890 pub fn note_subscriptions(&self, symbols: &[String]) {
1892 {
1893 let mut subs = self.subscriptions.lock().unwrap();
1894 for s in symbols {
1895 let s = s.trim();
1896 if !s.is_empty() && !subs.iter().any(|x| x == s) {
1897 subs.push(s.to_string());
1898 }
1899 }
1900 }
1901 self.bump();
1902 }
1903
1904 pub fn default_gateway(&self) -> Option<String> {
1911 let chosen = {
1912 let f = self.order_form.lock().unwrap();
1913 f.gateway.trim().to_string()
1914 };
1915 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
1916 return Some(chosen);
1917 }
1918 let acc = self.accounts.lock().unwrap();
1919 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
1920 }
1921
1922 pub fn kline_symbols(&self) -> Vec<String> {
1924 self.kline_windows.lock().unwrap().clone()
1925 }
1926
1927 pub fn open_kline(&self, symbol: impl Into<String>) {
1929 let s = symbol.into();
1930 let mut v = self.kline_windows.lock().unwrap();
1931 if !v.contains(&s) {
1932 v.push(s.clone());
1933 }
1934 drop(v);
1935 self.log_line(format!("[kline] 打开 K线窗 {s}"));
1936 }
1937
1938 pub fn close_kline(&self, symbol: &str) {
1940 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
1941 self.bump();
1942 }
1943
1944 pub fn contracts_window_open(&self) -> bool {
1948 self.contracts_window.load(Ordering::Acquire)
1949 }
1950
1951 pub fn set_contracts_window_open(&self, v: bool) {
1953 self.contracts_window.store(v, Ordering::Release);
1954 self.bump();
1955 }
1956
1957 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
1959 self.contract_store.lock().unwrap().put(c);
1960 self.bump();
1961 }
1962
1963 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1965 self.contract_store.lock().unwrap().query(pattern)
1966 }
1967
1968 pub fn contract_count(&self) -> usize {
1970 self.contract_store.lock().unwrap().len()
1971 }
1972
1973 pub fn contract_query(&self) -> String {
1975 self.contract_query.lock().unwrap().clone()
1976 }
1977
1978 pub fn set_contract_query(&self, q: impl Into<String>) {
1980 *self.contract_query.lock().unwrap() = q.into();
1981 self.bump();
1982 }
1983
1984 pub fn toggle_contract_selection(&self, en_symbol: &str) {
1986 let mut v = self.contract_selected.lock().unwrap();
1987 match v.iter().position(|s| s == en_symbol) {
1988 Some(i) => {
1989 v.remove(i);
1990 }
1991 None => v.push(en_symbol.to_string()),
1992 }
1993 self.bump();
1994 }
1995
1996 pub fn selected_contracts(&self) -> Vec<String> {
1998 self.contract_selected.lock().unwrap().clone()
1999 }
2000
2001 pub fn log_line(&self, line: impl Into<String>) {
2003 self.logs.lock().unwrap().push(line);
2004 self.bump();
2005 }
2006
2007 pub fn clear_logs(&self) {
2009 self.logs.lock().unwrap().clear();
2010 self.bump();
2011 }
2012
2013 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2015 let r = self.logs.lock().unwrap();
2016 (r.snapshot(), r.spilled())
2017 }
2018
2019 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2021 let r = self.logs.lock().unwrap();
2022 (r.entries(), r.spilled())
2023 }
2024
2025 pub fn logs_len(&self) -> usize {
2027 self.logs.lock().unwrap().len()
2028 }
2029
2030 pub fn set_snapshot(&self, snap: BusSnap) {
2032 *self.snap.lock().unwrap() = Arc::new(snap);
2033 self.bump();
2034 }
2035
2036 pub fn snapshot(&self) -> Arc<BusSnap> {
2038 Arc::clone(&self.snap.lock().unwrap())
2039 }
2040
2041 pub fn rev(&self) -> u64 {
2043 self.rev.load(Ordering::Acquire)
2044 }
2045
2046 pub fn bump(&self) {
2048 self.rev.fetch_add(1, Ordering::AcqRel);
2049 }
2050
2051 pub fn request_stop(&self) {
2053 self.stop.store(true, Ordering::Release);
2054 }
2055
2056 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2059 self.pending_alerts
2060 .lock()
2061 .unwrap_or_else(std::sync::PoisonError::into_inner)
2062 .push((level, text.into()));
2063 }
2064
2065 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2067 let mut q = self
2068 .pending_alerts
2069 .lock()
2070 .unwrap_or_else(std::sync::PoisonError::into_inner);
2071 std::mem::take(&mut *q)
2072 }
2073
2074 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2076 self.toasts.lock()
2077 }
2078
2079 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2081 let key = key.into();
2082 self.alerted
2083 .lock()
2084 .unwrap_or_else(std::sync::PoisonError::into_inner)
2085 .insert(key)
2086 }
2087
2088 #[must_use]
2090 pub fn pending_alert_count(&self) -> usize {
2091 self.pending_alerts
2092 .lock()
2093 .unwrap_or_else(std::sync::PoisonError::into_inner)
2094 .len()
2095 }
2096
2097 #[must_use]
2099 pub fn alerted_count(&self) -> usize {
2100 self.alerted
2101 .lock()
2102 .unwrap_or_else(std::sync::PoisonError::into_inner)
2103 .len()
2104 }
2105
2106 pub fn exit_confirm_open(&self) -> bool {
2108 self.exit_confirm_open
2109 .load(std::sync::atomic::Ordering::Relaxed)
2110 }
2111
2112 pub fn open_exit_confirm(&self) {
2113 self.exit_confirm_open
2114 .store(true, std::sync::atomic::Ordering::Relaxed);
2115 }
2116
2117 pub fn close_exit_confirm(&self) {
2118 self.exit_confirm_open
2119 .store(false, std::sync::atomic::Ordering::Relaxed);
2120 }
2121
2122 pub fn exit_confirmed(&self) -> bool {
2123 self.exit_confirmed
2124 .load(std::sync::atomic::Ordering::Relaxed)
2125 }
2126
2127 pub fn confirm_exit(&self) {
2128 self.exit_confirmed
2129 .store(true, std::sync::atomic::Ordering::Relaxed);
2130 }
2131
2132 pub fn is_stop(&self) -> bool {
2133 self.stop.load(Ordering::Acquire)
2134 }
2135
2136 pub fn push_cmd(&self, cmd: UiCmd) {
2140 self.cmds.lock().unwrap().push_back(cmd);
2141 self.bump();
2142 }
2143
2144 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2146 self.cmds.lock().unwrap().drain(..).collect()
2147 }
2148
2149 pub fn pending_confirm(&self) -> Option<String> {
2151 self.pending_confirm.lock().unwrap().clone()
2152 }
2153
2154 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2156 let mut g = self.pending_confirm.lock().unwrap();
2157 match g.take() {
2158 Some(p) if p == name => ConfirmStep::Fired,
2159 Some(_) => {
2160 *g = Some(name.to_string());
2161 ConfirmStep::Switched
2162 }
2163 None => {
2164 *g = Some(name.to_string());
2165 ConfirmStep::Armed
2166 }
2167 }
2168 }
2169
2170 pub fn confirm_cancel(&self) {
2172 *self.pending_confirm.lock().unwrap() = None;
2173 self.bump();
2174 }
2175
2176 pub fn market(&self) -> &MarketStore {
2180 &self.market
2181 }
2182
2183 pub fn set_trade_snap(&self, ts: TradeSnap) {
2187 *self.trade_snap.lock().unwrap() = Arc::new(ts);
2188 self.bump();
2189 }
2190
2191 pub fn trade_snap(&self) -> Arc<TradeSnap> {
2193 self.trade_snap.lock().unwrap().clone()
2194 }
2195
2196 pub fn ctp_form(&self) -> CtpForm {
2200 self.ctp_form.lock().unwrap().clone()
2201 }
2202
2203 pub fn set_ctp_form(&self, f: CtpForm) {
2205 *self.ctp_form.lock().unwrap() = f;
2206 self.bump();
2207 }
2208
2209 pub fn row_selected(&self, table: &str, key: &str) -> bool {
2213 self.row_sel
2214 .lock()
2215 .unwrap()
2216 .get(table)
2217 .is_some_and(|k| k == key)
2218 }
2219
2220 pub fn select_row(&self, table: &str, key: &str) {
2222 self.row_sel
2223 .lock()
2224 .unwrap()
2225 .insert(table.to_string(), key.to_string());
2226 }
2227}