# This script processes corporate action data from the NSE and generates a JSON file # containing the necessary information for dashboard integration. import pandas as pd import re import requests import os from datetime import datetime, timedelta # --- Configuration and Setup --- # Directory to save processed files (final destination) processed_dir = './processed' # Temporary directory for downloads temp_dir = './tempfiles' # Ensure both directories exist os.makedirs(processed_dir, exist_ok=True) os.makedirs(temp_dir, exist_ok=True) # Output JSON file path output_file = os.path.join(processed_dir, 'nse_dashboard_corpact.json') # Base URL for downloading corporate actions data DOWNLOAD_URL_TEMPLATE = "https://www.nseindia.com/api/corporates-corporateActions?index=equities&from_date=01-01-2020&to_date={}&csv=true" # URL for the main corporate actions page (to get session cookies) NSE_CORPORATE_ACTIONS_PAGE = "https://www.nseindia.com/companies-listing/corporate-filings-actions" # --- Helper Functions for Date and Download --- def get_last_nse_working_day(): """ Determines the last NSE working day (i.e., the most recent weekday). NSE is typically closed on Saturdays and Sundays. """ today = datetime.now() # Subtract days until a weekday is found (Monday=0, Tuesday=1, ..., Sunday=6) # If today is Saturday (5), subtract 1 to get Friday. # If today is Sunday (6), subtract 2 to get Friday. # If today is Monday-Friday, use today. while today.weekday() > 4: # 5 is Saturday, 6 is Sunday today -= timedelta(days=1) return today.strftime('%d-%m-%Y') def download_nse_corporate_actions_file(): """ Downloads the corporate actions CSV file from NSE to the tempfiles directory. It returns the path to the downloaded file in the temporary directory. Returns None if download fails. """ last_working_day = get_last_nse_working_day() download_url = DOWNLOAD_URL_TEMPLATE.format(last_working_day) # Construct the temporary file name and path temp_file_name = f"CF-CA-equities-01-01-2020-to-{last_working_day}.csv" temp_file_path = os.path.join(temp_dir, temp_file_name) # Note: We always attempt to download the latest to temp, no skipping here. # The check for existing file is done at the processed_dir level later. print(f"Attempting to download corporate actions data to temporary directory.") # Create a session to persist cookies and headers session = requests.Session() try: # Step 1: Visit the main corporate actions page to get necessary cookies print(f"Fetching cookies.") session_headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9', 'Accept-Language': 'en-US,en;q=0.9', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', } session.get(NSE_CORPORATE_ACTIONS_PAGE, headers=session_headers) # Step 2: Use the same session to download the CSV file print(f"Downloading CSV using established session.") download_headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'text/csv', # Specifically request CSV 'Accept-Language': 'en-US,en;q=0.9', 'Connection': 'keep-alive', 'Referer': NSE_CORPORATE_ACTIONS_PAGE, # Refer to the main page visited earlier 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'same-origin', 'Sec-Fetch-User': '?1', 'Upgrade-Insecure-Requests': '1', } response = session.get(download_url, headers=download_headers) response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx) # Get the raw binary content raw_bytes_content = response.content # --- BOM Preprocessing --- # Detect and remove UTF-8 BOM if present (EF BB BF) if raw_bytes_content.startswith(b'\xef\xbb\xbf'): # print("Detected and removed UTF-8 BOM from the file content.") raw_bytes_content = raw_bytes_content[3:] # --- End BOM Preprocessing --- # Decode to string for further preprocessing (only line stripping, no .strip('"')) raw_content_string = raw_bytes_content.decode('latin1', errors='ignore') # Preprocessing: Only strip leading/trailing whitespace from each line cleaned_lines = [line.strip() for line in raw_content_string.splitlines()] cleaned_content = "\n".join(cleaned_lines) # Save the cleaned content to the temporary file path with open(temp_file_path, 'w', encoding='latin1') as f: f.write(cleaned_content) print(f"Successfully downloaded and preprocessed to temporary file.") return temp_file_path except requests.exceptions.RequestException as e: print(f"Error downloading file: {e}") print("This often indicates an issue with the NSE API (e.g., authentication required, rate limiting, or changed endpoint).") print("You may still need to manually download the file if this persists.") return None # --- Existing Functions (Copied from original prompt) --- # Function to calculate adjustment factor for bonuses def calculate_bonus_adj_factor(purpose): """Extracts and calculates the adjustment factor for bonus shares.""" if "Bonus" in purpose: try: # Ratio format example: "1:4" in "Bonus 1:4" # It's safer to use regex to extract the ratio explicitly match = re.search(r"Bonus (\d+:\d+)", purpose) if match: ratio = match.group(1) bonus, base = map(int, ratio.split(":")) return f"{bonus}:{base}", "Bonus" except ValueError: print(f"Could not parse bonus ratio from: {purpose}") return None, None # Function to calculate adjustment factor for stock splits def calculate_split_adj_factor(purpose): """Extracts and calculates the adjustment factor for stock splits.""" if "Face Value Split" in purpose: try: # Use regex to find the 'From' and 'To' values # Example: "Face Value Split (Sub-Division) - From Rs 10/- Per Share To Re 1/- Per Share" match = re.search(r"From Rs (\d+(?:\.\d+)?)/?-? Per Share To (?:Re|Rs) (\d+(?:\.\d+)?)/?-? Per Share", purpose) if match: from_value_str = match.group(1) to_value_str = match.group(2) # Convert to float to handle potential decimal values from_value = float(from_value_str) to_value = float(to_value_str) # Return ratio as 'new_face_value:old_face_value' # For a split from 10 to 1, the ratio is 1:10 return f"{int(to_value)}:{int(from_value)}", "Split" else: pass # No match found except ValueError: print(f"Could not parse split values from: {purpose}") return None, None # --- Main Script Execution --- if __name__ == "__main__": temp_downloaded_file_path = None # Path of the file in tempfiles input_file = None # Final path of the file in processed_dir df = None # DataFrame for processing # Step 1: Attempt to download and preprocess the input file to tempfiles temp_downloaded_file_path = download_nse_corporate_actions_file() # Define the final target file path in the processed directory final_processed_file_name = 'Corporate_action.parquet' final_processed_file_path = os.path.join(processed_dir, final_processed_file_name) if temp_downloaded_file_path and os.path.exists(temp_downloaded_file_path): try: # Read downloaded CSV df_temp = pd.read_csv(temp_downloaded_file_path, encoding='latin1') # Save to Parquet df_temp.to_parquet(final_processed_file_path, index=False) # Remove temp CSV os.remove(temp_downloaded_file_path) # Also clean up any legacy Corporate_action.csv in the processed dir legacy_csv = os.path.join(processed_dir, 'Corporate_action.csv') if os.path.exists(legacy_csv): os.remove(legacy_csv) input_file = final_processed_file_path print("CSV downloaded file converted to Parquet and saved successfully.") except Exception as e: print(f"Error processing downloaded CSV file: {e}") input_file = final_processed_file_path if os.path.exists(final_processed_file_path) else None else: print("\nAutomatic download to temporary directory failed or file not found in tempfiles.") # Fallback: Check for the specifically named file in processed_dir print(f"Checking for '{final_processed_file_name}' in '{processed_dir}' as a fallback...") if os.path.exists(final_processed_file_path): input_file = final_processed_file_path print(f"Using existing file as input: '{input_file}'") else: # Check if there is an existing Corporate_action.csv to convert fallback_csv = os.path.join(processed_dir, 'Corporate_action.csv') if os.path.exists(fallback_csv): try: df_temp = pd.read_csv(fallback_csv, encoding='latin1') df_temp.to_parquet(final_processed_file_path, index=False) os.remove(fallback_csv) input_file = final_processed_file_path print(f"Auto-converted existing '{fallback_csv}' to Parquet.") except Exception as e: print(f"Failed to auto-convert existing CSV fallback: {e}") if not input_file: print(f"No '{final_processed_file_name}' found in '{processed_dir}'. Cannot proceed without an input file.") print(f"Please manually download the latest CSV from the following URL and place it in the '{temp_dir}' directory, then rerun the script:") print(f" {DOWNLOAD_URL_TEMPLATE.format(get_last_nse_working_day())}") exit() # Exit if no input file is available # Step 4: Use the determined input file as the dataframe source if input_file and os.path.exists(input_file): # Ensure input_file exists before reading try: df = pd.read_parquet(input_file) print(f"Successfully loaded data from '{input_file}'.") except Exception as e: print(f"Error reading input Parquet file '{input_file}': {e}") exit() # Exit if reading the Parquet fails else: print("No valid input file determined or file does not exist. Exiting without processing.") exit() # Initialize an output DataFrame output_data = [] # Only proceed if df was successfully loaded if df is not None: for _, row in df.iterrows(): symbol = row['SYMBOL'] purpose = str(row['PURPOSE']) # Ensure purpose is a string ex_date = row['EX-DATE'] ratio, corp_action = None, None # Calculate adjustment factor and corp action type if "Bonus" in purpose: ratio, corp_action = calculate_bonus_adj_factor(purpose) elif "Face Value Split" in purpose: ratio, corp_action = calculate_split_adj_factor(purpose) # Only add data if a valid corporate action and ratio were identified if ratio and corp_action: try: # Convert date to YYYY-MM-DD format, handling potential errors formatted_date = pd.to_datetime(ex_date, format='%d-%b-%Y').strftime('%Y-%m-%d') output_data.append({ "symbol": symbol, "type": corp_action, "date": formatted_date, "ratio": ratio }) except ValueError as ve: print(f"Error parsing date '{ex_date}' for symbol '{symbol}': {ve}") except Exception as e: print(f"An unexpected error occurred processing row for symbol '{symbol}': {e}") # Convert output data to a DataFrame output_df = pd.DataFrame(output_data) # Save to the target file in JSON format output_df.to_json(output_file, orient='records', indent=4) print(f"\nCorporate action data processed and saved to {output_file}") print(f"Total records processed: {len(df)}") print(f"Total valid corporate actions extracted: {len(output_df)}") else: print("Dataframe 'df' was not created due to previous errors. No processing performed.")