use crate::driver::{self, MdDriver, MdEvent, MdOut, TdDriver, TdEvent, TdOut};
use crate::wire::InputOrder;
use ek_core::context::PluginContext;
use ek_core::envelope::{Envelope, ReplyErr, e_req_type};
use ek_core::plugin::Plugin;
use ek_core::section::{ApplyReport, Configurable, Field, SectionMap};
use ek_core::service::ServiceDesc;
use ek_data::cancel_request::CancelRequest;
use ek_data::contract::ContractData;
use ek_data::log::LogData;
#[cfg(feature = "native")]
use ek_data::log::LEVEL_ERROR;
use ek_data::log::LEVEL_INFO;
use ek_data::order_request::OrderRequest;
use ek_event::Event;
use std::borrow::Cow;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
const W_TICK: &str = "eTick.";
const W_ORDER: &str = "eOrder.";
const W_TRADE: &str = "eTrade.";
const W_POSITION: &str = "ePosition.";
const W_ACCOUNT: &str = "eAccount.";
const W_CONTRACT: &str = "eContract.";
pub const CONTRACT_BATCH_ROWS: usize = 2000;
const W_POS_DONE: &str = "ePositionDone.";
const W_ACCOUNT_DONE: &str = "eAccountDone.";
pub const EVENT_CONTRACT_FINISH: &str = "eContractFinish";
pub const SVC_STATUS: &str = "status";
use ek_data::event::EVENT_LOG as W_LOG;
pub use ek_event::EVENT_TIMER as ROUTE_TIMER;
#[derive(Debug, Clone)]
pub struct SubscribeReq {
pub symbols: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubscribeReply {
pub ok: bool,
pub msg: String,
}
#[derive(Debug, Clone)]
pub struct SendOrderReq {
pub req: OrderRequest,
}
#[derive(Debug, Clone, PartialEq)]
pub struct SendOrderReply {
pub ok: bool,
pub en_orderid: String,
pub msg: String,
}
#[derive(Debug, Clone)]
pub struct CancelOrderReq {
pub req: CancelRequest,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CancelOrderReply {
pub ok: bool,
pub msg: String,
}
#[derive(Debug, Clone)]
pub struct QueryRateReq {
pub kind: String,
pub instrument_id: String,
pub hedge_flag: String,
pub exchange_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueryRateReply {
pub ok: bool,
pub msg: String,
}
#[derive(Clone, Default)]
pub struct ConfigureReq {
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 product_classes: Vec<String>,
}
impl std::fmt::Debug for ConfigureReq {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ConfigureReq")
.field("broker_id", &self.broker_id)
.field("td_addr", &self.td_addr)
.field("td_password", &"****")
.field("td_auth_code", &self.td_auth_code)
.field("td_app_id", &self.td_app_id)
.field("md_addr", &self.md_addr)
.field("md_user", &self.md_user)
.field("md_password", &"****")
.field("product_classes", &self.product_classes)
.finish()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConfigureReply {
pub ok: bool,
pub msg: String,
pub product_classes: Vec<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct StatusReq;
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct GatewayStatus {
pub gateway: String,
pub sampled_ms: i64,
pub td_handle_up: bool,
pub td_logged_in: bool,
pub td_login_failed: bool,
pub td_auth_ok: bool,
pub td_auth_failed: bool,
pub td_front_id: i64,
pub td_session_id: i64,
pub td_disconnects: u32,
pub td_last_login_ok_ms: Option<i64>,
pub td_last_account_ms: Option<i64>,
pub td_last_position_ms: Option<i64>,
pub td_query_rejects: u32,
pub td_query_reject_streak: u32,
pub td_instrument_rejects: u32,
pub td_last_error: Option<String>,
pub md_handle_up: bool,
pub md_logged_in: bool,
pub md_disconnects: u32,
pub md_last_login_ok_ms: Option<i64>,
pub md_last_tick_ms: Option<i64>,
pub md_sub_rejects: u32,
pub md_last_error: Option<String>,
pub md_subscribed: usize,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GatewayChange {
TdConnected,
TdDisconnected,
TdAuthOk,
TdAuthFailed,
TdLoginOk,
TdLoginFailed,
SettlementConfirmed,
ContractArrived,
ContractQueryStopped,
QueryRejected,
MdConnected,
MdDisconnected,
MdLoginOk,
MdLoginFailed,
SubscribeRejected,
}
impl GatewayChange {
pub fn kind(&self) -> &'static str {
match self {
Self::TdConnected => "td-connected",
Self::TdDisconnected => "td-disconnected",
Self::TdAuthOk => "td-auth-ok",
Self::TdAuthFailed => "td-auth-failed",
Self::TdLoginOk => "td-login-ok",
Self::TdLoginFailed => "td-login-failed",
Self::SettlementConfirmed => "settlement-confirmed",
Self::ContractArrived => "contract-arrived",
Self::ContractQueryStopped => "contract-query-stopped",
Self::QueryRejected => "query-rejected",
Self::MdConnected => "md-connected",
Self::MdDisconnected => "md-disconnected",
Self::MdLoginOk => "md-login-ok",
Self::MdLoginFailed => "md-login-failed",
Self::SubscribeRejected => "subscribe-rejected",
}
}
pub fn describe(&self, gw: &str, st: &GatewayStatus) -> String {
let p = driver::CHAIN_STATE_PREFIX;
match self {
Self::TdConnected => format!("{p}{gw} 交易前置已连接(下一步:认证)"),
Self::TdDisconnected => format!(
"{p}{gw} 交易前置断开(累计 {} 次;会话作废,查询阶段回 Idle)",
st.td_disconnects
),
Self::TdAuthOk => format!("{p}{gw} 授权通过(下一步:登录)"),
Self::TdAuthFailed => format!(
"{p}{gw} 授权失败:{}",
st.td_last_error.as_deref().unwrap_or("无错误文本")
),
Self::TdLoginOk => format!(
"{p}{gw} 登录成功(Front {} / Session {})",
st.td_front_id, st.td_session_id
),
Self::TdLoginFailed => format!(
"{p}{gw} 登录失败:{}",
st.td_last_error.as_deref().unwrap_or("无错误文本")
),
Self::SettlementConfirmed => {
format!("{p}{gw} 结算单已确认(开始问合约)")
}
Self::ContractArrived => {
format!("{p}{gw} 合约已到手,开始资金/持仓轮询")
}
Self::ContractQueryStopped => {
format!("{p}{gw} 合约查询问满:本轮查询已停止(不再问资金与持仓)")
}
Self::QueryRejected => format!(
"{p}{gw} 查询被拒(连续 {} 次 / 累计 {} 次):{}",
st.td_query_reject_streak,
st.td_query_rejects,
st.td_last_error.as_deref().unwrap_or("无错误文本")
),
Self::MdConnected => format!("{p}{gw} 行情前置已连接(下一步:登录)"),
Self::MdDisconnected => format!("{p}{gw} 行情前置断开(累计 {} 次)", st.md_disconnects),
Self::MdLoginOk => format!(
"{p}{gw} 行情登录成功(已登记订阅 {} 项,自会重推)",
st.md_subscribed
),
Self::MdLoginFailed => format!(
"{p}{gw} 行情登录失败:{}",
st.md_last_error.as_deref().unwrap_or("无错误文本")
),
Self::SubscribeRejected => format!(
"{p}{gw} 行情订阅被拒(累计 {} 次):{}",
st.md_sub_rejects,
st.md_last_error.as_deref().unwrap_or("无错误文本")
),
}
}
}
#[derive(Debug, Clone)]
pub struct GatewayStatusEvent {
pub gateway: String,
pub at_ms: i64,
pub change: GatewayChange,
pub status: Box<GatewayStatus>,
}
impl GatewayStatusEvent {
pub fn describe(&self) -> String {
self.change.describe(&self.gateway, &self.status)
}
}
pub const EVENT_STATUS_PREFIX: &str = "eStatus.";
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct DisconnectReq;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DisconnectReply {
pub ok: bool,
pub msg: String,
}
pub trait CtpTransport: Send + Sync {
fn md_connect(&self, addr: &str, flow_path: &str);
fn md_login(&self, user: &str, password: &str, broker: &str);
fn md_subscribe(&self, symbols: &[String]);
fn td_connect(&self, addr: &str, flow_path: &str);
fn td_auth(&self, user: &str, broker: &str, auth_code: &str, app_id: &str);
fn td_login(&self, user: &str, password: &str, broker: &str);
fn td_settlement_confirm(&self, broker: &str, user: &str);
fn td_qry_instrument(&self) -> i32;
fn td_qry_account(&self, broker: &str, user: &str) -> i32;
fn td_qry_position(&self, broker: &str, user: &str) -> i32;
fn td_insert(&self, order: &InputOrder) -> i32;
fn td_cancel(&self, action: &crate::wire::InputOrderAction) -> i32;
fn td_qry_margin_rate(&self, query: &crate::wire::QryInstrumentMarginRate) -> i32;
fn td_qry_commission_rate(&self, query: &crate::wire::QryInstrumentCommissionRate) -> i32;
fn shutdown(&self) {}
fn td_release(&self) {}
fn md_release(&self) {}
}
#[derive(Clone, Default)]
pub struct CtpConfig {
pub gateway_name: String,
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 product_classes: crate::map::ProductClasses,
}
impl std::fmt::Debug for CtpConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CtpConfig")
.field("gateway_name", &self.gateway_name)
.field("broker_id", &self.broker_id)
.field("td_addr", &self.td_addr)
.field("td_user", &self.td_user)
.field("td_password", &"****")
.field("td_auth_code", &self.td_auth_code)
.field("td_app_id", &self.td_app_id)
.field("md_addr", &self.md_addr)
.field("md_user", &self.md_user)
.field("md_password", &"****")
.field("product_classes", &self.product_classes)
.finish()
}
}
impl CtpConfig {
pub fn gateway_name_or_default(&self) -> String {
if self.gateway_name.is_empty() {
"ctp".to_string()
} else {
self.gateway_name.clone()
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialEntry {
pub field: String,
pub state: String,
pub ref_key: String,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct CredentialStateReq;
#[must_use]
pub fn derived_gateway_name(broker_id: &str, account: &str) -> Option<String> {
let broker = sanitize_id_field(broker_id)?;
let account = sanitize_id_field(account)?;
let name = format!("ctp_{broker}_{account}");
if name.len() > 32 {
return None;
}
Some(name)
}
fn sanitize_id_field(raw: &str) -> Option<String> {
let trimmed = raw.trim();
if trimmed.is_empty() || !trimmed.chars().all(|c| c.is_ascii_alphanumeric()) {
return None;
}
Some(trimmed.to_ascii_lowercase())
}
#[must_use]
pub fn is_derived_gateway_name(name: &str) -> bool {
let Some(rest) = name.strip_prefix("ctp_") else {
return false;
};
if name.len() > 32 {
return false;
}
let ok = |s: &str| {
!s.is_empty()
&& s.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
};
match rest.split_once('_') {
Some((broker, account)) => ok(broker) && ok(account),
None => false,
}
}
fn lock_clone<T: Clone>(l: &std::sync::Mutex<T>) -> T {
match l.lock() {
Ok(g) => g.clone(),
Err(e) => e.into_inner().clone(),
}
}
pub struct CtpPlugin {
gw: String,
ns: String,
cfg_rev: AtomicU32,
sec_refs: Mutex<Vec<(String, String)>>,
cfg: Mutex<CtpConfig>,
broker: Mutex<String>,
user: Mutex<String>,
md: Mutex<MdDriver>,
td: Mutex<TdDriver>,
timer_count: Mutex<u32>,
trans: Option<Box<dyn CtpTransport>>,
ctx: OnceLock<PluginContext>,
trans_warned: Mutex<bool>,
save_fallback: Mutex<bool>,
contract_batch: Mutex<Vec<ContractData>>,
}
impl CtpPlugin {
pub fn new(cfg: CtpConfig) -> Self {
let gw = cfg.gateway_name_or_default();
Self::with_parts(cfg, gw, None, None)
}
pub fn with_transport(cfg: CtpConfig, trans: Box<dyn CtpTransport>) -> Self {
let gw = cfg.gateway_name_or_default();
Self::with_parts(cfg, gw, Some(trans), None)
}
pub fn with_transport_and_clock(
cfg: CtpConfig,
trans: Box<dyn CtpTransport>,
now_ms: Arc<dyn Fn() -> u64 + Send + Sync>,
) -> Self {
let gw = cfg.gateway_name_or_default();
Self::with_parts(cfg, gw, Some(trans), Some(now_ms))
}
fn with_parts(
cfg: CtpConfig,
gw: String,
trans: Option<Box<dyn CtpTransport>>,
clock: Option<Arc<dyn Fn() -> u64 + Send + Sync>>,
) -> Self {
let contracts = driver::contracts_empty();
let cfg_broker = cfg.broker_id.clone();
let cfg_user = cfg.td_user.clone();
let md_cfg = driver::MdConfig {
addr: cfg.md_addr.clone(),
user: cfg.md_user.clone(),
password: cfg.md_password.clone(),
broker: cfg.broker_id.clone(),
};
let td_cfg = driver::TdConfig {
addr: cfg.td_addr.clone(),
user: cfg.td_user.clone(),
password: cfg.td_password.clone(),
broker: cfg.broker_id.clone(),
auth_code: cfg.td_auth_code.clone(),
app_id: cfg.td_app_id.clone(),
product_classes: cfg.product_classes.clone(),
};
let md = Mutex::new(MdDriver::new(
&gw,
md_cfg,
contracts.clone(),
crate::map::shanghai_now_date(),
));
let td = Mutex::new(match clock {
Some(f) => TdDriver::with_clock(&gw, td_cfg, contracts.clone(), f),
None => TdDriver::new(&gw, td_cfg, contracts.clone()),
});
let ns = if gw == "ctp" {
"ctp".to_owned()
} else {
format!("ctp.{gw}")
};
Self {
gw,
ns,
cfg_rev: AtomicU32::new(0),
sec_refs: Mutex::new(Vec::new()),
cfg: Mutex::new(cfg),
broker: Mutex::new(cfg_broker),
user: Mutex::new(cfg_user),
md,
td,
timer_count: Mutex::new(0),
trans,
ctx: OnceLock::new(),
trans_warned: Mutex::new(false),
save_fallback: Mutex::new(false),
contract_batch: Mutex::new(Vec::with_capacity(CONTRACT_BATCH_ROWS)),
}
}
#[cfg(feature = "native")]
pub fn with_native(cfg: CtpConfig, flow_root: impl Into<std::path::PathBuf>) -> Arc<Self> {
let (trans, ingress) = crate::native::transport::native_parts(flow_root);
let plugin = Arc::new(Self::with_transport(cfg, trans));
crate::native::transport::attach_plugin(&ingress, &plugin);
plugin
}
pub fn connect(&self) {
let (td_empty, md_empty) = {
let c = self.cfg.lock().unwrap();
(c.td_addr.is_empty(), c.md_addr.is_empty())
};
if td_empty && md_empty {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: "未配置前置地址,跳过连接(等待 configure)".to_string(),
level: 20,
});
return;
}
let td_outs = if td_empty {
Vec::new()
} else {
self.td.lock().unwrap().connect()
};
let md_outs = if md_empty {
Vec::new()
} else {
self.md.lock().unwrap().connect()
};
for o in td_outs {
self.emit_td(o);
}
for o in md_outs {
self.emit_md(o);
}
}
#[cfg(feature = "native")]
pub(crate) fn transport_log(&self, msg: String) {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg,
level: LEVEL_ERROR,
});
}
pub fn on_md_event(&self, ev: MdEvent) {
let outs = self.md.lock().unwrap().step(ev);
for o in outs {
self.emit_md(o);
}
}
pub fn on_td_event(&self, ev: TdEvent) {
let outs = self.td.lock().unwrap().step(ev);
for o in outs {
self.emit_td(o);
}
}
pub fn on_timer(&self) {
let outs = self.td.lock().unwrap().on_timer();
for o in outs {
self.emit_td(o);
}
let gate_pass = {
let mut c = self.timer_count.lock().unwrap();
*c += 1;
if *c < 2 {
false
} else {
*c = 0;
true
}
};
if !gate_pass {
return;
}
let date = crate::map::shanghai_now_date();
self.md.lock().unwrap().update_date(date);
}
fn emit_md(&self, o: MdOut) {
match o {
MdOut::Log(l) => self.put_log(l),
MdOut::Status(ch) => self.publish_status(ch),
MdOut::Tick(t) => self.put_data(W_TICK, t),
MdOut::ReqConnect { addr, flow_path } => {
if let Some(t) = &self.trans {
t.md_connect(&addr, &flow_path);
} else {
self.warn_no_transport();
}
}
MdOut::ReqLogin {
user,
password,
broker,
} => {
if let Some(t) = &self.trans {
t.md_login(&user, &password, &broker);
} else {
self.warn_no_transport();
}
}
MdOut::ReqSubscribe(symbols) => {
if let Some(t) = &self.trans {
t.md_subscribe(&symbols);
} else {
self.warn_no_transport();
}
}
}
}
fn emit_td(&self, o: TdOut) {
match o {
TdOut::Log(l) => self.put_log(l),
TdOut::Status(ch) => self.publish_status(ch),
TdOut::Contract(c) => self.emit_contract(c),
TdOut::Order(o) => self.publish_enriched(W_ORDER, o),
TdOut::Trade(t) => self.publish_enriched(W_TRADE, t),
TdOut::Position(p) => self.publish_enriched(W_POSITION, p),
TdOut::Account(a) => self.put_data(W_ACCOUNT, a),
TdOut::PositionQueryDone(gw) => {
if let Some(ctx) = self.ctx.get() {
let _ = ctx.put(Event::with_data(W_POS_DONE, Arc::new(gw)));
}
}
TdOut::AccountQueryDone(gw) => {
if let Some(ctx) = self.ctx.get() {
let _ = ctx.put(Event::with_data(W_ACCOUNT_DONE, Arc::new(gw)));
}
}
TdOut::ContractFinish(gw) => {
self.flush_contract();
if let Some(ctx) = self.ctx.get() {
let _ = ctx.put(Event::with_data(EVENT_CONTRACT_FINISH, Arc::new(gw)));
}
}
TdOut::ReqConnect { addr, flow_path } => {
if let Some(t) = &self.trans {
t.td_connect(&addr, &flow_path);
} else {
self.warn_no_transport();
}
}
TdOut::ReqAuthenticate {
user,
broker,
auth_code,
app_id,
} => {
if let Some(t) = &self.trans {
t.td_auth(&user, &broker, &auth_code, &app_id);
} else {
self.warn_no_transport();
}
}
TdOut::ReqLogin {
user,
password,
broker,
} => {
if let Some(t) = &self.trans {
t.td_login(&user, &password, &broker);
} else {
self.warn_no_transport();
}
}
TdOut::ReqSettlementConfirm { broker, user } => {
if let Some(t) = &self.trans {
t.td_settlement_confirm(&broker, &user);
} else {
self.warn_no_transport();
}
}
TdOut::ReqQryInstrument => {
if let Some(t) = &self.trans {
let n = t.td_qry_instrument();
for l in self.td.lock().unwrap().note_instrument_send_failure(n) {
self.put_log(l);
}
} else {
self.warn_no_transport();
}
}
TdOut::ReqQryAccount { broker, user } => {
if let Some(t) = &self.trans {
let n = t.td_qry_account(&broker, &user);
for l in self.td.lock().unwrap().note_query_send_failure("资金", n) {
self.put_log(l);
}
} else {
self.warn_no_transport();
}
}
TdOut::ReqQryPosition { broker, user } => {
if let Some(t) = &self.trans {
let n = t.td_qry_position(&broker, &user);
for l in self.td.lock().unwrap().note_query_send_failure("持仓", n) {
self.put_log(l);
}
} else {
self.warn_no_transport();
}
}
TdOut::Insert(o) => {
if let Some(t) = &self.trans {
t.td_insert(&o);
} else {
self.warn_no_transport();
}
}
TdOut::Cancel(a) => {
if let Some(t) = &self.trans {
t.td_cancel(&a);
} else {
self.warn_no_transport();
}
}
}
}
fn put_log(&self, l: LogData) {
if let Some(ctx) = self.ctx.get() {
let _ = ctx.put(Event::with_data(W_LOG, Arc::new(l)));
}
}
fn put_data<T: Send + Sync + 'static>(&self, prefix: &str, payload: T) {
if let Some(ctx) = self.ctx.get() {
let _ = ctx.put(Event::with_data(prefix, Arc::new(payload)));
}
}
fn emit_contract(&self, c: ContractData) {
let mut batch = self.contract_batch.lock().unwrap();
batch.push(c);
if batch.len() >= CONTRACT_BATCH_ROWS {
let payload = std::mem::take(&mut *batch);
drop(batch);
self.put_data(W_CONTRACT, payload);
}
}
pub fn flush_contract(&self) {
let payload = {
let mut batch = self.contract_batch.lock().unwrap();
std::mem::take(&mut *batch)
};
if !payload.is_empty() {
self.put_data(W_CONTRACT, payload);
}
}
fn publish_enriched<T>(&self, line: &str, mut row: T)
where
T: EnrichName + Send + Sync + 'static,
{
if row.name_mut().is_empty() {
let symbol = row.symbol().to_string();
if let Some(n) = self.td.lock().unwrap().contract_name(&symbol) {
*row.name_mut() = n;
}
}
self.put_data(line, row);
}
fn publish_status(&self, change: GatewayChange) {
let st = self.status();
self.put_data(
EVENT_STATUS_PREFIX,
GatewayStatusEvent {
gateway: self.gw.clone(),
at_ms: st.sampled_ms,
change,
status: Box::new(st),
},
);
}
fn warn_no_transport(&self) {
let mut warned = self.trans_warned.lock().unwrap();
if !*warned {
*warned = true;
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: "传输层未接入(默认构建):连接/查询/发送动作已被吸收".to_string(),
level: 20,
});
}
}
fn on_request(&self, event: &Event) {
let Some(env) = event.data.downcast_ref::<Envelope>() else {
return;
};
macro_rules! arm {
($ty:ty, $res:ty, $body:expr) => {
if let Some(req) = env.payload.downcast_ref::<$ty>() {
let res: $res = $body(req);
let _ = env.reply.send(Box::new(Ok::<$res, ReplyErr>(res)));
return;
}
};
}
arm!(SubscribeReq, SubscribeReply, |r: &SubscribeReq| self
.do_subscribe(r));
arm!(SendOrderReq, SendOrderReply, |r: &SendOrderReq| self
.do_send_order(r));
arm!(CancelOrderReq, CancelOrderReply, |r: &CancelOrderReq| self
.do_cancel_order(r));
arm!(QueryRateReq, QueryRateReply, |r: &QueryRateReq| self
.do_query_rate(r));
arm!(ConfigureReq, ConfigureReply, |r: &ConfigureReq| self
.do_configure(r));
arm!(StatusReq, GatewayStatus, |r: &StatusReq| self.do_status(r));
arm!(DisconnectReq, DisconnectReply, |r: &DisconnectReq| self
.do_disconnect(r));
arm!(
CredentialStateReq,
Vec<CredentialEntry>,
|_: &CredentialStateReq| self.credentials_state()
);
let _ = env
.reply
.send(Box::new(Err::<(), ReplyErr>(ReplyErr::InvalidPayload)));
}
fn do_subscribe(&self, r: &SubscribeReq) -> SubscribeReply {
let mut outs = Vec::new();
{
let mut md = self.md.lock().unwrap();
for s in &r.symbols {
outs.extend(md.subscribe(s));
}
}
for o in outs {
self.emit_md(o);
}
SubscribeReply {
ok: true,
msg: String::new(),
}
}
fn do_send_order(&self, r: &SendOrderReq) -> SendOrderReply {
let outcome = {
let mut td = self.td.lock().unwrap();
td.send_order(&r.req)
};
if let Some(log) = &outcome.error {
self.put_log(log.clone());
}
let mut submit_order: Option<ek_data::order::OrderData> = None;
let mut send_failed: Option<String> = None;
for o in outcome.outs {
match o {
TdOut::Insert(input) => match &self.trans {
Some(t) => {
let n = t.td_insert(&input);
if n != 0 {
send_failed = Some(format!("委托请求发送失败,错误代码:{n}"));
}
}
None => self.warn_no_transport(),
},
TdOut::Order(od) => submit_order = Some(od),
other => self.emit_td(other),
}
}
if let Some(msg) = send_failed {
let log = LogData {
gateway_name: self.gw.clone(),
msg: msg.clone(),
level: 20,
};
self.put_log(log);
return SendOrderReply {
ok: false,
en_orderid: String::new(),
msg,
};
}
if let Some(od) = submit_order {
self.emit_td(TdOut::Order(od));
}
if outcome.en_orderid.is_empty() {
let msg = outcome.error.map(|l| l.msg).unwrap_or_default();
SendOrderReply {
ok: false,
en_orderid: String::new(),
msg,
}
} else if self.trans.is_none() {
self.warn_no_transport();
SendOrderReply {
ok: false,
en_orderid: outcome.en_orderid,
msg: "传输层未接入(默认构建)".to_string(),
}
} else {
SendOrderReply {
ok: true,
en_orderid: outcome.en_orderid,
msg: String::new(),
}
}
}
fn do_query_rate(&self, r: &QueryRateReq) -> QueryRateReply {
let broker = self.broker.lock().unwrap().clone();
let user = self.user.lock().unwrap().clone();
let is_margin = r.kind == "margin";
let is_commission = r.kind == "commission";
if !is_margin && !is_commission {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: format!("费率查询类型未知:{}", r.kind),
level: 20,
});
return QueryRateReply {
ok: false,
msg: format!("费率查询类型未知:{}", r.kind),
};
}
if let Some(t) = &self.trans {
let n = if is_margin {
let query = crate::wire::QryInstrumentMarginRate {
broker_id: broker.clone(),
investor_id: user.clone(),
hedge_flag: r.hedge_flag.clone(),
exchange_id: r.exchange_id.clone(),
invest_unit_id: String::new(),
instrument_id: r.instrument_id.clone(),
};
t.td_qry_margin_rate(&query)
} else {
let query = crate::wire::QryInstrumentCommissionRate {
broker_id: broker.clone(),
investor_id: user.clone(),
exchange_id: r.exchange_id.clone(),
invest_unit_id: String::new(),
instrument_id: r.instrument_id.clone(),
};
t.td_qry_commission_rate(&query)
};
let label = if is_margin {
"保证金率"
} else {
"手续费率"
};
if n != 0 {
let msg =
format!("{label}查询请求发送失败,错误代码:{n}(不自动重发:费率查询按需调用)");
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: msg.clone(),
level: 20,
});
return QueryRateReply { ok: false, msg };
}
QueryRateReply {
ok: true,
msg: String::new(),
}
} else {
self.warn_no_transport();
QueryRateReply {
ok: false,
msg: "传输层未接入(默认构建)".to_string(),
}
}
}
fn do_cancel_order(&self, r: &CancelOrderReq) -> CancelOrderReply {
let outs = {
let mut td = self.td.lock().unwrap();
td.cancel_order(&r.req)
};
let mut rejected_log: Option<LogData> = None;
let outs: Vec<TdOut> = outs
.into_iter()
.filter(|o| match o {
TdOut::Log(l) => {
rejected_log = Some(l.clone());
false
}
_ => true,
})
.collect();
if let Some(l) = rejected_log {
let msg = l.msg.clone();
self.put_log(l);
return CancelOrderReply { ok: false, msg };
}
for o in outs {
self.emit_td(o);
}
if self.trans.is_none() {
self.warn_no_transport();
return CancelOrderReply {
ok: false,
msg: "传输层未接入(默认构建)".to_string(),
};
}
CancelOrderReply {
ok: true,
msg: String::new(),
}
}
fn do_configure(&self, r: &ConfigureReq) -> ConfigureReply {
self.configure(r)
}
fn do_status(&self, _r: &StatusReq) -> GatewayStatus {
self.status()
}
fn do_disconnect(&self, _r: &DisconnectReq) -> DisconnectReply {
self.disconnect()
}
pub fn status(&self) -> GatewayStatus {
let td = self.td.lock().unwrap().status_view();
let md = self.md.lock().unwrap().status_view();
GatewayStatus {
gateway: self.gw.clone(),
sampled_ms: crate::map::now_epoch_ms(),
td_handle_up: td.handle_up,
td_logged_in: td.logged_in,
td_login_failed: td.login_failed,
td_auth_ok: td.auth_ok,
td_auth_failed: td.auth_failed,
td_front_id: td.front_id,
td_session_id: td.session_id,
td_disconnects: td.disconnects,
td_last_login_ok_ms: td.last_login_ok_ms,
td_last_account_ms: td.last_account_ms,
td_last_position_ms: td.last_position_ms,
td_query_rejects: td.query_rejects,
td_query_reject_streak: td.query_reject_streak,
td_instrument_rejects: td.instrument_rejects,
td_last_error: td.last_error,
md_handle_up: md.handle_up,
md_logged_in: md.logged_in,
md_disconnects: md.disconnects,
md_last_login_ok_ms: md.last_login_ok_ms,
md_last_tick_ms: md.last_tick_ms,
md_sub_rejects: md.sub_rejects,
md_last_error: md.last_error,
md_subscribed: md.subscribed,
}
}
pub fn disconnect(&self) -> DisconnectReply {
if let Some(t) = &self.trans {
t.td_release();
t.md_release();
}
{
self.td.lock().unwrap().reset_session();
}
{
self.md.lock().unwrap().reset_session();
}
self.flush_contract();
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: "会话已停(实例与订阅保留;『连接』可恢复)".to_string(),
level: LEVEL_INFO,
});
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: format!("{}会话已复位", driver::CHAIN_STATE_PREFIX),
level: LEVEL_INFO,
});
DisconnectReply {
ok: true,
msg: "会话已停(实例与订阅保留;『连接』可恢复)".to_string(),
}
}
fn effective_classes(&self) -> Vec<String> {
self.td.lock().unwrap().product_classes()
}
pub fn configure(&self, r: &ConfigureReq) -> ConfigureReply {
if r.td_addr.is_empty() && r.md_addr.is_empty() {
return ConfigureReply {
ok: false,
msg: "配置校验: td_addr/md_addr 至少一个非空".to_string(),
product_classes: self.effective_classes(),
};
}
if !r.td_addr.is_empty()
&& (r.broker_id.is_empty() || r.td_user.is_empty() || r.td_password.is_empty())
{
return ConfigureReply {
ok: false,
msg: "配置校验: TD 侧需 broker_id/td_user/td_password 齐".to_string(),
product_classes: self.effective_classes(),
};
}
let md_user = if !r.md_user.is_empty() {
r.md_user.clone()
} else {
r.td_user.clone()
};
let md_password = if !r.md_password.is_empty() {
r.md_password.clone()
} else {
r.td_password.clone()
};
if !r.md_addr.is_empty() && (md_user.is_empty() || md_password.is_empty()) {
return ConfigureReply {
ok: false,
msg: "配置校验: MD 侧需 md_user/md_password(或 td 侧回落)".to_string(),
product_classes: self.effective_classes(),
};
}
let bad_codes: Vec<String> = r
.product_classes
.iter()
.map(|c| c.trim().to_string())
.filter(|c| !c.is_empty() && c.chars().count() != 1)
.collect();
if !bad_codes.is_empty() {
return ConfigureReply {
ok: false,
msg: format!(
"配置校验: product_classes 需为单字符 CTP 码(非法项:{})",
bad_codes.join(",")
),
product_classes: self.effective_classes(),
};
}
let classes = crate::map::ProductClasses(r.product_classes.clone());
let effective_classes = classes.normalize();
{
self.td.lock().unwrap().reconfigure(driver::TdConfig {
addr: r.td_addr.clone(),
user: r.td_user.clone(),
password: r.td_password.clone(),
broker: r.broker_id.clone(),
auth_code: r.td_auth_code.clone(),
app_id: r.td_app_id.clone(),
product_classes: classes.clone(),
});
}
{
self.md.lock().unwrap().reconfigure(driver::MdConfig {
addr: r.md_addr.clone(),
user: md_user,
password: md_password,
broker: r.broker_id.clone(),
});
}
*self.broker.lock().unwrap() = r.broker_id.clone();
*self.user.lock().unwrap() = r.td_user.clone();
*self.cfg.lock().unwrap() = CtpConfig {
gateway_name: self.gw.clone(),
broker_id: r.broker_id.clone(),
td_addr: r.td_addr.clone(),
td_user: r.td_user.clone(),
td_password: r.td_password.clone(),
td_auth_code: r.td_auth_code.clone(),
td_app_id: r.td_app_id.clone(),
md_addr: r.md_addr.clone(),
md_user: r.md_user.clone(),
md_password: r.md_password.clone(),
product_classes: classes,
};
if let Some(t) = &self.trans {
t.td_release();
t.md_release();
}
self.connect();
self.persist_config();
ConfigureReply {
ok: true,
msg: format!(
"已配置并发起连接(先 TD 后 MD);收录产品类:{}",
effective_classes.join(" ")
),
product_classes: effective_classes,
}
}
#[must_use]
pub fn credentials_state(&self) -> Vec<CredentialEntry> {
let refs = lock_clone(&self.sec_refs);
let cfg = lock_clone(&self.cfg);
let mut out = Vec::with_capacity(2);
for field in ["td_password", "md_password"] {
let sk = refs
.iter()
.find(|(f, _)| f == field)
.map(|(_, k)| k.clone())
.unwrap_or_default();
let resolved = if field == "td_password" {
!cfg.td_password.is_empty()
} else {
!cfg.md_password.is_empty()
};
let state = if resolved {
"RESOLVED"
} else if sk.is_empty() {
"ABSENT"
} else {
"REFERENCED"
};
out.push(CredentialEntry {
field: field.to_owned(),
state: state.to_owned(),
ref_key: sk,
});
}
out
}
fn resolve_pending_secrets(&self) {
let refs = lock_clone(&self.sec_refs);
if refs.is_empty() {
return;
}
let mut cfg = lock_clone(&self.cfg);
let mut hit = false;
for (field, sk) in &refs {
match ek_cfg::secret::get(&self.ns, sk) {
Some(v) => match std::str::from_utf8(v.as_bytes()) {
Ok(text) => {
if field == "td_password" {
cfg.td_password = text.to_owned();
hit = true;
} else if field == "md_password" {
cfg.md_password = text.to_owned();
hit = true;
}
}
Err(_) => {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: format!("{field} 的秘值不是文本(键 {sk}),按待手配处理"),
level: 40,
});
}
},
None => {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: format!("{field} 的引用在库值缺(键 {sk}),待手配"),
level: 20,
});
}
}
}
if !hit {
return;
}
let td_cfg = driver::TdConfig {
addr: cfg.td_addr.clone(),
user: cfg.td_user.clone(),
password: cfg.td_password.clone(),
broker: cfg.broker_id.clone(),
auth_code: cfg.td_auth_code.clone(),
app_id: cfg.td_app_id.clone(),
product_classes: cfg.product_classes.clone(),
};
let md_cfg = driver::MdConfig {
addr: cfg.md_addr.clone(),
user: if cfg.md_user.is_empty() {
cfg.td_user.clone()
} else {
cfg.md_user.clone()
},
password: if cfg.md_password.is_empty() {
cfg.td_password.clone()
} else {
cfg.md_password.clone()
},
broker: cfg.broker_id.clone(),
};
self.td.lock().unwrap().reconfigure(td_cfg);
self.md.lock().unwrap().reconfigure(md_cfg);
*self.cfg.lock().unwrap() = cfg;
}
fn persist_config(&self) {
let map = self.dump_config();
if map.is_empty() {
return; }
let ns = Configurable::config_ns(self).to_owned();
let req = ek_cfg::wire::SaveReq {
ns: ns.clone(),
map: map.clone(),
};
let ctx = self.ctx.get();
let res = match ctx {
Some(c) => c.ask_service::<ek_cfg::wire::SaveReq, ek_cfg::wire::SaveRes>(
"cfg.save",
req,
std::time::Duration::from_millis(1500),
),
None => Err(ek_core::error::AskError::UnknownService {
service: "cfg.save".to_owned(),
}),
};
match res {
Ok(r) if r.ok => {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: format!("配置已持久化(段 {});警告{:?}", ns, r.warns),
level: 20,
});
}
Ok(r) => {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: format!("配置落盘失败(段 {}):{}", ns, r.msg),
level: 40,
});
}
Err(_) => {
let once = {
let mut f = self.save_fallback.lock().unwrap();
let first = !*f;
*f = true;
first
};
if once {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: "cfg 缺席,已直写落盘".to_string(),
level: 20,
});
}
let ok = Self::store_direct(&ns, &map);
if !ok {
self.put_log(LogData {
gateway_name: self.gw.clone(),
msg: format!("直写托底也失败(段 {})", ns),
level: 40,
});
}
}
}
}
fn store_direct(ns: &str, map: &SectionMap) -> bool {
let store = match std::env::var("EK_CFG_FALLBACK_ROOT") {
Ok(v) if !v.is_empty() => ek_cfg::Store::with_root(std::path::PathBuf::from(v)),
_ => ek_cfg::Store::eco()
.unwrap_or_else(|_| ek_cfg::Store::with_root("./ekcfg-local".into())),
};
store.store(ns, map).is_ok()
}
}
impl Configurable for CtpPlugin {
fn config_ns(&self) -> &str {
&self.ns
}
fn apply_config(&self, m: &SectionMap) -> Result<ApplyReport, String> {
if m.rev > self.cfg_rev.load(Ordering::Relaxed) {
self.cfg_rev.store(m.rev, Ordering::Relaxed);
}
let mut rep = ApplyReport::all_adopted();
let mut patch = lock_clone(&self.cfg);
let mut new_refs: Vec<(String, String)> = Vec::new();
for (k, f) in m.iter() {
match k {
"broker_id" => {
patch.broker_id = f.plain_value().unwrap_or("").to_owned();
rep.adopted.push(k.to_owned());
}
"td_addr" => {
patch.td_addr = f.plain_value().unwrap_or("").to_owned();
rep.adopted.push(k.to_owned());
}
"td_user" => {
patch.td_user = f.plain_value().unwrap_or("").to_owned();
rep.adopted.push(k.to_owned());
}
"td_auth_code" => {
patch.td_auth_code = f.plain_value().unwrap_or("").to_owned();
rep.adopted.push(k.to_owned());
}
"td_app_id" => {
patch.td_app_id = f.plain_value().unwrap_or("").to_owned();
rep.adopted.push(k.to_owned());
}
"md_addr" => {
patch.md_addr = f.plain_value().unwrap_or("").to_owned();
rep.adopted.push(k.to_owned());
}
"md_user" => {
patch.md_user = f.plain_value().unwrap_or("").to_owned();
rep.adopted.push(k.to_owned());
}
"product_classes" => {
patch.product_classes = crate::map::ProductClasses(
f.plain_value()
.unwrap_or("")
.split(',')
.map(str::trim)
.filter(|c| !c.is_empty())
.map(str::to_owned)
.collect(),
);
rep.adopted.push(k.to_owned());
}
"td_password" | "md_password" => {
if let Some(sk) = f.secret_key() {
new_refs.push((k.to_owned(), sk.to_owned())); rep.adopted.push(k.to_owned());
} else {
rep.deferred.push(k.to_owned()); }
}
_ => rep.deferred.push(k.to_owned()), }
}
*self.sec_refs.lock().unwrap() = new_refs;
let td_cfg = driver::TdConfig {
addr: patch.td_addr.clone(),
user: patch.td_user.clone(),
password: patch.td_password.clone(),
broker: patch.broker_id.clone(),
auth_code: patch.td_auth_code.clone(),
app_id: patch.td_app_id.clone(),
product_classes: patch.product_classes.clone(),
};
let md_cfg = driver::MdConfig {
addr: patch.md_addr.clone(),
user: if patch.md_user.is_empty() {
patch.td_user.clone()
} else {
patch.md_user.clone()
},
password: if patch.md_password.is_empty() {
patch.td_password.clone()
} else {
patch.md_password.clone()
},
broker: patch.broker_id.clone(),
};
self.td
.lock()
.map_err(|e| e.to_string())?
.reconfigure(td_cfg);
self.md
.lock()
.map_err(|e| e.to_string())?
.reconfigure(md_cfg);
*self.broker.lock().unwrap() = patch.broker_id.clone();
*self.user.lock().unwrap() = patch.td_user.clone();
*self.cfg.lock().unwrap() = patch;
Ok(rep)
}
fn dump_config(&self) -> SectionMap {
let next = self.cfg_rev.load(Ordering::Relaxed) + 1;
self.cfg_rev.store(next, Ordering::Relaxed);
let cfg = lock_clone(&self.cfg);
let mut map = SectionMap::new(next);
for (k, v) in [
("broker_id", &cfg.broker_id),
("td_addr", &cfg.td_addr),
("td_user", &cfg.td_user),
("td_auth_code", &cfg.td_auth_code),
("td_app_id", &cfg.td_app_id),
("md_addr", &cfg.md_addr),
("md_user", &cfg.md_user),
] {
if !v.is_empty() {
map = map.set(k, Field::plain(v));
}
}
if !cfg.product_classes.0.is_empty() {
map = map.set(
"product_classes",
Field::plain(cfg.product_classes.0.join(",")),
);
}
for (field, value) in [
("td_password", &cfg.td_password),
("md_password", &cfg.md_password),
] {
let refs = lock_clone(&self.sec_refs);
if let Some((_, sk)) = refs.iter().find(|(f, _)| f == field) {
if !value.is_empty() {
let _ = ek_cfg::secret::set(
&self.ns,
sk,
ek_cfg::SecretValue::from_bytes(value.clone()),
);
}
map = map.set(field, Field::secret(sk.as_str()));
} else if !value.is_empty() {
let sk = format!("gw_{field}");
if ek_cfg::secret::set(
&self.ns,
&sk,
ek_cfg::SecretValue::from_bytes(value.clone()),
)
.is_ok()
{
self.sec_refs
.lock()
.unwrap()
.push((field.to_owned(), sk.clone()));
map = map.set(field, Field::secret(sk.as_str()));
}
}
}
map
}
}
impl Default for CtpPlugin {
fn default() -> Self {
Self::new(CtpConfig::default())
}
}
impl Plugin for CtpPlugin {
fn name(&self) -> &str {
&self.gw
}
fn routes(&self) -> Vec<Cow<'static, str>> {
vec![Cow::Owned(e_req_type(&self.gw)), Cow::Borrowed(ROUTE_TIMER)]
}
fn version(&self) -> Cow<'static, str> {
Cow::Borrowed(env!("CARGO_PKG_VERSION"))
}
fn services(&self) -> Vec<ServiceDesc> {
vec![
ServiceDesc::describe::<SubscribeReq, SubscribeReply>(
format!("{}.subscribe", self.gw),
"登记并(已登录时)下发行情订阅",
),
ServiceDesc::describe::<SendOrderReq, SendOrderReply>(
format!("{}.send_order", self.gw),
"下单(InputOrder + SUBMITTING 单已上总线)",
),
ServiceDesc::describe::<CancelOrderReq, CancelOrderReply>(
format!("{}.cancel_order", self.gw),
"撤单(三段订单号解析)",
),
ServiceDesc::describe::<QueryRateReq, QueryRateReply>(
format!("{}.query_rate", self.gw),
"费率查询(margin=保证金率/commission=手续费率;状态为纯被动推送)",
),
ServiceDesc::describe::<ConfigureReq, ConfigureReply>(
format!("{}.configure", self.gw),
"网关配置(插件自带配置面;手工输入账号/前置 → 复位守卫并发起连接)",
),
ServiceDesc::describe::<StatusReq, GatewayStatus>(
format!("{}.{}", self.gw, SVC_STATUS),
"状态读数(0.9.0;登录态 / 断开计数 / 查询被拒计数 / 最近回报时刻;只读、不写日志)",
),
ServiceDesc::describe::<DisconnectReq, DisconnectReply>(
format!("{}.disconnect", self.gw),
"停会话(0.9.0;实例与订阅保留,『连接』可恢复)",
),
ServiceDesc::describe::<CredentialStateReq, Vec<CredentialEntry>>(
format!("{}.credentials_state", self.gw),
"凭据档位(REFERENCED/RESOLVED/ABSENT;只读)",
),
]
}
fn on_init(&self, ctx: &PluginContext) {
let _ = self.ctx.set(ctx.clone());
}
fn on_start(&self) {
self.resolve_pending_secrets();
self.connect();
}
fn on_stop(&self) {
if let Some(t) = &self.trans {
t.shutdown();
}
self.flush_contract();
}
fn on_mail(&self, event: &Event) {
if event.event_type == ROUTE_TIMER {
self.on_timer();
return;
}
if event.event_type == e_req_type(&self.gw) {
self.on_request(event);
}
}
fn as_configurable(&self) -> Option<&dyn Configurable> {
Some(self)
}
}
trait EnrichName {
fn symbol(&self) -> &str;
fn name_mut(&mut self) -> &mut String;
}
impl EnrichName for ek_data::order::OrderData {
fn symbol(&self) -> &str {
&self.symbol
}
fn name_mut(&mut self) -> &mut String {
&mut self.name
}
}
impl EnrichName for ek_data::trade::TradeData {
fn symbol(&self) -> &str {
&self.symbol
}
fn name_mut(&mut self) -> &mut String {
&mut self.name
}
}
impl EnrichName for ek_data::position::PositionData {
fn symbol(&self) -> &str {
&self.symbol
}
fn name_mut(&mut self) -> &mut String {
&mut self.name
}
}