Python如何高效找出序列中出现次数最多的元素

def manual_counter(items):
    """手动计数实现"""
    counts = {}
    for item in items:
        counts[item] = counts.get(item, 0) + 1
    return counts

# 使用示例
data = ['apple', 'banana', 'apple', 'orange', 'banana', 'apple']
counts = manual_counter(data)
print("元素计数:", counts)
max_item = max(counts, key=counts.get)
print(f"出现次数最多的元素: {max_item} (出现{counts[max_item]}次)")

from collections import counter

# 基础使用
data = ['a', 'b', 'c', 'a', 'b', 'a', 'd']
counter = counter(data)

print("计数结果:", counter)
print("出现次数最多的元素:", counter.most_common(1)[0][0])

import timeit

# 测试数据
large_data = ['item_' + str(i % 1000) for i in range(1000000)]

# 性能测试
manual_time = timeit.timeit(
    lambda: manual_counter(large_data), 
    number=1
)
counter_time = timeit.timeit(
    lambda: counter(large_data), 
    number=1
)

print(f"手动计数耗时: {manual_time:.4f}秒")
print(f"counter计数耗时: {counter_time:.4f}秒")
print(f"counter效率提升: {(manual_time/counter_time):.1f}倍")

# 带权重的计数
def weighted_counter(items, weights):
    """带权重的元素计数"""
    if len(items) != len(weights):
        raise valueerror("项目和权重长度不一致")
    
    counter = counter()
    for item, weight in zip(items, weights):
        counter[item] += weight
    return counter

# 使用示例
products = ['apple', 'banana', 'apple', 'orange', 'banana']
sales = [10, 5, 8, 3, 7]  # 销售数量

weighted_counts = weighted_counter(products, sales)
print("加权计数结果:", weighted_counts.most_common())

class timedecaycounter:
    """时间衰减计数器"""
    
    def __init__(self, decay_rate=0.9):
        self.counter = counter()
        self.decay_rate = decay_rate
    
    def add(self, item, timestamp=none):
        """添加元素"""
        # 应用衰减
        self._apply_decay()
        self.counter[item] += 1
    
    def _apply_decay(self):
        """应用时间衰减"""
        for item in list(self.counter.keys()):
            self.counter[item] *= self.decay_rate
            if self.counter[item] < 0.001:  # 阈值清理
                del self.counter[item]
    
    def most_common(self, n=none):
        """获取高频元素"""
        return self.counter.most_common(n)

# 使用示例
decay_counter = timedecaycounter(decay_rate=0.95)

# 模拟事件流
events = ['login', 'search', 'purchase', 'search', 'login', 'logout']
for event in events:
    decay_counter.add(event)
    print(f"添加 '{event}' 后: {decay_counter.most_common(3)}")

class multidimensionalcounter:
    """多维度计数器"""
    
    def __init__(self):
        self.dimensions = {}
    
    def add(self, *dimension_values):
        """添加多维元素"""
        if len(dimension_values) not in self.dimensions:
            self.dimensions[len(dimension_values)] = counter()
        
        # 创建复合键
        composite_key = tuple(dimension_values)
        self.dimensions[len(dimension_values)][composite_key] += 1
    
    def most_common(self, n=1, dimension=none):
        """获取高频组合"""
        if dimension is none:
            # 返回所有维度中最常见的
            all_counts = counter()
            for counter in self.dimensions.values():
                all_counts.update(counter)
            return all_counts.most_common(n)
        else:
            return self.dimensions.get(dimension, counter()).most_common(n)

# 使用示例
user_actions = multidimensionalcounter()

# 添加用户行为 (用户id, 操作类型, 页面)
user_actions.add('user1', 'click', 'home')
user_actions.add('user2', 'view', 'product')
user_actions.add('user1', 'click', 'cart')
user_actions.add('user1', 'click', 'home')  # 重复

print("所有维度高频组合:", user_actions.most_common(3))
print("二维组合高频:", user_actions.most_common(2, dimension=2))

def chunked_counter(data, chunk_size=10000):
    """分块计数处理大型数据集"""
    total_counter = counter()
    
    for i in range(0, len(data), chunk_size):
        chunk = data[i:i+chunk_size]
        total_counter.update(chunk)
    
    return total_counter

# 生成大型数据集
big_data = ['item_' + str(i % 10000) for i in range(1000000)]

# 分块计数
counter = chunked_counter(big_data)
print(f"高频元素: {counter.most_common(1)[0][0]} (出现{counter.most_common(1)[0][1]}次)")

import mmh3  # murmurhash库

class countminsketch:
    """count-min sketch概率计数"""
    
    def __init__(self, width=1000, depth=5):
        self.width = width
        self.depth = depth
        self.counts = [[0] * width for _ in range(depth)]
        self.seeds = [i * 1000 for i in range(depth)]
    
    def add(self, item):
        """添加元素"""
        for i in range(self.depth):
            index = mmh3.hash(item, self.seeds[i]) % self.width
            self.counts[i][index] += 1
    
    def estimate(self, item):
        """估计元素频率"""
        min_count = float('inf')
        for i in range(self.depth):
            index = mmh3.hash(item, self.seeds[i]) % self.width
            if self.counts[i][index] < min_count:
                min_count = self.counts[i][index]
        return min_count
    
    def most_common(self, n=1):
        """估计高频元素（需额外存储键）"""
        # 实际实现需要跟踪键
        raise notimplementederror("完整实现需要键跟踪")

# 使用示例
cms = countminsketch(width=1000, depth=5)

text = "this is a sample text for testing count min sketch algorithm"
words = text.split()

for word in words:
    cms.add(word)

print("'sample'估计频率:", cms.estimate('sample'))

def memory_efficient_counter(items):
    """内存优化的计数器"""
    from collections import defaultdict
    import array
    
    # 使用数组存储计数
    index_map = {}
    counts = array.array('l')  # 无符号长整型
    free_list = []
    
    for item in items:
        if item in index_map:
            idx = index_map[item]
            counts[idx] += 1
        else:
            if free_list:
                idx = free_list.pop()
                index_map[item] = idx
                counts[idx] = 1
            else:
                idx = len(counts)
                index_map[item] = idx
                counts.append(1)
    
    # 重建结果
    result = {}
    for item, idx in index_map.items():
        result[item] = counts[idx]
    
    return result

# 内存对比
import sys
large_data = [str(i % 10000) for i in range(1000000)]

mem1 = sys.getsizeof(counter(large_data))
mem2 = sys.getsizeof(memory_efficient_counter(large_data))

print(f"counter内存占用: {mem1/1024:.1f}kb")
print(f"优化计数器内存: {mem2/1024:.1f}kb")

class streamingcounter:
    """实时流计数器"""
    
    def __init__(self, capacity=1000):
        self.capacity = capacity
        self.counter = counter()
        self.total = 0
    
    def add(self, item):
        """添加元素"""
        self.counter[item] += 1
        self.total += 1
        
        # 定期清理低频项
        if len(self.counter) > self.capacity * 1.5:
            self._prune()
    
    def _prune(self):
        """清理低频元素"""
        # 计算阈值（保留top n）
        threshold = sorted(self.counter.values())[-self.capacity]
        for item in list(self.counter.keys()):
            if self.counter[item] < threshold:
                del self.counter[item]
    
    def most_common(self, n=1):
        """获取高频元素"""
        return self.counter.most_common(n)
    
    def frequency(self, item):
        """获取元素频率"""
        return self.counter.get(item, 0) / self.total if self.total > 0 else 0

# 使用示例
stream_counter = streamingcounter(capacity=100)

# 模拟数据流
import random
items = ['a', 'b', 'c', 'd', 'e']

for i in range(1000):
    item = random.choices(items, weights=[5, 4, 3, 2, 1])[0]
    stream_counter.add(item)
    if i % 100 == 0:
        print(f"处理 {i} 项后高频元素: {stream_counter.most_common(1)}")

class slidingwindowcounter:
    """滑动窗口计数器"""
    
    def __init__(self, window_size=60):
        self.window_size = window_size
        self.window = deque()
        self.counter = counter()
    
    def add(self, item, timestamp=none):
        """添加元素"""
        ts = timestamp or time.time()
        self.window.append((item, ts))
        self.counter[item] += 1
        self._remove_expired(ts)
    
    def _remove_expired(self, current_time):
        """移除过期元素"""
        while self.window and current_time - self.window[0][1] > self.window_size:
            item, _ = self.window.popleft()
            self.counter[item] -= 1
            if self.counter[item] == 0:
                del self.counter[item]
    
    def most_common(self, n=1):
        """获取高频元素"""
        return self.counter.most_common(n)

# 使用示例
window_counter = slidingwindowcounter(window_size=5)  # 5秒窗口

# 添加带时间戳的元素
current_time = time.time()
events = [
    ('a', current_time),
    ('b', current_time + 1),
    ('a', current_time + 2),
    ('c', current_time + 3),
    ('a', current_time + 4),
    ('b', current_time + 6)  # 超出窗口
]

for item, ts in events:
    window_counter.add(item, ts)
    print(f"时间 {ts-current_time:.1f}s: 高频元素 {window_counter.most_common(1)}")

from multiprocessing import pool
from collections import counter

def map_function(chunk):
    """map阶段：局部计数"""
    local_counter = counter(chunk)
    return local_counter.items()

def reduce_function(mapped_results):
    """reduce阶段：合并计数"""
    total_counter = counter()
    for result in mapped_results:
        total_counter.update(dict(result))
    return total_counter

def mapreduce_counter(data, workers=4):
    """mapreduce计数框架"""
    # 分块数据
    chunk_size = len(data) // workers
    chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]
    
    # map阶段
    with pool(workers) as pool:
        mapped = pool.map(map_function, chunks)
    
    # reduce阶段
    return reduce_function(mapped)

# 使用示例
big_data = [random.choice(['a', 'b', 'c', 'd']) for _ in range(1000000)]
counter = mapreduce_counter(big_data, workers=4)
print("分布式计数结果:", counter.most_common(2))

import redis
import hashlib

class rediscounter:
    """基于redis的分布式计数器"""
    
    def __init__(self, host='localhost', port=6379, namespace='counter'):
        self.redis = redis.redis(host=host, port=port)
        self.namespace = namespace
        self.pipeline = self.redis.pipeline()
    
    def add(self, item):
        """添加元素"""
        key = f"{self.namespace}:{self._hash_item(item)}"
        self.pipeline.incr(key)
    
    def _hash_item(self, item):
        """哈希元素以节省空间"""
        return hashlib.md5(str(item).encode()).hexdigest()
    
    def commit(self):
        """提交批量操作"""
        self.pipeline.execute()
    
    def most_common(self, n=1):
        """获取高频元素"""
        # 注意：此实现需要额外映射哈希到原始值
        # 实际应用需要维护映射关系
        keys = self.redis.keys(f"{self.namespace}:*")
        counts = self.redis.mget(keys)
        
        items_counts = []
        for key, count in zip(keys, counts):
            # 此处应使用映射表获取原始值
            item = key.decode().split(':')[-1]
            items_counts.append((item, int(count)))
        
        return sorted(items_counts, key=lambda x: x[1], reverse=true)[:n]

# 使用示例
counter = rediscounter()

# 添加元素
for item in ['a', 'b', 'a', 'c', 'b', 'a']:
    counter.add(item)

counter.commit()
print("高频元素:", counter.most_common(1))

class productanalyzer:
    """电商热门商品分析系统"""
    
    def __init__(self):
        self.product_counter = counter()
        self.category_counter = counter()
        self.user_behavior = defaultdict(counter)
    
    def process_event(self, event):
        """处理用户行为事件"""
        if event['type'] == 'view':
            self.product_counter[event['product_id']] += 1
            self.category_counter[event['category']] += 1
            self.user_behavior[event['user_id']]['view'] += 1
        elif event['type'] == 'purchase':
            self.product_counter[event['product_id']] += 5  # 购买权重更高
            self.category_counter[event['category']] += 3
            self.user_behavior[event['user_id']]['purchase'] += 1
    
    def get_hot_products(self, n=10):
        """获取热门商品"""
        return self.product_counter.most_common(n)
    
    def get_popular_categories(self, n=5):
        """获取热门分类"""
        return self.category_counter.most_common(n)
    
    def get_active_users(self, n=5):
        """获取活跃用户"""
        user_activity = {
            user: sum(actions.values()) 
            for user, actions in self.user_behavior.items()
        }
        return sorted(user_activity.items(), key=lambda x: x[1], reverse=true)[:n]

# 使用示例
analyzer = productanalyzer()

# 模拟事件流
events = [
    {'type': 'view', 'user_id': 'u1', 'product_id': 'p100', 'category': 'electronics'},
    {'type': 'view', 'user_id': 'u2', 'product_id': 'p200', 'category': 'clothing'},
    {'type': 'purchase', 'user_id': 'u1', 'product_id': 'p100', 'category': 'electronics'},
    {'type': 'view', 'user_id': 'u3', 'product_id': 'p100', 'category': 'electronics'},
    {'type': 'view', 'user_id': 'u1', 'product_id': 'p300', 'category': 'books'},
]

for event in events:
    analyzer.process_event(event)

print("热门商品:", analyzer.get_hot_products(3))
print("热门分类:", analyzer.get_popular_categories())
print("活跃用户:", analyzer.get_active_users())

class loganalyzer:
    """日志错误分析系统"""
    
    def __init__(self):
        self.error_counter = counter()
        self.error_contexts = defaultdict(list)
    
    def process_log(self, log_entry):
        """处理日志条目"""
        if log_entry['level'] == 'error':
            error_type = log_entry['error_type']
            self.error_counter[error_type] += 1
            self.error_contexts[error_type].append({
                'timestamp': log_entry['timestamp'],
                'message': log_entry['message'],
                'source': log_entry['source']
            })
    
    def top_errors(self, n=5):
        """获取高频错误"""
        return self.error_counter.most_common(n)
    
    def get_error_context(self, error_type):
        """获取错误上下文"""
        return self.error_contexts.get(error_type, [])
    
    def generate_report(self):
        """生成错误报告"""
        report = []
        for error, count in self.top_errors(10):
            contexts = self.get_error_context(error)
            last_occurrence = max(ctx['timestamp'] for ctx in contexts) if contexts else none
            report.append({
                'error_type': error,
                'count': count,
                'last_occurrence': last_occurrence,
                'sources': counter(ctx['source'] for ctx in contexts)
            })
        return report

# 使用示例
log_analyzer = loganalyzer()

# 模拟日志
logs = [
    {'level': 'info', 'message': 'system started'},
    {'level': 'error', 'error_type': 'dbconnection', 'timestamp': '2023-08-01 10:00', 'source': 'api', 'message': 'failed to connect'},
    {'level': 'error', 'error_type': 'timeout', 'timestamp': '2023-08-01 10:05', 'source': 'worker', 'message': 'request timeout'},
    {'level': 'error', 'error_type': 'dbconnection', 'timestamp': '2023-08-01 11:30', 'source': 'api', 'message': 'failed to connect'},
]

for log in logs:
    log_analyzer.process_log(log)

print("错误报告:")
for item in log_analyzer.generate_report():
    print(f"- {item['error_type']}: {item['count']}次, 最后出现: {item['last_occurrence']}")

class dnaanalyzer:
    """dna序列分析系统"""
    
    def __init__(self, k=3):
        self.k = k  # k-mer长度
        self.kmer_counter = counter()
        self.sequence_counter = counter()
    
    def process_sequence(self, sequence):
        """处理dna序列"""
        # 计数完整序列
        self.sequence_counter[sequence] += 1
        
        # 计数k-mer
        for i in range(len(sequence) - self.k + 1):
            kmer = sequence[i:i+self.k]
            self.kmer_counter[kmer] += 1
    
    def most_common_sequence(self, n=1):
        """获取高频序列"""
        return self.sequence_counter.most_common(n)
    
    def most_common_kmers(self, n=5):
        """获取高频k-mer"""
        return self.kmer_counter.most_common(n)
    
    def find_anomalies(self, threshold=0.01):
        """发现异常k-mer"""
        total = sum(self.kmer_counter.values())
        avg_freq = total / len(self.kmer_counter)
        
        anomalies = []
        for kmer, count in self.kmer_counter.items():
            freq = count / total
            if freq > threshold or freq < avg_freq / 10:
                anomalies.append((kmer, count, freq))
        
        return sorted(anomalies, key=lambda x: x[1], reverse=true)

# 使用示例
dna_analyzer = dnaanalyzer(k=3)

# 模拟dna序列
sequences = [
    "atgcgatagctagctagct",
    "cgatagctagctagctagc",
    "atgcgatagctagctagct",  # 重复
    "ttacgatcgatcgatcga"
]

for seq in sequences:
    dna_analyzer.process_sequence(seq)

print("高频序列:", dna_analyzer.most_common_sequence())
print("高频3-mer:", dna_analyzer.most_common_kmers(3))
print("异常k-mer:", dna_analyzer.find_anomalies(threshold=0.05))

def optimized_counter(items):
    """内存优化的计数器"""
    from array import array
    import itertools
    
    # 使用排序分组计数
    sorted_items = sorted(items)
    groups = itertools.groupby(sorted_items)
    
    # 使用数组存储结果
    keys = []
    counts = array('i')  # 无符号整型
    
    for key, group in groups:
        keys.append(key)
        counts.append(sum(1 for _ in group))
    
    return dict(zip(keys, counts))

# 内存对比
large_data = [str(i % 10000) for i in range(1000000)]

mem_counter = sys.getsizeof(counter(large_data))
mem_optimized = sys.getsizeof(optimized_counter(large_data))

print(f"counter内存: {mem_counter/1024:.1f}kb")
print(f"优化方法内存: {mem_optimized/1024:.1f}kb")

from concurrent.futures import threadpoolexecutor
from collections import counter

def parallel_counter(data, workers=4):
    """并行计数器"""
    chunk_size = len(data) // workers
    chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]
    
    with threadpoolexecutor(max_workers=workers) as executor:
        # 并行计数
        futures = [executor.submit(counter, chunk) for chunk in chunks]
        
        # 合并结果
        total_counter = counter()
        for future in futures:
            total_counter.update(future.result())
    
    return total_counter

# 性能测试
big_data = [random.choice(['a', 'b', 'c', 'd']) for _ in range(10000000)]

t_seq = timeit.timeit(lambda: counter(big_data), number=1)
t_par = timeit.timeit(lambda: parallel_counter(big_data, workers=4), number=1)

print(f"串行计数耗时: {t_seq:.2f}秒")
print(f"并行计数耗时: {t_par:.2f}秒")
print(f"加速比: {t_seq/t_par:.1f}x")