feat(phase1): converge multi-strategy evidence contract (#2031)

Introduce the StrategyEngine facade and centralize deterministic opinion partitioning, aggregation, synthesis, and signal normalization.

Keep invalid opinions in diagnostics, protect synthesis ownership across fallback paths, and add localized strategy rendering for notification, history, Markdown, and WeChat outputs.

Harden legacy and malformed strategy_synthesis payload handling through shared renderer helpers, with public-entry regression coverage and an updated Phase 1 contract.

Co-authored-by: zhulinsen <42829555+ZhuLinsen@users.noreply.github.com>
This commit is contained in:
xushengasd
2026-07-19 18:55:42 +08:00
committed by GitHub
parent 748dba5074
commit 5d98d6d787
20 changed files with 3165 additions and 58 deletions

View File

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
> For user-friendly release highlights, see the [GitHub Releases](https://github.com/ZhuLinsen/daily_stock_analysis/releases) page. > For user-friendly release highlights, see the [GitHub Releases](https://github.com/ZhuLinsen/daily_stock_analysis/releases) page.
## [Unreleased] ## [Unreleased]
- [修复] 多策略综合器语义收敛:修复 Signal 枚举输入被误判为 invalid、缺失 signal 被静默伪装为有效 hold、opinion_count 错误包含 invalid opinions、deterministic synthesis 可被 LLM dashboard 覆盖等问题;新增并收敛 12 个 Phase 1 语义回归测试。
- [修复] 桌面与 Docker 发布显式安装 `orjson`,桌面 PyInstaller 产物同时冻结并执行运行时导入探针,避免 LiteLLM 调用时报 `No module named 'orjson'` - [修复] 桌面与 Docker 发布显式安装 `orjson`,桌面 PyInstaller 产物同时冻结并执行运行时导入探针,避免 LiteLLM 调用时报 `No module named 'orjson'`
- [改进] 个股报告不再单独展示“题材主线与个股位置”卡片,相关市场结构数据仍保留在分析上下文、模型 Prompt 与决策信号提取链路中。 - [改进] 个股报告不再单独展示“题材主线与个股位置”卡片,相关市场结构数据仍保留在分析上下文、模型 Prompt 与决策信号提取链路中。
- [改进] 通知推送与完整 Markdown/微信报告不再重复附加“AI 决策信号”摘要DecisionSignal 的存储、告警和 Web AI 建议页保持不变。 - [改进] 通知推送与完整 Markdown/微信报告不再重复附加“AI 决策信号”摘要DecisionSignal 的存储、告警和 Web AI 建议页保持不变。
@@ -19,6 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
- [修复] `codex_cli` 普通分析显式固定无人值守批准策略与只读沙箱,避免新版 Codex 在非交互任务中因请求人工批准而中断。 - [修复] `codex_cli` 普通分析显式固定无人值守批准策略与只读沙箱,避免新版 Codex 在非交互任务中因请求人工批准而中断。
- [新功能] Web AI 建议页支持确认保存基于历史报告快照重算的决策风格信号,以 created/existing/refreshed 区分新建、原样复用和既有记录续期或维度补齐,复用 profile-aware 去重与失效语义,将历史信号的创建时间、有效期和相反信号失效顺序锚定来源报告时间,并提供可审计 guardrail 提示与阻断。 - [新功能] Web AI 建议页支持确认保存基于历史报告快照重算的决策风格信号,以 created/existing/refreshed 区分新建、原样复用和既有记录续期或维度补齐,复用 profile-aware 去重与失效语义,将历史信号的创建时间、有效期和相反信号失效顺序锚定来源报告时间,并提供可审计 guardrail 提示与阻断。
- [新功能] 多策略观点结构化输出第一版:新增策略观点标准化、基础冲突检测与聚合 metadata作为 #1964 的阶段性基础契约本次不声明完成并发执行、24 策略完整调度 MVP 或前端完整多语言展示。
- [修复] 多策略综合报告统一兼容历史与外部 dashboard 的宽松字段形状避免非法计数、非字典综合块或异常策略列表导致通知、微信、Jinja 与历史 Markdown 渲染失败。
- [修复] MiniMax 分析与渠道 JSON 测试仅提取最终文本块,避免推理内容与 JSON 拼接后导致结果无法解析和持久化。 - [修复] MiniMax 分析与渠道 JSON 测试仅提取最终文本块,避免推理内容与 JSON 拼接后导致结果无法解析和持久化。
<!-- 新条目格式:- [类型] 描述(类型取值:新功能/改进/修复/文档/测试/chore--> <!-- 新条目格式:- [类型] 描述(类型取值:新功能/改进/修复/文档/测试/chore-->
<!-- 每条独立一行追加到本段末尾,无需分类标题,合并时冲突最小 --> <!-- 每条独立一行追加到本段末尾,无需分类标题,合并时冲突最小 -->

View File

@@ -0,0 +1,331 @@
# 多策略投资建议契约Baseline 语义、Phase 1 收敛、Phase 2/3/4 边界
本页是 Issue #1964「多策略投资建议」的专题文档,用于记录 2 个及以上策略/技能skill观点在系统内的**语义收敛边界**有效证据集合、无效观点隔离、阵营分组、共识度、跨消费面一致性。Baseline 负责契约边界和现状盘点Phase 1 只在 Baseline 契约内完成有效证据集合分拣、`strategy_synthesis` 确定性合成、DecisionAgent prompt 收敛、四条 renderer 一致性以及 E2E 反例覆盖Phase 2 只在 Phase 1 契约下新增 24 策略并发调度与阶段调度Phase 3 只在 Phase 2 之上补前端多语言完整展示Phase 4 只在同一 `CONTRACT_VERSION = "1.0"` 内补权重回测反馈闭环。Baseline 的所有约束对后续 Phase 均永久生效Phase N 不得静默降级 Baseline 中已经写死的边界。
## 术语与边界
当前仓库里有多种名为 opinion / signal / consensus / synthesis 的数据面Baseline 必须先消歧,避免把现有运行时结构误写成未来 phase。
| 术语 | 当前含义 | 当前主要消费方 | Baseline 边界 |
| --- | --- | --- | --- |
| `AgentOpinion` | `src/agent/protocols.py` 中所有 Agent含 SkillAgent、TechnicalAgent、IntelAgent、RiskAgent、DecisionAgent产出的观点数据类`agent_name` / `signal` / `confidence` / `reasoning` / `key_levels` / `raw_data`。 | Orchestrator、Aggregator、DecisionAgent、Disagreement、Renderer | 记录为原始观点承载体Baseline 不新增字段,也不把 `AgentOpinion` 分裂成两类。 |
| `StrategyOpinion` | `src/agent/protocols.py` 中的内部规范化视图,含 `skill_id` / `signal` / `original_signal` / `invalid_signal`;只在 Aggregator/Synthesizer 内部使用。 | `SkillAggregator``ConflictDetector``StrategySynthesizer` | 记录为内部计算的规范化视图,不进入 `ctx.opinions`、不进入公共 payload、不进入 DecisionAgent prompt。 |
| Signal / Canonical Signal | 交易信号规范化标签Canonical 取值仅限 `strong_buy` / `buy` / `hold` / `sell` / `strong_sell` 五个小写字符串。 | 全链路 | 记录为下游所有计算的唯一允许输入形式;大写别名、`"strong buy"`、Signal 枚举原值都必须先经 `normalize_strategy_signal()` 转成 canonical 再参与计算。 |
| Valid Opinion / Invalid Opinion | 通过 `is_valid_strategy_signal(signal) == True` 且未标记 `invalid_signal=True` 的观点为 Valid其余为 Invalid。 | Orchestrator 分拣、Aggregator、DecisionAgent | 记录为契约层的合法/非法判定Baseline 只定义判定函数与语义,不预设分拣位置。 |
| Evidence Chain | 进入 DecisionAgent prompt 与 `strategy_synthesis` 数值计算的**有效观点集合**。 | DecisionAgent、Aggregator | 记录为决策输入面Baseline 规定 Evidence Chain 只由 Valid Opinion 组成Invalid 不得混入。 |
| Diagnostics | 无效观点的诊断收纳位,仅供日志、调试、用户可见的“另有 N 个策略解析失败”计数使用。 | Renderer 展示、日志 | 记录为诊断面Baseline 规定 Invalid 必须落到 Diagnostics不得被静默转成 `hold` 混入 Evidence Chain。 |
| `strategy_synthesis` | `dashboard.strategy_synthesis` 顶层 payload`final_signal` / `consensus_level` / `conflict_severity` / `supporting_skills` / `opposing_skills` / `summary_params`。 | Markdown、WeChat、Notification、History 四条 renderer | 记录为公共低敏 payloadBaseline 规定该 payload 是**唯一权威合成来源**LLM dashboard 不得反向覆盖。 |
| `disagreement_summary` | `ctx.meta["agent_disagreement_summary"]`,低敏跨 Agent 分歧摘要,来自 `build_agent_disagreement_summary()`。 | DecisionAgent prompt、日志 | 记录为决策路径提示面Baseline 规定只从 Valid Opinion 建桶Invalid 不得进入 `bullish_agents` / `bearish_agents` / `neutral_agents`。 |
| Consensus Level | `strategy_synthesis.consensus_level`,取值 `high` / `medium` / `low` / `insufficient`。 | Renderer 展示、Aggregator 内部判定 | 记录为共识度枚举Baseline 规定 ≤ 1 valid 或 `sum(confidence) == 0` 时强制 `insufficient`,不得输出 `high`。 |
## Baseline 范围与非目标
Baseline 的目标是让 Phase 1/2/3/4 都基于同一份语义契约设计运行时改动,而不是每一轮 PR 重新定义"有效观点"、"共识"、"支持方"。
- Baseline 覆盖 SkillAgent → Orchestrator → Aggregator → Synthesizer → DecisionAgent → Disagreement → Renderer 七条消费面的语义收敛边界。
- Baseline 固定 Canonical Signal 枚举、Valid/Invalid 判定函数、Evidence Chain / Diagnostics 分离原则、动态二分阵营语义、共识门槛梯度、`strategy_synthesis` payload schema、不变量清单和反例矩阵Phase 1 是这些边界的第一版代码化实现。
- Baseline 不引入并发调度、不引入前端多语言完整展示、不引入权重回测反馈;这些留给 Phase 2/3/4。
- Baseline 不改变现有 `AgentOpinion` 字段、不新增数据库字段、不改变 API 返回结构(`strategy_synthesis` 已在此前 PR 加入)、不新增配置项。
- Baseline 不把契约扩展成通用 opinion registry`AgentOpinion` 结构由现有代码维护,本契约只规范其**语义处置流程**。
## Baseline 内部契约
### Canonical Signal 与 Valid 判定
Canonical Signal 是 Baseline 允许的**唯一评分/加权/分组输入形式**。规范化入口是 `src/agent/protocols.py` 中的两个函数:
- `normalize_strategy_signal(signal)` 返回 `(canonical, invalid, original)` 三元组。它接受 `Signal` 枚举、大小写字符串、`"strong buy"` / `"strong-buy"` 别名,统一映射到 canonical 集合。无法映射时 `invalid=True``canonical` 退化为 `default`(默认 `"hold"`)但**必须**配合 `invalid=True` 一并传递到下游,不得被单独使用。
- `is_valid_strategy_signal(signal)` 是 Baseline 全链路合法性判定的**单一真源**:任何模块判断“这条 opinion 是否有资格进入 Evidence Chain”都必须调用此函数。
Baseline 禁止在 `_STRATEGY_SIGNAL_ALIASES` 之外再维护第二份 canonical 映射表ConflictDetector 与 Synthesizer 内部的 `strategy_signal_score(canonical)` 只接受 canonical 值,禁止用 `op.original_signal` 或大小写变体查表。
### Evidence Chain 与 Diagnostics 分离
Baseline 规定:
- **Evidence Chain 是且仅是 Valid Opinion 集合**。DecisionAgent prompt、`strategy_synthesis` 数值计算、`disagreement_summary` 建桶都必须从同一个 Evidence Chain 读取。
- **Invalid Opinion 必须落到 Diagnostics**`ctx.meta["invalid_opinions"]` 或等价字段),仅用于日志、诊断、用户可见的“另有 N 个策略解析失败”计数。
- 两个集合**互斥且并集穷尽**:一条 opinion 要么在 Evidence Chain要么在 Diagnostics不得同时出现或都不出现。
- Invalid Opinion **不得**被静默转换成 `hold` / `confidence` 保留原值 / 匿名混入 `bullish_agents` / `bearish_agents` / `neutral_agents` 桶。
Diagnostics 结构:
```python
ctx.meta["invalid_opinions"] = [
{
"agent_name": str, # 原始 agent_name
"raw_signal": str | None, # 原始 signal 字面量(未归一化)
"confidence": float, # 原始 confidence仅诊断不参与任何计算
"reason": str, # "missing_signal" | "unrecognized_signal" | "invalid_flag"
},
...
]
```
Baseline 只规定该结构,不规定分拣发生的**代码位置**——Phase 1 会把分拣落到 Orchestrator。
### 动态二分阵营Supporting / Opposing
给定最终信号 `final_signal` 与 canonical score `final_score = strategy_signal_score(final_signal)`,对每个 Valid Opinion `op` 计算 `op_score = strategy_signal_score(op.signal)`
- **当 `final_signal == "hold"`(即 `final_score == 3.0`)时**
- `op_score == 3.0``supporting_skills`
- `op_score != 3.0``opposing_skills`(作为异议与分歧收纳,保证观望与分歧观点不被静默丢弃,避免展示时丢失异议背景)
- **当 `final_signal` 为方向性信号(`strong_buy` / `buy` / `sell` / `strong_sell`)时**
- 同向(都看涨 或 都看跌)且 `abs(op_score - final_score) ≤ 1.0``supporting_skills`
- 反向 且 `abs(op_score - final_score) ≥ 2.0``opposing_skills`
- 其余(`abs(diff) < 2.0` 且非同向)→ `opposing_skills`(并入异议,杜绝第三阵营 `neutral_skills`
Baseline 明确 **`neutral_skills` 不作为 payload 的正式字段**。每个 Valid Opinion 必须**恰好**落入 `supporting_skills``opposing_skills` 其一,分组结果总数必须等于 `summary_params.opinion_count`
### 共识度门槛
Baseline 固定共识度按 valid 样本数梯度判定:
| valid 数量 | consensus_level | 说明 |
| --- | --- | --- |
| 0 | `insufficient` | 无证据可综合final_signal 强制 `hold``confidence=0.0` |
| 1 | `insufficient` | 单样本不构成"共识",即使与 final 完全一致也不得输出 `high` |
| ≥ 2`sum(confidence) == 0` | `insufficient` | 有效证据的置信度为零,无从建立共识 |
| ≥ 2`sum(confidence) > 0` | 进入 aligned_ratio 判定 | 见下表 |
Aligned Ratio 判定valid ≥ 2 且 `sum(confidence) > 0`
| 条件 | consensus_level |
| --- | --- |
| `conflict_severity == "high"` | `low` |
| `aligned_ratio ≥ 2/3``conflict_count == 0`(等价 `conflict_severity == "none"` | `high` |
| `conflict_severity == "medium"``aligned_ratio < 0.5` | `low` |
| 其余 | `medium` |
其中 `aligned = 与 final_signal 同向且 score 距离 ≤ 1.0 的 valid 数量``aligned_ratio = aligned / len(valid)`
Baseline 禁止使用 `sum(...) or 1.0` 之类的兜底把零权重掩盖成分母 1零权重必须显式走 `insufficient` 分支,并让 `final_signal` 退回 `hold`
### `strategy_synthesis` Payload Schema
```json
{
"final_signal": "hold", // canonical signal
"weighted_score": 3.0, // 保留 4 位小数
"confidence": 0.72, // 折减后的置信度
"original_confidence": 0.80, // 折减前的加权置信度
"conflict_count": 0,
"conflict_severity": "none", // none | low | medium | high
"conflicts": [ /* ConflictDetector 输出的 dict 列表 */ ],
"supporting_skills": [ /* opinion item */ ],
"opposing_skills": [ /* opinion item */ ],
"consensus_level": "high", // high | medium | low | insufficient
"summary_key": "strategy_synthesis.no_conflicts", // 动态 i18n 摘要键名,随共识和冲突状态确定
"summary_params": {
"opinion_count": 2, // valid 样本数Evidence Chain 大小)
"total_opinion_count": 4, // valid + invalid分拣前原始输入总数
"invalid_opinion_count": 2, // Diagnostics 长度
"final_signal": "hold",
"consensus_level": "high",
"conflict_severity": "none",
"conflict_count": 0
}
}
```
Opinion Item 结构(`supporting_skills` / `opposing_skills` 每个元素):
```json
{
"skill_id": "trend_v1",
"agent_name": "skill_trend_v1",
"signal": "hold", // canonical
"confidence": 0.80, // 保留 4 位小数
"reasoning": "...",
"score_adjustment": 0,
"conditions_met": []
}
```
Baseline 明确 `strategy_synthesis` 是**由 SkillAggregator 确定性算法产出的唯一权威合成结果**。Orchestrator 的 `_collect_strategy_synthesis()` 必须优先使用 `ctx.get_data("skill_consensus")` 中的 synthesis只有在 SkillAggregator 未产出时才允许回退到 `ctx.opinions` 中的 `skill_consensus` opinion。**LLM 返回的 dashboard 不得覆盖或修改 `dashboard.strategy_synthesis`**`normalize_dashboard_payload` 收到 LLM 输出时应剥离 LLM 侧的 `strategy_synthesis` 字段,避免 LLM 幻觉污染权威合成结果。
### 关键不变量
Baseline 的语义边界收敛为八条不变量。所有 Phase N 的实现必须同时满足这八条,任一违反视为契约破坏。
| ID | 不变量 | 场景 | 期望 |
| --- | --- | --- | --- |
| I-1 | Evidence Chain 排他性 | 任何模块读取 Evidence Chain | 集合内每一条都必须 `is_valid_strategy_signal == True`Invalid 不允许出现 |
| I-2 | 禁止静默转换 | 缺失或无法识别的 signal | 归入 Diagnostics不得转换成 `hold` 后混入 Evidence Chain 或建桶 |
| I-3 | 零证据 → insufficient | 任意有效信号但 `sum(confidences) == 0`,或 valid 数量 = 0 | `final_signal="hold"`, `weighted_confidence=0.0`, `consensus_level="insufficient"`;禁止输出 `strong_sell` 或任何方向性信号 |
| I-4 | 单样本 → insufficient | 恰好 1 个 valid opinion | `consensus_level="insufficient"`,即使与 final 完全一致 |
| I-5 | Hold-final 一致性 | `final_signal == "hold"` 且存在 ≥ 2 个 hold valid opinion | 全部 hold opinion 必须归入 `supporting_skills`consensus_level 与 supporting_skills 数量关系必须自洽(`high` 时 supporting 覆盖 ≥ 2/3 |
| I-6 | Payload 与 renderer 语义一致 | `dashboard.strategy_synthesis` 值 | 四条 rendererMarkdown / WeChat / Notification / History实际文本必须与 payload 完全一致,不得出现"共识度:高 + 支持策略:无"等自相矛盾组合 |
| I-7 | Canonical-First 评分 | Aggregator / ConflictDetector / Synthesizer 内部的评分、加权、冲突判定、分组 | 必须使用 `normalize_strategy_signal()` 返回的 canonical 小写值;禁止用大写 `"BUY"`、别名等原始字符串直接查 `strategy_signal_score` |
| I-8 | 多语言空占位符 | `supporting_skills` / `opposing_skills` 为空时的展示 | 必须通过 `labels.none_label``report_language` 查表;禁止在代码或模板中硬编码中文 `"无"` / 英文 `"None"` / 韩文 `"없음"` 字面量 |
## Phase 1 语义收敛(本 PR 交付范围)
Phase 1 是 Baseline 契约的第一版代码化实现。Phase 1 **不新增契约条款**,只把 Baseline 已经写死的边界落到具体代码Orchestrator 分拣、Aggregator/Synthesizer 计算收敛、DecisionAgent prompt 收敛、Disagreement 收敛、四条 renderer 一致性、E2E 反例覆盖。
Phase 1 涉及的入口:
- `src/agent/protocols.py`:新增 `is_valid_strategy_signal()` 单一真源,`normalize_strategy_signal()` 保留 invalid 状态位。
- `src/agent/skills/engine.py``StrategyEngine.process()` 通过 `partition_only()` 完成唯一权威分拣,再由 `process_partition()` 驱动聚合与合成Valid 保留在 Evidence ChainInvalid 写入 Diagnostics。
- `src/agent/orchestrator.py`:在 DecisionAgent 运行前调用 `_run_strategy_engine(ctx)`timeout / budget-skip 早退路径调用 `_apply_partition_fallback(ctx)`,只分拣、不合成,避免 Invalid 回流证据链。
- `src/agent/skills/aggregator.py``StrategyEngine``valid_skill_opinions` 交给 `SkillAggregator.calculate()`;数学计算只使用 valid opinion`valid_weight_sum == 0` 显式走 `insufficient` 分支。
- `src/agent/skills/synthesis.py``ConflictDetector` / `StrategySynthesizer` 使用 canonical signal 计算;`_group_opinions()` 按 §"动态二分阵营" 实现;`_consensus_level()` 按 §"共识度门槛" 实现;`summary_params` 补齐 `invalid_opinion_count` / `total_opinion_count`
- `src/agent/agents/decision_agent.py``build_user_message()` 直接消费 `ctx.opinions`,不再二次过滤;在 prompt 中如实展示 `ctx.meta["invalid_opinions"]` 数量。
- `src/agent/disagreement.py``build_agent_disagreement_summary()` 直接消费 `ctx.opinions`(因 StrategyEngine 已完成分拣并由 Orchestrator 写回Invalid 完全不出现在 `bullish_agents` / `bearish_agents` / `neutral_agents` 三桶中。
- `src/services/report_renderer.py``templates/report_markdown.j2``templates/report_wechat.j2``src/notification.py``src/services/history_service.py`:读取 `strategy_synthesis.supporting_skills` / `opposing_skills` / `consensus_level` / `summary_params.invalid_opinion_count`;空列表通过 `labels.none_label` 输出;不再消费 `neutral_skills`
- `src/report_language.py``labels.none_label` 在 zh/en/ko 三语中完备;共识度、诊断计数文案完备。
- `tests/test_multi_agent.py`:新增 E2E-A..G 反例矩阵,从 SkillAgent 输入 → StrategyEngine 分拣/聚合 → DecisionAgent prompt → dashboard payload → renderer 实际文本全链路断言。
Phase 1 不改变 `AgentOpinion` 字段、不改变 API 返回结构、不改变数据库 schema、不新增配置项、不改变现有 skill 的执行方式。
## Phase 2 并发调度(本 PR 不做)
Phase 2 只在 Phase 1 契约下新增 24 策略并发调度与阶段调度:
- 策略执行从串行改为并发(`asyncio.gather` 或 thread pool阶段调度中按 `SKILL_CONCURRENCY` / `SKILL_TIMEOUT_PER_SKILL` 控制。
- 单个 skill 超时或异常,走 Baseline Invalid 处理路径(`reason="skill_timeout"` / `skill_error"`),进入 Diagnostics不阻塞其他 skill 与主流程。
- Phase 2 不改变 Baseline Evidence Chain / Diagnostics 分离原则、不改变阵营语义、不改变共识门槛、不改变 payload schema。
- Phase 2 不改变 renderer 展示逻辑;`invalid_opinion_count` 计数天然覆盖超时/异常 skill。
## Phase 3 前端多语言完整展示(本 PR 不做)
Phase 3 只在 Phase 2 之上补前端(`apps/dsa-web/``apps/dsa-desktop/`)对 `strategy_synthesis` 的完整多语言展示:
- Web 报告详情页展示 `final_signal` / `consensus_level` / `supporting_skills` / `opposing_skills` / `conflicts` / `invalid_opinion_count`
- 桌面端复用 Web 展示逻辑。
- 多语言 label 表复用 `src/report_language.py` 已有的 zh/en/ko 三语;前端只做投影,不重新定义。
- Phase 3 不改变 Baseline 契约、不新增 payload 字段、不新增 API 端点。
## Phase 4 权重回测反馈闭环(本 PR 不做)
Phase 4 在同一 `CONTRACT_VERSION = "1.0"` 内补权重回测反馈:
- `SkillAggregator._compute_weight()` 已有 `perf_weight` / `_backtest_factor()` 接线Phase 4 只补自动权重更新的闭环。
- Phase 4 不改变 Baseline canonical signal / valid 判定 / 共识门槛 / 阵营语义;权重变化只影响 `weighted_score``confidence`,不影响 `consensus_level` 判定路径。
## 消费面盘点
Baseline 的七条消费面必须严格按下表分工,不得越界互相消费对方的内部数据。
### SkillAgent
各 skill 通过 `src/agent/skills/skill_agent.py` 产出 `AgentOpinion`。Baseline 允许 skill 输出任意 signal 字面量(含大写、别名、`Signal` 枚举),也允许 skill 因数据不足产出 `signal=None` / 缺失字段——这些情况由下游分拣处理skill 本身不做自我过滤。
### StrategyEngine / Orchestrator分拣与接线
Phase 1 在 DecisionAgent 运行前由 Orchestrator 调用 `_run_strategy_engine(ctx)`
- `StrategyEngine.partition_only()` 遍历所有 `agent_name` 命中 `is_skill_agent_name()` 的观点,并使用 `normalize_strategy_signal()` 保留 canonical signal。
- Invalid 从 Evidence Chain 移除,写入 `StrategyResult.invalid_records`Orchestrator 再把它赋给 `ctx.meta["invalid_opinions"]`
- `StrategyEngine.process_partition()` 只把 `valid_skill_opinions` 交给 Aggregator/Synthesizer产出的 consensus opinion 和 `skill_consensus``_run_strategy_engine()` 一次写回 context。
- timeout / budget-skip 发生在完整 engine 运行前时,`_apply_partition_fallback()` 复用 `partition_only()`,只完成分拣和 Diagnostics 写回,不生成 consensus。
Baseline 规定 `StrategyEngine.partition_only()` 是**唯一权威分拣实现**。Aggregator / DecisionAgent / Disagreement 不再各自定义 Valid/Invalid 规则,直接消费 engine 收敛后的 Evidence ChainOrchestrator 中保留的旧 wrapper 仅用于兼容现有内部调用/测试,不属于正常运行时链路。
### SkillAggregator
正常运行时由 `StrategyEngine` 调用 `SkillAggregator.calculate(valid_skill_opinions)`。Aggregator 把输入转换为内部 `StrategyOpinion`,数学计算只使用 valid opinion并严格使用 canonical signal 查 `strategy_signal_score`;兼容入口即使收到未分拣输入也不得让 Invalid 参与权重。对以下三种状态显式走 `insufficient` 分支:
- `len(valid) == 0``final_signal="hold"`, `confidence=0.0`
- `len(valid) == 1`:按该 opinion 的 canonical signal 输出 `final_signal`,但 `consensus_level="insufficient"`
- `len(valid) ≥ 2``sum(confidence) == 0``final_signal="hold"`, `confidence=0.0`
产出的 `strategy_synthesis``StrategyEngine` 装入 `StrategyResult.skill_consensus_data`,再由 Orchestrator 挂到 `ctx.set_data("skill_consensus", {...})``_collect_strategy_synthesis()` 从这里读取,作为 dashboard 的权威合成源。
### DecisionAgent
`build_user_message()``ctx.opinions` 读取观点写入 prompt。因为 Orchestrator 分拣已保证 `ctx.opinions` 只含 ValidDecisionAgent **不再**做二次过滤。Prompt 中"另有 N 个策略解析失败"的展示直接读取 `ctx.meta["invalid_opinions"]` 长度。
DecisionAgent 输出的 dashboard JSON 不得覆盖 `dashboard.strategy_synthesis`;如果 LLM 返回中含有该字段,`normalize_dashboard_payload()` 必须剥离,保留 Aggregator 侧的权威合成。
### Disagreement
`build_agent_disagreement_summary()` 只从 `ctx.opinions``bullish_agents` / `bearish_agents` / `neutral_agents` 三桶。因为 `ctx.opinions` 已只含 ValidInvalid 完全不出现在三桶中,也不会被 `_normalize_signal()` 静默兜底为 `hold`
`ctx.meta["invalid_opinions"]` 长度作为 `disagreement_summary.diagnostics.invalid_count` 单独暴露给 DecisionAgent prompt供 LLM 生成 `data_limitations` 文案参考。
### Renderer四条
所有 renderer 读取 `dashboard.strategy_synthesis` 展示:
- `final_signal` / `consensus_level` / `conflict_severity` / `conflict_count`
- `supporting_skills` / `opposing_skills`(不再消费 `neutral_skills`)。
- `summary_params.invalid_opinion_count` → 按语言展示"另有 N 个策略无效/解析失败"。
空列表占位符必须通过 `labels.none_label`(按 `report_language` 查表)输出。四条 renderer 展示的最终文本必须与 payload 完全一致,不得出现"共识度:高 + 支持策略:无"这类内部矛盾。
历史记录和外部调用方可能保留契约落地前的宽松 shape。四条 renderer 必须先通过 `normalize_strategy_synthesis_payload()` 把非 dict 顶层值视为缺失,并过滤非 dict 的策略/冲突列表项;`strategy_invalid_opinion_count()` 统一读取诊断计数,只对纯十进制正整数字符串做窄转换,其余坏值降级为 0。禁止在 History、Notification 或模板中保留平行的手写读取逻辑。
### Diagnostics
`ctx.meta["invalid_opinions"]` 只允许被以下三类消费:
- 日志:记录 `agent_name` / `raw_signal` / `reason`,供排障。
- DecisionAgent prompt作为"另有 N 个策略解析失败"的计数来源。
- Renderer作为 `summary_params.invalid_opinion_count` 的来源。
禁止把 Diagnostics 里的 `confidence` 参与任何加权计算;禁止把 `raw_signal` 塞回 `ctx.opinions`
## 反例矩阵
Phase 1 必须提供如下 E2E 反例覆盖。E2E 定义为:从 SkillAgent 输入进,穿过 Orchestrator 分拣 → SkillAggregator → DecisionAgent prompt → 最终 dashboard payload → 四条 renderer 实际文本输出。禁止用局部单元测试冒充 E2E。
| 编号 | 输入 | 断言点 | 覆盖的不变量 |
| --- | --- | --- | --- |
| E2E-A | 1 valid `buy/0.8` + 2 invalid `moon/0.9` | ① DecisionAgent prompt 不含 `moon` 字面量、不含 invalid `agent_name`、不含 `0.9` 上下文;② `ctx.meta["invalid_opinions"]` 长度 = 2`strategy_synthesis.summary_params.opinion_count == 1``invalid_opinion_count == 2`;④ `consensus_level == "insufficient"`;⑤ 四条 renderer 输出文本包含"另有 2 个策略无效/解析失败"(按语言);⑥ `disagreement_summary.bullish_agents` / `neutral_agents` / `bearish_agents` 中都不出现 moon 转成的 hold/0.9 | I-1, I-2, I-4 |
| E2E-B | 2 valid `hold/0.0` | `final_signal="hold"``weighted_confidence=0.0``consensus_level="insufficient"`、**绝不**出现 `strong_sell`;所有 renderer 展示"证据不足(观望)"(按语言) | I-3 |
| E2E-C | 1 valid `buy/0.0` + 1 valid `hold/0.0` | 混合零权重场景:`final="hold"``confidence=0.0``consensus="insufficient"`、无 `strong_sell` | I-3 |
| E2E-D | 2 valid `hold/0.8` | ① `final_signal="hold"``consensus_level="high"`;② `supporting_skills` 长度 = 2、`opposing_skills` 长度 = 0③ 四条 renderer 实际文本同时包含"高共识"和两个 skill 名,不得出现"支持策略:无"配"共识度:高"的组合 | I-5, I-6 |
| E2E-E | 1 valid `buy/0.8` + 9 invalid | `consensus_level="insufficient"`**不得** high四条 renderer 展示"基于 1 个有效策略判断(另有 9 个策略无效/解析失败)" | I-4, I-6 |
| E2E-F | 2 valid opinion其中一个 `signal="BUY"`(大写) | Aggregator 内部计算 `weighted_score` 时使用 canonical `buy` 查分4.0**不得**因大写查表失败得到 0`strategy_synthesis.final_signal` 输出 canonical 小写 | I-7 |
| E2E-G | 空 `supporting_skills` + `report_language="en"` | 四条 renderer 输出中不出现中文 `"无"`,而是 `"None"`(或对应语言 `labels.none_label` | I-8 |
## 源码锚点
| 域 | 锚点 |
| --- | --- |
| Signal 规范化与 Valid 判定 | `src/agent/protocols.py::normalize_strategy_signal`, `is_valid_strategy_signal`, `strategy_signal_score` |
| StrategyEngine 分拣与合成门面 | `src/agent/skills/engine.py::StrategyEngine.partition_only`, `process`, `process_partition` |
| Orchestrator 接线与早退分拣 | `src/agent/orchestrator.py::_run_strategy_engine`, `_apply_partition_fallback` |
| SkillAggregator | `src/agent/skills/aggregator.py::SkillAggregator.calculate`, `aggregate`(兼容入口) |
| ConflictDetector / StrategySynthesizer | `src/agent/skills/synthesis.py::ConflictDetector`, `StrategySynthesizer` |
| DecisionAgent prompt | `src/agent/agents/decision_agent.py::build_user_message` |
| Disagreement | `src/agent/disagreement.py::build_agent_disagreement_summary` |
| Dashboard 合成挂载 | `src/agent/orchestrator.py::_collect_strategy_synthesis` |
| Renderer · Markdown | `src/services/report_renderer.py::render`, `templates/report_markdown.j2` |
| Renderer · WeChat | `templates/report_wechat.j2` |
| Renderer · Notification | `src/notification.py`(策略综合行渲染) |
| Renderer · History | `src/services/history_service.py`(历史详情策略综合块) |
| 多语言与宽松 payload 防腐 | `src/report_language.py::_REPORT_LABELS`, `normalize_strategy_synthesis_payload`, `strategy_invalid_opinion_count`, `localize_strategy_synthesis_summary`, `labels.none_label` |
| E2E 反例矩阵 | `tests/test_multi_agent.py::TestP1SemanticConvergence`, `TestStrategyEngineE2E` |
## 兼容与回滚
### 已废弃行为Phase 1 落地后)
| 旧行为 | 契约后 |
| --- | --- |
| `normalize_strategy_signal` 对未知信号静默返回 `default="hold"` 并混入证据链 | 未知信号必须归入 Diagnostics`ctx.opinions` 中不允许出现 |
| Aggregator 通过 `sum(...) or 1.0` 掩盖零权重 | 显式判 `valid_weight_sum == 0`,走 `insufficient` 分支,`final_signal="hold"` |
| Renderer 硬编码 `"无"` 展示空阵营 | 通过 `labels.none_label` 按语言查表 |
| DecisionAgent 在 prompt 层自己过滤 invalid | 分拣在 Orchestrator 完成DecisionAgent 直接消费 `ctx.opinions` |
| `strategy_synthesis` 输出 `neutral_skills` | 契约后该字段不存在renderer 不再消费 |
| LLM dashboard 覆盖 `strategy_synthesis` | 权威合成来自 AggregatorLLM 侧字段被 `normalize_dashboard_payload` 剥离 |
### 已新增字段
- `ctx.meta["invalid_opinions"]`Diagnostics 收纳位(结构见"Evidence Chain 与 Diagnostics 分离")。
- `strategy_synthesis.summary_params.invalid_opinion_count`Diagnostics 长度。
- `strategy_synthesis.summary_params.total_opinion_count`valid + invalid 的原始总数。
### 回滚方式
| 手段 | 作用 | 不能做什么 |
| --- | --- | --- |
| 版本回退 Phase 1 相关提交 | 移除 Orchestrator 分拣、Aggregator/Synthesizer 收敛、renderer 一致性改动 | 无法只回退部分不变量;契约是整体收敛 |
| 只保留契约文档、回退代码 | 保留 Baseline 文本、回到旧行为 | 只有文档意义,无运行时收益;不推荐 |
| Phase 2/3/4 独立回退 | 各自 Phase 的运行时改动独立回退 | 不能回退 Baseline任何 Phase 都必须始终满足 Baseline 八条不变量 |
Baseline 不新增配置项,因此无 env-level 回滚开关;这是刻意选择——契约边界应在代码中恒定生效,不通过环境变量降级。

View File

@@ -172,9 +172,12 @@ should sum to 100; all-zero means no effective signal and must not be faked.
"", "",
] ]
# Feed prior opinions # Feed prior opinions — Orchestrator已在 _partition_skill_opinions 中完成
# skill 观点的分拣ctx.opinions 中不再含 invalid skill opinion
# invalid skill 观点存于 ctx.meta["invalid_opinions"]。
# DecisionAgent 直接消费,不再二次过滤。
if ctx.opinions: if ctx.opinions:
parts.append("## Agent Opinions") parts.append("## Agent Opinions (Evidence Chain)")
for op in ctx.opinions: for op in ctx.opinions:
parts.append(f"\n### {op.agent_name}") parts.append(f"\n### {op.agent_name}")
parts.append(f"Signal: {op.signal} | Confidence: {op.confidence:.2f}") parts.append(f"Signal: {op.signal} | Confidence: {op.confidence:.2f}")
@@ -183,11 +186,20 @@ should sum to 100; all-zero means no effective signal and must not be faked.
parts.append(f"Key levels: {json.dumps(op.key_levels)}") parts.append(f"Key levels: {json.dumps(op.key_levels)}")
if op.raw_data: if op.raw_data:
extra_keys = {k: v for k, v in op.raw_data.items() extra_keys = {k: v for k, v in op.raw_data.items()
if k not in ("signal", "confidence", "reasoning", "key_levels")} if k not in ("signal", "confidence", "reasoning", "key_levels", "invalid_signal")}
if extra_keys: if extra_keys:
parts.append(f"Extra data: {json.dumps(extra_keys, ensure_ascii=False, default=str)}") parts.append(f"Extra data: {json.dumps(extra_keys, ensure_ascii=False, default=str)}")
parts.append("") parts.append("")
invalid_opinions = ctx.meta.get("invalid_opinions") or []
if invalid_opinions:
parts.append("## Invalid Skill Opinions (Diagnostics only — not in evidence chain)")
parts.append(
f"{len(invalid_opinions)} 个 skill 观点因 signal 缺失或无法识别,已从证据链移除;"
f"仅供你在 data_limitations 中标注,不得作为决策依据。"
)
parts.append("")
# Feed risk flags # Feed risk flags
if ctx.risk_flags: if ctx.risk_flags:
parts.append("## Risk Flags") parts.append("## Risk Flags")

View File

@@ -11,7 +11,7 @@ from __future__ import annotations
from collections.abc import Iterable from collections.abc import Iterable
from typing import Any, Dict, List from typing import Any, Dict, List
from src.agent.protocols import AgentContext from src.agent.protocols import AgentContext, is_valid_strategy_signal
from src.agent.risk_override import build_risk_override_plan from src.agent.risk_override import build_risk_override_plan
_BULLISH_SIGNALS = {"strong_buy", "buy"} _BULLISH_SIGNALS = {"strong_buy", "buy"}
@@ -25,14 +25,24 @@ def build_agent_disagreement_summary(
*, *,
risk_override_enabled: bool = True, risk_override_enabled: bool = True,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Build a structured, low-sensitivity summary of prior agent disagreement.""" """Build a structured, low-sensitivity summary of prior agent disagreement.
Per docs/multi-strategy-contract.md §"Disagreement": ctx.opinions is
already partitioned by the orchestrator, so invalid skill opinions
never appear here. Invalid counts come from ctx.meta["invalid_opinions"]
(Diagnostics), not from re-filtering.
"""
buckets = { buckets = {
"bullish_agents": [], "bullish_agents": [],
"bearish_agents": [], "bearish_agents": [],
"neutral_agents": [], "neutral_agents": [],
} }
valid_count = 0
for opinion in ctx.opinions: for opinion in ctx.opinions:
if not is_valid_strategy_signal(opinion.signal):
continue
valid_count += 1
signal = _effective_signal(opinion.agent_name, opinion.signal) signal = _effective_signal(opinion.agent_name, opinion.signal)
agent_summary = _summarize_opinion(opinion.agent_name, signal, opinion.confidence) agent_summary = _summarize_opinion(opinion.agent_name, signal, opinion.confidence)
if signal in _BULLISH_SIGNALS: if signal in _BULLISH_SIGNALS:
@@ -55,7 +65,11 @@ def build_agent_disagreement_summary(
degraded_result, degraded_result,
) )
return { invalid_bucket = ctx.meta.get("invalid_opinions") or []
if not isinstance(invalid_bucket, list):
invalid_bucket = []
result = {
**buckets, **buckets,
"conflict_type": conflict_type, "conflict_type": conflict_type,
"decision_path_hint": _decision_path_hint(conflict_type), "decision_path_hint": _decision_path_hint(conflict_type),
@@ -63,8 +77,24 @@ def build_agent_disagreement_summary(
and risk_override_plan.override_trigger_present, and risk_override_plan.override_trigger_present,
"risk_control": risk_override_plan.to_low_sensitivity_dict(), "risk_control": risk_override_plan.to_low_sensitivity_dict(),
"degraded_result": degraded_result, "degraded_result": degraded_result,
"valid_opinion_count": valid_count,
} }
if invalid_bucket:
result["diagnostics"] = {
"invalid_opinions": [
{
"agent_name": str(item.get("agent_name") or "unknown"),
"reason": str(item.get("reason") or "unrecognized_signal"),
}
for item in invalid_bucket
if isinstance(item, dict)
],
"invalid_count": len(invalid_bucket),
}
return result
def _summarize_opinion(agent_name: str, signal: Any, confidence: Any) -> Dict[str, Any]: def _summarize_opinion(agent_name: str, signal: Any, confidence: Any) -> Dict[str, Any]:
"""Keep only low-sensitivity opinion metadata for downstream synthesis.""" """Keep only low-sensitivity opinion metadata for downstream synthesis."""
@@ -76,12 +106,9 @@ def _summarize_opinion(agent_name: str, signal: Any, confidence: Any) -> Dict[st
def _normalize_signal(signal: Any) -> str: def _normalize_signal(signal: Any) -> str:
if not isinstance(signal, str): from src.agent.protocols import normalize_strategy_signal
return "hold" canonical, invalid, _ = normalize_strategy_signal(signal)
normalized = signal.strip().lower() return "hold" if invalid else canonical
if normalized in _BULLISH_SIGNALS or normalized in _BEARISH_SIGNALS or normalized == "hold":
return normalized
return "hold"
def _effective_signal(agent_name: str, signal: Any) -> str: def _effective_signal(agent_name: str, signal: Any) -> str:

View File

@@ -40,8 +40,11 @@ from src.agent.protocols import (
AgentRunStats, AgentRunStats,
StageResult, StageResult,
StageStatus, StageStatus,
is_valid_strategy_signal,
normalize_decision_signal, normalize_decision_signal,
) )
from src.agent.skills.defaults import is_skill_agent_name
from src.agent.skills.engine import EvidencePartition, StrategyEngine, StrategyResult, StrategyResultStatus
from src.agent.risk_override import build_risk_override_plan from src.agent.risk_override import build_risk_override_plan
from src.agent.runner import parse_dashboard_json from src.agent.runner import parse_dashboard_json
from src.agent.stock_scope import resolve_stock_scope from src.agent.stock_scope import resolve_stock_scope
@@ -112,6 +115,7 @@ class AgentOrchestrator:
self.mode = normalized_mode if normalized_mode in VALID_MODES else "standard" self.mode = normalized_mode if normalized_mode in VALID_MODES else "standard"
self.skill_manager = skill_manager self.skill_manager = skill_manager
self.config = config self.config = config
self.strategy_engine = StrategyEngine()
def _get_timeout_seconds(self) -> int: def _get_timeout_seconds(self) -> int:
"""Return the pipeline timeout in seconds. """Return the pipeline timeout in seconds.
@@ -487,6 +491,8 @@ class AgentOrchestrator:
elapsed=round(elapsed_s, 2), elapsed=round(elapsed_s, 2),
timeout=timeout_s, timeout=timeout_s,
)) ))
if ctx is not None:
self._apply_partition_fallback(ctx)
return self._build_timeout_result( return self._build_timeout_result(
stats, stats,
all_tool_calls, all_tool_calls,
@@ -518,6 +524,8 @@ class AgentOrchestrator:
"remaining budget" "remaining budget"
), ),
)) ))
if ctx is not None:
self._apply_partition_fallback(ctx)
return self._build_budget_skip_result( return self._build_budget_skip_result(
stats, stats,
all_tool_calls, all_tool_calls,
@@ -543,9 +551,8 @@ class AgentOrchestrator:
agents[index:index] = specialist_agents agents[index:index] = specialist_agents
continue continue
# Aggregate skill opinions before the decision agent if agent.agent_name == "decision":
if agent.agent_name == "decision" and getattr(self, "_skill_agent_names", None): self._run_strategy_engine(ctx)
self._aggregate_skill_opinions(ctx)
if agent.agent_name == "decision": if agent.agent_name == "decision":
self._prepare_decision_context(ctx) self._prepare_decision_context(ctx)
@@ -592,6 +599,7 @@ class AgentOrchestrator:
elapsed=round(elapsed_s, 2), elapsed=round(elapsed_s, 2),
timeout=timeout_s, timeout=timeout_s,
)) ))
self._apply_partition_fallback(ctx)
return self._build_timeout_result( return self._build_timeout_result(
stats, stats,
all_tool_calls, all_tool_calls,
@@ -748,6 +756,58 @@ class AgentOrchestrator:
# Skill aggregation # Skill aggregation
# ----------------------------------------------------------------- # -----------------------------------------------------------------
def _partition_skill_opinions(self, ctx: AgentContext) -> None:
"""Split skill opinions into Evidence Chain (valid) and Diagnostics (invalid).
Per docs/multi-strategy-contract.md §"Evidence Chain 与 Diagnostics 分离":
this is the ONLY partition point. After this method, ctx.opinions
contains only valid skill opinions; invalid ones are moved to
ctx.meta["invalid_opinions"] and never re-enter downstream evidence.
"""
kept: List = []
invalid_bucket: List[Dict[str, Any]] = ctx.meta.setdefault("invalid_opinions", [])
if not isinstance(invalid_bucket, list):
invalid_bucket = []
ctx.meta["invalid_opinions"] = invalid_bucket
for op in ctx.opinions:
if not is_skill_agent_name(op.agent_name):
kept.append(op)
continue
raw_signal = op.signal if op.signal else (
op.raw_data.get("signal") if isinstance(op.raw_data, dict) else None
)
if raw_signal is None or (isinstance(raw_signal, str) and not raw_signal.strip()):
reason = "missing_signal"
raw_display = raw_signal if isinstance(raw_signal, str) else None
is_valid = False
elif is_valid_strategy_signal(raw_signal):
is_valid = True
reason = ""
raw_display = str(raw_signal)
else:
is_valid = False
reason = "unrecognized_signal"
raw_display = str(raw_signal)
if is_valid:
kept.append(op)
else:
invalid_bucket.append({
"agent_name": op.agent_name,
"raw_signal": raw_display,
"confidence": op.confidence,
"reason": reason,
})
logger.info(
"[Orchestrator] invalid skill opinion moved to diagnostics: agent=%s raw_signal=%r reason=%s",
op.agent_name, raw_display, reason,
)
ctx.opinions = kept
def _aggregate_skill_opinions(self, ctx: AgentContext) -> None: def _aggregate_skill_opinions(self, ctx: AgentContext) -> None:
"""Run SkillAggregator to produce a consensus opinion. """Run SkillAggregator to produce a consensus opinion.
@@ -764,6 +824,9 @@ class AgentOrchestrator:
"signal": consensus.signal, "signal": consensus.signal,
"confidence": consensus.confidence, "confidence": consensus.confidence,
"reasoning": consensus.reasoning, "reasoning": consensus.reasoning,
"raw_data": consensus.raw_data,
"strategy_synthesis": consensus.raw_data.get("strategy_synthesis"),
"conflicts": consensus.raw_data.get("conflicts", []),
}) })
logger.info( logger.info(
"[Orchestrator] skill consensus: signal=%s confidence=%.2f", "[Orchestrator] skill consensus: signal=%s confidence=%.2f",
@@ -778,6 +841,54 @@ class AgentOrchestrator:
"""Compatibility wrapper for legacy tests/imports.""" """Compatibility wrapper for legacy tests/imports."""
self._aggregate_skill_opinions(ctx) self._aggregate_skill_opinions(ctx)
def _run_strategy_engine(self, ctx: AgentContext) -> None:
"""Run the full skill pipeline via StrategyEngine and update ctx.
Replaces the old two-step _partition_skill_opinions + _aggregate_skill_opinions
calls. The engine is the single authoritative owner of strategy_synthesis.
"""
result = self.strategy_engine.process(ctx.opinions)
ctx.meta["invalid_opinions"] = list(result.invalid_records)
ctx.opinions = list(result.non_skill_opinions) + list(result.valid_skill_opinions)
if result.consensus_opinion is not None:
ctx.opinions.append(result.consensus_opinion)
if result.skill_consensus_data is not None:
ctx.set_data("skill_consensus", result.skill_consensus_data)
if result.status == StrategyResultStatus.CONSENSUS:
logger.info(
"[Orchestrator] strategy engine: signal=%s confidence=%.2f",
result.consensus_opinion.signal,
result.consensus_opinion.confidence,
)
elif result.status == StrategyResultStatus.NO_CONSENSUS:
logger.info(
"[Orchestrator] strategy engine: NO_CONSENSUS invalid_count=%d",
result.invalid_count,
)
else:
logger.info("[Orchestrator] strategy engine: NO_SKILLS")
def _apply_partition_fallback(self, ctx: AgentContext) -> None:
"""Partition skill opinions for timeout/budget-skip early-exit paths.
Does not aggregate — only ensures invalid diagnostics are preserved
in ctx.meta["invalid_opinions"] before the pipeline bails out.
Idempotent: skips if the engine already ran fully (skill_consensus present).
"""
if ctx.get_data("skill_consensus") is not None:
return
partition = self.strategy_engine.partition_only(ctx.opinions)
ctx.opinions = list(partition.non_skill_opinions) + list(partition.valid_skill_opinions)
invalid_bucket = ctx.meta.get("invalid_opinions")
if not isinstance(invalid_bucket, list):
invalid_bucket = []
invalid_bucket.extend(partition.invalid_records)
ctx.meta["invalid_opinions"] = invalid_bucket
def _prepare_decision_context(self, ctx: AgentContext) -> None: def _prepare_decision_context(self, ctx: AgentContext) -> None:
"""Populate low-sensitivity summaries consumed by DecisionAgent.""" """Populate low-sensitivity summaries consumed by DecisionAgent."""
ctx.meta["agent_disagreement_summary"] = build_agent_disagreement_summary( ctx.meta["agent_disagreement_summary"] = build_agent_disagreement_summary(
@@ -978,6 +1089,8 @@ class AgentOrchestrator:
dashboard_block = {} dashboard_block = {}
else: else:
dashboard_block = dict(dashboard_block) dashboard_block = dict(dashboard_block)
# Strip any LLM-written strategy_synthesis — StrategyEngine is the sole writer.
dashboard_block.pop("strategy_synthesis", None)
core = dashboard_block.get("core_conclusion") core = dashboard_block.get("core_conclusion")
if not isinstance(core, dict): if not isinstance(core, dict):
@@ -1124,6 +1237,10 @@ class AgentOrchestrator:
if data_perspective: if data_perspective:
dashboard_block["data_perspective"] = data_perspective dashboard_block["data_perspective"] = data_perspective
strategy_synthesis = self._collect_strategy_synthesis(ctx, dashboard_block)
if strategy_synthesis:
dashboard_block["strategy_synthesis"] = strategy_synthesis
dashboard_block["core_conclusion"] = core dashboard_block["core_conclusion"] = core
dashboard_block["intelligence"] = intelligence dashboard_block["intelligence"] = intelligence
dashboard_block["battle_plan"] = battle dashboard_block["battle_plan"] = battle
@@ -1156,6 +1273,33 @@ class AgentOrchestrator:
payload["dashboard"] = dashboard_block payload["dashboard"] = dashboard_block
return payload return payload
def _collect_strategy_synthesis(
self,
ctx: AgentContext,
dashboard_block: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
# Deterministic synthesis from skill_consensus is the authoritative source
consensus_data = ctx.get_data("skill_consensus")
if isinstance(consensus_data, dict):
synthesis = consensus_data.get("strategy_synthesis")
if isinstance(synthesis, dict) and synthesis:
return synthesis
raw_data = consensus_data.get("raw_data")
if isinstance(raw_data, dict):
synthesis = raw_data.get("strategy_synthesis")
if isinstance(synthesis, dict) and synthesis:
return synthesis
# Fallback: scan opinions
for opinion in reversed(ctx.opinions):
if getattr(opinion, "agent_name", "") != "skill_consensus":
continue
raw_data = opinion.raw_data if isinstance(opinion.raw_data, dict) else {}
synthesis = raw_data.get("strategy_synthesis")
if isinstance(synthesis, dict) and synthesis:
return synthesis
return None
def _collect_key_levels( def _collect_key_levels(
self, self,
ctx: AgentContext, ctx: AgentContext,

View File

@@ -36,6 +36,69 @@ _CANONICAL_DECISION_SIGNAL_MAP: Dict[str, str] = {
"strong_sell": "sell", "strong_sell": "sell",
} }
_STRATEGY_SIGNAL_ALIASES: Dict[str, str] = {
"strong_buy": "strong_buy",
"strong buy": "strong_buy",
"strong-buy": "strong_buy",
"strongbuy": "strong_buy",
"buy": "buy",
"hold": "hold",
"neutral": "hold",
"sell": "sell",
"strong_sell": "strong_sell",
"strong sell": "strong_sell",
"strong-sell": "strong_sell",
"strongsell": "strong_sell",
}
def normalize_strategy_signal(signal: Any, default: str = "hold") -> tuple[str, bool, str]:
"""Normalize strategy signal labels while preserving invalid input state.
Single normalization entrypoint for the entire multi-strategy pipeline. See
docs/multi-strategy-contract.md §"Canonical Signal 与 Valid 判定".
Returns (canonical, invalid, original):
- canonical: canonical lowercase label; falls back to `default` when invalid
- invalid: True when input cannot be mapped to any canonical label
- original: original stripped string form (used for diagnostics only)
"""
if signal is None:
original = ""
elif hasattr(signal, "value"):
original = str(signal.value).strip()
else:
original = str(signal).strip()
normalized = original.lower().replace("/", "_")
canonical = _STRATEGY_SIGNAL_ALIASES.get(normalized)
if canonical is not None:
return canonical, False, original
return default, True, original
def is_valid_strategy_signal(signal: Any) -> bool:
"""Single source of truth for signal validity across the entire pipeline.
Consumers: SkillAgent → Orchestrator partitioning → SkillAggregator →
StrategySynthesizer → DecisionAgent → renderers. Delegates to
normalize_strategy_signal so alias/canonical rules stay consistent.
"""
_, invalid, _ = normalize_strategy_signal(signal)
return not invalid
def strategy_signal_score(signal: str) -> float:
scores = {
"strong_buy": 5.0,
"buy": 4.0,
"hold": 3.0,
"sell": 2.0,
"strong_sell": 1.0,
}
if signal not in scores:
raise ValueError(f"Unknown strategy signal: {signal!r}")
return scores[signal]
def normalize_decision_signal(signal: Any, default: str = "hold") -> str: def normalize_decision_signal(signal: Any, default: str = "hold") -> str:
"""Map model-facing signal labels to the dashboard's stable enum.""" """Map model-facing signal labels to the dashboard's stable enum."""
@@ -155,6 +218,39 @@ class AgentOpinion:
return None return None
@dataclass
class StrategyOpinion:
"""Normalized view of a skill/strategy opinion for synthesis."""
skill_id: str = ""
agent_name: str = ""
signal: str = "hold"
confidence: float = 0.0
reasoning: str = ""
score_adjustment: float = 0.0
conditions_met: List[str] = field(default_factory=list)
conditions_missed: List[str] = field(default_factory=list)
key_levels: Dict[str, float] = field(default_factory=dict)
raw_data: Dict[str, Any] = field(default_factory=dict)
original_signal: str = ""
invalid_signal: bool = False
def __post_init__(self) -> None:
self.confidence = max(0.0, min(1.0, float(self.confidence)))
@dataclass
class StrategyConflict:
"""Deterministic conflict found among strategy opinions."""
conflict_type: str = ""
severity: str = "medium"
description: str = ""
description_key: str = ""
participants: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
# ============================================================ # ============================================================
# StageResult — return type from a single pipeline stage # StageResult — return type from a single pipeline stage
# ============================================================ # ============================================================

View File

@@ -6,28 +6,27 @@ SkillAggregator — weighted aggregation of skill opinions.
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Dict, List, Optional from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from src.agent.memory import AgentMemory from src.agent.memory import AgentMemory
from src.agent.protocols import AgentContext, AgentOpinion from src.agent.protocols import AgentContext, AgentOpinion, StrategyConflict, StrategyOpinion
from src.agent.skills.defaults import ( from src.agent.skills.defaults import (
SKILL_CONSENSUS_AGENT_NAME, SKILL_CONSENSUS_AGENT_NAME,
extract_skill_id, extract_skill_id,
is_skill_agent_name, is_skill_agent_name,
) )
from src.agent.skills.synthesis import (
ConflictDetector,
StrategySynthesizer,
strategy_opinion_from_agent_opinion,
strategy_signal_score,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_MIN_BACKTEST_SAMPLES = 30 _MIN_BACKTEST_SAMPLES = 30
_SIGNAL_SCORES: Dict[str, float] = {
"strong_buy": 5.0,
"buy": 4.0,
"hold": 3.0,
"sell": 2.0,
"strong_sell": 1.0,
}
_SCORE_TO_SIGNAL = [ _SCORE_TO_SIGNAL = [
(4.5, "strong_buy"), (4.5, "strong_buy"),
(3.5, "buy"), (3.5, "buy"),
@@ -37,6 +36,21 @@ _SCORE_TO_SIGNAL = [
] ]
@dataclass
class AggregationData:
skill_opinions: List[AgentOpinion] = field(default_factory=list)
weights: List[float] = field(default_factory=list)
skill_names: List[str] = field(default_factory=list)
strategy_opinions: List[StrategyOpinion] = field(default_factory=list)
weighted_score: float = 3.0
weighted_confidence: float = 0.0
insufficient_evidence: bool = False
conflicts: List[StrategyConflict] = field(default_factory=list)
final_signal: str = "hold"
individual_signals: Dict[str, Dict[str, Any]] = field(default_factory=dict)
total_adjustment: float = 0.0
class SkillAggregator: class SkillAggregator:
"""Aggregate multiple skill-agent opinions into one consensus.""" """Aggregate multiple skill-agent opinions into one consensus."""
@@ -45,7 +59,28 @@ class SkillAggregator:
ctx: AgentContext, ctx: AgentContext,
min_samples: int = _MIN_BACKTEST_SAMPLES, min_samples: int = _MIN_BACKTEST_SAMPLES,
) -> Optional[AgentOpinion]: ) -> Optional[AgentOpinion]:
skill_opinions = [op for op in ctx.opinions if is_skill_agent_name(op.agent_name)] aggregation = self.calculate(ctx.opinions, min_samples=min_samples)
if aggregation is None:
return None
invalid_count = sum(1 for opinion in aggregation.strategy_opinions if opinion.invalid_signal)
synthesis = StrategySynthesizer().synthesize(
aggregation.strategy_opinions,
weighted_score=aggregation.weighted_score,
final_signal=aggregation.final_signal,
weighted_confidence=aggregation.weighted_confidence,
conflicts=aggregation.conflicts,
insufficient_evidence=aggregation.insufficient_evidence,
invalid_count=invalid_count,
)
return self.build_consensus_opinion(aggregation, synthesis)
def calculate(
self,
opinions: List[AgentOpinion],
min_samples: int = _MIN_BACKTEST_SAMPLES,
) -> Optional[AggregationData]:
skill_opinions = [op for op in opinions if is_skill_agent_name(op.agent_name)]
if not skill_opinions: if not skill_opinions:
return None return None
@@ -70,49 +105,100 @@ class SkillAggregator:
) )
weights.append(weight) weights.append(weight)
total_weight = sum(weights) or 1.0 strategy_opinions = [
weighted_score = sum( strategy_opinion_from_agent_opinion(op)
_SIGNAL_SCORES.get(op.signal, 3.0) * weight for op in skill_opinions
for op, weight in zip(skill_opinions, weights) ]
) / total_weight
weighted_confidence = sum( valid_opinions_with_weights = [
op.confidence * weight (op, strategy, weight)
for op, weight in zip(skill_opinions, weights) for op, strategy, weight in zip(skill_opinions, strategy_opinions, weights)
) / total_weight if not strategy.invalid_signal
]
valid_weight_sum = sum(weight for _, _, weight in valid_opinions_with_weights)
insufficient_evidence = (
not valid_opinions_with_weights or valid_weight_sum <= 0
)
if not insufficient_evidence:
weighted_score = sum(
strategy_signal_score(strategy.signal) * weight
for _, strategy, weight in valid_opinions_with_weights
) / valid_weight_sum
weighted_confidence = sum(
op.confidence * weight
for op, _, weight in valid_opinions_with_weights
) / valid_weight_sum
else:
weighted_score = 3.0
weighted_confidence = 0.0
total_adjustment = sum( total_adjustment = sum(
op.raw_data.get("score_adjustment", 0) op.raw_data.get("score_adjustment", 0)
for op in skill_opinions for op, strategy, weight in valid_opinions_with_weights
if isinstance(op.raw_data.get("score_adjustment"), (int, float)) if isinstance(op.raw_data.get("score_adjustment"), (int, float))
) )
final_signal = "hold" if insufficient_evidence:
for threshold, signal in _SCORE_TO_SIGNAL: final_signal = "hold"
if weighted_score >= threshold: else:
final_signal = signal final_signal = "hold"
break for threshold, signal in _SCORE_TO_SIGNAL:
if weighted_score >= threshold:
final_signal = signal
break
skill_names = [extract_skill_id(op.agent_name) or op.agent_name for op in skill_opinions] conflicts = ConflictDetector().detect(strategy_opinions, final_signal=final_signal)
individual_signals = {
op.agent_name: {
"signal": strategy.signal,
"confidence": op.confidence,
"original_signal": strategy.original_signal,
"invalid_signal": strategy.invalid_signal,
}
for op, strategy in zip(skill_opinions, strategy_opinions)
}
return AggregationData(
skill_opinions=skill_opinions,
weights=weights,
skill_names=skill_ids,
strategy_opinions=strategy_opinions,
weighted_score=weighted_score,
weighted_confidence=weighted_confidence,
insufficient_evidence=insufficient_evidence,
conflicts=conflicts,
final_signal=final_signal,
individual_signals=individual_signals,
total_adjustment=total_adjustment,
)
@staticmethod
def build_consensus_opinion(
aggregation: AggregationData,
synthesis: Dict[str, Any],
) -> AgentOpinion:
reasoning_parts = [ reasoning_parts = [
f"Skill consensus from {len(skill_opinions)} skills " f"Skill consensus from {len(aggregation.skill_opinions)} skills "
f"({', '.join(skill_names)}): weighted score {weighted_score:.2f}/5.0" f"({', '.join(aggregation.skill_names)}): weighted score {aggregation.weighted_score:.2f}/5.0, "
f"consensus={synthesis['consensus_level']}, conflicts={synthesis['conflict_severity']}({synthesis['conflict_count']})"
] ]
for op, weight in zip(skill_opinions, weights): for opinion, weight in zip(aggregation.skill_opinions, aggregation.weights):
name = extract_skill_id(op.agent_name) or op.agent_name name = extract_skill_id(opinion.agent_name) or opinion.agent_name
reasoning_parts.append(f" - {name}: {op.signal} ({op.confidence:.0%}) weight={weight:.2f}") reasoning_parts.append(f" - {name}: {opinion.signal} ({opinion.confidence:.0%}) weight={weight:.2f}")
return AgentOpinion( return AgentOpinion(
agent_name=SKILL_CONSENSUS_AGENT_NAME, agent_name=SKILL_CONSENSUS_AGENT_NAME,
signal=final_signal, signal=aggregation.final_signal,
confidence=min(1.0, weighted_confidence), confidence=synthesis["confidence"],
reasoning="\n".join(reasoning_parts), reasoning="\n".join(reasoning_parts),
raw_data={ raw_data={
"weighted_score": round(weighted_score, 2), "weighted_score": round(aggregation.weighted_score, 2),
"total_adjustment": total_adjustment, "total_adjustment": aggregation.total_adjustment,
"skill_count": len(skill_opinions), "skill_count": len(aggregation.skill_opinions),
"individual_signals": { "individual_signals": aggregation.individual_signals,
op.agent_name: {"signal": op.signal, "confidence": op.confidence} "strategy_synthesis": synthesis,
for op in skill_opinions "conflicts": synthesis["conflicts"],
}, "conflict_count": synthesis["conflict_count"],
"conflict_severity": synthesis["conflict_severity"],
"consensus_level": synthesis["consensus_level"],
}, },
) )

255
src/agent/skills/engine.py Normal file
View File

@@ -0,0 +1,255 @@
# -*- coding: utf-8 -*-
"""
StrategyEngine — authoritative multi-strategy pipeline facade.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
from src.agent.protocols import AgentOpinion, normalize_strategy_signal
from src.agent.skills.aggregator import AggregationData, SkillAggregator
from src.agent.skills.defaults import (
LEGACY_STRATEGY_CONSENSUS_AGENT_NAME,
SKILL_CONSENSUS_AGENT_NAME,
extract_skill_id,
is_skill_agent_name,
)
from src.agent.skills.synthesis import StrategySynthesizer
logger = logging.getLogger(__name__)
class StrategyResultStatus(str, Enum):
CONSENSUS = "consensus"
NO_CONSENSUS = "no_consensus"
NO_SKILLS = "no_skills"
@dataclass
class EvidencePartition:
valid_skill_opinions: List[AgentOpinion] = field(default_factory=list)
invalid_records: List[Dict[str, Any]] = field(default_factory=list)
invalid_count: int = 0
non_skill_opinions: List[AgentOpinion] = field(default_factory=list)
evidence_opinions: List[AgentOpinion] = field(default_factory=list)
@dataclass
class StrategyResult:
status: StrategyResultStatus = StrategyResultStatus.NO_SKILLS
synthesis_dict: Optional[Dict[str, Any]] = None
consensus_opinion: Optional[AgentOpinion] = None
skill_consensus_data: Optional[Dict[str, Any]] = None
valid_skill_opinions: List[AgentOpinion] = field(default_factory=list)
non_skill_opinions: List[AgentOpinion] = field(default_factory=list)
evidence_opinions: List[AgentOpinion] = field(default_factory=list)
invalid_records: List[Dict[str, Any]] = field(default_factory=list)
invalid_count: int = 0
class StrategyEngine:
"""Centralize the skill-opinion evidence chain into one facade."""
def __init__(self, aggregator: Optional[SkillAggregator] = None) -> None:
self.aggregator = aggregator or SkillAggregator()
def partition_only(self, opinions: List[AgentOpinion]) -> EvidencePartition:
valid_skill_opinions: List[AgentOpinion] = []
invalid_records: List[Dict[str, Any]] = []
non_skill_opinions: List[AgentOpinion] = []
evidence_opinions: List[AgentOpinion] = []
for opinion in opinions:
if opinion.agent_name in {SKILL_CONSENSUS_AGENT_NAME, LEGACY_STRATEGY_CONSENSUS_AGENT_NAME}:
continue
if not is_skill_agent_name(opinion.agent_name):
non_skill_opinions.append(opinion)
evidence_opinions.append(opinion)
continue
raw_data = opinion.raw_data if isinstance(opinion.raw_data, dict) else {}
raw_signal = opinion.signal if opinion.signal else raw_data.get("signal")
canonical, invalid_signal, original_signal = normalize_strategy_signal(raw_signal)
if raw_signal is None or (isinstance(raw_signal, str) and not raw_signal.strip()):
invalid_records.append({
"agent_name": opinion.agent_name,
"raw_signal": None if raw_signal is None else raw_signal,
"confidence": opinion.confidence,
"reason": "missing_signal",
})
logger.info(
"[StrategyEngine] invalid skill opinion moved to diagnostics: agent=%s raw_signal=%r reason=%s",
opinion.agent_name,
raw_signal,
"missing_signal",
)
continue
if invalid_signal:
invalid_records.append({
"agent_name": opinion.agent_name,
"raw_signal": original_signal,
"confidence": opinion.confidence,
"reason": "unrecognized_signal",
})
logger.info(
"[StrategyEngine] invalid skill opinion moved to diagnostics: agent=%s raw_signal=%r reason=%s",
opinion.agent_name,
original_signal,
"unrecognized_signal",
)
continue
if canonical != opinion.signal:
normalized_raw = dict(raw_data)
normalized_raw.setdefault("original_signal", original_signal)
normalized_raw["normalized_signal"] = canonical
opinion = AgentOpinion(
agent_name=opinion.agent_name,
signal=canonical,
confidence=opinion.confidence,
reasoning=opinion.reasoning,
key_levels=dict(opinion.key_levels or {}),
raw_data=normalized_raw,
timestamp=opinion.timestamp,
)
valid_skill_opinions.append(opinion)
evidence_opinions.append(opinion)
return EvidencePartition(
valid_skill_opinions=valid_skill_opinions,
invalid_records=invalid_records,
invalid_count=len(invalid_records),
non_skill_opinions=non_skill_opinions,
evidence_opinions=evidence_opinions,
)
def process(self, opinions: List[AgentOpinion]) -> StrategyResult:
return self.process_partition(self.partition_only(opinions))
def process_partition(self, partition: EvidencePartition) -> StrategyResult:
if not partition.valid_skill_opinions:
if partition.invalid_count > 0:
stub = self._build_no_consensus_stub(partition.invalid_count)
return StrategyResult(
status=StrategyResultStatus.NO_CONSENSUS,
synthesis_dict=stub,
skill_consensus_data={
"signal": "hold",
"confidence": 0.0,
"reasoning": "",
"raw_data": {},
"strategy_synthesis": stub,
"conflicts": [],
},
valid_skill_opinions=[],
non_skill_opinions=list(partition.non_skill_opinions),
invalid_records=list(partition.invalid_records),
invalid_count=partition.invalid_count,
)
return StrategyResult(
status=StrategyResultStatus.NO_SKILLS,
synthesis_dict=None,
valid_skill_opinions=[],
non_skill_opinions=list(partition.non_skill_opinions),
invalid_records=list(partition.invalid_records),
invalid_count=partition.invalid_count,
)
aggregation = self.aggregator.calculate(partition.valid_skill_opinions)
if aggregation is None:
return StrategyResult(
status=StrategyResultStatus.NO_SKILLS,
synthesis_dict=None,
valid_skill_opinions=list(partition.valid_skill_opinions),
non_skill_opinions=list(partition.non_skill_opinions),
invalid_records=list(partition.invalid_records),
invalid_count=partition.invalid_count,
)
synthesis = StrategySynthesizer().synthesize(
aggregation.strategy_opinions,
weighted_score=aggregation.weighted_score,
final_signal=aggregation.final_signal,
weighted_confidence=aggregation.weighted_confidence,
conflicts=aggregation.conflicts,
insufficient_evidence=aggregation.insufficient_evidence,
invalid_count=partition.invalid_count,
)
consensus_opinion = self._build_consensus_opinion(aggregation, synthesis)
return StrategyResult(
status=StrategyResultStatus.CONSENSUS,
synthesis_dict=synthesis,
consensus_opinion=consensus_opinion,
skill_consensus_data={
"signal": consensus_opinion.signal,
"confidence": consensus_opinion.confidence,
"reasoning": consensus_opinion.reasoning,
"raw_data": consensus_opinion.raw_data,
"strategy_synthesis": synthesis,
"conflicts": synthesis.get("conflicts", []),
},
valid_skill_opinions=list(partition.valid_skill_opinions),
non_skill_opinions=list(partition.non_skill_opinions),
invalid_records=list(partition.invalid_records),
invalid_count=partition.invalid_count,
)
@staticmethod
def _build_consensus_opinion(aggregation: AggregationData, synthesis: Dict[str, Any]) -> AgentOpinion:
reasoning_parts = [
f"Skill consensus from {len(aggregation.skill_opinions)} skills "
f"({', '.join(aggregation.skill_names)}): weighted score {aggregation.weighted_score:.2f}/5.0, "
f"consensus={synthesis['consensus_level']}, conflicts={synthesis['conflict_severity']}({synthesis['conflict_count']})"
]
for opinion, weight in zip(aggregation.skill_opinions, aggregation.weights):
name = extract_skill_id(opinion.agent_name) or opinion.agent_name
reasoning_parts.append(f" - {name}: {opinion.signal} ({opinion.confidence:.0%}) weight={weight:.2f}")
return AgentOpinion(
agent_name=SKILL_CONSENSUS_AGENT_NAME,
signal=aggregation.final_signal,
confidence=synthesis["confidence"],
reasoning="\n".join(reasoning_parts),
raw_data={
"weighted_score": round(aggregation.weighted_score, 2),
"total_adjustment": aggregation.total_adjustment,
"skill_count": len(aggregation.skill_opinions),
"individual_signals": aggregation.individual_signals,
"strategy_synthesis": synthesis,
"conflicts": synthesis["conflicts"],
"conflict_count": synthesis["conflict_count"],
"conflict_severity": synthesis["conflict_severity"],
"consensus_level": synthesis["consensus_level"],
},
)
@staticmethod
def _build_no_consensus_stub(invalid_count: int) -> Dict[str, Any]:
return {
"final_signal": "hold",
"weighted_score": 3.0,
"confidence": 0.0,
"original_confidence": 0.0,
"conflict_count": 0,
"conflict_severity": "none",
"conflicts": [],
"supporting_skills": [],
"opposing_skills": [],
"consensus_level": "insufficient",
"summary_key": "strategy_synthesis.no_conflicts",
"summary_params": {
"opinion_count": 0,
"total_opinion_count": invalid_count,
"invalid_opinion_count": invalid_count,
"final_signal": "hold",
"consensus_level": "insufficient",
"conflict_severity": "none",
"conflict_count": 0,
},
}

View File

@@ -108,7 +108,7 @@ Return **only** a JSON object:
return AgentOpinion( return AgentOpinion(
agent_name=self.agent_name, agent_name=self.agent_name,
signal=parsed.get("signal", "hold"), signal=parsed.get("signal"), # None if missing — no silent default
confidence=float(parsed.get("confidence", 0.5)), confidence=float(parsed.get("confidence", 0.5)),
reasoning=parsed.get("reasoning", ""), reasoning=parsed.get("reasoning", ""),
raw_data=parsed, raw_data=parsed,

View File

@@ -0,0 +1,397 @@
# -*- coding: utf-8 -*-
"""
Strategy synthesis helpers for skill-agent consensus.
"""
from __future__ import annotations
from dataclasses import asdict
from typing import Any, Dict, Iterable, List, Optional
from src.agent.protocols import (
AgentOpinion,
StrategyConflict,
StrategyOpinion,
normalize_strategy_signal,
strategy_signal_score,
)
from src.agent.skills.defaults import extract_skill_id
_SEVERITY_RANK = {"none": 0, "low": 1, "medium": 2, "high": 3}
def strategy_opinion_from_agent_opinion(opinion: AgentOpinion) -> StrategyOpinion:
raw_data = opinion.raw_data if isinstance(opinion.raw_data, dict) else {}
skill_id = str(raw_data.get("skill_id") or extract_skill_id(opinion.agent_name) or opinion.agent_name)
key_levels = opinion.key_levels or raw_data.get("key_levels") or {}
if not isinstance(key_levels, dict):
key_levels = {}
# Determine raw signal: prefer opinion.signal, fall back to raw_data
raw_signal = opinion.signal if opinion.signal else raw_data.get("signal")
# If signal is truly missing (None), mark as invalid — no silent hold default
if raw_signal is None:
return StrategyOpinion(
skill_id=skill_id,
agent_name=opinion.agent_name,
signal="hold",
confidence=opinion.confidence,
reasoning=str(opinion.reasoning or raw_data.get("reasoning") or ""),
score_adjustment=_as_float(raw_data.get("score_adjustment"), 0.0),
conditions_met=_as_string_list(raw_data.get("conditions_met")),
conditions_missed=_as_string_list(raw_data.get("conditions_missed")),
key_levels=key_levels,
raw_data={**raw_data, "normalized_signal": "hold", "original_signal": "", "invalid_signal": True},
original_signal="",
invalid_signal=True, # missing signal = invalid
)
signal, invalid_signal, original_signal = normalize_strategy_signal(raw_signal)
normalized_raw_data = dict(raw_data)
normalized_raw_data["normalized_signal"] = signal
normalized_raw_data["original_signal"] = original_signal
normalized_raw_data["invalid_signal"] = invalid_signal
return StrategyOpinion(
skill_id=skill_id,
agent_name=opinion.agent_name,
signal=signal,
confidence=opinion.confidence,
reasoning=str(opinion.reasoning or raw_data.get("reasoning") or ""),
score_adjustment=_as_float(raw_data.get("score_adjustment"), 0.0),
conditions_met=_as_string_list(raw_data.get("conditions_met")),
conditions_missed=_as_string_list(raw_data.get("conditions_missed")),
key_levels=key_levels,
raw_data=normalized_raw_data,
original_signal=original_signal,
invalid_signal=invalid_signal,
)
class ConflictDetector:
"""Detect deterministic disagreements among strategy opinions."""
def detect(
self,
opinions: List[StrategyOpinion],
*,
final_signal: Optional[str] = None,
) -> List[StrategyConflict]:
valid_opinions = [op for op in opinions if not op.invalid_signal]
if len(valid_opinions) < 2:
return []
conflicts: List[StrategyConflict] = []
conflicts.extend(self._detect_directional_opposition(valid_opinions))
conflicts.extend(self._detect_wide_score_dispersion(valid_opinions))
if final_signal:
conflicts.extend(self._detect_high_confidence_dissent(valid_opinions, final_signal))
conflicts.extend(self._detect_adjustment_contradiction(valid_opinions))
return sorted(conflicts, key=self._sort_key)
@staticmethod
def _detect_directional_opposition(opinions: List[StrategyOpinion]) -> List[StrategyConflict]:
bullish = [op for op in opinions if strategy_signal_score(op.signal) >= 4.0]
bearish = [op for op in opinions if strategy_signal_score(op.signal) <= 2.0]
if not bullish or not bearish:
return []
max_bull_conf = max(op.confidence for op in bullish)
max_bear_conf = max(op.confidence for op in bearish)
severity = "high" if max_bull_conf >= 0.7 and max_bear_conf >= 0.7 else "medium"
participants = _unique_ids([*bullish, *bearish])
return [
StrategyConflict(
conflict_type="directional_opposition",
severity=severity,
description_key="strategy_conflict.directional_opposition",
participants=participants,
metadata={
"bullish": [op.skill_id for op in bullish],
"bearish": [op.skill_id for op in bearish],
"max_bullish_confidence": round(max_bull_conf, 4),
"max_bearish_confidence": round(max_bear_conf, 4),
},
)
]
@staticmethod
def _detect_wide_score_dispersion(opinions: List[StrategyOpinion]) -> List[StrategyConflict]:
scored = [(op, strategy_signal_score(op.signal)) for op in opinions]
min_score = min(score for _, score in scored)
max_score = max(score for _, score in scored)
spread = max_score - min_score
if spread < 2.0:
return []
participants = [op.skill_id for op, score in scored if score in {min_score, max_score}]
return [
StrategyConflict(
conflict_type="wide_score_dispersion",
severity="high" if spread >= 3.0 else "medium",
description_key="strategy_conflict.wide_score_dispersion",
participants=_unique_strings(participants),
metadata={"min_score": min_score, "max_score": max_score, "spread": spread},
)
]
@staticmethod
def _detect_high_confidence_dissent(
opinions: List[StrategyOpinion],
final_signal: str,
) -> List[StrategyConflict]:
final_score = strategy_signal_score(final_signal)
dissenters = [
op
for op in opinions
if op.confidence >= 0.75 and abs(strategy_signal_score(op.signal) - final_score) >= 2.0
]
if not dissenters:
return []
return [
StrategyConflict(
conflict_type="high_confidence_dissent",
severity="medium",
description_key="strategy_conflict.high_confidence_dissent",
participants=[op.skill_id for op in dissenters],
metadata={
"final_signal": final_signal,
"dissenters": [
{"skill_id": op.skill_id, "signal": op.signal, "confidence": round(op.confidence, 4)}
for op in dissenters
],
},
)
]
@staticmethod
def _detect_adjustment_contradiction(opinions: List[StrategyOpinion]) -> List[StrategyConflict]:
positive = [op for op in opinions if op.score_adjustment >= 8]
negative = [op for op in opinions if op.score_adjustment <= -8]
if not positive or not negative:
return []
max_positive = max(op.score_adjustment for op in positive)
min_negative = min(op.score_adjustment for op in negative)
severity = "high" if max_positive >= 15 and min_negative <= -15 else "medium"
return [
StrategyConflict(
conflict_type="adjustment_contradiction",
severity=severity,
description_key="strategy_conflict.adjustment_contradiction",
participants=_unique_ids([*positive, *negative]),
metadata={"max_positive_adjustment": max_positive, "min_negative_adjustment": min_negative},
)
]
@staticmethod
def _sort_key(conflict: StrategyConflict) -> tuple[int, str, str]:
return (-_SEVERITY_RANK.get(conflict.severity, 0), conflict.conflict_type, ",".join(conflict.participants))
class StrategySynthesizer:
"""Build an explainable synthesis payload for strategy consensus."""
def synthesize(
self,
opinions: List[StrategyOpinion],
*,
weighted_score: float,
final_signal: str,
weighted_confidence: float,
conflicts: List[StrategyConflict],
insufficient_evidence: bool = False,
invalid_count: int = 0,
) -> Dict[str, Any]:
conflict_severity = _highest_severity(conflicts)
adjusted_confidence = self.adjust_confidence(weighted_confidence, conflict_severity)
final_score = strategy_signal_score(final_signal)
supporting, opposing = self._group_opinions(opinions, final_score)
consensus_level = self._consensus_level(
opinions,
conflicts,
final_signal,
insufficient_evidence=insufficient_evidence,
)
valid_opinions = [op for op in opinions if not op.invalid_signal]
# When called directly (e.g. unit tests), infer invalid_count from the opinions
# list itself. In the E2E path the explicit partition value takes precedence.
invalid_count = max(invalid_count, sum(1 for op in opinions if op.invalid_signal))
return {
"final_signal": final_signal,
"weighted_score": round(weighted_score, 4),
"confidence": round(adjusted_confidence, 4),
"original_confidence": round(max(0.0, min(1.0, weighted_confidence)), 4),
"conflict_count": len(conflicts),
"conflict_severity": conflict_severity,
"conflicts": [_conflict_to_dict(conflict) for conflict in conflicts],
"supporting_skills": supporting,
"opposing_skills": opposing,
"consensus_level": consensus_level,
"summary_key": "strategy_synthesis.with_conflicts" if conflicts else "strategy_synthesis.no_conflicts",
"summary_params": {
"opinion_count": len(valid_opinions),
"total_opinion_count": len(valid_opinions) + invalid_count,
"invalid_opinion_count": invalid_count,
"final_signal": final_signal,
"consensus_level": consensus_level,
"conflict_severity": conflict_severity,
"conflict_count": len(conflicts),
},
}
@staticmethod
def adjust_confidence(confidence: float, conflict_severity: str) -> float:
adjusted = max(0.0, min(1.0, confidence))
if conflict_severity == "high":
adjusted *= 0.85
elif conflict_severity == "medium":
adjusted *= 0.93
return max(0.0, min(1.0, adjusted))
@staticmethod
def _group_opinions(
opinions: List[StrategyOpinion],
final_score: float,
) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
"""Dynamic bipartite grouping per multi-strategy-contract §动态二分阵营.
Every valid opinion falls into exactly one of supporting/opposing.
`neutral_skills` is deliberately removed to prevent render mismatch
("high consensus" + "supporting: none").
"""
supporting: List[Dict[str, Any]] = []
opposing: List[Dict[str, Any]] = []
for op in opinions:
if op.invalid_signal:
continue
score = strategy_signal_score(op.signal)
item = _opinion_to_item(op)
if final_score == 3.0:
if score == 3.0:
supporting.append(item)
else:
opposing.append(item)
else:
opinion_bullish = score > 3.0
opinion_bearish = score < 3.0
same_side = (
(opinion_bullish and final_score > 3.0)
or (opinion_bearish and final_score < 3.0)
)
if same_side and abs(score - final_score) <= 1.0:
supporting.append(item)
else:
opposing.append(item)
return supporting, opposing
@staticmethod
def _consensus_level(
opinions: List[StrategyOpinion],
conflicts: List[StrategyConflict],
final_signal: str,
*,
insufficient_evidence: bool = False,
) -> str:
"""Consensus level per multi-strategy-contract §共识度门槛.
Precedence:
1. aggregator-signalled insufficient (zero valid_weight_sum) → insufficient
2. ≤ 1 valid opinion → insufficient
3. sum(confidence) == 0 → insufficient
4. conflict_severity == "high" → low
5. aligned_ratio ≥ 2/3 且 conflict_count == 0 → high
6. conflict_severity == "medium" 且 aligned_ratio < 0.5 → low
7. 其余 → medium
"""
if insufficient_evidence:
return "insufficient"
valid_opinions = [op for op in opinions if not op.invalid_signal]
if len(valid_opinions) <= 1:
return "insufficient"
if sum(op.confidence for op in valid_opinions) == 0:
return "insufficient"
conflict_severity = _highest_severity(conflicts)
if conflict_severity == "high":
return "low"
final_score = strategy_signal_score(final_signal)
aligned = sum(
1
for op in valid_opinions
if (
strategy_signal_score(op.signal) == final_score
or (
abs(strategy_signal_score(op.signal) - final_score) <= 1.0
and (
(strategy_signal_score(op.signal) > 3.0 and final_score > 3.0)
or (strategy_signal_score(op.signal) < 3.0 and final_score < 3.0)
)
)
)
)
aligned_ratio = aligned / len(valid_opinions)
if not conflicts and aligned_ratio >= 2 / 3:
return "high"
if conflict_severity == "medium" and aligned_ratio < 0.5:
return "low"
return "medium"
def _as_float(value: Any, default: float) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default
def _as_string_list(value: Any) -> List[str]:
if not isinstance(value, Iterable) or isinstance(value, (str, bytes, dict)):
return []
return [str(item) for item in value if item is not None]
def _unique_ids(opinions: Iterable[StrategyOpinion]) -> List[str]:
return _unique_strings(op.skill_id for op in opinions)
def _unique_strings(values: Iterable[str]) -> List[str]:
result: List[str] = []
for value in values:
if value and value not in result:
result.append(value)
return result
def _highest_severity(conflicts: List[StrategyConflict]) -> str:
if not conflicts:
return "none"
return max((conflict.severity for conflict in conflicts), key=lambda severity: _SEVERITY_RANK.get(severity, 0))
def _opinion_to_item(opinion: StrategyOpinion) -> Dict[str, Any]:
return {
"skill_id": opinion.skill_id,
"agent_name": opinion.agent_name,
"signal": opinion.signal,
"confidence": round(opinion.confidence, 4),
"reasoning": opinion.reasoning,
"score_adjustment": opinion.score_adjustment,
"conditions_met": opinion.conditions_met,
"invalid_signal": opinion.invalid_signal,
}
def _conflict_to_dict(conflict: StrategyConflict) -> Dict[str, Any]:
payload = asdict(conflict)
payload.pop("description", None)
return payload

View File

@@ -44,8 +44,16 @@ from src.report_language import (
get_chip_unavailable_reason, get_chip_unavailable_reason,
is_chip_structure_unavailable, is_chip_structure_unavailable,
localize_chip_health, localize_chip_health,
localize_conflict_severity,
localize_consensus_level,
localize_strategy_signal,
localize_strategy_skill,
localize_strategy_conflict_description,
localize_strategy_synthesis_summary,
localize_trend_prediction, localize_trend_prediction,
normalize_report_language, normalize_report_language,
normalize_strategy_synthesis_payload,
strategy_invalid_opinion_count,
) )
from src.schemas.decision_action import ( from src.schemas.decision_action import (
display_action_fields_for_result, display_action_fields_for_result,
@@ -101,6 +109,74 @@ def _safe_float(value: Any) -> Optional[float]:
except (TypeError, ValueError): except (TypeError, ValueError):
return None return None
def _format_strategy_skill_items(items: Any, report_language: str = "zh") -> str:
none_text = get_report_labels(report_language).get("none_label", "None")
if not isinstance(items, list):
return none_text
formatted: List[str] = []
for item in items:
if not isinstance(item, dict):
continue
skill_id = str(item.get("skill_id") or "").strip()
signal = str(item.get("signal") or "").strip()
confidence = item.get("confidence")
if not skill_id:
continue
suffix = f"/{localize_strategy_signal(signal, report_language)}" if signal else ""
if isinstance(confidence, (int, float)):
suffix += f"/{confidence:.0%}"
formatted.append(f"{localize_strategy_skill(skill_id, report_language)}{suffix}")
return "".join(formatted) if formatted else none_text
def _append_strategy_synthesis_block(lines: List[str], strategy_synthesis: Any, labels: Dict[str, str], report_language: str) -> None:
strategy_synthesis = normalize_strategy_synthesis_payload(strategy_synthesis)
if not strategy_synthesis:
return
confidence = strategy_synthesis.get("confidence")
confidence_text = f"{confidence:.0%}" if isinstance(confidence, (int, float)) else "N/A"
lines.extend([
f"### 🧩 {labels['strategy_synthesis_heading']}",
"",
(
f"- {labels['strategy_final_signal_label']}: "
f"{localize_strategy_signal(strategy_synthesis.get('final_signal', 'N/A'), report_language)} | "
f"{labels['strategy_consensus_level_label']}: "
f"{localize_consensus_level(strategy_synthesis.get('consensus_level', 'N/A'), report_language)} | "
f"{labels['strategy_conflict_label']}: "
f"{localize_conflict_severity(strategy_synthesis.get('conflict_severity', 'none'), report_language)} "
f"({strategy_synthesis.get('conflict_count', 0)}) | "
f"{labels['strategy_confidence_label']}: {confidence_text}"
),
])
summary = localize_strategy_synthesis_summary(strategy_synthesis, report_language)
if summary:
lines.append(f"- {labels['strategy_summary_label']}: {summary}")
lines.append(
f"- {labels['strategy_supporting_skills_label']}: "
f"{_format_strategy_skill_items(strategy_synthesis.get('supporting_skills'), report_language)}"
)
lines.append(
f"- {labels['strategy_opposing_skills_label']}: "
f"{_format_strategy_skill_items(strategy_synthesis.get('opposing_skills'), report_language)}"
)
invalid_opinion_count = strategy_invalid_opinion_count(strategy_synthesis)
if invalid_opinion_count:
invalid_label = labels.get("strategy_invalid_opinions_label", "")
if invalid_label:
lines.append(f"- {invalid_label.format(count=invalid_opinion_count)}")
for conflict in (strategy_synthesis.get("conflicts") or [])[:3]:
if isinstance(conflict, dict) and conflict.get("conflict_type"):
participants = conflict.get("participants") or []
participant_text = "".join(localize_strategy_skill(participant, report_language) for participant in participants)
suffix = f"{participant_text}" if participant_text else ""
lines.append(
f"- {localize_conflict_severity(conflict.get('severity', 'medium'), report_language)}: "
f"{localize_strategy_conflict_description(conflict.get('conflict_type'), report_language)}{suffix}"
)
lines.append("")
if TYPE_CHECKING: if TYPE_CHECKING:
from src.analyzer import AnalysisResult from src.analyzer import AnalysisResult
@@ -1445,6 +1521,12 @@ class NotificationService(
report_lines.append(f"**🐻 {labels['strongest_bearish_signal_label']}**: {signal_attr['strongest_bearish_signal']}") report_lines.append(f"**🐻 {labels['strongest_bearish_signal_label']}**: {signal_attr['strongest_bearish_signal']}")
report_lines.append("") report_lines.append("")
# ========== 多策略综合 ==========
strategy_synthesis = normalize_strategy_synthesis_payload(
dashboard.get('strategy_synthesis') if dashboard else None
)
_append_strategy_synthesis_block(report_lines, strategy_synthesis, labels, report_language)
# 财务摘要 / 股东回报 / 关联板块(数据缺失时自动隐藏对应小节) # 财务摘要 / 股东回报 / 关联板块(数据缺失时自动隐藏对应小节)
self._append_fundamental_blocks(report_lines, result) self._append_fundamental_blocks(report_lines, result)
@@ -1634,6 +1716,32 @@ class NotificationService(
lines.append(f"💼 {labels['has_position_label']}: {has_pos[:50]}") lines.append(f"💼 {labels['has_position_label']}: {has_pos[:50]}")
lines.append("") lines.append("")
# 多策略综合
strategy_synthesis = normalize_strategy_synthesis_payload(
dashboard.get('strategy_synthesis') if dashboard else None
)
if strategy_synthesis:
lines.append(
f"🧩 **{labels['strategy_synthesis_heading']}**: "
f"{localize_strategy_signal(strategy_synthesis.get('final_signal', 'N/A'), report_language)} | "
f"{labels['strategy_consensus_level_label']} "
f"{localize_consensus_level(strategy_synthesis.get('consensus_level', 'N/A'), report_language)} | "
f"{labels['strategy_conflict_label']} "
f"{localize_conflict_severity(strategy_synthesis.get('conflict_severity', 'none'), report_language)}"
f"({strategy_synthesis.get('conflict_count', 0)})"
)
invalid_count = strategy_invalid_opinion_count(strategy_synthesis)
if invalid_count:
lines.append(
labels.get(
'strategy_invalid_opinions_label', ''
).format(count=invalid_count)
)
summary = localize_strategy_synthesis_summary(strategy_synthesis, report_language)
if summary:
lines.append(summary[:80])
lines.append("")
# 检查清单简化版 # 检查清单简化版
checklist = battle.get('action_checklist', []) if battle else [] checklist = battle.get('action_checklist', []) if battle else []
if checklist: if checklist:

View File

@@ -133,6 +133,133 @@ _CONFIDENCE_LEVEL_TRANSLATIONS = {
"low": {"zh": "", "en": "Low", "ko": "낮음"}, "low": {"zh": "", "en": "Low", "ko": "낮음"},
} }
_STRATEGY_SIGNAL_CANONICAL_MAP = {
"strong buy": "strong_buy",
"strong_buy": "strong_buy",
"强烈买入": "strong_buy",
"buy": "buy",
"买入": "buy",
"hold": "hold",
"持有": "hold",
"sell": "sell",
"卖出": "sell",
"strong sell": "strong_sell",
"strong_sell": "strong_sell",
"强烈卖出": "strong_sell",
}
_STRATEGY_SIGNAL_TRANSLATIONS = {
"strong_buy": {"zh": "强烈买入", "en": "Strong Buy", "ko": "적극 매수"},
"buy": {"zh": "买入", "en": "Buy", "ko": "매수"},
"hold": {"zh": "持有", "en": "Hold", "ko": "보유"},
"sell": {"zh": "卖出", "en": "Sell", "ko": "매도"},
"strong_sell": {"zh": "强烈卖出", "en": "Strong Sell", "ko": "적극 매도"},
}
_CONSENSUS_LEVEL_CANONICAL_MAP = {
"high": "high",
"": "high",
"medium": "medium",
"": "medium",
"low": "low",
"": "low",
"insufficient": "insufficient",
"证据不足": "insufficient",
"Insufficient": "insufficient",
"증거 부족": "insufficient",
}
_CONSENSUS_LEVEL_TRANSLATIONS = {
"high": {"zh": "", "en": "High", "ko": "높음"},
"medium": {"zh": "", "en": "Medium", "ko": "보통"},
"low": {"zh": "", "en": "Low", "ko": "낮음"},
"insufficient": {"zh": "证据不足", "en": "Insufficient", "ko": "증거 부족"},
}
_CONFLICT_SEVERITY_CANONICAL_MAP = {
"none": "none",
"": "none",
"low": "low",
"": "low",
"medium": "medium",
"": "medium",
"high": "high",
"": "high",
}
_CONFLICT_SEVERITY_TRANSLATIONS = {
"none": {"zh": "", "en": "None", "ko": "없음"},
"low": {"zh": "", "en": "Low", "ko": "낮음"},
"medium": {"zh": "", "en": "Medium", "ko": "보통"},
"high": {"zh": "", "en": "High", "ko": "높음"},
}
_STRATEGY_SKILL_CANONICAL_MAP = {
"bull trend": "bull_trend",
"bull_trend": "bull_trend",
"默认多头趋势": "bull_trend",
"hot theme": "hot_theme",
"hot_theme": "hot_theme",
"热点题材": "hot_theme",
"volume breakout": "volume_breakout",
"volume_breakout": "volume_breakout",
"放量突破": "volume_breakout",
"ma golden cross": "ma_golden_cross",
"ma_golden_cross": "ma_golden_cross",
"均线金叉": "ma_golden_cross",
"growth quality": "growth_quality",
"growth_quality": "growth_quality",
"成长质量": "growth_quality",
"bottom volume": "bottom_volume",
"bottom_volume": "bottom_volume",
"底部放量": "bottom_volume",
"box oscillation": "box_oscillation",
"box_oscillation": "box_oscillation",
"箱体震荡": "box_oscillation",
"chan theory": "chan_theory",
"chan_theory": "chan_theory",
"缠论结构": "chan_theory",
"dragon head": "dragon_head",
"dragon_head": "dragon_head",
"龙头战法": "dragon_head",
"emotion cycle": "emotion_cycle",
"emotion_cycle": "emotion_cycle",
"情绪周期": "emotion_cycle",
"event driven": "event_driven",
"event_driven": "event_driven",
"事件驱动": "event_driven",
"expectation repricing": "expectation_repricing",
"expectation_repricing": "expectation_repricing",
"预期重估": "expectation_repricing",
"one yang three yin": "one_yang_three_yin",
"one_yang_three_yin": "one_yang_three_yin",
"一阳三阴": "one_yang_three_yin",
"shrink pullback": "shrink_pullback",
"shrink_pullback": "shrink_pullback",
"缩量回踩": "shrink_pullback",
"wave theory": "wave_theory",
"wave_theory": "wave_theory",
"波浪理论": "wave_theory",
}
_STRATEGY_SKILL_TRANSLATIONS = {
"bull_trend": {"zh": "默认多头趋势", "en": "Bull Trend", "ko": "기본 상승 추세"},
"hot_theme": {"zh": "热点题材", "en": "Hot Theme", "ko": "핫 테마"},
"volume_breakout": {"zh": "放量突破", "en": "Volume Breakout", "ko": "거래량 돌파"},
"ma_golden_cross": {"zh": "均线金叉", "en": "MA Golden Cross", "ko": "이평선 골든크로스"},
"growth_quality": {"zh": "成长质量", "en": "Growth Quality", "ko": "성장 품질"},
"bottom_volume": {"zh": "底部放量", "en": "Bottom Volume", "ko": "저점 거래량"},
"box_oscillation": {"zh": "箱体震荡", "en": "Box Oscillation", "ko": "박스권 등락"},
"chan_theory": {"zh": "缠论结构", "en": "Chan Theory", "ko": "찬 이론 구조"},
"dragon_head": {"zh": "龙头战法", "en": "Dragon Head", "ko": "대장주 전략"},
"emotion_cycle": {"zh": "情绪周期", "en": "Emotion Cycle", "ko": "심리 사이클"},
"event_driven": {"zh": "事件驱动", "en": "Event Driven", "ko": "이벤트 드리븐"},
"expectation_repricing": {"zh": "预期重估", "en": "Expectation Repricing", "ko": "기대 재평가"},
"one_yang_three_yin": {"zh": "一阳三阴", "en": "One Yang Three Yin", "ko": "일양삼음"},
"shrink_pullback": {"zh": "缩量回踩", "en": "Shrink Pullback", "ko": "거래량 축소 눌림"},
"wave_theory": {"zh": "波浪理论", "en": "Wave Theory", "ko": "파동 이론"},
}
_CHIP_HEALTH_CANONICAL_MAP = { _CHIP_HEALTH_CANONICAL_MAP = {
"健康": "healthy", "健康": "healthy",
"healthy": "healthy", "healthy": "healthy",
@@ -278,6 +405,7 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = {
"bullish_alignment_label": "多头排列", "bullish_alignment_label": "多头排列",
"yes_label": "", "yes_label": "",
"no_label": "", "no_label": "",
"none_label": "",
"trend_strength_label": "趋势强度", "trend_strength_label": "趋势强度",
"price_metrics_label": "价格指标", "price_metrics_label": "价格指标",
"ma5_label": "MA5", "ma5_label": "MA5",
@@ -356,6 +484,15 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = {
"market_conditions_label": "市场环境", "market_conditions_label": "市场环境",
"strongest_bullish_signal_label": "最强看多信号", "strongest_bullish_signal_label": "最强看多信号",
"strongest_bearish_signal_label": "最强看空信号", "strongest_bearish_signal_label": "最强看空信号",
"strategy_synthesis_heading": "多策略综合",
"strategy_final_signal_label": "综合信号",
"strategy_consensus_level_label": "共识度",
"strategy_conflict_label": "冲突",
"strategy_confidence_label": "置信度",
"strategy_summary_label": "综合说明",
"strategy_supporting_skills_label": "支持策略",
"strategy_opposing_skills_label": "反方策略",
"strategy_invalid_opinions_label": "另有 {count} 个策略解析失败",
}, },
"en": { "en": {
"dashboard_title": "Decision Dashboard", "dashboard_title": "Decision Dashboard",
@@ -402,6 +539,7 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = {
"bullish_alignment_label": "Bullish Alignment", "bullish_alignment_label": "Bullish Alignment",
"yes_label": "Yes", "yes_label": "Yes",
"no_label": "No", "no_label": "No",
"none_label": "None",
"trend_strength_label": "Trend Strength", "trend_strength_label": "Trend Strength",
"price_metrics_label": "Price Metrics", "price_metrics_label": "Price Metrics",
"ma5_label": "MA5", "ma5_label": "MA5",
@@ -480,6 +618,15 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = {
"market_conditions_label": "Market Conditions", "market_conditions_label": "Market Conditions",
"strongest_bullish_signal_label": "Strongest Bullish Signal", "strongest_bullish_signal_label": "Strongest Bullish Signal",
"strongest_bearish_signal_label": "Strongest Bearish Signal", "strongest_bearish_signal_label": "Strongest Bearish Signal",
"strategy_synthesis_heading": "Strategy Synthesis",
"strategy_final_signal_label": "Final Signal",
"strategy_consensus_level_label": "Consensus",
"strategy_conflict_label": "Conflict",
"strategy_confidence_label": "Confidence",
"strategy_summary_label": "Summary",
"strategy_supporting_skills_label": "Supporting Strategies",
"strategy_opposing_skills_label": "Opposing Strategies",
"strategy_invalid_opinions_label": "{count} additional strategies failed to produce valid signals",
}, },
"ko": { "ko": {
"dashboard_title": "결정 대시보드", "dashboard_title": "결정 대시보드",
@@ -526,6 +673,7 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = {
"bullish_alignment_label": "정배열", "bullish_alignment_label": "정배열",
"yes_label": "", "yes_label": "",
"no_label": "아니오", "no_label": "아니오",
"none_label": "없음",
"trend_strength_label": "추세 강도", "trend_strength_label": "추세 강도",
"price_metrics_label": "가격 지표", "price_metrics_label": "가격 지표",
"ma5_label": "MA5", "ma5_label": "MA5",
@@ -604,6 +752,15 @@ _REPORT_LABELS: Dict[str, Dict[str, str]] = {
"market_conditions_label": "시장 환경", "market_conditions_label": "시장 환경",
"strongest_bullish_signal_label": "최강 상승 신호", "strongest_bullish_signal_label": "최강 상승 신호",
"strongest_bearish_signal_label": "최강 하락 신호", "strongest_bearish_signal_label": "최강 하락 신호",
"strategy_synthesis_heading": "전략 종합",
"strategy_final_signal_label": "종합 신호",
"strategy_consensus_level_label": "공감도",
"strategy_conflict_label": "충돌",
"strategy_confidence_label": "신뢰도",
"strategy_summary_label": "종합 설명",
"strategy_supporting_skills_label": "지지 전략",
"strategy_opposing_skills_label": "반대 전략",
"strategy_invalid_opinions_label": "추가로 {count}개 전략이 유효한 신호를 생성하지 못했습니다",
}, },
} }
@@ -839,6 +996,46 @@ def localize_confidence_level(value: Any, language: Optional[str]) -> str:
) )
def localize_strategy_signal(value: Any, language: Optional[str]) -> str:
"""Translate strategy signal labels when recognized."""
return _translate_from_map(
value,
language,
canonical_map=_STRATEGY_SIGNAL_CANONICAL_MAP,
translations=_STRATEGY_SIGNAL_TRANSLATIONS,
)
def localize_consensus_level(value: Any, language: Optional[str]) -> str:
"""Translate strategy consensus levels when recognized."""
return _translate_from_map(
value,
language,
canonical_map=_CONSENSUS_LEVEL_CANONICAL_MAP,
translations=_CONSENSUS_LEVEL_TRANSLATIONS,
)
def localize_conflict_severity(value: Any, language: Optional[str]) -> str:
"""Translate strategy conflict severity when recognized."""
return _translate_from_map(
value,
language,
canonical_map=_CONFLICT_SEVERITY_CANONICAL_MAP,
translations=_CONFLICT_SEVERITY_TRANSLATIONS,
)
def localize_strategy_skill(value: Any, language: Optional[str]) -> str:
"""Translate strategy skill names when recognized."""
return _translate_from_map(
value,
language,
canonical_map=_STRATEGY_SKILL_CANONICAL_MAP,
translations=_STRATEGY_SKILL_TRANSLATIONS,
)
def localize_chip_health(value: Any, language: Optional[str]) -> str: def localize_chip_health(value: Any, language: Optional[str]) -> str:
"""Translate chip health labels between Chinese and English when recognized.""" """Translate chip health labels between Chinese and English when recognized."""
return _translate_from_map( return _translate_from_map(
@@ -879,6 +1076,120 @@ def is_chip_structure_unavailable(chip_data: Any) -> bool:
return all(is_chip_placeholder_value(value) for value in chip_data.values()) return all(is_chip_placeholder_value(value) for value in chip_data.values())
def localize_strategy_conflict_description(conflict_type: Any, language: Optional[str]) -> str:
"""Translate strategy conflict type into a display sentence at render boundaries."""
lang = normalize_report_language(language)
key = str(conflict_type or "").strip()
translations = {
"directional_opposition": {
"zh": "策略方向出现对立:部分策略看多,部分策略看空,综合结论需要降低确定性。",
"en": "Strategy directions diverge: some strategies are bullish while others are bearish, so conviction should be reduced.",
"ko": "전략 방향이 엇갈립니다. 일부 전략은 상승을, 일부 전략은 하락을 보며 확신도를 낮춰야 합니다.",
},
"wide_score_dispersion": {
"zh": "策略信号分数分布较宽,说明多策略对行情结构存在明显分歧。",
"en": "Strategy signal scores are widely dispersed, indicating meaningful disagreement on market structure.",
"ko": "전략 신호 점수 분포가 넓어 시장 구조에 대한 전략 간 이견이 큽니다.",
},
"high_confidence_dissent": {
"zh": "存在高置信少数派策略与综合信号明显不一致,应保留反方观点。",
"en": "A high-confidence minority strategy materially disagrees with the final signal and should be kept as a dissenting view.",
"ko": "높은 확신도의 소수 전략이 종합 신호와 크게 달라 반대 관점으로 보존해야 합니다.",
},
"adjustment_contradiction": {
"zh": "策略加减分方向相互矛盾,说明不同策略对同一标的的边际评分分歧较大。",
"en": "Strategy score adjustments contradict each other, showing large disagreement in marginal scoring.",
"ko": "전략별 점수 조정 방향이 서로 충돌해 동일 종목의 한계 평가 차이가 큽니다.",
},
}
localized = translations.get(key, {})
return localized.get(lang) or localized.get("zh") or key
def normalize_strategy_synthesis_payload(value: Any) -> Dict[str, Any]:
"""Return a renderer-safe copy of a strategy synthesis payload.
Historical records and external callers may contain pre-contract values.
Renderers must treat a malformed top-level payload as absent and must not
iterate malformed collection fields as strategy/conflict entries.
"""
if not isinstance(value, dict) or not value:
return {}
payload = dict(value)
for key in ("supporting_skills", "opposing_skills", "conflicts"):
items = payload.get(key)
payload[key] = (
[item for item in items if isinstance(item, dict)]
if isinstance(items, list)
else []
)
return payload
def strategy_invalid_opinion_count(strategy_synthesis: Any) -> int:
"""Safely extract invalid_opinion_count from a possibly-malformed synthesis payload.
Guards against `summary_params` being absent OR present-but-not-a-dict
(e.g. a legacy string value). `d.get(k, {})` only uses the default when
the key is missing; if the key exists with a bad value it returns that value
and the subsequent `.get()` crashes. This helper eliminates that footgun
for all renderers.
"""
strategy_synthesis = normalize_strategy_synthesis_payload(strategy_synthesis)
if not strategy_synthesis:
return 0
summary_params = strategy_synthesis.get("summary_params")
if not isinstance(summary_params, dict):
return 0
count = summary_params.get("invalid_opinion_count")
if isinstance(count, bool):
return 0
if isinstance(count, int):
return count if count > 0 else 0
if isinstance(count, str):
normalized = count.strip()
if normalized.isascii() and normalized.isdecimal():
parsed = int(normalized)
return parsed if parsed > 0 else 0
return 0
def localize_strategy_synthesis_summary(strategy_synthesis: Any, language: Optional[str]) -> str:
"""Render a language-specific summary from the structured synthesis payload."""
strategy_synthesis = normalize_strategy_synthesis_payload(strategy_synthesis)
if not strategy_synthesis:
return ""
lang = normalize_report_language(language)
summary_params = strategy_synthesis.get("summary_params")
if not isinstance(summary_params, dict):
summary_params = {}
opinion_count = summary_params.get("opinion_count")
if not isinstance(opinion_count, int):
opinion_count = len(strategy_synthesis.get("supporting_skills") or []) + len(strategy_synthesis.get("opposing_skills") or [])
final_signal = localize_strategy_signal(strategy_synthesis.get("final_signal"), lang)
consensus_level = localize_consensus_level(strategy_synthesis.get("consensus_level"), lang)
conflict_severity = localize_conflict_severity(strategy_synthesis.get("conflict_severity"), lang)
conflict_count = strategy_synthesis.get("conflict_count", 0)
if lang == "en":
if conflict_count:
base = f"Strategy synthesis from {opinion_count} strategies: final signal is {final_signal}, consensus level is {consensus_level}, conflict severity is {conflict_severity}."
else:
base = f"Strategy synthesis from {opinion_count} strategies: final signal is {final_signal}, consensus level is {consensus_level}, with no detected conflicts."
return base
if lang == "ko":
if conflict_count:
base = f"{opinion_count}개 전략의 종합 판단: 종합 신호는 {final_signal}, 공감도는 {consensus_level}, 충돌 강도는 {conflict_severity}입니다."
else:
base = f"{opinion_count}개 전략의 종합 판단: 종합 신호는 {final_signal}, 공감도는 {consensus_level}, 감지된 전략 충돌은 없습니다."
return base
if conflict_count:
base = f"来自 {opinion_count} 个策略的综合判断:综合信号为{final_signal},共识度为{consensus_level},冲突强度为{conflict_severity}"
else:
base = f"来自 {opinion_count} 个策略的综合判断:综合信号为{final_signal},共识度为{consensus_level},未检测到策略冲突。"
return base
def get_chip_unavailable_reason(value: Any, language: Optional[str]) -> str: def get_chip_unavailable_reason(value: Any, language: Optional[str]) -> str:
"""Return the explicit or default chip unavailable reason for rendering.""" """Return the explicit or default chip unavailable reason for rendering."""
if not isinstance(value, dict) or not value: if not isinstance(value, dict) or not value:

View File

@@ -26,8 +26,15 @@ from src.report_language import (
is_chip_structure_unavailable, is_chip_structure_unavailable,
localize_bias_status, localize_bias_status,
localize_chip_health, localize_chip_health,
localize_conflict_severity,
localize_consensus_level,
localize_strategy_signal,
localize_strategy_skill,
localize_strategy_synthesis_summary,
localize_trend_prediction, localize_trend_prediction,
normalize_report_language, normalize_report_language,
normalize_strategy_synthesis_payload,
strategy_invalid_opinion_count,
) )
from src.storage import DatabaseManager from src.storage import DatabaseManager
from src.services.run_diagnostics import build_run_diagnostic_summary from src.services.run_diagnostics import build_run_diagnostic_summary
@@ -1160,6 +1167,51 @@ class HistoryService:
report_lines.append(f"**🐻 {labels.get('strongest_bearish_signal_label', '最强看空信号')}**: {bearish}") report_lines.append(f"**🐻 {labels.get('strongest_bearish_signal_label', '最强看空信号')}**: {bearish}")
report_lines.append("") report_lines.append("")
# ========== 多策略综合 ==========
strategy_synthesis = normalize_strategy_synthesis_payload(
dashboard.get('strategy_synthesis') if dashboard else None
)
if strategy_synthesis:
confidence = strategy_synthesis.get('confidence')
confidence_text = f"{confidence:.0%}" if isinstance(confidence, (int, float)) else "N/A"
report_lines.extend([
f"### 🧩 {labels.get('strategy_synthesis_heading', '多策略综合')}",
"",
(
f"- {labels.get('strategy_final_signal_label', '综合信号')}: "
f"{localize_strategy_signal(strategy_synthesis.get('final_signal', 'N/A'), report_language)} | "
f"{labels.get('strategy_consensus_level_label', '共识度')}: "
f"{localize_consensus_level(strategy_synthesis.get('consensus_level', 'N/A'), report_language)} | "
f"{labels.get('strategy_conflict_label', '冲突')}: "
f"{localize_conflict_severity(strategy_synthesis.get('conflict_severity', 'none'), report_language)} "
f"({strategy_synthesis.get('conflict_count', 0)}) | "
f"{labels.get('strategy_confidence_label', '置信度')}: {confidence_text}"
),
])
summary = localize_strategy_synthesis_summary(strategy_synthesis, report_language)
if summary:
report_lines.append(f"- {labels.get('strategy_summary_label', '综合说明')}: {summary}")
report_lines.append(
f"- {labels.get('strategy_supporting_skills_label', '支持策略')}: "
f"{self._format_strategy_skill_items(strategy_synthesis.get('supporting_skills'), report_language)}"
)
report_lines.append(
f"- {labels.get('strategy_opposing_skills_label', '反方策略')}: "
f"{self._format_strategy_skill_items(strategy_synthesis.get('opposing_skills'), report_language)}"
)
invalid_count = strategy_invalid_opinion_count(strategy_synthesis)
if invalid_count:
invalid_label_template = labels.get(
"strategy_invalid_opinions_label",
"另有 {count} 个策略解析失败",
)
try:
invalid_text = invalid_label_template.format(count=invalid_count)
except (KeyError, IndexError):
invalid_text = f"{invalid_label_template}: {invalid_count}"
report_lines.append(f"- {invalid_text}")
report_lines.append("")
# ========== 如果没有 dashboard显示传统格式 ========== # ========== 如果没有 dashboard显示传统格式 ==========
if not dashboard: if not dashboard:
# 操作理由 # 操作理由
@@ -1202,6 +1254,26 @@ class HistoryService:
return "\n".join(report_lines) return "\n".join(report_lines)
@staticmethod
def _format_strategy_skill_items(items: Any, report_language: str = "zh") -> str:
none_text = get_report_labels(report_language).get("none_label", "None")
if not isinstance(items, list):
return none_text
formatted: List[str] = []
for item in items:
if not isinstance(item, dict):
continue
skill_id = str(item.get("skill_id") or "").strip()
signal = str(item.get("signal") or "").strip()
confidence = item.get("confidence")
if not skill_id:
continue
suffix = f"/{localize_strategy_signal(signal, report_language)}" if signal else ""
if isinstance(confidence, (int, float)):
suffix += f"/{confidence:.0%}"
formatted.append(f"{localize_strategy_skill(skill_id, report_language)}{suffix}")
return "".join(formatted) if formatted else none_text
@staticmethod @staticmethod
def _escape_md(text: Optional[str]) -> str: def _escape_md(text: Optional[str]) -> str:
"""Escape markdown special characters.""" """Escape markdown special characters."""

View File

@@ -23,9 +23,17 @@ from src.report_language import (
get_chip_unavailable_reason, get_chip_unavailable_reason,
is_chip_structure_unavailable, is_chip_structure_unavailable,
localize_chip_health, localize_chip_health,
localize_conflict_severity,
localize_consensus_level,
localize_operation_advice, localize_operation_advice,
localize_strategy_signal,
localize_strategy_skill,
localize_strategy_conflict_description,
localize_strategy_synthesis_summary,
localize_trend_prediction, localize_trend_prediction,
normalize_report_language, normalize_report_language,
normalize_strategy_synthesis_payload,
strategy_invalid_opinion_count,
) )
from src.schemas.decision_action import ( from src.schemas.decision_action import (
display_action_fields_for_result, display_action_fields_for_result,
@@ -225,6 +233,14 @@ def render(
"localize_action_label": localize_action_label, "localize_action_label": localize_action_label,
"localize_trend_prediction": localize_trend_prediction, "localize_trend_prediction": localize_trend_prediction,
"localize_chip_health": localize_chip_health, "localize_chip_health": localize_chip_health,
"localize_strategy_signal": localize_strategy_signal,
"localize_strategy_skill": localize_strategy_skill,
"localize_strategy_conflict_description": localize_strategy_conflict_description,
"localize_strategy_synthesis_summary": localize_strategy_synthesis_summary,
"localize_consensus_level": localize_consensus_level,
"localize_conflict_severity": localize_conflict_severity,
"normalize_strategy_synthesis_payload": normalize_strategy_synthesis_payload,
"strategy_invalid_opinion_count": strategy_invalid_opinion_count,
"signal_attribution_has_content": signal_attribution_has_content, "signal_attribution_has_content": signal_attribution_has_content,
"signal_attribution_weight_items": signal_attribution_weight_items, "signal_attribution_weight_items": signal_attribution_weight_items,
} }

View File

@@ -164,6 +164,26 @@
{% endif %} {% endif %}
{% set strategy_synthesis = normalize_strategy_synthesis_payload(dashboard.get('strategy_synthesis')) %}
{% if strategy_synthesis %}
### 🧩 {{ labels.strategy_synthesis_heading }}
{% set strategy_confidence = strategy_synthesis.get('confidence') %}
- {{ labels.strategy_final_signal_label }}: {{ localize_strategy_signal(strategy_synthesis.get('final_signal', 'N/A'), report_language) }} | {{ labels.strategy_consensus_level_label }}: {{ localize_consensus_level(strategy_synthesis.get('consensus_level', 'N/A'), report_language) }} | {{ labels.strategy_conflict_label }}: {{ localize_conflict_severity(strategy_synthesis.get('conflict_severity', 'none'), report_language) }} ({{ strategy_synthesis.get('conflict_count', 0) }}) | {{ labels.strategy_confidence_label }}: {{ '%.0f%%' | format(strategy_confidence * 100) if strategy_confidence is number else 'N/A' }}
{% if localize_strategy_synthesis_summary(strategy_synthesis, report_language) %}- {{ labels.strategy_summary_label }}: {{ localize_strategy_synthesis_summary(strategy_synthesis, report_language) }}{% endif %}
{% set supporting_skills = strategy_synthesis.get('supporting_skills') or [] %}
{% set opposing_skills = strategy_synthesis.get('opposing_skills') or [] %}
- {{ labels.strategy_supporting_skills_label }}: {% if supporting_skills %}{% for item in supporting_skills %}{{ localize_strategy_skill(item.get('skill_id'), report_language) }}{% if item.get('signal') %}/{{ localize_strategy_signal(item.get('signal'), report_language) }}{% endif %}{% if item.get('confidence') is number %}/{{ '%.0f%%' | format(item.get('confidence') * 100) }}{% endif %}{{ '、' if not loop.last else '' }}{% endfor %}{% else %}{{ labels.none_label }}{% endif %}
- {{ labels.strategy_opposing_skills_label }}: {% if opposing_skills %}{% for item in opposing_skills %}{{ localize_strategy_skill(item.get('skill_id'), report_language) }}{% if item.get('signal') %}/{{ localize_strategy_signal(item.get('signal'), report_language) }}{% endif %}{% if item.get('confidence') is number %}/{{ '%.0f%%' | format(item.get('confidence') * 100) }}{% endif %}{{ '、' if not loop.last else '' }}{% endfor %}{% else %}{{ labels.none_label }}{% endif %}
{% set invalid_opinion_count = strategy_invalid_opinion_count(strategy_synthesis) %}
{% if invalid_opinion_count and invalid_opinion_count > 0 %}- {{ labels.strategy_invalid_opinions_label.format(count=invalid_opinion_count) }}
{% endif %}
{% for conflict in (strategy_synthesis.get('conflicts') or [])[:3] %}
{% if conflict.get('conflict_type') %}- {{ localize_conflict_severity(conflict.get('severity', 'medium'), report_language) }}: {{ localize_strategy_conflict_description(conflict.get('conflict_type'), report_language) }}{% if conflict.get('participants') %}{% for participant in conflict.get('participants') %}{{ localize_strategy_skill(participant, report_language) }}{{ '、' if not loop.last else '' }}{% endfor %}{% endif %}{% endif %}
{% endfor %}
{% endif %}
{% if battle %} {% if battle %}
### 🎯 {{ labels.battle_plan_heading }} ### 🎯 {{ labels.battle_plan_heading }}

View File

@@ -64,6 +64,21 @@
{% endif %} {% endif %}
{% endif %} {% endif %}
{% set strategy_synthesis = normalize_strategy_synthesis_payload(dashboard.get('strategy_synthesis')) %}
{% if strategy_synthesis %}
🧩 **{{ labels.strategy_synthesis_heading }}**: {{ localize_strategy_signal(strategy_synthesis.get('final_signal', 'N/A'), report_language) }} | {{ labels.strategy_consensus_level_label }} {{ localize_consensus_level(strategy_synthesis.get('consensus_level', 'N/A'), report_language) }} | {{ labels.strategy_conflict_label }} {{ localize_conflict_severity(strategy_synthesis.get('conflict_severity', 'none'), report_language) }}({{ strategy_synthesis.get('conflict_count', 0) }})
{% set supporting_skills = strategy_synthesis.get('supporting_skills') or [] %}
{% set opposing_skills = strategy_synthesis.get('opposing_skills') or [] %}
- {{ labels.strategy_supporting_skills_label }}: {% if supporting_skills %}{% for item in supporting_skills %}{{ localize_strategy_skill(item.get('skill_id'), report_language) }}{{ '、' if not loop.last else '' }}{% endfor %}{% else %}{{ labels.none_label }}{% endif %}
- {{ labels.strategy_opposing_skills_label }}: {% if opposing_skills %}{% for item in opposing_skills %}{{ localize_strategy_skill(item.get('skill_id'), report_language) }}{{ '、' if not loop.last else '' }}{% endfor %}{% else %}{{ labels.none_label }}{% endif %}
{% set invalid_opinion_count = strategy_invalid_opinion_count(strategy_synthesis) %}
{% if invalid_opinion_count and invalid_opinion_count > 0 %}- {{ labels.strategy_invalid_opinions_label.format(count=invalid_opinion_count) }}
{% endif %}
{% if localize_strategy_synthesis_summary(strategy_synthesis, report_language) %}
{{ localize_strategy_synthesis_summary(strategy_synthesis, report_language)[:80] }}
{% endif %}
{% endif %}
{% set checklist = battle.get('action_checklist', []) if battle else [] %} {% set checklist = battle.get('action_checklist', []) if battle else [] %}
{% set fc = failed_checks(checklist) %} {% set fc = failed_checks(checklist) %}
{% if fc %} {% if fc %}

View File

@@ -1761,6 +1761,59 @@ class AnalysisHistoryTestCase(unittest.TestCase):
self.assertIn("**🟡 Avoid** | Bullish", markdown) self.assertIn("**🟡 Avoid** | Bullish", markdown)
self.assertNotIn("Strong Buy", markdown) self.assertNotIn("Strong Buy", markdown)
def test_history_markdown_handles_legacy_strategy_synthesis_shapes(self) -> None:
service = HistoryService(self.db)
record = MagicMock(created_at=None)
for malformed in ("bad-shape", ["bad-shape"], 42, True):
result = AnalysisResult(
code="600519",
name="贵州茅台",
sentiment_score=50,
trend_prediction="震荡",
operation_advice="观望",
report_language="zh",
dashboard={
"core_conclusion": {"one_sentence": "测试"},
"intelligence": {},
"battle_plan": {},
"strategy_synthesis": malformed,
},
)
markdown = service._generate_single_stock_markdown(result, record)
self.assertNotIn("多策略综合", markdown)
result = AnalysisResult(
code="600519",
name="贵州茅台",
sentiment_score=50,
trend_prediction="震荡",
operation_advice="观望",
report_language="zh",
dashboard={
"core_conclusion": {"one_sentence": "测试"},
"intelligence": {},
"battle_plan": {},
"strategy_synthesis": {
"final_signal": "hold",
"consensus_level": "insufficient",
"conflict_severity": "none",
"conflict_count": 0,
"supporting_skills": "bad-shape",
"opposing_skills": ["bad-shape"],
"conflicts": "bad-shape",
"summary_params": {"invalid_opinion_count": "3"},
},
},
)
markdown = service._generate_single_stock_markdown(result, record)
self.assertIn("多策略综合", markdown)
self.assertIn("另有 3 个策略解析失败", markdown)
def test_history_markdown_returns_persisted_market_review_report(self) -> None: def test_history_markdown_returns_persisted_market_review_report(self) -> None:
"""Market review history should return the saved Markdown without rebuilding a stock report.""" """Market review history should return the saved Markdown without rebuilding a stock report."""
result = AnalysisResult( result = AnalysisResult(

File diff suppressed because it is too large Load Diff

View File

@@ -952,6 +952,65 @@ class TestNotificationServiceReportGeneration(unittest.TestCase):
self.assertNotIn("AI 决策信号", out) self.assertNotIn("AI 决策信号", out)
self.assertNotIn("理由: 技术面走弱", out) self.assertNotIn("理由: 技术面走弱", out)
@mock.patch("src.notification.get_config")
def test_strategy_synthesis_legacy_shapes_are_safe_in_fallback_reports(
self, mock_get_config: mock.MagicMock
):
mock_get_config.return_value = _make_config(report_renderer_enabled=False)
service = NotificationService()
for malformed in ("bad-shape", ["bad-shape"], 42, True):
result = AnalysisResult(
code="600519",
name="贵州茅台",
sentiment_score=50,
trend_prediction="震荡",
operation_advice="观望",
report_language="zh",
dashboard={
"core_conclusion": {"one_sentence": "测试"},
"intelligence": {},
"battle_plan": {},
"strategy_synthesis": malformed,
},
)
markdown = service.generate_dashboard_report([result], report_date="2026-07-19")
wechat = service.generate_wechat_dashboard([result])
self.assertNotIn("多策略综合", markdown)
self.assertNotIn("多策略综合", wechat)
result = AnalysisResult(
code="600519",
name="贵州茅台",
sentiment_score=50,
trend_prediction="震荡",
operation_advice="观望",
report_language="zh",
dashboard={
"core_conclusion": {"one_sentence": "测试"},
"intelligence": {},
"battle_plan": {},
"strategy_synthesis": {
"final_signal": "hold",
"consensus_level": "insufficient",
"conflict_severity": "none",
"conflict_count": 0,
"supporting_skills": "bad-shape",
"opposing_skills": ["bad-shape"],
"conflicts": "bad-shape",
"summary_params": {"invalid_opinion_count": "3"},
},
},
)
markdown = service.generate_dashboard_report([result], report_date="2026-07-19")
wechat = service.generate_wechat_dashboard([result])
self.assertIn("另有 3 个策略解析失败", markdown)
self.assertIn("另有 3 个策略解析失败", wechat)
@mock.patch("src.notification.get_config") @mock.patch("src.notification.get_config")
def test_generate_wechat_summary_omits_decision_signal_excerpt( def test_generate_wechat_summary_omits_decision_signal_excerpt(
self, mock_get_config: mock.MagicMock self, mock_get_config: mock.MagicMock

View File

@@ -382,6 +382,88 @@ class TestReportRenderer(unittest.TestCase):
self.assertIn("**筹码**: 筹码分布未启用或数据源暂不可用,未纳入筹码判断。", out) self.assertIn("**筹码**: 筹码分布未启用或数据源暂不可用,未纳入筹码判断。", out)
self.assertEqual(out.count("数据缺失,无法判断"), 0) self.assertEqual(out.count("数据缺失,无法判断"), 0)
def test_render_markdown_renders_strategy_synthesis_with_localized_labels(self) -> None:
r = _make_result(
dashboard={
"core_conclusion": {"one_sentence": "持有观望"},
"strategy_synthesis": {
"final_signal": "buy",
"confidence": 0.8,
"conflict_count": 1,
"conflict_severity": "medium",
"consensus_level": "medium",
"summary_key": "strategy_synthesis.with_conflicts",
"summary_params": {
"opinion_count": 2,
"final_signal": "buy",
"consensus_level": "medium",
"conflict_severity": "medium",
"conflict_count": 1,
},
"supporting_skills": [{"skill_id": "bull_trend", "signal": "buy", "confidence": 0.8}],
"opposing_skills": [{"skill_id": "hot_theme", "signal": "sell", "confidence": 0.75}],
"conflicts": [
{
"conflict_type": "directional_opposition",
"severity": "medium",
"description_key": "strategy_conflict.directional_opposition",
"participants": ["bull_trend", "hot_theme"],
}
],
},
}
)
out = render("markdown", [r], summary_only=False)
self.assertIsNotNone(out)
self.assertIn("多策略综合", out)
self.assertIn("综合信号: 买入", out)
self.assertIn("默认多头趋势/买入/80%", out)
self.assertIn("热点题材/卖出/75%", out)
self.assertNotIn("bull_trend/买入", out)
def test_render_templates_handle_legacy_strategy_synthesis_shapes(self) -> None:
for platform in ("markdown", "wechat"):
for malformed in ("bad-shape", ["bad-shape"], 42, True):
result = _make_result(
dashboard={
"core_conclusion": {"one_sentence": "持有观望"},
"intelligence": {},
"battle_plan": {},
"strategy_synthesis": malformed,
}
)
out = render(platform, [result], summary_only=False)
self.assertIsNotNone(out)
self.assertNotIn("多策略综合", out)
result = _make_result(
dashboard={
"core_conclusion": {"one_sentence": "持有观望"},
"intelligence": {},
"battle_plan": {},
"strategy_synthesis": {
"final_signal": "hold",
"consensus_level": "insufficient",
"conflict_severity": "none",
"conflict_count": 0,
"supporting_skills": "bad-shape",
"opposing_skills": ["bad-shape"],
"conflicts": "bad-shape",
"summary_params": {"invalid_opinion_count": "3"},
},
}
)
out = render(platform, [result], summary_only=False)
self.assertIsNotNone(out)
self.assertIn("多策略综合", out)
self.assertIn("另有 3 个策略解析失败", out)
def test_render_unknown_platform_returns_none(self) -> None: def test_render_unknown_platform_returns_none(self) -> None:
"""Unknown platform returns None (caller fallback).""" """Unknown platform returns None (caller fallback)."""
r = _make_result() r = _make_result()