Datamanager predictions handlers

applications/datamanager/.claude/settings.local.json

@@ -19,4 +19,5 @@
     "deny": [],
     "defaultMode": "acceptEdits"
   }
-}
+}
+

applications/datamanager/Cargo.toml

@@ -14,7 +14,14 @@ path = "src/main.rs"
 [dependencies]
 axum = "0.8.4"
 chrono = { version = "0.4.41", features = ["serde"] }
-polars = { version = "0.50.0", features = ["json", "lazy", "parquet", "temporal"] }
+polars = { version = "0.50.0", features = [
+    "json",
+    "lazy",
+    "parquet",
+    "temporal",
+    "serde",
+    "polars-io",
+] }
 reqwest = "0.12.23"
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.143"
@@ -27,6 +34,8 @@ aws-config = "1.5.8"
 aws-sdk-s3 = "1.48.0"
 aws-credential-types = "1.2.6"
 duckdb = { version = "1.0", features = ["r2d2", "chrono"] }
+validator = { version = "0.18", features = ["derive"] }
+thiserror = "2.0.3"
 
 [dev-dependencies]
 tokio-test = "0.4"

applications/datamanager/src/lib.rs

@@ -2,11 +2,11 @@ use aws_sdk_s3::Client as S3Client;
 use axum::{routing::get, Router};
 use reqwest::Client;
 use tower_http::trace::TraceLayer;
-use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 pub mod routes;
 use routes::equity;
 use routes::health;
+use routes::prediction;
 
 #[derive(Clone)]
 pub struct AlpacaSecrets {
@@ -69,8 +69,8 @@ pub async fn create_app() -> Router<AppState> {
 
     Router::<AppState>::new()
         .route("/health", get(health::check))
+        .merge(prediction::router())
         .merge(equity::router())
         .with_state(state)
         .layer(TraceLayer::new_for_http())
 }
-

applications/datamanager/src/routes/equity.rs

@@ -32,16 +32,14 @@ struct DateRangeQuery {
 struct BarResult {
     #[serde(rename = "T")]
     ticker: String,
-    // TODO: money types
-    c: Option<f64>,
-    h: Option<f64>,
-    l: Option<f64>,
-    n: Option<f64>,
-    o: Option<f64>,
-    // otc: bool,
-    t: i64,
-    v: Option<f64>,
-    vw: Option<f64>,
+    c: Option<u64>,
+    h: Option<u64>,
+    l: Option<u64>,
+    n: Option<u64>,
+    o: Option<u64>,
+    t: u64,
+    v: Option<u64>,
+    vw: Option<u64>,
 }
 
 #[derive(serde::Deserialize, Debug)]
@@ -337,12 +335,12 @@ async fn sync(State(state): State<AppState>, payload: Json<DailySync>) -> impl I
 
     let tickers: Vec<String> = bars.iter().map(|b| b.ticker.clone()).collect();
     let volumes: Vec<Option<u64>> = bars.iter().map(|b| b.v.map(|v| v as u64)).collect();
-    let vw_prices: Vec<Option<f64>> = bars.iter().map(|b| b.vw).collect();
-    let open_prices: Vec<Option<f64>> = bars.iter().map(|b| b.o).collect();
-    let close_prices: Vec<Option<f64>> = bars.iter().map(|b| b.c).collect();
-    let high_prices: Vec<Option<f64>> = bars.iter().map(|b| b.h).collect();
-    let low_prices: Vec<Option<f64>> = bars.iter().map(|b| b.l).collect();
-    let timestamps: Vec<i64> = bars.iter().map(|b| b.t).collect();
+    let vw_prices: Vec<Option<f64>> = bars.iter().map(|b| b.vw.map(|vw| vw as f64)).collect();
+    let open_prices: Vec<Option<f64>> = bars.iter().map(|b| b.o.map(|o| o as f64)).collect();
+    let close_prices: Vec<Option<f64>> = bars.iter().map(|b| b.c.map(|c| c as f64)).collect();
+    let high_prices: Vec<Option<f64>> = bars.iter().map(|b| b.h.map(|h| h as f64)).collect();
+    let low_prices: Vec<Option<f64>> = bars.iter().map(|b| b.l.map(|l| l as f64)).collect();
+    let timestamps: Vec<i64> = bars.iter().map(|b| b.t as i64).collect();
     let num_transactions: Vec<Option<u64>> = bars.iter().map(|b| b.n.map(|n| n as u64)).collect();
 
     let df_result = df! {

applications/datamanager/src/routes/mod.rs

@@ -1,2 +1,3 @@
 pub mod equity;
 pub mod health;
+pub mod prediction;