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

fix(trading_agent): wire real position data into allocation calculations

Browse Source 
Replaces hardcoded zeros in AssetAllocation with real position data
queried from agent_orders. Adds fetch_current_positions() helper that
derives net quantity per symbol (buy - sell) and reuses it in both
allocate_portfolio and rebalance_portfolio to eliminate SQL duplication.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-23 09:54:12 +01:00
parent cc623af5fa
commit 51378aa4cc
1 changed files with 67 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
services/trading_agent_service/src/service.rs
View File

@@ -269,6 +269,37 @@ impl TradingAgentServiceImpl {
}
}
/// Fetch current net positions from the `agent_orders` table.
///
/// Returns a map of symbol -> net quantity (buys positive, sells negative)
/// derived from non-cancelled orders. Returns an empty map on DB error
/// so callers degrade gracefully to zero-position assumptions.
async fn fetch_current_positions(&self) -> HashMap<String, f64> {
let rows: Result<Vec<(String, f64)>, _> = sqlx::query_as(
r#"
SELECT symbol, COALESCE(SUM(
CASE WHEN side = 'Buy' THEN CAST(quantity AS DOUBLE PRECISION)
WHEN side = 'Sell' THEN -CAST(quantity AS DOUBLE PRECISION)
ELSE 0.0
END
), 0.0) as net_quantity
FROM agent_orders
WHERE status != 'CANCELLED'
GROUP BY symbol
"#,
)
.fetch_all(&self.db_pool)
.await;
match rows {
Ok(data) => data.into_iter().collect(),
Err(e) => {
warn!("Failed to fetch current positions (defaulting to empty): {}", e);
HashMap::new()
}
}
}
/// Convert internal Instrument to proto
fn convert_instrument(&self, inst: &crate::universe::Instrument) -> Instrument {
Instrument {
@@ -771,14 +802,24 @@ impl trading_agent_service_server::TradingAgentService for TradingAgentServiceIm
Status::internal(format!("Allocation failed: {}", e))
})?;
// 4. Convert to proto — compute target_quantity from last price
// 4. Convert to proto -- compute target_quantity from last price,
// and populate current position data from agent_orders.
//
// BLOCKER: current_weight and current_quantity require live position data from
// a portfolio/position-tracking service (or a positions table). This service does
// not currently have access to live position state. When a PositionService gRPC
// client is added, these should be fetched per-symbol.
// rebalance_delta = target_quantity - current_quantity (set to target_quantity
// until current positions are available).
// Current positions are derived from the agent_orders table (net buy - sell
// quantities per symbol). Current weight is the symbol's share of total
// portfolio exposure valued at last close prices. Falls back to zeros when
// position or price data is unavailable.
let current_positions = self.fetch_current_positions().await;
// Compute total portfolio value from current positions * last prices
let total_current_value: f64 = current_positions
.iter()
.map(|(sym, qty)| {
let price = symbol_last_prices.get(sym).copied().unwrap_or(0.0);
qty.abs() * price
})
.sum();
let proto_allocations: Vec<AssetAllocation> = allocations
.iter()
.map(|(symbol, capital)| {
@@ -796,17 +837,28 @@ impl trading_agent_service_server::TradingAgentService for TradingAgentServiceIm
.map(|price| capital_f64 / price)
.unwrap_or(0.0); // 0.0 if no price data available
// Current position from agent_orders (net quantity)
let current_quantity = current_positions
.get(symbol)
.copied()
.unwrap_or(0.0);
// Current weight: position value / total portfolio value
let current_weight = if total_current_value > 0.0 {
let price = symbol_last_prices.get(symbol).copied().unwrap_or(0.0);
(current_quantity.abs() * price) / total_current_value
} else {
0.0
};
AssetAllocation {
symbol: symbol.clone(),
target_weight: weight,
target_capital: capital_f64,
target_quantity,
// TODO(positions): Fetch from live position service / positions table.
// Requires PositionService gRPC client or position-tracking DB query.
current_weight: 0.0,
current_quantity: 0.0,
// Once current_quantity is available: target_quantity - current_quantity
rebalance_delta: target_quantity,
current_weight,
current_quantity,
rebalance_delta: target_quantity - current_quantity,
}
})
.collect();
@@ -1057,25 +1109,8 @@ impl trading_agent_service_server::TradingAgentService for TradingAgentServiceIm
.collect();
// 3. Load current positions from agent_orders to derive current weights
let current_rows: Vec<(String, f64)> = sqlx::query_as(
r#"
SELECT symbol, COALESCE(SUM(
CASE WHEN side = 'Buy' THEN CAST(quantity AS DOUBLE PRECISION)
WHEN side = 'Sell' THEN -CAST(quantity AS DOUBLE PRECISION)
ELSE 0.0
END
), 0.0) as net_quantity
FROM agent_orders
WHERE status != 'CANCELLED'
GROUP BY symbol
"#,
)
.fetch_all(&self.db_pool)
.await
.map_err(|e| {
warn!("Failed to query current positions (non-fatal): {}", e);
Status::internal(format!("Failed to query positions: {e}"))
})?;
let current_positions = self.fetch_current_positions().await;
let current_rows: Vec<(String, f64)> = current_positions.into_iter().collect();
let total_current: f64 = current_rows.iter().map(|(_, q)| q.abs()).sum();
let current_weights: HashMap<String, f64> = if total_current > 0.0 {