Source code for langchain_community.utilities.alpha_vantage

"""Util that calls AlphaVantage for Currency Exchange Rate."""

from typing import Any, Dict, List, Optional

import requests
from langchain_core.pydantic_v1 import BaseModel, root_validator
from langchain_core.utils import get_from_dict_or_env


[docs]class AlphaVantageAPIWrapper(BaseModel): """Wrapper for AlphaVantage API for Currency Exchange Rate. Docs for using: 1. Go to AlphaVantage and sign up for an API key 2. Save your API KEY into ALPHAVANTAGE_API_KEY env variable """ alphavantage_api_key: Optional[str] = None class Config: extra = "forbid" @root_validator(pre=True) def validate_environment(cls, values: Dict) -> Dict: """Validate that api key exists in environment.""" values["alphavantage_api_key"] = get_from_dict_or_env( values, "alphavantage_api_key", "ALPHAVANTAGE_API_KEY" ) return values
[docs] 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]: """Make a request to the AlphaVantage API to get the exchange rate.""" response = requests.get( "https://www.alphavantage.co/query/", params={ "function": "CURRENCY_EXCHANGE_RATE", "from_currency": from_currency, "to_currency": to_currency, "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 @property def standard_currencies(self) -> List[str]: return ["USD", "EUR", "GBP", "JPY", "CHF", "CAD", "AUD", "NZD"]
[docs] def run(self, from_currency: str, to_currency: str) -> str: """Get the current exchange rate for a specified currency pair.""" if to_currency not in self.standard_currencies: from_currency, to_currency = to_currency, from_currency data = self._get_exchange_rate(from_currency, to_currency) return data["Realtime Currency Exchange Rate"]