Spaces:
Sleeping
Sleeping
File size: 4,989 Bytes
9419f40 | 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 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | import yfinance as yf
from typing import Dict, Any, List
from base_tool import BaseTool
from registry import ToolRegistry
import logging
logger = logging.getLogger("FinancialAgent")
# backend/tools.py
# backend/tools.py
from typing import Dict, Any, List
# import BaseTool if it's in a separate file, or assume it's available
class CalculatorTool(BaseTool):
name = "calculator"
description = "Perform basic arithmetic operations. Use this for calculating differences, ratios, or percentages."
categories = ["utils"]
def get_schema(self) -> Dict[str, Any]:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": {
"operation": {
"type": "string",
"enum": ["add", "subtract", "multiply", "divide"],
"description": "The math operation to perform."
},
"x": {
"type": "number",
"description": "The first number."
},
"y": {
"type": "number",
"description": "The second number."
}
},
"required": ["operation", "x", "y"]
}
}
}
def run(self, operation: str, x: float, y: float) -> Any:
try:
# Ensure numbers are floats (in case string is passed)
x, y = float(x), float(y)
if operation == "add":
return {"result": x + y}
elif operation == "subtract":
return {"result": x - y}
elif operation == "multiply":
return {"result": x * y}
elif operation == "divide":
if y == 0:
return {"error": "Error: Division by zero"}
return {"result": x / y}
else:
return {"error": f"Unknown operation: {operation}"}
except Exception as e:
return {"error": f"Math execution failed: {str(e)}"}
class StockPriceTool(BaseTool):
name = "get_stock_price"
description = "Get the current price of a stock using its Ticker Symbol."
categories = ["finance"]
def get_schema(self) -> Dict[str, Any]:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": {
"ticker_symbol": {"type": "string", "description": "The stock ticker (e.g., AAPL)"}
},
"required": ["ticker_symbol"]
}
}
}
def run(self, ticker_symbol: str) -> Dict:
try:
ticker = yf.Ticker(ticker_symbol)
info = ticker.fast_info
return {"ticker": ticker_symbol, "price": info.last_price, "currency": info.currency}
except Exception as e:
return {"error": str(e)}
class CompanyNewsTool(BaseTool):
name = "get_company_news"
description = "Get the latest news summaries about a company."
categories = ["news"]
def get_schema(self) -> Dict[str, Any]:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": {
"ticker_symbol": {"type": "string", "description": "The stock ticker"},
"num_stories": {"type": "integer", "description": "Number of stories"}
},
"required": ["ticker_symbol", "num_stories"]
}
}
}
def run(self, ticker_symbol: str, num_stories: int = 5) -> Dict:
try:
ticker = yf.Ticker(ticker_symbol)
news = ticker.news[:num_stories] if ticker.news else []
if len(news) == 0:
return {"error": f'HTTP Error 404: ${ticker_symbol}: possibly delisted; Quote not found for symbol'}
return {"ticker": ticker_symbol, "news": [n['content']['title'] for n in news], "storiesFetched": len(news)}
except Exception as e:
return {"error": str(e)}
def initialize_registry() -> ToolRegistry:
registry = ToolRegistry()
tools_list = [
StockPriceTool(),
CompanyNewsTool(),
CalculatorTool()
]
for tool in tools_list:
registry.register(tool)
return registry
if __name__ == "__main__":
pass |