如何利用腾讯云函数 (SCF) 零成本搭建每日自动化的金融行情数据同步通道？

┌─────────────────┐      ┌─────────────────┐      ┌────────────────────────┐      ┌─────────────────┐
│  腾讯云定时触发器 │ ───> │  腾讯云函数 (SCF) │ ───> │  QuantDash 数据源 (API) │ ───> │  腾讯云数据库 CDB│
│ (Timer Trigger) │      │ (Python 运行环境) │      │ (获取 A/港/美 历史日K线)│      │  (MySQL/PG 存储) │
└─────────────────┘      └─────────────────┘      └────────────────────────┘      └─────────────────┘

import os
import sys
import pandas as pd
from sqlalchemy import create_engine
from quantdash import QuantDash

# 获取环境变量（建议将敏感配置信息放在云函数环境变量中）
API_KEY = os.environ.get("QUANTDASH_API_KEY")
DB_HOST = os.environ.get("DB_HOST")
DB_PORT = os.environ.get("DB_PORT", "3306")
DB_USER = os.environ.get("DB_USER")
DB_PASS = os.environ.get("DB_PASS")
DB_NAME = os.environ.get("DB_NAME")

# 1. 初始化数据库连接引擎 (以 MySQL 为例)
db_url = f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4"
engine = create_engine(db_url)

# 2. 初始化 QuantDash 客户端
qd = QuantDash(api_key=API_KEY)

def sync_daily_data(symbols: list):
    """
    获取指定标的的最新日K线并增量写入云数据库
    """
    for symbol in symbols:
        try:
            print(f"开始同步标的数据: {symbol}")
            # 拉取最新 2 天的日K线（覆盖当天和前一交易日，避免因时差或收盘结算延迟导致漏数据）
            df = qd.klines.get(
                symbol=symbol,
                period="1d",
                count=2,
                to_dataframe=True
            )
            
            if df.empty:
                print(f"[{symbol}] 未获取到行情数据，跳过")
                continue
                
            # 标准化字段为小写
            df.columns = [col.lower() for col in df.columns]
            # 增加一个标识标的代码的列
            df["symbol"] = symbol
            
            # 使用 Pandas 的 to_sql 增量写入云数据库
            # 'append' 模式可以避免覆盖历史数据
            # 如果需要去重，建议在数据库表结构中将 (symbol, time) 设为唯一联合索引，并捕获 DuplicateKey 异常
            df.to_sql(
                name="stock_daily_kline",
                con=engine,
                if_exists="append",
                index=False
            )
            print(f"[{symbol}] 同步成功，成功写入 {len(df)} 条记录。")
            
        except Exception as e:
            print(f"[{symbol}] 同步异常: {e}")

def main_handler(event, context):
    """
    腾讯云 SCF 云函数标准入口
    """
    print("=== 开始运行每日行情同步任务 ===")
    
    # 需要监控并同步的跨市场标的池
    watch_pool = [
        "600519.SH",  # 贵州茅台
        "000001.SZ",  # 平安银行
        "00700.HK",   # 腾讯控股
        "AAPL.US"     # 苹果
    ]
    
    sync_daily_data(watch_pool)
    
    return {
        "status": "success",
        "message": "Daily kline sync completed successfully."
    }

0 30 16 * * MON-FRI *