diff --git a/libs/community/langchain_community/tools/yahoo_finance_news.py b/libs/community/langchain_community/tools/yahoo_finance_news.py index 7a9a49d2e..dcbc5c7b6 100644 --- a/libs/community/langchain_community/tools/yahoo_finance_news.py +++ b/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) diff --git a/libs/community/tests/unit_tests/tools/test_yahoo_finance_news.py b/libs/community/tests/unit_tests/tools/test_yahoo_finance_news.py new file mode 100644 index 000000000..e861ec9f7 --- /dev/null +++ b/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"