Compare commits
20
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
518bd5771e | ||
|
|
16522056ba | ||
|
|
51178d85d3 | ||
|
|
86b50e58f4 | ||
|
|
8431c05dfe | ||
|
|
18c05e2c8c | ||
|
|
06535c0938 | ||
|
|
0d8f01c8bb | ||
|
|
36f85b1e62 | ||
|
|
502a22a2d9 | ||
|
|
daa49a5568 | ||
|
|
0ef371cc40 | ||
|
|
b23f0fc21b | ||
|
|
d0dc1b8e51 | ||
|
|
0bbc4f737b | ||
|
|
29fb1b427b | ||
|
|
9de6d3ffe6 | ||
|
|
a168d4da99 | ||
|
|
d65d4b3258 | ||
|
|
b4771abad0 |
No files matched your search
+2
-2
@@ -113,8 +113,8 @@
|
||||
"description": "Debug build compiled w/ gcc",
|
||||
"binaryDir": "${sourceDir}/build/gcc/debug",
|
||||
"cacheVariables": {
|
||||
"CMAKE_C_COMPILER": "gcc",
|
||||
"CMAKE_CXX_COMPILER": "g++",
|
||||
"CMAKE_C_COMPILER": "gcc-12",
|
||||
"CMAKE_CXX_COMPILER": "g++-12",
|
||||
"CMAKE_CXX_FLAGS_DEBUG": "-g -fPIC -Wall -pedantic -Wextra -Werror -Wno-missing-braces -Wno-ignored-qualifiers --std=c++20 -fdiagnostics-color -Wno-stringop-overread -DDEBUG",
|
||||
"CMAKE_INSTALL_PREFIX": "${sourceDir}/install/gcc/debug/sparkd"
|
||||
}
|
||||
|
||||
Submodule
+1
Submodule docker/linux added at 56c61231f3.
Submodule
+1
Submodule docker/windows added at de7d40c09d.
+1
-1
Submodule externals updated: cae5fd20da...1166db1211.
@@ -36,7 +36,7 @@ namespace Spark.Api.MarketData {
|
||||
public override int GetHashCode() =>
|
||||
security.Id.GetHashCode();
|
||||
public static Object ToJsonObj(BookTuple obj) => new {
|
||||
secuirty = obj.security.ToString(),
|
||||
security = obj.security.ToString(),
|
||||
book = obj.book.ToJsonObj()
|
||||
};
|
||||
public static bool operator ==(BookTuple left, BookTuple right) =>
|
||||
|
||||
@@ -67,7 +67,10 @@ namespace mbts.co2 {
|
||||
[DllImport("libspark")]
|
||||
internal unsafe static extern int md_consumer_read_trades(
|
||||
IntPtr consumer, UserSymbol channel, ref long cursor, raw_receive_pb cb);
|
||||
|
||||
[DllImport("libspark")]
|
||||
internal unsafe static extern int md_consumer_read_trades_linux(
|
||||
IntPtr consumer, string symbol, int venue, int type, ref long cursor, raw_receive_pb cb);
|
||||
|
||||
[DllImport("libspark")]
|
||||
internal unsafe static extern int md_consumer_read_rfqs(
|
||||
IntPtr consumer, UserSymbol channel, ref long cursor, raw_receive_pb cb);
|
||||
|
||||
@@ -23,7 +23,7 @@ namespace mbts.co2 {
|
||||
|
||||
public unsafe IList<TradeMessage> read_trades(ref long cursor) {
|
||||
IList<TradeMessage> trades = Array.Empty<TradeMessage>();
|
||||
Co2NativeMethods.md_consumer_read_trades(_consumer.Handle, _symbol, ref cursor, callback);
|
||||
Co2NativeMethods.md_consumer_read_trades_linux(_consumer.Handle, _symbol.symbol,(int)_symbol.exchange,(int)_symbol.type , ref cursor, callback);
|
||||
return trades;
|
||||
|
||||
void callback(byte* buffer, ulong size) =>
|
||||
|
||||
@@ -28,7 +28,7 @@ namespace mbts.spark {
|
||||
public TickDisplaySettingsImmutable qty_display => security_md.quantity_display;
|
||||
|
||||
public static System.Object ToJsonObj(TradeInfo obj) => new {
|
||||
secuirty = obj.security.Exchange.ToString() + "." + obj.security.ToString(),
|
||||
security = obj.security.Exchange.ToString() + "." + obj.security.ToString(),
|
||||
price = obj.trade.price(),
|
||||
timestamp = obj.trade.Time,
|
||||
quantity = obj.volume,
|
||||
|
||||
@@ -38,9 +38,9 @@
|
||||
var gwy = new bts.websocket_core.WebsocketConnectionGateway();
|
||||
var prms = new WebsocketServerConfig {
|
||||
DefaultEndPoint = "/",
|
||||
Host = "titan",
|
||||
Host = "0.0.0.0",
|
||||
Scheme = "http",
|
||||
Port = "8080",
|
||||
Port = "3000",
|
||||
KeepAliveIntervalSecs = 0.1,
|
||||
};
|
||||
gwy.init(sync);
|
||||
|
||||
@@ -424,6 +424,7 @@ namespace bts.websocket_core {
|
||||
/// </summary>
|
||||
/// <param name="server_params">The parameters for the websocket server includind the URL and endpoint,</param>
|
||||
public Task start_listening(WebsocketServerConfig server_params) {
|
||||
LoggerContext logger = new(Log.Gen, $"WebsocketGateway.start_listening ", "");
|
||||
// start listening, assert that init has been called already
|
||||
bts.Log.notnull(_sync);
|
||||
bts.Log.isnull(_server_params);
|
||||
@@ -433,6 +434,7 @@ namespace bts.websocket_core {
|
||||
_server_params = server_params;
|
||||
var builder = WebApplication.CreateBuilder();
|
||||
builder.WebHost.UseUrls(_server_params.Uri());
|
||||
logger.warn(_server_params.Uri());
|
||||
// var MyAllowSpecificOrigins = "_myAllowSpecificOrigins";
|
||||
builder.Services.AddLogging(builder => {
|
||||
builder.AddFilter("Microsoft", Microsoft.Extensions.Logging.LogLevel.Warning)
|
||||
|
||||
@@ -181,7 +181,7 @@ namespace bts.websocket_core {
|
||||
public void init(WebsocketConnectionGateway gtwy) {
|
||||
Gateway = gtwy;
|
||||
clients = new ClientsStore();
|
||||
TimeSpan md_throttle = new TimeSpan(10000);
|
||||
TimeSpan md_throttle = TimeSpan.FromSeconds(0.1);
|
||||
MD_Source = new MDSource(md_throttle);
|
||||
Md_Client = new(MD_Source);
|
||||
_securities_sub = MD_Source.subscribe_securities().Subscribe(sec_map => _securities = sec_map);
|
||||
|
||||
@@ -160,7 +160,7 @@ namespace bts.websocket_core {
|
||||
user_id = (uint)user.Id;
|
||||
BTSAccount? account = admin.Accounts.FirstOrDefault(x => x.Name.ToLowerInvariant() == accountname.ToLowerInvariant());
|
||||
if (account == null) {
|
||||
return (false, null, "account not corrent");
|
||||
return (false, null, "account not correct");
|
||||
}
|
||||
account_id = (uint)account.Id;
|
||||
if (!WebsocketJsonParsers.TryParse<Side>(
|
||||
|
||||
@@ -6,17 +6,19 @@ package bts.api.md;
|
||||
|
||||
option csharp_namespace = "bts.api.md";
|
||||
|
||||
message GlbxActivePlatform {
|
||||
string host = 1;
|
||||
string user = 2;
|
||||
string password = 3;
|
||||
}
|
||||
message GlbxOnixPlatform {
|
||||
string temp =1; // will add everything related to onix here.
|
||||
}
|
||||
|
||||
message GlbxMarketDataConfig {
|
||||
uint32 unique_check_value = 1;
|
||||
//uint32 num_depth_levels = 1;
|
||||
//uint32 depth_throttle_ms = 2;
|
||||
bts.api.domain.SecurityCriteria security_criteria = 3;
|
||||
// Market data isn't subscribed for securities listed here (even if they are
|
||||
// included by 'security_criteria'). There are some securities that don't
|
||||
// fit into our system just yet, such as 'BTCBIDR', which trades in prices
|
||||
// larger than our cpp 'domain::Price' class supports.
|
||||
//repeated string excluded_securities = 4;
|
||||
//string url_rest_base = 5;
|
||||
//string url_ws_base = 6;
|
||||
bts.api.domain.SecurityCriteria security_criteria = 2;
|
||||
oneof Platform {
|
||||
GlbxActivePlatform glbx_active_platform = 10;
|
||||
GlbxOnixPlatform glbx_onix_platform = 11;
|
||||
}
|
||||
}
|
||||
@@ -25,8 +25,23 @@ function(get_link_libraries dest)
|
||||
Packet++
|
||||
Common++
|
||||
ssh2
|
||||
ACE-5.7.5
|
||||
ActivBase_x86-64_ubuntu22-04_gcc123_mdid
|
||||
ActivOneApi_x86-64_ubuntu22-04_gcc123_mdid
|
||||
ActivAgentApplication_x86-64_ubuntu22-04_gcc123_mdid
|
||||
ActivDatabase_x86-64_ubuntu22-04_gcc123_mdid
|
||||
ActivFramework_x86-64_ubuntu22-04_gcc123_mdid
|
||||
ActivServices_x86-64_ubuntu22-04_gcc123_mdid
|
||||
BoostAtomic_x86-64_ubuntu22-04_gcc123_mdid
|
||||
BoostChrono_x86-64_ubuntu22-04_gcc123_mdid
|
||||
BoostProgramOptions_x86-64_ubuntu22-04_gcc123_mdid
|
||||
BoostFilesystem_x86-64_ubuntu22-04_gcc123_mdid
|
||||
-lm
|
||||
-lgcc_s
|
||||
-lc
|
||||
-lrt
|
||||
-ldl
|
||||
-lstdc++
|
||||
-lpcap
|
||||
-static-libgcc
|
||||
-static-libstdc++
|
||||
|
||||
@@ -23,7 +23,18 @@ TimeStamp::TimeStamp(uint32_t year, uint32_t month, uint32_t day) {
|
||||
value_ =
|
||||
std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count();
|
||||
}
|
||||
bool TimeStamp::from_ymd(
|
||||
std::string yearMonthDay, TimeStamp *result) {
|
||||
uint32_t year, month, day;
|
||||
std::istringstream ss(yearMonthDay);
|
||||
char dash; // to consume the dash character in the input
|
||||
|
||||
// Use stringstream to extract year, month, and day from the input string
|
||||
ss >> year >> dash >> month >> dash >> day;
|
||||
if(ss.fail())
|
||||
return false;
|
||||
return TimeStamp::from_ymd(year,month,day,result);
|
||||
}
|
||||
bool TimeStamp::from_ymd(
|
||||
std::uint32_t year, std::uint32_t month, std::uint32_t day, TimeStamp *result) {
|
||||
if (!is_valid_ymd(year, month, day)) {
|
||||
|
||||
@@ -41,6 +41,9 @@ struct TimeStamp {
|
||||
std::uint32_t month,
|
||||
std::uint32_t day,
|
||||
TimeStamp *result);
|
||||
static bool from_ymd(
|
||||
std::string yearMonthDay,
|
||||
TimeStamp *result);
|
||||
|
||||
constexpr std::uint64_t nanos() const;
|
||||
constexpr std::uint64_t micros() const;
|
||||
|
||||
@@ -128,6 +128,32 @@ CAPI bool md_consumer_is_heart_beating(const ColdHeader *header) {
|
||||
return is_heart_beat_ok(*header);
|
||||
}
|
||||
|
||||
typedef void (*raw_receive_pb)(const char *, size_t);
|
||||
CAPI int md_consumer_read_trades_linux(
|
||||
MDConsumer *consumer,
|
||||
const char* symbol,
|
||||
int venue_i,
|
||||
int type_i,
|
||||
int64_t &cursor,
|
||||
raw_receive_pb cb) {
|
||||
auto type = static_cast<SecurityType>(type_i);
|
||||
auto venue = static_cast<Venue>(venue_i);
|
||||
bts_assert(consumer);
|
||||
google::protobuf::Arena arena;
|
||||
co2::TradeList *ptrades =
|
||||
google::protobuf::Arena::CreateMessage<co2::TradeList>(&arena);
|
||||
auto channel =
|
||||
consumer->get_channel(symbol, type, venue);
|
||||
if (channel.is_initialized()) {
|
||||
cursor = channel.read_trades(*ptrades->mutable_items(), cursor);
|
||||
std::string buffer;
|
||||
ptrades->SerializeToString(&buffer);
|
||||
cb(buffer.c_str(), buffer.size());
|
||||
return ptrades->items_size();
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
typedef void (*raw_receive_pb)(const char *, size_t);
|
||||
CAPI int md_consumer_read_trades(
|
||||
MDConsumer *consumer,
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
#if !defined(_WIN32) //&& defined(SPARK_ENABLE_CCAPI)
|
||||
#include "lib/api/md/GlbxMarketDataConfig.pb.h"
|
||||
#if !defined(_WIN32) && defined(SPARK_ENABLE_CCAPI)
|
||||
#include "MarketDataGateway.h"
|
||||
#include "lib/api/md/EmumbaDeribitMarketDataConfig.pb.h"
|
||||
#include "lib/api/md/GlbxMarketDataConfig.pb.h"
|
||||
#include "lib/api/md/MarketDataConfig.pb.h"
|
||||
#include "lib/api/md/MarketDataConfigUtils.h"
|
||||
#include "lib/api/md/MarketDataReplayConfig.pb.h"
|
||||
@@ -40,6 +42,7 @@
|
||||
#include "lib/utils/thread_manager.h"
|
||||
#include "server/sparkd/md/platforms/emumba/EmumbaMain.h"
|
||||
#include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
|
||||
#include "lib/md/cme/CMEActivSession.h"
|
||||
|
||||
#include <bits/stdint-uintn.h>
|
||||
#include <boost/filesystem.hpp>
|
||||
@@ -135,15 +138,19 @@ void MarketDataGateway::run() {
|
||||
}
|
||||
LOG(md, Info, "MarketDataGateway::run starting t_id:{} t_name:{}",
|
||||
os::get_thread_id(), os::get_thread_name());
|
||||
|
||||
_start_cc(me_->md_config);
|
||||
|
||||
if (!me_->md_config.emumba_deribit().empty()) {
|
||||
bts_assert(me_->md_config.emumba_deribit().size() == 1);
|
||||
_start_emumba_deribit(me_->md_config.emumba_deribit().at(0));
|
||||
}
|
||||
|
||||
if (!me_->md_config.glbx().empty()) {
|
||||
bts_assert(me_->md_config.glbx().size() == 1);
|
||||
_start_onix_glbx(me_->md_config.glbx().at(0));
|
||||
_start_glbx(me_->md_config.glbx().at(0));
|
||||
}
|
||||
|
||||
_start_replays(me_->md_config);
|
||||
while (true) {
|
||||
std::this_thread::sleep_for(std::chrono::seconds(1));
|
||||
@@ -239,7 +246,26 @@ void MarketDataGateway::unsubscribe(MarketDataListenerID id) {
|
||||
LOG(md, Info, "Unsubscribing listener {}", id);
|
||||
me_->listeners.erase(id);
|
||||
}
|
||||
/// glbx //////////////////////////////////////////////////////////////////////
|
||||
/// onix active //////////////////////////////////////////////////////////////////////
|
||||
//////////////////////////////////////////////////////////////////////////////
|
||||
static void run_active(bts::api::md::GlbxMarketDataConfig cfg, persistence::IDGenerator *sec_id_store) {
|
||||
LOG(md, Info, "MarketDataGateway::run_active t_id:{} t_name", os::get_thread_id(),
|
||||
os::get_thread_name());
|
||||
bts::md::GeneralSecurityIndex sec_index;
|
||||
bts::md::MDProducer md_producer;
|
||||
std::shared_ptr<GeneralMarketDataListener> listener =std::make_shared<bts::md::GeneralMarketDataListener>(sec_index,md_producer,bts::api::VENUE_GLBX);
|
||||
bts::md::cme::CMEActivSession::run(cfg.glbx_active_platform().host(), cfg.glbx_active_platform().user(), cfg.glbx_active_platform().password(), listener, sec_id_store);
|
||||
// Timers
|
||||
// auto &disp = utils::Dispatcher::current();
|
||||
// utils::DispatcherTimer heartbeat_timer(disp, 250, [&]() {
|
||||
// for (auto &p : platforms)
|
||||
// p->send_heartbeats();
|
||||
// });
|
||||
// heartbeat_timer.start();
|
||||
|
||||
// disp.run();
|
||||
}
|
||||
/// onix glbx //////////////////////////////////////////////////////////////////////
|
||||
//////////////////////////////////////////////////////////////////////////////
|
||||
static void run_onixs() {
|
||||
LOG(md, Info, "MarketDataGateway::run_onixs t_id:{} t_name",
|
||||
@@ -268,8 +294,29 @@ static void run_onixs() {
|
||||
disp.run();
|
||||
}
|
||||
|
||||
void MarketDataGateway::_start_onix_glbx(
|
||||
bts::api::md::GlbxMarketDataConfig cfg) {
|
||||
void MarketDataGateway::_start_active_glbx(bts::api::md::GlbxMarketDataConfig cfg){
|
||||
LOG(md, Info,
|
||||
"MarketDataGateway::_start_active_glbx starting t_id:{} t_name:{}. "
|
||||
"cfg:\n{}",
|
||||
os::get_thread_id(), os::get_thread_name(), cfg.DebugString());
|
||||
LOG(md, Info,
|
||||
"MarketDataGateway::_start_active_glbx");
|
||||
bts::md::run_active(cfg, &me_->sec_id_store);
|
||||
}
|
||||
void MarketDataGateway::_start_glbx( bts::api::md::GlbxMarketDataConfig cfg){
|
||||
LOG(md, Info,
|
||||
"MarketDataGateway::_start_glbx starting t_id:{} t_name:{}. "
|
||||
"cfg:\n{}",
|
||||
os::get_thread_id(), os::get_thread_name(), cfg.DebugString());
|
||||
if(cfg.has_glbx_active_platform()){
|
||||
_start_active_glbx(cfg);
|
||||
}
|
||||
else if(cfg.has_glbx_onix_platform()){
|
||||
_start_onix_glbx(cfg);
|
||||
}
|
||||
|
||||
}
|
||||
void MarketDataGateway::_start_onix_glbx( bts::api::md::GlbxMarketDataConfig cfg){
|
||||
LOG(md, Info,
|
||||
"MarketDataGateway::_start_onix_glbx starting t_id:{} t_name:{}. "
|
||||
"cfg:\n{}",
|
||||
|
||||
@@ -35,6 +35,10 @@ class MarketDataGateway {
|
||||
bts::api::md::EmumbaDeribitMarketDataConfig cfg);
|
||||
static void _start_onix_glbx(
|
||||
bts::api::md::GlbxMarketDataConfig cfg);
|
||||
void _start_glbx(
|
||||
bts::api::md::GlbxMarketDataConfig cfg);
|
||||
void _start_active_glbx(
|
||||
bts::api::md::GlbxMarketDataConfig cfg);
|
||||
void _start_replays(const bts::api::md::MarketDataConfig &cfg);
|
||||
};
|
||||
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
#include "CMEActivSession.h"
|
||||
#include "lib/logging/logging.h"
|
||||
#include "lib/md/GeneralMarketDataListener.h"
|
||||
#include "lib/md/cme/CMEActivSessionHandler.h"
|
||||
#include "lib/md/cme/CMEActivSubscriptionHandler.h"
|
||||
#include "lib/md/cme/CMEActiveQueryHandler.h"
|
||||
#include "lib/md/cme/CMEActiveSnapshotHandler.h"
|
||||
#include <ActivMiddleware/ActivBase/Exception.h>
|
||||
#include <ActivMiddleware/ActivBase/FieldTypes/ActivFieldTypes.h>
|
||||
#include <ActivMiddleware/ActivBase/FieldTypes/FieldMap.h>
|
||||
#include <ActivOneApi/Constants/FieldIds.h>
|
||||
#include <ActivOneApi/Session.h>
|
||||
#include <exception>
|
||||
#include <ActivOneApi/Constants/DataSourceIds.h>
|
||||
|
||||
namespace bts::md::cme {
|
||||
|
||||
using Activ::FieldMap;
|
||||
using Activ::OneApi::Session;
|
||||
using Activ::OneApi::Handle;
|
||||
using std::string;
|
||||
void CMEActivSession::run(const string &host, const string &user, const string &password,std::shared_ptr<GeneralMarketDataListener> listner, persistence::IDGenerator *sec_id_store) {
|
||||
try {
|
||||
// Session callback handler.
|
||||
CMEActivSessionHandler sessionHandler;
|
||||
|
||||
FieldMap fields;
|
||||
// Construct the Session with a default configuration passing in our sessionHandler.
|
||||
Session session(fields, sessionHandler);
|
||||
|
||||
FieldMap connectParameters;
|
||||
|
||||
connectParameters.Set(Activ::OneApi::FID_HOST, host.c_str());
|
||||
connectParameters.Set(Activ::OneApi::FID_USER_ID, user.c_str());
|
||||
connectParameters.Set(Activ::OneApi::FID_PASSWORD, password.c_str());
|
||||
|
||||
// Connect synchronously with a 10 second timeout (will throw on failure).
|
||||
session.Connect(connectParameters, 10000);
|
||||
std::cerr << "connected" << std::endl;
|
||||
|
||||
CMEActivQueryHandler query_handler(listner, sec_id_store);
|
||||
// Specifying a data source and symbol.
|
||||
// Activ::OneApi::Handle handle = session.Subscribe(Activ::OneApi::DATA_SOURCE_ACTIV,"CSIQ.Q",subscriptionHandler);
|
||||
|
||||
// Activ::OneApi::Handle handle = session.Query(DATA_SOURCE_ACTIV,
|
||||
// "type=option and country=US and symbol=KC*",
|
||||
// subscriptionHandler);
|
||||
|
||||
md::cme::SnapshotHandler snapshotHandler;
|
||||
|
||||
// Request snapshot of the Microsoft US Equity topic.
|
||||
// Returned Handle for a Snapshot is only required to cancel the request prior to a response.
|
||||
// (void) session.Snapshot(DATA_SOURCE_ACTIV, "type=option and country=US and symbol=CSIQ/*", snapshotHandler);
|
||||
|
||||
// (void) session.Query(DATA_SOURCE_ACTIV, "symbol=CSIQ/*", query_handler);
|
||||
(void) session.Query(DATA_SOURCE_ACTIV, "type=future and symbol=NK/*", query_handler);
|
||||
// QueryResults queryResults;
|
||||
// std::string query;
|
||||
// while(true)
|
||||
// {
|
||||
// query ="";
|
||||
// std::getline(std::cin, query);
|
||||
// if(query.substr(0,3) == "end")
|
||||
// break;
|
||||
// session.Query(DATA_SOURCE_ACTIV,
|
||||
// query,
|
||||
// queryResults);
|
||||
|
||||
// // Prin t out the topics.
|
||||
// for (const auto &message : queryResults)
|
||||
// {
|
||||
// std::cout << "Found symbol " << message.GetSymbol() << std::endl;
|
||||
// }
|
||||
// std::cout<<"END "<<std::endl;
|
||||
// }
|
||||
// Alternatively an asynchronous connection can be used.
|
||||
// Connection completion is notified via ISessionHandler::OnSessionConnect().
|
||||
// session.AsyncConnect(connectParameters);
|
||||
session.Run();
|
||||
|
||||
} catch (const Activ::Exception &ex) {
|
||||
std::cerr << "Error running CMEActivSession (activ): " << ex.GetMessage()
|
||||
<< ex.ToString()<< std::endl;
|
||||
} catch (const std::exception &ex) {
|
||||
std::cerr << "Error running CMEActivSession: " << ex.what() << std::endl;
|
||||
// LOG(md, Error, "Error running CMEActivSession: {}",
|
||||
// std::string{ex.what()});
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace bts::md::cme
|
||||
@@ -0,0 +1,19 @@
|
||||
#pragma once
|
||||
|
||||
#include "lib/md/GeneralMarketDataListener.h"
|
||||
#include "lib/persistence/IDGenerator.h"
|
||||
#include <string>
|
||||
|
||||
namespace bts::md::cme {
|
||||
|
||||
class CMEActivSession {
|
||||
public:
|
||||
static void run(
|
||||
const std::string &host,
|
||||
const std::string &user,
|
||||
const std::string &password,
|
||||
std::shared_ptr<GeneralMarketDataListener> listner,
|
||||
persistence::IDGenerator *sec_id_store);
|
||||
};
|
||||
|
||||
} // namespace bts::md::cme
|
||||
@@ -0,0 +1,33 @@
|
||||
#include "CMEActivSessionHandler.h"
|
||||
#include <ActivMiddleware/ActivBase/ActivBase.h>
|
||||
#include <ActivOneApi/Session.h>
|
||||
#include "CMEActivSubscriptionHandler.h"
|
||||
|
||||
namespace bts::md::cme {
|
||||
|
||||
using Activ::LogType;
|
||||
using Activ::StatusCode;
|
||||
using Activ::String;
|
||||
using Activ::OneApi::Session;
|
||||
|
||||
void CMEActivSessionHandler::OnSessionLogMessage(
|
||||
Session & /*session*/, const LogType log_type, const String &message) {
|
||||
std::cout << Activ::LogTypeToString(log_type) << ": " << message
|
||||
<< std::endl;
|
||||
}
|
||||
|
||||
void CMEActivSessionHandler::OnSessionConnect(Session & /*session*/) {
|
||||
std::cerr << "connected" << std::endl;
|
||||
}
|
||||
|
||||
void CMEActivSessionHandler::OnSessionDisconnect(Session & /*session*/) {
|
||||
std::cout << "disconnected" << std::endl;
|
||||
}
|
||||
|
||||
void CMEActivSessionHandler::OnSessionError(
|
||||
Session & /*session*/, const StatusCode statusCode) {
|
||||
std::string err{Activ::StatusCodeToString(statusCode)};
|
||||
std::cerr << "CMEActivSessionHandler::OnSessionError. " << err << std::endl;
|
||||
}
|
||||
|
||||
} // namespace bts::md::cme
|
||||
@@ -0,0 +1,24 @@
|
||||
#pragma once
|
||||
|
||||
#include <ActivMiddleware/ActivBase/ActivBase.h>
|
||||
#include <ActivOneApi/ISessionHandler.h>
|
||||
|
||||
namespace bts::md::cme {
|
||||
|
||||
class CMEActivSessionHandler : public Activ::OneApi::ISessionHandler {
|
||||
public:
|
||||
static void run();
|
||||
|
||||
void OnSessionLogMessage(
|
||||
Activ::OneApi::Session &session,
|
||||
const Activ::LogType log_type,
|
||||
const Activ::String &message) override;
|
||||
|
||||
void OnSessionConnect(Activ::OneApi::Session &session) override;
|
||||
void OnSessionDisconnect(Activ::OneApi::Session &session) override;
|
||||
void OnSessionError(
|
||||
Activ::OneApi::Session &session,
|
||||
const Activ::StatusCode statusCode) override;
|
||||
};
|
||||
|
||||
} // namespace bts::md::cme
|
||||
@@ -0,0 +1,365 @@
|
||||
#pragma once
|
||||
#include "lib/core/TimeStamp.h"
|
||||
#include "lib/domain/AggressorSide.h"
|
||||
#include "lib/domain/Book.h"
|
||||
#include "lib/domain/BookLevel.h"
|
||||
#include "lib/domain/FutureSecurity.h"
|
||||
#include "lib/domain/Quantity.h"
|
||||
#include "lib/domain/SecurityID.h"
|
||||
#include "lib/domain/Side.h"
|
||||
#include "lib/domain/TradeRecord.h"
|
||||
#include "lib/md/GeneralMarketDataListener.h"
|
||||
#include "lib/md/MarketDataGateway.h"
|
||||
#include "lib/orderman/orderman_utils/Book.h"
|
||||
#include "lib/persistence/IDGenerator.h"
|
||||
#include "lib/utils/bts_assert.h"
|
||||
#include <ActivOneApi/MarketData/EventTypes.h>
|
||||
#include <ActivOneApi/MarketData/FieldIds.h>
|
||||
#include <ActivOneApi/MarketData/Separators.h>
|
||||
#include <cassert>
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
#define UNUSED(x) (void)(x)
|
||||
#include "ActivOneApi/MarketData/MarketData.h"
|
||||
#include <ActivMiddleware/ActivBase/ActivBase.h>
|
||||
|
||||
#include "ActivOneApi/OneApi.h"
|
||||
#include <ActivOneApi/ISessionHandler.h>
|
||||
#include <ActivOneApi/ISubscriptionHandler.h>
|
||||
#include <ActivOneApi/Messages/RefreshMessage.h>
|
||||
#include <ActivOneApi/Messages/UpdateMessage.h>
|
||||
using namespace Activ;
|
||||
using namespace Activ::OneApi;
|
||||
namespace bts::md::cme {
|
||||
class CMEActivSubscriptionHandler : public Activ::OneApi::ISubscriptionHandler {
|
||||
|
||||
FieldListValidator m_fieldListValidator;
|
||||
|
||||
Activ::OneApi::TopicSubscriptionState *m_state;
|
||||
|
||||
public:
|
||||
std::shared_ptr<GeneralMarketDataListener> listner;
|
||||
persistence::IDGenerator *sec_id_store;
|
||||
|
||||
void init(
|
||||
std::shared_ptr<GeneralMarketDataListener> listner,
|
||||
persistence::IDGenerator *id_store,
|
||||
Activ::OneApi::TopicSubscriptionState *m_state) {
|
||||
this->listner = listner;
|
||||
sec_id_store = id_store;
|
||||
this->m_state = m_state;
|
||||
}
|
||||
CMEActivSubscriptionHandler() {}
|
||||
CMEActivSubscriptionHandler(
|
||||
std::shared_ptr<GeneralMarketDataListener> listner,
|
||||
persistence::IDGenerator *sec_id_store,
|
||||
Activ::OneApi::TopicSubscriptionState *m_state)
|
||||
: m_state(m_state), listner(listner), sec_id_store(sec_id_store) {}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
std::string removeCharacters(const std::string &input) {
|
||||
std::string result;
|
||||
|
||||
for (char c : input) {
|
||||
if (c != '/' && c != '.') {
|
||||
result += c;
|
||||
} else if (c == '.') {
|
||||
break; // Stop adding characters after encountering a dot
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
void OnSubscriptionRefresh(
|
||||
const RefreshMessage &refreshMessage, const Context &context) {
|
||||
UNUSED(context);
|
||||
std::cout << "REFRESH received for " << refreshMessage.m_symbol
|
||||
<< std::endl;
|
||||
std::cout << "Symbol .................... " << refreshMessage.m_symbol
|
||||
<< std::endl;
|
||||
std::cout << "Topic subscription state .. "
|
||||
<< TopicSubscriptionStateToString(
|
||||
refreshMessage.m_topicSubscriptionState)
|
||||
<< std::endl;
|
||||
StatusCode statusCode = m_fieldListValidator.Initialize(refreshMessage);
|
||||
|
||||
for (const Activ::OneApi::FlvDetail::Field &field :
|
||||
m_fieldListValidator) {
|
||||
if(field.IsDefined())
|
||||
std::cout<<"onsub Feild: "<<MarketData::GetFieldName(field.GetId())<<", Value: "<<field.GetValue<std::string>() <<std::endl;
|
||||
}
|
||||
|
||||
if (STATUS_CODE_SUCCESS != statusCode) {
|
||||
std::cout << "ERROR decoding field data for "
|
||||
<< refreshMessage.m_symbol << " - "
|
||||
<< StatusCodeToString(statusCode) << std::endl;
|
||||
return;
|
||||
}
|
||||
const Activ::OneApi::FlvDetail::Field *pField;
|
||||
statusCode = m_fieldListValidator.GetField(
|
||||
MarketData::FID_EXPIRATION_DATE, pField);
|
||||
|
||||
if (STATUS_CODE_SUCCESS != statusCode) {
|
||||
std::cout << "ERROR fetching "
|
||||
<< "exp date"
|
||||
<< " field for " << refreshMessage.m_symbol << " - "
|
||||
<< StatusCodeToString(statusCode) << std::endl;
|
||||
return;
|
||||
}
|
||||
std::string exp_date = pField->GetValue<String>();
|
||||
core::TimeStamp exp_time;
|
||||
if (!core::TimeStamp::from_ymd(exp_date, &exp_time)) {
|
||||
return;
|
||||
}
|
||||
|
||||
String result = removeCharacters(refreshMessage.m_symbol);
|
||||
domain::SecurityID sec_id{
|
||||
sec_id_store->get_or_add(api::domain::VENUE_GLBX, result)};
|
||||
std::vector<domain::Security> complete_security_list;
|
||||
domain::Security &sec = complete_security_list.emplace_back();
|
||||
sec.security_common.venue = api::domain::VENUE_GLBX;
|
||||
sec.security_common.id = sec_id;
|
||||
sec.security_common.exchange_sec_id =
|
||||
refreshMessage.ToString(); // Need to fix this
|
||||
sec.security_common.desc = result;
|
||||
sec.security_common.symbol = result;
|
||||
sec.security_common.asset = "a1";
|
||||
sec.security_common.group = result;
|
||||
|
||||
sec.security_common.min_price_increment = 0.1;
|
||||
sec.security_common.min_quantity = 1;
|
||||
sec.security_common.min_quantity_increment = 0.1;
|
||||
sec.security_common.currency = "name";
|
||||
auto &t = sec.security.emplace<domain::FutureSecurity>();
|
||||
t.expiration = exp_time;
|
||||
std::vector<const domain::Security *> const_security_list;
|
||||
for (const auto &security : complete_security_list) {
|
||||
const_security_list.push_back(&security);
|
||||
}
|
||||
listner->on_securities(const_security_list);
|
||||
|
||||
|
||||
// Alternatively use random access.
|
||||
|
||||
// Print out the current value for FID_TRADE. FID_TRADE is defined in the Activ::MarketData namespace.
|
||||
// ProcessField(
|
||||
// MarketData::FID_EXPIRATION_DATE, "Exp ", refreshMessage.m_symbol);
|
||||
}
|
||||
template <typename T>
|
||||
bool _get_field(String field_name, Activ::OneApi::FieldId fieldId, T &res ){
|
||||
const Activ::OneApi::FlvDetail::Field *pField;
|
||||
auto statusCode =
|
||||
m_fieldListValidator.GetField(fieldId, pField);
|
||||
if (STATUS_CODE_SUCCESS != statusCode) {
|
||||
std::cout << "ERROR fetching "
|
||||
<< field_name
|
||||
<< " field Error: "
|
||||
<< StatusCodeToString(statusCode) << std::endl;
|
||||
return false;
|
||||
}
|
||||
res = pField->GetValue<T>();
|
||||
return true;
|
||||
}
|
||||
bool _parse_trade(std::vector<const domain::TradeRecord *> &trades, domain::SecurityID id) {
|
||||
UNUSED(trades);
|
||||
for (const Activ::OneApi::FlvDetail::Field &field :
|
||||
m_fieldListValidator) {
|
||||
if (field.IsDefined()) {
|
||||
|
||||
std::cout << "Has Feild: "
|
||||
<< MarketData::GetFieldName(field.GetId())
|
||||
<< ", Value: " << field.GetValue<std::string>()
|
||||
<< std::endl;
|
||||
}
|
||||
}
|
||||
std::vector<domain::TradeRecord> parsed_trades;
|
||||
auto &t = parsed_trades.emplace_back();
|
||||
double tp;
|
||||
if(!_get_field<double>("FID_TRADE", MarketData::FID_TRADE, tp))
|
||||
return false;
|
||||
uint ts;
|
||||
if(!_get_field<uint>("FID_TRADE_SIZE", MarketData::FID_TRADE_SIZE, ts))
|
||||
return false;
|
||||
String time, date;
|
||||
if(!_get_field<String>("FID_TRADE_TIME", MarketData::FID_TRADE_TIME, time))
|
||||
return false;
|
||||
if(!_get_field<String>("FID_TRADE_DATE", MarketData::FID_TRADE_DATE, date))
|
||||
return false;
|
||||
|
||||
t.trade_price=bts::domain::Price::from_double(tp);
|
||||
t.trade_quantity=bts::domain::Quantity(ts);
|
||||
t.sec_id=id;
|
||||
t.agg_side=bts::domain::AggressorSide::Buy;
|
||||
t.time_stamp=core::TimeStamp::from_iso_string(date+"T"+time);;
|
||||
trades.push_back(&parsed_trades[0]);
|
||||
bts_assert(trades[0]->sec_id==id);
|
||||
listner->on_trades(trades);
|
||||
return true;
|
||||
}
|
||||
// void _parse_side()
|
||||
bool _parse_book(domain::Book &result, size_t n_bid_levels = 5) {
|
||||
UNUSED(n_bid_levels);
|
||||
uint ask_size, bid_size;
|
||||
if(!_get_field<uint>("FID_ASK_SIZE", MarketData::FID_ASK_SIZE, ask_size))
|
||||
return false;
|
||||
if(!_get_field<uint>("FID_BID_SIZE", MarketData::FID_BID_SIZE, bid_size))
|
||||
return false;
|
||||
double bid_val,ask_val;
|
||||
if(!_get_field<double>("FID_BID", MarketData::FID_BID, bid_val))
|
||||
return false;
|
||||
if(!_get_field<double>("FID_ASK", MarketData::FID_ASK, ask_val))
|
||||
return false;
|
||||
domain::BookSideT<10> &b = result.bids();
|
||||
domain::BookSideT<10> &a = result.asks();
|
||||
domain::BookLevel b_l;
|
||||
b_l.price=domain::Price::from_double(bid_val);
|
||||
b_l.total_quantity=domain::Quantity(bid_size);
|
||||
domain::BookLevel a_l;
|
||||
a_l.price=domain::Price::from_double(ask_val);
|
||||
a_l.total_quantity=domain::Quantity(ask_size);
|
||||
b[0]=b_l;
|
||||
a[0]=a_l;
|
||||
|
||||
return true;
|
||||
}
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
void ProcessField(
|
||||
const FieldId fieldId, const String &fieldName, const String &symbol) {
|
||||
const FieldListValidator::Field *pField;
|
||||
|
||||
// FieldListValidator::GetField() provides access to the internally stored cached field.
|
||||
StatusCode statusCode = m_fieldListValidator.GetField(fieldId, pField);
|
||||
if (STATUS_CODE_SUCCESS != statusCode) {
|
||||
std::cout << "ERROR fetching " << fieldName << " field for " << symbol
|
||||
<< " - " << StatusCodeToString(statusCode) << std::endl;
|
||||
return;
|
||||
}
|
||||
|
||||
std::cout << fmt::format("{:.<27s} ", fieldName + " ");
|
||||
|
||||
if (!pField->IsDefined()) {
|
||||
std::cout << "undefined" << std::endl;
|
||||
} else {
|
||||
// Field contents are only valid if the Field is defined.
|
||||
std::cout << pField->GetValue<String>() << std::endl;
|
||||
}
|
||||
}
|
||||
void OnSubscriptionUpdate(
|
||||
const UpdateMessage &updateMessage, const Context &context) {
|
||||
UNUSED(context);
|
||||
m_fieldListValidator.ApplyUpdate(updateMessage);
|
||||
String result = removeCharacters(updateMessage.m_symbol);
|
||||
auto id = sec_id_store->get_id(api::domain::VENUE_GLBX, result);
|
||||
assert(id.has_value());
|
||||
|
||||
for (const Activ::OneApi::FlvDetail::Field &field :
|
||||
m_fieldListValidator) {
|
||||
if(field.IsDefined())
|
||||
std::cout<<"Update Feild: "<<MarketData::GetFieldName(field.GetId())<<", Value: "<<field.GetValue<std::string>() <<std::endl;
|
||||
}
|
||||
switch (updateMessage.m_eventType) {
|
||||
case MarketData::EVENT_TYPE_BBO_QUOTE:
|
||||
case MarketData::EVENT_TYPE_BBO_DEPTH: {
|
||||
domain::Book book;
|
||||
book.set_security_id(domain::SecurityID(id.value()));
|
||||
if (_parse_book(book)) {
|
||||
listner->on_book(book);
|
||||
std::cout << "Book UPDATE received for " << updateMessage.m_symbol
|
||||
<< std::endl;
|
||||
}
|
||||
break;
|
||||
}
|
||||
case MarketData::EVENT_TYPE_TRADE:
|
||||
{
|
||||
std::vector<const domain::TradeRecord *> trades;
|
||||
if(_parse_trade(trades,domain::SecurityID(id.value()))){
|
||||
|
||||
std::cout << "trade UPDATE received for " << updateMessage.m_symbol
|
||||
<< std::endl;
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
std::cout << "case not defined \n";
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnSubscriptionTopicStatus(
|
||||
const TopicStatusMessage &topicStatusMessage, const Context &context) {
|
||||
UNUSED(context);
|
||||
std::cout << "TOPIC STATUS received for " << topicStatusMessage.m_symbol
|
||||
<< " "
|
||||
<< TopicSubscriptionStateToString(
|
||||
topicStatusMessage.m_topicSubscriptionState)
|
||||
<< std::endl;
|
||||
|
||||
*m_state = topicStatusMessage.m_topicSubscriptionState;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnSubscriptionError(
|
||||
const StatusCode statusCode, const Context &context) {
|
||||
UNUSED(context);
|
||||
std::cout << "Subscription error " << StatusCodeToString(statusCode)
|
||||
<< std::endl;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnSubscriptionFailure(
|
||||
const StatusCode statusCode, const Context &context) {
|
||||
UNUSED(context);
|
||||
std::cout << "Subscription failure " << StatusCodeToString(statusCode)
|
||||
<< std::endl;
|
||||
}
|
||||
// void OnSubscriptionUpdate(
|
||||
// const Activ::OneApi::UpdateMessage &updateMessage,
|
||||
// const Activ::OneApi::Context &context) {
|
||||
// UNUSED(context);
|
||||
// std::cout << "UPDATE: " << updateMessage.m_symbol
|
||||
// << updateMessage.ToString() << std::endl;
|
||||
|
||||
// // ApplyUpdate is used here, unlike Initialize, ApplyUpdate does not clear existing
|
||||
// // field content.
|
||||
// // m_fieldListValidator.ApplyUpdate(updateMessage);
|
||||
|
||||
// // ... use m_fieldListValidator to access content.
|
||||
// }
|
||||
// void OnSubscriptionRefresh(
|
||||
// const Activ::OneApi::RefreshMessage &refreshMessage,
|
||||
// const Activ::OneApi::Context &context) {
|
||||
// UNUSED(context);
|
||||
// std::cout << "REFRESH: " << refreshMessage.m_symbol << std::endl;
|
||||
|
||||
// // Access to field data is obtained using a FieldListValidator.
|
||||
// // m_fieldListValidator.Initialize(refreshMessage);
|
||||
|
||||
// // ... use m_fieldListValidator to access content.
|
||||
// }
|
||||
|
||||
// void OnSubscriptionTopicStatus(
|
||||
// const TopicStatusMessage &topicStatusMessage, const Context &context) {
|
||||
// UNUSED(context);
|
||||
// // If topicStatusMessage.m_topicSubscriptionState is not TOPIC_SUBSCRIPTION_STATE_OK
|
||||
// // then any cached fields may be stale.
|
||||
// std::cout << "TOPIC STATUS: " << topicStatusMessage.m_symbol
|
||||
// << ", state=" << topicStatusMessage.ToString() << std::endl;
|
||||
// }
|
||||
// void OnSubscriptionError(
|
||||
// const StatusCode statusCode, const Context &context) {
|
||||
// std::cout << "error :" << StatusCodeToString(statusCode) << std::endl;
|
||||
// UNUSED(statusCode);
|
||||
// UNUSED(context);
|
||||
// }
|
||||
// void OnSubscriptionFailure(
|
||||
// const StatusCode statusCode, const Context &context) {
|
||||
// std::cout << "Failure :" << StatusCodeToString(statusCode) << std::endl;
|
||||
// UNUSED(statusCode);
|
||||
// UNUSED(context);
|
||||
// }
|
||||
};
|
||||
|
||||
} // namespace bts::md::cme
|
||||
@@ -0,0 +1,108 @@
|
||||
#pragma once
|
||||
#include "lib/md/cme/CMEActivSubscriptionHandler.h"
|
||||
#include "lib/md/GeneralMarketDataListener.h"
|
||||
#include <ActivMiddleware/CxxYZ/unordered_map.h>
|
||||
#define UNUSED(x) (void)(x)
|
||||
#include "ActivOneApi/OneApi.h"
|
||||
namespace bts::md::cme {
|
||||
class CMEActivQueryHandler : public Activ::OneApi::ITopicQueryHandler {
|
||||
class TopicDetails {
|
||||
public:
|
||||
//! The subscription handle for this topic
|
||||
Activ::OneApi::ScopedHandle m_handle;
|
||||
CMEActivSubscriptionHandler sub_handler;
|
||||
//! The current topic subscription state. Is this topic waiting for a refresh.
|
||||
Activ::OneApi::TopicSubscriptionState m_state;
|
||||
|
||||
/**
|
||||
* @brief Has this topic been seen for the current topic query. Whilst a query is in an error state there may be missed
|
||||
* topic deletions, these need to be cleaned up once the OnTopicQueryComplete callback is received.
|
||||
*/
|
||||
size_t m_queryVersion;
|
||||
};
|
||||
//! Used to track how many times the query has restarted.
|
||||
// size_t m_queryVersion;
|
||||
//! Map of subscriptions.
|
||||
typedef Activ::CxxYZ::unordered_map<Activ::String, TopicDetails> SubscriptionMap;
|
||||
SubscriptionMap m_subscriptionMap;
|
||||
//! Used to track how many times the query has restarted.
|
||||
size_t m_queryVersion;
|
||||
public:
|
||||
std::shared_ptr<GeneralMarketDataListener> listner;
|
||||
persistence::IDGenerator *sec_id_store;
|
||||
CMEActivQueryHandler(
|
||||
std::shared_ptr<GeneralMarketDataListener> listner,
|
||||
persistence::IDGenerator *sec_id_store)
|
||||
: m_queryVersion(0), listner(listner), sec_id_store(sec_id_store) {}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnTopicQueryStart(const Context &context) {
|
||||
UNUSED(context);
|
||||
std::cout << "QUERY start" << std::endl;
|
||||
++m_queryVersion;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnTopicQueryAdd(Topic &topic, const Context &context) {
|
||||
const String &symbol = topic.GetSymbol();
|
||||
TopicDetails &topicDetails = m_subscriptionMap[symbol];
|
||||
|
||||
if (!topicDetails.m_handle) {
|
||||
topicDetails.m_queryVersion = m_queryVersion;
|
||||
topicDetails.m_state = TOPIC_SUBSCRIPTION_STATE_PENDING;
|
||||
topicDetails.sub_handler = CMEActivSubscriptionHandler(listner, sec_id_store, &topicDetails.m_state);
|
||||
topicDetails.m_handle =
|
||||
context.GetSession().Subscribe(DATA_SOURCE_ACTIV, symbol, topicDetails.sub_handler);
|
||||
} else {
|
||||
topicDetails.m_queryVersion = m_queryVersion;
|
||||
}
|
||||
}
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnTopicQueryRemove(const TopicId &topicId, const Context &context) {
|
||||
UNUSED(context);
|
||||
std::cout << "QUERY topic removed: " << topicId.m_symbol << std::endl;
|
||||
|
||||
SubscriptionMap::iterator iter = m_subscriptionMap.find(topicId.m_symbol);
|
||||
if (m_subscriptionMap.end() != iter)
|
||||
m_subscriptionMap.erase(iter);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnTopicQueryComplete(const Context &context) {
|
||||
std::cout << "QUERY complete" << std::endl;
|
||||
UNUSED(context);
|
||||
// Close the subscriptions for any topics that haven't been received.
|
||||
SubscriptionMap::iterator iter = m_subscriptionMap.begin();
|
||||
while (m_subscriptionMap.end() != iter) {
|
||||
if (iter->second.m_queryVersion != m_queryVersion)
|
||||
m_subscriptionMap.erase(iter++);
|
||||
else
|
||||
++iter;
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnTopicQueryError(const StatusCode statusCode, const Context &context) {
|
||||
std::cout << "QUERY error " << StatusCodeToString(statusCode)
|
||||
<< std::endl;
|
||||
UNUSED(context);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
void OnTopicQueryFailure(
|
||||
const StatusCode statusCode, const Context &context) {
|
||||
std::cout << "QUERY failure " << StatusCodeToString(statusCode)
|
||||
<< std::endl;
|
||||
|
||||
// Shutdown session.
|
||||
Session::Shutdown(context.GetSession());
|
||||
UNUSED(context);
|
||||
}
|
||||
};
|
||||
} // namespace bts::md::cme
|
||||
@@ -0,0 +1,100 @@
|
||||
/**
|
||||
* @file source/ActivOnePlatform/Samples/SnapshotFieldListValidatorSample/SnapshotHandler.h
|
||||
*
|
||||
* @brief SnapshotHandler definition.
|
||||
*/
|
||||
|
||||
#if (!defined ACTIV_SAMPLE_SNAPSHOT_HANDLER_H)
|
||||
#define ACTIV_SAMPLE_SNAPSHOT_HANDLER_H
|
||||
|
||||
#include "ActivOneApi/OneApi.h"
|
||||
#include "ActivOneApi/MarketData/MarketData.h"
|
||||
|
||||
|
||||
#include <ostream>
|
||||
|
||||
using namespace Activ;
|
||||
using namespace Activ::OneApi;
|
||||
namespace bts::md::cme {
|
||||
//! SnapshotHandler.
|
||||
class SnapshotHandler : public ISnapshotHandler
|
||||
{
|
||||
public:
|
||||
virtual void OnSnapshot(const SnapshotMessage &snapshotMessage, const Context &context) ACTIV_OVERRIDE;
|
||||
virtual void OnSnapshotFailure(const SnapshotFailureMessage &snapshotFailureMessage, const Context &context) ACTIV_OVERRIDE;
|
||||
|
||||
private:
|
||||
void ProcessField(const FieldId fieldId, const String &fieldName, const String &symbol);
|
||||
|
||||
// FieldListValidator will parse SnapshotMessage, if possible reuse FieldListValidator objects.
|
||||
FieldListValidator m_fieldListValidator;
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
inline void SnapshotHandler::OnSnapshot(const SnapshotMessage &snapshotMessage, const Context &context)
|
||||
{
|
||||
std::cout << "SNAPSHOT received for " << snapshotMessage.m_symbol << std::endl;
|
||||
std::cout << "Symbol .................... " << snapshotMessage.m_symbol << std::endl;
|
||||
|
||||
// Initialize will clear any existing cache fields. SnapshotMessage contains a complete topic field list if no filters have
|
||||
// been requested.
|
||||
StatusCode statusCode = m_fieldListValidator.Initialize(snapshotMessage);
|
||||
|
||||
if (STATUS_CODE_SUCCESS != statusCode)
|
||||
{
|
||||
std::cout << "ERROR decoding field data for " << snapshotMessage.m_symbol << " - " << StatusCodeToString(statusCode) << std::endl;
|
||||
return;
|
||||
}
|
||||
for (const Activ::OneApi::FlvDetail::Field &field :
|
||||
m_fieldListValidator) {
|
||||
if(field.IsDefined())
|
||||
std::cout<<"Has snapshot Feild: "<<MarketData::GetFieldName(field.GetId())<<", Value: "<<field.GetValue<std::string>() <<std::endl;
|
||||
}
|
||||
|
||||
// Print out the current value for FID_TRADE. FID_TRADE is defined in the Activ::MarketData namespace.
|
||||
ProcessField(MarketData::FID_TRADE, "FID_TRADE", snapshotMessage.m_symbol);
|
||||
|
||||
std::cout << std::endl;
|
||||
|
||||
context.GetSession().CloseHandle(context.GetHandle());
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
inline void SnapshotHandler::OnSnapshotFailure(const SnapshotFailureMessage &snapshotFailureMessage, const Context &context)
|
||||
{
|
||||
std::cout << "SNAPSHOT FAILURE for " << snapshotFailureMessage.m_symbol << std::endl;
|
||||
std::cout << snapshotFailureMessage.ToString() << std::endl;
|
||||
|
||||
context.GetSession().CloseHandle(context.GetHandle());
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
inline void SnapshotHandler::ProcessField(const FieldId fieldId, const String &fieldName, const String &symbol)
|
||||
{
|
||||
const FieldListValidator::Field *pField;
|
||||
|
||||
// FieldListValidator::GetField() provides access to the internally stored cached field.
|
||||
StatusCode statusCode = m_fieldListValidator.GetField(fieldId, pField);
|
||||
if (STATUS_CODE_SUCCESS != statusCode)
|
||||
{
|
||||
std::cout << "ERROR fetching " << fieldName << " field for " << symbol << " - " << StatusCodeToString(statusCode) << std::endl;
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
if (!pField->IsDefined())
|
||||
{
|
||||
std::cout << "undefined" << std::endl;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Field contents are only valid if the Field is defined.
|
||||
std::cout << pField->GetValue<String>() << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
#endif // (!defined ACTIV_SAMPLE_SNAPSHOT_HANDLER_H)
|
||||
@@ -1,3 +1,5 @@
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
|
||||
#include "CCAPIDeribitOrderMux.h"
|
||||
#include "CCAPIOrderState.h"
|
||||
#include "CCAPISingleOrderManager.h"
|
||||
@@ -953,4 +955,6 @@ void CCAPIDeribitOrderMux::_send_buffered_orders(uint32_t sec_id) {
|
||||
"n_buffered_orders: {}",
|
||||
me->buffered_creates.size());
|
||||
}
|
||||
} // namespace bts::orderman
|
||||
} // namespace bts::orderman
|
||||
|
||||
#endif
|
||||
@@ -1,4 +1,6 @@
|
||||
#pragma once
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
|
||||
#include "ICCAPIOrderMux.h"
|
||||
#include "lib/md/SecurityOracle.h"
|
||||
#include "lib/orderman/CreateOrderMsg.h"
|
||||
@@ -72,4 +74,6 @@ class CCAPIDeribitOrderMux : public ICCAPIOrderMux {
|
||||
void _handle_generic_pvt_request(const nlohmann::json &json_obj);
|
||||
};
|
||||
|
||||
} // namespace bts::orderman
|
||||
} // namespace bts::orderman
|
||||
|
||||
#endif
|
||||
@@ -1,3 +1,5 @@
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
|
||||
#include "CCAPIExecutionGateway.h"
|
||||
#include "lib/api/service_helpers/AdminServiceUtils.h"
|
||||
#include "lib/api/services/AdminService.pb.h"
|
||||
@@ -492,4 +494,6 @@ SingleOrderMux *CCAPIExecutionGateway::_find_mux_from_orderkey(OrderKey key) {
|
||||
return _find_mux(it->second.routing_record);
|
||||
}
|
||||
|
||||
} // namespace bts::orderman
|
||||
} // namespace bts::orderman
|
||||
|
||||
#endif
|
||||
@@ -1,4 +1,5 @@
|
||||
#pragma once
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
|
||||
#include "IExecutionGateway.h"
|
||||
#include "lib/api/services/AdminService.pb.h"
|
||||
@@ -73,3 +74,5 @@ class CCAPIExecutionGateway : public IExecutionGateway {
|
||||
|
||||
} // namespace orderman
|
||||
} // namespace bts
|
||||
|
||||
#endif
|
||||
@@ -1,3 +1,4 @@
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
#include "EMExecutionGateway.h"
|
||||
#include "lib/api/service_helpers/AdminServiceUtils.h"
|
||||
#include "lib/api/services/AdminService.pb.h"
|
||||
@@ -705,4 +706,6 @@ const api::Security *EMExecutionGateway::_get_security(
|
||||
}
|
||||
|
||||
} // namespace orderman
|
||||
} // namespace bts
|
||||
} // namespace bts
|
||||
|
||||
#endif
|
||||
@@ -34,9 +34,11 @@ std::unique_ptr<IExecutionGateway> ExecutionGatewayFactory::create_gateway(
|
||||
return std::make_unique<ilink3::GLBXiLink3ExecutionGateway>(
|
||||
conn, pl, o);
|
||||
return std::make_unique<GLBXExecutionGateway>(conn, pl, o);
|
||||
} else if (conn.venue() == bts::api::VENUE_DERIBIT)
|
||||
// return std::make_unique<EMExecutionGateway>(o);
|
||||
} else if (conn.venue() == bts::api::VENUE_DERIBIT) {
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
return std::make_unique<CCAPIExecutionGateway>(conn.venue(), o);
|
||||
#endif
|
||||
}
|
||||
return std::make_unique<FakeExecutionGateway>(o);
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
#pragma once
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
|
||||
#include "lib/core/EnumMacros.h"
|
||||
#include "lib/core/FMT_IMPL.h"
|
||||
@@ -26,7 +25,6 @@ ENUM_CLASS_DEF(
|
||||
#undef FOR_EACH_GLBX_ILINK3_EXECUTION_GATEWAY_STATE
|
||||
|
||||
} // namespace bts::orderman::ilink3
|
||||
#endif
|
||||
|
||||
FMT_IMPL(
|
||||
bts::orderman::ilink3::GLBXiLink3ExecutionGatewayState, s, "{}", s.name());
|
||||
@@ -1,5 +1,4 @@
|
||||
#pragma once
|
||||
#ifdef SPARK_ENABLE_CCAPI
|
||||
|
||||
#include "lib/core/EnumMacros.h"
|
||||
#include "lib/core/FMT_IMPL.h"
|
||||
@@ -27,6 +26,5 @@ ENUM_CLASS_DEF(IL3State, FOR_EACH_IL3_STATE)
|
||||
#undef FOR_EACH_IL3_STATE
|
||||
|
||||
} // namespace bts::orderman::ilink3
|
||||
#endif
|
||||
|
||||
FMT_IMPL(bts::orderman::ilink3::IL3State, s, "{}", s.name());
|
||||
@@ -1,3 +1,4 @@
|
||||
#include "lib/md/cme/CMEActivSession.h"
|
||||
#include "lib/testing/catch_test_runner.h"
|
||||
#include "lib/utils/init_spark_lib.h"
|
||||
#include "server/sparkd/cmds/AdminReport.h"
|
||||
@@ -10,6 +11,9 @@
|
||||
#include "server/sparkd/cmds/crypto/CryptoCmd.h"
|
||||
#include <CLI11/CLI11.hpp>
|
||||
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
|
||||
using namespace bts;
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
@@ -30,6 +34,9 @@ int main(int argc, char **argv) {
|
||||
app.add_subcommand(CLI::App_p(new cmds::Run()));
|
||||
app.add_subcommand(CLI::App_p(new cmds::KeyCmd()));
|
||||
app.add_subcommand(CLI::App_p(new cmds::Web()));
|
||||
|
||||
|
||||
|
||||
app.add_subcommand(
|
||||
"test", "Run unit test(s)"); // so this shows up in help
|
||||
CLI11_PARSE(app, argc, argv);
|
||||
|
||||
Reference in new issue
Block a user