def manual_counter(items):
    """手動(dòng)計(jì)數(shù)實(shí)現(xiàn)"""
    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("元素計(jì)數(shù):", counts)
max_item = max(counts, key=counts.get)
print(f"出現(xiàn)次數(shù)最多的元素: {max_item} (出現(xiàn){counts[max_item]}次)")

from collections import Counter

# 基礎(chǔ)使用
data = ['a', 'b', 'c', 'a', 'b', 'a', 'd']
counter = Counter(data)

print("計(jì)數(shù)結(jié)果:", counter)
print("出現(xiàn)次數(shù)最多的元素:", counter.most_common(1)[0][0])

import timeit

# 測試數(shù)據(jù)
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"手動(dòng)計(jì)數(shù)耗時(shí): {manual_time:.4f}秒")
print(f"Counter計(jì)數(shù)耗時(shí): {counter_time:.4f}秒")
print(f"Counter效率提升: {(manual_time/counter_time):.1f}倍")

# 帶權(quán)重的計(jì)數(shù)
def weighted_counter(items, weights):
    """帶權(quán)重的元素計(jì)數(shù)"""
    if len(items) != len(weights):
        raise ValueError("項(xiàng)目和權(quán)重長度不一致")
    
    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]  # 銷售數(shù)量

weighted_counts = weighted_counter(products, sales)
print("加權(quán)計(jì)數(shù)結(jié)果:", weighted_counts.most_common())

class TimeDecayCounter:
    """時(shí)間衰減計(jì)數(shù)器"""
    
    def __init__(self, decay_rate=0.9):
        self.counter = Counter()
        self.decay_rate = decay_rate
    
    def add(self, item, timestamp=None):
        """添加元素"""
        # 應(yīng)用衰減
        self._apply_decay()
        self.counter[item] += 1
    
    def _apply_decay(self):
        """應(yīng)用時(shí)間衰減"""
        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:
    """多維度計(jì)數(shù)器"""
    
    def __init__(self):
        self.dimensions = {}
    
    def add(self, *dimension_values):
        """添加多維元素"""
        if len(dimension_values) not in self.dimensions:
            self.dimensions[len(dimension_values)] = Counter()
        
        # 創(chuàng)建復(fù)合鍵
        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')  # 重復(fù)

print("所有維度高頻組合:", user_actions.most_common(3))
print("二維組合高頻:", user_actions.most_common(2, dimension=2))

def chunked_counter(data, chunk_size=10000):
    """分塊計(jì)數(shù)處理大型數(shù)據(jù)集"""
    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

# 生成大型數(shù)據(jù)集
big_data = ['item_' + str(i % 10000) for i in range(1000000)]

# 分塊計(jì)數(shù)
counter = chunked_counter(big_data)
print(f"高頻元素: {counter.most_common(1)[0][0]} (出現(xiàn){counter.most_common(1)[0][1]}次)")

import mmh3  # MurmurHash庫

class CountMinSketch:
    """Count-Min Sketch概率計(jì)數(shù)"""
    
    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):
        """估計(jì)元素頻率"""
        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):
        """估計(jì)高頻元素（需額外存儲鍵）"""
        # 實(shí)際實(shí)現(xiàn)需要跟蹤鍵
        raise NotImplementedError("完整實(shí)現(xiàn)需要鍵跟蹤")

# 使用示例
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'估計(jì)頻率:", cms.estimate('sample'))

def memory_efficient_counter(items):
    """內(nèi)存優(yōu)化的計(jì)數(shù)器"""
    from collections import defaultdict
    import array
    
    # 使用數(shù)組存儲計(jì)數(shù)
    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)
    
    # 重建結(jié)果
    result = {}
    for item, idx in index_map.items():
        result[item] = counts[idx]
    
    return result

# 內(nèi)存對比
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內(nèi)存占用: {mem1/1024:.1f}KB")
print(f"優(yōu)化計(jì)數(shù)器內(nèi)存: {mem2/1024:.1f}KB")

class StreamingCounter:
    """實(shí)時(shí)流計(jì)數(shù)器"""
    
    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
        
        # 定期清理低頻項(xiàng)
        if len(self.counter) > self.capacity * 1.5:
            self._prune()
    
    def _prune(self):
        """清理低頻元素"""
        # 計(jì)算閾值（保留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)

# 模擬數(shù)據(jù)流
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} 項(xiàng)后高頻元素: {stream_counter.most_common(1)}")

class SlidingWindowCounter:
    """滑動(dòng)窗口計(jì)數(shù)器"""
    
    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秒窗口

# 添加帶時(shí)間戳的元素
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"時(shí)間 {ts-current_time:.1f}s: 高頻元素 {window_counter.most_common(1)}")

from multiprocessing import Pool
from collections import Counter

def map_function(chunk):
    """Map階段：局部計(jì)數(shù)"""
    local_counter = Counter(chunk)
    return local_counter.items()

def reduce_function(mapped_results):
    """Reduce階段：合并計(jì)數(shù)"""
    total_counter = Counter()
    for result in mapped_results:
        total_counter.update(dict(result))
    return total_counter

def mapreduce_counter(data, workers=4):
    """MapReduce計(jì)數(shù)框架"""
    # 分塊數(shù)據(jù)
    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("分布式計(jì)數(shù)結(jié)果:", counter.most_common(2))

import redis
import hashlib

class RedisCounter:
    """基于Redis的分布式計(jì)數(shù)器"""
    
    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):
        """哈希元素以節(jié)省空間"""
        return hashlib.md5(str(item).encode()).hexdigest()
    
    def commit(self):
        """提交批量操作"""
        self.pipeline.execute()
    
    def most_common(self, n=1):
        """獲取高頻元素"""
        # 注意：此實(shí)現(xiàn)需要額外映射哈希到原始值
        # 實(shí)際應(yīng)用需要維護(hù)映射關(guān)系
        keys = self.redis.keys(f"{self.namespace}:*")
        counts = self.redis.mget(keys)
        
        items_counts = []
        for key, count in zip(keys, counts):
            # 此處應(yīng)使用映射表獲取原始值
            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:
    """電商熱門商品分析系統(tǒng)"""
    
    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  # 購買權(quán)重更高
            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:
    """日志錯(cuò)誤分析系統(tǒng)"""
    
    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):
        """獲取高頻錯(cuò)誤"""
        return self.error_counter.most_common(n)
    
    def get_error_context(self, error_type):
        """獲取錯(cuò)誤上下文"""
        return self.error_contexts.get(error_type, [])
    
    def generate_report(self):
        """生成錯(cuò)誤報(bào)告"""
        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("錯(cuò)誤報(bào)告:")
for item in log_analyzer.generate_report():
    print(f"- {item['error_type']}: {item['count']}次, 最后出現(xiàn): {item['last_occurrence']}")

class DNAAnalyzer:
    """DNA序列分析系統(tǒng)"""
    
    def __init__(self, k=3):
        self.k = k  # k-mer長度
        self.kmer_counter = Counter()
        self.sequence_counter = Counter()
    
    def process_sequence(self, sequence):
        """處理DNA序列"""
        # 計(jì)數(shù)完整序列
        self.sequence_counter[sequence] += 1
        
        # 計(jì)數(shù)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):
        """發(fā)現(xiàn)異常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",  # 重復(fù)
    "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):
    """內(nèi)存優(yōu)化的計(jì)數(shù)器"""
    from array import array
    import itertools
    
    # 使用排序分組計(jì)數(shù)
    sorted_items = sorted(items)
    groups = itertools.groupby(sorted_items)
    
    # 使用數(shù)組存儲結(jié)果
    keys = []
    counts = array('I')  # 無符號整型
    
    for key, group in groups:
        keys.append(key)
        counts.append(sum(1 for _ in group))
    
    return dict(zip(keys, counts))

# 內(nèi)存對比
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內(nèi)存: {mem_counter/1024:.1f}KB")
print(f"優(yōu)化方法內(nèi)存: {mem_optimized/1024:.1f}KB")

from concurrent.futures import ThreadPoolExecutor
from collections import Counter

def parallel_counter(data, workers=4):
    """并行計(jì)數(shù)器"""
    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:
        # 并行計(jì)數(shù)
        futures = [executor.submit(Counter, chunk) for chunk in chunks]
        
        # 合并結(jié)果
        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"串行計(jì)數(shù)耗時(shí): {t_seq:.2f}秒")
print(f"并行計(jì)數(shù)耗時(shí): {t_par:.2f}秒")
print(f"加速比: {t_seq/t_par:.1f}x")