プロキシを使用したパース時のAPIレート制限を回避する方法：ローテーション設定とIP選択

あなたはパーサーを設定し、プロキシを接続しましたが、APIは依然として「429 Too Many Requests」エラーを返したり、アクセスをブロックしたりしますか？問題はプロキシそのものではなく、それらの使用方法の戦略が間違っていることです。レート制限は、特定の期間内に1つのIPアドレスからのリクエスト数を制限するAPIの保護メカニズムです。この記事では、プロキシを介して作業する際にブロックが発生する理由と、制限を回避するためのシステムの正しい設定方法を説明します。

APIレート制限とは何か、どのように機能するか

レート制限（リクエスト頻度制限）は、APIが過負荷や悪用から保護するためのメカニズムです。サービスは、特定の時間枠内で1つのソースから実行できるリクエストの数に制限を設けます。たとえば、人気のあるAPIは次のような制限を使用します：

• Twitter API： 標準アクセスで15分あたり300リクエスト
• Instagram Graph API： アプリケーションあたり1時間に200リクエスト
• Google Maps API： プランによって異なり、通常は1日あたり100〜1000リクエスト
• Wildberries API： 非公式の制限は、1つのIPから1分あたり約60リクエスト
• Avito API： 広告をパースするために1秒あたり10リクエスト

レート制限が適用されるリクエストのソースを特定する方法はいくつかあります：

制限を超えると、APIはHTTPステータス429「Too Many Requests」を返し、制限リセットまでの時間を示すRetry-Afterヘッダーを返します。一部のサービスは「ローリングウィンドウ」を使用しており、制限が徐々に更新されるのに対し、他のサービスは特定の時間にリセットされる固定ウィンドウを使用します。

なぜプロキシは自動的にレート制限から救ってくれないのか

多くの開発者は、プロキシを接続するだけで無制限のリクエストを送信できると誤解しています。実際には次のような問題が発生します：

すべてのリクエストに同じプロキシを使用する

スクリプトがすべてのリクエストに同じプロキシのIPアドレスを使用している場合、APIはこれを通常のユーザーとして認識し、標準の制限を適用します。たとえば、あなたがWildberriesを介して1つのレジデンシャルプロキシを使用して価格パーサーを設定したとします。パーサーは1分あたり100リクエストを行いますが、制限は60リクエストです。結果：IPが10〜30分間ブロックされます。

遅いIPローテーション

一部の人は5〜10のプロキシのプールを使用し、それらの間で順番に切り替えます。問題は、各IPが完全なローテーションが行われる前に制限に達してしまうことです。たとえば、あなたが10のプロキシを持ち、IPあたりの制限が1時間に100リクエストの場合、1時間に1000リクエストを行うと、各プロキシは100リクエストを受け取ります—ちょうど制限の境界で。分配の不均等はブロックを引き起こします。

他の識別要因の無視

IPの理想的なローテーションがあっても、次のような場合にはブロックされる可能性があります：

• すべてのリクエストが同じUser-Agentから来ている（例：python-requests/2.28.0）
• すべてのリクエストに同じAPIキーを使用している
• リクエストが理想的な周期性で来る（0.5秒ごと）— これはボットのように見えます
• プロキシのIPアドレスが同じサブネットにある（例：すべて192.168.1.xの範囲内）

IPアドレスの評判

データセンターのプロキシは、他のユーザーがパースのために使用するため、しばしばブラックリストに載ります。APIはそのようなアドレスに対してより厳しい制限を適用したり、即座にブロックしたりすることがあります。たとえば、InstagramやFacebookは、公式の制限を超えなくてもデータセンターを積極的に禁止します。

制限を回避するためのIPローテーション戦略

正しいプロキシのローテーションは、レート制限を回避する鍵です。タスクに応じた効果的な戦略を見てみましょう。

各リクエスト後のローテーション

最も攻撃的な戦略：各リクエストが新しいIPを経由します。これは、非常に厳しい制限（IPあたり1〜5リクエスト）や、負荷を最大限に分散させる必要がある場合に適しています。これには、自宅用プロキシの自動ローテーションを使用します—これにより、数百万のIPのプールが提供され、各リクエストは自動的に新しいアドレスを取得します。

使用例：1つのIPから1分あたり5リクエストの制限がある非公式APIを介してInstagramをパースする場合、各リクエスト後にローテーションを行うことで、300の異なるIPを介して1分あたり300リクエストを行うことができます。

時間によるローテーション（スティッキーセッション）

IPアドレスは特定の時間（5〜30分）使用され、その後新しいものに変更されます。この戦略は、セッションの保持が必要なAPIや、1人の「ユーザー」からの関連リクエストをいくつか行う必要がある場合に適しています。

最適なローテーション時間の計算：APIの制限が1時間に100リクエストで、1つのIPを介して50リクエストを行う予定がある場合、30分のスティッキーセッションを使用します。この時間内に25リクエストを行うことができ（均等な負荷の場合）、制限の半分になります。

制限を考慮したプールによるローテーション

進んだ戦略：スクリプトは各IPからのリクエスト数を追跡し、制限に近づくと自動的に新しいものに切り替えます。たとえば、20のプロキシのプールがあり、APIの制限が1時間に100リクエストの場合、スクリプトは各IPのカウンターを追跡し、90リクエストに達すると次に切り替えます。

この戦略はロジックのプログラミングを必要としますが、最大の効率を提供します：各プロキシを最大限に活用し、制限を超えません。

地理的ローテーション

一部のAPIは地域によって異なる制限を適用します。たとえば、サービスは米国からのリクエストをヨーロッパからのリクエストよりも厳しく制限する場合があります。このような場合は、異なる国のプロキシを使用し、それらの間で負荷を分散させてください。

リクエスト間の遅延設定

理想的なIPローテーションがあっても、リクエスト間の遅延を正しく設定することが重要です。あまりにも速いリクエストは攻撃のように見えます、たとえそれらが異なるIPから来ていても。

最小遅延の計算

フォーミュラ： 遅延 = (時間枠（秒） / リクエスト制限) × セキュリティ係数

例：APIは1時間に100リクエストを許可します（3600秒）。最小遅延 = 3600 / 100 = 36秒。セキュリティ係数1.2を追加します：36 × 1.2 = 43秒、1つのIPからのリクエスト間の遅延。

10のプロキシを使用してローテーションする場合、1つのIPで制限を超えずに4.3秒ごとにリクエストを行うことができます（43 / 10）。

ランダムな遅延（ジッター）

固定の5秒の遅延の代わりに、3〜7秒のランダムな間隔を使用します。これにより、トラフィックが実際のユーザーの行動に似たものになります。多くのボット防止システムはパターンを分析します：リクエストが正確に5.0秒ごとに来る場合、これは疑わしいです。

エラー時の指数遅延

エラー429を受け取った場合、すぐにリクエストを送信し続けないでください。指数遅延を使用します：最初の試行は1秒後、2回目は2秒後、3回目は4秒後、4回目は8秒後、そしてその後も続けます。これは、APIがクライアントから期待する標準的なプラクティスです。

APIで使用するプロキシの種類を選ぶには

プロキシの種類の選択は、レート制限を回避する成功に重要な影響を与えます。APIでの各オプションの利点と欠点を見てみましょう。

自宅用プロキシ

自宅用プロキシは、インターネットプロバイダーによって割り当てられた実際のユーザーのIPアドレスを使用します。APIにとっては、これは通常の家庭用インターネットのように見えます。

使用するタイミング： Instagram、Facebook、TikTokのパース、マーケットプレイスAPI（Wildberries、Ozon）との作業、IPの評判が重要なタスク。

モバイルプロキシ

モバイルプロキシは、モバイルオペレーター（4G/5G）のIPを使用します。1つのIPは、同時に数千の実際のユーザーによって使用されることが多いため、APIはそれらをブロックすることは非常に稀です。

使用するタイミング： モバイルAPIとの作業、大量のInstagram/TikTokのパース、ブロックからの最大の保護が必要な場合。

データセンターのプロキシ

データセンターのプロキシは、データセンターに配置されたサーバーのIPアドレスです。これらは実際のユーザーとは関係ありません。

使用するタイミング： 厳しい保護がない公開APIのパース（天気、為替レート、ニュース）、独自のAPIとの作業、テストや開発。

実践的な実装：Pythonのコード例

プロキシを使用してレート制限を回避する具体的な実装例を見てみましょう。すべての例は、requestsライブラリを使用したPythonで書かれています。

プロキシプールからの単純なローテーション

リストからのIPの循環ローテーションを使用した基本的な実装：

import requests
import time
from itertools import cycle

# プロキシのリスト（形式：protocol://user:pass@host:port）
PROXY_LIST = [
    'http://user1:pass1@proxy1.example.com:8080',
    'http://user2:pass2@proxy2.example.com:8080',
    'http://user3:pass3@proxy3.example.com:8080',
]

# 循環イテレータを作成
proxy_pool = cycle(PROXY_LIST)

def make_request(url):
    proxy = next(proxy_pool)  # プールから次のプロキシを取得
    proxies = {
        'http': proxy,
        'https': proxy
    }
    
    try:
        response = requests.get(url, proxies=proxies, timeout=10)
        return response
    except requests.exceptions.RequestException as e:
        print(f"プロキシ {proxy} に関するエラー: {e}")
        return None

# 使用例
for i in range(10):
    response = make_request('https://api.example.com/data')
    if response and response.status_code == 200:
        print(f"リクエスト {i+1}: 成功")
    time.sleep(2)  # リクエスト間の遅延

制限を追跡するローテーション

各プロキシのリクエストをカウントし、制限に近づくと切り替えるより進んだバージョン：

import requests
import time
from collections import defaultdict

class ProxyRotator:
    def __init__(self, proxy_list, max_requests_per_ip=90, time_window=3600):
        self.proxy_list = proxy_list
        self.max_requests = max_requests_per_ip  # IPあたりのリクエスト制限
        self.time_window = time_window  # 秒単位の時間枠
        self.request_counts = defaultdict(list)  # 各IPのリクエスト履歴
        self.current_index = 0
    
    def get_proxy(self):
        """リクエスト数が最も少ないプロキシを返す"""
        current_time = time.time()
        
        # 時間枠を超えた古い記録をクリア
        for proxy in self.request_counts:
            self.request_counts[proxy] = [
                t for t in self.request_counts[proxy]
                if current_time - t < self.time_window
            ]
        
        # リクエスト数が最も少ないプロキシを探す
        available_proxies = []
        for proxy in self.proxy_list:
            count = len(self.request_counts[proxy])
            if count < self.max_requests:
                available_proxies.append((proxy, count))
        
        if not available_proxies:
            # すべてのプロキシが制限に達した場合、待機
            oldest_request = min(
                min(times) for times in self.request_counts.values() if times
            )
            wait_time = self.time_window - (current_time - oldest_request) + 1
            print(f"すべてのプロキシが制限に達しました。{wait_time:.0f}秒待機...")
            time.sleep(wait_time)
            return self.get_proxy()
        
        # リクエスト数が最も少ないプロキシを選択
        proxy = min(available_proxies, key=lambda x: x[1])[0]
        self.request_counts[proxy].append(current_time)
        return proxy
    
    def make_request(self, url, **kwargs):
        proxy = self.get_proxy()
        proxies = {'http': proxy, 'https': proxy}
        
        try:
            response = requests.get(url, proxies=proxies, timeout=10, **kwargs)
            return response
        except requests.exceptions.RequestException as e:
            print(f"プロキシ {proxy} に関するエラー: {e}")
            return None

# 使用例
PROXY_LIST = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
]

rotator = ProxyRotator(PROXY_LIST, max_requests_per_ip=100, time_window=3600)

for i in range(500):  # 500リクエストを行う
    response = rotator.make_request('https://api.example.com/data')
    if response and response.status_code == 200:
        print(f"リクエスト {i+1}: 成功")
    time.sleep(1)  # 最小遅延

エラー429の処理と指数遅延

"Too Many Requests"の応答を正しく処理し、Retry-Afterヘッダーを考慮する：

import requests
import time

def make_request_with_retry(url, proxies, max_retries=5):
    """エラー429の場合に自動的に再試行するリクエストを行う"""
    
    for attempt in range(max_retries):
        try:
            response = requests.get(url, proxies=proxies, timeout=10)
            
            if response.status_code == 200:
                return response
            
            elif response.status_code == 429:
                # Retry-Afterヘッダーを確認
                retry_after = response.headers.get('Retry-After')
                
                if retry_after:
                    wait_time = int(retry_after)
                    print(f"レート制限。{wait_time}秒待機（Retry-Afterから）")
                else:
                    # 指数遅延：2^attempt秒
                    wait_time = 2 ** attempt
                    print(f"レート制限。試行 {attempt+1}、{wait_time}秒待機")
                
                time.sleep(wait_time)
                continue
            
            else:
                print(f"HTTPエラー {response.status_code}")
                return None
        
        except requests.exceptions.RequestException as e:
            print(f"接続エラー: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            return None
    
    print(f"試行回数が超過しました ({max_retries})")
    return None

# 使用例
proxies = {
    'http': 'http://user:pass@proxy.example.com:8080',
    'https': 'http://user:pass@proxy.example.com:8080'
}

response = make_request_with_retry('https://api.example.com/data', proxies)
if response:
    print("データを取得しました:", response.json())

自宅用プロキシの自動ローテーションを使用

多くの自宅用プロキシプロバイダーは、各リクエストごとにIPを自動的に変更する1つのエンドポイントを提供します。設定の例：

import requests
import random
import time

# 自宅用プロキシの自動ローテーション
# 形式：protocol://username:password@gateway:port
ROTATING_PROXY = 'http://customer-USER:PASS@proxy.provider.com:12321'

def make_request_rotating(url):
    """ローテーションプロキシを介してリクエストを行う（毎回新しいIP）"""
    proxies = {
        'http': ROTATING_PROXY,
        'https': ROTATING_PROXY
    }
    
    # より高い匿名性のためにランダムなUser-Agentを追加
    user_agents = [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
        'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',
    ]
    
    headers = {
        'User-Agent': random.choice(user_agents)
    }
    
    try:
        response = requests.get(url, proxies=proxies, headers=headers, timeout=15)
        return response
    except requests.exceptions.RequestException as e:
        print(f"エラー: {e}")
        return None

# 異なるIPを介して100リクエストを行う
for i in range(100):
    response = make_request_rotating('https://api.example.com/data')
    if response and response.status_code == 200:
        print(f"リクエスト {i+1}: 成功、IPが変更されました")
    
    # 1〜3秒のランダムな遅延
    time.sleep(random.uniform(1, 3))

制限の監視とエラー処理

APIとの効果的な作業には、制限の継続的な監視とエラーの正しい処理が必要です。以下は重要なプラクティスです：

応答ヘッダーの分析

多くのAPIは、応答ヘッダーに制限に関する情報を返します。標準的なヘッダー：

• X-RateLimit-Limit — ウィンドウ内の最大リクエスト数
• X-RateLimit-Remaining — 残りのリクエスト数
• X-RateLimit-Reset — 制限リセットの時間（Unixタイムスタンプ）
• Retry-After — 何秒後にリクエストを再試行できるか

これらのヘッダーを読む例：

response = requests.get(url, proxies=proxies)

# 制限ヘッダーを確認
limit = response.headers.get('X-RateLimit-Limit')
remaining = response.headers.get('X-RateLimit-Remaining')
reset_time = response.headers.get('X-RateLimit-Reset')

if remaining:
    remaining = int(remaining)
    if remaining < 10:
        print(f"注意！残りのリクエストはわずか {remaining} 件です")
        
if reset_time:
    import datetime
    reset_dt = datetime.datetime.fromtimestamp(int(reset_time))
    print(f"制限は {reset_dt} にリセットされます")

ロギングと統計

各プロキシのリクエストの詳細な統計を維持します。これにより、問題のあるIPを特定し、ローテーションを最適化できます：

import json
from datetime import datetime

class RequestLogger:
    def __init__(self):
        self.stats = {}
    
    def log_request(self, proxy, status_code, response_time):
        if proxy not in self.stats:
            self.stats[proxy] = {
                'total': 0,
                'success': 0,
                'rate_limited': 0,
                'errors': 0,
                'avg_response_time': 0
            }
        
        self.stats[proxy]['total'] += 1
        
        if status_code == 200:
            self.stats[proxy]['success'] += 1
        elif status_code == 429:
            self.stats[proxy]['rate_limited'] += 1
        else:
            self.stats[proxy]['errors'] += 1
        
        # 平均応答時間を更新
        current_avg = self.stats[proxy]['avg_response_time']
        total = self.stats[proxy]['total']
        self.stats[proxy]['avg_response_time'] = (
            (current_avg * (total - 1) + response_time) / total
        )
    
    def print_stats(self):
        print("\n=== プロキシの統計 ===")
        for proxy, data in self.stats.items():
            success_rate = (data['success'] / data['total'] * 100) if data['total'] > 0 else 0
            print(f"\nプロキシ: {proxy}")
            print(f"  総リクエスト数: {data['total']}")
            print(f"  成功数: {data['success']} ({success_rate:.1f}%)")
            print(f"  レート制限: {data['rate_limited']}")
            print(f"  エラー数: {data['errors']}")
            print(f"  平均応答時間: {data['avg_response_time']:.2f}s")

# 使用例
logger = RequestLogger()

start_time = time.time()
response = requests.get(url, proxies=proxies)
response_time = time.time() - start_time

logger.log_request(proxy, response.status_code, response_time)
logger.print_stats()

戦略の自動切り替え

もしあなたが常にエラー429を受け取る場合、リクエストを自動的に遅くするか、プロキシプールを増やします：

class AdaptiveRateLimiter:
    def __init__(self, initial_delay=1.0):
        self.delay = initial_delay
        self.consecutive_429 = 0
    
    def on_success(self):
        """成功したリクエスト - 少しスピードアップできます"""
        self.consecutive_429 = 0
        self.delay = max(0.5, self.delay * 0.95)  # 遅延を5%減少させます
    
    def on_rate_limit(self):
        """429を受け取った - 遅くする必要があります"""
        self.consecutive_429 += 1
        self.delay *= 1.5  # 遅延を1.5倍に増加させます
        
        if self.consecutive_429 > 5:
            print("注意: 429エラーが多すぎます。設定を確認してください！")
    
    def wait(self):
        """次のリクエストの前に待機します"""
        time.sleep(self.delay)
        return self.delay

# 使用例
limiter = AdaptiveRateLimiter(initial_delay=2.0)

for i in range(1000):
    response = make_request(url)
    
    if response.status_code == 200:
        limiter.on_success()
    elif response.status_code == 429:
        limiter.on_rate_limit()
    
    delay = limiter.wait()
    print(f"リクエスト {i+1}、遅延: {delay:.2f}s")

キャプチャやその他のブロックの処理

一部のAPIは、制限を超えた場合に直接ブロックするのではなく、キャプチャを表示します。兆候：

• レスポンスボディに「captcha」または「recaptcha」を含む403ステータスコード
• キャプチャページへのリダイレクト（302ステータス）
• X-Captcha-Required: trueのような特別なヘッダー

このような場合は、次のことを行う必要があります：

1. このIPの使用を直ちに中止する
2. プール内の別のプロキシに切り替える
3. リクエスト間の遅延を増やす
4. User-Agentや他のヘッダーにもっと多様性を追加する

結論

プロキシを使用したAPIレート制限の回避は、単なる技術的な設定ではなく、プロキシの種類の正しい選択、IPのローテーションの設定、遅延の管理、制限の監視を含む包括的な戦略です。この記事からの重要な結論：

• プロキシ自体はレート制限の問題を解決しません—正しいローテーション戦略が必要です
• ソーシャルメディアや厳しいAPIには自宅用またはモバイルプロキシを使用し、公開APIにはデータセンターが適しています
• APIの制限と希望するパース速度に基づいてプロキシプールのサイズを計算してください
• 常に適切な遅延を設定し、エラーを監視してください