Pengumpulan Data Pasar Saham dengan Penanganan CAPTCHA

CAPTCHA di portal keuangan hampir tidak pernah muncul secara acak. Ia datang pada tiga momen yang bisa diprediksi: saat Anda memeriksa puluhan simbol dalam hitungan detik, saat mengunduh riwayat harga dalam jumlah besar, dan saat menjalankan kueri screener dengan banyak filter. Begitu ketiga momen itu terpetakan, penanganannya berhenti menjadi tebak-tebakan dan berubah menjadi satu langkah tambahan di dalam pipeline: deteksi tantangan, kirim ke CaptchaAI, ambil token, ulangi request.

Artikel ini menunjukkan pola tersebut untuk Cloudflare Turnstile dan reCAPTCHA v2 pada portal keuangan, lengkap dengan kode Python dan Node.js yang bisa langsung Anda adaptasi ke kolektor yang sudah berjalan.

Tiga pemicu CAPTCHA di portal keuangan

Sebelum menambahkan solver, cocokkan dulu jenis data yang Anda ambil dengan tantangan yang biasanya menghadangnya. Pemetaan ini menentukan method API mana yang Anda panggil dan berapa lama satu siklus bisa tertahan.

Tetapkan ritme pengumpulan sebelum menulis kode

Sebagian besar volume CAPTCHA yang ditemui tim data adalah hasil polling yang terlalu rapat, bukan hasil pengetatan proteksi. Kolektor yang menarik kutipan setiap menit selama 24 jam penuh akan memicu jauh lebih banyak tantangan daripada kolektor yang aktif pada jam bursa 09:00–16:00 WIB lalu beralih ke satu snapshot penutupan.

Menjadwalkan job berat di luar jam sibuk juga menurunkan latensi. Tim yang menjalankan worker di AWS ap-southeast-3 (Jakarta) atau ap-southeast-1 (Singapura) biasanya melihat waktu round-trip yang lebih pendek dibanding region Amerika, dan selisih itu berlipat ketika satu siklus berisi ratusan simbol.

Kolektor kutipan dan harga historis dengan Python

Kelas berikut membungkus pola empat langkah yang sama pada setiap request: kirim task ke in.php, simpan task ID, polling res.php, lalu ulangi request dengan token. Deteksi tantangan dijalankan lebih dulu, sehingga solver hanya dipanggil ketika benar-benar diperlukan.

import requests
import time
import re
from datetime import datetime, timedelta

class StockDataCollector:
    def __init__(self, api_key):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })

    def get_quote(self, portal_url, symbol):
        """Get current stock quote, solving CAPTCHAs if needed."""
        url = f"{portal_url}/quote/{symbol}"
        response = self.session.get(url)

        if self._is_captcha_page(response):
            response = self._solve_and_retry(response, url)

        return self._parse_quote(response.text, symbol)

    def get_historical(self, portal_url, symbol, days=365):
        """Download historical price data."""
        url = f"{portal_url}/history/{symbol}"
        params = {
            "period": f"{days}d",
            "interval": "1d"
        }
        response = self.session.get(url, params=params)

        if self._is_captcha_page(response):
            response = self._solve_and_retry(response, url)

        return self._parse_historical(response.text)

    def scan_symbols(self, portal_url, symbols, delay=2):
        """Collect quotes for multiple symbols."""
        results = {}

        for symbol in symbols:
            try:
                results[symbol] = self.get_quote(portal_url, symbol)
                time.sleep(delay)
            except Exception as e:
                results[symbol] = {"error": str(e)}

        return results

    def _is_captcha_page(self, response):
        return (
            response.status_code == 403 or
            "cf-turnstile" in response.text or
            "challenges.cloudflare.com" in response.text
        )

    def _solve_and_retry(self, response, url):
        match = re.search(r'data-sitekey="(0x[^"]+)"', response.text)
        if not match:
            # Fall back to reCAPTCHA detection
            match = re.search(r'data-sitekey="([^"]+)"', response.text)
            if match:
                return self._solve_recaptcha_and_retry(match.group(1), url)
            raise ValueError("No CAPTCHA sitekey found")

        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "turnstile",
            "sitekey": match.group(1),
            "pageurl": url,
            "json": 1
        })
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(3)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = result.json()
            if data["status"] == 1:
                return self.session.post(url, data={
                    "cf-turnstile-response": data["request"]
                })

        raise TimeoutError("CAPTCHA solve timed out")

    def _solve_recaptcha_and_retry(self, site_key, url):
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
            "json": 1
        })
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(3)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = result.json()
            if data["status"] == 1:
                return self.session.post(url, data={
                    "g-recaptcha-response": data["request"]
                })

        raise TimeoutError("reCAPTCHA solve timed out")

    def _parse_quote(self, html, symbol):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")

        def text_or_none(node):
            return node.text.strip() if node and node.text else None

        return {
            "symbol": symbol,
            "price": text_or_none(soup.select_one("[data-field='regularMarketPrice'], .price")),
            "change": text_or_none(soup.select_one("[data-field='regularMarketChange'], .change")),
            "volume": text_or_none(soup.select_one("[data-field='regularMarketVolume'], .volume")),
            "market_cap": text_or_none(soup.select_one("[data-field='marketCap'], .market-cap")),
            "timestamp": datetime.now().isoformat()
        }

    def _parse_historical(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        rows = []

        for row in soup.select("table tr")[1:]:  # Skip header
            cells = [td.text.strip() for td in row.select("td")]
            if len(cells) >= 6:
                rows.append({
                    "date": cells[0],
                    "open": cells[1],
                    "high": cells[2],
                    "low": cells[3],
                    "close": cells[4],
                    "volume": cells[5]
                })

        return rows


# Usage
collector = StockDataCollector("YOUR_API_KEY")

# Single quote
quote = collector.get_quote("https://finance.example.com", "AAPL")
print(f"AAPL: ${quote['price']} ({quote['change']})")

# Scan multiple symbols
portfolio = collector.scan_symbols(
    "https://finance.example.com",
    ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"]
)

Dua hal membuat kolektor ini bertahan lama. Pertama, requests.Session() dipertahankan di seluruh siklus, jadi cookie hasil penyelesaian ikut terbawa dan halaman berikutnya sering lolos tanpa tantangan baru. Kedua, _solve_and_retry mencoba pola sitekey Turnstile (0x...) lebih dulu lalu jatuh ke jalur reCAPTCHA, sehingga satu kolektor tetap berfungsi walau portal mengganti proteksinya di tengah jalan.

Screener pasar dengan penanganan Turnstile di Node.js

Kueri screener adalah pemicu paling konsisten karena setiap kombinasi filter menghasilkan URL baru yang belum pernah dilihat cache. Implementasi Node.js berikut memakai alur yang identik dengan versi Python, hanya dengan fetch dan URLSearchParams.

class MarketScreener {
  constructor(apiKey) {
    this.apiKey = apiKey;
  }

  async screenStocks(portalUrl, filters) {
    const params = new URLSearchParams(filters);
    const response = await fetch(`${portalUrl}/screener?${params}`);
    const html = await response.text();

    if (html.includes('cf-turnstile') || response.status === 403) {
      return this.solveAndScreen(portalUrl, filters, html);
    }

    return this.parseScreenerResults(html);
  }

  async solveAndScreen(portalUrl, filters, html) {
    const match = html.match(/data-sitekey="(0x[^"]+)"/);
    if (!match) throw new Error('Turnstile sitekey not found');

    const submitResp = await fetch('https://ocr.captchaai.com/in.php', {
      method: 'POST',
      body: new URLSearchParams({
        key: this.apiKey,
        method: 'turnstile',
        sitekey: match[1],
        pageurl: portalUrl,
        json: '1'
      })
    });
    const { request: taskId } = await submitResp.json();

    for (let i = 0; i < 60; i++) {
      await new Promise(r => setTimeout(r, 3000));
      const result = await fetch(
        `https://ocr.captchaai.com/res.php?key=${this.apiKey}&action=get&id=${taskId}&json=1`
      );
      const data = await result.json();
      if (data.status === 1) {
        const response = await fetch(`${portalUrl}/screener`, {
          method: 'POST',
          body: new URLSearchParams({
            ...filters,
            'cf-turnstile-response': data.request
          })
        });
        return this.parseScreenerResults(await response.text());
      }
    }
    throw new Error('Turnstile solve timed out');
  }

  parseScreenerResults(html) {
    const rows = [];
    const tableMatch = html.match(/<table[^>]*>[\s\S]*?<\/table>/i);
    if (!tableMatch) return rows;

    const rowMatches = tableMatch[0].matchAll(/<tr[^>]*>([\s\S]*?)<\/tr>/gi);
    for (const row of rowMatches) {
      const cells = [...row[1].matchAll(/<td[^>]*>([\s\S]*?)<\/td>/gi)]
        .map(m => m[1].replace(/<[^>]+>/g, '').trim());
      if (cells.length >= 4) {
        rows.push({
          symbol: cells[0],
          price: cells[1],
          change: cells[2],
          volume: cells[3]
        });
      }
    }
    return rows;
  }
}

// Usage
const screener = new MarketScreener('YOUR_API_KEY');
const results = await screener.screenStocks('https://finance.example.com', {
  sector: 'technology',
  marketCap: 'large',
  peRatio: '<25'
});

Perhatikan bahwa hasil solve dikirim sebagai field cf-turnstile-response pada POST berikutnya, bukan ditempelkan ke URL. Jika portal tetap menolak dengan 403 setelah token dikirim, penyebabnya hampir selalu pageurl yang tidak sama persis dengan halaman tempat sitekey ditemukan.

Menghitung kebutuhan thread untuk pipeline data pasar

CaptchaAI menagih per thread bersamaan, bukan per solve, dan setiap paket mencakup solve tanpa batas selama bulan berjalan. Satu thread berarti satu CAPTCHA yang sedang dikerjakan; begitu selesai, thread itu langsung menerima antrean berikutnya.

Untuk pekerjaan data pasar, angka yang menentukan adalah berapa banyak tantangan yang berjalan bersamaan pada puncak siklus, bukan total tantangan per hari. Kolektor satu portal dengan beberapa simbol per menit umumnya nyaman di BASIC ($15/bulan, 5 thread). Agensi price monitoring atau pekerja lepas yang menangani beberapa klien sekaligus biasanya butuh STANDARD ($30/bulan, 15 thread), sedangkan pipeline yang memindai ratusan simbol plus screener harian lebih tepat di ADVANCE ($90/bulan, 50 thread). Biaya bulanan yang tetap inilah yang membuat model thread masuk akal untuk tim dengan volume naik-turun mengikuti musim laporan keuangan.

Masalah umum saat mengumpulkan data pasar

Catatan kepatuhan untuk tim data di Indonesia

Kumpulkan hanya data pasar yang memang dipublikasikan untuk umum, dan hindari data pribadi pengguna. UU Pelindungan Data Pribadi (UU 27/2022) dan UU ITE membuat pemisahan itu penting bagi tim yang menyimpan hasil scraping dalam jangka panjang. Patuhi pula syarat layanan portal yang Anda akses; artikel ini bukan nasihat hukum.

Pertanyaan Umum

Apakah CaptchaAI mendukung semua CAPTCHA yang muncul di situs keuangan?

Tidak semuanya. Yang didukung mencakup reCAPTCHA v2 dan v3 (termasuk versi Enterprise), Cloudflare Turnstile, Cloudflare Challenge, image CAPTCHA/OCR, grid image, dan GeeTest v3. hCaptcha dan FunCaptcha belum didukung, GeeTest v4 berstatus segera hadir, sedangkan CaptchaFox (beta), Friendly Captcha (beta), dan Lemin (beta) masih tahap beta.

Seberapa besar penyelesaian CAPTCHA memperlambat satu siklus pengumpulan?

Cloudflare Turnstile biasanya selesai di bawah 10 detik dan reCAPTCHA v2 di bawah 60 detik, dengan tingkat keberhasilan tinggi pada tipe yang didukung. Karena tantangan hanya muncul pada sebagian request, dampaknya terhadap siklus penuh kecil — apalagi jika Anda memproses beberapa simbol secara paralel sesuai jumlah thread paket Anda.

Bisakah pipeline ini berjalan tanpa browser?

Bisa. Kedua contoh di atas hanya memakai HTTP client biasa (requests dan fetch), tanpa Selenium maupun browser headless. Anda mengirim sitekey dan pageurl ke in.php, menerima token, lalu memasukkannya ke request berikutnya — cocok untuk worker kecil, cron job, maupun fungsi serverless.

Bagaimana menjaga konsistensi dataset ketika solve gagal atau timeout?

Jangan menulis nilai kosong ke tabel harga. Tandai baris sebagai belum lengkap, catat task ID beserta simbolnya, lalu jadwalkan percobaan ulang dengan jeda yang meningkat eksponensial. Kolektor yang memisahkan "gagal diambil" dari "harga nol" jauh lebih mudah diaudit ketika laporan harian tidak cocok.

Artikel Terkait

• Pengumpulan data untuk riset pasar
• Perbandingan GeeTest dan Cloudflare Turnstile
• Mengatasi 403 Turnstile setelah token diperbaiki

Langkah Selanjutnya

Mulai dari API key: buat akun CaptchaAI, tempelkan API key Anda ke kolektor di atas, lalu jalankan satu siklus simbol malam ini.