tag.py钩子函数和上下文获取方法

tag.py 里定义一个类，继承 TagHooks，实现 calculate_tag 方法。NTQ 在计算标签时按日期推进，每个交易日对每只股票（或全局）调用你的钩子，让你计算并返回标签值。

简单的tag.py的例子

from core.modules.tag.contracts import TagHooks, TagContext

class VolumeSpikeTag(TagHooks):
    def calculate_tag(self, ctx: TagContext):
        klines = ctx.data.items.get("stock.kline.daily")
        if not klines or len(klines) < 20:
            return None

        recent_vol = klines[-1]["volume"]
        avg_vol = sum(bar["volume"] for bar in klines[-20:]) / 20
        spike_ratio = recent_vol / avg_vol if avg_vol > 0 else 0

        return {
            "value": round(spike_ratio, 4),
            "start_date": ctx.data.now,
        }

类名随意，NTQ 通过 settings.py 里的 meta.key 找到它。

钩子函数详解

calculate_tag(ctx) -> Optional[Dict[str, Any]]

必须实现：是

调用时机：

entity_based 模式：每个交易日每只股票调用一次
slice_based 模式：每个时间切片调用（除非实现了 on_calendar_asof）
global 模式：每个交易日调用一次，不区分股票
non_time_series 模式：只调用一次

用途描述：标签计算的核心方法。NTQ 在回测时间窗口内，每个交易日对每个实体（或全局）调用一次。返回一个字典表示标签值，返回 None 表示当天不写入。

函数参数：ctx: TagContext — 只读上下文，提供当日数据、实体信息、标签定义等。

函数返回：Optional[Dict[str, Any]] — 返回字典表示写入标签值，返回 None 表示不写。

返回字典的字段：

字段	类型	必填	说明
value	Any	是	标签值，可以是数字、字符串、布尔等
start_date	str	否	标签值生效的起始日期，默认为当天
end_date	str	否	标签值生效的截止日期

代码示例：

def calculate_tag(self, ctx: TagContext):
    klines = ctx.data.items.get("stock.kline.daily")
    if not klines or len(klines) < 14:
        return None

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

    # 从 settings.core 读取阈值
    threshold = ctx.settings.raw.get("core", {}).get("rsi_threshold", 30)

    # 标签值：RSI 是否低于阈值
    return {
        "value": rsi < threshold,
        "start_date": ctx.data.now,
    }

读取上下文数据：

ctx.data.items 是一个映射，key 是 data_key（如 "stock.kline.daily"），value 是该数据源当天及之前的数据行列表。你在 settings.py 的 data.base 和 data.required 里声明的数据源，都会出现在这里。

# 读取 base 数据
klines = ctx.data.items.get("stock.kline.daily")

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

读取已有标签值（增量模式）：

在 incremental 模式下，ctx.data.prior_value 包含上一次计算写入的标签值。你可以用它判断"标签值是否变化"，避免重复写入。

def calculate_tag(self, ctx: TagContext):
    new_value = compute_my_tag(ctx)

    # 增量模式：和上次一样就不写
    if ctx.data.prior_value is not None and ctx.data.prior_value == new_value:
        return None

    return {"value": new_value, "start_date": ctx.data.now}

使用内存袋：

ctx.custom 是一个可写字典，在同一个实体的多次调用之间共享。适合存储中间状态。

def calculate_tag(self, ctx: TagContext):
    # 读上次存的累计值
    cumulative = ctx.custom.get("cumulative", 0)

    klines = ctx.data.items.get("stock.kline.daily")
    if klines:
        cumulative += klines[-1]["volume"]
        ctx.custom["cumulative"] = cumulative

    return {"value": cumulative, "start_date": ctx.data.now}

on_calendar_asof(ctx) -> TagCalendarAsOfResult

必须实现：否

调用时机：slice_based 模式下，每个日历时间点调用一次。entity_based、global、non_time_series 模式不调用。

用途描述：仅在 slice_based 模式下，当你需要做跨股票的横截面计算时使用。普通逐股计算走 calculate_tag；横截面计算走 on_calendar_asof。典型场景：每天对全部股票排序，取成交量前 10% 的股票打上"放量"标签。这种计算需要一次性拿到当天所有股票的数据，不能逐股独立算。

函数参数：ctx: TagContext — ctx.data.by_entity 包含当天所有实体的数据。

函数返回：TagCalendarAsOfResult — 包含当前日期、各实体的标签值、跨日会话状态。

from core.modules.tag.contracts import TagCalendarAsOfResult

def on_calendar_asof(self, ctx: TagContext):
    by_entity = ctx.data.by_entity or {}

    # 收集所有股票今天的成交量
    volumes = {}
    for entity_id, items in by_entity.items():
        klines = items.get("stock.kline.daily")
        if klines:
            volumes[entity_id] = klines[-1]["volume"]

    if not volumes:
        return TagCalendarAsOfResult(as_of_date=ctx.data.now)

    # 按成交量排序，取前 10% 打标签
    sorted_stocks = sorted(volumes.items(), key=lambda x: x[1], reverse=True)
    top_n = max(1, len(sorted_stocks) // 10)

    entity_tags = {}
    for entity_id, vol in sorted_stocks[:top_n]:
        entity_tags[entity_id] = [
            {"tag_name": "volume_top", "value": True, "start_date": ctx.data.now}
        ]
    for entity_id, vol in sorted_stocks[top_n:]:
        entity_tags[entity_id] = [
            {"tag_name": "volume_top", "value": False, "start_date": ctx.data.now}
        ]

    return TagCalendarAsOfResult(
        as_of_date=ctx.data.now,
        entity_tags=entity_tags,
    )

返回值 TagCalendarAsOfResult 的结构：

字段	类型	说明
as_of_date	str	当前日期
entity_tags	Dict[str, List[Dict]]	entity_id → 标签值列表
session_state	Dict[str, Any]	跨日会话状态，下次调用时可通过 ctx.data.calendar 取回

entity_tags 里每个条目的格式：

字段	类型	说明
tag_name	str	对应 settings.tag_definitions 里的 name
value	Any	标签值
start_date	str	可选，生效起始日期

session_state 适合存跨日的状态——比如"连续放量天数"计数器：

def on_calendar_asof(self, ctx: TagContext):
    # 从上次的状态里取连续放量计数
    prev_state = ctx.data.calendar.get("session_state", {}) if ctx.data.calendar else {}
    counts = prev_state.get("volume_streak", {})

    by_entity = ctx.data.by_entity or {}
    entity_tags = {}

    for entity_id, items in by_entity.items():
        klines = items.get("stock.kline.daily")
        if not klines:
            continue

        vol = klines[-1]["volume"]
        avg_vol = sum(bar["volume"] for bar in klines[-20:]) / min(20, len(klines))
        is_spike = vol > avg_vol * 2

        if is_spike:
            counts[entity_id] = counts.get(entity_id, 0) + 1
        else:
            counts[entity_id] = 0

        entity_tags[entity_id] = [
            {"tag_name": "consecutive_spike", "value": counts[entity_id]}
        ]

    return TagCalendarAsOfResult(
        as_of_date=ctx.data.now,
        entity_tags=entity_tags,
        session_state={"volume_streak": counts},
    )

TagContext 方法

方法	用途
ctx.data.items	获取当前实体的数据（DataKey → rows 映射）
ctx.data.by_entity	获取全部实体的数据（on_calendar_asof 专用）
ctx.data.now	当前日期
ctx.data.entity_id	当前实体 ID
ctx.data.tag_definition	当前标签定义
ctx.data.prior_value	增量模式下上次的标签值
ctx.custom	可写字典，跨调用共享状态
ctx.base_data_key	返回 settings.data.base.data_key
ctx.effective_settings_dict()	返回 settings 的字典视图

静态工具方法

TagHooks 基类暂不提供静态工具方法。如需读取 settings.core 里的参数，直接从 ctx.settings 取：

def calculate_tag(self, ctx: TagContext):
    threshold = ctx.settings.raw.get("core", {}).get("threshold", 30)
    period = ctx.settings.raw.get("core", {}).get("period", 14)