ohlc_time_series_circular_buffer.clj

(import 'com.google.common.collect.EvictingQueue)
(let [queues (atom {})]
  (defn ohlc-circular-cache [size seconds ^String id ^Fn get-ohlc-data-fn]
    "`size`: Size of circular buffer.
`seconds`: Time frame of data series. I.e. number of seconds between OHLC .open and .close.
`id`: Unique ID string used to identify this circular buffer.
`get-ohlc-data-fn`: (fn [timestamp] ...) Get the most recent data series from timestamp. If timestamp is NIL get as much data as possible. In both cases the data returned will include the most recently closed OHLC entry (candle). Note that `get-ohlc-data-fn` will only be called once – so if stale data is returned you must deal with this either there or in code that depends on OHLC-CIRCULAR-CACHE directly or indirectly."
    (let [seconds (long seconds)
          ^com.google.common.collect.EvictingQueue
          queue (locking queues
                  (or (get @queues id)
                      (with1 (EvictingQueue/create size)
                        (swap! queues #(assoc % id it)))))]
      (locking queue
        (let [prev-oseries (into-array OHLC queue)]
          (if (or (empty? prev-oseries)
                  (time/after? (time/now) ;; Has enough time passed for there to even be a point in requesting new data?
                               (time/plus (.timestamp ^OHLC (last prev-oseries))
                                          (time/seconds (* 2 seconds)))))
            (let [next-oseries (if (empty? prev-oseries)
                                 (get-ohlc-data-fn nil)
                                 (get-ohlc-data-fn (time/plus (.timestamp ^OHLC (last prev-oseries))
                                                              (time/seconds seconds))))]
              (when (or (empty? prev-oseries)
                        (not (empty? next-oseries)))
                (.addAll queue next-oseries))
              (into-array OHLC queue))
            prev-oseries))))))