woohoo
This commit is contained in:
1 parent
b036f1e7fd
commit
1dbae11b21
5 files changed
+82
-71
No files matched your search
@@ -1,6 +1,6 @@
|
|||||||
# Skrimish
|
# Skrimish
|
||||||
|
|
||||||

|

|
||||||
|
|
||||||
A stock exchange simulator. The scope of this project is not yet defined but it should
|
A stock exchange simulator. The scope of this project is not yet defined but it should
|
||||||
allow end user(s?) to exchange securities via the common exchange rules.
|
allow end user(s?) to exchange securities via the common exchange rules.
|
||||||
@@ -9,6 +9,21 @@ allow end user(s?) to exchange securities via the common exchange rules.
|
|||||||
## Installation
|
## Installation
|
||||||
Okay, this should be easy but it is hard...
|
Okay, this should be easy but it is hard...
|
||||||
|
|
||||||
|
## Hack me
|
||||||
|
|
||||||
|
### Configuration
|
||||||
|
All configuration are set in the `resources/config.edn` file.
|
||||||
|
Configuration uses [aero](https://github.com/juxt/aero).
|
||||||
|
|
||||||
|
### Technologies
|
||||||
|
|
||||||
|
#### Server
|
||||||
|
* clojure
|
||||||
|
* kafka
|
||||||
|
|
||||||
|
#### In the browser
|
||||||
|
* https://aftertheflood.com/projects/sparks/
|
||||||
|
* http://metricsgraphicsjs.org/
|
||||||
|
|
||||||
## Usage
|
## Usage
|
||||||
|
|
||||||
|
|||||||
@@ -3,8 +3,6 @@
|
|||||||
{:bootstrap.servers "localhost:9092"
|
{:bootstrap.servers "localhost:9092"
|
||||||
:topics-prefix "skirmish-01-"}
|
:topics-prefix "skirmish-01-"}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
:webserver ;; possible config from http-kit's run-server
|
:webserver ;; possible config from http-kit's run-server
|
||||||
{:port #long #or [#env PORT 8080]
|
{:port #long #or [#env PORT 8080]
|
||||||
:ip "0.0.0.0"
|
:ip "0.0.0.0"
|
||||||
@@ -16,6 +14,7 @@
|
|||||||
{:sec-id :bar}
|
{:sec-id :bar}
|
||||||
{:sec-id :toto}
|
{:sec-id :toto}
|
||||||
{:sec-id :tata}]
|
{:sec-id :tata}]
|
||||||
|
|
||||||
:benchmark {
|
:benchmark {
|
||||||
:number-of-requests-per-securities 30}}
|
:number-of-requests-per-securities 30}}
|
||||||
|
|
||||||
|
|||||||
@@ -16,17 +16,6 @@
|
|||||||
(ws/connect url :headers headers :on-receive on-receive)))
|
(ws/connect url :headers headers :on-receive on-receive)))
|
||||||
|
|
||||||
|
|
||||||
(defn create-trades-channel [securities]
|
|
||||||
(let [output (chan 1 (map edn/read-string))
|
|
||||||
url #(format "ws://localhost:%s/api/1/securities/%s/trades" 8080 %)
|
|
||||||
xf (comp (map :sec-id)
|
|
||||||
(map name)
|
|
||||||
(map url)
|
|
||||||
(map #(web-socket-to-channel % output)))]
|
|
||||||
(doall (sequence xf securities))
|
|
||||||
(async/pub output :book.order/order-id)))
|
|
||||||
|
|
||||||
|
|
||||||
(defn create-orders-channel [securities]
|
(defn create-orders-channel [securities]
|
||||||
(let [output (chan 1 (map edn/read-string))
|
(let [output (chan 1 (map edn/read-string))
|
||||||
url #(format "ws://localhost:%s/api/1/securities/%s/orders" 8080 %)
|
url #(format "ws://localhost:%s/api/1/securities/%s/orders" 8080 %)
|
||||||
@@ -79,7 +68,7 @@
|
|||||||
ret))
|
ret))
|
||||||
|
|
||||||
|
|
||||||
(defn benchmark [requests order trades]
|
(defn benchmark [requests orders]
|
||||||
(let [decoded-responses (chan 1 (map decode-response))
|
(let [decoded-responses (chan 1 (map decode-response))
|
||||||
responses (no-more-then-n-simultaneously #(%) (to-chan requests) 10)]
|
responses (no-more-then-n-simultaneously #(%) (to-chan requests) 10)]
|
||||||
(go (async/pipe responses decoded-responses))
|
(go (async/pipe responses decoded-responses))
|
||||||
@@ -108,9 +97,8 @@
|
|||||||
(defn -main [config]
|
(defn -main [config]
|
||||||
(let [benchmark-config (:benchmark config)
|
(let [benchmark-config (:benchmark config)
|
||||||
requests (-> benchmark-config :number-of-requests-per-securities (make-requests (:securities config)))
|
requests (-> benchmark-config :number-of-requests-per-securities (make-requests (:securities config)))
|
||||||
trades (create-trades-channel (:securities config))
|
|
||||||
orders (create-orders-channel (:securities config))
|
orders (create-orders-channel (:securities config))
|
||||||
results (benchmark requests orders trades)]
|
results (benchmark requests orders)]
|
||||||
(display-results (async/<!! (async/reduce compute-result {} results)))))
|
(display-results (async/<!! (async/reduce compute-result {} results)))))
|
||||||
|
|
||||||
|
|
||||||
+59
-47
@@ -4,51 +4,16 @@
|
|||||||
create-sell-book
|
create-sell-book
|
||||||
match-order-in-book]]
|
match-order-in-book]]
|
||||||
[skirmish.util :refer :all]
|
[skirmish.util :refer :all]
|
||||||
[clojure.core.async :refer [chan go >! <! pub]]))
|
[clojure.core.async :refer [chan go >! <! pub] :as async]))
|
||||||
|
|
||||||
|
|
||||||
|
;; utility functions
|
||||||
|
|
||||||
(def empty-book {:sell (create-sell-book)
|
(def empty-book {:sell (create-sell-book)
|
||||||
:buy (create-buy-book)
|
:buy (create-buy-book)
|
||||||
:id 0})
|
:id 0})
|
||||||
|
|
||||||
|
;; Bellow are all the functions to match the received order in the correct book
|
||||||
(defn retrieve-new-book-order-from-trade [trade book]
|
|
||||||
(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))
|
|
||||||
totally-fulfilled? (= (:book.order/quantity trade) (:quantity ret-order))]
|
|
||||||
(into ret-order {:status (if totally-fulfilled? :totally-fulfilled :partially-fulfilled)
|
|
||||||
:status-dual-id (:book.order/order-id trade)
|
|
||||||
:quantity new-quantity})));;;;
|
|
||||||
|
|
||||||
|
|
||||||
(defn ff [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)
|
|
||||||
:status-dual-id dual-id
|
|
||||||
:quantity new-order-quantity})))
|
|
||||||
|
|
||||||
|
|
||||||
(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 ff order trades))]
|
|
||||||
(interleave trades-orders processed-orders)))
|
|
||||||
|
|
||||||
|
|
||||||
(defn get-processed-orders [order leftover-order trades book]
|
|
||||||
"Computes all the intermediary orders that took place during
|
|
||||||
a the trade."
|
|
||||||
(let [ret [(assoc order :status :inserted)]
|
|
||||||
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))
|
|
||||||
leftover-order)]
|
|
||||||
ret))
|
|
||||||
|
|
||||||
|
|
||||||
(defn process-abstract [order reference-book same-type-book]
|
(defn process-abstract [order reference-book same-type-book]
|
||||||
(let [[trades new-reference-book leftover-order] (match-order-in-book reference-book order)
|
(let [[trades new-reference-book leftover-order] (match-order-in-book reference-book order)
|
||||||
new-same-type-book (if leftover-order (conj same-type-book leftover-order) same-type-book)]
|
new-same-type-book (if leftover-order (conj same-type-book leftover-order) same-type-book)]
|
||||||
@@ -72,17 +37,64 @@
|
|||||||
(process-sell-order book order)
|
(process-sell-order book order)
|
||||||
(process-buy-order book order)))
|
(process-buy-order book order)))
|
||||||
|
|
||||||
|
;; Bellow are all the function to transform the order and trades to
|
||||||
|
;; a list of order
|
||||||
|
|
||||||
(defn process-orders [initial-book incoming-order-channel]
|
(defn retrieve-new-book-order-from-trade [trade book]
|
||||||
|
(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))
|
||||||
|
totally-fulfilled? (= (:book.order/quantity trade) (:quantity ret-order))]
|
||||||
|
(into ret-order {:status (if totally-fulfilled? :totally-fulfilled :partially-fulfilled)
|
||||||
|
:status-dual-id (:book.order/order-id trade)
|
||||||
|
: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)
|
||||||
|
:status-dual-id dual-id
|
||||||
|
:quantity new-order-quantity})))
|
||||||
|
|
||||||
|
|
||||||
|
(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)))
|
||||||
|
|
||||||
|
|
||||||
|
(defn get-processed-orders [order leftover-order trades book]
|
||||||
|
"Computes all the intermediary orders that took place during
|
||||||
|
a the trade."
|
||||||
|
(let [ret [(assoc order :status :inserted)]
|
||||||
|
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))
|
||||||
|
leftover-order)]
|
||||||
|
ret))
|
||||||
|
|
||||||
|
|
||||||
|
;; bellow are the really interesting functions.
|
||||||
|
|
||||||
|
(defn create-exchange-for-security [initial-book incoming-order-channel]
|
||||||
|
;; output - trades channel, pub channel
|
||||||
|
;; - orders channel, pub channel
|
||||||
(let [trades-channel (chan)
|
(let [trades-channel (chan)
|
||||||
new-book-order-channel (chan)]
|
outcoming-orders-channel (chan)]
|
||||||
(go (loop [order (<! incoming-order-channel) book initial-book]
|
(async/go
|
||||||
(let [[new-book new-order new-trades] (process-order book order)]
|
(loop [order (<! incoming-order-channel) book initial-book]
|
||||||
(when (not (nil? new-order)) (>! new-book-order-channel new-order))
|
(println "received an order! " order)
|
||||||
(doseq [trade new-trades] (>! trades-channel trade)) ;; TODO: user clojure.core.async/onto-chan
|
(let [[new-book new-order new-trades] (process-order book order)
|
||||||
|
orders (get-processed-orders order new-order new-trades book)
|
||||||
|
orders-copy (async/onto-chan outcoming-orders-channel orders false)
|
||||||
|
trades-copy (async/onto-chan trades-channel new-trades false)]
|
||||||
|
(<! orders-copy)
|
||||||
|
(<! trades-copy)
|
||||||
(recur (<! incoming-order-channel) new-book))))
|
(recur (<! incoming-order-channel) new-book))))
|
||||||
[(pub trades-channel :book.order/type) (pub new-book-order-channel :type)]))
|
[(pub trades-channel :book.order/type) (pub outcoming-orders-channel :type)]))
|
||||||
|
|
||||||
|
|
||||||
(def create-exchange-for-security process-orders)
|
|
||||||
|
|
||||||
@@ -105,19 +105,16 @@
|
|||||||
exchange-api (get-exchange-api (keyword sec-id-str) exchange-apis)
|
exchange-api (get-exchange-api (keyword sec-id-str) exchange-apis)
|
||||||
incoming-order-channel (:incoming-order-channel exchange-api)
|
incoming-order-channel (:incoming-order-channel exchange-api)
|
||||||
order (validate-order asked-order (:sec-id exchange-api))]
|
order (validate-order asked-order (:sec-id exchange-api))]
|
||||||
@(kinsky.client/send! kafka-producer
|
(>!! incoming-order-channel order)
|
||||||
(str "new-orders-" sec-id-str)
|
|
||||||
(:book.order/id order)
|
|
||||||
order)
|
|
||||||
(build-response request order))
|
(build-response request order))
|
||||||
(catch [:error-type :4XX] e (build-response request e))
|
(catch [:error-type :4XX] e (build-response request e))
|
||||||
(catch [:error-type :5XX] e (build-response request e)))))
|
(catch [:error-type :5XX] e (build-response request e)))))
|
||||||
|
|
||||||
|
|
||||||
(defn chan-to-websocket [request chan ws-chan]
|
(defn chan-to-websocket [request input-chan ws-chan]
|
||||||
(on-close ws-chan (fn [status] (close! chan)))
|
(on-close ws-chan (fn [status] (close! input-chan)))
|
||||||
(go-loop []
|
(go-loop []
|
||||||
(when-let [e (<! chan)]
|
(when-let [e (<! input-chan)]
|
||||||
(send! ws-chan (build-response-body request e))
|
(send! ws-chan (build-response-body request e))
|
||||||
(recur))))
|
(recur))))
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user