第 8 章 · 02 策略执行与推送编排

第 8 章 · 02 策略执行与推送编排

本节摘要：本节讲 Sequoia-X 「日常模式」的完整流程——主程序入口如何把「拉数据」「跑策略」「推飞书」三件事串成「一条两分钟跑完的流水线」。我们会先讲「初始化顺序」——为什么是「配置 → 日志 → 引擎」的固定顺序（依赖关系决定）；再讲「策略注册」的「列表追加」模式；然后讲「遍历执行」的三层 try/except 异常隔离；最后讲「推送编排」的「结果非空才推」逻辑。学完本节，你就完整理解了 Sequoia-X 「从命令行到飞书」的最后一公里——也是「整套系统的入口」。

内容来源：原项目主程序入口精读。

💡 核心心法：好的「主流程编排」的精髓是「顺序合理 + 边界清晰 + 失败隔离」——每一步都有「为什么这么放」的理由。

学习目标

阅读完本节，你应当能够：

1. 讲清 Sequoia-X 主流程的六步固定顺序及依赖关系。
2. 读懂「策略注册 + 遍历执行 + 路由推送」的完整结构。
3. 理解「结果非空才推」「异常隔离三层」「主流程非零退出」的工程价值。
4. 写一段可运行的「主程序入口」教学示例（Mock 全部依赖）。

一、主流程的六步固定顺序

Sequoia-X 日常模式的主流程按以下顺序执行，顺序不能乱（依赖关系决定）：

为什么是这个顺序？每一步都依赖前一步：

「配置先行」是整套系统的依赖根——任何模块都可能读 settings，所以它必须第一个初始化。

二、初始化细节

2.1 配置初始化

settings = get_settings()
​

get_settings() 是全局单例（第 4 章讲过）——整个进程共享同一份配置。这一步可能抛 ValidationError（必填字段缺失）——fail-fast 在此生效。

2.2 日志初始化

logger = get_logger(__name__)
​

get_logger(__name__) 让 logger 自动带模块名前缀（__main__ 或 main）。日志配置 rich 输出，带颜色、带级别、带时间。

2.3 引擎初始化

engine = DataEngine(settings)
​

DataEngine.__init__ 内部会自动：

• Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)：创建数据库目录
• 初始化 SQLite 表（CREATE TABLE IF NOT EXISTS）：幂等——重复跑不会出错

引擎初始化是「幂等」的——多次运行不会出问题。

三、策略注册：列表追加

strategies: list[BaseStrategy] = [
    MaVolumeStrategy(engine=engine, settings=settings),
    TurtleTradeStrategy(engine=engine, settings=settings),
    HighTightFlagStrategy(engine=engine, settings=settings),
    LimitUpShakeoutStrategy(engine=engine, settings=settings),
    UptrendLimitDownStrategy(engine=engine, settings=settings),
    RpsBreakoutStrategy(engine=engine, settings=settings),
    PrivatePlacementStrategy(engine=engine, settings=settings),
]
​

关键观察：

• 每个策略构造时传 engine 和 settings——显式依赖注入；
• 列表是构造顺序——也就是执行顺序；
• 新增策略只需要追加一行——主流程零修改。

四、推送器初始化

notifier = FeishuNotifier(settings)
​

FeishuNotifier 在初始化时不连接飞书——只在 send() 调用时才发起 HTTP 请求。这让「构造」和「发送」解耦——构造时不需要网络。

五、遍历执行：三层异常隔离

for strategy in strategies:
    strategy_name = type(strategy).__name__
    logger.info(f"执行策略：{strategy_name}")
    
    try:
        selected: list[str] = strategy.run()
        logger.info(f"{strategy_name} 选出 {len(selected)} 只股票")
        
        if selected:
            notifier.send(
                symbols=selected,
                strategy_name=strategy_name,
                webhook_key=strategy.webhook_key,
            )
        else:
            logger.info(f"{strategy_name} 无选股结果，跳过推送")
    
    except Exception as exc:
        logger.exception(f"{strategy_name} 出现未捕获异常，跳过该策略")
        # continue 隐式：继续下一个策略
​

三层异常隔离（第 5 章讲过）：

任意一层挂不影响整体——这是 Sequoia-X 「故障隔离」的完整呈现。

六、推送编排：「结果非空才推」

if selected:
    notifier.send(symbols=selected, ...)
else:
    logger.info(f"{strategy_name} 无选股结果，跳过推送")
​

为什么结果非空才推？

• 飞书机器人频繁推无信息会让用户「对消息麻木」；
• 选股信号常有显著为空的时段（如震荡市）——没必要打扰；
• 节省飞书 Webhook 调用次数——避免触发限流。

这是「产品视角的工程决策」——什么时候不发比什么时候发更重要。

七、主流程兜底

def main() -> None:
    try:
        # ... 整个流程 ...
    except Exception:
        try:
            _logger = get_logger(__name__)
            _logger.exception("主流程发生未捕获异常，程序终止")
        except Exception:
            import traceback
            traceback.print_exc()
        sys.exit(1)                 # 非零退出码
    
    logger.info("Sequoia-X V2 运行完成")
​

**主流程层（L3）**兜底的关键设计：

1. 再包一层 try/except——避免「记日志时再挂」的灾难；
2. 记录 + 非零退出——sys.exit(1) 让操作系统知道「这次运行失败了」；
3. 正常完成时打 INFO——Sequoia-X V2 运行完成 是「健康心跳」。

「非零退出码」是与定时任务系统通信的标准方式——crontab 会把非零退出码视为失败并可能发邮件。

八、可运行教学代码：完整主流程（Mock）

下面这段代码完整演示 Sequoia-X 日常模式的主流程。它Mock 数据引擎和飞书通知，让你无需真实环境就能看到主流程的全貌：

# main_orchestration_demo.py  # 可运行：日常模式主流程（Mock）
import time
from abc import ABC, abstractmethod


# === Mock Settings / Logger / Engine / Notifier ===
class Settings:
    feishu_webhook_url = "https://open.feishu.cn/hook/default"
    strategy_webhooks = {
        "turtle": "https://open.feishu.cn/hook/turtle",
        "ma_volume": "https://open.feishu.cn/hook/ma",
    }
    db_path = "data/test.db"
    start_date = "2024-01-01"
    
    def get_webhook_url(self, key: str) -> str:
        return self.strategy_webhooks.get(key.lower(), self.feishu_webhook_url)


def get_logger(name: str):
    import logging
    logging.basicConfig(level=logging.INFO, format="%(name)s - %(message)s")
    return logging.getLogger(name)


class MockEngine:
    def __init__(self, settings):
        self.settings = settings
    
    def sync_today_bulk(self) -> int:
        # 模拟 8 进程并行拉数据
        time.sleep(0.2)
        return 5200   # 模拟 5200 只股票


class MockNotifier:
    def __init__(self, settings):
        self.settings = settings
    
    def send(self, symbols, strategy_name, webhook_key):
        url = self.settings.get_webhook_url(webhook_key)
        print(f"  📤 推送 {strategy_name}（{webhook_key}）→ {url}：{symbols}")


# === 策略抽象基类（与第 5 章一致）===
class BaseStrategy(ABC):
    webhook_key: str = "default"
    
    def __init__(self, engine, settings):
        self.engine = engine
        self.settings = settings
    
    @abstractmethod
    def run(self) -> list[str]:
        ...


class MaVolumeStrategy(BaseStrategy):
    webhook_key = "ma_volume"
    def run(self) -> list[str]:
        return ["300750", "002594"]


class TurtleTradeStrategy(BaseStrategy):
    webhook_key = "turtle"
    def run(self) -> list[str]:
        return ["600519", "000001"]


class EmptyStrategy(BaseStrategy):
    """故意产出空结果，演示「无选股不推」逻辑。"""
    webhook_key = "empty"
    def run(self) -> list[str]:
        return []


class FailingStrategy(BaseStrategy):
    """故意抛异常，演示「异常隔离」逻辑。"""
    webhook_key = "failing"
    def run(self) -> list[str]:
        raise RuntimeError("模拟策略挂掉")


# === 主流程（Sequoia-X 简化版）===
def main():
    try:
        # 1. 初始化
        settings = Settings()
        logger = get_logger("main")
        engine = MockEngine(settings)
        
        logger.info("Sequoia-X V2 启动")
        
        # 2. 拉增量
        count = engine.sync_today_bulk()
        logger.info(f"快照同步完成，写入 {count} 只股票")
        
        # 3. 策略注册
        strategies = [
            MaVolumeStrategy(engine, settings),
            TurtleTradeStrategy(engine, settings),
            EmptyStrategy(engine, settings),
            FailingStrategy(engine, settings),
        ]
        notifier = MockNotifier(settings)
        
        # 4. 遍历执行 + 推送
        for strategy in strategies:
            name = type(strategy).__name__
            logger.info(f"执行策略：{name}")
            
            try:
                selected = strategy.run()
                logger.info(f"{name} 选出 {len(selected)} 只股票")
                
                if selected:
                    notifier.send(selected, name, webhook_key=strategy.webhook_key)
                else:
                    logger.info(f"{name} 无选股结果，跳过推送")
            
            except Exception:
                logger.exception(f"{name} 出现未捕获异常，跳过该策略")
    
    except Exception:
        logger.exception("主流程发生未捕获异常，程序终止")
        import sys
        sys.exit(1)
    
    logger.info("Sequoia-X V2 运行完成")


if __name__ == "__main__":
    main()
​

预期输出（部分）：

main - Sequoia-X V2 启动
main - 快照同步完成，写入 5200 只股票
main - 执行策略：MaVolumeStrategy
main - MaVolumeStrategy 选出 2 只股票
  📤 推送 MaVolumeStrategy（ma_volume）→ https://open.feishu.cn/hook/ma：['300750', '002594']
main - 执行策略：TurtleTradeStrategy
main - TurtleTradeStrategy 选出 2 只股票
  📤 推送 TurtleTradeStrategy（turtle）→ https://open.feishu.cn/hook/turtle：['600519', '000001']
main - 执行策略：EmptyStrategy
main - EmptyStrategy 选出 0 只股票
main - EmptyStrategy 无选股结果，跳过推送
main - 执行策略：FailingStrategy
main - 执行策略：FailingStrategy
main - FailingStrategy 出现未捕获异常，跳过该策略
main - Sequoia-X V2 运行完成
​

注意看：

• 4 个策略全部执行（包括故意挂掉的 FailingStrategy）——异常隔离；
• 空结果不推——EmptyStrategy 没有推送；
• 主流程最终打印「运行完成」——整体仍然成功。

九、本节在全景图的位置

策略执行与推送编排 ✅ ← 本节
        │
        ▼
下一章（第 9 章 质量保障与生产实践）：hypothesis 属性测试 + crontab 部署
​

本节要点回顾

1. 六步固定顺序：配置 → 日志 → 引擎 → 拉数据 → 注册策略 → 遍历执行——依赖关系决定。
2. 策略注册：列表追加模式——新增策略 = 列表里加一行。
3. 三层异常隔离：单股 → 单策略 → 主流程——任意一层挂不影响整体。
4. 结果非空才推：避免「对消息麻木」、节省 webhook 调用、产品级克制。
5. 主流程兜底：再包一层 try + 非零退出码——与定时任务系统通信。
6. 可运行示例：4 套策略（含空结果 + 异常）演示异常隔离 + 非空才推。

配套教学脚本：images/main_orchestration_demo.py——完整主流程 Mock 演示。

至此，第 8 章两节完成。下一章进入质量保障与生产实践——讲 hypothesis 属性测试与 crontab 部署。