6. Estudiar factores impulsados ​​por eventos con HistoryPanel

Factor impulsado por eventos: señal de patrón de velas -> máscara de ventana de eventos -> CAGR y explicación visual

En la investigación real, los “factores de tipo evento” suelen estar más cerca de nuestra intuición que los factores continuos: por ejemplo, “aparece un martillo”, “aparece un patrón envolvente”, “aparece una larga vela alcista con un gran volumen”… Una vez que ocurren estos eventos, a menudo preguntamos instintivamente:

Si sólo elijo acciones en función de eventos, ¿será mejor a largo plazo?

Este tutorial se centra en este problema: use HistoryPanel.candle_pattern() para convertir eventos de patrón en datos investigables, luego transfórelos en condiciones de detección transversal y, finalmente, ejecute todo el ciclo cerrado: señal de evento -> ventana de evento -> where/mask -> portfolio + benchmark -> cum_return + CAGR -> plot(highlight) interpretación.

Restricciones unificadas (consistentes con las dos publicaciones anteriores):

• Céntrese en acciones individuales y también incluya 000300.SH como punto de referencia

• Ventana del evento (K=5) (cuenta como “válida” si ocurrió dentro de los últimos 5 días)

---

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

Primero, la prueba mínima ejecutable: solo hacemos dos cosas:

1. Calcular la señal del patrón;

2. Resalte los puntos del evento en el gráfico de velas de una sola acción.

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

Sin embargo, el simple hecho de poder “calcularlo/trazarlo” está lejos de ser suficiente. Si realmente queremos utilizarlo como método de investigación reutilizable, nos encontraremos con al menos los siguientes problemas:

1. Los eventos son demasiado escasos: muchas señales de patrones aparecen como “puntos”: aparecen hoy y desaparecen mañana. Si se construye la canasta “seleccionándola cuando aparezca hoy”, la cartera será muy inestable: el número de acciones seleccionadas oscilará violentamente y la curva de rendimiento fácilmente se volverá entrecortada y discontinua. Al final, es posible que ni siquiera estés seguro de si estás estudiando el patrón o la “escasez de muestras”.

2. Cómo abrirlo en ventana como en el trading real: una convención más natural no suele ser “elegirlo inmediatamente si aparece hoy”, sino “contarlo como válido si apareció en el período de tiempo más reciente”. Porque en realidad es posible que no podamos captar el punto del patrón perfecto el mismo día; más comúnmente, lo tratamos como una «señal de atención dentro de un período corto».

3. Cómo convertirlo en una selección de acciones transversal: El valor de los estudios de eventos a menudo reside en «seleccionar, en el mismo día, el subconjunto que tuvo el evento entre muchas acciones». Esto requiere convertir la señal del evento en una matriz de condición (M,L) y luego estandarizarla en una máscara where.

4. Devoluciones sin explicación: incluso si ha calculado la CAGR, aún necesita algo que pueda revisar: ¿a qué velas correspondieron los puntos del evento? ¿Ocurrieron en una tendencia alcista, bajista o en un rango? Volver a marcar los eventos en el gráfico es el paso más crítico en la investigación basada en eventos.

Afortunadamente, todas estas capacidades se pueden completar paso a paso. Este artículo comienza con la «estructura de las señales de los patrones de velas».

---

6.2. 0.5 Primero, muestra el resultado final (con qué terminaremos)

Si sigue este artículo hasta el final, obtendrá tres tipos de resultados:

1. Curva de cartera a partir de la evaluación de la ventana de eventos frente al índice de referencia: trate las acciones que «han tenido el patrón en los últimos 5 días» como una canasta dinámica, calcule su curva de cartera y compárela con 000300.SH.

2. Tabla de resumen CAGR: convierta el rendimiento terminal de la cartera en una cifra anualizada, lo que facilita la comparación de diferentes períodos de investigación uno al lado del otro.

3. Gráfico de velas de una sola acción con eventos destacados: marque la fecha del evento en el gráfico de velas para que los resultados sean revisables y discutibles.

---

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

• Obtenga un panel OHLC para múltiples recursos compartidos y agregue 000300.SH como punto de referencia

• Utilice candle_pattern para obtener una matriz de patrón-señal (tiempo x porcentajes)

• Ventana las señales de evento (K=5) para obtener condiciones de evento de (M,L)

• Utilice where() para normalizarlo en una máscara de investigación y crear curvas de cartera largas/cortas.

• Utilice cum_return para derivar una tabla de resumen CAGR

• Utilice plot(highlight=...) para resaltar puntos de eventos en el gráfico de velas, cerrando el ciclo de interpretabilidad

• Al final, proporcione el código completo «ejecutable de función única»

---

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

2.1 Qué pretende resolver esta sección

El reconocimiento de patrones depende de open/high/low/close. Entonces, en esta sección solo hacemos una cosa: asegurarnos de que las columnas OHLC estén completas e incluir también el punto de referencia para que podamos comparar directamente más adelante.

2.2 Principios mínimos necesarios

candle_pattern(name=...) comprueba si el price_htypes que pasas existe en hp.htypes. Por lo tanto, primero validamos los campos para evitar descubrir columnas faltantes a mitad del proceso.

2.3 Código ejecutable + resultados esperados

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 Qué pretende resolver esta sección

Queremos una “tabla de señales de eventos”: para cada día y cada acción, si aparece el patrón. Este ejemplo utiliza cdlhammer (Martillo); Más tarde podrá reemplazarlo con otros nombres de funciones de patrón.

3.2 Principios mínimos necesarios

El valor de retorno de signals = hp.candle_pattern(name='cdlhammer', as_panel=False) es:

• DataFrame (índice=hora, columnas=compartidos)

• Los valores son flotantes (normalmente 0 significa que no hay evento; positivo/negativo indica dirección/fuerza, según lo define TA-Lib)

Posteriormente lo convertiremos en una matriz de eventos booleanos y luego aplicaremos ventanas.

Aquí utilizamos una «simplificación de la convención de investigación» muy práctica: independientemente de si es +100 o -100, siempre que sea distinto de cero, lo tratamos como «un evento ocurrido». Si le importa más la dirección, también puede dividirla en dos ventanas, >0 y <0 (este artículo demostrará las ventanas larga y corta para mantener la coherencia con los dos artículos anteriores).

3.3 Código ejecutable + resultado esperado

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 Qué pretende resolver esta sección

Esta es la sección más crítica del artículo: convertir señales de eventos dispersas en una condición más práctica, «válida dentro de la ventana».

Elegimos K=5: si un evento ocurrió dentro de los 5 días más recientes, tratamos la acción como perteneciente al conjunto de candidatos de ese día.

4.2 Principios mínimos necesarios

La forma de signals es (L, M) (tiempo x acciones). Lo convertimos a (M, L) (comparte x tiempo), luego hacemos una ventana móvil any:

• events_ml[i, t] = True significa compartir. Tuve el evento el día t.

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

El window_ml resultante sigue siendo una matriz booleana (M, L), por lo que puede introducirlo directamente en hp.where(window_ml).

Puede entender esta «ventana» como una convención de investigación muy intuitiva:

No te estoy pidiendo que sigas el día exacto del patrón hoy; Siempre que el patrón haya aparecido una vez en los últimos 5 días, lo trataré como un candidato cuyo «evento sigue siendo válido».

Esto hará que la canasta sea más estable (no quedará con solo una o dos señales de una aguja en un pajar), y estará más cerca de lo que realmente hacemos en las revisiones: los patrones son a menudo una «señal de fase», no un «disparador de nivel de milisegundos».

4.3 Código ejecutable + resultado esperado (correspondiente a la hoja de muestra 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()))

Le sugiero que agregue otra verificación de cordura: observe la distribución del “recuento diario seleccionado” para evitar terminar con una canasta vacía debido a un filtrado demasiado estricto:

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 Qué pretende resolver esta sección

Tenemos el filtrado de ventana de eventos, pero ¿realmente “funciona”? En esta sección, damos una conclusión clara utilizando la curva de cartera + una tabla CAGR: la diferencia entre los grupos de eventos largos/cortos y el índice de referencia durante el período de investigación.

5.2 Los principios mínimos necesarios

Aquí todavía utilizamos agregación portfolio orientada a la investigación, sin ejecución comercial. Luego use el valor final de cum_return para derivar CAGR (rendimiento anualizado equivalente).

5.3 Código ejecutable (ilustración)

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

Asimismo, te sugiero que lo organices en una tabla resumen (al menos 3 filas: EVENT_LONG / EVENT_SHORT / benchmark) para que los lectores puedan comparar de un vistazo:

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 Qué pretende resolver esta sección

El mayor temor en los estudios de eventos es «tener una sola tabla de retornos». Necesitamos trazar los puntos del evento en el gráfico de velas y verificar visualmente “qué patrón es realmente y dentro de qué tendencia ocurre”, para que la conclusión pueda reproducirse y discutirse.

6.2 El principio de necesidad mínima

plot(highlight=...) puede usar un bool 1D (eje de tiempo) para resaltar puntos de eventos. Por lo tanto, para un primary_share determinado, extraemos una condición 1D de signals, luego trazamos el gráfico de velas y lo resaltamos.

6.3 Código ejecutable + resultados esperados

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

Resultado esperado: verá una serie de puntos resaltados (días de eventos) en el gráfico de velas. Este paso es crucial: los estudios de eventos no temen los “malos resultados”; tienen miedo de «resultados que no se pueden explicar». Al volver a marcar los eventos en el gráfico, puede ver directamente dónde ocurren: al final de una caída, como una continuación dentro de una tendencia alcista o simplemente como ruido lateral, para que pueda decidir si el siguiente paso es agregar filtros de tendencia o filtros de volatilidad.

---

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

A continuación se muestra una versión completa “ejecutable con una sola función” que puede copiar en una computadora portátil y ejecutar con un solo clic. Cubre:

• candle_pattern extrae las señales de eventos;

• K=5 ventanas (cualquiera en los últimos 5 días);

• La investigación cerró el circuito de where -> portfolio -> cum_return -> CAGR;

• Resalte los puntos de eventos en el gráfico de velas de una sola acción (explicable mediante revisión/repetición).

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. 小结与边界

En este punto, ya hemos recorrido todo el proceso de investigación de factores de eventos: señal de patrón -> ventanas -> filtrado de sección transversal -> curva compuesta -> CAGR -> resaltado y explicación de velas japonesas.

Es necesario enfatizar nuevamente: este canal es una agregación burda orientada a la investigación y no contiene semántica de ejecución comercial. Si desea convertir la señal del evento en una estrategia real comprobable, debe convertir las «reglas de filtrado de la ventana de eventos» en señales de estrategia y entregárselas al Operator/Backtester para manejar los detalles de la capa comercial (costos, liquidación, restricciones de órdenes, etc.).

---

6.11. Apéndice: índice de figuras (recomendado generar/captura de pantalla en el Notebook)

Figura	Ubicación sugerida	lo que veras
img/3.3_minimal_run.png	§0	Ejecutable mínimo: calcular señales + trazar el gráfico básico
img/3.3_pf_compare.png	§0.5 o §5	Curva compuesta de la canasta de ventana de eventos versus el índice de referencia
img/3.3_highlight_pattern.png	§6	Resalte el día del evento en un gráfico de velas de una sola acción (reproducible y explicable)