Repository: pandas-dev/pandas (python)

Issue:
ENH: Allow args to be specified for the pivot_table aggfunc
### Feature Type

- [X] Adding new functionality to pandas

- [ ] Changing existing functionality in pandas

- [ ] Removing existing functionality in pandas


### Problem Description

Currently the `pivot_table` has an `aggfunc` parameter which is used to do a groupby aggregation [here](https://github.com/pandas-dev/pandas/blob/main/pandas/core/reshape/pivot.py#L170). However, no additional arguments can be passed into that `agg` call. I'm specifically referring to the `*args` which can be specified in (df/series) [groupby.agg](https://pandas.pydata.org/docs/reference/api/pandas.core.groupby.DataFrameGroupBy.agg.html) function. It would be very useful if `pivot_table` could accept additional arguments for the `aggfunc`.

### Feature Description

I'm not 100% sure, but I think it would be something like this:
1. Add a `**aggfunc_args` or `aggfunc_args: dict` parameter to the `pivot_table` function.
2. Do the same for the `__internal_pivot_table` function
3. Change the line [here](https://github.com/pandas-dev/pandas/blob/main/pandas/core/reshape/pivot.py#L170), from `agged = grouped.agg(aggfunc)` to `agged = grouped.agg(aggfunc, **aggfunc_args)`.
4. Update the docs

### Alternative Solutions

The same functionality can currently be achieved by specifying a custom function as `aggfunc`, but using that is much slower. My use case is pretty much the same as [this](https://stackoverflow.com/questions/56890105/how-to-keep-nan-in-pivot-table).
Instead of using `pd.pivot_table(... , aggfunc=lambda x: x.sum(min_count=1))`, I would like to be able to do `pd.pivot_table(... , aggfunc=sum, min_count=1)` or similiar.

### Additional Context

_No response_

Pull request description:
ENH: Add **kwargs to pivot_table to allow the specification of aggfunc keyword arguments #57884
Add the option of passing keyword arguments to DataFrame.pivot_table and pivot_table's aggfunc through **kwargs.

- [X] closes #57884 
- [X] Tests added and passed after adding a new feature
- [X] All code checks passed
- [X] Added an entry in the latest `doc/source/whatsnew/v3.0.0.rst` file after adding a new feature.

Patch:
diff --git a/doc/source/getting_started/intro_tutorials/09_timeseries.rst b/doc/source/getting_started/intro_tutorials/09_timeseries.rst
index 14db38c382..6ba3c17fac 100644
--- a/doc/source/getting_started/intro_tutorials/09_timeseries.rst
+++ b/doc/source/getting_started/intro_tutorials/09_timeseries.rst
@@ -295,7 +295,7 @@ Aggregate the current hourly time series values to the monthly maximum value in
 
 .. ipython:: python
 
-    monthly_max = no_2.resample("ME").max()
+    monthly_max = no_2.resample("MS").max()
     monthly_max
 
 A very powerful method on time series data with a datetime index, is the
diff --git a/doc/source/user_guide/boolean.rst b/doc/source/user_guide/boolean.rst
index 3c361d4de1..7de0430123 100644
--- a/doc/source/user_guide/boolean.rst
+++ b/doc/source/user_guide/boolean.rst
@@ -37,6 +37,19 @@ If you would prefer to keep the ``NA`` values you can manually fill them with ``
 
    s[mask.fillna(True)]
 
+If you create a column of ``NA`` values (for example to fill them later)
+with ``df['new_col'] = pd.NA``, the ``dtype`` would be set to ``object`` in the
+new column. The performance on this column will be worse than with
+the appropriate type. It's better to use
+``df['new_col'] = pd.Series(pd.NA, dtype="boolean")``
+(or another ``dtype`` that supports ``NA``).
+
+.. ipython:: python
+
+   df = pd.DataFrame()
+   df['objects'] = pd.NA
+   df.dtypes
+
 .. _boolean.kleene:
 
 Kleene logical operations
diff --git a/doc/source/user_guide/integer_na.rst b/doc/source/user_guide/integer_na.rst
index 1a727cd78a..76a2f22b79 100644
--- a/doc/source/user_guide/integer_na.rst
+++ b/doc/source/user_guide/integer_na.rst
@@ -84,6 +84,19 @@ with the dtype.
    In the future, we may provide an option for :class:`Series` to infer a
    nullable-integer dtype.
 
+If you create a column of ``NA`` values (for example to fill them later)
+with ``df['new_col'] = pd.NA``, the ``dtype`` would be set to ``object`` in the
+new column. The performance on this column will be worse than with
+the appropriate type. It's better to use
+``df['new_col'] = pd.Series(pd.NA, dtype="Int64")``
+(or another ``dtype`` that supports ``NA``).
+
+.. ipython:: python
+
+   df = pd.DataFrame()
+   df['objects'] = pd.NA
+   df.dtypes
+
 Operations
 ----------
 
diff --git a/doc/source/user_guide/timeseries.rst b/doc/source/user_guide/timeseries.rst
index ab3f5b314e..d5137baa95 100644
--- a/doc/source/user_guide/timeseries.rst
+++ b/doc/source/user_guide/timeseries.rst
@@ -1864,7 +1864,7 @@ to resample based on datetimelike column in the frame, it can passed to the
        ),
    )
    df
-   df.resample("ME", on="date")[["a"]].sum()
+   df.resample("MS", on="date")[["a"]].sum()
 
 Similarly, if you instead want to resample by a datetimelike
 level of ``MultiIndex``, its name or location can be passed to the
@@ -1872,7 +1872,7 @@ level of ``MultiIndex``, its name or location can be passed to the
 
 .. ipython:: python
 
-   df.resample("ME", level="d")[["a"]].sum()
+   df.resample("MS", level="d")[["a"]].sum()
 
 .. _timeseries.iterating-label:
 
diff --git a/doc/source/whatsnew/v3.0.0.rst b/doc/source/whatsnew/v3.0.0.rst
index 802a9fd7e2..abf1896807 100644
--- a/doc/source/whatsnew/v3.0.0.rst
+++ b/doc/source/whatsnew/v3.0.0.rst
@@ -42,6 +42,7 @@ Other enhancements
 - :meth:`DataFrame.corrwith` now accepts ``min_periods`` as optional arguments, as in :meth:`DataFrame.corr` and :meth:`Series.corr` (:issue:`9490`)
 - :meth:`DataFrame.cummin`, :meth:`DataFrame.cummax`, :meth:`DataFrame.cumprod` and :meth:`DataFrame.cumsum` methods now have a ``numeric_only`` parameter (:issue:`53072`)
 - :meth:`DataFrame.fillna` and :meth:`Series.fillna` can now accept ``value=None``; for non-object dtype the corresponding NA value will be used (:issue:`57723`)
+- :meth:`DataFrame.pivot_table` and :func:`pivot_table` now allow the passing of keyword arguments to ``aggfunc`` through ``**kwargs`` (:issue:`57884`)
 - :meth:`Series.cummin` and :meth:`Series.cummax` now supports :class:`CategoricalDtype` (:issue:`52335`)
 - :meth:`Series.plot` now correctly handle the ``ylabel`` parameter for pie charts, allowing for explicit control over the y-axis label (:issue:`58239`)
 - Restore support for reading Stata 104-format and enable reading 103-format dta files (:issue:`58554`)
diff --git a/pandas/core/frame.py b/pandas/core/frame.py
index c37dfa225d..a6c0e1e372 100644
--- a/pandas/core/frame.py
+++ b/pandas/core/frame.py
@@ -9275,6 +9275,11 @@ class DataFrame(NDFrame, OpsMixin):
 
             .. versionadded:: 1.3.0
 
+        **kwargs : dict
+            Optional keyword arguments to pass to ``aggfunc``.
+
+            .. versionadded:: 3.0.0
+
         Returns
         -------
         DataFrame
@@ -9382,6 +9387,7 @@ class DataFrame(NDFrame, OpsMixin):
         margins_name: Level = "All",
         observed: bool = True,
         sort: bool = True,
+        **kwargs,
     ) -> DataFrame:
         from pandas.core.reshape.pivot import pivot_table
 
@@ -9397,6 +9403,7 @@ class DataFrame(NDFrame, OpsMixin):
             margins_name=margins_name,
             observed=observed,
             sort=sort,
+            **kwargs,
         )
 
     def stack(
diff --git a/pandas/core/internals/managers.py b/pandas/core/internals/managers.py
index 82b88d090f..64109f5c16 100644
--- a/pandas/core/internals/managers.py
+++ b/pandas/core/internals/managers.py
@@ -821,11 +821,13 @@ class BaseBlockManager(PandasObject):
             raise IndexError("Requested axis not found in manager")
 
         if axis == 0:
-            new_blocks = self._slice_take_blocks_ax0(
-                indexer,
-                fill_value=fill_value,
-                only_slice=only_slice,
-                use_na_proxy=use_na_proxy,
+            new_blocks = list(
+                self._slice_take_blocks_ax0(
+                    indexer,
+                    fill_value=fill_value,
+                    only_slice=only_slice,
+                    use_na_proxy=use_na_proxy,
+                )
             )
         else:
             new_blocks = [
@@ -857,7 +859,7 @@ class BaseBlockManager(PandasObject):
         *,
         use_na_proxy: bool = False,
         ref_inplace_op: bool = False,
-    ) -> list[Block]:
+    ) -> Generator[Block, None, None]:
         """
         Slice/take blocks along axis=0.
 
@@ -875,9 +877,9 @@ class BaseBlockManager(PandasObject):
         ref_inplace_op: bool, default False
             Don't track refs if True because we operate inplace
 
-        Returns
-        -------
-        new_blocks : list of Block
+        Yields
+        ------
+        Block : New Block
         """
         allow_fill = fill_value is not lib.no_default
 
@@ -892,9 +894,10 @@ class BaseBlockManager(PandasObject):
                 # GH#32959 EABlock would fail since we can't make 0-width
                 # TODO(EA2D): special casing unnecessary with 2D EAs
                 if sllen == 0:
-                    return []
+                    return
                 bp = BlockPlacement(slice(0, sllen))
-                return [blk.getitem_block_columns(slobj, new_mgr_locs=bp)]
+                yield blk.getitem_block_columns(slobj, new_mgr_locs=bp)
+                return
             elif not allow_fill or self.ndim == 1:
                 if allow_fill and fill_value is None:
                     fill_value = blk.fill_value
@@ -902,25 +905,21 @@ class BaseBlockManager(PandasObject):
                 if not allow_fill and only_slice:
                     # GH#33597 slice instead of take, so we get
                     #  views instead of copies
-                    blocks = [
-                        blk.getitem_block_columns(
+                    for i, ml in enumerate(slobj):
+                        yield blk.getitem_block_columns(
                             slice(ml, ml + 1),
                             new_mgr_locs=BlockPlacement(i),
                             ref_inplace_op=ref_inplace_op,
                         )
-                        for i, ml in enumerate(slobj)
-                    ]
-                    return blocks
                 else:
                     bp = BlockPlacement(slice(0, sllen))
-                    return [
-                        blk.take_nd(
-                            slobj,
-                            axis=0,
-                            new_mgr_locs=bp,
-                            fill_value=fill_value,
-                        )
-                    ]
+                    yield blk.take_nd(
+                        slobj,
+                        axis=0,
+                        new_mgr_locs=bp,
+                        fill_value=fill_value,
+                    )
+                return
 
         if sl_type == "slice":
             blknos = self.blknos[slobj]
@@ -935,18 +934,15 @@ class BaseBlockManager(PandasObject):
 
         # When filling blknos, make sure blknos is updated before appending to
         # blocks list, that way new blkno is exactly len(blocks).
-        blocks = []
         group = not only_slice
         for blkno, mgr_locs in libinternals.get_blkno_placements(blknos, group=group):
             if blkno == -1:
                 # If we've got here, fill_value was not lib.no_default
 
-                blocks.append(
-                    self._make_na_block(
-                        placement=mgr_locs,
-                        fill_value=fill_value,
-                        use_na_proxy=use_na_proxy,
-                    )
+                yield self._make_na_block(
+                    placement=mgr_locs,
+                    fill_value=fill_value,
+                    use_na_proxy=use_na_proxy,
                 )
             else:
                 blk = self.blocks[blkno]
@@ -961,7 +957,7 @@ class BaseBlockManager(PandasObject):
                     for mgr_loc in mgr_locs:
                         newblk = blk.copy(deep=deep)
                         newblk.mgr_locs = BlockPlacement(slice(mgr_loc, mgr_loc + 1))
-                        blocks.append(newblk)
+                        yield newblk
 
                 else:
                     # GH#32779 to avoid the performance penalty of copying,
@@ -972,7 +968,7 @@ class BaseBlockManager(PandasObject):
 
                     if isinstance(taker, slice):
                         nb = blk.getitem_block_columns(taker, new_mgr_locs=mgr_locs)
-                        blocks.append(nb)
+                        yield nb
                     elif only_slice:
                         # GH#33597 slice instead of take, so we get
                         #  views instead of copies
@@ -981,12 +977,10 @@ class BaseBlockManager(PandasObject):
                             bp = BlockPlacement(ml)
                             nb = blk.getitem_block_columns(slc, new_mgr_locs=bp)
                             # We have np.shares_memory(nb.values, blk.values)
-                            blocks.append(nb)
+                            yield nb
                     else:
                         nb = blk.take_nd(taker, axis=0, new_mgr_locs=mgr_locs)
-                        blocks.append(nb)
-
-        return blocks
+                        yield nb
 
     def _make_na_block(
         self, placement: BlockPlacement, fill_value=None, use_na_proxy: bool = False
diff --git a/pandas/core/reshape/pivot.py b/pandas/core/reshape/pivot.py
index ff993c039b..8c2c2053b0 100644
--- a/pandas/core/reshape/pivot.py
+++ b/pandas/core/reshape/pivot.py
@@ -61,6 +61,7 @@ def pivot_table(
     margins_name: Hashable = "All",
     observed: bool = True,
     sort: bool = True,
+    **kwargs,
 ) -> DataFrame:
     """
     Create a spreadsheet-style pivot table as a DataFrame.
@@ -119,6 +120,11 @@ def pivot_table(
 
         .. versionadded:: 1.3.0
 
+    **kwargs : dict
+        Optional keyword arguments to pass to ``aggfunc``.
+
+        .. versionadded:: 3.0.0
+
     Returns
     -------
     DataFrame
@@ -246,6 +252,7 @@ def pivot_table(
                 margins_name=margins_name,
                 observed=observed,
                 sort=sort,
+                kwargs=kwargs,
             )
             pieces.append(_table)
             keys.append(getattr(func, "__name__", func))
@@ -265,6 +272,7 @@ def pivot_table(
         margins_name,
         observed,
         sort,
+        kwargs,
     )
     return table.__finalize__(data, method="pivot_table")
 
@@ -281,6 +289,7 @@ def __internal_pivot_table(
     margins_name: Hashable,
     observed: bool,
     sort: bool,
+    kwargs,
 ) -> DataFrame:
     """
     Helper of :func:`pandas.pivot_table` for any non-list ``aggfunc``.
@@ -323,7 +332,7 @@ def __internal_pivot_table(
         values = list(values)
 
     grouped = data.groupby(keys, observed=observed, sort=sort, dropna=dropna)
-    agged = grouped.agg(aggfunc)
+    agged = grouped.agg(aggfunc, **kwargs)
 
     if dropna and isinstance(agged, ABCDataFrame) and len(agged.columns):
         agged = agged.dropna(how="all")
@@ -378,6 +387,7 @@ def __internal_pivot_table(
             rows=index,
             cols=columns,
             aggfunc=aggfunc,
+            kwargs=kwargs,
             observed=dropna,
             margins_name=margins_name,
             fill_value=fill_value,
@@ -403,6 +413,7 @@ def _add_margins(
     rows,
     cols,
     aggfunc,
+    kwargs,
     observed: bool,
     margins_name: Hashable = "All",
     fill_value=None,
@@ -415,7 +426,7 @@ def _add_margins(
         if margins_name in table.index.get_level_values(level):
             raise ValueError(msg)
 
-    grand_margin = _compute_grand_margin(data, values, aggfunc, margins_name)
+    grand_margin = _compute_grand_margin(data, values, aggfunc, kwargs, margins_name)
 
     if table.ndim == 2:
         # i.e. DataFrame
@@ -436,7 +447,15 @@ def _add_margins(
 
     elif values:
         marginal_result_set = _generate_marginal_results(
-            table, data, values, rows, cols, aggfunc, observed, margins_name
+            table,
+            data,
+            values,
+            rows,
+            cols,
+            aggfunc,
+            kwargs,
+            observed,
+            margins_name,
         )
         if not isinstance(marginal_result_set, tuple):
             return marginal_result_set
@@ -445,7 +464,7 @@ def _add_margins(
         # no values, and table is a DataFrame
         assert isinstance(table, ABCDataFrame)
         marginal_result_set = _generate_marginal_results_without_values(
-            table, data, rows, cols, aggfunc, observed, margins_name
+            table, data, rows, cols, aggfunc, kwargs, observed, margins_name
         )
         if not isinstance(marginal_result_set, tuple):
             return marginal_result_set
@@ -482,26 +501,26 @@ def _add_margins(
 
 
 def _compute_grand_margin(
-    data: DataFrame, values, aggfunc, margins_name: Hashable = "All"
+    data: DataFrame, values, aggfunc, kwargs, margins_name: Hashable = "All"
 ):
     if values:
         grand_margin = {}
         for k, v in data[values].items():
             try:
                 if isinstance(aggfunc, str):
-                    grand_margin[k] = getattr(v, aggfunc)()
+                    grand_margin[k] = getattr(v, aggfunc)(**kwargs)
                 elif isinstance(aggfunc, dict):
                     if isinstance(aggfunc[k], str):
-                        grand_margin[k] = getattr(v, aggfunc[k])()
+                        grand_margin[k] = getattr(v, aggfunc[k])(**kwargs)
                     else:
-                        grand_margin[k] = aggfunc[k](v)
+                        grand_margin[k] = aggfunc[k](v, **kwargs)
                 else:
-                    grand_margin[k] = aggfunc(v)
+                    grand_margin[k] = aggfunc(v, **kwargs)
             except TypeError:
                 pass
         return grand_margin
     else:
-        return {margins_name: aggfunc(data.index)}
+        return {margins_name: aggfunc(data.index, **kwargs)}
 
 
 def _generate_marginal_results(
@@ -511,6 +530,7 @@ def _generate_marginal_results(
     rows,
     cols,
     aggfunc,
+    kwargs,
     observed: bool,
     margins_name: Hashable = "All",
 ):
@@ -524,7 +544,11 @@ def _generate_marginal_results(
             return (key, margins_name) + ("",) * (len(cols) - 1)
 
         if len(rows) > 0:
-            margin = data[rows + values].groupby(rows, observed=observed).agg(aggfunc)
+            margin = (
+                data[rows + values]
+                .groupby(rows, observed=observed)
+                .agg(aggfunc, **kwargs)
+            )
             cat_axis = 1
 
             for key, piece in table.T.groupby(level=0, observed=observed):
@@ -549,7 +573,7 @@ def _generate_marginal_results(
                 table_pieces.append(piece)
                 # GH31016 this is to calculate margin for each group, and assign
                 # corresponded key as index
-                transformed_piece = DataFrame(piece.apply(aggfunc)).T
+                transformed_piece = DataFrame(piece.apply(aggfunc, **kwargs)).T
                 if isinstance(piece.index, MultiIndex):
                     # We are adding an empty level
                     transformed_piece.index = MultiIndex.from_tuples(
@@ -579,7 +603,9 @@ def _generate_marginal_results(
         margin_keys = table.columns
 
     if len(cols) > 0:
-        row_margin = data[cols + values].groupby(cols, observed=observed).agg(aggfunc)
+        row_margin = (
+            data[cols + values].groupby(cols, observed=observed).agg(aggfunc, **kwargs)
+        )
         row_margin = row_margin.stack()
 
         # GH#26568. Use names instead of indices in case of numeric names
@@ -598,6 +624,7 @@ def _generate_marginal_results_without_values(
     rows,
     cols,
     aggfunc,
+    kwargs,
     observed: bool,
     margins_name: Hashable = "All",
 ):
@@ -612,14 +639,16 @@ def _generate_marginal_results_without_values(
             return (margins_name,) + ("",) * (len(cols) - 1)
 
         if len(rows) > 0:
-            margin = data.groupby(rows, observed=observed)[rows].apply(aggfunc)
+            margin = data.groupby(rows, observed=observed)[rows].apply(
+                aggfunc, **kwargs
+            )
             all_key = _all_key()
             table[all_key] = margin
             result = table
             margin_keys.append(all_key)
 
         else:
-            margin = data.groupby(level=0, observed=observed).apply(aggfunc)
+            margin = data.groupby(level=0, observed=observed).apply(aggfunc, **kwargs)
             all_key = _all_key()
             table[all_key] = margin
             result = table
@@ -630,7 +659,9 @@ def _generate_marginal_results_without_values(
         margin_keys = table.columns
 
     if len(cols):
-        row_margin = data.groupby(cols, observed=observed)[cols].apply(aggfunc)
+        row_margin = data.groupby(cols, observed=observed)[cols].apply(
+            aggfunc, **kwargs
+        )
     else:
         row_margin = Series(np.nan, index=result.columns)
 


Test patch:
diff --git a/pandas/tests/reshape/test_pivot.py b/pandas/tests/reshape/test_pivot.py
index 4a13c1f5e1..728becc76b 100644
--- a/pandas/tests/reshape/test_pivot.py
+++ b/pandas/tests/reshape/test_pivot.py
@@ -2058,6 +2058,60 @@ class TestPivotTable:
         ).rename_axis("A")
         tm.assert_frame_equal(result, expected)
 
+    @pytest.mark.parametrize("kwargs", [{"a": 2}, {"a": 2, "b": 3}, {"b": 3, "a": 2}])
+    def test_pivot_table_kwargs(self, kwargs):
+        # GH#57884
+        def f(x, a, b=3):
+            return x.sum() * a + b
+
+        def g(x):
+            return f(x, **kwargs)
+
+        df = DataFrame(
+            {
+                "A": ["good", "bad", "good", "bad", "good"],
+                "B": ["one", "two", "one", "three", "two"],
+                "X": [2, 5, 4, 20, 10],
+            }
+        )
+        result = pivot_table(
+            df, index="A", columns="B", values="X", aggfunc=f, **kwargs
+        )
+        expected = pivot_table(df, index="A", columns="B", values="X", aggfunc=g)
+        tm.assert_frame_equal(result, expected)
+
+    @pytest.mark.parametrize(
+        "kwargs", [{}, {"b": 10}, {"a": 3}, {"a": 3, "b": 10}, {"b": 10, "a": 3}]
+    )
+    def test_pivot_table_kwargs_margin(self, data, kwargs):
+        # GH#57884
+        def f(x, a=5, b=7):
+            return (x.sum() + b) * a
+
+        def g(x):
+            return f(x, **kwargs)
+
+        result = data.pivot_table(
+            values="D",
+            index=["A", "B"],
+            columns="C",
+            aggfunc=f,
+            margins=True,
+            fill_value=0,
+            **kwargs,
+        )
+
+        expected = data.pivot_table(
+            values="D",
+            index=["A", "B"],
+            columns="C",
+            aggfunc=g,
+            margins=True,
+            fill_value=0,
+        )
+
+        tm.assert_frame_equal(result, expected)
+
     @pytest.mark.parametrize(
         "f, f_numpy",
         [