Files
spark/src/lib/md/SecurityDB.cpp
T

210 lines
7.8 KiB
C++

#include "SecurityDB.h"
#ifndef WIN32
#include "lib/utils/Sqlite_utils.h"
#include "lib/config/config.h"
#include "lib/config/protos/system.pb.h"
#include "lib/md/Shm/FixedSecurity.h"
#include <boost/container/flat_set.hpp>
#include <boost/filesystem.hpp>
#include <string>
using std::string;
using namespace bts::utils;
namespace bts {
namespace md {
namespace db {
/*
* Schema:
* key, created_date, expiration_date
*/
sqlite3_ptr open_db() {
// Directory location on disk
auto db_dir = config::get<config::System>().data_path() + "databases/md/";
auto db_path = db_dir + "/securities.sqlite";
if (!boost::filesystem::exists(db_dir)) {
boost::filesystem::create_directories(db_dir);
}
// Create table if it does not exist
auto dbptr = utils::open_db(db_path.c_str());
auto db = dbptr.get();
sqlite3_busy_timeout(db, 9999);
sqlite3_exec_nofail(db, "PRAGMA journal_mode=OFF");
sqlite3_exec_nofail(db, "PRAGMA syncronous=0");
sqlite3_exec_nofail(db,
"CREATE TABLE IF NOT EXISTS securities (key INTEGER primary key, symbol TEXT NOT NULL, venue INTEGER NOT NULL, type INTEGER NOT NULL, security_blob BLOB NOT NULL, expiration INTEGER NOT NULL);");
sqlite3_exec_nofail(db,
"CREATE TABLE IF NOT EXISTS metadata (key INTEGER PRIMARY KEY, data BLOB NOT NULL);");
return dbptr;
}
size_t load_securities(
const std::set<UserSymbol>& symbols,
const std::function<void(const api::Security&)>& callback) {
size_t count = 0;
auto dbptr = open_db();
// Load all existing securities and call the callback
auto db = dbptr.get();
auto select_stmt = prepare_query(db, "select key,symbol,venue,type,security_blob from securities where expiration > ?");
auto select = select_stmt.get();
// sqlite3_bind_int(select, 1, datetime::today().to_short());
sqlite3_bind_int(select, 1, 0);
auto result = sqlite3_step(select);
Security security;
while (result == SQLITE_ROW) {
security_key security_key = sqlite3_column_int(select, 0);
auto symbol = std::string(
(char*)sqlite3_column_text(select, 1),
sqlite3_column_bytes(select, 1));
auto venue = (api::Venue)sqlite3_column_int(select, 2);
auto type = (api::SecurityType)sqlite3_column_int(select, 3);
UserSymbol us(symbol, venue, type);
if (symbols.find(us) != symbols.end()) {
auto data = sqlite3_column_blob(select, 4);
auto len = sqlite3_column_bytes(select, 4);
// Migrate from old tick stuff
if (security.ParseFromArray(data, len)) {
if (security.has_glbx()) {
if (security.contract_spec().main_fraction() == 0) {
security.mutable_contract_spec()->set_main_fraction(security.glbx().do_not_use_main_fraction());
}
if (security.contract_spec().sub_fraction() == 0) {
security.mutable_contract_spec()->set_sub_fraction(security.glbx().do_not_use_sub_fraction());
}
if (security.contract_spec().price_display_format() == 0) {
security.mutable_contract_spec()->set_price_display_format(security.glbx().do_not_use_price_display_format());
}
}
bts_assert(security_key == security.id());
count++;
callback(security);
}
else {
LOG(md, Warn, "Failed to parse security from database. id = {}", security_key);
}
}
result = sqlite3_step(select);
}
return count;
}
void store_if_new(const std::vector<Security*>& securities) {
auto dbptr = open_db();
auto insert = utils::prepare_query(dbptr, "insert or ignore into securities (key, symbol, venue, type, security_blob, expiration) values (?,?,?,?,?,?)");
sqlite3_exec_nofail(dbptr, "BEGIN TRANSACTION");
std::string buffer;
for (const api::Security* psecurity : securities) {
if (psecurity && psecurity->SerializeToString(&buffer)) {
SQLITE_CHK(sqlite3_bind_int(insert.get(), 1, psecurity->id()), dbptr);
SQLITE_CHK(sqlite3_bind_text(insert.get(), 2, psecurity->symbol().c_str(), psecurity->symbol().size(), SQLITE_STATIC), dbptr);
SQLITE_CHK(sqlite3_bind_int(insert.get(), 3, psecurity->exchange()), dbptr);
SQLITE_CHK(sqlite3_bind_int(insert.get(), 4, securitytype_of(*psecurity)), dbptr);
SQLITE_CHK(sqlite3_bind_blob(insert.get(), 5, buffer.c_str(), buffer.size(), SQLITE_STATIC), dbptr);
SQLITE_CHK(sqlite3_bind_int(insert.get(), 6, expiration(*psecurity).to_short()), dbptr);
sqlite3_step(insert.get());
sqlite3_reset(insert.get());
}
}
sqlite3_exec_nofail(dbptr, "COMMIT TRANSACTION");
}
std::vector<api::synthetic::Metadata> load_metadata() {
std::vector<api::synthetic::Metadata> values;
auto dbptr = open_db();
auto db = dbptr.get();
auto select_stmt = prepare_query(db, "select key,data from metadata");
auto select = select_stmt.get();
auto result = sqlite3_step(select);
while(result == SQLITE_ROW) {
uint32_t key = sqlite3_column_int(select, 0);
auto blob = sqlite3_column_blob(select, 1);
auto len = sqlite3_column_bytes(select, 1);
api::synthetic::Metadata data;
if (data.ParseFromArray(blob, len)) {
bts_assert(key == data.security_key());
values.push_back(data);
}
else {
LOG(md, Warn, "Failed to parse security metadata for security_key={}", key);
}
result = sqlite3_step(select);
}
return values;
}
bool store_metadata(const api::synthetic::Metadata& data) {
auto dbptr = open_db();
auto insert = utils::prepare_query(dbptr, "insert or replace into metadata (key, data) values (?,?)");
SQLITE_CHK(sqlite3_bind_int(insert.get(), 1, data.security_key()), dbptr);
std::string buffer = data.SerializeAsString();
SQLITE_CHK(sqlite3_bind_blob(insert.get(), 2, buffer.c_str(), buffer.size(), SQLITE_STATIC), dbptr);
auto result = sqlite3_step(insert.get());
if (result != SQLITE_DONE) {
LOG(md, Warn, "Failed to insert metadata. sqlite result {}", result);
return false;
}
return true;
}
#else
namespace bts {
namespace md {
namespace db {
size_t load_securities(
const std::set<UserSymbol>& symbols,
const std::function<void(const api::Security&)>& callback) {
return 0;
}
void store_if_new(const std::vector<Security*>& handles) {}
// std::vector<api::synthetic::Metadata> load_metadata() {
// std::vector<api::synthetic::Metadata> result;
// return result;
// }
// void store_metadata(const api::synthetic::Metadata&) {
// }
#endif
template <class L>
datetime::Date smallest_leg(L l) {
// Iterate though all legs and choose the smallest
datetime::Date date(std::numeric_limits<uint16_t>::max(),0,0);
for (const auto& leg : l) {
auto legDate = expiration(leg.security());
if (legDate < date) {
date = legDate;
}
}
return date;
}
datetime::Date expiration(const api::Security& security) {
switch(security.Contract_case()) {
case api::Security::kFuture:
return datetime::Date(security.future().expiration_year(), security.future().expiration_month(), security.future().expiration_day());
case api::Security::kOption:
return datetime::Date(security.option().expiration_year(), security.option().expiration_month(), security.option().expiration_day());
case api::Security::kSpread:
return smallest_leg(security.spread().legs());
case api::Security::kSynthetic:
return smallest_leg(security.synthetic().legs());
default:
// return a 0 date which will always be considered expired.
return datetime::Date();
}
}
}}}//namespace