BUG: Rolling groupby should not maintain the by column in the resulting DataFrame

doc/source/whatsnew/v1.1.0.rst

@@ -162,6 +162,31 @@ key and type of :class:`Index`.  These now consistently raise ``KeyError`` (:iss
     ...
     KeyError: Timestamp('1970-01-01 00:00:00')
 
+GroupBy.rolling no longer returns grouped-by column in values
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+*Previous behavior*:
+
+.. code-block:: ipython
+
+    In [1]: df = pd.DataFrame({"A": [1, 1, 2, 3], "B": [0, 1, 2, 3]})
+
+    In [2]: df.groupby("A").rolling(2).sum()
+    Out[2]:
+           A    B
+    A
+    1 0  NaN  NaN
+    1    2.0  1.0
+    2 2  NaN  NaN
+    3 3  NaN  NaN
+
+*New behavior*:
+
+.. ipython:: python
+
+    df = pd.DataFrame({"A": [1, 1, 2, 3], "B": [0, 1, 2, 3]})
+    df.groupby("A").rolling(2).sum()
+
 .. ---------------------------------------------------------------------------
 
 .. _whatsnew_110.deprecations:

pandas/core/groupby/groupby.py

@@ -1577,6 +1577,7 @@ def rolling(self, *args, **kwargs):
         """
         from pandas.core.window import RollingGroupby
 
+        kwargs["exclusions"] = self.exclusions
         return RollingGroupby(self, *args, **kwargs)
 
     @Substitution(name="groupby")

pandas/core/window/ewm.py

@@ -129,6 +129,7 @@ def __init__(
         adjust=True,
         ignore_na=False,
         axis=0,
+        **kwargs,
     ):
         self.obj = obj
         self.com = _get_center_of_mass(com, span, halflife, alpha)

pandas/core/window/rolling.py

@@ -93,6 +93,18 @@ def __init__(
         self.axis = obj._get_axis_number(axis) if axis is not None else None
         self.validate()
         self._numba_func_cache: Dict[Optional[str], Callable] = dict()
+        self.exclusions = kwargs.get("exclusions", set())
+
+    def _shallow_copy(self, obj: FrameOrSeries, **kwargs) -> ShallowMixin:
+        if isinstance(obj, ABCDataFrame):
+            # there may be elements in self.exclusions that are no longer
+            # in obj, see GH 32468
+            exclusions = self.exclusions.intersection(obj.columns)
+            new_obj = super()._shallow_copy(obj, exclusions=exclusions, **kwargs)
+            new_obj.obj = new_obj._obj_with_exclusions
+        else:
+            new_obj = super()._shallow_copy(obj, **kwargs)
+        return new_obj
 
     @property
     def _constructor(self):

pandas/tests/window/test_grouper.py

@@ -62,11 +62,15 @@ def test_rolling(self):
         for f in ["sum", "mean", "min", "max", "count", "kurt", "skew"]:
             result = getattr(r, f)()
             expected = g.apply(lambda x: getattr(x.rolling(4), f)())
+            # groupby.apply doesn't drop the grouped-by column
+            expected = expected.drop("A", axis=1)
             tm.assert_frame_equal(result, expected)
 
         for f in ["std", "var"]:
             result = getattr(r, f)(ddof=1)
             expected = g.apply(lambda x: getattr(x.rolling(4), f)(ddof=1))
+            # groupby.apply doesn't drop the grouped-by column
+            expected = expected.drop("A", axis=1)
             tm.assert_frame_equal(result, expected)
 
     @pytest.mark.parametrize(
@@ -79,6 +83,8 @@ def test_rolling_quantile(self, interpolation):
         expected = g.apply(
             lambda x: x.rolling(4).quantile(0.4, interpolation=interpolation)
         )
+        # groupby.apply doesn't drop the grouped-by column
+        expected = expected.drop("A", axis=1)
         tm.assert_frame_equal(result, expected)
 
     def test_rolling_corr_cov(self):
@@ -92,6 +98,8 @@ def func(x):
                 return getattr(x.rolling(4), f)(self.frame)
 
             expected = g.apply(func)
+            # groupby.apply doesn't drop the grouped-by column
+            expected = expected.drop("A", axis=1)
             tm.assert_frame_equal(result, expected)
 
             result = getattr(r.B, f)(pairwise=True)
@@ -109,6 +117,8 @@ def test_rolling_apply(self, raw):
         # reduction
         result = r.apply(lambda x: x.sum(), raw=raw)
         expected = g.apply(lambda x: x.rolling(4).apply(lambda y: y.sum(), raw=raw))
+        # groupby.apply doesn't drop the grouped-by column
+        expected = expected.drop("A", axis=1)
         tm.assert_frame_equal(result, expected)
 
     def test_rolling_apply_mutability(self):

pandas/tests/window/test_rolling.py

@@ -465,3 +465,15 @@ def test_rolling_count_default_min_periods_with_null_values(constructor):
     result = constructor(values).rolling(3).count()
     expected = constructor(expected_counts)
     tm.assert_equal(result, expected)
+
+
+def test_by_column_not_in_values():
+    # GH 32262
+    df = pd.DataFrame({"A": [1] * 20 + [2] * 12 + [3] * 8, "B": np.arange(40)})
+
+    g = df.groupby("A")
+    original_obj = g.obj.copy(deep=True)
+    r = g.rolling(4)
+    result = r.sum()
+    assert "A" not in result.columns
+    tm.assert_frame_equal(g.obj, original_obj)  # check for side-effects