mirror of
https://github.com/hwchase17/langchain.git
synced 2025-09-05 04:55:14 +00:00
community: Add PolygonAPIWrapper and get_last_quote endpoint (#15971)
- **Description:** Added a `PolygonAPIWrapper` and an initial `get_last_quote` endpoint, which allows us to get the last price quote for a given `ticker`. Once merged, I can add a Polygon tool in `tools/` for agents to use. - **Twitter handle:** [@virattt](https://twitter.com/virattt) The Polygon.io Stocks API provides REST endpoints that let you query the latest market data from all US stock exchanges.
This commit is contained in:
46
libs/community/langchain_community/utilities/polygon.py
Normal file
46
libs/community/langchain_community/utilities/polygon.py
Normal file
@@ -0,0 +1,46 @@
|
||||
"""
|
||||
Util that calls several of Polygon's stock market REST APIs.
|
||||
Docs: https://polygon.io/docs/stocks/getting-started
|
||||
"""
|
||||
import json
|
||||
from typing import Dict, Optional
|
||||
|
||||
import requests
|
||||
from langchain_core.pydantic_v1 import BaseModel, root_validator
|
||||
from langchain_core.utils import get_from_dict_or_env
|
||||
|
||||
POLYGON_BASE_URL = "https://api.polygon.io/"
|
||||
|
||||
|
||||
class PolygonAPIWrapper(BaseModel):
|
||||
"""Wrapper for Polygon API."""
|
||||
|
||||
polygon_api_key: Optional[str] = None
|
||||
|
||||
@root_validator()
|
||||
def validate_environment(cls, values: Dict) -> Dict:
|
||||
"""Validate that api key in environment."""
|
||||
polygon_api_key = get_from_dict_or_env(
|
||||
values, "polygon_api_key", "POLYGON_API_KEY"
|
||||
)
|
||||
values["polygon_api_key"] = polygon_api_key
|
||||
|
||||
return values
|
||||
|
||||
def get_last_quote(self, ticker: str) -> Optional[dict]:
|
||||
"""Get the most recent National Best Bid and Offer (Quote) for a ticker."""
|
||||
url = f"{POLYGON_BASE_URL}v2/last/nbbo/{ticker}?apiKey={self.polygon_api_key}"
|
||||
response = requests.get(url)
|
||||
data = response.json()
|
||||
|
||||
status = data.get("status", None)
|
||||
if status != "OK":
|
||||
raise ValueError(f"API Error: {data}")
|
||||
|
||||
return data.get("results", None)
|
||||
|
||||
def run(self, mode: str, ticker: str) -> str:
|
||||
if mode == "get_last_quote":
|
||||
return json.dumps(self.get_last_quote(ticker))
|
||||
else:
|
||||
raise ValueError(f"Invalid mode {mode} for Polygon API.")
|
Reference in New Issue
Block a user