import warnings warnings.filterwarnings("ignore") import yfinance as yf import pandas as pd import numpy as np import plotly.graph_objects as go from plotly.subplots import make_subplots from datetime import timedelta, datetime import feedparser import urllib.parse import html import re import time import gradio as gr # -------------------------- # ML libraries # -------------------------- HAS_PROPHET = False HAS_ARIMA = False HAS_SKLEARN = False HAS_TF = False try: from prophet import Prophet HAS_PROPHET = True except: Prophet = None try: from statsmodels.tsa.arima.model import ARIMA HAS_ARIMA = True except: ARIMA = None try: from sklearn.ensemble import RandomForestRegressor from sklearn.preprocessing import MinMaxScaler HAS_SKLEARN = True except: RandomForestRegressor = None; MinMaxScaler=None try: import tensorflow as tf from tensorflow import keras from tensorflow.keras import layers HAS_TF = True except: tf=None; keras=None; layers=None # -------------------------- # Data Utilities # -------------------------- def fetch_yahoo_data(ticker: str, period="6mo", interval="1d") -> pd.DataFrame: try: t = yf.Ticker(ticker) df = t.history(period=period, interval=interval) if df is None or df.empty: return pd.DataFrame() df = df.reset_index() df['Date'] = pd.to_datetime(df['Date']).dt.tz_localize(None) for col in ['Open','High','Low','Close','Volume']: if col in df.columns: df[col] = pd.to_numeric(df[col], errors='coerce') df = df.dropna(subset=['Date','Close']).reset_index(drop=True) return df except: return pd.DataFrame() def compute_indicators(df: pd.DataFrame) -> pd.DataFrame: df = df.copy().reset_index(drop=True) df['MA20'] = df['Close'].rolling(20).mean() df['MA50'] = df['Close'].rolling(50).mean() df['EMA12'] = df['Close'].ewm(span=12, adjust=False).mean() df['EMA26'] = df['Close'].ewm(span=26, adjust=False).mean() df['MACD'] = df['EMA12'] - df['EMA26'] df['Signal'] = df['MACD'].ewm(span=9, adjust=False).mean() df['BB_MID'] = df['Close'].rolling(20).mean() df['BB_STD'] = df['Close'].rolling(20).std(ddof=0).fillna(0) df['BB_UP'] = df['BB_MID'] + 2*df['BB_STD'] df['BB_LOW'] = df['BB_MID'] - 2*df['BB_STD'] delta = df['Close'].diff() up = delta.clip(lower=0) down = -delta.clip(upper=0) roll_up = up.rolling(14).mean() roll_down = down.rolling(14).mean() rs = roll_up / (roll_down + 1e-8) df['RSI'] = 100 - (100 / (1 + rs)) df['RSI'] = df['RSI'].clip(0,100).fillna(50) return df # -------------------------- # Plotting # -------------------------- def plot_advanced(df: pd.DataFrame, show_indicators: bool=True) -> go.Figure: fig = make_subplots(rows=3, cols=1, shared_xaxes=True, row_heights=[0.6,0.18,0.22], vertical_spacing=0.03) fig.add_trace(go.Candlestick(x=df['Date'], open=df['Open'], high=df['High'], low=df['Low'], close=df['Close'], name='Price'), row=1, col=1) if show_indicators: fig.add_trace(go.Scatter(x=df['Date'], y=df['MA20'], name='MA20', line=dict(width=1.4)), row=1, col=1) fig.add_trace(go.Scatter(x=df['Date'], y=df['MA50'], name='MA50', line=dict(width=1.4, dash='dash')), row=1, col=1) fig.add_trace(go.Scatter(x=df['Date'], y=df['BB_UP'], name='BB_UP', line=dict(width=1, dash='dot')), row=1, col=1) fig.add_trace(go.Scatter(x=df['Date'], y=df['BB_LOW'], name='BB_LOW', line=dict(width=1, dash='dot')), row=1, col=1) fig.add_trace(go.Bar(x=df['Date'], y=df['Volume'], name='Volume', marker_color='gray', opacity=0.3), row=2, col=1) fig.add_trace(go.Scatter(x=df['Date'], y=df['RSI'], name='RSI', line=dict(width=1.2)), row=3, col=1) fig.add_hline(y=70, line_dash='dot', row=3, col=1) fig.add_hline(y=30, line_dash='dot', row=3, col=1) fig.update_layout(template='plotly_dark', height=600, paper_bgcolor='#000000', plot_bgcolor='#000000') return fig # -------------------------- # Forecasting # -------------------------- def forecast_all(df: pd.DataFrame, periods: int = 30): forecasts = {} # Prophet if HAS_PROPHET: try: prophet_df = df[['Date','Close']].rename(columns={'Date':'ds','Close':'y'}).copy() m = Prophet(daily_seasonality=True, yearly_seasonality=True, weekly_seasonality=True) m.fit(prophet_df) future = m.make_future_dataframe(periods=periods, freq='D') fc = m.predict(future)[['ds','yhat']].rename(columns={'ds':'Date'}) forecasts['Prophet'] = fc except: pass # ARIMA if HAS_ARIMA: try: series = df.set_index('Date')['Close'].sort_index() series.index = pd.to_datetime(series.index) series = series.asfreq('D').ffill() model = ARIMA(series, order=(5,1,0)).fit() fc = model.forecast(steps=periods) dates = pd.date_range(start=series.index[-1]+timedelta(days=1), periods=periods) forecasts['ARIMA'] = pd.DataFrame({'Date':dates,'yhat':fc.values}) except: pass # Random Forest if HAS_SKLEARN: try: data = df[['Close']].copy() n_lags = 5 for lag in range(1,n_lags+1): data[f'lag_{lag}'] = data['Close'].shift(lag) data = data.dropna() X = data[[f'lag_{i}' for i in range(1,n_lags+1)]].values y = data['Close'].values model = RandomForestRegressor(n_estimators=200, random_state=42) model.fit(X,y) last_window = X[-1].tolist() preds=[] for _ in range(periods): p=float(model.predict([last_window])); preds.append(p); last_window=[p]+last_window[:-1] dates=pd.date_range(start=df['Date'].iloc[-1]+timedelta(days=1), periods=periods) forecasts['RandomForest'] = pd.DataFrame({'Date':dates,'yhat':preds}) except: pass # LSTM if HAS_TF and HAS_SKLEARN: try: values=df['Close'].values.astype('float32') n_lags=20 scaler=MinMaxScaler() scaled=scaler.fit_transform(values.reshape(-1,1)).flatten() X=[];y=[] for i in range(n_lags,len(scaled)): X.append(scaled[i-n_lags:i]); y.append(scaled[i]) X=np.array(X).reshape(-1,n_lags,1);y=np.array(y) tf.keras.backend.clear_session() model=keras.Sequential([layers.Input(shape=(n_lags,1)),layers.LSTM(64),layers.Dense(32,activation='relu'),layers.Dense(1)]) model.compile(optimizer='adam',loss='mse');model.fit(X,y,epochs=10,batch_size=16,verbose=0) last_window=list(scaled[-n_lags:]);preds_scaled=[] for _ in range(periods): x=np.array(last_window).reshape(1,n_lags,1);p=float(model.predict(x,verbose=0)[0,0]);preds_scaled.append(p);last_window=last_window[1:]+[p] preds = scaler.inverse_transform(np.array(preds_scaled).reshape(-1,1)).flatten().tolist() dates=pd.date_range(start=df['Date'].iloc[-1]+timedelta(days=1), periods=periods) forecasts['LSTM']=pd.DataFrame({'Date':dates,'yhat':preds}) except: pass return forecasts # -------------------------- # News # -------------------------- def get_company_name(ticker:str) -> str: try: return yf.Ticker(ticker).info.get('shortName','') except: return '' def _format_time(entry) -> str: try: if entry.get('published_parsed'): ts = time.mktime(entry.published_parsed) return datetime.fromtimestamp(ts).strftime("%a, %d %b %Y %H:%M:%S") if entry.get('published'): return str(entry.get('published')) except: pass return 'Unknown' def _extract_image(entry) -> str|None: mc=entry.get('media_content') or entry.get('mediaContents') if mc and isinstance(mc,(list,tuple)) and mc: return mc[0].get('url') mt=entry.get('media_thumbnail') if mt and isinstance(mt,(list,tuple)) and mt: return mt[0].get('url') summary = entry.get('summary','') or entry.get('description','') m=re.search(r']+src=["\']([^"\']+)["\']',summary or '',re.I) return m.group(1) if m else None def _resolve_link(entry, feed) -> str: for l in entry.get('links',[]): href=l.get('href') or '' if href and 'news.google.com' not in href: return href return entry.get('link','') def fetch_google_news(query:str,max_items:int=5): if not query: return [] encoded = urllib.parse.quote_plus(query) rss_url=f'https://news.google.com/rss/search?q={encoded}' feed=feedparser.parse(rss_url) items=[];seen=set() for entry in feed.entries: try: title=html.unescape(entry.get('title','')).strip() or '(no title)' published=_format_time(entry) src = feed.feed.get('title','') image=_extract_image(entry) link=_resolve_link(entry, feed) if not link or link in seen: continue seen.add(link) items.append({'title':title,'link':link,'source':src,'published':published,'image':image}) if len(items)>=max_items: break except: continue return items # -------------------------- # Gradio function # -------------------------- def analyze_stock(ticker, period, interval, show_indicators): df = fetch_yahoo_data(ticker, period, interval) if df.empty: return None, None, None, None df = compute_indicators(df) fig = plot_advanced(df, show_indicators) forecasts = forecast_all(df) news_articles = fetch_google_news(ticker) return fig, forecasts, df.tail(20), news_articles # -------------------------- # Gradio Interface # -------------------------- tickers_list = ["AAPL","MSFT","GOOG","AMZN","TSLA","META","NVDA","JPM","V","JNJ","WMT", "RELIANCE.NS","TCS.NS","INFY.NS","HDFCBANK.NS","ICICIBANK.NS","SBIN.NS", "BTC-USD","ETH-USD"] iface = gr.Interface( fn=analyze_stock, inputs=[ gr.Dropdown(tickers_list, label="Ticker"), gr.Dropdown(["1mo","3mo","6mo","1y","2y","5y","10y","max"], value="6mo", label="History Period"), gr.Dropdown(["1d","1wk","1mo"], value="1d", label="Interval"), gr.Checkbox(value=True, label="Show Indicators") ], outputs=[ gr.Plot(label="Price & Indicators"), gr.JSON(label="Forecasts"), gr.Dataframe(label="Recent Historical Data"), gr.JSON(label="News Articles") ], title="POINT.BLANK — Stock Analysis & News", description="Complete Stock Analysis & News Platform with technical indicators, AI forecasts, and latest news." ) iface.launch()