| | from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool |
| | import datetime |
| | import requests |
| | import pytz |
| | import yaml |
| | from tools.final_answer import FinalAnswerTool |
| | import yfinance as yf |
| | from textblob import TextBlob |
| | from Gradio_UI import GradioUI |
| | import os |
| |
|
| | |
| |
|
| | |
| |
|
| |
|
| | @tool |
| | def get_stock_price(symbol: str) -> str: |
| | """Fetches the latest stock price and historical trend for a given stock symbol. |
| | |
| | Args: |
| | symbol: The stock ticker symbol (e.g., 'AAPL' for Apple, 'TSLA' for Tesla). |
| | """ |
| | try: |
| | stock = yf.Ticker(symbol) |
| | hist = stock.history(period="1mo") |
| | latest_price = stock.history(period="1d")['Close'].iloc[-1] |
| | if hist['Close'].iloc[-1] > hist['Close'].iloc[0]: |
| | trend = "increasing" |
| | else: |
| | trend = "decreasing" |
| |
|
| | return f"The latest price of {symbol} is ${latest_price:.2f}. The trend over the last month is {trend}." |
| | except Exception as e: |
| | return f"Error fetching stock data for {symbol}: {str(e)}" |
| | @tool |
| | def stock_news_sentiment(symbol: str) -> str: |
| | """Fetches recent news articles on a stock and performs sentiment analysis. |
| | |
| | Args: |
| | symbol: The stock ticker symbol (e.g., 'AAPL' for Apple, 'TSLA' for Tesla). |
| | """ |
| | try: |
| | search_tool = DuckDuckGoSearchTool() |
| | query = f"{symbol} stock market news" |
| | news_results = search_tool.forward(query) |
| | |
| | if not news_results: |
| | return f"No recent news found for {symbol}." |
| | |
| | articles = news_results.split("\n\n")[:10] |
| | sentiment_scores = [] |
| | for article in articles: |
| | try: |
| | lines = article.split("\n") |
| | if len(lines) > 1: |
| | snippet = lines[1] |
| | else: |
| | snippet = article |
| |
|
| | analysis = TextBlob(snippet) |
| | sentiment_scores.append(analysis.sentiment.polarity) |
| | except Exception as inner_e: |
| | print(f"Skipping an article due to error: {inner_e}") |
| |
|
| | if not sentiment_scores: |
| | return f"No valid articles found for sentiment analysis of {symbol}." |
| |
|
| | avg_sentiment = sum(sentiment_scores) / len(sentiment_scores) |
| | sentiment = "positive" if avg_sentiment > 0 else "negative" if avg_sentiment < 0 else "neutral" |
| |
|
| | return f"The sentiment for {symbol} based on recent news is {sentiment}." |
| | |
| | except Exception as e: |
| | return f"Error analyzing sentiment for {symbol}: {str(e)}" |
| |
|
| | @tool |
| | def get_current_time_in_timezone(timezone: str) -> str: |
| | """A tool that fetches the current local time in a specified timezone. |
| | Args: |
| | timezone: A string representing a valid timezone (e.g., 'America/New_York'). |
| | """ |
| | try: |
| | tz = pytz.timezone(timezone) |
| | local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S") |
| | return f"The current local time in {timezone} is: {local_time}" |
| | except Exception as e: |
| | return f"Error fetching time for timezone '{timezone}': {str(e)}" |
| |
|
| |
|
| | final_answer = FinalAnswerTool() |
| |
|
| | |
| | |
| |
|
| | model = HfApiModel( |
| | max_tokens=2096, |
| | temperature=0.5, |
| | model_id='Qwen/Qwen2.5-Coder-32B-Instruct', |
| | custom_role_conversions=None, |
| | ) |
| |
|
| |
|
| | |
| | image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True) |
| |
|
| | with open("prompts.yaml", 'r') as stream: |
| | prompt_templates = yaml.safe_load(stream) |
| | |
| | agent = CodeAgent( |
| | model=model, |
| | tools=[final_answer, get_stock_price, stock_news_sentiment], |
| | max_steps=6, |
| | verbosity_level=1, |
| | grammar=None, |
| | planning_interval=None, |
| | name="Stock Investment Advisor", |
| | description="Analyzes stock trends and news sentiment to provide investment advice.", |
| | prompt_templates=prompt_templates |
| | ) |
| |
|
| |
|
| | GradioUI(agent).launch() |