405 lines
13 KiB
C++
405 lines
13 KiB
C++
#ifndef NO_CPP0X
|
|
#include "RelayClient.h"
|
|
#include "RelayRecorder.h"
|
|
|
|
#include "lib/config/config.h"
|
|
#include "lib/utils/TimeStamp.h"
|
|
#include "lib/utils/bts_chrono.h"
|
|
#include "../MDProducer.h"
|
|
#include "../get_symbols.h"
|
|
#include "lib/logging/logging.h"
|
|
|
|
#include "lib/config/protos/system.pb.h"
|
|
#include "lib/config/protos/MarketData.pb.h"
|
|
#include "lib/md/protos/co2.pb.h"
|
|
#include "lib/api/services/MarketDataRelay.client.pb.h"
|
|
#include "lib/api/public/ServiceContext.h"
|
|
|
|
#include <memory>
|
|
#include <set>
|
|
#include <vector>
|
|
#include <array>
|
|
#include <algorithm>
|
|
#include <unordered_map>
|
|
#include <iostream>
|
|
#include <iomanip>
|
|
#include <chrono>
|
|
|
|
using namespace std;
|
|
|
|
namespace bts {
|
|
namespace md {
|
|
using namespace co2;
|
|
|
|
PriceSize convert(const PriceSizeMessage& psm) {
|
|
return psm.size() > 0 ? PriceSize(psm.mantissa(), psm.exponent(), psm.size()) : PriceSize::NO_MARKET;
|
|
}
|
|
PriceSize trade_pricesize(const co2::TradeMessage& trade) {
|
|
return trade.size() > 0 ?
|
|
PriceSize(trade.mantissa(), trade.exponent(), trade.size()) : PriceSize::NO_MARKET;
|
|
}
|
|
PriceSize book_at(const google::protobuf::RepeatedPtrField<PriceSizeMessage>& side, int index) {
|
|
if (side.size() > index) {
|
|
return convert(side.Get(index));
|
|
}
|
|
return PriceSize::NO_MARKET;
|
|
}
|
|
|
|
class RelayProducer : public IRelayWriter {
|
|
bool _shared;
|
|
std::map<Platform, MDProducer> _producers;
|
|
unordered_map<security_key, MDProducerHandle> _handles;
|
|
|
|
void record_update(MDProducerHandle& handle, const ChannelUpdate& update) {
|
|
utils::TimeStamp updated = utils::TimeStamp::now();
|
|
bool snapshot = update.has_security();
|
|
// If this is a snapshot an empty book() means "no market" rather than "no change"
|
|
if (update.has_book() || snapshot) {
|
|
std::array<PriceSize,COCONUT_MAX_DEPTH> bids, asks;
|
|
for (size_t x = 0; x < COCONUT_MAX_DEPTH; ++x) {
|
|
bids[x] = book_at(update.book().bids(), x);
|
|
asks[x] = book_at(update.book().asks(), x);
|
|
}
|
|
handle.update_book(updated, bids, asks);
|
|
}
|
|
if (update.has_customer_bbo()) {
|
|
handle.update_customer_bbo(updated, convert(update.customer_bbo().bid()), convert(update.customer_bbo().ask()));
|
|
}
|
|
if (update.has_misc()) {
|
|
auto& misc = update.misc();
|
|
handle.update_settlement(updated, misc.settlement_price(), datetime::Date(misc.settlement_date()));
|
|
handle.update_previous_settlement(updated, misc.previous_settlement_price(), datetime::Date(misc.previous_settlement_date()));
|
|
handle.update_opening(updated, misc.open_price(), datetime::Date(misc.open_date()));
|
|
handle.update_closing(updated, misc.close_price(), datetime::Date(misc.close_date()));
|
|
handle.update_session_high(updated, misc.session_high());
|
|
handle.update_session_low(updated, misc.session_low());
|
|
handle.update_open_interest(updated, misc.open_interest());
|
|
handle.update_session_high_bid(updated, misc.session_high_bid());
|
|
handle.update_session_low_offer(updated, misc.session_low_offer());
|
|
handle.update_fixing_price(updated, misc.fixing_price());
|
|
}
|
|
if (update.has_last_trade()) {
|
|
handle.set_last_trade(updated, trade_pricesize(update.last_trade()), update.last_trade().time(), update.last_trade().total_volume());
|
|
}
|
|
}
|
|
|
|
public:
|
|
RelayProducer(bool shared) :
|
|
_shared(shared),
|
|
_producers(),
|
|
_handles()
|
|
{}
|
|
|
|
MDProducer& ensure_producer(Platform platform) {
|
|
auto find_result = _producers.find(platform);
|
|
if (find_result == _producers.end()) {
|
|
auto result = _producers.insert(make_pair(platform, MDProducer(platform, _shared)));
|
|
if (result.second) {
|
|
return result.first->second;
|
|
}
|
|
}
|
|
return find_result->second;
|
|
}
|
|
|
|
void write(const co2::MDReply& data, bool* phas_new_securities) override {
|
|
for (const auto& ch : data.channels()) {
|
|
// Per-security data
|
|
int added_security_count = 0;
|
|
for (const auto& update : ch.updates()) {
|
|
auto find_result = _handles.find(update.key());
|
|
if (update.has_security() && find_result == _handles.end()) {
|
|
if (phas_new_securities) *phas_new_securities = true;
|
|
added_security_count++;
|
|
MDProducer& producer = ensure_producer(ch.platform());
|
|
with_security_like(update.security(), [&](FixedSecurity& fs) {
|
|
convert_security(update.security(), fs, true);
|
|
MDProducerHandle handle = producer.add_security(fs);
|
|
_handles.insert(make_pair(handle.key(), handle));
|
|
record_update(handle, update);
|
|
});
|
|
}
|
|
else if (find_result != _handles.end()) {
|
|
MDProducerHandle& handle = find_result->second;
|
|
record_update(handle, update);
|
|
}
|
|
}
|
|
// Heartbeat
|
|
auto iter = _producers.find(ch.platform());
|
|
if (iter != _producers.end()) {
|
|
auto& producer = iter->second;
|
|
auto channel = producer.get_channel(ch.channel().symbol().c_str(), ch.channel().type(), ch.channel().exchange());
|
|
channel.heartbeat();
|
|
|
|
// RFQ and TRADE LOG
|
|
if (ch.trade_log_size() > 0 || ch.rfq_log_size() > 0) {
|
|
if (ch.trade_log_size() > 0) {
|
|
LOG(md,Diag, "Added {} trades", ch.trade_log_size());
|
|
}
|
|
for (const auto& trade : ch.trade_log()) {
|
|
channel.add_trade(trade);
|
|
}
|
|
for (const auto& rfq : ch.rfq_log()) {
|
|
channel.add_rfq(rfq);
|
|
}
|
|
}
|
|
}
|
|
if (added_security_count > 0) {
|
|
LOG(md,Diag,"Added {} {} - {}" ,
|
|
added_security_count , ch.channel().symbol() ,SecurityType_Name(ch.channel().type()));
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
shared_ptr<IRelayWriter> mkrelayproducer(bool shared) {
|
|
return make_shared<RelayProducer>(shared);
|
|
}
|
|
|
|
// Single Threaded
|
|
class RelayClient {
|
|
MDRequest _request;
|
|
MarketDataRelay& _rpc;
|
|
MDConsumer* _pconsumer;
|
|
IRelayWriter& _writer;
|
|
utils::TimeStamp _request_log;
|
|
|
|
public:
|
|
RelayClient(
|
|
MarketDataRelay& rpc,
|
|
MDConsumer* pconsumer,
|
|
const std::set<UserSymbol>& symbols,
|
|
IRelayWriter& writer) :
|
|
_request(),
|
|
_rpc(rpc),
|
|
_pconsumer(pconsumer),
|
|
_writer(writer),
|
|
_request_log(utils::TimeStamp::now())
|
|
{
|
|
ostringstream fmt;
|
|
fmt << config::get<config::System>().host_name() << "(" << os::process_id() << ")";
|
|
_request.set_host(fmt.str());
|
|
for (const auto& s : symbols) {
|
|
SecurityChannel ch;
|
|
ch.set_symbol(s.symbol());
|
|
ch.set_type(s.type());
|
|
ch.set_exchange(s.exchange());
|
|
ch.set_flags(0xff);
|
|
add_symbol(ch);
|
|
}
|
|
}
|
|
|
|
// Can be called safely with the same symbol
|
|
void add_symbol(const SecurityChannel& ch) {
|
|
auto find_result =
|
|
std::find_if(_request.mutable_channels()->begin(), _request.mutable_channels()->end(),
|
|
[&](const SecurityChannel& c) {
|
|
return c.symbol() == ch.symbol() &&
|
|
c.type() == ch.type() &&
|
|
c.exchange() == ch.exchange(); });
|
|
if (find_result == _request.mutable_channels()->end()) {
|
|
auto* new_channel = _request.add_channels();
|
|
new_channel->CopyFrom(ch);
|
|
new_channel->set_time(0);
|
|
}
|
|
else {
|
|
// update flags
|
|
auto flags = find_result->flags() | ch.flags();
|
|
if (flags != find_result->flags()) {
|
|
find_result->set_flags(flags);
|
|
// After changing flags request a new snapshot
|
|
find_result->set_time(0);
|
|
}
|
|
}
|
|
}
|
|
|
|
void request_md(std::function<void()> cont) {
|
|
auto now = utils::TimeStamp::now();
|
|
if ((now - _request_log).seconds() > 60) {
|
|
_request_log = now;
|
|
LOG(md, Info, "Requesting MD: {}", _request.DebugString());
|
|
}
|
|
_rpc.requestmd(_request, [=](const api::Result<MDReply>& r) { receive_md(r); cont(); }, 500u);
|
|
}
|
|
void receive_md(const api::Result<MDReply>& result) {
|
|
if (result.success()) {
|
|
// Write data to shared mem
|
|
bool has_new_security = false;
|
|
_writer.write(*result.result(), &has_new_security);
|
|
// If the reply included and we have a consumer have it scan them
|
|
if (has_new_security && _pconsumer) {
|
|
scan_securities_thread().post([=]() {
|
|
auto count = _pconsumer->scan_securities(true);
|
|
LOG(md, Info, "Discovered {} securities.", count);
|
|
});
|
|
}
|
|
// Update reply timetamps
|
|
for (auto& req : *_request.mutable_channels()) {
|
|
for (auto& reply : result.result()->channels()) {
|
|
if (req.type() == reply.channel().type() &&
|
|
req.exchange() == reply.channel().exchange() &&
|
|
req.symbol() == reply.channel().symbol()) {
|
|
|
|
req.set_time(reply.channel().time());
|
|
req.set_trade_cursor(reply.channel().trade_cursor());
|
|
req.set_rfq_cursor(reply.channel().rfq_cursor());
|
|
if (req.segment_id() != reply.channel().segment_id()) {
|
|
LOG(md, Diag, "New segment_id = {}", reply.channel().segment_id());
|
|
req.set_segment_id(reply.channel().segment_id());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
struct RelayScheduler {
|
|
uint32_t _interval;
|
|
api::ServiceContext& _ctx;
|
|
RelayClient& _client;
|
|
bool running;
|
|
double speed_factor;
|
|
|
|
RelayScheduler(api::ServiceContext& context, uint32_t interval, RelayClient& client) :
|
|
_interval(interval),
|
|
_ctx(context),
|
|
_client(client),
|
|
running(true),
|
|
speed_factor(config::get<ReplayOptions>().speed())
|
|
{ }
|
|
|
|
void run() {
|
|
if (running) {
|
|
utils::TimeStamp start = utils::TimeStamp::now();
|
|
_client.request_md([=]() {
|
|
int64_t elapsed_ms = (utils::TimeStamp::now() - start).milliseconds();
|
|
int64_t delay = (_interval - elapsed_ms) / speed_factor;
|
|
_ctx.post([=]() { run(); }, std::max(delay, (int64_t)0));
|
|
});
|
|
}
|
|
}
|
|
};
|
|
|
|
|
|
void relay_everything() {
|
|
api::ServiceContext ctx;
|
|
std::set<UserSymbol> system_symbols;
|
|
get_system_md_symbols(system_symbols);
|
|
RelayProducer producer(true);
|
|
relay_selected(ctx, system_symbols, producer);
|
|
}
|
|
|
|
void relay_selected(api::ServiceContext& ctx, const std::set<UserSymbol>& symbols, IRelayWriter& writer) {
|
|
auto relay = ctx.locate<MarketDataRelay>();
|
|
RelayClient client(relay, nullptr, symbols, writer);
|
|
RelayScheduler scheduler(ctx, config::get<RelayOptions>().naggle_ms(), client);
|
|
ctx.post([&]() { scheduler.run(); });
|
|
|
|
// block forever, I guess
|
|
utils::Dispatcher::current().run();
|
|
}
|
|
void local_relay(IRelayWriter& writer) {
|
|
api::ServiceContext ctx;
|
|
auto relay = ctx.locate<MarketDataRelay>();
|
|
std::set<UserSymbol> symbols;
|
|
{
|
|
std::set<UserSymbol> system_symbols;
|
|
get_system_md_symbols(system_symbols);
|
|
|
|
// Filter out VENUE_SIM
|
|
for (const auto& s : system_symbols) {
|
|
if (s.exchange() != api::VENUE_SIM) {
|
|
symbols.insert(s);
|
|
}
|
|
}
|
|
}
|
|
relay_selected(ctx, symbols, writer);
|
|
}
|
|
|
|
void record_everything(std::string path, std::string filename) {
|
|
auto options = config::get<RecordOptions>();
|
|
if (filename.empty()) {
|
|
ostringstream file;
|
|
file << "marketdata" << "-";
|
|
int year, month, day, hour, minute, unused;
|
|
datetime::ll_to_parts(datetime::ll_local_now(), year, month, day, hour, minute, unused, unused);
|
|
file
|
|
<< year
|
|
<< setw(2) << setfill('0') << month
|
|
<< setw(2) << setfill('0') << day
|
|
<< "-"
|
|
<< setw(2) << setfill('0') << hour
|
|
<< setw(2) << setfill('0') << minute
|
|
<< ".coconut";
|
|
filename = file.str();
|
|
}
|
|
|
|
if (path.empty()) {
|
|
path = os::sparkdata_path("record");
|
|
}
|
|
|
|
auto naggle_ms = config::get<RelayOptions>().naggle_ms();
|
|
RelayRecorder recorder(path, filename, naggle_ms);
|
|
|
|
api::ServiceContext ctx;
|
|
auto relay = ctx.locate<MarketDataRelay>();
|
|
|
|
std::set<UserSymbol> system_symbols;
|
|
get_system_md_symbols(system_symbols);
|
|
|
|
RelayClient client(relay, nullptr, system_symbols, recorder);
|
|
RelayScheduler scheduler(ctx, naggle_ms, client);
|
|
scheduler.run();
|
|
|
|
utils::Dispatcher::current().run();
|
|
}
|
|
|
|
/*
|
|
* extern api for c#
|
|
*/
|
|
|
|
struct RelayControl {
|
|
RelayControl(MDConsumer* pconsumer, uint32_t interval) :
|
|
context(),
|
|
relay(context.locate<MarketDataRelay>()),
|
|
producer(false),
|
|
client(relay, pconsumer, std::set<UserSymbol>(), producer),
|
|
scheduler(context, interval, client)
|
|
{
|
|
scheduler.run();
|
|
}
|
|
api::ServiceContext context;
|
|
MarketDataRelay relay;
|
|
RelayProducer producer;
|
|
RelayClient client;
|
|
RelayScheduler scheduler;
|
|
};
|
|
|
|
RelayControl* start_private_relay(MDConsumer* pconsumer, uint32_t interval) {
|
|
bts_assert(pconsumer);
|
|
return new RelayControl(pconsumer, interval);
|
|
}
|
|
void stop_relay(RelayControl* relay) {
|
|
if (relay) {
|
|
relay->scheduler.running = false;
|
|
bool terminated = false;
|
|
relay->context.post([&]() {
|
|
terminated = true;
|
|
}, 500);
|
|
}
|
|
}
|
|
void add_relay_symbol(RelayControl* relay, const SecurityChannel& chan) {
|
|
bts_assert(relay);
|
|
relay->context.post([=]() {
|
|
relay->client.add_symbol(chan);
|
|
});
|
|
}
|
|
|
|
utils::Dispatcher& scan_securities_thread() {
|
|
return utils::Dispatcher::background();
|
|
}
|
|
|
|
|
|
} }//namespace
|
|
#endif
|