docs: add data pipeline and asset selection design

Covers automated data downloading/caching for ML training and a tiered asset selection funnel (universe → predictability → regime → signal). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-23 12:28:02 +01:00
parent f0bafdf30b
commit df251a9d7e
1 changed files with 324 additions and 0 deletions
--- a/docs/plans/2026-02-23-data-pipeline-asset-selection-design.md
+++ b/docs/plans/2026-02-23-data-pipeline-asset-selection-design.md
@@ -0,0 +1,324 @@
+# Data Pipeline & Asset Selection Design
+
+**Date**: 2026-02-23
+**Status**: Approved
+**Goal**: Automated data downloading/caching for ML training and a tiered asset selection funnel that operationalizes "trade what we're good at predicting."
+
+## Problem
+
+- Data downloading is manual (TLI commands, no automation)
+- No train/val/test split management — each training run re-downloads and re-processes
+- No mechanism to track which assets the ML ensemble predicts well (predictability scoring)
+- UniverseSelectionEngine exists (`ml/src/universe/`) but isn't wired into the training or trading pipeline
+- Batch sizes hardcoded per dataset, not tied to GPU capabilities (now fixed via gpu/ module)
+- Starting with limited capital — need to trade assets where prediction accuracy matters more than speed
+
+## Strategic Decisions
+
+- **Edge**: Prediction accuracy ("being right"), not latency
+- **Initial assets**: 5-10 US equities/ETFs (Databento data)
+- **Primary broker**: IBKR (fractional shares, direct exchange access, TWS API partially integrated)
+- **Secondary broker**: ICMarkets (cTrader OpenAPI, for future forex/CFD expansion — separate work)
+- **Data source**: Databento (exchange-level data matches IBKR execution — no train/execute mismatch)
+- **Broker integration**: Out of scope for this design (broker-agnostic)
+
+## Approach: Data Pipeline + Tiered Asset Selection
+
+Two new modules, each doing one thing:
+
+```
+ml/src/data_pipeline/     — Automated data downloading, caching, train/test splits
+ml/src/asset_selection/   — Tiered funnel: universe → predictability → regime → signal
+```
+
+Both are broker-agnostic — they work with market data symbols and produce trading candidates.
+
+## Part 1: Data Pipeline
+
+### Module Structure
+
+```
+ml/src/data_pipeline/
+├── mod.rs              // DatasetSpec, DatasetMode, re-exports
+├── manager.rs          // DatasetManager — download, cache, gap detection
+├── cache.rs            // Local Parquet cache with manifest
+└── prepared.rs         // PreparedDataset — train/val/test iterators
+```
+
+### DatasetSpec — what data you need
+
+```rust
+pub struct DatasetSpec {
+    pub name: String,                   // "us-equities-dev"
+    pub symbols: Vec<SymbolSpec>,       // Symbol + exchange + asset class
+    pub date_range: DateRange,          // Absolute or Rolling
+    pub bar_size: BarSize,              // OneMinute, FiveMinute, Daily
+    pub split_ratio: SplitRatio,        // default 70/15/15
+    pub mode: DatasetMode,             // Dev / Backtest / Full
+}
+
+pub enum DatasetMode {
+    Dev,        // Last 30 days — fast iteration, ~40K bars per symbol
+    Backtest,   // Last 6 months — proper regime diversity
+    Full,       // 2+ years — production training
+}
+
+pub struct SymbolSpec {
+    pub symbol: String,                // "AAPL", "SPY"
+    pub exchange: String,              // "XNAS", "XNYS"
+    pub asset_class: AssetClass,       // Equity, ETF, Future
+}
+
+pub enum AssetClass {
+    Equity,
+    ETF,
+    Future,
+}
+
+pub struct SplitRatio {
+    pub train: f64,   // 0.70
+    pub val: f64,     // 0.15
+    pub test: f64,    // 0.15
+}
+```
+
+### DatasetManager — orchestrates downloads and caching
+
+```rust
+pub struct DatasetManager {
+    cache_dir: PathBuf,           // data/cache/
+    acquisition_client: Option<DataAcquisitionClient>,  // gRPC to data_acquisition_service
+}
+
+impl DatasetManager {
+    /// Main entry point: ensure data is downloaded, cached, and ready
+    pub async fn prepare(&self, spec: &DatasetSpec) -> Result<PreparedDataset, MLError>;
+
+    /// Check what's already cached for a symbol
+    pub fn cached_ranges(&self, symbol: &str) -> Vec<DateRange>;
+
+    /// Identify date gaps that need downloading
+    pub fn find_gaps(&self, symbol: &str, required: &DateRange) -> Vec<DateRange>;
+}
+```
+
+Internally:
+1. Check local cache manifest for each symbol
+2. Identify date gaps
+3. Call `data_acquisition_service` gRPC to download missing data
+4. Run feature extraction (reuses existing `parquet_utils.rs`)
+5. Split into train/val/test
+6. Return `PreparedDataset`
+
+### Cache — local Parquet storage with manifest
+
+```rust
+pub struct CacheManifest {
+    pub last_updated: DateTime<Utc>,
+    pub symbols: HashMap<String, SymbolCache>,
+}
+
+pub struct SymbolCache {
+    pub ranges: Vec<DateRange>,         // What date ranges are cached
+    pub bar_size: BarSize,
+    pub bar_count: usize,
+    pub file_paths: Vec<PathBuf>,
+}
+```
+
+Cache layout:
+```
+data/cache/
+├── manifest.json
+├── AAPL/
+│   ├── 2024-01_1m.parquet
+│   ├── 2024-02_1m.parquet
+│   └── ...
+├── SPY/
+│   └── ...
+└── features/
+    ├── AAPL_128d.parquet
+    └── SPY_128d.parquet
+```
+
+### PreparedDataset — ready for training
+
+```rust
+pub struct PreparedDataset {
+    pub spec: DatasetSpec,
+    pub train: DataSlice,
+    pub val: DataSlice,
+    pub test: DataSlice,
+    pub feature_dim: usize,            // 128
+    pub total_bars: usize,
+}
+
+pub struct DataSlice {
+    pub bars: usize,
+    pub symbols: Vec<String>,
+}
+
+impl PreparedDataset {
+    pub fn train_batches(&self, batch_size: usize) -> impl Iterator<Item = Batch>;
+    pub fn val_batches(&self, batch_size: usize) -> impl Iterator<Item = Batch>;
+    pub fn test_batches(&self, batch_size: usize) -> impl Iterator<Item = Batch>;
+}
+```
+
+## Part 2: Asset Selection
+
+### Module Structure
+
+```
+ml/src/asset_selection/
+├── mod.rs              // AssetUniverse, TradingCandidate re-exports
+├── universe.rs         // Static universe definition + config
+├── scorer.rs           // PredictabilityScorer + composite scoring
+└── selector.rs         // ActiveSetSelector — final funnel stage
+```
+
+### Three-Tier Funnel
+
+```
+AssetUniverse (5-10 configured assets)
+   ↓ filter by enabled + min volume
+Eligible Assets
+   ↓ PredictabilityScorer (rolling ML accuracy per asset)
+   ↓ + liquidity, regime_fit scoring
+Scored Assets
+   ↓ ActiveSetSelector (top N by composite score)
+Active Trading Set (3-5 assets)
+   ↓ Ensemble prediction per active asset
+   ↓ Conviction gates (from operational maturity plan)
+Executed Trades (0-3)
+```
+
+### Tier 1: Universe Definition
+
+```rust
+pub struct AssetUniverse {
+    pub assets: Vec<UniverseAsset>,
+}
+
+pub struct UniverseAsset {
+    pub symbol: String,
+    pub exchange: String,
+    pub asset_class: AssetClass,
+    pub sector: Option<String>,
+    pub min_daily_volume: f64,
+    pub enabled: bool,
+}
+
+impl AssetUniverse {
+    /// Load from config (TOML/JSON)
+    pub fn from_config(path: &Path) -> Result<Self, MLError>;
+
+    /// Get enabled assets
+    pub fn eligible(&self) -> Vec<&UniverseAsset>;
+}
+```
+
+### Tier 2: Asset Scoring
+
+```rust
+pub struct AssetScore {
+    pub symbol: String,
+    pub predictability: f64,     // Rolling ML accuracy (0.0-1.0)
+    pub liquidity: f64,          // Normalized liquidity score
+    pub regime_fit: f64,         // How well current regime suits this asset
+    pub composite: f64,          // Weighted combination
+}
+
+pub struct PredictabilityScorer {
+    history: HashMap<String, VecDeque<PredictionResult>>,
+    window_size: usize,          // 30 trading days default
+}
+
+pub struct PredictionResult {
+    pub symbol: String,
+    pub predicted_direction: f64,  // Signal from ensemble
+    pub actual_direction: f64,     // Realized return
+    pub timestamp: DateTime<Utc>,
+}
+
+impl PredictabilityScorer {
+    /// Record a prediction outcome
+    pub fn record(&mut self, result: PredictionResult);
+
+    /// Get rolling accuracy for a symbol
+    pub fn accuracy(&self, symbol: &str) -> Option<f64>;
+
+    /// Score all tracked symbols
+    pub fn scores(&self) -> Vec<(String, f64)>;
+}
+```
+
+Composite score formula:
+```
+composite = w_pred * predictability + w_liq * liquidity + w_regime * regime_fit
+```
+
+Default weights: predictability=0.5, liquidity=0.3, regime_fit=0.2 (predictability dominates because "being right" is the edge).
+
+### Tier 3: Active Set Selection
+
+```rust
+pub struct ActiveSetSelector {
+    pub max_active: usize,              // 3-5
+    pub min_score_threshold: f64,       // Don't trade below this
+    pub rebalance_interval: Duration,   // Daily default
+}
+
+pub struct TradingCandidate {
+    pub symbol: String,
+    pub score: AssetScore,
+    pub dataset: DatasetSpec,           // What data to use for this asset
+}
+
+impl ActiveSetSelector {
+    /// Select top N assets by composite score
+    pub fn select(&self, scores: &[AssetScore]) -> Vec<TradingCandidate>;
+}
+```
+
+## Integration Points
+
+| From | To | How |
+|------|-----|-----|
+| DataPipeline → AssetSelection | `DatasetManager::prepare()` uses `AssetUniverse` symbols | Shared SymbolSpec |
+| AssetSelection → Ensemble | `ActiveSetSelector` feeds `EnsembleCoordinator` | Only predict on active assets |
+| AssetSelection → ConvictionGates | Gates are the final filter (operational maturity plan) | Sequential pipeline |
+| PredictabilityScorer → QuestDB | Historical accuracy from QuestDB time-series | Query rolling window |
+| DataPipeline → GPU | `PreparedDataset` uses `GpuCapabilities` for batch sizing | Already implemented |
+
+## Per-Mode Data Estimates
+
+| Mode | Duration | Bars/Symbol (1m) | Symbols | Total Bars | Disk |
+|------|----------|-------------------|---------|------------|------|
+| Dev | 30 days | ~10K | 5 | ~50K | ~20MB |
+| Backtest | 6 months | ~60K | 5 | ~300K | ~120MB |
+| Full | 2 years | ~240K | 10 | ~2.4M | ~1GB |
+
+## Fallback Behavior
+
+1. `data_acquisition_service` unavailable → use cached data, warn if stale
+2. No cached data for a symbol → skip symbol, reduce active set
+3. PredictabilityScorer has no history → equal scoring (all assets start equal)
+4. All assets score below threshold → no trading (stay flat)
+
+## What This Does NOT Include
+
+- Broker-specific execution routing (IBKR/ICMarkets — separate work)
+- Real-time streaming data ingestion (this is batch/historical)
+- Multi-timeframe feature engineering (future enhancement)
+- Cross-asset correlation portfolio optimization (after initial results)
+- Forex/CFD data sources (ICMarkets expansion — future)
+
+## Testing
+
+- Unit tests with synthetic data (no Databento API calls needed)
+- DatasetManager with mock gRPC client
+- PredictabilityScorer with known prediction histories
+- ActiveSetSelector with deterministic scores
+- Cache manifest serialization/deserialization
+- All tests run on CPU (no GPU required for CI)