API Reference

This page contains the complete API reference for Qanta, automatically generated from the source code.

qanta

Qanta - A Python library for analysts and quants.

binned_means

binned_means(df: DataFrame, bins: Mapping[str, int | Collection[float]], *, quantile: bool = True, weights: str | None = None, output_weights: str | None = None, compute_std_errors: bool = False) -> pd.DataFrame

Compute n-dimensional binned means.

Splits observations (rows) into n-dimensional hyper-rectangles using the specified binning strategy, then computes the weighted mean for each bin.

Rows with NaN values in any binned column are excluded. Bins that contain no observations are dropped from the output, so the result may have fewer rows than the total number of bins requested.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame containing the data.	required
bins	Mapping[str, int | Collection[float]]	Dictionary mapping column names to either: - An integer specifying the number of bins. - A sequence of values whose meaning depends on quantile: - quantile=False: explicit bin edges in data units (e.g. [0, 3, 7, 11]). - quantile=True: quantile break-points between 0 and 1 (e.g. [0.0, 0.5, 1.0] for a median split).	required
quantile	bool	If True (default), use quantile-based binning (equal-count bins). If False, use equal-width binning.	True
weights	str | None	Column name to use as observation weights. If None, all observations are weighted equally. Weights must be finite and positive.	None
output_weights	str | None	If provided, include the sum of weights per bin in the output under this column name. Must not collide with any other output column name.	None
compute_std_errors	bool	If True, include standard errors of the means as additional columns with '_se' suffix. Returns NaN for bins that contain a single observation.	False

Returns:

Type	Description
DataFrame	DataFrame with one row per non-empty bin, containing: - Mean values for each column specified in bins - Sum of weights per bin (if output_weights is provided) - Standard errors (if compute_std_errors=True)

Raises:

Type	Description
ValueError	If weights contains NaN, inf, zero, or negative values.
ValueError	If output_weights collides with a bin column or a standard-error column name.

Examples:

import pandas as pd

df = pd.DataFrame(
    {
        'x': [0, 1, 7, 8, 9, 10],
        'y': [0, 1, 7, 1, 3, 9],
        'w': [1, 3, 1, 1, 2, 1],
    }
)

result = binned_means(
    df,
    bins={'x': 2, 'y': 1},
    weights='w',
    output_weights='w',
)
#      x    y    w
# 0  2.0  2.0  5.0
# 1  9.0  4.0  4.0

Source code in src/qanta/binning.py

def binned_means(
    df: pd.DataFrame,
    bins: Mapping[str, int | Collection[float]],
    *,
    quantile: bool = True,
    weights: str | None = None,
    output_weights: str | None = None,
    compute_std_errors: bool = False,
) -> pd.DataFrame:
    """Compute n-dimensional binned means.

    Splits observations (rows) into n-dimensional hyper-rectangles using the
    specified binning strategy, then computes the weighted mean for each
    bin.

    Rows with NaN values in any binned column are excluded. Bins that
    contain no observations are dropped from the output, so the result
    may have fewer rows than the total number of bins requested.

    Args:
        df: Input DataFrame containing the data.
        bins: Dictionary mapping column names to either:
            - An integer specifying the number of bins.
            - A sequence of values whose meaning depends on ``quantile``:
              - ``quantile=False``: explicit bin edges in data units
                (e.g. ``[0, 3, 7, 11]``).
              - ``quantile=True``: quantile break-points between 0 and 1
                (e.g. ``[0.0, 0.5, 1.0]`` for a median split).
        quantile: If True (default), use quantile-based binning
            (equal-count bins). If False, use equal-width binning.
        weights: Column name to use as observation weights. If None, all
            observations are weighted equally. Weights must be finite
            and positive.
        output_weights: If provided, include the **sum** of weights per
            bin in the output under this column name. Must not collide
            with any other output column name.
        compute_std_errors: If True, include standard errors of the
            means as additional columns with ``'_se'`` suffix. Returns
            NaN for bins that contain a single observation.

    Returns:
        DataFrame with one row per non-empty bin, containing:
            - Mean values for each column specified in ``bins``
            - Sum of weights per bin (if ``output_weights`` is provided)
            - Standard errors (if ``compute_std_errors=True``)

    Raises:
        ValueError: If ``weights`` contains NaN, inf, zero, or negative
            values.
        ValueError: If ``output_weights`` collides with a bin column or
            a standard-error column name.

    Examples:
        ```python
        import pandas as pd

        df = pd.DataFrame(
            {
                'x': [0, 1, 7, 8, 9, 10],
                'y': [0, 1, 7, 1, 3, 9],
                'w': [1, 3, 1, 1, 2, 1],
            }
        )

        result = binned_means(
            df,
            bins={'x': 2, 'y': 1},
            weights='w',
            output_weights='w',
        )
        #      x    y    w
        # 0  2.0  2.0  5.0
        # 1  9.0  4.0  4.0
        ```
    """
    # Sort columns for deterministic output
    column_bins = sorted(bins.items())
    columns = [col for col, _ in column_bins]

    # Validate output_weights doesn't collide with other output columns
    if output_weights is not None:
        reserved = set(columns)
        if compute_std_errors:
            reserved |= {f'{col}_se' for col in columns}
        if output_weights in reserved:
            raise ValueError(
                f"output_weights='{output_weights}' collides with an output column name"
            )

    # Handle empty input
    if df.empty:
        result_columns = (
            columns
            + ([output_weights] if output_weights else [])
            + ([f'{col}_se' for col in columns] if compute_std_errors else [])
        )
        return pd.DataFrame(columns=result_columns)

    # Extract data as numpy arrays
    values = df[columns].to_numpy(dtype=np.float64)
    weight_values = (
        df[weights].to_numpy(dtype=np.float64) if weights is not None else None
    )

    # Validate weights
    if weight_values is not None:
        if not np.all(np.isfinite(weight_values)):
            raise ValueError('Weights must be finite (no NaN or inf values)')
        if np.any(weight_values <= 0):
            raise ValueError('Weights must be positive')

    # Compute bin assignments for each column
    categories = []
    for i, (_, bin_spec) in enumerate(column_bins):
        column_values = values[:, i]
        if quantile:
            category = _create_quantile_bins(column_values, bin_spec, weight_values)
        else:
            category = _create_equal_width_bins(column_values, bin_spec)
        categories.append(category)

    # Group observations into bins
    filtered_values, filtered_weights, bin_codes = _group_by(
        values, weight_values, categories
    )

    # Compute weighted means for each bin
    means, standard_errors, agg_weights = _compute_means(
        filtered_values, filtered_weights, bin_codes, compute_std_errors
    )

    # Build result DataFrame
    result = pd.DataFrame(means, columns=columns)

    if output_weights is not None:
        result[output_weights] = agg_weights

    if compute_std_errors:
        standard_errors_df = pd.DataFrame(
            standard_errors, columns=[f'{col}_se' for col in columns]
        )
        result = pd.concat([result, standard_errors_df], axis=1)

    return result

weighted_quantile

weighted_quantile(values: Collection[float], weights: Collection[float], quantiles: Collection[float]) -> list[float]

Compute weighted quantiles using staircase interpolation.

Each observation is given a flat plateau in the empirical CDF proportional to its weight. Quantiles that fall within a single observation's plateau return that observation's exact value; quantiles that fall between plateaus are linearly interpolated.

Non-finite values (NaN, inf) in values are silently excluded. If no finite values remain, returns a list of NaN.

Parameters:

Name	Type	Description	Default
values	Collection[float]	Data values.	required
weights	Collection[float]	Weights for each value. Must be positive. Non-integer weights are internally normalized to integers for the staircase construction.	required
quantiles	Collection[float]	Quantiles to compute (values between 0 and 1).	required

Returns:

Type	Description
list[float]	List of percentile values corresponding to the requested
list[float]	quantiles, in the same order.

Examples:

Heavy weight on the last value pulls the median to 3.0:

>>> weighted_quantile([1, 2, 3], [1, 1, 10], [0.5])
[3.0]

With equal weights, q=0.25 falls between the plateaus of 1 and 2 and is linearly interpolated:

>>> weighted_quantile([1, 2, 3], [1, 1, 1], [0.25])
[1.5]

Source code in src/qanta/binning.py

def weighted_quantile(
    values: Collection[float],
    weights: Collection[float],
    quantiles: Collection[float],
) -> list[float]:
    """Compute weighted quantiles using staircase interpolation.

    Each observation is given a flat plateau in the empirical CDF
    proportional to its weight. Quantiles that fall within a single
    observation's plateau return that observation's exact value;
    quantiles that fall between plateaus are linearly interpolated.

    Non-finite values (NaN, inf) in ``values`` are silently excluded.
    If no finite values remain, returns a list of NaN.

    Args:
        values: Data values.
        weights: Weights for each value. Must be positive. Non-integer
            weights are internally normalized to integers for the
            staircase construction.
        quantiles: Quantiles to compute (values between 0 and 1).

    Returns:
        List of percentile values corresponding to the requested
        quantiles, in the same order.

    Examples:
        Heavy weight on the last value pulls the median to 3.0:

        ```python
        >>> weighted_quantile([1, 2, 3], [1, 1, 10], [0.5])
        [3.0]
        ```

        With equal weights, q=0.25 falls between the plateaus of 1
        and 2 and is linearly interpolated:

        ```python
        >>> weighted_quantile([1, 2, 3], [1, 1, 1], [0.25])
        [1.5]
        ```
    """
    values_array = np.asarray(values, dtype=np.float64)
    weight_array = np.asarray(weights, dtype=np.float64)
    quantile_points = np.asarray(quantiles, dtype=np.float64)

    # Filter out non-finite values
    valid_mask = np.isfinite(values_array)
    values_array = values_array[valid_mask]
    weight_array = weight_array[valid_mask]

    if len(values_array) == 0:
        return [np.nan] * len(quantile_points)

    # Sort by values
    sort_indices = np.argsort(values_array)
    values_array = values_array[sort_indices]
    weight_array = weight_array[sort_indices]

    # Normalize weights if they're not integers
    if np.any(weight_array.astype(np.int64) != weight_array):
        weight_array = weight_array / weight_array.max()
        weight_array = np.clip(weight_array, 1e-6, 1.0)
        weight_array = weight_array / weight_array.min()
        weight_array = (weight_array * 1000).astype(np.int64)

    # Build cumulative distribution for interpolation
    cumulative_weights = np.cumsum(weight_array)
    x_coords = (
        np.concatenate([[0], cumulative_weights[:-1], cumulative_weights - 1])
        .reshape(2, -1)
        .T.ravel()
    )
    y_coords = np.repeat(values_array, 2)

    # Interpolate to find percentile values
    interpolated: FloatArray = np.interp(
        quantile_points * x_coords[-1], x_coords, y_coords
    )
    return [float(val) for val in interpolated]