本文介绍一种基于自定义逻辑的高效方法，用于在多产品价格时间序列中识别跨周末的连续价格不变区间，并输出每个稳定区间的起止日期、价格及持续天数。
本文介绍一种基于自定义逻辑的高效方法，用于在多产品价格时间序列中识别跨周末的连续价格不变区间，并输出每个稳定区间的起止日期、价格及持续天数。
在金融、零售或供应链分析中，常需识别价格“冻结期”——即同一产品价格在连续工作日（含跨周末的自然日逻辑连续性，如周五→周一视为连续）中保持不变的时段。标准的 shift() + cumsum() 方法无法直接满足该需求，因其默认按原始行序判断连续性，未考虑业务规则中的“工作日连续性”与“分组隔离性”。
核心挑战有三点：
- ✅ 工作日连续性：2024-02-01（周四）→ 2024-02-04（周一）应视为连续2天（跳过周六、周日）；
- ✅ 组内独立性：仅当同一 ID 下价格相同且日期逻辑连续时才合并；
- ✅ 严格区间划分：相同价格在不同时间段重复出现（如 ID=1 的 1.30 出现在 2月6–7日，而非2月1–4日）须作为独立区间处理。
为此，我们采用状态驱动的逐行遍历法，替代依赖索引的向量化操作，确保逻辑完全可控：
步骤说明
- 预处理日期：将 Date 列转为 datetime64[ns] 类型，便于日期运算；
- 定义工作日递进函数：get_next_day() 模拟真实交易日逻辑——若当前日为周五（.dayofweek == 4），则下一日为周一（+3天）；若为周六（5），则下一日为周一（+2天）；其他情况+1天；
-
分组生成器 generate_groups()：
- 维护 current_group 存储当前连续区间内的 (ID, Date, Price) 元组；
- 关键判断条件：expected_next_day != d or prev_id != i or prev_price != p —— 任一不满足即触发新组；
- 每组至少含2条记录才输出（单日不构成“重复”）；
- 结果组装：用 pd.DataFrame() 直接接收生成器输出，自动构建目标结构。
完整可运行代码
import pandas as pd
# 示例数据（注意：Price 使用字符串格式以匹配原始示例中的逗号小数点）
df = pd.DataFrame({
"ID": [1,1,1,1,1,2,2,2,2,2],
"Date": ["2024-02-01", "2024-02-04", "2024-02-05", "2024-02-06", "2024-02-07",
"2024-02-01", "2024-02-04", "2024-02-05", "2024-02-06", "2024-02-07"],
"Price": ["1,00", "1,00", "1,20", "1,30", "1,30",
"1,30", "0,90", "0,90", "0,90", "1,30"]
})
df["Date"] = pd.to_datetime(df["Date"])
def get_next_day(d):
d += pd.Timedelta("1 day")
if d.dayofweek == 4: # Friday → Monday (+3)
d += pd.Timedelta("2 days")
elif d.dayofweek == 5: # Saturday → Monday (+2)
d += pd.Timedelta("1 day")
return d
def yield_group(g):
if len(g) > 1:
yield {
"ID": g[0][0],
"start_date": g[0][1],
"end_date": g[-1][1],
"price": g[0][2],
"repeated_days_count": len(g),
}
def generate_groups(ids, dates, prices):
current_group = []
prev_id = None
prev_price = None
expected_next_day = None
for i, d, p in zip(ids, dates, prices):
if prev_id is None:
current_group.append((i, d, p))
elif expected_next_day != d or prev_id != i or prev_price != p:
yield from yield_group(current_group)
current_group = [(i, d, p)]
else:
current_group.append((i, d, p))
prev_id = i
prev_price = p
expected_next_day = get_next_day(d)
yield from yield_group(current_group)
# 执行并输出
result = pd.DataFrame(generate_groups(df["ID"], df["Date"], df["Price"]))
print(result)
注意事项与优化建议
- ? 性能考量：该方法为 O(n) 时间复杂度，对百万级记录仍高效；若数据量极大，可先按 ID 排序并分块处理；
- ? 价格精度：原始数据中 Price 为字符串（如 "1,00"），若后续需数值计算，请在输入前统一转换：df["Price"] = df["Price"].str.replace(",", ".").astype(float)；
- ? 扩展性：如需支持节假日（非仅周末），可将 get_next_day() 替换为基于 pandas.tseries.offsets.CustomBusinessDay 的逻辑；
- ? 边界验证：确保输入数据已按 ID 和 Date 升序排序，否则 generate_groups 将失效。
该方案兼顾准确性、可读性与业务适配性，是处理“逻辑连续但物理日期不连续”时间序列问题的稳健实践。
