ParquetDataCatalog append mode missing in v1.219?

[trylovetom]

Hello,

I use ParquetDataCatalog to save real-time market data. In version 1.218, there was append mode. But in version 1.219, I cannot find it.

Problem: tick1 is gone! Only tick2 in file

catalog = ParquetDataCatalog("/path/to/catalog")
start_ns, end_ns = get_day_range_nanos()  # Today 00:00 to 23:59

# 09:30 AM - First trade comes
catalog.write_data(data=[tick1], start=start_ns, end=end_ns)

# 09:31 AM - Second trade comes (1 minutes later)
catalog.write_data(data=[tick2], start=start_ns, end=end_ns)

What I need:

• One file per day (all trades from 00:00-23:59 in same file)
• Each write_data() should ADD to file, not replace
• Data comes every few seconds/minutes during trading hours

Questions:

1. Version 1.218 had mode=CatalogWriteMode.APPEND. Where is it in 1.219?
2. Did I miss something?
3. How to save continuous data to same daily file now?

[faysou]

The catalog has changed. Each write data creates a new file. There's also consolidate by period function that allows to rewrite files. Look at the documentation, I've updated it.

[faysou]

It's not designed for this. Use instead streamingfeatherwriter then convert it to parquet. There are functions for this, look in the code, the documentation, or ask an LLM agent.

[trylovetom]

Oh! That’s great, thank you for your reply. This is exactly what I need. I will ask my code pilot,

but may I ask if there is an example I can look at?

[trylovetom]

I am still learning. So far, I only found documents that use backtest node to do convert parquet.
Could you please tell me how to do it using live node? Thanks!

[trylovetom]

Stay calm, bro. I’m not trying to give you a hard time.
Thanks for your reply.
I’ve been writing Python for 10 years — when I said I’m “learning,” I was just being humble.

I did try to find the answer myself, but the documentation at https://nautilustrader.io/docs/latest/concepts/data/#feather-streaming-and-conversion is not very helpful.
Right now, Nautilus Trader still seems unstable, and the sample code in the docs cannot run properly. Also, when I try the following code in a live node, I get an error. This has made the development process quite time-consuming for me:

node.trader.add_actor(actor)
node.build()
node.run()

catalog = ParquetDataCatalog(path="catalog")
# subdirectory must be "live" for live node or it won’t work
catalog.convert_stream_to_data(node.instance_id, data_cls=TradeTick, subdirectory="live")

I understand Nautilus Trader is still undergoing major updates, and many areas in the documentation still need improvement. The developer experience could definitely be better — but I get it. As long as we stay friendly, we can work together to build something great.

[cjdsellers]

Hi @trylovetom

Thanks for taking the time to provide the detailed feedback and for your patience as we maintain this high pace of development and find bandwidth to bring the documentation up to date.

[trylovetom]

Here is an example that was found. I'm sharing it with the community. If there are any mistakes, please feel free to point them out.

from os import environ

from dotenv import load_dotenv
from nautilus_trader.config import (
  InstrumentProviderConfig,
  LiveDataClientConfig,
  TradingNodeConfig,
)
from nautilus_trader.live.node import TradingNode
from nautilus_trader.model.data import (
  InstrumentStatus,
  OrderBookDepth10,
  TradeTick,
)
from nautilus_trader.model.identifiers import InstrumentId
from nautilus_trader.persistence.catalog import ParquetDataCatalog
from nautilus_trader.persistence.config import RotationMode, StreamingConfig
from pandas import Timedelta

# my custom adapter
from tw_adapter.live_market_data_client_factory import TwLiveDataClientFactory
from tw_adapter.twse_api import TwseMarket, TwseType
from tw_adapter.utils import tw_name
from tw_persist_actor import TwPersistActor, TwPersistActorConfig


def main():
  load_dotenv()
  instrument_ids = [
    InstrumentId.from_str("1101.TW"),
    InstrumentId.from_str("2330.TW"),
  ]
  actor = TwPersistActor(TwPersistActorConfig(instrument_ids))
  markets: list[TwseMarket] = ["上市", "上櫃"]
  types: list[TwseType] = ["股票"]
  query = f"market in {markets} and type in {types}"
  node = TradingNode(
    config=TradingNodeConfig(
      data_clients={
        tw_name: LiveDataClientConfig(
          instrument_provider=InstrumentProviderConfig(
            load_all=True, filters={"query": query}
          )
        )
      },
      streaming=StreamingConfig(
        catalog_path=environ.get("CATALOG_PATH"),
        fs_protocol="file",
        include_types=[TradeTick, InstrumentStatus, OrderBookDepth10],
        flush_interval_ms=1000,  # Flush every second
        replace_existing=False,
        rotation_mode=RotationMode.INTERVAL,
        rotation_interval=Timedelta(days=1),
        rotation_timezone="Asia/Taipei",
        max_file_size=1024 * 1024 * 100,  # 100MB max file size
      ),
    )
  )
  node.add_data_client_factory(tw_name, TwLiveDataClientFactory)
  node.trader.add_strategy(actor)
  node.build()
  node.run()

  catalog = ParquetDataCatalog(path=environ.get("CATALOG_PATH"))
  catalog.convert_stream_to_data(
    node.instance_id.value,
    data_cls=TradeTick,
    subdirectory="live",
  )
  catalog.convert_stream_to_data(
    node.instance_id.value,
    data_cls=InstrumentStatus,
    subdirectory="live",
  )
  catalog.convert_stream_to_data(
    node.instance_id.value,
    data_cls=OrderBookDepth10,
    subdirectory="live",
  )


if __name__ == "__main__":
  main()