Skip to content

storage.md

Latest commit

 

History

History
270 lines (204 loc) · 7.92 KB

storage.md

File metadata and controls

270 lines (204 loc) · 7.92 KB
Raw

Storage

ml4t-data stores time-series data as Parquet files with two backend strategies: Hive (partitioned by time) and Flat (single file per key). Both backends share atomic writes, file locking, metadata tracking, and Polars lazy evaluation.

Choosing a Backend

Hive Flat
Best for Large datasets, time-range queries Small datasets, simple access
Layout Directory tree with year=.../month=.../data.parquet Single .parquet file per key
Date filtering Prunes partitions before reading Reads the single file and filters rows
Write cost Higher (one file per partition) Lower (single file)
Default Yes No 
from ml4t.data.storage import create_storage

# Hive storage (default)
storage = create_storage("./data", strategy="hive")

# Flat storage
storage = create_storage("./data", strategy="flat")

Partition Granularity

Hive storage partitions data by time. Choose granularity based on your data frequency to keep partition sizes in the 200-5,000 row range:

Granularity Partition columns Best for Rows/partition (stocks)
year year/ Daily data ~252
month year=/month=/ Hourly data ~720
day year=/month=/day=/ Minute data ~1,440
hour year=/month=/day=/hour=/ Second/tick data ~3,600 
from ml4t.data.storage import HiveStorage, StorageConfig

# Daily equity data -- partition by year
config = StorageConfig(
    base_path="./data",
    partition_granularity="year",
)
storage = HiveStorage(config)

# Minute crypto data -- partition by day
config = StorageConfig(
    base_path="./data",
    partition_granularity="day",
)
storage = HiveStorage(config)

On-disk layout for month-level partitioning. Storage keys can use slashes, such as equities/daily/AAPL. Each key is encoded into one filesystem-safe directory and published through a CURRENT commit pointer:

data/
  k1_<encoded-key>/
    CURRENT
    commits/
    generations/
      <generation-id>/
        year=2024/
          month=1/
            data.parquet
          month=2/
            data.parquet
  .metadata/
    k1_<encoded-key>.lock

Upgrading Pre-0.1 Storage

Pre-0.1 development releases used ambiguous flattened filesystem names. The current storage backends report those entries but do not guess their logical keys. Inventory the old names, supply an explicit mapping, and run the verified migration:

from ml4t.data.storage import find_legacy_storage_entries, migrate_legacy_storage

entries = find_legacy_storage_entries("./data", "hive")
print([entry.physical_key for entry in entries])

migrate_legacy_storage(
    "./data",
    "hive",
    {"equities_daily_BRK_B": "equities/daily/BRK_B"},
)

The mapping must cover the complete inventory. Migration writes and reads back each new dataset before moving its legacy files to .legacy-v0-backup. Pass the same StorageConfig used by the application through storage_config= when Hive partition granularity or compression differs from the defaults.

Reading and Writing

Both backends use the same StorageBackend interface.

Writing Data

import polars as pl

df = pl.DataFrame({
    "timestamp": [...],
    "open": [...],
    "high": [...],
    "low": [...],
    "close": [...],
    "volume": [...],
})

# Write with a hierarchical storage key
storage.write(df, "equities/daily/AAPL")

The write operation:

  1. Adds partition columns derived from timestamp (Hive only)
  2. Groups data by partition values
  3. Writes each partition atomically (temp file + rename)
  4. Updates the JSON metadata manifest

Reading Data

All reads return a Polars LazyFrame for deferred execution:

# Read all data for a key
lf = storage.read("equities/daily/AAPL")
df = lf.collect()

# Date-filtered read (Hive prunes partitions before scanning)
from datetime import datetime

lf = storage.read(
    "equities/daily/AAPL",
    start_date=datetime(2024, 6, 1),
    end_date=datetime(2024, 12, 31),
)

# Column projection (only read what you need)
lf = storage.read("equities/daily/AAPL", columns=["timestamp", "close", "volume"])

Date ranges use half-open intervals: start_date is inclusive and end_date is exclusive. This convention is consistent across Flat, Hive, and Chunked storage.

With Hive storage, date filters prune entire partition directories before any Parquet file is opened, giving measured 7x speedup on typical queries.

Other Operations

# List all stored keys
keys = storage.list_keys()  # ["AAPL", "BTC-USD", ...]

# Check existence
if storage.exists("AAPL"):
    ...

# Delete all data for a key
storage.delete("AAPL")

# Read metadata
meta = storage.get_metadata("AAPL")
# {"last_updated": "...", "row_count": 5040, "schema": [...], ...}

StorageConfig Options

Field Default Description
base_path required Base directory for all data
strategy "hive" "hive" or "flat"
compression "zstd" Parquet compression: zstd, lz4, snappy, or None
partition_granularity "month" year, month, day, hour (Hive only)
lock_timeout 30 Seconds to wait for another writer on the same key
metadata_tracking True JSON manifest files in .metadata/

Writes are always staged and published atomically. Generate profiles explicitly with the profiling API described below.

Incremental Updates

Hive storage supports chunk-based incremental updates for streaming workflows:

from datetime import datetime

# Save a new data chunk
chunk_path = storage.save_chunk(
    data=new_df,
    symbol="AAPL",
    provider="yahoo",
    start_time=datetime(2024, 12, 1),
    end_time=datetime(2024, 12, 31),
)

# Merge chunk into the main combined file (deduplicates by timestamp)
new_rows = storage.update_combined_file(new_df, symbol="AAPL", provider="yahoo")

# Get latest timestamp for a symbol (to know where to resume)
latest = storage.get_latest_timestamp("AAPL", "yahoo")

Data Profiling

The ProfileMixin adds dataset profiling to any data manager class. Profiles contain column-level statistics (dtype, null count, min/max, mean, std) stored as JSON alongside the data.

from ml4t.data.storage import generate_profile, save_profile, load_profile

# Generate a profile from a DataFrame
profile = generate_profile(df, source="ETFDataManager")
print(profile.summary())
# Dataset Profile (generated: 2024-12-15T10:30:00)
#   Rows: 25,200
#   Columns: 7
#   Date range: 2020-01-02 to 2024-12-13
#   Column Details:
#     close: Float64 (25200 unique, 0.0% null) [mean=156.3, std=42.1]
#     volume: Int64 (24891 unique, 0.0% null) [mean=7.83e+07, std=4.21e+07]

# Save and load profiles
save_profile(profile, Path("data/AAPL_profile.json"))
profile = load_profile(Path("data/AAPL_profile.json"))

# Convert to DataFrame for analysis
profile_df = profile.to_dataframe()

To add profiling to a custom data manager, implement the ProfileMixin:

from ml4t.data.storage import ProfileMixin

class MyDataManager(ProfileMixin):
    def _get_profile_data(self) -> pl.DataFrame:
        return self.load_all()

    def _get_profile_data_path(self) -> Path:
        return self.storage_path / "data.parquet"

    def _get_profile_source_name(self) -> str:
        return "MyDataManager"

# Now available:
profile = manager.generate_profile()  # generates and saves
profile = manager.load_profile()      # loads existing

Concurrency and Safety

Both backends use two mechanisms for safe concurrent access:

  • Atomic writes: Data is written to a temporary file first, then renamed to the target path. This prevents readers from seeing partial writes.
  • File locking: Metadata updates use filelock to prevent corruption when multiple processes write simultaneously. Lock timeout is 10 seconds.

These are enabled by default and can be toggled via StorageConfig.