# encoding: UTF-8 # frozen_string_literal: true require_relative 'constants' module Matching class Executor ExecutorError = Class.new(StandardError) def initialize(payload) @payload = payload @trade_payload = payload[:trade].symbolize_keys if payload[:trade] end def process case @payload[:action] when 'execute' execute when 'cancel' publish_cancel else raise ExecutorError.new("Unknown action: #{@payload[:action]}") end end def publish_cancel AMQP::Queue.enqueue(:order_processor, { action: 'cancel', order: @payload[:order] }, { persistent: false }) end def execute execute! # TODO: Queue should exist even if none is listening. rescue TradeExecutionError => e AMQP::Queue.enqueue(:trade_error, e.options) [@maker_order, @taker_order].each do |order| order.with_lock do next unless order.state == Order::WAIT AMQP::Queue.enqueue(:matching, action: 'submit', order: order.to_matching_attributes) end end report_exception_to_screen(e) false end def execute! # NOTE: Run matching engine for disabled markets. @market = Market.find(@trade_payload[:market_id]) @price = @trade_payload[:strike_price].to_d @amount = @trade_payload[:amount].to_d @total = @trade_payload[:total].to_d create_trade_and_strike_orders publish_trade @trade end private def validate! ask, bid = @maker_order.side == 'sell' ? [@maker_order, @taker_order] : [@taker_order, @maker_order] raise_error(3001, 'Ask price exceeds strike price.') if ask.ord_type == 'limit' && ask.price > @price raise_error(3002, 'Bid price is less than strike price.') if bid.ord_type == 'limit' && bid.price < @price raise_error(3003, "Maker order state isn\'t equal to «wait» (#{@maker_order.state}).") unless @maker_order.state == Order::WAIT raise_error(3004, "Taker order state isn\'t equal to «wait» (#{@taker_order.state}).") unless @taker_order.state == Order::WAIT unless @total > ZERO && [@maker_order.volume, @taker_order.volume].min >= @amount raise_error(3005, 'Not enough funds.') end end def create_trade_and_strike_orders ActiveRecord::Base.transaction do Order.lock.where(id: [@trade_payload[:maker_order_id], @trade_payload[:taker_order_id]]) .includes(:ask_currency, :bid_currency) .to_a .tap do |orders| @maker_order = orders.find { |order| order.id == @trade_payload[:maker_order_id] } @taker_order = orders.find { |order| order.id == @trade_payload[:taker_order_id] } end # Check if accounts exists or create them. @maker_order.member.get_account(@maker_order.income_currency) @taker_order.member.get_account(@taker_order.income_currency) validate! accounts_table = Account .lock .select(:id, :member_id, :currency_id, :balance, :locked) .where(member_id: [@maker_order.member_id, @taker_order .member_id].uniq, currency_id: [@market.base_unit, @market.quote_unit]) .each_with_object({}) { |record, memo| memo["#{record.currency_id}:#{record.member_id}"] = record } @trade = Trade.new \ maker_order: @maker_order, maker_id: @maker_order.member_id, taker_order: @taker_order, taker_id: @taker_order.member_id, price: @price, amount: @amount, total: @total, market: @market strike(@trade, @maker_order, accounts_table["#{@maker_order.outcome_currency.id}:#{@maker_order.member_id}"], accounts_table["#{@maker_order.income_currency.id}:#{@maker_order.member_id}"]) strike(@trade, @taker_order, accounts_table["#{@taker_order.outcome_currency.id}:#{@taker_order.member_id}"], accounts_table["#{@taker_order.income_currency.id}:#{@taker_order.member_id}"]) @trade.record_complete_operations! ([@maker_order, @taker_order] + accounts_table.values).map do |record| table = record.class.arel_table statement = Arel::UpdateManager.new statement.table(table) statement.where(table[:id].eq(record.id)) updates = record.changed_attributes.map do |(attribute, _)| if Order === record value = record.public_send(attribute) [table[attribute], ::Order::STATES.with_indifferent_access.fetch(value, value)] else [table[attribute], record.public_send(attribute)] end end statement.set updates statement.to_sql end.join('; ').tap do |sql| Rails.logger.debug { sql } client = ActiveRecord::Base.connection.raw_connection if Rails.configuration.database_adapter.downcase == 'PostgreSQL'.downcase client.send_query(sql) client.set_single_row_mode loop do (res = client.get_result) or break res.check res.each do |_row| end end else client.query(sql) while client.next_result end end end @trade.save(validate: false) end end def publish_trade AMQP::Queue.publish :trade, @trade.as_json, { headers: { type: :local, market: @market.id, maker_id: @maker_id, taker_id: @taker_id } } @trade.trigger_event [@maker_order, @taker_order].each do |order| event = case order.state when 'cancel' then 'order_canceled' when 'done' then 'order_completed' else 'order_updated' end order.trigger_event next unless order.ord_type == 'limit' # Skip market orders. EventAPI.notify ['market', order.market_id, event].join('.'), \ Serializers::EventAPI.const_get(event.camelize).call(order) end end def raise_error(code, message) raise TradeExecutionError.new \ maker_order: @maker_order.attributes, taker_order: @taker_order.attributes, price: @price, amount: @amount, total: @total, code: code, message: message end def strike(trade, order, outcome_account, income_account) #amount is crypto and total is price so total is rls or usdt outcome_value, income_value = OrderAsk === order ? [trade.amount, trade.total] : [trade.total, trade.amount] fee = income_value * trade.order_fee(order) real_income_value = income_value - fee outcome_account.assign_attributes outcome_account.attributes_after_unlock_and_sub_funds!(outcome_value) income_account.assign_attributes income_account.attributes_after_plus_funds!(real_income_value) order.volume -= trade.amount order.locked -= outcome_value order.funds_received += income_value order.trades_count += 1 order.updated_at = Time.now if order.volume.zero? order.state = Order::DONE # Unlock not used funds. unless order.locked.zero? outcome_account.assign_attributes outcome_account.attributes_after_unlock_funds!(order.locked) end elsif order.ord_type == 'market' && order.locked.zero? # Partially filled market order has run out it's locked funds. order.state = Order::CANCEL order.record_cancel_operations! end end end end