第 7 章 · 03 多机器人路由与兜底机制

第 7 章 · 03 多机器人路由与兜底机制

本节摘要：本节讲 Sequoia-X 触达层的「分发机制」——多机器人路由。A 股选股策略有 7 套（行情 6 套 + 事件 1 套），但不是所有结果都该发到同一个群。海龟突破的票常是趋势型投资者关注的，涨停洗盘的票常是短线型投资者关注的——两类人应该看到不同的信号。Sequoia-X 的解法是「多机器人路由」：每个策略可配独立的飞书机器人（甚至是不同的飞书群），没配的策略自动兜底到默认机器人。本节会讲清「为什么需要多机器人」「怎么配置」「路由的查询逻辑」「兜底机制的实现」四个工程要点，并演示「缺配置时如何优雅降级」。

内容来源：原项目配置管理 + 飞书通知模块的路由代码精读。

💡 核心心法：好的「路由系统」应该「约定优于配置 + 优雅降级」——约定让新增策略零成本接入，降级让缺配置不会阻塞主流程。

学习目标

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

1. 说出「为什么需要多机器人」的产品理由与工程理由。
2. 理解「约定优于配置 + 动态键扫描」在路由中的体现。
3. 读懂 get_webhook_url() 的两段式查询逻辑。
4. 写一段可运行的路由查询函数。

一、为什么需要多机器人

1.1 产品理由：信息分层

不同投资者关注不同信号：

如果所有策略结果都发到同一个群：

• 趋势型被涨停洗盘的短线信号淹没（信号过载）；
• 短线型被海龟突破的趋势信号误导（信号错配）。

「多机器人」本质是信息分层——让每类信号到达合适的人。

1.2 工程理由：失败隔离

如果所有策略共用一个 Webhook：

• 某条策略频繁推送（每日 50 条）→ 整个 Webhook 触发限流 → 其他策略也被影响；
• 某条策略触发飞书安全机制（敏感词）→ 整个 Webhook 被封禁 → 全盘推送失败。

「多机器人」让单个策略的失败不会扩散到其他策略。

二、配置约定

Sequoia-X 的路由配置采用「约定优于配置」：

# 默认 Webhook（必填，作为兜底）
FEISHU_WEBHOOK_URL=https://open.feishu.cn/hook/default

# 策略专属 Webhook（可选）
STRATEGY_WEBHOOK_MA_VOLUME=https://open.feishu.cn/hook/ma
STRATEGY_WEBHOOK_TURTLE=https://open.feishu.cn/hook/turtle
STRATEGY_WEBHOOK_FLAG=https://open.feishu.cn/hook/flag
STRATEGY_WEBHOOK_SHAKEOUT=https://open.feishu.cn/hook/shakeout
STRATEGY_WEBHOOK_LIMIT_DOWN=https://open.feishu.cn/hook/limit_down
STRATEGY_WEBHOOK_RPS=https://open.feishu.cn/hook/rps
STRATEGY_WEBHOOK_PRIVATE_PLACEMENT=https://open.feishu.cn/hook/private_placement
​

约定：

策略标识与配置键名的对应关系（Sequoia-X 约定）：

第 4 章「配置管理」讲过——Sequoia-X 启动时扫描 STRATEGY_WEBHOOK_* 键，自动收编到 strategy_webhooks 字典——新增策略不需要改任何配置代码。

三、查询逻辑：两段式路由

Sequoia-X 的路由查询逻辑非常简洁——两段式：

def get_webhook_url(self, webhook_key: str) -> str:
    """
    优先从 strategy_webhooks 查找，找不到则 fallback 到 feishu_webhook_url。
    """
    return self.strategy_webhooks.get(webhook_key.lower(), self.feishu_webhook_url)
​

两段式：

两个值得细抠的点：

3.1 webhook_key.lower()

策略类里的 webhook_key 可能是大写或小写，配置键一定是大写。lower() 把策略标识统一转小写做查询——避免大小写不一致导致的查询失败。

3.2 dict.get(key, default) 的二段式语义

dict.get(key, default) 是 Python 字典的两段式查询——key 存在就返回值，不存在就用 default。

这是 Python 写「带默认值的查询」的惯用法——比 if key in dict: ... else: ... 简洁 10 倍。

四、兜底机制的多层意义

兜底机制有三层意义：

「兜底」只解决「配置缺失」——其他两类错误有更严格的上层兜底。

五、可运行教学代码

下面这段代码完整演示「两段式路由 + 多层兜底」的查询逻辑：

# routing_demo.py  # 可运行：飞书多机器人路由查询
import os


def make_routes(env: dict) -> dict[str, str]:
    """扫描 STRATEGY_WEBHOOK_ 前缀的环境变量。"""
    prefix = "STRATEGY_WEBHOOK_"
    return {
        k[len(prefix):].lower(): v
        for k, v in env.items()
        if k.startswith(prefix)
    }


def get_webhook_url(
    strategy_webhooks: dict[str, str],
    default_url: str,
    webhook_key: str,
) -> tuple[str, str]:
    """
    两段式路由：先查专属 URL，再兜底到默认 URL。
    返回 (url, 来源标签)。
    """
    url = strategy_webhooks.get(webhook_key.lower(), default_url)
    source = "专属" if webhook_key.lower() in strategy_webhooks else "兜底"
    return url, source


# === 演示场景 ===
if __name__ == "__main__":
    # 场景 1：完整配置（7 个策略全有专属 URL）
    env_full = {
        "FEISHU_WEBHOOK_URL": "https://open.feishu.cn/hook/default",
        "STRATEGY_WEBHOOK_TURTLE":  "https://open.feishu.cn/hook/turtle",
        "STRATEGY_WEBHOOK_RPS":     "https://open.feishu.cn/hook/rps",
        "STRATEGY_WEBHOOK_FLAG":    "https://open.feishu.cn/hook/flag",
    }
    routes = make_routes(env_full)
    print("【场景 1：完整配置】")
    for k in ["turtle", "rps", "flag", "ma_volume", "private_placement"]:
        url, src = get_webhook_url(routes, env_full["FEISHU_WEBHOOK_URL"], k)
        print(f"  {k:>15s} → [{src}] {url}")
    print()
    
    # 场景 2：最小配置（只配默认）
    env_min = {"FEISHU_WEBHOOK_URL": "https://open.feishu.cn/hook/default"}
    routes = make_routes(env_min)
    print("【场景 2：最小配置（全部兜底）】")
    for k in ["turtle", "rps", "flag"]:
        url, src = get_webhook_url(routes, env_min["FEISHU_WEBHOOK_URL"], k)
        print(f"  {k:>15s} → [{src}] {url}")
    print()
    
    # 场景 3：大小写不敏感
    print("【场景 3：大小写不敏感】")
    url, src = get_webhook_url(routes, env_min["FEISHU_WEBHOOK_URL"], "TURTLE")
    print(f"  'TURTLE' → [{src}] {url}")
    print(f"  'Turtle' → [{src}] {url}")
​

预期输出：

【场景 1：完整配置】
           turtle → [专属] https://open.feishu.cn/hook/turtle
              rps → [专属] https://open.feishu.cn/hook/rps
             flag → [专属] https://open.feishu.cn/hook/flag
        ma_volume → [兜底] https://open.feishu.cn/hook/default
private_placement → [兜底] https://open.feishu.cn/hook/default

【场景 2：最小配置（全部兜底）】
  turtle → [兜底] https://open.feishu.cn/hook/default
     rps → [兜底] https://open.feishu.cn/hook/default
    flag → [兜底] https://open.feishu.cn/hook/default

【场景 3：大小写不敏感】
  'TURTLE' → [兜底] https://open.feishu.cn/hook/default
  'Turtle' → [兜底] https://open.feishu.cn/hook/default
​

这段代码清晰展示：两段式路由 + 大小写不敏感 + 优雅兜底三件套的完整工作方式。

六、配置层级的「金字塔」

把 Sequoia-X 的飞书配置层级画成金字塔：

                        ┌──────────┐
                        │ 默认 Webhook│  ← 必填，兜底
                        └──────────┘
                              ▲
                              │ 兜底
                              │
        ┌──────────┐  ┌──────────┐  ┌──────────┐
        │ 海龟   │  │ RPS     │  │ 定增公告 │  ← 可选，越多越好
        └──────────┘  └──────────┘  └──────────┘
              ↑              ↑              ↑
              │              │              │
         趋势型群       牛市型群       价值型群
​

金字塔顶端 = 必填（默认 Webhook）；
金字塔底部 = 可选（策略专属 Webhook）。

系统的「最低生存要求」只有默认 Webhook——全部专属 URL 都不配也能跑（全部兜底）。这种「最弱可工作条件」的设计很友好。

七、为什么不用「Web 界面管理」

读者可能会问：「为什么不搞个 Web 界面管理 Webhook？」答案和 Sequoia-X 的整体设计哲学一致：

Sequoia-X 的选股系统只需要「能跑」——Web 界面是过度设计。

八、常见配置错误速查

九、本节在全景图的位置

多机器人路由 ✅ ← 本节
        │
        ▼
下一节（第 04 节）触达链路细节：代码转换、名称查询与容错
​

本节要点回顾

1. 为什么需要多机器人：信息分层（不同投资者关注不同信号）+ 失败隔离（单策略失败不扩散）。
2. 配置约定：STRATEGY_WEBHOOK_<策略标识>=<URL>——约定优于配置。
3. 两段式路由：strategy_webhooks.get(key, default_url)——先查专属、再兜底。
4. 大小写不敏感：webhook_key.lower() 统一转小写做查询。
5. 配置金字塔：默认 Webhook 是「必填」、策略专属是「可选」、最弱可工作条件友好。
6. 可运行示例：3 种场景（完整/最小/大小写），亲眼看两段式路由的过滤。

配套教学脚本：images/routing_demo.py——演示两段式路由查询。

下一节，把所有「细节」串起来——讲飞书触达链路的最后三件小事：雪球代码转换、股票名称二次查询、HTTP 失败容错。