[BUG]: Rolling selected too large windows with PeriodIndex

doc/source/whatsnew/v1.2.0.rst

@@ -407,6 +407,7 @@ Groupby/resample/rolling
 - Bug in :meth:`DataFrameGroupBy.apply` raising error with ``np.nan`` group(s) when ``dropna=False`` (:issue:`35889`)
 - Bug in :meth:`Rolling.sum()` returned wrong values when dtypes where mixed between float and integer and axis was equal to one (:issue:`20649`, :issue:`35596`)
 - Bug in :meth:`Rolling.count` returned ``np.nan`` with :class:`pandas.api.indexers.FixedForwardWindowIndexer` as window, ``min_periods=0`` and only missing values in window (:issue:`35579`)
+- Bug where :class:`pandas.core.window.Rolling` produces incorrect window sizes when using a ``PeriodIndex`` (:issue:`34225`)
 
 Reshaping
 ^^^^^^^^^

pandas/core/window/rolling.py

@@ -1932,7 +1932,6 @@ def validate(self):
         ):
 
             self._validate_monotonic()
-            freq = self._validate_freq()
 
             # we don't allow center
             if self.center:
@@ -1943,7 +1942,7 @@ def validate(self):
 
             # this will raise ValueError on non-fixed freqs
             self.win_freq = self.window
-            self.window = freq.nanos
+            self.window = self._determine_window_length()
             self.win_type = "freq"
 
             # min_periods must be an integer
@@ -1963,6 +1962,16 @@ def validate(self):
                 "closed only implemented for datetimelike and offset based windows"
             )
 
+    def _determine_window_length(self) -> Union[int, float]:
+        """
+        Calculate freq for PeriodIndexes based on Index freq. Can not use
+        nanos, because asi8 of PeriodIndex is not in nanos
+        """
+        freq = self._validate_freq()
+        if isinstance(self._on, ABCPeriodIndex):
+            return freq.nanos / (self._on.freq.nanos / self._on.freq.n)
+        return freq.nanos
+
     def _validate_monotonic(self):
         """
         Validate monotonic (increasing or decreasing).

pandas/tests/window/test_rolling.py

@@ -837,3 +837,34 @@ def test_rolling_on_df_transposed():
 
     result = df.T.rolling(min_periods=1, window=2).sum().T
     tm.assert_frame_equal(result, expected)
+
+
+@pytest.mark.parametrize(
+    ("index", "window"),
+    [
+        (
+            pd.period_range(start="2020-01-01 08:00", end="2020-01-01 08:08", freq="T"),
+            "2T",
+        ),
+        (
+            pd.period_range(
+                start="2020-01-01 08:00", end="2020-01-01 12:00", freq="30T"
+            ),
+            "1h",
+        ),
+    ],
+)
+@pytest.mark.parametrize(
+    ("func", "values"),
+    [
+        ("min", [np.nan, 0, 0, 1, 2, 3, 4, 5, 6]),
+        ("max", [np.nan, 0, 1, 2, 3, 4, 5, 6, 7]),
+        ("sum", [np.nan, 0, 1, 3, 5, 7, 9, 11, 13]),
+    ],
+)
+def test_rolling_period_index(index, window, func, values):
+    # GH: 34225
+    ds = pd.Series([0, 1, 2, 3, 4, 5, 6, 7, 8], index=index)
+    result = getattr(ds.rolling(window, closed="left"), func)()
+    expected = pd.Series(values, index=index)
+    tm.assert_series_equal(result, expected)