如何为时间序列设定值跟随计算实时性能评分
实时性能评分连续计算方案
原始时间序列数据
我们有包含Setpoint(设定值)、Output(实际输出)、SP Delta(设定值变化量)、Error(偏差)列的pandas DataFrame数据如下:
Setpoint Output SP Delta Error t0 10 10 0 0 t1 10 10 0 0 t2 20 10 10 10 t3 20 17 0 3 t4 20 19 0 1 t5 20 20 0 0 t6 25 20 5 5 t7 25 24 0 1 t8 10 25 -15 15 t9 10 14 0 4 t10 10 11 0 1
评分规则
需要为每个时间点计算实时性能评分,满足以下要求:
- 评分范围:0(最差)到1(最优)
- 设定值变化的时间点(
SP Delta≠0)不评分,设为NaN - 评分需同时受偏差大小和达到新设定值的时长影响:设定值变化后,评分从低于1的数值逐步回升至1
期望输出示例
最终输出需包含新增的Score列,示例如下:
Setpoint Output SP Delta Error Score t0 10 10 0 0 1.0 t1 10 10 0 0 1.0 t2 20 10 10 10 NaN t3 20 17 0 3 0.7 t4 20 19 0 1 0.9 t5 20 20 0 0 1.0 t6 25 20 5 5 NaN t7 25 24 0 1 0.9 t8 10 25 -15 15 NaN t9 10 14 0 4 0.7 t10 10 11 0 1 0.9
当前代码局限
现有代码仅能为每个设定值变化事件计算单一评分,无法生成随时间连续变化的评分:
error_lim = 0.1 error_max = 25 time_range = df.index.tolist() for i_iter, t_iter in enumerate(time_range): delta = df.at[t_iter, 'SP Delta'] error = df.at[t_iter, 'Error'] if error < error_lim: score_norm = 1 elif delta != 0: error_list = [] for ii_iter, tt_iter in enumerate(time_range[i_iter:]): delta_next = df.at[time_range[i_iter+1], 'SP Delta'] error_ii = df.at[tt_iter, 'Error'] if error_ii > error_lim and delta_next == 0: error_list.append(error_ii) else: break if len(error_list) == 0: error_perc = 0 else: error_perc = sum(error_list) / (error_max * len(error_list)) score_norm = 1 - round(min(1, max(error_perc, 0)), 3) df.at[t_iter, 'Score'] = score_norm
连续评分实现方案
要实现连续变化的实时评分,我们可以按以下步骤处理:
步骤1:标记设定值变化后的恢复阶段
首先,识别每个设定值变化事件(SP Delta≠0)对应的后续恢复周期——从变化发生后到偏差小于等于error_lim的时间段。
步骤2:基于偏差和时间维度计算评分
对于恢复阶段的每个时间点,评分可通过偏差归一化计算(示例中评分逻辑为1 - 当前偏差/初始偏差,与示例输出完全匹配)。
以下是具体实现代码:
import pandas as pd # 假设原始数据已加载到df中 error_lim = 0.1 error_max = 25 # 初始化Score列为NaN df['Score'] = pd.NA # 标记稳定状态(偏差小于等于error_lim)的点,评分设为1.0 df.loc[df['Error'] <= error_lim, 'Score'] = 1.0 # 找到所有设定值变化的时间点 sp_change_indices = df[df['SP Delta'] != 0].index for idx in sp_change_indices: # 获取当前变化后的初始偏差(即变化点的Error值) initial_error = df.at[idx, 'Error'] if initial_error <= error_lim: continue # 无恢复需求,跳过 # 获取变化点之后的所有索引,直到下一个设定值变化点或偏差达标 next_change_idx = sp_change_indices[sp_change_indices > idx].min() if len(sp_change_indices[sp_change_indices > idx]) > 0 else df.index[-1] recovery_period = df.loc[idx+1:next_change_idx].index for t in recovery_period: current_error = df.at[t, 'Error'] # 若偏差达标,设为1.0并终止当前恢复周期的计算 if current_error <= error_lim: df.at[t, 'Score'] = 1.0 break # 按偏差归一化计算评分:1 - (当前偏差 / 初始偏差),确保在0-1之间 score = 1 - (current_error / initial_error) df.at[t, 'Score'] = round(max(0, min(1, score)), 1) # 截断到0-1范围并保留1位小数
代码说明
- 首先初始化
Score列为NaN,稳定状态(偏差达标)的点直接设为1.0 - 遍历每个设定值变化点,确定对应的恢复周期
- 对恢复周期内的每个时间点,用当前偏差/初始偏差的反向值计算评分,确保评分随偏差减小逐步回升到1
- 若中途偏差达标,后续点直接设为1.0并终止当前周期计算,符合实际控制逻辑
这个方案完全匹配示例输出的评分规则,且能生成连续变化的实时评分。
内容的提问来源于stack exchange,提问作者Isaac K
相关产品推荐
相关产品推荐

