1use std::collections::{BTreeMap, HashSet, VecDeque};
7use std::sync::atomic::{AtomicBool, 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
222#[must_use]
228pub fn mailbox_cap_of(name: &str) -> usize {
229 if name == crate::plugin::UI_NAME {
230 crate::plugin::UI_MAILBOX_CAP
231 } else {
232 ek_core::plugin::MAILBOX_CAP
233 }
234}
235
236#[derive(Debug, Clone, Copy, PartialEq, Eq)]
238pub enum ConfirmStep {
239 Armed,
240 Fired,
241 Switched,
242}
243
244#[must_use]
247pub fn is_quarantined(state: &str) -> bool {
248 state.contains("Quarantin")
249}
250
251#[derive(Debug, Default)]
255pub struct MarketStore {
256 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
257 frames: AtomicU64,
259 updates: AtomicU64,
261 reloads: AtomicU64,
263}
264
265#[derive(Debug, Clone, PartialEq)]
267pub struct CellView {
268 pub symbol: String,
269 pub interval: i64,
270 pub bars: Vec<BarData>,
271}
272
273impl MarketStore {
274 pub fn on_frame(&self, frame: &BarsFrame) {
276 let mut cells = self.cells.lock().unwrap();
277 for b in &frame.bars {
278 let key = (b.symbol.clone(), frame.interval);
279 let cell = cells.entry(key).or_default();
280 match cell.last() {
281 Some(last) if b.datetime < last.datetime => continue,
282 Some(last) if b.datetime == last.datetime => {
283 let slot = cell.last_mut().expect("len>1");
284 *slot = b.clone();
285 }
286 _ => {
287 cell.push(b.clone());
288 if cell.len() > KLINE_CELL_CAP {
289 cell.remove(0);
290 }
291 }
292 }
293 }
294 self.frames.fetch_add(1, Ordering::AcqRel);
295 }
296
297 pub fn on_update(&self, b: &BarData) {
299 let mut cells = self.cells.lock().unwrap();
300 let iv = b
301 .interval
302 .unwrap_or(ek_data::constant::Interval::MINUTE)
303 .seconds();
304 let key = (b.symbol.clone(), iv);
305 let cell = cells.entry(key).or_default();
306 match cell.last() {
307 Some(last) if b.datetime == last.datetime => {
308 let slot = cell.last_mut().expect("len>1");
309 *slot = b.clone();
310 }
311 Some(last) if b.datetime < last.datetime => {}
312 _ => {
313 cell.push(b.clone());
314 if cell.len() > KLINE_CELL_CAP {
315 cell.remove(0);
316 }
317 }
318 }
319 self.updates.fetch_add(1, Ordering::AcqRel);
320 }
321
322 pub fn note_reload(&self) {
324 self.reloads.fetch_add(1, Ordering::AcqRel);
325 }
326
327 pub fn cells(&self) -> Vec<CellView> {
329 self.cells
330 .lock()
331 .unwrap()
332 .iter()
333 .map(|((sym, iv), bars)| CellView {
334 symbol: sym.clone(),
335 interval: *iv,
336 bars: bars.clone(),
337 })
338 .collect()
339 }
340
341 pub fn counts(&self) -> (u64, u64, u64) {
343 (
344 self.frames.load(Ordering::Acquire),
345 self.updates.load(Ordering::Acquire),
346 self.reloads.load(Ordering::Acquire),
347 )
348 }
349}
350
351#[derive(Debug, Default, Clone, PartialEq)]
354pub struct ReloadForm {
355 pub interval: String,
356 pub symbol: String,
357 pub csv: String,
358}
359
360pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
367 let g = raw.trim();
368 if !g.is_empty() {
369 if g.chars().any(|c| c.is_whitespace()) {
370 return Err(format!("网关解析: 含空白:{raw:?}"));
371 }
372 return Ok(g.to_string());
373 }
374 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
375 Some(f) => Ok(f.to_string()),
376 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
377 }
378}
379
380#[derive(Debug, Clone, PartialEq)]
383pub struct CtpForm {
384 pub broker_id: String,
385 pub td_addr: String,
386 pub td_user: String,
387 pub td_password: String,
388 pub td_auth_code: String,
389 pub td_app_id: String,
390 pub md_addr: String,
391 pub md_user: String,
392 pub md_password: String,
393 pub symbols: String,
395 pub accept_futures: bool,
398 pub accept_options: bool,
399}
400
401impl Default for CtpForm {
404 fn default() -> Self {
405 Self {
406 broker_id: String::new(),
407 td_addr: String::new(),
408 td_user: String::new(),
409 td_password: String::new(),
410 td_auth_code: String::new(),
411 td_app_id: String::new(),
412 md_addr: String::new(),
413 md_user: String::new(),
414 md_password: String::new(),
415 symbols: String::new(),
416 accept_futures: true,
417 accept_options: false,
418 }
419 }
420}
421
422impl CtpForm {
423 #[must_use]
427 pub fn product_classes(&self) -> Vec<String> {
428 let mut codes = Vec::new();
429 if self.accept_futures {
430 codes.push("1".to_string());
431 }
432 if self.accept_options {
433 codes.push("2".to_string());
434 }
435 codes
436 }
437
438 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
440 ek_ctp::plugin::ConfigureReq {
441 broker_id: self.broker_id.trim().to_string(),
442 td_addr: self.td_addr.trim().to_string(),
443 td_user: self.td_user.trim().to_string(),
444 td_password: self.td_password.clone(),
445 td_auth_code: self.td_auth_code.trim().to_string(),
446 td_app_id: self.td_app_id.trim().to_string(),
447 md_addr: self.md_addr.trim().to_string(),
448 md_user: self.md_user.trim().to_string(),
449 md_password: self.md_password.clone(),
450 product_classes: self.product_classes(),
451 }
452 }
453
454 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
456 let list: Vec<String> = self
457 .symbols
458 .split([',', ',', ' '])
459 .map(|s| s.trim().to_string())
460 .filter(|s| !s.is_empty())
461 .collect();
462 if list.is_empty() {
463 return Err("订阅解析: 标的位空".to_string());
464 }
465 Ok(list)
466 }
467
468 pub fn validate(&self) -> Result<(), String> {
477 fn front(name: &str, v: &str) -> Result<(), String> {
478 let v = v.trim();
479 let Some(rest) = v.strip_prefix("tcp://") else {
480 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
481 };
482 let Some((host, port)) = rest.rsplit_once(':') else {
483 return Err(format!("{name} 缺端口:{name} {:?}", v));
484 };
485 if host.trim().is_empty() {
486 return Err(format!("{name} 主机位空:{}", v));
487 }
488 let port: u32 = port
489 .parse()
490 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
491 if port == 0 || port > 65535 {
492 return Err(format!("{name} 端口越界(1..=65535):{}", v));
493 }
494 Ok(())
495 }
496 if self.broker_id.trim().is_empty() {
497 return Err("broker 位空".to_string());
498 }
499 if self.td_user.trim().is_empty() {
500 return Err("交易账号位空".to_string());
501 }
502 front("td 前置", &self.td_addr)?;
503 if !self.md_addr.trim().is_empty() {
504 front("md 前置", &self.md_addr)?;
505 }
506 if self.td_app_id.trim().is_empty() {
507 return Err("AppID 位空(CTP 登录必需)".to_string());
508 }
509 if self.td_auth_code.trim().is_empty() {
510 return Err("AuthCode 位空(CTP 登录必需)".to_string());
511 }
512 Ok(())
513 }
514}
515
516#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
523pub enum AccountStage {
524 #[default]
526 NotAssembled,
527 Assembling,
529 Assembled,
531 AssembleFailed,
533}
534
535impl AccountStage {
536 #[must_use]
538 pub fn label(self) -> &'static str {
539 match self {
540 AccountStage::NotAssembled => "未装配",
541 AccountStage::Assembling => "装配中…",
542 AccountStage::Assembled => "已装配·登录态未知",
545 AccountStage::AssembleFailed => "装配失败",
546 }
547 }
548}
549
550#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
555pub enum LoginState {
556 #[default]
561 Unknown,
562 Online,
564 LoginFailed,
566 FrontDown,
568 QueryStarved,
570 Stopped,
572}
573
574impl LoginState {
575 #[must_use]
577 pub fn label(self) -> &'static str {
578 match self {
579 LoginState::Unknown => "登录态未知",
580 LoginState::Online => "在线",
581 LoginState::LoginFailed => "登录失败",
582 LoginState::FrontDown => "掉线",
583 LoginState::QueryStarved => "查询断供",
584 LoginState::Stopped => "已断开",
585 }
586 }
587
588 #[must_use]
590 pub fn is_alarm(self) -> bool {
591 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
592 }
593}
594
595fn count_note(label: &str, n: u32) -> Option<String> {
597 (n > 0).then(|| format!("{label} {n} 次"))
598}
599
600fn brief_counter_reason(full: &str) -> Option<String> {
607 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
608 let body = tail.trim().trim_start_matches("CTP:").trim();
609 if body.is_empty() {
610 return None;
611 }
612 let code = full
613 .rfind("代码:")
614 .map(|i| {
615 let rest = &full[i + "代码:".len()..];
616 let end = rest.find(',').unwrap_or(rest.len());
617 rest[..end].trim().to_string()
618 })
619 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
620 Some(match code {
621 Some(c) => format!("{body}({c})"),
622 None => body.to_string(),
623 })
624}
625
626#[derive(Debug, Clone, PartialEq, Default)]
632pub struct AccountRow {
633 pub gateway: String,
635 pub alias: String,
637 pub broker_id: String,
638 pub account: String,
639 pub md_source: bool,
641 pub form: CtpForm,
643 pub stage: AccountStage,
645 pub fail_note: Option<String>,
647 pub effective_classes: Vec<String>,
652 pub connected: bool,
655 pub login: LoginState,
659 pub status_note: Option<String>,
662 pub login_brief: Option<String>,
669 pub chain_note: Option<String>,
677}
678
679impl AccountRow {
680 pub fn derive(form: &CtpForm) -> Result<Self, String> {
685 let broker = form.broker_id.trim();
686 let account = form.td_user.trim();
687 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
688 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
689 })?;
690 Ok(Self {
691 gateway,
692 alias: String::new(),
693 broker_id: broker.to_string(),
694 account: account.to_string(),
695 md_source: false,
696 form: form.clone(),
697 stage: AccountStage::NotAssembled,
698 fail_note: None,
699 effective_classes: Vec::new(),
700 connected: false,
701 login: LoginState::Unknown,
702 status_note: None,
703 login_brief: None,
704 chain_note: None,
705 })
706 }
707
708 #[must_use]
714 pub fn status_label(&self) -> String {
715 if self.stage == AccountStage::Assembled {
716 match self.login {
717 LoginState::Unknown => "已装配·登录态未知".to_string(),
718 other => format!("已装配·{}", other.label()),
719 }
720 } else {
721 self.stage.label().to_string()
722 }
723 }
724
725 pub fn apply_status(
738 &mut self,
739 st: &ek_ctp::plugin::GatewayStatus,
740 now_ms: i64,
741 reject_limit: u32,
742 stale_secs: i64,
743 ) {
744 let td_side = !self.form.td_addr.trim().is_empty();
745 let (login, note) = if td_side {
746 if st.td_login_failed && !st.td_logged_in {
747 (
748 LoginState::LoginFailed,
749 st.td_last_error
750 .clone()
751 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
752 )
753 } else if st.td_logged_in {
754 if st.td_query_reject_streak >= reject_limit {
755 (
756 LoginState::QueryStarved,
757 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
758 )
759 } else if let Some(last) = st.td_last_account_ms {
760 let idle_ms = now_ms.saturating_sub(last);
761 if idle_ms > stale_secs.saturating_mul(1_000) {
762 (
763 LoginState::QueryStarved,
764 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
765 )
766 } else {
767 (LoginState::Online, count_note("掉线", st.td_disconnects))
768 }
769 } else {
770 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
777 match since_login_ms {
778 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
779 LoginState::QueryStarved,
780 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
781 ),
782 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
783 }
784 }
785 } else if self.login == LoginState::Stopped {
786 let extra = if st.td_disconnects > 0 {
787 format!("(此前掉线 {} 次)", st.td_disconnects)
788 } else {
789 String::new()
790 };
791 (
792 LoginState::Stopped,
793 Some(format!("会话已停{extra};『连接』可恢复")),
794 )
795 } else if st.td_disconnects > 0 {
796 (
797 LoginState::FrontDown,
798 Some(format!("掉线 {} 次", st.td_disconnects)),
799 )
800 } else {
801 (LoginState::Unknown, st.td_last_error.clone())
802 }
803 } else if st.md_logged_in {
804 (
805 LoginState::Online,
806 count_note("掉线", st.md_disconnects)
807 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
808 )
809 } else if self.login == LoginState::Stopped {
810 (
811 LoginState::Stopped,
812 Some("会话已停;『连接』可恢复".to_string()),
813 )
814 } else if st.md_disconnects > 0 {
815 (
816 LoginState::FrontDown,
817 Some(format!("掉线 {} 次", st.md_disconnects)),
818 )
819 } else {
820 (LoginState::Unknown, st.md_last_error.clone())
821 };
822 let note = if td_side && st.td_instrument_rejects > 0 {
827 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
828 match note {
829 Some(base) if base.contains("合约查询被拒") => Some(base),
831 Some(base) => Some(format!("{base};{add}")),
832 None => Some(add),
833 }
834 } else {
835 note
836 };
837 self.login = login;
838 self.status_note = note;
839 self.login_brief = if login.is_alarm() {
842 st.td_last_error
843 .as_deref()
844 .or(st.md_last_error.as_deref())
845 .and_then(brief_counter_reason)
846 } else {
847 None
848 };
849 }
850
851 #[must_use]
853 pub fn display_name(&self) -> String {
854 let a = self.alias.trim();
855 if a.is_empty() {
856 self.account.clone()
857 } else {
858 a.to_string()
859 }
860 }
861}
862
863#[derive(Debug, Clone, PartialEq, Default)]
866pub struct TradeSnap {
867 pub orders: Vec<ek_data::order::OrderData>,
868 pub trades: Vec<ek_data::trade::TradeData>,
869 pub positions: Vec<ek_data::position::PositionData>,
870 pub accounts: Vec<ek_data::account::AccountData>,
872}
873
874pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
877 let interval: i64 = form
878 .interval
879 .trim()
880 .parse()
881 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
882 let symbol = form.symbol.trim().to_string();
883 if symbol.is_empty() {
884 return Err("回灌解析: symbol 空".to_string());
885 }
886 if form.csv.trim().is_empty() {
887 return Err("回灌解析: csv 空".to_string());
888 }
889 let mut bars = Vec::new();
890 for (i, raw) in form.csv.lines().enumerate() {
891 let line = raw.trim();
892 if line.is_empty() || line.starts_with('#') {
893 continue;
894 }
895 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
896 if parts.len() != 6 {
897 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
898 }
899 let num = |s: &str| -> Result<f64, String> {
900 s.parse::<f64>()
901 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
902 };
903 let dt_s: i64 = parts[0]
904 .parse()
905 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
906 bars.push(BarData {
907 symbol: symbol.clone(),
908 datetime: dt_s * 1_000_000_000,
909 interval: Some(
910 ek_dm::wire::interval_of_seconds(interval)
911 .unwrap_or(ek_data::constant::Interval::MINUTE),
912 ),
913 open_price: num(parts[1])?,
914 high_price: num(parts[2])?,
915 low_price: num(parts[3])?,
916 close_price: num(parts[4])?,
917 volume: num(parts[5])?,
918 ..BarData::default()
919 });
920 }
921 if bars.is_empty() {
922 return Err("回灌解析: 无有效行".to_string());
923 }
924 Ok((interval, symbol, bars))
925}
926
927#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
932pub enum TopLeftTab {
933 #[default]
934 Orders,
935 Active,
936}
937
938impl TopLeftTab {
939 pub fn label(self) -> &'static str {
940 match self {
941 Self::Orders => "委托",
942 Self::Active => "未成交",
943 }
944 }
945}
946
947#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
949pub enum BottomRightTab {
950 #[default]
951 Account,
952 Logs,
953}
954
955impl BottomRightTab {
956 pub fn label(self) -> &'static str {
957 match self {
958 Self::Account => "资金",
959 Self::Logs => "日志",
960 }
961 }
962}
963
964#[derive(Debug, Clone, Copy, PartialEq)]
967pub struct DataSplit {
968 pub col: f32,
970 pub row: f32,
972}
973
974impl Default for DataSplit {
975 fn default() -> Self {
976 Self {
977 col: crate::splitters::DATA_SPLIT_DEFAULT,
978 row: crate::splitters::DATA_SPLIT_DEFAULT,
979 }
980 }
981}
982
983pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
986 ek_data::constant::Exchange::SHFE,
987 ek_data::constant::Exchange::DCE,
988 ek_data::constant::Exchange::CZCE,
989 ek_data::constant::Exchange::CFFEX,
990 ek_data::constant::Exchange::INE,
991 ek_data::constant::Exchange::GFEX,
992];
993
994#[derive(Debug, Default)]
1004pub struct TickStore {
1005 ticks: BTreeMap<String, ek_data::tick::TickData>,
1006}
1007
1008impl TickStore {
1009 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1012 self.ticks
1013 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1014 }
1015
1016 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1018 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1019 v.sort_by(|a, b| {
1020 (a.exchange.as_str().to_string(), a.symbol.clone())
1021 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1022 });
1023 v
1024 }
1025
1026 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1028 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1029 for t in self.rows() {
1030 let ex = t.exchange.as_str().to_string();
1031 match out.last_mut() {
1032 Some((e, v)) if *e == ex => v.push(t),
1033 _ => out.push((ex, vec![t])),
1034 }
1035 }
1036 out
1037 }
1038
1039 pub fn len(&self) -> usize {
1040 self.ticks.len()
1041 }
1042
1043 pub fn is_empty(&self) -> bool {
1044 self.ticks.is_empty()
1045 }
1046}
1047
1048#[derive(Debug, Clone, PartialEq)]
1050pub struct OrderForm {
1051 pub exchange: String,
1053 pub symbol: String,
1055 pub direction: ek_data::constant::Direction,
1056 pub offset: ek_data::constant::Offset,
1057 pub order_type: ek_data::constant::OrderType,
1058 pub price: String,
1060 pub volume: String,
1061 pub gateway: String,
1063 pub follow_price: bool,
1065 pub auto_split: bool,
1068}
1069
1070impl Default for OrderForm {
1071 fn default() -> Self {
1072 Self {
1073 exchange: "SHFE".to_string(),
1074 symbol: String::new(),
1075 direction: ek_data::constant::Direction::LONG,
1076 offset: ek_data::constant::Offset::OPEN,
1077 order_type: ek_data::constant::OrderType::LIMIT,
1078 price: String::new(),
1079 volume: "1".to_string(),
1080 gateway: String::new(),
1081 follow_price: true,
1082 auto_split: true,
1083 }
1084 }
1085}
1086
1087impl OrderForm {
1088 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1090 use std::str::FromStr;
1091 let symbol = self.symbol.trim();
1092 if symbol.is_empty() {
1093 return Err("下单: 代码为空".to_string());
1094 }
1095 if symbol.contains('.') {
1096 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1097 }
1098 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1099 .map_err(|e| format!("下单: {e}"))?;
1100 let price: f64 = self
1101 .price
1102 .trim()
1103 .parse()
1104 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1105 let volume: f64 = self
1106 .volume
1107 .trim()
1108 .parse()
1109 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1110 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1111 return Err("下单: 限价单价格须为正".to_string());
1112 }
1113 if volume <= 0.0 {
1114 return Err("下单: 数量须为正".to_string());
1115 }
1116 Ok(ek_data::order_request::OrderRequest {
1117 symbol: symbol.to_string(),
1118 exchange,
1119 direction: self.direction,
1120 order_type: self.order_type,
1121 volume,
1122 price,
1123 offset: self.offset,
1124 ..ek_data::order_request::OrderRequest::default()
1125 })
1126 }
1127}
1128
1129#[derive(Debug, Default)]
1135pub struct ContractStore {
1136 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1137}
1138
1139impl ContractStore {
1140 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1141 self.contracts
1142 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1143 }
1144
1145 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1147 let p = pattern.trim();
1148 self.contracts
1149 .values()
1150 .filter(|c| {
1151 p.is_empty()
1152 || c.symbol.contains(p)
1153 || c.exchange.as_str().eq_ignore_ascii_case(p)
1154 || c.name.contains(p)
1155 })
1156 .cloned()
1157 .collect()
1158 }
1159
1160 pub fn len(&self) -> usize {
1161 self.contracts.len()
1162 }
1163
1164 pub fn is_empty(&self) -> bool {
1165 self.contracts.is_empty()
1166 }
1167}
1168
1169#[derive(Debug, Default)]
1171pub struct UiShared {
1172 logs: Mutex<LogRing>,
1173 snap: Mutex<Arc<BusSnap>>,
1174 market: MarketStore,
1175 ctp_form: Mutex<CtpForm>,
1176 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1179 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1180 trade_ids: Mutex<HashSet<String>>,
1188 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1189 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1190 trade_seq: AtomicU64,
1192 trade_dropped: AtomicU64,
1194 trade_dirty: AtomicBool,
1196 trade_snap: Mutex<Arc<TradeSnap>>,
1198 tick_store: Mutex<TickStore>,
1199 order_form: Mutex<OrderForm>,
1200 data_split: Mutex<DataSplit>,
1203 selected_exchange: Mutex<String>,
1205 topleft_tab: Mutex<TopLeftTab>,
1207 bottomright_tab: Mutex<BottomRightTab>,
1209 ops_window: AtomicBool,
1210 accounts_window: AtomicBool,
1213 contracts_window: AtomicBool,
1214 contract_store: Mutex<ContractStore>,
1215 contract_query: Mutex<String>,
1216 contract_selected: Mutex<Vec<String>>,
1217 kline_windows: Mutex<Vec<String>>,
1218 rev: AtomicU64,
1219 stop: AtomicBool,
1221 cmds: Mutex<VecDeque<UiCmd>>,
1223 pending_confirm: Mutex<Option<String>>,
1225 accounts: Mutex<Vec<AccountRow>>,
1227 account_edit_window: AtomicBool,
1229 editing_account: Mutex<Option<String>>,
1231 edit_alias: Mutex<String>,
1233 edit_error: Mutex<Option<String>>,
1235 md_source: Mutex<String>,
1237 subscriptions: Mutex<Vec<String>>,
1239 row_sel: Mutex<BTreeMap<String, String>>,
1242 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1246 status_cursor: AtomicUsize,
1250 list_tick: AtomicU32,
1252 list_due: AtomicBool,
1256 seed_due: AtomicBool,
1260 pipeline_due: AtomicBool,
1262 cfg_due: AtomicBool,
1264 cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
1267 cfg_delete_arm: Mutex<Option<String>>,
1269 bad_payload: Mutex<BTreeMap<String, u64>>,
1271 pipeline_lines: Mutex<Vec<String>>,
1274 status_thresholds: Mutex<Option<(u32, i64)>>,
1277 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1279 toasts: crate::toast::ToastsCell,
1282 alerted: Mutex<std::collections::BTreeSet<String>>,
1284 exit_confirm_open: AtomicBool,
1286 exit_confirmed: AtomicBool,
1288}
1289
1290impl UiShared {
1291 pub fn new() -> Self {
1293 Self::default()
1294 }
1295
1296 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1300 self.tick_store.lock().unwrap().put(t);
1301 self.bump();
1302 }
1303
1304 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1306 self.tick_store.lock().unwrap().rows()
1307 }
1308
1309 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1311 self.tick_store.lock().unwrap().groups()
1312 }
1313
1314 pub fn tick_count(&self) -> usize {
1316 self.tick_store.lock().unwrap().len()
1317 }
1318
1319 pub fn order_form(&self) -> OrderForm {
1321 self.order_form.lock().unwrap().clone()
1322 }
1323
1324 pub fn set_order_form(&self, f: OrderForm) {
1326 *self.order_form.lock().unwrap() = f;
1327 self.bump();
1328 }
1329
1330 pub fn selected_exchange(&self) -> String {
1334 let s = self.selected_exchange.lock().unwrap().clone();
1335 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1336 s
1337 } else {
1338 MARKET_ORDER[0].as_str().to_string()
1339 }
1340 }
1341
1342 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1344 *self.selected_exchange.lock().unwrap() = s.into();
1345 self.bump();
1346 }
1347
1348 pub fn topleft_tab(&self) -> TopLeftTab {
1350 *self.topleft_tab.lock().unwrap()
1351 }
1352
1353 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1355 *self.topleft_tab.lock().unwrap() = t;
1356 self.bump();
1357 }
1358
1359 pub fn bottomright_tab(&self) -> BottomRightTab {
1361 *self.bottomright_tab.lock().unwrap()
1362 }
1363
1364 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1366 *self.bottomright_tab.lock().unwrap() = t;
1367 self.bump();
1368 }
1369
1370 pub fn data_split(&self) -> DataSplit {
1372 *self.data_split.lock().unwrap()
1373 }
1374
1375 pub fn set_data_split(&self, s: DataSplit) {
1378 let s = DataSplit {
1379 col: crate::splitters::clamp_frac(s.col),
1380 row: crate::splitters::clamp_frac(s.row),
1381 };
1382 let mut g = self.data_split.lock().unwrap();
1383 if g.col != s.col || g.row != s.row {
1384 *g = s;
1385 self.bump();
1386 }
1387 }
1388
1389 pub fn reset_layout(&self) {
1393 *self.data_split.lock().unwrap() = DataSplit::default();
1394 *self.selected_exchange.lock().unwrap() = String::new();
1395 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1396 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1397 self.bump();
1398 }
1399
1400 pub fn ops_window_open(&self) -> bool {
1404 self.ops_window.load(Ordering::Acquire)
1405 }
1406
1407 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1410 self.table_sorts
1411 .lock()
1412 .unwrap()
1413 .get(table)
1414 .copied()
1415 .flatten()
1416 }
1417
1418 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1420 let mut m = self.table_sorts.lock().unwrap();
1421 let asc = match m.get(table).copied().flatten() {
1422 Some((c, a)) if c == col => !a,
1423 _ => true,
1424 };
1425 m.insert(table.to_string(), Some((col, asc)));
1426 }
1427
1428 pub fn set_ops_window_open(&self, v: bool) {
1429 self.ops_window.store(v, Ordering::Release);
1430 self.bump();
1431 }
1432
1433 pub fn accounts_window_open(&self) -> bool {
1437 self.accounts_window.load(Ordering::Acquire)
1438 }
1439
1440 pub fn set_accounts_window_open(&self, v: bool) {
1442 self.accounts_window.store(v, Ordering::Release);
1443 self.bump();
1444 }
1445
1446 pub fn account_edit_open(&self) -> bool {
1448 self.account_edit_window.load(Ordering::Acquire)
1449 }
1450
1451 pub fn editing_account(&self) -> Option<String> {
1453 self.editing_account.lock().unwrap().clone()
1454 }
1455
1456 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1460 match gateway {
1461 Some(gw) => {
1462 let Some(row) = self.account(gw) else {
1463 return false;
1464 };
1465 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1466 *self.edit_alias.lock().unwrap() = row.alias.clone();
1467 self.set_ctp_form(row.form.clone());
1468 }
1469 None => {
1470 *self.editing_account.lock().unwrap() = None;
1471 *self.edit_alias.lock().unwrap() = String::new();
1472 let draft = self.draft_from_existing();
1473 self.set_ctp_form(draft);
1474 }
1475 }
1476 *self.edit_error.lock().unwrap() = None;
1477 self.account_edit_window.store(true, Ordering::Release);
1478 self.bump();
1479 true
1480 }
1481
1482 pub fn end_account_edit(&self) {
1484 *self.editing_account.lock().unwrap() = None;
1485 *self.edit_alias.lock().unwrap() = String::new();
1486 *self.edit_error.lock().unwrap() = None;
1487 self.account_edit_window.store(false, Ordering::Release);
1488 let draft = self.draft_from_existing();
1489 self.set_ctp_form(draft);
1490 self.bump();
1491 }
1492
1493 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1499 if self.editing_account().as_deref() == Some(gateway) {
1500 self.end_account_edit();
1501 }
1502 }
1503
1504 pub fn edit_alias(&self) -> String {
1506 self.edit_alias.lock().unwrap().clone()
1507 }
1508
1509 pub fn set_edit_alias(&self, v: impl Into<String>) {
1511 *self.edit_alias.lock().unwrap() = v.into();
1512 self.bump();
1513 }
1514
1515 pub fn edit_error(&self) -> Option<String> {
1517 self.edit_error.lock().unwrap().clone()
1518 }
1519
1520 pub fn set_edit_error(&self, e: Option<String>) {
1522 *self.edit_error.lock().unwrap() = e;
1523 self.bump();
1524 }
1525
1526 pub fn account_edit_dirty(&self) -> bool {
1528 let editing = self.editing_account();
1529 let draft = self.ctp_form();
1530 match editing {
1531 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1532 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1533 }
1534 }
1535
1536 pub fn accounts(&self) -> Vec<AccountRow> {
1538 self.accounts.lock().unwrap().clone()
1539 }
1540
1541 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1543 self.accounts
1544 .lock()
1545 .unwrap()
1546 .iter()
1547 .find(|a| a.gateway == gateway)
1548 .cloned()
1549 }
1550
1551 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1557 let mut acc = self.accounts.lock().unwrap();
1558 if acc.iter().any(|a| a.gateway == row.gateway) {
1559 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1560 }
1561 let alias = row.alias.trim();
1562 if !alias.is_empty()
1563 && acc
1564 .iter()
1565 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1566 {
1567 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1568 }
1569 if acc.len() >= MAX_ACCOUNTS {
1570 return Err(format!(
1571 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1572 ));
1573 }
1574 acc.push(row);
1575 drop(acc);
1576 self.bump();
1577 Ok(())
1578 }
1579
1580 #[must_use]
1582 pub fn can_add_account(&self) -> bool {
1583 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1584 }
1585
1586 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1592 let alias = alias.trim().to_string();
1593 let mut acc = self.accounts.lock().unwrap();
1594 if !alias.is_empty()
1595 && acc
1596 .iter()
1597 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1598 {
1599 return false;
1600 }
1601 match acc.iter_mut().find(|a| a.gateway == gateway) {
1602 Some(a) => {
1603 a.alias = alias;
1604 drop(acc);
1605 self.bump();
1606 true
1607 }
1608 None => false,
1609 }
1610 }
1611
1612 pub fn note_connect_sent(&self, gateway: &str) {
1617 let mut acc = self.accounts.lock().unwrap();
1618 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1619 a.stage = AccountStage::Assembling;
1620 a.fail_note = None;
1621 a.login = LoginState::Unknown;
1622 a.status_note = None;
1623 a.login_brief = None;
1624 a.chain_note = None;
1626 drop(acc);
1627 self.bump();
1628 }
1629 }
1630
1631 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1635 let mut acc = self.accounts.lock().unwrap();
1636 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1637 match result {
1638 Ok(()) => {
1639 a.stage = AccountStage::Assembled;
1640 a.fail_note = None;
1641 }
1642 Err(e) => {
1643 a.stage = AccountStage::AssembleFailed;
1644 a.fail_note = Some(e);
1645 }
1646 }
1647 drop(acc);
1648 self.bump();
1649 }
1650 }
1651
1652 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1656 let mut acc = self.accounts.lock().unwrap();
1657 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1658 a.effective_classes = classes.to_vec();
1659 drop(acc);
1660 self.bump();
1661 }
1662 }
1663
1664 pub fn note_chain(&self, gateway: &str, note: &str) {
1669 let mut acc = self.accounts.lock().unwrap();
1670 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1671 if a.chain_note.as_deref() == Some(note) {
1672 return; }
1674 a.chain_note = Some(note.to_string());
1675 drop(acc);
1676 self.bump();
1677 }
1678 }
1679
1680 #[must_use]
1684 pub fn status_thresholds(&self) -> (u32, i64) {
1685 self.status_thresholds
1686 .lock()
1687 .unwrap()
1688 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1689 }
1690
1691 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1693 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1694 self.bump();
1695 }
1696
1697 pub fn list_poll_due(&self) -> bool {
1702 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1703 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1704 }
1705
1706 pub fn request_list_refresh(&self) {
1708 self.list_due.store(true, Ordering::Release);
1709 }
1710
1711 pub fn request_trade_reseed(&self) {
1713 self.seed_due.store(true, Ordering::Release);
1714 }
1715
1716 pub fn take_trade_reseed(&self) -> bool {
1718 self.seed_due.swap(false, Ordering::AcqRel)
1719 }
1720
1721 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1726 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1730 for gw in present {
1731 if !uniq.iter().any(|u| u == gw) {
1732 uniq.push(gw.clone());
1733 }
1734 }
1735 let n = uniq.len();
1736 if n == 0 {
1737 return Vec::new();
1738 }
1739 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1740 let take = STATUS_ASKS_PER_TICK.min(n);
1741 let mut out = Vec::with_capacity(take);
1742 for i in 0..take {
1743 out.push(uniq[(start + i) % n].clone());
1744 }
1745 out
1746 }
1747
1748 pub fn request_pipeline_refresh(&self) {
1750 self.pipeline_due.store(true, Ordering::Release);
1751 }
1752 pub fn take_pipeline_due(&self) -> bool {
1754 self.pipeline_due.swap(false, Ordering::AcqRel)
1755 }
1756
1757 pub fn request_cfg_refresh(&self) {
1759 self.cfg_due.store(true, Ordering::Release);
1760 }
1761 pub fn take_cfg_due(&self) -> bool {
1763 self.cfg_due.swap(false, Ordering::AcqRel)
1764 }
1765 pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
1767 *self.cfg_briefs.lock().unwrap() = Some(b);
1768 self.bump();
1769 }
1770 pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
1772 self.cfg_briefs.lock().unwrap().clone()
1773 }
1774
1775 pub fn cfg_delete_armed(&self) -> Option<String> {
1777 self.cfg_delete_arm.lock().unwrap().clone()
1778 }
1779 pub fn arm_cfg_delete(&self, ns: Option<String>) {
1780 *self.cfg_delete_arm.lock().unwrap() = ns;
1781 self.bump();
1782 }
1783
1784 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
1786 *self.pipeline_lines.lock().unwrap() = lines;
1787 self.bump();
1788 }
1789
1790 pub fn pipeline_lines(&self) -> Vec<String> {
1792 self.pipeline_lines.lock().unwrap().clone()
1793 }
1794
1795 pub fn note_bad_payload(&self, line: &str) {
1801 let mut book = self.bad_payload.lock().unwrap();
1802 let n = book.entry(line.to_string()).or_insert(0);
1803 *n += 1;
1804 if *n == 1 {
1805 self.log_line(format!(
1806 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
1807 ));
1808 }
1809 }
1810
1811 #[must_use]
1813 pub fn bad_payload_count(&self, line: &str) -> u64 {
1814 self.bad_payload
1815 .lock()
1816 .unwrap()
1817 .get(line)
1818 .copied()
1819 .unwrap_or(0)
1820 }
1821
1822 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
1831 for r in rows {
1832 if r.spilled == 0 {
1833 continue;
1834 }
1835 let cap = mailbox_cap_of(&r.name);
1836 if self.alert_once(format!("spill:{}", r.name)) {
1837 self.push_alert(
1838 crate::toast::AlertLevel::Warning,
1839 format!(
1840 "插件 {}:邮箱溢出 {} 条(刷新停顿时行情会堆;溢出丢的是新到的事件)",
1841 r.name, r.spilled
1842 ),
1843 );
1844 }
1845 if r.spilled >= cap as u64 && self.alert_once(format!("spill-full:{}", r.name)) {
1846 self.push_alert(
1847 crate::toast::AlertLevel::Error,
1848 format!(
1849 "插件 {}:溢出 {} 条 ≥ 该插件邮箱容量 {}(丢的是一段时期的事件,不是零星几条)",
1850 r.name, r.spilled, cap
1851 ),
1852 );
1853 }
1854 }
1855 }
1856
1857 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
1865 let (limit, stale) = self.status_thresholds();
1866 let now = ek_ctp::map::now_epoch_ms();
1867 let mut alarm: Option<(LoginState, String)> = None;
1870 let mut found = false;
1871 {
1872 let mut acc = self.accounts.lock().unwrap();
1873 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1874 found = true;
1875 a.apply_status(st, now, limit, stale);
1876 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
1877 alarm = Some((
1878 a.login,
1879 a.status_note
1880 .clone()
1881 .unwrap_or_else(|| a.login.label().to_string()),
1882 ));
1883 }
1884 }
1885 }
1886 if found {
1887 self.bump();
1888 }
1889 if let Some((login, note)) = alarm
1890 && self.alert_once(format!("{gateway}|{}", login.label()))
1891 {
1892 self.push_alert(
1893 crate::toast::AlertLevel::Error,
1894 format!("账户 {gateway}:{note}"),
1895 );
1896 }
1897 }
1898
1899 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
1903 let mut acc = self.accounts.lock().unwrap();
1904 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1905 a.login = LoginState::Unknown;
1906 a.status_note = Some(format!("状态口问不到:{why}"));
1907 a.login_brief = None;
1908 drop(acc);
1909 self.bump();
1910 }
1911 }
1912
1913 pub fn mark_account_stopped(&self, gateway: &str) {
1915 let mut acc = self.accounts.lock().unwrap();
1916 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1917 a.login = LoginState::Stopped;
1918 a.status_note = Some("会话已停;『连接』可恢复".to_string());
1919 a.login_brief = None;
1920 drop(acc);
1921 self.bump();
1922 }
1923 }
1924
1925 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
1928 let source = self.md_source();
1929 let mut f = form;
1930 if source.as_str() != gateway {
1931 f.md_addr.clear();
1932 f.md_user.clear();
1933 f.md_password.clear();
1934 }
1935 self.update_account_form(gateway, f.clone());
1936 f
1937 }
1938
1939 #[must_use]
1943 pub fn draft_from_existing(&self) -> CtpForm {
1944 let acc = self.accounts.lock().unwrap();
1945 let mut draft = CtpForm::default();
1946 if let Some(a) = acc.last() {
1947 draft.broker_id = a.form.broker_id.clone();
1948 draft.td_auth_code = a.form.td_auth_code.clone();
1949 draft.td_app_id = a.form.td_app_id.clone();
1950 draft.accept_futures = a.form.accept_futures;
1951 draft.accept_options = a.form.accept_options;
1952 }
1953 draft
1954 }
1955
1956 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
1958 let mut acc = self.accounts.lock().unwrap();
1959 match acc.iter_mut().find(|a| a.gateway == gateway) {
1960 Some(a) => {
1961 a.form = form;
1962 drop(acc);
1963 self.bump();
1964 true
1965 }
1966 None => false,
1967 }
1968 }
1969
1970 pub fn remove_account(&self, gateway: &str) -> bool {
1972 let mut acc = self.accounts.lock().unwrap();
1973 let before = acc.len();
1974 acc.retain(|a| a.gateway != gateway);
1975 let removed = acc.len() != before;
1976 if removed {
1977 let mut md = self.md_source.lock().unwrap();
1979 if md.as_str() == gateway {
1980 md.clear();
1981 }
1982 drop(acc);
1983 self.bump();
1984 }
1985 removed
1986 }
1987
1988 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
1997 let mut newly = Vec::new();
1998 {
1999 let mut acc = self.accounts.lock().unwrap();
2000 for a in acc.iter_mut() {
2001 let up = rows.iter().any(|r| r.name == a.gateway);
2002 if up && !a.connected {
2003 newly.push(a.gateway.clone());
2004 }
2005 a.connected = up;
2006 if up {
2007 a.stage = AccountStage::Assembled;
2008 a.fail_note = None;
2009 } else {
2010 a.chain_note = None;
2013 if a.stage == AccountStage::Assembled {
2014 a.stage = AccountStage::NotAssembled;
2015 a.login = LoginState::Unknown;
2016 a.status_note = None;
2017 a.login_brief = None;
2018 }
2019 }
2020 }
2021 }
2022 if !newly.is_empty() {
2023 self.bump();
2024 }
2025 newly
2026 }
2027
2028 pub fn md_source(&self) -> String {
2030 self.md_source.lock().unwrap().clone()
2031 }
2032
2033 pub fn accounts_locked(&self) -> bool {
2035 self.accounts.lock().unwrap().iter().any(|a| a.connected)
2036 }
2037
2038 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
2040 {
2041 let acc = self.accounts.lock().unwrap();
2042 if acc.iter().any(|a| a.connected) {
2043 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2044 }
2045 if !acc.iter().any(|a| a.gateway == gateway) {
2046 return Err(format!("账户 {gateway} 不在清册"));
2047 }
2048 }
2049 let mut acc = self.accounts.lock().unwrap();
2050 for a in acc.iter_mut() {
2051 a.md_source = a.gateway == gateway;
2052 }
2053 drop(acc);
2054 let mut md = self.md_source.lock().unwrap();
2055 *md = gateway.to_string();
2056 self.bump();
2057 Ok(())
2058 }
2059
2060 pub fn subscriptions(&self) -> Vec<String> {
2062 self.subscriptions.lock().unwrap().clone()
2063 }
2064
2065 pub fn note_subscriptions(&self, symbols: &[String]) {
2067 {
2068 let mut subs = self.subscriptions.lock().unwrap();
2069 for s in symbols {
2070 let s = s.trim();
2071 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2072 subs.push(s.to_string());
2073 }
2074 }
2075 }
2076 self.bump();
2077 }
2078
2079 pub fn default_gateway(&self) -> Option<String> {
2086 let chosen = {
2087 let f = self.order_form.lock().unwrap();
2088 f.gateway.trim().to_string()
2089 };
2090 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2091 return Some(chosen);
2092 }
2093 let acc = self.accounts.lock().unwrap();
2094 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2095 }
2096
2097 pub fn kline_symbols(&self) -> Vec<String> {
2099 self.kline_windows.lock().unwrap().clone()
2100 }
2101
2102 pub fn open_kline(&self, symbol: impl Into<String>) {
2104 let s = symbol.into();
2105 let mut v = self.kline_windows.lock().unwrap();
2106 if !v.contains(&s) {
2107 v.push(s.clone());
2108 }
2109 drop(v);
2110 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2111 }
2112
2113 pub fn close_kline(&self, symbol: &str) {
2115 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2116 self.bump();
2117 }
2118
2119 pub fn contracts_window_open(&self) -> bool {
2123 self.contracts_window.load(Ordering::Acquire)
2124 }
2125
2126 pub fn set_contracts_window_open(&self, v: bool) {
2128 self.contracts_window.store(v, Ordering::Release);
2129 self.bump();
2130 }
2131
2132 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2134 self.contract_store.lock().unwrap().put(c);
2135 self.bump();
2136 }
2137
2138 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2140 self.contract_store.lock().unwrap().query(pattern)
2141 }
2142
2143 pub fn contract_count(&self) -> usize {
2145 self.contract_store.lock().unwrap().len()
2146 }
2147
2148 pub fn contract_query(&self) -> String {
2150 self.contract_query.lock().unwrap().clone()
2151 }
2152
2153 pub fn set_contract_query(&self, q: impl Into<String>) {
2155 *self.contract_query.lock().unwrap() = q.into();
2156 self.bump();
2157 }
2158
2159 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2161 let mut v = self.contract_selected.lock().unwrap();
2162 match v.iter().position(|s| s == en_symbol) {
2163 Some(i) => {
2164 v.remove(i);
2165 }
2166 None => v.push(en_symbol.to_string()),
2167 }
2168 self.bump();
2169 }
2170
2171 pub fn selected_contracts(&self) -> Vec<String> {
2173 self.contract_selected.lock().unwrap().clone()
2174 }
2175
2176 pub fn log_line(&self, line: impl Into<String>) {
2178 self.logs.lock().unwrap().push(line);
2179 self.bump();
2180 }
2181
2182 pub fn clear_logs(&self) {
2184 self.logs.lock().unwrap().clear();
2185 self.bump();
2186 }
2187
2188 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2190 let r = self.logs.lock().unwrap();
2191 (r.snapshot(), r.spilled())
2192 }
2193
2194 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2196 let r = self.logs.lock().unwrap();
2197 (r.entries(), r.spilled())
2198 }
2199
2200 pub fn logs_len(&self) -> usize {
2202 self.logs.lock().unwrap().len()
2203 }
2204
2205 pub fn set_snapshot(&self, snap: BusSnap) {
2207 *self.snap.lock().unwrap() = Arc::new(snap);
2208 self.bump();
2209 }
2210
2211 pub fn snapshot(&self) -> Arc<BusSnap> {
2213 Arc::clone(&self.snap.lock().unwrap())
2214 }
2215
2216 pub fn rev(&self) -> u64 {
2218 self.rev.load(Ordering::Acquire)
2219 }
2220
2221 pub fn bump(&self) {
2223 self.rev.fetch_add(1, Ordering::AcqRel);
2224 }
2225
2226 pub fn request_stop(&self) {
2228 self.stop.store(true, Ordering::Release);
2229 }
2230
2231 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2234 self.pending_alerts
2235 .lock()
2236 .unwrap_or_else(std::sync::PoisonError::into_inner)
2237 .push((level, text.into()));
2238 }
2239
2240 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2242 let mut q = self
2243 .pending_alerts
2244 .lock()
2245 .unwrap_or_else(std::sync::PoisonError::into_inner);
2246 std::mem::take(&mut *q)
2247 }
2248
2249 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2251 self.toasts.lock()
2252 }
2253
2254 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2256 let key = key.into();
2257 self.alerted
2258 .lock()
2259 .unwrap_or_else(std::sync::PoisonError::into_inner)
2260 .insert(key)
2261 }
2262
2263 #[must_use]
2265 pub fn pending_alert_count(&self) -> usize {
2266 self.pending_alerts
2267 .lock()
2268 .unwrap_or_else(std::sync::PoisonError::into_inner)
2269 .len()
2270 }
2271
2272 #[must_use]
2274 pub fn alerted_count(&self) -> usize {
2275 self.alerted
2276 .lock()
2277 .unwrap_or_else(std::sync::PoisonError::into_inner)
2278 .len()
2279 }
2280
2281 pub fn exit_confirm_open(&self) -> bool {
2283 self.exit_confirm_open
2284 .load(std::sync::atomic::Ordering::Relaxed)
2285 }
2286
2287 pub fn open_exit_confirm(&self) {
2288 self.exit_confirm_open
2289 .store(true, std::sync::atomic::Ordering::Relaxed);
2290 }
2291
2292 pub fn close_exit_confirm(&self) {
2293 self.exit_confirm_open
2294 .store(false, std::sync::atomic::Ordering::Relaxed);
2295 }
2296
2297 pub fn exit_confirmed(&self) -> bool {
2298 self.exit_confirmed
2299 .load(std::sync::atomic::Ordering::Relaxed)
2300 }
2301
2302 pub fn confirm_exit(&self) {
2303 self.exit_confirmed
2304 .store(true, std::sync::atomic::Ordering::Relaxed);
2305 }
2306
2307 pub fn is_stop(&self) -> bool {
2308 self.stop.load(Ordering::Acquire)
2309 }
2310
2311 pub fn push_cmd(&self, cmd: UiCmd) {
2315 self.cmds.lock().unwrap().push_back(cmd);
2316 self.bump();
2317 }
2318
2319 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2321 self.cmds.lock().unwrap().drain(..).collect()
2322 }
2323
2324 #[must_use]
2326 pub fn cmd_pending(&self) -> bool {
2327 !self.cmds.lock().unwrap().is_empty()
2328 }
2329
2330 pub fn pending_confirm(&self) -> Option<String> {
2332 self.pending_confirm.lock().unwrap().clone()
2333 }
2334
2335 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2337 let mut g = self.pending_confirm.lock().unwrap();
2338 match g.take() {
2339 Some(p) if p == name => ConfirmStep::Fired,
2340 Some(_) => {
2341 *g = Some(name.to_string());
2342 ConfirmStep::Switched
2343 }
2344 None => {
2345 *g = Some(name.to_string());
2346 ConfirmStep::Armed
2347 }
2348 }
2349 }
2350
2351 pub fn confirm_cancel(&self) {
2353 *self.pending_confirm.lock().unwrap() = None;
2354 self.bump();
2355 }
2356
2357 pub fn market(&self) -> &MarketStore {
2361 &self.market
2362 }
2363
2364 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2373 {
2374 let mut book = self.trade_orders.lock().unwrap();
2375 book.clear();
2376 for o in ts.orders.into_iter().rev() {
2377 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2378 book.insert(o.en_orderid(), (seq, o));
2379 }
2380 }
2381 {
2382 let mut book = self.trade_trades.lock().unwrap();
2383 book.clear();
2384 let mut ids = self.trade_ids.lock().unwrap();
2385 ids.clear();
2386 for t in ts.trades.into_iter().rev() {
2387 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2388 ids.insert(t.en_tradeid());
2389 book.push_front((seq, t));
2390 }
2391 }
2392 {
2393 let mut book = self.trade_positions.lock().unwrap();
2394 book.clear();
2395 for p in ts.positions.into_iter().rev() {
2396 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2397 book.insert(p.en_positionid(), (seq, p));
2398 }
2399 }
2400 {
2401 let mut book = self.trade_funds.lock().unwrap();
2402 book.clear();
2403 for a in ts.accounts.into_iter().rev() {
2404 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2405 book.insert(a.en_accountid(), (seq, a));
2406 }
2407 }
2408 self.trade_dirty.store(true, Ordering::Release);
2409 self.bump();
2410 }
2411
2412 pub fn note_order(&self, o: &ek_data::order::OrderData) {
2420 let key = o.en_orderid();
2421 {
2422 let mut book = self.trade_orders.lock().unwrap();
2423 match book.get_mut(&key) {
2424 Some((_, cur)) => {
2425 cur.name.clone_from(&o.name);
2426 cur.traded = o.traded;
2427 cur.traded_price = o.traded_price;
2428 cur.status = o.status;
2429 cur.status_msg.clone_from(&o.status_msg);
2430 cur.datetime = o.datetime;
2431 }
2432 None => {
2433 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2434 book.insert(key, (seq, o.clone()));
2435 }
2436 }
2437 }
2438 self.trade_dirty.store(true, Ordering::Release);
2439 self.bump();
2440 }
2441
2442 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
2450 let id = t.en_tradeid();
2451 if self.trade_ids.lock().unwrap().contains(&id) {
2452 return;
2453 }
2454 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2455 let dropped = {
2456 let mut book = self.trade_trades.lock().unwrap();
2457 book.push_front((seq, t.clone()));
2458 self.trade_ids.lock().unwrap().insert(id);
2459 let over = book.len().saturating_sub(TRADE_CAP);
2460 for _ in 0..over {
2461 if let Some((_, old)) = book.pop_back() {
2463 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
2464 }
2465 }
2466 over
2467 };
2468 if dropped > 0 {
2469 let total = self
2470 .trade_dropped
2471 .fetch_add(dropped as u64, Ordering::AcqRel)
2472 + dropped as u64;
2473 if total == dropped as u64 {
2475 self.push_alert(
2476 crate::toast::AlertLevel::Warning,
2477 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
2478 );
2479 }
2480 }
2481 self.trade_dirty.store(true, Ordering::Release);
2482 self.bump();
2483 }
2484
2485 pub fn note_position(&self, p: &ek_data::position::PositionData) {
2490 let key = p.en_positionid();
2491 {
2492 let mut book = self.trade_positions.lock().unwrap();
2493 match book.get_mut(&key) {
2494 Some((_, cur)) => {
2495 cur.volume = p.volume;
2496 cur.yd_volume = p.yd_volume;
2497 cur.can_use_volume = p.can_use_volume;
2498 cur.frozen = p.frozen;
2499 cur.on_road_volume = p.on_road_volume;
2500 cur.market_value = p.market_value;
2501 cur.avg_price = p.avg_price;
2502 cur.price = p.price;
2503 cur.last_price = p.last_price;
2504 cur.pnl = p.pnl;
2505 cur.profit_rate = p.profit_rate;
2506 cur.name.clone_from(&p.name);
2507 }
2508 None => {
2509 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2510 book.insert(key, (seq, p.clone()));
2511 }
2512 }
2513 }
2514 self.trade_dirty.store(true, Ordering::Release);
2515 self.bump();
2516 }
2517
2518 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
2520 let key = a.en_accountid();
2521 {
2522 let mut book = self.trade_funds.lock().unwrap();
2523 match book.get_mut(&key) {
2524 Some((_, cur)) => {
2525 cur.balance = a.balance;
2526 cur.frozen = a.frozen;
2527 }
2528 None => {
2529 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2530 book.insert(key, (seq, a.clone()));
2531 }
2532 }
2533 }
2534 self.trade_dirty.store(true, Ordering::Release);
2535 self.bump();
2536 }
2537
2538 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
2540 let mut removed = 0usize;
2541 {
2542 let mut book = self.trade_orders.lock().unwrap();
2543 let before = book.len();
2544 book.retain(|_, (_, o)| o.gateway_name != gateway);
2545 removed += before - book.len();
2546 }
2547 {
2548 let mut book = self.trade_trades.lock().unwrap();
2549 let before = book.len();
2550 book.retain(|(_, t)| t.gateway_name != gateway);
2551 removed += before - book.len();
2552 let prefix = format!("{gateway}.");
2555 self.trade_ids
2556 .lock()
2557 .unwrap()
2558 .retain(|id| !id.starts_with(&prefix));
2559 }
2560 {
2561 let mut book = self.trade_positions.lock().unwrap();
2562 let before = book.len();
2563 book.retain(|_, (_, p)| p.gateway_name != gateway);
2564 removed += before - book.len();
2565 }
2566 {
2567 let mut book = self.trade_funds.lock().unwrap();
2568 let before = book.len();
2569 book.retain(|_, (_, a)| a.gateway_name != gateway);
2570 removed += before - book.len();
2571 }
2572 if removed > 0 {
2573 self.trade_dirty.store(true, Ordering::Release);
2574 self.bump();
2575 }
2576 removed
2577 }
2578
2579 #[must_use]
2581 pub fn trade_dropped(&self) -> u64 {
2582 self.trade_dropped.load(Ordering::Acquire)
2583 }
2584
2585 pub fn trade_snap(&self) -> Arc<TradeSnap> {
2590 let mut cache = self.trade_snap.lock().unwrap();
2591 if self.trade_dirty.swap(false, Ordering::AcqRel) {
2592 *cache = Arc::new(self.build_trade_snap());
2593 }
2594 Arc::clone(&cache)
2595 }
2596
2597 fn build_trade_snap(&self) -> TradeSnap {
2600 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
2601 where
2602 T: Clone,
2603 {
2604 let mut v: Vec<(u64, &String, &T)> =
2605 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
2606 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
2607 v.into_iter().map(|(_, _, v)| v.clone()).collect()
2608 }
2609 TradeSnap {
2610 orders: by_seq(&self.trade_orders.lock().unwrap()),
2611 trades: self
2612 .trade_trades
2613 .lock()
2614 .unwrap()
2615 .iter()
2616 .map(|(_, t)| t.clone())
2617 .collect(),
2618 positions: by_seq(&self.trade_positions.lock().unwrap()),
2619 accounts: by_seq(&self.trade_funds.lock().unwrap()),
2620 }
2621 }
2622
2623 pub fn ctp_form(&self) -> CtpForm {
2627 self.ctp_form.lock().unwrap().clone()
2628 }
2629
2630 pub fn set_ctp_form(&self, f: CtpForm) {
2632 *self.ctp_form.lock().unwrap() = f;
2633 self.bump();
2634 }
2635
2636 pub fn row_selected(&self, table: &str, key: &str) -> bool {
2640 self.row_sel
2641 .lock()
2642 .unwrap()
2643 .get(table)
2644 .is_some_and(|k| k == key)
2645 }
2646
2647 pub fn select_row(&self, table: &str, key: &str) {
2649 self.row_sel
2650 .lock()
2651 .unwrap()
2652 .insert(table.to_string(), key.to_string());
2653 }
2654}