Join schemas on instrument ID

import asyncio
import databento as db

async def fetch_data():
    client = db.Historical(key="$YOUR_API_KEY")
    mbp_task = client.timeseries.get_range_async(
        dataset="GLBX.MDP3",
        schema="trades",
        symbols=["NG.FUT"],
        start="2023-04-02T00:00:00",
        end="2023-04-05T00:00:00",
        stype_in="parent",
    )
    def_task = client.timeseries.get_range_async(
        dataset="GLBX.MDP3",
        schema="definition",
        symbols=["NG.FUT"],
        start="2023-04-02T00:00:00",
        end="2023-04-05T00:00:00",
        stype_in="parent",
    )
    return await asyncio.gather(mbp_task, def_task)

# Concurrently fetch definitions and trades
mbp_store, def_store = asyncio.run(fetch_data())
mbp_df = mbp_store.to_df()
def_df = def_store.to_df()
# Join futures with their definitions on instrument ID
combined_df = mbp_df.merge(
    def_df,
    on="instrument_id",
    how="inner",
    suffixes=("", "_def"),
).set_index("ts_event")
# Remove spreads
combined_df = combined_df.loc[combined_df["instrument_class"] == "F", :]
print(combined_df[["raw_symbol", "expiration", "price"]].head())

                          raw_symbol                expiration  price
ts_event
2023-04-02 22:00:00+00:00       NGK3 2023-04-26 18:30:00+00:00  2.085
2023-04-02 22:00:00+00:00       NGK3 2023-04-26 18:30:00+00:00  2.085
2023-04-02 22:00:00+00:00       NGK3 2023-04-26 18:30:00+00:00  2.085
2023-04-02 22:00:00+00:00       NGK3 2023-04-26 18:30:00+00:00  2.080
2023-04-02 22:00:00+00:00       NGK3 2023-04-26 18:30:00+00:00  2.080