Compare commits

...
Author SHA1 Message Date
Mohammad Arham 420f6d7cc0 all exectuion work 2024-10-18 21:26:43 +00:00
Mohammad Arham 36b6168dec Taifex State working 2024-10-18 21:13:56 +00:00
John Colvin e4c87c267d Fix DateTimeConstant token regex for DataStreams
The previous regex was overly ambitious (and needlessly complex as a
result). The new regex is simpler, and doesn't impose bounds on month
and day ints, aside from requiring 1-2 digits.

Bugs (fixed):
  * The DataStreamExpressionInterpretter fails to parse certain date
    constants, such as "8/30/2024"
2024-10-15 11:46:58 -04:00
Mohammad Arham 86899a6ff0 FIX: TAifex state working 2024-10-14 17:19:03 +00:00
Adeel Khan 2324fd3ad2 CHANGE: Taifex State Mgmt 2024-10-14 20:59:34 +05:00
Mohammad Arham 8850bf4903 Remove: Assert in taifex code 2024-10-09 14:53:20 +00:00
Mohammad Arham 11112b41d1 Add: Missing config entry databento 2024-10-09 02:36:35 +00:00
Adeel Khan 2a98889b67 UPDATE: TaifexDataGateway 2024-10-09 02:18:28 +00:00
Adeel Khan 8060518a28 FIX: Removed Taifex Bugs 2024-10-08 21:29:44 +00:00
Mohammad Arham 68440d3df3 ADD: Config for one year security filter 2024-10-08 13:50:38 +00:00
Mohammad Arham 7d775afc4c Add: Taifex after hours ip change config 2024-10-08 13:49:36 +00:00
Mohammad Arham 50c7764ef7 Add: Databento security filter 2024-10-07 20:33:38 +00:00
Mohammad Arham fb8b2d28f3 Fix: DATABENTO live working 2024-10-07 20:05:38 +00:00
Mohammad Arham bcc2c4a514 CHANGE: Taifex Refactored 2024-10-07 16:26:12 +00:00
Mohammad Arham 9352298be0 Add: Taifex statehandler 2024-10-07 08:51:55 +00:00
Mohammad Arham 180bd11740 FIX: Taifex replay types of files handled 2024-10-06 21:31:34 +00:00
Mohammad Arham 57ea841381 ADD: i084 taifex 2024-10-06 20:31:07 +00:00
Mohammad Arham 7d9d176fa2 Change: Rename taifex files 2024-10-06 18:38:36 +00:00
Mohammad Arham a8f4ec07c8 Change: moved file paths to config for taifex file replay 2024-10-06 08:29:27 +00:00
Mohammad Arham 58c5be268f Add: Refactored taifex code to break it down 2024-10-06 08:01:22 +00:00
Mohammad Arham bf4859a2a9 Change: Refactored read binary file taifex 2024-10-06 07:28:12 +00:00
Adeel Khan db1cf93994 Add: Incremental Books 2024-10-04 23:23:54 +05:00
Adeel Khan ecb1c4fea0 Add: Taifex Trades 2024-10-04 23:23:41 +05:00
Mohammad Arham a513933044 Add: Exceptions handling for databento 2024-10-04 13:06:56 +00:00
Mohammad Arham 764d65b595 Remove: code to remove secuties 2024-10-03 12:06:55 +00:00
Mohammad Arham bc0d606252 Add: code to connect to Live databento 2024-10-03 12:03:48 +00:00
Mohammad Arham ceb8a68902 Add: databento repllay filename to cofig 2024-09-27 03:20:20 +00:00
Mohammad Arham 0b3183fd30 Add: added onix fix to lib to build dir 2024-09-26 16:23:19 +00:00
Mohammad Arham b6dbe37818 Add: entry for onix fix 2024-09-24 13:41:45 +00:00
Mohammad Arham c78b8bc8e1 Add databento to external 2024-09-24 13:37:57 +00:00
Taha Afzal 314f3e3715 Adding comments 2024-09-23 18:40:06 +00:00
Taha Afzal 26a1d5fc03 Creating scaffolding for incremental book updates in GeneralMarketDataListener 2024-09-23 18:39:12 +00:00
51 changed files with 4243 additions and 1591 deletions

No files matched your search

+12
View File
@@ -26,6 +26,18 @@
"stopAtEntry": false,
"console": "integratedTerminal"
},
{
"name": "(gdb) Launch",
"type": "cppdbg",
"request": "launch",
"program": "/home/bts/spark/build/sparkd-debug-clang/server/sparkd",
"args": ["run"],
"stopAtEntry": false,
"cwd": "/home/bts/spark/build/sparkd-debug-clang/server",
"environment": [],
"externalConsole": false,
"MIMode": "gdb",
},
{
"name": "DataStreamQueryAnalyzer",
"type": "coreclr",
+2
View File
@@ -132,6 +132,8 @@ else()
COMMAND ln ARGS -sf ${CMAKE_BINARY_DIR}/src/server/sparkd/sparkd ${CMAKE_BINARY_DIR}/server
COMMAND ln ARGS -sf ${CMAKE_BINARY_DIR}/data_files ${CMAKE_BINARY_DIR}/server
COMMAND ln ARGS -sf ${CMAKE_BINARY_DIR}/src/lib/libspark.so ${CMAKE_BINARY_DIR}/server
COMMAND ln ARGS -sf ${EXTERNALS_ROOT}/lib/${CMAKE_BUILD_TYPE}/libonixs-fix-engine.so.4 ${CMAKE_BINARY_DIR}/server
COMMAND
ln ARGS -sf
${EXTERNALS_ROOT}/lib/Release/libOnixS.IceImpactMulticastPriceFeedHandlerCpp.so.3.16.0.0
@@ -9,7 +9,15 @@
Letter [a-zA-Z]
BooleanConstant 'true'|'false'
CSStringConstant \"([^\"\\]*|\\.)*\"
DateTimeConstant ([1-9]|'10'|'11'|'12')\/([1-9]|([1-2][0-9])|'30'|'31')\/[0-9][0-9][0-9]*
YearInt [0-9]{4}
MonthInt [0-9]{1,2}
DayInt [0-9]{1,2}
/**
* DateTimeConstant supports American (mm/dd/yyyy) and European (yyyy/mm/dd)
* formats. Month and day can be 1 or 2 digits. It does not validate that the
* numbers represent an actual date.
*/
DateTimeConstant ({MonthInt}\/{DayInt}\/{YearInt}|{YearInt}\/{MonthInt}\/{DayInt})
IntegerConstant -?[0-9]+
NullConstant 'null'
RealConstant -?[0-9]+\.[0-9]+
+3 -11
View File
@@ -1,6 +1,7 @@
syntax = "proto3";
import "lib/api/domain/SecurityCriteria.proto";
import "lib/api/md/DatabentoMarketDataConfig.proto";
package bts.api.md;
@@ -9,16 +10,7 @@ option csharp_namespace = "bts.api.md";
message CboeMarketDataConfig {
uint32 unique_check_value = 1;
bool use_databento = 10;
bool read_from_file = 20;
//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;
bool use_databento = 10;
DatabentoMarketDataConfig databento_settings = 20;
}
@@ -0,0 +1,17 @@
syntax = "proto3";
package bts.api.md;
option csharp_namespace = "bts.api.md";
option cc_enable_arenas = true;
message DatabentoMarketDataConfig {
uint32 unique_check_value = 1;
bool read_from_file = 20;
string contracts_file_name = 30;
string bbo_file_name = 40;
double replay_sleep_timer_sec = 50;
string key = 60;
string dataset = 70;
bool filter_securities_one_year= 80;
}
+3 -3
View File
@@ -1,7 +1,7 @@
syntax = "proto3";
import "lib/api/domain/SecurityCriteria.proto";
import "lib/api/md/DatabentoMarketDataConfig.proto";
package bts.api.md;
option csharp_namespace = "bts.api.md";
@@ -9,9 +9,9 @@ option csharp_namespace = "bts.api.md";
message GlbxMarketDataConfig {
uint32 unique_check_value = 1;
bool use_databento = 10;
bool read_from_file = 20;
bts.api.domain.SecurityCriteria security_criteria = 3;
bool use_databento = 10;
DatabentoMarketDataConfig databento_settings =20;
//uint32 num_depth_levels = 1;
//uint32 depth_throttle_ms = 2;
// Market data isn't subscribed for securities listed here (even if they are
+15 -10
View File
@@ -5,19 +5,24 @@ import "lib/api/domain/SecurityCriteria.proto";
package bts.api.md;
option csharp_namespace = "bts.api.md";
message TaifexConnectionDetails{
string live_data_ip = 30;
string reference_data_ip = 31;
string refresh_data_ip = 32;
string live_data_port = 40;
string reference_data_port = 41;
string refresh_data_port = 42;
string interface_ip = 50;
}
message TaifexMarketDataConfig {
uint32 unique_check_value = 1;
bool read_from_file = 20;
//uint32 num_depth_levels = 1;
//uint32 depth_throttle_ms = 2;
string ref_data_file_name = 21;
string realtime_data_file_name = 22;
TaifexConnectionDetails regular_hours_connection_details = 30;
TaifexConnectionDetails after_hours_connection_details = 31;
bool use_after_hours = 40;
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;
}
@@ -156,6 +156,8 @@ AdminServiceUtils::resolve_connection(
const bts::api::admin::ExchangeConnection *rv = nullptr;
switch (sec.exchange()) {
case api::Venue::VENUE_GLBX: {
//CME EXECUTION NEEDS TO BE UPDATED, probably merge with VENUE_BINANCE_USDS_FUTURES
auto deployment_type =
boost::to_lower_copy(config::get<config::DeploymentType>().type());
rv = resolve_cme_connection(data, account, sec);
@@ -227,6 +229,7 @@ void AdminServiceUtils::fill_connections_from_config(
connection.set_name(c.name());
switch (c.venue()) {
//CME EXECUTION NEEDS TO BE UPDATED, probably merge default case
case bts::api::VENUE_GLBX:
if (c.cme_market_segment_id() == 0)
connection.mutable_cme_connection()->mutable_cgw_connection();
@@ -236,9 +239,7 @@ void AdminServiceUtils::fill_connections_from_config(
->set_market_segment_id(c.cme_market_segment_id());
add_connection(admin_data, connection);
break;
case bts::api::VENUE_CBOE:
break;
case bts::api::VENUE_TAIFEX:
case bts::api::VENUE_TAIFEX:
break;
case bts::api::VENUE_SIM:
case bts::api::VENUE_SIMICE:
@@ -250,6 +251,7 @@ void AdminServiceUtils::fill_connections_from_config(
c.sender_compid());
add_connection(admin_data, connection);
break;
case bts::api::VENUE_CBOE:
default:
connection.mutable_general_connection()->set_exchange(c.venue());
add_connection(admin_data, connection);
@@ -571,6 +573,13 @@ bts::api::orderman::RoutingRecord AdminServiceUtils::get_routing_record(
return rv;
}
switch (sec.exchange()) {
case api::Venue::VENUE_CBOE:{
rv.set_session("cboe_1");
rv.set_account(account_ptr->name());
rv.set_operator_id(user_ptr->name());
out_result.set_is_success(true);
return rv;
}
case api::Venue::VENUE_DERIBIT: {
rv.set_session("deribit_1");
rv.set_account(account_ptr->name());
@@ -741,7 +750,6 @@ bts::api::orderman::RoutingRecord AdminServiceUtils::get_routing_record(
case api::VENUE_OKX:
case api::Venue::VENUE_UNSET:
case api::Venue::VENUE_BTS:
case api::Venue::VENUE_CBOE:
case api::Venue::VENUE_TAIFEX:
case api::Venue::VENUE_CBOECFE:
case api::Venue::Venue_INT_MIN_SENTINEL_DO_NOT_USE_:
+2
View File
@@ -409,6 +409,7 @@ message RoutingRecord {
message NativeOrderState {
string exchange_order_id = 1;
string external_order_id = 11; //"string : orderkey"
uint64 last_msg_seq_num = 4;
enum SubError {
SUB_ERROR_NONE = 0;
@@ -752,6 +753,7 @@ message RetrievedExchangeOpenOrder{
string exchange_sec_desc = 10;
string exchange_account = 20;
string exchange_order_id = 30;
string external_order_id = 11; //"string : orderkey"
uint32 security_key = 40;
bool security_resolved = 50;
string account_name = 60;
@@ -25,6 +25,7 @@ function(get_link_libraries dest)
Packet++
Common++
ssh2
onixs-fix-engine
-lrt
-ldl
-lpcap
+13 -1
View File
@@ -4028,5 +4028,17 @@ JetFIX {
name: "deribit_1"
venue: VENUE_DERIBIT
}
# CME EXECUTION NEEDS TO BE UPDATED, probably below
# connection {
# name: "onixs_sim"
# venue: VENUE_GLBX
# }
}
}
}
# NotSimExecution {
# value {
# entries {
# venue: VENUE_CBOE
# }
# }
# }
@@ -0,0 +1,12 @@
OnixFixCfg {
value{
name: "trafix",
msg_type: "Order_Single",
protocol_version: "FIX_42",
handl_inst: 2,
sender_sub_id: "BTS",
target_sub_id: "TRFXMRX",
client_id: "BTS",
rule80a: 1
}
}
+3 -3
View File
@@ -105,10 +105,10 @@ message CMECannedSecureLogon {
}
message SimExecution{
message SimExecutionEntry {
message NotSimExecution{
message NotSimExecutionEntry {
bts.api.Venue venue = 100;
bool incorporate_crosses_into_book = 200;
}
repeated SimExecutionEntry entries = 100;
repeated NotSimExecutionEntry entries = 100;
}
+23
View File
@@ -0,0 +1,23 @@
syntax = "proto3";
package config;
message OnixFixCfg {
enum HandlInst {
UNDEFINED = 0;
DMA = 1;
ALGO = 2;
CARE = 3;
}
enum Rule80A {
UNUSED = 0;
AGENCY = 1;
PRINCIPAL = 2;
}
string name = 9;
string msg_type = 1;
string protocol_version = 2;
HandlInst handl_inst = 3;
string sender_sub_id = 4;
string target_sub_id = 5;
string client_id = 6;
Rule80A rule80a = 7;
}
+148 -25
View File
@@ -21,6 +21,7 @@
#include "lib/domain/TradeRecord.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/logging/logging.h"
#include "lib/md/IMarketDataListener.h"
#include "lib/md/IntPrice.h"
#include "lib/md/MDProducer.h"
#include "lib/md/MDProducerHandle.h"
@@ -73,6 +74,103 @@ struct CCSecSearchResult {
bool exists_now;
};
struct IncrementalBookData {
std::map<domain::SecurityID, domain::Book> book_map;
bool update_incremental(const IMarketDataListener::BookOp & op) {
// Retrieve the Book from the map and update it
domain::Book* book = get_book(op.id);
if (book == nullptr)
return false;
// Choosing the correct side of the book (bids or asks)
auto& book_side = (op.side == domain::Side::Buy) ? book->bids() : book->asks();
std::string side_str = (op.side == domain::Side::Buy) ? "Buy" : "Sell";
// Capturing the book state before the update
std::stringstream book_before;
book_before << "Book before update (" << side_str << " side):\n";
for (const auto& level : book_side) {
book_before << "Price: " << level.price << ", Quantity: " << level.direct_quantity << "\n";
}
size_t insert_level = 0;
bool level_found = false;
// Logic to find where to insert or update the level
if (op.side == domain::Side::Buy) {
while (insert_level < book_side.size()) {
if (op.price == book_side[insert_level].price) {
level_found = true;
break;
}
if (op.price > book_side[insert_level].price) {
break;
}
insert_level++;
}
} else {
while (insert_level < book_side.size()) {
if (op.price == book_side[insert_level].price) {
level_found = true;
break;
}
if (op.price < book_side[insert_level].price) {
break;
}
insert_level++;
}
}
// Insert or update the book level
if (!level_found) {
book_side.insert(book_side.begin() + insert_level, {op.price, op.quantity});
} else {
book_side[insert_level].direct_quantity = op.quantity;
}
// Capturing the book state after the update
std::stringstream book_after;
book_after << "Book after update (" << side_str << " side):\n";
for (const auto& level : book_side) {
book_after << "Price: " << level.price << ", Quantity: " << level.direct_quantity << "\n";
}
// Capturing the operation details
std::stringstream op_details;
op_details << "Operation: Price: " << op.price << ", Quantity: " << op.quantity
<< ", Side: " << side_str << "\n";
// Combine the before/after and operation details into one string
std::string book_before_after_op = book_before.str() + "\n" + book_after.str() + "\n" + op_details.str();
// Assertion to check the book is sorted
bts_assert_msgf(book->is_sorted(), "book not sorted {}", book_before_after_op);
return true;
}
domain::Book const *get_const_book(domain::SecurityID id) const {
auto it = book_map.find(id);
if (it == book_map.end()) {
return nullptr;
}
return &(it->second);
}
domain::Book *get_book(domain::SecurityID id) {
auto it = book_map.find(id);
if (it == book_map.end()) {
return nullptr;
}
return &(it->second);
}
void overwrite_book(const domain::Book &book) {
domain::SecurityID id = book.security_id();
book_map[id] = book;
}
};
struct GeneralMarketDataListener::PImpl {
// if this is set to true, then the security key passed on in the
// on_securities function is used. If this is not set to true, then a new key
@@ -94,6 +192,7 @@ struct GeneralMarketDataListener::PImpl {
std::unordered_map<bts::domain::SecurityID, uint64_t> cc_sec_id_map = {};
MDArray bids = {};
MDArray asks = {};
IncrementalBookData inc_book_data = {};
explicit PImpl(
GeneralSecurityIndex &index,
MDProducer &md_producer,
@@ -234,7 +333,8 @@ void GeneralMarketDataListener::on_security_statistics(
auto handle = _find_handle(sec_stats.security_id);
bts_assert_msgf(
handle != nullptr,
"GeneralMarketDataListener::on_security_statistics null handle. id: {}",
"GeneralMarketDataListener::on_security_statistics null handle. id: "
"{}",
sec_stats.security_id);
auto now = utils::TimeStamp::now();
bts::datetime::Date date_temp = {};
@@ -246,31 +346,23 @@ void GeneralMarketDataListener::on_security_statistics(
handle->handle.set_total_volume(sec_stats.volume.value());
}
void GeneralMarketDataListener::on_book(const bts::domain::Book &book, std::optional<utils::TimeStamp> time) {
void GeneralMarketDataListener::on_book(
const bts::domain::Book &book, std::optional<utils::TimeStamp> time) {
LOG(md, Diag, "GeneralMarketDataListener::on_book sec_id: {}",
book.security_id());
if(!time.has_value()){
time = utils::TimeStamp::now();
}
me->n_bk_updates++;
auto handle = _find_handle(book.security_id());
if (handle == nullptr) {
LOG(md, Error,
"GeneralMarketDataListener::on_book, could not find sec_id: {}",
book.security_id());
// in case this is called externally in conjunction iwth inc book udpates
me->inc_book_data.overwrite_book(book);
_on_book(book, time);
}
void GeneralMarketDataListener::on_book_increment(
const IMarketDataListener::BookOp &book_op) {
if(!me->inc_book_data.update_incremental(book_op)){
return;
}
copy_books(book, me->bids, bts::domain::Side::Value::Buy);
copy_books(book, me->asks, bts::domain::Side::Value::Sell);
_possibly_match_with_sim_orders(
bts::md::SecurityKey(book.security_id().value()));
LOG(md, Diag,
"GeneralMarketDataListener::on_book post id: {} bid: {}/{}/{} ask: "
"{}/{}/{}",
handle->id, me->bids.at(0).price(), me->bids.at(0).direct_size(),
me->bids.at(0).size(), me->asks.at(0).price(),
me->asks.at(0).direct_size(), me->asks.at(0).size());
handle->handle.update_direct(time.value(), me->bids, me->asks);
domain::Book const *book_ptr = me->inc_book_data.get_const_book(book_op.id);
bts_assert(book_ptr != nullptr);
_on_book(*book_ptr);
}
void GeneralMarketDataListener::on_trades(
@@ -351,7 +443,8 @@ void GeneralMarketDataListener::_add_future(const bts::domain::Security &sec) {
const auto &f_sec = std::get<bts::domain::FutureSecurity>(sec.security);
get_date_from_epoch_nanos(f_sec.expiration.nanos(), year, month, day);
LOG(md, Info,
"GeneralMarketDataListener::_add_future nanos: {} yy: {} mm: {} dd: {}",
"GeneralMarketDataListener::_add_future nanos: {} yy: {} mm: {} dd: "
"{}",
f_sec.expiration.nanos(), year, month, day);
ff_sec.expiration_year = year;
ff_sec.expiration_month = month;
@@ -394,7 +487,8 @@ void GeneralMarketDataListener::_add_spread(const bts::domain::Security &sec) {
auto it = me->cc_sec_id_map.find(l.security_id);
if (it == me->cc_sec_id_map.end()) {
LOG(md, Error,
"GeneralMarketDataListener::_add_spread, could not resolve leg. "
"GeneralMarketDataListener::_add_spread, could not resolve "
"leg. "
"id: {}; leg_id: {}",
sec.id(), l.security_id);
return;
@@ -431,7 +525,8 @@ void GeneralMarketDataListener::_commit_security(const FixedSecurity &sec) {
// Add it to the index so it can receive messages
auto kp = me->index.add_security(sec.numeric_id, handle);
LOG(md, Info,
"GeneralMarketDataListener::_add_security_to_shm added sec: {} id: {} "
"GeneralMarketDataListener::_add_security_to_shm added sec: {} id: "
"{} "
"num_id: {} n_secs: {}",
std::string(handle.security().security_desc()), kp->id,
handle.security().numeric_id, me->index.n_secs());
@@ -468,5 +563,33 @@ void GeneralMarketDataListener::_possibly_match_with_sim_orders(
}
}
void GeneralMarketDataListener::_on_book(
const domain::Book &book, std::optional<utils::TimeStamp> time) {
LOG(md, Diag, "GeneralMarketDataListener::_on_book sec_id: {}",
book.security_id());
if (!time.has_value()) {
time = utils::TimeStamp::now();
}
me->n_bk_updates++;
auto handle = _find_handle(book.security_id());
if (handle == nullptr) {
LOG(md, Error,
"GeneralMarketDataListener::_on_book, could not find sec_id: {}",
book.security_id());
return;
}
copy_books(book, me->bids, bts::domain::Side::Value::Buy);
copy_books(book, me->asks, bts::domain::Side::Value::Sell);
_possibly_match_with_sim_orders(
bts::md::SecurityKey(book.security_id().value()));
LOG(md, Diag,
"GeneralMarketDataListener::_on_book post id: {} bid: {}/{}/{} ask: "
"{}/{}/{}",
handle->id, me->bids.at(0).price(), me->bids.at(0).direct_size(),
me->bids.at(0).size(), me->asks.at(0).price(),
me->asks.at(0).direct_size(), me->asks.at(0).size());
handle->handle.update_direct(time.value(), me->bids, me->asks);
}
} // namespace bts::md
//#endif
+7 -1
View File
@@ -23,7 +23,10 @@ class GeneralMarketDataListener : public IMarketDataListener {
void on_security_state(const domain::SecurityState &security_state) override;
void on_security_statistics(
const domain::SecurityMarketDataStatistics &security_stats) override;
void on_book(const domain::Book &book, std::optional<utils::TimeStamp> time = std::nullopt) override;
void on_book(
const domain::Book &book,
std::optional<utils::TimeStamp> time = std::nullopt) override;
void on_book_increment(const BookOp &) override;
void on_trades(
const std::vector<domain::TradeRecord const *> &trades) override;
void on_rfqs(const std::vector<domain::RFQ const *> &rfqs) override;
@@ -49,5 +52,8 @@ class GeneralMarketDataListener : public IMarketDataListener {
FixedSecurity &to) const;
uint64_t _generate_security_key(const bts::domain::Security &sec);
void _possibly_match_with_sim_orders(bts::md::SecurityKey key);
void _on_book(
const domain::Book &book,
std::optional<utils::TimeStamp> time = std::nullopt);
};
} // namespace bts::md
+11 -5
View File
@@ -1,6 +1,7 @@
#pragma once
#include "lib/api/services/OrderService.pb.h"
#include "lib/domain/Book.h"
#include "lib/domain/SecurityID.h"
#include "lib/md/PriceSize.h"
#include <vector>
@@ -19,12 +20,15 @@ class MarketDataError;
// The is the interface that registers itself to IMarketDataGateway and receives
// all market data via callbacks. These callbacks are always on the same thread.
class IMarketDataListener {
enum BookOpType { Add = 0, REMOVE, MODIFY };
public:
enum BookOpType { Add = 0, MODIFY, REMOVE };
struct BookOp {
bts::api::Side side;
domain::SecurityID id;
domain::Side side;
BookOpType op_type;
// for op == REMOVE size in price_size is no-op
PriceSize price_size;
domain::Price price;
domain::Quantity quantity;
};
public:
@@ -38,8 +42,10 @@ class IMarketDataListener {
const domain::SecurityState &security_state) = 0;
virtual void on_security_statistics(
const domain::SecurityMarketDataStatistics &security_stats) = 0;
virtual void on_book(const domain::Book &book, std::optional<utils::TimeStamp> time = std::nullopt) = 0;
virtual void on_book_increment(const BookOp &) {}
virtual void on_book(
const domain::Book &book,
std::optional<utils::TimeStamp> time = std::nullopt) = 0;
virtual void on_book_increment(const IMarketDataListener::BookOp &) {}
virtual void on_trades(
const std::vector<domain::TradeRecord const *> &trades) = 0;
virtual void on_rfqs(const std::vector<domain::RFQ const *> &rfqs) = 0;
@@ -60,7 +60,7 @@ class OrderStateUpdateEvent : public mercury::Event<OrderStateUpdateEvent> {
} else {
auto native_state = get_native_state(state);
if (native_state) {
exchange_order_id = native_state->exchange_order_id();
exchange_order_id = native_state->exchange_order_id(); //use this
exchange_order_id_u64 = native_state->exchange_order_id_u64();
last_msg_seq_num = native_state->last_msg_seq_num();
sub_error_code = native_state->sub_error_code();
@@ -4,9 +4,12 @@
#include "lib/orderman/gateways/CCAPIExecutionGateway.h"
// #include "lib/orderman/gateways/EMExecutionGateway.h"
#include "lib/orderman/gateways/FakeExecutionGateway.h"
#include "lib/orderman/gateways/GeneralExecutionGateway.h"
#include "lib/orderman/gateways/GLBXExecutionGateway.h"
#include "lib/orderman/gateways/GLBXiLink3ExecutionGateway.h"
#include "lib/orderman/gateways/ICEExecutionGateway.h"
#include "lib/orderman/gateways/IFIXGenericAdapter.h"
#include "lib/orderman/gateways/OnixFIXAdapter.h"
#include "lib/orderman/gateways/sim/SimExecutionGatewayFactory.h"
#include "lib/orderman/gateways/sim/SimExecutionGatewayUniqueWrapper.h"
#include "lib/tracing/PerfLog.h"
@@ -21,23 +24,30 @@ std::unique_ptr<IExecutionGateway> ExecutionGatewayFactory::create_gateway(
const config::JetFIX::Connection &conn,
trace::PerfLog &pl,
md::SecurityOracle o) {
LOG(OrderMan, Info, "ExecutionGatewayFactory::create_gateway conn: {}",
utils::pb::to_string(conn));
if (SimExecutionGatewayFactory::is_sim_execution_on(conn.venue())) {
return std::make_unique<SimExecutionGatewayUniqueWrapper>(conn.venue());
}
LOG(OrderMan, Info, "ExecutionGatewayFactory::create_gateway conn: {}",
utils::pb::to_string(conn));
if (conn.venue() == bts::api::VENUE_ICE ||
conn.venue() == bts::api::VENUE_SIMICE)
return std::make_unique<ICEExecutionGateway>(conn, pl, o);
else if (conn.venue() == bts::api::VENUE_GLBX) {
if (conn.use_ilink3())
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);
return std::make_unique<CCAPIExecutionGateway>(conn.venue(), o);
return std::make_unique<FakeExecutionGateway>(o);
switch(conn.venue()){
case bts::api::VENUE_ICE:
case bts::api::VENUE_SIMICE:
return std::make_unique<ICEExecutionGateway>(conn, pl, o);
case bts::api::VENUE_GLBX:{
if (conn.use_ilink3())
return std::make_unique<ilink3::GLBXiLink3ExecutionGateway>( conn, pl, o);
return std::make_unique<GLBXExecutionGateway>(conn, pl, o);
}
case bts::api::VENUE_DERIBIT:
return std::make_unique<CCAPIExecutionGateway>(conn.venue(), o);
case bts::api::VENUE_CBOE:
{
std::unique_ptr<IFIXGenericAdapter> adapter = std::make_unique<OnixFIXAdapter>();
return std::make_unique<GeneralExecutionGateway>(conn.venue(),std::move(adapter),o);
}
default:
return std::make_unique<FakeExecutionGateway>(o);
}
}
} // namespace bts::orderman
@@ -0,0 +1,85 @@
#include "GeneralExecutionGateway.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/logging/logging.h"
#include "lib/orderman/CreateOrderMsg.h"
#include "lib/orderman/OrderMsg.h"
#include "lib/orderman/gateways/IExecutionGateway.h"
#include "lib/orderman/orderman_utils/CancelReject.h"
#include <memory>
#include <type_traits>
namespace bts::orderman {
struct GeneralExecutionGateway::PImpl {
PImpl(bts::api::Venue venue,
std::unique_ptr<IFIXGenericAdapter> adapter,
md::SecurityOracle o)
: venue(venue), _adapter(std::move(adapter)), sec_oracle(o) {}
bts::api::Venue venue;
std::unique_ptr<IFIXGenericAdapter> _adapter;
md::SecurityOracle sec_oracle;
OnExeReport on_exe_report;
OnGwStatusChange on_gw_status_change;
OnCancelReject on_cancel_reject;
OnExchangePositionState on_exchange_position_state;
OnExchangeOpenOrdersState on_exchange_open_orders;
};
OnGwStatusChange &GeneralExecutionGateway::on_gw_status_change() {
return me->on_gw_status_change;
}
// Modified constructor to accept the adapter from outside
GeneralExecutionGateway::GeneralExecutionGateway(bts::api::Venue venue,
std::unique_ptr<IFIXGenericAdapter> adapter, md::SecurityOracle o)
: IExecutionGateway(), me(new PImpl(venue, std::move(adapter),o)) {
LOG(OrderMan, Info, "ExecutionGateway::ExecutionGateway constructor");
}
GeneralExecutionGateway::~GeneralExecutionGateway() {}
void GeneralExecutionGateway::start() {
me->on_gw_status_change(true);
me->_adapter->start();
me->_adapter->register_callback(me->on_exe_report);
LOG(OrderMan,Info,"ExecutionGateway::start started" );
}
void GeneralExecutionGateway::send_msgs(
std::vector<OrderMsg> const &msgs, trace::CorID) {
LOG(OrderMan,Info,"ExecutionGateway::send_msgs" );
for (const auto &msg : msgs) {
std::visit(
[this](auto &&order) {
using T = std::decay_t<decltype(order)>;
if constexpr (std::is_same_v<T, CreateOrderMsg>) {
LOG(OrderMan,Warn,"ExecutionGateway::CreateOrderMsg" );
me->_adapter->launch_new_order(order);
} else if constexpr (std::is_same_v<T, CancelOrderMsg>) {
LOG(OrderMan,Warn,"ExecutionGateway::CancelOrderMsg" );
me->_adapter->cancel_order(order);
} else if constexpr (std::is_same_v<T, CancelReplaceOrderMsg>) {
LOG(OrderMan,Warn,"ExecutionGateway::CancelReplaceOrderMsg" );
me->_adapter->modify_order(order);
} else if constexpr (std::is_same_v<T, RFQMsg>) {
// Handle RFQMsg if necessary
// adapter->handle_rfq(order); // Example placeholder
}
},
msg);
}
}
void GeneralExecutionGateway::dispatch_incoming_fix_messages() {}
void GeneralExecutionGateway::on_securities(
const mercury::Update<NewSecuritiesEvent> &) {}
void GeneralExecutionGateway::on_admin_data(
const mercury::Update<NewAdminDataEvent> &) {}
void GeneralExecutionGateway::send_fix(const jetfix::OutFIXPtr &, trace::CorID) {}
OnExeReport &GeneralExecutionGateway::on_exe_report() { return me->on_exe_report; }
OnCancelReject &GeneralExecutionGateway::on_cancel_reject() {
static OnCancelReject default_cancel_reject;
return default_cancel_reject;
}
OnExchangePositionState &GeneralExecutionGateway::on_exchange_position_state() {
static OnExchangePositionState default_exchange_position_state;
return default_exchange_position_state;
}
OnExchangeOpenOrdersState &GeneralExecutionGateway::on_exchange_open_orders() {
static OnExchangeOpenOrdersState default_exchange_open_orders;
return default_exchange_open_orders;
}
} //namespace bts::orderman
@@ -0,0 +1,29 @@
#pragma once
#include "lib/orderman/gateways/IExecutionGateway.h"
#include "lib/orderman/gateways/IFIXGenericAdapter.h"
namespace bts {
namespace orderman {
class GeneralExecutionGateway : public IExecutionGateway {
public:
GeneralExecutionGateway(bts::api::Venue venue,std::unique_ptr<IFIXGenericAdapter> adapter,
md::SecurityOracle o);
virtual ~GeneralExecutionGateway();
void start() override;
void send_msgs(std::vector<OrderMsg> const & msgs, trace::CorID) override;
void dispatch_incoming_fix_messages() override;
void on_securities(const mercury::Update<NewSecuritiesEvent> &) override;
void on_admin_data(const mercury::Update<NewAdminDataEvent> &) override;
void send_fix(const jetfix::OutFIXPtr &, trace::CorID) override;
OnExeReport &on_exe_report() override;
OnGwStatusChange &on_gw_status_change() override;
OnCancelReject &on_cancel_reject() override;
OnExchangePositionState &on_exchange_position_state() override;
OnExchangeOpenOrdersState &on_exchange_open_orders() override;
private:
struct PImpl;
std::unique_ptr<PImpl> me;
};
} //namespace bts
} //namespace orderman
@@ -1,3 +1,6 @@
#include "IFIXGenericAdapter.h"
namespace bts::orderman {}
namespace bts::orderman {
IFIXGenericAdapter::~IFIXGenericAdapter(){}
}
+19 -19
View File
@@ -15,8 +15,8 @@
#include "lib/utils/bts_no_copy.h"
#include <functional>
#include <memory>
#include <variant>
namespace bts::orderman {
@@ -29,33 +29,33 @@ class NewAdminDataEvent;
class ExecutionReport;
class CancelReject;
struct OrderStateWithFills {
api::orderman::OrderState ordr_state;
std::vector<api::orderman::Fill> fills = {};
};
// struct OrderStateWithFills {
// api::orderman::OrderState ordr_state;
// std::vector<api::orderman::Fill> fills = {};
// };
using OnOrderUpdate = std::function<void(OrderStateWithFills const &)>;
using OnExecutionReport = std::function<void(ExecutionReport const &, const trace::CorID &)>;
// fix execution gateway for a single FIX session
class IFIXGenericAdapter {
public:
IFIXGenericAdapter() = default;
virtual ~IFIXGenericAdapter() {}
virtual ~IFIXGenericAdapter();
virtual void start() = 0;
// this callback should be added in a list, and on receving an update from the other side, it should be passed here
virtual void register_callback(const OnOrderUpdate &);
virtual void launch_new_order(const CreateOrderMsg &);
virtual void modify_order(const CancelReplaceOrderMsg &);
virtual void cancel_order(const CancelOrderMsg &);
// // this callback should be added in a list, and on receving an update from the other side, it should be passed here
virtual void register_callback(const OnExecutionReport &) =0 ;
virtual void launch_new_order(const CreateOrderMsg &) =0;
virtual void modify_order(const CancelReplaceOrderMsg &) = 0;
virtual void cancel_order(const CancelOrderMsg &) =0;
// as new securities are added
virtual void update_securities(
const mercury::Update<NewSecuritiesEvent> &) = 0;
virtual void update_admin_data(
const mercury::Update<NewAdminDataEvent> &) = 0;
// virtual void update_securities(
// const mercury::Update<NewSecuritiesEvent> &) = 0;
// virtual void update_admin_data(
// const mercury::Update<NewAdminDataEvent> &) = 0;
private:
BTS_NO_COPY(IFIXGenericAdapter);
// private:
// BTS_NO_COPY(IFIXGenericAdapter);
};
} // namespace bts::orderman
@@ -0,0 +1,406 @@
#include "OnixFIXAdapter.h"
#include "OnixS/FIXEngine/FIX/Engine.h"
#include "lib/config/config.h"
#include "lib/config/protos/system_config.pb.h"
#include "lib/logging/logging.h"
#include "lib/orderman/CancelReplaceOrderMsg.h"
#include "lib/orderman/CreateOrderMsg.h"
#include "lib/orderman/gateways/OnixFIXListener.h"
#include "lib/orderman/gateways/TrafixOrderStateHandler.h"
#include "lib/utils/bts_assert.h"
#include <OnixS/FIXEngine/FIX/Exception.h>
#include <OnixS/FIXEngine/FIX/FIX42.h>
#include <OnixS/FIXEngine/FIX/Message.h>
#include <OnixS/FIXEngine/FIX/ProtocolVersion.h>
#include <cassert>
#include <memory>
#include <quickfix/FixFields.h>
#include <string>
namespace bts::orderman {
OnixFIXAdapter::OnixFIXAdapter()
: state_handler(std::make_shared<TrafixOrderStateHandler>()), listener(state_handler), // Initialize listener
session(nullptr) // Initialize session pointer to nullptr
{
try {
// Initialization phase.
LOG(OrderMan, Warn, "OnixFIXAdapter::OnixFIXAdapter constructor");
OnixS::FIX::Engine::init();
session = std::make_unique<Session>(
"BTS", "TRFXMRX", OnixS::FIX::ProtocolVersion::FIX_42, &listener);
// Session-related logic goes here.
} catch (const OnixS::FIX::Exception &ex) {
std::string errorMessage = ex.what();
LOG(OrderMan, Error, "Exception: {}", errorMessage);
Engine::shutdown();
bts_assert_msg(false, "OnixFIXAdapter constructor failed");
}
}
OnixFIXAdapter::~OnixFIXAdapter() {
try {
session->logout();
Engine::instance()->unregisterListener();
Engine::shutdown();
LOG(OrderMan, Warn, "Object removed");
} catch (const OnixS::FIX::Exception &ex) {
std::string s = ex.what();
LOG(OrderMan, Error, "Exception {}", s);
bts_assert_msg(false, "OnixFIXAdapter deconstructor failed");
}
}
void OnixFIXAdapter::start() {
try {
LOG(OrderMan, Warn, "OnixFIXAdapter::start logging on");
const std::string counterpartyHost = "74.217.55.230";
const int heartBtInt = 30;
const int counterpartyPort = 22754;
session->logonAsInitiator(counterpartyHost, counterpartyPort, heartBtInt);
LOG(OrderMan, Warn, "OnixFIXAdapter::start logging on done");
} catch (const OnixS::FIX::Exception &ex) {
std::string errorMessage = ex.what();
LOG(OrderMan, Error, "OnixFIXAdapter::start Exception: {}", errorMessage);
session->logout();
Engine::shutdown();
bts_assert_msg(false, "OnixFIXAdapter session failed");
}
}
template <typename T>
uint translateEnum(T enum_value, const std::map<int, int> &translationMap) {
auto it = translationMap.find(enum_value);
assert(it != translationMap.end()); // Using assert instead of bts_assert
return it->second;
}
uint translateTimeInForce(api::orderman::TimeInForce time_in_force) {
//0 = Day 1 = GTC 2 = At the Open 3 = IOC 4 = FOK 5 = GTX
static const std::map<int, int> tifMap = {
{api::orderman::TimeInForce::TIME_IN_FORCE_DAY, 0},
{api::orderman::TimeInForce::TIME_IN_FORCE_GTC, 1},
{api::orderman::TimeInForce::TIME_IN_FORCE_IOC, 3}};
return translateEnum(time_in_force, tifMap);
}
uint translateOrderType(api::orderman::NativeOrderType order_type) {
//1 = Market 2 = Limit 3 = Stop 4 = Stop Limit 5 = Market on close A = On Close B = Limit on Close
static const std::map<int, int> orderTypeMap = {
{api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_MARKET, 1},
{api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_LIMIT, 2},
{api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_STOP, 3},
{api::orderman::NativeOrderType::NATIVE_ORDER_TYPE_STOP_LIMIT, 4}};
return translateEnum(order_type, orderTypeMap);
}
uint translateSide(api::Side side) {
//1 = Buy 2 = Sell 5 = Sell short 6 = Sell short Exempt
static const std::map<int, int> sideMap = {
{api::Side::SIDE_BUY, 1}, {api::Side::SIDE_SELL, 2}};
return translateEnum(side, sideMap);
}
// void waitUntilEnterKey(const std::string &waitText = std::string()) {
// OnixS::Threading::ThisThread::sleep(1000);
// if (!waitText.empty())
// std::clog << std::endl << waitText << std::endl;
// #ifdef _WIN32
// _getch();
// #else
// getchar();
// #endif
// }
// static std::string get_value_from_file(const std::string &fallback, const std::string &name) {
// std::string filename =
// config::get<config::SystemConfig>().data_path() + name;
// std::ifstream infile(filename);
// if (infile.is_open()) {
// std::string file_value;
// infile >> file_value;
// infile.close();
// return file_value;
// } else {
// std::ofstream outfile(filename);
// if (!outfile) {
// throw std::runtime_error("Could not create the file.");
// }
// outfile << fallback;
// outfile.close();
// return fallback;
// }
// }
// static std::string get_symbol_from_file(const std::string &symbol) {
// return get_value_from_file(symbol, "symbol.txt");
// }
// static std::string get_exp_day_from_file(const std::string &exp_day) {
// return get_value_from_file(exp_day, "exp_day.txt");
// }
// static std::string get_exp_month_from_file(const std::string &exp_month) {
// return get_value_from_file(exp_month, "exp_month.txt");
// }
// static std::string get_exp_year_from_file(const std::string &exp_year) {
// return get_value_from_file(exp_year, "exp_year.txt");
// }
// static std::string get_put_from_file(const std::string &put) {
// return get_value_from_file(put, "put.txt");
// }
// static std::string get_strike_from_file(const std::string &strike) {
// return get_value_from_file(strike, "strike.txt");
// }
std::string pad_string(const std::string& str, size_t length, char padding = ' ') {
if (str.size() > length) {
return str.substr(0, length);
}
return str + std::string(length - str.size(), padding);
}
// Function to create the OSI Symbol
static std::string create_osi_symbol(const std::string &symbol, const uint32_t &year, const uint32_t &month,const uint32_t &day,const char &call_put_indicator,const double &strike_price) {
if (symbol.size() > 6) {
throw std::invalid_argument("Symbol length cannot exceed 6 characters.");
}
std::ostringstream osi_stream;
// Symbol: 6 bytes, padded with spaces
osi_stream << pad_string(symbol, 6);
// Year: last 2 digits of the year
osi_stream << std::setw(2) << std::setfill('0') << (year % 100);
// Month: 2 digits
osi_stream << std::setw(2) << std::setfill('0') << month;
// Day: 2 digits
osi_stream << std::setw(2) << std::setfill('0') << day;
// Call/Put Indicator: 1 byte
osi_stream << call_put_indicator;
// Strike Price: 5 bytes for dollars, 3 bytes for decimal part
int strike_dollars = static_cast<int>(strike_price);
int strike_decimals = static_cast<int>((strike_price - strike_dollars) * 1000);
osi_stream << std::setw(5) << std::setfill('0') << strike_dollars;
osi_stream << std::setw(3) << std::setfill('0') << strike_decimals;
return osi_stream.str();
}
void parse_option(Message &order, const api::Security &sec) {
// std::string put_call = get_put_from_file(sec.option().call_put() == bts::api::PutOrCall::PUTORCALL_CALL ? "1" : "0");
// std::string exp_day = get_exp_day_from_file(std::to_string(sec.option().expiration_day()));
// std::string exp_month = get_exp_month_from_file(std::to_string(sec.option().expiration_month()));
// std::string exp_year = get_exp_year_from_file(std::to_string(sec.option().expiration_year()));
// std::string maturity_month_year = exp_year + exp_month;
// std::string strike_price = get_strike_from_file(std::to_string(sec.option().strike()));
auto osi = create_osi_symbol(sec.symbol(),sec.option().expiration_year(),sec.option().expiration_month(), sec.option().expiration_day(), sec.option().call_put()== bts::api::PutOrCall::PUTORCALL_CALL?'c':'p', sec.option().strike());
// std::string osi_symbol = get_value_from_file(osi, "osi_symbol.txt");
order
.set(OnixS::FIX::FIX42::Tags::PutOrCall, //put_call)
sec.option().call_put() == bts::api::PutOrCall::PUTORCALL_CALL ? "1" : "0")
.set(OnixS::FIX::FIX42::Tags::StrikePrice, //strike_price)
std::to_string(sec.option().strike()))
.set(OnixS::FIX::FIX42::Tags::MaturityMonthYear,// maturity_month_year)
std::to_string(sec.option().expiration_month()) +std::to_string(sec.option().expiration_year()))
.set(OnixS::FIX::FIX42::Tags::MaturityDay, //exp_day)
std::to_string(sec.option().expiration_day()))
.set(OnixS::FIX::FIX42::Tags::SecurityType,"OPT")
.set(OnixS::FIX::FIX42::Tags::Symbol,// osi_symbol)
osi)
.set(OnixS::FIX::FIX42::Tags::SecurityExchange, "SMART"); // Example value, adjust accordingly
}
void OnixFIXAdapter::modify_order(const CancelReplaceOrderMsg &msg) {
try {
//use this file:///C:/Users/arham/Downloads/OnixS.FixEngineCpp-Ubuntu2204-gcc1210-c++17-x64-4.10.1/OnixS.FixEngineCpp-Ubuntu2204-gcc1210-c++17-x64-4.10.1/doc/html/group__double-decimal-conversion.html
LOG(OrderMan, Warn, "OnixFIXAdapter::modify_order modifying order");
// auto order_cfg = config::get<config::OnixFixCfg>();
//message logic
//create message
// Get the symbol from the file or write it if the file doesn't exist
std::string symbol =msg.create_fields.security->symbol();
Message order(
OnixS::FIX::FIX42::Values::MsgType::Order_Cancel_Replace_Request,
OnixS::FIX::ProtocolVersion::FIX_42);
order
.set(OnixS::FIX::FIX42::Tags::HandlInst, 2) // 1: DMA 2: ALGO 3:CARE
.set(
OnixS::FIX::FIX42::Tags::OrigClOrdID,
std::to_string(msg.create_fields.key.value()))
.set(
OnixS::FIX::FIX42::Tags::ClOrdID,
"upd:"+std::to_string(msg.create_fields.key.value())) // ID for order
.set(
OnixS::FIX::FIX42::Tags::Account,
msg.create_fields.routing_record->account())
.set(
OnixS::FIX::FIX42::Tags::TimeInForce,
translateTimeInForce(msg.create_fields.time_in_force))
.set(
OnixS::FIX::FIX42::Tags::OrdType,
translateOrderType(msg.create_fields.order_type))
.set(OnixS::FIX::FIX42::Tags::Price, msg.create_fields.price.value())
.set(OnixS::FIX::FIX42::Tags::Rule80A, "A") //A = AGENCY P = PRINCIPAL
.set(OnixS::FIX::FIX42::Tags::SenderSubID, "BTC")
.set(OnixS::FIX::FIX42::Tags::TargetSubID, "TRFXMRX")
.set(OnixS::FIX::FIX42::Tags::ClientID, "BTC")
.set(OnixS::FIX::FIX42::Tags::SecurityType,"OPT")
// .set(
// OnixS::FIX::FIX42::Tags::Symbol,
// msg.create_fields.security->symbol())
.set(OnixS::FIX::FIX42::Tags::CustomerOrFirm,0)
.set(OnixS::FIX::FIX42::Tags::OpenClose,"O")
.set(
OnixS::FIX::FIX42::Tags::Side,
translateSide(msg.create_fields.side))
.set(
OnixS::FIX::FIX42::Tags::OrderQty,
msg.create_fields.quantity.value())
.set(
OnixS::FIX::FIX42::Tags::TransactTime,
OnixS::FIX::Timestamp::utc(),
OnixS::FIX::TimestampFormat::YYYYMMDDHHMMSSMsec);
if (msg.create_fields.security->has_option()) {
parse_option(order, *msg.create_fields.security);}
else{
order.set(OnixS::FIX::FIX42::Tags::Symbol, symbol);
}
order.validate();
LOG(OrderMan, Warn, "order Validated");
order.set(847, "SMART");
std::string order_string = order.toString('|');
updated = true;
session->send(&order);
LOG(OrderMan, Warn, "Order update sent :{}", order_string);
}
// Shut down all reactors if they exist.
catch (const OnixS::FIX::Exception &ex) {
std::cout << "Exception: " << ex.what() << std::endl;
}
}
void OnixFIXAdapter::launch_new_order(const CreateOrderMsg &msg) {
LOG(OrderMan, Warn, "launch order {}", msg);
try {
// Listener listener2;
LOG(OrderMan, Warn, "OnixFIXAdapter::launch_new_order creating order");
// auto order_cfg = config::get<config::OnixFixCfg>() //add this
std::string symbol = msg.security->symbol();
Message order(
OnixS::FIX::FIX42::Values::MsgType::Order_Single,
OnixS::FIX::ProtocolVersion::FIX_42);
order
.set(OnixS::FIX::FIX42::Tags::HandlInst, 2) // 1: DMA 2: ALGO 3:CARE
.set(
OnixS::FIX::FIX42::Tags::ClOrdID,
std::to_string(msg.key.value())) // ID for order
.set(OnixS::FIX::FIX42::Tags::Account, msg.routing_record->account())
.set(
OnixS::FIX::FIX42::Tags::TimeInForce,
translateTimeInForce(msg.time_in_force))
.set(
OnixS::FIX::FIX42::Tags::OrdType,
translateOrderType(msg.order_type))
.set(OnixS::FIX::FIX42::Tags::Price, msg.price.value())
.set(OnixS::FIX::FIX42::Tags::Rule80A, "A") //A = AGENCY P = PRINCIPAL
.set(OnixS::FIX::FIX42::Tags::SenderSubID, "BTC")
.set(OnixS::FIX::FIX42::Tags::TargetSubID, "TRFXMRX")
.set(OnixS::FIX::FIX42::Tags::ClientID, "BTC")
.set(OnixS::FIX::FIX42::Tags::CustomerOrFirm,0)
.set(OnixS::FIX::FIX42::Tags::OpenClose,"O")
.set(OnixS::FIX::FIX42::Tags::Side, translateSide(msg.side))
.set(OnixS::FIX::FIX42::Tags::OrderQty, msg.quantity.value())
.set(
OnixS::FIX::FIX42::Tags::TransactTime,
OnixS::FIX::Timestamp::utc(),
OnixS::FIX::TimestampFormat::YYYYMMDDHHMMSSMsec);
if (msg.security->has_option()) {
parse_option(order, *msg.security);}
else{
order.set(OnixS::FIX::FIX42::Tags::Symbol, symbol)
.set(OnixS::FIX::FIX42::Tags::Currency,msg.security->currency());
}
std::string order_string = order.toString('|');
order.validate();
LOG(OrderMan, Warn, "order Validated");
order.set(847, "SMART");
state_handler->launch_order(msg);
session->send(&order);
// TrafixOrderStateHandler
// auto er = listener.fire_dummy_report(msg);
LOG(OrderMan, Warn, "Order sent :{}", order_string);
}
// Shut down all reactors if they exist.
catch (const OnixS::FIX::Exception &ex) {
std::cout << "Exception: " << ex.what() << std::endl;
}
}
void OnixFIXAdapter::register_callback(const OnExecutionReport &cb) {
LOG(OrderMan, Warn, "OnixFIXAdapter::register_callback callback registered");
listener.setCallback(cb);
}
void OnixFIXAdapter::cancel_order(const CancelOrderMsg &msg) {
LOG(OrderMan, Warn, "cancel order {}", msg.key);
// std::string symbol =
// get_symbol_from_file(msg.security->symbol());
try {
Message order_cancel(
OnixS::FIX::FIX42::Values::MsgType::Order_Cancel_Request,
OnixS::FIX::ProtocolVersion::FIX_42);
order_cancel.set(OnixS::FIX::FIX42::Tags::HandlInst, 2);
order_cancel.set(
OnixS::FIX::FIX42::Tags::ClOrdID, "cancel:"+std::to_string(msg.key.value()));
// .set(OnixS::FIX::FIX42::Tags::OrderQty, msg.manual_order_indicator)
std::string prefix = updated?"upd:":"";
order_cancel
.set(
OnixS::FIX::FIX42::Tags::OrigClOrdID,
prefix+std::to_string(msg.key.value()))
.set(OnixS::FIX::FIX42::Tags::Side, translateSide(msg.side))
// .set(OnixS::FIX::FIX42::Tags::Symbol, symbol)
.set(OnixS::FIX::FIX42::Tags::SecurityType,"OPT")
.set(
OnixS::FIX::FIX42::Tags::TransactTime,
OnixS::FIX::Timestamp::utc(),
OnixS::FIX::TimestampFormat::YYYYMMDDHHMMSSMsec);
if (msg.security->has_option()) {
parse_option(order_cancel, *msg.security);}
else{
order_cancel.set(OnixS::FIX::FIX42::Tags::Symbol, msg.security->symbol())
.set(OnixS::FIX::FIX42::Tags::Currency,msg.security->currency());
}
std::string order_str = order_cancel.toString('|');
LOG(OrderMan, Warn, "order_cancel Created: {}", order_str);
// order_cancel.validate(); fails becasuse of HandlInst
LOG(OrderMan, Warn, "order Validated");
order_cancel.set(847, "SMART");
updated = false;
session->send(&order_cancel);
LOG(OrderMan, Warn, "order cancel sent");
}
// Shut down all reactors if they exist.
catch (const OnixS::FIX::Exception &ex) {
std::cout << "Exception: " << ex.what() << std::endl;
}
}
} // namespace bts::orderman
@@ -0,0 +1,27 @@
#pragma once
#include "IFIXGenericAdapter.h"
#include "lib/orderman/gateways/OnixFIXListener.h"
#include "lib/orderman/gateways/TrafixOrderStateHandler.h"
#include <memory>
namespace bts::orderman {
class OnixFIXAdapter : public IFIXGenericAdapter {
public:
OnixFIXAdapter();
~OnixFIXAdapter() override;
void register_callback(const OnExecutionReport &report) override;
void launch_new_order(const CreateOrderMsg &msg) override;
void cancel_order(const CancelOrderMsg &msg) override;
void modify_order(const CancelReplaceOrderMsg &) override;
void start() override;
private:
std::shared_ptr<TrafixOrderStateHandler> state_handler;
OnixFixListener listener;
std::unique_ptr<Session> session;
bool updated = false;
};
} // namespace bts::orderman
@@ -0,0 +1,175 @@
#include "OnixFIXListener.h"
#include "lib/api/services/OrderService.pb.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/domain_model/common.h"
#include "lib/logging/logging.h"
#include "lib/orderman/gateways/IExecutionGateway.h"
#include "lib/orderman/gateways/IFIXGenericAdapter.h"
#include "lib/orderman/orderman_utils/ExecutionReport.h"
#include "lib/orderman/orderman_utils/orderman_common.h"
#include <OnixS/FIXEngine/FIX/FIX42.h>
#include <OnixS/FIXEngine/FIX/Tag.h>
#include <string>
namespace bts::orderman {
ExecutionReport OnixFixListener::fire_dummy_report(CreateOrderMsg msg) {
ExecutionReport *er_p = exe_report_pool.construct();
ExecutionReport &er = *er_p;
std::string account_name = "BTSIM";
std::array<char, 12> an_array = {};
size_t i = 0;
for (const char c : account_name) {
an_array[i++] = c;
}
er.account_name = AccountName(an_array);
to_array(msg.routing_record->account(), &er.account_name.value());
to_array(
msg.routing_record->operator_id(), &er.operator_id.value());
er.order_key = msg.key;
er.msg_seq_num = 0;
er.security = msg.security;
er.leg_security = nullptr;
er.side = msg.side;
er.time_in_force = msg.time_in_force;
er.order_type = msg.order_type;
er.exec_type = api::orderman::EXEC_TYPE_NEW;
er.order_status = api::orderman::NATIVE_ORDER_STATUS_NEW;
er.manual_order_indicator = api::orderman::MANUAL_ORDER_INDICATOR_NO;
ExchangeOrderID64 eoid = ExchangeOrderID64(exchange_oid++);
ExchangeOrderIDVar oid = eoid;
er.exchange_order_id = oid;
er.price = msg.price;
er.quantity = msg.quantity;
er.filled_quantity = Quantity(0);
er.last_shares = Quantity(0);
er.msg = "Message Placed";
er.exchange = msg.security->exchange();
// Set transaction time
ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
er.transact_time = now;
cb(er,{});
return er;
}
ExecutionReport OnixFixListener::get_exe_report(Message msg) {
ExecutionReport *er_p = exe_report_pool.construct();
ExecutionReport &er = *er_p;
auto msg_str = msg.toString();
LOG(OrderMan,Info, "OnixFixListener::get_exe_report {}", msg_str);
std::string account_name = "BTSIM";
std::string ops_id = "op_id_bts";
std::array<char, 12> an_array = {};
size_t i = 0;
for (const char c : account_name) {
an_array[i++] = c;
}
er.account_name = AccountName(an_array);
to_array(
ops_id, &er.operator_id.value());
auto extractUInt = [](const std::string& input) -> uint {
std::string numericPart;
std::copy_if(input.begin(), input.end(), std::back_inserter(numericPart), [](char c) {
return std::isdigit(c);
});
if (numericPart.empty()) {
throw std::invalid_argument("Invalid format: no numeric characters found");
}
uint number = std::stoul(numericPart);
return number;
};
er.order_key = OrderKey(extractUInt(msg.get(OnixS::FIX::FIX42::Tags::OrderID).toString()));
er.msg_seq_num = 0;
// er.security = ;
er.leg_security = nullptr;
er.side = msg.get(OnixS::FIX::FIX42::Tags::Side).toString()=="1"?api::Side::SIDE_BUY:api::Side::SIDE_SELL;
er.time_in_force = api::orderman::TIME_IN_FORCE_DAY;
er.order_type = api::orderman::NATIVE_ORDER_TYPE_LIMIT;
er.exec_type = api::orderman::ExecType(std::stoi(msg.get(OnixS::FIX::FIX42::Tags::ExecType).toString()));
er.order_status = api::orderman::NATIVE_ORDER_STATUS_NEW;
er.manual_order_indicator = api::orderman::MANUAL_ORDER_INDICATOR_NO;
ExchangeOrderID64 eoid = ExchangeOrderID64(exchange_oid++);
ExchangeOrderIDVar oid = eoid;
er.exchange_order_id = oid;
er.price = Price(std::stod(msg.get(OnixS::FIX::FIX42::Tags::Price)));
er.quantity = Quantity(std::stoi(msg.get(FIX42::Tags::OrderQty)));
er.filled_quantity = Quantity(0);
er.last_shares = Quantity(0);
er.msg = "Message Placed";
er.exchange = api::VENUE_CBOE;
// Set transaction time
ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
er.transact_time = now;
return er;
}
void OnixFixListener::onInboundApplicationMsg(Message &msg, Session * /*sn*/) {
std::string msg_str = msg.toString();
std::string msg_json = msg.toJson();
LOG(OrderMan, Warn,
"Incoming application-level message: {} As json {}",
msg_str, msg_json);
FieldValueRef t = msg.type();
char message_type;
if(!t.toChar(message_type)){
LOG(OrderMan, Warn,
"no message type message: {} As json {}",
msg_str, msg_json);
return;
}
switch (message_type) {
case 'D':
{
//exection report
auto er = get_exe_report(msg);
cb(er,{});
}
}
// LOG(OrderMan, Warn,
// "onInboundApplicationMsg::inbound firing er: {}",
// er->to_string());
// cb(*er, {});
}
void OnixFixListener::onInboundSessionMsg(Message & msg, Session * /*sn*/)
{
std::string msg_str = msg.toString();
std::string msg_json = msg.toJson();
LOG( OrderMan, Warn,
"Incoming session-level message:\n {} As json {}",
msg_str,
msg_json);
}
void OnixFixListener::onStateChange(SessionState::Enum newState, SessionState::Enum prevState,
Session * /*sn*/)
{
std::string prevst = SessionState::toString(prevState);
std::string newst = SessionState::toString(newState);
LOG(OrderMan, Warn,
"Session's state is changed, prevState = {} , newState= {}",
prevst,
newst);
}
void OnixFixListener::onError(ErrorReason::Enum /*reason*/, const std::string & description, Session * /*sn*/)
{
LOG( OrderMan, Error, "Session-level error: {}", description);
}
void OnixFixListener::onWarning(WarningReason::Enum /*reason*/, const std::string & description,
Session * /*sn*/)
{
LOG( OrderMan, Error, "Session-level warning: {}", description);
}
} // namespace bts::orderman
@@ -0,0 +1,52 @@
#pragma once
#include "lib/orderman/CreateOrderMsg.h"
#include "lib/orderman/gateways/IExecutionGateway.h"
#include "lib/orderman/gateways/TrafixOrderStateHandler.h"
#include "lib/orderman/orderman_utils/ExecutionReport.h"
#include <OnixS/FIXEngine.h>
#include <OnixS/FIXEngine/FIX/Message.h>
#include <boost/pool/object_pool.hpp>
namespace bts::orderman {
using namespace OnixS::FIX;
class OnixFixListener : public OnixS::FIX::ISessionListener {
public:
OnixFixListener( std::shared_ptr<TrafixOrderStateHandler> state_handler) : state_handler(state_handler),cb(dummy) {
LOG(OrderMan, Info, "OnixFixListener::cb set callback registered");
}
std::shared_ptr<TrafixOrderStateHandler> state_handler;
unsigned long exchange_oid = 0;
/// Is called when the application-level message is received from the counterparty.
void onInboundApplicationMsg(Message &msg, Session *sn)
ONIXS_FIXENGINE_FINAL;
// Set the real callback function
void setCallback(const OnExeReport &callback) {
LOG(OrderMan, Info, "OnixFixListener::setCallback callback registered");
cb = callback;
}
/// Is called when the session-level message is received from the counterparty.
void onInboundSessionMsg(Message &msg, Session *sn) ONIXS_FIXENGINE_FINAL;
void onStateChange(
SessionState::Enum newState,
SessionState::Enum prevState,
Session *sn) ONIXS_FIXENGINE_FINAL;
OnExeReport cb;
ExecutionReport fire_dummy_report(CreateOrderMsg msg);
ExecutionReport get_exe_report(Message msg);
static void dummy(const ExecutionReport &, const trace::CorID &) {
LOG(OrderMan, Info, "OnixFixListener::Dummy fired");
}
void onError(
ErrorReason::Enum reason,
const std::string &description,
Session *sn) ONIXS_FIXENGINE_FINAL;
void onWarning(
WarningReason::Enum reason,
const std::string &description,
Session *sn) ONIXS_FIXENGINE_FINAL;
boost::object_pool<ExecutionReport> exe_report_pool;
};
} // namespace bts::orderman
@@ -0,0 +1,82 @@
#include "TrafixOrderStateHandler.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/orderman/CreateOrderMsg.h"
#include "lib/orderman/order_factory/IOrderFactory.h"
#include "lib/orderman/orderman_utils/orderman_common.h"
#include <string>
namespace bts::orderman {
// Internal structure for storing order info
struct TrafixOrderStateHandler::OrderInfo {
OrderInfo(const CreateOrderMsg &msg)
: update_count(0), routing_record(msg.routing_record) {}
uint update_count;
const api::orderman::RoutingRecord* routing_record;
std::string update_id(bts::orderman::OrderKey key) {
return std::to_string(update_count++) + ":" + std::to_string(key.value());
}
};
// PImpl structure containing the actual implementation details, including SecurityOracle
struct TrafixOrderStateHandler::PImpl {
PImpl() {}
std::unordered_map<bts::orderman::OrderKey, OrderInfo> orders;
// md::SecurityOracle sec_oracle;
};
// Constructor
TrafixOrderStateHandler::TrafixOrderStateHandler()
: me(std::make_unique<PImpl>()) {}
// Destructor
TrafixOrderStateHandler::~TrafixOrderStateHandler() = default;
// Method to launch an order
bts::orderman::OrderKey TrafixOrderStateHandler::launch_order(const CreateOrderMsg &msg) {
me->orders.emplace(msg.key, OrderInfo(msg));
return bts::orderman::OrderKey(msg.key);
}
// Method to modify an order
bts::orderman::OrderKey TrafixOrderStateHandler::modify_order(const std::string& key) {
auto order_key = get_order_key(key);
auto it = me->orders.find(order_key);
if (it != me->orders.end()) {
// Modify the order here
return bts::orderman::null_order_key();
}
return it->first;
}
// Utility method to get order key from a string
bts::orderman::OrderKey TrafixOrderStateHandler::get_order_key(const std::string& id) {
std::size_t pos = id.find(':');
if (pos == std::string::npos) {
return null_order_key();
}
int key_value = std::stoi(id.substr(pos + 1));
return bts::orderman::OrderKey(key_value);
}
// Method to get update number from string
int TrafixOrderStateHandler::get_update_number(const std::string& id) {
std::size_t pos = id.find(':');
if (pos == std::string::npos) {
return -1; // No colon found
}
std::string prefix = id.substr(0, pos);
// Check if the prefix contains only digits
if (prefix.empty() || !std::all_of(prefix.begin(), prefix.end(), ::isdigit)) {
return -1; // Not a valid number
}
// Convert the valid numeric prefix to an integer
return std::stoi(prefix);
}
} // namespace b
@@ -0,0 +1,28 @@
#pragma once
#include "IFIXGenericAdapter.h"
#include "lib/orderman/gateways/OnixFIXListener.h"
#include "lib/orderman/orderman_utils/orderman_common.h"
#include <memory>
namespace bts::orderman {
class TrafixOrderStateHandler {
public:
TrafixOrderStateHandler();
~TrafixOrderStateHandler();
bts::orderman::OrderKey launch_order(const CreateOrderMsg &msg);
bts::orderman::OrderKey modify_order(const std::string &key);
bts::orderman::OrderKey cancel_order(const std::string &key);
std::string get_marex_id(const bts::orderman::OrderKey& key);
static bts::orderman::OrderKey get_order_key(const std::string& id);
static int get_update_number(const std::string& id);
private:
struct OrderInfo;
struct PImpl;
std::unique_ptr<PImpl> me;
};
} // namespace bts::orderman
@@ -61,15 +61,14 @@ SimExecutionGatewayFactory *SimExecutionGatewayFactory::get() {
}
bool SimExecutionGatewayFactory::is_sim_execution_on(api::Venue venue) {
// TODO: Arham, fix this !
return true;
const auto &sim_exec = config::get<bts::config::SimExecution>();
for (const auto &e : sim_exec.entries()) {
// return true;
const auto &not_sim_exec = config::get<bts::config::NotSimExecution>();
for (const auto &e : not_sim_exec.entries()) {
if (e.venue() == venue) {
return true;
return false;
}
}
return false;
return true;
}
} // namespace bts::orderman
@@ -129,7 +129,7 @@ NativeOrderBase *OrderFactory::make_order(const api::Venue exchange) {
return new ICENativeOrder();
else if (exchange == api::VENUE_GLBX || exchange == api::VENUE_SIM)
return new GLBXNativeOrder();
return new EMNativeOrder();
return new EMNativeOrder(exchange);
}
NativeOrderBase *OrderFactory::make_native_order(
+15 -6
View File
@@ -5,6 +5,7 @@
#include "lib/config/protos/toggles.pb.h"
#include "lib/domain/OrderStatus.h"
#include "lib/domain_model/Security.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/domain_model/common.h"
#include "lib/jetfix/codec/FIXFragment.h"
#include "lib/jetfix/codec/FieldDefinitions.h"
@@ -52,10 +53,10 @@ struct InflightUpdateState {
};
struct EMNativeOrder::PImpl {
PImpl()
PImpl(const api::Venue exchange)
: instance_id(EMNativeOrder::count++), services(nullptr),
session_services(), sec(nullptr), previous_working_position(0),
definition(), state(), exchange_order_id() {}
definition(), state(), exchange_order_id(), exchange(exchange) {}
const int instance_id;
bool inited = false;
OrderServices *services;
@@ -67,7 +68,7 @@ struct EMNativeOrder::PImpl {
OrderDefinition definition;
OrderState state;
std::optional<ExchangeOrderIDVar> exchange_order_id;
const api::Venue exchange;
OrderServices::ExeReportSubID exe_report_sid;
OrderServices::CancelRejectSubID cancel_reject_sid;
// If a cancel is sent before the order placement is ack'ed this is set to
@@ -82,7 +83,7 @@ struct EMNativeOrder::PImpl {
RoutingRecord routing_record;
};
EMNativeOrder::EMNativeOrder() : me(new PImpl()) {
EMNativeOrder::EMNativeOrder(const api::Venue exchange) : me(new PImpl(exchange)) {
LOG(OrderMan, Info,
"EMNativeOrder::EMNativeOrder instance_id: {} t_id: {} t_name: {}",
me->instance_id, os::get_thread_id(), os::get_application_name());
@@ -501,11 +502,19 @@ OrderOpResult EMNativeOrder::_on_activate(
me->state.set_filled_quantity(0);
me->state.set_last_price(0);
me->state.set_avg_price(0);
me->session_services = &me->services->get_session_services(
get_native_state(me->state).routing_record().session());
// TODO: just to get going
me->session_services = &me->services->get_session_services("deribit_1");
std::string session_name;
switch (me->exchange) {
case api::VENUE_CBOE:
session_name = "cboe_1";
break;
default:
session_name ="deribit_1";
}
me->session_services = &me->services->get_session_services(session_name);
// Checks for the following:
// 1. Gateway connection
+2 -1
View File
@@ -1,6 +1,7 @@
#pragma once
#include "lib/api/services/OrderService.pb.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/domain_model/common.h"
#include "lib/orderman/CreateOrderMsg.h"
#include "lib/orderman/orders/NativeOrderBase.h"
@@ -19,7 +20,7 @@ class SessionServices;
// Wrapper around Emumba's order
class EMNativeOrder : public NativeOrderBase {
public:
EMNativeOrder();
EMNativeOrder(const api::Venue exchange = api::VENUE_DERIBIT);
~EMNativeOrder() override;
api::orderman::OrderDefinition const &definition() const override;
void on_init_new(
+4 -4
View File
@@ -478,7 +478,7 @@ OrderOpResult GLBXNativeOrder::on_activate(
now.to_uint64(), (now - corid.app_arrival).nanoseconds());
// in sim mode, we dont need this,
me->encoder->send_create(*me->services, *me->session_services);
if (me->in_sim_mode) {
if (me->in_sim_mode) { //CME EXECUTION NEEDS TO BE UPDATED
CreateOrderMsg c_msg = {};
_create_create_msg(c_msg);
me->services->queue_order_msg(*me->session_services, c_msg);
@@ -579,7 +579,7 @@ OrderOpResult GLBXNativeOrder::on_update(api::orderman::Update const &update) {
auto p = Price(update.price());
auto q = Quantity(update.quantity());
auto send_update_in_sim_mode = [&]() {
if (me->in_sim_mode) {
if (me->in_sim_mode) { //CME EXECUTION NEEDS TO BE UPDATED
CancelReplaceOrderMsg cxl_repl_msg = {};
_create_create_msg(cxl_repl_msg.create_fields);
Price new_p = Price(update.price());
@@ -799,7 +799,7 @@ OrderOpResult GLBXNativeOrder::on_cancel() {
me->encoder->update_cancel_encoding(*me->services);
me->encoder->send_cancel(*me->services, *me->session_services);
// TODO: check for sim mode here
if (me->in_sim_mode) {
if (me->in_sim_mode) { //CME EXECUTION NEEDS TO BE UPDATED
CancelOrderMsg cxl_msg = {};
cxl_msg.key = key();
LOG(OrderMan, Info, "GLBXNativeOrder queue_order_msg for cxl");
@@ -1381,7 +1381,7 @@ void GLBXNativeOrder::_create_create_msg(CreateOrderMsg &c) {
bts::orderman::get_quantity(me->state, StateQuantityType::OrderQuantity);
c.security = me->sec;
c.side = me->definition.side();
c.order_type = get_glbx_native_ord_type(me->definition);
c.order_type = get_glbx_native_ord_type(me->definition); //CME EXECUTION NEEDS TO BE UPDATED
c.time_in_force = me->definition.time_in_force();
c.order_type = api::orderman::NATIVE_ORDER_TYPE_LIMIT;
c.manual_order_indicator = me->definition.manual_order_indicator();
@@ -277,6 +277,7 @@ void iLink2OrderEncoder::update_exchange_order_id(ExchangeOrderIDVar const &v) {
auto oid = to_trim_str(std::get<ExchangeOrderID>(v));
me->update_encoding.add<jetfix::fix::OrderID>(oid);
me->cancel_encoding.add<jetfix::fix::OrderID>(oid);
//CME EXECUTION NEEDS TO BE UPDATED, can be commented out
}
void iLink2OrderEncoder::update_create_order_encoding(
+34 -29
View File
@@ -35,7 +35,7 @@
#include "server/sparkd/md/okx/OkxSubscription.h"
#include "server/sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
#include "server/sparkd/md/recording/MDReplayer.h"
#include "server/sparkd/md/taifex/TaifexDataHandler.h"
#include "server/sparkd/md/taifex/TaifexDataGateway.h"
#include <bits/stdint-uintn.h>
#include <boost/filesystem.hpp>
@@ -43,6 +43,7 @@
#include <emumba-deribit-connector/deribit_utils.h>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
@@ -117,10 +118,6 @@ struct MarketDataGateway::PImpl {
std::vector<std::unique_ptr<MDReplayData>> replays = {};
// we create these and pass them to the cc md sources
std::vector<std::shared_ptr<VenueListenerData>> venue_listeners = {};
void start_glbx();
void start_cboe();
void start_taifex();
};
MarketDataGateway::MarketDataGateway()
@@ -142,9 +139,9 @@ void MarketDataGateway::run() {
bts_assert(me_->md_config.emumba_deribit().size() == 1);
_start_emumba_deribit(me_->md_config.emumba_deribit().at(0));
}
me_->start_glbx();
me_->start_taifex();
me_->start_cboe();
start_glbx(me_->md_config);
start_taifex(me_->md_config);
start_cboe(me_->md_config);
_start_replays(me_->md_config);
while (true) {
std::this_thread::sleep_for(std::chrono::seconds(1));
@@ -247,14 +244,17 @@ void MarketDataGateway::unsubscribe(MarketDataListenerID id) {
}
/// glbx //////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////
void MarketDataGateway::PImpl::start_glbx(){
for (const auto &glbx : md_config.glbx()) {
void MarketDataGateway::start_glbx(const bts::api::md::MarketDataConfig &cfg){
for (const auto &glbx : cfg.glbx()) {
if(glbx.use_databento()){
auto p = venue_listeners.emplace_back(
auto p = me_->venue_listeners.emplace_back(
std::make_shared<VenueListenerData>(bts::api::VENUE_GLBX));
std::string tid_name = "glbx";
bts::threading::make_thread(tid_name, [&]() {
bts::md::glbx::GlbxMarketDatahandler::run(glbx.read_from_file(), p->listener,sec_id_store,api::domain::VENUE_GLBX);
api::md::glbx::GlbxMarketDatahandler::run(
p->listener,
me_->sec_id_store,
glbx);
}).detach();
}
else{
@@ -372,31 +372,36 @@ void MarketDataGateway::_start_replays(
os::get_thread_id(), os::get_thread_name());
}
void MarketDataGateway::PImpl::start_taifex() {
void MarketDataGateway::start_taifex(const bts::api::md::MarketDataConfig &cfg) {
// taifex
for (const auto &taifex : md_config.taifex()) {
auto p = venue_listeners.emplace_back(
int counter = 0;
for (const auto &taifex : cfg.taifex()) {
counter++;
auto p = me_->venue_listeners.emplace_back(
std::make_shared<VenueListenerData>(bts::api::VENUE_TAIFEX));
std::string tid_name = "taifex";
std::string tid_name = "taifex" +std::to_string(counter);
bts::threading::make_thread(tid_name, [&]() {
bts::md::taifex::TaifexDatahandler::run(
taifex.read_from_file(), p->listener, sec_id_store);
bts::md::taifex::TaifexDataGateway::run(
taifex, p->listener, me_->sec_id_store);
}).detach();
}
}
void MarketDataGateway::PImpl::start_cboe(){
void MarketDataGateway::start_cboe(const bts::api::md::MarketDataConfig &cfg){
// cboe
for (const auto &cboe : md_config.cboe()) {
auto p = venue_listeners.emplace_back(
std::make_shared<VenueListenerData>(bts::api::VENUE_CBOE));
std::string tid_name = "CBOE";
bts::threading::make_thread(tid_name, [&]() {
bts::md::cboe::CboeMarketDatahandler::run(cboe.read_from_file(), p->listener,sec_id_store,api::domain::VENUE_CBOE);
}).detach();
UNUSED(cboe);
}
for (const api::md::CboeMarketDataConfig &cboe : cfg.cboe()) {
auto p = me_->venue_listeners.emplace_back(
std::make_shared<VenueListenerData>(bts::api::VENUE_CBOE));
std::string tid_name = "CBOE";
bts::threading::make_thread(tid_name, [&]() {
api::md::cboe::CboeMarketDatahandler::run(
p->listener,
me_->sec_id_store,
cboe);
}).detach();
UNUSED(cboe);
}
}
+3
View File
@@ -36,6 +36,9 @@ class MarketDataGateway {
static void _start_onix_glbx(
bts::api::md::GlbxMarketDataConfig cfg);
void _start_replays(const bts::api::md::MarketDataConfig &cfg);
void start_glbx(const bts::api::md::MarketDataConfig &cfg);
void start_cboe(const bts::api::md::MarketDataConfig &cfg);
void start_taifex(const bts::api::md::MarketDataConfig &cfg);
};
} // namespace bts::md
@@ -2,25 +2,35 @@
#include "lib/api/domain/Venue.pb.h"
#include "lib/config/config.h"
#include "lib/config/protos/system_config.pb.h"
#include "lib/utils/bts_assert.h"
#include "server/sparkd/md/databento/DatabentoDataHandler.h"
#include <cassert>
#include <chrono>
#include <cmath>
#include <databento/constants.hpp>
#include <databento/historical.hpp>
#include <memory>
#include <string>
namespace bts::md::cboe {
namespace bts::api::md::cboe {
void CboeMarketDatahandler::run(
bool use_file,
std::shared_ptr<IMarketDataListener> listner,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
bts::api::domain::Venue venue) {
std::string contratcs_file_name = "/opra-pillar-20240815.definition.dbn.zst";
std::string bbo_file_name = "/opra-pillar-20240815.mbp-1.dbn.zst";
std::string base_path =
config::get<bts::config::SystemConfig>().data_path();
DatabentoHandler::DatabentoDatahandler::run(use_file, listner,sec_id_store,venue,base_path + contratcs_file_name,base_path + bbo_file_name);
std::shared_ptr<::bts::md::IMarketDataListener> listner,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
const CboeMarketDataConfig &config) {
bts_assert_msg(config.use_databento(), "CboeMarketDatahandler only supports databento");
std::string base_path =
config::get<bts::config::SystemConfig>().data_path();
bts_assert(config.has_databento_settings());
bts::md::DatabentoHandler::DatabentoDatahandler::run(
config.databento_settings().read_from_file(),
listner,
sec_id_store,
domain::VENUE_CBOE,
base_path + config.databento_settings().contracts_file_name(),
base_path + config.databento_settings().bbo_file_name(),
config.databento_settings().key(),
config.databento_settings().dataset(),
config.databento_settings().filter_securities_one_year());
}
} // namespace bts::md::cboe
@@ -1,12 +1,12 @@
#pragma once
#include "lib/api/md/CboeMarketDataConfig.pb.h"
#include "lib/md/IMarketDataListener.h"
#include "lib/persistence/IDGenerator.h"
namespace bts::md::cboe {
namespace bts::api::md::cboe {
class CboeMarketDatahandler {
public:
static void run(bool use_file,
std::shared_ptr<IMarketDataListener> listner,
static void run(std::shared_ptr<::bts::md::IMarketDataListener> listner,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
bts::api::domain::Venue venue);
const CboeMarketDataConfig &config);
};
} // namespace bts::md::cboe
@@ -1,5 +1,9 @@
#include "DatabentoDataHandler.h"
#include "databento/datetime.hpp"
#include "databento/enums.hpp"
#include "databento/exceptions.hpp"
#include "databento/live_blocking.hpp"
#include "databento/log.hpp"
#include "databento/timeseries.hpp"
#include "lib/api/domain/Venue.pb.h"
#include "lib/config/config.h"
@@ -8,6 +12,8 @@
#include "lib/domain/Quantity.h"
#include "lib/domain/Security.h"
#include "lib/domain/SecurityID.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/logging/logging.h"
#include "lib/utils/TimeStamp.h"
#include "lib/utils/bts_assert.h"
#include "lib/utils/bts_chrono.h"
@@ -16,6 +22,7 @@
#include <chrono>
#include <databento/constants.hpp>
#include <databento/historical.hpp>
#include <databento/live.hpp>
#include <optional>
#include <string>
@@ -39,30 +46,47 @@ void DatabentoDatahandler::run(
bts::api::domain::Venue venue,
const std::string &securities_path,
const std::string &books_path,
const std::string &key,
const std::string &dataset,
const bool filter_securities,
const std::chrono::duration<long double> sleep_duration) {
auto client = databento::HistoricalBuilder{}
.SetKey("db-Q4eCbTQfrAHaG6sAcuFUYB9YqkF6j")
.Build();
auto sleep_duration_updated = sleep_duration;
if (!use_file) {
sleep_duration_updated = std::chrono::duration<long double>::zero(); // Set to zero for live data
}
auto process_security = [&](const databento::Record &record) {
const auto &msg = record.Get<databento::InstrumentDefMsg>();
std::string id = remove_spaces(msg.RawSymbol());
std::string symbol;
switch (venue) {
case api::domain::Venue::VENUE_CBOE:
symbol = id.substr(0, 3);
case api::domain::Venue::VENUE_CBOE:{
symbol = remove_spaces(std::string(msg.underlying.data(),msg.underlying.size()));
break;
}
case api::domain::Venue::VENUE_GLBX:
if (id.length() > 5)
return databento::KeepGoing::Continue;
symbol = id.substr(0, 2);
// sec.security_common.cpp = msg.unit_of_measure_qty;
break;
default:
bts_assert_msg(false, "Exchange not supported on databento");
return databento::KeepGoing::Stop;
}
if(filter_securities){
const auto exp_nanos = msg.expiration;
std::string exp_string = databento::ToIso8601(exp_nanos);
const auto expiration_time = bts::core::TimeStamp::from_iso_string(exp_string);
if(expiration_time.seconds() < expiration_time.now().seconds() || expiration_time.seconds() > (expiration_time.now().seconds() + 60*60*24*365)){
return databento::KeepGoing::Continue;
}
}
bts::domain::SecurityID sec_id{sec_id_store->get_or_add(
venue, std::to_string(msg.hd.instrument_id))};
bts::domain::Security sec;
@@ -71,13 +95,14 @@ void DatabentoDatahandler::run(
sec.security_common.id = sec_id;
sec.security_common.exchange_sec_id = msg.hd.instrument_id;
sec.security_common.desc = msg.RawSymbol();
sec.security_common.symbol = symbol;
sec.security_common.asset = symbol;
sec.security_common.group = symbol;
sec.security_common.symbol = msg.Asset();
sec.security_common.asset = msg.Asset();
sec.security_common.group = msg.Group();
sec.security_common.min_price_increment = 0.01;
sec.security_common.min_quantity = 0.01;
sec.security_common.min_quantity_increment = 0.01;
sec.security_common.currency = {msg.currency.data(), msg.currency.size()};
// sec.security_common.cpp = msg.hd.
const std::string_view sec_type = {msg.security_type.data(), msg.security_type.size()};
const auto exp_nanos = msg.expiration;
@@ -85,8 +110,15 @@ void DatabentoDatahandler::run(
const auto expiration_time = bts::core::TimeStamp::from_iso_string(exp_string);
if (sec_type.find("OPT") != std::string::npos) {
auto call_put = api::domain::CallPut::PUT;
if(msg.instrument_class == 'C' ){
call_put=api::domain::CallPut::CALL;
}
else{
bts_assert(msg.instrument_class == 'P');
}
sec.security.emplace<bts::domain::OptionSecurity>(
expiration_time, bts::domain::Price::from_double(msg.strike_price));
expiration_time, bts::domain::Price(msg.strike_price, -9),call_put);
} else if (sec_type.find("FUT") != std::string::npos) {
sec.security.emplace<bts::domain::FutureSecurity>(expiration_time);
} else {
@@ -105,7 +137,7 @@ void DatabentoDatahandler::run(
if (!sec_id_opt.has_value())
return databento::KeepGoing::Continue;
sleep_for(sleep_duration);
sleep_for(sleep_duration_updated);
bts::domain::SecurityID sec_id = bts::domain::SecurityID(sec_id_opt.value());
domain::Book book;
book.set_security_id(sec_id);
@@ -126,15 +158,81 @@ void DatabentoDatahandler::run(
utils::TimeStamp{datetime::ptime_to_timestamp(recv_time.to_ptime())});
return databento::KeepGoing::Continue;
};
if (use_file) {
std::string base_path =
config::get<bts::config::SystemConfig>().data_path();
auto client = databento::HistoricalBuilder{}
.SetKey(key)
.Build();
databento::DbnFileStore dbn_file_store(securities_path);
dbn_file_store.Replay(process_security);
databento::DbnFileStore dbn_file_store2(books_path);
dbn_file_store2.Replay(process_bbo);
}else {
bts_assert(dataset!="");
// auto client = databento::LiveBuilder{}
// .SetKey(key)
// .SetDataset(dataset)
// .BuildThreaded();
auto client = databento::LiveBlocking{
databento::ILogReceiver::Default(),
key,
dataset,
true,
databento::VersionUpgradePolicy::Upgrade,
{}
};
//hardcoding for now will update later
switch (venue) {
case api::domain::VENUE_CBOE:
client.Subscribe({"SPX.OPT","SPXW.OPT"}, databento::Schema::Definition, databento::SType::Parent,databento::UnixNanos{});
client.Subscribe({"SPX.OPT","SPXW.OPT"}, databento::Schema::Mbp1, databento::SType::Parent);
break;
case api::domain::VENUE_GLBX:
client.Subscribe({"ES.FUT"}, databento::Schema::Definition, databento::SType::Parent);
client.Subscribe({"ES.FUT"}, databento::Schema::Mbp1, databento::SType::Parent);
break;
default:
bts_assert_msg(false, "ONLY GLBX AND CBOE SUPPORTED FOR NOW");
}
client.Start();
while (true) {
try {
auto* record = client.NextRecord(std::chrono::seconds{500});
if (record == nullptr) {
LOG(md, Error, "Timed out waiting for data");
break;
}
if (auto* bbo = record->GetIf<databento::Mbp1Msg>()) {
process_bbo(*record);
}
if (auto* def = record->GetIf<databento::InstrumentDefMsg>()) {
process_security(*record);
}
} catch (const databento::TcpError& error) {
std::string err = error.what();
LOG(md, Error, "Caught TCP error: {} \nAttempting reconnect\n",err);
client.Reconnect();
} catch(const databento::DbnResponseError& err){
std::string msg = err.Message();
std::string what = err.what();
LOG(md, Error, "Caught DbnResponseError what: {} ,msg:{} \nShutting down\n",msg, what);
client.Stop();
// bts_assert(false);
}
catch(const databento::Exception& excp){
std::string err = excp.Message();
std::string err2 = excp.what();
LOG(md, Error, "Caught unknown dadtabento::exception what: {}, msg: {} \nShutting down\n",err, err2);
client.Stop();
// bts_assert(false);
}catch (const std::exception& exc) {
std::string err = exc.what();
LOG(md, Error, "Caught unknown exception: {} \nShutting down\n",err);
client.Stop();
// bts_assert(false);
}
}
}
}
@@ -12,6 +12,9 @@ class DatabentoDatahandler {
bts::api::domain::Venue venue,
const std::string& securities_path,
const std::string& books_path,
const std::string &dataset="",
const std::string& key = "db-Q4eCbTQfrAHaG6sAcuFUYB9YqkF6j",
const bool filter_securities=false,
const std::chrono::duration<long double> sleep_duration = 0.04s);
};
} // namespace bts::md::databento
@@ -1,7 +1,9 @@
#include "GlbxMarketDataHandler.h"
#include "lib/api/domain/Venue.pb.h"
#include "lib/api/md/GlbxMarketDataConfig.pb.h"
#include "lib/config/config.h"
#include "lib/config/protos/system_config.pb.h"
#include "lib/utils/bts_assert.h"
#include "server/sparkd/md/databento/DatabentoDataHandler.h"
#include <cassert>
#include <cmath>
@@ -9,18 +11,27 @@
#include <databento/historical.hpp>
#include <memory>
#include <string>
namespace bts::md::glbx {
namespace bts::api::md::glbx {
void GlbxMarketDatahandler::run(
bool use_file,
std::shared_ptr<IMarketDataListener> listner,
std::shared_ptr<::bts::md::IMarketDataListener> listner,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
bts::api::domain::Venue venue) {
std::string contratcs_file_name = "/glbx-mdp3-20240815.definition.dbn.zst";
std::string bbo_file_name = "/glbx-mdp3-20240815.mbp-1.dbn.zst";
std::string base_path =
config::get<bts::config::SystemConfig>().data_path();
DatabentoHandler::DatabentoDatahandler::run(use_file, listner,sec_id_store,venue,base_path + contratcs_file_name,base_path + bbo_file_name);
const GlbxMarketDataConfig &config) {
bts_assert_msg(config.use_databento(),
"GlbxMarketDatahandler only supports databento");
std::string base_path =
config::get<bts::config::SystemConfig>().data_path();
bts_assert(config.has_databento_settings());
bts::md::DatabentoHandler::DatabentoDatahandler::run(
config.databento_settings().read_from_file(),
listner,
sec_id_store,
domain::VENUE_GLBX,
base_path + config.databento_settings().contracts_file_name(),
base_path + config.databento_settings().bbo_file_name(),
config.databento_settings().key(),
config.databento_settings().dataset(),
config.databento_settings().filter_securities_one_year());
}
} // namespace bts::md::glbx
} // namespace bts::api::md::glbx
@@ -1,12 +1,13 @@
#pragma once
#include "lib/api/md/GlbxMarketDataConfig.pb.h"
#include "lib/md/IMarketDataListener.h"
#include "lib/persistence/IDGenerator.h"
namespace bts::md::glbx {
namespace bts::api::md::glbx {
class GlbxMarketDatahandler {
public:
static void run(bool use_file,
std::shared_ptr<IMarketDataListener> listner,
static void run(
std::shared_ptr<::bts::md::IMarketDataListener> listner,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
bts::api::domain::Venue venue);
const GlbxMarketDataConfig& config);
};
} // namespace bts::md::glbx
@@ -0,0 +1,848 @@
#include "TaifexDataGateway.h"
#include "TaifexDataProcessors.h"
#include "lib/api/md/TaifexMarketDataConfig.pb.h"
#include "lib/config/config.h"
#include "lib/config/protos/system_config.pb.h"
#include "lib/logging/logging.h"
#include "lib/md/IMarketDataListener.h"
#include "lib/persistence/IDGenerator.h"
#include "lib/utils/bts_assert.h"
#include "lib/utils/bts_chrono.h"
#include "lib/utils/overloaded.h"
#include "lib/utils/thread_manager.h"
#include <memory>
#include <optional>
#include <ostream>
#include <sstream>
#include <string>
#include <tbb/concurrent_hash_map.h>
#include <tbb/concurrent_queue.h>
#include <thread>
#include <type_traits>
#include <variant>
#include <vector>
namespace bts::md::taifex {
enum class LIVE_DATA_STATES {
START = 0,
WAITING_FOR_REFRESH_DATA,
REFRESH_DATA_DONE,
WAITING_TO_SYNC_LIVE,
LIVE_RUNNING // add state to shift to after hours
};
constexpr size_t MSG_FOOTER_SIZE = 3;
constexpr size_t MSG_HEADER_SIZE = 19;
constexpr size_t FEED_BUFFER_SIZE = 1024;
std::atomic_bool stop_refresh_thread = true;
std::atomic_bool stop_reference_thread = true;
std::atomic_bool stop_live_buffer_thread = true;
std::atomic<LIVE_DATA_STATES> state;
using UNPROCESSED_MSG = std::vector<unsigned char>;
using REFERENCE_MSG =
std::variant<TaifexMessages::I010Data, TaifexMessages::I011Data>;
using LIVE_MSG = std::variant<
TaifexMessages::I081Data,
TaifexMessages::I083Data,
TaifexMessages::I024Data>;
using REFRESH_MSG = TaifexMessages::I083Data;
using SNAPSHOT_BUFFER_MSG =
std::shared_ptr<std::variant<REFERENCE_MSG, REFRESH_MSG>>;
// Refactored handleReferenceDataParsedMessage
void handleReferenceDataParsedMessage(
const REFERENCE_MSG &msg,
const std::shared_ptr<bts::md::IMarketDataListener> &listener,
const std::shared_ptr<persistence::IDGenerator> &sec_id_store,
const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
std::visit(
[&](auto &&msg) {
using T = std::decay_t<decltype(msg)>;
if constexpr (std::is_same_v<T, TaifexMessages::I010Data>) {
TaifexDataProcessors::processI010Data(
msg, listener, sec_id_store, security_details);
} else if constexpr (std::is_same_v<T, TaifexMessages::I011Data>) {
TaifexDataProcessors::processI011Data(msg, security_details);
}
},
msg);
}
std::optional<REFERENCE_MSG> getParsedReferenceMessage(
const UNPROCESSED_MSG &buffer,
size_t position,
const TaifexMessages::Header_Fields &Header) {
switch (Header.message_type.identifier) {
case TaifexMessages::MessageIdentifier::I010:
return TaifexMessages::I010Data(
buffer, position + MSG_HEADER_SIZE,
Header.Body_Length + MSG_FOOTER_SIZE);
case TaifexMessages::MessageIdentifier::I011:
return TaifexMessages::I011Data(
buffer, position + MSG_HEADER_SIZE,
Header.Body_Length + MSG_FOOTER_SIZE);
default:
return std::nullopt;
}
}
void handleReferenceDataUnparsedMessage(
const UNPROCESSED_MSG &buffer,
size_t position,
const TaifexMessages::Header_Fields &Header,
std::shared_ptr<bts::md::IMarketDataListener> listener,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
std::stringstream ss;
ss << Header.message_type.identifier;
auto s = ss.str();
LOG(md, Info, "handleReferenceDataParsedMessage: type:{}", s);
auto parsed = getParsedReferenceMessage(buffer, position, Header);
if (!parsed.has_value())
return;
handleReferenceDataParsedMessage(
parsed.value(), listener, sec_id_store, security_details);
}
void handleLiveDataParsedMessage(
const LIVE_MSG &msg,
std::shared_ptr<bts::md::IMarketDataListener> listener,
std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
std::visit(
[&](auto &&msg) {
using T = std::decay_t<decltype(msg)>;
if constexpr (std::is_same_v<T, TaifexMessages::I024Data>) {
TaifexDataProcessors::processTradeRecords(
msg, listener, security_details);
} else if constexpr (std::is_same_v<T, TaifexMessages::I081Data>) {
TaifexDataProcessors::processBookIncrements(
msg, listener, security_details);
} else if constexpr (std::is_same_v<T, TaifexMessages::I083Data>) {
TaifexDataProcessors::processBookSnapShotMessage(
msg, listener, security_details);
}
},
msg);
}
std::optional<LIVE_MSG> getParsedLiveDataMessage(
const UNPROCESSED_MSG &buffer,
size_t position,
const TaifexMessages::Header_Fields &Header) {
switch (Header.message_type.identifier) {
case TaifexMessages::MessageIdentifier::I024:
return TaifexMessages::I024Data(
buffer, position + MSG_HEADER_SIZE,
Header.Body_Length + MSG_FOOTER_SIZE);
case TaifexMessages::MessageIdentifier::I083:
return TaifexMessages::I083Data(
buffer, position + MSG_HEADER_SIZE,
Header.Body_Length + MSG_FOOTER_SIZE);
case TaifexMessages::MessageIdentifier::I081:
return TaifexMessages::I081Data(buffer, position + MSG_HEADER_SIZE);
default:
return std::nullopt;
}
}
void handleLiveDataUnparsedMessage(
const UNPROCESSED_MSG &buffer,
size_t position,
const TaifexMessages::Header_Fields &Header,
std::shared_ptr<bts::md::IMarketDataListener> listener,
std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
auto parsed = getParsedLiveDataMessage(buffer, position, Header);
if (!parsed.has_value())
return;
handleLiveDataParsedMessage(parsed.value(), listener, security_details);
}
std::shared_ptr<TaifexMessages::I084Data> handleRefreshDataUnparsedMessage(
const UNPROCESSED_MSG &buffer,
size_t position,
const TaifexMessages::Header_Fields &Header) {
switch (Header.message_type.identifier) {
case TaifexMessages::MessageIdentifier::I084:
return std::make_shared<TaifexMessages::I084Data>(
TaifexMessages::I084Data(buffer, position));
//handle this
break;
default:
bts_assert_msg(false, "NOT IN REFRESH DATA");
}
bts_assert_msg(false, "Control should not reach here");
return std::make_shared<TaifexMessages::I084Data>(
TaifexMessages::I084Data(buffer, position));
;
}
void HandleRefreshFeed(
int socket,
unsigned long seq_num,
tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG> &snapshot_buffered_messages,
unsigned long &refresh_seq_num) {
LOG(md, Info, "HandleRefreshFeed looking for {}", seq_num);
enum {
WAITING_FOR_A = 0,
WAITING_FOR_Z,
} refresh_state = WAITING_FOR_A;
size_t counter = 0;
while (!stop_refresh_thread && !stop_live_buffer_thread) {
UNPROCESSED_MSG buffer(FEED_BUFFER_SIZE);
std::string desc = "Refresh Socket";
size_t bytes_received =
TaifexUtils::recv_socket_data(socket, buffer, FEED_BUFFER_SIZE, desc);
if (bytes_received <= 0) {
continue;
}
auto Header = TaifexMessages::Header_Fields(buffer, 0);
switch (refresh_state) {
case WAITING_FOR_A: {
auto i084_message =
TaifexMessages::I084Data(buffer, MSG_HEADER_SIZE, true);
LOG(md, Info, "HandleRefreshFeed WAITING_FOR_A looking for {}, got {}",
seq_num, i084_message.last_seq);
if (i084_message.message_type == TaifexMessages::I084Data::A &&
i084_message.last_seq > seq_num) {
refresh_state = WAITING_FOR_Z;
}
break;
}
case WAITING_FOR_Z: {
LOG(md, Info, "HandleRefreshFeed WAITING_FOR_Z");
auto i084_message = TaifexMessages::I084Data(buffer, MSG_HEADER_SIZE);
// bts_assert(i084_message.last_seq > seq_num);
if (i084_message.message_type == TaifexMessages::I084Data::O) {
for (auto m : i084_message.data) {
auto mes =
std::make_shared<std::variant<REFERENCE_MSG, REFRESH_MSG>>(
m);
if (std::holds_alternative<REFERENCE_MSG>(*mes)) {
LOG(md, Error, "Adding REFer in refresh");
} else if (std::holds_alternative<REFRESH_MSG>(*mes)) {
LOG(md, Info, "Adding REFresh in refresh");
} else {
LOG(md, Error,
"Unexpected message type in variant in refresh");
}
std::vector<std::string> contracts = {
"TXF", "MXF", "TMF", "CDF", "QFF"};
bool found = false;
for (const auto &contract : contracts) {
if (m.prod_id.find(contract) != std::string::npos &&
m.prod_id.find("/") == std::string::npos) {
found = true;
break;
}
}
if (found) {
snapshot_buffered_messages.push(mes);
}
counter++;
}
} else if (i084_message.message_type == TaifexMessages::I084Data::Z) {
LOG(md, Info, "HandleRefreshFeed Z found total secs {}", counter);
refresh_seq_num = i084_message.last_seq;
stop_refresh_thread = true;
return;
}
break;
}
}
}
}
void truncateFile(const std::string &filename, size_t validSize) {
// Truncate the file to remove invalid data
std::fstream file(filename, std::ios::in | std::ios::out | std::ios::binary);
if (file) {
file.seekp(validSize); // Set the position where truncation should happen
file.close();
std::cout << "File truncated to " << validSize << " bytes." << std::endl;
} else {
std::cerr << "Failed to truncate file: " << filename << std::endl;
}
}
bool LoadReferenceFromFile(
const std::string &file_name,
tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG> &snapshot_buffered_messages,
std::set<std::string> I010_seen,
std::set<std::string> I011_seen) {
const size_t BUFFER_SIZE = 1024;
std::ifstream file(file_name, std::ios::binary);
if (!file) {
LOG(md, Error, "File not found: {}", file_name);
return false;
}
UNPROCESSED_MSG buffer(
BUFFER_SIZE + 1024); // Extra space for possible leftovers
UNPROCESSED_MSG prev_data(1024); // Stores leftover data
unsigned long total_read = 0;
bool start = true;
size_t prev_data_len = 0; // Length of leftover data
while (file.read(
reinterpret_cast<char *>(buffer.data() + prev_data_len), BUFFER_SIZE)) {
size_t bytes_read = file.gcount();
if (bytes_read < BUFFER_SIZE) {
break; // No more data to read
}
total_read += bytes_read;
if (!start) {
// Move data from prev_data to the start of the buffer
std::memcpy(buffer.data(), prev_data.data(), prev_data_len);
}
size_t position = 0;
while (position < (BUFFER_SIZE + prev_data_len - MSG_HEADER_SIZE)) {
auto Header = TaifexMessages::Header_Fields(buffer, position);
size_t message_length =
Header.Body_Length + MSG_HEADER_SIZE + MSG_FOOTER_SIZE;
// Check if the full message fits within the buffer
if ((BUFFER_SIZE + prev_data_len - position) >= message_length) {
// Handle the parsed message
auto msg = getParsedReferenceMessage(buffer, position, Header);
if (!msg.has_value())
continue;
bool seen = true;
std::visit(
overloaded{
[&](const TaifexMessages::I010Data &i010) {
if (!I010_seen.contains(i010.prod_id)) {
seen = false;
I010_seen.insert(i010.prod_id);
}
},
[&](const TaifexMessages::I011Data &i011) {
if (!I011_seen.contains(i011.kind_id)) {
seen = false;
I011_seen.insert(i011.kind_id);
}
}},
msg.value());
if (!seen) {
auto mes =
std::make_shared<std::variant<REFERENCE_MSG, REFRESH_MSG>>(
msg.value());
snapshot_buffered_messages.push(mes);
}
position += message_length;
} else {
// Incomplete message, break and handle leftover data
break;
}
}
// Move remaining (incomplete) data from buffer into prev_data and set prev_data_len
prev_data_len = BUFFER_SIZE + prev_data_len - position;
if (prev_data_len > 0) {
std::memcpy(prev_data.data(), buffer.data() + position, prev_data_len);
}
start = false;
}
truncateFile(file_name, total_read);
return true;
}
void writeBufferWithSeparator(
const std::vector<unsigned char> &buffer,
size_t bytes_received,
unsigned long position = 0) {
auto filename = config::get<bts::config::SystemConfig>().data_path() +
"taifex_default_reference";
std::ofstream outputFile(
filename, std::ios::binary | std::ios::app); // Open file in append mode
if (!outputFile) {
std::cerr << "Failed to open file for writing: " << filename << std::endl;
return;
}
// Write the received bytes from the buffer to the file
outputFile.write(
reinterpret_cast<const char *>(buffer.data() + position),
bytes_received);
// Write the separator after the message
// outputFile.write(MESSAGE_SEPARATOR.c_str(), MESSAGE_SEPARATOR.size());
if (!outputFile) {
std::cerr << "Failed to write to file." << std::endl;
} else {
std::cout << "Successfully wrote " << bytes_received
<< " bytes with separator to " << filename << std::endl;
}
}
void HandleReferenceFeed(
int socket,
tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG> &snapshot_buffered_messages) {
std::set<std::string> I010_seen;
std::set<std::string> I011_seen;
LoadReferenceFromFile(
config::get<bts::config::SystemConfig>().data_path() +
"taifex_default_reference",
snapshot_buffered_messages, I010_seen, I011_seen);
// bool file_done =
UNPROCESSED_MSG buffer(FEED_BUFFER_SIZE);
LOG(md, Info, "HandleReferenceFeed");
std::string desc = "Reference socket";
while (!stop_reference_thread) {
// LOG(md, Info, "HandleReferenceFeed: Inside While");
uint bytes_received =
TaifexUtils::recv_socket_data(socket, buffer, FEED_BUFFER_SIZE, desc);
// LOG(md, Info, "HandleReferenceFeed: Inside While {}th pakcket recved",counter++);
if (bytes_received <= 0) {
// LOG(md, Info, "HandleReferenceFeed: Inside While {}th pakcket not valid",counter-1);
continue;
}
auto Header = TaifexMessages::Header_Fields(buffer, 0);
size_t message_length =
Header.Body_Length + MSG_HEADER_SIZE + MSG_FOOTER_SIZE;
if (bytes_received != message_length) {
LOG(md, Error, "Reference data recv: {}, should be:{}", bytes_received,
message_length);
} else {
auto msg = getParsedReferenceMessage(buffer, 0, Header);
if (!msg.has_value())
continue;
bool seen = true;
std::visit(
overloaded{
[&](const TaifexMessages::I010Data &i010) {
if (!I010_seen.contains(i010.prod_id)) {
seen = false;
I010_seen.insert(i010.prod_id);
}
},
[&](const TaifexMessages::I011Data &i011) {
if (!I011_seen.contains(i011.kind_id)) {
seen = false;
I011_seen.insert(i011.kind_id);
}
}},
msg.value());
if (!seen) {
auto mes =
std::make_shared<std::variant<REFERENCE_MSG, REFRESH_MSG>>(
msg.value());
writeBufferWithSeparator(buffer, bytes_received);
snapshot_buffered_messages.push(mes);
}
// if (std::holds_alternative<REFERENCE_MSG>(*mes)) {
// LOG(md, Error, "Adding REFer from refrence data");
// } else if (std::holds_alternative<REFRESH_MSG>(*mes)) {
// LOG(md, Info, "Adding REFresh from refrence data");
// } else {
// LOG(md, Error,
// "Unexpected message type in variant from refrence data");
// }
// snapshot_buffered_messages.push(mes);
}
}
// 6. Close the socket after done processing
close(socket);
}
void HandleLiveBufferFeed(
int socket,
tbb::concurrent_queue<UNPROCESSED_MSG> &live_buffered_messages,
std::string desc) {
int bytes_received = -1;
while (!stop_live_buffer_thread) {
UNPROCESSED_MSG b(FEED_BUFFER_SIZE);
// auto &b = live_buffered_messages.emplace();
bytes_received =
TaifexUtils::recv_socket_data(socket, b, FEED_BUFFER_SIZE, desc);
if (bytes_received <= 0) {
LOG(md, Error, "HandleLiveBufferFeed: recv failed resseting state");
state = LIVE_DATA_STATES::START;
break;
}
if (bytes_received > 0) {
b.resize(bytes_received);
live_buffered_messages.push(
std::move(b)); // have to use this instead of emplace
}
}
stop_live_buffer_thread = true;
}
bool runOnBinaryFile(
const std::string &file_name,
std::shared_ptr<bts::md::IMarketDataListener> listener,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
std::shared_ptr<TaifexSecurityDetailsMapper> security_details,
TaifexDataGateway::DATA_TYPE data_type) {
const size_t BUFFER_SIZE = 1024 * 1024;
std::ifstream file(file_name, std::ios::binary);
if (!file) {
LOG(md, Error, "File not found: {}", file_name);
bts_assert(false);
}
UNPROCESSED_MSG buffer(
BUFFER_SIZE + 1024); // Extra space for possible leftovers
UNPROCESSED_MSG prev_data(1024); // Stores leftover data
bool start = true;
size_t prev_data_len = 0; // Length of leftover data
while (file.read(
reinterpret_cast<char *>(buffer.data() + prev_data_len), BUFFER_SIZE)) {
size_t bytes_read = file.gcount();
if (bytes_read < BUFFER_SIZE) {
break; // No more data to read
}
if (!start) {
// Move data from prev_data to the start of the buffer
std::memcpy(buffer.data(), prev_data.data(), prev_data_len);
}
size_t position = 0;
while (position < (BUFFER_SIZE + prev_data_len - MSG_HEADER_SIZE)) {
auto Header = TaifexMessages::Header_Fields(buffer, position);
size_t message_length =
Header.Body_Length + MSG_HEADER_SIZE + MSG_FOOTER_SIZE;
// Check if the full message fits within the buffer
if ((BUFFER_SIZE + prev_data_len - position) >= message_length) {
// Handle the parsed message
switch (data_type) {
case TaifexDataGateway::DATA_TYPE::LIVE:
handleLiveDataUnparsedMessage(
buffer, position, Header, listener, security_details);
break;
case TaifexDataGateway::DATA_TYPE::REFERENCE:
handleReferenceDataUnparsedMessage(
buffer, position, Header, listener, sec_id_store,
security_details);
break;
case TaifexDataGateway::DATA_TYPE::REFRESH:
handleRefreshDataUnparsedMessage(buffer, position, Header);
break;
default:
bts_assert(false);
}
position += message_length;
} else {
// Incomplete message, break and handle leftover data
break;
}
}
// Move remaining (incomplete) data from buffer into prev_data and set prev_data_len
prev_data_len = BUFFER_SIZE + prev_data_len - position;
if (prev_data_len > 0) {
std::memcpy(prev_data.data(), buffer.data() + position, prev_data_len);
}
start = false; // Only move leftover data on subsequent iterations
}
return true;
}
void runLive(
std::shared_ptr<IMarketDataListener> listener,
std::shared_ptr<persistence::IDGenerator> sec_id_store,
std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
long reset_counter = -1;
state = LIVE_DATA_STATES::START;
std::set<int> sockets;
unsigned long last_seq_num;
std::unordered_map<std::string, REFRESH_MSG> products_waiting;
// auto print =
// [](std::unordered_map<std::string, REFRESH_MSG> &products_waiting) {
// std::string s = "\n";
// for (auto &pair : products_waiting) {
// s = s + "Key: " + pair.first + ", value: '" + pair.second.prod_id +
// "'\n";
// }
// return s;
// };
tbb::concurrent_queue<SNAPSHOT_BUFFER_MSG>
snapshot_buffered_messages; // Shared concurrent queue
tbb::concurrent_queue<UNPROCESSED_MSG>
live_buffered_messages; // Shared concurrent queue
// Load from config
std::string live_data_ip = "225.0.140.140";
const int live_data_port = 14000;
std::string futures_market_data_ip =
"192.168.201.94"; // Replace with actual IP
std::string reference_data_ip = "225.0.140.141"; // Placeholder
const int reference_data_port = 14100; // Placeholder
std::string refresh_data_ip = "225.0.140.147"; // Placeholder
const int refresh_data_port = 14700; // Placeholder
// if (use_after_hours) {
// live_data_ip = "225.10.140.140";
// reference_data_ip = "225.10.140.141"; // Placeholder
// refresh_data_ip = "225.10.140.147"; // Placeholder
// }
std::vector<TaifexMessages::I083Data> refresh_messages;
int live_socket = -1;
int reference_socket = -1;
int last_channel_id = -1;
while (true) {
switch (state) {
case LIVE_DATA_STATES::START: {
live_data_ip = TaifexUtils::getIP("realtime");
reference_data_ip = TaifexUtils::getIP("reference");
refresh_data_ip = TaifexUtils::getIP("refresh");
reset_counter++;
LOG(md, Error, "State: START counter:{}", reset_counter);
stop_reference_thread = stop_refresh_thread = stop_live_buffer_thread =
true;
security_details->reset();
live_buffered_messages.clear();
TaifexUtils::close_sockets(sockets);
last_seq_num = 0;
bts_assert(reset_counter < 5);
state = LIVE_DATA_STATES::WAITING_FOR_REFRESH_DATA;
break;
}
case LIVE_DATA_STATES::WAITING_FOR_REFRESH_DATA: {
LOG(md, Info, "State: WAITING_FOR_REFRESH_DATA");
reference_socket = TaifexUtils::create_multicast_socket_assert(
reference_data_ip, reference_data_port, futures_market_data_ip);
sockets.insert(reference_socket);
stop_reference_thread = false;
bts::threading::make_thread("Taifex_reference_data", [&]() {
HandleReferenceFeed(
reference_socket, std::ref(snapshot_buffered_messages));
}).detach();
LOG(md, Info,
"State: WAITING_FOR_REFRESH_DATA Reference thread started");
std::string desc = "WAITING_FOR_REFRESH_DATA Live data";
live_socket = TaifexUtils::create_multicast_socket_assert(
live_data_ip, live_data_port, futures_market_data_ip);
sockets.insert(live_socket);
UNPROCESSED_MSG buffer(FEED_BUFFER_SIZE);
size_t bytes_received = TaifexUtils::recv_socket_data(
live_socket, buffer, FEED_BUFFER_SIZE, desc);
if (bytes_received <= 0) {
LOG(md, Error,
"LIVE_DATA_STATES::WAITING_FOR_REFRESH_DATA recv failed "
"resseting state");
state = LIVE_DATA_STATES::START;
break;
}
auto s = TaifexUtils::getBufferString(buffer);
LOG(md, Info,
"State: WAITING_FOR_REFRESH_DATA first live message recieved: {}",
s);
TaifexMessages::Header_Fields Header =
TaifexMessages::Header_Fields(buffer, 0);
if (!Header.valid) {
LOG(md, Error, "Header not valid");
state = LIVE_DATA_STATES::START;
break;
}
last_seq_num = Header.Channel_Seq;
stop_refresh_thread = false;
stop_live_buffer_thread = false;
bts::threading::make_thread("Taifex_live_buffer", [&]() {
HandleLiveBufferFeed(
live_socket, live_buffered_messages, "Taifex Live Buffer");
}).detach();
LOG(md, Info,
"State: WAITING_FOR_REFRESH_DATA refresh thread started");
auto refresh_socket = TaifexUtils::create_multicast_socket_assert(
refresh_data_ip, refresh_data_port, futures_market_data_ip);
sockets.insert(refresh_socket);
unsigned long refresh_seq_num = 0;
HandleRefreshFeed(
refresh_socket, last_seq_num, snapshot_buffered_messages,
refresh_seq_num);
last_seq_num = refresh_seq_num;
if (state != LIVE_DATA_STATES::START)
state = LIVE_DATA_STATES::REFRESH_DATA_DONE;
break;
}
case LIVE_DATA_STATES::REFRESH_DATA_DONE: {
LOG(md, Info,
"State: REFRESH_DATA_DONE snapshot_buffered_messages size{}",
snapshot_buffered_messages.unsafe_size());
bts_assert(stop_refresh_thread);
while (!snapshot_buffered_messages.empty() ||
!products_waiting.empty()) {
SNAPSHOT_BUFFER_MSG msg;
if (!snapshot_buffered_messages.try_pop(msg)) {
continue;
}
// LOG(md, Info,
// "State: REFRESH_DATA_DONE Waiting for Reference data products "
// "waiting:{}, buffer_size:{} \n waiting: {}\n sec_deets: {}",
// products_waiting.size(),
// snapshot_buffered_messages.unsafe_size(),
// print(products_waiting), security_details->to_string());
std::visit(
overloaded{
[&](const REFERENCE_MSG &ref_msg) {
handleReferenceDataParsedMessage(
ref_msg, listener, sec_id_store, security_details);
std::visit(
overloaded{
[&](const TaifexMessages::I010Data &data) {
// std::string s = "looking for: '"+TaifexUtils::remove_spaces(data.prod_id)+"', in"+print(products_waiting);
// LOG(md, Error, "{}",s);
if (products_waiting.contains(
TaifexUtils::remove_spaces(
data.prod_id))) {
auto m = products_waiting.at(
TaifexUtils::remove_spaces(
data.prod_id));
SNAPSHOT_BUFFER_MSG mes =
std::make_shared<std::variant<
REFERENCE_MSG, REFRESH_MSG>>(m);
snapshot_buffered_messages.push(mes);
products_waiting.erase(data.prod_id);
}
},
[&](const TaifexMessages::I011Data) {
return; // Exit early for I011Data
}},
ref_msg);
return;
},
[&](const REFRESH_MSG &refresh_msg) {
if (!TaifexDataProcessors::processBookSnapShotMessage(
refresh_msg, listener, security_details)) {
products_waiting.insert_or_assign(
TaifexUtils::remove_spaces(refresh_msg.prod_id),
refresh_msg);
}
return;
}},
*msg);
}
stop_reference_thread = true;
state = LIVE_DATA_STATES::WAITING_TO_SYNC_LIVE;
break;
// stop reference data
// apply all refresh data
}
case LIVE_DATA_STATES::WAITING_TO_SYNC_LIVE: {
LOG(md, Info, "State: WAITING_TO_SYNC_LIVE");
while (true) {
// Process messages from the concurrent queue
UNPROCESSED_MSG buffer;
live_buffered_messages.try_pop(buffer);
if (buffer.empty()) {
continue;
}
TaifexMessages::Header_Fields Header(buffer, 0);
if (!Header.valid) {
LOG(md, Error, "Header not valid");
state = LIVE_DATA_STATES::START;
break;
}
if (Header.Channel_Seq < last_seq_num) {
continue; // Skip messages with outdated sequence numbers
} else {
break;
}
// If no message was available to pop, break the outer while loop
if (live_buffered_messages.empty()) {
break;
}
}
state = LIVE_DATA_STATES::LIVE_RUNNING;
break;
}
// while 1
// process buffer:
// if seq_num > last_seq_num + 1
// if seq_num <= last_seq_num
// skip message
// state = LIVE_RUNNING;
// break
case LIVE_DATA_STATES::LIVE_RUNNING: {
// if seq_num != last_seq_num + 1
LOG(md, Info, "STATE: LIVE RUNNING !!");
while (state == LIVE_DATA_STATES::LIVE_RUNNING) {
UNPROCESSED_MSG buffer;
live_buffered_messages.try_pop(buffer);
if (buffer.empty()) {
continue;
}
TaifexMessages::Header_Fields Header =
TaifexMessages::Header_Fields(buffer, 0);
if (!Header.valid) {
LOG(md, Error, "Header not valid");
state = LIVE_DATA_STATES::START;
break;
}
if (Header.Channel_Seq != last_seq_num + 1) {
LOG(md, Error, "Header.Channel_Seq != last_seq_num+1");
LOG(md, Info, "header.channel_seq {}", Header.Channel_Seq);
LOG(md, Info, "header.channel_id {}", Header.Channel_Id);
LOG(md, Info, "last_seq_number {}", last_seq_num);
LOG(md, Info, "last channel id {}", last_channel_id);
state = LIVE_DATA_STATES::START;
break;
}
last_seq_num = Header.Channel_Seq;
last_channel_id = Header.Channel_Id;
handleLiveDataUnparsedMessage(
buffer, 0, Header, listener, security_details);
}
break;
}
// else
// process live
default:
bts_assert(false);
};
}
}
//const std::string &filename,
void bts::md::taifex::TaifexDataGateway::run(
const api::md::TaifexMarketDataConfig &config,
std::shared_ptr<IMarketDataListener> listner,
std::shared_ptr<persistence::IDGenerator> sec_id_store) {
std::shared_ptr<TaifexSecurityDetailsMapper> sec_details =
std::make_shared<TaifexSecurityDetailsMapper>();
if (config.read_from_file()) {
bts_assert(!config.ref_data_file_name().empty());
bts_assert(!config.realtime_data_file_name().empty());
std::string reference_file_name =
config.ref_data_file_name(); //"reference_data_20240806_1";
std::string realtime_file_name =
config.realtime_data_file_name(); //"real_time_data_20240806_274";
std::string base_path =
config::get<bts::config::SystemConfig>().data_path();
runOnBinaryFile(
base_path + reference_file_name, listner, sec_id_store, sec_details,
REFERENCE);
runOnBinaryFile(
base_path + realtime_file_name, listner, sec_id_store, sec_details,
LIVE);
while (1) { //confirm with taha
sleep_for(10s);
}
} else {
runLive(
listner, sec_id_store,
sec_details); //regular_hours_connection_details,after_hours_connection_details);
}
}
} //namespace bts::md::taifex
@@ -1,5 +1,6 @@
#pragma once
#include "lib/api/md/TaifexMarketDataConfig.pb.h"
#include <memory>
namespace bts::persistence {
@@ -12,10 +13,15 @@ class IMarketDataListener;
namespace bts::md::taifex {
// const std::string &filename,
class TaifexDatahandler {
class TaifexDataGateway {
public:
enum DATA_TYPE{
LIVE=0,
REFERENCE,
REFRESH
};
static void run(
bool use_file,
const api::md::TaifexMarketDataConfig &config,
std::shared_ptr<IMarketDataListener> listner,
std::shared_ptr<persistence::IDGenerator> sec_id_store);
};
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,232 @@
#pragma once
#include "TaifexMessages.h"
#include "TaifexSecurityDetailsMapper.h"
#include "lib/domain/Security.h"
#include "lib/domain/TradeRecord.h"
#include "lib/md/IMarketDataListener.h"
#include "lib/persistence/IDGenerator.h"
#include "lib/utils/bts_assert.h"
namespace bts::md::taifex {
class TaifexDataProcessors {
public:
static bool processBookSnapShotMessage(
const TaifexMessages::I083Data &i083_message,
std::shared_ptr<bts::md::IMarketDataListener> listener,
std::shared_ptr<TaifexSecurityDetailsMapper> security_details) {
auto [sec_id, dec_loc] = security_details->getSecurityDetails(
TaifexUtils::remove_spaces(i083_message.prod_id));
if (!sec_id || !dec_loc)
return false;
domain::Book book;
auto &bids = book.bids();
auto &asks = book.asks();
book.set_security_id(sec_id.value());
for (const auto &entry : i083_message.entries) {
bts::domain::Price price(
entry.price *
pow(10,
9 - dec_loc
.value())); // Assuming the price needs to be adjusted by a factor of -2
bts::domain::Quantity quantity(entry.quantity);
// Process each MD_ENTRY
if (entry.md_entry_type == '0') { // Buy or derived buy
// Update the corresponding bid level
if (entry.level <= bids.size()) {
bids[entry.level - 1].price = price;
bids[entry.level - 1].total_quantity = quantity;
}
} else if (
entry.md_entry_type == '1') { // Sell or derived sell
// Update the corresponding ask level
if (entry.level <= asks.size()) {
asks[entry.level - 1].price = price;
asks[entry.level - 1].total_quantity = quantity;
}
}
}
std::stringstream book_after;
book_after << "Book after snapshot update:\n";
book_after << "Bids:\n";
for (const auto &level : bids) {
book_after << "Price: " << level.price.to_double()
<< ", Quantity: " << level.total_quantity.value() << "\n";
}
book_after << "Asks:\n";
for (const auto &level : asks) {
book_after << "Price: " << level.price.to_double()
<< ", Quantity: " << level.total_quantity.value() << "\n";
}
book_after<<"MSG:\n"<<i083_message;
std::string book_snapshot_details = book_after.str() + "\n";
bts_assert_msgf(
book.is_sorted(), "book not sorted {}", book_snapshot_details);
listener->on_book(book);
return true;
}
static void processBookIncrements(
const TaifexMessages::I081Data &msg,
const std::shared_ptr<bts::md::IMarketDataListener> &listener,
const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
auto [sec_id, dec_loc] = security_details->getSecurityDetails(
TaifexUtils::remove_spaces(msg.prod_id));
if (!sec_id || !dec_loc)
return;
for (const auto &entry : msg.entries) {
IMarketDataListener::BookOp inc;
inc.id = *sec_id;
inc.price =
bts::domain::Price(entry.price * std::pow(10, 9 - *dec_loc));
inc.quantity = bts::domain::Quantity(entry.quantity);
inc.side = (entry.md_entry_type == '0' || entry.md_entry_type == 'E')
? bts::domain::Side::Buy
: bts::domain::Side::Sell;
switch (entry.md_update_action) {
case '0':
inc.op_type = IMarketDataListener::BookOpType::Add;
break;
case '1':
inc.op_type = IMarketDataListener::BookOpType::MODIFY;
break;
case '2':
inc.op_type = IMarketDataListener::BookOpType::REMOVE;
break;
default:
continue;
}
listener->on_book_increment(inc);
}
}
// Refactored trade processing function
static void processTradeRecords(
const TaifexMessages::I024Data &msg,
const std::shared_ptr<bts::md::IMarketDataListener> &listener,
const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
auto [sec_id, dec_loc] = security_details->getSecurityDetails(
TaifexUtils::remove_spaces(msg.prod_id));
if (!sec_id || !dec_loc)
return;
std::vector<bts::domain::TradeRecord> trades;
for (const auto &trade_record : msg.trades) {
auto &record = trades.emplace_back();
record.sec_id = *sec_id;
record.time_stamp = bts::core::TimeStamp::from_iso_string(
TaifexUtils::convertToISO(msg.match_time, "datetime_no_timezone"));
record.trade_quantity = bts::domain::Quantity(trade_record.match_qty);
record.trade_price = bts::domain::Price(
trade_record.match_price * std::pow(10, 9 - *dec_loc));
// TODO: Set the aggressor side (record.agg_side)
}
std::vector<const domain::TradeRecord *> trade_ptrs;
trade_ptrs.reserve(trades.size());
for (const auto &trade : trades) {
trade_ptrs.push_back(&trade);
}
// for (const auto *trade_ptr : trade_ptrs) {
// if (trade_ptr != nullptr) {
// LOG(md, Error, "Trade Quantity: {}, Trade Price: {}",
// trade_ptr->trade_quantity.value(),
// trade_ptr->trade_price);
// }
// }
listener->on_trades(trade_ptrs);
}
static bts::domain::Security initializeSecurity(
const std::string &id,
const bts::domain::SecurityID &sec_id,
const std::string &symbol,
const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
bts::domain::Security sec;
sec.security_common.venue = bts::api::domain::VENUE_TAIFEX;
sec.security_common.id = sec_id;
sec.security_common.exchange_sec_id = id;
sec.security_common.desc = id;
sec.security_common.symbol = symbol;
sec.security_common.asset = id;
sec.security_common.group = id;
sec.security_common.min_price_increment =
security_details->get_min_inc(id);
sec.security_common.min_quantity = security_details->get_min_inc(id);
sec.security_common.min_quantity_increment =
security_details->get_min_inc(id);
sec.security_common.currency = security_details->get_currency(id);
return sec;
}
// Process I010Data
static void processI010Data(
const TaifexMessages::I010Data &msg,
const std::shared_ptr<bts::md::IMarketDataListener> &listener,
const std::shared_ptr<persistence::IDGenerator> &sec_id_store,
const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
const auto id = TaifexUtils::remove_spaces(msg.prod_id);
const auto symbol = id.substr(0, 3);
bts::domain::SecurityID sec_id{
sec_id_store->get_or_add(api::domain::VENUE_TAIFEX, id)};
auto check =
security_details->add_i010_mapping(id, sec_id, msg.decimal_locator);
if (!check) {
return;
}
// Initialize the security object
bts::domain::Security sec =
initializeSecurity(id, sec_id, symbol, security_details);
// Handle product-specific details
switch (msg.parsedProductInfo.productType) {
case TaifexMessages::I010Data::ProductInfo::ProductType::FUTURES: {
auto &futureSec = sec.security.emplace<domain::FutureSecurity>();
std::tm tm = {};
std::istringstream iss(msg.end_date);
iss >> std::get_time(&tm, "%Y%m%d");
std::time_t time = std::mktime(&tm);
futureSec.expiration =
bts::core::TimeStamp(static_cast<uint64_t>(time));
break;
}
case TaifexMessages::I010Data::ProductInfo::ProductType::OPTIONS: {
auto &optionSec = sec.security.emplace<domain::OptionSecurity>();
std::tm tm = {};
std::istringstream iss(msg.end_date);
iss >> std::get_time(&tm, "%Y%m%d");
std::time_t time = std::mktime(&tm);
optionSec.expiration =
bts::core::TimeStamp(static_cast<uint64_t>(time));
optionSec.strike =
domain::Price::from_double(msg.parsedProductInfo.strikePrice);
break;
}
}
std::vector<const domain::Security *> const_security_list;
const_security_list.push_back(&sec);
listener->on_securities(const_security_list);
}
// Process I011Data
static void processI011Data(
const TaifexMessages::I011Data &msg,
const std::shared_ptr<TaifexSecurityDetailsMapper> &security_details) {
const auto id = TaifexUtils::remove_spaces(msg.kind_id);
security_details->add_i011_mapping(
id, msg.contract_size, msg.getCurrencyString());
}
};
} //namespace bts::md::taifex
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,140 @@
#pragma once
#include "lib/domain/SecurityID.h"
#include "lib/logging/logging.h"
#include <optional>
#include <sstream>
#include <string>
namespace bts::md::taifex {
class TaifexSecurityDetailsMapper {
struct product_details {
std::string instrument_id; // use this as
bts::domain::SecurityID sec_id;//uint64_t
uint dec_loc;
};
struct contract_details{
std::unordered_map<std::string,product_details> products;
std::string kind_id;
double min_increment;
std::string currency;
};
public:
static std::string get_kind_id(const std::string &prod_id ){
return prod_id.substr(0,3);
}
// Add a mapping from instrument_id to sec_id
bool add_i010_mapping(
std::string instrument_id,
const bts::domain::SecurityID sec_id,
uint decimal_locator) {
auto kind_id = get_kind_id(instrument_id);
if(!sec_deets_map.contains(kind_id)){
return false;
}
auto &contract = sec_deets_map[kind_id];
auto &details = contract.products[instrument_id];
details.sec_id = sec_id;
details.dec_loc = decimal_locator;
details.instrument_id = instrument_id;
auto s = this->to_string();
LOG(md, Info, "add_i010_mapping: SEC_ID STORE= {}",s);
return true;
}
// Add a mapping for i011
void add_i011_mapping(
std::string kind_id, double min_increment, std::string currency) {
auto &contract = sec_deets_map[kind_id];
contract.currency= currency!=""?currency:"usd";
contract.kind_id=kind_id;
contract.min_increment=min_increment;
}
// Find sec_id by instrument_id
std::optional<bts::domain::SecurityID> get_sec_id(
std::string instrument_id) const {
auto kind_id = get_kind_id(instrument_id);
auto it = sec_deets_map.find(kind_id);
if (it != sec_deets_map.end()) {
auto &contract = it->second;
auto it2 = contract.products.find(instrument_id);
if (it2 != contract.products.end())
return it2->second.sec_id;
}
return std::nullopt;
}
// Find dec_loc by instrument_id
std::optional<uint> get_dec_loc(std::string instrument_id) const {
auto kind_id = get_kind_id(instrument_id);
auto it = sec_deets_map.find(kind_id);
if (it != sec_deets_map.end()) {
auto &contract = it->second;
auto it2 = contract.products.find(instrument_id);
if (it2 != contract.products.end())
return it2->second.dec_loc;
}
return std::nullopt;
}
std::string get_currency(std::string instrument_id) const {
auto kind_id = get_kind_id(instrument_id);
auto it = sec_deets_map.find(kind_id);
if (it != sec_deets_map.end()) {
return it->second.currency;
}
return "usd";
}
double get_min_inc(std::string instrument_id) const {
auto kind_id = get_kind_id(instrument_id);
auto it = sec_deets_map.find(kind_id);
if (it != sec_deets_map.end()) {
return it->second.min_increment;
}
return 1;
}
std::pair<std::optional<bts::domain::SecurityID>, std::optional<int>> getSecurityDetails(
const std::string &prod_id) {
auto sec_id = get_sec_id(prod_id);
auto dec_loc = get_dec_loc(prod_id);
if (!sec_id.has_value() || !dec_loc.has_value()) {
if (sec_id.has_value() != dec_loc.has_value())
LOG(md, Error, "Mismatch in sec_id: {}, dec_loc: {}", sec_id.has_value(), dec_loc.has_value());
return {std::nullopt, std::nullopt};
}
return {sec_id, dec_loc};
}
// to_string function to print all data in both maps
std::string to_string() const {
std::stringstream ss;
for (const auto &[kind_id, contract] : sec_deets_map) {
ss << "Kind ID: '" << kind_id << "'\n"
<< " Currency: '" << contract.currency << "'\n"
<< " Min Increment: " << contract.min_increment << "\n";
for (const auto &[instrument_id, details] : contract.products) {
ss << " Instrument ID: '" << instrument_id << "'\n"
<< " Sec ID: " << details.sec_id << "\n"
<< " Dec Loc: " << details.dec_loc << "\n";
}
}
return ss.str();
}
// Reset function to clear the data map
void reset() {
for(auto &contract:sec_deets_map){
contract.second.products.clear();
}
sec_deets_map.clear();
}
private:
std::unordered_map<std::string, contract_details> sec_deets_map;
};
} // namespace bts::md::taifex
+508
View File
@@ -0,0 +1,508 @@
#include "lib/logging/logging.h"
#include "lib/utils/bts_assert.h"
#include <boost/date_time/posix_time/ptime.hpp>
#include <boost/date_time/posix_time/time_parsers.hpp>
#include <chrono>
#include <cstddef>
#include <cstdio>
#include <limits>
#include <netinet/in.h>
#include <string>
#include <vector>
namespace bts::md::taifex {
class TaifexUtils {
using UNPROCESSED_MSG = std::vector<unsigned char>;
public:
static int recv_socket_data(int socket, UNPROCESSED_MSG& buffer, const size_t FEED_BUFFER_SIZE, const std::string & socket_desc ="") {
int bytes_received = -1;
try {
bytes_received = recvfrom(socket, buffer.data(), FEED_BUFFER_SIZE, 0, nullptr, nullptr);
} catch (...) {
std::string s = strerror(errno);
LOG(md, Error, "{}: Error receiving data from socket, {}",socket_desc, s);
}
if (bytes_received < 0) {
std::string s = strerror(errno);
LOG(md, Error, "{}: Socket error: {}",socket_desc, s);
} else if (bytes_received == 0) {
std::string s = strerror(errno);
LOG(md, Error, "{} Received 0 bytes {}",socket_desc, s);
}
return bytes_received;
}
// Reusable function to create and join a multicast group
static int create_multicast_socket(
const std::string &multicast_ip,
const int port,
const std::string &interface_ip) {
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
// Create UDP socket
auto sock = socket(AF_INET, SOCK_DGRAM, 0);
if (sock < 0) {
LOG(md, Error, "Socket creation failed IP{} PORT{}", multicast_ip,
port);
return -1;
}
// Allow multiple sockets to use the same port
int reuse = 1;
if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) <
0) {
LOG(md, Error, "Setting SO_REUSEADDR failed IP{} PORT{}", multicast_ip,
port);
close(sock);
return -1;
}
// Bind to the multicast port
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
addr.sin_addr.s_addr = htonl(INADDR_ANY);
if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
LOG(md, Error, "Bind failed IP{} PORT{}", multicast_ip, port);
close(sock);
return -1;
}
// Specify the network interface for multicast
struct in_addr localInterface;
localInterface.s_addr = inet_addr(interface_ip.c_str());
if (setsockopt(
sock, IPPROTO_IP, IP_MULTICAST_IF, (char *)&localInterface,
sizeof(localInterface)) < 0) {
LOG(md, Error, "Setting local interface failed IP{} PORT{}",
multicast_ip, port);
close(sock);
return -1;
}
// Join the multicast group
struct ip_mreq mreq;
mreq.imr_multiaddr.s_addr = inet_addr(multicast_ip.c_str());
mreq.imr_interface.s_addr = inet_addr(interface_ip.c_str());
if (setsockopt(
sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *)&mreq,
sizeof(mreq)) < 0) {
LOG(md, Error, "adding multicast group failed IP{} PORT{}",
multicast_ip, port);
close(sock);
return -1;
}
LOG(md, Info, "Joined multicast group on IP: {}, Port: {}", multicast_ip,
port);
// struct timeval tv;
// tv.tv_sec = 5;
// tv.tv_usec = 0;
// setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof tv);
return sock;
}
static int create_multicast_socket_assert(
const std::string &ip, int port, const std::string &market_data_ip) {
int socket =
TaifexUtils::create_multicast_socket(ip, port, market_data_ip);
if (socket == -1) {
bts_assert_msgf(false, "Couldn't open socket for {}:{}", ip, port);
}
return socket;
}
static void close_sockets(std::set<int> &sockets) {
for (int socket : sockets) {
try {
close(socket);
} catch (...) {
LOG(md, Error, "Couldnt properly close socket {}", socket);
}
}
sockets.clear();
}
static std::string getIP(const std::string &ipType) {
auto isLastTradingDay=[](){
return false;
};
const std::string regular_referenceIP = "225.0.140.141";
const std::string regular_refreshIP = "225.0.140.147";
const std::string regular_realTimeIP = "225.0.140.140";
const std::string afterHours_referenceIP = "225.10.140.141";
const std::string afterHours_refreshIP = "225.10.140.147";
const std::string afterHours_realTimeIP = "225.10.140.140";
// Get current time
std::time_t now = std::time(nullptr);
std::tm *localTime = std::localtime(&now);
int hour = localTime->tm_hour;
// int minute = localTime->tm_min;
bool lastTradingDay = isLastTradingDay();
// Check if it's regular trading hours
// bool isRegularTrading = (hour == 8 && minute >= 45) ||
// (hour > 8 && hour < 14 && (lastTradingDay ? minute < 30 : minute < 45));
// // Check if it's after-hours trading
bool isAfterHoursTrading = !lastTradingDay &&
((hour == 15) || (hour > 15 && hour < 24) || (hour < 5));
// Determine the IP based on the time of day and type of IP requested
if (isAfterHoursTrading) {
if (ipType == "reference")
return afterHours_referenceIP;
if (ipType == "refresh")
return afterHours_refreshIP;
if (ipType == "realtime")
return afterHours_realTimeIP;
} //else if (isAfterHoursTrading) {
// }
if (ipType == "reference")
return regular_referenceIP;
if (ipType == "refresh")
return regular_refreshIP;
if (ipType == "realtime")
return regular_realTimeIP;
// Return an error message if an invalid IP type is passed
return "Invalid IP type!";
}
static std::string getBufferString(
const UNPROCESSED_MSG &buffer,
size_t offset = 0,
size_t length = std::numeric_limits<size_t>::max()) {
// Adjust length if it exceeds buffer size
if (length == std::numeric_limits<size_t>::max() ||
offset + length > buffer.size()) {
length = buffer.size() - offset;
}
if (offset >= buffer.size()) {
offset = buffer.size() - 1;
}
std::stringstream ss; // Use stringstream to build the string
for (size_t i = 0; i < length; ++i) {
ss << " offset: " << offset << ", i: " << i << ", value: " << std::hex
<< std::uppercase << std::setw(2) << std::setfill('0')
<< (unsigned int)buffer[offset + i] << "|";
}
return ss.str(); // Return the constructed string
}
static void printBuffer(
const UNPROCESSED_MSG &buffer,
size_t offset = 0,
size_t length = std::numeric_limits<size_t>::max()) {
// Adjust length if it exceeds buffer size
if (length == std::numeric_limits<size_t>::max() ||
offset + length > buffer.size()) {
length = buffer.size() - offset;
}
if (offset >= buffer.size())
offset = buffer.size() - 1;
for (size_t i = 0; i < length; ++i) {
printf(
" offset: %zu, i: %zu, value: %.2X|", offset, i,
(unsigned int)buffer[offset + i]);
}
// std::cout << std::dec << std::endl;
}
//Replace std::localtime with localtime_s (use localtime_r for POSIX)
static std::string getCurrentDate() {
auto now = std::chrono::system_clock::now();
std::time_t now_c = std::chrono::system_clock::to_time_t(now);
std::tm local_tm = *std::localtime(&now_c);
char buffer[11]; // Buffer to hold formatted date
std::strftime(buffer, sizeof(buffer), "%Y-%m-%d", &local_tm);
return std::string(buffer);
}
static boost::posix_time::ptime convertToPTime(
const std::string &match_time) {
if (match_time.length() != 12) {
throw std::invalid_argument(
"match_time must be exactly 12 characters long");
}
// Extract hours, minutes, seconds, and microseconds from match_time
std::string hours = match_time.substr(0, 2);
std::string minutes = match_time.substr(2, 2);
std::string seconds = match_time.substr(4, 2);
std::string microseconds = match_time.substr(6, 6);
// Get the current date
std::string date = getCurrentDate();
// Combine the date and time
std::string datetime = date + " " + hours + ":" + minutes + ":" + seconds;
// Convert to ptime
boost::posix_time::ptime pt =
boost::posix_time::time_from_string(datetime);
// Add fractional seconds (microseconds)
boost::posix_time::ptime result =
pt + boost::posix_time::microseconds(std::stoi(microseconds));
return result;
}
static std::string convertToISO(
const std::string &match_time,
const std::string &format = "datetime_no_timezone") {
if (match_time.length() != 12) {
throw std::invalid_argument(
"in function convertToISO: match_time must be exactly 12 "
"characters "
"long: -> " +
match_time);
}
try {
// Extract time components
std::string hours = match_time.substr(0, 2);
std::string minutes = match_time.substr(2, 2);
std::string seconds = match_time.substr(4, 2);
// Validate and sanitize time components
int h = std::abs(std::stoi(hours)) % 24;
int m = std::abs(std::stoi(minutes)) % 60;
int s = std::abs(std::stoi(seconds)) % 60;
// Get the current date dynamically
std::string current_date = getCurrentDate();
// Construct the desired ISO format
std::ostringstream iso_stream;
if (format == "datetime_with_timezone") {
// Full date-time with time zone: "YYYY-MM-DDTHH:MM:SS+02:00" (for simplicity using UTC "Z")
iso_stream << current_date << "T" << std::setfill('0')
<< std::setw(2) << h << ":" << std::setfill('0')
<< std::setw(2) << m << ":" << std::setfill('0')
<< std::setw(2) << s << "Z"; // Assuming UTC for now
} else if (format == "datetime_no_timezone") {
// Date-time without time zone: "YYYY-MM-DDTHH:MM:SS"
iso_stream << current_date << "T" << std::setfill('0')
<< std::setw(2) << h << ":" << std::setfill('0')
<< std::setw(2) << m << ":" << std::setfill('0')
<< std::setw(2) << s;
} else if (format == "date_only") {
// Date only: "YYYY-MM-DD"
iso_stream << current_date;
} else {
throw std::invalid_argument(
"Unknown format specified for convertToISO: " + format);
}
return iso_stream.str();
} catch (const std::exception &e) {
std::cerr << "Error in convertToISO: " << e.what() << std::endl;
throw;
}
}
static std::string sanitizeISOString(const std::string &isoString) {
std::string result = isoString;
// Remove any whitespace
result.erase(
std::remove_if(result.begin(), result.end(), ::isspace),
result.end());
// Ensure it's at least 20 characters long (e.g., YYYY-MM-DDTHH:MM:SS)
if (result.size() != 20) {
if (result.size() < 20) {
// If it's too short, pad with zeros at the end (assuming it's missing time components)
result.append(20 - result.size(), '0');
} else {
// If it's too long, truncate to 20 characters
result = result.substr(0, 20);
}
}
// Check for and correct any time zone issues or missing Z (UTC indicator)
size_t posT = result.find('T');
if (posT != std::string::npos) {
size_t posZ = result.find('Z', posT);
if (posZ == std::string::npos) {
size_t posMinus = result.find('-', posT);
size_t posPlus = result.find('+', posT);
if (posMinus != std::string::npos && posMinus > posT) {
// Check if it's "-0" or "-00" or "-000", etc.
if (std::all_of(
result.begin() + posMinus + 1, result.end(),
[](char c) { return c == '0'; })) {
result = result.substr(0, posMinus) + "Z";
}
} else if (posPlus != std::string::npos && posPlus > posT) {
// Check if it's "+0" or "+00" or "+000", etc.
if (std::all_of(
result.begin() + posPlus + 1, result.end(),
[](char c) { return c == '0'; })) {
result = result.substr(0, posPlus) + "Z";
}
} else {
// If there's no timezone indicator, assume UTC and add 'Z'
result += "Z";
}
}
}
return result;
}
static std::string time_decodePackBCD(
const unsigned char *data, size_t length) {
std::stringstream ss;
for (size_t i = 0; i < length; ++i) {
int upper = (data[i] >> 4) & 0x0F;
int lower = data[i] & 0x0F;
// Check if the decoded values are valid digits
if (upper > 9 || lower > 9) {
// Instead of throwing an error, replace invalid digits with '0'
upper = upper > 9 ? 0 : upper;
lower = lower > 9 ? 0 : lower;
}
ss << upper << lower;
}
std::string result = ss.str();
// Ensure the result is exactly 12 characters long (hhmmssxxxxxx)
if (result.length() < 12) {
// If less than 12 characters, pad with zeros at the end
result = result + std::string(12 - result.length(), '0');
} else if (result.length() > 12) {
// If more than 12 characters, keep the first 12
result = result.substr(0, 12);
}
// Ensure the time part (hhmmss) is valid
int hours = std::stoi(result.substr(0, 2));
int minutes = std::stoi(result.substr(2, 2));
int seconds = std::stoi(result.substr(4, 2));
hours = std::min(23, std::max(0, hours));
minutes = std::min(59, std::max(0, minutes));
seconds = std::min(59, std::max(0, seconds));
// Reconstruct the result with valid time
std::stringstream final_ss;
final_ss << std::setfill('0') << std::setw(2) << hours
<< std::setfill('0') << std::setw(2) << minutes
<< std::setfill('0') << std::setw(2) << seconds
<< result.substr(6); // Keep the last 6 characters (xxxxxx) as is
return final_ss.str();
}
// need to change parameters
static std::string decodePackBCD(const unsigned char *data, size_t length) {
std::stringstream ss;
for (size_t i = 0; i < length; ++i) {
// Extract the upper and lower nibble (4 bits each)
int upper = (data[i] >> 4) & 0x0F;
int lower = data[i] & 0x0F;
// Append both nibbles to the stringstream
ss << upper << lower;
}
std::string result = ss.str();
// Remove leading zeros
result.erase(0, result.find_first_not_of('0'));
// If the result is empty (all zeros), return "0"
return result.empty() ? "0" : result;
}
static std::string extractString(
const UNPROCESSED_MSG &buffer,
size_t &position,
size_t length) {
// Extract the substring from the buffer
std::string result(
buffer.begin() + position, buffer.begin() + position + length);
position += length;
// Log the raw result
return result;
}
// Helper function to extract a BCD value from the buffer and convert to
// unsigned long
static unsigned long extractBCD_STOUL(
const UNPROCESSED_MSG &buffer,
size_t &position,
size_t length) {
std::string bcd_str = decodePackBCD(&buffer[position], length);
position += length;
return std::stoul(bcd_str);
}
static std::string extractBCD_STRING(
const UNPROCESSED_MSG &buffer,
size_t &position,
size_t length) {
std::string bcd_str = decodePackBCD(&buffer[position], length);
position += length;
return bcd_str;
}
// Helper function to extract a single character and convert to bool
static bool extractBool(
const UNPROCESSED_MSG &buffer, size_t &position) {
bool result = (buffer[position] == 'Y');
position += 1;
return result;
}
static std::string extractBCDDate(
const UNPROCESSED_MSG &buffer,
size_t &current_position,
size_t size) {
if (size != 4) {
throw std::invalid_argument("BCD date must be 4 bytes long");
}
std::string date;
for (size_t i = 0; i < size; ++i) {
unsigned char byte = buffer[current_position++];
char high = (byte >> 4) + '0';
char low = (byte & 0x0F) + '0';
date.push_back(high);
date.push_back(low);
}
// Ensure the date is in YYYYMMDD format
if (date.length() != 8) {
throw std::runtime_error("Extracted date is not in YYYYMMDD format");
}
return date;
}
static std::string remove_spaces(const std::string &name) {
std::string removed;
removed.reserve(name.size()); // Reserve space to avoid reallocations
const char space = ' ';
for (auto c : name) {
if (c != space) {
removed += c;
}
}
return removed;
}
};
} //namespace bts::md::taifex