更多用例 - 创建特征因子标签

场景分析

我的策略只需要对我认为的财务健康的股票进行投资，我在我的策略里已经写了，但是跑起来太慢了，我该怎么办？

我有一套我自己的判断经济大环境的逻辑算法，需要PPI，CPI，GDP等等一些数据。我现在想看看我当前的策略在不同经济环境下的表现，每次跑策略前我的经济判断逻辑都要先跑一遍，产生了大量重复计算。我有没有办法做一套标记，让我的算法算出的经济环境在不同的时间点放置不同的经济大环境的标签？

我有一套我自己判断高潜力股票的算法，但我不想在每次跑策略的时候都现去算，我有办法把我的这些算法一次性安插在所有股票上，策略里发现不是高潜力股票就自动跳过吗？

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

用户自定义位置：userspace/tags/{your tag}

需要先了解的概念

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

  • Scenario - 场景：可以理解为“这套标签在描述什么问题”。比如：哪些公司市值超过 50 亿，当前宏观经济状况，这些都可以是一个场景。

  • Tag - 标签：是在某个场景下，对某个对象打上的分类结果。比如您把 成交量高于某个阈值的情况定义为交易活跃（例如日成交量1000万手是分界点），那么标签就是交易活跃和不活跃。

  • Tag value - 标签备忘录。用于记录标签产生时的重要信息。还是成交量的例子：当前股票在 2 月 1 日 交易量是1001万手，当天会产生“活跃”标签，2月2日变成了990万手，此时会产生平淡的标签。tag_value 可以把“从多少到多少”这类原因和上下文一起记录下来（JSON 格式）比如2月2日，股票不再活跃，因为交易总量从1001万手变成了990万手

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

  • entity_based：一种是针对某个数据的，比如公司市值标签，它有明确的数据实体对象。

  • general：另一种是不绑定单一实体的，泛类的，比如您用 GDP、LPR 等指标判断宏观环境，这类标签描述的是一段时间的总体状态，并不针对某个数据或者股票这种实体类型

这两种标签的声明方式略有不同，请在配置时注意区分。

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

第一步：先建场景目录

userspace/tags/my_tag_scenario/
├── settings.py
└── tag_worker.py

第二步：修改配置 settings.py（决定“算什么、按谁算”）

首先在tags根目录拷贝一份 example_settings.py 重命名为 settings.py

至少包含以下设置：

  • name 您的标签场景的唯一名字

  • is_enabled 是不是开启的

  • tag_target_type 是针对某个数据还是泛型的

  • data.required 您在计算标签时需要注入的数据

  • update_mode 您是每次重算？还是增加在上次计算结果的后边？

  • tags 在当前场景下产生的所有标签定义（如：标签1:繁荣的经济环境，标签2:开始衰退的经济环境 等等 ...）

示例（entity_based）：

from core.global_enums.enums import EntityType, UpdateMode
Settings = {
    "is_enabled": True,
    "name": "stock_activity_scenario",
    "tag_target_type": "entity_based",
    "target_entity": {"type": EntityType.STOCK_KLINE_DAILY.value},
    "data": {
        "required": [{"data_id": "stock.kline", "params": {"term": "daily", "adjust": "qfq"}}],
    },
    "update_mode": UpdateMode.INCREMENTAL.value,
    "core": {
    	"activity_threshold": 10_000_000
    },
    "performance": {"max_workers": "auto"},
    "tags": [
    	{
    		"name": "high_activity", 
    		"display_name": "高成交量" 
    	},
    	{
    		"name": "low_activity", 
    		"display_name": "低成交量" 
    	}
    ],
}

 

第三步：写 tag_worker.py（决定“怎么计算” )

核心是实现 calculate_tag(as_of_date, historical_data, tag_definition)：(以下只是示意代码，实际可以根据不同版本有差别)

from typing import Any, Dict, Optional
from core.modules.tag.base_tag_worker import BaseTagWorker

class MyTagWorker(BaseTagWorker):
    def calculate_tag(
        self,
        as_of_date: str,
        historical_data: Dict[str, Any],
        tag_definition: Any,
    ) -> Optional[Dict[str, Any]]:
        latest = historical_data.get("stock.kline", [])
        if latest.get("volume") >= self.settings["core"].get("activity_threshold")
        	return tag_definition.get("high_activity")
        return tag_definition.get("low_activity")
        #如果你要返回tag value，可以这样：
        # return {"value": "high_activity", "your_values": "xxx"}

 

第四步：运行并验证

python start-cli.py tag --scenario stock_activity_scenario

当系统提示完成后就代表计算已经成功了，您可以在db中查看生成的tag。生成的这些tags可以直接在策略中声明并自动注入以供使用。