借助 DeepSeek 辅助优化 Python 脚本：从数据分析到性能提升

import pandas as pd
import numpy as np
import os

def process_file(file_path):
    """处理单个文件"""
    df = pd.read_csv(file_path)
    
    # 数据清洗
    df = df[df['value'] > 0]  # 去除负值
    df = df.dropna()          # 去除空值
    
    # 计算统计量
    stats = {
        'mean': df['value'].mean(),
        'std': df['value'].std(),
        'max': df['value'].max(),
        'min': df['value'].min()
    }
    
    # 输出结果
    output_path = f"processed_{os.path.basename(file_path)}"
    df.to_csv(output_path, index=False)
    
    return stats

def main():
    file_dir = "data/"
    all_stats = []
    
    for file_name in os.listdir(file_dir):
        if file_name.endswith('.csv'):
            file_path = os.path.join(file_dir, file_name)
            stats = process_file(file_path)
            all_stats.append(stats)
    
    # 保存汇总统计
    pd.DataFrame(all_stats).to_csv("summary_stats.csv", index=False)

if __name__ == "__main__":
    main()

# DeepSeek 建议的性能分析代码
import cProfile
import pstats

def profile_main():
    profiler = cProfile.Profile()
    profiler.enable()
    
    main()
    
    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumtime').print_stats(10)

# profile_main()  # 取消注释运行性能分析

def process_file_optimized(file_path):
    """优化后的单文件处理"""
    # 指定数据类型避免自动推断
    dtypes = {
        'timestamp': 'str',
        'value': 'float32',    # 使用 float32 减少内存
        'sensor_id': 'int32'
    }
    
    # 使用更高效的参数
    df = pd.read_csv(
        file_path,
        dtype=dtypes,
        usecols=['timestamp', 'value', 'sensor_id'],  # 只读取需要的列
        parse_dates=['timestamp'],
        engine='c'  # 使用C引擎加速
    )
    
    # 使用 query() 方法加速过滤
    df = df.query('value > 0').dropna()
    
    # 使用更高效的内存操作
    stats = {
        'mean': df['value'].mean(),
        'std': df['value'].std(ddof=0),  # 更快的标准差计算
        'max': df['value'].max(),
        'min': df['value'].min()
    }
    
    # 使用更快的输出格式
    output_path = f"processed_{os.path.basename(file_path)}"
    df.to_parquet(output_path, index=False)  # Parquet 比 CSV 更快更小
    
    return stats

from concurrent.futures import ProcessPoolExecutor
import multiprocessing

def parallel_process_files():
    """并行处理文件"""
    file_dir = "data/"
    csv_files = [
        os.path.join(file_dir, f) 
        for f in os.listdir(file_dir) 
        if f.endswith('.csv')
    ]
    
    # 根据CPU核心数调整进程数
    num_workers = min(multiprocessing.cpu_count(), len(csv_files))
    
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        results = list(executor.map(process_file_optimized, csv_files))
    
    return results

def process_file_memory_optimized(file_path):
    """内存优化的处理函数"""
    # 使用分块读取处理大文件
    chunk_size = 100000
    chunks = []
    
    for chunk in pd.read_csv(
        file_path,
        chunksize=chunk_size,
        dtype={'value': 'float32'},
        usecols=['timestamp', 'value']
    ):
        # 立即过滤减少内存占用
        chunk = chunk[chunk['value'] > 0]
        chunk = chunk.dropna()
        chunks.append(chunk)
    
    # 合并结果
    if chunks:
        df = pd.concat(chunks, ignore_index=True)
    else:
        df = pd.DataFrame()
    
    # 后续处理...
    return df

# DeepSeek 建议的高效代码模式示例

# 原始代码
result = []
for item in large_list:
    if condition(item):
        result.append(transform(item))

# 优化建议：使用生成器表达式
result = (transform(item) for item in large_list if condition(item))