journal.hpp

#pragma once
 
#include "./clock.hpp"
#include "./execution.hpp"
 
#include "../strategy/reconcile.hpp"
 
#include <algotradelib/core>
 
#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <iterator>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
 
namespace algotradelib {
namespace runner {

enum class JournalAction {
  PLACE,
  CANCEL,
};

inline constexpr std::array<JournalAction, 2> all_journal_actions{JournalAction::PLACE, JournalAction::CANCEL};

inline constexpr std::string_view journal_action_name(JournalAction action) noexcept {
  switch (action) {
    case JournalAction::PLACE:
      return "place";
    case JournalAction::CANCEL:
      return "cancel";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_journal_actions, [](JournalAction action) { return !journal_action_name(action).empty(); }),
  "all_journal_actions must list every JournalAction"
);

inline constexpr bool journal_action_matches_by_order_id(JournalAction action) noexcept {
  switch (action) {
    case JournalAction::CANCEL:
      return true;
    case JournalAction::PLACE:
      return false;
  }
  return false;
}

inline std::optional<JournalAction> journal_action_from(std::string_view text) noexcept {
  return value_named(all_journal_actions, journal_action_name, text);
}

enum class JournalOutcome {
  PENDING,
  ACCEPTED,
  REJECTED,
  INDETERMINATE,
  RESOLVED,
  CONFIRMED,
  HELD,
};

inline constexpr std::array<JournalOutcome, 7> all_journal_outcomes{
  JournalOutcome::PENDING,
  JournalOutcome::ACCEPTED,
  JournalOutcome::REJECTED,
  JournalOutcome::INDETERMINATE,
  JournalOutcome::RESOLVED,
  JournalOutcome::CONFIRMED,
  JournalOutcome::HELD,
};

inline constexpr std::string_view journal_outcome_name(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::PENDING:
      return "pending";
    case JournalOutcome::ACCEPTED:
      return "accepted";
    case JournalOutcome::REJECTED:
      return "rejected";
    case JournalOutcome::INDETERMINATE:
      return "indeterminate";
    case JournalOutcome::RESOLVED:
      return "resolved";
    case JournalOutcome::CONFIRMED:
      return "confirmed";
    case JournalOutcome::HELD:
      return "held";
  }
  return "";
}
 
static_assert(
  lists_every_value(all_journal_outcomes, [](JournalOutcome outcome) { return !journal_outcome_name(outcome).empty(); }),
  "all_journal_outcomes must list every JournalOutcome"
);

inline std::optional<JournalOutcome> journal_outcome_from(std::string_view text) noexcept {
  return value_named(all_journal_outcomes, journal_outcome_name, text);
}

inline constexpr JournalOutcome journal_outcome_of(SendOutcome outcome) noexcept {
  switch (outcome) {
    case SendOutcome::ACCEPTED:
      return JournalOutcome::ACCEPTED;
    case SendOutcome::REJECTED:
      return JournalOutcome::REJECTED;
    case SendOutcome::INDETERMINATE:
      return JournalOutcome::INDETERMINATE;
    case SendOutcome::UNAVAILABLE:
      // Nothing was sent, so the order does not exist.
      return JournalOutcome::REJECTED;
    case SendOutcome::REFUSED_LOCALLY:
      // Nothing was sent, so the order does not exist.
      return JournalOutcome::REJECTED;
  }
  return JournalOutcome::INDETERMINATE;
}

inline constexpr bool journal_outcome_is_terminal(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::PENDING:
      return false;
    case JournalOutcome::ACCEPTED:
    case JournalOutcome::REJECTED:
    case JournalOutcome::INDETERMINATE:
    case JournalOutcome::RESOLVED:
    case JournalOutcome::CONFIRMED:
    case JournalOutcome::HELD:
      return true;
  }
  return true;
}

inline constexpr bool journal_outcome_answers_send(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::ACCEPTED:
    case JournalOutcome::REJECTED:
    case JournalOutcome::INDETERMINATE:
      return true;
    case JournalOutcome::PENDING:
    case JournalOutcome::RESOLVED:
    case JournalOutcome::CONFIRMED:
    case JournalOutcome::HELD:
      return false;
  }
  return false;
}

inline constexpr bool journal_outcome_names_order(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::ACCEPTED:
    case JournalOutcome::RESOLVED:
      return true;
    case JournalOutcome::PENDING:
    case JournalOutcome::REJECTED:
    case JournalOutcome::INDETERMINATE:
    case JournalOutcome::CONFIRMED:
    case JournalOutcome::HELD:
      return false;
  }
  return false;
}

inline constexpr bool journal_outcome_refuses(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::REJECTED:
      return true;
    case JournalOutcome::PENDING:
    case JournalOutcome::ACCEPTED:
    case JournalOutcome::INDETERMINATE:
    case JournalOutcome::RESOLVED:
    case JournalOutcome::CONFIRMED:
    case JournalOutcome::HELD:
      return false;
  }
  return false;
}

inline constexpr bool journal_outcome_is_closing_account(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::CONFIRMED:
    case JournalOutcome::HELD:
      return true;
    case JournalOutcome::PENDING:
    case JournalOutcome::ACCEPTED:
    case JournalOutcome::REJECTED:
    case JournalOutcome::INDETERMINATE:
    case JournalOutcome::RESOLVED:
      return false;
  }
  return false;
}

inline constexpr bool journal_outcome_requires_operator(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::HELD:
      return true;
    case JournalOutcome::PENDING:
    case JournalOutcome::ACCEPTED:
    case JournalOutcome::REJECTED:
    case JournalOutcome::INDETERMINATE:
    case JournalOutcome::RESOLVED:
    case JournalOutcome::CONFIRMED:
      return false;
  }
  return false;
}

inline constexpr bool journal_outcome_leaves_open(JournalOutcome outcome) noexcept {
  switch (outcome) {
    case JournalOutcome::PENDING:
    case JournalOutcome::INDETERMINATE:
      return true;
    case JournalOutcome::ACCEPTED:
    case JournalOutcome::REJECTED:
    case JournalOutcome::RESOLVED:
    case JournalOutcome::CONFIRMED:
    case JournalOutcome::HELD:
      return false;
  }
  return true;
}

struct JournalEntry {
  std::string id_;
  JournalAction action_{JournalAction::PLACE};
  JournalOutcome outcome_{JournalOutcome::PENDING};
  std::chrono::system_clock::time_point recorded_at_{};
  InstrumentId instrument_;
  OrderSide side_{OrderSide::BUY};
  OrderType type_{OrderType::LIMIT};
  TimeInForce time_in_force_{TimeInForce::DAY};
  OrderSession session_{OrderSession::REGULAR};
  std::optional<Venue> venue_{Venue::UNSPECIFIED};
  Decimal64 quantity_;
  std::optional<Decimal64> price_;
  std::optional<Decimal64> trigger_price_;
  std::optional<Decimal64> baseline_;
  std::string order_id_;
  std::string detail_;
  bool requires_operator_{false};
  std::string scope_;
};

enum class JournalReadStatus {
  MISSING,
  READ,
  UNREADABLE,
};

inline constexpr std::string_view journal_read_status_name(JournalReadStatus status) noexcept {
  switch (status) {
    case JournalReadStatus::MISSING:
      return "missing";
    case JournalReadStatus::READ:
      return "read";
    case JournalReadStatus::UNREADABLE:
      return "unreadable";
  }
  return "unknown";
}

struct JournalReplay {
  std::vector<JournalEntry> entries_;
  std::size_t unreadable_lines_{0};
  bool truncated_tail_{false};
  std::string detail_;
  std::string scope_;
  bool has_header_{false};
  JournalReadStatus status_{JournalReadStatus::READ};

  bool readable() const noexcept {
    switch (status_) {
      case JournalReadStatus::MISSING:
      case JournalReadStatus::READ:
        return true;
      case JournalReadStatus::UNREADABLE:
        return false;
    }
    return false;
  }

  bool intact() const noexcept {
    return readable() && unreadable_lines_ == 0 && !truncated_tail_;
  }
};

struct OpenInstruction {
  JournalEntry entry_;
  std::chrono::system_clock::time_point opened_at_{};
  std::optional<std::chrono::system_clock::time_point> closed_at_;
  bool answered_{false};
  std::string detail_;
};

inline std::vector<OpenInstruction> unresolved_instructions(const JournalReplay& replay) {
  std::vector<OpenInstruction> open;
  for (const auto& entry : replay.entries_) {
    if (!journal_outcome_is_terminal(entry.outcome_)) {
      OpenInstruction instruction;
      instruction.entry_ = entry;
      instruction.opened_at_ = entry.recorded_at_;
      instruction.detail_ = entry.detail_;
      open.push_back(std::move(instruction));
      continue;
    }
 
    bool paired = false;
    for (auto at = open.begin(); at != open.end(); ++at) {
      if (at->entry_.id_ != entry.id_) {
        continue;
      }
      paired = true;
      if (journal_outcome_leaves_open(entry.outcome_)) {
        at->answered_ = true;
        at->closed_at_ = entry.recorded_at_;
        at->detail_ = entry.detail_;
        // The broker issued no reference, so the placement keeps the one its pending line had.
        at->entry_.order_id_ = entry.order_id_.empty() ? at->entry_.order_id_ : entry.order_id_;
      } else {
        open.erase(at);
      }
      break;
    }
    if (!paired && journal_outcome_leaves_open(entry.outcome_)) {
      OpenInstruction instruction;
      instruction.entry_ = entry;
      instruction.opened_at_ = entry.recorded_at_;
      instruction.closed_at_ = entry.recorded_at_;
      instruction.answered_ = true;
      instruction.detail_ = entry.detail_;
      open.push_back(std::move(instruction));
    }
  }
  return open;
}

inline std::vector<OpenInstruction> placements_of(const JournalReplay& replay, const std::vector<std::string>& ids) {
  std::vector<OpenInstruction> found;
  for (const auto& entry : replay.entries_) {
    if (entry.action_ != JournalAction::PLACE || entry.requires_operator_ ||
        !journal_outcome_leaves_open(entry.outcome_) ||
        std::find(ids.begin(), ids.end(), entry.id_) == ids.end()) {
      continue;
    }
    auto known = std::find_if(found.begin(), found.end(), [&](const OpenInstruction& instruction) {
      return instruction.entry_.id_ == entry.id_;
    });
    if (known == found.end()) {
      OpenInstruction instruction;
      instruction.entry_ = entry;
      instruction.opened_at_ = entry.recorded_at_;
      instruction.detail_ = entry.detail_;
      found.push_back(std::move(instruction));
      known = std::prev(found.end());
    }
    if (journal_outcome_is_terminal(entry.outcome_)) {
      known->answered_ = true;
      known->closed_at_ = entry.recorded_at_;
      known->detail_ = entry.detail_;
    }
  }
  return found;
}

class MemoryJournal {
public:
  MemoryJournal() = default;

  explicit MemoryJournal(std::string scope) : scope_(std::move(scope)) {}

  bool append(const JournalEntry& entry) {
    entries_.push_back(entry);
    if (entries_.back().scope_.empty()) {
      entries_.back().scope_ = scope_;
    }
    return true;
  }

  JournalReplay replay() const {
    JournalReplay replay;
    replay.entries_ = entries_;
    return replay;
  }

  const std::vector<JournalEntry>& records() const noexcept { return entries_; }

  void clear() noexcept { entries_.clear(); }

  const std::string& scope() const noexcept { return scope_; }
 
private:
  std::vector<JournalEntry> entries_;
  std::string scope_;
};

template <typename Inner, typename Journal, typename Clock = SteadyClock>
class JournalingExecutor {
public:
  static constexpr bool reaches_a_broker = runner_detail::sends_to_a_broker<Inner>::value;

  JournalingExecutor(Inner inner, Journal& journal, Clock clock = {})
  : inner_(std::move(inner)), journal_(&journal), clock_(std::move(clock)) {
    prefix_ = default_prefix();
  }

  void set_id_prefix(std::string prefix) { prefix_ = std::move(prefix); }

  template <typename E = Inner>
  auto route_of(const strategy::PlannedOrder& order, int) const
    -> decltype(std::declval<const E&>().effective_route(order.venue_)) {
    return inner_.effective_route(order.venue_);
  }

  Venue route_of(const strategy::PlannedOrder& order, long) const { return order.venue_; }

  Venue effective_route_of(const strategy::PlannedOrder& order) const {
    return route_of(order, 0);
  }

  SendOutcome place(const strategy::PlannedOrder& order) { return place(order, std::nullopt); }

  SendOutcome place(const strategy::PlannedOrder& order, std::optional<Decimal64> baseline) {
    JournalEntry entry;
    entry.id_ = next_id();
    entry.action_ = JournalAction::PLACE;
    entry.outcome_ = JournalOutcome::PENDING;
    entry.recorded_at_ = clock_.now();
    entry.instrument_ = order.instrument_;
    entry.side_ = order.side_;
    entry.type_ = order.type_;
    entry.time_in_force_ = order.time_in_force_;
    entry.session_ = order.session_;
    // Where the order will actually go (an unrouted KIS order reaches KRX), asked of the binding;
    // a pure function of the order, so the line is still written before any I/O.
    entry.venue_ = effective_route_of(order);
    entry.quantity_ = order.quantity_;
    entry.price_ = order.price_;
    entry.trigger_price_ = order.trigger_price_;
    entry.baseline_ = std::move(baseline);
    if (!journal_->append(entry)) {
      return refuse_unjournaled();
    }
 
    const auto outcome = inner_.place(order);
 
    entry.outcome_ = journal_outcome_of(outcome);
    entry.recorded_at_ = clock_.now();
    entry.order_id_ = accepted_order_id(outcome);
    entry.detail_ = failure_detail(outcome);
    last_entry_ = entry;
    if (!journal_->append(entry)) {
      // The send happened; only its answer is lost, so recovery will see it unresolved. Recorded so
      // the runner stops.
      note_failure("the order journal could not record an instruction's outcome");
    }
    return outcome;
  }

  const std::optional<JournalEntry>& last_entry() const noexcept { return last_entry_; }

  bool record(JournalEntry& entry) {
    entry.recorded_at_ = clock_.now();
    if (journal_->append(entry)) {
      return true;
    }
    note_failure("the order journal could not record that a placement was accounted for");
    return false;
  }

  template <typename BrokerTag>
  SendOutcome cancel(const WorkingOrder<BrokerTag>& working) {
    JournalEntry entry;
    entry.id_ = next_id();
    entry.action_ = JournalAction::CANCEL;
    entry.outcome_ = JournalOutcome::PENDING;
    entry.recorded_at_ = clock_.now();
    entry.instrument_ = working.order_.instrument_;
    entry.side_ = working.order_.side_;
    entry.type_ = working.order_.type_;
    entry.quantity_ = working.order_.quantity_;
    entry.price_ = working.order_.price_;
    entry.trigger_price_ = working.order_.trigger_price_;
    // Known from the start: a cancellation names an existing order.
    entry.order_id_ = working.ref_.primary_;
    if (!journal_->append(entry)) {
      return refuse_unjournaled();
    }
 
    const auto outcome = inner_.cancel(working);
 
    entry.outcome_ = journal_outcome_of(outcome);
    entry.recorded_at_ = clock_.now();
    entry.detail_ = failure_detail(outcome);
    if (!journal_->append(entry)) {
      note_failure("the order journal could not record an instruction's outcome");
    }
    return outcome;
  }

  const Inner& inner() const noexcept { return inner_; }
  Inner& inner() noexcept { return inner_; }

  const Journal& journal() const noexcept { return *journal_; }

  const std::string& last_journal_error() const noexcept { return journal_error_; }

  std::size_t unjournaled_refusals() const noexcept { return refusals_; }

  std::size_t journal_failures() const noexcept { return failures_; }

  template <typename Executor = Inner>
  auto requested_route(Venue asked) const noexcept
    -> decltype(std::declval<const Executor&>().requested_route(asked)) {
    return inner_.requested_route(asked);
  }

  template <typename Executor = Inner>
  auto last_ref() const noexcept -> decltype(std::declval<const Executor&>().last_ref()) {
    return inner_.last_ref();
  }

  template <typename Executor = Inner>
  auto last_error() const noexcept -> decltype(std::declval<const Executor&>().last_error()) {
    return inner_.last_error();
  }
 
private:
  // Refuses an instruction that could not be recorded, without attempting it.
  SendOutcome refuse_unjournaled() {
    ++refusals_;
    note_failure("the order journal could not be written, so nothing was sent");
    return SendOutcome::UNAVAILABLE;
  }
 
  // Records why the journal failed, keeping the first reason and counting every failure.
  //
  // The two halves answer different questions. The reason is for a person reading a report, and the
  // first one is the useful one. The count is for the runner, which needs to know whether anything
  // has gone wrong *since* it last acknowledged a problem.
  void note_failure(std::string detail) {
    ++failures_;
    if (journal_error_.empty()) {
      journal_error_ = std::move(detail);
    }
  }
 
  // Returns a line id unique within this process and, by its prefix, across restarts.
  std::string next_id() { return prefix_ + '-' + std::to_string(++sequence_); }
 
  // Returns a prefix derived from the moment of construction, on the wall clock rather than `clock_`:
  // the prefix tells process runs apart, and an injected clock (a test's, a replay's) may start two
  // runs at the same instant.
  std::string default_prefix() const {
    const auto since = std::chrono::system_clock::now().time_since_epoch();
    const auto micros = std::chrono::duration_cast<std::chrono::microseconds>(since).count();
    return std::to_string(static_cast<std::int64_t>(micros));
  }
 
  // Returns the broker's id for an accepted order, where the inner executor reports one.
  //
  // Detected rather than required, because `RecordingExecutor` has no references to report and must
  // still be wrappable — it is what every journal test runs against.
  std::string accepted_order_id(SendOutcome outcome) const {
    switch (outcome) {
      case SendOutcome::ACCEPTED:
        return accepted_order_id_impl(0);
      case SendOutcome::REJECTED:
      case SendOutcome::INDETERMINATE:
      case SendOutcome::UNAVAILABLE:
      case SendOutcome::REFUSED_LOCALLY:
        return {};
    }
    return {};
  }
 
  template <typename Executor = Inner>
  auto accepted_order_id_impl(int) const
    -> decltype(std::declval<const Executor&>().last_ref(), std::string()) {
    const auto& ref = inner_.last_ref();
    return ref ? ref->primary_ : std::string();
  }
 
  std::string accepted_order_id_impl(...) const { return {}; }
 
  // Returns what the inner executor said went wrong, where it says anything.
  std::string failure_detail(SendOutcome outcome) const {
    switch (outcome) {
      case SendOutcome::ACCEPTED:
        return {};
      case SendOutcome::REJECTED:
      case SendOutcome::INDETERMINATE:
      case SendOutcome::UNAVAILABLE:
      case SendOutcome::REFUSED_LOCALLY:
        return failure_detail_impl(0);
    }
    return failure_detail_impl(0);
  }
 
  template <typename Executor = Inner>
  auto failure_detail_impl(int) const
    -> decltype(std::declval<const Executor&>().last_error(), std::string()) {
    const auto& error = inner_.last_error();
    return error ? error->message_ : std::string();
  }
 
  std::string failure_detail_impl(...) const { return {}; }
 
  Inner inner_;
  Journal* journal_;
  Clock clock_;
  std::optional<JournalEntry> last_entry_;
  std::string prefix_;
  std::string journal_error_;
  std::size_t refusals_{0};
  std::size_t failures_{0};
  std::uint64_t sequence_{0};
};
 
} // namespace runner
} // namespace algotradelib