Add tests for ewm and notimplementederrors

Author: mroeschke

streamz/dataframe/aggregations.py

@@ -148,24 +148,27 @@ def initial(self, new):
         return new.iloc[:0]
 
 
-class EWMean(Mean):
+class EWMean(Aggregation):
     def __init__(self, com):
         self.com = com
         alpha = 1. / (1. + self.com)
         self.old_wt_factor = 1. - alpha
         self.new_wt = 1.
 
     def on_new(self, acc, new):
-        result, old_wt = acc
-        for i in range(len(new)):
-            result = ((old_wt * result) + (self.new_wt * new.iloc[i])) / (old_wt + self.new_wt)
+        result, old_wt, is_first = acc
+        for i in range(int(is_first), len(new)):
             old_wt *= self.old_wt_factor
+            result = ((old_wt * result) + (self.new_wt * new.iloc[i])) / (old_wt + self.new_wt)
             old_wt += self.new_wt
-        return (result, old_wt), result
+        return (result, old_wt, False), result
 
     def on_old(self, acc, old):
         pass
 
+    def initial(self, new):
+        return new.iloc[:1], 1, True
+
 
 def diff_iloc(dfs, new, window=None):
     """ Emit new list of dfs and decayed data

streamz/dataframe/tests/test_dataframes.py

@@ -731,6 +731,33 @@ def test_expanding(func):
     assert_eq(result, expected)
 
 
+def test_ewm_mean():
+    sdf = DataFrame(example=pd.DataFrame(columns=['x', 'y']))
+    L = sdf.ewm(1).mean().stream.gather().sink_to_list()
+    sdf.emit(pd.DataFrame({'x': [1.], 'y': [2.]}))
+    sdf.emit(pd.DataFrame({'x': [2.], 'y': [3.]}))
+    sdf.emit(pd.DataFrame({'x': [3.], 'y': [4.]}))
+    result = pd.concat(L, ignore_index=True)
+
+    df = pd.DataFrame({'x': [1., 2., 3.], 'y': [2., 3., 4.]})
+    expected = df.ewm(1).mean()
+    assert_eq(result, expected)
+
+
+@pytest.mark.parametrize('func', [
+    lambda x: x.sum(),
+    lambda x: x.count(),
+    lambda x: x.apply(lambda x: x),
+    lambda x: x.full(),
+    lambda x: x.var(),
+    lambda x: x.std()
+], ids=["sum", "count", "apply", "full", "var", "std"])
+def test_ewm_notimplemented(func):
+    sdf = DataFrame(example=pd.DataFrame(columns=['x', 'y']))
+    with pytest.raises(NotImplementedError):
+        func(sdf.ewm(1))
+
+
 @pytest.mark.parametrize('func', [
     lambda x: x.sum(),
     lambda x: x.mean(),