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}
168
169pub const KLINE_CELL_CAP: usize = 4096;
171
172pub const MAX_ACCOUNTS: usize = 8;
175
176pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
181
182pub const ACCOUNT_STALE_SECS: i64 = 30;
188
189pub use ek_ctp::driver::CHAIN_STATE_PREFIX;
195
196pub const LIST_POLL_TICKS: u32 = 4;
202
203pub const STATUS_ASKS_PER_TICK: usize = 2;
208
209pub const TRADE_CAP: usize = 5000;
214
215#[must_use]
221pub fn mailbox_cap_of(name: &str) -> usize {
222 if name == crate::plugin::UI_NAME {
223 crate::plugin::UI_MAILBOX_CAP
224 } else {
225 ek_core::plugin::MAILBOX_CAP
226 }
227}
228
229#[derive(Debug, Clone, Copy, PartialEq, Eq)]
231pub enum ConfirmStep {
232 Armed,
233 Fired,
234 Switched,
235}
236
237#[must_use]
240pub fn is_quarantined(state: &str) -> bool {
241 state.contains("Quarantin")
242}
243
244#[derive(Debug, Default)]
248pub struct MarketStore {
249 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
250 frames: AtomicU64,
252 updates: AtomicU64,
254 reloads: AtomicU64,
256}
257
258#[derive(Debug, Clone, PartialEq)]
260pub struct CellView {
261 pub symbol: String,
262 pub interval: i64,
263 pub bars: Vec<BarData>,
264}
265
266impl MarketStore {
267 pub fn on_frame(&self, frame: &BarsFrame) {
269 let mut cells = self.cells.lock().unwrap();
270 for b in &frame.bars {
271 let key = (b.symbol.clone(), frame.interval);
272 let cell = cells.entry(key).or_default();
273 match cell.last() {
274 Some(last) if b.datetime < last.datetime => continue,
275 Some(last) if b.datetime == last.datetime => {
276 let slot = cell.last_mut().expect("len>1");
277 *slot = b.clone();
278 }
279 _ => {
280 cell.push(b.clone());
281 if cell.len() > KLINE_CELL_CAP {
282 cell.remove(0);
283 }
284 }
285 }
286 }
287 self.frames.fetch_add(1, Ordering::AcqRel);
288 }
289
290 pub fn on_update(&self, b: &BarData) {
292 let mut cells = self.cells.lock().unwrap();
293 let iv = b
294 .interval
295 .unwrap_or(ek_data::constant::Interval::MINUTE)
296 .seconds();
297 let key = (b.symbol.clone(), iv);
298 let cell = cells.entry(key).or_default();
299 match cell.last() {
300 Some(last) if b.datetime == last.datetime => {
301 let slot = cell.last_mut().expect("len>1");
302 *slot = b.clone();
303 }
304 Some(last) if b.datetime < last.datetime => {}
305 _ => {
306 cell.push(b.clone());
307 if cell.len() > KLINE_CELL_CAP {
308 cell.remove(0);
309 }
310 }
311 }
312 self.updates.fetch_add(1, Ordering::AcqRel);
313 }
314
315 pub fn note_reload(&self) {
317 self.reloads.fetch_add(1, Ordering::AcqRel);
318 }
319
320 pub fn cells(&self) -> Vec<CellView> {
322 self.cells
323 .lock()
324 .unwrap()
325 .iter()
326 .map(|((sym, iv), bars)| CellView {
327 symbol: sym.clone(),
328 interval: *iv,
329 bars: bars.clone(),
330 })
331 .collect()
332 }
333
334 pub fn counts(&self) -> (u64, u64, u64) {
336 (
337 self.frames.load(Ordering::Acquire),
338 self.updates.load(Ordering::Acquire),
339 self.reloads.load(Ordering::Acquire),
340 )
341 }
342}
343
344#[derive(Debug, Default, Clone, PartialEq)]
347pub struct ReloadForm {
348 pub interval: String,
349 pub symbol: String,
350 pub csv: String,
351}
352
353pub fn parse_gateway(raw: &str, fallback: Option<&str>) -> Result<String, String> {
360 let g = raw.trim();
361 if !g.is_empty() {
362 if g.chars().any(|c| c.is_whitespace()) {
363 return Err(format!("网关解析: 含空白:{raw:?}"));
364 }
365 return Ok(g.to_string());
366 }
367 match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
368 Some(f) => Ok(f.to_string()),
369 None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
370 }
371}
372
373#[derive(Debug, Clone, PartialEq)]
376pub struct CtpForm {
377 pub broker_id: String,
378 pub td_addr: String,
379 pub td_user: String,
380 pub td_password: String,
381 pub td_auth_code: String,
382 pub td_app_id: String,
383 pub md_addr: String,
384 pub md_user: String,
385 pub md_password: String,
386 pub symbols: String,
388 pub accept_futures: bool,
391 pub accept_options: bool,
392}
393
394impl Default for CtpForm {
397 fn default() -> Self {
398 Self {
399 broker_id: String::new(),
400 td_addr: String::new(),
401 td_user: String::new(),
402 td_password: String::new(),
403 td_auth_code: String::new(),
404 td_app_id: String::new(),
405 md_addr: String::new(),
406 md_user: String::new(),
407 md_password: String::new(),
408 symbols: String::new(),
409 accept_futures: true,
410 accept_options: false,
411 }
412 }
413}
414
415impl CtpForm {
416 #[must_use]
420 pub fn product_classes(&self) -> Vec<String> {
421 let mut codes = Vec::new();
422 if self.accept_futures {
423 codes.push("1".to_string());
424 }
425 if self.accept_options {
426 codes.push("2".to_string());
427 }
428 codes
429 }
430
431 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
433 ek_ctp::plugin::ConfigureReq {
434 broker_id: self.broker_id.trim().to_string(),
435 td_addr: self.td_addr.trim().to_string(),
436 td_user: self.td_user.trim().to_string(),
437 td_password: self.td_password.clone(),
438 td_auth_code: self.td_auth_code.trim().to_string(),
439 td_app_id: self.td_app_id.trim().to_string(),
440 md_addr: self.md_addr.trim().to_string(),
441 md_user: self.md_user.trim().to_string(),
442 md_password: self.md_password.clone(),
443 product_classes: self.product_classes(),
444 }
445 }
446
447 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
449 let list: Vec<String> = self
450 .symbols
451 .split([',', ',', ' '])
452 .map(|s| s.trim().to_string())
453 .filter(|s| !s.is_empty())
454 .collect();
455 if list.is_empty() {
456 return Err("订阅解析: 标的位空".to_string());
457 }
458 Ok(list)
459 }
460
461 pub fn validate(&self) -> Result<(), String> {
470 fn front(name: &str, v: &str) -> Result<(), String> {
471 let v = v.trim();
472 let Some(rest) = v.strip_prefix("tcp://") else {
473 return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
474 };
475 let Some((host, port)) = rest.rsplit_once(':') else {
476 return Err(format!("{name} 缺端口:{name} {:?}", v));
477 };
478 if host.trim().is_empty() {
479 return Err(format!("{name} 主机位空:{}", v));
480 }
481 let port: u32 = port
482 .parse()
483 .map_err(|_| format!("{name} 端口不是数字:{}", v))?;
484 if port == 0 || port > 65535 {
485 return Err(format!("{name} 端口越界(1..=65535):{}", v));
486 }
487 Ok(())
488 }
489 if self.broker_id.trim().is_empty() {
490 return Err("broker 位空".to_string());
491 }
492 if self.td_user.trim().is_empty() {
493 return Err("交易账号位空".to_string());
494 }
495 front("td 前置", &self.td_addr)?;
496 if !self.md_addr.trim().is_empty() {
497 front("md 前置", &self.md_addr)?;
498 }
499 if self.td_app_id.trim().is_empty() {
500 return Err("AppID 位空(CTP 登录必需)".to_string());
501 }
502 if self.td_auth_code.trim().is_empty() {
503 return Err("AuthCode 位空(CTP 登录必需)".to_string());
504 }
505 Ok(())
506 }
507}
508
509#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
516pub enum AccountStage {
517 #[default]
519 NotAssembled,
520 Assembling,
522 Assembled,
524 AssembleFailed,
526}
527
528impl AccountStage {
529 #[must_use]
531 pub fn label(self) -> &'static str {
532 match self {
533 AccountStage::NotAssembled => "未装配",
534 AccountStage::Assembling => "装配中…",
535 AccountStage::Assembled => "已装配·登录态未知",
538 AccountStage::AssembleFailed => "装配失败",
539 }
540 }
541}
542
543#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
548pub enum LoginState {
549 #[default]
554 Unknown,
555 Online,
557 LoginFailed,
559 FrontDown,
561 QueryStarved,
563 Stopped,
565}
566
567impl LoginState {
568 #[must_use]
570 pub fn label(self) -> &'static str {
571 match self {
572 LoginState::Unknown => "登录态未知",
573 LoginState::Online => "在线",
574 LoginState::LoginFailed => "登录失败",
575 LoginState::FrontDown => "掉线",
576 LoginState::QueryStarved => "查询断供",
577 LoginState::Stopped => "已断开",
578 }
579 }
580
581 #[must_use]
583 pub fn is_alarm(self) -> bool {
584 matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
585 }
586}
587
588fn count_note(label: &str, n: u32) -> Option<String> {
590 (n > 0).then(|| format!("{label} {n} 次"))
591}
592
593fn brief_counter_reason(full: &str) -> Option<String> {
600 let tail = full.rfind("信息:").map(|i| &full[i + "信息:".len()..])?;
601 let body = tail.trim().trim_start_matches("CTP:").trim();
602 if body.is_empty() {
603 return None;
604 }
605 let code = full
606 .rfind("代码:")
607 .map(|i| {
608 let rest = &full[i + "代码:".len()..];
609 let end = rest.find(',').unwrap_or(rest.len());
610 rest[..end].trim().to_string()
611 })
612 .filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
613 Some(match code {
614 Some(c) => format!("{body}({c})"),
615 None => body.to_string(),
616 })
617}
618
619#[derive(Debug, Clone, PartialEq, Default)]
625pub struct AccountRow {
626 pub gateway: String,
628 pub alias: String,
630 pub broker_id: String,
631 pub account: String,
632 pub md_source: bool,
634 pub form: CtpForm,
636 pub stage: AccountStage,
638 pub fail_note: Option<String>,
640 pub effective_classes: Vec<String>,
645 pub connected: bool,
648 pub login: LoginState,
652 pub status_note: Option<String>,
655 pub login_brief: Option<String>,
662 pub chain_note: Option<String>,
670}
671
672impl AccountRow {
673 pub fn derive(form: &CtpForm) -> Result<Self, String> {
678 let broker = form.broker_id.trim();
679 let account = form.td_user.trim();
680 let gateway = ek_ctp::derived_gateway_name(broker, account).ok_or_else(|| {
681 "账户: broker/账号 不能为空且只能含字母数字(且派生名 ≤32)".to_string()
682 })?;
683 Ok(Self {
684 gateway,
685 alias: String::new(),
686 broker_id: broker.to_string(),
687 account: account.to_string(),
688 md_source: false,
689 form: form.clone(),
690 stage: AccountStage::NotAssembled,
691 fail_note: None,
692 effective_classes: Vec::new(),
693 connected: false,
694 login: LoginState::Unknown,
695 status_note: None,
696 login_brief: None,
697 chain_note: None,
698 })
699 }
700
701 #[must_use]
707 pub fn status_label(&self) -> String {
708 if self.stage == AccountStage::Assembled {
709 match self.login {
710 LoginState::Unknown => "已装配·登录态未知".to_string(),
711 other => format!("已装配·{}", other.label()),
712 }
713 } else {
714 self.stage.label().to_string()
715 }
716 }
717
718 pub fn apply_status(
731 &mut self,
732 st: &ek_ctp::plugin::GatewayStatus,
733 now_ms: i64,
734 reject_limit: u32,
735 stale_secs: i64,
736 ) {
737 let td_side = !self.form.td_addr.trim().is_empty();
738 let (login, note) = if td_side {
739 if st.td_login_failed && !st.td_logged_in {
740 (
741 LoginState::LoginFailed,
742 st.td_last_error
743 .clone()
744 .or_else(|| Some("登录失败(网关未回带原因)".to_string())),
745 )
746 } else if st.td_logged_in {
747 if st.td_query_reject_streak >= reject_limit {
748 (
749 LoginState::QueryStarved,
750 Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
751 )
752 } else if let Some(last) = st.td_last_account_ms {
753 let idle_ms = now_ms.saturating_sub(last);
754 if idle_ms > stale_secs.saturating_mul(1_000) {
755 (
756 LoginState::QueryStarved,
757 Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
758 )
759 } else {
760 (LoginState::Online, count_note("掉线", st.td_disconnects))
761 }
762 } else {
763 let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
770 match since_login_ms {
771 Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
772 LoginState::QueryStarved,
773 Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
774 ),
775 _ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
776 }
777 }
778 } else if self.login == LoginState::Stopped {
779 let extra = if st.td_disconnects > 0 {
780 format!("(此前掉线 {} 次)", st.td_disconnects)
781 } else {
782 String::new()
783 };
784 (
785 LoginState::Stopped,
786 Some(format!("会话已停{extra};『连接』可恢复")),
787 )
788 } else if st.td_disconnects > 0 {
789 (
790 LoginState::FrontDown,
791 Some(format!("掉线 {} 次", st.td_disconnects)),
792 )
793 } else {
794 (LoginState::Unknown, st.td_last_error.clone())
795 }
796 } else if st.md_logged_in {
797 (
798 LoginState::Online,
799 count_note("掉线", st.md_disconnects)
800 .or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
801 )
802 } else if self.login == LoginState::Stopped {
803 (
804 LoginState::Stopped,
805 Some("会话已停;『连接』可恢复".to_string()),
806 )
807 } else if st.md_disconnects > 0 {
808 (
809 LoginState::FrontDown,
810 Some(format!("掉线 {} 次", st.md_disconnects)),
811 )
812 } else {
813 (LoginState::Unknown, st.md_last_error.clone())
814 };
815 let note = if td_side && st.td_instrument_rejects > 0 {
820 let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
821 match note {
822 Some(base) if base.contains("合约查询被拒") => Some(base),
824 Some(base) => Some(format!("{base};{add}")),
825 None => Some(add),
826 }
827 } else {
828 note
829 };
830 self.login = login;
831 self.status_note = note;
832 self.login_brief = if login.is_alarm() {
835 st.td_last_error
836 .as_deref()
837 .or(st.md_last_error.as_deref())
838 .and_then(brief_counter_reason)
839 } else {
840 None
841 };
842 }
843
844 #[must_use]
846 pub fn display_name(&self) -> String {
847 let a = self.alias.trim();
848 if a.is_empty() {
849 self.account.clone()
850 } else {
851 a.to_string()
852 }
853 }
854}
855
856#[derive(Debug, Clone, PartialEq, Default)]
859pub struct TradeSnap {
860 pub orders: Vec<ek_data::order::OrderData>,
861 pub trades: Vec<ek_data::trade::TradeData>,
862 pub positions: Vec<ek_data::position::PositionData>,
863 pub accounts: Vec<ek_data::account::AccountData>,
865}
866
867pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
870 let interval: i64 = form
871 .interval
872 .trim()
873 .parse()
874 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
875 let symbol = form.symbol.trim().to_string();
876 if symbol.is_empty() {
877 return Err("回灌解析: symbol 空".to_string());
878 }
879 if form.csv.trim().is_empty() {
880 return Err("回灌解析: csv 空".to_string());
881 }
882 let mut bars = Vec::new();
883 for (i, raw) in form.csv.lines().enumerate() {
884 let line = raw.trim();
885 if line.is_empty() || line.starts_with('#') {
886 continue;
887 }
888 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
889 if parts.len() != 6 {
890 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
891 }
892 let num = |s: &str| -> Result<f64, String> {
893 s.parse::<f64>()
894 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
895 };
896 let dt_s: i64 = parts[0]
897 .parse()
898 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
899 bars.push(BarData {
900 symbol: symbol.clone(),
901 datetime: dt_s * 1_000_000_000,
902 interval: Some(
903 ek_dm::wire::interval_of_seconds(interval)
904 .unwrap_or(ek_data::constant::Interval::MINUTE),
905 ),
906 open_price: num(parts[1])?,
907 high_price: num(parts[2])?,
908 low_price: num(parts[3])?,
909 close_price: num(parts[4])?,
910 volume: num(parts[5])?,
911 ..BarData::default()
912 });
913 }
914 if bars.is_empty() {
915 return Err("回灌解析: 无有效行".to_string());
916 }
917 Ok((interval, symbol, bars))
918}
919
920#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
925pub enum TopLeftTab {
926 #[default]
927 Orders,
928 Active,
929}
930
931impl TopLeftTab {
932 pub fn label(self) -> &'static str {
933 match self {
934 Self::Orders => "委托",
935 Self::Active => "未成交",
936 }
937 }
938}
939
940#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
942pub enum BottomRightTab {
943 #[default]
944 Account,
945 Logs,
946}
947
948impl BottomRightTab {
949 pub fn label(self) -> &'static str {
950 match self {
951 Self::Account => "资金",
952 Self::Logs => "日志",
953 }
954 }
955}
956
957#[derive(Debug, Clone, Copy, PartialEq)]
960pub struct DataSplit {
961 pub col: f32,
963 pub row: f32,
965}
966
967impl Default for DataSplit {
968 fn default() -> Self {
969 Self {
970 col: crate::splitters::DATA_SPLIT_DEFAULT,
971 row: crate::splitters::DATA_SPLIT_DEFAULT,
972 }
973 }
974}
975
976pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
979 ek_data::constant::Exchange::SHFE,
980 ek_data::constant::Exchange::DCE,
981 ek_data::constant::Exchange::CZCE,
982 ek_data::constant::Exchange::CFFEX,
983 ek_data::constant::Exchange::INE,
984 ek_data::constant::Exchange::GFEX,
985];
986
987#[derive(Debug, Default)]
997pub struct TickStore {
998 ticks: BTreeMap<String, ek_data::tick::TickData>,
999}
1000
1001impl TickStore {
1002 pub fn put(&mut self, t: &ek_data::tick::TickData) {
1005 self.ticks
1006 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
1007 }
1008
1009 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
1011 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
1012 v.sort_by(|a, b| {
1013 (a.exchange.as_str().to_string(), a.symbol.clone())
1014 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
1015 });
1016 v
1017 }
1018
1019 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1021 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
1022 for t in self.rows() {
1023 let ex = t.exchange.as_str().to_string();
1024 match out.last_mut() {
1025 Some((e, v)) if *e == ex => v.push(t),
1026 _ => out.push((ex, vec![t])),
1027 }
1028 }
1029 out
1030 }
1031
1032 pub fn len(&self) -> usize {
1033 self.ticks.len()
1034 }
1035
1036 pub fn is_empty(&self) -> bool {
1037 self.ticks.is_empty()
1038 }
1039}
1040
1041#[derive(Debug, Clone, PartialEq)]
1043pub struct OrderForm {
1044 pub exchange: String,
1046 pub symbol: String,
1048 pub direction: ek_data::constant::Direction,
1049 pub offset: ek_data::constant::Offset,
1050 pub order_type: ek_data::constant::OrderType,
1051 pub price: String,
1053 pub volume: String,
1054 pub gateway: String,
1056 pub follow_price: bool,
1058 pub auto_split: bool,
1061}
1062
1063impl Default for OrderForm {
1064 fn default() -> Self {
1065 Self {
1066 exchange: "SHFE".to_string(),
1067 symbol: String::new(),
1068 direction: ek_data::constant::Direction::LONG,
1069 offset: ek_data::constant::Offset::OPEN,
1070 order_type: ek_data::constant::OrderType::LIMIT,
1071 price: String::new(),
1072 volume: "1".to_string(),
1073 gateway: String::new(),
1074 follow_price: true,
1075 auto_split: true,
1076 }
1077 }
1078}
1079
1080impl OrderForm {
1081 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
1083 use std::str::FromStr;
1084 let symbol = self.symbol.trim();
1085 if symbol.is_empty() {
1086 return Err("下单: 代码为空".to_string());
1087 }
1088 if symbol.contains('.') {
1089 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
1090 }
1091 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
1092 .map_err(|e| format!("下单: {e}"))?;
1093 let price: f64 = self
1094 .price
1095 .trim()
1096 .parse()
1097 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
1098 let volume: f64 = self
1099 .volume
1100 .trim()
1101 .parse()
1102 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
1103 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
1104 return Err("下单: 限价单价格须为正".to_string());
1105 }
1106 if volume <= 0.0 {
1107 return Err("下单: 数量须为正".to_string());
1108 }
1109 Ok(ek_data::order_request::OrderRequest {
1110 symbol: symbol.to_string(),
1111 exchange,
1112 direction: self.direction,
1113 order_type: self.order_type,
1114 volume,
1115 price,
1116 offset: self.offset,
1117 ..ek_data::order_request::OrderRequest::default()
1118 })
1119 }
1120}
1121
1122#[derive(Debug, Default)]
1128pub struct ContractStore {
1129 contracts: BTreeMap<String, ek_data::contract::ContractData>,
1130}
1131
1132impl ContractStore {
1133 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
1134 self.contracts
1135 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
1136 }
1137
1138 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
1140 let p = pattern.trim();
1141 self.contracts
1142 .values()
1143 .filter(|c| {
1144 p.is_empty()
1145 || c.symbol.contains(p)
1146 || c.exchange.as_str().eq_ignore_ascii_case(p)
1147 || c.name.contains(p)
1148 })
1149 .cloned()
1150 .collect()
1151 }
1152
1153 pub fn len(&self) -> usize {
1154 self.contracts.len()
1155 }
1156
1157 pub fn is_empty(&self) -> bool {
1158 self.contracts.is_empty()
1159 }
1160}
1161
1162#[derive(Debug, Default)]
1164pub struct UiShared {
1165 logs: Mutex<LogRing>,
1166 snap: Mutex<Arc<BusSnap>>,
1167 market: MarketStore,
1168 ctp_form: Mutex<CtpForm>,
1169 trade_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
1172 trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
1173 trade_ids: Mutex<HashSet<String>>,
1181 trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
1182 trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
1183 trade_seq: AtomicU64,
1185 trade_dropped: AtomicU64,
1187 trade_dirty: AtomicBool,
1189 trade_snap: Mutex<Arc<TradeSnap>>,
1191 tick_store: Mutex<TickStore>,
1192 order_form: Mutex<OrderForm>,
1193 data_split: Mutex<DataSplit>,
1196 selected_exchange: Mutex<String>,
1198 topleft_tab: Mutex<TopLeftTab>,
1200 bottomright_tab: Mutex<BottomRightTab>,
1202 ops_window: AtomicBool,
1203 accounts_window: AtomicBool,
1206 contracts_window: AtomicBool,
1207 contract_store: Mutex<ContractStore>,
1208 contract_query: Mutex<String>,
1209 contract_selected: Mutex<Vec<String>>,
1210 kline_windows: Mutex<Vec<String>>,
1211 rev: AtomicU64,
1212 stop: AtomicBool,
1214 cmds: Mutex<VecDeque<UiCmd>>,
1216 pending_confirm: Mutex<Option<String>>,
1218 accounts: Mutex<Vec<AccountRow>>,
1220 account_edit_window: AtomicBool,
1222 editing_account: Mutex<Option<String>>,
1224 edit_alias: Mutex<String>,
1226 edit_error: Mutex<Option<String>>,
1228 md_source: Mutex<String>,
1230 subscriptions: Mutex<Vec<String>>,
1232 row_sel: Mutex<BTreeMap<String, String>>,
1235 table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
1239 status_cursor: AtomicUsize,
1243 list_tick: AtomicU32,
1245 list_due: AtomicBool,
1249 seed_due: AtomicBool,
1253 pipeline_due: AtomicBool,
1255 bad_payload: Mutex<BTreeMap<String, u64>>,
1257 pipeline_lines: Mutex<Vec<String>>,
1260 status_thresholds: Mutex<Option<(u32, i64)>>,
1263 pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
1265 toasts: crate::toast::ToastsCell,
1268 alerted: Mutex<std::collections::BTreeSet<String>>,
1270 exit_confirm_open: AtomicBool,
1272 exit_confirmed: AtomicBool,
1274}
1275
1276impl UiShared {
1277 pub fn new() -> Self {
1279 Self::default()
1280 }
1281
1282 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
1286 self.tick_store.lock().unwrap().put(t);
1287 self.bump();
1288 }
1289
1290 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
1292 self.tick_store.lock().unwrap().rows()
1293 }
1294
1295 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
1297 self.tick_store.lock().unwrap().groups()
1298 }
1299
1300 pub fn tick_count(&self) -> usize {
1302 self.tick_store.lock().unwrap().len()
1303 }
1304
1305 pub fn order_form(&self) -> OrderForm {
1307 self.order_form.lock().unwrap().clone()
1308 }
1309
1310 pub fn set_order_form(&self, f: OrderForm) {
1312 *self.order_form.lock().unwrap() = f;
1313 self.bump();
1314 }
1315
1316 pub fn selected_exchange(&self) -> String {
1320 let s = self.selected_exchange.lock().unwrap().clone();
1321 if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
1322 s
1323 } else {
1324 MARKET_ORDER[0].as_str().to_string()
1325 }
1326 }
1327
1328 pub fn set_selected_exchange(&self, s: impl Into<String>) {
1330 *self.selected_exchange.lock().unwrap() = s.into();
1331 self.bump();
1332 }
1333
1334 pub fn topleft_tab(&self) -> TopLeftTab {
1336 *self.topleft_tab.lock().unwrap()
1337 }
1338
1339 pub fn set_topleft_tab(&self, t: TopLeftTab) {
1341 *self.topleft_tab.lock().unwrap() = t;
1342 self.bump();
1343 }
1344
1345 pub fn bottomright_tab(&self) -> BottomRightTab {
1347 *self.bottomright_tab.lock().unwrap()
1348 }
1349
1350 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
1352 *self.bottomright_tab.lock().unwrap() = t;
1353 self.bump();
1354 }
1355
1356 pub fn data_split(&self) -> DataSplit {
1358 *self.data_split.lock().unwrap()
1359 }
1360
1361 pub fn set_data_split(&self, s: DataSplit) {
1364 let s = DataSplit {
1365 col: crate::splitters::clamp_frac(s.col),
1366 row: crate::splitters::clamp_frac(s.row),
1367 };
1368 let mut g = self.data_split.lock().unwrap();
1369 if g.col != s.col || g.row != s.row {
1370 *g = s;
1371 self.bump();
1372 }
1373 }
1374
1375 pub fn reset_layout(&self) {
1379 *self.data_split.lock().unwrap() = DataSplit::default();
1380 *self.selected_exchange.lock().unwrap() = String::new();
1381 *self.topleft_tab.lock().unwrap() = TopLeftTab::default();
1382 *self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
1383 self.bump();
1384 }
1385
1386 pub fn ops_window_open(&self) -> bool {
1390 self.ops_window.load(Ordering::Acquire)
1391 }
1392
1393 pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
1396 self.table_sorts
1397 .lock()
1398 .unwrap()
1399 .get(table)
1400 .copied()
1401 .flatten()
1402 }
1403
1404 pub fn toggle_table_sort(&self, table: &str, col: usize) {
1406 let mut m = self.table_sorts.lock().unwrap();
1407 let asc = match m.get(table).copied().flatten() {
1408 Some((c, a)) if c == col => !a,
1409 _ => true,
1410 };
1411 m.insert(table.to_string(), Some((col, asc)));
1412 }
1413
1414 pub fn set_ops_window_open(&self, v: bool) {
1415 self.ops_window.store(v, Ordering::Release);
1416 self.bump();
1417 }
1418
1419 pub fn accounts_window_open(&self) -> bool {
1423 self.accounts_window.load(Ordering::Acquire)
1424 }
1425
1426 pub fn set_accounts_window_open(&self, v: bool) {
1428 self.accounts_window.store(v, Ordering::Release);
1429 self.bump();
1430 }
1431
1432 pub fn account_edit_open(&self) -> bool {
1434 self.account_edit_window.load(Ordering::Acquire)
1435 }
1436
1437 pub fn editing_account(&self) -> Option<String> {
1439 self.editing_account.lock().unwrap().clone()
1440 }
1441
1442 pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
1446 match gateway {
1447 Some(gw) => {
1448 let Some(row) = self.account(gw) else {
1449 return false;
1450 };
1451 *self.editing_account.lock().unwrap() = Some(gw.to_string());
1452 *self.edit_alias.lock().unwrap() = row.alias.clone();
1453 self.set_ctp_form(row.form.clone());
1454 }
1455 None => {
1456 *self.editing_account.lock().unwrap() = None;
1457 *self.edit_alias.lock().unwrap() = String::new();
1458 let draft = self.draft_from_existing();
1459 self.set_ctp_form(draft);
1460 }
1461 }
1462 *self.edit_error.lock().unwrap() = None;
1463 self.account_edit_window.store(true, Ordering::Release);
1464 self.bump();
1465 true
1466 }
1467
1468 pub fn end_account_edit(&self) {
1470 *self.editing_account.lock().unwrap() = None;
1471 *self.edit_alias.lock().unwrap() = String::new();
1472 *self.edit_error.lock().unwrap() = None;
1473 self.account_edit_window.store(false, Ordering::Release);
1474 let draft = self.draft_from_existing();
1475 self.set_ctp_form(draft);
1476 self.bump();
1477 }
1478
1479 pub fn end_account_edit_if_editing(&self, gateway: &str) {
1485 if self.editing_account().as_deref() == Some(gateway) {
1486 self.end_account_edit();
1487 }
1488 }
1489
1490 pub fn edit_alias(&self) -> String {
1492 self.edit_alias.lock().unwrap().clone()
1493 }
1494
1495 pub fn set_edit_alias(&self, v: impl Into<String>) {
1497 *self.edit_alias.lock().unwrap() = v.into();
1498 self.bump();
1499 }
1500
1501 pub fn edit_error(&self) -> Option<String> {
1503 self.edit_error.lock().unwrap().clone()
1504 }
1505
1506 pub fn set_edit_error(&self, e: Option<String>) {
1508 *self.edit_error.lock().unwrap() = e;
1509 self.bump();
1510 }
1511
1512 pub fn account_edit_dirty(&self) -> bool {
1514 let editing = self.editing_account();
1515 let draft = self.ctp_form();
1516 match editing {
1517 Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
1518 None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
1519 }
1520 }
1521
1522 pub fn accounts(&self) -> Vec<AccountRow> {
1524 self.accounts.lock().unwrap().clone()
1525 }
1526
1527 pub fn account(&self, gateway: &str) -> Option<AccountRow> {
1529 self.accounts
1530 .lock()
1531 .unwrap()
1532 .iter()
1533 .find(|a| a.gateway == gateway)
1534 .cloned()
1535 }
1536
1537 pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
1543 let mut acc = self.accounts.lock().unwrap();
1544 if acc.iter().any(|a| a.gateway == row.gateway) {
1545 return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
1546 }
1547 let alias = row.alias.trim();
1548 if !alias.is_empty()
1549 && acc
1550 .iter()
1551 .any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
1552 {
1553 return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
1554 }
1555 if acc.len() >= MAX_ACCOUNTS {
1556 return Err(format!(
1557 "账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
1558 ));
1559 }
1560 acc.push(row);
1561 drop(acc);
1562 self.bump();
1563 Ok(())
1564 }
1565
1566 #[must_use]
1568 pub fn can_add_account(&self) -> bool {
1569 self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
1570 }
1571
1572 pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
1578 let alias = alias.trim().to_string();
1579 let mut acc = self.accounts.lock().unwrap();
1580 if !alias.is_empty()
1581 && acc
1582 .iter()
1583 .any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
1584 {
1585 return false;
1586 }
1587 match acc.iter_mut().find(|a| a.gateway == gateway) {
1588 Some(a) => {
1589 a.alias = alias;
1590 drop(acc);
1591 self.bump();
1592 true
1593 }
1594 None => false,
1595 }
1596 }
1597
1598 pub fn note_connect_sent(&self, gateway: &str) {
1603 let mut acc = self.accounts.lock().unwrap();
1604 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1605 a.stage = AccountStage::Assembling;
1606 a.fail_note = None;
1607 a.login = LoginState::Unknown;
1608 a.status_note = None;
1609 a.login_brief = None;
1610 a.chain_note = None;
1612 drop(acc);
1613 self.bump();
1614 }
1615 }
1616
1617 pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
1621 let mut acc = self.accounts.lock().unwrap();
1622 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1623 match result {
1624 Ok(()) => {
1625 a.stage = AccountStage::Assembled;
1626 a.fail_note = None;
1627 }
1628 Err(e) => {
1629 a.stage = AccountStage::AssembleFailed;
1630 a.fail_note = Some(e);
1631 }
1632 }
1633 drop(acc);
1634 self.bump();
1635 }
1636 }
1637
1638 pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
1642 let mut acc = self.accounts.lock().unwrap();
1643 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1644 a.effective_classes = classes.to_vec();
1645 drop(acc);
1646 self.bump();
1647 }
1648 }
1649
1650 pub fn note_chain(&self, gateway: &str, note: &str) {
1655 let mut acc = self.accounts.lock().unwrap();
1656 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1657 if a.chain_note.as_deref() == Some(note) {
1658 return; }
1660 a.chain_note = Some(note.to_string());
1661 drop(acc);
1662 self.bump();
1663 }
1664 }
1665
1666 #[must_use]
1670 pub fn status_thresholds(&self) -> (u32, i64) {
1671 self.status_thresholds
1672 .lock()
1673 .unwrap()
1674 .unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
1675 }
1676
1677 pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
1679 *self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
1680 self.bump();
1681 }
1682
1683 pub fn list_poll_due(&self) -> bool {
1688 let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
1689 self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
1690 }
1691
1692 pub fn request_list_refresh(&self) {
1694 self.list_due.store(true, Ordering::Release);
1695 }
1696
1697 pub fn request_trade_reseed(&self) {
1699 self.seed_due.store(true, Ordering::Release);
1700 }
1701
1702 pub fn take_trade_reseed(&self) -> bool {
1704 self.seed_due.swap(false, Ordering::AcqRel)
1705 }
1706
1707 pub fn status_batch(&self, present: &[String]) -> Vec<String> {
1712 let mut uniq: Vec<String> = Vec::with_capacity(present.len());
1716 for gw in present {
1717 if !uniq.iter().any(|u| u == gw) {
1718 uniq.push(gw.clone());
1719 }
1720 }
1721 let n = uniq.len();
1722 if n == 0 {
1723 return Vec::new();
1724 }
1725 let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
1726 let take = STATUS_ASKS_PER_TICK.min(n);
1727 let mut out = Vec::with_capacity(take);
1728 for i in 0..take {
1729 out.push(uniq[(start + i) % n].clone());
1730 }
1731 out
1732 }
1733
1734 pub fn request_pipeline_refresh(&self) {
1736 self.pipeline_due.store(true, Ordering::Release);
1737 }
1738 pub fn take_pipeline_due(&self) -> bool {
1740 self.pipeline_due.swap(false, Ordering::AcqRel)
1741 }
1742
1743 pub fn set_pipeline_lines(&self, lines: Vec<String>) {
1745 *self.pipeline_lines.lock().unwrap() = lines;
1746 self.bump();
1747 }
1748
1749 pub fn pipeline_lines(&self) -> Vec<String> {
1751 self.pipeline_lines.lock().unwrap().clone()
1752 }
1753
1754 pub fn note_bad_payload(&self, line: &str) {
1760 let mut book = self.bad_payload.lock().unwrap();
1761 let n = book.entry(line.to_string()).or_insert(0);
1762 *n += 1;
1763 if *n == 1 {
1764 self.log_line(format!(
1765 "[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
1766 ));
1767 }
1768 }
1769
1770 #[must_use]
1772 pub fn bad_payload_count(&self, line: &str) -> u64 {
1773 self.bad_payload
1774 .lock()
1775 .unwrap()
1776 .get(line)
1777 .copied()
1778 .unwrap_or(0)
1779 }
1780
1781 pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
1790 for r in rows {
1791 if r.spilled == 0 {
1792 continue;
1793 }
1794 let cap = mailbox_cap_of(&r.name);
1795 if self.alert_once(format!("spill:{}", r.name)) {
1796 self.push_alert(
1797 crate::toast::AlertLevel::Warning,
1798 format!(
1799 "插件 {}:邮箱溢出 {} 条(刷新停顿时行情会堆;溢出丢的是新到的事件)",
1800 r.name, r.spilled
1801 ),
1802 );
1803 }
1804 if r.spilled >= cap as u64 && self.alert_once(format!("spill-full:{}", r.name)) {
1805 self.push_alert(
1806 crate::toast::AlertLevel::Error,
1807 format!(
1808 "插件 {}:溢出 {} 条 ≥ 该插件邮箱容量 {}(丢的是一段时期的事件,不是零星几条)",
1809 r.name, r.spilled, cap
1810 ),
1811 );
1812 }
1813 }
1814 }
1815
1816 pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
1824 let (limit, stale) = self.status_thresholds();
1825 let now = ek_ctp::map::now_epoch_ms();
1826 let mut alarm: Option<(LoginState, String)> = None;
1829 let mut found = false;
1830 {
1831 let mut acc = self.accounts.lock().unwrap();
1832 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1833 found = true;
1834 a.apply_status(st, now, limit, stale);
1835 if a.login.is_alarm() || a.login == LoginState::QueryStarved {
1836 alarm = Some((
1837 a.login,
1838 a.status_note
1839 .clone()
1840 .unwrap_or_else(|| a.login.label().to_string()),
1841 ));
1842 }
1843 }
1844 }
1845 if found {
1846 self.bump();
1847 }
1848 if let Some((login, note)) = alarm
1849 && self.alert_once(format!("{gateway}|{}", login.label()))
1850 {
1851 self.push_alert(
1852 crate::toast::AlertLevel::Error,
1853 format!("账户 {gateway}:{note}"),
1854 );
1855 }
1856 }
1857
1858 pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
1862 let mut acc = self.accounts.lock().unwrap();
1863 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1864 a.login = LoginState::Unknown;
1865 a.status_note = Some(format!("状态口问不到:{why}"));
1866 a.login_brief = None;
1867 drop(acc);
1868 self.bump();
1869 }
1870 }
1871
1872 pub fn mark_account_stopped(&self, gateway: &str) {
1874 let mut acc = self.accounts.lock().unwrap();
1875 if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
1876 a.login = LoginState::Stopped;
1877 a.status_note = Some("会话已停;『连接』可恢复".to_string());
1878 a.login_brief = None;
1879 drop(acc);
1880 self.bump();
1881 }
1882 }
1883
1884 pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
1887 let source = self.md_source();
1888 let mut f = form;
1889 if source.as_str() != gateway {
1890 f.md_addr.clear();
1891 f.md_user.clear();
1892 f.md_password.clear();
1893 }
1894 self.update_account_form(gateway, f.clone());
1895 f
1896 }
1897
1898 #[must_use]
1902 pub fn draft_from_existing(&self) -> CtpForm {
1903 let acc = self.accounts.lock().unwrap();
1904 let mut draft = CtpForm::default();
1905 if let Some(a) = acc.last() {
1906 draft.broker_id = a.form.broker_id.clone();
1907 draft.td_auth_code = a.form.td_auth_code.clone();
1908 draft.td_app_id = a.form.td_app_id.clone();
1909 draft.accept_futures = a.form.accept_futures;
1910 draft.accept_options = a.form.accept_options;
1911 }
1912 draft
1913 }
1914
1915 pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
1917 let mut acc = self.accounts.lock().unwrap();
1918 match acc.iter_mut().find(|a| a.gateway == gateway) {
1919 Some(a) => {
1920 a.form = form;
1921 drop(acc);
1922 self.bump();
1923 true
1924 }
1925 None => false,
1926 }
1927 }
1928
1929 pub fn remove_account(&self, gateway: &str) -> bool {
1931 let mut acc = self.accounts.lock().unwrap();
1932 let before = acc.len();
1933 acc.retain(|a| a.gateway != gateway);
1934 let removed = acc.len() != before;
1935 if removed {
1936 let mut md = self.md_source.lock().unwrap();
1938 if md.as_str() == gateway {
1939 md.clear();
1940 }
1941 drop(acc);
1942 self.bump();
1943 }
1944 removed
1945 }
1946
1947 pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
1956 let mut newly = Vec::new();
1957 {
1958 let mut acc = self.accounts.lock().unwrap();
1959 for a in acc.iter_mut() {
1960 let up = rows.iter().any(|r| r.name == a.gateway);
1961 if up && !a.connected {
1962 newly.push(a.gateway.clone());
1963 }
1964 a.connected = up;
1965 if up {
1966 a.stage = AccountStage::Assembled;
1967 a.fail_note = None;
1968 } else {
1969 a.chain_note = None;
1972 if a.stage == AccountStage::Assembled {
1973 a.stage = AccountStage::NotAssembled;
1974 a.login = LoginState::Unknown;
1975 a.status_note = None;
1976 a.login_brief = None;
1977 }
1978 }
1979 }
1980 }
1981 if !newly.is_empty() {
1982 self.bump();
1983 }
1984 newly
1985 }
1986
1987 pub fn md_source(&self) -> String {
1989 self.md_source.lock().unwrap().clone()
1990 }
1991
1992 pub fn accounts_locked(&self) -> bool {
1994 self.accounts.lock().unwrap().iter().any(|a| a.connected)
1995 }
1996
1997 pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
1999 {
2000 let acc = self.accounts.lock().unwrap();
2001 if acc.iter().any(|a| a.connected) {
2002 return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
2003 }
2004 if !acc.iter().any(|a| a.gateway == gateway) {
2005 return Err(format!("账户 {gateway} 不在清册"));
2006 }
2007 }
2008 let mut acc = self.accounts.lock().unwrap();
2009 for a in acc.iter_mut() {
2010 a.md_source = a.gateway == gateway;
2011 }
2012 drop(acc);
2013 let mut md = self.md_source.lock().unwrap();
2014 *md = gateway.to_string();
2015 self.bump();
2016 Ok(())
2017 }
2018
2019 pub fn subscriptions(&self) -> Vec<String> {
2021 self.subscriptions.lock().unwrap().clone()
2022 }
2023
2024 pub fn note_subscriptions(&self, symbols: &[String]) {
2026 {
2027 let mut subs = self.subscriptions.lock().unwrap();
2028 for s in symbols {
2029 let s = s.trim();
2030 if !s.is_empty() && !subs.iter().any(|x| x == s) {
2031 subs.push(s.to_string());
2032 }
2033 }
2034 }
2035 self.bump();
2036 }
2037
2038 pub fn default_gateway(&self) -> Option<String> {
2045 let chosen = {
2046 let f = self.order_form.lock().unwrap();
2047 f.gateway.trim().to_string()
2048 };
2049 if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
2050 return Some(chosen);
2051 }
2052 let acc = self.accounts.lock().unwrap();
2053 acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
2054 }
2055
2056 pub fn kline_symbols(&self) -> Vec<String> {
2058 self.kline_windows.lock().unwrap().clone()
2059 }
2060
2061 pub fn open_kline(&self, symbol: impl Into<String>) {
2063 let s = symbol.into();
2064 let mut v = self.kline_windows.lock().unwrap();
2065 if !v.contains(&s) {
2066 v.push(s.clone());
2067 }
2068 drop(v);
2069 self.log_line(format!("[kline] 打开 K线窗 {s}"));
2070 }
2071
2072 pub fn close_kline(&self, symbol: &str) {
2074 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
2075 self.bump();
2076 }
2077
2078 pub fn contracts_window_open(&self) -> bool {
2082 self.contracts_window.load(Ordering::Acquire)
2083 }
2084
2085 pub fn set_contracts_window_open(&self, v: bool) {
2087 self.contracts_window.store(v, Ordering::Release);
2088 self.bump();
2089 }
2090
2091 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
2093 self.contract_store.lock().unwrap().put(c);
2094 self.bump();
2095 }
2096
2097 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
2099 self.contract_store.lock().unwrap().query(pattern)
2100 }
2101
2102 pub fn contract_count(&self) -> usize {
2104 self.contract_store.lock().unwrap().len()
2105 }
2106
2107 pub fn contract_query(&self) -> String {
2109 self.contract_query.lock().unwrap().clone()
2110 }
2111
2112 pub fn set_contract_query(&self, q: impl Into<String>) {
2114 *self.contract_query.lock().unwrap() = q.into();
2115 self.bump();
2116 }
2117
2118 pub fn toggle_contract_selection(&self, en_symbol: &str) {
2120 let mut v = self.contract_selected.lock().unwrap();
2121 match v.iter().position(|s| s == en_symbol) {
2122 Some(i) => {
2123 v.remove(i);
2124 }
2125 None => v.push(en_symbol.to_string()),
2126 }
2127 self.bump();
2128 }
2129
2130 pub fn selected_contracts(&self) -> Vec<String> {
2132 self.contract_selected.lock().unwrap().clone()
2133 }
2134
2135 pub fn log_line(&self, line: impl Into<String>) {
2137 self.logs.lock().unwrap().push(line);
2138 self.bump();
2139 }
2140
2141 pub fn clear_logs(&self) {
2143 self.logs.lock().unwrap().clear();
2144 self.bump();
2145 }
2146
2147 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
2149 let r = self.logs.lock().unwrap();
2150 (r.snapshot(), r.spilled())
2151 }
2152
2153 pub fn logs_entries(&self) -> (Vec<LogEntry>, u64) {
2155 let r = self.logs.lock().unwrap();
2156 (r.entries(), r.spilled())
2157 }
2158
2159 pub fn logs_len(&self) -> usize {
2161 self.logs.lock().unwrap().len()
2162 }
2163
2164 pub fn set_snapshot(&self, snap: BusSnap) {
2166 *self.snap.lock().unwrap() = Arc::new(snap);
2167 self.bump();
2168 }
2169
2170 pub fn snapshot(&self) -> Arc<BusSnap> {
2172 Arc::clone(&self.snap.lock().unwrap())
2173 }
2174
2175 pub fn rev(&self) -> u64 {
2177 self.rev.load(Ordering::Acquire)
2178 }
2179
2180 pub fn bump(&self) {
2182 self.rev.fetch_add(1, Ordering::AcqRel);
2183 }
2184
2185 pub fn request_stop(&self) {
2187 self.stop.store(true, Ordering::Release);
2188 }
2189
2190 pub fn push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
2193 self.pending_alerts
2194 .lock()
2195 .unwrap_or_else(std::sync::PoisonError::into_inner)
2196 .push((level, text.into()));
2197 }
2198
2199 pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
2201 let mut q = self
2202 .pending_alerts
2203 .lock()
2204 .unwrap_or_else(std::sync::PoisonError::into_inner);
2205 std::mem::take(&mut *q)
2206 }
2207
2208 pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
2210 self.toasts.lock()
2211 }
2212
2213 pub fn alert_once(&self, key: impl Into<String>) -> bool {
2215 let key = key.into();
2216 self.alerted
2217 .lock()
2218 .unwrap_or_else(std::sync::PoisonError::into_inner)
2219 .insert(key)
2220 }
2221
2222 #[must_use]
2224 pub fn pending_alert_count(&self) -> usize {
2225 self.pending_alerts
2226 .lock()
2227 .unwrap_or_else(std::sync::PoisonError::into_inner)
2228 .len()
2229 }
2230
2231 #[must_use]
2233 pub fn alerted_count(&self) -> usize {
2234 self.alerted
2235 .lock()
2236 .unwrap_or_else(std::sync::PoisonError::into_inner)
2237 .len()
2238 }
2239
2240 pub fn exit_confirm_open(&self) -> bool {
2242 self.exit_confirm_open
2243 .load(std::sync::atomic::Ordering::Relaxed)
2244 }
2245
2246 pub fn open_exit_confirm(&self) {
2247 self.exit_confirm_open
2248 .store(true, std::sync::atomic::Ordering::Relaxed);
2249 }
2250
2251 pub fn close_exit_confirm(&self) {
2252 self.exit_confirm_open
2253 .store(false, std::sync::atomic::Ordering::Relaxed);
2254 }
2255
2256 pub fn exit_confirmed(&self) -> bool {
2257 self.exit_confirmed
2258 .load(std::sync::atomic::Ordering::Relaxed)
2259 }
2260
2261 pub fn confirm_exit(&self) {
2262 self.exit_confirmed
2263 .store(true, std::sync::atomic::Ordering::Relaxed);
2264 }
2265
2266 pub fn is_stop(&self) -> bool {
2267 self.stop.load(Ordering::Acquire)
2268 }
2269
2270 pub fn push_cmd(&self, cmd: UiCmd) {
2274 self.cmds.lock().unwrap().push_back(cmd);
2275 self.bump();
2276 }
2277
2278 pub fn drain_cmds(&self) -> Vec<UiCmd> {
2280 self.cmds.lock().unwrap().drain(..).collect()
2281 }
2282
2283 #[must_use]
2285 pub fn cmd_pending(&self) -> bool {
2286 !self.cmds.lock().unwrap().is_empty()
2287 }
2288
2289 pub fn pending_confirm(&self) -> Option<String> {
2291 self.pending_confirm.lock().unwrap().clone()
2292 }
2293
2294 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
2296 let mut g = self.pending_confirm.lock().unwrap();
2297 match g.take() {
2298 Some(p) if p == name => ConfirmStep::Fired,
2299 Some(_) => {
2300 *g = Some(name.to_string());
2301 ConfirmStep::Switched
2302 }
2303 None => {
2304 *g = Some(name.to_string());
2305 ConfirmStep::Armed
2306 }
2307 }
2308 }
2309
2310 pub fn confirm_cancel(&self) {
2312 *self.pending_confirm.lock().unwrap() = None;
2313 self.bump();
2314 }
2315
2316 pub fn market(&self) -> &MarketStore {
2320 &self.market
2321 }
2322
2323 pub fn seed_trade_snap(&self, ts: TradeSnap) {
2332 {
2333 let mut book = self.trade_orders.lock().unwrap();
2334 book.clear();
2335 for o in ts.orders.into_iter().rev() {
2336 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2337 book.insert(o.en_orderid(), (seq, o));
2338 }
2339 }
2340 {
2341 let mut book = self.trade_trades.lock().unwrap();
2342 book.clear();
2343 let mut ids = self.trade_ids.lock().unwrap();
2344 ids.clear();
2345 for t in ts.trades.into_iter().rev() {
2346 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2347 ids.insert(t.en_tradeid());
2348 book.push_front((seq, t));
2349 }
2350 }
2351 {
2352 let mut book = self.trade_positions.lock().unwrap();
2353 book.clear();
2354 for p in ts.positions.into_iter().rev() {
2355 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2356 book.insert(p.en_positionid(), (seq, p));
2357 }
2358 }
2359 {
2360 let mut book = self.trade_funds.lock().unwrap();
2361 book.clear();
2362 for a in ts.accounts.into_iter().rev() {
2363 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2364 book.insert(a.en_accountid(), (seq, a));
2365 }
2366 }
2367 self.trade_dirty.store(true, Ordering::Release);
2368 self.bump();
2369 }
2370
2371 pub fn note_order(&self, o: &ek_data::order::OrderData) {
2379 let key = o.en_orderid();
2380 {
2381 let mut book = self.trade_orders.lock().unwrap();
2382 match book.get_mut(&key) {
2383 Some((_, cur)) => {
2384 cur.name.clone_from(&o.name);
2385 cur.traded = o.traded;
2386 cur.traded_price = o.traded_price;
2387 cur.status = o.status;
2388 cur.status_msg.clone_from(&o.status_msg);
2389 cur.datetime = o.datetime;
2390 }
2391 None => {
2392 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2393 book.insert(key, (seq, o.clone()));
2394 }
2395 }
2396 }
2397 self.trade_dirty.store(true, Ordering::Release);
2398 self.bump();
2399 }
2400
2401 pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
2409 let id = t.en_tradeid();
2410 if self.trade_ids.lock().unwrap().contains(&id) {
2411 return;
2412 }
2413 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2414 let dropped = {
2415 let mut book = self.trade_trades.lock().unwrap();
2416 book.push_front((seq, t.clone()));
2417 self.trade_ids.lock().unwrap().insert(id);
2418 let over = book.len().saturating_sub(TRADE_CAP);
2419 for _ in 0..over {
2420 if let Some((_, old)) = book.pop_back() {
2422 self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
2423 }
2424 }
2425 over
2426 };
2427 if dropped > 0 {
2428 let total = self
2429 .trade_dropped
2430 .fetch_add(dropped as u64, Ordering::AcqRel)
2431 + dropped as u64;
2432 if total == dropped as u64 {
2434 self.push_alert(
2435 crate::toast::AlertLevel::Warning,
2436 format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
2437 );
2438 }
2439 }
2440 self.trade_dirty.store(true, Ordering::Release);
2441 self.bump();
2442 }
2443
2444 pub fn note_position(&self, p: &ek_data::position::PositionData) {
2449 let key = p.en_positionid();
2450 {
2451 let mut book = self.trade_positions.lock().unwrap();
2452 match book.get_mut(&key) {
2453 Some((_, cur)) => {
2454 cur.volume = p.volume;
2455 cur.yd_volume = p.yd_volume;
2456 cur.can_use_volume = p.can_use_volume;
2457 cur.frozen = p.frozen;
2458 cur.on_road_volume = p.on_road_volume;
2459 cur.market_value = p.market_value;
2460 cur.avg_price = p.avg_price;
2461 cur.price = p.price;
2462 cur.last_price = p.last_price;
2463 cur.pnl = p.pnl;
2464 cur.profit_rate = p.profit_rate;
2465 cur.name.clone_from(&p.name);
2466 }
2467 None => {
2468 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2469 book.insert(key, (seq, p.clone()));
2470 }
2471 }
2472 }
2473 self.trade_dirty.store(true, Ordering::Release);
2474 self.bump();
2475 }
2476
2477 pub fn note_funds(&self, a: &ek_data::account::AccountData) {
2479 let key = a.en_accountid();
2480 {
2481 let mut book = self.trade_funds.lock().unwrap();
2482 match book.get_mut(&key) {
2483 Some((_, cur)) => {
2484 cur.balance = a.balance;
2485 cur.frozen = a.frozen;
2486 }
2487 None => {
2488 let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
2489 book.insert(key, (seq, a.clone()));
2490 }
2491 }
2492 }
2493 self.trade_dirty.store(true, Ordering::Release);
2494 self.bump();
2495 }
2496
2497 pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
2499 let mut removed = 0usize;
2500 {
2501 let mut book = self.trade_orders.lock().unwrap();
2502 let before = book.len();
2503 book.retain(|_, (_, o)| o.gateway_name != gateway);
2504 removed += before - book.len();
2505 }
2506 {
2507 let mut book = self.trade_trades.lock().unwrap();
2508 let before = book.len();
2509 book.retain(|(_, t)| t.gateway_name != gateway);
2510 removed += before - book.len();
2511 let prefix = format!("{gateway}.");
2514 self.trade_ids
2515 .lock()
2516 .unwrap()
2517 .retain(|id| !id.starts_with(&prefix));
2518 }
2519 {
2520 let mut book = self.trade_positions.lock().unwrap();
2521 let before = book.len();
2522 book.retain(|_, (_, p)| p.gateway_name != gateway);
2523 removed += before - book.len();
2524 }
2525 {
2526 let mut book = self.trade_funds.lock().unwrap();
2527 let before = book.len();
2528 book.retain(|_, (_, a)| a.gateway_name != gateway);
2529 removed += before - book.len();
2530 }
2531 if removed > 0 {
2532 self.trade_dirty.store(true, Ordering::Release);
2533 self.bump();
2534 }
2535 removed
2536 }
2537
2538 #[must_use]
2540 pub fn trade_dropped(&self) -> u64 {
2541 self.trade_dropped.load(Ordering::Acquire)
2542 }
2543
2544 pub fn trade_snap(&self) -> Arc<TradeSnap> {
2549 let mut cache = self.trade_snap.lock().unwrap();
2550 if self.trade_dirty.swap(false, Ordering::AcqRel) {
2551 *cache = Arc::new(self.build_trade_snap());
2552 }
2553 Arc::clone(&cache)
2554 }
2555
2556 fn build_trade_snap(&self) -> TradeSnap {
2559 fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
2560 where
2561 T: Clone,
2562 {
2563 let mut v: Vec<(u64, &String, &T)> =
2564 rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
2565 v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
2566 v.into_iter().map(|(_, _, v)| v.clone()).collect()
2567 }
2568 TradeSnap {
2569 orders: by_seq(&self.trade_orders.lock().unwrap()),
2570 trades: self
2571 .trade_trades
2572 .lock()
2573 .unwrap()
2574 .iter()
2575 .map(|(_, t)| t.clone())
2576 .collect(),
2577 positions: by_seq(&self.trade_positions.lock().unwrap()),
2578 accounts: by_seq(&self.trade_funds.lock().unwrap()),
2579 }
2580 }
2581
2582 pub fn ctp_form(&self) -> CtpForm {
2586 self.ctp_form.lock().unwrap().clone()
2587 }
2588
2589 pub fn set_ctp_form(&self, f: CtpForm) {
2591 *self.ctp_form.lock().unwrap() = f;
2592 self.bump();
2593 }
2594
2595 pub fn row_selected(&self, table: &str, key: &str) -> bool {
2599 self.row_sel
2600 .lock()
2601 .unwrap()
2602 .get(table)
2603 .is_some_and(|k| k == key)
2604 }
2605
2606 pub fn select_row(&self, table: &str, key: &str) {
2608 self.row_sel
2609 .lock()
2610 .unwrap()
2611 .insert(table.to_string(), key.to_string());
2612 }
2613}