use std::collections::{BTreeMap, VecDeque};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use ek_data::bar::BarData;
use ek_dm::wire::BarsFrame;
use ek_station::wire::PlugRow;
pub const RING_CAP: usize = 512;
#[derive(Debug)]
pub struct LogRing {
lines: VecDeque<String>,
cap: usize,
spilled: u64,
}
impl Default for LogRing {
fn default() -> Self {
Self::new(RING_CAP)
}
}
impl LogRing {
pub fn new(cap: usize) -> Self {
Self {
lines: VecDeque::with_capacity(cap),
cap,
spilled: 0,
}
}
pub fn push(&mut self, line: impl Into<String>) {
if self.cap == 0 {
self.spilled += 1;
return;
}
if self.lines.len() == self.cap {
self.lines.pop_front();
self.spilled += 1;
}
self.lines.push_back(line.into());
}
pub fn len(&self) -> usize {
self.lines.len()
}
pub fn is_empty(&self) -> bool {
self.lines.is_empty()
}
pub fn spilled(&self) -> u64 {
self.spilled
}
pub fn snapshot(&self) -> Vec<String> {
self.lines.iter().cloned().collect()
}
pub fn clear(&mut self) {
self.lines.clear();
}
}
#[derive(Debug, Clone, Default)]
pub struct BusSnap {
pub rows: Vec<PlugRow>,
pub services: Vec<(String, Vec<String>)>,
}
#[derive(Debug, Clone, PartialEq)]
pub enum UiCmd {
Start { factory: String },
Remove { name: String },
Reload {
interval: i64,
symbol: String,
bars: Vec<BarData>,
},
SendOrder {
req: ek_data::order_request::OrderRequest,
gateway: String,
mode: ek_oms::LockMode,
},
CancelOrder { orderid: String, gateway: String },
QueryRate,
CtpConfigure(CtpForm),
CtpSubscribe { symbols: Vec<String> },
}
pub const KLINE_CELL_CAP: usize = 4096;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConfirmStep {
Armed,
Fired,
Switched,
}
#[must_use]
pub fn is_quarantined(state: &str) -> bool {
state.contains("Quarantin")
}
#[derive(Debug, Default)]
pub struct MarketStore {
cells: Mutex<BTreeMap<(String, i64), Vec<BarData>>>,
frames: AtomicU64,
updates: AtomicU64,
reloads: AtomicU64,
}
#[derive(Debug, Clone, PartialEq)]
pub struct CellView {
pub symbol: String,
pub interval: i64,
pub bars: Vec<BarData>,
}
impl MarketStore {
pub fn on_frame(&self, frame: &BarsFrame) {
let mut cells = self.cells.lock().unwrap();
for b in &frame.bars {
let key = (b.symbol.clone(), frame.interval);
let cell = cells.entry(key).or_default();
match cell.last() {
Some(last) if b.datetime < last.datetime => continue,
Some(last) if b.datetime == last.datetime => {
let slot = cell.last_mut().expect("len>1");
*slot = b.clone();
}
_ => {
cell.push(b.clone());
if cell.len() > KLINE_CELL_CAP {
cell.remove(0);
}
}
}
}
self.frames.fetch_add(1, Ordering::AcqRel);
}
pub fn on_update(&self, b: &BarData) {
let mut cells = self.cells.lock().unwrap();
let iv = b
.interval
.unwrap_or(ek_data::constant::Interval::MINUTE)
.seconds();
let key = (b.symbol.clone(), iv);
let cell = cells.entry(key).or_default();
match cell.last() {
Some(last) if b.datetime == last.datetime => {
let slot = cell.last_mut().expect("len>1");
*slot = b.clone();
}
Some(last) if b.datetime < last.datetime => {}
_ => {
cell.push(b.clone());
if cell.len() > KLINE_CELL_CAP {
cell.remove(0);
}
}
}
self.updates.fetch_add(1, Ordering::AcqRel);
}
pub fn note_reload(&self) {
self.reloads.fetch_add(1, Ordering::AcqRel);
}
pub fn cells(&self) -> Vec<CellView> {
self.cells
.lock()
.unwrap()
.iter()
.map(|((sym, iv), bars)| CellView {
symbol: sym.clone(),
interval: *iv,
bars: bars.clone(),
})
.collect()
}
pub fn counts(&self) -> (u64, u64, u64) {
(
self.frames.load(Ordering::Acquire),
self.updates.load(Ordering::Acquire),
self.reloads.load(Ordering::Acquire),
)
}
}
#[derive(Debug, Default, Clone, PartialEq)]
pub struct ReloadForm {
pub interval: String,
pub symbol: String,
pub csv: String,
}
pub fn parse_gateway(raw: &str) -> Result<String, String> {
let g = raw.trim();
if g.is_empty() {
return Ok("sim".to_string());
}
if g.chars().any(|c| c.is_whitespace()) {
return Err(format!("网关解析: 含空白:{raw:?}"));
}
Ok(g.to_string())
}
#[derive(Debug, Default, Clone, PartialEq)]
pub struct CtpForm {
pub broker_id: String,
pub td_addr: String,
pub td_user: String,
pub td_password: String,
pub td_auth_code: String,
pub td_app_id: String,
pub md_addr: String,
pub md_user: String,
pub md_password: String,
pub symbols: String,
}
impl CtpForm {
pub fn to_configure_req(&self) -> ek_ctp::plugin::ConfigureReq {
ek_ctp::plugin::ConfigureReq {
broker_id: self.broker_id.trim().to_string(),
td_addr: self.td_addr.trim().to_string(),
td_user: self.td_user.trim().to_string(),
td_password: self.td_password.clone(),
td_auth_code: self.td_auth_code.trim().to_string(),
td_app_id: self.td_app_id.trim().to_string(),
md_addr: self.md_addr.trim().to_string(),
md_user: self.md_user.trim().to_string(),
md_password: self.md_password.clone(),
}
}
pub fn parse_symbols(&self) -> Result<Vec<String>, String> {
let list: Vec<String> = self
.symbols
.split([',', ',', ' '])
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect();
if list.is_empty() {
return Err("订阅解析: 标的位空".to_string());
}
Ok(list)
}
}
#[derive(Debug, Clone, PartialEq, Default)]
pub struct TradeSnap {
pub orders: Vec<ek_data::order::OrderData>,
pub trades: Vec<ek_data::trade::TradeData>,
pub positions: Vec<ek_data::position::PositionData>,
pub accounts: Vec<ek_data::account::AccountData>,
}
pub fn parse_reload_csv(form: &ReloadForm) -> Result<(i64, String, Vec<BarData>), String> {
let interval: i64 = form
.interval
.trim()
.parse()
.map_err(|_| format!("回灌 解析: interval 非整数:{:?}", form.interval))?;
let symbol = form.symbol.trim().to_string();
if symbol.is_empty() {
return Err("回灌解析: symbol 空".to_string());
}
if form.csv.trim().is_empty() {
return Err("回灌解析: csv 空".to_string());
}
let mut bars = Vec::new();
for (i, raw) in form.csv.lines().enumerate() {
let line = raw.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
let parts: Vec<&str> = line.split(',').map(str::trim).collect();
if parts.len() != 6 {
return Err(format!("回灌解析: 第 {} 列数 ≠ 6", i + 1));
}
let num = |s: &str| -> Result<f64, String> {
s.parse::<f64>()
.map_err(|_| format!("回灌解析: 第 {} 列数值坏:{:?}", i + 1, s))
};
let dt_s: i64 = parts[0]
.parse()
.map_err(|_| format!("回灌解析: 第 {} 列 dt 坏:{:?}", i + 1, parts[0]))?;
bars.push(BarData {
symbol: symbol.clone(),
datetime: dt_s * 1_000_000_000,
interval: Some(
ek_dm::wire::interval_of_seconds(interval)
.unwrap_or(ek_data::constant::Interval::MINUTE),
),
open_price: num(parts[1])?,
high_price: num(parts[2])?,
low_price: num(parts[3])?,
close_price: num(parts[4])?,
volume: num(parts[5])?,
..BarData::default()
});
}
if bars.is_empty() {
return Err("回灌解析: 无有效行".to_string());
}
Ok((interval, symbol, bars))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum TopLeftTab {
#[default]
Orders,
Active,
}
impl TopLeftTab {
pub fn label(self) -> &'static str {
match self {
Self::Orders => "委托",
Self::Active => "未成交",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum BottomRightTab {
#[default]
Account,
Logs,
}
impl BottomRightTab {
pub fn label(self) -> &'static str {
match self {
Self::Account => "资金",
Self::Logs => "日志",
}
}
}
#[derive(Debug, Default)]
pub struct TickStore {
ticks: BTreeMap<String, ek_data::tick::TickData>,
}
impl TickStore {
pub fn put(&mut self, t: &ek_data::tick::TickData) {
self.ticks
.insert(format!("{}.{}", t.exchange.as_str(), t.symbol), t.clone());
}
pub fn rows(&self) -> Vec<ek_data::tick::TickData> {
let mut v: Vec<ek_data::tick::TickData> = self.ticks.values().cloned().collect();
v.sort_by(|a, b| {
(a.exchange.as_str().to_string(), a.symbol.clone())
.cmp(&(b.exchange.as_str().to_string(), b.symbol.clone()))
});
v
}
pub fn groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
let mut out: Vec<(String, Vec<ek_data::tick::TickData>)> = Vec::new();
for t in self.rows() {
let ex = t.exchange.as_str().to_string();
match out.last_mut() {
Some((e, v)) if *e == ex => v.push(t),
_ => out.push((ex, vec![t])),
}
}
out
}
pub fn len(&self) -> usize {
self.ticks.len()
}
pub fn is_empty(&self) -> bool {
self.ticks.is_empty()
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct OrderForm {
pub exchange: String,
pub symbol: String,
pub direction: ek_data::constant::Direction,
pub offset: ek_data::constant::Offset,
pub order_type: ek_data::constant::OrderType,
pub price: String,
pub volume: String,
pub gateway: String,
pub follow_price: bool,
pub auto_split: bool,
}
impl Default for OrderForm {
fn default() -> Self {
Self {
exchange: "SHFE".to_string(),
symbol: String::new(),
direction: ek_data::constant::Direction::LONG,
offset: ek_data::constant::Offset::OPEN,
order_type: ek_data::constant::OrderType::LIMIT,
price: String::new(),
volume: "1".to_string(),
gateway: String::new(),
follow_price: true,
auto_split: true,
}
}
}
impl OrderForm {
pub fn to_request(&self) -> Result<ek_data::order_request::OrderRequest, String> {
use std::str::FromStr;
let symbol = self.symbol.trim();
if symbol.is_empty() {
return Err("下单: 代码为空".to_string());
}
if symbol.contains('.') {
return Err("下单: 代码不带交易所前缀(交易所走下拉)".to_string());
}
let exchange = ek_data::constant::Exchange::from_str(self.exchange.trim())
.map_err(|e| format!("下单: {e}"))?;
let price: f64 = self
.price
.trim()
.parse()
.map_err(|_| format!("下单: 价格无效 {:?}", self.price))?;
let volume: f64 = self
.volume
.trim()
.parse()
.map_err(|_| format!("下单: 数量无效 {:?}", self.volume))?;
if self.order_type != ek_data::constant::OrderType::MARKET && price <= 0.0 {
return Err("下单: 限价单价格须为正".to_string());
}
if volume <= 0.0 {
return Err("下单: 数量须为正".to_string());
}
Ok(ek_data::order_request::OrderRequest {
symbol: symbol.to_string(),
exchange,
direction: self.direction,
order_type: self.order_type,
volume,
price,
offset: self.offset,
..ek_data::order_request::OrderRequest::default()
})
}
}
#[derive(Debug, Default)]
pub struct ContractStore {
contracts: BTreeMap<String, ek_data::contract::ContractData>,
}
impl ContractStore {
pub fn put(&mut self, c: &ek_data::contract::ContractData) {
self.contracts
.insert(format!("{}.{}", c.exchange.as_str(), c.symbol), c.clone());
}
pub fn query(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
let p = pattern.trim();
self.contracts
.values()
.filter(|c| {
p.is_empty()
|| c.symbol.contains(p)
|| c.exchange.as_str().eq_ignore_ascii_case(p)
|| c.name.contains(p)
})
.cloned()
.collect()
}
pub fn len(&self) -> usize {
self.contracts.len()
}
pub fn is_empty(&self) -> bool {
self.contracts.is_empty()
}
}
#[derive(Debug, Default)]
pub struct UiShared {
logs: Mutex<LogRing>,
snap: Mutex<Arc<BusSnap>>,
market: MarketStore,
ctp_form: Mutex<CtpForm>,
trade_snap: Mutex<Arc<TradeSnap>>,
tick_store: Mutex<TickStore>,
order_form: Mutex<OrderForm>,
selected_exchange: Mutex<String>,
topleft_tab: Mutex<TopLeftTab>,
bottomright_tab: Mutex<BottomRightTab>,
ops_window: AtomicBool,
ctp_window: AtomicBool,
contracts_window: AtomicBool,
contract_store: Mutex<ContractStore>,
contract_query: Mutex<String>,
contract_selected: Mutex<Vec<String>>,
kline_windows: Mutex<Vec<String>>,
rev: AtomicU64,
stop: AtomicBool,
cmds: Mutex<VecDeque<UiCmd>>,
pending_confirm: Mutex<Option<String>>,
}
impl UiShared {
pub fn new() -> Self {
Self::default()
}
pub fn update_tick(&self, t: &ek_data::tick::TickData) {
self.tick_store.lock().unwrap().put(t);
self.bump();
}
pub fn tick_rows(&self) -> Vec<ek_data::tick::TickData> {
self.tick_store.lock().unwrap().rows()
}
pub fn tick_groups(&self) -> Vec<(String, Vec<ek_data::tick::TickData>)> {
self.tick_store.lock().unwrap().groups()
}
pub fn tick_count(&self) -> usize {
self.tick_store.lock().unwrap().len()
}
pub fn order_form(&self) -> OrderForm {
self.order_form.lock().unwrap().clone()
}
pub fn set_order_form(&self, f: OrderForm) {
*self.order_form.lock().unwrap() = f;
self.bump();
}
pub fn selected_exchange(&self) -> String {
self.selected_exchange.lock().unwrap().clone()
}
pub fn set_selected_exchange(&self, s: impl Into<String>) {
*self.selected_exchange.lock().unwrap() = s.into();
self.bump();
}
pub fn topleft_tab(&self) -> TopLeftTab {
*self.topleft_tab.lock().unwrap()
}
pub fn set_topleft_tab(&self, t: TopLeftTab) {
*self.topleft_tab.lock().unwrap() = t;
self.bump();
}
pub fn bottomright_tab(&self) -> BottomRightTab {
*self.bottomright_tab.lock().unwrap()
}
pub fn set_bottomright_tab(&self, t: BottomRightTab) {
*self.bottomright_tab.lock().unwrap() = t;
self.bump();
}
pub fn ops_window_open(&self) -> bool {
self.ops_window.load(Ordering::Acquire)
}
pub fn set_ops_window_open(&self, v: bool) {
self.ops_window.store(v, Ordering::Release);
self.bump();
}
pub fn ctp_window_open(&self) -> bool {
self.ctp_window.load(Ordering::Acquire)
}
pub fn set_ctp_window_open(&self, v: bool) {
self.ctp_window.store(v, Ordering::Release);
self.bump();
}
pub fn kline_symbols(&self) -> Vec<String> {
self.kline_windows.lock().unwrap().clone()
}
pub fn open_kline(&self, symbol: impl Into<String>) {
let s = symbol.into();
let mut v = self.kline_windows.lock().unwrap();
if !v.contains(&s) {
v.push(s.clone());
}
drop(v);
self.log_line(format!("[kline] 打开 K线窗 {s}"));
}
pub fn close_kline(&self, symbol: &str) {
self.kline_windows.lock().unwrap().retain(|s| s != symbol);
self.bump();
}
pub fn contracts_window_open(&self) -> bool {
self.contracts_window.load(Ordering::Acquire)
}
pub fn set_contracts_window_open(&self, v: bool) {
self.contracts_window.store(v, Ordering::Release);
self.bump();
}
pub fn update_contract(&self, c: &ek_data::contract::ContractData) {
self.contract_store.lock().unwrap().put(c);
self.bump();
}
pub fn contracts(&self, pattern: &str) -> Vec<ek_data::contract::ContractData> {
self.contract_store.lock().unwrap().query(pattern)
}
pub fn contract_count(&self) -> usize {
self.contract_store.lock().unwrap().len()
}
pub fn contract_query(&self) -> String {
self.contract_query.lock().unwrap().clone()
}
pub fn set_contract_query(&self, q: impl Into<String>) {
*self.contract_query.lock().unwrap() = q.into();
self.bump();
}
pub fn toggle_contract_selection(&self, en_symbol: &str) {
let mut v = self.contract_selected.lock().unwrap();
match v.iter().position(|s| s == en_symbol) {
Some(i) => {
v.remove(i);
}
None => v.push(en_symbol.to_string()),
}
self.bump();
}
pub fn selected_contracts(&self) -> Vec<String> {
self.contract_selected.lock().unwrap().clone()
}
pub fn log_line(&self, line: impl Into<String>) {
self.logs.lock().unwrap().push(line);
self.bump();
}
pub fn clear_logs(&self) {
self.logs.lock().unwrap().clear();
self.bump();
}
pub fn logs_snapshot(&self) -> (Vec<String>, u64) {
let r = self.logs.lock().unwrap();
(r.snapshot(), r.spilled())
}
pub fn logs_len(&self) -> usize {
self.logs.lock().unwrap().len()
}
pub fn set_snapshot(&self, snap: BusSnap) {
*self.snap.lock().unwrap() = Arc::new(snap);
self.bump();
}
pub fn snapshot(&self) -> Arc<BusSnap> {
Arc::clone(&self.snap.lock().unwrap())
}
pub fn rev(&self) -> u64 {
self.rev.load(Ordering::Acquire)
}
pub fn bump(&self) {
self.rev.fetch_add(1, Ordering::AcqRel);
}
pub fn request_stop(&self) {
self.stop.store(true, Ordering::Release);
}
pub fn is_stop(&self) -> bool {
self.stop.load(Ordering::Acquire)
}
pub fn push_cmd(&self, cmd: UiCmd) {
self.cmds.lock().unwrap().push_back(cmd);
self.bump();
}
pub fn drain_cmds(&self) -> Vec<UiCmd> {
self.cmds.lock().unwrap().drain(..).collect()
}
pub fn pending_confirm(&self) -> Option<String> {
self.pending_confirm.lock().unwrap().clone()
}
pub fn confirm_step(&self, name: &str) -> ConfirmStep {
let mut g = self.pending_confirm.lock().unwrap();
match g.take() {
Some(p) if p == name => ConfirmStep::Fired,
Some(_) => {
*g = Some(name.to_string());
ConfirmStep::Switched
}
None => {
*g = Some(name.to_string());
ConfirmStep::Armed
}
}
}
pub fn confirm_cancel(&self) {
*self.pending_confirm.lock().unwrap() = None;
self.bump();
}
pub fn market(&self) -> &MarketStore {
&self.market
}
pub fn set_trade_snap(&self, ts: TradeSnap) {
*self.trade_snap.lock().unwrap() = Arc::new(ts);
self.bump();
}
pub fn trade_snap(&self) -> Arc<TradeSnap> {
self.trade_snap.lock().unwrap().clone()
}
pub fn ctp_form(&self) -> CtpForm {
self.ctp_form.lock().unwrap().clone()
}
pub fn set_ctp_form(&self, f: CtpForm) {
*self.ctp_form.lock().unwrap() = f;
self.bump();
}
}