import streamlit as st import pandas as pd import yfinance as yf import ta from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime import logging import time # ------------------------------- # Streamlit App Configuration # ------------------------------- # Set the page layout to wide for better visibility st.set_page_config(layout="wide") # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger() # Set the title of the app st.title("NIFTY 500 RSI Divergence Scanner") # Add a description st.markdown(""" This application scans **NIFTY 500** stocks to identify RSI (Relative Strength Index) divergences. RSI divergence occurs when the price movement and RSI indicator move in opposite directions, potentially signaling a trend reversal. """) # ------------------------------- # Load NIFTY 500 Stock List # ------------------------------- @st.cache_data def load_stock_list(csv_path='ind_nifty500list.csv'): """ Loads the NIFTY 500 stock list from a CSV file. """ try: stock_list = pd.read_csv(csv_path) required_columns = {'Symbol', 'Exchange'} if not required_columns.issubset(stock_list.columns): st.error(f"CSV must contain the following columns: {required_columns}") return pd.DataFrame() return stock_list except FileNotFoundError: st.error(f"CSV file '{csv_path}' not found. Please ensure it is uploaded correctly.") return pd.DataFrame() except Exception as e: st.error(f"Error loading CSV file '{csv_path}': {e}") return pd.DataFrame() # Load the stock list stock_list = load_stock_list() if stock_list.empty: st.stop() tickers = stock_list['Symbol'].tolist() # ------------------------------- # User Inputs (Sidebar) # ------------------------------- # Sidebar for user inputs st.sidebar.header("User Inputs") # Multi-select Dropdown for Stock Selection selected_stocks = st.sidebar.multiselect( 'Select Stocks for Screening', tickers, default=tickers # Select all by default ) # RSI Parameters st.sidebar.subheader("RSI Parameters") rsi_period = st.sidebar.slider('RSI Period', min_value=7, max_value=28, value=14, step=1) rsi_lower_bound = st.sidebar.slider('RSI Lower Bound', min_value=10, max_value=50, value=30, step=5) rsi_upper_bound = st.sidebar.slider('RSI Upper Bound', min_value=50, max_value=90, value=70, step=5) # ------------------------------- # Processing Functions # ------------------------------- @st.cache_data def fetch_stock_data(ticker, exchange, period="6mo", interval="1d"): """ Fetches historical stock data using yfinance. Appends suffix based on the exchange. """ suffix_map = { 'NSE': '.NS', 'NASDAQ': '', 'NYSE': '', # Add other exchanges and their suffixes if needed } suffix = suffix_map.get(exchange.upper(), '') yf_ticker = f"{ticker}{suffix}" try: st.write(f"Fetching data for: {yf_ticker}") # Debug statement data = yf.download(yf_ticker, period=period, interval=interval, progress=False) if data.empty: st.warning(f"No data fetched for {yf_ticker}. Please check the ticker symbol.") logger.warning(f"No data for {yf_ticker}") else: logger.info(f"Data fetched for {yf_ticker}") return data except Exception as e: st.error(f"Error fetching data for {yf_ticker}: {e}") logger.error(f"Error fetching data for {yf_ticker}: {e}") return pd.DataFrame() def calculate_rsi(data, window): """ Calculates the RSI using the ta library. """ try: rsi_indicator = ta.momentum.RSIIndicator(close=data['Close'], window=window) data['RSI'] = rsi_indicator.rsi() return data except Exception as e: st.error(f"Error calculating RSI: {e}") logger.error(f"Error calculating RSI: {e}") return data def identify_divergence(data, lower_bound, upper_bound): """ Identifies RSI divergence in the stock data. """ try: data['Price Change'] = data['Close'].diff() data['RSI Change'] = data['RSI'].diff() # Divergence Condition: Price and RSI moving in opposite directions data['Divergence'] = (data['Price Change'] * data['RSI Change'] < 0).astype(int) # Filter for RSI levels and Divergence divergence_days = data[ ((data['RSI'] < lower_bound) | (data['RSI'] > upper_bound)) & (data['Divergence'] == 1) ] return divergence_days except Exception as e: st.error(f"Error identifying divergence: {e}") logger.error(f"Error identifying divergence: {e}") return pd.DataFrame() def process_ticker(ticker, exchange, rsi_period, lower_bound, upper_bound): """ Processes a single ticker to identify RSI divergence. """ time.sleep(0.1) # Slight delay to prevent rate limiting st.write(f"Processing ticker: {ticker}") # Debug statement data = fetch_stock_data(ticker, exchange) if data.empty: logger.warning(f"No data fetched for {ticker}") return {'Ticker': ticker, 'Divergence Dates': [], 'Error': 'No data fetched.'} data = calculate_rsi(data, rsi_period) divergence_days = identify_divergence(data, lower_bound, upper_bound) if not divergence_days.empty: # Convert Timestamps to date strings divergence_dates = [date.strftime('%Y-%m-%d') for date in divergence_days.index] st.success(f"RSI Divergence found for {ticker} on {divergence_dates}") return {'Ticker': ticker, 'Divergence Dates': divergence_dates, 'Error': None} else: st.info(f"No RSI Divergence found for {ticker}.") return {'Ticker': ticker, 'Divergence Dates': [], 'Error': None} # ------------------------------- # Scan Stocks for RSI Divergence # ------------------------------- st.header("RSI Divergence Results") # Initialize a list to hold divergence results divergence_results = [] # Initialize a list to hold errors error_results = [] # Progress bar setup progress_bar = st.progress(0) status_text = st.empty() total_stocks = len(selected_stocks) completed_stocks = 0 if selected_stocks: # Use ThreadPoolExecutor for parallel processing with ThreadPoolExecutor(max_workers=10) as executor: # Dictionary to keep track of futures future_to_ticker = { executor.submit(process_ticker, ticker, stock_list.loc[stock_list['Symbol'] == ticker, 'Exchange'].values[0], rsi_period, rsi_lower_bound, rsi_upper_bound): ticker for ticker in selected_stocks } for future in as_completed(future_to_ticker): ticker = future_to_ticker[future] try: result = future.result() if result['Error']: error_results.append(result) elif result['Divergence Dates']: divergence_results.append(result) except Exception as e: error_results.append({'Ticker': ticker, 'Divergence Dates': [], 'Error': str(e)}) finally: completed_stocks += 1 progress_percentage = completed_stocks / total_stocks progress_bar.progress(progress_percentage) status_text.text(f"Processing {completed_stocks} of {total_stocks} stocks...") # Finalize progress bar progress_bar.empty() status_text.empty() # ------------------------------- # Display Divergence Results # ------------------------------- if divergence_results: st.success("RSI Divergence Found in the Following Stocks:") for item in divergence_results: divergence_dates = ', '.join(item['Divergence Dates']) st.markdown(f"**{item['Ticker']}** - Divergence Dates: {divergence_dates}") else: st.info("No RSI divergences found for the selected stocks.") # ------------------------------- # Display Errors (If Any) # ------------------------------- if error_results: st.header("Errors Encountered") for item in error_results: st.error(f"**{item['Ticker']}** - Error: {item['Error']}") else: st.warning("Please select at least one stock to begin screening.") # ------------------------------- # Show Raw Data Option # ------------------------------- if st.sidebar.checkbox('Show Stock Data'): st.header("Raw Stock Data") for ticker in selected_stocks: exchange = stock_list.loc[stock_list['Symbol'] == ticker, 'Exchange'].values[0] st.subheader(f"Stock Data for: {ticker} ({exchange})") data = fetch_stock_data(ticker, exchange) if data.empty: st.write("No data available.") continue data = calculate_rsi(data, rsi_period) st.dataframe(data) # ------------------------------- # Footer # ------------------------------- st.markdown(""" --- **Disclaimer:** This tool is for informational purposes only and does not constitute financial advice. Please consult with a financial advisor before making investment decisions. """)