4. 用 HistoryPanel 研究縱向擇時因子
縱向擇時:因子閾值 -> bool mask -> 組合收益與 CAGR 對比
本篇教程面向一個非常常見、也非常“接地氣”的研究場景:我們手裏只有少量股票(幾隻到十幾只),想沿每隻股票自己的時間軸做出持有/不持有的擇時決策;然後把這些決策彙總成一條組合曲線,最後再和基準(HS300)對比,得到一個能複用的年化指標(CAGR)。
先把定位說清楚:HistoryPanel 在這裏扮演的是輕量因子研究容器。它的強項是把“數據 -> 條件 -> 研究口徑(mask)-> 組合聚合 -> 可視化解釋”打通,讓我們能快速驗證一個規則是不是值得繼續深挖。它不是交易回測引擎:不處理交易成本、交割、滑點、資金約束、最小成交單位等完整交易語義。本文的目標是做研究閉環,不是做交易閉環。
4.1. 0. 开场:先跑通一个最小可用的研究闭环
如下代碼所示,我們只用幾行就能拿到一個 HistoryPanel,並畫出第一張圖(不管是折線還是 K 線),證明“數據和可視化鏈路是通的”。
import qteasy as qt
stocks = ['000001.SZ', '600519.SH', '300750.SZ']
benchmark = '000300.SH'
shares = stocks + [benchmark]
hp = qt.get_kline(
shares=shares,
start='20220101',
end='20221231',
freq='D',
as_panel=True,
)
# 最小可跑:能出图即可
fig = hp.plot(interactive=True)
fig
如果你在 Notebook 裏運行,上面這一段通常就能立即看到一張圖。
不過,僅僅“能畫出來”還遠遠不夠。 如果我們真的要把它當成日常研究工具來用,至少會遇到下面這些問題:
擇時規則怎麼落地:我們腦子裏很容易有一句“MACD>0 就持有”,但寫成代碼後常常會變形:列名記錯、數據缺列、算出來全是 NaN、或不同股票對齊方式不一致。結果就是研究腳本一跑就報錯,或者更糟——不報錯但結論不可信。
mask 形狀怎麼對齊:條件本質是二維的(股票×時間),但
HistoryPanel是三維的(股票×時間×字段)。如果形狀對不齊,你會出現“以爲在篩選股票,其實在篩選字段”的隱性 bug;而這種 bug 往往不會立刻報錯,只會讓收益曲線看起來“有點怪”。沒有 benchmark 很難判斷好壞:只看組合曲線很容易“自嗨”。它可能只是踩中了 2022 年某一段風格或大盤趨勢。把
000300.SH拉進來做對照,至少能回答一個更接近實戰的問題:這套擇時規則,是在創造超額,還是隻是跟着市場波動?有結果但難解釋:就算 long 組合跑贏了,也很難解釋“爲什麼贏”。真正的研究需要可覆盤:當收益出現明顯拐點時,我們能不能迅速回到圖上定位“是哪些觸發點改變了持倉狀態”?
好在這些能力都可以一步步補齊。本文就按“先跑通,再增強”的節奏,把它補齊成一個真正可用的小研究流程。
4.2. 0.5 先貼最終效果(我們最後會得到什麼)
爲了讓節奏更穩,我們先把“終點形態”講清楚。按本文做完,你至少會得到兩類輸出:
組合層面的對比:
LONG / SHORT / 000300.SH三條曲線(統一從 1.0 起點歸一化)放在一張圖裏,你能一眼看出“擇時規則在研究期內是否有效”。單標的層面的解釋:對某隻股票畫 K 線(或價格曲線),並把“觸發持有條件的時點”用高亮標出來。這樣當你看到組合曲線某段表現異常時,可以快速回到單股圖上覆盤“觸發點是否符合直覺”。
注:本文不強制你輸出 GIF。更推薦的做法是:先在 Notebook 裏把圖跑通,截圖即可;如果你要寫博客或做分享,再把關鍵步驟錄成 GIF。
4.3. 1. 目标(我们这篇文章要完成什么)
在開始動手之前,我們先把目標講清楚,這樣每一步都知道自己在解決什麼問題。
獲取少量個股 +
000300.SH的HistoryPanel派生一個可解釋的擇時因子(示例:MACD)
用比較運算得到 bool 條件,並用
where()生成研究 mask用
portfolio(mask=...)聚合成組合曲線,並與 benchmark 對比用
normalize/cum_return得到累計收益,並推導 CAGR 摘要用
plot(highlight=...)把“觸發時點”解釋回圖上文末給出一段“單函數可跑”的完整代碼
明確研究口徑邊界:這不是交易回測引擎
整篇文章保持同一個節奏: 先講本節要解決什麼 -> 再說最小必要原理 -> 最後給關鍵代碼與預期效果。 重複代碼會適當省略,但每一節都保證可以順着文章直接復現。
4.4. 1.1 復現前置:數據是否已在本地準備好?
本文的示例默認你已經在本地配置好了數據源,並且 qt.get_kline() 能成功取到 20220101–20221231 的日線數據。
如果你的環境裏暫時沒有數據,最常見的現象是:qt.get_kline() 返回空面板,或者在後續計算指標時出現全 NaN。爲了避免“跑到一半才發現沒數據”,建議你在第 2 節做完後務必檢查:
hp.shape的時間長度是否大於 100(一年日線一般在 200 附近);hp.htypes是否至少包含open/high/low/close;hp.hdates是否連續覆蓋你想研究的區間。
如果你在自己的環境裏需要先下載數據,請先完成“數據下載與數據源配置”相關章節;本文不展開數據通道細節,專注在
HistoryPanel的研究鏈路本身。
4.5. 2. 准备数据:三只个股 + 一个基准指数(HS300)
2.1 本節要解決什麼
我們先把數據準備乾淨: 用 3 只個股做縱向擇時,同時把 000300.SH(滬深 300)加入面板,作爲後面 portfolio(..., benchmark=...) 的對照。
這裏最關鍵的一步是:確認 OHLC 列齊全。 因爲後面不管是畫 K 線、還是做形態識別,都繞不開 open/high/low/close。
2.2 最小必要原理
HistoryPanel 是三維數據:(share, time, htype)。 後續 where/mask/portfolio/cum_return 都會依賴“形狀對齊”的前提。 因此我們在一開始就列印 shape/shares/htypes,並做一個最低限度的字段校驗,可以把後面大部分“寫到一半才報錯”的坑提前堵掉。
2.3 可運行代碼 + 預期效果
import qteasy as qt
stocks = ['000001.SZ', '600519.SH', '300750.SZ']
benchmark = '000300.SH'
shares = stocks + [benchmark]
hp = qt.get_kline(
shares=shares,
start='20220101',
end='20221231',
freq='D',
as_panel=True,
)
print('hp.shape:', hp.shape)
print('hp.shares:', hp.shares)
print('hp.htypes:', hp.htypes)
print('last_date:', hp.hdates[-1])
required = {'open', 'high', 'low', 'close'}
missing = [c for c in required if c not in set(hp.htypes)]
if missing:
raise ValueError(f'Missing required OHLC columns in htypes: {missing}')
if benchmark not in hp.shares:
raise ValueError(f'Benchmark {benchmark} not found in shares: {hp.shares}')
# 额外做一个“是否真的有数据”的快速检查(避免空面板/全 NaN 继续往下跑)
if hp.shape[1] < 50:
raise ValueError(
'Not enough data points loaded (too few hdates). '
'Please check your local datasource and date range.'
)
你應該看到:
hp.shares裏包含 3 只個股 +000300.SHhp.htypes至少包含open/high/low/close(以及可能還有vol)
4.6. 3. 派生择时因子:MACD(把信号落成一列可复用的数据)
3.1 本節要解決什麼
我們先選一個足夠常見、足夠可解釋、也足夠“能馬上用”的擇時因子:MACD。 這一節的目標很簡單:把 MACD 計算出來,併成爲 HistoryPanel 的新列,讓後續條件篩選可以直接對列做比較。
3.2 最小必要原理
hp.kline.macd() 會返回一個新 HistoryPanel,並在 htypes 裏追加三列:
macd_12_26_9macd_signal_12_26_9macd_hist_12_26_9
注意這個命名:默認是帶後綴的,不是裸的 macd_hist。 這一步我們把列名列印出來,後面寫條件時就不會猜錯。
另外,很多朋友第一次做因子研究時,會不自覺地在不同地方“重複計算指標”。短期看沒問題,但一旦你開始加更多列、畫更多圖,就容易出現“同名列覆蓋/同義列多份”的混亂。更穩妥的做法是:先把指標算成列,明確寫進 htypes,後續的條件、聚合、可視化都只依賴這些列。
3.3 可運行代碼 + 預期效果
hp_macd = hp.kline.macd(price_htype='close', fastperiod=12, slowperiod=26, signalperiod=9)
print('new htypes (tail):', hp_macd.htypes[-6:])
預期:你會看到 macd_hist_12_26_9 出現在 htypes 裏。
4.7. 4. 因子阈值 -> bool 条件 -> where() 研究 mask
4.1 本節要解決什麼
現在我們把擇時規則寫清楚: 例如“MACD 柱狀圖大於 0 的時候屬於 long,小於等於 0 屬於 short”。 這一步要交付的產物是兩個 mask:
mask_long:哪些格點參與 long 組合聚合mask_short:哪些格點參與 short 組合聚合
4.2 最小必要原理
從 2.2.8 起,HistoryPanel 支持直接做比較運算: 例如 hp_macd > 0 會返回 numpy.ndarray(bool)。 而 hp.where(condition) 會把各種可廣播條件規整爲與 hp.values 同形的 (M,L,N) mask。
這一步非常關鍵:我們不是在“刪除數據”,而是在“定義研究口徑”: mask 爲 False 的格點,在 portfolio/cum_return 裏會被視作缺失(不參與聚合或導致路徑斷開)。 因此 mask 就是“研究規則本身”。
這裏我們再把“形狀”說得更直白一點:
你腦子裏的擇時規則是“每隻股票每天一個 True/False”,因此最自然的條件形狀是
(M, L)。但
HistoryPanel的 values 是(M, L, N),有 N 個字段列。where()做的事情就是:把你的條件擴展/廣播成(M, L, N),讓後續任何需要按格點過濾的計算都有一個統一的入口。
一旦你養成“條件都先過 where()”的習慣,後面把條件餵給 portfolio(mask=...)、cum_return(mask=...) 時,就不容易在形狀上踩坑。
4.3 可運行代碼 + 預期效果
import numpy as np
factor_col = 'macd_hist_12_26_9'
# 取单列子面板(形状 (M,L,1)),再与标量比较得到 bool ndarray
cond_long = (hp_macd[factor_col] > 0.0) # numpy.ndarray(bool)
cond_short = (hp_macd[factor_col] <= 0.0) # numpy.ndarray(bool)
# 规整为 (M,L,N) 研究 mask
mask_long = hp.where(cond_long)
mask_short = hp.where(cond_short)
print('cond_long shape:', getattr(cond_long, 'shape', None))
print('mask_long shape:', mask_long.shape)
print('mask_long dtype:', mask_long.dtype)
# 只做一个直觉检查:最后一天 long 有多少格点为 True
print('true_count_last_day(long):', int(mask_long[:, -1, 0].sum()))
預期:
mask_long.shape == hp.shapemask_long.dtype == bool
4.8. 5. 用 portfolio(mask=...) 聚合组合曲线,并与 benchmark 对比
5.1 本節要解決什麼
我們不做交易回測,只做研究向的“粗聚合”: 對每一天,把滿足條件的股票(long/short)聚合成一條組合曲線,然後把 000300.SH 作爲 benchmark 拉出來對比。
5.2 最小必要原理
HistoryPanel.portfolio() 的幾個關鍵點(我們用到的):
mask=:與where()同形規則;False 的格點不參與聚合benchmark=+benchmark_output='tag_along':把基準行追加到輸出中輸出仍是一個
HistoryPanel,時間軸不變
再次強調:這是研究向聚合,不含交易成本、無資金約束、也不做調倉執行。
你可以把它理解成一種“研究口徑上的等權籃子”:每天把符合條件的股票等權平均成一條曲線。它的意義不是模擬真實交易,而是回答一個更基礎的問題:
如果我用這個條件去定義一個“持有集合”,它在樣本期內表現有沒有系統性差異?
只有當這個問題的答案是“有”,我們才值得繼續投入,把它升級成真正的交易策略回測。
5.3 可運行代碼 + 預期效果
benchmark = '000300.SH'
pf_long = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_long,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='LONG',
)
pf_short = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_short,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='SHORT',
)
print('pf_long.shares:', pf_long.shares)
print('pf_long.htypes:', pf_long.htypes)
print('pf_long.shape:', pf_long.shape)
預期:pf_long.shares 裏會有 LONG 和 000300.SH 兩行。
4.9. 6. normalize / cum_return + CAGR:把曲线变成可比较的年化摘要
6.1 本節要解決什麼
兩條曲線擺在一起,人眼能看趨勢,但很難快速總結: “這一段研究期,long 比 benchmark 到底強多少?” 因此我們做兩件事:
normalize:統一起點(便於目視比較)cum_return + CAGR:給出一個可複用的年化摘要表
6.2 最小必要原理
normalize(base_index=0)會把基準點縮放到 1.0(研究口徑),非常適合畫同圖比較cum_return(method='simple')給出累計收益cumret_*CAGR 的本質是“等效年化增長率”:
[ \text{CAGR}=(1+R)^{1/T}-1 ]
其中 (R) 是區間累計收益(末值),(T) 是年數。
這裏我們再補一句“爲什麼要算 CAGR”:很多時候我們會拿不同長度的區間做對比(半年、一年、兩年)。如果只看累計收益,很容易“區間越長看起來越厲害”。CAGR 把它折算成“每年等效增長多少”,這樣不同研究期的結果才更容易橫向比較。
6.3 可運行代碼 + 預期效果
import pandas as pd
def _years_between(hdates) -> float:
idx = pd.DatetimeIndex(hdates)
days = (idx[-1] - idx[0]).days
return max(1e-9, days / 365.25)
def _cagr_from_cumret(cumret_end: float, years: float) -> float:
return (1.0 + cumret_end) ** (1.0 / years) - 1.0
years = _years_between(pf_long.hdates)
cr_long = pf_long.cum_return(htypes='close', method='simple') # 输出列 cumret_close
cr_short = pf_short.cum_return(htypes='close', method='simple')
cumret_long_end = float(cr_long.values[cr_long.shares.index('LONG'), -1, 0])
cumret_short_end = float(cr_short.values[cr_short.shares.index('SHORT'), -1, 0])
# benchmark 行同样在 shares 里(tag_along)
cumret_bm_long_end = float(cr_long.values[cr_long.shares.index('000300.SH'), -1, 0])
cumret_bm_short_end = float(cr_short.values[cr_short.shares.index('000300.SH'), -1, 0])
summary = pd.DataFrame(
{
'cum_return_end': [cumret_long_end, cumret_short_end, cumret_bm_long_end],
'CAGR': [
_cagr_from_cumret(cumret_long_end, years),
_cagr_from_cumret(cumret_short_end, years),
_cagr_from_cumret(cumret_bm_long_end, years),
],
},
index=['LONG', 'SHORT', '000300.SH'],
)
print(summary)
你應該能看到一張 3 行的摘要表。建議你至少關注兩件事:
LONG vs 000300.SH:是否真的跑贏(CAGR 更高/或累計收益更高)?
SHORT 的表現:它不一定要“虧”,但它能幫助我們判斷條件是否真的在分離樣本(LONG/SHORT 兩端的差異是否明顯)。
4.10. 7. 可视化解释:用 plot(highlight=...) 把“触发点”标回图上
7.1 本節要解決什麼
我們已經有收益曲線了,但還差最後一環:解釋。 這一步我們做一個“覆盤友好”的動作:在單隻股票的 K 線圖上,把觸發點高亮出來,讓讀者能一眼看到“什麼時候觸發了擇時條件”。
7.2 最小必要原理
HistoryPanel.plot(highlight=...) 支持兩種常用寫法:
简写:
highlight='max'/'min'显式:
highlight={'condition': <1D bool over time>, 'style': {...}}
注意一點:靜態渲染路徑裏 condition 更偏向 1D 時間軸(用於在圖上散點標記)。因此我們在這裏用單股時間軸的 1D bool 來做高亮更穩。
7.3 可運行代碼 + 預期效果
import numpy as np
primary_share = '000001.SZ'
# 从 cond_long(M,L,1 或 M,L,N 的 bool)里抽出该 share 的时间轴 1D 条件
si = hp_macd.shares.index(primary_share)
cond_1d = np.asarray(cond_long[si, :, 0], dtype=bool).ravel()
# 为了让图更清爽,这里只看最后 200 个交易日(你也可以改成全区间)
hp_one = hp.loc[-200:]
fig = hp_one.plot(
shares=[primary_share],
interactive=True,
highlight={'condition': cond_1d[-200:], 'style': {'marker': 'x', 's': 50}},
)
fig
預期效果:你會在圖上看到一系列被高亮標出的點(對應 macd_hist_12_26_9 > 0 的時點)。 這一步的價值在於:當你看到組合曲線某段突然變差時,可以回到單股圖快速確認:觸發點是否集中出現在震盪區、是否“來回打臉”,從而決定下一步要不要加過濾條件(例如趨勢過濾、波動過濾等)。
4.11. 8. 完整代码(单函数可跑版本)
下面給出一段“單函數可跑”的完整版本,方便你複製進 Notebook 一鍵運行。它做了三件事:
跑通
MACD -> mask -> portfolio -> CAGR的研究閉環;畫出
LONG/SHORT/benchmark的歸一化曲線;畫出單股圖並高亮觸發點(便於解釋與覆盤)。
import numpy as np
import pandas as pd
import qteasy as qt
def demo_vertical_timing(
stocks: list,
benchmark: str = '000300.SH',
start: str = '20220101',
end: str = '20221231',
primary_share: str = '000001.SZ',
):
\"\"\"演示纵向择时研究闭环:MACD 阈值 -> mask -> 组合曲线 -> CAGR -> 高亮解释。
Parameters
----------
stocks : list
个股代码列表(建议 3~15 只,太多不利于解释)。
benchmark : str, default '000300.SH'
基准指数代码(示例使用沪深 300)。
start : str, default '20220101'
起始日期(YYYYMMDD)。
end : str, default '20221231'
结束日期(YYYYMMDD)。
primary_share : str, default '000001.SZ'
用于做“触发点高亮解释”的单只股票代码。
Returns
-------
dict
结果对象集合,便于你在 Notebook 里继续查看:
- hp: 原始面板
- hp_macd: 含 MACD 列的面板
- pf_long/pf_short: 组合面板(含 benchmark 行)
- summary: CAGR 摘要表(DataFrame)
- fig_pf: 组合对比图
- fig_one: 单股高亮图
\"\"\"
shares = list(stocks) + [benchmark]
hp = qt.get_kline(
shares=shares,
start=start,
end=end,
freq='D',
as_panel=True,
)
required = {'open', 'high', 'low', 'close'}
missing = [c for c in required if c not in set(hp.htypes)]
if missing:
raise ValueError(f'Missing required OHLC columns in htypes: {missing}')
if benchmark not in hp.shares:
raise ValueError(f'Benchmark {benchmark} not found in shares: {hp.shares}')
if hp.shape[1] < 50:
raise ValueError(
'Not enough data points loaded (too few hdates). '
'Please check your local datasource and date range.'
)
if primary_share not in hp.shares:
raise ValueError(f'primary_share "{primary_share}" not found in shares')
# 1) 派生因子:MACD(默认 12_26_9)
hp_macd = hp.kline.macd(price_htype='close', fastperiod=12, slowperiod=26, signalperiod=9)
factor_col = 'macd_hist_12_26_9'
if factor_col not in hp_macd.htypes:
raise ValueError(f'Required factor htype "{factor_col}" not found after macd()')
# 2) 条件 -> mask(研究口径)
cond_long = (hp_macd[factor_col] > 0.0)
cond_short = (hp_macd[factor_col] <= 0.0)
mask_long = hp.where(cond_long)
mask_short = hp.where(cond_short)
# 3) 组合聚合 + benchmark
pf_long = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_long,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='LONG',
)
pf_short = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_short,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='SHORT',
)
# 4) cum_return -> CAGR 摘要
def _years_between(hdates) -> float:
idx = pd.DatetimeIndex(hdates)
days = (idx[-1] - idx[0]).days
return max(1e-9, days / 365.25)
def _cagr_from_cumret(cumret_end: float, years: float) -> float:
return (1.0 + cumret_end) ** (1.0 / years) - 1.0
years = _years_between(pf_long.hdates)
cr_long = pf_long.cum_return(htypes='close', method='simple')
cr_short = pf_short.cum_return(htypes='close', method='simple')
cumret_long_end = float(cr_long.values[cr_long.shares.index('LONG'), -1, 0])
cumret_short_end = float(cr_short.values[cr_short.shares.index('SHORT'), -1, 0])
cumret_bm_end = float(cr_long.values[cr_long.shares.index(benchmark), -1, 0])
summary = pd.DataFrame(
{
'cum_return_end': [cumret_long_end, cumret_short_end, cumret_bm_end],
'CAGR': [
_cagr_from_cumret(cumret_long_end, years),
_cagr_from_cumret(cumret_short_end, years),
_cagr_from_cumret(cumret_bm_end, years),
],
},
index=['LONG', 'SHORT', benchmark],
)
print('\\n[CAGR summary]')
print(summary)
# 5) 图:组合曲线对比(先 normalize,便于肉眼比较)
pf_view = pf_long.normalize(htypes='close', base_index=0)
fig_pf = pf_view.plot(interactive=True)
# 6) 图:单股触发点解释(1D 时间轴条件)
si = hp_macd.shares.index(primary_share)
cond_1d = np.asarray(cond_long[si, :, 0], dtype=bool).ravel()
lookback = min(200, len(cond_1d))
fig_one = hp.loc[-lookback:].plot(
shares=[primary_share],
interactive=True,
highlight={'condition': cond_1d[-lookback:], 'style': {'marker': 'x', 's': 50}},
)
return {
'hp': hp,
'hp_macd': hp_macd,
'pf_long': pf_long,
'pf_short': pf_short,
'summary': summary,
'fig_pf': fig_pf,
'fig_one': fig_one,
}
res = demo_vertical_timing(
stocks=['000001.SZ', '600519.SH', '300750.SZ'],
benchmark='000300.SH',
start='20220101',
end='20221231',
primary_share='000001.SZ',
)
res['fig_pf']
說明:上面
normalize只爲了畫圖對比更直觀;你做統計時仍以cum_return或原始價格序列爲準。
4.12. 9. 小结与边界
到這裏,我們已經把一個“少量標的擇時”的研究閉環跑通了: 數據 -> 因子 -> bool 條件 -> where mask -> portfolio + benchmark -> cum_return + CAGR -> plot(highlight) 解釋。
需要注意的是:這裏的 portfolio/cum_return 都是研究向計算,不包含交易成本、滑點、交割、資金約束等完整回測語義。 如果你希望把研究邏輯遷移到真正的策略回測,建議把因子/條件輸出成策略可用的數據列或信號,並交給 Operator/Backtester 處理交易層語義。
4.13. 附錄:插圖索引(建議你在 Notebook 裏生成/截圖)
插圖 |
建議放置位置 |
你將看到什麼 |
|---|---|---|
|
§0 開場最小可跑 |
證明 |
|
§0.5 或 §6 |
|
|
§7 |
單股圖上高亮觸發點(便於覆盤解釋) |