plot_cross_correlation

yohou.plotting.diagnostics.plot_cross_correlation(df, *, columns=None, max_lags=None, confidence_level=0.95, groups=None, facet_n_cols=2, color_palette=None, show_legend=True, title=None, x_label=None, y_label=None, width=None, height=None)

Plot cross-correlation function (CCF) between time series pairs.

Computes and visualizes the cross-correlation between pairs of series at various lags, useful for identifying lead-lag relationships and temporal dependencies.

When columns is None (or has more than 2 entries), all unique upper-triangle pairs are computed and arranged in a subplot grid. When exactly two columns are given, a single CCF plot is produced (matching legacy behaviour).

Parameters

Name	Type	Description	Default
df	DataFrame	Input DataFrame with 'time' column and numeric columns.	required
columns	list[str] | None	Column names to cross-correlate. When None, all numeric columns are used and every unique upper-triangle pair is plotted. When exactly two columns are given, a single CCF plot is produced.	None
max_lags	int | None	Number of lags to compute (both positive and negative). If None, uses min(len(df) // 2, 40).	None
confidence_level	float	Confidence level for confidence bands (e.g. 0.95 for 95%).	0.95
groups	list[str] | None	Panel group prefixes. When set (or auto-detected), CCF is computed between members within each group using an upper-triangle matrix layout.	None
facet_n_cols	int	Number of columns in the subplot grid.	2
color_palette	list[str] | None	Custom color palette for pair traces.	None
show_legend	bool	Whether to show the legend.	True
title	str | None	Plot title.	None
x_label	str | None	X-axis label. Defaults to "Lag".	None
y_label	str | None	Y-axis label. Defaults to "Cross-Correlation".	None
width	int | None	Plot width in pixels.	None
height	int | None	Plot height in pixels.	None

Returns

Type	Description
Figure	Plotly figure object.

Examples

>>> import polars as pl
>>> from yohou.plotting import plot_cross_correlation

>>> # Create two time series with lag relationship
>>> df = pl.DataFrame({
...     "time": pl.date_range(pl.date(2020, 1, 1), pl.date(2020, 3, 31), "1d", eager=True),
...     "x": [100 + i for i in range(91)],
...     "y": [105 + i for i in range(91)],  # y leads x by 5 units
... })

>>> # Plot cross-correlation
>>> fig = plot_cross_correlation(df, columns=["x", "y"], max_lags=20)
>>> len(fig.data) > 0
True

See Also

plot_autocorrelation : Plot autocorrelation function.

Source Code

View on GitHub

Show/Hide source

def plot_cross_correlation(
    df: pl.DataFrame,
    *,
    columns: list[str] | None = None,
    max_lags: int | None = None,
    confidence_level: float = 0.95,
    groups: list[str] | None = None,
    facet_n_cols: int = 2,
    color_palette: list[str] | None = None,
    show_legend: bool = True,
    title: str | None = None,
    x_label: str | None = None,
    y_label: str | None = None,
    width: int | None = None,
    height: int | None = None,
) -> go.Figure:
    """Plot cross-correlation function (CCF) between time series pairs.

    Computes and visualizes the cross-correlation between pairs of series at
    various lags, useful for identifying lead-lag relationships and temporal
    dependencies.

    When ``columns`` is ``None`` (or has more than 2 entries), all unique
    upper-triangle pairs are computed and arranged in a subplot grid.  When
    exactly two columns are given, a single CCF plot is produced (matching
    legacy behaviour).

    Parameters
    ----------
    df : pl.DataFrame
        Input DataFrame with 'time' column and numeric columns.
    columns : list[str] | None, default=None
        Column names to cross-correlate.  When ``None``, all numeric
        columns are used and every unique upper-triangle pair is plotted.
        When exactly two columns are given, a single CCF plot is produced.
    max_lags : int | None, default=None
        Number of lags to compute (both positive and negative).
        If None, uses ``min(len(df) // 2, 40)``.
    confidence_level : float, default=0.95
        Confidence level for confidence bands (e.g. ``0.95`` for 95%).
    groups : list[str] | None, default=None
        Panel group prefixes.  When set (or auto-detected), CCF is
        computed between members within each group using an
        upper-triangle matrix layout.
    facet_n_cols : int, default=2
        Number of columns in the subplot grid.
    color_palette : list[str] | None, default=None
        Custom color palette for pair traces.
    show_legend : bool, default=True
        Whether to show the legend.
    title : str | None, default=None
        Plot title.
    x_label : str | None, default=None
        X-axis label. Defaults to ``"Lag"``.
    y_label : str | None, default=None
        Y-axis label. Defaults to ``"Cross-Correlation"``.
    width : int | None, default=None
        Plot width in pixels.
    height : int | None, default=None
        Plot height in pixels.

    Returns
    -------
    go.Figure
        Plotly figure object.

    Examples
    --------
    >>> import polars as pl
    >>> from yohou.plotting import plot_cross_correlation

    >>> # Create two time series with lag relationship
    >>> df = pl.DataFrame({
    ...     "time": pl.date_range(pl.date(2020, 1, 1), pl.date(2020, 3, 31), "1d", eager=True),
    ...     "x": [100 + i for i in range(91)],
    ...     "y": [105 + i for i in range(91)],  # y leads x by 5 units
    ... })

    >>> # Plot cross-correlation
    >>> fig = plot_cross_correlation(df, columns=["x", "y"], max_lags=20)
    >>> len(fig.data) > 0
    True

    See Also
    --------
    [`plot_autocorrelation`][yohou.plotting.plot_autocorrelation] : Plot autocorrelation function.
    """
    # Validate inputs
    validate_plotting_data(df)
    validate_plotting_params(width=width, height=height)

    if max_lags is None:
        max_lags = min(len(df) // 2, 40)

    _lag_vals = list(range(-max_lags, max_lags + 1))

    def _add_ccf_bar(
        fig: go.Figure,
        lag_vals: list[int],
        ccf: list[float],
        pair_color: str,
        pair_label: str,
        show_in_legend: bool,
        *,
        row: int | None = None,
        col: int | None = None,
    ) -> None:
        """Add a single ACF-style bar trace."""
        trace_kwargs: dict = {}
        if row is not None and col is not None:
            trace_kwargs["row"] = row
            trace_kwargs["col"] = col
        fig.add_trace(
            go.Bar(
                x=lag_vals,
                y=ccf,
                marker={"color": pair_color},
                name=pair_label,
                showlegend=show_in_legend,
                legendgroup=pair_label,
                hovertemplate=_make_hovertemplate(pair_label, "Lag", "CCF", decimals=3),
            ),
            **trace_kwargs,
        )

    def _add_ccf_bands(
        fig: go.Figure,
        n_pts: int,
        *,
        row: int | None = None,
        col: int | None = None,
    ) -> None:
        """Add confidence bands and zero line."""
        hline_kw: dict = {}
        if row is not None and col is not None:
            hline_kw["row"] = row
            hline_kw["col"] = col
        if confidence_level > 0:
            cb = norm.ppf(1 - (1 - confidence_level) / 2) / math.sqrt(n_pts)
            fig.add_hline(y=cb, line={"dash": "dash", "color": "#DC2626", "width": 1}, **hline_kw)
            fig.add_hline(y=-cb, line={"dash": "dash", "color": "#DC2626", "width": 1}, **hline_kw)
        fig.add_hline(y=0, line={"color": "#64748B", "width": 1}, **hline_kw)

    # Auto-detect panel data
    if groups is None and _auto_detect_panel(df) and columns is None:
        groups = []

    if groups is not None:
        _panel_cols = resolve_panel_columns(df, groups, columns)
        grouped, _all_members = _group_panel_columns(_panel_cols)
        color_mgr = PanelColorManager(color_palette)

        # Build per-group pair lists
        group_pairs: list[tuple[str, list[tuple[str, str]]]] = []
        for gname, gcols in grouped.items():
            member_names = [_member_name(c) for c in gcols]
            pairs = _upper_triangle_pairs(member_names)
            if pairs:
                group_pairs.append((gname, pairs))

        if not group_pairs:
            msg = f"Need at least 2 members per group for cross-correlation. Groups: {list(grouped.keys())}"
            raise ValueError(msg)

        # All groups + pairs in one figure
        all_cells: list[tuple[str, str, str]] = []
        for gname, pairs in group_pairs:
            for a, b in pairs:
                all_cells.append((gname, a, b))

        n = len(all_cells)
        n_cols_grid = min(n, facet_n_cols)
        n_rows_grid = math.ceil(n / n_cols_grid)
        legend_tracker = LegendTracker(show_legend=show_legend)

        fig = make_subplots(
            rows=n_rows_grid,
            cols=n_cols_grid,
            subplot_titles=[f"{gn}: {a} vs {b}" for gn, a, b in all_cells],
            vertical_spacing=_subplot_spacing(n_rows_grid),
            horizontal_spacing=_subplot_spacing(n_cols_grid) if n_cols_grid > 1 else 0.08,
        )

        for idx, (gname, a_name, b_name) in enumerate(all_cells):
            r = idx // n_cols_grid + 1
            c = idx % n_cols_grid + 1
            col_a = f"{gname}__{a_name}"
            col_b = f"{gname}__{b_name}"
            clean = df.select(col_a, col_b).drop_nulls()
            x_arr = clean[col_a].to_numpy()
            y_arr = clean[col_b].to_numpy()
            ccf = _compute_ccf(x_arr, y_arr, max_lags)
            pair_label = f"{a_name} vs {b_name}"
            pair_color = color_mgr.get_color(pair_label)
            _add_ccf_bar(
                fig,
                _lag_vals,
                ccf,
                pair_color,
                pair_label,
                legend_tracker.should_show(pair_label),
                row=r,
                col=c,
            )
            _add_ccf_bands(fig, len(x_arr), row=r, col=c)

        fig = apply_default_layout(
            fig,
            title=title or "Cross-Correlation",
            x_label=x_label or "Lag",
            y_label=y_label or "Cross-Correlation",
            width=width,
            height=height,
        )
        fig.update_layout(showlegend=show_legend)
        return fig

    plot_columns = validate_plotting_data(df, columns=columns, exclude=["time"])

    if len(plot_columns) < 2:  # noqa: PLR2004
        msg = f"Cross-correlation requires at least 2 columns, got {len(plot_columns)}: {plot_columns}"
        raise ValueError(msg)

    pairs = _upper_triangle_pairs(plot_columns)
    color_mgr = PanelColorManager(color_palette)
    legend_tracker = LegendTracker()

    if len(pairs) == 1:
        # Single pair - flat figure
        x_column, y_column = pairs[0]
        clean = df.select(x_column, y_column).drop_nulls()
        x_arr = clean[x_column].to_numpy()
        y_arr = clean[y_column].to_numpy()
        n_pts = len(x_arr)
        ccf = _compute_ccf(x_arr, y_arr, max_lags)
        pair_label = f"{x_column} vs {y_column}"
        pair_color = color_mgr.get_color(pair_label)

        fig = go.Figure()
        _add_ccf_bar(fig, _lag_vals, ccf, pair_color, pair_label, True)

        if confidence_level > 0:
            cb = norm.ppf(1 - (1 - confidence_level) / 2) / math.sqrt(n_pts)
            ci_pct = f"{confidence_level:.0%}"
            fig.add_hline(
                y=cb,
                line={"dash": "dash", "color": "#DC2626", "width": 1},
                annotation_text=f"{ci_pct} CI",
                annotation_position="right",
            )
            fig.add_hline(y=-cb, line={"dash": "dash", "color": "#DC2626", "width": 1})
        fig.add_hline(y=0, line={"color": "#64748B", "width": 1})

        fig = apply_default_layout(
            fig,
            title=title or "Cross-Correlation",
            x_label=x_label or "Lag",
            y_label=y_label or "Cross-Correlation",
            width=width,
            height=height,
        )
        fig.update_layout(showlegend=show_legend)
        return fig

    # Multiple pairs - upper-triangle subplot grid
    n = len(pairs)
    n_cols_grid = min(n, facet_n_cols)
    n_rows_grid = math.ceil(n / n_cols_grid)

    fig = make_subplots(
        rows=n_rows_grid,
        cols=n_cols_grid,
        subplot_titles=[f"{a} vs {b}" for a, b in pairs],
        vertical_spacing=_subplot_spacing(n_rows_grid),
        horizontal_spacing=_subplot_spacing(n_cols_grid) if n_cols_grid > 1 else 0.08,
    )

    for idx, (col_a, col_b) in enumerate(pairs):
        r = idx // n_cols_grid + 1
        c = idx % n_cols_grid + 1
        clean = df.select(col_a, col_b).drop_nulls()
        x_arr = clean[col_a].to_numpy()
        y_arr = clean[col_b].to_numpy()
        ccf = _compute_ccf(x_arr, y_arr, max_lags)
        pair_label = f"{col_a} vs {col_b}"
        pair_color = color_mgr.get_color(pair_label)
        _add_ccf_bar(
            fig,
            _lag_vals,
            ccf,
            pair_color,
            pair_label,
            legend_tracker.should_show(pair_label),
            row=r,
            col=c,
        )
        _add_ccf_bands(fig, len(x_arr), row=r, col=c)

    fig = apply_default_layout(
        fig,
        title=title or "Cross-Correlation",
        x_label=x_label or "Lag",
        y_label=y_label or "Cross-Correlation",
        width=width,
        height=height,
    )
    fig.update_layout(showlegend=show_legend)
    return fig

Tutorials

The following example notebooks use this component:

• How to Visualize Correlations

  ---

  Visualization

  Pairwise correlation heatmaps, scatter matrices, cross-correlation at multiple lags, and lag scatter plots for multivariate time series diagnostics.

  View · Open in marimo