本文重點摘要
- 文章難度：★★☆☆☆
- 本文將簡介如何將 TEJToolAPI 取得的資訊導入 Pipeline，並將其使用於策略架構。
- 透過董監持股轉讓時機找尋進場點，形成董監轉讓策略，並利用 TQuant Lab 回測其績效表現。
前言
當今的股市，股價往往都因為大戶進出場而有大幅的波動，研究籌碼面的變動已成投資股市的一大課題，這當中董監轉讓持股對公司的股價即可能造成很大的影響，其原因在於董監對公司的營運有較為深入的了解，因此只要有董監轉讓持股的訊息釋出，投資人便會跟隨董監進出場，以賺取波段獲利。
根據臺灣證券交易所統計，2024 年第一季成交值為新台幣 1 億元以下的自然人有 487 萬 9506 人，創 7 年新高，顯示市場買氣濃厚，股市也不斷創新高，因此如何挑選好的進場時機避免追高也成了重要的投資關鍵。本次範例將結合上述董監持股轉讓的概念，以董監轉讓時機作為進場點，並利用 TQuant Lab 回測董監轉讓策略的績效，讓我們來一探究竟吧！
董監轉讓策略
本範例以 MSCI 成分股作為股票池，並挑選董監轉讓方式中的：一般交易、鉅額逐筆交易以及一般交易鉅額逐筆交易作為進場依據，利用以下的規則建立董監轉讓策略並進行回測：
- 若發生董監轉讓，則買入或加碼 15,000 元。 p.s. 本策略預設本金為 1,000,000 元。
- 若 RSI > 95，代表市場超買情緒濃厚，因此賣出一半的部位。
編輯環境與模組需求
本文使用 Windows 11 並以 Jupyter Lab 作為編輯器。
載入套件與環境設定
import os import numpy as np import pandas as pd # tej_key tej_key = 'your key' api_base = 'https://api.tej.com.tw' os.environ['TEJAPI_KEY'] = tej_key os.environ['TEJAPI_BASE'] = api_base
抓取 MSCI 成分股
我們透過以下兩個步驟取得 MSCI 成分股：
- 利用 get_universe 函式取得 2018-05-01 到 2019-05-01 一年內所有上市櫃普通股的股票代碼。
- 利用 TEJ Tool API 篩選上述股票中屬於 MSCI 成分股的股票，作為董監轉讓策略的股票池。 from zipline.sources.TEJ_Api_Data import get_universe pool = get_universe(start = '2018-05-01', end = '2019-05-01', mkt = ['TWSE', 'OTC'], stktp_e=['Common Stock-Foreign', 'Common Stock'] ) len(pool) import TejToolAPI df_MSCI = TejToolAPI.get_history_data(start = '2018-05-01', end = '2019-05-01', ticker = pool, columns = ['Component_Stock_of_MSCI_TW_Fg'], transfer_to_chinese = False) df_MSCI = df_MSCI.drop_duplicates(subset=['coid']) df_filtered = df_MSCI[df_MSCI['Component_Stock_of_MSCI_TW_Fg'] == 'Y'] MSCI_pool = df_filtered['coid'].to_list() # MSCI stocks len(MSCI_pool)
導入股票池價量資料
回測期間從 2019-05-02 至 2024-05-01，共五年，並導入上述 89 檔 MSCI 成分股的價量資料與加權股價報酬指數 ( IR0001 ) 作為績效比較基準。
start = '2019-05-02' end = '2024-05-01' os.environ['mdate'] = start + ' ' + end os.environ['ticker'] = ' '.join(MSCI_pool) + ' ' + 'IR0001' !zipline ingest -b tquant
使用 TEJ Tool API 取得董監轉讓資料
在這裡，我們先利用 TEJ Tool API 取得每檔股票的轉讓起始日、轉讓結束日以及轉讓方式的日資料。
import TejToolAPI columns = ['Transfer_Method_English', 'Transfer_Start_Date', 'Transfer_End_Date'] data = TejToolAPI.get_history_data(start = start, end = end, ticker = MSCI_pool, columns = columns, transfer_to_chinese = False ) data = data.sort_values(['coid','mdate']) data
接著，我們將原始資料加上 transfer 欄位，代表當天是否有進行一般交易、鉅額逐筆交易或一般交易鉅額逐筆交易。
tran_key = ['Trade', 'Great each a transaction trade', 'Trade/Fix the price after the plate'] tran_data = data[data['Transfer_Method_English'].isin(tran_key)] tran_data['transfer'] = False tran_data.loc[tran_data['Transfer_Start_Date'].shift(1) != tran_data['Transfer_Start_Date'], 'transfer'] = True data_all = pd.merge(data, tran_data[['coid', 'mdate', 'transfer']], on = ['coid', 'mdate'], how = 'left') data_all['transfer'] = data_all['transfer'].fillna(False) data_all
將每日轉讓訊息導入 Pipeline
CustomDataset 可以將資料庫中的內容導入 Pipeline 中，方便後續回測使用。於本範例我們用以將上述 transfer 欄位紀錄的每日轉讓資訊導入 Pipeline。擷取部分程式碼如下：
from zipline.pipeline.data.dataset import Column, DataSet from zipline.pipeline.domain import TW_EQUITIES class CustomDataset(DataSet): Transfer_Method_English = Column(object) transfer = Column(bool) domain = TW_EQUITIES
建立 Pipeline 函式
Pipeline() 提供使用者快速處理多檔標的的量化指標與價量資料的功能，於本次案例我們用以處理：
- 利用 TQuant Lab 內建因子計算 RSI 值。
- 導入每日轉讓資訊。from zipline.pipeline import Pipeline from zipline.pipeline.filters import StaticAssets from zipline.pipeline.factors import RSI, ExponentialWeightedMovingAverage benchmark_asset = bundle.asset_finder.lookup_symbol('IR0001',as_of_date = None) def make_pipeline(): return Pipeline( columns = { 'price': TWEquityPricing.close.latest, 'RSI': RSI(inputs = [TWEquityPricing.close], window_length = 14), 'Transfer_Method_English': CustomDataset.Transfer_Method_English.latest, 'transfer': CustomDataset.transfer.latest }, screen = ~StaticAssets([benchmark_asset]) ) start_dt = pd.Timestamp(start, tz = 'UTC') end_dt = pd.Timestamp(end, tz = 'UTC') pipeline_result = engine.run_pipeline(make_pipeline(), start_dt, end_dt) pipeline_result
建立 initialize 函式
initialize() 函式用於定義交易開始前的每日交易環境，與此例中我們設置：
- 流動性滑價
- 交易手續費
- 買入持有加權股價報酬指數 ( IR0001 ) 的報酬作為基準
- 設定交易不可出現空單部位
- 將上述計算的 Pipeline 導入交易流程中from zipline.finance import slippage, commission from zipline.utils.calendar_utils import get_calendar from logbook import Logger, StderrHandler, INFO from zipline.api import * # Set up 'log' so we can see trading details when backtesting. log_handler = StderrHandler(format_string='[{record.time:%Y-%m-%d %H:%M:%S.%f}]: ' + '{record.level_name}: {record.func_name}: {record.message}', level=INFO) log_handler.push_application() log = Logger('Algorithm') def initialize(context): set_slippage(slippage.VolumeShareSlippage()) set_commission(commission.Custom_TW_Commission(min_trade_cost = 20, discount = 1.0, tax = 0.003)) set_benchmark(symbol('IR0001')) set_long_only(on_error = 'fail') attach_pipeline(make_pipeline(), 'mystrategy')
建立 handle_data 函式
handle_data() 為構建交易策略的重要函式，會在回測開始後每天被呼叫，主要任務為設定交易策略、下單與紀錄交易資訊。
本範例運用每日董監轉讓與否作為進場訊號，並透過 RSI 找出市場超買點進行賣出。
董監轉讓策略的進出場規則：
Long Entry:
轉讓發生且現金水位充足，配置帳戶資金 15,000 元。
Short Entry:
RSI > 95，賣出一半的持股。
# Buy when there is no position, transfer occurs and there is enough cash. if stock_position == 0: if (transfer == True) and (cash_position > 15000): order_value(i, 15000) buy = True record(**{f'buy{sym}': buy}) # When there are short positions, cancel the order. # This is to prevent the unexpected occurrence of short positions. elif stock_position < 0: open_orders = get_open_orders() for asset in open_orders: for o in open_orders[asset]: cancel_order(o) elif stock_position > 0: # overweight if another transfer occur if (transfer == True) and (cash_position > 15000): order_value(i, 15000) buy = True record(**{f'buy{sym}': buy}) # sell half of our holding positions when rsi > 95 elif rsi > 95: order(i, -0.5 * stock_position) sell = True record(**{f'sell_{sym}': sell})
執行交易策略
使用 run_algorithm() 執行上述設定的董監轉讓策略，設置交易期間為 start_dt ( 2019-05-02 ) 到 end_dt ( 2024-05-01 )，所使用資料集為 tquant，初始資金為 1,000,000 元。其中輸出的 results 就是每日績效與交易的明細表。
from zipline import run_algorithm results = run_algorithm( start = start_dt, end = end_dt, initialize = initialize, bundle = 'tquant', analyze = analyze, capital_base = 1e6, handle_data = handle_data, custom_loader = custom_loader )
利用 Pyfolio 進行績效評估
import pyfolio as pf returns, positions, transactions = pf.utils.extract_rets_pos_txn_from_zipline(results) benchmark_rets = results.benchmark_return # Creating a Full Tear Sheet pf.create_full_tear_sheet(returns=returns, positions=positions, transactions=transactions, benchmark_rets=benchmark_rets )
藉由上表我們可以看到董監轉讓策略的年化報酬率達到 19.348 %，另外夏普比率與索提諾比率分別有 1.22 以及 1.75，加上 α 值為 0.06，代表董監轉讓策略在承受單位風險下確實能賺取超額報酬，在下行風險下也有一定的保本能力。觀察右圖，可以發現董監轉讓策略在 2021 年 4 月前的績效略輸大盤，而後正式超越大盤的績效，其原因在於策略初期還在建倉階段，因此績效不及大盤，不過當倉位建置完畢加上透過 RSI 在超買期間出場，能有效避免空頭風險，也能獲取超過大盤的績效。
長短部位曝險圖幫助我們了解資金於長短部位的使用狀況。透過上圖我們可以發現董監轉讓策略於前期屬於建倉階段，並且在 2021 年底，藉由 RSI 的輔助賣出了部分持股，一定程度上避免了 2022 年的空頭走勢。
結論
本次策略由董監轉讓持股進行發想，欲探討在股市屢創新高的股市中，跟隨董監進場是否有良好的擇時能力。我們透過董監轉讓時機加上 RSI 指標建構董監轉讓策略，另外也簡單介紹如何將 TEJToolAPI 取得的資料導入 Pipeline 中進行回測，最後在回測結果上，董監轉讓策略取得了略贏大盤的績效。不過要提醒投資朋友的是，本範例使用的 MSCI 股票屬於大型股票，若套用在中小型股，回測結果可能不如預期；另外，每次投入的金額也應評估當下的股價走勢與自身承受的風險程度而定，才能較穩健的獲取報酬。
溫馨提醒，本次策略與標的僅供參考，不代表任何商品或投資上的建議。之後也會介紹使用 TEJ 資料庫來建構各式指標，並回測指標績效，所以歡迎對各種交易回測有興趣的讀者，選購 TQuant Lab 的相關方案，用高品質的資料庫，建構出適合自己的交易策略。
【TQuant Lab回測系統】解決你的量化金融痛點
全方位提供交易回測所需工具