Xây Dựng Hệ Thống MLOps End-to-End Dự Đoán Giá Cổ Phiếu Ngành Bảo Hiểm

Bài viết hướng dẫn xây dựng hệ thống MLOps End-to-End dự đoán giá đóng cửa sau 1 tuần của 5 mã cổ phiếu bảo hiểm sàn HOSE (BVH, BMI, MIG, BIC, PGI) từ 10 năm dữ liệu lịch sử qua API vnstock. Toàn bộ pipeline tự động hóa khép kín: DVC kết hợp MinIO/S3 quản lý phiên bản dữ liệu, Apache Airflow lập lịch ingest và retrain định kỳ, MLflow theo dõi và tự động promote model đạt chuẩn R² ≥ 0.70 lên Production, cùng FastAPI phục vụ suy luận thời gian thực được đóng gói hoàn chỉnh bằng Docker Compose.

1. Tại sao lại cần một hệ thống MLOps cho bài toán Chứng khoán?

Trong phân tích tài chính và đầu tư định lượng (Quantitative Trading), nhiều kỹ sư dữ liệu dừng lại ở mức thử nghiệm mô hình trong Jupyter Notebook. Tuy nhiên, thị trường chứng khoán biến động liên tục: dữ liệu giao dịch mới sinh ra mỗi ngày, xu hướng vĩ mô thay đổi khiến mô hình rất nhanh bị lỗi thời (hiện tượngConcept Drift).

Nếu không có một quy trình vận hành tự động gọi là MLOps, bạn sẽ phải đối mặt với các vấn đề như:

Dữ liệu phân mảnh: Khó kiểm soát phiên bản dữ liệu sạch dùng để huấn luyện từng model.
Triển khai thủ công: Mỗi lần cập nhật dữ liệu mới phải chạy lại code bằng tay và copy file model .pkl lên server.
Mất dấu chất lượng: Không đo lường được mô hình nào đang chạy trên production và hiệu suất thực tế ra sao.

Dự án này được xây dựng nhằm giải quyết trọn vẹn vòng đời đó: biến một mô hình dự đoán giá cổ phiếu đóng cửa sau 1 tuần thành một hệ thống tự động hóa khép kín, tự nạp dữ liệu, tự đánh giá và phục vụ dự đoán qua API.

Ứng dụng thực tế của Hệ thống
Tự động hóa báo cáo phân tích tuần	Hệ thống tự động kích hoạt sau phiên giao dịch thứ Sáu, tải toàn bộ dữ liệu 10 năm của nhóm cổ phiếu bảo hiểm (BVH, BMI, MIG, BIC, PGI), xử lý dữ liệu khuyết và đưa ra dự báo mức giá kỳ vọng cho tuần kế tiếp.
Tích hợp vào nền tảng giao dịch	Cung cấp Rest API độ trễ thấp thông qua FastAPI để các ứng dụng Web/Mobile hoặc hệ thống Trading Bot có thể truy vấn mức giá dự báo ngay lập tức.
Quản trị mô hình tin cậy (Governance)	Chỉ những mô hình vượt qua bài kiểm tra chỉ số R² khắt khe mới được MLflow tự động đẩy lên nhãn Production, loại bỏ hoàn toàn rủi ro triển khai nhầm mô hình kém chất lượng.

Trong các bài toán Machine Learning thực tế, việc huấn luyện mô hình trên Jupyter Notebook chỉ chiếm khoảng 10-20% khối lượng công việc. Phần lớn thách thức nằm ở việc quản lý phiên bản dữ liệu, tự động hóa quy trình huấn luyện, đóng gói API và giám sát mô hình.

2. Thiết lập Môi trường & Quản lý Phiên bản Dữ liệu (DVC + S3/MinIO)

Bước 2.1: Thu thập và tiền xử lý dữ liệu với vnstock

Dữ liệu chuỗi thời gian chứng khoán thường gặp tình trạng khuyết thiếu do ngày nghỉ giao dịch hoặc nghẽn mạng. Ta xử lý bằng phương pháp nội suy tuyến tính (Linear Interpolation) kết hợp ffill/bfill.

# src/data_pipeline.py
import pandas as pd
from vnstock import stock_historical_data
from datetime import datetime

def fetch_and_clean_stock(symbol: str, start_date: str, end_date: str) -> pd.DataFrame:
    df = stock_historical_data(
        symbol=symbol, 
        start_date=start_date, 
        end_date=end_date, 
        resolution="1D", 
        type="stock"
    )
    # Đặt index dạng Datetime để nội suy chuỗi thời gian
    df['time'] = pd.to_datetime(df['time'])
    df = df.set_index('time').sort_index()

    # Xử lý missing values: forward fill trước, sau đó nội suy tuyến tính
    df['close'] = df['close'].ffill().interpolate(method='linear').bfill()
    
    # Tạo Target: Giá đóng cửa của 5 phiên tiếp theo (1 tuần giao dịch)
    df['target_close_1w'] = df['close'].shift(-5)
    df = df.dropna()
    return df

if __name__ == "__main__":
    data = fetch_and_clean_stock("BVH", "2016-01-01", "2026-01-01")
    data.to_csv("data/processed/BVH_processed.csv")

Bước 2.2: Versioning dữ liệu với DVC

Thay vì lưu file .csv nặng trực tiếp vào Git, ta sử dụng DVC để đồng bộ lên Object Storage (S3 hoặc MinIO cục bộ).

# Khởi tạo DVC và cấu hình MinIO Remote Storage
dvc init
dvc remote add -d local_storage s3://stock-mlops-bucket
dvc remote modify local_storage endpointurl http://localhost:9000
dvc remote modify local_storage access_key_id minioadmin
dvc remote modify local_storage secret_access_key minioadmin

# Track file dữ liệu và push
dvc add data/processed/BVH_processed.csv
git add data/processed/BVH_processed.csv.dvc .gitignore
git commit -m "feat: track cleaned stock dataset version 1"
dvc push

3. Huấn luyện & Quản lý Vòng đời Mô hình với MLflow

Mô hình được huấn luyện và lưu lại artifact kèm theo siêu tham số. Nếu R² đạt ngưỡng quy định (R² > 0.7), mô hình được tự động gán nhãn lên Production.

# src/train.py
import mlflow
import mlflow.sklearn
from sklearn.linear_model import Ridge
from sklearn.metrics import r2_score, mean_squared_error
from sklearn.model_selection import train_test_split
import pandas as pd

mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("stock_price_prediction")

def train_model(symbol: str = "BVH"):
    df = pd.read_csv(f"data/processed/{symbol}_processed.csv")
    
    features = ['open', 'high', 'low', 'close', 'volume']
    X = df[features]
    y = df['target_close_1w']
    
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
    
    with mlflow.start_run(run_name=f"{symbol}_linear_ridge"):
        alpha = 1.0
        model = Ridge(alpha=alpha)
        model.fit(X_train, y_train)
        
        preds = model.predict(X_test)
        r2 = r2_score(y_test, preds)
        rmse = mean_squared_error(y_test, preds, squared=False)
        
        # Log Metrics & Parameters
        mlflow.log_param("alpha", alpha)
        mlflow.log_metric("r2_score", r2)
        mlflow.log_metric("rmse", rmse)
        
        # Log Model và đăng ký vào Model Registry
        model_name = f"{symbol}_model"
        mlflow.sklearn.log_model(
            sk_model=model,
            artifact_path="model",
            registered_model_name=model_name
        )
        
        # Tự động promote lên Production nếu vượt qua tiêu chuẩn chất lượng
        if r2 >= 0.70:
            client = mlflow.tracking.MlflowClient()
            latest_version = client.get_latest_versions(model_name, stages=["None"])[0].version
            client.transition_model_version_stage(
                name=model_name,
                version=latest_version,
                stage="Production",
                archive_existing_versions=True
            )
            print(f"Model version {latest_version} promoted to PRODUCTION with R2: {r2:.4f}")

if __name__ == "__main__":
    train_model("BVH")

4. Tự động hóa Pipeline với Apache Airflow DAG

Thiết lập workflow định kỳ chạy vào cuối tuần: Tải dữ liệu mới -> Tiền xử lý -> Huấn luyện -> Đánh giá & Promote.

# airflow/dags/stock_retrain_dag.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'mlops_engineer',
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'stock_pipeline_retrain',
    default_args=default_args,
    description='Tự động cập nhật dữ liệu và huấn luyện lại mô hình giá cổ phiếu',
    schedule_interval='@weekly',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    ingest_and_clean_data = BashOperator(
        task_id='fetch_and_clean_data',
        bash_command='python /opt/airflow/dags/src/data_pipeline.py'
    )

    train_and_evaluate = BashOperator(
        task_id='train_and_register_model',
        bash_command='python /opt/airflow/dags/src/train.py'
    )

    ingest_and_clean_data >> train_and_evaluate

5. Xây dựng Serving API với FastAPI

Dịch vụ API sẽ tự động tải mô hình mới nhất đang ở trạng thái Production từ MLflow để thực hiện suy luận.

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import mlflow.pyfunc
import pandas as pd

app = FastAPI(title="Stock Prediction Inference Service", version="1.0")

mlflow.set_tracking_uri("http://mlflow_server:5000")

class StockFeatures(BaseModel):
    symbol: str = "BVH"
    open: float
    high: float
    low: float
    close: float
    volume: float

@app.post("/predict")
def predict_stock_price(payload: StockFeatures):
    try:
        # Load mô hình đang mang tag Production từ Registry
        model_uri = f"models:/{payload.symbol}_model/Production"
        loaded_model = mlflow.pyfunc.load_model(model_uri)
        
        input_data = pd.DataFrame([{
            'open': payload.open,
            'high': payload.high,
            'low': payload.low,
            'close': payload.close,
            'volume': payload.volume
        }])
        
        prediction = loaded_model.predict(input_data)
        return {
            "symbol": payload.symbol,
            "predicted_close_1w": float(prediction[0]),
            "status": "success"
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

6. Đóng gói & Triển khai Hệ thống với Docker Compose

Để tránh lỗi xung đột thư viện giữa Airflow core và các gói machine learning, ta tùy biến Dockerfile riêng cho Airflow worker:

# docker-compose.yml
version: '3.8'

services:
  # Object Storage cục bộ (thay thế S3/GCS khi phát triển local)
  minio:
    image: minio/minio:latest
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: server /data --console-address ":9001"

  # MLflow Tracking Server
  mlflow_server:
    image: ghcr.io/mlflow/mlflow:latest
    ports:
      - "5000:5000"
    environment:
      - MLFLOW_S3_ENDPOINT_URL=http://minio:9000
      - AWS_ACCESS_KEY_ID=minioadmin
      - AWS_SECRET_ACCESS_KEY=minioadmin
    command: mlflow server --backend-store-uri sqlite:///mlflow.db --default-artifact-root s3://stock-mlops-bucket/ --host 0.0.0.0 --port 5000

  # FastAPI Serving Service
  prediction_api:
    build:
      context: .
      dockerfile: Dockerfile.api
    ports:
      - "8000:8000"
    environment:
      - MLFLOW_TRACKING_URI=http://mlflow_server:5000
      - AWS_ACCESS_KEY_ID=minioadmin
      - AWS_SECRET_ACCESS_KEY=minioadmin
      - MLFLOW_S3_ENDPOINT_URL=http://minio:9000

7. Các lỗi thực tế thường gặp & Cách xử lý

Xung đột thư viện trong Airflow Container: Tránh cài thủ công bằng docker exec.
Không nên dùng docker exec để cài thư viện tạm thời vì container sẽ mất hết cài đặt khi khởi động lại. Hãy viết một Dockerfile kế thừa từ apache/airflow:2.x và thêm các thư viện cần thiết (vnstock, scikit-learn, mlflow) vào _PIP_ADDITIONAL_REQUIREMENTS hoặc chạy lệnh pip install ngay trong lúc build image.
Lỗi Network & Port Mapping khi kết nối Storage: Ưu tiên MinIO cho môi trường local.
Khi chạy trên Windows/WSL2, việc kết nối ra Cloud Storage bên ngoài thường gặp lỗi chứng chỉ hoặc cổng mạng bị chặn. Việc tích hợp sẵn MinIO container trong Docker network giúp giải quyết triệt để vấn đề endpoint mà vẫn giữ nguyên cú pháp S3 API.
Hiện tượng Data Leakage trong Time-series Split: Không dùng K-Fold ngẫu nhiên.
Với dữ liệu tài chính, tuyệt đối không sử dụng phép chia ngẫu nhiên (train_test_split(shuffle=True)) vì sẽ làm lộ thông tin tương lai về quá khứ (lookahead bias). Luôn đặt shuffle=False hoặc dùng TimeSeriesSplit.

8. Các câu hỏi thường gặp

Tại sao cần sử dụng DVC thay vì lưu trực tiếp file dữ liệu vào Git?
Git được thiết kế để quản lý mã nguồn nhẹ dạng văn bản, không phù hợp với các file dữ liệu lớn (CSV, Parquet, hình ảnh) dẫn đến phình to kích thước repository và giảm tốc độ clone/pull. DVC lưu trữ siêu dữ liệu phiên bản (con trỏ .dvc) trong Git, trong khi dữ liệu thực tế được đẩy lên Remote Object Storage (S3, GCS, MinIO).

Làm thế nào để tránh rò rỉ dữ liệu (Data Leakage) khi huấn luyện chuỗi thời gian chứng khoán?
Tuyệt đối không sử dụng phép chia ngẫu nhiên (shuffle=True) hoặc K-Fold Cross-Validation thông thường vì sẽ khiến dữ liệu tương lai bị lộ vào tập huấn luyện của quá khứ. Luôn thiết lập shuffle=False theo thứ tự thời gian hoặc sử dụng TimeSeriesSplit của Scikit-Learn.

Khi nào MLflow tự động chuyển đổi mô hình sang trạng thái Production?
Trong pipeline đánh giá (train_and_evaluate), hệ thống tính toán hệ số xác định R² (R-squared) trên tập kiểm thử độc lập. Nếu R² ≥ 0.70 (hoặc ngưỡng chuẩn định sẵn), MLflow Client sẽ tự động chuyển nhãn mô hình mới từ Staging/None sang Production và lưu trữ (Archive) phiên bản cũ.

Tại sao nên dùng MinIO thay vì kết nối trực tiếp AWS S3 khi phát triển ở môi trường Local/WSL2?
MinIO cung cấp Object Storage tương thích hoàn toàn với S3 API nhưng chạy trực tiếp dưới dạng Docker container cục bộ. Điều này giúp tránh phát sinh chi phí Cloud, không cần cấu hình IAM/mạng phức tạp, giảm độ trễ mạng và loại bỏ lỗi chặn cổng kết nối giữa container và internet ngoài.

9. Kết luận & Hướng phát triển tiếp theo

Hệ thống trên đã hoàn thiện chu trình MLOps khép kín từ khâu nạp dữ liệu, đóng gói artifact đến phục vụ dự đoán tự động. Việc thành thạo những công cụ này giúp các kỹ sư dữ liệu rút ngắn khoảng cách từ môi trường lab đến hệ thống thực tế doanh nghiệp. Để hoàn thiện hơn trong môi trường production, các bước nâng cấp tiếp theo bao gồm:

Explainable AI (XAI): Tích hợp SHAP/LIME để giải thích tại sao mô hình đưa ra mức giá đó.
Monitoring: Kết nối Prometheus và Grafana để theo dõi hiện tượng suy thoái dữ liệu (Data Drift) và suy thoái mô hình (Concept Drift).

Kiến trúc MLOps & Docker	★★★★★
Quản lý Data Versioning (DVC)	★★★★★
Pipeline Automation (Airflow)	★★★★
API Serving & Inference	★★★★★
Khả năng ứng dụng thực tế	★★★★★