Sequoia-X 项目研究站 · 新增策略
Sequoia-X 新增策略:四步把私有逻辑接进流水线
这套系统的扩展点很明确:继承 BaseStrategy、实现 run() 返回股票代码列表、给类设一个 webhook_key、把它加进 main.py 的策略列表。做完这四步,它就会和其他七个策略一样,在每天补完增量之后自动执行,命中结果推到对应的飞书群。
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 |
给 Sequoia-X 接自定义策略:五步
下面的骨架与现有策略结构一致,命名与注释风格照仓库保持。
新建策略文件
在
sequoia_x/strategy/下新增my_signal.py,导入BaseStrategy、get_logger与 pandas。文件名与类名风格参照现有策略(如ma_volume.py对应MaVolumeStrategy)。实现 run() 并返回代码列表
用
self.engine.get_local_symbols()拿全市场代码,逐只self.engine.get_ohlcv(symbol)取日 K,做向量化计算后把命中的代码append进列表返回。需要 60 日均线这类长窗口时,先判断len(df)是否够。设置 webhook_key
给类加一行
webhook_key = "my_signal",并在.env里补STRATEGY_WEBHOOK_MY_SIGNAL(不配则回落到默认 Webhook,也能跑)。注册到 main.py
在导入区加一行 import,在
strategies列表里追加实例YourSignalStrategy(engine=engine, settings=settings)。列表顺序就是执行顺序。跑一次验证
先跑
python main.py看日志里是否出现「YourSignalStrategy 选出 N 只股票」,再确认命中的推送落在了预期群里。空命中时只会有日志、不会有卡片。
ma_volume.py 是最短的一份(约 60 行),适合当模板;rps_breakout.py 展示了「一次性读全表做横截面排名」的另一种取数方式。把这两个文件对着读一遍,比从零设计快得多。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)而不是函数内魔法数字 | 需要试验不同阈值时 | 参数埋在下标里时很难批量调整与比对 |
| 结果不排序或明确排序 | 要排序就写明依据(如海龟按流通市值降序),不排就保持遍历顺序 | 结果可解释性 | 隐式顺序会让人误以为清单有优先级含义 |
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 里,异常不影响其他标的 | 读取失败时整体返回空列表 | 整表读取的失败面更大 |
| 额外联网 | 一般不联网(海龟查市值是例外) | 不联网 | 定增公告策略则完全走外部接口,不走库 |
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 = 100、target-version = "py311"、select = ["E","F","I","UP"] | 保持风格一致 | import 顺序(I)与语法升级(UP)也在检查范围内 |
| Python 版本 | requires-python >= 3.10,ruff 目标 py311 | 类型标注与语法可用范围 | 写 list[str] 这类内置泛型标注需要 3.9+ |
| 依赖锁定 | 仓库含 uv.lock | 可复现的环境 | 新增第三方依赖时应同步更新锁文件 |
二次开发常见问题
给 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 与你的环境管理方式为准。