Windowing fix conforming to Pandas

Author: chinmaychandak

streamz/dataframe/aggregations.py

@@ -209,9 +209,9 @@ def diff_loc(dfs, new, window=None):
     old = []
     if len(dfs) > 0:
         mx = max(df.index.max() for df in dfs)
-        mn = mx - pd.Timedelta(window)
+        mn = mx - pd.Timedelta(window) + pd.Timedelta('1ns')
         while pd.Timestamp(dfs[0].index.min()) < mn:
-            o = dfs[0].loc[:mn - pd.Timedelta('1ns')]
+            o = dfs[0].loc[:mn]
             old.append(o)  # TODO: avoid copy if fully lost
             dfs[0] = dfs[0].iloc[len(o):]
             if not len(dfs[0]):

streamz/dataframe/tests/test_dataframes.py

@@ -755,11 +755,11 @@ def f(x):
     assert len(L) == 5
 
     first = df.iloc[:diff]
-    first = first[first.index.max() - value - pd.Timedelta('1ns'):]
+    first = first[first.index.max() - value + pd.Timedelta('1ns'):]
 
     assert_eq(L[0], f(first))
 
-    last = df.loc[index.max() - value - pd.Timedelta('1ns'):]
+    last = df.loc[index.max() - value + pd.Timedelta('1ns'):]
 
     assert_eq(L[-1], f(last))