策略文件里定义一个类，需要继承 StrategyHooks，实现 has_opportunity 方法。NTQ 在回测时逐日逐股调你的钩子，让你判断"今天这只股票有没有机会"。

StrategyHooks提供的API一览：

has_opportunity(ctx: StrategyContext) -> bool：

必须实现：是

调用时机：回测中的每个交易日对每只股票都调用一次

用途描述：枚举阶段的核心方法。NTQ 在回测时间窗口内，每个交易日对每只股票调用一次。返回 True 表示发现交易机会，返回 False 表示没有。

函数参数：ctx: StrategyContext - 只读上下文，提供当日为止的所有数据。包含扫描过程中的 remember 数据

函数返回：Boolean

代码示例：

def has_opportunity(self, ctx: StrategyContext) -> bool:
    klines = ctx.data("stock.kline.daily")
    if klines is None or len(klines) < 14:
        return False

    close = [bar["close"] for bar in klines]
    rsi = self._calc_rsi(close, 14)

    # 记录信号上下文，事后可在 signal_snapshots.csv 里查看
    ctx.capture("rsi", rsi)
    ctx.capture("close", close[-1])

    return rsi < 30

注意：ctx.data() 返回的是当天及之前的数据序列，不是只有当天一根 K 线。你可以用全部历史 K 线计算指标。ctx.capture() 记录的值会在报告的信号快照里保存——帮助你事后分析策略在什么条件下触发了。

on_before_scan(ctx) -> None：

必须实现：否

调用时机：扫描阶段开始前，只调用一次。（不是每只股票一次）。

用途描述：扫描阶段开始前调用一次。适合做初始化工作：预加载参考数据、设置全局阈值、初始化计数器。

函数参数：ctx: StrategyContext - 只读上下文，提供当日为止的所有数据。包含扫描过程中的 remember 数据

函数返回：无返回

代码示例：

def on_before_scan(self, ctx: StrategyContext) -> None:
    # 预加载一个行业分类表，后续 has_opportunity 里用
    industry_map = ctx.data("stock.industry_map")
    ctx.remember("industry_map", industry_map)

    # 从 settings.core 读取阈值，后续复用
    threshold = self.core_float(ctx.settings, "rsi_threshold", clamp=(0.0, 100.0))
    ctx.remember("threshold", threshold)

和 ctx.capture() 不同——remember() 存的值不进报告，只在本次回测的钩子之间传递。has_opportunity 里用 ctx.recall("threshold") 取出来。

on_after_scan(ctx) -> None：

必须实现：否

调用时机：扫描阶段结束后，只调用一次。（不是每只股票一次）。

用途描述：扫描阶段结束后调用一次。适合做清理工作或输出统计。

函数参数：ctx: StrategyContext - 只读上下文，提供当日为止的所有数据。包含扫描过程中的 remember 数据

函数返回：无返回

代码示例：

def on_after_scan(self, ctx: StrategyContext) -> None:
   # 输出本次扫描触发了多少次机会
   count = ctx.recall("hit_count", default=0)
   print(f"扫描完成，共发现 {count} 次机会")
   ctx.forget("hit_count")

on_calendar_asof(ctx) -> CalendarAsOfResult：

必须实现：否

调用时机：slice_based 模式的时间推进点。entity_based 模式不会调用当前钩子函数。

用途描述：slice_based 切片模式下，每当时间轴推进到一个新的日历点时调用。ctx里包括当前日期和回测里所有的股票数据 - 你可以在这里做跨股票的全局判断，比如交易量排前五的股票。

函数参数：ctx: StrategyContext - 只读上下文，此时 ctx.data 尚未填充具体股票数据。

函数返回：CalendarAsOfResult - 控制后续行为。默认返回 None，表示无特殊操作

代码示例：

def on_calendar_asof(self, ctx: StrategyContext):
   # 读取全局市场数据（非单只股票）
   market_data = ctx.data("market.index.daily")
   if market_data and len(market_data) > 0:
       today_change = market_data[-1]["pct_change"]
       # 大盘暴跌超过 4%，设一个标记，has_opportunity 里检查
       if today_change < -0.04:
           ctx.remember("market_crash", True)
       else:
           ctx.remember("market_crash", False)

配套方法 calendar_asof_needs_by_entity(ctx) 控制是否需要 ctx.data.by_entity 数据，默认 True。如果你只读全局数据、不需要逐股数据，可以覆盖为 False 以节省内存：

def calendar_asof_needs_by_entity(self, ctx: StrategyContext) -> bool:
   return False  # 我只用全局数据，不需要 by_entity

on_pick_portfolio_member(ctx) -> Sequence[Union[Opportunity, str]]

必须实现：否

调用时机：资金回测阶段，每日入场前。

用途描述：资金回测阶段，每日入场前调用。比如：RSI低于20的机会可能连续很多天出现，你是要马上投资，还是再等等？再或者是如果有5个机会摆在你面前，你只能选3个，选哪三个？这个函数就是来描述怎么选的。

函数参数：ctx: StrategyContext - 只读上下文，ctx.data.opportunity 包含当日所有机会列表。

函数返回：Sequence[Union[Opportunity, str]] — 选中的机会列表。返回空列表表示今日不买入。可以返回 Opportunity 对象或股票 ID 字符串。

代码示例：

def on_pick_portfolio_member(self, ctx: StrategyContext):
   opportunities = ctx.data.opportunity or []
   # 今天最多买 3 只
   selected = opportunities[:3]
   # 已经持仓的股票不再加仓
   held_stocks = ctx.recall("held_stocks", default=set())
   selected = [o for o in selected if o.stock.code not in held_stocks]
   return selected

如果不实现这个方法，系统默认把当日所有机会都交给资金分配引擎处理，由 portfolio.allocation 的配置决定怎么分配资金。

is_stop_loss(ctx, *, custom: str, stage: Any) -> bool

必须实现：否

调用时机：止损检查阶段，对持仓中的每笔投资调用。

用途描述：当 settings.goal.stop_loss 的某个 stage 用了 custom 字段而不是 ratio 字段时，NTQ 会调用这个方法判断是否触发止损。让你用自定义逻辑替代固定比例——比如"跌破 20 日均线止损"。

函数参数：

• ctx: StrategyContext — 包含当日数据。
• custom: str — 你在 settings 里设的自定义名称。
• stage: Any — 该 stage 的完整配置对象。

函数返回：bool — True 表示触发止损，False 表示不触发。

代码示例：

settings 里的配置：

"goal": {
    "stop_loss": {
        "stages": [
            # 用 custom 而不是 ratio
            {"custom": "below_ma20", "close_invest": True},
        ],
    },
}

钩子实现：

def is_stop_loss(self, ctx: StrategyContext, *, custom: str, stage) -> bool:
    if custom == "below_ma20":
        klines = ctx.data("stock.kline.daily")
        if klines is None or len(klines) < 20:
            return False
        ma20 = sum(bar["close"] for bar in klines[-20:]) / 20
        today_close = klines[-1]["close"]
        return today_close < ma20
    return False

一个 stage 要么用 ratio（固定比例触发），要么用 custom（自定义逻辑触发），两者互斥。

is_take_profit(ctx, *, custom: str, stage: Any) -> bool

必须实现：否

调用时机：止盈检查阶段，对持仓中的每笔投资调用。

用途描述：和 is_stop_loss 完全对称，只是用于止盈。当 settings.goal.take_profit 的某个 stage 用了 custom 字段时调用。

函数参数：

• ctx: StrategyContext — 包含当日数据。
• custom: str — 你在 settings 里设的自定义名称。
• stage: Any — 该 stage 的完整配置对象。

函数返回：bool — True 表示触发止盈，False 表示不触发。

代码示例：

settings 里的配置：

"goal": {
    "take_profit": {
        "stages": [
            {"custom": "rsi_overbought", "close_invest": True},
        ],
    },
}

钩子实现：

def is_take_profit(self, ctx: StrategyContext, *, custom: str, stage) -> bool:
    if custom == "rsi_overbought":
        klines = ctx.data("stock.kline.daily")
        if klines is None or len(klines) < 14:
            return False
        close = [bar["close"] for bar in klines]
        rsi = self._calc_rsi(close, 14)
        return rsi > 70
    return False

典型用途：RSI 策略买入条件是 RSI < 30（超卖），止盈条件是 RSI > 70（超买）。用 custom 可以让进出条件都基于技术指标，而不是固定的涨跌幅比例。

StrategyContext 方法

钩子的参数 ctx 是一个只读上下文，提供数据访问和信号记录能力。

读数据

# 读取 base 数据（K线）
klines = ctx.data("stock.kline.daily")

# 读取 required 数据（在 settings.data.required 里声明的）
finance = ctx.data("stock.finance.quarterly")

# 读取全局数据
calendar = ctx.data("trade.calendar")

ctx.data() 返回的是当天及之前的数据序列。你拿到的是前复权数据，不需要关心复权逻辑。

记录信号

ctx.capture("rsi", rsi)
ctx.capture("volume", klines[-1]["volume"])

capture 记录的值会保存到 signal_snapshots.csv。在报告里可以看到每次触发时这些指标的具体值。用途是事后分析"策略在什么条件下触发了"。

内存袋

# 在 on_before_scan 里存
ctx.remember("my_threshold", 0.3)

# 在 has_opportunity 里取
threshold = ctx.recall("my_threshold", default=0.3)

# 用完可以删
ctx.forget("my_threshold")

内存袋在单次回测内有效，用于在不同钩子之间传递状态。

读取 settings

# 从 settings.core 读私有参数
period = self.core_int(ctx.settings, "rsi_period")
threshold = self.core_float(ctx.settings, "rsi_threshold", clamp=(0.0, 100.0))

core_int 和 core_float 是 StrategyHooks 提供的静态方法，从 settings.core 里安全取值。

静态工具方法