Tác vụ nạp dữ liệu

Tác vụ nạp dữ liệu gọi các hàm của vnstock 6 bằng khoá của bạn, kiểm tra chất lượng rồi ghi vào thư mục lưu trữ. Mỗi tác vụ là một lệnh vnstock-pipeline run …, hoặc một phương thức của lớp Runner trong Python.

Cú pháp chung

vnstock-pipeline run <tác vụ> [chọn mã] [tuỳ chọn của tác vụ]

Chọn mã

Nhóm theo sàn và theo chỉ số lấy từ danh mục đã nạp hôm nay. Nếu hôm nay chưa có, pipeline tự nạp trước. Xem một nhóm gồm những mã nào bằng vnstock-pipeline universe show --group VN30.

Kết quả và mã thoát

Mọi lệnh run in JSON gồm phần tóm tắt và kết quả từng đơn vị việc (một mã, một ngày, một ảnh chụp):

{
 "summary": {"units": 30, "actions": {"merged": 28, "up_to_date": 2}, "rows_new": 28,
             "errors": [], "quarantined": [], "warnings": 0},
 "results": [{"key": "FPT", "action": "merged", "rows_new": 1, "source": "<nguồn>", "note": null}]
}

Lệnh thoát với mã 1 khi có error hoặc quarantined, mã 0 khi mọi đơn vị đều ổn. Nhờ vậy lịch chạy và script biết được lần chạy nào cần xem lại.

Nguồn dự phòng

Với mỗi đơn vị việc, pipeline thử lần lượt các nguồn trong [sources] (hoặc --source). Nguồn đầu lỗi tạm thời, như mất mạng hay nguồn bận, thì chuyển sang nguồn kế tiếp. Lỗi do tham số sai thì dừng ngay, không thử tiếp. Cột source trong kết quả và metadata của tệp cho biết nguồn nào đã trả dữ liệu.

Các tác vụ

Nến (run ohlcv)

vnstock-pipeline run ohlcv --group VN30                          # nến ngày, cập nhật phiên gần nhất
vnstock-pipeline run ohlcv --symbols FPT,MWG --mode init         # tải từ đầu cho mã mới theo dõi
vnstock-pipeline run ohlcv --group LIQUIDITY --top 50 --interval 1m
vnstock-pipeline run ohlcv --group INDEX                         # nến chỉ số
vnstock-pipeline run ohlcv --group FUTURES --interval 1D         # hợp đồng tương lai liên tục

Cách tải nối tiếp:

• Pipeline đọc lại overlap_days nến cuối, mặc định 10.
• Giá đóng cửa trong khoảng đọc lại khác giá đã lưu quá 0,5% nghĩa là nguồn đã điều chỉnh lịch sử, ví dụ sau chia cổ tức hay tách cổ phiếu. Khi đó pipeline tải lại toàn bộ lịch sử của mã và thay tệp cũ.
• Nến phút của hôm nay được giữ, vì mỗi nến phút đóng dần trong ngày.

Giá theo đơn vị của vnstock 6: cổ phiếu tính bằng nghìn đồng, chỉ số và phái sinh tính bằng điểm. Lịch sử là giá đã điều chỉnh do nguồn trả về.

Khớp lệnh trong ngày (run trades)

vnstock-pipeline run trades --symbols FPT,HPG                    # phiên gần nhất
vnstock-pipeline run trades --group VN30 --dates 2026-10-30,2026-10-31

Lấy toàn bộ khớp lệnh của từng mã cho từng ngày, rồi ghi lại cả tệp market/trades/<ngày>/<MÃ>.parquet. Ngày nào lấy được tuỳ nguồn lưu lại bao lâu.

Bảng giá và sổ lệnh (run quote)

vnstock-pipeline run quote --group VN30                          # bảng giá và sổ lệnh
vnstock-pipeline run quote --symbols FPT --kinds order_book

Mỗi lần chạy là một ảnh chụp, được nối vào tệp của ngày theo mốc giờ. Chạy lặp theo lịch, ví dụ mỗi phút, sẽ có bản ghi trong ngày mà không cần mở phiên stream.

Chỉ số trong phiên (run index)

vnstock-pipeline run index                                        # các chỉ số trong [reference] indexes
vnstock-pipeline run index --symbols VNINDEX,VN30,HNXINDEX

Khối ngoại, thống kê phiên, thoả thuận

vnstock-pipeline run foreign --group VN30                        # khối ngoại theo ngày, tải nối tiếp
vnstock-pipeline run session-stats --group VN30                  # thống kê phiên theo ngày
vnstock-pipeline run block-trades                                # giao dịch thoả thuận cả thị trường

foreign và session-stats là chuỗi theo ngày của từng mã, nhận --start, --end, --mode như run ohlcv. Hai tác vụ này không có bước kiểm điều chỉnh lịch sử.

Tham chiếu (run reference)

vnstock-pipeline run reference
vnstock-pipeline run reference --parts equity_list,index_members

Nên chạy run reference mỗi sáng trước các tác vụ khác, vì nhóm mã theo sàn, theo chỉ số và lịch giao dịch đều đọc từ đây.

Hồ sơ doanh nghiệp (run company)

vnstock-pipeline run company --symbols FPT,MWG                   # thông tin chung
vnstock-pipeline run company --group VN30 --parts all

--parts nhận info, shareholders, ownership, insider_trading, hoặc all. Tác vụ chạy trên nguồn có hồ sơ doanh nghiệp, mặc định fhsc.

Nhóm thanh khoản

vnstock-pipeline universe build-liquidity --days 20 --top 300    # xếp hạng trên nến ngày đã lưu
vnstock-pipeline universe show --group LIQUIDITY --top 50
vnstock-pipeline run ohlcv --group LIQUIDITY --top 50 --interval 1m

build-liquidity xếp các mã theo giá trị giao dịch trung bình của --days phiên gần nhất, tính trên nến ngày đã lưu, rồi lưu danh sách --top mã đầu. Nhóm LIQUIDITY đọc danh sách này. Nếu chưa có danh sách, pipeline tự dựng. Nên chạy lại sau khi cập nhật nến ngày.

Đặt lịch chạy

Lệnh run không tự lặp. Bạn đặt lịch bằng công cụ của hệ điều hành.

macOS, Linux (crontab -e):

# 08:30 thứ Hai tới thứ Sáu: danh mục, thành phần, trần sàn, ngày giao dịch
30 8 * * 1-5  ~/.venv_pipeline6/bin/vnstock-pipeline run reference >> ~/pipeline.log 2>&1
# 15:45: nến ngày, khối ngoại, thống kê phiên cho VN30
45 15 * * 1-5 ~/.venv_pipeline6/bin/vnstock-pipeline run ohlcv --group VN30 >> ~/pipeline.log 2>&1
50 15 * * 1-5 ~/.venv_pipeline6/bin/vnstock-pipeline run foreign --group VN30 >> ~/pipeline.log 2>&1

Windows: tạo tác vụ trong Task Scheduler, chương trình là vnstock-pipeline.exe trong thư mục Scripts của môi trường ảo, tham số là phần sau tên lệnh, ví dụ run ohlcv --group VN30.

Pipeline tự bỏ qua mã đã mới, nên chạy lại một lệnh nhiều lần trong ngày không tải trùng. Nguồn tcbs cần mở phiên bằng OTP mỗi 24 giờ; lịch chạy tự động nên ưu tiên nguồn không cần OTP trong [sources].

Gọi từ Python

from vnstock_pipeline import Runner
from vnstock_pipeline.tasks import summary

run = Runner()                                   # Store() + vnstock 6; Runner(store, strict=True) nếu muốn chặt
results = run.reference()
results += run.ohlcv(group="VN30", interval="1D")
results += run.trades(symbols=["FPT", "HPG"], dates=["2026-10-30"])
results += run.snapshot(symbols=["FPT"], kinds=("quote", "order_book"))
results += run.foreign_flow(watchlist="ngan_hang")

print(summary(results))                          # như phần "summary" của dòng lệnh
for r in results:
    if r.action == "error":
        print(r.key, r.source, r.error)

Mỗi phương thức trả danh sách TaskResult (task, key, source, action, rows, rows_new, path, issues, error) và không in gì ra màn hình.

Bước tiếp theo

Dữ liệu trong phiên.