CSP compatibility with pandas 3.0 (#669)

Signed-off-by: Adam Glustein <adam.glustein@point72.com>

Author: AdamGlustein

.github/workflows/build.yml

@@ -628,6 +628,9 @@ jobs:
           package:
             - sqlalchemy<2
             - perspective-python<3
+            # - pandas<3
+            # uncomment the pandas pin once we move off Python 3.10 to ensure we maintain pandas 2.x compatibility
+            # pandas 3.0 does not support 3.10 so we currently get free coverage, but won't after 3.10 goes EOL
             # Min supported version of pandas 2.2
             - numpy==1.22.4
 

cpp/csp/cppnodes/statsimpl.h

@@ -776,7 +776,7 @@ class StandardError
 
         double compute() const
         {
-            return ( m_count > m_ddof ? sqrt( m_var.compute() / ( m_count - m_ddof ) ) : std::numeric_limits<double>::quiet_NaN() );
+            return ( m_count > m_ddof ? sqrt( m_var.compute() / m_count ) : std::numeric_limits<double>::quiet_NaN() );
         }
 
     private:
@@ -822,7 +822,7 @@ class WeightedStandardError
 
         double compute() const
         {
-            return ( m_wsum > m_ddof && m_wsum > EPSILON ? sqrt( m_var.compute() / ( m_wsum - m_ddof ) ) : std::numeric_limits<double>::quiet_NaN() );
+            return ( m_wsum > m_ddof && m_wsum > EPSILON ? sqrt( m_var.compute() / m_wsum ) : std::numeric_limits<double>::quiet_NaN() );
         }
 
     private:

csp/impl/pandas_ext_type.py

@@ -203,6 +203,7 @@ class _NumPyBackedExtensionArrayMixin(ExtensionArray, ExtensionScalarOpsMixin):
     # Code inspired by https://github.com/hgrecco/pint-pandas/blob/master/pint_pandas/pint_array.py
     # and https://github.com/ContinuumIO/cyberpandas/blob/master/cyberpandas/ip_array.py
     _data: np.ndarray
+    _readonly: bool = False
 
     @property
     def dtype(self) -> "TsDtype":
@@ -235,9 +236,14 @@ def __getitem__(self, item) -> Any:
 
         item = check_array_indexer(self, item)
 
-        return self.__class__(self._data[item], self.dtype)
+        cls = self.__class__(self._data[item], self.dtype)
+        cls._readonly = self._readonly
+        return cls
 
     def __setitem__(self, key, value):
+        if self._readonly:
+            raise ValueError("Cannot modify read-only array")
+
         # need to not use `not value` on numpy arrays
         if isinstance(value, (list, tuple)) and (not value):
             # doing nothing here seems to be ok
@@ -390,6 +396,9 @@ def __init__(self, values, dtype=None, copy=False):
             self._data = values._data
         else:
             self._data = np.atleast_1d(np.asarray(values, dtype="O"))
+            # Ensure the array is writeable (may be read-only in pandas 3.x)
+            if not self._data.flags.writeable:
+                self._data = self._data.copy()
 
         if dtype is None:
             try:
@@ -439,7 +448,7 @@ def formatting_function(x):
 
         return formatting_function
 
-    def _reduce(self, name, skipna=True, **kwds):
+    def _reduce(self, name, skipna=True, keepdims: bool = False, **kwds):
         """
         Return a scalar result of performing the reduction operation.
         Parameters
@@ -450,12 +459,15 @@ def _reduce(self, name, skipna=True, **kwds):
             std, var, sem, kurt, skew }.
         skipna : bool, default True
             If True, skip NaN values (both missing Edges and nan's inside of Edges)
+        keepdims : bool, default False
+            If True, return a TsArray with shape (1,) instead of a scalar, keeping the dimensions.
+            Used by DataFrame reductions. Required argument as of pandas 3.0: https://github.com/pandas-dev/pandas/pull/52788.
         **kwargs
             Additional keyword arguments passed to the reduction function.
             Currently, `ddof` is the only supported kwarg.
         Returns
         -------
-        scalar
+        scalar or TsArray
         Raises
         ------
         TypeError : subclass does not define reductions
@@ -486,9 +498,24 @@ def _reduce(self, name, skipna=True, **kwds):
             quantity = self._data
 
         if len(quantity) == 0:
-            return None
+            result = None
+        else:
+            result = functions[name](quantity.tolist(), self.dtype.subtype, kwargs=dict(skipna=skipna))
+
+        if keepdims:
+            # Determine the output dtype based on the reduction operation
+            if name in ("any", "all"):
+                result_dtype = TsDtype(bool)
+            elif name in ("mean", "median", "std", "var", "sem", "skew", "kurt"):
+                result_dtype = TsDtype(float)
+            else:
+                result_dtype = self.dtype
+
+            if result is None:
+                return TsArray([], dtype=result_dtype)
+            return TsArray([result], dtype=result_dtype)
 
-        return functions[name](quantity.tolist(), self.dtype.subtype, kwargs=dict(skipna=skipna))
+        return result
 
     @classmethod
     def _create_comparison_method(cls, op):
@@ -537,6 +564,36 @@ def astype(self, dtype, copy=True):
                 return TsArray(values, dtype)
         return super(TsArray, self).astype(dtype, copy=copy)
 
+    def to_numpy(self, dtype=None, copy=False, na_value=None):
+        """Convert to a NumPy ndarray.
+
+        Parameters
+        ----------
+        dtype : str or numpy.dtype, optional
+            The dtype to pass to numpy.asarray.
+        copy : bool, default False
+            Whether to ensure that the returned value is not a view on another array.
+        na_value : Any, optional
+            The value to use for missing values.
+
+        Returns
+        -------
+        numpy.ndarray
+        """
+        result = np.asarray(self._data, dtype=dtype)
+
+        if copy:
+            result = result.copy()
+        elif self._readonly:
+            # Need to make array not writeable if this TsArray is read-only
+            if np.may_share_memory(result, self._data):
+                result = result.view()
+                result.flags.writeable = False
+
+        if na_value is not None:
+            result[self.isna()] = na_value
+        return result
+
 
 @node
 def _cast(x: ts[object], typ: "T") -> ts["T"]:

csp/tests/adapters/test_parquet.py

@@ -205,10 +205,11 @@ def graph():
                 csp.run(graph, starttime=start_time)
                 df = pandas.read_parquet(filename)
                 expected_columns = {}
+                # pandas 3.0+: need to explicitly force ns unit and UTC tz, no longer the default
                 if timestamp_column_name:
-                    expected_columns[timestamp_column_name] = [
-                        start_time + timedelta(seconds=sec + 1) for sec in range(10)
-                    ]
+                    expected_columns[timestamp_column_name] = pandas.date_range(
+                        start=start_time + timedelta(seconds=1), periods=10, unit="ns", freq="1s", tz="UTC"
+                    )
                 expected_columns.update(
                     {
                         "x": list(range(1, 11)),
@@ -241,7 +242,10 @@ def graph():
 
             expected_df = pandas.DataFrame.from_dict(
                 {
-                    "timestamp": [start_time + timedelta(seconds=sec + 1) for sec in range(10)],
+                    # pandas 3.0+: need to explicitly force ns unit and UTC tz, no longer the default
+                    "timestamp": pandas.date_range(
+                        start=start_time + timedelta(seconds=1), periods=10, unit="ns", freq="1s", tz="UTC"
+                    ),
                     "x": list(range(1, 11)),
                     "y": list(map(float, range(0, 100, 10))),
                 }

csp/tests/impl/test_pandas.py

@@ -1,5 +1,6 @@
 import unittest
 from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
 
 import pandas as pd
 from numpy import nan
@@ -27,7 +28,7 @@ def test_make_pandas_basic(self):
             self.assertEqual(out[2][0], start + 3 * dt2)
 
         ## First Tick
-        idx = pd.DatetimeIndex([start + dt2])
+        idx = pd.DatetimeIndex([start + dt2], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [2], "y": [1]}, index=idx)
         pd.testing.assert_frame_equal(out1[0][1], target)
         pd.testing.assert_frame_equal(out2[0][1], target)
@@ -37,7 +38,7 @@ def test_make_pandas_basic(self):
         # Notes:
         # - pandas is a bit inconsistent with whether or not it sets freq on the DateTimeIndex, so we drop for comparison.
         # - when there is missing data in an integer column, it uses float NaN and hence the column becomes float type
-        idx = pd.DatetimeIndex([start + dt2, start + dt2 + dt1, start + dt2 + 2 * dt1])
+        idx = pd.DatetimeIndex([start + dt2, start + dt2 + dt1, start + dt2 + 2 * dt1], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [2, 3, 4], "y": [1.0, nan, 2.0]}, index=idx)
         out1[1][1].index.freq = None
         pd.testing.assert_frame_equal(out1[1][1], target)
@@ -50,7 +51,8 @@ def test_make_pandas_basic(self):
 
         ## Third Tick
         idx = pd.DatetimeIndex(
-            [start + dt2, start + dt2 + dt1, start + dt2 + 2 * dt1, start + dt2 + 3 * dt1, start + dt2 + 4 * dt1]
+            [start + dt2, start + dt2 + dt1, start + dt2 + 2 * dt1, start + dt2 + 3 * dt1, start + dt2 + 4 * dt1],
+            dtype="datetime64[ns]",
         )
         target = pd.DataFrame({"x": [2, 3, 4, 5, 6], "y": [1.0, nan, 2.0, nan, 3.0]}, index=idx)
         out1[2][1].index.freq = None
@@ -83,7 +85,7 @@ def test_make_pandas_window(self):
             self.assertEqual(out[1][0], start + 2 * dt2)
 
         ## First Tick
-        idx = pd.DatetimeIndex([start + dt2])
+        idx = pd.DatetimeIndex([start + dt2], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [2], "y": [1]}, index=idx)
         pd.testing.assert_frame_equal(out1[0][1], target)
         pd.testing.assert_frame_equal(out2[0][1], target)
@@ -93,17 +95,17 @@ def test_make_pandas_window(self):
         # Notes:
         # - pandas is a bit inconsistent with whether or not it sets freq on the DateTimeIndex, so we drop for comparison.
         # - when there is missing data in an integer column, it uses float NaN and hence the column becomes float type
-        idx = pd.DatetimeIndex([start + dt2, start + dt2 + dt1, start + dt2 + 2 * dt1])
+        idx = pd.DatetimeIndex([start + dt2, start + dt2 + dt1, start + dt2 + 2 * dt1], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [nan, 3.0, 4.0], "y": [1.0, nan, 2.0]}, index=idx)
         out1[1][1].index.freq = None
         pd.testing.assert_frame_equal(out1[1][1], target)
 
-        idx = pd.DatetimeIndex([start + dt2 + dt1, start + dt2 + 2 * dt1])
+        idx = pd.DatetimeIndex([start + dt2 + dt1, start + dt2 + 2 * dt1], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [3, 4], "y": [nan, 2.0]}, index=idx)
         out2[1][1].index.freq = None
         pd.testing.assert_frame_equal(out2[1][1], target)
 
-        idx = pd.DatetimeIndex([start + dt2 + 2 * dt1])
+        idx = pd.DatetimeIndex([start + dt2 + 2 * dt1], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [4], "y": [2]}, index=idx)
         out3[1][1].index.freq = None
         pd.testing.assert_frame_equal(out3[1][1], target)
@@ -135,19 +137,19 @@ def test_make_pandas_init(self):
         target = pd.DataFrame({"x": pd.Series(dtype="int64"), "y": pd.Series(dtype="int64")}, index=idx)
         pd.testing.assert_frame_equal(out1[0][1], target)
 
-        idx = pd.DatetimeIndex([start + dt1])
+        idx = pd.DatetimeIndex([start + dt1], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [1], "y": [nan]}, index=idx)
         pd.testing.assert_frame_equal(out1[1][1], target)
         pd.testing.assert_frame_equal(out1[2][1], target)
 
-        idx = pd.DatetimeIndex([start + dt1, start + dt2])
+        idx = pd.DatetimeIndex([start + dt1, start + dt2], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [1, 2], "y": [nan, 1.0]}, index=idx)
         pd.testing.assert_frame_equal(out1[3][1], target)
 
         ## out2
         self.assertEqual(len(out2), 1)
         self.assertEqual(out2[0][0], start + dt2)
-        idx = pd.DatetimeIndex([start + dt2])
+        idx = pd.DatetimeIndex([start + dt2], dtype="datetime64[ns]")
         target = pd.DataFrame({"x": [2], "y": [1]}, index=idx)
         out2[0][1].index.freq = None
         pd.testing.assert_frame_equal(out2[0][1], target)
@@ -201,9 +203,16 @@ def test_pandas_ts_types(self):
         df = csp.DataFrame({("A", "x"): x, ("B", "x"): csp.const("Test"), ("A", "y"): y, ("B", "y"): csp.const(start)})
         out = csp.run(lambda: df.to_pandas_ts(("A", "y"), 2, ("A", "x")), starttime=start, endtime=end)[0]
         self.assertEqual(len(out), 3)
-        idx = pd.DatetimeIndex([start + dt2 + dt1, start + dt2 + 2 * dt1])
+        idx = pd.DatetimeIndex([start + dt2 + dt1, start + dt2 + 2 * dt1], dtype="datetime64[ns]")
         target = pd.DataFrame(
-            {("A", "x"): [3, 4], ("B", "x"): ["Test", "Test"], ("A", "y"): [1, 2], ("B", "y"): [start, start]},
+            {
+                ("A", "x"): [3, 4],
+                ("B", "x"): ["Test", "Test"],
+                ("A", "y"): [1, 2],
+                ("B", "y"): [
+                    pd.Timestamp(start.replace(tzinfo=ZoneInfo("UTC")).timestamp() * 1e9, unit="ns") for i in range(2)
+                ],
+            },
             index=idx,
         )
         self.assertEqual(out[1][0], start + dt2 + 2 * dt1)