課程「建立自己的套利機器人」第 13 課。從這裡進入世界 4:機器人能運作了,現在必須夠快,也要準備好展示。今天你會了解掃描器為什麼變慢,以及該怎麼處理。
為什麼這裡的速度很重要
你讀到的價格只在短暫的一刻有價值。第 11 課已經說明這一點:其他機器人看見同一個價差,較慢的掃描器到達時訂單簿已經空了。需要兩秒才能檢查市場的掃描器,檢查的是一個已經不存在的市場。
所以今天的問題很簡單。機器人每秒能檢查多少個環路,又怎樣把這個數字變大?
直接的方法:三層迴圈
幾乎每個人寫的第一個掃描器,都會檢查每個由三種貨幣組成的環:把 A 換成 B,把 B 換成 C,再把 C 換回 A,並將三個匯率相乘。如果乘積大於 1,錢就出現了。在程式碼裡,這就是三層巢狀迴圈。
建立一個有 120 種貨幣的市場。匯率來自隱藏的數值,每次交易收取 0.2% 的費用,所以沒有任何交易獲利。接著刻意放入一個錯誤定價的交易對:把貨幣 7 換成貨幣 40 時,會比應有的數量多得到 2%。
import random, time
import numpy as np
random.seed(5)
N = 120
value = [random.uniform(1, 100) for _ in range(N)]
# rate[i][j] = how much of j you get for 1 of i, with a 0.2% fee on every trade
rate = [[value[i] / value[j] * 0.998 if i != j else 1.0 for j in range(N)] for i in range(N)]
rate[7][40] *= 1.02 # the planted mistake
def best_loops():
best, arg = 0.0, None
for i in range(N):
for j in range(N):
for k in range(N):
if i == j or j == k or k == i:
continue
m = rate[i][j] * rate[j][k] * rate[k][i]
if m > best:
best, arg = m, (i, j, k)
return best, arg
t = time.perf_counter(); b = best_loops(); print(round(b[0], 4), b[1], round(time.perf_counter() - t, 2), "s")
一次測試執行輸出:
1.0139 (0, 7, 40) 0.55 s
掃描器找到了刻意放入的錯誤:把 0 換成 7,把 7 換成 40,再把 40 換回 0,1 個單位變成 1.0139。2% 的人為偏差,扣掉三次各 0.2% 的費用後,剩下約 1.4%。
它花了 0.55 秒。作為展示這沒有問題,對市場來說卻遠遠太慢。環路數量隨貨幣數量的立方成長:120 種貨幣約有 1.7 百萬個環路,240 種約有 13.8 百萬個,是原來的八倍。市場規模加倍,迴圈就慢八倍。
想法 1:一次做很多乘法
電腦將兩個各含百萬個數字的清單相乘,遠比 Python 迴圈逐對處理快。原因是 Python 迴圈的大部分時間花在管理工作上:這是什麼型別,下一個元素在哪裡。像 numpy 這樣的函式庫會把整個清單交給快速的編譯程式碼,管理工作只做一次。
這個技巧有個名字:向量化。你不再寫「對每個元素做這個」,而是寫「對所有元素做這個」。
我們的匯率是一張 120 列、120 欄的表。在數學中它叫矩陣,三維表則是一疊這樣的矩陣。每個三元組 (i, j, k) 需要的乘積是 rate[i][j] * rate[j][k] * rate[k][i]。建立同一張表的三個視圖,讓每個視圖沿不同方向對齊,再把它們相乘:
def best_numpy():
R = np.array(rate)
# m[i, j, k] = R[i, j] * R[j, k] * R[k, i], all at once
m = R[:, :, None] * R[None, :, :] * R.T[:, None, :]
idx = np.arange(N)
m[idx, idx, :] = 0; m[idx, :, idx] = 0; m[:, idx, idx] = 0 # drop triples with a repeat
flat = int(m.argmax())
return float(m.flat[flat]), np.unravel_index(flat, m.shape)
t = time.perf_counter(); b = best_numpy(); print(round(b[0], 4), tuple(int(x) for x in b[1]), round(time.perf_counter() - t, 3), "s")
怎樣理解函式的第一行。R[:, :, None] 是加上一個空的第三方向後的表,因此它沿 i 和 j 變化。R[None, :, :] 沿 j 和 k 變化。R.T[:, None, :] 是轉置後的表,列與欄互換,因此它在 (i, k) 位置的項目是 R[k][i]。將形狀為 120 × 120 × 1、1 × 120 × 120 和 120 × 1 × 120 的三個物件相乘,會得到一個 120 × 120 × 120 的立方體,每個儲存格剛好儲存一個乘積。接下來幾行會將索引重複的儲存格設為零,argmax 會找出最大的儲存格。
一次測試執行輸出:
1.0139 (0, 7, 40) 0.009 s
同一個答案,0.009 秒而不是 0.55 秒:約快 60 倍,用三行程式碼而不是一個迴圈。自己檢查一下:如果答案和迴圈版不同,其中一個一定有 bug。
這個技巧的代價
這個立方體有 N × N × N 個儲存格,每個儲存格是一個 8 位元組數字。對 120 種貨幣來說是 14 MB,沒有問題。對 240 種貨幣是 110 MB,對 1000 種貨幣則是 8 GB,裝不下。以下是一次測試執行的計時,只計算乘法:
60 numpy 0.001 s
120 numpy 0.007 s
240 numpy 0.105 s
規模加倍仍會慢約 8 到 15 倍,因為立方體仍隨 N 的立方成長。向量化改變的是常數,而不是成長形狀。面對大市場,你需要減少工作量,不能只讓工作變快。第 6 到 8 課講的就是這個:Bellman-Ford 及其同類演算法會跳過大部分環路。實務上要結合兩者:用聰明的演算法檢查少量環路,用快速程式碼迅速檢查每一個。
想法 2:可編譯的語言
Rust 會在程式執行前把它變成機器碼,它的迴圈每一步幾乎沒有額外成本。這樣可以重新使用最簡單的三層迴圈程式碼,也不必在記憶體中保留立方體。用 Python 把匯率儲存到文字檔,每列一行:
open("rates.csv", "w").write("\n".join(",".join(repr(x) for x in row) for row in rate))
接著是 Rust 程式:
use std::fs;
use std::time::Instant;
fn main() {
let text = fs::read_to_string("rates.csv").expect("rates.csv");
let rate: Vec<Vec<f64>> = text
.lines()
.map(|l| l.split(',').map(|x| x.parse().unwrap()).collect())
.collect();
let n = rate.len();
let start = Instant::now();
let (mut best, mut arg) = (0.0_f64, (0, 0, 0));
for i in 0..n {
for j in 0..n {
if j == i { continue; }
for k in 0..n {
if k == i || k == j { continue; }
let m = rate[i][j] * rate[j][k] * rate[k][i];
if m > best { best = m; arg = (i, j, k); }
}
}
}
println!("rust {:.4} {:?} {:.3} s", best, arg, start.elapsed().as_secs_f64());
}
使用 rustc -O best.rs -o best && ./best 建置並執行。-O 旗標會開啟最佳化器,沒有它程式會慢得多。一次測試執行輸出:
rust 1.0139 (0, 7, 40) 0.005 s
0.005 秒:比 Python 迴圈快約 110 倍,比 numpy 稍快,而且記憶體中沒有大立方體。程式和第一個 Python 版本有同樣的三層迴圈,改變的只有語言。
把三個結果並排比較:
| 版本 | 120 種貨幣的時間 | 付出的代價 |
|---|---|---|
| Python 迴圈 | 0.55 s | 沒有,最容易讀 |
| numpy 矩陣 | 0.009 s | 記憶體按 N³ 成長,技巧較難讀 |
| Rust 迴圈 | 0.005 s | 第二種語言和一次編譯步驟 |
為你的機器人選擇什麼
- 先在 Python 中做原型,讓演算法清楚且正確。先不要最佳化。
- 測量。用計時器找出最慢的部分,就像本課所做的那樣。你對慢點的猜測出錯的次數,比預想的更多。
- 如果慢的部分是數字上的大迴圈,先試 numpy。只需幾行程式碼。
- 如果機器人必須在毫秒內反應,或者 numpy 耗盡記憶體,就把那一個函式移到 Rust,再從 Python 呼叫它,或者用 Rust 寫熱點路徑。機器人的其他部分,設定、記錄和模擬交易,可以繼續留在 Python。
- 更快的迴圈永遠不能替代更好的演算法。即使把檢查 14 百萬個環路的速度提高 100 倍,仍然比不上一個只檢查 1000 個環路的演算法。
讓 AI 代理人寫它並測試
一個模糊的要求:
讓我的掃描器變快。
一個帶有待比對數字的要求:
撰寫 fast.py。按第 13 課的程式碼建立市場,seed 為 5,120 種貨幣,手續費為 0.2%,rate[7][40] 乘以 1.02。實作 best_loops 和 best_numpy,印出各自的結果和時間。兩者都必須為環路 (0, 7, 40) 回傳乘數 1.0139;如果不同,要說明並修正 bug。接著撰寫 best.rs,它從 rates.csv 讀取匯率,用三層迴圈找出同一環路,用 rustc -O 建置並印出時間。不要更改市場的數字。
先確認三個答案完全一致,再看速度。快速的錯誤答案沒有用。
練習
- 執行 Python 程式碼,確認兩個版本都印出 1.0139 和環路 (0, 7, 40)。記下你的計時。
- 將 N 改為 60,再改為 200。迴圈版本的時間如何變化?與它按 N 的立方成長這一說法比較:200 / 120 的立方約為 4.6。
- 加入第二個錯誤,rate[50][90] *= 1.03,並修改函式,讓它回傳最好的三個環路,而不是只回傳最好的那一個。提示:用 np.argsort 排序立方體,或在迴圈中保留一個清單。
- 用 rustc 不帶 -O 執行 Rust 版本,再帶上 -O 執行。最佳化器將時間改變了多少?
接下來
現在你知道時間花在哪裡,也知道怎樣把它拿回來。第 14 課會加入第二種策略:期貨對現貨的資金費率策略,機器人從每隔數小時重複一次的小差異中賺錢,而不是從幾秒內就會關閉的價差中賺錢。