| | from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool |
| | import datetime |
| | import requests |
| | import pytz |
| | import yaml |
| | import yfinance as yf |
| | from ta.momentum import RSIIndicator, StochasticOscillator |
| | from ta.trend import MACD |
| | from ta.volume import volume_weighted_average_price |
| | from tools.final_answer import FinalAnswerTool |
| |
|
| |
|
| | from Gradio_UI import GradioUI |
| |
|
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| |
|
| | @tool |
| | def get_stock_price(ticker: str) -> Union[Dict, str]: |
| | """ |
| | A tool that fetches the historical stock price data and technical indicators for a given ticker. |
| | Args: |
| | ticker: A string representing a stocke ticker name (e.g AAPL) |
| | """ |
| |
|
| | try: |
| | data = yf.download( |
| | ticker, |
| | start=dt.datetime.now() - dt.timedelta(weeks=24 * 3), |
| | end=dt.datetime.now(), |
| | interval="1wk", |
| | ) |
| | df = data.copy() |
| | data.reset_index(inplace=True) |
| | data.Date = data.Date.astype(str) |
| |
|
| | indicators = {} |
| |
|
| | rsi_series = RSIIndicator(df["Close"], window=14).rsi().iloc[-12:] |
| | indicators["RSI"] = { |
| | date.strftime("%Y-%m-%d"): int(value) |
| | for date, value in rsi_series.dropna().to_dict().items() |
| | } |
| |
|
| | stochastic_series = ( |
| | StochasticOscillator(df["High"], df["Low"], df["Close"], window=14) |
| | .stoch() |
| | .iloc[-12:] |
| | ) |
| | indicators["Stochastic Oscillator"] = { |
| | date.strftime("%Y-%m-%d"): int(value) |
| | for date, value in stochastic_series.dropna().to_dict().items() |
| | } |
| |
|
| | macd = MACD(df["Close"]) |
| | macd_series = macd.macd().iloc[-12:] |
| | indicators["MACD"] = { |
| | date.strftime("%Y-%m-%d"): int(value) |
| | for date, value in macd_series.to_dict().items() |
| | } |
| |
|
| | macd_signal_series = macd.macd_signal().iloc[-12:] |
| | indicators["MACD Signal"] = { |
| | date.strftime("%Y-%m-%d"): int(value) |
| | for date, value in macd_signal_series.to_dict().items() |
| | } |
| |
|
| | vwap_series = volume_weighted_average_price( |
| | df["High"], df["Low"], df["Close"], df["Volume"] |
| | ).iloc[-12:] |
| | indicators["vwap"] = { |
| | date.strftime("%Y-%m-%d"): int(value) |
| | for date, value in vwap_series.to_dict().items() |
| | } |
| |
|
| | return {"stock_price": data.to_dict(orient="records"), "indicators": indicators} |
| | except Exception as e: |
| | return f"Error fetching price data: {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], |
| | max_steps=6, |
| | verbosity_level=1, |
| | grammar=None, |
| | planning_interval=None, |
| | name=None, |
| | description=None, |
| | prompt_templates=prompt_templates |
| | ) |
| |
|
| |
|
| | GradioUI(agent).launch() |