跳轉到

Block 產生的程式碼 code

所屬章節:積木 Block 以 2026-09-21 實際執行結果為準(資料日 2026-09-18)

Overview

screener_tw() / screener_us() 回傳的 code 是這組積木的等價 python 程式碼。 用途有三個:

  1. 看懂積木實際算了什麼(積木名稱是中文描述,程式碼是真正的運算)
  2. 貼進回測腳本或自己的流程,不必依賴 api.Block
  3. 當作範本改寫成自訂條件

程式碼自給自足:作用域裡只要有已登入的 api,貼進 cell 就能直接跑, 不需要任何 import。它自己會做三件事:

  • 開頭 msmp = api.MSMP(美股是 usmsmp = api.USMSMP)取得資料源
  • 用到的 helper(_continue、index_str_to_date、_rank…)直接寫成函式定義, 和 _operators 一樣,不從 SnowyOwl 內部匯入
  • 只寫出這組積木真正用到的部分,沒用到的 helper 不會出現

執行完會留下一個 sig 變數,即訊號矩陣。

Manual Sample Input/Output

Example 1:取出並閱讀

input

b = api.Block.con_tw('日K<>日D', operator='>')
r = api.Block.screener_tw([b])
print(b)
print(r['dataframe'].shape, r['date'])
print(r['code'])

output

{'id': 't2', 'operator': '>'}
(2004, 5) 2026-09-18 00:00:00

msmp = api.MSMP

# Stock pool
pool = msmp.list['all']

_operators = {
    'crossup': lambda x, y: (x > y) & (x.shift(1) < (y if isinstance(y, (int, float)) else y.shift(1))),
    'crossdown': lambda x, y: (x < y) & (x.shift(1) > (y if isinstance(y, (int, float)) else y.shift(1))),
    '>': lambda x, y: x > y,
    '<': lambda x, y: x < y,
    '>=': lambda x, y: x >= y,
    '<=': lambda x, y: x <= y,
    '==': lambda x, y: x == y,
}

K = msmp.日_技術指標.K
D = msmp.日_技術指標.D
op_func = _operators['>']
sig1 = op_func(K, D)
# sig1 variable is ready

# Combine conditions
sig1 = sig1.loc[:, sig1.columns.isin(pool)]
sig = sig1

程式碼結構固定為這幾段(沒用到的段落會省略):

段落 內容
msmp = api.MSMP 資料源。美股是 usmsmp = api.USMSMP
# Helper functions 這組積木用到的 helper 函式定義,見 Example 5
# Stock pool 由 pool 參數展開成 msmp.list[...] 的運算
_operators 比較運算子的 lambda 表(crossup / crossdown 是穿越,不是單純比大小)
各積木的運算 一個積木一段,變數名為 sig1、sig2…
# Combine conditions 套用股池過濾,並把各 sigN 以 & 合併成 sig

Example 2:實際執行,驗證與 screener 結果一致

命名空間只要給 api,其餘由程式碼自己建立:

input

ns = {'api': api}
exec(r['code'], ns)

sig = ns['sig']
print('sig shape =', sig.shape)
print('最後一列 True 檔數 =', int(sig.iloc[-1].sum()))
print('screener 回報檔數  =', r['dataframe'].shape[0])

output

sig shape = (1000, 2874)
最後一列 True 檔數 = 2004
screener 回報檔數  = 2004

兩者相同,code 與 screener 是同一套運算。

執行後可用的名稱:

['api', 'msmp', 'pool', '_operators', 'K', 'D', 'op_func', 'sig1', 'sig']

Example 3:美股用 usmsmp

美股積木產生的程式碼開頭是 usmsmp = api.USMSMP,一樣只要給 api:

b = api.Block.con_us('今日收盤價<>開高低P%',
                     operator='>', field1='Open', percent=1)
r = api.Block.screener_us([b])

ns = {'api': api}
exec(r['code'], ns)
sig = ns['sig']

美股沒有股池,所以產生的程式碼不含 # Stock pool 段落。


Example 4:多積木時的合併段落

兩個積木時,# Combine conditions 會變成:

# Combine conditions
sig = sig1 & sig2
sig = sig.loc[:, sig.columns.isin(pool)]

& 是 pandas 的元素級 AND,兩個矩陣的 index/columns 會自動對齊。 不同頻率(日/月/季)的積木由 screener 內部先用 union + forward fill 對齊, 產生的程式碼中已反映對齊後的結果。


Example 5:helper 函式會寫進程式碼

通用積木(a1/a2/a3)與部分原生積木會呼叫 helper。這些函式直接以原始碼 形式寫進輸出,不是 import 進來的,所以貼到任何地方都能跑:

input

b1 = api.Block.con_tw(id='a1', left='日_K.Volume', operator='>', const=3000)
b2 = api.Block.con_tw(id='a1', left='日_三大法人.三大法人買賣超_張數',
                      operator='>', const=300, rank=30)
r = api.Block.screener_tw([b1, b2], pool=['tse', 'otc', '~etf'])
print(r['code'])

output(節錄開頭)

msmp = api.MSMP

# Helper functions
def _continue(sig, days):
    """連續 days 根都成立。days<=1 時原樣回傳。"""
    if days is None or days <= 1:
        return sig
    return sig.rolling(days).sum() == days

def index_str_to_date(df):
    """月/季/年頻的字串 index 轉成日頻公告日;日頻與常數原樣回傳。"""
    if not hasattr(df, 'index_str_to_date'):
        return df
    return df.index_str_to_date().ffill()

def _build_market_pool(pool):
    ...

def _rank(sig, value, n, operator, pool):
    ...

# Stock pool
ls_tse = msmp.list['tse']
...
helper 何時出現
_continue 積木有 continue_days 參數時
index_str_to_date 讀月/季/年頻資料時(把字串 index 轉成公告日)
_rank 積木有 rank 參數時
_build_market_pool _rank 的相依,會自動一起帶出來

沒用到的不會出現——第一個例子(日K<>日D)就完全沒有 # Helper functions 段落。


Example 6:全部 665 個積木都驗證過

台股 405 + 美股 260,每個積木都實際跑過一次 screener 並檢查產生的程式碼:

con() 建構        665 / 665 成功
screener 實際選股  665 / 665 成功
產生的 code 可編譯  665 / 665 通過
自由名稱檢查       0 個未定義(除了 api)

所以 code 可以直接信任——貼進 cell 執行的結果,與 screener 回報的一致。