Python量化回测完整实战用本地股票数据引擎构建多因子选股系统附完整代码

Python量化回测完整实战：用本地股票数据引擎构建多因子选股系统（附完整代码）

一、背景介绍

在量化投资领域，数据是一切策略的基石。无论是机构投资者还是个人量化开发者，如何快速、稳定、低成本地获取高质量的金融数据，始终是摆在面前的第一道门槛。

传统的量化数据获取方案主要依赖在线API接口，例如Tushare、AkShare、Wind、Choice等。这些方案各有优劣：在线API方便快捷，但往往面临调用频率限制、网络延迟、数据完整性不足以及长期使用成本高昂等问题。对于需要进行大规模回测、高频策略研发或对数据安全性有较高要求的开发者而言，一套本地部署、极速读取、全量覆盖的股票数据引擎几乎是刚需。

这正是 ig50本地股票数据引擎 应运而生的原因。ig50是一款面向量化开发者的专业级本地金融数据解决方案，它将A股、港股、美股、基金等全市场数据完整打包到本地，通过统一的文件系统路径接口供开发者调用，彻底告别了对外部网络的依赖。

本文将带领读者从零开始，使用 ig50本地股票数据引擎 构建一套完整的多因子选股系统。我们将从股票列表获取、历史行情读取、财务指标提取，到最终的因子加权选股和回测绩效分析，全程提供可直接运行的Python代码。

1.1 为什么选择ig50本地股票数据引擎？

在正式进入代码实战之前，先简要阐述ig50相较于传统在线数据方案的核心优势：

• 本地极速读取：所有数据存储在本地磁盘，无需网络请求，单只股票K线读取耗时在毫秒级，特别适合大规模回测场景。
• 全量数据覆盖：涵盖A股、港股、美股、基金等多个市场，共计221个数据集，从行情数据到财务指标、从股票列表到指数成分，一应俱全。
• 接口简洁统一：所有数据通过本地文件路径访问，如 base/gplist（股票列表）、time/history/trade（历史K线）、time/f10/fi（财务指标），学习成本极低。
• 数据格式统一：所有数据采用JSON格式存储，数组或对象结构，字段命名与官方文档一致，无需额外的格式转换。
• Python友好：配合pandas、numpy等数据科学库可以无缝衔接。

1.2 多因子选股策略概述

多因子选股模型是量化投资中应用最广泛的策略之一。其核心思想是：通过多个具有逻辑支撑的因子（如估值因子、成长因子、质量因子、动量因子等）对股票进行综合评分，选择得分最高的一篮子股票构建投资组合，以期获得超越市场基准的超额收益。

在本文中，我们将选取四个经典的基本面因子来构建选股模型：

通过ig50本地股票数据引擎，我们可以轻松获取计算这四个因子所需的全部数据。

二、数据接口说明

ig50本地股票数据引擎采用本地文件路径作为统一的数据访问接口。安装ig50后，所有数据将按照固定的目录结构存放在本地磁盘中。开发者只需通过文件路径即可读取对应的数据表，无需任何网络请求。

下面详细介绍本文将使用的三个核心数据接口。

2.1 股票列表：base/gplist

数据名称：沪深京A股-股票基础信息-股票列表-沪深京A股列表

本地路径：数据存放目录/base/gplist

数据说明：该接口返回全市场A股的基础信息列表，包括股票代码、股票名称、所属交易所、上市日期等字段。这是量化策略开发的起点——只有先获取完整的股票池，才能进行后续的选股和回测。

数据格式：[{},…]（JSON数组）

关键字段：

更新频率：每天16点，完成更新耗时约10分钟。

2.2 历史K线：time/history/trade/{股票代码}/{分时级别}

数据名称：沪深京A股-股票交易数据-股票K线-历史K线（数据范围：短分时两年，日线及以上所有）

本地路径：数据存放目录/time/history/trade/{股票代码}/{分时级别}

数据说明：该接口返回单只A股的历史K线数据，包括开盘价、最高价、最低价、收盘价、成交量、成交额等核心行情字段。这是计算价格类因子（如动量、波动率）和进行回测的基础数据。

数据格式：[{},…]（JSON数组，交易时间升序）

关键字段：

更新频率：分钟级别盘中每5分钟更新，日线及以上级别盘后15:35更新。

2.3 财务指标：time/f10/fi/{股票代码}

数据名称：沪深京A股-股票基础信息-上市公司详情-财务指标

本地路径：数据存放目录/time/f10/fi/{股票代码}

数据说明：该接口返回单只股票近四个季度的主要财务指标，按报告日期倒序。包括盈利能力（ROE、ROA、毛利率）、估值水平（PE、PB）、偿债能力、营运能力等数十个核心财务指标。这是构建基本面因子模型的核心数据源。

数据格式：[{}]（JSON数组）

关键字段（本文用到的）：

更新频率：每天15:30（更新完成约12小时）。

2.4 实时行情：time/real/{股票代码}

数据名称：沪深京A股-股票交易数据-股票行情-实时行情数据（3秒落盘）

本地路径：数据存放目录/time/real/{股票代码}

数据说明：该接口返回单只股票的实时交易数据（可理解为日线的最新数据），包括PE、PB、总市值等因子所需的实时数据。

数据格式：{}（JSON对象）

关键字段（本文用到的）：

更新频率：交易时间段每10秒，完成更新耗时约3秒。

三、完整Python代码示例

下面进入本文的核心部分：使用ig50本地股票数据引擎构建多因子选股系统的完整Python代码实现。

3.1 环境准备

首先确保你的Python环境中已安装以下依赖库：

pip install pandas numpy matplotlib tqdm

ig50本地数据引擎安装后，数据将按照目录结构存放在本地磁盘。请参考官方文档（ig50.com）完成安装和数据初始化。

3.2 完整代码

# -*- coding: utf-8 -*-
"""
多因子选股系统 – 基于ig50本地股票数据引擎
策略逻辑：PE + PB + ROE + 市值 四因子等权选股
回测周期：2020-01-01 至 2025-12-31
调仓频率：每月第一个交易日调仓
持仓数量：20只股票
"""

import os
import json
import pandas as pd
import numpy as np
from datetime import datetime
from tqdm import tqdm
import matplotlib.pyplot as plt
import matplotlib.dates as mdates

# ==========================================
# 1. 基础配置
# ==========================================

# ig50本地数据根目录（请根据实际安装路径修改）
IG50_DATA_ROOT = r"D:\\ig50\\data"

# 回测参数
START_DATE = "2020-01-01"
END_DATE = "2025-12-31"
HOLDING_COUNT = 20 # 持仓股票数量
REBALANCE_DAY = 1 # 每月第几个交易日调仓
INITIAL_CAPITAL = 1000000 # 初始资金（元）

def read_ig50_json(relative_path):
"""
读取ig50本地JSON数据
ig50的数据以JSON格式存储，数组或对象结构
:param relative_path: 相对路径，如 "base/gplist", "time/history/trade/000001/day"
:return: list 或 dict
"""
full_path = os.path.join(IG50_DATA_ROOT, relative_path)

if not os.path.exists(full_path):
return None

try:
with open(full_path, 'r', encoding='utf-8') as f:
data = json.load(f)
return data
except Exception as e:
print(f"[ig50] 读取数据失败 {relative_path}: {e}")
return None

# ==========================================
# 2. 获取股票池（base/gplist）
# ==========================================

def get_stock_universe():
"""
从ig50的base/gplist接口获取A股股票列表
并进行基础筛选（排除ST、退市股票，排除新股）
"""
print("[ig50] 正在读取股票列表 base/gplist …")
data = read_ig50_json("base/gplist")

if data is None:
raise ValueError("无法读取股票列表，请检查ig50数据路径配置")

# 转换为DataFrame，字段名使用ig50官方关键字
df_stocks = pd.DataFrame(data)

# 查看ig50返回的实际列名
print(f"[ig50] 股票列表字段: {df_stocks.columns.tolist()}")

# 基础筛选（使用ig50的中文字段名）
# 排除ST股
if '是否ST' in df_stocks.columns:
df_stocks = df_stocks[df_stocks['是否ST'] == 0]

# 排除新股
if '是否是新股' in df_stocks.columns:
df_stocks = df_stocks[df_stocks['是否是新股'] == 0]

# 仅保留沪深主板（北交所可根据需求加入）
if '交易所' in df_stocks.columns:
df_stocks = df_stocks[df_stocks['交易所'].isin(['sh', 'sz'])]

print(f"[ig50] 初始股票池筛选完成，共 {len(df_stocks)} 只股票")
return df_stocks.reset_index(drop=True)

# ==========================================
# 3. 读取单只股票的历史K线
# ==========================================

def get_daily_kline(code):
"""
从ig50的 time/history/trade/{code}/day 读取日K线
"""
data = read_ig50_json(f"time/history/trade/{code}/day")
if data is None or len(data) == 0:
return None

df = pd.DataFrame(data)

# 使用ig50的中文字段名处理
if '交易时间' in df.columns:
df['trade_date'] = pd.to_datetime(df['交易时间'], errors='coerce')
df = df.sort_values('trade_date').reset_index(drop=True)

return df

# ==========================================
# 4. 读取财务指标（获取ROE）
# ==========================================

def get_financial_indicators(code):
"""
从ig50的 time/f10/fi/{code} 读取财务指标
返回最新一期的ROE等核心指标
"""
data = read_ig50_json(f"time/f10/fi/{code}")
if data is None or len(data) == 0:
return {}

# 取最新一期（按报告日期倒序，第一条是最新）
latest = data[0]

result = {}
if '净资产收益率(%)' in latest:
result['roe'] = latest['净资产收益率(%)']
if '销售毛利率(%)' in latest:
result['gross_margin'] = latest['销售毛利率(%)']
if '净利润增长率(%)' in latest:
result['net_profit_growth'] = latest['净利润增长率(%)']
if '资产负债率(%)' in latest:
result['debt_ratio'] = latest['资产负债率(%)']

return result

# ==========================================
# 5. 读取实时行情（获取PE、PB、市值）
# ==========================================

def get_realtime_quote(code):
"""
从ig50的 time/real/{code} 读取实时行情数据
"""
data = read_ig50_json(f"time/real/{code}")
if data is None:
return {}

result = {}
if '市盈率（动态）' in data:
result['pe'] = data['市盈率（动态）']
if '市净率' in data:
result['pb'] = data['市净率']
if '总市值（元）' in data:
result['market_cap'] = data['总市值（元）']
if '每股收益（元）' in data:
result['eps'] = data['每股收益（元）']
if '每股净资产（元）' in data:
result['bps'] = data['每股净资产（元）']

return result

# ==========================================
# 6. 因子计算与选股
# ==========================================

def calculate_factors_and_rank(df_stocks):
"""
计算四因子并对股票进行综合评分
因子：PE（越低分越高）、PB（越低分越高）、ROE（越高分越高）、市值（越小分越高）
"""
print("[ig50] 正在计算多因子评分 …")

factors_list = []

for _, row in tqdm(df_stocks.iterrows(), total=len(df_stocks), desc="计算因子"):
code = row.get('股票代码') or row.get('code')
if not code:
continue

# 跳过北交所（可根据需求调整）
if code.startswith('8') or code.startswith('4'):
continue

# 读取实时行情（PE、PB、市值）
rt = get_realtime_quote(code)
pe = rt.get('pe')
pb = rt.get('pb')
market_cap = rt.get('market_cap')

# 读取财务指标（ROE）
fi = get_financial_indicators(code)
roe = fi.get('roe')

# 跳过数据缺失的股票
if pe is None or pb is None or market_cap is None or roe is None:
continue

# 跳过异常值
if pe <= 0 or pb <= 0 or market_cap <= 0:
continue
if pe > 500 or pb > 50: # 过滤极端值
continue

factors_list.append({
'code': code,
'name': row.get('股票名称', ''),
'pe': float(pe),
'pb': float(pb),
'roe': float(roe),
'market_cap': float(market_cap),
})

if len(factors_list) == 0:
raise ValueError("没有符合条件的股票，请检查数据是否完整")

df_factors = pd.DataFrame(factors_list)
print(f"[ig50] 有效因子样本数: {len(df_factors)}")

# ============== 因子标准化与评分 ==============
# PE：越低越好，排名越靠前分数越高
df_factors['pe_rank'] = df_factors['pe'].rank(ascending=True)

# PB：越低越好
df_factors['pb_rank'] = df_factors['pb'].rank(ascending=True)

# ROE：越高越好
df_factors['roe_rank'] = df_factors['roe'].rank(ascending=False)

# 市值：越小越好（小盘股溢价）
df_factors['cap_rank'] = df_factors['market_cap'].rank(ascending=True)

# 等权综合评分（四个因子排名相加，分数越低越好）
df_factors['total_score'] = (
df_factors['pe_rank'] +
df_factors['pb_rank'] +
df_factors['roe_rank'] +
df_factors['cap_rank']
)

# 按综合评分排序，分数越低排名越靠前
df_factors = df_factors.sort_values('total_score', ascending=True).reset_index(drop=True)

return df_factors

# ==========================================
# 7. 回测引擎
# ==========================================

def run_backtest(df_stocks, df_factors):
"""
简单回测：等权买入排名前N的股票，持有一个月后调仓
"""
print("[ig50] 正在执行回测 …")

# 获取排名前N的股票
top_stocks = df_factors.head(HOLDING_COUNT)
print(f"\\n[ig50] 选股结果（Top {HOLDING_COUNT}）:")
print(top_stocks[['code', 'name', 'pe', 'pb', 'roe', 'market_cap', 'total_score']].to_string(index=False))

# 获取这些股票的日K线进行回测
portfolio_returns = []
dates = []

print("\\n[ig50] 正在获取持仓股票的历史K线 …")
valid_count = 0

for _, stock in tqdm(top_stocks.iterrows(), total=len(top_stocks), desc="读取K线"):
code = stock['code']
df_k = get_daily_kline(code)
if df_k is None:
continue

# 筛选回测区间
df_k = df_k[(df_k['trade_date'] >= START_DATE) & (df_k['trade_date'] <= END_DATE)]
if len(df_k) < 30:
continue

valid_count += 1

# 计算日收益率
if '收盘价（元）' in df_k.columns:
df_k['daily_return'] = df_k['收盘价（元）'].pct_change().fillna(0)
else:
df_k['daily_return'] = df_k['收盘价'].pct_change().fillna(0) if '收盘价' in df_k.columns else 0

# 对齐日期
if len(dates) == 0:
dates = df_k['trade_date'].tolist()
portfolio_returns = df_k['daily_return'].tolist()
else:
# 简化处理：取交集日期
date_map = dict(zip(df_k['trade_date'], df_k['daily_return']))
new_returns = []
new_dates = []
for i, d in enumerate(dates):
if d in date_map:
new_returns.append(portfolio_returns[i] + date_map[d])
new_dates.append(d)
dates = new_dates
portfolio_returns = new_returns

if valid_count == 0:
raise ValueError("没有股票有足够的历史数据用于回测")

# 等权：除以持仓数
portfolio_returns = [r / valid_count for r in portfolio_returns]

# 计算累计净值
nav = [1.0]
for r in portfolio_returns:
nav.append(nav[–1] * (1 + r))

# 计算绩效指标
total_return = (nav[–1] – 1) * 100
annual_days = 252
trading_days = len(portfolio_returns)
annual_return = ((nav[–1]) ** (annual_days / trading_days) – 1) * 100 if trading_days > 0 else 0

returns_array = np.array(portfolio_returns)
annual_vol = returns_array.std() * np.sqrt(annual_days) * 100
sharpe = (annual_return – 3) / annual_vol if annual_vol > 0 else 0 # 无风险利率取3%

max_dd = 0
peak = nav[0]
for n in nav:
if n > peak:
peak = n
dd = (peak – n) / peak
if dd > max_dd:
max_dd = dd
max_dd *= 100

print(f"\\n{'='*50}")
print(f"[ig50] 回测绩效统计")
print(f"{'='*50}")
print(f"回测区间: {START_DATE} 至 {END_DATE}")
print(f"交易天数: {trading_days}")
print(f"有效持仓数: {valid_count}")
print(f"总收益率: {total_return:.2f}%")
print(f"年化收益率: {annual_return:.2f}%")
print(f"年化波动率: {annual_vol:.2f}%")
print(f"夏普比率: {sharpe:.2f}")
print(f"最大回撤: {max_dd:.2f}%")
print(f"{'='*50}")

# 可视化
fig, axes = plt.subplots(2, 1, figsize=(14, 10), gridspec_kw={'height_ratios': [3, 1]})

# 净值曲线
axes[0].plot(dates, nav[1:], 'b-', linewidth=1.5, label='多因子组合净值')
axes[0].axhline(y=1.0, color='gray', linestyle='–', alpha=0.5)
axes[0].set_title(f'ig50本地数据引擎 – 多因子选股回测净值曲线\\n(PE+PB+ROE+市值 四因子等权)', fontsize=14)
axes[0].set_ylabel('净值', fontsize=12)
axes[0].legend(loc='best')
axes[0].grid(True, alpha=0.3)

# 回撤曲线
drawdowns = []
peak = nav[0]
for n in nav:
if n > peak:
peak = n
drawdowns.append(–(peak – n) / peak * 100)

axes[1].fill_between(dates, drawdowns[1:], 0, color='red', alpha=0.3)
axes[1].plot(dates, drawdowns[1:], 'r-', linewidth=1)
axes[1].set_ylabel('回撤(%)', fontsize=12)
axes[1].set_xlabel('日期', fontsize=12)
axes[1].grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('ig50_multifactor_backtest.png', dpi=150, bbox_inches='tight')
print(f"\\n[ig50] 回测图表已保存: ig50_multifactor_backtest.png")

return {
'top_stocks': top_stocks,
'nav': nav,
'dates': dates,
'metrics': {
'total_return': total_return,
'annual_return': annual_return,
'annual_vol': annual_vol,
'sharpe': sharpe,
'max_drawdown': max_dd,
'trading_days': trading_days,
}
}

# ==========================================
# 8. 主程序入口
# ==========================================

def main():
print("=" * 60)
print("ig50本地股票数据引擎 – 多因子选股系统")
print("=" * 60)

# 步骤1：获取股票池
df_stocks = get_stock_universe()

# 步骤2：计算因子并排名
df_factors = calculate_factors_and_rank(df_stocks)

# 步骤3：执行回测
result = run_backtest(df_stocks, df_factors)

# 导出选股结果
result['top_stocks'].to_csv('ig50_multifactor_top20.csv', index=False, encoding='utf-8-sig')
print(f"\\n[ig50] 选股结果已导出: ig50_multifactor_top20.csv")

print("\\n" + "=" * 60)
print("完成！如需扩展策略，可修改因子权重、增加动量因子、或优化调仓逻辑。")
print("更多ig50数据接口请参考官方文档：ig50.com")
print("=" * 60)

if __name__ == "__main__":
main()

3.3 代码说明

以上代码包含以下核心模块：

3.4 运行效果

运行脚本后，你将看到：

• ig50数据读取的详细日志
• Top 20选股结果列表（含各因子数值和综合评分）
• 回测绩效统计（收益率、波动率、夏普比率、最大回撤等）
• 自动保存的净值曲线图和选股结果CSV文件

四、总结与扩展

4.1 本文总结

本文基于 ig50本地股票数据引擎，从零构建了一套完整的多因子选股回测系统。通过本文的实践，你可以看到：

4.2 策略扩展方向

本文实现的是一个基础版本的多因子策略，你可以从以下方向进行扩展：

• 增加因子：加入动量因子（20日涨幅）、波动率因子、换手率因子等
• 因子加权优化：使用机器学习方法（如线性回归、随机森林、XGBoost）进行因子权重优化
• 风险控制：加入行业中性化、市值中性化处理，控制组合的行业暴露和风格暴露
• 调仓优化：从月度调仓改为周度或日度调仓，加入交易成本和滑点模拟
• 高频信号：使用ig50的分钟级K线（time/history/trade/{code}/min）和L2数据构建高频因子

4.3 更多ig50数据接口

本文仅使用了ig50 221个数据集中的4个，还有大量有价值的数据等待你去探索：

• L2行情数据：五档盘口 time/real/trace/level5、逐笔交易 time/real/trace/onebyone、L2指标 time/real/trace/l2sign
• 资金流向数据：资金走势 time/zijin/zlzjzs、资金流入趋势 time/zijin/zjlrqs、行业资金流向 all/zjlx/zjhhy
• 龙虎榜数据：机构席位追踪 all/lhb/jgcz、营业部上榜统计 all/lhb/yybtj
• 融资融券数据：两融标的 all/rzrq/bd、两融交易明细 time/rzrq/mx/{code}
• 指数成分数据：指数成分列表 time/indextree/{index_code}

ig50本地股票数据引擎为量化开发者提供了完整的数据基础设施，让你可以把精力集中在策略研发上，而不是数据采集和维护上。更多接口详情请参考官方文档ig50.com。