VER: Release 0.6.0

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.6.0 - 2023-03-24
+- Added support for imbalance schema
+- Added support for decoding `ts_out` field
+- Removed `record_count` from `Metadata`
+- Changed `Historical::BatchDownload` to return the paths of the downloaded files
+- Added flags `kSnapshot` and `kMaybeBadBook`
+
 ## 0.5.0 - 2023-03-13
 - Added `Historical::MetadataGetDatasetRange`
 - Changed `MetadataGetDatasetCondition` to return `vector<DatasetConditionDetail>`

CMakeLists.txt

@@ -4,7 +4,7 @@ cmake_minimum_required(VERSION 3.14)
 # Project details
 #
 
-project("databento" VERSION 0.5.0 LANGUAGES CXX)
+project("databento" VERSION 0.6.0 LANGUAGES CXX)
 string(TOUPPER ${PROJECT_NAME} PROJECT_NAME_UPPERCASE)
 
 #

README.md

@@ -80,7 +80,7 @@ cmake --install build
 
 Then in your project's `CMakeLists.txt`, add the following:
 ```cmake
-find_package(databento 0.5.0 REQUIRED)
+find_package(databento 0.6.0 REQUIRED)
 add_library(my_library)
 target_link_libraries(my_library PRIVATE databento::databento)
 ```

cmake/SourcesAndHeaders.cmake

@@ -17,6 +17,7 @@ set(headers
   include/databento/record.hpp
   include/databento/symbology.hpp
   include/databento/timeseries.hpp
+  include/databento/with_ts_out.hpp
   include/databento/detail/file_stream.hpp
   include/databento/detail/http_client.hpp
   include/databento/detail/scoped_fd.hpp

include/databento/dbn.hpp

@@ -40,8 +40,6 @@ struct Metadata {
   UnixNanos end;
   // The maximum number of records for the query.
   std::uint64_t limit;
-  // The total number of records.
-  std::uint64_t record_count;
   // The input symbology type.
   SType stype_in;
   // The output symbology type.
@@ -71,10 +69,9 @@ inline bool operator==(const Metadata& lhs, const Metadata& rhs) {
   return lhs.version == rhs.version && lhs.dataset == rhs.dataset &&
          lhs.schema == rhs.schema && lhs.start == rhs.start &&
          lhs.end == rhs.end && lhs.limit == rhs.limit &&
-         lhs.record_count == rhs.record_count && lhs.stype_in == rhs.stype_in &&
-         lhs.stype_out == rhs.stype_out && lhs.symbols == rhs.symbols &&
-         lhs.partial == rhs.partial && lhs.not_found == rhs.not_found &&
-         lhs.mappings == rhs.mappings;
+         lhs.stype_in == rhs.stype_in && lhs.stype_out == rhs.stype_out &&
+         lhs.symbols == rhs.symbols && lhs.partial == rhs.partial &&
+         lhs.not_found == rhs.not_found && lhs.mappings == rhs.mappings;
 }
 
 std::string ToString(const Metadata& metadata);

include/databento/dbn_decoder.hpp

@@ -29,8 +29,9 @@ class DbnDecoder {
 
   // Should only be called once
   Metadata DecodeMetadata();
-  // Lifetime of returned Record is until next call to ParseRecord.
-  Record DecodeRecord();
+  // Lifetime of returned Record is until next call to DecodeRecord. Returns
+  // nullptr once the end of the input has been reached.
+  const Record* DecodeRecord();
 
  private:
   static std::string DecodeSymbol(
@@ -51,5 +52,6 @@ class DbnDecoder {
   std::unique_ptr<IReadable> input_;
   std::vector<std::uint8_t> buffer_;
   std::size_t buffer_idx_{};
+  Record current_record_{nullptr};
 };
 }  // namespace databento

include/databento/enums.hpp

@@ -31,6 +31,7 @@ enum class Schema : std::uint16_t {
   Definition = 9,
   Statistics = 10,
   Status = 11,
+  Imbalance = 12,
 };
 
 // Represents a data output encoding.
@@ -105,6 +106,7 @@ enum RType : std::uint8_t {
   Ohlcv1H = 0x22,
   Ohlcv1D = 0x23,
   InstrumentDef = 0x13,
+  Imbalance = 0x14,
   Error = 0x15,
   SymbolMapping = 0x16,
   Mbo = 0xA0,

include/databento/flag_set.hpp

@@ -6,16 +6,23 @@
 #include <string>
 
 namespace databento {
-/// Transparent wrapper around the bit flags used in several DBN record types.
+// Transparent wrapper around the bit flags used in several DBN record types.
 class FlagSet {
  public:
   using Repr = std::uint8_t;
-  // Last message in the packet from the venue for a given `product_id`.
+  // Indicates it's the last message in the packet from the venue for a given
+  // `product_id`.
   static constexpr Repr kLast = 1 << 7;
-  // Aggregated price level message, not an individual order.
+  // Indicates the message was sourced from a replay, such as a snapshot
+  // server.
+  static constexpr Repr kSnapshot = 1 << 5;
+  // Indicates an aggregated price level message, not an individual order.
   static constexpr Repr kMbp = 1 << 4;
-  // The `ts_recv` value is inaccurate due to clock issues or packet reordering.
+  // Indicates the `ts_recv` value is inaccurate due to clock issues or packet
+  // reordering.
   static constexpr Repr kBadTsRecv = 1 << 3;
+  // Indicates an unrecoverable gap was detected in the channel.
+  static constexpr Repr kMaybeBadBook = 1 << 2;
 
   friend std::ostream& operator<<(std::ostream&, FlagSet);
 

include/databento/historical.hpp

@@ -62,9 +62,13 @@ class Historical {
   std::vector<BatchJob> BatchListJobs(const std::vector<JobState>& states,
                                       const std::string& since);
   std::vector<BatchFileDesc> BatchListFiles(const std::string& job_id);
-  void BatchDownload(const std::string& output_dir, const std::string& job_id);
-  void BatchDownload(const std::string& output_dir, const std::string& job_id,
-                     const std::string& filename_to_download);
+  // Returns the paths of the downloaded files.
+  std::vector<std::string> BatchDownload(const std::string& output_dir,
+                                         const std::string& job_id);
+  // Returns the path of the downloaded file.
+  std::string BatchDownload(const std::string& output_dir,
+                            const std::string& job_id,
+                            const std::string& filename_to_download);
 
   /*
    * Metadata API

include/databento/record.hpp

@@ -13,6 +13,8 @@
 namespace databento {
 // Common data for all Databento Records.
 struct RecordHeader {
+  static constexpr std::size_t kLengthMultiplier = 4;
+
   // The length of the message in 32-bit words.
   std::uint8_t length;
   // The record type.
@@ -232,6 +234,37 @@ struct InstrumentDefMsg {
 static_assert(sizeof(InstrumentDefMsg) == 360,
               "InstrumentDefMsg size must match C");
 
+// An order imbalance message.
+struct ImbalanceMsg {
+  static bool HasRType(RType rtype) { return rtype == RType::Imbalance; }
+
+  RecordHeader hd;
+  UnixNanos ts_recv;
+  std::int64_t ref_price;
+  UnixNanos auction_time;
+  std::int64_t cont_book_clr_price;
+  std::int64_t auct_interest_clr_price;
+  std::int64_t ssr_filling_price;
+  std::int64_t ind_match_price;
+  std::int64_t upper_collar;
+  std::int64_t lower_collar;
+  std::uint32_t paired_qty;
+  std::uint32_t total_imbalance_qty;
+  std::uint32_t market_imbalance_qty;
+  std::uint32_t unpaired_qty;
+  char auction_type;
+  Side side;
+  std::uint8_t auction_status;
+  std::uint8_t freeze_status;
+  std::uint8_t num_extensions;
+  Side unpaired_side;
+  char significant_imbalance;
+  // padding for alignment
+  std::array<char, 1> dummy[1];
+};
+
+static_assert(sizeof(ImbalanceMsg) == 112, "ImbalanceMsg size must match C");
+
 // An error message from the Live Subscription Gateway (LSG). This will never
 // be present in historical data.
 struct ErrorMsg {
@@ -335,6 +368,11 @@ inline bool operator!=(const InstrumentDefMsg& lhs,
   return !(lhs == rhs);
 }
 
+bool operator==(const ImbalanceMsg& lhs, const ImbalanceMsg& rhs);
+inline bool operator!=(const ImbalanceMsg& lhs, const ImbalanceMsg& rhs) {
+  return !(lhs == rhs);
+}
+
 inline bool operator==(const ErrorMsg& lhs, const ErrorMsg& rhs) {
   return lhs.hd == rhs.hd && lhs.err == rhs.err;
 }
@@ -366,6 +404,9 @@ std::ostream& operator<<(std::ostream& stream, const OhlcvMsg& ohlcv_msg);
 std::string ToString(const InstrumentDefMsg& instr_def_msg);
 std::ostream& operator<<(std::ostream& stream,
                          const InstrumentDefMsg& instr_def_msg);
+std::string ToString(const ImbalanceMsg& imbalance_msg);
+std::ostream& operator<<(std::ostream& stream,
+                         const ImbalanceMsg& imbalance_msg);
 std::string ToString(const ErrorMsg& err_msg);
 std::ostream& operator<<(std::ostream& stream, const ErrorMsg& err_msg);
 std::string ToString(const SymbolMappingMsg& symbol_mapping_msg);

include/databento/with_ts_out.hpp

@@ -0,0 +1,27 @@
+#pragma once
+
+#include "databento/datetime.hpp"  // UnixNanos
+#include "databento/enums.hpp"
+
+namespace databento {
+// Record wrapper to read records with their live gateway send
+// timestamp (ts_out).
+template <typename R>
+struct WithTsOut {
+  static bool HasRType(RType rtype) { return R::HasRType(rtype); }
+
+  // The base record.
+  R rec;
+  // The end timestamp from the Databento live gateway.
+  UnixNanos ts_out;
+};
+
+template <typename R>
+inline bool operator==(const WithTsOut<R>& lhs, const WithTsOut<R>& rhs) {
+  return lhs.rec == rhs.rec && lhs.ts_out == rhs.ts_out;
+}
+template <typename R>
+inline bool operator!=(const WithTsOut<R>& lhs, const WithTsOut<R>& rhs) {
+  return !(lhs == rhs);
+}
+}  // namespace databento

pkg/PKGBUILD

@@ -1,7 +1,7 @@
 # Maintainer: Databento <[email protected]>
 _pkgname=databento-cpp
 pkgname=databento-cpp-git
-pkgver=0.5.0
+pkgver=0.6.0
 pkgrel=1
 pkgdesc="Official C++ client for Databento"
 arch=('any')

src/dbn.cpp

@@ -19,7 +19,6 @@ std::ostream& operator<<(std::ostream& stream, const Metadata& metadata) {
                     .AddField("start", metadata.start)
                     .AddField("end", metadata.end)
                     .AddField("limit", metadata.limit)
-                    .AddField("record_count", metadata.record_count)
                     .AddField("stype_in", metadata.stype_in)
                     .AddField("stype_out", metadata.stype_out);
 

src/dbn_decoder.cpp

@@ -105,7 +105,8 @@ databento::Metadata DbnDecoder::DecodeMetadataFields(
   res.end =
       UnixNanos{std::chrono::nanoseconds{Consume<std::uint64_t>(buffer_it)}};
   res.limit = Consume<std::uint64_t>(buffer_it);
-  res.record_count = Consume<std::uint64_t>(buffer_it);
+  // skip deprecated record_count
+  buffer_it += 8;
   res.stype_in = static_cast<SType>(Consume<std::uint8_t>(buffer_it));
   res.stype_out = static_cast<SType>(Consume<std::uint8_t>(buffer_it));
   // skip reserved
@@ -136,23 +137,23 @@ databento::Metadata DbnDecoder::DecodeMetadata() {
 }
 
 // assumes ParseMetadata has been called
-databento::Record DbnDecoder::DecodeRecord() {
+const databento::Record* DbnDecoder::DecodeRecord() {
   // need some unread unread_bytes
   const auto unread_bytes = buffer_.size() - buffer_idx_;
   if (unread_bytes == 0) {
     if (FillBuffer() == 0) {
-      throw DbnResponseError{"Reached end of DBN stream"};
+      return nullptr;
     }
   }
   // check length
   while (buffer_.size() - buffer_idx_ < BufferRecordHeader()->Size()) {
     if (FillBuffer() == 0) {
-      throw DbnResponseError{"Reached end of DBN stream"};
+      return nullptr;
     }
   }
-  Record res{BufferRecordHeader()};
-  buffer_idx_ += res.Size();
-  return res;
+  current_record_ = Record{BufferRecordHeader()};
+  buffer_idx_ += current_record_.Size();
+  return &current_record_;
 }
 
 size_t DbnDecoder::FillBuffer() {

src/dbn_file_store.cpp

@@ -12,13 +12,12 @@ DbnFileStore::DbnFileStore(const std::string& file_path)
 void DbnFileStore::Replay(const MetadataCallback& metadata_callback,
                           const RecordCallback& record_callback) {
   auto metadata = parser_.DecodeMetadata();
-  const auto record_count = metadata.record_count;
   if (metadata_callback) {
     metadata_callback(std::move(metadata));
   }
-  for (std::size_t i = 0; i < record_count; ++i) {
-    const auto record = parser_.DecodeRecord();
-    if (record_callback(record) == KeepGoing::Stop) {
+  const databento::Record* record;
+  while ((record = parser_.DecodeRecord()) != nullptr) {
+    if (record_callback(*record) == KeepGoing::Stop) {
       break;
     }
   }

src/enums.cpp

@@ -53,6 +53,9 @@ const char* ToString(Schema schema) {
     case Schema::Definition: {
       return "definition";
     }
+    case Schema::Imbalance: {
+      return "imbalance";
+    }
     case Schema::Statistics: {
       return "statistics";
     }
@@ -250,6 +253,9 @@ const char* ToString(RType rtype) {
     case RType::InstrumentDef: {
       return "InstrumentDef";
     }
+    case RType::Imbalance: {
+      return "Imbalance";
+    }
     case RType::Mbo: {
       return "Mbo";
     }
@@ -396,6 +402,9 @@ Schema FromString(const std::string& str) {
   if (str == "definition") {
     return Schema::Definition;
   }
+  if (str == "imbalance") {
+    return Schema::Imbalance;
+  }
   if (str == "statistics") {
     return Schema::Statistics;
   }