Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pandas/core/apply.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import abc
4from collections import defaultdict
5from collections.abc import Callable
6import functools
7from functools import partial
8import inspect
9from typing import (
10 TYPE_CHECKING,
11 Any,
12 Literal,
13 TypeAlias,
14 cast,
15)
17import numpy as np
19from pandas._libs.internals import BlockValuesRefs
20from pandas._typing import (
21 AggFuncType,
22 AggFuncTypeBase,
23 AggFuncTypeDict,
24 AggObjType,
25 Axis,
26 AxisInt,
27 NDFrameT,
28 npt,
29)
30from pandas.compat._optional import import_optional_dependency
31from pandas.errors import SpecificationError
32from pandas.util._decorators import (
33 cache_readonly,
34 set_module,
35)
37from pandas.core.dtypes.cast import is_nested_object
38from pandas.core.dtypes.common import (
39 is_dict_like,
40 is_extension_array_dtype,
41 is_list_like,
42 is_numeric_dtype,
43 is_sequence,
44)
45from pandas.core.dtypes.dtypes import ExtensionDtype
46from pandas.core.dtypes.generic import (
47 ABCDataFrame,
48 ABCNDFrame,
49 ABCSeries,
50)
52from pandas.core._numba.executor import generate_apply_looper
53import pandas.core.common as com
54from pandas.core.construction import ensure_wrapped_if_datetimelike
55from pandas.core.util.numba_ import (
56 get_jit_arguments,
57 prepare_function_arguments,
58)
60if TYPE_CHECKING:
61 from collections.abc import (
62 Generator,
63 Hashable,
64 Iterable,
65 MutableMapping,
66 Sequence,
67 )
69 from pandas import (
70 DataFrame,
71 Index,
72 Series,
73 )
74 from pandas.core.groupby import GroupBy
75 from pandas.core.resample import Resampler
76 from pandas.core.window.rolling import BaseWindow
78ResType: TypeAlias = dict[int, Any]
81@set_module("pandas.api.executors")
82class BaseExecutionEngine(abc.ABC):
83 """
84 Base class for execution engines for map and apply methods.
86 An execution engine receives all the parameters of a call to
87 ``apply`` or ``map``, such as the data container, the function,
88 etc. and takes care of running the execution.
90 Supporting different engines allows functions to be JIT compiled,
91 run in parallel, and others. Besides the default executor which
92 simply runs the code with the Python interpreter and pandas.
93 """
95 @staticmethod
96 @abc.abstractmethod
97 def map(
98 data: Series | DataFrame | np.ndarray,
99 func: AggFuncType,
100 args: tuple,
101 kwargs: dict[str, Any],
102 decorator: Callable | None,
103 skip_na: bool,
104 ):
105 """
106 Executor method to run functions elementwise.
108 In general, pandas uses ``map`` for running functions elementwise,
109 but ``Series.apply`` with the default ``by_row='compat'`` will also
110 call this executor function.
112 Parameters
113 ----------
114 data : Series, DataFrame or NumPy ndarray
115 The object to use for the data. Some methods implement a ``raw``
116 parameter which will convert the original pandas object to a
117 NumPy array, which will then be passed here to the executor.
118 func : function or NumPy ufunc
119 The function to execute.
120 args : tuple
121 Positional arguments to be passed to ``func``.
122 kwargs : dict
123 Keyword arguments to be passed to ``func``.
124 decorator : function, optional
125 For JIT compilers and other engines that need to decorate the
126 function ``func``, this is the decorator to use. While the
127 executor may already know which is the decorator to use, this
128 is useful as for a single executor the user can specify for
129 example ``numba.jit`` or ``numba.njit(nogil=True)``, and this
130 decorator parameter will contain the exact decorator from the
131 executor the user wants to use.
132 skip_na : bool
133 Whether the function should be called for missing values or not.
134 This is specified by the pandas user as ``map(na_action=None)``
135 or ``map(na_action='ignore')``.
136 """
138 @staticmethod
139 @abc.abstractmethod
140 def apply(
141 data: Series | DataFrame | np.ndarray,
142 func: AggFuncType,
143 args: tuple,
144 kwargs: dict[str, Any],
145 decorator: Callable,
146 axis: Axis,
147 ):
148 """
149 Executor method to run functions by an axis.
151 While we can see ``map`` as executing the function for each cell
152 in a ``DataFrame`` (or ``Series``), ``apply`` will execute the
153 function for each column (or row).
155 Parameters
156 ----------
157 data : Series, DataFrame or NumPy ndarray
158 The object to use for the data. Some methods implement a ``raw``
159 parameter which will convert the original pandas object to a
160 NumPy array, which will then be passed here to the executor.
161 func : function or NumPy ufunc
162 The function to execute.
163 args : tuple
164 Positional arguments to be passed to ``func``.
165 kwargs : dict
166 Keyword arguments to be passed to ``func``.
167 decorator : function, optional
168 For JIT compilers and other engines that need to decorate the
169 function ``func``, this is the decorator to use. While the
170 executor may already know which is the decorator to use, this
171 is useful as for a single executor the user can specify for
172 example ``numba.jit`` or ``numba.njit(nogil=True)``, and this
173 decorator parameter will contain the exact decorator from the
174 executor the user wants to use.
175 axis : {0 or 'index', 1 or 'columns'}
176 0 or 'index' should execute the function passing each column as
177 parameter. 1 or 'columns' should execute the function passing
178 each row as parameter. The default executor engine passes rows
179 as pandas ``Series``. Other executor engines should probably
180 expect functions to be implemented this way for compatibility.
181 But passing rows as other data structures is technically possible
182 as far as the function ``func`` is implemented accordingly.
183 """
186def frame_apply(
187 obj: DataFrame,
188 func: AggFuncType,
189 axis: Axis = 0,
190 raw: bool = False,
191 result_type: str | None = None,
192 by_row: Literal[False, "compat"] = "compat",
193 engine: str = "python",
194 engine_kwargs: dict[str, bool] | None = None,
195 args=None,
196 kwargs=None,
197) -> FrameApply:
198 """construct and return a row or column based frame apply object"""
199 _, func, columns, _ = reconstruct_func(func, **kwargs)
201 axis = obj._get_axis_number(axis)
202 klass: type[FrameApply]
203 if axis == 0:
204 klass = FrameRowApply
205 elif axis == 1:
206 if columns:
207 raise NotImplementedError(
208 f"Named aggregation is not supported when {axis=}."
209 )
210 klass = FrameColumnApply
212 return klass(
213 obj,
214 func,
215 raw=raw,
216 result_type=result_type,
217 by_row=by_row,
218 engine=engine,
219 engine_kwargs=engine_kwargs,
220 args=args,
221 kwargs=kwargs,
222 )
225class Apply(metaclass=abc.ABCMeta):
226 axis: AxisInt
228 def __init__(
229 self,
230 obj: AggObjType,
231 func: AggFuncType,
232 raw: bool,
233 result_type: str | None,
234 *,
235 by_row: Literal[False, "compat", "_compat"] = "compat",
236 engine: str = "python",
237 engine_kwargs: dict[str, bool] | None = None,
238 args,
239 kwargs,
240 ) -> None:
241 self.obj = obj
242 self.raw = raw
244 assert by_row is False or by_row in ["compat", "_compat"]
245 self.by_row = by_row
247 self.args = args or ()
248 self.kwargs = kwargs or {}
250 self.engine = engine
251 self.engine_kwargs = {} if engine_kwargs is None else engine_kwargs
253 if result_type not in [None, "reduce", "broadcast", "expand"]:
254 raise ValueError(
255 "invalid value for result_type, must be one "
256 "of {None, 'reduce', 'broadcast', 'expand'}"
257 )
259 self.result_type = result_type
261 self.func = func
263 @abc.abstractmethod
264 def apply(self) -> DataFrame | Series:
265 pass
267 @abc.abstractmethod
268 def agg_or_apply_list_like(
269 self, op_name: Literal["agg", "apply"]
270 ) -> DataFrame | Series:
271 pass
273 @abc.abstractmethod
274 def agg_or_apply_dict_like(
275 self, op_name: Literal["agg", "apply"]
276 ) -> DataFrame | Series:
277 pass
279 def agg(self) -> DataFrame | Series | None:
280 """
281 Provide an implementation for the aggregators.
283 Returns
284 -------
285 Result of aggregation, or None if agg cannot be performed by
286 this method.
287 """
288 func = self.func
290 if isinstance(func, str):
291 return self.apply_str()
293 if is_dict_like(func):
294 return self.agg_dict_like()
295 elif is_list_like(func):
296 # we require a list, but not a 'str'
297 return self.agg_list_like()
299 # caller can react
300 return None
302 def transform(self) -> DataFrame | Series:
303 """
304 Transform a DataFrame or Series.
306 Returns
307 -------
308 DataFrame or Series
309 Result of applying ``func`` along the given axis of the
310 Series or DataFrame.
312 Raises
313 ------
314 ValueError
315 If the transform function fails or does not transform.
316 """
317 obj = self.obj
318 func = self.func
319 axis = self.axis
320 args = self.args
321 kwargs = self.kwargs
323 is_series = obj.ndim == 1
325 if obj._get_axis_number(axis) == 1:
326 assert not is_series
327 return obj.T.transform(func, 0, *args, **kwargs).T
329 if is_list_like(func) and not is_dict_like(func):
330 func = cast(list[AggFuncTypeBase], func)
331 # Convert func equivalent dict
332 if is_series:
333 func = {com.get_callable_name(v) or v: v for v in func}
334 else:
335 func = dict.fromkeys(obj, func)
337 if is_dict_like(func):
338 func = cast(AggFuncTypeDict, func)
339 return self.transform_dict_like(func)
341 # func is either str or callable
342 func = cast(AggFuncTypeBase, func)
343 try:
344 result = self.transform_str_or_callable(func)
345 except TypeError:
346 raise
347 except Exception as err:
348 raise ValueError("Transform function failed") from err
350 # Functions that transform may return empty Series/DataFrame
351 # when the dtype is not appropriate
352 if (
353 isinstance(result, (ABCSeries, ABCDataFrame))
354 and result.empty
355 and not obj.empty
356 ):
357 raise ValueError("Transform function failed")
358 if not isinstance(result, (ABCSeries, ABCDataFrame)) or not result.index.equals(
359 obj.index
360 ):
361 raise ValueError("Function did not transform")
363 return result
365 def transform_dict_like(self, func) -> DataFrame:
366 """
367 Compute transform in the case of a dict-like func
368 """
369 from pandas.core.reshape.concat import concat
371 obj = self.obj
372 args = self.args
373 kwargs = self.kwargs
375 # transform is currently only for Series/DataFrame
376 assert isinstance(obj, ABCNDFrame)
378 if len(func) == 0:
379 raise ValueError("No transform functions were provided")
381 func = self.normalize_dictlike_arg("transform", obj, func)
383 results: dict[Hashable, DataFrame | Series] = {}
384 for name, how in func.items():
385 colg = obj._gotitem(name, ndim=1)
386 results[name] = colg.transform(how, 0, *args, **kwargs)
387 return concat(results, axis=1)
389 def transform_str_or_callable(self, func) -> DataFrame | Series:
390 """
391 Compute transform in the case of a string or callable func
392 """
393 obj = self.obj
394 args = self.args
395 kwargs = self.kwargs
397 if isinstance(func, str):
398 return self._apply_str(obj, func, *args, **kwargs)
400 # Two possible ways to use a UDF - apply or call directly
401 try:
402 return obj.apply(func, args=args, **kwargs)
403 except Exception:
404 return func(obj, *args, **kwargs)
406 def agg_list_like(self) -> DataFrame | Series:
407 """
408 Compute aggregation in the case of a list-like argument.
410 Returns
411 -------
412 Result of aggregation.
413 """
414 return self.agg_or_apply_list_like(op_name="agg")
416 def compute_list_like(
417 self,
418 op_name: Literal["agg", "apply"],
419 selected_obj: Series | DataFrame,
420 kwargs: dict[str, Any],
421 ) -> tuple[list[Hashable] | Index, list[Any]]:
422 """
423 Compute agg/apply results for like-like input.
425 Parameters
426 ----------
427 op_name : {"agg", "apply"}
428 Operation being performed.
429 selected_obj : Series or DataFrame
430 Data to perform operation on.
431 kwargs : dict
432 Keyword arguments to pass to the functions.
434 Returns
435 -------
436 keys : list[Hashable] or Index
437 Index labels for result.
438 results : list
439 Data for result. When aggregating with a Series, this can contain any
440 Python objects.
441 """
442 func = cast(list[AggFuncTypeBase], self.func)
443 obj = self.obj
445 results = []
446 keys = []
448 # degenerate case
449 if selected_obj.ndim == 1:
450 for a in func:
451 colg = obj._gotitem(selected_obj.name, ndim=1, subset=selected_obj)
452 args = (
453 [self.axis, *self.args]
454 if include_axis(op_name, colg)
455 else self.args
456 )
457 new_res = getattr(colg, op_name)(a, *args, **kwargs)
458 results.append(new_res)
460 # make sure we find a good name
461 name = com.get_callable_name(a) or a
462 keys.append(name)
464 else:
465 indices = []
466 for index, col in enumerate(selected_obj):
467 colg = obj._gotitem(col, ndim=1, subset=selected_obj.iloc[:, index])
468 args = (
469 [self.axis, *self.args]
470 if include_axis(op_name, colg)
471 else self.args
472 )
473 new_res = getattr(colg, op_name)(func, *args, **kwargs)
474 results.append(new_res)
475 indices.append(index)
476 # error: Incompatible types in assignment (expression has type "Any |
477 # Index", variable has type "list[Any | Callable[..., Any] | str]")
478 keys = selected_obj.columns.take(indices) # type: ignore[assignment]
480 return keys, results
482 def wrap_results_list_like(
483 self, keys: Iterable[Hashable], results: list[Series | DataFrame]
484 ):
485 from pandas.core.reshape.concat import concat
487 obj = self.obj
489 try:
490 return concat(results, keys=keys, axis=1, sort=False)
491 except TypeError as err:
492 # we are concatting non-NDFrame objects,
493 # e.g. a list of scalars
494 from pandas import Series
496 result = Series(results, index=keys, name=obj.name)
497 if is_nested_object(result):
498 raise ValueError(
499 "cannot combine transform and aggregation operations"
500 ) from err
501 return result
503 def agg_dict_like(self) -> DataFrame | Series:
504 """
505 Compute aggregation in the case of a dict-like argument.
507 Returns
508 -------
509 Result of aggregation.
510 """
511 return self.agg_or_apply_dict_like(op_name="agg")
513 def compute_dict_like(
514 self,
515 op_name: Literal["agg", "apply"],
516 selected_obj: Series | DataFrame,
517 selection: Hashable | Sequence[Hashable],
518 kwargs: dict[str, Any],
519 ) -> tuple[list[Hashable], list[Any]]:
520 """
521 Compute agg/apply results for dict-like input.
523 Parameters
524 ----------
525 op_name : {"agg", "apply"}
526 Operation being performed.
527 selected_obj : Series or DataFrame
528 Data to perform operation on.
529 selection : hashable or sequence of hashables
530 Used by GroupBy, Window, and Resample if selection is applied to the object.
531 kwargs : dict
532 Keyword arguments to pass to the functions.
534 Returns
535 -------
536 keys : list[hashable]
537 Index labels for result.
538 results : list
539 Data for result. When aggregating with a Series, this can contain any
540 Python object.
541 """
542 from pandas.core.groupby.generic import (
543 DataFrameGroupBy,
544 SeriesGroupBy,
545 )
547 obj = self.obj
548 is_groupby = isinstance(obj, (DataFrameGroupBy, SeriesGroupBy))
549 func = cast(AggFuncTypeDict, self.func)
550 func = self.normalize_dictlike_arg(op_name, selected_obj, func)
552 is_non_unique_col = (
553 selected_obj.ndim == 2
554 and selected_obj.columns.nunique() < len(selected_obj.columns)
555 )
557 if selected_obj.ndim == 1:
558 # key only used for output
559 colg = obj._gotitem(selection, ndim=1)
560 results = [getattr(colg, op_name)(how, **kwargs) for _, how in func.items()]
561 keys = list(func.keys())
562 elif not is_groupby and is_non_unique_col:
563 # key used for column selection and output
564 # GH#51099
565 results = []
566 keys = []
567 for key, how in func.items():
568 indices = selected_obj.columns.get_indexer_for([key])
569 labels = selected_obj.columns.take(indices)
570 label_to_indices = defaultdict(list)
571 for index, label in zip(indices, labels, strict=True):
572 label_to_indices[label].append(index)
574 key_data = [
575 getattr(selected_obj._ixs(indice, axis=1), op_name)(how, **kwargs)
576 for label, indices in label_to_indices.items()
577 for indice in indices
578 ]
580 keys += [key] * len(key_data)
581 results += key_data
582 elif is_groupby:
583 # key used for column selection and output
585 df = selected_obj
586 results, keys = [], []
587 for key, how in func.items():
588 cols = df[key]
590 if cols.ndim == 1:
591 series = obj._gotitem(key, ndim=1, subset=cols)
592 results.append(getattr(series, op_name)(how, **kwargs))
593 keys.append(key)
594 else:
595 for _, col in cols.items():
596 series = obj._gotitem(key, ndim=1, subset=col)
597 results.append(getattr(series, op_name)(how, **kwargs))
598 keys.append(key)
599 else:
600 results = [
601 getattr(obj._gotitem(key, ndim=1), op_name)(how, **kwargs)
602 for key, how in func.items()
603 ]
604 keys = list(func.keys())
606 return keys, results
608 def wrap_results_dict_like(
609 self,
610 selected_obj: Series | DataFrame,
611 result_index: list[Hashable],
612 result_data: list,
613 ):
614 from pandas import Index
615 from pandas.core.reshape.concat import concat
617 obj = self.obj
619 # Avoid making two isinstance calls in all and any below
620 is_ndframe = [isinstance(r, ABCNDFrame) for r in result_data]
622 if all(is_ndframe):
623 results = [result for result in result_data if not result.empty]
624 keys_to_use: Iterable[Hashable]
625 keys_to_use = [
626 k for k, v in zip(result_index, result_data, strict=True) if not v.empty
627 ]
628 # Have to check, if at least one DataFrame is not empty.
629 if keys_to_use == []:
630 keys_to_use = result_index
631 results = result_data
633 if selected_obj.ndim == 2:
634 # keys are columns, so we can preserve names
635 ktu = Index(keys_to_use)
636 ktu._set_names(selected_obj.columns.names)
637 keys_to_use = ktu
639 axis: AxisInt = 0 if isinstance(obj, ABCSeries) else 1
640 result = concat(
641 results,
642 axis=axis,
643 keys=keys_to_use,
644 sort=False,
645 )
646 elif any(is_ndframe):
647 # There is a mix of NDFrames and scalars
648 raise ValueError(
649 "cannot perform both aggregation "
650 "and transformation operations "
651 "simultaneously"
652 )
653 else:
654 from pandas import Series
656 # we have a list of scalars
657 # GH 36212 use name only if obj is a series
658 if obj.ndim == 1:
659 obj = cast("Series", obj)
660 name = obj.name
661 else:
662 name = None
664 result = Series(result_data, index=result_index, name=name)
666 return result
668 def apply_str(self) -> DataFrame | Series:
669 """
670 Compute apply in case of a string.
672 Returns
673 -------
674 result: Series or DataFrame
675 """
676 # Caller is responsible for checking isinstance(self.f, str)
677 func = cast(str, self.func)
679 obj = self.obj
681 from pandas.core.groupby.generic import (
682 DataFrameGroupBy,
683 SeriesGroupBy,
684 )
686 # Support for `frame.transform('method')`
687 # Some methods (shift, etc.) require the axis argument, others
688 # don't, so inspect and insert if necessary.
689 method = getattr(obj, func, None)
690 if callable(method):
691 sig = inspect.getfullargspec(method)
692 arg_names = (*sig.args, *sig.kwonlyargs)
693 if self.axis != 0 and (
694 "axis" not in arg_names or func in ("corrwith", "skew")
695 ):
696 raise ValueError(f"Operation {func} does not support axis=1")
697 if "axis" in arg_names and not isinstance(
698 obj, (SeriesGroupBy, DataFrameGroupBy)
699 ):
700 self.kwargs["axis"] = self.axis
701 return self._apply_str(obj, func, *self.args, **self.kwargs)
703 def apply_list_or_dict_like(self) -> DataFrame | Series:
704 """
705 Compute apply in case of a list-like or dict-like.
707 Returns
708 -------
709 result: Series, DataFrame, or None
710 Result when self.func is a list-like or dict-like, None otherwise.
711 """
713 if self.engine == "numba":
714 raise NotImplementedError(
715 "The 'numba' engine doesn't support list-like/"
716 "dict likes of callables yet."
717 )
719 if self.axis == 1 and isinstance(self.obj, ABCDataFrame):
720 return self.obj.T.apply(self.func, 0, args=self.args, **self.kwargs).T
722 func = self.func
723 kwargs = self.kwargs
725 if is_dict_like(func):
726 result = self.agg_or_apply_dict_like(op_name="apply")
727 else:
728 result = self.agg_or_apply_list_like(op_name="apply")
730 result = reconstruct_and_relabel_result(result, func, **kwargs)
732 return result
734 def normalize_dictlike_arg(
735 self, how: str, obj: DataFrame | Series, func: AggFuncTypeDict
736 ) -> AggFuncTypeDict:
737 """
738 Handler for dict-like argument.
740 Ensures that necessary columns exist if obj is a DataFrame, and
741 that a nested renamer is not passed. Also normalizes to all lists
742 when values consists of a mix of list and non-lists.
743 """
744 assert how in ("apply", "agg", "transform")
746 # Can't use func.values(); wouldn't work for a Series
747 if (
748 how == "agg"
749 and isinstance(obj, ABCSeries)
750 and any(is_list_like(v) for _, v in func.items())
751 ) or (any(is_dict_like(v) for _, v in func.items())):
752 # GH 15931 - deprecation of renaming keys
753 raise SpecificationError("nested renamer is not supported")
755 if obj.ndim != 1:
756 # Check for missing columns on a frame
757 from pandas import Index
759 cols = Index(list(func.keys())).difference(obj.columns, sort=True)
760 if len(cols) > 0:
761 # GH 58474
762 raise KeyError(f"Label(s) {list(cols)} do not exist")
764 aggregator_types = (list, tuple, dict)
766 # if we have a dict of any non-scalars
767 # eg. {'A' : ['mean']}, normalize all to
768 # be list-likes
769 # Cannot use func.values() because arg may be a Series
770 if any(isinstance(x, aggregator_types) for _, x in func.items()):
771 new_func: AggFuncTypeDict = {}
772 for k, v in func.items():
773 if not isinstance(v, aggregator_types):
774 new_func[k] = [v]
775 else:
776 new_func[k] = v
777 func = new_func
778 return func
780 def _apply_str(self, obj, func: str, *args, **kwargs):
781 """
782 if arg is a string, then try to operate on it:
783 - try to find a function (or attribute) on obj
784 - try to find a numpy function
785 - raise
786 """
787 assert isinstance(func, str)
789 if hasattr(obj, func):
790 f = getattr(obj, func)
791 if callable(f):
792 return f(*args, **kwargs)
794 # people may aggregate on a non-callable attribute
795 # but don't let them think they can pass args to it
796 assert len(args) == 0
797 assert not any(kwarg == "axis" for kwarg in kwargs)
798 return f
799 elif hasattr(np, func) and hasattr(obj, "__array__"):
800 # in particular exclude Window
801 f = getattr(np, func)
802 return f(obj, *args, **kwargs)
803 else:
804 msg = f"'{func}' is not a valid function for '{type(obj).__name__}' object"
805 raise AttributeError(msg)
808class NDFrameApply(Apply):
809 """
810 Methods shared by FrameApply and SeriesApply but
811 not GroupByApply or ResamplerWindowApply
812 """
814 obj: DataFrame | Series
816 @property
817 def index(self) -> Index:
818 return self.obj.index
820 @property
821 def agg_axis(self) -> Index:
822 return self.obj._get_agg_axis(self.axis)
824 def agg_or_apply_list_like(
825 self, op_name: Literal["agg", "apply"]
826 ) -> DataFrame | Series:
827 obj = self.obj
828 kwargs = self.kwargs
830 if op_name == "apply":
831 if isinstance(self, FrameApply):
832 by_row = self.by_row
834 elif isinstance(self, SeriesApply):
835 by_row = "_compat" if self.by_row else False
836 else:
837 by_row = False
838 kwargs = {**kwargs, "by_row": by_row}
840 if getattr(obj, "axis", 0) == 1:
841 raise NotImplementedError("axis other than 0 is not supported")
843 keys, results = self.compute_list_like(op_name, obj, kwargs)
844 result = self.wrap_results_list_like(keys, results)
845 return result
847 def agg_or_apply_dict_like(
848 self, op_name: Literal["agg", "apply"]
849 ) -> DataFrame | Series:
850 assert op_name in ["agg", "apply"]
851 obj = self.obj
853 kwargs = {}
854 if op_name == "apply":
855 by_row = "_compat" if self.by_row else False
856 kwargs.update({"by_row": by_row})
858 if getattr(obj, "axis", 0) == 1:
859 raise NotImplementedError("axis other than 0 is not supported")
861 selection = None
862 result_index, result_data = self.compute_dict_like(
863 op_name, obj, selection, kwargs
864 )
865 result = self.wrap_results_dict_like(obj, result_index, result_data)
866 return result
869class FrameApply(NDFrameApply):
870 obj: DataFrame
872 def __init__(
873 self,
874 obj: AggObjType,
875 func: AggFuncType,
876 raw: bool,
877 result_type: str | None,
878 *,
879 by_row: Literal[False, "compat"] = False,
880 engine: str = "python",
881 engine_kwargs: dict[str, bool] | None = None,
882 args,
883 kwargs,
884 ) -> None:
885 if by_row is not False and by_row != "compat":
886 raise ValueError(f"by_row={by_row} not allowed")
887 super().__init__(
888 obj,
889 func,
890 raw,
891 result_type,
892 by_row=by_row,
893 engine=engine,
894 engine_kwargs=engine_kwargs,
895 args=args,
896 kwargs=kwargs,
897 )
899 # ---------------------------------------------------------------
900 # Abstract Methods
902 @property
903 @abc.abstractmethod
904 def result_index(self) -> Index:
905 pass
907 @property
908 @abc.abstractmethod
909 def result_columns(self) -> Index:
910 pass
912 @property
913 @abc.abstractmethod
914 def series_generator(self) -> Generator[Series]:
915 pass
917 @staticmethod
918 @functools.cache
919 @abc.abstractmethod
920 def generate_numba_apply_func(
921 func, nogil=True, nopython=True, parallel=False
922 ) -> Callable[[npt.NDArray, Index, Index], dict[int, Any]]:
923 pass
925 @abc.abstractmethod
926 def apply_with_numba(self):
927 pass
929 def validate_values_for_numba(self) -> None:
930 # Validate column dtyps all OK
931 for colname, dtype in self.obj.dtypes.items():
932 if not is_numeric_dtype(dtype):
933 raise ValueError(
934 f"Column {colname} must have a numeric dtype. "
935 f"Found '{dtype}' instead"
936 )
937 if is_extension_array_dtype(dtype):
938 raise ValueError(
939 f"Column {colname} is backed by an extension array, "
940 f"which is not supported by the numba engine."
941 )
943 @abc.abstractmethod
944 def wrap_results_for_axis(
945 self, results: ResType, res_index: Index
946 ) -> DataFrame | Series:
947 pass
949 # ---------------------------------------------------------------
951 @property
952 def res_columns(self) -> Index:
953 return self.result_columns
955 @property
956 def columns(self) -> Index:
957 return self.obj.columns
959 @cache_readonly
960 def values(self):
961 return self.obj.values
963 def apply(self) -> DataFrame | Series:
964 """compute the results"""
966 # dispatch to handle list-like or dict-like
967 if is_list_like(self.func):
968 if self.engine == "numba":
969 raise NotImplementedError(
970 "the 'numba' engine doesn't support lists of callables yet"
971 )
972 return self.apply_list_or_dict_like()
974 # all empty
975 if len(self.columns) == 0 and len(self.index) == 0:
976 return self.apply_empty_result()
978 # string dispatch
979 if isinstance(self.func, str):
980 if self.engine == "numba":
981 raise NotImplementedError(
982 "the 'numba' engine doesn't support using "
983 "a string as the callable function"
984 )
985 return self.apply_str()
987 # ufunc
988 elif isinstance(self.func, np.ufunc):
989 if self.engine == "numba":
990 raise NotImplementedError(
991 "the 'numba' engine doesn't support "
992 "using a numpy ufunc as the callable function"
993 )
994 with np.errstate(all="ignore"):
995 results = self.obj._mgr.apply("apply", func=self.func)
996 # _constructor will retain self.index and self.columns
997 return self.obj._constructor_from_mgr(results, axes=results.axes)
999 # broadcasting
1000 if self.result_type == "broadcast":
1001 if self.engine == "numba":
1002 raise NotImplementedError(
1003 "the 'numba' engine doesn't support result_type='broadcast'"
1004 )
1005 return self.apply_broadcast(self.obj)
1007 # one axis empty
1008 elif not all(self.obj.shape):
1009 return self.apply_empty_result()
1011 # raw
1012 elif self.raw:
1013 return self.apply_raw(engine=self.engine, engine_kwargs=self.engine_kwargs)
1015 return self.apply_standard()
1017 def agg(self):
1018 obj = self.obj
1019 axis = self.axis
1021 # TODO: Avoid having to change state
1022 self.obj = self.obj if self.axis == 0 else self.obj.T
1023 self.axis = 0
1025 result = None
1026 try:
1027 result = super().agg()
1028 finally:
1029 self.obj = obj
1030 self.axis = axis
1032 if axis == 1:
1033 result = result.T if result is not None else result
1035 if result is None:
1036 result = self.obj.apply(self.func, axis, args=self.args, **self.kwargs)
1038 return result
1040 def apply_empty_result(self):
1041 """
1042 we have an empty result; at least 1 axis is 0
1044 we will try to apply the function to an empty
1045 series in order to see if this is a reduction function
1046 """
1047 assert callable(self.func)
1049 # we are not asked to reduce or infer reduction
1050 # so just return a copy of the existing object
1051 if self.result_type not in ["reduce", None]:
1052 return self.obj.copy()
1054 # we may need to infer
1055 should_reduce = self.result_type == "reduce"
1057 from pandas import Series
1059 if not should_reduce:
1060 try:
1061 if self.axis == 0:
1062 r = self.func(
1063 Series([], dtype=np.float64), *self.args, **self.kwargs
1064 )
1065 else:
1066 r = self.func(
1067 Series(index=self.columns, dtype=np.float64),
1068 *self.args,
1069 **self.kwargs,
1070 )
1071 except Exception:
1072 pass
1073 else:
1074 should_reduce = not isinstance(r, Series)
1076 if should_reduce:
1077 if len(self.agg_axis):
1078 r = self.func(Series([], dtype=np.float64), *self.args, **self.kwargs)
1079 else:
1080 r = np.nan
1082 return self.obj._constructor_sliced(r, index=self.agg_axis)
1083 else:
1084 return self.obj.copy()
1086 def apply_raw(self, engine="python", engine_kwargs=None):
1087 """apply to the values as a numpy array"""
1089 def wrap_function(func):
1090 """
1091 Wrap user supplied function to work around numpy issue.
1093 see https://github.com/numpy/numpy/issues/8352
1094 """
1096 def wrapper(*args, **kwargs):
1097 result = func(*args, **kwargs)
1098 if isinstance(result, str):
1099 result = np.array(result, dtype=object)
1100 return result
1102 return wrapper
1104 if engine == "numba":
1105 args, kwargs = prepare_function_arguments(
1106 self.func, # type: ignore[arg-type]
1107 self.args,
1108 self.kwargs,
1109 num_required_args=1,
1110 )
1111 # error: Argument 1 to "__call__" of "_lru_cache_wrapper" has
1112 # incompatible type "Callable[..., Any] | str | list[Callable
1113 # [..., Any] | str] | dict[Hashable,Callable[..., Any] | str |
1114 # list[Callable[..., Any] | str]]"; expected "Hashable"
1115 nb_looper = generate_apply_looper(
1116 self.func, # type: ignore[arg-type]
1117 **get_jit_arguments(engine_kwargs),
1118 )
1119 result = nb_looper(self.values, self.axis, *args)
1120 # If we made the result 2-D, squeeze it back to 1-D
1121 result = np.squeeze(result)
1122 else:
1123 result = np.apply_along_axis(
1124 wrap_function(self.func),
1125 self.axis,
1126 self.values,
1127 *self.args,
1128 **self.kwargs,
1129 )
1131 # TODO: mixed type case
1132 if result.ndim == 2:
1133 return self.obj._constructor(result, index=self.index, columns=self.columns)
1134 else:
1135 return self.obj._constructor_sliced(result, index=self.agg_axis)
1137 def apply_broadcast(self, target: DataFrame) -> DataFrame:
1138 assert callable(self.func)
1140 result_values = np.empty_like(target.values)
1142 # axis which we want to compare compliance
1143 result_compare = target.shape[0]
1145 for i, col in enumerate(target.columns):
1146 res = self.func(target[col], *self.args, **self.kwargs)
1147 ares = np.asarray(res).ndim
1149 # must be a scalar or 1d
1150 if ares > 1:
1151 raise ValueError("too many dims to broadcast")
1152 if ares == 1:
1153 # must match return dim
1154 if result_compare != len(res):
1155 raise ValueError("cannot broadcast result")
1157 result_values[:, i] = res
1159 # we *always* preserve the original index / columns
1160 result = self.obj._constructor(
1161 result_values, index=target.index, columns=target.columns
1162 )
1163 return result
1165 def apply_standard(self):
1166 if self.engine == "python":
1167 results, res_index = self.apply_series_generator()
1168 else:
1169 results, res_index = self.apply_series_numba()
1171 # wrap results
1172 return self.wrap_results(results, res_index)
1174 def apply_series_generator(self) -> tuple[ResType, Index]:
1175 assert callable(self.func)
1177 series_gen = self.series_generator
1178 res_index = self.result_index
1180 results = {}
1182 for i, v in enumerate(series_gen):
1183 results[i] = self.func(v, *self.args, **self.kwargs)
1184 if isinstance(results[i], ABCSeries):
1185 # If we have a view on v, we need to make a copy because
1186 # series_generator will swap out the underlying data
1187 results[i] = results[i].copy(deep=False)
1189 return results, res_index
1191 def apply_series_numba(self):
1192 if self.engine_kwargs.get("parallel", False):
1193 raise NotImplementedError(
1194 "Parallel apply is not supported when raw=False and engine='numba'"
1195 )
1196 if not self.obj.index.is_unique or not self.columns.is_unique:
1197 raise NotImplementedError(
1198 "The index/columns must be unique when raw=False and engine='numba'"
1199 )
1200 self.validate_values_for_numba()
1201 results = self.apply_with_numba()
1202 return results, self.result_index
1204 def wrap_results(self, results: ResType, res_index: Index) -> DataFrame | Series:
1205 from pandas import Series
1207 # see if we can infer the results
1208 if len(results) > 0 and 0 in results and is_sequence(results[0]):
1209 return self.wrap_results_for_axis(results, res_index)
1211 # dict of scalars
1213 # the default dtype of an empty Series is `object`, but this
1214 # code can be hit by df.mean() where the result should have dtype
1215 # float64 even if it's an empty Series.
1216 constructor_sliced = self.obj._constructor_sliced
1217 if len(results) == 0 and constructor_sliced is Series:
1218 result = constructor_sliced(results, dtype=np.float64)
1219 else:
1220 result = constructor_sliced(results)
1221 result.index = res_index
1223 return result
1225 def apply_str(self) -> DataFrame | Series:
1226 # Caller is responsible for checking isinstance(self.func, str)
1227 # TODO: GH#39993 - Avoid special-casing by replacing with lambda
1228 if self.func == "size":
1229 # Special-cased because DataFrame.size returns a single scalar
1230 obj = self.obj
1231 value = obj.shape[self.axis]
1232 return obj._constructor_sliced(value, index=self.agg_axis)
1233 return super().apply_str()
1236class FrameRowApply(FrameApply):
1237 axis: AxisInt = 0
1239 @property
1240 def series_generator(self) -> Generator[Series]:
1241 return (self.obj._ixs(i, axis=1) for i in range(len(self.columns)))
1243 @staticmethod
1244 @functools.cache
1245 def generate_numba_apply_func(
1246 func, nogil=True, nopython=True, parallel=False
1247 ) -> Callable[[npt.NDArray, Index, Index], dict[int, Any]]:
1248 numba = import_optional_dependency("numba")
1249 from pandas import Series
1251 # Import helper from extensions to cast string object -> np strings
1252 # Note: This also has the side effect of loading our numba extensions
1253 from pandas.core._numba.extensions import maybe_cast_str
1255 jitted_udf = numba.extending.register_jitable(func)
1257 # Currently the parallel argument doesn't get passed through here
1258 # (it's disabled) since the dicts in numba aren't thread-safe.
1259 @numba.jit(nogil=nogil, nopython=nopython, parallel=parallel)
1260 def numba_func(values, col_names, df_index, *args):
1261 results = {}
1262 for j in range(values.shape[1]):
1263 # Create the series
1264 ser = Series(
1265 values[:, j], index=df_index, name=maybe_cast_str(col_names[j])
1266 )
1267 results[j] = jitted_udf(ser, *args)
1268 return results
1270 return numba_func
1272 def apply_with_numba(self) -> dict[int, Any]:
1273 func = cast(Callable, self.func)
1274 args, kwargs = prepare_function_arguments(
1275 func, self.args, self.kwargs, num_required_args=1
1276 )
1277 nb_func = self.generate_numba_apply_func(
1278 func, **get_jit_arguments(self.engine_kwargs)
1279 )
1280 from pandas.core._numba.extensions import set_numba_data
1282 index = self.obj.index
1283 columns = self.obj.columns
1285 # Convert from numba dict to regular dict
1286 # Our isinstance checks in the df constructor don't pass for numbas typed dict
1287 with set_numba_data(index) as index, set_numba_data(columns) as columns:
1288 res = dict(nb_func(self.values, columns, index, *args))
1289 return res
1291 @property
1292 def result_index(self) -> Index:
1293 return self.columns
1295 @property
1296 def result_columns(self) -> Index:
1297 return self.index
1299 def wrap_results_for_axis(
1300 self, results: ResType, res_index: Index
1301 ) -> DataFrame | Series:
1302 """return the results for the rows"""
1304 if self.result_type == "reduce":
1305 # e.g. test_apply_dict GH#8735
1306 res = self.obj._constructor_sliced(results)
1307 res.index = res_index
1308 return res
1310 elif self.result_type is None and all(
1311 isinstance(x, dict) for x in results.values()
1312 ):
1313 # Our operation was a to_dict op e.g.
1314 # test_apply_dict GH#8735, test_apply_reduce_to_dict GH#25196 #37544
1315 res = self.obj._constructor_sliced(results)
1316 res.index = res_index
1317 return res
1319 try:
1320 result = self.obj._constructor(data=results)
1321 except ValueError as err:
1322 if "All arrays must be of the same length" in str(err):
1323 # e.g. result = [[2, 3], [1.5], ['foo', 'bar']]
1324 # see test_agg_listlike_result GH#29587
1325 res = self.obj._constructor_sliced(results)
1326 res.index = res_index
1327 return res
1328 else:
1329 raise
1331 if not isinstance(results[0], ABCSeries):
1332 if len(result.index) == len(self.res_columns):
1333 result.index = self.res_columns
1335 if len(result.columns) == len(res_index):
1336 result.columns = res_index
1338 return result
1341class FrameColumnApply(FrameApply):
1342 axis: AxisInt = 1
1344 def apply_broadcast(self, target: DataFrame) -> DataFrame:
1345 result = super().apply_broadcast(target.T)
1346 return result.T
1348 @property
1349 def series_generator(self) -> Generator[Series]:
1350 values = self.values
1351 values = ensure_wrapped_if_datetimelike(values)
1352 assert len(values) > 0
1354 # We create one Series object, and will swap out the data inside
1355 # of it. Kids: don't do this at home.
1356 ser = self.obj._ixs(0, axis=0)
1357 mgr = ser._mgr
1359 is_view = mgr.blocks[0].refs.has_reference()
1361 if isinstance(ser.dtype, ExtensionDtype):
1362 # values will be incorrect for this block
1363 # TODO(EA2D): special case would be unnecessary with 2D EAs
1364 obj = self.obj
1365 for i in range(len(obj)):
1366 yield obj._ixs(i, axis=0)
1368 else:
1369 for arr, name in zip(values, self.index, strict=True):
1370 # GH#35462 re-pin mgr in case setitem changed it
1371 ser._mgr = mgr
1372 mgr.set_values(arr)
1373 object.__setattr__(ser, "_name", name)
1374 if not is_view:
1375 # In apply_series_generator we store the a shallow copy of the
1376 # result, which potentially increases the ref count of this reused
1377 # `ser` object (depending on the result of the applied function)
1378 # -> if that happened and `ser` is already a copy, then we reset
1379 # the refs here to avoid triggering a unnecessary CoW inside the
1380 # applied function (https://github.com/pandas-dev/pandas/pull/56212)
1381 mgr.blocks[0].refs = BlockValuesRefs(mgr.blocks[0])
1382 yield ser
1384 @staticmethod
1385 @functools.cache
1386 def generate_numba_apply_func(
1387 func, nogil=True, nopython=True, parallel=False
1388 ) -> Callable[[npt.NDArray, Index, Index], dict[int, Any]]:
1389 numba = import_optional_dependency("numba")
1390 from pandas import Series
1391 from pandas.core._numba.extensions import maybe_cast_str
1393 jitted_udf = numba.extending.register_jitable(func)
1395 @numba.jit(nogil=nogil, nopython=nopython, parallel=parallel)
1396 def numba_func(values, col_names_index, index, *args):
1397 results = {}
1398 # Currently the parallel argument doesn't get passed through here
1399 # (it's disabled) since the dicts in numba aren't thread-safe.
1400 for i in range(values.shape[0]):
1401 # Create the series
1402 # TODO: values corrupted without the copy
1403 ser = Series(
1404 values[i].copy(),
1405 index=col_names_index,
1406 name=maybe_cast_str(index[i]),
1407 )
1408 results[i] = jitted_udf(ser, *args)
1410 return results
1412 return numba_func
1414 def apply_with_numba(self) -> dict[int, Any]:
1415 func = cast(Callable, self.func)
1416 args, kwargs = prepare_function_arguments(
1417 func, self.args, self.kwargs, num_required_args=1
1418 )
1419 nb_func = self.generate_numba_apply_func(
1420 func, **get_jit_arguments(self.engine_kwargs)
1421 )
1423 from pandas.core._numba.extensions import set_numba_data
1425 # Convert from numba dict to regular dict
1426 # Our isinstance checks in the df constructor don't pass for numbas typed dict
1427 with (
1428 set_numba_data(self.obj.index) as index,
1429 set_numba_data(self.columns) as columns,
1430 ):
1431 res = dict(nb_func(self.values, columns, index, *args))
1433 return res
1435 @property
1436 def result_index(self) -> Index:
1437 return self.index
1439 @property
1440 def result_columns(self) -> Index:
1441 return self.columns
1443 def wrap_results_for_axis(
1444 self, results: ResType, res_index: Index
1445 ) -> DataFrame | Series:
1446 """return the results for the columns"""
1447 result: DataFrame | Series
1449 # we have requested to expand
1450 if self.result_type == "expand":
1451 result = self.infer_to_same_shape(results, res_index)
1453 # we have a non-series and don't want inference
1454 elif not isinstance(results[0], ABCSeries):
1455 result = self.obj._constructor_sliced(results)
1456 result.index = res_index
1458 # we may want to infer results
1459 else:
1460 result = self.infer_to_same_shape(results, res_index)
1462 return result
1464 def infer_to_same_shape(self, results: ResType, res_index: Index) -> DataFrame:
1465 """infer the results to the same shape as the input object"""
1466 result = self.obj._constructor(data=results)
1467 result = result.T
1469 # set the index
1470 result.index = res_index
1472 # infer dtypes
1473 result = result.infer_objects()
1475 return result
1478class SeriesApply(NDFrameApply):
1479 obj: Series
1480 axis: AxisInt = 0
1481 by_row: Literal[False, "compat", "_compat"] # only relevant for apply()
1483 def __init__(
1484 self,
1485 obj: Series,
1486 func: AggFuncType,
1487 *,
1488 by_row: Literal[False, "compat", "_compat"] = "compat",
1489 args,
1490 kwargs,
1491 ) -> None:
1492 super().__init__(
1493 obj,
1494 func,
1495 raw=False,
1496 result_type=None,
1497 by_row=by_row,
1498 args=args,
1499 kwargs=kwargs,
1500 )
1502 def apply(self) -> DataFrame | Series:
1503 obj = self.obj
1505 if len(obj) == 0:
1506 return self.apply_empty_result()
1508 # dispatch to handle list-like or dict-like
1509 if is_list_like(self.func):
1510 return self.apply_list_or_dict_like()
1512 if isinstance(self.func, str):
1513 # if we are a string, try to dispatch
1514 return self.apply_str()
1516 if self.by_row == "_compat":
1517 return self.apply_compat()
1519 # self.func is Callable
1520 return self.apply_standard()
1522 def agg(self):
1523 result = super().agg()
1524 if result is None:
1525 obj = self.obj
1526 func = self.func
1527 # string, list-like, and dict-like are entirely handled in super
1528 assert callable(func)
1529 result = func(obj, *self.args, **self.kwargs)
1530 return result
1532 def apply_empty_result(self) -> Series:
1533 obj = self.obj
1534 return obj._constructor(dtype=obj.dtype, index=obj.index).__finalize__(
1535 obj, method="apply"
1536 )
1538 def apply_compat(self):
1539 """compat apply method for funcs in listlikes and dictlikes.
1541 Used for each callable when giving listlikes and dictlikes of callables to
1542 apply. Needed for compatibility with Pandas < v2.1.
1544 .. versionadded:: 2.1.0
1545 """
1546 obj = self.obj
1547 func = self.func
1549 if callable(func):
1550 f = com.get_cython_func(func)
1551 if f and not self.args and not self.kwargs:
1552 return obj.apply(func, by_row=False)
1554 try:
1555 result = obj.apply(func, by_row="compat")
1556 except (ValueError, AttributeError, TypeError):
1557 result = obj.apply(func, by_row=False)
1558 return result
1560 def apply_standard(self) -> DataFrame | Series:
1561 # caller is responsible for ensuring that f is Callable
1562 func = cast(Callable, self.func)
1563 obj = self.obj
1565 if isinstance(func, np.ufunc):
1566 with np.errstate(all="ignore"):
1567 return func(obj, *self.args, **self.kwargs)
1568 elif not self.by_row:
1569 return func(obj, *self.args, **self.kwargs)
1571 if self.args or self.kwargs:
1572 # _map_values does not support args/kwargs
1573 def curried(x):
1574 return func(x, *self.args, **self.kwargs)
1576 else:
1577 curried = func
1578 mapped = obj._map_values(mapper=curried)
1580 if len(mapped) and isinstance(mapped[0], ABCSeries):
1581 # GH#43986 Need to do list(mapped) in order to get treated as nested
1582 # See also GH#25959 regarding EA support
1583 return obj._constructor_expanddim(list(mapped), index=obj.index)
1584 else:
1585 return obj._constructor(mapped, index=obj.index).__finalize__(
1586 obj, method="apply"
1587 )
1590class GroupByApply(Apply):
1591 obj: GroupBy | Resampler | BaseWindow
1593 def __init__(
1594 self,
1595 obj: GroupBy[NDFrameT],
1596 func: AggFuncType,
1597 *,
1598 args,
1599 kwargs,
1600 ) -> None:
1601 kwargs = kwargs.copy()
1602 self.axis = obj.obj._get_axis_number(kwargs.get("axis", 0))
1603 super().__init__(
1604 obj,
1605 func,
1606 raw=False,
1607 result_type=None,
1608 args=args,
1609 kwargs=kwargs,
1610 )
1612 def apply(self):
1613 raise NotImplementedError
1615 def transform(self):
1616 raise NotImplementedError
1618 def agg_or_apply_list_like(
1619 self, op_name: Literal["agg", "apply"]
1620 ) -> DataFrame | Series:
1621 obj = self.obj
1622 kwargs = self.kwargs
1623 if op_name == "apply":
1624 kwargs = {**kwargs, "by_row": False}
1626 if getattr(obj, "axis", 0) == 1:
1627 raise NotImplementedError("axis other than 0 is not supported")
1629 if obj._selected_obj.ndim == 1:
1630 # For SeriesGroupBy this matches _obj_with_exclusions
1631 selected_obj = obj._selected_obj
1632 else:
1633 selected_obj = obj._obj_with_exclusions
1635 # Only set as_index=True on groupby objects, not Window or Resample
1636 # that inherit from this class.
1637 with com.temp_setattr(
1638 obj, "as_index", True, condition=hasattr(obj, "as_index")
1639 ):
1640 keys, results = self.compute_list_like(op_name, selected_obj, kwargs)
1641 result = self.wrap_results_list_like(keys, results)
1642 return result
1644 def agg_or_apply_dict_like(
1645 self, op_name: Literal["agg", "apply"]
1646 ) -> DataFrame | Series:
1647 from pandas.core.groupby.generic import (
1648 DataFrameGroupBy,
1649 SeriesGroupBy,
1650 )
1652 assert op_name in ["agg", "apply"]
1654 obj = self.obj
1655 kwargs: dict[str, Any] = {}
1656 if op_name == "apply":
1657 by_row = "_compat" if self.by_row else False
1658 kwargs.update({"by_row": by_row})
1660 if getattr(obj, "axis", 0) == 1:
1661 raise NotImplementedError("axis other than 0 is not supported")
1663 selected_obj = obj._selected_obj
1664 selection = obj._selection
1666 is_groupby = isinstance(obj, (DataFrameGroupBy, SeriesGroupBy))
1668 # Numba Groupby engine/engine-kwargs passthrough
1669 if is_groupby:
1670 engine = self.kwargs.get("engine", None)
1671 engine_kwargs = self.kwargs.get("engine_kwargs", None)
1672 kwargs.update({"engine": engine, "engine_kwargs": engine_kwargs})
1674 with com.temp_setattr(
1675 obj, "as_index", True, condition=hasattr(obj, "as_index")
1676 ):
1677 result_index, result_data = self.compute_dict_like(
1678 op_name, selected_obj, selection, kwargs
1679 )
1680 result = self.wrap_results_dict_like(selected_obj, result_index, result_data)
1681 return result
1684class ResamplerWindowApply(GroupByApply):
1685 axis: AxisInt = 0
1686 obj: Resampler | BaseWindow
1688 def __init__(
1689 self,
1690 obj: Resampler | BaseWindow,
1691 func: AggFuncType,
1692 *,
1693 args,
1694 kwargs,
1695 ) -> None:
1696 super(GroupByApply, self).__init__(
1697 obj,
1698 func,
1699 raw=False,
1700 result_type=None,
1701 args=args,
1702 kwargs=kwargs,
1703 )
1705 def apply(self):
1706 raise NotImplementedError
1708 def transform(self):
1709 raise NotImplementedError
1712def reconstruct_func(
1713 func: AggFuncType | None, **kwargs
1714) -> tuple[bool, AggFuncType, tuple[str, ...] | None, npt.NDArray[np.intp] | None]:
1715 """
1716 This is the internal function to reconstruct func given if there is relabeling
1717 or not and also normalize the keyword to get new order of columns.
1719 If named aggregation is applied, `func` will be None, and kwargs contains the
1720 column and aggregation function information to be parsed;
1721 If named aggregation is not applied, `func` is either string (e.g. 'min') or
1722 Callable, or list of them (e.g. ['min', np.max]), or the dictionary of column name
1723 and str/Callable/list of them (e.g. {'A': 'min'}, or {'A': [np.min, lambda x: x]})
1725 If relabeling is True, will return relabeling, reconstructed func, column
1726 names, and the reconstructed order of columns.
1727 If relabeling is False, the columns and order will be None.
1729 Parameters
1730 ----------
1731 func: agg function (e.g. 'min' or Callable) or list of agg functions
1732 (e.g. ['min', np.max]) or dictionary (e.g. {'A': ['min', np.max]}).
1733 **kwargs: dict, kwargs used in is_multi_agg_with_relabel and
1734 normalize_keyword_aggregation function for relabelling
1736 Returns
1737 -------
1738 relabelling: bool, if there is relabelling or not
1739 func: normalized and mangled func
1740 columns: tuple of column names
1741 order: array of columns indices
1743 Examples
1744 --------
1745 >>> reconstruct_func(None, **{"foo": ("col", "min")})
1746 (True, defaultdict(<class 'list'>, {'col': ['min']}), ('foo',), array([0]))
1748 >>> reconstruct_func("min")
1749 (False, 'min', None, None)
1750 """
1751 from pandas.core.groupby.generic import NamedAgg
1753 relabeling = func is None and (
1754 is_multi_agg_with_relabel(**kwargs)
1755 or any(isinstance(v, NamedAgg) for v in kwargs.values())
1756 )
1758 columns: tuple[str, ...] | None = None
1759 order: npt.NDArray[np.intp] | None = None
1761 if not relabeling:
1762 if isinstance(func, list) and len(func) > len(set(func)):
1763 # GH 28426 will raise error if duplicated function names are used and
1764 # there is no reassigned name
1765 raise SpecificationError(
1766 "Function names must be unique if there is no new column names assigned"
1767 )
1768 if func is None:
1769 # nicer error message
1770 raise TypeError("Must provide 'func' or tuples of '(column, aggfunc).")
1772 if relabeling:
1773 # error: Incompatible types in assignment (expression has type
1774 # "MutableMapping[Hashable, list[Callable[..., Any] | str]]", variable has type
1775 # "Callable[..., Any] | str | list[Callable[..., Any] | str] |
1776 # MutableMapping[Hashable, Callable[..., Any] | str | list[Callable[..., Any] |
1777 # str]] | None")
1778 converted_kwargs = {}
1779 for key, val in kwargs.items():
1780 if isinstance(val, NamedAgg):
1781 aggfunc = val.aggfunc
1782 if val.args or val.kwargs:
1783 aggfunc = lambda x, func=aggfunc, a=val.args, kw=val.kwargs: func(
1784 x, *a, **kw
1785 )
1786 converted_kwargs[key] = (val.column, aggfunc)
1787 else:
1788 converted_kwargs[key] = val
1790 func, columns, order = normalize_keyword_aggregation( # type: ignore[assignment]
1791 converted_kwargs
1792 )
1794 assert func is not None
1796 return relabeling, func, columns, order
1799def is_multi_agg_with_relabel(**kwargs) -> bool:
1800 """
1801 Check whether kwargs passed to .agg look like multi-agg with relabeling.
1803 Parameters
1804 ----------
1805 **kwargs : dict
1807 Returns
1808 -------
1809 bool
1811 Examples
1812 --------
1813 >>> is_multi_agg_with_relabel(a="max")
1814 False
1815 >>> is_multi_agg_with_relabel(a_max=("a", "max"), a_min=("a", "min"))
1816 True
1817 >>> is_multi_agg_with_relabel()
1818 False
1819 """
1820 return all(isinstance(v, tuple) and len(v) == 2 for v in kwargs.values()) and (
1821 len(kwargs) > 0
1822 )
1825def normalize_keyword_aggregation(
1826 kwargs: dict,
1827) -> tuple[
1828 MutableMapping[Hashable, list[AggFuncTypeBase]],
1829 tuple[str, ...],
1830 npt.NDArray[np.intp],
1831]:
1832 """
1833 Normalize user-provided "named aggregation" kwargs.
1834 Transforms from the new ``Mapping[str, NamedAgg]`` style kwargs
1835 to the old Dict[str, List[scalar]]].
1837 Parameters
1838 ----------
1839 kwargs : dict
1841 Returns
1842 -------
1843 aggspec : dict
1844 The transformed kwargs.
1845 columns : tuple[str, ...]
1846 The user-provided keys.
1847 col_idx_order : List[int]
1848 List of columns indices.
1850 Examples
1851 --------
1852 >>> normalize_keyword_aggregation({"output": ("input", "sum")})
1853 (defaultdict(<class 'list'>, {'input': ['sum']}), ('output',), array([0]))
1854 """
1855 from pandas.core.indexes.base import Index
1857 # Normalize the aggregation functions as Mapping[column, List[func]],
1858 # process normally, then fixup the names.
1859 # TODO: aggspec type: typing.Dict[str, List[AggScalar]]
1860 aggspec = defaultdict(list)
1861 order = []
1862 columns = tuple(kwargs.keys())
1864 for column, aggfunc in kwargs.values():
1865 aggspec[column].append(aggfunc)
1866 order.append((column, com.get_callable_name(aggfunc) or aggfunc))
1868 # uniquify aggfunc name if duplicated in order list
1869 uniquified_order = _make_unique_kwarg_list(order)
1871 # GH 25719, due to aggspec will change the order of assigned columns in aggregation
1872 # uniquified_aggspec will store uniquified order list and will compare it with order
1873 # based on index
1874 aggspec_order = [
1875 (column, com.get_callable_name(aggfunc) or aggfunc)
1876 for column, aggfuncs in aggspec.items()
1877 for aggfunc in aggfuncs
1878 ]
1879 uniquified_aggspec = _make_unique_kwarg_list(aggspec_order)
1881 # get the new index of columns by comparison
1882 col_idx_order = Index(uniquified_aggspec).get_indexer(uniquified_order)
1883 return aggspec, columns, col_idx_order
1886def _make_unique_kwarg_list(
1887 seq: Sequence[tuple[Any, Any]],
1888) -> Sequence[tuple[Any, Any]]:
1889 """
1890 Uniquify aggfunc name of the pairs in the order list
1892 Examples:
1893 --------
1894 >>> kwarg_list = [("a", "<lambda>"), ("a", "<lambda>"), ("b", "<lambda>")]
1895 >>> _make_unique_kwarg_list(kwarg_list)
1896 [('a', '<lambda>_0'), ('a', '<lambda>_1'), ('b', '<lambda>')]
1897 """
1898 return [
1899 (pair[0], f"{pair[1]}_{seq[:i].count(pair)}") if seq.count(pair) > 1 else pair
1900 for i, pair in enumerate(seq)
1901 ]
1904def relabel_result(
1905 result: DataFrame | Series,
1906 func: dict[str, list[Callable | str]],
1907 columns: Iterable[Hashable],
1908 order: Iterable[int],
1909) -> dict[Hashable, Series]:
1910 """
1911 Internal function to reorder result if relabelling is True for
1912 dataframe.agg, and return the reordered result in dict.
1914 Parameters:
1915 ----------
1916 result: Result from aggregation
1917 func: Dict of (column name, funcs)
1918 columns: New columns name for relabelling
1919 order: New order for relabelling
1921 Examples
1922 --------
1923 >>> from pandas.core.apply import relabel_result
1924 >>> result = pd.DataFrame(
1925 ... {"A": [np.nan, 2, np.nan], "C": [6, np.nan, np.nan], "B": [np.nan, 4, 2.5]},
1926 ... index=["max", "mean", "min"],
1927 ... )
1928 >>> funcs = {"A": ["max"], "C": ["max"], "B": ["mean", "min"]}
1929 >>> columns = ("foo", "aab", "bar", "dat")
1930 >>> order = [0, 1, 2, 3]
1931 >>> result_in_dict = relabel_result(result, funcs, columns, order)
1932 >>> pd.DataFrame(result_in_dict, index=columns)
1933 A C B
1934 foo 2.0 NaN NaN
1935 aab NaN 6.0 NaN
1936 bar NaN NaN 4.0
1937 dat NaN NaN 2.5
1938 """
1939 from pandas.core.indexes.base import Index
1941 reordered_indexes = [
1942 pair[0] for pair in sorted(zip(columns, order, strict=True), key=lambda t: t[1])
1943 ]
1944 reordered_result_in_dict: dict[Hashable, Series] = {}
1945 idx = 0
1947 reorder_mask = not isinstance(result, ABCSeries) and len(result.columns) > 1
1948 for col, fun in func.items():
1949 s = result[col].dropna()
1951 # In the `_aggregate`, the callable names are obtained and used in `result`, and
1952 # these names are ordered alphabetically. e.g.
1953 # C2 C1
1954 # <lambda> 1 NaN
1955 # amax NaN 4.0
1956 # max NaN 4.0
1957 # sum 18.0 6.0
1958 # Therefore, the order of functions for each column could be shuffled
1959 # accordingly so need to get the callable name if it is not parsed names, and
1960 # reorder the aggregated result for each column.
1961 # e.g. if df.agg(c1=("C2", sum), c2=("C2", lambda x: min(x))), correct order is
1962 # [sum, <lambda>], but in `result`, it will be [<lambda>, sum], and we need to
1963 # reorder so that aggregated values map to their functions regarding the order.
1965 # However there is only one column being used for aggregation, not need to
1966 # reorder since the index is not sorted, and keep as is in `funcs`, e.g.
1967 # A
1968 # min 1.0
1969 # mean 1.5
1970 # mean 1.5
1971 if reorder_mask:
1972 fun = [
1973 com.get_callable_name(f) if not isinstance(f, str) else f for f in fun
1974 ]
1975 col_idx_order = Index(s.index, copy=False).get_indexer(fun)
1976 valid_idx = col_idx_order != -1
1977 if valid_idx.any():
1978 s = s.iloc[col_idx_order[valid_idx]]
1979 # assign the new user-provided "named aggregation" as index names, and reindex
1980 # it based on the whole user-provided names.
1981 if not s.empty:
1982 s.index = reordered_indexes[idx : idx + len(fun)]
1983 reordered_result_in_dict[col] = s.reindex(columns)
1984 idx = idx + len(fun)
1985 return reordered_result_in_dict
1988def reconstruct_and_relabel_result(result, func, **kwargs) -> DataFrame | Series:
1989 from pandas import DataFrame
1991 relabeling, func, columns, order = reconstruct_func(func, **kwargs)
1993 if relabeling:
1994 # This is to keep the order to columns occurrence unchanged, and also
1995 # keep the order of new columns occurrence unchanged
1997 # For the return values of reconstruct_func, if relabeling is
1998 # False, columns and order will be None.
1999 assert columns is not None
2000 assert order is not None
2002 result_in_dict = relabel_result(result, func, columns, order)
2003 result = DataFrame(result_in_dict, index=columns)
2005 return result
2008# TODO: Can't use, because mypy doesn't like us setting __name__
2009# error: "partial[Any]" has no attribute "__name__"
2010# the type is:
2011# typing.Sequence[Callable[..., ScalarResult]]
2012# -> typing.Sequence[Callable[..., ScalarResult]]:
2015def _managle_lambda_list(aggfuncs: Sequence[Any]) -> Sequence[Any]:
2016 """
2017 Possibly mangle a list of aggfuncs.
2019 Parameters
2020 ----------
2021 aggfuncs : Sequence
2023 Returns
2024 -------
2025 mangled: list-like
2026 A new AggSpec sequence, where lambdas have been converted
2027 to have unique names.
2029 Notes
2030 -----
2031 If just one aggfunc is passed, the name will not be mangled.
2032 """
2033 if len(aggfuncs) <= 1:
2034 # don't mangle for .agg([lambda x: .])
2035 return aggfuncs
2036 i = 0
2037 mangled_aggfuncs = []
2038 for aggfunc in aggfuncs:
2039 if com.get_callable_name(aggfunc) == "<lambda>":
2040 aggfunc = partial(aggfunc)
2041 # error: "partial[Any]" has no attribute "__name__"; maybe "__new__"?
2042 aggfunc.__name__ = f"<lambda_{i}>" # type: ignore[attr-defined]
2043 i += 1
2044 mangled_aggfuncs.append(aggfunc)
2046 return mangled_aggfuncs
2049def maybe_mangle_lambdas(agg_spec: Any) -> Any:
2050 """
2051 Make new lambdas with unique names.
2053 Parameters
2054 ----------
2055 agg_spec : Any
2056 An argument to GroupBy.agg.
2057 Non-dict-like `agg_spec` are pass through as is.
2058 For dict-like `agg_spec` a new spec is returned
2059 with name-mangled lambdas.
2061 Returns
2062 -------
2063 mangled : Any
2064 Same type as the input.
2066 Examples
2067 --------
2068 >>> maybe_mangle_lambdas("sum")
2069 'sum'
2070 >>> maybe_mangle_lambdas([lambda: 1, lambda: 2]) # doctest: +SKIP
2071 [<function __main__.<lambda_0>,
2072 <function pandas...._make_lambda.<locals>.f(*args, **kwargs)>]
2073 """
2074 is_dict = is_dict_like(agg_spec)
2075 if not (is_dict or is_list_like(agg_spec)):
2076 return agg_spec
2077 mangled_aggspec = type(agg_spec)() # dict or OrderedDict
2079 if is_dict:
2080 for key, aggfuncs in agg_spec.items():
2081 if is_list_like(aggfuncs) and not is_dict_like(aggfuncs):
2082 mangled_aggfuncs = _managle_lambda_list(aggfuncs)
2083 else:
2084 mangled_aggfuncs = aggfuncs
2086 mangled_aggspec[key] = mangled_aggfuncs
2087 else:
2088 mangled_aggspec = _managle_lambda_list(agg_spec)
2090 return mangled_aggspec
2093def validate_func_kwargs(
2094 kwargs: dict,
2095) -> tuple[list[str], list[str | Callable[..., Any]]]:
2096 """
2097 Validates types of user-provided "named aggregation" kwargs.
2098 `TypeError` is raised if aggfunc is not `str` or callable.
2100 Parameters
2101 ----------
2102 kwargs : dict
2104 Returns
2105 -------
2106 columns : List[str]
2107 List of user-provided keys.
2108 func : List[Union[str, callable[...,Any]]]
2109 List of user-provided aggfuncs
2111 Examples
2112 --------
2113 >>> validate_func_kwargs({"one": "min", "two": "max"})
2114 (['one', 'two'], ['min', 'max'])
2115 """
2116 tuple_given_message = "func is expected but received {} in **kwargs."
2117 columns = list(kwargs)
2118 func = []
2119 for col_func in kwargs.values():
2120 if not (isinstance(col_func, str) or callable(col_func)):
2121 raise TypeError(tuple_given_message.format(type(col_func).__name__))
2122 func.append(col_func)
2123 if not columns:
2124 no_arg_message = "Must provide 'func' or named aggregation **kwargs."
2125 raise TypeError(no_arg_message)
2126 return columns, func
2129def include_axis(op_name: Literal["agg", "apply"], colg: Series | DataFrame) -> bool:
2130 return isinstance(colg, ABCDataFrame) or (
2131 isinstance(colg, ABCSeries) and op_name == "agg"
2132 )