全面解析Python如何高效查找最大/最小N个元素

import heapq

# 创建堆
data = [5, 7, 9, 1, 3]
heapq.heapify(data)  # 线性时间建堆

# 添加元素
heapq.heappush(data, 4)

# 弹出最小值
min_val = heapq.heappop(data)

# 获取top n
largest = heapq.nlargest(3, data)
smallest = heapq.nsmallest(3, data)

class minheap:
    """最小堆实现"""
    
    def __init__(self):
        self.heap = []
    
    def push(self, item):
        """添加元素"""
        heapq.heappush(self.heap, item)
    
    def pop(self):
        """弹出最小值"""
        return heapq.heappop(self.heap)
    
    def pushpop(self, item):
        """添加并弹出最小值"""
        return heapq.heappushpop(self.heap, item)
    
    def replace(self, item):
        """弹出最小值并添加新元素"""
        return heapq.heapreplace(self.heap, item)
    
    def top_k(self, k):
        """获取最小的k个元素"""
        return heapq.nsmallest(k, self.heap)
    
    def __len__(self):
        return len(self.heap)

# 使用示例
heap = minheap()
for num in [10, 2, 8, 5, 3]:
    heap.push(num)
print(f"最小3个元素: {heap.top_k(3)}")  # [2, 3, 5]

import random

def quickselect(arr, k):
    """快速选择算法 - 查找第k小的元素"""
    if len(arr) == 1:
        return arr[0]
    
    pivot = random.choice(arr)
    lows = [x for x in arr if x < pivot]
    highs = [x for x in arr if x > pivot]
    pivots = [x for x in arr if x == pivot]
    
    if k < len(lows):
        return quickselect(lows, k)
    elif k < len(lows) + len(pivots):
        return pivots[0]
    else:
        return quickselect(highs, k - len(lows) - len(pivots))

def top_k(arr, k):
    """获取最小的k个元素"""
    kth_smallest = quickselect(arr, k-1)
    return sorted([x for x in arr if x <= kth_smallest])[:k]

# 使用示例
data = [random.randint(1, 1000) for _ in range(1000000)]
top_100 = top_k(data, 100)
print(f"最小的100个数: {top_100[:10]}...")

def distributed_top_k(data, k, chunk_size=1000000):
    """分布式top k查找"""
    chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]
    
    # 第一阶段：每个分块找到top k
    chunk_top_k = []
    for chunk in chunks:
        heapq.heapify(chunk)
        chunk_top_k.append(heapq.nsmallest(k, chunk))
    
    # 第二阶段：合并所有分块的top k
    merged = []
    for top in chunk_top_k:
        merged.extend(top)
    
    heapq.heapify(merged)
    return heapq.nsmallest(k, merged)

# 10亿数据查找top 100
big_data = [random.random() for _ in range(10**9)]
top_100 = distributed_top_k(big_data, 100)

from concurrent.futures import processpoolexecutor

def parallel_top_k(data, k, workers=8):
    """并行top k查找"""
    chunk_size = len(data) // workers
    chunks = [data[i*chunk_size:(i+1)*chunk_size] for i in range(workers)]
    
    with processpoolexecutor(max_workers=workers) as executor:
        # 并行处理每个分块
        futures = [executor.submit(heapq.nsmallest, k, chunk) for chunk in chunks]
        
        # 收集结果
        results = []
        for future in futures:
            results.extend(future.result())
    
    # 合并结果
    return heapq.nsmallest(k, results)

# 使用示例
import numpy as np
large_data = np.random.uniform(0, 100, 100000000)  # 1亿个随机数
top_100 = parallel_top_k(large_data, 100, workers=8)

class product:
    def __init__(self, id, name, price, sales):
        self.id = id
        self.name = name
        self.price = price
        self.sales = sales
    
    def __repr__(self):
        return f"{self.name} (¥{self.price}, 销量:{self.sales})"

# 创建产品列表
products = [
    product(1, "iphone 15", 8999, 12000),
    product(2, "ipad pro", 6999, 8500),
    product(3, "macbook air", 10999, 6500),
    product(4, "apple watch", 2999, 15000),
    product(5, "airpods pro", 1999, 28000)
]

# 查找最畅销的3个产品
top_selling = heapq.nlargest(3, products, key=lambda p: p.sales)
print("最畅销产品:")
for p in top_selling:
    print(f"- {p}")

# 查找最贵的2个产品
most_expensive = heapq.nlargest(2, products, key=lambda p: p.price)
print("\n最贵产品:")
for p in most_expensive:
    print(f"- {p}")

def top_k_complex(items, k, key_func):
    """多条件top k查找"""
    # 创建堆
    heap = []
    for item in items:
        # 计算排序键
        key = key_func(item)
        
        # 维护大小为k的堆
        if len(heap) < k:
            heapq.heappush(heap, (key, item))
        elif key > heap[0][0]:
            heapq.heapreplace(heap, (key, item))
    
    # 提取结果
    return [item for _, item in sorted(heap, reverse=true)]

# 使用示例：查找性价比最高的产品（销量/价格）
best_value = top_k_complex(
    products, 
    k=3, 
    key_func=lambda p: p.sales / p.price
)

print("\n性价比最高产品:")
for p in best_value:
    value = p.sales / p.price
    print(f"- {p.name}: ¥{p.price}, 销量:{p.sales}, 性价比:{value:.2f}")

class streamingtopk:
    """实时top k维护系统"""
    
    def __init__(self, k):
        self.k = k
        self.heap = []  # 最小堆维护当前top k
    
    def add(self, item, value):
        """添加新元素"""
        # 使用负值构建最小堆模拟最大堆
        entry = (-value, item)
        
        if len(self.heap) < self.k:
            heapq.heappush(self.heap, entry)
        elif entry > self.heap[0]:
            heapq.heapreplace(self.heap, entry)
    
    def get_top_k(self):
        """获取当前top k"""
        return [item for value, item in sorted(self.heap, reverse=true)]

# 使用示例
stream_processor = streamingtopk(k=3)

# 模拟数据流
data_stream = [
    ("a", 15), ("b", 20), ("c", 10), 
    ("d", 25), ("e", 18), ("f", 30)
]

for item, value in data_stream:
    stream_processor.add(item, value)
    print(f"添加 {item}={value} 后top 3: {stream_processor.get_top_k()}")

from collections import deque
import heapq
import time

class timewindowtopk:
    """时间窗口top k维护"""
    
    def __init__(self, k, window_size):
        """
        :param k: top k数量
        :param window_size: 时间窗口大小（秒）
        """
        self.k = k
        self.window_size = window_size
        self.data = deque()  # (timestamp, value, data)
        self.heap = []       # 当前top k
    
    def add(self, value, data):
        """添加新数据点"""
        now = time.time()
        self.data.append((now, value, data))
        
        # 维护时间窗口
        while self.data and now - self.data[0][0] > self.window_size:
            self.data.popleft()
        
        # 重建堆
        self._rebuild_heap()
    
    def _rebuild_heap(self):
        """重建top k堆"""
        self.heap = []
        for timestamp, value, data in self.data:
            if len(self.heap) < self.k:
                heapq.heappush(self.heap, (value, data))
            elif value > self.heap[0][0]:
                heapq.heapreplace(self.heap, (value, data))
    
    def get_top_k(self):
        """获取当前top k"""
        return sorted(self.heap, reverse=true)

# 使用示例
window_topk = timewindowtopk(k=3, window_size=10)

# 添加数据
window_topk.add(15, "event a")
time.sleep(1)
window_topk.add(20, "event b")
time.sleep(1)
window_topk.add(10, "event c")
print(f"当前top 3: {window_topk.get_top_k()}")

# 添加新数据
time.sleep(3)
window_topk.add(25, "event d")
print(f"添加后top 3: {window_topk.get_top_k()}")

# 等待窗口滑动
time.sleep(8)
print(f"窗口滑动后top 3: {window_topk.get_top_k()}")

class stockanalyzer:
    """股票交易分析系统"""
    
    def __init__(self, k=10):
        self.top_gainers = []  # 最大涨幅
        self.top_losers = []   # 最大跌幅
        self.top_volume = []   # 最高交易量
        self.k = k
    
    def process_trades(self, trades):
        """处理交易数据"""
        for trade in trades:
            # 计算涨跌幅
            change = (trade['price'] - trade['prev_close']) / trade['prev_close'] * 100
            
            # 更新最大涨幅
            self._update_heap(self.top_gainers, change, trade, max_heap=true)
            
            # 更新最大跌幅
            self._update_heap(self.top_losers, -change, trade, max_heap=true)
            
            # 更新最高交易量
            self._update_heap(self.top_volume, trade['volume'], trade, max_heap=true)
    
    def _update_heap(self, heap, value, data, max_heap=true):
        """更新堆状态"""
        # 使用负值转换最大堆为最小堆
        key = value if max_heap else -value
        entry = (key, data)
        
        if len(heap) < self.k:
            heapq.heappush(heap, entry)
        elif key > heap[0][0]:
            heapq.heapreplace(heap, entry)
    
    def get_top_gainers(self):
        """获取涨幅最大的股票"""
        return [data for _, data in sorted(self.top_gainers, reverse=true)]
    
    def get_top_losers(self):
        """获取跌幅最大的股票"""
        return [data for _, data in sorted(self.top_losers, reverse=true)]
    
    def get_top_volume(self):
        """获取交易量最大的股票"""
        return [data for _, data in sorted(self.top_volume, reverse=true)]

# 使用示例
trades = [
    {'symbol': 'aapl', 'price': 185.5, 'prev_close': 182.3, 'volume': 1000000},
    {'symbol': 'msft', 'price': 340.2, 'prev_close': 345.6, 'volume': 850000},
    {'symbol': 'googl', 'price': 135.7, 'prev_close': 132.5, 'volume': 1200000},
    # ...更多交易数据
]

analyzer = stockanalyzer(k=5)
analyzer.process_trades(trades)

print("涨幅top 5:")
for stock in analyzer.get_top_gainers():
    print(f"{stock['symbol']}: {stock['price']}")

print("\n交易量top 5:")
for stock in analyzer.get_top_volume():
    print(f"{stock['symbol']}: {stock['volume']}")

class recommendersystem:
    """实时推荐系统"""
    
    def __init__(self, k=10):
        self.user_preferences = {}  # 用户偏好向量
        self.item_features = {}     # 物品特征向量
        self.k = k
    
    def update_user_preference(self, user_id, item_id, rating):
        """更新用户偏好"""
        if user_id not in self.user_preferences:
            self.user_preferences[user_id] = {}
        self.user_preferences[user_id][item_id] = rating
    
    def update_item_features(self, item_id, features):
        """更新物品特征"""
        self.item_features[item_id] = features
    
    def recommend(self, user_id, n=10):
        """为用户生成推荐"""
        if user_id not in self.user_preferences:
            return []
        
        # 获取用户评分过的物品
        user_ratings = self.user_preferences[user_id]
        
        # 计算未评分物品的预测评分
        scores = []
        for item_id, features in self.item_features.items():
            if item_id not in user_ratings:
                # 简化计算：实际中应使用更复杂的预测模型
                score = sum(
                    user_ratings.get(other_item, 0) * self._similarity(features, self.item_features[other_item])
                    for other_item in user_ratings
                )
                scores.append((score, item_id))
        
        # 获取top n推荐
        return heapq.nlargest(n, scores, key=lambda x: x[0])
    
    def _similarity(self, features1, features2):
        """计算特征相似度（简化版）"""
        # 实际应用中应使用余弦相似度等
        return sum(a * b for a, b in zip(features1, features2))

# 使用示例
recommender = recommendersystem()

# 添加物品特征
recommender.update_item_features("item1", [0.8, 0.2, 0.5])
recommender.update_item_features("item2", [0.6, 0.3, 0.7])
# ...添加更多物品

# 更新用户评分
recommender.update_user_preference("user1", "item1", 5)
recommender.update_user_preference("user1", "item2", 4)
# ...添加更多评分

# 生成推荐
recommendations = recommender.recommend("user1", n=5)
print("推荐物品:")
for score, item_id in recommendations:
    print(f"- {item_id} (预测评分: {score:.2f})")

class loganalyzer:
    """日志分析系统"""
    
    def __init__(self, k=10):
        self.error_counter = {}  # 错误计数
        self.slow_requests = []  # 慢请求
        self.k = k
    
    def process_log(self, log_entry):
        """处理日志条目"""
        # 错误日志统计
        if log_entry['level'] == 'error':
            error_type = log_entry['error_type']
            self.error_counter[error_type] = self.error_counter.get(error_type, 0) + 1
        
        # 慢请求记录
        if 'response_time' in log_entry and log_entry['response_time'] > 1000:
            self._update_heap(
                self.slow_requests, 
                log_entry['response_time'], 
                log_entry
            )
    
    def _update_heap(self, heap, value, data):
        """更新堆状态"""
        entry = (value, data)
        if len(heap) < self.k:
            heapq.heappush(heap, entry)
        elif value > heap[0][0]:
            heapq.heapreplace(heap, entry)
    
    def top_errors(self, k=none):
        """获取top k错误类型"""
        k = k or self.k
        return heapq.nlargest(k, self.error_counter.items(), key=lambda x: x[1])
    
    def top_slow_requests(self, k=none):
        """获取top k慢请求"""
        k = k or self.k
        return heapq.nlargest(k, self.slow_requests, key=lambda x: x[0])

# 使用示例
logs = [
    {'level': 'info', 'message': 'request received'},
    {'level': 'error', 'error_type': 'timeout', 'message': 'request timeout'},
    {'level': 'error', 'error_type': 'dberror', 'message': 'database connection failed'},
    {'level': 'info', 'response_time': 1200, 'endpoint': '/api/users'},
    # ...更多日志
]

analyzer = loganalyzer(k=5)
for log in logs:
    analyzer.process_log(log)

print("top 5错误类型:")
for error, count in analyzer.top_errors():
    print(f"- {error}: {count}次")

print("\ntop 5慢请求:")
for time, log in analyzer.top_slow_requests():
    print(f"- {log['endpoint']}: {time}ms")

def memory_efficient_top_k(data, k, chunk_size=1000000):
    """内存优化的top k查找"""
    # 初始化堆
    heap = []
    
    # 分块处理
    for i in range(0, len(data), chunk_size):
        chunk = data[i:i+chunk_size]
        
        # 处理当前分块
        for value in chunk:
            if len(heap) < k:
                heapq.heappush(heap, value)
            elif value > heap[0]:
                heapq.heapreplace(heap, value)
    
    return sorted(heap, reverse=true)

# 使用示例：处理10亿数据只需o(k)内存
big_data = (random.random() for _ in range(10**9))  # 生成器表达式减少内存
top_100 = memory_efficient_top_k(big_data, 100)

class multiindextopk:
    """多维度top k索引系统"""
    
    def __init__(self, k=10):
        self.k = k
        self.heaps = {
            'price': [],   # 价格最高
            'sales': [],   # 销量最高
            'rating': []   # 评分最高
        }
        self.data = {}     # 存储完整数据
    
    def add_item(self, item_id, price, sales, rating):
        """添加商品"""
        self.data[item_id] = {'price': price, 'sales': sales, 'rating': rating}
        
        # 更新各维度堆
        self._update_heap('price', price, item_id)
        self._update_heap('sales', sales, item_id)
        self._update_heap('rating', rating, item_id)
    
    def _update_heap(self, dimension, value, item_id):
        """更新指定维度堆"""
        heap = self.heaps[dimension]
        entry = (value, item_id)
        
        if len(heap) < self.k:
            heapq.heappush(heap, entry)
        elif value > heap[0][0]:
            heapq.heapreplace(heap, entry)
    
    def get_top_k(self, dimension):
        """获取指定维度top k"""
        return sorted(self.heaps[dimension], reverse=true)

# 使用示例
index = multiindextopk(k=3)
index.add_item("a", price=100, sales=500, rating=4.5)
index.add_item("b", price=200, sales=300, rating=4.8)
index.add_item("c", price=150, sales=400, rating=4.2)
index.add_item("d", price=250, sales=200, rating=4.9)

print("价格top 3:")
for price, item_id in index.get_top_k('price'):
    print(f"- {item_id}: ¥{price}")

print("\n销量top 3:")
for sales, item_id in index.get_top_k('sales'):
    print(f"- {item_id}: {sales}件")