Как создать и развернуть веб-скрепер акций на Python

TL;DR

Этот проект собирает ежедневные OHLCV через документированный API рыночных данных, валидирует записи и загружает их в SQLite.
Включенная фикция работает без учетных данных и содержит синтетические цены, а не рыночные наблюдения.
Даты представляют собой рыночные сессии; время получения, валюта, биржа и основание корректировки являются отдельными полями.
Развертывание использует однократный контейнер Docker и внешнее расписание, с постоянной базой данных и секретами вне образа.
Доступ к живому API, права на рыночные данные и развертывание контейнера требуют проверки в вашем окружении. Локальная фикция не подтверждает живые права или актуальность в реальном времени.

Что вы построите: Скрапер данных фондового рынка

Вы создадите ежедневный сборщик рыночных данных: символ → поддерживаемый API → валидация → SQLite → запланированное обновление. Та же структура инженерии может поддерживать авторизованный HTML источник, но документированный API избегает ненадежных селекторов страниц котировок.

Учебник использует документированный контракт TIME_SERIES_DAILY от Alpha Vantage. Это сырые ежедневные бары, а не живые исполняемые цены или автоматически откорректированный ряд полной доходности. Данные могут быть задержаны или историческими в зависимости от прав на источник; этот проект не является торговой лентой.

Предварительные требования и настройка проекта

Используйте Python 3.12, предоставленный скрипт и фикцию, а также Docker, если хотите путь развертывания. Python проект использует только стандартную библиотеку.

  stock_scraper.py
  stock_fixture.json
  Dockerfile
  .dockerignore
  requirements.txt

Поле	Значение в этом проекте	Проблема валидации
Символ и биржа	Идентичность инструмента, специфичная для источника	Один и тот же короткий тикер может идентифицировать разные инструменты
Дата сессии	Торговая дата, связанная с ежедневным баром	Не дата UTC получения
Открытие/максимум/минимум/закрытие	Сырые цены сессии	Положительные, конечные, внутренне согласованные
Объем	Отчетные единицы для инструмента/сессии	Неотрицательные; определение единицы зависит от источника
Валюта	Явно предоставленные метаданные инструмента	Не делайте выводы по символу в одиночку
Основание	raw_daily	Храните отдельно от данных с корректировкой сплита/дивидендов
Наблюдаемое в	Временная метка UTC сборщика	Отслеживает свежесть получения

"""Ежедневный сборщик OHLCV. По умолчанию требуется явная локальная фикция."""
import argparse
import json
import logging
import os
import re
import sqlite3
import time
from datetime import date, datetime, timezone
from decimal import Decimal
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen


def fetch(symbol):
key = os.environ["ALPHAVANTAGE_API_KEY"]
    query = urlencode(dict(function="TIME_SERIES_DAILY", symbol=symbol,
                           outputsize="compact", datatype="json", apikey=key))
    for attempt in range(3):
        try:
            request = Request("https://www.alphavantage.co/query?" + query,
                              headers={"User-Agent": "DailyDataCollector/1.0"})
            with urlopen(request, timeout=30) as response:
                return json.load(response)
        except HTTPError as exc:
            # Никогда не выводите exc или его URL: запрос содержит API ключ.
            if exc.code not in (429, 502, 503, 504) or attempt == 2:
                raise RuntimeError(f"Ошибка HTTP источника: {exc.code}") from None
            delay = exc.headers.get("Retry-After", "")
            if delay:
                # Запланировать более позднюю попытку; никогда не пытайтесь раньше, чем запрашивалось.
                raise RuntimeError("Превышен лимит запросов; перенастройтесь, используя Retry-After") from None
        except (URLError, TimeoutError):
            if attempt == 2:
                raise RuntimeError("Источник недоступен после трех попыток") from None
        time.sleep(2 ** attempt)


def normalize(payload, symbol, currency, exchange):
    if not re.fullmatch(r"[A-Z]{3}", currency) or not exchange.strip():
        raise ValueError("Укажите действительный код валюты и идентификатор биржи")
    if payload.get("Meta Data", {}).get("2. Symbol", "").upper() != symbol.upper():
        raise ValueError("Отсутствует или несовпадающий символ провайдера")
    series = payload.get("Time Series (Daily)")
    if not isinstance(series, dict) or not series:
        raise ValueError("Нет дневных баров; проверьте лимиты аккаунта или ошибку источника")
    observed = datetime.now(timezone.utc).isoformat()
    rows = []
    for session, bar in sorted(series.items()):
        date.fromisoformat(session)
        prices = [Decimal(bar[f"{i}. {name}"]) for i, name in
                  ((1, "open"), (2, "high"), (3, "low"), (4, "close"))]
        if any(not p.is_finite() or p <= 0 for p in prices):
            raise ValueError(f"Недопустимые OHLC: {session}")
        opening, high, low, close = prices
        if low > min(opening, close) or high < max(opening, close) or low > high:
            raise ValueError(f"Несоответствующая свеча: {session}")
        volume = int(bar["5. volume"])
        if volume < 0:
            raise ValueError("Отрицательный объем")
        rows.append((symbol, exchange, currency, "raw_daily", session,
                     *(str(p) for p in prices), volume, "alpha_vantage", observed))
    return rows


def save(rows, database):
    with sqlite3.connect(database) as db:
        db.execute("""CREATE TABLE IF NOT EXISTS bars(
            symbol TEXT, exchange TEXT, currency TEXT, basis TEXT, session_date TEXT,
            open TEXT, high TEXT, low TEXT, close TEXT, volume INTEGER,
            source TEXT, observed_at TEXT,
            PRIMARY KEY(symbol,exchange,currency,basis,session_date))""")
        db.executemany("""INSERT INTO bars VALUES(?,?,?,?,?,?,?,?,?,?,?,?)
            ON CONFLICT(symbol,exchange,currency,basis,session_date) DO UPDATE SET
            open=excluded.open,high=excluded.high,low=excluded.low,
            close=excluded.close,volume=excluded.volume,observed_at=excluded.observed_at
            """, rows)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--symbol", default="IBM")
    parser.add_argument("--currency", required=True)
    parser.add_argument("--exchange", required=True)
    parser.add_argument("--fixture")
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--db", default="stocks.sqlite")
    args = parser.parse_args()
    if bool(args.fixture) == args.live:
        parser.error("Выберите точно одно из --fixture или --live")
    payload = json.loads(Path(args.fixture).read_text()) if args.fixture else fetch(args.symbol)
    rows = normalize(payload, args.symbol, args.currency, args.exchange)
    save(rows, args.db)
    logging.info("accepted_bars=%s latest_session=%s", len(rows), rows[-1][4])


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    try:
        main()
    except (ValueError, KeyError, RuntimeError, OSError) as exc:
        logging.error("collector_failed: %s", type(exc).__name__)
        raise SystemExit(1)

Как создать и развернуть веб-скребок для акций: Практическое руководство

TL;DR

Что вы построите: Скрапер данных фондового рынка

Предварительные требования и настройка проекта

Популярные статьи

Сбор данных о недвижимости: Руководство по сбору данных о государственной собственности

Dify SERP Scraping: Как создать рабочий процесс данных поиска

Выбор надежного источника данных акций

API фондового рынка против веб-скрейпинга

Понимание полей данных акций

Шаг 1: Создайте проект Python Scraper

Шаг 2: Получите данные акций из источника

Шаг 3: Парсинг и проверка цен на акции

Шаг 4: Сохранение данных акций в CSV или SQLite

Шаг 5: Добавление журналирования, попыток и обработки ошибок

Шаг 6: Запланировать автоматические обновления данных акций

Шаг 7: Развертывание скрапера в производственной среде

Dockerize скрапер акций

Настройка переменных среды и секретов

Настройка планирования и мониторинга

Тестирование развернутого скрапера акций

Масштабирование от одной акции до сотен тикеров

Общие проблемы и решения при скрейпинге акций

Затраты на скрейпинг данных об акциях и юридические соображения

Финальный контрольный список проекта

Подтвердите одобренные источники страниц

Часто задаваемые вопросы