场景分析

我的回测中需要频繁使用对一个公司市值大小的评判，我有没有办法提前评判好，回测的时候直接用就行？而且多个策略都可以用？

我的回测已经变得非常复杂了，其中每推进一步都需要大量计算，我有办法把一部分计算摘出来提前算完吗？这样我每次回测都会更快看到结果。

Tag 模块就是为了解决上述问题而设计的一套持续的、可追溯的、可复用的标签（也可以说是因子）的系统。

用户自定义位置：userspace/extensions/tags/{your_tag}/

需要先了解的概念

在自己定义标签（tag）之前，我们先需要了解几个基本概念：

• Scenario（场景）：可以理解为"这套标签在描述什么问题"。比如：哪些公司市值超过 50 亿，当前宏观经济状况，这些都可以是一个场景。
• Tag（标签）：是在某个场景下，对某个对象打上的分类结果。比如你把成交量高于某个阈值的情况定义为交易活跃（例如日成交量 1000 万手是分界点），那么标签就是"交易活跃"和"不活跃"。
• Tag value（标签值）：标签的具体值。还是成交量的例子：2 月 1 日成交量是 1001 万手，当天产生"活跃"标签；2 月 2 日变成了 990 万手，产生"不活跃"标签。返回值里可以带上 start_date / end_date，记录这个标签生效的时间区间。

注意，Tag 有两种种类——以实体为基准的（entity_based）和泛型的（general）：

• entity_based：针对某个具体数据实体的，比如公司市值标签，它有明确的股票作为对象。系统通过 data.base.data_key 自动推断。
• general：不绑定单一实体，比如用 GDP、LPR 等指标判断宏观环境，这类标签描述的是一段时间的总体状态，不针对某只股票。

这两种标签的执行路由不同——entity_based 走逐实体并行管线，general 走全局管线。你不需要手动指定，系统根据 data.base 引用的数据类型自动判断。

举例：从 0 创建一个新标签场景

第一步：建场景目录

userspace/extensions/tags/stock_activity/
├── settings.py      # 声明：算什么、按谁算、怎么算
└── tag.py           # 逻辑：具体的计算代码

第二步：写 settings.py

settings.py 里定义一个 Settings 字典，描述这个标签场景的元信息、数据依赖和计算方式：

Settings = {
    "is_enabled": True,
    "meta": {
        "key": "stock_activity",          # 必填，全局唯一标识
        "display_name": "成交量活跃度",
        "description": "根据日成交量判断股票交易活跃程度",
        "keywords": ["volume", "activity"],
        "category": ["technical"]
    },
    "core": {
        "activity_threshold": 10_000_000   # 你的自定义参数，在 tag.py 里读取
    },
    "data": {
        "base": {
            "data_key": "stock.kline.daily",   # 基础数据：日K线
            "params": {"adjust": "qfq"},
            "indicators": {}
        },
        "required": []    # 额外数据，留空表示只需要 base
    },
    "calculation": {
        "update_mode": "incremental",       # incremental（增量续跑）或 refresh（全量重算）
        "recompute": False,
        "execution": {
            "start_date": "",    # 空=用系统默认起始日期
            "end_date": "",      # 空=用最新交易日
            "mode": "entity_based"   # entity_based 或 slice_based
        }
    },
    "tag_definitions": [
        {"name": "high_activity", "display_name": "高成交量", "description": "日成交量达到阈值"},
        {"name": "low_activity", "display_name": "低成交量", "description": "日成交量低于阈值"},
    ],
}

几个关键配置块的意思：

第三步：写 tag.py

tag.py 里继承 TagHooks，实现 calculate_tag 方法。系统会对每个实体、每个 tag_definition 调用一次：

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

class StockActivityHooks(TagHooks):
    def calculate_tag(self, ctx: TagContext):
        """
        ctx.data.now              — 当前 as-of 日期（YYYYMMDD）
        ctx.data.entity_id        — 当前股票代码
        ctx.data.items            — 数据字典，key 是 data_key
        ctx.data.tag_definition   — 当前正在计算的标签定义
        ctx.settings              — 完整的 settings 对象
        ctx.custom                — 可变状态，跨调用共享
        """
        threshold = ctx.settings.raw_settings.get("core", {}).get("activity_threshold", 10_000_000)

        kline_rows = ctx.data.items.get("stock.kline.daily", [])
        if not kline_rows:
            return None    # 无数据，跳过

        latest = kline_rows[-1]    # 最新一天的数据
        volume = latest.get("volume", 0)

        tag_name = ctx.data.tag_definition.name if ctx.data.tag_definition else ""

        if tag_name == "high_activity":
            if volume >= threshold:
                return {"value": True, "start_date": ctx.data.now}
            return None

        if tag_name == "low_activity":
            if volume < threshold:
                return {"value": True, "start_date": ctx.data.now}
            return None

        return None

关键约定：

• 返回 None 表示不打标签
• 返回 {"value": ..., "start_date": ..., "end_date": ...} 表示写入标签
• value 可以是任意 JSON 可序列化的值（布尔、数字、字符串、对象）
• start_date / end_date 可选，不填默认用 ctx.data.now

第四步：运行并验证

# 执行所有已启用的标签场景
python cli.py t

# 只执行这一个场景
python cli.py t --scenario stock_activity

# 试运行（只计算不落库）
python cli.py t --scenario stock_activity --dry-run

# 限制只算 10 只股票（调试用）
python cli.py t --scenario stock_activity --entity-limit 10

计算完成后，标签数据写入 sys_tag_value 表。这些标签可以在策略中直接声明并自动注入使用。

其他可用配置

general 模式（泛型标签）

如果你的标签不针对某只股票，而是描述宏观状态（比如"LPR 是否下调"），把 data.base 指向 global 数据：

"data": {
    "base": {
        "data_key": "macro.lpr",    # global 时序数据
        "params": {},
        "indicators": {}
    }
}

系统会自动走 global 管线，ctx.data.entity_id 此时为空。你可以在 calculate_tag 里直接读 ctx.data.items 里的宏观时间序列。

增量 vs 全量重算

设 recompute: True 会强制使用 refresh 模式，不管 update_mode 写的是什么。

slice_based 模式

默认的 entity_based 模式是逐实体并行计算。如果你的标签需要横截面比较（比如"成交量排名前 10%"），可以用 slice_based 模式——按日期切片，同一天所有实体一起算：

"calculation": {
    "execution": {
        "mode": "slice_based"
    }
}

slice_based 模式下可以覆写 on_calendar_asof 方法，一次性处理当天所有实体：

def on_calendar_asof(self, ctx: TagContext):
    """横截面钩子：一次性拿到当天所有实体的数据。"""
    all_data = ctx.data.by_entity    # {entity_id: items}
    # 计算、排序、取前 N%
    # ...
    return TagCalendarAsOfResult(
        as_of_date=ctx.data.now,
        entity_tags={entity_id: [{"tag_name": "top_volume", "value": True}]},
    )

在策略中使用标签

标签算好之后，在策略的 settings.py 里声明数据依赖即可自动注入：

data = {
    "required": [
        {"data_key": "tag.stock_activity", "params": {}},
    ]
}

策略代码里通过 ctx.get_data("tag.stock_activity", entity_id) 读取该股票的标签值。