Day 04 的 SMA 和 Day 05 的 EMA 都在回答同一個問題:價格往哪個方向走。這個問題在趨勢明顯的時候答得很好,快線在慢線之上就是往上,穿下去就是轉折。
麻煩的是市場大部分時間不長那樣。把 Day 05 存下來的那張 BTC/USDT 現貨 1 小時 K 線攤開,會看到很長一段時間裡 EMA(12) 和 EMA(26) 幾乎黏在一起、反覆交錯,價格在一個區間裡上上下下。這時候均線給不出任何有用的東西,因為它只有「上」和「下」兩種答案,而現在的答案是「橫著」。
今天換一個問題來問:走得多急。同樣是上漲 3%,如果是連續八根小陽線慢慢爬上去,跟一根長紅棒一次拉上去,這兩件事在均線上長得差不多,但市場的狀態完全不同。RSI(Relative Strength Index,相對強弱指標)就是把「多急」這件事壓成一個 0 到 100 的數字。
它也是這三天的最後一個指標。它跟前兩天一樣繼承 Day 04 訂好的 Indicator,所以今天寫完之後,三個指標會一起進到一張註冊表裡——那是 Day 15 要把所有特徵模組化的前置作業。
今天有三個新名詞。
動能(momentum)指的是價格變動的速度與力道,不是方向。它回答的是「最近這段時間,往上的力氣和往下的力氣哪一邊比較大、大多少」。
用工程的話講,均線是對價格本身做平滑,動能類指標是對價格的一階差分做統計。輸入從 close 換成 close.diff(),問題就從「在哪」變成「怎麼動」。這也是為什麼動能指標畫不進 K 線圖裡:它的單位跟價格不一樣,必須另外開一個子圖。
RSI 的值域是 0 到 100。習慣上把 70 以上叫超買(overbought)、30 以下叫超賣(oversold)。
這兩條線是慣例,不是定律。RSI 最早提出時建議的就是 70/30,後來大家沿用了幾十年,看盤軟體也預設畫在那裡。有人用 80/20(觸發次數更少但更極端),有人在明顯的上升趨勢裡把兩條線一起往上挪。它們是我們自己挑的閾值,跟 14 這個週期一樣,沒有神聖性。
至於怎麼決定該用哪一組,那需要一整套判斷「這組參數是不是自己挑出來的假象」的方法,系列後面有一整天在處理。在那之前,不要用「多試幾組挑數字最好看的那組」來決定。
背離(divergence)指的是價格創了新高,但 RSI 沒有跟著創新高(或反過來,價格創新低而 RSI 沒有)。它描述的是「價格還在往上,但推它上去的力氣在變小」。
今天只解釋概念,不實作。要程式化偵測背離,得先有一套找出價格局部高低點的方法,那個東西後面幾天才會建起來,硬用 rolling().max() 湊會很容易用到當根之後的資料。這裡先記住它是個「兩條線的形狀比對」問題,不是一個單根 K 線就能判斷的閾值問題。
這是零基礎讀者對 RSI 最常見的誤解,而且它會長成一個一路虧錢的策略:看到 RSI > 70 就準備放空。
RSI 高只代表最近的上漲幅度佔總變動幅度的比例很大,它完全沒有說接下來會怎樣。在一段強勢的上漲行情裡,RSI 可以貼著 75 到 85 走上好幾天,期間價格繼續漲,那張空單就一路虧。
這不是經驗談,是可以從定義直接推出來的。RSI 的公式(下一節會拆)長這樣:
RSI = 100 × 平均漲幅 / (平均漲幅 + 平均跌幅)
如果一段序列裡每一根都在漲,平均跌幅是 0,RSI 恆等於 100。實際跑一下確實如此:
>>> rising = make_series(np.arange(100, 140, 1.0)) # 每一根都比前一根高
>>> RSI(14).compute(rising).iloc[-1]
100.0
也就是說,只要「跌的那幾根很小」,RSI 就會一直很高,跟「該不該回檔」沒有任何關係。RSI 是一個對已經發生的事情所做的描述,不是對接下來的預測。這句話對後面所有特徵都成立,只是 RSI 因為有 70/30 這兩條線,特別容易讓人誤讀成紅綠燈。
實務上的用法比較接近「條件」而不是「指令」:RSI > 70 可以拿來當過濾條件(這套進場邏輯在追高的時候先不要動作),而不是進場條件本身。這個區分會在 Day 16 變成程式碼裡的兩種不同角色。
拆開來只有三步,每一步都很短。
拿到相鄰兩根 K 線的收盤價差之後,把它拆成兩條序列:漲的那些(跌的記 0)與跌的那些取絕對值(漲的記 0)。
| 收盤價 | 差分 | 漲幅 | 跌幅 |
|---|---|---|---|
| 100 | — | — | — |
| 103 | +3 | 3 | 0 |
| 101 | −2 | 0 | 2 |
| 101 | 0 | 0 | 0 |
| 105 | +4 | 4 | 0 |
分開之後,「漲的力氣」與「跌的力氣」就變成兩個可以獨立統計的量。
對這兩條序列各做一次平滑,得到平均漲幅與平均跌幅。這一步是整篇最容易寫錯的地方,等一下用一整節處理。
兩個平均值相除得到相對強度(RS),再套進一個把值域壓到 0–100 的式子:
RS = 平均漲幅 / 平均跌幅
RSI = 100 − 100 / (1 + RS)
這個式子等價於前面那個看起來比較直覺的寫法:
RSI = 100 × 平均漲幅 / (平均漲幅 + 平均跌幅)
兩邊做個代數就能對上。實作要選後者,理由在邊界那節會講。
看第二種寫法就很清楚 RSI 在量什麼了:在最近這段時間的總變動幅度裡,往上的部分佔多少比例。 全部往上是 100,全部往下是 0,漲跌力道相當是 50。
14 是這個指標最初被提出時用的預設值,來源是「大約半個月的交易日」這種很粗的直覺,不是最佳化出來的。它沿用至今主要是因為所有看盤軟體都預設 14,導致很多人看的是同一組數字。
週期短(例如 7)反應快、進出 70/30 的次數多;週期長(例如 21)平滑、次數少但延遲大。這個取捨跟 Day 04 講 SMA 視窗長度時完全一樣。今天預設用 14,是為了能跟 pandas-ta 的預設值直接對數字。
第一步是純粹的向量化操作,不需要迴圈:
change = series.frame["close"].diff()
gain = change.clip(lower=0.0) # 跌的變成 0
loss = (-change).clip(lower=0.0) # 漲的變成 0,跌的取正值
clip(lower=0) 跟 where(change > 0, 0) 在數值上等價,挑一個順眼的就好:
>>> np.allclose(change.clip(lower=0).fillna(0), change.where(change > 0, 0).fillna(0))
True
要注意 diff() 的第一筆是 NaN,這個 NaN 會一路帶到最後的暖機期,是預期行為,不要急著 fillna(0)。填 0 等於宣稱第一根 K 線之前的價格跟它一樣,那是捏造出來的資料。
網路上有不少 RSI 教學是用 for i in range(len(candles)) 逐根累加寫的。那個寫法在幾千根 K 線上還跑得動,到了 Day 09 之後處理逐筆成交的資料量就會直接卡死。全系列的規矩不變:NEVER 用 for loop 遍歷 K 線。
這一節是今天最有價值的一段。
Day 05 已經講過 EMA 是遞迴的,每一根的值是「上一根的值」與「這一根的新資料」的加權平均,權重由平滑係數 α 決定:
ema[i] = α × price[i] + (1 − α) × ema[i−1]
也知道 pandas 的 ewm(span=n) 用的 α 是 2 / (n + 1)。
問題來了:RSI 用的平滑不是這個。 RSI 的原始定義用的是另一種遞迴平均,通常叫 Wilder 平滑(也叫 RMA、SMMA),它的更新式長這樣:
avg[i] = (avg[i−1] × (n − 1) + value[i]) / n
把它整理一下就會發現這就是 EMA,只是 α 換成 1 / n:
avg[i] = (1/n) × value[i] + (1 − 1/n) × avg[i−1]
兩個係數差多少,代進 n = 14 就知道:
| 平滑方式 | α 的定義 | n = 14 時的 α | 等價的 EMA span |
|---|---|---|---|
| Wilder(RSI 用的) | 1 / n | 0.0714 | 27 |
| 一般 EMA | 2 / (n + 1) | 0.1333 | 14 |
差了將近一倍。反過來解 1/n = 2/(span+1) 會得到 span = 2n − 1,所以一個 14 期的 RSI,平滑力道相當於 27 期的 EMA。順手寫成 ewm(span=14) 的話,算出來的東西實際上接近一個 7.5 期的 RSI,會比正確版本敏感得多。
正確的寫法是直接指定 α,不要用 span:
avg_gain = gain.ewm(alpha=1 / period, adjust=False).mean() # 對
avg_gain = gain.ewm(span=period, adjust=False).mean() # 錯,α 變成 2/(n+1)
這一行等一下會被關進 WilderSmoother,讓它只出現一次。
adjust=False 也不能省。Day 05 講過 adjust=True 算的是「到目前為止所有資料的加權平均」,adjust=False 才是遞迴式本身。RSI 的定義要的是後者。
遞迴式需要一個起點。RSI 的慣例是:前 n 筆漲幅(跌幅)的算術平均當作第 n 根的初始值,從第 n+1 根開始才進遞迴。
這件事 ewm() 不會自己做,它預設拿第一個非 NaN 的值當起點。所以要手動塞種子:把前 n 個位置設成 NaN、在第 n 個位置放進算術平均,再讓 ewm() 從那裡開始遞迴。
用不用種子,差別會在前兩三百根 K 線上,這點在驗證那節會有具體數字。
兩個類別。平滑自己是一個東西,因為「α 該是多少、種子怎麼放」是這篇最容易寫錯的地方,把它獨立出來就有一個可以單獨測試、也可以被之後的指標重用的對象:
# quantbot/domain/indicators/rsi.py
from __future__ import annotations
from typing import ClassVar
import numpy as np
import pandas as pd
from quantbot.domain.indicators.indicator import Indicator
class WilderSmoother:
"""Wilder 平滑(也叫 RMA、SMMA):alpha = 1/period。
獨立成一個類別是因為它是 RSI 最容易寫錯的一行:alpha 是 1/n,
NEVER 寫成 ewm(span=n)——那是 EMA 的 2/(n+1),差了將近一倍,
而算出來的東西照樣落在 0 到 100 之間,只有對照組看得出差別。
它跟 RSI 住同一個檔案,因為目前只為 RSI 存在;哪天 ATR 也要用,再搬出去。
"""
def __init__(self, period: int) -> None:
if period < 1:
raise ValueError(f"period 必須 >= 1,收到 {period}")
self.period = period
@property
def alpha(self) -> float:
return 1.0 / self.period
def smooth(self, values: pd.Series) -> pd.Series:
"""平滑一條序列,起始值用前 period 筆的算術平均。
values 的第 0 筆是 diff() 產生的 NaN,所以種子取 values[1:period+1],
放在第 period 個位置,遞迴從第 period + 1 筆開始。
"""
seeded = values.copy()
seeded.iloc[: self.period] = np.nan
seeded.iloc[self.period] = values.iloc[1 : self.period + 1].mean()
return seeded.ewm(alpha=self.alpha, adjust=False).mean()
class RSI(Indicator):
"""相對強弱指標,值域 0-100。
全部上漲回傳 100、全部下跌回傳 0、完全持平回傳 NEUTRAL_VALUE(50)。
最後那個 50 是補的、不是算出來的,所以它是一個具名的類別常數。
"""
NEUTRAL_VALUE: ClassVar[float] = 50.0
def __init__(self, period: int = 14, *, column: str = "close") -> None:
super().__init__(period, column=column)
self._smoother = WilderSmoother(period)
@property
def name(self) -> str:
return f"rsi_{self.period}"
def _compute(self, values: pd.Series) -> pd.Series:
# 資料不足以算出任何有效值時回等長的全 NaN,而不是丟例外:
# 呼叫端處理的是 NaN,不是 try/except。
if len(values) <= self.period:
return pd.Series(np.nan, index=values.index, dtype="float64")
change = values.diff()
average_gain = self._smoother.smooth(change.clip(lower=0.0))
average_loss = self._smoother.smooth((-change).clip(lower=0.0))
total = average_gain + average_loss
strength = 100.0 * average_gain / total
# total == 0 表示這段完全沒動,漲跌力道相當,補中性值。
# 用 mask 而不是 where(total > 0, 50):NaN > 0 是 False,
# 那樣寫會把暖機期的 NaN 一起填成 50。
return strength.mask(total == 0.0, self.NEUTRAL_VALUE)
RSI 繼承 Day 04 訂好的 Indicator,所以它只剩下 name 與 _compute 那十幾行,也就是真正屬於 RSI 的東西。檢查欄位、轉型別、確認 index 沒被動過、把 name 貼上去,這四件事由基底類別統一處理,新增指標時不會漏。
warmup_bar_count 是 period:前 14 根是 NaN,第 15 根(index 14)開始有值。這個數字由 Indicator 提供,Day 15 的 pipeline 會靠它自動決定整條特徵管線要丟掉前面幾根。
向量化寫法最怕的是「跑得很快但算錯」。最省事的驗證方式是照教科書的遞迴定義寫一支慢的迴圈版本,只當測試用的參考實作,然後跟正式版對數字:
介面跟 Day 05 的 ReferenceEMA 一致,兩個參考實作放在同一個 tests/reference/ 底下:
# tests/reference/reference_rsi.py
"""教科書定義的迴圈版 RSI。只在測試裡當對照組,NEVER 進正式路徑。"""
from __future__ import annotations
import numpy as np
import pandas as pd
class ReferenceRSI:
"""照 avg[i] = (avg[i-1] * (n-1) + value[i]) / n 這條原始更新式寫。
連 100 - 100/(1+RS) 那個式子都照抄,好跟教科書逐行對照——正式版用的是
等價但不會產生 inf 的另一種寫法,兩邊對得起來才表示代數沒推錯。
"""
def __init__(self, period: int = 14) -> None:
self.period = period
def compute(self, closes: pd.Series) -> pd.Series:
period = self.period
gain = closes.diff().clip(lower=0.0).to_numpy()
loss = (-closes.diff()).clip(lower=0.0).to_numpy()
count = len(closes)
average_gain = np.full(count, np.nan)
average_loss = np.full(count, np.nan)
average_gain[period] = gain[1 : period + 1].mean()
average_loss[period] = loss[1 : period + 1].mean()
for index in range(period + 1, count):
average_gain[index] = (
average_gain[index - 1] * (period - 1) + gain[index]
) / period
average_loss[index] = (
average_loss[index - 1] * (period - 1) + loss[index]
) / period
strength = average_gain / average_loss
return pd.Series(100 - 100 / (1 + strength), index=closes.index)
在 2,000 根序列上,兩者最大差異 3.55e-14,是浮點累加順序造成的,沒有邏輯差異。
接著上真正的對照組。Day 04 和 Day 05 都跟 pandas-ta 對過數字,RSI 這裡要多一個提醒:pandas-ta 的實作跟教科書定義在暖機期不一致,不能直接期待全序列誤差在 1e-9 以內。
pandas-ta(0.3.14b)的 RSI 走的是它自己的 rma(),而 rma() 的核心是這一行:
close.ewm(alpha=1 / length, min_periods=length).mean()
α 是對的(1/length),但它沒有指定 adjust,所以走 pandas 的預設值 True,也沒有塞算術平均的種子。這兩件事都只影響前面那段,後面會收斂。
實際差多少?用一段固定種子的合成序列跑(用合成資料是為了複製下去能得到一模一樣的數字;換成真實的 BTC/USDT 1h 資料,數量級一樣):
import pandas_ta
from quantbot.domain.indicators.rsi import RSI
series = make_series(random_closes(2000)) # 測試裡那兩支 helper
mine = RSI(14).compute(series)
theirs = pandas_ta.rsi(series.frame["close"], length=14)
for start in (0, 100, 200, 300, 400):
difference = (mine - theirs).abs().iloc[start:].dropna()
print(f"從第 {start:>3} 根之後比對:最大誤差 {difference.max():.3e}")
結果:
| 從第幾根之後開始比 | 最大誤差 |
|---|---|
| 0(全序列) | 6.44 |
| 100 | 1.05e−02 |
| 200 | 6.54e−06 |
| 300 | 4.56e−09 |
| 400 | 1.61e−12 |
暖機期差到 6.44 個 RSI 點,然後每過一段就縮小一個數量級。原因是起始值不同造成的偏差每根乘上一次 1 − 1/n,n = 14 時每 14 根縮成原來的 0.354 倍,要從 6.4 縮到 1e-9 大約需要 300 根。
這給出兩個實務結論:
warmup_bars 等於 period 只保證「算得出值」,不保證「值已經穩定」。任何跨實作要對得起來的場合,安全的暖機長度是 20 倍週期左右。這件事會在 Day 15 設計 pipeline 的 warm-up 處理時再出現一次。現在回答本篇最開始那個問題:把 alpha=1/14 寫成 span=14,實際上差多少?同一段序列,兩種寫法並排:
| 比較項目 | 數值 |
|---|---|
| 誤差中位數 | 4.94 點 |
| 誤差 90% 分位 | 10.05 點 |
| 最大誤差 | 15.97 點 |
| 誤差超過 5 點的 K 線比例 | 49.2% |
| 向上穿越 70 的次數(Wilder / 錯誤版) | 16 / 65 |
| 向下穿越 30 的次數(Wilder / 錯誤版) | 28 / 83 |
近一半的 K 線差超過 5 個 RSI 點,而穿越 70 的次數是四倍。
值得注意的是這種錯誤的症狀:它不會噴例外,輸出仍然乖乖落在 0 到 100 之間,畫出來的圖形狀也很像 RSI,波動大一點而已。肉眼幾乎看不出來。這正是 Day 01 講的第二個常見錯誤——把指標算錯卻沒發現——最具體的例子。沒有對照組,這件事不會被發現。
四種情況要處理,而且每一種都要有測試。
資料不足。 少於 period + 1 根時算不出任何有效值。這裡的選擇是回傳等長的全 NaN,而不是丟例外。理由是 Day 15 的 pipeline 會一次算十幾個特徵,各自的暖機長度不一樣,用 NaN 表達「這裡還沒有值」比讓呼叫端到處 try/except 乾淨得多。順帶一提,period 本身不合法(例如 0)走的是另一條路:那是建構參數,Indicator.__init__ 直接丟 ValueError,連物件都建不出來。
分母為 0。 全部上漲時平均跌幅是 0,RS = 平均漲幅 / 0。這裡就看出兩種公式寫法的差別了:
100 − 100 / (1 + RS):RS 變成 inf,pandas 對浮點除以 0 不會丟例外,最後算出 100,答案碰巧是對的,但那是靠 inf 的傳播僥倖拿到的。100 × 平均漲幅 / (平均漲幅 + 平均跌幅):分母是 平均漲幅 + 0,直接得到 100,中間不出現任何 inf。所以實作選第二種。真正需要顯式處理的是完全持平:漲幅與跌幅都是 0,兩種寫法都會得到 0/0 = NaN。這種情況實務上會出現在冷門交易對或交易所維護期間,補中性值 50 是合理的處理,但要在 docstring 裡寫清楚,不要讓呼叫端以為 50 是「算出來的」。
>>> indicator = RSI(14)
>>> indicator.compute(pd.DataFrame({"close": np.arange(100, 140, 1.0)})).iloc[-1]
100.0
>>> indicator.compute(pd.DataFrame({"close": np.arange(140, 100, -1.0)})).iloc[-1]
0.0
>>> indicator.compute(pd.DataFrame({"close": [100.0] * 40})).iloc[-1]
50.0
資料有缺漏。 這一個最危險,因為它不會以任何形式報錯。把中間五根收盤價改成 NaN 再算一次:
gapped = df.copy()
gapped.iloc[300:305, 0] = np.nan
輸出的 NaN 只有暖機期那 14 個,缺漏那五根不是 NaN,而是原封不動重複缺漏前的最後一個值(44.604)。原因是 ewm() 預設會跳過 NaN 觀測值,平均值就停在那裡不動。跟完整資料相比,最大偏差 6.03 個 RSI 點,之後慢慢收斂:
| 缺漏之後第幾根 | 與完整資料的差異 |
|---|---|
| 10 | 1.27 點 |
| 50 | 0.03 點 |
| 100 | 0.002 點 |
Day 01 提過的「一根缺漏的 K 線可能讓均線算錯十天,而且不會收到任何錯誤訊息」,在 RSI 上就是這個樣子。結論不是要在指標裡補值——指標不該負責修資料——而是缺漏必須在入庫之前就被偵測並處理掉。那是 Day 07 和 Day 08 的事,今天要做的是寫一個測試把這個行為釘住,讓它至少變成「已知且有紀錄的行為」,而不是某天有人踩到才發現。
# tests/domain/indicators/test_rsi.py
import numpy as np
import pandas as pd
import pytest
from quantbot.domain.entities.candle_series import CandleSeries
from quantbot.domain.indicators.ema import EMA
from quantbot.domain.indicators.registry import INDICATORS
from quantbot.domain.indicators.rsi import RSI, WilderSmoother
from quantbot.domain.values.instrument import Instrument
from quantbot.domain.values.market import Market
from quantbot.domain.values.timeframe import Timeframe
from tests.reference.reference_rsi import ReferenceRSI
PERIOD = 14
# 起始值差異需要約 20 倍週期才會衰減到浮點精度以下,跨實作比對前要丟掉這段
STABLE_AFTER_BARS = 300
INSTRUMENT = Instrument(
symbol="BTC/USDT", market=Market.SPOT, timeframe=Timeframe("1h")
)
def make_series(closes: np.ndarray | list[float]) -> CandleSeries:
values = list(closes)
index = pd.date_range(
"2026-01-01", periods=len(values), freq="1h", tz="UTC", name="open_time"
)
return CandleSeries(
INSTRUMENT,
pd.DataFrame(
{
"open": values,
"high": values,
"low": values,
"close": values,
"volume": 1.0,
},
index=index,
),
)
def random_closes(count: int = 2000, seed: int = 20260920) -> np.ndarray:
generator = np.random.default_rng(seed)
return 60_000 * np.exp(np.cumsum(generator.normal(0, 0.004, count)))
def test_matches_the_loop_reference():
series = make_series(random_closes())
assert np.allclose(
RSI(PERIOD).compute(series).to_numpy(),
ReferenceRSI(PERIOD).compute(series.frame["close"]).to_numpy(),
equal_nan=True,
atol=1e-10,
)
def test_wilder_alpha_is_one_over_n_not_two_over_n_plus_one():
"""寫成 span=n 的話這裡會變成 EMA 的係數,是本篇最容易出的錯。
兩個係數並排斷言,比寫十行註解有用:n=14 時差了將近一倍。
"""
assert WilderSmoother(PERIOD).alpha == pytest.approx(1 / PERIOD)
assert EMA(PERIOD).alpha == pytest.approx(2 / (PERIOD + 1))
assert EMA(PERIOD).alpha > WilderSmoother(PERIOD).alpha * 1.8
def test_matches_pandas_ta_after_the_warmup():
pandas_ta = pytest.importorskip("pandas_ta")
series = make_series(random_closes())
difference = (
RSI(PERIOD).compute(series)
- pandas_ta.rsi(series.frame["close"], length=PERIOD)
).abs()
assert difference.iloc[STABLE_AFTER_BARS:].max() < 1e-9
def test_output_contract():
series = make_series(random_closes(100))
indicator = RSI(PERIOD)
result = indicator.compute(series)
assert result.index.equals(series.frame.index) # index 不動
assert result.name == f"rsi_{PERIOD}" # 命名慣例
assert indicator.warmup_bar_count == PERIOD # 暖機期問得到
assert result.iloc[:PERIOD].isna().all()
assert result.iloc[PERIOD:].notna().all()
assert result.iloc[PERIOD:].between(0, 100).all()
def test_registry_builds_a_working_indicator():
"""Day 15 的 pipeline 只會這樣用它:用字串建物件、問暖機、再算。"""
indicator = INDICATORS["rsi"](PERIOD)
assert isinstance(indicator, RSI)
assert indicator.warmup_bar_count == PERIOD
assert indicator.compute(make_series(random_closes(100))).name == f"rsi_{PERIOD}"
def test_registry_warmup_is_the_longest_of_the_chain():
wanted = [("ema", 12), ("ema", 26), ("rsi", 14)]
indicators = [INDICATORS[name](period) for name, period in wanted]
assert max(indicator.warmup_bar_count for indicator in indicators) == 26
def test_insufficient_data_returns_all_nan():
result = RSI(PERIOD).compute(make_series(random_closes(PERIOD)))
assert len(result) == PERIOD and result.isna().all()
@pytest.mark.parametrize(
("closes", "expected"),
[
(np.arange(100, 140, 1.0), 100.0), # 全漲
(np.arange(140, 100, -1.0), 0.0), # 全跌
(np.full(40, 100.0), 50.0), # 完全持平,補中性值
],
)
def test_degenerate_series(closes, expected):
assert RSI(PERIOD).compute(make_series(closes)).iloc[-1] == pytest.approx(expected)
def test_gap_does_not_surface_as_nan():
"""釘住已知行為:缺漏不會變成 NaN,只會讓數值悄悄偏掉。
這是 ewm() 跳過 NaN 的結果。缺漏 MUST 在入庫階段處理(Day 07-08),
指標不負責修資料。
"""
closes = random_closes()
complete = make_series(closes)
gapped_closes = closes.copy()
gapped_closes[300:305] = np.nan
gapped = make_series(gapped_closes)
result = RSI(PERIOD).compute(gapped)
assert result.iloc[300:305].notna().all()
assert result.iloc[300:305].eq(result.iloc[300]).all()
assert (RSI(PERIOD).compute(complete) - result).abs().max() > 1.0
def test_rejects_bad_input():
with pytest.raises(ValueError):
RSI(period=0) # 建構時就擋掉
with pytest.raises(KeyError):
RSI(PERIOD, column="typical_price").compute(make_series(random_closes(50)))
test_gap_does_not_surface_as_nan 這種測試值得多說一句。它斷言的不是「正確行為」,是「已知的錯誤行為」。把它寫下來的意義在於:哪天有人改了實作、缺漏突然變成 NaN 或被補值了,這個測試會紅,那個決定就得重新被想一次。沒寫下來的話,這種行為改變是靜悄悄的。
test_registry_warmup_is_the_longest_of_the_chain 則是先把 Day 15 的用法寫出來:一條特徵鏈要丟掉前面幾根,是問出來的,不是算完之後猜的。
三個指標到今天到齊,而且它們從 Day 04 就繼承同一個 Indicator,所以不需要「回頭統一簽章」這個步驟——今天要補的只有一張表:
# quantbot/domain/indicators/registry.py
from collections.abc import Mapping
from types import MappingProxyType
from quantbot.domain.indicators.ema import EMA
from quantbot.domain.indicators.indicator import Indicator
from quantbot.domain.indicators.rsi import RSI
from quantbot.domain.indicators.sma import SMA
# 註冊表裡放的是**類別**而不是函式,所以取出來之後可以先問它問題、再算。
# Day 15 會把它擴充成完整的特徵註冊表。
INDICATORS: Mapping[str, type[Indicator]] = MappingProxyType(
{
"sma": SMA,
"ema": EMA,
"rsi": RSI,
}
)
註冊表裡放的是類別而不是函式,所以取出來之後可以先問它問題、再算:
wanted = [("ema", 12), ("ema", 26), ("rsi", 14)]
indicators = [INDICATORS[name](period) for name, period in wanted]
warmup = max(indicator.warmup_bar_count for indicator in indicators) # 26
features = pd.concat([indicator.compute(series) for indicator in indicators], axis=1)
features = features.iloc[warmup:] # 暖機期一次切掉
那個 warmup 是把類別當註冊項的直接好處:整條特徵管線需要多少暖機資料,可以在算之前問出來,不必等到算完再猜哪幾根不能用。這幾行就是 Day 15 的雛形,到那天要補的是快取、參數驗證,以及讓字串設定能從 YAML 讀進來。
順帶提一件跟分層有關的事:INDICATORS 住在 domain/indicators/,而不是 application。理由是它描述的是「這個系統認得哪些指標」,那是領域知識;至於「這次要算哪幾個」是設定,屬於 application 與設定檔。兩者混在一起的話,加一個指標就得改到用例。
動能指標的單位跟價格不一樣,不能疊在 K 線上,要另外開子圖。用 Plotly 的 make_subplots 做上下兩格、共用 x 軸,這樣縮放時兩邊會同步:
# quantbot/infrastructure/charting/plotly_relative_strength_chart_renderer.py
from __future__ import annotations
from typing import ClassVar
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from quantbot.domain.entities.candle_series import CandleSeries
from quantbot.domain.indicators.rsi import RSI
class PlotlyRelativeStrengthChartRenderer:
"""上格 K 線、下格 RSI,共用同一條 x 軸。
動能指標的單位跟價格不一樣,不能疊在 K 線上,所以要另外開子圖。
三條水平線的位置與顏色是類別常數:70 / 30 是慣例閾值,50 是中線。
"""
LEVELS: ClassVar[tuple[tuple[int, str, str], ...]] = (
(70, "dash", "#c0392b"),
(50, "dot", "#95a5a6"),
(30, "dash", "#27ae60"),
)
def __init__(self, *, period: int = 14) -> None:
self._indicator = RSI(period)
def render(self, series: CandleSeries) -> go.Figure:
line = self._indicator.compute(series)
candles = series.frame
figure = make_subplots(
rows=2,
cols=1,
shared_xaxes=True,
vertical_spacing=0.04,
row_heights=[0.7, 0.3],
subplot_titles=(
series.instrument.storage_key,
f"RSI({self._indicator.period})",
),
)
figure.add_trace(
go.Candlestick(
x=candles.index,
open=candles["open"],
high=candles["high"],
low=candles["low"],
close=candles["close"],
name=series.instrument.symbol,
),
row=1,
col=1,
)
figure.add_trace(
go.Scatter(x=line.index, y=line, name=str(line.name), line={"width": 1.4}),
row=2,
col=1,
)
for level, dash, color in self.LEVELS:
figure.add_hline(
y=level, line_dash=dash, line_color=color, line_width=1, row=2, col=1
)
figure.update_yaxes(title_text="價格(USDT)", row=1, col=1)
figure.update_yaxes(title_text="RSI", range=[0, 100], row=2, col=1)
figure.update_xaxes(title_text="時間(UTC)", row=2, col=1)
figure.update_layout(
height=760,
xaxis_rangeslider_visible=False,
hovermode="x unified",
showlegend=False,
)
return figure
一樣由 entrypoints/ 把指標、註冊表與圖表接起來。這支多做一件事:把超買超賣的停留長度統計出來,因為那個數字等一下要用來說明「超買不代表要跌」:
# quantbot/entrypoints/relative_strength_command.py
"""讀回補好的 parquet,算 RSI、統計超買超賣的停留長度,輸出雙軸互動圖。
uv run python -m quantbot.entrypoints.relative_strength_command \
--symbol BTC/USDT --market spot --timeframe 1h --period 14
"""
from __future__ import annotations
import argparse
from pathlib import Path
import pandas as pd
from quantbot.domain.entities.candle_series import CandleSeries
from quantbot.domain.indicators.registry import INDICATORS
from quantbot.domain.values.instrument import Instrument
from quantbot.domain.values.market import Market
from quantbot.domain.values.timeframe import Timeframe
from quantbot.infrastructure.charting.plotly_relative_strength_chart_renderer import (
PlotlyRelativeStrengthChartRenderer,
)
OVERBOUGHT = 70.0
OVERSOLD = 30.0
def parse_arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--symbol", default="BTC/USDT")
parser.add_argument("--market", default="spot", choices=[m.value for m in Market])
parser.add_argument("--timeframe", default="1h")
parser.add_argument("--period", type=int, default=14)
parser.add_argument("--source", type=Path, default=Path("data/klines"))
parser.add_argument("--out", type=Path, default=Path("notebooks"))
return parser.parse_args()
def longest_run(flags: pd.Series) -> int:
"""最長一段連續為真有幾根。用來回答「超買狀態能撐多久」。"""
blocks = (flags != flags.shift()).cumsum()
lengths = flags.groupby(blocks).sum()
return int(lengths.max()) if len(lengths) > 0 else 0
def main() -> int:
arguments = parse_arguments()
instrument = Instrument(
symbol=arguments.symbol,
market=Market(arguments.market),
timeframe=Timeframe(arguments.timeframe),
)
series = CandleSeries(
instrument,
pd.read_parquet(arguments.source / f"{instrument.storage_key}.parquet"),
)
# 從註冊表取類別再建實例,走的是 Day 15 特徵管線未來會走的那條路
line = INDICATORS["rsi"](arguments.period).compute(series)
valid = line.dropna()
overbought = valid > OVERBOUGHT
oversold = valid < OVERSOLD
print(f"{len(series)} 根 K 線:{series.open_times[0]} → {series.open_times[-1]}")
print(
f"RSI > {OVERBOUGHT:.0f}:{int(overbought.sum())} 根"
f"({overbought.mean():.2%}),最長連續 {longest_run(overbought)} 根"
)
print(
f"RSI < {OVERSOLD:.0f}:{int(oversold.sum())} 根"
f"({oversold.mean():.2%}),最長連續 {longest_run(oversold)} 根"
)
arguments.out.mkdir(parents=True, exist_ok=True)
chart_path = arguments.out / f"day06-{instrument.storage_key}-rsi.html"
PlotlyRelativeStrengthChartRenderer(period=arguments.period).render(
series
).write_html(chart_path)
print(f"圖:{chart_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
指標的實例是從 INDICATORS 取類別再建出來的,不是直接 RSI(14)。今天只有一個指標,看起來多繞一步,但這正是 Day 15 特徵管線會走的路徑,現在先讓它跑在真實資料上。
這支指令讀的是 parquet。手上還沒有 1 小時那份的話,先用 Day 03 的回補指令補一份——同一條管線,只換 --timeframe:
uv run python -m quantbot.entrypoints.backfill_command \
--symbol BTC/USDT --market spot --timeframe 1h \
--start 2025-01-01 --end 2026-08-02 --out data/klines
13872 根,缺 0 根,覆蓋率 100.0000%
Day 04、Day 05 已經補過的話,這一步會命中 data/raw/ 的快取,不會重下。接著:
uv run python -m quantbot.entrypoints.relative_strength_command \
--symbol BTC/USDT --market spot --timeframe 1h --period 14
13872 根 K 線:2025-01-01 00:00:00+00:00 → 2026-08-01 23:00:00+00:00
RSI > 70:675 根(4.87%),最長連續 23 根
RSI < 30:738 根(5.33%),最長連續 23 根
圖:notebooks/day06-spot_BTCUSDT_1h-rsi.html
這三行數字剛好把前面那節「超買不代表要跌」變成可查證的東西。一年半的 1 小時線上,RSI 待在 70 以上的時間只佔 4.87%——所以它確實是個少見的狀態,這是 70 這個閾值有意義的部分。但最長一段連續 23 根,也就是將近一整天,價格在這 23 小時裡並沒有義務回頭。看到 RSI 破 70 就準備放空,這 23 根就是要承受的東西,而且它不是極端個案,是同一份樣本裡真實出現過的最長紀錄。超賣那側的形狀幾乎對稱。
打開產出的 html,把游標移到 RSI 衝過 70 的那幾段,往右看價格接下來走了什麼。有幾次確實在那之後回落,也有幾次 RSI 在 70 以上待了二十幾根、價格一路往上。兩種都有,而且後者不少。
這是「RSI 是描述,不是預測」這句話在資料上的樣子。
quantbot/
├── domain/indicators/
│ ├── indicator.py Day 04
│ ├── sma.py / ema.py Day 04 / Day 05
│ ├── rsi.py 今天:RSI + WilderSmoother
│ └── registry.py 今天:INDICATORS
├── infrastructure/charting/
│ └── plotly_relative_strength_chart_renderer.py 今天
├── entrypoints/
│ └── relative_strength_command.py 今天:組裝根
└── tests/
├── reference/reference_rsi.py 今天
└── domain/indicators/test_rsi.py 今天
六項全過才算完成:
uv run pytest tests/domain/indicators/test_rsi.py 全綠,包含資料不足、全漲、全跌、完全持平、缺漏五種邊界。RSI(14).compute(series) 與 ReferenceRSI(14) 的最大誤差小於 1e-10。pandas-ta 的話,丟掉前 300 根之後的最大誤差小於 1e-9;沒裝的話該測試 skip 而不是 fail。SMA、EMA、RSI 都繼承 Indicator,INDICATORS 能用字串取到任何一個並建出實例,compute() 回傳的 Series 名稱符合 {indicator}_{period},warmup_bar_count 問得出來。backfill_command 補一份(--timeframe 1h,缺漏要是 0 根),再跑 uv run python -m quantbot.entrypoints.relative_strength_command --symbol BTC/USDT --market spot --timeframe 1h --period 14。它會印出超買超賣的根數與最長連續長度,並產出 notebooks/day06-spot_BTCUSDT_1h-rsi.html:上格 K 線、下格 RSI 與 70/50/30 三條水平線,x 軸縮放時兩格同步。uv run mypy quantbot 與 uv run lint-imports 全過。第一階段的指標到今天收尾,這是它們進 Day 15 特徵管線前的最後一道把關。第 4 項是今天真正的重點。前三項是 RSI 本身算得對不對,第 4 項是三天下來累積出的介面能不能撐住後面二十幾天。Day 15 會有八個東西要塞進同一套介面,那時候要是還得回頭改 domain/indicators/ 底下三個類別的介面,連帶所有測試與 notebook 都要動。
免責聲明:本文為程式與資料工程的技術分享,所有數字皆為教學範例,不構成投資建議;RSI 是對已發生價格變動的描述,不預測後續走勢。
第一階段的三個指標到今天告一段落。從 Day 02 到現在,資料一直是躺在 parquet 檔裡的:抓下來、存成檔案、要用的時候整份讀進 pandas。
這個做法在只有一個交易對、一種 timeframe、幾千根 K 線的時候完全沒問題。但明天要面對的情況是:五個交易對、三種 timeframe,每天有新資料要補進去,而查詢只想拿「2026 年 3 月那一週的 1 小時線」。用檔案做這件事,會開始出現「先列出哪些檔案的日期範圍可能相交、各自讀進來、concat、去重、再篩時間」這種程式碼,而且每加一個 timeframe 就要重寫一次。
明天 Day 07,我們談為什麼時序資料庫才是這種資料該待的地方:市場資料只增不改、幾乎都用時間範圍查、量隨時間線性成長,這三個特性剛好落在 TimescaleDB 的 hypertable 分區設計上。也會講為什麼不是純 PostgreSQL、不是 InfluxDB,以及一個之後會很有用的東西——continuous aggregate 讓 1 分鐘資料自動聚合出 5 分鐘與 1 小時線,不必自己算,也就不會算錯。