6. 用 HistoryPanel 研究事件型因子
事件型因子：K线形态信号 -> 事件窗口 mask -> CAGR 与可视化解释
在真实研究里，“事件型因子”经常比连续因子更贴近我们的直觉：
比如“锤头线出现”“吞没形态出现”“放量长阳出现”……这些事件一旦发生，我们往往会下意识地问一句：
如果我只在事件附近挑股票，长期看会不会更好？
本篇教程就围绕这个问题，用 HistoryPanel.candle_pattern() 把形态事件变成可研究的数据，再把它窗口化成横向筛选条件，最后跑通完整闭环：
事件信号 -> 事件窗口 -> where/mask -> portfolio + benchmark -> cum_return + CAGR -> plot(highlight) 解释。
统一约束（与前两篇保持一致）：
以个股为主，同时加入
000300.SH作为 benchmark事件窗口 (K=5)（最近 5 天内发生过就算“有效”）
6.1. 0. 开场：先跑通“形态信号 -> 在 K 线上高亮事件”的最小版本
先看最小可跑证明：我们只做两件事：
计算形态信号；
在单只股票的 K 线上把事件点高亮出来。
import qteasy as qt
share = '000001.SZ'
benchmark = '000300.SH'
hp = qt.get_kline(
shares=[share, benchmark],
start='20220101',
end='20221231',
freq='D',
as_panel=True,
)
signals = hp.candle_pattern(name='cdlhammer', as_panel=False)
print(signals.tail())
fig = hp.plot(shares=[share], interactive=True, highlight='max')
fig
不过，仅仅“能算出来/能画出来”还远远不够。
如果我们真的要把它当成可复用的研究方法来用，至少会遇到下面这些问题：
事件太稀疏：很多形态信号都像“针尖”一样跳出来——今天出现、明天又没了。如果你按“当天出现就选”来做篮子，组合会非常不稳定：入选股票数量忽多忽少，收益曲线也容易断断续续，最后你甚至分不清是在研究形态，还是在研究“样本稀疏性”。
怎么窗口化才像实战：更自然的口径通常不是“今天出现就立刻选”，而是“最近一段时间出现过就算有效”。因为现实里我们不一定能在当天捕捉到完美的形态点；更常见的做法是把它当成一个“短窗口内的关注信号”。
怎么变成横向选股：事件研究的价值，往往在于“同一天从很多股票里挑出发生过事件的那一批”。这要求我们把事件信号变成
(M,L)的条件矩阵，再统一规整成where的 mask。有收益没解释：就算你算出了 CAGR，你也需要可复盘：事件点到底对应了哪些 K 线？发生在上升趋势、下跌趋势还是震荡？把事件标回图上，是事件型研究里最关键的一步。
好在这些能力都可以一步步补齐。本文就从“形态信号的结构”讲起。
6.2. 0.5 先贴最终效果（我们最后会得到什么）
按本文做完，你会得到三类输出：
事件窗口筛选的组合曲线 vs benchmark：把“最近 5 天发生过形态”的股票当作一个动态篮子，算出它的组合曲线，并与
000300.SH对比。CAGR 摘要表：把组合末值收益折算成年化口径，便于横向比较不同研究期。
单股 K 线事件高亮图：把事件发生日标回 K 线上，让结果可复盘、可讨论。
6.3. 1. 目标（我们这篇文章要完成什么）
获取多 shares 的 OHLC 面板，并加入
000300.SH作为 benchmark用
candle_pattern得到形态信号矩阵（time x shares）把事件信号窗口化（K=5），得到
(M,L)的事件条件用
where()规整为研究 mask，并构建 long/short 两组组合曲线用
cum_return推导 CAGR 摘要表用
plot(highlight=...)在 K 线上高亮事件点，做到可解释闭环文末给出“单函数可跑”的完整代码
6.4. 2. 准备数据：必须有 OHLC（事件型因子离不开它）
2.1 本节要解决什么
形态识别依赖 open/high/low/close。
所以本节我们只做一件事：确保 OHLC 列齐全，并把 benchmark 一并放进来，后面直接对比。
2.2 最小必要原理
candle_pattern(name=...) 会检查你传入的 price_htypes 是否在 hp.htypes 中存在。
因此我们先做字段校验，避免写到一半才发现缺列。
2.3 可运行代码 + 预期效果
import qteasy as qt
benchmark = '000300.SH'
shares = ['000001.SZ', '600519.SH', '300750.SZ', benchmark]
hp = qt.get_kline(
shares=shares,
start='20220101',
end='20221231',
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}')
print('hp.shape:', hp.shape)
print('hp.htypes:', hp.htypes)
if hp.shape[1] < 50:
raise ValueError(
'Not enough data points loaded (too few hdates). '
'Please check your local datasource and date range.'
)
6.5. 3. 形态因子提取：candle_pattern 得到事件信号矩阵
3.1 本节要解决什么
我们要得到一张“事件信号表”：每一天、每只股票，是否出现该形态。
这里示例用 cdlhammer（锤头线），你后面可以替换成其他形态函数名。
3.2 最小必要原理
signals = hp.candle_pattern(name='cdlhammer', as_panel=False) 的返回是：
DataFrame（index=时间，columns=shares）值为浮点数（通常 0 表示无事件，正/负表示方向/强度，具体由 ta-lib 定义）
后面我们会把它转成 bool 事件矩阵，再做窗口化。
这里我们用一个很实用的“研究口径简化”：
不管它是 +100 还是 -100，只要 非 0，我们就认为“发生过事件”。如果你更关心方向，也可以分成 >0 和 <0 两套窗口（本文会同时演示 long/short 两套窗口，保持与前两篇一致）。
3.3 可运行代码 + 预期效果
signals = hp.candle_pattern(name='cdlhammer', as_panel=False)
print('signals shape:', signals.shape)
# 只看非 0 的事件（便于验证确实有触发）
nonzero = signals.where(signals != 0.0)
print(nonzero.dropna(how='all').tail())
6.6. 4. 事件窗口化（K=5）：不是“今天发生就选”，而是“最近 5 天发生过就选”
4.1 本节要解决什么
这是本文最关键的一节：
把稀疏的事件信号变成更贴近实战的“窗口内有效”条件。
我们选择 K=5：
最近 5 天内出现过事件，就认为该股票在当日属于候选集合。
4.2 最小必要原理
signals 的形状是 (L, M)（time x shares），我们把它转成 (M, L)（shares x time），再做窗口滚动的 any：
events_ml[i, t] = True表示 share i 在 t 日发生了事件window_ml[i, t] = any(events_ml[i, t-K+1 : t+1])
最终得到的 window_ml 仍然是 (M, L) 的 bool，就可以直接喂给 hp.where(window_ml)。
你可以把这个“窗口化”理解成一句非常直觉的研究口径：
我不要求你今天刚好踩在形态当天；只要你在最近 5 天里出现过一次形态，我就把你当作“事件仍然有效”的候选。
这会让篮子更稳定（不会只剩一两个针尖信号），也更接近我们做复盘时的真实行为：形态往往是一个“阶段信号”，不是一个“毫秒级触发器”。
4.3 可运行代码 + 预期效果（样张3对应，K=5）
import numpy as np
K = 5
sig_ml = signals.to_numpy().T # (M, L)
long_events_ml = sig_ml > 0.0
short_events_ml = sig_ml < 0.0
def any_in_last_k(events_ml: np.ndarray, k: int) -> np.ndarray:
m, l = events_ml.shape
out = np.zeros((m, l), dtype=bool)
for t in range(l):
left = max(0, t - k + 1)
out[:, t] = np.any(events_ml[:, left:t+1], axis=1)
return out
long_window_ml = any_in_last_k(long_events_ml, K)
short_window_ml = any_in_last_k(short_events_ml, K)
mask_long = hp.where(long_window_ml)
mask_short = hp.where(short_window_ml)
print('mask_long shape:', mask_long.shape) # 期望 (M,L,N)
print('selected_count_last_day(long):', int(long_window_ml[:, -1].sum()))
建议你再加一段 sanity check：看看“每天入选数量”的分布，避免筛选过严导致空篮子：
selected_count_by_day = long_window_ml.sum(axis=0)
print('selected_count stats (event_long):')
print(' min/max:', int(selected_count_by_day.min()), int(selected_count_by_day.max()))
print(' mean:', float(selected_count_by_day.mean()))
print(' p10/p50/p90:', np.quantile(selected_count_by_day.astype(float), [0.1, 0.5, 0.9]))
6.7. 5. portfolio + benchmark + cum_return + CAGR：把事件窗口筛选变成可比较的结果
5.1 本节要解决什么
事件窗口筛选得到了，但它到底“有没有用”？
这一节我们用组合曲线 + CAGR 表给出一个清晰的结论：
事件 long/short 两组在研究期内与 benchmark 的差异。
5.2 最小必要原理
这里仍然使用研究向的 portfolio 聚合，不做交易执行。
然后用 cum_return 取末值推导 CAGR（等效年化）。
5.3 可运行代码（示意）
import pandas as pd
benchmark = '000300.SH'
pf_long = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_long,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='EVENT_LONG',
)
pf_short = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_short,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='EVENT_SHORT',
)
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 = pf_long.cum_return(htypes='close', method='simple')
cumret_long_end = float(cr.values[cr.shares.index('EVENT_LONG'), -1, 0])
cumret_bm_end = float(cr.values[cr.shares.index('000300.SH'), -1, 0])
print('CAGR(event_long):', _cagr_from_cumret(cumret_long_end, years))
print('CAGR(benchmark):', _cagr_from_cumret(cumret_bm_end, years))
同样建议你整理成摘要表（至少 3 行：EVENT_LONG / EVENT_SHORT / benchmark），便于读者一眼比较：
cr2 = pf_short.cum_return(htypes='close', method='simple')
cumret_short_end = float(cr2.values[cr2.shares.index('EVENT_SHORT'), -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=['EVENT_LONG', 'EVENT_SHORT', '000300.SH'],
)
print('\\n[CAGR summary]')
print(summary)
6.8. 6. 可视化解释：在 K 线上高亮事件发生日（让结论可复盘）
6.1 本节要解决什么
事件型研究最怕“只有一张收益表”。
我们需要把事件点标回到 K 线图上，肉眼核对“这到底是什么形态、发生在什么趋势里”，这样结论才可复盘、可讨论。
6.2 最小必要原理
plot(highlight=...) 可以用 1D bool（时间轴）高亮事件点。
因此我们对某只 primary_share 从 signals 里抽出 1D 条件，再画 K 线并高亮。
6.3 可运行代码 + 预期效果
import numpy as np
primary_share = '000001.SZ'
event_1d = (signals[primary_share].to_numpy() != 0.0).astype(bool)
fig = hp.plot(
shares=[primary_share],
interactive=True,
highlight={'condition': event_1d, 'style': {'marker': 'x', 's': 60}},
)
fig
预期效果：你会在 K 线上看到一串被高亮的点（事件发生日）。
这一步很重要：事件型研究不怕“结果不好”，怕的是“结果无法解释”。把事件标回图上，你就能直观看到它出现在哪些位置：是下跌末端、是上涨中继、还是震荡噪声，从而决定下一步要不要叠加趋势过滤或波动过滤。
6.9. 7. 完整代码（单函数可跑版本）
下面给出一段“单函数可跑”的完整版本，方便你复制进 Notebook 一键运行。它覆盖：
candle_pattern提取事件信号；K=5 窗口化（最近 5 天 any）；
where -> portfolio -> cum_return -> CAGR的研究闭环；单股 K 线高亮事件点（可复盘解释）。
import numpy as np
import pandas as pd
import qteasy as qt
def demo_event_pattern(
shares: list,
benchmark: str = '000300.SH',
pattern_name: str = 'cdlhammer',
k: int = 5,
start: str = '20220101',
end: str = '20221231',
primary_share: str = '000001.SZ',
):
\"\"\"演示事件型因子研究闭环：形态信号 -> 窗口化 -> 组合曲线 -> CAGR -> K线高亮解释。
Parameters
----------
shares : list
股票池（必须包含 benchmark；建议 3~30 只即可演示横向筛选）。
benchmark : str, default '000300.SH'
基准指数代码。
pattern_name : str, default 'cdlhammer'
形态名称（ta-lib 风格名称）。
k : int, default 5
事件窗口长度：最近 k 天出现过就算有效。
start : str, default '20220101'
起始日期（YYYYMMDD）。
end : str, default '20221231'
结束日期（YYYYMMDD）。
primary_share : str, default '000001.SZ'
用于高亮解释的单只股票代码。
Returns
-------
dict
包含 hp/signals/pf_long/pf_short/summary/fig_pf/fig_one 等结果对象。
\"\"\"
if benchmark not in shares:
raise ValueError('benchmark must be included in shares')
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 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) 形态信号：DataFrame (L, M)
signals = hp.candle_pattern(name=pattern_name, as_panel=False)
print('\\n[signals]')
print(' shape:', signals.shape)
print(' nonzero tail:')
print(signals.where(signals != 0.0).dropna(how='all').tail())
# 2) 事件 -> bool -> 窗口化 any-in-last-k（得到 (M, L)）
sig_ml = signals.to_numpy().T # (M, L)
long_events_ml = sig_ml > 0.0
short_events_ml = sig_ml < 0.0
def any_in_last_k(events_ml: np.ndarray, kk: int) -> np.ndarray:
m, l = events_ml.shape
out = np.zeros((m, l), dtype=bool)
for t in range(l):
left = max(0, t - kk + 1)
out[:, t] = np.any(events_ml[:, left:t + 1], axis=1)
return out
long_window_ml = any_in_last_k(long_events_ml, k)
short_window_ml = any_in_last_k(short_events_ml, k)
selected_count_by_day = long_window_ml.sum(axis=0)
print('\\n[Selection count stats]')
print(' min/max:', int(selected_count_by_day.min()), int(selected_count_by_day.max()))
print(' mean:', float(selected_count_by_day.mean()))
print(' p10/p50/p90:', np.quantile(selected_count_by_day.astype(float), [0.1, 0.5, 0.9]))
mask_long = hp.where(long_window_ml)
mask_short = hp.where(short_window_ml)
# 3) 组合聚合 + benchmark
pf_long = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_long,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='EVENT_LONG',
)
pf_short = hp.portfolio(
htypes='close',
mode='equal',
mask=mask_short,
benchmark=benchmark,
benchmark_output='tag_along',
new_share_name='EVENT_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('EVENT_LONG'), -1, 0])
cumret_short_end = float(cr_short.values[cr_short.shares.index('EVENT_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=['EVENT_LONG', 'EVENT_SHORT', benchmark],
)
print('\\n[CAGR summary]')
print(summary)
# 5) 图：组合对比（归一化更直观）
fig_pf = pf_long.normalize(htypes='close', base_index=0).plot(interactive=True)
# 6) 图：单股事件高亮（使用 1D 时间轴 bool）
event_1d = (signals[primary_share].to_numpy() != 0.0).astype(bool)
lookback = min(200, len(event_1d))
fig_one = hp.loc[-lookback:].plot(
shares=[primary_share],
interactive=True,
highlight={'condition': event_1d[-lookback:], 'style': {'marker': 'x', 's': 60}},
)
return {
'hp': hp,
'signals': signals,
'pf_long': pf_long,
'pf_short': pf_short,
'summary': summary,
'fig_pf': fig_pf,
'fig_one': fig_one,
}
res = demo_event_pattern(
shares=['000001.SZ', '600519.SH', '300750.SZ', '000300.SH'],
benchmark='000300.SH',
pattern_name='cdlhammer',
k=5,
start='20220101',
end='20221231',
primary_share='000001.SZ',
)
res['fig_one']
6.10. 8. 小结与边界
到这里，我们已经把事件型因子研究跑通了：
形态信号 -> 窗口化 -> 横向筛选 -> 组合曲线 -> CAGR -> K线高亮解释。
需要再次强调：这条链路是研究向粗聚合，不含交易执行语义。
如果你希望把事件信号做成真实可回测的策略，应把“事件窗口的筛选规则”转成策略信号，并交给 Operator/Backtester 处理交易层细节（成本、交割、下单约束等）。
6.11. 附录：插图索引（建议你在 Notebook 里生成/截图）
插图 |
建议放置位置 |
你将看到什么 |
|---|---|---|
|
§0 |
最小可跑：算 signals + 画出基础图 |
|
§0.5 或 §5 |
事件窗口篮子 vs benchmark 的组合曲线 |
|
§6 |
单股 K 线高亮事件发生日（可复盘解释） |