From df251a9d7e8ebdfce84b13a4319d582256083b29 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 23 Feb 2026 12:28:02 +0100 Subject: [PATCH] docs: add data pipeline and asset selection design MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- ...23-data-pipeline-asset-selection-design.md | 324 ++++++++++++++++++ 1 file changed, 324 insertions(+) create mode 100644 docs/plans/2026-02-23-data-pipeline-asset-selection-design.md diff --git a/docs/plans/2026-02-23-data-pipeline-asset-selection-design.md b/docs/plans/2026-02-23-data-pipeline-asset-selection-design.md new file mode 100644 index 000000000..321ed38a5 --- /dev/null +++ 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, // 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, // 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; + + /// Check what's already cached for a symbol + pub fn cached_ranges(&self, symbol: &str) -> Vec; + + /// Identify date gaps that need downloading + pub fn find_gaps(&self, symbol: &str, required: &DateRange) -> Vec; +} +``` + +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, + pub symbols: HashMap, +} + +pub struct SymbolCache { + pub ranges: Vec, // What date ranges are cached + pub bar_size: BarSize, + pub bar_count: usize, + pub file_paths: Vec, +} +``` + +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, +} + +impl PreparedDataset { + pub fn train_batches(&self, batch_size: usize) -> impl Iterator; + pub fn val_batches(&self, batch_size: usize) -> impl Iterator; + pub fn test_batches(&self, batch_size: usize) -> impl Iterator; +} +``` + +## 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, +} + +pub struct UniverseAsset { + pub symbol: String, + pub exchange: String, + pub asset_class: AssetClass, + pub sector: Option, + pub min_daily_volume: f64, + pub enabled: bool, +} + +impl AssetUniverse { + /// Load from config (TOML/JSON) + pub fn from_config(path: &Path) -> Result; + + /// 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>, + 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, +} + +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; + + /// 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; +} +``` + +## 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)