community: Add PolygonTickerNews Tool (#17808)

Description:
In this PR, I am adding a PolygonTickerNews Tool, which can be used to
get the latest news for a given ticker / stock.

Twitter handle: [@virattt](https://twitter.com/virattt)

libs/community/langchain_community/agent_toolkits/polygon/toolkit.py

@@ -2,7 +2,7 @@ from typing import List
 
 from langchain_community.agent_toolkits.base import BaseToolkit
 from langchain_community.tools import BaseTool
-from langchain_community.tools.polygon import PolygonLastQuote
+from langchain_community.tools.polygon import PolygonLastQuote, PolygonTickerNews
 from langchain_community.utilities.polygon import PolygonAPIWrapper
 
 
@@ -18,7 +18,10 @@ class PolygonToolkit(BaseToolkit):
         tools = [
             PolygonLastQuote(
                 api_wrapper=polygon_api_wrapper,
-            )
+            ),
+            PolygonTickerNews(
+                api_wrapper=polygon_api_wrapper,
+            ),
         ]
         return cls(tools=tools)
 

libs/community/langchain_community/tools/__init__.py

@@ -510,6 +510,12 @@ def _import_polygon_tool_PolygonLastQuote() -> Any:
     return PolygonLastQuote
 
 
+def _import_polygon_tool_PolygonTickerNews() -> Any:
+    from langchain_community.tools.polygon.ticker_news import PolygonTickerNews
+
+    return PolygonTickerNews
+
+
 def _import_powerbi_tool_InfoPowerBITool() -> Any:
     from langchain_community.tools.powerbi.tool import InfoPowerBITool
 
@@ -957,6 +963,8 @@ def __getattr__(name: str) -> Any:
         return _import_plugin()
     elif name == "PolygonLastQuote":
         return _import_polygon_tool_PolygonLastQuote()
+    elif name == "PolygonTickerNews":
+        return _import_polygon_tool_PolygonTickerNews()
     elif name == "InfoPowerBITool":
         return _import_powerbi_tool_InfoPowerBITool()
     elif name == "ListPowerBITool":
@@ -1141,6 +1149,7 @@ __all__ = [
     "OpenWeatherMapQueryRun",
     "PubmedQueryRun",
     "PolygonLastQuote",
+    "PolygonTickerNews",
     "RedditSearchRun",
     "QueryCheckerTool",
     "QueryPowerBITool",

libs/community/langchain_community/tools/polygon/__init__.py

@@ -1,7 +1,9 @@
 """Polygon IO tools."""
 
 from langchain_community.tools.polygon.last_quote import PolygonLastQuote
+from langchain_community.tools.polygon.ticker_news import PolygonTickerNews
 
 __all__ = [
     "PolygonLastQuote",
+    "PolygonTickerNews",
 ]

libs/community/langchain_community/tools/polygon/ticker_news.py

@@ -0,0 +1,36 @@
+from typing import Optional, Type
+
+from langchain_core.callbacks import CallbackManagerForToolRun
+from langchain_core.pydantic_v1 import BaseModel
+from langchain_core.tools import BaseTool
+
+from langchain_community.utilities.polygon import PolygonAPIWrapper
+
+
+class Inputs(BaseModel):
+    """Inputs for Polygon's Ticker News API"""
+
+    query: str
+
+
+class PolygonTickerNews(BaseTool):
+    """Tool that gets the latest news for a given ticker from Polygon"""
+
+    mode: str = "get_ticker_news"
+    name: str = "polygon_ticker_news"
+    description: str = (
+        "A wrapper around Polygon's Ticker News API. "
+        "This tool is useful for fetching the latest news for a stock. "
+        "Input should be the ticker that you want to get the latest news for."
+    )
+    args_schema: Type[BaseModel] = Inputs
+
+    api_wrapper: PolygonAPIWrapper
+
+    def _run(
+        self,
+        query: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> str:
+        """Use the Polygon API tool."""
+        return self.api_wrapper.run(self.mode, ticker=query)

libs/community/langchain_community/utilities/polygon.py

@@ -39,8 +39,29 @@ class PolygonAPIWrapper(BaseModel):
 
         return data.get("results", None)
 
+    def get_ticker_news(self, ticker: str) -> Optional[dict]:
+        """
+        Get the most recent news articles relating to a stock ticker symbol,
+        including a summary of the article and a link to the original source.
+        """
+        url = (
+            f"{POLYGON_BASE_URL}v2/reference/news?"
+            f"ticker={ticker}&"
+            f"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))
+        elif mode == "get_ticker_news":
+            return json.dumps(self.get_ticker_news(ticker))
         else:
             raise ValueError(f"Invalid mode {mode} for Polygon API.")

libs/community/tests/unit_tests/tools/test_imports.py

@@ -85,6 +85,7 @@ EXPECTED_ALL = [
     "OpenWeatherMapQueryRun",
     "PubmedQueryRun",
     "PolygonLastQuote",
+    "PolygonTickerNews",
     "RedditSearchRun",
     "QueryCheckerTool",
     "QueryPowerBITool",

libs/community/tests/unit_tests/tools/test_public_api.py

@@ -87,6 +87,7 @@ _EXPECTED = [
     "OpenWeatherMapQueryRun",
     "PubmedQueryRun",
     "PolygonLastQuote",
+    "PolygonTickerNews",
     "RedditSearchRun",
     "QueryCheckerTool",
     "QueryPowerBITool",