nepse_client.AsyncNepseClient

class nepse_client.AsyncNepseClient[source]

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)[source]

Initialize asynchronous NEPSE client.

Parameters:

logger (Logger | None)
mask_request_data (bool)
timeout (float)

Methods

`__init__`([logger, mask_request_data, timeout])	Initialize asynchronous NEPSE client.
`close`()	Close HTTP client and cleanup resources.
`getAuthorizationHeaders`()	Get headers with authorization token.
`getCompanyDetails`(symbol)	Get detailed information for a specific company.
`getCompanyIDKeyMap`([force_update])	Get mapping of company symbols to IDs.
`getCompanyList`()	Get list of all listed companies.
`getCompanyPriceVolumeHistory`(symbol[, ...])	Get price and volume history for a company.
`getDailyScripPriceGraph`(symbol)	Get daily price graph data for a scrip.
`getDummyData`()	Get dummy data array.
`getDummyID`()	Get dummy ID for current market state.
`getFloorSheet`([show_progress])	Get complete floor sheet data.
`getFloorSheetOf`(symbol[, business_date])	Get floor sheet for a specific company.
`getLiveMarket`()	Get live market data.
`getMarketStatus`()	Get current market status (open/closed).
`getNepseIndex`()	Get NEPSE index data.
`getNepseSubIndices`()	Get all NEPSE sub-indices.
`getPOSTPayloadID`()	Generate general payload ID.
`getPOSTPayloadIDForFloorSheet`([business_date])	Generate payload ID for floor sheet requests.
`getPOSTPayloadIDForScrips`()	Generate payload ID for scrip-related requests.
`getPriceVolume`()	Get current price and volume data for all securities.
`getSectorScrips`()	Get scrips grouped by sector.
`getSecurityIDKeyMap`([force_update])	Get mapping of security symbols to IDs.
`getSecurityList`()	Get list of all securities (non-delisted).
`getSummary`()	Get market summary with turnover, trades, etc.
`getSupplyDemand`()	Get supply and demand data.
`getSymbolMarketDepth`(symbol)	Get market depth for a symbol.
`getTopGainers`()	Get list of top gaining stocks.
`getTopLosers`()	Get list of top losing stocks.
`getTopTenTradeScrips`()	Get top 10 scrips by trade volume.
`getTopTenTransactionScrips`()	Get top 10 scrips by transaction count.
`getTopTenTurnoverScrips`()	Get top 10 scrips by turnover.
`getTradingAverage`([business_date, nDays])	Get trading average data.
`get_full_url`(api_url)	Construct full URL from API endpoint.
`get_random_user_agent`()	Get random user agent.
`handle_response`(response[, request_data])	Process HTTP response and handle errors.
`init_client`(tls_verify)	Initialize async HTTP client with specified settings.
`requestGETAPI`(url[, ...])	Make async GET request to NEPSE API.
`requestPOSTAPI`(url, payload_generator)	Make async POST request to NEPSE API.
`setTLSVerification`([flag])	Enable or disable TLS certificate verification.

Attributes

headers

__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 __aenter__()[source]: Async context manager entry.

async __aexit__(exc_type, exc_val, exc_tb)[source]: Async context manager exit - cleanup resources.

async close()[source]

Close HTTP client and cleanup resources.

Return type:: None

async requestGETAPI(url, include_authorization_headers=True)[source]

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)[source]

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()[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 getPOSTPayloadID()[source]

Generate general payload ID.

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 getTopGainers()[source]

Get list of top gaining stocks.

Return type:: list[dict[str, Any]]

async getTopLosers()[source]

Get list of top losing stocks.

Return type:: list[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 getSupplyDemand()[source]

Get supply and demand data.

Return type:: dict[str, Any]

async getNepseIndex()[source]

Get NEPSE index data.

Return type:: dict[str, Any]

async getNepseSubIndices()[source]

Get all NEPSE sub-indices.

Return type:: list[dict[str, Any]]

async getLiveMarket()[source]

Get live market data.

Return type:: 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 getCompanyList()[source]

Get list of all listed companies.

Return type:: list[dict[str, Any]]

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 getSectorScrips()[source]

Get scrips grouped by sector.

Return type:: dict[str, list[str]]

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

async getFloorSheetOf(symbol, business_date=None)[source]

Get floor sheet for a specific company.

Return type:

list[dict[str, Any]]

Parameters:

symbol (str)
business_date (str | date | None)

async getSymbolMarketDepth(symbol)[source]

Get market depth for a symbol.

Return type:: dict[str, Any]
Parameters:: symbol (str)