Improve YahooFinanceNewsTool to resolve tickers automatically

libs/community/langchain_community/tools/yahoo_finance_news.py

@@ -1,19 +1,28 @@
-from typing import Iterable, Optional, Type
+import re
+from typing import Optional, Type
 
 from langchain_core.callbacks import CallbackManagerForToolRun
-from langchain_core.documents import Document
 from langchain_core.tools import BaseTool
 from pydantic import BaseModel, Field
 from requests.exceptions import HTTPError, ReadTimeout
 from urllib3.exceptions import ConnectionError
 
-from langchain_community.document_loaders.web_base import WebBaseLoader
-
 
 class YahooFinanceNewsInput(BaseModel):
     """Input for the YahooFinanceNews tool."""
 
-    query: str = Field(description="company ticker query to look up")
+    query: str = Field(
+        description=(
+            "Ticker symbol to look up (e.g., 'MSFT'). "
+            "If you only have the company name, convert it to the ticker before "
+            "calling."
+        )
+    )
+
+
+def _looks_like_ticker(text: str) -> bool:
+    """Return True if the text resembles a stock ticker."""
+    return bool(re.fullmatch(r"[A-Za-z0-9.\-]{1,7}", text.strip()))
 
 
 class YahooFinanceNewsTool(BaseTool):
@@ -23,14 +32,58 @@ class YahooFinanceNewsTool(BaseTool):
     description: str = (
         "Useful for when you need to find financial news "
         "about a public company. "
-        "Input should be a company ticker. "
-        "For example, AAPL for Apple, MSFT for Microsoft."
+        "Input must be a ticker symbol (for example, AAPL for Apple, MSFT for "
+        "Microsoft). Convert company names to tickers before invoking this tool."
     )
     top_k: int = 10
     """The number of results to return."""
 
     args_schema: Type[BaseModel] = YahooFinanceNewsInput
 
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._last_resolved_symbol: Optional[str] = None
+
+    def _resolve_symbol(self, query: str) -> Optional[str]:
+        """Resolve a free-form query to a ticker symbol using Yahoo Finance search."""
+        try:
+            from yfinance.search import Search
+        except (
+            ImportError
+        ) as exc:  # pragma: no cover - yfinance should be installed by user
+            raise ImportError(
+                "Could not import yfinance python package. "
+                "Please install it with `pip install yfinance`."
+            ) from exc
+
+        stripped = query.strip()
+        if _looks_like_ticker(stripped):
+            return stripped.upper()
+
+        try:
+            search = Search(stripped, max_results=10)
+            matches = search.quotes or []
+        except (HTTPError, ReadTimeout, ConnectionError):
+            matches = []
+
+        if matches:
+            symbol = matches[0].get("symbol")
+            if symbol:
+                return symbol.upper()
+
+        return None
+
+    def _parse_input(self, tool_input, tool_call_id=None):
+        parsed = super()._parse_input(tool_input, tool_call_id=tool_call_id)
+        if isinstance(parsed, dict):
+            query = parsed.get("query", "")
+            if isinstance(query, str) and not _looks_like_ticker(query):
+                symbol = self._resolve_symbol(query)
+                if symbol:
+                    parsed["query"] = symbol
+                    self._last_resolved_symbol = symbol
+        return parsed
+
     def _run(
         self,
         query: str,
@@ -53,38 +106,60 @@ def _run(
                 "Could not import yfinance python package. "
                 "Please install it with `pip install yfinance`."
             )
-        company = yfinance.Ticker(query)
+
+        if _looks_like_ticker(query):
+            symbol = query.strip().upper()
+        else:
+            symbol = self._last_resolved_symbol or self._resolve_symbol(query)
+
+        self._last_resolved_symbol = None
+
+        if not symbol:
+            return f"Could not find a company for query '{query}'."
+
+        company = yfinance.Ticker(symbol)
         try:
             if company.isin is None:
-                return f"Company ticker {query} not found."
+                return f"Company ticker {symbol} not found."
         except (HTTPError, ReadTimeout, ConnectionError):
-            return f"Company ticker {query} not found."
+            return f"Company ticker {symbol} not found."
 
-        links = []
         try:
-            links = [
-                n["content"]["canonicalUrl"]["url"]
-                for n in company.news
-                if n["content"]["contentType"] == "STORY"
-            ]
+            news_items = company.get_news() or []
         except (HTTPError, ReadTimeout, ConnectionError):
-            if not links:
-                return f"No news found for company that searched with {query} ticker."
-        if not links:
-            return f"No news found for company that searched with {query} ticker."
-        loader = WebBaseLoader(web_paths=links)
-        docs = loader.load()
-        result = self._format_results(docs, query)
-        if not result:
-            return f"No news found for company that searched with {query} ticker."
-        return result
-
-    @staticmethod
-    def _format_results(docs: Iterable[Document], query: str) -> str:
-        doc_strings = [
-            "\n".join([doc.metadata["title"], doc.metadata.get("description", "")])
-            for doc in docs
-            if query in doc.metadata.get("description", "")
-            or query in doc.metadata["title"]
+            return f"Failed to fetch news for {symbol}."
+
+        stories = [
+            item
+            for item in news_items
+            if item.get("content", {}).get("contentType") == "STORY"
         ]
-        return "\n\n".join(doc_strings)
+        if not stories:
+            return f"No news found for company with ticker {symbol}."
+
+        summaries = []
+        for story in stories[: self.top_k]:
+            content = story.get("content", {})
+            title = content.get("title") or story.get("title") or ""
+            summary = content.get("summary") or content.get("description") or ""
+            publisher = (content.get("provider") or {}).get("displayName", "")
+            link = (
+                (content.get("canonicalUrl") or {}).get("url")
+                or (content.get("clickThroughUrl") or {}).get("url")
+                or ""
+            )
+            if not title and not summary:
+                continue
+            parts = [
+                title.strip(),
+                summary.strip(),
+                f"Источник: {publisher}" if publisher else "",
+                link,
+            ]
+            formatted = "\n".join(filter(None, parts))
+            summaries.append(formatted)
+
+        if not summaries:
+            return f"No news summaries available for ticker {symbol}."
+
+        return "\n\n---\n\n".join(summaries)

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

@@ -0,0 +1,78 @@
+import sys
+import types
+
+import pytest
+
+from langchain_community.tools.yahoo_finance_news import YahooFinanceNewsTool
+
+
+@pytest.fixture()
+def fake_yfinance(monkeypatch):
+    """Install a minimal fake `yfinance` module to avoid network requests."""
+
+    class DummySearch:
+        def __init__(self, query: str, max_results: int = 10):
+            self.query = query
+            self.max_results = max_results
+            if query.lower() == "microsoft":
+                self.quotes = [{"symbol": "MSFT"}]
+            else:
+                self.quotes = [{"symbol": query.upper()}]
+
+    class DummyTicker:
+        def __init__(self, symbol: str):
+            self.symbol = symbol
+            self.isin = "DUMMYISIN"
+            yf_module.last_symbol = symbol
+            self._news = [
+                {
+                    "content": {
+                        "contentType": "STORY",
+                        "title": "Sample Headline",
+                        "summary": "Sample Summary",
+                        "provider": {"displayName": "Sample Publisher"},
+                        "canonicalUrl": {"url": "https://example.com/article"},
+                    }
+                }
+            ]
+
+        def get_news(self):
+            return self._news
+
+    search_module = types.ModuleType("yfinance.search")
+    search_module.Search = DummySearch
+
+    yf_module = types.ModuleType("yfinance")
+    yf_module.search = search_module
+    yf_module.Ticker = DummyTicker
+
+    monkeypatch.setitem(sys.modules, "yfinance", yf_module)
+    monkeypatch.setitem(sys.modules, "yfinance.search", search_module)
+
+    yf_module.last_symbol = None
+
+    return yf_module
+
+
+def test_tool_resolves_company_name(fake_yfinance):
+    """The tool should resolve a company name to a ticker before executing."""
+    tool = YahooFinanceNewsTool()
+    output = tool.run("Microsoft")
+
+    assert output
+    assert fake_yfinance.last_symbol == "MSFT"
+    assert "Sample Headline" in output
+    assert "Sample Summary" in output
+    assert "Sample Publisher" in output
+    assert "https://example.com/article" in output
+
+
+def test_tool_accepts_ticker(fake_yfinance):
+    """The tool should accept ticker symbols directly."""
+    tool = YahooFinanceNewsTool()
+    output = tool.run("MSFT")
+
+    # Output should still contain the formatted news snippet.
+    assert "Sample Headline" in output
+    assert "Sample Summary" in output
+    assert fake_yfinance.last_symbol == "MSFT"