更多用例 - 用自定义策略扫描市场机会

策略扫描做什么?

策略扫描(scan)做的是:使用我们定义过的策略去当前市场上扫描看看有没有什么投资机会。一句话说:就是应用我们的策略去找机会。

两种扫描模式

  • 严格模式(默认):要求数据更新到最新交易日才能扫描。如果数据不是最新的,扫描会被门闸拦截并提示你先更新数据。
  • 演示模式(--demo):放宽严格交易日门闸,使用数据里最近的一个日期来扫描。方便测试和演示。

在策略的 settings.py 里配置:

Settings = {
    # ...
    "scanner": {
        "use_strict_previous_trading_day": True,    # True=严格模式,False=宽松
        "adapters": ["console"],                       # 输出适配器
    },
    # ...
}

扫描的步骤

前提条件:更新数据到最新

框架不自带付费数据源,但内置了 Tushare 数据更新逻辑(需要购买 Tushare 数据服务)。你也可以接入自己的数据源或使用免费数据源。

# 更新数据
python cli.py renew

# 简写
python cli.py r

 

运行扫描

# 扫描所有已启用的策略
python cli.py scan

# 简写
python cli.py c

# 只用某个策略扫描
python cli.py scan --strategy my_strategy

# 演示模式(不检查数据是否最新)
python cli.py scan --demo

系统会自动查找所有 enabled 策略并放入策略集,然后逐个扫描所有股票(你也可以定制股票候选池),看看有什么投资机会。

结果在哪看

终端

console adapter 会直接打印本期扫到的机会摘要,包括扫描日期、机会数、命中股票数、涨停数等。

输出目录

每次扫描的落盘结果在策略目录下,按日期组织:

策略目录/results/scan/{YYYYMMDD}/
├── scan_summary.json       # 扫描汇总(日期、机会数、命中股票等)
└── opportunities.csv       # 机会明细(股票代码、信号、指标等)

同一日期重复扫描时,如果已存在 scan_summary.json,会直接复用缓存结果,不会重跑。需要强制重跑时删除该日期目录或使用 --demo(宽松模式下会跳过门闸检查)。

扫描结果的保留版本数由 data.json 里的 retention.scan_results_max_versions 控制,默认保留 10 个日期目录,超出的自动清理。

自定义输出 Adapter

如果你想把扫描结果通过邮件、微信等方式推送,可以自定义 adapter。

创建 adapter

在 userspace/extensions/adapters/ 下创建:

userspace/extensions/adapters/
└── my_notifier/
    ├── adapter.py       # 必需:继承 BaseOpportunityAdapter
    └── settings.py      # 可选:adapter 配置

写 adapter.py

from core.modules.adapter.contracts import BaseOpportunityAdapter

class MyNotifierAdapter(BaseOpportunityAdapter):
    def process(self, opportunities, context):
        """
        opportunities: 机会列表,每个是 Opportunity 对象
        context: 上下文信息
            - date: 扫描日期
            - strategy_name: 策略名称
            - scan_summary: 扫描汇总统计
        """
        date = context.get("date", "")
        strategy = context.get("strategy_name", "")

        if not opportunities:
            self.log_info(f"{date} {strategy} 无机会")
            return

        # 你的通知逻辑:发邮件、调 webhook、写文件等
        for opp in opportunities:
            self.log_info(f"  {opp.stock_id} {opp.signal}")

        # 示例:调 webhook
        # import requests
        # requests.post("https://your-webhook", json={...})

在策略里启用

在策略的 settings.py 里加上 adapter 名称:

"scanner": {
    "adapters": ["console", "my_notifier"],   # 可以同时用多个
}

如果指定的 adapter 不存在或执行报错,系统会回退到默认输出,不会中断扫描。

scanner 配置字段一览

字段

默认值

说明

use_strict_previous_trading_day

True

是否严格要求最新交易日数据

adapters

["console"]

输出适配器名称列表,可同时配多个

max_cache_days

取 data.json 的 retention.scan_results_max_versions

扫描结果保留多少个日期目录

watch_list

""

限定扫描范围,传入股票代码列表