File size: 3,054 Bytes
cc9a0e5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
531bb0a
cc9a0e5
531bb0a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
from typing import Any, Optional
from smolagents.tools import Tool
import yfinance as yf

class StockPriceTool(Tool):
    """
    A tool that fetches the latest stock price for a given stock ticker.
    This class extends the Tool base class from smolagents.
    """

    name = "stock_price_tool"
    description = "Fetches the latest stock price for a given stock ticker using yfinance."
    inputs = {'ticker_symbol': {'type': 'any', 'description': 'Stock Ticker Symbol'}}
    output_type = "Any"

    def _call(self, ticker_symbol: str) -> Any:
        """
        Fetches the latest stock price for the given ticker symbol.
        Args:
            ticker_symbol (str): The stock ticker symbol (e.g., "AAPL" for Apple, "GOOGL" for Google).
        Returns:
            float: The current stock price of the given ticker symbol.
            None: If an error occurs or data is unavailable.
        Example:
            >>> StockPriceTool()._call("AAPL")
            150.25
        """
        try:
            # Fetch the stock data using yfinance
            stock = yf.Ticker(ticker_symbol)
            
            # Get the current price (the last close price)
            stock_data = stock.history(period="1d")
            if stock_data.empty:
                return None
            
            current_price = stock_data['Close'].iloc[0]
            return float(current_price)
        except Exception as e:
            # If there's an error, log it and return None
            print(f"Error fetching price for {ticker_symbol}: {str(e)}")
            return None

    def __init__(self, *args, **kwargs):
        self.is_initialized = False

    def forward_stage(self, ticker_symbol: str) -> Optional[Any]:
        """
        Forward stage to handle further analysis or additional steps after fetching the stock price.
        Args:
            ticker_symbol (str): The stock ticker symbol.
        Returns:
            Any: Additional stock analysis or further actions based on the stock data.
        Example:
            >>> StockPriceTool().forward_stage("AAPL")
            {"price": 150.25, "price_change": -0.5}
        """
        # Fetch the current stock price
        current_price = self._call(ticker_symbol)
        
        if current_price is None:
            return {"error": f"Unable to fetch data for {ticker_symbol}"}

        # For forward analysis, we can fetch historical data for further insights
        stock = yf.Ticker(ticker_symbol)
        stock_data = stock.history(period="5d")  # Fetch last 5 days of stock data
        
        if stock_data.empty:
            return {"error": f"No historical data available for {ticker_symbol}"}

        # Calculate the price change over the last 5 days
        price_change = current_price - stock_data['Close'].iloc[-2]
        price_change_percent = (price_change / stock_data['Close'].iloc[-2]) * 100

        return {
            "price": current_price,
            "price_change": price_change,
            "price_change_percent": price_change_percent
        }