dataflow.clj

...
(defn mk-cbm []
  (a/mult (chan 20000 (comp (map lib/parse-cb-msg)))))

(defn mk-cbp [cbm]
  (a/pub (a/tap cbm (chan 1000)) topic1))

(defn mk-userm [cbm]
  (a/mult (a/tap cbm (chan 1000 (filter :cb/profile_id)))))

(defn mk-odb [cbp opt] (a/sub cbp ::order (chan (a/sliding-buffer 1000) (order-flow->odb-xf opt))))

(defn mk-lobm []
  (let [lobs-ch (a/chan (a/sliding-buffer 1))
        lobm    (a/mult lobs-ch)]
    lobm))

(defn mk-px-domain-mult [cbp {:keys [ttl] :as opt}]
  (a/mult (a/sub cbp ::trade (chan (a/sliding-buffer 1) (price-domain-xf opt)))))

(defn mk-trade-scale [domain-mult]
  (a/tap domain-mult
    (chan (sb1) (map (fn [[mn mx]] (scale/scale-map [:linear] {:domain [mn mx]}))))))

(defn mk-trade-line [cbp {:keys [ttl] :as opt}]
  (a/sub cbp ::trade (chan (sb1) (time-price-line-xf opt))))

(defn mk-best-bids-line [lobm opt]
  (a/tap lobm (chan (sb1) (lob->best-xf ::lob/bids opt))))

(defn mk-best-asks-line [lobm opt]
  (a/tap lobm (chan (sb1) (lob->best-xf ::lob/asks opt))))

(defn mk-by-px-mult [lobm domain-mult]
  (a/mult (bounded-px-lobs (a/tap lobm (chan (sb1))) (a/tap domain-mult (chan (sb1))))))

(defn mk-side-by-px [by-px-mult side-key]
  (a/tap by-px-mult (chan (sb1) (map side-key))))

(defn mk-agg-side [by-px-mult side-key]
  (a/tap by-px-mult
    (chan (sb1) (map (fn [lob] (val-reductions + (side-key lob)))))))