賈布利‧瓦森因子模型:量化動能投資的成長公式
前言
賈布利‧瓦森(Gabriel Watson)是美國知名的成長動能型投資組合經理人,早年曾任職於摩根添惠與威廉歐尼爾公司,累積深厚的市場研究經驗。自 1998 年加入黑玫瑰資本管理公司後,他跳脫傳統價值投資窠臼,發展出一套以「營收動能與股價強勢」為核心的機械式選股方法 —— 「The Machine」,在資訊迅速變動、資金輪動快速的環境中,成功捕捉強勢股的主升段波動。
瓦森策略的核心精神,在於鎖定具備高速營收成長的企業,並要求其股價與成交量同步展現強勢,反映基本面成長已獲市場認同。這套方法曾在 1998 年至 1999 年期間成功選中 Netbank,僅不到一年股價漲幅即超過 700%,奠定其在動能投資領域的聲望。不同於價值型投資的低估進場邏輯,瓦森的方法重視「成長中的企業」,並以數據為依據進行全程量化操作,是一套風格明確、進出彈性且強調即時性的投資系統。
投資標的 & 回測期間
本研究以台灣證券交易所與櫃檯買賣中心掛牌之所有上市櫃公司為投資標的,蒐集2013年起的股價、財務報表與董監事持股等基礎資料,並完成資料清洗與整合。由於策略邏輯需引用最近三年之財務資訊,實際回測期間訂為2017年1月1日至2025年5月1日,僅對符合條件之樣本進行歷史績效模擬,以確保數據完整性與策略檢驗的嚴謹性。
策略邏輯
本策略參考賈布利.瓦森(Gabriel Watson)於 Blackrose Capital 所提出的「營收成長 + 股價動能」選股邏輯,透過量化方式轉換為六項具體條件,並適應本地市場特性進行調整與排序評分。選股邏輯結合基本面成長性與市場認同度,目標為找出營運快速成長、股價強勢且市場關注度高的企業。
- 市值 > 市場第 80 百分位排除規模過小、資訊不透明或流動性風險較高的公司,策略首先以總市值作為基本門檻。具體方式為:選取市值在全市場前 80% 的公司,即排除落在市值排名最低 20% 的標的,以確保具備機構資金介入與法人追蹤的基礎條件。
- 近 12 個月營收成長率排名前 20%企業近一年的營收若快速成長,代表該公司近期基本面改善、產業需求回升或市占率擴張,有高度機會成為市場焦點。因此,本策略設定:「近 12 個月營收成長率」需位於市場前 20%(> 第80 百分位)。
- 近三年營收成長率排名前 40%短期成長若能與長期趨勢一致,將提高營運穩定性與持續性。因此,加入「近三年營收成長率」為條件,需排名在市場前 40%(> 第60 百分位),篩除僅短期爆發、長期成長性存疑的企業。
- 近 12 個月平均月成交量週轉率 > 市場第 60 百分位成交量週轉率是衡量個股市場關注度與流動性的重要指標。為避免選入交易冷清、難以實際進出之標的,本策略排除「近 12 個月平均月成交量週轉率」排名在市場最底部 40% 的股票(即僅保留前 60% 排名)。
- 近 12 個月股價漲幅排名前 20%價格是所有訊息的綜合反映,若公司股價在過去一年有強勁表現,代表市場已對其營收與成長前景做出積極評價。此策略僅納入「近 12 個月股價漲幅」位於市場前 20%(> 80 百分位)的個股。
- 近三月累計營收成長率排名前 40%
加入「近三月累計營收成長率」排名篩選,需在市場前 40%(> 60 百分位),補足傳統年度營收成長指標對於當前季度趨勢反應遲緩的問題。
實務操作上是通過上述的條件進行股票的篩選,篩選出來的股票以等權重的方式進行買入並且持有至下一次的再平衡日。再平衡天數的設計,此策略設定每21天進行換股。
股票篩選程式碼展示
導入套件
import pandas as pd import numpy as np import tejapi import os import datetime start='2013-01-01' end='2025-05-01' os.environ['TEJAPI_KEY'] = 'Your Key' from logbook import Logger, StderrHandler, INFO 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('get_universe')
匯入股票池(所有曾經上下市櫃的普通股)
from zipline.sources.TEJ_Api_Data import get_universe # 由文字型態轉為Timestamp,供回測使用 tz = 'UTC' start_dt, end_dt = pd.Timestamp(start, tz = tz), pd.Timestamp(end, tz = tz) from zipline.sources.TEJ_Api_Data import get_universe pool = get_universe(start = start_dt, end = end_dt, mkt=['TWSE','OTC'], stktp_e=['Common Stock-Foreign', 'Common Stock'])
匯入股票財務資訊
import TejToolAPI columns = ['主產業別_中文', '開盤價', '最高價','最低價', '收盤價', '本益比', '個股市值_元', '報酬率', '單月營收_千元', '近12月累計營收成長率', '近3月累計營收成長率', '成交量_千股', '流通在外股數_千股', '調整係數'] data__ = TejToolAPI.get_history_data(start = start_dt, end = end_dt, ticker = pool, fin_type = ['Q'], columns = columns, transfer_to_chinese = True)
處理資料
data = data.loc[:, ~data.columns.duplicated()] # 刪除重複欄位 data['日期'] = pd.to_datetime(data['日期'], format='%Y%m%d') # 將日期轉為datetime格式 data['成交量週轉率'] = data['成交量_千股']/ data['流通在外股數_千股'] # 計算成交量佔比 data['近12個月平均月成交量週轉率'] = data.groupby('股票代碼')['成交量週轉率'].rolling(252).mean().reset_index(0, drop=True) data['近12個月平均月成交量週轉率_排名'] = data.groupby('日期')['近12個月平均月成交量週轉率'].rank(pct=True) data['收盤價'] = data['收盤價'] * data['調整係數'] data['近12個月股價漲幅'] = data.groupby('股票代碼')['收盤價'].transform(lambda x: x.pct_change(252)) data['近12個月股價漲幅_排名'] = data.groupby('日期')['近12個月股價漲幅'].rank(pct=True) data['近12個月營收成長率_排名'] = data.groupby('日期')['近12月累計營收成長率'].rank(pct=True) data['近3年營收成長率'] = (data['單月營收_千元'] - data.groupby('股票代碼')['單月營收_千元'].shift(252*3)) / data.groupby('股票代碼')['單月營收_千元'].shift(252*3).abs() data['近3年營收成長率_排名'] = data.groupby('日期')['近3年營收成長率'].rank(pct=True) data['近3月累計營收成長率_排名'] = data.groupby('日期')['近3月累計營收成長率'].rank(pct=True) data['個股市值_元_排名'] = data.groupby('日期')['個股市值_元'].rank(pct=True) data['報酬率'] = data['報酬率']/100 data['sharpe_ratio'] = data.groupby('股票代碼')['報酬率'].rolling(252).mean().reset_index(0, drop=True) / data.groupby('股票代碼')['報酬率'].rolling(252).std().reset_index(0, drop=True) # 計算 True Range (TR) data['前日收盤價'] = data.groupby('股票代碼')['收盤價'].shift(1) data['TR1'] = data['最高價'] - data['最低價'] data['TR2'] = (data['最高價'] - data['前日收盤價']).abs() data['TR3'] = (data['最低價'] - data['前日收盤價']).abs() data['TR'] = data[['TR1', 'TR2', 'TR3']].max(axis=1) # ATR: 14日移動平均 data['ATR_14'] = data.groupby('股票代碼')['TR'].transform(lambda x: x.rolling(14).mean()) # 清理暫時欄位 data.drop(columns=['前日收盤價', 'TR1', 'TR2', 'TR3'], inplace=True)
匯入回測資料
from zipline.data.run_ingest import simple_ingest pools = pool + ['IR0001'] start_ingest = start.replace('-', '') end_ingest = end.replace('-', '') print(f'開始匯入回測資料') simple_ingest(name = 'tquant' , tickers = pools , start_date = start_ingest , end_date = end_ingest) print(f'結束匯入回測資料')
回測架構
from zipline.pipeline import Pipeline from zipline.pipeline.factors import Returns from zipline.pipeline.filters import SingleAsset from zipline.api import set_slippage, set_commission, set_benchmark, symbol, record, order_target_percent, pipeline_output, attach_pipeline from zipline.finance import commission, slippage def initialize(context): set_slippage(slippage.VolumeShareSlippage(volume_limit=1, price_impact=0.01))#調整volume_limit to 1 and price impact to 0.01 set_commission(commission.Custom_TW_Commission()) set_benchmark(symbol('IR0001')) # attach_pipeline(make_pipeline(), 'my_pipeline') context.i = 0 context.state = False context.order_tickers = [] context.last_tickers = [] def compute_stock(date, data): """ 根據指定的日期進行選股,篩選出符合條件的股票列表。 Parameters: date (str): 選股的日期,格式為 'YYYY-MM-DD'。 data (DataFrame): 包含股票數據的 DataFrame。 Returns: list: 符合條件的股票代碼列表。 """ # 提取指定日期的股票資訊 df = data[data['日期'] == pd.Timestamp(date)].reset_index(drop=True) # 條件 1:選取總市值>市場平 df['平均個股市值_元'] = df.groupby(['日期'])['個股市值_元'].transform('mean') # set_1 = set(df[df['個股市值_元'] > df['平均個股市值_元']]['股票代碼 ']) set_1 = set(df[df['個股市值_元_排名'] > 0.2 ]['股票代碼']) # 條件 2:最近12個月營收成長率>60%。 # set_2 = set(df[df['近12個月營收成長率'] > 0.60]['股票代碼']) set_2 = set(df[df['近12個月營收成長率_排名'] > 0.8]['股票代碼']) # 條件 3: 最近3年營收成長率>30%。 # set_3 = set(df[df['近3年營收成長率'] > 0.30]['股票代碼']) set_3 = set(df[df['近3年營收成長率_排名'] > 0.6]['股票代碼']) # 條件 4: 最近12個月平均月成交量週轉率>市場平均值件 4: 最近12個月平均月成交量週轉率>市場平均值 # df['平均成交量週轉率'] = df.groupby(['日期'])['近12個月平均月成交量週轉率'].transform('mean') # set_4 = set(df[df['近12個月平均月成交量週轉率'] > df['平均成交量週轉率']]['股票代碼']) set_4 = set(df[df['近12個月平均月成交量週轉率_排名'] < 0.6]['股票代碼']) # 條件 5: 最近12個月股價漲幅>75% # set_5 = set(df[df['近12個月股價漲幅'] > 0.75]['股票代碼']) set_5 = set(df[df['近12個月股價漲幅_排名'] > 0.8]['股票代碼']) set_6 = set(df[df['近3月累計營收成長率_排名'] > 0.6]['股票代碼']) tickers = list(set_1 & set_2 & set_3 & set_4 & set_5 & set_6) filtered_df = df[df['股票代碼'].isin(tickers)].copy() filtered_df = filtered_df[filtered_df['sharpe_ratio'].notna()] sorted_df = filtered_df.sort_values(by='sharpe_ratio', ascending=False) # 取前二十檔股票代碼,若不足二十檔則全部輸出 top_20 = sorted_df['股票代碼'].tolist()[:20] return top_20 rebalance_period = 21 # 調倉週期 def handle_data_1(context, data, rebalance = rebalance_period): # 避免前視偏誤,在篩選股票下一交易日下單 if context.state == True: print(f"下單日期:{data.current_dt.date()}, 擇股股票數量:{len(context.order_tickers)}") for i in context.last_tickers: if i not in context.order_tickers: order_target_percent(symbol(i), 0) for i in context.order_tickers: order_target_percent(symbol(i), 1 / len(context.order_tickers)) curr = data.current(symbol(i), 'price') record(price = curr, days = context.i) context.last_tickers = context.order_tickers context.state = False backtest_date = data.current_dt.date() # 若今天是月底交易日,執行轉倉前的擇股邏輯 if context.i % rebalance == 0: context.state = True context.order_tickers = compute_stock(date = backtest_date, data = data_) # e.g., ['2330', '2317', …] print(f"月末擇股名單:{context.order_tickers}") context.i += 1 import matplotlib.pyplot as plt import matplotlib def analyze(context, perf): PASS return from zipline import run_algorithm from zipline.utils.calendar_utils import get_calendar calendar_name = 'TEJ' import logging # 關掉 Zipline 訊息 logging.getLogger('zipline').setLevel(logging.WARNING) logging.getLogger('zipline.finance').setLevel(logging.WARNING) logging.getLogger('exchange_calendars').setLevel(logging.WARNING) start = pd.Timestamp('2017-01-01', tz = 'UTC') end_dt = pd.Timestamp('2025-05-01', tz = "UTC") results = run_algorithm( start = start_, end = end_dt, initialize = initialize, handle_data = handle_data_1, analyze = analyze, bundle = 'tquant', capital_base = 1e6, trading_calendar=get_calendar(calendar_name), data_frequency='daily')
查看績效
from pyfolio.utils import extract_rets_pos_txn_from_zipline returns, positions, transactions = extract_rets_pos_txn_from_zipline(results) benchmark_rets = results.benchmark_return # 時區標準化 returns.index = returns.index.tz_localize(None).tz_localize('UTC') positions.index = positions.index.tz_localize(None).tz_localize('UTC') transactions.index = transactions.index.tz_localize(None).tz_localize('UTC') benchmark_rets.index = benchmark_rets.index.tz_localize(None).tz_localize('UTC') import pyfolio from pyfolio.utils import extract_rets_pos_txn_from_zipline import matplotlib.pyplot as plt import matplotlib matplotlib.rcParams['font.family'] = 'SimHei' matplotlib.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False returns, positions, transactions = extract_rets_pos_txn_from_zipline(results) benchmark_rets = results.benchmark_return pyfolio.plot_gross_leverage(returns, positions) pyfolio.tears.create_full_tear_sheet(returns=returns, positions=positions, transactions=transactions, benchmark_rets=benchmark_rets )
績效圖表&分析
從 2017 至 2025 年的累積報酬圖來看,該策略在多數期間內顯著超越基準指數(benchmark),尤其在 2020~2024 年間展現強勁上漲趨勢,最終累計報酬高達508.783%,高於同期間大盤的累計報酬194.032%。
然而該策略的資金管理或部位控制在極端行情下仍需加強,例如2025年的空頭行情未能有停損機制,導致策略在劇烈震盪中大幅回吐獲利。
歡迎投資朋友參考,之後也會持續介紹使用 TEJ 資料庫來建構各式指標,並回測指標績效,所以歡迎對各種交易回測有興趣的讀者,選購 TQuant Lab 的相關方案,用高品質的資料庫,建構出適合自己的交易策略。
溫馨提醒,本次分析僅供參考,不代表任何商品或投資上的建議。
【TQuant Lab 回測系統】解決你的量化金融痛點
全方位提供交易回測所需工具