摊还分析与在线算法

import math
from typing import Any, Iterator

class DynamicArray:
    """动态数组实现，演示摊还分析

    时间复杂度（摊还）：
        - append:     O(1) 摊还
        - pop:        O(1) 摊还
        - access:     O(1) 最坏
        - insert(0):  O(n) 摊还
        - delete(0):  O(n) 摊还
    """

    def __init__(self, capacity: int = 1):
        self._capacity = max(1, capacity)
        self._size = 0
        self._data = [None] * self._capacity
        self._copy_count = 0  # 统计元素拷贝次数

    def __len__(self) -> int:
        return self._size

    def __getitem__(self, i: int) -> Any:
        if i < 0:
            i += self._size
        if not 0 <= i < self._size:
            raise IndexError("index out of range")
        return self._data[i]

    def __setitem__(self, i: int, value: Any) -> None:
        if i < 0:
            i += self._size
        if not 0 <= i < self._size:
            raise IndexError("index out of range")
        self._data[i] = value

    def append(self, value: Any) -> None:
        """追加元素，必要时扩容"""
        if self._size == self._capacity:
            self._resize(self._capacity * 2)
        self._data[self._size] = value
        self._size += 1

    def _resize(self, new_capacity: int) -> None:
        """调整容量，统计元素拷贝"""
        new_data = [None] * new_capacity
        for i in range(self._size):
            new_data[i] = self._data[i]
            self._copy_count += 1
        self._data = new_data
        self._capacity = new_capacity

    def pop(self) -> Any:
        """弹出末尾元素，必要时缩容"""
        if self._size == 0:
            raise IndexError("pop from empty array")
        value = self._data[self._size - 1]
        self._data[self._size - 1] = None
        self._size -= 1
        # 缩容：当 size < capacity / 4 时，容量减半
        if self._size < self._capacity // 4 and self._capacity > 1:
            self._resize(max(1, self._capacity // 2))
        return value

    def __iter__(self) -> Iterator:
        for i in range(self._size):
            yield self._data[i]

def verify_amortized_analysis(n: int = 10000) -> None:
    """验证 append 的摊还 O(1) 性质

    聚合分析：n 次 append 总拷贝次数 < 2n
    """
    arr = DynamicArray()
    for i in range(n):
        arr.append(i)

    # 总拷贝次数应 < 2n（理论上 n + n/2 + n/4 + ... = 2n）
    print(f"n = {n}")
    print(f"实际拷贝次数: {arr._copy_count}")
    print(f"理论上界 (2n): {2 * n}")
    print(f"摊还代价 (拷贝次数/n): {arr._copy_count / n:.4f}")
    # 输出示例：摊还代价约 1-2，符合 O(1) 摊还

    # 验证势能函数 Φ(D) = 2*size - capacity 的非负性
    # 当 size > capacity/2 时 Φ > 0；扩容后 size = capacity/2，Φ = 0
    # 操作序列中 Φ 始终非负

verify_amortized_analysis(10000)

import java.util.*;

/**
 * LRU（Least Recently Used）缓存实现
 *
 * 竞争分析（Sleator-Tarjan 1985）：
 *   对容量 k 的缓存，LRU 是 k-竞争的
 *   即对任意请求序列 σ，cost(LRU, σ) ≤ k · cost(OPT, σ) + k
 *
 * 证明思路：
 *   将 σ 划分为 k-相位（k-phase），每个相位至多 k 个不同页面
 *   LRU 每相位至多 k 次未命中；OPT 每相位至少 1 次未命中
 *   故 cost(LRU) / cost(OPT) ≤ k
 */
public class LRUCache<K, V> {
    private final int capacity;
    private final Map<K, V> cache;
    private long hitCount = 0;
    private long missCount = 0;

    public LRUCache(int capacity) {
        this.capacity = capacity;
        // LinkedHashMap 按 access-order 维护，最近访问在末尾
        this.cache = new LinkedHashMap<K, V>(capacity, 0.75f, true) {
            @Override
            protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
                return size() > LRUCache.this.capacity;
            }
        };
    }

    public V get(K key) {
        V value = cache.get(key);
        if (value != null) {
            hitCount++;
            return value;
        }
        missCount++;
        return null;
    }

    public void put(K key, V value) {
        if (!cache.containsKey(key)) {
            missCount++;
        } else {
            hitCount++;
        }
        cache.put(key, value);
    }

    public long getHitCount() { return hitCount; }
    public long getMissCount() { return missCount; }

    public static void main(String[] args) {
        // 模拟 paging 请求序列
        int k = 3;  // 缓存容量
        int[] requests = {1, 2, 3, 4, 1, 2, 5, 1, 2, 3, 4, 5};

        LRUCache<Integer, Boolean> lru = new LRUCache<>(k);
        for (int page : requests) {
            if (lru.get(page) == null) {
                lru.put(page, true);
            }
        }

        System.out.println("LRU 竞争比验证 (k = " + k + ")");
        System.out.println("命中: " + lru.getHitCount());
        System.out.println("未命中: " + lru.getMissCount());
        System.out.println("理论竞争比上界: " + k);
        // 输出：LRU 应是 3-竞争的
    }
}

import numpy as np
import mmh3  # MurmurHash3
from typing import List

class CountMinSketch:
    """Count-Min Sketch 频率估计算法

    空间复杂度：O(d * w) = O((1/ε) · log(1/δ))
    估计误差：|f̂ - f| ≤ ε·||f||_1，概率 ≥ 1 - δ

    应用：
        - 网络流量监控（按 IP 统计包数）
        - 搜索引擎热门查询
        - 推荐系统频率统计
    """

    def __init__(self, epsilon: float, delta: float):
        """
        epsilon: 频率估计误差上界（相对于总流量）
        delta:  估计失败概率
        """
        self.w = int(np.ceil(1 / epsilon))   # 列数
        self.d = int(np.ceil(np.log(1 / delta)))  # 行数（哈希函数数）
        self.table = np.zeros((self.d, self.w), dtype=np.int64)
        # 使用不同 seed 生成 d 个独立哈希函数
        self.seeds = [i * 31 + 17 for i in range(self.d)]

    def update(self, key: str, count: int = 1) -> None:
        """更新 key 的频率估计"""
        for i in range(self.d):
            j = mmh3.hash(key, self.seeds[i], signed=False) % self.w
            self.table[i][j] += count

    def estimate(self, key: str) -> int:
        """估计 key 的频率（上界）"""
        return min(
            self.table[i][mmh3.hash(key, self.seeds[i], signed=False) % self.w]
            for i in range(self.d)
        )

# 测试
cms = CountMinSketch(epsilon=0.01, delta=0.01)
print(f"Sketch 大小: {cms.d} 行 × {cms.w} 列 = {cms.d * cms.w} 计数器")

# 模拟数据流
stream = ["apple", "banana", "apple", "cherry", "apple", "banana"] * 1000
for item in stream:
    cms.update(item)

# 估计频率
for item in ["apple", "banana", "cherry", "grape"]:
    print(f"  {item}: 估计 = {cms.estimate(item)}, 真实 = {stream.count(item)}")
# 输出示例：估计值略大于真实值（Count-Min 总是高估）