feat(store): LocalTieredStore — Tier-1 SQLite + Tier-2 Parquet (epic #540 phase 3b)

Second slice of Phase 3 of epic #540. Adds a real tiered storage
implementation that ships the analytics value (cross-run DuckDB /
Polars queries) without yet replacing the canonical .iafbt bundle.

Layout under <root>:
  index.sqlite                                    Tier-1 (always in sync)
  bundles/<handle>.iafbt                          canonical bytes
  parquet/portfolio_snapshots/run_id=<h>/...      Tier-2 hive-partitioned
  parquet/trades/run_id=<h>/...                   Tier-2
  parquet/orders/run_id=<h>/...                   Tier-2

Phase 3b deliberately keeps the bundle as the canonical representation;
Tier-2 sidecars are auxiliary, written best-effort, and a malformed
sidecar never blocks a write or a read. This trivially preserves
byte-identical Backtest round-trips today. Byte-identical
Tier-2 -> Backtest reassembly (no bundle on the read path) is Phase 3d.

- decompose.py: Backtest -> flat record lists for snapshots / trades /
  orders, adding run_id and window_name columns so downstream tools
  group cleanly across walk-forward windows. Extension point for
  metric_series and any future kind is the DATASETS tuple.

- LocalTieredStore: implements BacktestStore + SupportsCopyFrom.
  write() saves the bundle, upserts the Tier-1 row, and writes
  hive-partitioned Parquet sidecars per dataset. delete() removes all
  three tiers. iter_index_rows() serves from SQLite directly.
  rebuild_index() recreates Tier-1 from the bundles (useful after a
  software upgrade that adds new index columns).

- scan('portfolio_snapshots' | 'trades' | 'orders') returns a
  pyarrow.dataset.Dataset that DuckDB / Polars can query across every
  run with partition pruning on run_id.

- 15 new tests: Protocol + SupportsCopyFrom conformance, three-tier
  layout, handle normalisation, round-trip, summary_only, Tier-1
  always-in-sync (write/delete/len), Tier-2 cross-run scan, copy_from
  from LocalDirStore, rebuild_index, missing-handle errors. Includes
  a synthetic-records test that asserts hive partitions are written
  and that scan() returns the expected rows + columns.

Targeted suite (backtest_store + backtest_index + cli): 101 / 101 passing.

Author: MDUYN

investing_algorithm_framework/services/backtest_store/__init__.py

@@ -19,6 +19,7 @@
     SupportsCopyFrom,
 )
 from .local_dir_store import LocalDirStore
+from .local_tiered_store import LocalTieredStore
 
 __all__ = [
     "BacktestStore",
@@ -27,4 +28,5 @@
     "StoreHandleNotFoundError",
     "SupportsCopyFrom",
     "LocalDirStore",
+    "LocalTieredStore",
 ]

investing_algorithm_framework/services/backtest_store/decompose.py

@@ -0,0 +1,70 @@
+"""Backtest → flat-records decomposer for Tier-2 Parquet writes
+(epic #540 phase 3b).
+
+Given a :class:`Backtest`, yields flat record lists for each Tier-2
+dataset (``portfolio_snapshots`` / ``trades`` / ``orders``). The
+output is shaped for direct ingestion by
+:func:`pyarrow.dataset.write_dataset` with hive partitioning on
+``run_id``.
+
+The bundle (``.iafbt``) remains the canonical source of truth in
+Phase 3b; these Parquet datasets are *auxiliary* — they make
+DuckDB / Polars analytics work across thousands of runs without
+re-decoding bundles. Round-tripping Tier-2 back to a Backtest is
+Phase 3d territory.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Iterator, List
+
+from investing_algorithm_framework.domain import Backtest
+
+
+def _windowed_records(
+    backtest: Backtest, attr: str, run_id: str,
+) -> Iterator[Dict[str, Any]]:
+    """Yield ``to_dict()``-shaped records from every BacktestRun.
+
+    Adds ``run_id`` and ``window_name`` columns so downstream
+    columnar tools can group / partition cleanly across a Backtest's
+    walk-forward windows.
+    """
+    if not backtest.backtest_runs:
+        return
+    for window in backtest.backtest_runs:
+        items = getattr(window, attr, None) or []
+        window_name = getattr(window, "backtest_date_range_name", None) or (
+            window.create_directory_name()
+            if hasattr(window, "create_directory_name") else ""
+        )
+        for obj in items:
+            d = obj.to_dict() if hasattr(obj, "to_dict") else dict(obj)
+            d["run_id"] = run_id
+            d["window_name"] = window_name
+            yield d
+
+
+def snapshots(backtest: Backtest, run_id: str) -> List[Dict[str, Any]]:
+    """Flat portfolio_snapshots records, one per BacktestRun timestep."""
+    return list(_windowed_records(backtest, "portfolio_snapshots", run_id))
+
+
+def trades(backtest: Backtest, run_id: str) -> List[Dict[str, Any]]:
+    """Flat trades records (one per trade across all windows)."""
+    return list(_windowed_records(backtest, "trades", run_id))
+
+
+def orders(backtest: Backtest, run_id: str) -> List[Dict[str, Any]]:
+    """Flat orders records (one per order across all windows)."""
+    return list(_windowed_records(backtest, "orders", run_id))
+
+
+# Datasets exposed by LocalTieredStore. Each entry is
+# (dataset_name, decomposer_fn). Add new kinds here (e.g.
+# metric_series) once their decomposer is in place.
+DATASETS = (
+    ("portfolio_snapshots", snapshots),
+    ("trades", trades),
+    ("orders", orders),
+)