BUG: .rolling() returns incorrect values when ts index is not nano seconds

doc/source/whatsnew/v2.1.2.rst

@@ -15,6 +15,7 @@ Fixed regressions
 ~~~~~~~~~~~~~~~~~
 - Fixed bug in :meth:`DataFrame.resample` where bin edges were not correct for :class:`~pandas.tseries.offsets.MonthBegin` (:issue:`55271`)
 - Fixed bug where PDEP-6 warning about setting an item of an incompatible dtype was being shown when creating a new conditional column (:issue:`55025`)
+- Fixed bug in :class:`pandas.core.window.Rolling` where non-nanosecond ``on`` would produce incorrect results (:issue:`55026`, :issue:`55106`, :issue:`55299`)
 -
 
 .. ---------------------------------------------------------------------------

pandas/core/window/rolling.py

@@ -405,6 +405,15 @@ def _index_array(self):
         # TODO: why do we get here with e.g. MultiIndex?
         if needs_i8_conversion(self._on.dtype):
             idx = cast("PeriodIndex | DatetimeIndex | TimedeltaIndex", self._on)
+            if (type(idx) == DatetimeIndex) and (idx.T.tz is not None):
+                if idx.T.unit == "s":
+                    return idx.asi8 * int(1e9)
+                elif idx.T.unit == "ms":
+                    return idx.asi8 * int(1e6)
+                elif idx.T.unit == "us":
+                    return idx.asi8 * int(1e3)
+                elif idx.T.unit == "ns":
+                    return idx.asi8
             return idx.asi8
         return None
 
@@ -1889,6 +1898,13 @@ def _validate(self):
             else:
                 self._win_freq_i8 = freq.nanos
 
+            if self._on.dtype == "M8[us]":
+                self._win_freq_i8 /= 1e3
+            elif self._on.dtype == "M8[ms]":
+                self._win_freq_i8 /= 1e6
+            elif self._on.dtype == "M8[s]":
+                self._win_freq_i8 /= 1e9
+
             # min_periods must be an integer
             if self.min_periods is None:
                 self.min_periods = 1

pandas/tests/window/test_rolling.py

@@ -1950,3 +1950,41 @@ def test_numeric_only_corr_cov_series(kernel, use_arg, numeric_only, dtype):
         op2 = getattr(rolling2, kernel)
         expected = op2(*arg2, numeric_only=numeric_only)
         tm.assert_series_equal(result, expected)
+
+
+@pytest.mark.parametrize("unit", ["s", "ms", "us", "ns"])
+@pytest.mark.parametrize("tz", [None, "UTC"])
+def test_rolling_rolling_sum_window_microseconds_conflict_timestamp(unit, tz):
+    # GH#55106
+    df_time = DataFrame(
+        {"A": range(5)}, index=date_range("2013-01-01", freq="1s", periods=5, tz=tz)
+    )
+    sum_in_nanosecs = df_time.rolling("1s").sum()
+    # microseconds / milliseconds should not break the correct rolling
+    df_time.index = df_time.index.as_unit(unit)
+    sum_in_microsecs = df_time.rolling("1s").sum()
+    sum_in_microsecs.index = sum_in_microsecs.index.as_unit("ns")
+    tm.assert_frame_equal(sum_in_nanosecs, sum_in_microsecs)
+
+
+@pytest.mark.parametrize("unit", ["s", "ms", "us", "ns"])
+@pytest.mark.parametrize("tz", [None, "UTC"])
+def test_rolling_rolling_max_window_nanoseconds_conflict_timestamp(unit, tz):
+    # GH#55026
+    window = Timedelta(days=4)
+
+    ref_dates = date_range("2023-01-01", "2023-01-10", unit="ns", tz=tz)
+    ref_series = Series(0, index=ref_dates)
+    ref_series.iloc[0] = 1
+    ref_max_series = ref_series.rolling(window).max()
+
+    dates = date_range("2023-01-01", "2023-01-10", unit=unit, tz=tz)
+    series = Series(0, index=dates)
+    series.iloc[0] = 1
+    max_series = series.rolling(window).max()
+
+    ref_df = DataFrame(ref_max_series)
+    df = DataFrame(max_series)
+    df.index = df.index.as_unit("ns")
+
+    tm.assert_frame_equal(ref_df, df)