""" 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)