给出股票区间极值统计模块的设计：按业务选择复权口径，毫秒时间窗取日K并显式指定数量防截断，用盘中high/low定位极值与日期，批量计算与校验清单，附实跑结果。

原创

用户9138916

发布于 2026-11-01 14:01:30

1330

在研究和风控场景里，“某只股票在一段时间内的最高价、最低价及发生日期”是使用频率很高的基础统计：止损位设定、区间突破判断、回撤统计、报表生成都依赖它。这个需求看似简单，实际落地时常见三类错误：区间被静默截断、复权口径混用、用收盘价代替盘中极值。本文给出一个区间极值统计模块的设计与实现，示例使用 Python 与 pandas，于 2026-11-01 实跑，数据截至 2026-10-30 收盘。

一、需求拆解

输入	说明
symbol	股票代码，如 600519.SH
start / end	区间起止日期
adjust	复权口径

输出	说明
high / high_date	区间最高价及日期
low / low_date	区间最低价及日期
bars	区间内实际交易日数
amplitude	区间振幅

二、口径设计：三种问题对应三种复权

同一个“最高价”，在不同业务问题下应取不同口径：

业务问题	口径	理由
当时盘面真实成交到过多少	不复权	与当日行情一致，适合复盘、合规留痕
区间内低点到高点的涨幅	前复权	消除分红送转造成的价格断层
跨多年比较历史高点	后复权	早期价格不被压缩，便于长期对比

模块接口上把 adjust 作为必填参数，并在输出中回写所用口径，避免下游误用。

三、整体流程

 参数校验 ──▶ 日期转毫秒时间戳 ──▶ 按时间窗取日K(count上限)
                                        │
                                        ▼
            输出结果 ◀── 极值定位(high/low) ◀── 完整性检查(首尾日期/根数)

四、时间窗取数与防截断

行情接口通常以毫秒时间戳表示区间，并对单次返回数量设有默认值。示例数据源默认返回 100 根，上限 10000 根。如果区间超过 100 个交易日而未显式指定数量，结果会被截断且不报错——这是区间极值统计中最隐蔽的错误。

from datetime import datetime
from quantdash import QuantDash

qd = QuantDash()  # 凭证从环境变量 QUANTDASH_API_KEY 读取

def ms(date_str):
    return int(datetime.strptime(date_str, "%Y-%m-%d").timestamp() * 1000)

df = qd.klines.get("600519.SH", period="1d", adjust="none",
                   start_time=ms("2026-07-01"), end_time=ms("2026-10-31"),
                   count=10000, to_dataframe=True)
print(len(df), df["trade_date"].iloc[0], "->", df["trade_date"].iloc[-1])

82 2026-07-01 -> 2026-10-30

取数后立即检查首尾日期是否落在预期区间内，作为完整性校验的第一道关。end_time 取结束日次日零点，可确保结束日当天被包含。

五、极值定位

hi = df.loc[df["high"].idxmax()]
lo = df.loc[df["low"].idxmin()]
amp = (hi["high"] - lo["low"]) / lo["low"]
print(hi["high"], hi["trade_date"], lo["low"], lo["trade_date"], f"{amp:.2%}")

1393.37 2026-10-30 1166.33 2026-07-01 19.47%

设计要点：

最高价取 high 列、最低价取 low 列。若用收盘价，极值会被低估，对止损和突破类逻辑影响明显。
idxmax/idxmin 返回行索引，可同时得到价格与日期。
若区间内全部停牌，返回空表，需在定位前判断并返回空结果。

六、模块封装

def range_extreme(symbol, start, end, adjust="none"):
    k = qd.klines.get(symbol, period="1d", adjust=adjust,
                      start_time=ms(start), end_time=ms(end),
                      count=10000, to_dataframe=True)
    if k.empty:
        return {"symbol": symbol, "bars": 0, "adjust": adjust}
    h = k.loc[k["high"].idxmax()]
    l = k.loc[k["low"].idxmin()]
    return {"symbol": symbol, "adjust": adjust, "bars": len(k),
            "high": float(h["high"]), "high_date": h["trade_date"],
            "low": float(l["low"]), "low_date": l["trade_date"]}

print(range_extreme("000001.SZ", "2026-07-01", "2026-10-31"))

{'symbol': '000001.SZ', 'adjust': 'none', 'bars': 82, 'high': 12.08, 'high_date': '2026-09-03', 'low': 9.99, 'low_date': '2026-07-01'}

七、批量计算

股票池较大时，逐只请求既慢又容易触发频率限制。取数层改用批量接口，返回“代码 → DataFrame”的字典，计算逻辑不变：

bars = qd.klines.batch(["600519.SH", "000001.SZ"], period="1d",
                       count=120, adjust="none", to_dataframe=True)
for sym, x in bars.items():
    print(sym, x["high"].max(), x["low"].min())

600519.SH 1393.37 1151.01
000001.SZ 12.08 9.99

批量接口内部自动分批并发，计算本身为向量化操作，几千只股票的统计耗时主要在网络传输。

八、校验清单

校验项	方法
区间完整	首尾日期落在区间内；根数与交易日历大致相符
口径一致	输出回写 adjust，报表中标注
空结果	停牌或代码错误时返回 bars=0，不抛异常
盘中隔离	统计任务放在收盘后运行