レジデンシャルプロキシでPythonのGoogleランキングトラッカーを構築する

SEOエンジニアやPython開発者にとって、レジデンシャルプロキシでPythonのGoogleランキングトラッカーを構築することは、順位変動の可視化と競合分析を自動化する第一歩です。本記事では、Googleが2025年9月にnum=100パラメータを廃止した後のSERP取得方法、TLS/JA3-JA4フィンガープリント回避、ProxyHatによる都市単位のジオターゲティング、SQLiteへの履歴保存、そして本番運用向けのリトライ・CAPTCHA検出・並行制御までをコード中心に解説します。

なぜレジデンシャルプロキシでPythonのGoogleランキングトラッカーを構築する必要があるのか

Googleは2025年9月に検索URLのnum=100パラメータのサポートを終了し、1ページあたりの有機結果は最大10件に固定されました。これにより、上位100位を追跡するにはstart=0,10,20…でページネーションする必要があります。同時に、GoogleはTLSクライアントハロー（JA3/JA4）フィンガープリントとIPレピュテーションスコアを組み合わせてボットトラフィックを検出する仕組みを強化しています。データセンタープロキシのIP帯域は高いスコアでフラグ付けされるため、レジデンシャルプロキシが事実上の必須となっています。

1回限りの順位チェックでは、日々のボラティリティ（順位変動）を追跡できません。毎日SERPスナップショットを保存することで、アルゴリズム更新の影響範囲、インデックス再編成のタイミング、競合の急上昇を特定できます。これは Google Search Central Blog が推奨する「トレンドの監視」アプローチとも一致します。

データモデル設計：キーワード・ドメイン・国・デバイス・順位・取得日時

ランクトラッカーの中核は、時系列データモデルです。以下のフィールドを保持するテーブルを設計します。

• keyword：追跡対象の検索クエリ
• target_domain：順位を測定する対象ドメイン（例: example.com）
• country：ジオターゲット（例: US、JP）
• device：desktop または mobile
• position：有機結果の順位（1〜100）
• captured_at：取得日時（UTCタイムスタンプ）

このモデルにより、「特定キーワードで対象ドメインが過去30日間で何位から何位に変動したか」をSQL1本で回答できます。

プロキシ方式の比較：レジデンシャル vs データセンター vs モバイル

方式	IPレピュテーション	Google検索の成功率	コスト	適している用途
データセンタープロキシ	低い（フラグされやすい）	〜30%	低	静的ページの軽量スクレイピング
レジデンシャルプロキシ	高い（実在ISPのIP）	95%以上	中	SERP追跡、e-commerce価格監視
モバイルプロキシ	非常に高い	98%以上	高	モバイル専用SERP、アプリストア追跡

SERP追跡の大部分はレジデンシャルプロキシで十分です。モバイル専用デバイスでの順位を追跡する場合のみモバイルプロキシを併用します。ProxyHatの料金体系は /ja/pricing で確認できます。

SERP取得の基本：curl_cffiでChromeを偽装する

Python標準のrequestsはTLSハローがPythonのデフォルトであり、Googleに即座に検出されます。curl_cffiのimpersonate='chrome'を使うことで、Chromeと同一のTLSフィンガープリントを生成できます。

import asyncio
from curl_cffi.requests import AsyncSession

PROXY = "http://user-country-US-city-chicago-session-rank001:pass@gate.proxyhat.com:8080"

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Accept-Language": "en-US,en;q=0.9",
}

async def fetch_serp(keyword: str, start: int = 0) -> str:
    url = "https://www.google.com/search"
    params = {"q": keyword, "start": start, "hl": "en", "gl": "us"}
    async with AsyncSession(impersonate="chrome") as s:
        resp = await s.get(
            url,
            params=params,
            headers=HEADERS,
            proxies={"https": PROXY, "http": PROXY},
            timeout=30,
        )
        resp.raise_for_status()
        return resp.text

if __name__ == "__main__":
    html = asyncio.run(fetch_serp("build a rank tracker python"))
    print(f"取得したHTML長: {len(html)}")

ページネーションで上位100位を取得する

num=100が廃止されたため、start=0,10,20,…,90の10ページを順番に取得します。

async def fetch_top_100(keyword: str) -> list[str]:
    pages = []
    for start in range(0, 100, 10):
        try:
            html = await fetch_serp(keyword, start=start)
            pages.append(html)
            await asyncio.sleep(2)  # 礼儀正しい間隔
        except Exception as e:
            print(f"start={start} で失敗: {e}")
            break
    return pages

有機結果のパース：広告とSERP機能を除外する

Googleの検索結果HTMLは頻繁に変更されます。堅牢な抽出にはCSSセレクタと正規表現を組み合わせ、広告やナレッジパネルを除外します。

from bs4 import BeautifulSoup
import re

def parse_organic_results(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    results = []
    # 広告ブロックを除外
    for ad in soup.select("[data-text-ad], .uEierd, .commercial-unit"):
        ad.decompose()

    for div in soup.select("div.g, div.MjjYud"):
        link = div.select_one("a[href^='/url?q='], a[href^='http']")
        if not link:
            continue
        href = link.get("href", "")
        # /url?q= 形式のリダイレクトを展開
        m = re.search(r"/url\?q=([^&]+)", href)
        url = m.group(1) if m else href
        title_el = div.select_one("h3")
        title = title_el.get_text(strip=True) if title_el else ""
        if url and title:
            results.append({"url": url, "title": title})
    return results

対象ドメインの順位を抽出してSQLiteに保存

import sqlite3
from datetime import datetime, timezone
from urllib.parse import urlparse

DB_PATH = "rank_history.db"

def init_db(db_path: str = DB_PATH) -> None:
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS rankings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            keyword TEXT NOT NULL,
            target_domain TEXT NOT NULL,
            country TEXT NOT NULL,
            device TEXT NOT NULL,
            position INTEGER,
            captured_at TEXT NOT NULL
        )
    """)
    conn.commit()
    conn.close()

def find_position(results: list[dict], target_domain: str) -> int | None:
    target = target_domain.lower().replace("www.", "")
    for i, r in enumerate(results, start=1):
        host = urlparse(r["url"]).netloc.lower().replace("www.", "")
        if host == target or host.endswith("." + target):
            return i
    return None

def save_ranking(keyword: str, target_domain: str, country: str,
                 device: str, position: int | None) -> None:
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "INSERT INTO rankings (keyword, target_domain, country, device, position, captured_at) "
        "VALUES (?, ?, ?, ?, ?, ?)",
        (keyword, target_domain, country, device, position,
         datetime.now(timezone.utc).isoformat()),
    )
    conn.commit()
    conn.close()

ProxyHat SDKでIPローテーションを統合

キーワードごとにスティッキーセッションを使うことで、ページネーション中にIPが変わらないようにします。これにより、Googleが「同じユーザーがページめくりをしている」と認識しやすくなります。

import hashlib

def build_proxy_url(username: str, password: str, country: str,
                    city: str | None, keyword: str) -> str:
    # キーワードから安定したセッションIDを生成
    session_id = hashlib.md5(keyword.encode()).hexdigest()[:12]
    parts = [f"user-country-{country}"]
    if city:
        parts.append(f"city-{city.lower()}")
    parts.append(f"session-{session_id}")
    user = "-".join(parts)
    return f"http://{user}:{password}@gate.proxyhat.com:8080"

async def track_keyword(keyword: str, target_domain: str,
                        country: str = "US", city: str | None = None) -> int | None:
    proxy = build_proxy_url("user", "pass", country, city, keyword)
    all_results = []
    for start in range(0, 100, 10):
        html = await fetch_serp_with_proxy(keyword, start, proxy)
        all_results.extend(parse_organic_results(html))
        await asyncio.sleep(2)
    pos = find_position(all_results, target_domain)
    save_ranking(keyword, target_domain, country, "desktop", pos)
    return pos

async def fetch_serp_with_proxy(keyword: str, start: int, proxy: str) -> str:
    url = "https://www.google.com/search"
    params = {"q": keyword, "start": start, "hl": "en", "gl": "us"}
    async with AsyncSession(impersonate="chrome") as s:
        resp = await s.get(
            url,
            params=params,
            headers=HEADERS,
            proxies={"https": proxy, "http": proxy},
            timeout=30,
        )
        resp.raise_for_status()
        return resp.text

本番運用のハードニング：リトライ・CAPTCHA検出・並行制御

指数バックオフ付きリトライ

import random

async def fetch_with_retry(keyword: str, start: int, proxy: str,
                           max_retries: int = 4) -> str:
    for attempt in range(max_retries):
        try:
            html = await fetch_serp_with_proxy(keyword, start, proxy)
            if "Our systems have detected unusual traffic" in html:
                raise RuntimeError("CAPTCHA/ボット検出ページ")
            return html
        except Exception as e:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"リトライ {attempt+1}/{max_retries} ({e}) — {wait:.1f}秒待機")
            await asyncio.sleep(wait)
    raise RuntimeError(f"{keyword} start={start} の取得に失敗")

セマファで並行数を制限

CONCURRENCY = 5
sem = asyncio.Semaphore(CONCURRENCY)

async def safe_track(keyword: str, target_domain: str, country: str) -> None:
    async with sem:
        try:
            pos = await track_keyword(keyword, target_domain, country)
            print(f"{keyword}: {pos or '圏外'}")
        except Exception as e:
            print(f"{keyword} 追跡失敗: {e}")

async def main():
    init_db()
    keywords = ["build a rank tracker python", "serp scraping python 2026",
                "google rank tracker proxies"]
    tasks = [safe_track(kw, "proxyhat.com", "US") for kw in keywords]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

CSVエクスポートで分析しやすく

import csv

def export_csv(output_path: str = "rank_history.csv") -> None:
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        "SELECT keyword, target_domain, country, device, position, captured_at "
        "FROM rankings ORDER BY captured_at"
    ).fetchall()
    conn.close()
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["keyword", "target_domain", "country", "device", "position", "captured_at"])
        w.writerows(rows)

順位ボラティリティの平滑化

1日のスパイクでパニックにならないよう、7日間移動平均を算出してトレンドを可視化します。

def moving_average(keyword: str, target_domain: str, window: int = 7) -> list[float]:
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        "SELECT position FROM rankings WHERE keyword=? AND target_domain=? "
        "ORDER BY captured_at", (keyword, target_domain)
    ).fetchall()
    conn.close()
    positions = [r[0] for r in rows if r[0] is not None]
    if len(positions) < window:
        return positions
    return [
        sum(positions[i:i+window]) / window
        for i in range(len(positions) - window + 1)
    ]

ProxyHat固有の設定とジオターゲティング

ProxyHatではユーザー名にフラグを埋め込むでジオターゲティングとスティッキーセッションを制御します。公式ドキュメントは docs.proxyhat.com で確認できます。

• 国指定: user-country-US:pass@gate.proxyhat.com:8080
• 都市指定: user-country-DE-city-berlin:pass@gate.proxyhat.com:8080
• スティッキーセッション: user-session-abc123:pass@gate.proxyhat.com:8080
• SOCKS5（モバイルSERP用）: socks5://user-country-US:pass@gate.proxyhat.com:1080

対応ロケーションの一覧は /ja/locations で確認できます。SERP追跡のユースケース詳細は /ja/use-cases/serp-tracking を参照してください。Webスクレイピング全般のユースケースは /ja/use-cases/web-scraping にまとめてあります。

よくある失敗と対策

• 1キーワードで100ページ連続取得：IPが即座にブロックされます。2〜3秒の間隔とセッション固定を必ず入れてください。
• デバイスを指定しない：モバイルとデスクトップで順位が異なります。deviceフィールドで必ず区別してください。
• 広告を有機結果としてカウント：data-text-ad属性のブロックをパース前に除外してください。
• HTML構造変更への無対策：セレクタが壊れた場合に空リストを返すフォールバックを用意してください。

倫理と制限：自分の順位と公開ランキングを追跡する

Googleの利用規約は自動化された検索クエリを禁止しています。本番運用では以下を実践してください。

• 追跡は自分の所有ドメインと公開情報のランキングに限定する
• 1キーワードあたり1日1回まで、低頻度を保つ
• 低ボリュームの場合は公式のGoogle Custom Search APIまたは公式SERP APIの利用を優先する
• robots.txtと対象サイトのToSを尊重する
• GDPR/CCPA対象の個人データを収集・保存しない

Key Takeaways — まとめ

本番対応のランクトラッカーを構築するために必要な要素:

• データモデルは時系列（keyword, target_domain, country, device, position, captured_at）で設計する
• num=100廃止後はstartパラメータでページネーションする
• curl_cffiのimpersonate='chrome'でTLS/JA3-JA4フィンガープリントを回避する
• レジデンシャルプロキシ＋都市単位ジオターゲティング＋キーワードごとのスティッキーセッションが成功率95%以上の鍵
• 指数バックオフ、CAPTCHA検出、セマファによる並行制限で本番運用をハードニングする
• 7日間移動平均でノイズを平滑化し、トレンドを可視化する

ProxyHatのレジデンシャルプロキシを使えば、gate.proxyhat.com:8080に接続するだけで都市単位のジオターゲティングとIPローテーションが利用できます。まずは少人数のキーワードセットで1週間のスナップショットを蓄積し、ボラティリティのベースラインを確立することをお勧めします。