Cấu hình streaming mẫu

Sau khi đã tìm hiểu các thành phần nền tảng trong hướng dẫn Khung streaming dữ liệu, bạn có thể lựa chọn phương thức thiết lập nhanh và tự động hóa cao hơn: cấu hình streaming mẫu khai báo bằng định dạng TOML. Thay vì phải tự viết mã Python tùy biến lớp phân tích cú pháp (BaseDataParser) hay tự xử lý handshake/heartbeat thủ công, bạn chỉ cần mô tả cấu hình nguồn WebSocket trong một tệp TOML duy nhất. Thư viện sẽ tự động quản lý toàn bộ quy trình: từ kết nối mạng, gửi thông điệp đăng ký kênh, duy trì heartbeat, tự động reconnect khi gián đoạn, cho tới việc bóc tách từng bản ghi thành các sự kiện dữ liệu chuẩn hóa.

Ba kiểu giao thức đóng khung (Framing)

Khi nhận dữ liệu trong phiên qua WebSocket, mỗi máy chủ nguồn có thể đóng gói thông điệp theo một chuẩn khác nhau. vnstock_pipeline hỗ trợ 3 kiểu đóng khung thông điệp (framing) phổ biến:

Cách nhận biết giao thức của nguồn

Để xác định nguồn dữ liệu của bạn đang sử dụng giao thức nào, hãy kiểm tra gói tin đầu tiên máy chủ trả về ngay sau khi thiết lập kết nối:

• Engine.IO: Gói tin đầu tiên có dạng chuỗi 0{...}. Với giao thức này, thư viện sẽ tự động duy trì heartbeat bằng cách phản hồi mã 3 (pong) mỗi khi nhận được mã 2 (ping) từ máy chủ.
• Socket.IO: Máy chủ phản hồi mã 40 sau khi client gửi gói tin 40.
• JSON thuần: Máy chủ gửi trực tiếp đối tượng JSON hợp lệ (ví dụ: {"stream": ..., "data": ...}).

Cấu trúc cú pháp của tệp cấu hình TOML

Tệp cấu hình TOML được tổ chức thành 3 khối thiết lập chính: nguồn kết nối ([source]), thông điệp đăng ký ([[subscribe]]) và quy tắc định tuyến dữ liệu ([[routes]]).

1. Khối nguồn kết nối [source]

Thiết lập thông số kết nối cơ bản tới máy chủ WebSocket:

2. Khối đăng ký luồng dữ liệu [[subscribe]]

Mô tả cấu trúc thông điệp gửi lên máy chủ để đăng ký nhận luồng dữ liệu cho danh sách mã. Bạn có thể khai báo nhiều khối [[subscribe]] cho các loại dữ liệu khác nhau (như trades, nến kline, sổ lệnh):

3. Khối định tuyến và bóc tách dữ liệu [[routes]]

Xác định cách nhận diện thông điệp nhận về và chuyển đổi các trường dữ liệu thô thành sự kiện chuẩn hóa. Một cấu hình có thể chứa nhiều khối [[routes]]:

Cơ chế định tuyến: Mỗi thông điệp nhận về từ WebSocket sẽ được kiểm tra với toàn bộ các route. Nếu một thông điệp khớp với nhiều route, tất cả các route đó đều được xử lý độc lập. Các thông điệp không khớp với bất kỳ route nào sẽ được ghi nhận vào bộ đếm thống kê và có thể ghi ra tệp riêng khi bạn thiết lập tham số capture_path.

Ví dụ thực tế: Tệp cấu hình Binance Spot

Dưới đây là tệp cấu hình mẫu hoàn chỉnh binance_spot.toml để tiếp nhận dữ liệu khớp lệnh (trades) và nến 1 phút (klines) từ luồng WebSocket công khai của Binance Spot:

[source]
name = "binance_spot"
uri = "wss://stream.binance.com:9443/stream"
framing = "json"

# Mỗi mã được định dạng thành "<mã>@trade" và "<mã>@kline_1m"; gộp tối đa 50 stream mỗi gói tin
[[subscribe]]
name = "trades"
template = '{{"method":"SUBSCRIBE","params":{symbols},"id":{seq}}}'
unsubscribe_template = '{{"method":"UNSUBSCRIBE","params":{symbols},"id":{seq}}}'
item = "{symbol}@trade"
case = "lower"
chunk_size = 50

[[subscribe]]
name = "klines"
template = '{{"method":"SUBSCRIBE","params":{symbols},"id":{seq}}}'
unsubscribe_template = '{{"method":"UNSUBSCRIBE","params":{symbols},"id":{seq}}}'
item = "{symbol}@kline_1m"
case = "lower"
chunk_size = 50

[[routes]]
event_type = "trade"
match = { path = "data.e", equals = "trade" }
records = "data"
symbol = "s"
fields = { trade_id = "t", price = "p", quantity = "q", buyer_is_maker = "m" }
numbers = ["price", "quantity"]
time = { field = "T", unit = "ms" }

[[routes]]
event_type = "kline"
match = { path = "data.e", equals = "kline" }
records = "data"
symbol = "s"
fields = { interval = "k.i", open_time = "k.t", open = "k.o", high = "k.h", low = "k.l", close = "k.c", volume = "k.v", quote_volume = "k.q", trades = "k.n", is_closed = "k.x" }
numbers = ["open", "high", "low", "close", "volume", "quote_volume"]
time = { field = "E", unit = "ms" }

Giải thích chi tiết các khối cấu hình:

• Khối [source]: Binance gửi thông điệp dạng JSON thuần, do đó ta khai báo framing = "json".
• Khối [[subscribe]]:
  • template: Định dạng JSON yêu cầu của Binance có dạng {"method": "SUBSCRIBE", "params": [...], "id": 1}. Dấu ngoặc nhọn JSON được nhân đôi {{ }} để phân biệt với {symbols} và {seq}.
  • item = "{symbol}@trade" và case = "lower": Tự động chuyển đổi mã tài sản bạn truyền vào (ví dụ BTCUSDT) thành định dạng stream viết thường của Binance (btcusdt@trade).
  • chunk_size = 50: Gộp tối đa 50 kênh trong một thông điệp gửi đi, giúp giảm thiểu số lượng gói tin đăng ký.
• Khối [[routes]]:
  • match = { path = "data.e", equals = "trade" }: Kiểm tra trường e (event type) trong đối tượng data; nếu có giá trị "trade" thì xử lý theo route này.
  • fields: Ánh xạ các trường viết tắt của Binance (t, p, q, m) sang tên cột rõ nghĩa (trade_id, price, quantity, buyer_is_maker). Đối với nến kline, ta dùng cú pháp dấu chấm như open = "k.o" để lấy trường o nằm trong đối tượng k.
  • numbers: Binance trả về giá trị số học (như giá và khối lượng) dưới dạng chuỗi string trong JSON nhằm tránh hiện tượng làm tròn số của JavaScript; khai báo numbers giúp thư viện tự động chuyển đổi các trường này thành số thực để sẵn sàng tính toán.

Khởi chạy luồng stream từ tệp cấu hình

Thư viện cung cấp hai phương thức vận hành: chạy nhanh bằng hàm tiện ích hoặc quản lý linh hoạt qua lớp client. Cả hai đều tích hợp trực tiếp với bộ ghi dữ liệu CSVProcessor (được giới thiệu trong hướng dẫn Khung streaming dữ liệu) hoặc các bộ xử lý tùy biến kế thừa từ BaseProcessor.

Cách 1: Khởi chạy nhanh với run_profile

Thích hợp cho các tiến trình tiếp nhận dữ liệu chạy nền với danh sách mã cố định:

import asyncio
from pathlib import Path
from vnstock_pipeline.stream import CSVProcessor
from vnstock_pipeline.stream.profiles import load_profile, run_profile

# Thư mục lưu trữ tệp đầu ra trên máy bạn
OUT = Path.home() / "vnstock_stream"
OUT.mkdir(exist_ok=True)

# Nạp tệp cấu hình TOML
profile = load_profile("binance_spot.toml")

# Chạy luồng tiếp nhận dữ liệu
asyncio.run(run_profile(
    profile,
    ["BTCUSDT", "ETHUSDT"],
    [CSVProcessor(str(OUT / "binance_{event_type}_%Y-%m-%d.csv"))],
    rotate_seconds=23 * 3600,   # Binance chủ động ngắt mỗi kết nối sau 24 giờ; xoay vòng sau 23 giờ để duy trì ổn định
))

Mỗi bản ghi sự kiện sau khi qua xử lý sẽ bao gồm các trường chuẩn: event_type, symbol, ts_ms (mốc thời gian từ nguồn, mili-giây), received_ms (mốc thời gian máy bạn nhận được gói tin) cùng các trường dữ liệu nghiệp vụ đã ánh xạ trong fields.

Cách 2: Điều khiển nâng cao với ProfileClient

Khi cần thay đổi danh sách mã động trong phiên giao dịch (thêm hoặc bớt mã theo dõi) mà không muốn làm ngắt quãng toàn bộ kết nối, bạn sử dụng ProfileClient:

import asyncio
from pathlib import Path
from vnstock_pipeline.stream import CSVProcessor
from vnstock_pipeline.stream.profiles import load_profile, ProfileClient

OUT = Path.home() / "vnstock_stream"
OUT.mkdir(exist_ok=True)

profile = load_profile("binance_spot.toml")

async def main():
    # Khởi tạo client với danh sách theo dõi ban đầu
    client = ProfileClient(profile, ["BTCUSDT"])
    client.add_processor(CSVProcessor(str(OUT / "binance_{event_type}_%Y-%m-%d.csv")))

    # Kết nối tới máy chủ nguồn
    await client.connect()

    # Đăng ký thêm mã mới ngay trong phiên
    await client.subscribe(["SOLUSDT"])

    # Hủy đăng ký mã không còn theo dõi
    await client.unsubscribe(["BTCUSDT"])

    # Giữ phiên chạy cho đến khi mất kết nối
    await client.wait_until_disconnected()

# asyncio.run(main())

Phân tích dữ liệu tick ghi nhận được với vnstock_quant

Sau khi lưu trữ luồng dữ liệu khớp lệnh (tick) ra tệp CSV, bạn có thể kết hợp với mô-đun micro của thư viện vnstock_quant để dựng nến và phân tích cấu trúc vi mô thị trường:

from datetime import date
from pathlib import Path
import pandas as pd
from vnstock_quant import micro

OUT = Path.home() / "vnstock_stream"

# Đọc dữ liệu tick ghi nhận trong ngày
ticks = pd.read_csv(OUT / f"binance_trade_{date.today():%Y-%m-%d}.csv")
ticks["time"] = pd.to_datetime(ticks["ts_ms"], unit="ms", utc=True)
btc = ticks[ticks["symbol"] == "BTCUSDT"]

# Dựng nến 1 phút theo thời gian truyền thống (time bars)
m1 = micro.bars(btc, by="time", size="1min")

# Dựng nến theo ngưỡng giá trị giao dịch 500.000 USDT (value bars)
vb = micro.bars(btc, by="value", size=500_000)

Ghi chú về phân loại giao dịch chủ động: Trong dữ liệu của Binance, cờ buyer_is_maker = True có nghĩa là bên mua đóng vai trò cung cấp thanh khoản (maker trong sổ lệnh), do đó giao dịch phát sinh từ phía người bán chủ động (taker). Hàm micro.classify của vnstock_quant nhận diện trực tiếp cột này để phân loại dòng tiền mua/bán chủ động.

Quy trình 4 bước tự tạo tệp cấu hình cho nguồn mới

Khi cần kết nối tới một nguồn dữ liệu WebSocket mới, bạn có thể thực hiện theo quy trình 4 bước chuẩn sau (kết hợp với cơ chế handshake và quản lý phiên đã giới thiệu trong hướng dẫn Khung streaming dữ liệu):

1. Nghiên cứu tài liệu và điều khoản nguồn: Tìm hiểu cấu trúc gói tin, quy trình handshake và kiểm tra kỹ điều khoản sử dụng của nhà cung cấp trước khi tích hợp.
2. Bảo mật thông tin xác thực qua biến môi trường: Tuyệt đối không ghi cứng (hardcode) các thông tin nhạy cảm (như token, mã phiên hoặc định danh thiết bị) vào tệp TOML. Hãy lưu chúng vào biến môi trường trên máy bạn và tham chiếu trong tệp cấu hình bằng cú pháp ${TEN_BIEN} tại trường uri.
3. Ghi nhận gói tin mẫu với capture_path: Tạo một tệp cấu hình cơ bản chứa khối [source], [[subscribe]] và một route chưa thiết lập điều kiện lọc, sau đó khởi chạy với tham số capture_path để ghi lại các gói tin thô thực tế:
   Python

   client = ProfileClient(
       load_profile("nguon_cua_ban.toml"),
       ["AAA"],
       capture_path="du_lieu_mau.jsonl",
       capture_limit=500,
   )

4. Phân tích dữ liệu mẫu và hoàn thiện các khối [[routes]]: Mở tệp .jsonl vừa ghi nhận, quan sát cấu trúc thực tế của các thông điệp để xác định chính xác đường dẫn trường (records, symbol, fields) và quy tắc lọc (match).

Một số lưu ý kỹ thuật và giới hạn vận hành

• Giới hạn số lượng stream và tần suất gửi tin: Mỗi nhà cung cấp đều có quy định riêng về giới hạn kết nối (ví dụ Binance cho phép tối đa 1024 stream trên một kết nối và 5 thông điệp gửi lên mỗi giây). Thư viện vnstock_pipeline đã tích hợp sẵn cơ chế giãn cách tự động ít nhất 0,2 giây giữa các gói tin đăng ký để tránh bị ngắt kết nối đột ngột.
• Xử lý lỗi HTTP khi handshake: Nếu quá trình handshake WebSocket bị từ chối với các mã lỗi HTTP 401 (Unauthorized), 403 (Forbidden) hoặc 451 (Unavailable For Legal Reasons), run_profile sẽ chủ động dừng hoàn toàn tiến trình và không thử lại nhằm tránh việc gửi yêu cầu dồn dập khiến IP của bạn bị đưa vào danh sách chặn. Mã HTTP 429 sẽ được ghi nhận là trạng thái chạm trần tần suất (rate limited).
• Nguyên tắc an toàn cho tệp cấu hình: Tệp cấu hình TOML được thiết kế có chủ đích không cho phép cấu hình trực tiếp các header tùy biến, cookie, Origin hay Referer. Bất kỳ thông số phiên cần thiết nào đều phải được truyền qua chuỗi truy vấn (query string) trong uri thông qua biến môi trường ${TEN_BIEN}.
• Mục đích sử dụng: Toàn bộ dữ liệu tiếp nhận được lưu trữ trực tiếp trên máy của bạn nhằm phục vụ cho mục đích nghiên cứu và phân tích định lượng cá nhân. Hãy lưu ý các quy định của nhà cung cấp về việc chia sẻ hoặc phân phối lại dữ liệu.