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