# FB Page: https://www.facebook.com/AIsparking import streamlit as st import yfinance as yf import pandas as pd import numpy as np from ta.trend import MACD from ta.momentum import RSIIndicator, StochasticOscillator from ta.volatility import AverageTrueRange, BollingerBands from ta.volume import VolumeWeightedAveragePrice import plotly.graph_objects as go from datetime import datetime, timedelta import plotly.subplots as sp from ta.trend import IchimokuIndicator # Set page config st.set_page_config(layout="wide", page_title="Stock Technical Analysis") # Functions from previous implementations remain the same # Add error handling wrapper def safe_execute(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: st.error(f"Error: {str(e)}") return None return wrapper @safe_execute def get_stock_data(symbol, market, nYear): if market == 'HK': symbol = f'{symbol}.HK' end_date = datetime.now() start_date = end_date - timedelta(days=nYear*365) df = yf.download(symbol, start=start_date, end=end_date) if df.empty: raise ValueError("No data found for this symbol") return df # Combine all technical indicators @safe_execute def generate_recommendation(df): last_row = df.iloc[-1] signals = [] # RSI signals if last_row['RSI'] < 30: signals.append(('BUY', 'RSI oversold')) elif last_row['RSI'] > 70: signals.append(('SELL', 'RSI overbought')) # Stochastic signals if last_row['%K'] < 20 and last_row['%D'] < 20: signals.append(('BUY', 'Stochastic oversold')) elif last_row['%K'] > 80 and last_row['%D'] > 80: signals.append(('SELL', 'Stochastic overbought')) # MACD signals if last_row['MACD'] > last_row['MACD_Signal']: signals.append(('BUY', 'MACD crossover')) elif last_row['MACD'] < last_row['MACD_Signal']: signals.append(('SELL', 'MACD crossunder')) return signals def calculate_indicators(df): # Calculate RSI rsi = RSIIndicator(df['Close']) df['RSI'] = rsi.rsi() # Calculate Stochastic stoch = StochasticOscillator(df['High'], df['Low'], df['Close']) df['%K'] = stoch.stoch() df['%D'] = stoch.stoch_signal() # Calculate MACD macd = MACD(df['Close']) df['MACD'] = macd.macd() df['MACD_Signal'] = macd.macd_signal() # Calculate ATR atr = AverageTrueRange(df['High'], df['Low'], df['Close']) df['ATR'] = atr.average_true_range() # Add multiple SMAs df['SMA20'] = df['Close'].rolling(window=20).mean() df['SMA50'] = df['Close'].rolling(window=50).mean() df['SMA100'] = df['Close'].rolling(window=100).mean() df['SMA200'] = df['Close'].rolling(window=200).mean() # Calculate ATR atr = AverageTrueRange(df['High'], df['Low'], df['Close']) df['ATR'] = atr.average_true_range() # Calculate VWAP vwap = VolumeWeightedAveragePrice(high=df['High'], low=df['Low'], close=df['Close'], volume=df['Volume']) df['VWAP'] = vwap.volume_weighted_average_price() return df def calculate_additional_indicators(df): # Add Bollinger Bands bb = BollingerBands(df['Close']) df['BB_upper'] = bb.bollinger_hband() df['BB_lower'] = bb.bollinger_lband() df['BB_middle'] = bb.bollinger_mavg() # Add Moving Averages df['MA50'] = df['Close'].rolling(window=50).mean() df['MA200'] = df['Close'].rolling(window=200).mean() # Add VWAP vwap = VolumeWeightedAveragePrice(high=df['High'], low=df['Low'], close=df['Close'], volume=df['Volume']) df['VWAP'] = vwap.volume_weighted_average_price() return df def calculate_ema(df): # Calculate EMAs df['EMA9'] = df['Close'].ewm(span=9, adjust=False).mean() df['EMA21'] = df['Close'].ewm(span=21, adjust=False).mean() return df def calculate_obv(df): # Calculate OBV obv = [] prev_obv = 0 for i in range(len(df)): if i == 0: obv.append(prev_obv) continue if df['Close'].iloc[i] > df['Close'].iloc[i-1]: current_obv = prev_obv + df['Volume'].iloc[i] elif df['Close'].iloc[i] < df['Close'].iloc[i-1]: current_obv = prev_obv - df['Volume'].iloc[i] else: current_obv = prev_obv obv.append(current_obv) prev_obv = current_obv df['OBV'] = obv return df def calculate_ichimoku(df): ichimoku = IchimokuIndicator(high=df['High'], low=df['Low']) df['ichimoku_a'] = ichimoku.ichimoku_a() df['ichimoku_b'] = ichimoku.ichimoku_b() df['ichimoku_base'] = ichimoku.ichimoku_base_line() df['ichimoku_conversion'] = ichimoku.ichimoku_conversion_line() return df def calculate_all_indicators(df): df = calculate_indicators(df) df = calculate_additional_indicators(df) df = calculate_ema(df) df = calculate_obv(df) df = calculate_ichimoku(df) return df def enhanced_recommendation(df): last_row = df.iloc[-1] prev_row = df.iloc[-2] signals = [] # Add EMA signals if last_row['EMA9'] > last_row['EMA21']: signals.append(('BUY', 'EMA9 crossed above EMA21')) elif last_row['EMA9'] < last_row['EMA21']: signals.append(('SELL', 'EMA9 crossed below EMA21')) # Add Ichimoku signals if (last_row['ichimoku_conversion'] > last_row['ichimoku_base'] and last_row['Close'] > last_row['ichimoku_a']): signals.append(('BUY', 'Ichimoku bullish signal')) elif (last_row['ichimoku_conversion'] < last_row['ichimoku_base'] and last_row['Close'] < last_row['ichimoku_b']): signals.append(('SELL', 'Ichimoku bearish signal')) # Add OBV signals if df['OBV'].iloc[-1] > df['OBV'].iloc[-2]: signals.append(('BUY', 'OBV increasing')) else: signals.append(('SELL', 'OBV decreasing')) # Add SMA signals if (last_row['SMA20'] > last_row['SMA50'] and prev_row['SMA20'] <= prev_row['SMA50']): signals.append(('BUY', 'SMA20 crossed above SMA50')) elif (last_row['SMA20'] < last_row['SMA50'] and prev_row['SMA20'] >= prev_row['SMA50']): signals.append(('SELL', 'SMA20 crossed below SMA50')) # Add VWAP signals if last_row['Close'] > last_row['VWAP']: signals.append(('BUY', 'Price above VWAP')) else: signals.append(('SELL', 'Price below VWAP')) # Add ATR-based volatility signals atr_threshold = df['ATR'].mean() * 1.5 if last_row['ATR'] > atr_threshold: signals.append(('HOLD', 'High volatility detected by ATR')) return signals # Main app layout st.title('Advanced Stock Technical Analysis') # Sidebar for inputs with st.sidebar: st.header('Input Parameters') market = st.selectbox('Select Market', ['HK','US']) symbol = st.text_input('Enter Stock Symbol (e.g. 0700 for HK):') # Add analysis timeframe option timeframe = st.selectbox('Select Timeframe', ['1y','2y','3y','5y','8y','10y']) nYear = int(timeframe.split("y")[0]) if st.sidebar.button('Analyze'): if symbol: with st.spinner('Fetching and analyzing data...'): df = get_stock_data(symbol, market,nYear) if df is not None: df = calculate_all_indicators(df) # Create tabs for different analyses tab1, tab2, tab3 = st.tabs(['Price Analysis', 'Technical Indicators', 'Technical Analysis']) with tab1: # Main price chart with volume fig = sp.make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.03, row_heights=[0.7, 0.3]) fig.add_trace(go.Candlestick(x=df.index, open=df['Open'], high=df['High'], low=df['Low'], close=df['Close'], name='Price'), row=1, col=1) # Add Moving Averages fig.add_trace(go.Scatter(x=df.index, y=df['MA50'], name='MA50', line=dict(color='orange')), row=1, col=1) fig.add_trace(go.Scatter(x=df.index, y=df['MA200'], name='MA200', line=dict(color='blue')), row=1, col=1) # Add volume bars colors = ['red' if row['Open'] > row['Close'] else 'green' for index, row in df.iterrows()] fig.add_trace(go.Bar(x=df.index, y=df['Volume'], marker_color=colors, name='Volume'), row=2, col=1) fig.update_layout(height=800) st.plotly_chart(fig, use_container_width=True) with tab2: col1, col2 = st.columns(2) with col1: # RSI Plot fig_rsi = go.Figure() fig_rsi.add_trace(go.Scatter(x=df.index, y=df['RSI'], name='RSI')) fig_rsi.add_hline(y=70, line_dash="dash", line_color="red") fig_rsi.add_hline(y=30, line_dash="dash", line_color="green") fig_rsi.update_layout(title='RSI Indicator') st.plotly_chart(fig_rsi) # MACD Plot fig_macd = go.Figure() fig_macd.add_trace(go.Scatter(x=df.index, y=df['MACD'], name='MACD')) fig_macd.add_trace(go.Scatter(x=df.index, y=df['MACD_Signal'], name='Signal')) fig_macd.update_layout(title='MACD Indicator') st.plotly_chart(fig_macd) # EMA Plot fig_ema = go.Figure() fig_ema.add_trace(go.Scatter(x=df.index, y=df['EMA9'], name='EMA9')) fig_ema.add_trace(go.Scatter(x=df.index, y=df['EMA21'], name='EMA21')) fig_ema.update_layout(title='EMA Indicators') st.plotly_chart(fig_ema) # OBV Plot fig_obv = go.Figure() fig_obv.add_trace(go.Scatter(x=df.index, y=df['OBV'], name='OBV')) fig_obv.update_layout(title='On-Balance Volume (OBV)') st.plotly_chart(fig_obv) # VWAP Plot fig_vwap = go.Figure() fig_vwap.add_trace(go.Scatter(x=df.index, y=df['Close'], name='Price')) fig_vwap.add_trace(go.Scatter(x=df.index, y=df['VWAP'], name='VWAP', line=dict(color='orange'))) fig_vwap.update_layout(title='Volume Weighted Average Price (VWAP)') st.plotly_chart(fig_vwap) with col2: # Stochastic Plot fig_stoch = go.Figure() fig_stoch.add_trace(go.Scatter(x=df.index, y=df['%K'], name='%K')) fig_stoch.add_trace(go.Scatter(x=df.index, y=df['%D'], name='%D')) fig_stoch.update_layout(title='Stochastic Oscillator') st.plotly_chart(fig_stoch) # Bollinger Bands fig_bb = go.Figure() fig_bb.add_trace(go.Scatter(x=df.index, y=df['BB_upper'], name='Upper Band')) fig_bb.add_trace(go.Scatter(x=df.index, y=df['Close'], name='Price')) fig_bb.add_trace(go.Scatter(x=df.index, y=df['BB_lower'], name='Lower Band')) fig_bb.update_layout(title='Bollinger Bands') st.plotly_chart(fig_bb) # Ichimoku Cloud fig_ichimoku = go.Figure() fig_ichimoku.add_trace(go.Scatter(x=df.index, y=df['ichimoku_a'], name='Senkou Span A')) fig_ichimoku.add_trace(go.Scatter(x=df.index, y=df['ichimoku_b'], name='Senkou Span B')) fig_ichimoku.add_trace(go.Scatter(x=df.index, y=df['ichimoku_base'], name='Kijun-sen')) fig_ichimoku.add_trace(go.Scatter(x=df.index, y=df['ichimoku_conversion'], name='Tenkan-sen')) fig_ichimoku.update_layout(title='Ichimoku Cloud') st.plotly_chart(fig_ichimoku) # SMA Plot fig_sma = go.Figure() fig_sma.add_trace(go.Scatter(x=df.index, y=df['Close'], name='Price')) fig_sma.add_trace(go.Scatter(x=df.index, y=df['SMA20'], name='SMA20')) fig_sma.add_trace(go.Scatter(x=df.index, y=df['SMA50'], name='SMA50')) fig_sma.add_trace(go.Scatter(x=df.index, y=df['SMA100'], name='SMA100')) fig_sma.add_trace(go.Scatter(x=df.index, y=df['SMA200'], name='SMA200')) fig_sma.update_layout(title='Simple Moving Averages (SMA)') st.plotly_chart(fig_sma) # ATR Plot fig_atr = go.Figure() fig_atr.add_trace(go.Scatter(x=df.index, y=df['ATR'], name='ATR')) fig_atr.update_layout(title='Average True Range (ATR)') st.plotly_chart(fig_atr) with tab3: st.subheader('Technical Analysis') # Combine all signals signals = generate_recommendation(df) enhanced_signals = enhanced_recommendation(df) all_signals = signals + enhanced_signals # Count buy and sell signals buy_signals = len([s for s in all_signals if s[0] == 'BUY']) sell_signals = len([s for s in all_signals if s[0] == 'SELL']) # Display recommendation summary col1, col2, col3 = st.columns(3) with col1: st.metric("Buy Signals", buy_signals) with col2: st.metric("Sell Signals", sell_signals) with col3: overall_rec = "BUY" if buy_signals > sell_signals else "SELL" if sell_signals > buy_signals else "HOLD" if overall_rec == "BUY": st.success(f"Overall: {overall_rec} 📈") elif overall_rec == "SELL": st.error(f"Overall: {overall_rec} 📉") else: st.warning(f"Overall: {overall_rec} ↔️") # Display detailed signals st.subheader("Detailed Signals:") for signal, reason in all_signals: if signal == 'BUY': st.success(f"🟢 {signal}: {reason}") else: st.error(f"🔴 {signal}: {reason}") # Technical Indicators Summary st.subheader("Technical Indicators Summary") # Create expandable section for current price levels with st.expander("📊 Current Price Levels", expanded=True): current_price = df['Close'].iloc[-1] prev_close = df['Close'].iloc[-2] price_change = ((current_price - prev_close) / prev_close) * 100 # Price information with colored indicators if price_change > 0: st.success(f"📈 Current Price: ${current_price:.2f} (+{price_change:.2f}%)") else: st.error(f"📉 Current Price: ${current_price:.2f} ({price_change:.2f}%)") # Display technical levels in a more organized way col1, col2, col3 = st.columns(3) with col1: st.info(f"💹 VWAP\n${df['VWAP'].iloc[-1]:.2f}") with col2: st.info(f"📏 ATR\n${df['ATR'].iloc[-1]:.2f}") with col3: st.info(f"📈 SMA50\n${df['SMA50'].iloc[-1]:.2f}") # Moving Averages Analysis with st.expander("📈 Moving Averages Analysis", expanded=True): sma_status = "Bullish" if (df['SMA20'].iloc[-1] > df['SMA50'].iloc[-1]) else "Bearish" sma_icon = "🟢" if sma_status == "Bullish" else "🔴" st.write(f"{sma_icon} SMA20 vs SMA50: {sma_status}") # Add more SMA comparisons sma_100_status = "Bullish" if (df['Close'].iloc[-1] > df['SMA100'].iloc[-1]) else "Bearish" sma_100_icon = "🟢" if sma_100_status == "Bullish" else "🔴" st.write(f"{sma_100_icon} Price vs SMA100: {sma_100_status}") sma_200_status = "Bullish" if (df['Close'].iloc[-1] > df['SMA200'].iloc[-1]) else "Bearish" sma_200_icon = "🟢" if sma_200_status == "Bullish" else "🔴" st.write(f"{sma_200_icon} Price vs SMA200: {sma_200_status}") # Volatility Analysis with st.expander("📊 Volatility Analysis", expanded=True): atr_avg = df['ATR'].mean() current_atr = df['ATR'].iloc[-1] atr_ratio = current_atr / atr_avg if atr_ratio > 1.5: volatility = "High" vol_icon = "⚠️" st.warning(f"{vol_icon} Volatility: {volatility}") elif atr_ratio < 0.5: volatility = "Low" vol_icon = "💤" st.info(f"{vol_icon} Volatility: {volatility}") else: volatility = "Normal" vol_icon = "✅" st.success(f"{vol_icon} Volatility: {volatility}") st.write(f"ATR Ratio: {atr_ratio:.2f}x average") # VWAP Analysis with st.expander("💹 VWAP Analysis", expanded=True): vwap_diff = ((df['Close'].iloc[-1] - df['VWAP'].iloc[-1]) / df['VWAP'].iloc[-1]) * 100 if vwap_diff > 0: st.success(f"🟢 Price is ABOVE VWAP by {abs(vwap_diff):.2f}%") else: st.error(f"🔴 Price is BELOW VWAP by {abs(vwap_diff):.2f}%") # Add risk warning st.warning("⚠️ Disclaimer: Above analysis is for AI Research and Learning purposes only. DO NOT make any investment decisions according to the displayed information and analysis result.") else: st.error('Error fetching stock data. Please check the symbol.') else: st.warning('Please enter a stock symbol.') with st.sidebar: # Add a horizontal line st.markdown("---") # Add text at the bottom st.warning("⚠️ Disclaimer: This is for AI Research and Learning purposes only. DO NOT make any investment decisions according to the displayed information and analysis result.") st.markdown("---") st.write("https://www.facebook.com/AIsparking")