Python Polars模块革命性解析:比Pandas快100倍的数据处理神器,让大数据分析飞起来!

import polars as pl
import numpy as np

# 示例1：创建DataFrame的多种方式
# 方式1：从字典创建
df1 = pl.DataFrame({
"姓名": ["张三", "李四", "王五", "赵六"],
"年龄": [25, 30, 35, 28],
"城市": ["北京", "上海", "广州", "深圳"],
"薪资": [15000, 20000, 18000, 22000]
})

print("DataFrame 1（字典创建）：")
print(df1)
print(f"形状：{df1.shape}")  # (4, 4)
print(f"数据类型：{df1.dtypes}")  # [Utf8, Int64, Utf8, Int64]

# 方式2：从NumPy数组创建
names = np.array(["张三", "李四", "王五", "赵六"])
ages = np.array([25, 30, 35, 28], dtype=np.int64)
salaries = np.array([15000, 20000, 18000, 22000], dtype=np.int64)

df2 = pl.DataFrame({
"姓名": names,
"年龄": ages,
"薪资": salaries
})

print("\nDataFrame 2（NumPy数组创建）：")
print(df2)

# 方式3：从Pandas DataFrame转换（兼容现有代码）
import pandas as pd
pd_df = pd.DataFrame({
"A": [1, 2, 3, 4],
"B": ["a", "b", "c", "d"]
})
df3 = pl.from_pandas(pd_df)
print("\nDataFrame 3（从Pandas转换）：")
print(df3)

# 示例2：高级数据筛选
# 创建示例数据集
employees = pl.DataFrame({
"id": range(1, 100001),  # 10万行数据
"name": [f"员工{i}"for i in range(1, 100001)],
"department": np.random.choice(["技术部", "市场部", "人事部", "财务部", "产品部"], 100000),
"salary": np.random.randint(3000, 50000, 100000),
"years_experience": np.random.randint(0, 20, 100000),
"join_date": pl.date_range(
        start=pl.datetime(2020, 1, 1),
        end=pl.datetime(2026, 4, 8),
        interval="1d",
        eager=True
    )[:100000]
})

print("原始数据集信息：")
print(f"行数：{employees.shape[0]}")
print(f"列数：{employees.shape[1]}")
print("\n数据预览：")
print(employees.head())

# 复杂条件筛选：链式表达式
high_salary_tech = employees.filter(
    (pl.col("department") == "技术部") &
    (pl.col("salary") > 30000) &
    (pl.col("years_experience") >= 5)
)

print(f"\n技术部高薪资深员工（薪资>30000，经验≥5年）：{len(high_salary_tech)}人")

# 多条件组合筛选
market_or_product = employees.filter(
    pl.col("department").is_in(["市场部", "产品部"])
).filter(
    pl.col("salary").is_between(20000, 40000)
)

print(f"\n市场部或产品部中等薪资员工：{len(market_or_product)}人")

# 时间范围筛选
recent_employees = employees.filter(
    pl.col("join_date") >= pl.datetime(2025, 1, 1)
)

print(f"\n2025年后入职的新员工：{len(recent_employees)}人")

# 示例3：高效分组聚合
# 部门薪资统计分析
dept_stats = employees.group_by("department").agg([
    pl.col("salary").mean().alias("平均薪资"),
    pl.col("salary").median().alias("薪资中位数"),
    pl.col("salary").std().alias("薪资标准差"),
    pl.col("salary").min().alias("最低薪资"),
    pl.col("salary").max().alias("最高薪资"),
    pl.col("id").count().alias("员工数量")
]).sort("平均薪资", descending=True)

print("各部门薪资统计分析：")
print(dept_stats)

# 多级分组与聚合
# 按部门和经验级别分组
experience_levels = pl.when(pl.col("years_experience") < 3).then("初级")\
                     .when(pl.col("years_experience") < 8).then("中级")\
                     .otherwise("高级")

dept_exp_stats = employees.with_columns(
    experience_levels.alias("经验级别")
).group_by(["department", "经验级别"]).agg([
    pl.col("salary").mean().alias("平均薪资"),
    pl.col("id").count().alias("人数")
]).sort(["department", "经验级别"])

print("\n部门-经验级别薪资分析：")
print(dept_exp_stats)

# 示例4：高性能数据连接
# 创建部门信息表
departments = pl.DataFrame({
"department": ["技术部", "市场部", "人事部", "财务部", "产品部"],
"manager": ["张总", "李总", "王总", "赵总", "刘总"],
"budget": [5000000, 3000000, 800000, 2000000, 2500000],
"location": ["A栋", "B栋", "C栋", "D栋", "E栋"]
})

# 创建项目信息表
projects = pl.DataFrame({
"project_id": range(1, 21),
"project_name": [f"项目{i}"for i in range(1, 21)],
"department": np.random.choice(["技术部", "市场部", "人事部", "财务部", "产品部"], 20),
"budget": np.random.randint(100000, 1000000, 20),
"status": np.random.choice(["进行中", "已完成", "延期"], 20)
})

# 内连接：员工表 + 部门表
employees_with_dept = employees.join(
    departments,
on="department",
    how="inner"
)

print("员工部门连接结果：")
print(employees_with_dept.select(["id", "name", "department", "manager", "salary"]).head())

# 左连接：项目表 + 部门表
projects_with_dept = projects.join(
    departments,
on="department",
    how="left"
)

print("\n项目部门连接结果：")
print(projects_with_dept.head())

# 多表连接：复杂业务场景
# 首先员工连接到部门，再部门连接到项目
complex_join = employees.join(
    departments,
on="department",
    how="inner"
).join(
    projects.filter(pl.col("status") == "进行中"),
on="department",
    how="inner",
    suffix="_project"
)

print("\n复杂多表连接结果（员工-部门-进行中项目）：")
print(complex_join.select([
"id", "name", "department", "manager", "project_name", "budget_project"
]).head())

# 示例5：惰性计算实战
# 创建大型数据集（模拟1000万行销售数据）
sales_data = pl.DataFrame({
"sale_id": range(1, 10000001),
"product_id": np.random.randint(1, 1000, 10000000),
"customer_id": np.random.randint(1, 50000, 10000000),
"quantity": np.random.randint(1, 100, 10000000),
"price": np.random.uniform(10, 500, 10000000),
"sale_date": pl.date_range(
        start=pl.datetime(2024, 1, 1),
        end=pl.datetime(2026, 4, 8),
        interval="10s",
        eager=True
    )[:10000000],
"region": np.random.choice(["华北", "华东", "华南", "华中", "西南", "西北"], 10000000)
})

print("销售数据集信息：")
print(f"数据量：{sales_data.shape[0]:,} 行")
print(f"占用内存：{sales_data.estimated_size() / 1024**2:.2f} MB")

# 使用惰性API处理（不立即执行）
lazy_sales = sales_data.lazy()

# 构建复杂查询计划
complex_query = lazy_sales.filter(
    pl.col("sale_date") >= pl.datetime(2025, 1, 1)
).filter(
    pl.col("price") > 100
).group_by(["region", "product_id"]).agg([
    pl.col("quantity").sum().alias("总销量"),
    (pl.col("quantity") * pl.col("price")).sum().alias("总收入"),
    pl.col("customer_id").n_unique().alias("客户数")
]).sort("总收入", descending=True).limit(100)

# 显示优化后的查询计划
print("\n查询计划：")
print(complex_query.explain())

# 执行查询（触发实际计算）
start_time = pl.datetime.now()
result = complex_query.collect()
end_time = pl.datetime.now()

print(f"\n查询执行时间：{(end_time - start_time).total_seconds():.3f}秒")
print(f"结果行数：{result.shape[0]}")
print("\n结果预览：")
print(result.head())

# 示例6：流式处理大数据
# 创建超大数据集（模拟1亿行日志数据）
# 注意：实际生成1亿行数据会消耗大量内存，这里用示例说明
import time

print("开始流式处理模拟...")

# 模拟流式读取和处理大文件
defprocess_large_file_streaming():
"""模拟流式处理大文件"""
    total_rows = 100_000_000  # 1亿行
    chunk_size = 1_000_000   # 每批处理100万行

# 模拟流式处理
    total_processed = 0
    start_time = time.time()

for chunk_num in range(total_rows // chunk_size):
# 模拟生成一个数据块
        chunk_data = pl.DataFrame({
"log_id": range(chunk_num * chunk_size, (chunk_num + 1) * chunk_size),
"timestamp": pl.date_range(
                start=pl.datetime(2026, 1, 1),
                end=pl.datetime(2026, 4, 8),
                interval="2s",
                eager=True
            )[:chunk_size],
"user_id": np.random.randint(1, 1000000, chunk_size),
"action": np.random.choice(["login", "click", "purchase", "view", "search"], chunk_size),
"response_time": np.random.exponential(100, chunk_size)
        })

# 流式处理：筛选和聚合
        processed_chunk = chunk_data.lazy().filter(
            pl.col("response_time") < 1000
        ).group_by("action").agg([
            pl.col("response_time").mean().alias("平均响应时间"),
            pl.col("user_id").n_unique().alias("独立用户数")
        ]).collect()

        total_processed += chunk_size

# 每处理10个数据块输出一次进度
if (chunk_num + 1) % 10 == 0:
            elapsed = time.time() - start_time
            speed = total_processed / elapsed
            print(f"已处理：{total_processed:,} 行，耗时：{elapsed:.1f}秒，速度：{speed:,.0f} 行/秒")

    end_time = time.time()
    print(f"\n流式处理完成！")
    print(f"总耗时：{end_time - start_time:.2f}秒")
    print(f"平均速度：{total_processed / (end_time - start_time):,.0f} 行/秒")

# 执行流式处理模拟
process_large_file_streaming()

# 示例7：高级表达式应用
# 创建金融交易数据集
transactions = pl.DataFrame({
"txn_id": range(1, 500001),
"account_id": np.random.randint(1000, 5000, 500000),
"txn_type": np.random.choice(["存款", "取款", "转账", "缴费", "理财"], 500000),
"amount": np.random.uniform(-50000, 50000, 500000),
"fee": np.random.uniform(0, 50, 500000),
"txn_date": pl.date_range(
        start=pl.datetime(2025, 1, 1),
        end=pl.datetime(2026, 4, 8),
        interval="1m",
        eager=True
    )[:500000],
"channel": np.random.choice(["网银", "手机银行", "ATM", "柜台", "第三方支付"], 500000)
})

# 复杂表达式：多条件计算和分类
analysis_result = transactions.lazy().with_columns([
    # 交易净额（考虑手续费）
    (pl.col("amount") - pl.col("fee")).alias("net_amount"),

    # 交易规模分类
    pl.when(pl.col("amount").abs() < 1000).then("小额")
     .when(pl.col("amount").abs() < 10000).then("中额")
     .otherwise("大额").alias("amount_category"),

    # 时间分段
    pl.col("txn_date").dt.hour().alias("hour_of_day"),

    # 是否为异常交易（金额过大或手续费过高）
    ((pl.col("amount").abs() > 30000) | 
     (pl.col("fee") > pl.col("amount").abs() * 0.05)).alias("is_suspicious")
]).group_by(["account_id", "txn_type"]).agg([
    pl.col("amount").sum().alias("总金额"),
    pl.col("net_amount").sum().alias("净金额"),
    pl.col("amount").mean().alias("平均金额"),
    pl.col("txn_id").count().alias("交易笔数"),
    pl.col("is_suspicious").sum().alias("可疑交易数"),

    # 时间分布统计
    pl.col("hour_of_day").mode().first().alias("最活跃时段")
]).filter(
    pl.col("交易笔数") > 10
).sort("总金额", descending=True).collect()

print("账户交易分析结果：")
print(analysis_result.head(10))

# 窗口函数应用：计算移动平均和累计
window_analysis = transactions.lazy().filter(
    pl.col("txn_type") == "存款"
).with_columns([
    # 按账户和时间排序
    pl.col("amount").cum_sum().over("account_id").alias("累计存款"),

    # 7天移动平均
    pl.col("amount").rolling_mean(window_size="7d", by="txn_date").alias("7日移动平均"),

    # 存款排名
    pl.col("amount").rank(method="dense", descending=True).over("account_id").alias("存款排名")
]).sort(["account_id", "txn_date"]).collect()

print("\n窗口函数分析结果（存款交易）：")
print(window_analysis.head(10))

# 优化前：先连接后过滤（效率低）
result = df1.lazy().join(df2.lazy(), on="id").filter(pl.col("value") > 100).collect()

# 优化后：先过滤后连接（利用谓词下推）
result = df1.lazy().filter(pl.col("value") > 100).join(
    df2.lazy().filter(pl.col("value") > 100), 
on="id"
).collect()

# 优化前：选择所有列（内存占用高）
result = df.lazy().select(["*"]).group_by("category").agg([
    pl.col("value").mean()
]).collect()

# 优化后：只选择需要的列
result = df.lazy().select(["category", "value"]).group_by("category").agg([
    pl.col("value").mean()
]).collect()

# 显式指定连接算法
result = df1.lazy().join(
    df2.lazy(),
on="id",
    how="inner",
    join_nulls=False,  # 优化空值处理
    suffix="_right"
).collect()

# 查看DataFrame内存占用
print(f"内存占用：{df.estimated_size() / 1024**2:.2f} MB")

# 查看列级内存占用
for col in df.columns:
    col_size = df[col].estimated_size()
    print(f"{col}: {col_size / 1024:.2f} KB")

# 分块处理大文件
chunk_size = 1_000_000
results = []

for i in range(0, total_rows, chunk_size):
    chunk = df.slice(i, min(chunk_size, total_rows - i))
    processed = chunk.lazy().group_by("category").agg([
        pl.col("value").sum()
    ]).collect()
    results.append(processed)

# 合并结果
final_result = pl.concat(results)

# 示例8：Polars vs Pandas性能基准测试
import pandas as pd
import time

# 创建测试数据集（100万行）
test_size = 1_000_000
test_data = {
"id": range(test_size),
"value1": np.random.randn(test_size),
"value2": np.random.randn(test_size),
"category": np.random.choice(["A", "B", "C", "D", "E"], test_size),
"flag": np.random.choice([True, False], test_size)
}

# 转换为Pandas和Polars DataFrame
print("创建测试数据集...")
pd_df = pd.DataFrame(test_data)
pl_df = pl.DataFrame(test_data)

# 测试1：数据筛选
print("\n=== 测试1：数据筛选 ===")
print("筛选条件：category == 'A' AND value1 > 0")

# Pandas筛选
start = time.time()
pd_result = pd_df[(pd_df["category"] == "A") & (pd_df["value1"] > 0)]
pd_time = time.time() - start
print(f"Pandas 耗时：{pd_time:.4f}秒，结果行数：{len(pd_result)}")

# Polars筛选
start = time.time()
pl_result = pl_df.filter(
    (pl.col("category") == "A") & (pl.col("value1") > 0)
)
pl_time = time.time() - start
print(f"Polars 耗时：{pl_time:.4f}秒，结果行数：{len(pl_result)}")
print(f"性能提升：{pd_time/pl_time:.1f}倍")

# 测试2：分组聚合
print("\n=== 测试2：分组聚合 ===")
print("按category分组，计算value1和value2的平均值")

# Pandas分组
start = time.time()
pd_group = pd_df.groupby("category")[["value1", "value2"]].mean().reset_index()
pd_time = time.time() - start
print(f"Pandas 耗时：{pd_time:.4f}秒")

# Polars分组
start = time.time()
pl_group = pl_df.group_by("category").agg([
    pl.col("value1").mean(),
    pl.col("value2").mean()
])
pl_time = time.time() - start
print(f"Polars 耗时：{pl_time:.4f}秒")
print(f"性能提升：{pd_time/pl_time:.1f}倍")

# 测试3：排序操作
print("\n=== 测试3：排序操作 ===")
print("按value1降序排序")

# Pandas排序
start = time.time()
pd_sorted = pd_df.sort_values("value1", ascending=False)
pd_time = time.time() - start
print(f"Pandas 耗时：{pd_time:.4f}秒")

# Polars排序
start = time.time()
pl_sorted = pl_df.sort("value1", descending=True)
pl_time = time.time() - start
print(f"Polars 耗时：{pl_time:.4f}秒")
print(f"性能提升：{pd_time/pl_time:.1f}倍")

# 测试4：多列计算
print("\n=== 测试4：多列计算 ===")
print("计算value1和value2的和、差、乘积")

# Pandas计算
start = time.time()
pd_df["sum"] = pd_df["value1"] + pd_df["value2"]
pd_df["diff"] = pd_df["value1"] - pd_df["value2"]
pd_df["product"] = pd_df["value1"] * pd_df["value2"]
pd_time = time.time() - start
print(f"Pandas 耗时：{pd_time:.4f}秒")

# Polars计算
start = time.time()
pl_df = pl_df.with_columns([
    (pl.col("value1") + pl.col("value2")).alias("sum"),
    (pl.col("value1") - pl.col("value2")).alias("diff"),
    (pl.col("value1") * pl.col("value2")).alias("product")
])
pl_time = time.time() - start
print(f"Polars 耗时：{pl_time:.4f}秒")
print(f"性能提升：{pd_time/pl_time:.1f}倍")

print("\n=== 总结 ===")
print("Polars在各项操作上均显著优于Pandas，特别在处理大规模数据时，性能优势更加明显！")

# 场景1代码：电商用户分析
# 模拟电商用户行为数据
user_behavior = pl.DataFrame({
"user_id": np.random.randint(1, 1000000, 10000000),
"session_id": np.random.randint(1, 5000000, 10000000),
"action": np.random.choice(["view", "click", "add_cart", "purchase", "search"], 10000000),
"product_id": np.random.randint(1, 10000, 10000000),
"timestamp": pl.date_range(
        start=pl.datetime(2026, 1, 1),
        end=pl.datetime(2026, 4, 8),
        interval="1s",
        eager=True
    )[:10000000],
"duration": np.random.exponential(30, 10000000)
})

# 分析用户价值
user_value_analysis = user_behavior.lazy().group_by("user_id").agg([
    pl.col("action").filter(pl.col("action") == "purchase").count().alias("购买次数"),
    pl.col("action").filter(pl.col("action") == "add_cart").count().alias("加购次数"),
    pl.col("duration").sum().alias("总停留时间"),
    pl.col("session_id").n_unique().alias("活跃会话数"),
    pl.col("action").count().alias("总行为数")
]).with_columns([
    # 计算用户价值得分
    (pl.col("购买次数") * 0.5 + 
     pl.col("加购次数") * 0.3 + 
     pl.col("总停留时间") * 0.001 + 
     pl.col("活跃会话数") * 0.2).alias("价值得分")
]).sort("价值得分", descending=True).limit(1000).collect()

print("高价值用户分析完成！")
print(f"分析数据量：{user_behavior.shape[0]:,} 行")
print(f"识别高价值用户：{len(user_value_analysis)} 人")
print("\nTop 10高价值用户：")
print(user_value_analysis.head(10))

# 场景2代码：金融风控监控
# 模拟实时交易流
real_time_transactions = pl.DataFrame({
"txn_id": range(1, 1000001),
"account_id": np.random.randint(1000, 10000, 1000000),
"counterparty": np.random.randint(5000, 20000, 1000000),
"amount": np.random.exponential(1000, 1000000),
"currency": np.random.choice(["CNY", "USD", "EUR", "JPY"], 1000000),
"txn_time": pl.datetime_range(
        start=pl.datetime(2026, 4, 8, 0, 0, 0),
        end=pl.datetime(2026, 4, 8, 23, 59, 59),
        interval="86ms",  # 每86毫秒一笔交易
        eager=True
    )[:1000000]
})

# 实时风控规则检测
risk_monitoring = real_time_transactions.lazy().with_columns([
# 规则1：大额交易标记
    (pl.col("amount") > 100000).alias("is_large_amount"),

# 规则2：高频交易标记（窗口内交易次数）
    pl.col("txn_id").count().over(
"account_id", 
        window_size="5m", 
        by="txn_time"
    ).alias("transactions_5m"),

# 规则3：异常时间交易（凌晨交易）
    ((pl.col("txn_time").dt.hour() >= 0) & 
     (pl.col("txn_time").dt.hour() < 6)).alias("is_night_txn")
]).filter(
# 触发风控规则的条件
    (pl.col("is_large_amount") == True) |
    (pl.col("transactions_5m") > 10) |
    (pl.col("is_night_txn") == True)
).collect()

print("实时风控监控完成！")
print(f"监控交易数：{real_time_transactions.shape[0]:,} 笔")
print(f"触发风控规则：{len(risk_monitoring)} 笔")
print("\n可疑交易示例：")
print(risk_monitoring.select(["txn_id", "account_id", "amount", "txn_time"]).head())

# 场景3代码：物联网数据处理
# 模拟传感器数据流（使用流式处理）
sensor_data = pl.DataFrame({
"sensor_id": np.random.randint(1, 1000, 5000000),
"timestamp": pl.datetime_range(
        start=pl.datetime(2026, 4, 1, 0, 0, 0),
        end=pl.datetime(2026, 4, 8, 0, 0, 0),
        interval="1s",
        eager=True
    )[:5000000],
"temperature": np.random.normal(25, 5, 5000000),
"humidity": np.random.normal(60, 10, 5000000),
"pressure": np.random.normal(1013, 5, 5000000)
})

# 流式异常检测
anomaly_detection = sensor_data.lazy().with_columns([
    # 计算移动统计量
    pl.col("temperature").rolling_mean(window_size="1h", by="timestamp").alias("temp_1h_avg"),
    pl.col("temperature").rolling_std(window_size="1h", by="timestamp").alias("temp_1h_std"),

    # 检测异常（超过3个标准差）
    ((pl.col("temperature") - pl.col("temp_1h_avg")).abs() > 
3 * pl.col("temp_1h_std")).alias("is_temperature_anomaly"),

    # 趋势分析（与前一小时比较）
    ((pl.col("temperature") - 
      pl.col("temperature").shift(3600).over("sensor_id", by="timestamp")) > 
5).alias("temp_spike")
]).filter(
    pl.col("is_temperature_anomaly") == True
).collect()

print("物联网传感器异常检测完成！")
print(f"处理数据量：{sensor_data.shape[0]:,} 条")
print(f"检测到异常：{len(anomaly_detection)} 条")
print("\n异常数据示例：")
print(anomaly_detection.select(["sensor_id", "timestamp", "temperature"]).head())

# 检查并优化配置
import polars as pl

# 1. 启用惰性计算（默认启用）
df = pl.read_csv("large_file.csv").lazy()

# 2. 检查并行线程数
print(f"可用CPU核心数：{pl.thread_pool_size()}")

# 3. 调整内存管理策略
pl.Config.set_tbl_rows(100)  # 限制显示行数
pl.Config.set_tbl_cols(20)   # 限制显示列数

# 4. 使用合适的列数据类型
optimized_df = df.with_columns([
    pl.col("category").cast(pl.Categorical),  # 分类数据使用Categorical类型
    pl.col("amount").cast(pl.Float32),        # 浮点数使用Float32节省内存
])

# 5. 批量处理大文件
result = (
    pl.scan_csv("huge_file.csv", n_rows=1000000)  # 分批读取
    .filter(pl.col("value") > 0)
    .collect(streaming=True)  # 启用流式处理
)

# 内存优化策略
# 1. 使用适当的数据类型
df = df.with_columns([
    pl.col("id").cast(pl.UInt32),          # 使用无符号整数
    pl.col("price").cast(pl.Float32),       # 使用32位浮点数
    pl.col("category").cast(pl.Categorical), # 分类数据
])

# 2. 分块处理大数据
chunk_size = 1000000
results = []

for i in range(0, total_rows, chunk_size):
    chunk = df.slice(i, min(chunk_size, total_rows - i))
# 处理chunk
    processed = chunk.lazy().group_by("category").agg([
        pl.col("value").sum()
    ]).collect()
    results.append(processed)

# 3. 及时释放内存
import gc
del df
gc.collect()

# 分阶段迁移策略
# 阶段1：并行运行，对比验证
defpandas_workflow(df_pandas):
"""原Pandas工作流"""
    result = df_pandas.groupby("category")["value"].agg(["mean", "sum"])
return result

defpolars_workflow(df_polars):
"""对应Polars工作流"""
    result = df_polars.group_by("category").agg([
        pl.col("value").mean().alias("mean"),
        pl.col("value").sum().alias("sum")
    ])
return result

# 阶段2：逐步替换模块
classHybridDataProcessor:
"""混合处理器，逐步迁移"""
def__init__(self, use_polars=True):
        self.use_polars = use_polars

defprocess(self, df):
if self.use_polars:
# 使用Polars处理
if isinstance(df, pd.DataFrame):
                df = pl.from_pandas(df)
return self._polars_process(df)
else:
# 使用Pandas处理
return self._pandas_process(df)

def_polars_process(self, df):
# Polars处理逻辑
return df.lazy().filter(pl.col("value") > 0).collect()

def_pandas_process(self, df):
# Pandas处理逻辑
return df[df["value"] > 0]

# 阶段3：性能监控和优化
defbenchmark_migration(original_func, new_func, test_data):
"""迁移性能对比"""
import time

# 原实现
    start = time.time()
    original_result = original_func(test_data.copy())
    original_time = time.time() - start

# 新实现
    start = time.time()
    new_result = new_func(test_data.copy())
    new_time = time.time() - start

    print(f"原实现耗时：{original_time:.3f}秒")
    print(f"新实现耗时：{new_time:.3f}秒")
    print(f"性能提升：{original_time/new_time:.1f}倍")

return original_result, new_result

# 分析Pandas使用模式
import ast

defanalyze_pandas_usage(file_path):
with open(file_path, 'r') as f:
        tree = ast.parse(f.read())

    pandas_calls = []
for node in ast.walk(tree):
if isinstance(node, ast.Call):
# 检测Pandas方法调用
pass

return pandas_calls

classDataProcessor:
def__init__(self, use_polars=True):
self.use_polars = use_polars

defprocess(self, data):
ifself.use_polars:
returnself._polars_process(data)
else:
returnself._pandas_process(data)

def_polars_process(self, data):
# Polars实现
           pass

def_pandas_process(self, data):
# 保持原有Pandas实现
           pass

classMigrationMonitor:
def__init__(self):
self.original_times = []
self.new_times = []

defbenchmark(self, func_original, func_new, test_data):
# 记录性能对比
           pass