| 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 |
|
|
|
|
| |
| |
| |
|
|
| 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 |
|
|
|
|
| |
| |
| |
|
|
|
|
| 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 |
|
|
| |
| 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 |
|
|
|
|
| |
| |
| |
|
|
|
|
| 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 |
|
|
|
|
| |
| |
| |
|
|
| |
| priority_tickers = tickers_to_ids.copy() |
|
|
| def find_company_online(query): |
| query_clean = query.strip() |
|
|
| |
| 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 |
|
|
| |
| results = [ |
| (item.get("shortname") or item.get("longname") or item.get("symbol"), item["symbol"]) |
| for item in data["quotes"] |
| if "symbol" in item |
| ] |
|
|
| |
| for name, ticker in results: |
| if ticker in priority_tickers: |
| return name, 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 |
|
|
| |
| 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 |
|
|
| |
| return results[0] if results else (None, None) |
|
|
|
|
