Chất lượng dữ liệu

Dữ liệu từ nguồn có lúc thiếu phiên, trùng dòng, sai đơn vị hay đổi cấu trúc cột mà không báo trước. Pipeline kiểm từng khung dữ liệu trước khi ghi, để lỗi của một lần tải không làm hỏng dữ liệu bạn đã có.

Ba lớp kiểm

Lớp	Khi nào	Làm gì
Kiểm nội dung	Mỗi tác vụ `run …`, trước khi ghi	Giá, khối lượng, mốc giờ, độ tươi, khoảng hở theo lịch giao dịch. Lỗi thì không ghi; cảnh báo thì vẫn ghi, trừ khi bật `--strict`
Kiểm cấu trúc	Sau khi kiểm nội dung	So tên cột và kiểu cột với cấu trúc chuẩn của bộ dữ liệu. Lệch thì khung mới vào vùng cách ly, tệp cũ giữ nguyên
Kiểm định kỳ	Khi bạn chạy `store audit`	Kiểm lại nến đã lưu: độ tươi, khoảng hở, giá bất thường

Dữ liệu trong phiên đi theo hợp đồng sự kiện chuẩn có kiểu cột cố định (Sự kiện chuẩn), nên không qua lớp kiểm cấu trúc. Chất lượng của phiên được báo trong tóm tắt phiên: lỗi giải mã, lệnh trùng, lệnh tới muộn, khối lượng hụt.

Mã kiểm nội dung

Mã	Mức	Nghĩa
`PIPELINE_QUALITY_EMPTY`	cảnh báo	Nguồn trả khung rỗng
`PIPELINE_QUALITY_COLUMNS`	lỗi / cảnh báo	Thiếu cột thời gian (lỗi), hoặc thiếu cột khoá khác (cảnh báo; gộp theo cột còn lại)
`PIPELINE_QUALITY_TIME_NULL`	lỗi	Có dòng không có mốc giờ
`PIPELINE_QUALITY_UNSORTED`	cảnh báo	Mốc giờ không tăng dần; pipeline sắp xếp lại khi ghi
`PIPELINE_QUALITY_DUPLICATE`	cảnh báo	Dòng trùng khoá; giữ dòng sau
`PIPELINE_QUALITY_OHLC`	lỗi / cảnh báo	Cao, thấp không bao giá mở, đóng. Vài dòng lẻ trong lịch sử dài là cảnh báo; quá 1% số dòng là lỗi
`PIPELINE_QUALITY_PRICE_ZERO`	lỗi	Giá bằng 0 hoặc âm
`PIPELINE_QUALITY_PRICE_UNIT`	lỗi	Giá cổ phiếu lớn bất thường: có vẻ là đồng trong khi đơn vị chuẩn là nghìn đồng
`PIPELINE_QUALITY_VOLUME`	lỗi	Khối lượng âm
`PIPELINE_QUALITY_VOLUME_ZERO`	cảnh báo	Hơn nửa số nến ngày có khối lượng 0
`PIPELINE_QUALITY_STALE`	cảnh báo (lỗi khi `--strict`)	Nến ngày cuối chậm hơn 5 phiên so với ngày giao dịch gần nhất
`PIPELINE_QUALITY_GAP`	cảnh báo	Thiếu phiên giữa ngày đầu và ngày cuối, theo lịch giao dịch

Kiểm độ tươi và khoảng hở dựa vào lịch giao dịch. Lịch lấy từ ngày giao dịch đã nạp bằng run reference và ngày nghỉ trong [calendar] holidays. Khi khoảng ngày chưa có lịch, pipeline không kết luận thiếu phiên, vì ngày thường có thể là ngày nghỉ lễ.

Mã và nội dung cảnh báo nằm trong kết quả của lệnh run (--verbose để xem từng mã) và trong TaskResult.issues khi gọi từ Python.

Cấu trúc chuẩn và vùng cách ly

Lần đầu một bộ dữ liệu được ghi, tên cột và kiểu cột của nó thành cấu trúc chuẩn (_schema/<bộ dữ liệu>.json). Từ đó, mỗi khung mới được so với cấu trúc chuẩn:

thêm cột, bớt cột, hoặc đổi kiểu không tương thích, ví dụ số thành chữ: khung vào _quarantine/<bộ dữ liệu>/<giờ>_<khoá>.parquet, kết quả của đơn vị việc là quarantined, lệnh thoát với mã 1;
đổi kiểu tương thích, ví dụ số nguyên sang số thực: vẫn ghi.

Shell

vnstock-pipeline quarantine list                 # mọi tệp đang cách ly, kèm cột thêm, bớt, đổi
vnstock-pipeline quarantine list ohlcv
vnstock-pipeline quarantine adopt <đường dẫn>     # nhận cấu trúc mới làm chuẩn, rồi xoá tệp cách ly
vnstock-pipeline quarantine discard <đường dẫn>   # bỏ tệp cách ly
vnstock-pipeline schema show ohlcv               # xem cấu trúc chuẩn
vnstock-pipeline schema reset ohlcv              # xoá cấu trúc chuẩn; lần ghi sau đặt lại
vnstock-pipeline schema file <tệp.parquet>       # cấu trúc cột của một tệp bất kỳ

Khi nguồn đổi cấu trúc thật, ví dụ thêm một cột hợp lệ:

Xem khác biệt bằng quarantine list.
Nhận cấu trúc mới bằng quarantine adopt.
Chạy lại tác vụ để ghi dữ liệu.

adopt chỉ đổi cấu trúc chuẩn, không tự ghi khung bị cách ly vào thư mục lưu trữ.

Tắt kiểm cấu trúc bằng schema_guard = false trong [store].

Kiểm định kỳ: store audit

Shell

vnstock-pipeline store audit                          # nến ngày của mọi mã đã lưu
vnstock-pipeline store audit --interval 1m --stale-days 2

Lệnh đọc lại nến đã lưu của từng mã, chạy các kiểm nội dung ở trên, rồi in danh sách vấn đề theo mã dạng JSON. Nên chạy sau các tác vụ cuối ngày, hoặc trước khi xuất dữ liệu sang công cụ khác.

Dùng trong Python

Python

from vnstock_pipeline.quality import DataGuard, SchemaGuard, Quarantine, TradingCalendar
from vnstock_pipeline import Store

store = Store()
guard = DataGuard(TradingCalendar(holidays=["2026-09-02"]), strict=False, stale_days=5)
report = guard.validate(df, dataset_name="ohlcv", key="FPT", interval="1D")
report.ok, [i.to_dict() for i in report.issues]

ok, diff = SchemaGuard(store).check("ohlcv", "FPT", df)   # False và một tệp cách ly khi lệch
Quarantine(store).list("ohlcv")

Bước tiếp theo

Tra cứu và xuất dữ liệu.