Trading-Daten mit inkrementellen Updates und partitioniertem Speicher verwalten

# Data Management: From Download to Production Pipeline


# Data Management: From Download to Production Pipeline

**Docker image**: `ml4t`

**Chapter 2: The Financial Data Universe**

Previous notebooks fetched and validated data. This notebook shows how to
**manage** it at scale using ml4t-data's production features:

- **DataManager**: Unified entry point for fetching, storing, and updating
- **Universe**: Predefined symbol lists (S&P 500, NASDAQ 100, etc.)
- **HiveStorage**: Partitioned Parquet for fast queries and incremental writes
- **Incremental Updates**: Keep data fresh without re-downloading history
- **CLI**: Command-line interface for scripted workflows

## Learning Objectives

By completing this notebook, you will:
1. Use `DataManager` as a single entry point for all data operations
2. Load predefined universes with the `Universe` class
3. Store and query data with Hive-partitioned Parquet
4. Perform incremental updates and detect gaps
5. Use the `ml4t-data` CLI for scripted workflows

## Why This Matters

A one-time download is fine for a tutorial. A trading system needs:
- **Daily updates** that only fetch new data (10x faster than full refresh)
- **Partitioned storage** that supports fast date-range queries
- **Gap detection** to ensure completeness before backtesting
- **Metadata tracking** so you know what you have and when it was updated

> **ml4t-data docs**: See the [Incremental Updates Guide](https://ml4trading.io/docs/data/user-guide/incremental-updates/)
> and [Storage Guide](https://ml4trading.io/docs/data/user-guide/storage/) for full reference.

**Prerequisites**: ml4t-data installed; live network access for Yahoo Finance.

## Setup

```python
"""Data Management — DataManager, Universe, HiveStorage, and incremental updates."""

import logging
import shutil
from datetime import datetime
from pathlib import Path

import plotly.graph_objects as go
import polars as pl
import structlog

# ml4t-data emits structured debug logs on every fetch/store; route them
# through stdlib logging at WARNING so the notebook output stays focused on
# the demonstration.
structlog.configure(
    wrapper_class=structlog.make_filtering_bound_logger(logging.WARNING),
)

# ml4t-data core imports
from ml4t.data import DataManager
from ml4t.data.storage import HiveStorage
from ml4t.data.storage.backend import StorageConfig
from ml4t.data.universe import Universe

from utils.downloading import update_through_last_complete_bar
from utils.paths import REPO_ROOT, get_output_dir
from utils.style import COLORS, show_plotly_with_alt


def _rel(path):
    """Repo-relative display path (keeps absolute machine paths out of outputs)."""
    try:
        return path.relative_to(REPO_ROOT)
    except ValueError:
        return path


# Working directory for this notebook's storage examples. Wipe any artifacts
# from a previous run so the demo is fully reproducible.
STORAGE_DIR = get_output_dir(2, "data_management")
if STORAGE_DIR.exists():
    shutil.rmtree(STORAGE_DIR)
STORAGE_DIR.mkdir(parents=True, exist_ok=True)

print(f"Storage directory: {_rel(STORAGE_DIR)}")
```

### Declared parameters

The demo universe, its window and the batch worker count are the three things a CI run
would want to narrow, so they are declared rather than written into the call that uses
them. `REBASE_LEVEL` is the value each series is rebased to in the figure below, and it
appears in the axis label as well as the arithmetic.

```python
ETF_SYMBOLS = ["SPY", "QQQ", "IWM", "TLT", "GLD"]
DEMO_START = "2024-01-01"
DEMO_END = "2024-12-31"
DEMO_PROVIDER = "yahoo"
MAX_WORKERS = 4
STORAGE_SYMBOLS = ["AAPL", "MSFT", "GOOGL"]
DEMO_SYMBOL = STORAGE_SYMBOLS[0]  # the symbol the partition walkthrough follows
STORAGE_START = "2023-01-01"
STORAGE_END = "2024-12-31"
# The read-back below demonstrates partition pruning, so it asks for a window inside what
# was stored rather than the whole of it.
READBACK_START = "2024-01-01"
READBACK_END = "2024-12-31"
PIPELINE_UNIVERSE = "etf_momentum"
REBASE_LEVEL = 100.0
```

---

## 1. DataManager: The Unified Entry Point

`DataManager` puts provider selection, storage and updates behind one interface. Compare:

```python
# Without DataManager (manual)
provider = YahooFinanceProvider()
df = provider.fetch_ohlcv("AAPL", "2024-01-01", "2024-12-31", "daily")

# With DataManager (unified)
dm = DataManager()
df = dm.fetch("AAPL", "2024-01-01", "2024-12-31")
```

The real power shows with batch operations, storage integration, and updates.

### Fetch: Single Symbol

```python
# DataManager without storage — pure fetch mode
dm = DataManager()

# Fetch a single symbol (defaults to Yahoo Finance for equities)
aapl = dm.fetch(DEMO_SYMBOL, DEMO_START, DEMO_END, provider=DEMO_PROVIDER)

print(f"AAPL: {aapl.shape[0]} rows, {aapl.shape[1]} columns")
print(f"Date range: {aapl['timestamp'].min().date()} to {aapl['timestamp'].max().date()}")
print(f"Columns: {aapl.columns}")
aapl.head(3)
```

### Batch Fetch: Multiple Symbols

`batch_load` fetches multiple symbols in parallel and returns a single
stacked DataFrame — the standard multi-asset format used throughout the book.

```python
# Fetch 5 ETFs in parallel
etf_symbols = ETF_SYMBOLS
etf_data = dm.batch_load(
    symbols=etf_symbols,
    start=DEMO_START,
    end=DEMO_END,
    provider=DEMO_PROVIDER,
    max_workers=MAX_WORKERS,
)

print(f"Combined: {etf_data.shape[0]:,} rows across {etf_data['symbol'].n_unique()} symbols")
```

A row count says the call returned something; it does not say what. Rebasing each series to
a common level at the first session makes the frame legible, and colouring by asset class
rather than by ticker makes the point of a multi-asset loader visible: the three equity
lines travel together and the bond and gold lines do not, which is the dispersion a batch
spanning three asset classes exists to capture and a single-asset loader cannot show.

```python
etf_rebased = etf_data.sort("timestamp").with_columns(
    (pl.col("close") / pl.col("close").first().over("symbol") * REBASE_LEVEL).alias("rebased")
)

etf_class = {
    "SPY": "Equities",
    "QQQ": "Equities",
    "IWM": "Equities",
    "TLT": "Long bonds",
    "GLD": "Gold",
}
class_color = {"Equities": COLORS["blue"], "Long bonds": COLORS["copper"], "Gold": COLORS["amber"]}

fig = go.Figure()
seen: set[str] = set()
for sym in etf_symbols:
    s = etf_rebased.filter(pl.col("symbol") == sym)
    cls = etf_class[sym]
    fig.add_trace(
        go.Scatter(
            x=s["timestamp"].to_list(),
            y=s["rebased"].to_list(),
            mode="lines",
            line=dict(color=class_color[cls], width=1.5),
            name=cls,
            legendgroup=cls,
            showlegend=cls not in seen,
            text=sym,
            hovertemplate="%{text}: %{y:.1f}<extra></extra>",
        )
    )
    seen.add(cls)
fig.add_hline(y=REBASE_LEVEL, line=dict(color=COLORS["neutral"], width=1, dash="dot"))
fig.update_layout(
    title="Batch-loaded ETFs, rebased and coloured by asset class",
    xaxis_title="Date",
    yaxis_title=f"Rebased close (first session = {REBASE_LEVEL:.0f})",
    height=420,
    legend_title="Asset class",
)
show_plotly_with_alt(
    fig,
    "Five lines over one year, rebased to a common level at the first session and coloured "
    "in three groups for equities, long bonds and gold, with a dotted reference line at "
    "the rebase level. The three equity lines rise together into the autumn and then "
    "separate, two finishing well above the reference line and one falling back toward "
    "it. The gold line runs highest through the autumn and eases back to join the "
    "equity leaders. The bond line spends almost the whole year below the reference "
    "line and ends furthest below it.",
)
```

---

## 2. Universe: Predefined Symbol Lists

Instead of maintaining symbol lists in YAML or hardcoding them, ml4t-data
ships curated universes that stay current with index rebalances.

```python
print("Available universes:")
for name in Universe.list_universes():
    symbols = Universe.get(name)
    print(f"  {name}: {len(symbols)} symbols")
```

```python
# Access a universe directly
sp500 = Universe.SP500
print(f"\nS&P 500: {len(sp500)} symbols")
print(f"First 10: {sp500[:10]}")
print(f"Last 10:  {sp500[-10:]}")
```

A universe feeds straight into `batch_load`. The full list is the whole index, which is
more fetching than a demonstration needs, so this takes a slice of it.

```python
sp500_sample = dm.batch_load(
    symbols=sp500[:5],
    start="2024-06-01",
    end="2024-12-31",
    provider=DEMO_PROVIDER,
)
print(
    f"S&P 500 sample: {sp500_sample.shape[0]:,} rows, {sp500_sample['symbol'].n_unique()} symbols"
)
```

```python
# Custom universes for your strategy
Universe.add_custom("etf_momentum", ["SPY", "QQQ", "IWM", "EFA", "EEM", "TLT", "GLD"])
Universe.add_custom("crypto_arb", ["BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT"])

print("\nCustom universes registered:")
for name in ["etf_momentum", "crypto_arb"]:
    print(f"  {name}: {Universe.get(name)}")
```

---

## 3. HiveStorage: Partitioned Parquet

For data you'll query repeatedly, Hive-partitioned Parquet is the storage
layer used throughout ml4t-data. The HiveStorage backend encodes a logical key such as
`equities/daily/AAPL` as a filesystem-safe directory name and nests Hive-style year and
month partitions underneath:

```
hive_demo/
├── .metadata/
│   └── equities_daily_AAPL.json
└── equities_daily_AAPL/
    ├── year=2024/month=1/data.parquet
    ├── year=2024/month=2/data.parquet
    └── ...
```

**Benefits over flat files**:
- **Partition pruning**: Query "last 30 days" reads 1 file, not all of history
- **Incremental writes**: New data appends without rewriting existing partitions
- **Metadata tracking**: Know when each symbol was last updated

### DataManager with Storage

```python
# Initialize storage
storage_config = StorageConfig(
    base_path=STORAGE_DIR / "hive_demo",
    compression="zstd",
    partition_granularity="month",
)
storage = HiveStorage(config=storage_config)

# Given a storage backend, the same manager gains load, update and metadata operations.
dm_stored = DataManager(storage=storage)
```

### Load and Store

`DataManager.load()` fetches from the provider and writes to Hive
partitions in one call. The storage key encodes the asset class, frequency,
and symbol.

```python
symbols = STORAGE_SYMBOLS
stored_keys = {}
for symbol in symbols:
    key = dm_stored.load(symbol, STORAGE_START, STORAGE_END, provider=DEMO_PROVIDER)
    stored_keys[symbol] = key
    print(f"  Stored {symbol} → key: {key}")
```

### Query Stored Data

`storage.list_keys()` walks the on-disk layout rather than reading a manifest, so it
reports what is actually there even if a metadata file is missing or stale.

The read below carries a date range, and only the partitions overlapping that range are
opened. Reading one year out of a two-year archive touches about half the files.

```python
stored_symbols = sorted(storage.list_keys())
print(f"Symbols in storage: {stored_symbols}")
aapl_2024 = storage.read(
    stored_keys[DEMO_SYMBOL],
    start_date=datetime.strptime(READBACK_START, "%Y-%m-%d"),
    end_date=datetime.strptime(READBACK_END, "%Y-%m-%d"),
).collect()
print(
    f"\n{DEMO_SYMBOL} {READBACK_START} to {READBACK_END}: "
    f"{len(aapl_2024)} rows out of {STORAGE_START} to {STORAGE_END} stored (partition-pruned)"
)
print(f"Date range: {aapl_2024['timestamp'].min().date()} to {aapl_2024['timestamp'].max().date()}")
```

### Metadata

```python
for symbol in symbols:
    meta = dm_stored.get_metadata(symbol)
    if meta:
        print(f"\n{symbol}:")
        for k, v in list(meta.items())[:5]:
            print(f"  {k}: {v}")
```

### Inspect Partition Structure

The directory names are not addressable from outside: the key is encoded for filesystem
safety, and each write commits into a new generation directory so a failed write cannot
leave a half-written partition visible. `partitions()` is therefore how a caller asks what
the store wrote, rather than listing a path it guessed.

```python
for symbol in symbols:
    parts = storage.partitions(stored_keys[symbol])
    print(f"{symbol}: {len(parts)} partitions, {sum(p.size_bytes for p in parts) / 1024:.1f} KB")

print("\nAAPL partitions (first 8):")
for part in storage.partitions(stored_keys[DEMO_SYMBOL])[:8]:
    print(f"  {part.label}  ({part.size_bytes / 1024:.1f} KB)")
```

The two-year load lands as one Parquet file per calendar month, which is the
`partition_granularity="month"` setting above. Two things follow from that and the
chart below shows both. A date-range query reads only the months it overlaps, so the
partition-pruned read printed earlier touched a fraction of the files. And an
incremental update writes one new file rather than rewriting anything, because a month
that has closed never changes.

The near-uniform file sizes are the visible consequence: every month holds about the
same number of trading days, so no partition is large enough to dominate a read.

```python
aapl_sizes = pl.DataFrame(
    [
        {"period": part.label, "size_kb": part.size_bytes / 1024}
        for part in storage.partitions(stored_keys[DEMO_SYMBOL])
    ]
)

fig = go.Figure(
    go.Bar(
        x=aapl_sizes["period"].to_list(),
        y=aapl_sizes["size_kb"].to_list(),
        marker_color=COLORS["blue"],
    )
)
fig.update_layout(
    title=f"{DEMO_SYMBOL} Hive partitions, one per month",
    xaxis_title="Partition (year-month)",
    yaxis_title="Partition size (KB)",
    height=420,
    showlegend=False,
)
show_plotly_with_alt(
    fig,
    "A bar per calendar month showing the size in kilobytes of that month's Parquet "
    "partition. The bars are of similar height across the whole span, with small "
    "variation and no month standing out.",
)
```

---

## 4. Incremental Updates

The key workflow: download history once, then **update daily** with only new data.

### Update a Symbol

The delta is everything since the last stored bar, up to the newest bar the
vendor has actually published. That bound is not optional: Yahoo returns the
current exchange date as a row with accumulating volume and no
open/high/low/close, and the provider rejects a bar whose prices are null.
`DataManager.update()` fetches to `datetime.now()`, so it asks for that row on
every trading day. `update_through_last_complete_bar` asks for the same delta
and finds the end of the window instead of computing it - it steps back a day
at a time while the provider refuses the window, because the placeholder row
usually resolves a few hours after the close and sometimes does not. A refused
window is logged at error level, so an error line followed by a row count is
the retreat working rather than a failure.

```python
# Fetch every bar since the last stored one, up to the last complete session.
for symbol in symbols:
    rows = update_through_last_complete_bar(
        dm_stored, storage, symbol, provider=DEMO_PROVIDER, lookback_days=7
    )
    print(f"  Updated {symbol} → {rows} rows")

# Verify data is current
for symbol in symbols:
    meta = dm_stored.get_metadata(symbol)
    if meta and "last_updated" in meta:
        print(f"  {symbol} last updated: {meta['last_updated']}")
```

### Understanding Update Strategies

ml4t-data supports four update strategies:

| Strategy | Behavior | Use Case |
|----------|----------|----------|
| `INCREMENTAL` | Only fetch data after last stored timestamp | Daily updates (default) |
| `APPEND_ONLY` | Never modify existing rows | Audit-safe archives |
| `FULL_REFRESH` | Replace all data for the symbol | Recovery after corruption |
| `BACKFILL` | Fill gaps in historical data | Fix missing periods |

The default `INCREMENTAL` strategy is correct for most workflows.
`DataManager.update()` uses it automatically.

### Gap Detection

`exclude_weekends=True` stops Saturdays and Sundays counting as gaps, which is the only
part of the trading calendar a detector can infer without being handed one.

What it reports therefore depends on the shape of the series it is given, and the two
shapes give opposite answers. The cached series here is calendar-dense: every non-trading
day carries the prior close forward, so there is nothing missing to find and the detector
reports none. A sparse feed carrying only trading days would have the detector flag every
holiday, because a holiday and an outage are the same absence to anything without an
exchange calendar. Neither answer is wrong and neither is a coverage check on its own.

Before backtesting, verify data completeness. The IncrementalUpdater
can detect missing trading days.

```python
from ml4t.data.update_manager import GapDetector

gap_detector = GapDetector(exclude_weekends=True)

for symbol, key in stored_keys.items():
    df = storage.read(key).collect()
    gaps = gap_detector.detect_gaps(df, frequency="daily")
    if gaps:
        print(f"{symbol}: {len(gaps)} gap(s) detected")
        for gap in gaps[:3]:
            print(f"  {gap['start'].date()} to {gap['end'].date()} ({gap['size_days']} days)")
    else:
        print(f"{symbol}: No gaps (complete)")
```

---

## 5. Command-Line Interface

ml4t-data includes a CLI for scripted workflows and cron jobs.
Here are the key commands:

### Fetch Data
```bash
# Single symbol
ml4t-data fetch AAPL --start 2024-01-01 --end 2024-12-31

# Multiple symbols
ml4t-data fetch SPY QQQ IWM TLT --provider yahoo --output data/etfs.parquet
```

### Update Stored Data
```bash
# Update a symbol (incremental — only fetches new data)
ml4t-data update AAPL --storage-path ./data

# Update all stored symbols
ml4t-data update --all --storage-path ./data
```

### Validate Data Quality
```bash
# Run OHLCV validation on stored data
ml4t-data validate ./data/etfs.parquet
```

### List Available Data
```bash
# List symbols in storage
ml4t-data list --storage-path ./data

# List available providers
ml4t-data info --providers
```

### Automated Daily Updates (Cron)
```bash
# Daily at 6 PM EST (after US market close), Monday-Friday
0 18 * * 1-5 cd ~/ml4t && ml4t-data update --all --storage-path ./data >> logs/update.log 2>&1
```

---

## 6. Putting It Together: Production Data Pipeline

Here's the complete workflow combining everything above — the pattern
used by the book's `data/download_all.py` orchestrator.

```python
def production_pipeline(
    universe_name: str,
    start: str,
    end: str,
    storage_path: Path,
) -> pl.DataFrame:
    """Fetch, store, validate, and assemble a stacked DataFrame for a universe.

    The same pattern drives `data/download_all.py` for every asset class —
    only the universe and provider differ.
    """
    from ml4t.data.validation import OHLCVValidator

    symbols = Universe.get(universe_name)
    print(f"Universe '{universe_name}': {len(symbols)} symbols")

    config = StorageConfig(base_path=storage_path, compression="zstd")
    store = HiveStorage(config=config)
    manager = DataManager(storage=store, enable_validation=True)

    stored = {}
    for symbol in symbols:
        stored[symbol] = manager.load(symbol, start, end, provider=DEMO_PROVIDER)
    print(f"Fetched: {len(stored)} symbols")

    validator = OHLCVValidator(max_return_threshold=0.5)
    issues = 0
    for symbol, key in stored.items():
        df = store.read(key).collect()
        result = validator.validate(df)
        if not result.passed:
            issues += result.error_count
            print(f"  {symbol}: {result.error_count} validation issues")
    print(f"Validated: {issues} total issue(s) across {len(stored)} symbols")

    frames = [
        store.read(key).collect().with_columns(pl.lit(symbol).alias("symbol"))
        for symbol, key in stored.items()
    ]
    combined = pl.concat(frames)
    print(f"Result: {combined.shape[0]:,} rows, {combined['symbol'].n_unique()} symbols")
    return combined
```

```python
pipeline_output = production_pipeline(
    universe_name=PIPELINE_UNIVERSE,
    start=DEMO_START,
    end=DEMO_END,
    storage_path=STORAGE_DIR / "pipeline_demo",
)

pipeline_output.head()
```

The issue count printed above is the extreme-return check firing, not corrupt data. The
validator is configured with a maximum daily move, and on a one-year ETF panel the largest
move in each series is the one it flags. That is the check working: it surfaces candidates
and says nothing about which of them are defects.

Deciding what to do with a candidate is a separate step, and it belongs downstream where
the context is. `13_data_quality_framework` scores exactly this kind of flag against the
corporate-action columns of the same file, which is the difference between a candidate
list and a defect list.

---

## Summary

| Component | Purpose | Key Method |
|-----------|---------|------------|
| **DataManager** | Unified entry point | `fetch()`, `batch_load()`, `load()`, `update()` |
| **Universe** | Predefined symbol lists | `Universe.SP500`, `Universe.get("nasdaq100")` |
| **HiveStorage** | Partitioned Parquet | `read()`, `write()`, partition pruning |
| **GapDetector** | Gap detection in time series | `detect_gaps()`, `detect_gaps_in_storage()` |
| **CLI** | Scripted workflows & cron | `ml4t-data fetch`, `ml4t-data update` |

### The ml4t-data Workflow

```
1. Initial load:    dm.load("AAPL", "2020-01-01", "2024-12-31")
2. Daily update:    dm.update("AAPL", lookback_days=7)
3. Gap check:       gap_detector.detect_gaps(df, frequency="daily")
4. Batch load:      dm.batch_load_universe("sp500", start, end)
5. Automate:        cron + ml4t-data update --all
```

### Key Takeaways

- **One entry point, many providers.** `DataManager.fetch()` hides whether the
  bytes come from Yahoo, Binance, AlgoSeek, or local Hive parquet; the user
  code does not change when providers do.
- **`load()` is cache-first, `fetch()` is provider-first.** Use `load()` for
  research / backtesting (fast, offline, deterministic); use `fetch()` only
  when the cache must be refreshed.
- **Universes are first-class.** `Universe.SP500` and friends keep symbol
  lists out of notebook code and version-controlled in the library.
- **Gap detection is a separate concern.** `GapDetector` runs against
  already-stored data; missing trading days surface as findings, not silent
  nulls.
- **The CLI is the production surface.** Cron-driven `ml4t-data update --all`
  is the same code path the notebook exercises.

### Further Reading

- **Incremental updates**: `19_incremental_updates` walks the update strategies
  from this notebook in detail and shows how to schedule them.
- **Storage formats**: `20_storage_benchmark_file` compares Parquet, CSV, and HDF5;
  `21_storage_benchmark_database` benchmarks Postgres-backed alternatives.
- **Data quality**: `13_data_quality_framework` covers validation and anomaly detection.
- **Provider comparison**: `16_provider_comparison` demonstrates multi-source acquisition.
- **ml4t-data docs**: [ml4trading.io/docs/data/](https://ml4trading.io/docs/data/)
![notebook output](figures/p1_1.png)
![notebook output](figures/p1_2.png)