migration: preserve sentiment and pools slice

backend/application.py

@@ -53,14 +53,13 @@ from screener import (
 from backend.features.accounts.http import AccountHttpMixin
 from backend.features.accounts.security import SecretVault
 from backend.features.accounts.service import AccountService
-from backend.features.system import SystemHttpMixin
+from backend.features.pools import PoolServiceMixin
-from sentiment_engine import (
+from backend.features.sentiment import SentimentServiceMixin
-    COMPONENT_WEIGHTS,
+from backend.features.sentiment.engine import (
-    SENTIMENT_ENGINE_VERSION,
-    apply_sentiment_to_dashboard,
     build_sentiment_history,
     latest_contiguous_history,
 )
+from backend.features.system import SystemHttpMixin
 from backend.data.providers.tushare_client import TushareClient, TushareError, _sector_coverage_issue
 
 
@@ -141,7 +140,7 @@ MENTOR_ETF_UNIVERSE = (
 )
 
 
-class DashboardService(MarketServiceMixin):
+class DashboardService(MarketServiceMixin, SentimentServiceMixin, PoolServiceMixin):
     def __init__(self) -> None:
         runtime = load_runtime_settings()
         self.vault = SecretVault(runtime.encryption_key)
@@ -734,31 +733,6 @@ class DashboardService(MarketServiceMixin):
         return AccountService.public_personal_profile(personal)
 
 
-    def _enrich_dashboard_sentiment(
-        self,
-        dashboard: dict[str, Any],
-        end_date: str,
-    ) -> dict[str, Any]:
-        history = self.database.list_snapshot_payloads(end_date, 260)
-        return apply_sentiment_to_dashboard(dashboard, history)
-
-    def sentiment_history(self, trade_date: str, limit: int = 20) -> dict[str, Any]:
-        normalized_date = normalize_date(trade_date)
-        limit = max(10, min(120, int(limit)))
-        full_series = build_sentiment_history(
-            self.database.list_snapshot_payloads(normalized_date, 240)
-        )
-        series = latest_contiguous_history(full_series)
-        rows = series[-limit:]
-        return {
-            "trade_date": rows[-1]["trade_date"] if rows else normalized_date,
-            "available_days": len(series),
-            "stored_days": len(full_series),
-            "requested_days": limit,
-            "rows": rows,
-            "weights": COMPONENT_WEIGHTS,
-            "normalization": rows[-1]["normalization"] if rows else "固定锚点",
-        }
 
     def rotation_history(self, trade_date: str, limit: int = 9) -> dict[str, Any]:
         normalized_date = normalize_date(trade_date)
@@ -3380,146 +3354,6 @@ class DashboardService(MarketServiceMixin):
         }
 
 
-    def save_reason(self, trade_date: str, code: str, reason: str) -> None:
-        normalized_date = normalize_date(trade_date)
-        code = validate_stock_code(code)
-        reason = reason.strip()
-        if not reason or len(reason) > 200:
-            raise ValueError("涨停原因应为 1 至 200 个字符。")
-        self.database.save_reason_override(normalized_date, code, reason)
-
-
-    def _apply_reason_overrides(self, dashboard: dict[str, Any]) -> dict[str, Any]:
-        trade_date = str(dashboard.get("meta", {}).get("trade_date", "")).replace("-", "")
-        enrichment = self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date)
-        if enrichment:
-            self._merge_ifind_event_enrichment(dashboard, enrichment)
-        else:
-            self._schedule_ifind_event_enrichment(trade_date)
-        overrides = self.database.reason_overrides(trade_date)
-        if not overrides:
-            return dashboard
-        for key in ("limits", "broken", "down_limits"):
-            for row in dashboard.get(key) or []:
-                if row.get("code") in overrides:
-                    row["reason"] = overrides[row["code"]]
-                    row["reason_source"] = "manual"
-        return dashboard
-
-    def _schedule_ifind_event_enrichment(self, trade_date: str) -> None:
-        ifind = getattr(self, "ifind", None)
-        if not ifind or not ifind.configured or not re.fullmatch(r"\d{8}", trade_date):
-            return
-        now = datetime.now().astimezone()
-        if trade_date == now.strftime("%Y%m%d") and now.time().replace(tzinfo=None) < dt_time(15, 0):
-            return
-        self.jobs.submit(
-            "market.ifind-event-enrichment",
-            f"{trade_date}:v1",
-            lambda: self._refresh_ifind_event_enrichment(trade_date),
-            {"trade_date": trade_date, "trigger": "dashboard-enrichment"},
-        )
-
-    def _refresh_ifind_event_enrichment(self, trade_date: str) -> None:
-        if not self._ifind_event_lock.acquire(blocking=False):
-            return
-        try:
-            if self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date):
-                return
-            ifind = getattr(self, "ifind", None)
-            if not ifind or not ifind.configured:
-                return
-            current = datetime.strptime(trade_date, "%Y%m%d")
-            display_date = f"{current.year}年{current.month}月{current.day}日"
-            requests = {
-                "limits": (
-                    f"{display_date}涨停股票，股票代码、股票简称、涨停原因、"
-                    "首次涨停时间、最终涨停时间、开板次数"
-                ),
-                "broken": (
-                    f"{display_date}曾涨停但收盘未涨停的股票，股票代码、股票简称、"
-                    "涨停原因、首次涨停时间、开板次数"
-                ),
-                "down_limits": (
-                    f"{display_date}跌停股票，股票代码、股票简称、跌停原因"
-                ),
-            }
-            result: dict[str, Any] = {
-                "trade_date": trade_date,
-                "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
-                "limits": {}, "broken": {}, "down_limits": {}, "partial": False,
-            }
-            for kind, query in requests.items():
-                try:
-                    rows = ifind.wencai(query, "stock", cache_ttl=900)
-                except IfindError:
-                    result["partial"] = True
-                    continue
-                for raw in rows:
-                    code = self._ifind_row_code(raw)
-                    if not code:
-                        continue
-                    reason_tokens = (
-                        ("跌停原因", "风险线索", "原因")
-                        if kind == "down_limits"
-                        else ("涨停原因类别", "涨停原因", "触板逻辑", "原因")
-                    )
-                    reason = str(self._ifind_field(raw, reason_tokens) or "").strip()
-                    first_time = self._normalize_ifind_event_time(
-                        self._ifind_field(raw, ("首次涨停时间", "首次触板时间", "首次封板时间"))
-                    )
-                    last_time = self._normalize_ifind_event_time(
-                        self._ifind_field(raw, ("最终涨停时间", "最后涨停时间", "最后封板时间"))
-                    )
-                    open_times = self._ifind_field(raw, ("开板次数", "打开涨停次数"))
-                    try:
-                        open_count = max(0, int(float(open_times))) if open_times not in (None, "") else None
-                    except (TypeError, ValueError):
-                        open_count = None
-                    result[kind][code] = {
-                        "reason": reason,
-                        "first_time": first_time,
-                        "last_time": last_time,
-                        "open_times": open_count,
-                    }
-            if any(result[kind] for kind in ("limits", "broken", "down_limits")):
-                self.database.save_data_snapshot(
-                    "ifind_event_enrichment_v1", trade_date, "ifind", result
-                )
-        finally:
-            self._ifind_event_lock.release()
-
-    @staticmethod
-    def _normalize_ifind_event_time(value: Any) -> str:
-        text = str(value or "").strip()
-        match = re.search(r"(?:^|\s)(\d{1,2}:\d{2}(?::\d{2})?)(?:$|\s)", text)
-        if not match:
-            match = re.search(r"(?<!\d)(\d{6})(?!\d)", text)
-            if match:
-                compact = match.group(1)
-                return f"{compact[:2]}:{compact[2:4]}:{compact[4:]}"
-            return ""
-        parts = match.group(1).split(":")
-        return ":".join(part.zfill(2) for part in parts)
-
-    @staticmethod
-    def _merge_ifind_event_enrichment(
-        dashboard: dict[str, Any], enrichment: dict[str, Any]
-    ) -> None:
-        for kind in ("limits", "broken", "down_limits"):
-            records = enrichment.get(kind) or {}
-            for row in dashboard.get(kind) or []:
-                event = records.get(str(row.get("code") or "")) or {}
-                reason = str(event.get("reason") or "").strip()
-                if reason:
-                    row["reason"] = reason
-                    row["reason_source"] = "market_event"
-                if event.get("first_time"):
-                    row["first_time"] = event["first_time"]
-                if event.get("last_time"):
-                    row["last_time"] = event["last_time"]
-                if event.get("open_times") is not None:
-                    row["open_times"] = event["open_times"]
 
     def _apply_seat_aliases(self, payload: dict[str, Any]) -> dict[str, Any]:
         aliases = self.database.list_seat_aliases()

backend/data/providers/tushare_client.py

@@ -11,7 +11,7 @@ from datetime import datetime, time as dt_time, timedelta
 from threading import Lock
 from typing import Any, ClassVar
 
-from sentiment_engine import apply_sentiment_to_dashboard
+from backend.features.sentiment.engine import apply_sentiment_to_dashboard
 
 
 TUSHARE_URL = "http://api.tushare.pro"

backend/features/market/service.py

@@ -14,7 +14,7 @@ from backend.bootstrap.config import (
 from backend.data.providers.ifind_client import IfindError
 from backend.data.providers.tushare_client import TushareClient, TushareError
 from backend.features.market.charts import ChartDataError
-from sentiment_engine import SENTIMENT_ENGINE_VERSION
+from backend.features.sentiment.engine import SENTIMENT_ENGINE_VERSION
 
 
 SEARCH_INDEXES = (

backend/features/pools/__init__.py

@@ -0,0 +1,6 @@
+"""Limit-up, broken-board, limit-down and prior-limit pool feature."""
+
+from .repository import PoolRepositoryMixin
+from .service import PoolServiceMixin
+
+__all__ = ["PoolRepositoryMixin", "PoolServiceMixin"]

backend/features/pools/repository.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+
+class PoolRepositoryMixin:
+    def save_reason_override(self, trade_date: str, code: str, reason: str) -> None:
+        now = datetime.now().astimezone().isoformat(timespec="seconds")
+        with self.connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO reason_overrides (trade_date, code, reason, updated_at)
+                VALUES (?, ?, ?, ?)
+                ON CONFLICT(trade_date, code) DO UPDATE SET
+                    reason = excluded.reason,
+                    updated_at = excluded.updated_at
+                """,
+                (trade_date, code, reason, now),
+            )
+
+    def reason_overrides(self, trade_date: str) -> dict[str, str]:
+        with self.connect() as connection:
+            rows = connection.execute(
+                "SELECT code, reason FROM reason_overrides WHERE trade_date = ?",
+                (trade_date,),
+            ).fetchall()
+        return {row["code"]: row["reason"] for row in rows}

backend/features/pools/service.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime, time as dt_time
+from typing import Any
+
+from backend.bootstrap.config import normalize_date, validate_stock_code
+from backend.data.providers.ifind_client import IfindError
+
+
+class PoolServiceMixin:
+    def save_reason(self, trade_date: str, code: str, reason: str) -> None:
+        normalized_date = normalize_date(trade_date)
+        code = validate_stock_code(code)
+        reason = reason.strip()
+        if not reason or len(reason) > 200:
+            raise ValueError("涨停原因应为 1 至 200 个字符。")
+        self.database.save_reason_override(normalized_date, code, reason)
+
+    def _apply_reason_overrides(self, dashboard: dict[str, Any]) -> dict[str, Any]:
+        trade_date = str(dashboard.get("meta", {}).get("trade_date", "")).replace("-", "")
+        enrichment = self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date)
+        if enrichment:
+            self._merge_ifind_event_enrichment(dashboard, enrichment)
+        else:
+            self._schedule_ifind_event_enrichment(trade_date)
+        overrides = self.database.reason_overrides(trade_date)
+        if not overrides:
+            return dashboard
+        for key in ("limits", "broken", "down_limits"):
+            for row in dashboard.get(key) or []:
+                if row.get("code") in overrides:
+                    row["reason"] = overrides[row["code"]]
+                    row["reason_source"] = "manual"
+        return dashboard
+
+    def _schedule_ifind_event_enrichment(self, trade_date: str) -> None:
+        ifind = getattr(self, "ifind", None)
+        if not ifind or not ifind.configured or not re.fullmatch(r"\d{8}", trade_date):
+            return
+        now = datetime.now().astimezone()
+        if trade_date == now.strftime("%Y%m%d") and now.time().replace(tzinfo=None) < dt_time(15, 0):
+            return
+        self.jobs.submit(
+            "market.ifind-event-enrichment",
+            f"{trade_date}:v1",
+            lambda: self._refresh_ifind_event_enrichment(trade_date),
+            {"trade_date": trade_date, "trigger": "dashboard-enrichment"},
+        )
+
+    def _refresh_ifind_event_enrichment(self, trade_date: str) -> None:
+        if not self._ifind_event_lock.acquire(blocking=False):
+            return
+        try:
+            if self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date):
+                return
+            ifind = getattr(self, "ifind", None)
+            if not ifind or not ifind.configured:
+                return
+            current = datetime.strptime(trade_date, "%Y%m%d")
+            display_date = f"{current.year}年{current.month}月{current.day}日"
+            requests = {
+                "limits": (
+                    f"{display_date}涨停股票，股票代码、股票简称、涨停原因、"
+                    "首次涨停时间、最终涨停时间、开板次数"
+                ),
+                "broken": (
+                    f"{display_date}曾涨停但收盘未涨停的股票，股票代码、股票简称、"
+                    "涨停原因、首次涨停时间、开板次数"
+                ),
+                "down_limits": (
+                    f"{display_date}跌停股票，股票代码、股票简称、跌停原因"
+                ),
+            }
+            result: dict[str, Any] = {
+                "trade_date": trade_date,
+                "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+                "limits": {}, "broken": {}, "down_limits": {}, "partial": False,
+            }
+            for kind, query in requests.items():
+                try:
+                    rows = ifind.wencai(query, "stock", cache_ttl=900)
+                except IfindError:
+                    result["partial"] = True
+                    continue
+                for raw in rows:
+                    code = self._ifind_row_code(raw)
+                    if not code:
+                        continue
+                    reason_tokens = (
+                        ("跌停原因", "风险线索", "原因")
+                        if kind == "down_limits"
+                        else ("涨停原因类别", "涨停原因", "触板逻辑", "原因")
+                    )
+                    reason = str(self._ifind_field(raw, reason_tokens) or "").strip()
+                    first_time = self._normalize_ifind_event_time(
+                        self._ifind_field(raw, ("首次涨停时间", "首次触板时间", "首次封板时间"))
+                    )
+                    last_time = self._normalize_ifind_event_time(
+                        self._ifind_field(raw, ("最终涨停时间", "最后涨停时间", "最后封板时间"))
+                    )
+                    open_times = self._ifind_field(raw, ("开板次数", "打开涨停次数"))
+                    try:
+                        open_count = max(0, int(float(open_times))) if open_times not in (None, "") else None
+                    except (TypeError, ValueError):
+                        open_count = None
+                    result[kind][code] = {
+                        "reason": reason,
+                        "first_time": first_time,
+                        "last_time": last_time,
+                        "open_times": open_count,
+                    }
+            if any(result[kind] for kind in ("limits", "broken", "down_limits")):
+                self.database.save_data_snapshot(
+                    "ifind_event_enrichment_v1", trade_date, "ifind", result
+                )
+        finally:
+            self._ifind_event_lock.release()
+
+    @staticmethod
+    def _normalize_ifind_event_time(value: Any) -> str:
+        text = str(value or "").strip()
+        match = re.search(r"(?:^|\s)(\d{1,2}:\d{2}(?::\d{2})?)(?:$|\s)", text)
+        if not match:
+            match = re.search(r"(?<!\d)(\d{6})(?!\d)", text)
+            if match:
+                compact = match.group(1)
+                return f"{compact[:2]}:{compact[2:4]}:{compact[4:]}"
+            return ""
+        parts = match.group(1).split(":")
+        return ":".join(part.zfill(2) for part in parts)
+
+    @staticmethod
+    def _merge_ifind_event_enrichment(
+        dashboard: dict[str, Any], enrichment: dict[str, Any]
+    ) -> None:
+        for kind in ("limits", "broken", "down_limits"):
+            records = enrichment.get(kind) or {}
+            for row in dashboard.get(kind) or []:
+                event = records.get(str(row.get("code") or "")) or {}
+                reason = str(event.get("reason") or "").strip()
+                if reason:
+                    row["reason"] = reason
+                    row["reason_source"] = "market_event"
+                if event.get("first_time"):
+                    row["first_time"] = event["first_time"]
+                if event.get("last_time"):
+                    row["last_time"] = event["last_time"]
+                if event.get("open_times") is not None:
+                    row["open_times"] = event["open_times"]

backend/features/sentiment/__init__.py

@@ -0,0 +1,19 @@
+"""Market sentiment cycle and history feature."""
+
+from .engine import (
+    COMPONENT_WEIGHTS,
+    SENTIMENT_ENGINE_VERSION,
+    apply_sentiment_to_dashboard,
+    build_sentiment_history,
+    latest_contiguous_history,
+)
+from .service import SentimentServiceMixin
+
+__all__ = [
+    "COMPONENT_WEIGHTS",
+    "SENTIMENT_ENGINE_VERSION",
+    "SentimentServiceMixin",
+    "apply_sentiment_to_dashboard",
+    "build_sentiment_history",
+    "latest_contiguous_history",
+]

backend/features/sentiment/engine.py

@@ -0,0 +1,496 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from statistics import mean, median
+from typing import Any
+
+
+COMPONENT_WEIGHTS = {
+    "breadth": 20,
+    "limit_ecology": 25,
+    "profit_effect": 30,
+    "ladder_structure": 15,
+    "liquidity": 10,
+}
+
+SENTIMENT_ENGINE_VERSION = 2
+
+
+def _number(value: Any, default: float = 0.0) -> float:
+    try:
+        number = float(value)
+        return number if number == number else default
+    except (TypeError, ValueError):
+        return default
+
+
+def _clamp(value: float, lower: float = 0.0, upper: float = 100.0) -> float:
+    return min(upper, max(lower, value))
+
+
+def _linear(value: float, low: float, high: float) -> float:
+    if high <= low:
+        return 50.0
+    return _clamp((value - low) / (high - low) * 100)
+
+
+def _percentile(value: float, history: list[float]) -> float:
+    if not history:
+        return 50.0
+    below = sum(item < value for item in history)
+    equal = sum(item == value for item in history)
+    return _clamp((below + equal * 0.5) / len(history) * 100)
+
+
+def _adaptive_score(value: float, fixed: float, history: list[float]) -> float:
+    if len(history) < 20:
+        return fixed
+    return fixed * 0.25 + _percentile(value, history[-250:]) * 0.75
+
+
+def _trade_date(payload: dict[str, Any]) -> str:
+    meta = payload.get("meta") or {}
+    return str(meta.get("trade_date") or payload.get("_snapshot_date") or "").replace("-", "")
+
+
+def _deduplicate_snapshots(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    by_trade_date: dict[str, dict[str, Any]] = {}
+    for payload in snapshots:
+        trade_date = _trade_date(payload)
+        if trade_date:
+            by_trade_date[trade_date] = payload
+    return [by_trade_date[key] for key in sorted(by_trade_date)]
+
+
+def _snapshot_stats(payload: dict[str, Any]) -> dict[str, Any]:
+    overview = payload.get("overview") or {}
+    meta = payload.get("meta") or {}
+    limits = list(payload.get("limits") or [])
+    broken = list(payload.get("broken") or [])
+    down_limits = list(payload.get("down_limits") or [])
+    yesterday = list(payload.get("yesterday_limits") or [])
+
+    limit_up = len(limits) if limits else int(_number(overview.get("limit_up_count")))
+    broken_count = len(broken) if broken else int(_number(overview.get("broken_count")))
+    limit_down = len(down_limits) if down_limits else int(_number(overview.get("limit_down_count")))
+    streaks = [max(1, int(_number(row.get("streak"), 1))) for row in limits]
+    first_board = sum(streak == 1 for streak in streaks)
+    second_board = sum(streak == 2 for streak in streaks)
+    three_plus = sum(streak >= 3 for streak in streaks)
+    max_height = max(streaks, default=0)
+    present_levels = set(streaks)
+    ladder_completeness = (
+        sum(level in present_levels for level in range(1, max_height + 1)) / max_height * 100
+        if max_height else 0.0
+    )
+
+    up_count = int(_number(overview.get("up_count")))
+    down_count = int(_number(overview.get("down_count")))
+    flat_count = int(_number(overview.get("flat_count")))
+    active_count = up_count + down_count
+    breadth_ratio = up_count / max(active_count, 1) * 100
+    seal_rate = _number(overview.get("seal_rate"))
+    if not seal_rate and limit_up + broken_count:
+        seal_rate = limit_up / (limit_up + broken_count) * 100
+
+    previous_limit_count = len(yesterday)
+    previous_positive_count = sum(_number(row.get("current_change")) > 0 for row in yesterday)
+    previous_positive_rate = previous_positive_count / max(previous_limit_count, 1) * 100
+    advanced_count = sum(row.get("outcome") == "晋级" for row in yesterday)
+    advance_rate = advanced_count / max(previous_limit_count, 1) * 100
+    average_previous_change = (
+        mean(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0
+    )
+    median_previous_change = (
+        median(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0
+    )
+    severe_loss_count = sum(_number(row.get("current_change")) <= -5 for row in yesterday)
+    severe_loss_rate = severe_loss_count / max(previous_limit_count, 1) * 100
+    previous_down_count = sum(row.get("outcome") == "跌停" for row in yesterday)
+    high_previous = [row for row in yesterday if int(_number(row.get("prior_streak"), 1)) >= 2]
+    high_positive_rate = (
+        sum(_number(row.get("current_change")) > 0 for row in high_previous)
+        / max(len(high_previous), 1)
+        * 100
+    )
+
+    amount_billion = _number(overview.get("amount_billion"))
+    limit_amount_billion = sum(_number(row.get("amount_billion")) for row in limits)
+    return {
+        "trade_date": _trade_date(payload),
+        "previous_trade_date": str(meta.get("previous_trade_date") or "").replace("-", ""),
+        "up_count": up_count,
+        "down_count": down_count,
+        "flat_count": flat_count,
+        "breadth_ratio": round(breadth_ratio, 1),
+        "limit_up_count": limit_up,
+        "first_board_count": first_board,
+        "second_board_count": second_board,
+        "three_plus_count": three_plus,
+        "max_height": max_height,
+        "ladder_completeness": round(ladder_completeness, 1),
+        "broken_count": broken_count,
+        "limit_down_count": limit_down,
+        "seal_rate": round(seal_rate, 1),
+        "previous_limit_count": previous_limit_count,
+        "previous_positive_count": previous_positive_count,
+        "previous_positive_rate": round(previous_positive_rate, 1),
+        "advance_rate": round(advance_rate, 1),
+        "average_previous_change": round(average_previous_change, 2),
+        "median_previous_change": round(median_previous_change, 2),
+        "severe_loss_count": severe_loss_count,
+        "severe_loss_rate": round(severe_loss_rate, 1),
+        "previous_down_count": previous_down_count,
+        "high_positive_rate": round(high_positive_rate, 1),
+        "amount_billion": round(amount_billion, 1),
+        "limit_amount_billion": round(limit_amount_billion, 2),
+    }
+
+
+def _sentiment_label(score: float) -> str:
+    if score >= 80:
+        return "情绪高涨"
+    if score >= 60:
+        return "情绪偏强"
+    if score >= 40:
+        return "情绪中性"
+    if score >= 20:
+        return "情绪偏弱"
+    return "情绪冰点"
+
+
+def _phase_signal(score: float, momentum: float, profit_score: float) -> str:
+    if score < 25:
+        return "修复" if momentum > 3 else "冰点"
+    if score < 45:
+        return "修复" if momentum > 3 else "退潮"
+    if score >= 80:
+        return "高潮" if momentum >= -2 and profit_score >= 60 else "分化"
+    if score >= 65:
+        return "分化" if momentum < -3 or profit_score < 50 else "发酵"
+    if momentum < -5:
+        return "退潮"
+    return "发酵" if momentum >= 0 and profit_score >= 45 else "分化"
+
+
+def _confirmed_phase(
+    previous: dict[str, Any] | None,
+    score: float,
+    day_change: float,
+    systemic_health: float,
+    profit_score: float,
+    ecology_score: float,
+    phase_signal: str,
+    extreme_ice: bool,
+    fermentation_signal_count: int,
+) -> tuple[str, str]:
+    if previous is None:
+        return phase_signal, "首个连续交易日，采用原始阶段信号"
+    previous_phase = str(previous.get("phase") or phase_signal)
+    if extreme_ice:
+        return "冰点", "市场宽度与跌停数量触发极端冰点"
+
+    recovery = day_change >= 6 and score >= 25 and systemic_health >= 24
+    fermentation_confirmed = fermentation_signal_count >= 2
+    climax_ready = (
+        score >= 80
+        and profit_score >= 60
+        and systemic_health >= 60
+        and ecology_score >= 70
+    )
+
+    if previous_phase == "冰点":
+        return ("修复", "冰点后首次有效回升") if recovery else ("冰点", "冰点尚未形成有效修复")
+
+    if previous_phase == "退潮":
+        if score < 25:
+            return "冰点", "退潮继续下探至冰点区间"
+        return ("修复", "退潮后出现有效回升") if recovery else ("退潮", "退潮尚未形成有效修复")
+
+    if previous_phase == "修复":
+        if score < 25:
+            return "冰点", "修复失败并重新跌入冰点区间"
+        if day_change <= -6 and score < 45:
+            return "退潮", "修复失败且温度显著回落"
+        if fermentation_confirmed:
+            return "发酵", "发酵条件连续两个交易日成立"
+        return "修复", "修复延续，等待发酵确认"
+
+    if previous_phase == "发酵":
+        if score < 25:
+            return "冰点", "发酵阶段出现极端情绪坍塌"
+        if score < 45 and (day_change < 0 or systemic_health < 35):
+            return "退潮", "发酵阶段温度与系统健康度同步转弱"
+        if climax_ready:
+            return "高潮", "温度、赚钱效应与涨停生态共同达到高潮条件"
+        if phase_signal in {"分化", "退潮"} or day_change <= -6:
+            return "分化", "发酵阶段出现降温或赚钱效应弱化"
+        return "发酵", "发酵状态延续"
+
+    if previous_phase == "高潮":
+        if score < 25:
+            return "冰点", "高潮后出现极端情绪坍塌"
+        if climax_ready:
+            return "高潮", "高潮条件继续成立"
+        if score < 45 or systemic_health < 30:
+            return "退潮", "高潮后风险快速释放"
+        return "分化", "高潮条件消退，进入分化"
+
+    if previous_phase == "分化":
+        if score < 25:
+            return "冰点", "分化继续恶化至冰点区间"
+        if score < 45 or systemic_health < 30:
+            return "退潮", "分化后温度或系统健康度继续下降"
+        if fermentation_confirmed:
+            return "发酵", "分化转强条件连续两个交易日成立"
+        return "分化", "分化延续，等待方向确认"
+
+    return phase_signal, "采用原始阶段信号"
+
+
+def build_sentiment_history(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    payloads = _deduplicate_snapshots(snapshots)
+    raw_rows = [_snapshot_stats(payload) for payload in payloads]
+    results: list[dict[str, Any]] = []
+
+    for index, stats in enumerate(raw_rows):
+        previous = raw_rows[:index]
+        limit_history = [float(row["limit_up_count"]) for row in previous]
+        down_limit_history = [float(row["limit_down_count"]) for row in previous]
+        height_history = [float(row["max_height"]) for row in previous]
+        three_plus_history = [float(row["three_plus_count"]) for row in previous]
+        amount_history = [float(row["amount_billion"]) for row in previous[-20:] if row["amount_billion"]]
+
+        breadth_score = _clamp(float(stats["breadth_ratio"]))
+        limit_strength = _adaptive_score(
+            float(stats["limit_up_count"]),
+            _linear(float(stats["limit_up_count"]), 10, 100),
+            limit_history,
+        )
+        down_relief = 100 - _adaptive_score(
+            float(stats["limit_down_count"]),
+            _linear(float(stats["limit_down_count"]), 0, 50),
+            down_limit_history,
+        )
+        seal_quality = _linear(float(stats["seal_rate"]), 35, 90)
+        systemic_health = breadth_score * 0.60 + down_relief * 0.40
+        systemic_gate = 1.0 if systemic_health >= 35 else 0.35 + systemic_health / 35 * 0.65
+        ecology_base_score = limit_strength * 0.35 + seal_quality * 0.35 + down_relief * 0.30
+        # Systemic risk is applied once to the final temperature. Reapplying it here
+        # would count market breadth and limit-down pressure twice.
+        limit_ecology_score = ecology_base_score
+
+        if stats["previous_limit_count"]:
+            positive_score = float(stats["previous_positive_rate"])
+            average_change_score = _clamp(50 + float(stats["average_previous_change"]) * 6)
+            median_change_score = _clamp(50 + float(stats["median_previous_change"]) * 7)
+            advance_score = _clamp(float(stats["advance_rate"]) * 2.5)
+            severe_loss_safety = _clamp(100 - float(stats["severe_loss_rate"]) * 3)
+            down_safety = _clamp(100 - float(stats["previous_down_count"]) / stats["previous_limit_count"] * 700)
+            tail_safety_score = severe_loss_safety * 0.70 + down_safety * 0.30
+            profit_effect_score = (
+                positive_score * 0.30
+                + median_change_score * 0.25
+                + average_change_score * 0.10
+                + advance_score * 0.20
+                + tail_safety_score * 0.15
+            )
+        else:
+            profit_effect_score = 50.0
+
+        max_height_score = _adaptive_score(
+            float(stats["max_height"]),
+            _linear(float(stats["max_height"]), 1, 7),
+            height_history,
+        )
+        continuation_rate = (
+            (float(stats["second_board_count"]) + float(stats["three_plus_count"]))
+            / max(float(stats["limit_up_count"]), 1)
+            * 100
+        )
+        three_plus_density = float(stats["three_plus_count"]) / max(float(stats["limit_up_count"]), 1) * 100
+        three_plus_score = _adaptive_score(
+            float(stats["three_plus_count"]),
+            _clamp(three_plus_density * 5),
+            three_plus_history,
+        )
+        ladder_structure_score = (
+            max_height_score * 0.30
+            + _clamp(continuation_rate * 3) * 0.25
+            + three_plus_score * 0.25
+            + float(stats["ladder_completeness"]) * 0.20
+        )
+
+        amount_baseline = mean(amount_history) if amount_history else float(stats["amount_billion"] or 1)
+        amount_ratio = float(stats["amount_billion"]) / max(amount_baseline, 1)
+        amount_score = _clamp(50 + (amount_ratio - 1) * 100)
+        limit_amount_share = float(stats["limit_amount_billion"]) / max(float(stats["amount_billion"]), 1) * 100
+        liquidity_score = amount_score * 0.70 + _clamp(limit_amount_share * 20) * 0.30
+
+        component_scores = {
+            "breadth": breadth_score,
+            "limit_ecology": limit_ecology_score,
+            "profit_effect": profit_effect_score,
+            "ladder_structure": ladder_structure_score,
+            "liquidity": liquidity_score,
+        }
+        raw_score = sum(component_scores[key] * weight / 100 for key, weight in COMPONENT_WEIGHTS.items())
+        score = round(
+            raw_score * systemic_gate
+        )
+        extreme_ice = float(stats["breadth_ratio"]) <= 15 and float(stats["limit_down_count"]) >= 100
+        if extreme_ice:
+            score = min(score, 15)
+        elif float(stats["breadth_ratio"]) <= 25 and float(stats["limit_down_count"]) >= 50:
+            score = min(score, 24)
+        previous_scores: list[float] = []
+        expected_date = str(stats.get("previous_trade_date") or "")
+        for prior_result in reversed(results):
+            if not expected_date or str(prior_result.get("trade_date") or "") != expected_date:
+                break
+            previous_scores.append(float(prior_result["score"]))
+            expected_date = str(prior_result.get("previous_trade_date") or "")
+            if len(previous_scores) == 3:
+                break
+        momentum = score - mean(previous_scores) if previous_scores else 0.0
+        direction = "升温" if momentum > 3 else "降温" if momentum < -3 else "持平"
+        normalization = "历史百分位" if len(previous) >= 20 else "固定锚点"
+        previous_result = (
+            results[-1]
+            if results and str(stats.get("previous_trade_date") or "") == str(results[-1].get("trade_date") or "")
+            else None
+        )
+        day_change = score - float(previous_result["score"]) if previous_result else 0.0
+        ema_score = round(
+            score if not previous_result
+            else score * 0.5 + float(previous_result.get("ema_score", previous_result["score"])) * 0.5,
+            1,
+        )
+        phase_signal = _phase_signal(score, momentum, profit_effect_score)
+        fermentation_ready = (
+            phase_signal == "发酵"
+            and score >= 45
+            and profit_effect_score >= 45
+            and systemic_health >= 35
+            and not extreme_ice
+        )
+        previous_fermentation_count = int(previous_result.get("fermentation_signal_count") or 0) if previous_result else 0
+        fermentation_signal_count = previous_fermentation_count + 1 if fermentation_ready else 0
+        phase, transition_reason = _confirmed_phase(
+            previous_result,
+            score,
+            day_change,
+            systemic_health,
+            profit_effect_score,
+            limit_ecology_score,
+            phase_signal,
+            extreme_ice,
+            fermentation_signal_count,
+        )
+        previous_phase = str(previous_result.get("phase") or "") if previous_result else ""
+        if phase not in {"修复", "分化"}:
+            fermentation_signal_count = 0
+        elif phase == "分化" and previous_phase != "分化":
+            fermentation_signal_count = 0
+
+        components = {
+            "breadth": {
+                "label": "市场宽度",
+                "score": round(breadth_score, 1),
+                "weight": COMPONENT_WEIGHTS["breadth"],
+                "summary": f"上涨占比 {stats['breadth_ratio']:.1f}%",
+            },
+            "limit_ecology": {
+                "label": "涨停生态",
+                "score": round(limit_ecology_score, 1),
+                "weight": COMPONENT_WEIGHTS["limit_ecology"],
+                "summary": (
+                    f"涨停 {stats['limit_up_count']} · 跌停 {stats['limit_down_count']} · "
+                    f"封板 {stats['seal_rate']:.1f}%"
+                ),
+            },
+            "profit_effect": {
+                "label": "赚钱效应",
+                "score": round(profit_effect_score, 1),
+                "weight": COMPONENT_WEIGHTS["profit_effect"],
+                "summary": (
+                    f"昨涨停红盘 {stats['previous_positive_rate']:.1f}% · "
+                    f"中位 {stats['median_previous_change']:+.2f}% · "
+                    f"重亏 {stats['severe_loss_rate']:.1f}%"
+                    if stats["previous_limit_count"] else "缺少前一交易日样本"
+                ),
+            },
+            "ladder_structure": {
+                "label": "连板结构",
+                "score": round(ladder_structure_score, 1),
+                "weight": COMPONENT_WEIGHTS["ladder_structure"],
+                "summary": f"最高 {stats['max_height']} 板 · 三板以上 {stats['three_plus_count']} 家",
+            },
+            "liquidity": {
+                "label": "成交活跃度",
+                "score": round(liquidity_score, 1),
+                "weight": COMPONENT_WEIGHTS["liquidity"],
+                "summary": f"成交 {stats['amount_billion']:.1f} 亿 · 均值比 {amount_ratio:.2f}",
+            },
+        }
+        results.append(
+            {
+                **stats,
+                "score": score,
+                "ema_score": ema_score,
+                "label": _sentiment_label(score),
+                "phase": phase,
+                "phase_signal": phase_signal,
+                "transition_reason": transition_reason,
+                "fermentation_signal_count": fermentation_signal_count,
+                "day_change": round(day_change, 1),
+                "direction": direction,
+                "momentum": round(momentum, 1),
+                "normalization": "250日历史百分位" if len(previous) >= 20 else normalization,
+                "history_days": len(previous) + 1,
+                "systemic_health": round(systemic_health, 1),
+                "risk_multiplier": round(systemic_gate, 3),
+                "components": components,
+            }
+        )
+    return results
+
+
+def latest_contiguous_history(series: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    if not series:
+        return []
+    contiguous = [series[-1]]
+    for row in reversed(series[:-1]):
+        expected_previous = str(contiguous[0].get("previous_trade_date") or "")
+        if not expected_previous or expected_previous != str(row.get("trade_date") or ""):
+            break
+        contiguous.insert(0, row)
+    return contiguous
+
+
+def apply_sentiment_to_dashboard(
+    dashboard: dict[str, Any],
+    historical_snapshots: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    result = deepcopy(dashboard)
+    history = list(historical_snapshots or [])
+    history.append(result)
+    series = build_sentiment_history(history)
+    target_date = _trade_date(result)
+    sentiment = next((row for row in reversed(series) if row["trade_date"] == target_date), None)
+    if not sentiment:
+        return result
+    overview = dict(result.get("overview") or {})
+    overview.update(
+        {
+            "sentiment_score": sentiment["score"],
+            "sentiment_trend_score": sentiment["ema_score"],
+            "sentiment_label": sentiment["label"],
+            "sentiment_phase": sentiment["phase"],
+            "sentiment_direction": sentiment["direction"],
+            "sentiment_components": sentiment["components"],
+            "sentiment_engine_version": SENTIMENT_ENGINE_VERSION,
+        }
+    )
+    result["overview"] = overview
+    return result

backend/features/sentiment/service.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.bootstrap.config import normalize_date
+from backend.features.sentiment.engine import (
+    COMPONENT_WEIGHTS,
+    apply_sentiment_to_dashboard,
+    build_sentiment_history,
+    latest_contiguous_history,
+)
+
+
+class SentimentServiceMixin:
+    def _enrich_dashboard_sentiment(
+        self,
+        dashboard: dict[str, Any],
+        end_date: str,
+    ) -> dict[str, Any]:
+        history = self.database.list_snapshot_payloads(end_date, 260)
+        return apply_sentiment_to_dashboard(dashboard, history)
+
+    def sentiment_history(self, trade_date: str, limit: int = 20) -> dict[str, Any]:
+        normalized_date = normalize_date(trade_date)
+        limit = max(10, min(120, int(limit)))
+        full_series = build_sentiment_history(
+            self.database.list_snapshot_payloads(normalized_date, 240)
+        )
+        series = latest_contiguous_history(full_series)
+        rows = series[-limit:]
+        return {
+            "trade_date": rows[-1]["trade_date"] if rows else normalized_date,
+            "available_days": len(series),
+            "stored_days": len(full_series),
+            "requested_days": limit,
+            "rows": rows,
+            "weights": COMPONENT_WEIGHTS,
+            "normalization": rows[-1]["normalization"] if rows else "固定锚点",
+        }

database.py

@@ -9,6 +9,7 @@ from typing import Any
 from backend.database import MIGRATIONS, MigrationRunner, SQLiteConnectionFactory
 from backend.features.accounts.repository import AccountRepositoryMixin
 from backend.features.market.repository import MarketRepositoryMixin
+from backend.features.pools.repository import PoolRepositoryMixin
 from backend.features.system.repository import SystemSettingsRepositoryMixin
 
 
@@ -24,6 +25,7 @@ def _optional_float(value: Any) -> float | None:
 class ReviewDatabase(
     AccountRepositoryMixin,
     MarketRepositoryMixin,
+    PoolRepositoryMixin,
     SystemSettingsRepositoryMixin,
 ):
     def __init__(self, path: Path) -> None:
@@ -836,27 +838,6 @@ class ReviewDatabase(
             )
             return cursor.rowcount > 0
 
-    def save_reason_override(self, trade_date: str, code: str, reason: str) -> None:
-        now = datetime.now().astimezone().isoformat(timespec="seconds")
-        with self.connect() as connection:
-            connection.execute(
-                """
-                INSERT INTO reason_overrides (trade_date, code, reason, updated_at)
-                VALUES (?, ?, ?, ?)
-                ON CONFLICT(trade_date, code) DO UPDATE SET
-                    reason = excluded.reason,
-                    updated_at = excluded.updated_at
-                """,
-                (trade_date, code, reason, now),
-            )
-
-    def reason_overrides(self, trade_date: str) -> dict[str, str]:
-        with self.connect() as connection:
-            rows = connection.execute(
-                "SELECT code, reason FROM reason_overrides WHERE trade_date = ?",
-                (trade_date,),
-            ).fetchall()
-        return {row["code"]: row["reason"] for row in rows}
 
     def list_seat_aliases(self) -> dict[str, str]:
         with self.connect() as connection:

demo_data.py

@@ -5,7 +5,7 @@ import math
 from datetime import datetime, timedelta
 from typing import Any
 
-from sentiment_engine import apply_sentiment_to_dashboard
+from backend.features.sentiment.engine import apply_sentiment_to_dashboard
 
 
 DEMO_LIMITS = [

screener.py

@@ -10,7 +10,7 @@ from typing import Any
 
 from advanced_strategies import ADVANCED_CURATED_STRATEGIES
 from database import ReviewDatabase
-from sentiment_engine import build_sentiment_history, latest_contiguous_history
+from backend.features.sentiment.engine import build_sentiment_history, latest_contiguous_history
 from tushare_client import TushareClient, TushareError
 
 

sentiment_engine.py

@@ -1,496 +1,7 @@
-from __future__ import annotations
+"""Compatibility alias for the canonical sentiment engine implementation."""
 
-from copy import deepcopy
+import sys
-from statistics import mean, median
-from typing import Any
 
+from backend.features.sentiment import engine as _implementation
 
-COMPONENT_WEIGHTS = {
+sys.modules[__name__] = _implementation
-    "breadth": 20,
-    "limit_ecology": 25,
-    "profit_effect": 30,
-    "ladder_structure": 15,
-    "liquidity": 10,
-}
-
-SENTIMENT_ENGINE_VERSION = 2
-
-
-def _number(value: Any, default: float = 0.0) -> float:
-    try:
-        number = float(value)
-        return number if number == number else default
-    except (TypeError, ValueError):
-        return default
-
-
-def _clamp(value: float, lower: float = 0.0, upper: float = 100.0) -> float:
-    return min(upper, max(lower, value))
-
-
-def _linear(value: float, low: float, high: float) -> float:
-    if high <= low:
-        return 50.0
-    return _clamp((value - low) / (high - low) * 100)
-
-
-def _percentile(value: float, history: list[float]) -> float:
-    if not history:
-        return 50.0
-    below = sum(item < value for item in history)
-    equal = sum(item == value for item in history)
-    return _clamp((below + equal * 0.5) / len(history) * 100)
-
-
-def _adaptive_score(value: float, fixed: float, history: list[float]) -> float:
-    if len(history) < 20:
-        return fixed
-    return fixed * 0.25 + _percentile(value, history[-250:]) * 0.75
-
-
-def _trade_date(payload: dict[str, Any]) -> str:
-    meta = payload.get("meta") or {}
-    return str(meta.get("trade_date") or payload.get("_snapshot_date") or "").replace("-", "")
-
-
-def _deduplicate_snapshots(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    by_trade_date: dict[str, dict[str, Any]] = {}
-    for payload in snapshots:
-        trade_date = _trade_date(payload)
-        if trade_date:
-            by_trade_date[trade_date] = payload
-    return [by_trade_date[key] for key in sorted(by_trade_date)]
-
-
-def _snapshot_stats(payload: dict[str, Any]) -> dict[str, Any]:
-    overview = payload.get("overview") or {}
-    meta = payload.get("meta") or {}
-    limits = list(payload.get("limits") or [])
-    broken = list(payload.get("broken") or [])
-    down_limits = list(payload.get("down_limits") or [])
-    yesterday = list(payload.get("yesterday_limits") or [])
-
-    limit_up = len(limits) if limits else int(_number(overview.get("limit_up_count")))
-    broken_count = len(broken) if broken else int(_number(overview.get("broken_count")))
-    limit_down = len(down_limits) if down_limits else int(_number(overview.get("limit_down_count")))
-    streaks = [max(1, int(_number(row.get("streak"), 1))) for row in limits]
-    first_board = sum(streak == 1 for streak in streaks)
-    second_board = sum(streak == 2 for streak in streaks)
-    three_plus = sum(streak >= 3 for streak in streaks)
-    max_height = max(streaks, default=0)
-    present_levels = set(streaks)
-    ladder_completeness = (
-        sum(level in present_levels for level in range(1, max_height + 1)) / max_height * 100
-        if max_height else 0.0
-    )
-
-    up_count = int(_number(overview.get("up_count")))
-    down_count = int(_number(overview.get("down_count")))
-    flat_count = int(_number(overview.get("flat_count")))
-    active_count = up_count + down_count
-    breadth_ratio = up_count / max(active_count, 1) * 100
-    seal_rate = _number(overview.get("seal_rate"))
-    if not seal_rate and limit_up + broken_count:
-        seal_rate = limit_up / (limit_up + broken_count) * 100
-
-    previous_limit_count = len(yesterday)
-    previous_positive_count = sum(_number(row.get("current_change")) > 0 for row in yesterday)
-    previous_positive_rate = previous_positive_count / max(previous_limit_count, 1) * 100
-    advanced_count = sum(row.get("outcome") == "晋级" for row in yesterday)
-    advance_rate = advanced_count / max(previous_limit_count, 1) * 100
-    average_previous_change = (
-        mean(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0
-    )
-    median_previous_change = (
-        median(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0
-    )
-    severe_loss_count = sum(_number(row.get("current_change")) <= -5 for row in yesterday)
-    severe_loss_rate = severe_loss_count / max(previous_limit_count, 1) * 100
-    previous_down_count = sum(row.get("outcome") == "跌停" for row in yesterday)
-    high_previous = [row for row in yesterday if int(_number(row.get("prior_streak"), 1)) >= 2]
-    high_positive_rate = (
-        sum(_number(row.get("current_change")) > 0 for row in high_previous)
-        / max(len(high_previous), 1)
-        * 100
-    )
-
-    amount_billion = _number(overview.get("amount_billion"))
-    limit_amount_billion = sum(_number(row.get("amount_billion")) for row in limits)
-    return {
-        "trade_date": _trade_date(payload),
-        "previous_trade_date": str(meta.get("previous_trade_date") or "").replace("-", ""),
-        "up_count": up_count,
-        "down_count": down_count,
-        "flat_count": flat_count,
-        "breadth_ratio": round(breadth_ratio, 1),
-        "limit_up_count": limit_up,
-        "first_board_count": first_board,
-        "second_board_count": second_board,
-        "three_plus_count": three_plus,
-        "max_height": max_height,
-        "ladder_completeness": round(ladder_completeness, 1),
-        "broken_count": broken_count,
-        "limit_down_count": limit_down,
-        "seal_rate": round(seal_rate, 1),
-        "previous_limit_count": previous_limit_count,
-        "previous_positive_count": previous_positive_count,
-        "previous_positive_rate": round(previous_positive_rate, 1),
-        "advance_rate": round(advance_rate, 1),
-        "average_previous_change": round(average_previous_change, 2),
-        "median_previous_change": round(median_previous_change, 2),
-        "severe_loss_count": severe_loss_count,
-        "severe_loss_rate": round(severe_loss_rate, 1),
-        "previous_down_count": previous_down_count,
-        "high_positive_rate": round(high_positive_rate, 1),
-        "amount_billion": round(amount_billion, 1),
-        "limit_amount_billion": round(limit_amount_billion, 2),
-    }
-
-
-def _sentiment_label(score: float) -> str:
-    if score >= 80:
-        return "情绪高涨"
-    if score >= 60:
-        return "情绪偏强"
-    if score >= 40:
-        return "情绪中性"
-    if score >= 20:
-        return "情绪偏弱"
-    return "情绪冰点"
-
-
-def _phase_signal(score: float, momentum: float, profit_score: float) -> str:
-    if score < 25:
-        return "修复" if momentum > 3 else "冰点"
-    if score < 45:
-        return "修复" if momentum > 3 else "退潮"
-    if score >= 80:
-        return "高潮" if momentum >= -2 and profit_score >= 60 else "分化"
-    if score >= 65:
-        return "分化" if momentum < -3 or profit_score < 50 else "发酵"
-    if momentum < -5:
-        return "退潮"
-    return "发酵" if momentum >= 0 and profit_score >= 45 else "分化"
-
-
-def _confirmed_phase(
-    previous: dict[str, Any] | None,
-    score: float,
-    day_change: float,
-    systemic_health: float,
-    profit_score: float,
-    ecology_score: float,
-    phase_signal: str,
-    extreme_ice: bool,
-    fermentation_signal_count: int,
-) -> tuple[str, str]:
-    if previous is None:
-        return phase_signal, "首个连续交易日，采用原始阶段信号"
-    previous_phase = str(previous.get("phase") or phase_signal)
-    if extreme_ice:
-        return "冰点", "市场宽度与跌停数量触发极端冰点"
-
-    recovery = day_change >= 6 and score >= 25 and systemic_health >= 24
-    fermentation_confirmed = fermentation_signal_count >= 2
-    climax_ready = (
-        score >= 80
-        and profit_score >= 60
-        and systemic_health >= 60
-        and ecology_score >= 70
-    )
-
-    if previous_phase == "冰点":
-        return ("修复", "冰点后首次有效回升") if recovery else ("冰点", "冰点尚未形成有效修复")
-
-    if previous_phase == "退潮":
-        if score < 25:
-            return "冰点", "退潮继续下探至冰点区间"
-        return ("修复", "退潮后出现有效回升") if recovery else ("退潮", "退潮尚未形成有效修复")
-
-    if previous_phase == "修复":
-        if score < 25:
-            return "冰点", "修复失败并重新跌入冰点区间"
-        if day_change <= -6 and score < 45:
-            return "退潮", "修复失败且温度显著回落"
-        if fermentation_confirmed:
-            return "发酵", "发酵条件连续两个交易日成立"
-        return "修复", "修复延续，等待发酵确认"
-
-    if previous_phase == "发酵":
-        if score < 25:
-            return "冰点", "发酵阶段出现极端情绪坍塌"
-        if score < 45 and (day_change < 0 or systemic_health < 35):
-            return "退潮", "发酵阶段温度与系统健康度同步转弱"
-        if climax_ready:
-            return "高潮", "温度、赚钱效应与涨停生态共同达到高潮条件"
-        if phase_signal in {"分化", "退潮"} or day_change <= -6:
-            return "分化", "发酵阶段出现降温或赚钱效应弱化"
-        return "发酵", "发酵状态延续"
-
-    if previous_phase == "高潮":
-        if score < 25:
-            return "冰点", "高潮后出现极端情绪坍塌"
-        if climax_ready:
-            return "高潮", "高潮条件继续成立"
-        if score < 45 or systemic_health < 30:
-            return "退潮", "高潮后风险快速释放"
-        return "分化", "高潮条件消退，进入分化"
-
-    if previous_phase == "分化":
-        if score < 25:
-            return "冰点", "分化继续恶化至冰点区间"
-        if score < 45 or systemic_health < 30:
-            return "退潮", "分化后温度或系统健康度继续下降"
-        if fermentation_confirmed:
-            return "发酵", "分化转强条件连续两个交易日成立"
-        return "分化", "分化延续，等待方向确认"
-
-    return phase_signal, "采用原始阶段信号"
-
-
-def build_sentiment_history(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    payloads = _deduplicate_snapshots(snapshots)
-    raw_rows = [_snapshot_stats(payload) for payload in payloads]
-    results: list[dict[str, Any]] = []
-
-    for index, stats in enumerate(raw_rows):
-        previous = raw_rows[:index]
-        limit_history = [float(row["limit_up_count"]) for row in previous]
-        down_limit_history = [float(row["limit_down_count"]) for row in previous]
-        height_history = [float(row["max_height"]) for row in previous]
-        three_plus_history = [float(row["three_plus_count"]) for row in previous]
-        amount_history = [float(row["amount_billion"]) for row in previous[-20:] if row["amount_billion"]]
-
-        breadth_score = _clamp(float(stats["breadth_ratio"]))
-        limit_strength = _adaptive_score(
-            float(stats["limit_up_count"]),
-            _linear(float(stats["limit_up_count"]), 10, 100),
-            limit_history,
-        )
-        down_relief = 100 - _adaptive_score(
-            float(stats["limit_down_count"]),
-            _linear(float(stats["limit_down_count"]), 0, 50),
-            down_limit_history,
-        )
-        seal_quality = _linear(float(stats["seal_rate"]), 35, 90)
-        systemic_health = breadth_score * 0.60 + down_relief * 0.40
-        systemic_gate = 1.0 if systemic_health >= 35 else 0.35 + systemic_health / 35 * 0.65
-        ecology_base_score = limit_strength * 0.35 + seal_quality * 0.35 + down_relief * 0.30
-        # Systemic risk is applied once to the final temperature. Reapplying it here
-        # would count market breadth and limit-down pressure twice.
-        limit_ecology_score = ecology_base_score
-
-        if stats["previous_limit_count"]:
-            positive_score = float(stats["previous_positive_rate"])
-            average_change_score = _clamp(50 + float(stats["average_previous_change"]) * 6)
-            median_change_score = _clamp(50 + float(stats["median_previous_change"]) * 7)
-            advance_score = _clamp(float(stats["advance_rate"]) * 2.5)
-            severe_loss_safety = _clamp(100 - float(stats["severe_loss_rate"]) * 3)
-            down_safety = _clamp(100 - float(stats["previous_down_count"]) / stats["previous_limit_count"] * 700)
-            tail_safety_score = severe_loss_safety * 0.70 + down_safety * 0.30
-            profit_effect_score = (
-                positive_score * 0.30
-                + median_change_score * 0.25
-                + average_change_score * 0.10
-                + advance_score * 0.20
-                + tail_safety_score * 0.15
-            )
-        else:
-            profit_effect_score = 50.0
-
-        max_height_score = _adaptive_score(
-            float(stats["max_height"]),
-            _linear(float(stats["max_height"]), 1, 7),
-            height_history,
-        )
-        continuation_rate = (
-            (float(stats["second_board_count"]) + float(stats["three_plus_count"]))
-            / max(float(stats["limit_up_count"]), 1)
-            * 100
-        )
-        three_plus_density = float(stats["three_plus_count"]) / max(float(stats["limit_up_count"]), 1) * 100
-        three_plus_score = _adaptive_score(
-            float(stats["three_plus_count"]),
-            _clamp(three_plus_density * 5),
-            three_plus_history,
-        )
-        ladder_structure_score = (
-            max_height_score * 0.30
-            + _clamp(continuation_rate * 3) * 0.25
-            + three_plus_score * 0.25
-            + float(stats["ladder_completeness"]) * 0.20
-        )
-
-        amount_baseline = mean(amount_history) if amount_history else float(stats["amount_billion"] or 1)
-        amount_ratio = float(stats["amount_billion"]) / max(amount_baseline, 1)
-        amount_score = _clamp(50 + (amount_ratio - 1) * 100)
-        limit_amount_share = float(stats["limit_amount_billion"]) / max(float(stats["amount_billion"]), 1) * 100
-        liquidity_score = amount_score * 0.70 + _clamp(limit_amount_share * 20) * 0.30
-
-        component_scores = {
-            "breadth": breadth_score,
-            "limit_ecology": limit_ecology_score,
-            "profit_effect": profit_effect_score,
-            "ladder_structure": ladder_structure_score,
-            "liquidity": liquidity_score,
-        }
-        raw_score = sum(component_scores[key] * weight / 100 for key, weight in COMPONENT_WEIGHTS.items())
-        score = round(
-            raw_score * systemic_gate
-        )
-        extreme_ice = float(stats["breadth_ratio"]) <= 15 and float(stats["limit_down_count"]) >= 100
-        if extreme_ice:
-            score = min(score, 15)
-        elif float(stats["breadth_ratio"]) <= 25 and float(stats["limit_down_count"]) >= 50:
-            score = min(score, 24)
-        previous_scores: list[float] = []
-        expected_date = str(stats.get("previous_trade_date") or "")
-        for prior_result in reversed(results):
-            if not expected_date or str(prior_result.get("trade_date") or "") != expected_date:
-                break
-            previous_scores.append(float(prior_result["score"]))
-            expected_date = str(prior_result.get("previous_trade_date") or "")
-            if len(previous_scores) == 3:
-                break
-        momentum = score - mean(previous_scores) if previous_scores else 0.0
-        direction = "升温" if momentum > 3 else "降温" if momentum < -3 else "持平"
-        normalization = "历史百分位" if len(previous) >= 20 else "固定锚点"
-        previous_result = (
-            results[-1]
-            if results and str(stats.get("previous_trade_date") or "") == str(results[-1].get("trade_date") or "")
-            else None
-        )
-        day_change = score - float(previous_result["score"]) if previous_result else 0.0
-        ema_score = round(
-            score if not previous_result
-            else score * 0.5 + float(previous_result.get("ema_score", previous_result["score"])) * 0.5,
-            1,
-        )
-        phase_signal = _phase_signal(score, momentum, profit_effect_score)
-        fermentation_ready = (
-            phase_signal == "发酵"
-            and score >= 45
-            and profit_effect_score >= 45
-            and systemic_health >= 35
-            and not extreme_ice
-        )
-        previous_fermentation_count = int(previous_result.get("fermentation_signal_count") or 0) if previous_result else 0
-        fermentation_signal_count = previous_fermentation_count + 1 if fermentation_ready else 0
-        phase, transition_reason = _confirmed_phase(
-            previous_result,
-            score,
-            day_change,
-            systemic_health,
-            profit_effect_score,
-            limit_ecology_score,
-            phase_signal,
-            extreme_ice,
-            fermentation_signal_count,
-        )
-        previous_phase = str(previous_result.get("phase") or "") if previous_result else ""
-        if phase not in {"修复", "分化"}:
-            fermentation_signal_count = 0
-        elif phase == "分化" and previous_phase != "分化":
-            fermentation_signal_count = 0
-
-        components = {
-            "breadth": {
-                "label": "市场宽度",
-                "score": round(breadth_score, 1),
-                "weight": COMPONENT_WEIGHTS["breadth"],
-                "summary": f"上涨占比 {stats['breadth_ratio']:.1f}%",
-            },
-            "limit_ecology": {
-                "label": "涨停生态",
-                "score": round(limit_ecology_score, 1),
-                "weight": COMPONENT_WEIGHTS["limit_ecology"],
-                "summary": (
-                    f"涨停 {stats['limit_up_count']} · 跌停 {stats['limit_down_count']} · "
-                    f"封板 {stats['seal_rate']:.1f}%"
-                ),
-            },
-            "profit_effect": {
-                "label": "赚钱效应",
-                "score": round(profit_effect_score, 1),
-                "weight": COMPONENT_WEIGHTS["profit_effect"],
-                "summary": (
-                    f"昨涨停红盘 {stats['previous_positive_rate']:.1f}% · "
-                    f"中位 {stats['median_previous_change']:+.2f}% · "
-                    f"重亏 {stats['severe_loss_rate']:.1f}%"
-                    if stats["previous_limit_count"] else "缺少前一交易日样本"
-                ),
-            },
-            "ladder_structure": {
-                "label": "连板结构",
-                "score": round(ladder_structure_score, 1),
-                "weight": COMPONENT_WEIGHTS["ladder_structure"],
-                "summary": f"最高 {stats['max_height']} 板 · 三板以上 {stats['three_plus_count']} 家",
-            },
-            "liquidity": {
-                "label": "成交活跃度",
-                "score": round(liquidity_score, 1),
-                "weight": COMPONENT_WEIGHTS["liquidity"],
-                "summary": f"成交 {stats['amount_billion']:.1f} 亿 · 均值比 {amount_ratio:.2f}",
-            },
-        }
-        results.append(
-            {
-                **stats,
-                "score": score,
-                "ema_score": ema_score,
-                "label": _sentiment_label(score),
-                "phase": phase,
-                "phase_signal": phase_signal,
-                "transition_reason": transition_reason,
-                "fermentation_signal_count": fermentation_signal_count,
-                "day_change": round(day_change, 1),
-                "direction": direction,
-                "momentum": round(momentum, 1),
-                "normalization": "250日历史百分位" if len(previous) >= 20 else normalization,
-                "history_days": len(previous) + 1,
-                "systemic_health": round(systemic_health, 1),
-                "risk_multiplier": round(systemic_gate, 3),
-                "components": components,
-            }
-        )
-    return results
-
-
-def latest_contiguous_history(series: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    if not series:
-        return []
-    contiguous = [series[-1]]
-    for row in reversed(series[:-1]):
-        expected_previous = str(contiguous[0].get("previous_trade_date") or "")
-        if not expected_previous or expected_previous != str(row.get("trade_date") or ""):
-            break
-        contiguous.insert(0, row)
-    return contiguous
-
-
-def apply_sentiment_to_dashboard(
-    dashboard: dict[str, Any],
-    historical_snapshots: list[dict[str, Any]] | None = None,
-) -> dict[str, Any]:
-    result = deepcopy(dashboard)
-    history = list(historical_snapshots or [])
-    history.append(result)
-    series = build_sentiment_history(history)
-    target_date = _trade_date(result)
-    sentiment = next((row for row in reversed(series) if row["trade_date"] == target_date), None)
-    if not sentiment:
-        return result
-    overview = dict(result.get("overview") or {})
-    overview.update(
-        {
-            "sentiment_score": sentiment["score"],
-            "sentiment_trend_score": sentiment["ema_score"],
-            "sentiment_label": sentiment["label"],
-            "sentiment_phase": sentiment["phase"],
-            "sentiment_direction": sentiment["direction"],
-            "sentiment_components": sentiment["components"],
-            "sentiment_engine_version": SENTIMENT_ENGINE_VERSION,
-        }
-    )
-    result["overview"] = overview
-    return result

tests/test_preservation_slice_market.py

@@ -127,12 +127,15 @@ class MarketSliceSourceEquivalenceTests(unittest.TestCase):
 
     def test_provider_logic_is_the_original_implementation(self) -> None:
         exact_moves = (
-            ("tushare_client.py", "backend/data/providers/tushare_client.py"),
             ("ifind_client.py", "backend/data/providers/ifind_client.py"),
             ("realtime_aggregator.py", "backend/data/realtime.py"),
         )
         for original, migrated in exact_moves:
             self.assertEqual(sha256(ORIGINAL_ROOT / original), sha256(APP_ROOT / migrated))
+        self.assertEqual(
+            top_level_definitions(ORIGINAL_ROOT / "tushare_client.py"),
+            top_level_definitions(APP_ROOT / "backend/data/providers/tushare_client.py"),
+        )
         self.assertEqual(
             top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py"),
             top_level_definitions(APP_ROOT / "backend/features/market/charts.py"),

tests/test_preservation_slice_sentiment_pools.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import ast
+import hashlib
+import unittest
+from pathlib import Path
+
+import sentiment_engine
+from backend.features.sentiment import engine as canonical_engine
+
+
+APP_ROOT = Path(__file__).resolve().parents[1]
+ORIGINAL_ROOT = APP_ROOT.parent
+
+SENTIMENT_METHODS = {
+    "_enrich_dashboard_sentiment",
+    "sentiment_history",
+}
+POOL_METHODS = {
+    "save_reason",
+    "_apply_reason_overrides",
+    "_schedule_ifind_event_enrichment",
+    "_refresh_ifind_event_enrichment",
+    "_normalize_ifind_event_time",
+    "_merge_ifind_event_enrichment",
+}
+POOL_REPOSITORY_METHODS = {
+    "save_reason_override",
+    "reason_overrides",
+}
+
+
+def class_methods(path: Path, class_name: str) -> dict[str, str]:
+    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    owner = next(
+        node
+        for node in tree.body
+        if isinstance(node, ast.ClassDef) and node.name == class_name
+    )
+    return {
+        node.name: ast.dump(node, include_attributes=False)
+        for node in owner.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+    }
+
+
+def sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+class SentimentPoolSliceSourceEquivalenceTests(unittest.TestCase):
+    def test_sentiment_service_methods_are_exact_original_ast(self) -> None:
+        original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
+        migrated = class_methods(
+            APP_ROOT / "backend" / "features" / "sentiment" / "service.py",
+            "SentimentServiceMixin",
+        )
+        self.assertEqual(set(migrated), SENTIMENT_METHODS)
+        for name in sorted(SENTIMENT_METHODS):
+            self.assertEqual(migrated[name], original[name], name)
+
+    def test_pool_service_methods_are_exact_original_ast(self) -> None:
+        original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
+        migrated = class_methods(
+            APP_ROOT / "backend" / "features" / "pools" / "service.py",
+            "PoolServiceMixin",
+        )
+        self.assertEqual(set(migrated), POOL_METHODS)
+        for name in sorted(POOL_METHODS):
+            self.assertEqual(migrated[name], original[name], name)
+
+    def test_pool_repository_methods_are_exact_original_ast(self) -> None:
+        original = class_methods(ORIGINAL_ROOT / "database.py", "ReviewDatabase")
+        migrated = class_methods(
+            APP_ROOT / "backend" / "features" / "pools" / "repository.py",
+            "PoolRepositoryMixin",
+        )
+        self.assertEqual(set(migrated), POOL_REPOSITORY_METHODS)
+        for name in sorted(POOL_REPOSITORY_METHODS):
+            self.assertEqual(migrated[name], original[name], name)
+
+    def test_original_classes_no_longer_duplicate_moved_methods(self) -> None:
+        remaining_service = class_methods(
+            APP_ROOT / "backend" / "application.py", "DashboardService"
+        )
+        remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase")
+        self.assertTrue((SENTIMENT_METHODS | POOL_METHODS).isdisjoint(remaining_service))
+        self.assertTrue(POOL_REPOSITORY_METHODS.isdisjoint(remaining_database))
+
+    def test_sentiment_engine_is_exact_original_with_legacy_alias(self) -> None:
+        self.assertEqual(
+            sha256(ORIGINAL_ROOT / "sentiment_engine.py"),
+            sha256(APP_ROOT / "backend" / "features" / "sentiment" / "engine.py"),
+        )
+        self.assertIs(sentiment_engine, canonical_engine)
+
+    def test_api_and_frontend_assets_are_unchanged(self) -> None:
+        for relative in (
+            "config/api.config.json",
+            "static/index.html",
+            "static/app.js",
+            "static/styles.css",
+            "static/pages/sentiment/page.js",
+            "static/pages/pools/page.js",
+        ):
+            self.assertEqual(
+                sha256(APP_ROOT / relative),
+                sha256(ORIGINAL_ROOT / relative),
+                relative,
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()