""" title: Yahoo Finance Stock and Crypto Analyzer with Historical Data description: A comprehensive stock and cryptocurrency analysis tool using Yahoo Finance data, including historical price analysis. author: AI Assistant author_url: https://github.com/your-github-username github: https://github.com/your-github-username/yahoo-finance-analyzer funding_url: https://github.com/open-webui version: 0.3.0 license: MIT requirements: yfinance """ import yfinance as yf import requests import aiohttp import asyncio from bs4 import BeautifulSoup from transformers import pipeline from pydantic import BaseModel from datetime import datetime, timedelta from typing import Dict, Any, List, Union, Callable, Awaitable from functools import lru_cache import pandas as pd import numpy as np # Helper functions @lru_cache(maxsize=128) def _get_sentiment_model(): return pipeline("sentiment-analysis", model="ProsusAI/finbert") async def _async_web_scrape(session: aiohttp.ClientSession, url: str) -> str: try: async with session.get(url) as response: response.raise_for_status() content = await response.text() soup = BeautifulSoup(content, "html.parser") return soup.get_text()[:1000] # Limit to first 1000 characters except Exception as e: return f"Error scraping web page: {str(e)}" async def _async_sentiment_analysis( content: str, model ) -> Dict[str, Union[str, float]]: result = model(content)[0] return {"sentiment": result["label"], "confidence": result["score"]} # Main functions def _get_asset_info(ticker: str) -> Dict[str, Any]: asset = yf.Ticker(ticker) info = asset.info if info.get("quoteType") == "CRYPTOCURRENCY": return { "name": info.get("name", "N/A"), "symbol": info.get("symbol", "N/A"), "market_cap": info.get("marketCap", "N/A"), "circulating_supply": info.get("circulatingSupply", "N/A"), "max_supply": info.get("maxSupply", "N/A"), "24h_volume": info.get("volume24Hr", "N/A"), "7d_change": info.get("sevenDayAvgChangePercent", "N/A"), "is_crypto": True, } else: return { "name": info.get("longName", "N/A"), "symbol": info.get("symbol", "N/A"), "sector": info.get("sector", "N/A"), "industry": info.get("industry", "N/A"), "market_cap": info.get("marketCap", "N/A"), "pe_ratio": info.get("trailingPE", "N/A"), "dividend_yield": info.get("dividendYield", "N/A"), "beta": info.get("beta", "N/A"), "52_week_high": info.get("fiftyTwoWeekHigh", "N/A"), "52_week_low": info.get("fiftyTwoWeekLow", "N/A"), "is_crypto": False, } def _get_current_price(ticker: str) -> Dict[str, float]: asset = yf.Ticker(ticker) data = asset.history(period="1d") return { "current_price": data["Close"].iloc[-1], "open": data["Open"].iloc[-1], "high": data["High"].iloc[-1], "low": data["Low"].iloc[-1], "volume": data["Volume"].iloc[-1], } def _get_asset_news(ticker: str) -> List[Dict[str, str]]: url = f"https://finance.yahoo.com/quote/{ticker}/news" response = requests.get(url) soup = BeautifulSoup(response.text, "html.parser") news_items = soup.find_all("h3", class_="Mb(5px)") return [ { "title": item.text, "url": "https://finance.yahoo.com" + item.find("a")["href"], } for item in news_items[:5] ] def _get_historical_data(ticker: str, period: str = "1y") -> pd.DataFrame: asset = yf.Ticker(ticker) history = asset.history(period=period) return history def _analyze_historical_data(df: pd.DataFrame) -> Dict[str, Any]: analysis = {} # Calculate simple moving averages df["SMA50"] = df["Close"].rolling(window=50).mean() df["SMA200"] = df["Close"].rolling(window=200).mean() # Calculate price change and volatility df["Daily_Return"] = df["Close"].pct_change() analysis["price_change"] = (df["Close"].iloc[-1] / df["Close"].iloc[0] - 1) * 100 analysis["volatility"] = ( df["Daily_Return"].std() * np.sqrt(252) * 100 ) # Annualized volatility # Trend analysis current_price = df["Close"].iloc[-1] analysis["trend_sma50"] = ( "Bullish" if current_price > df["SMA50"].iloc[-1] else "Bearish" ) analysis["trend_sma200"] = ( "Bullish" if current_price > df["SMA200"].iloc[-1] else "Bearish" ) # Highest and lowest prices analysis["highest_price"] = df["High"].max() analysis["lowest_price"] = df["Low"].min() # Average volume analysis["avg_volume"] = df["Volume"].mean() return analysis async def _async_gather_asset_data(ticker: str) -> Dict[str, Any]: asset_info = _get_asset_info(ticker) current_price = _get_current_price(ticker) news_items = _get_asset_news(ticker) historical_data = _get_historical_data(ticker) historical_analysis = _analyze_historical_data(historical_data) async with aiohttp.ClientSession() as session: scrape_tasks = [_async_web_scrape(session, item["url"]) for item in news_items] contents = await asyncio.gather(*scrape_tasks) model = _get_sentiment_model() sentiment_tasks = [ _async_sentiment_analysis(content, model) for content in contents if content ] sentiments = await asyncio.gather(*sentiment_tasks) sentiment_results = [ { "url": news_items[i]["url"], "title": news_items[i]["title"], "sentiment": sentiment["sentiment"], "confidence": sentiment["confidence"], } for i, sentiment in enumerate(sentiments) if contents[i] ] return { "asset_info": asset_info, "current_price": current_price, "sentiments": sentiment_results, "historical_analysis": historical_analysis, } def _get_rsi_macd(symbol="BTC-USD", period="1mo", interval="1d"): ticker = yf.Ticker(symbol) data = ticker.history(period=period, interval=interval) close_prices = data["Close"] # Calculate RSI delta = close_prices.diff() gain = delta.where(delta > 0, 0) loss = -delta.where(delta < 0, 0) avg_gain = gain.rolling(window=14).mean() avg_loss = loss.rolling(window=14).mean() rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) # Calculate MACD ema_12 = close_prices.ewm(span=12, adjust=False).mean() ema_26 = close_prices.ewm(span=26, adjust=False).mean() macd_line = ema_12 - ema_26 signal_line = macd_line.ewm(span=9, adjust=False).mean() macd_hist = macd_line - signal_line # Format output output = f"Results for {symbol} over period {period} with interval {interval}:\n\n" output += f"Latest RSI: {rsi.iloc[-1]:.2f}\n" output += f"Latest MACD: {macd_line.iloc[-1]:.2f}\n" output += f"Latest MACD Signal: {signal_line.iloc[-1]:.2f}\n" output += f"Latest MACD Histogram: {macd_hist.iloc[-1]:.2f}\n\n" output += "RSI Evolution:\n" output += f"{rsi.tail(10).to_string(header=False)}\n\n" output += "MACD Evolution:\n" output_df = pd.DataFrame( {"MACD": macd_line, "Signal": signal_line, "Histogram": macd_hist}, index=macd_line.index, ) output += f"{output_df.tail(10).to_string(index=True)}" return output def _compile_report(data: Dict[str, Any], price_history: str, tech_data: str) -> str: info = data["asset_info"] price = data["current_price"] historical = data["historical_analysis"] if info["is_crypto"]: report = """ Cryptocurrency Analysis Report for {name} ({symbol}) Basic Information: Market Cap: ${market_cap:,.0f} Circulating Supply: {circulating_supply:,.0f} Current Trading Information: Current Price: ${current_price:.2f} 24h Range: ${low:.2f} - ${high:.2f} 24h Volume: {volume:,.0f} Historical Analysis: Price Change (1 year): {price_change:.2f}% Annualized Volatility: {volatility:.2f}% 50-day SMA Trend: {trend_sma50} 200-day SMA Trend: {trend_sma200} Highest Price (1 year): ${highest_price:.2f} Lowest Price (1 year): ${lowest_price:.2f} Average Daily Volume: {avg_volume:,.0f} Price history: {price_history} Technical data: {tech_data} """.format( name=info["name"], symbol=info["symbol"], market_cap=info["market_cap"], circulating_supply=info["circulating_supply"], current_price=price["current_price"], low=price["low"], high=price["high"], volume=price["volume"], price_change=historical["price_change"], volatility=historical["volatility"], trend_sma50=historical["trend_sma50"], trend_sma200=historical["trend_sma200"], highest_price=historical["highest_price"], lowest_price=historical["lowest_price"], avg_volume=historical["avg_volume"], price_history=price_history, tech_data=tech_data, ) else: report = """ Stock Analysis Report for {name} ({symbol}) Basic Information: Sector: {sector} Industry: {industry} Market Cap: ${market_cap:,.0f} Current Trading Information: Current Price: ${current_price:.2f} Day's Range: ${low:.2f} - ${high:.2f} Volume: {volume:,.0f} Key Financial Metrics: P/E Ratio: {pe_ratio} Dividend Yield: {dividend_yield:.2%} Beta: {beta} 52 Week Range: ${week_low:.2f} - ${week_high:.2f} Historical Analysis: Price Change (1 year): {price_change:.2f}% Annualized Volatility: {volatility:.2f}% 50-day SMA Trend: {trend_sma50} 200-day SMA Trend: {trend_sma200} Highest Price (1 year): ${highest_price:.2f} Lowest Price (1 year): ${lowest_price:.2f} Average Daily Volume: {avg_volume:,.0f} """.format( name=info["name"], symbol=info["symbol"], sector=info["sector"], industry=info["industry"], market_cap=info["market_cap"], current_price=price["current_price"], low=price["low"], high=price["high"], volume=price["volume"], pe_ratio=info["pe_ratio"], dividend_yield=info["dividend_yield"], beta=info["beta"], week_low=info["52_week_low"], week_high=info["52_week_high"], price_change=historical["price_change"], volatility=historical["volatility"], trend_sma50=historical["trend_sma50"], trend_sma200=historical["trend_sma200"], highest_price=historical["highest_price"], lowest_price=historical["lowest_price"], avg_volume=historical["avg_volume"], ) report += "Recent News and Sentiment Analysis:\n" for item in data["sentiments"]: report += """ Title: {title} URL: {url} Sentiment: {sentiment} (Confidence: {confidence:.2f}) """.format( title=item["title"], url=item["url"], sentiment=item["sentiment"], confidence=item["confidence"], ) return report class Tools: class UserValves(BaseModel): pass def __init__(self): pass async def analyze_asset( self, ticker: str, __user__: dict = {}, __event_emitter__: Callable[[Any], Awaitable[None]] = None, ) -> str: """ Perform a comprehensive analysis and compile a detailed report for a given stock or cryptocurrency ticker using Yahoo Finance data. :param ticker: The asset ticker symbol (e.g., "AAPL" for Apple Inc. or "BTC-USD" for Bitcoin). :return: A comprehensive analysis report of the asset as a formatted string. """ await __event_emitter__( { "type": "status", "data": {"description": "Initializing analysis", "done": False}, } ) await __event_emitter__( { "type": "status", "data": {"description": "Retrieving asset data", "done": False}, } ) price = _get_historical_data(ticker) tech_data = _get_rsi_macd(ticker) data = await _async_gather_asset_data(ticker) await __event_emitter__( { "type": "status", "data": {"description": "Compiling asset report", "done": False}, } ) report = _compile_report(data, price, tech_data) last_price = data["current_price"]["current_price"] asset_type = "cryptocurrency" if data["asset_info"]["is_crypto"] else "stock" await __event_emitter__( { "type": "status", "data": { "description": "Finished creating report for {0} - latest price: ${1:.4f}".format( ( "cryptocurrency" if data["asset_info"]["is_crypto"] else "stock" ), last_price, ), "done": True, }, } ) return report