Files
daily_stock_analysis/tests/test_report_integrity.py
743ca5b1b4 feat: 添加信号归因分析功能 (Issue #1742) (#1796)
* feat: 修复 reviewer blocker 问题

- 同步 agent 路径(executor.py, decision_agent.py)
- 添加 SignalAttribution 字段验证器(自动转换、归零、归一化)
- 更新 docs/CHANGELOG.md
- 添加回归测试(tests/test_signal_attribution.py)
- 修复 notification.py 和模板的 None 值显示问题

Closes #1742

* fix: 修复 signal_attribution 完整契约

Reviewer feedback 指出的完整契约收敛:

## 1. [Correctness] 归一化接入真实 parse 路径
- 问题:Pydantic validator 没有进入主分析路径(dashboard 是 raw dict)
- 修复:将归一化函数移到 src/utils/data_processing.py,
  在 _parse_response() 和 agent runner.py 的 parse_dashboard_json() 中调用
- 确保 LLM 返回的字符串/负数/总和≠100 被正确处理

## 2. [Correctness] 同步 HistoryService 路径
- 问题:_generate_single_stock_markdown() 不读取 signal_attribution
- 修复:在 history_service.py 中添加信号归因展示代码

## 3. [Process] 修复 CHANGELOG.md 格式
- 问题:两行 [Unreleased] 条目拼在同一行
- 修复:分割成独立行

## 4. [验证] 添加真实路径回归测试
- tests/test_signal_attribution_real_paths.py:
  - 归一化函数测试(9个)
  - _parse_response 集成测试(1个)
  - HistoryService 展示测试(2个)

## 5. [Process] 修复 executor.py prompt 模板格式
- 问题:signal_attribution JSON 例子没转义花括号,导致 .format() 报错
- 修复:将 { 转成 {{,} 转成 }}

Co-authored-by: qyj <jiangqiyuan@tencent.com>

* fix: remove trailing whitespace in notification.py and report_schema.py

* fix: converge signal_attribution contract across all paths

- Add signal_attribution to check_content_integrity() as recommended field
- Normalize signal_attribution in _parse_response() and parse_dashboard_json()
- Sync HistoryService._generate_single_stock_markdown() to render signal_attribution
- Update CHANGELOG.md to reflect actual implementation (explicit normalization, not schema-level)
- Add end-to-end tests covering all paths: _parse_response, notification, Jinja2, HistoryService
- Fix tests to accept signal_attribution as recommended field (missing does not fail integrity check)

* fix: address all reviewer blockers

- Fix generate_single_stock_report() to render signal_attribution
- Fix normalization: clamp values to [0, 100], keep all-zero as 0 (not 25)
- Update docs/full-guide.md and docs/full-guide_EN.md with signal_attribution description
- Add supplement tests covering generate_single_stock_report, normalization edge cases, and _parse_response integration

* fix: 修复 CI 静态检查失败和文档表述不一致

- 修复 tests/test_signal_attribution_supplement.py 的 flake8 错误(F821 undefined name 'AnalysisResult')
- 将 AnalysisResult import 移到文件顶部
- 更新 docs/CHANGELOG.md 表述,反映实际行为(all-zero 保留为 0,有效贡献度归一化到 100)
- 所有 40 个 signal_attribution 测试通过

* fix: converge signal attribution runtime contract

* fix: hide empty signal attribution blocks

* fix: reject non-finite signal attribution weights

---------

Co-authored-by: qiyuanjiang <qiyuanjiang@tencent.com>
Co-authored-by: qyj <jiangqiyuan@tencent.com>
Co-authored-by: hsms4710-pixel <228664208+hsms4710-pixel@users.noreply.github.com>
Co-authored-by: zhulinsen <zhuls97@163.com>
2026-06-27 22:41:44 +08:00

529 lines
21 KiB
Python

# -*- coding: utf-8 -*-
"""
===================================
Report Engine - Content integrity tests
===================================
Tests for check_content_integrity, apply_placeholder_fill, and retry/placeholder behavior.
"""
import json
import sys
import unittest
from unittest.mock import MagicMock, patch
try:
import litellm # noqa: F401
except ModuleNotFoundError:
sys.modules["litellm"] = MagicMock()
from src.analyzer import AnalysisResult, GeminiAnalyzer, check_content_integrity, apply_placeholder_fill
class TestCheckContentIntegrity(unittest.TestCase):
"""Content integrity check tests."""
def test_pass_when_all_required_present(self) -> None:
"""Integrity passes when all mandatory fields are present."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有观望"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": "110元"}},
},
)
ok, missing = check_content_integrity(result)
self.assertTrue(ok)
self.assertEqual(missing, [])
def test_pass_when_signal_attribution_missing(self) -> None:
"""Signal attribution is optional and does not enter missing_fields."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有观望"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": "110元"}},
},
)
ok, missing = check_content_integrity(result)
self.assertTrue(ok)
self.assertEqual(missing, [])
def test_fail_when_analysis_summary_empty(self) -> None:
"""Integrity fails when analysis_summary is empty."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": "110"}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("analysis_summary", missing)
def test_fail_when_one_sentence_missing(self) -> None:
"""Integrity fails when core_conclusion.one_sentence is missing."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": "110"}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("dashboard.core_conclusion.one_sentence", missing)
def test_fail_when_one_sentence_blank(self) -> None:
"""Integrity fails when one_sentence is blank whitespace."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": " "},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": "110"}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("dashboard.core_conclusion.one_sentence", missing)
def test_fail_when_stop_loss_missing_for_buy(self) -> None:
"""Integrity fails when stop_loss missing and decision_type is buy."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="买入",
analysis_summary="稳健",
decision_type="buy",
dashboard={
"core_conclusion": {"one_sentence": "可买入"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("dashboard.battle_plan.sniper_points.stop_loss", missing)
def test_pass_when_stop_loss_missing_for_sell(self) -> None:
"""Integrity passes when stop_loss missing and decision_type is sell."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看空",
sentiment_score=35,
operation_advice="卖出",
analysis_summary="弱势",
decision_type="sell",
dashboard={
"core_conclusion": {"one_sentence": "建议卖出"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {}},
},
)
ok, missing = check_content_integrity(result)
self.assertTrue(ok)
self.assertEqual(missing, [])
def test_fail_when_risk_alerts_missing(self) -> None:
"""Integrity fails when intelligence.risk_alerts field is missing."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有"},
"intelligence": {},
"battle_plan": {"sniper_points": {"stop_loss": "110"}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("dashboard.intelligence.risk_alerts", missing)
def test_phase_decision_missing_only_when_required(self) -> None:
"""Phase decision fields are required only for phase-aware analysis."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": "110"}},
},
)
ok, missing = check_content_integrity(result)
self.assertTrue(ok)
self.assertEqual(missing, [])
ok, missing = check_content_integrity(result, require_phase_decision=True)
self.assertFalse(ok)
self.assertIn("dashboard.phase_decision.phase_context", missing)
self.assertIn("dashboard.phase_decision.watch_conditions", missing)
self.assertIn("dashboard.phase_decision.data_limitations", missing)
def test_fail_when_risk_alerts_is_none(self) -> None:
"""Integrity fails when risk_alerts is None."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有"},
"intelligence": {"risk_alerts": None},
"battle_plan": {"sniper_points": {"stop_loss": "110"}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("dashboard.intelligence.risk_alerts", missing)
def test_fail_when_risk_alerts_is_invalid_type(self) -> None:
"""Integrity fails when risk_alerts is not list."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有"},
"intelligence": {"risk_alerts": "需留意"},
"battle_plan": {"sniper_points": {"stop_loss": "110"}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("dashboard.intelligence.risk_alerts", missing)
def test_fail_when_stop_loss_is_blank(self) -> None:
"""Integrity fails when stop_loss is blank whitespace."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="买入",
analysis_summary="稳健",
decision_type="buy",
dashboard={
"core_conclusion": {"one_sentence": "可买入"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": " "}},
},
)
ok, missing = check_content_integrity(result)
self.assertFalse(ok)
self.assertIn("dashboard.battle_plan.sniper_points.stop_loss", missing)
class TestApplyPlaceholderFill(unittest.TestCase):
"""Placeholder fill tests."""
def test_fills_missing_analysis_summary(self) -> None:
"""Placeholder fills analysis_summary when missing."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="",
decision_type="hold",
dashboard={},
)
apply_placeholder_fill(result, ["analysis_summary"])
self.assertEqual(result.analysis_summary, "待补充")
def test_fills_missing_analysis_summary_in_english(self) -> None:
"""English report should use English placeholder text for missing analysis_summary."""
result = AnalysisResult(
code="600519",
name="MacaoTech",
report_language="en",
trend_prediction="Bullish",
sentiment_score=70,
operation_advice="Buy",
analysis_summary="",
decision_type="buy",
dashboard={},
)
apply_placeholder_fill(result, ["analysis_summary"])
self.assertEqual(result.analysis_summary, "TBD")
def test_fills_missing_stop_loss(self) -> None:
"""Placeholder fills stop_loss when missing."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="买入",
analysis_summary="稳健",
decision_type="buy",
dashboard={"battle_plan": {"sniper_points": {}}},
)
apply_placeholder_fill(result, ["dashboard.battle_plan.sniper_points.stop_loss"])
self.assertEqual(
result.dashboard["battle_plan"]["sniper_points"]["stop_loss"],
"待补充",
)
def test_fills_risk_alerts_empty_list(self) -> None:
"""Placeholder fills risk_alerts with empty list when missing."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={"intelligence": {}},
)
apply_placeholder_fill(result, ["dashboard.intelligence.risk_alerts"])
self.assertEqual(result.dashboard["intelligence"]["risk_alerts"], [])
def test_fills_risk_alerts_when_none(self) -> None:
"""Placeholder fills risk_alerts when value is None."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
risk_warning="注意融资",
dashboard={"intelligence": {"risk_alerts": None}},
)
apply_placeholder_fill(result, ["dashboard.intelligence.risk_alerts"])
self.assertEqual(result.dashboard["intelligence"]["risk_alerts"], ["注意融资"])
def test_fills_risk_alerts_when_invalid_type(self) -> None:
"""Placeholder fills risk_alerts when value is non-list."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
dashboard={"intelligence": {"risk_alerts": "注意回撤"}},
)
apply_placeholder_fill(result, ["dashboard.intelligence.risk_alerts"])
self.assertEqual(result.dashboard["intelligence"]["risk_alerts"], [])
def test_fills_risk_alerts_when_risk_warning_is_list(self) -> None:
"""Placeholder handles list risk_warning and flattens valid text values."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
risk_warning=["回撤风险", "波动加大"],
dashboard={"intelligence": {"risk_alerts": ""}},
)
apply_placeholder_fill(result, ["dashboard.intelligence.risk_alerts"])
self.assertEqual(result.dashboard["intelligence"]["risk_alerts"], ["回撤风险", "波动加大"])
def test_fills_risk_alerts_when_risk_warning_is_dict(self) -> None:
"""Placeholder serializes dict risk_warning into a string risk alert."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="持有",
analysis_summary="稳健",
decision_type="hold",
risk_warning={"note": "技术面偏弱"},
dashboard={"intelligence": {"risk_alerts": ""}},
)
apply_placeholder_fill(result, ["dashboard.intelligence.risk_alerts"])
self.assertEqual(
json.loads(result.dashboard["intelligence"]["risk_alerts"][0]),
{"note": "技术面偏弱"},
)
def test_fills_stop_loss_when_blank(self) -> None:
"""Placeholder fills stop_loss when blank whitespace."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="买入",
analysis_summary="稳健",
decision_type="buy",
dashboard={"battle_plan": {"sniper_points": {"stop_loss": " "}}},
)
apply_placeholder_fill(result, ["dashboard.battle_plan.sniper_points.stop_loss"])
self.assertEqual(
result.dashboard["battle_plan"]["sniper_points"]["stop_loss"],
"待补充",
)
def test_fills_stop_loss_when_invalid_type(self) -> None:
"""Placeholder fills stop_loss when value is invalid type."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="买入",
analysis_summary="稳健",
decision_type="buy",
dashboard={"battle_plan": {"sniper_points": {"stop_loss": {}}}},
)
apply_placeholder_fill(result, ["dashboard.battle_plan.sniper_points.stop_loss"])
self.assertEqual(
result.dashboard["battle_plan"]["sniper_points"]["stop_loss"],
"待补充",
)
def test_fills_none_dashboard_blocks_from_existing_context(self) -> None:
"""Placeholder fill handles null dashboard blocks and reuses existing result text."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="看多",
sentiment_score=70,
operation_advice="买入",
analysis_summary="已有趋势摘要",
risk_warning="跌破支撑需减仓",
decision_type="buy",
dashboard={
"core_conclusion": None,
"intelligence": None,
"battle_plan": None,
},
)
apply_placeholder_fill(
result,
[
"dashboard.core_conclusion.one_sentence",
"dashboard.intelligence.risk_alerts",
"dashboard.battle_plan.sniper_points.stop_loss",
],
)
self.assertEqual(result.dashboard["core_conclusion"]["one_sentence"], "已有趋势摘要")
self.assertEqual(result.dashboard["intelligence"]["risk_alerts"], ["跌破支撑需减仓"])
self.assertEqual(result.dashboard["battle_plan"]["sniper_points"]["stop_loss"], "待补充")
def test_phase_decision_placeholder_fill_satisfies_integrity_contract(self) -> None:
"""Phase placeholders close the retry-exhausted integrity contract without fake conditions."""
result = AnalysisResult(
code="600519",
name="贵州茅台",
trend_prediction="震荡",
sentiment_score=50,
operation_advice="持有",
analysis_summary="已有摘要",
decision_type="hold",
dashboard={
"core_conclusion": {"one_sentence": "持有观望"},
"intelligence": {"risk_alerts": []},
"battle_plan": {"sniper_points": {"stop_loss": "100"}},
"phase_decision": {
"phase_context": "invalid",
"watch_conditions": "invalid",
"data_limitations": None,
},
},
)
ok, missing = check_content_integrity(result, require_phase_decision=True)
self.assertFalse(ok)
apply_placeholder_fill(result, missing)
ok, missing = check_content_integrity(result, require_phase_decision=True)
self.assertTrue(ok)
self.assertEqual(missing, [])
phase_decision = result.dashboard["phase_decision"]
self.assertEqual(phase_decision["phase_context"], {})
self.assertEqual(phase_decision["watch_conditions"], [])
self.assertEqual(phase_decision["data_limitations"], [])
self.assertEqual(phase_decision["action_window"], "模型未提供阶段化行动窗口")
self.assertEqual(phase_decision["immediate_action"], "模型未提供阶段化即时动作")
self.assertEqual(phase_decision["next_check_time"], "模型未提供下一次检查点")
self.assertEqual(phase_decision["confidence_reason"], "模型未提供阶段化置信度理由")
class TestIntegrityRetryPrompt(unittest.TestCase):
"""Retry prompt construction tests."""
def test_retry_prompt_includes_previous_response(self) -> None:
"""Retry prompt should carry previous response so补全是增量的。"""
with patch.object(GeminiAnalyzer, "_init_litellm", return_value=None):
analyzer = GeminiAnalyzer()
prompt = analyzer._build_integrity_retry_prompt(
"原始提示",
'{"analysis_summary": "已有内容"}',
["dashboard.core_conclusion.one_sentence"],
)
self.assertIn("原始提示", prompt)
self.assertIn('{"analysis_summary": "已有内容"}', prompt)
self.assertIn("dashboard.core_conclusion.one_sentence", prompt)