market_arbitrage.hpp

// A market arbitrage (latency) trading agent.
// Copyright 2020 Christian Kauten
//
// Author: Christian Kauten (kautenja@auburn.edu)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
//
 
#ifndef STRATEGIES_MARKET_ARBITRAGE_HPP
#define STRATEGIES_MARKET_ARBITRAGE_HPP
 
#include "data_feed/receiver.hpp"
#include "order_entry/client.hpp"
#include "maths/probability.hpp"
#include <nlohmann/json.hpp>
#include <atomic>
#include <iostream>
 
namespace Strategies {
 
class MarketArbitrage {
 private:
    typedef DataFeed::Receiver<MarketArbitrage> DataFeedReceiver;
    DataFeedReceiver receiverA;
    DataFeedReceiver receiverB;
    OrderEntry::Client clientA;
    OrderEntry::Client clientB;
    asio::steady_timer timer;
    uint32_t sleep_time;
    float P_act;
    std::atomic<bool> is_running;
 
    OrderEntry::Quantity size;
    DataFeed::Quantity spread;
 
    inline void start_strategy() {
        timer.expires_after(std::chrono::milliseconds(sleep_time));
        timer.async_wait([this](const std::error_code& error) {
            if (error) {  // TODO: handle error code
                std::cout << "MarketArbitrage::start_strategy - " << error << std::endl;
                return;
            }
            if (Maths::Probability::boolean(P_act)) do_strategy();
            start_strategy();
        });
    }
 
    void do_strategy() {
        // check if A is selling for less than B is buying
        if (receiverA.get_book().does_cross(receiverB.get_book(), spread)) {
            // buy on A and sell on B
            // std::cout << "buy on A and sell on B" << std::endl;
            clientA.send<OrderEntry::Messages::OrderRequest>(
                OrderEntry::Messages::ORDER_PRICE_MARKET,
                size,
                OrderEntry::Side::Buy
            );
            clientB.send<OrderEntry::Messages::OrderRequest>(
                OrderEntry::Messages::ORDER_PRICE_MARKET,
                size,
                OrderEntry::Side::Sell
            );
        // check if B is selling for less than A is buying
        } else if (receiverB.get_book().does_cross(receiverA.get_book(), spread)) {
            // buy on B and sell on A
            // std::cout << "buy on B and sell on A" << std::endl;
            clientB.send<OrderEntry::Messages::OrderRequest>(
                OrderEntry::Messages::ORDER_PRICE_MARKET,
                size,
                OrderEntry::Side::Buy
            );
            clientA.send<OrderEntry::Messages::OrderRequest>(
                OrderEntry::Messages::ORDER_PRICE_MARKET,
                size,
                OrderEntry::Side::Sell
            );
        }
    }
 
 public:
    MarketArbitrage(
        asio::io_context& feedA_context,
        asio::io_context& feedB_context,
        asio::io_context& context,
        nlohmann::json options
    ) :
        receiverA(
            feedA_context,
            asio::ip::make_address(options["data_feed"]["A"]["listen"].get<std::string>()),
            asio::ip::make_address(options["data_feed"]["A"]["group"].get<std::string>()),
            options["data_feed"]["A"]["port"].get<uint16_t>(),
            *this
        ),
        receiverB(
            feedB_context,
            asio::ip::make_address(options["data_feed"]["B"]["listen"].get<std::string>()),
            asio::ip::make_address(options["data_feed"]["B"]["group"].get<std::string>()),
            options["data_feed"]["B"]["port"].get<uint16_t>(),
            *this
        ),
        clientA(
            context,
            options["order_entry"]["A"]["host"].get<std::string>(),
            std::to_string(options["order_entry"]["A"]["port"].get<uint16_t>())
        ),
        clientB(
            context,
            options["order_entry"]["B"]["host"].get<std::string>(),
            std::to_string(options["order_entry"]["B"]["port"].get<uint16_t>())
        ),
        timer(context),
        sleep_time(options["strategy"]["sleep_time"].get<uint32_t>()),
        P_act(options["strategy"]["P_act"].get<double>()),
        is_running(false),
        size(options["strategy"]["size"].get<OrderEntry::Quantity>()),
        spread(options["strategy"]["spread"].get<DataFeed::Quantity>()) {
            // login to client A
            auto usernameA = OrderEntry::make_username(options["order_entry"]["A"]["username"].get<std::string>());
            auto passwordA = OrderEntry::make_password(options["order_entry"]["A"]["password"].get<std::string>());
            clientA.send<OrderEntry::Messages::LoginRequest>(usernameA, passwordA);
            // login to client B
            auto usernameB = OrderEntry::make_username(options["order_entry"]["B"]["username"].get<std::string>());
            auto passwordB = OrderEntry::make_password(options["order_entry"]["B"]["password"].get<std::string>());
            clientB.send<OrderEntry::Messages::LoginRequest>(usernameB, passwordB);
        }
 
    inline void did_receive(DataFeedReceiver* receiver, const DataFeed::Messages::StartOfSession& message) {
        if (is_running) {
            std::cerr << "received start of session when already running" << std::endl;
            return;
        }
        // check if both markets are active
        if (receiverA.is_session_active() and receiverB.is_session_active()) {
            // start the strategy
            start_strategy();
            // set the running flag to true
            is_running = true;
        }
    }
 
    inline void did_receive(DataFeedReceiver* receiver, const DataFeed::Messages::EndOfSession& message) {
        if (not is_running) {
            std::cerr << "received end of session when not running" << std::endl;
            return;
        }
        // check if either market has stopped the trading session
        if (not receiverA.is_session_active() or not receiverB.is_session_active()) {
            // stop the strategy
            timer.cancel();
            is_running = false;
        }
    }
 
    inline void did_receive(DataFeedReceiver* receiver, const DataFeed::Messages::Clear& message) { }
 
    inline void did_receive(DataFeedReceiver* receiver, const DataFeed::Messages::AddOrder& message) { }
 
    inline void did_receive(DataFeedReceiver* receiver, const DataFeed::Messages::DeleteOrder& message) { }
 
    inline void did_receive(DataFeedReceiver* receiver, const DataFeed::Messages::Trade& message) { }
};
 
}  // namespace Strategies
 
#endif  // STRATEGIES_MARKET_ARBITRAGE_HPP