docs: TUI live data wiring design
Streaming-first architecture: 11 new poll-to-stream gateway adapters, channel-based DataFetcher in the TUI, auto-reconnect with backoff. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
178
docs/plans/2026-03-04-tui-live-data-design.md
Normal file
178
docs/plans/2026-03-04-tui-live-data-design.md
Normal file
@@ -0,0 +1,178 @@
|
||||
# TUI Live Data Wiring — Design
|
||||
|
||||
**Date:** 2026-03-04
|
||||
**Goal:** Replace all hardcoded mock data in `fxt watch` with real gRPC streaming data from the API Gateway.
|
||||
|
||||
## Architecture
|
||||
|
||||
### Data Flow
|
||||
|
||||
```
|
||||
gRPC Streams (19 total)
|
||||
│
|
||||
▼
|
||||
DataFetcher (tokio tasks, one per stream)
|
||||
│
|
||||
▼ mpsc::UnboundedSender<StateUpdate>
|
||||
│
|
||||
event_loop (drains channel each tick, applies to AppState, renders)
|
||||
```
|
||||
|
||||
- `DataFetcher` spawns one tokio task per gRPC stream
|
||||
- Each task sends `StateUpdate` enum variants through an unbounded mpsc channel
|
||||
- Event loop owns `AppState` exclusively — no shared mutable state, no locks
|
||||
- On stream disconnect, tasks auto-reconnect with exponential backoff (1s → 2s → 4s → max 30s)
|
||||
|
||||
### StateUpdate Enum
|
||||
|
||||
```rust
|
||||
enum StateUpdate {
|
||||
Positions(Vec<PositionData>),
|
||||
Executions(Vec<ExecutionData>),
|
||||
Orders(Vec<OrderData>),
|
||||
AccountState(AccountData),
|
||||
PortfolioSummary(PortfolioData),
|
||||
OrderBook(OrderBookData),
|
||||
Services(Vec<ServiceData>),
|
||||
SystemStatus(SystemStatusData),
|
||||
Metrics(MetricsData),
|
||||
RiskMetrics(RiskData),
|
||||
CircuitBreakers(Vec<CircuitBreakerData>),
|
||||
VaRUpdate(VaRData),
|
||||
RiskAlerts(Vec<RiskAlertData>),
|
||||
TrainingSessions(Vec<TrainingSessionData>),
|
||||
TrainingJobStatus(TrainingJobData),
|
||||
ModelStatus(ModelStatusData),
|
||||
ModelMetrics(ModelMetricsData),
|
||||
AgentStatus(AgentStatusData),
|
||||
AgentActivity(AgentActivityData),
|
||||
DataFeeds(Vec<DataFeedData>),
|
||||
DownloadStatus(Vec<DownloadJobData>),
|
||||
SessionStatus(SessionStatusData),
|
||||
SignalStrength(SignalStrengthData),
|
||||
ConnectionStatus(ConnectionStatus),
|
||||
}
|
||||
```
|
||||
|
||||
## Proto Changes — 11 New Streaming RPCs
|
||||
|
||||
All new RPCs are gateway-level poll-to-stream adapters. The backend services remain unchanged.
|
||||
|
||||
| # | Proto | New RPC | Wrapped Unary | Poll Interval |
|
||||
|---|---|---|---|---|
|
||||
| 1 | `broker_gateway.proto` | `StreamAccountState` | `GetAccountState` | 3s |
|
||||
| 2 | `broker_gateway.proto` | `StreamSessionStatus` | `GetSessionStatus` | 5s |
|
||||
| 3 | `risk.proto` | `StreamCircuitBreakerStatus` | `GetCircuitBreakerStatus` | 2s |
|
||||
| 4 | `risk.proto` | `StreamRiskMetrics` | `GetRiskMetrics` | 3s |
|
||||
| 5 | `data_acquisition.proto` | `StreamDownloadStatus` | `ListDownloadJobs` | 5s |
|
||||
| 6 | `ml.proto` | `StreamModelStatus` | `GetModelStatus` | 5s |
|
||||
| 7 | `trading_agent.proto` | `StreamAgentStatus` | `GetAgentStatus` | 3s |
|
||||
| 8 | `monitoring.proto` | `StreamServiceHealth` | `GetHealthCheck` | 5s |
|
||||
| 9 | `trading.proto` | `StreamPortfolioSummary` | `GetPortfolioSummary` | 3s |
|
||||
| 10 | `ml_training.proto` | `StreamTrainingJobStatus` | `GetTrainingJobDetails` | 3s |
|
||||
| 11 | `trading.proto` | `StreamOrderBook` | `GetOrderBook` | 1s |
|
||||
|
||||
### Existing Streaming RPCs (8 — already proxied, just need TUI clients)
|
||||
|
||||
| Proto | Existing Stream | TUI Cockpit |
|
||||
|---|---|---|
|
||||
| `trading.proto` | `StreamPositions` | Trading |
|
||||
| `trading.proto` | `StreamOrders` | Trading |
|
||||
| `trading.proto` | `StreamExecutions` | Trading |
|
||||
| `trading.proto` | `StreamMarketData` | Data |
|
||||
| `risk.proto` | `StreamVaRUpdates` | Risk |
|
||||
| `risk.proto` | `StreamRiskAlerts` | Risk |
|
||||
| `monitoring.proto` | `StreamSystemStatus` | Services/Overview |
|
||||
| `monitoring.proto` | `StreamMetrics` | Services/Overview |
|
||||
| `monitoring.proto` | `StreamTrainingMetrics` | Training |
|
||||
| `monitoring.proto` | `StreamAlerts` | Services |
|
||||
| `ml.proto` | `StreamPredictions` | Overview |
|
||||
| `ml.proto` | `StreamModelMetrics` | Training/Overview |
|
||||
| `ml.proto` | `StreamSignalStrength` | Overview |
|
||||
| `trading_agent.proto` | `StreamAgentActivity` | Overview |
|
||||
|
||||
## API Gateway — Poll-to-Stream Adapter Pattern
|
||||
|
||||
Each new streaming RPC in the gateway follows this pattern:
|
||||
|
||||
```rust
|
||||
async fn stream_foo(&self, _request: Request<StreamFooRequest>)
|
||||
-> Result<Response<Self::StreamFooStream>, Status>
|
||||
{
|
||||
let mut client = self.client.clone();
|
||||
let (tx, rx) = tokio::sync::mpsc::channel(16);
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
match client.get_foo(Request::new(GetFooRequest {})).await {
|
||||
Ok(resp) => {
|
||||
if tx.send(Ok(resp.into_inner())).await.is_err() {
|
||||
break; // client disconnected
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Backend GetFoo failed: {}", e);
|
||||
let _ = tx.send(Err(e)).await;
|
||||
break;
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(Duration::from_secs(N)).await;
|
||||
}
|
||||
});
|
||||
|
||||
Ok(Response::new(ReceiverStream::new(rx)))
|
||||
}
|
||||
```
|
||||
|
||||
## TUI Changes
|
||||
|
||||
### New File: `bin/fxt/src/tui/data_fetcher.rs`
|
||||
|
||||
- `DataFetcher::new(client: FoxhuntClient, tx: UnboundedSender<StateUpdate>)`
|
||||
- `DataFetcher::start()` — spawns all stream tasks
|
||||
- Each task: open stream → loop { recv → convert to StateUpdate → send on channel }
|
||||
- Reconnect on error: exponential backoff 1s → 2s → 4s → max 30s
|
||||
- Graceful shutdown via `CancellationToken`
|
||||
|
||||
### Modified Files
|
||||
|
||||
- `event_loop.rs`: Accept `FoxhuntClient`, create `DataFetcher`, drain mpsc channel each tick
|
||||
- `state.rs`: Add `ConnectionStatus` enum, make `Default` start with empty vecs
|
||||
- `watch.rs`: Create `FoxhuntClient` with auth, pass to event loop
|
||||
- 6 cockpit files: Handle empty state (show "Connecting..." or "No data")
|
||||
|
||||
### ConnectionStatus
|
||||
|
||||
```rust
|
||||
enum ConnectionStatus {
|
||||
Connecting,
|
||||
Connected,
|
||||
Reconnecting { attempt: u32 },
|
||||
Error(String),
|
||||
}
|
||||
```
|
||||
|
||||
Displayed in the status bar: "Connected to api.fxhnt.ai" / "Reconnecting (attempt 3)..." / etc.
|
||||
|
||||
## Cockpit → Stream Mapping
|
||||
|
||||
| Cockpit | Streams Used |
|
||||
|---|---|
|
||||
| Overview | StreamSystemStatus, StreamPortfolioSummary, StreamAgentStatus, StreamModelMetrics |
|
||||
| Training | StreamTrainingMetrics, StreamTrainingJobStatus, StreamModelStatus |
|
||||
| Trading | StreamPositions, StreamExecutions, StreamOrders, StreamAccountState, StreamOrderBook, StreamPortfolioSummary, StreamSessionStatus |
|
||||
| Services | StreamServiceHealth, StreamSystemStatus, StreamMetrics, StreamAlerts |
|
||||
| Risk | StreamVaRUpdates, StreamRiskAlerts, StreamCircuitBreakerStatus, StreamRiskMetrics |
|
||||
| Data | StreamMarketData, StreamDownloadStatus |
|
||||
|
||||
## File Impact Summary
|
||||
|
||||
| Layer | Files | Type |
|
||||
|---|---|---|
|
||||
| Proto | 7 `.proto` files | Modify (additive) |
|
||||
| Gateway build | `build.rs` | Modify (regenerate) |
|
||||
| Gateway proxies | 7 proxy files | Modify (add stream methods) |
|
||||
| FXT TUI | `data_fetcher.rs` | New |
|
||||
| FXT TUI | `event_loop.rs`, `state.rs`, `watch.rs` | Modify |
|
||||
| FXT TUI | 6 cockpit files | Modify (handle empty state) |
|
||||
| FXT build | `bin/fxt/build.rs` | May need regeneration |
|
||||
Reference in New Issue
Block a user