优化热点循环

少做事，而不是更快地做同样的事

当分析器指出某个热点循环后，最大的收益来自改变它做多少工作，而不是榨取同样的工作。一个在一百万个位置上每次都把 250 个价格的窗口重新相加的循环要做 2.5 亿次加法；滑动窗口的版本只需要两百万次。这种差异，即算法复杂度，让内层那一行上的任何小技巧都相形见绌。所以先看循环在重复什么：在列表里查找、从头重算的求和、永远不变却每次重建的值。

# Membership in a list scans it; in a set it is a hash lookup. Same answer, very different work
import time

ids = list(range(20_000))
as_set = set(ids)
wanted = list(range(19_000, 20_000))

start = time.perf_counter()
found_list = sum(1 for x in wanted if x in ids)             # scans up to 20,000 items for each of 1,000 lookups
slow = time.perf_counter() - start

start = time.perf_counter()
found_set = sum(1 for x in wanted if x in as_set)
fast = time.perf_counter() - start

print(found_list, found_set)                                # 1000 1000
print(slow > 20 * fast)                                     # True  the set is at least twenty times faster here

滑动、只查一次、提到循环外

三种模式涵盖了大多数热点循环。滑动而不是重算：移动求和加上进入窗口的值、减去离开的值，所以无论窗口多长，每个位置都只花两次运算。只建一次查找表：把配对列表变成 dict 只需一遍，之后每次查找都是瞬时的，而不是对每个查询都扫描列表。把不变的东西提到循环外：循环里计算出来、却只依赖循环外值的东西（常量、len(x)、属性查找）应该放到循环之前。像本实验这样，用读取或运算次数的计数器来测试：计数不依赖机器速度，因此是衡量工作量的稳定指标。

# A sliding sum does two operations per position; recomputing each window does `window` of them
def moving_sum_slow(values, window):
    operations = 0
    sums = []
    for i in range(len(values) - window + 1):
        total = 0
        for v in values[i : i + window]:
            total += v
            operations += 1
        sums.append(total)
    return sums, operations

def moving_sum_fast(values, window):
    operations = window
    total = sum(values[:window])
    sums = [total]
    for i in range(window, len(values)):
        total += values[i] - values[i - window]
        operations += 2
        sums.append(total)
    return sums, operations

values = list(range(1_000))
slow, slow_ops = moving_sum_slow(values, 250)
fast, fast_ops = moving_sum_fast(values, 250)
print(slow == fast)                                   # True  the same answers
print(slow_ops, fast_ops)                             # 187750 1750  about a hundred times less work

# Hoisting: a value that never changes is computed once, outside the loop
import math

calls = {"count": 0}

def annualisation_factor():                          # imagine something costly that never changes
    calls["count"] += 1
    return math.sqrt(252)

def scale_slow(returns):
    return [r * annualisation_factor() for r in returns]

def scale_fast(returns):
    factor = annualisation_factor()                  # hoisted out of the loop
    return [r * factor for r in returns]

returns = [0.001 * i for i in range(2_000)]
slow = scale_slow(returns)
slow_calls, calls["count"] = calls["count"], 0
fast = scale_fast(returns)
print(slow == fast)                                  # True  the same numbers
print(slow_calls, calls["count"])                    # 2000 1   one call instead of two thousand

让 NumPy 替你循环

当工作确实是对大量数字做算术时，把循环移出 Python。NumPy 数组把数字存放在一块连续内存中，并在编译后的代码里对所有元素执行运算，所以 np.log(prices[1:] / prices[:-1]) 用一次调用取代了 Python 循环，通常快几十倍。这就是向量化，是数值代码中最先要尝试的优化。它最适合对整个数组做运算、且每个元素没有 if 的情形；需要条件时用 np.where 或布尔掩码。也要留意内存：向量化会产生与数据同样大小的临时数组。

# Vectorising: one NumPy call replaces a Python loop and gives the same numbers
import math
import time
import numpy as np

rng = np.random.default_rng(3)
prices = 100 * np.exp(np.cumsum(rng.normal(0, 0.001, size=300_000)))

start = time.perf_counter()
loop = [math.log(prices[i] / prices[i - 1]) for i in range(1, len(prices))]
loop_time = time.perf_counter() - start

start = time.perf_counter()
vector = np.log(prices[1:] / prices[:-1])
vector_time = time.perf_counter() - start

print(np.allclose(loop, vector))                    # True  the same returns
print(loop_time > 5 * vector_time)                  # True  at least five times faster here

• 先减少工作量：滑动窗口而不是重新相加，避免重算已有的结果。
• 使用合适的结构：查找用（只建一次的）集合或字典，而不是每次查询都扫描列表；把常量提到循环外。
• 用 NumPy 向量化数值循环，测试中统计工作量而非秒数，并持续把快版本与慢而显然正确的版本对照。

下一节：Numba，用一个装饰器把 Python 数值循环编译成机器码。

实验：评分内容

• 在 moving_sum 中滑动窗口25 分
• 在 has_duplicates 中使用集合25 分
• 在 lookup_prices 中只构建一次字典25 分
• 将 log_returns 向量化25 分