如何生成符合时间逻辑的任务样本DataFrame？

需求梳理 #

要生成符合以下规则的任务时序样本DataFrame：

• 核心列：id（任务流ID）、step（任务节点，格式如TaskX_Start/TaskX_End）、timestamp（时间戳）
• 包含11个任务，其中Task2、Task3、Task7存在分支选择路径
• 基础时间规则：单个任务的End时间必须晚于Start时间；后续任务的Start时间必须晚于前序任务的End时间
• 特殊回退规则：支持从Task8回到Task7，此时需满足时间顺序：Task7_Start(回退版) > Task8_End > Task8_Start > Task7_End(初始版)

完整实现代码 #

import pandas as pd
from datetime import datetime, timedelta
import random

def generate_task_sequence(task_flow_id):
    # 初始化时间起点，用当前时间作为基准
    current_time = datetime.now()
    sequence = []
    
    # 通用函数：添加任务的Start和End时间
    def add_task(task_num, duration_min=random.randint(5, 30)):
        nonlocal current_time
        start_time = current_time
        end_time = start_time + timedelta(minutes=duration_min)
        sequence.append({"id": task_flow_id, "step": f"Task{task_num}_Start", "timestamp": start_time})
        sequence.append({"id": task_flow_id, "step": f"Task{task_num}_End", "timestamp": end_time})
        current_time = end_time
    
    # 任务1：无分支，直接执行
    add_task(1)
    
    # Task2：分支选择（分支A/分支B，对应不同后续路径）
    task2_branch = random.choice(["A", "B"])
    add_task(2)
    if task2_branch == "A":
        # 分支A：Task2后走Task4
        add_task(4)
    else:
        # 分支B：Task2后走Task5
        add_task(5)
    
    # Task3：分支选择（分支X/分支Y）
    task3_branch = random.choice(["X", "Y"])
    add_task(3)
    if task3_branch == "X":
        add_task(6)
    else:
        add_task(9)
    
    # Task7：分支选择，包含回退到自身的可能
    task7_branch = random.choice(["normal", "rollback"])
    # 先执行第一次Task7
    add_task(7)
    if task7_branch == "rollback":
        # 执行Task8，然后回退到Task7
        add_task(8)
        # 回退版Task7：Start时间必须晚于Task8的End
        rollback_start = current_time
        rollback_end = rollback_start + timedelta(minutes=random.randint(5, 30))
        sequence.append({"id": task_flow_id, "step": "Task7_Start", "timestamp": rollback_start})
        sequence.append({"id": task_flow_id, "step": "Task7_End", "timestamp": rollback_end})
        current_time = rollback_end
    
    # 剩余无分支任务：Task10、Task11
    add_task(10)
    add_task(11)
    
    return sequence

# 生成10条任务流样本
sample_data = []
for idx in range(1, 11):
    sample_data.extend(generate_task_sequence(idx))

# 转为DataFrame并格式化时间戳
df = pd.DataFrame(sample_data)
df["timestamp"] = df["timestamp"].dt.strftime("%Y-%m-%d %H:%M:%S")

# 查看结果
print(df.head(20))

关键逻辑说明 #

1. 时间戳生成：用datetime和timedelta控制时间流动，每个任务的Start时间继承前一个任务的End时间，确保时序合规
2. 分支处理：通过random.choice()模拟分支选择，Task2/Task3的分支对应不同后续任务，Task7的分支区分正常流程和回退流程
3. 回退场景实现：先执行第一次Task7→Task8，再生成回退版Task7，其Start时间直接沿用Task8的End时间，天然满足Task7_Start(回退) > Task8_End > Task8_Start > Task7_End(初始)的规则
4. 扩展性：可以通过调整duration_min的范围控制任务耗时，或新增分支选项丰富样本多样性

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