tools »
21天驯化AI打工仔 - 日线数据的定时获取（2）
数据如同血液，字段则是血型标记。本章带你深入A股数据的"涨跌停"与"ST"世界，让007助手为你揭秘如何用Tushare API完美修复那些缺失的关键字段，让量化策略在真实市场环境中游刃有余！
前言
根据上一章节，我们基本实现了日线定时获取的基本架构，但是其中有部分字段是不正确的，我们今天重点来对这些字段进行修复和完善......
"007，我们的日线数据定时获取系统已经基本成型，但数据字段还不够完整，特别是涨跌停价格、ST状态和复权因子这些关键信息都缺失了。"我一边查看ClickHouse中的数据，一边对我的AI助手说道。
"收到🫡，我马上为您解决这个问题。"007立刻回应道。
这是我们量化交易系统开发的第七天。前几天，我们已经搭建了一个能够定时从Tushare获取日线数据并存入ClickHouse的系统。但在实际使用过程中，我发现数据中缺少了几个关键字段，这些字段对于量化策略的开发至关重要。
问题分析：缺失的关键字段
在量化交易中，有几个字段对策略制定和回测至关重要：
-
涨跌停价格（limit_up和limit_down）：A股市场有涨跌停限制，这直接影响了交易策略的执行。如果不知道涨跌停价格，可能会制定出无法执行的交易计划。
-
ST状态（is_st）：ST股票有特殊的涨跌停限制（±5%而非±10%），且风险较高，许多策略会选择避开ST股票。
-
复权因子（adjust）：对于长期回测，正确的复权因子是保证数据一致性的关键。没有复权因子，就无法正确处理除权除息对股价的影响。
查看我们现有的代码，发现这些字段要么完全缺失，要么只是简单地填充了默认值：
| # 添加涨跌停和ST信息
df['limit_up'] = None
df['limit_down'] = None
df['is_st'] = 0
# 添加复权因子（这里简化处理，实际应该从adj_factor接口获取）
df['adjust'] = 1.0
|
这显然不能满足实际需求。我们需要从Tushare获取真实的数据来填充这些字段。
解决方案：使用Tushare API获取完整字段
经过讨论，我和007决定使用以下Tushare API来获取缺失的字段：
- stk_limit：获取涨跌停价格数据
- namechange：获取股票名称变更信息，用于判断ST状态
- adj_factor：获取复权因子数据
1. 获取涨跌停价格
涨跌停价格是交易策略中的重要参考点。在A股市场，普通股票的涨跌幅限制为±10%，ST股票为±5%，科创板和创业板为±20%。我们需要从Tushare的stk_limit接口获取这些数据。
007修改了_enrich_daily_data方法，添加了获取涨跌停价格的代码：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 | # 获取涨跌停价格
limit_df = self._call_tushare_api(
api_name='stk_limit',
params={'trade_date': trade_date},
fields='ts_code,trade_date,up_limit,down_limit'
)
# 添加涨跌停信息
if not limit_df.empty:
# 合并涨跌停信息
df = pd.merge(df, limit_df[['ts_code', 'up_limit', 'down_limit']],
on='ts_code', how='left')
else:
# 如果没有涨跌停数据，则添加空列
df['up_limit'] = None
df['down_limit'] = None
|
这段代码首先调用Tushare的stk_limit接口获取指定交易日的涨跌停价格数据，然后通过pd.merge将这些数据与原始日线数据合并。如果获取失败或数据为空，则添加空列作为占位符。
2. 判断ST状态
ST（Special Treatment）是对存在财务问题或其他风险的上市公司的特殊处理。ST股票有更严格的涨跌幅限制，且风险较高，因此在量化策略中通常需要特别对待。
判断一只股票是否为ST股票并不是简单地查询一个接口就能得到答案。我们需要通过股票名称变更记录来判断。如果股票名称中包含"ST"，且当前日期在ST开始日期和结束日期之间，则该股票在当前日期为ST股票。
007实现了一个复杂但高效的ST状态判断逻辑：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36 | # 获取ST股票信息
namechange_df = self._call_tushare_api(
api_name='namechange',
params={},
fields='ts_code,name,start_date,end_date,change_reason'
)
# 处理ST信息
st_codes = []
if not namechange_df.empty:
# 筛选出名称中包含ST的股票
st_df = namechange_df[namechange_df['name'].str.contains('ST', na=False)]
# 检查每个股票在当前日期是否为ST
for _, row in st_df.iterrows():
ts_code = row['ts_code']
start_date_str = row['start_date']
end_date_str = row['end_date']
# 转换日期格式
start_date = datetime.datetime.strptime(start_date_str, '%Y%m%d')
target_date = datetime.datetime.strptime(trade_date, '%Y%m%d')
# 处理结束日期
if pd.isna(end_date_str) or end_date_str is None:
end_date = datetime.datetime.now()
else:
end_date = datetime.datetime.strptime(end_date_str, '%Y%m%d')
# 判断当前日期是否在ST日期范围内
if (target_date - start_date).days >= 0 and (target_date - end_date).days <= 0:
st_codes.append(ts_code)
logger.debug(f"股票 {ts_code} 在 {trade_date} 为ST股票")
# 添加ST标志
df['is_st'] = df['ts_code'].isin(st_codes).astype(int)
|
这段代码首先获取所有股票的名称变更记录，然后筛选出名称中包含"ST"的记录。对于每条记录，检查目标日期是否在ST开始日期和结束日期之间。如果是，则将该股票添加到ST股票列表中。最后，通过检查每只股票是否在ST股票列表中，为原始数据添加is_st字段，值为0或1。
3. 获取复权因子
复权因子是处理股票除权除息的关键。在A股市场，上市公司分红送股会导致股价变动，使得历史数据不连续。通过复权因子，我们可以将历史价格调整为连续的序列，便于分析和回测。
007添加了获取复权因子的代码：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 | # 获取复权因子
adj_factor_df = self._call_tushare_api(
api_name='adj_factor',
params={'trade_date': trade_date},
fields='ts_code,trade_date,adj_factor'
)
# 添加复权因子
if not adj_factor_df.empty:
# 合并复权因子信息
df = pd.merge(df, adj_factor_df[['ts_code', 'adj_factor']],
on='ts_code', how='left')
# 将缺失值填充为1.0
df['adj_factor'] = df['adj_factor'].fillna(1.0)
else:
# 如果没有复权因子数据，则添加默认值
df['adj_factor'] = 1.0
# 重命名adj_factor为adjust
df = df.rename(columns={'adj_factor': 'adjust'})
|
这段代码调用Tushare的adj_factor接口获取复权因子数据，然后通过pd.merge将这些数据与原始日线数据合并。对于缺失的复权因子，填充默认值1.0（表示不需要复权）。最后，将列名从adj_factor重命名为adjust，以符合我们的数据库结构。
处理历史数据的特殊挑战
对于当日数据，我们可以直接获取当日的涨跌停价格、ST状态和复权因子。但对于历史数据，情况会更复杂，因为我们需要处理一段时间内的数据。
007为历史数据设计了更高效的处理方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29 | # 将涨跌停信息转换为字典，方便查找
limit_info = {}
for _, row in limit_df.iterrows():
ts_code = row['ts_code']
trade_date = row['trade_date']
key = f"{ts_code}_{trade_date}"
limit_info[key] = {
'up_limit': row['up_limit'],
'down_limit': row['down_limit']
}
# 添加涨跌停和ST信息
df['limit_up'] = None
df['limit_down'] = None
df['is_st'] = 0
for i, row in df.iterrows():
ts_code = row['ts_code']
trade_date = row['trade_date']
key = f"{ts_code}_{trade_date}"
# 添加涨跌停信息
if key in limit_info:
df.at[i, 'limit_up'] = limit_info[key]['up_limit']
df.at[i, 'limit_down'] = limit_info[key]['down_limit']
# 添加ST信息
if key in st_info:
df.at[i, 'is_st'] = st_info[key]
|
这种方法首先将涨跌停信息和ST信息转换为字典，键为股票代码和交易日期的组合，值为对应的数据。然后遍历原始数据的每一行，根据股票代码和交易日期构建键，从字典中查找对应的数据。这种方法比直接使用pd.merge更灵活，可以处理更复杂的情况。
对于复权因子，也采用了类似的方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 | # 将复权因子信息转换为字典，方便查找
adj_factor_info = {}
for _, row in adj_factor_df.iterrows():
ts_code = row['ts_code']
trade_date = row['trade_date']
key = f"{ts_code}_{trade_date}"
adj_factor_info[key] = row['adj_factor']
# 为每条记录添加复权因子
df['adjust'] = 1.0 # 默认值
for i, row in df.iterrows():
ts_code = row['ts_code']
trade_date = row['trade_date']
key = f"{ts_code}_{trade_date}"
# 添加复权因子信息
if key in adj_factor_info:
df.at[i, 'adjust'] = adj_factor_info[key]
|
优化API调用：直接使用Tushare库
在实现过程中，我们发现一个重要的问题：最初的代码尝试通过HTTP请求直接调用Tushare API，这不仅效率低，还可能导致各种网络问题。
"007，我们应该直接使用Tushare库进行数据获取，而不是通过HTTP请求。"我提醒道。
"收到🫡，我马上修改代码。"007迅速响应。
修改后的代码如下：
| # 初始化Tushare API
ts.set_token(self.token)
self.pro = ts.pro_api() # 不再传入api_url参数
|
这种方式直接使用Tushare官方库的API，不仅代码更简洁，而且更可靠，避免了HTTP请求可能带来的各种问题。
测试与验证
完成代码修改后，我们进行了测试，确保新添加的字段能够正确获取和存储。
首先，我们获取当日数据：
| python main.py daily --batch-size 1000
|
然后，我们获取历史数据：
| python main.py history --days 7 --batch-size 1000
|
最后，我们检查ClickHouse中的数据：
| 2025-05-22 17:10:24,352 - day_bar_fetcher - INFO - Tushare API初始化成功
2025-05-22 17:10:24,393 - day_bar_fetcher - INFO - Redis连接成功
2025-05-22 17:10:24,538 - day_bar_fetcher - INFO - ClickHouse连接成功
2025-05-22 17:10:24,554 - day_bar_fetcher - INFO - 已确保表 RealTime_DailyLine_DB.day_bar 存在
2025-05-22 17:10:24,555 - day_bar_fetcher - INFO - 调度器初始化成功
2025-05-22 17:10:24,585 - day_bar_fetcher - INFO - ==================================================
2025-05-22 17:10:24,585 - day_bar_fetcher - INFO - ClickHouse中已有数据的时间范围: 20250515 - 20250521
2025-05-22 17:10:24,588 - day_bar_fetcher - INFO - ClickHouse中共有 26953 条数据记录
2025-05-22 17:10:24,588 - day_bar_fetcher - INFO - ==================================================
|
测试结果显示，所有字段都已正确获取和存储。特别是，我们可以看到涨跌停价格、ST状态和复权因子这些关键字段都已经填充了真实的数据，而不是简单的默认值。
成果与思考
通过这次字段修复，我们的日线数据定时获取系统变得更加完善和实用。现在，系统可以获取并存储以下关键字段：
- 基本OHLC数据：开盘价、最高价、最低价、收盘价、成交量、成交额
- 涨跌停价格：上涨限制价格、下跌限制价格
- ST状态：是否为ST股票
- 复权因子：用于前复权或后复权计算
这些字段为后续的量化策略开发提供了坚实的数据基础。特别是，有了正确的涨跌停价格和ST状态，我们可以更准确地模拟实际交易环境；有了复权因子，我们可以正确处理历史数据的连续性问题。
"007，你做得很好！这次的字段修复让我们的系统更加完善了。"我由衷地赞赏道。
下一步计划
完成字段修复后，我们的日线数据定时获取系统已经相当完善。但在量化交易系统的开发道路上，我们还有很长的路要走。接下来，我们计划：
- 系统稳定性测试：在实际环境中长期运行系统，确保其稳定性和可靠性。
- 数据质量监控：开发监控工具，实时检查数据的完整性和准确性。
- 扩展数据源：除了日线数据，还可以考虑添加分钟线、tick数据等更细粒度的数据。
- 策略回测模块：开发策略回测模块，利用已获取的数据进行策略回测。
这些计划将在接下来的日子里逐步实施。我相信，在007的协助下，我们的量化交易系统将会越来越强大。
21天的挑战仍在继续，期待更多的突破和成果！