更多用例 - 使用Strategy模块创建交易策略

策略模拟和扫描是整个框架的核心,需要理解的概念略多一些,我会尽量用最简单的方法描述清楚:

首先,我们需要理解一些基本的概念和理念

回测 vs 扫描

回测:就是用您定义好的一套交易规则在历史数据里测试它是不是达到了您要的交易目标

扫描:就是用您定义的策略跑到最新的数据(需要通过 DataSource 模块拉取最近数据)上去,对所有您觉得有价值的股票扫描一遍,看看有没有现成的机会

多层回测模拟

  • 机会枚举:我通过您定义的找到机会的方法去在历史记录里一条一条测,发现机会我就记录,一直到机会结束。枚举器的作用是记录整个历史中出现的机会(其实在当前系统里,它最重要的作用是缓存 ^_^)

  • 价格因子模拟:我通过枚举出来的所有机会去投资1股,一只到投资完整个时间轴,我看看这1个股票的价格是怎么波动的,我的ROI(回报率)是不是正的。这个回测的主要作用是快速验证策略是不是个有效策略

  • 资金模拟:我在按照我定义的策略枚举出的所有股票的所有时间段上进行模拟持仓投资,这个模拟加入了自己管理和分配策略,更加真实地模拟现实生活中的投资。里边包括我怎么持有多只股票,怎么分配仓位等等因素的记入

请注意:当前三层模拟的重要(参考)的含义是:

  • 机会枚举:看看我的策略能不能找到机会,找到的多不多。(机会枚举的中间数据对机器学习也比较友好)

  • 价格因子模拟:初步看看我的策略是不是对目标股票有效,但注意这里得到的ROI与您的实际策略投资回报不是一回事,这里只能证明您的策略是不是对目标股票产生正向收益,正向收益是不是足够大。

  • 资金模拟:这个模拟才是更接近实际的投资模拟。您还需要加入仓位管理才能真正实现盈利。有可能我的策略普遍对大部分股票都是盈利的,但我运气不好就是分配了大部分资金去了亏损的股票,那您的总体收益很可能是负的。

从 0 开始创建一个新策略

步骤1:先建一个策略目录

例如创建:

userspace/strategies/my_strategy/
├── settings.py
└── strategy_worker.py

目录名建议和 settings["name"] 一致,便于排查与管理。

 

步骤2:创建 settings.py(决定“怎么跑”)

拷贝根目录下的settings_example.py并且重命名为settings.py。您可以从 settings_example.py 精简出最小版本。最少建议包含:

  • name - 这个策略叫什么名字

  • is_enabled - 这个策略是不是开启的

  • core - 把您自己需要的参数放进这里边

  • data.base_required_data - 声明您的策略需要什么数据,框架会自动注入

  • goal - 策略的止损,止赢,投资时间限制等等目标

例如(最小可读示例):

Settings = {
    "name": "my_strategy",
    "description": "my first strategy",
    "is_enabled": True,
    "core": {"rsi_oversold_threshold": 20},
    "data": {
        "base_required_data": {"params": {"term": "daily", "adjust": "qfq"}},
        "extra_required_data_sources": [],
        "min_required_records": 30,
        "indicators": {"rsi": [{"period": 14}]},
    },
    "goal": {
        "expiration": {"fixed_window_in_days": 30, "is_trading_days": True},
        "stop_loss": {"stages": [{"name": "loss10%", "ratio": -0.1, "close_invest": True}]},
        "take_profit": {"stages": [{"name": "win20%", "ratio": 0.2, "close_invest": True}]},
    },
}

 

步骤3:写 strategy_worker.py(决定“什么时机发信号”)

最核心就是实现 scan_opportunity(data, settings),命中条件就返回 Opportunity,否则返回 None。

from typing import Dict, Any, Optional
from core.modules.strategy.base_strategy_worker import BaseStrategyWorker
from core.modules.strategy.models.opportunity import Opportunity


class MyStrategyWorker(BaseStrategyWorker):
    def scan_opportunity(self, data: Dict[str, Any], settings: Dict[str, Any]) -> Optional[Opportunity]:
        klines = data.get("klines", [])
        if not klines:
            return None
        latest = klines[-1]
        rsi = latest.get("rsi14")
        if rsi is None or rsi >= settings["core"]["rsi_oversold_threshold"]:
            return None
        return Opportunity(stock=self.stock_info, record_of_today=latest)

步骤4:运行并验证

python start-cli.py scan --strategy my_strategy
python start-cli.py enumerate --strategy my_strategy
python start-cli.py simulate --strategy my_strategy

 

步骤5:看结果目录

结果通常在:

userspace/strategies/my_strategy/results/

这个目录默认是本地产物目录,不建议提交大结果文件到 Git