# 用 HistoryPanel 研究事件型因子

事件型因子：K线形态信号 -> 事件窗口 mask -> CAGR 与可视化解释

在真实研究里，“事件型因子”经常比连续因子更贴近我们的直觉：  
比如“锤头线出现”“吞没形态出现”“放量长阳出现”……这些事件一旦发生，我们往往会下意识地问一句：

> 如果我只在事件附近挑股票，长期看会不会更好？

本篇教程就围绕这个问题，用 `HistoryPanel.candle_pattern()` 把形态事件变成可研究的数据，再把它窗口化成横向筛选条件，最后跑通完整闭环：  
**事件信号 -> 事件窗口 -> `where/mask` -> `portfolio + benchmark` -> `cum_return + CAGR` -> `plot(highlight)` 解释。**

统一约束（与前两篇保持一致）：

- 以个股为主，同时加入 `000300.SH` 作为 benchmark  
- 事件窗口 \(K=5\)（最近 5 天内发生过就算“有效”）

---

## 0. 开场：先跑通“形态信号 -> 在 K 线上高亮事件”的最小版本

先看最小可跑证明：我们只做两件事：

1) 计算形态信号；  
2) 在单只股票的 K 线上把事件点高亮出来。

```python
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
```

不过，仅仅“能算出来/能画出来”还远远不够。  
如果我们真的要把它当成可复用的研究方法来用，至少会遇到下面这些问题：

1. **事件太稀疏**：很多形态信号都像“针尖”一样跳出来——今天出现、明天又没了。如果你按“当天出现就选”来做篮子，组合会非常不稳定：入选股票数量忽多忽少，收益曲线也容易断断续续，最后你甚至分不清是在研究形态，还是在研究“样本稀疏性”。  
2. **怎么窗口化才像实战**：更自然的口径通常不是“今天出现就立刻选”，而是“最近一段时间出现过就算有效”。因为现实里我们不一定能在当天捕捉到完美的形态点；更常见的做法是把它当成一个“短窗口内的关注信号”。  
3. **怎么变成横向选股**：事件研究的价值，往往在于“同一天从很多股票里挑出发生过事件的那一批”。这要求我们把事件信号变成 `(M,L)` 的条件矩阵，再统一规整成 `where` 的 mask。  
4. **有收益没解释**：就算你算出了 CAGR，你也需要可复盘：事件点到底对应了哪些 K 线？发生在上升趋势、下跌趋势还是震荡？把事件标回图上，是事件型研究里最关键的一步。  

好在这些能力都可以一步步补齐。本文就从“形态信号的结构”讲起。

---

## 0.5 先贴最终效果（我们最后会得到什么）

按本文做完，你会得到三类输出：

1) **事件窗口筛选的组合曲线 vs benchmark**：把“最近 5 天发生过形态”的股票当作一个动态篮子，算出它的组合曲线，并与 `000300.SH` 对比。  
2) **CAGR 摘要表**：把组合末值收益折算成年化口径，便于横向比较不同研究期。  
3) **单股 K 线事件高亮图**：把事件发生日标回 K 线上，让结果可复盘、可讨论。

---

## 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 线上高亮事件点，做到可解释闭环
- [ ] 文末给出“单函数可跑”的完整代码

---

## 2. 准备数据：必须有 OHLC（事件型因子离不开它）

### 2.1 本节要解决什么

形态识别依赖 `open/high/low/close`。  
所以本节我们只做一件事：确保 OHLC 列齐全，并把 benchmark 一并放进来，后面直接对比。

### 2.2 最小必要原理

`candle_pattern(name=...)` 会检查你传入的 `price_htypes` 是否在 `hp.htypes` 中存在。  
因此我们先做字段校验，避免写到一半才发现缺列。

### 2.3 可运行代码 + 预期效果

```python
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.'
    )
```

---

## 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 可运行代码 + 预期效果

```python
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())
```

---

## 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）

```python
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：看看“每天入选数量”的分布，避免筛选过严导致空篮子：

```python
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]))
```

---

## 5. `portfolio + benchmark + cum_return + CAGR`：把事件窗口筛选变成可比较的结果

### 5.1 本节要解决什么

事件窗口筛选得到了，但它到底“有没有用”？  
这一节我们用组合曲线 + CAGR 表给出一个清晰的结论：  
事件 long/short 两组在研究期内与 benchmark 的差异。

### 5.2 最小必要原理

这里仍然使用研究向的 `portfolio` 聚合，不做交易执行。  
然后用 `cum_return` 取末值推导 CAGR（等效年化）。

### 5.3 可运行代码（示意）

```python
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），便于读者一眼比较：

```python
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. 可视化解释：在 K 线上高亮事件发生日（让结论可复盘）

### 6.1 本节要解决什么

事件型研究最怕“只有一张收益表”。  
我们需要把事件点标回到 K 线图上，肉眼核对“这到底是什么形态、发生在什么趋势里”，这样结论才可复盘、可讨论。

### 6.2 最小必要原理

`plot(highlight=...)` 可以用 1D bool（时间轴）高亮事件点。  
因此我们对某只 `primary_share` 从 `signals` 里抽出 1D 条件，再画 K 线并高亮。

### 6.3 可运行代码 + 预期效果

```python
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 线上看到一串被高亮的点（事件发生日）。  
这一步很重要：事件型研究不怕“结果不好”，怕的是“结果无法解释”。把事件标回图上，你就能直观看到它出现在哪些位置：是下跌末端、是上涨中继、还是震荡噪声，从而决定下一步要不要叠加趋势过滤或波动过滤。

---

## 7. 完整代码（单函数可跑版本）

下面给出一段“单函数可跑”的完整版本，方便你复制进 Notebook 一键运行。它覆盖：

- `candle_pattern` 提取事件信号；  
- K=5 窗口化（最近 5 天 any）；  
- `where -> portfolio -> cum_return -> CAGR` 的研究闭环；  
- 单股 K 线高亮事件点（可复盘解释）。

```python
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']
```

---

## 8. 小结与边界

到这里，我们已经把事件型因子研究跑通了：  
形态信号 -> 窗口化 -> 横向筛选 -> 组合曲线 -> CAGR -> K线高亮解释。

需要再次强调：这条链路是研究向粗聚合，不含交易执行语义。  
如果你希望把事件信号做成真实可回测的策略，应把“事件窗口的筛选规则”转成策略信号，并交给 `Operator/Backtester` 处理交易层细节（成本、交割、下单约束等）。

---

## 附录：插图索引（建议你在 Notebook 里生成/截图）

| 插图 | 建议放置位置 | 你将看到什么 |
|------|--------------|--------------|
| `img/3.3_minimal_run.png` | §0 | 最小可跑：算 signals + 画出基础图 |
| `img/3.3_pf_compare.png` | §0.5 或 §5 | 事件窗口篮子 vs benchmark 的组合曲线 |
| `img/3.3_highlight_pattern.png` | §6 | 单股 K 线高亮事件发生日（可复盘解释） |