Table of Contents
賈布利‧瓦森（Gabriel Watson）是美國知名的成長動能型投資組合經理人，早年曾任職於摩根添惠與威廉歐尼爾公司，累積深厚的市場研究經驗。自 1998 年加入黑玫瑰資本管理公司後，他跳脫傳統價值投資窠臼，發展出一套以「營收動能與股價強勢」為核心的機械式選股方法 —— 「The Machine」，在資訊迅速變動、資金輪動快速的環境中，成功捕捉強勢股的主升段波動。
瓦森策略的核心精神，在於鎖定具備高速營收成長的企業，並要求其股價與成交量同步展現強勢，反映基本面成長已獲市場認同。這套方法曾在 1998 年至 1999 年期間成功選中 Netbank，僅不到一年股價漲幅即超過 700%，奠定其在動能投資領域的聲望。不同於價值型投資的低估進場邏輯，瓦森的方法重視「成長中的企業」，並以數據為依據進行全程量化操作，是一套風格明確、進出彈性且強調即時性的投資系統。
本研究以台灣證券交易所與櫃檯買賣中心掛牌之所有上市櫃公司為投資標的，蒐集2013年起的股價、財務報表與董監事持股等基礎資料，並完成資料清洗與整合。由於策略邏輯需引用最近三年之財務資訊，實際回測期間訂為2017年1月1日至2025年5月1日，僅對符合條件之樣本進行歷史績效模擬，以確保數據完整性與策略檢驗的嚴謹性。
本策略參考賈布利．瓦森（Gabriel Watson）於 Blackrose Capital 所提出的「營收成長 + 股價動能」選股邏輯，透過量化方式轉換為六項具體條件，並適應本地市場特性進行調整與排序評分。選股邏輯結合基本面成長性與市場認同度，目標為找出營運快速成長、股價強勢且市場關注度高的企業。
加入「近三月累計營收成長率」排名篩選，需在市場前 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 的相關方案，用高品質的資料庫，建構出適合自己的交易策略。
溫馨提醒，本次分析僅供參考，不代表任何商品或投資上的建議。
