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 里运行，上面这一段通常就能立即看到一张图。

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

1. 择时规则怎么落地：我们脑子里很容易有一句“MACD>0 就持有”，但写成代码后常常会变形：列名记错、数据缺列、算出来全是 NaN、或不同股票对齐方式不一致。结果就是研究脚本一跑就报错，或者更糟——不报错但结论不可信。

2. mask 形状怎么对齐：条件本质是二维的（股票×时间），但 HistoryPanel 是三维的（股票×时间×字段）。如果形状对不齐，你会出现“以为在筛选股票，其实在筛选字段”的隐性 bug；而这种 bug 往往不会立刻报错，只会让收益曲线看起来“有点怪”。

3. 没有 benchmark 很难判断好坏：只看组合曲线很容易“自嗨”。它可能只是踩中了 2022 年某一段风格或大盘趋势。把 000300.SH 拉进来做对照，至少能回答一个更接近实战的问题：这套择时规则，是在创造超额，还是只是跟着市场波动？

4. 有结果但难解释：就算 long 组合跑赢了，也很难解释“为什么赢”。真正的研究需要可复盘：当收益出现明显拐点时，我们能不能迅速回到图上定位“是哪些触发点改变了持仓状态”？

好在这些能力都可以一步步补齐。本文就按“先跑通，再增强”的节奏，把它补齐成一个真正可用的小研究流程。

---

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

为了让节奏更稳，我们先把“终点形态”讲清楚。按本文做完，你至少会得到两类输出：

1. 组合层面的对比：LONG / SHORT / 000300.SH 三条曲线（统一从 1.0 起点归一化）放在一张图里，你能一眼看出“择时规则在研究期内是否有效”。

2. 单标的层面的解释：对某只股票画 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.SH

• hp.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_9

• macd_signal_12_26_9

• macd_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.shape

• mask_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 到底强多少？”
因此我们做两件事：

1. normalize：统一起点（便于目视比较）

2. 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 里生成/截图）

插图	建议放置位置	你将看到什么
img/3.1_minimal_run.png	§0 开场最小可跑	证明 get_kline -> plot 链路通了
img/3.1_pf_compare.png	§0.5 或 §6	LONG/SHORT/000300.SH 归一化曲线对比
img/3.1_highlight_one_share.png	§7	单股图上高亮触发点（便于复盘解释）