foxhunt/crates/market-data/src/prices.rs

//! # Price Data Module
//!
//! This module provides repository abstractions and implementations for storing,
//! retrieving, and managing price data including tick data, OHLCV candles,
//! and historical price records for trading instruments.
//!
//! ## Features
//!
//! - Storage and retrieval of tick-level price data
//! - OHLCV candle data management
//! - Historical price queries with time range filtering
//! - Multi-symbol price retrieval
//! - Data cleanup and maintenance operations
//! - Batch operations for efficient data processing
//!
//! ## Usage
//!
//! ```rust
//! use market_data::prices::{PriceRepository, PostgresPriceRepository};
//! use market_data::models::{PriceRecord, TimePeriod};
//! use sqlx::PgPool;
//!
//! # async fn example(pool: PgPool) -> Result<(), Box<dyn std::error::Error>> {
//! let repo = PostgresPriceRepository::new(pool);
//!
//! // Get latest price for a symbol
//! let price = repo.get_latest_price("AAPL").await?;
//!
//! // Get candle data
//! let start = Utc::now() - Duration::days(30);
//! let end = Utc::now();
//! let candles = repo.get_candles("AAPL", TimePeriod::Daily, start, end).await?;
//! # Ok(())
//! # }
//! ```

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use sqlx::{PgPool, Row};
use std::collections::HashMap;

use crate::{
    error::{MarketDataError, MarketDataResult},
    models::{Candle, PriceRecord, TimePeriod},
};

/// Repository trait for price data operations
///
/// This trait defines the interface for storing, retrieving, and managing
/// price data including tick-level prices and aggregated candles.
///
/// Implementations should provide efficient access patterns optimized
/// for both real-time and historical data queries.
///
/// The trait supports:
/// - Individual and batch price storage
///
/// - Historical price data retrieval
/// - Multi-symbol price queries
///
/// - OHLCV candle data management
/// - Data maintenance and cleanup
#[async_trait]
pub trait PriceRepository {
    /// Store a single price record
    ///
    /// Stores a price record in the repository. If a record with the same ID
    /// already exists, it will be updated with the new price information.
    ///
    /// # Arguments
    ///
    /// * `price` - The price record to store
    ///
    /// # Returns
    ///
    /// `Ok(())` on success, or a `MarketDataError` if the operation fails
    ///
    /// # Errors
    ///
    /// - `MarketDataError::InvalidSymbol` if the symbol is invalid
    /// - `MarketDataError::Database` if the database operation fails
    async fn store_price(&self, price: &PriceRecord) -> MarketDataResult<()>;

    /// Store multiple price records in a batch
    ///
    /// Efficiently stores multiple price records in a single transaction.
    ///
    /// This is optimized for bulk data loading and reduces database overhead.
    ///
    /// # Arguments
    ///
    /// * `prices` - Slice of price records to store
    ///
    /// # Returns
    ///
    /// `Ok(())` on success, or a `MarketDataError` if any operation fails
    ///
    /// # Errors
    ///
    /// - `MarketDataError::InvalidSymbol` if any symbol is invalid
    /// - `MarketDataError::Database` if the database transaction fails
    async fn store_prices(&self, prices: &[PriceRecord]) -> MarketDataResult<()>;

    /// Get the latest price for a symbol
    ///
    /// Retrieves the most recent price record for the specified symbol,
    /// ordered by timestamp.
    ///
    /// # Arguments
    ///
    /// * `symbol` - Trading symbol to query
    ///
    /// # Returns
    ///
    /// `Some(price)` if found, `None` if no data exists, or a `MarketDataError`
    ///
    /// # Errors
    ///
    /// - `MarketDataError::InvalidSymbol` if the symbol is invalid
    /// - `MarketDataError::Database` if the query fails
    async fn get_latest_price(&self, symbol: &str) -> MarketDataResult<Option<PriceRecord>>;

    /// Get price history for a symbol within a time range
    ///
    /// Retrieves historical price records within the specified time range,
    /// ordered chronologically. This is useful for backtesting and analysis.
    ///
    /// # Arguments
    ///
    /// * `symbol` - Trading symbol to query
    /// * `from` - Start of time range (inclusive)
    ///
    /// * `to` - End of time range (inclusive)
    ///
    /// # Returns
    ///
    /// Vector of price records ordered by timestamp, or a `MarketDataError`
    ///
    /// # Errors
    ///
    /// - `MarketDataError::InvalidSymbol` if the symbol is invalid
    /// - `MarketDataError::InvalidTimeRange` if from >= to
    ///
    /// - `MarketDataError::Database` if the query fails
    async fn get_price_history(
        &self,
        symbol: &str,
        from: DateTime<Utc>,
        to: DateTime<Utc>,
    ) -> MarketDataResult<Vec<PriceRecord>>;

    /// Get latest prices for multiple symbols
    ///
    /// Efficiently retrieves the latest price records for multiple symbols
    /// in a single query. Useful for portfolio analysis and screening.
    ///
    /// # Arguments
    ///
    /// * `symbols` - List of trading symbols to query
    ///
    /// # Returns
    ///
    /// HashMap mapping symbols to their latest price records, or a `MarketDataError`
    ///
    /// # Errors
    ///
    /// - `MarketDataError::InvalidSymbol` if any symbol is invalid
    /// - `MarketDataError::Database` if the query fails
    async fn get_latest_prices(
        &self,
        symbols: &[String],
    ) -> MarketDataResult<HashMap<String, PriceRecord>>;

    /// Store a candle (OHLCV) record
    ///
    /// Stores an OHLCV candle record for a specific time period.
    ///
    /// If a candle with the same symbol, period, and timestamp exists, it will be updated.
    ///
    /// # Arguments
    ///
    /// * `candle` - The candle record to store
    ///
    /// # Returns
    ///
    /// `Ok(())` on success, or a `MarketDataError` if the operation fails
    ///
    /// # Errors
    ///
    /// - `MarketDataError::InvalidSymbol` if the symbol is invalid
    /// - `MarketDataError::Database` if the database operation fails
    async fn store_candle(&self, candle: &Candle) -> MarketDataResult<()>;

    /// Get candle history for a symbol and period
    ///
    /// Retrieves historical OHLCV candle data for a specific symbol and time period
    /// within the specified time range. Useful for technical analysis and charting.
    ///
    /// # Arguments
    ///
    /// * `symbol` - Trading symbol to query
    /// * `period` - Time period for the candles (e.g., Daily, Hour)
    ///
    /// * `from` - Start of time range (inclusive)
    /// * `to` - End of time range (inclusive)
    ///
    /// # Returns
    ///
    /// Vector of candles ordered by timestamp, or a `MarketDataError`
    ///
    /// # Errors
    ///
    /// - `MarketDataError::InvalidSymbol` if the symbol is invalid
    /// - `MarketDataError::InvalidTimeRange` if from >= to
    ///
    /// - `MarketDataError::Database` if the query fails
    async fn get_candles(
        &self,
        symbol: &str,
        period: TimePeriod,
        from: DateTime<Utc>,
        to: DateTime<Utc>,
    ) -> MarketDataResult<Vec<Candle>>;

    /// Delete old price data before a given timestamp
    ///
    /// Removes historical price data older than the specified timestamp.
    ///
    /// This is useful for data retention management and storage optimization.
    ///
    /// # Arguments
    ///
    /// * `before` - Timestamp before which all data will be deleted
    ///
    /// # Returns
    ///
    /// Number of records deleted, or a `MarketDataError`
    ///
    /// # Errors
    ///
    /// - `MarketDataError::Database` if the deletion fails
    async fn cleanup_old_prices(&self, before: DateTime<Utc>) -> MarketDataResult<u64>;
}

/// PostgreSQL implementation of PriceRepository
///
/// Provides a production-ready implementation of the `PriceRepository` trait
/// using PostgreSQL as the backend storage. This implementation is optimized
/// for high-frequency price updates and efficient historical data retrieval.
///
/// ## Features
///
/// - Transactional batch operations for atomic updates
/// - Optimized time-series queries with proper indexing
///
/// - Input validation and error handling
/// - Conflict resolution with upsert semantics
///
/// - Support for both tick data and aggregated candles
pub struct PostgresPriceRepository {
    /// PostgreSQL connection pool for database operations
    pool: PgPool,
}

impl PostgresPriceRepository {
    /// Create a new PostgreSQL price repository
    ///
    /// # Arguments
    ///
    /// * `pool` - PostgreSQL connection pool
    ///
    /// # Returns
    ///
    /// A new `PostgresPriceRepository` instance
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    /// Validate that a symbol meets format requirements
    ///
    /// Ensures the symbol is non-empty and within length limits.
    ///
    /// # Arguments
    ///
    /// * `symbol` - Symbol to validate
    ///
    /// # Returns
    ///
    /// `Ok(())` if valid, `MarketDataError::InvalidSymbol` otherwise
    async fn validate_symbol(&self, symbol: &str) -> MarketDataResult<()> {
        if symbol.is_empty() || symbol.len() > 20 {
            return Err(MarketDataError::InvalidSymbol {
                symbol: symbol.to_string(),
            });
        }
        Ok(())
    }

    /// Validate that a time range is logically correct
    ///
    /// Ensures the start time is before the end time.
    ///
    /// # Arguments
    ///
    /// * `from` - Start time
    /// * `to` - End time
    ///
    /// # Returns
    ///
    /// `Ok(())` if valid, `MarketDataError::InvalidTimeRange` otherwise
    async fn validate_time_range(
        &self,
        from: DateTime<Utc>,
        to: DateTime<Utc>,
    ) -> MarketDataResult<()> {
        if from >= to {
            return Err(MarketDataError::InvalidTimeRange { from, to });
        }
        Ok(())
    }
}

#[async_trait]
impl PriceRepository for PostgresPriceRepository {
    async fn store_price(&self, price: &PriceRecord) -> MarketDataResult<()> {
        self.validate_symbol(&price.symbol).await?;

        sqlx::query(
            r#"
            INSERT INTO prices (id, symbol, timestamp, bid, ask, last, volume, open, high, low, close, created_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
            ON CONFLICT (id) DO UPDATE SET
                bid = EXCLUDED.bid,
                ask = EXCLUDED.ask,
                last = EXCLUDED.last,
                volume = EXCLUDED.volume,
                open = EXCLUDED.open,
                high = EXCLUDED.high,
                low = EXCLUDED.low,
                close = EXCLUDED.close
            "#
        )
        .bind(price.id)
        .bind(&price.symbol)
        .bind(price.timestamp)
        .bind(price.bid)
        .bind(price.ask)
        .bind(price.last)
        .bind(price.volume)
        .bind(price.open)
        .bind(price.high)
        .bind(price.low)
        .bind(price.close)
        .bind(price.created_at)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    async fn store_prices(&self, prices: &[PriceRecord]) -> MarketDataResult<()> {
        if prices.is_empty() {
            return Ok(());
        }

        // Validate all symbols first
        for price in prices {
            self.validate_symbol(&price.symbol).await?;
        }

        let mut tx = self.pool.begin().await?;

        for price in prices {
            sqlx::query(
                r#"
                INSERT INTO prices (id, symbol, timestamp, bid, ask, last, volume, open, high, low, close, created_at)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
                ON CONFLICT (id) DO UPDATE SET
                    bid = EXCLUDED.bid,
                    ask = EXCLUDED.ask,
                    last = EXCLUDED.last,
                    volume = EXCLUDED.volume,
                    open = EXCLUDED.open,
                    high = EXCLUDED.high,
                    low = EXCLUDED.low,
                    close = EXCLUDED.close
                "#
            )
            .bind(price.id)
            .bind(&price.symbol)
            .bind(price.timestamp)
            .bind(price.bid)
            .bind(price.ask)
            .bind(price.last)
            .bind(price.volume)
            .bind(price.open)
            .bind(price.high)
            .bind(price.low)
            .bind(price.close)
            .bind(price.created_at)
            .execute(&mut *tx)
            .await?;
        }

        tx.commit().await?;
        Ok(())
    }

    async fn get_latest_price(&self, symbol: &str) -> MarketDataResult<Option<PriceRecord>> {
        self.validate_symbol(symbol).await?;

        let row = sqlx::query(
            r#"
            SELECT id, symbol, timestamp, bid, ask, last, volume, open, high, low, close, created_at
            FROM prices
            WHERE symbol = $1
            ORDER BY timestamp DESC
            LIMIT 1
            "#,
        )
        .bind(symbol)
        .fetch_optional(&self.pool)
        .await?;

        if let Some(row) = row {
            Ok(Some(PriceRecord {
                id: row.get("id"),
                symbol: row.get("symbol"),
                timestamp: row.get("timestamp"),
                bid: row.get("bid"),
                ask: row.get("ask"),
                last: row.get("last"),
                volume: row.get("volume"),
                open: row.get("open"),
                high: row.get("high"),
                low: row.get("low"),
                close: row.get("close"),
                created_at: row.get("created_at"),
            }))
        } else {
            Ok(None)
        }
    }

    async fn get_price_history(
        &self,
        symbol: &str,
        from: DateTime<Utc>,
        to: DateTime<Utc>,
    ) -> MarketDataResult<Vec<PriceRecord>> {
        self.validate_symbol(symbol).await?;
        self.validate_time_range(from, to).await?;

        let rows = sqlx::query(
            r#"
            SELECT id, symbol, timestamp, bid, ask, last, volume, open, high, low, close, created_at
            FROM prices
            WHERE symbol = $1 AND timestamp >= $2 AND timestamp <= $3
            ORDER BY timestamp ASC
            "#,
        )
        .bind(symbol)
        .bind(from)
        .bind(to)
        .fetch_all(&self.pool)
        .await?;

        let prices = rows
            .into_iter()
            .map(|row| PriceRecord {
                id: row.get("id"),
                symbol: row.get("symbol"),
                timestamp: row.get("timestamp"),
                bid: row.get("bid"),
                ask: row.get("ask"),
                last: row.get("last"),
                volume: row.get("volume"),
                open: row.get("open"),
                high: row.get("high"),
                low: row.get("low"),
                close: row.get("close"),
                created_at: row.get("created_at"),
            })
            .collect();

        Ok(prices)
    }

    async fn get_latest_prices(
        &self,
        symbols: &[String],
    ) -> MarketDataResult<HashMap<String, PriceRecord>> {
        if symbols.is_empty() {
            return Ok(HashMap::new());
        }

        // Validate all symbols
        for symbol in symbols {
            self.validate_symbol(symbol).await?;
        }

        let rows = sqlx::query(
            r#"
            SELECT DISTINCT ON (symbol) id, symbol, timestamp, bid, ask, last, volume, open, high, low, close, created_at
            FROM prices
            WHERE symbol = ANY($1)
            ORDER BY symbol, timestamp DESC
            "#
        )
        .bind(symbols)
        .fetch_all(&self.pool)
        .await?;

        let mut prices = HashMap::new();
        for row in rows {
            let price = PriceRecord {
                id: row.get("id"),
                symbol: row.get::<String, _>("symbol"),
                timestamp: row.get("timestamp"),
                bid: row.get("bid"),
                ask: row.get("ask"),
                last: row.get("last"),
                volume: row.get("volume"),
                open: row.get("open"),
                high: row.get("high"),
                low: row.get("low"),
                close: row.get("close"),
                created_at: row.get("created_at"),
            };
            prices.insert(price.symbol.clone(), price);
        }

        Ok(prices)
    }

    async fn store_candle(&self, candle: &Candle) -> MarketDataResult<()> {
        self.validate_symbol(&candle.symbol).await?;

        sqlx::query(
            r#"
            INSERT INTO candles (id, symbol, period, timestamp, open, high, low, close, volume, created_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
            ON CONFLICT (symbol, period, timestamp) DO UPDATE SET
                open = EXCLUDED.open,
                high = EXCLUDED.high,
                low = EXCLUDED.low,
                close = EXCLUDED.close,
                volume = EXCLUDED.volume
            "#
        )
        .bind(candle.id)
        .bind(&candle.symbol)
        .bind(&candle.period)
        .bind(candle.timestamp)
        .bind(candle.open)
        .bind(candle.high)
        .bind(candle.low)
        .bind(candle.close)
        .bind(candle.volume)
        .bind(candle.created_at)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    async fn get_candles(
        &self,
        symbol: &str,
        period: TimePeriod,
        from: DateTime<Utc>,
        to: DateTime<Utc>,
    ) -> MarketDataResult<Vec<Candle>> {
        self.validate_symbol(symbol).await?;
        self.validate_time_range(from, to).await?;

        let period_str = format!("{:?}", period);

        let rows = sqlx::query(
            r#"
            SELECT id, symbol, period, timestamp, open, high, low, close, volume, created_at
            FROM candles
            WHERE symbol = $1 AND period = $2 AND timestamp >= $3 AND timestamp <= $4
            ORDER BY timestamp ASC
            "#,
        )
        .bind(symbol)
        .bind(&period_str)
        .bind(from)
        .bind(to)
        .fetch_all(&self.pool)
        .await?;

        let candles = rows
            .into_iter()
            .map(|row| Candle {
                id: row.get("id"),
                symbol: row.get("symbol"),
                period: row.get("period"),
                timestamp: row.get("timestamp"),
                open: row.get("open"),
                high: row.get("high"),
                low: row.get("low"),
                close: row.get("close"),
                volume: row.get("volume"),
                created_at: row.get("created_at"),
            })
            .collect();

        Ok(candles)
    }

    async fn cleanup_old_prices(&self, before: DateTime<Utc>) -> MarketDataResult<u64> {
        let result = sqlx::query("DELETE FROM prices WHERE timestamp < $1")
            .bind(before)
            .execute(&self.pool)
            .await?;

        Ok(result.rows_affected())
    }
}