# Copyright 2023 Google LLC

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

"""Series is a 1 dimensional data structure."""

from __future__ import annotations

import functools

import itertools

import numbers

import os

import textwrap

import typing

from typing import Any, Literal, Mapping, Optional, Tuple, Union

import bigframes_vendored.pandas.core.series as vendored_pandas_series

import google.cloud.bigquery as bigquery

import numpy

import pandas

import pandas.core.dtypes.common

import typing_extensions

import bigframes.constants as constants

import bigframes.core

from bigframes.core import log_adapter

import bigframes.core.block_transforms as block_ops

import bigframes.core.blocks as blocks

import bigframes.core.expression as ex

import bigframes.core.groupby as groupby

import bigframes.core.indexers

import bigframes.core.indexes as indexes

import bigframes.core.ordering as order

import bigframes.core.scalar as scalars

import bigframes.core.utils as utils

import bigframes.core.window

import bigframes.core.window_spec

import bigframes.dataframe

import bigframes.dtypes

import bigframes.formatting_helpers as formatter

import bigframes.operations as ops

import bigframes.operations.aggregations as agg_ops

import bigframes.operations.base

import bigframes.operations.datetimes as dt

import bigframes.operations.plotting as plotting

import bigframes.operations.strings as strings

import bigframes.operations.structs as structs

LevelType = typing.Union[str, int]

LevelsType = typing.Union[LevelType, typing.Sequence[LevelType]]

_remote_function_recommendation_message = (

"Your functions could not be applied directly to the Series."

" Try converting it to a remote function."

)

@log_adapter.class_logger

class Series(bigframes.operations.base.SeriesMethods, vendored_pandas_series.Series):

def __init__(self, *args, **kwargs):

self._query_job: Optional[bigquery.QueryJob] = None

super().__init__(*args, **kwargs)

# Runs strict validations to ensure internal type predictions and ibis are completely in sync

# Do not execute these validations outside of testing suite.

if "PYTEST_CURRENT_TEST" in os.environ:

self._block.expr.validate_schema()

@property

def dt(self) -> dt.DatetimeMethods:

return dt.DatetimeMethods(self._block)

@property

def dtype(self):

return self._dtype

@property

def dtypes(self):

return self._dtype

@property

def loc(self) -> bigframes.core.indexers.LocSeriesIndexer:

return bigframes.core.indexers.LocSeriesIndexer(self)

@property

def iloc(self) -> bigframes.core.indexers.IlocSeriesIndexer:

return bigframes.core.indexers.IlocSeriesIndexer(self)

@property

def iat(self) -> bigframes.core.indexers.IatSeriesIndexer:

return bigframes.core.indexers.IatSeriesIndexer(self)

@property

def at(self) -> bigframes.core.indexers.AtSeriesIndexer:

return bigframes.core.indexers.AtSeriesIndexer(self)

@property

def name(self) -> blocks.Label:

return self._name

@name.setter

def name(self, label: blocks.Label):

new_block = self._block.with_column_labels([label])

self._set_block(new_block)

@property

def shape(self) -> typing.Tuple[int]:

return (self._block.shape[0],)

@property

def size(self) -> int:

return self.shape[0]

@property

def ndim(self) -> int:

return 1

@property

def empty(self) -> bool:

return self.shape[0] == 0

@property

def values(self) -> numpy.ndarray:

return self.to_numpy()

@property

def index(self) -> indexes.Index:

return indexes.Index.from_frame(self)

@property

def query_job(self) -> Optional[bigquery.QueryJob]:

"""BigQuery job metadata for the most recent query.

Returns:

The most recent `QueryJob

<https://cloud.google.com/python/docs/reference/bigquery/latest/google.cloud.bigquery.job.QueryJob>`_.

"""

if self._query_job is None:

self._set_internal_query_job(self._compute_dry_run())

return self._query_job

@property

def struct(self) -> structs.StructAccessor:

return structs.StructAccessor(self._block)

@property

def T(self) -> Series:

return self.transpose()

@property

def _info_axis(self) -> indexes.Index:

return self.index

@property

def _session(self) -> bigframes.Session:

return self._get_block().expr.session

def transpose(self) -> Series:

return self

def _set_internal_query_job(self, query_job: bigquery.QueryJob):

self._query_job = query_job

def __len__(self):

return self.shape[0]

def __iter__(self) -> typing.Iterator:

return itertools.chain.from_iterable(

map(lambda x: x.squeeze(axis=1), self._block.to_pandas_batches())

)

def copy(self) -> Series:

return Series(self._block)

def rename(

self, index: Union[blocks.Label, Mapping[Any, Any]] = None, **kwargs

) -> Series:

if len(kwargs) != 0:

raise NotImplementedError(

f"rename does not currently support any keyword arguments. {constants.FEEDBACK_LINK}"

)

# rename the Series name

if index is None or isinstance(

index, str

): # Python 3.9 doesn't allow isinstance of Optional

index = typing.cast(Optional[str], index)

block = self._block.with_column_labels([index])

return Series(block)

# rename the index

if isinstance(index, Mapping):

index = typing.cast(Mapping[Any, Any], index)

block = self._block

for k, v in index.items():

new_idx_ids = []

for idx_id, idx_dtype in zip(block.index_columns, block.index.dtypes):

# Will throw if key type isn't compatible with index type, which leads to invalid SQL.

block.create_constant(k, dtype=idx_dtype)

# Will throw if value type isn't compatible with index type.

block, const_id = block.create_constant(v, dtype=idx_dtype)

block, cond_id = block.project_expr(

ops.ne_op.as_expr(idx_id, ex.const(k))

)

block, new_idx_id = block.apply_ternary_op(

idx_id, cond_id, const_id, ops.where_op

)

new_idx_ids.append(new_idx_id)

block = block.drop_columns([const_id, cond_id])

block = block.set_index(new_idx_ids, index_labels=block.index.names)

return Series(block)

# rename the Series name

if isinstance(index, typing.Hashable):

index = typing.cast(Optional[str], index)

block = self._block.with_column_labels([index])

return Series(block)

raise ValueError(f"Unsupported type of parameter index: {type(index)}")

def rename_axis(

self,

mapper: typing.Union[blocks.Label, typing.Sequence[blocks.Label]],

**kwargs,

) -> Series:

if len(kwargs) != 0:

raise NotImplementedError(

f"rename_axis does not currently support any keyword arguments. {constants.FEEDBACK_LINK}"

)

# limited implementation: the new index name is simply the 'mapper' parameter

if _is_list_like(mapper):

labels = mapper

else:

labels = [mapper]

return Series(self._block.with_index_labels(labels))

def equals(

self, other: typing.Union[Series, bigframes.dataframe.DataFrame]

) -> bool:

# Must be same object type, same column dtypes, and same label values

if not isinstance(other, Series):

return False

return block_ops.equals(self._block, other._block)

def reset_index(

self,

*,

name: typing.Optional[str] = None,

drop: bool = False,

) -> bigframes.dataframe.DataFrame | Series:

block = self._block.reset_index(drop)

if drop:

return Series(block)

else:

if name:

block = block.assign_label(self._value_column, name)

return bigframes.dataframe.DataFrame(block)

def __repr__(self) -> str:

# TODO(swast): Add a timeout here? If the query is taking a long time,

# maybe we just print the job metadata that we have so far?

# TODO(swast): Avoid downloading the whole series by using job

# metadata, like we do with DataFrame.

opts = bigframes.options.display

max_results = opts.max_rows

if opts.repr_mode == "deferred":

return formatter.repr_query_job(self.query_job)

self._cached()

pandas_df, _, query_job = self._block.retrieve_repr_request_results(max_results)

self._set_internal_query_job(query_job)

return repr(pandas_df.iloc[:, 0])

def astype(

self,

dtype: Union[bigframes.dtypes.DtypeString, bigframes.dtypes.Dtype],

) -> Series:

return self._apply_unary_op(bigframes.operations.AsTypeOp(to_type=dtype))

def to_pandas(

self,

max_download_size: Optional[int] = None,

sampling_method: Optional[str] = None,

random_state: Optional[int] = None,

*,

ordered: bool = True,

) -> pandas.Series:

"""Writes Series to pandas Series.

Args:

max_download_size (int, default None):

Download size threshold in MB. If max_download_size is exceeded when downloading data

(e.g., to_pandas()), the data will be downsampled if

bigframes.options.sampling.enable_downsampling is True, otherwise, an error will be

raised. If set to a value other than None, this will supersede the global config.

sampling_method (str, default None):

Downsampling algorithms to be chosen from, the choices are: "head": This algorithm

returns a portion of the data from the beginning. It is fast and requires minimal

computations to perform the downsampling; "uniform": This algorithm returns uniform

random samples of the data. If set to a value other than None, this will supersede

the global config.

random_state (int, default None):

The seed for the uniform downsampling algorithm. If provided, the uniform method may

take longer to execute and require more computation. If set to a value other than

None, this will supersede the global config.

ordered (bool, default True):

Determines whether the resulting pandas series will be deterministically ordered.

In some cases, unordered may result in a faster-executing query.

Returns:

pandas.Series: A pandas Series with all rows of this Series if the data_sampling_threshold_mb

is not exceeded; otherwise, a pandas Series with downsampled rows of the DataFrame.

"""

df, query_job = self._block.to_pandas(

max_download_size=max_download_size,

sampling_method=sampling_method,

random_state=random_state,

ordered=ordered,

)

self._set_internal_query_job(query_job)

series = df.squeeze(axis=1)

series.name = self._name

return series

def _compute_dry_run(self) -> bigquery.QueryJob:

return self._block._compute_dry_run((self._value_column,))

def drop(

self,

labels: typing.Any = None,

*,

axis: typing.Union[int, str] = 0,

index: typing.Any = None,

columns: Union[blocks.Label, typing.Iterable[blocks.Label]] = None,

level: typing.Optional[LevelType] = None,

) -> Series:

if labels and index:

raise ValueError("Must specify exacly one of 'labels' or 'index'")

index = labels or index

# ignore axis, columns params

block = self._block

level_id = self._resolve_levels(level or 0)[0]

if _is_list_like(index):

block, inverse_condition_id = block.apply_unary_op(

level_id, ops.IsInOp(values=tuple(index), match_nulls=True)

)

block, condition_id = block.apply_unary_op(

inverse_condition_id, ops.invert_op

)

else:

block, condition_id = block.project_expr(

ops.ne_op.as_expr(level_id, ex.const(index))

)

block = block.filter_by_id(condition_id, keep_null=True)

block = block.drop_columns([condition_id])

return Series(block.select_column(self._value_column))

def droplevel(self, level: LevelsType, axis: int | str = 0):

resolved_level_ids = self._resolve_levels(level)

return Series(self._block.drop_levels(resolved_level_ids))

def swaplevel(self, i: int = -2, j: int = -1):

level_i = self._block.index_columns[i]

level_j = self._block.index_columns[j]

mapping = {level_i: level_j, level_j: level_i}

reordering = [

mapping.get(index_id, index_id) for index_id in self._block.index_columns

]

return Series(self._block.reorder_levels(reordering))

def reorder_levels(self, order: LevelsType, axis: int | str = 0):

resolved_level_ids = self._resolve_levels(order)

return Series(self._block.reorder_levels(resolved_level_ids))

def _resolve_levels(self, level: LevelsType) -> typing.Sequence[str]:

return self._block.index.resolve_level(level)

def between(self, left, right, inclusive="both"):

if inclusive not in ["both", "neither", "left", "right"]:

raise ValueError(

"Must set 'inclusive' to one of 'both', 'neither', 'left', or 'right'"

)

left_op = ops.ge_op if (inclusive in ["left", "both"]) else ops.gt_op

right_op = ops.le_op if (inclusive in ["right", "both"]) else ops.lt_op

return self._apply_binary_op(left, left_op).__and__(

self._apply_binary_op(right, right_op)

)

def cumsum(self) -> Series:

return self._apply_window_op(

agg_ops.sum_op, bigframes.core.window_spec.WindowSpec(following=0)

)

def ffill(self, *, limit: typing.Optional[int] = None) -> Series:

window = bigframes.core.window_spec.WindowSpec(preceding=limit, following=0)

return self._apply_window_op(agg_ops.LastNonNullOp(), window)

pad = ffill

def bfill(self, *, limit: typing.Optional[int] = None) -> Series:

window = bigframes.core.window_spec.WindowSpec(preceding=0, following=limit)

return self._apply_window_op(agg_ops.FirstNonNullOp(), window)

def cummax(self) -> Series:

return self._apply_window_op(

agg_ops.max_op, bigframes.core.window_spec.WindowSpec(following=0)

)

def cummin(self) -> Series:

return self._apply_window_op(

agg_ops.min_op, bigframes.core.window_spec.WindowSpec(following=0)

)

def cumprod(self) -> Series:

return self._apply_window_op(

agg_ops.product_op, bigframes.core.window_spec.WindowSpec(following=0)

)

def shift(self, periods: int = 1) -> Series:

window = bigframes.core.window_spec.WindowSpec(

preceding=periods if periods > 0 else None,

following=-periods if periods < 0 else None,

)

return self._apply_window_op(agg_ops.ShiftOp(periods), window)

def diff(self, periods: int = 1) -> Series:

window = bigframes.core.window_spec.WindowSpec(

preceding=periods if periods > 0 else None,

following=-periods if periods < 0 else None,

)

return self._apply_window_op(agg_ops.DiffOp(periods), window)

def pct_change(self, periods: int = 1) -> Series:

# Future versions of pandas will not perfrom ffill automatically

series = self.ffill()

return Series(block_ops.pct_change(series._block, periods=periods))

def rank(

self,

axis=0,

method: str = "average",

numeric_only=False,

na_option: str = "keep",

ascending: bool = True,

) -> Series:

return Series(block_ops.rank(self._block, method, na_option, ascending))

def fillna(self, value=None) -> Series:

return self._apply_binary_op(value, ops.fillna_op)

def replace(

self, to_replace: typing.Any, value: typing.Any = None, *, regex: bool = False

):

if regex:

# No-op unless to_replace and series dtype are both string type

if not isinstance(to_replace, str) or not isinstance(

self.dtype, pandas.StringDtype

):

return self

return self._regex_replace(to_replace, value)

elif utils.is_dict_like(to_replace):

return self._mapping_replace(to_replace) # type: ignore

elif utils.is_list_like(to_replace):

replace_list = to_replace

else: # Scalar

replace_list = [to_replace]

replace_list = [

i for i in replace_list if bigframes.dtypes.is_compatible(i, self.dtype)

]

return self._simple_replace(replace_list, value) if replace_list else self

def _regex_replace(self, to_replace: str, value: str):

if not bigframes.dtypes.is_dtype(value, self.dtype):

raise NotImplementedError(

f"Cannot replace {self.dtype} elements with incompatible item {value} as mixed-type columns not supported. {constants.FEEDBACK_LINK}"

)

block, result_col = self._block.apply_unary_op(

self._value_column,

ops.RegexReplaceStrOp(to_replace, value),

result_label=self.name,

)

return Series(block.select_column(result_col))

def _simple_replace(self, to_replace_list: typing.Sequence, value):

result_type = bigframes.dtypes.is_compatible(value, self.dtype)

if not result_type:

raise NotImplementedError(

f"Cannot replace {self.dtype} elements with incompatible item {value} as mixed-type columns not supported. {constants.FEEDBACK_LINK}"

)

if result_type != self.dtype:

return self.astype(result_type)._simple_replace(to_replace_list, value)

block, cond = self._block.apply_unary_op(

self._value_column, ops.IsInOp(tuple(to_replace_list))

)

block, result_col = block.project_expr(

ops.where_op.as_expr(ex.const(value), cond, self._value_column), self.name

)

return Series(block.select_column(result_col))

def _mapping_replace(self, mapping: dict[typing.Hashable, typing.Hashable]):

tuples = []

lcd_types: list[typing.Optional[bigframes.dtypes.Dtype]] = []

for key, value in mapping.items():

lcd_type = bigframes.dtypes.is_compatible(key, self.dtype)

if not lcd_type:

continue

if not bigframes.dtypes.is_dtype(value, self.dtype):

raise NotImplementedError(

f"Cannot replace {self.dtype} elements with incompatible item {value} as mixed-type columns not supported. {constants.FEEDBACK_LINK}"

)

tuples.append((key, value))

lcd_types.append(lcd_type)

result_dtype = functools.reduce(

lambda t1, t2: bigframes.dtypes.lcd_type(t1, t2) if (t1 and t2) else None,

lcd_types,

)

if not result_dtype:

raise NotImplementedError(

f"Cannot replace {self.dtype} elements with incompatible mapping {mapping} as mixed-type columns not supported. {constants.FEEDBACK_LINK}"

)

block, result = self._block.apply_unary_op(

self._value_column, ops.MapOp(tuple(tuples))

)

return Series(block.select_column(result))

def interpolate(self, method: str = "linear") -> Series:

if method == "pad":

return self.ffill()

result = block_ops.interpolate(self._block, method)

return Series(result)

def dropna(

self,

*,

axis: int = 0,

inplace: bool = False,

how: typing.Optional[str] = None,

ignore_index: bool = False,

) -> Series:

if inplace:

raise NotImplementedError("'inplace'=True not supported")

result = block_ops.dropna(self._block, [self._value_column], how="any")

if ignore_index:

result = result.reset_index()

return Series(result)

def head(self, n: int = 5) -> Series:

return typing.cast(Series, self.iloc[0:n])

def tail(self, n: int = 5) -> Series:

return typing.cast(Series, self.iloc[-n:])

def nlargest(self, n: int = 5, keep: str = "first") -> Series:

if keep not in ("first", "last", "all"):

raise ValueError("'keep must be one of 'first', 'last', or 'all'")

return Series(

block_ops.nlargest(self._block, n, [self._value_column], keep=keep)

)

def nsmallest(self, n: int = 5, keep: str = "first") -> Series:

if keep not in ("first", "last", "all"):

raise ValueError("'keep must be one of 'first', 'last', or 'all'")

return Series(

block_ops.nsmallest(self._block, n, [self._value_column], keep=keep)

)

def isin(self, values) -> "Series" | None:

if not _is_list_like(values):

raise TypeError(

"only list-like objects are allowed to be passed to "

f"isin(), you passed a [{type(values).__name__}]"

)

return self._apply_unary_op(

ops.IsInOp(values=tuple(values), match_nulls=True)

).fillna(value=False)

def isna(self) -> "Series":

return self._apply_unary_op(ops.isnull_op)

isnull = isna

def notna(self) -> "Series":

return self._apply_unary_op(ops.notnull_op)

notnull = notna

def __and__(self, other: bool | int | Series) -> Series:

return self._apply_binary_op(other, ops.and_op)

__rand__ = __and__

def __or__(self, other: bool | int | Series) -> Series:

return self._apply_binary_op(other, ops.or_op)

__ror__ = __or__

def __add__(self, other: float | int | Series) -> Series:

return self.add(other)

def __radd__(self, other: float | int | Series) -> Series:

return self.radd(other)

def add(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.add_op)

def radd(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.add_op, reverse=True)

def __sub__(self, other: float | int | Series) -> Series:

return self.sub(other)

def __rsub__(self, other: float | int | Series) -> Series:

return self.rsub(other)

def sub(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.sub_op)

def rsub(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.sub_op, reverse=True)

subtract = sub

def __mul__(self, other: float | int | Series) -> Series:

return self.mul(other)

def __rmul__(self, other: float | int | Series) -> Series:

return self.rmul(other)

def mul(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.mul_op)

def rmul(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.mul_op, reverse=True)

multiply = mul

def __truediv__(self, other: float | int | Series) -> Series:

return self.truediv(other)

def __rtruediv__(self, other: float | int | Series) -> Series:

return self.rtruediv(other)

def truediv(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.div_op)

def rtruediv(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.div_op, reverse=True)

div = truediv

divide = truediv

rdiv = rtruediv

def __floordiv__(self, other: float | int | Series) -> Series:

return self.floordiv(other)

def __rfloordiv__(self, other: float | int | Series) -> Series:

return self.rfloordiv(other)

def floordiv(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.floordiv_op)

def rfloordiv(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.floordiv_op, reverse=True)

def __pow__(self, other: float | int | Series) -> Series:

return self.pow(other)

def __rpow__(self, other: float | int | Series) -> Series:

return self.rpow(other)

def pow(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.pow_op)

def rpow(self, other: float | int | Series) -> Series:

return self._apply_binary_op(other, ops.pow_op, reverse=True)

def __lt__(self, other: float | int | Series) -> Series: # type: ignore

return self.lt(other)

def __le__(self, other: float | int | Series) -> Series: # type: ignore

return self.le(other)

def lt(self, other) -> Series:

return self._apply_binary_op(other, ops.lt_op)

def le(self, other) -> Series:

return self._apply_binary_op(other, ops.le_op)

def __gt__(self, other: float | int | Series) -> Series: # type: ignore

return self.gt(other)

def __ge__(self, other: float | int | Series) -> Series: # type: ignore

return self.ge(other)

def gt(self, other) -> Series:

return self._apply_binary_op(other, ops.gt_op)

def ge(self, other) -> Series:

return self._apply_binary_op(other, ops.ge_op)

def __mod__(self, other) -> Series: # type: ignore

return self.mod(other)

def __rmod__(self, other) -> Series: # type: ignore

return self.rmod(other)

def mod(self, other) -> Series: # type: ignore

return self._apply_binary_op(other, ops.mod_op)

def rmod(self, other) -> Series: # type: ignore

return self._apply_binary_op(other, ops.mod_op, reverse=True)

def divmod(self, other) -> Tuple[Series, Series]: # type: ignore

# TODO(huanc): when self and other both has dtype int and other contains zeros,

# the output should be dtype float, both floordiv and mod returns dtype int in this case.

return (self.floordiv(other), self.mod(other))

def rdivmod(self, other) -> Tuple[Series, Series]: # type: ignore

# TODO(huanc): when self and other both has dtype int and self contains zeros,

# the output should be dtype float, both floordiv and mod returns dtype int in this case.

return (self.rfloordiv(other), self.rmod(other))

def __matmul__(self, other):

return (self * other).sum()

dot = __matmul__

def abs(self) -> Series:

return self._apply_unary_op(ops.abs_op)

def round(self, decimals=0) -> "Series":

return self._apply_binary_op(decimals, ops.round_op)

def corr(self, other: Series, method="pearson", min_periods=None) -> float:

# TODO(tbergeron): Validate early that both are numeric

# TODO(tbergeron): Handle partially-numeric columns

if method != "pearson":

raise NotImplementedError(

f"Only Pearson correlation is currently supported. {constants.FEEDBACK_LINK}"

)

if min_periods:

raise NotImplementedError(

f"min_periods not yet supported. {constants.FEEDBACK_LINK}"

)

return self._apply_binary_aggregation(other, agg_ops.CorrOp())

def cov(self, other: Series) -> float:

return self._apply_binary_aggregation(other, agg_ops.CovOp())

def all(self) -> bool:

return typing.cast(bool, self._apply_aggregation(agg_ops.all_op))

def any(self) -> bool:

return typing.cast(bool, self._apply_aggregation(agg_ops.any_op))

def count(self) -> int:

return typing.cast(int, self._apply_aggregation(agg_ops.count_op))

def nunique(self) -> int:

return typing.cast(int, self._apply_aggregation(agg_ops.nunique_op))

def max(self) -> scalars.Scalar:

return self._apply_aggregation(agg_ops.max_op)

def min(self) -> scalars.Scalar:

return self._apply_aggregation(agg_ops.min_op)

def std(self) -> float:

return typing.cast(float, self._apply_aggregation(agg_ops.std_op))

def var(self) -> float:

return typing.cast(float, self._apply_aggregation(agg_ops.var_op))

def _central_moment(self, n: int) -> float:

"""Useful helper for calculating central moment statistics"""

# Nth central moment is mean((x-mean(x))^n)

# See: https://en.wikipedia.org/wiki/Moment_(mathematics)

mean_deltas = self - self.mean()

delta_powers = mean_deltas**n

return delta_powers.mean()

def agg(self, func: str | typing.Sequence[str]) -> scalars.Scalar | Series:

if _is_list_like(func):

if self.dtype not in bigframes.dtypes.NUMERIC_BIGFRAMES_TYPES_PERMISSIVE:

raise NotImplementedError(

f"Multiple aggregations only supported on numeric series. {constants.FEEDBACK_LINK}"

)

aggregations = [agg_ops.lookup_agg_func(f) for f in func]

return Series(

self._block.summarize(

[self._value_column],

aggregations,

)

)

else:

return self._apply_aggregation(

agg_ops.lookup_agg_func(typing.cast(str, func))

)

aggregate = agg

def skew(self):

count = self.count()

if count < 3:

return pandas.NA

moment3 = self._central_moment(3)

moment2 = self.var() * (count - 1) / count # Convert sample var to pop var

# See G1 estimator:

# https://en.wikipedia.org/wiki/Skewness#Sample_skewness

numerator = moment3

denominator = moment2 ** (3 / 2)

adjustment = (count * (count - 1)) ** 0.5 / (count - 2)

return (numerator / denominator) * adjustment

def kurt(self):

count = self.count()

if count < 4:

return pandas.NA

moment4 = self._central_moment(4)

moment2 = self.var() * (count - 1) / count # Convert sample var to pop var

# Kurtosis is often defined as the second standardize moment: moment(4)/moment(2)**2

# Pandas however uses Fisher’s estimator, implemented below

numerator = (count + 1) * (count - 1) * moment4

denominator = (count - 2) * (count - 3) * moment2**2

adjustment = 3 * (count - 1) ** 2 / ((count - 2) * (count - 3))

return (numerator / denominator) - adjustment

kurtosis = kurt

def mode(self) -> Series:

block = self._block

# Approach: Count each value, return each value for which count(x) == max(counts))

block, agg_ids = block.aggregate(

by_column_ids=[self._value_column],

aggregations=((self._value_column, agg_ops.count_op),),

)

value_count_col_id = agg_ids[0]

block, max_value_count_col_id = block.apply_window_op(

value_count_col_id,

agg_ops.max_op,

window_spec=bigframes.core.window_spec.WindowSpec(),

)

block, is_mode_col_id = block.apply_binary_op(

value_count_col_id,

max_value_count_col_id,

ops.eq_op,

)

block = block.filter_by_id(is_mode_col_id)

# use temporary name for reset_index to avoid collision, restore after dropping extra columns

block = (

block.with_index_labels(["mode_temp_internal"])

.order_by([order.ascending_over(self._value_column)])

.reset_index(drop=False)

)

block = block.select_column(self._value_column).with_column_labels([self.name])

mode_values_series = Series(block.select_column(self._value_column))

return typing.cast(Series, mode_values_series)

def mean(self) -> float:

return typing.cast(float, self._apply_aggregation(agg_ops.mean_op))

def median(self, *, exact: bool = False) -> float:

if exact:

raise NotImplementedError(

f"Only approximate median is supported. {constants.FEEDBACK_LINK}"

)

return typing.cast(float, self._apply_aggregation(agg_ops.median_op))

def sum(self) -> float:

return typing.cast(float, self._apply_aggregation(agg_ops.sum_op))

def prod(self) -> float:

return typing.cast(float, self._apply_aggregation(agg_ops.product_op))

product = prod

def __eq__(self, other: object) -> Series: # type: ignore

return self.eq(other)

def __ne__(self, other: object) -> Series: # type: ignore

return self.ne(other)

def __invert__(self) -> Series:

return self._apply_unary_op(ops.invert_op)

def eq(self, other: object) -> Series:

# TODO: enforce stricter alignment

return self._apply_binary_op(other, ops.eq_op)

def ne(self, other: object) -> Series:

# TODO: enforce stricter alignment

return self._apply_binary_op(other, ops.ne_op)

def where(self, cond, other=None):

value_id, cond_id, other_id, block = self._align3(cond, other)

block, result_id = block.apply_ternary_op(

value_id, cond_id, other_id, ops.where_op

)

return Series(block.select_column(result_id).with_column_labels([self.name]))

def clip(self, lower, upper):

if lower is None and upper is None:

return self

if lower is None:

return self._apply_binary_op(upper, ops.clipupper_op, alignment="left")

if upper is None:

return self._apply_binary_op(lower, ops.cliplower_op, alignment="left")

value_id, lower_id, upper_id, block = self._align3(lower, upper)

block, result_id = block.apply_ternary_op(

value_id, lower_id, upper_id, ops.clip_op

)

return Series(block.select_column(result_id).with_column_labels([self.name]))

def argmax(self) -> int:

block, row_nums = self._block.promote_offsets()

block = block.order_by(

[

order.descending_over(self._value_column),

order.ascending_over(row_nums),

]

)

return typing.cast(

scalars.Scalar, Series(block.select_column(row_nums)).iloc[0]

)

def argmin(self) -> int:

block, row_nums = self._block.promote_offsets()

block = block.order_by(

[

order.ascending_over(self._value_column),

order.ascending_over(row_nums),

]

)

return typing.cast(

scalars.Scalar, Series(block.select_column(row_nums)).iloc[0]

)

def unstack(self, level: LevelsType = -1):

if isinstance(level, int) or isinstance(level, str):

level = [level]

block = self._block

if self.index.nlevels == 1:

raise ValueError("Series must have multi-index to unstack")

# Pivot by index levels

unstack_ids = self._resolve_levels(level)

block = block.reset_index(drop=False)

block = block.set_index(

[col for col in self._block.index_columns if col not in unstack_ids]

)

pivot_block = block.pivot(

columns=unstack_ids,

values=self._block.value_columns,

values_in_index=False,

)

return bigframes.dataframe.DataFrame(pivot_block)

def idxmax(self) -> blocks.Label:

block = self._block.order_by(

[

order.descending_over(self._value_column),

*[

order.ascending_over(idx_col)

for idx_col in self._block.index_columns

],

]