fix: address P6 signal linkage gaps (#1724)

Author: massif-01

docs/CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 - [新功能] #1390 P6 将 DecisionSignal 复用到告警、通知和组合风险：告警触发关联 latest active 信号或创建最小 alert 信号，通知追加低敏信号摘要，持仓风险聚合 active sell/reduce/alert 信号并保持 fail-open。
+- [修复] #1722 修复 #1390 P6 DecisionSignal 在组合风险快照语义和默认聚合通知展示中的遗漏。
 - [新功能] #1707 资讯源新增 `newsnow` 类型、`NEWSNOW_BASE_URL` 配置和 `/api/v1/intelligence/sources/defaults` 默认源初始化接口，内置财联社热门、雪球热门股票、华尔街见闻快讯、金十数据和格隆汇事件等财经热点源，可直接拉取落库并进入既有分析证据链路；官方 NewsNow 部署指南见 https://github.com/qqhann/newsnow，生产环境建议自建实例而非使用公开示例。
 
 - [修复] AlphaSift 热点题材刷新在 EastMoney 瞬断且无缓存时返回友好空态，并让桌面更新保留 AlphaSift 热点缓存。

src/notification.py

@@ -850,6 +850,9 @@ def generate_daily_report(
                     f"{labels['score_label']} {r.sentiment_score} | "
                     f"{localize_trend_prediction(r.trend_prediction, report_language)}"
                 )
+                signal_excerpt = self._decision_signal_excerpt(r, report_language)
+                if signal_excerpt:
+                    report_lines.append(signal_excerpt)
         else:
             report_lines.extend([f"## 📈 {labels['report_title']}", ""])
             # 逐个股票的详细分析
@@ -866,6 +869,9 @@ def generate_daily_report(
                     f"**Confidence：{confidence_stars}**",
                     "",
                 ])
+                signal_excerpt = self._decision_signal_excerpt(result, report_language)
+                if signal_excerpt:
+                    report_lines.extend([signal_excerpt, ""])
                 self._append_market_snapshot(report_lines, result)
 
                 # 核心看点
@@ -1099,6 +1105,9 @@ def generate_dashboard_report(
                     f"{labels['score_label']} {r.sentiment_score} | "
                     f"{localize_trend_prediction(r.trend_prediction, report_language)}"
                 )
+                signal_excerpt = self._decision_signal_excerpt(r, report_language)
+                if signal_excerpt:
+                    report_lines.append(signal_excerpt)
             report_lines.extend([
                 "",
                 "---",
@@ -1401,6 +1410,9 @@ def generate_wechat_dashboard(self, results: List[AnalysisResult]) -> str:
                     f"{labels['score_label']} {r.sentiment_score} | "
                     f"{localize_trend_prediction(r.trend_prediction, report_language)}"
                 )
+                signal_excerpt = self._decision_signal_excerpt(r, report_language)
+                if signal_excerpt:
+                    lines.append(signal_excerpt)
         else:
             for result in sorted_results:
                 signal_text, signal_emoji, _ = self._get_signal_level(result)
@@ -1421,6 +1433,10 @@ def generate_wechat_dashboard(self, results: List[AnalysisResult]) -> str:
                 if one_sentence:
                     lines.append(f"📌 **{one_sentence[:80]}**")
                     lines.append("")
+                signal_excerpt = self._decision_signal_excerpt(result, report_language)
+                if signal_excerpt:
+                    lines.append(signal_excerpt)
+                    lines.append("")
 
                 # 重要信息区（舆情+基本面）
                 info_lines = []
@@ -1551,6 +1567,9 @@ def generate_wechat_summary(self, results: List[AnalysisResult]) -> str:
                 f"{labels['score_label']}:{result.sentiment_score} | "
                 f"{localize_trend_prediction(result.trend_prediction, report_language)}"
             )
+            signal_excerpt = self._decision_signal_excerpt(result, report_language)
+            if signal_excerpt:
+                lines.append(signal_excerpt)
 
             # 操作理由（截断）
             if hasattr(result, 'buy_reason') and result.buy_reason:

src/services/decision_signal_service.py

@@ -115,6 +115,7 @@ def list_signals(
         expires_to: Optional[Any] = None,
         holding_only: bool = False,
         account_id: Optional[int] = None,
+        stock_identities: Optional[List[Tuple[str, str]]] = None,
         page: int = 1,
         page_size: int = 20,
     ) -> Dict[str, Any]:
@@ -133,9 +134,27 @@ def list_signals(
         expires_from_dt = self._parse_datetime(expires_from)
         expires_to_dt = self._parse_datetime(expires_to)
         stock_codes = self._stock_filter_codes(stock_code, market=market_norm)
-        stock_identities = None
-
-        if holding_only:
+        stock_identity_filters: Optional[List[Tuple[str, str]]] = None
+
+        if stock_identities is not None:
+            # Explicit identities come from a caller-owned snapshot; skip cached holdings entirely.
+            requested_codes = set(stock_codes or [])
+            normalized_identities: set[Tuple[str, str]] = set()
+            for identity_market, identity_code in stock_identities:
+                if not str(identity_code or "").strip():
+                    continue
+                identity_market_norm = self._normalize_market(identity_market)
+                if market_norm and identity_market_norm != market_norm:
+                    continue
+                identity_code_norm = self._normalize_stock_code(identity_code, market=identity_market_norm)
+                if requested_codes and identity_code_norm not in requested_codes:
+                    continue
+                normalized_identities.add((identity_market_norm, identity_code_norm))
+            stock_identity_filters = sorted(normalized_identities)
+            stock_codes = None
+            if not stock_identity_filters:
+                return {"items": [], "total": 0, "page": safe_page, "page_size": safe_page_size}
+        elif holding_only:
             held_identities = self._cached_holding_identities(account_id=account_id)
             if market_norm:
                 held_identities = {
@@ -146,14 +165,14 @@ def list_signals(
                 held_identities = {
                     identity for identity in held_identities if identity[1] in requested_codes
                 }
-            stock_identities = sorted(held_identities)
+            stock_identity_filters = sorted(held_identities)
             stock_codes = None
-            if not stock_identities:
+            if not stock_identity_filters:
                 return {"items": [], "total": 0, "page": safe_page, "page_size": safe_page_size}
 
         rows, total = self.repo.list(
             stock_codes=stock_codes,
-            stock_identities=stock_identities,
+            stock_identities=stock_identity_filters,
             market=market_norm,
             action=action_norm,
             market_phase=market_phase_norm,
@@ -183,12 +202,13 @@ def list_signals(
             created_to=created_to_dt,
             expires_from=expires_from_dt,
             expires_to=expires_to_dt,
+            stock_identities=stock_identity_filters,
             holding_only=holding_only,
         ):
             self._backfill_analysis_signal_from_history(source_report_id_norm)
             rows, total = self.repo.list(
                 stock_codes=stock_codes,
-                stock_identities=stock_identities,
+                stock_identities=stock_identity_filters,
                 market=market_norm,
                 action=action_norm,
                 market_phase=market_phase_norm,
@@ -276,6 +296,7 @@ def _should_backfill_history_bound_analysis_signal(
         created_to: Optional[datetime],
         expires_from: Optional[datetime],
         expires_to: Optional[datetime],
+        stock_identities: Optional[List[Tuple[str, str]]],
         holding_only: bool,
     ) -> bool:
         """Only lazy-backfill for the exact report section query used by Web."""
@@ -296,6 +317,7 @@ def _should_backfill_history_bound_analysis_signal(
                 created_to,
                 expires_from,
                 expires_to,
+                stock_identities,
                 holding_only,
             )
         )

src/services/portfolio_risk_service.py

@@ -82,7 +82,7 @@ def get_risk_report(
             lookback_days=thresholds["lookback_days"],
         )
         stop_loss = self._build_stop_loss(snapshot, thresholds)
-        decision_signal_risk = self._build_decision_signal_risk(snapshot, account_id=account_id)
+        decision_signal_risk = self._build_decision_signal_risk(snapshot)
 
         return {
             "as_of": as_of_date.isoformat(),
@@ -100,21 +100,22 @@ def get_risk_report(
     def _build_decision_signal_risk(
         self,
         snapshot: Dict[str, Any],
-        *,
-        account_id: Optional[int],
     ) -> Dict[str, Any]:
         try:
             held_positions = self._held_position_identities(snapshot)
             if not held_positions:
                 return self._empty_decision_signal_risk(available=True)
+            stock_identities = sorted({
+                (position["market"], position["signal_stock_code"])
+                for position in held_positions
+            })
 
             defensive_actions = set(DEFENSIVE_DECISION_SIGNAL_ACTIONS)
             latest_by_identity: Dict[Tuple[str, str], Dict[str, Any]] = {}
             page = 1
             while True:
                 response = self.decision_signal_service.list_signals(
-                    holding_only=True,
-                    account_id=account_id,
+                    stock_identities=stock_identities,
                     status="active",
                     page=page,
                     page_size=100,

templates/report_markdown.j2

@@ -10,6 +10,10 @@
 
 {% for e in enriched %}
 {{ e.signal_emoji }} **{{ e.stock_name }}({{ e.result.code }})**: {{ e.localized_operation_advice }} | {{ labels.score_label }} {{ e.result.sentiment_score }} | {{ e.localized_trend_prediction }}
+{% set signal_excerpt = decision_signal_excerpt(e.result) %}
+{% if signal_excerpt %}
+{{ signal_excerpt }}
+{% endif %}
 {% endfor %}
 
 ---

templates/report_wechat.j2

@@ -9,6 +9,10 @@
 **📊 {{ labels.summary_heading }}**
 {% for e in enriched %}
 {{ e.signal_emoji }} **{{ e.stock_name }}({{ e.result.code }})**: {{ e.localized_operation_advice }} | {{ labels.score_label }} {{ e.result.sentiment_score }} | {{ e.localized_trend_prediction }}
+{% set signal_excerpt = decision_signal_excerpt(e.result) %}
+{% if signal_excerpt %}
+{{ signal_excerpt }}
+{% endif %}
 {% endfor %}
 {% else %}
 {% for e in enriched %}
@@ -24,6 +28,10 @@
 {% if one_sentence %}
 📌 **{{ one_sentence[:80] }}**
 {% endif %}
+{% set signal_excerpt = decision_signal_excerpt(result) %}
+{% if signal_excerpt %}
+{{ signal_excerpt }}
+{% endif %}
 
 {% if intel.get('earnings_outlook') %}
 📊 {{ labels.earnings_outlook_label }}: {{ intel.earnings_outlook[:60] }}

tests/test_decision_signal_service.py

@@ -494,6 +494,84 @@ def test_list_signals_does_not_backfill_ambiguous_history_default_decision_type_
         assert session.query(DecisionSignalRecord).count() == 0
 
 
+def test_list_signals_explicit_stock_identities_override_holding_only_and_intersect_filters(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    service.create_signal(
+        _payload(
+            source_report_id=171501,
+            trace_id="trace-explicit-identity-000001",
+            stock_code="000001",
+            stock_name="平安银行",
+            action="sell",
+        )
+    )
+    service.create_signal(
+        _payload(
+            source_report_id=171502,
+            trace_id="trace-explicit-identity-600519",
+            stock_code="600519",
+            action="reduce",
+        )
+    )
+
+    listed = service.list_signals(
+        stock_identities=[("cn", "000001")],
+        holding_only=True,
+        status="active",
+    )
+
+    assert listed["total"] == 1
+    assert listed["items"][0]["stock_code"] == "000001"
+    assert listed["items"][0]["action"] == "sell"
+
+    mismatched_stock_filter = service.list_signals(
+        stock_code="600519",
+        market="cn",
+        stock_identities=[("cn", "000001")],
+        status="active",
+    )
+
+    assert mismatched_stock_filter == {"items": [], "total": 0, "page": 1, "page_size": 20}
+
+
+def test_list_signals_explicit_empty_stock_identities_returns_empty_without_widening(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    service.create_signal(
+        _payload(
+            source_report_id=171503,
+            trace_id="trace-empty-identity-600519",
+            stock_code="600519",
+            action="sell",
+        )
+    )
+
+    listed = service.list_signals(stock_identities=[], status="active")
+
+    assert listed == {"items": [], "total": 0, "page": 1, "page_size": 20}
+
+
+def test_list_signals_explicit_stock_identities_do_not_trigger_history_backfill(isolated_db) -> None:
+    record_id = isolated_db.save_analysis_history(
+        result=_history_result(operation_advice="卖出", decision_type="sell", action="sell", action_label="卖出"),
+        query_id="query-explicit-identity-no-backfill",
+        report_type="simple",
+        news_content="新闻摘要",
+        context_snapshot={"market_phase_summary": {"phase": "postmarket"}},
+        save_snapshot=True,
+    )
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    listed = service.list_signals(
+        source_type="analysis",
+        source_report_id=record_id,
+        stock_identities=[("cn", "600519")],
+    )
+
+    assert listed == {"items": [], "total": 0, "page": 1, "page_size": 20}
+    with isolated_db.get_session() as session:
+        assert session.query(DecisionSignalRecord).count() == 0
+
+
 def test_service_plan_quality_slots_and_explicit_override(isolated_db) -> None:
     service = DecisionSignalService(db_manager=isolated_db)