Client Classes
This page documents the client classes in detail.
Base Client
Base client implementation for NEPSE API.
This module provides the foundation for both sync and async clients, including common utilities, configuration loading, and response handling.
- nepse_client.client.mask_sensitive_data(data, keys=('token', 'password', 'Authorization'))[source]
Mask sensitive fields in data for safe logging.
- Parameters:
data (
dict[str,Any]) – Dictionary containing potentially sensitive datakeys (
tuple) – Tuple of keys to mask
- Return type:
dict[str,Any]- Returns:
Dictionary with masked sensitive values
- nepse_client.client.safe_serialize(obj)[source]
Safely serialize objects for logging.
- Parameters:
obj (
Any) – Object to serialize- Return type:
Union[str,dict,list]- Returns:
Serialized representation
The base client provides common functionality for both synchronous and asynchronous implementations.
NepseClient (Synchronous)
- class nepse_client.sync_client.NepseClient[source]
Bases:
_NepseBaseSynchronous client for NEPSE API.
This client provides blocking methods to access Nepal Stock Exchange data including market status, company information, trading data, and more.
- Parameters:
logger (
Optional[Logger]) – Optional custom logger instancemask_request_data (
bool) – Whether to mask sensitive data in logs (default: True)timeout (
float) – Request timeout in seconds (default: 100.0)
Example
Basic usage:
from nepse_client import NepseClient client = NepseClient() # Get market status status = client.getMarketStatus() print(f"Market is {status['isOpen']}") # Get company details nabil = client.getCompanyDetails("NABIL") print(f"NABIL LTP: {nabil['lastTradedPrice']}")
Note
The client automatically manages authentication tokens and handles token expiration transparently.
- __init__(logger=None, mask_request_data=True, timeout=100.0)[source]
Initialize synchronous NEPSE client.
- Parameters:
logger (Logger | None)
mask_request_data (bool)
timeout (float)
- init_client(tls_verify)[source]
Initialize HTTP client with specified settings.
- Parameters:
tls_verify (
bool) – Whether to verify TLS certificates- Return type:
None
- requestGETAPI(url, include_authorization_headers=True)[source]
Make GET request to NEPSE API.
- Parameters:
url (
str) – API endpoint URLinclude_authorization_headers (
bool) – Whether to include auth headers
- Return type:
Any- Returns:
Parsed response data
- requestPOSTAPI(url, payload_generator)[source]
Make POST request to NEPSE API.
- Parameters:
url (
str) – API endpoint URLpayload_generator – Function to generate payload
- Return type:
Any- Returns:
Parsed response data
- getAuthorizationHeaders()[source]
Get headers with authorization token.
- Return type:
dict[str,str]- Returns:
Dictionary of HTTP headers
- getPOSTPayloadIDForScrips()[source]
Generate payload ID for scrip-related requests.
- Return type:
int
- getPOSTPayloadIDForFloorSheet(business_date=None)[source]
Generate payload ID for floor sheet requests.
- Parameters:
business_date (
Union[str,date,None]) – Business date (YYYY-MM-DD string or date object)- Return type:
int- Returns:
Payload ID integer
- getCompanyList()[source]
Get list of all listed companies.
- Return type:
list[dict[str,Any]]- Returns:
List of company dictionaries
Note
Results are cached internally. Subsequent calls return cached data unless cache is cleared.
- getSecurityList()[source]
Get list of all securities (non-delisted).
- Return type:
list[dict[str,Any]]- Returns:
List of security dictionaries
- getCompanyIDKeyMap(force_update=False)[source]
Get mapping of company symbols to IDs.
- Parameters:
force_update (
bool) – Force refresh of cached data- Return type:
dict[str,int]- Returns:
Dictionary mapping symbol to company ID
- getSecurityIDKeyMap(force_update=False)[source]
Get mapping of security symbols to IDs.
- Parameters:
force_update (
bool) – Force refresh of cached data- Return type:
dict[str,int]- Returns:
Dictionary mapping symbol to security ID
- getSectorScrips()[source]
Get scrips grouped by sector.
- Return type:
dict[str,list[str]]- Returns:
Dictionary mapping sector name to list of symbols
- getCompanyDetails(symbol)[source]
Get detailed information for a specific company.
- Parameters:
symbol (
str) – Company stock symbol (e.g., “NABIL”)- Return type:
dict[str,Any]- Returns:
Dictionary with company details
- Raises:
KeyError – If symbol not found
- getCompanyFinancialDetails(company_id=None)[source]
Get financial details for a specific company.
- Parameters:
company_id (str, optional) – The unique identifier for the company. If not provided, the behavior depends on the API endpoint’s default or requirements.
- Returns:
- A list of dictionaries containing financial details for the company.
Each dictionary might include keys like ‘period’, ‘financialMetrics’, ‘applicationDocumentDetailsList’, etc., as returned by the API. Returns an empty list if no data is found or an error occurs during the API request or processing.
The ‘applicationDocumentDetailsList’ within each item may be augmented with ‘fullFilePath’ and ‘fullEncryptedPath’ keys for
- Return type:
list
- getCompanyAGM(company_id=None)[source]
Get Annual General Meeting (AGM) information for a specific company.
- Parameters:
company_id (str, optional) – The unique identifier for the company. If not provided, the behavior depends on the API endpoint’s default or requirements.
- Returns:
- A list of dictionaries containing AGM details for the company.
Each dictionary might include keys like ‘meetingDate’, ‘agenda’, ‘applicationDocumentDetailsList’, etc., as returned by the API. Returns an empty list if no data is found or an error occurs during the API request or processing.
The ‘applicationDocumentDetailsList’ within each item may be augmented with a ‘fullFilePath’ key for document access.
- Return type:
list
- getCompanyDividend(company_id=None)[source]
Get dividend information for a specific company.
- Parameters:
company_id (str, optional) – The unique identifier for the company. If not provided, the behavior depends on the API endpoint’s default or requirements.
- Returns:
- A list of dictionaries containing dividend details for the company.
Each dictionary might include keys like ‘dividendType’, ‘rate’, ‘applicationDocumentDetailsList’, etc., as returned by the API. Returns an empty list if no data is found or an error occurs during the API request or processing.
The ‘applicationDocumentDetailsList’ within each item may be augmented with a ‘fullFilePath’ key for document access.
- Return type:
list
- getCompanyMarketDepth(company_id=None)[source]
Get market depth information for a specific company.
Market depth typically includes buy and sell orders at various price levels.
- Parameters:
company_id (str, optional) – The unique identifier (symbol or code) for the company/security. If not provided, the behavior depends on the API endpoint’s default or requirements.
- Returns:
- A list containing market depth data for the company.
The structure of the data depends on the API response, but it usually includes buy/sell orders with prices and volumes. Returns an empty list if no data is found or an error occurs during the API request.
- Return type:
list
- getCompanyPriceVolumeHistory(symbol, start_date=None, end_date=None)[source]
Get price and volume history for a company.
- Parameters:
symbol (
str) – Company symbolstart_date (
Union[str,date,None]) – Start date (YYYY-MM-DD or date object)end_date (
Union[str,date,None]) – End date (YYYY-MM-DD or date object)
- Return type:
dict[str,Any]- Returns:
Dictionary with paginated history data
- getDailyScripPriceGraph(symbol)[source]
Get daily price graph data for a scrip.
- Parameters:
symbol (
str) – Company symbol- Return type:
dict[str,Any]- Returns:
Graph data dictionary
- getFloorSheet(show_progress=False, paginated=False, page=None)[source]
Get floor sheet data.
- Parameters:
show_progress (
bool) – Show progress bar during downloadpaginated (
bool) – Return list of pages instead of flattened listpage (
Optional[int]) – Get specific page number (0-indexed)
- Return type:
Union[list[dict[str,Any]],list[list[dict[str,Any]]],dict[str,Any]]- Returns:
Floor sheet data (format depends on parameters)
- getFloorSheetOf(symbol, business_date=None, size=500)[source]
Get floor sheet for a specific company.
- Parameters:
symbol (
str) – Company symbolbusiness_date (
Union[str,date,None]) – Business date (YYYY-MM-DD string or date object)size (int)
- Return type:
list[dict[str,Any]]- Returns:
List of floor sheet records
- getSymbolMarketDepth(symbol)[source]
Get market depth for a symbol.
- Parameters:
symbol (
str) – Company symbol- Return type:
dict[str,Any]- Returns:
Market depth data
- getHolidayList(year=2025)[source]
Get list of market holidays for specified year.
- Return type:
list[dict[str,Any]]- Parameters:
year (int)
- getDebentureAndBondList(bond_type='debenture')[source]
Get list of debentures and bonds.
- Return type:
list[dict[str,Any]]- Parameters:
bond_type (str)
- getCompanyNewsList(page=0, page_size=100, is_strip_tags=True)[source]
Get list of company news.
- Return type:
dict[str,Any]- Parameters:
page (int)
page_size (int)
is_strip_tags (bool)
- getNewsAndAlertList(page=0, page_size=100, is_strip_tags=True)[source]
Get list of News and Alert.
- Return type:
dict[str,Any]- Parameters:
page (int)
page_size (int)
is_strip_tags (bool)
- getPressRelease(page=0, size=20)[source]
Get list of Press release.
- Parameters:
page (int, optional) – _description_. Defaults to None.
size (int, optional) – _description_. Defaults to 20.
- Returns:
Response containing press releases with full file paths
- Return type:
dict[str, Any]
- getNepseNotice(page=0, size=10)[source]
Get NEPSE Notice data.
- Parameters:
page (int, optional) – _description_. Defaults to 0.
size (int, optional) – _description_. Defaults to 10.
- Returns:
_description_
- Return type:
dict[str, Any]
- getPriceVolumeHistory(business_date=None)[source]
Get price volume history for a business date.
- Return type:
dict[str,Any]- Parameters:
business_date (str | None)
- getDailyNepseIndexGraph()[source]
Get price volume history for a business date.
- Return type:
list[Any]
- getDailySensitiveIndexGraph()[source]
Get NEPSE Daily Sensitive Index Graph.
- Returns:
_description_
- Return type:
list[Any]
- getDailyFloatIndexGraph()[source]
Get NEPSE Daily Float Index Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailySensitiveFloatIndexGraph()[source]
Get NEPSE Daily Sensitive Float Index Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyBankSubindexGraph()[source]
Get NEPSE Daily Bank Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyDevelopmentBankSubindexGraph()[source]
Get NEPSE Development Bank Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyFinanceSubindexGraph()[source]
Get NEPSE Daily Finance Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyHotelTourismSubindexGraph()[source]
Gat NEPSE Daily Hotel Tourism Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyHydroSubindexGraph()[source]
Get NEPSE Daily Hydro Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyInvestmentSubindexGraph()[source]
Get NEPSE Daily Investment Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyLifeInsuranceSubindexGraph()[source]
Get NEPSE Daily Life Insurance Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyManufacturingSubindexGraph()[source]
Get NEPSE Daily Manufacturing Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyMicrofinanceSubindexGraph()[source]
Get NEPSE Daily Microfinance Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyMutualfundSubindexGraph()[source]
Get NEPSE Daily Mutual Fund Subindex Graph Data.
- Returns:
_description_
- Return type:
list[Any]
- getDailyNonLifeInsuranceSubindexGraph()[source]
Get NEPSE Daily Non Life Insurance Subindex Graph data.
- Returns:
_description_
- Return type:
list[Any]
The synchronous client is ideal for:
Simple scripts and applications
Jupyter notebooks
Sequential data fetching
Environments without async support
Initialization:
from nepse_client import NepseClient
# Basic initialization
client = NepseClient()
# With custom settings
client = NepseClient(
logger=my_logger,
mask_request_data=True,
timeout=120.0
)
Context Manager:
# Automatic resource cleanup
with NepseClient() as client:
status = client.getMarketStatus()
companies = client.getCompanyList()
# Client automatically closed
Market Data Methods
- NepseClient.getMarketStatus()
Get current market status (open/closed).
- Return type:
dict[str,Any]- Returns:
Dictionary with market status information
- NepseClient.getSummary()
Get market summary with turnover, trades, etc.
- Return type:
dict[str,Any]- Returns:
Dictionary with market summary data
- NepseClient.getNepseIndex()
Get NEPSE index data.
- Return type:
dict[str,Any]
- NepseClient.getNepseSubIndices()
Get all NEPSE sub-indices.
- Return type:
list[dict[str,Any]]
- NepseClient.getLiveMarket()
Get live market data.
- Return type:
dict[str,Any]
- NepseClient.getPriceVolume()
Get current price and volume data for all securities.
- Return type:
list[dict[str,Any]]- Returns:
List of price/volume records
- NepseClient.getSupplyDemand()
Get supply and demand data.
- Return type:
dict[str,Any]
Company Information Methods
- NepseClient.getCompanyList()[source]
Get list of all listed companies.
- Return type:
list[dict[str,Any]]- Returns:
List of company dictionaries
Note
Results are cached internally. Subsequent calls return cached data unless cache is cleared.
- NepseClient.getSecurityList()[source]
Get list of all securities (non-delisted).
- Return type:
list[dict[str,Any]]- Returns:
List of security dictionaries
- NepseClient.getCompanyDetails(symbol)[source]
Get detailed information for a specific company.
- Parameters:
symbol (
str) – Company stock symbol (e.g., “NABIL”)- Return type:
dict[str,Any]- Returns:
Dictionary with company details
- Raises:
KeyError – If symbol not found
- NepseClient.getCompanyPriceVolumeHistory(symbol, start_date=None, end_date=None)[source]
Get price and volume history for a company.
- Parameters:
symbol (
str) – Company symbolstart_date (
Union[str,date,None]) – Start date (YYYY-MM-DD or date object)end_date (
Union[str,date,None]) – End date (YYYY-MM-DD or date object)
- Return type:
dict[str,Any]- Returns:
Dictionary with paginated history data
- NepseClient.getDailyScripPriceGraph(symbol)[source]
Get daily price graph data for a scrip.
- Parameters:
symbol (
str) – Company symbol- Return type:
dict[str,Any]- Returns:
Graph data dictionary
- NepseClient.getCompanyFinancialDetails(company_id=None)[source]
Get financial details for a specific company.
- Parameters:
company_id (str, optional) – The unique identifier for the company. If not provided, the behavior depends on the API endpoint’s default or requirements.
- Returns:
- A list of dictionaries containing financial details for the company.
Each dictionary might include keys like ‘period’, ‘financialMetrics’, ‘applicationDocumentDetailsList’, etc., as returned by the API. Returns an empty list if no data is found or an error occurs during the API request or processing.
The ‘applicationDocumentDetailsList’ within each item may be augmented with ‘fullFilePath’ and ‘fullEncryptedPath’ keys for
- Return type:
list
- NepseClient.getCompanyAGM(company_id=None)[source]
Get Annual General Meeting (AGM) information for a specific company.
- Parameters:
company_id (str, optional) – The unique identifier for the company. If not provided, the behavior depends on the API endpoint’s default or requirements.
- Returns:
- A list of dictionaries containing AGM details for the company.
Each dictionary might include keys like ‘meetingDate’, ‘agenda’, ‘applicationDocumentDetailsList’, etc., as returned by the API. Returns an empty list if no data is found or an error occurs during the API request or processing.
The ‘applicationDocumentDetailsList’ within each item may be augmented with a ‘fullFilePath’ key for document access.
- Return type:
list
- NepseClient.getCompanyDividend(company_id=None)[source]
Get dividend information for a specific company.
- Parameters:
company_id (str, optional) – The unique identifier for the company. If not provided, the behavior depends on the API endpoint’s default or requirements.
- Returns:
- A list of dictionaries containing dividend details for the company.
Each dictionary might include keys like ‘dividendType’, ‘rate’, ‘applicationDocumentDetailsList’, etc., as returned by the API. Returns an empty list if no data is found or an error occurs during the API request or processing.
The ‘applicationDocumentDetailsList’ within each item may be augmented with a ‘fullFilePath’ key for document access.
- Return type:
list
Trading Data Methods
- NepseClient.getFloorSheet(show_progress=False, paginated=False, page=None)[source]
Get floor sheet data.
- Parameters:
show_progress (
bool) – Show progress bar during downloadpaginated (
bool) – Return list of pages instead of flattened listpage (
Optional[int]) – Get specific page number (0-indexed)
- Return type:
Union[list[dict[str,Any]],list[list[dict[str,Any]]],dict[str,Any]]- Returns:
Floor sheet data (format depends on parameters)
- NepseClient.getFloorSheetOf(symbol, business_date=None, size=500)[source]
Get floor sheet for a specific company.
- Parameters:
symbol (
str) – Company symbolbusiness_date (
Union[str,date,None]) – Business date (YYYY-MM-DD string or date object)size (int)
- Return type:
list[dict[str,Any]]- Returns:
List of floor sheet records
- NepseClient.getSymbolMarketDepth(symbol)[source]
Get market depth for a symbol.
- Parameters:
symbol (
str) – Company symbol- Return type:
dict[str,Any]- Returns:
Market depth data
- NepseClient.getTradingAverage(business_date=None, nDays=180)
Get trading average data.
- Parameters:
business_date (
Optional[str]) – Business date in YYYY-MM-DD formatnDays (
int) – Number of days
- Return type:
dict[str,Any]- Returns:
Trading average data
Top Performers Methods
- NepseClient.getTopGainers()
Get list of top gaining stocks.
- Return type:
list[dict[str,Any]]- Returns:
List of top gainer records
- NepseClient.getTopLosers()
Get list of top losing stocks.
- Return type:
list[dict[str,Any]]- Returns:
List of top loser records
- NepseClient.getTopTenTradeScrips()
Get top 10 scrips by trade volume.
- Return type:
list[dict[str,Any]]
- NepseClient.getTopTenTransactionScrips()
Get top 10 scrips by transaction count.
- Return type:
list[dict[str,Any]]
- NepseClient.getTopTenTurnoverScrips()
Get top 10 scrips by turnover.
- Return type:
list[dict[str,Any]]
Utility Methods
- NepseClient.getCompanyIDKeyMap(force_update=False)[source]
Get mapping of company symbols to IDs.
- Parameters:
force_update (
bool) – Force refresh of cached data- Return type:
dict[str,int]- Returns:
Dictionary mapping symbol to company ID
- NepseClient.getSecurityIDKeyMap(force_update=False)[source]
Get mapping of security symbols to IDs.
- Parameters:
force_update (
bool) – Force refresh of cached data- Return type:
dict[str,int]- Returns:
Dictionary mapping symbol to security ID
- NepseClient.getSectorScrips()[source]
Get scrips grouped by sector.
- Return type:
dict[str,list[str]]- Returns:
Dictionary mapping sector name to list of symbols
- NepseClient.setTLSVerification(flag=False)
Enable or disable TLS certificate verification.
- Parameters:
flag (
bool) – True to enable, False to disable- Return type:
None
Warning
Disabling TLS verification is insecure and should only be used for testing purposes.
AsyncNepseClient (Asynchronous)
- class nepse_client.async_client.AsyncNepseClient[source]
Bases:
_NepseBaseAsynchronous client for NEPSE API.
This client provides non-blocking async methods to access Nepal Stock Exchange data, enabling concurrent operations and better performance for bulk requests.
- Parameters:
logger (
Optional[Logger]) – Optional custom logger instancemask_request_data (
bool) – Whether to mask sensitive data in logs (default: True)timeout (
float) – Request timeout in seconds (default: 100.0)
Example
Basic usage:
import asyncio from nepse_client import AsyncNepseClient async def main(): client = AsyncNepseClient() # Get market status status = await client.getMarketStatus() print(f"Market is {status['isOpen']}") # Concurrent requests status, summary, gainers = await asyncio.gather( client.getMarketStatus(), client.getSummary(), client.getTopGainers() ) asyncio.run(main())
Note
All methods are coroutines and must be awaited. The client automatically manages authentication tokens and handles token expiration.
- __init__(logger=None, mask_request_data=True, timeout=100.0)[source]
Initialize asynchronous NEPSE client.
- Parameters:
logger (Logger | None)
mask_request_data (bool)
timeout (float)
- init_client(tls_verify)[source]
Initialize async HTTP client with specified settings.
- Parameters:
tls_verify (
bool) – Whether to verify TLS certificates- Return type:
None
- async requestGETAPI(url, include_authorization_headers=True)[source]
Make async GET request to NEPSE API.
- Parameters:
url (
str) – API endpoint URLinclude_authorization_headers (
bool) – Whether to include auth headers
- Return type:
Any- Returns:
Parsed response data
- async requestPOSTAPI(url, payload_generator)[source]
Make async POST request to NEPSE API.
- Parameters:
url (
str) – API endpoint URLpayload_generator – Async function to generate payload
- Return type:
Any- Returns:
Parsed response data
- async getAuthorizationHeaders()[source]
Get headers with authorization token.
- Return type:
dict[str,str]- Returns:
Dictionary of HTTP headers
- async getPOSTPayloadIDForScrips()[source]
Generate payload ID for scrip-related requests.
- Return type:
int
- async getPOSTPayloadIDForFloorSheet(business_date=None)[source]
Generate payload ID for floor sheet requests.
- Parameters:
business_date (
Union[str,date,None]) – Business date (YYYY-MM-DD string or date object)- Return type:
int- Returns:
Payload ID integer
- async getMarketStatus()[source]
Get current market status (open/closed).
- Return type:
dict[str,Any]
- async getPriceVolume()[source]
Get current price and volume data for all securities.
- Return type:
list[dict[str,Any]]
- async getSummary()[source]
Get market summary with turnover, trades, etc.
- Return type:
dict[str,Any]
- async getTopTenTradeScrips()[source]
Get top 10 scrips by trade volume.
- Return type:
list[dict[str,Any]]
- async getTopTenTransactionScrips()[source]
Get top 10 scrips by transaction count.
- Return type:
list[dict[str,Any]]
- async getTopTenTurnoverScrips()[source]
Get top 10 scrips by turnover.
- Return type:
list[dict[str,Any]]
- async getTradingAverage(business_date=None, nDays=180)[source]
Get trading average data.
- Return type:
dict[str,Any]- Parameters:
business_date (str | None)
nDays (int)
- async getSecurityList()[source]
Get list of all securities (non-delisted).
- Return type:
list[dict[str,Any]]
- async getCompanyIDKeyMap(force_update=False)[source]
Get mapping of company symbols to IDs.
- Return type:
dict[str,int]- Parameters:
force_update (bool)
- async getSecurityIDKeyMap(force_update=False)[source]
Get mapping of security symbols to IDs.
- Return type:
dict[str,int]- Parameters:
force_update (bool)
- async getCompanyDetails(symbol)[source]
Get detailed information for a specific company.
- Return type:
dict[str,Any]- Parameters:
symbol (str)
- async getCompanyPriceVolumeHistory(symbol, start_date=None, end_date=None)[source]
Get price and volume history for a company.
- Return type:
dict[str,Any]- Parameters:
symbol (str)
start_date (str | date | None)
end_date (str | date | None)
- async getDailyScripPriceGraph(symbol)[source]
Get daily price graph data for a scrip.
- Return type:
dict[str,Any]- Parameters:
symbol (str)
- async getFloorSheet(show_progress=False)[source]
Get complete floor sheet data.
- Parameters:
show_progress (
bool) – Show progress bar during download- Return type:
list[dict[str,Any]]- Returns:
List of all floor sheet records
The asynchronous client is ideal for:
High-performance applications
Concurrent data fetching
Web servers (FastAPI, aiohttp)
Real-time data processing
Initialization:
from nepse_client import AsyncNepseClient
# Basic initialization
client = AsyncNepseClient()
# With custom settings
client = AsyncNepseClient(
logger=my_logger,
mask_request_data=True,
timeout=120.0
)
Context Manager:
import asyncio
async def main():
# Automatic resource cleanup
async with AsyncNepseClient() as client:
status = await client.getMarketStatus()
companies = await client.getCompanyList()
# Client automatically closed
asyncio.run(main())
Concurrent Requests:
import asyncio
from nepse_client import AsyncNepseClient
async def fetch_multiple():
async with AsyncNepseClient() as client:
# Fetch multiple data concurrently
results = await asyncio.gather(
client.getMarketStatus(),
client.getSummary(),
client.getTopGainers(),
client.getTopLosers()
)
status, summary, gainers, losers = results
asyncio.run(fetch_multiple())
Token Management
Token management for NEPSE API authentication.
This module handles automatic token generation, refresh, and validation for both synchronous and asynchronous clients.
- class nepse_client.token_manager.TokenManager[source]
Bases:
_TokenManagerBaseSynchronous token manager.
Manages authentication tokens for synchronous NEPSE client, automatically refreshing tokens when they expire.
- getAccessToken()[source]
Get valid access token, refreshing if necessary.
- Return type:
str- Returns:
Valid access token
- class nepse_client.token_manager.AsyncTokenManager[source]
Bases:
_TokenManagerBaseAsynchronous token manager.
Manages authentication tokens for asynchronous NEPSE client, with support for concurrent token refresh operations.
- async getAccessToken()[source]
Get valid access token, refreshing if necessary.
- Return type:
str- Returns:
Valid access token
- class nepse_client.token_manager.TokenParser[source]
Bases:
objectParse authentication tokens using WebAssembly module.
This class uses a WASM module to decode and parse the authentication tokens returned by the NEPSE API.
TokenManager (Synchronous)
- class nepse_client.token_manager.TokenManager[source]
Bases:
_TokenManagerBaseSynchronous token manager.
Manages authentication tokens for synchronous NEPSE client, automatically refreshing tokens when they expire.
- getAccessToken()[source]
Get valid access token, refreshing if necessary.
- Return type:
str- Returns:
Valid access token
Manages authentication tokens for synchronous client:
Automatic token refresh
Token validity checking
Salt management for payload generation
AsyncTokenManager (Asynchronous)
- class nepse_client.token_manager.AsyncTokenManager[source]
Bases:
_TokenManagerBaseAsynchronous token manager.
Manages authentication tokens for asynchronous NEPSE client, with support for concurrent token refresh operations.
- async getAccessToken()[source]
Get valid access token, refreshing if necessary.
- Return type:
str- Returns:
Valid access token
Manages authentication tokens for asynchronous client:
Automatic token refresh
Concurrent request handling
Event-based synchronization
Dummy ID Management
Dummy ID management for NEPSE API requests.
This module manages the generation and caching of dummy IDs used in POST request payloads, ensuring they stay synchronized with market status.
- class nepse_client.dummy_id_manager.DummyIDManager[source]
Bases:
_DummyIDManagerBaseSynchronous dummy ID manager.
Manages dummy IDs for synchronous NEPSE client, automatically updating when the date changes or market status is updated.
- __init__(market_status_function=None, date_function=<built-in method now of type object>)[source]
Initialize synchronous dummy ID manager.
- Parameters:
market_status_function (Callable | None)
date_function (Callable)
- class nepse_client.dummy_id_manager.AsyncDummyIDManager[source]
Bases:
_DummyIDManagerBaseAsynchronous dummy ID manager.
Manages dummy IDs for asynchronous NEPSE client, with support for concurrent operations and proper async synchronization.
- __init__(market_status_function=None, date_function=<built-in method now of type object>)[source]
Initialize asynchronous dummy ID manager.
- Parameters:
market_status_function (Callable | None)
date_function (Callable)
DummyIDManager (Synchronous)
- class nepse_client.dummy_id_manager.DummyIDManager[source]
Bases:
_DummyIDManagerBaseSynchronous dummy ID manager.
Manages dummy IDs for synchronous NEPSE client, automatically updating when the date changes or market status is updated.
- __init__(market_status_function=None, date_function=<built-in method now of type object>)[source]
Initialize synchronous dummy ID manager.
- Parameters:
market_status_function (Callable | None)
date_function (Callable)
Manages dummy IDs for POST request payloads:
Date-aware caching
Automatic updates on date change
Market status integration
AsyncDummyIDManager (Asynchronous)
- class nepse_client.dummy_id_manager.AsyncDummyIDManager[source]
Bases:
_DummyIDManagerBaseAsynchronous dummy ID manager.
Manages dummy IDs for asynchronous NEPSE client, with support for concurrent operations and proper async synchronization.
- __init__(market_status_function=None, date_function=<built-in method now of type object>)[source]
Initialize asynchronous dummy ID manager.
- Parameters:
market_status_function (Callable | None)
date_function (Callable)
Manages dummy IDs for asynchronous POST requests:
Date-aware caching
Concurrent request support
Event-based synchronization
Configuration
All clients support the following configuration options:
Parameters:
logger(Optional[logging.Logger]): Custom logger instancemask_request_data(bool): Mask sensitive data in logs (default: True)timeout(float): Request timeout in seconds (default: 100.0)
Example:
import logging
from nepse_client import NepseClient
# Setup logging
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)
# Initialize with custom config
client = NepseClient(
logger=logger,
mask_request_data=True,
timeout=60.0
)
# Disable TLS verification (testing only!)
client.setTLSVerification(False)
Performance Considerations
Synchronous vs Asynchronous
Use Synchronous Client when:
Making sequential requests
Simple scripts or notebooks
Straightforward data fetching
No need for concurrency
Use Asynchronous Client when:
Fetching multiple resources concurrently
Building web applications
High-performance requirements
Processing real-time data
Performance Comparison:
# Synchronous - Sequential (slower)
client = NepseClient()
status = client.getMarketStatus() # ~500ms
summary = client.getSummary() # ~500ms
gainers = client.getTopGainers() # ~500ms
# Total: ~1500ms
# Asynchronous - Concurrent (faster)
async with AsyncNepseClient() as client:
status, summary, gainers = await asyncio.gather(
client.getMarketStatus(), # \
client.getSummary(), # } ~500ms (concurrent)
client.getTopGainers() # /
)
# Total: ~500ms
Caching
The client automatically caches:
Company ID mappings
Security ID mappings
Sector scrips
To force cache refresh:
# Force refresh cache
company_map = client.getCompanyIDKeyMap(force_update=True)
security_map = client.getSecurityIDKeyMap(force_update=True)
Best Practices
Use Context Managers
# Good with NepseClient() as client: data = client.getMarketStatus() # Avoid client = NepseClient() data = client.getMarketStatus() # Resource not cleaned up!
Handle Errors Properly
from nepse_client import NepseClient, NepseError with NepseClient() as client: try: data = client.getMarketStatus() except NepseError as e: logger.error(f"Error: {e}")
Use Async for Multiple Requests
# Fetch 10 companies concurrently async with AsyncNepseClient() as client: symbols = ['NABIL', 'NICA', 'SCB', ...] tasks = [client.getCompanyDetails(s) for s in symbols] results = await asyncio.gather(*tasks)
Configure Appropriate Timeouts
# For slow connections client = NepseClient(timeout=180.0)
Enable Logging for Debugging
import logging logging.basicConfig(level=logging.DEBUG) client = NepseClient()