Spaces:
Runtime error
Runtime error
File size: 6,156 Bytes
105b369 |
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 148 149 150 151 152 153 154 |
import json
from os import getenv
from typing import Optional, Literal, Any
from phi.tools import Toolkit
from phi.utils.log import logger
try:
from openbb import obb as openbb_app
except ImportError:
raise ImportError("`openbb` not installed. Please install using `pip install 'openbb[all]'`.")
class OpenBBTools(Toolkit):
def __init__(
self,
obb: Optional[Any] = None,
openbb_pat: Optional[str] = None,
provider: Literal["benzinga", "fmp", "intrinio", "polygon", "tiingo", "tmx", "yfinance"] = "yfinance",
stock_price: bool = True,
search_symbols: bool = False,
company_news: bool = False,
company_profile: bool = False,
price_targets: bool = False,
):
super().__init__(name="yfinance_tools")
self.obb = obb or openbb_app
try:
if openbb_pat or getenv("OPENBB_PAT"):
self.obb.account.login(pat=openbb_pat or getenv("OPENBB_PAT")) # type: ignore
except Exception as e:
logger.error(f"Error logging into OpenBB: {e}")
self.provider: Literal["benzinga", "fmp", "intrinio", "polygon", "tiingo", "tmx", "yfinance"] = provider
if stock_price:
self.register(self.get_stock_price)
if search_symbols:
self.register(self.search_company_symbol)
if company_news:
self.register(self.get_company_news)
if company_profile:
self.register(self.get_company_profile)
if price_targets:
self.register(self.get_price_targets)
def get_stock_price(self, symbol: str) -> str:
"""Use this function to get the current stock price for a stock symbol or list of symbols.
Args:
symbol (str): The stock symbol or list of stock symbols.
Eg: "AAPL" or "AAPL,MSFT,GOOGL"
Returns:
str: The current stock prices or error message.
"""
try:
result = self.obb.equity.price.quote(symbol=symbol, provider=self.provider).to_polars() # type: ignore
clean_results = []
for row in result.to_dicts():
clean_results.append(
{
"symbol": row.get("symbol"),
"last_price": row.get("last_price"),
"currency": row.get("currency"),
"name": row.get("name"),
"high": row.get("high"),
"low": row.get("low"),
"open": row.get("open"),
"close": row.get("close"),
"prev_close": row.get("prev_close"),
"volume": row.get("volume"),
"ma_50d": row.get("ma_50d"),
"ma_200d": row.get("ma_200d"),
}
)
return json.dumps(clean_results, indent=2, default=str)
except Exception as e:
return f"Error fetching current price for {symbol}: {e}"
def search_company_symbol(self, company_name: str) -> str:
"""Use this function to get a list of ticker symbols for a company.
Args:
company_name (str): The name of the company.
Returns:
str: A JSON string containing the ticker symbols.
"""
logger.debug(f"Search ticker for {company_name}")
result = self.obb.equity.search(company_name).to_polars() # type: ignore
clean_results = []
if len(result) > 0:
for row in result.to_dicts():
clean_results.append({"symbol": row.get("symbol"), "name": row.get("name")})
return json.dumps(clean_results, indent=2, default=str)
def get_price_targets(self, symbol: str) -> str:
"""Use this function to get consensus price target and recommendations for a stock symbol or list of symbols.
Args:
symbol (str): The stock symbol or list of stock symbols.
Eg: "AAPL" or "AAPL,MSFT,GOOGL"
Returns:
str: JSON containing consensus price target and recommendations.
"""
try:
result = self.obb.equity.estimates.consensus(symbol=symbol, provider=self.provider).to_polars() # type: ignore
return json.dumps(result.to_dicts(), indent=2, default=str)
except Exception as e:
return f"Error fetching company news for {symbol}: {e}"
def get_company_news(self, symbol: str, num_stories: int = 10) -> str:
"""Use this function to get company news for a stock symbol or list of symbols.
Args:
symbol (str): The stock symbol or list of stock symbols.
Eg: "AAPL" or "AAPL,MSFT,GOOGL"
num_stories (int): The number of news stories to return. Defaults to 10.
Returns:
str: JSON containing company news and press releases.
"""
try:
result = self.obb.news.company(symbol=symbol, provider=self.provider, limit=num_stories).to_polars() # type: ignore
clean_results = []
if len(result) > 0:
for row in result.to_dicts():
row.pop("images")
clean_results.append(row)
return json.dumps(clean_results[:num_stories], indent=2, default=str)
except Exception as e:
return f"Error fetching company news for {symbol}: {e}"
def get_company_profile(self, symbol: str) -> str:
"""Use this function to get company profile and overview for a stock symbol or list of symbols.
Args:
symbol (str): The stock symbol or list of stock symbols.
Eg: "AAPL" or "AAPL,MSFT,GOOGL"
Returns:
str: JSON containing company profile and overview.
"""
try:
result = self.obb.equity.profile(symbol=symbol, provider=self.provider).to_polars() # type: ignore
return json.dumps(result.to_dicts(), indent=2, default=str)
except Exception as e:
return f"Error fetching company profile for {symbol}: {e}"
|