在工业设备智能运维领域，基于大模型实现断路器寿命预测是典型的时序+结构化数据微调场景。与通用NLP任务不同，设备寿命预测的核心是处理多维度时序监测数据+设备属性数据，数据准备需兼顾工业数据的专业性、时序特征的完整性和模型输入的适配性。本文以10kV高压断路器寿命预测为场景，完整讲解从样本数据设计、质量分析、清洗标准化到Tokenizer编码的全流程，通过可复现的代码案例拆解每一步技术细节。

一、需求背景与样本数据设计

1.1 场景定义

本次案例聚焦10kV高压断路器寿命预测，目标是微调一个适配工业时序数据的大模型（如TimeLLM、BERT-Industrial或基于LLaMA的时序适配模型），输入断路器运行过程中的多维度监测数据，输出剩余寿命等级（短/中/长）或剩余运行时长。

1.2 断路器寿命预测的核心数据维度

断路器寿命主要受以下核心指标影响，样本数据需完整覆盖：

数据类型	核心指标
时序监测数据	分合闸次数、累计运行时长(h)、触头温度(℃)、操作电压(V)、机械振动值(g)
设备属性数据	型号、生产厂家、投运年份、安装环境(室内/室外)、维护频次(次/年)
标签数据	剩余寿命等级(短：≤1年；中：1-5年；长：>5年)、实际剩余寿命(月)

1.3 样本数据生成（含脏数据）

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

# 1. 定义基础配置
# 断路器型号与厂家映射
model_manufacturer = {
    "ZW32-10": ["平高集团", "正泰电器", "德力西电气"],
    "VS1-10": ["许继集团", "施耐德", "ABB"],
    "ZN63-10": ["西门子", "华光科技", "泰开电气"]
}
# 安装环境与维护频次区间
env_maintain = {
    "室内": (2, 4),   # 维护频次：2-4次/年
    "室外": (4, 6)    # 维护频次：4-6次/年
}
# 寿命等级映射
life_level_map = {
    "短": (0, 12),    # 剩余寿命≤12个月
    "中": (13, 60),   # 剩余寿命13-60个月
    "长": (61, 120)   # 剩余寿命>60个月
}

# 2. 生成模拟样本数据（含脏数据）
def generate_breaker_life_data(num_samples=200):
    samples = []
    for idx in range(num_samples):
        # 基础属性
        model = random.choice(list(model_manufacturer.keys()))
        manufacturer = random.choice(model_manufacturer[model])
        install_env = random.choice(["室内", "室外"])
        maintain_freq = random.randint(*env_maintain[install_env])
        put_in_year = random.randint(2010, 2022)
        put_in_date = datetime(put_in_year, random.randint(1, 12), random.randint(1, 28)).strftime("%Y-%m-%d")
        
        # 时序监测数据（模拟5年运行数据，取均值）
        switch_times = random.randint(100, 5000)  # 分合闸次数
        total_run_hours = (datetime.now() - datetime.strptime(put_in_date, "%Y-%m-%d")).days * 24 * random.uniform(0.7, 0.95)  # 累计运行时长
        contact_temp = random.uniform(45, 85)     # 触头温度（均值）
        operate_voltage = random.uniform(10.0, 10.5)  # 操作电压（均值）
        vibration = random.uniform(0.1, 0.8)      # 机械振动值（均值）
        
        # 剩余寿命（月）与等级
        base_life = 120 - (2024 - put_in_year) * 12  # 基础寿命（10年）
        # 基于运行参数调整寿命（分合闸次数多、温度高则寿命短）
        life_adjust = - (switch_times / 100) - (contact_temp - 60) * 2 - vibration * 10
        remain_life = max(1, base_life + life_adjust)
        # 匹配寿命等级
        if remain_life <= life_level_map["短"][1]:
            life_level = "短"
        elif remain_life <= life_level_map["中"][1]:
            life_level = "中"
        else:
            life_level = "长"
        
        # 故意混入脏数据（10%比例）
        if idx % 10 == 0:
            # 重复数据
            if samples:
                switch_times = samples[-1]["switch_times"]
                contact_temp = samples[-1]["contact_temp"]
                remain_life = samples[-1]["remain_life"]
                life_level = samples[-1]["life_level"]
        if idx % 15 == 0:
            # 异常值（如温度超范围、电压为0）
            contact_temp = random.choice([150, -10])  # 异常温度
            operate_voltage = 0  # 异常电压
        if idx % 20 == 0:
            # 缺失值
            vibration = None
        if idx % 25 == 0:
            # 格式错误（日期格式混乱）
            put_in_date = "2020/13/32"  # 无效日期
        
        # 构造样本
        sample = {
            "device_id": f"CB{random.randint(1000, 9999)}",  # 设备ID
            "model": model,                  # 型号
            "manufacturer": manufacturer,    # 厂家
            "install_env": install_env,      # 安装环境
            "put_in_date": put_in_date,      # 投运日期
            "maintain_freq": maintain_freq,  # 维护频次（次/年）
            "switch_times": switch_times,    # 分合闸次数
            "total_run_hours": round(total_run_hours, 2),  # 累计运行时长(h)
            "contact_temp": round(contact_temp, 2) if contact_temp else None,  # 触头温度(℃)
            "operate_voltage": round(operate_voltage, 2) if operate_voltage else None,  # 操作电压(V)
            "vibration": round(vibration, 2) if vibration else None,  # 机械振动值(g)
            "remain_life": round(remain_life, 1),  # 剩余寿命(月)
            "life_level": life_level          # 寿命等级
        }
        samples.append(sample)
    
    # 保存为JSON和CSV（工业场景常用格式）
    with open("raw_breaker_life_data.json", "w", encoding="utf-8") as f:
        json.dump(samples, f, ensure_ascii=False, indent=2)
    df = pd.DataFrame(samples)
    df.to_csv("raw_breaker_life_data.csv", index=False, encoding="utf-8-sig")
    
    print(f"生成原始样本数据{len(samples)}条")
    return samples, df

# 执行数据生成
raw_samples, raw_df = generate_breaker_life_data(200)
print("前5条原始数据示例：")
print(raw_df.head().to_string())

1.4 高质量工业数据的核心特性

针对断路器寿命预测场景，高质量数据需满足：

1. 领域专业性：覆盖断路器寿命相关的核心工业指标（分合闸次数、触头温度等），指标取值符合工业实际范围（如操作电压10.0-10.5V）；
2. 时序完整性：关键监测指标无大面积缺失，数值无明显异常（如温度不会超过150℃）；
3. 标签准确性：剩余寿命等级与实际剩余寿命数值严格匹配，无标签错误；
4. 格式规范性：日期、数值格式统一，无无效字符或格式混乱；
5. 模型适配性：数据可转换为「文本化时序序列+结构化属性」的格式，适配大模型的输入要求。

1.5 模型输入要求

本次选择适配工业时序数据的TimeLLM（基于BERT改造）作为微调模型，其输入要求：

• 输入格式：将结构化/时序数据转换为自然语言描述的序列文本（如「设备CB1234，型号ZW32-10，分合闸次数1200次，触头温度65℃，剩余寿命等级：中」）；
• 文本长度：单条输入token数≤128；
• 数值标准化：所有数值型指标需归一化到[0,1]区间；
• 标签编码：寿命等级（短/中/长）转换为数字编码（0/1/2）；
• 无缺失/异常值：输入数据需完成缺失值填充、异常值修正。

二、数据清洗：剔除低质、异常、重复数据

工业数据清洗的核心是处理缺失值、异常值、重复值、格式错误，保留符合工业逻辑的有效数据。

2.1 清洗步骤与代码实现

import json
import pandas as pd
import numpy as np
from datetime import datetime
from collections import Counter

# 1. 加载原始数据
df = pd.read_csv("raw_breaker_life_data.csv", encoding="utf-8-sig")
print(f"原始数据条数：{len(df)}")
print(f"原始数据缺失值统计：\n{df.isnull().sum()}")

# 2. 定义清洗规则与函数
class BreakerDataCleaner:
    def __init__(self):
        # 工业指标正常范围
        self.normal_ranges = {
            "contact_temp": (40, 90),    # 触头温度正常范围：40-90℃
            "operate_voltage": (9.5, 11.0), # 操作电压正常范围：9.5-11.0V
            "vibration": (0.05, 1.0),    # 振动值正常范围：0.05-1.0g
            "switch_times": (0, 10000),  # 分合闸次数正常范围：0-10000
            "maintain_freq": (0, 10)     # 维护频次正常范围：0-10次/年
        }
    
    def validate_date(self, date_str):
        """验证日期格式是否有效"""
        try:
            datetime.strptime(date_str, "%Y-%m-%d")
            return True
        except:
            return False
    
    def fill_missing_values(self, df):
        """缺失值填充（工业数据适配策略）"""
        # 数值型字段：用同型号设备的均值填充
        numeric_cols = ["contact_temp", "operate_voltage", "vibration", "maintain_freq"]
        for col in numeric_cols:
            # 按型号分组计算均值
            model_means = df.groupby("model")[col].transform("mean")
            df[col] = df[col].fillna(model_means)
            # 若仍有缺失（如该型号无数据），用全局均值填充
            df[col] = df[col].fillna(df[col].mean())
        return df
    
    def correct_outliers(self, df):
        """异常值修正（替换为正常范围边界值）"""
        for col, (min_val, max_val) in self.normal_ranges.items():
            if col in df.columns:
                # 低于最小值→替换为最小值
                df.loc[df[col] < min_val, col] = min_val
                # 高于最大值→替换为最大值
                df.loc[df[col] > max_val, col] = max_val
        return df
    
    def remove_duplicates(self, df):
        """去重：基于设备ID+核心指标去重"""
        # 按设备ID分组，保留第一条有效数据
        df = df.drop_duplicates(subset=["device_id"], keep="first")
        return df
    
    def filter_invalid_date(self, df):
        """过滤日期格式错误的数据"""
        df["date_valid"] = df["put_in_date"].apply(self.validate_date)
        valid_df = df[df["date_valid"]].drop("date_valid", axis=1)
        invalid_count = len(df) - len(valid_df)
        print(f"过滤无效日期数据{invalid_count}条")
        return valid_df
    
    def clean(self, df):
        """完整清洗流程"""
        # 步骤1：过滤无效日期数据
        df = self.filter_invalid_date(df)
        # 步骤2：去重
        df = self.remove_duplicates(df)
        # 步骤3：缺失值填充
        df = self.fill_missing_values(df)
        # 步骤4：异常值修正
        df = self.correct_outliers(df)
        # 重置索引
        df = df.reset_index(drop=True)
        return df

# 3. 执行清洗
cleaner = BreakerDataCleaner()
cleaned_df = cleaner.clean(raw_df)

# 保存清洗后数据
cleaned_df.to_csv("cleaned_breaker_life_data.csv", index=False, encoding="utf-8-sig")
with open("cleaned_breaker_life_data.json", "w", encoding="utf-8") as f:
    json.dump(cleaned_df.to_dict("records"), f, ensure_ascii=False, indent=2)

print(f"清洗后数据条数：{len(cleaned_df)}")
print(f"清洗后缺失值统计：\n{cleaned_df.isnull().sum()}")
print("前5条清洗后数据示例：")
print(cleaned_df.head().to_string())

2.2 清洗步骤说明

1. 无效日期过滤：通过datetime.strptime验证投运日期格式，剔除如「2020/13/32」的无效日期数据；
2. 重复值去除：基于device_id去重（断路器设备ID唯一），避免同一设备多条重复数据导致模型过拟合；
3. 缺失值填充：工业数据缺失值不建议直接删除（样本量宝贵），采用「同型号设备均值→全局均值」的分层填充策略，保证数据的领域相关性；
4. 异常值修正：基于断路器工业标准，将超出正常范围的指标（如触头温度150℃）修正为范围边界值（90℃），而非直接删除，保留样本的其他有效信息；
5. 结果验证：清洗后需检查缺失值、异常值是否处理完成，确保数据符合工业逻辑。

三、数据标准化：统一格式、数值归一化、文本化转换

工业数据标准化需兼顾「数值一致性」和「模型输入适配性」，核心是将结构化时序数据转换为大模型可理解的文本格式，并完成数值归一化。

3.1 标准化步骤与代码实现

import json
import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler

# 1. 加载清洗后的数据
cleaned_df = pd.read_csv("cleaned_breaker_life_data.csv", encoding="utf-8-sig")

# 2. 定义标准化类
class BreakerDataStandardizer:
    def __init__(self):
        # 数值归一化器
        self.scaler = MinMaxScaler(feature_range=(0, 1))
        # 需要归一化的数值列
        self.numeric_cols = [
            "switch_times", "total_run_hours", "contact_temp",
            "operate_voltage", "vibration", "maintain_freq"
        ]
    
    def normalize_numeric(self, df):
        """数值型指标归一化（[0,1]区间）"""
        # 拟合归一化器
        self.scaler.fit(df[self.numeric_cols])
        # 执行归一化
        normalized_vals = self.scaler.transform(df[self.numeric_cols])
        # 构造归一化后的数据框
        normalized_df = pd.DataFrame(
            normalized_vals,
            columns=[f"{col}_norm" for col in self.numeric_cols],
            index=df.index
        )
        # 合并到原数据框
        df = pd.concat([df, normalized_df], axis=1)
        return df
    
    def convert_to_text(self, row):
        """将结构化数据转换为模型输入的文本序列"""
        # 构造文本模板（兼顾指标完整性和可读性）
        text_template = (
            "断路器设备{device_id}，型号{model}，生产厂家{manufacturer}，安装环境{install_env}，"
            "投运日期{put_in_date}，维护频次{maintain_freq}次/年；"
            "运行指标：分合闸次数{switch_times}次，累计运行时长{total_run_hours}小时，"
            "触头温度{contact_temp}℃，操作电压{operate_voltage}V，机械振动值{vibration}g；"
            "剩余寿命{remain_life}个月，寿命等级{life_level}。"
        )
        # 填充模板（保留原始数值，便于人工核验）
        text = text_template.format(
            device_id=row["device_id"],
            model=row["model"],
            manufacturer=row["manufacturer"],
            install_env=row["install_env"],
            put_in_date=row["put_in_date"],
            maintain_freq=row["maintain_freq"],
            switch_times=row["switch_times"],
            total_run_hours=row["total_run_hours"],
            contact_temp=row["contact_temp"],
            operate_voltage=row["operate_voltage"],
            vibration=row["vibration"],
            remain_life=row["remain_life"],
            life_level=row["life_level"]
        )
        # 构造归一化数值的文本（模型输入用）
        norm_text_template = (
            "断路器{device_id}，型号{model}，维护频次{maintain_freq_norm:.3f}，"
            "分合闸次数{switch_times_norm:.3f}，触头温度{contact_temp_norm:.3f}，"
            "操作电压{operate_voltage_norm:.3f}，振动值{vibration_norm:.3f}，"
            "累计运行时长{total_run_hours_norm:.3f}。"
        )
        norm_text = norm_text_template.format(
            device_id=row["device_id"],
            model=row["model"],
            maintain_freq_norm=row["maintain_freq_norm"],
            switch_times_norm=row["switch_times_norm"],
            contact_temp_norm=row["contact_temp_norm"],
            operate_voltage_norm=row["operate_voltage_norm"],
            vibration_norm=row["vibration_norm"],
            total_run_hours_norm=row["total_run_hours_norm"]
        )
        return text, norm_text
    
    def standardize(self, df):
        """完整标准化流程"""
        # 步骤1：数值归一化
        df = self.normalize_numeric(df)
        # 步骤2：结构化数据文本化
        df[["raw_text", "model_input_text"]] = df.apply(
            lambda row: pd.Series(self.convert_to_text(row)),
            axis=1
        )
        # 步骤3：标签编码（寿命等级→数字）
        life_level_mapping = {"短": 0, "中": 1, "长": 2}
        df["life_level_label"] = df["life_level"].map(life_level_mapping)
        return df

# 3. 执行标准化
standardizer = BreakerDataStandardizer()
standardized_df = standardizer.standardize(cleaned_df)

# 4. 保存标准化后数据
standardized_df.to_csv("standardized_breaker_life_data.csv", index=False, encoding="utf-8-sig")
with open("standardized_breaker_life_data.json", "w", encoding="utf-8") as f:
    json.dump(standardized_df.to_dict("records"), f, ensure_ascii=False, indent=2)

# 保存归一化器（后续推理时复用）
import joblib
joblib.dump(standardizer.scaler, "breaker_scaler.pkl")

print(f"标准化后数据条数：{len(standardized_df)}")
print("\n标准化示例：")
sample_row = standardized_df.iloc[0]
print(f"原始结构化数据→文本：\n{sample_row['raw_text']}")
print(f"\n模型输入文本（归一化数值）：\n{sample_row['model_input_text']}")
print(f"\n寿命等级标签编码：{sample_row['life_level']} → {sample_row['life_level_label']}")
print(f"\n归一化示例：触头温度{sample_row['contact_temp']}℃ → {sample_row['contact_temp_norm']:.3f}")

3.2 标准化步骤说明

1. 数值归一化：
   • 选择MinMaxScaler将所有数值型指标归一化到[0,1]区间（大模型对归一化数值更敏感）；
   • 保存归一化器（breaker_scaler.pkl），后续模型推理时需用相同的归一化规则处理新数据；
2. 结构化数据文本化：
   • 构造两类文本：raw_text（含原始数值，用于人工核验）、model_input_text（仅含核心归一化数值，适配模型输入长度限制）；
   • 文本模板需保留断路器的核心属性（型号、厂家）和关键运行指标，符合大模型的文本理解逻辑；
3. 标签编码：将「短/中/长」寿命等级转换为数字编码（0/1/2），便于模型进行分类任务；
4. 格式统一：所有文本统一为UTF-8编码，数值保留3位小数，确保格式一致性。

四、数据编码：Tokenizer转换为模型可读取格式

工业场景大模型微调的编码核心是将文本化的时序数据转换为模型可读取的张量格式，需兼顾文本长度控制和工业语义保留。

4.1 Tokenizer编码步骤与代码实现

import json
import torch
import pandas as pd
from transformers import BertTokenizer, BertForSequenceClassification

# 1. 加载标准化后的数据
standardized_df = pd.read_csv("standardized_breaker_life_data.csv", encoding="utf-8-sig")

# 2. 初始化Tokenizer（适配工业中文文本）
# 使用工业领域适配的BERT中文Tokenizer（或TimeLLM Tokenizer）
tokenizer = BertTokenizer.from_pretrained("bert-base-chinese")
# 模型输入最大长度（根据文本长度统计设置）
MAX_LENGTH = 128

# 3. 定义编码函数
def encode_breaker_data(df, tokenizer, max_length):
    """
    将文本化的断路器数据转换为模型可读取的张量
    """
    # 初始化存储列表
    input_ids = []
    attention_masks = []
    token_type_ids = []
    labels = []
    metadata = []
    
    for idx, row in df.iterrows():
        # 提取模型输入文本和标签
        text = row["model_input_text"]
        label = row["life_level_label"]
        
        # Tokenizer编码
        encoding = tokenizer(
            text,
            max_length=max_length,
            padding="max_length",  # 不足max_length补0
            truncation=True,       # 超过max_length截断
            return_attention_mask=True,
            return_token_type_ids=True,
            return_tensors="pt"    # 返回PyTorch张量
        )
        
        # 存储编码结果（去除batch维度）
        input_ids.append(encoding["input_ids"].squeeze(0))
        attention_masks.append(encoding["attention_mask"].squeeze(0))
        token_type_ids.append(encoding["token_type_ids"].squeeze(0))
        labels.append(torch.tensor(label, dtype=torch.long))
        
        # 存储元数据（便于后续分析）
        metadata.append({
            "device_id": row["device_id"],
            "raw_text": row["raw_text"],
            "life_level": row["life_level"],
            "remain_life": row["remain_life"]
        })
    
    # 转换为批量张量
    batch_encoding = {
        "input_ids": torch.stack(input_ids),          # 形状：(样本数, max_length)
        "attention_mask": torch.stack(attention_masks),# 形状：(样本数, max_length)
        "token_type_ids": torch.stack(token_type_ids), # 形状：(样本数, max_length)
        "labels": torch.stack(labels),                # 形状：(样本数,)
        "metadata": metadata
    }
    
    return batch_encoding

# 4. 执行编码
encoded_data = encode_breaker_data(standardized_df, tokenizer, MAX_LENGTH)

# 5. 保存编码后的数据
torch.save(encoded_data, "encoded_breaker_life_data.pt")
# 保存Tokenizer配置
tokenizer.save_pretrained("breaker_tokenizer")

print(f"编码后数据维度：")
print(f"input_ids形状：{encoded_data['input_ids'].shape}")
print(f"attention_mask形状：{encoded_data['attention_mask'].shape}")
print(f"labels形状：{encoded_data['labels'].shape}")

# 6. 验证编码结果
sample_idx = 0
print("\n编码结果验证（第1条样本）：")
print(f"原始模型输入文本：{standardized_df.iloc[sample_idx]['model_input_text']}")
print(f"Input IDs（前10个）：{encoded_data['input_ids'][sample_idx][:10]}")
print(f"Attention Mask（前10个）：{encoded_data['attention_mask'][sample_idx][:10]}")
print(f"标签：{encoded_data['labels'][sample_idx]}（对应寿命等级：{standardized_df.iloc[sample_idx]['life_level']}）")
# 解码验证
decoded_text = tokenizer.decode(encoded_data['input_ids'][sample_idx], skip_special_tokens=True)
print(f"解码后文本：{decoded_text}")

4.2 Tokenizer编码步骤说明

1. Tokenizer选择：选用bert-base-chinese（适配中文工业文本），若使用TimeLLM等专用时序大模型，需替换为对应Tokenizer；
2. 编码参数设置：
   • max_length=128：根据断路器文本长度统计（单条文本约80-100个字符）设置，兼顾语义完整性和模型效率；
   • padding="max_length"：确保所有样本输入长度一致；
   • truncation=True：截断超长文本（工业数据文本一般不会超长，此为兜底）；
3. 张量整理：将单条样本的编码结果拼接为批量张量，形状为(样本数, max_length)，符合大模型的批量输入要求；
4. 结果保存：
   • 编码数据保存为.pt文件，便于微调时直接加载；
   • Tokenizer配置单独保存，确保推理时使用相同的编码规则；
5. 编码验证：通过解码input_ids验证编码准确性，确保工业指标的语义未丢失。

五、完整流程总结与扩展建议

5.1 断路器寿命预测数据准备全流程

5.2 核心要点回顾

1. 工业数据特性适配：断路器寿命预测数据需兼顾时序性和领域专业性，清洗时优先「修正而非删除」，缺失值采用分层填充策略；
2. 文本化转换关键：将结构化工业数据转换为自然语言文本是适配大模型的核心，需保留核心指标（分合闸次数、触头温度），控制文本长度；
3. 归一化复用：数值归一化器需保存，后续推理新设备数据时，必须使用相同的归一化规则；
4. 编码一致性：Tokenizer的max_length、padding等参数需与模型输入要求严格匹配，确保张量维度正确。

5.3 扩展建议

1. 数据增强：工业样本量通常较少，可通过「数值小幅度扰动+文本同义改写」增强数据（如触头温度±2℃，文本中「分合闸次数」改为「操作次数」）；
2. 多模态扩展：若有断路器运行振动波形、温度曲线等数据，可将其转换为文本描述（如「振动曲线呈上升趋势，斜率0.02g/月」）融入输入；
3. 模型适配：若使用大语言模型（如LLaMA-3）做寿命预测，需将输入改为指令格式（如「请预测以下断路器的剩余寿命等级：{文本}」）；
4. 数据校验：增加工业专家核验环节，确保数据的专业性和标签准确性，避免模型学习错误的工业逻辑。

通过以上流程，你可将断路器的工业时序/结构化数据，完整转换为大模型微调所需的高质量、结构化、可读取格式，为断路器寿命预测模型的微调奠定核心基础。