from transformers import pipeline from alpaca_trade_api import REST import os from dotenv import load_dotenv from datetime import datetime import pandas as pd import matplotlib.pyplot as plt from datetime import date, timedelta from pydantic import BaseModel, Field from langchain.tools import BaseTool from typing import Optional, Type from langchain.tools import StructuredTool def sentimental_analysis_tools(): class AlpacaNewsFetcher: """ A class for fetching news articles related to a specific stock from Alpaca API. Attributes: - api_key (str): Alpaca API key for authentication. - api_secret (str): Alpaca API secret for authentication. - rest_client (alpaca_trade_api.REST): Alpaca REST API client. """ def __init__(self): """ Initializes the AlpacaNewsFetcher object. Args: - api_key (str): Alpaca API key for authentication. - api_secret (str): Alpaca API secret for authentication. """ load_dotenv() self.api_key = os.environ["ALPACA_API_KEY"] self.api_secret = os.environ["ALPACA_SECRET"] self.rest_client = REST(self.api_key, self.api_secret) #No of news articles to fetch for the input stock ticker. self.no_of_newsarticles_to_fetch = os.environ["NO_OF_NEWSARTICLES_TO_FETCH"] #No of days to fetch news articles for self.no_of_days = os.environ["NO_OF_DAYS_TO_FETCH_NEWS_ARTICLES"] def fetch_news(self, stockticker): """ Fetches news articles for a given stock symbol within a specified date range. Args: - stockticker (str): Stock symbol for which news articles are to be fetched (e.g., "AAPL"). Returns: - list: A list of dictionaries containing relevant information for each news article. """ #Date range for which to get the news start_date = date.today() end_date = date.today() - timedelta(self.no_of_days) news_articles = self.rest_client.get_news(stockticker, start_date, end_date, limit=self.no_of_newsarticles_to_fetch ) formatted_news = [] for article in news_articles: summary = article.summary title = article.headline timestamp = article.created_at relevant_info = { 'timestamp': timestamp, 'title': title, 'summary': summary } formatted_news.append(relevant_info) return formatted_news class NewsSentimentAnalysis: """ A class for sentiment analysis of news articles using the Transformers library. Attributes: - classifier (pipeline): Sentiment analysis pipeline from Transformers. """ def __init__(self): """ Initializes the NewsSentimentAnalysis object. """ self.classifier = pipeline('sentiment-analysis') def analyze_sentiment(self, news_article): """ Analyzes the sentiment of a given news article. Args: - news_article (dict): Dictionary containing 'summary', 'headline', and 'created_at' keys. Returns: - dict: A dictionary containing sentiment analysis results. """ summary = news_article['summary'] title = news_article['title'] timestamp = news_article['timestamp'] relevant_text = summary + title sentiment_result = self.classifier(relevant_text) analysis_result = { 'timestamp': timestamp, 'title': title, 'summary': summary, 'sentiment': sentiment_result } return analysis_result def plot_sentiment_graph(self, sentiment_analysis_result): """ Plots a sentiment analysis graph Args: - sentiment_analysis_result): (dict): Dictionary containing 'summary', 'headline', and 'created_at' keys. Returns: - dict: A dictionary containing sentiment analysis results. """ df = pd.DataFrame(sentiment_analysis_result) df['Timestamp'] = pd.to_datetime(df['Timestamp']) df['Date'] = df['Timestamp'].dt.date #Group by Date, sentiment value count grouped = df.groupby(by='Date')['Sentiment'].value_counts() grouped.plot.pie() def get_dominant_sentiment (self, sentiment_analysis_result): """ Returns overall sentiment, negative or positive or neutral depending on the count of negative sentiment vs positive sentiment Args: - sentiment_analysis_result): (dict): Dictionary containing 'summary', 'headline', and 'created_at' keys. Returns: - dict: A dictionary containing sentiment analysis results. """ df = pd.DataFrame(sentiment_analysis_result) df['Timestamp'] = pd.to_datetime(df['Timestamp']) df['Date'] = df['Timestamp'].dt.date #Group by Date, sentiment value count grouped = df.groupby(by='Date')['Sentiment'].value_counts() df = pd.DataFrame(list(grouped.items()), columns=['Sentiment', 'count']) df['date'] = df['Sentiment'].apply(lambda x: x[0]) df['sentiment'] = df['Sentiment'].apply(lambda x: x[1]) df.drop('Sentiment', axis=1, inplace=True) result = df.groupby('sentiment')['count'].sum().reset_index() # Determine the sentiment with the most count dominant_sentiment = result.loc[result['count'].idxmax()] return dominant_sentiment #Function to get the stock sentiment def get_stock_sentiment(stockticker: str): #Initialize AlpacaNewsFetcher, a class for fetching news articles related to a specific stock from Alpaca API. news_fetcher = AlpacaNewsFetcher() # Fetch news (contains - title of the news, timestamp and summary) for specified stocksticker news_data = news_fetcher.fetch_news(stockticker) # Initialize the NewsSentimentAnalysis object news_sentiment_analyzer = NewsSentimentAnalysis() analysis_result = [] # Assume 'news_data' is a list of news articles (each as a dictionary), analyze sentiment of each news for article in news_data: sentiment_analysis_result = news_sentiment_analyzer.analyze_sentiment(article) # Display sentiment analysis results print(f'Timestamp: {sentiment_analysis_result["timestamp"]}, ' f'Title: {sentiment_analysis_result["title"]}, ' f'Summary: {sentiment_analysis_result["summary"]}') print(f'Sentiment: {sentiment_analysis_result["sentiment"]}', '\n') result = { 'Timestamp': sentiment_analysis_result["timestamp"], 'News- Title:Summar': sentiment_analysis_result["title"] + sentiment_analysis_result["summary"], 'Sentiment': sentiment_analysis_result["sentiment"][0]['label'] } analysis_result.append(result) #Extracting timestamp of article and sentiment of article for graphing """ result_for_graph = { 'Timestamp': sentiment_analysis_result["timestamp"], 'Sentiment': sentiment_analysis_result["sentiment"][0]['label'] } analysis_result.append(result_for_graph) """ #Get dominant sentiment dominant_sentiment = news_sentiment_analyzer.get_dominant_sentiment(sentiment_analysis_result) #Build response string for news sentiment output_string = "" for result in analysis_result: output_string = output_string + f'{result["Timestamp"]} : {result["News- Title:Summary"]} : {result["Sentiment"]}' + '\n' final_result = { 'Sentiment-analysis-result' : output_string, 'Dominant-sentiment' : dominant_sentiment['sentiment'] } return final_result class StockSentimentCheckInput(BaseModel): """Input for Stock price check.""" stockticker: str = Field(..., description="Ticker symbol for stock or index") class StockSentimentAnalysisTool(BaseTool): name = "get_stock_sentiment" description = """Useful for finding sentiment of stock, based on published news articles. Fetches configured number of news items for the sentiment, determines sentiment of each news items and then returns List of sentiment analysit result & domainant sentiment of the news """ """Input for Stock sentiment analysis.""" stockticker: str = Field(..., description="Ticker symbol for stock or index") def _run(self, stockticker: str): # print("i'm running") sentiment_response = get_stock_sentiment(stockticker) print("++++++++++++++++++++++++++++++++++++++++++++++++++++++") print(str(sentiment_response)) print("++++++++++++++++++++++++++++++++++++++++++++++++++++++") return sentiment_response def _arun(self, stockticker: str): raise NotImplementedError("This tool does not support async") args_schema: Optional[Type[BaseModel]] = StockSentimentCheckInput tools_sentiment_analyst = [StructuredTool.from_function( func=StockSentimentAnalysisTool, args_schema=StockSentimentCheckInput, description="Function to get stock sentiment.", ) ] return tools_sentiment_analyst