Files
spark/src/lib/orderman/gateways/CCAPIExecutionGateway.cpp
T

494 lines
19 KiB
C++

#include "CCAPIExecutionGateway.h"
#include "lib/api/service_helpers/AdminServiceUtils.h"
#include "lib/api/services/AdminService.pb.h"
#include "lib/api/services/OrderService.pb.h"
#include "lib/core/TimeStamp.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/domain_model/SecurityKey.h"
#include "lib/domain_model/common.h"
#include "lib/logging/logging.h"
#include "lib/md/SecurityOracle.h"
#include "lib/orderman/CancelOrderMsg.h"
#include "lib/orderman/CancelReplaceOrderMsg.h"
#include "lib/orderman/CreateOrderMsg.h"
#include "lib/orderman/OrderMsg.h"
#include "lib/orderman/RFQMsg.h"
#include "lib/orderman/ccapi/CCAPIDeribitOrderMux.h"
#include "lib/orderman/ccapi/ICCAPIOrderMux.h"
#include "lib/orderman/events/NewAdminDataEvent.h"
#include "lib/orderman/events/NewExecutionReportEvent.h"
#include "lib/orderman/events/NewSecuritiesEvent.h"
#include "lib/orderman/orderman_utils/CancelReject.h"
#include "lib/orderman/orderman_utils/ExchangeOpenOrders.h"
#include "lib/orderman/orderman_utils/ExchangePosition.h"
#include "lib/orderman/orderman_utils/ExecutionReport.h"
#include "lib/orderman/orderman_utils/orderman_common.h"
#include "lib/os/os.h"
#include "lib/utils/bts_assert.h"
#include "lib/utils/bts_chrono.h"
#include "lib/utils/datetime.h"
#include "lib/api/service_helpers/AdminServiceUtils.h"
#include "lib/config/config.h"
#include "lib/config/protos/toggles.pb.h"
#include "lib/domain_model/Security.h"
#include "lib/orderman/ccapi/CCAPIOrderState.h"
#include "lib/orderman/ccapi/ICCAPIOrderMux.h"
#include "lib/utils/PBUtils.h"
#include "lib/utils/Uuid.h"
#include <atomic>
#include <bits/stdint-uintn.h>
//#include <boost/container/small_vector.hpp>
#include "lib/orderman/ccapi/CCAPIConnectionConfig.h"
#include <cmath>
#include <cstdint>
#include <optional>
#include <string>
#include <tbb/concurrent_queue.h>
#include <unordered_map>
#include <variant>
#include <vector>
namespace bts::orderman {
api::orderman::NativeOrderStatus native_order_status(const CCAPIOrderState &s) {
switch (s.status) {
case api::orderman::OrderStatus::ORDER_STATUS_INACTIVE:
return api::orderman::NativeOrderStatus::
NATIVE_ORDER_STATUS_UNACKNOWLEDGED;
case api::orderman::OrderStatus::ORDER_STATUS_WORKING:
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_NEW;
case api::orderman::OrderStatus::ORDER_STATUS_COMPLETED:
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_FILLED;
case api::orderman::OrderStatus::ORDER_STATUS_CANCELED:
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_CANCELED;
case api::orderman::OrderStatus::ORDER_STATUS_ERROR:
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_REJECTED;
case api::orderman::OrderStatus::ORDER_STATUS_DISCONNECTED:
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_UNDEFINED;
case api::orderman::OrderStatus::ORDER_STATUS_PENDING_WORKING:
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_PENDING_NEW;
case api::orderman::OrderStatus::ORDER_STATUS_REJECTED_BY_SPARK:
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_REJECTED;
case api::orderman::OrderStatus::OrderStatus_INT_MIN_SENTINEL_DO_NOT_USE_:
case api::orderman::OrderStatus::OrderStatus_INT_MAX_SENTINEL_DO_NOT_USE_:
bts_assert(false);
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_UNDEFINED;
}
bts_assert(false);
return api::orderman::NativeOrderStatus::NATIVE_ORDER_STATUS_UNDEFINED;
}
struct SingleOrderMux {
ICCAPIOrderMux *order_mux;
std::string account;
uint32_t account_id;
};
struct InitState {
enum class StateType { Start = 0, Securities = 1, Admin = 2 };
std::map<StateType, bool> state_map = {
{StateType::Start, false},
{StateType::Securities, false},
{StateType::Admin, false}};
bool set(StateType state) {
bool s0 = inited();
auto it = state_map.find(state);
bts_assert(it != state_map.end());
it->second = true;
bool s1 = inited();
return !s0 && s1;
}
bool inited() const {
for (const auto &kv : state_map)
if (!kv.second)
return false;
return true;
}
};
struct CCAPIExecutionGateway::PImpl {
PImpl(bts::api::Venue venue, md::SecurityOracle o)
: venue(venue), sec_oracle(o) {
LOG(OrderMan, Info, "CCAPIExecutionGateway::PImpl constructor");
}
// TODO: get it from config
std::vector<CCAPIConnectionConfig> configs = {
CCAPIConnectionConfig{
"MAIN",
CCAPIConnectionConfig::DeribitConnectionConfig{
"ZmIKL6fb", "_FY2PcXYNc8VdlfkZO5c-pkFJQFhn8P9z4_p8YA2fS0"}},
CCAPIConnectionConfig{
"SMALL",
CCAPIConnectionConfig::DeribitConnectionConfig{
"v1iSzZrY", "fFOX8SBB2ZKj43xoab2f3YV70DDK5064STWK3WOCrK4"}}};
const bts::api::Venue venue;
InitState init_state = {};
std::vector<SingleOrderMux> mux_vec = {};
const size_t pos_query_period_ms = 5000;
core::TimeStamp last_pos_query_time = core::TimeStamp::now();
api::admin::AdminData admin_data = {};
md::SecurityOracle sec_oracle;
// interface (IExecutionGateway) callbacks
OnExeReport on_exe_report;
OnGwStatusChange on_gw_status_change;
OnCancelReject on_cancel_reject;
OnExchangePositionState on_exchange_position_state;
OnExchangeOpenOrdersState on_exchange_open_orders;
// orders are sent out and callbacks received via IEMOrderMuxer
size_t msg_seq_num = 0;
std::unordered_map<OrderKey, CreateOrderMsg> order_map = {};
// this is updated as we get exchange positions
ExchangePosition exchange_position;
ExchangeOpenOrders exchange_open_orders;
};
CCAPIExecutionGateway::CCAPIExecutionGateway(
bts::api::Venue venue, md::SecurityOracle o)
: IExecutionGateway(), me(new PImpl(venue, o)) {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::CCAPIExecutionGateway constructor venue: {} "
"t_id: {} "
"t_name: {}",
bts::api::Venue_Name(venue), os::get_thread_id(),
os::get_application_name());
}
CCAPIExecutionGateway::~CCAPIExecutionGateway() {}
void CCAPIExecutionGateway::start() {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::start t_id: {} "
"t_name: {}",
os::get_thread_id(), os::get_thread_name());
bool just_inited = me->init_state.set(InitState::StateType::Start);
if (just_inited) {
_init();
}
}
// this is a periodic timer
void CCAPIExecutionGateway::dispatch_incoming_fix_messages() {
if (!me->init_state.inited())
return;
for (auto &mux : me->mux_vec)
mux.order_mux->dispatch_incoming_messages();
if (core::TimeStamp::now().millis() - me->last_pos_query_time.millis() >=
me->pos_query_period_ms) {
me->last_pos_query_time = core::TimeStamp::now();
for (auto &mux : me->mux_vec) {
mux.order_mux->query_positions();
mux.order_mux->query_open_orders();
}
}
}
void CCAPIExecutionGateway::on_securities(
const mercury::Update<NewSecuritiesEvent> &update) {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::on_securities t_id: {} t_name: {} n_secs: {}",
os::get_thread_id(), os::get_thread_name(),
update.events == nullptr ? 0 : update.events->size());
bool just_inited = me->init_state.set(InitState::StateType::Securities);
if (just_inited)
_init();
}
void CCAPIExecutionGateway::on_admin_data(
const mercury::Update<NewAdminDataEvent> &update) {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::on_admin_data t_id: {} t_name: {} size: {}",
os::get_thread_id(), os::get_thread_name(),
update.events != nullptr ? update.events->size() : 0);
bool has_data = update.events != nullptr && update.events->size() == 1;
if (!has_data)
return;
me->admin_data.CopyFrom(*(update.events)->at(0).admin_data);
bool just_inited = me->init_state.set(InitState::StateType::Admin);
if (just_inited) {
_init();
}
}
void CCAPIExecutionGateway::send_fix(const jetfix::OutFIXPtr &, trace::CorID) {
bts_assert_msg(false, "CCAPIExecutionGateway does not support fix");
}
void CCAPIExecutionGateway::send_msgs(
std::vector<OrderMsg> const &msg_vec, trace::CorID) {
if (!me->init_state.inited())
return;
LOG(OrderMan, Info, "CCAPIExecutionGateway::send_msgs count: {}",
msg_vec.size());
for (const auto &m : msg_vec) {
LOG(OrderMan, Info, "CCAPIExecutionGateway::send_msgs {}", m);
if (std::holds_alternative<CreateOrderMsg>(m)) {
const CreateOrderMsg &cr_msg = std::get<CreateOrderMsg>(m);
SingleOrderMux *mux = _find_mux(cr_msg.routing_record);
if (mux == nullptr) {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::send_msgs could not resolve acount for "
"cr: {}",
m);
// TODO: send execution report
continue;
}
LOG(OrderMan, Info,
"CCAPIExecutionGateway::send_msgs Create sec: {} order_key",
cr_msg.security->id(), cr_msg.key);
me->order_map.insert({cr_msg.key, cr_msg});
mux->order_mux->launch_new_order(cr_msg);
} else if (std::holds_alternative<CancelReplaceOrderMsg>(m)) {
const CancelReplaceOrderMsg &cxl_repl_msg =
std::get<CancelReplaceOrderMsg>(m);
SingleOrderMux *mux =
_find_mux_from_orderkey(cxl_repl_msg.create_fields.key);
if (mux == nullptr) {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::send_msgs could not resolve acount for "
"cr: {}",
m);
continue;
}
LOG(OrderMan, Info,
"CCAPIExecutionGateway::send_msgs CancelReplace sec: {} "
"order_key: {}",
cxl_repl_msg.create_fields.security->id(),
cxl_repl_msg.create_fields.key);
mux->order_mux->update_order(cxl_repl_msg);
} else if (std::holds_alternative<CancelOrderMsg>(m)) {
const CancelOrderMsg &cxl_msg = std::get<CancelOrderMsg>(m);
SingleOrderMux *mux = _find_mux_from_orderkey(cxl_msg.key);
if (mux == nullptr) {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::send_msgs could not resolve acount for "
"cr: {}",
m);
continue;
}
LOG(OrderMan, Info,
"CCAPIExecutionGateway::send_msgs Cancel order_key: {}",
cxl_msg.key);
mux->order_mux->cancel_order(cxl_msg);
} else if (std::holds_alternative<RFQMsg>(m)) {
const RFQMsg &rfq_msg = std::get<RFQMsg>(m);
LOG(OrderMan, Info, "CCAPIExecutionGateway::send_msgs RFQ: {}",
utils::pb::to_string(rfq_msg.rfq));
// me->order_mux->send_rfq(rfq_msg);
}
}
}
OnExeReport &CCAPIExecutionGateway::on_exe_report() {
return me->on_exe_report;
}
OnGwStatusChange &CCAPIExecutionGateway::on_gw_status_change() {
return me->on_gw_status_change;
}
OnCancelReject &CCAPIExecutionGateway::on_cancel_reject() {
return me->on_cancel_reject;
}
OnExchangePositionState &CCAPIExecutionGateway::on_exchange_position_state() {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::on_exchange_position_state subscribed");
return me->on_exchange_position_state;
}
OnExchangeOpenOrdersState &CCAPIExecutionGateway::on_exchange_open_orders() {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::on_exchange_open_orders subscribed");
return me->on_exchange_open_orders;
}
void CCAPIExecutionGateway::_init() {
bts_assert(me->init_state.inited());
LOG(OrderMan, Info, "CCAPIExecutionGateway::_init");
for (const auto &p : me->configs) {
api::admin::BTSAccount *acct_ptr = admin::AdminServiceUtils::find_account(
me->admin_data, p.internal_account);
if (acct_ptr == nullptr) {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::_init could not resolve account: {}",
p.internal_account);
continue;
}
CCAPIDeribitOrderMux *mux = new CCAPIDeribitOrderMux(p, me->sec_oracle);
me->mux_vec.push_back({mux, p.internal_account, acct_ptr->id()});
mux->order_update_cb() = [this](const CCAPIOrderState &order_state) {
_on_order_cb(order_state);
};
mux->exchange_position_update_cb() = [this](const auto &pos_vec) {
_on_exchange_positions(pos_vec);
};
mux->exchange_open_orders_update_cb() = [this](const auto &ord_vec) {
_on_exchange_working_orders(ord_vec);
};
}
}
void CCAPIExecutionGateway::_on_order_cb(const CCAPIOrderState &s) {
LOG(OrderMan, Info, "CCAPIExecutionGateway::_on_order_cb order_state: {}",
s);
auto it = me->order_map.find(s.order_key);
if (it == me->order_map.end()) {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::_process_cb_and_fire_er key: {} not found",
s.order_key);
}
const auto &cr_msg = it->second;
// cant create one on the stack
ExecutionReport er;
// length of these have to be <= 12
to_array(cr_msg.routing_record->account(), &er.account_name.value());
to_array(cr_msg.routing_record->operator_id(), &er.operator_id.value());
er.order_key = cr_msg.key;
er.msg_seq_num =
me->msg_seq_num++; // this is needed for fix, here we just ++ it
er.security = cr_msg.security;
er.leg_security = nullptr;
er.side = cr_msg.side;
er.time_in_force = cr_msg.time_in_force;
er.order_type = cr_msg.order_type;
er.exec_type = api::orderman::ExecType::EXEC_TYPE_NEW; // TODO
er.order_status = native_order_status(s);
er.manual_order_indicator =
api::orderman::ManualOrderIndicator::MANUAL_ORDER_INDICATOR_NO;
auto &er_exch_oid = std::get<ExchangeOrderID>(er.exchange_order_id).value();
to_array(s.cc_order_key, &er_exch_oid);
er.price.value() = s.order_px;
er.quantity = s.order_qty;
er.msg = s.msg;
// fractional qty
// fills
er.avg_price.value() = s.avg_filled_price;
er.filled_quantity = s.total_fills;
if (!s.cb_trades.empty()) {
if (s.cb_trades.size() > 1) {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::_on_order_cb multiple trade fills for "
"order_key: {}",
s.order_key);
}
er.exec_type = api::orderman::ExecType::EXEC_TYPE_PARTIAL_FILL;
if (s.status == api::orderman::ORDER_STATUS_COMPLETED) {
er.exec_type = api::orderman::ExecType::EXEC_TYPE_FILL;
}
auto &tr = s.cb_trades.back();
er.last_price = Price(tr.last_exec_price);
er.last_shares = Quantity::create_new(tr.last_exec_qty);
to_array(tr.trade_id, &er.exchange_fill_id.value());
} else {
er.last_shares = Quantity(0);
}
er.max_show = Quantity(0);
er.exchange = cr_msg.security->exchange();
ExchangeTimeStamp64 now = ExchangeTimeStamp64(datetime::unix_nano_now());
er.transact_time = now;
LOG(OrderMan, Info,
"CCAPIExecutionGateway::_process_cb_and_fire_er firing er: {}",
er.to_string());
me->on_exe_report(std::make_shared<ExecutionReport>(std::move(er)), {});
// _fire_leg_er(em_cb, er);
}
api::Security const *CCAPIExecutionGateway::_find_sec(
// TODO
api::Venue venue,
const std::string &exchange_security_desc) {
for (auto p : me->sec_oracle.get_securities()) {
if (p->exchange() == venue) {
bts_assert(venue == api::VENUE_DERIBIT);
if (p->deribit().security_desc() == exchange_security_desc)
return p;
}
}
return nullptr;
}
void CCAPIExecutionGateway::_on_exchange_positions(
const std::vector<CCAPIPosition> &pos_vec) {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::_on_exchange_positions cb sz: {}",
pos_vec.size());
ExchangePosition out_pos = {me->venue, {}};
for (const CCAPIPosition &pos : pos_vec) {
auto sec_ptr = _find_sec(pos.venue, pos.exchange_sec_desc);
api::admin::BTSAccount *acct_ptr = admin::AdminServiceUtils::find_account(
me->admin_data, pos.internal_account);
if (sec_ptr != nullptr && acct_ptr != nullptr) {
out_pos.total_exchange_positions.push_back(
ExchangePosition::UnitPosition{
pos.exchange_sec_desc, sec_ptr, pos.internal_account,
acct_ptr->id(), pos.position, Price{pos.average_price}});
} else {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::_on_exchange_positions could not "
"resolve "
"exchange: {} desc: {} account: {} , this position will not be "
"reported",
api::Venue_Name(pos.venue), pos.exchange_sec_desc,
pos.internal_account);
}
}
LOG(OrderMan, Info,
"CCAPIExecutionGateway::_on_exchange_positions firing positions sz: "
"{}",
out_pos.total_exchange_positions.size());
me->on_exchange_position_state(out_pos);
}
void CCAPIExecutionGateway::_on_exchange_working_orders(
const std::vector<CCAPIExchangeOpenOrder> &ord_vec) {
LOG(OrderMan, Info,
"CCAPIExecutionGateway::_on_exchange_working_orders sz: {}",
ord_vec.size());
ExchangeOpenOrders out_ords = {me->venue, {}};
for (const auto &ord : ord_vec) {
auto sec_ptr = _find_sec(me->venue, ord.exchange_sec_desc);
api::admin::BTSAccount *acct_ptr = admin::AdminServiceUtils::find_account(
me->admin_data, ord.account_name);
if (sec_ptr != nullptr && acct_ptr != nullptr) {
out_ords.all_open_orders.push_back(ExchangeOpenOrders::UnitOrder{
ord.exchange_sec_desc, ord.exchange_account, sec_ptr,
ord.exchange_order_id, ord.account_name, acct_ptr->id(), ord.tif,
ord.side, Price{ord.price}, Quantity::create_new(ord.qty),
Price{ord.filled_price}, Quantity::create_new(ord.filled_qty),
ord.creation_timestamp, ord.last_update_timestamp, !ord.web,
ord.web});
} else {
LOG(OrderMan, Error,
"CCAPIExecutionGateway::_on_exchange_working_orders could not "
"resolve exchange: {} desc: {} account: {} this order will not be "
"reported",
me->venue, ord.exchange_sec_desc, ord.account_name);
}
}
LOG(OrderMan, Info,
"CCAPIExecutionGateway::_on_exchange_working_orders firing orders "
"sz: "
"{}",
out_ords.all_open_orders.size());
me->on_exchange_open_orders(out_ords);
}
SingleOrderMux *CCAPIExecutionGateway::_find_mux(
api::orderman::RoutingRecord const *r) {
if (r == nullptr)
return nullptr;
for (auto &m : me->mux_vec) {
if (m.account == r->account())
return &m;
}
return nullptr;
}
SingleOrderMux *CCAPIExecutionGateway::_find_mux_from_orderkey(OrderKey key) {
auto it = me->order_map.find(key);
if (it == me->order_map.end())
return nullptr;
return _find_mux(it->second.routing_record);
}
} // namespace bts::orderman