[#697] fix: 新闻时效硬过滤与策略分窗 (#707)

* fix: enforce strict news freshness windows (#697)

* fix: normalize timezone-aware publish date parsing (#697)

* fix: align analyzer news window with runtime pipeline config (#697)

.env.example

@@ -110,6 +110,9 @@ SEARXNG_BASE_URLS=
 # ===================================
 # 新闻时效与分析筛选配置
 # ===================================
+# 新闻策略窗口档位：ultra_short(1天) / short(3天) / medium(7天) / long(30天)
+# 实际窗口 = min(策略窗口, NEWS_MAX_AGE_DAYS)
+# NEWS_STRATEGY_PROFILE=short
 # 新闻最大时效（天），搜索时限制结果在近期内，避免使用过时信息
 # NEWS_MAX_AGE_DAYS=3
 # 乖离率阈值（%），偏离 MA5 超过此值提示不追高；强势趋势股自动放宽到 1.5 倍

README.md

@@ -164,6 +164,7 @@
 | `TUSHARE_TOKEN` | [Tushare Pro](https://tushare.pro/weborder/#/login?reg=834638 ) Token | 可选 |
 | `PREFETCH_REALTIME_QUOTES` | 实时行情预取开关：设为 `false` 可禁用全市场预取（默认 `true`） | 可选 |
 | `WECHAT_MSG_TYPE` | 企微消息类型，默认 markdown，支持配置 text 类型，发送纯 markdown 文本 | 可选 |
+| `NEWS_STRATEGY_PROFILE` | 新闻策略窗口档位：`ultra_short`(1天) / `short`(3天) / `medium`(7天) / `long`(30天)，默认 `short` | 可选 |
 | `NEWS_MAX_AGE_DAYS` | 新闻最大时效（天），默认 3，避免使用过时信息 | 可选 |
 | `BIAS_THRESHOLD` | 乖离率阈值（%），默认 5.0，超过提示不追高；强势趋势股自动放宽 | 可选 |
 | `AGENT_MODE` | 开启 Agent 策略问股模式（`true`/`false`，默认 false） | 可选 |

docs/CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 ### 修复
 
 - 🐛 **港股 Agent 实时行情去重与快速路由** — 统一 `HK01810` / `1810.HK` / `01810` 等港股代码归一规则；港股实时行情改为直接走单次 `akshare_hk` 路径，避免按 A 股 source priority 重复触发同一失败接口；Agent 运行期对显式 `retriable=false` 的工具失败增加短路缓存，减少同轮分析中的重复失败调用。
+- 📰 **新闻时效硬过滤与策略分窗**（#697）— 新增 `NEWS_STRATEGY_PROFILE`（`ultra_short/short/medium/long`）并与 `NEWS_MAX_AGE_DAYS` 统一计算有效窗口；搜索结果在返回后执行发布时间硬过滤（时间未知剔除、超窗剔除、未来仅容忍 1 天），并在历史 fallback 链路追加相同约束，避免旧闻再次进入“最新动态/风险警报”。
 
 ## [3.7.0] - 2026-03-15
 

docs/full-guide.md

@@ -228,6 +228,7 @@ daily_stock_analysis/
 | `BRAVE_API_KEYS` | Brave Search API Key（美股优化） | 可选 |
 | `SERPAPI_API_KEYS` | SerpAPI 备用搜索 | 可选 |
 | `SEARXNG_BASE_URLS` | SearXNG 自建实例（无配额兜底，需在 settings.yml 启用 format: json） | 可选 |
+| `NEWS_STRATEGY_PROFILE` | 新闻策略窗口档位：`ultra_short`(1天)/`short`(3天)/`medium`(7天)/`long`(30天)；实际窗口取与 `NEWS_MAX_AGE_DAYS` 的最小值 | 默认 `short` |
 | `NEWS_MAX_AGE_DAYS` | 新闻最大时效（天），搜索时限制结果在近期内 | 默认 `3` |
 | `BIAS_THRESHOLD` | 乖离率阈值（%），超过提示不追高；强势趋势股自动放宽到 1.5 倍 | 默认 `5.0` |
 

main.py

@@ -642,6 +642,7 @@ def main() -> int:
                     minimax_keys=config.minimax_api_keys,
                     searxng_base_urls=config.searxng_base_urls,
                     news_max_age_days=config.news_max_age_days,
+                    news_strategy_profile=getattr(config, "news_strategy_profile", "short"),
                 )
 
             if config.gemini_api_key or config.openai_api_key:

src/analyzer.py

@@ -22,7 +22,14 @@ from json_repair import repair_json
 from litellm import Router
 
 from src.agent.llm_adapter import get_thinking_extra_body
-from src.config import Config, get_config, get_api_keys_for_model, extra_litellm_params, get_configured_llm_models
+from src.config import (
+    Config,
+    extra_litellm_params,
+    get_api_keys_for_model,
+    get_config,
+    get_configured_llm_models,
+    resolve_news_window_days,
+)
 from src.storage import persist_llm_usage
 from src.data.stock_mapping import STOCK_NAME_MAP
 from src.schemas.report_schema import AnalysisReportSchema
@@ -1161,6 +1168,22 @@ class GeminiAnalyzer:
 """
         
         # 添加新闻搜索结果（重点区域）
+        news_window_days: Optional[int] = None
+        context_window = context.get("news_window_days")
+        try:
+            if context_window is not None:
+                parsed_window = int(context_window)
+                if parsed_window > 0:
+                    news_window_days = parsed_window
+        except (TypeError, ValueError):
+            news_window_days = None
+
+        if news_window_days is None:
+            prompt_config = get_config()
+            news_window_days = resolve_news_window_days(
+                news_max_age_days=getattr(prompt_config, "news_max_age_days", 3),
+                news_strategy_profile=getattr(prompt_config, "news_strategy_profile", "short"),
+            )
         prompt += """
 ---
 
@@ -1168,10 +1191,14 @@ class GeminiAnalyzer:
 """
         if news_context:
             prompt += f"""
-以下是 **{stock_name}({code})** 近7日的新闻搜索结果，请重点提取：
+以下是 **{stock_name}({code})** 近{news_window_days}日的新闻搜索结果，请重点提取：
 1. 🚨 **风险警报**：减持、处罚、利空
 2. 🎯 **利好催化**：业绩、合同、政策
 3. 📊 **业绩预期**：年报预告、业绩快报
+4. 🕒 **时间规则（强制）**：
+   - 输出到 `risk_alerts` / `positive_catalysts` / `latest_news` 的每一条都必须带具体日期（YYYY-MM-DD）
+   - 超出近{news_window_days}日窗口的新闻一律忽略
+   - 时间未知、无法确定发布日期的新闻一律忽略
 
 ```
 {news_context}
@@ -1226,6 +1253,7 @@ class GeminiAnalyzer:
 - **持仓分类建议**：空仓者怎么做 vs 持仓者怎么做
 - **具体狙击点位**：买入价、止损价、目标价（精确到分）
 - **检查清单**：每项用 ✅/⚠️/❌ 标记
+- **消息面时间合规**：`latest_news`、`risk_alerts`、`positive_catalysts` 不得包含超出近{news_window_days}日或时间未知的信息
 
 请输出完整的 JSON 格式决策仪表盘。"""
         

src/config.py

@@ -43,6 +43,12 @@ class ConfigIssue:
 _MANAGED_LITELLM_KEY_PROVIDERS = {"gemini", "vertex_ai", "anthropic", "openai", "deepseek"}
 SUPPORTED_LLM_CHANNEL_PROTOCOLS = ("openai", "anthropic", "gemini", "vertex_ai", "deepseek", "ollama")
 _FALSEY_ENV_VALUES = {"0", "false", "no", "off"}
+NEWS_STRATEGY_WINDOWS: Dict[str, int] = {
+    "ultra_short": 1,
+    "short": 3,
+    "medium": 7,
+    "long": 30,
+}
 
 
 def parse_env_bool(value: Optional[str], default: bool = False) -> bool:
@@ -55,6 +61,19 @@ def parse_env_bool(value: Optional[str], default: bool = False) -> bool:
     return normalized not in _FALSEY_ENV_VALUES
 
 
+def normalize_news_strategy_profile(value: Optional[str]) -> str:
+    """Normalize news strategy profile to known values."""
+    candidate = (value or "short").strip().lower()
+    return candidate if candidate in NEWS_STRATEGY_WINDOWS else "short"
+
+
+def resolve_news_window_days(news_max_age_days: int, news_strategy_profile: Optional[str]) -> int:
+    """Resolve effective news window days from profile and global max-age."""
+    profile = normalize_news_strategy_profile(news_strategy_profile)
+    profile_days = NEWS_STRATEGY_WINDOWS.get(profile, NEWS_STRATEGY_WINDOWS["short"])
+    return max(1, min(max(1, int(news_max_age_days)), profile_days))
+
+
 def canonicalize_llm_channel_protocol(value: Optional[str]) -> str:
     """Normalize a protocol label into a LiteLLM provider identifier."""
     candidate = (value or "").strip().lower().replace("-", "_")
@@ -335,6 +354,7 @@ class Config:
 
     # === 新闻与分析筛选配置 ===
     news_max_age_days: int = 3   # 新闻最大时效（天）
+    news_strategy_profile: str = "short"  # 新闻窗口策略档位：ultra_short/short/medium/long
     bias_threshold: float = 5.0  # 乖离率阈值（%），超过此值提示不追高
 
     # === Agent 模式配置 ===
@@ -896,6 +916,9 @@ class Config:
             social_sentiment_api_key=os.getenv('SOCIAL_SENTIMENT_API_KEY') or None,
             social_sentiment_api_url=os.getenv('SOCIAL_SENTIMENT_API_URL', 'https://api.adanos.org').rstrip('/'),
             news_max_age_days=max(1, int(os.getenv('NEWS_MAX_AGE_DAYS', '3'))),
+            news_strategy_profile=cls._parse_news_strategy_profile(
+                os.getenv('NEWS_STRATEGY_PROFILE', 'short')
+            ),
             bias_threshold=max(1.0, float(os.getenv('BIAS_THRESHOLD', '5.0'))),
             agent_mode=os.getenv('AGENT_MODE', 'false').lower() == 'true',
             _agent_mode_explicit=os.getenv('AGENT_MODE') is not None,
@@ -1310,6 +1333,26 @@ class Config:
         )
         return 'simple'
 
+    @classmethod
+    def _parse_news_strategy_profile(cls, value: Optional[str]) -> str:
+        """Parse NEWS_STRATEGY_PROFILE, fallback to short for invalid values."""
+        normalized = normalize_news_strategy_profile(value)
+        raw = (value or "short").strip().lower()
+        if raw != normalized:
+            logging.getLogger(__name__).warning(
+                "NEWS_STRATEGY_PROFILE '%s' invalid, fallback to 'short' "
+                "(valid: ultra_short/short/medium/long)",
+                value,
+            )
+        return normalized
+
+    def get_effective_news_window_days(self) -> int:
+        """Return effective news window days after profile + max-age merge."""
+        return resolve_news_window_days(
+            news_max_age_days=self.news_max_age_days,
+            news_strategy_profile=self.news_strategy_profile,
+        )
+
     @classmethod
     def _parse_market_review_region(cls, value: str) -> str:
         """解析大盘复盘市场区域，非法值记录警告后回退为 cn"""

src/core/config_registry.py

@@ -372,6 +372,20 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
         "validation": {"min": 1, "max": 30},
         "display_order": 60,
     },
+    "NEWS_STRATEGY_PROFILE": {
+        "title": "News Strategy Profile",
+        "description": "News window profile: ultra_short(1d), short(3d), medium(7d), long(30d). Effective window = min(profile, NEWS_MAX_AGE_DAYS).",
+        "category": "data_source",
+        "data_type": "string",
+        "ui_control": "select",
+        "is_sensitive": False,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": "short",
+        "options": ["ultra_short", "short", "medium", "long"],
+        "validation": {"enum": ["ultra_short", "short", "medium", "long"]},
+        "display_order": 61,
+    },
     "BIAS_THRESHOLD": {
         "title": "Bias Threshold (%)",
         "description": "Deviation threshold from MA5 (%). Exceeding this triggers 'do not chase' warning. Strong trend stocks auto-widen to 1.5x.",
@@ -384,7 +398,7 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
         "default_value": "5.0",
         "options": [],
         "validation": {"min": 0.0, "max": 50.0},
-        "display_order": 61,
+        "display_order": 62,
     },
     "PYTDX_HOST": {
         "title": "Pytdx Host",

src/core/pipeline.py

@@ -91,6 +91,7 @@ class StockAnalysisPipeline:
             serpapi_keys=self.config.serpapi_keys,
             minimax_keys=self.config.minimax_api_keys,
             news_max_age_days=self.config.news_max_age_days,
+            news_strategy_profile=getattr(self.config, "news_strategy_profile", "short"),
         )
         
         logger.info(f"调度器初始化完成，最大并发数: {self.max_workers}")
@@ -450,6 +451,9 @@ class StockAnalysisPipeline:
             enhanced['stock_name'] = stock_name
         elif realtime_quote and getattr(realtime_quote, 'name', None):
             enhanced['stock_name'] = realtime_quote.name
+
+        # 将运行时搜索窗口透传给 analyzer，避免与全局配置重新读取产生窗口不一致
+        enhanced['news_window_days'] = getattr(self.search_service, "news_window_days", 3)
         
         # 添加实时行情（兼容不同数据源的字段差异）
         if realtime_quote:

src/search_service.py

@@ -13,10 +13,12 @@ A股自选股智能分析系统 - 搜索服务模块
 
 import logging
 import random
+import re
 import time
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from datetime import datetime
+from dataclasses import dataclass
+from datetime import date, datetime, timedelta, timezone
+from email.utils import parsedate_to_datetime
 from typing import List, Dict, Any, Optional, Tuple
 from itertools import cycle
 import requests
@@ -30,6 +32,11 @@ from tenacity import (
 )
 
 from data_provider.us_index_mapping import is_us_index_code
+from src.config import (
+    NEWS_STRATEGY_WINDOWS,
+    normalize_news_strategy_profile,
+    resolve_news_window_days,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -1338,6 +1345,9 @@ class SearchService:
         "{name} technical analysis",
         "{name} {code} performance volume",
     ]
+    NEWS_OVERSAMPLE_FACTOR = 2
+    NEWS_OVERSAMPLE_MAX = 10
+    FUTURE_TOLERANCE_DAYS = 1
     
     def __init__(
         self,
@@ -1348,6 +1358,7 @@ class SearchService:
         minimax_keys: Optional[List[str]] = None,
         searxng_base_urls: Optional[List[str]] = None,
         news_max_age_days: int = 3,
+        news_strategy_profile: str = "short",
     ):
         """
         初始化搜索服务
@@ -1360,9 +1371,25 @@ class SearchService:
             minimax_keys: MiniMax API Key 列表
             searxng_base_urls: SearXNG 实例地址列表（自建无配额兜底）
             news_max_age_days: 新闻最大时效（天）
+            news_strategy_profile: 新闻窗口策略档位（ultra_short/short/medium/long）
         """
         self._providers: List[BaseSearchProvider] = []
         self.news_max_age_days = max(1, news_max_age_days)
+        raw_profile = (news_strategy_profile or "short").strip().lower()
+        self.news_strategy_profile = normalize_news_strategy_profile(news_strategy_profile)
+        if raw_profile != self.news_strategy_profile:
+            logger.warning(
+                "NEWS_STRATEGY_PROFILE '%s' 无效，已回退为 'short'",
+                news_strategy_profile,
+            )
+        self.news_window_days = resolve_news_window_days(
+            news_max_age_days=self.news_max_age_days,
+            news_strategy_profile=self.news_strategy_profile,
+        )
+        self.news_profile_days = NEWS_STRATEGY_WINDOWS.get(
+            self.news_strategy_profile,
+            NEWS_STRATEGY_WINDOWS["short"],
+        )
 
         # 初始化搜索引擎（按优先级排序）
         # 1. Bocha 优先（中文搜索优化，AI摘要）
@@ -1402,6 +1429,13 @@ class SearchService:
         self._cache: Dict[str, Tuple[float, 'SearchResponse']] = {}
         # Default cache TTL in seconds (10 minutes)
         self._cache_ttl: int = 600
+        logger.info(
+            "新闻时效策略已启用: profile=%s, profile_days=%s, NEWS_MAX_AGE_DAYS=%s, effective_window=%s",
+            self.news_strategy_profile,
+            self.news_profile_days,
+            self.news_max_age_days,
+            self.news_window_days,
+        )
     
     @staticmethod
     def _is_foreign_stock(stock_code: str) -> bool:
@@ -1482,6 +1516,228 @@ class SearchService:
                 for k in oldest:
                     del self._cache[k]
         self._cache[key] = (time.time(), response)
+
+    def _effective_news_window_days(self) -> int:
+        """Resolve effective news window from strategy profile and global max-age."""
+        return resolve_news_window_days(
+            news_max_age_days=self.news_max_age_days,
+            news_strategy_profile=self.news_strategy_profile,
+        )
+
+    @classmethod
+    def _provider_request_size(cls, max_results: int) -> int:
+        """Apply light overfetch before time filtering to avoid sparse outputs."""
+        target = max(1, int(max_results))
+        return max(target, min(target * cls.NEWS_OVERSAMPLE_FACTOR, cls.NEWS_OVERSAMPLE_MAX))
+
+    @staticmethod
+    def _parse_relative_news_date(text: str, now: datetime) -> Optional[date]:
+        """Parse common Chinese/English relative-time strings."""
+        raw = (text or "").strip()
+        if not raw:
+            return None
+
+        lower = raw.lower()
+        if raw in {"今天", "今日", "刚刚"} or lower in {"today", "just now", "now"}:
+            return now.date()
+        if raw == "昨天" or lower == "yesterday":
+            return (now - timedelta(days=1)).date()
+        if raw == "前天":
+            return (now - timedelta(days=2)).date()
+
+        zh = re.match(r"^\s*(\d+)\s*(分钟|小时|天|周|个月|月|年)\s*前\s*$", raw)
+        if zh:
+            amount = int(zh.group(1))
+            unit = zh.group(2)
+            if unit == "分钟":
+                return (now - timedelta(minutes=amount)).date()
+            if unit == "小时":
+                return (now - timedelta(hours=amount)).date()
+            if unit == "天":
+                return (now - timedelta(days=amount)).date()
+            if unit == "周":
+                return (now - timedelta(weeks=amount)).date()
+            if unit in {"个月", "月"}:
+                return (now - timedelta(days=amount * 30)).date()
+            if unit == "年":
+                return (now - timedelta(days=amount * 365)).date()
+
+        en = re.match(
+            r"^\s*(\d+)\s*(minute|minutes|min|mins|hour|hours|day|days|week|weeks|month|months|year|years)\s*ago\s*$",
+            lower,
+        )
+        if en:
+            amount = int(en.group(1))
+            unit = en.group(2)
+            if unit in {"minute", "minutes", "min", "mins"}:
+                return (now - timedelta(minutes=amount)).date()
+            if unit in {"hour", "hours"}:
+                return (now - timedelta(hours=amount)).date()
+            if unit in {"day", "days"}:
+                return (now - timedelta(days=amount)).date()
+            if unit in {"week", "weeks"}:
+                return (now - timedelta(weeks=amount)).date()
+            if unit in {"month", "months"}:
+                return (now - timedelta(days=amount * 30)).date()
+            if unit in {"year", "years"}:
+                return (now - timedelta(days=amount * 365)).date()
+
+        return None
+
+    @classmethod
+    def _normalize_news_publish_date(cls, value: Any) -> Optional[date]:
+        """Normalize provider date value into a date object."""
+        if value is None:
+            return None
+        if isinstance(value, datetime):
+            if value.tzinfo is not None:
+                local_tz = datetime.now().astimezone().tzinfo or timezone.utc
+                return value.astimezone(local_tz).date()
+            return value.date()
+        if isinstance(value, date):
+            return value
+
+        text = str(value).strip()
+        if not text:
+            return None
+        now = datetime.now()
+        local_tz = now.astimezone().tzinfo or timezone.utc
+
+        relative_date = cls._parse_relative_news_date(text, now)
+        if relative_date:
+            return relative_date
+
+        # Unix timestamp fallback
+        if text.isdigit() and len(text) in (10, 13):
+            try:
+                ts = int(text[:10]) if len(text) == 13 else int(text)
+                # Provider timestamps are typically UTC epoch seconds.
+                # Normalize to local date to keep window checks aligned with local "today".
+                return datetime.fromtimestamp(ts, tz=timezone.utc).astimezone(local_tz).date()
+            except (OSError, OverflowError, ValueError):
+                pass
+
+        iso_candidate = text.replace("Z", "+00:00")
+        try:
+            parsed_iso = datetime.fromisoformat(iso_candidate)
+            if parsed_iso.tzinfo is not None:
+                return parsed_iso.astimezone(local_tz).date()
+            return parsed_iso.date()
+        except ValueError:
+            pass
+
+        normalized = re.sub(r"(\d+)(st|nd|rd|th)", r"\1", text, flags=re.IGNORECASE)
+
+        try:
+            parsed_rfc = parsedate_to_datetime(normalized)
+            if parsed_rfc:
+                if parsed_rfc.tzinfo is not None:
+                    return parsed_rfc.astimezone(local_tz).date()
+                return parsed_rfc.date()
+        except (TypeError, ValueError):
+            pass
+
+        zh_match = re.search(r"(\d{4})\s*[年/\-.]\s*(\d{1,2})\s*[月/\-.]\s*(\d{1,2})\s*日?", text)
+        if zh_match:
+            try:
+                return date(int(zh_match.group(1)), int(zh_match.group(2)), int(zh_match.group(3)))
+            except ValueError:
+                pass
+
+        for fmt in (
+            "%Y-%m-%d %H:%M:%S",
+            "%Y-%m-%d %H:%M",
+            "%Y-%m-%d",
+            "%Y/%m/%d %H:%M:%S",
+            "%Y/%m/%d %H:%M",
+            "%Y/%m/%d",
+            "%Y.%m.%d %H:%M:%S",
+            "%Y.%m.%d %H:%M",
+            "%Y.%m.%d",
+            "%Y%m%d",
+            "%b %d, %Y",
+            "%B %d, %Y",
+            "%d %b %Y",
+            "%d %B %Y",
+            "%a, %d %b %Y %H:%M:%S %z",
+        ):
+            try:
+                parsed_dt = datetime.strptime(normalized, fmt)
+                if parsed_dt.tzinfo is not None:
+                    return parsed_dt.astimezone(local_tz).date()
+                return parsed_dt.date()
+            except ValueError:
+                continue
+
+        return None
+
+    def _filter_news_response(
+        self,
+        response: SearchResponse,
+        *,
+        search_days: int,
+        max_results: int,
+        log_scope: str,
+    ) -> SearchResponse:
+        """Hard-filter results by published_date recency and normalize date strings."""
+        if not response.success or not response.results:
+            return response
+
+        today = datetime.now().date()
+        earliest = today - timedelta(days=max(0, int(search_days) - 1))
+        latest = today + timedelta(days=self.FUTURE_TOLERANCE_DAYS)
+
+        filtered: List[SearchResult] = []
+        dropped_unknown = 0
+        dropped_old = 0
+        dropped_future = 0
+
+        for item in response.results:
+            published = self._normalize_news_publish_date(item.published_date)
+            if published is None:
+                dropped_unknown += 1
+                continue
+            if published < earliest:
+                dropped_old += 1
+                continue
+            if published > latest:
+                dropped_future += 1
+                continue
+
+            filtered.append(
+                SearchResult(
+                    title=item.title,
+                    snippet=item.snippet,
+                    url=item.url,
+                    source=item.source,
+                    published_date=published.isoformat(),
+                )
+            )
+            if len(filtered) >= max_results:
+                break
+
+        if dropped_unknown or dropped_old or dropped_future:
+            logger.info(
+                "[新闻过滤] %s: provider=%s, total=%s, kept=%s, drop_unknown=%s, drop_old=%s, drop_future=%s, window=[%s,%s]",
+                log_scope,
+                response.provider,
+                len(response.results),
+                len(filtered),
+                dropped_unknown,
+                dropped_old,
+                dropped_future,
+                earliest.isoformat(),
+                latest.isoformat(),
+            )
+
+        return SearchResponse(
+            query=response.query,
+            results=filtered,
+            provider=response.provider,
+            success=response.success,
+            error_message=response.error_message,
+            search_time=response.search_time,
+        )
     
     def search_stock_news(
         self,
@@ -1502,20 +1758,10 @@ class SearchService:
         Returns:
             SearchResponse 对象
         """
-        # 智能确定搜索时间范围
-        # 策略：
-        # 1. 周二至周五：搜索近1天（24小时）
-        # 2. 周六、周日：搜索近2-3天（覆盖周末）
-        # 3. 周一：搜索近3天（覆盖周末）
-        # 4. 用 NEWS_MAX_AGE_DAYS 限制上限
-        today_weekday = datetime.now().weekday()
-        if today_weekday == 0:  # 周一
-            weekday_days = 3
-        elif today_weekday >= 5:  # 周六(5)、周日(6)
-            weekday_days = 2
-        else:  # 周二(1) - 周五(4)
-            weekday_days = 1
-        search_days = min(weekday_days, self.news_max_age_days)
+        # 策略窗口优先：ultra_short/short/medium/long = 1/3/7/30 天，
+        # 并统一受 NEWS_MAX_AGE_DAYS 上限约束。
+        search_days = self._effective_news_window_days()
+        provider_max_results = self._provider_request_size(max_results)
 
         # 构建搜索查询（优化搜索效果）
         is_foreign = self._is_foreign_stock(stock_code)
@@ -1529,7 +1775,15 @@ class SearchService:
             # 默认主查询：股票名称 + 核心关键词
             query = f"{stock_name} {stock_code} 股票 最新消息"
 
-        logger.info(f"搜索股票新闻: {stock_name}({stock_code}), query='{query}', 时间范围: 近{search_days}天")
+        logger.info(
+            "搜索股票新闻: %s(%s), query='%s', 时间范围: 近%s天, 目标条数=%s, provider请求条数=%s",
+            stock_name,
+            stock_code,
+            query,
+            search_days,
+            max_results,
+            provider_max_results,
+        )
 
         # Check cache first
         cache_key = self._cache_key(query, max_results, search_days)
@@ -1538,19 +1792,46 @@ class SearchService:
             logger.info(f"使用缓存搜索结果: {stock_name}({stock_code})")
             return cached
 
-        # 依次尝试各个搜索引擎
+        # 依次尝试各个搜索引擎（若过滤后为空，继续尝试下一引擎）
+        had_provider_success = False
         for provider in self._providers:
             if not provider.is_available:
                 continue
             
-            response = provider.search(query, max_results, days=search_days)
-            
-            if response.success and response.results:
+            response = provider.search(query, provider_max_results, days=search_days)
+            filtered_response = self._filter_news_response(
+                response,
+                search_days=search_days,
+                max_results=max_results,
+                log_scope=f"{stock_code}:{provider.name}:stock_news",
+            )
+            had_provider_success = had_provider_success or bool(response.success)
+
+            if filtered_response.success and filtered_response.results:
                 logger.info(f"使用 {provider.name} 搜索成功")
-                self._put_cache(cache_key, response)
-                return response
+                self._put_cache(cache_key, filtered_response)
+                return filtered_response
             else:
-                logger.warning(f"{provider.name} 搜索失败: {response.error_message}，尝试下一个引擎")
+                if response.success and not filtered_response.results:
+                    logger.info(
+                        "%s 搜索成功但过滤后无有效新闻，继续尝试下一引擎",
+                        provider.name,
+                    )
+                else:
+                    logger.warning(
+                        "%s 搜索失败: %s，尝试下一个引擎",
+                        provider.name,
+                        response.error_message,
+                    )
+
+        if had_provider_success:
+            return SearchResponse(
+                query=query,
+                results=[],
+                provider="Filtered",
+                success=True,
+                error_message=None,
+            )
         
         # 所有引擎都失败
         return SearchResponse(
@@ -1673,7 +1954,18 @@ class SearchService:
                 ), 'desc': '行业分析'},
             ]
         
-        logger.info(f"开始多维度情报搜索: {stock_name}({stock_code})")
+        search_days = self._effective_news_window_days()
+        target_per_dimension = 3
+        provider_max_results = self._provider_request_size(target_per_dimension)
+
+        logger.info(
+            "开始多维度情报搜索: %s(%s), 时间范围: 近%s天, 目标条数=%s, provider请求条数=%s",
+            stock_name,
+            stock_code,
+            search_days,
+            target_per_dimension,
+            provider_max_results,
+        )
         
         # 轮流使用不同的搜索引擎
         provider_index = 0
@@ -1692,12 +1984,27 @@ class SearchService:
             
             logger.info(f"[情报搜索] {dim['desc']}: 使用 {provider.name}")
             
-            response = provider.search(dim['query'], max_results=3, days=self.news_max_age_days)
-            results[dim['name']] = response
+            response = provider.search(
+                dim['query'],
+                max_results=provider_max_results,
+                days=search_days,
+            )
+            filtered_response = self._filter_news_response(
+                response,
+                search_days=search_days,
+                max_results=target_per_dimension,
+                log_scope=f"{stock_code}:{provider.name}:{dim['name']}",
+            )
+            results[dim['name']] = filtered_response
             search_count += 1
             
             if response.success:
-                logger.info(f"[情报搜索] {dim['desc']}: 获取 {len(response.results)} 条结果")
+                logger.info(
+                    "[情报搜索] %s: 原始=%s条, 过滤后=%s条",
+                    dim['desc'],
+                    len(response.results),
+                    len(filtered_response.results),
+                )
             else:
                 logger.warning(f"[情报搜索] {dim['desc']}: 搜索失败 - {response.error_message}")
             
@@ -1978,6 +2285,7 @@ def get_search_service() -> SearchService:
             minimax_keys=config.minimax_api_keys,
             searxng_base_urls=config.searxng_base_urls,
             news_max_age_days=config.news_max_age_days,
+            news_strategy_profile=getattr(config, "news_strategy_profile", "short"),
         )
     
     return _search_service

src/services/history_service.py

@@ -12,9 +12,10 @@ Responsibilities:
 from __future__ import annotations
 import json
 import logging
-from datetime import datetime, timedelta
+from datetime import date, datetime, timedelta
 from typing import Optional, Dict, Any, List, Tuple, TYPE_CHECKING
 
+from src.config import get_config, resolve_news_window_days
 from src.storage import DatabaseManager
 from src.utils.data_processing import normalize_model_used, parse_json_field
 
@@ -381,7 +382,31 @@ class HistoryService:
             if item.fetched_at and start_time <= item.fetched_at <= end_time
         ]
 
-        return matched[:limit]
+        # 历史兜底链路也做发布时间硬过滤，避免旧库脏数据重新冒出。
+        cfg = get_config()
+        window_days = resolve_news_window_days(
+            news_max_age_days=getattr(cfg, "news_max_age_days", 3),
+            news_strategy_profile=getattr(cfg, "news_strategy_profile", "short"),
+        )
+        # Anchor to analysis date instead of "today" to preserve historical context.
+        anchor_date = analysis.created_at.date()
+        latest_allowed = anchor_date + timedelta(days=1)
+        earliest_allowed = anchor_date - timedelta(days=max(0, window_days - 1))
+
+        filtered = []
+        for item in matched:
+            if not item.published_date:
+                continue
+            if isinstance(item.published_date, datetime):
+                published = item.published_date.date()
+            elif isinstance(item.published_date, date):
+                published = item.published_date
+            else:
+                continue
+            if earliest_allowed <= published <= latest_allowed:
+                filtered.append(item)
+
+        return filtered[:limit]
     
     def _get_sentiment_label(self, score: int) -> str:
         """

src/services/name_to_code_resolver.py

@@ -86,6 +86,19 @@ def _get_akshare_name_to_code() -> Optional[Dict[str, str]]:
         return None
 
 
+def _is_single_char_typo(input_name: str, candidate_name: str) -> bool:
+    """Return True when two names only differ by one character position."""
+    if not input_name or not candidate_name:
+        return False
+    if len(input_name) != len(candidate_name):
+        return False
+    # Keep typo fallback conservative: only for names with enough signal.
+    if len(input_name) < 3:
+        return False
+    diff = sum(1 for a, b in zip(input_name, candidate_name) if a != b)
+    return diff == 1
+
+
 def resolve_name_to_code(name: str) -> Optional[str]:
     """
     Resolve stock name to code.
@@ -147,10 +160,19 @@ def resolve_name_to_code(name: str) -> Optional[str]:
     # e.g. '中国' matching arbitrary company names in a pool of 5000+ stocks.
     # Use a higher cutoff (0.8) to reduce mis-hits on longer inputs as well.
     if len(s) > 2:
-        matches = difflib.get_close_matches(s, list(all_name_to_code.keys()), n=1, cutoff=0.8)
+        names = list(all_name_to_code.keys())
+        matches = difflib.get_close_matches(s, names, n=1, cutoff=0.8)
         if matches:
             logger.debug(f"[NameResolver] 命中模糊匹配: input={s}, matched={matches[0]}")
             return all_name_to_code[matches[0]]
 
+        # Conservative fallback for one-character typo in medium/long names.
+        # This keeps the strict default threshold while fixing obvious misspellings
+        # such as "贵州茅苔" -> "贵州茅台".
+        typo_matches = difflib.get_close_matches(s, names, n=1, cutoff=0.7)
+        if typo_matches and _is_single_char_typo(s, typo_matches[0]):
+            logger.debug(f"[NameResolver] 命中单字误写兜底: input={s}, matched={typo_matches[0]}")
+            return all_name_to_code[typo_matches[0]]
+
     logger.debug(f"[NameResolver] 解析失败: {s}")
     return None

tests/test_analyzer_news_prompt.py

@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+"""Tests for analyzer news prompt hard constraints (Issue #697)."""
+
+import sys
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+try:
+    import litellm  # noqa: F401
+except ModuleNotFoundError:
+    sys.modules["litellm"] = MagicMock()
+
+from src.analyzer import GeminiAnalyzer
+
+
+class AnalyzerNewsPromptTestCase(unittest.TestCase):
+    def test_prompt_contains_time_constraints(self) -> None:
+        with patch.object(GeminiAnalyzer, "_init_litellm", return_value=None):
+            analyzer = GeminiAnalyzer()
+
+        context = {
+            "code": "600519",
+            "stock_name": "贵州茅台",
+            "date": "2026-03-16",
+            "today": {},
+        }
+        fake_cfg = SimpleNamespace(
+            news_max_age_days=30,
+            news_strategy_profile="medium",  # 7 days
+        )
+        with patch("src.analyzer.get_config", return_value=fake_cfg):
+            prompt = analyzer._format_prompt(context, "贵州茅台", news_context="news")
+
+        self.assertIn("近7日的新闻搜索结果", prompt)
+        self.assertIn("每一条都必须带具体日期（YYYY-MM-DD）", prompt)
+        self.assertIn("超出近7日窗口的新闻一律忽略", prompt)
+        self.assertIn("时间未知、无法确定发布日期的新闻一律忽略", prompt)
+
+    def test_prompt_prefers_context_news_window_days(self) -> None:
+        with patch.object(GeminiAnalyzer, "_init_litellm", return_value=None):
+            analyzer = GeminiAnalyzer()
+
+        context = {
+            "code": "600519",
+            "stock_name": "贵州茅台",
+            "date": "2026-03-16",
+            "today": {},
+            "news_window_days": 1,
+        }
+        fake_cfg = SimpleNamespace(
+            news_max_age_days=30,
+            news_strategy_profile="long",  # 30 days if fallback is used
+        )
+        with patch("src.analyzer.get_config", return_value=fake_cfg):
+            prompt = analyzer._format_prompt(context, "贵州茅台", news_context="news")
+
+        self.assertIn("近1日的新闻搜索结果", prompt)
+        self.assertIn("超出近1日窗口的新闻一律忽略", prompt)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_history_news_fallback.py

@@ -0,0 +1,77 @@
+# -*- coding: utf-8 -*-
+"""Tests for history fallback published_date hard filtering (Issue #697)."""
+
+import unittest
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from src.services.history_service import HistoryService
+
+
+class HistoryNewsFallbackTestCase(unittest.TestCase):
+    def test_fallback_filters_by_published_date_window(self) -> None:
+        now = datetime.now()
+        analysis = SimpleNamespace(code="600519", created_at=now)
+
+        # All entries are within fetched_at window; only one should pass published_date window.
+        candidates = [
+            SimpleNamespace(
+                fetched_at=now,
+                published_date=now - timedelta(days=20),  # too old
+                title="old",
+            ),
+            SimpleNamespace(
+                fetched_at=now,
+                published_date=None,  # unknown -> drop
+                title="unknown",
+            ),
+            SimpleNamespace(
+                fetched_at=now,
+                published_date=now - timedelta(days=1),  # valid
+                title="fresh",
+            ),
+        ]
+
+        mock_db = MagicMock()
+        mock_db.get_analysis_history.return_value = [analysis]
+        mock_db.get_recent_news.return_value = candidates
+
+        svc = HistoryService(db_manager=mock_db)
+        fake_cfg = SimpleNamespace(news_max_age_days=30, news_strategy_profile="short")
+        with patch("src.services.history_service.get_config", return_value=fake_cfg):
+            result = svc._fallback_news_by_analysis_context("q-1", limit=20)
+
+        self.assertEqual([item.title for item in result], ["fresh"])
+
+    def test_fallback_uses_analysis_date_as_window_anchor(self) -> None:
+        analysis_time = datetime.now() - timedelta(days=40)
+        analysis = SimpleNamespace(code="600519", created_at=analysis_time)
+
+        candidates = [
+            SimpleNamespace(
+                fetched_at=analysis_time,
+                published_date=analysis_time - timedelta(days=10),  # too old for short profile
+                title="too_old_for_analysis_window",
+            ),
+            SimpleNamespace(
+                fetched_at=analysis_time,
+                published_date=analysis_time - timedelta(days=1),  # valid around analysis date
+                title="valid_near_analysis_date",
+            ),
+        ]
+
+        mock_db = MagicMock()
+        mock_db.get_analysis_history.return_value = [analysis]
+        mock_db.get_recent_news.return_value = candidates
+
+        svc = HistoryService(db_manager=mock_db)
+        fake_cfg = SimpleNamespace(news_max_age_days=30, news_strategy_profile="short")
+        with patch("src.services.history_service.get_config", return_value=fake_cfg):
+            result = svc._fallback_news_by_analysis_context("q-1", limit=20)
+
+        self.assertEqual([item.title for item in result], ["valid_near_analysis_date"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_news_strategy_config.py

@@ -0,0 +1,23 @@
+# -*- coding: utf-8 -*-
+"""Tests for NEWS_STRATEGY_PROFILE parsing and effective window calculation."""
+
+import unittest
+
+from src.config import Config, resolve_news_window_days
+
+
+class NewsStrategyConfigTestCase(unittest.TestCase):
+    def test_invalid_profile_fallback_to_short(self) -> None:
+        self.assertEqual(Config._parse_news_strategy_profile("bad_value"), "short")
+
+    def test_window_respects_news_max_age_days(self) -> None:
+        # medium=7 but max-age=3 -> effective=3
+        self.assertEqual(resolve_news_window_days(3, "medium"), 3)
+        # long=30 with max-age=30 -> effective=30
+        self.assertEqual(resolve_news_window_days(30, "long"), 30)
+        # ultra_short=1 with max-age=30 -> effective=1
+        self.assertEqual(resolve_news_window_days(30, "ultra_short"), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pipeline_realtime_indicators.py

@@ -197,6 +197,16 @@ class TestEnhanceContextRealtimeOverride(unittest.TestCase):
         self.assertIn("price_change_ratio", enhanced)
         self.assertIn("volume_change_ratio", enhanced)
 
+    def test_enhance_context_injects_runtime_news_window_days(self) -> None:
+        context = {"code": "600519", "today": {"close": 15.0}}
+        enhanced = self.pipeline._enhance_context(
+            context, None, None, None, "贵州茅台"
+        )
+        self.assertEqual(
+            enhanced["news_window_days"],
+            self.pipeline.search_service.news_window_days,
+        )
+
     def test_today_not_overridden_when_trend_missing(self) -> None:
         context = {"code": "600519", "today": {"close": 15.0}}
         quote = _make_realtime_quote(price=15.72)

tests/test_search_news_freshness.py

@@ -1,10 +1,12 @@
 # -*- coding: utf-8 -*-
 """
-Unit tests for search_stock_news and search_comprehensive_intel news_max_age_days logic (Issue #296).
+Unit tests for strict news freshness filtering and strategy window logic (Issue #697).
 """
 
 import sys
 import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 # Mock newspaper before search_service import (optional dependency)
@@ -17,107 +19,199 @@ if "newspaper" not in sys.modules:
 from src.search_service import SearchResponse, SearchResult, SearchService
 
 
-def _fake_search_response() -> SearchResponse:
-    """Return a successful SearchResponse for mocking."""
+def _result(title: str, published_date: str | None) -> SearchResult:
+    return SearchResult(
+        title=title,
+        snippet="snippet",
+        url=f"https://example.com/{title}",
+        source="example.com",
+        published_date=published_date,
+    )
+
+
+def _response(results) -> SearchResponse:
     return SearchResponse(
         query="test",
-        results=[
-            SearchResult(
-                title="Test",
-                snippet="snippet",
-                url="https://example.com/1",
-                source="example.com",
-                published_date=None,
-            )
-        ],
+        results=results,
         provider="Mock",
         success=True,
     )
 
 
 class SearchNewsFreshnessTestCase(unittest.TestCase):
-    """Tests for news_max_age_days in search_stock_news and search_comprehensive_intel."""
+    """Tests for strategy window and strict published_date filtering."""
 
-    def _create_service_with_mock_provider(self, news_max_age_days: int = 3):
-        """Create SearchService with a mock provider that records search() calls."""
+    def _create_service_with_mock_provider(
+        self,
+        *,
+        news_max_age_days: int = 3,
+        news_strategy_profile: str = "short",
+        response: SearchResponse | None = None,
+    ):
         service = SearchService(
             bocha_keys=["dummy_key"],
             news_max_age_days=news_max_age_days,
+            news_strategy_profile=news_strategy_profile,
+        )
+        mock_search = MagicMock(
+            return_value=response
+            or _response([_result("default", datetime.now().date().isoformat())])
         )
-        mock_search = MagicMock(return_value=_fake_search_response())
         service._providers[0].search = mock_search
         return service, mock_search
 
-    @patch("src.search_service.datetime")
-    def test_search_stock_news_days_monday_limit_by_news_max_age(
-        self, mock_dt: MagicMock
-    ) -> None:
-        """Monday + news_max_age_days=1 -> search_days=1 (min(3,1)=1)."""
-        mock_dt.now.return_value.weekday.return_value = 0  # Monday -> weekday_days=3
+    def test_effective_window_uses_profile_and_news_max_age(self) -> None:
+        """window = min(profile_days, NEWS_MAX_AGE_DAYS)."""
         service, mock_search = self._create_service_with_mock_provider(
-            news_max_age_days=1
+            news_max_age_days=3,
+            news_strategy_profile="medium",  # 7
         )
-        service.search_stock_news("600519", "贵州茅台")
-        mock_search.assert_called_once()
-        call_kwargs = mock_search.call_args[1]
-        self.assertEqual(call_kwargs["days"], 1)
+        service.search_stock_news("600519", "贵州茅台", max_results=5)
+        kwargs = mock_search.call_args[1]
+        self.assertEqual(kwargs["days"], 3)
 
-    @patch("src.search_service.datetime")
-    def test_search_stock_news_days_tuesday_weekday_dominates(
-        self, mock_dt: MagicMock
-    ) -> None:
-        """Tuesday + news_max_age_days=3 -> search_days=1 (min(1,3)=1)."""
-        mock_dt.now.return_value.weekday.return_value = 1  # Tuesday -> weekday_days=1
+    def test_invalid_profile_falls_back_to_short(self) -> None:
+        """Invalid profile should fallback to short (3 days)."""
         service, mock_search = self._create_service_with_mock_provider(
-            news_max_age_days=3
+            news_max_age_days=30,
+            news_strategy_profile="invalid_profile",
         )
-        service.search_stock_news("600519", "贵州茅台")
-        mock_search.assert_called_once()
-        call_kwargs = mock_search.call_args[1]
-        self.assertEqual(call_kwargs["days"], 1)
+        service.search_stock_news("600519", "贵州茅台", max_results=5)
+        kwargs = mock_search.call_args[1]
+        self.assertEqual(kwargs["days"], 3)
 
-    @patch("src.search_service.datetime")
-    def test_search_stock_news_days_monday_news_max_age_dominates(
-        self, mock_dt: MagicMock
-    ) -> None:
-        """Monday + news_max_age_days=5 -> search_days=3 (min(3,5)=3)."""
-        mock_dt.now.return_value.weekday.return_value = 0  # Monday -> weekday_days=3
-        service, mock_search = self._create_service_with_mock_provider(
-            news_max_age_days=5
-        )
-        service.search_stock_news("600519", "贵州茅台")
-        mock_search.assert_called_once()
-        call_kwargs = mock_search.call_args[1]
-        self.assertEqual(call_kwargs["days"], 3)
+    def test_search_stock_news_strict_filters(self) -> None:
+        """Drop old/unknown/future+2, keep future+1 and within-window dates."""
+        today = datetime.now().date()
+        fresh = today.isoformat()
+        old = (today - timedelta(days=30)).isoformat()
+        future_1 = (today + timedelta(days=1)).isoformat()
+        future_2 = (today + timedelta(days=2)).isoformat()
 
-    @patch("src.search_service.datetime")
-    def test_search_stock_news_days_weekend(self, mock_dt: MagicMock) -> None:
-        """Saturday + news_max_age_days=5 -> search_days=2 (min(2,5)=2)."""
-        mock_dt.now.return_value.weekday.return_value = 5  # Saturday -> weekday_days=2
-        service, mock_search = self._create_service_with_mock_provider(
-            news_max_age_days=5
+        service, _ = self._create_service_with_mock_provider(
+            news_max_age_days=7,
+            news_strategy_profile="medium",
+            response=_response(
+                [
+                    _result("old", old),
+                    _result("unknown", None),
+                    _result("future_2", future_2),
+                    _result("future_1", future_1),
+                    _result("fresh", fresh),
+                ]
+            ),
         )
-        service.search_stock_news("600519", "贵州茅台")
-        mock_search.assert_called_once()
-        call_kwargs = mock_search.call_args[1]
-        self.assertEqual(call_kwargs["days"], 2)
 
-    def test_search_comprehensive_intel_uses_news_max_age_days(self) -> None:
-        """search_comprehensive_intel passes news_max_age_days directly to provider.search."""
+        resp = service.search_stock_news("600519", "贵州茅台", max_results=5)
+        titles = [r.title for r in resp.results]
+        self.assertEqual(titles, ["future_1", "fresh"])
+        for item in resp.results:
+            self.assertRegex(item.published_date or "", r"^\d{4}-\d{2}-\d{2}$")
+
+    def test_search_stock_news_overfetch_before_filter(self) -> None:
+        """Provider request size should be increased before filtering."""
         service, mock_search = self._create_service_with_mock_provider(
-            news_max_age_days=2
+            news_max_age_days=3,
+            news_strategy_profile="short",
         )
-        with patch("src.search_service.time.sleep"):  # avoid delay in tests
-            service.search_comprehensive_intel(
+        service.search_stock_news("600519", "贵州茅台", max_results=4)
+        args, kwargs = mock_search.call_args
+        requested = kwargs.get("max_results")
+        if requested is None:
+            requested = args[1]
+        self.assertEqual(requested, 8)
+
+    def test_search_stock_news_try_next_provider_when_filtered_empty(self) -> None:
+        """If provider-A passes API call but all results are filtered, continue to provider-B."""
+        today = datetime.now().date()
+        old = (today - timedelta(days=90)).isoformat()
+        fresh = today.isoformat()
+
+        service = SearchService(
+            bocha_keys=["dummy_key"],
+            news_max_age_days=3,
+            news_strategy_profile="short",
+        )
+
+        p1 = SimpleNamespace(
+            is_available=True,
+            name="P1",
+            search=MagicMock(return_value=_response([_result("too_old", old)])),
+        )
+        p2 = SimpleNamespace(
+            is_available=True,
+            name="P2",
+            search=MagicMock(return_value=_response([_result("fresh", fresh)])),
+        )
+        service._providers = [p1, p2]
+
+        resp = service.search_stock_news("600519", "贵州茅台", max_results=3)
+        self.assertEqual([r.title for r in resp.results], ["fresh"])
+        p1.search.assert_called_once()
+        p2.search.assert_called_once()
+
+    def test_search_comprehensive_intel_uses_same_filter(self) -> None:
+        """Comprehensive intel should use same strict date filtering logic."""
+        today = datetime.now().date()
+        old = (today - timedelta(days=20)).isoformat()
+        fresh = (today - timedelta(days=1)).isoformat()
+
+        service, mock_search = self._create_service_with_mock_provider(
+            news_max_age_days=3,
+            news_strategy_profile="medium",  # min(7,3)=3
+            response=_response([_result("old", old), _result("fresh", fresh)]),
+        )
+        with patch("src.search_service.time.sleep"):
+            intel = service.search_comprehensive_intel(
                 stock_code="600519",
                 stock_name="贵州茅台",
                 max_searches=2,
             )
+
         self.assertGreaterEqual(mock_search.call_count, 1)
         for call in mock_search.call_args_list:
-            call_kwargs = call[1]
-            self.assertEqual(
-                call_kwargs["days"],
-                2,
-                msg=f"Expected days=2, got {call_kwargs.get('days')}",
-            )
+            kwargs = call[1]
+            self.assertEqual(kwargs["days"], 3)
+            self.assertEqual(kwargs["max_results"], 6)  # target 3 -> overfetch 6
+
+        for resp in intel.values():
+            self.assertTrue(all(item.title != "old" for item in resp.results))
+
+    def test_effective_window_helper_has_no_side_effect(self) -> None:
+        """_effective_news_window_days should not mutate stored news_window_days."""
+        service, _ = self._create_service_with_mock_provider(
+            news_max_age_days=3,
+            news_strategy_profile="short",
+        )
+        service.news_window_days = 99
+        resolved = service._effective_news_window_days()
+        self.assertEqual(resolved, 3)
+        self.assertEqual(service.news_window_days, 99)
+
+    def test_unix_timestamp_normalizes_to_local_date(self) -> None:
+        """Unix timestamp should be converted to local date before window filtering."""
+        dt_utc = datetime(2026, 3, 15, 23, 30, tzinfo=timezone.utc)
+        timestamp = str(int(dt_utc.timestamp()))
+        expected_local_date = dt_utc.astimezone().date()
+        parsed = SearchService._normalize_news_publish_date(timestamp)
+        self.assertEqual(parsed, expected_local_date)
+
+    def test_iso_utc_string_normalizes_to_local_date(self) -> None:
+        """ISO datetime with timezone should be converted to local date."""
+        dt_utc = datetime(2026, 3, 15, 23, 30, tzinfo=timezone.utc)
+        iso_text = "2026-03-15T23:30:00Z"
+        expected_local_date = dt_utc.astimezone().date()
+        parsed = SearchService._normalize_news_publish_date(iso_text)
+        self.assertEqual(parsed, expected_local_date)
+
+    def test_rfc_utc_string_normalizes_to_local_date(self) -> None:
+        """RFC datetime with timezone should be converted to local date."""
+        dt_utc = datetime(2026, 3, 15, 23, 30, tzinfo=timezone.utc)
+        rfc_text = "Sun, 15 Mar 2026 23:30:00 +0000"
+        expected_local_date = dt_utc.astimezone().date()
+        parsed = SearchService._normalize_news_publish_date(rfc_text)
+        self.assertEqual(parsed, expected_local_date)
+
+
+if __name__ == "__main__":
+    unittest.main()