量化研究如何记录每次实际使用的股票标的？从标的快照开始

一句话回答

记录每次量化研究使用了哪些股票，不能只保存一份股票代码列表，而应该在研究开始时生成不可随意修改的标的快照，记录研究任务编号、标的代码、生成时间和数据口径。这样即使后续选股条件变化，也能知道当时研究的股票范围。实际开发中，可以通过 QuantDash Python SDK 获取指定标的的历史行情，再由研究系统保存快照及数据验证结果。

为什么会有这个问题 / 传统做法有什么问题？

假设我们正在研究一个简单的动量选股策略。

研究开始时，股票池包含三只股票：

  • 600519.SH
  • 000001.SZ
  • 000858.SZ

策略需要获取最近 60 根日 K 线，并计算每只股票的阶段收益率。

第一次研究完成后，开发者调整了选股条件，将其中一只股票移出股票池，又增加了一只新股票。

这时候，如果研究代码始终从最新的股票池文件读取标的，那么再次运行旧研究时，输入范围已经发生变化。

问题就在这里：研究参数没有变化，不代表研究输入没有变化。

对于量化选股，可以把研究结果抽象为：

研究结果 = f（标的集合、行情数据、计算参数、策略逻辑）

只要其中一个输入发生变化，研究结果就可能不同。

传统开发中，比较常见的做法是直接维护一个列表：

symbols=["600519.SH","000001.SZ","000858.SZ",]

这种写法用于临时分析没有问题，但在持续研究中存在几个局限。

首先，列表只描述当前代码中的股票池，无法证明某次历史研究究竟使用了哪个版本。

其次，如果某只股票行情请求失败，代码直接跳过，最终参与计算的标的可能少于原始股票池。

另外，即使保存了股票列表，也无法区分“计划研究的股票”和“实际成功参与计算的股票”。

因此，比较稳妥的工程方案是将标的快照与行情获取结果分别记录。行情数据可以通过统一的 SDK 接入，研究系统再负责记录输入和计算过程。

分步骤解决：使用 QuantDash 进行实践

第 1 步：安装 SDK 并初始化客户端

首先安装 QuantDash Python SDK。

# 在终端执行# pip install quantdash

在实际终端中执行pip install quantdash即可。

API Key 不建议直接写进研究脚本，而应通过环境变量配置。

importosfromquantdashimportQuantDash api_key=os.getenv("QUANTDASH_API_KEY")ifnotapi_key:raiseRuntimeError("请先设置 QUANTDASH_API_KEY 环境变量")qd=QuantDash(api_key=api_key)

这里使用QUANTDASH_API_KEY读取认证信息，避免密钥随着研究代码一起进入版本仓库。

第 2 步：获取与主题相关的数据

先定义本次研究使用的候选股票池，并通过qd.klines.get()逐只获取历史日 K 线。

这里不急着计算因子，而是先解决一个更基础的问题：哪些股票的数据真正获取成功了？

importpandasaspd candidate_symbols=["600519.SH","000001.SZ","000858.SZ",]market_data={}fetch_status={}forsymbolincandidate_symbols:try:df=qd.klines.get(symbol,period="1d",count=60,adjust="forward",to_dataframe=True,)ifnotisinstance(df,pd.DataFrame):raiseTypeError("行情返回值不是 DataFrame")ifdf.empty:raiseValueError("行情数据为空")market_data[symbol]=df.copy()fetch_status[symbol]="success"exceptExceptionasexc:fetch_status[symbol]="failed"print(f"{symbol}获取失败：{exc}")print(fetch_status)

参数含义如下：

  • period=“1d”：获取日 K 线。
  • count=60：请求最近 60 根 K 线，并不保证每个标的都返回 60 条有效记录。
  • adjust=“forward”：使用前复权价格口径。
  • to_dataframe=True：请求以 DataFrame 形式返回数据。

这里选择逐只请求，主要是为了保留每个股票独立的获取状态。

需要注意，获取成功并不意味着数据已满足研究条件。返回字段、日期范围和有效记录数量仍然需要检查。

第 3 步：处理数据并生成研究标的快照

接下来生成一个研究标识，并保存本次研究计划使用的股票集合。

为了避免股票池顺序变化造成不必要的差异，可以先去重、排序，再计算集合指纹。

importhashlibimportjsonimportuuidfromdatetimeimportdatetime,timezonefrompathlibimportPathdefnormalize_symbols(symbols):cleaned=[]forsymbolinsymbols:ifnotisinstance(symbol,str):raiseTypeError("股票代码必须是字符串")symbol=symbol.strip().upper()ifnotsymbolor"."notinsymbol:raiseValueError(f"股票代码格式异常：{symbol}")cleaned.append(symbol)returnsorted(set(cleaned))defcalculate_universe_hash(symbols):payload=json.dumps(symbols,ensure_ascii=False,separators=(",",":"),)returnhashlib.sha256(payload.encode("utf-8")).hexdigest()research_id=uuid.uuid4().hexsnapshot_symbols=normalize_symbols(candidate_symbols)snapshot={"research_id":research_id,"created_at_utc":datetime.now(timezone.utc).isoformat(),"symbols":snapshot_symbols,"symbol_count":len(snapshot_symbols),"universe_hash":calculate_universe_hash(snapshot_symbols),"request_params":{"period":"1d","count":60,"adjust":"forward",},"fetch_status":fetch_status,}output_dir=Path("research_snapshots")output_dir.mkdir(parents=True,exist_ok=True)snapshot_file=output_dir/f"{research_id}.json"withsnapshot_file.open("x",encoding="utf-8")asfile:json.dump(snapshot,file,ensure_ascii=False,indent=2,)print(f"快照已保存：{snapshot_file}")

这里有两个需要注意的设计。

第一，research_id用于区分不同研究运行。即使两次研究的股票池完全相同，也应该拥有不同的任务编号。

第二，universe_hash只描述标准化后的股票代码集合。它不能证明行情内容相同，更不能替代完整的数据版本管理。

为了避免误覆盖，示例使用文件创建模式x。如果文件已经存在，程序会抛出异常。

第 4 步：展示完整业务示例或进阶处理

前面保存的是候选股票池，下面进一步记录真正具备计算条件的股票。

假设研究需要至少 20 个有效收盘价，并使用首尾有效价格计算一个简单的区间价格收益率。

由于不同数据接口的字段结构可能存在差异，以下代码要求先确认实际返回的收盘价字段。如果不是close，需要根据实际响应调整映射。

importmath MIN_OBSERVATIONS=20defvalidate_price_data(df):required={"close"}missing=required-set(df.columns)ifmissing:raiseValueError(f"缺少字段：{missing}")prices=pd.to_numeric(df["close"],errors="coerce",).dropna()iflen(prices)<MIN_OBSERVATIONS:raiseValueError("有效收盘价数量不足")ifnotprices.map(lambdavalue:math.isfinite(value)andvalue>0).all():raiseValueError("收盘价存在非法数值")# 使用前必须确认数据已按真实交易时间升序排列first_price=float(prices.iloc[0])last_price=float(prices.iloc[-1])returnlast_price/first_price-1calculated_symbols=[]research_results={}calculation_errors={}forsymbol,dfinmarket_data.items():try:result=validate_price_data(df)calculated_symbols.append(symbol)research_results[symbol]=resultexceptExceptionasexc:calculation_errors[symbol]=str(exc)snapshot["calculated_symbols"]=sorted(calculated_symbols)snapshot["calculation_errors"]=calculation_errors snapshot["result_count"]=len(research_results)# 更新之前创建的快照：这里只演示单进程写入withsnapshot_file.open("w",encoding="utf-8")asfile:json.dump(snapshot,file,ensure_ascii=False,indent=2,)print("实际参与计算的股票：")print(snapshot["calculated_symbols"])

这段代码展示了候选集合、获取成功集合和计算成功集合之间的区别。

需要额外强调两个边界。

首先，首尾价格的选择依赖 K 线排序。正式使用时，应根据实际返回结构确认交易日期或时间戳字段，并按真实时间升序排序；不能默认接口返回顺序永远固定。

其次，这里计算的是前复权价格序列上的区间价格比值，不是包含现金分红、交易费用和税费的真实账户收益。

在生产环境中，建议将快照拆分为候选标的记录、有效标的记录和数据质量记录，并采用原子写入或事务机制，避免研究中断时留下不完整的最终记录。

关键坑与注意事项

  • 候选股票池不等于实际研究股票池。原始股票列表只能说明计划研究哪些股票，必须单独记录行情获取成功以及最终参与计算的集合。
  • 股票代码需要统一格式。例如600519.SH和其他缺少交易所后缀的写法不应直接混用。标准化时也不能随意猜测交易所。
  • 行情请求成功不代表数据完整。即使 SDK 返回非空 DataFrame，也应验证价格字段、日期范围、记录数量和非法数值。
  • 研究快照必须包含数据参数。单纯保存股票代码还不够，至少应记录周期、请求条数、复权方式和研究时间。
  • 哈希只能比较其覆盖的内容。股票池哈希相同，只代表标准化后的代码集合一致，无法证明行情数据没有发生修订。
  • 快照的修改应受控。示例演示了研究期间更新记录的方式。正式系统更适合使用状态流转和不可变的最终归档版本。
  • 历史股票池可能存在未来信息问题。如果用于历史回测，股票池应按当时可获得的信息构建，而不是直接使用今天的股票名单。

常见问题（FAQ）

Q：保存一份股票代码列表，不就能知道研究了哪些股票吗？

A：只能知道预设的研究范围。若某些股票行情获取失败或未通过数据检查，还需要单独记录实际参与计算的集合。

Q：股票池哈希值能否代替完整的股票列表？

A：不能。哈希适合快速比较集合是否一致，但完整股票列表仍应保存，方便审计和人工排查。

Q：QuantDash SDK 会自动生成研究快照吗？

A：本文没有使用此类 SDK 功能。示例通过 QuantDash 获取行情，再由 Python 代码自行生成和保存研究快照。

Q：两次研究股票池相同，为什么结果仍可能不同？

A：可能存在行情数据修订、复权处理变化、因子参数变化或者研究程序版本变化。股票池一致只是可复现的一个条件。

Q：实际研究股票池应该在什么时候保存？

A：建议研究开始时保存候选集合，数据验证后记录有效集合，研究完成后再生成最终归档记录。

小结

量化研究中的股票池不应该只存在于代码变量里。只要研究需要持续运行、重复验证或多人协作，就需要明确区分计划研究的股票、成功获取数据的股票以及最终参与计算的股票。

通过 QuantDash Python SDK，可以按指定标的和行情参数获取研究数据，减少在不同研究脚本中重复编写行情接入逻辑。不过，标的快照、计算资格检查、研究任务关联以及最终归档仍然是研究系统需要自行实现的工程能力。

把这些记录保存下来，研究结果才有清晰的输入边界，也更容易排查不同研究批次之间的差异。

参考

QuantDash Python SDK 文档：https://docs.quantdash.net/zh-Hans/sdk/python-quickstart