Skip to content

Distribution Metrics

ACV (All Commodity Volume) metric.

ACV measures total dollar sales across all products in a set of stores, expressed in millions ($MM).

Acv

Calculates ACV (All Commodity Volume) for a set of stores.

ACV represents total dollar sales across all products, expressed in millions ($MM). NaN values in the spend column are excluded from the sum.

Results are accessible via the table attribute (ibis Table) or the df property (materialized pandas DataFrame).

Parameters:

Name Type Description Default
df DataFrame | Table

Transaction data containing at least a unit_spend column.

 required
group_by str | list[str] | None

Optional column(s) to group the ACV calculation by (e.g., store_id). Defaults to None for total ACV.

 None
acv_scale_factor float

Factor to scale the ACV result (default is 1,000,000 for $MM).

 1000000

Raises:

Type Description
TypeError

If df is not a pandas DataFrame or an Ibis Table.
ValueError

If required columns are missing from the data or if acv_scale_factor is not positive.
Source code in pyretailscience/metrics/distribution/acv.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class Acv:
    """Calculates ACV (All Commodity Volume) for a set of stores.

    ACV represents total dollar sales across all products, expressed in millions ($MM).
    NaN values in the spend column are excluded from the sum.

    Results are accessible via the `table` attribute (ibis Table) or the `df` property
    (materialized pandas DataFrame).

    Args:
        df (pd.DataFrame | ibis.Table): Transaction data containing at least a unit_spend column.
        group_by (str | list[str] | None): Optional column(s) to group the ACV calculation by
            (e.g., store_id). Defaults to None for total ACV.
        acv_scale_factor (float): Factor to scale the ACV result (default is 1,000,000 for $MM).

    Raises:
        TypeError: If df is not a pandas DataFrame or an Ibis Table.
        ValueError: If required columns are missing from the data or if acv_scale_factor is not positive.
    """

    def __init__(
        self,
        df: pd.DataFrame | ibis.Table,
        group_by: str | list[str] | None = None,
        acv_scale_factor: float = 1_000_000,
    ) -> None:
        """Initializes the ACV calculation."""
        self._df: pd.DataFrame | None = None
        self.table: ibis.Table

        if acv_scale_factor <= 0:
            raise ValueError("acv_scale_factor must be positive.")

        if isinstance(df, pd.DataFrame):
            df = ibis.memtable(df)
        elif not isinstance(df, ibis.Table):
            raise TypeError("df must be either a pandas DataFrame or an Ibis Table.")

        unit_spend_col = get_option("column.unit_spend")

        if isinstance(group_by, str):
            group_by = [group_by]

        required_cols = [unit_spend_col]
        if group_by is not None:
            required_cols.extend(group_by)
            validate_columns(df, required_cols)
            df = df.group_by(group_by)
        else:
            validate_columns(df, required_cols)

        self.table = df.aggregate(acv=_[unit_spend_col].sum() / acv_scale_factor)

    @property
    def df(self) -> pd.DataFrame:
        """Returns the materialized pandas DataFrame of ACV results.

        Returns:
            pd.DataFrame: DataFrame with ACV values. Cached after first access.
        """
        if self._df is None:
            self._df = self.table.execute()
        return self._df

df: pd.DataFrame property

Returns the materialized pandas DataFrame of ACV results.

Returns:

Type Description
DataFrame

pd.DataFrame: DataFrame with ACV values. Cached after first access.

__init__(df, group_by=None, acv_scale_factor=1000000)

Initializes the ACV calculation.

Source code in pyretailscience/metrics/distribution/acv.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    df: pd.DataFrame | ibis.Table,
    group_by: str | list[str] | None = None,
    acv_scale_factor: float = 1_000_000,
) -> None:
    """Initializes the ACV calculation."""
    self._df: pd.DataFrame | None = None
    self.table: ibis.Table

    if acv_scale_factor <= 0:
        raise ValueError("acv_scale_factor must be positive.")

    if isinstance(df, pd.DataFrame):
        df = ibis.memtable(df)
    elif not isinstance(df, ibis.Table):
        raise TypeError("df must be either a pandas DataFrame or an Ibis Table.")

    unit_spend_col = get_option("column.unit_spend")

    if isinstance(group_by, str):
        group_by = [group_by]

    required_cols = [unit_spend_col]
    if group_by is not None:
        required_cols.extend(group_by)
        validate_columns(df, required_cols)
        df = df.group_by(group_by)
    else:
        validate_columns(df, required_cols)

    self.table = df.aggregate(acv=_[unit_spend_col].sum() / acv_scale_factor)