ctp_md_adapter.hpp

Go to the documentation of this file.

 
#pragma once
 
#ifdef ENABLE_CTP
 
#include <thread>
#include <atomic>
#include <mutex>
#include <set>
#include <string>
#include "market/md_adapter.hpp"
#include "ThostFtdcMdApi.h"
 
namespace fix40 {
 
struct CtpMdConfig {
    std::string mdFront;      
    std::string brokerId;     
    std::string userId;       
    std::string password;     
    std::string flowPath;     
};
 
// 前向声明
class CtpMdAdapter;
 
class CtpMdSpi final : public CThostFtdcMdSpi {
public:
    explicit CtpMdSpi(CtpMdAdapter* adapter) : adapter_(adapter) {}
    ~CtpMdSpi() = default;
 
    void OnFrontConnected() override;
    void OnFrontDisconnected(int nReason) override;
    void OnRspUserLogin(CThostFtdcRspUserLoginField* pRspUserLogin,
                        CThostFtdcRspInfoField* pRspInfo,
                        int nRequestID, bool bIsLast) override;
    void OnRspSubMarketData(CThostFtdcSpecificInstrumentField* pSpecificInstrument,
                            CThostFtdcRspInfoField* pRspInfo,
                            int nRequestID, bool bIsLast) override;
    void OnRspUnSubMarketData(CThostFtdcSpecificInstrumentField* pSpecificInstrument,
                              CThostFtdcRspInfoField* pRspInfo,
                              int nRequestID, bool bIsLast) override;
    void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField* pDepthMarketData) override;
    void OnRspError(CThostFtdcRspInfoField* pRspInfo, int nRequestID, bool bIsLast) override;
    void OnHeartBeatWarning(int nTimeLapse) override;
 
private:
    CtpMdAdapter* adapter_;
};
 
class CtpMdAdapter : public MdAdapter {
    friend class CtpMdSpi;
 
public:
    CtpMdAdapter(moodycamel::BlockingConcurrentQueue<MarketData>& queue,
                 const CtpMdConfig& config);
 
    ~CtpMdAdapter() override;
 
    // 禁止拷贝
    CtpMdAdapter(const CtpMdAdapter&) = delete;
    CtpMdAdapter& operator=(const CtpMdAdapter&) = delete;
 
    // MdAdapter 接口实现
    bool start() override;
    void stop() override;
    bool isRunning() const override { return running_.load(); }
    MdAdapterState getState() const override { return state_.load(); }
 
    bool subscribe(const std::vector<std::string>& instruments) override;
    bool unsubscribe(const std::vector<std::string>& instruments) override;
 
    void setStateCallback(StateCallback callback) override;
 
    std::string getName() const override { return "CTP"; }
    std::string getTradingDay() const override;
 
private:
    void doLogin();
 
    void doSubscribePending();
 
    void notifyState(MdAdapterState state, const std::string& message);
 
    static MarketData convertMarketData(const CThostFtdcDepthMarketDataField* pData);
 
    CtpMdConfig config_;
    CThostFtdcMdApi* api_ = nullptr;
    std::unique_ptr<CtpMdSpi> spi_;
 
    std::atomic<bool> running_{false};
    std::atomic<MdAdapterState> state_{MdAdapterState::DISCONNECTED};
    std::atomic<int> requestId_{0};
 
    mutable std::mutex mutex_;
    std::set<std::string> subscribedInstruments_;
    std::set<std::string> pendingSubscribe_;
    std::string tradingDay_;
 
    StateCallback stateCallback_;
};
 
CtpMdConfig loadCtpConfig(const std::string& filename);
 
} // namespace fix40
 
#endif // ENABLE_CTP