| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| from __future__ import annotations |
|
|
| from typing import Dict, Iterable, List, Optional, Tuple |
|
|
| from config import ( |
| SCRAPERAPI_KEY, CRYPTOCOMPARE_API_KEY, TIMEZONE_TARGET, GRADIO_SHARE, |
| CRYPTOCOMPARE_BASE_URL, BITCOINTREASURIES_STOCKS_URL, BITCOINTREASURIES_ETFS_URL, |
| YF_SEARCH_ENDPOINT, TICKERS_TO_IDS |
| ) |
|
|
|
|
| import requests |
| from rapidfuzz import process |
|
|
|
|
| YF_SEARCH_ENDPOINT = ( |
| "https://query1.finance.yahoo.com/v1/finance/search" |
| ) |
|
|
|
|
| def _fetch_yahoo_finance_candidates( |
| query: str, |
| quotes_count: int = 10, |
| timeout: float = 15.0, |
| user_agent: str = "Mozilla/5.0", |
| ) -> List[Tuple[str, str]]: |
| """ |
| Llama a Yahoo Finance Search API y devuelve una lista de candidatos (name, ticker). |
| |
| Parámetros |
| ---------- |
| query : str |
| Texto de búsqueda (p. ej., 'MicroStrategy', 'MSTR', 'Tesla'). |
| quotes_count : int |
| Número máximo de resultados (por defecto: 10). |
| timeout : float |
| Timeout de la petición (segundos). |
| user_agent : str |
| Cabecera 'User-Agent' para la petición. |
| |
| Devuelve |
| -------- |
| list[tuple[str, str]] |
| Lista de pares (name, ticker) en el orden que devuelve la API. |
| Si no hay resultados o ocurre un error, devuelve lista vacía. |
| """ |
| q = (query or "").strip() |
| if not q: |
| return [] |
|
|
| url = f"{YF_SEARCH_ENDPOINT}?q={q}"esCount={quotes_count}&newsCount=0" |
| headers = {"User-Agent": user_agent} |
|
|
| try: |
| r = requests.get(url, headers=headers, timeout=timeout) |
| r.raise_for_status() |
| data = r.json() |
| except Exception: |
| return [] |
|
|
| quotes = data.get("quotes", []) |
| results: List[Tuple[str, str]] = [] |
|
|
| for item in quotes: |
| symbol = item.get("symbol") |
| if not symbol: |
| continue |
| |
| name = item.get("shortname") or item.get("longname") or symbol |
| results.append((name, symbol)) |
|
|
| return results |
|
|
|
|
| def _prioritize_with_whitelist( |
| results: Iterable[Tuple[str, str]], |
| priority_tickers: Optional[Dict[str, object]] = None, |
| ) -> Optional[Tuple[str, str]]: |
| """ |
| Devuelve el primer candidato cuyo ticker esté en la lista/dict de prioridad. |
| Si no hay prioridad o no hay coincidencias, devuelve None. |
| """ |
| if not priority_tickers: |
| return None |
|
|
| priority_keys = set(priority_tickers.keys()) |
| for name, ticker in results: |
| if ticker in priority_keys: |
| return name, ticker |
| return None |
|
|
|
|
| def _fuzzy_match_ticker( |
| query: str, |
| results: Iterable[Tuple[str, str]], |
| score_cutoff: int = 80, |
| ) -> Optional[Tuple[str, str]]: |
| """ |
| Fuzzy matching por TICKER (mayúsculas). Devuelve el (name, ticker) con |
| mejor score si supera el umbral; en caso contrario, None. |
| """ |
| tickers = [sym for _, sym in results] |
| if not tickers: |
| return None |
|
|
| match = process.extractOne(query.upper(), tickers, score_cutoff=score_cutoff) |
| if not match: |
| return None |
|
|
| best_ticker = match[0] |
| |
| for name, ticker in results: |
| if ticker == best_ticker: |
| return name, ticker |
| return None |
|
|
|
|
| def _fuzzy_match_name( |
| query: str, |
| results: Iterable[Tuple[str, str]], |
| score_cutoff: int = 60, |
| ) -> Optional[Tuple[str, str]]: |
| """ |
| Fuzzy matching por NOMBRE visible. Devuelve el (name, ticker) con mejor |
| score si supera el umbral; si no, None. |
| """ |
| names = [name for name, _ in results] |
| if not names: |
| return None |
|
|
| match = process.extractOne(query, names, score_cutoff=score_cutoff) |
| if not match: |
| return None |
|
|
| best_name = match[0] |
| for name, ticker in results: |
| if name == best_name: |
| return name, ticker |
| return None |
|
|
|
|
| def find_company_online( |
| query: str, |
| priority_tickers: Optional[Dict[str, object]] = None, |
| quotes_count: int = 10, |
| timeout: float = 15.0, |
| ) -> Tuple[Optional[str], Optional[str]]: |
| """ |
| Realiza la búsqueda de una empresa/ticker en Yahoo Finance y aplica |
| una estrategia de resolución en cuatro pasos (igual a tu lógica): |
| |
| 1) Si 'priority_tickers' está definido, devuelve el primer candidato |
| cuyo ticker esté en esa lista. |
| 2) Fuzzy matching por ticker (score >= 80). |
| 3) Fuzzy matching por nombre (score >= 60). |
| 4) Si todo falla, el primer resultado. |
| 5) Si no hay resultados, (None, None). |
| |
| Parámetros |
| ---------- |
| query : str |
| Texto a buscar (se recomienda recortar en UI a 6 caracteres como haces). |
| priority_tickers : dict[str, Any], opcional |
| Diccionario o conjunto de tickers priorizados (p. ej., tickers_to_ids). |
| quotes_count : int |
| Límite de resultados solicitados a la API de Yahoo Finance. |
| timeout : float |
| Timeout de la petición a la API. |
| |
| Devuelve |
| -------- |
| (str | None, str | None) |
| Tupla (name, ticker) o (None, None). |
| """ |
| |
| results = _fetch_yahoo_finance_candidates( |
| query=query, quotes_count=quotes_count, timeout=timeout |
| ) |
| if not results: |
| return None, None |
|
|
| |
| prioritized = _prioritize_with_whitelist(results, priority_tickers) |
| if prioritized: |
| return prioritized |
|
|
| |
| by_ticker = _fuzzy_match_ticker(query, results, score_cutoff=80) |
| if by_ticker: |
| return by_ticker |
|
|
| |
| by_name = _fuzzy_match_name(query, results, score_cutoff=60) |
| if by_name: |
| return by_name |
|
|
| |
| return results[0] |
|
|
|
|
| __all__ = [ |
| "find_company_online", |
| ] |
|
|
