Determine publishing times for statistics

import datetime as dt
import databento as db
import matplotlib.pyplot as plt
import pandas as pd
from matplotlib.dates import DateFormatter, MinuteLocator

# Create mapping of statistic type to human-readable name
stat_type_dict = {x.value: x.name.replace("_", " ").capitalize() for x in db.StatType.variants()}

def plot_statistics_distribution(
    symbol: str,
    stats_data: pd.DataFrame,
    status_data: pd.DataFrame,
):
    """Plot publishing time distribution of each statistic type"""
    stats_data = (
        stats_data.groupby("stat_type")
        .resample("15min", include_groups=False)
        .size().unstack(level=0).fillna(0)
    ).transpose()
    stats_data = stats_data.rename(index=stat_type_dict)

    _, ax = plt.subplots(figsize=(16, 8))

    times = pd.to_datetime(stats_data.columns)
    time_width = pd.Timedelta(minutes=15)

    # Plot horizontal bars for each statistic type
    for idx, stat_type in enumerate(stats_data.index):
        mask = stats_data.loc[stat_type] > 0
        if mask.any():
            ax.barh(
                idx,
                width=time_width * 1.1,
                left= times[mask] - time_width / 2,
                label=stat_type,
            )

    # Plot vertical lines for changes in market state
    for idx, row in status_data.iterrows():
        ax.axvline(
            x=idx,
            color="dimgray",
            linestyle=":",
            linewidth=1.5,
        )
        ax.text(
            x=idx,
            y=ax.get_ylim()[1],
            s=str(row["action"]),
            ha="left",
            va="top",
            rotation=-90,
            color="dimgray",
        )

    # Format axes
    ax.set(
        yticks=range(len(stats_data)),
        yticklabels=stats_data.index,
        xlabel="Time (UTC)",
        ylabel="Stat Type",
        title=f"{symbol} Statistics Message Distribution",
    )
    ax.xaxis.set_major_formatter(DateFormatter("%H:%M"))
    ax.xaxis.set_major_locator(MinuteLocator([0, 60]))
    ax.set_axisbelow(True)
    ax.grid()

    plt.setp(ax.xaxis.get_majorticklabels(), rotation=45, ha="right")
    plt.tight_layout()
    plt.show()

# Set parameters
dataset = "IFUS.IMPACT"
product = "CC"
start = "2025-09-17"

client = db.Historical("$YOUR_API_KEY")

# Request statistics data for the continuous contract
stats_df = client.timeseries.get_range(
    dataset=dataset,
    symbols=f"{product}.v.0",
    stype_in="continuous",
    schema="statistics",
    start=start,
).to_df()

# Request status data for the continuous contract
status_df = client.timeseries.get_range(
    dataset=dataset,
    symbols=f"{product}.v.0",
    stype_in="continuous",
    schema="status",
    start=start,
).to_df()

# Filter out midnight snapshot
status_df = status_df[status_df.index.time != dt.time.min]

# Only keep certain market states
# For a full list of variants, see https://databento.com/docs/schemas-and-data-formats/status#status-action-variants
status_map = {1: "Pre-Open", 7: "Open", 12: "Close"}
status_df = status_df[status_df["action"].isin(status_map)]
status_df = status_df[status_df["trading_event"] != 1]
status_df["action"] = status_df["action"].map(status_map)

# Print out the number of occurrences for each statistic type
print(stats_df["stat_type"].map(stat_type_dict).value_counts())

# Plot distribution of statistic publishing time
plot_statistics_distribution(
    symbol=product,
    stats_data=stats_df,
    status_data=status_df,
)

stat_type
Trading session high price    4482
Trading session low price     4482
Vwap                          4482
Cleared volume                4482
Indicative opening price      3060
Opening price                    3
Open interest                    3
Settlement price                 3
Name: count, dtype: int64