Create matching.py

matching.py

@@ -0,0 +1,210 @@
+# ============================================================
+# 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
+
+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}&quotesCount={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",
+]