Construir un rastreador de ranking de Google en Python con proxies residenciales

Construir un rastreador de ranking de Google en Python con proxies residenciales: por qué importa

Si gestionas SEO para múltiples clientes o monitorizas tus propias posiciones, necesitas snapshots diarios de SERP, no comprobaciones puntuales. Un rastreador de ranking de Google en Python con proxies residenciales te da visibilidad sobre la volatilidad de posiciones, te alerta de caídas repentinas y te permite correlacionar cambios con actualizaciones del algoritmo. En esta guía construimos uno desde cero: modelo de datos, paginación de SERPs, rotación de IPs residenciales con ProxyHat, parsing de resultados orgánicos y endurecimiento para producción.

El enfoque es code-first: cada sección incluye código ejecutable que puedes copiar y adaptar. Usamos curl_cffi para impersonar el fingerprint TLS de Chrome y SQLite para almacenar el historial de posiciones sin infraestructura adicional.

El problema: Google endurece el scraping de SERPs

Google ha incrementado significativamente sus defensas contra scraping. En septiembre de 2025, Google eliminó el parámetro num=100, que permitía obtener 100 resultados en una sola petición. Ahora, el máximo por página es 10 resultados, lo que obliga a paginar con start=0,10,20... para cubrir el top 100.

Además, Google implementa fingerprinting TLS/JA3-JA4 y scoring de reputación IP. Las peticiones desde IPs de datacenter son detectadas y bloqueadas con mayor frecuencia, devolviendo CAPTCHAs o resultados vacíos. Según la documentación oficial de Google Search, el scraping automatizado viola los términos de servicio, lo que hace crítico usar proxies residenciales con rotación adecuada para distribuir las peticiones.

Estos cambios significan que:

• No puedes usar requests estándar — su fingerprint TLS es detectable.
• No puedes usar IPs de datacenter — se bloquean tras unas pocas peticiones.
• Necesitas sesiones estables por keyword para evitar sesgos geográficos.

Modelo de datos para seguimiento de rankings

Un rastreador de rankings necesita almacenar snapshots temporales. Cada snapshot captura la posición de un dominio para una keyword específica en un momento dado. El modelo mínimo incluye: keyword, dominio objetivo, país, dispositivo, posición y fecha de captura.

Los snapshots diarios son superiores a las comprobaciones puntuales porque:

• Detectan volatilidad: una keyword que pasa de la posición 3 a la 15 y vuelve a la 5 en tres días.
• Permiten calcular tendencias con medias móviles de 7 y 30 días.
• Identifican actualizaciones del algoritmo: caídas simultáneas en múltiples keywords.

Usamos SQLite por su simplicidad — no requiere servidor y soporta hasta 50,000 keywords sin problemas de rendimiento.

import sqlite3
from datetime import datetime, timezone

def init_db(db_path="rankings.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS rank_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            keyword TEXT NOT NULL,
            target_domain TEXT NOT NULL,
            country TEXT NOT NULL DEFAULT 'US',
            device TEXT NOT NULL DEFAULT 'desktop',
            position INTEGER,
            page INTEGER,
            found_url TEXT,
            captured_at TEXT NOT NULL,
            UNIQUE(keyword, target_domain, country, device, captured_at)
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_keyword_domain
        ON rank_snapshots(keyword, target_domain, captured_at)
    """)
    conn.commit()
    return conn

init_db()

Paginación de resultados tras la eliminación de num=100

Con la eliminación de num=100 en septiembre de 2025, necesitas paginar las SERPs. Para cubrir el top 100, haces 10 peticiones con start=0,10,20...90. Cada petición devuelve hasta 10 resultados orgánicos, que debes parsear saltándote anuncios y features de SERP.

La clave es mantener una sesión proxy estable por keyword durante toda la paginación. Si rotas IPs entre páginas, Google puede devolver resultados diferentes por sesgo geográfico, corrompiendo tu snapshot.

Primero, un ejemplo rápido con curl usando ProxyHat:

curl -x "http://user-country-US-city-chicago-session-kw123:pass@gate.proxyhat.com:8080" \
  "https://www.google.com/search?q=python+web+scraping&start=0&num=10&hl=en&gl=us" \
  -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" \
  --compressed

Ahora la función de fetching en Python con curl_cffi, que impersona el fingerprint TLS de Chrome:

import asyncio
from curl_cffi.requests import AsyncSession

GOOGLE_SEARCH_URL = "https://www.google.com/search"

def build_proxy_url(username, password, country="US", city=None, session_id=None):
    parts = [f"user-country-{country}"]
    if city:
        parts.append(f"city-{city.lower()}")
    if session_id:
        parts.append(f"session-{session_id}")
    user = "-".join(parts)
    return f"http://{user}:{password}@gate.proxyhat.com:8080"

async def fetch_serp_page(
    keyword: str,
    start: int,
    country: str,
    session: AsyncSession,
    proxy_url: str,
    max_retries: int = 3,
):
    params = {
        "q": keyword,
        "start": start,
        "num": 10,
        "hl": "en",
        "gl": country.lower(),
    }
    for attempt in range(max_retries):
        try:
            resp = await session.get(
                GOOGLE_SEARCH_URL,
                params=params,
                proxies={"https": proxy_url, "http": proxy_url},
                impersonate="chrome",
                timeout=30,
            )
            if resp.status_code == 200 and "captcha" not in resp.text.lower():
                return resp.text
            elif resp.status_code == 429:
                await asyncio.sleep(2 ** attempt)
            else:
                return None
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)
    return None

Por qué los proxies residenciales son imprescindibles

Google emplea dos capas de detección principales: fingerprinting TLS (JA3/JA4) y reputación de IP. Las IPs de datacenter tienen una reputación conocida y son bloqueadas rápidamente. Los proxies residenciales usan IPs asignadas por ISPs reales, lo que les confiere una reputación mucho más alta.

Además, el geo-targeting a nivel de ciudad es crucial. Los resultados de Google varían según la ubicación del usuario. Si rastreas rankings para "pizza delivery" en Chicago, necesitas una IP residencial en Chicago, no una en Miami. ProxyHat permite geo-targeting granular con el formato user-country-US-city-chicago.

La página de Wikipedia sobre SERPs documenta cómo los resultados incluyen anuncios, features y resultados orgánicos. Solo los orgánicos importan para rank tracking.

Para rank tracking, los proxies residenciales ofrecen el mejor equilibrio entre costo y fiabilidad. Puedes consultar las ubicaciones disponibles de ProxyHat y revisar los planes de precios para elegir el pool adecuado.

Implementación completa con curl_cffi y ProxyHat

Ahora ensamblamos todo: fetching paginado, parsing de resultados orgánicos y almacenamiento en SQLite. Usamos curl_cffi con impersonate='chrome' para replicar el fingerprint TLS de Chrome.

El parsing de resultados orgánicos requiere cuidado: Google mezcla anuncios, featured snippets, People Also Ask y resultados orgánicos. Solo nos interesan los orgánicos.

from bs4 import BeautifulSoup
import re

def parse_organic_results(html: str):
    soup = BeautifulSoup(html, "html.parser")
    results = []
    # Los resultados orgánicos usan div.g con enlaces a[href]
    for div in soup.select("div.g"):
        link = div.select_one("a[href]")
        if not link:
            continue
        href = link.get("href", "")
        if not href.startswith("/url?q=") and not href.startswith("http"):
            continue
        # Extraer URL real
        if href.startswith("/url?q="):
            url = href.split("/url?q=")[1].split("&")[0]
        else:
            url = href
        title_el = div.select_one("h3")
        title = title_el.get_text(strip=True) if title_el else ""
        results.append({"url": url, "title": title})
    return results

def find_domain_position(results: list, target_domain: str):
    for i, r in enumerate(results, 1):
        if target_domain in r["url"]:
            return i, r["url"]
    return None, None

El rastreador completo con concurrencia controlada y sesiones estables por keyword:

import asyncio
import sqlite3
from datetime import datetime, timezone
from curl_cffi.requests import AsyncSession

async def track_keyword(
    keyword: str,
    target_domain: str,
    country: str = "US",
    city: str = None,
    device: str = "desktop",
    proxy_username: str = "YOUR_USER",
    proxy_password: str = "YOUR_PASS",
    db_path: str = "rankings.db",
):
    session_id = f"kw-{abs(hash(keyword)) % 100000}"
    proxy_url = build_proxy_url(
        proxy_username, proxy_password, country, city, session_id
    )

    async with AsyncSession() as http:
        all_results = []
        for start in range(0, 100, 10):
            html = await fetch_serp_page(keyword, start, country, http, proxy_url)
            if not html:
                break
            page_results = parse_organic_results(html)
            if not page_results:
                break
            all_results.extend(page_results)
            await asyncio.sleep(2)  # Rate limit entre páginas

        position, found_url = find_domain_position(all_results, target_domain)
        page = (position // 10) + 1 if position else None

        conn = sqlite3.connect(db_path)
        conn.execute(
            """INSERT OR REPLACE INTO rank_snapshots
            (keyword, target_domain, country, device, position,
             page, found_url, captured_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (keyword, target_domain, country, device, position,
             page, found_url, datetime.now(timezone.utc).isoformat())
        )
        conn.commit()
        conn.close()

        return {
            "keyword": keyword,
            "position": position,
            "page": page,
            "url": found_url,
        }

async def main():
    keywords = ["python web scraping", "best proxy service", "seo tools 2026"]
    target = "example.com"
    semaphore = asyncio.Semaphore(5)  # Máx 5 concurrentes

    async def limited_track(kw):
        async with semaphore:
            return await track_keyword(
                kw, target, country="US", city="chicago"
            )

    tasks = [limited_track(kw) for kw in keywords]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"{r['keyword']}: posición {r['position']}")

asyncio.run(main())

Endurecimiento para producción

Un rastreador en producción necesita manejar fallos con elegancia. Estos son los patrones esenciales:

Reintentos con backoff exponencial

Ya implementamos reintentos en fetch_serp_page con 2 ** attempt segundos de espera. Para mayor robustez, usa tenacity:

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rank-tracker")

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    reraise=True,
)
async def fetch_with_retry(session, url, **kwargs):
    resp = await session.get(url, **kwargs)
    if resp.status_code == 429:
        logger.warning("Rate limited, reintentando...")
        raise Exception("429 Too Many Requests")
    return resp

Detección de CAPTCHA

Google devuelve CAPTCHAs como respuestas 200 con contenido específico. Detectarlos evita registrar posiciones falsas:

def detect_captcha(html: str) -> bool:
    captcha_signals = [
        "captcha", "g-recaptcha", "unusual traffic",
        "detected unusual traffic", "sorry/image"
    ]
    html_lower = html.lower()
    return any(signal in html_lower for signal in captcha_signals)

# En fetch_serp_page, añadir antes de retornar:
# if detect_captcha(resp.text):
#     logger.error("CAPTCHA detectado, rotando sesión proxy")
#     return None

Pools de proxy por país y límites de concurrencia

Usa asyncio.Semaphore para limitar la concurrencia a 5-10 keywords simultáneas. Para múltiples países, asigna un pool de sesiones proxy distinto por país. Mantén 200ms de latencia mínima entre peticiones al mismo proxy.

Suavizado de volatilidad de rankings

Las posiciones diarias pueden ser ruidosas. Aplica una media móvil de 7 días para suavizar y exporta a CSV:

import csv
from collections import defaultdict

def calculate_moving_average(db_path="rankings.db", window=7):
    conn = sqlite3.connect(db_path)
    rows = conn.execute("""
        SELECT keyword, captured_at, position
        FROM rank_snapshots ORDER BY keyword, captured_at
    """).fetchall()
    conn.close()

    by_keyword = defaultdict(list)
    for kw, date, pos in rows:
        by_keyword[kw].append((date, pos))

    averages = []
    for kw, data in by_keyword.items():
        for i in range(len(data)):
            start = max(0, i - window + 1)
            window_positions = [d[1] for d in data[start:i+1] if d[1]]
            if window_positions:
                avg = sum(window_positions) / len(window_positions)
                averages.append((kw, data[i][0], data[i][1], round(avg, 1)))
    return averages

def export_to_csv(db_path="rankings.db", output="rankings_history.csv"):
    conn = sqlite3.connect(db_path)
    rows = conn.execute("""
        SELECT keyword, target_domain, country, device, position,
               page, found_url, captured_at
        FROM rank_snapshots ORDER BY keyword, captured_at
    """).fetchall()
    conn.close()
    with open(output, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["keyword", "target_domain", "country", "device",
                         "position", "page", "found_url", "captured_at"])
        writer.writerows(rows)
    print(f"Exportadas {len(rows)} filas a {output}")

Consideraciones éticas y límites

El scraping de Google existe en un área legal gris. Antes de implementar tu rastreador:

• Consulta el robots.txt de Google: https://www.google.com/robots.txt restringe el acceso a /search.
• Usa volúmenes bajos: 50-100 keywords por día es razonable; 10,000 no lo es.
• Prefiere fuentes oficiales: la Search Console API de Google ofrece datos de impresiones y clics de forma legítima y gratuita hasta 5,000 consultas al día.
• Respeta los ToS de Google: el scraping automatizado puede violar los términos de servicio.
• Rastrea tu propio dominio o datos públicos: no rastrees posiciones de competidores sin autorización.

Para casos de uso legítimo, consulta nuestra guía de casos de uso de web scraping y seguimiento de SERP.

Conclusiones clave

Un rastreador de rankings en producción requiere tres piezas: fetching con fingerprint TLS realista, proxies residenciales con geo-targeting, y almacenamiento temporal para detectar volatilidad.

• Usa curl_cffi con impersonate='chrome' — requests estándar es detectable por su fingerprint TLS.
• Pagina con start=0,10,20...90 para cubrir el top 100 tras la eliminación de num=100.
• Mantén sesiones proxy estables por keyword: user-country-US-city-chicago-session-kw123.
• Limita la concurrencia a 5-10 keywords simultáneas con asyncio.Semaphore.
• Detecta CAPTCHAs antes de registrar posiciones para evitar datos corruptos.
• Para volúmenes bajos, prefiere la Search Console API — es gratuita y legítima.