性能优化

# Python 3.12+
import timeit

# 对比列表推导式 vs map 的性能
setup = "data = list(range(10_000))"

stmt_listcomp = "[x * 2 for x in data]"
stmt_map = "list(map(lambda x: x * 2, data))"

t1 = timeit.timeit(stmt_listcomp, setup=setup, number=10_000)
t2 = timeit.timeit(stmt_map, setup=setup, number=10_000)

print(f"列表推导式: {t1:.4f}s")
print(f"map+lambda:  {t2:.4f}s")
print(f"列表推导式快 {t2/t1:.1f}x")

# Python 3.12+
import cProfile
import pstats
import io


def fibonacci(n: int) -> int:
    """故意用低效递归来演示 profiling"""
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)


def main() -> None:
    result = fibonacci(30)
    print(f"结果: {result}")


# 方式一：直接运行并输出统计
cProfile.run("main()", sort="cumtime")

# 方式二：保存到文件，后续用 pstats 或可视化工具分析
profiler = cProfile.Profile()
profiler.enable()
main()
profiler.disable()

# 输出到字符串
stream = io.StringIO()
stats = pstats.Stats(profiler, stream=stream)
stats.sort_stats("cumtime")
stats.print_stats(10)  # 只打印前 10 条
print(stream.getvalue())

# 对运行中的进程采样（PID 为 12345）
py-spy top --pid 12345

# 生成火焰图
py-spy record -o flamegraph.svg --pid 12345

# 直接对脚本采样
py-spy record -o profile.svg -- python my_script.py

# Python 3.12+
# 安装: pip install line_profiler


@profile  # 装饰器由 kernprof 在运行时注入，直接运行会报错
def process_data(items: list[dict]) -> list[dict]:
    results = []
    for item in items:  # line_profiler 会显示这行的耗时
        value = item.get("value", 0)
        processed = value**2 + value * 3 - 1
        if processed > 100:
            results.append({"id": item["id"], "result": processed})
    return results


# 使用方式（命令行）:
# kernprof -l -v my_script.py

# Python 3.12+
# 安装: pip install memory_profiler


@profile
def build_large_list() -> list[int]:
    """对比不同方式构建大列表的内存使用"""
    # 方式一：一次性构建
    data = [i * 2 for i in range(1_000_000)]  # 内存峰值高
    total = sum(data)
    del data

    # 方式二：生成器惰性求值
    total2 = sum(i * 2 for i in range(1_000_000))  # 内存峰值低

    return [total, total2]


# 使用方式（命令行）:
# python -m memory_profiler my_script.py

# Python 3.12+
import random

data = [random.randint(0, 100) for _ in range(100_000)]

# 差：手动循环求和
total = 0
for x in data:
    total += x

# 好：使用内置 sum（C 实现）
total = sum(data)

# 差：手动找最大值
max_val = data[0]
for x in data:
    if x > max_val:
        max_val = x

# 好：使用内置 max
max_val = max(data)

# 差：手动去重
seen: set[int] = set()
unique: list[int] = []
for x in data:
    if x not in seen:
        seen.add(x)
        unique.append(x)

# 好：利用 set 的特性
unique = list(set(data))

# Python 3.12+
data = list(range(100_000))

# 列表推导式（推荐）：可读性好，性能优秀
result1 = [x * 2 for x in data if x % 2 == 0]

# map + filter + lambda：性能略优但可读性差
result2 = list(map(lambda x: x * 2, filter(lambda x: x % 2 == 0, data)))

# 结论：优先用列表推导式。只有当函数已经是现成的 C 实现时，
# map(str.lower, names) 才比 [n.lower() for n in names] 略快。

# Python 3.12+
from collections.abc import Iterator


def read_large_file(path: str) -> Iterator[str]:
    """逐行读取大文件，内存占用恒定"""
    with open(path, encoding="utf-8") as f:
        for line in f:
            yield line.strip()


# 差：一次性加载所有行到内存
# lines = open("huge_file.txt").readlines()

# 好：生成器管道处理
def process_logs(path: str) -> dict[str, int]:
    error_counts: dict[str, int] = {}
    for line in read_large_file(path):
        if "ERROR" in line:
            # 提取错误类型（简化示例）
            error_type = line.split("ERROR")[-1].strip().split()[0]
            error_counts[error_type] = error_counts.get(error_type, 0) + 1
    return error_counts

# Python 3.12+
import math


def compute_distances_slow(points: list[tuple[float, float]]) -> list[float]:
    """慢：每次循环都查找 math.sqrt 和 math.atan2"""
    return [
        math.sqrt(x**2 + y**2) * math.atan2(y, x)
        for x, y in points
    ]


def compute_distances_fast(points: list[tuple[float, float]]) -> list[float]:
    """快：将函数引用缓存为局部变量"""
    sqrt = math.sqrt
    atan2 = math.atan2
    return [
        sqrt(x**2 + y**2) * atan2(y, x)
        for x, y in points
    ]


# 同样适用于方法调用
def process_items_slow(items: list[str]) -> list[str]:
    result: list[str] = []
    for item in items:
        result.append(item.strip().upper())  # 每次查找 append
    return result


def process_items_fast(items: list[str]) -> list[str]:
    result: list[str] = []
    append = result.append
    strip = str.strip
    upper = str.upper
    for item in items:
        append(upper(strip(item)))
    return result

# Python 3.12+
words = ["hello", "world", "python", "performance"] * 10_000

# 差：+ 拼接每次创建新字符串，O(n²) 复杂度
# result = ""
# for w in words:
#     result += w + ","

# 好：join() 一次性分配结果空间
result = ",".join(words)

# Python 3.12+
from collections import deque


class RecentItems:
    """保存最近 N 条记录的高效队列"""

    def __init__(self, maxlen: int = 100) -> None:
        self._items: deque[str] = deque(maxlen=maxlen)

    def add(self, item: str) -> None:
        self._items.append(item)  # O(1)

    def get_recent(self, n: int = 10) -> list[str]:
        return list(self._items)[-n:]

    @property
    def count(self) -> int:
        return len(self._items)

# Python 3.12+
import asyncio
import time
from typing import Any

import httpx  # pip install httpx


async def fetch_url(client: httpx.AsyncClient, url: str) -> dict[str, Any]:
    """异步获取 URL"""
    response = await client.get(url)
    return {"url": url, "status": response.status_code, "length": len(response.text)}


async def fetch_all(urls: list[str]) -> list[dict[str, Any]]:
    async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
        tasks = [fetch_url(client, url) for url in urls]
        return await asyncio.gather(*tasks)


# 使用
# results = asyncio.run(fetch_all(["https://httpbin.org/delay/1"] * 10))

# Python 3.12+
from concurrent.futures import ProcessPoolExecutor
import math


def is_prime(n: int) -> bool:
    """CPU 密集：判断素数"""
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True


def count_primes_in_range(start: int, end: int) -> int:
    return sum(1 for n in range(start, end) if is_prime(n))


def parallel_prime_count(total_range: int, workers: int = 4) -> int:
    """使用进程池并行计算素数个数"""
    chunk_size = total_range // workers
    ranges = [
        (i * chunk_size, (i + 1) * chunk_size)
        for i in range(workers)
    ]

    with ProcessPoolExecutor(max_workers=workers) as executor:
        futures = [executor.submit(count_primes_in_range, s, e) for s, e in ranges]
        return sum(f.result() for f in futures)

# Python 3.12+
import functools
import time


@functools.lru_cache(maxsize=128)
def expensive_query(user_id: int, month: str) -> dict:
    """模拟一次昂贵的数据库查询"""
    time.sleep(0.1)  # 模拟查询延迟
    return {"user_id": user_id, "month": month, "data": "some result"}


# 第一次调用：执行查询
result1 = expensive_query(42, "2026-01")  # 耗时 ~100ms

# 第二次调用：命中缓存
result2 = expensive_query(42, "2026-01")  # 耗时 ~0ms

# 查看缓存统计
print(expensive_query.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=128, currsize=1)

# Python 3.12+
import time
from collections.abc import Callable
from typing import TypeVar

T = TypeVar("T")


class TTLCache:
    """简单的带过期时间的本地缓存"""

    def __init__(self, ttl_seconds: float = 60.0) -> None:
        self._ttl = ttl_seconds
        self._store: dict[str, tuple[float, object]] = {}

    def get(self, key: str) -> object | None:
        entry = self._store.get(key)
        if entry is None:
            return None
        timestamp, value = entry
        if time.monotonic() - timestamp > self._ttl:
            del self._store[key]
            return None
        return value

    def set(self, key: str, value: object) -> None:
        self._store[key] = (time.monotonic(), value)

    def clear_expired(self) -> int:
        """清理过期条目，返回清理数量"""
        now = time.monotonic()
        expired = [
            k for k, (ts, _) in self._store.items()
            if now - ts > self._ttl
        ]
        for k in expired:
            del self._store[k]
        return len(expired)

# Python 3.12+
# pip install redis
import json
from typing import Any

import redis


class RedisCache:
    def __init__(
        self,
        host: str = "localhost",
        port: int = 6379,
        default_ttl: int = 300,
    ) -> None:
        self._client = redis.Redis(host=host, port=port, decode_responses=True)
        self._default_ttl = default_ttl

    def get(self, key: str) -> Any | None:
        value = self._client.get(key)
        if value is None:
            return None
        return json.loads(value)

    def set(self, key: str, value: Any, ttl: int | None = None) -> None:
        self._client.setex(
            key,
            ttl or self._default_ttl,
            json.dumps(value),
        )

    def delete(self, key: str) -> None:
        self._client.delete(key)

# Python 3.12+
# pip install numba
from numba import jit
import math


@jit(nopython=True)
def count_primes_numba(limit: int) -> int:
    """Numba 加速的素数计数"""
    count = 0
    for n in range(2, limit):
        is_p = True
        for i in range(2, int(math.sqrt(n)) + 1):
            if n % i == 0:
                is_p = False
                break
        if is_p:
            count += 1
    return count

# GitHub Actions 示例片段
perf:
  runs-on: ubuntu-latest
  steps:
    - uses: actions/checkout@v4
    - uses: actions/setup-python@v5
      with:
        python-version: "3.12"
    - name: Install dependencies
      run: pip install -e .[dev]
    - name: Run performance benchmarks
      run: pytest tests/benchmarks/ --benchmark-only
    - name: Compare with baseline
      run: python scripts/compare_benchmarks.py