USER
#include <thread>
#include "TradeClient.h"
#include "Msg.h"
#include "message_generated.h"
#include "Utils/PackMsg.h"
#include "Utils/ZmqHelper.hpp"
#include "Utils/Logger.h"
#include "Utils/Config.h"
#include "Utils/TimeUtils.hpp"
#include <ctime>
#include <iomanip>
#include <iostream>
#include <iomanip>
#include <sstream>
#include <chrono>
TradeClient::TradeClient()
:m_pSendTradeReqContext(nullptr)
,m_pSendTradeReqSocket(nullptr)
,m_pRecvTradeRtnContext(nullptr)
,m_pRecvTradeRtnSocket(nullptr)
,m_pSendQueryReqContext(nullptr)
,m_pSendQueryReqSocket(nullptr)
,m_pRecvQueryRtnContext(nullptr)
,m_pRecvQueryRtnSocket(nullptr)
{
m_tcp_keep_alive = 1;
m_tcp_keep_idle = 30;
m_tcp_keep_intvl = 5;
m_tcp_keep_cnt = 5;
m_router_handover = 1;
m_heartbeat_ivl = 1000;
m_heartbeat_timeout = 3000;
m_heartbeat_ttl = 6000;
m_last_heartbeat.store(-1);
m_gateway_is_live.store(false);
}
TradeClient::~TradeClient()
{
}
void TradeClient::handleLoginRsp(FbMsg::LoginRspT *loginRspT)
{
Logger::getInstance().log->debug("TradeClient::handleLoginRsp");
LoginRsp loginRsp;
loginRsp.AccountId = loginRspT->AccountId;
loginRsp.ChannelId = loginRspT->ChannelId;
loginRsp.LoginTime = loginRspT->LoginTime;
loginRsp.TradingDay = loginRspT->TradingDay;
loginRsp.MaxLocalId = loginRspT->MaxLocalId;
}
void TradeClient::handleEntrustRtn(FbMsg::OrderT *orderT)
{
Logger::getInstance().log->info("TradeClient::handleEntrustRtn ChannelId:{} ExchangeId:{} InstrumentId:{} "
"AccountId:{} Volume:{} LimitPrice:{} BuyOrSell:{} OrderPriceType:{} OrderStatus:{} OrderId:{} OrderSysId:{} VolumeLeft:{} VolumeTraded:{} InsertTime:{} StatusMsg:{}"
,orderT->ChannelId,orderT->ExchangeId,orderT->InstrumentId,orderT->AccountId,orderT->Volume
,orderT->LimitPrice,orderT->BuyOrSell,orderT->OrderPriceType,orderT->OrderStatus,orderT->OrderId,orderT->OrderSysId,orderT->VolumeLeft,orderT->VolumeTraded,orderT->InsertTime,orderT->StatusMsg);
Order order;
order.ChannelId = orderT->ChannelId;
order.ExchangeId = orderT->ExchangeId;
order.InstrumentId = orderT->InstrumentId;
order.AccountId = orderT->AccountId;
order.OrderSysId = orderT->OrderSysId;
order.Volume = orderT->Volume;
order.VolumeLeft = orderT->VolumeLeft;
order.VolumeTraded = orderT->VolumeTraded;
order.LimitPrice = orderT->LimitPrice;
order.TradeAmount = orderT->TradeAmount;
order.BuyOrSell = orderT->BuyOrSell;
order.HedgeFlag = orderT->HedgeFlag;
order.OrderPriceType = orderT->OrderPriceType;
order.OpenOrClose = orderT->OpenOrClose;
order.OrderStatus = orderT->OrderStatus;
order.InsertTime = orderT->InsertTime;
order.LocalInsertTime = orderT->LocalInsertTime;
order.OrderId = orderT->OrderId;
order.OrigOrderSysId = orderT->OrigOrderSysId;
order.OrigOrderId = orderT->OrigOrderId;
order.StatusMsg = orderT->StatusMsg;
}
void TradeClient::handleCancelOrderRtn(FbMsg::OrderT *orderT)
{
Logger::getInstance().log->info("TradeClient::handleCancelOrderRtn ChannelId:{} AccountId:{} OrderId:{} OrderSysId:{} OrderStatus:{} StatusMsg:{}",orderT->ChannelId,orderT->AccountId,orderT->OrderId,orderT->OrderSysId,orderT->OrderStatus,orderT->StatusMsg);
Order order;
order.ChannelId = orderT->ChannelId;
order.ExchangeId = orderT->ExchangeId;
order.InstrumentId = orderT->InstrumentId;
order.AccountId = orderT->AccountId;
order.OrderSysId = orderT->OrderSysId;
order.Volume = orderT->Volume;
order.VolumeLeft = orderT->VolumeLeft;
order.VolumeTraded = orderT->VolumeTraded;
order.LimitPrice = orderT->LimitPrice;
order.TradeAmount = orderT->TradeAmount;
order.BuyOrSell = orderT->BuyOrSell;
order.HedgeFlag = orderT->HedgeFlag;
order.OrderPriceType = orderT->OrderPriceType;
order.OpenOrClose = orderT->OpenOrClose;
order.OrderStatus = orderT->OrderStatus;
order.InsertTime = orderT->InsertTime;
order.LocalInsertTime = orderT->LocalInsertTime;
order.OrderId = orderT->OrderId;
order.OrigOrderSysId = orderT->OrigOrderSysId;
order.OrigOrderId = orderT->OrigOrderId;
order.StatusMsg = orderT->StatusMsg;
}
void TradeClient::handleTradeRtn(FbMsg::TradeT *tradeT)
{
Logger::getInstance().log->info("TradeClient::handleTradeRtn ChannelId:{} ExchangeId:{} InstrumentId:{} "
"AccountId:{} OrderId:{} OrderSysId:{} TradeId:{} Price:{} Volume:{} TradeTime:{} BuyOrSell:{} "
,tradeT->ChannelId,tradeT->ExchangeId,tradeT->InstrumentId,tradeT->AccountId,tradeT->OrderId
,tradeT->OrderSysId,tradeT->TradeId,tradeT->Price,tradeT->Volume,tradeT->TradeTime,tradeT->BuyOrSell);
Trade trade;
trade.ChannelId = tradeT->ChannelId;
trade.ExchangeId = tradeT->ExchangeId;
trade.InstrumentId = tradeT->InstrumentId;
trade.AccountId = tradeT->AccountId;
trade.OrderSysId = tradeT->OrderSysId;
trade.OrderId = tradeT->OrderId;
trade.TradeId = tradeT->TradeId;
trade.Price = tradeT->Price;
trade.Volume = tradeT->Volume;
trade.TradeTime = TimeUtils::getCurrentYYYYMMDDHHMMSS();
trade.BuyOrSell = tradeT->BuyOrSell;
trade.HedgeFlag = tradeT->HedgeFlag;
trade.OpenOrClose = tradeT->OpenOrClose;
}
void TradeClient::handleQryFundAccountRsp(FbMsg::FundAccountT *fundAccountT)
{
Logger::getInstance().log->debug("TradeClient::handleQryFundAccountRsp ChannelId:{} AccountId:{}",fundAccountT->ChannelId,fundAccountT->AccountId);
FundAccount fundAccount;
fundAccount.ChannelId = fundAccountT->ChannelId;
fundAccount.AccountId = fundAccountT->AccountId;
fundAccount.Available = fundAccountT->Available;
fundAccount.Balance = fundAccountT->Balance;
}
void TradeClient::handleQryPositionsRsp(FbMsg::PositionT *positionT)
{
Logger::getInstance().log->debug("TradeClient::handleQryPositionsRsp ChannelId:{} AccountId:{}",positionT->ChannelId,positionT->AccountId);
Position position;
position.ChannelId = positionT->ChannelId;
position.AccountId = positionT->AccountId;
position.ExchangeId = positionT->ExchangeId;
position.InstrumentId = positionT->InstrumentId;
position.Name = positionT->Name;
position.Volume = positionT->Volume;
position.EnableVolume = positionT->EnableVolume;
position.YdVolume = positionT->YdVolume;
position.TdVolume = positionT->TdVolume;
position.AvgPrice = positionT->AvgPrice;
position.BuyOrSell = positionT->BuyOrSell;
position.HedgeFlag = positionT->HedgeFlag;
}
void TradeClient::handleQryOrdersRsp(FbMsg::OrderT *orderT)
{
Logger::getInstance().log->debug("TradeClient::handleQryOrdersRsp ChannelId:{} AccountId:{}",orderT->ChannelId,orderT->AccountId);
Order order;
order.ChannelId = orderT->ChannelId;
order.ExchangeId = orderT->ExchangeId;
order.InstrumentId = orderT->InstrumentId;
order.AccountId = orderT->AccountId;
order.OrderSysId = orderT->OrderSysId;
order.Volume = orderT->Volume;
order.VolumeLeft = orderT->VolumeLeft;
order.VolumeTraded = orderT->VolumeTraded;
order.LimitPrice = orderT->LimitPrice;
order.TradeAmount = orderT->TradeAmount;
order.BuyOrSell = orderT->BuyOrSell;
order.HedgeFlag = orderT->HedgeFlag;
order.OrderPriceType = orderT->OrderPriceType;
order.OpenOrClose = orderT->OpenOrClose;
order.OrderStatus = orderT->OrderStatus;
order.InsertTime = orderT->InsertTime;
order.LocalInsertTime = orderT->LocalInsertTime;
order.OrderId = orderT->OrderId;
order.OrigOrderSysId = orderT->OrigOrderSysId;
order.OrigOrderId = orderT->OrigOrderId;
order.StatusMsg = orderT->StatusMsg;
}
void TradeClient::handleQryTradesRsp(FbMsg::TradeT *tradeT)
{
Logger::getInstance().log->debug("TradeClient::handleQryTradesRsp ChannelId:{} AccountId:{}",tradeT->ChannelId,tradeT->AccountId);
Trade trade;
trade.ChannelId = tradeT->ChannelId;
trade.ExchangeId = tradeT->ExchangeId;
trade.InstrumentId = tradeT->InstrumentId;
trade.AccountId = tradeT->AccountId;
trade.OrderSysId = tradeT->OrderSysId;
trade.OrderId = tradeT->OrderId;
trade.TradeId = tradeT->TradeId;
trade.Price = tradeT->Price;
trade.Volume = tradeT->Volume;
trade.TradeTime = tradeT->TradeTime;
trade.BuyOrSell = tradeT->BuyOrSell;
trade.HedgeFlag = tradeT->HedgeFlag;
trade.OpenOrClose = tradeT->OpenOrClose;
}
void TradeClient::handleQryShareholderRsp(FbMsg::ShareHolderT *shareHolderT)
{
Logger::getInstance().log->debug("TradeClient::handleQryShareholderRsp ChannelId:{} AccountId:{}",shareHolderT->ChannelId,shareHolderT->AccountId);
ShareHolder shareHolder;
shareHolder.ChannelId = shareHolderT->ChannelId;
shareHolder.AccountId = shareHolderT->AccountId;
shareHolder.ExchangeId = shareHolderT->ExchangeId;
shareHolder.ShareHolderId = shareHolderT->ShareHolderId;
shareHolder.ShareHolderName = shareHolderT->ShareHolderName;
}
void TradeClient::handleError(FbMsg::ErrorInfoT *errorInfoT)
{
Logger::getInstance().log->error("TradeClient::handleError ChannelId:{} AccountId:{} ErrorId:{} ErrorMsg:{}",errorInfoT->ChannelId,errorInfoT->AccountId,errorInfoT->ErrorId,errorInfoT->ErrorMsg);
ErrorInfo error;
error.ChannelId = errorInfoT->ChannelId;
error.AccountId = errorInfoT->AccountId;
error.ErrorId = errorInfoT->ErrorId;
error.ErrorMsg = errorInfoT->ErrorMsg;
}
void TradeClient::handleHeartBeatRsp(FbMsg::HeartBeatT *heartT)
{
Logger::getInstance().log->debug("TradeClient::handleHeartBeatRsp ChannelId:{} AccountId:{}",heartT->ChannelId,heartT->AccountId);
HeartBeat heart;
heart.ChannelId = heartT->ChannelId;
heart.AccountId = heartT->AccountId;
heart.LastTime = heartT->LastTime;
heart.ErrorId = heartT->ErrorId;
heart.ErrorMsg = heartT->ErrorMsg;
if(!m_gateway_is_live.load())
{
m_gateway_is_live.store(true);
}
m_last_heartbeat.store(std::stoll(TimeUtils::getCurrentTimestamp()));
}
void TradeClient::handleAgentRegister(FbMsg::AgentInfoT *agentT)
{
// Logger::getInstance().log->debug("TradeClient::handleAgentRegister ChannelId:{} AccountId:{}" ,agentT->ChannelId, agentT->AccountId);
// std::string agentid = agentT->ChannelId+agentT->AccountId;
// AgentInfo agent;
// agent.ChannelId = agentT->ChannelId;
// agent.AccountId = agentT->AccountId;
// agent.LastTime = TimeUtils::getCurrentTimestamp();
// agent.Status = agentT->Status;
// {
// std::lock_guard<std::mutex> lock(m_mtxAgent);
// m_mapAgent[agentid] = agent;
// }
// _agentRegisterRsp(agent);
}
void TradeClient::run()
{
while(true)
{
LoginReq loginReq;
loginReq.ChannelId = m_channelid;
loginReq.AccountId = m_accountid;
loginReq.Password = m_pass;
login(loginReq);
std::this_thread::sleep_for(std::chrono::seconds(1));
Order insOrder;
insOrder.ChannelId = m_channelid;
insOrder.AccountId = m_accountid;
insOrder.ExchangeId = EXCHANGE_SH;
insOrder.InstrumentId = "600018";
insOrder.LimitPrice = 76;
insOrder.Volume = 200;
insOrder.BuyOrSell = ORDER_DIRECTION_BUY;
insOrder.OrderPriceType = ORDER_PRICE_TYPE_LIMIT;
insOrder.OrderId = "1000000";
insOrder.ChannelId = m_channelid;
insOrder.AccountId = m_accountid;
insertOrder(insOrder);
std::this_thread::sleep_for(std::chrono::seconds(1));
Order canOrder;
canOrder.ChannelId = m_channelid;
canOrder.AccountId = m_accountid;
canOrder.OrderId = "1000001";
canOrder.OrderSysId="1000000";
cancelOrder(canOrder);
std::this_thread::sleep_for(std::chrono::seconds(1));
QryReq query;
query.ChannelId = m_channelid;
query.AccountId = m_accountid;
queryAccount(query);
std::this_thread::sleep_for(std::chrono::seconds(1));
queryPositions(query);
std::this_thread::sleep_for(std::chrono::seconds(1));
queryOrders(query);
std::this_thread::sleep_for(std::chrono::seconds(1));
queryTrades(query);
std::this_thread::sleep_for(std::chrono::seconds(1));
queryShareHolderInfo(query);
std::this_thread::sleep_for(std::chrono::seconds(1));
}
}
void TradeClient::init(ConfigParams ¶ms)
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_send_trade_req_addr = params["client"]["send_trade_req_addr"].as<std::string>();
m_recv_trade_rtn_addr = params["client"]["recv_trade_rtn_addr"].as<std::string>();
m_send_query_req_addr = params["client"]["send_query_req_addr"].as<std::string>();
m_recv_query_rtn_addr = params["client"]["recv_query_rtn_addr"].as<std::string>();
m_heartBeatInterval = params["client"]["heart_beat_interval"].as<int>();
m_checkHeartBeatTimeout = params["client"]["check_heart_beat_timeout"].as<int>();
m_channelid = params["client"]["channelid"].as<std::string>();
m_accountid = params["client"]["accountid"].as<std::string>();
m_pass = params["client"]["pass"].as<std::string>();
Logger::getInstance().log->debug("{} load config recv_trade_req_addr:{} send_trade_rtn_addr:{} recv_query_req_addr:{} send_query_rtn_addr:{}",GET_CLASS_FUNCTION_NAME,m_send_trade_req_addr,m_recv_trade_rtn_addr,m_send_query_req_addr,m_recv_query_rtn_addr);
_createZmqCtx();
initTrade();
initQuery();
std::thread sendHeartBeat(&TradeClient::_sendHeartBeat, this);
sendHeartBeat.detach();
std::thread checkTimeout(&TradeClient::_checkHeartBeatTimeout, this);
checkTimeout.detach();
}
void TradeClient::initTrade()
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_send_trade_req_identity = m_channelid+m_accountid;
m_recv_trade_rtn_identity = m_channelid+m_accountid;
_setDealerSocket(m_pSendTradeReqSocket, m_send_trade_req_identity, m_send_trade_req_addr);
_setDealerSocket(m_pRecvTradeRtnSocket, m_recv_trade_rtn_identity, m_recv_trade_rtn_addr);
std::thread recvTraderRtn(&TradeClient::listenTradeRtnMessages, this);
recvTraderRtn.detach();
}
void TradeClient::initQuery()
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_send_query_req_identity = m_channelid+m_accountid;
m_recv_query_rtn_identity = m_channelid+m_accountid;
_setDealerSocket(m_pSendQueryReqSocket, m_send_query_req_identity, m_send_query_req_addr);
_setDealerSocket(m_pRecvQueryRtnSocket, m_recv_query_rtn_identity, m_recv_query_rtn_addr);
std::thread recvQueryRtn(&TradeClient::listenQueryRtnMessages, this);
recvQueryRtn.detach();
}
void TradeClient::listenTradeRtnMessages()
{
Logger::getInstance().log->debug("{} start",GET_CLASS_FUNCTION_NAME);
const int BUFFSIZE = 1024;
uchar_t buff[BUFFSIZE];
while (1)
{
zmq_pollitem_t items[] = {{ m_pRecvTradeRtnSocket, 0, ZMQ_POLLIN, 0}};
int rc = zmq_poll(items, 1, -1);
if (rc == -1) {
Logger::getInstance().log->error("{} Poll error:{}",GET_CLASS_FUNCTION_NAME,zmq_strerror(errno));
continue;
}
if (items[0].revents & ZMQ_POLLIN)
{
memset(buff,0,sizeof(buff));
ZmqHelper::recvBuffMsg(m_pRecvTradeRtnSocket,buff);
Header header;
memcpy(&header,buff,sizeof(Header));
if(header.MsgType==MsgType::LoginRspType)
{
auto loginRsp = flatbuffers::GetRoot<FbMsg::LoginRsp>(buff+sizeof(Header));
FbMsg::LoginRspT* loginRspT = loginRsp->UnPack();
handleLoginRsp(loginRspT);
}
else if(header.MsgType==MsgType::EntrustRtnType)
{
auto order = flatbuffers::GetRoot<FbMsg::Order>(buff+sizeof(Header));
FbMsg::OrderT* orderT = order->UnPack();
handleEntrustRtn(orderT);
}
else if(header.MsgType==MsgType::CancelOrderRtnType)
{
auto order = flatbuffers::GetRoot<FbMsg::Order>(buff+sizeof(Header));
FbMsg::OrderT* orderT = order->UnPack();
handleCancelOrderRtn(orderT);
}
else if(header.MsgType==MsgType::TradeRtnType)
{
auto trade = flatbuffers::GetRoot<FbMsg::Trade>(buff+sizeof(Header));
FbMsg::TradeT* tradeT = trade->UnPack();
handleTradeRtn(tradeT);
}
else if(header.MsgType==MsgType::ErrorInfoType)
{
auto error = flatbuffers::GetRoot<FbMsg::ErrorInfo>(buff+sizeof(Header));
FbMsg::ErrorInfoT* errorT = error->UnPack();
handleError(errorT);
}
else if(header.MsgType==MsgType::HeartBeatTypeRspType)
{
auto heart = flatbuffers::GetRoot<FbMsg::HeartBeat>(buff+sizeof(Header));
FbMsg::HeartBeatT* heartT = heart->UnPack();
handleHeartBeatRsp(heartT);
}
else if(header.MsgType==MsgType::AgentRegisterType)
{
auto agent = flatbuffers::GetRoot<FbMsg::AgentInfo>(buff+sizeof(Header));
FbMsg::AgentInfoT* agentT = agent->UnPack();
handleAgentRegister(agentT);
}
memset(buff,0,sizeof(buff));
}
}
zmq_close(m_pRecvTradeRtnSocket);
zmq_ctx_destroy(m_pRecvTradeRtnContext);
return;
}
void TradeClient::listenQueryRtnMessages()
{
Logger::getInstance().log->debug("{} start",GET_CLASS_FUNCTION_NAME);
const int BUFFSIZE = 1024;
uchar_t buff[BUFFSIZE];
while (1)
{
zmq_pollitem_t items[] = {{ m_pRecvQueryRtnSocket, 0, ZMQ_POLLIN, 0}};
int rc = zmq_poll(items, 1, -1);
if (rc == -1) {
Logger::getInstance().log->error("{} Poll error:{}",GET_CLASS_FUNCTION_NAME,zmq_strerror(errno));
continue;
}
if (items[0].revents & ZMQ_POLLIN)
{
int ret = 0;
memset(buff,0,sizeof(buff));
ZmqHelper::recvBuffMsg(m_pRecvQueryRtnSocket,buff);
Header header;
memcpy(&header,buff,sizeof(Header));
if(header.MsgType==MsgType::QryFundAccRspType)
{
auto fundAccount = flatbuffers::GetRoot<FbMsg::FundAccount>(buff+sizeof(Header));
FbMsg::FundAccountT* fundAccountT = fundAccount->UnPack();
handleQryFundAccountRsp(fundAccountT);
}
else if(header.MsgType==MsgType::QryPositionsRspType)
{
auto position = flatbuffers::GetRoot<FbMsg::Position>(buff+sizeof(Header));
FbMsg::PositionT* positionT = position->UnPack();
handleQryPositionsRsp(positionT);
}
else if(header.MsgType==MsgType::QryOrdersRspType)
{
auto order = flatbuffers::GetRoot<FbMsg::Order>(buff+sizeof(Header));
FbMsg::OrderT* orderT = order->UnPack();
handleQryOrdersRsp(orderT);
}
else if(header.MsgType==MsgType::QryTradesRspType)
{
auto trade = flatbuffers::GetRoot<FbMsg::Trade>(buff+sizeof(Header));
FbMsg::TradeT* tradeT = trade->UnPack();
handleQryTradesRsp(tradeT);
}
else if(header.MsgType==MsgType::QryShareHolderRspType)
{
auto shareHolder = flatbuffers::GetRoot<FbMsg::ShareHolder>(buff+sizeof(Header));
FbMsg::ShareHolderT* shareHolderT = shareHolder->UnPack();
handleQryShareholderRsp(shareHolderT);
}
else if(header.MsgType==MsgType::ErrorInfoType)
{
auto error = flatbuffers::GetRoot<FbMsg::ErrorInfo>(buff+sizeof(Header));
FbMsg::ErrorInfoT* errorT = error->UnPack();
handleError(errorT);
}
memset(buff,0,sizeof(buff));
}
}
zmq_close(m_pRecvQueryRtnSocket);
zmq_ctx_destroy(m_pRecvQueryRtnContext);
return;
}
int TradeClient::login(LoginReq &loginReq)
{
if(!_checkGatewayStatus(loginReq.ChannelId+loginReq.AccountId)) // agent不在线
{
ErrorInfo error;
error.ChannelId = loginReq.ChannelId;
error.AccountId = loginReq.AccountId;
error.ErrorId = ERROR_ID_LOGIN;
error.ErrorMsg = "login gateway error";
LoginRsp loginRsp;
loginRsp.AccountId = loginReq.AccountId;
loginRsp.ChannelId = loginReq.ChannelId;
loginRsp.LoginTime = TimeUtils::getHH_MM_SSTime();
loginRsp.TradingDay = TimeUtils::getDateYYYYMMDD();
loginRsp.MaxLocalId = 10000000;
return 0;
}
FbMsg::LoginReqT loginReqT;
loginReqT.AccountId = loginReq.AccountId;
loginReqT.Password = loginReq.Password;
loginReqT.ChannelId = loginReq.ChannelId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbLoginReq = FbMsg::CreateLoginReq(builder,&loginReqT);
builder.Finish(fbLoginReq);
Header header;
header.MsgLength = sizeof(Header) + builder.GetSize();
header.MsgType = MsgType::LoginReqType;
header.ChannelId = 101;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
return 0;
}
int TradeClient::logout()
{
return 0;
}
int TradeClient::insertOrder(Order &order)
{
if(!_checkGatewayStatus(order.ChannelId+order.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::insertOrder client not online, gateway insert order error ChannelId:{} ExchangeId:{} InstrumentId:{} "
"AccountId:{} Volume:{} LimitPrice:{} BuyOrSell:{} OrderPriceType:{} OrderId:{}"
,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.Volume
,order.LimitPrice,order.BuyOrSell,order.OrderPriceType,order.OrderId);
ErrorInfo error;
error.ChannelId = order.ChannelId;
error.AccountId = order.AccountId;
error.ErrorId = ERROR_ID_INSERT;
error.ErrorMsg = "TradeClient::insertOrder gateway insert order error|client not online";
order.OrderStatus = ORDER_STATUS_REJECTED;
order.VolumeTraded = 0;
order.TradeAmount = 0;
order.OrderSysId = std::string("#gateway-")+TimeUtils::getCurrentTimestamp();
order.StatusMsg = error.ErrorMsg;
return 0;
}
Logger::getInstance().log->info("TradeClient::insertOrder ChannelId:{} ExchangeId:{} InstrumentId:{} "
"AccountId:{} Volume:{} LimitPrice:{} BuyOrSell:{} OrderPriceType:{} OrderId:{}"
,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.Volume
,order.LimitPrice,order.BuyOrSell,order.OrderPriceType,order.OrderId);
FbMsg::OrderT orderT;
orderT.ChannelId = order.ChannelId;
orderT.ExchangeId = order.ExchangeId;
orderT.InstrumentId = order.InstrumentId;
orderT.AccountId = order.AccountId;
orderT.OrderSysId = order.OrderSysId;
orderT.Volume = order.Volume;
orderT.VolumeLeft = order.VolumeLeft;
orderT.VolumeTraded = order.VolumeTraded;
orderT.LimitPrice = order.LimitPrice;
orderT.TradeAmount = order.TradeAmount;
orderT.BuyOrSell = order.BuyOrSell;
orderT.HedgeFlag = order.HedgeFlag;
orderT.OrderPriceType = order.OrderPriceType;
orderT.OpenOrClose = order.OpenOrClose;
orderT.OrderStatus = order.OrderStatus;
orderT.InsertTime = order.InsertTime;
orderT.LocalInsertTime = order.LocalInsertTime;
orderT.OrderId = order.OrderId;
orderT.OrigOrderSysId = order.OrigOrderSysId;
orderT.OrigOrderId = order.OrigOrderId;
orderT.StatusMsg = order.StatusMsg;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbOrder = FbMsg::CreateOrder(builder,&orderT);
builder.Finish(fbOrder);
Header header;
header.MsgLength = sizeof(Header) + builder.GetSize();
header.MsgType = MsgType::EntrustType;
header.ChannelId = 101;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_trade_mtx);;
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
return 0;
}
int TradeClient::cancelOrder(Order &order)
{
if(!_checkGatewayStatus(order.ChannelId+order.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::cancelOrder client not online, gateway cancel order error ChannelId:{} ExchangeId:{} InstrumentId:{} AccountId:{} OrderId:{} OrderSysId:{}"
,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.OrderId,order.OrderSysId);
ErrorInfo error;
error.ChannelId = order.ChannelId;
error.AccountId = order.AccountId;
error.ErrorId = ERROR_ID_CANCEL;
error.ErrorMsg = "TradeClient::cancelOrder gateway cancel order error|client not online";
order.OrderStatus = ORDER_STATUS_REJECTED;
order.StatusMsg = error.ErrorMsg;
return 0;
}
Logger::getInstance().log->info("TradeClient::cancelOrder ChannelId:{} ExchangeId:{} InstrumentId:{} AccountId:{} OrderId:{} OrderSysId:{}"
,order.ChannelId,order.ExchangeId,order.InstrumentId,order.AccountId,order.OrderId,order.OrderSysId);
FbMsg::OrderT orderT;
orderT.ChannelId = order.ChannelId;
orderT.ExchangeId = order.ExchangeId;
orderT.InstrumentId = order.InstrumentId;
orderT.AccountId = order.AccountId;
orderT.OrderSysId = order.OrderSysId;
orderT.Volume = order.Volume;
orderT.VolumeLeft = order.VolumeLeft;
orderT.VolumeTraded = order.VolumeTraded;
orderT.LimitPrice = order.LimitPrice;
orderT.TradeAmount = order.TradeAmount;
orderT.BuyOrSell = order.BuyOrSell;
orderT.HedgeFlag = order.HedgeFlag;
orderT.OrderPriceType = order.OrderPriceType;
orderT.OpenOrClose = order.OpenOrClose;
orderT.OrderStatus = order.OrderStatus;
orderT.InsertTime = order.InsertTime;
orderT.LocalInsertTime = order.LocalInsertTime;
orderT.OrderId = order.OrderId;
orderT.OrigOrderSysId = order.OrigOrderSysId;
orderT.OrigOrderId = order.OrigOrderId;
orderT.StatusMsg = order.StatusMsg;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbOrder = FbMsg::CreateOrder(builder,&orderT);
builder.Finish(fbOrder);
Header header;
header.MsgLength = sizeof(Header) + builder.GetSize();
header.MsgType = MsgType::CancelOrderType;
header.ChannelId = 101;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
return 0;
}
int TradeClient::cancelBatchOrder(std::vector<Order> &vecOrders)
{
Logger::getInstance().log->info("TradeClient::cancelBatchOrder start");
for(auto& order : vecOrders)
{
cancelOrder(order);
}
Logger::getInstance().log->info("TradeClient::cancelBatchOrder end");
return 0;
}
int TradeClient::queryAccount(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryAccount client not online, gateway query account error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryAccErrCnt = 0;
if(qryAccErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYACC;
error.ErrorMsg = "gateway query account error";
qryAccErrCnt++;
qryAccErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryAccount ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);
Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryFundAccType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);
return 0;
}
int TradeClient::queryPositions(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryPositions client not online, gateway query positions error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryPosErrCnt = 0;
if(qryPosErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYPOS;
error.ErrorMsg = "gateway query positions error";
qryPosErrCnt++;
qryPosErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryPositions ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);
Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryPositionsType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);
return 0;
}
int TradeClient::queryOrders(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryOrders client not online, gateway query orders error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryOrdErrCnt = 0;
if(qryOrdErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYORDERS;
error.ErrorMsg = "gateway query orders error";
qryOrdErrCnt++;
qryOrdErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryOrders ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);
Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryOrdersType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);
return 0;
}
int TradeClient::queryTrades(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryTrades client not online, gateway query trades error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryTrdErrCnt = 0;
if(qryTrdErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYTRADES;
error.ErrorMsg = "gateway query trades error";
qryTrdErrCnt++;
qryTrdErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryTrades ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);
Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryTradesType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);
return 0;
}
int TradeClient::queryShareHolderInfo(QryReq& query)
{
if(!_checkGatewayStatus(query.ChannelId+query.AccountId)) // agent不在线
{
Logger::getInstance().log->error("TradeClient::queryShareHolderInfo client not online, gateway query shareholder error ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
static int qryShdErrCnt = 0;
if(qryShdErrCnt > STACK_DEEP)
{
return 0;
}
ErrorInfo error;
error.ChannelId = query.ChannelId;
error.AccountId = query.AccountId;
error.ErrorId = ERROR_ID_QRYSHARE;
error.ErrorMsg = "gateway query shareholder info error";
qryShdErrCnt++;
qryShdErrCnt--;
return 0;
}
Logger::getInstance().log->info("TradeClient::queryShareHolderInfo ChannelId:{} AccountId:{}",query.ChannelId,query.AccountId);
FbMsg::QryReqT queryT;
queryT.ChannelId = query.ChannelId;
queryT.AccountId = query.AccountId;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbQryReq = FbMsg::CreateQryReq(builder,&queryT);
builder.Finish(fbQryReq);
Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::QryShareHolderType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_qry_mtx);
ZmqHelper::sendPackMsg(m_pSendQueryReqSocket, &msg);
return 0;
}
int TradeClient::heartBeat(HeartBeat &heart)
{
FbMsg::HeartBeatT heartT;
heartT.ChannelId = heart.ChannelId;
heartT.AccountId = heart.AccountId;
heartT.LastTime = heart.LastTime;
heartT.ErrorId = heart.ErrorId;
heartT.ErrorMsg = heart.ErrorMsg;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbheart = FbMsg::CreateHeartBeat(builder,&heartT);
builder.Finish(fbheart);
Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::HeartBeatType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
return 0;
}
void TradeClient::_sendHeartBeat()
{
while (true)
{
HeartBeat heart;
heart.ChannelId = m_channelid;
heart.AccountId = m_accountid;
heart.LastTime = TimeUtils::getCurrentTimestamp();
heart.ErrorId = 0;
heart.ErrorMsg = "OK";
heartBeat(heart);
Logger::getInstance().log->info("TradeClient::_sendHeartBeat ChannelId:{} AccountId:{}",heart.ChannelId,heart.AccountId);
std::this_thread::sleep_for(std::chrono::seconds(m_heartBeatInterval));
}
}
void TradeClient::_checkHeartBeatTimeout()
{
while(true)
{
long long last_time = m_last_heartbeat.load();
if(last_time > 0)
{
long long curtime = std::stoll(TimeUtils::getCurrentTimestamp());
long long dealt = curtime - last_time;
if(dealt > m_checkHeartBeatTimeout)
{
m_gateway_is_live.store(false);
Logger::getInstance().log->error("TradeClient::_checkHeartBeatTimeout channel:{} account:{} curtime:{} LastTime:{} dealt:{} heartBeatInterval:{} checkHeartBeatTimeout:{} connect agent timeout",m_channelid,m_accountid,curtime,last_time,dealt,m_heartBeatInterval,m_checkHeartBeatTimeout);
}
}
std::this_thread::sleep_for(std::chrono::seconds(1));
}
}
void TradeClient::_agentRegisterRsp(AgentInfo &agent)
{
FbMsg::AgentInfoT agentT;
agentT.ChannelId = agent.ChannelId;
agentT.AccountId = agent.AccountId;
agentT.LastTime = agent.LastTime;
agentT.Status = agent.Status;
flatbuffers::FlatBufferBuilder builder(1024);
auto fbagent = FbMsg::CreateAgentInfo(builder,&agentT);
builder.Finish(fbagent);
Header header;
header.MsgLength = sizeof(Header);
header.MsgType = MsgType::AgentRegisterRspType;
PackMsg msg;
msg.pack(&header,builder.GetBufferPointer(),builder.GetSize());
std::lock_guard<std::mutex> lock(m_trade_mtx);
ZmqHelper::sendPackMsg(m_pSendTradeReqSocket, &msg);
}
bool TradeClient::_checkGatewayStatus(std::string agentid)
{
return m_gateway_is_live.load();
}
void TradeClient::_createZmqCtx()
{
Logger::getInstance().log->debug("{}",GET_CLASS_FUNCTION_NAME);
m_pSendTradeReqContext = zmq_ctx_new();
m_pSendTradeReqSocket = zmq_socket(m_pSendTradeReqContext, ZMQ_DEALER);
m_pRecvTradeRtnContext = zmq_ctx_new();
m_pRecvTradeRtnSocket = zmq_socket(m_pRecvTradeRtnContext, ZMQ_DEALER);
m_pSendQueryReqContext = zmq_ctx_new();
m_pSendQueryReqSocket = zmq_socket(m_pSendQueryReqContext, ZMQ_DEALER);
m_pRecvQueryRtnContext = zmq_ctx_new();
m_pRecvQueryRtnSocket = zmq_socket(m_pRecvQueryRtnContext, ZMQ_DEALER);
}
void TradeClient::_setDealerSocket(ZmqSocketPtr socket, std::string identity, std::string addr)
{
Logger::getInstance().log->debug("{} identity:{} bind addr:{}",GET_CLASS_FUNCTION_NAME,identity,addr);
zmq_setsockopt(socket, ZMQ_IDENTITY, identity.c_str(), identity.length());
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE, &m_tcp_keep_alive, sizeof(m_tcp_keep_alive));
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_IDLE, &m_tcp_keep_idle, sizeof(m_tcp_keep_idle));
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_INTVL, &m_tcp_keep_intvl, sizeof(m_tcp_keep_intvl));
zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_CNT, &m_tcp_keep_cnt, sizeof(m_tcp_keep_cnt));
zmq_setsockopt(socket, ZMQ_HEARTBEAT_IVL, &m_heartbeat_ivl, sizeof(m_heartbeat_ivl));
zmq_setsockopt(socket, ZMQ_HEARTBEAT_TIMEOUT, &m_heartbeat_timeout, sizeof(m_heartbeat_timeout));
zmq_setsockopt(socket, ZMQ_HEARTBEAT_TTL, &m_heartbeat_ttl, sizeof(m_heartbeat_ttl));
zmq_connect(socket, addr.c_str());
}
bool TradeClient::_checkClientStatus(std::string clientid)
{
return false;
}
转成C#代码