bucket.clj

(defn bucket
  [accf init-acc splitf]
  (fn [rf]
    (let [acc-atom (atom init-acc)]
      (fn
        ([] (rf))
        ([result]
         (rf result))
        ([result input]
         (let [old-acc @acc-atom
               new-acc (accf old-acc input)
               split?  (splitf new-acc)]
           (if split?
             (do (reset! acc-atom (accf init-acc input))
                 (rf result old-acc))
             (do (reset! acc-atom new-acc)
                 result))))))))

(defn current-bucket
  [accf init-acc splitf  ]
  (fn [rf]
    (let [acc-atom (atom init-acc)]
      (fn
        ([] (rf))
        ([result]
         (rf result))
        ([result input]
         (let [old-acc @acc-atom
               new-acc (accf old-acc input)
               split?  (splitf new-acc)]
           (if split?
             (rf result (reset! acc-atom (accf init-acc input)))
             (do (reset! acc-atom new-acc)
                 (rf result new-acc)))))))))

(def pts
  [{:size 0.01M, :price 10660.06M}
   {:size 0.02M, :price 10657.21M}
   {:size 0.03M, :price 10657.21M}
   {:size 0.02M, :price 10657.21M}
   {:size 0.0272M, :price 10656.3M}
   {:size 0.0913M, :price 10654.16M}
   {:size 0.1143M, :price 10653.34M}
   {:size 0.0045M, :price 10651.92M}])

(->> pts
     (into []
       (bucket
         (fn [acc x]
           (-> acc
               (update :agg-size + (:size x))
               (update :points conj x)))
         {:agg-size 0 :points []}
         (fn [acc] (> (:agg-size acc) 0.1)))))

;;=>
[{:agg-size 0.08M,
  :points
  [{:size 0.01M, :price 10660.06M}
   {:size 0.02M, :price 10657.21M}
   {:size 0.03M, :price 10657.21M}
   {:size 0.02M, :price 10657.21M}]}
 {:agg-size 0.0272M, :points [{:size 0.0272M, :price 10656.3M}]}
 {:agg-size 0.0913M, :points [{:size 0.0913M, :price 10654.16M}]}
 {:agg-size 0.1143M, :points [{:size 0.1143M, :price 10653.34M}]}]


(->> pts
     (into []
       (current-bucket
         (fn [acc x]
           (-> acc
               (update :agg-size + (:size x))
               (update :points conj x)))
         {:agg-size 0 :points []}
         (fn [acc] (> (:agg-size acc) 0.1)))))

;;=>

[{:agg-size 0.01M, :points [{:size 0.01M, :price 10660.06M}]}
 {:agg-size 0.03M,
  :points [{:size 0.01M, :price 10660.06M}
           {:size 0.02M, :price 10657.21M}]}
 {:agg-size 0.06M,
  :points [{:size 0.01M, :price 10660.06M}
           {:size 0.02M, :price 10657.21M}
           {:size 0.03M, :price 10657.21M}]}
 {:agg-size 0.08M,
  :points [{:size 0.01M, :price 10660.06M}
           {:size 0.02M, :price 10657.21M}
           {:size 0.03M, :price 10657.21M}
           {:size 0.02M, :price 10657.21M}]}
 {:agg-size 0.0272M, :points [{:size 0.0272M, :price 10656.3M}]}
 {:agg-size 0.0913M, :points [{:size 0.0913M, :price 10654.16M}]}
 {:agg-size 0.1143M, :points [{:size 0.1143M, :price 10653.34M}]}
 {:agg-size 0.0045M, :points [{:size 0.0045M, :price 10651.92M}]}]