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

fix(trading): fix QuestDB metrics provider for real integration testing

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-23 14:13:19 +01:00
parent 69d435cb52
commit 3f9f6ebe17
1 changed files with 92 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
services/trading_service/src/questdb_metrics.rs
View File

@@ -109,7 +109,7 @@ impl QuestDBMetricsProvider {
/// Query 30-day rolling Sharpe per model
async fn query_model_sharpe(&self) -> HashMap<String, f64> {
let result: Result<Vec<(String, f64)>, _> = 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<String, (f64, u64)> {
let result: Result<Vec<(String, f64, i64)>, _> = 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<Vec<(f64, f64, f64, i64, f64)>, _> = 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<ModelRollingMetrics> {
// 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<GateBucketMetrics> {
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<String, f64> {
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};