From c937be9d94f2228a3ac5a10515eceef6a2656e65 Mon Sep 17 00:00:00 2001
From: jinliyl <6469360+jinliyl@users.noreply.github.com>
Date: Mon, 27 Jul 2026 20:15:27 +0800
Subject: [PATCH] refactor(auto_fin): normalize data models and selection logic
 across agents (#396)

- Introduce tolerant AutoFinAgentModel base class allowing extra fields in raw Agent outputs
- Replace strict models with tolerant ones for ETF, historical event, market selection, and report outputs
- Remove redundant field validators and allow empty defaults for key string fields
- Enhance historical source path resolution to safely filter invalid or out-of-workspace paths
- Add normalization of whitespace and validation to historical event references before processing
- Implement normalization in Topic and Market Agent selections to eliminate duplicates, blanks, unknowns
- Limit Topic Agent output to top 20 ETFs and ensure sorting and deduplication of events
- Normalize final Markdown report by removing redundant headers and providing safe fallbacks
- Update agent prompts to clarify task constraints and improve instruction consistency
- Add extensive tests for normalization, filtering, and safe source resolution for historical events
---
 reme/schema/auto_fin.py                       | 109 +++-------
 .../steps/cookbook/auto_fin/history_search.py |  42 ++--
 .../cookbook/auto_fin/history_search.yaml     |  35 +--
 reme/steps/cookbook/auto_fin/market.py        |  35 ++-
 reme/steps/cookbook/auto_fin/market.yaml      |  30 +--
 reme/steps/cookbook/auto_fin/merge.py         |  20 +-
 reme/steps/cookbook/auto_fin/merge.yaml       |  36 +---
 reme/steps/cookbook/auto_fin/topic.py         |  59 +++--
 reme/steps/cookbook/auto_fin/topic.yaml       |  25 +--
 tests/unit/test_auto_fin.py                   | 202 +++++++++++++++---
 10 files changed, 358 insertions(+), 235 deletions(-)

diff --git a/reme/schema/auto_fin.py b/reme/schema/auto_fin.py
index e1647294..55680a24 100644
--- a/reme/schema/auto_fin.py
+++ b/reme/schema/auto_fin.py
@@ -26,26 +26,23 @@ ShanghaiDateTime = Annotated[datetime, BeforeValidator(_shanghai_local_time)]
 
 
 class AutoFinModel(BaseModel):
-    """Strict base for Agent output."""
+    """Strict base for program-owned Auto Fin data."""
 
     model_config = ConfigDict(extra="forbid")
 
 
-class AutoFinEtfEventReference(AutoFinModel):
+class AutoFinAgentModel(AutoFinModel):
+    """Tolerant base for raw Agent output."""
+
+    model_config = ConfigDict(extra="ignore")
+
+
+class AutoFinEtfEventReference(AutoFinAgentModel):
     """One selected news item and why it is relevant to an ETF."""
 
     reason: str
     news_id: str
 
-    @model_validator(mode="after")
-    def non_empty_values(self) -> "AutoFinEtfEventReference":
-        """Reject blank event references."""
-        self.reason = self.reason.strip()
-        self.news_id = self.news_id.strip()
-        if not self.reason or not self.news_id:
-            raise ValueError("ETF event reason and news ID must be non-empty")
-        return self
-
 
 class AutoFinSelectedEvent(AutoFinModel):
     """A selected current event with its source news reference."""
@@ -57,62 +54,33 @@ class AutoFinSelectedEvent(AutoFinModel):
     event_title: str = ""
 
 
-class AutoFinEtfSelection(AutoFinModel):
-    """One liquid ETF selected for current news."""
+class AutoFinEtfSelection(AutoFinAgentModel):
+    """One ETF selection returned by the Topic Agent."""
 
     etf_code: str
-    etf_name: str
-    events: list[AutoFinEtfEventReference] = Field(min_length=1)
-
-    @model_validator(mode="after")
-    def valid_news_ids(self) -> "AutoFinEtfSelection":
-        """Reject blank or duplicate news references."""
-        news_ids = [event.news_id for event in self.events]
-        if len(news_ids) != len(set(news_ids)):
-            raise ValueError("ETF event news IDs must be unique")
-        return self
+    etf_name: str = ""
+    events: list[AutoFinEtfEventReference] = Field(default_factory=list)
 
 
-class AutoFinEtfsOutput(AutoFinModel):
-    """Liquid ETFs related to current news, deduplicated by name and theme."""
+class AutoFinEtfsOutput(AutoFinAgentModel):
+    """ETF selections returned by the Topic Agent before normalization."""
 
-    etfs: list[AutoFinEtfSelection] = Field(default_factory=list, max_length=20)
-
-    @model_validator(mode="after")
-    def unique_etfs(self) -> "AutoFinEtfsOutput":
-        """Reject duplicate ETF codes or names."""
-        codes = [item.etf_code.strip().upper() for item in self.etfs]
-        names = [item.etf_name.strip().casefold() for item in self.etfs]
-        if any(not code for code in codes) or any(not name for name in names):
-            raise ValueError("ETF codes and names must be non-empty")
-        if len(codes) != len(set(codes)) or len(names) != len(set(names)):
-            raise ValueError("ETF codes and names must be unique")
-        return self
+    etfs: list[AutoFinEtfSelection] = Field(default_factory=list)
 
 
-class AutoFinHistoricalEventReference(AutoFinModel):
+class AutoFinHistoricalEventReference(AutoFinAgentModel):
     """One historical news item selected by the search Agent."""
 
     reason: str
     news_id: str
-    source_path: str
-
-    @model_validator(mode="after")
-    def non_empty_values(self) -> "AutoFinHistoricalEventReference":
-        """Reject references that cannot be resolved deterministically."""
-        for field in ("reason", "news_id", "source_path"):
-            value = getattr(self, field).strip()
-            if not value:
-                raise ValueError(f"historical event {field} must not be empty")
-            setattr(self, field, value)
-        return self
+    source_path: str = ""
 
 
-class AutoFinEtfHistoricalEvents(AutoFinModel):
+class AutoFinEtfHistoricalEvents(AutoFinAgentModel):
     """Historical news references returned by the search Agent."""
 
-    etf_code: str
-    etf_name: str
+    etf_code: str = ""
+    etf_name: str = ""
     historical_events: list[AutoFinHistoricalEventReference] = Field(default_factory=list)
 
 
@@ -235,36 +203,19 @@ class AutoFinEtfHistoricalResearch(AutoFinModel):
         return self
 
 
-class AutoFinHistoricalDirectionReference(AutoFinModel):
+class AutoFinHistoricalDirectionReference(AutoFinAgentModel):
     """One direction-classified historical event returned by the Market Agent."""
 
     reason: str
     news_id: str
 
-    @model_validator(mode="after")
-    def non_empty_values(self) -> "AutoFinHistoricalDirectionReference":
-        """Reject a direction judgment without source identity or rationale."""
-        self.reason = self.reason.strip()
-        self.news_id = self.news_id.strip()
-        if not self.reason or not self.news_id:
-            raise ValueError("historical direction reason and news ID must be non-empty")
-        return self
 
-
-class AutoFinMarketSelection(AutoFinModel):
+class AutoFinMarketSelection(AutoFinAgentModel):
     """Same- and opposite-direction historical events returned by the Market Agent."""
 
     same_direction_events: list[AutoFinHistoricalDirectionReference] = Field(default_factory=list)
     opposite_direction_events: list[AutoFinHistoricalDirectionReference] = Field(default_factory=list)
 
-    @model_validator(mode="after")
-    def unique_historical_news(self) -> "AutoFinMarketSelection":
-        """Reject news IDs repeated within or across direction groups."""
-        news_ids = [event.news_id for event in (*self.same_direction_events, *self.opposite_direction_events)]
-        if len(news_ids) != len(set(news_ids)):
-            raise ValueError("direction-classified historical event news IDs must be unique")
-        return self
-
 
 class AutoFinHistoricalMatch(AutoFinModel):
     """One direction-classified historical event used by the equal-weight forecast."""
@@ -341,18 +292,8 @@ class AutoFinEtfHistoryDetail(AutoFinModel):
         return self
 
 
-class AutoFinReportOutput(AutoFinModel):
+class AutoFinReportOutput(AutoFinAgentModel):
     """Final Markdown title and body for all selected ETFs."""
 
-    title: str
-    body: str
-
-    @model_validator(mode="after")
-    def non_empty_report(self) -> "AutoFinReportOutput":
-        """Require both Markdown report fields."""
-        for field in ("title", "body"):
-            value = getattr(self, field).strip()
-            if not value:
-                raise ValueError(f"{field} must not be empty")
-            setattr(self, field, value)
-        return self
+    title: str = ""
+    body: str = ""
diff --git a/reme/steps/cookbook/auto_fin/history_search.py b/reme/steps/cookbook/auto_fin/history_search.py
index f6ebe962..706dba6c 100644
--- a/reme/steps/cookbook/auto_fin/history_search.py
+++ b/reme/steps/cookbook/auto_fin/history_search.py
@@ -45,20 +45,20 @@ class AutoFinHistorySearchStep(AutoFinStep):
         return number if number > 0 else None
 
     def _historical_source_candidates(self, source_path_value: str, news_id: str) -> list[Path]:
-        """Return the declared source and a date-derived fallback within the workspace."""
+        """Return safe declared and date-derived source candidates within the workspace."""
         workspace = self.workspace_path.resolve()
         relative_path = Path(source_path_value)
-        if relative_path.is_absolute() or ".." in relative_path.parts:
-            raise ValueError(f"Historical source path must be workspace-relative: {source_path_value}")
-        source_path = (workspace / relative_path).resolve()
-        try:
-            source_path.relative_to(workspace)
-        except ValueError as exc:
-            raise ValueError(f"Historical source path is outside the workspace: {source_path_value}") from exc
-        if source_path.name != "auto_fin_news_data.jsonl":
-            raise ValueError(f"Historical source must be an Auto Fin news file: {source_path_value}")
+        candidates = []
+        if not relative_path.is_absolute() and ".." not in relative_path.parts:
+            source_path = (workspace / relative_path).resolve()
+            try:
+                source_path.relative_to(workspace)
+            except ValueError:
+                pass
+            else:
+                if source_path.name == "auto_fin_news_data.jsonl":
+                    candidates.append(source_path)
 
-        candidates = [source_path]
         news_date = news_id.partition("_")[0][:8]
         try:
             parsed_date = datetime.strptime(news_date, "%Y%m%d").date()
@@ -68,6 +68,8 @@ class AutoFinHistorySearchStep(AutoFinStep):
             inferred_path = workspace / "daily" / parsed_date.isoformat() / "auto_fin_news_data.jsonl"
             if inferred_path not in candidates:
                 candidates.append(inferred_path)
+        if not candidates:
+            raise ValueError(f"Historical source cannot be inferred safely: {source_path_value}")
         return candidates
 
     async def _resolve_historical_event(
@@ -129,6 +131,18 @@ class AutoFinHistorySearchStep(AutoFinStep):
         events_by_news_id: dict[str, AutoFinHistoricalEvent] = {}
         limitations = []
         for reference in references:
+            reference = reference.model_copy(
+                update={
+                    "reason": reference.reason.strip(),
+                    "news_id": reference.news_id.strip(),
+                    "source_path": reference.source_path.strip(),
+                },
+            )
+            if not reference.reason or not reference.news_id:
+                limitation = "跳过缺少 reason 或 news_id 的历史新闻"
+                self.logger.warning(f"[{self.name}] {limitation}")
+                limitations.append(limitation)
+                continue
             try:
                 event = await self._resolve_historical_event(
                     reference,
@@ -287,8 +301,6 @@ class AutoFinHistorySearchStep(AutoFinStep):
                     contexts.pop(tool_context_id, None)
                     if not contexts:
                         self.app_context.metadata.pop(_ToolContextDedupMixin.TOOL_CONTEXTS_KEY, None)
-        if (history.etf_code, history.etf_name) != (item.etf_code, item.etf_name):
-            raise ValueError(f"History Agent changed ETF {label!r}")
         resolved_events, resolution_limitations = await self._resolve_historical_events(
             history.historical_events,
             {event.news_id for event in events},
@@ -309,8 +321,8 @@ class AutoFinHistorySearchStep(AutoFinStep):
             for event, sample in zip(resolved_events, samples, strict=True)
         ]
         enriched_history = AutoFinEtfHistoricalResearch(
-            etf_code=history.etf_code,
-            etf_name=history.etf_name,
+            etf_code=item.etf_code,
+            etf_name=item.etf_name,
             historical_events=enriched_events,
             limitations=list(dict.fromkeys([*resolution_limitations, *market_limitations])),
         )
diff --git a/reme/steps/cookbook/auto_fin/history_search.yaml b/reme/steps/cookbook/auto_fin/history_search.yaml
index c8f480d6..b94a13a9 100644
--- a/reme/steps/cookbook/auto_fin/history_search.yaml
+++ b/reme/steps/cookbook/auto_fin/history_search.yaml
@@ -1,33 +1,19 @@
 history_search_user: |
-  你只负责为一个已选定 ETF 选择历史相似新闻，不查询行情、不复制新闻正文、不更换或增加 ETF、
-  不提供预测。时间、标题、正文和行情都由程序在你返回后补充。
-  指定 ETF：{etf_code}（{etf_name}）
-  当前事件时间线（仅作为检索线索，不含 news_id）：{events}
+  为 {etf_code}（{etf_name}）寻找与当前事件相似的历史新闻。不查询行情、不计算收益或预测。
+
+  当前事件：{events}
   历史截止时间：{window_start}
   ReMe workspace：{workspace_root}
 
-  必须遵守：
-  1. 使用 memory_search 多轮搜索 {window_start} 之前的相似新闻；系统会自动过滤本次研究中已经
-     返回过的结果，因此结果不足时继续尝试，并调整事件类型、关键实体、传导机制或影响方向等
-     查询角度。memory_search 无更多有效结果后，使用 Read 或 Python 扫描 workspace 下过去
-     360 天的 `daily/YYYY-MM-DD/auto_fin_news_data.jsonl`。历史 Auto Fin Markdown 只能作为
-     检索线索，最终选择必须能回溯到上述新闻 JSONL。
-  2. 根据当前事件提取事件类型、关键实体、传导机制和影响方向进行搜索，不能只搜索 ETF 名称。
-  3. 当前事件时间线来自最近一次收盘后至本次分析时点，只能作为检索线索，不属于历史事件。
-     即使搜索命中同一条当前新闻，也不得放入 historical_events。
-  4. 每个 historical_events 项只需从原始 JSONL 逐字复制 news_id 和 workspace 相对
-     source_path，并用 reason 简洁说明它与当前事件相似的原因。source_path 的 YYYY-MM-DD 必须
-     是实际文件所在日期，通常与 news_id 开头的日期一致。不要返回时间、标题或正文。
-  5. 只选择原始记录中真实存在且带有 news_id 的新闻。程序会严格使用 source_path + news_id
-     回查，找不到、重复命中、路径越界或不是 auto_fin_news_data.jsonl 都会失败；严禁编造。
-  6. etf_code 和 etf_name 必须原样返回 `{etf_code}` 和 `{etf_name}`。不要读写或下载行情文件，
-     不要计算收益。没有合适新闻时直接返回空 historical_events，无需说明限制。
-  7. 最终只生成 etf_code、etf_name 和 historical_events 三个维度；
-     historical_events 的每一项只包含 reason、news_id 和 source_path。JSON 示例：
+  工作要求：
+  1. 使用 memory_search 从事件类型、关键实体、传导机制和影响方向等角度搜索截止时间之前的
+     相似新闻；结果不足时调整角度继续搜索。必要时扫描 workspace 过去 360 天的
+     `daily/YYYY-MM-DD/auto_fin_news_data.jsonl`。
+  2. 选择机制可比的历史事件，说明相似原因，并返回原始新闻的 news_id 和 source_path。
+     当前事件不属于历史事件。程序会回查、过滤、去重、排序并补充行情。
+  3. 返回以下 JSON；没有合适新闻时 historical_events 为空：
      ```json
      {{
-       "etf_code": "{etf_code}",
-       "etf_name": "{etf_name}",
        "historical_events": [
          {{
            "reason": "事件类型、关键实体、传导机制和影响方向相似",
@@ -37,4 +23,3 @@ history_search_user: |
        ]
      }}
      ```
-  8. 最终只生成上述 JSON 对象，不附加解释或 Markdown 正文。无需排序或计算，程序会完成。
diff --git a/reme/steps/cookbook/auto_fin/market.py b/reme/steps/cookbook/auto_fin/market.py
index 74c4ae4d..8b77ff99 100644
--- a/reme/steps/cookbook/auto_fin/market.py
+++ b/reme/steps/cookbook/auto_fin/market.py
@@ -19,6 +19,35 @@ from ._base import AutoFinStep, _write
 class AutoFinMarketStep(AutoFinStep):
     """Classify historical event directions and calculate one ETF forecast."""
 
+    @staticmethod
+    def _normalize_selection(
+        selection: AutoFinMarketSelection,
+        history: AutoFinEtfHistoricalResearch,
+    ) -> tuple[AutoFinMarketSelection, bool]:
+        """Filter unknown, blank, and duplicate direction references."""
+        known_news_ids = {event.news_id for event in history.historical_events}
+        seen_news_ids: set[str] = set()
+
+        def valid_events(events):
+            normalized = []
+            for event in events:
+                reason = event.reason.strip()
+                news_id = event.news_id.strip()
+                if not reason or news_id not in known_news_ids or news_id in seen_news_ids:
+                    continue
+                normalized.append({"reason": reason, "news_id": news_id})
+                seen_news_ids.add(news_id)
+            return normalized
+
+        normalized_selection = AutoFinMarketSelection.model_validate(
+            {
+                "same_direction_events": valid_events(selection.same_direction_events),
+                "opposite_direction_events": valid_events(selection.opposite_direction_events),
+            },
+        )
+        changed = normalized_selection.model_dump(mode="json") != selection.model_dump(mode="json")
+        return normalized_selection, changed
+
     @staticmethod
     def _calculate_analysis(
         item: AutoFinEtfSelection,
@@ -31,9 +60,6 @@ class AutoFinMarketStep(AutoFinStep):
             *((match, "same", 1.0) for match in selection.same_direction_events),
             *((match, "opposite", -1.0) for match in selection.opposite_direction_events),
         ]
-        unknown_news_ids = {match.news_id for match, _, _ in selected if match.news_id not in history_by_news_id}
-        if unknown_news_ids:
-            raise ValueError(f"Market Agent referenced unknown historical news: {sorted(unknown_news_ids)}")
 
         weight = 1.0 / len(selected) if selected else 0.0
         matches = [
@@ -126,6 +152,9 @@ class AutoFinMarketStep(AutoFinStep):
                 self.workspace_path / "resource" / str(self._required("auto_fin_date")) / f"{resource_name}_output.json"
             )
             self.logger.warning(f"[{self.name}] skip direction Agent for {item.etf_code}: no valid history")
+        selection, normalized = self._normalize_selection(selection, history)
+        if normalized:
+            self.logger.info(f"[{self.name}] normalized historical direction selections")
         analysis = self._calculate_analysis(item, history, selection)
         _write(
             selection_path,
diff --git a/reme/steps/cookbook/auto_fin/market.yaml b/reme/steps/cookbook/auto_fin/market.yaml
index 7e3d3edd..56273c89 100644
--- a/reme/steps/cookbook/auto_fin/market.yaml
+++ b/reme/steps/cookbook/auto_fin/market.yaml
@@ -1,30 +1,21 @@
 market_user: |
-  你只负责筛选与当前事件机制可比的历史事件，并判断影响方向相同还是相反；不计算收益、
-  不生成预测、不总结报告。
+  筛选与当前事件机制可比的历史事件，并判断其影响方向与当前事件相同还是相反。
+  不计算收益或生成预测。
+
   ETF：{etf_code}（{etf_name}）
   分析截止时间：{decision_at}
-
   当前事件：
   {events}
+  历史事件文件：{history_path}
 
-  已补全的历史事件文件：
-  {history_path}
-
-  必须遵守：
-  1. 读取历史文件，只从 historical_events 中选择与当前事件相似的事件。
-  2. 综合事件类型、关键实体、传导机制和影响方向进行判断：
+  工作要求：
+  1. 根据历史事件的时间、标题、正文和 reason，综合事件类型、关键实体、传导机制和影响方向：
      - 机制可比且影响方向相同，放入 same_direction_events；
      - 机制可比但影响方向相反，放入 opposite_direction_events；
-     - 没有有效关系，不要返回。
-     这里的“相同/相反”是历史事件相对当前事件的影响方向，不是简单判断新闻利好或利空。例如
-     当前事件是黄金涨价、历史事件是黄金降价时，应放入 opposite_direction_events。
-  3. 只根据历史事件的 event_time、event_title、event_content 和 reason 判断相似性；不要依据
-     market_entry 或 future_returns 选择事件，避免使用事后行情影响相似度判断。
-  4. reason 简洁说明相似之处，news_id 必须从历史文件逐字复制，严禁编造。
-  5. 不需要返回 ETF、事件时间、权重、预测、持有天数、代码、总结或 limitations；程序会校验
-     news_id，并完成所有计算。没有相似事件时两组都返回空列表。
-  6. 同一 news_id 只能出现一次，不能同时放入两组。每项只包含 reason、news_id。
-  7. 最终只生成 same_direction_events 和 opposite_direction_events。JSON 示例：
+     - 机制不可比则不选择。
+     不要依据 market_entry 或 future_returns 选择事件，避免使用事后行情。
+  2. 返回 news_id 和判断理由。程序会过滤、去重并完成计算。
+  3. 返回以下 JSON：
      ```json
      {{
        "same_direction_events": [
@@ -41,4 +32,3 @@ market_user: |
        ]
      }}
      ```
-  8. 最终只生成上述 JSON 对象，不附加解释或 Markdown 正文。
diff --git a/reme/steps/cookbook/auto_fin/merge.py b/reme/steps/cookbook/auto_fin/merge.py
index b0b2e422..21c7c3ac 100644
--- a/reme/steps/cookbook/auto_fin/merge.py
+++ b/reme/steps/cookbook/auto_fin/merge.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import re
 from types import SimpleNamespace
 
 from ....components import R
@@ -15,6 +16,16 @@ from ._base import AutoFinStep, _write, _write_jsonl
 class AutoFinMergeStep(AutoFinStep):
     """Ask a fresh Agent for the final Markdown and persist it directly."""
 
+    @staticmethod
+    def _normalize_report(output: AutoFinReportOutput) -> AutoFinReportOutput:
+        """Normalize cosmetic report fields and provide safe empty fallbacks."""
+        title = re.sub(r"^#+\s*", "", output.title.strip()) or "Auto Fin ETF 结论"
+        body = output.body.strip() or "## 结论\n\n暂无可用结论。"
+        first_line, separator, remainder = body.partition("\n")
+        if first_line.lstrip().startswith("# "):
+            body = remainder.lstrip() if separator else "## 结论\n\n暂无可用结论。"
+        return AutoFinReportOutput(title=title, body=body)
+
     @staticmethod
     def _calculation_results(history_details: list[AutoFinEtfHistoryDetail]) -> list[dict]:
         """Return the program-calculated forecast for every analyzed ETF."""
@@ -45,7 +56,7 @@ class AutoFinMergeStep(AutoFinStep):
         self.logger.info(
             f"[{self.name}] start etfs={len(etfs)}",
         )
-        output, _ = await self._reply(
+        output, output_path = await self._reply(
             "merge_user",
             "auto_fin_merge",
             AutoFinReportOutput,
@@ -55,6 +66,13 @@ class AutoFinMergeStep(AutoFinStep):
             history_path=str(self._required("auto_fin_history_resource")),
             calculation_results=json.dumps(calculation_results, ensure_ascii=False),
         )
+        normalized_output = self._normalize_report(output)
+        if normalized_output != output:
+            _write(
+                output_path,
+                json.dumps(normalized_output.model_dump(mode="json"), ensure_ascii=False, separators=(",", ":")) + "\n",
+            )
+        output = normalized_output
         markdown = f"# {output.title}\n\n{output.body}\n\n"
         markdown += "> 仅为事件研究和持有时间参考，不构成投资建议，不会执行交易。\n"
         day_dir = self.workspace_path / str(self.config_value("daily_dir")) / str(self._required("auto_fin_date"))
diff --git a/reme/steps/cookbook/auto_fin/merge.yaml b/reme/steps/cookbook/auto_fin/merge.yaml
index dcd72784..b4c156f6 100644
--- a/reme/steps/cookbook/auto_fin/merge.yaml
+++ b/reme/steps/cookbook/auto_fin/merge.yaml
@@ -1,39 +1,23 @@
 merge_user: |
-  你只负责把已经完成的结构化分析写成一份中文 Markdown 报告，不重新搜索新闻、不下载行情、
-  不修改任何数值。
+  把已经完成的结构化分析写成中文 Markdown 结论，不重新搜索新闻、下载行情或修改数值。
 
   分析截止时间：{decision_at}
   新闻窗口：({window_start}, {decision_at}]
+  ETF 与当前事件：{etfs_path}
+  历史事件、行情样本及分析：{history_path}
+  程序计算结果：{calculation_results}
 
-  已筛选 ETF 及其当前事件时间线：
-  {etfs_path}
-
-  各 ETF 的完整历史事件、实际行情样本和程序计算结果：
-  {history_path}
-
-  全部 ETF 的程序计算结果（包含正向、负向、零值和缺失值）：
-  {calculation_results}
-
-  必须遵守：
-  1. 只读取上述两个文件，严格使用其中已有的 ETF、事件、收益、权重、最佳持有天数和 limitations。
-  2. body 只写最终结论，以推荐 ETF 为主要内容。
-  3. 由你根据文件中的当前事件内容和传导关系，自行判断事件对 ETF 的影响方向。推荐 ETF 必须
-     同时满足：
+  工作要求：
+  1. 根据当前事件内容和传导关系判断影响方向。只推荐同时满足以下条件的 ETF：
      - 经你判断，当前事件对该 ETF 所代表的资产、行业或主题影响明确为正向；
      - 程序给出了最佳持有天数，且该天数对应的 expected_return 大于 0。
-     不得使用程序计算结果反推事件方向。事件影响为负向、中性、方向不明，或计算结果为 null、
-     0、负数时，一律不得推荐。
-  4. 若没有同时满足两项条件的 ETF，只输出观望结论。不得因为计算为正就把事件方向不明的 ETF
-     推荐出来。
-  5. 如有推荐，结论需要列 ETF code、name、最佳持有天数、对应预估收益。
-  6. 对事件影响为负向或计算结果没有正值的 ETF，在推荐结论后用一句话合并简述；不要逐只展开。
-     中性、方向不明或结果缺失的 ETF 可以不写。
-  7. title 不包含 Markdown 标记，body 不重复一级标题，不生成 YAML frontmatter。
-  8. 最终只生成 title 和 body。JSON 示例：
+     不使用计算结果反推事件方向；不满足条件时给出观望结论。
+  2. 推荐结论包含 ETF code、name、最佳持有天数和对应预估收益；负向或无正收益的情况可以合并
+     简述。
+  3. 返回以下 JSON：
      ```json
      {{
        "title": "Auto Fin ETF 结论",
        "body": "## 结论\n\n推荐 518880.SH（黄金ETF），参考持有 3 个交易日，当前加权预估收益 +1.2%；......。\n\n负向提示：相关能源 ETF 事件影响偏负，不推荐。"
      }}
      ```
-  9. 最终只生成上述 JSON 对象，不附加解释或代码块标记。
diff --git a/reme/steps/cookbook/auto_fin/topic.py b/reme/steps/cookbook/auto_fin/topic.py
index a38970f0..bd59c591 100644
--- a/reme/steps/cookbook/auto_fin/topic.py
+++ b/reme/steps/cookbook/auto_fin/topic.py
@@ -16,6 +16,7 @@ NEWS_TITLE_MAX_CHARS = 200
 NEWS_CONTENT_MAX_CHARS = 1200
 NEWS_TOTAL_CONTENT_MAX_CHARS = 60_000
 ETF_CANDIDATE_LIMIT = 150
+ETF_OUTPUT_LIMIT = 20
 
 
 class _NewsTextExtractor(HTMLParser):
@@ -167,7 +168,9 @@ class AutoFinTopicStep(AutoFinStep):
         repairs: dict[str, str] = {}
         for item in data["etfs"]:
             for event in item["events"]:
-                news_id = event["news_id"]
+                event["reason"] = event["reason"].strip()
+                news_id = event["news_id"].strip()
+                event["news_id"] = news_id
                 if news_id in news_ids:
                     continue
                 _, separator, suffix = news_id.rpartition("_")
@@ -178,26 +181,47 @@ class AutoFinTopicStep(AutoFinStep):
         return AutoFinEtfsOutput.model_validate(data), repairs
 
     @staticmethod
-    def _validate_selection(
+    def _normalize_selection(
         output: AutoFinEtfsOutput,
         news: list[dict[str, Any]],
         etfs: list[dict[str, str]],
-    ) -> None:
+    ) -> tuple[AutoFinEtfsOutput, bool]:
+        """Canonicalize, filter, deduplicate, sort, and limit Agent selections."""
         news_order = {str(row["news_id"]): index for index, row in enumerate(news)}
         news_ids = set(news_order)
-        candidates = {row["code"]: row["name"] for row in etfs}
+        candidates = {str(row["code"]).strip().upper(): str(row["name"]).strip() for row in etfs}
+        selected: dict[str, dict[str, Any]] = {}
         for item in output.etfs:
-            if candidates.get(item.etf_code) != item.etf_name:
-                raise ValueError(f"Topic Agent returned an ETF outside filtered_etf.jsonl: {item.etf_code}")
-            selected_news_ids = [event.news_id for event in item.events]
-            unknown = set(selected_news_ids) - news_ids
-            if unknown:
-                raise ValueError(f"Topic Agent returned unknown news IDs: {sorted(unknown)}")
-            if len(selected_news_ids) != len(set(selected_news_ids)):
-                raise ValueError(f"Topic Agent returned duplicate news IDs for ETF: {item.etf_code}")
-            event_order = [news_order[news_id] for news_id in selected_news_ids]
-            if event_order != sorted(event_order):
-                raise ValueError(f"Topic Agent returned unsorted news IDs for ETF: {item.etf_code}")
+            code = item.etf_code.strip().upper()
+            name = candidates.get(code)
+            if name is None:
+                continue
+            normalized = selected.setdefault(
+                code,
+                {"etf_code": code, "etf_name": name, "events": [], "seen_news_ids": set()},
+            )
+            for event in item.events:
+                if not event.reason or event.news_id not in news_ids or event.news_id in normalized["seen_news_ids"]:
+                    continue
+                normalized["events"].append(event.model_dump(mode="json"))
+                normalized["seen_news_ids"].add(event.news_id)
+
+        normalized_items = []
+        for item in selected.values():
+            events = sorted(item["events"], key=lambda event: news_order[event["news_id"]])
+            if events:
+                normalized_items.append(
+                    {
+                        "etf_code": item["etf_code"],
+                        "etf_name": item["etf_name"],
+                        "events": events,
+                    },
+                )
+            if len(normalized_items) >= ETF_OUTPUT_LIMIT:
+                break
+        normalized_output = AutoFinEtfsOutput.model_validate({"etfs": normalized_items})
+        changed = normalized_output.model_dump(mode="json") != output.model_dump(mode="json")
+        return normalized_output, changed
 
     async def execute(self):
         assert self.context is not None
@@ -227,10 +251,13 @@ class AutoFinTopicStep(AutoFinStep):
             filtered_etf_path=str(etf_path),
         )
         output, repairs = self._repair_news_ids(output, news)
+        output, normalized = self._normalize_selection(output, news, etfs)
         if repairs:
             self.logger.warning(f"[{self.name}] repaired mistyped news IDs: {repairs}")
+        if normalized:
+            self.logger.info(f"[{self.name}] normalized ETF selections")
+        if repairs or normalized:
             _write_jsonl(output_path, output.model_dump(mode="json")["etfs"])
-        self._validate_selection(output, news, etfs)
         self.context["auto_fin_window_start"] = window_start.isoformat()
         self.context["auto_fin_etfs"] = output.model_dump(mode="json")["etfs"]
         self.context["auto_fin_etfs_resource"] = str(output_path)
diff --git a/reme/steps/cookbook/auto_fin/topic.yaml b/reme/steps/cookbook/auto_fin/topic.yaml
index 722ba487..7f1bbda0 100644
--- a/reme/steps/cookbook/auto_fin/topic.yaml
+++ b/reme/steps/cookbook/auto_fin/topic.yaml
@@ -1,27 +1,15 @@
 topic_user: |
-  你只负责从候选文件中筛选与当前新闻直接相关的代表性 ETF，并返回对应新闻 ID。
+  你只负责筛选与当前新闻直接相关的代表性 ETF，并说明每条相关新闻与 ETF 的关系。
   不搜索或补充新闻、不下载数据、不计算收益、不提供预测或投资建议。
 
-  时间边界：
-  - 新闻窗口严格为 ({window_start}, {decision_at}]。
-  - 分析截止时间为 {decision_at}。
-
   输入文件：
-  - 当前窗口内的全部新闻：{filtered_news_path}
-  - 已按最近收盘成交额关联、去重并截取 Top 150 的 ETF：{filtered_etf_path}
+  - 新闻（时间窗口为 ({window_start}, {decision_at}]）：{filtered_news_path}
+  - 候选 ETF：{filtered_etf_path}
 
   输出要求：
-  1. 必须读取上述两个 JSONL 文件；只能从 filtered_etf.jsonl 选择 ETF，只能引用
-     filtered_news.jsonl 中的 news_id。
-  2. 只保留其 name 所代表的指数、行业、主题或资产与新闻存在明确传导关系的代表性 ETF。
-  3. 最多返回 20 只 ETF，按与新闻的相关性排序；不要为了凑数纳入弱相关 ETF。
-  4. events 只填写与该 ETF 相关的事件对象，每项包含 reason 和 news_id；reason 简洁说明该条新闻
-     与 ETF 所代表资产之间的直接传导关系。合并重复报道，并按新闻时间升序排列。
-  5. etf_code、etf_name 必须逐字复制候选文件中的 code、name；无相关 ETF 时返回空 etfs。
-     news_id 也必须整段逐字复制，禁止把一条新闻的时间前缀与另一条新闻的哈希后缀拼接。
-  6. reason 不得加入预测；同一 ETF 的不同新闻应分别说明关联理由。
-  7. 最终生成一个对象，只包含 etfs；每只 ETF 只包含 etf_code、etf_name、events，每个 event
-     只包含 reason 和 news_id。JSON 示例：
+  1. 从候选 ETF 中选择与新闻有明确传导关系的代表性 ETF，不要为了凑数纳入弱相关 ETF。
+  2. 为每只 ETF 返回相关新闻的 news_id，并用 reason 简洁说明直接传导关系。
+  3. 返回以下 JSON：
      ```json
      {{
        "etfs": [
@@ -42,4 +30,3 @@ topic_user: |
        ]
      }}
      ```
-  8. 最终只生成上述 JSON 对象，不附加解释或 Markdown 正文，不写入 ReMe workspace。
diff --git a/tests/unit/test_auto_fin.py b/tests/unit/test_auto_fin.py
index 1da5784d..9a19505a 100644
--- a/tests/unit/test_auto_fin.py
+++ b/tests/unit/test_auto_fin.py
@@ -114,6 +114,76 @@ def test_topic_does_not_repair_ambiguous_content_hash():
     assert not repairs
 
 
+def test_topic_normalizes_duplicate_and_invalid_selections():
+    output = AutoFinEtfsOutput.model_validate(
+        {
+            "etfs": [
+                {
+                    "etf_code": "159516.SZ",
+                    "etf_name": "半导体设备ETF",
+                    "events": [
+                        {"reason": "较晚新闻", "news_id": "20260724164534_9368"},
+                        {"reason": "较早新闻", "news_id": "20260724164043_7332"},
+                        {"reason": "重复新闻", "news_id": "20260724164043_7332"},
+                        {"reason": "未知新闻", "news_id": "20260724170000_ffff"},
+                        {"reason": " ", "news_id": "20260724170000_abcd"},
+                    ],
+                },
+                {
+                    "etf_code": "159516.SZ",
+                    "etf_name": "模型返回的错误名称",
+                    "events": [{"reason": "另一条新闻", "news_id": "20260724170000_abcd"}],
+                },
+                {
+                    "etf_code": "000000.SZ",
+                    "etf_name": "候选范围外ETF",
+                    "events": [{"reason": "范围外", "news_id": "20260724164043_7332"}],
+                },
+            ],
+        },
+    )
+    news = [
+        {"news_id": "20260724164043_7332"},
+        {"news_id": "20260724164534_9368"},
+        {"news_id": "20260724170000_abcd"},
+    ]
+    etfs = [{"code": "159516.SZ", "name": "国泰中证半导体材料设备主题ETF"}]
+
+    normalized, changed = AutoFinTopicStep._normalize_selection(output, news, etfs)
+
+    assert changed
+    assert len(normalized.etfs) == 1
+    assert normalized.etfs[0].etf_name == "国泰中证半导体材料设备主题ETF"
+    assert [event.news_id for event in normalized.etfs[0].events] == [
+        "20260724164043_7332",
+        "20260724164534_9368",
+        "20260724170000_abcd",
+    ]
+
+
+def test_topic_limits_normalized_output_in_code():
+    output = AutoFinEtfsOutput.model_validate(
+        {
+            "etfs": [
+                {
+                    "etf_code": f"{index:06d}.SZ",
+                    "etf_name": f"ETF {index}",
+                    "events": [{"reason": "相关事件", "news_id": "20260724164043_7332"}],
+                }
+                for index in range(21)
+            ],
+        },
+    )
+    news = [{"news_id": "20260724164043_7332"}]
+    etfs = [{"code": f"{index:06d}.SZ", "name": f"ETF {index}"} for index in range(21)]
+
+    normalized, changed = AutoFinTopicStep._normalize_selection(output, news, etfs)
+
+    assert changed
+    assert len(normalized.etfs) == 20
+    assert [item.etf_code for item in normalized.etfs] == [f"{index:06d}.SZ" for index in range(20)]
+
+
 def test_published_time_is_normalized_once_to_shanghai_local_time():
     parsed = AutoFinDataStep._published_at({"published_at": "2026-07-24T01:00:00+00:00"})
 
@@ -315,8 +385,7 @@ class _Agent(BaseAgentWrapper):
         if schema is AutoFinEtfsOutput:
             assert "filtered_news.jsonl" in task
             assert "filtered_etf.jsonl" in task
-            assert "Top 150" in task
-            assert "最多返回 20" in task
+            assert "最多返回 20" not in task
             value = {
                 "etfs": [
                     {
@@ -349,19 +418,16 @@ class _Agent(BaseAgentWrapper):
                 f"20260724090000_" f"{hashlib.sha256('财联社供应恢复时间仍不确定'.encode()).hexdigest()[:4]}"
             )
             assert current_news_id not in task
-            assert "当前事件时间线（仅作为检索线索，不含 news_id）" in task
-            assert "时间、标题、正文和行情都由程序在你返回后补充" in task
-            assert "每一项只包含 reason、news_id 和 source_path" in task
+            assert "程序会回查、过滤、去重、排序并补充行情" in task
             tool_context_id = kwargs.get("tool_context_id", "")
             assert tool_context_id.startswith("auto_fin_history_01_159018.SZ_")
             assert tool_context_id not in task
-            assert "系统会自动过滤本次研究中已经" in task
             self.app_context.metadata.setdefault("tool_contexts", {})[tool_context_id] = {
                 "search_seen_chunk_ids": {},
             }
             value = {
-                "etf_code": "159018.SZ",
-                "etf_name": "油气ETF",
+                "etf_code": "changed",
+                "etf_name": "changed",
                 "historical_events": [
                     {
                         "reason": "供应中断的事件类型和传导机制相同",
@@ -375,13 +441,16 @@ class _Agent(BaseAgentWrapper):
         elif schema is AutoFinMarketSelection:
             assert "ETF：159018.SZ（油气ETF）" in task
             assert "[2026-07-23T16:00:00] 原油供应中断" in task
-            assert "判断影响方向相同还是相反" in task
+            assert "影响方向与当前事件相同还是相反" in task
             assert "不要依据" in task
-            assert "程序会校验" in task
-            assert "每项只包含 reason、news_id" in task
+            assert "程序会过滤、去重并完成计算" in task
             assert "$tushare-data" not in task
             history_path = Path(
-                next(line.strip() for line in task.splitlines() if line.strip().endswith("_output.json")),
+                next(
+                    line.rsplit("：", 1)[-1].strip()
+                    for line in task.splitlines()
+                    if line.strip().endswith("_output.json")
+                ),
             )
             history = json.loads(history_path.read_text(encoding="utf-8"))
             assert "historical_samples" not in history
@@ -400,15 +469,12 @@ class _Agent(BaseAgentWrapper):
         elif schema is AutoFinReportOutput:
             assert "不重新搜索新闻" in task
             assert "auto_fin_history_output.jsonl" in task
-            assert "不生成 YAML frontmatter" in task
             assert '"etf_code": "159018.SZ"' in task
             assert '"suggested_holding_days": 10' in task
             assert '"horizon": 1' in task
             assert '"horizon": 10' in task
-            assert "自行判断事件对 ETF 的影响方向" in task
-            assert "不得使用程序计算结果反推事件方向" in task
-            assert "以推荐 ETF 为主要内容" in task
-            assert "用一句话合并简述" in task
+            assert "不使用计算结果反推事件方向" in task
+            assert "负向或无正收益的情况可以合并" in task
             value = {
                 "title": "Auto Fin ETF 结论",
                 "body": "## 结论\n\n推荐 159018.SZ（油气ETF），参考持有 10 个交易日，"
@@ -570,7 +636,7 @@ async def test_four_step_pipeline_writes_plain_markdown_and_cleans_temporary_dat
     assert sum("agent input prompt=" in line for line in logs) == 2
     assert sum("agent output prompt=" in line for line in logs) == 2
     assert all("agent start prompt=" not in line and "agent done prompt=" not in line for line in logs)
-    assert any('query="你只负责从候选文件中筛选' in line for line in logs)
+    assert any('query="你只负责筛选与当前新闻直接相关' in line for line in logs)
     assert any('output={"etfs":[{"etf_code":"159018.SZ"' in line for line in logs)
     resource_dir = tmp_path / "resource" / "2026-07-24"
     filtered_news = AutoFinDataStep._read_jsonl_sync(resource_dir / "filtered_news.jsonl")
@@ -721,16 +787,61 @@ def test_market_calculation_equal_weights_and_reverses_opposite_direction_event(
     assert "相似历史样本的收益方向存在分歧" in analysis.limitations
 
 
-def test_market_selection_rejects_news_repeated_across_direction_groups():
+def test_market_selection_filters_duplicate_unknown_and_blank_references():
     duplicate = {"reason": "方向判断", "news_id": "20260601100000_abcd"}
+    selection = AutoFinMarketSelection.model_validate(
+        {
+            "same_direction_events": [
+                duplicate,
+                {"reason": " ", "news_id": "20260602100000_efgh"},
+                {"reason": "不存在", "news_id": "20260603100000_dead"},
+            ],
+            "opposite_direction_events": [duplicate],
+            "ignored_extra_field": True,
+        },
+    )
+    history = AutoFinEtfHistoricalResearch.model_validate(
+        {
+            "etf_code": "518880.SH",
+            "etf_name": "黄金ETF",
+            "historical_events": [
+                {
+                    "reason": "历史事件",
+                    "news_id": "20260601100000_abcd",
+                    "source_path": "daily/2026-06-01/auto_fin_news_data.jsonl",
+                    "event_time": "2026-06-01T10:00:00",
+                    "event_title": "黄金上涨",
+                    "event_content": "黄金价格上涨。",
+                },
+                {
+                    "reason": "历史事件",
+                    "news_id": "20260602100000_efgh",
+                    "source_path": "daily/2026-06-02/auto_fin_news_data.jsonl",
+                    "event_time": "2026-06-02T10:00:00",
+                    "event_title": "美元变化",
+                    "event_content": "美元发生变化。",
+                },
+            ],
+        },
+    )
 
-    with pytest.raises(ValueError, match="news IDs must be unique"):
-        AutoFinMarketSelection.model_validate(
-            {
-                "same_direction_events": [duplicate],
-                "opposite_direction_events": [duplicate],
-            },
-        )
+    normalized, changed = AutoFinMarketStep._normalize_selection(selection, history)
+
+    assert changed
+    assert [event.news_id for event in normalized.same_direction_events] == ["20260601100000_abcd"]
+    assert not normalized.opposite_direction_events
+
+
+def test_merge_normalizes_cosmetic_or_empty_report_fields():
+    normalized = AutoFinMergeStep._normalize_report(
+        AutoFinReportOutput(title="## Auto Fin ETF 结论 ", body="# 重复标题\n\n## 结论\n\n观望。"),
+    )
+    fallback = AutoFinMergeStep._normalize_report(AutoFinReportOutput.model_validate({}))
+
+    assert normalized.title == "Auto Fin ETF 结论"
+    assert normalized.body == "## 结论\n\n观望。"
+    assert fallback.title == "Auto Fin ETF 结论"
+    assert fallback.body == "## 结论\n\n暂无可用结论。"
 
 
 @pytest.mark.asyncio
@@ -878,6 +989,45 @@ async def test_history_search_recovers_source_path_from_news_id_date(tmp_path: P
     assert not limitations
 
 
+@pytest.mark.asyncio
+async def test_history_search_ignores_unsafe_source_path_and_uses_news_id_date(tmp_path: Path):
+    actual_path = tmp_path / "daily" / "2026-04-26" / "auto_fin_news_data.jsonl"
+    actual_path.parent.mkdir(parents=True)
+    actual_path.write_text(
+        json.dumps(
+            {
+                "news_id": "20260426221449_de86",
+                "pub_time": "2026-04-26 22:14:49",
+                "title": "有效历史新闻",
+                "content": "有效内容",
+            },
+            ensure_ascii=False,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    step = AutoFinHistorySearchStep(
+        app_context=ApplicationContext(workspace_dir=str(tmp_path), timezone="Asia/Shanghai"),
+    )
+    references = [
+        AutoFinHistoricalEventReference(
+            reason="相似事件",
+            news_id="20260426221449_de86",
+            source_path="/tmp/not-allowed.jsonl",
+        ),
+    ]
+
+    events, limitations = await step._resolve_historical_events(
+        references,
+        set(),
+        datetime.fromisoformat("2026-07-24T15:00:00"),
+    )
+
+    assert [event.news_id for event in events] == ["20260426221449_de86"]
+    assert events[0].source_path == "daily/2026-04-26/auto_fin_news_data.jsonl"
+    assert not limitations
+
+
 @pytest.mark.asyncio
 async def test_history_search_skips_one_invalid_reference_and_continues(tmp_path: Path):
     actual_path = tmp_path / "daily" / "2026-04-26" / "auto_fin_news_data.jsonl"