# ============================================================ # matching.py — Búsqueda y emparejamiento de empresas/tickers # ------------------------------------------------------------ # NOTA PARA AUDITORÍA: # - Este módulo encapsula la lógica de: # (1) Consulta a Yahoo Finance Search API para obtener candidatos # (nombre, ticker) a partir de una cadena de búsqueda. # (2) Priorización de resultados con una lista/dict opcional # de tickers preferentes (p. ej., los que tienen holdings BTC). # (3) Fuzzy matching (rapidfuzz) primero por ticker y luego por nombre. # - No tiene efectos secundarios al importarlo. # - La salida mantiene la firma usada en tu app: (name, ticker) o (None, None). # ============================================================ 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 # Preferir shortname/longname sobre symbol como nombre visible 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] # Recuperar el 'name' correspondiente 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). """ # 0) Descargar candidatos results = _fetch_yahoo_finance_candidates( query=query, quotes_count=quotes_count, timeout=timeout ) if not results: return None, None # 1) Priorizar tickers conocidos prioritized = _prioritize_with_whitelist(results, priority_tickers) if prioritized: return prioritized # 2) Fuzzy por ticker by_ticker = _fuzzy_match_ticker(query, results, score_cutoff=80) if by_ticker: return by_ticker # 3) Fuzzy por nombre by_name = _fuzzy_match_name(query, results, score_cutoff=60) if by_name: return by_name # 4) Fallback: primer resultado return results[0] __all__ = [ "find_company_online", ]