基于 Python + QuantDash 搭建跨市场多因子选股系统：从因子计算到动态组合构建

pip install pandas numpy quantdash

import os
import numpy as np
import pandas as pd
from quantdash import QuantDash

# 1. 客户端初始化与选股池定义
# 混合选股池：包含 A股头部白马 与 港股互联网巨头
TICKER_POOL = [
    "600519.SH", "601318.SH", "600900.SH", "000333.SZ",  # A股
    "00700.HK", "03690.HK", "09988.HK", "01810.HK"       # 港股
]

api_key = os.getenv("QUANTDASH_API_KEY", "YOUR_API_KEY")
qd = QuantDash(api_key=api_key)

# 2. 批量拉取前复权历史日线数据
def fetch_pool_data(tickers, start_date):
    pool_data = {}
    for symbol in tickers:
        try:
            print(f"正在拉取标的数据: {symbol}")
            # 获取前复权历史K线，确保因子计算不因除权除息产生杂音
            df = qd.stock.get_kline(symbol=symbol, start_date=start_date, adjust="forward")
            if not df.empty:
                df['date'] = pd.to_datetime(df['date'])
                df.set_index('date', inplace=True)
                # 只保留收盘价进行特征计算
                pool_data[symbol] = df['close']
        except Exception as e:
            print(f"标的 {symbol} 数据获取失败: {e}")
    return pd.DataFrame(pool_data)

# 3. 计算多因子特征
def calculate_factors(price_df):
    factors_df = pd.DataFrame(index=price_df.columns)
    
    # 计算日收益率
    returns_df = price_df.pct_change()
    
    # 因子 1: 20日动量因子 (20-day cumulative return)
    # 计算公式：(今日收盘价 - 20天前收盘价) / 20天前收盘价
    factors_df['momentum_20'] = (price_df.iloc[-1] - price_df.iloc[-20]) / price_df.iloc[-20]
    
    # 因子 2: 20日低波动因子 (20-day Volatility)
    # 计算20日收益率的标准差，并取负值（波动越小，因子值越大）
    factors_df['low_vol_20'] = -returns_df.tail(20).std()
    
    return factors_df

# 4. 截面打分与排序构建组合
def generate_portfolio(factors_df, top_n=3):
    # 对因子进行 Z-Score 标准化，使其在同一维度下可比
    z_scores = factors_df.apply(lambda x: (x - x.mean()) / x.std() if x.std() != 0 else x)
    
    # 等权综合得分
    z_scores['composite_score'] = z_scores['momentum_20'] * 0.5 + z_scores['low_vol_20'] * 0.5
    
    # 按综合得分从高到低排序，筛选 Top N
    portfolio = z_scores.sort_values(by='composite_score', ascending=False)
    
    return portfolio

# --- 主程序运行 ---
if __name__ == "__main__":
    # 拉取最近 60 个交易日的数据，以确保有足够的历史长度计算20日因子
    start_date = "2026-03-01" 
    
    print("--- 步骤1：正在从 QuantDash 获取多市场历史行情数据 ---")
    raw_prices = fetch_pool_data(TICKER_POOL, start_date)
    
    # 剔除存在缺失值的数据行（保持截面对齐）
    clean_prices = raw_prices.dropna()
    print(f"\n有效交易日天数: {len(clean_prices)}")
    
    print("\n--- 步骤2：计算多因子特征指标 ---")
    factors = calculate_factors(clean_prices)
    print(factors)
    
    print("\n--- 步骤3：截面标准化并生成打分推荐组合 ---")
    result_portfolio = generate_portfolio(factors, top_n=3)
    print(result_portfolio[['momentum_20', 'low_vol_20', 'composite_score']])
    
    print("\n[推荐买入组合]：")
    for i, symbol in enumerate(result_portfolio.head(3).index):
        print(f"Top {i+1}: {symbol} (综合得分: {result_portfolio.loc[symbol, 'composite_score']:.4f})")