gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(ml): DST-aware sessions, normalize ensemble weights, filter non-finite predictions

Browse Source 
- Use chrono-tz America/New_York for correct EST/EDT trading session
  boundaries (was hardcoded UTC-5, off by 1h during daylight saving)
- Normalize effective weights to sum to 1.0 before signal aggregation
  (raw weights could sum to anything, biasing the ensemble)
- Skip adapter predictions that return NaN/Inf direction or confidence
  instead of letting them poison the weighted average

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-23 23:03:20 +01:00
parent 17e60a48de
commit 5deb618864
1 changed files with 443 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

468
ml/src/ensemble/coordinator.rs
View File

@@ -10,7 +10,8 @@ use crate::ensemble::conviction_gates::{
use crate::ensemble::inference_adapter::{FeatureVector, ModelInferenceAdapter};
use crate::ensemble::{EnsembleDecision, ModelVote, ModelWeight, TradingAction};
use crate::{Features, MLError, MLResult, ModelPrediction};
use chrono::{Timelike, Utc};
use chrono::{DateTime, TimeZone, Timelike, Utc};
use chrono_tz::America::New_York;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
@@ -235,6 +236,16 @@ impl EnsembleCoordinator {
};
match adapter.predict(&fv) {
Ok(ensemble_pred) => {
if !ensemble_pred.direction.is_finite()
|| !ensemble_pred.confidence.is_finite()
{
warn!(
"Adapter {} returned non-finite prediction \
(direction={}, confidence={}), skipping",
model_id, ensemble_pred.direction, ensemble_pred.confidence
);
continue;
}
let prediction = ModelPrediction::new(
model_id.clone(),
ensemble_pred.direction,
@@ -377,12 +388,18 @@ impl EnsembleCoordinator {
agreeing as f64 / decision.model_votes.len() as f64
}
/// Determine current trading session (Eastern Time, simplified UTC-5)
/// Determine current trading session (Eastern Time, DST-aware)
fn current_trading_session() -> TradingSession {
let now = Utc::now();
let et_hour = (now.hour() + 24 - 5) % 24;
let et_minute = now.minute();
let et_time = et_hour * 60 + et_minute;
Self::trading_session_at(Utc::now())
}
/// Determine trading session for an arbitrary UTC timestamp.
///
/// Uses `chrono-tz` America/New_York so the boundaries are correct
/// during both EST (UTC-5, Nov-Mar) and EDT (UTC-4, Mar-Nov).
fn trading_session_at(utc_time: DateTime<Utc>) -> TradingSession {
let et = utc_time.with_timezone(&New_York);
let et_time = et.hour() * 60 + et.minute();
match et_time {
t if t < 240 => TradingSession::AfterHours, // 00:00-04:00 ET
@@ -513,12 +530,17 @@ impl SignalAggregator {
});
}
// Normalize effective weights so they sum to 1.0 before any calculations.
// Raw effective_weight = static_weight * dynamic_weight can range 0.5-1.5
// per model, so the sum across models is not guaranteed to be 1.0.
let normalized = Self::normalize_effective_weights(&predictions, weights);
// Calculate weighted average signal
let (weighted_signal, _total_weight) =
self.calculate_weighted_signal(&predictions, weights);
self.calculate_weighted_signal(&predictions, &normalized);
// Calculate ensemble confidence
let confidence = self.calculate_ensemble_confidence(&predictions, weights);
let confidence = self.calculate_ensemble_confidence(&predictions, &normalized);
// Calculate disagreement rate
let disagreement_rate = self.calculate_disagreement_rate(&predictions);
@@ -526,8 +548,8 @@ impl SignalAggregator {
// Determine trading action
let action = TradingAction::from_signal(weighted_signal, self.signal_threshold);
// Build model votes
let model_votes = self.build_model_votes(&predictions, weights);
// Build model votes (uses normalized weights)
let model_votes = self.build_model_votes(&predictions, &normalized);
let decision = EnsembleDecision::new(
action,
@@ -540,20 +562,62 @@ impl SignalAggregator {
Ok(decision)
}
/// Calculate weighted average signal
/// Normalize effective weights across all predictions so they sum to 1.0.
///
/// Models without an explicit weight entry receive equal share (1/N).
/// If total weight is zero or non-finite, all models get uniform 1/N weight.
fn normalize_effective_weights(
predictions: &[ModelPrediction],
weights: &HashMap<String, ModelWeight>,
) -> HashMap<String, f64> {
let n = predictions.len();
let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 };
// Collect raw effective weights
let raw: Vec<(String, f64)> = predictions
.iter()
.map(|pred| {
let w = weights
.get(&pred.model_id)
.map(|mw| mw.effective_weight())
.unwrap_or(uniform);
(pred.model_id.clone(), w)
})
.collect();
let total: f64 = raw.iter().map(|(_, w)| w).sum();
let mut normalized = HashMap::new();
if total > 0.0 && total.is_finite() {
for (id, w) in raw {
normalized.insert(id, w / total);
}
} else {
// Fallback: uniform weights
for (id, _) in raw {
normalized.insert(id, uniform);
}
}
normalized
}
/// Calculate weighted average signal using pre-normalized weights
fn calculate_weighted_signal(
&self,
predictions: &[ModelPrediction],
weights: &HashMap<String, ModelWeight>,
normalized_weights: &HashMap<String, f64>,
) -> (f64, f64) {
let n = predictions.len();
let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 };
let mut weighted_sum = 0.0;
let mut total_weight = 0.0;
for pred in predictions {
let weight = weights
let weight = normalized_weights
.get(&pred.model_id)
.map(|w| w.effective_weight())
.unwrap_or(1.0 / predictions.len() as f64);
.copied()
.unwrap_or(uniform);
weighted_sum += pred.value * pred.confidence * weight;
total_weight += weight * pred.confidence;
@@ -568,20 +632,22 @@ impl SignalAggregator {
(signal, total_weight)
}
/// Calculate ensemble confidence
/// Calculate ensemble confidence using pre-normalized weights
fn calculate_ensemble_confidence(
&self,
predictions: &[ModelPrediction],
weights: &HashMap<String, ModelWeight>,
normalized_weights: &HashMap<String, f64>,
) -> f64 {
let n = predictions.len();
let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 };
let mut confidence_sum = 0.0;
let mut weight_sum = 0.0;
for pred in predictions {
let weight = weights
let weight = normalized_weights
.get(&pred.model_id)
.map(|w| w.effective_weight())
.unwrap_or(1.0 / predictions.len() as f64);
.copied()
.unwrap_or(uniform);
confidence_sum += pred.confidence * weight;
weight_sum += weight;
@@ -613,19 +679,21 @@ impl SignalAggregator {
disagreements as f64 / predictions.len() as f64
}
/// Build model votes map
/// Build model votes map using pre-normalized weights
fn build_model_votes(
&self,
predictions: &[ModelPrediction],
weights: &HashMap<String, ModelWeight>,
normalized_weights: &HashMap<String, f64>,
) -> HashMap<String, ModelVote> {
let n = predictions.len();
let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 };
let mut votes = HashMap::new();
for pred in predictions {
let weight = weights
let weight = normalized_weights
.get(&pred.model_id)
.map(|w| w.effective_weight())
.unwrap_or(1.0 / predictions.len() as f64);
.copied()
.unwrap_or(uniform);
let vote = ModelVote::new(pred.model_id.clone(), pred.value, pred.confidence, weight);
@@ -879,6 +947,184 @@ mod tests {
assert_eq!(decision.model_count(), 1);
}
#[tokio::test]
async fn test_ensemble_filters_nan_inf_predictions() {
use crate::ensemble::inference_adapter::{
EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta,
};
struct NanAdapter;
impl ModelInferenceAdapter for NanAdapter {
fn model_name(&self) -> &str {
"NAN_MODEL"
}
fn predict(
&self,
_features: &FeatureVector,
) -> crate::MLResult<EnsemblePrediction> {
Ok(EnsemblePrediction {
model_name: "NAN_MODEL".to_string(),
direction: f64::NAN,
confidence: 0.9,
metadata: PredictionMeta::default(),
})
}
fn is_ready(&self) -> bool {
true
}
}
unsafe impl Send for NanAdapter {}
unsafe impl Sync for NanAdapter {}
struct InfAdapter;
impl ModelInferenceAdapter for InfAdapter {
fn model_name(&self) -> &str {
"INF_MODEL"
}
fn predict(
&self,
_features: &FeatureVector,
) -> crate::MLResult<EnsemblePrediction> {
Ok(EnsemblePrediction {
model_name: "INF_MODEL".to_string(),
direction: 0.5,
confidence: f64::INFINITY,
metadata: PredictionMeta::default(),
})
}
fn is_ready(&self) -> bool {
true
}
}
unsafe impl Send for InfAdapter {}
unsafe impl Sync for InfAdapter {}
struct GoodAdapter;
impl ModelInferenceAdapter for GoodAdapter {
fn model_name(&self) -> &str {
"GOOD_MODEL"
}
fn predict(
&self,
_features: &FeatureVector,
) -> crate::MLResult<EnsemblePrediction> {
Ok(EnsemblePrediction {
model_name: "GOOD_MODEL".to_string(),
direction: 0.6,
confidence: 0.85,
metadata: PredictionMeta::default(),
})
}
fn is_ready(&self) -> bool {
true
}
}
unsafe impl Send for GoodAdapter {}
unsafe impl Sync for GoodAdapter {}
let mut coordinator = EnsembleCoordinator::new();
coordinator.add_adapter(Box::new(NanAdapter));
coordinator.add_adapter(Box::new(InfAdapter));
coordinator.add_adapter(Box::new(GoodAdapter));
coordinator
.register_model("NAN_MODEL".to_string(), 0.33)
.await
.unwrap();
coordinator
.register_model("INF_MODEL".to_string(), 0.33)
.await
.unwrap();
coordinator
.register_model("GOOD_MODEL".to_string(), 0.34)
.await
.unwrap();
let features = Features::new(
vec![0.5; 5],
vec!["f1", "f2", "f3", "f4", "f5"]
.into_iter()
.map(String::from)
.collect(),
);
let decision = coordinator.predict(&features).await.unwrap();
// Only the good model should survive filtering
assert_eq!(
decision.model_count(),
1,
"NaN and Inf predictions should be filtered out"
);
assert!(
decision.confidence.is_finite(),
"Ensemble confidence must be finite"
);
assert!(
decision.signal.is_finite(),
"Ensemble signal must be finite"
);
}
#[tokio::test]
async fn test_ensemble_all_nan_returns_error() {
use crate::ensemble::inference_adapter::{
EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta,
};
struct AllNanAdapter {
name: &'static str,
}
impl ModelInferenceAdapter for AllNanAdapter {
fn model_name(&self) -> &str {
self.name
}
fn predict(
&self,
_features: &FeatureVector,
) -> crate::MLResult<EnsemblePrediction> {
Ok(EnsemblePrediction {
model_name: self.name.to_string(),
direction: f64::NAN,
confidence: f64::NAN,
metadata: PredictionMeta::default(),
})
}
fn is_ready(&self) -> bool {
true
}
}
unsafe impl Send for AllNanAdapter {}
unsafe impl Sync for AllNanAdapter {}
let mut coordinator = EnsembleCoordinator::new();
coordinator.add_adapter(Box::new(AllNanAdapter { name: "A" }));
coordinator.add_adapter(Box::new(AllNanAdapter { name: "B" }));
coordinator
.register_model("A".to_string(), 0.5)
.await
.unwrap();
coordinator
.register_model("B".to_string(), 0.5)
.await
.unwrap();
let features = Features::new(
vec![0.5; 5],
vec!["f1", "f2", "f3", "f4", "f5"]
.into_iter()
.map(String::from)
.collect(),
);
let result = coordinator.predict(&features).await;
assert!(
result.is_err(),
"All-NaN predictions should produce an error"
);
}
#[tokio::test]
async fn test_ensemble_rejects_no_adapters() {
let coordinator = EnsembleCoordinator::new();
@@ -907,4 +1153,176 @@ mod tests {
err_msg
);
}
#[tokio::test]
async fn test_normalized_weights_sum_to_one() {
let aggregator = SignalAggregator::new();
let predictions = vec![
ModelPrediction::new("DQN".to_string(), 0.8, 0.9),
ModelPrediction::new("PPO".to_string(), 0.7, 0.85),
ModelPrediction::new("TFT".to_string(), 0.6, 0.8),
];
// Create weights with dynamic adjustments that cause unnormalized sums.
// DQN: high perf => dynamic ~1.2, effective = 0.5 * 1.2 = 0.6
// PPO: low perf => dynamic ~0.6, effective = 0.3 * 0.6 = 0.18
// TFT: default => dynamic 1.0, effective = 0.2 * 1.0 = 0.2
// Raw sum = 0.98, NOT 1.0
let mut weights = HashMap::new();
let mut dqn_w = ModelWeight::new("DQN".to_string(), 0.5);
dqn_w.performance_metrics.sharpe_ratio = 2.0;
dqn_w.performance_metrics.accuracy = 0.7;
dqn_w.update_dynamic_weight();
weights.insert("DQN".to_string(), dqn_w);
let mut ppo_w = ModelWeight::new("PPO".to_string(), 0.3);
ppo_w.performance_metrics.sharpe_ratio = 0.3;
ppo_w.performance_metrics.accuracy = 0.3;
ppo_w.update_dynamic_weight();
weights.insert("PPO".to_string(), ppo_w);
let mut tft_w = ModelWeight::new("TFT".to_string(), 0.2);
// Keep default performance (sharpe=1.0, accuracy=0.5) => dynamic ~0.95
tft_w.update_dynamic_weight();
weights.insert("TFT".to_string(), tft_w);
// Verify raw effective weights do NOT sum to 1.0
let raw_sum: f64 = weights.values().map(|w| w.effective_weight()).sum();
assert!(
(raw_sum - 1.0).abs() > 0.01,
"Raw effective weights should NOT sum to 1.0 (got {}), \
otherwise the test does not exercise normalization",
raw_sum
);
// After aggregation, model_votes weights must sum to 1.0
let decision = aggregator
.aggregate(predictions, &weights)
.await
.ok()
.expect("aggregation should succeed");
let vote_weight_sum: f64 = decision.model_votes.values().map(|v| v.weight).sum();
assert!(
(vote_weight_sum - 1.0).abs() < 1e-10,
"Normalized vote weights must sum to 1.0, got {}",
vote_weight_sum
);
// Each individual weight must be positive
for (model_id, vote) in &decision.model_votes {
assert!(
vote.weight > 0.0,
"Weight for {} must be positive, got {}",
model_id,
vote.weight
);
}
}
#[test]
fn test_normalize_effective_weights_uniform_fallback() {
// When weights map is empty, all models get 1/N
let predictions = vec![
ModelPrediction::new("A".to_string(), 0.5, 0.8),
ModelPrediction::new("B".to_string(), 0.3, 0.7),
];
let weights = HashMap::new();
let normalized = SignalAggregator::normalize_effective_weights(&predictions, &weights);
assert_eq!(normalized.len(), 2);
let sum: f64 = normalized.values().sum();
assert!(
(sum - 1.0).abs() < 1e-10,
"Uniform fallback weights must sum to 1.0, got {}",
sum
);
for (_, w) in &normalized {
assert!(
(*w - 0.5).abs() < 1e-10,
"Each weight should be 0.5 for 2 models, got {}",
w
);
}
}
#[test]
fn test_normalize_effective_weights_single_model() {
let predictions = vec![ModelPrediction::new("SOLO".to_string(), 0.9, 0.95)];
let mut weights = HashMap::new();
weights.insert(
"SOLO".to_string(),
ModelWeight::new("SOLO".to_string(), 0.7),
);
let normalized = SignalAggregator::normalize_effective_weights(&predictions, &weights);
assert_eq!(normalized.len(), 1);
let w = normalized.get("SOLO").copied().unwrap_or(0.0);
assert!(
(w - 1.0).abs() < 1e-10,
"Single model must get weight 1.0, got {}",
w
);
}
/// Verify DST-aware trading session boundaries.
///
/// June 15 2026 is during EDT (UTC-4), so 13:30 UTC = 09:30 ET (Regular).
/// The old hardcoded UTC-5 would have mapped 13:30 UTC to 08:30 ET (PreMarket) -- wrong.
#[test]
fn test_trading_session_dst_correctness() {
// --- EDT (summer): UTC-4 ---
// June 15 2026 13:30 UTC => 09:30 ET => Regular
let june_1330 = Utc.with_ymd_and_hms(2026, 6, 15, 13, 30, 0).unwrap();
assert_eq!(
EnsembleCoordinator::trading_session_at(june_1330),
TradingSession::Regular,
"13:30 UTC in June (EDT) should be 09:30 ET = Regular, not PreMarket"
);
// June 15 2026 20:00 UTC => 16:00 ET => AfterHours
let june_2000 = Utc.with_ymd_and_hms(2026, 6, 15, 20, 0, 0).unwrap();
assert_eq!(
EnsembleCoordinator::trading_session_at(june_2000),
TradingSession::AfterHours,
"20:00 UTC in June (EDT) should be 16:00 ET = AfterHours"
);
// June 15 2026 08:00 UTC => 04:00 ET => PreMarket
let june_0800 = Utc.with_ymd_and_hms(2026, 6, 15, 8, 0, 0).unwrap();
assert_eq!(
EnsembleCoordinator::trading_session_at(june_0800),
TradingSession::PreMarket,
"08:00 UTC in June (EDT) should be 04:00 ET = PreMarket"
);
// --- EST (winter): UTC-5 ---
// January 15 2026 14:30 UTC => 09:30 ET => Regular
let jan_1430 = Utc.with_ymd_and_hms(2026, 1, 15, 14, 30, 0).unwrap();
assert_eq!(
EnsembleCoordinator::trading_session_at(jan_1430),
TradingSession::Regular,
"14:30 UTC in January (EST) should be 09:30 ET = Regular"
);
// January 15 2026 21:00 UTC => 16:00 ET => AfterHours
let jan_2100 = Utc.with_ymd_and_hms(2026, 1, 15, 21, 0, 0).unwrap();
assert_eq!(
EnsembleCoordinator::trading_session_at(jan_2100),
TradingSession::AfterHours,
"21:00 UTC in January (EST) should be 16:00 ET = AfterHours"
);
// January 15 2026 03:00 UTC => 22:00 ET (prev day) => AfterHours
let jan_0300 = Utc.with_ymd_and_hms(2026, 1, 15, 3, 0, 0).unwrap();
assert_eq!(
EnsembleCoordinator::trading_session_at(jan_0300),
TradingSession::AfterHours,
"03:00 UTC in January (EST) should be 22:00 ET = AfterHours"
);
}
}