一句话回答
当量化研究需要指定截止日期时,不能只依靠count获取最近若干条行情,而应通过end_time限定查询范围,并在数据返回后使用真实交易日期再次校验。QuantDash Python SDK 的qd.klines.get()支持按时间范围获取历史 K 线,研究代码可以在此基础上增加日期过滤和边界检查,避免截止日期之后的数据进入因子计算和历史回测。
为什么会有这个问题 / 传统做法有什么问题?
假设我们正在研究一只股票在 2025 年 6 月 30 日之前的价格变化,准备根据最近 20 个交易日的收盘价计算动量指标。
最直接的写法可能是:
df=qd.klines.get("600519.SH",period="1d",count=20,to_dataframe=True,)这段代码能获取历史日 K 线,但存在一个容易忽略的问题:count=20只指定返回数量,并没有明确研究需要的截止日期。
如果在更晚的时间运行相同代码,获取的数据可能已经包含 2025 年 7 月甚至之后的行情。
于是,我们原本想计算 6 月 30 日的历史动量,却可能使用了未来交易日的收盘价。
这属于典型的未来数据泄漏。
例如,20 日价格动量可以定义为:
[
M_{20}(t)=\frac{P_t}{P_{t-20}}-1
]
其中:
- (P_t) 表示截止研究日期的收盘价。
- (P_{t-20}) 表示之前第 20 个交易日的收盘价。
- (M_{20}(t)) 表示对应区间的价格变化率。
如果 (P_t) 实际来自研究日期之后,那么计算公式虽然正确,研究的时间边界却已经被破坏。
传统处理方式通常有两个问题。
第一个是只使用count,没有给请求设置历史截止时间。
第二个是完全相信接口返回结果,没有验证实际交易日期。
尤其在交易日期、自然日期和时间戳混用的系统中,时间边界很容易出现偏差。
更稳妥的方案是把数据获取和日期校验拆成两层:先通过标准化行情接口限制查询时间,再由研究代码检查最终参与计算的数据。
分步骤解决:使用 QuantDash 进行实践
第 1 步:安装 SDK 并初始化客户端
首先安装 QuantDash Python SDK:
# 在终端执行# pip install quantdash实际安装命令为pip install quantdash。
初始化客户端时,建议从环境变量读取 API Key,而不是直接写在源代码中。
importosimportpandasaspdfromdatetimeimportdatetime,date,timedeltafromzoneinfoimportZoneInfofromquantdashimportQuantDash api_key=os.getenv("QUANTDASH_API_KEY")ifnotapi_key:raiseRuntimeError("请先设置 QUANTDASH_API_KEY 环境变量")qd=QuantDash(api_key=api_key)这样可以避免 API Key 被误提交到代码仓库,也方便不同研究环境使用独立配置。
第 2 步:获取指定截止日期之前的数据
假设我们要获取 2025 年 6 月 30 日及之前的日 K 线。
QuantDash 的qd.klines.get()支持end_time参数,可以传入毫秒时间戳。
这里需要特别注意:如果研究截止条件是“包含 6 月 30 日整个交易日”,那么可以构造下一个自然日零点作为查询上界,再由本地代码执行包含截止日的过滤。
这样无需猜测服务端对于边界时间戳的包含规则。
symbol="600519.SH"cutoff_date=date(2025,6,30)market_tz=ZoneInfo("Asia/Shanghai")next_day=cutoff_date+timedelta(days=1)end_datetime=datetime(next_day.year,next_day.month,next_day.day,tzinfo=market_tz,)end_timestamp=int(end_datetime.timestamp()*1000)try:df=qd.klines.get(symbol,period="1d",count=100,end_time=end_timestamp,adjust="none",to_dataframe=True,)exceptExceptionasexc:raiseRuntimeError(f"获取{symbol}历史行情失败")fromexcifnotisinstance(df,pd.DataFrame):raiseTypeError("行情结果不是 DataFrame")ifdf.empty:raiseValueError("指定时间范围没有返回行情")required_columns={"trade_date","close"}missing_columns=required_columns-set(df.columns)ifmissing_columns:raiseValueError(f"行情缺少必要字段:{missing_columns}")这段代码中的几个参数需要明确:
- period=“1d”:获取日 K 线。
- count=100:最多请求一定数量的历史记录,实际返回条数仍以接口为准。
- end_time:指定查询结束时间。
- adjust=“none”:使用不复权价格,方便先检查原始价格日期。
- to_dataframe=True:返回便于后续处理的 DataFrame。
当前文档中的日 K 线示例使用trade_date字段。实际接入时仍需检查字段是否存在,以及返回值能否正确解析为日期。
另外,count并不等于交易日覆盖保证。如果需要严格覆盖固定研究窗口,应进一步检查最早交易日期是否足够。
第 3 步:根据真实交易日期执行二次过滤
即使接口已经设置end_time,研究代码仍然应执行一次日期校验。
原因很简单:查询参数是请求约束,而参与研究的数据必须以实际返回内容为准。
data=df.copy()data["trade_date"]=pd.to_datetime(data["trade_date"],errors="coerce",)ifdata["trade_date"].isna().any():raiseValueError("存在无法解析的交易日期")data["close"]=pd.to_numeric(data["close"],errors="coerce",)ifdata["close"].isna().any():raiseValueError("存在无效收盘价")if(data["close"]<=0).any():raiseValueError("收盘价必须大于零")data=data.loc[data["trade_date"].dt.date<=cutoff_date].copy()ifdata.empty:raiseValueError("截止日期之前没有有效行情")data=data.sort_values("trade_date")ifdata["trade_date"].duplicated().any():raiseValueError("存在重复交易日期")assert(data["trade_date"].dt.date.max()<=cutoff_date),"检测到超过截止日期的数据"print(data[["trade_date","close"]].tail())这里实际做了四件事。
首先,将交易日期转换为统一类型,避免字符串比较带来的错误。
其次,检查收盘价是否能够正常转换为数值。
然后,按照指定截止日期过滤数据。
最后,使用断言验证最终数据的最大交易日期。
需要强调的是,本地过滤是最后一道数据边界保护,不应替代服务端查询范围设置。
两者结合,才能同时控制请求内容和实际计算输入。
第 4 步:计算截止日期之前的动量指标
完成时间过滤后,就可以计算研究指标了。
假设我们需要计算 20 个交易日的价格动量。
这需要至少 21 个有效交易日收盘价,因为要比较当前收盘价与向前第 20 个交易日的收盘价。
lookback=20iflen(data)<lookback+1:raiseValueError(f"计算{lookback}日动量至少需要 "f"{lookback+1}条行情")current_close=float(data["close"].iloc[-1])previous_close=float(data["close"].iloc[-lookback-1])ifprevious_close<=0:raiseValueError("历史基准价格无效")momentum=current_close/previous_close-1latest_trade_date=(data["trade_date"].iloc[-1].date())print("研究截止日期:",cutoff_date)print("实际最后交易日:",latest_trade_date)print("20日价格动量:",momentum)这里还存在一个工程细节:研究截止日期不一定是实际交易日。
如果截止日期为节假日或休市日,那么最后一条行情可能来自更早的交易日。
这不一定是错误,但需要在研究结果中保留实际最后交易日期。
另外,这段代码使用不复权价格计算价格变化率,遇到分红、拆股等公司行动时,结果可能产生不连续变化。
如果研究目的是跨公司行动比较价格趋势,可以根据策略定义选择相应复权口径,但仍需评估复权基准是否包含研究截止日之后的信息。
关键坑与注意事项
把 count 当成截止日期控制。count主要约束返回数量,不负责表达历史研究时点。需要指定历史边界时,应明确设置end_time。
截止日与查询时间边界混淆。日 K 线研究通常按交易日期判断,而 API 查询使用毫秒时间戳。建议先明确是否包含截止日,再转换为查询上界。
字符串日期直接比较。不同日期字符串格式可能导致错误判断,应先通过pd.to_datetime()转换,再进行日期比较。
忽略非交易日。截止日期当天没有行情,不意味着数据缺失。应结合交易日历或实际交易日期判断,但不能简单假定每个自然日都有 K 线。
只校验时间,不校验价格。即使数据没有超过截止日期,收盘价为空、非数值或非正数仍然会影响计算,应执行必要的数据质量检查。
复权数据的历史一致性。控制 K 线日期不代表复权结果天然符合历史信息可得性要求。涉及公司行动时,还需要验证复权因子的时点与基准。
缺乏研究参数记录。建议保存截止日期、实际最后交易日、复权方式、SDK 版本和数据获取时间,方便以后复现研究。
常见问题(FAQ)
Q:设置 end_time 后,还需要在 pandas 中再次过滤吗?
A:建议保留。end_time用于约束接口查询,DataFrame 过滤用于确认实际参与计算的数据符合研究条件。两者解决的是不同层面的问题。
Q:为什么不直接使用 count=20?
A:因为count=20并不指定历史研究截止日。如果研究日期固定,单独使用count容易在不同执行日期得到不同的数据窗口。
Q:截止日期是周末怎么办?
A:对于日 K 线研究,通常使用截止日期之前最近一个有效交易日的数据,并记录这个实际交易日。如果业务要求截止日期必须为交易日,则应额外验证交易日历。
Q:使用前复权价格后,就能避免未来函数吗?
A:不能。前复权解决的是价格序列的调整口径问题,未来函数涉及历史时点的信息可得性。两者需要分别控制。
Q:QuantDash SDK 会自动校验研究截止日期吗?
A:SDK 提供时间范围查询能力。示例中的日期解析、数据过滤、重复日期检查和研究有效性判断属于开发者在业务层实现的逻辑,不应认为这些检查由 SDK 自动完成。
小结
历史研究最容易忽略的问题之一,就是代码虽然使用了历史 K 线,但实际获取的数据不一定停留在指定研究日期。尤其是通过最近 N 条记录计算指标时,如果没有明确截止时间,就可能把未来行情带入历史研究。
QuantDash Python SDK 可以通过qd.klines.get()和end_time简化历史行情接入。研究系统则需要继续完成日期校验、数据过滤、字段质量检查、复权口径确认以及研究参数记录。
将“请求时间边界”和“实际数据边界”同时纳入校验流程,是提高量化研究可复现性的重要基础。
参考
QuantDash Python SDK 文档:https://docs.quantdash.net/zh-Hans/sdk/python-quickstart