1use std::collections::{BTreeMap, VecDeque};
7use std::sync::atomic::{AtomicBool, AtomicU8, 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
17#[derive(Debug)]
19pub struct LogRing {
20 lines: VecDeque<String>,
21 cap: usize,
22 spilled: u64,
23}
24
25impl Default for LogRing {
26 fn default() -> Self {
27 Self::new(RING_CAP)
28 }
29}
30
31impl LogRing {
32 pub fn new(cap: usize) -> Self {
34 Self {
35 lines: VecDeque::with_capacity(cap),
36 cap,
37 spilled: 0,
38 }
39 }
40
41 pub fn push(&mut self, line: impl Into<String>) {
43 if self.cap == 0 {
44 self.spilled += 1;
45 return;
46 }
47 if self.lines.len() == self.cap {
48 self.lines.pop_front();
49 self.spilled += 1;
50 }
51 self.lines.push_back(line.into());
52 }
53
54 pub fn len(&self) -> usize {
56 self.lines.len()
57 }
58
59 pub fn is_empty(&self) -> bool {
61 self.lines.is_empty()
62 }
63
64 pub fn spilled(&self) -> u64 {
66 self.spilled
67 }
68
69 pub fn snapshot(&self) -> Vec<String> {
71 self.lines.iter().cloned().collect()
72 }
73
74 pub fn clear(&mut self) {
76 self.lines.clear();
77 }
78}
79
80#[derive(Debug, Clone, Default)]
82pub struct BusSnap {
83 pub rows: Vec<PlugRow>,
85 pub services: Vec<(String, Vec<String>)>,
87}
88
89#[derive(Debug, Clone, PartialEq)]
91pub enum UiCmd {
92 Start { factory: String, name: String },
95 Remove { name: String },
97 Reload {
99 interval: i64,
100 symbol: String,
101 bars: Vec<BarData>,
102 },
103 SendOrder {
109 req: ek_data::order_request::OrderRequest,
110 gateway: String,
111 mode: ek_oms::LockMode,
112 },
113 CancelOrder { orderid: String, gateway: String },
117 CtpConfigure { gateway: String, form: CtpForm },
121 Connect { gateway: String, form: CtpForm },
126 CtpSubscribe {
129 gateway: String,
130 symbols: Vec<String>,
131 },
132 Disconnect { gateway: String },
136}
137
138pub const KLINE_CELL_CAP: usize = 4096;
140
141pub const MAX_ACCOUNTS: usize = 8;
144
145pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
150
151pub const ACCOUNT_STALE_SECS: i64 = 30;
156
157pub const STATUS_POLL_TICKS: u32 = 2;
162
163#[derive(Debug, Clone, Copy, PartialEq, Eq)]
165pub enum ConfirmStep {
166 Armed,
167 Fired,
168 Switched,
169}
170
171#[must_use]
174pub fn is_quarantined(state: &str) -> bool {
175 state.contains("Quarantin")
176}
177
178#[derive(Debug, Default)]
182pub struct MarketStore {
183 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
184 frames: AtomicU64,
186 updates: AtomicU64,
188 reloads: AtomicU64,
190}
191
192#[derive(Debug, Clone, PartialEq)]
194pub struct CellView {
195 pub symbol: String,
196 pub interval: i64,
197 pub bars: Vec<BarData>,
198}
199
200impl MarketStore {
201 pub fn on_frame(&self, frame: &BarsFrame) {
203 let mut cells = self.cells.lock().unwrap();
204 for b in &frame.bars {
205 let key = (b.symbol.clone(), frame.interval);
206 let cell = cells.entry(key).or_default();
207 match cell.last() {
208 Some(last) if b.datetime < last.datetime => continue,
209 Some(last) if b.datetime == last.datetime => {
210 let slot = cell.last_mut().expect("len>1");
211 *slot = b.clone();
212 }
213 _ => {
214 cell.push(b.clone());
215 if cell.len() > KLINE_CELL_CAP {
216 cell.remove(0);
217 }
218 }
219 }
220 }
221 self.frames.fetch_add(1, Ordering::AcqRel);
222 }
223
224 pub fn on_update(&self, b: &BarData) {
226 let mut cells = self.cells.lock().unwrap();
227 let iv = b
228 .interval
229 .unwrap_or(ek_data::constant::Interval::MINUTE)
230 .seconds();
231 let key = (b.symbol.clone(), iv);
232 let cell = cells.entry(key).or_default();
233 match cell.last() {
234 Some(last) if b.datetime == last.datetime => {
235 let slot = cell.last_mut().expect("len>1");
236 *slot = b.clone();
237 }
238 Some(last) if b.datetime < last.datetime => {}
239 _ => {
240 cell.push(b.clone());
241 if cell.len() > KLINE_CELL_CAP {
242 cell.remove(0);
243 }
244 }
245 }
246 self.updates.fetch_add(1, Ordering::AcqRel);
247 }
248
249 pub fn note_reload(&self) {
251 self.reloads.fetch_add(1, Ordering::AcqRel);
252 }
253
254 pub fn cells(&self) -> Vec<CellView> {
256 self.cells
257 .lock()
258 .unwrap()
259 .iter()
260 .map(|((sym, iv), bars)| CellView {
261 symbol: sym.clone(),
262 interval: *iv,
263 bars: bars.clone(),
264 })
265 .collect()
266 }
267
268 pub fn counts(&self) -> (u64, u64, u64) {
270 (
271 self.frames.load(Ordering::Acquire),
272 self.updates.load(Ordering::Acquire),
273 self.reloads.load(Ordering::Acquire),
274 )
275 }
276}
277
278#[derive(Debug, Default, Clone, PartialEq)]
281pub struct ReloadForm {
282 pub interval: String,
283 pub symbol: String,
284 pub csv: String,
285}
286
287pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
294 let g = raw.trim();
295 if !g.is_empty() {
296 if g.chars().any(|c| c.is_whitespace()) {
297 return Err(format!("网关解析: 含空白:{raw:?}"));
298 }
299 return Ok(g.to_string());
300 }
301 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
302 Some(f) => Ok(f.to_string()),
303 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
304 }
305}
306
307#[derive(Debug, Clone, PartialEq)]
310pub struct CtpForm {
311 pub broker_id: String,
312 pub td_addr: String,
313 pub td_user: String,
314 pub td_password: String,
315 pub td_auth_code: String,
316 pub td_app_id: String,
317 pub md_addr: String,
318 pub md_user: String,
319 pub md_password: String,
320 pub symbols: String,
322 pub accept_futures: bool,
325 pub accept_options: bool,
326}
327
328impl Default for CtpForm {
331 fn default() -> Self {
332 Self {
333 broker_id: String::new(),
334 td_addr: String::new(),
335 td_user: String::new(),
336 td_password: String::new(),
337 td_auth_code: String::new(),
338 td_app_id: String::new(),
339 md_addr: String::new(),
340 md_user: String::new(),
341 md_password: String::new(),
342 symbols: String::new(),
343 accept_futures: true,
344 accept_options: false,
345 }
346 }
347}
348
349impl CtpForm {
350 #[must_use]
354 pub fn product_classes(&self) -> Vec<String> {
355 let mut codes = Vec::new();
356 if self.accept_futures {
357 codes.push("1".to_string());
358 }
359 if self.accept_options {
360 codes.push("2".to_string());
361 }
362 codes
363 }
364
365 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
367 ek_ctp::plugin::ConfigureReq {
368 broker_id: self.broker_id.trim().to_string(),
369 td_addr: self.td_addr.trim().to_string(),
370 td_user: self.td_user.trim().to_string(),
371 td_password: self.td_password.clone(),
372 td_auth_code: self.td_auth_code.trim().to_string(),
373 td_app_id: self.td_app_id.trim().to_string(),
374 md_addr: self.md_addr.trim().to_string(),
375 md_user: self.md_user.trim().to_string(),
376 md_password: self.md_password.clone(),
377 product_classes: self.product_classes(),
378 }
379 }
380
381 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
383 let list: Vec<String> = self
384 .symbols
385 .split([',', ',', ' '])
386 .map(|s| s.trim().to_string())
387 .filter(|s| !s.is_empty())
388 .collect();
389 if list.is_empty() {
390 return Err("订阅解析: 标的位空".to_string());
391 }
392 Ok(list)
393 }
394
395 pub fn validate(&self) -> Result<(), String> {
404 fn front(name: &str, v: &str) -> Result<(), String> {
405 let v = v.trim();
406 let Some(rest) = v.strip_prefix("tcp://") else {
407 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
408 };
409 let Some((host, port)) = rest.rsplit_once(':') else {
410 return Err(format!("{name} 缺端口:{name} {:?}", v));
411 };
412 if host.trim().is_empty() {
413 return Err(format!("{name} 主机位空:{}", v));
414 }
415 let port: u32 = port
416 .parse()
417 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
418 if port == 0 || port > 65535 {
419 return Err(format!("{name} 端口越界(1..=65535):{}", v));
420 }
421 Ok(())
422 }
423 if self.broker_id.trim().is_empty() {
424 return Err("broker 位空".to_string());
425 }
426 if self.td_user.trim().is_empty() {
427 return Err("交易账号位空".to_string());
428 }
429 front("td 前置", &self.td_addr)?;
430 if !self.md_addr.trim().is_empty() {
431 front("md 前置", &self.md_addr)?;
432 }
433 if self.td_app_id.trim().is_empty() {
434 return Err("AppID 位空(CTP 登录必需)".to_string());
435 }
436 if self.td_auth_code.trim().is_empty() {
437 return Err("AuthCode 位空(CTP 登录必需)".to_string());
438 }
439 Ok(())
440 }
441}
442
443#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
450pub enum AccountStage {
451 #[default]
453 NotAssembled,
454 Assembling,
456 Assembled,
458 AssembleFailed,
460}
461
462impl AccountStage {
463 #[must_use]
465 pub fn label(self) -> &'static str {
466 match self {
467 AccountStage::NotAssembled => "未装配",
468 AccountStage::Assembling => "装配中…",
469 AccountStage::Assembled => "已装配·登录态未知",
472 AccountStage::AssembleFailed => "装配失败",
473 }
474 }
475}
476
477#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
482pub enum LoginState {
483 #[default]
488 Unknown,
489 Online,
491 LoginFailed,
493 FrontDown,
495 QueryStarved,
497 Stopped,
499}
500
501impl LoginState {
502 #[must_use]
504 pub fn label(self) -> &'static str {
505 match self {
506 LoginState::Unknown => "登录态未知",
507 LoginState::Online => "在线",
508 LoginState::LoginFailed => "登录失败",
509 LoginState::FrontDown => "掉线",
510 LoginState::QueryStarved => "查询断供",
511 LoginState::Stopped => "已断开",
512 }
513 }
514
515 #[must_use]
517 pub fn is_alarm(self) -> bool {
518 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
519 }
520}
521
522fn count_note(label: &str, n: u32) -> Option<String> {
524 (n > 0).then(|| format!("{label} {n} 次"))
525}
526
527#[derive(Debug, Clone, PartialEq, Default)]
533pub struct AccountRow {
534 pub gateway: String,
536 pub alias: String,
538 pub broker_id: String,
539 pub account: String,
540 pub md_source: bool,
542 pub form: CtpForm,
544 pub stage: AccountStage,
546 pub fail_note: Option<String>,
548 pub effective_classes: Vec<String>,
553 pub connected: bool,
556 pub login: LoginState,
560 pub status_note: Option<String>,
563}
564
565impl AccountRow {
566 pub fn derive(form: &CtpForm) -> Result<Self, String> {
571 let broker = form.broker_id.trim();
572 let account = form.td_user.trim();
573 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
574 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
575 })?;
576 Ok(Self {
577 gateway,
578 alias: String::new(),
579 broker_id: broker.to_string(),
580 account: account.to_string(),
581 md_source: false,
582 form: form.clone(),
583 stage: AccountStage::NotAssembled,
584 fail_note: None,
585 effective_classes: Vec::new(),
586 connected: false,
587 login: LoginState::Unknown,
588 status_note: None,
589 })
590 }
591
592 #[must_use]
598 pub fn status_label(&self) -> String {
599 if self.stage == AccountStage::Assembled {
600 match self.login {
601 LoginState::Unknown => "已装配·登录态未知".to_string(),
602 other => format!("已装配·{}", other.label()),
603 }
604 } else {
605 self.stage.label().to_string()
606 }
607 }
608
609 pub fn apply_status(
622 &mut self,
623 st: &ek_ctp::plugin::GatewayStatus,
624 now_ms: i64,
625 reject_limit: u32,
626 stale_secs: i64,
627 ) {
628 let td_side = !self.form.td_addr.trim().is_empty();
629 let (login, note) = if td_side {
630 if st.td_login_failed && !st.td_logged_in {
631 (
632 LoginState::LoginFailed,
633 st.td_last_error
634 .clone()
635 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
636 )
637 } else if st.td_logged_in {
638 if st.td_query_reject_streak >= reject_limit {
639 (
640 LoginState::QueryStarved,
641 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
642 )
643 } else if let Some(last) = st.td_last_account_ms {
644 let idle_ms = now_ms.saturating_sub(last);
645 if idle_ms > stale_secs.saturating_mul(1_000) {
646 (
647 LoginState::QueryStarved,
648 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
649 )
650 } else {
651 (LoginState::Online, count_note("掉线", st.td_disconnects))
652 }
653 } else {
654 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
661 match since_login_ms {
662 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
663 LoginState::QueryStarved,
664 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
665 ),
666 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
667 }
668 }
669 } else if self.login == LoginState::Stopped {
670 let extra = if st.td_disconnects > 0 {
671 format!("(此前掉线 {} 次)", st.td_disconnects)
672 } else {
673 String::new()
674 };
675 (
676 LoginState::Stopped,
677 Some(format!("会话已停{extra};『连接』可恢复")),
678 )
679 } else if st.td_disconnects > 0 {
680 (
681 LoginState::FrontDown,
682 Some(format!("掉线 {} 次", st.td_disconnects)),
683 )
684 } else {
685 (LoginState::Unknown, st.td_last_error.clone())
686 }
687 } else if st.md_logged_in {
688 (
689 LoginState::Online,
690 count_note("掉线", st.md_disconnects)
691 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
692 )
693 } else if self.login == LoginState::Stopped {
694 (
695 LoginState::Stopped,
696 Some("会话已停;『连接』可恢复".to_string()),
697 )
698 } else if st.md_disconnects > 0 {
699 (
700 LoginState::FrontDown,
701 Some(format!("掉线 {} 次", st.md_disconnects)),
702 )
703 } else {
704 (LoginState::Unknown, st.md_last_error.clone())
705 };
706 let note = if td_side && st.td_instrument_rejects > 0 {
711 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
712 match note {
713 Some(base) if base.contains("合约查询被拒") => Some(base),
715 Some(base) => Some(format!("{base};{add}")),
716 None => Some(add),
717 }
718 } else {
719 note
720 };
721 self.login = login;
722 self.status_note = note;
723 }
724
725 #[must_use]
727 pub fn display_name(&self) -> String {
728 let a = self.alias.trim();
729 if a.is_empty() {
730 self.account.clone()
731 } else {
732 a.to_string()
733 }
734 }
735}
736
737#[derive(Debug, Clone, PartialEq, Default)]
739pub struct TradeSnap {
740 pub orders: Vec<ek_data::order::OrderData>,
741 pub trades: Vec<ek_data::trade::TradeData>,
742 pub positions: Vec<ek_data::position::PositionData>,
743 pub accounts: Vec<ek_data::account::AccountData>,
745}
746
747pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
750 let interval: i64 = form
751 .interval
752 .trim()
753 .parse()
754 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
755 let symbol = form.symbol.trim().to_string();
756 if symbol.is_empty() {
757 return Err("回灌解析: symbol 空".to_string());
758 }
759 if form.csv.trim().is_empty() {
760 return Err("回灌解析: csv 空".to_string());
761 }
762 let mut bars = Vec::new();
763 for (i, raw) in form.csv.lines().enumerate() {
764 let line = raw.trim();
765 if line.is_empty() || line.starts_with('#') {
766 continue;
767 }
768 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
769 if parts.len() != 6 {
770 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
771 }
772 let num = |s: &str| -> Result<f64, String> {
773 s.parse::<f64>()
774 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
775 };
776 let dt_s: i64 = parts[0]
777 .parse()
778 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
779 bars.push(BarData {
780 symbol: symbol.clone(),
781 datetime: dt_s * 1_000_000_000,
782 interval: Some(
783 ek_dm::wire::interval_of_seconds(interval)
784 .unwrap_or(ek_data::constant::Interval::MINUTE),
785 ),
786 open_price: num(parts[1])?,
787 high_price: num(parts[2])?,
788 low_price: num(parts[3])?,
789 close_price: num(parts[4])?,
790 volume: num(parts[5])?,
791 ..BarData::default()
792 });
793 }
794 if bars.is_empty() {
795 return Err("回灌解析: 无有效行".to_string());
796 }
797 Ok((interval, symbol, bars))
798}
799
800#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
805pub enum TopLeftTab {
806 #[default]
807 Orders,
808 Active,
809}
810
811impl TopLeftTab {
812 pub fn label(self) -> &'static str {
813 match self {
814 Self::Orders => "委托",
815 Self::Active => "未成交",
816 }
817 }
818}
819
820#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
822pub enum BottomRightTab {
823 #[default]
824 Account,
825 Logs,
826}
827
828impl BottomRightTab {
829 pub fn label(self) -> &'static str {
830 match self {
831 Self::Account => "资金",
832 Self::Logs => "日志",
833 }
834 }
835}
836
837#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
841pub enum Theme {
842 #[default]
844 FinanceDark,
845 EguiDark,
847 EguiLight,
849 QtClassic,
851}
852
853impl Theme {
854 pub fn label(self) -> &'static str {
855 match self {
856 Self::FinanceDark => "金融深色",
857 Self::EguiDark => "egui深色",
858 Self::EguiLight => "egui浅色",
859 Self::QtClassic => "Qt灰白",
860 }
861 }
862
863 pub fn next(self) -> Self {
865 match self {
866 Self::FinanceDark => Self::EguiDark,
867 Self::EguiDark => Self::EguiLight,
868 Self::EguiLight => Self::QtClassic,
869 Self::QtClassic => Self::FinanceDark,
870 }
871 }
872}
873
874#[derive(Debug, Default)]
877pub struct TickStore {
878 ticks: BTreeMap<String, ek_data::tick::TickData>,
879}
880
881impl TickStore {
882 pub fn put(&mut self, t: &ek_data::tick::TickData) {
885 self.ticks
886 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
887 }
888
889 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
891 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
892 v.sort_by(|a, b| {
893 (a.exchange.as_str().to_string(), a.symbol.clone())
894 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
895 });
896 v
897 }
898
899 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
901 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
902 for t in self.rows() {
903 let ex = t.exchange.as_str().to_string();
904 match out.last_mut() {
905 Some((e, v)) if *e == ex => v.push(t),
906 _ => out.push((ex, vec![t])),
907 }
908 }
909 out
910 }
911
912 pub fn len(&self) -> usize {
913 self.ticks.len()
914 }
915
916 pub fn is_empty(&self) -> bool {
917 self.ticks.is_empty()
918 }
919}
920
921#[derive(Debug, Clone, PartialEq)]
923pub struct OrderForm {
924 pub exchange: String,
926 pub symbol: String,
928 pub direction: ek_data::constant::Direction,
929 pub offset: ek_data::constant::Offset,
930 pub order_type: ek_data::constant::OrderType,
931 pub price: String,
933 pub volume: String,
934 pub gateway: String,
936 pub follow_price: bool,
938 pub auto_split: bool,
941}
942
943impl Default for OrderForm {
944 fn default() -> Self {
945 Self {
946 exchange: "SHFE".to_string(),
947 symbol: String::new(),
948 direction: ek_data::constant::Direction::LONG,
949 offset: ek_data::constant::Offset::OPEN,
950 order_type: ek_data::constant::OrderType::LIMIT,
951 price: String::new(),
952 volume: "1".to_string(),
953 gateway: String::new(),
954 follow_price: true,
955 auto_split: true,
956 }
957 }
958}
959
960impl OrderForm {
961 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
963 use std::str::FromStr;
964 let symbol = self.symbol.trim();
965 if symbol.is_empty() {
966 return Err("下单: 代码为空".to_string());
967 }
968 if symbol.contains('.') {
969 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
970 }
971 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
972 .map_err(|e| format!("下单: {e}"))?;
973 let price: f64 = self
974 .price
975 .trim()
976 .parse()
977 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
978 let volume: f64 = self
979 .volume
980 .trim()
981 .parse()
982 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
983 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
984 return Err("下单: 限价单价格须为正".to_string());
985 }
986 if volume <= 0.0 {
987 return Err("下单: 数量须为正".to_string());
988 }
989 Ok(ek_data::order_request::OrderRequest {
990 symbol: symbol.to_string(),
991 exchange,
992 direction: self.direction,
993 order_type: self.order_type,
994 volume,
995 price,
996 offset: self.offset,
997 ..ek_data::order_request::OrderRequest::default()
998 })
999 }
1000}
1001
1002#[derive(Debug, Default)]
1008pub struct ContractStore {
1009 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1010}
1011
1012impl ContractStore {
1013 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1014 self.contracts
1015 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1016 }
1017
1018 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1020 let p = pattern.trim();
1021 self.contracts
1022 .values()
1023 .filter(|c| {
1024 p.is_empty()
1025 || c.symbol.contains(p)
1026 || c.exchange.as_str().eq_ignore_ascii_case(p)
1027 || c.name.contains(p)
1028 })
1029 .cloned()
1030 .collect()
1031 }
1032
1033 pub fn len(&self) -> usize {
1034 self.contracts.len()
1035 }
1036
1037 pub fn is_empty(&self) -> bool {
1038 self.contracts.is_empty()
1039 }
1040}
1041
1042#[derive(Debug, Default)]
1044pub struct UiShared {
1045 logs: Mutex<LogRing>,
1046 snap: Mutex<Arc<BusSnap>>,
1047 market: MarketStore,
1048 ctp_form: Mutex<CtpForm>,
1049 trade_snap: Mutex<Arc<TradeSnap>>,
1050 tick_store: Mutex<TickStore>,
1051 order_form: Mutex<OrderForm>,
1052 selected_exchange: Mutex<String>,
1053 topleft_tab: Mutex<TopLeftTab>,
1054 bottomright_tab: Mutex<BottomRightTab>,
1055 ops_window: AtomicBool,
1056 accounts_window: AtomicBool,
1059 contracts_window: AtomicBool,
1060 theme: AtomicU8,
1062 contract_store: Mutex<ContractStore>,
1063 contract_query: Mutex<String>,
1064 contract_selected: Mutex<Vec<String>>,
1065 kline_windows: Mutex<Vec<String>>,
1066 rev: AtomicU64,
1067 stop: AtomicBool,
1069 cmds: Mutex<VecDeque<UiCmd>>,
1071 pending_confirm: Mutex<Option<String>>,
1073 accounts: Mutex<Vec<AccountRow>>,
1075 account_edit_window: AtomicBool,
1077 editing_account: Mutex<Option<String>>,
1079 edit_alias: Mutex<String>,
1081 edit_error: Mutex<Option<String>>,
1083 md_source: Mutex<String>,
1085 subscriptions: Mutex<Vec<String>>,
1087 row_sel: Mutex<BTreeMap<String, String>>,
1090 status_tick: AtomicU32,
1092 status_thresholds: Mutex<Option<(u32, i64)>>,
1095}
1096
1097impl UiShared {
1098 pub fn new() -> Self {
1100 Self::default()
1101 }
1102
1103 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1107 self.tick_store.lock().unwrap().put(t);
1108 self.bump();
1109 }
1110
1111 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1113 self.tick_store.lock().unwrap().rows()
1114 }
1115
1116 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1118 self.tick_store.lock().unwrap().groups()
1119 }
1120
1121 pub fn tick_count(&self) -> usize {
1123 self.tick_store.lock().unwrap().len()
1124 }
1125
1126 pub fn order_form(&self) -> OrderForm {
1128 self.order_form.lock().unwrap().clone()
1129 }
1130
1131 pub fn set_order_form(&self, f: OrderForm) {
1133 *self.order_form.lock().unwrap() = f;
1134 self.bump();
1135 }
1136
1137 pub fn selected_exchange(&self) -> String {
1139 self.selected_exchange.lock().unwrap().clone()
1140 }
1141
1142 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1144 *self.selected_exchange.lock().unwrap() = s.into();
1145 self.bump();
1146 }
1147
1148 pub fn topleft_tab(&self) -> TopLeftTab {
1150 *self.topleft_tab.lock().unwrap()
1151 }
1152
1153 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1155 *self.topleft_tab.lock().unwrap() = t;
1156 self.bump();
1157 }
1158
1159 pub fn bottomright_tab(&self) -> BottomRightTab {
1161 *self.bottomright_tab.lock().unwrap()
1162 }
1163
1164 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1166 *self.bottomright_tab.lock().unwrap() = t;
1167 self.bump();
1168 }
1169
1170 pub fn theme(&self) -> Theme {
1174 match self.theme.load(Ordering::Acquire) {
1175 1 => Theme::EguiDark,
1176 2 => Theme::EguiLight,
1177 3 => Theme::QtClassic,
1178 _ => Theme::FinanceDark,
1179 }
1180 }
1181
1182 pub fn set_theme(&self, t: Theme) {
1184 self.theme.store(t as u8, Ordering::Release);
1185 self.bump();
1186 }
1187
1188 pub fn cycle_theme(&self) -> Theme {
1190 let t = self.theme().next();
1191 self.set_theme(t);
1192 t
1193 }
1194
1195 pub fn ops_window_open(&self) -> bool {
1197 self.ops_window.load(Ordering::Acquire)
1198 }
1199
1200 pub fn set_ops_window_open(&self, v: bool) {
1202 self.ops_window.store(v, Ordering::Release);
1203 self.bump();
1204 }
1205
1206 pub fn accounts_window_open(&self) -> bool {
1210 self.accounts_window.load(Ordering::Acquire)
1211 }
1212
1213 pub fn set_accounts_window_open(&self, v: bool) {
1215 self.accounts_window.store(v, Ordering::Release);
1216 self.bump();
1217 }
1218
1219 pub fn account_edit_open(&self) -> bool {
1221 self.account_edit_window.load(Ordering::Acquire)
1222 }
1223
1224 pub fn editing_account(&self) -> Option<String> {
1226 self.editing_account.lock().unwrap().clone()
1227 }
1228
1229 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1233 match gateway {
1234 Some(gw) => {
1235 let Some(row) = self.account(gw) else {
1236 return false;
1237 };
1238 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1239 *self.edit_alias.lock().unwrap() = row.alias.clone();
1240 self.set_ctp_form(row.form.clone());
1241 }
1242 None => {
1243 *self.editing_account.lock().unwrap() = None;
1244 *self.edit_alias.lock().unwrap() = String::new();
1245 let draft = self.draft_from_existing();
1246 self.set_ctp_form(draft);
1247 }
1248 }
1249 *self.edit_error.lock().unwrap() = None;
1250 self.account_edit_window.store(true, Ordering::Release);
1251 self.bump();
1252 true
1253 }
1254
1255 pub fn end_account_edit(&self) {
1257 *self.editing_account.lock().unwrap() = None;
1258 *self.edit_alias.lock().unwrap() = String::new();
1259 *self.edit_error.lock().unwrap() = None;
1260 self.account_edit_window.store(false, Ordering::Release);
1261 let draft = self.draft_from_existing();
1262 self.set_ctp_form(draft);
1263 self.bump();
1264 }
1265
1266 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1272 if self.editing_account().as_deref() == Some(gateway) {
1273 self.end_account_edit();
1274 }
1275 }
1276
1277 pub fn edit_alias(&self) -> String {
1279 self.edit_alias.lock().unwrap().clone()
1280 }
1281
1282 pub fn set_edit_alias(&self, v: impl Into<String>) {
1284 *self.edit_alias.lock().unwrap() = v.into();
1285 self.bump();
1286 }
1287
1288 pub fn edit_error(&self) -> Option<String> {
1290 self.edit_error.lock().unwrap().clone()
1291 }
1292
1293 pub fn set_edit_error(&self, e: Option<String>) {
1295 *self.edit_error.lock().unwrap() = e;
1296 self.bump();
1297 }
1298
1299 pub fn account_edit_dirty(&self) -> bool {
1301 let editing = self.editing_account();
1302 let draft = self.ctp_form();
1303 match editing {
1304 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1305 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1306 }
1307 }
1308
1309 pub fn accounts(&self) -> Vec<AccountRow> {
1311 self.accounts.lock().unwrap().clone()
1312 }
1313
1314 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1316 self.accounts
1317 .lock()
1318 .unwrap()
1319 .iter()
1320 .find(|a| a.gateway == gateway)
1321 .cloned()
1322 }
1323
1324 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1330 let mut acc = self.accounts.lock().unwrap();
1331 if acc.iter().any(|a| a.gateway == row.gateway) {
1332 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1333 }
1334 let alias = row.alias.trim();
1335 if !alias.is_empty()
1336 && acc
1337 .iter()
1338 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1339 {
1340 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1341 }
1342 if acc.len() >= MAX_ACCOUNTS {
1343 return Err(format!(
1344 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1345 ));
1346 }
1347 acc.push(row);
1348 drop(acc);
1349 self.bump();
1350 Ok(())
1351 }
1352
1353 #[must_use]
1355 pub fn can_add_account(&self) -> bool {
1356 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1357 }
1358
1359 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1365 let alias = alias.trim().to_string();
1366 let mut acc = self.accounts.lock().unwrap();
1367 if !alias.is_empty()
1368 && acc
1369 .iter()
1370 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1371 {
1372 return false;
1373 }
1374 match acc.iter_mut().find(|a| a.gateway == gateway) {
1375 Some(a) => {
1376 a.alias = alias;
1377 drop(acc);
1378 self.bump();
1379 true
1380 }
1381 None => false,
1382 }
1383 }
1384
1385 pub fn note_connect_sent(&self, gateway: &str) {
1390 let mut acc = self.accounts.lock().unwrap();
1391 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1392 a.stage = AccountStage::Assembling;
1393 a.fail_note = None;
1394 a.login = LoginState::Unknown;
1395 a.status_note = None;
1396 drop(acc);
1397 self.bump();
1398 }
1399 }
1400
1401 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1405 let mut acc = self.accounts.lock().unwrap();
1406 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1407 match result {
1408 Ok(()) => {
1409 a.stage = AccountStage::Assembled;
1410 a.fail_note = None;
1411 }
1412 Err(e) => {
1413 a.stage = AccountStage::AssembleFailed;
1414 a.fail_note = Some(e);
1415 }
1416 }
1417 drop(acc);
1418 self.bump();
1419 }
1420 }
1421
1422 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1426 let mut acc = self.accounts.lock().unwrap();
1427 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1428 a.effective_classes = classes.to_vec();
1429 drop(acc);
1430 self.bump();
1431 }
1432 }
1433
1434 #[must_use]
1438 pub fn status_thresholds(&self) -> (u32, i64) {
1439 self.status_thresholds
1440 .lock()
1441 .unwrap()
1442 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1443 }
1444
1445 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1447 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1448 self.bump();
1449 }
1450
1451 pub fn status_poll_due(&self) -> bool {
1453 let n = self.status_tick.fetch_add(1, Ordering::Relaxed) + 1;
1454 n % STATUS_POLL_TICKS == 0
1455 }
1456
1457 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
1459 let (limit, stale) = self.status_thresholds();
1460 let now = ek_ctp::map::now_epoch_ms();
1461 let mut acc = self.accounts.lock().unwrap();
1462 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1463 a.apply_status(st, now, limit, stale);
1464 drop(acc);
1465 self.bump();
1466 }
1467 }
1468
1469 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
1473 let mut acc = self.accounts.lock().unwrap();
1474 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1475 a.login = LoginState::Unknown;
1476 a.status_note = Some(format!("状态口问不到:{why}"));
1477 drop(acc);
1478 self.bump();
1479 }
1480 }
1481
1482 pub fn mark_account_stopped(&self, gateway: &str) {
1484 let mut acc = self.accounts.lock().unwrap();
1485 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1486 a.login = LoginState::Stopped;
1487 a.status_note = Some("会话已停;『连接』可恢复".to_string());
1488 drop(acc);
1489 self.bump();
1490 }
1491 }
1492
1493 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
1496 let source = self.md_source();
1497 let mut f = form;
1498 if source.as_str() != gateway {
1499 f.md_addr.clear();
1500 f.md_user.clear();
1501 f.md_password.clear();
1502 }
1503 self.update_account_form(gateway, f.clone());
1504 f
1505 }
1506
1507 #[must_use]
1511 pub fn draft_from_existing(&self) -> CtpForm {
1512 let acc = self.accounts.lock().unwrap();
1513 let mut draft = CtpForm::default();
1514 if let Some(a) = acc.last() {
1515 draft.broker_id = a.form.broker_id.clone();
1516 draft.td_auth_code = a.form.td_auth_code.clone();
1517 draft.td_app_id = a.form.td_app_id.clone();
1518 draft.accept_futures = a.form.accept_futures;
1519 draft.accept_options = a.form.accept_options;
1520 }
1521 draft
1522 }
1523
1524 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
1526 let mut acc = self.accounts.lock().unwrap();
1527 match acc.iter_mut().find(|a| a.gateway == gateway) {
1528 Some(a) => {
1529 a.form = form;
1530 drop(acc);
1531 self.bump();
1532 true
1533 }
1534 None => false,
1535 }
1536 }
1537
1538 pub fn remove_account(&self, gateway: &str) -> bool {
1540 let mut acc = self.accounts.lock().unwrap();
1541 let before = acc.len();
1542 acc.retain(|a| a.gateway != gateway);
1543 let removed = acc.len() != before;
1544 if removed {
1545 let mut md = self.md_source.lock().unwrap();
1547 if md.as_str() == gateway {
1548 md.clear();
1549 }
1550 drop(acc);
1551 self.bump();
1552 }
1553 removed
1554 }
1555
1556 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
1565 let mut newly = Vec::new();
1566 {
1567 let mut acc = self.accounts.lock().unwrap();
1568 for a in acc.iter_mut() {
1569 let up = rows.iter().any(|r| r.name == a.gateway);
1570 if up && !a.connected {
1571 newly.push(a.gateway.clone());
1572 }
1573 a.connected = up;
1574 if up {
1575 a.stage = AccountStage::Assembled;
1576 a.fail_note = None;
1577 } else if a.stage == AccountStage::Assembled {
1578 a.stage = AccountStage::NotAssembled;
1579 a.login = LoginState::Unknown;
1580 a.status_note = None;
1581 }
1582 }
1583 }
1584 if !newly.is_empty() {
1585 self.bump();
1586 }
1587 newly
1588 }
1589
1590 pub fn md_source(&self) -> String {
1592 self.md_source.lock().unwrap().clone()
1593 }
1594
1595 pub fn accounts_locked(&self) -> bool {
1597 self.accounts.lock().unwrap().iter().any(|a| a.connected)
1598 }
1599
1600 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
1602 {
1603 let acc = self.accounts.lock().unwrap();
1604 if acc.iter().any(|a| a.connected) {
1605 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
1606 }
1607 if !acc.iter().any(|a| a.gateway == gateway) {
1608 return Err(format!("账户 {gateway} 不在清册"));
1609 }
1610 }
1611 let mut acc = self.accounts.lock().unwrap();
1612 for a in acc.iter_mut() {
1613 a.md_source = a.gateway == gateway;
1614 }
1615 drop(acc);
1616 let mut md = self.md_source.lock().unwrap();
1617 *md = gateway.to_string();
1618 self.bump();
1619 Ok(())
1620 }
1621
1622 pub fn subscriptions(&self) -> Vec<String> {
1624 self.subscriptions.lock().unwrap().clone()
1625 }
1626
1627 pub fn note_subscriptions(&self, symbols: &[String]) {
1629 {
1630 let mut subs = self.subscriptions.lock().unwrap();
1631 for s in symbols {
1632 let s = s.trim();
1633 if !s.is_empty() && !subs.iter().any(|x| x == s) {
1634 subs.push(s.to_string());
1635 }
1636 }
1637 }
1638 self.bump();
1639 }
1640
1641 pub fn default_gateway(&self) -> Option<String> {
1648 let chosen = {
1649 let f = self.order_form.lock().unwrap();
1650 f.gateway.trim().to_string()
1651 };
1652 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
1653 return Some(chosen);
1654 }
1655 let acc = self.accounts.lock().unwrap();
1656 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
1657 }
1658
1659 pub fn kline_symbols(&self) -> Vec<String> {
1661 self.kline_windows.lock().unwrap().clone()
1662 }
1663
1664 pub fn open_kline(&self, symbol: impl Into<String>) {
1666 let s = symbol.into();
1667 let mut v = self.kline_windows.lock().unwrap();
1668 if !v.contains(&s) {
1669 v.push(s.clone());
1670 }
1671 drop(v);
1672 self.log_line(format!("[kline] 打开 K线窗 {s}"));
1673 }
1674
1675 pub fn close_kline(&self, symbol: &str) {
1677 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
1678 self.bump();
1679 }
1680
1681 pub fn contracts_window_open(&self) -> bool {
1685 self.contracts_window.load(Ordering::Acquire)
1686 }
1687
1688 pub fn set_contracts_window_open(&self, v: bool) {
1690 self.contracts_window.store(v, Ordering::Release);
1691 self.bump();
1692 }
1693
1694 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
1696 self.contract_store.lock().unwrap().put(c);
1697 self.bump();
1698 }
1699
1700 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1702 self.contract_store.lock().unwrap().query(pattern)
1703 }
1704
1705 pub fn contract_count(&self) -> usize {
1707 self.contract_store.lock().unwrap().len()
1708 }
1709
1710 pub fn contract_query(&self) -> String {
1712 self.contract_query.lock().unwrap().clone()
1713 }
1714
1715 pub fn set_contract_query(&self, q: impl Into<String>) {
1717 *self.contract_query.lock().unwrap() = q.into();
1718 self.bump();
1719 }
1720
1721 pub fn toggle_contract_selection(&self, en_symbol: &str) {
1723 let mut v = self.contract_selected.lock().unwrap();
1724 match v.iter().position(|s| s == en_symbol) {
1725 Some(i) => {
1726 v.remove(i);
1727 }
1728 None => v.push(en_symbol.to_string()),
1729 }
1730 self.bump();
1731 }
1732
1733 pub fn selected_contracts(&self) -> Vec<String> {
1735 self.contract_selected.lock().unwrap().clone()
1736 }
1737
1738 pub fn log_line(&self, line: impl Into<String>) {
1740 self.logs.lock().unwrap().push(line);
1741 self.bump();
1742 }
1743
1744 pub fn clear_logs(&self) {
1746 self.logs.lock().unwrap().clear();
1747 self.bump();
1748 }
1749
1750 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
1752 let r = self.logs.lock().unwrap();
1753 (r.snapshot(), r.spilled())
1754 }
1755
1756 pub fn logs_len(&self) -> usize {
1758 self.logs.lock().unwrap().len()
1759 }
1760
1761 pub fn set_snapshot(&self, snap: BusSnap) {
1763 *self.snap.lock().unwrap() = Arc::new(snap);
1764 self.bump();
1765 }
1766
1767 pub fn snapshot(&self) -> Arc<BusSnap> {
1769 Arc::clone(&self.snap.lock().unwrap())
1770 }
1771
1772 pub fn rev(&self) -> u64 {
1774 self.rev.load(Ordering::Acquire)
1775 }
1776
1777 pub fn bump(&self) {
1779 self.rev.fetch_add(1, Ordering::AcqRel);
1780 }
1781
1782 pub fn request_stop(&self) {
1784 self.stop.store(true, Ordering::Release);
1785 }
1786
1787 pub fn is_stop(&self) -> bool {
1789 self.stop.load(Ordering::Acquire)
1790 }
1791
1792 pub fn push_cmd(&self, cmd: UiCmd) {
1796 self.cmds.lock().unwrap().push_back(cmd);
1797 self.bump();
1798 }
1799
1800 pub fn drain_cmds(&self) -> Vec<UiCmd> {
1802 self.cmds.lock().unwrap().drain(..).collect()
1803 }
1804
1805 pub fn pending_confirm(&self) -> Option<String> {
1807 self.pending_confirm.lock().unwrap().clone()
1808 }
1809
1810 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
1812 let mut g = self.pending_confirm.lock().unwrap();
1813 match g.take() {
1814 Some(p) if p == name => ConfirmStep::Fired,
1815 Some(_) => {
1816 *g = Some(name.to_string());
1817 ConfirmStep::Switched
1818 }
1819 None => {
1820 *g = Some(name.to_string());
1821 ConfirmStep::Armed
1822 }
1823 }
1824 }
1825
1826 pub fn confirm_cancel(&self) {
1828 *self.pending_confirm.lock().unwrap() = None;
1829 self.bump();
1830 }
1831
1832 pub fn market(&self) -> &MarketStore {
1836 &self.market
1837 }
1838
1839 pub fn set_trade_snap(&self, ts: TradeSnap) {
1843 *self.trade_snap.lock().unwrap() = Arc::new(ts);
1844 self.bump();
1845 }
1846
1847 pub fn trade_snap(&self) -> Arc<TradeSnap> {
1849 self.trade_snap.lock().unwrap().clone()
1850 }
1851
1852 pub fn ctp_form(&self) -> CtpForm {
1856 self.ctp_form.lock().unwrap().clone()
1857 }
1858
1859 pub fn set_ctp_form(&self, f: CtpForm) {
1861 *self.ctp_form.lock().unwrap() = f;
1862 self.bump();
1863 }
1864
1865 pub fn row_selected(&self, table: &str, key: &str) -> bool {
1869 self.row_sel
1870 .lock()
1871 .unwrap()
1872 .get(table)
1873 .is_some_and(|k| k == key)
1874 }
1875
1876 pub fn select_row(&self, table: &str, key: &str) {
1878 self.row_sel
1879 .lock()
1880 .unwrap()
1881 .insert(table.to_string(), key.to_string());
1882 }
1883}