從零認識 SPFA:Bellman-Ford 的懶人版

套利課程第 7 課。SPFA 如何只重新檢查可能改變的道路、一步步的佇列範例、含負環警報的可執行 Python 程式,以及和 Bellman-Ford 的速度比較。

課程「建立自己的套利機器人」第 7 課。你在第 6 課做出了 Bellman-Ford。它有效,但浪費力氣。今天要讓它變得更懶。

Bellman-Ford 在哪裡浪費工作

第 6 課的 Bellman-Ford 是本課直接硬做的版本:每一輪重新檢查所有東西,以確保正確。我們保留它作為基準,因為任何更快的方法都必須給出相同答案。

Bellman-Ford 每一輪都檢查每條路。但只有在 A 的數字剛變小時,從 A 到 B 的道路才可能提供一條到 B 更好的路。如果上次查看 A 的道路後,A 沒有改變,再看一次也不會改變任何事。

想像一位老師每天重新檢查全部 30 本作業簿,即使只有三個學生交了新作業。一位更懶也更聰明的老師會保留一張「有東西改變的學生」清單,只檢查那些學生。這張清單就是 SPFA(Shortest Path Faster Algorithm)的全部想法。

想法

  1. 保留一個佇列:一排等待處理的節點。節點從後面加入,從前面離開。
  2. 一開始只有起始節點在佇列中。
  3. 從前面取出節點 A。對從 A 到某個 B 的每條道路,試著鬆弛它(A 的數字加上成本,是否比 B 的數字更好?)。
  4. 如果 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,少見的圖可能觸發它。加速幅度取決於資料的形狀,所以要用你自己的資料測量。

何時用哪一種

Bellman-FordSPFA
想法重新檢查所有道路,V − 1 輪只重新檢查改變節點的道路
最糟情況V × EV × E
市場圖中的典型情況大約 V × E少得多
負環警報多做一輪一個節點進入佇列 V 次
簡單程度最簡單多幾行程式

讓 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 如何處理這個問題。