feat(api_gateway): add all stream adapters + monitoring streams

Gateway poll-to-stream adapters:
- broker_gateway: StreamAccountState (3s), StreamSessionStatus (5s)
- data_acquisition: StreamDownloadStatus (5s)
- risk: StreamCircuitBreakerStatus (2s), StreamRiskMetrics (3s)
- ml: StreamModelStatus (5s)
- trading: StreamPortfolioSummary (3s), StreamOrderBook (1s)
- trading_agent: StreamAgentStatus (3s)

Monitoring handler real implementations:
- stream_system_status: fan-out health checks
- stream_metrics: Prometheus query loop
- stream_alerts: empty placeholder (no alertmanager yet)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

services/api_gateway/src/grpc/broker_gateway_proxy.rs

@@ -4,6 +4,7 @@
 
 use futures::Stream;
 use std::pin::Pin;
+use std::time::Duration;
 use tonic::{Request, Response, Status};
 use tracing::{error, instrument};
 
@@ -13,7 +14,8 @@ use crate::broker_gateway::{
     CancelOrderRequest, CancelOrderResponse, ExecutionEvent, GetAccountStateRequest,
     GetAccountStateResponse, GetPositionsRequest, GetPositionsResponse,
     GetSessionStatusRequest, GetSessionStatusResponse, HealthCheckRequest, HealthCheckResponse,
-    RouteOrderRequest, RouteOrderResponse, StreamExecutionsRequest,
+    RouteOrderRequest, RouteOrderResponse, StreamAccountStateRequest, StreamExecutionsRequest,
+    StreamSessionStatusRequest,
 };
 
 #[derive(Debug, Clone)]
@@ -122,4 +124,74 @@ impl BrokerGatewayService for BrokerGatewayProxy {
             e
         })
     }
+
+    type StreamAccountStateStream =
+        Pin<Box<dyn Stream<Item = Result<GetAccountStateResponse, Status>> + Send>>;
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_account_state(
+        &self,
+        request: Request<StreamAccountStateRequest>,
+    ) -> Result<Response<Self::StreamAccountStateStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            3
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_account_state(Request::new(GetAccountStateRequest {
+                    account_id: String::new(),
+                })).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetAccountState failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    type StreamSessionStatusStream =
+        Pin<Box<dyn Stream<Item = Result<GetSessionStatusResponse, Status>> + Send>>;
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_session_status(
+        &self,
+        request: Request<StreamSessionStatusRequest>,
+    ) -> Result<Response<Self::StreamSessionStatusStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            5
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_session_status(Request::new(GetSessionStatusRequest::default())).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetSessionStatus failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
 }

services/api_gateway/src/grpc/data_acquisition_proxy.rs

@@ -3,6 +3,9 @@
 //! Forwards to data-acquisition-service:50057 (separate service).
 //! All RPCs are unary (no streaming) — simplest proxy.
 
+use futures::Stream;
+use std::pin::Pin;
+use std::time::Duration;
 use tonic::{Request, Response, Status};
 use tracing::{error, instrument};
 
@@ -12,7 +15,7 @@ use crate::data_acquisition::{
     CancelDownloadRequest, CancelDownloadResponse, GetDownloadStatusRequest,
     GetDownloadStatusResponse, HealthCheckRequest, HealthCheckResponse,
     ListDownloadJobsRequest, ListDownloadJobsResponse, ScheduleDownloadRequest,
-    ScheduleDownloadResponse,
+    ScheduleDownloadResponse, StreamDownloadStatusRequest,
 };
 
 #[derive(Debug, Clone)]
@@ -98,4 +101,38 @@ impl DataAcquisitionService for DataAcquisitionProxy {
             e
         })
     }
+
+    type StreamDownloadStatusStream =
+        Pin<Box<dyn Stream<Item = Result<ListDownloadJobsResponse, Status>> + Send>>;
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_download_status(
+        &self,
+        request: Request<StreamDownloadStatusRequest>,
+    ) -> Result<Response<Self::StreamDownloadStatusStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            5
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.list_download_jobs(Request::new(ListDownloadJobsRequest::default())).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend ListDownloadJobs failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
 }

services/api_gateway/src/grpc/ml_proxy.rs

@@ -4,6 +4,7 @@
 
 use futures::Stream;
 use std::pin::Pin;
+use std::time::Duration;
 use tonic::{Request, Response, Status};
 use tracing::{error, instrument};
 
@@ -15,7 +16,8 @@ use crate::ml_inference::{
     GetModelPerformanceRequest, GetModelPerformanceResponse, GetModelStatusRequest,
     GetModelStatusResponse, GetPredictionRequest, GetPredictionResponse, ModelMetricsEvent,
     PredictionEvent, RetrainModelRequest, RetrainModelResponse, SignalStrengthEvent,
-    StreamModelMetricsRequest, StreamPredictionsRequest, StreamSignalStrengthRequest,
+    StreamModelMetricsRequest, StreamModelStatusRequest, StreamPredictionsRequest,
+    StreamSignalStrengthRequest,
 };
 
 #[derive(Debug, Clone)]
@@ -37,6 +39,8 @@ impl MlService for MlServiceProxy {
         Pin<Box<dyn Stream<Item = Result<ModelMetricsEvent, Status>> + Send>>;
     type StreamSignalStrengthStream =
         Pin<Box<dyn Stream<Item = Result<SignalStrengthEvent, Status>> + Send>>;
+    type StreamModelStatusStream =
+        Pin<Box<dyn Stream<Item = Result<GetModelStatusResponse, Status>> + Send>>;
 
     #[instrument(skip(self, request), err)]
     async fn get_prediction(
@@ -189,4 +193,38 @@ impl MlService for MlServiceProxy {
             })?;
         Ok(Response::new(Box::pin(stream.into_inner())))
     }
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_model_status(
+        &self,
+        request: Request<StreamModelStatusRequest>,
+    ) -> Result<Response<Self::StreamModelStatusStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            5
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let model_name = req.model_name;
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_model_status(Request::new(GetModelStatusRequest {
+                    model_name: if model_name.is_empty() { None } else { Some(model_name.clone()) },
+                })).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetModelStatus failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
 }

services/api_gateway/src/grpc/monitoring_handler.rs

@@ -21,15 +21,16 @@ use tracing::{error, info, instrument};
 
 use crate::monitoring::monitoring_service_server::{MonitoringService, MonitoringServiceServer};
 use crate::monitoring::{
-    AcknowledgeAlertRequest, AcknowledgeAlertResponse, AlertEvent, EpochFinancialSnapshot,
-    GetActiveAlertsRequest, GetActiveAlertsResponse, GetEpochHistoryRequest,
-    GetEpochHistoryResponse, GetHealthCheckRequest, GetHealthCheckResponse,
-    GetLatencyMetricsRequest, GetLatencyMetricsResponse, GetLiveTrainingMetricsRequest,
-    GetLiveTrainingMetricsResponse, GetMetricsRequest, GetMetricsResponse,
-    GetSystemStatusRequest, GetSystemStatusResponse, GetThroughputMetricsRequest,
-    GetThroughputMetricsResponse, GpuSnapshot, HealthCheck, HealthStatus, Metric, MetricType,
-    MetricsEvent, ServiceHealth, ServiceState, ServiceStatus, StreamTrainingMetricsRequest,
-    SystemHealth, SystemMetrics, SystemStatus, SystemStatusEvent, TrainingSession,
+    AcknowledgeAlertRequest, AcknowledgeAlertResponse, AlertEvent, AlertEventType,
+    EpochFinancialSnapshot, GetActiveAlertsRequest, GetActiveAlertsResponse,
+    GetEpochHistoryRequest, GetEpochHistoryResponse, GetHealthCheckRequest,
+    GetHealthCheckResponse, GetLatencyMetricsRequest, GetLatencyMetricsResponse,
+    GetLiveTrainingMetricsRequest, GetLiveTrainingMetricsResponse, GetMetricsRequest,
+    GetMetricsResponse, GetSystemStatusRequest, GetSystemStatusResponse,
+    GetThroughputMetricsRequest, GetThroughputMetricsResponse, GpuSnapshot, HealthCheck,
+    HealthStatus, Metric, MetricType, MetricsEvent, ServiceHealth, ServiceState, ServiceStatus,
+    StreamTrainingMetricsRequest, SystemHealth, SystemMetrics, SystemStatus,
+    SystemStatusChangeType, SystemStatusEvent, TrainingSession,
 };
 
 // ============================================================================
@@ -494,13 +495,101 @@ impl MonitoringService for MonitoringServiceHandler {
         Ok(Response::new(response))
     }
 
+    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
     async fn stream_system_status(
         &self,
-        _request: Request<crate::monitoring::StreamSystemStatusRequest>,
+        request: Request<crate::monitoring::StreamSystemStatusRequest>,
     ) -> Result<Response<Self::StreamSystemStatusStream>, Status> {
-        Err(Status::unimplemented(
-            "StreamSystemStatus not yet implemented",
-        ))
+        let req = request.into_inner();
+        let interval_secs = req.update_frequency_seconds.unwrap_or(0);
+        let interval_secs = if interval_secs == 0 {
+            5u32
+        } else {
+            (interval_secs as u32).clamp(1, 60)
+        };
+        let backends = self.backends.clone();
+        let start_time = self.start_time;
+        let service_filter = req.service_names;
+
+        info!(
+            "StreamSystemStatus: interval={}s, filter={:?}",
+            interval_secs,
+            if service_filter.is_empty() {
+                "<all>"
+            } else {
+                "filtered"
+            }
+        );
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+
+                let checks: Vec<_> = backends
+                    .iter()
+                    .filter(|(name, _)| {
+                        service_filter.is_empty() || service_filter.contains(name)
+                    })
+                    .map(|(name, url)| {
+                        let name = name.clone();
+                        let url = url.clone();
+                        async move { Self::check_backend_health(&name, &url).await }
+                    })
+                    .collect();
+
+                let statuses = futures::future::join_all(checks).await;
+
+                let healthy_count = statuses
+                    .iter()
+                    .filter(|s| s.health == i32::from(ServiceHealth::Healthy))
+                    .count();
+                let total = statuses.len();
+
+                #[allow(clippy::cast_possible_truncation)]
+                let overall_health = if healthy_count == total {
+                    SystemHealth::Healthy
+                } else if healthy_count == 0 {
+                    SystemHealth::Critical
+                } else if healthy_count * 2 >= total {
+                    SystemHealth::Degraded
+                } else {
+                    SystemHealth::Unhealthy
+                };
+
+                let critical_issues: Vec<String> = statuses
+                    .iter()
+                    .filter(|s| s.health != i32::from(ServiceHealth::Healthy))
+                    .map(|s| {
+                        format!(
+                            "{}: {}",
+                            s.service_name,
+                            s.error_message.as_deref().unwrap_or("unhealthy")
+                        )
+                    })
+                    .collect();
+
+                #[allow(clippy::cast_possible_truncation)]
+                let system_status = SystemStatus {
+                    overall_health: overall_health.into(),
+                    healthy_services: healthy_count as i32,
+                    total_services: total as i32,
+                    critical_issues,
+                    system_uptime_seconds: start_time.elapsed().as_secs() as i64,
+                    system_metrics: Some(SystemMetrics::default()),
+                };
+
+                let event = SystemStatusEvent {
+                    system_status: Some(system_status),
+                    change_type: SystemStatusChangeType::Unspecified.into(),
+                    timestamp: chrono::Utc::now().timestamp(),
+                };
+
+                yield Ok(event);
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
     }
 
     #[instrument(skip(self, request), err)]
@@ -606,13 +695,77 @@ impl MonitoringService for MonitoringServiceHandler {
         }))
     }
 
+    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
     async fn stream_metrics(
         &self,
-        _request: Request<crate::monitoring::StreamMetricsRequest>,
+        request: Request<crate::monitoring::StreamMetricsRequest>,
     ) -> Result<Response<Self::StreamMetricsStream>, Status> {
-        Err(Status::unimplemented(
-            "StreamMetrics not yet implemented",
-        ))
+        let req = request.into_inner();
+        let interval_secs = req.update_frequency_seconds.unwrap_or(0);
+        let interval_secs = if interval_secs == 0 {
+            5u32
+        } else {
+            (interval_secs as u32).clamp(1, 60)
+        };
+        let metric_names: Vec<String> = if req.metric_names.is_empty() {
+            vec![
+                "cpu_usage".to_owned(),
+                "memory_usage".to_owned(),
+                "gpu_utilization".to_owned(),
+            ]
+        } else {
+            req.metric_names
+        };
+        let prom = self.prom.clone();
+
+        info!(
+            "StreamMetrics: interval={}s, metrics={:?}",
+            interval_secs, metric_names
+        );
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+
+                let mut metrics = Vec::new();
+                for name in &metric_names {
+                    match prom.query(name).await {
+                        Ok(results) => {
+                            for r in results {
+                                if let Ok(value) = r.value.1.parse::<f64>() {
+                                    metrics.push(Metric {
+                                        name: r
+                                            .metric
+                                            .get("__name__")
+                                            .cloned()
+                                            .unwrap_or_else(|| name.clone()),
+                                        metric_type: MetricType::Gauge.into(),
+                                        value,
+                                        unit: String::new(),
+                                        labels: r.metric,
+                                        timestamp: chrono::Utc::now().timestamp(),
+                                        statistics: None,
+                                    });
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            error!("StreamMetrics: Prometheus query for '{}' failed: {}", name, e);
+                        }
+                    }
+                }
+
+                let event = MetricsEvent {
+                    metrics,
+                    timestamp: chrono::Utc::now().timestamp(),
+                };
+
+                yield Ok(event);
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
     }
 
     async fn get_latency_metrics(
@@ -636,13 +789,33 @@ impl MonitoringService for MonitoringServiceHandler {
     type StreamAlertsStream =
         Pin<Box<dyn Stream<Item = Result<AlertEvent, Status>> + Send>>;
 
+    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
     async fn stream_alerts(
         &self,
-        _request: Request<crate::monitoring::StreamAlertsRequest>,
+        request: Request<crate::monitoring::StreamAlertsRequest>,
     ) -> Result<Response<Self::StreamAlertsStream>, Status> {
-        Err(Status::unimplemented(
-            "StreamAlerts not yet implemented",
-        ))
+        let _req = request.into_inner();
+
+        info!("StreamAlerts: interval=10s (placeholder, no alertmanager integration yet)");
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(10));
+            loop {
+                interval.tick().await;
+
+                // Placeholder: yields an empty AlertEvent each tick.
+                // Will be wired to Prometheus Alertmanager when available.
+                let event = AlertEvent {
+                    alert: None,
+                    event_type: AlertEventType::Unspecified.into(),
+                    timestamp: chrono::Utc::now().timestamp(),
+                };
+
+                yield Ok(event);
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
     }
 
     async fn acknowledge_alert(

services/api_gateway/src/grpc/risk_proxy.rs

@@ -4,6 +4,7 @@
 
 use futures::Stream;
 use std::pin::Pin;
+use std::time::Duration;
 use tonic::{Request, Response, Status};
 use tracing::{error, instrument};
 
@@ -13,7 +14,8 @@ use crate::risk::{
     EmergencyStopRequest, EmergencyStopResponse, GetCircuitBreakerStatusRequest,
     GetCircuitBreakerStatusResponse, GetPositionRiskRequest, GetPositionRiskResponse,
     GetRiskMetricsRequest, GetRiskMetricsResponse, GetVaRRequest, GetVaRResponse,
-    RiskAlertEvent, StreamRiskAlertsRequest, StreamVaRRequest, VaREvent, ValidateOrderRequest,
+    RiskAlertEvent, StreamCircuitBreakerStatusRequest, StreamRiskAlertsRequest,
+    StreamRiskMetricsRequest, StreamVaRRequest, VaREvent, ValidateOrderRequest,
     ValidateOrderResponse,
 };
 
@@ -33,6 +35,10 @@ impl RiskService for RiskServiceProxy {
     type StreamVaRUpdatesStream = Pin<Box<dyn Stream<Item = Result<VaREvent, Status>> + Send>>;
     type StreamRiskAlertsStream =
         Pin<Box<dyn Stream<Item = Result<RiskAlertEvent, Status>> + Send>>;
+    type StreamCircuitBreakerStatusStream =
+        Pin<Box<dyn Stream<Item = Result<GetCircuitBreakerStatusResponse, Status>> + Send>>;
+    type StreamRiskMetricsStream =
+        Pin<Box<dyn Stream<Item = Result<GetRiskMetricsResponse, Status>> + Send>>;
 
     #[instrument(skip(self, request), err)]
     async fn get_va_r(
@@ -153,4 +159,72 @@ impl RiskService for RiskServiceProxy {
                 e
             })
     }
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_circuit_breaker_status(
+        &self,
+        request: Request<StreamCircuitBreakerStatusRequest>,
+    ) -> Result<Response<Self::StreamCircuitBreakerStatusStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            2
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let symbol = req.symbol;
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_circuit_breaker_status(Request::new(GetCircuitBreakerStatusRequest {
+                    symbol: symbol.clone(),
+                })).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetCircuitBreakerStatus failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_risk_metrics(
+        &self,
+        request: Request<StreamRiskMetricsRequest>,
+    ) -> Result<Response<Self::StreamRiskMetricsStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            3
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let portfolio_id = req.portfolio_id;
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_risk_metrics(Request::new(GetRiskMetricsRequest {
+                    portfolio_id: portfolio_id.clone(),
+                })).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetRiskMetrics failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
 }

services/api_gateway/src/grpc/trading_agent_proxy.rs

@@ -7,8 +7,10 @@
 //! - Efficient streaming support for agent activity events
 //! - Health checking integration
 
-use futures::Stream;
 use std::pin::Pin;
+use std::time::Duration;
+
+use futures::Stream;
 use tonic::{Request, Response, Status};
 use tracing::{error, info, instrument, warn};
 
@@ -56,6 +58,7 @@ use crate::trading_agent::{
     SelectUniverseRequest,
     SelectUniverseResponse,
     StreamAgentActivityRequest,
+    StreamAgentStatusRequest,
     SubmitAgentOrdersRequest,
     SubmitAgentOrdersResponse,
 
@@ -101,6 +104,9 @@ impl TradingAgentService for TradingAgentProxy {
     /// Server streaming type for agent activity events
     type StreamAgentActivityStream =
         Pin<Box<dyn Stream<Item = Result<AgentActivityEvent, Status>> + Send>>;
+    /// Server streaming type for poll-based agent status
+    type StreamAgentStatusStream =
+        Pin<Box<dyn Stream<Item = Result<GetAgentStatusResponse, Status>> + Send>>;
 
     // ===== Universe Management Methods =====
 
@@ -499,6 +505,40 @@ impl TradingAgentService for TradingAgentProxy {
         info!("HealthCheck request forwarded successfully");
         Ok(response)
     }
+
+    /// Poll-to-stream adapter for agent status
+    ///
+    /// Polls GetAgentStatus at a configurable interval and yields results as a server stream.
+    #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)]
+    async fn stream_agent_status(
+        &self,
+        request: Request<StreamAgentStatusRequest>,
+    ) -> Result<Response<Self::StreamAgentStatusStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            3
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_agent_status(Request::new(GetAgentStatusRequest::default())).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetAgentStatus failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
 }
 
 #[cfg(test)]

services/api_gateway/src/grpc/trading_direct_proxy.rs

@@ -3,8 +3,10 @@
 //! Forwards requests from fxt CLI (package: trading) directly to trading-service backend.
 //! Distinct from TradingServiceProxy which serves the monolithic foxhunt.tli.TradingService.
 
-use futures::Stream;
 use std::pin::Pin;
+use std::time::Duration;
+
+use futures::Stream;
 use tonic::{Request, Response, Status};
 use tracing::{error, instrument};
 
@@ -18,8 +20,9 @@ use crate::trading_backend::{
     GetRegimeStateResponse, GetRegimeTransitionsRequest, GetRegimeTransitionsResponse,
     MarketDataEvent, MlOrderRequest, MlOrderResponse, MlPerformanceRequest,
     MlPerformanceResponse, MlPredictionsRequest, MlPredictionsResponse, OrderEvent,
-    PositionEvent, StreamExecutionsRequest, StreamMarketDataRequest, StreamOrdersRequest,
-    StreamPositionsRequest, SubmitOrderRequest, SubmitOrderResponse,
+    PositionEvent, StreamExecutionsRequest, StreamMarketDataRequest, StreamOrderBookRequest,
+    StreamOrdersRequest, StreamPortfolioSummaryRequest, StreamPositionsRequest,
+    SubmitOrderRequest, SubmitOrderResponse,
 };
 
 #[derive(Debug, Clone)]
@@ -42,6 +45,10 @@ impl TradingService for TradingDirectProxy {
         Pin<Box<dyn Stream<Item = Result<MarketDataEvent, Status>> + Send>>;
     type StreamExecutionsStream =
         Pin<Box<dyn Stream<Item = Result<ExecutionEvent, Status>> + Send>>;
+    type StreamPortfolioSummaryStream =
+        Pin<Box<dyn Stream<Item = Result<GetPortfolioSummaryResponse, Status>> + Send>>;
+    type StreamOrderBookStream =
+        Pin<Box<dyn Stream<Item = Result<GetOrderBookResponse, Status>> + Send>>;
 
     #[instrument(skip(self, request), err)]
     async fn submit_order(
@@ -256,4 +263,74 @@ impl TradingService for TradingDirectProxy {
                 e
             })
     }
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_portfolio_summary(
+        &self,
+        request: Request<StreamPortfolioSummaryRequest>,
+    ) -> Result<Response<Self::StreamPortfolioSummaryStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            3
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let account_id = req.account_id;
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_portfolio_summary(Request::new(GetPortfolioSummaryRequest {
+                    account_id: account_id.clone(),
+                })).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetPortfolioSummary failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
+
+    #[instrument(skip(self, request), err)]
+    async fn stream_order_book(
+        &self,
+        request: Request<StreamOrderBookRequest>,
+    ) -> Result<Response<Self::StreamOrderBookStream>, Status> {
+        let req = request.into_inner();
+        let interval_secs = if req.interval_seconds == 0 {
+            1
+        } else {
+            req.interval_seconds.clamp(1, 60)
+        };
+        let symbol = req.symbol;
+        let depth = req.depth;
+        let mut client = self.client.clone();
+
+        let stream = async_stream::stream! {
+            let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs)));
+            loop {
+                interval.tick().await;
+                match client.get_order_book(Request::new(GetOrderBookRequest {
+                    symbol: symbol.clone(),
+                    depth: if depth == 0 { None } else { Some(depth) },
+                })).await {
+                    Ok(resp) => yield Ok(resp.into_inner()),
+                    Err(e) => {
+                        error!("Backend GetOrderBook failed: {}", e);
+                        yield Err(e);
+                        break;
+                    }
+                }
+            }
+        };
+
+        Ok(Response::new(Box::pin(stream)))
+    }
 }