use std::collections::{BTreeMap, HashSet, VecDeque};
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, AtomicUsize, 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 = 4096;
pub(crate) fn beijing_tz() -> chrono::FixedOffset {
chrono::FixedOffset::east_opt(8 * 3600).expect("+08 offset is always valid")
}
#[derive(Debug, Clone)]
pub struct LogEntry {
pub at: String,
pub text: String,
}
#[derive(Debug)]
pub struct LogRing {
lines: VecDeque<LogEntry>,
cap: usize,
spilled: u64,
spill_dir: Option<std::path::PathBuf>,
}
pub(crate) fn default_spill_dir() -> std::path::PathBuf {
if let Ok(d) = std::env::var("EK_UI_LOG_DIR")
&& !d.trim().is_empty()
{
return std::path::PathBuf::from(d);
}
std::env::var("APPDATA")
.map(|r| std::path::PathBuf::from(r).join("ek").join("logs"))
.unwrap_or_else(|_| std::path::PathBuf::from(".ek-logs"))
}
impl Default for LogRing {
fn default() -> Self {
Self::new_spilling(RING_CAP, default_spill_dir())
}
}
impl LogRing {
pub fn new(cap: usize) -> Self {
Self {
lines: VecDeque::with_capacity(cap),
cap,
spilled: 0,
spill_dir: None,
}
}
pub fn new_spilling(cap: usize, dir: std::path::PathBuf) -> Self {
Self {
lines: VecDeque::with_capacity(cap),
cap,
spilled: 0,
spill_dir: Some(dir),
}
}
fn spill_write(&self, entry: &LogEntry) {
let Some(dir) = self.spill_dir.as_ref() else {
return;
};
let day = chrono::Local::now().format("%Y%m%d");
let path = dir.join(format!("ekapp-{day}.log"));
let _ = std::fs::create_dir_all(dir);
if let Ok(mut f) = std::fs::OpenOptions::new().append(true).create(true).open(&path) {
use std::io::Write as _;
let _ = writeln!(f, "{} | {}", entry.at, entry.text);
}
}
pub fn push(&mut self, line: impl Into<String>) {
let at = chrono::Utc::now()
.with_timezone(&beijing_tz())
.format("%H:%M:%S")
.to_string();
let entry = LogEntry {
at,
text: line.into(),
};
if self.spill_dir.is_some() {
self.spill_write(&entry);
}
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(entry);
}
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().map(|e| e.text.clone()).collect()
}
pub fn entries(&self) -> Vec<LogEntry> {
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, name: 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 },
CtpConfigure { gateway: String, form: CtpForm },
Connect { gateway: String, form: CtpForm },
CtpSubscribe {
gateway: String,
symbols: Vec<String>,
},
Disconnect { gateway: String },
CfgList,
CfgDelete { ns: String },
CfgResave { ns: String },
}
pub const KLINE_CELL_CAP: usize = 4096;
pub const MAX_ACCOUNTS: usize = 64;
pub const QUERY_REJECT_STREAK_LIMIT: u32 = 3;
pub const ACCOUNT_STALE_SECS: i64 = 30;
pub use ek_ctp::driver::CHAIN_STATE_PREFIX;
pub const LIST_POLL_TICKS: u32 = 4;
pub const STATUS_ASKS_PER_TICK: usize = 2;
pub const TRADE_CAP: usize = 5000;
pub const PENDING_WARN_FLOOR: u64 = 4_096;
pub const PENDING_CRIT_FLOOR: u64 = 32_768;
#[must_use]
pub fn pending_thresholds(soft_cap: u64) -> (u64, u64) {
let warn = (soft_cap / 16).max(PENDING_WARN_FLOOR);
let crit = (soft_cap / 4).max(PENDING_CRIT_FLOOR);
(warn.min(soft_cap.max(1)), crit.min(soft_cap.max(1)))
}
#[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, Clone)]
pub struct KlineFetch {
pub symbol: String,
pub interval_secs: i64,
}
#[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_history(&self, symbol: &str, iv_secs: i64, bars: &[BarData]) {
let mut cells = self.cells.lock().unwrap();
let cell = cells.entry((symbol.to_string(), iv_secs)).or_default();
cell.extend(bars.iter().cloned());
cell.sort_unstable_by_key(|b| b.datetime);
cell.dedup_by(|a, b| a.datetime == b.datetime);
if cell.len() > KLINE_CELL_CAP {
let excess = cell.len() - KLINE_CELL_CAP;
cell.drain(0..excess);
}
self.frames.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 find_cell(&self, sym: &str) -> Option<CellView> {
self.cells()
.into_iter()
.find(|c| c.symbol == sym || c.symbol.ends_with(&format!(".{sym}")))
}
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, fallback: Option<&str>) -> Result<String, String> {
let g = raw.trim();
if !g.is_empty() {
if g.chars().any(|c| c.is_whitespace()) {
return Err(format!("网关解析: 含空白:{raw:?}"));
}
return Ok(g.to_string());
}
match fallback.map(|f| f.trim()).filter(|f| !f.is_empty()) {
Some(f) => Ok(f.to_string()),
None => Err("网关解析: 未选择账户(先在账户面板添加账户)".to_string()),
}
}
#[derive(Debug, 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,
pub accept_futures: bool,
pub accept_options: bool,
}
impl Default for CtpForm {
fn default() -> Self {
Self {
broker_id: String::new(),
td_addr: String::new(),
td_user: String::new(),
td_password: String::new(),
td_auth_code: String::new(),
td_app_id: String::new(),
md_addr: String::new(),
md_user: String::new(),
md_password: String::new(),
symbols: String::new(),
accept_futures: true,
accept_options: false,
}
}
}
impl CtpForm {
#[must_use]
pub fn product_classes(&self) -> Vec<String> {
let mut codes = Vec::new();
if self.accept_futures {
codes.push("1".to_string());
}
if self.accept_options {
codes.push("2".to_string());
}
codes
}
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().into(),
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().into(),
product_classes: self.product_classes(),
}
}
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)
}
pub fn validate(&self) -> Result<(), String> {
fn front(name: &str, v: &str) -> Result<(), String> {
let v = v.trim();
let Some(rest) = v.strip_prefix("tcp://") else {
return Err(format!("{name} 形制应为 tcp://host:port(当前 {:?})", v));
};
let Some((host, port)) = rest.rsplit_once(':') else {
return Err(format!("{name} 缺端口:{name} {:?}", v));
};
if host.trim().is_empty() {
return Err(format!("{name} 主机位空:{}", v));
}
let port: u32 = port
.parse()
.map_err(|_| format!("{name} 端口不是数字:{}", v))?;
if port == 0 || port > 65535 {
return Err(format!("{name} 端口越界(1..=65535):{}", v));
}
Ok(())
}
if self.broker_id.trim().is_empty() {
return Err("broker 位空".to_string());
}
if self.td_user.trim().is_empty() {
return Err("交易账号位空".to_string());
}
front("td 前置", &self.td_addr)?;
if !self.md_addr.trim().is_empty() {
front("md 前置", &self.md_addr)?;
}
if self.td_app_id.trim().is_empty() {
return Err("AppID 位空(CTP 登录必需)".to_string());
}
if self.td_auth_code.trim().is_empty() {
return Err("AuthCode 位空(CTP 登录必需)".to_string());
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum AccountStage {
#[default]
NotAssembled,
Assembling,
Assembled,
AssembleFailed,
}
impl AccountStage {
#[must_use]
pub fn label(self) -> &'static str {
match self {
AccountStage::NotAssembled => "未装配",
AccountStage::Assembling => "装配中…",
AccountStage::Assembled => "已装配·登录态未知",
AccountStage::AssembleFailed => "装配失败",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum LoginState {
#[default]
Unknown,
Online,
LoginFailed,
FrontDown,
QueryStarved,
Stopped,
}
impl LoginState {
#[must_use]
pub fn label(self) -> &'static str {
match self {
LoginState::Unknown => "登录态未知",
LoginState::Online => "在线",
LoginState::LoginFailed => "登录失败",
LoginState::FrontDown => "掉线",
LoginState::QueryStarved => "查询断供",
LoginState::Stopped => "已断开",
}
}
#[must_use]
pub fn is_alarm(self) -> bool {
matches!(self, LoginState::LoginFailed | LoginState::FrontDown)
}
}
fn count_note(label: &str, n: u32) -> Option<String> {
(n > 0).then(|| format!("{label} {n} 次"))
}
fn brief_counter_reason(full: &str) -> Option<String> {
let tail = full.rfind("信息：").map(|i| &full[i + "信息：".len()..])?;
let body = tail.trim().trim_start_matches("CTP:").trim();
if body.is_empty() {
return None;
}
let code = full
.rfind("代码：")
.map(|i| {
let rest = &full[i + "代码：".len()..];
let end = rest.find('，').unwrap_or(rest.len());
rest[..end].trim().to_string()
})
.filter(|c| !c.is_empty() && c.parse::<i64>().is_ok());
Some(match code {
Some(c) => format!("{body}({c})"),
None => body.to_string(),
})
}
#[derive(Debug, Clone, PartialEq, Default)]
pub struct AccountRow {
pub gateway: String,
pub alias: String,
pub broker_id: String,
pub account: String,
pub md_source: bool,
pub form: CtpForm,
pub restored: bool,
pub auto_connect: bool,
pub password_ref: bool,
pub stage: AccountStage,
pub fail_note: Option<String>,
pub effective_classes: Vec<String>,
pub connected: bool,
pub login: LoginState,
pub status_note: Option<String>,
pub login_brief: Option<String>,
pub chain_note: Option<String>,
}
impl AccountRow {
pub fn derive(form: &CtpForm) -> Result<Self, String> {
Self::derive_with(form, &[])
}
pub fn derive_with(
form: &CtpForm,
taken: &[(String, String, String, String)],
) -> Result<Self, String> {
let broker = form.broker_id.trim();
let account = form.td_user.trim();
let gateway = crate::appconfig::unique_gw_name(
taken,
broker,
account,
form.td_addr.trim(),
)?;
Ok(Self {
gateway,
alias: String::new(),
broker_id: broker.to_string(),
account: account.to_string(),
md_source: false,
form: form.clone(),
stage: AccountStage::NotAssembled,
fail_note: None,
effective_classes: Vec::new(),
connected: false,
login: LoginState::Unknown,
status_note: None,
login_brief: None,
chain_note: None,
restored: false,
auto_connect: false,
password_ref: false,
})
}
#[must_use]
pub fn status_label(&self) -> String {
if self.stage == AccountStage::Assembled {
match self.login {
LoginState::Unknown => "已装配·登录态未知".to_string(),
other => format!("已装配·{}", other.label()),
}
} else {
self.stage.label().to_string()
}
}
pub fn apply_status(
&mut self,
st: &ek_ctp::plugin::GatewayStatus,
now_ms: i64,
reject_limit: u32,
stale_secs: i64,
) {
let td_side = !self.form.td_addr.trim().is_empty();
let (login, note) = if td_side {
if st.td_login_failed && !st.td_logged_in {
(
LoginState::LoginFailed,
st.td_last_error
.clone()
.or_else(|| Some("登录失败(网关未回带原因)".to_string())),
)
} else if st.td_logged_in {
if st.td_query_reject_streak >= reject_limit {
(
LoginState::QueryStarved,
Some(format!("查询连续被拒 {} 次", st.td_query_reject_streak)),
)
} else if let Some(last) = st.td_last_account_ms {
let idle_ms = now_ms.saturating_sub(last);
if idle_ms > stale_secs.saturating_mul(1_000) {
(
LoginState::QueryStarved,
Some(format!("资金回报已 {}s 未更新", idle_ms / 1_000)),
)
} else {
(LoginState::Online, count_note("掉线", st.td_disconnects))
}
} else {
let since_login_ms = st.td_last_login_ok_ms.map(|ok| now_ms.saturating_sub(ok));
match since_login_ms {
Some(ms) if ms > stale_secs.saturating_mul(1_000) => (
LoginState::QueryStarved,
Some(format!("资金回报从未到手(登录已 {}s)", ms / 1_000)),
),
_ => (LoginState::Online, Some("资金回报尚未到手".to_string())),
}
}
} else if self.login == LoginState::Stopped {
let extra = if st.td_disconnects > 0 {
format!("(此前掉线 {} 次)", st.td_disconnects)
} else {
String::new()
};
(
LoginState::Stopped,
Some(format!("会话已停{extra};『连接』可恢复")),
)
} else if st.td_disconnects > 0 {
(
LoginState::FrontDown,
Some(format!("掉线 {} 次", st.td_disconnects)),
)
} else {
(LoginState::Unknown, st.td_last_error.clone())
}
} else if st.md_logged_in {
(
LoginState::Online,
count_note("掉线", st.md_disconnects)
.or_else(|| Some("纯行情账户(无 TD 会话)".to_string())),
)
} else if self.login == LoginState::Stopped {
(
LoginState::Stopped,
Some("会话已停;『连接』可恢复".to_string()),
)
} else if st.md_disconnects > 0 {
(
LoginState::FrontDown,
Some(format!("掉线 {} 次", st.md_disconnects)),
)
} else {
(LoginState::Unknown, st.md_last_error.clone())
};
let note = if td_side && st.td_instrument_rejects > 0 {
let add = format!("合约查询被拒 {} 次", st.td_instrument_rejects);
match note {
Some(base) if base.contains("合约查询被拒") => Some(base),
Some(base) => Some(format!("{base};{add}")),
None => Some(add),
}
} else {
note
};
self.login = login;
self.status_note = note;
self.login_brief = if login.is_alarm() {
st.td_last_error
.as_deref()
.or(st.md_last_error.as_deref())
.and_then(brief_counter_reason)
} else {
None
};
}
#[must_use]
pub fn display_name(&self) -> String {
let a = self.alias.trim();
if a.is_empty() {
self.account.clone()
} else {
a.to_string()
}
}
}
#[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, Clone, Copy, PartialEq)]
pub struct DataSplit {
pub col: f32,
pub row: f32,
}
impl Default for DataSplit {
fn default() -> Self {
Self {
col: crate::splitters::DATA_SPLIT_DEFAULT,
row: crate::splitters::DATA_SPLIT_DEFAULT,
}
}
}
pub const MARKET_ORDER: [ek_data::constant::Exchange; 6] = [
ek_data::constant::Exchange::SHFE,
ek_data::constant::Exchange::DCE,
ek_data::constant::Exchange::CZCE,
ek_data::constant::Exchange::CFFEX,
ek_data::constant::Exchange::INE,
ek_data::constant::Exchange::GFEX,
];
#[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_orders: Mutex<BTreeMap<String, (u64, ek_data::order::OrderData)>>,
trade_trades: Mutex<VecDeque<(u64, ek_data::trade::TradeData)>>,
trade_ids: Mutex<HashSet<String>>,
trade_positions: Mutex<BTreeMap<String, (u64, ek_data::position::PositionData)>>,
trade_funds: Mutex<BTreeMap<String, (u64, ek_data::account::AccountData)>>,
trade_seq: AtomicU64,
trade_dropped: AtomicU64,
trade_dirty: AtomicBool,
trade_snap: Mutex<Arc<TradeSnap>>,
tick_store: Mutex<TickStore>,
order_form: Mutex<OrderForm>,
data_split: Mutex<DataSplit>,
selected_exchange: Mutex<String>,
topleft_tab: Mutex<TopLeftTab>,
bottomright_tab: Mutex<BottomRightTab>,
ops_window: AtomicBool,
accounts_window: AtomicBool,
contracts_window: AtomicBool,
contract_store: Mutex<ContractStore>,
contract_query: Mutex<String>,
contract_selected: Mutex<Vec<String>>,
kline_windows: Mutex<Vec<String>>,
kline_fetches: Mutex<Vec<KlineFetch>>,
kline_fetched: Mutex<HashSet<(String, i64)>>,
rev: AtomicU64,
stop: AtomicBool,
cmds: Mutex<VecDeque<UiCmd>>,
pending_confirm: Mutex<Option<String>>,
accounts: Mutex<Vec<AccountRow>>,
account_edit_window: AtomicBool,
editing_account: Mutex<Option<String>>,
edit_alias: Mutex<String>,
edit_error: Mutex<Option<String>>,
md_source: Mutex<String>,
subscriptions: Mutex<Vec<String>>,
row_sel: Mutex<BTreeMap<String, String>>,
table_sorts: Mutex<BTreeMap<String, Option<(usize, bool)>>>,
status_cursor: AtomicUsize,
status_seed: Mutex<Vec<String>>,
list_tick: AtomicU32,
list_due: AtomicBool,
seed_due: AtomicBool,
pipeline_due: AtomicBool,
cfg_due: AtomicBool,
cfg_briefs: Mutex<Option<Vec<ek_cfg::wire::NsBrief>>>,
cfg_delete_arm: Mutex<Option<String>>,
theme: AtomicU8,
theme_persist_due: AtomicBool,
auto_connect_persist_due: AtomicBool,
alias_persist_due: AtomicBool,
archive_purge: Mutex<Vec<String>>,
bad_payload: Mutex<BTreeMap<String, u64>>,
pipeline_lines: Mutex<Vec<String>>,
status_thresholds: Mutex<Option<(u32, i64)>>,
pending_alerts: Mutex<Vec<(crate::toast::AlertLevel, String)>>,
toasts: crate::toast::ToastsCell,
alerted: Mutex<std::collections::BTreeSet<String>>,
exit_confirm_open: AtomicBool,
exit_confirmed: AtomicBool,
assembly_locked: AtomicBool,
}
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 {
let s = self.selected_exchange.lock().unwrap().clone();
if MARKET_ORDER.iter().any(|e| e.as_str() == s) {
s
} else {
MARKET_ORDER[0].as_str().to_string()
}
}
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 data_split(&self) -> DataSplit {
*self.data_split.lock().unwrap()
}
pub fn set_data_split(&self, s: DataSplit) {
let s = DataSplit {
col: crate::splitters::clamp_frac(s.col),
row: crate::splitters::clamp_frac(s.row),
};
let mut g = self.data_split.lock().unwrap();
if g.col != s.col || g.row != s.row {
*g = s;
self.bump();
}
}
pub fn reset_layout(&self) {
*self.data_split.lock().unwrap() = DataSplit::default();
*self.selected_exchange.lock().unwrap() = String::new();
*self.topleft_tab.lock().unwrap() = TopLeftTab::default();
*self.bottomright_tab.lock().unwrap() = BottomRightTab::default();
self.bump();
}
pub fn ops_window_open(&self) -> bool {
self.ops_window.load(Ordering::Acquire)
}
pub fn table_sort(&self, table: &str) -> Option<(usize, bool)> {
self.table_sorts
.lock()
.unwrap()
.get(table)
.copied()
.flatten()
}
pub fn toggle_table_sort(&self, table: &str, col: usize) {
let mut m = self.table_sorts.lock().unwrap();
let asc = match m.get(table).copied().flatten() {
Some((c, a)) if c == col => !a,
_ => true,
};
m.insert(table.to_string(), Some((col, asc)));
}
pub fn set_ops_window_open(&self, v: bool) {
self.ops_window.store(v, Ordering::Release);
self.bump();
}
pub fn accounts_window_open(&self) -> bool {
self.accounts_window.load(Ordering::Acquire)
}
pub fn set_accounts_window_open(&self, v: bool) {
self.accounts_window.store(v, Ordering::Release);
self.bump();
}
pub fn account_edit_open(&self) -> bool {
self.account_edit_window.load(Ordering::Acquire)
}
pub fn editing_account(&self) -> Option<String> {
self.editing_account.lock().unwrap().clone()
}
pub fn begin_account_edit(&self, gateway: Option<&str>) -> bool {
match gateway {
Some(gw) => {
let Some(row) = self.account(gw) else {
return false;
};
*self.editing_account.lock().unwrap() = Some(gw.to_string());
*self.edit_alias.lock().unwrap() = row.alias.clone();
self.set_ctp_form(row.form.clone());
}
None => {
*self.editing_account.lock().unwrap() = None;
*self.edit_alias.lock().unwrap() = String::new();
let draft = self.draft_from_existing();
self.set_ctp_form(draft);
}
}
*self.edit_error.lock().unwrap() = None;
self.account_edit_window.store(true, Ordering::Release);
self.bump();
true
}
pub fn end_account_edit(&self) {
*self.editing_account.lock().unwrap() = None;
*self.edit_alias.lock().unwrap() = String::new();
*self.edit_error.lock().unwrap() = None;
self.account_edit_window.store(false, Ordering::Release);
let draft = self.draft_from_existing();
self.set_ctp_form(draft);
self.bump();
}
pub fn end_account_edit_if_editing(&self, gateway: &str) {
if self.editing_account().as_deref() == Some(gateway) {
self.end_account_edit();
}
}
pub fn edit_alias(&self) -> String {
self.edit_alias.lock().unwrap().clone()
}
pub fn set_edit_alias(&self, v: impl Into<String>) {
*self.edit_alias.lock().unwrap() = v.into();
self.bump();
}
pub fn edit_error(&self) -> Option<String> {
self.edit_error.lock().unwrap().clone()
}
pub fn set_edit_error(&self, e: Option<String>) {
*self.edit_error.lock().unwrap() = e;
self.bump();
}
pub fn account_edit_dirty(&self) -> bool {
let editing = self.editing_account();
let draft = self.ctp_form();
match editing {
Some(gw) => self.account(&gw).map(|a| a.form != draft).unwrap_or(false),
None => draft != CtpForm::default() && !draft.td_user.trim().is_empty(),
}
}
pub fn accounts(&self) -> Vec<AccountRow> {
self.accounts.lock().unwrap().clone()
}
pub fn account(&self, gateway: &str) -> Option<AccountRow> {
self.accounts
.lock()
.unwrap()
.iter()
.find(|a| a.gateway == gateway)
.cloned()
}
pub fn add_account(&self, row: AccountRow) -> Result<(), String> {
let mut acc = self.accounts.lock().unwrap();
if acc.iter().any(|a| a.gateway == row.gateway) {
return Err(format!("账户 {} 已存在(同 broker + 账号)", row.gateway));
}
let alias = row.alias.trim();
if !alias.is_empty()
&& acc
.iter()
.any(|a| a.alias.trim().eq_ignore_ascii_case(alias))
{
return Err(format!("别名 {alias:?} 已被其他账户占用(别名须唯一)"));
}
if acc.len() >= MAX_ACCOUNTS {
return Err(format!(
"账户已达软上限 {MAX_ACCOUNTS} 个(先移除一个再添加)"
));
}
let has_alias = !row.alias.trim().is_empty();
acc.push(row);
drop(acc);
if has_alias {
self.alias_persist_due.store(true, Ordering::Release);
}
self.bump();
Ok(())
}
#[must_use]
pub fn can_add_account(&self) -> bool {
self.accounts.lock().unwrap().len() < MAX_ACCOUNTS
}
pub fn set_account_alias(&self, gateway: &str, alias: &str) -> bool {
let alias = alias.trim().to_string();
let mut acc = self.accounts.lock().unwrap();
if !alias.is_empty()
&& acc
.iter()
.any(|a| a.gateway != gateway && a.alias.eq_ignore_ascii_case(&alias))
{
return false;
}
match acc.iter_mut().find(|a| a.gateway == gateway) {
Some(a) => {
a.alias = alias;
drop(acc);
self.alias_persist_due.store(true, Ordering::Release);
self.bump();
true
}
None => false,
}
}
pub fn note_connect_sent(&self, gateway: &str) {
let mut acc = self.accounts.lock().unwrap();
if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
a.stage = AccountStage::Assembling;
a.fail_note = None;
a.login = LoginState::Unknown;
a.status_note = None;
a.login_brief = None;
a.chain_note = None;
drop(acc);
self.bump();
}
}
pub fn note_connect_result(&self, gateway: &str, result: Result<(), String>) {
let mut acc = self.accounts.lock().unwrap();
if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
match result {
Ok(()) => {
a.stage = AccountStage::Assembled;
a.fail_note = None;
}
Err(e) => {
a.stage = AccountStage::AssembleFailed;
a.fail_note = Some(e);
}
}
drop(acc);
self.bump();
}
}
pub fn note_effective_product_classes(&self, gateway: &str, classes: &[String]) {
let mut acc = self.accounts.lock().unwrap();
if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
a.effective_classes = classes.to_vec();
drop(acc);
self.bump();
}
}
pub fn note_chain(&self, gateway: &str, note: &str) {
let mut acc = self.accounts.lock().unwrap();
if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
if a.chain_note.as_deref() == Some(note) {
return; }
a.chain_note = Some(note.to_string());
drop(acc);
self.bump();
}
}
#[must_use]
pub fn status_thresholds(&self) -> (u32, i64) {
self.status_thresholds
.lock()
.unwrap()
.unwrap_or((QUERY_REJECT_STREAK_LIMIT, ACCOUNT_STALE_SECS))
}
pub fn set_status_thresholds(&self, reject_limit: u32, stale_secs: i64) {
*self.status_thresholds.lock().unwrap() = Some((reject_limit, stale_secs));
self.bump();
}
pub fn list_poll_due(&self) -> bool {
let n = self.list_tick.fetch_add(1, Ordering::Relaxed) + 1;
self.list_due.swap(false, Ordering::AcqRel) || n == 1 || n.is_multiple_of(LIST_POLL_TICKS)
}
pub fn request_list_refresh(&self) {
self.list_due.store(true, Ordering::Release);
}
pub fn request_trade_reseed(&self) {
self.seed_due.store(true, Ordering::Release);
}
pub fn take_trade_reseed(&self) -> bool {
self.seed_due.swap(false, Ordering::AcqRel)
}
pub fn status_batch(&self, present: &[String]) -> Vec<String> {
let mut uniq: Vec<String> = Vec::with_capacity(present.len());
for gw in present {
if !uniq.iter().any(|u| u == gw) {
uniq.push(gw.clone());
}
}
let n = uniq.len();
if n == 0 {
return Vec::new();
}
let start = self.status_cursor.fetch_add(1, Ordering::Relaxed) % n;
let take = STATUS_ASKS_PER_TICK.min(n);
let mut out = Vec::with_capacity(take);
for i in 0..take {
out.push(uniq[(start + i) % n].clone());
}
out
}
pub fn request_status_seed(&self, gateway: &str) {
let mut q = self.status_seed.lock().unwrap();
if !q.iter().any(|g| g == gateway) {
q.push(gateway.to_string());
}
}
pub fn take_status_seeds(&self) -> Vec<String> {
let mut q = self.status_seed.lock().unwrap();
std::mem::take(&mut *q)
}
pub fn note_status_event(&self, ev: &ek_ctp::GatewayStatusEvent) {
let gw = ev.gateway.as_str();
let text = ev.describe();
self.log_line(format!("[INFO][{gw}] {text}"));
let note = text
.strip_prefix(CHAIN_STATE_PREFIX)
.map(|s| s.trim().to_string())
.unwrap_or_else(|| text.clone());
self.note_chain(gw, ¬e);
self.note_gateway_status(gw, &ev.status);
let alarm_kind = matches!(
ev.change,
ek_ctp::GatewayChange::TdDisconnected
| ek_ctp::GatewayChange::TdAuthFailed
| ek_ctp::GatewayChange::TdLoginFailed
| ek_ctp::GatewayChange::ContractQueryStopped
| ek_ctp::GatewayChange::QueryRejected
| ek_ctp::GatewayChange::MdDisconnected
| ek_ctp::GatewayChange::MdLoginFailed
| ek_ctp::GatewayChange::SubscribeRejected
);
if alarm_kind {
let session = if ev.change.kind().starts_with("md-") {
ev.status.md_disconnects
} else {
ev.status.td_disconnects
};
let key = format!("estatus:{gw}:{}:{session}", ev.change.kind());
if self.alert_once(key) {
self.push_alert(
crate::toast::AlertLevel::Warning,
format!("账户 {gw}:{}", note.trim()),
);
}
}
}
pub fn request_pipeline_refresh(&self) {
self.pipeline_due.store(true, Ordering::Release);
}
pub fn take_pipeline_due(&self) -> bool {
self.pipeline_due.swap(false, Ordering::AcqRel)
}
pub fn request_cfg_refresh(&self) {
self.cfg_due.store(true, Ordering::Release);
}
pub fn take_cfg_due(&self) -> bool {
self.cfg_due.swap(false, Ordering::AcqRel)
}
pub fn theme(&self) -> crate::tokens::Theme {
crate::tokens::Theme::from_u8(self.theme.load(Ordering::Acquire))
}
pub fn set_theme(&self, theme: crate::tokens::Theme) {
self.theme.store(theme.as_u8(), Ordering::Release);
self.bump();
}
pub fn toggle_theme(&self) -> crate::tokens::Theme {
let next = self.theme().flipped();
self.theme.store(next.as_u8(), Ordering::Release);
self.theme_persist_due.store(true, Ordering::Release);
self.bump();
next
}
pub fn take_theme_persist_due(&self) -> bool {
self.theme_persist_due.swap(false, Ordering::AcqRel)
}
pub fn set_auto_connect(&self, gateway: &str, on: bool) -> Result<(), String> {
{
let mut acc = self.accounts.lock().unwrap();
match acc.iter_mut().find(|a| a.gateway == gateway) {
Some(a) => {
if a.auto_connect != on {
a.auto_connect = on;
self.auto_connect_persist_due.store(true, Ordering::Release);
self.bump();
}
}
None => return Err(format!("账户 {gateway} 不在清册")),
}
}
Ok(())
}
pub fn take_auto_connect_persist_due(&self) -> bool {
self.auto_connect_persist_due.swap(false, Ordering::AcqRel)
}
pub fn take_alias_persist_due(&self) -> bool {
self.alias_persist_due.swap(false, Ordering::AcqRel)
}
pub fn take_archive_purges(&self) -> Vec<String> {
let mut q = self.archive_purge.lock().unwrap();
std::mem::take(&mut *q)
}
pub fn restore_saved_accounts_from(&self, store: &ek_cfg::Store) -> Vec<String> {
let mut flags: std::collections::BTreeMap<String, bool> = Default::default();
let mut aliases: std::collections::BTreeMap<String, String> = Default::default();
match store.load(crate::plugin::THEME_NS) {
Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => {
for (k, _f) in map.iter() {
if let Some(gw) = k.strip_prefix("auto_connect.") {
flags.insert(gw.to_string(), map.get_plain(k) == Some("1"));
}
if let Some(gw) = k.strip_prefix("alias.")
&& let Some(v) = map.get_plain(k)
{
let v = v.trim();
if !v.is_empty() {
aliases.insert(gw.to_string(), v.to_string());
}
}
}
}
Ok(_) => {} Err(_) => {} }
let mut restored: Vec<String> = Vec::new();
let ns_list = match store.list_namespaces() {
Ok(v) => v,
Err(_) => {
let msg = "[ui] 档案清点失败:存储根不可读(具体见运行日志)".to_string();
self.log_line(msg.clone());
return restored;
}
};
for ns in ns_list {
let Some(gw) = ns.strip_prefix("ctp.") else {
continue;
};
if self.account(gw).is_some() {
self.log_line(format!(
"[ui] 档案 {ns} 与在册同名行撞名,清点不覆盖(清册行为唯一事实源)"
));
continue;
}
let map = match store.load(&ns) {
Ok(ek_cfg::LoadOutcome::Loaded { map, .. }) => map,
Ok(_) => {
self.log_line(format!("[ui] 档案 {ns} 一段也读不出(跳过)"));
continue;
}
Err(e) => {
self.log_line(format!("[ui] 档案 {ns} 读不了:{e:?}(跳过)"));
continue;
}
};
let password_ref = map
.get_secret_key("td_password")
.is_some_and(|k| !k.is_empty());
let mut form = CtpForm {
broker_id: map.get_plain("broker_id").unwrap_or_default().to_owned(),
td_addr: map.get_plain("td_addr").unwrap_or_default().to_owned(),
td_user: map.get_plain("td_user").unwrap_or_default().to_owned(),
td_password: String::new(),
td_auth_code: map.get_plain("td_auth_code").unwrap_or_default().to_owned(),
td_app_id: map.get_plain("td_app_id").unwrap_or_default().to_owned(),
md_addr: map.get_plain("md_addr").unwrap_or_default().to_owned(),
md_user: String::new(),
md_password: String::new(),
symbols: String::new(),
accept_futures: false,
accept_options: false,
};
for c in map
.get_plain("product_classes")
.unwrap_or_default()
.split(',')
{
if c.trim() == "1" {
form.accept_futures = true;
}
}
let Ok(mut row) = AccountRow::derive(&form) else {
self.log_line(format!(
"[ui] 档案 {ns} 无法派生(账号/前置非法别名冲突?),跳过 —— 界面手输重配更省心"
));
continue;
};
row.gateway = gw.to_string();
row.restored = true;
row.auto_connect = flags.get(gw).copied().unwrap_or(false);
row.password_ref = password_ref;
if let Some(a) = aliases.get(gw) {
let taken = self
.accounts
.lock()
.unwrap()
.iter()
.any(|x| x.alias.trim().eq_ignore_ascii_case(a));
if taken {
self.log_line(format!(
"[ui] 档案别名 {a:?} 已被在册行占用,{gw} 不带标签复原(账户面板改别名即可)"
));
} else {
row.alias = a.clone();
}
}
restored.push(row.gateway.clone());
self.add_account(row).ok();
}
if !restored.is_empty() {
self.log_line(format!(
"[ui] 账户档案清点复原 {} 行(手点「连接」直连存档;自动连接按行上勾选)",
restored.len()
));
if self.md_source.lock().unwrap().is_empty() {
let first = self
.accounts
.lock()
.unwrap()
.iter()
.find(|a| a.auto_connect)
.map(|a| a.gateway.clone());
if let Some(gw) = first
&& self.set_md_source(&gw).is_ok()
{
self.log_line(format!(
"[账户] 行情源(默认=第一个勾选自动连接的)= {gw}"
));
}
}
self.bump();
}
restored
}
pub fn restore_saved_accounts(&self) -> Vec<String> {
match ek_cfg::Store::eco() {
Ok(store) => self.restore_saved_accounts_from(&store),
Err(_) => {
self.log_line(
"[ui] 档案清点跳过:config_dir() 不可用(BD-CFG-20;配置根异常该显形)".to_string(),
);
Vec::new()
}
}
}
pub fn set_cfg_briefs(&self, b: Vec<ek_cfg::wire::NsBrief>) {
*self.cfg_briefs.lock().unwrap() = Some(b);
self.bump();
}
pub fn cfg_briefs(&self) -> Option<Vec<ek_cfg::wire::NsBrief>> {
self.cfg_briefs.lock().unwrap().clone()
}
pub fn cfg_delete_armed(&self) -> Option<String> {
self.cfg_delete_arm.lock().unwrap().clone()
}
pub fn arm_cfg_delete(&self, ns: Option<String>) {
*self.cfg_delete_arm.lock().unwrap() = ns;
self.bump();
}
pub fn set_pipeline_lines(&self, lines: Vec<String>) {
*self.pipeline_lines.lock().unwrap() = lines;
self.bump();
}
pub fn pipeline_lines(&self) -> Vec<String> {
self.pipeline_lines.lock().unwrap().clone()
}
pub fn note_bad_payload(&self, line: &str) {
let mut book = self.bad_payload.lock().unwrap();
let n = book.entry(line.to_string()).or_insert(0);
*n += 1;
if *n == 1 {
self.log_line(format!(
"[ui] {line} 载荷类型异常(第一次):这一路的行进不了账本"
));
}
}
#[must_use]
pub fn bad_payload_count(&self, line: &str) -> u64 {
self.bad_payload
.lock()
.unwrap()
.get(line)
.copied()
.unwrap_or(0)
}
pub fn note_plugin_spill(&self, rows: &[ek_station::wire::PlugRow]) {
for r in rows {
let (warn, crit) = pending_thresholds(r.soft_cap);
if r.pending >= warn && self.alert_once(format!("pend-warn:{}", r.name)) {
self.push_alert(
crate::toast::AlertLevel::Warning,
format!(
"插件 {}：邮箱积压 {} 帧(峰值 {};该插件软顶 {})——它的 worker 消化不过来,刷新会变慢",
r.name, r.pending, r.peak, r.soft_cap
),
);
}
if r.pending >= crit && self.alert_once(format!("pend-crit:{}", r.name)) {
self.push_alert(
crate::toast::AlertLevel::Error,
format!(
"插件 {}：邮箱积压 {} 帧,越过软顶 {} 的四分之一——再涨就开始丢新到的事件",
r.name, r.pending, r.soft_cap
),
);
}
if r.spilled > 0 && self.alert_once(format!("spill:{}", r.name)) {
self.push_alert(
crate::toast::AlertLevel::Error,
format!(
"插件 {}：已触软顶 {} 并丢弃新到事件累计 {} 条(峰值 {})——它的消化速度跟不上生产速度",
r.name, r.soft_cap, r.spilled, r.peak
),
);
}
}
}
pub fn note_gateway_status(&self, gateway: &str, st: &ek_ctp::plugin::GatewayStatus) {
let (limit, stale) = self.status_thresholds();
let now = ek_ctp::map::now_epoch_ms();
let mut alarm: Option<(LoginState, String)> = None;
let mut found = false;
{
let mut acc = self.accounts.lock().unwrap();
if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
found = true;
a.apply_status(st, now, limit, stale);
if a.login.is_alarm() || a.login == LoginState::QueryStarved {
alarm = Some((
a.login,
a.status_note
.clone()
.unwrap_or_else(|| a.login.label().to_string()),
));
}
}
}
if found {
self.bump();
}
if let Some((login, note)) = alarm
&& self.alert_once(format!("{gateway}|{}", login.label()))
{
self.push_alert(
crate::toast::AlertLevel::Error,
format!("账户 {gateway}：{note}"),
);
}
}
pub fn note_status_unreachable(&self, gateway: &str, why: &str) {
let mut acc = self.accounts.lock().unwrap();
if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
a.login = LoginState::Unknown;
a.status_note = Some(format!("状态口问不到:{why}"));
a.login_brief = None;
drop(acc);
self.bump();
}
}
pub fn mark_account_stopped(&self, gateway: &str) {
let mut acc = self.accounts.lock().unwrap();
if let Some(a) = acc.iter_mut().find(|a| a.gateway == gateway) {
a.login = LoginState::Stopped;
a.status_note = Some("会话已停;『连接』可恢复".to_string());
a.login_brief = None;
drop(acc);
self.bump();
}
}
pub fn prepare_connect_form(&self, gateway: &str, form: CtpForm) -> CtpForm {
let source = self.md_source();
let mut f = form;
if source.as_str() != gateway {
f.md_addr.clear();
f.md_user.clear();
f.md_password.clear();
}
self.update_account_form(gateway, f.clone());
f
}
#[must_use]
pub fn draft_from_existing(&self) -> CtpForm {
let acc = self.accounts.lock().unwrap();
let mut draft = CtpForm::default();
if let Some(a) = acc.last() {
draft.broker_id = a.form.broker_id.clone();
draft.td_auth_code = a.form.td_auth_code.clone();
draft.td_app_id = a.form.td_app_id.clone();
draft.accept_futures = a.form.accept_futures;
draft.accept_options = a.form.accept_options;
}
draft
}
pub fn update_account_form(&self, gateway: &str, form: CtpForm) -> bool {
let mut acc = self.accounts.lock().unwrap();
match acc.iter_mut().find(|a| a.gateway == gateway) {
Some(a) => {
a.form = form;
drop(acc);
self.bump();
true
}
None => false,
}
}
pub fn remove_account(&self, gateway: &str) -> bool {
let mut acc = self.accounts.lock().unwrap();
let before = acc.len();
acc.retain(|a| a.gateway != gateway);
let removed = acc.len() != before;
if removed {
let mut md = self.md_source.lock().unwrap();
if md.as_str() == gateway {
md.clear();
}
drop(acc);
self.alias_persist_due.store(true, Ordering::Release);
self.archive_purge.lock().unwrap().push(gateway.to_string());
self.bump();
}
removed
}
pub fn sync_accounts_connected(&self, rows: &[PlugRow]) -> Vec<String> {
let mut newly = Vec::new();
{
let mut acc = self.accounts.lock().unwrap();
for a in acc.iter_mut() {
let up = rows.iter().any(|r| r.name == a.gateway);
if up && !a.connected {
newly.push(a.gateway.clone());
}
a.connected = up;
if up {
a.stage = AccountStage::Assembled;
a.fail_note = None;
} else {
a.chain_note = None;
if a.stage == AccountStage::Assembled {
a.stage = AccountStage::NotAssembled;
a.login = LoginState::Unknown;
a.status_note = None;
a.login_brief = None;
}
}
}
}
if !newly.is_empty() {
self.bump();
}
newly
}
pub fn md_source(&self) -> String {
self.md_source.lock().unwrap().clone()
}
pub fn accounts_locked(&self) -> bool {
self.accounts.lock().unwrap().iter().any(|a| a.connected)
}
pub fn set_md_source(&self, gateway: &str) -> Result<(), String> {
{
let acc = self.accounts.lock().unwrap();
if acc.iter().any(|a| a.connected) {
return Err("元账号已锁定:有账户在线,切换行情源需先断开该账户".to_string());
}
if !acc.iter().any(|a| a.gateway == gateway) {
return Err(format!("账户 {gateway} 不在清册"));
}
}
let mut acc = self.accounts.lock().unwrap();
for a in acc.iter_mut() {
a.md_source = a.gateway == gateway;
}
drop(acc);
let mut md = self.md_source.lock().unwrap();
*md = gateway.to_string();
self.bump();
Ok(())
}
pub fn subscriptions(&self) -> Vec<String> {
self.subscriptions.lock().unwrap().clone()
}
pub fn note_subscriptions(&self, symbols: &[String]) {
{
let mut subs = self.subscriptions.lock().unwrap();
for s in symbols {
let s = s.trim();
if !s.is_empty() && !subs.iter().any(|x| x == s) {
subs.push(s.to_string());
}
}
}
self.bump();
}
pub fn default_gateway(&self) -> Option<String> {
let chosen = {
let f = self.order_form.lock().unwrap();
f.gateway.trim().to_string()
};
if !chosen.is_empty() && self.account(&chosen).is_some_and(|a| a.connected) {
return Some(chosen);
}
let acc = self.accounts.lock().unwrap();
acc.iter().find(|a| a.connected).map(|a| a.gateway.clone())
}
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}"));
let fq = if s.contains('.') {
s.clone()
} else {
format!("{}.{}", self.selected_exchange(), s)
};
const KLINE_INTERVAL_SECS: i64 = 60;
{
let mut fetched = self.kline_fetched.lock().unwrap();
if fetched.contains(&(fq.clone(), KLINE_INTERVAL_SECS)) {
return;
}
fetched.insert((fq.clone(), KLINE_INTERVAL_SECS));
}
self.kline_fetches.lock().unwrap().push(KlineFetch {
symbol: fq,
interval_secs: KLINE_INTERVAL_SECS,
});
self.log_line(format!("[kline] 入队历史补数 {s}(1min×300,tq 桥)"));
}
pub fn take_kline_fetches(&self) -> Vec<KlineFetch> {
std::mem::take(&mut self.kline_fetches.lock().unwrap())
}
pub fn kline_fetch_pending(&self) -> bool {
!self.kline_fetches.lock().unwrap().is_empty()
}
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);
if !c.name.is_empty() {
self.backfill_names(&c.symbol, &c.name);
}
self.bump();
}
fn backfill_names(&self, symbol: &str, name: &str) {
let mut touched = false;
{
let mut rows = self.trade_orders.lock().unwrap();
for (_, (_, o)) in rows.iter_mut() {
if o.symbol == symbol && o.name.is_empty() {
o.name = name.to_string();
touched = true;
}
}
}
{
let mut rows = self.trade_trades.lock().unwrap();
for (_, t) in rows.iter_mut() {
if t.symbol == symbol && t.name.is_empty() {
t.name = name.to_string();
touched = true;
}
}
}
{
let mut rows = self.trade_positions.lock().unwrap();
for (_, (_, p)) in rows.iter_mut() {
if p.symbol == symbol && p.name.is_empty() {
p.name = name.to_string();
touched = true;
}
}
}
if touched {
self.trade_dirty.store(true, Ordering::Release);
}
}
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()
}
fn contract_name_of(&self, symbol: &str) -> Option<String> {
self.contract_store
.lock()
.unwrap()
.query(symbol)
.into_iter()
.find(|c| c.symbol == symbol && !c.name.is_empty())
.map(|c| c.name)
}
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()
}
#[must_use]
pub fn cancel_reject_note(msg: &str) -> String {
let head = msg.split(' ').next().unwrap_or(msg);
let note = if head.starts_with("unknown:order") {
"这台网关的账本里没有这笔委托(登录前的旧单,或不属于这个账户)"
} else if head.starts_with("mismatch:gateway") {
"该委托不属于这个账户(归属守卫挡住,不会撤到别人头上)"
} else if head.starts_with("terminal:status") {
"这笔委托已经到终态(已成交/已撤/已拒),撤不了了"
} else if head.starts_with("limit:cancel-guard") {
"撤单尝试次数已到上限(每单有上限,防连点打爆柜台)"
} else {
""
};
if note.is_empty() {
format!("撤单被拒:{msg}")
} else {
format!("撤单被拒:{note}(机器码:{msg})")
}
}
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_entries(&self) -> (Vec<LogEntry>, u64) {
let r = self.logs.lock().unwrap();
(r.entries(), 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 push_alert(&self, level: crate::toast::AlertLevel, text: impl Into<String>) {
self.pending_alerts
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.push((level, text.into()));
}
pub(crate) fn take_alerts(&self) -> Vec<(crate::toast::AlertLevel, String)> {
let mut q = self
.pending_alerts
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
std::mem::take(&mut *q)
}
pub(crate) fn toasts(&self) -> std::sync::MutexGuard<'_, Option<egui_notify::Toasts>> {
self.toasts.lock()
}
pub fn alert_once(&self, key: impl Into<String>) -> bool {
let key = key.into();
self.alerted
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(key)
}
#[must_use]
pub fn pending_alert_count(&self) -> usize {
self.pending_alerts
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.len()
}
#[must_use]
pub fn alerted_count(&self) -> usize {
self.alerted
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.len()
}
pub fn exit_confirm_open(&self) -> bool {
self.exit_confirm_open
.load(std::sync::atomic::Ordering::Relaxed)
}
pub fn open_exit_confirm(&self) {
self.exit_confirm_open
.store(true, std::sync::atomic::Ordering::Relaxed);
}
pub fn close_exit_confirm(&self) {
self.exit_confirm_open
.store(false, std::sync::atomic::Ordering::Relaxed);
}
pub fn exit_confirmed(&self) -> bool {
self.exit_confirmed
.load(std::sync::atomic::Ordering::Relaxed)
}
pub fn confirm_exit(&self) {
self.exit_confirmed
.store(true, std::sync::atomic::Ordering::Relaxed);
}
pub fn is_stop(&self) -> bool {
self.stop.load(Ordering::Acquire)
}
#[must_use]
pub fn assembly_locked(&self) -> bool {
self.assembly_locked.load(Ordering::Relaxed)
}
pub fn set_assembly_locked(&self, on: bool) {
self.assembly_locked.store(on, Ordering::Relaxed);
}
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()
}
#[must_use]
pub fn cmd_pending(&self) -> bool {
!self.cmds.lock().unwrap().is_empty()
}
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 seed_trade_snap(&self, ts: TradeSnap) {
let named = |symbol: &str, name: String| -> String {
if name.is_empty() {
self.contract_name_of(symbol).unwrap_or(name)
} else {
name
}
};
{
let mut book = self.trade_orders.lock().unwrap();
book.clear();
for mut o in ts.orders.into_iter().rev() {
o.name = named(&o.symbol, o.name);
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
book.insert(o.en_orderid(), (seq, o));
}
}
{
let mut book = self.trade_trades.lock().unwrap();
book.clear();
let mut ids = self.trade_ids.lock().unwrap();
ids.clear();
for mut t in ts.trades.into_iter().rev() {
t.name = named(&t.symbol, t.name);
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
ids.insert(t.en_tradeid());
book.push_front((seq, t));
}
}
{
let mut book = self.trade_positions.lock().unwrap();
book.clear();
for mut p in ts.positions.into_iter().rev() {
p.name = named(&p.symbol, p.name);
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
book.insert(p.en_positionid(), (seq, p));
}
}
{
let mut book = self.trade_funds.lock().unwrap();
book.clear();
for a in ts.accounts.into_iter().rev() {
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
book.insert(a.en_accountid(), (seq, a));
}
}
self.trade_dirty.store(true, Ordering::Release);
self.bump();
}
pub fn note_order(&self, o: &ek_data::order::OrderData) {
let key = o.en_orderid();
{
let mut book = self.trade_orders.lock().unwrap();
match book.get_mut(&key) {
Some((_, cur)) => {
cur.name.clone_from(&o.name);
cur.traded = o.traded;
cur.traded_price = o.traded_price;
cur.status = o.status;
cur.status_msg.clone_from(&o.status_msg);
cur.datetime = o.datetime;
}
None => {
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
book.insert(key, (seq, o.clone()));
}
}
}
self.trade_dirty.store(true, Ordering::Release);
self.bump();
}
pub fn note_trade(&self, t: &ek_data::trade::TradeData) {
let id = t.en_tradeid();
if self.trade_ids.lock().unwrap().contains(&id) {
return;
}
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
let dropped = {
let mut book = self.trade_trades.lock().unwrap();
book.push_front((seq, t.clone()));
self.trade_ids.lock().unwrap().insert(id);
let over = book.len().saturating_sub(TRADE_CAP);
for _ in 0..over {
if let Some((_, old)) = book.pop_back() {
self.trade_ids.lock().unwrap().remove(&old.en_tradeid());
}
}
over
};
if dropped > 0 {
let total = self
.trade_dropped
.fetch_add(dropped as u64, Ordering::AcqRel)
+ dropped as u64;
if total == dropped as u64 {
self.push_alert(
crate::toast::AlertLevel::Warning,
format!("成交表已达 {TRADE_CAP} 行上限:更早的成交不再显示"),
);
}
}
self.trade_dirty.store(true, Ordering::Release);
self.bump();
}
pub fn note_position(&self, p: &ek_data::position::PositionData) {
let key = p.en_positionid();
{
let mut book = self.trade_positions.lock().unwrap();
match book.get_mut(&key) {
Some((_, cur)) => {
cur.volume = p.volume;
cur.yd_volume = p.yd_volume;
cur.can_use_volume = p.can_use_volume;
cur.frozen = p.frozen;
cur.on_road_volume = p.on_road_volume;
cur.market_value = p.market_value;
cur.avg_price = p.avg_price;
cur.price = p.price;
cur.last_price = p.last_price;
cur.pnl = p.pnl;
cur.profit_rate = p.profit_rate;
cur.name.clone_from(&p.name);
}
None => {
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
book.insert(key, (seq, p.clone()));
}
}
}
self.trade_dirty.store(true, Ordering::Release);
self.bump();
}
pub fn note_funds(&self, a: &ek_data::account::AccountData) {
let key = a.en_accountid();
{
let mut book = self.trade_funds.lock().unwrap();
match book.get_mut(&key) {
Some((_, cur)) => {
cur.balance = a.balance;
cur.frozen = a.frozen;
}
None => {
let seq = self.trade_seq.fetch_add(1, Ordering::Relaxed) + 1;
book.insert(key, (seq, a.clone()));
}
}
}
self.trade_dirty.store(true, Ordering::Release);
self.bump();
}
pub fn clear_gateway_rows(&self, gateway: &str) -> usize {
let mut removed = 0usize;
{
let mut book = self.trade_orders.lock().unwrap();
let before = book.len();
book.retain(|_, (_, o)| o.gateway_name != gateway);
removed += before - book.len();
}
{
let mut book = self.trade_trades.lock().unwrap();
let before = book.len();
book.retain(|(_, t)| t.gateway_name != gateway);
removed += before - book.len();
let prefix = format!("{gateway}.");
self.trade_ids
.lock()
.unwrap()
.retain(|id| !id.starts_with(&prefix));
}
{
let mut book = self.trade_positions.lock().unwrap();
let before = book.len();
book.retain(|_, (_, p)| p.gateway_name != gateway);
removed += before - book.len();
}
{
let mut book = self.trade_funds.lock().unwrap();
let before = book.len();
book.retain(|_, (_, a)| a.gateway_name != gateway);
removed += before - book.len();
}
if removed > 0 {
self.trade_dirty.store(true, Ordering::Release);
self.bump();
}
removed
}
#[must_use]
pub fn trade_dropped(&self) -> u64 {
self.trade_dropped.load(Ordering::Acquire)
}
pub fn trade_snap(&self) -> Arc<TradeSnap> {
let mut cache = self.trade_snap.lock().unwrap();
if self.trade_dirty.swap(false, Ordering::AcqRel) {
*cache = Arc::new(self.build_trade_snap());
}
Arc::clone(&cache)
}
fn build_trade_snap(&self) -> TradeSnap {
fn by_seq<T>(rows: &BTreeMap<String, (u64, T)>) -> Vec<T>
where
T: Clone,
{
let mut v: Vec<(u64, &String, &T)> =
rows.iter().map(|(k, (s, v))| (*s, k, v)).collect();
v.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(b.1)));
v.into_iter().map(|(_, _, v)| v.clone()).collect()
}
TradeSnap {
orders: by_seq(&self.trade_orders.lock().unwrap()),
trades: self
.trade_trades
.lock()
.unwrap()
.iter()
.map(|(_, t)| t.clone())
.collect(),
positions: by_seq(&self.trade_positions.lock().unwrap()),
accounts: by_seq(&self.trade_funds.lock().unwrap()),
}
}
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();
}
pub fn row_selected(&self, table: &str, key: &str) -> bool {
self.row_sel
.lock()
.unwrap()
.get(table)
.is_some_and(|k| k == key)
}
pub fn select_row(&self, table: &str, key: &str) {
self.row_sel
.lock()
.unwrap()
.insert(table.to_string(), key.to_string());
}
}