nseprocessor.py

"""
NSE Bhavcopy Data Processor

This script processes NSE bhavcopy files with proper DataFrame handling and enhanced logging.
Supports both fresh processing and incremental updates.

Features:
- Date-ordered processing
- Proper DataFrame operations
- Enhanced logging
- Progress tracking
- Data validation
- Optional corporate action application for full processing runs

Author: Your Name
Date: June 2025 (Updated)
"""

import pandas as pd
import numpy as np
import zipfile
import re
import logging
from pathlib import Path
from datetime import datetime, timedelta
import time
from pathlib import Path
import logging
from tqdm import tqdm
import sys
import gc
import json
import talib
import fastparquet
import warnings
from sklearn.preprocessing import MinMaxScaler
from sklearn.feature_selection import mutual_info_regression

warnings.filterwarnings('ignore')
warnings.simplefilter('ignore', FutureWarning)

class NSEDataProcessor:
    """Process NSE bhavcopy data with optimized display and relative paths"""

    def __init__(self):
        self.setup_paths()
        self.setup_logging()
        self.initialize_metrics()
        
        # Define schema (existing schema code)
        self.schema = {
            'SYMBOL': 'category',
            'SECURITY': 'category', 
            'SERIES': 'category',
            'TRAN_DATE': 'datetime64[ns]',
            'OPEN_PRICE': 'float32',
            'HIGH_PRICE': 'float32', 
            'LOW_PRICE': 'float32',
            'CLOSE_PRICE': 'float32',
            'PREV_CL_PR': 'float32',
            'NET_TRDVAL': 'float64',
            'NET_TRDQTY': 'int64',
            'TRADES': 'int64',
            'HI_52_WK': 'float32',
            'LO_52_WK': 'float32',
            'HI_DIFF': 'float32',
            'LO_DIFF': 'float32',
            'DAILY_RETURN': 'float32',
            'VOLATILITY': 'float32'
            # Additional feature engineering columns, added runtime and saved to extract file
        }
        
        self.string_cols = ['MKT', 'SERIES', 'SYMBOL', 'SECURITY']
        self.numeric_cols = [
            'PREV_CL_PR', 'OPEN_PRICE', 'HIGH_PRICE', 'LOW_PRICE',
            'CLOSE_PRICE', 'NET_TRDVAL', 'NET_TRDQTY', 'TRADES',
            'HI_52_WK', 'LO_52_WK'
        ]
        self.corporate_actions = [] # Initialize corporate actions list

    def setup_paths(self):
        """Setup using relative paths"""
        try:
            self.base_dir = Path.cwd()
            self.downloads_dir = self.base_dir / 'downloads'
            self.processed_dir = self.base_dir / 'processed'
            self.corpact_file = self.processed_dir / 'nse_dashboard_corpact.json'
            self.extract_file = self.processed_dir / 'nse_extract_data.parquet'
            self.parquet_path = self.processed_dir / 'nse_processed_data.parquet'
            
            if not self.downloads_dir.exists():
                raise FileNotFoundError(f"Downloads directory not found: {self.downloads_dir}")
                
            self.processed_dir.mkdir(parents=True, exist_ok=True)
            
        except Exception as e:
            print(f"Path setup failed: {str(e)}")
            sys.exit(1)

    def setup_logging(self):
        """Configure minimal logging"""
        try:
            log_format = '%(asctime)s - %(levelname)s - %(message)s'
            logging.basicConfig(
                filename=self.downloads_dir / 'bulk_data_processing.log',
                level=logging.INFO,
                format=log_format
            )
            self.logger = logging.getLogger(__name__)
            
        except Exception as e:
            print(f"Logging setup failed: {str(e)}")
            sys.exit(1)
           
    def initialize_metrics(self):
        """Initialize processing metrics"""
        self.metrics = {
            'start_time': None,
            'files_found': 0,
            'files_processed': 0,
            'files_failed': 0,
            'rows_processed': 0,
            'processing_errors': [],
            'processed_files': [],
            'failed_files': [],
            'start_datetime': None,
            'end_datetime': None,
            'duration': None,
            'batch_metrics': [],
            'corporate_actions_applied_count': 0 # New metric for corporate actions
        }
        
    def get_latest_processed_date(self):
        """Get latest processed date from existing output"""
        try:
            if self.parquet_path.exists():
                df = pd.read_parquet(self.parquet_path, columns=['TRAN_DATE'])
                latest_date = pd.to_datetime(df['TRAN_DATE']).max()
                self.logger.info(f"Latest processed date: {latest_date.strftime('%Y-%m-%d')}")
                print(f"Found existing data - processing files newer than {latest_date.strftime('%Y-%m-%d')}")
                return latest_date
            else:
                self.logger.info("No existing data found - performing fresh processing")
                print("Starting fresh processing - no existing data found")
                return None
        except Exception as e:
            self.logger.error(f"Failed to get latest date: {str(e)}")
            return None

    def extract_date_from_filename(self, filename):
        """Extract and validate date from filename (supports PdDDMMYY, PRDDMMYY, pdDDMMYYYY)"""
        try:
            name = Path(filename).stem

            # Match old format: Pd or PR + ddmmyy
            match_old = re.search(r'(?i)(Pd|PR)(\d{2})(\d{2})(\d{2})', name)

            # Match new format: pd + ddmmyyyy
            match_new = re.search(r'(?i)pd(\d{2})(\d{2})(\d{4})', name)

            if match_new:
                day, month, year = match_new.groups()
            elif match_old:
                day, month, year = match_old.groups()[1:]  # skip prefix
                year = str(2000 + int(year))  # convert yy to yyyy
            else:
                return None

            date_obj = datetime(int(year), int(month), int(day))

            if date_obj.year < 2000 or date_obj > datetime.now():
                raise ValueError(f"Date out of valid range: {date_obj}")

            return date_obj.strftime('%Y-%m-%d')

        except Exception as e:
            self.logger.error(f"Date extraction failed for {filename}: {str(e)}")
            return None

    def process_zip_file(self, zip_path):
        """Process ZIP file with error handling"""
        try:
            with zipfile.ZipFile(zip_path) as zf:
                csv_files = [f for f in zf.namelist()
                             if re.match(r'^pd\d{6}(\d{2})?\.csv$', Path(f).name, re.IGNORECASE)]
                
                if not csv_files:
                    self.logger.warning(f"No matching CSV in {zip_path}")
                    return None
                    
                csv_file = csv_files[0]
                tran_date = self.extract_date_from_filename(Path(csv_file).name)
                if not tran_date:
                    return None
                    
                with zf.open(csv_file) as f:
                    df = pd.read_csv(f, thousands=',')
                    return self.process_dataframe(df, tran_date)
                    
        except Exception as e:
            self.logger.error(f"ZIP processing failed - {zip_path}: {str(e)}")
            self.metrics['files_failed'] += 1
            return None

    def process_dataframe(self, df, tran_date):
        """Process DataFrame with proper indexing"""
        try:
            # Create explicit copy
            df = df.copy()
            
            # Clean string columns
            for col in self.string_cols:
                if col in df.columns:
                    df.loc[:, col] = df[col].astype(str).str.strip()
            
            # Filter rows and create new copy
            mask = (
                (df['SERIES'].str.upper() == 'EQ') & 
                df['MKT'].notna()
            )
            mask_index = (
                df['MKT'].str.upper() == 'Y'
            )
            df_index = df[mask_index].copy()
            df = df[mask].copy()
            
            # Create SYMBOL column for df_index
            if not df_index.empty:
                df_index.loc[:, 'SYMBOL'] = df_index['SECURITY'].str.upper().str.replace(r'\s+','', regex=True)
                df_index = df_index[df_index['SYMBOL'].notna()]
                df_index = df_index.drop_duplicates(subset=['SYMBOL'])

            # Concat df_index to df
            df = pd.concat([df, df_index], ignore_index=True)

            # Add transaction date
            df.loc[:, 'TRAN_DATE'] = pd.to_datetime(tran_date)
            
            # Process numeric columns
            for col in self.numeric_cols:
                if col in df.columns:
                    df.loc[:, col] = pd.to_numeric(
                        df[col].astype(str).str.replace(',', ''),
                        errors='coerce'
                    )
            
            # Drop rows with missing numerics
            df = df.dropna(subset=self.numeric_cols)
            
            if df.empty:
                return None
                
            # Calculate indicators and validate
            df = self.add_technical_indicators(df)
            df = self.validate_schema(df)
            
            if df is not None:
                self.metrics['rows_processed'] += len(df)
                
            return df
            
        except Exception as e:
            self.logger.error(f"DataFrame processing error: {str(e)}")
            return None

    def add_technical_indicators(self, df):
        """Calculate technical indicators with proper dtype handling"""
        try:
            df = df.copy()
            
            # Initialize indicator columns with np.nan
            indicators = ['HI_DIFF', 'LO_DIFF', 'DAILY_RETURN', 'VOLATILITY']
            for col in indicators:
                df[col] = np.nan
            
            # Helper function for safe calculation
            def calculate_pct_change(a, b, divisor):
                """Calculate percentage change with proper dtype handling"""
                result = (
                    ((a.astype('float32') - b.astype('float32')) / 
                    divisor.astype('float32') * 100)
                    .round(2)
                    .astype('float32')
                )
                return result
            
            # Calculate HI_DIFF
            mask = df['HI_52_WK'] > 0
            if mask.any():
                df.loc[mask, 'HI_DIFF'] = calculate_pct_change(
                    df.loc[mask, 'HI_52_WK'],
                    df.loc[mask, 'CLOSE_PRICE'],
                    df.loc[mask, 'HI_52_WK']
                )
            
            # Calculate LO_DIFF
            mask = df['LO_52_WK'] > 0
            if mask.any():
                df.loc[mask, 'LO_DIFF'] = calculate_pct_change(
                    df.loc[mask, 'CLOSE_PRICE'],
                    df.loc[mask, 'LO_52_WK'],
                    df.loc[mask, 'LO_52_WK']
                )
            
            # Calculate DAILY_RETURN
            mask = df['PREV_CL_PR'] > 0
            if mask.any():
                df.loc[mask, 'DAILY_RETURN'] = calculate_pct_change(
                    df.loc[mask, 'CLOSE_PRICE'],
                    df.loc[mask, 'PREV_CL_PR'],
                    df.loc[mask, 'PREV_CL_PR']
                )
            
            # Calculate VOLATILITY
            mask = df['OPEN_PRICE'] > 0
            if mask.any():
                high_low_diff = (df.loc[mask, 'HIGH_PRICE'] - df.loc[mask, 'LOW_PRICE'])
                df.loc[mask, 'VOLATILITY'] = calculate_pct_change(
                    high_low_diff,
                    pd.Series(0, index=high_low_diff.index),
                    df.loc[mask, 'OPEN_PRICE']
                )
            
            # Drop rows where any indicator is missing
            return df.dropna(subset=indicators)
            
        except Exception as e:
            self.logger.error(f"Error calculating indicators: {str(e)}")
            return df
        
    def _create_forward_target(self, group, lookahead_days=1):
        """
        Creates a forward-looking regression target: the next closing price after `lookahead_days`.
        Optionally allows for volatility computation or future annotation.
        Returns a DataFrame with 'TARGET_PRICES' column containing single future prices.
        """
        # Ensure chronological order
        group = group.sort_values(by='TRAN_DATE').copy()

        # Extract future single-step targets
        close_prices = group['CLOSE_PRICE'].values
        future_targets = []

        for i in range(len(close_prices)):
            if i + lookahead_days < len(close_prices):
                future_targets.append(close_prices[i + lookahead_days])
            else:
                future_targets.append(np.nan)

        group['TARGET_PRICES'] = future_targets

        # Drop rows where TARGET_PRICES is NaN
        group = group[~pd.isna(group['TARGET_PRICES'])].reset_index(drop=True)

        return group

    def _create_simple_target(self, group, lookahead_days=1, method="single"):
        """
            Generates regression targets for each row based on future closing prices.

        Parameters:
        - group: DataFrame containing at least 'TRAN_DATE' and 'CLOSE_PRICE'
        - lookahead_days: Number of days to look ahead for target generation
        - method: 'single' for next-day price, 'mean' for average of next N days

        Returns:
        - Modified DataFrame with a new column 'TARGET_PRICES'
        """
        # Ensure chronological order
        group = group.sort_values(by='TRAN_DATE').copy()

        # Extract close prices
        close_prices = group['CLOSE_PRICE'].values

        # Generate targets using vectorized slicing
        if method == "single":
            target = np.roll(close_prices, -lookahead_days)
        elif method == "mean":
            target = np.array([
                np.mean(close_prices[i+1:i+1+lookahead_days])
                if i + lookahead_days < len(close_prices) else np.nan
                for i in range(len(close_prices))
            ])
        else:
            raise ValueError(f"Unsupported method: {method}")

        # Assign and clean
        group['TARGET_PRICES'] = target
        group = group[~pd.isna(group['TARGET_PRICES'])].reset_index(drop=True)

        return group


    def create_features(self):
        logging.info("Calculating daily returns and volatility...")

        # Wrap around feature generation in a tqdm progress bar
        with tqdm(total=25, desc="Generating features") as pbar:

            # Daily Return
            self.df['DAILY_RETURN'] = self.df.groupby('SYMBOL', observed=False)['CLOSE_PRICE'].pct_change() * 100
            self.df['DAILY_RETURN'] = self.df['DAILY_RETURN'].fillna(0)

            # Volatility
            self.df['VOLATILITY'] = self.df.groupby('SYMBOL', observed=False)['DAILY_RETURN'].transform(
                lambda x: x.rolling(window=30, min_periods=1).std()
            )
            self.df['VOLATILITY'] = self.df['VOLATILITY'].fillna(0)
            pbar.update(1)

            # RSI
            logging.info("Calculating RSI, MACD, Bollinger Bands, and Moving Averages...")
            def compute_rsi(series, window=14):
                delta = series.diff()
                gain = delta.where(delta > 0, 0.0)
                loss = -delta.where(delta < 0, 0.0)
                avg_gain = gain.rolling(window=window, min_periods=1).mean()
                avg_loss = loss.rolling(window=window, min_periods=1).mean()
                rs = avg_gain / (avg_loss + 1e-6)
                rsi = 100 - (100 / (1 + rs))
                return rsi.fillna(0)

            self.df['RSI'] = self.df.groupby('SYMBOL', observed=False)['CLOSE_PRICE'].transform(compute_rsi)
            pbar.update(1)

            # --- MACD Calculation ---
            def compute_macd(series, fast=12, slow=26, signal=9):
                ema_fast = series.ewm(span=fast, min_periods=1).mean()
                ema_slow = series.ewm(span=slow, min_periods=1).mean()
                macd_line = ema_fast - ema_slow
                signal_line = macd_line.ewm(span=signal, min_periods=1).mean()
                return pd.DataFrame({'MACD': macd_line, 'MACD_SIGNAL': signal_line})

            macd_df = (
                self.df.groupby('SYMBOL', observed=False)['CLOSE_PRICE']
                .apply(compute_macd)
                .reset_index(level=0, drop=True)
            )
            self.df[['MACD', 'MACD_SIGNAL']] = macd_df
            pbar.update(1)

            # --- Bollinger Bands ---
            def compute_bollinger(series, window=20):
                sma = series.rolling(window=window, min_periods=1).mean()
                std = series.rolling(window=window, min_periods=1).std()
                upper = sma + (2 * std)
                lower = sma - (2 * std)
                return pd.DataFrame({'BB_UPPER': upper, 'BB_LOWER': lower})

            bb_df = (
                self.df.groupby('SYMBOL', observed=False)['CLOSE_PRICE']
                .apply(compute_bollinger)
                .reset_index(level=0, drop=True)
            )
            self.df[['BB_UPPER', 'BB_LOWER']] = bb_df
            pbar.update(1)

            # --- Simple Moving Averages ---
            for window in [9, 14, 20, 30]:
                self.df[f'DMA_{window}'] = (
                    self.df.groupby('SYMBOL', observed=False)['CLOSE_PRICE']
                    .transform(lambda x: x.rolling(window=window, min_periods=1).mean())
                )
            pbar.update(1)

            # --- Type Conversion for TA-Lib Compatibility ---
            float_columns = [
                'HIGH_PRICE', 'LOW_PRICE', 'CLOSE_PRICE', 'NET_TRDQTY',
                'DAILY_RETURN', 'RSI', 'MACD', 'TRADES'
            ]
            self.df[float_columns] = self.df[float_columns].astype('float64')
            pbar.update(1)

            # --- ADX ---
            self.df['ADX'] = (
                self.df[['SYMBOL', 'HIGH_PRICE', 'LOW_PRICE', 'CLOSE_PRICE']]
                .groupby('SYMBOL', observed=False)
                .apply(lambda x: pd.Series(
                    talib.ADX(x['HIGH_PRICE'].values, x['LOW_PRICE'].values, x['CLOSE_PRICE'].values, timeperiod=14),
                    index=x.index
                ))
                .droplevel(0)  # Remove group index
            )
            pbar.update(1)

            # --- OBV ---
            logging.info("Calculating OBV (On-Balance Volume)...")
            self.df['OBV'] = (
                self.df[['SYMBOL', 'CLOSE_PRICE', 'NET_TRDQTY']]
                .groupby('SYMBOL', observed=False)
                .apply(lambda x: pd.Series(
                    talib.OBV(x['CLOSE_PRICE'].values, x['NET_TRDQTY'].values),
                    index=x.index
                ))
                .droplevel(0)
            )
            pbar.update(1)

            # --- Stochastic Oscillator ---
            logging.info("Calculating Stochastic Oscillator (%K, %D)...")
            self.df[['STOCH_K', 'STOCH_D']] = (
                self.df[['SYMBOL', 'HIGH_PRICE', 'LOW_PRICE', 'CLOSE_PRICE']]
                .groupby('SYMBOL', observed=False)
                .apply(lambda x: pd.DataFrame({
                    'STOCH_K': talib.STOCH(x['HIGH_PRICE'].values, x['LOW_PRICE'].values, x['CLOSE_PRICE'].values)[0],
                    'STOCH_D': talib.STOCH(x['HIGH_PRICE'].values, x['LOW_PRICE'].values, x['CLOSE_PRICE'].values)[1]
                }, index=x.index))
                .droplevel(0)
            )
            pbar.update(1)

            # --- CCI ---
            logging.info("Calculating CCI (Commodity Channel Index)...")
            self.df['CCI'] = (
                self.df[['SYMBOL', 'HIGH_PRICE', 'LOW_PRICE', 'CLOSE_PRICE']]
                .groupby('SYMBOL', observed=False)
                .apply(lambda x: pd.Series(
                    talib.CCI(x['HIGH_PRICE'].values, x['LOW_PRICE'].values, x['CLOSE_PRICE'].values, timeperiod=20),
                    index=x.index
                ))
                .droplevel(0)
            )
            pbar.update(1)

            # --- ATR ---
            logging.info("Calculating ATR (Average True Range)...")
            self.df['ATR'] = (
                self.df[['SYMBOL', 'HIGH_PRICE', 'LOW_PRICE', 'CLOSE_PRICE']]
                .groupby('SYMBOL', observed=False)
                .apply(lambda x: pd.Series(
                    talib.ATR(x['HIGH_PRICE'].values, x['LOW_PRICE'].values, x['CLOSE_PRICE'].values, timeperiod=14),
                    index=x.index
                ))
                .droplevel(0)
            )
            pbar.update(1)

            # --- Momentum ---
            logging.info("Calculating Momentum (n-days)...")
            self.df['MOMENTUM_10'] = self.df.groupby('SYMBOL', observed=False)['CLOSE_PRICE'].transform(lambda x: x.diff(periods=10))
            pbar.update(1)

            # --- Rolling Mean & Std Dev ---
            logging.info("Calculating Rolling Mean & Std Dev of DAILY_RETURN...")
            self.df['ROLL_MEAN_10'] = self.df.groupby('SYMBOL', observed=False)['DAILY_RETURN'].transform(lambda x: x.rolling(window=10).mean())
            self.df['ROLL_STD_10'] = self.df.groupby('SYMBOL', observed=False)['DAILY_RETURN'].transform(lambda x: x.rolling(window=10).std())
            pbar.update(1)

            # --- Z-score ---
            logging.info("Calculating Z-score of CLOSE_PRICE...")
            self.df['Z_SCORE'] = self.df.groupby('SYMBOL', observed=False)['CLOSE_PRICE'].transform(lambda x: (x - x.mean()) / (x.std() + 1e-6))
            pbar.update(1)

            # --- Lag Features ---
            logging.info("Creating lag features...")
            self.df['RSI_LAG_1'] = self.df.groupby('SYMBOL', observed=False)['RSI'].shift(1)
            self.df['MACD_LAG_2'] = self.df.groupby('SYMBOL', observed=False)['MACD'].shift(2)
            pbar.update(1)

            # --- Cumulative Return ---
            logging.info("Calculating cumulative return (5-day)...")
            self.df['CUM_RETURN_5'] = self.df.groupby('SYMBOL', observed=False)['DAILY_RETURN'].transform(lambda x: x.rolling(window=5).sum())
            pbar.update(1)

            # --- Feature Interactions ---
            logging.info("Calculating feature interactions...")
            self.df['MACD_RSI_RATIO'] = self.df['MACD'] / (self.df['RSI'] + 1e-6)
            self.df['VOL_TRADES'] = self.df['VOLATILITY'] * self.df['TRADES']
            pbar.update(1)

            # --- Time-Based Features ---
            logging.info("Calculating time-based features...")
            self.df['DAY_OF_WEEK'] = pd.to_datetime(self.df['TRAN_DATE']).dt.dayofweek
            self.df['MONTH'] = pd.to_datetime(self.df['TRAN_DATE']).dt.month
            pbar.update(1)

            # --- Price Delta Features ---
            logging.info("Calculating price delta features...")
            self.df['PRICE_DELTA_9'] = self.df['CLOSE_PRICE'] - self.df['DMA_9']
            self.df['PRICE_DELTA_9_14'] = self.df['DMA_9'] - self.df['DMA_14']
            pbar.update(1)

            # --- Momentum Slope ---
            logging.info("Calculating momentum slope features...")
            self.df['MACD_LAG_1'] = self.df.groupby('SYMBOL', observed=False)['MACD'].shift(1)
            self.df['MOMENTUM_SLOPE'] = self.df['MACD'] - self.df['MACD_LAG_1']
            pbar.update(1)

            # --- Volatility Breakout Flag ---
            logging.info("Calculating volatility breakout flag features...")
            self.df['ATR_ROLL_MEAN_14'] = self.df.groupby('SYMBOL', observed=False)['ATR'].transform(lambda x: x.rolling(window=14).mean())
            self.df['VOLATILITY_BREAKOUT'] = (self.df['ATR'] > self.df['ATR_ROLL_MEAN_14']).astype(int)
            pbar.update(1)

            # --- RSI Divergence ---
            logging.info("Calculating RSI divergence features...")
            self.df['RSI_DIVERGENCE'] = self.df['RSI'] - self.df['RSI_LAG_1']
            pbar.update(1)

            # --- Bollinger Band Width ---
            logging.info("Calculating Bollinger band width features...")
            self.df['BB_WIDTH'] = (self.df['BB_UPPER'] - self.df['BB_LOWER']) / (self.df['DMA_20'] + 1e-6)
            pbar.update(1)

            # --- Lagged Returns ---
            logging.info("Calculating lagged returns features...")
            self.df['DAILY_RETURN_LAG_1'] = self.df.groupby('SYMBOL', observed=False)['DAILY_RETURN'].shift(1)
            self.df['DAILY_RETURN_LAG_2'] = self.df.groupby('SYMBOL', observed=False)['DAILY_RETURN'].shift(2)
            pbar.update(1)

            # --- NIFTY50 Index Features ---
            logging.info("Checking for NIFTY50 index data...")
            if 'NIFTY50' in self.df['SYMBOL'].values:
                logging.info("Calculating NIFTY50 index features...")
                nifty_df = self.df[self.df['SYMBOL'] == 'NIFTY50'].copy()
                nifty_df = nifty_df[['TRAN_DATE', 'CLOSE_PRICE']].rename(columns={'CLOSE_PRICE': 'NIFTY50_CLOSE'})
                self.df = self.df.merge(nifty_df, on='TRAN_DATE', how='left')
                self.df['NIFTY50_RETURN'] = self.df.groupby('SYMBOL', observed=False)['NIFTY50_CLOSE'].pct_change() * 100
                self.df['NIFTY50_RETURN'] = self.df['NIFTY50_RETURN'].fillna(0)
            pbar.update(1)

    def create_target(self):
        # Add a column for the symbol name
        self.df['SYMBOL'] = self.df['SYMBOL'].astype('category')
        # Ensure the DataFrame is sorted by SYMBOL and TRAN_DATE
        self.df.sort_values(by=['SYMBOL', 'TRAN_DATE'], inplace=True)
        self.df.reset_index(drop=True, inplace=True)
        
        # Calculate and populate RECOMMEND column using the chosen strategy
        # Options: 'strategy' (Ultra Profit Maximizer), 'forward' (future price movement), 'simple' (basic up/down/flat)
        target_creator_type = 'simple' # Options: 'forward', 'simple'
        logging.info(f"Generating Type: {target_creator_type} target price for model training...")
        
        if target_creator_type == 'forward':
            self.df = self.df.groupby('SYMBOL').apply(
                self._create_forward_target, 
                lookahead_days=1
            ).reset_index(drop=True)
        elif target_creator_type == 'simple':
            self.df = self.df.groupby('SYMBOL').apply(
                self._create_simple_target, 
                lookahead_days=1
            ).reset_index(drop=True)

        # Verify if TARGET_PRICES column is created
        if 'TARGET_PRICES' not in self.df.columns:
            logging.error("TARGET_PRICES column not found after target creation...")

    def run_correlation_analysis(self, df):
        try:
            print("\nRunning correlation analysis.")
            numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()

            # Drop rows with NaNs to ensure clean correlation
            clean_df = df[numeric_cols].dropna()

            # Compute correlation matrix
            corr_matrix = clean_df.corr()

            # Focus on correlation with CLOSE_PRICE
            target_corr = corr_matrix['CLOSE_PRICE'].sort_values(ascending=False)

            print("\n📈 Correlation with CLOSE_PRICE:")
            print(target_corr.to_string(float_format=lambda x: f"{x:.4f}"))

            # Optionally, show top 5 positively and negatively correlated features
            print("\n⬆️ Top 5 positively correlated features:")
            print(target_corr.drop('CLOSE_PRICE').head(5).to_string(float_format=lambda x: f"{x:.4f}"))

            print("\n⬇️ Top 5 negatively correlated features:")
            print(target_corr.drop('CLOSE_PRICE').tail(5).to_string(float_format=lambda x: f"{x:.4f}"))

            # Mutual Information Analysis
            print("\nCalculating Mutual Information Scores.")
            X = clean_df.drop(columns=['CLOSE_PRICE'], errors='ignore')
            y = clean_df['CLOSE_PRICE']
            scaler = MinMaxScaler()
            X_scaled = scaler.fit_transform(X)

            mi_scores = mutual_info_regression(X_scaled, y)
            mi_series = pd.Series(mi_scores, index=X.columns).sort_values(ascending=False)

            print("\n📊 Mutual Information Scores:")
            print(mi_series.to_string(float_format=lambda x: f"{x:.4f}"))

        except Exception as e:
            logging.error(f"Correlation analysis failed: {e}")


    def save_extract_file(self):
        # Final cleanup and save to Parquet
        logging.info("Saving processed data to Parquet file...")
        self.df.dropna(inplace=True)
        self.df.sort_values(by=['SYMBOL', 'TRAN_DATE'], inplace=True)
        self.df.reset_index(drop=True, inplace=True)
        self.df.to_parquet(self.extract_file, index=False, compression='snappy')
        logging.info(f"Processed data saved to {self.extract_file}")

    def load_data(self, label_creator_type = 'simple'):
            """
            Loads stock data from CSV, performs date filtering, and calculates
            various technical indicators and lag features.
            """
            logging.info("Loading stock data... This may take a moment.")
            try:
                # Define data types for efficient memory usage and accurate loading
                dtypes = {
                    'SYMBOL': 'category', 'TRAN_DATE': str, 'OPEN_PRICE': 'float32',
                    'HIGH_PRICE': 'float32', 'LOW_PRICE': 'float32', 'PREV_CL_PR': 'float32',
                    'CLOSE_PRICE': 'float32', 'NET_TRDVAL': 'float32', 'NET_TRDQTY': 'float32',
                    'TRADES': 'float32', 'HI_52_WK': 'float32', 'LO_52_WK': 'float32',
                    'HI_DIFF': 'float32', 'LO_DIFF': 'float32'
                }

                # Load data from Parquet
                self.df = pd.read_parquet(
                    self.parquet_path,
                    columns=list(dtypes.keys())
                )

                self.df['TRAN_DATE'] = pd.to_datetime(self.df['TRAN_DATE'], errors='coerce')
                self.df.dropna(subset=['TRAN_DATE'], inplace=True)

                self.first_date = self.df['TRAN_DATE'].min()
                self.last_date = self.df['TRAN_DATE'].max()

                # Filter data for the last two years
                scale_data = False
                training_days = 365 * 2 # Years of data to consider for scaling
                if scale_data:
                    self.scaled_date = self.last_date - timedelta(days=training_days)
                    logging.info(f"Processing transaction dates and filtering data from: {self.scaled_date.strftime('%Y-%m-%d')}...")
                    mask = self.df['TRAN_DATE'] >= self.scaled_date
                    latest_data = self.df[mask].copy()
                    self.df = latest_data
                else:
                    logging.info(f"Processing transaction dates and filtering data from: {self.first_date.strftime('%Y-%m-%d')}...")
        
                # Sort data by SYMBOL and TRAN_DATE
                self.df = self.df.sort_values(by=['SYMBOL', 'TRAN_DATE']).reset_index(drop=True)

                # Create feature set from dataframe
                logging.info(f"Extract file not found. Generating features and labels for model training...")
                self.create_features()
                print("Generating target prices.")
                self.create_target()
                # Drop null values from dataframe and drop index
                self.df.dropna(inplace=True)
                print("Saving extract file.")
                self.save_extract_file()
                print("Extract data generated and saved successfully.")
                
                logging.info("Data loading and preprocessing completed successfully...")
                # Show correlation analysis results
                #self.run_correlation_analysis(self.df)
                # Log the final DataFrame statistics
                logging.info(f"Data loaded with {len(self.df)} rows and {len(self.df.columns)} columns.")
                # Show all the column from dataframe
                logging.info(f"Data columns: {', '.join(self.df.columns)}")
                logging.info(f"Data contains {self.df['SYMBOL'].nunique()} unique symbols.")
                logging.info(f"Data last date: {self.last_date.strftime('%Y-%m-%d')}, Data first date: {self.first_date.strftime('%Y-%m-%d')}")
                logging.info(f"Data contains {self.df['TRAN_DATE'].nunique()} unique transaction dates.")  
                # Show the first and lst few rows of the DataFrame for verification
                logging.info(f"First 5 rows of the DataFrame:\n{self.df.head()}")
                logging.info(f"Last 5 rows of the DataFrame:\n{self.df.tail()}")

            except FileNotFoundError:
                logging.error(f"Error: Input file not found at {self.input_file}. Please ensure 'nse_processed_data.csv' exists.")
                sys.exit(1) # Exit if the critical input file is missing
            except Exception as e:
                logging.error(f"Failed to load and preprocess data: {e}")
                sys.exit(1) # Exit on general data processing failure

    def apply_corporate_actions(self, df):
        """
        Adjusts historical prices (OPEN_PRICE,HIGH_PRICE,LOW_PRICE,CLOSE_PRICE,PREV_CL_PR,HI_52_WK,LO_52_WK)
        in the DataFrame based on corporate actions. Returns the adjusted DataFrame and the count of symbols for which actions were applied.
        """
        adjusted_symbols_count = 0

        # Load corporate actions from JSON file
        try:
            if not self.corpact_file.exists():
                self.logger.info(f"Corporate actions file not found: {self.corpact_file}. Skipping corporate actions.")
                return df, 0 # Return original DataFrame if file not found
            
            with open(self.corpact_file, 'r') as f:
                self.corporate_actions = json.load(f)
            
            if not self.corporate_actions:
                self.logger.info("No corporate actions found in file.")
                return df, 0 # Return original DataFrame if no actions are defined
        except FileNotFoundError:
            self.logger.error(f"Corporate actions file not found: {self.corpact_file}")
            return df, 0
        except json.JSONDecodeError:
            self.logger.error(f"Error decoding corporate actions JSON from {self.corpact_file}. Check JSON format.")
            return df, 0
        except Exception as e:
            self.logger.error(f"Failed to load corporate actions: {str(e)}")
            return df, 0
        
        df_adjusted = df.copy()
        df_adjusted['TRAN_DATE'] = pd.to_datetime(df_adjusted['TRAN_DATE']) # Ensure datetime type

        print("Applying corporate actions...", end='', flush=True)
        # Sort corporate actions by date in descending order to apply older actions first
        # This ensures that if multiple actions for the same symbol happen, the adjustments cascade correctly
        self.corporate_actions.sort(key=lambda x: datetime.strptime(x['date'], '%Y-%m-%d'), reverse=True)

        for action in tqdm(self.corporate_actions, desc="Applying corporate actions"):
            symbol = action['symbol']
            action_date = datetime.strptime(action['date'], '%Y-%m-%d')
            ratio_str = action['ratio']

            # Corrected part: Check if 'type' key exists and its value is not None
            if 'type' not in action or action['type'] is None:
                self.logger.warning(f"Corporate action for {symbol} on {action_date.strftime('%Y-%m-%d')} is missing 'type' or 'type' is null. Skipping.")
                continue
            
            action_type = action['type'].lower() # Now it's safe to call .lower()

            # Parse ratio (e.g., "1:2" for split, "1:1" for bonus)
            try:
                if action_type == "split":
                    old_shares, new_shares = map(int, ratio_str.split(':'))
                    adjustment_factor = new_shares / old_shares
                elif action_type == "bonus":
                    bonus_shares, held_shares = map(int, ratio_str.split(':'))
                    adjustment_factor = (held_shares + bonus_shares) / held_shares
                else:
                    self.logger.warning(f"Unsupported corporate action type: {action_type} for {symbol} on {action_date.strftime('%Y-%m-%d')}. Skipping.")
                    continue
            except ValueError:
                self.logger.error(f"Invalid ratio format for {symbol} {action_type}: {ratio_str}. Skipping.")
                continue

            # Select data for the specific symbol and dates *before* the corporate action
            # Prices before the action date need to be adjusted
            mask = (df_adjusted['SYMBOL'] == symbol) & (df_adjusted['TRAN_DATE'] < action_date)

            if df_adjusted[mask].empty:
                continue

            # Apply adjustment to relevant price columns
            # HI_52_WK and LO_52_WK are adjusted here as well
            price_columns = ['OPEN_PRICE', 'HIGH_PRICE', 'LOW_PRICE', 'CLOSE_PRICE', 'PREV_CL_PR', 'HI_52_WK', 'LO_52_WK']
            
            for col in price_columns:
                if col in df_adjusted.columns:
                    # Apply adjustment. Rounding to 2 decimal places for prices.
                    df_adjusted.loc[mask, col] = (df_adjusted.loc[mask, col] / adjustment_factor).round(2).astype('float32')
            
            adjusted_symbols_count += 1
            self.logger.info(f"Applied {action_type} action for {symbol} on {action_date.strftime('%Y-%m-%d')}. Factor: {adjustment_factor:.2f}")
        
        print("Done")
        return df_adjusted, adjusted_symbols_count

    def validate_schema(self, df):
        """Validate and convert DataFrame against schema with safe type conversion"""
        try:
            validated_df = pd.DataFrame()
            
            for col, dtype in self.schema.items():
                if col in df.columns:
                    try:
                        # Handle integer columns safely
                        if dtype in ['int32', 'int64']:
                            # First convert to float to handle NaN values
                            validated_df[col] = pd.to_numeric(df[col], errors='coerce')
                            # Then convert to integer, replacing NaN with 0
                            validated_df[col] = validated_df[col].fillna(0).astype(dtype)
                        else:
                            validated_df[col] = df[col].astype(dtype)
                    except Exception as e:
                        self.logger.warning(f"Type conversion failed for {col} to {dtype}: {str(e)}. Keeping original type.")
                        # If conversion fails, keep original column but ensure it's copied
                        validated_df[col] = df[col] 
            
            return validated_df
            
        except Exception as e:
            self.logger.error(f"Schema validation error: {str(e)}")
            return None

    def save_processed_data(self, df):
        """Save processed data with proper sorting and validation"""
        try:
            if df.empty:
                raise ValueError("No data to save")
            
            # Check if this is fresh processing
            is_fresh_save = not self.parquet_path.exists()
            
            # Ensure proper date format 
            df['TRAN_DATE'] = pd.to_datetime(df['TRAN_DATE'])
            
            # If existing file, merge with new data
            if not is_fresh_save:
                self.logger.info("Reading existing data for merge...")
                existing_df = pd.read_parquet(self.parquet_path)
                if not pd.api.types.is_datetime64_any_dtype(existing_df['TRAN_DATE']):
                    existing_df['TRAN_DATE'] = pd.to_datetime(existing_df['TRAN_DATE'])
                
                # Concatenate and remove duplicates
                self.logger.info("Merging with existing data...")
                df = pd.concat([existing_df, df], ignore_index=True)
                df = df.drop_duplicates(subset=['TRAN_DATE', 'SYMBOL'], keep='last')
            
            # Convert SYMBOL to category for efficient sorting
            df['SYMBOL'] = df['SYMBOL'].astype('category')
            
            # Sort the final dataset
            self.logger.info("Sorting output data...")
            df = df.sort_values(
                by=['TRAN_DATE', 'SYMBOL'],
                ascending=[True, True]
            ).reset_index(drop=True)
            
            print("\nSaving processed data...", end='', flush=True)
            with tqdm(total=1, desc="Saving outputs") as pbar:
                # Save Parquet
                df.to_parquet(self.parquet_path, index=False)
                pbar.update(1)
            
            # Log summary
            self.logger.info(
                f"{'Created new' if is_fresh_save else 'Updated'} output files\n"
                f"Total records: {len(df):,}\n"
                f"Date range: {df['TRAN_DATE'].min():%Y-%m-%d} to "
                f"{df['TRAN_DATE'].max():%Y-%m-%d}\n"
                f"Files:\n{self.parquet_path}"
            )
            
            # Print summary
            print(f"\nData {'created' if is_fresh_save else 'updated'} successfully")
            print(f"Records: {len(df):,}")
            print(f"Date range: {df['TRAN_DATE'].min():%Y-%m-%d} to "
                f"{df['TRAN_DATE'].max():%Y-%m-%d}")
            
            # Return final DataFrame
            return df
            
        except Exception as e:
            self.logger.error(f"Failed to save output: {str(e)}")
            raise
            return None

    def process_all_files(self):
        """
        Process files with simplified progress display,
        and optionally apply corporate actions for full processing runs.
        """
        self.metrics['start_time'] = time.time()
        
        # Determine if this is a full processing run
        latest_date = self.get_latest_processed_date()
        is_full_processing = (latest_date is None)
        
        # Find and sort ZIP files
        zip_files = []
        for zip_path in self.downloads_dir.rglob("*.zip"):
            try:
                date_str = self.extract_date_from_filename(zip_path.name)
                if date_str:
                    file_date = pd.to_datetime(date_str)
                    if latest_date is None or file_date > latest_date:
                        zip_files.append((file_date, zip_path))
            except Exception:
                # Log the error but continue if a single file has issues
                self.logger.warning(f"Could not extract date or process filename for {zip_path.name}. Skipping.")
                continue
        
        zip_files.sort()
        zip_files = [p for _, p in zip_files]
        
        if not zip_files:
            print("No new files to process.")
            if is_full_processing: # If no files and it's a full run, maybe apply corporate actions to existing data
                print("However, performing corporate action check on existing data (if any).")
                try:
                    # Load existing data if available to apply corporate actions
                    if self.parquet_path.exists():
                        final_df = pd.read_parquet(self.parquet_path)
                        if not pd.api.types.is_datetime64_any_dtype(final_df['TRAN_DATE']):
                            final_df['TRAN_DATE'] = pd.to_datetime(final_df['TRAN_DATE'])
                        final_df, applied_count = self.apply_corporate_actions(final_df)
                        self.metrics['corporate_actions_applied_count'] = applied_count
                        if applied_count > 0:
                            print(f"\nApplied {applied_count} corporate actions to existing data.")
                            self.save_processed_data(final_df) # Save updated data
                        else:
                            print("No new corporate actions applied to existing data.")
                    else:
                        print("No existing data found for corporate action application.")
                except Exception as e:
                    self.logger.error(f"Error during corporate action application on existing data: {str(e)}")
                    print(f"Error applying corporate actions to existing data: {str(e)}")
            # Check if extract file exists, if not then try to load data
            if not self.extract_file.exists():
                print("No extract file found, generating now.")
                try:
                    self.load_data()
                except Exception as e:
                    self.logger.error(f"Error loading extract data: {str(e)}")
                    print(f"Error loading extract data: {str(e)}")
            return
            
        print(f"Processing {len(zip_files)} files...", end='', flush=True)
        
        # Process in batches
        batch_size = 10
        processed_batches = []

        with tqdm(total=len(zip_files), desc="Processing ZIP files", bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt}') as pbar:
            for i in range(0, len(zip_files), batch_size):
                batch = zip_files[i:i + batch_size]
                batch_data = []

                for zip_file in batch:
                    try:
                        df = self.process_zip_file(zip_file)
                        if df is not None and not df.empty:
                            batch_data.append(df)
                            self.metrics['files_processed'] += 1
                    except Exception as e:
                        logging.warning(f"Failed to process {zip_file}: {e}")
                    finally:
                        pbar.update(1)

                if batch_data:
                    batch_df = pd.concat(batch_data, ignore_index=True)
                    processed_batches.append(batch_df)
                    gc.collect()
        
        if processed_batches:
            print("\nMerging data...", end='', flush=True)
            final_df = pd.concat(processed_batches, ignore_index=True)
            # Sorting will happen inside save_processed_data as well if merging with existing.
            # However, for consistency before corporate actions, we can sort here too.
            final_df = final_df.sort_values(['TRAN_DATE', 'SYMBOL'])
            print("Done")

            # Apply corporate actions if it's a full processing run
            if is_full_processing:
                print("It's a full processing run. Checking for corporate actions...")
                final_df, applied_count = self.apply_corporate_actions(final_df)
                self.metrics['corporate_actions_applied_count'] = applied_count
                if applied_count > 0:
                    print(f"Successfully applied {applied_count} corporate actions.")
                    self.logger.info(f"Applied {applied_count} corporate actions during full processing.")
                else:
                    print("No corporate actions found or applied.")
                    self.logger.info("No corporate actions found or applied during full processing.")

            # Save processed data
            df = self.save_processed_data(final_df)

            # Reload data to ensure consistency
            self.load_data()

            """
            # Update self.df and create features/target
            if df is not None:
                self.df = df
                self.create_features()
                self.create_target()
                self.save_extract_file()
            else:
                print("No data to save after processing.")
                self.logger.warning("No data to save after processing.")
            """
            
        duration = time.time() - self.metrics['start_time']
        self.metrics['duration'] = duration
        self.metrics['end_datetime'] = datetime.now()
        start_dt = datetime.fromtimestamp(self.metrics['start_time']).strftime('%Y-%m-%d %H:%M:%S')

        print(f"\n--- Processing Summary ---")
        print(f"Started: {start_dt}")
        print(f"Ended: {self.metrics['end_datetime'].strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"Duration: {self.metrics['duration']:.1f}s")
        print(f"Files Found: {len(zip_files)}") # This is the count of files considered for processing
        print(f"Files Processed (new/updated): {self.metrics['files_processed']}")
        print(f"Files Failed: {self.metrics['files_failed']}")
        print(f"Rows Processed: {self.metrics['rows_processed']:,}")
        if is_full_processing:
            print(f"Corporate Actions Applied: {self.metrics['corporate_actions_applied_count']}")
        print(f"--------------------------")


if __name__ == "__main__":
    try:
        processor = NSEDataProcessor()
        processor.process_all_files()
    except KeyboardInterrupt:
        print("\nProcessing interrupted by user.")
    except Exception as e:
        print(f"\nAn unhandled error occurred: {str(e)}")
        logging.exception("An unhandled error occurred in main execution block.")
        sys.exit(1)