Python实现能耗状态判定函数报错求助（异常检测项目）

我正在开展一项异常检测项目，已获取能耗预测值（predicted_values），希望基于预测值的标准差区间为实际能耗（powerconsumption）判定状态标签。尝试两段Python代码后均报错，请求协助排查错误并实现正确的判定函数。

第一段代码错误分析 #

代码内容 #

def status(x):
    if df_an['powerconsumption'] <= (df_an['predicted_values'] + (1*df_an['predicted_values'].std())):
        return "Normal"
    elif df_an['powerconsumption'] >= (df_an['predicted_values'] - (1*df_an['predicted_values'].std())):
        return "Normal"
    elif df_an['powerconsumption'] <= (df_an['predicted_values'] + (2*df_an['predicted_values'].std())):
        return "Above Normal"
    elif df_an['powerconsumption'] >= (df_an['predicted_values'] - (2*df_an['predicted_values'].std())):
        return "Above Normal"
    elif df_an['powerconsumption'] <= (df_an['predicted_values'] + (3*df_an['predicted_values'].std())):
        return "Normal"
    elif df_an['powerconsumption'] >= (df_an['predicted_values'] - (3*df_an['predicted_values'].std())):
        return "Normal"
    else:
        return "Anomalous"

for col in df_an.columns:
    df_an['status'] = df_an['powerconsumption'].apply(lambda x: status(x))

报错信息 #

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
C:\Users\Public\Documents\Wondershare\CreatorTemp/ipykernel_4112/2640136280.py in <module>
     16 
     17 for col in df_an.columns:
---&gt; 18     df_an['status'] = df_an['powerconsumption'].apply(lambda x: status(x))

c:\users\lenovo\anaconda3\envs\project\lib\site-packages\pandas\core\series.py in apply(self, func, convert_dtype, args, **kwargs)
   4355         dtype: float64
   4356         """
-&gt; 4357         return SeriesApply(self, func, convert_dtype, args, kwargs).apply()
   4358 
   4359     def _reduce(

c:\users\lenovo\anaconda3\envs\project\lib\site-packages\pandas\core\apply.py in apply(self)
   1041             return self.apply_str()
   1042 
-&gt; 1043         return self.apply_standard()
   1044 
   1045     def agg(self):

c:\users\lenovo\anaconda3\envs\project\lib\site-packages\pandas\core\apply.py in apply_standard(self)
   1100                     values,
   1101                     f,  # type: ignore[arg-type]
-&gt; 1102                     convert=self.convert_dtype,
   1103                 )
   1104 

c:\users\lenovo\anaconda3\envs\project\lib\site-packages\pandas\_libs\lib.pyx in pandas._libs.lib.map_infer()

C:\Users\Public\Documents\Wondershare\CreatorTemp/ipykernel_4112/2640136280.py in <lambda>(x)
     16 
     17 for col in df_an.columns:
---&gt; 18     df_an['status'] = df_an['powerconsumption'].apply(lambda x: status(x))

C:\Users\Public\Documents\Wondershare\CreatorTemp/ipykernel_4112/2640136280.py in status(x)
      1 def status(x):
----&gt; 2     if df_an['powerconsumption'] <= (df_an['predicted_values'] + (1*df_an['predicted_values'].std())):
      3         return "Normal"
      4     elif df_an['powerconsumption'] >= (df_an['predicted_values'] - (1*df_an['predicted_values'].std())):
      5         return "Normal"

c:\users\lenovo\anaconda3\envs\project\lib\site-packages\pandas\core\generic.py in __nonzero__(self)
   1536     def __nonzero__(self):
   1537         raise ValueError(
-&gt; 1538             f"The truth value of a {type(self).__name__} is ambiguous. "
   1539             "Use a.empty, a.bool(), a.item(), a.any() or a.all()."
   1540         )

ValueError: The truth value of a Series is ambiguous. Use a.empty, a.bool(), a.item(), a.any() or a.all().

错误原因 #

1. 函数逻辑错误：status函数直接操作整个DataFrame列，而apply是逐行传入单个值，每次判断都会生成布尔Series，Python无法直接判断整个Series的真假，触发ValueError。
2. 区间逻辑混乱：条件覆盖重叠且矛盾，前两个条件几乎覆盖所有数据，后续条件无法触发。
3. 循环冗余：遍历所有列赋值status列完全没必要，只需处理一次powerconsumption列即可。

第二段代码错误分析 #

代码内容 #

filter_method = lambda x: 
'Normal' if (df_an['powerconsumption'] <= (df_an['predicted_values'].mean() + (1*df_an['predicted_values'].std())))
else 'Normal' if df_an['powerconsumption'] >= df_an['predicted_values'].mean() - (1*df_an['predicted_values'].std())
else 'Above Normal' if df_an['powerconsumption'] <= df_an['predicted_values'].mean() + (2*df_an['predicted_values'].std())
else 'Above Normal' if df_an['powerconsumption'] >= df_an['predicted_values'].mean() - (2*df_an['predicted_values'].std())
else 'Nearing Anomalous' if df_an['powerconsumption'] <= df_an['predicted_values'].mean() + (3*df_an['predicted_values'].std())
else 'Nearing Anomalous' if df_an['powerconsumption'] >= df_an['predicted_values'].mean() - (3*df_an['predicted_values'].std())
else 'Anomalous'df_an['powerconsumption'] > df_an['predicted_values'].mean() + (3*df_an['predicted_values'].std())
else 'Anomalous'df_an['powerconsumption'] < df_an['predicted_values'].mean() - (3*df_an['predicted_values'].std())  

报错信息 #

File "C:\Users\Public\Documents\Wondershare\CreatorTemp/ipykernel_4112/3716694249.py", line 1
    filter_method = lambda x: 'Normal' if (df_an['powerconsumption'] > (df_an['predicted_values'].mean() + (1*df_an['predicted_values'].std())))
                                                                                                                                                ^
SyntaxError: invalid syntax

错误原因 #

1. 语法错误：
   • lambda表达式不能直接换行，需用括号或续行符。
   • 存在错误运算符=<、>=，正确应为<=、>=。
   • 最后两个else分支缺少逻辑连接，语法结构错误。
2. 逻辑错误：同样直接操作整个DataFrame列而非传入的单个值，区间条件重叠导致判定逻辑混乱。

正确实现方案 #

明确区间规则 #

基于需求整理合理的判定逻辑：

• Normal：实际能耗在预测值±1倍标准差范围内
• Above Normal：实际能耗在预测值±1~2倍标准差范围内
• Nearing Anomalous：实际能耗在预测值±2~3倍标准差范围内
• Anomalous：实际能耗超出预测值±3倍标准差范围

方法1：逐行处理（代码直观） #

# 预先计算预测值标准差，避免重复计算
pred_std = df_an['predicted_values'].std()

def get_status(row):
    actual = row['powerconsumption']
    pred = row['predicted_values']
    # 计算各区间边界
    lower_1, upper_1 = pred - pred_std, pred + pred_std
    lower_2, upper_2 = pred - 2*pred_std, pred + 2*pred_std
    lower_3, upper_3 = pred - 3*pred_std, pred + 3*pred_std
    
    if lower_1 <= actual <= upper_1:
        return "Normal"
    elif lower_2 <= actual <= upper_2:
        return "Above Normal"
    elif lower_3 <= actual <= upper_3:
        return "Nearing Anomalous"
    else:
        return "Anomalous"

# 对DataFrame逐行应用函数
df_an['status'] = df_an.apply(get_status, axis=1)

方法2：向量化操作（高效处理大数据） #

利用pandas和numpy的内置函数，避免循环开销：

import numpy as np

pred_std = df_an['predicted_values'].std()

# 计算所有行的区间边界
df_an['lower_1'] = df_an['predicted_values'] - pred_std
df_an['upper_1'] = df_an['predicted_values'] + pred_std
df_an['lower_2'] = df_an['predicted_values'] - 2*pred_std
df_an['upper_2'] = df_an['predicted_values'] + 2*pred_std
df_an['lower_3'] = df_an['predicted_values'] - 3*pred_std
df_an['upper_3'] = df_an['predicted_values'] + 3*pred_std

# 批量判定状态
conditions = [
    df_an['powerconsumption'].between(df_an['lower_1'], df_an['upper_1']),
    df_an['powerconsumption'].between(df_an['lower_2'], df_an['upper_2']),
    df_an['powerconsumption'].between(df_an['lower_3'], df_an['upper_3'])
]
choices = ["Normal", "Above Normal", "Nearing Anomalous"]

df_an['status'] = np.select(conditions, choices, default="Anomalous")

# 可选：删除中间计算的边界列
df_an.drop(['lower_1','upper_1','lower_2','upper_2','lower_3','upper_3'], axis=1, inplace=True)

关键优化点 #

1. 预先计算标准差，减少重复计算提升效率。
2. 逐行处理时使用当前行的实际值和预测值，避免操作整个列。
3. 区间判断采用连续范围，消除逻辑重叠。
4. 向量化操作利用内置函数，处理速度远快于apply。

内容的提问来源于stack exchange，提问作者Joseph Badana Rivera