Stats History

history

CLASS	DESCRIPTION
RunningHistory	Accumulates complete history of all data batches.

Classes

RunningHistory(data: torch.Tensor | None = None, state: dict | None = None)

Bases: Stats

Accumulates complete history of all data batches.

Concatenates all added batches into a single tensor. Unlike other running statistics, this stores all data in memory. Useful for collecting data across batches without manual concatenation logic.

PARAMETER	DESCRIPTION
data	Initial data tensor. TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
state	Previous state to restore from. TYPE: dict | None, optional, by default None DEFAULT: None

Properties

data : torch.Tensor | None Concatenated data from all batches.

Notes

Data is concatenated in chunks of 100 batches for efficiency. Use history() to retrieve the complete concatenated tensor.

Warnings

This statistic stores ALL data in memory. For large datasets, consider using other running statistics that maintain only summary information.

Examples:

Basic usage:

>>> from spectre.stats import RunningHistory
>>> import torch
>>>
>>> stat = RunningHistory()
>>> for i in range(5):
...     batch = torch.randn(10, 3)
...     stat.add(batch)
>>>
>>> all_data = stat.history()
>>> all_data.shape
torch.Size([50, 3])

Collecting predictions across batches:

>>> stat = RunningHistory()
>>> model = MyModel()
>>> for batch in dataloader:
...     predictions = model(batch)
...     stat.add(predictions)
>>>
>>> all_predictions = stat.history()

METHOD	DESCRIPTION
add	Add batch to history.
history	Get complete concatenated history.
load_state_dict	Load state from dictionary.
state_dict	Save state to dictionary.
to_	Move internal tensors to specified device.

Source code in spectre/stats/history.py

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

    self._data = data
    self._added = []

Functions

add(X: torch.Tensor, weights: torch.Tensor | None = None, *args, **kwargs) -> None

Add batch to history.

Batches are buffered and concatenated periodically for efficiency.

PARAMETER	DESCRIPTION
X	Data batch to add to history. TYPE: Tensor

Source code in spectre/stats/history.py

def add(
    self, X: torch.Tensor, weights: torch.Tensor | None = None, *args, **kwargs
) -> None:
    """
    Add batch to history.

    Batches are buffered and concatenated periodically for efficiency.

    Parameters
    ----------
    X : torch.Tensor
        Data batch to add to history.
    """
    self._added.append(X)
    if len(self._added) > 100:
        self._cat_added()

history() -> torch.Tensor

Get complete concatenated history.

Triggers final concatenation of any buffered batches.

RETURNS	DESCRIPTION
Tensor	All data concatenated along first dimension.

Source code in spectre/stats/history.py

def history(self) -> torch.Tensor:
    """
    Get complete concatenated history.

    Triggers final concatenation of any buffered batches.

    Returns
    -------
    torch.Tensor
        All data concatenated along first dimension.
    """
    self._cat_added()
    return self._data

load_state_dict(state: dict) -> None

Load state from dictionary.

PARAMETER	DESCRIPTION
state	State dictionary from state_dict() containing: "data": Concatenated history tensor (or None) TYPE: dict

Source code in spectre/stats/history.py

def load_state_dict(self, state: dict) -> None:
    """
    Load state from dictionary.

    Parameters
    ----------
    state : dict
        State dictionary from `state_dict()` containing:

        - "data": Concatenated history tensor (or None)
    """
    data = state["data"]
    self._data = None if data is None else torch.from_numpy(data)
    self._added = []

state_dict() -> dict

Save state to dictionary.

Triggers final concatenation before saving.

RETURNS	DESCRIPTION
dict	State dictionary containing concatenated history.

Source code in spectre/stats/history.py

def state_dict(self) -> dict:
    """
    Save state to dictionary.

    Triggers final concatenation before saving.

    Returns
    -------
    dict
        State dictionary containing concatenated history.
    """
    self._cat_added()
    return dict(
        constructor=self.__module__ + "." + self.__class__.__name__ + "()",
        data=None if self._data is None else self._data.cpu().numpy(),
    )

to_(device: str) -> None

Move internal tensors to specified device.

Triggers final concatenation before moving.

PARAMETER	DESCRIPTION
device	Target device ("cpu", "cuda", "cuda:0", etc.). TYPE: str

Source code in spectre/stats/history.py

def to_(self, device: str) -> None:
    """
    Move internal tensors to specified device.

    Triggers final concatenation before moving.

    Parameters
    ----------
    device : str
        Target device ("cpu", "cuda", "cuda:0", etc.).
    """
    self._cat_added()
    if self._data is not None:
        self._data = self._data.to(device)