Cách Tạo và Triển Khai Một Công Cụ Thu Thập Dữ Liệu Chứng Khoán Trong Python

TL;DR

Dự án này thu thập OHLCV hàng ngày thông qua một API dữ liệu thị trường đã được tài liệu, xác thực các bản ghi, và cập nhật chúng vào SQLite.
Tệp đính kèm chạy mà không cần thông tin xác thực và chứa giá tổng hợp, không phải quan sát thị trường.
Ngày tháng đại diện cho các phiên giao dịch; thời gian truy xuất, tiền tệ, sàn giao dịch, và cơ sở điều chỉnh là các trường riêng biệt.
Triển khai sử dụng một container Docker một lần và một lịch trình bên ngoài, với một cơ sở dữ liệu lâu dài và bí mật bên ngoài hình ảnh.
Truy cập API trực tiếp, quyền dữ liệu thị trường, và việc triển khai container cần được xác thực trong môi trường của bạn. Tệp đính kèm địa phương không chứng minh quyền lợi sống hoặc tính mới trong thời gian thực.

Những Gì Bạn Sẽ Xây Dựng: Một Trình Thu Thập Dữ Liệu Thị Trường Chứng Khoán

Bạn sẽ xây dựng một trình thu thập dữ liệu thị trường hàng ngày: biểu tượng → API được hỗ trợ → xác thực → SQLite → làm mới theo lịch trình. Cấu trúc kỹ thuật tương tự có thể hỗ trợ một nguồn HTML được ủy quyền, nhưng một API đã được tài liệu tránh được các trình chọn trang báo giá dễ bị hỏng.

Hướng dẫn sử dụng hợp đồng TIME_SERIES_DAILY đã được tài liệu của Alpha Vantage. Đây là các thanh hàng ngày thô chứ không phải giá có thể thực thi ngay lập tức hoặc một chuỗi tổng lợi nhuận tự động điều chỉnh. Dữ liệu có thể bị trễ hoặc lịch sử tùy thuộc vào quyền lợi nguồn; dự án này không phải là một chuỗi cung cấp giao dịch.

Các Yêu Cầu và Thiết Lập Dự Án

Sử dụng Python 3.12, tệp kịch bản và tệp đính kèm được cung cấp, và Docker nếu bạn muốn con đường triển khai. Dự án Python chỉ sử dụng thư viện tiêu chuẩn.

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

Trường	Ý nghĩa trong dự án này	Mối quan tâm xác thực
Biểu tượng và sàn giao dịch	Danh tính công cụ cụ thể nguồn	Cùng một mã ngắn có thể xác định các công cụ khác nhau
Ngày phiên giao dịch	Ngày giao dịch gắn với thanh hàng ngày	Không phải ngày truy xuất UTC
Mở/cao/thấp/đóng	Giá phiên thô	Dương, hữu hạn, nhất quán nội bộ
Khối lượng	Đơn vị báo cáo cho công cụ/phiên giao dịch	Không âm; định nghĩa đơn vị phụ thuộc vào nguồn
Tiền tệ	Siêu dữ liệu công cụ được cung cấp rõ ràng	Không suy diễn chỉ từ một biểu tượng
Cơ sở	raw_daily	Giữ tách biệt với dữ liệu đã điều chỉnh chia/cổ tức
Quan sát tại	Dấu thời gian UTC của trình thu thập	Theo dõi tính mới của việc thu thập

"""Bộ thu thập OHLCV hàng ngày. Thực thi mặc định yêu cầu một tệp đính kèm địa phương rõ ràng."""
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:
            # Never print exc or its URL: the query contains the API key.
            if exc.code not in (429, 502, 503, 504) or attempt == 2:
                raise RuntimeError(f"Source HTTP failure: {exc.code}") from None
            delay = exc.headers.get("Retry-After", "")
            if delay:
                # Schedule a later run; never retry earlier than requested.
                raise RuntimeError("Rate limited; reschedule using Retry-After") from None
        except (URLError, TimeoutError):
            if attempt == 2:
                raise RuntimeError("Source unavailable after three attempts") 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("Cung cấp mã tiền tệ và mã sàn đã được xác thực")
    if payload.get("Meta Data", {}).get("2. Symbol", "").upper() != symbol.upper():
        raise ValueError("Thiếu hoặc không khớp mã nhà cung cấp")
    series = payload.get("Time Series (Daily)")
    if not isinstance(series, dict) or not series:
        raise ValueError("Không có thanh hàng ngày; kiểm tra giới hạn tài khoản hoặc lỗi nguồn")
    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, "mở cửa"), (2, "cao nhất"), (3, "thấp nhất"), (4, "đóng cửa"))]
        if any(not p.is_finite() or p <= 0 for p in prices):
            raise ValueError(f"OHLC không hợp lệ: {session}")
        opening, high, low, close = prices
        if low > min(opening, close) or high < max(opening, close) or low > high:
            raise ValueError(f"Nến không đồng nhất: {session}")
        volume = int(bar["5. volume"])
        if volume < 0:
            raise ValueError("Khối lượng âm")
        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("Chọn chính xác một trong hai --fixture hoặc --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)

Cách tạo và triển khai một công cụ thu thập dữ liệu web chứng khoán: Hướng dẫn thực tiễn

TL;DR

Những Gì Bạn Sẽ Xây Dựng: Một Trình Thu Thập Dữ Liệu Thị Trường Chứng Khoán

Các Yêu Cầu và Thiết Lập Dự Án

Bai viet pho bien

Thu thập Dữ liệu Tài sản: Hướng dẫn về Thu thập Dữ liệu Tài sản Công cộng

Dify SERP Scraping: Cách Xây Dựng Quy Trình Dữ Liệu Tìm Kiếm

Chọn Một Nguồn Dữ Liệu Chứng Khoán Đáng Tin Cậy

APIs Thị Trường Chứng Khoán so với Thu Thập Dữ Liệu Từ Website

Hiểu Các Trường Dữ Liệu Chứng Khoán

Bước 1: Tạo Dự Án Trình Thu Thập Dữ Liệu Python

Bước 2: Lấy Dữ Liệu Chứng Khoán Từ Nguồn

Bước 3: Phân Tích và Xác Thực Giá Cổ Phiếu

Step 4: Save Stock Data to CSV or SQLite

Step 5: Add Logging, Retries, and Error Handling

Step 6: Schedule Automatic Stock Data Updates

Step 7: Deploy the Scraper to Production

Dockerize the Stock Scraper

Configure Environment Variables and Secrets

Set Up Scheduling and Monitoring

Testing the Deployed Stock Scraper

Scaling from One Stock to Hundreds of Tickers

Các Vấn Đề và Giải Pháp Khi Thu Thập Dữ Liệu Cổ Phiếu

Chi Phí và Cân Nhắc Pháp Lý Khi Thu Thập Dữ Liệu Cổ Phiếu

Danh Sách Kiểm Tra Dự Án Cuối Cùng

Xác Thực Các Nguồn Trang Được Phê Duyệt

Câu Hỏi Thường Gặp