1use std::collections::{BTreeMap, VecDeque};
7use std::sync::atomic::{AtomicBool, 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 },
94 Remove { name: String },
96 Reload {
98 interval: i64,
99 symbol: String,
100 bars: Vec<BarData>,
101 },
102 SendOrder {
108 req: ek_data::order_request::OrderRequest,
109 gateway: String,
110 mode: ek_oms::LockMode,
111 },
112 CancelOrder { orderid: String, gateway: String },
116 QueryRate,
118 CtpConfigure(CtpForm),
121 CtpSubscribe { symbols: Vec<String> },
123}
124
125pub const KLINE_CELL_CAP: usize = 4096;
127
128#[derive(Debug, Clone, Copy, PartialEq, Eq)]
130pub enum ConfirmStep {
131 Armed,
132 Fired,
133 Switched,
134}
135
136#[must_use]
139pub fn is_quarantined(state: &str) -> bool {
140 state.contains("Quarantin")
141}
142
143#[derive(Debug, Default)]
147pub struct MarketStore {
148 cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
149 frames: AtomicU64,
151 updates: AtomicU64,
153 reloads: AtomicU64,
155}
156
157#[derive(Debug, Clone, PartialEq)]
159pub struct CellView {
160 pub symbol: String,
161 pub interval: i64,
162 pub bars: Vec<BarData>,
163}
164
165impl MarketStore {
166 pub fn on_frame(&self, frame: &BarsFrame) {
168 let mut cells = self.cells.lock().unwrap();
169 for b in &frame.bars {
170 let key = (b.symbol.clone(), frame.interval);
171 let cell = cells.entry(key).or_default();
172 match cell.last() {
173 Some(last) if b.datetime < last.datetime => continue,
174 Some(last) if b.datetime == last.datetime => {
175 let slot = cell.last_mut().expect("len>1");
176 *slot = b.clone();
177 }
178 _ => {
179 cell.push(b.clone());
180 if cell.len() > KLINE_CELL_CAP {
181 cell.remove(0);
182 }
183 }
184 }
185 }
186 self.frames.fetch_add(1, Ordering::AcqRel);
187 }
188
189 pub fn on_update(&self, b: &BarData) {
191 let mut cells = self.cells.lock().unwrap();
192 let iv = b
193 .interval
194 .unwrap_or(ek_data::constant::Interval::MINUTE)
195 .seconds();
196 let key = (b.symbol.clone(), iv);
197 let cell = cells.entry(key).or_default();
198 match cell.last() {
199 Some(last) if b.datetime == last.datetime => {
200 let slot = cell.last_mut().expect("len>1");
201 *slot = b.clone();
202 }
203 Some(last) if b.datetime < last.datetime => {}
204 _ => {
205 cell.push(b.clone());
206 if cell.len() > KLINE_CELL_CAP {
207 cell.remove(0);
208 }
209 }
210 }
211 self.updates.fetch_add(1, Ordering::AcqRel);
212 }
213
214 pub fn note_reload(&self) {
216 self.reloads.fetch_add(1, Ordering::AcqRel);
217 }
218
219 pub fn cells(&self) -> Vec<CellView> {
221 self.cells
222 .lock()
223 .unwrap()
224 .iter()
225 .map(|((sym, iv), bars)| CellView {
226 symbol: sym.clone(),
227 interval: *iv,
228 bars: bars.clone(),
229 })
230 .collect()
231 }
232
233 pub fn counts(&self) -> (u64, u64, u64) {
235 (
236 self.frames.load(Ordering::Acquire),
237 self.updates.load(Ordering::Acquire),
238 self.reloads.load(Ordering::Acquire),
239 )
240 }
241}
242
243#[derive(Debug, Default, Clone, PartialEq)]
246pub struct ReloadForm {
247 pub interval: String,
248 pub symbol: String,
249 pub csv: String,
250}
251
252pub fn parse_gateway(raw: &str) -> Result<String, String> {
254 let g = raw.trim();
255 if g.is_empty() {
256 return Ok("sim".to_string());
257 }
258 if g.chars().any(|c| c.is_whitespace()) {
259 return Err(format!("网关解析: 含空白:{raw:?}"));
260 }
261 Ok(g.to_string())
262}
263
264#[derive(Debug, Default, Clone, PartialEq)]
267pub struct CtpForm {
268 pub broker_id: String,
269 pub td_addr: String,
270 pub td_user: String,
271 pub td_password: String,
272 pub td_auth_code: String,
273 pub td_app_id: String,
274 pub md_addr: String,
275 pub md_user: String,
276 pub md_password: String,
277 pub symbols: String,
279}
280
281impl CtpForm {
282 pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
284 ek_ctp::plugin::ConfigureReq {
285 broker_id: self.broker_id.trim().to_string(),
286 td_addr: self.td_addr.trim().to_string(),
287 td_user: self.td_user.trim().to_string(),
288 td_password: self.td_password.clone(),
289 td_auth_code: self.td_auth_code.trim().to_string(),
290 td_app_id: self.td_app_id.trim().to_string(),
291 md_addr: self.md_addr.trim().to_string(),
292 md_user: self.md_user.trim().to_string(),
293 md_password: self.md_password.clone(),
294 }
295 }
296
297 pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
299 let list: Vec<String> = self
300 .symbols
301 .split([',', ',', ' '])
302 .map(|s| s.trim().to_string())
303 .filter(|s| !s.is_empty())
304 .collect();
305 if list.is_empty() {
306 return Err("订阅解析: 标的位空".to_string());
307 }
308 Ok(list)
309 }
310}
311
312#[derive(Debug, Clone, PartialEq, Default)]
314pub struct TradeSnap {
315 pub orders: Vec<ek_data::order::OrderData>,
316 pub trades: Vec<ek_data::trade::TradeData>,
317 pub positions: Vec<ek_data::position::PositionData>,
318 pub accounts: Vec<ek_data::account::AccountData>,
320}
321
322pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
325 let interval: i64 = form
326 .interval
327 .trim()
328 .parse()
329 .map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
330 let symbol = form.symbol.trim().to_string();
331 if symbol.is_empty() {
332 return Err("回灌解析: symbol 空".to_string());
333 }
334 if form.csv.trim().is_empty() {
335 return Err("回灌解析: csv 空".to_string());
336 }
337 let mut bars = Vec::new();
338 for (i, raw) in form.csv.lines().enumerate() {
339 let line = raw.trim();
340 if line.is_empty() || line.starts_with('#') {
341 continue;
342 }
343 let parts: Vec<&str> = line.split(',').map(str::trim).collect();
344 if parts.len() != 6 {
345 return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
346 }
347 let num = |s: &str| -> Result<f64, String> {
348 s.parse::<f64>()
349 .map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
350 };
351 let dt_s: i64 = parts[0]
352 .parse()
353 .map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
354 bars.push(BarData {
355 symbol: symbol.clone(),
356 datetime: dt_s * 1_000_000_000,
357 interval: Some(
358 ek_dm::wire::interval_of_seconds(interval)
359 .unwrap_or(ek_data::constant::Interval::MINUTE),
360 ),
361 open_price: num(parts[1])?,
362 high_price: num(parts[2])?,
363 low_price: num(parts[3])?,
364 close_price: num(parts[4])?,
365 volume: num(parts[5])?,
366 ..BarData::default()
367 });
368 }
369 if bars.is_empty() {
370 return Err("回灌解析: 无有效行".to_string());
371 }
372 Ok((interval, symbol, bars))
373}
374
375#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
380pub enum TopLeftTab {
381 #[default]
382 Orders,
383 Active,
384}
385
386impl TopLeftTab {
387 pub fn label(self) -> &'static str {
388 match self {
389 Self::Orders => "委托",
390 Self::Active => "未成交",
391 }
392 }
393}
394
395#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
397pub enum BottomRightTab {
398 #[default]
399 Account,
400 Logs,
401}
402
403impl BottomRightTab {
404 pub fn label(self) -> &'static str {
405 match self {
406 Self::Account => "资金",
407 Self::Logs => "日志",
408 }
409 }
410}
411
412#[derive(Debug, Default)]
415pub struct TickStore {
416 ticks: BTreeMap<String, ek_data::tick::TickData>,
417}
418
419impl TickStore {
420 pub fn put(&mut self, t: &ek_data::tick::TickData) {
423 self.ticks
424 .insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
425 }
426
427 pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
429 let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
430 v.sort_by(|a, b| {
431 (a.exchange.as_str().to_string(), a.symbol.clone())
432 .cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
433 });
434 v
435 }
436
437 pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
439 let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
440 for t in self.rows() {
441 let ex = t.exchange.as_str().to_string();
442 match out.last_mut() {
443 Some((e, v)) if *e == ex => v.push(t),
444 _ => out.push((ex, vec![t])),
445 }
446 }
447 out
448 }
449
450 pub fn len(&self) -> usize {
451 self.ticks.len()
452 }
453
454 pub fn is_empty(&self) -> bool {
455 self.ticks.is_empty()
456 }
457}
458
459#[derive(Debug, Clone, PartialEq)]
461pub struct OrderForm {
462 pub exchange: String,
464 pub symbol: String,
466 pub direction: ek_data::constant::Direction,
467 pub offset: ek_data::constant::Offset,
468 pub order_type: ek_data::constant::OrderType,
469 pub price: String,
471 pub volume: String,
472 pub gateway: String,
474 pub follow_price: bool,
476 pub auto_split: bool,
479}
480
481impl Default for OrderForm {
482 fn default() -> Self {
483 Self {
484 exchange: "SHFE".to_string(),
485 symbol: String::new(),
486 direction: ek_data::constant::Direction::LONG,
487 offset: ek_data::constant::Offset::OPEN,
488 order_type: ek_data::constant::OrderType::LIMIT,
489 price: String::new(),
490 volume: "1".to_string(),
491 gateway: String::new(),
492 follow_price: true,
493 auto_split: true,
494 }
495 }
496}
497
498impl OrderForm {
499 pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
501 use std::str::FromStr;
502 let symbol = self.symbol.trim();
503 if symbol.is_empty() {
504 return Err("下单: 代码为空".to_string());
505 }
506 if symbol.contains('.') {
507 return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
508 }
509 let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
510 .map_err(|e| format!("下单: {e}"))?;
511 let price: f64 = self
512 .price
513 .trim()
514 .parse()
515 .map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
516 let volume: f64 = self
517 .volume
518 .trim()
519 .parse()
520 .map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
521 if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
522 return Err("下单: 限价单价格须为正".to_string());
523 }
524 if volume <= 0.0 {
525 return Err("下单: 数量须为正".to_string());
526 }
527 Ok(ek_data::order_request::OrderRequest {
528 symbol: symbol.to_string(),
529 exchange,
530 direction: self.direction,
531 order_type: self.order_type,
532 volume,
533 price,
534 offset: self.offset,
535 ..ek_data::order_request::OrderRequest::default()
536 })
537 }
538}
539
540#[derive(Debug, Default)]
546pub struct ContractStore {
547 contracts: BTreeMap<String, ek_data::contract::ContractData>,
548}
549
550impl ContractStore {
551 pub fn put(&mut self, c: &ek_data::contract::ContractData) {
552 self.contracts
553 .insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
554 }
555
556 pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
558 let p = pattern.trim();
559 self.contracts
560 .values()
561 .filter(|c| {
562 p.is_empty()
563 || c.symbol.contains(p)
564 || c.exchange.as_str().eq_ignore_ascii_case(p)
565 || c.name.contains(p)
566 })
567 .cloned()
568 .collect()
569 }
570
571 pub fn len(&self) -> usize {
572 self.contracts.len()
573 }
574
575 pub fn is_empty(&self) -> bool {
576 self.contracts.is_empty()
577 }
578}
579
580#[derive(Debug, Default)]
582pub struct UiShared {
583 logs: Mutex<LogRing>,
584 snap: Mutex<Arc<BusSnap>>,
585 market: MarketStore,
586 ctp_form: Mutex<CtpForm>,
587 trade_snap: Mutex<Arc<TradeSnap>>,
588 tick_store: Mutex<TickStore>,
589 order_form: Mutex<OrderForm>,
590 selected_exchange: Mutex<String>,
591 topleft_tab: Mutex<TopLeftTab>,
592 bottomright_tab: Mutex<BottomRightTab>,
593 ops_window: AtomicBool,
594 ctp_window: AtomicBool,
595 contracts_window: AtomicBool,
596 contract_store: Mutex<ContractStore>,
597 contract_query: Mutex<String>,
598 contract_selected: Mutex<Vec<String>>,
599 kline_windows: Mutex<Vec<String>>,
600 rev: AtomicU64,
601 stop: AtomicBool,
603 cmds: Mutex<VecDeque<UiCmd>>,
605 pending_confirm: Mutex<Option<String>>,
607}
608
609impl UiShared {
610 pub fn new() -> Self {
612 Self::default()
613 }
614
615 pub fn update_tick(&self, t: &ek_data::tick::TickData) {
619 self.tick_store.lock().unwrap().put(t);
620 self.bump();
621 }
622
623 pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
625 self.tick_store.lock().unwrap().rows()
626 }
627
628 pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
630 self.tick_store.lock().unwrap().groups()
631 }
632
633 pub fn tick_count(&self) -> usize {
635 self.tick_store.lock().unwrap().len()
636 }
637
638 pub fn order_form(&self) -> OrderForm {
640 self.order_form.lock().unwrap().clone()
641 }
642
643 pub fn set_order_form(&self, f: OrderForm) {
645 *self.order_form.lock().unwrap() = f;
646 self.bump();
647 }
648
649 pub fn selected_exchange(&self) -> String {
651 self.selected_exchange.lock().unwrap().clone()
652 }
653
654 pub fn set_selected_exchange(&self, s: impl Into<String>) {
656 *self.selected_exchange.lock().unwrap() = s.into();
657 self.bump();
658 }
659
660 pub fn topleft_tab(&self) -> TopLeftTab {
662 *self.topleft_tab.lock().unwrap()
663 }
664
665 pub fn set_topleft_tab(&self, t: TopLeftTab) {
667 *self.topleft_tab.lock().unwrap() = t;
668 self.bump();
669 }
670
671 pub fn bottomright_tab(&self) -> BottomRightTab {
673 *self.bottomright_tab.lock().unwrap()
674 }
675
676 pub fn set_bottomright_tab(&self, t: BottomRightTab) {
678 *self.bottomright_tab.lock().unwrap() = t;
679 self.bump();
680 }
681
682 pub fn ops_window_open(&self) -> bool {
686 self.ops_window.load(Ordering::Acquire)
687 }
688
689 pub fn set_ops_window_open(&self, v: bool) {
691 self.ops_window.store(v, Ordering::Release);
692 self.bump();
693 }
694
695 pub fn ctp_window_open(&self) -> bool {
697 self.ctp_window.load(Ordering::Acquire)
698 }
699
700 pub fn set_ctp_window_open(&self, v: bool) {
702 self.ctp_window.store(v, Ordering::Release);
703 self.bump();
704 }
705
706 pub fn kline_symbols(&self) -> Vec<String> {
708 self.kline_windows.lock().unwrap().clone()
709 }
710
711 pub fn open_kline(&self, symbol: impl Into<String>) {
713 let s = symbol.into();
714 let mut v = self.kline_windows.lock().unwrap();
715 if !v.contains(&s) {
716 v.push(s.clone());
717 }
718 drop(v);
719 self.log_line(format!("[kline] 打开 K线窗 {s}"));
720 }
721
722 pub fn close_kline(&self, symbol: &str) {
724 self.kline_windows.lock().unwrap().retain(|s| s != symbol);
725 self.bump();
726 }
727
728 pub fn contracts_window_open(&self) -> bool {
732 self.contracts_window.load(Ordering::Acquire)
733 }
734
735 pub fn set_contracts_window_open(&self, v: bool) {
737 self.contracts_window.store(v, Ordering::Release);
738 self.bump();
739 }
740
741 pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
743 self.contract_store.lock().unwrap().put(c);
744 self.bump();
745 }
746
747 pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
749 self.contract_store.lock().unwrap().query(pattern)
750 }
751
752 pub fn contract_count(&self) -> usize {
754 self.contract_store.lock().unwrap().len()
755 }
756
757 pub fn contract_query(&self) -> String {
759 self.contract_query.lock().unwrap().clone()
760 }
761
762 pub fn set_contract_query(&self, q: impl Into<String>) {
764 *self.contract_query.lock().unwrap() = q.into();
765 self.bump();
766 }
767
768 pub fn toggle_contract_selection(&self, en_symbol: &str) {
770 let mut v = self.contract_selected.lock().unwrap();
771 match v.iter().position(|s| s == en_symbol) {
772 Some(i) => {
773 v.remove(i);
774 }
775 None => v.push(en_symbol.to_string()),
776 }
777 self.bump();
778 }
779
780 pub fn selected_contracts(&self) -> Vec<String> {
782 self.contract_selected.lock().unwrap().clone()
783 }
784
785 pub fn log_line(&self, line: impl Into<String>) {
787 self.logs.lock().unwrap().push(line);
788 self.bump();
789 }
790
791 pub fn clear_logs(&self) {
793 self.logs.lock().unwrap().clear();
794 self.bump();
795 }
796
797 pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
799 let r = self.logs.lock().unwrap();
800 (r.snapshot(), r.spilled())
801 }
802
803 pub fn logs_len(&self) -> usize {
805 self.logs.lock().unwrap().len()
806 }
807
808 pub fn set_snapshot(&self, snap: BusSnap) {
810 *self.snap.lock().unwrap() = Arc::new(snap);
811 self.bump();
812 }
813
814 pub fn snapshot(&self) -> Arc<BusSnap> {
816 Arc::clone(&self.snap.lock().unwrap())
817 }
818
819 pub fn rev(&self) -> u64 {
821 self.rev.load(Ordering::Acquire)
822 }
823
824 pub fn bump(&self) {
826 self.rev.fetch_add(1, Ordering::AcqRel);
827 }
828
829 pub fn request_stop(&self) {
831 self.stop.store(true, Ordering::Release);
832 }
833
834 pub fn is_stop(&self) -> bool {
836 self.stop.load(Ordering::Acquire)
837 }
838
839 pub fn push_cmd(&self, cmd: UiCmd) {
843 self.cmds.lock().unwrap().push_back(cmd);
844 self.bump();
845 }
846
847 pub fn drain_cmds(&self) -> Vec<UiCmd> {
849 self.cmds.lock().unwrap().drain(..).collect()
850 }
851
852 pub fn pending_confirm(&self) -> Option<String> {
854 self.pending_confirm.lock().unwrap().clone()
855 }
856
857 pub fn confirm_step(&self, name: &str) -> ConfirmStep {
859 let mut g = self.pending_confirm.lock().unwrap();
860 match g.take() {
861 Some(p) if p == name => ConfirmStep::Fired,
862 Some(_) => {
863 *g = Some(name.to_string());
864 ConfirmStep::Switched
865 }
866 None => {
867 *g = Some(name.to_string());
868 ConfirmStep::Armed
869 }
870 }
871 }
872
873 pub fn confirm_cancel(&self) {
875 *self.pending_confirm.lock().unwrap() = None;
876 self.bump();
877 }
878
879 pub fn market(&self) -> &MarketStore {
883 &self.market
884 }
885
886 pub fn set_trade_snap(&self, ts: TradeSnap) {
890 *self.trade_snap.lock().unwrap() = Arc::new(ts);
891 self.bump();
892 }
893
894 pub fn trade_snap(&self) -> Arc<TradeSnap> {
896 self.trade_snap.lock().unwrap().clone()
897 }
898
899 pub fn ctp_form(&self) -> CtpForm {
903 self.ctp_form.lock().unwrap().clone()
904 }
905
906 pub fn set_ctp_form(&self, f: CtpForm) {
908 *self.ctp_form.lock().unwrap() = f;
909 self.bump();
910 }
911}