Create data_sources.py

data_sources.py

@@ -0,0 +1,406 @@
+"""
+Real-time market data integration for Indian and global stocks
+Optimized for Hugging Face free tier
+"""
+
+import requests
+import yfinance as yf
+import pandas as pd
+import numpy as np
+from typing import Dict, List, Optional, Tuple
+import time
+from datetime import datetime, timedelta
+
+class IndianStockAPI:
+    """
+    Free Indian Stock Market API client for NSE/BSE data
+    No API key required [citation:1]
+    """
+    
+    BASE_URL = "https://military-jobye-haiqstudios-14f59639.koyeb.app"
+    
+    def __init__(self):
+        self.session = requests.Session()
+        self.cache = {}
+        self.cache_timestamp = {}
+        self.cache_duration = 60  # Cache for 60 seconds to avoid rate limits
+    
+    def search_stocks(self, query: str) -> List[Dict]:
+        """
+        Search for stocks by company name or symbol
+        Example: /search?q=reliance [citation:1]
+        """
+        try:
+            url = f"{self.BASE_URL}/search"
+            params = {'q': query}
+            response = self.session.get(url, params=params, timeout=10)
+            
+            if response.status_code == 200:
+                data = response.json()
+                if data.get('status') == 'success':
+                    return data.get('results', [])
+            return []
+        except Exception as e:
+            print(f"Search error: {e}")
+            return []
+    
+    def get_stock_price(self, symbol: str, exchange: str = "NSE") -> Optional[Dict]:
+        """
+        Get real-time stock price and details
+        Examples: 
+          - ITC → NSE (default) [citation:1]
+          - ITC.NS → NSE explicit
+          - ITC.BO → BSE
+        """
+        # Check cache first
+        cache_key = f"{symbol}_{exchange}"
+        if cache_key in self.cache:
+            if time.time() - self.cache_timestamp[cache_key] < self.cache_duration:
+                return self.cache[cache_key]
+        
+        try:
+            # Format symbol with exchange suffix
+            if exchange.upper() == "NSE":
+                api_symbol = f"{symbol}.NS" if not symbol.endswith('.NS') else symbol
+            else:  # BSE
+                api_symbol = f"{symbol}.BO" if not symbol.endswith('.BO') else symbol
+            
+            url = f"{self.BASE_URL}/stock"
+            params = {
+                'symbol': api_symbol,
+                'res': 'val'  # Get values with units [citation:1]
+            }
+            
+            response = self.session.get(url, params=params, timeout=10)
+            
+            if response.status_code == 200:
+                data = response.json()
+                if data.get('status') == 'success':
+                    # Cache the result
+                    self.cache[cache_key] = data
+                    self.cache_timestamp[cache_key] = time.time()
+                    return data
+            return None
+        except Exception as e:
+            print(f"Stock price error: {e}")
+            return None
+    
+    def get_multiple_stocks(self, symbols: List[str], exchange: str = "NSE") -> Dict:
+        """
+        Get prices for multiple stocks efficiently
+        """
+        results = {}
+        for symbol in symbols:
+            data = self.get_stock_price(symbol, exchange)
+            if data:
+                results[symbol] = data
+            time.sleep(0.5)  # Rate limiting protection
+        return results
+    
+    def format_for_index(self, stock_data: Dict) -> Dict:
+        """
+        Convert stock API response to index calculator format
+        """
+        if not stock_data or 'data' not in stock_data:
+            return {}
+        
+        d = stock_data['data']
+        
+        # Handle both numeric and unit formats [citation:1]
+        def extract_value(field):
+            if isinstance(field, dict):
+                return field.get('value', 0)
+            return field
+        
+        return {
+            'company_name': d.get('company_name', ''),
+            'last_price': extract_value(d.get('last_price', 0)),
+            'change': extract_value(d.get('change', 0)),
+            'percent_change': extract_value(d.get('percent_change', 0)),
+            'previous_close': extract_value(d.get('previous_close', 0)),
+            'open': extract_value(d.get('open', 0)),
+            'day_high': extract_value(d.get('day_high', 0)),
+            'day_low': extract_value(d.get('day_low', 0)),
+            'year_high': extract_value(d.get('year_high', 0)),
+            'year_low': extract_value(d.get('year_low', 0)),
+            'volume': extract_value(d.get('volume', 0)),
+            'market_cap': extract_value(d.get('market_cap', 0)),
+            'pe_ratio': extract_value(d.get('pe_ratio', 0)),
+            'dividend_yield': extract_value(d.get('dividend_yield', 0)),
+            'sector': d.get('sector', ''),
+            'currency': d.get('currency', 'INR')
+        }
+
+
+class GlobalStockAPI:
+    """
+    Yahoo Finance client for global stocks and indices
+    Uses yfinance library [citation:2]
+    """
+    
+    def __init__(self):
+        self.cache = {}
+        self.cache_timestamp = {}
+        self.cache_duration = 60  # Cache for 60 seconds
+    
+    def get_stock_info(self, symbol: str) -> Optional[Dict]:
+        """
+        Get real-time stock data from Yahoo Finance
+        Examples: 'RELIANCE.NS', 'AAPL', '^NSEI' (Nifty 50)
+        """
+        # Check cache
+        if symbol in self.cache:
+            if time.time() - self.cache_timestamp[symbol] < self.cache_duration:
+                return self.cache[symbol]
+        
+        try:
+            ticker = yf.Ticker(symbol)
+            info = ticker.info
+            
+            # Get current price (fastest method)
+            hist = ticker.history(period="1d", interval="1m")
+            
+            if not hist.empty:
+                current_price = hist['Close'].iloc[-1]
+            else:
+                current_price = info.get('regularMarketPrice', info.get('currentPrice', 0))
+            
+            result = {
+                'symbol': symbol,
+                'company_name': info.get('longName', info.get('shortName', symbol)),
+                'last_price': current_price,
+                'previous_close': info.get('regularMarketPreviousClose', 0),
+                'open': info.get('regularMarketOpen', 0),
+                'day_high': info.get('dayHigh', info.get('regularMarketDayHigh', 0)),
+                'day_low': info.get('dayLow', info.get('regularMarketDayLow', 0)),
+                'volume': info.get('volume', info.get('regularMarketVolume', 0)),
+                'market_cap': info.get('marketCap', 0),
+                'pe_ratio': info.get('trailingPE', 0),
+                'dividend_yield': info.get('dividendYield', 0) * 100 if info.get('dividendYield') else 0,
+                'sector': info.get('sector', ''),
+                'industry': info.get('industry', ''),
+                'currency': info.get('currency', 'USD'),
+                'timestamp': datetime.now().isoformat()
+            }
+            
+            # Cache the result
+            self.cache[symbol] = result
+            self.cache_timestamp[symbol] = time.time()
+            
+            return result
+            
+        except Exception as e:
+            print(f"YFinance error for {symbol}: {e}")
+            return None
+    
+    def get_nifty50_components(self) -> List[str]:
+        """
+        Get Nifty 50 constituent symbols
+        """
+        # Common Nifty 50 stocks with .NS suffix for Yahoo Finance
+        return [
+            'RELIANCE.NS', 'TCS.NS', 'HDFCBANK.NS', 'INFY.NS', 'ICICIBANK.NS',
+            'HINDUNILVR.NS', 'ITC.NS', 'SBIN.NS', 'BHARTIARTL.NS', 'KOTAKBANK.NS',
+            'BAJFINANCE.NS', 'LT.NS', 'WIPRO.NS', 'AXISBANK.NS', 'TITAN.NS',
+            'ASIANPAINT.NS', 'MARUTI.NS', 'SUNPHARMA.NS', 'HCLTECH.NS', 'ULTRACEMCO.NS'
+        ]
+    
+    def get_multiple_stocks(self, symbols: List[str]) -> Dict:
+        """
+        Get data for multiple symbols efficiently
+        """
+        results = {}
+        for symbol in symbols:
+            data = self.get_stock_info(symbol)
+            if data:
+                results[symbol] = data
+        return results
+
+
+class IndexBuilder:
+    """
+    Build custom indices from real-time stock data
+    """
+    
+    def __init__(self):
+        self.indian_api = IndianStockAPI()
+        self.global_api = GlobalStockAPI()
+    
+    def create_price_weighted_index(self, symbols: List[str], exchange: str = "NSE") -> Dict:
+        """
+        Create a price-weighted index (like Dow Jones)
+        Index = (Sum of stock prices) / Divisor
+        """
+        stocks_data = []
+        total_price = 0
+        
+        for symbol in symbols:
+            if exchange in ["NSE", "BSE"]:
+                data = self.indian_api.get_stock_price(symbol, exchange)
+                if data and 'data' in data:
+                    price_data = self.indian_api.format_for_index(data)
+                    stocks_data.append(price_data)
+                    total_price += price_data['last_price']
+            else:
+                data = self.global_api.get_stock_info(symbol)
+                if data:
+                    stocks_data.append(data)
+                    total_price += data['last_price']
+        
+        # Simple divisor (number of stocks for demo)
+        divisor = len(stocks_data)
+        index_value = total_price / divisor if divisor > 0 else 0
+        
+        return {
+            'index_name': f"Price-Weighted Index ({exchange})",
+            'index_value': round(index_value, 2),
+            'components': len(stocks_data),
+            'total_price': round(total_price, 2),
+            'divisor': divisor,
+            'stocks': stocks_data,
+            'timestamp': datetime.now().isoformat()
+        }
+    
+    def create_market_cap_weighted_index(self, symbols: List[str], exchange: str = "NSE") -> Dict:
+        """
+        Create a market capitalization-weighted index (like S&P 500, Nifty 50)
+        Index = (Sum of market caps) / Base market cap * Base index value
+        """
+        stocks_data = []
+        total_market_cap = 0
+        
+        for symbol in symbols:
+            if exchange in ["NSE", "BSE"]:
+                data = self.indian_api.get_stock_price(symbol, exchange)
+                if data and 'data' in data:
+                    price_data = self.indian_api.format_for_index(data)
+                    stocks_data.append(price_data)
+                    total_market_cap += price_data['market_cap']
+            else:
+                data = self.global_api.get_stock_info(symbol)
+                if data:
+                    stocks_data.append(data)
+                    total_market_cap += data['market_cap']
+        
+        # Use first stock's market cap as base for demo
+        base_market_cap = stocks_data[0]['market_cap'] if stocks_data else 1
+        index_value = (total_market_cap / base_market_cap) * 100 if base_market_cap > 0 else 0
+        
+        return {
+            'index_name': f"Market Cap-Weighted Index ({exchange})",
+            'index_value': round(index_value, 2),
+            'components': len(stocks_data),
+            'total_market_cap': total_market_cap,
+            'base_market_cap': base_market_cap,
+            'stocks': stocks_data,
+            'timestamp': datetime.now().isoformat()
+        }
+    
+    def calculate_sector_index(self, sector: str, exchange: str = "NSE") -> Dict:
+        """
+        Create an index for a specific sector
+        """
+        # This would normally filter stocks by sector
+        # For demo, we'll use a few known stocks
+        sector_stocks = {
+            'technology': ['TCS.NS', 'INFY.NS', 'WIPRO.NS', 'HCLTECH.NS'],
+            'banking': ['HDFCBANK.NS', 'ICICIBANK.NS', 'SBIN.NS', 'KOTAKBANK.NS'],
+            'energy': ['RELIANCE.NS', 'ONGC.NS', 'POWERGRID.NS']
+        }
+        
+        symbols = sector_stocks.get(sector.lower(), [])
+        if not symbols:
+            return {'error': f'Sector {sector} not found'}
+        
+        return self.create_market_cap_weighted_index(symbols, exchange)
+
+
+class RealTimeIndexCalculator:
+    """
+    Main class that combines real-time data with index calculations
+    """
+    
+    def __init__(self):
+        self.index_builder = IndexBuilder()
+        self.indian_api = IndianStockAPI()
+        self.global_api = GlobalStockAPI()
+        from index_calculator import IndexCalculator
+        self.calculator = IndexCalculator()
+    
+    def calculate_live_laspeyres(self, symbols: List[str], base_prices: List[float], 
+                                 base_quantities: List[float], exchange: str = "NSE") -> Dict:
+        """
+        Calculate Laspeyres index using live prices
+        """
+        # Get live prices
+        live_prices = []
+        stock_details = []
+        
+        for i, symbol in enumerate(symbols):
+            if exchange in ["NSE", "BSE"]:
+                data = self.indian_api.get_stock_price(symbol, exchange)
+                if data and 'data' in data:
+                    price_data = self.indian_api.format_for_index(data)
+                    live_prices.append(price_data['last_price'])
+                    stock_details.append(price_data)
+            else:
+                data = self.global_api.get_stock_info(symbol)
+                if data:
+                    live_prices.append(data['last_price'])
+                    stock_details.append(data)
+        
+        # Create DataFrame for index calculator
+        df = pd.DataFrame({
+            'item': symbols,
+            'p0': base_prices[:len(live_prices)],
+            'q0': base_quantities[:len(live_prices)],
+            'p1': live_prices
+        })
+        
+        self.calculator.load_data(df)
+        laspeyres = self.calculator.calculate_laspeyres()
+        
+        return {
+            'index_value': laspeyres['value'],
+            'interpretation': laspeyres['interpretation'],
+            'live_prices': live_prices,
+            'stock_details': stock_details,
+            'timestamp': datetime.now().isoformat()
+        }
+    
+    def compare_indices(self, symbols: List[str], exchange: str = "NSE") -> pd.DataFrame:
+        """
+        Compare multiple index calculations on the same basket
+        """
+        # Get live data
+        stocks_data = []
+        for symbol in symbols:
+            if exchange in ["NSE", "BSE"]:
+                data = self.indian_api.get_stock_price(symbol, exchange)
+                if data and 'data' in data:
+                    stocks_data.append(self.indian_api.format_for_index(data))
+            else:
+                data = self.global_api.get_stock_info(symbol)
+                if data:
+                    stocks_data.append(data)
+        
+        if not stocks_data:
+            return pd.DataFrame()
+        
+        # Prepare data for index calculator
+        df = pd.DataFrame({
+            'item': [s.get('company_name', f'Stock{i}') for i, s in enumerate(stocks_data)],
+            'p0': [100.0] * len(stocks_data),  # Assume base price 100
+            'q0': [1.0] * len(stocks_data),    # Equal quantities
+            'p1': [s['last_price'] for s in stocks_data]
+        })
+        
+        self.calculator.load_data(df)
+        results = self.calculator.calculate_all()
+        
+        # Add live data
+        results['live_data'] = stocks_data
+        results['symbols'] = symbols
+        results['exchange'] = exchange
+        
+        return results