nepse_client.async_client

Asynchronous NEPSE client implementation.

This module provides a non-blocking, asynchronous interface to the NEPSE API, suitable for concurrent operations and high-performance applications.

Classes

AsyncNepseClient

Asynchronous client for NEPSE API.

class nepse_client.async_client.AsyncNepseClient

Bases: _NepseBase

Asynchronous 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 instance

• mask_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)

Initialize asynchronous NEPSE client.

Parameters:

• logger (Logger | None)

• mask_request_data (bool)

• timeout (float)

init_client(tls_verify)

Initialize async HTTP client with specified settings.

Parameters:

tls_verify (bool) – Whether to verify TLS certificates

Return type:

None

async __aenter__()

Async context manager entry.

async __aexit__(exc_type, exc_val, exc_tb)

Async context manager exit - cleanup resources.

async close()

Close HTTP client and cleanup resources.

Return type:

None

async requestGETAPI(url, include_authorization_headers=True)

Make async GET request to NEPSE API.

Parameters:

• url (str) – API endpoint URL

• include_authorization_headers (bool) – Whether to include auth headers

Return type:

Any

Returns:

Parsed response data

async requestPOSTAPI(url, payload_generator)

Make async POST request to NEPSE API.

Parameters:

• url (str) – API endpoint URL

• payload_generator – Async function to generate payload

Return type:

Any

Returns:

Parsed response data

async getAuthorizationHeaders()

Get headers with authorization token.

Return type:

dict[str, str]

Returns:

Dictionary of HTTP headers

async getPOSTPayloadIDForScrips()

Generate payload ID for scrip-related requests.

Return type:

int

async getPOSTPayloadID()

Generate general payload ID.

Return type:

int

async getPOSTPayloadIDForFloorSheet(business_date=None)

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()

Get current market status (open/closed).

Return type:

dict[str, Any]

async getPriceVolume()

Get current price and volume data for all securities.

Return type:

list[dict[str, Any]]

async getSummary()

Get market summary with turnover, trades, etc.

Return type:

dict[str, Any]

async getTopGainers()

Get list of top gaining stocks.

Return type:

list[dict[str, Any]]

async getTopLosers()

Get list of top losing stocks.

Return type:

list[dict[str, Any]]

async getTopTenTradeScrips()

Get top 10 scrips by trade volume.

Return type:

list[dict[str, Any]]

async getTopTenTransactionScrips()

Get top 10 scrips by transaction count.

Return type:

list[dict[str, Any]]

async getTopTenTurnoverScrips()

Get top 10 scrips by turnover.

Return type:

list[dict[str, Any]]

async getSupplyDemand()

Get supply and demand data.

Return type:

dict[str, Any]

async getNepseIndex()

Get NEPSE index data.

Return type:

dict[str, Any]

async getNepseSubIndices()

Get all NEPSE sub-indices.

Return type:

list[dict[str, Any]]

async getLiveMarket()

Get live market data.

Return type:

dict[str, Any]

async getTradingAverage(business_date=None, nDays=180)

Get trading average data.

Return type:

dict[str, Any]

Parameters:

• business_date (str | None)

• nDays (int)

async getCompanyList()

Get list of all listed companies.

Return type:

list[dict[str, Any]]

async getSecurityList()

Get list of all securities (non-delisted).

Return type:

list[dict[str, Any]]

async getCompanyIDKeyMap(force_update=False)

Get mapping of company symbols to IDs.

Return type:

dict[str, int]

Parameters:

force_update (bool)

async getSecurityIDKeyMap(force_update=False)

Get mapping of security symbols to IDs.

Return type:

dict[str, int]

Parameters:

force_update (bool)

async getSectorScrips()

Get scrips grouped by sector.

Return type:

dict[str, list[str]]

async getCompanyDetails(symbol)

Get detailed information for a specific company.

Return type:

dict[str, Any]

Parameters:

symbol (str)

async getCompanyPriceVolumeHistory(symbol, start_date=None, end_date=None)

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)

Get daily price graph data for a scrip.

Return type:

dict[str, Any]

Parameters:

symbol (str)

async getFloorSheet(show_progress=False)

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

async getFloorSheetOf(symbol, business_date=None)

Get floor sheet for a specific company.

Return type:

list[dict[str, Any]]

Parameters:

• symbol (str)

• business_date (str | date | None)

async getSymbolMarketDepth(symbol)

Get market depth for a symbol.

Return type:

dict[str, Any]

Parameters:

symbol (str)

headers: dict[str, str]

company_symbol_id_keymap: Optional[dict[str, int]]

security_symbol_id_keymap: Optional[dict[str, int]]

company_list: Optional[list[dict]]

security_list: Optional[list[dict]]

holiday_list: Optional[list[dict]]

sector_scrips: Optional[dict[str, list[str]]]