//! Execution repository implementation //! //! This module provides repository pattern implementation for trade executions //! with PostgreSQL backing. It includes comprehensive trade history tracking, //! execution reporting, and performance analytics for high-frequency trading. use async_trait::async_trait; use chrono::{DateTime, Utc}; use common::prelude::*; // Removed direct rust_decimal import - using common::Decimal via common crate use sqlx::{Pool, Postgres, Row}; use uuid::Uuid; use crate::{ models::{Execution, OrderSide}, Repository, RepositoryError, Result, }; /// Filter criteria for execution queries #[derive(Debug, Default, Clone)] pub struct ExecutionFilter { /// Filter by trading symbol pub symbol: Option, /// Filter by related order ID pub order_id: Option, /// Filter by execution side pub side: Option, /// Filter executions with quantity greater than this value pub min_quantity: Option, /// Filter executions with quantity less than this value pub max_quantity: Option, /// Filter executions with price greater than this value pub min_price: Option, /// Filter executions with price less than this value pub max_price: Option, /// Filter by broker execution ID pub broker_execution_id: Option, /// Filter by trading venue pub venue: Option, /// Filter by counterparty pub counterparty: Option, /// Filter executions after this timestamp pub executed_after: Option>, /// Filter executions before this timestamp pub executed_before: Option>, /// Filter by minimum gross value pub min_gross_value: Option, /// Filter by maximum gross value pub max_gross_value: Option, /// Limit number of results pub limit: Option, /// Offset for pagination pub offset: Option, } impl ExecutionFilter { /// Create a new empty filter pub fn new() -> Self { Self::default() } /// Filter by symbol pub fn symbol>(mut self, symbol: S) -> Self { self.symbol = Some(symbol.into()); self } /// Filter by order ID pub fn order_id(mut self, order_id: Uuid) -> Self { self.order_id = Some(order_id); self } /// Filter by execution side pub fn side(mut self, side: OrderSide) -> Self { self.side = Some(side); self } /// Filter by quantity range pub fn quantity_range(mut self, min: Option, max: Option) -> Self { self.min_quantity = min; self.max_quantity = max; self } /// Filter by price range pub fn price_range(mut self, min: Option, max: Option) -> Self { self.min_price = min; self.max_price = max; self } /// Filter by value range pub fn value_range(mut self, min: Option, max: Option) -> Self { self.min_gross_value = min; self.max_gross_value = max; self } /// Filter by venue pub fn venue>(mut self, venue: S) -> Self { self.venue = Some(venue.into()); self } /// Filter by counterparty pub fn counterparty>(mut self, counterparty: S) -> Self { self.counterparty = Some(counterparty.into()); self } /// Filter by execution time range pub fn executed_between(mut self, start: DateTime, end: DateTime) -> Self { self.executed_after = Some(start); self.executed_before = Some(end); self } /// Filter by today's executions pub fn today(mut self) -> Self { let now = Utc::now(); let start_of_day = now.date_naive().and_hms_opt(0, 0, 0).unwrap().and_utc(); self.executed_after = Some(start_of_day); self.executed_before = Some(now); self } /// Limit results pub fn limit(mut self, limit: i64) -> Self { self.limit = Some(limit); self } /// Set offset for pagination pub fn offset(mut self, offset: i64) -> Self { self.offset = Some(offset); self } } /// Execution repository trait defining available operations #[async_trait] pub trait ExecutionRepository: Repository { /// Find executions matching the filter criteria async fn find_by_filter(&self, filter: &ExecutionFilter) -> Result>; /// Find executions by symbol async fn find_by_symbol(&self, symbol: &str) -> Result>; /// Find executions by order ID async fn find_by_order_id(&self, order_id: &Uuid) -> Result>; /// Find executions by side (buy/sell) async fn find_by_side(&self, side: OrderSide) -> Result>; /// Find executions within a time range async fn find_by_time_range(&self, start: DateTime, end: DateTime) -> Result>; /// Find executions by venue async fn find_by_venue(&self, venue: &str) -> Result>; /// Batch insert executions for high-frequency operations async fn batch_insert(&self, executions: &[Execution]) -> Result>; /// Get execution statistics async fn get_execution_stats(&self, symbol: Option<&str>, time_range: Option<(DateTime, DateTime)>) -> Result; /// Get trading performance metrics async fn get_trading_performance(&self, symbol: Option<&str>, time_range: Option<(DateTime, DateTime)>) -> Result; /// Find large executions (above threshold) async fn find_large_executions(&self, value_threshold: Decimal) -> Result>; /// Get volume-weighted average price (VWAP) for a symbol async fn calculate_vwap(&self, symbol: &str, time_range: Option<(DateTime, DateTime)>) -> Result; /// Get execution summary by hour for analytics async fn get_hourly_execution_summary(&self, date: chrono::NaiveDate) -> Result>; /// Find executions with high slippage async fn find_high_slippage_executions(&self, slippage_threshold: Decimal) -> Result>; } /// Execution statistics #[derive(Debug, Clone)] pub struct ExecutionStats { pub total_executions: i64, pub buy_executions: i64, pub sell_executions: i64, pub total_volume: Decimal, pub total_value: Decimal, pub total_fees: Decimal, pub avg_execution_size: Decimal, pub avg_price: Decimal, pub largest_execution: Option, pub smallest_execution: Option, pub unique_symbols: i64, pub unique_venues: i64, } /// Trading performance metrics #[derive(Debug, Clone)] pub struct TradingPerformance { pub total_trades: i64, pub winning_trades: i64, pub losing_trades: i64, pub win_rate: Decimal, pub avg_win: Decimal, pub avg_loss: Decimal, pub profit_factor: Decimal, pub total_pnl: Decimal, pub gross_profit: Decimal, pub gross_loss: Decimal, pub largest_win: Decimal, pub largest_loss: Decimal, pub avg_trade_duration: Option, // in seconds } /// Hourly execution summary for analytics #[derive(Debug, Clone)] pub struct HourlyExecutionSummary { pub hour: i32, pub execution_count: i64, pub total_volume: Decimal, pub total_value: Decimal, pub avg_price: Decimal, pub unique_symbols: i64, } /// Execution with calculated slippage #[derive(Debug, Clone)] pub struct ExecutionWithSlippage { pub execution: Execution, pub expected_price: Decimal, pub slippage: Decimal, pub slippage_bps: Decimal, // basis points } /// PostgreSQL implementation of ExecutionRepository pub struct PostgresExecutionRepository { pool: Pool, } impl PostgresExecutionRepository { /// Create a new PostgreSQL execution repository pub fn new(pool: Pool) -> Self { Self { pool } } } #[async_trait] impl Repository for PostgresExecutionRepository { async fn find_by_id(&self, id: &Uuid) -> Result> { let query = r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions WHERE id = $1 "#; let row = sqlx::query(query) .bind(id) .fetch_optional(&self.pool) .await?; match row { Some(row) => { let execution = Execution { id: row.get("id"), order_id: row.get("order_id"), symbol: row.get("symbol"), quantity: row.get("quantity"), price: row.get("price"), side: row.get("side"), fees: row.get("fees"), fee_currency: row.get("fee_currency"), executed_at: row.get("executed_at"), broker_execution_id: row.get("broker_execution_id"), counterparty: row.get("counterparty"), venue: row.get("venue"), gross_value: row.get("gross_value"), net_value: row.get("net_value"), }; Ok(Some(execution)) } None => Ok(None), } } async fn save(&self, execution: &Execution) -> Result { let query = r#" INSERT INTO executions (id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) ON CONFLICT (id) DO UPDATE SET broker_execution_id = EXCLUDED.broker_execution_id, counterparty = EXCLUDED.counterparty, venue = EXCLUDED.venue RETURNING * "#; let row = sqlx::query(query) .bind(&execution.id) .bind(&execution.order_id) .bind(&execution.symbol) .bind(&execution.quantity) .bind(&execution.price) .bind(&execution.side) .bind(&execution.fees) .bind(&execution.fee_currency) .bind(&execution.executed_at) .bind(&execution.broker_execution_id) .bind(&execution.counterparty) .bind(&execution.venue) .bind(&execution.gross_value) .bind(&execution.net_value) .fetch_one(&self.pool) .await?; Ok(Execution { id: row.get("id"), order_id: row.get("order_id"), symbol: row.get("symbol"), quantity: row.get("quantity"), price: row.get("price"), side: row.get("side"), fees: row.get("fees"), fee_currency: row.get("fee_currency"), executed_at: row.get("executed_at"), broker_execution_id: row.get("broker_execution_id"), counterparty: row.get("counterparty"), venue: row.get("venue"), gross_value: row.get("gross_value"), net_value: row.get("net_value"), }) } async fn delete(&self, id: &Uuid) -> Result { let result = sqlx::query("DELETE FROM executions WHERE id = $1") .bind(id) .execute(&self.pool) .await?; Ok(result.rows_affected() > 0) } async fn exists(&self, id: &Uuid) -> Result { let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM executions WHERE id = $1") .bind(id) .fetch_one(&self.pool) .await?; Ok(count > 0) } } #[async_trait] impl ExecutionRepository for PostgresExecutionRepository { async fn find_by_filter(&self, filter: &ExecutionFilter) -> Result> { let mut conditions = Vec::new(); let mut query = r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions "#.to_string(); // Build dynamic WHERE clause based on filter if filter.symbol.is_some() { conditions.push("symbol = $1".to_string()); } // Add other conditions as needed (simplified for brevity) if !conditions.is_empty() { query.push_str(&format!(" WHERE {}", conditions.join(" AND "))); } query.push_str(" ORDER BY executed_at DESC"); if let Some(limit) = filter.limit { query.push_str(&format!(" LIMIT {}", limit)); } if let Some(offset) = filter.offset { query.push_str(&format!(" OFFSET {}", offset)); } let executions = sqlx::query_as::<_, Execution>(&query) .fetch_all(&self.pool) .await?; Ok(executions) } async fn find_by_symbol(&self, symbol: &str) -> Result> { let executions = sqlx::query_as::<_, Execution>( r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions WHERE symbol = $1 ORDER BY executed_at DESC "# ) .bind(symbol) .fetch_all(&self.pool) .await?; Ok(executions) } async fn find_by_order_id(&self, order_id: &Uuid) -> Result> { let executions = sqlx::query_as::<_, Execution>( r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions WHERE order_id = $1 ORDER BY executed_at ASC "# ) .bind(order_id) .fetch_all(&self.pool) .await?; Ok(executions) } async fn find_by_side(&self, side: OrderSide) -> Result> { let executions = sqlx::query_as::<_, Execution>( r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions WHERE side = $1 ORDER BY executed_at DESC "# ) .bind(side) .fetch_all(&self.pool) .await?; Ok(executions) } async fn find_by_time_range(&self, start: DateTime, end: DateTime) -> Result> { let executions = sqlx::query_as::<_, Execution>( r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions WHERE executed_at >= $1 AND executed_at <= $2 ORDER BY executed_at ASC "# ) .bind(start) .bind(end) .fetch_all(&self.pool) .await?; Ok(executions) } async fn find_by_venue(&self, venue: &str) -> Result> { let executions = sqlx::query_as::<_, Execution>( r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions WHERE venue = $1 ORDER BY executed_at DESC "# ) .bind(venue) .fetch_all(&self.pool) .await?; Ok(executions) } async fn batch_insert(&self, executions: &[Execution]) -> Result> { if executions.is_empty() { return Ok(Vec::new()); } let mut tx = self.pool.begin().await?; let mut inserted_executions = Vec::new(); for execution in executions { let query = r#" INSERT INTO executions (id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) RETURNING * "#; let row = sqlx::query(query) .bind(&execution.id) .bind(&execution.order_id) .bind(&execution.symbol) .bind(&execution.quantity) .bind(&execution.price) .bind(&execution.side) .bind(&execution.fees) .bind(&execution.fee_currency) .bind(&execution.executed_at) .bind(&execution.broker_execution_id) .bind(&execution.counterparty) .bind(&execution.venue) .bind(&execution.gross_value) .bind(&execution.net_value) .fetch_one(&mut *tx) .await?; let inserted_execution = Execution { id: row.get("id"), order_id: row.get("order_id"), symbol: row.get("symbol"), quantity: row.get("quantity"), price: row.get("price"), side: row.get("side"), fees: row.get("fees"), fee_currency: row.get("fee_currency"), executed_at: row.get("executed_at"), broker_execution_id: row.get("broker_execution_id"), counterparty: row.get("counterparty"), venue: row.get("venue"), gross_value: row.get("gross_value"), net_value: row.get("net_value"), }; inserted_executions.push(inserted_execution); } tx.commit().await?; Ok(inserted_executions) } async fn get_execution_stats(&self, symbol: Option<&str>, time_range: Option<(DateTime, DateTime)>) -> Result { let mut conditions = Vec::new(); let mut params: Vec + Send>> = Vec::new(); let mut param_count = 0; if let Some(symbol) = symbol { param_count += 1; conditions.push(format!("symbol = ${}", param_count)); params.push(Box::new(symbol.to_string())); } if let Some((start, end)) = time_range { param_count += 1; conditions.push(format!("executed_at >= ${}", param_count)); params.push(Box::new(start)); param_count += 1; conditions.push(format!("executed_at <= ${}", param_count)); params.push(Box::new(end)); } let where_clause = if conditions.is_empty() { String::new() } else { format!("WHERE {}", conditions.join(" AND ")) }; let query = format!( r#" SELECT COUNT(*) as total_executions, COUNT(CASE WHEN side = 'buy' THEN 1 END) as buy_executions, COUNT(CASE WHEN side = 'sell' THEN 1 END) as sell_executions, COALESCE(SUM(quantity), 0) as total_volume, COALESCE(SUM(gross_value), 0) as total_value, COALESCE(SUM(fees), 0) as total_fees, COALESCE(AVG(quantity), 0) as avg_execution_size, COALESCE(AVG(price), 0) as avg_price, MAX(quantity) as largest_execution, MIN(quantity) as smallest_execution, COUNT(DISTINCT symbol) as unique_symbols, COUNT(DISTINCT venue) as unique_venues FROM executions {} "#, where_clause ); let row = sqlx::query(&query) .fetch_one(&self.pool) .await?; Ok(ExecutionStats { total_executions: row.get("total_executions"), buy_executions: row.get("buy_executions"), sell_executions: row.get("sell_executions"), total_volume: row.get("total_volume"), total_value: row.get("total_value"), total_fees: row.get("total_fees"), avg_execution_size: row.get("avg_execution_size"), avg_price: row.get("avg_price"), largest_execution: row.get("largest_execution"), smallest_execution: row.get("smallest_execution"), unique_symbols: row.get("unique_symbols"), unique_venues: row.get("unique_venues"), }) } async fn get_trading_performance(&self, symbol: Option<&str>, time_range: Option<(DateTime, DateTime)>) -> Result { // This is a simplified implementation // In practice, you'd need more complex logic to calculate trading performance metrics let stats = self.get_execution_stats(symbol, time_range).await?; // Simplified calculations (real implementation would be more complex) let win_rate = if stats.total_executions > 0 { Decimal::from(stats.buy_executions) / Decimal::from(stats.total_executions) * Decimal::from(100) } else { Decimal::ZERO }; Ok(TradingPerformance { total_trades: stats.total_executions, winning_trades: stats.buy_executions, // Simplified losing_trades: stats.sell_executions, // Simplified win_rate, avg_win: Decimal::ZERO, // Would need P&L calculation avg_loss: Decimal::ZERO, // Would need P&L calculation profit_factor: Decimal::ONE, // Would need P&L calculation total_pnl: Decimal::ZERO, // Would need P&L calculation gross_profit: Decimal::ZERO, // Would need P&L calculation gross_loss: Decimal::ZERO, // Would need P&L calculation largest_win: Decimal::ZERO, // Would need P&L calculation largest_loss: Decimal::ZERO, // Would need P&L calculation avg_trade_duration: None, // Would need order matching }) } async fn find_large_executions(&self, value_threshold: Decimal) -> Result> { let executions = sqlx::query_as::<_, Execution>( r#" SELECT id, order_id, symbol, quantity, price, side, fees, fee_currency, executed_at, broker_execution_id, counterparty, venue, gross_value, net_value FROM executions WHERE gross_value >= $1 ORDER BY gross_value DESC "# ) .bind(value_threshold) .fetch_all(&self.pool) .await?; Ok(executions) } async fn calculate_vwap(&self, symbol: &str, time_range: Option<(DateTime, DateTime)>) -> Result { let (query, start, end) = if let Some((start, end)) = time_range { ( r#" SELECT COALESCE(SUM(quantity * price) / NULLIF(SUM(quantity), 0), 0) as vwap FROM executions WHERE symbol = $1 AND executed_at >= $2 AND executed_at <= $3 "#, Some(start), Some(end), ) } else { ( r#" SELECT COALESCE(SUM(quantity * price) / NULLIF(SUM(quantity), 0), 0) as vwap FROM executions WHERE symbol = $1 "#, None, None, ) }; let vwap: Decimal = if let (Some(start), Some(end)) = (start, end) { sqlx::query_scalar(query) .bind(symbol) .bind(start) .bind(end) .fetch_one(&self.pool) .await? } else { sqlx::query_scalar(query) .bind(symbol) .fetch_one(&self.pool) .await? }; Ok(vwap) } async fn get_hourly_execution_summary(&self, date: chrono::NaiveDate) -> Result> { let start_date = date.and_hms_opt(0, 0, 0).unwrap().and_utc(); let end_date = date.succ_opt().unwrap().and_hms_opt(0, 0, 0).unwrap().and_utc(); let rows = sqlx::query( r#" SELECT EXTRACT(HOUR FROM executed_at) as hour, COUNT(*) as execution_count, COALESCE(SUM(quantity), 0) as total_volume, COALESCE(SUM(gross_value), 0) as total_value, COALESCE(AVG(price), 0) as avg_price, COUNT(DISTINCT symbol) as unique_symbols FROM executions WHERE executed_at >= $1 AND executed_at < $2 GROUP BY EXTRACT(HOUR FROM executed_at) ORDER BY hour "# ) .bind(start_date) .bind(end_date) .fetch_all(&self.pool) .await?; let mut summaries = Vec::new(); for row in rows { summaries.push(HourlyExecutionSummary { hour: row.get::("hour"), execution_count: row.get("execution_count"), total_volume: row.get("total_volume"), total_value: row.get("total_value"), avg_price: row.get("avg_price"), unique_symbols: row.get("unique_symbols"), }); } Ok(summaries) } async fn find_high_slippage_executions(&self, slippage_threshold: Decimal) -> Result> { // This is a placeholder implementation // Real slippage calculation would need reference prices (expected vs actual) let executions = self.find_by_filter(&ExecutionFilter::new()).await?; let mut high_slippage_executions = Vec::new(); for execution in executions { // Simplified slippage calculation (would need actual reference prices) let expected_price = execution.price; // Placeholder let slippage = execution.price - expected_price; let slippage_bps = if !expected_price.is_zero() { slippage / expected_price * Decimal::from(10000) } else { Decimal::ZERO }; if slippage_bps.abs() >= slippage_threshold { high_slippage_executions.push(ExecutionWithSlippage { execution, expected_price, slippage, slippage_bps, }); } } Ok(high_slippage_executions) } } #[cfg(test)] mod tests { use super::*; use crate::models::OrderSide; use rust_decimal_macros::dec; use uuid::Uuid; #[test] fn test_execution_filter_builder() { let order_id = Uuid::new_v4(); let filter = ExecutionFilter::new() .symbol("EURUSD") .order_id(order_id) .side(OrderSide::Buy) .limit(100); assert_eq!(filter.symbol.as_ref().unwrap(), "EURUSD"); assert_eq!(filter.order_id.unwrap(), order_id); assert_eq!(filter.side.unwrap(), OrderSide::Buy); assert_eq!(filter.limit.unwrap(), 100); } #[test] fn test_execution_stats() { let stats = ExecutionStats { total_executions: 1000, buy_executions: 600, sell_executions: 400, total_volume: dec!(50000000), total_value: dec!(55000000), total_fees: dec!(5500), avg_execution_size: dec!(50000), avg_price: dec!(1.1000), largest_execution: Some(dec!(500000)), smallest_execution: Some(dec!(1000)), unique_symbols: 15, unique_venues: 3, }; assert_eq!(stats.total_executions, 1000); assert_eq!(stats.buy_executions, 600); assert_eq!(stats.total_fees, dec!(5500)); } #[test] fn test_hourly_execution_summary() { let summary = HourlyExecutionSummary { hour: 14, execution_count: 150, total_volume: dec!(5000000), total_value: dec!(5500000), avg_price: dec!(1.1000), unique_symbols: 8, }; assert_eq!(summary.hour, 14); assert_eq!(summary.execution_count, 150); assert_eq!(summary.unique_symbols, 8); } #[test] fn test_execution_with_slippage() { let execution = Execution::new( Uuid::new_v4(), "EURUSD".to_string(), dec!(100000), dec!(1.1005), OrderSide::Buy, dec!(5.0), ); let execution_with_slippage = ExecutionWithSlippage { execution: execution.clone(), expected_price: dec!(1.1000), slippage: dec!(0.0005), slippage_bps: dec!(4.545), // Approximately 4.5 bps }; assert_eq!(execution_with_slippage.expected_price, dec!(1.1000)); assert_eq!(execution_with_slippage.slippage, dec!(0.0005)); assert!(execution_with_slippage.slippage_bps > dec!(4.0)); assert!(execution_with_slippage.slippage_bps < dec!(5.0)); } // Note: Database integration tests would require a test database // and are typically run separately from unit tests }