🪶 Airflow로 완성하는 AI 퀀트 트레이딩 파이프라인

🪶 Airflow로 완성하는 AI 퀀트 트레이딩 파이프라인

— 데이터 수집부터 모델 배포까지 한 번에 자동화하기

지난 글에서는 MLflow로 모델 버전 관리와 자동 배포까지 구현했습니다.
이제 남은 마지막 단계는,
이 모든 작업을 Airflow로 스케줄링·자동화해 완전한 MLOps 시스템으로 만드는 것입니다.

오늘은 데이터 수집 → 학습 → 검증 → 배포 → 리포트 생성을
하나의 Airflow DAG(Directed Acyclic Graph)로 묶는 실전 구축 편입니다.

---

🎯 목표

“매일 아침 6시, 자동으로 퀀트 모델이 학습되고,
성능이 개선되면 자동으로 배포되는 시스템”

즉, AI가 스스로 데이터와 코드를 업데이트하며 진화하는 환경을 만드는 것입니다.

---

⚙️ 1️⃣ Airflow 환경 구성

설치

pip install apache-airflow==2.9.0
airflow db init
airflow users create \
    --username admin \
    --password admin \
    --firstname Quant \
    --lastname AI \
    --role Admin \
    --email quant@ai.com

Airflow Web UI 실행:

airflow webserver -p 8080
airflow scheduler

👉 http://localhost:8080 접속 → “quant_ai_pipeline” DAG이 보이게 됩니다.

---

🧱 2️⃣ DAG 기본 구조

반응형

# dags/quant_ai_pipeline.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import os, subprocess, requests

default_args = {
    "owner": "quant_ai",
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

def collect_data():
    subprocess.run(["python", "scripts/collect_factors.py"], check=True)

def train_model():
    subprocess.run(["python", "scripts/train_transformer.py"], check=True)

def validate_model():
    subprocess.run(["python", "scripts/validate_model.py"], check=True)

def deploy_model():
    subprocess.run(["mlflow", "models", "serve", "-m", "models:/quant_ai/Production", "-p", "6000"])

def notify_slack(msg="Pipeline 완료"):
    requests.post(os.getenv("SLACK_WEBHOOK_URL"), json={"text": f"🚀 {msg}"})

with DAG(
    "quant_ai_pipeline",
    default_args=default_args,
    description="AI Quant End-to-End MLOps Pipeline",
    schedule_interval="0 6 * * *",  # 매일 오전 6시
    start_date=datetime(2025, 1, 1),
    catchup=False,
) as dag:

    t1 = PythonOperator(task_id="collect_data", python_callable=collect_data)
    t2 = PythonOperator(task_id="train_model", python_callable=train_model)
    t3 = PythonOperator(task_id="validate_model", python_callable=validate_model)
    t4 = PythonOperator(task_id="deploy_model", python_callable=deploy_model)
    t5 = PythonOperator(task_id="notify", python_callable=lambda: notify_slack("✅ 오늘의 파이프라인 완료"))

    t1 >> t2 >> t3 >> t4 >> t5

---

📊 3️⃣ DAG 단계별 설명

단계 역할 도구

collect_data	yfinance, PostgreSQL, Airflow XCom	실시간 팩터 데이터 수집
train_model	MLflow, PyTorch	Transformer 또는 RL 모델 학습
validate_model	Scikit-learn	MSE, R², Sharpe 등 검증
deploy_model	MLflow Serve, Docker	모델 자동 배포
notify	Slack	결과 알림 전송

각 단계는 의존 관계(task dependency) 로 연결되어 있습니다.
즉, 데이터가 수집되어야 학습이 시작되고,
검증 통과 후에만 배포가 진행됩니다.

---

🧩 4️⃣ XCom으로 단계 간 데이터 전달

Airflow의 XCom을 활용하면 이전 단계의 출력을 다음 태스크에 넘길 수 있습니다.

def validate_model(**context):
    result = subprocess.run(["python", "scripts/validate_model.py"], capture_output=True, text=True)
    context['ti'].xcom_push(key='validation_score', value=result.stdout)

def deploy_model(**context):
    score = float(context['ti'].xcom_pull(key='validation_score'))
    if score > 0.8:
        subprocess.run(["python", "scripts/deploy_model.py"])
    else:
        print("🚫 모델 성능 미달, 배포 취소")

👉 이렇게 하면 성능 기준을 자동으로 평가하고,
성공한 모델만 배포됩니다.

---

☁️ 5️⃣ MLflow + Airflow 통합

Airflow DAG에서 MLflow 실험 기록을 자동화합니다.

def train_model():
    import mlflow
    mlflow.set_tracking_uri("http://localhost:5001")
    mlflow.set_experiment("quant_ai_pipeline")
    with mlflow.start_run(run_name="daily_train"):
        subprocess.run(["python", "scripts/train_transformer.py"], check=True)
        mlflow.log_param("source", "Airflow-DAG")

MLflow UI에서 각 Airflow 실행을 “하루 단위 모델 실험”으로 관리할 수 있습니다.

---

📈 6️⃣ Streamlit 실시간 모니터링

import streamlit as st
import pandas as pd
import requests

AIRFLOW_URL = "http://localhost:8080/api/v1/dags/quant_ai_pipeline/dagRuns"
MLFLOW_URL = "http://localhost:5001/api/2.0/preview/mlflow/experiments/list"

st.title("📊 Quant AI Pipeline Monitoring")
st.subheader("Airflow DAG 상태")
st.json(requests.get(AIRFLOW_URL, auth=('admin', 'admin')).json())

st.subheader("최근 MLflow 실험 기록")
data = requests.get(MLFLOW_URL).json()
st.write(pd.DataFrame(data['experiments']))

이 Streamlit 대시보드에서는

• Airflow DAG 실행 상태
• MLflow 실험 결과
• 최근 배포된 모델 버전
  을 한 화면에서 모니터링할 수 있습니다.

---

🔁 7️⃣ Airflow + Slack 알림

def notify_slack(msg):
    requests.post(os.getenv("SLACK_WEBHOOK_URL"), json={"text": f"📢 {msg}"})

def on_failure_callback(context):
    notify_slack(f"❌ 실패: {context['task_instance'].task_id}")

t1 = PythonOperator(
    task_id="collect_data",
    python_callable=collect_data,
    on_failure_callback=on_failure_callback
)

어떤 단계에서 실패하더라도 Slack으로 즉시 알림이 전송됩니다.

---

🧠 8️⃣ Docker Compose로 통합 실행

version: "3.9"
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
  airflow:
    image: apache/airflow:2.9.0
    depends_on: [postgres]
    ports:
      - "8080:8080"
    volumes:
      - ./dags:/opt/airflow/dags
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__WEBSERVER__RBAC: "True"
      SLACK_WEBHOOK_URL: ${SLACK_WEBHOOK_URL}

---

📌 정리

단계 설명

1	Airflow DAG 정의 및 스케줄 등록
2	MLflow 통합으로 모델 실험 기록
3	XCom을 통한 성능 검증 결과 전달
4	성공 모델만 자동 배포
5	Slack + Streamlit으로 실시간 모니터링

이제 매일 아침, 당신이 출근하기 전에
AI가 알아서 새로운 데이터를 학습하고, 모델을 평가하고, 필요하면 배포합니다.

이게 바로 **“스스로 진화하는 퀀트 시스템”**의 핵심입니다.

---

📘 다음 글 예고

다음 편에서는 **“실전 퀀트 펀드 운영 전략 – 백테스트와 실시간 거래 데이터 동기화 시스템”**을 다룹니다.
AI 모델이 학습한 결과를 실제 거래소 데이터와 동기화하여
실시간 포트폴리오 추적 및 위험 관리 시스템을 완성할 예정입니다.

---

 

Airflow,MLflow,MLOps,퀀트AI,자동배포,데이터파이프라인,파이썬자동화,PostgreSQL,Streamlit,AI투자

---