Skip to content

API Reference

This page contains the auto-generated API reference for aynse.

NSE Module

The main module for fetching NSE data.

Historical Data

aynse.nse.history.stock_raw module-attribute

stock_raw = stock_raw

aynse.nse.history.stock_df

stock_df(symbol, from_date, to_date, series='EQ')

aynse.nse.history.stock_csv

stock_csv(symbol, from_date, to_date, series='EQ', output='', show_progress=True)

aynse.nse.history.derivatives_raw module-attribute

derivatives_raw = derivatives_raw

aynse.nse.history.derivatives_df

derivatives_df(symbol, from_date, to_date, expiry_date, instrument_type, strike_price=None, option_type=None)

aynse.nse.history.derivatives_csv

derivatives_csv(symbol, from_date, to_date, expiry_date, instrument_type, strike_price=None, option_type=None, output='', show_progress=False)

aynse.nse.history.index_raw module-attribute

index_raw = index_raw

aynse.nse.history.index_df

index_df(symbol, from_date, to_date)

aynse.nse.history.index_csv

index_csv(symbol, from_date, to_date, output='', show_progress=False)

aynse.nse.history.index_pe_raw module-attribute

index_pe_raw = index_pe_raw

aynse.nse.history.index_pe_df

index_pe_df(symbol, from_date, to_date)

aynse.nse.history.set_stock_history_backend

set_stock_history_backend(backend: str) -> None

Set the stock history backend globally for stock_raw/stock_df/stock_csv.

Backends: - auto: try NSE historical API, then fallback to bhavcopy reconstruction - nse: use NSE historical API only - bhavcopy: use bhavcopy reconstruction only - custom: use user-registered provider only

aynse.nse.history.get_stock_history_backend

get_stock_history_backend() -> str

Return currently configured global stock history backend.

aynse.nse.history.register_stock_history_provider

register_stock_history_provider(provider: Optional[StockHistoryProvider]) -> None

Register a custom stock history provider used when backend is 'custom'.

The provider signature must be

provider(symbol: str, from_date: date, to_date: date, series: str) -> list[dict]

Archives

aynse.nse.archives.bhavcopy_raw module-attribute

bhavcopy_raw = bhavcopy_raw

aynse.nse.archives.bhavcopy_save module-attribute

bhavcopy_save = bhavcopy_save

aynse.nse.archives.full_bhavcopy_raw module-attribute

full_bhavcopy_raw = full_bhavcopy_raw

aynse.nse.archives.full_bhavcopy_save module-attribute

full_bhavcopy_save = full_bhavcopy_save

aynse.nse.archives.bhavcopy_fo_raw module-attribute

bhavcopy_fo_raw = bhavcopy_fo_raw

aynse.nse.archives.bhavcopy_fo_save module-attribute

bhavcopy_fo_save = bhavcopy_fo_save

aynse.nse.archives.bhavcopy_index_raw module-attribute

bhavcopy_index_raw = bhavcopy_index_raw

aynse.nse.archives.bhavcopy_index_save module-attribute

bhavcopy_index_save = bhavcopy_index_save

aynse.nse.archives.bulk_deals_raw module-attribute

bulk_deals_raw = bulk_deals_raw

aynse.nse.archives.bulk_deals_save module-attribute

bulk_deals_save = bulk_deals_save

aynse.nse.archives.index_constituent_raw module-attribute

index_constituent_raw = index_constituent_raw

aynse.nse.archives.index_constituent_save module-attribute

index_constituent_save = index_constituent_save

aynse.nse.archives.index_constituent_save_all module-attribute

index_constituent_save_all = index_constituent_save_all

aynse.nse.archives.expiry_dates

expiry_dates(dt: date, instrument_type: str = '', symbol: str = '', contracts: int = 0, months_ahead: int = 6) -> list

Algorithmically calculate expiry dates based on NSE rules.

PARAMETER DESCRIPTION
dt

Reference date for calculation

TYPE: date

instrument_type

Type of derivative instrument (e.g., "FUTIDX", "OPTIDX", "FUTSTK", "OPTSTK")

TYPE: str DEFAULT: ''

symbol

Symbol name (used to determine contract cycle)

TYPE: str DEFAULT: ''

contracts

Minimum open interest filter (deprecated in algorithmic approach)

TYPE: int DEFAULT: 0

months_ahead

Number of months ahead to calculate expiries for

TYPE: int DEFAULT: 6

RETURNS DESCRIPTION
list

List of expiry dates

Live Data

aynse.nse.live.NSELive

NSELive()

Initialize NSELive with centralized HTTP client

stock_quote

stock_quote(symbol)

stock_quote_fno

stock_quote_fno(symbol)

trade_info

trade_info(symbol)

market_status

market_status()

chart_data

chart_data(symbol, indices=False, flag='1D')

Fetch intraday/period graph data.

NSE moved live chart data to NextApi: - Equities: /api/NextApi/apiClient/GetQuoteApi?functionName=getSymbolChartData symbol must be EQN (library adds suffix automatically) uses days parameter (e.g., 1D, 5D, 3M) - Indices: /api/NextApi/apiClient?functionName=getGraphChart uses type and flag (e.g., 1D, 5D, 6M)

We normalize responses so callers still receive a dict with grapthData.

tick_data

tick_data(symbol, indices=False, flag='1D')

market_turnover

market_turnover()

eq_derivative_turnover

eq_derivative_turnover(type='allcontracts')

all_indices

all_indices()

live_index

live_index(symbol='NIFTY 50')

index_option_chain

index_option_chain(symbol='NIFTY')

equities_option_chain

equities_option_chain(symbol)

currency_option_chain

currency_option_chain(symbol='USDINR')

live_fno

live_fno()

pre_open_market

pre_open_market(key='NIFTY')

holiday_list

holiday_list()

corporate_announcements

corporate_announcements(segment='equities', from_date=None, to_date=None, symbol=None)

This function returns the corporate annoucements (https://www.nseindia.com/companies-listing/corporate-filings-announcements)

bulk_equities_option_chain

bulk_equities_option_chain(symbols, max_workers=3) -> Dict[str, Any]

Fetch option chains for multiple equity symbols concurrently.

PARAMETER DESCRIPTION
symbols

List of equity symbols

TYPE: list

max_workers

Maximum number of concurrent requests

TYPE: int DEFAULT: 3

RETURNS DESCRIPTION
dict

Dictionary with symbol as key and option chain data as value

TYPE: Dict[str, Any]

get_options_around_date

get_options_around_date(symbol, target_date, days_before=5, days_after=5) -> Dict[str, Any]

Get option chain data and analyze it in context of a target date (like earnings).

PARAMETER DESCRIPTION
symbol

Equity symbol

TYPE: str

target_date

Target date (e.g., earnings date)

TYPE: date

days_before

Days before target date to consider

TYPE: int DEFAULT: 5

days_after

Days after target date to consider

TYPE: int DEFAULT: 5

RETURNS DESCRIPTION
dict

Analysis of options around the target date

TYPE: Dict[str, Any]

analyze_earnings_options

analyze_earnings_options(symbols_and_dates, max_workers=3) -> Dict[str, Any]

Analyze option chains for multiple stocks around their earnings dates.

PARAMETER DESCRIPTION
symbols_and_dates

List of tuples (symbol, earnings_date)

TYPE: list

max_workers

Maximum concurrent requests

TYPE: int DEFAULT: 3

RETURNS DESCRIPTION
dict

Analysis results for each symbol

TYPE: Dict[str, Any]

Holidays Module

aynse.holidays

Trading holidays for NSE (National Stock Exchange of India).

This module provides a comprehensive list of trading holidays and utilities for checking if a given date is a trading day.

The holiday list is based on the Zipline calendar library and is updated with official NSE holiday announcements.

Reference

https://github.com/quantopian/trading_calendars/blob/master/trading_calendars/exchange_calendar_xbom.py

holidays

holidays(year: Optional[int] = None, month: Optional[int] = None) -> List[date]

Get list of NSE trading holidays.

PARAMETER DESCRIPTION
year

Filter by year (optional). If None, returns all holidays.

TYPE: Optional[int] DEFAULT: None

month

Filter by month (optional, requires year to be set).

TYPE: Optional[int] DEFAULT: None

RETURNS DESCRIPTION
List[date]

List of holiday dates, sorted in chronological order.

Examples:

>>> # Get all holidays
>>> all_holidays = holidays()
>>> # Get holidays for 2024
>>> holidays_2024 = holidays(year=2024)
>>> # Get holidays for January 2024
>>> jan_2024 = holidays(year=2024, month=1)

is_holiday

is_holiday(dt: date) -> bool

Check if a date is an NSE trading holiday.

PARAMETER DESCRIPTION
dt

Date to check

TYPE: date

RETURNS DESCRIPTION
bool

True if the date is a holiday, False otherwise

Example

from datetime import date is_holiday(date(2024, 1, 26)) # Republic Day True

is_trading_day

is_trading_day(dt: date) -> bool

Check if a date is a trading day.

A trading day is a weekday (Monday-Friday) that is not a holiday.

PARAMETER DESCRIPTION
dt

Date to check

TYPE: date

RETURNS DESCRIPTION
bool

True if the date is a trading day, False otherwise

Example

from datetime import date is_trading_day(date(2024, 1, 15)) # Monday, not a holiday True is_trading_day(date(2024, 1, 13)) # Saturday False

get_trading_days

get_trading_days(from_date: date, to_date: date) -> List[date]

Get all trading days in a date range.

PARAMETER DESCRIPTION
from_date

Start date (inclusive)

TYPE: date

to_date

End date (inclusive)

TYPE: date

RETURNS DESCRIPTION
List[date]

List of trading days in the range, sorted chronologically

Example

from datetime import date trading_days = get_trading_days(date(2024, 1, 1), date(2024, 1, 31))

count_trading_days

count_trading_days(from_date: date, to_date: date) -> int

Count the number of trading days in a date range.

PARAMETER DESCRIPTION
from_date

Start date (inclusive)

TYPE: date

to_date

End date (inclusive)

TYPE: date

RETURNS DESCRIPTION
int

Number of trading days in the range

RBI Module

aynse.rbi.RBI

RBI()

Client for fetching RBI (Reserve Bank of India) data.

Provides access to policy rate archives and other monetary policy information from the RBI website.

Example

rbi = RBI() rates = rbi.policy_rate_archive(n=10) print(rates[0])

Initialize RBI client with a requests session.

current_rates

current_rates() -> None

Get current RBI policy rates.

.. deprecated:: This function is broken due to website changes. Use policy_rate_archive() instead.

RAISES DESCRIPTION
DeprecationWarning

Always raised as this method is deprecated

policy_rate_archive

policy_rate_archive(n: int = 10) -> List[Dict[str, Any]]

Fetch historical policy rates from RBI website.

PARAMETER DESCRIPTION
n

Number of past rate records to fetch (default: 10)

TYPE: int DEFAULT: 10

RETURNS DESCRIPTION
List[Dict[str, Any]]

List of dictionaries containing policy rate data.

List[Dict[str, Any]]

Keys are normalized to lowercase with underscores.

Example

rbi = RBI() rates = rbi.policy_rate_archive(n=5) for rate in rates: ... print(rate.get('date'), rate.get('repo_rate'))

aynse.rbi.policy_rate_archive

policy_rate_archive(n: int = 10) -> List[Dict[str, Any]]

Fetch historical policy rates from RBI website.

Scrapes the RBI policy rate archive page to extract historical policy rate data including repo rate, reverse repo rate, etc.

PARAMETER DESCRIPTION
n

Number of past rate records to fetch (default: 10). Higher values will fetch more historical data.

TYPE: int DEFAULT: 10

RETURNS DESCRIPTION
List[Dict[str, Any]]

List of dictionaries containing policy rate data.

List[Dict[str, Any]]

Each dictionary represents one rate announcement.

List[Dict[str, Any]]

Column names are normalized to lowercase with underscores.

RAISES DESCRIPTION
RequestException

If the HTTP request fails

ValueError

If no data table is found on the page

Example

rates = policy_rate_archive(n=5) for rate in rates: ... print(f"Date: {rate.get('date')}, Repo: {rate.get('repo_rate')}")

HTTP Client

The library includes resilient HTTP clients with retry logic, rate limiting, and circuit breakers.

NSEHttpClient

from aynse.nse.http_client import NSEHttpClient

client = NSEHttpClient(
    base_url="https://www.nseindia.com",
    timeout=15.0,
    max_connections=20,
    rate_limit_per_sec=10
)

# Make requests
response = client.get("/api/marketStatus")
data = client.get_json("/api/marketStatus")

Connection Pool

from aynse.nse.connection_pool import get_connection_pool

pool = get_connection_pool()
client = pool.get_client("https://www.nseindia.com")

# The pool manages client lifecycle and reuse
data = client.get_json("/api/marketStatus")

# Get pool statistics
stats = pool.get_pool_stats()
print(f"Active clients: {stats['sync_clients']}")

Request Batching

aynse.nse.request_batcher.RequestBatcher

RequestBatcher(max_batch_size: int = 10, max_concurrent_batches: int = 3, timeout: float = 30.0, strategy: BatchStrategy = ADAPTIVE)

Efficient request batching system for NSE API calls.

Supports both synchronous and asynchronous batching with different strategies.

Initialize the request batcher.

PARAMETER DESCRIPTION
max_batch_size

Maximum requests per batch

TYPE: int DEFAULT: 10

max_concurrent_batches

Maximum concurrent batch processors

TYPE: int DEFAULT: 3

timeout

Request timeout in seconds

TYPE: float DEFAULT: 30.0

strategy

Batching strategy to use

TYPE: BatchStrategy DEFAULT: ADAPTIVE

abatch_requests async

abatch_requests(requests: List[Dict[str, Any]], request_coro_func: Callable[..., Future[Any]], max_concurrency: Optional[int] = None, **kwargs: Any) -> List[BatchResult]

Async batch processing with bounded concurrency.

PARAMETER DESCRIPTION
requests

List of request parameter dicts

TYPE: List[Dict[str, Any]]

request_coro_func

Async function to call per request

TYPE: Callable[..., Future[Any]]

max_concurrency

Optional concurrency cap (defaults to max_concurrent_batches * max_batch_size)

TYPE: Optional[int] DEFAULT: None

**kwargs

Extra kwargs for request function

TYPE: Any DEFAULT: {}

batch_requests

batch_requests(requests: List[Dict[str, Any]], request_func: Callable[..., Any], **kwargs: Any) -> List[BatchResult]

Process multiple requests in batches.

PARAMETER DESCRIPTION
requests

List of request parameters

TYPE: List[Dict[str, Any]]

request_func

Function to process individual requests

TYPE: Callable[..., Any]

**kwargs

Additional arguments for request_func

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
List[BatchResult]

List of BatchResult objects

get_stats

get_stats() -> BatchStats

Get current batch processing statistics

reset_stats

reset_stats()

Reset batch processing statistics

aynse.nse.request_batcher.BatchStrategy

Bases: Enum

Different strategies for batching requests

Streaming Processor

aynse.nse.streaming_processor.StreamingProcessor

StreamingProcessor(config: Optional[StreamConfig] = None)

Efficient streaming processor for large NSE datasets.

Processes data in chunks to minimize memory usage and handle large files that wouldn't fit in memory.

Initialize streaming processor with configuration

process_csv_file

process_csv_file(file_path: str, processor_func: Callable[[List[Dict[str, Any]]], Any], skip_header: bool = True) -> Any

Process a CSV file in streaming chunks.

PARAMETER DESCRIPTION
file_path

Path to CSV file

TYPE: str

processor_func

Function to process each chunk of data

TYPE: Callable[[List[Dict[str, Any]]], Any]

skip_header

Whether to skip the first row (header)

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
Any

Result of processing all chunks

process_csv_string

process_csv_string(csv_data: str, processor_func: Callable[[List[Dict[str, Any]]], Any], skip_header: bool = True) -> Any

Process CSV data from string in streaming chunks.

PARAMETER DESCRIPTION
csv_data

CSV data as string

TYPE: str

processor_func

Function to process each chunk

TYPE: Callable[[List[Dict[str, Any]]], Any]

skip_header

Whether to skip the first row

TYPE: bool DEFAULT: True

RETURNS DESCRIPTION
Any

Result of processing all chunks

process_json_file

process_json_file(file_path: str, processor_func: Callable[[List[Dict[str, Any]]], Any]) -> Any

Process a JSON file in streaming chunks.

PARAMETER DESCRIPTION
file_path

Path to JSON file (assumed to be array of objects)

TYPE: str

processor_func

Function to process each chunk

TYPE: Callable[[List[Dict[str, Any]]], Any]

RETURNS DESCRIPTION
Any

Result of processing all chunks

process_zip_file

process_zip_file(zip_path: str, processor_func: Callable[[List[Dict[str, Any]]], Any], csv_filename: Optional[str] = None) -> Any

Process CSV data from within a ZIP file in streaming chunks.

PARAMETER DESCRIPTION
zip_path

Path to ZIP file

TYPE: str

processor_func

Function to process each chunk

TYPE: Callable[[List[Dict[str, Any]]], Any]

csv_filename

Specific CSV file within ZIP (if None, uses first CSV found)

TYPE: Optional[str] DEFAULT: None

RETURNS DESCRIPTION
Any

Result of processing all chunks

aynse.nse.streaming_processor.StreamConfig dataclass

StreamConfig(chunk_size: int = 1000, max_memory_mb: int = 100, buffer_size: int = 8192, encoding: str = 'utf-8')

Configuration for streaming data processing

Utilities

aynse.util.break_dates

break_dates(from_date: date, to_date: date, max_days_per_chunk: int = 31) -> List[Tuple[date, date]]

Break a date range into monthly chunks for efficient API calls.

The NSE API typically limits requests to one month at a time, so this function splits larger date ranges into monthly chunks.

PARAMETER DESCRIPTION
from_date

Start date of the range

TYPE: date

to_date

End date of the range

TYPE: date

max_days_per_chunk

Maximum days per chunk (default: 31)

TYPE: int DEFAULT: 31

RETURNS DESCRIPTION
List[Tuple[date, date]]

List of (start_date, end_date) tuples for each chunk

RAISES DESCRIPTION
ValueError

If from_date is after to_date

Example

from datetime import date break_dates(date(2024, 1, 15), date(2024, 3, 20)) [(date(2024, 1, 15), date(2024, 1, 31)), (date(2024, 2, 1), date(2024, 2, 29)), (date(2024, 3, 1), date(2024, 3, 20))]

aynse.util.cached

cached(app_name: str, max_age_seconds: Optional[int] = None) -> Callable[[F], F]

Decorator for caching function results to disk.

Results are stored as gzip-compressed pickle files in a user-specific cache directory. The cache key is generated from the function arguments.

PARAMETER DESCRIPTION
app_name

Name of the application (used for cache directory)

TYPE: str

max_age_seconds

Optional TTL for cache entries. If set, expired cache files are ignored and rebuilt.

TYPE: Optional[int] DEFAULT: None

RETURNS DESCRIPTION
Callable[[F], F]

Decorator function

Environment Variables

J_CACHE_DIR: Override default cache directory

Example

@cached("myapp") def expensive_function(symbol, date): # ... expensive API call ... return data

aynse.util.live_cache

live_cache(func: F) -> F

Decorator for time-based caching of live data functions.

Caches the result of a function for a configurable timeout period. This is useful for live market data where we want to avoid hitting the API too frequently but still get reasonably fresh data.

The decorated class must have a time_out attribute specifying the cache duration in seconds.

PARAMETER DESCRIPTION
func

Method to cache (must be an instance method)

TYPE: F

RETURNS DESCRIPTION
F

Wrapped method with time-based caching

Example

class LiveData: time_out = 5 # Cache for 5 seconds

@live_cache
def get_quote(self, symbol):
    # ... expensive API call ...
    return quote

aynse.util.is_trading_day

is_trading_day(dt: date, holidays_list: Optional[List[date]] = None) -> bool

Check if a given date is a trading day.

A trading day is a weekday (Monday-Friday) that is not a holiday.

PARAMETER DESCRIPTION
dt

Date to check

TYPE: date

holidays_list

Optional list of holiday dates. If None, uses the holidays from the holidays module.

TYPE: Optional[List[date]] DEFAULT: None

RETURNS DESCRIPTION
bool

True if the date is a trading day, False otherwise