云原生量化实践：基于 Docker + Python 搭建全自动多市场「量化数据 ETL 跑批系统」

┌────────────────────────────────────────────────────────┐
│               腾讯云服务器 (Docker Container)            │
│                                                        │
│  ┌────────────┐     1. 触发任务     ┌──────────────┐   │
│  │ 定时调度器  │ ─────────────────> │ Python ETL   │   │
│  │ (APScheduler)│                    │ 跑批主程序   │   │
│  └────────────┘                    └──────┬───────┘   │
│                                           │            │
│  ┌────────────┐     3. 转换并落库   ┌─────▼────────┐   │
│  │ 结构化数据库│ <───────────────── │ 2. 抓取原始  │   │
│  │ (SQLite/PG) │                    │ 行情数据     │   │
│  └────────────┘                    └──────▲───────┘   │
└───────────────────────────────────────────┼────────────┘
                                            │
                                  ┌─────────┴────────┐
                                  │ QuantDash 云 API │
                                  └──────────────────┘

import os
import sys
import logging
from datetime import datetime
from sqlalchemy import create_engine
from quantdash import QuantDash

# 配置日志系统
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

# 1. 初始化环境变量
API_KEY = os.getenv("QUANTDASH_API_KEY")
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///quant_market_data.db")
TARGET_SYMBOLS = os.getenv("TARGET_SYMBOLS", "600519.SH,AAPL.US,00700.HK")

if not API_KEY:
    logging.error("未配置 QUANTDASH_API_KEY 环境变量，程序退出。")
    sys.exit(1)

# 初始化云端数据库引擎与 QuantDash 客户端
engine = create_engine(DATABASE_URL)
qd = QuantDash(api_key=API_KEY)

def run_etl():
    logging.info("================ 开始运行每日行情 ETL 跑批任务 ================")
    symbols = [s.strip() for s in TARGET_SYMBOLS.split(",")]
    
    for symbol in symbols:
        logging.info(f"[Extract] 正在从 QuantDash 提取标的 {symbol} 的日K线...")
        try:
            # 1. Extraction: 提取原始数据 (直接获取 DataFrame 格式)
            df = qd.klines.get(symbol=symbol, period="1d", to_dataframe=True)
            if df.empty:
                logging.warning(f"标的 {symbol} 返回数据为空，跳过。")
                continue
                
            # 2. Transformation: 数据清洗与转换
            # 统一将交易日期列转换为标准的 datetime 格式
            df['trade_date'] = pd.to_datetime(df['trade_date'])
            # 增量处理：增加更新时间戳，便于审计数据
            df['updated_at'] = datetime.utcnow()
            
            # 计算简单衍生指标：例如 5 日均线、10 日均线
            df['ma_5'] = df['close'].rolling(window=5).mean()
            df['ma_10'] = df['close'].rolling(window=10).mean()
            
            # 3. Loading: 写入数据库
            # 使用 SQLAlchemy 引擎，由于是跑批更新，我们对表采用 append (增量追加) 或 replace
            # 生产环境建议通过唯一索引 (symbol + trade_date) 进行 upsert 处理，这里简化为 replace/append 演示
            table_name = "kline_daily"
            df.to_sql(name=table_name, con=engine, if_exists="replace", index=False)
            logging.info(f"[Load] 标的 {symbol} 成功导入数据库表 {table_name}，共 {len(df)} 行数据。")
            
        except Exception as e:
            logging.error(f"处理标的 {symbol} 时发生异常: {e}")
            
    logging.info("================ ETL 跑批任务顺利完成 ================")

if __name__ == "__main__":
    run_etl()

import time
import logging
from apscheduler.schedulers.blocking import BlockingScheduler
from etl_job import run_etl

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

if __name__ == "__main__":
    logging.info("启动量化数据调度服务，等待定时触发...")
    
    scheduler = BlockingScheduler()
    # 设定每日北京时间 23:30 自动执行 ETL 跑批（通常此时多市场已收盘并完成结算）
    scheduler.add_job(run_etl, 'cron', hour=23, minute=30, timezone='Asia/Shanghai')
    
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        logging.info("调度服务安全退出。")

quantdash>=0.1.0
pandas>=1.5.0
sqlalchemy>=1.4.0
apscheduler>=3.9.0
pyarrow>=9.0.0

# 使用轻量级 Python 基础镜像
FROM python:3.10-slim

# 设置容器内工作目录
WORKDIR /app

# 设置时区为亚洲/上海，确保 cron/scheduler 定时逻辑准确
ENV TZ=Asia/Shanghai
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone

# 复制依赖描述文件并安装
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码到容器内
COPY . .

# 启动定时调度主程序
CMD ["python", "main.py"]

# 1. 构建 Docker 镜像
docker build -t quant-etl-system:v1 .

# 2. 运行容器，通过环境变量传入 API Key 以及你要监控的自选股
docker run -d \
  --name my-quant-etl \
  -e QUANTDASH_API_KEY="您的QuantDash真实APIKEY" \
  -e TARGET_SYMBOLS="600519.SH,AAPL.US,00700.HK,000001.SZ" \
  -v $(pwd)/data:/app/data \
  quant-etl-system:v1