Python实现价格区间筛选与排序功能详解

引言

在电商系统、金融分析或数据可视化场景中，价格区间筛选和排序是高频需求。本文将系统讲解如何使用Python实现这两个核心功能，从基础实现到性能优化，提供完整的代码示例和工程化建议。

一、基础实现方案

1.1 价格区间筛选实现

使用列表推导式是最简洁的实现方式：

def filter_by_price_range(products, min_price, max_price):
    """
    筛选指定价格区间的商品
    :param products: 商品列表，每个商品为字典格式{'name': str, 'price': float}
    :param min_price: 最低价格
    :param max_price: 最高价格
    :return: 筛选后的商品列表
    """
    return [p for p in products if min_price <= p['price'] <= max_price]

实际应用示例：

products = [
    {'name': '手机', 'price': 2999},
    {'name': '笔记本', 'price': 5999},
    {'name': '耳机', 'price': 399},
    {'name': '平板', 'price': 1999}
]
# 筛选1000-3000价格区间的商品
filtered = filter_by_price_range(products, 1000, 3000)
print(filtered)
# 输出: [{'name': '手机', 'price': 2999}, {'name': '平板', 'price': 1999}]

1.2 价格排序实现

Python内置的sorted()函数配合lambda表达式可轻松实现：

def sort_by_price(products, ascending=True):
    """
    按价格排序商品
    :param products: 商品列表
    :param ascending: 是否升序排列，默认为True
    :return: 排序后的商品列表
    """
    return sorted(products, key=lambda x: x['price'], reverse=not ascending)

多字段排序扩展：

def sort_by_multiple_fields(products, price_asc=True, name_asc=True):
    """
    多字段排序（价格+名称）
    """
    return sorted(products, 
                 key=lambda x: (x['price'], x['name']),
                 reverse=(not price_asc or not name_asc))

二、进阶优化方案

2.1 使用Pandas处理大数据

当数据量超过10万条时，建议使用Pandas提升性能：

import pandas as pd
def pandas_filter_sort(products_df, min_price, max_price, ascending=True):
    """
    Pandas版本的价格筛选与排序
    :param products_df: Pandas DataFrame，包含'name'和'price'列
    :return: 处理后的DataFrame
    """
    # 筛选
    filtered = products_df[(products_df['price'] >= min_price) & 
                          (products_df['price'] <= max_price)]
    # 排序
    return filtered.sort_values('price', ascending=ascending)

性能对比测试：

# 生成100万条测试数据
import random
data = [{'name': f'商品{i}', 'price': round(random.uniform(100, 10000), 2)} 
        for i in range(1000000)]
# 转换为DataFrame
df = pd.DataFrame(data)
# 传统方法耗时测试
%timeit filter_by_price_range(data, 1000, 5000)  # 约1.2秒
# Pandas方法耗时测试
%timeit pandas_filter_sort(df, 1000, 5000)  # 约0.08秒

2.2 数据库查询优化

对于数据库存储的商品数据，推荐使用SQL的BETWEEN和ORDER BY：

import sqlite3
def db_filter_sort(db_path, min_price, max_price, ascending=True):
    """
    数据库查询实现
    """
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    order = 'ASC' if ascending else 'DESC'
    query = f"""
    SELECT * FROM products 
    WHERE price BETWEEN {min_price} AND {max_price}
    ORDER BY price {order}
    """
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()
    # 转换为字典列表
    return [{'name': row[0], 'price': row[1]} for row in results]

三、工程化实现建议

3.1 类封装实现

class ProductManager:
    def __init__(self, products):
        self.products = products
    def filter_price_range(self, min_price, max_price):
        """价格区间筛选"""
        return [p for p in self.products if min_price <= p['price'] <= max_price]
    def sort_price(self, ascending=True):
        """价格排序"""
        self.products = sorted(self.products, 
                              key=lambda x: x['price'],
                              reverse=not ascending)
        return self.products
    def get_stats(self):
        """获取价格统计信息"""
        prices = [p['price'] for p in self.products]
        return {
            'min': min(prices),
            'max': max(prices),
            'avg': sum(prices)/len(prices),
            'count': len(prices)
        }

3.2 异常处理机制

def safe_filter_sort(products, min_price=None, max_price=None, ascending=True):
    """
    带异常处理的安全版本
    """
    try:
        # 参数校验
        if min_price is not None and max_price is not None and min_price > max_price:
            raise ValueError("最小价格不能大于最大价格")
        # 筛选逻辑
        filtered = products
        if min_price is not None:
            filtered = [p for p in filtered if p['price'] >= min_price]
        if max_price is not None:
            filtered = [p for p in filtered if p['price'] <= max_price]
        # 排序逻辑
        return sorted(filtered, key=lambda x: x['price'], reverse=not ascending)
    except Exception as e:
        print(f"处理失败: {str(e)}")
        return []

四、实际应用场景

4.1 电商系统实现

class ECommerceSystem:
    def __init__(self):
        self.products = []
    def add_product(self, name, price):
        self.products.append({'name': name, 'price': price})
    def search_products(self, keyword=None, min_price=None, max_price=None, sort_by='price'):
        """综合搜索功能"""
        # 1. 关键词筛选（简化版）
        results = self.products
        if keyword:
            results = [p for p in results if keyword.lower() in p['name'].lower()]
        # 2. 价格区间筛选
        if min_price is not None or max_price is not None:
            results = [p for p in results if 
                      (min_price is None or p['price'] >= min_price) and
                      (max_price is None or p['price'] <= max_price)]
        # 3. 排序
        if sort_by == 'price':
            results = sorted(results, key=lambda x: x['price'])
        elif sort_by == 'name':
            results = sorted(results, key=lambda x: x['name'])
        return results

4.2 数据分析报表

def generate_price_report(products, bins=[0, 500, 1000, 2000, 5000, 10000]):
    """生成价格区间分布报表"""
    import numpy as np
    prices = [p['price'] for p in products]
    counts, _ = np.histogram(prices, bins=bins)
    report = []
    for i in range(len(bins)-1):
        lower = bins[i]
        upper = bins[i+1]
        count = counts[i]
        report.append({
            'range': f'{lower}-{upper}',
            'count': count,
            'percentage': round(count/len(prices)*100, 2)
        })
    return report

五、性能优化技巧

1. 索引优化：对频繁查询的价格字段建立索引
2. 批量处理：对于百万级数据，采用分批处理策略
3. 缓存机制：对常用查询结果进行缓存
4. 并行计算：使用多进程处理超大数据集

from multiprocessing import Pool
def parallel_filter(products_chunk, min_price, max_price):
    return [p for p in products_chunk if min_price <= p['price'] <= max_price]
def parallel_filter_sort(products, min_price, max_price, workers=4):
    """并行处理版本"""
    chunk_size = len(products) // workers
    chunks = [products[i:i + chunk_size] for i in range(0, len(products), chunk_size)]
    with Pool(workers) as pool:
        filtered_chunks = pool.starmap(parallel_filter, 
                                     [(chunk, min_price, max_price) for chunk in chunks])
    # 合并结果并排序
    merged = []
    for chunk in filtered_chunks:
        merged.extend(chunk)
    return sorted(merged, key=lambda x: x['price'])

结论

本文系统阐述了Python实现价格区间筛选和排序的完整方案，从基础实现到工程化优化，覆盖了：

1. 基础列表推导式实现
2. Pandas大数据处理方案
3. 数据库查询优化
4. 类封装与异常处理
5. 实际应用场景示例
6. 性能优化技巧

根据实际业务场景，开发者可以选择最适合的实现方式。对于小型应用，基础实现足够；对于中大型系统，建议采用Pandas或数据库方案；对于高并发场景，则需要考虑并行计算和缓存优化。