# 方法1：简单阈值过滤（适合你已经知道合理范围的场景）
df = df[(df['temperature'] >= -40) & (df['temperature'] <= 150)]

# 方法2：基于统计的异常检测（IQR 方法，适合不确定合理范围时）
Q1 = df['temperature'].quantile(0.25)
Q3 = df['temperature'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 标记异常但不直接删除，方便后续审查
df['is_outlier'] = (df['temperature'] < lower_bound) | (df['temperature'] > upper_bound)
outlier_count = df['is_outlier'].sum()
print(f"发现 {outlier_count} 个异常值，范围外: [{lower_bound:.2f}, {upper_bound:.2f}]")

# 将异常值替换为 NaN，再用插值填补
df.loc[df['is_outlier'], 'temperature'] = None
df['temperature'] = df['temperature'].interpolate(method='linear')
df = df.drop(columns=['is_outlier'])

# 模拟数据
import pandas as pd
import numpy as np

# 加速度计数据：100Hz
accel_times = pd.date_range('2024-05-20 08:00:00', periods=500, freq='10ms')
accel_df = pd.DataFrame({
    'timestamp': accel_times,
    'accel_x': np.random.randn(500) * 0.5,  # 模拟振动数据
    'accel_y': np.random.randn(500) * 0.3,
    'accel_z': np.random.randn(500) * 0.4 + 9.8  # Z轴含重力分量
})

# 温度数据：10Hz
temp_times = pd.date_range('2024-05-20 08:00:00.05', periods=50, freq='100ms')
temp_df = pd.DataFrame({
    'timestamp': temp_times,
    'temperature': np.linspace(25.0, 32.5, 50)  # 模拟温升曲线
})

# 电压数据：1Hz
volt_times = pd.date_range('2024-05-20 08:00:00.5', periods=5, freq='1s')
volt_df = pd.DataFrame({
    'timestamp': volt_times,
    'voltage': [3.30, 3.29, 3.27, 3.25, 3.28]
})

# 使用 merge_asof 将温度"附着"到加速度计的时间轴上
# 方向='nearest' 表示找时间上最近的（无论前后）
merged = pd.merge_asof(
    accel_df.sort_values('timestamp'),
    temp_df.sort_values('timestamp'),
    on='timestamp',
    direction='nearest'    # 关键参数：找最近的时间戳
)

# 继续将电压也合入
merged = pd.merge_asof(
    merged.sort_values('timestamp'),
    volt_df.sort_values('timestamp'),
    on='timestamp',
    direction='nearest',
    tolerance=pd.Timedelta('2s')  # 可选：超过2秒的差距就不匹配了，避免"张冠李戴"
)

print(merged.head(10))
print(f"合并后总行数: {len(merged)}")  # 仍然是500行（以加速度计为基准）

原始数据（不同频率）
    │
    ▼
各数据源分别读取 → 转 datetime → 设置索引
    │
    ▼
降采样到统一频率（可选，取决于分析粒度需求）
    │
    ▼
merge_asof 逐个合入（以最高频数据为基准）
    │
    ▼
interpolate + ffill/bfill 兜底填补
    │
    ▼
对齐完成的统一 DataFrame ✅

import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime

# ============================================================
# 配置区（只改这里！）
# ============================================================
INPUT_DIR = Path('./test_logs/')           # CSV文件所在的文件夹
OUTPUT_FILE = 'test_report_20240520.xlsx'  # 输出的Excel报表文件名
TEMP_MAX_LIMIT = 85.0      # 最高温度阈值（°C）
TEMP_RISE_RATE_LIMIT = 2.0 # 温升速率阈值（°C/min），超过即标记
EXPECTED_SAMPLES = 10000   # 预期每块板子的最低采样数（数据完整性检查）

# ============================================================
# 第一步：批量读取并合并所有 CSV 文件
# ============================================================
def load_all_csv(directory: Path) -> pd.DataFrame:
    """读取目录下所有CSV文件，添加'board_id'列，合并为一个大DataFrame"""
    all_dfs = []
    csv_files = sorted(directory.glob('*.csv'))

    if not csv_files:
        raise FileNotFoundError(f"在 {directory} 下未找到任何CSV文件")

    for file_path in csv_files:
        # 从文件名提取板子编号，例如 'board_001_temp.csv' → '001'
        board_id = file_path.stem.split('_')[1]

        try:
            df = pd.read_csv(file_path)
            df['board_id'] = board_id
            all_dfs.append(df)
        except Exception as e:
            print(f"⚠️ 读取 {file_path.name} 失败: {e}")
            continue

    combined = pd.concat(all_dfs, ignore_index=True)
    print(f"✅ 共读取 {len(csv_files)} 个文件, 合并后共 {len(combined)} 行数据")
    return combined

# ============================================================
# 第二步：数据清洗流水线
# ============================================================
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """标准化的数据清洗流水线"""
    print("\n--- 数据清洗开始 ---")
    original_count = len(df)

    # 2.1 去重
    df = df.drop_duplicates(subset=['timestamp', 'board_id', 'sensor_id'], keep='first')
    print(f"  去重后: {original_count} → {len(df)} 行 (删除 {original_count - len(df)} 行重复)")

    # 2.2 格式转换
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['temperature'] = pd.to_numeric(df['temperature'], errors='coerce')
    df['voltage'] = pd.to_numeric(df['voltage'], errors='coerce')
    df['sensor_id'] = df['sensor_id'].str.strip()

    # 2.3 缺失值处理
    missing_temp = df['temperature'].isnull().sum()
    df['temperature'] = df['temperature'].interpolate(method='linear')
    df['temperature'] = df['temperature'].fillna(method='ffill').fillna(method='bfill')
    df['status'] = df['status'].fillna('UNKNOWN')
    print(f"  温度缺失值: {missing_temp} 个（已插值填补）")

    # 2.4 异常值标记与处理
    df['is_temp_outlier'] = (df['temperature'] < -40) | (df['temperature'] > 150)
    outlier_count = df['is_temp_outlier'].sum()
    if outlier_count > 0:
        print(f"  发现 {outlier_count} 个温度异常值（<-40°C 或 >150°C），已标记")
        # 用前向填充替换异常值
        df.loc[df['is_temp_outlier'], 'temperature'] = None
        df['temperature'] = df['temperature'].fillna(method='ffill')

    # 2.5 排序
    df = df.sort_values(['board_id', 'sensor_id', 'timestamp']).reset_index(drop=True)

    print(f"--- 清洗完成，最终 {len(df)} 行 ---\n")
    return df

# ============================================================
# 第三步：按板子分组聚合，生成汇总指标
# ============================================================
def generate_summary(df: pd.DataFrame) -> pd.DataFrame:
    """对每块板子计算关键统计指标"""
    # 按板子分组
    grouped = df.groupby('board_id')

    # 计算每个板子的统计量
    summary = grouped.agg(
        temp_max=('temperature', 'max'),
        temp_min=('temperature', 'min'),
        temp_mean=('temperature', 'mean'),
        temp_std=('temperature', 'std'),
        sample_count=('temperature', 'count'),
        voltage_mean=('voltage', 'mean'),
    ).round(2)

    # 计算温升速率：对每块板子的温度数据求差分，再除以时间差
    def calc_max_rise_rate(group):
        """计算最大温升速率（°C/min）"""
        group = group.sort_values('timestamp')
        temp_diff = group['temperature'].diff()
        time_diff = group['timestamp'].diff().dt.total_seconds() / 60  # 转为分钟
        # 避免除以零
        valid = time_diff > 0
        if valid.any():
            rates = temp_diff[valid] / time_diff[valid]
            return round(rates.max(), 2)
        return 0.0

    rise_rates = grouped.apply(calc_max_rise_rate)
    summary['max_rise_rate'] = rise_rates

    # 计算超标次数
    def count_exceeded(group):
        """计算温度超过阈值的数据点数量"""
        return (group['temperature'] > TEMP_MAX_LIMIT).sum()

    exceed_counts = grouped.apply(count_exceeded)
    summary['exceed_count'] = exceed_counts

    # 数据完整性检查
    summary['data_complete'] = summary['sample_count'] >= EXPECTED_SAMPLES

    # 最终判定
    def judge(row):
        if not row['data_complete']:
            return 'INCOMPLETE'
        if row['temp_max'] > TEMP_MAX_LIMIT:
            return 'FAIL'
        if row['max_rise_rate'] > TEMP_RISE_RATE_LIMIT:
            return 'WARN'
        return 'PASS'

    summary['verdict'] = summary.apply(judge, axis=1)

    # 重置索引，让 board_id 变回普通列
    summary = summary.reset_index()

    # 重新排列列顺序，让结论放最前面
    cols = ['board_id', 'verdict', 'temp_max', 'temp_mean', 'temp_min',
            'temp_std', 'max_rise_rate', 'exceed_count',
            'sample_count', 'data_complete', 'voltage_mean']
    summary = summary[cols]

    return summary

# ============================================================
# 第四步：提取异常明细
# ============================================================
def extract_anomalies(df: pd.DataFrame) -> pd.DataFrame:
    """提取所有需要关注的异常数据点"""
    # 条件1：温度超过阈值
    high_temp = df[df['temperature'] > TEMP_MAX_LIMIT].copy()
    high_temp['anomaly_type'] = 'HIGH_TEMP'

    # 条件2：原数据中的异常值标记
    flagged = df[df['is_temp_outlier']].copy()
    flagged['anomaly_type'] = 'SENSOR_ERROR'

    # 合并异常记录
    anomalies = pd.concat([high_temp, flagged], ignore_index=True)
    anomalies = anomalies.drop(columns=['is_temp_outlier'], errors='ignore')

    if not anomalies.empty:
        anomalies = anomalies.sort_values(['board_id', 'timestamp'])

    return anomalies

# ============================================================
# 第五步：输出 Excel 报表
# ============================================================
def export_report(summary: pd.DataFrame, anomalies: pd.DataFrame, output_path: str):
    """将汇总表和异常明细写入同一个Excel文件的不同Sheet"""
    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
        # Sheet 1: 汇总表
        summary.to_excel(writer, sheet_name='板子汇总', index=False)

        # Sheet 2: 异常明细
        if not anomalies.empty:
            anomalies.to_excel(writer, sheet_name='异常明细', index=False)
        else:
            pd.DataFrame({'说明': ['未发现任何异常数据']}).to_excel(
                writer, sheet_name='异常明细', index=False
            )

    print(f"✅ 报表已生成: {output_path}")
    print(f"   - 汇总: {len(summary)} 块板子")
    verdict_counts = summary['verdict'].value_counts()
    for v, c in verdict_counts.items():
        print(f"   - {v}: {c} 块")

# ============================================================
# 主程序入口
# ============================================================
if __name__ == '__main__':
    # 1. 加载数据
    raw_df = load_all_csv(INPUT_DIR)

    # 2. 清洗数据
    clean_df = clean_data(raw_df)

    # 3. 生成汇总
    summary_df = generate_summary(clean_df)

    # 4. 提取异常
    anomalies_df = extract_anomalies(clean_df)

    # 5. 输出报表
    export_report(summary_df, anomalies_df, OUTPUT_FILE)

【数据描述】
我的CSV文件路径是: xxx.csv
列名及含义:
  - timestamp: 采样时间，格式 "2024-05-20 08:00:01.123"
  - sensor_id: 传感器编号，值为 "NTC_1", "NTC_2", "NTC_3"
  - temperature: 温度值（°C），浮点数，可能有缺失值
  - voltage: 电压值（V），浮点数
  - status: 状态标记，"OK" 或 "ERROR" 或空值
数据特征:
  - 约20000行，单个传感器约6600行
  - 采样频率约10Hz
  - 存在重复行、缺失值、传感器偶发异常尖峰

【期望输出】
我想得到一份按 sensor_id 分组的统计报表:
  1. 每个传感器的温度: 最大值、最小值、均值、标准差
  2. 每个传感器温度超过 85°C 的次数
  3. 每个传感器的数据完整性（实际行数 vs 预期行数）
  4. 最后输出为 Excel 文件，包含两个 Sheet: "汇总" 和 "高温明细"

【附加要求】
  - 代码要有清晰的函数拆分和中文注释
  - 清洗步骤包括: 去重、缺失值插值、异常值过滤（<-40 或 >150）
  - 输出代码要能直接运行