从零认识 SPFA：懒人版 Bellman-Ford

《打造自己的套利机器人》课程第 7 课。第 6 课你实现了 Bellman-Ford。它能用，但做了不少无用功。今天我们让它“懒”一点。

Bellman-Ford 把力气浪费在哪里

第 6 课的 Bellman-Ford 是本课的直接版本：每一轮重新检查所有内容，确保结果正确。我们保留它作为参照，因为任何更快的方法都必须给出相同的答案。

Bellman-Ford 每一轮都会检查每条道路。但从 A 到 B 的道路，只有在 A 的数值刚变小后，才可能让到达 B 的路线变得更好。如果上次检查 A 的道路后，A 没有变化，再检查也不会有结果。

想象一位老师每天都重新检查 30 本作业本，尽管只有 3 个学生交了新作业。更聪明也更省力的老师会记下一张“有变化的学生”名单，只检查这些学生。这就是 SPFA（最短路径快速算法）的核心想法。

基本思路

1. 准备一个队列：里面放着还需要处理的节点。节点从队尾加入，从队首取出。
2. 一开始，队列里只放起点。
3. 取出队首节点 A。对从 A 出发的每条道路，尝试松弛它（A 的数值加上道路成本，是否比 B 的数值更好？）。
4. 如果 B 的数值变小了，而且 B 还不在队列中，就把 B 放到队尾。
5. 重复以上步骤，直到队列为空。

队列为空时，所有数值都已经确定。

一步步看一个例子

和第 6 课的道路相同：从 S 到 A 成本为 4，从 S 到 B 成本为 5，从 A 到 B 成本为 −2。

开始时：S = 0，A = ∞，B = ∞。队列：[S]。

• 取出 S。道路 S 到 A：0 + 4 = 4 < ∞，所以 A = 4，A 加入队列。道路 S 到 B：0 + 5 = 5 < ∞，所以 B = 5，B 加入队列。队列：[A, B]。
• 取出 A。道路 A 到 B：4 − 2 = 2 < 5，所以 B = 2。B 已经在队列里，不再重复添加。队列：[B]。
• 取出 B。B 没有出边。队列：[]。

检查 3 条道路后就完成了。Bellman-Ford 为了确认结果，会在第 1 轮检查 3 条道路，第 2 轮再检查 3 条：共检查 6 次。SPFA 在这里只做了一半的工作。

负环怎么办？

如果能到达一个负环，有些节点会不断变小，也会不断重新进入队列，程序就永远停不下来。我们需要一个警报。

警报的做法是：记录每个节点进入队列的次数。在没有负环的图中，一个节点最多进入 V − 1 次（到达它的最便宜路线最多使用 V − 1 条道路，每次进入都对应一条新的、更长的路线）。所以，如果一个节点进入队列 V 次，就说明经过它的路径上有负环。

代码

import math
from collections import deque

rates = {("USD", "EUR"): 0.90, ("EUR", "GBP"): 0.88, ("GBP", "USD"): 1.30,
         ("EUR", "USD"): 1.10, ("GBP", "EUR"): 1.10, ("USD", "GBP"): 0.75}
nodes = ["USD", "EUR", "GBP"]
out = {n: [] for n in nodes}                       # 每个节点的出边
for (a, b), r in rates.items():
    out[a].append((b, -math.log(r)))

def spfa(nodes, out, start):
    dist = {n: math.inf for n in nodes}
    prev = {n: None for n in nodes}
    entered = {n: 0 for n in nodes}                # 节点 n 加入队列的次数
    dist[start] = 0.0
    queue, waiting = deque([start]), {start}
    checks = 0
    while queue:
        a = queue.popleft()
        waiting.discard(a)
        for b, w in out[a]:
            checks += 1
            if dist[a] + w < dist[b]:
                dist[b] = dist[a] + w
                prev[b] = a
                if b not in waiting:
                    entered[b] += 1
                    if entered[b] >= len(nodes):   # 警报
                        return ("negative cycle through", b, "after", checks, "checks")
                    queue.append(b)
                    waiting.add(b)
    return ("no negative cycle", {n: round(d, 4) for n, d in dist.items()}, checks)

print(spfa(nodes, out, "USD"))

输出：

('negative cycle through', 'EUR', 'after', 13, 'checks')

SPFA 拉响了警报。要找出环本身，可以像第 6 课那样使用前驱节点：从触发警报的节点往回走 V 次，然后沿着这个环走一圈。把它当作练习试试看。

速度：自己动手测试

这里有一个包含 300 种货币的市场，每种货币随机配出 6 个交易对。汇率根据隐藏的真实价值生成，每笔交易都收取 0.2% 的费用，因此没有任何环能盈利（这是一个平稳的市场）。我们从同一个起点开始，统计道路检查次数。

import math, random
from collections import deque
random.seed(1)
N = 300
nodes = list(range(N))
value = [random.uniform(1, 100) for _ in nodes]
edges, out = [], {n: [] for n in nodes}
for a in nodes:
    for b in random.sample(nodes, 6):
        if a != b:
            w = -math.log(value[a] / value[b] * 0.998)
            edges.append((a, b, w)); out[a].append((b, w))

def bf_work():
    dist = [math.inf] * N; dist[0] = 0.0; work = 0
    for _ in range(N - 1):
        changed = False
        for a, b, w in edges:
            work += 1
            if dist[a] + w < dist[b]:
                dist[b] = dist[a] + w; changed = True
        if not changed: break
    return work

def spfa_work():
    dist = [math.inf] * N; dist[0] = 0.0; work = 0
    q, inq = deque([0]), {0}
    while q:
        a = q.popleft(); inq.discard(a)
        for b, w in out[a]:
            work += 1
            if dist[a] + w < dist[b]:
                dist[b] = dist[a] + w
                if b not in inq: q.append(b); inq.add(b)
    return work

print("edges:", len(edges), "Bellman-Ford checks:", bf_work(), "SPFA checks:", spfa_work())

一次测试的输出：

edges: 1799 Bellman-Ford checks: 7196 SPFA checks: 1799

这次 SPFA 只做了四分之一的工作。实际使用中，SPFA 在稀疏图上通常快得多，而加密货币市场往往是稀疏图（每种币只和少数其他币交易）。不过 SPFA 的最坏情况和 Bellman-Ford 一样，要做 V × E 次操作；少数图会触发这种情况。速度提升取决于数据的结构，所以要用自己的数据测量。

什么时候用哪一种

让 AI 智能体编写并测试代码

一个含糊的请求：

让 Bellman-Ford 快一点。

一个说明了方法和检查方式的请求：

在 bellman.py 中添加 spfa_find_cycle(rates)，使用队列、一个“在队列中”集合、每个节点进入队列的次数计数，以及前驱节点。在六种汇率的示例中，它必须返回和 find_cycle 相同的环；当所有汇率都乘以 0.95 时，必须返回 None。然后编写 bench.py，生成一个包含 300 种货币、每种货币有 6 个交易对的随机市场（汇率由隐藏的价值乘以 0.998 得到），从节点 0 开始统计两个算法检查道路的次数，并打印两个数字。不要修改 find_cycle。

你应该会看到 SPFA 的检查次数更少。如果智能体在这组数据上显示 SPFA 检查次数更多，就让它解释原因，不要接受“这个算法理论上更好”这样的回答。

练习

1. 运行第一个程序。修改它，让程序用前驱节点返回环本身。
2. 运行基准测试。把每种货币的交易对数从 6 改成 30。SPFA 还能保持同样的优势吗？图变得更密集后，结果为什么可能改变？
3. 把费用系数 0.998 改成 1.001（这是一个有免费钱可赚的市场）。从节点 0 运行基准测试中的 spfa_work。它不会停，因为没有警报。按 Ctrl+C 停下来，然后加上第一个程序中的“进入队列 V 次”警报，再运行一次。警报响起前需要检查多少次？
4. 把代码保存为一个新版本。

接下来学什么

这两种算法都能回答“某处是否存在可盈利的环？”。真正的机器人还需要更多：在步骤数有限的情况下，反复找出最佳环，并跟上不断变化的价格。第 8 课将介绍一种近期研究算法 RICH 如何解决这个问题。