//! Repository implementations that wrap existing storage infrastructure use anyhow::Result; use async_trait::async_trait; use chrono::{DateTime, Utc}; use std::collections::HashMap; use std::sync::Arc; use common::MarketDataEvent; use data::providers::benzinga::{BenzingaConfig, BenzingaHistoricalProvider}; use data::providers::databento::{DatabentoConfig, DatabentoHistoricalProvider}; use data::providers::traits::{HistoricalProvider, HistoricalSchema}; use data::types::TimeRange; use crate::dbn_repository::DbnMarketDataRepository; use crate::foxhunt::tli::BacktestStatus; use crate::performance::PerformanceMetrics; use crate::repositories::{ DefaultRepositories, MarketDataRepository, NewsRepository, TradingRepository, }; use crate::storage::{BacktestSummary, StorageManager}; use crate::strategy_engine::{BacktestTrade, MarketData}; /// Market data repository implementation using data providers pub struct DataProviderMarketDataRepository { databento_provider: Arc, } impl DataProviderMarketDataRepository { /// Create a new market data repository with Databento provider pub async fn new() -> Result { let databento_config = DatabentoConfig::default(); // DatabentoHistoricalProvider::new returns Result, use await let databento_provider = Arc::new(DatabentoHistoricalProvider::new(databento_config).await?); Ok(Self { databento_provider }) } } #[async_trait] impl MarketDataRepository for DataProviderMarketDataRepository { async fn load_historical_data( &self, symbols: &[String], start_time: i64, end_time: i64, ) -> Result> { let start_date = DateTime::from_timestamp_nanos(start_time); let end_date = DateTime::from_timestamp_nanos(end_time); // Create TimeRange for the request let time_range = TimeRange::new(start_date, end_date) .map_err(|e| anyhow::anyhow!("Failed to create time range: {}", e))?; // Convert symbols to Symbol type and load data for each let mut all_market_data = Vec::new(); for symbol_str in symbols { let symbol = common::Symbol::from(symbol_str.as_str()); // Fetch historical OHLCV bars from Databento let market_events = self .databento_provider .fetch(&symbol, HistoricalSchema::OHLCV, time_range) .await?; // Convert MarketDataEvents to MarketData format for event in market_events { if let MarketDataEvent::Bar(bar_event) = event { all_market_data.push(MarketData { symbol: bar_event.symbol.to_string(), timestamp: bar_event.end_timestamp, open: bar_event.open, high: bar_event.high, low: bar_event.low, close: bar_event.close, volume: bar_event.volume, }); } } } // Sort by timestamp all_market_data.sort_by(|a, b| a.timestamp.cmp(&b.timestamp)); Ok(all_market_data) } } /// Trading repository implementation that wraps StorageManager pub struct StorageManagerTradingRepository { storage_manager: Arc, } impl StorageManagerTradingRepository { /// Create a new trading repository with storage manager pub fn new(storage_manager: Arc) -> Self { Self { storage_manager } } } #[async_trait] impl TradingRepository for StorageManagerTradingRepository { async fn save_backtest_results( &self, backtest_id: &str, trades: &[BacktestTrade], metrics: &PerformanceMetrics, ) -> Result<()> { self.storage_manager .save_backtest_results(backtest_id, trades, metrics) .await } async fn load_backtest_results( &self, backtest_id: &str, ) -> Result<(Vec, PerformanceMetrics)> { self.storage_manager .load_backtest_results(backtest_id) .await } async fn list_backtests( &self, limit: u32, offset: u32, strategy_name: Option, status_filter: Option, ) -> Result> { self.storage_manager .list_backtests(limit, offset, strategy_name, status_filter) .await } } /// News repository implementation using Benzinga provider pub struct BenzingaNewsRepository { benzinga_provider: Arc, } impl BenzingaNewsRepository { /// Create a new news repository with Benzinga provider pub async fn new() -> Result { let benzinga_config = BenzingaConfig::default(); // BenzingaHistoricalProvider::new returns Result (not async) let benzinga_provider = Arc::new(BenzingaHistoricalProvider::new(benzinga_config)?); Ok(Self { benzinga_provider }) } } #[async_trait] impl NewsRepository for BenzingaNewsRepository { async fn load_news_events( &self, symbols: &[String], start_time: DateTime, end_time: DateTime, ) -> Result> { // Convert &[String] to Vec<&str> let symbol_refs: Vec<&str> = symbols.iter().map(|s| s.as_str()).collect(); let news_events = self .benzinga_provider .get_all_events(Some(&symbol_refs), start_time, end_time) .await?; // Convert data::providers::common::NewsEvent to backtesting NewsEvent (unit struct) let converted_events: Vec<_> = news_events .iter() .map(|_| crate::strategy_engine::NewsEvent) .collect(); Ok(converted_events) } } /// Factory function to create repository implementation with dependency injection /// /// Supports two modes controlled by USE_DBN_DATA environment variable: /// - "true": Use local DBN files (for backtesting with real historical data) /// - "false" or unset: Use Databento API (default for production) /// /// When using DBN mode, the following environment variables are required: /// - DBN_SYMBOL_MAPPINGS: Comma-separated list of symbol:path pairs /// Example: "ES.FUT:test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn" pub async fn create_repositories( storage_manager: Arc, ) -> Result { // Check if we should use DBN files instead of API let use_dbn_data = std::env::var("USE_DBN_DATA") .ok() .and_then(|v| v.parse::().ok()) .unwrap_or(false); let market_data: Box = if use_dbn_data { tracing::info!("Using DBN file-based market data repository"); // Parse DBN file mappings from environment variable (symbol:path pairs) let mappings_str = std::env::var("DBN_SYMBOL_MAPPINGS").unwrap_or_else(|_| { "ES.FUT:test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn".to_string() }); let mut file_mapping = HashMap::new(); for pair in mappings_str.split(',') { let parts: Vec<&str> = pair.trim().split(':').collect(); if parts.len() == 2 { #[allow(clippy::indexing_slicing)] // Length checked above: parts.len() == 2 let symbol = parts[0].trim().to_string(); #[allow(clippy::indexing_slicing)] // Length checked above: parts.len() == 2 let path = parts[1].trim().to_string(); file_mapping.insert(symbol.clone(), path.clone()); tracing::info!(" DBN file mapping: {} -> {}", symbol, path); } } if file_mapping.is_empty() { return Err(anyhow::anyhow!( "USE_DBN_DATA=true but no valid DBN_SYMBOL_MAPPINGS provided. \ Expected format: 'SYMBOL1:path1,SYMBOL2:path2'" )); } // Parse symbol mappings (for test compatibility, e.g., BTC/USD -> ES.FUT) let symbol_map_str = std::env::var("DBN_SYMBOL_MAP").unwrap_or_default(); let mut symbol_mappings = HashMap::new(); if !symbol_map_str.is_empty() { for pair in symbol_map_str.split(',') { let parts: Vec<&str> = pair.trim().split(':').collect(); if parts.len() == 2 { let from_symbol = parts[0].trim().to_string(); let to_symbol = parts[1].trim().to_string(); symbol_mappings.insert(from_symbol.clone(), to_symbol.clone()); tracing::info!(" Symbol mapping: {} -> {}", from_symbol, to_symbol); } } } Box::new(DbnMarketDataRepository::new_with_mappings(file_mapping, symbol_mappings).await?) } else { tracing::info!("Using Databento API-based market data repository"); Box::new(DataProviderMarketDataRepository::new().await?) }; let trading = Box::new(StorageManagerTradingRepository::new(storage_manager)) as Box; let news = Box::new(BenzingaNewsRepository::new().await?) as Box; Ok(DefaultRepositories { market_data, trading, news, }) }