#!/usr/bin/env python3 """ NEPSE Stock Map Updater This script updates the stockmap.json file by fetching the latest security list and sector information from the NEPSE API. It should be run periodically to keep the stock map current as the exchange adds new stocks. Author: NEPSE API Project License: MIT """ import asyncio import json import os import sys from typing import Dict, Any, Optional import logging import argparse from datetime import datetime try: import httpx except ImportError: print("Error: httpx not installed. Please install with: pip install httpx") sys.exit(1) # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('stock_update.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # Constants STOCK_MAP_FILE = "stockmap.json" API_BASE_URL = "http://localhost:8088" # Internal sector mapping based on NEPSE indices INTERNAL_SECTOR_MAP = { "Commercial Banks": "Banking SubIndex", "Development Banks": "Development Bank Ind.", "Finance": "Finance Index", "Hotels And Tourism": "Hotels And Tourism", "Hydro Power": "HydroPower Index", "Investment": "Investment", "Life Insurance": "Life Insurance", "Manufacturing And Processing": "Manufacturing And Pr.", "Microfinance": "Microfinance Index", "Mutual Fund": "Mutual Fund", "NEPSE": "NEPSE Index", "Non Life Insurance": "Non Life Insurance", "Others": "Others Index", "Tradings": "Trading Index", "Promoter Share": "Promoter Share", } class StockMapUpdater: """Updates the stock map from NEPSE API endpoints""" def __init__(self, api_base_url: str = API_BASE_URL): self.api_base_url = api_base_url self.client = httpx.AsyncClient(timeout=30.0) async def __aenter__(self): return self async def __aexit__(self, exc_type, exc_val, exc_tb): await self.client.aclose() async def check_server_health(self) -> bool: """Check if the API server is running""" try: response = await self.client.get(f"{self.api_base_url}/health") return response.status_code == 200 except Exception as e: logger.error(f"Server health check failed: {e}") return False async def fetch_company_list(self) -> list: """Fetch the full company list (primary source — has sectorName built-in)""" try: response = await self.client.get(f"{self.api_base_url}/CompanyList") response.raise_for_status() data = response.json() logger.info(f"CompanyList fetched: {len(data)} items") return data except httpx.HTTPStatusError as e: logger.error(f"CompanyList API error: {e.response.status_code}") raise except Exception as e: logger.error(f"Failed to fetch company list: {e}") raise async def fetch_security_list(self) -> list: """Fetch the security list (supplemental — catches symbols missing from CompanyList)""" try: response = await self.client.get(f"{self.api_base_url}/SecurityList") response.raise_for_status() data = response.json() logger.info(f"SecurityList fetched: {len(data)} items") return data except Exception as e: logger.warning(f"SecurityList fetch failed (non-fatal): {e}") return [] def create_stock_map(self, company_data: list, security_data: list) -> dict: """Build stockmap from CompanyList (primary) + SecurityList (supplement).""" stock_map = {} # Primary: CompanyList has sectorName directly for item in company_data: if item.get("status") != "A": continue symbol = item.get("symbol", "") if not symbol or not isinstance(symbol, str): continue sector = item.get("sectorName", "Unknown") internal_sector = INTERNAL_SECTOR_MAP.get(sector, "Others Index") stock_map[symbol] = { "name": item.get("securityName") or item.get("companyName") or symbol, "sector": sector, "internalSector": internal_sector, } # Supplement: add any active symbols from SecurityList not already present for item in security_data: if item.get("activeStatus") != "A": continue symbol = item.get("symbol", "") if not symbol or not isinstance(symbol, str) or symbol in stock_map: continue stock_map[symbol] = { "name": item.get("securityName") or item.get("name") or symbol, "sector": "Unknown", "internalSector": "Others Index", } logger.info(f"Stock map built: {len(stock_map)} total entries") return stock_map def save_stock_map(self, stock_map: dict) -> bool: """Save the stock map to file""" try: with open(STOCK_MAP_FILE, 'w', encoding='utf-8') as f: json.dump(stock_map, f, indent=2, ensure_ascii=False) logger.info(f"Stock map saved successfully to {STOCK_MAP_FILE}") return True except Exception as e: logger.error(f"Failed to save stock map: {e}") return False async def update_stock_map(self) -> bool: """Main function to update the stock map""" try: # Check server health logger.info("Checking API server health...") if not await self.check_server_health(): logger.error("API server is not responding. Please start the server first.") return False # Fetch data — CompanyList is primary (has sector info + more symbols) logger.info("Fetching company list (primary)...") company_data = await self.fetch_company_list() logger.info("Fetching security list (supplemental)...") security_data = await self.fetch_security_list() # Process data logger.info("Processing data...") stock_map = self.create_stock_map(company_data, security_data) # Save to file logger.info("Saving stock map...") if self.save_stock_map(stock_map): logger.info("Stock map update completed successfully!") return True else: logger.error("Failed to save stock map") return False except Exception as e: logger.error(f"Stock map update failed: {e}") return False async def main(): """Main entry point""" parser = argparse.ArgumentParser(description="Update NEPSE stock map") parser.add_argument("--api-url", default=API_BASE_URL, help=f"API base URL (default: {API_BASE_URL})") parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose logging") args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.DEBUG) logger.info("=== NEPSE Stock Map Updater ===") logger.info(f"API URL: {args.api_url}") logger.info(f"Target file: {STOCK_MAP_FILE}") async with StockMapUpdater(args.api_url) as updater: success = await updater.update_stock_map() if success: logger.info("Update completed successfully!") sys.exit(0) else: logger.error("Update failed!") sys.exit(1) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("Update cancelled by user") sys.exit(1) except Exception as e: logger.error(f"Unexpected error: {e}") sys.exit(1)