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