ParallelLLC/algorithmic_trading

1import pandas as pd2import numpy as np3import logging4import os5from typing import Dict, Any, Optional6from datetime import datetime, timedelta7 8logger = logging.getLogger(__name__)9 10def load_data(config: Dict[str, Any]) -> Optional[pd.DataFrame]:11    """12    Load market data based on configuration.13    14    Args:15        config: Configuration dictionary16        17    Returns:18        DataFrame with market data or None if error19    """20    try:21        data_source = config['data_source']['type']22        logger.info(f"Loading data from source: {data_source}")23        24        if data_source == 'alpaca':25            return _load_alpaca_data(config)26        elif data_source == 'csv':27            return _load_csv_data(config)28        elif data_source == 'synthetic':29            return _load_synthetic_data(config)30        elif data_source == 'yahoo':31            return _load_yahoo_data(config)32        else:33            logger.error(f"Unsupported data source: {data_source}")34            return None35            36    except Exception as e:37        logger.error(f"Error loading data: {e}")38        return None39 40def _load_alpaca_data(config: Dict[str, Any]) -> Optional[pd.DataFrame]:41    """Load market data from Alpaca"""42    try:43        from .alpaca_broker import AlpacaBroker44        45        # Initialize Alpaca broker46        alpaca_broker = AlpacaBroker(config)47        48        # Get symbol and timeframe from config49        symbol = config['trading']['symbol']50        timeframe = config['trading']['timeframe']51        52        # Convert timeframe to Alpaca format53        tf_map = {54            '1m': '1Min',55            '5m': '5Min', 56            '15m': '15Min',57            '1h': '1Hour',58            '1d': '1Day'59        }60        alpaca_timeframe = tf_map.get(timeframe, '1Min')61        62        # Get market data63        data = alpaca_broker.get_market_data(64            symbol=symbol,65            timeframe=alpaca_timeframe,66            limit=100067        )68        69        if data is not None and not data.empty:70            logger.info(f"Loaded {len(data)} data points from Alpaca for {symbol}")71            return data72        else:73            logger.error(f"No data returned from Alpaca for {symbol}")74            return None75            76    except Exception as e:77        logger.error(f"Error loading Alpaca data: {e}")78        return None79 80def _load_yahoo_data(config: Dict[str, Any]) -> Optional[pd.DataFrame]:81    """Load OHLCV bars from Yahoo Finance via yfinance. Does not replace Alpaca."""82    try:83        from .yahoo_data_stream import YahooDataStream84 85        stream = YahooDataStream(config)86        trading = config.get('trading', {})87        symbol = trading.get('symbol') or (trading.get('symbols') or ['AAPL'])[0]88        yahoo_cfg = config.get('yahoo', {})89        start_date = yahoo_cfg.get('start_date', '2024-01-01')90        end_date = yahoo_cfg.get('end_date', '2026-12-31')91 92        data = stream.get_historical_data(symbol, start_date, end_date)93        if data is None or data.empty:94            logger.error("No Yahoo data returned for %s", symbol)95            return None96 97        logger.info("Loaded %s Yahoo bars for %s", len(data), symbol)98        return data99    except Exception as e:100        logger.error("Error loading Yahoo data: %s", e)101        return None102 103 104def _load_csv_data(config: Dict[str, Any]) -> Optional[pd.DataFrame]:105    """Load market data from CSV file"""106    try:107        file_path = config['data_source']['path']108        109        if not os.path.exists(file_path):110            logger.error(f"CSV file not found: {file_path}")111            return None112        113        # Load CSV data114        data = pd.read_csv(file_path)115        116        # Handle both 'timestamp' and 'date' column names117        if 'date' in data.columns and 'timestamp' not in data.columns:118            data = data.rename(columns={'date': 'timestamp'})119        120        # Ensure required columns exist121        required_columns = ['timestamp', 'open', 'high', 'low', 'close', 'volume']122        missing_columns = [col for col in required_columns if col not in data.columns]123        124        if missing_columns:125            logger.error(f"Missing required columns: {missing_columns}")126            return None127        128        # Convert timestamp to datetime129        data['timestamp'] = pd.to_datetime(data['timestamp'])130        131        # Sort by timestamp132        data = data.sort_values('timestamp').reset_index(drop=True)133        134        logger.info(f"Loaded {len(data)} data points from CSV: {file_path}")135        return data136        137    except Exception as e:138        logger.error(f"Error loading CSV data: {e}")139        return None140 141def _load_synthetic_data(config: Dict[str, Any]) -> Optional[pd.DataFrame]:142    """Load or generate synthetic market data"""143    try:144        synthetic_config = config.get('synthetic_data', {})145        data_path = synthetic_config.get('data_path', 'data/synthetic_market_data.csv')146        147        # Check if synthetic data file exists148        if os.path.exists(data_path):149            logger.info(f"Loading existing synthetic data from: {data_path}")150            return _load_csv_data({'data_source': {'path': data_path}})151        152        # Generate new synthetic data153        logger.info("Generating new synthetic market data")154        from .synthetic_data_generator import SyntheticDataGenerator155        156        generator = SyntheticDataGenerator(config)157        data = generator.generate_data()158        159        if data is not None and not data.empty:160            # Save generated data161            os.makedirs(os.path.dirname(data_path), exist_ok=True)162            data.to_csv(data_path, index=False)163            logger.info(f"Saved synthetic data to: {data_path}")164            return data165        else:166            logger.error("Failed to generate synthetic data")167            return None168        169    except Exception as e:170        logger.error(f"Error loading synthetic data: {e}")171        return None172 173def validate_data(data: pd.DataFrame) -> bool:174    """175    Validate market data quality.176    177    Args:178        data: DataFrame with market data179        180    Returns:181        True if data is valid, False otherwise182    """183    try:184        if data is None or data.empty:185            logger.error("Data is None or empty")186            return False187        188        # Handle both 'timestamp' and 'date' column names189        if 'date' in data.columns and 'timestamp' not in data.columns:190            data = data.rename(columns={'date': 'timestamp'})191        192        # Check required columns193        required_columns = ['timestamp', 'open', 'high', 'low', 'close', 'volume']194        missing_columns = [col for col in required_columns if col not in data.columns]195        196        if missing_columns:197            logger.error(f"Missing required columns: {missing_columns}")198            return False199        200        # Check for NaN values201        nan_counts = data[required_columns].isna().sum()202        if nan_counts.sum() > 0:203            logger.warning(f"Found NaN values: {nan_counts.to_dict()}")204            # Remove rows with NaN values205            data.dropna(subset=required_columns, inplace=True)206            logger.info(f"Removed {nan_counts.sum()} rows with NaN values")207        208        # Check for negative prices209        price_columns = ['open', 'high', 'low', 'close']210        negative_prices = data[price_columns] < 0211        if negative_prices.any().any():212            logger.error("Found negative prices in data")213            return False214        215        # Check for zero volumes216        zero_volumes = data['volume'] == 0217        if zero_volumes.sum() > len(data) * 0.5:  # More than 50% zero volumes218            logger.warning("High percentage of zero volumes detected")219        220        # Check OHLC consistency221        invalid_ohlc = (222            (data['high'] < data['low']) |223            (data['open'] > data['high']) |224            (data['close'] > data['high']) |225            (data['open'] < data['low']) |226            (data['close'] < data['low'])227        )228        229        if invalid_ohlc.any():230            logger.error("Found invalid OHLC relationships")231            return False232        233        # Check timestamp consistency234        if 'timestamp' in data.columns:235            timestamps = pd.to_datetime(data['timestamp'])236            if not timestamps.is_monotonic_increasing:237                logger.warning("Timestamps are not in ascending order")238                data = data.sort_values('timestamp').reset_index(drop=True)239        240        logger.info(f"Data validation passed: {len(data)} valid records")241        return True242        243    except Exception as e:244        logger.error(f"Error validating data: {e}")245        return False246 247def add_technical_indicators(data: pd.DataFrame) -> pd.DataFrame:248    """249    Add technical indicators to market data.250    251    Args:252        data: DataFrame with OHLCV data253        254    Returns:255        DataFrame with technical indicators added256    """257    try:258        df = data.copy()259        260        # Simple Moving Averages261        df['sma_20'] = df['close'].rolling(window=20).mean()262        df['sma_50'] = df['close'].rolling(window=50).mean()263        df['sma_200'] = df['close'].rolling(window=200).mean()264        265        # Exponential Moving Averages266        df['ema_12'] = df['close'].ewm(span=12).mean()267        df['ema_26'] = df['close'].ewm(span=26).mean()268        269        # MACD270        df['macd'] = df['ema_12'] - df['ema_26']271        df['macd_signal'] = df['macd'].ewm(span=9).mean()272        df['macd_histogram'] = df['macd'] - df['macd_signal']273        274        # RSI275        delta = df['close'].diff()276        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()277        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()278        rs = gain / loss279        df['rsi'] = 100 - (100 / (1 + rs))280        281        # Bollinger Bands282        df['bb_middle'] = df['close'].rolling(window=20).mean()283        bb_std = df['close'].rolling(window=20).std()284        df['bb_upper'] = df['bb_middle'] + (bb_std * 2)285        df['bb_lower'] = df['bb_middle'] - (bb_std * 2)286        287        # Average True Range (ATR)288        high_low = df['high'] - df['low']289        high_close = np.abs(df['high'] - df['close'].shift())290        low_close = np.abs(df['low'] - df['close'].shift())291        true_range = np.maximum(high_low, np.maximum(high_close, low_close))292        df['atr'] = true_range.rolling(window=14).mean()293        294        # Volume indicators295        df['volume_sma'] = df['volume'].rolling(window=20).mean()296        df['volume_ratio'] = df['volume'] / df['volume_sma']297        298        # Price momentum299        df['price_change'] = df['close'].pct_change()300        df['price_change_5'] = df['close'].pct_change(periods=5)301        df['price_change_20'] = df['close'].pct_change(periods=20)302        303        logger.info("Technical indicators added successfully")304        return df305        306    except Exception as e:307        logger.error(f"Error adding technical indicators: {e}")308        return data309 310def get_latest_data(data: pd.DataFrame, n_periods: int = 100) -> pd.DataFrame:311    """312    Get the latest n periods of data.313    314    Args:315        data: DataFrame with market data316        n_periods: Number of periods to return317        318    Returns:319        DataFrame with latest n periods320    """321    try:322        if len(data) <= n_periods:323            return data324        325        return data.tail(n_periods).reset_index(drop=True)326        327    except Exception as e:328        logger.error(f"Error getting latest data: {e}")329        return data330