docs: implementation plan for feature extraction pipeline fix
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
@@ -0,0 +1,746 @@
|
||||
# Feature Extraction Pipeline Fix
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** Make the fxcache data pipeline correct, strict, and consistent. One shared discovery function, strict key matching, symbol + data_source in the cache key, explicit symbol parameter on trainer API.
|
||||
|
||||
**Architecture:** Bottom-up: fix cache key first (foundation), then shared discovery function, then update all callers (trainer API, CLI binaries, tests). Each task produces a compiling, testable codebase.
|
||||
|
||||
**Tech Stack:** Rust, SHA256 (sha2 crate), cargo test
|
||||
|
||||
---
|
||||
|
||||
## File Map
|
||||
|
||||
| File | Change |
|
||||
|------|--------|
|
||||
| `crates/ml/src/feature_cache.rs` | Add `symbol` + `data_source` to `calculate_dbn_cache_key_full()` |
|
||||
| `crates/ml/src/fxcache.rs` | Add `has_ofi` to `FxCacheData` + header, add `discover_and_load()` |
|
||||
| `crates/ml/src/trainers/dqn/trainer/mod.rs` | `train(data_dir, symbol, cb)`, `load_training_data(data_dir, symbol)` |
|
||||
| `crates/ml/src/trainers/dqn/data_loading.rs` | Replace inline discovery with `discover_and_load()`, delete fallback |
|
||||
| `crates/ml/src/hyperopt/adapters/dqn.rs` | Replace `preload_data()` discovery, pass symbol |
|
||||
| `crates/ml/examples/train_baseline_rl.rs` | Replace inline discovery with `discover_and_load()` |
|
||||
| `crates/ml/examples/hyperopt_baseline_rl.rs` | Pass symbol to discovery |
|
||||
| `crates/ml/examples/precompute_features.rs` | Pass symbol + data_source to cache key, set `has_ofi` |
|
||||
| `crates/ml/tests/dqn_training_pipeline_test.rs` | Pass symbol to `train()` |
|
||||
| `crates/ml/tests/dqn_early_stopping_termination_test.rs` | Pass symbol to `train()` |
|
||||
| 10+ other test files | Add symbol arg to `train()` calls |
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Add symbol + data_source to cache key
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/ml/src/feature_cache.rs`
|
||||
|
||||
The cache key must include symbol and data_source so different instruments / modes get different cache files.
|
||||
|
||||
- [ ] **Step 1: Update `calculate_dbn_cache_key_full` signature**
|
||||
|
||||
In `crates/ml/src/feature_cache.rs`, change the function signature (line 24-28) and add symbol + data_source to the hash:
|
||||
|
||||
```rust
|
||||
pub fn calculate_dbn_cache_key_full(
|
||||
data_dir: &Path,
|
||||
mbp10_dir: Option<&Path>,
|
||||
trades_dir: Option<&Path>,
|
||||
symbol: &str,
|
||||
data_source: &str,
|
||||
) -> Result<String> {
|
||||
let mut hasher = Sha256::new();
|
||||
|
||||
// Include symbol and data_source in hash — different instruments/modes get different keys
|
||||
hasher.update(symbol.as_bytes());
|
||||
hasher.update(b"|");
|
||||
hasher.update(data_source.as_bytes());
|
||||
hasher.update(b"|");
|
||||
|
||||
// ... rest of existing hash logic unchanged ...
|
||||
```
|
||||
|
||||
Also update the convenience wrapper `calculate_dbn_cache_key` (line 16-18):
|
||||
|
||||
```rust
|
||||
pub fn calculate_dbn_cache_key(data_dir: &Path) -> Result<String> {
|
||||
calculate_dbn_cache_key_full(data_dir, None, None, "ES.FUT", "ohlcv")
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Write test for symbol differentiation**
|
||||
|
||||
Add at the bottom of `feature_cache.rs` (or in a separate test module):
|
||||
|
||||
```rust
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::path::Path;
|
||||
|
||||
#[test]
|
||||
fn test_cache_key_includes_symbol() {
|
||||
let dir = Path::new("test_data/futures-baseline");
|
||||
if !dir.exists() { return; } // skip if no test data
|
||||
let key_es = calculate_dbn_cache_key_full(dir, None, None, "ES.FUT", "ohlcv").unwrap();
|
||||
let key_nq = calculate_dbn_cache_key_full(dir, None, None, "NQ.FUT", "ohlcv").unwrap();
|
||||
assert_ne!(key_es, key_nq, "Different symbols must produce different cache keys");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cache_key_includes_data_source() {
|
||||
let dir = Path::new("test_data/futures-baseline");
|
||||
if !dir.exists() { return; }
|
||||
let key_ohlcv = calculate_dbn_cache_key_full(dir, None, None, "ES.FUT", "ohlcv").unwrap();
|
||||
let key_mbp10 = calculate_dbn_cache_key_full(dir, None, None, "ES.FUT", "mbp10").unwrap();
|
||||
assert_ne!(key_ohlcv, key_mbp10, "Different data sources must produce different cache keys");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cache_key_stable() {
|
||||
let dir = Path::new("test_data/futures-baseline");
|
||||
if !dir.exists() { return; }
|
||||
let key1 = calculate_dbn_cache_key_full(dir, None, None, "ES.FUT", "ohlcv").unwrap();
|
||||
let key2 = calculate_dbn_cache_key_full(dir, None, None, "ES.FUT", "ohlcv").unwrap();
|
||||
assert_eq!(key1, key2, "Same inputs must produce same key");
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Fix all callers to pass symbol + data_source**
|
||||
|
||||
Every caller of `calculate_dbn_cache_key_full` needs updating. There are 5 call sites:
|
||||
|
||||
1. `crates/ml/examples/precompute_features.rs:214` — pass `&opts.symbol`, `"ohlcv"` (or from opts)
|
||||
2. `crates/ml/examples/precompute_features.rs:394` — same
|
||||
3. `crates/ml/examples/train_baseline_rl.rs:574` — pass `&args.symbol`, `"ohlcv"`
|
||||
4. `crates/ml/src/trainers/dqn/data_loading.rs:160` — pass `&self.hyperparams.symbol`, `&self.hyperparams.data_source`
|
||||
5. `crates/ml/src/hyperopt/adapters/dqn.rs:821` — pass symbol from `self.dbn_data_dir` context, data_source from hyperparams
|
||||
|
||||
For each, add the two new arguments. The compiler will flag any missed call sites.
|
||||
|
||||
- [ ] **Step 4: Verify compilation + run tests**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check --workspace
|
||||
SQLX_OFFLINE=true cargo test -p ml --lib -- feature_cache
|
||||
```
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add crates/ml/src/feature_cache.rs crates/ml/examples/precompute_features.rs crates/ml/examples/train_baseline_rl.rs crates/ml/src/trainers/dqn/data_loading.rs crates/ml/src/hyperopt/adapters/dqn.rs
|
||||
git commit -m "feat: cache key includes symbol + data_source (prevents cross-instrument collisions)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: Add `has_ofi` to FxCacheData + header
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/ml/src/fxcache.rs`
|
||||
|
||||
Use one byte from the `reserved` field to store `has_ofi` flag. No backward compat needed.
|
||||
|
||||
- [ ] **Step 1: Add `has_ofi` field to `FxCacheData`**
|
||||
|
||||
In `crates/ml/src/fxcache.rs`, add to the struct (after line 200):
|
||||
|
||||
```rust
|
||||
pub struct FxCacheData {
|
||||
pub timestamps: Vec<i64>,
|
||||
pub features: Vec<[f64; FEAT_DIM]>,
|
||||
pub targets: Vec<[f64; TARGET_DIM]>,
|
||||
pub ofi: Vec<[f64; OFI_DIM]>,
|
||||
pub cache_key: [u8; 32],
|
||||
pub bar_count: usize,
|
||||
/// Explicit flag: true if OFI was computed from real MBP-10 data.
|
||||
/// False means OFI is zero-filled (no MBP-10 data was available during precompute).
|
||||
pub has_ofi: bool,
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Store `has_ofi` in header reserved[0]**
|
||||
|
||||
Update `FxCacheHeader::new()` to accept `has_ofi`:
|
||||
|
||||
```rust
|
||||
pub fn new(version: u16, bar_count: u64, cache_key: [u8; 32], has_ofi: bool) -> Self {
|
||||
let mut reserved = [0u8; 8];
|
||||
reserved[0] = if has_ofi { 1 } else { 0 };
|
||||
Self {
|
||||
magic: FXCACHE_MAGIC,
|
||||
version,
|
||||
feat_dim: FEAT_DIM as u16,
|
||||
target_dim: TARGET_DIM as u16,
|
||||
ofi_dim: OFI_DIM as u16,
|
||||
bar_count,
|
||||
cache_key,
|
||||
reserved,
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Read `has_ofi` from header on load**
|
||||
|
||||
In `load_fxcache()`, after parsing the header, extract `has_ofi`:
|
||||
|
||||
```rust
|
||||
let has_ofi = header.reserved[0] == 1;
|
||||
```
|
||||
|
||||
And set it on the returned `FxCacheData`:
|
||||
|
||||
```rust
|
||||
Ok(FxCacheData {
|
||||
timestamps, features, targets, ofi,
|
||||
cache_key: header.cache_key,
|
||||
bar_count,
|
||||
has_ofi,
|
||||
})
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Update `write_fxcache()` to accept `has_ofi`**
|
||||
|
||||
Add `has_ofi: bool` parameter to `write_fxcache()`:
|
||||
|
||||
```rust
|
||||
pub fn write_fxcache(
|
||||
path: &Path,
|
||||
features: &[[f64; FEAT_DIM]],
|
||||
targets: &[[f64; TARGET_DIM]],
|
||||
ofi: &[[f64; OFI_DIM]],
|
||||
timestamps: &[i64],
|
||||
cache_key: [u8; 32],
|
||||
bf16: bool,
|
||||
has_ofi: bool,
|
||||
) -> Result<u64> {
|
||||
```
|
||||
|
||||
And pass it to `FxCacheHeader::new(version, bar_count as u64, cache_key, has_ofi)`.
|
||||
|
||||
- [ ] **Step 5: Update all `write_fxcache` callers**
|
||||
|
||||
The main caller is `precompute_features.rs`. Find the `write_fxcache` call and add `has_ofi`:
|
||||
|
||||
```rust
|
||||
// has_ofi = true if mbp10_dir was provided and files were found
|
||||
let has_ofi = mbp10_dir.is_some();
|
||||
ml::fxcache::write_fxcache(&output_path, &features, &targets, &ofi, ×tamps, cache_key, bf16, has_ofi)?;
|
||||
```
|
||||
|
||||
- [ ] **Step 6: Update all `FxCacheData` constructors**
|
||||
|
||||
Search for `FxCacheData {` — there's one in `train_baseline_rl.rs` (DBN fallback path, line 627). Add `has_ofi: false` (DBN fallback has no MBP-10).
|
||||
|
||||
- [ ] **Step 7: Write test for has_ofi round-trip**
|
||||
|
||||
```rust
|
||||
#[test]
|
||||
fn test_has_ofi_roundtrip() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("test.fxcache");
|
||||
let features = vec![[1.0_f64; 42]; 10];
|
||||
let targets = vec![[0.0_f64; 4]; 10];
|
||||
let ofi = vec![[0.5_f64; 8]; 10];
|
||||
let timestamps = vec![1_i64; 10];
|
||||
let key = [0u8; 32];
|
||||
|
||||
// Write with has_ofi=true
|
||||
write_fxcache(&path, &features, &targets, &ofi, ×tamps, key, false, true).unwrap();
|
||||
let loaded = load_fxcache(&path).unwrap();
|
||||
assert!(loaded.has_ofi, "has_ofi should be true");
|
||||
|
||||
// Write with has_ofi=false
|
||||
let path2 = dir.path().join("test2.fxcache");
|
||||
write_fxcache(&path2, &features, &targets, &ofi, ×tamps, key, false, false).unwrap();
|
||||
let loaded2 = load_fxcache(&path2).unwrap();
|
||||
assert!(!loaded2.has_ofi, "has_ofi should be false");
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 8: Verify + commit**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check --workspace
|
||||
SQLX_OFFLINE=true cargo test -p ml --lib -- fxcache
|
||||
git add crates/ml/src/fxcache.rs crates/ml/examples/precompute_features.rs crates/ml/examples/train_baseline_rl.rs
|
||||
git commit -m "feat: has_ofi flag in fxcache header (explicit, no zero-detection)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: Create shared `discover_and_load()` function
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/ml/src/fxcache.rs`
|
||||
|
||||
This is the single source of truth for fxcache discovery. All callers will use this instead of inline logic.
|
||||
|
||||
- [ ] **Step 1: Implement `discover_and_load()`**
|
||||
|
||||
Add to `crates/ml/src/fxcache.rs`:
|
||||
|
||||
```rust
|
||||
/// Discover and load an fxcache file. Single source of truth for all callers.
|
||||
///
|
||||
/// Cache dir priority: `cache_dir_override` > `FOXHUNT_FEATURE_CACHE_DIR` env > walk-up sibling.
|
||||
/// Strict key match only — returns `None` on miss (no "most recent" fallback).
|
||||
pub fn discover_and_load(
|
||||
data_dir: &Path,
|
||||
symbol: &str,
|
||||
mbp10_dir: Option<&Path>,
|
||||
trades_dir: Option<&Path>,
|
||||
data_source: &str,
|
||||
cache_dir_override: Option<&Path>,
|
||||
) -> Option<FxCacheData> {
|
||||
// 1. Resolve cache directory
|
||||
let cache_dir = resolve_cache_dir(data_dir, cache_dir_override)?;
|
||||
|
||||
// 2. Compute cache key (includes symbol + data_source)
|
||||
let key_hex = crate::feature_cache::calculate_dbn_cache_key_full(
|
||||
data_dir, mbp10_dir, trades_dir, symbol, data_source,
|
||||
).ok()?;
|
||||
let key: [u8; 32] = hex::decode(&key_hex).ok()?.try_into().ok()?;
|
||||
|
||||
// 3. Strict key match — no fallback
|
||||
let path = find_fxcache(&cache_dir, &key)?;
|
||||
|
||||
// 4. Load
|
||||
match load_fxcache(&path) {
|
||||
Ok(data) => {
|
||||
info!("fxcache hit: {} bars, OFI={} from {:?}", data.bar_count, data.has_ofi, path);
|
||||
Some(data)
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("fxcache load failed: {e}");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve fxcache directory: explicit override > env var > walk-up sibling.
|
||||
fn resolve_cache_dir(data_dir: &Path, override_dir: Option<&Path>) -> Option<PathBuf> {
|
||||
// Explicit override
|
||||
if let Some(dir) = override_dir {
|
||||
if dir.exists() {
|
||||
return Some(dir.to_path_buf());
|
||||
}
|
||||
}
|
||||
|
||||
// Environment variable
|
||||
if let Ok(dir) = std::env::var("FOXHUNT_FEATURE_CACHE_DIR") {
|
||||
let p = PathBuf::from(dir);
|
||||
if p.exists() {
|
||||
return Some(p);
|
||||
}
|
||||
}
|
||||
|
||||
// Walk up from data_dir to find sibling feature-cache/
|
||||
let mut dir = data_dir;
|
||||
loop {
|
||||
if let Some(parent) = dir.parent() {
|
||||
let candidate = parent.join("feature-cache");
|
||||
if candidate.exists() {
|
||||
return Some(candidate);
|
||||
}
|
||||
if parent == dir { break; }
|
||||
dir = parent;
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
None
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Write tests for discover_and_load**
|
||||
|
||||
```rust
|
||||
#[cfg(test)]
|
||||
mod discover_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_discover_cache_miss_no_dir() {
|
||||
let result = discover_and_load(
|
||||
Path::new("/nonexistent"),
|
||||
"ES.FUT",
|
||||
None, None,
|
||||
"ohlcv",
|
||||
None,
|
||||
);
|
||||
assert!(result.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_discover_cache_miss_wrong_key() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let cache_dir = dir.path().join("feature-cache");
|
||||
std::fs::create_dir_all(&cache_dir).unwrap();
|
||||
// Write a cache file with a known key
|
||||
let path = cache_dir.join("0000000000000000000000000000000000000000000000000000000000000000.fxcache");
|
||||
let features = vec![[1.0_f64; 42]; 5];
|
||||
let targets = vec![[0.0_f64; 4]; 5];
|
||||
let ofi = vec![[0.0_f64; 8]; 5];
|
||||
let timestamps = vec![1_i64; 5];
|
||||
write_fxcache(&path, &features, &targets, &ofi, ×tamps, [0u8; 32], false, false).unwrap();
|
||||
|
||||
// Try to discover with a different data_dir (different key)
|
||||
let result = discover_and_load(
|
||||
Path::new("test_data/futures-baseline"),
|
||||
"ES.FUT",
|
||||
None, None,
|
||||
"ohlcv",
|
||||
Some(&cache_dir),
|
||||
);
|
||||
// Should be None — strict match, no fallback
|
||||
assert!(result.is_none(), "Wrong key should not match");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_discover_explicit_dir_override() {
|
||||
// If cache_dir_override is set but doesn't exist, falls through
|
||||
let result = discover_and_load(
|
||||
Path::new("test_data/futures-baseline"),
|
||||
"ES.FUT",
|
||||
None, None,
|
||||
"ohlcv",
|
||||
Some(Path::new("/nonexistent/cache")),
|
||||
);
|
||||
assert!(result.is_none());
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Verify + commit**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo test -p ml --lib -- fxcache
|
||||
git add crates/ml/src/fxcache.rs
|
||||
git commit -m "feat: shared discover_and_load() — strict key match, no fallback"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 4: Update `DQNTrainer::train()` and `load_training_data()` signatures
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/ml/src/trainers/dqn/trainer/mod.rs`
|
||||
- Modify: `crates/ml/src/trainers/dqn/data_loading.rs`
|
||||
|
||||
- [ ] **Step 1: Change `train()` signature**
|
||||
|
||||
In `crates/ml/src/trainers/dqn/trainer/mod.rs`, change `train()` (line 431):
|
||||
|
||||
```rust
|
||||
pub async fn train<F>(
|
||||
&mut self,
|
||||
dbn_data_dir: &str,
|
||||
symbol: &str,
|
||||
checkpoint_callback: F,
|
||||
) -> Result<TrainingMetrics>
|
||||
```
|
||||
|
||||
Update the body to pass `symbol` to `load_training_data`:
|
||||
|
||||
```rust
|
||||
let (training_data, val_data) = self.load_training_data(dbn_data_dir, symbol).await?;
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Change `load_training_data()` signature**
|
||||
|
||||
In `data_loading.rs` (line 93):
|
||||
|
||||
```rust
|
||||
pub async fn load_training_data(
|
||||
&mut self,
|
||||
dbn_data_dir: &str,
|
||||
symbol: &str,
|
||||
) -> Result<(Vec<(FeatureVector, Vec<f64>)>, Vec<(FeatureVector, Vec<f64>)>)>
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Replace inline fxcache discovery with `discover_and_load()`**
|
||||
|
||||
Replace lines 100-200 in `data_loading.rs` with:
|
||||
|
||||
```rust
|
||||
// ── fxcache discovery (strict key match, no fallback) ────────────────
|
||||
let cache_dir_override = self.feature_cache_dir.as_deref();
|
||||
let mbp10_path = if self.hyperparams.mbp10_data_dir.is_empty() {
|
||||
None
|
||||
} else {
|
||||
let p = Path::new(&self.hyperparams.mbp10_data_dir);
|
||||
if p.exists() { Some(p) } else { None }
|
||||
};
|
||||
let trades_path = if self.hyperparams.trades_data_dir.is_empty() {
|
||||
None
|
||||
} else {
|
||||
let p = Path::new(&self.hyperparams.trades_data_dir);
|
||||
if p.exists() { Some(p) } else { None }
|
||||
};
|
||||
|
||||
if let Some(cached) = crate::fxcache::discover_and_load(
|
||||
Path::new(dbn_data_dir),
|
||||
symbol,
|
||||
mbp10_path,
|
||||
trades_path,
|
||||
&self.hyperparams.data_source,
|
||||
cache_dir_override,
|
||||
) {
|
||||
// Set OFI from cache if present
|
||||
if cached.has_ofi {
|
||||
self.ofi_features = Some(std::sync::Arc::from(cached.ofi));
|
||||
}
|
||||
|
||||
// Combine features + targets
|
||||
let all_data: Vec<(FeatureVector, Vec<f64>)> = cached.features
|
||||
.into_iter()
|
||||
.zip(cached.targets.into_iter())
|
||||
.map(|(f, t)| {
|
||||
let fv: FeatureVector = f.to_vec();
|
||||
(fv, t.to_vec())
|
||||
})
|
||||
.collect();
|
||||
|
||||
let split = (all_data.len() * 80) / 100;
|
||||
let train = all_data[..split].to_vec();
|
||||
let val = all_data[split..].to_vec();
|
||||
return Ok((train, val));
|
||||
}
|
||||
|
||||
// ── DBN fallback: load bars scoped to symbol ─────────────────────────
|
||||
let symbol_data_dir = Path::new(dbn_data_dir).join(symbol);
|
||||
let effective_dir = if symbol_data_dir.exists() {
|
||||
info!("Symbol filter: loading {} from {}", symbol, symbol_data_dir.display());
|
||||
symbol_data_dir.to_string_lossy().to_string()
|
||||
} else {
|
||||
dbn_data_dir.to_string()
|
||||
};
|
||||
```
|
||||
|
||||
Then the rest of the existing DBN loading code uses `&effective_dir` instead of `dbn_data_dir`.
|
||||
|
||||
- [ ] **Step 4: Delete the dangerous "most recent" fallback**
|
||||
|
||||
The old lines 167-181 (`if fxcache_path.is_none() { ... most recent ... }`) are gone — replaced by the strict `discover_and_load()` call above.
|
||||
|
||||
- [ ] **Step 5: Update `load_training_data` caller in `helpers.rs`**
|
||||
|
||||
In `crates/ml/src/trainers/dqn/smoke_tests/helpers.rs` line 124:
|
||||
|
||||
```rust
|
||||
let (train, val) = rt.block_on(trainer.load_training_data(&data_dir, "ES.FUT"))?;
|
||||
```
|
||||
|
||||
- [ ] **Step 6: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p ml
|
||||
```
|
||||
|
||||
This will show all callers that need updating (compiler errors). Don't fix them yet — that's Task 5.
|
||||
|
||||
- [ ] **Step 7: Commit**
|
||||
|
||||
```bash
|
||||
git add crates/ml/src/trainers/dqn/trainer/mod.rs crates/ml/src/trainers/dqn/data_loading.rs crates/ml/src/trainers/dqn/smoke_tests/helpers.rs
|
||||
git commit -m "feat: train(data_dir, symbol, cb) + strict fxcache discovery in load_training_data"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 5: Update all `train()` callers to pass symbol
|
||||
|
||||
**Files:**
|
||||
- Modify: ~17 test files + hyperopt adapter
|
||||
|
||||
This is mechanical. Every `.train(&data_dir, callback)` becomes `.train(&data_dir, "ES.FUT", callback)`.
|
||||
|
||||
- [ ] **Step 1: Fix all test callers**
|
||||
|
||||
Run `cargo check -p ml --tests` and fix each error. The pattern for every test:
|
||||
|
||||
```rust
|
||||
// Before:
|
||||
trainer.train(&data_dir, |epoch, data, is_best| { ... }).await
|
||||
// After:
|
||||
trainer.train(&data_dir, "ES.FUT", |epoch, data, is_best| { ... }).await
|
||||
```
|
||||
|
||||
Files to update (from grep):
|
||||
- `dqn_training_pipeline_test.rs` — 6 call sites
|
||||
- `dqn_early_stopping_termination_test.rs` — 3 call sites
|
||||
- `dqn_gradient_accumulation_test.rs` — 1
|
||||
- `dqn_long_training_test.rs` — 1
|
||||
- `production_training_smoke_test.rs` — 1
|
||||
- `dqn_accumulation_convergence_test.rs` — 2
|
||||
- `dqn_inference_test.rs` — 1
|
||||
- `smoke_test_real_data.rs` — 1
|
||||
|
||||
- [ ] **Step 2: Fix hyperopt adapter**
|
||||
|
||||
In `crates/ml/src/hyperopt/adapters/dqn.rs:2229`:
|
||||
|
||||
```rust
|
||||
// Before:
|
||||
handle.block_on(internal_trainer.train(data_path_str, checkpoint_callback))
|
||||
// After:
|
||||
let symbol = &self.hyperparams.symbol;
|
||||
handle.block_on(internal_trainer.train(data_path_str, symbol, checkpoint_callback))
|
||||
```
|
||||
|
||||
Also update `preload_data()` discovery (lines 794-925) to use `discover_and_load()`:
|
||||
|
||||
```rust
|
||||
let cached = crate::fxcache::discover_and_load(
|
||||
data_dir_path,
|
||||
&self.hyperparams.symbol,
|
||||
mbp10.as_deref(),
|
||||
trades.as_deref(),
|
||||
&self.hyperparams.data_source,
|
||||
self.feature_cache_dir.as_deref(),
|
||||
);
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Verify full workspace compiles + run tests**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check --workspace --tests
|
||||
SQLX_OFFLINE=true cargo test -p ml --lib -- feature_cache fxcache training_profile
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add -A
|
||||
git commit -m "refactor: all train() callers pass explicit symbol"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 6: Update `train_baseline_rl.rs` to use shared discovery
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/ml/examples/train_baseline_rl.rs`
|
||||
|
||||
Replace the ~40-line inline discovery (lines 545-589) with a single call.
|
||||
|
||||
- [ ] **Step 1: Replace inline discovery**
|
||||
|
||||
Replace lines 545-589 with:
|
||||
|
||||
```rust
|
||||
// 1. Try fxcache first, fall back to DBN loading + feature extraction
|
||||
info!("Step 1/5: Loading data...");
|
||||
let data_load_start = std::time::Instant::now();
|
||||
|
||||
let cache_dir_override = args.feature_cache_dir.as_ref().map(|s| PathBuf::from(s));
|
||||
let mbp10 = args.mbp10_data_dir.as_ref().filter(|p| p.exists());
|
||||
let trades = args.trades_data_dir.as_ref().filter(|p| p.exists());
|
||||
|
||||
let fxcache_data = ml::fxcache::discover_and_load(
|
||||
&args.data_dir,
|
||||
&args.symbol,
|
||||
mbp10.map(|p| p.as_path()),
|
||||
trades.map(|p| p.as_path()),
|
||||
"ohlcv",
|
||||
cache_dir_override.as_deref(),
|
||||
);
|
||||
```
|
||||
|
||||
The rest of the loading code (fxcache hit → use it, miss → DBN fallback) stays the same.
|
||||
|
||||
- [ ] **Step 2: Verify + commit**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p ml --examples
|
||||
git add crates/ml/examples/train_baseline_rl.rs
|
||||
git commit -m "refactor: train_baseline_rl uses shared fxcache::discover_and_load()"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 7: Update `precompute_features.rs` to set `has_ofi`
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/ml/examples/precompute_features.rs`
|
||||
|
||||
- [ ] **Step 1: Pass `has_ofi` and updated cache key args**
|
||||
|
||||
At the `write_fxcache` call site, add `has_ofi`:
|
||||
|
||||
```rust
|
||||
let has_ofi = mbp10_dir.is_some();
|
||||
ml::fxcache::write_fxcache(
|
||||
&output_path, &features, &targets, &ofi, ×tamps,
|
||||
cache_key, bf16, has_ofi,
|
||||
)?;
|
||||
```
|
||||
|
||||
At both `calculate_dbn_cache_key_full` calls (lines 214 and 394), add symbol + data_source:
|
||||
|
||||
```rust
|
||||
let hex_key = ml::feature_cache::calculate_dbn_cache_key_full(
|
||||
&data_dir,
|
||||
mbp10_dir.as_deref(),
|
||||
trades_dir.as_deref(),
|
||||
&opts.symbol,
|
||||
"ohlcv", // precompute always uses ohlcv source
|
||||
)?;
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Verify + commit**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p ml --examples
|
||||
git add crates/ml/examples/precompute_features.rs
|
||||
git commit -m "feat: precompute sets has_ofi flag + symbol/data_source in cache key"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 8: Full verification — all tests pass
|
||||
|
||||
- [ ] **Step 1: Run unit tests**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo test -p ml --lib -- feature_cache fxcache training_profile
|
||||
```
|
||||
|
||||
Expected: all pass (cache key tests, has_ofi roundtrip, discover tests).
|
||||
|
||||
- [ ] **Step 2: Run integration tests (release mode, one at a time)**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo test -p ml --release --test dqn_early_stopping_termination_test -- --nocapture
|
||||
SQLX_OFFLINE=true FOXHUNT_TEST_DATA=test_data/futures-baseline cargo test -p ml --release --test dqn_training_pipeline_test -- --nocapture
|
||||
SQLX_OFFLINE=true cargo test -p ml --release --test dqn_action_collapse_fix_test -- --nocapture
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Run precompute → train round-trip**
|
||||
|
||||
```bash
|
||||
# Precompute
|
||||
SQLX_OFFLINE=true cargo run -p ml --example precompute_features --release -- \
|
||||
--data-dir test_data/futures-baseline \
|
||||
--mbp10-data-dir test_data/futures-baseline-mbp10 \
|
||||
--trades-data-dir test_data/futures-baseline-trades \
|
||||
--output-dir /tmp/fxcache-verify \
|
||||
--symbol ES.FUT --yes
|
||||
|
||||
# Train (should hit cache, OFI=true)
|
||||
SQLX_OFFLINE=true cargo run -p ml --example train_baseline_rl --release -- \
|
||||
--model dqn --data-dir test_data/futures-baseline \
|
||||
--mbp10-data-dir test_data/futures-baseline-mbp10 \
|
||||
--trades-data-dir test_data/futures-baseline-trades \
|
||||
--feature-cache-dir /tmp/fxcache-verify \
|
||||
--symbol ES.FUT --epochs 1 --output-dir /tmp/train-verify
|
||||
```
|
||||
|
||||
Verify: "fxcache hit" + "OFI=true" in output.
|
||||
|
||||
- [ ] **Step 4: Push**
|
||||
|
||||
```bash
|
||||
git push origin main
|
||||
```
|
||||
Reference in New Issue
Block a user