Stats Bincount

bincount

CLASS	DESCRIPTION
RunningBincount	Running histogram for integer-valued data with optional sample weights.

Classes

RunningBincount(state: dict | None = None)

Bases: Stats

Running histogram for integer-valued data with optional sample weights.

Maintains a count (or weighted sum) of occurrences for each unique integer value, incrementally updated across batches. Useful for computing histograms of categorical or discrete data without storing all samples. Supports weighted samples for robust statistics and importance sampling.

PARAMETER	DESCRIPTION
state	Previous state to restore from. TYPE: dict | None, optional, by default None DEFAULT: None

Properties

count : int Total number of elements processed.

bin_count : torch.Tensor Histogram counts (or weighted sums) for each integer value (length = max_value + 1).

Notes

Input tensors must contain non-negative integers. The bin_count tensor automatically expands to accommodate larger values as they are encountered.

When weights are provided, each bin accumulates the sum of weights for samples with that value instead of just counting occurrences.

Examples:

Basic usage with categorical data:

>>> from spectre.stats import RunningBincount
>>> import torch
>>>
>>> stat = RunningBincount()
>>> for _ in range(10):
...     batch = torch.randint(0, 5, (100,))  # Categories 0-4
...     stat.add(batch)
>>>
>>> histogram = stat.bin_count()
>>> histogram.shape
torch.Size([5])

Weighted histogram:

>>> stat = RunningBincount()
>>> labels = torch.tensor([0, 1, 1, 2, 2, 2, 3])
>>> weights = torch.tensor([1.0, 2.0, 3.0, 0.5, 1.5, 2.0, 1.0])
>>> stat.add(labels, weights=weights)
>>> weighted_counts = stat.bin_count()
>>> # weighted_counts[0] = 1.0, weighted_counts[1] = 5.0, etc.

Counting class labels:

>>> stat = RunningBincount()
>>> labels = torch.tensor([0, 1, 1, 2, 2, 2, 3])
>>> stat.add(labels)
>>> counts = stat.bin_count()
>>> counts  # [1, 2, 3, 1] - counts for classes 0, 1, 2, 3

METHOD	DESCRIPTION
add	Add batch of integer values to histogram with optional sample weights.
to	Move internal tensors to specified device.
state_dict	Save state to dictionary.
load_state_dict	Load state from dictionary.

ATTRIBUTE	DESCRIPTION
count	Get total number of elements processed. TYPE: int
bin_count	Get current histogram counts. TYPE: Tensor

Source code in spectre/stats/bincount.py

def __init__(self, state: dict | None = None):
    if state is not None:
        return super().__init__(state=state)

    self._count = 0
    self._bin_count = None

Attributes

count: int property

Get total number of elements processed.

bin_count: torch.Tensor property

Get current histogram counts.

Functions

add(X: torch.Tensor, weights: torch.Tensor | None = None, count: int | None = None) -> None

Add batch of integer values to histogram with optional sample weights.

Automatically expands bin_count tensor to accommodate new larger values. When weights are provided, each bin accumulates the sum of weights for samples with that value instead of just counting occurrences.

PARAMETER	DESCRIPTION
X	Integer tensor with non-negative values. Automatically flattened. TYPE: Tensor
weights	Sample weights with same shape as X. If None, all samples have weight 1.0 (equivalent to standard counting). By default None. TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
count	Explicit size to add to count (overrides X.numel()). TYPE: int | None, optional, by default None DEFAULT: None

Source code in spectre/stats/bincount.py

def add(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    count: int | None = None,
) -> None:  # ty: ignore
    """
    Add batch of integer values to histogram with optional sample weights.

    Automatically expands bin_count tensor to accommodate new larger values.
    When weights are provided, each bin accumulates the sum of weights for
    samples with that value instead of just counting occurrences.

    Parameters
    ----------
    X : torch.Tensor
        Integer tensor with non-negative values. Automatically flattened.

    weights : torch.Tensor | None, optional, by default None
        Sample weights with same shape as X. If None, all samples have
        weight 1.0 (equivalent to standard counting). By default None.

    count : int | None, optional, by default None
        Explicit size to add to count (overrides X.numel()).
    """
    X = X.view(-1)

    if weights is not None:
        weights = weights.view(-1)
        check_same_len(X, weights)

        # Use weighted bincount: sum weights for each bin
        max_val = int(X.max().item()) if len(X) > 0 else 0
        w_bin_count = torch.zeros(max_val + 1, dtype=weights.dtype, device=X.device)
        w_bin_count.scatter_add_(0, X, weights)
    else:
        # Standard bincount: count occurrences
        w_bin_count = X.bincount()

    if self._bin_count is None:
        self._bin_count = w_bin_count
    elif len(self._bin_count) < len(w_bin_count):
        # Expand existing bins and add new ones
        expanded_count = torch.zeros_like(w_bin_count)
        expanded_count[: len(self._bin_count)] = self._bin_count
        self._bin_count = expanded_count + w_bin_count
    else:
        # Add to existing bins
        self._bin_count[: len(w_bin_count)] += w_bin_count

    if count is None:
        self._count += len(X)
    else:
        self._count += count

to(device: str = 'cpu') -> None

Move internal tensors to specified device.

Source code in spectre/stats/bincount.py

def to(self, device: str = "cpu") -> None:
    """Move internal tensors to specified device."""
    if self._bin_count is not None:
        self._bin_count = self._bin_count.to(device)

state_dict() -> dict

Save state to dictionary.

Source code in spectre/stats/bincount.py

def state_dict(self) -> dict:
    """Save state to dictionary."""
    return dict(
        constructor=f"{self.__module__}.{self.__class__.__name__}()",
        count=self._count,
        bin_count=self._bin_count,
    )

load_state_dict(state: dict) -> None

Load state from dictionary.

Source code in spectre/stats/bincount.py

def load_state_dict(self, state: dict) -> None:
    """Load state from dictionary."""
    self._count = int(state["count"])
    self._bin_count = state["bin_count"]

Functions