Track NSE Derivatives Top Contracts Easily – Copy This Ready-to-Use Code

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import pandas as pd
import time
import random
from datetime import datetime
from colorama import init, Fore, Back, Style
import os

init(autoreset=True)

class NSE:
    def __init__(self):
        self.url = "https://www.nseindia.com/api/liveEquity-derivatives?index=top20_contracts"
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'application/json, text/plain, */*',
            'Accept-Language': 'en-US,en;q=0.9',
            'Referer': 'https://www.nseindia.com/market-data/equity-derivatives-watch'
        }
        self.limit = self.get_choice()
        self.session = self.setup_session()

    def setup_session(self):
        s = requests.Session()
        retry = Retry(total=3, backoff_factor=2, status_forcelist=[401, 429, 500, 502, 503, 504])
        s.mount('https://', HTTPAdapter(max_retries=retry))
        return s

    def get_choice(self):
        print(f"{Fore.CYAN}{'='*50}")
        print(f"{Fore.CYAN}{'NSE DERIVATIVES MONITOR':^50}")
        print(f"{Fore.CYAN}{'='*50}")
        print(f"{Fore.YELLOW}Select contracts to display:")
        choices = ["5", "10", "15", "20"]
        for i, c in enumerate(choices, 1):
            print(f"{Fore.GREEN}{i}. Top {c} contracts")

        while True:
            try:
                choice = input(f"\n{Fore.WHITE}Enter choice (1-4): ").strip()
                if choice in ['1', '2', '3', '4']:
                    val = [5, 10, 15, 20][int(choice)-1]
                    print(f"{Fore.YELLOW}Selected: Top {val} contracts")
                    return val
                else:
                    print(f"{Fore.RED}Invalid choice. Enter 1-4.")
            except KeyboardInterrupt:
                print(f"\n{Fore.YELLOW}Exiting...")
                exit(0)
            except Exception as e:
                print(f"{Fore.RED}Error: {e}. Try again.")

    def fetch(self):
        try:
            r = self.session.get(self.url, headers=self.headers, timeout=10)
            r.raise_for_status()
            return r.json()
        except requests.RequestException as e:
            print(f"{Fore.RED}Fetch error: {e}")
            return None

    def format(self, data):
        if not data or 'data' not in data:
            return None, {}

        df = pd.DataFrame(data['data'])
        print(f"{Fore.CYAN}API: {len(df)} contracts, showing: {self.limit}")

        df = df.head(self.limit)
        cols = ['underlying', 'optionType', 'strikePrice', 'lastPrice', 
                'change', 'pChange', 'volume', 'openInterest', 'noOfTrades']

        df = df[cols].copy()
        df.columns = ['Stock', 'Type', 'Strike', 'LTP', 'Change', 'Change%', 'Volume', 'OI', 'Trades']

        df['LTP'] = df['LTP'].round(2)
        df['Change'] = df['Change'].round(2)
        df['Change%'] = df['Change%'].round(2)
        df['Volume'] = df['Volume'].apply(lambda x: f"{x/1e6:.1f}M")
        df['OI'] = df['OI'].apply(lambda x: f"{x/1000:.0f}K")
        df['Trades'] = df['Trades'].apply(lambda x: f"{x/1000:.0f}K")

        return df, data.get('marketStatus', {})

    def header(self, status):
        os.system('cls' if os.name == 'nt' else 'clear')

        print(f"{Back.BLUE}{Fore.WHITE}{'='*80}")
        print(f"{Back.BLUE}{Fore.WHITE}{'NSE DERIVATIVES LIVE DATA':^80}")
        print(f"{Back.BLUE}{Fore.WHITE}{'='*80}{Style.RESET_ALL}")

        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        mkt_date = status.get('marketCurrentTradingDate', 'N/A')
        mkt_status = status.get('marketStatusMessage', 'N/A')

        print(f"{Fore.CYAN}Time: {now}")
        print(f"{Fore.CYAN}Market Date: {mkt_date}")
        print(f"{Fore.CYAN}Status: {mkt_status}")
        print(f"{Fore.MAGENTA}Showing: Top {self.limit}")
        print(f"{Fore.YELLOW}{'-'*80}")

    def color(self, row):
        pct = float(row['Change%'])
        if pct > 5: return f"{Back.GREEN}{Fore.BLACK}"
        elif pct > 0: return f"{Fore.GREEN}"
        elif pct < -5: return f"{Back.RED}{Fore.WHITE}"
        elif pct < 0: return f"{Fore.RED}"
        else: return f"{Fore.WHITE}"

    def table(self, df):
        if df is None or df.empty:
            print(f"{Fore.RED}No data")
            return

        hdr = f"{Fore.YELLOW}{'Stock':^8} {'Type':^4} {'Strike':^7} {'LTP':^8} {'Change':^7} {'Change%':^8} {'Volume':^8} {'OI':^8} {'Trades':^8}"
        print(hdr)
        print(f"{Fore.YELLOW}{'-'*80}")

        for _, row in df.iterrows():
            c = self.color(row)
            print(f"{c}{row['Stock']:^8} {row['Type']:^4} {row['Strike']:^7} "
                  f"{row['LTP']:^8} {row['Change']:^7} {row['Change%']:^8} "
                  f"{row['Volume']:^8} {row['OI']:^8} {row['Trades']:^8}{Style.RESET_ALL}")

    def run(self):
        print(f"\n{Fore.GREEN}Starting NSE Monitor...")
        print(f"{Fore.YELLOW}Press Ctrl+C to stop")
        time.sleep(2)

        try:
            while True:
                data = self.fetch()
                delay = random.uniform(60, 90)

                if data:
                    df, status = self.format(data)
                    self.header(status)
                    self.table(df)
                    print(f"\n{Fore.MAGENTA}Next update: {delay:.0f}s...")
                else:
                    print(f"{Fore.RED}Failed. Retry: {delay:.0f}s...")

                time.sleep(delay)

        except KeyboardInterrupt:
            print(f"\n{Fore.YELLOW}Stopped by user")
        except Exception as e:
            print(f"{Fore.RED}Error: {e}")

def main():
    nse = NSE()
    nse.run()

if __name__ == "__main__":
    main()