Sequoia-X 项目研究站 · 新增策略

Sequoia-X 新增策略:四步把私有逻辑接进流水线

这套系统的扩展点很明确:继承 BaseStrategy、实现 run() 返回股票代码列表、给类设一个 webhook_key、把它加进 main.py 的策略列表。做完这四步,它就会和其他七个策略一样,在每天补完增量之后自动执行,命中结果推到对应的飞书群。

扩展方式:新增策略类 + 注册不需要改动框架其他部分依据 strategy/base.py 与 main.py(2026-09 核验)
继承基类BaseStrategy,拿到 engine 与 settings
实现 run()返回股票代码列表
设置 webhook_key决定推送到哪个群
注册到 main.py策略列表顺序即执行顺序
策略契约示意(依据 strategy/base.py 与 main.py 整理,非官方流程图)。
Contract

Sequoia-X 的 BaseStrategy 契约

约束很少,但每一条都有明确用途。

契约项定义适用场景注意点
基类class BaseStrategy(ABC),用抽象方法强制子类实现所有具体策略的共同父类不实现 run() 的类无法实例化
抽象方法@abstractmethod def run(self) -> list[str]执行选股逻辑的入口返回空列表表示当天没有命中,不会触发推送
类属性webhook_key: str = "default"决定推送到哪个飞书群子类覆盖它(如 "rps"),再在 .env 里配同名键
构造参数__init__(self, engine: DataEngine, settings: Settings)拿到数据引擎与配置源码把它存成 self.engine / self.settings,新策略直接复用
异常边界现有策略在逐只循环内 try/except 并记 warning单只标的异常不应影响全市场扫描这是既有实现的写法约定,新策略建议保持一致
日志get_logger(__name__) 取命名 logger输出可追溯的执行记录同名 logger 幂等,不会重复叠加 handler
Steps

给 Sequoia-X 接自定义策略:五步

下面的骨架与现有策略结构一致,命名与注释风格照仓库保持。

  1. 新建策略文件

    sequoia_x/strategy/ 下新增 my_signal.py,导入 BaseStrategyget_logger 与 pandas。文件名与类名风格参照现有策略(如 ma_volume.py 对应 MaVolumeStrategy)。

  2. 实现 run() 并返回代码列表

    self.engine.get_local_symbols() 拿全市场代码,逐只 self.engine.get_ohlcv(symbol) 取日 K,做向量化计算后把命中的代码 append 进列表返回。需要 60 日均线这类长窗口时,先判断 len(df) 是否够。

  3. 设置 webhook_key

    给类加一行 webhook_key = "my_signal",并在 .env 里补 STRATEGY_WEBHOOK_MY_SIGNAL(不配则回落到默认 Webhook,也能跑)。

  4. 注册到 main.py

    在导入区加一行 import,在 strategies 列表里追加实例 YourSignalStrategy(engine=engine, settings=settings)。列表顺序就是执行顺序。

  5. 跑一次验证

    先跑 python main.py 看日志里是否出现「YourSignalStrategy 选出 N 只股票」,再确认命中的推送落在了预期群里。空命中时只会有日志、不会有卡片。

策略骨架(结构与仓库现有策略一致)"""自定义信号策略:说明你判断什么形态。""" from sequoia_x.core.logger import get_logger from sequoia_x.strategy.base import BaseStrategy logger = get_logger(__name__) class YourSignalStrategy(BaseStrategy): """一句话说明选股条件。""" webhook_key: str = "my_signal" _MIN_BARS: int = 20 def run(self) -> list[str]: symbols = self.engine.get_local_symbols() selected: list[str] = [] for symbol in symbols: try: df = self.engine.get_ohlcv(symbol) if len(df) < self._MIN_BARS: continue # 向量化计算:rolling / shift / rank,不要用 iterrows df["ma20"] = df["close"].rolling(20).mean() last = df.iloc[-1] if last["close"] > last["ma20"]: selected.append(symbol) except Exception as exc: # noqa: BLE001 logger.warning(f"[{symbol}] YourSignalStrategy 计算失败:{exc}") continue logger.info(f"YourSignalStrategy 选出 {len(selected)} 只股票") return selected
先看现成的再动手:ma_volume.py 是最短的一份(约 60 行),适合当模板;rps_breakout.py 展示了「一次性读全表做横截面排名」的另一种取数方式。把这两个文件对着读一遍,比从零设计快得多。
Conventions

Sequoia-X 策略的四条写法约定

这些约定来自现有七个策略的一致写法,不是硬性限制,但破坏它们会带来实际问题。

约定具体做法适用场景破坏后的后果
向量化计算rolling / shift / rank 等 pandas 方法,源码注释直接写「严禁 iterrows」全市场逐只计算逐行遍历会让日常运行从分钟级退化到不可接受
窗口隔离比较基准要排除当日,例如 high.shift(1).rolling(20).max()判断「突破前 N 日高点」这类条件不隔离会形成「当日包含自身」的自指比较,条件恒成立
最小样本用类常量声明(如 _MIN_BARS)并提前 continue需要 20 / 40 / 60 根窗口的策略窗口不足时得到 NaN,静默漏选或报错
单只容错逐只 try/except,异常写 warning 后 continue个别标的脏数据一个异常会让整个策略中断,后续股票全部漏算
可调参数外置用类属性(如 rps_period = 120)而不是函数内魔法数字需要试验不同阈值时参数埋在下标里时很难批量调整与比对
结果不排序或明确排序要排序就写明依据(如海龟按流通市值降序),不排就保持遍历顺序结果可解释性隐式顺序会让人误以为清单有优先级含义
Data access

Sequoia-X 的两条取数路径

现有策略里两种写法都有,选错了要么慢要么算不出。

对比项逐只读库 get_ohlcv一次性读全表 read_sql适用场景
调用方式self.engine.get_ohlcv(symbol),内部按 symbol 查询并排序sqlite3.connect(self.engine.db_path) + pd.read_sql前者适合形态类,后者适合横截面类
内存占用每次一只,峰值小整表进内存,随库增长机器内存有限时优先前者
能否做横向排名不能(看不到其他标的)可以,如 rank(pct=True)相对强度、行业内排名这类需求
现有例子海龟、均线放量、高窄旗形、涨停洗盘、上升跌停RPS 突破可以在仓库里各找一份对照
异常表现包在逐只 try 里,异常不影响其他标的读取失败时整体返回空列表整表读取的失败面更大
额外联网一般不联网(海龟查市值是例外)不联网定增公告策略则完全走外部接口,不走库
Tests

Sequoia-X 的测试与代码规范

仓库自带测试与静态检查配置,扩展时建议沿用。

项目配置适用场景注意点
测试框架pytest(testpaths = ["tests"]运行 pytest 即可执行全部测试新增策略建议在 tests/ 下补对应用例
属性测试hypothesis(dev 依赖 >=6.100)对边界条件做随机化验证阈值类策略尤其适合验证「窗口不足时的行为」
模拟对象pytest-mock(dev 依赖 >=3.12)替换数据引擎与网络请求现有测试文件名可按模块对照:config / data_engine / feishu / logger / main / strategy
代码检查ruff:line-length = 100target-version = "py311"select = ["E","F","I","UP"]保持风格一致import 顺序(I)与语法升级(UP)也在检查范围内
Python 版本requires-python >= 3.10,ruff 目标 py311类型标注与语法可用范围list[str] 这类内置泛型标注需要 3.9+
依赖锁定仓库含 uv.lock可复现的环境新增第三方依赖时应同步更新锁文件
关于测试的边界:仓库里的测试覆盖配置、数据引擎、通知、日志、主入口与策略几个模块,属于工程正确性层面的自测;项目本身不含回测或绩效评估测试,因此新增策略的「有效性」不在代码测试能回答的范围内。
FAQ

二次开发常见问题

给 Sequoia-X 加策略必须继承 BaseStrategy 吗?

如果在 main.py 的策略列表里使用,是的——列表元素类型标注为 BaseStrategy,而基类用抽象方法强制子类实现 run()。若只是想临时算一下,也可以直接写脚本调用数据引擎,但那样不会进入每日流水线。

Sequoia-X 新策略怎么推到自己的群?

给类设一个 webhook_key(如 "my_signal"),然后在 .env 里加 STRATEGY_WEBHOOK_MY_SIGNAL。键名不匹配时会回落到 FEISHU_WEBHOOK_URL,也就是说「配错了也能跑,但会进错群」。

Sequoia-X 能用分钟线或财务数据写策略吗?

不能直接支持。本地库只存日 K(后复权),表里也没有财务字段;要用其他数据需要自己扩展取数与存储,那已经超出「加一个策略类」的范围。数据边界见数据层页。

Sequoia-X 策略执行顺序会影响结果吗?

不影响各自的判定结果——每个策略独立读库、独立返回清单,彼此不共享状态。顺序只决定日志打印与推送的先后。反过来说,同一只股票可能同时被多个策略选中,推送里会出现多条。

怎么判断 Sequoia-X 策略是不是太慢?

看日志里策略行之间的时间差。逐只读库的方式耗时与标的数成正比,全表读取则受内存与分组计算影响;如果某个策略明显拖慢整体,优先检查是否误用了逐行遍历。

改了源码要重装吗?

pip install . 安装的话,源码改动需要重新安装或使用可编辑安装(pip install -e .)才能生效;用 uv sync 管理环境时同理。具体以官方 README 与你的环境管理方式为准。

扩展前先看清失败路径

排障页把配置、数据、交易日与推送四类问题分开处理,并给出安全重跑方式。