community: Add PolygonFinancials Tool (#18324)

**Description:**
In this PR, I am adding a `PolygonFinancials` tool, which can be used to
get financials data for a given ticker. The financials data is the
fundamental data that is found in income statements, balance sheets, and
cash flow statements of public US companies.

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

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

@@ -2,7 +2,11 @@ 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, PolygonTickerNews
+from langchain_community.tools.polygon import (
+    PolygonFinancials,
+    PolygonLastQuote,
+    PolygonTickerNews,
+)
 from langchain_community.utilities.polygon import PolygonAPIWrapper
 
 
@@ -22,6 +26,9 @@ class PolygonToolkit(BaseToolkit):
             PolygonTickerNews(
                 api_wrapper=polygon_api_wrapper,
             ),
+            PolygonFinancials(
+                api_wrapper=polygon_api_wrapper,
+            ),
         ]
         return cls(tools=tools)
 

libs/community/langchain_community/tools/__init__.py

@@ -504,6 +504,12 @@ def _import_plugin() -> Any:
     return AIPluginTool
 
 
+def _import_polygon_tool_PolygonFinancials() -> Any:
+    from langchain_community.tools.polygon.financials import PolygonFinancials
+
+    return PolygonFinancials
+
+
 def _import_polygon_tool_PolygonLastQuote() -> Any:
     from langchain_community.tools.polygon.last_quote import PolygonLastQuote
 
@@ -961,6 +967,8 @@ def __getattr__(name: str) -> Any:
         return _import_playwright_NavigateTool()
     elif name == "AIPluginTool":
         return _import_plugin()
+    elif name == "PolygonFinancials":
+        return _import_polygon_tool_PolygonFinancials()
     elif name == "PolygonLastQuote":
         return _import_polygon_tool_PolygonLastQuote()
     elif name == "PolygonTickerNews":
@@ -1148,6 +1156,7 @@ __all__ = [
     "OpenAPISpec",
     "OpenWeatherMapQueryRun",
     "PubmedQueryRun",
+    "PolygonFinancials",
     "PolygonLastQuote",
     "PolygonTickerNews",
     "RedditSearchRun",

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

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

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

@@ -0,0 +1,38 @@
+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 Financials API"""
+
+    query: str
+
+
+class PolygonFinancials(BaseTool):
+    """Tool that gets the financials of a ticker from Polygon"""
+
+    mode: str = "get_financials"
+    name: str = "polygon_financials"
+    description: str = (
+        "A wrapper around Polygon's Stock Financials API. "
+        "This tool is useful for fetching fundamental financials from "
+        "balance sheets, income statements, and cash flow statements "
+        "for a stock ticker. The input should be the ticker that you want "
+        "to get the latest fundamental financial data 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

@@ -27,6 +27,25 @@ class PolygonAPIWrapper(BaseModel):
 
         return values
 
+    def get_financials(self, ticker: str) -> Optional[dict]:
+        """
+        Get fundamental financial data, which is found in balance sheets,
+        income statements, and cash flow statements for a given ticker.
+        """
+        url = (
+            f"{POLYGON_BASE_URL}vX/reference/financials?"
+            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 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}"
@@ -59,7 +78,9 @@ class PolygonAPIWrapper(BaseModel):
         return data.get("results", None)
 
     def run(self, mode: str, ticker: str) -> str:
-        if mode == "get_last_quote":
+        if mode == "get_financials":
+            return json.dumps(self.get_financials(ticker))
+        elif 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))

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

@@ -84,6 +84,7 @@ EXPECTED_ALL = [
     "OpenAPISpec",
     "OpenWeatherMapQueryRun",
     "PubmedQueryRun",
+    "PolygonFinancials",
     "PolygonLastQuote",
     "PolygonTickerNews",
     "RedditSearchRun",

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

@@ -86,6 +86,7 @@ _EXPECTED = [
     "OpenAPISpec",
     "OpenWeatherMapQueryRun",
     "PubmedQueryRun",
+    "PolygonFinancials",
     "PolygonLastQuote",
     "PolygonTickerNews",
     "RedditSearchRun",