REF: Move Resampler/Window.agg into core.apply (#39459)

Author: rhshadrach

Diff for: pandas/core/aggregation.py

@@ -534,56 +534,6 @@ def transform_str_or_callable(
         return func(obj, *args, **kwargs)
 
 
-def aggregate(
-    obj: AggObjType,
-    arg: AggFuncType,
-    *args,
-    **kwargs,
-):
-    """
-    Provide an implementation for the aggregators.
-
-    Parameters
-    ----------
-    obj : Pandas object to compute aggregation on.
-    arg : string, dict, function.
-    *args : args to pass on to the function.
-    **kwargs : kwargs to pass on to the function.
-
-    Returns
-    -------
-    tuple of result, how.
-
-    Notes
-    -----
-    how can be a string describe the required post-processing, or
-    None if not required.
-    """
-    _axis = kwargs.pop("_axis", None)
-    if _axis is None:
-        _axis = getattr(obj, "axis", 0)
-
-    if isinstance(arg, str):
-        return obj._try_aggregate_string_function(arg, *args, **kwargs), None
-    elif is_dict_like(arg):
-        arg = cast(AggFuncTypeDict, arg)
-        return agg_dict_like(obj, arg, _axis), True
-    elif is_list_like(arg):
-        # we require a list, but not an 'str'
-        arg = cast(List[AggFuncTypeBase], arg)
-        return agg_list_like(obj, arg, _axis=_axis), None
-    else:
-        result = None
-
-    if callable(arg):
-        f = obj._get_cython_func(arg)
-        if f and not args and not kwargs:
-            return getattr(obj, f)(), None
-
-    # caller can react
-    return result, True
-
-
 def agg_list_like(
     obj: AggObjType,
     arg: List[AggFuncTypeBase],

Diff for: pandas/core/apply.py

@@ -47,6 +47,8 @@
 if TYPE_CHECKING:
     from pandas import DataFrame, Index, Series
     from pandas.core.groupby import DataFrameGroupBy, SeriesGroupBy
+    from pandas.core.resample import Resampler
+    from pandas.core.window.rolling import BaseWindow
 
 ResType = Dict[int, Any]
 
@@ -684,3 +686,27 @@ def __init__(
 
     def apply(self):
         raise NotImplementedError
+
+
+class ResamplerWindowApply(Apply):
+    axis = 0
+    obj: Union[Resampler, BaseWindow]
+
+    def __init__(
+        self,
+        obj: Union[Resampler, BaseWindow],
+        func: AggFuncType,
+        args,
+        kwds,
+    ):
+        super().__init__(
+            obj,
+            func,
+            raw=False,
+            result_type=None,
+            args=args,
+            kwds=kwds,
+        )
+
+    def apply(self):
+        raise NotImplementedError

Diff for: pandas/core/resample.py

@@ -23,8 +23,8 @@
 
 from pandas.core.dtypes.generic import ABCDataFrame, ABCSeries
 
-from pandas.core.aggregation import aggregate
 import pandas.core.algorithms as algos
+from pandas.core.apply import ResamplerWindowApply
 from pandas.core.base import DataError
 from pandas.core.generic import NDFrame, _shared_docs
 from pandas.core.groupby.base import GotItemMixin, ShallowMixin
@@ -301,7 +301,7 @@ def pipe(
     def aggregate(self, func, *args, **kwargs):
 
         self._set_binner()
-        result, how = aggregate(self, func, *args, **kwargs)
+        result, how = ResamplerWindowApply(self, func, args=args, kwds=kwargs).agg()
         if result is None:
             how = func
             grouper = None

Diff for: pandas/core/window/rolling.py

@@ -48,7 +48,7 @@
 )
 from pandas.core.dtypes.missing import notna
 
-from pandas.core.aggregation import aggregate
+from pandas.core.apply import ResamplerWindowApply
 from pandas.core.base import DataError, SelectionMixin
 from pandas.core.construction import extract_array
 from pandas.core.groupby.base import GotItemMixin, ShallowMixin
@@ -479,7 +479,7 @@ def calc(x):
             return self._apply_tablewise(homogeneous_func, name)
 
     def aggregate(self, func, *args, **kwargs):
-        result, how = aggregate(self, func, *args, **kwargs)
+        result, how = ResamplerWindowApply(self, func, args=args, kwds=kwargs).agg()
         if result is None:
             return self.apply(func, raw=False, args=args, kwargs=kwargs)
         return result
@@ -1150,7 +1150,7 @@ def calc(x):
         axis="",
     )
     def aggregate(self, func, *args, **kwargs):
-        result, how = aggregate(self, func, *args, **kwargs)
+        result, how = ResamplerWindowApply(self, func, args=args, kwds=kwargs).agg()
         if result is None:
 
             # these must apply directly