第4章性能优化

4. 性能优化：内存视图、批量处理与缓存策略

行情数据的处理，说白了就是跟时间赛跑。你想想看，交易所每秒可能推送几千笔增量，你要是每来一笔就拷贝一次、处理一次，那延迟肯定高得吓人。我早年做高频系统时，就吃过这个亏——代码逻辑明明没问题，但就是跑不过别人，后来一分析，发现大量时间都花在了内存拷贝上。

今天咱们就聊聊三个实战技巧：内存视图、批量处理和缓存策略。这三板斧用好了，你的行情合并性能至少能翻个倍。

4.1 用 memoryview 告别无谓拷贝

Python 里有个好东西叫 memoryview，它允许你在不复制底层数据的情况下，直接操作内存缓冲区。我刚开始也不太用它，觉得麻烦，直到有一次处理 100 万条快照数据，内存直接飙到 2GB……嗯，从那以后我就学乖了。

来看个对比：

# 传统方式：每次切片都会产生新对象
snapshot = b'\x00\x01\x02\x03\x04\x05'
price_bytes = snapshot[2:6]  # 这里发生了拷贝

# 使用 memoryview：零拷贝
mv = memoryview(snapshot)
price_view = mv[2:6]  # 只是创建了一个视图，没有复制数据
price = int.from_bytes(price_view, 'big')

为什么说这个重要？因为行情快照通常包含上千只股票的字段，如果你每取一个字段就拷贝一次，那开销是巨大的。用 memoryview 后，你只是创建了一个“窗口”，数据还在原地。

我的习惯：在解析行情协议时，我会先把整个数据包转成 memoryview，然后所有字段解析都基于这个视图操作。这样既安全又高效。

4.2 批量处理：别来一条处理一条

很多新手容易犯一个错误：收到一条增量，立刻合并到快照里。这样做有两个问题：

锁竞争激烈：如果多线程处理，每次合并都要加锁
上下文切换频繁：处理一条就切换一次，CPU 缓存都浪费了

我建议的做法是——攒一批，再处理。具体来说：

用一个队列缓存增量数据
每 10 毫秒或攒够 100 条，批量处理一次
处理时一次性合并到快照中

代码示例：

class BatchMerger:
    def __init__(self, batch_size=100, interval_ms=10):
        self.buffer = []
        self.batch_size = batch_size
        self.interval = interval_ms / 1000.0
        self.last_process_time = time.time()

    def add_increment(self, inc_data):
        self.buffer.append(inc_data)
        # 达到批量阈值或超时，触发处理
        if (len(self.buffer) >= self.batch_size or
            time.time() - self.last_process_time >= self.interval):
            self.flush()

    def flush(self):
        if not self.buffer:
            return
        # 批量合并逻辑
        merged = self._merge_batch(self.buffer)
        self._apply_to_snapshot(merged)
        self.buffer.clear()
        self.last_process_time = time.time()

你可能会问：这样会不会增加延迟？其实不会。因为 10 毫秒对大多数策略来说完全可以接受，而且批量处理能大幅提升吞吐量。我在项目中实测过，单条处理每秒最多 5 万笔，批量处理能到 30 万笔以上。

4.3 缓存策略：别重复计算

行情合并中有很多重复计算。比如，某只股票的价格在 1 秒内变了 10 次，但你的策略只关心每秒的 OHLC（开盘、最高、最低、收盘）。那中间那 8 次变化，其实没必要每次都重新计算。

我的缓存策略很简单：

快照缓存：保留最近一次完整快照，增量只更新变化字段
计算结果缓存：比如技术指标，只在快照更新时才重新计算
序列化缓存：如果要把行情数据发给其他进程，缓存序列化后的字节流

核心原则：能复用就不重算，能引用就不拷贝。

举个例子，假设你要计算某只股票的 5 日均线：

class CachedIndicator:
    def __init__(self, window=5):
        self.window = window
        self.cache = {}
        self.last_snapshot_id = None

    def get_ma5(self, snapshot_id, price_data):
        # 如果快照没变，直接返回缓存
        if snapshot_id == self.last_snapshot_id:
            return self.cache.get('ma5')

        # 否则重新计算并缓存
        ma5 = self._calculate_ma5(price_data)
        self.cache['ma5'] = ma5
        self.last_snapshot_id = snapshot_id
        return ma5

这样做的好处是，即使增量来了 100 次，只要快照 ID 没变，指标计算就只做一次。我见过有些系统每来一条增量就重新算一遍所有指标，CPU 直接拉满，其实完全没必要。

4.4 实战中的避坑指南

讲几个我踩过的坑：

我曾经犯过的错：

用 memoryview 后忘记释放引用，导致内存泄漏——记得在不用时显式删除视图
批量处理时没考虑极端行情，缓冲区积压到几万条——要加一个最大积压限制，超了就强制处理
缓存没设置过期时间，导致旧数据一直占用内存——给缓存加个 TTL（生存时间）

另外，缓存粒度也很重要。是按股票缓存，还是按交易所缓存？我个人的经验是：按股票粒度缓存最灵活，但内存开销大；按交易所粒度缓存省内存，但冲突多。具体选哪个，取决于你的策略是偏重个股还是偏重全市场。

4.5 一张图总结

下面这张图展示了整个优化流程的核心逻辑：

说白了，性能优化就是三板斧：少拷贝、攒批处理、多缓存。你把这三点做到位了，行情合并这块基本就稳了。我见过不少系统，代码写得花里胡哨，但性能就是上不去，一查全是拷贝和重复计算的问题。

嗯，今天就聊到这儿。下一节咱们会深入讲讲如何用这些技术搭建一个完整的行情处理管线，到时候再结合实战案例细说。

📚 课程目录

4. 性能优化：内存视图、批量处理与缓存策略

4.1 用 memoryview 告别无谓拷贝

4.2 批量处理：别来一条处理一条

4.3 缓存策略：别重复计算

4.4 实战中的避坑指南

4.5 一张图总结