<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def _grouped(input_type, output_type, base_class, output_type_method):
 """Define a user-defined function that is applied per group. Parameters input_type : List[ibis.expr.datatypes.DataType] A list of the types found in :mod:`~ibis.expr.datatypes`. The length of this list must match the number of arguments to the function. Variadic arguments are not yet supported. output_type : ibis.expr.datatypes.DataType The return type of the function. base_class : Type[T] The base class of the generated Node output_type_method : Callable A callable that determines the method to call to get the expression type of the UDF See Also -------- ibis.pandas.udf.reduction ibis.pandas.udf.analytic """


        def wrapper(func):
            funcsig = valid_function_signature(input_type, func)

            UDAFNode = type(
                func.__name__,
                (base_class,),
                {
                    'signature': sig.TypeSignature.from_dtypes(input_type),
                    'output_type': output_type_method(output_type),
                },
            )

            # An execution rule for a simple aggregate node
            @execute_node.register(
                UDAFNode, *udf_signature(input_type, pin=None, klass=pd.Series)
            )
            def execute_udaf_node(op, *args, **kwargs):
                args, kwargs = arguments_from_signature(
                    funcsig, *args, **kwargs
                )
                return func(*args, **kwargs)

            # An execution rule for a grouped aggregation node. This
            # includes aggregates applied over a window.
            nargs = len(input_type)
            group_by_signatures = [
                udf_signature(input_type, pin=pin, klass=SeriesGroupBy)
                for pin in range(nargs)
            ]

            @toolz.compose(
                *(
                    execute_node.register(UDAFNode, *types)
                    for types in group_by_signatures
                )
            )
            def execute_udaf_node_groupby(op, *args, **kwargs):
                # construct a generator that yields the next group of data
                # for every argument excluding the first (pandas performs
                # the iteration for the first argument) for each argument
                # that is a SeriesGroupBy.
                #
                # If the argument is not a SeriesGroupBy then keep
                # repeating it until all groups are exhausted.
                aggcontext = kwargs.pop('aggcontext', None)
                assert aggcontext is not None, 'aggcontext is None'
                iters = (
                    (data for _, data in arg)
                    if isinstance(arg, SeriesGroupBy)
                    else itertools.repeat(arg)
                    for arg in args[1:]
                )
                funcsig = signature(func)

                def aggregator(first, *rest, **kwargs):
                    # map(next, *rest) gets the inputs for the next group
                    # TODO: might be inefficient to do this on every call
                    args, kwargs = arguments_from_signature(
                        funcsig, first, *map(next, rest), **kwargs
                    )
                    return func(*args, **kwargs)

                result = aggcontext.agg(args[0], aggregator, *iters, **kwargs)
                return result

            @functools.wraps(func)
            def wrapped(*args):
                return UDAFNode(*args).to_expr()

            return wrapped

        return wrapper