pointblank.py

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'<img[^>]+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()