Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update search #2237

Open
wants to merge 3 commits into
base: dev
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
156 changes: 126 additions & 30 deletions yfinance/search.py
Original file line number Diff line number Diff line change
@@ -24,9 +24,73 @@
from . import utils
from .const import _BASE_URL_
from .data import YfData
from typing import TypedDict
import typing as t

SEARCH_CACHE = TypedDict('SEARCH_CACHE', {
"query": str,
"fields": list[str],
"results": dict
})

class SearchCache:
data = YfData()
logger = utils.get_yf_logger()
def __init__(self):
self.cache = {}

def get(self, query:'tuple[str, list[str]]') -> 't.Union[dict, None]':
if query in self.cache:
return self.cache[query]
return None

def set(self, query:'tuple[str, list[str]]', data:'dict') -> 'None':
self.cache[query] = data

def fetch(self, query:'str', fields:'list[str]', session=None, proxy=None, timeout=30, **kwargs:'dict') -> 'dict':
if res := self.get((query, fields)):
return res

params = {
"q": query,
"enableFuzzyQuery": "enable_fuzzy_query" in fields,
"quotesQueryId": "tss_match_phrase_query",
"newsQueryId": "news_cie_vespa",
"enableCb": "cb" in fields,
"enableNavLinks": "nav_links" in fields,
"enableResearchReports": "research" in fields,
"enableCulturalAssets": "cultural_assets" in fields,
}


if "quotes" in fields:
params["quotesCount"] = kwargs.get("max_results", 8)
params["recommendedCount"] = kwargs.get("recommended", 8)
if "news" in fields:
params["newsCount"] = kwargs.get("news_count", 8)
if "lists" in fields:
params["listsCount"] = kwargs.get("lists_count", 8)

url = f"{_BASE_URL_}/v1/finance/search"
data = self.data.cache_get(url=url, params=params, proxy=proxy, timeout=timeout, session=session)
if data is None or "Will be right back" in data.text:
raise RuntimeError("*** YAHOO! FINANCE IS CURRENTLY DOWN! ***\n"
"Our engineers are working quickly to resolve "
"the issue. Thank you for your patience.")
try:
data = data.json()
self.set((query, fields), data)
except _json.JSONDecodeError:
self.logger.error(f"{query}: Failed to retrieve search results and received faulty response instead.")
data = {}
return data





class Search:
cache = SearchCache()
def __init__(self, query, max_results=8, news_count=8, lists_count=8, include_cb=True, include_nav_links=False,
include_research=False, include_cultural_assets=False, enable_fuzzy_query=False, recommended=8,
session=None, proxy=None, timeout=30, raise_errors=True):
@@ -49,6 +113,7 @@ def __init__(self, query, max_results=8, news_count=8, lists_count=8, include_cb
timeout: Request timeout in seconds (default 30).
raise_errors: Raise exceptions on error (default True).
"""
self.fields = ["quotes" if max_results else "", "news" if news_count else "", "lists" if lists_count else "", "cb" if include_cb else "", "nav_links" if include_nav_links else "", "research" if include_research else "", "cultural_assets" if include_cultural_assets else ""]
self.query = query
self.max_results = max_results
self.enable_fuzzy_query = enable_fuzzy_query
@@ -65,7 +130,6 @@ def __init__(self, query, max_results=8, news_count=8, lists_count=8, include_cb
self.enable_cultural_assets = include_cultural_assets
self.recommended = recommended

self._data = YfData(session=self.session)
self._logger = utils.get_yf_logger()

self._response = {}
@@ -80,34 +144,9 @@ def __init__(self, query, max_results=8, news_count=8, lists_count=8, include_cb

def search(self) -> 'Search':
"""Search using the query parameters defined in the constructor."""
url = f"{_BASE_URL_}/v1/finance/search"
params = {
"q": self.query,
"quotesCount": self.max_results,
"enableFuzzyQuery": self.enable_fuzzy_query,
"newsCount": self.news_count,
"quotesQueryId": "tss_match_phrase_query",
"newsQueryId": "news_cie_vespa",
"listsCount": self.lists_count,
"enableCb": self.include_cb,
"enableNavLinks": self.nav_links,
"enableResearchReports": self.enable_research,
"enableCulturalAssets": self.enable_cultural_assets,
"recommendedCount": self.recommended
}
self._logger.debug(f'{self.query}: Fields: {self.fields}')

self._logger.debug(f'{self.query}: Yahoo GET parameters: {str(dict(params))}')

data = self._data.cache_get(url=url, params=params, proxy=self.proxy, timeout=self.timeout)
if data is None or "Will be right back" in data.text:
raise RuntimeError("*** YAHOO! FINANCE IS CURRENTLY DOWN! ***\n"
"Our engineers are working quickly to resolve "
"the issue. Thank you for your patience.")
try:
data = data.json()
except _json.JSONDecodeError:
self._logger.error(f"{self.query}: Failed to retrieve search results and received faulty response instead.")
data = {}
data = self.cache.fetch(self.query, self.fields, session=self.session, proxy=self.proxy, timeout=self.timeout)

self._response = data
# Filter quotes to only include symbols
@@ -117,11 +156,68 @@ def search(self) -> 'Search':
self._research = data.get("researchReports", [])
self._nav = data.get("nav", [])

self._all = {"quotes": self._quotes, "news": self._news, "lists": self._lists, "research": self._research,
"nav": self._nav}
self._all = {
"quotes": self._quotes,
"news": self._news,
"lists": self._lists,
"research": self._research,
"nav": self._nav
}

return self



def search_quotes(self) -> 'list':
"""Search using the query parameters defined in the constructor, but only return the quotes."""
self._logger.debug(f'{self.query}: Fields: [quotes]')
data = self.cache.fetch(self.query, ["quotes"], session=self.session, proxy=self.proxy, timeout=self.timeout)

self._quotes = data.get("quotes", [])
self._response = data
self._all["quotes"] = self._quotes
return self.quotes

def search_news(self) -> 'list':
"""Search using the query parameters defined in the constructor, but only return the news."""
self._logger.debug(f'{self.query}: Fields: [news]')
data = self.cache.fetch(self.query, ["news"], session=self.session, proxy=self.proxy, timeout=self.timeout)

self._news = data.get("news", [])
self._response = data
self._all["news"] = self._news
return self.news

def search_lists(self) -> 'list':
"""Search using the query parameters defined in the constructor, but only return the lists."""
self._logger.debug(f'{self.query}: Fields: [lists]')
data = self.cache.fetch(self.query, ["lists"], session=self.session, proxy=self.proxy, timeout=self.timeout)

self._lists = data.get("lists", [])
self._response = data
self._all["lists"] = self._lists
return self.lists

def search_research(self) -> 'list':
"""Search using the query parameters defined in the constructor, but only return the research reports."""
self._logger.debug(f'{self.query}: Fields: [research]')
data = self.cache.fetch(self.query, ["research"], session=self.session, proxy=self.proxy, timeout=self.timeout)

self._research = data.get("researchReports", [])
self._response = data
self._all["research"] = self._research
return self.research

def search_nav(self) -> 'list':
"""Search using the query parameters defined in the constructor, but only return the navigation links."""
self._logger.debug(f'{self.query}: Fields: [nav]')
data = self.cache.fetch(self.query, ["nav"], session=self.session, proxy=self.proxy, timeout=self.timeout)

self._nav = data.get("nav", [])
self._response = data
self._all["nav"] = self._nav
return self.nav

@property
def quotes(self) -> 'list':
"""Get the quotes from the search results."""