petefisher78/spark
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4d85106fbfe2dfa10d242d44f3a252e0a3662d61
spark/src/lib/orderman/OrderServiceServer.cpp
T

594 lines
22 KiB
C++
Raw Blame History

#include "OrderServiceServer.h"
#include "lib/api/public/ServiceContext.h"
#include "lib/api/public/StatefulPublisher.h"
#include "lib/api/services/AdminService.client.pb.h"
#include "lib/api/services/MarginService.client.pb.h"
#include "lib/api/services/OrderService.pb.h"
#include "lib/api/services/PositionService.client.pb.h"
#include "lib/config/config.h"
#include "lib/config/protos/jetfix.pb.h"
#include "lib/config/protos/orderman_cfg.pb.h"
#include "lib/domain_model/Security.pb.h"
#include "lib/domain_model/common.h"
#include "lib/logging/logging.h"
#include "lib/md/MDClient.h"
#include "lib/md/MDConsumer.h"
#include "lib/mercury/actors/Timer.h"
#include "lib/mercury/executor/Executor.h"
#include "lib/metrics/metrics.h"
#include "lib/orderman/actors/AsyncFactory.h"
#include "lib/orderman/actors/FillStore.h"
#include "lib/orderman/actors/MDSource.h"
#include "lib/orderman/actors/MDSupplementarySource.h"
#include "lib/orderman/actors/OrderStore.h"
#include "lib/orderman/actors/SecuritySrc.h"
#include "lib/orderman/actors/ServiceBridge.h"
#include "lib/orderman/actors/order_manager/OrderManager.h"
#include "lib/orderman/events/ExchangePositionEvent.h"
#include "lib/orderman/events/OrderOpResultEvent.h"
#include "lib/orderman/order_factory/DynamicOrderRegistry.h"
#include "lib/orderman/orderman_utils/ExchangeOpenOrders.h"
#include "lib/plugin/SparkServices.h"
#include "lib/plugin/plugin.h"
#include "lib/utils/PBUtils.h"
#include "lib/utils/bts_chrono.h"
#include "lib/utils/thread_manager.h"
#include "lib/utils/type_name.h"
#include <bits/stdint-uintn.h>
#include <boost/algorithm/string/predicate.hpp>
#include <functional>
#include <optional>
#include <unordered_map>
namespace bts {
namespace orderman {
struct ActorThread {
ActorThread(
const config::OrdermanCfg::Executor &cfg,
metrics::MetricsContainer &stats)
: cfg(cfg), thread(), executor(), loop_metric(), done(false) {
std::array<double, 30> buckets;
auto i = 0;
for (auto &b : buckets)
b = (i += 1000);
loop_metric = std::make_shared<metrics::HistogramMetric<30>>(
fmt::format("spark_metrics_{}_thread_loop_ns", cfg.thread_name()),
buckets);
stats.publish(loop_metric);
}
void start(trace::PerfLog &log) {
if (cfg.poll())
thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
for (;;)
executor.do_work(log);
});
else {
thread = threading::make_thread(cfg.thread_name(), [this, &log]() {
const auto inner_iter = 1000;
while (!done) {
auto start = Timer::now();
for (auto i = 0; i < inner_iter; i++)
executor.do_work(log);
auto stop = Timer::now();
loop_metric->observe(ns_diff(stop, start) / inner_iter);
std::this_thread::sleep_for(10ms);
}
});
}
}
~ActorThread() {
done = true;
thread.join();
}
config::OrdermanCfg::Executor cfg;
std::thread thread;
mercury::Executor executor;
std::shared_ptr<metrics::HistogramMetric<30>> loop_metric;
std::atomic<bool> done;
};
ActorThread *find_thread_for_actor(
std::string const &actor_name,
std::vector<std::unique_ptr<ActorThread>> const &all_executors) {
for (auto const &ex : all_executors)
for (auto const &hosted : ex->cfg.hosted_actor())
if (hosted == "*" || boost::algorithm::to_lower_copy(hosted) ==
boost::algorithm::to_lower_copy(actor_name))
return ex.get();
return nullptr;
}
class OpenOrderRecordIDStore {
public:
std::uint32_t get_rec_id(const std::string &exch_order_id) {
auto [it, is_new] = rec_ids_.try_emplace(exch_order_id, rec_id_counter_);
if (is_new) {
rec_id_counter_++;
}
return it->second;
}
private:
std::uint32_t rec_id_counter_ = 1;
std::unordered_map<std::string, std::uint32_t> rec_ids_;
};
struct OrderServiceServer::PImpl {
PImpl(
bts::api::ServiceContext &ctx,
const Exchange_positionsStatefulPublisher &exchange_pos_publisher,
const Exchange_open_ordersStatefulPublisher &open_orders_publisher,
md::MDClient &mdc,
DynamicOrderRegistry const &d,
metrics::MetricsContainer &stats,
trace::PerfLog &log)
: service_bridge(), service_ready(false), known_orders(), known_fills(),
publish_orders(), publish_fills(),
exchange_pos_publisher(exchange_pos_publisher),
open_orders_publisher(open_orders_publisher),
admin_service(ctx.locate<api::admin::AdminService>()),
position_service(ctx.locate<api::position::PositionService>()),
margin_service(ctx.locate<api::position::MarginService>()),
md_consumer(mdc.current_consumer()), stats(stats),
dyn_order_registry(d), log(log), in_flight_order_ops(),
current_req_op_id(0) {}
ServiceBridge *service_bridge;
std::vector<std::unique_ptr<ActorThread>> actor_threads;
std::atomic<bool> service_ready;
std::unordered_set<OrderKey> known_orders;
std::unordered_set<std::string> known_fills;
OrderServiceServer::OrdersPublisher publish_orders;
OrderServiceServer::FillsPublisher publish_fills;
Exchange_positionsStatefulPublisher exchange_pos_publisher;
Exchange_open_ordersStatefulPublisher open_orders_publisher;
api::admin::AdminService admin_service;
api::position::PositionService position_service;
api::position::MarginService margin_service;
bts::md::MDConsumer &md_consumer;
metrics::MetricsContainer &stats;
DynamicOrderRegistry const &dyn_order_registry;
trace::PerfLog &log;
std::unordered_map<uint32_t, api::Response<api::orderman::Response>>
in_flight_order_ops;
uint32_t current_req_op_id;
OpenOrderRecordIDStore open_order_id_store;
};
OrderServiceServer::OrderServiceServer(
bts::api::ServiceContext &ctx,
md::MDClient &mdc,
DynamicOrderRegistry const &d,
metrics::MetricsContainer &stats,
trace::PerfLog &log,
md::TradeSummarySignalDispatcher &tssd)
: api::orderman::OrderServiceBase(ctx),
me(new PImpl(
ctx,
make_exchange_positions_stateful_publisher(),
make_exchange_open_orders_stateful_publisher(),
mdc,
d,
stats,
log)) {
LOG(md, Info, "OrderServiceServer cons t_id: {} t_name: {}",
os::get_thread_id(), os::get_thread_name());
me->publish_orders = make_orders_publisher();
me->publish_fills = make_fills_publisher();
std::vector<mercury::Executor *> all_executors;
for (auto const &ex_cfg : config::get<config::OrdermanCfg>().executor()) {
auto at = std::make_unique<ActorThread>(ex_cfg, stats);
all_executors.push_back(&at->executor);
me->actor_threads.push_back(std::move(at));
}
for (auto &t : me->actor_threads)
t->executor.init_neighbours(all_executors);
// MDSource
auto thread = find_thread_for_actor("MDSource", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for MDSource. See the "
"OrdermanCfg key");
thread->executor.spawn<MDSource>(me->md_consumer, me->stats, log, &tssd);
// OrderManager
thread = find_thread_for_actor("OrderManager", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for OrderManager. See "
"the OrdermanCfg key");
thread->executor.spawn<OrderManager>(d, me->log, mdc.oracle());
// Timer
thread = find_thread_for_actor("Timer", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for Timer. See the "
"OrdermanCfg key");
thread->executor.spawn<mercury::Timer>();
// AsyncFactory
thread = find_thread_for_actor("AsyncFactory", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for AsyncFactory. See "
"the OrdermanCfg key");
thread->executor.spawn<AsyncFactory>(d, mdc.oracle());
// OrderStore
thread = find_thread_for_actor("OrderStore", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for OrderStore. See "
"the OrdermanCfg key");
thread->executor.spawn<OrderStore>(mdc.oracle());
// FillStore
thread = find_thread_for_actor("FillStore", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for FilllStore. See "
"the OrdermanCfg key");
thread->executor.spawn<FillStore>(mdc.oracle());
// SecuritySrc
thread = find_thread_for_actor("SecuritySrc", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for SecuritySrc. See "
"the OrdermanCfg key");
thread->executor.spawn<SecuritySrc>(me->md_consumer);
// MDSupplementarySource
thread = find_thread_for_actor("MDSupplementarySource", me->actor_threads);
bts_assert_msg(
thread,
"Could not find a configured thread for MDSupplementarySource. See "
"the OrdermanCfg key");
thread->executor.spawn<MDSupplementarySource>(me->md_consumer);
// ServiceBridge
thread = find_thread_for_actor("ServiceBridge", me->actor_threads);
bts_assert_msg(
thread, "Could not find a configured thread for ServiceBridge. See "
"the OrdermanCfg key");
me->service_bridge = new ServiceBridge(
[this]() { me->service_ready = true; },
[this, &ctx](auto const &orders) {
ctx.post([this, orders] { this->on_orders(orders); });
},
[this, &ctx](auto const &fill) {
ctx.post([this, fill] { this->on_fills(fill); });
},
[this, &ctx](auto const &order_op_result) {
ctx.post([this, order_op_result] {
this->on_order_op_result(order_op_result);
});
},
[this, &ctx](const ExchangePosition &exch_pos) {
LOG(OrderMan, Info, "OrderServiceServer exchange position n_pos: {} ",
exch_pos.total_exchange_positions.size());
ctx.post([this, exch_pos] { this->_on_exchange_position(exch_pos); });
},
[this, &ctx](const ExchangeOpenOrders &exch_open_orders) {
LOG(OrderMan, Info,
"OrderServiceServer open orders n_open_orders: {} ",
exch_open_orders.all_open_orders.size());
ctx.post([this, exch_open_orders] {
this->_on_exchange_open_orders(exch_open_orders);
});
},
me->dyn_order_registry, thread->executor.make_proxy(), mdc.oracle());
thread->executor.activate(me->service_bridge);
// spin up threads
for (auto &t : me->actor_threads)
t->start(log);
// wait for the service bridge to be ready
me->service_bridge->block_until_ready();
// subscribe to admin model updates and forward them to the service bridge
me->admin_service.subscribe_admin_data() |
rxcpp::operators::subscribe<api::admin::Admin_dataUpdate>(
[this](auto const &update) {
bts_assert(update.all().size() == 1);
me->service_bridge->update_admin_data(update.all()[0]);
});
// Subscribe to position updates and foward them to the service bridge
me->position_service.subscribe_pnl_account_data() |
rxcpp::operators::subscribe<api::position::Pnl_account_dataUpdate>(
[this](auto const &update) {
me->service_bridge->update_account_pnl(update);
});
// Subscribe
me->position_service.subscribe_position_data() |
rxcpp::operators::subscribe<api::position::Position_dataUpdate>(
[this](auto const &update) {
me->service_bridge->update_position(update);
});
me->margin_service.subscribe_excess_liquidity_report() |
rxcpp::operators::subscribe<
api::position::Excess_liquidity_reportUpdate>(
[this](auto const &update) {
me->service_bridge->update_excess_liquidity_report(update);
});
}
OrderServiceServer::~OrderServiceServer() {
bts_assert(false); // this is badly broken at the moment
}
void OrderServiceServer::on_orders(
std::vector<api::orderman::OrderRecord> const &orders) {
LOG(OrderMan, Info, "Received order update for publishing");
std::vector<api::orderman::OrderRecord> created;
created.reserve(orders.size());
std::vector<api::orderman::OrderRecord> updated;
updated.reserve(orders.size());
for (auto const &rec : orders) {
LOG(OrderMan, Info, "OrderServiceServer::on_orders order: {}",
utils::pb::to_string(rec));
auto const &k = OrderKey(rec.key());
if (me->known_orders.count(k) == 0) {
me->known_orders.insert(k);
created.push_back(rec);
} else
updated.push_back(rec);
}
LOG(OrderMan, Info,
"OrderServiceServer::on_orders publishing orders to the service API. "
"created {}, updated {}",
created.size(), updated.size());
me->publish_orders(&created, &updated, nullptr);
}
void OrderServiceServer::on_fills(
std::vector<api::orderman::Fill> const &fills) {
LOG(OrderMan, Info, "Received fill update for publishing");
std::vector<api::orderman::Fill> created;
created.reserve(fills.size());
std::vector<api::orderman::Fill> updated;
updated.reserve(fills.size());
for (auto const &f : fills) {
auto const &k = f.fill_key();
if (me->known_fills.count(k) == 0) {
me->known_fills.insert(k);
created.push_back(f);
} else
updated.push_back(f);
}
LOG(OrderMan, Info, "OrderServiceServer::on_fills created: {} updated: {}",
created.size(), updated.size());
me->publish_fills(&created, &updated, nullptr);
}
void OrderServiceServer::create_order(
const api::orderman::Create &c,
const api::Response<api::orderman::Response> &r) {
LOG(OrderMan, Info, "OrderServiceServer::create_order create: {}",
utils::pb::to_string(c));
if (!me->service_ready) {
api::orderman::Response response;
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
r(response);
} else {
me->in_flight_order_ops[++me->current_req_op_id] = r;
me->service_bridge->create_order(me->current_req_op_id, c);
}
}
void OrderServiceServer::cancel_order(
const api::orderman::Cancel &cancel,
const api::Response<api::orderman::Response> &r) {
if (!me->service_ready) {
api::orderman::Response response;
response.set_order_key(cancel.order_key());
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
r(response);
} else {
me->in_flight_order_ops[++me->current_req_op_id] = r;
me->service_bridge->cancel_order(me->current_req_op_id, cancel);
}
}
void OrderServiceServer::update_order(
const api::orderman::Update &update,
const api::Response<api::orderman::Response> &r) {
LOG(OrderMan, Info, "OrderServiceServer::update_order update: {}",
utils::pb::to_string(update));
if (!me->service_ready) {
api::orderman::Response response;
response.set_order_key(update.order_key());
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
r(response);
} else {
me->in_flight_order_ops[++me->current_req_op_id] = r;
me->service_bridge->update_order(me->current_req_op_id, update);
}
}
void OrderServiceServer::detach_order(
const api::orderman::Detach &detach,
const api::Response<api::orderman::Response> &r) {
if (!me->service_ready) {
api::orderman::Response response;
response.set_order_key(detach.order_key());
response.set_error_code(api::orderman::ORDER_ERROR_SERVER_NOT_READY);
r(response);
} else {
me->in_flight_order_ops[++me->current_req_op_id] = r;
me->service_bridge->detach_order(me->current_req_op_id, detach);
}
}
void OrderServiceServer::get_registered_dynamic_order_types(
const api::VoidMessage &,
const api::Response<api::orderman::RegisteredDynamicOrderTypes> &r) {
api::orderman::RegisteredDynamicOrderTypes response;
auto all_orders = me->dyn_order_registry.get_all_dynamic_order_types();
for (auto const &desc : all_orders)
*response.add_dynamic_order_desc() = desc;
r(response);
}
void OrderServiceServer::manually_bust_all_fills(
const api::orderman::ManualBustParameters &p,
const api::Response<api::orderman::Response> &r) {
LOG(OrderMan, Info, "Received manually_bust_all_fills: {}",
utils::pb::to_string(p));
me->service_bridge->manually_bust_all_fills(p);
api::orderman::Response response;
r(response);
}
void OrderServiceServer::send_rfq(
const api::orderman::RFQRequest &rfq,
const api::Response<api::orderman::Response> &r) {
LOG(OrderMan, Info, "OrderServiceServer::send_rfq rfq_request: {}",
utils::pb::to_string(rfq));
me->service_bridge->send_rfq(rfq);
api::orderman::Response response;
r(response);
}
void OrderServiceServer::set_precomputed_theos(
const bts::api::orderman::PrecomputedTheos &precomp_theos,
const api::Response<::bts::api::ResultMessage> &reply) {
LOG(Gen, Info, "OrderServiceServer::set_precomputed_theos count: {}",
precomp_theos.theos().size());
me->service_bridge->set_precomputed_theos(precomp_theos);
bts::api::ResultMessage res = {};
reply(res);
}
void OrderServiceServer::on_order_op_result(
OrderOpResultEvent const &op_result) {
auto it = me->in_flight_order_ops.find(op_result.req_id);
bts_assert(it != me->in_flight_order_ops.end());
auto &responder = it->second;
api::orderman::Response r;
r.set_order_key(op_result.ok.value());
r.set_error_code(op_result.result.error);
r.set_msg(op_result.result.error_msg);
responder(r);
me->in_flight_order_ops.erase(it);
}
void OrderServiceServer::_on_exchange_position(
const ExchangePositionEvent &ev) {
LOG(OrderMan, Info, "OrderServiceServer::_on_exchange_position n_pos: {}",
ev.position.total_exchange_positions.size());
for (const auto &p : ev.position.total_exchange_positions) {
bts::api::orderman::RetrievedExchangePosition ret_pos = {};
std::string id = p.exchange_sec_desc + "_" +
bts::api::Venue_Name(ev.position.exchange) + "_" +
p.account_name;
ret_pos.set_record_id(id);
ret_pos.set_security_key(p.security == nullptr ? 0 : p.security->id());
ret_pos.set_security_resolved(p.security != nullptr);
ret_pos.set_account_name(p.account_name);
ret_pos.set_account_id(
p.account_id == std::nullopt ? 0 : p.account_id.value());
ret_pos.set_account_resolved(p.account_id != std::nullopt);
ret_pos.set_exchange_desc(p.exchange_sec_desc);
ret_pos.set_exchange(ev.position.exchange);
ret_pos.set_avg_filled_price(p.avg_filled_price.value());
ret_pos.set_position(p.position);
me->exchange_pos_publisher.add_or_update(ret_pos);
LOG(OrderMan, Info,
"OrderServiceServer::_on_exchange_position publishing pos id: {} "
"key: {} res: {} account_name: {} account_id: {} "
"account_resolved: "
"{} desc: {} exchange: {} filled_price: {} fills: {}",
ret_pos.record_id(), ret_pos.security_key(),
ret_pos.security_resolved(), ret_pos.account_name(),
ret_pos.account_id(), ret_pos.account_resolved(),
ret_pos.exchange_desc(), bts::api::Venue_Name(ret_pos.exchange()),
ret_pos.avg_filled_price(), ret_pos.position());
}
LOG(OrderMan, Info,
"OrderServiceServer::_on_exchange_position publishing {} records",
ev.position.total_exchange_positions.size());
me->exchange_pos_publisher.flush();
}
void OrderServiceServer::_on_exchange_open_orders(
const ExchangeOpenOrdersEvent &ev) {
using namespace bts::api::orderman;
LOG(OrderMan, Info,
"OrderServiceServer::_on_exchange_open_orders n_open_orders: {} "
"current_sz: {}",
ev.open_orders.all_open_orders.size(),
me->open_orders_publisher.keys().size());
std::map<uint32_t, RetrievedExchangeOpenOrder> new_state = {};
auto get_signed_qty = [](const auto &o, Quantity qty) {
double qty_d = std::abs(qty.full_value());
if (o.side == api::Side::SIDE_SELL) {
qty_d = -1.0 * qty_d;
}
return qty_d;
};
for (const auto &o : ev.open_orders.all_open_orders) {
RetrievedExchangeOpenOrder o_ord = {};
auto rec_id = me->open_order_id_store.get_rec_id(o.exchange_order_id);
o_ord.set_record_id(rec_id);
o_ord.set_exchange_order_id(o.exchange_order_id);
o_ord.set_exchange_sec_desc(o.exchange_sec_desc);
o_ord.set_exchange_account(o.exchange_account);
o_ord.set_security_key(o.security == nullptr ? 0 : o.security->id());
o_ord.set_security_resolved(o.security != nullptr);
o_ord.set_account_id(o.account_id.has_value() ? o.account_id.value() : 0);
o_ord.set_account_name(o.account_id.has_value() ? o.account_name : "");
o_ord.set_account_resolved(o.account_id.has_value());
o_ord.set_tif(o.tif);
o_ord.set_price(o.price.value());
o_ord.set_qty(get_signed_qty(o, o.qty));
o_ord.set_filled_price(o.filled_price.value());
o_ord.set_filled_qty(get_signed_qty(o, o.filled_qty));
o_ord.set_creation_timestamp_ms(o.creation_timestamp);
o_ord.set_last_update_timestamp_ms(o.last_update_timestamp);
o_ord.set_api(o.api);
o_ord.set_web(o.web);
new_state[rec_id] = o_ord;
}
const auto &prev_state = me->open_orders_publisher.keys();
std::vector<RetrievedExchangeOpenOrder const *> added = {};
std::vector<RetrievedExchangeOpenOrder const *> updated = {};
std::vector<uint32_t> deleted = {};
bts::api::StatefulPublisher<RetrievedExchangeOpenOrder, uint32_t>::diff(
new_state, prev_state, added, updated, deleted);
for (const auto &k : deleted)
me->open_orders_publisher.erase(k);
for (RetrievedExchangeOpenOrder const *o : updated)
me->open_orders_publisher.add_or_update(*o);
for (RetrievedExchangeOpenOrder const *o : added)
me->open_orders_publisher.add_or_update(*o);
LOG(OrderMan, Info,
"OrderServiceServer::_on_exchange_open_orders publishing {} records",
ev.open_orders.all_open_orders.size());
me->open_orders_publisher.flush();
}
} // namespace orderman
} // namespace bts
Reference in new issue View Git Blame Copy Permalink