Skip to content

Composite Rank

Composite ranking across multiple metrics.

Ranks each metric column (ascending or descending per column), then aggregates the per-column ranks — not the raw values — into a composite_rank column (lower is better overall).

CompositeRank

Creates composite rankings by aggregating per-metric ranks.

Ranks are 1-based with rank 1 = best in the given sort direction, so a lower composite_rank is better overall. Tied values share the lowest rank (1, 2, 2, 4). Nulls rank last, except in an "asc" column on BigQuery, which ranks them first (best). composite_rank aggregates the per-column ranks, not the raw values: mean = average rank (a float), sum = total rank, min = best single-metric rank, max = worst single-metric rank. Rows come back in no particular order; sort on composite_rank yourself. .table is a lazy ibis Table; .df materializes on first access and is cached.

Source code in openretailscience/analysis/composite_rank.py 
 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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
class CompositeRank:
    """Creates composite rankings by aggregating per-metric ranks.

    Ranks are 1-based with rank 1 = best in the given sort direction, so a lower
    `composite_rank` is better overall. Tied values share the lowest rank (1, 2, 2, 4). Nulls
    rank last, except in an "asc" column on BigQuery, which ranks them first (best).
    `composite_rank` aggregates the per-column ranks, not the raw values:
    mean = average rank (a float), sum = total rank, min = best single-metric rank, max =
    worst single-metric rank. Rows come back in no particular order; sort on
    `composite_rank` yourself. `.table` is a lazy ibis Table; `.df` materializes on first
    access and is cached.
    """

    def __init__(
        self,
        df: pd.DataFrame | ibis.Table,
        rank_cols: list[tuple[str, str] | str],
        agg_func: str,
        ignore_ties: bool = False,
        group_col: str | list[str] | None = None,
    ) -> None:
        """Initialize a CompositeRank over `df`.

        Args:
            df (pd.DataFrame | ibis.Table): Data to rank.
            rank_cols (list[tuple[str, str] | str]): Metrics to rank with their sort
                direction. Each entry is a column name (defaults to "asc") or a
                (column, sort_order) tuple; sort_order is "asc", "ascending", "desc", or
                "descending" (case-insensitive).
            agg_func (str): How to combine the per-column ranks: "mean", "sum", "min" or "max".
            ignore_ties (bool, optional): If True, every row gets a unique rank and ties are
                broken arbitrarily, so tied rows can swap between runs. Defaults to False.
            group_col (str | list[str], optional): Column(s) to partition the ranking by.
                None (default) ranks globally; otherwise ranks are computed independently
                within each group.

        Raises:
            TypeError: If `df` is not a pandas DataFrame or an Ibis Table.
            ValueError: If `rank_cols` is empty.
            ValueError: If a specified metric is not in the data or the sort order is invalid.
            ValueError: If the aggregation function is not supported.
            ValueError: If `group_col` is specified but doesn't exist in the data.

        Examples:
            >>> # Group-based ranking: each product competes with its own category
            >>> ranker = CompositeRank(
            ...     df=product_data,
            ...     rank_cols=[("weekly_sales", "desc"), ("margin_percentage", "desc"), ("stock_cover_days", "asc")],
            ...     agg_func="mean",
            ...     group_col="product_category",
            ... )
        """
        df = ensure_ibis_table(df)

        if group_col is not None:
            group_col = ensure_columns(df, group_col, "group_col")
        # Validate agg_func up-front so an invalid value fails before any per-column work runs.
        agg_func = ensure_value_choice(agg_func, VALID_AGG_FUNCS, "agg_func")
        rank_mutates = self._process_rank_columns(rank_cols, df, group_col, ignore_ties)
        df = df.mutate(**rank_mutates)
        self.table = self._create_composite_ranking(df, rank_mutates, agg_func)

    def _process_rank_columns(
        self,
        rank_cols: list[tuple[str, str] | str],
        df: ibis.Table,
        group_col: list[str] | None,
        ignore_ties: bool,
    ) -> dict[str, ir.IntegerColumn]:
        """Build an ibis ranking expression for each rank column spec.

        Args:
            rank_cols (list[tuple[str, str] | str]): Column specifications to rank. Each element is
                either a string (column name, defaults to ascending) or a tuple of (column_name, sort_order).
            df (ibis.Table): The table containing the columns to rank.
            group_col (list[str] | None): Columns to partition the ranking window by, or None
                for global ranking. Pre-normalized to a list by the caller.
            ignore_ties (bool): If True, uses row_number for unique ranks. If False, uses rank
                which assigns the same rank to tied values.

        Returns:
            dict[str, ir.IntegerColumn]: Mapping of output column names (`{col_name}_rank`)
                to ibis ranking expressions.

        Raises:
            ValueError: If `rank_cols` is empty.
            ValueError: If a specified column is not found in the DataFrame.
            ValueError: If a sort order is not one of "asc", "ascending", "desc", or "descending".
        """
        if len(rank_cols) == 0:
            msg = "rank_cols must contain at least one column specification"
            raise ValueError(msg)

        rank_mutates = {}

        for col_spec in rank_cols:
            col_name, sort_order = self._parse_column_spec(col_spec)

            ensure_columns(df, col_name, "rank_cols")
            sort_order = ensure_value_choice(sort_order, VALID_SORT_ORDERS, "sort_order")

            order_by = ibis.asc(df[col_name]) if sort_order in ["asc", "ascending"] else ibis.desc(df[col_name])
            window = (
                ibis.window(order_by=order_by)
                if group_col is None
                else ibis.window(group_by=[df[col] for col in group_col], order_by=order_by)
            )

            # Calculate rank based on ignore_ties parameter (using 1-based ranks)
            rank_col = ibis.row_number().over(window) + 1 if ignore_ties else ibis.rank().over(window) + 1
            rank_mutates[f"{col_name}_rank"] = rank_col

        return rank_mutates

    def _parse_column_spec(self, col_spec: tuple[str, str] | str) -> tuple[str, str]:
        """Parse a column specification into a column name and sort order.

        Args:
            col_spec (tuple[str, str] | str): Either a column name string (defaults to "asc")
                or a tuple of (column_name, sort_order).

        Returns:
            tuple[str, str]: A tuple of (column_name, sort_order).

        Raises:
            ValueError: If a tuple is provided but does not contain exactly two elements.
        """
        if isinstance(col_spec, str):
            return col_spec, "asc"
        if len(col_spec) != 2:  # noqa: PLR2004 - Error message below explains the value
            msg = f"Column specification must be a string or a tuple of (column_name, sort_order). Got {col_spec}"
            raise ValueError(msg)
        return col_spec

    def _create_composite_ranking(
        self,
        df: ibis.Table,
        rank_mutates: dict[str, ir.IntegerColumn],
        agg_func: str,
    ) -> ibis.Table:
        """Aggregate per-metric rank columns into a `composite_rank` column.

        Args:
            df (ibis.Table): The table with individual rank columns already added.
            rank_mutates (dict[str, ir.IntegerColumn]): Mapping of rank column names to their expressions,
                used to identify which columns to aggregate.
            agg_func (str): Aggregation function to combine ranks. Must be one of
                "mean", "sum", "min", or "max".

        Returns:
            ibis.Table: The input table with an additional composite_rank column.
        """
        column_refs = [df[col] for col in rank_mutates]
        agg_expr = {
            "mean": sum(column_refs) / len(column_refs),
            "sum": sum(column_refs),
            "min": ibis.least(*column_refs),
            "max": ibis.greatest(*column_refs),
        }
        return df.mutate(composite_rank=agg_expr[agg_func])

    @functools.cached_property
    def df(self) -> pd.DataFrame:
        """Materialized ranked data (computed lazily on first access, then cached).

        Returns:
            pd.DataFrame: The input metrics plus one `{col}_rank` column per ranked
                metric and the final `composite_rank` column.
        """
        return self.table.execute()

df cached property

Materialized ranked data (computed lazily on first access, then cached).

Returns:

Type Description
DataFrame

pd.DataFrame: The input metrics plus one {col}_rank column per ranked metric and the final composite_rank column.

__init__(df, rank_cols, agg_func, ignore_ties=False, group_col=None)

Initialize a CompositeRank over df.

Parameters:

Name Type Description Default
df DataFrame | Table

Data to rank.

 required
rank_cols list[tuple[str, str] | str]

Metrics to rank with their sort direction. Each entry is a column name (defaults to "asc") or a (column, sort_order) tuple; sort_order is "asc", "ascending", "desc", or "descending" (case-insensitive).

 required
agg_func str

How to combine the per-column ranks: "mean", "sum", "min" or "max".

 required
ignore_ties bool

If True, every row gets a unique rank and ties are broken arbitrarily, so tied rows can swap between runs. Defaults to False.

 False
group_col str | list[str]

Column(s) to partition the ranking by. None (default) ranks globally; otherwise ranks are computed independently within each group.

 None

Raises:

Type Description
TypeError

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

If rank_cols is empty.
ValueError

If a specified metric is not in the data or the sort order is invalid.
ValueError

If the aggregation function is not supported.
ValueError

If group_col is specified but doesn't exist in the data.

Examples:

>>> # Group-based ranking: each product competes with its own category
>>> ranker = CompositeRank(
...     df=product_data,
...     rank_cols=[("weekly_sales", "desc"), ("margin_percentage", "desc"), ("stock_cover_days", "asc")],
...     agg_func="mean",
...     group_col="product_category",
... )
Source code in openretailscience/analysis/composite_rank.py 
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
def __init__(
    self,
    df: pd.DataFrame | ibis.Table,
    rank_cols: list[tuple[str, str] | str],
    agg_func: str,
    ignore_ties: bool = False,
    group_col: str | list[str] | None = None,
) -> None:
    """Initialize a CompositeRank over `df`.

    Args:
        df (pd.DataFrame | ibis.Table): Data to rank.
        rank_cols (list[tuple[str, str] | str]): Metrics to rank with their sort
            direction. Each entry is a column name (defaults to "asc") or a
            (column, sort_order) tuple; sort_order is "asc", "ascending", "desc", or
            "descending" (case-insensitive).
        agg_func (str): How to combine the per-column ranks: "mean", "sum", "min" or "max".
        ignore_ties (bool, optional): If True, every row gets a unique rank and ties are
            broken arbitrarily, so tied rows can swap between runs. Defaults to False.
        group_col (str | list[str], optional): Column(s) to partition the ranking by.
            None (default) ranks globally; otherwise ranks are computed independently
            within each group.

    Raises:
        TypeError: If `df` is not a pandas DataFrame or an Ibis Table.
        ValueError: If `rank_cols` is empty.
        ValueError: If a specified metric is not in the data or the sort order is invalid.
        ValueError: If the aggregation function is not supported.
        ValueError: If `group_col` is specified but doesn't exist in the data.

    Examples:
        >>> # Group-based ranking: each product competes with its own category
        >>> ranker = CompositeRank(
        ...     df=product_data,
        ...     rank_cols=[("weekly_sales", "desc"), ("margin_percentage", "desc"), ("stock_cover_days", "asc")],
        ...     agg_func="mean",
        ...     group_col="product_category",
        ... )
    """
    df = ensure_ibis_table(df)

    if group_col is not None:
        group_col = ensure_columns(df, group_col, "group_col")
    # Validate agg_func up-front so an invalid value fails before any per-column work runs.
    agg_func = ensure_value_choice(agg_func, VALID_AGG_FUNCS, "agg_func")
    rank_mutates = self._process_rank_columns(rank_cols, df, group_col, ignore_ties)
    df = df.mutate(**rank_mutates)
    self.table = self._create_composite_ranking(df, rank_mutates, agg_func)