       (fn [[ev' ev]] (assoc ev' :metric (- (:metric ev') (:metric ev))))
       (map vector (rest l) l)))
     children))
+
+
+(defn |>| [& args] (apply > (map #(Math/abs %) args)))
+(defn |<| [& args] (apply < (map #(Math/abs %) args)))
+
+(defn maxpdiff
+  "计算最大变化率，与 aggregate 搭配使用(MAX Percentage DIFFerence)"
+  [& m]
+  (let [r (last m)]
+    (->> (map #(/ (- r %) r) m)
+         (reduce (fn [v v'] (if (> (Math/abs v') (Math/abs v)) v' v))))))
 
 
 (defn feed-dog

  [from to & children]
  (apply smap #(assoc % to (from %)) children))


(defn |>| [& args] (apply > (map #(Math/abs %) args)))
(defn |<| [& args] (apply < (map #(Math/abs %) args)))

(defn maxpdiff
  "计算最大变化率，与 aggregate 搭配使用(MAX Percentage DIFFerence)"
  [& m]
  (let [r (last m)]
    (->> (map #(/ (- r %) r) m)
         (reduce (fn [v v'] (if (> (Math/abs v') (Math/abs v)) v' v))))))


(defn ->difference
  "接受事件的数组，变换成另外一个事件数组。
   新的事件数组中，每一个事件的 metric 是之前相邻两个事件 metric 的差。

        (smap :metric
          (with {:state :ok}
            (apply sdo children)))))))

(defn group-window
  "将事件分组后向下传递，类似 fixed-time-window，但不使用时间切割，
  而是通过 (group-fn event) 的值进行切割。(group-fn event) 的值会被记录下来，
  每一次出现重复值的时候，会将当前缓存住的事件数组向下传递。

  比如你有一组同质的机器，跑了相同的服务，但是机器名不一样，可以通过
  (group-window :host
    ....)
  将事件分组后处理（e.g. 对单台的容量求和获得总体容量）
  e.g.: 一个事件流中的事件先后到达，其中 :host 的值如下
      a b c d b a c a b
  那么会被这个流分成
    [a b c d] [b a c] [a b]
  分成 3 次向下游传递
  "
  [group-fn & children]
  (let [buffer (ref [])
        group-keys (ref #{})]
    (fn stream [event]
      (let [evkey (group-fn event)]
        (-> (dosync
              (if (@group-keys evkey)
                (let [rst @buffer]
                  (ref-set buffer [event])
                  (ref-set group-keys #{evkey})
                  rst)
                (do
                  (alter buffer conj event)
                  (alter group-keys conj evkey)
                  nil)))
            ((fn [rst]
              (when rst
                (call-rescue rst children)))))))))
