dataprocessor.py

import yfinance as yf
import pandas as pd
import numpy as np
from typing import List, Dict, Optional, Tuple
import os
import logging
from datetime import datetime, timedelta
import pickle
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
from sklearn.preprocessing import MinMaxScaler, StandardScaler
import warnings
warnings.filterwarnings('ignore')

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class StockDataProcessor:
    """

    A comprehensive class for downloading, processing, and preprocessing stock data

    from Yahoo Finance for reinforcement learning applications.

    """
    
    def __init__(self, data_dir: str = "stock_data", cache_dir: str = "cache"):
        self.data_dir = data_dir
        self.cache_dir = cache_dir
        self.scalers = {}
        
        # Create directories if they don't exist
        os.makedirs(data_dir, exist_ok=True)
        os.makedirs(cache_dir, exist_ok=True)
        
    def get_sp500_tickers(self) -> List[str]:
        """Get S&P 500 stock tickers"""
        try:
            # Download S&P 500 list from Wikipedia
            url = 'https://en.wikipedia.org/wiki/List_of_S%26P_500_companies'
            tables = pd.read_html(url)
            sp500_table = tables[0]
            tickers = sp500_table['Symbol'].tolist()
            # Clean tickers (remove dots, etc.)
            tickers = [ticker.replace('.', '-') for ticker in tickers]
            logger.info(f"Retrieved {len(tickers)} S&P 500 tickers")
            return tickers
        except Exception as e:
            logger.error(f"Error fetching S&P 500 tickers: {e}")
            # Fallback to a smaller list of popular stocks
            return ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA', 'META', 'NVDA', 'JPM', 'JNJ', 'V']
    
    def download_stock_data(self, 

                          ticker: str, 

                          period: str = "10y", 

                          interval: str = "1d") -> Optional[pd.DataFrame]:
        """

        Download stock data for a single ticker

        

        Args:

            ticker: Stock symbol

            period: Time period (1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max)

            interval: Data interval (1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo)

        """
        try:
            stock = yf.Ticker(ticker)
            data = stock.history(period=period, interval=interval)
            
            if data.empty:
                logger.warning(f"No data found for {ticker}")
                return None
                
            # Add ticker column
            data['Ticker'] = ticker
            data.reset_index(inplace=True)
            
            logger.info(f"Downloaded {len(data)} records for {ticker}")
            return data
            
        except Exception as e:
            logger.error(f"Error downloading data for {ticker}: {e}")
            return None
    
    def download_multiple_stocks(self, 

                                tickers: List[str], 

                                period: str = "10y",

                                interval: str = "1d",

                                max_workers: int = 10) -> pd.DataFrame:
        """

        Download stock data for multiple tickers using parallel processing

        """
        all_data = []
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all download tasks
            future_to_ticker = {
                executor.submit(self.download_stock_data, ticker, period, interval): ticker 
                for ticker in tickers
            }
            
            # Collect results
            for future in as_completed(future_to_ticker):
                ticker = future_to_ticker[future]
                try:
                    data = future.result()
                    if data is not None:
                        all_data.append(data)
                except Exception as e:
                    logger.error(f"Error processing {ticker}: {e}")
                
                # Rate limiting
                time.sleep(0.1)
        
        if all_data:
            combined_data = pd.concat(all_data, ignore_index=True)
            logger.info(f"Combined data shape: {combined_data.shape}")
            return combined_data
        else:
            return pd.DataFrame()
    
    def calculate_technical_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
        """

        Calculate technical indicators for each stock

        """
        logger.info("Calculating technical indicators...")
        
        result_dfs = []
        
        for ticker in df['Ticker'].unique():
            ticker_data = df[df['Ticker'] == ticker].copy()
            ticker_data = ticker_data.sort_values('Date')
            
            # Moving averages
            ticker_data['SMA_5'] = ticker_data['Close'].rolling(window=5).mean()
            ticker_data['SMA_10'] = ticker_data['Close'].rolling(window=10).mean()
            ticker_data['SMA_20'] = ticker_data['Close'].rolling(window=20).mean()
            ticker_data['SMA_50'] = ticker_data['Close'].rolling(window=50).mean()
            
            # Exponential moving averages
            ticker_data['EMA_12'] = ticker_data['Close'].ewm(span=12).mean()
            ticker_data['EMA_26'] = ticker_data['Close'].ewm(span=26).mean()
            
            # MACD
            ticker_data['MACD'] = ticker_data['EMA_12'] - ticker_data['EMA_26']
            ticker_data['MACD_Signal'] = ticker_data['MACD'].ewm(span=9).mean()
            ticker_data['MACD_Histogram'] = ticker_data['MACD'] - ticker_data['MACD_Signal']
            
            # RSI
            delta = ticker_data['Close'].diff()
            gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
            loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
            rs = gain / loss
            ticker_data['RSI'] = 100 - (100 / (1 + rs))
            
            # Bollinger Bands
            ticker_data['BB_Middle'] = ticker_data['Close'].rolling(window=20).mean()
            bb_std = ticker_data['Close'].rolling(window=20).std()
            ticker_data['BB_Upper'] = ticker_data['BB_Middle'] + (bb_std * 2)
            ticker_data['BB_Lower'] = ticker_data['BB_Middle'] - (bb_std * 2)
            ticker_data['BB_Width'] = ticker_data['BB_Upper'] - ticker_data['BB_Lower']
            ticker_data['BB_Position'] = (ticker_data['Close'] - ticker_data['BB_Lower']) / ticker_data['BB_Width']
            
            # Volatility
            ticker_data['Volatility'] = ticker_data['Close'].rolling(window=20).std()
            
            # Price change features
            ticker_data['Price_Change'] = ticker_data['Close'].pct_change()
            ticker_data['Price_Change_5d'] = ticker_data['Close'].pct_change(periods=5)
            ticker_data['High_Low_Ratio'] = ticker_data['High'] / ticker_data['Low']
            ticker_data['Open_Close_Ratio'] = ticker_data['Open'] / ticker_data['Close']
            
            # Volume features
            ticker_data['Volume_SMA'] = ticker_data['Volume'].rolling(window=20).mean()
            ticker_data['Volume_Ratio'] = ticker_data['Volume'] / ticker_data['Volume_SMA']
            
            result_dfs.append(ticker_data)
        
        result = pd.concat(result_dfs, ignore_index=True)
        logger.info(f"Technical indicators calculated. New shape: {result.shape}")
        return result
    
    def create_lagged_features(self, df: pd.DataFrame, lags: List[int] = [1, 2, 3, 5, 10]) -> pd.DataFrame:
        """

        Create lagged features for time series analysis

        """
        logger.info("Creating lagged features...")
        
        result_dfs = []
        feature_columns = ['Close', 'Volume', 'Price_Change', 'RSI', 'MACD', 'Volatility']
        
        for ticker in df['Ticker'].unique():
            ticker_data = df[df['Ticker'] == ticker].copy()
            ticker_data = ticker_data.sort_values('Date')
            
            for col in feature_columns:
                if col in ticker_data.columns:
                    for lag in lags:
                        ticker_data[f'{col}_lag_{lag}'] = ticker_data[col].shift(lag)
            
            result_dfs.append(ticker_data)
        
        result = pd.concat(result_dfs, ignore_index=True)
        logger.info(f"Lagged features created. New shape: {result.shape}")
        return result
    
    def create_future_returns(self, df: pd.DataFrame, horizons: List[int] = [1, 5, 10, 20]) -> pd.DataFrame:
        """

        Create future return targets for prediction

        """
        logger.info("Creating future return targets...")
        
        result_dfs = []
        
        for ticker in df['Ticker'].unique():
            ticker_data = df[df['Ticker'] == ticker].copy()
            ticker_data = ticker_data.sort_values('Date')
            
            for horizon in horizons:
                ticker_data[f'Future_Return_{horizon}d'] = ticker_data['Close'].shift(-horizon) / ticker_data['Close'] - 1
                
                # Create binary classification targets
                ticker_data[f'Future_Up_{horizon}d'] = (ticker_data[f'Future_Return_{horizon}d'] > 0).astype(int)
                
                # Create categorical targets (strong down, down, up, strong up)
                returns = ticker_data[f'Future_Return_{horizon}d']
                ticker_data[f'Future_Category_{horizon}d'] = pd.cut(
                    returns, 
                    bins=[-np.inf, -0.02, 0, 0.02, np.inf], 
                    labels=[0, 1, 2, 3]
                ).astype(float)
            
            result_dfs.append(ticker_data)
        
        result = pd.concat(result_dfs, ignore_index=True)
        logger.info(f"Future return targets created. New shape: {result.shape}")
        return result
    
    def clean_and_normalize_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """

        Clean and normalize the data for ML/RL

        """
        logger.info("Cleaning and normalizing data...")
        
        # Remove rows with too many NaN values
        df = df.dropna(thresh=len(df.columns) * 0.7)
        
        # Forward fill remaining NaN values
        numeric_columns = df.select_dtypes(include=[np.number]).columns
        df[numeric_columns] = df[numeric_columns].fillna(method='ffill')
        
        # Remove infinite values
        df = df.replace([np.inf, -np.inf], np.nan)
        df = df.dropna()
        
        logger.info(f"Data cleaned. Final shape: {df.shape}")
        return df
    
    def create_rl_states_actions(self, df: pd.DataFrame) -> Dict:
        """

        Create state and action spaces suitable for reinforcement learning

        """
        logger.info("Creating RL state and action representations...")
        
        # Define state features (technical indicators and market data)
        state_features = [
            'Open', 'High', 'Low', 'Close', 'Volume',
            'SMA_5', 'SMA_10', 'SMA_20', 'SMA_50',
            'EMA_12', 'EMA_26', 'MACD', 'MACD_Signal', 'RSI',
            'BB_Position', 'BB_Width', 'Volatility',
            'Price_Change', 'High_Low_Ratio', 'Volume_Ratio'
        ]
        
        # Add lagged features to state
        lag_features = [col for col in df.columns if '_lag_' in col]
        state_features.extend(lag_features)
        
        # Filter existing features
        state_features = [feat for feat in state_features if feat in df.columns]
        
        # Normalize state features
        scaler = StandardScaler()
        df_scaled = df.copy()
        df_scaled[state_features] = scaler.fit_transform(df[state_features])
        
        # Define action space (0: Hold, 1: Buy, 2: Sell)
        # You can expand this based on your RL strategy
        
        # Create sequences for each stock
        rl_data = {}
        sequence_length = 60  # Number of days to look back
        
        for ticker in df_scaled['Ticker'].unique():
            ticker_data = df_scaled[df_scaled['Ticker'] == ticker].sort_values('Date')
            
            states = []
            rewards = []
            dates = []
            
            for i in range(sequence_length, len(ticker_data)):
                # State: sequence of technical indicators
                state_sequence = ticker_data.iloc[i-sequence_length:i][state_features].values
                states.append(state_sequence)
                
                # Reward: next day return (can be modified based on your RL objective)
                if 'Future_Return_1d' in ticker_data.columns:
                    reward = ticker_data.iloc[i]['Future_Return_1d']
                else:
                    current_price = ticker_data.iloc[i]['Close']
                    if i < len(ticker_data) - 1:
                        next_price = ticker_data.iloc[i+1]['Close']
                        reward = (next_price - current_price) / current_price
                    else:
                        reward = 0
                
                rewards.append(reward)
                dates.append(ticker_data.iloc[i]['Date'])
            
            rl_data[ticker] = {
                'states': np.array(states),
                'rewards': np.array(rewards),
                'dates': dates,
                'state_features': state_features
            }
        
        logger.info(f"RL data created for {len(rl_data)} stocks")
        return rl_data, scaler
    
    def save_processed_data(self, data: pd.DataFrame, rl_data: Dict, scaler, filename_prefix: str = "processed_stock_data"):
        """

        Save processed data to files

        """
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        
        # Save CSV data
        csv_filename = f"{self.data_dir}/{filename_prefix}_{timestamp}.csv"
        data.to_csv(csv_filename, index=False)
        logger.info(f"CSV data saved to {csv_filename}")
        
        # Save RL data
        rl_filename = f"{self.data_dir}/{filename_prefix}_rl_{timestamp}.pkl"
        with open(rl_filename, 'wb') as f:
            pickle.dump(rl_data, f)
        logger.info(f"RL data saved to {rl_filename}")
        
        # Save scaler
        scaler_filename = f"{self.data_dir}/{filename_prefix}_scaler_{timestamp}.pkl"
        with open(scaler_filename, 'wb') as f:
            pickle.dump(scaler, f)
        logger.info(f"Scaler saved to {scaler_filename}")
        
        return csv_filename, rl_filename, scaler_filename
    
    def process_stocks_pipeline(self, 

                               tickers: Optional[List[str]] = None,

                               period: str = "10y",

                               interval: str = "1d",

                               use_sp500: bool = True) -> Tuple[pd.DataFrame, Dict, object]:
        """

        Complete pipeline for processing stock data

        """
        logger.info("Starting stock data processing pipeline...")
        
        # Get tickers
        if tickers is None:
            if use_sp500:
                tickers = self.get_sp500_tickers()
            else:
                tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA']  # Default list
        
        # Download data
        logger.info(f"Downloading data for {len(tickers)} tickers...")
        raw_data = self.download_multiple_stocks(tickers, period, interval)
        
        if raw_data.empty:
            logger.error("No data downloaded. Exiting.")
            return None, None, None
        
        # Process data
        data_with_indicators = self.calculate_technical_indicators(raw_data)
        data_with_lags = self.create_lagged_features(data_with_indicators)
        data_with_targets = self.create_future_returns(data_with_lags)
        cleaned_data = self.clean_and_normalize_data(data_with_targets)
        
        # Create RL data
        rl_data, scaler = self.create_rl_states_actions(cleaned_data)
        
        # Save data
        self.save_processed_data(cleaned_data, rl_data, scaler)
        
        logger.info("Pipeline completed successfully!")
        return cleaned_data, rl_data, scaler

# Example usage and utility functions
def example_usage():
    """

    Example of how to use the StockDataProcessor

    """
    # Initialize processor
    processor = StockDataProcessor()
    
    # Option 1: Process S&P 500 stocks
    print("Processing S&P 500 stocks...")
    data, rl_data, scaler = processor.process_stocks_pipeline(use_sp500=True, period="5y")
    
    # Option 2: Process specific stocks
    # custom_tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA', 'META', 'NVDA']
    # data, rl_data, scaler = processor.process_stocks_pipeline(tickers=custom_tickers, period="10y")
    
    if data is not None:
        print(f"Processed data shape: {data.shape}")
        print(f"Features: {data.columns.tolist()}")
        print(f"RL data available for {len(rl_data)} stocks")
        
        # Example: Access RL data for a specific stock
        if 'AAPL' in rl_data:
            aapl_states = rl_data['AAPL']['states']
            aapl_rewards = rl_data['AAPL']['rewards']
            print(f"AAPL: {aapl_states.shape[0]} sequences, each with {aapl_states.shape[1]} timesteps and {aapl_states.shape[2]} features")

def load_processed_data(rl_filename: str, scaler_filename: str) -> Tuple[Dict, object]:
    """

    Load previously processed RL data

    """
    with open(rl_filename, 'rb') as f:
        rl_data = pickle.load(f)
    
    with open(scaler_filename, 'rb') as f:
        scaler = pickle.load(f)
    
    return rl_data, scaler

if __name__ == "__main__":
    example_usage()