From 3f9f6ebe17ca783671f25bfda12bee4db26a5e5d Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 23 Feb 2026 14:13:19 +0100 Subject: [PATCH] fix(trading): fix QuestDB metrics provider for real integration testing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Use tokio::task::block_in_place for safe sync→async bridging - Replace count(*) with count(1) in division context (QuestDB parser bug) - Add coalesce() for stddev NULL handling (single-row edge case) - Consolidate integration tests into single lifecycle test with DROP+CREATE - All 4 QuestDB tests pass against real QuestDB 8.2.3 Co-Authored-By: Claude Opus 4.6 --- .../trading_service/src/questdb_metrics.rs | 157 ++++++++++-------- 1 file changed, 92 insertions(+), 65 deletions(-) diff --git a/services/trading_service/src/questdb_metrics.rs b/services/trading_service/src/questdb_metrics.rs index 578f1890f..4aae8a2e0 100644 --- a/services/trading_service/src/questdb_metrics.rs +++ b/services/trading_service/src/questdb_metrics.rs @@ -109,7 +109,7 @@ impl QuestDBMetricsProvider { /// Query 30-day rolling Sharpe per model async fn query_model_sharpe(&self) -> HashMap { let result: Result, _> = sqlx::query_as( - "SELECT model_id, avg(signal) / (stddev(signal) + 1e-10) as sharpe + "SELECT model_id, coalesce(avg(signal) / (stddev(signal) + 1e-10), 0.0) as sharpe FROM model_predictions WHERE timestamp > dateadd('d', -30, now()) GROUP BY model_id" @@ -130,7 +130,7 @@ impl QuestDBMetricsProvider { async fn query_model_win_rates(&self) -> HashMap { let result: Result, _> = sqlx::query_as( "SELECT model_id, - sum(CASE WHEN signal_alignment > 0 THEN 1.0 ELSE 0.0 END) / count(*) as win_rate, + sum(CASE WHEN signal_alignment > 0 THEN 1.0 ELSE 0.0 END) / count(1) as win_rate, count(*) as trade_count FROM trade_outcomes WHERE timestamp > dateadd('d', -30, now()) @@ -157,7 +157,7 @@ impl QuestDBMetricsProvider { let result: Result, _> = sqlx::query_as( "SELECT confidence_bucket_lower, confidence_bucket_upper, - sum(CASE WHEN signal_alignment > 0 THEN 1.0 ELSE 0.0 END) / count(*) as win_rate, + sum(CASE WHEN signal_alignment > 0 THEN 1.0 ELSE 0.0 END) / count(1) as win_rate, count(*) as trade_count, avg(realized_pnl) as avg_pnl FROM trade_outcomes @@ -208,49 +208,56 @@ impl QuestDBMetricsProvider { impl MetricsProvider for QuestDBMetricsProvider { fn get_model_metrics(&self) -> Vec { - // Use tokio::runtime::Handle to run async queries synchronously - // This is acceptable since the feedback loop runs periodically (24h) + // block_in_place allows blocking inside a multi-threaded tokio runtime + // by moving other tasks off this thread. Safe because the feedback loop + // runs periodically (24h cycle) — not on the hot path. let handle = tokio::runtime::Handle::current(); - let sharpes = handle.block_on(self.query_model_sharpe()); - let win_rates = handle.block_on(self.query_model_win_rates()); + tokio::task::block_in_place(|| { + let sharpes = handle.block_on(self.query_model_sharpe()); + let win_rates = handle.block_on(self.query_model_win_rates()); - let mut metrics = Vec::new(); - for (model_id, sharpe) in &sharpes { - let (win_rate, trade_count) = win_rates - .get(model_id) - .copied() - .unwrap_or((0.5, 0)); + let mut metrics = Vec::new(); + for (model_id, sharpe) in &sharpes { + let (win_rate, trade_count) = win_rates + .get(model_id) + .copied() + .unwrap_or((0.5, 0)); - metrics.push(ModelRollingMetrics { - model_id: model_id.clone(), - sharpe_30d: *sharpe, - win_rate_30d: win_rate, - prediction_accuracy: win_rate, // Using win_rate as proxy - trade_count, - deployed_at: Instant::now() - std::time::Duration::from_secs(30 * 24 * 3600), - }); - } - metrics + metrics.push(ModelRollingMetrics { + model_id: model_id.clone(), + sharpe_30d: *sharpe, + win_rate_30d: win_rate, + prediction_accuracy: win_rate, // Using win_rate as proxy + trade_count, + deployed_at: Instant::now() - std::time::Duration::from_secs(30 * 24 * 3600), + }); + } + metrics + }) } fn get_gate_buckets(&self) -> Vec { let handle = tokio::runtime::Handle::current(); - handle.block_on(self.query_confidence_buckets()) + tokio::task::block_in_place(|| handle.block_on(self.query_confidence_buckets())) } fn get_current_weights(&self) -> HashMap { let handle = tokio::runtime::Handle::current(); - handle.block_on(async { self.current_weights.read().await.clone() }) + tokio::task::block_in_place(|| { + handle.block_on(async { self.current_weights.read().await.clone() }) + }) } fn get_gate_config(&self) -> ConvictionGateConfig { let handle = tokio::runtime::Handle::current(); - handle.block_on(async { self.gate_config.read().await.clone() }) + tokio::task::block_in_place(|| { + handle.block_on(async { self.gate_config.read().await.clone() }) + }) } fn get_ensemble_sharpe_7d(&self) -> f64 { let handle = tokio::runtime::Handle::current(); - handle.block_on(self.query_ensemble_sharpe()) + tokio::task::block_in_place(|| handle.block_on(self.query_ensemble_sharpe())) } fn get_model_correlations(&self) -> Vec<((String, String), f64)> { @@ -264,48 +271,65 @@ impl MetricsProvider for QuestDBMetricsProvider { mod tests { use super::*; - /// Integration test that requires QuestDB running on localhost:8812. - /// Run with: docker compose up questdb -d - /// Then: SQLX_OFFLINE=true cargo test -p trading_service --lib -- questdb_metrics --ignored - #[tokio::test] - #[ignore = "Requires QuestDB on localhost:8812"] - async fn test_questdb_provider_connects() { - let provider = QuestDBMetricsProvider::new("postgresql://admin:quest@localhost:8812/qdb") - .await - .expect("QuestDB should be running"); - - provider.ensure_tables().await.expect("Tables should be created"); - - // With empty tables, metrics should return empty/default - let metrics = provider.get_model_metrics(); - assert!(metrics.is_empty()); - - let buckets = provider.get_gate_buckets(); - assert!(buckets.is_empty()); - - let sharpe = provider.get_ensemble_sharpe_7d(); - assert!((sharpe - 0.0).abs() < 1e-10); + /// Drop and recreate all test tables for clean isolation between runs. + /// QuestDB TRUNCATE via PG wire can be flaky; DROP + CREATE is reliable. + async fn reset_tables(pool: &PgPool) { + for table in &["model_predictions", "trade_outcomes", "ensemble_metrics"] { + let drop_q = format!("DROP TABLE IF EXISTS {table};"); + let _ = sqlx::query(&drop_q).execute(pool).await; + } } - /// Integration test that writes data and reads it back. - #[tokio::test] + /// Poll QuestDB until a table has at least `min_rows` rows (up to 5s). + async fn wait_for_rows(pool: &PgPool, table: &str, min_rows: i64) { + for _ in 0..50 { + let query = format!("SELECT count(*) FROM {table}"); + let row: Option<(i64,)> = sqlx::query_as(&query) + .fetch_optional(pool) + .await + .ok() + .flatten(); + if row.map(|(c,)| c).unwrap_or(0) >= min_rows { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + } + + /// Comprehensive integration test: connect, write, read, verify metrics. + /// Single test avoids parallel interference on shared QuestDB tables. + /// + /// Run with: docker-compose up -d questdb + /// Then: SQLX_OFFLINE=true cargo test -p trading_service --lib -- questdb_metrics --include-ignored + #[tokio::test(flavor = "multi_thread")] #[ignore = "Requires QuestDB on localhost:8812"] - async fn test_questdb_provider_reads_written_data() { + async fn test_questdb_provider_full_lifecycle() { let provider = QuestDBMetricsProvider::new("postgresql://admin:quest@localhost:8812/qdb") .await .expect("QuestDB should be running"); + // Drop + recreate for clean slate (prior test runs may have left data) + reset_tables(&provider.pool).await; provider.ensure_tables().await.expect("Tables should be created"); - // Write some test data + // Phase 1: Empty tables — metrics should return defaults + let metrics = provider.get_model_metrics(); + assert!(metrics.is_empty(), "Empty tables should yield no metrics"); + + let sharpe = provider.get_ensemble_sharpe_7d(); + assert!((sharpe - 0.0).abs() < 1e-10, "Empty ensemble Sharpe should be 0.0"); + + // Phase 2: Write test data (multiple rows per model for valid stddev) sqlx::query( "INSERT INTO model_predictions(model_id, signal, confidence, prediction_accuracy, win_rate, timestamp) VALUES ('dqn', 0.8, 0.7, 0.62, 0.58, systimestamp()), - ('ppo', 0.3, 0.6, 0.55, 0.52, systimestamp())" + ('dqn', 0.6, 0.8, 0.65, 0.60, systimestamp()), + ('ppo', 0.3, 0.6, 0.55, 0.52, systimestamp()), + ('ppo', 0.5, 0.7, 0.58, 0.54, systimestamp())" ) .execute(&provider.pool) .await - .expect("Insert should succeed"); + .expect("Insert predictions should succeed"); sqlx::query( "INSERT INTO trade_outcomes(model_id, realized_pnl, signal_alignment, confidence_bucket_lower, confidence_bucket_upper, timestamp) @@ -315,7 +339,7 @@ mod tests { ) .execute(&provider.pool) .await - .expect("Insert should succeed"); + .expect("Insert outcomes should succeed"); sqlx::query( "INSERT INTO ensemble_metrics(sharpe_7d, sharpe_30d, total_pnl, timestamp) @@ -323,37 +347,40 @@ mod tests { ) .execute(&provider.pool) .await - .expect("Insert should succeed"); + .expect("Insert ensemble metrics should succeed"); - // Read back - model metrics + // Wait for QuestDB WAL to commit all tables + wait_for_rows(&provider.pool, "model_predictions", 4).await; + wait_for_rows(&provider.pool, "trade_outcomes", 3).await; + wait_for_rows(&provider.pool, "ensemble_metrics", 1).await; + + // Phase 3: Read back and verify let metrics = provider.get_model_metrics(); - assert!(!metrics.is_empty(), "Should have model metrics"); + assert!(!metrics.is_empty(), "Should have model metrics after insert"); + assert!(metrics.len() >= 2, "Should have at least DQN and PPO"); - // Read back - ensemble Sharpe let sharpe = provider.get_ensemble_sharpe_7d(); - assert!((sharpe - 1.5).abs() < 0.1, "Sharpe should be ~1.5, got {}", sharpe); + assert!((sharpe - 1.5).abs() < 0.1, "Sharpe should be ~1.5, got {sharpe}"); - // Read back - buckets let buckets = provider.get_gate_buckets(); - // Should have at least one bucket with data assert!(!buckets.is_empty(), "Should have confidence buckets"); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] #[ignore = "Requires QuestDB on localhost:8812"] async fn test_questdb_provider_try_new_succeeds() { let provider = QuestDBMetricsProvider::try_new("postgresql://admin:quest@localhost:8812/qdb").await; assert!(provider.is_some(), "Should connect to QuestDB"); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_questdb_provider_try_new_fails_gracefully() { // Connect to a non-existent QuestDB — should return None, not panic let provider = QuestDBMetricsProvider::try_new("postgresql://admin:quest@localhost:19999/qdb").await; assert!(provider.is_none(), "Should fail gracefully"); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] #[ignore = "Requires QuestDB on localhost:8812"] async fn test_feedback_loop_with_questdb() { use crate::feedback_loop::{FeedbackLoop, FeedbackLoopConfig};