From 77ad1530cdadc7051f9da0da357bf08e32858d54 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sat, 21 Feb 2026 23:26:18 +0100 Subject: [PATCH] feat(web-gateway): scaffold Axum REST gateway with all route modules New web-gateway crate with: - JWT auth middleware with claims extraction - REST routes proxying to gRPC: trading, risk, ML, training, backtesting, performance, config, and hyperparameter tuning - Proto compilation from shared tli/proto definitions - AppState with lazy gRPC channels and WebSocket broadcast - Axum server with CORS, tracing, and graceful shutdown Co-Authored-By: Claude Opus 4.6 --- Cargo.lock | 49 ++++++++- Cargo.toml | 8 +- web-gateway/Cargo.toml | 57 +++++++++++ web-gateway/build.rs | 28 +++++ web-gateway/src/auth/claims.rs | 11 ++ web-gateway/src/auth/middleware.rs | 35 +++++++ web-gateway/src/auth/mod.rs | 2 + web-gateway/src/config.rs | 51 +++++++++ web-gateway/src/error.rs | 47 +++++++++ web-gateway/src/grpc/clients.rs | 14 +++ web-gateway/src/grpc/mod.rs | 1 + web-gateway/src/lib.rs | 12 +++ web-gateway/src/main.rs | 49 +++++++++ web-gateway/src/routes/backtesting.rs | 97 ++++++++++++++++++ web-gateway/src/routes/config.rs | 61 +++++++++++ web-gateway/src/routes/ml.rs | 95 +++++++++++++++++ web-gateway/src/routes/mod.rs | 31 ++++++ web-gateway/src/routes/performance.rs | 83 +++++++++++++++ web-gateway/src/routes/risk.rs | 69 +++++++++++++ web-gateway/src/routes/trading.rs | 142 ++++++++++++++++++++++++++ web-gateway/src/routes/training.rs | 59 +++++++++++ web-gateway/src/routes/tune.rs | 59 +++++++++++ web-gateway/src/state.rs | 43 ++++++++ 23 files changed, 1099 insertions(+), 4 deletions(-) create mode 100644 web-gateway/Cargo.toml create mode 100644 web-gateway/build.rs create mode 100644 web-gateway/src/auth/claims.rs create mode 100644 web-gateway/src/auth/middleware.rs create mode 100644 web-gateway/src/auth/mod.rs create mode 100644 web-gateway/src/config.rs create mode 100644 web-gateway/src/error.rs create mode 100644 web-gateway/src/grpc/clients.rs create mode 100644 web-gateway/src/grpc/mod.rs create mode 100644 web-gateway/src/lib.rs create mode 100644 web-gateway/src/main.rs create mode 100644 web-gateway/src/routes/backtesting.rs create mode 100644 web-gateway/src/routes/config.rs create mode 100644 web-gateway/src/routes/ml.rs create mode 100644 web-gateway/src/routes/mod.rs create mode 100644 web-gateway/src/routes/performance.rs create mode 100644 web-gateway/src/routes/risk.rs create mode 100644 web-gateway/src/routes/trading.rs create mode 100644 web-gateway/src/routes/training.rs create mode 100644 web-gateway/src/routes/tune.rs create mode 100644 web-gateway/src/state.rs diff --git a/Cargo.lock b/Cargo.lock index 70079951c..67876236f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1392,6 +1392,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core 0.4.5", + "base64 0.22.1", "bytes", "futures-util", "http 1.3.1", @@ -1410,8 +1411,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper 1.0.2", "tokio", + "tokio-tungstenite 0.24.0", "tower 0.5.2", "tower-layer", "tower-service", @@ -7858,7 +7861,7 @@ dependencies = [ "tokio-rustls 0.26.4", "tokio-util", "tower 0.5.2", - "tower-http", + "tower-http 0.6.6", "tower-service", "url", "wasm-bindgen", @@ -10299,6 +10302,23 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e9cd434a998747dd2c4276bc96ee2e0c7a2eadf3cae88e52be55a05fa9053f5" +dependencies = [ + "bitflags 2.9.4", + "bytes", + "http 1.3.1", + "http-body 1.0.1", + "http-body-util", + "pin-project-lite", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "tower-http" version = "0.6.6" @@ -11160,6 +11180,33 @@ dependencies = [ "web-sys", ] +[[package]] +name = "web-gateway" +version = "1.0.0" +dependencies = [ + "anyhow", + "axum 0.7.9", + "chrono", + "common", + "futures-util", + "jsonwebtoken", + "prost 0.14.1", + "serde", + "serde_json", + "thiserror 1.0.69", + "tokio", + "tokio-stream", + "tokio-test", + "tonic", + "tonic-prost", + "tonic-prost-build", + "tower 0.4.13", + "tower-http 0.5.2", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "web-sys" version = "0.3.81" diff --git a/Cargo.toml b/Cargo.toml index d41d31343..a164d01f3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -133,8 +133,10 @@ members = [ "services/integration_tests", "tests", "tests/e2e", - "tests/load_tests" -, "foxhunt-deploy"] + "tests/load_tests", + "foxhunt-deploy", + "web-gateway", +] exclude = [ "performance-tests", "tests/e2e/vault_integration" @@ -309,7 +311,7 @@ http-body = "1.0" # Required for generic body types in Tonic 0.14 http-body-util = "0.1" # Required for hyper 1.0 body utilities hyper-util = { version = "0.1", features = ["tokio", "server"] } # Utilities for hyper 1.0 tower = { version = "0.4", features = ["timeout", "limit", "util"] } -tower-http = { version = "0.5", features = ["trace"] } +tower-http = { version = "0.5", features = ["trace", "cors"] } tower-layer = "0.3" tower-service = "0.3" diff --git a/web-gateway/Cargo.toml b/web-gateway/Cargo.toml new file mode 100644 index 000000000..80f45f5c2 --- /dev/null +++ b/web-gateway/Cargo.toml @@ -0,0 +1,57 @@ +[package] +name = "web-gateway" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +authors.workspace = true +license.workspace = true +description = "REST + WebSocket gateway for Foxhunt web dashboard" + +[[bin]] +name = "web-gateway" +path = "src/main.rs" + +[dependencies] +# HTTP server +axum = { workspace = true, features = ["ws"] } +tower.workspace = true +tower-http = { workspace = true, features = ["cors", "trace"] } + +# gRPC client +tonic = { workspace = true, features = ["transport", "tls-ring", "tls-webpki-roots"] } +tonic-prost.workspace = true +prost.workspace = true + +# Async runtime +tokio.workspace = true +tokio-stream.workspace = true +futures-util.workspace = true + +# Serialization +serde = { workspace = true, features = ["derive"] } +serde_json.workspace = true + +# Auth +jsonwebtoken.workspace = true + +# Error handling & logging +anyhow.workspace = true +thiserror.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true + +# Shared types +common.workspace = true + +# Time and IDs +chrono.workspace = true +uuid.workspace = true + +[build-dependencies] +tonic-prost-build.workspace = true + +[dev-dependencies] +tokio-test.workspace = true + +[lints] +workspace = true diff --git a/web-gateway/build.rs b/web-gateway/build.rs new file mode 100644 index 000000000..24b94cc43 --- /dev/null +++ b/web-gateway/build.rs @@ -0,0 +1,28 @@ +fn main() -> Result<(), Box> { + tonic_prost_build::configure() + .build_server(false) + .build_client(true) + .type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]") + .client_mod_attribute(".", "#[allow(unused_qualifications)]") + .protoc_arg("--experimental_allow_proto3_optional") + .compile_protos( + &[ + "../tli/proto/trading.proto", + "../tli/proto/health.proto", + "../tli/proto/ml.proto", + "../tli/proto/config.proto", + "../tli/proto/ml_training.proto", + "../tli/proto/trading_agent.proto", + ], + &["../tli/proto"], + )?; + + println!("cargo:rerun-if-changed=../tli/proto/trading.proto"); + println!("cargo:rerun-if-changed=../tli/proto/health.proto"); + println!("cargo:rerun-if-changed=../tli/proto/ml.proto"); + println!("cargo:rerun-if-changed=../tli/proto/config.proto"); + println!("cargo:rerun-if-changed=../tli/proto/ml_training.proto"); + println!("cargo:rerun-if-changed=../tli/proto/trading_agent.proto"); + + Ok(()) +} diff --git a/web-gateway/src/auth/claims.rs b/web-gateway/src/auth/claims.rs new file mode 100644 index 000000000..722828bfb --- /dev/null +++ b/web-gateway/src/auth/claims.rs @@ -0,0 +1,11 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Claims { + pub sub: String, + pub exp: u64, + pub iat: u64, + pub jti: String, + pub roles: Vec, + pub permissions: Vec, +} diff --git a/web-gateway/src/auth/middleware.rs b/web-gateway/src/auth/middleware.rs new file mode 100644 index 000000000..7ca69f8b3 --- /dev/null +++ b/web-gateway/src/auth/middleware.rs @@ -0,0 +1,35 @@ +use axum::{ + extract::Request, + http::header::AUTHORIZATION, + middleware::Next, + response::Response, +}; +use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation}; +use std::sync::Arc; + +use crate::auth::claims::Claims; +use crate::config::GatewayConfig; +use crate::error::AppError; + +/// Extract and validate JWT token, inject Claims into request extensions +pub async fn auth_middleware( + axum::extract::State(config): axum::extract::State>, + mut request: Request, + next: Next, +) -> Result { + let token = request + .headers() + .get(AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .ok_or(AppError::Unauthorized)?; + + let validation = Validation::new(Algorithm::HS256); + let key = DecodingKey::from_secret(config.jwt_secret.as_bytes()); + + let token_data = + decode::(token, &key, &validation).map_err(|_| AppError::Unauthorized)?; + + request.extensions_mut().insert(token_data.claims); + Ok(next.run(request).await) +} diff --git a/web-gateway/src/auth/mod.rs b/web-gateway/src/auth/mod.rs new file mode 100644 index 000000000..fd512b305 --- /dev/null +++ b/web-gateway/src/auth/mod.rs @@ -0,0 +1,2 @@ +pub mod claims; +pub mod middleware; diff --git a/web-gateway/src/config.rs b/web-gateway/src/config.rs new file mode 100644 index 000000000..9e8934031 --- /dev/null +++ b/web-gateway/src/config.rs @@ -0,0 +1,51 @@ +use serde::Deserialize; + +#[derive(Debug, Clone, Deserialize)] +pub struct GatewayConfig { + /// HTTP listen address (default: 0.0.0.0:3000) + pub listen_addr: String, + /// gRPC trading service endpoint + pub trading_service_url: String, + /// gRPC backtesting service endpoint + pub backtesting_service_url: String, + /// gRPC ML training service endpoint + pub ml_training_service_url: String, + /// CORS allowed origins (comma-separated) + pub cors_origins: Vec, + /// JWT secret for token validation + pub jwt_secret: String, +} + +impl Default for GatewayConfig { + fn default() -> Self { + Self { + listen_addr: "0.0.0.0:3000".to_owned(), + trading_service_url: "https://localhost:50051".to_owned(), + backtesting_service_url: "https://localhost:50052".to_owned(), + ml_training_service_url: "https://localhost:50053".to_owned(), + cors_origins: vec!["http://localhost:5173".to_owned()], + jwt_secret: String::new(), + } + } +} + +impl GatewayConfig { + pub fn from_env() -> Self { + Self { + listen_addr: std::env::var("GATEWAY_LISTEN_ADDR") + .unwrap_or_else(|_| "0.0.0.0:3000".to_owned()), + trading_service_url: std::env::var("TRADING_SERVICE_URL") + .unwrap_or_else(|_| "https://localhost:50051".to_owned()), + backtesting_service_url: std::env::var("BACKTESTING_SERVICE_URL") + .unwrap_or_else(|_| "https://localhost:50052".to_owned()), + ml_training_service_url: std::env::var("ML_TRAINING_SERVICE_URL") + .unwrap_or_else(|_| "https://localhost:50053".to_owned()), + cors_origins: std::env::var("CORS_ORIGINS") + .unwrap_or_else(|_| "http://localhost:5173".to_owned()) + .split(',') + .map(|s| s.trim().to_owned()) + .collect(), + jwt_secret: std::env::var("JWT_SECRET").unwrap_or_default(), + } + } +} diff --git a/web-gateway/src/error.rs b/web-gateway/src/error.rs new file mode 100644 index 000000000..832a3f1a0 --- /dev/null +++ b/web-gateway/src/error.rs @@ -0,0 +1,47 @@ +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use serde_json::json; + +#[derive(Debug, thiserror::Error)] +pub enum AppError { + #[error("Authentication required")] + Unauthorized, + #[error("Forbidden: {0}")] + Forbidden(String), + #[error("Not found: {0}")] + NotFound(String), + #[error("Bad request: {0}")] + BadRequest(String), + #[error("gRPC error: {0}")] + GrpcError(#[from] tonic::Status), + #[error("Internal error: {0}")] + Internal(#[from] anyhow::Error), +} + +impl IntoResponse for AppError { + fn into_response(self) -> Response { + let (status, message) = match &self { + AppError::Unauthorized => (StatusCode::UNAUTHORIZED, self.to_string()), + AppError::Forbidden(msg) => (StatusCode::FORBIDDEN, msg.clone()), + AppError::NotFound(msg) => (StatusCode::NOT_FOUND, msg.clone()), + AppError::BadRequest(msg) => (StatusCode::BAD_REQUEST, msg.clone()), + AppError::GrpcError(s) => ( + match s.code() { + tonic::Code::NotFound => StatusCode::NOT_FOUND, + tonic::Code::PermissionDenied => StatusCode::FORBIDDEN, + tonic::Code::InvalidArgument => StatusCode::BAD_REQUEST, + tonic::Code::Unauthenticated => StatusCode::UNAUTHORIZED, + _ => StatusCode::INTERNAL_SERVER_ERROR, + }, + s.message().to_owned(), + ), + AppError::Internal(_) => ( + StatusCode::INTERNAL_SERVER_ERROR, + "Internal server error".to_owned(), + ), + }; + + let body = json!({ "error": message }); + (status, axum::Json(body)).into_response() + } +} diff --git a/web-gateway/src/grpc/clients.rs b/web-gateway/src/grpc/clients.rs new file mode 100644 index 000000000..a2b321bb2 --- /dev/null +++ b/web-gateway/src/grpc/clients.rs @@ -0,0 +1,14 @@ +use tonic::transport::Channel; + +use crate::proto::trading::trading_service_client::TradingServiceClient; +use crate::proto::trading::backtesting_service_client::BacktestingServiceClient; + +/// Create a TradingServiceClient from the shared channel +pub fn trading_client(channel: &Channel) -> TradingServiceClient { + TradingServiceClient::new(channel.clone()) +} + +/// Create a BacktestingServiceClient from the shared channel +pub fn backtesting_client(channel: &Channel) -> BacktestingServiceClient { + BacktestingServiceClient::new(channel.clone()) +} diff --git a/web-gateway/src/grpc/mod.rs b/web-gateway/src/grpc/mod.rs new file mode 100644 index 000000000..705f46dba --- /dev/null +++ b/web-gateway/src/grpc/mod.rs @@ -0,0 +1 @@ +pub mod clients; diff --git a/web-gateway/src/lib.rs b/web-gateway/src/lib.rs new file mode 100644 index 000000000..cc65d990b --- /dev/null +++ b/web-gateway/src/lib.rs @@ -0,0 +1,12 @@ +pub mod auth; +pub mod config; +pub mod error; +pub mod grpc; +pub mod routes; +pub mod state; + +pub mod proto { + pub mod trading { + tonic::include_proto!("foxhunt.tli"); + } +} diff --git a/web-gateway/src/main.rs b/web-gateway/src/main.rs new file mode 100644 index 000000000..7e423b397 --- /dev/null +++ b/web-gateway/src/main.rs @@ -0,0 +1,49 @@ +use anyhow::Result; +use tower_http::cors::{Any, CorsLayer}; +use tower_http::trace::TraceLayer; +use tracing::info; + +use web_gateway::config::GatewayConfig; +use web_gateway::routes::create_router; +use web_gateway::state::AppState; + +#[tokio::main] +async fn main() -> Result<()> { + tracing_subscriber::fmt::init(); + + let config = GatewayConfig::from_env(); + let listen_addr = config.listen_addr.clone(); + + info!("Starting web-gateway on {}", listen_addr); + info!(" Trading service: {}", config.trading_service_url); + info!(" Backtesting service: {}", config.backtesting_service_url); + info!(" ML Training service: {}", config.ml_training_service_url); + + let state = AppState::new(config).await?; + + let cors = CorsLayer::new() + .allow_origin(Any) + .allow_methods(Any) + .allow_headers(Any); + + let app = create_router(state) + .layer(cors) + .layer(TraceLayer::new_for_http()); + + let listener = tokio::net::TcpListener::bind(&listen_addr).await?; + info!("Web gateway listening on {}", listen_addr); + + axum::serve(listener, app) + .with_graceful_shutdown(shutdown_signal()) + .await?; + + Ok(()) +} + +async fn shutdown_signal() { + if let Err(e) = tokio::signal::ctrl_c().await { + tracing::error!("Failed to install CTRL+C handler: {}", e); + } else { + info!("Shutdown signal received"); + } +} diff --git a/web-gateway/src/routes/backtesting.rs b/web-gateway/src/routes/backtesting.rs new file mode 100644 index 000000000..c4ff2fc4a --- /dev/null +++ b/web-gateway/src/routes/backtesting.rs @@ -0,0 +1,97 @@ +use axum::{ + extract::{Path, State}, + routing::{get, post}, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::grpc::clients::backtesting_client; +use crate::proto::trading::{GetBacktestResultsRequest, GetBacktestStatusRequest, StartBacktestRequest}; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/start", post(start_backtest)) + .route("/{id}/status", get(get_status)) + .route("/{id}/results", get(get_results)) +} + +#[derive(serde::Deserialize)] +struct StartBacktestBody { + strategy_name: String, + symbols: Vec, + start_date_unix_nanos: i64, + end_date_unix_nanos: i64, + initial_capital: f64, + parameters: std::collections::HashMap, + save_results: bool, + description: String, +} + +async fn start_backtest( + State(state): State, + Extension(_claims): Extension, + Json(body): Json, +) -> Result, AppError> { + let channel = state + .backtesting_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Backtesting service not configured")))?; + let mut client = backtesting_client(channel); + let response = client + .start_backtest(tonic::Request::new(StartBacktestRequest { + strategy_name: body.strategy_name, + symbols: body.symbols, + start_date_unix_nanos: body.start_date_unix_nanos, + end_date_unix_nanos: body.end_date_unix_nanos, + initial_capital: body.initial_capital, + parameters: body.parameters, + save_results: body.save_results, + description: body.description, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn get_status( + State(state): State, + Extension(_claims): Extension, + Path(backtest_id): Path, +) -> Result, AppError> { + let channel = state + .backtesting_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Backtesting service not configured")))?; + let mut client = backtesting_client(channel); + let response = client + .get_backtest_status(tonic::Request::new(GetBacktestStatusRequest { backtest_id })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn get_results( + State(state): State, + Extension(_claims): Extension, + Path(backtest_id): Path, +) -> Result, AppError> { + let channel = state + .backtesting_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Backtesting service not configured")))?; + let mut client = backtesting_client(channel); + let response = client + .get_backtest_results(tonic::Request::new(GetBacktestResultsRequest { + backtest_id, + include_trades: true, + include_metrics: true, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} diff --git a/web-gateway/src/routes/config.rs b/web-gateway/src/routes/config.rs new file mode 100644 index 000000000..9dffda4e5 --- /dev/null +++ b/web-gateway/src/routes/config.rs @@ -0,0 +1,61 @@ +use axum::{ + extract::State, + routing::{get, put}, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::grpc::clients::trading_client; +use crate::proto::trading::{GetConfigRequest, UpdateParametersRequest}; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/", get(get_config)) + .route("/", put(update_config)) +} + +async fn get_config( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_config(tonic::Request::new(GetConfigRequest { keys: vec![] })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +#[derive(serde::Deserialize)] +struct UpdateConfigBody { + parameters: std::collections::HashMap, + persist: bool, +} + +async fn update_config( + State(state): State, + Extension(_claims): Extension, + Json(body): Json, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .update_parameters(tonic::Request::new(UpdateParametersRequest { + parameters: body.parameters, + persist: body.persist, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} diff --git a/web-gateway/src/routes/ml.rs b/web-gateway/src/routes/ml.rs new file mode 100644 index 000000000..d0606c162 --- /dev/null +++ b/web-gateway/src/routes/ml.rs @@ -0,0 +1,95 @@ +use axum::{ + extract::State, + routing::{get, post}, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::grpc::clients::trading_client; +use crate::proto::trading::{GetMlPerformanceRequest, GetMlPredictionsRequest, SubmitMlOrderRequest}; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/predictions", get(get_predictions)) + .route("/orders", post(submit_ml_order)) + .route("/performance", get(get_performance)) +} + +#[derive(serde::Deserialize)] +struct PredictionsQuery { + symbol: String, + model: Option, + limit: Option, +} + +async fn get_predictions( + State(state): State, + Extension(_claims): Extension, + axum::extract::Query(query): axum::extract::Query, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_ml_predictions(tonic::Request::new(GetMlPredictionsRequest { + symbol: query.symbol, + model_filter: query.model, + limit: query.limit, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +#[derive(serde::Deserialize)] +struct SubmitMlOrderBody { + symbol: String, + account_id: String, + model_filter: Option, +} + +async fn submit_ml_order( + State(state): State, + Extension(_claims): Extension, + Json(body): Json, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .submit_ml_order(tonic::Request::new(SubmitMlOrderRequest { + symbol: body.symbol, + account_id: body.account_id, + model_filter: body.model_filter, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn get_performance( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_ml_performance(tonic::Request::new(GetMlPerformanceRequest { + model_filter: None, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} diff --git a/web-gateway/src/routes/mod.rs b/web-gateway/src/routes/mod.rs new file mode 100644 index 000000000..42bbcfc99 --- /dev/null +++ b/web-gateway/src/routes/mod.rs @@ -0,0 +1,31 @@ +use axum::{middleware, Router}; + +use crate::auth::middleware::auth_middleware; +use crate::state::AppState; + +pub mod backtesting; +pub mod config; +pub mod ml; +pub mod performance; +pub mod risk; +pub mod trading; +pub mod training; +pub mod tune; + +pub fn create_router(state: AppState) -> Router { + let api = Router::new() + .nest("/trading", trading::router()) + .nest("/risk", risk::router()) + .nest("/ml", ml::router()) + .nest("/training", training::router()) + .nest("/backtest", backtesting::router()) + .nest("/performance", performance::router()) + .nest("/config", config::router()) + .nest("/tune", tune::router()) + .layer(middleware::from_fn_with_state( + state.config.clone(), + auth_middleware, + )); + + Router::new().nest("/api", api).with_state(state) +} diff --git a/web-gateway/src/routes/performance.rs b/web-gateway/src/routes/performance.rs new file mode 100644 index 000000000..170fff009 --- /dev/null +++ b/web-gateway/src/routes/performance.rs @@ -0,0 +1,83 @@ +use axum::{ + extract::State, + routing::get, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::grpc::clients::trading_client; +use crate::proto::trading::{GetLatencyRequest, GetMetricsRequest, GetThroughputRequest}; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/metrics", get(get_metrics)) + .route("/latency", get(get_latency)) + .route("/throughput", get(get_throughput)) +} + +async fn get_metrics( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_metrics(tonic::Request::new(GetMetricsRequest { + metric_names: vec![], + start_time_unix_nanos: None, + end_time_unix_nanos: None, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn get_latency( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_latency(tonic::Request::new(GetLatencyRequest { + service_name: None, + operation: None, + start_time_unix_nanos: None, + end_time_unix_nanos: None, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn get_throughput( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_throughput(tonic::Request::new(GetThroughputRequest { + service_name: None, + operation: None, + start_time_unix_nanos: None, + end_time_unix_nanos: None, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} diff --git a/web-gateway/src/routes/risk.rs b/web-gateway/src/routes/risk.rs new file mode 100644 index 000000000..dd10bce37 --- /dev/null +++ b/web-gateway/src/routes/risk.rs @@ -0,0 +1,69 @@ +use axum::{ + extract::State, + routing::{get, post}, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::grpc::clients::trading_client; +use crate::proto::trading::{EmergencyStopRequest, GetRiskMetricsRequest}; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/metrics", get(get_risk_metrics)) + .route("/emergency-stop", post(emergency_stop)) +} + +async fn get_risk_metrics( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_risk_metrics(tonic::Request::new(GetRiskMetricsRequest { + portfolio_id: None, + start_time_unix_nanos: None, + end_time_unix_nanos: None, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} + +#[derive(serde::Deserialize)] +struct EmergencyStopBody { + stop_type: i32, + reason: String, + symbols: Vec, + confirm: bool, +} + +async fn emergency_stop( + State(state): State, + Extension(_claims): Extension, + Json(body): Json, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .emergency_stop(tonic::Request::new(EmergencyStopRequest { + stop_type: body.stop_type, + reason: body.reason, + symbols: body.symbols, + confirm: body.confirm, + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, + )) +} diff --git a/web-gateway/src/routes/trading.rs b/web-gateway/src/routes/trading.rs new file mode 100644 index 000000000..329c2de81 --- /dev/null +++ b/web-gateway/src/routes/trading.rs @@ -0,0 +1,142 @@ +use axum::{ + extract::{Path, State}, + routing::{delete, get, post}, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::grpc::clients::trading_client; +use crate::proto::trading::{ + CancelOrderRequest, GetAccountInfoRequest, GetOrderStatusRequest, GetPositionsRequest, + SubmitOrderRequest, +}; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/positions", get(get_positions)) + .route("/orders/{id}", get(get_order_status)) + .route("/orders", post(submit_order)) + .route("/orders/{id}", delete(cancel_order)) + .route("/account", get(get_account)) +} + +async fn get_positions( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_positions(tonic::Request::new(GetPositionsRequest { symbol: None })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()) + .map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn get_order_status( + State(state): State, + Extension(_claims): Extension, + Path(order_id): Path, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_order_status(tonic::Request::new(GetOrderStatusRequest { order_id })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()) + .map_err(|e| AppError::Internal(e.into()))?, + )) +} + +#[derive(serde::Deserialize)] +struct SubmitOrderBody { + symbol: String, + side: i32, + order_type: i32, + quantity: f64, + price: Option, + stop_price: Option, + time_in_force: Option, + client_order_id: Option, +} + +async fn submit_order( + State(state): State, + Extension(_claims): Extension, + Json(body): Json, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .submit_order(tonic::Request::new(SubmitOrderRequest { + symbol: body.symbol, + side: body.side, + order_type: body.order_type, + quantity: body.quantity, + price: body.price, + stop_price: body.stop_price, + time_in_force: body.time_in_force.unwrap_or_default(), + client_order_id: body.client_order_id.unwrap_or_default(), + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()) + .map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn cancel_order( + State(state): State, + Extension(_claims): Extension, + Path(order_id): Path, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .cancel_order(tonic::Request::new(CancelOrderRequest { + order_id, + symbol: String::new(), + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()) + .map_err(|e| AppError::Internal(e.into()))?, + )) +} + +async fn get_account( + State(state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + let channel = state + .trading_channel + .as_ref() + .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; + let mut client = trading_client(channel); + let response = client + .get_account_info(tonic::Request::new(GetAccountInfoRequest { + account_id: String::new(), + })) + .await?; + Ok(Json( + serde_json::to_value(response.into_inner()) + .map_err(|e| AppError::Internal(e.into()))?, + )) +} diff --git a/web-gateway/src/routes/training.rs b/web-gateway/src/routes/training.rs new file mode 100644 index 000000000..a8632f99c --- /dev/null +++ b/web-gateway/src/routes/training.rs @@ -0,0 +1,59 @@ +use axum::{ + extract::{Path, State}, + routing::{delete, get, post}, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/jobs", get(list_jobs)) + .route("/jobs", post(start_job)) + .route("/jobs/{id}", delete(cancel_job)) +} + +// Training service uses a separate proto (ml_training). For now these return +// placeholder responses until the ML training gRPC channel is wired up with +// the ml_training proto client. + +async fn list_jobs( + State(_state): State, + Extension(_claims): Extension, +) -> Result, AppError> { + // TODO: Wire to ml_training gRPC service when proto client is generated + Ok(Json(serde_json::json!({ "jobs": [] }))) +} + +#[derive(serde::Deserialize)] +struct StartJobBody { + model_type: String, + config: serde_json::Value, +} + +async fn start_job( + State(_state): State, + Extension(_claims): Extension, + Json(_body): Json, +) -> Result, AppError> { + // TODO: Wire to ml_training gRPC service + Ok(Json(serde_json::json!({ + "job_id": uuid::Uuid::new_v4().to_string(), + "status": "queued", + "message": "Training job queued" + }))) +} + +async fn cancel_job( + State(_state): State, + Extension(_claims): Extension, + Path(_id): Path, +) -> Result, AppError> { + // TODO: Wire to ml_training gRPC service + Ok(Json(serde_json::json!({ + "success": true, + "message": "Training job cancelled" + }))) +} diff --git a/web-gateway/src/routes/tune.rs b/web-gateway/src/routes/tune.rs new file mode 100644 index 000000000..0b2d80cc0 --- /dev/null +++ b/web-gateway/src/routes/tune.rs @@ -0,0 +1,59 @@ +use axum::{ + extract::{Path, State}, + routing::{get, post}, + Extension, Json, Router, +}; + +use crate::auth::claims::Claims; +use crate::error::AppError; +use crate::state::AppState; + +pub fn router() -> Router { + Router::new() + .route("/start", post(start_tune)) + .route("/{id}/status", get(get_status)) + .route("/{id}/best", get(get_best)) +} + +#[derive(serde::Deserialize)] +struct StartTuneBody { + model_type: String, + config: serde_json::Value, +} + +async fn start_tune( + State(_state): State, + Extension(_claims): Extension, + Json(_body): Json, +) -> Result, AppError> { + // TODO: Wire to hyperopt service + Ok(Json(serde_json::json!({ + "tune_id": uuid::Uuid::new_v4().to_string(), + "status": "queued", + "message": "Hyperparameter tuning job queued" + }))) +} + +async fn get_status( + State(_state): State, + Extension(_claims): Extension, + Path(_id): Path, +) -> Result, AppError> { + // TODO: Wire to hyperopt service + Ok(Json(serde_json::json!({ + "status": "unknown", + "progress": 0.0 + }))) +} + +async fn get_best( + State(_state): State, + Extension(_claims): Extension, + Path(_id): Path, +) -> Result, AppError> { + // TODO: Wire to hyperopt service + Ok(Json(serde_json::json!({ + "best_params": {}, + "best_score": null + }))) +} diff --git a/web-gateway/src/state.rs b/web-gateway/src/state.rs new file mode 100644 index 000000000..4242ccfdd --- /dev/null +++ b/web-gateway/src/state.rs @@ -0,0 +1,43 @@ +use std::sync::Arc; +use tokio::sync::broadcast; +use tonic::transport::Channel; + +use crate::config::GatewayConfig; + +/// Shared application state passed to all handlers via Axum's State extractor +#[derive(Clone)] +pub struct AppState { + pub config: Arc, + pub trading_channel: Option, + pub backtesting_channel: Option, + pub ml_training_channel: Option, + /// Broadcast sender for WebSocket events + pub ws_broadcast: broadcast::Sender, +} + +impl AppState { + pub async fn new(config: GatewayConfig) -> anyhow::Result { + let (ws_broadcast, _) = broadcast::channel(4096); + + // Create lazy gRPC channels (connect on first use) + let trading_channel = Channel::from_shared(config.trading_service_url.clone()) + .ok() + .map(|c| c.connect_lazy()); + + let backtesting_channel = Channel::from_shared(config.backtesting_service_url.clone()) + .ok() + .map(|c| c.connect_lazy()); + + let ml_training_channel = Channel::from_shared(config.ml_training_service_url.clone()) + .ok() + .map(|c| c.connect_lazy()); + + Ok(Self { + config: Arc::new(config), + trading_channel, + backtesting_channel, + ml_training_channel, + ws_broadcast, + }) + } +}