somehow, things start to work togheter

project.clj

@@ -19,6 +19,7 @@
                  [slingshot "0.12.2"]
                  [spootnik/kinsky "0.1.22"]
                  [org.clojars.fossi/kafka-streams-clojure "0.1.0-SNAPSHOT"] ;; JEAH!
+                 [orchestra "2017.11.12-1"]
                  [cheshire "5.8.0"]]
   :java-source-paths ["src/java"]
   :main ^:skip-aot skirmish.core

resources/config.edn

@@ -1,28 +1,37 @@
 {
- ;;
+ ;; Specify which middleware to use, it will default to :chan.
  :middleware :kafka
- :kafka  ;; all the kafka configuration
+
+ ;; If set to true, orchestra will instrument all the spec'ed functions
+ :instrument-spec true
+
+ ;; all the kafka configuration goes bellow.
+ :kafka
  {:bootstrap.servers #or [#env KAFKA_BOOTSTRAP_SERVERS "localhost:9092"]
   :topics-prefix "skirmish-01-"}
 
- :api-server ;; possible config from http-kit's run-server that will affect the api-server
+ ;; possible config from http-kit's run-server that will affect the api-server
+ :api-server
  {:port #long #or [#env API_SERVER_PORT 1602]
   :ip "0.0.0.0"
   :max-body 524288 ;; 512kb, small but enough
   :max-ws 524288}
 
- :api-bookkeeper ;; possible config from http-kit's run-server that will affect the api-bookkeeper
+ ;; possible config from http-kit's run-server that will affect the api-bookkeeper
+ :api-bookkeeper
  {:port #long #or [#env API_SERVER_PORT 1603]
   :ip "0.0.0.0"
   :max-body 524288 ;; 512kb, small but enough
   :max-ws 524288}
 
- :securities ;; list of tradable securities
+ ;; list of tradable securities
+ :securities
  [{:sec-id :foo}
   {:sec-id :bar}
   {:sec-id :toto}
   {:sec-id :tata}]
 
+ ;; stupid parameters about benchmarking tools
  :benchmark {
              :number-of-requests-per-securities 30}}
 

src/skirmish/core.clj

@@ -5,6 +5,7 @@
             [skirmish.exchange.core]
             [skirmish.middleware.core :refer [create-exchange-apis-for]]
             [aero.core :refer (read-config)]
+            [orchestra.spec.test :as st]
             [clojure.core.async :as async])
   (:gen-class))
 
@@ -12,17 +13,16 @@
              ;; arguments:  the config map and the "exchange-api"
   {:api-server     skirmish.rest.core/api-main
    :exchange-core  skirmish.exchange.core/-main ;; should be skirmish.exchange.core/api-main
-   :api-bookkeeper skirmish.rest.core/bookeeper-main})
+   :api-bookkeeper skirmish.rest.core/bookkeeper-main})
    ;:benchmark      skirmish.cli.benchmark/-main})
 
 (defn run-all-targets-in-separate-threads [target-fns config exchange-apis]
   (println "running  multiple targets. This is okay in dev but horrible in production")
   (println target-fns)
   (let [threads (map #(async/thread (% config exchange-apis)) target-fns)
-        finished-threads (async/merge threads)];; need to lock here
+        finished-threads (async/merge threads)]
     (loop [finished-thread (async/<!! finished-threads)]
       (when (some? finished-thread)
-          (println (str "thread " finished-thread " finished"))
           (recur (async/<!! finished-threads))))))
 
 
@@ -36,6 +36,8 @@
   (let [config (-> "config.edn" clojure.java.io/resource read-config)
         xf (comp (map keyword) (map targets) (filter some?))
         target-fns (sequence xf argv)]
+    (if (true? (:instrument-spec config))
+      (st/instrument))
     (if (seq target-fns)
       (run-all-targets target-fns config)
       (do

src/skirmish/exchange/exchange.clj

@@ -4,6 +4,7 @@
                                              create-sell-book
                                              match-order-in-book]]
             [skirmish.util :refer :all]
+            [clojure.spec.alpha :as s]
             [clojure.core.async :refer [chan go >! <! pub] :as async]))
 
 
@@ -13,6 +14,18 @@
                  :buy (create-buy-book)
                  :id 0})
 
+(defn get-reference-book [order book]
+  "The reference book is the book where the order will be fulfilled"
+  (if (sell? order)
+    (:buy book)
+    (:sell book)))
+
+(defn get-same-type-order [order book]
+  "the same type order book is the book where the order will be added if not fulfilled"
+  (if (sell? order)
+    (:sell book)
+    (:buy book)))
+
 ;;  Bellow are all the functions to match the received order in the correct book
 (defn process-abstract [order reference-book same-type-book]
   (let [[trades new-reference-book leftover-order] (match-order-in-book reference-book order)
@@ -44,23 +57,23 @@
   (let [book-id (:book.order/book-id trade)
         f #(= book-id (:book.order/id %))
         ret-order (first (filter f book))
-        new-quantity (- (:quantity ret-order 10) (:book.order/quantity trade));; (- (:quantity ret-order) (:book.order/quantity trade))
+        new-quantity (- (:quantity ret-order) (:book.order/quantity trade));; (- (:quantity ret-order) (:book.order/quantity trade))
         totally-fulfilled? (= (:book.order/quantity trade) (:quantity ret-order))]
-    (into ret-order {:status         (if totally-fulfilled? :totally-fulfilled :partially-fulfilled)
+    (into ret-order {:book.order/status         (if totally-fulfilled? :totally-fulfilled :partially-fulfilled)
-                     :status-dual-id (:book.order/order-id trade)
+                     :book.order/status-dual-id (:book.order/order-id trade)
-                     :quantity new-quantity})));;;;
+                     :quantity new-quantity})))
 
 
 (defn get-intermediate-order [order trade]
   (let [new-order-quantity (- (:quantity order) (:book.order/quantity trade))
         totally-fulfilled? (= 0 new-order-quantity)
         dual-id (:book.order/book-id trade)]
-    (into order {:status (if totally-fulfilled? :totally-fulfilled :partially-fulfilled)
+    (into order {:book.order/status (if totally-fulfilled? :totally-fulfilled :partially-fulfilled)
-                 :status-dual-id dual-id
+                 :book.order/status-dual-id dual-id
                  :quantity new-order-quantity})))
 
 
-(defn get-processed-orders-in-trades [order book trades]
+(defn  get-processed-orders-in-trades[order book trades]
   (let [trades-orders (sequence  (map #(retrieve-new-book-order-from-trade % book)) trades)
         processed-orders (rest (reductions get-intermediate-order order trades))]
     (interleave trades-orders processed-orders)))
@@ -68,22 +81,26 @@
 
 (defn get-processed-orders [order leftover-order trades book]
   "Computes all the intermediary orders that took place during
-  a the trade."
+  a the trade.
-  (let [ret [(assoc order :status :inserted)]
+  The interal logic is very complicated.  Probably too complicated."
+  (let [ret [(assoc order :book.order/status :received)]
         ret (into ret (get-processed-orders-in-trades order book trades))
         ret (if (seq leftover-order)
-              (conj ret (assoc leftover-order :status :inserted-into-book))
+              (conj ret (assoc leftover-order :book.order/status :inserted-into-book))
-              leftover-order)]
+              ret)]
     ret))
 
+(s/fdef get-processed-orders
+        :ret (s/* :book.order/order))
 
-;; bellow are the really interesting functions.
+
+;; bellow are the really interesting functions; aka public functions.
 
 (defn create-exchange-for-security [initial-book new-order-publisher trades-submitter orders-submitter]
   (async/go
     (loop [order (<! new-order-publisher) book initial-book]
       (let [[new-book new-order new-trades] (process-order book order)
-            orders (get-processed-orders order new-order new-trades book)
+            orders (get-processed-orders order new-order new-trades (get-reference-book order book))
             orders-copy (async/onto-chan orders-submitter orders false)
             trades-copy (async/onto-chan trades-submitter new-trades false)]
         (<! orders-copy)

src/skirmish/middleware/core.clj

@@ -13,7 +13,7 @@
      :new-order-submitter    new-order
      :trades-publisher       (async/pub trades :book.order/type)
      :trades-submitter       trades
-     :orders-publisher       (async/pub orders :type)
+     :orders-publisher       (async/pub orders :book.order/status)
      :orders-submitter       orders}))
 
 (def api-creators {})

src/skirmish/order.clj

@@ -7,18 +7,19 @@
 ;;  :type    one of :sell :buy
 ;;  ::id     maybe an id
 
-(s/def :book.order/type #{:buy :sell})
+(def order-status #{:received  ;; maybe called acknowledged.
-(s/def :book.order/price pos-int?)
-(s/def :book.order/quantity pos-int?)
-(s/def :book.order/id    pos-int?)
-(s/def :book.order/status #{:inserted  ;; maybe called acknklowdlge.
                     :partially-matched
                     :totally-matched
                     :inserted-into-book
                     :partially-fulfilled
                     :totally-fulfilled
-                            :withdrawn}) ;; for instance the order has been canceled
+                    :withdrawn}) ;; for instance the order has been canceled)
 
+(s/def :book.order/type #{:buy :sell})
+(s/def :book.order/price nat-int?)
+(s/def :book.order/quantity nat-int?)
+(s/def :book.order/id    nat-int?)
+(s/def :book.order/status order-status)
 
 
 (s/def :book.order/status-dual-id :book.order/id)

src/skirmish/rest/bookkeeper.clj

@@ -1 +1,118 @@
-(ns skirmish.rest.marketdata)
+(ns skirmish.rest.bookkeeper
+  (:require [skirmish.exchange.exchange :refer [empty-book create-exchange-for-security]]
+            [org.httpkit.server :refer [run-server with-channel send! close on-close]]
+            [compojure.route :refer [not-found]]
+            [compojure.handler :refer [api]]
+            [compojure.core :refer :all]
+            [skirmish.order]  ;; for spec side effect
+            [skirmish.rest.util :refer [build-response build-response-body]]
+            [clojure.spec.alpha :as s]
+            [kinsky.client]
+            [slingshot.slingshot :refer [try+ throw+]]
+            [clojure.core.async :refer [chan go-loop close! <! sub]]))
+
+
+(def order-id (atom -1))
+
+
+(defn api-description [req]
+  (let [body {"/api/1/securities" "list all securities."
+              "/api/1/book"    "For all information about the book."}]
+    (build-response req body)))
+
+(defn subscribe-to-orders-publisher [exchange-api status-to-listen-on]
+  (let [ret (chan)
+        new-book-order-channel (:orders-publisher exchange-api)]
+    (doseq [status status-to-listen-on]
+      (sub new-book-order-channel status ret))
+    ret))
+
+
+(defn get-new-order-channel [exchange-apis sec-id status-to-listen-on]
+  (let [exchange (get exchange-apis sec-id)]
+    (subscribe-to-orders-publisher exchange status-to-listen-on)))
+
+
+(defn get-new-trades-channel [exchange-apis sec-id]
+  (let [exchange (get exchange-apis sec-id)
+        new-trades-channel (:trades-publisher exchange)
+        ret (chan)]
+    (sub new-trades-channel :sell ret)
+    (sub new-trades-channel :buy ret)
+    ret))
+
+
+(defn securities [exchange-apis]
+  (fn [request]
+    (build-response request {:securities (keys exchange-apis)})))
+
+
+(defn get-security [exchange-apis]
+  (fn [{{sec-id-str :sec-id} :route-params body :body :as request}]
+    (build-response request {:security {:sec-id sec-id-str}})))
+
+
+(defn chan-to-websocket [request input-chan ws-chan]
+  (on-close ws-chan (fn [status] (close! input-chan)))
+  (go-loop []
+    (when-let [e (<! input-chan)]
+      (send! ws-chan (build-response-body request e))
+      (recur))))
+
+
+(defn get-status-to-listen-on [{params :query-params}]
+  (let [param-status (:status params)
+        status (if (seq? param-status) param-status skirmish.order/order-status)]
+    (filter #(s/valid? :book.order/status %) status)))
+
+
+(defn listen-to-orders [exchange-apis]
+  (fn [{{sec-id-str :sec-id} :route-params :as request}]
+    (with-channel request ws-chan
+                  (let [status-to-listen-on (get-status-to-listen-on request)
+                        order-channel (get-new-order-channel exchange-apis (keyword sec-id-str) status-to-listen-on)]
+                    (chan-to-websocket request order-channel ws-chan)))))
+
+
+(defn listen-to-trades [exchange-apis]
+  (fn [{{sec-id-str :sec-id} :route-params :as request}]
+    (with-channel request ws-chan
+                  (let [order-channel (get-new-trades-channel exchange-apis (keyword sec-id-str))]
+                    (chan-to-websocket request order-channel ws-chan)))))
+
+
+(defn create-book [exchange-api]
+  (let [ret (atom {})
+        new-book-order-channel (subscribe-to-orders-publisher exchange-api skirmish.order/order-status)]
+    (go-loop [order (<! new-book-order-channel)]
+      (let [order-id (:book.order/id order)
+            to-remove (contains? #{:withdrawn :totally-fulfilled} (:book.order/status order))]
+        (println ret)
+        (swap! ret (fn [book]
+                     (if to-remove
+                       (dissoc book order-id)
+                       (assoc book order-id order)))))
+      (recur (<! new-book-order-channel)))
+    ret))
+
+(defn map-values [f m]
+  (into {} (for [[k v] m] [k (f v)])))
+
+(defn get-books [exchange-apis]
+  (let [books (map-values #(create-book %) exchange-apis)]
+    (fn [{{sec-id-str :sec-id} :route-params body :body :as request}]
+      (let [sec-id (keyword sec-id-str)]
+        (build-response request @(get books sec-id))))))
+
+
+(defn create-routes-for-bookkeeper [exchange-apis]
+  (let [get-book (get-books exchange-apis)]
+    (routes
+      (context "/api/1" []
+        (GET "/" [] api-description)
+        (GET "/securities" [] (securities exchange-apis))
+        (GET "/securities/:sec-id" [] (get-security exchange-apis))
+        (GET "/securities/:sec-id/trades" [] (listen-to-trades exchange-apis))
+        (GET "/securities/:sec-id/orders" [] (listen-to-orders exchange-apis))
+        (GET "/securities/:sec-id/book" [] get-book))
+      (not-found "not found"))))

src/skirmish/rest/core.clj

@@ -3,8 +3,8 @@
             [compojure.handler :refer [api]]
             [skirmish.util :refer [block-thread]]
             [skirmish.middleware.core :refer [create-exchange-apis-for]]
-            [skirmish.rest.exchange :refer [create-routes-for-exchange
+            [skirmish.rest.bookkeeper :refer [create-routes-for-bookkeeper]]
-                                            create-routes-for-bookkeeper]]))
+            [skirmish.rest.exchange :refer [create-routes-for-exchange]]))
 
 
 (defn api-main [config exchange-apis]
@@ -15,7 +15,7 @@
     (run-server (api api-routes) webserver-config)))
 
 
-(defn bookeeper-main [config exchange-apis]
+(defn bookkeeper-main [config exchange-apis]
   (let [webserver-config (:api-bookkeeper config)
         api-routes (create-routes-for-bookkeeper exchange-apis)]
     (println "Running server with bellow config")

src/skirmish/rest/exchange.clj

@@ -16,48 +16,6 @@
 (def order-id (atom -1))
 
 
-(defn api-description [req]
-  (let [body {"/api/1/securities" "list all securities."
-              "/api/1/book"    "For all information about the book."}]
-    (build-response req body)))
-
-
-;; the api should look like...
-;; trades ...
-;;    /:sec-id/trades/:trade-id
-;; book ...
-;;    /:sec-id/book/:order
-;;
-
-
-(defn get-new-order-channel [exchange-apis sec-id]
-  (let [exchange (get exchange-apis sec-id)
-        new-book-order-channel (:orders-publisher exchange)
-        ret (chan)]
-    (sub new-book-order-channel :sell ret)
-    (sub new-book-order-channel :buy ret)
-    ret))
-
-
-(defn get-new-trades-channel [exchange-apis sec-id]
-  (let [exchange (get exchange-apis sec-id)
-        new-trades-channel (:trades-publisher exchange)
-        ret (chan)]
-    (sub new-trades-channel :sell ret)
-    (sub new-trades-channel :buy ret)
-    ret))
-
-
-(defn securities [exchange-apis]
-  (fn [request]
-    (build-response request {:securities (keys exchange-apis)})))
-
-
-(defn get-security [exchange-apis]
-  (fn [{{sec-id-str :sec-id} :route-params body :body :as request}]
-    (build-response request {:security {:sec-id sec-id-str}})))
-
-
 (defn validate-order [asked-order sec-id]
   (let [now (System/currentTimeMillis)]
     (if (not (s/valid? :book.order/order asked-order))
@@ -96,41 +54,9 @@
       (catch [:error-type :5XX] e (build-response request e)))))
 
 
-(defn chan-to-websocket [request input-chan ws-chan]
-  (on-close ws-chan (fn [status] (close! input-chan)))
-  (go-loop []
-    (when-let [e (<! input-chan)]
-      (send! ws-chan (build-response-body request e))
-      (recur))))
-
-
-(defn listen-to-orders [exchange-apis]
-  (fn [{{sec-id-str :sec-id} :route-params :as request}]
-    (with-channel request ws-chan
-      (let [order-channel (get-new-order-channel exchange-apis (keyword sec-id-str))]
-        (chan-to-websocket request order-channel ws-chan)))))
-
-
-(defn listen-to-trades [exchange-apis]
-  (fn [{{sec-id-str :sec-id} :route-params :as request}]
-    (with-channel request ws-chan
-                  (let [order-channel (get-new-trades-channel exchange-apis (keyword sec-id-str))]
-                    (chan-to-websocket request order-channel ws-chan)))))
-
-
 (defn create-routes-for-exchange [exchange-apis]
   (routes
     (context "/api/1" []
       (POST "/securities/:sec-id" [] (create-order exchange-apis)))
     (not-found "not found")))
 
-
-(defn create-routes-for-bookkeeper [exchange-apis]
-  (routes
-    (context "/api/1" []
-      (GET "/" [] api-description)
-      (GET "/securities" [] (securities exchange-apis))
-      (GET "/securities/:sec-id" [] (get-security exchange-apis))
-      (GET "/securities/:sec-id/trades" [] (listen-to-trades exchange-apis))
-      (GET "/securities/:sec-id/orders" [] (listen-to-orders exchange-apis)))
-    (not-found "not found")))

test/skirmish/exchange/exchange_test.clj

@@ -1,11 +1,14 @@
 (ns skirmish.exchange.exchange-test
   (:import [skirmish.PersistentSortedList])
   (:require [midje.sweet :refer :all]
+            [orchestra.spec.test :as st]
             [skirmish.exchange.exchange :refer :all]
             [skirmish.exchange.match :refer [create-buy-book
                                              create-sell-book
                                              match-order-in-book]]))
 
+(st/instrument)
+
 (def PersistentSortedList? (comp #(= % skirmish.PersistentSortedList) type))
 
 (facts "the exchange as an whole"
@@ -70,13 +73,21 @@
                book-4 => expt-book-4)))
 
 (facts "get-processed-orders"
-       (fact "ss"
+       (fact "bug numero uno"
              (let [sell-book (create-buy-book {:type :buy :quantity 2 :price 14 :book.order/id 116}
                                               {:type :buy :quantity 8 :price 13 :book.order/id 117})
                    order {:type :sell :quantity 21 :price 13 :book.order/id 118}
                    [trades new-book leftover-order] (match-order-in-book sell-book order)
                    processed-order (get-processed-orders order leftover-order trades sell-book)]
-               processed-order => seq)));;;
+               processed-order => seq));;;
+       (fact "bug numero dos"
+             (let [order   {:type :sell, :quantity 10, :price 2, :book.order/id 1}
+                   leftover-order nil
+                   trades [#:book.order{:type :buy, :quantity 10, :price 3, :order-id 1, :book-id 0}]
+                   book  {:sell (create-sell-book)
+                          :buy (create-buy-book {:type :buy, :quantity 15, :price 3, :book.order/id 0})}
+                   ret (get-processed-orders order leftover-order trades (:buy book))]
+                (count ret) => 3)))