Merge pull request #1 from TenzT/feat/master_md

Feat/master md

CMakeLists.txt

@@ -40,7 +40,7 @@ elseif(CMAKE_HOST_LINUX)
     link_directories(thirdparty/ctp/lib/lin64)
 endif()
 
-add_executable(CTPTest src/main.cpp
+add_executable(md src/md.cpp
         src/md/CMduserHandler.cpp
         src/md/CMduserHandler.h
         src/yijinjing/locator/MyLocator.cpp
@@ -50,4 +50,26 @@ add_executable(CTPTest src/main.cpp
         src/entity/Quotes.h
 )
 
-target_link_libraries(CTPTest thostmduserapi_se yijinjing)
+target_link_libraries(md thostmduserapi_se yijinjing)
+
+add_executable(master src/master_cmd.cpp
+        src/yijinjing/master/MyMaster.cpp
+        src/yijinjing/locator/MyLocator.cpp
+        src/yijinjing/locator/MyLocator.h
+        src/yijinjing/master/MyMaster.h
+)
+
+target_link_libraries(master yijinjing)
+
+add_executable(strategy_runner src/strategy_runner.cpp
+        src/yijinjing/locator/MyLocator.cpp
+        src/yijinjing/locator/MyLocator.h
+        src/strategy/runner.cpp
+        src/strategy/runner.h)
+target_link_libraries(strategy_runner yijinjing)
+
+
+ADD_COMPILE_DEFINITIONS(SPDLOG_ACTIVE_LEVEL=SPDLOG_LEVEL_INFO)
+ADD_COMPILE_DEFINITIONS(SPDLOG_FMT_EXTERNAL)
+ADD_COMPILE_DEFINITIONS(SPDLOG_NO_NAME)
+ADD_COMPILE_DEFINITIONS(SPDLOG_NO_ATOMIC_LEVELS)

README.md

@@ -1,4 +1,23 @@
 # CTP接入测试
 
 - [x] 接入openctp 7x24行情服务
-- [x] 接入yijinjing
+- [x] 接入yijinjing
+- [x] 仿照kungfu分布式架构，分离出apprentice和master
+
+## 用法
+
+- 编译
+```
+mkdir build && cd build
+cmake .. && make
+```
+- 运行
+```
+./master
+sleep 3
+
+./md tcp://121.37.80.177:20004 sc2410 sc2411 sc2412 sc2501 sc2502
+sleep 3
+
+./strategy_runner
+```

src/entity/Quotes.h

@@ -12,6 +12,10 @@
 #define INSTRUMENT_ID_LEN 32
 #define EXCHANGE_ID_LEN 16
 
+enum msg_type: int32_t {
+    QUOTES = 101
+};
+
 struct Quotes {
     char source_id[SOURCE_ID_LEN];              //柜台ID
     char trading_day[DATE_LEN];                 //交易日

src/master_cmd.cpp

@@ -0,0 +1,18 @@
+#include <iostream>
+
+#include "yijinjing/locator/MyLocator.h"
+#include "yijinjing/master/MyMaster.h"
+
+using namespace kungfu::yijinjing::data;
+
+int main(int argc, const char * argv[]) {
+    std::cout << "Hello, Master!" << std::endl;
+
+    locator_ptr locator = std::make_shared<MyLocator>("./home");
+    location_ptr location = location::make(mode::LIVE, category::SYSTEM, "master", "master", locator);
+    MyMaster my_master(location, false);
+
+    my_master.run();
+
+    return 0;
+}

src/main.cpp → src/md.cpp

@@ -1,8 +1,6 @@
 #include <iostream>
-#include <kungfu/practice/master.h>
 
 #include "../thirdparty/yijinjing/include/kungfu/practice/apprentice.h"
-#include "../thirdparty/yijinjing/include/kungfu/practice/hero.h"
 #include "md/CMduserHandler.h"
 
 std::unique_ptr<CMduserHandler> InitMDHandler(int argc, const char* argv[]) {
@@ -22,7 +20,7 @@ int main(int argc, const char * argv[]) {
     std::cout << "Hello, World!" << std::endl;
     auto mduser_handler = InitMDHandler(argc, argv);
 
-    while (true) {}
+    mduser_handler->run();
 
     return 0;
 }

src/md/CMduserHandler.cpp

@@ -59,8 +59,15 @@ void CMduserHandler::subscribe(char* instrumentId) {
     ++nInstrument;
 }
 
+void CMduserHandler::run() const {
+    journal_->run();
+}
+
 void CMduserHandler::OnRtnDepthMarketData(CThostFtdcDepthMarketDataField *pDepthMarketData) {
-    journal_->write(*pDepthMarketData);
+    if (journal_->has_writer(0)) {
+        journal_->write(*pDepthMarketData);
+        std::cout << "write success" << std::endl;
+    }
     // std::cout << "TradingDay: " << pDepthMarketData->TradingDay << std::endl;
     // std::cout << "ExchangeID: " << pDepthMarketData->ExchangeID << std::endl;
     // std::cout << "InstrumentID: " << pDepthMarketData->InstrumentID << std::endl;

src/md/CMduserHandler.h

@@ -23,6 +23,7 @@ class CMduserHandler : public CThostFtdcMdSpi {
         void connect(const std::string &frontString);
         void login() const;
         void subscribe(char * instrumentId);
+        void run() const;
 
         void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField *pDepthMarketData) override;
 

src/strategy/runner.cpp

@@ -0,0 +1,24 @@
+//
+// Created by 谭德志 on 2024/9/26.
+//
+
+#include "runner.h"
+#include "../entity/Quotes.h"
+using namespace kungfu::rx;
+using namespace kungfu::yijinjing;
+
+void runner::on_start() {
+    auto md_location = location::make(mode::LIVE, category::MD, "ctp", "ctp", home_->locator);
+    this->request_read_from(this->now(), md_location->uid, true);
+    // this->request_write_to(this->now(), md_location->uid);   // 写方向
+    market_data_["ctp"] = md_location->uid;
+    SPDLOG_INFO("added md {} [{:08x}]", "ctp", md_location->uid);
+
+    events_ | is(QUOTES) |
+        $([&](event_ptr event)
+          {
+             std::cout << "quotes received: instrumentId: " << event->data<Quotes>().instrument_id << std::endl;
+          });
+
+    apprentice::on_start();
+}

src/strategy/runner.h

@@ -0,0 +1,25 @@
+//
+// Created by 谭德志 on 2024/9/26.
+//
+
+#ifndef RUNNER_H
+#define RUNNER_H
+#include <kungfu/practice/apprentice.h>
+#include <kungfu/yijinjing/common.h>
+
+using namespace kungfu::yijinjing::data;
+
+class runner : public kungfu::practice::apprentice {
+private:
+    location_ptr home_;
+    std::unordered_map<std::string, uint32_t> market_data_;
+public:
+    explicit runner(location_ptr home, bool low_latency = false) : apprentice(home, low_latency) {
+        home_ = home;
+    }
+    void on_start() override;
+};
+
+
+
+#endif //RUNNER_H

src/strategy_runner.cpp

@@ -0,0 +1,19 @@
+#include <iostream>
+
+#include "strategy/runner.h"
+#include "yijinjing/locator/MyLocator.h"
+#include "yijinjing/master/MyMaster.h"
+
+using namespace kungfu::yijinjing::data;
+
+int main(int argc, const char * argv[]) {
+    std::cout << "Hello, Runner!" << std::endl;
+
+    locator_ptr locator = std::make_shared<MyLocator>("./home");
+    location_ptr location = location::make(mode::LIVE, category::STRATEGY, "ctp", "ctp", locator);
+    runner r(location, false);
+
+    r.run();
+
+    return 0;
+}

src/yijinjing/journal/MyJournal.cpp

@@ -3,3 +3,11 @@
 //
 
 #include "MyJournal.h"
+#include "../../entity/Quotes.h"
+
+void MyJournal::write(const CThostFtdcDepthMarketDataField &depthMarketData) {
+    auto writer = get_writer(0);
+    Quotes& quotes = writer->open_data<Quotes>(this->now() , QUOTES);
+    from_ctp(depthMarketData, quotes);
+    get_writer(0)->close_data();
+}

src/yijinjing/journal/MyJournal.h

@@ -4,23 +4,19 @@
 
 #ifndef MYJOURNAL_H
 #define MYJOURNAL_H
+#include <kungfu/practice/apprentice.h>
 #include <kungfu/practice/master.h>
 
 #include "../locator/MyLocator.h"
 #include "../../entity/Quotes.h"
 
 namespace kf_data = kungfu::yijinjing::data;
 
-class MyJournal : public kungfu::practice::master {
+class MyJournal : public kungfu::practice::apprentice {
 public:
-    explicit MyJournal(location_ptr home, bool low_latency = false): master(home, low_latency) {
-    };
-    void write(const CThostFtdcDepthMarketDataField& depthMarketData) {
-        auto writer = get_writer(0);
-        Quotes& quotes = writer->open_data<Quotes>(0 , 0);
-        from_ctp(depthMarketData, quotes);
-        get_writer(0)->close_data();
-    }
+    explicit MyJournal(location_ptr home, bool low_latency = false): apprentice(home, low_latency) {}
+
+    void write(const CThostFtdcDepthMarketDataField& depthMarketData);
 };
 
 

src/yijinjing/master/MyMaster.cpp

@@ -0,0 +1,5 @@
+//
+// Created by 谭德志 on 2024/9/26.
+//
+
+#include "MyMaster.h"

src/yijinjing/master/MyMaster.h

@@ -0,0 +1,18 @@
+//
+// Created by 谭德志 on 2024/9/26.
+//
+
+#ifndef MASTER_H
+#define MASTER_H
+#include <kungfu/practice/master.h>
+
+using namespace kungfu::yijinjing::data;
+
+class MyMaster : public kungfu::practice::master{
+public:
+    explicit MyMaster(location_ptr home, bool low_latency = false): master(home, low_latency) {};
+};
+
+
+
+#endif //MASTER_H