stocklens/pipeline/ — 分析流水线编排

职责

核心调度模块。编排完整的股票分析流程：共享数据预取 → 数据预采集 → enrich + LLM 流水线并行 → 报告生成 → 推送。

文件清单

文件	职责
`orchestrator.py`	`StockAnalysisPipeline` 主编排器（仅装配 + 编排，业务步骤已抽到 `stages/`）
`stages/`	原子化的分析步骤子包（双路径共享，下表展开）
`shared_prefetch.py`	2026-05 新增：Phase 1 共享数据预取助手（`warm_us_market_env` / `fetch_us_macro_or_none` / `fetch_market_news`）；市场新闻 IO 委托 `stocklens.market.market_news`
`data_collector.py`	`DataCollector` 数据采集与预处理（共享 IO 池 `DC_IO`）
`stock_indicators.py`	`compute_trend_and_indicators()` — 趋势 + 多周期技术指标共用 helper
`us_data_enricher.py`	`enrich_us_hk_context()` — 美股/港股增强数据并行（线程池 `US_ENRICH`）
`analysis_flow.py`	`_PipelineHelpersMixin`（上下文增强、Agent 路径、辅助方法）
`push_handler.py`	`_PipelineNotifyMixin`（报告保存 + 推送调度）
`trading_calendar.py`	交易日历，判断市场是否开市（fail-open）
`batch_context.py`	2026-05 新增：批次数据对齐 — `BatchContext` / `compute_batch_as_of` / `validate_alignment`（详见下方"BatchContext"节）
`metrics.py`	流水线计时与统计

`stages/` 子包

把原本散落在 analyze_stock 与 _prepare_stock_context 中的重复逻辑提炼为独立、可单测的 stage 函数。每个 stage 自行兜底领域异常并降级返回，主调度器只在最外层捕获 DSAError / Exception。

模块	入口	职责
`realtime_stage.py`	`fetch_realtime_quote()`, `resolve_stock_name()`	Step 1：实时行情 + 股票名补全
`chip_stage.py`	`fetch_chip_distribution()`	Step 2：筹码分布（A 股专属，US/HK 跳过）
`fundamental_stage.py`	`fetch_fundamental_context()`, `save_fundamental_snapshot()`	Step 2.5：基本面聚合 + 快照保存
`agent_mode_check.py`	`should_use_agent()`	Step 3：Agent 模式判定（agent_mode / agent_skills 优先级）
`intel_stage.py`	`collect_news_context()`	Step 4 + 4.5：多维情报检索 + 大盘新闻 + Jin10 + Finviz + Social sentiment
`enrichment_stage.py`	`enrich_full_context()`	Step 6：context enhance + tech_indicators + RS vs SPY + US/HK enrich
`scoring_stage.py`	`run_quantitative_scoring()`	Step 7：18 量化模型评分 + 综合分 + IV/分析师快照
`metadata_stage.py`	`build_and_save_metadata_report()`	Step 7b：完整性校验 + 元数据报告生成 + 落地
`llm_stage.py`	`run_llm_and_persist()`	Step 8：LLM 调用 + 字段回填 + 历史保存 + 单股推送
`persistence_stage.py`	`persist_all_snapshots()`	2026-05 新增：Step 7a — 把 chip / tech indicators / 社交情绪 / 相关性 / OpenBB 机构数据 / 市场环境 / 大盘新闻写入 7 张快照表（详见 `storage.md`），best-effort 不阻断

核心类

`StockAnalysisPipeline`（orchestrator.py）

主编排器（987 行，从历史 1442 行降下来）。继承 _PipelineHelpersMixin 与 _PipelineNotifyMixin 两个 Mixin。

初始化签名

pipeline = StockAnalysisPipeline(
    config: Optional[Config] = None,
    source_message: Optional[BotMessage] = None,
    query_id: Optional[str] = None,
    query_source: Optional[str] = None,
    save_context_snapshot: Optional[bool] = None,
)
# self.max_workers = get_pool_workers("PIPELINE")  ← 来自 utils.concurrency

关键方法

公开 API（兼容外部调用方）

run(stock_codes, send_notification=True, metadata_only=False) -> List[AnalysisResult] — 完整批量分析入口
- 调用链：_prefetch_shared_data() → _batch_fetch_phase() → _pipeline_phase() → _send_notifications()
analyze_stock(code, report_type, query_id) -> Optional[AnalysisResult] — 单股完整分析（API/Bot 同步路径）
- 内部委托到 _collect_full_context() + Agent 路径或 run_llm_and_persist()
process_single_stock(code, single_stock_notify=False, ..., metadata_only=False) — 单股工作流
- 处理报告查重、metadata-only 短路、单股推送
- 被 task queue / API / CLI 调用
fetch_and_save_stock_data(code) / collect_stock_data(code) — 委托到 DataCollector

内部流水线 helpers

_collect_full_context(code, query_id, report_type) -> Optional[dict] — 双路径共享核心
- 串行调 9 个 stage：realtime → chip → fundamental → agent check → trend+indicators → news → enrich → scoring → metadata
- Agent 模式自动短路（不跑 news/enrich/scoring/metadata，直接返回供 _analyze_with_agent 使用）
- 返回结构化 dict，供 analyze_stock / _prepare_stock_context 各自后处理
_prepare_stock_context(code, ...) — Phase 3 enrich worker
- 报告查重 → fetch → metadata-only 短路 → _collect_full_context() → 返回 {needs_llm: bool, ...} 给 LLM 池
- Agent 模式在本 worker 同步执行（不进 LLM 池）
_run_stock_llm(ctx) — Phase 3 LLM worker
- 薄壳，直接转发到 stages.run_llm_and_persist()

Phase 子方法

_prefetch_shared_data(stock_codes) — Phase 1 共享数据预取（PREFETCH 池，默认 4）
- 实时行情批量预取（≥ 2 只，F5）+ 股票名预热
- 美股相关：_warm_market_env() + _warm_macro() + _fetch_market_news() 并行
_batch_fetch_phase(stock_codes) — Phase 2 数据预采集（DATA_FETCH 池）
_pipeline_phase(stock_codes, ...) — Phase 3 enrich + LLM 真流水线（F2，2026-05）
- 两个独立池：enrich_executor + llm_executor
- add_done_callback 触发：enrich 完成一只 → 立即向 LLM 池提交
- F2 改动：弃用 with enrich_executor: 的 hard barrier；主线程改为事件驱动 drain，enrich 仍在 in-flight 时已开始消费 LLM future。Phase 3 总耗时从 max(enrich_total, llm_total) 降到 max(enrich_max_one + llm_max_one, llm_total)——长尾股票不再阻塞其它股票的 LLM 收尾。
_run_metadata_only(code) — metadata-only 路径（采集 + 评分 + 报告，无 LLM）

流水线核心代码（F2 之后，简化）

# Phase 3：两个池真流水线（F2）
llm_executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="llm")
enrich_executor = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="enrich")
llm_futures: dict = {}
results: list = []
enrich_remaining = len(stock_codes)

def _on_enrich_done(future, code):
    nonlocal enrich_remaining
    ctx = future.result()
    if ctx and ctx.get("needs_llm"):
        f = llm_executor.submit(self._run_stock_llm, ctx)
        llm_futures[f] = code
    elif ctx and "result" in ctx:           # Agent 路径短路
        results.append(ctx["result"])
    enrich_remaining -= 1

# 提交 enrich 但**不等**它结束（关键 F2 改动）
for code in stock_codes:
    future = enrich_executor.submit(self._prepare_stock_context, code, ...)
    future.add_done_callback(lambda f, c=code: _on_enrich_done(f, c))
enrich_executor.shutdown(wait=False)        # 不阻塞主线程

# 事件驱动 drain：每完成一只 LLM 立刻消费，enrich 继续 in-flight
consumed = set()
while True:
    pending = [f for f in llm_futures if f not in consumed]
    if not pending and enrich_remaining <= 0:
        break
    if not pending:
        time.sleep(0.05); continue
    try:
        done = next(as_completed(pending, timeout=1.0))
    except FuturesTimeoutError:
        continue
    consumed.add(done)
    if (r := done.result()):
        results.append(r)

回归测试：tests/unit/pipeline/test_pipeline_phase_drain.py（5 用例，含 test_long_tail_enrich_does_not_block_fast_llm 直接断言 F2 行为）。

重构收益（2026-05 重构）

维度	之前	之后
`orchestrator.py` 行数	1442	987
双路径重复代码	`analyze_stock` 与 `_prepare_stock_context` 几乎逐行复制	共享 `_collect_full_context()`
业务步骤可单测性	无（嵌在 1400 行类里）	9 个独立 stage 函数，可独立 mock
异常处理	32 处 `except Exception` 散落主类	各 stage 自行兜底领域异常，主类仅最外层 2 处

`DataCollector`（data_collector.py）

数据采集与预处理。共享内部 IO 线程池（DC_IO，避免每只股票重建池）。

关键方法

fetch_and_save_stock_data(code) -> (success, error) — 采集并保存日线/周线 + 60m/月/季/年 + 当日行情快照
collect_stock_data(code) -> dict — 元数据模式下的完整采集（不调 LLM）
augment_historical_with_realtime(df, quote, code) -> DataFrame — 盘中将实时价格追加到历史
enhance_context(context, realtime, chip, trend, name, fundamental) -> dict — 上下文增强
validate_metadata(code, name, ec) -> List[str] — 元数据完整性校验（静态）

多周期 K 线同步（2026-05 新增）

fetch_and_save_stock_data 在保存日线后顺序触发：

_sync_weekly_data(code) — 从日线 resample 周线（已有）
_sync_multi_period_bars(code) — 新流程：
- 60m 直接拉取: fetcher_manager.fetch_bars(code, '60m', count=500) → stock_bars（仅雪球支持）
- 月线 resample: 日线 → pd.resample('ME') → stock_bars(interval='1mo')
- 季线 resample: pd.resample('QE') → stock_bars(interval='1q')
- 年线 resample: pd.resample('YE') → stock_bars(interval='1y')
_save_quote_snapshot_if_present(code) — 实时行情面板落库 stock_quote_snapshot（一天一行/股，重跑覆盖）

每一步独立 try/except，失败仅 warning 不阻断主流程。月/季/年都从本地日线生成，零 API 配额成本。

`compute_trend_and_indicators()`（stock_indicators.py）

抽出的共用纯函数，被双路径与单股路径共用。

trend_result, df, tech_indicators = compute_trend_and_indicators(
    code=code, stock_name=stock_name,
    realtime_quote=realtime_quote,
    stock_repo=self.stock_repo,
    data_collector=self.data_collector,
    trend_analyzer=self.trend_analyzer,
    fetcher_manager=self.fetcher_manager,
)

内部：

趋势分析 1 年日线
周线 2 年（DB 优先 → 远程 fallback）
月线 3 年（重采样 resample('ME')）
compute_tech_indicators() 多周期斐波那契/ATR/OBV

`enrich_us_hk_context()`（us_data_enricher.py）

enrich_us_hk_context(
    code, stock_name, enhanced_context,
    fetcher_manager, fundamental_repo,
    mode="full" | "metadata",
)

并行获取：宏观注入 / Finviz 前瞻 / 同行对比 / PE 百分位 / 板块 ETF 表现。线程池 US_ENRICH。

`TradingCalendar`（trading_calendar.py）

依赖 exchange-calendars（可选）。未安装时所有市场视为开市（fail-open）。

get_market_for_stock(code) → "cn" / "hk" / "us" / None
is_market_open(market, date) → bool
get_open_markets_today() → Set
compute_effective_region(...)

2026-05 修复：get_market_for_stock 原来引用了 tickbridge.codes.normalize.is_hk_stock_code（实际不存在；模块只导出 _is_hk_market），未走到该 codepath 时一直没暴露。BatchContext 接入后立刻触发 ImportError。修复：lazy import _is_hk_market，tickbridge 不可用时退到 HK* 前缀 fallback。

`BatchContext` — 批次数据对齐（batch_context.py，2026-05 新增）

问题：盘后复盘工具一次 batch 拉的数据必须对齐到同一交易日的盘后收盘。原实现里 date.today() 散落各处，跨市场（CN+US 混批 / 服务器时区与目标市场时区不一致）容易导致 SPY 是周五、个股 K 线是周四、宏观是上周。

解决方案：stocklens.pipeline.batch_context 单一来源。

名称	类型	用途
`BatchContext`	frozen dataclass	`as_of_date` / `markets` / `fail_open` / `notes`；`as_of_iso` 给字符串持久化用
`compute_batch_as_of(markets, reference=None)`	factory	算 batch as-of：per-market 取本地时区今天 → 用 `exchange_calendars` 走回到最后交易日 → 跨市场取 min
`AlignmentResult`	dataclass	`aligned` / `actual_date` / `expected_date` / `drift_days` / `severity`
`validate_alignment(code, actual_date, ctx)`	function	比较 df 最后一根 bar 与 `ctx.as_of_date`；severity = `ok` / `warn` (≤3d) / `error` (>3d)

算法关键点：

跨市场取 min：CN+US 混批时 = 「今天 CN 盘后」与「昨天 US 盘后」的较早者，保证两边都已收盘。
fail-open：exchange_calendars 未装 / 未识别市场 → 用 reference 原值 + fail_open=True。validate_alignment 在 fail-open 下把 1 日漂移视为对齐（as_of 本身就是近似）。
单股入口兜底：StockAnalysisPipeline._ensure_batch_ctx(codes) 在 analyze_stock / process_single_stock / _run_metadata_only 三个非批次入口 lazy 建一只股票的 BatchContext。

接入点（修改了哪些方法）：

位置	改动
`_BatchOrchestratorMixin.run()` / `async_run()`	入口先 `compute_batch_as_of(markets)` 并打 `[batch_ctx] BatchContext(...)` 日志
`StockAnalysisPipeline.__init__`	`self.batch_ctx: Optional[BatchContext] = None`
`_ensure_batch_ctx(codes)`	lazy 构建 + 复用
`_collect_full_context`	戳 `enhanced_context["as_of_date"]` / `["batch_markets"]` / `["batch_started_at"]`；`persist_all_snapshots(snapshot_date=ctx.as_of_date)`
`_validate_data_alignment(code, df, ec, ctx)`	解析日线最后一根 bar，写 `enhanced_context["data_alignment"]`，漂移 `WARNING` 但不阻断

enhanced_context 新增字段：

ec["as_of_date"]          # "2024-05-13"
ec["batch_markets"]       # ["cn", "us"]
ec["batch_started_at"]    # ISO datetime
ec["data_alignment"] = {
    "aligned": False,
    "actual_date": "2024-05-10",
    "expected_date": "2024-05-13",
    "drift_days": 3,
    "severity": "warn",   # ok / warn / error
}

已知 follow-up：stocklens/pipeline/data_collector.py 内 10+ 处 date.today() 仍按调用时刻取，未强制收口到 ctx.as_of_date。当前依赖采集层产出的 df 真实日期 + _validate_data_alignment 兜底告警。后续若要让 cache key 用 as_of_date，需把 ctx 沿调用链传下去。

测试：tests/unit/pipeline/test_batch_context.py（19 用例）：

compute_batch_as_of：单 US 工作日 / 周六周日 rollback / mixed CN+US / 美国独立日 / 空 markets / 未知市场 / iso 格式（8）
validate_alignment：对齐 / 数据更新 / 1d/3d warn / 4d error / 无数据 / fail-open 1d 容忍 / fail-open 拒 2d（8）
BatchContext：frozen / __str__ / as_of_iso（3）

`_PipelineHelpersMixin`（analysis_flow.py）

提供：_enhance_context()、_analyze_with_agent()、_agent_result_to_analysis_result()、_build_context_snapshot()、_safe_to_dict()、_resolve_query_source()、_build_query_context() 等 helper。

`_PipelineNotifyMixin`（push_handler.py）

单股报告保存
日报汇总保存
调度 NotificationDispatcher

异常处理约定

每个 stage 函数自行处理领域异常，按以下原则：

异常类型	处理
`DSAError` 子类（`DataSourceError` / `LLMError` / `StorageWriteError` / `SearchError` / `NotificationError` / …）	在 stage 内捕获，记 `logger.warning`，降级返回 `None` 或失败结构
`Exception`（兜底）	在 stage 内最外层捕获，记 `logger.warning/exception`，降级返回；禁止 silent `pass`
致命未预期异常	主调度器最外层（`analyze_stock` / `_prepare_stock_context`）捕获 `DSAError` + `Exception` 并 `logger.exception`，返回 `None`

依赖关系

依赖：tickbridge、config、storage、repositories、analyzer、services.scoring、services.tech_indicators_service、search、market、notification、utils.concurrency、contracts、exceptions
被依赖：main.py、api/v1/endpoints/*（通过 TaskQueue 异步调用）

并发资源

池	用途	默认	调节
`PIPELINE`	enrich + LLM 池	cpu_count	`STOCKLENS_PIPELINE_WORKERS`
`PREFETCH`	共享数据预取	4	`STOCKLENS_PREFETCH_WORKERS`
`DATA_FETCH`	Phase 2	cpu_count	`STOCKLENS_DATA_FETCH_WORKERS`
`DC_IO`	DataCollector 内部	cpu_count	`STOCKLENS_DC_IO_WORKERS`
`US_ENRICH`	US/HK 增强	cpu_count	`STOCKLENS_US_ENRICH_WORKERS`

配置项

变量	默认	说明
`PREFETCH_REALTIME_QUOTES`	true	预取实时行情
`REPORT_INTEGRITY_ENABLED`	true	元数据完整性校验
`REPORT_INTEGRITY_RETRY`	1	校验失败重试次数
`NEWS_MAX_AGE_DAYS`	3	新闻最大时效
`BACKTEST_ENABLED`	false	分析后是否自动回测

注意事项

同一天重复运行会跳过已有报告的股票（_prepare_stock_context 开头检查）
删除 stocklens/reports/{YYYYMMDD}/ 目录可强制重跑
所有线程池并发数都通过 get_pool_workers(name) 取，可用环境变量调节
批量分析采用流水线架构：enrichment 池和 LLM 池同时运行，enrichment 完成一只就立即提交 LLM
单股分析（API/Bot 调用）走 analyze_stock → _collect_full_context 共享路径
analyze_stock 与 _prepare_stock_context 共享 _collect_full_context()，业务步骤完全等价
趋势分析依赖 tickbridge.codes.us_index 导出的 is_us_stock_code 等函数
exchange-calendars 未安装时所有市场视为开市（fail-open）
月线重采样使用 pandas resample('ME')，OHLC 聚合（'M' 已弃用）

性能优化（2026-05）

历经一轮系统性性能审计，覆盖以下高 ROI 修复：

数据流去重

SPY 复用：enrichment_stage._compute_rs_vs_spy 改读 kvcache spy_qqq 命名空间，与 compute_correlation 共享 SPY/QQQ 历史数据，消除每只美股都重新拉一次 akshare 的冗余下载。
月线读取：stock_indicators._load_monthly_df 优先从 stock_bars（Phase 2 已写）读取，其次复用调用方已加载的 1 年日线 resample，仅在两者都不可用时才重新查 3 年日线。
context 复用：orchestrator Step 6 用新方法 stock_repo.build_context_from_df 从已加载的 1 年 df 派生 today / yesterday / daily_history / ma_status，省去第三次 get_latest(15) 查询。失败时自动回退到 get_analysis_context。
macro 单写：宏观指标持久化从每只股票循环里移到 Phase 1 _warm_macro 一次性写入，消除 O(stocks × 12) 次重复 upsert。

并发与池化

_collect_result 重叠化：us_data_enricher.enrich_us_hk_context 把结果收集移入 with pool: 块内，让 macro/mv_trend/PE 等快任务的处理与 OpenBB 等慢任务的等待重叠。
DataCollector._io_pool 懒初始化：池在首次调用 _get_io_pool() 时才创建，并通过 atexit.register 注册 shutdown，避免长跑 API/Bot 进程累积空闲线程。
industry_service 并行：fetch_industry_snapshot 中的 sections 4-9（insider trades / analyst ratings / description / peer tickers / etf holders / news）从 6 次顺序请求改为 ThreadPoolExecutor 4 worker 并行，预计单股节省 4-8 秒墙钟时间。

数据库优化

save_news_intel UPSERT：news_repo 改用 INSERT ... ON CONFLICT(url) DO UPDATE，替换原先 N+1 的 select-then-insert + savepoint 模式（30 stocks × 30 items ≈ 1800 round-trips → 30 statements）。
count_by_code 改 SQL COUNT()：analysis_repo.count_by_code 不再加载 ORM 对象到内存，直接 SELECT COUNT(*) WHERE code AND created_at >= cutoff。
DataFrame 向量化写入：stock_repo.{save_dataframe, save_weekly_data, save_bars} 用 _normalize_date_column + _df_to_records_with_meta 替换 iterrows()，10-50× 提速。
新增复合索引：
- MacroSnapshot: ix_macro_indicator_date(indicator, data_date) 优化日期范围扫描。
- LLMUsage: ix_llm_usage_model_time(model, called_at) + ix_llm_usage_code_time(stock_code, called_at) 匹配典型审计查询模式。

LLM 与外部 IO

Anthropic prompt caching：llm_analyzer._call_litellm 检测到 anthropic/ 系列模型时，在 system 消息上注入 cache_control: {type: "ephemeral"}。LiteLLM 透传该标记；非 Anthropic provider 静默忽略。usage 字段额外暴露 cache_creation_input_tokens / cache_read_input_tokens 以便审计。预期减少 ~90% 系统提示符重复输入 token 成本。
HTTP Session/Client 复用：
- tickbridge: akshare.py（Sina + Tencent 实时行情）/ twelvedata.py / tushare.py / domain_sources/social_sentiment.py 引入模块级 requests.Session 单例。
- stocklens: services/scoring/_cboe_fetcher.py 改用模块级 httpx.Client。
- 每次调用消除 50-200ms 的 TCP+TLS 握手开销。

计算向量化

OBV：quantcore.indicators.tech.compute_tech_indicators 中的 OBV 改为 np.cumsum(np.sign(diff) * volume)，30-100× 提速。
Swing Points：detect_swing_points 改用 Series.rolling(window, center=True).max/min，从 O(n × window) 降为 O(n)。
Volume rolling：StockTrendAnalyzer._calc_volume 预先用 rolling(5).mean().shift(1) 一次算好 5 日均量基线，避免循环里 lookback 次重复计算。
Max Pain (期权情绪)：quantcore.scoring.calc_options_sentiment 用 numpy 广播向量化 Max Pain 计算，从 O(M²) Python 双循环降为 O(M²) numpy 矢量（实测 100× 提速；对 200+ 行权价的 SPY/QQQ/AAPL 期权链尤其显著）。

缓存集中化

finviz 五个 ad-hoc 缓存迁移到 kvcache：tickbridge/domain_sources/finviz.py 中的 _page_cache / _cache / _news_cache / _peer_cache / _sector_etf_cache 全部替换为 kvcache.get_manager().namespace("finviz_*")。新增 namespace：finviz_page (1800s) / finviz_forward (3600s) / finviz_news (1800s) / finviz_peer (3600s) / finviz_sector_etf (3600s)。 fundamental_ctx namespace 也已注册（120s），但 fundamental_mixin 因紧耦合 self-state 暂未迁移。
etf_quote TTL 1s → 5s：原 1s 短于多数非 dashboard 调用方的请求延迟，等于零缓存收益。 Dashboard SSE 自身 polling 间隔 ≥2s，5s 仍保持实时感。

热点函数 lru_cache

tickbridge.codes.us_index.is_us_stock_code 加 lru_cache(maxsize=2048) —— 每只股票每次分析会被调用十几次，且参数空间小。
quantcore.scoring.types.classify_industry 加 lru_cache(maxsize=512) —— pipeline 一次运行 industry/sector pair 基数极小。
新增/修改 stage 时同步更新本文件的 stages/ 子包表

M2 + M7 — sync/async dedup + StageContext（2026-05）

`stages/llm_stage.py` 抽出 `_postprocess_and_persist`

run_llm_and_persist 与 arun_llm_and_persist 此前重复了完整的「写 query_id → 取价格 → 填筹码 → enforce_trend_status → enforce_score_band → 保存历史 → 推送」~80 行后处理逻辑。M2 把这段抽到 _postprocess_and_persist(result, *, log_tag, ...)，sync/async 两个入口都在 try 块里调它（区别只剩 analyzer.analyze vs await analyzer.aanalyze）。

`stages/stage_context.py` — 新增 `StageContext` dataclass（M7）

把过去到处传递的 code / stock_name / query_id / fetcher_manager / config / batch_ctx 6 个 kwargs 收口到一个 frozen dataclass：

@dataclass(frozen=True)
class StageContext:
    code: str
    stock_name: str
    query_id: str
    fetcher_manager: Any = None
    config: Any = None
    batch_ctx: Any = None
    extras: Dict[str, Any] = field(default_factory=dict)

提供 StageContext.create(...) 构造方法和 with_(**overrides) 派生方法（支持 extras_patch= 增量更新 extras）。

这是过渡层：现有 stage 函数的 keyword-only 签名保留，调用方零改动；新写的 stage 推荐 def my_stage(ctx: StageContext, **extras): ...。后续可逐个迁移现有 stage —— 不强制一次性大 diff。

`pipeline/push_handler.py` — channel dispatch 拆分（M4）

详见 docs/modules/notification.md 的 M4 条目。_send_notifications 从 180 行 if/elif ladder 拆成 4 个聚焦 helper（_convert_to_image_with_hint / _push_wechat / _push_email_groups / _push_other_channel）。Email 的分组路由逻辑保留在 _push_email_groups 内部。

F7-F10 — 全部 stage 接 StageContext + orchestrator 一次构造（2026-05）

M7 引入了 StageContext dataclass 但保留 stage 旧签名作为过渡层。F7-F9 把过渡走完：每个 stage 函数都暴露 *_ctx 兄弟版本，与旧版本共享 _do_* 内部实现。F10 把 orchestrator 切到 ctx 变体。

Stage 公开 API（双版本并存）

旧签名（保留，向后兼容）	StageContext 版本（推荐）
`fetch_realtime_quote(fetcher_manager, code, stock_name)`	`fetch_realtime_quote_ctx(ctx)`
`fetch_chip_distribution(fetcher_manager, code, stock_name)`	`fetch_chip_distribution_ctx(ctx)`
`fetch_fundamental_context(fetcher_manager, code, stock_name, budget_seconds=...)`	`fetch_fundamental_context_ctx(ctx, *, budget_seconds=...)`
`save_fundamental_snapshot(fundamental_repo, query_id, code, stock_name, payload)`	`save_fundamental_snapshot_ctx(ctx, *, fundamental_repo, payload)`
`should_use_agent(config, stock_name, code)`	`should_use_agent_ctx(ctx)`
`collect_news_context(*, code, stock_name, query_id, search_service, ...)`	`collect_news_context_ctx(ctx, *, search_service, ...)`
`enrich_full_context(*, code, stock_name, ...)`	`enrich_full_context_ctx(ctx, *, ...)`
`run_quantitative_scoring(*, code, stock_name, enhanced_context, db)`	`run_quantitative_scoring_ctx(ctx, *, enhanced_context, db)`
`build_and_save_metadata_report(*, code, stock_name, enhanced_context, news_context, notifier)`	`build_and_save_metadata_report_ctx(ctx, *, enhanced_context, news_context, notifier)`
`run_llm_and_persist(*, code, stock_name, query_id, ...)`	`run_llm_and_persist_ctx(ctx, *, ...)`
`arun_llm_and_persist(*, code, stock_name, query_id, ...)`	`arun_llm_and_persist_ctx(ctx, *, ...)`

每对函数共享 _do_* 内部实现，行为可证明一致。回归保护：tests/unit/stages/test_stage_context_variants.py 14 用例显式断言每个 ctx 变体与旧版本结果相同。

Orchestrator 改造

StockAnalysisPipeline._collect_full_context：开头一次性构造 stage_ctx = StageContext.create(code=..., stock_name=..., query_id=..., fetcher_manager=..., config=...)；实时行情之后通过 stage_ctx = stage_ctx.with_(stock_name=resolved_name) 派生新 ctx；后续 9 个 stage 全部通过 *_ctx(stage_ctx, ...) 调用。

_run_metadata_only / _run_stock_llm / _arun_stock_llm / analyze_stock 的 LLM 路径同样改为构造 StageContext 后调 *_ctx。run_llm_and_persist 的旧名仍然导出，但在 orchestrator 内部已无引用。

新增跨切关注点（如 tracing span / request_id / batch metadata）只需在 StageContext 加字段，stage 逐步消费即可——不需要改 11 个 stage 的签名。

H4 — `_collect_full_context` 拆解（2026-05）

_collect_full_context 之前 180 行单方法承担 7 件事（实时行情 / 筹码 / 基本面 / agent 决策 / 趋势 / 新闻 / 增强 / 持久化 / 元数据报告）。重写为 50 行编排器调 5 个聚焦 helper：

Helper	职责
`_collect_prelude`	Phase A：实时行情 + 筹码分布 + 基本面聚合 + agent-mode 决策。返回 dict 含 `stage_ctx` / `stock_name` / `realtime_quote` / `chip_data` / `fundamental_context` / `use_agent`。
`_build_agent_context`（@staticmethod）	Phase C：Agent 路径的 early-return dict。
`_build_llm_context`	Phase D：增强 + 评分 + 持久化 + 元数据报告（仅 LLM 路径走）。
`_build_base_context`	选择最便宜的有效 base context（已加载 df 优先，否则 SQL 回退）。
`_stamp_batch_metadata`	在 `enhanced_context` 上盖 batch as-of-date / markets / start time + 校验数据对齐。返回 `BatchContext` 给持久化用。
`_persist_all_snapshots`	调 `persist_all_snapshots` + 标记 `_market_news_persisted` 防止下一只股票重复 dedup。

_collect_full_context 自身现在是 50 行，从上往下读 = Phase A → B → C → D。每个 sub-method 有自己的 docstring + 可独立单测。新加 stage 的 modus operandi：写一个 method + 在主编排器加一行调用。

FilesExpand file tree

pipeline.md

Latest commit

History

pipeline.md

File metadata and controls

stocklens/pipeline/ — 分析流水线编排

职责

文件清单

stages/ 子包

核心类

StockAnalysisPipeline（orchestrator.py）

初始化签名

关键方法

流水线核心代码（F2 之后，简化）

重构收益（2026-05 重构）

DataCollector（data_collector.py）

关键方法

多周期 K 线同步（2026-05 新增）

compute_trend_and_indicators()（stock_indicators.py）

enrich_us_hk_context()（us_data_enricher.py）

TradingCalendar（trading_calendar.py）

BatchContext — 批次数据对齐（batch_context.py，2026-05 新增）

_PipelineHelpersMixin（analysis_flow.py）

_PipelineNotifyMixin（push_handler.py）

异常处理约定

依赖关系

并发资源

配置项

注意事项

性能优化（2026-05）

数据流去重

并发与池化

数据库优化

LLM 与外部 IO

计算向量化

缓存集中化

热点函数 lru_cache

M2 + M7 — sync/async dedup + StageContext（2026-05）

stages/llm_stage.py 抽出 _postprocess_and_persist

stages/stage_context.py — 新增 StageContext dataclass（M7）

pipeline/push_handler.py — channel dispatch 拆分（M4）

F7-F10 — 全部 stage 接 StageContext + orchestrator 一次构造（2026-05）

Stage 公开 API（双版本并存）

Orchestrator 改造

H4 — _collect_full_context 拆解（2026-05）

`stages/` 子包

`StockAnalysisPipeline`（orchestrator.py）

`DataCollector`（data_collector.py）

`compute_trend_and_indicators()`（stock_indicators.py）

`enrich_us_hk_context()`（us_data_enricher.py）

`TradingCalendar`（trading_calendar.py）

`BatchContext` — 批次数据对齐（batch_context.py，2026-05 新增）

`_PipelineHelpersMixin`（analysis_flow.py）

`_PipelineNotifyMixin`（push_handler.py）

`stages/llm_stage.py` 抽出 `_postprocess_and_persist`

`stages/stage_context.py` — 新增 `StageContext` dataclass（M7）

`pipeline/push_handler.py` — channel dispatch 拆分（M4）

H4 — `_collect_full_context` 拆解（2026-05）