基于两个时序DataFrame的系统状态切换与连续小时统计问题

一、修复连续小时统计代码 #

原代码问题分析 #

• 错误按sum列分组：连续小时统计基于时间连续性，与sum值无关
• 变量名不一致：定义了fgroups但使用未定义的groups
• 逻辑偏差：连续小时分组应基于全局时间序列的连续性，而非按sum拆分后的局部判断

修复后代码（以df_A为例，df_B可复用相同逻辑） #

import pandas as pd
import datetime

# 示例DataFrame A
data = [[datetime.datetime(2021,6,7,13,0), 1], [datetime.datetime(2021,7,30,8,0), 1], [datetime.datetime(2022,4,10,1,0), 49],
    [datetime.datetime(2022,4,10,2,0), 81], [datetime.datetime(2022,4,10,3,0), 67], [datetime.datetime(2022,4,10,4,0), 93],
    [datetime.datetime(2022,5,10,3,0), 81], [datetime.datetime(2022,5,10,4,0), 1], [datetime.datetime(2022,7,10,1,0), 106]]
df_A = pd.DataFrame(data, columns=['col_0', 'sum'])

# 修复后的连续小时统计逻辑
# 1. 确保时间列排序（冗余处理保证鲁棒性）
df_A_sorted = df_A.sort_values('col_0').reset_index(drop=True)
# 2. 判断相邻时间是否间隔1小时，生成连续分组标识
df_A_sorted['is_break'] = df_A_sorted['col_0'].diff().ne(pd.Timedelta(hours=1))
df_A_sorted['group_key'] = df_A_sorted['is_break'].cumsum()
# 3. 分组统计连续小时数、起止时间
grouped = df_A_sorted.groupby('group_key')['col_0']
df_A_upd = df_A_sorted.assign(
    count=grouped.transform('size'),
    Period_start=grouped.transform('first'),
    Period_end=grouped.transform('last')
).drop(['is_break', 'group_key'], axis=1)

print(df_A_upd)

二、状态切换统计与重叠检测 #

完整实现代码 #

# 示例DataFrame B
data1 = [[datetime.datetime(2020,5,1,16,0), 7], [datetime.datetime(2020,5,1,17,0), 9], [datetime.datetime(2021,6,7,20,0), 8],
    [datetime.datetime(2021,7,30,15,0), 17], [datetime.datetime(2022,6,10,3,0), 32], [datetime.datetime(2022,8,10,13,0), 54],
    [datetime.datetime(2022,8,10,14,0), 19], [datetime.datetime(2022,8,10,15,0), 1], [datetime.datetime(2022,8,10,16,0), 3]]
df_B = pd.DataFrame(data1, columns=['col_0', 'sum'])

# 1. 合并数据并标记来源
df_A['frame'] = 'df_A'
df_B['frame'] = 'df_B'
combined = pd.concat([df_A, df_B]).sort_values('col_0').reset_index(drop=True)

# 2. 检测同一小时重叠事件
hourly_groups = combined.groupby(combined['col_0'].dt.floor('H'))['frame'].unique()
overlap_hours = hourly_groups[hourly_groups.apply(lambda x: len(x) == 2)].index.tolist()
if overlap_hours:
    print("检测到重叠小时：")
    for hour in overlap_hours:
        print(f"- {hour}")
else:
    print("未检测到同一小时的重叠事件")

# 3. 生成状态切换统计结果
combined['state'] = combined['frame'].map({'df_A':0, 'df_B':1})
# 标记状态切换点
combined['is_switch'] = combined['state'].ne(combined['state'].shift())
# 提取每个状态段的起始记录
result = combined[combined['is_switch'] | (combined.index == 0)].reset_index(drop=True)
# 整理输出格式
result = result[['frame', 'state', 'col_0']].rename(columns={'col_0':'from'})
result['change_of_state'] = range(len(result))

# 输出最终结果
print("\n状态切换统计结果：")
print(result)

输出说明 #

• 重叠检测：会列出所有同时存在df_A和df_B事件的小时（示例数据无重叠，输出对应提示）
• 状态切换结果：与需求示例格式一致，包含数据源、状态、起始时间、切换次数

内容的提问来源于stack exchange，提问作者Miss.Pepper