[docs]classMaxConsecutiveZeros(AggregationPrimitive):"""Determines the maximum number of consecutive zero values in the input Args: skipna (bool): Ignore any `NaN` values in the input. Default is True. Examples: >>> max_consecutive_zeros = MaxConsecutiveZeros() >>> max_consecutive_zeros([1.0, -1.4, 0, 0.0, 0, -4.3]) 3 `NaN` values can be ignored with the `skipna` parameter >>> max_consecutive_zeros_skipna = MaxConsecutiveZeros(skipna=False) >>> max_consecutive_zeros_skipna([1.0, -1.4, 0, None, 0.0, -4.3]) 1 """name="max_consecutive_zeros"input_types=[[ColumnSchema(logical_type=Integer)],[ColumnSchema(logical_type=Double)],]return_type=ColumnSchema(logical_type=Integer,semantic_tags={"numeric"})stack_on_self=Falsedefault_value=0
defget_function(self):defmax_consecutive_zeros(x):ifself.skipna:x=x.dropna()# convert the numeric values to booleans for processingx[x.notnull()]=x[x.notnull()].eq(0)# find the locations where the value changes from the previous valuenot_equal=x!=x.shift()# Use cumulative sum to determine where consecutive values occur. When the# sum changes, consecutive non-zero values are present, when the cumulative# sum remains unchnaged, consecutive zero values are present.not_equal_sum=not_equal.cumsum()# group the input by the cumulative sum values and use cumulative count# to count the number of consecutive values. Add 1 to account for the cumulative# sum starting at zero where the first zero occursconsecutive=x.groupby(not_equal_sum).cumcount()+1# multiply by the boolean input to keep only the counts that correspond to# zero valuesconsecutive_zero=consecutive*x# return the max of all the consecutive zero valuesreturnconsecutive_zero.max()returnmax_consecutive_zeros