6. HistoryPanel によるイベント駆動因子の研究

イベントドリブン要因: ローソク足パターンシグナル -> イベントウィンドウマスク -> CAGR と視覚的な説明

実際の研究では、「イベント型の要因」は、連続的な要因よりも直感に近いことがよくあります。たとえば、「ハンマーが現れる」、「巻き込むパターンが現れる」、「高出来高の長い強気のローソク足が現れる」…これらのイベントが発生すると、私たちはしばしば本能的に次のように尋ねます。

イベント周辺でのみ銘柄を選んだ方が、長期的には良いでしょうか?

このチュートリアルでは、この問題を中心に説明します。HistoryPanel.candle_pattern() を使用してパターン イベントを調査可能なデータに変換し、それを断面スクリーニング条件にウィンドウ処理して、最後に完全な閉ループを実行します: イベント信号 -> イベント ウィンドウ -> where/mask -> portfolio + benchmark -> cum_return + CAGR -> plot(highlight) 解釈。

統合された制約 (前の 2 つの投稿と一致):

• 個別銘柄に焦点を当て、ベンチマークとして000300.SHも含めます

• イベントウィンドウ (K=5) (最近 5 日以内に発生した場合は「有効」としてカウントされます)

---

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

まず、最小限の実行可能な証明です。行うことは 2 つだけです。

1. パターン信号を計算します。

2. 単一銘柄のローソク足チャート上のイベント ポイントを強調表示します。

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 マスクに標準化する必要があります。

4. 説明なしで返されます: CAGR を計算したとしても、イベント ポイントがどのローソク足に対応していたかを確認できるものが必要です。それらは上昇トレンド、下降トレンド、またはレンジで発生しましたか?イベントをチャート上にマークし直すことは、イベントベースの調査において最も重要なステップです。

幸いなことに、これらの機能はすべて段階的に満たすことができます。この記事は「ローソク足パターンシグナルの構造」から始まります。

---

6.2. 0.5 まず、最終結果 (最終的に何が得られるか) を表示します。

この記事を最後まで実行すると、次の 3 種類の出力が得られます。

1. イベントウィンドウスクリーニングとベンチマークによるポートフォリオ曲線: 「過去 5 日以内にパターンが発生した」銘柄を動的バスケットとして扱い、そのポートフォリオ曲線を計算し、000300.SH と比較します。

2. CAGR 概要表: ポートフォリオの最終収益率を年換算した数値に変換し、異なる調査期間を並べて比較しやすくします。

3. イベントのハイライトを含む単一銘柄のローソク足チャート: 結果を確認して議論できるように、イベントの日付をローソク足チャートにマークします。

---

6.3. 1. 目标（我们这篇文章要完成什么）

• 複数の共有の OHLC パネルを取得し、000300.SH をベンチマークとして追加します

• candle_pattern を使用してパターン信号行列 (時間 x シェア) を取得します。

• イベント信号をウィンドウ処理 (K=5) して、(M,L) のイベント条件を取得します

• where() を使用してリサーチマスクに正規化し、ロング/ショートポートフォリオ曲線を構築します

• cum_return を使用して CAGR 概要テーブルを導出する

• plot(highlight=...) を使用してローソク足チャート上のイベント ポイントを強調表示し、解釈可能性ループを閉じます

• 最後に、完全な「単一関数実行可能」コードを提供します。

---

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

2.1 このセクションの解決の目的

パターン認識はopen/high/low/closeに依存します。 したがって、このセクションで行うことは 1 つだけです。OHLC カラムが完全であることを確認し、後で直接比較できるようにベンチマークも含めます。

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 (インデックス=時間、列=シェア)

• 値は浮動小数点です (通常、0 はイベントがないことを意味します。TA-Lib で定義されているように、正/負は方向/強度を示します)。

後でそれをブール値イベント行列に変換し、ウィンドウ処理を適用します。

ここでは、非常に実用的な「研究規約の簡略化」を使用します。+100 であっても -100 であっても、ゼロ以外である限り、それを「イベントが発生した」ものとして扱います。方向を重視する場合は、>0 と <0 の 2 つのウィンドウに分割することもできます (この記事では、前の 2 つの記事との一貫性を保つために、長いウィンドウと短いウィンドウの両方を示します)。

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) (時間 x シェア) です。これを (M, L) (シェア x 時間) に変換し、ローリング ウィンドウ any を実行します。

• events_ml[i, t] = True は、t 日にイベントがあったことを共有することを意味します

• window_ml[i, t] = any(events_ml[i, t-K+1 : t+1])

結果の window_ml は依然として (M, L) ブール配列であるため、それを hp.where(window_ml) に直接入力できます。

この「ウィンドウ処理」は、非常に直観的な研究上の慣例として理解できます。

今日はまさにそのパターンの日を迎えるように求めているわけではありません。過去 5 日間にそのパターンが 1 回出現した限り、「イベントがまだ有効である」候補として扱います。

これにより、バスケットがより安定します (干し草の中の針が 1 つまたは 2 つだけの信号のままになることはありません)。また、レビューで実際に行っていることに近づきます。パターンは、多くの場合、「ミリ秒レベルのトリガー」ではなく、「位相信号」です。

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

別の健全性チェックを追加することをお勧めします。厳密すぎるフィルタリングによってバスケットが空になることを避けるために、「毎日選択された数」の分布を確認します。

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 テーブルを使用して、調査期間にわたるイベントのロング/ショート グループとベンチマークの差を明確に結論付けます。

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 / ベンチマーク) に整理することをお勧めします。

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 このセクションの解決の目的

イベント研究における最大の懸念は、「リターン テーブルが 1 つしかない」ことです。 イベント ポイントをローソク足チャートにプロットし、「これが実際にどのようなパターンで、どのようなトレンド内で発生するか」を視覚的に検証して、結論を再現して議論できるようにする必要があります。

6.2 必要最小限の原則

plot(highlight=...) は 1D ブール (時間軸) を使用してイベント ポイントを強調表示できます。 したがって、特定の primary_share に対して、signals から 1D 条件を抽出し、ローソク足チャートをプロットして強調表示します。

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

期待される結果: ローソク足チャートに一連のハイライトされたポイント (イベント日) が表示されます。 このステップは非常に重要です。イベント研究は「悪い結果」を恐れません。彼らは「説明できない結果」を恐れているのです。チャート上でイベントをマークし直すと、下落の終わり、上昇トレンド内の継続、または単なる横ばいのノイズなど、イベントが発生した場所を直接確認できるため、次のステップでトレンド フィルターを追加するかボラティリティ フィルターを追加するかを決定できます。

---

6.9. 7.完全なコード (単一関数実行可能バージョン)

以下は、ノートブックにコピーしてワンクリックで実行できる完全な「単一関数実行可能」バージョンです。内容は次のとおりです。

• candle_pattern はイベント信号を抽出します。

• K=5 ウィンドウ処理 (過去 5 日間のいずれか)。

• where -> portfolio -> cum_return -> CAGRの研究閉ループ。

• 単一銘柄のローソク足チャート上のイベント ポイントを強調表示します (レビュー/リプレイで説明可能)。

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 -> ローソク足の強調表示と説明という、イベント要因の調査パイプライン全体をすでに実行しています。

もう一度強調しておきますが、このパイプラインはリサーチ指向の粗い集計であり、取引実行のセマンティクスは含まれていません。 イベントシグナルを実際のバックテスト可能な戦略に変えたい場合は、「イベントウィンドウフィルタリングルール」を戦略シグナルに変換し、それをOperator/Backtesterに渡してトレーディングレイヤーの詳細（コスト、決済、注文制約など）を処理する必要があります。

---

6.11. 付録: 図のインデックス (ノートブックで生成/スクリーンショットを作成することをお勧めします)

図	推奨される配置場所	何が見えるか
img/3.3_minimal_run.png	§0	最小限の実行可能機能: 信号を計算し、基本的なチャートをプロットする
img/3.3_pf_compare.png	§0.5 または §5	イベントウィンドウバスケットとベンチマークの複合曲線
img/3.3_highlight_pattern.png	§6	単一銘柄のローソク足チャートでイベント日を強調表示します (再現可能で説明可能)