Table of Contents
TEJToolAPI 取得的資訊導入 Pipeline，並將其使用於策略架構。當今的股市，股價往往都因為大戶進出場而有大幅的波動，研究籌碼面的變動已成投資股市的一大課題，這當中董監轉讓持股對公司的股價即可能造成很大的影響，其原因在於董監對公司的營運有較為深入的了解，因此只要有董監轉讓持股的訊息釋出，投資人便會跟隨董監進出場，以賺取波段獲利。
根據臺灣證券交易所統計，2024 年第一季成交值為新台幣 1 億元以下的自然人有 487 萬 9506 人，創 7 年新高，顯示市場買氣濃厚，股市也不斷創新高，因此如何挑選好的進場時機避免追高也成了重要的投資關鍵。本次範例將結合上述董監持股轉讓的概念，以董監轉讓時機作為進場點，並利用 TQuant Lab 回測董監轉讓策略的績效，讓我們來一探究竟吧！
本範例以 MSCI 成分股作為股票池，並挑選董監轉讓方式中的：一般交易、鉅額逐筆交易以及一般交易鉅額逐筆交易作為進場依據，利用以下的規則建立董監轉讓策略並進行回測：
本文使用 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 成分股：
get_universe 函式取得 2018-05-01 到 2019-05-01 一年內所有上市櫃普通股的股票代碼。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 取得每檔股票的轉讓起始日、轉讓結束日以及轉讓方式的日資料。
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
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_EQUITIESPipeline() 提供使用者快速處理多檔標的的量化指標與價量資料的功能，於本次案例我們用以處理：
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() 函式用於定義交易開始前的每日交易環境，與此例中我們設置：
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() 為構建交易策略的重要函式，會在回測開始後每天被呼叫，主要任務為設定交易策略、下單與紀錄交易資訊。
本範例運用每日董監轉讓與否作為進場訊號，並透過 RSI 找出市場超買點進行賣出。
董監轉讓策略的進出場規則：
# 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
)
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 的相關方案，用高品質的資料庫，建構出適合自己的交易策略。
