community[minor]: Add Alpha Vantage API Tool (#14332)

### Description
This implementation adds functionality from the AlphaVantage API,
renowned for its comprehensive financial data. The class encapsulates
various methods, each dedicated to fetching specific types of financial
information from the API.

### Implemented Functions

- **`search_symbols`**: 
- Searches the AlphaVantage API for financial symbols using the provided
keywords.

- **`_get_market_news_sentiment`**: 
- Retrieves market news sentiment for a specified stock symbol from the
AlphaVantage API.

- **`_get_time_series_daily`**: 
- Fetches daily time series data for a specific symbol from the
AlphaVantage API.

- **`_get_quote_endpoint`**: 
- Obtains the latest price and volume information for a given symbol
from the AlphaVantage API.

- **`_get_time_series_weekly`**: 
- Gathers weekly time series data for a particular symbol from the
AlphaVantage API.

- **`_get_top_gainers_losers`**: 
- Provides details on top gainers, losers, and most actively traded
tickers in the US market from the AlphaVantage API.

  ### Issue: 
  - #11994 
  
### Dependencies: 
  - 'requests' library for HTTP requests. (import requests)
  - 'pytest' library for testing. (import pytest)

---------

Co-authored-by: Adam Badar <94140103+adam-badar@users.noreply.github.com>
Co-authored-by: Harrison Chase <hw.chase.17@gmail.com>
Co-authored-by: Bagatur <baskaryan@gmail.com>
This commit is contained in:
anshaneel
2024-03-29 20:44:01 -04:00
committed by GitHub
parent a9bc212bf2
commit 0884e5de7f
3 changed files with 344 additions and 25 deletions

View File

@@ -30,6 +30,117 @@ class AlphaVantageAPIWrapper(BaseModel):
)
return values
def search_symbols(self, keywords: str) -> Dict[str, Any]:
"""Make a request to the AlphaVantage API to search for symbols."""
response = requests.get(
"https://www.alphavantage.co/query/",
params={
"function": "SYMBOL_SEARCH",
"keywords": keywords,
"apikey": self.alphavantage_api_key,
},
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
raise ValueError(f"API Error: {data['Error Message']}")
return data
def _get_market_news_sentiment(self, symbol: str) -> Dict[str, Any]:
"""Make a request to the AlphaVantage API to get market news sentiment for a
given symbol."""
response = requests.get(
"https://www.alphavantage.co/query/",
params={
"function": "NEWS_SENTIMENT",
"symbol": symbol,
"apikey": self.alphavantage_api_key,
},
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
raise ValueError(f"API Error: {data['Error Message']}")
return data
def _get_time_series_daily(self, symbol: str) -> Dict[str, Any]:
"""Make a request to the AlphaVantage API to get the daily time series."""
response = requests.get(
"https://www.alphavantage.co/query/",
params={
"function": "TIME_SERIES_DAILY",
"symbol": symbol,
"apikey": self.alphavantage_api_key,
},
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
raise ValueError(f"API Error: {data['Error Message']}")
return data
def _get_quote_endpoint(self, symbol: str) -> Dict[str, Any]:
"""Make a request to the AlphaVantage API to get the
latest price and volume information."""
response = requests.get(
"https://www.alphavantage.co/query/",
params={
"function": "GLOBAL_QUOTE",
"symbol": symbol,
"apikey": self.alphavantage_api_key,
},
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
raise ValueError(f"API Error: {data['Error Message']}")
return data
def _get_time_series_weekly(self, symbol: str) -> Dict[str, Any]:
"""Make a request to the AlphaVantage API
to get the Weekly Time Series."""
response = requests.get(
"https://www.alphavantage.co/query/",
params={
"function": "TIME_SERIES_WEEKLY",
"symbol": symbol,
"apikey": self.alphavantage_api_key,
},
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
raise ValueError(f"API Error: {data['Error Message']}")
return data
def _get_top_gainers_losers(self) -> Dict[str, Any]:
"""Make a request to the AlphaVantage API to get the top gainers, losers,
and most actively traded tickers in the US market."""
response = requests.get(
"https://www.alphavantage.co/query/",
params={
"function": "TOP_GAINERS_LOSERS",
"apikey": self.alphavantage_api_key,
},
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
raise ValueError(f"API Error: {data['Error Message']}")
return data
def _get_exchange_rate(
self, from_currency: str, to_currency: str
) -> Dict[str, Any]: