//说明部分 //系统 #include "stdafx.h" #include #include //Boost #define BOOST_PYTHON_STATIC_LIB #include //python封装 #include //python封装 #include //python封装 #include //python封装 #include //python封装 #include //任务队列的线程功能 #include //任务队列的线程功能 #include //任务队列的任务实现 //API #include "SecurityFtdcMdApi.h" //命名空间 using namespace std; using namespace boost::python; using namespace boost; //常量 #define ONFRONTCONNECTED 0 #define ONFRONTDISCONNECTED 1 #define ONHEARTBEATWARNING 2 #define ONRSPERROR 3 #define ONRSPUSERLOGIN 4 #define ONRSPUSERLOGOUT 5 #define ONRSPSUBMARKETDATA 6 #define ONRSPUNSUBMARKETDATA 7 #define ONRTNDEPTHMARKETDATA 8 #define EXIT 0 ///------------------------------------------------------------------------------------- ///API中的部分组件 ///------------------------------------------------------------------------------------- //GIL全局锁简化获取用， //用于帮助C++线程获得GIL锁，从而防止python崩溃 class PyLock { private: PyGILState_STATE gil_state; public: //在某个函数方法中创建该对象时，获得GIL锁 PyLock() { gil_state = PyGILState_Ensure(); } //在某个函数完成后销毁该对象时，解放GIL锁 ~PyLock() { PyGILState_Release(gil_state); } }; //任务结构体 struct Task { int task_name; //回调函数名称对应的常量 any task_data; //数据结构体 any task_error; //错误结构体 int task_id; //请求id bool task_last; //是否为最后返回 }; ///线程安全的队列 template class ConcurrentQueue { private: queue the_queue; //标准库队列 mutable mutex the_mutex; //boost互斥锁 condition_variable the_condition_variable; //boost条件变量 public: //存入新的任务 void push(Data const& data) { mutex::scoped_lock lock(the_mutex); //获取互斥锁 the_queue.push(data); //向队列中存入数据 lock.unlock(); //释放锁 the_condition_variable.notify_one(); //通知正在阻塞等待的线程 } //检查队列是否为空 bool empty() const { mutex::scoped_lock lock(the_mutex); return the_queue.empty(); } //取出 Data wait_and_pop() { mutex::scoped_lock lock(the_mutex); while (the_queue.empty()) //当队列为空时 { the_condition_variable.wait(lock); //等待条件变量通知 } Data popped_value = the_queue.front(); //获取队列中的最后一个任务 the_queue.pop(); //删除该任务 return popped_value; //返回该任务 } }; //从字典中获取某个建值对应的整数，并赋值到请求结构体对象的值上 void getInt(dict d, string key, int* value); //从字典中获取某个建值对应的浮点数，并赋值到请求结构体对象的值上 void getDouble(dict d, string key, double* value); //从字典中获取某个建值对应的字符串，并赋值到请求结构体对象的值上 void getChar(dict d, string key, char* value); ///------------------------------------------------------------------------------------- ///C++ SPI的回调函数方法实现 ///------------------------------------------------------------------------------------- //API的继承实现 class MdApi : public CSecurityFtdcMdSpi { private: CSecurityFtdcMdApi* api; //API对象 thread *task_thread; //工作线程指针（向python中推送数据） ConcurrentQueue task_queue; //任务队列 public: MdApi() { function0 f = boost::bind(&MdApi::processTask, this); thread t(f); this->task_thread = &t; }; ~MdApi() { }; //------------------------------------------------------------------------------------- //API回调函数 //------------------------------------------------------------------------------------- //当客户端与交易后台建立起通信连接时（还未登录前），该方法被调用。 virtual void OnFrontConnected(); //当客户端与交易后台通信连接断开时，该方法被调用。当发生这个情况后，API会自动重新连接，客户端可不做处理。 //@param nReason 错误原因 // 0x1001 网络读失败 // 0x1002 网络写失败 // 0x2001 接收心跳超时 // 0x2002 发送心跳失败 // 0x2003 收到错误报文 virtual void OnFrontDisconnected(int nReason); //心跳超时警告。当长时间未收到报文时，该方法被调用。 //@param nTimeLapse 距离上次接收报文的时间 virtual void OnHeartBeatWarning(int nTimeLapse); //错误应答 virtual void OnRspError(CSecurityFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast); //登录请求响应 virtual void OnRspUserLogin(CSecurityFtdcRspUserLoginField *pRspUserLogin, CSecurityFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast); //登出请求响应 virtual void OnRspUserLogout(CSecurityFtdcUserLogoutField *pUserLogout, CSecurityFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast); //订阅行情应答 virtual void OnRspSubMarketData(CSecurityFtdcSpecificInstrumentField *pSpecificInstrument, CSecurityFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast); //取消订阅行情应答 virtual void OnRspUnSubMarketData(CSecurityFtdcSpecificInstrumentField *pSpecificInstrument, CSecurityFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast); //深度行情通知 virtual void OnRtnDepthMarketData(CSecurityFtdcDepthMarketDataField *pDepthMarketData); //------------------------------------------------------------------------------------- //task：任务 //------------------------------------------------------------------------------------- void processTask(); void processFrontConnected(Task task); void processFrontDisconnected(Task task); void processHeartBeatWarning(Task task); void processRspError(Task task); void processRspUserLogin(Task task); void processRspUserLogout(Task task); void processRspSubMarketData(Task task); void processRspUnSubMarketData(Task task); void processRtnDepthMarketData(Task task); //------------------------------------------------------------------------------------- //data：回调函数的数据字典 //error：回调函数的错误字典 //id：请求id //last：是否为最后返回 //i：整数 //------------------------------------------------------------------------------------- virtual void onFrontConnected(){}; virtual void onFrontDisconnected(int i){}; virtual void onHeartBeatWarning(int i){}; virtual void onRspError(dict data, int id, bool last) {}; virtual void onRspUserLogin(dict data, dict error, int id, bool last) {}; virtual void onRspUserLogout(dict data, dict error, int id, bool last) {}; virtual void onRspSubMarketData(dict data, dict error, int id, bool last) {}; virtual void onRspUnSubMarketData(dict data, dict error, int id, bool last) {}; virtual void onRtnDepthMarketData(dict data) {}; //------------------------------------------------------------------------------------- //req:主动函数的请求字典 //------------------------------------------------------------------------------------- void createFtdcMdApi(string pszFlowPath = ""); void release(); void init(); int join(); string getTradingDay(); void registerFront(string pszFrontAddress); int subscribeMarketData(dict req); int unSubscribeMarketData(dict req); int reqUserLogin(dict req, int nRequestID); int reqUserLogout(dict req, int nRequestID); };