Dữ liệu trong phiên

Trong giờ giao dịch, pipeline nhận thông điệp liên tục từ nguồn, giải mã về sự kiện chuẩn, dựng nến từ khớp lệnh và ghi tệp Parquet theo nguồn, theo ngày. Một lệnh làm đủ các bước đó:

vnstock-pipeline stream run <nguồn> <kênh...> --symbols <mã> --seconds 600 --intervals 1m,5m

Luồng xử lý

1. Kết nối. Phần kết nối vnstock-oapi mở kết nối tới cổng stream chính thức của nguồn bằng khoá của bạn, rồi đăng ký các kênh bạn chọn. Kênh chưa khai sẵn, kênh lệnh, vị thế, tài khoản đều bị từ chối trước khi gửi đăng ký. Mất kết nối thì phần kết nối tự nối lại với thời gian chờ tăng dần và đánh dấu khoảng bị mất.
2. Bản thô (tuỳ chọn). Khi bật keep_raw hoặc --keep-raw, mỗi thông điệp được ghi nguyên văn vào raw/<nguồn>/<ngày>/stream/.
3. Giải mã. Thông điệp được giải mã về sự kiện chuẩn: lệnh khớp, bảng giá, chỉ số, khối ngoại… (Sự kiện chuẩn và bộ cột). Phần giải mã chạy trong lõi Rust.
4. Dựng nến. Lệnh khớp ở bảng chính được gom thành nến theo các khung bạn chọn. Nến của chính nguồn (nếu có kênh nến) được ghi riêng, giữ nhãn nguồn phát.
5. Ghi tệp. Cứ vài giây hoặc vài nghìn sự kiện, pipeline ghi một tệp phần cho mỗi loại sự kiện. Khi phiên kết thúc, các tệp phần trong ngày được gộp thành một tệp.

Phiên dừng khi hết --seconds, khi bạn bấm Ctrl+C, hoặc khi phần kết nối hết lượt nối lại. Phiên nào dừng cũng đóng nốt nến đang mở, ghi phần còn lại và gộp tệp.

Nguồn hỗ trợ

Liệt kê kênh của một nguồn ngay trên máy, kèm mẫu tham số:

vnstock-pipeline stream channels <nguồn>

Kênh của dnse

{board}: G1 lô chẵn, G4 lô lẻ, T1 thoả thuận. {resolution}: 1, 3, 5, 15, 30, 1H, 1D, 1W. Tên chỉ số theo nguồn: VN30, VNINDEX, HNX, UPCOM (pipeline lưu thành HNXINDEX, UPCOMINDEX). Hợp đồng tương lai dùng mã KRX, ví dụ 41I1GB000.

Kênh của tcbs

Kênh match_history chỉ nhận một mã mỗi phiên. Đăng ký nhiều mã thì phần kết nối báo lỗi ngay.

Kênh của ssi

Kênh index nhận VN30, VNINDEX, HNXINDEX, HNX30, UPCOMINDEX…, không phân biệt hoa thường. Đăng ký ALL để nhận mọi mã của một kênh.

Kênh của binance

{interval} của nến: 1s, 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d, 3d, 1w, 1M.

Lệnh stream

Kết thúc, stream run in tóm tắt phiên dạng JSON:

{
 "source": "<nguồn>", "date": "2026-11-04", "engine": "rust",
 "started_at": "2026-11-04T10:08:43+07:00", "ended_at": "2026-11-04T10:10:13+07:00",
 "frames": 19789,
 "control": {"connected": 1},
 "events": {"trade": 954, "quote": 8559, "stats": 586, "bar": 950, "index_tick": 29},
 "bars_closed": {"1m": 9},
 "decode_errors": {},
 "parts": 102,
 "quality": {"1m": {"trades": 864, "duplicates": 0, "off_board": 0, "late": 0, "replayed": 68, "gaps": []}}
}

Nến dựng từ khớp lệnh

• Chỉ bảng chính. Nến chỉ gom lệnh lô chẵn, như nến của sàn và của nguồn. Lô lẻ và thoả thuận vẫn được lưu ở stream_trades, nhưng không vào nến.
• Không đếm hai lần.
  • Khi hai kênh cùng mang một lệnh, ví dụ tick và tick_extra của dnse, pipeline dùng tổng khối lượng cộng dồn của nguồn để bỏ bản trùng.
  • Mỗi nguồn có một nhóm kênh dựng nến: dnse dùng tick/tick_extra, tcbs dùng history_1m/derivative_match (không có history_1m thì match), ssi dùng trade, binance dùng trade (không có thì agg_trade).
• Lệnh gửi lại không vào nến. Lệnh nguồn gửi lại khi vừa đăng ký mang replay = true và được bỏ qua: lịch sử khớp của tcbs, bản ghi cuối của ssi, thông điệp khớp của tcbs lặp lại mà tổng cộng dồn không tăng, lệnh ssi có tổng cộng dồn đã thấy, và nói chung lệnh cũ hơn 60 giây lúc nhận.
• Lệnh đến lệch thứ tự. Lệnh ssi trong cùng một giây có thể đến không theo thứ tự khớp. Lệnh có tổng nhỏ hơn bản trước vẫn được tính; giá mở và giá đóng của nến theo thứ tự tổng cộng dồn, không theo thứ tự đến.
• Khối lượng theo tổng cộng dồn ở nguồn dạng ảnh chụp. Với kênh match và derivative_match của tcbs, một thông điệp khớp là ảnh chụp của mã và có thể gộp nhiều lệnh. Vì vậy khối lượng nến lấy bằng phần tăng của tổng cộng dồn, không cộng khối lượng từng thông điệp. Nhờ vậy nến không hụt khi nguồn gộp lệnh.
• Giá trị nến tính bằng đồng.
  • Cổ phiếu: giá × 1.000 × khối lượng.
  • Hợp đồng tương lai chỉ số VN30, VN100: 100.000 đồng mỗi điểm.
  • Phái sinh khác chưa có hệ số, nên giá trị để trống, không ghi số đoán.
  • Với nguồn dạng ảnh chụp, giá trị lấy theo phần tăng của tổng giá trị cộng dồn khi tổng đó khớp với giá.
• Mỗi mã một phiên cho mỗi nguồn. Hai phiên cùng nguồn cùng có một mã, ví dụ phiên bảng giá và phiên match_history của tcbs cùng có FPT, ghi vào một thư mục gốc sẽ cho hai bộ nến của mã đó.
• Giờ.
  • Nến cổ phiếu, chỉ số, phái sinh chia theo giờ Việt Nam; nến ngày bắt đầu 00:00 giờ Việt Nam.
  • Nến đóng 2 giây sau khi hết khung (close_delay), vì ssi ghi giờ tới giây và thông điệp đến chậm tới khoảng một giây. Phát lại tệp dùng cùng độ trễ, nên cho cùng nến như phiên trực tiếp.
  • Nến binance chia theo UTC như nến của chính Binance. Giá và giá trị tính theo tài sản định giá (ví dụ USDT), khối lượng theo tài sản gốc.

Kiểm thử nội bộ ngày 04/11/2026 trong giờ giao dịch so ba cặp nến, kết quả như sau:

Đây là quan sát tại thời điểm kiểm, không phải cam kết về dữ liệu của nguồn.

Ghi lại và chạy lại

Khi cần đối chiếu hay thử cách xử lý khác mà không mở lại phiên, hãy ghi thông điệp ra tệp rồi chạy lại:

vnstock-pipeline stream record <nguồn> <kênh...> --symbols <mã> --seconds 600 -o ~/phien/<nguồn>-0930.jsonl
vnstock-pipeline stream replay ~/phien/<nguồn>-0930.jsonl --intervals 1m,5m

stream replay không nối tới nguồn, nhưng cần giấy phép như stream run, vì giải mã và dựng nến là phần được cấp phép. Tệp đi qua đúng luồng của phiên trực tiếp: giải mã, dựng nến với cùng quy tắc của nguồn, ghi vào ngày của thông điệp đầu tiên. Tệp ghi bằng --keep-raw cũng chạy lại được.

Dùng trong Python

from vnstock_pipeline import Session, Store

def on_bar(bars):                      # nến 1m, 5m vừa cập nhật hoặc vừa đóng
    for bar in bars:
        if bar["final"]:
            print(bar["symbol"], bar["interval"], bar["close"], bar["volume"])

session = Session(
    Store(), "<nguồn>",
    ["<kênh 1>", "<kênh 2>"],            # tên kênh: vnstock-pipeline stream channels <nguồn>
    ["<mã>"],
    intervals=("1m", "5m"),
    on_bar=on_bar,                     # on_event=... nhận mọi sự kiện đã giải mã
)
stats = session.run(seconds=600)       # chặn tới khi hết giờ hoặc Ctrl+C
print(stats.to_dict())

Trong mã bất đồng bộ, dùng await session.arun(seconds=...) và session.stop().

Chạy lại tệp đã ghi, hoặc chỉ giải mã, dựng nến mà không ghi vào thư mục lưu trữ:

from vnstock_pipeline import replay_events, replay_bars
from vnstock_pipeline.stream import replay_into_store

events, errors = replay_events("ssi-0930.jsonl")          # danh sách sự kiện chuẩn, lỗi kèm số dòng
bars, quality = replay_bars("ssi-0930.jsonl", "1m")        # nến đã đóng và báo cáo của bộ dựng nến
replay_into_store(Store(), "ssi-0930.jsonl", intervals=("1m",))

Chạy nhiều nguồn cùng lúc

Mỗi nguồn chạy một tiến trình, cùng ghi vào một thư mục gốc:

vnstock-pipeline stream run <nguồn 1> <kênh...> --symbols FPT,HPG &
vnstock-pipeline stream run <nguồn 2> <kênh...> --symbols FPT,HPG,VN30 &
vnstock-pipeline stream run <nguồn 3> <kênh...> --symbols BTCUSDT,ETHUSDT &

Tệp của mỗi nguồn nằm ở thư mục source=<nguồn> riêng, còn mục lục được cập nhật dưới khoá chung. Đọc gộp nhiều nguồn bằng Store().read("stream_trades", date="2026-11-04"), rồi lọc theo cột source.

Lưu ý khi chạy

• Mỗi nguồn có giới hạn số kết nối và số mã mỗi kết nối riêng.
  • Với binance, khi số luồng vượt giới hạn của một kết nối, phần kết nối tự mở thêm kết nối. Phần kết nối cũng tự nối lại trước mốc nguồn đóng kết nối định kỳ.
  • Với các nguồn khác, đăng ký vượt giới hạn thì nguồn từ chối và phiên báo lỗi.
• Phiên của nguồn mở bằng OTP hết hạn thì phần kết nối báo OtpRequiredError; mở lại bằng vnstock-oapi auth <nguồn>.
• Phiên chạy qua nửa đêm, như binance chạy cả đêm, tự sang thư mục của ngày mới và gộp tệp của ngày cũ.
• Dữ liệu trong phiên dùng cho nghiên cứu của bạn. Lưu trữ, hiển thị hay chia sẻ dữ liệu theo điều khoản của từng nguồn (vnstock-oapi terms <nguồn>).

Bước tiếp theo

Sự kiện chuẩn và bộ cột.