File size: 4,784 Bytes
9bb3deb c86d0be 9bb3deb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | import re
import requests
import pandas as pd
import numpy as np
from bs4 import BeautifulSoup
from fuzzywuzzy import process
from datetime import datetime, timedelta
from config import tickers_to_ids
from config import api_key
# -----------------------------------------------
# HERRAMIENTAS DE FROMATO
# -----------------------------------------------
def limpiar_texto(texto):
"""Elimina caracteres especiales y espacios extra."""
texto_limpio_parcial = re.sub(r'[^a-zA-Z0-9.,\s]', '', texto)
texto_final = texto_limpio_parcial.strip()
return texto_final
def expand_money(val):
"""Convierte strings de dinero (ej. '$1.2B') a valores flotantes."""
if pd.isna(val):
return np.nan
val = str(val).replace('$', '').replace(',','').upper()
try:
if 'B' in val:
return float(val.replace('B', '')) * 1_000_000_000
elif 'M' in val:
return float(val.replace('M', '')) * 1_000_000
elif 'K' in val:
return float(val.replace('K', '')) * 1_000
else:
return float(val)
except:
return np.nan
# -----------------------------------------------
# FUNCIÓN DE SCRAPER API
# -----------------------------------------------
def descargar_tabla_web(url, indice_tabla=0):
"""
Descarga una tabla de una URL usando ScraperAPI para evitar bloqueos.
"""
api_url = f"http://api.scraperapi.com?api_key={api_key}&url={url}"
try:
response = requests.get(api_url)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
tables = soup.find_all('table')
if len(tables) > indice_tabla:
table = tables[indice_tabla]
else:
print(f"No se encontró la tabla con índice {indice_tabla} en la página: {url}.")
return None
# El resto del código para extraer los datos de la tabla sigue igual
headers = [th.text.strip() for th in table.find_all('th')]
data = []
for row in table.find_all('tr'):
cols = row.find_all('td')
cols = [ele.text.strip() for ele in cols]
if cols:
data.append(cols)
if headers and data and len(headers) == len(data[0]):
df = pd.DataFrame(data, columns=headers)
else:
df = pd.DataFrame(data)
return df
except requests.exceptions.RequestException as e:
print(f"Error al acceder a la URL: {e}")
return None
except Exception as e:
print(f"Ocurrió un error inesperado: {e}")
return None
# -----------------------------------------------
# FUNCIÓN PARA LEER LA PLANTILLA
# -----------------------------------------------
def extract_file_id(url):
"""Extrae el ID del archivo de una URL de Google Drive."""
match = re.search(r'/d/([a-zA-Z0-9_-]+)', url)
if match:
return match.group(1)
return None
# -----------------------------------------------
# FUNCIÓN PARA EL CUADRO DE BÚSQUEDA
# -----------------------------------------------
# Lista de tickers prioritarios
priority_tickers = tickers_to_ids.copy()
def find_company_online(query):
query_clean = query.strip()
# Petición a la API de Yahoo Finance
url = f"https://query1.finance.yahoo.com/v1/finance/search?q={query_clean}"esCount=10&newsCount=0"
headers = {"User-Agent": "Mozilla/5.0"}
r = requests.get(url, headers=headers)
data = r.json()
if "quotes" not in data or not data["quotes"]:
return None, None
# Lista de resultados: (nombre, ticker)
results = [
(item.get("shortname") or item.get("longname") or item.get("symbol"), item["symbol"])
for item in data["quotes"]
if "symbol" in item
]
# Priorizar tickers de la lista
for name, ticker in results:
if ticker in priority_tickers:
return name, ticker
# Buscar coincidencia exacta/aproximada por ticker
tickers = [sym for _, sym in results]
best_ticker_match = process.extractOne(query_clean.upper(), tickers, score_cutoff=80)
if best_ticker_match:
ticker = best_ticker_match[0]
name = next(name for name, sym in results if sym == ticker)
return name, ticker
# Buscar coincidencia por nombre
names = [name for name, _ in results]
best_name_match = process.extractOne(query_clean, names, score_cutoff=60)
if best_name_match:
name = best_name_match[0]
ticker = next(sym for nm, sym in results if nm == name)
return name, ticker
# Si no hay nada claro, devolver el primer resultado
return results[0] if results else (None, None)
|