Blame for matching.py

# ============================================================
# 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}&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",
]