querybus — 统一搜索/资讯层

独立 Git 仓库：github.com/W-M-R/querybus（见 ADR-0007）。本目录在主仓库已 gitignore，通过 querybus @ file://./querybus 安装。多搜索引擎 + cache → manager fall-over 抽象层；目标是可独立开源（PyPI 候选）。顶层目录：querybus/（不是 stocklens 的子包，与 tickbridge-server/ 同级）。

产品愿景背景

querybus 是 stocklens 主仓「拼凑独立好仓 → 整合成 Web UI」路线上的第二个 bridge 包，跟 tickbridge 同构：

  ┌──────────────────────────────────────────────┐
  │   web UI (FastAPI + 模板) — 行情 / 分析 / 资讯  │
  └────────────┬─────────────────────────────────┘
               │ HTTP / API
               ▼
  ┌──────────────────────────────────────────────┐
  │   stocklens (业务编排：pipeline / scoring /     │
  │              agent / portfolio …)              │
  └─┬──────────┬──────────┬──────────┬───────────┘
    │          │          │          │
    ▼          ▼          ▼          ▼
 tickbridge querybus   pulsefan   llm-bridge ...
 (行情)     (本文)      (推送)      (规划)

querybus 抽象 7 个搜索 Provider 进同一条门面：

在 stocklens 主仓里 in-tree 共存（开发期）；
自带 pyproject.toml / LICENSE / README.md，未来可独立成 pip install querybus；
web UI 可以直接驱动它 —— REST endpoint 简单转发到 provider.search(...)。

角色

唯一对外门面：querybus.SearchProvider。
多源搜索插件：querybus.contracts.BaseSearchProvider 的子类（tavily / serpapi / bocha / brave / minimax / searxng / akshare_news），按 priority 顺序 fall-over。
可注入缓存：CacheProtocol —— 默认走 querybus 内置的 InMemoryTTLCache，但宿主可注入自有实现（Redis / Memcached）。
业务壳留在 stocklens：股票特化的查询模板、ETF/指数检测、多维度情报路由 —— 一律放在 stocklens/search/stock_search_service.py，不污染 querybus。

querybus/
├── __init__.py                顶层暴露 SearchProvider, SearchResult, SearchResponse, DimensionSpec, __version__
├── pyproject.toml             独立包定义（pip install -e querybus/）
├── LICENSE                    MIT
├── README.md                  独立项目说明（不引用 stocklens）
├── core/
│   ├── provider.py            SearchProvider 门面（唯一对外类）
│   ├── manager.py             SearchManager — 多源 fall-over 调度
│   ├── cache.py               InMemoryTTLCache（cachetools TTLCache 薄壳，独立于 kvcache）
│   ├── filters.py             filter_by_time_window / dedupe_results
│   └── reporting.py           format_intel_report + DimensionSpec（通用化）
├── contracts/
│   ├── base_provider.py       BaseSearchProvider (ABC)
│   ├── cache_protocol.py      CacheProtocol（duck-typed）
│   └── retry.py               _post_with_retry / _get_with_retry / _SEARCH_TRANSIENT_EXCEPTIONS
├── models/
│   ├── result.py              SearchResult
│   └── response.py            SearchResponse
├── providers/                 7 个 BaseSearchProvider 子类
│   ├── tavily.py              Tavily AI 搜索
│   ├── serpapi.py             SerpAPI Google 搜索
│   ├── bocha.py               博查 AI 搜索（中文友好）
│   ├── brave.py               Brave Search
│   ├── minimax.py             MiniMax MCP 搜索
│   ├── searxng.py             SearXNG 自托管
│   └── akshare_news.py        东方财富免费兜底
├── parsing/
│   ├── relative_dates.py      parse_relative_date（CN+EN 相对时间）
│   ├── absolute_dates.py      normalize_publish_date（ISO/RFC/CN/UNIX）
│   └── content.py             fetch_url_content（lazy newspaper3k）
└── tests/                     32 个测试（cache / filters / parsing / reporting / skeleton）

快速使用

from querybus import SearchProvider, DimensionSpec

# 默认：仅 Akshare 兜底 + 内存 TTL 缓存
provider = SearchProvider()

# 配置多 key：
provider = SearchProvider(
    tavily_keys=[...],
    serpapi_keys=[...],
    bocha_keys=[...],
    brave_keys=[...],
    minimax_keys=[...],
    searxng_base_urls=[...],
    default_max_age_days=3,
    cache=True,                  # True / False / 自定义 CacheProtocol
    cache_ttl_seconds=600,
)

# 通用 API
resp = provider.search("Apple Q4 earnings", max_results=5, days=7)
batch = provider.batch_search(["Apple", "Google", "Meta"])

# 通用情报报告（dimensions 由 caller 决定）
md = provider.format_intel_report(
    {"news": resp1, "analysis": resp2},
    subject="AAPL",
    dimension_specs=[
        DimensionSpec(key="news", label="📰 News"),
        DimensionSpec(key="analysis", label="📈 Analyst"),
    ],
)

# 逃生口
tavily = provider.provider("Tavily")
hourly = tavily.search(...)

DI cookbook

# 1) 默认：InMemoryTTLCache + Akshare 兜底
provider = SearchProvider()

# 2) 自定义缓存（满足 CacheProtocol：get/set 即可）
class RedisCacheAdapter:
    def get(self, key): ...
    def set(self, key, value, ttl_seconds): ...

provider = SearchProvider(tavily_keys=[...], cache=RedisCacheAdapter())

# 3) 完全无缓存（一次性脚本）
provider = SearchProvider(tavily_keys=[...], cache=False)

# 4) 完全自定义 provider 链（绕过默认 7 个内置）
from querybus.providers import TavilySearchProvider
provider = SearchProvider(providers=[TavilySearchProvider(api_keys=[...])])

与 stocklens 的边界

querybus 不知道「股票」这个概念：

不依赖 stocklens.* / tickbridge.* / api.* / bot.* / pulsefan.* / kvcache.*（CLAUDE.md 强制）
运行时依赖：httpx>=0.27 / tenacity>=8.0 / cachetools>=5.3.0
不读环境变量；所有配置走构造参数（DI）
不持有领域查询模板；调用方自己拼 query 字符串
filter_by_time_window / dedupe_results 已从 querybus.__init__ 公开导出，调用方无需 import querybus.core.filters

股票特化的逻辑全部留在 stocklens/search/stock_search_service.py：

在 stocklens 这一层	不在 querybus 这一层
`_is_foreign_stock` / `is_index_or_etf` 分类	querybus 不关心代码格式
`search_stock_news` / `search_stock_events`	querybus 只暴露通用 `search()`
`search_comprehensive_intel` 5 维度并发	querybus 提供 `format_intel_report` 渲染骨架
`search_stock_price_fallback`	querybus 不知道"数据源失败"
`STOCK_INTEL_DIMENSIONS`（emoji 标签）	querybus 接受任意 `DimensionSpec` 列表
`news_strategy_profile`（业务术语）	querybus 直接接 `default_max_age_days`

唯一对外类：SearchProvider

构造参数

参数	类型	说明
`tavily_keys` 等 6 个 `*_keys`	`List[str] \| None`	各 provider 的 API key 列表（多 key 自动负载均衡 + 错误计数）
`searxng_base_urls`	`List[str] \| None`	SearXNG 自托管实例 URL 列表
`default_max_age_days`	`int = 3`	`search()` 不传 `days` 时使用的窗口
`cache`	`bool \| CacheProtocol`	True=默认 InMemoryTTL；False=禁用；或自定义
`cache_ttl_seconds`	`int = 600`	默认 TTL
`providers`	`Sequence[BaseSearchProvider] \| None`	完全自定义的 provider 链（绕过内置工厂）
`include_akshare_news`	`bool = True`	是否在末尾追加免费兜底

关键方法

search(query, max_results=5, days=None, *, filter_results=True, log_scope="") -> SearchResponse
batch_search(queries, max_results=5, days=None, *, delay_between=0.0, filter_results=True) -> List[SearchResponse]
provider(name) -> BaseSearchProvider — escape hatch
format_intel_report(intel_results, *, subject, dimension_specs, ...) -> str — 静态方法
filter_by_time_window(response, *, days, max_results, log_scope="") -> SearchResponse — 静态方法
dedupe(responses) -> None — 静态方法（in-place 改动 results）
属性：is_available、provider_names

核心数据流

SearchProvider.search(query, days=7)
    │
    ▼
┌─────────────────┐  hit
│ InMemoryTTL     ├────────► return cached SearchResponse
│ Cache (L1)      │
└────────┬────────┘
         │ miss
         ▼
┌─────────────────┐
│ SearchManager   │  iterate providers in priority order
│ (failover)      │  Tavily → Bocha → Brave → SerpAPI → MiniMax → SearXNG → AkshareNews
└────────┬────────┘
         │ first non-empty filtered response wins
         ▼
┌─────────────────┐
│ filter_by_time_ │  hard-filter by published_date recency
│ window + dedupe │  + dedupe across dimensions (when called by caller)
└────────┬────────┘
         │
         ▼
   cache + return

测试覆盖

文件	测试
`test_skeleton.py`	4 个 — 顶层 import / SearchProvider 构造 / Akshare 默认开 / 关闭 / ABC 抽象性
`test_cache.py`	6 个 — get/set/expire/eviction/clear
`test_filters.py`	5 个 — 时间窗保留/丢弃/补全 + dedupe 跨 response
`test_date_parsing.py`	12 个 — 中英文相对时间 + ISO/RFC/CN/UNIX/strptime 兜底
`test_reporting.py`	4 个 — 通用 dimension specs / fallback / missing keys

合计 31 个独立测试；pytest querybus/tests/ -q 应全部通过。

与 tickbridge 的对照

维度	tickbridge	querybus
唯一对外类	`DataProvider`	`SearchProvider`
多源插件契约	`BaseFetcher`	`BaseSearchProvider`
默认缓存	`InMemoryTTLCache + SqliteTTLCache + TieredCache`	`InMemoryTTLCache`（搜索结果短时性，无需 L2）
默认存储	`SqliteRepository`（持久化 OHLCV）	无 — 搜索结果不需要持久化
失败模式	per-source 熔断冷却	per-key 错误计数 + key 轮换
单源领域插件	`DomainFetcher`（finviz / social_sentiment）	暂无（如未来需要可加）

注意事项（变更日志）

2026-05 修复 providers/brave.py 解析 bug：原 web_results = web_data.get("results", []) 中 web_data 从未定义；results.append(...) 中 results 也从未初始化。从 commit history 看是某次 reformat 把 # Parse search results 注释和 data = data.get('web', {}) 误合并到了一行。补回 web_data = data.get("web", {}) + results: list[SearchResult] = []。该 provider 在搜索 fall-over 链中的优先级靠后，所以 bug 一直没暴露。
httpx 迁移进度：retry helper / brave.py / searxng.py / minimax.py 已迁；bocha.py 还在 requests，详见 docs/CONTEXT.md 的 httpx 迁移债务表。
2026-05 M10 — provider 共享 helper：BaseSearchProvider 新增两个共享方法：
- _error_response(query, message, search_time=None) — 标准化空结果 SearchResponse 构造，避免每个 provider 在 4-5 处自己拼 SearchResponse(query=..., results=[], success=False, ...)。
- _extract_domain(url) — URL → 主机名（剥 www. 前缀），之前 tavily.py 与 brave.py 各自有几乎一模一样的实现。 Tavily 与 Brave 的内部错误返回点全部切到 self._error_response(...)；两者的 _extract_domain 静态方法已删除（继承基类）。新加 provider 直接 from querybus.contracts.base_provider import BaseSearchProvider，错误返回写一行即可。

Navigation Menu

querybus — 统一搜索/资讯层

产品愿景背景

角色

目录

快速使用

DI cookbook

与 stocklens 的边界

唯一对外类：SearchProvider

构造参数

关键方法

核心数据流

测试覆盖

与 tickbridge 的对照

注意事项（变更日志）

相关文档

FilesExpand file tree

querybus.md

Latest commit

History

querybus.md

File metadata and controls

querybus — 统一搜索/资讯层

产品愿景背景

角色

目录

快速使用

DI cookbook

与 stocklens 的边界

唯一对外类：SearchProvider

构造参数

关键方法

核心数据流

测试覆盖

与 tickbridge 的对照

注意事项（变更日志）

相关文档