Merge pull request #54 from realyarilabs/develop

version 0.2.2

lib/exchange.ex

@@ -13,36 +13,12 @@ defmodule Exchange do
 
   """
   @spec place_order(order_params :: map(), ticker :: atom()) :: atom() | {atom(), String.t()}
-  def place_order(%{type: :limit} = order_params, ticker) do
+  def place_order(order_params, ticker) do
     order_params = Map.put(order_params, :ticker, ticker)
 
     case Exchange.Validations.cast_order(order_params) do
       {:ok, limit_order} ->
-        Exchange.MatchingEngine.place_limit_order(ticker, limit_order)
-
-      {:error, errors} ->
-        {:error, errors}
-    end
-  end
-
-  def place_order(%{type: :market} = order_params, ticker) do
-    order_params = Map.put(order_params, :ticker, ticker)
-
-    case Exchange.Validations.cast_order(order_params) do
-      {:ok, market_order} ->
-        Exchange.MatchingEngine.place_market_order(ticker, market_order)
-
-      {:error, errors} ->
-        {:error, errors}
-    end
-  end
-
-  def place_order(%{type: :marketable_limit} = order_params, ticker) do
-    order_params = Map.put(order_params, :ticker, ticker)
-
-    case Exchange.Validations.cast_order(order_params) do
-      {:ok, marketable_limit_order} ->
-        Exchange.MatchingEngine.place_marketable_limit_order(ticker, marketable_limit_order)
+        Exchange.MatchingEngine.place_order(ticker, limit_order)
 
       {:error, errors} ->
         {:error, errors}
@@ -70,7 +46,7 @@ defmodule Exchange do
 
     - ticker: Atom that represents on which market the order should be canceled
   """
-  @spec spread(ticker :: atom) :: {atom, number}
+  @spec spread(ticker :: atom) :: {atom, Money}
   def spread(ticker) do
     Exchange.MatchingEngine.spread(ticker)
   end
@@ -82,7 +58,7 @@ defmodule Exchange do
 
     - ticker: Atom that represents on which market the query should be placed
   """
-  @spec highest_bid_price(ticker :: atom) :: {atom, number}
+  @spec highest_bid_price(ticker :: atom) :: {atom, Money}
   def highest_bid_price(ticker) do
     Exchange.MatchingEngine.bid_max(ticker)
   end
@@ -106,7 +82,7 @@ defmodule Exchange do
 
     - ticker: Atom that represents on which market the query should be placed
   """
-  @spec lowest_ask_price(ticker :: atom) :: {atom, number}
+  @spec lowest_ask_price(ticker :: atom) :: {atom, Money}
   def lowest_ask_price(ticker) do
     Exchange.MatchingEngine.ask_min(ticker)
   end
@@ -160,10 +136,28 @@ defmodule Exchange do
     Exchange.MatchingEngine.open_orders_by_trader(ticker, trader_id)
   end
 
-  def last_price do
+  @doc """
+  Returns the lastest price from a side of an Exchange
+
+  ## Parameters
+    - ticker: Exchange identifier
+    - side: Atom to decide which side of the book is used
+  """
+  @spec last_price(ticker :: atom, side :: atom) :: {atom, number}
+  def last_price(ticker, side) do
+    Exchange.MatchingEngine.last_price(ticker, side)
   end
 
-  def last_size do
+  @doc """
+  Returns the lastest size from a side of an Exchange
+
+  ## Parameters
+    - ticker: Exchange identifier
+    - side: Atom to decide which side of the book is used
+  """
+  @spec last_size(ticker :: atom, ticker :: atom) :: {atom, number}
+  def last_size(ticker, side) do
+    Exchange.MatchingEngine.last_size(ticker, side)
   end
 
   @doc """
@@ -216,8 +210,21 @@ defmodule Exchange do
 
     - ticker: Atom that represents on which market the query should made
   """
-  @spec completed_trades(ticker :: atom) :: {atom, list()}
+  @spec completed_trades(ticker :: atom) :: list
   def completed_trades(ticker) do
     Exchange.Utils.fetch_all_completed_trades(ticker)
   end
+
+  @doc """
+  Returns the trade with trade_id
+
+  ## Parameters
+
+    - ticker: Atom that represents on which market the query should made
+    - trade_id: Id of the requested trade
+  """
+  @spec completed_trade_by_trade_id(ticker :: atom, trade_id :: String.t()) :: Exchange.Trade.t()
+  def completed_trade_by_trade_id(ticker, trade_id) do
+    Exchange.Utils.fetch_completed_trade_by_trade_id(ticker, trade_id)
+  end
 end

lib/exchange/adapters/event_bus.ex

@@ -6,8 +6,7 @@ defmodule Exchange.Adapters.EventBus do
   alias Exchange.Adapters.MessageBus
 
   use Exchange.MessageBus, required_config: [], required_deps: []
-  @events ~w(trade_executed order_queued order_cancelled order_expired
-             transaction_open order_placed trade_processed price_broadcast)a
+  @events ~w(trade_executed order_queued order_cancelled order_expired price_broadcast)a
 
   @doc """
   Adds the process calling this function to the `Registry` under the given `key`
@@ -51,11 +50,9 @@ defmodule Exchange.Adapters.EventBus do
   @spec cast_event(
           :order_cancelled
           | :order_expired
-          | :order_placed
           | :order_queued
           | :price_broadcast
-          | :trade_executed
-          | :trade_processed,
+          | :trade_executed,
           any
         ) :: nil | :ok
   def cast_event(:order_cancelled, payload),
@@ -67,15 +64,9 @@ defmodule Exchange.Adapters.EventBus do
   def cast_event(:order_expired, payload),
     do: dispatch_event(:order_expired, %MessageBus.OrderExpired{order: payload})
 
-  def cast_event(:order_placed, payload),
-    do: dispatch_event(:order_placed, %MessageBus.OrderPlaced{} = payload)
-
   def cast_event(:order_queued, payload),
     do: dispatch_event(:order_queued, %MessageBus.OrderQueued{order: payload})
 
-  def cast_event(:trade_processed, payload),
-    do: dispatch_event(:trade_processed, %MessageBus.TradeProcessed{} = payload)
-
   def cast_event(:price_broadcast, payload) do
     price_broadcast_event = %MessageBus.PriceBroadcast{
       ticker: payload.ticker,

lib/exchange/adapters/flux.ex

@@ -79,5 +79,32 @@ if Code.ensure_loaded?(Instream.Connection) do
         }
       end)
     end
+
+    @spec get_completed_trade_by_trade_id(ticker :: atom, trade_id :: String.t()) ::
+            Exchange.Trade
+    def get_completed_trade_by_trade_id(ticker, trade_id) do
+      Trades.get_completed_trade_by_trade_id(ticker, trade_id)
+      |> Enum.map(fn flux_trade ->
+        trade = %Exchange.Trade{}
+
+        %{
+          trade
+          | trade_id: flux_trade.fields.trade_id,
+            ticker: String.to_atom(flux_trade.tags.ticker),
+            currency: flux_trade.tags.currency,
+            buyer_id: flux_trade.tags.buyer_id,
+            seller_id: flux_trade.tags.seller_id,
+            buy_order_id: flux_trade.fields.buy_order_id,
+            sell_order_id: flux_trade.fields.sell_order_id,
+            price: flux_trade.fields.price,
+            size: flux_trade.fields.size,
+            buy_init_size: flux_trade.fields.buy_init_size,
+            sell_init_size: flux_trade.fields.sell_init_size,
+            type: flux_trade.fields.type,
+            acknowledged_at: flux_trade.fields.acknowledged_at
+        }
+      end)
+      |> List.first()
+    end
   end
 end

lib/exchange/adapters/flux/orders.ex

@@ -62,7 +62,7 @@ if Code.ensure_loaded?(Instream.Connection) do
               initial_size: order_params.initial_size,
               price: order_params.price,
               type: Atom.to_string(order_params.type),
-              modified_at: :os.system_time(:nanosecond)
+              modified_at: DateTime.utc_now() |> DateTime.to_unix(:nanosecond)
           },
           tags: %{
             data.tags

lib/exchange/adapters/flux/trades.ex

@@ -54,6 +54,20 @@ if Code.ensure_loaded?(Instream.Connection) do
       end
     end
 
+    @spec get_completed_trade_by_trade_id(ticker :: atom, trade_id :: String.t()) ::
+            Exchange.Trade | atom
+    def get_completed_trade_by_trade_id(ticker, trade_id) do
+      response =
+        ~s(SELECT * FROM trades WHERE trade_id = '#{trade_id}' and ticker = '#{ticker}')
+        |> Flux.Connection.query(precision: :nanosecond)
+
+      if response.results == [%{statement_id: 0}] do
+        nil
+      else
+        Flux.Trades.from_result(response)
+      end
+    end
+
     def process_trade!(trade_params) do
       data = %Flux.Trades{}
       t = trade_params

lib/exchange/adapters/in_memory_time_series.ex

@@ -107,6 +107,20 @@ defmodule Exchange.Adapters.InMemoryTimeSeries do
     {:reply, {:ok, trades_by_id}, state}
   end
 
+  def handle_call({:trade_by_trade_id, ticker, trade_id}, _from, state) do
+    {:ok, trades} = Map.fetch(state, :trades)
+
+    trade =
+      trades
+      |> Enum.flat_map(fn {_ts, queue} -> queue end)
+      |> Enum.filter(fn trade ->
+        trade.trade_id == trade_id and trade.ticker == ticker
+      end)
+      |> List.first()
+
+    {:reply, {:ok, trade}, state}
+  end
+
   def handle_call({:live_orders, ticker}, _from, state) do
     {:ok, orders} = Map.fetch(state, :orders)
 
@@ -153,13 +167,13 @@ defmodule Exchange.Adapters.InMemoryTimeSeries do
 
   @spec save_price(price :: map, state :: map) :: map
   def save_price(price, state) do
-    current_time = :os.system_time(:nanosecond)
+    current_time = DateTime.utc_now() |> DateTime.to_unix(:nanosecond)
     {:ok, prices} = Map.fetch(state, :prices)
     update_prices = save(price, current_time, prices)
     Map.put(state, :prices, update_prices)
   end
 
-  @spec save_order(Exchange.Order, map) :: map
+  @spec save_order(Exchange.Order.order(), map) :: map
   def save_order(order, state) do
     ack_time = order.acknowledged_at
     {:ok, orders} = Map.fetch(state, :orders)
@@ -210,4 +224,10 @@ defmodule Exchange.Adapters.InMemoryTimeSeries do
     {:ok, orders} = GenServer.call(:in_memory_time_series, {:live_orders, ticker})
     orders
   end
+
+  @spec get_completed_trade_by_trade_id(ticker :: atom, trade_id :: String.t()) :: Exchange.Trade
+  def get_completed_trade_by_trade_id(ticker, trade_id) do
+    {:ok, trade} = GenServer.call(:in_memory_time_series, {:trade_by_trade_id, ticker, trade_id})
+    trade
+  end
 end

lib/exchange/adapters/message_bus/trade_executed.ex

@@ -15,7 +15,7 @@ defmodule Exchange.Adapters.MessageBus.TradeExecuted do
   ## Parameters
     - params: map with necessary parameters to populate the struct
   """
-  @spec decode_from_jason(map) :: Exchange.Adapters.MessageBus.TradeExecuted.t()
+  @spec decode_from_jason(map) :: %Exchange.Adapters.MessageBus.TradeExecuted{}
   def decode_from_jason(params) do
     trade = Map.get(params, :trade)