Add allow_non_overlapping parameter for bivariate statistics cov, corr and ema_cov

Signed-off-by: Adam Glustein <adamglustein@gmail.com>

Author: AdamGlustein

cpp/csp/cppnodes/statsimpl.cpp

@@ -108,6 +108,33 @@ DECLARE_CPPNODE( _in_sequence_check )
 
 EXPORT_CPPNODE( _in_sequence_check );
 
+/*
+@csp.node
+def _discard_non_overlapping(x: ts[float], y: ts[float]):
+*/
+
+DECLARE_CPPNODE( _discard_non_overlapping )
+{
+    TS_INPUT( double, x );
+    TS_INPUT( double, y );
+    
+    TS_NAMED_OUTPUT( double, x_sync );
+    TS_NAMED_OUTPUT( double, y_sync );
+
+    INIT_CPPNODE( _discard_non_overlapping ) { }
+
+    INVOKE()
+    {
+        if( csp.ticked( x ) && csp.ticked( y ) )
+        {
+            x_sync.output( x );
+            y_sync.output( y );
+        }
+    };
+};
+
+EXPORT_CPPNODE( _discard_non_overlapping );
+
 /*
 @csp.node(cppimpl=_cspstatsimpl._sync_nan_f)
 def _sync_nan_f(x: ts[float], y: ts[float]) -> csp.Outputs(x_sync=ts[float], y_sync=ts[float]):

cpp/csp/python/cspstatsimpl.cpp

@@ -7,6 +7,7 @@ REGISTER_CPPNODE( csp::cppnodes,  _time_window_updates );
 REGISTER_CPPNODE( csp::cppnodes,  _cross_sectional_as_list );
 REGISTER_CPPNODE( csp::cppnodes,  _min_hit_by_tick );
 REGISTER_CPPNODE( csp::cppnodes,  _in_sequence_check );
+REGISTER_CPPNODE( csp::cppnodes,  _discard_non_overlapping );
 REGISTER_CPPNODE( csp::cppnodes,  _sync_nan_f );
 
 // Base statistics

csp/stats.py

@@ -151,6 +151,12 @@ def _in_sequence_check(x: ts["T"], y: ts["T"]):
     raise NotImplementedError("_in_sequence_check only implemented in C++")
 
 
+@csp.node(cppimpl=_cspstatsimpl._discard_non_overlapping)
+def _discard_non_overlapping(x: ts[float], y: ts[float]) -> csp.Outputs(x_sync=ts[float], y_sync=ts[float]):
+    raise NotImplementedError("_discard_non_overlapping only implemented in C++")
+    return csp.output(x_sync=0, y_sync=0)
+
+
 @csp.node(cppimpl=_cspstatsimpl._sync_nan_f)
 def _sync_nan_f(x: ts[float], y: ts[float]) -> csp.Outputs(x_sync=ts[float], y_sync=ts[float]):
     raise NotImplementedError("_sync_nan_f only implemented in C++")
@@ -176,12 +182,6 @@ def _combine_signal(x: ts["T"], y: ts["U"]) -> ts[bool]:
         return True
 
 
-@csp.node
-def _combine_signal(x: ts["T"], y: ts["U"]) -> ts[bool]:
-    if csp.ticked(x, y):
-        return True
-
-
 @csp.node
 def _np_log(x: ts[np.ndarray]) -> ts[np.ndarray]:
     return np.log(x)
@@ -280,21 +280,32 @@ def _setup(x, interval, min_window, trigger, sampler, reset, weights=None, ignor
     return series, interval, min_window, trigger, min_hit, updates, sampler, reset, weights, recalc, clear_stat
 
 
+def _synchronize_bivariate(x, y, allow_non_overlapping):
+    """
+    If allow_non_overlapping=True, discard any out-of-sync ticks between and y. Else, raise an exception when this occurs.
+    """
+    in_seq = None
+    if x is not y:
+        if allow_non_overlapping:
+            sync = _discard_non_overlapping(x, y)
+            x, y = sync.x_sync, sync.y_sync
+        else:
+            in_seq = _in_sequence_check(x, y)
+    return x, y, in_seq
+
+
 def _bivariate_setup(
-    x, y, interval, min_window, trigger, sampler, reset, weights=None, ignore_weights=False, recalc=None
+    x, y, interval, min_window, trigger, sampler, reset, weights=None, recalc=None, allow_non_overlapping=False
 ):
     """
     Sets up time-series window updates and triggers for a bivariate stats calculation
     """
+    x, y, in_seq = _synchronize_bivariate(x, y, allow_non_overlapping)
     x_upd = _setup(x, interval, min_window, trigger, sampler, reset, weights, False, recalc)[5]
     series, interval, min_window, trigger, min_hit, y_upd, sampler, reset, weights, recalc, clear_stat = _setup(
         y, interval, min_window, trigger, sampler, reset, weights, False, recalc
     )
 
-    in_seq = None
-    if x is not y:
-        in_seq = _in_sequence_check(x, y)
-
     return (
         series,
         interval,
@@ -2181,25 +2192,27 @@ def cov(
     reset: ts[object] = None,
     recalc: ts[object] = None,
     min_data_points: int = 0,
+    allow_non_overlapping: bool = False,
 ) -> ts[Union[float, np.ndarray]]:
     """
 
     Returns the covariance between two in-sequence time-series within the given window. If the time-series are of type np.ndarray, the covariance is calculated elementwise.
 
     Inputs
-    x:              time series data, of type float or np.ndarray
-    y:              time series data, of type float or np.ndarray, which ticks at the same time as x
-    interval:       the window interval (either time or tick specified)
-    min_window:     the minimum window (either time or tick specified) before statistics are returned. Must be the same type as interval
-    ddof:           delta degrees of freedom
-    ignore_na:      if True, will treat NaN values as missing data. If False, a NaN present in the window will make the computed statistic NaN as well
-    trigger:        another time-series which specifies when you want to recalculate the statistic
-    weights:        another time-series which specifies the weights to use on each x value, if a weighted covariance is desired
-    sampler:        another time-series which specifies when x should tick. If x ticks when sampler does not, the data is ignored. If sampler ticks when x does not,
-                    the data point is treated as NaN
-    reset:          another time-series which will clear the data in the window when it ticks
-    recalc:         another time-series which triggers a clean recalculation of the window statistic, and in doing so clears any accumulated floating-point error
-    min_data_points: minimum number of current ticks in the interval needed for a valid computation. If there are fewer ticks, NaN is returned.
+    x:                      time series data, of type float or np.ndarray
+    y:                      time series data, of type float or np.ndarray, which ticks at the same time as x
+    interval:               the window interval (either time or tick specified)
+    min_window:             the minimum window (either time or tick specified) before statistics are returned. Must be the same type as interval
+    ddof:                   delta degrees of freedom
+    ignore_na:              if True, will treat NaN values as missing data. If False, a NaN present in the window will make the computed statistic NaN as well
+    trigger:                another time-series which specifies when you want to recalculate the statistic
+    weights:                another time-series which specifies the weights to use on each x value, if a weighted covariance is desired
+    sampler:                another time-series which specifies when x should tick. If x ticks when sampler does not, the data is ignored. If sampler ticks when x does not,
+                            the data point is treated as NaN
+    reset:                  another time-series which will clear the data in the window when it ticks
+    recalc:                 another time-series which triggers a clean recalculation of the window statistic, and in doing so clears any accumulated floating-point error
+    min_data_points:        minimum number of current ticks in the interval needed for a valid computation. If there are fewer ticks, NaN is returned.
+    allow_non_overlapping:  if True, discard any ticks of x and y that occur out-of-sync with one another. If False, raise an exception on any out-of-sync ticks.
 
     """
 
@@ -2217,7 +2230,7 @@ def cov(
         recalc,
         clear_stat,
         in_seq,
-    ) = _bivariate_setup(x, y, interval, min_window, trigger, sampler, reset, weights, True, recalc)
+    ) = _bivariate_setup(x, y, interval, min_window, trigger, sampler, reset, weights, recalc, allow_non_overlapping)
 
     # Use same "debiasing" for weighted/non-weighted
     edge = None
@@ -2543,24 +2556,26 @@ def corr(
     reset: ts[object] = None,
     recalc: ts[object] = None,
     min_data_points: int = 0,
+    allow_non_overlapping: bool = False,
 ) -> ts[Union[float, np.ndarray]]:
     """
 
     Returns the correlation between x and y within the given window. If the time-series are of type np.ndarray, the correlation is calculated elementwise.
 
     Inputs
-    x:              time series data, of type float or np.ndarray
-    y:              time series data, of type float or np.ndarray, which ticks at the same time x ticks
-    interval:       the window interval (either time or tick specified)
-    min_window:     the minimum window (either time or tick specified) before statistics are returned. Must be the same type as interval
-    ignore_na:      if True, will treat NaN values as missing data. If False, a NaN present in the window will make the computed statistic NaN as well
-    trigger:        another time-series which specifies when you want to recalculate the statistic
-    weights:        another time-series which specifies the weights to use on each x value, if a weighted correlation is desired
-    sampler:        another time-series which specifies when x should tick. If x ticks when sampler does not, the data is ignored. If sampler ticks when x does not,
-                    the data point is treated as NaN
-    reset:          another time-series which will clear the data in the window when it ticks
-    recalc:         another time-series which triggers a clean recalculation of the window statistic, and in doing so clears any accumulated floating-point error
-    min_data_points: minimum number of current ticks in the interval needed for a valid computation. If there are fewer ticks, NaN is returned.
+    x:                      time series data, of type float or np.ndarray
+    y:                      time series data, of type float or np.ndarray, which ticks at the same time x ticks
+    interval:               the window interval (either time or tick specified)
+    min_window:             the minimum window (either time or tick specified) before statistics are returned. Must be the same type as interval
+    ignore_na:              if True, will treat NaN values as missing data. If False, a NaN present in the window will make the computed statistic NaN as well
+    trigger:                another time-series which specifies when you want to recalculate the statistic
+    weights:                another time-series which specifies the weights to use on each x value, if a weighted correlation is desired
+    sampler:                another time-series which specifies when x should tick. If x ticks when sampler does not, the data is ignored. If sampler ticks when x does not,
+                            the data point is treated as NaN
+    reset:                  another time-series which will clear the data in the window when it ticks
+    recalc:                 another time-series which triggers a clean recalculation of the window statistic, and in doing so clears any accumulated floating-point error
+    min_data_points:        minimum number of current ticks in the interval needed for a valid computation. If there are fewer ticks, NaN is returned.
+    allow_non_overlapping:  if True, discard any ticks of x and y that occur out-of-sync with one another. If False, raise an exception on any out-of-sync ticks.
 
     """
     (
@@ -2577,7 +2592,7 @@ def corr(
         recalc,
         clear_stat,
         in_seq,
-    ) = _bivariate_setup(x, y, interval, min_window, trigger, sampler, reset, weights, True, recalc)
+    ) = _bivariate_setup(x, y, interval, min_window, trigger, sampler, reset, weights, recalc, allow_non_overlapping)
 
     if series.tstype.typ is float:
         if weights is not None:
@@ -2965,38 +2980,40 @@ def ema_cov(
     reset: ts[object] = None,
     recalc: ts[object] = None,
     min_data_points: int = 0,
+    allow_non_overlapping: bool = False,
 ) -> ts[Union[float, np.ndarray]]:
     """
 
     Returns the exponential moving covariance between two time series.
 
     Inputs
-    x:              time series data, of type float or np.ndarray
-    y:              time series data, of type float or np.ndarray, which ticks at the same time as x
-    min_periods:    the minimum number of data points before statistics are returned
-    alpha:          specify the decay parameter in terms of alpha
-    span:           specify the decay parameter in terms of span
-    com:            specify the decay parameter in terms of com
-    halflife:       specify the decay parameter in terms of halflife
-    adjust:         if True, an adjusted EMA will be computed. If False, a standard (unadjusted) EMA will be computed
-    horizon:        if specified, values that are older than the horizon will be removed entirely from the computation (essentially making EMA a window computation)
-    bias:           if True, a biased EMA covariance is computed. If False, the covariance estimate is unbiased
-    ignore_na:      if True, NaNs will be ignored and have no effect on the computation. If False, a NaN will shift the observation window once new non-NaN data comes in
-    trigger:        another time-series which specifies when you want to recalculate the statistic
-    sampler:        another time-series which specifies when x should tick. If x ticks when sampler does not, the data is ignored. If sampler ticks when x does not,
-                    the data point is treated as NaN
-    reset:          another time-series which will clear the data in the window when it ticks
-    recalc:         only valid when a finite-horizon EMA is used. Another time-series which triggers a clean recalculation of the window statistic, and in doing so clears any accumulated floating-point error
-    min_data_points: minimum number of current ticks in the interval needed for a valid computation. If there are fewer ticks, NaN is returned.
+    x:                      time series data, of type float or np.ndarray
+    y:                      time series data, of type float or np.ndarray, which ticks at the same time as x
+    min_periods:            the minimum number of data points before statistics are returned
+    alpha:                  specify the decay parameter in terms of alpha
+    span:                   specify the decay parameter in terms of span
+    com:                    specify the decay parameter in terms of com
+    halflife:               specify the decay parameter in terms of halflife
+    adjust:                 if True, an adjusted EMA will be computed. If False, a standard (unadjusted) EMA will be computed
+    horizon:                if specified, values that are older than the horizon will be removed entirely from the computation (essentially making EMA a window computation)
+    bias:                   if True, a biased EMA covariance is computed. If False, the covariance estimate is unbiased
+    ignore_na:              if True, NaNs will be ignored and have no effect on the computation. If False, a NaN will shift the observation window once new non-NaN data comes in
+    trigger:                another time-series which specifies when you want to recalculate the statistic
+    sampler:                another time-series which specifies when x should tick. If x ticks when sampler does not, the data is ignored. If sampler ticks when x does not,
+                            the data point is treated as NaN
+    reset:                  another time-series which will clear the data in the window when it ticks
+    recalc:                 only valid when a finite-horizon EMA is used. Another time-series which triggers a clean recalculation of the window statistic, and in doing so clears any accumulated floating-point error
+    min_data_points:        minimum number of current ticks in the interval needed for a valid computation. If there are fewer ticks, NaN is returned.
+    allow_non_overlapping:  if True, discard any ticks of x and y that occur out-of-sync with one another. If False, raise an exception on any out-of-sync ticks.
 
     """
 
     alpha, interval, min_window, debias_horizon, recalc = _validate_ema(
         alpha, span, com, halflife, adjust, horizon, recalc
     )
 
+    x, y, _ = _synchronize_bivariate(x, y, allow_non_overlapping)
     if x is not y:
-        _in_sequence_check(x, y)
         sync = _sync_nan(x, y)
         x, y = sync.x_sync, sync.y_sync
 

csp/tests/test_stats.py

@@ -3599,6 +3599,47 @@ def g():
         self.assertTrue(pd.Series(res["kurt"][1]).isna().all())
         self.assertTrue(pd.Series(res["corr"][1]).isna().all())
 
+    def test_allow_non_overlapping_bivariate(self):
+        st = datetime(2020, 1, 1)
+
+        @csp.graph
+        def g(allow_non_overlapping: bool):
+            x = csp.curve(
+                typ=float,
+                data=[
+                    (datetime(2020, 1, 1), 1),
+                    (datetime(2020, 1, 2), 2),
+                    (datetime(2020, 1, 4), 4),  # discarded
+                    (datetime(2020, 1, 6), 6),  # discarded
+                    (datetime(2020, 1, 7), 7),
+                ],
+            )
+            y = csp.curve(
+                typ=float,
+                data=[
+                    (datetime(2020, 1, 1), 1),
+                    (datetime(2020, 1, 2), 2),
+                    (datetime(2020, 1, 3), -1),  # discarded
+                    (datetime(2020, 1, 5), -2),  # discarded
+                    (datetime(2020, 1, 7), 7),
+                ],
+            )
+            cov = csp.stats.cov(x, y, 5, 1, allow_non_overlapping=allow_non_overlapping)
+            corr = csp.stats.corr(x, y, 5, 1, allow_non_overlapping=allow_non_overlapping)
+            ema_cov = csp.stats.ema_cov(x, y, 1, alpha=0.1, allow_non_overlapping=allow_non_overlapping)
+            csp.add_graph_output("cov", cov)
+            csp.add_graph_output("corr", corr)
+            csp.add_graph_output("ema_cov", ema_cov)
+
+        res = csp.run(g, True, starttime=st, endtime=timedelta(days=8), output_numpy=True)
+        for output in res.values():
+            # Convert expected datetimes to the same type as output
+            expected_dates = pd.to_datetime([datetime(2020, 1, i) for i in (1, 2, 7)])
+            np.testing.assert_array_equal(output[0], expected_dates.values)
+
+        # Additional sanity check is that correlation should be 1 as middle ticks are ignored
+        np.testing.assert_allclose(res["corr"][1], np.array([np.nan, 1.0, 1.0]), equal_nan=True)
+
 
 if __name__ == "__main__":
     unittest.main()