live_program.hpp

Go to the documentation of this file.

#pragma once

 
#include <algotradelib/live>
#include <algotradelib/runner>
 
#include <chrono>
#include <cstddef>
#include <ostream>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
 
namespace live_program {
 
namespace atl = algotradelib;
namespace live = algotradelib::live;
namespace runner = algotradelib::runner;
namespace strategy = algotradelib::strategy;

struct Settings {
  std::vector<atl::InstrumentId> universe_;
  strategy::RiskLimits limits_;
  runner::PollSchedule schedule_;
  atl::BarInterval bar_interval_{atl::BarInterval::ONE_DAY};
  int bar_count_{100};
  std::chrono::nanoseconds maximum_portfolio_age_{std::chrono::minutes(2)};
  std::chrono::nanoseconds maximum_price_limits_age_{std::chrono::minutes(20)};
  std::string journal_path_;
  std::string journal_scope_;
  std::string control_file_;
  std::size_t max_iterations_{0};
  std::chrono::nanoseconds closing_wait_{std::chrono::seconds(30)};
  std::chrono::nanoseconds owed_order_limit_{std::chrono::minutes(10)};
  atl::Venue default_route_{atl::Venue::UNSPECIFIED};
};

enum class Exit : int {
  CLEAN = 0,
  CLOSING_INCOMPLETE = 1,
  JOURNAL_REFUSED = 2,
  RECOVERY_HALTED = 3,
  REFERENCE_UNREADABLE = 4,
  LOOP_ABANDONED = 5,
  LOCAL_FAILURE = 6,
  ACCOUNT_UNRESOLVED = 7,
  OWED_ORDER = 8,
};

template <typename Session, typename Runner>
bool load_reference_data(Session& session, Runner& bound, const Settings& settings, std::ostream& out) {
  for (const auto& instrument : settings.universe_) {
    auto reference = session.instrument(instrument);
    if (!reference) {
      out << "could not read reference data for " << instrument.symbol_ << ": "
          << reference.error().message_ << '\n';
      return false;
    }
    bound.context().set_instrument(std::move(reference.value()));
  }
  return true;
}

inline void print_recovery(const runner::RecoveryReport& recovery, std::ostream& out) {
  if (recovery.clean()) {
    out << "recovery: clean -- the journal owes nothing\n";
    return;
  }
  out << "recovery: NOT clean\n";
  for (const auto& instrument : recovery.marked_) {
    out << "  withheld until the broker's lists resolve it: " << instrument.symbol_ << '\n';
  }
  if (recovery.truncated_) {
    out << "  the last line was torn: the previous process died while writing it\n";
  }
  if (recovery.unreadable_lines_ > 0) {
    out << "  " << recovery.unreadable_lines_ << " line(s) this library did not write\n";
  }
  if (recovery.unreadable_file_) {
    out << "  the journal exists and could not be read\n";
  }
  if (recovery.unattributable_) {
    out << "  an unresolved instruction names no instrument\n";
  }
  if (!recovery.detail_.empty()) {
    out << "  detail: " << recovery.detail_ << '\n';
  }
}

inline void print_how_to_stop(const Settings& settings, std::ostream& out) {
  out << "to stop: Ctrl-C or SIGTERM (quiesce), or\n"
      << "  echo quiesce  > " << settings.control_file_ << "   # stop placing, leave orders resting\n"
      << "  echo withdraw > " << settings.control_file_ << "   # cancel the account's working orders\n"
      << "  echo flatten  > " << settings.control_file_ << "   # withdraw, then close the account's positions\n"
      << "the file is not deleted: a word left in it is carried out at the next start; remove it\n"
      << "before trading again\n";
  if (settings.closing_wait_ > std::chrono::nanoseconds::zero()) {
    out << "a withdraw or a flatten may first wait up to "
        << std::chrono::duration_cast<std::chrono::seconds>(settings.closing_wait_).count()
        << " s for the order list and holdings to account for this\n"
        << "program's own orders. ";
  }
  out << "A supervisor's grace period must cover that stop: the wait, the calls in\n"
      << "flight (a paginated read, a paced request) and the closing orders themselves, with margin; a\n"
      << "SIGTERM does not turn a pending flatten into a quiesce\n";
}

inline void print_loop(const runner::LoopReport& report, std::ostream& out) {
  out << "\niterations      : " << report.iterations_ << '\n'
      << "order list read : " << (report.order_list_read_ ? "yes" : "no") << '\n'
      << "polls failed    : bars " << report.failed(runner::PollKind::BARS) << '/'
      << report.attempted(runner::PollKind::BARS) << ", holdings "
      << report.failed(runner::PollKind::HOLDINGS) << '/' << report.attempted(runner::PollKind::HOLDINGS)
      << ", orders " << report.failed(runner::PollKind::WORKING_ORDERS) << '/'
      << report.attempted(runner::PollKind::WORKING_ORDERS) << '\n'
      << "blocked         : " << report.blocked_ << " iteration(s)\n"
      << "stopped by      : "
      << (report.stopped_by_ ? runner::stop_intent_name(*report.stopped_by_) : std::string_view("nothing"))
      << '\n'
      << "final gate      : "
      << (report.final_gate_ ? strategy::halt_reason_name(*report.final_gate_) : std::string_view("none"))
      << '\n'
      << "market phase    : " << strategy::market_phase_name(report.final_phase_) << '\n'
      // Every way an order the strategy wanted did not go out, each with its own count, so an empty
      // placement count explains itself.
      << "plans withheld  : " << report.plans_withheld_ << " (account unread, too old or older than the order listing, or a hold while it would cancel)\n"
      << "not buildable   : " << report.orders_unbuildable_ << '\n'
      << "withheld        : " << report.orders_reconcile_withheld_ << " (unresolved earlier orders)\n"
      << "refused by guard: " << report.orders_guard_refused_ << '\n'
      << "refused locally : " << report.orders_binding_refused_ << " (by the adapter, before sending)\n"
      << "refused by broker: " << report.orders_broker_refused_ << '\n'
      << "unknown outcome : " << report.orders_indeterminate_ << " (read the broker's order list)\n"
      // Cancellations that did not go through, which no line above counts. Accepted ones are not here.
      << "cancels         : " << report.cancels_rejected_ << " rejected (by the broker or the adapter), "
      << report.cancels_indeterminate_ << " of unknown outcome, " << report.cancels_refused_
      << " refused by a hold, " << report.cancels_unresolved_ << " for an order not in the picture\n"
      << "unconfirmed     : " << report.orders_unconfirmed_timed_out_
      << " (sent, and not confirmed in time: its instrument waits for holdings or its history, or for you if "
         "it was never named or the runner overlaps orders)\n";
  if (report.last_unbuildable_) {
    out << "last unbuildable: " << strategy::reject_reason_name(report.last_unbuildable_->reason_) << " for "
        << report.last_unbuildable_->instrument_.symbol_
        << (report.last_unbuildable_->tag_.empty() ? "" : " (" + report.last_unbuildable_->tag_ + ")") << '\n';
  }
  if (report.last_broker_refusal_) {
    out << "last refusal    : " << report.last_broker_refusal_->code_ << ' '
        << report.last_broker_refusal_->message_ << '\n';
  }
  if (report.abandoned_) {
    out << "ABANDONED       : after repeated failures polling "
        << (report.abandoned_kind_ ? runner::poll_kind_name(*report.abandoned_kind_) : std::string_view("?"))
        << '\n';
  }
  for (const auto& note : report.notes_) {
    out << "note            : " << note << '\n';
  }
}

inline void print_report(const runner::LoopReport& report, std::ostream& out) {
  print_loop(report, out);
  out << "closing complete: " << (report.closing_complete() ? "yes" : "NO")
      << " -- whether every closing instruction was taken, not whether the account is flat;\n"
      << "                  read the broker's order list and holdings to learn that\n";
}

inline bool stop_closes(const runner::LoopReport& report) noexcept {
  return report.stopped_by_ && *report.stopped_by_ != runner::StopIntent::QUIESCE;
}

template <typename Loop>
void configure(Loop& loop, const Settings& settings) {
  loop.set_maximum_portfolio_age(settings.maximum_portfolio_age_);
  loop.set_maximum_price_limits_age(settings.maximum_price_limits_age_);
  loop.set_closing_wait(settings.closing_wait_);
  loop.set_owed_order_limit(settings.owed_order_limit_);
  loop.set_control_file(runner::FileStopChannel(settings.control_file_));
}

template <typename Strategy, typename Session, typename MarketData, typename Clock = runner::SteadyClock>
Exit trade(
  Session& session,
  MarketData& market_data,
  Strategy strategy_instance,
  const Settings& settings,
  runner::StopSource& stop,
  std::ostream& out,
  Clock clock = {}
) {
  runner::FileJournal journal(settings.journal_path_, settings.journal_scope_);
  if (!journal.open()) {
    out << "journal refused: " << journal.rejection() << '\n'
        << "a runner that cannot record what it sends does not send\n";
    return Exit::JOURNAL_REFUSED;
  }
 
  // One clock for the loop and the journal: lines are stamped, and recovery's sweep asks the broker
  // about, the times the loop runs at, which are not the wall clock's in a paper run or a test.
  runner::ClockReference<Clock> shared_clock(clock);
  auto bound = live::bind_journaled(
    session, journal, std::move(strategy_instance), settings.limits_, live::default_series_capacity,
    settings.default_route_, shared_clock
  );
 
  const auto recovery = runner::recover_runner(bound, journal);
  print_recovery(recovery, out);
  if (const auto failure = bound.local_failure()) {
    // Damage that names no single instrument halts everything, and restarting will find the same
    // line. There is deliberately no flag to trade past it: a flag that is needed on every restart
    // ends up in a unit file. The way out is by hand, and it keeps the evidence.
    out << "HALTED: " << *failure << '\n'
        << "nothing was started. This program's only way on gives up the whole journal. That is safe\n"
        << "only once no order the journal may track can execute any more:\n"
        << "  1. read the broker's order list, order history and holdings for this account yourself.\n"
        << "     A damaged journal may not name every order it tracks: take every order sent on this\n"
        << "     account since the journal began, on every day and every page of the order history, and\n"
        << "     every order still working;\n"
        << "  2. each must have ended: filled in full, or cancelled with its history row saying so. A\n"
        << "     cancellation only requested, an order list that no longer shows it, and the close of the\n"
        << "     session do not show that an order has ended;\n"
        << "  3. holdings must show the final fills of each;\n"
        << "  4. only then move the journal aside, keeping it:  mv " << settings.journal_path_ << ' '
        << settings.journal_path_ << ".damaged-$(date +%Y%m%d)\n"
        << "the next start then begins a new journal and sizes from the holdings and working orders the\n"
        << "broker reports. If steps 2 and 3 cannot be shown for every order, leave the journal where it\n"
        << "is. Do NOT move it aside while an order it may track is left working: a new journal does not\n"
        << "know the order, and when it fills before holdings show it, the order is sent again.\n";
    return Exit::RECOVERY_HALTED;
  }
 
  if (!load_reference_data(market_data, bound, settings, out)) {
    return Exit::REFERENCE_UNREADABLE;
  }
  // Said before trading as well as in the report at the end, so it is read before it matters.
  if constexpr (decltype(bound)::on_fill_never_fires) {
    out << "note: this strategy defines on_fill, which never fires on a live runner; anything it\n"
        << "      does there (a protective stop, say) will not happen. See writing-a-strategy.md.\n";
  }
 
  live::SessionSource<Session, runner::FileJournal, MarketData> source(
    session, market_data, settings.universe_
  );
  source.set_journal(journal);
  source.set_bar_query(settings.bar_interval_, settings.bar_count_);
 
  runner::PollingLoop<decltype(bound), decltype(source), runner::ClockReference<Clock>> loop(
    bound, source, settings.schedule_, shared_clock
  );
  configure(loop, settings);
 
  print_how_to_stop(settings, out);
  const auto report = loop.run(stop, settings.max_iterations_);
  print_report(report, out);
 
  // Checked before abandonment: sticky, set by a failed journal write as much as by recovery, and
  // the more urgent of the two -- what the runner was sending at that moment is unknown. A run can
  // end both ways, and the report above still says ABANDONED.
  if (const auto failure = bound.local_failure()) {
    out << "LOCAL FAILURE   : " << *failure << '\n'
        << "                  read the broker's order list before starting again\n";
    return Exit::LOCAL_FAILURE;
  }
  if (report.abandoned_) {
    return Exit::LOOP_ABANDONED;
  }
  if (stop_closes(report) && !report.closing_complete()) {
    return Exit::CLOSING_INCOMPLETE;
  }
  if (report.owed_stop_) {
    out << "OWED ORDER      : the run stopped placing by itself (see the note above). A supervisor must not\n"
        << "                  restart it: the order is restored from the journal and the run stops again.\n"
        << "this journal still owes:\n";
    for (const auto& owed : bound.obligations()) {
      out << "  " << owed.instrument_.symbol_ << ' ' << (owed.side_ == atl::OrderSide::BUY ? "buy " : "sell ")
          << owed.quantity_.str() << ", sent " << atl::format_iso8601_utc(owed.sent_at_) << ", "
          << (owed.state_ == runner::ObligationState::OPERATOR ? "held for an operator" :
              owed.state_ == runner::ObligationState::UNNAMED  ? "no order found" :
                                                                 "order " + owed.order_id_ + ", not yet in holdings")
          << '\n';
    }
    out << "this program has no command to release one order, and its only way on gives up the whole\n"
        << "journal. That is safe only once none of the orders above can execute any more:\n"
        << "  1. read the broker's order list, order history and holdings for every order above;\n"
        << "  2. each must have ended: filled in full, or cancelled with its history row saying so. A\n"
        << "     cancellation only requested, an order list that no longer shows it, and the close of the\n"
        << "     session do not show that an order has ended;\n"
        << "  3. holdings must show the final fills of each;\n"
        << "  4. only then move the journal aside, keeping it:  mv " << settings.journal_path_ << ' '
        << settings.journal_path_ << ".owed-$(date +%Y%m%d)\n"
        << "the next start then begins a new journal and sizes from the account as it is.\n"
        << "Do NOT move the journal aside while an order it tracks is left working: a new journal does not\n"
        << "know the order, and when it fills before holdings show it, the order is sent again. A program\n"
        << "of your own releases that instrument's obligations instead (Runner::release_obligations; see\n"
        << "set_owed_order_limit) and keeps the rest of the journal.\n";
    return Exit::OWED_ORDER;
  }
  return Exit::CLEAN;
}

template <typename Strategy, typename Session, typename Clock = runner::SteadyClock>
Exit trade(
  Session& session,
  Strategy strategy_instance,
  const Settings& settings,
  runner::StopSource& stop,
  std::ostream& out,
  Clock clock = {}
) {
  return trade(session, session, std::move(strategy_instance), settings, stop, out, std::move(clock));
}

template <typename Strategy, typename Session, typename MarketData, typename Clock = runner::SteadyClock>
Exit observe(
  Session& session,
  MarketData& market_data,
  Strategy strategy_instance,
  const Settings& settings,
  runner::StopSource& stop,
  std::ostream& out,
  Clock clock = {}
) {
  auto bound = live::bind_observer(session, std::move(strategy_instance), settings.limits_);
  if (!load_reference_data(market_data, bound, settings, out)) {
    return Exit::REFERENCE_UNREADABLE;
  }
 
  live::SessionSource<Session, runner::MemoryJournal, MarketData> source(
    session, market_data, settings.universe_
  );
  source.set_bar_query(settings.bar_interval_, settings.bar_count_);
  runner::PollingLoop<decltype(bound), decltype(source), Clock> loop(
    bound, source, settings.schedule_, std::move(clock)
  );
  configure(loop, settings);
 
  out << "observing: nothing can be sent (pass --live to trade)\n";
  print_how_to_stop(settings, out);
  const auto report = loop.run(stop, settings.max_iterations_);
  print_loop(report, out);
  // `closing_complete()` is deliberately not printed: an observer's executor accepts every
  // instruction without sending it, so "every closing instruction was taken" would be true and would
  // say nothing about the broker.
  out << "closing         : simulated -- the observer sent nothing, so nothing at the broker was\n"
      << "                  placed or withdrawn by this run\n";
  return report.abandoned_ ? Exit::LOOP_ABANDONED : Exit::CLEAN;
}

template <typename Strategy, typename Session, typename Clock = runner::SteadyClock>
Exit observe(
  Session& session,
  Strategy strategy_instance,
  const Settings& settings,
  runner::StopSource& stop,
  std::ostream& out,
  Clock clock = {}
) {
  return observe(session, session, std::move(strategy_instance), settings, stop, out, std::move(clock));
}
 
} // namespace live_program