series_buffer.hpp

#pragma once
 
#include "./conversion.hpp"
#include "./frame.hpp"
#include "./time_index.hpp"
 
#include "../math/matrix.hpp"
 
#include <chrono>
#include <cstddef>
#include <stdexcept>
#include <vector>
 
namespace algotradelib {
namespace analytics {

class SeriesBuffer {
public:
  using Instant = TimeIndex::Instant;

  SeriesBuffer(std::size_t capacity, std::size_t fields = bar_field_count)
  : capacity_(capacity), fields_(fields) {
    if (capacity == 0 || fields == 0) {
      throw std::invalid_argument("a series buffer needs a capacity and at least one field");
    }
    // Twice the capacity bounds compaction to one row move per push, amortized.
    values_ = math::unobserved(capacity * 2, fields);
    stamps_.resize(capacity * 2);
  }

  std::size_t capacity() const noexcept { return capacity_; }

  std::size_t fields() const noexcept { return fields_; }

  std::size_t size() const noexcept { return count_; }

  bool empty() const noexcept { return count_ == 0; }

  bool full() const noexcept { return count_ == capacity_; }

  std::size_t compactions() const noexcept { return compactions_; }

  void push(Instant when, math::SeriesView row) {
    if (static_cast<std::size_t>(row.size()) != fields_) {
      throw std::invalid_argument("a series buffer observation must carry every field");
    }
    if (count_ > 0 && !(newest_instant() < when)) {
      throw std::invalid_argument("a series buffer observation must be later than the last");
    }
 
    if (count_ == capacity_) {
      ++begin_;
      --count_;
    }
    if (begin_ + count_ == static_cast<std::size_t>(values_.rows())) {
      compact();
    }
 
    const auto row_index = static_cast<math::Index>(begin_ + count_);
    for (std::size_t field = 0; field < fields_; ++field) {
      values_(row_index, static_cast<math::Index>(field)) =
        row(static_cast<math::Index>(field));
    }
    stamps_[begin_ + count_] = when;
    ++count_;
  }

  void push(Instant when, const std::vector<Scalar>& row) {
    if (row.size() != fields_) {
      throw std::invalid_argument("a series buffer observation must carry every field");
    }
    const math::Vector values = math::Vector::Map(row.data(), static_cast<math::Index>(row.size()));
    push(when, values);
  }

  void push(const Bar& bar) {
    if (fields_ != bar_field_count) {
      throw std::invalid_argument("pushing a bar needs a buffer over every bar field");
    }
    math::Vector row = math::unobserved(bar_field_count);
    if (bar.open_) {
      row(static_cast<math::Index>(bar_field_index(BarField::OPEN))) = to_scalar(*bar.open_);
    }
    row(static_cast<math::Index>(bar_field_index(BarField::HIGH))) = to_scalar(bar.high_);
    row(static_cast<math::Index>(bar_field_index(BarField::LOW))) = to_scalar(bar.low_);
    row(static_cast<math::Index>(bar_field_index(BarField::CLOSE))) = to_scalar(bar.close_);
    row(static_cast<math::Index>(bar_field_index(BarField::VOLUME))) = to_scalar(bar.volume_);
    if (bar.traded_value_) {
      row(static_cast<math::Index>(bar_field_index(BarField::TRADED_VALUE))) =
        to_scalar(*bar.traded_value_);
    }
    push(bar.timestamp_, row);
  }

  math::SeriesView column(std::size_t field) const {
    if (field >= fields_) {
      throw std::out_of_range("series buffer field is past the end");
    }
    return values_.col(static_cast<math::Index>(field))
      .segment(static_cast<math::Index>(begin_), static_cast<math::Index>(count_));
  }

  math::SeriesView column(BarField field) const { return column(bar_field_index(field)); }

  Scalar at(std::size_t position, std::size_t field) const {
    if (position >= count_ || field >= fields_) {
      throw std::out_of_range("series buffer position is past the end");
    }
    return values_(
      static_cast<math::Index>(begin_ + position), static_cast<math::Index>(field)
    );
  }

  Instant instant(std::size_t position) const {
    if (position >= count_) {
      throw std::out_of_range("series buffer position is past the end");
    }
    return stamps_[begin_ + position];
  }

  Instant newest_instant() const {
    if (count_ == 0) {
      throw std::out_of_range("an empty series buffer has no newest observation");
    }
    return stamps_[begin_ + count_ - 1];
  }

  void clear() noexcept {
    begin_ = 0;
    count_ = 0;
  }
 
private:
  // Moves the held observations to the front. Source and destination never overlap: this runs
  // only when the held run reaches the end of storage, so begin_ = 2 * capacity - count_ >= count_.
  void compact() {
    const auto rows = static_cast<math::Index>(count_);
    const auto from = static_cast<math::Index>(begin_);
    values_.topRows(rows) = values_.middleRows(from, rows);
    for (std::size_t position = 0; position < count_; ++position) {
      stamps_[position] = stamps_[begin_ + position];
    }
    begin_ = 0;
    ++compactions_;
  }
 
  std::size_t capacity_{0};
  std::size_t fields_{0};
  std::size_t begin_{0};
  std::size_t count_{0};
  std::size_t compactions_{0};
  math::Matrix values_;
  std::vector<Instant> stamps_;
};
 
} // namespace analytics
} // namespace algotradelib