根据您提供的配网设备故障分类表和需求，我可以帮您设计一个程序来自动判断遥测数据是否存在故障。以下是一个基于阈值检测的方案，它能够检测常见的配网故障类型并返回1(正常)或0(异常)。
import numpy as np
import pandas as pd
def detect_power_grid_fault(telemetry_data):
"""
根据遥测数据检测配网设备是否存在故障
参数:
telemetry_data: 包含电流、电压等数据的字典，格式如下:
{
'phase_a_current': float, # A相电流值
'phase_b_current': float, # B相电流值
'phase_c_current': float, # C相电流值
'phase_a_voltage': float, # A相电压值
'phase_b_voltage': float, # B相电压值
'phase_c_voltage': float, # C相电压值
'zero_sequence_current': float, # 零序电流
'nominal_current': float, # 额定电流
'nominal_voltage': float, # 额定电压
}
返回:
int: 1表示正常，0表示异常
"""
# 从遥测数据中提取关键值
i_a = telemetry_data['phase_a_current']
i_b = telemetry_data['phase_b_current']
i_c = telemetry_data['phase_c_current']
v_a = telemetry_data['phase_a_voltage']
v_b = telemetry_data['phase_b_voltage']
v_c = telemetry_data['phase_c_voltage']
i_0 = telemetry_data.get('zero_sequence_current', 0)
i_nom = telemetry_data['nominal_current']
v_nom = telemetry_data['nominal_voltage']
# 定义阈值常数
CURRENT_THRESHOLD = 1.5 # 电流超过额定值的倍数
VOLTAGE_DROP_THRESHOLD = 0.8 # 电压低于额定值的比例
VOLTAGE_IMBALANCE_THRESHOLD = 0.1 # 电压不平衡阈值
ZERO_SEQUENCE_THRESHOLD = 0.1 * i_nom # 零序电流阈值
# 故障检测逻辑
# 1. 三相短路检测
three_phase_short = (i_a > CURRENT_THRESHOLD * i_nom and
i_b > CURRENT_THRESHOLD * i_nom and
i_c > CURRENT_THRESHOLD * i_nom and
v_a < 0.2 * v_nom and v_b < 0.2 * v_nom and v_c < 0.2 * v_nom)
# 2. 两相短路检测
two_phase_short_ab = (i_a > CURRENT_THRESHOLD * i_nom and
i_b > CURRENT_THRESHOLD * i_nom and
abs(v_a - v_b) < 0.5 * v_nom)
two_phase_short_bc = (i_b > CURRENT_THRESHOLD * i_nom and
i_c > CURRENT_THRESHOLD * i_nom and
abs(v_b - v_c) < 0.5 * v_nom)
two_phase_short_ca = (i_c > CURRENT_THRESHOLD * i_nom and
i_a > CURRENT_THRESHOLD * i_nom and
abs(v_c - v_a) < 0.5 * v_nom)
two_phase_short = two_phase_short_ab or two_phase_short_bc or two_phase_short_ca
# 3. 单相接地故障检测
single_phase_ground_a = (i_0 > ZERO_SEQUENCE_THRESHOLD and
v_a < VOLTAGE_DROP_THRESHOLD * v_nom and
v_b > v_nom and v_c > v_nom)
single_phase_ground_b = (i_0 > ZERO_SEQUENCE_THRESHOLD and
v_b < VOLTAGE_DROP_THRESHOLD * v_nom and
v_a > v_nom and v_c > v_nom)
single_phase_ground_c = (i_0 > ZERO_SEQUENCE_THRESHOLD and
v_c < VOLTAGE_DROP_THRESHOLD * v_nom and
v_a > v_nom and v_b > v_nom)
single_phase_ground = single_phase_ground_a or single_phase_ground_b or single_phase_ground_c
# 4. 失压故障检测
voltage_loss = (v_a < 0.1 * v_nom and v_b < 0.1 * v_nom and v_c < 0.1 * v_nom)
# 5. 电压不平衡故障检测
avg_voltage = (v_a + v_b + v_c) / 3
voltage_imbalance = (max(abs(v_a - avg_voltage),
abs(v_b - avg_voltage),
abs(v_c - avg_voltage)) / avg_voltage > VOLTAGE_IMBALANCE_THRESHOLD)
# 6. 过载故障检测
overload = (i_a > CURRENT_THRESHOLD * i_nom or
i_b > CURRENT_THRESHOLD * i_nom or
i_c > CURRENT_THRESHOLD * i_nom)
# 综合判断是否存在故障
is_fault = (three_phase_short or two_phase_short or
single_phase_ground or voltage_loss or
voltage_imbalance or overload)
# 返回结果：1表示正常，0表示异常
return 0 if is_fault else 1
# 测试示例
if __name__ == "__main__":
# 正常数据示例
normal_data = {
'phase_a_current': 80,
'phase_b_current': 82,
'phase_c_current': 81,
'phase_a_voltage': 10100,
'phase_b_voltage': 10050,
'phase_c_voltage': 10080,
'zero_sequence_current': 0.5,
'nominal_current': 100,
'nominal_voltage': 10000
}
# 两相短路故障示例
fault_data = {
'phase_a_current': 200, # 电流明显增大
'phase_b_current': 190, # 电流明显增大
'phase_c_current': 80,
'phase_a_voltage': 8000, # 相间电压降低
'phase_b_voltage': 8100, # 相间电压降低
'phase_c_voltage': 10100,
'zero_sequence_current': 5,
'nominal_current': 100,
'nominal_voltage': 10000
}
print(f"正常数据检测结果: {detect_power_grid_fault(normal_data)}") # 应该输出1
print(f"故障数据检测结果: {detect_power_grid_fault(fault_data)}") # 应该输出0
如果您想要采用机器学习方法，以下是一个基于决策树的示例，它可以通过历史数据学习故障模式：
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
def train_fault_detector():
"""
训练配网故障检测模型
"""
# 此处应该是读取历史遥测数据和对应的故障标签
# 为了示例，我们创建一些模拟数据
# 实际应用中需要替换为真实的历史数据
# 创建模拟数据
np.random.seed(42)
n_samples = 1000
# 创建特征
data = {
'phase_a_current': np.random.normal(100, 20, n_samples),
'phase_b_current': np.random.normal(100, 20, n_samples),
'phase_c_current': np.random.normal(100, 20, n_samples),
'phase_a_voltage': np.random.normal(10000, 500, n_samples),
'phase_b_voltage': np.random.normal(10000, 500, n_samples),
'phase_c_voltage': np.random.normal(10000, 500, n_samples),
'zero_sequence_current': np.random.normal(1, 0.5, n_samples),
}
# 计算一些派生特征
data['current_imbalance'] = np.max([
np.abs(data['phase_a_current'] - data['phase_b_current']),
np.abs(data['phase_b_current'] - data['phase_c_current']),
np.abs(data['phase_c_current'] - data['phase_a_current'])
], axis=0)
data['voltage_imbalance'] = np.max([
np.abs(data['phase_a_voltage'] - data['phase_b_voltage']),
np.abs(data['phase_b_voltage'] - data['phase_c_voltage']),
np.abs(data['phase_c_voltage'] - data['phase_a_voltage'])
], axis=0)
# 将数据转为DataFrame
df = pd.DataFrame(data)
# 模拟一些故障情况
# 两相短路
two_phase_short_indices = np.random.choice(n_samples, 100, replace=False)
for idx in two_phase_short_indices:
df.loc[idx, 'phase_a_current'] = np.random.normal(300, 30) # 增大电流
df.loc[idx, 'phase_b_current'] = np.random.normal(300, 30) # 增大电流
df.loc[idx, 'phase_a_voltage'] = np.random.normal(7000, 300) # 降低电压
df.loc[idx, 'phase_b_voltage'] = np.random.normal(7000, 300) # 降低电压
# 单相接地
single_phase_ground_indices = np.random.choice(
list(set(range(n_samples)) - set(two_phase_short_indices)),
100, replace=False
)
for idx in single_phase_ground_indices:
df.loc[idx, 'phase_a_voltage'] = np.random.normal(5000, 300) # 降低电压
df.loc[idx, 'zero_sequence_current'] = np.random.normal(20, 5) # 增大零序电流
# 创建标签（1正常，0异常）
labels = np.ones(n_samples, dtype=int)
labels[list(two_phase_short_indices) + list(single_phase_ground_indices)] = 0
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(df, labels, test_size=0.2, random_state=42)
# 训练随机森林模型
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
# 评估模型
y_pred = model.predict(X_test)
print("模型评估报告:")
print(classification_report(y_test, y_pred))
return model
def detect_fault_ml(model, telemetry_data):
"""
使用训练好的模型检测故障
参数:
model: 训练好的机器学习模型
telemetry_data: 包含电流、电压等数据的字典
返回:
int: 1表示正常，0表示异常
"""
# 计算派生特征
telemetry_data['current_imbalance'] = max([
abs(telemetry_data['phase_a_current'] - telemetry_data['phase_b_current']),
abs(telemetry_data['phase_b_current'] - telemetry_data['phase_c_current']),
abs(telemetry_data['phase_c_current'] - telemetry_data['phase_a_current'])
])
telemetry_data['voltage_imbalance'] = max([
abs(telemetry_data['phase_a_voltage'] - telemetry_data['phase_b_voltage']),
abs(telemetry_data['phase_b_voltage'] - telemetry_data['phase_c_voltage']),
abs(telemetry_data['phase_c_voltage'] - telemetry_data['phase_a_voltage'])
])
# 删除不需要的字段
if 'nominal_current' in telemetry_data:
del telemetry_data['nominal_current']
if 'nominal_voltage' in telemetry_data:
del telemetry_data['nominal_voltage']
# 将数据转为DataFrame
df = pd.DataFrame([telemetry_data])
# 使用模型预测
prediction = model.predict(df)[0]
return prediction
# 测试示例
if __name__ == "__main__":
# 训练模型
print("训练故障检测模型...")
model = train_fault_detector()
# 正常数据示例
normal_data = {
'phase_a_current': 95,
'phase_b_current': 98,
'phase_c_current': 100,
'phase_a_voltage': 10100,
'phase_b_voltage': 10050,
'phase_c_voltage': 10080,
'zero_sequence_current': 1.2,
}
# 两相短路故障示例
fault_data = {
'phase_a_current': 290, # 电流明显增大
'phase_b_current': 285, # 电流明显增大
'phase_c_current': 100,
'phase_a_voltage': 7100, # 相间电压降低
'phase_b_voltage': 7200, # 相间电压降低
'phase_c_voltage': 10100,
'zero_sequence_current': 3.5,
}
print(f"正常数据检测结果: {detect_fault_ml(model, normal_data)}") # 应该输出1
print(f"故障数据检测结果: {detect_fault_ml(model, fault_data)}") # 应该输出0
这两种方法各有优势：
基于阈值的方法：
基于机器学习的方法：
您可以根据实际情况选择适合的方法。如果有足够的历史故障数据，机器学习方法可能会有更好的表现；如果数据有限但有明确的故障判断规则，基于阈值的方法更为合适。