//! Order repository implementation //! //! This module provides repository pattern implementation for trading orders //! with `PostgreSQL` backing. It includes comprehensive CRUD operations, //! filtering, and batch operations for high-frequency trading. use async_trait::async_trait; use chrono::{DateTime, Utc}; use common::types::Quantity; use rust_decimal::Decimal; // Removed direct rust_decimal import - using common::Decimal via common crate use sqlx::{Pool, Postgres, Row}; use uuid::Uuid; use crate::{Repository, Result}; // Import trading types directly from common use common::types::{Order, OrderSide, OrderStatus, OrderType}; // Import Volume from common use common::types::Volume; /// Filter criteria for order queries #[derive(Debug, Default, Clone)] pub struct OrderFilter { /// Filter by symbol pub symbol: Option, /// Filter by order status pub status: Option, /// Filter by order side pub side: Option, /// Filter by order type pub order_type: Option, /// Filter by client order ID pub client_order_id: Option, /// Filter by broker order ID pub broker_order_id: Option, /// Filter orders created after this timestamp pub created_after: Option>, /// Filter orders created before this timestamp pub created_before: Option>, /// Limit number of results pub limit: Option, /// Offset for pagination pub offset: Option, } impl OrderFilter { /// Create a new empty filter #[must_use] pub fn new() -> Self { Self::default() } /// Filter by symbol #[must_use] pub fn symbol>(mut self, symbol: S) -> Self { self.symbol = Some(symbol.into()); self } /// Filter by status #[must_use] pub fn status(mut self, status: OrderStatus) -> Self { self.status = Some(status); self } /// Filter by side #[must_use] pub fn side(mut self, side: OrderSide) -> Self { self.side = Some(side); self } /// Filter by order type #[must_use] pub fn order_type(mut self, order_type: OrderType) -> Self { self.order_type = Some(order_type); self } /// Filter by client order ID #[must_use] pub fn client_order_id>(mut self, client_order_id: S) -> Self { self.client_order_id = Some(client_order_id.into()); self } /// Filter by date range #[must_use] pub fn created_between(mut self, start: DateTime, end: DateTime) -> Self { self.created_after = Some(start); self.created_before = Some(end); self } /// Limit results #[must_use] pub fn limit(mut self, limit: i64) -> Self { self.limit = Some(limit); self } /// Set offset for pagination #[must_use] pub fn offset(mut self, offset: i64) -> Self { self.offset = Some(offset); self } } /// Order repository trait defining available operations #[async_trait] pub trait OrderRepository: Repository { /// Find orders matching the filter criteria async fn find_by_filter(&self, filter: &OrderFilter) -> Result>; /// Find orders by symbol async fn find_by_symbol(&self, symbol: &str) -> Result>; /// Find orders by status async fn find_by_status(&self, status: OrderStatus) -> Result>; /// Find active orders (submitted or partially filled) async fn find_active_orders(&self) -> Result>; /// Update order status async fn update_status(&self, order_id: &Uuid, status: OrderStatus) -> Result<()>; /// Update order fill information async fn update_fill( &self, order_id: &Uuid, filled_quantity: Decimal, avg_fill_price: Decimal, ) -> Result<()>; /// Batch insert orders for high-frequency operations async fn batch_insert(&self, orders: &[Order]) -> Result>; /// Cancel order by ID async fn cancel_order(&self, order_id: &Uuid) -> Result; /// Find orders ready for expiration async fn find_expired_orders(&self) -> Result>; /// Get order statistics async fn get_order_stats(&self, symbol: Option<&str>) -> Result; } /// Order statistics for performance analysis #[derive(Debug, Clone)] pub struct OrderStats { /// Total number of orders in the dataset pub total_orders: i64, /// Number of completely filled orders pub filled_orders: i64, /// Number of cancelled orders pub cancelled_orders: i64, /// Number of orders still pending or partially filled pub pending_orders: i64, /// Total trading volume across all orders pub total_volume: Volume, /// Average fill rate as a percentage pub avg_fill_rate: Decimal, } /// `PostgreSQL` implementation of `OrderRepository` pub struct PostgresOrderRepository { pool: Pool, } impl PostgresOrderRepository { /// Create a new `PostgreSQL` order repository #[must_use] pub fn new(pool: Pool) -> Self { Self { pool } } /// Build WHERE clause and parameters from filter fn build_filter_query( filter: &OrderFilter, ) -> (String, Vec + Send>>) { let mut conditions = Vec::new(); let mut params: Vec + Send>> = Vec::new(); let mut param_count = 0; if let Some(symbol) = &filter.symbol { param_count += 1; conditions.push(format!("symbol = ${param_count}")); params.push(Box::new(symbol.clone())); } if let Some(status) = &filter.status { param_count += 1; conditions.push(format!("status = ${param_count}")); params.push(Box::new(*status)); } if let Some(side) = &filter.side { param_count += 1; conditions.push(format!("side = ${param_count}")); params.push(Box::new(*side)); } if let Some(order_type) = &filter.order_type { param_count += 1; conditions.push(format!("order_type = ${param_count}")); params.push(Box::new(*order_type)); } if let Some(client_order_id) = &filter.client_order_id { param_count += 1; conditions.push(format!("client_order_id = ${param_count}")); params.push(Box::new(client_order_id.clone())); } if let Some(broker_order_id) = &filter.broker_order_id { param_count += 1; conditions.push(format!("broker_order_id = ${param_count}")); params.push(Box::new(broker_order_id.clone())); } if let Some(created_after) = &filter.created_after { param_count += 1; conditions.push(format!("created_at >= ${param_count}")); params.push(Box::new(*created_after)); } if let Some(created_before) = &filter.created_before { param_count += 1; conditions.push(format!("created_at <= ${param_count}")); params.push(Box::new(*created_before)); } let where_clause = if conditions.is_empty() { String::new() } else { format!("WHERE {}", conditions.join(" AND ")) }; // Add ORDER BY, LIMIT, OFFSET let mut query_parts = vec![where_clause]; query_parts.push("ORDER BY created_at DESC".to_owned()); if let Some(limit) = filter.limit { param_count += 1; query_parts.push(format!("LIMIT ${param_count}")); params.push(Box::new(limit)); } if let Some(offset) = filter.offset { param_count += 1; query_parts.push(format!("OFFSET ${param_count}")); params.push(Box::new(offset)); } (query_parts.join(" "), params) } } #[async_trait] impl Repository for PostgresOrderRepository { async fn find_by_id(&self, id: &Uuid) -> Result> { let query = r" SELECT id, client_order_id, broker_order_id, account_id, symbol, side, order_type, status, time_in_force, quantity, price, stop_price, filled_quantity, remaining_quantity, average_price, avg_fill_price, parent_id, execution_algorithm, execution_params, stop_loss, take_profit, created_at, updated_at, expires_at, metadata FROM orders WHERE id = $1 "; let row = sqlx::query(query) .bind(id) .fetch_optional(&self.pool) .await?; match row { Some(row) => { let order = Order { id: row.get("id"), client_order_id: row.get("client_order_id"), broker_order_id: row.get("broker_order_id"), account_id: row.get("account_id"), symbol: row.get::("symbol").into(), side: row.get("side"), order_type: row.get("order_type"), status: row.get("status"), time_in_force: row .get::, _>("time_in_force") .unwrap_or(common::types::TimeInForce::GoodTillCancel), quantity: row.get("quantity"), price: row.get("price"), stop_price: row.get("stop_price"), filled_quantity: row.get("filled_quantity"), remaining_quantity: row.get("remaining_quantity"), average_price: row.get("average_price"), avg_fill_price: row.get("avg_fill_price"), parent_id: row.get("parent_id"), execution_algorithm: row.get("execution_algorithm"), execution_params: row .get::, _>("execution_params") .unwrap_or(serde_json::Value::Null), stop_loss: row.get("stop_loss"), take_profit: row.get("take_profit"), created_at: row.get("created_at"), average_fill_price: None, exchange_order_id: None, updated_at: row.get("updated_at"), expires_at: row.get("expires_at"), metadata: row .get::, _>("metadata") .unwrap_or(serde_json::Value::Null), }; Ok(Some(order)) }, None => Ok(None), } } async fn save(&self, order: &Order) -> Result { let query = r" INSERT INTO orders (id, client_order_id, broker_order_id, account_id, symbol, side, order_type, status, time_in_force, quantity, price, stop_price, filled_quantity, remaining_quantity, average_price, avg_fill_price, parent_id, execution_algorithm, execution_params, stop_loss, take_profit, created_at, updated_at, expires_at, metadata) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, filled_quantity = EXCLUDED.filled_quantity, remaining_quantity = EXCLUDED.remaining_quantity, average_price = EXCLUDED.average_price, avg_fill_price = EXCLUDED.avg_fill_price, updated_at = EXCLUDED.updated_at, broker_order_id = EXCLUDED.broker_order_id, stop_loss = EXCLUDED.stop_loss, take_profit = EXCLUDED.take_profit, metadata = EXCLUDED.metadata RETURNING * "; let row = sqlx::query(query) .bind(order.id) .bind(&order.client_order_id) .bind(&order.broker_order_id) .bind(&order.account_id) .bind(order.symbol.to_string()) .bind(order.side) .bind(order.order_type) .bind(order.status) .bind(order.time_in_force) .bind(order.quantity) .bind(order.price) .bind(order.stop_price) .bind(order.filled_quantity) .bind(order.remaining_quantity) .bind(order.average_price) .bind(order.avg_fill_price) .bind(&order.parent_id) .bind(&order.execution_algorithm) .bind(&order.execution_params) .bind(order.stop_loss) .bind(order.take_profit) .bind(order.created_at) .bind(order.updated_at) .bind(order.expires_at) .bind(&order.metadata) .fetch_one(&self.pool) .await?; Ok(Order { id: row.get("id"), client_order_id: row.get("client_order_id"), broker_order_id: row.get("broker_order_id"), account_id: row.get("account_id"), symbol: row.get("symbol"), side: row.get("side"), order_type: row.get("order_type"), status: row.get("status"), time_in_force: row .get::, _>("time_in_force") .unwrap_or(common::types::TimeInForce::GoodTillCancel), quantity: row.get("quantity"), price: row.get("price"), stop_price: row.get("stop_price"), filled_quantity: row.get("filled_quantity"), remaining_quantity: row.get("remaining_quantity"), average_price: row.get("average_price"), avg_fill_price: row.get("avg_fill_price"), parent_id: row.get("parent_id"), execution_algorithm: row.get("execution_algorithm"), execution_params: row .get::, _>("execution_params") .unwrap_or(serde_json::Value::Null), stop_loss: row.get("stop_loss"), take_profit: row.get("take_profit"), created_at: row.get("created_at"), average_fill_price: None, exchange_order_id: None, updated_at: row.get("updated_at"), expires_at: row.get("expires_at"), metadata: row .get::, _>("metadata") .unwrap_or(serde_json::Value::Null), }) } async fn delete(&self, id: &Uuid) -> Result { let result = sqlx::query("DELETE FROM orders 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 orders WHERE id = $1") .bind(id) .fetch_one(&self.pool) .await?; Ok(count > 0) } } #[async_trait] impl OrderRepository for PostgresOrderRepository { async fn find_by_filter(&self, filter: &OrderFilter) -> Result> { let base_query = r" SELECT id, symbol, side, quantity, price, order_type, status, filled_quantity, remaining_quantity, avg_fill_price, created_at, updated_at, expires_at, client_order_id, broker_order_id, stop_loss, take_profit FROM orders "; let (filter_clause, _params) = Self::build_filter_query(filter); let full_query = format!("{base_query} {filter_clause}"); // Note: Due to sqlx limitations with dynamic parameters, we'll build the query manually // In a real implementation, you might use a query builder or handle this more elegantly let query = sqlx::query_as::<_, Order>(&full_query); // This is a simplified version - in practice you'd need to bind parameters dynamically let rows = query.fetch_all(&self.pool).await?; Ok(rows) } async fn find_by_symbol(&self, symbol: &str) -> Result> { let orders = sqlx::query_as::<_, Order>( r" SELECT id, symbol, side, quantity, price, order_type, status, filled_quantity, remaining_quantity, avg_fill_price, created_at, updated_at, expires_at, client_order_id, broker_order_id, stop_loss, take_profit FROM orders WHERE symbol = $1 ORDER BY created_at DESC ", ) .bind(symbol) .fetch_all(&self.pool) .await?; Ok(orders) } async fn find_by_status(&self, status: OrderStatus) -> Result> { let orders = sqlx::query_as::<_, Order>( r" SELECT id, symbol, side, quantity, price, order_type, status, filled_quantity, remaining_quantity, avg_fill_price, created_at, updated_at, expires_at, client_order_id, broker_order_id, stop_loss, take_profit FROM orders WHERE status = $1 ORDER BY created_at DESC ", ) .bind(status) .fetch_all(&self.pool) .await?; Ok(orders) } async fn find_active_orders(&self) -> Result> { let orders = sqlx::query_as::<_, Order>( r" SELECT id, symbol, side, quantity, price, order_type, status, filled_quantity, remaining_quantity, avg_fill_price, created_at, updated_at, expires_at, client_order_id, broker_order_id, stop_loss, take_profit FROM orders WHERE status IN ('submitted', 'partiallyfilled') ORDER BY created_at ASC ", ) .fetch_all(&self.pool) .await?; Ok(orders) } async fn update_status(&self, order_id: &Uuid, status: OrderStatus) -> Result<()> { sqlx::query("UPDATE orders SET status = $1, updated_at = $2 WHERE id = $3") .bind(status) .bind(Utc::now()) .bind(order_id) .execute(&self.pool) .await?; Ok(()) } async fn update_fill( &self, order_id: &Uuid, filled_quantity: Decimal, avg_fill_price: Decimal, ) -> Result<()> { sqlx::query( r" UPDATE orders SET filled_quantity = $1, avg_fill_price = $2, remaining_quantity = quantity - $1, updated_at = $3, status = CASE WHEN $1 >= quantity THEN 'filled'::order_status WHEN $1 > 0 THEN 'partiallyfilled'::order_status ELSE status END WHERE id = $4 ", ) .bind(filled_quantity) .bind(avg_fill_price) .bind(Utc::now()) .bind(order_id) .execute(&self.pool) .await?; Ok(()) } async fn batch_insert(&self, orders: &[Order]) -> Result> { if orders.is_empty() { return Ok(Vec::new()); } let mut tx = self.pool.begin().await?; let mut inserted_orders = Vec::new(); for order in orders { let query = r" INSERT INTO orders (id, symbol, side, quantity, price, order_type, status, filled_quantity, remaining_quantity, avg_fill_price, created_at, updated_at, expires_at, client_order_id, broker_order_id, stop_loss, take_profit) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) RETURNING * "; let row = sqlx::query(query) .bind(order.id) .bind(&order.symbol) .bind(order.side) .bind(order.quantity) .bind(order.price) .bind(order.order_type) .bind(order.status) .bind(order.filled_quantity) .bind(order.remaining_quantity) .bind(order.avg_fill_price) .bind(order.created_at) .bind(order.updated_at) .bind(order.expires_at) .bind(&order.client_order_id) .bind(&order.broker_order_id) .bind(order.stop_loss) .bind(order.take_profit) .fetch_one(&mut *tx) .await?; let inserted_order = Order { id: row.get("id"), client_order_id: row.get("client_order_id"), broker_order_id: row.get("broker_order_id"), account_id: row.get("account_id"), symbol: row.get("symbol"), side: row.get("side"), order_type: row.get("order_type"), status: row.get("status"), time_in_force: row .get::, _>("time_in_force") .unwrap_or(common::types::TimeInForce::GoodTillCancel), quantity: row.get("quantity"), price: row.get("price"), stop_price: row.get("stop_price"), filled_quantity: row.get("filled_quantity"), remaining_quantity: row.get("remaining_quantity"), average_price: row.get("average_price"), avg_fill_price: row.get("avg_fill_price"), parent_id: row.get("parent_id"), execution_algorithm: row.get("execution_algorithm"), execution_params: row .get::, _>("execution_params") .unwrap_or(serde_json::Value::Null), stop_loss: row.get("stop_loss"), take_profit: row.get("take_profit"), created_at: row.get("created_at"), average_fill_price: None, exchange_order_id: None, updated_at: row.get("updated_at"), expires_at: row.get("expires_at"), metadata: row .get::, _>("metadata") .unwrap_or(serde_json::Value::Null), }; inserted_orders.push(inserted_order); } tx.commit().await?; Ok(inserted_orders) } async fn cancel_order(&self, order_id: &Uuid) -> Result { let result = sqlx::query( r" UPDATE orders SET status = 'cancelled', updated_at = $1 WHERE id = $2 AND status IN ('pending', 'submitted', 'partiallyfilled') ", ) .bind(Utc::now()) .bind(order_id) .execute(&self.pool) .await?; Ok(result.rows_affected() > 0) } async fn find_expired_orders(&self) -> Result> { let orders = sqlx::query_as::<_, Order>( r" SELECT id, symbol, side, quantity, price, order_type, status, filled_quantity, remaining_quantity, avg_fill_price, created_at, updated_at, expires_at, client_order_id, broker_order_id, stop_loss, take_profit FROM orders WHERE expires_at IS NOT NULL AND expires_at <= $1 AND status IN ('pending', 'submitted', 'partiallyfilled') ORDER BY expires_at ASC ", ) .bind(Utc::now()) .fetch_all(&self.pool) .await?; Ok(orders) } async fn get_order_stats(&self, symbol: Option<&str>) -> Result { let (base_query, symbol_filter) = if let Some(symbol) = symbol { ( r" SELECT COUNT(*) as total_orders, COUNT(CASE WHEN status = 'filled' THEN 1 END) as filled_orders, COUNT(CASE WHEN status = 'cancelled' THEN 1 END) as cancelled_orders, COUNT(CASE WHEN status IN ('pending', 'submitted', 'partiallyfilled') THEN 1 END) as pending_orders, COALESCE(SUM(CASE WHEN status = 'filled' THEN filled_quantity * COALESCE(avg_fill_price, price, 0) END), 0) as total_volume FROM orders WHERE symbol = $1 ", Some(symbol), ) } else { ( r" SELECT COUNT(*) as total_orders, COUNT(CASE WHEN status = 'filled' THEN 1 END) as filled_orders, COUNT(CASE WHEN status = 'cancelled' THEN 1 END) as cancelled_orders, COUNT(CASE WHEN status IN ('pending', 'submitted', 'partiallyfilled') THEN 1 END) as pending_orders, COALESCE(SUM(CASE WHEN status = 'filled' THEN filled_quantity * COALESCE(avg_fill_price, price, 0) END), 0) as total_volume FROM orders ", None, ) }; let row = if let Some(symbol) = symbol_filter { sqlx::query(base_query) .bind(symbol) .fetch_one(&self.pool) .await? } else { sqlx::query(base_query).fetch_one(&self.pool).await? }; let total_orders: i64 = row.get("total_orders"); let filled_orders: i64 = row.get("filled_orders"); let cancelled_orders: i64 = row.get("cancelled_orders"); let pending_orders: i64 = row.get("pending_orders"); let total_volume_decimal: Decimal = row.get("total_volume"); let total_volume = Quantity::from_f64(total_volume_decimal.try_into().unwrap_or(0.0)) .unwrap_or(Quantity::ZERO); let avg_fill_rate = if total_orders > 0 { Decimal::from(filled_orders) / Decimal::from(total_orders) * Decimal::from(100) } else { Decimal::ZERO }; Ok(OrderStats { total_orders, filled_orders, cancelled_orders, pending_orders, total_volume, avg_fill_rate, }) } } #[cfg(test)] mod tests { use super::*; use common::OrderStatus; use rust_decimal_macros::dec; #[test] fn test_order_filter_builder() { let filter = OrderFilter::new() .symbol("EURUSD") .status(OrderStatus::Filled) .limit(50); assert_eq!(filter.symbol.as_ref().unwrap(), "EURUSD"); assert_eq!(filter.status.unwrap(), OrderStatus::Filled); assert_eq!(filter.limit.unwrap(), 50); } #[test] fn test_order_stats() { use common::Quantity; let stats = OrderStats { total_orders: 100, filled_orders: 80, cancelled_orders: 15, pending_orders: 5, total_volume: Quantity::from_decimal(dec!(1000000)).unwrap(), avg_fill_rate: dec!(80.0), }; assert_eq!(stats.total_orders, 100); assert_eq!(stats.avg_fill_rate, dec!(80.0)); } // Note: Database integration tests would require a test database // and are typically run separately from unit tests }