1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

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())

1
2
3
4
5
6
7

                          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       NGM3 2023-05-26 18:30:00+00:00  2.345
2023-04-02 22:00:00+00:00       NGM3 2023-05-26 18:30:00+00:00  2.345