tsam.api

tsam.api

New simplified API for tsam aggregation.

aggregate

aggregate(
    data: DataFrame,
    n_clusters: int,
    *,
    period_duration: int | float | str = 24,
    temporal_resolution: float | str | None = None,
    cluster: ClusterConfig | None = None,
    segments: SegmentConfig | None = None,
    extremes: ExtremeConfig | None = None,
    weights: dict[str, float] | None = None,
    preserve_column_means: bool = True,
    rescale_exclude_columns: list[str] | None = None,
    round_decimals: int | None = None,
    numerical_tolerance: float = 1e-13,
) -> AggregationResult

Aggregate time series data into typical periods.

This function reduces a time series dataset to a smaller set of representative "typical periods" using clustering algorithms.

Parameters:

Name	Type	Description	Default
data	DataFrame	Input time series data with a datetime index. Each column represents a different variable (e.g., solar, wind, demand). The index should be a DatetimeIndex with regular intervals.	required
n_clusters	int	Number of clusters (typical periods) to create. Higher values = more accuracy but less data reduction. Typical range: 4-20 for energy system models.	required
period_duration	int, float, or str	Length of each period. Accepts: - int/float: hours (e.g., 24 for daily, 168 for weekly) - str: pandas Timedelta string (e.g., '24h', '1d', '1w')	24
temporal_resolution	float or str	Time resolution of input data. Accepts: - float: hours (e.g., 1.0 for hourly, 0.25 for 15-minute) - str: pandas Timedelta string (e.g., '1h', '15min', '30min') If not provided, inferred from the datetime index.	None
cluster	ClusterConfig	Clustering configuration. If not provided, uses defaults: - method: "hierarchical" - representation: "medoid"	None
segments	SegmentConfig	Segmentation configuration for reducing temporal resolution within periods. If not provided, no segmentation is applied.	None
extremes	ExtremeConfig	Configuration for preserving extreme periods. If not provided, no extreme period handling is applied.	None
weights	dict[str, float]	Per-column weights that influence all pipeline stages (clustering, segmentation, representation, rescaling). Higher weight = more influence on distance calculations. Example: {"demand": 2.0, "solar": 1.0}	None
preserve_column_means	bool	Rescale typical periods so each column's weighted mean matches the original data's mean. Ensures total energy/load is preserved when weights represent occurrence counts.	True
rescale_exclude_columns	list[str]	Column names to exclude from rescaling when preserve_column_means=True. Useful for binary/indicator columns (0/1 values) that should not be rescaled. If None (default), all columns are rescaled.	None
round_decimals	int	Round output values to this many decimal places. If not provided, no rounding is applied.	None
numerical_tolerance	float	Tolerance for numerical precision issues. Controls when warnings are raised for aggregated values exceeding the original time series bounds. Increase this value to silence warnings caused by floating-point precision errors.	1e-13

Returns:

Type	Description
AggregationResult	Object containing: - cluster_representatives: DataFrame with aggregated periods - cluster_assignments: Which cluster each original period belongs to - cluster_weights: Occurrence count per cluster - accuracy: RMSE, MAE metrics - Methods: to_dict()

Raises:

Type	Description
ValueError	If input data is invalid or parameters are inconsistent.
TypeError	If parameter types are incorrect.

Examples:

Basic usage with defaults:

>>> import tsam
>>> result = tsam.aggregate(df, n_clusters=8)
>>> typical = result.cluster_representatives

With custom clustering:

>>> from tsam import aggregate, ClusterConfig
>>> result = aggregate(
...     df,
...     n_clusters=8,
...     cluster=ClusterConfig(method="kmeans", representation="mean"),
... )

With segmentation (reduce to 12 timesteps per period):

>>> from tsam import aggregate, SegmentConfig
>>> result = aggregate(
...     df,
...     n_clusters=8,
...     segments=SegmentConfig(n_segments=12),
... )

Preserving peak demand periods:

>>> from tsam import aggregate, ExtremeConfig
>>> result = aggregate(
...     df,
...     n_clusters=8,
...     extremes=ExtremeConfig(max_value=["demand"]),
... )

Transferring assignments to new data:

>>> result1 = aggregate(df_wind, n_clusters=8)
>>> result2 = result1.clustering.apply(df_all)

See Also

ClusterConfig : Clustering algorithm configuration SegmentConfig : Temporal segmentation configuration ExtremeConfig : Extreme period preservation configuration AggregationResult : Result object with all outputs

Source code in src/tsam/api.py

def aggregate(
    data: pd.DataFrame,
    n_clusters: int,
    *,
    period_duration: int | float | str = 24,
    temporal_resolution: float | str | None = None,
    cluster: ClusterConfig | None = None,
    segments: SegmentConfig | None = None,
    extremes: ExtremeConfig | None = None,
    weights: dict[str, float] | None = None,
    preserve_column_means: bool = True,
    rescale_exclude_columns: list[str] | None = None,
    round_decimals: int | None = None,
    numerical_tolerance: float = 1e-13,
) -> AggregationResult:
    """Aggregate time series data into typical periods.

    This function reduces a time series dataset to a smaller set of
    representative "typical periods" using clustering algorithms.

    Parameters
    ----------
    data : pd.DataFrame
        Input time series data with a datetime index.
        Each column represents a different variable (e.g., solar, wind, demand).
        The index should be a DatetimeIndex with regular intervals.

    n_clusters : int
        Number of clusters (typical periods) to create.
        Higher values = more accuracy but less data reduction.
        Typical range: 4-20 for energy system models.

    period_duration : int, float, or str, default 24
        Length of each period. Accepts:
        - int/float: hours (e.g., 24 for daily, 168 for weekly)
        - str: pandas Timedelta string (e.g., '24h', '1d', '1w')

    temporal_resolution : float or str, optional
        Time resolution of input data. Accepts:
        - float: hours (e.g., 1.0 for hourly, 0.25 for 15-minute)
        - str: pandas Timedelta string (e.g., '1h', '15min', '30min')
        If not provided, inferred from the datetime index.

    cluster : ClusterConfig, optional
        Clustering configuration. If not provided, uses defaults:
        - method: "hierarchical"
        - representation: "medoid"

    segments : SegmentConfig, optional
        Segmentation configuration for reducing temporal resolution
        within periods. If not provided, no segmentation is applied.

    extremes : ExtremeConfig, optional
        Configuration for preserving extreme periods.
        If not provided, no extreme period handling is applied.

    weights : dict[str, float], optional
        Per-column weights that influence all pipeline stages
        (clustering, segmentation, representation, rescaling).
        Higher weight = more influence on distance calculations.
        Example: {"demand": 2.0, "solar": 1.0}

    preserve_column_means : bool, default True
        Rescale typical periods so each column's weighted mean matches
        the original data's mean. Ensures total energy/load is preserved
        when weights represent occurrence counts.

    rescale_exclude_columns : list[str], optional
        Column names to exclude from rescaling when preserve_column_means=True.
        Useful for binary/indicator columns (0/1 values) that should not be
        rescaled. If None (default), all columns are rescaled.

    round_decimals : int, optional
        Round output values to this many decimal places.
        If not provided, no rounding is applied.

    numerical_tolerance : float, default 1e-13
        Tolerance for numerical precision issues.
        Controls when warnings are raised for aggregated values exceeding
        the original time series bounds. Increase this value to silence
        warnings caused by floating-point precision errors.

    Returns
    -------
    AggregationResult
        Object containing:
        - cluster_representatives: DataFrame with aggregated periods
        - cluster_assignments: Which cluster each original period belongs to
        - cluster_weights: Occurrence count per cluster
        - accuracy: RMSE, MAE metrics
        - Methods: to_dict()

    Raises
    ------
    ValueError
        If input data is invalid or parameters are inconsistent.
    TypeError
        If parameter types are incorrect.

    Examples
    --------
    Basic usage with defaults:

    >>> import tsam
    >>> result = tsam.aggregate(df, n_clusters=8)
    >>> typical = result.cluster_representatives

    With custom clustering:

    >>> from tsam import aggregate, ClusterConfig
    >>> result = aggregate(
    ...     df,
    ...     n_clusters=8,
    ...     cluster=ClusterConfig(method="kmeans", representation="mean"),
    ... )

    With segmentation (reduce to 12 timesteps per period):

    >>> from tsam import aggregate, SegmentConfig
    >>> result = aggregate(
    ...     df,
    ...     n_clusters=8,
    ...     segments=SegmentConfig(n_segments=12),
    ... )

    Preserving peak demand periods:

    >>> from tsam import aggregate, ExtremeConfig
    >>> result = aggregate(
    ...     df,
    ...     n_clusters=8,
    ...     extremes=ExtremeConfig(max_value=["demand"]),
    ... )

    Transferring assignments to new data:

    >>> result1 = aggregate(df_wind, n_clusters=8)
    >>> result2 = result1.clustering.apply(df_all)

    See Also
    --------
    ClusterConfig : Clustering algorithm configuration
    SegmentConfig : Temporal segmentation configuration
    ExtremeConfig : Extreme period preservation configuration
    AggregationResult : Result object with all outputs
    """
    # Validate input
    if not isinstance(data, pd.DataFrame):
        raise TypeError(f"data must be a pandas DataFrame, got {type(data).__name__}")

    _warn_column_order_change(data)

    if not isinstance(n_clusters, int) or n_clusters < 1:
        raise ValueError(f"n_clusters must be a positive integer, got {n_clusters}")

    # Parse duration parameters to hours
    period_duration = _parse_duration_hours(period_duration, "period_duration")
    if period_duration <= 0:
        raise ValueError(f"period_duration must be positive, got {period_duration}")

    temporal_resolution = (
        _parse_duration_hours(temporal_resolution, "temporal_resolution")
        if temporal_resolution is not None
        else None
    )
    if temporal_resolution is not None and temporal_resolution <= 0:
        raise ValueError(
            f"temporal_resolution must be positive, got {temporal_resolution}"
        )

    # Apply defaults
    if cluster is None:
        cluster = ClusterConfig()

    # Validate segments against data
    if segments is not None:
        # Calculate timesteps per period
        if temporal_resolution is not None:
            timesteps_per_period = int(period_duration / temporal_resolution)
        else:
            # Infer resolution from data index
            if isinstance(data.index, pd.DatetimeIndex) and len(data.index) > 1:
                inferred_resolution = (
                    data.index[1] - data.index[0]
                ).total_seconds() / 3600
                timesteps_per_period = int(period_duration / inferred_resolution)
            else:
                # Fall back to assuming hourly resolution
                timesteps_per_period = int(period_duration)

        if segments.n_segments > timesteps_per_period:
            raise ValueError(
                f"n_segments ({segments.n_segments}) cannot exceed "
                f"timesteps per period ({timesteps_per_period})"
            )

    # Validate extreme columns exist in data
    if extremes is not None:
        all_extreme_cols = (
            extremes.max_value
            + extremes.min_value
            + extremes.max_period
            + extremes.min_period
        )
        missing = set(all_extreme_cols) - set(data.columns)
        if missing:
            raise ValueError(f"Extreme period columns not found in data: {missing}")

    # Resolve weights: top-level takes precedence, ClusterConfig.weights is deprecated
    if cluster.weights is not None and weights is not None:
        raise ValueError(
            "weights specified both as top-level parameter and in ClusterConfig. "
            "Use only the top-level weights parameter."
        )
    if cluster.weights is not None:
        # Deprecation warning already emitted by ClusterConfig.__post_init__
        weights = cluster.weights

    # Validate weight columns exist
    if weights is not None:
        missing = set(weights.keys()) - set(data.columns)
        if missing:
            raise ValueError(f"Weight columns not found in data: {missing}")

    # Build old API parameters
    old_params = _build_old_params(
        data=data,
        n_clusters=n_clusters,
        period_duration=period_duration,
        temporal_resolution=temporal_resolution,
        cluster=cluster,
        segments=segments,
        extremes=extremes,
        weights=weights,
        preserve_column_means=preserve_column_means,
        rescale_exclude_columns=rescale_exclude_columns,
        round_decimals=round_decimals,
        numerical_tolerance=numerical_tolerance,
    )

    # Run aggregation using old implementation (suppress deprecation warning for internal use)
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", LegacyAPIWarning)
        agg = TimeSeriesAggregation(**old_params)
        cluster_representatives = agg.createTypicalPeriods()

    # Rename index levels for consistency with new API terminology
    cluster_representatives = cluster_representatives.rename_axis(
        index={"PeriodNum": "cluster", "TimeStep": "timestep"}
    )

    # Build accuracy metrics
    accuracy_df = agg.accuracyIndicators()

    # Build rescale deviations DataFrame
    rescale_deviations_dict = getattr(agg, "_rescaleDeviations", {})
    if rescale_deviations_dict:
        rescale_deviations = pd.DataFrame.from_dict(
            rescale_deviations_dict, orient="index"
        )
        rescale_deviations.index.name = "column"
    else:
        rescale_deviations = pd.DataFrame(
            columns=["deviation_pct", "converged", "iterations"]
        )

    accuracy = AccuracyMetrics(
        rmse=accuracy_df["RMSE"],
        mae=accuracy_df["MAE"],
        rmse_duration=accuracy_df["RMSE_duration"],
        rescale_deviations=rescale_deviations,
        weighted_rmse=_weighted_rms(accuracy_df["RMSE"], weights),
        weighted_mae=_weighted_mean(accuracy_df["MAE"], weights),
        weighted_rmse_duration=_weighted_rms(accuracy_df["RMSE_duration"], weights),
    )

    # Build ClusteringResult
    time_index = data.index if isinstance(data.index, pd.DatetimeIndex) else None
    clustering_result = _build_clustering_result(
        agg=agg,
        n_segments=segments.n_segments if segments else None,
        cluster_config=cluster,
        segment_config=segments,
        extremes_config=extremes,
        weights=weights,
        preserve_column_means=preserve_column_means,
        rescale_exclude_columns=rescale_exclude_columns,
        temporal_resolution=temporal_resolution,
        time_index=time_index,
    )

    # Compute segment_durations as tuple of tuples
    segment_durations_tuple = None
    if segments and hasattr(agg, "segmentedNormalizedTypicalPeriods"):
        segmented_df = agg.segmentedNormalizedTypicalPeriods
        segment_durations_tuple = tuple(
            tuple(
                int(seg_dur)
                for _seg_step, seg_dur, _orig_start in segmented_df.loc[
                    period_idx
                ].index
            )
            for period_idx in segmented_df.index.get_level_values(0).unique()
        )

    # Build result object
    return AggregationResult(
        cluster_representatives=cluster_representatives,
        cluster_weights=dict(agg.clusterPeriodNoOccur),
        n_timesteps_per_period=agg.timeStepsPerPeriod,
        segment_durations=segment_durations_tuple,
        accuracy=accuracy,
        clustering_duration=getattr(agg, "clusteringDuration", 0.0),
        clustering=clustering_result,
        is_transferred=False,
        _aggregation=agg,
    )

unstack_to_periods

unstack_to_periods(
    data: DataFrame, period_duration: int | float | str = 24
) -> pd.DataFrame

Reshape time series data into period structure for visualization.

Transforms a flat time series into a DataFrame with periods as rows and timesteps as a MultiIndex level, suitable for creating heatmaps with plotly.

Parameters:

Name	Type	Description	Default
data	DataFrame	Time series data with datetime index.	required
period_duration	int, float, or str	Length of each period. Accepts: - int/float: hours (e.g., 24 for daily, 168 for weekly) - str: pandas Timedelta string (e.g., '24h', '1d', '1w')	24

Returns:

Type	Description
DataFrame	Reshaped data with shape (n_periods, n_timesteps_per_period) for each column. Suitable for px.imshow(result["column"].values.T) to create heatmaps.

Examples:

>>> import tsam
>>> import plotly.express as px
>>>
>>> # Reshape data for heatmap visualization
>>> unstacked = tsam.unstack_to_periods(df, period_duration=24)
>>>
>>> # Create heatmap with plotly
>>> px.imshow(
...     unstacked["Load"].values.T,
...     labels={"x": "Day", "y": "Hour", "color": "Load"},
...     title="Load Heatmap"
... )

Source code in src/tsam/api.py

def unstack_to_periods(
    data: pd.DataFrame,
    period_duration: int | float | str = 24,
) -> pd.DataFrame:
    """Reshape time series data into period structure for visualization.

    Transforms a flat time series into a DataFrame with periods as rows and
    timesteps as a MultiIndex level, suitable for creating heatmaps with plotly.

    Parameters
    ----------
    data : pd.DataFrame
        Time series data with datetime index.
    period_duration : int, float, or str, default 24
        Length of each period. Accepts:
        - int/float: hours (e.g., 24 for daily, 168 for weekly)
        - str: pandas Timedelta string (e.g., '24h', '1d', '1w')

    Returns
    -------
    pd.DataFrame
        Reshaped data with shape (n_periods, n_timesteps_per_period) for each column.
        Suitable for ``px.imshow(result["column"].values.T)`` to create heatmaps.

    Examples
    --------
    >>> import tsam
    >>> import plotly.express as px
    >>>
    >>> # Reshape data for heatmap visualization
    >>> unstacked = tsam.unstack_to_periods(df, period_duration=24)
    >>>
    >>> # Create heatmap with plotly
    >>> px.imshow(
    ...     unstacked["Load"].values.T,
    ...     labels={"x": "Day", "y": "Hour", "color": "Load"},
    ...     title="Load Heatmap"
    ... )
    """
    period_hours = _parse_duration_hours(period_duration, "period_duration")

    # Infer timestep resolution from data index
    timestep_hours = 1.0  # Default to hourly
    if isinstance(data.index, pd.DatetimeIndex) and len(data.index) > 1:
        timestep_hours = (data.index[1] - data.index[0]).total_seconds() / 3600

    # Calculate timesteps per period
    timesteps_per_period = round(period_hours / timestep_hours)
    if timesteps_per_period < 1:
        raise ValueError(
            f"period_duration ({period_hours}h) is smaller than "
            f"data timestep resolution ({timestep_hours}h)"
        )

    with warnings.catch_warnings():
        warnings.simplefilter("ignore", LegacyAPIWarning)
        unstacked, _ = unstackToPeriods(data.copy(), timesteps_per_period)
    return cast("pd.DataFrame", unstacked)