app.py

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.
""")