Issue with loading multiple instruments data to data catalog

[amitsinghpgs]

Expected Behavior

Using the process_files function we should be able to add multiple instruments data in a single data catalog

Actual Behavior

I am facing issues in adding multiple instruments to data catalog.
I am seeing the behavior when ever I add 2 instruments to the catalog. If POLYUSDT is added after YFIUSDT then the instrument id of YFIUSDT becomes NA in catalog.trade_ticks(). if YFIUSDT is added after POLYUSDT then instrument id of POLYUSDT is shown as NA

Steps to Reproduce the Problem

def parser(data, instrument_id):
    """ Parser function for hist_data FX data, for use with CSV Reader """
    dt_ = pd.Timestamp(dt.datetime.fromtimestamp(int(data['timestamp'])/1e3), tz='UTC')
    
    yield TradeTick(
        instrument_id=instrument_id,
        price=Price(data['price'], 2),
        size=Quantity(data['volume'], 5),
        aggressor_side =AggressorSideParser.from_str_py("BUY" if int(data['side'])==1 else "SELL"),
        trade_id=str(data['id']),
        ts_event=dt_to_unix_nanos(dt_),
        ts_init=dt_to_unix_nanos(dt_)
    )



YFI = Currency("YFI", precision=8, iso4217=0, name="Yearn", currency_type=CurrencyType.CRYPTO)
YFIUSDT = CurrencySpot(
            instrument_id=InstrumentId(
                symbol=Symbol("YFI/USDT"),
                venue=Venue("BINANCE"),
            ),
            native_symbol=Symbol("YFIUSDT"),
            base_currency=YFI,
            quote_currency=USDT,
            price_precision=2,
            size_precision=5,
            price_increment=Price(1e-02, precision=2),
            size_increment=Quantity(1e-05, precision=5),
            lot_size=None,
            max_quantity=Quantity(9000, precision=5),
            min_quantity=Quantity(1e-05, precision=5),
            max_notional=None,
            min_notional=Money(10.00, USDT),
            max_price=Price(1000000, precision=2),
            min_price=Price(0.01, precision=2),
            margin_init=Decimal("1.00"),
            margin_maint=Decimal("0.35"),
            maker_fee=Decimal("0.0001"),
            taker_fee=Decimal("0.0001"),
            ts_event=0,
            ts_init=0,
        )
POLY = Currency("POLY", precision=8, iso4217=0, name="Poly", currency_type=CurrencyType.CRYPTO)
POLYUSDT = CurrencySpot(
            instrument_id=InstrumentId(
                symbol=Symbol("POLY/USDT"),
                venue=Venue("BINANCE"),
            ),
            native_symbol=Symbol("POLYUSDT"),
            base_currency=POLY,
            quote_currency=USDT,
            price_precision=2,
            size_precision=5,
            price_increment=Price(1e-02, precision=2),
            size_increment=Quantity(1e-05, precision=5),
            lot_size=None,
            max_quantity=Quantity(9000, precision=5),
            min_quantity=Quantity(1e-05, precision=5),
            max_notional=None,
            min_notional=Money(10.00, USDT),
            max_price=Price(1000000, precision=2),
            min_price=Price(0.01, precision=2),
            margin_init=Decimal("1.00"),
            margin_maint=Decimal("0.35"),
            maker_fee=Decimal("0.0001"),
            taker_fee=Decimal("0.0001"),
            ts_event=0,
            ts_init=0,
        )



import os, shutil

CATALOG_PATH =  "/home/amit//catalog"
# Clear if it already exists, then create fresh
if os.path.exists(CATALOG_PATH):
    shutil.rmtree(CATALOG_PATH)
os.mkdir(CATALOG_PATH)


catalog = DataCatalog(CATALOG_PATH)



write_objects(catalog, [YFIUSDT, POLYUSDT])



process_files(
    glob_path="~/DATA/crypto_archive/YFIUSDT.csv.gz",
    reader=CSVReader_(
        block_parser=lambda x: parser(x, instrument_id=YFIUSDT.id), 
        header=['id', 'price', 'volume', 'timestamp', 'side'],
        chunked=False, 
        as_dataframe=True,
        sep='|'
    ),
    catalog=catalog,
)



process_files(
    glob_path="~/DATA/crypto_archive/POLYUSDT.csv.gz",
    reader=CSVReader_(
        block_parser=lambda x: parser(x, instrument_id=POLYUSDT.id), 
        header=['id', 'price', 'volume', 'timestamp', 'side'],
        chunked=False, 
        as_dataframe=True,
        sep='|'
    ),
    catalog=catalog,

)

catalog.trade_ticks()

Specifications

• OS Platform:
• NautilusTrader Version:

[amitsinghpgs]

Hi @cjdsellers @limx0 I did some debugging yesterday and found the error comes from

nautilus_trader/nautilus_trader/persistence/catalog.py

Line 154 in 70e87ee

df.loc[:, col] = df[col].map(mappings[col])

Somehow the partition_mapping.json file was having only one instrument which was causing an issue in pandas column
$ cat catalog/data/trade_tick.parquet/_partition_mappings.json
{"instrument_id":{"POLY/USDT.BINANCE":"POLY-USDT.BINANCE"}}

The issue does not arise if we pass as_nautilus=True to catalog.trade_ticks()

[limx0]

hey @amitsinghpgs - yep this is definitely the source of the bug. For now you can just update this file to include your other instrument:

# _partition_mappings.json
{"instrument_id":{"POLY/USDT.BINANCE":"POLY-USDT.BINANCE","YFI/USDT.BINANCE":"YFI-USDT.BINANCE"}}

And we'll fix this up in the next release.

[cjdsellers]

Fixed from #593
Many thanks to @limx0 🚀