如何用Pandas/Python统计无中断连续失败测试次数及后续状态

我正在练习Pandas技能，目前卡在一道练习题上：
我创建了一个包含自行车测试数据的DataFrame，测试按test_id升序（时间顺序）排列。需要实现两个需求：

1. 为每个bike+test_type组获取最近的失败测试（最大test_id），统计该组无中断的连续失败总次数；
2. 添加ends_with_pass列，判断该失败组之后是否出现了通过测试。

我已有以下代码，但不知道如何继续：

fail = data[data["test_result"] == "fail"]
max_fail_tests = fail.groupby(["bike", "test_type"])["test_id"].max().reset_index()
res = pd.merge(max_failed_tests, data, on=["bike", "test_type"])

res["fails_in_a_row"] = res.groupby(
    ["bike", "test_type"]
)["test_result"].apply(
    lambda x: (
        x.eq("fail") & x.shift().ne("pass")
    ).cumsum()
)

输入数据 #

期望输出 #

核心思路 #

你的现有代码没聚焦到「最近的连续失败序列」，且未处理后续是否有通过测试的判断。我们需要先按bike+test_type分组，划分连续的失败/通过序列块，再定位最后一个失败块的长度，同时对比组内最大测试ID判断后续是否有通过记录。

完整代码 #

import pandas as pd

# 构造输入数据（替换为你的原始DataFrame即可）
data = pd.DataFrame([
    [1, "a", "slow", "pass"],
    [1, "a", "fast", "pass"],
    [15, "c", "fast", "pass"],
    [15, "c", "slow", "pass"],
    [34, "b", "slow", "fail"],
    [34, "b", "fast", "fail"],
    [36, "a", "slow", "pass"],
    [36, "a", "fast", "pass"],
    [37, "c", "fast", "fail"],
    [37, "c", "slow", "fail"],
    [87, "c", "fast", "fail"],
    [87, "c", "slow", "fail"],
    [99, "b", "slow", "fail"],
    [99, "b", "fast", "fail"],
    [124, "b", "slow", "pass"],
    [124, "b", "fast", "pass"]
], columns=["test_id", "bike", "test_type", "test_result"])

# 1. 确保数据按时间顺序排列，为每个连续结果序列分配块ID
data = data.sort_values("test_id")
data["block_id"] = data.groupby(["bike", "test_type"])["test_result"].apply(
    lambda x: (x != x.shift()).cumsum()
).reset_index(drop=True)

# 2. 统计每个块的核心信息：结果类型、最后测试ID、块长度
block_stats = data.groupby(["bike", "test_type", "block_id"]).agg(
    result=("test_result", "first"),
    last_test_id=("test_id", "max"),
    length=("test_id", "count")
).reset_index()

# 3. 筛选每个bike+test_type组的最后一个失败块
last_fail_blocks = block_stats[block_stats["result"] == "fail"].groupby(["bike", "test_type"]).last().reset_index()

# 4. 判断失败块后是否有通过测试：对比组内最大测试ID和失败块最后测试ID
max_test_per_group = data.groupby(["bike", "test_type"])["test_id"].max().reset_index(name="max_test_id")
last_fail_blocks = last_fail_blocks.merge(max_test_per_group, on=["bike", "test_type"])
last_fail_blocks["ends_with_pass"] = last_fail_blocks.apply(
    lambda row: "yes" if row["max_test_id"] > row["last_test_id"] else "no",
    axis=1
)

# 5. 整理为期望输出格式
final_output = last_fail_blocks[["bike", "test_type", "length", "ends_with_pass"]].rename(columns={"length": "fails_in_a_row"})
print(final_output)

关键步骤解释 #

• 块ID分配：通过(x != x.shift()).cumsum()识别结果变化的节点，为每个连续相同结果的序列分配唯一ID，区分不同阶段的失败/通过周期。
• 块统计：提取每个块的结果类型、结束时间和长度，快速定位最后一个失败块的信息。
• 后续通过判断：如果组内最大test_id大于最后失败块的结束test_id，说明失败后出现了通过测试。

内容的提问来源于stack exchange，提问作者konichiwa