nsecorpact.py

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