Improve catalog file names

Author: faysou

nautilus_trader/persistence/catalog/parquet.py

@@ -42,7 +42,9 @@
 from nautilus_trader.core.correctness import PyCondition
 from nautilus_trader.core.data import Data
 from nautilus_trader.core.datetime import dt_to_unix_nanos
+from nautilus_trader.core.datetime import maybe_dt_to_unix_nanos
 from nautilus_trader.core.datetime import time_object_to_dt
+from nautilus_trader.core.datetime import unix_nanos_to_iso8601
 from nautilus_trader.core.inspect import is_nautilus_class
 from nautilus_trader.core.message import Event
 from nautilus_trader.core.nautilus_pyo3 import DataBackendSession
@@ -313,7 +315,8 @@ def _write_chunk(
 
         start = start if start else data[0].ts_init
         end = end if end else data[-1].ts_init
-        parquet_file = f"{directory}/{start}-{end}.parquet"
+        filename = _timestamps_to_filename(start, end)
+        parquet_file = f"{directory}/{filename}"
         pq.write_table(
             table,
             where=parquet_file,
@@ -390,13 +393,19 @@ def extend_file_name(
 
         for interval in intervals:
             if interval[0] == end + 1:
-                old_path = os.path.join(directory, f"{interval[0]}-{interval[1]}.parquet")
-                new_path = os.path.join(directory, f"{start}-{interval[1]}.parquet")
+                old_path = os.path.join(
+                    directory,
+                    _timestamps_to_filename(interval[0], interval[1]),
+                )
+                new_path = os.path.join(directory, _timestamps_to_filename(start, interval[1]))
                 self.fs.rename(old_path, new_path)
                 break
             elif interval[1] == start - 1:
-                old_path = os.path.join(directory, f"{interval[0]}-{interval[1]}.parquet")
-                new_path = os.path.join(directory, f"{interval[0]}-{end}.parquet")
+                old_path = os.path.join(
+                    directory,
+                    _timestamps_to_filename(interval[0], interval[1]),
+                )
+                new_path = os.path.join(directory, _timestamps_to_filename(interval[0], end))
                 self.fs.rename(old_path, new_path)
                 break
 
@@ -482,7 +491,7 @@ def _reset_file_names(self, directory: str) -> None:
             if first_ts == -1:
                 continue
 
-            new_filename = f"{first_ts}-{last_ts}.parquet"
+            new_filename = _timestamps_to_filename(first_ts, last_ts)
             new_path = os.path.join(os.path.dirname(file), new_filename)
             self.fs.rename(file, new_path)
 
@@ -609,7 +618,10 @@ def _consolidate_directory(
         if ensure_contiguous_files:
             assert _are_intervals_contiguous(intervals)
 
-        new_file_name = os.path.join(directory, f"{intervals[0][0]}-{intervals[-1][1]}.parquet")
+        new_file_name = os.path.join(
+            directory,
+            _timestamps_to_filename(intervals[0][0], intervals[-1][1]),
+        )
         files_to_consolidate.sort()
         self._combine_parquet_files(files_to_consolidate, new_file_name)
 
@@ -1378,6 +1390,18 @@ def convert_stream_to_data(
         used_catalog.write_data(all_data)
 
 
+def _timestamps_to_filename(timestamp_1: int, timestamp_2: int) -> str:
+    datetime_1 = _iso_timestamp_to_file_timestamp(unix_nanos_to_iso8601(timestamp_1))
+    datetime_2 = _iso_timestamp_to_file_timestamp(unix_nanos_to_iso8601(timestamp_2))
+
+    return f"{datetime_1}_{datetime_2}.parquet"
+
+
+def _iso_timestamp_to_file_timestamp(iso_timestamp: str) -> str:
+    # Assumes format YYYY-MM-DDTHH:MM:SS.nanosecondsZ, "2023-10-26T07:30:50.123456789Z" becomes "2023-10-26T07-30-50-123456789Z"
+    return iso_timestamp.replace(":", "-").replace(".", "-")
+
+
 def _query_intersects_filename(
     filename: str,
     start: pd.Timestamp | None,
@@ -1395,17 +1419,31 @@ def _query_intersects_filename(
 
 def _parse_filename_timestamps(filename: str) -> tuple[int, int] | None:
     base_filename = os.path.splitext(os.path.basename(filename))[0]
-    match = re.match(r"(\d+)-(\d+)", base_filename)
+    match = re.match(r"(.*?)_(.*)", base_filename)
 
     if not match:
         return None
 
-    first_ts = int(match.group(1))
-    last_ts = int(match.group(2))
+    first_ts = maybe_dt_to_unix_nanos(_file_timestamp_to_iso_timestamp(match.group(1)))
+    last_ts = maybe_dt_to_unix_nanos(_file_timestamp_to_iso_timestamp(match.group(2)))
+
+    if not first_ts or not last_ts:
+        return None
 
     return (first_ts, last_ts)
 
 
+def _file_timestamp_to_iso_timestamp(file_timestamp: str) -> str:
+    # Assumes format YYYY-MM-DDTHH-MM-SS-nanosecondsZ, "2023-10-26T07-30-50-123456789Z" becomes "2023-10-26T07:30:50.123456789Z"
+    date_part, time_part = file_timestamp.split("T")
+    time_part = time_part[:-1]
+    last_hyphen_idx = time_part.rfind("-")
+    time_with_dot_for_nanos = time_part[:last_hyphen_idx] + "." + time_part[last_hyphen_idx + 1 :]
+    final_time_part = time_with_dot_for_nanos.replace("-", ":")
+
+    return f"{date_part}T{final_time_part}Z"
+
+
 def _min_max_from_parquet_metadata(file_path: str, column_name: str) -> tuple[int, int]:
     parquet_file = pq.ParquetFile(file_path)
     metadata = parquet_file.metadata

tests/unit_tests/data/test_engine.py

@@ -77,6 +77,7 @@
 from nautilus_trader.model.instruments.base import Instrument
 from nautilus_trader.model.objects import Price
 from nautilus_trader.model.objects import Quantity
+from nautilus_trader.persistence.catalog.parquet import _timestamps_to_filename
 from nautilus_trader.portfolio.portfolio import Portfolio
 from nautilus_trader.test_kit.mocks.data import MockMarketDataClient
 from nautilus_trader.test_kit.mocks.data import setup_catalog
@@ -2521,7 +2522,7 @@ def test_request_bars_when_catalog_and_client_registered(self):
             "data",
             "bar",
             str(bar_type),
-            "1711238400000000000-1711324800000000000.parquet",
+            _timestamps_to_filename(1711238400000000000, 1711324800000000000),
         )
         other_name = os.path.join(catalog.path, "data", "bar", str(bar_type), "other.parquet")
         os.rename(parquet_file, other_name)