migration: preserve market insights slice

app/backend/application.py

@@ -42,7 +42,6 @@ from heaven_engine import (
 from backend.data.providers.ifind_client import IfindError
 from llm_strategy import LLMCompilerError, compile_strategy_with_llm, test_llm_connection
 from mentor_agent import MentorAgentError, stream_with_mentor
-from market_insights import MarketInsightsService
 from screener import (
     FACTOR_FIELDS,
     FACTOR_GROUPS,
@@ -53,13 +52,14 @@ 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.auction import AuctionServiceMixin
+from backend.features.dragon_tiger import DragonTigerServiceMixin
 from backend.features.pools import PoolServiceMixin
+from backend.features.popularity import PopularityServiceMixin
+from backend.features.rotation import RotationServiceMixin
 from backend.features.sentiment import SentimentServiceMixin
-from backend.features.sentiment.engine import (
-    build_sentiment_history,
-    latest_contiguous_history,
-)
 from backend.features.system import SystemHttpMixin
+from backend.features.themes import ThemeServiceMixin
 from backend.data.providers.tushare_client import TushareClient, TushareError, _sector_coverage_issue
 
 
@@ -140,7 +140,16 @@ MENTOR_ETF_UNIVERSE = (
 )
 
 
-class DashboardService(MarketServiceMixin, SentimentServiceMixin, PoolServiceMixin):
+class DashboardService(
+    MarketServiceMixin,
+    SentimentServiceMixin,
+    PoolServiceMixin,
+    RotationServiceMixin,
+    AuctionServiceMixin,
+    ThemeServiceMixin,
+    PopularityServiceMixin,
+    DragonTigerServiceMixin,
+):
     def __init__(self) -> None:
         runtime = load_runtime_settings()
         self.vault = SecretVault(runtime.encryption_key)
@@ -734,158 +743,6 @@ class DashboardService(MarketServiceMixin, SentimentServiceMixin, PoolServiceMix
 
 
 
-    def rotation_history(self, trade_date: str, limit: int = 9) -> dict[str, Any]:
-        normalized_date = normalize_date(trade_date)
-        # 板块轮动固定展示最近 9 个交易日，按由近到远排列。
-        limit = 9
-        snapshots = self.database.list_snapshot_payloads(normalized_date, 240)
-        by_trade_date: dict[str, dict[str, Any]] = {}
-        for snapshot in snapshots:
-            meta = snapshot.get("meta") or {}
-            actual_date = str(meta.get("trade_date") or snapshot.get("_snapshot_date") or "")
-            compact_date = actual_date.replace("-", "")
-            if len(compact_date) == 8:
-                by_trade_date[compact_date] = snapshot
-
-        sentiment_dates = {
-            str(row.get("trade_date") or "").replace("-", "")
-            for row in latest_contiguous_history(build_sentiment_history(snapshots))
-        }
-        ordered_dates = sorted(
-            date_key for date_key in by_trade_date
-            if not sentiment_dates or date_key in sentiment_dates
-        )[-limit:][::-1]
-        rows = []
-        for date_key in ordered_dates:
-            snapshot = by_trade_date[date_key]
-            sector_context = {
-                str(item.get("name") or ""): item
-                for item in snapshot.get("sectors") or []
-            }
-            sectors = []
-            for item in (snapshot.get("sector_rotation") or [])[:12]:
-                name = str(item.get("name") or "").strip()
-                context = sector_context.get(name, {})
-                sectors.append(
-                    {
-                        "name": name,
-                        "rank": int(item.get("rank") or len(sectors) + 1),
-                        "trend": item.get("trend") or "持平",
-                        "count": int(item.get("count") or 0),
-                        "strength": float(item.get("strength") or context.get("strength") or 0),
-                        "change": float(context.get("change") or 0),
-                        "leader": item.get("leader") or context.get("leader") or "--",
-                    }
-                )
-            rows.append(
-                {
-                    "trade_date": f"{date_key[:4]}-{date_key[4:6]}-{date_key[6:]}",
-                    "sectors": sectors,
-                }
-            )
-        return {
-            "trade_date": rows[0]["trade_date"] if rows else normalized_date,
-            "available_days": len(ordered_dates),
-            "requested_days": limit,
-            "rows": rows,
-        }
-
-    def rotation_sector_members(self, trade_date: str, sector_name: str) -> dict[str, Any]:
-        normalized_date = normalize_date(trade_date)
-        sector_name = validate_text(sector_name, "板块名称", 60, required=True)
-        dashboard = self.get_dashboard(normalized_date)
-        actual_date = normalize_date(
-            str((dashboard.get("meta") or {}).get("trade_date") or normalized_date)
-        )
-        cache_key = f"{actual_date}:{sector_name}"
-        cached = self.database.get_data_snapshot("rotation_sector_members_v1", cache_key)
-        if cached:
-            cached["meta"] = {**(cached.get("meta") or {}), "cached": True}
-            return cached
-        if not self.configured:
-            raise ValueError("板块成分数据暂不可用。")
-
-        representative = next(
-            (
-                item for item in dashboard.get("limits") or []
-                if str(item.get("sector") or "").strip() == sector_name
-            ),
-            None,
-        )
-        if not representative:
-            raise ValueError("未找到该板块的代表股票，暂时无法核验成分股。")
-        raw_code = str(representative.get("ts_code") or representative.get("code") or "")
-        if "." in raw_code:
-            ts_code = raw_code
-        elif raw_code.startswith(("4", "8", "92")):
-            ts_code = f"{raw_code}.BJ"
-        elif raw_code.startswith(("6", "68", "90")):
-            ts_code = f"{raw_code}.SH"
-        else:
-            ts_code = f"{raw_code}.SZ"
-        client = self._tushare_client()
-        try:
-            industry = client.sw_stock_industry(ts_code, actual_date)
-            sector_code = str(industry.get("l2_code") or "")
-            members = client.sw_sector_members(sector_code, actual_date)
-        except TushareError as exc:
-            raise ValueError(f"该板块成分股暂不可用：{exc}") from exc
-
-        daily_rows = self.database.daily_bars_for_date(actual_date)
-        if len(daily_rows) < 1000:
-            try:
-                daily_rows = client.query(
-                    "daily",
-                    {"trade_date": actual_date},
-                    "ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
-                )
-                if daily_rows:
-                    self.database.upsert_daily_bars(daily_rows)
-            except TushareError:
-                daily_rows = self.database.daily_bars_for_date(actual_date)
-        daily_map = {str(item.get("ts_code") or ""): item for item in daily_rows}
-        rows = []
-        for member in members:
-            member_code = str(member.get("ts_code") or "")
-            quote = daily_map.get(member_code) or {}
-            rows.append(
-                {
-                    "code": member_code.split(".")[0],
-                    "ts_code": member_code,
-                    "name": str(member.get("name") or "--"),
-                    "change": quote.get("pct_chg"),
-                    "open": quote.get("open"),
-                    "close": quote.get("close"),
-                    "amount_billion": (
-                        round(float(quote.get("amount") or 0) / 100000, 2)
-                        if quote else None
-                    ),
-                    "quoted": bool(quote),
-                }
-            )
-        rows.sort(
-            key=lambda item: (
-                bool(item.get("quoted")),
-                float(item.get("change") or -999),
-                float(item.get("amount_billion") or 0),
-            ),
-            reverse=True,
-        )
-        result = {
-            "meta": {
-                "trade_date": self._display_compact_date(actual_date),
-                "sector_name": str(industry.get("l2_name") or sector_name),
-                "sector_code": sector_code,
-                "member_count": len(rows),
-                "quoted_count": sum(bool(item.get("quoted")) for item in rows),
-                "cached": False,
-            },
-            "rows": rows,
-        }
-        self.database.save_data_snapshot(
-            "rotation_sector_members_v1", cache_key, "tushare", result
-        )
-        return result
 
     def status(self) -> dict[str, Any]:
         llm_access = self.llm_access_status()
@@ -903,28 +760,9 @@ class DashboardService(MarketServiceMixin, SentimentServiceMixin, PoolServiceMix
         }
 
 
-    def _market_insights(self) -> MarketInsightsService:
-        if not self.configured:
-            raise ValueError("行情数据尚未配置。")
-        return MarketInsightsService(
-            self.database,
-            self._tushare_client(),
-            ifind=self.ifind,
-        )
 
-    def auction_center(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        return self._market_insights().auction_center(
-            normalize_date(trade_date), force, self.current_user_id
-        )
 
-    def theme_library(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        return self._market_insights().theme_library(normalize_date(trade_date), force)
 
-    def theme_detail(self, code: str, trade_date: str) -> dict[str, Any]:
-        return self._market_insights().theme_detail(code, normalize_date(trade_date))
-
-    def popularity(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        return self._market_insights().popularity(normalize_date(trade_date), force)
 
     @staticmethod
     def _ifind_field(row: dict[str, Any], tokens: tuple[str, ...]) -> Any:
@@ -3224,287 +3062,6 @@ class DashboardService(MarketServiceMixin, SentimentServiceMixin, PoolServiceMix
         )
         return result
 
-    def get_hot_money_profiles(self, force: bool = False) -> dict[str, Any]:
-        cache_kind = "hot_money_profiles_v1"
-        cache_key = "directory"
-        cached = self.database.get_data_snapshot(cache_kind, cache_key)
-        if cached and not force:
-            cached["meta"] = {**cached.get("meta", {}), "cached": True}
-            return cached
-        if self.configured:
-            try:
-                payload = self._tushare_client().hot_money_profiles()
-            except TushareError:
-                if cached:
-                    cached["meta"] = {
-                        **cached.get("meta", {}),
-                        "cached": True,
-                        "stale": True,
-                        "notice": "名录暂未完成更新，当前展示最近一次收录结果。",
-                    }
-                    return cached
-                return {
-                    "meta": {
-                        "source": "unavailable",
-                        "status": "unavailable",
-                        "schema_version": 1,
-                        "cached": False,
-                        "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
-                        "notice": "游资名录暂不可用，请稍后重试。",
-                    },
-                    "summary": {
-                        "profile_count": 0,
-                        "described_count": 0,
-                        "organization_count": 0,
-                    },
-                    "profiles": [],
-                }
-            payload["meta"]["cached"] = False
-            if payload.get("meta", {}).get("status") == "success":
-                self.database.save_data_snapshot(cache_kind, cache_key, "tushare", payload)
-            return payload
-        if cached:
-            cached["meta"] = {**cached.get("meta", {}), "cached": True}
-            return cached
-        return {
-            "meta": {
-                "source": "unavailable",
-                "status": "unavailable",
-                "schema_version": 1,
-                "cached": False,
-                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
-                "notice": "游资名录暂不可用，请联系管理员检查行情配置。",
-            },
-            "summary": {
-                "profile_count": 0,
-                "described_count": 0,
-                "organization_count": 0,
-            },
-            "profiles": [],
-        }
-
-    def get_dragon_tiger(self, trade_date: str, force: bool = False) -> dict[str, Any]:
-        normalized_date = normalize_date(trade_date)
-        cache_kind = "hot_money_detail_v3"
-        if not force:
-            cached = self.database.get_data_snapshot(cache_kind, normalized_date)
-            if (
-                cached
-                and cached.get("meta", {}).get("source") == "tushare"
-                and cached.get("meta", {}).get("status") == "success"
-                and int(cached.get("meta", {}).get("schema_version") or 0) == 3
-            ):
-                cached["meta"] = {**cached.get("meta", {}), "cached": True}
-                return cached
-        if self.configured:
-            try:
-                payload = self._tushare_client().dragon_tiger(normalized_date)
-            except TushareError as exc:
-                return {
-                    "meta": {
-                        "requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
-                        "trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
-                        "source": "tushare_error",
-                        "status": "error",
-                        "schema_version": 3,
-                        "cached": False,
-                        "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
-                        "notice": "龙虎榜数据暂不可用，请稍后重试。",
-                    },
-                    "summary": {
-                        "trader_count": 0,
-                        "identity_count": 0,
-                        "operation_count": 0,
-                        "active_stock_count": 0,
-                        "seat_net_buy_million": 0,
-                        "unclassified_count": 0,
-                        "directory_count": 0,
-                    },
-                    "traders": [],
-                    "unclassified_seats": [],
-                    "rows": [],
-                }
-            payload["meta"]["cached"] = False
-            if payload.get("meta", {}).get("status") == "success":
-                self.database.save_data_snapshot(cache_kind, normalized_date, "tushare", payload)
-            return payload
-
-        return {
-            "meta": {
-            "requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
-            "trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
-            "source": "unavailable",
-            "status": "unavailable",
-            "schema_version": 3,
-            "cached": False,
-            "notice": "龙虎榜数据暂不可用，请联系管理员检查行情配置。",
-            },
-            "summary": {
-                "trader_count": 0,
-                "identity_count": 0,
-                "operation_count": 0,
-                "active_stock_count": 0,
-                "seat_net_buy_million": 0,
-                "unclassified_count": 0,
-                "directory_count": 0,
-            },
-            "traders": [],
-            "unclassified_seats": [],
-            "rows": [],
-        }
-
-
-
-    def _apply_seat_aliases(self, payload: dict[str, Any]) -> dict[str, Any]:
-        aliases = self.database.list_seat_aliases()
-        result = dict(payload)
-        rows = payload.get("rows") or []
-        for row in rows:
-            for institution in row.get("institutions") or []:
-                institution["alias"] = aliases.get(institution.get("seat_name", ""), "")
-        traders: dict[tuple[str, str], dict[str, Any]] = {}
-        unclassified: dict[str, dict[str, Any]] = {}
-        seen_operations: set[tuple[Any, ...]] = set()
-        builtin_aliases = {
-            "国泰海通证券股份有限公司南京太平南路证券营业部": "作手新一",
-        }
-
-        for row in rows:
-            for institution in row.get("institutions") or []:
-                seat_name = str(institution.get("seat_name") or "未知席位").strip()
-                saved_alias = str(institution.get("alias") or "").strip()
-                builtin_alias = builtin_aliases.get(seat_name, "")
-                if saved_alias or builtin_alias:
-                    identity_name = saved_alias or builtin_alias
-                    identity_type = "trader"
-                    recognized = True
-                    identity_source = "manual" if saved_alias else "builtin"
-                elif "机构专用" in seat_name:
-                    identity_name = "机构专用"
-                    identity_type = "institution"
-                    recognized = True
-                    identity_source = "system"
-                elif "沪股通专用" in seat_name or "深股通专用" in seat_name:
-                    identity_name = "北向资金"
-                    identity_type = "channel"
-                    recognized = True
-                    identity_source = "system"
-                else:
-                    identity_name = seat_name
-                    identity_type = "unclassified"
-                    recognized = False
-                    identity_source = "raw"
-
-                buy = round(float(institution.get("buy_million") or 0), 2)
-                sell = round(float(institution.get("sell_million") or 0), 2)
-                net_buy = round(float(institution.get("net_buy_million") or 0), 2)
-                operation_key = (row.get("code"), seat_name, buy, sell, net_buy)
-                if operation_key in seen_operations:
-                    continue
-                seen_operations.add(operation_key)
-
-                group_key = (identity_type, identity_name)
-                group = traders.setdefault(
-                    group_key,
-                    {
-                        "name": identity_name,
-                        "identity_type": identity_type,
-                        "identity_source": identity_source,
-                        "recognized": recognized,
-                        "buy_million": 0.0,
-                        "sell_million": 0.0,
-                        "net_buy_million": 0.0,
-                        "seat_names": set(),
-                        "stock_codes": set(),
-                        "operations": [],
-                    },
-                )
-                group["buy_million"] += buy
-                group["sell_million"] += sell
-                group["net_buy_million"] += net_buy
-                group["seat_names"].add(seat_name)
-                group["stock_codes"].add(str(row.get("code") or ""))
-                group["operations"].append(
-                    {
-                        "code": row.get("code") or "",
-                        "name": row.get("name") or "--",
-                        "change": row.get("change") or 0,
-                        "direction": "买入" if net_buy > 0 else "卖出" if net_buy < 0 else "持平",
-                        "buy_million": buy,
-                        "sell_million": sell,
-                        "net_buy_million": net_buy,
-                        "reason": row.get("reason") or "--",
-                        "seat_name": seat_name,
-                        "seat_alias": identity_name if recognized else "",
-                    }
-                )
-
-                if not recognized:
-                    pending = unclassified.setdefault(
-                        seat_name,
-                        {
-                            "seat_name": seat_name,
-                            "stock_codes": set(),
-                            "operation_count": 0,
-                            "buy_million": 0.0,
-                            "sell_million": 0.0,
-                            "net_buy_million": 0.0,
-                        },
-                    )
-                    pending["stock_codes"].add(str(row.get("code") or ""))
-                    pending["operation_count"] += 1
-                    pending["buy_million"] += buy
-                    pending["sell_million"] += sell
-                    pending["net_buy_million"] += net_buy
-
-        type_order = {"trader": 0, "institution": 1, "channel": 2, "unclassified": 3}
-        aggregated = list(traders.values())
-        aggregated.sort(
-            key=lambda item: (
-                type_order.get(item["identity_type"], 9),
-                -abs(item["net_buy_million"]),
-                item["name"],
-            )
-        )
-        for index, group in enumerate(aggregated, start=1):
-            group["id"] = f"identity-{index}"
-            group["buy_million"] = round(group["buy_million"], 2)
-            group["sell_million"] = round(group["sell_million"], 2)
-            group["net_buy_million"] = round(group["net_buy_million"], 2)
-            group["seat_count"] = len(group.pop("seat_names"))
-            group["stock_count"] = len(group.pop("stock_codes"))
-            group["operation_count"] = len(group["operations"])
-            group["operations"].sort(
-                key=lambda item: abs(float(item.get("net_buy_million") or 0)), reverse=True
-            )
-
-        pending_seats = list(unclassified.values())
-        for pending in pending_seats:
-            pending["stock_count"] = len(pending.pop("stock_codes"))
-            pending["buy_million"] = round(pending["buy_million"], 2)
-            pending["sell_million"] = round(pending["sell_million"], 2)
-            pending["net_buy_million"] = round(pending["net_buy_million"], 2)
-        pending_seats.sort(key=lambda item: abs(item["net_buy_million"]), reverse=True)
-
-        operation_count = sum(item["operation_count"] for item in aggregated)
-        active_stocks = {
-            operation["code"] for item in aggregated for operation in item["operations"]
-        }
-        seat_net_buy = round(sum(item["net_buy_million"] for item in aggregated), 2)
-        result["rows"] = rows
-        result["traders"] = aggregated
-        result["unclassified_seats"] = pending_seats
-        result["summary"] = {
-            **(payload.get("summary") or {}),
-            "trader_count": sum(item["identity_type"] == "trader" for item in aggregated),
-            "identity_count": len(aggregated),
-            "operation_count": operation_count,
-            "active_stock_count": len(active_stocks),
-            "seat_net_buy_million": seat_net_buy,
-            "unclassified_count": len(pending_seats),
-        }
-        return result
-
 
 SERVICE = DashboardService()
 

app/backend/features/auction/__init__.py

@@ -0,0 +1,4 @@
+from .repository import AuctionRepositoryMixin
+from .service import AuctionServiceMixin
+
+__all__ = ["AuctionRepositoryMixin", "AuctionServiceMixin"]

app/backend/features/auction/repository.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class AuctionRepositoryMixin:
+    def upsert_auction_factors(self, rows: list[dict[str, Any]]) -> int:
+        values = []
+        for row in rows:
+            trade_date = str(row.get("trade_date") or "")
+            ts_code = str(row.get("ts_code") or "")
+            price = float(row.get("price") or 0)
+            pre_close = float(row.get("pre_close") or 0)
+            if not trade_date or not ts_code or price <= 0 or pre_close <= 0:
+                continue
+            values.append(
+                (
+                    trade_date,
+                    ts_code,
+                    price,
+                    pre_close,
+                    (price / pre_close - 1) * 100,
+                    float(row.get("vol") or 0),
+                    float(row.get("amount") or 0),
+                    float(row.get("turnover_rate") or 0),
+                    float(row.get("volume_ratio") or 0),
+                )
+            )
+        with self.connect() as connection:
+            connection.executemany(
+                """
+                INSERT INTO auction_factors
+                    (trade_date, ts_code, price, pre_close, change, vol, amount,
+                     turnover_rate, volume_ratio)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
+                    price=excluded.price, pre_close=excluded.pre_close,
+                    change=excluded.change, vol=excluded.vol, amount=excluded.amount,
+                    turnover_rate=excluded.turnover_rate,
+                    volume_ratio=excluded.volume_ratio
+                """,
+                values,
+            )
+        return len(values)
+
+    def auction_factor_dates(self, end_date: str = "", limit: int = 80) -> list[str]:
+        where = "WHERE trade_date <= ?" if end_date else ""
+        parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
+        with self.connect() as connection:
+            rows = connection.execute(
+                f"SELECT DISTINCT trade_date FROM auction_factors {where} "
+                "ORDER BY trade_date DESC LIMIT ?",
+                parameters,
+            ).fetchall()
+        return [row["trade_date"] for row in reversed(rows)]
+
+    def auction_factors_for_date(self, trade_date: str) -> list[dict[str, Any]]:
+        with self.connect() as connection:
+            rows = connection.execute(
+                "SELECT * FROM auction_factors WHERE trade_date = ? ORDER BY ts_code",
+                (trade_date,),
+            ).fetchall()
+        return [dict(row) for row in rows]

app/backend/features/auction/service.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.bootstrap.config import normalize_date
+from backend.features.market.insights import MarketInsightsService
+
+
+class AuctionServiceMixin:
+    def auction_center(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self._market_insights().auction_center(
+            normalize_date(trade_date), force, self.current_user_id
+        )

app/backend/features/dragon_tiger/__init__.py

@@ -0,0 +1,4 @@
+from .repository import DragonTigerRepositoryMixin
+from .service import DragonTigerServiceMixin
+
+__all__ = ["DragonTigerRepositoryMixin", "DragonTigerServiceMixin"]

app/backend/features/dragon_tiger/repository.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+
+class DragonTigerRepositoryMixin:
+    def list_seat_aliases(self) -> dict[str, str]:
+        with self.connect() as connection:
+            rows = connection.execute("SELECT seat_name, alias FROM seat_aliases").fetchall()
+        return {row["seat_name"]: row["alias"] for row in rows}
+
+    def save_seat_alias(self, seat_name: str, alias: str) -> None:
+        now = datetime.now().astimezone().isoformat(timespec="seconds")
+        with self.connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO seat_aliases (seat_name, alias, updated_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(seat_name) DO UPDATE SET
+                    alias = excluded.alias,
+                    updated_at = excluded.updated_at
+                """,
+                (seat_name, alias, now),
+            )
+
+    def upsert_lhb_institutions(self, rows: list[dict[str, Any]]) -> int:
+        grouped: dict[tuple[str, str], dict[str, float | int]] = {}
+        for row in rows:
+            trade_date = str(row.get("trade_date") or "")
+            ts_code = str(row.get("ts_code") or "")
+            seat_name = str(row.get("exalter") or row.get("seat_name") or "")
+            if not trade_date or not ts_code or "机构专用" not in seat_name:
+                continue
+            group = grouped.setdefault(
+                (trade_date, ts_code),
+                {"net": 0.0, "buy": 0.0, "sell": 0.0, "seats": 0},
+            )
+            group["net"] = float(group["net"]) + float(row.get("net_buy") or row.get("net_amount") or 0)
+            group["buy"] = float(group["buy"]) + float(row.get("buy") or row.get("buy_amount") or 0)
+            group["sell"] = float(group["sell"]) + float(row.get("sell") or row.get("sell_amount") or 0)
+            group["seats"] = int(group["seats"]) + 1
+        values = [
+            (trade_date, ts_code, item["net"], item["buy"], item["sell"], item["seats"])
+            for (trade_date, ts_code), item in grouped.items()
+        ]
+        with self.connect() as connection:
+            connection.executemany(
+                """
+                INSERT INTO lhb_institution_daily
+                    (trade_date, ts_code, net_buy_amount, buy_amount, sell_amount, seat_count)
+                VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
+                    net_buy_amount=excluded.net_buy_amount,
+                    buy_amount=excluded.buy_amount,
+                    sell_amount=excluded.sell_amount,
+                    seat_count=excluded.seat_count
+                """,
+                values,
+            )
+        return len(values)

app/backend/features/dragon_tiger/service.py

@@ -0,0 +1,288 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from backend.bootstrap.config import normalize_date
+from backend.data.providers.tushare_client import TushareError
+
+
+class DragonTigerServiceMixin:
+    def get_hot_money_profiles(self, force: bool = False) -> dict[str, Any]:
+        cache_kind = "hot_money_profiles_v1"
+        cache_key = "directory"
+        cached = self.database.get_data_snapshot(cache_kind, cache_key)
+        if cached and not force:
+            cached["meta"] = {**cached.get("meta", {}), "cached": True}
+            return cached
+        if self.configured:
+            try:
+                payload = self._tushare_client().hot_money_profiles()
+            except TushareError:
+                if cached:
+                    cached["meta"] = {
+                        **cached.get("meta", {}),
+                        "cached": True,
+                        "stale": True,
+                        "notice": "名录暂未完成更新，当前展示最近一次收录结果。",
+                    }
+                    return cached
+                return {
+                    "meta": {
+                        "source": "unavailable",
+                        "status": "unavailable",
+                        "schema_version": 1,
+                        "cached": False,
+                        "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+                        "notice": "游资名录暂不可用，请稍后重试。",
+                    },
+                    "summary": {
+                        "profile_count": 0,
+                        "described_count": 0,
+                        "organization_count": 0,
+                    },
+                    "profiles": [],
+                }
+            payload["meta"]["cached"] = False
+            if payload.get("meta", {}).get("status") == "success":
+                self.database.save_data_snapshot(cache_kind, cache_key, "tushare", payload)
+            return payload
+        if cached:
+            cached["meta"] = {**cached.get("meta", {}), "cached": True}
+            return cached
+        return {
+            "meta": {
+                "source": "unavailable",
+                "status": "unavailable",
+                "schema_version": 1,
+                "cached": False,
+                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+                "notice": "游资名录暂不可用，请联系管理员检查行情配置。",
+            },
+            "summary": {
+                "profile_count": 0,
+                "described_count": 0,
+                "organization_count": 0,
+            },
+            "profiles": [],
+        }
+
+    def get_dragon_tiger(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        normalized_date = normalize_date(trade_date)
+        cache_kind = "hot_money_detail_v3"
+        if not force:
+            cached = self.database.get_data_snapshot(cache_kind, normalized_date)
+            if (
+                cached
+                and cached.get("meta", {}).get("source") == "tushare"
+                and cached.get("meta", {}).get("status") == "success"
+                and int(cached.get("meta", {}).get("schema_version") or 0) == 3
+            ):
+                cached["meta"] = {**cached.get("meta", {}), "cached": True}
+                return cached
+        if self.configured:
+            try:
+                payload = self._tushare_client().dragon_tiger(normalized_date)
+            except TushareError as exc:
+                return {
+                    "meta": {
+                        "requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
+                        "trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
+                        "source": "tushare_error",
+                        "status": "error",
+                        "schema_version": 3,
+                        "cached": False,
+                        "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+                        "notice": "龙虎榜数据暂不可用，请稍后重试。",
+                    },
+                    "summary": {
+                        "trader_count": 0,
+                        "identity_count": 0,
+                        "operation_count": 0,
+                        "active_stock_count": 0,
+                        "seat_net_buy_million": 0,
+                        "unclassified_count": 0,
+                        "directory_count": 0,
+                    },
+                    "traders": [],
+                    "unclassified_seats": [],
+                    "rows": [],
+                }
+            payload["meta"]["cached"] = False
+            if payload.get("meta", {}).get("status") == "success":
+                self.database.save_data_snapshot(cache_kind, normalized_date, "tushare", payload)
+            return payload
+
+        return {
+            "meta": {
+            "requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
+            "trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
+            "source": "unavailable",
+            "status": "unavailable",
+            "schema_version": 3,
+            "cached": False,
+            "notice": "龙虎榜数据暂不可用，请联系管理员检查行情配置。",
+            },
+            "summary": {
+                "trader_count": 0,
+                "identity_count": 0,
+                "operation_count": 0,
+                "active_stock_count": 0,
+                "seat_net_buy_million": 0,
+                "unclassified_count": 0,
+                "directory_count": 0,
+            },
+            "traders": [],
+            "unclassified_seats": [],
+            "rows": [],
+        }
+
+    def _apply_seat_aliases(self, payload: dict[str, Any]) -> dict[str, Any]:
+        aliases = self.database.list_seat_aliases()
+        result = dict(payload)
+        rows = payload.get("rows") or []
+        for row in rows:
+            for institution in row.get("institutions") or []:
+                institution["alias"] = aliases.get(institution.get("seat_name", ""), "")
+        traders: dict[tuple[str, str], dict[str, Any]] = {}
+        unclassified: dict[str, dict[str, Any]] = {}
+        seen_operations: set[tuple[Any, ...]] = set()
+        builtin_aliases = {
+            "国泰海通证券股份有限公司南京太平南路证券营业部": "作手新一",
+        }
+
+        for row in rows:
+            for institution in row.get("institutions") or []:
+                seat_name = str(institution.get("seat_name") or "未知席位").strip()
+                saved_alias = str(institution.get("alias") or "").strip()
+                builtin_alias = builtin_aliases.get(seat_name, "")
+                if saved_alias or builtin_alias:
+                    identity_name = saved_alias or builtin_alias
+                    identity_type = "trader"
+                    recognized = True
+                    identity_source = "manual" if saved_alias else "builtin"
+                elif "机构专用" in seat_name:
+                    identity_name = "机构专用"
+                    identity_type = "institution"
+                    recognized = True
+                    identity_source = "system"
+                elif "沪股通专用" in seat_name or "深股通专用" in seat_name:
+                    identity_name = "北向资金"
+                    identity_type = "channel"
+                    recognized = True
+                    identity_source = "system"
+                else:
+                    identity_name = seat_name
+                    identity_type = "unclassified"
+                    recognized = False
+                    identity_source = "raw"
+
+                buy = round(float(institution.get("buy_million") or 0), 2)
+                sell = round(float(institution.get("sell_million") or 0), 2)
+                net_buy = round(float(institution.get("net_buy_million") or 0), 2)
+                operation_key = (row.get("code"), seat_name, buy, sell, net_buy)
+                if operation_key in seen_operations:
+                    continue
+                seen_operations.add(operation_key)
+
+                group_key = (identity_type, identity_name)
+                group = traders.setdefault(
+                    group_key,
+                    {
+                        "name": identity_name,
+                        "identity_type": identity_type,
+                        "identity_source": identity_source,
+                        "recognized": recognized,
+                        "buy_million": 0.0,
+                        "sell_million": 0.0,
+                        "net_buy_million": 0.0,
+                        "seat_names": set(),
+                        "stock_codes": set(),
+                        "operations": [],
+                    },
+                )
+                group["buy_million"] += buy
+                group["sell_million"] += sell
+                group["net_buy_million"] += net_buy
+                group["seat_names"].add(seat_name)
+                group["stock_codes"].add(str(row.get("code") or ""))
+                group["operations"].append(
+                    {
+                        "code": row.get("code") or "",
+                        "name": row.get("name") or "--",
+                        "change": row.get("change") or 0,
+                        "direction": "买入" if net_buy > 0 else "卖出" if net_buy < 0 else "持平",
+                        "buy_million": buy,
+                        "sell_million": sell,
+                        "net_buy_million": net_buy,
+                        "reason": row.get("reason") or "--",
+                        "seat_name": seat_name,
+                        "seat_alias": identity_name if recognized else "",
+                    }
+                )
+
+                if not recognized:
+                    pending = unclassified.setdefault(
+                        seat_name,
+                        {
+                            "seat_name": seat_name,
+                            "stock_codes": set(),
+                            "operation_count": 0,
+                            "buy_million": 0.0,
+                            "sell_million": 0.0,
+                            "net_buy_million": 0.0,
+                        },
+                    )
+                    pending["stock_codes"].add(str(row.get("code") or ""))
+                    pending["operation_count"] += 1
+                    pending["buy_million"] += buy
+                    pending["sell_million"] += sell
+                    pending["net_buy_million"] += net_buy
+
+        type_order = {"trader": 0, "institution": 1, "channel": 2, "unclassified": 3}
+        aggregated = list(traders.values())
+        aggregated.sort(
+            key=lambda item: (
+                type_order.get(item["identity_type"], 9),
+                -abs(item["net_buy_million"]),
+                item["name"],
+            )
+        )
+        for index, group in enumerate(aggregated, start=1):
+            group["id"] = f"identity-{index}"
+            group["buy_million"] = round(group["buy_million"], 2)
+            group["sell_million"] = round(group["sell_million"], 2)
+            group["net_buy_million"] = round(group["net_buy_million"], 2)
+            group["seat_count"] = len(group.pop("seat_names"))
+            group["stock_count"] = len(group.pop("stock_codes"))
+            group["operation_count"] = len(group["operations"])
+            group["operations"].sort(
+                key=lambda item: abs(float(item.get("net_buy_million") or 0)), reverse=True
+            )
+
+        pending_seats = list(unclassified.values())
+        for pending in pending_seats:
+            pending["stock_count"] = len(pending.pop("stock_codes"))
+            pending["buy_million"] = round(pending["buy_million"], 2)
+            pending["sell_million"] = round(pending["sell_million"], 2)
+            pending["net_buy_million"] = round(pending["net_buy_million"], 2)
+        pending_seats.sort(key=lambda item: abs(item["net_buy_million"]), reverse=True)
+
+        operation_count = sum(item["operation_count"] for item in aggregated)
+        active_stocks = {
+            operation["code"] for item in aggregated for operation in item["operations"]
+        }
+        seat_net_buy = round(sum(item["net_buy_million"] for item in aggregated), 2)
+        result["rows"] = rows
+        result["traders"] = aggregated
+        result["unclassified_seats"] = pending_seats
+        result["summary"] = {
+            **(payload.get("summary") or {}),
+            "trader_count": sum(item["identity_type"] == "trader" for item in aggregated),
+            "identity_count": len(aggregated),
+            "operation_count": operation_count,
+            "active_stock_count": len(active_stocks),
+            "seat_net_buy_million": seat_net_buy,
+            "unclassified_count": len(pending_seats),
+        }
+        return result

app/backend/features/market/service.py

@@ -14,6 +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 backend.features.market.insights import MarketInsightsService
 from backend.features.sentiment.engine import SENTIMENT_ENGINE_VERSION
 
 
@@ -36,6 +37,14 @@ THS_SEARCH_TYPES = {
 
 
 class MarketServiceMixin:
+    def _market_insights(self) -> MarketInsightsService:
+        if not self.configured:
+            raise ValueError("行情数据尚未配置。")
+        return MarketInsightsService(
+            self.database,
+            self._tushare_client(),
+            ifind=self.ifind,
+        )
     def _tushare_client(self) -> TushareClient:
         gateway = getattr(self, "data_gateway", None)
         if gateway is not None:

app/backend/features/popularity/__init__.py

@@ -0,0 +1,4 @@
+from .repository import PopularityRepositoryMixin
+from .service import PopularityServiceMixin
+
+__all__ = ["PopularityRepositoryMixin", "PopularityServiceMixin"]

app/backend/features/popularity/repository.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class PopularityRepositoryMixin:
+    def upsert_popularity_factors(self, rows: list[dict[str, Any]]) -> int:
+        values = [
+            (
+                str(row.get("trade_date") or ""),
+                str(row.get("ts_code") or ""),
+                int(row["ths_rank"]) if row.get("ths_rank") not in (None, "") else None,
+                int(row["dc_rank"]) if row.get("dc_rank") not in (None, "") else None,
+                float(row.get("combined_score") or 0),
+                int(row["rank_change"]) if row.get("rank_change") not in (None, "") else None,
+                int(bool(row.get("dual_source"))),
+            )
+            for row in rows
+            if row.get("trade_date") and row.get("ts_code")
+        ]
+        with self.connect() as connection:
+            connection.executemany(
+                """
+                INSERT INTO popularity_factors
+                    (trade_date, ts_code, ths_rank, dc_rank, combined_score,
+                     rank_change, dual_source)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
+                    ths_rank=excluded.ths_rank,
+                    dc_rank=excluded.dc_rank,
+                    combined_score=excluded.combined_score,
+                    rank_change=excluded.rank_change,
+                    dual_source=excluded.dual_source
+                """,
+                values,
+            )
+        return len(values)

app/backend/features/popularity/service.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.bootstrap.config import normalize_date
+from backend.features.market.insights import MarketInsightsService
+
+
+class PopularityServiceMixin:
+    def popularity(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self._market_insights().popularity(normalize_date(trade_date), force)

app/backend/features/rotation/__init__.py

@@ -0,0 +1,5 @@
+"""Sector rotation history and constituent detail feature."""
+
+from .service import RotationServiceMixin
+
+__all__ = ["RotationServiceMixin"]

app/backend/features/rotation/service.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.bootstrap.config import normalize_date, validate_text
+from backend.data.providers.tushare_client import TushareError
+from backend.features.sentiment.engine import (
+    build_sentiment_history,
+    latest_contiguous_history,
+)
+
+
+class RotationServiceMixin:
+    def rotation_history(self, trade_date: str, limit: int = 9) -> dict[str, Any]:
+        normalized_date = normalize_date(trade_date)
+        # 板块轮动固定展示最近 9 个交易日，按由近到远排列。
+        limit = 9
+        snapshots = self.database.list_snapshot_payloads(normalized_date, 240)
+        by_trade_date: dict[str, dict[str, Any]] = {}
+        for snapshot in snapshots:
+            meta = snapshot.get("meta") or {}
+            actual_date = str(meta.get("trade_date") or snapshot.get("_snapshot_date") or "")
+            compact_date = actual_date.replace("-", "")
+            if len(compact_date) == 8:
+                by_trade_date[compact_date] = snapshot
+
+        sentiment_dates = {
+            str(row.get("trade_date") or "").replace("-", "")
+            for row in latest_contiguous_history(build_sentiment_history(snapshots))
+        }
+        ordered_dates = sorted(
+            date_key for date_key in by_trade_date
+            if not sentiment_dates or date_key in sentiment_dates
+        )[-limit:][::-1]
+        rows = []
+        for date_key in ordered_dates:
+            snapshot = by_trade_date[date_key]
+            sector_context = {
+                str(item.get("name") or ""): item
+                for item in snapshot.get("sectors") or []
+            }
+            sectors = []
+            for item in (snapshot.get("sector_rotation") or [])[:12]:
+                name = str(item.get("name") or "").strip()
+                context = sector_context.get(name, {})
+                sectors.append(
+                    {
+                        "name": name,
+                        "rank": int(item.get("rank") or len(sectors) + 1),
+                        "trend": item.get("trend") or "持平",
+                        "count": int(item.get("count") or 0),
+                        "strength": float(item.get("strength") or context.get("strength") or 0),
+                        "change": float(context.get("change") or 0),
+                        "leader": item.get("leader") or context.get("leader") or "--",
+                    }
+                )
+            rows.append(
+                {
+                    "trade_date": f"{date_key[:4]}-{date_key[4:6]}-{date_key[6:]}",
+                    "sectors": sectors,
+                }
+            )
+        return {
+            "trade_date": rows[0]["trade_date"] if rows else normalized_date,
+            "available_days": len(ordered_dates),
+            "requested_days": limit,
+            "rows": rows,
+        }
+
+    def rotation_sector_members(self, trade_date: str, sector_name: str) -> dict[str, Any]:
+        normalized_date = normalize_date(trade_date)
+        sector_name = validate_text(sector_name, "板块名称", 60, required=True)
+        dashboard = self.get_dashboard(normalized_date)
+        actual_date = normalize_date(
+            str((dashboard.get("meta") or {}).get("trade_date") or normalized_date)
+        )
+        cache_key = f"{actual_date}:{sector_name}"
+        cached = self.database.get_data_snapshot("rotation_sector_members_v1", cache_key)
+        if cached:
+            cached["meta"] = {**(cached.get("meta") or {}), "cached": True}
+            return cached
+        if not self.configured:
+            raise ValueError("板块成分数据暂不可用。")
+
+        representative = next(
+            (
+                item for item in dashboard.get("limits") or []
+                if str(item.get("sector") or "").strip() == sector_name
+            ),
+            None,
+        )
+        if not representative:
+            raise ValueError("未找到该板块的代表股票，暂时无法核验成分股。")
+        raw_code = str(representative.get("ts_code") or representative.get("code") or "")
+        if "." in raw_code:
+            ts_code = raw_code
+        elif raw_code.startswith(("4", "8", "92")):
+            ts_code = f"{raw_code}.BJ"
+        elif raw_code.startswith(("6", "68", "90")):
+            ts_code = f"{raw_code}.SH"
+        else:
+            ts_code = f"{raw_code}.SZ"
+        client = self._tushare_client()
+        try:
+            industry = client.sw_stock_industry(ts_code, actual_date)
+            sector_code = str(industry.get("l2_code") or "")
+            members = client.sw_sector_members(sector_code, actual_date)
+        except TushareError as exc:
+            raise ValueError(f"该板块成分股暂不可用：{exc}") from exc
+
+        daily_rows = self.database.daily_bars_for_date(actual_date)
+        if len(daily_rows) < 1000:
+            try:
+                daily_rows = client.query(
+                    "daily",
+                    {"trade_date": actual_date},
+                    "ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
+                )
+                if daily_rows:
+                    self.database.upsert_daily_bars(daily_rows)
+            except TushareError:
+                daily_rows = self.database.daily_bars_for_date(actual_date)
+        daily_map = {str(item.get("ts_code") or ""): item for item in daily_rows}
+        rows = []
+        for member in members:
+            member_code = str(member.get("ts_code") or "")
+            quote = daily_map.get(member_code) or {}
+            rows.append(
+                {
+                    "code": member_code.split(".")[0],
+                    "ts_code": member_code,
+                    "name": str(member.get("name") or "--"),
+                    "change": quote.get("pct_chg"),
+                    "open": quote.get("open"),
+                    "close": quote.get("close"),
+                    "amount_billion": (
+                        round(float(quote.get("amount") or 0) / 100000, 2)
+                        if quote else None
+                    ),
+                    "quoted": bool(quote),
+                }
+            )
+        rows.sort(
+            key=lambda item: (
+                bool(item.get("quoted")),
+                float(item.get("change") or -999),
+                float(item.get("amount_billion") or 0),
+            ),
+            reverse=True,
+        )
+        result = {
+            "meta": {
+                "trade_date": self._display_compact_date(actual_date),
+                "sector_name": str(industry.get("l2_name") or sector_name),
+                "sector_code": sector_code,
+                "member_count": len(rows),
+                "quoted_count": sum(bool(item.get("quoted")) for item in rows),
+                "cached": False,
+            },
+            "rows": rows,
+        }
+        self.database.save_data_snapshot(
+            "rotation_sector_members_v1", cache_key, "tushare", result
+        )
+        return result

app/backend/features/themes/__init__.py

@@ -0,0 +1,3 @@
+from .service import ThemeServiceMixin
+
+__all__ = ["ThemeServiceMixin"]

app/backend/features/themes/service.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.bootstrap.config import normalize_date
+from backend.features.market.insights import MarketInsightsService
+
+
+class ThemeServiceMixin:
+    def theme_library(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self._market_insights().theme_library(normalize_date(trade_date), force)
+
+    def theme_detail(self, code: str, trade_date: str) -> dict[str, Any]:
+        return self._market_insights().theme_detail(code, normalize_date(trade_date))

app/database.py

@@ -8,8 +8,11 @@ from typing import Any
 
 from backend.database import MIGRATIONS, MigrationRunner, SQLiteConnectionFactory
 from backend.features.accounts.repository import AccountRepositoryMixin
+from backend.features.auction.repository import AuctionRepositoryMixin
+from backend.features.dragon_tiger.repository import DragonTigerRepositoryMixin
 from backend.features.market.repository import MarketRepositoryMixin
 from backend.features.pools.repository import PoolRepositoryMixin
+from backend.features.popularity.repository import PopularityRepositoryMixin
 from backend.features.system.repository import SystemSettingsRepositoryMixin
 
 
@@ -24,8 +27,11 @@ def _optional_float(value: Any) -> float | None:
 
 class ReviewDatabase(
     AccountRepositoryMixin,
+    AuctionRepositoryMixin,
+    DragonTigerRepositoryMixin,
     MarketRepositoryMixin,
     PoolRepositoryMixin,
+    PopularityRepositoryMixin,
     SystemSettingsRepositoryMixin,
 ):
     def __init__(self, path: Path) -> None:
@@ -839,24 +845,6 @@ class ReviewDatabase(
             return cursor.rowcount > 0
 
 
-    def list_seat_aliases(self) -> dict[str, str]:
-        with self.connect() as connection:
-            rows = connection.execute("SELECT seat_name, alias FROM seat_aliases").fetchall()
-        return {row["seat_name"]: row["alias"] for row in rows}
-
-    def save_seat_alias(self, seat_name: str, alias: str) -> None:
-        now = datetime.now().astimezone().isoformat(timespec="seconds")
-        with self.connect() as connection:
-            connection.execute(
-                """
-                INSERT INTO seat_aliases (seat_name, alias, updated_at)
-                VALUES (?, ?, ?)
-                ON CONFLICT(seat_name) DO UPDATE SET
-                    alias = excluded.alias,
-                    updated_at = excluded.updated_at
-                """,
-                (seat_name, alias, now),
-            )
 
     def list_sector_phase_overrides(self) -> dict[str, str]:
         with self.connect() as connection:
@@ -1056,44 +1044,6 @@ class ReviewDatabase(
             )
         return len(values)
 
-    def upsert_auction_factors(self, rows: list[dict[str, Any]]) -> int:
-        values = []
-        for row in rows:
-            trade_date = str(row.get("trade_date") or "")
-            ts_code = str(row.get("ts_code") or "")
-            price = float(row.get("price") or 0)
-            pre_close = float(row.get("pre_close") or 0)
-            if not trade_date or not ts_code or price <= 0 or pre_close <= 0:
-                continue
-            values.append(
-                (
-                    trade_date,
-                    ts_code,
-                    price,
-                    pre_close,
-                    (price / pre_close - 1) * 100,
-                    float(row.get("vol") or 0),
-                    float(row.get("amount") or 0),
-                    float(row.get("turnover_rate") or 0),
-                    float(row.get("volume_ratio") or 0),
-                )
-            )
-        with self.connect() as connection:
-            connection.executemany(
-                """
-                INSERT INTO auction_factors
-                    (trade_date, ts_code, price, pre_close, change, vol, amount,
-                     turnover_rate, volume_ratio)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
-                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
-                    price=excluded.price, pre_close=excluded.pre_close,
-                    change=excluded.change, vol=excluded.vol, amount=excluded.amount,
-                    turnover_rate=excluded.turnover_rate,
-                    volume_ratio=excluded.volume_ratio
-                """,
-                values,
-            )
-        return len(values)
 
     def upsert_earnings_events(self, rows: list[dict[str, Any]]) -> int:
         values = [
@@ -1130,84 +1080,7 @@ class ReviewDatabase(
             )
         return len(values)
 
-    def upsert_popularity_factors(self, rows: list[dict[str, Any]]) -> int:
-        values = [
-            (
-                str(row.get("trade_date") or ""),
-                str(row.get("ts_code") or ""),
-                int(row["ths_rank"]) if row.get("ths_rank") not in (None, "") else None,
-                int(row["dc_rank"]) if row.get("dc_rank") not in (None, "") else None,
-                float(row.get("combined_score") or 0),
-                int(row["rank_change"]) if row.get("rank_change") not in (None, "") else None,
-                int(bool(row.get("dual_source"))),
-            )
-            for row in rows
-            if row.get("trade_date") and row.get("ts_code")
-        ]
-        with self.connect() as connection:
-            connection.executemany(
-                """
-                INSERT INTO popularity_factors
-                    (trade_date, ts_code, ths_rank, dc_rank, combined_score,
-                     rank_change, dual_source)
-                VALUES (?, ?, ?, ?, ?, ?, ?)
-                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
-                    ths_rank=excluded.ths_rank,
-                    dc_rank=excluded.dc_rank,
-                    combined_score=excluded.combined_score,
-                    rank_change=excluded.rank_change,
-                    dual_source=excluded.dual_source
-                """,
-                values,
-            )
-        return len(values)
 
-    def upsert_lhb_institutions(self, rows: list[dict[str, Any]]) -> int:
-        grouped: dict[tuple[str, str], dict[str, float | int]] = {}
-        for row in rows:
-            trade_date = str(row.get("trade_date") or "")
-            ts_code = str(row.get("ts_code") or "")
-            seat_name = str(row.get("exalter") or row.get("seat_name") or "")
-            if not trade_date or not ts_code or "机构专用" not in seat_name:
-                continue
-            group = grouped.setdefault(
-                (trade_date, ts_code),
-                {"net": 0.0, "buy": 0.0, "sell": 0.0, "seats": 0},
-            )
-            group["net"] = float(group["net"]) + float(row.get("net_buy") or row.get("net_amount") or 0)
-            group["buy"] = float(group["buy"]) + float(row.get("buy") or row.get("buy_amount") or 0)
-            group["sell"] = float(group["sell"]) + float(row.get("sell") or row.get("sell_amount") or 0)
-            group["seats"] = int(group["seats"]) + 1
-        values = [
-            (trade_date, ts_code, item["net"], item["buy"], item["sell"], item["seats"])
-            for (trade_date, ts_code), item in grouped.items()
-        ]
-        with self.connect() as connection:
-            connection.executemany(
-                """
-                INSERT INTO lhb_institution_daily
-                    (trade_date, ts_code, net_buy_amount, buy_amount, sell_amount, seat_count)
-                VALUES (?, ?, ?, ?, ?, ?)
-                ON CONFLICT(trade_date, ts_code) DO UPDATE SET
-                    net_buy_amount=excluded.net_buy_amount,
-                    buy_amount=excluded.buy_amount,
-                    sell_amount=excluded.sell_amount,
-                    seat_count=excluded.seat_count
-                """,
-                values,
-            )
-        return len(values)
-
-    def auction_factor_dates(self, end_date: str = "", limit: int = 80) -> list[str]:
-        where = "WHERE trade_date <= ?" if end_date else ""
-        parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
-        with self.connect() as connection:
-            rows = connection.execute(
-                f"SELECT DISTINCT trade_date FROM auction_factors {where} "
-                "ORDER BY trade_date DESC LIMIT ?",
-                parameters,
-            ).fetchall()
-        return [row["trade_date"] for row in reversed(rows)]
 
     def daily_indicator_dates(self, end_date: str = "", limit: int = 400) -> list[str]:
         where = "WHERE trade_date <= ?" if end_date else ""
@@ -1227,13 +1100,6 @@ class ReviewDatabase(
             ).fetchall()
         return [str(row["end_date"]) for row in rows]
 
-    def auction_factors_for_date(self, trade_date: str) -> list[dict[str, Any]]:
-        with self.connect() as connection:
-            rows = connection.execute(
-                "SELECT * FROM auction_factors WHERE trade_date = ? ORDER BY ts_code",
-                (trade_date,),
-            ).fetchall()
-        return [dict(row) for row in rows]
 
     def daily_bars_for_date(self, trade_date: str) -> list[dict[str, Any]]:
         with self.connect() as connection:

app/tests/test_preservation_slice_ladder_rotation.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import ast
+import hashlib
+import unittest
+from pathlib import Path
+
+
+APP_ROOT = Path(__file__).resolve().parents[1]
+ORIGINAL_ROOT = APP_ROOT.parent
+
+ROTATION_METHODS = {
+    "rotation_history",
+    "rotation_sector_members",
+}
+LADDER_ROTATION_BUILDERS = {
+    "_build_ladders",
+    "_build_sector_rotation",
+}
+
+
+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 top_level_functions(path: Path) -> dict[str, str]:
+    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    return {
+        node.name: ast.dump(node, include_attributes=False)
+        for node in tree.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and node.name in LADDER_ROTATION_BUILDERS
+    }
+
+
+def sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+class LadderRotationSliceSourceEquivalenceTests(unittest.TestCase):
+    def test_rotation_service_methods_are_exact_original_ast(self) -> None:
+        original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
+        migrated = class_methods(
+            APP_ROOT / "backend" / "features" / "rotation" / "service.py",
+            "RotationServiceMixin",
+        )
+        self.assertEqual(set(migrated), ROTATION_METHODS)
+        for name in sorted(ROTATION_METHODS):
+            self.assertEqual(migrated[name], original[name], name)
+
+    def test_dashboard_service_no_longer_duplicates_rotation_methods(self) -> None:
+        remaining = class_methods(
+            APP_ROOT / "backend" / "application.py", "DashboardService"
+        )
+        self.assertTrue(ROTATION_METHODS.isdisjoint(remaining))
+
+    def test_ladder_and_rotation_builders_are_exact_original_ast(self) -> None:
+        self.assertEqual(
+            top_level_functions(ORIGINAL_ROOT / "tushare_client.py"),
+            top_level_functions(
+                APP_ROOT / "backend" / "data" / "providers" / "tushare_client.py"
+            ),
+        )
+
+    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/ladder/page.js",
+            "static/pages/rotation/page.js",
+        ):
+            self.assertEqual(
+                sha256(APP_ROOT / relative),
+                sha256(ORIGINAL_ROOT / relative),
+                relative,
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

app/tests/test_preservation_slice_market.py

@@ -20,6 +20,7 @@ ORIGINAL_ROOT = APP_ROOT.parent
 
 MARKET_METHODS = {
     "_tushare_client",
+    "_market_insights",
     "get_dashboard",
     "_dashboard_sentiment_ready",
     "_display_compact_date",

app/tests/test_preservation_slice_market_insights.py

@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+import ast
+import hashlib
+import unittest
+from pathlib import Path
+
+import market_insights
+from backend.features.market import insights as canonical_insights
+
+
+APP_ROOT = Path(__file__).resolve().parents[1]
+ORIGINAL_ROOT = APP_ROOT.parent
+
+MARKET_INSIGHT_METHODS = {
+    "__init__",
+    "_trade_context",
+    "_latest_feature_snapshot",
+    "_auction_session",
+    "_stock_master",
+    "_expectation_label",
+    "_auction_confirmation",
+    "_attention_score",
+    "_auction_candidates",
+    "_auction_theme_evidence",
+    "_auction_amount_history",
+    "_ensure_auction_amount_history",
+    "_with_auction_watchlist",
+    "_dynamic_auction_rows",
+    "auction_center",
+    "_theme_directory",
+    "theme_library",
+    "theme_detail",
+    "_parse_concepts",
+    "popularity",
+    "_hot_rows",
+    "_normalize_hot",
+}
+MARKET_SERVICE_METHODS = {"_market_insights"}
+AUCTION_SERVICE_METHODS = {"auction_center"}
+THEME_SERVICE_METHODS = {"theme_library", "theme_detail"}
+POPULARITY_SERVICE_METHODS = {"popularity"}
+DRAGON_TIGER_SERVICE_METHODS = {
+    "get_hot_money_profiles",
+    "get_dragon_tiger",
+    "_apply_seat_aliases",
+}
+AUCTION_REPOSITORY_METHODS = {
+    "upsert_auction_factors",
+    "auction_factor_dates",
+    "auction_factors_for_date",
+}
+POPULARITY_REPOSITORY_METHODS = {"upsert_popularity_factors"}
+DRAGON_TIGER_REPOSITORY_METHODS = {
+    "list_seat_aliases",
+    "save_seat_alias",
+    "upsert_lhb_institutions",
+}
+TUSHARE_METHODS = {"hot_money_profiles", "dragon_tiger"}
+
+
+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 MarketInsightsSliceSourceEquivalenceTests(unittest.TestCase):
+    def assert_methods_equal(
+        self,
+        original_path: Path,
+        original_class: str,
+        migrated_path: Path,
+        migrated_class: str,
+        names: set[str],
+    ) -> None:
+        original = class_methods(original_path, original_class)
+        migrated = class_methods(migrated_path, migrated_class)
+        self.assertEqual(set(migrated), names)
+        for name in sorted(names):
+            self.assertEqual(migrated[name], original[name], name)
+
+    def test_shared_market_insight_service_is_exact_original_ast(self) -> None:
+        self.assert_methods_equal(
+            ORIGINAL_ROOT / "market_insights.py",
+            "MarketInsightsService",
+            APP_ROOT / "backend" / "features" / "market" / "insights.py",
+            "MarketInsightsService",
+            MARKET_INSIGHT_METHODS,
+        )
+        self.assertIs(market_insights.MarketInsightsService, canonical_insights.MarketInsightsService)
+
+    def test_dashboard_service_methods_are_exact_original_ast(self) -> None:
+        original = ORIGINAL_ROOT / "server.py"
+        mappings = (
+            ("auction/service.py", "AuctionServiceMixin", AUCTION_SERVICE_METHODS),
+            ("themes/service.py", "ThemeServiceMixin", THEME_SERVICE_METHODS),
+            ("popularity/service.py", "PopularityServiceMixin", POPULARITY_SERVICE_METHODS),
+            ("dragon_tiger/service.py", "DragonTigerServiceMixin", DRAGON_TIGER_SERVICE_METHODS),
+        )
+        for relative, class_name, names in mappings:
+            with self.subTest(relative=relative):
+                self.assert_methods_equal(
+                    original,
+                    "DashboardService",
+                    APP_ROOT / "backend" / "features" / relative,
+                    class_name,
+                    names,
+                )
+        original_methods = class_methods(original, "DashboardService")
+        market_methods = class_methods(
+            APP_ROOT / "backend" / "features" / "market" / "service.py",
+            "MarketServiceMixin",
+        )
+        for name in MARKET_SERVICE_METHODS:
+            self.assertEqual(market_methods[name], original_methods[name], name)
+
+    def test_repository_methods_are_exact_original_ast(self) -> None:
+        original = ORIGINAL_ROOT / "database.py"
+        mappings = (
+            ("auction/repository.py", "AuctionRepositoryMixin", AUCTION_REPOSITORY_METHODS),
+            ("popularity/repository.py", "PopularityRepositoryMixin", POPULARITY_REPOSITORY_METHODS),
+            ("dragon_tiger/repository.py", "DragonTigerRepositoryMixin", DRAGON_TIGER_REPOSITORY_METHODS),
+        )
+        for relative, class_name, names in mappings:
+            with self.subTest(relative=relative):
+                self.assert_methods_equal(
+                    original,
+                    "ReviewDatabase",
+                    APP_ROOT / "backend" / "features" / relative,
+                    class_name,
+                    names,
+                )
+
+    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")
+        moved_service = (
+            MARKET_SERVICE_METHODS
+            | AUCTION_SERVICE_METHODS
+            | THEME_SERVICE_METHODS
+            | POPULARITY_SERVICE_METHODS
+            | DRAGON_TIGER_SERVICE_METHODS
+        )
+        moved_repository = (
+            AUCTION_REPOSITORY_METHODS
+            | POPULARITY_REPOSITORY_METHODS
+            | DRAGON_TIGER_REPOSITORY_METHODS
+        )
+        self.assertTrue(moved_service.isdisjoint(remaining_service))
+        self.assertTrue(moved_repository.isdisjoint(remaining_database))
+
+    def test_tushare_dragon_tiger_implementations_are_exact_original_ast(self) -> None:
+        original = class_methods(ORIGINAL_ROOT / "tushare_client.py", "TushareClient")
+        migrated = class_methods(
+            APP_ROOT / "backend" / "data" / "providers" / "tushare_client.py",
+            "TushareClient",
+        )
+        for name in sorted(TUSHARE_METHODS):
+            self.assertEqual(migrated[name], original[name], name)
+
+    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/auction/page.js",
+            "static/pages/themes/page.js",
+            "static/pages/popularity/page.js",
+            "static/pages/dragon-tiger/page.js",
+        ):
+            self.assertEqual(
+                sha256(APP_ROOT / relative),
+                sha256(ORIGINAL_ROOT / relative),
+                relative,
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

app/tools/compare_preservation_apis.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import http.cookiejar
+import json
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+def request_json(
+    opener: urllib.request.OpenerDirector,
+    url: str,
+    payload: dict[str, Any] | None = None,
+) -> tuple[int, Any]:
+    data = None
+    headers = {"Accept": "application/json"}
+    if payload is not None:
+        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(url, data=data, headers=headers)
+    try:
+        with opener.open(request, timeout=90) as response:
+            return response.status, json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        return exc.code, json.loads(exc.read().decode("utf-8"))
+
+
+def session(base_url: str, username: str, password: str) -> urllib.request.OpenerDirector:
+    opener = urllib.request.build_opener(
+        urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())
+    )
+    status, body = request_json(
+        opener,
+        f"{base_url.rstrip('/')}/api/auth/login",
+        {"username": username, "password": password},
+    )
+    if status != 200 or not body.get("ok"):
+        raise RuntimeError(f"Login failed for {base_url}: HTTP {status} {body}")
+    return opener
+
+
+def digest(value: Any) -> str:
+    content = json.dumps(
+        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+    return hashlib.sha256(content).hexdigest()
+
+
+def comparable(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {
+            key: comparable(item)
+            for key, item in value.items()
+            if key != "request_id"
+        }
+    if isinstance(value, list):
+        return [comparable(item) for item in value]
+    return value
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Compare authenticated preservation APIs")
+    parser.add_argument("--original", required=True)
+    parser.add_argument("--migrated", required=True)
+    parser.add_argument("--username", required=True)
+    parser.add_argument("--password", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("endpoints", nargs="+")
+    args = parser.parse_args()
+
+    original = session(args.original, args.username, args.password)
+    migrated = session(args.migrated, args.username, args.password)
+    rows = []
+    all_equal = True
+    for endpoint in args.endpoints:
+        original_status, original_body = request_json(
+            original, f"{args.original.rstrip('/')}{endpoint}"
+        )
+        migrated_status, migrated_body = request_json(
+            migrated, f"{args.migrated.rstrip('/')}{endpoint}"
+        )
+        original_comparable = comparable(original_body)
+        migrated_comparable = comparable(migrated_body)
+        equal = original_status == migrated_status and original_comparable == migrated_comparable
+        all_equal = all_equal and equal
+        rows.append(
+            {
+                "endpoint": endpoint,
+                "original_status": original_status,
+                "migrated_status": migrated_status,
+                "original_sha256": digest(original_comparable),
+                "migrated_sha256": digest(migrated_comparable),
+                "equal": equal,
+            }
+        )
+
+    result = {"all_equal": all_equal, "endpoints": rows}
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(
+        json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+    )
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    if not all_equal:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

app/tools/compare_preservation_databases.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sqlite3
+from pathlib import Path
+from typing import Any
+
+
+def digest(value: Any) -> str:
+    content = json.dumps(
+        value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str
+    ).encode("utf-8")
+    return hashlib.sha256(content).hexdigest()
+
+
+def schema(connection: sqlite3.Connection) -> list[dict[str, Any]]:
+    rows = connection.execute(
+        """
+        SELECT type, name, tbl_name, sql
+        FROM sqlite_master
+        WHERE name NOT LIKE 'sqlite_%'
+        ORDER BY type, name
+        """
+    ).fetchall()
+    return [dict(row) for row in rows]
+
+
+def table_rows(connection: sqlite3.Connection, table: str) -> list[dict[str, Any]]:
+    quoted = '"' + table.replace('"', '""') + '"'
+    rows = [dict(row) for row in connection.execute(f"SELECT * FROM {quoted}").fetchall()]
+    return sorted(rows, key=lambda row: json.dumps(row, ensure_ascii=False, sort_keys=True, default=str))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Compare preservation SQLite databases")
+    parser.add_argument("--original", type=Path, required=True)
+    parser.add_argument("--migrated", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("tables", nargs="+")
+    args = parser.parse_args()
+
+    original = sqlite3.connect(args.original)
+    migrated = sqlite3.connect(args.migrated)
+    original.row_factory = sqlite3.Row
+    migrated.row_factory = sqlite3.Row
+    try:
+        original_schema = schema(original)
+        migrated_schema = schema(migrated)
+        tables = []
+        all_equal = original_schema == migrated_schema
+        for table in args.tables:
+            original_rows = table_rows(original, table)
+            migrated_rows = table_rows(migrated, table)
+            equal = original_rows == migrated_rows
+            all_equal = all_equal and equal
+            tables.append(
+                {
+                    "table": table,
+                    "original_count": len(original_rows),
+                    "migrated_count": len(migrated_rows),
+                    "original_sha256": digest(original_rows),
+                    "migrated_sha256": digest(migrated_rows),
+                    "equal": equal,
+                }
+            )
+        result = {
+            "all_equal": all_equal,
+            "schema": {
+                "object_count": len(original_schema),
+                "original_sha256": digest(original_schema),
+                "migrated_sha256": digest(migrated_schema),
+                "equal": original_schema == migrated_schema,
+            },
+            "tables": tables,
+        }
+    finally:
+        original.close()
+        migrated.close()
+
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(
+        json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+    )
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    if not all_equal:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

app/tools/run_preservation_runtime.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import argparse
+import sys
+from http.server import ThreadingHTTPServer
+from pathlib import Path
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run an isolated preservation runtime")
+    parser.add_argument("--runtime-root", type=Path, required=True)
+    parser.add_argument("--data-dir", type=Path, required=True)
+    parser.add_argument("--port", type=int, required=True)
+    args = parser.parse_args()
+
+    runtime_root = args.runtime_root.resolve()
+    data_dir = args.data_dir.resolve()
+    data_dir.mkdir(parents=True, exist_ok=True)
+    sys.path.insert(0, str(runtime_root))
+
+    if (runtime_root / "backend" / "bootstrap" / "config.py").is_file():
+        from backend.bootstrap import config
+
+        config.DATA_DIR = data_dir
+        config.PRIVATE_MENTOR_SKILLS_DIR = data_dir / "private-mentor-skills"
+    else:
+        import app_config as config
+
+        config.DATA_DIR = data_dir
+        config.PRIVATE_MENTOR_SKILLS_DIR = data_dir / "private-mentor-skills"
+
+    from server import RequestHandler, SERVICE
+
+    server = ThreadingHTTPServer(("127.0.0.1", args.port), RequestHandler)
+    print(f"Preservation runtime is running at http://127.0.0.1:{args.port}", flush=True)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        SERVICE._background_stop.set()
+        server.server_close()
+
+
+if __name__ == "__main__":
+    main()

docs/migration/evidence/slice-04/README.md

@@ -0,0 +1,58 @@
+# 切片 04：市场天梯与板块轮动
+
+> 基线：`b3555d2`（切片 03）
+> 回档标签：`xiaobai-preservation-slice-04-20260731`
+> 结论：源码、API、真实页面和浏览器回归通过；最终视觉仍等待全站人工验收
+
+## 1. 原实现归位
+
+本切片从原版副本机械移动板块轮动服务，没有从 `next/` 取用代码，也没有修改天梯、轮动的计算、
+排序、展开、配色、页面结构或交互。
+
+| 原位置 | 新的唯一实现位置 | 原位置兼容 |
+|---|---|---|
+| `app/backend/application.py` 的 2 个轮动方法 | `app/backend/features/rotation/service.py` | `DashboardService` 继承 `RotationServiceMixin` |
+| Tushare Provider 的天梯与轮动构造函数 | 保持 `app/backend/data/providers/tushare_client.py` | 切片 02 已归位的公共数据实现 |
+
+市场天梯没有独立后端 API 或第二套计算，直接展示 `/api/dashboard` 中原 Tushare 实现生成的
+`ladders`；因此没有为目录形式建立空的天梯服务。
+
+## 2. 等价证据
+
+- `test_preservation_slice_ladder_rotation.py` 对 2 个轮动服务方法逐项执行无位置信息 AST 比较，
+  全部与根目录原版 `server.py` 完全相同。
+- `_build_ladders` 与 `_build_sector_rotation` 两个原数据构造函数的 AST 与根目录原版完全相同。
+- 原版 `8784` 与迁移版 `8785` 在相同账号、日期和数据库副本上返回的天梯数据及 9 日轮动历史
+  JSON 逐字段完全相同。
+- 成分股接口在当前外部网络状态下两版均返回 HTTP 400、`bad_request` 和相同的
+  `该板块成分股暂不可用：Tushare request failed:`，没有改变错误或增加静默降级。
+- `config/api.config.json`、API 路径、鉴权、数据库 schema 和 `app/static/` 未修改。
+
+## 3. 真实运行检查
+
+- 市场天梯：8 个层级（含断层）、18 个首屏股票单元格、3 个结构分析模块正常；1920×1080 下
+  页面宽度无溢出，首板展开入口保留。
+- 板块轮动：9 个交易日、每日 Top 12 共 108 个板块单元格、由远到近/由近到远两个排序入口正常；
+  1920×1080 下页面宽度无溢出并保持全页滚动。
+- 日间模式页面控制台没有错误或警告。
+- `app-light-ladder-1920x1080.png` SHA-256：
+  `9e57d18d92e745fd92131f7bf08f21faaaa745476dd942cdaa2a703b9a7a303a`。
+- `app-light-rotation-1920x1080.png` SHA-256：
+  `03c092bb40bd0eb672136dff853abffc87e9790840107c2f22e6cf31d5f83c09`。
+
+## 4. 自动验证
+
+| 验证 | 结果 |
+|---|---:|
+| `python -m unittest discover -s tests -q` | 252 项通过 |
+| `python -m unittest tests.test_preservation_slice_ladder_rotation -q` | 4 项通过 |
+| 切片 02 至 04 与总览缓存专项集合 | 24 项通过 |
+| `npx.cmd playwright test --reporter=dot` | 45 项通过 |
+| `git diff --check` | 通过 |
+
+## 5. 保留边界
+
+- 成分股接口依赖的日行情与因子持久化方法仍由原 `ReviewDatabase` 提供，因其同时服务智能选股，
+  待切片 06 随完整共享职责归位。
+- 天梯和轮动前端资产保持原位，切片 10 再按页面职责归档；当前没有复制或改写。
+- 没有删除待定代码、没有改动根目录正式数据库、没有切换 Docker/NAS。

docs/migration/evidence/slice-05/README.md

@@ -0,0 +1,69 @@
+# 切片 05：集合竞价、题材库、人气热榜与龙虎榜
+
+> 基线：`814e757`（切片 04）
+> 回档标签：`xiaobai-preservation-slice-05-20260731`
+> 结论：源码、API、数据库、真实页面和全量回归通过；最终视觉仍等待全站人工验收
+
+## 1. 原实现归位
+
+本切片没有从`next/`取用代码，也没有重写计算、页面或接口。集合竞价、题材库和人气热榜原本
+共享`MarketInsightsService`，其中竞价候选会直接调用人气榜热度数据，因此整体移动到公共行情
+领域，避免拆出互相复制的实现；各页面入口仍按功能目录归位。
+
+| 原位置 | 新的唯一实现位置 | 兼容方式 |
+|---|---|---|
+| `app/market_insights.py` | `app/backend/features/market/insights.py` | 根级模块导出同一类对象 |
+| `DashboardService`竞价入口 | `app/backend/features/auction/service.py` | `AuctionServiceMixin` |
+| `DashboardService`题材入口 | `app/backend/features/themes/service.py` | `ThemeServiceMixin` |
+| `DashboardService`人气入口 | `app/backend/features/popularity/service.py` | `PopularityServiceMixin` |
+| `DashboardService`龙虎榜及游资档案 | `app/backend/features/dragon_tiger/service.py` | `DragonTigerServiceMixin` |
+| 竞价、人气、龙虎榜持久化方法 | 对应功能目录的`repository.py` | `ReviewDatabase`继承原接口 |
+
+Tushare Provider 中`hot_money_profiles`与`dragon_tiger`继续保持切片02归位的唯一实现，没有为目录
+形式再制造一套数据构造逻辑。
+
+## 2. 源码与接口等价
+
+- `test_preservation_slice_market_insights.py`逐项比较22个市场洞察方法、8个页面服务方法、7个
+  Repository方法和2个Tushare方法，全部与根目录原版无位置信息AST一致。
+- `DashboardService`与`ReviewDatabase`不再重复保留已移动方法；根级`market_insights`与新模块
+  暴露同一个`MarketInsightsService`类对象。
+- 原版`8784`和迁移版`8785`使用同一数据库的独立副本，集合竞价、题材库、题材详情、人气热榜、
+  龙虎榜、游资档案和席位别名共7个真实API状态码及JSON一致。
+- 题材详情在当前外部网络条件下两版均返回HTTP 400；差分只排除每次请求随机生成的
+  `request_id`，错误码与错误内容仍完全一致。
+- 完整接口摘要见`api-diff.json`。
+
+## 3. 数据库差分
+
+- 两个副本均为62个schema对象，哈希均为
+  `60a4e044f0ddb6502e44b45f65bedc1a0a31d4bd596f386d4ccfe153a6c8ddd1`。
+- `auction_factors` 511914行、`popularity_factors` 232行、`lhb_institution_daily` 47行、
+  `seat_aliases` 0行、`stock_master` 5535行均逐行一致。
+- 完整表计数与哈希见`database-diff.json`；运行数据库副本已在验收后删除，未提交凭据或正式数据。
+
+## 4. 真实浏览器检查
+
+- 1920×1080日间模式检查集合竞价、题材库、人气热榜和龙虎榜四页；均无横向溢出，控制台无
+  错误或警告。
+- 集合竞价载入30行重点候选；题材库载入394个题材及选中题材成分股；人气热榜载入3个摘要
+  模块和200行综合榜；龙虎榜按当前缓存显示既有不可用空态。
+- 四页HTML、主JS、CSS及各自页面JS与根目录原版字节哈希一致。
+- 截图SHA-256：
+  - `app-light-auction-1920x1080.png`：`8065086c8f2b360aeb1004bd60429f3e2d7f1b8c872ec4a965e9a5d0c016b915`
+  - `app-light-themes-1920x1080.png`：`80e1e41101497ee7213e1dadfaa4c9572a1c47b3495edd09e36745ccdb639402`
+  - `app-light-popularity-1920x1080.png`：`614d6b7770f4e9ec72c059ab8a4129486df9eff5c4dd7e8279cc68ebcb80a36b`
+  - `app-light-dragon-tiger-1920x1080.png`：`2f1e2ad3a9bc3874c73bae884fc8744177cef354c7176559da1453fab1f85993`
+
+## 5. 自动验证与保留边界
+
+| 验证 | 结果 |
+|---|---:|
+| `python -m unittest discover -s tests -q` | 258项通过 |
+| `python -m unittest tests.test_preservation_slice_market_insights -q` | 6项通过 |
+| `npx.cmd playwright test --reporter=dot` | 45项通过 |
+| `git diff --check` | 通过 |
+
+- 竞价、人气和龙虎榜因子同时服务切片06智能选股，迁移后仍由`ReviewDatabase`原方法名暴露。
+- 前端资产保持原位置，切片10再按页面职责归档；本切片没有改DOM、CSS、动画或交互。
+- 没有删除待定代码、没有修改根目录正式数据库、没有切换Docker/NAS。

docs/migration/evidence/slice-05/api-diff.json

@@ -0,0 +1,61 @@
+{
+  "all_equal": true,
+  "endpoints": [
+    {
+      "endpoint": "/api/auction?trade_date=2026-07-29",
+      "original_status": 200,
+      "migrated_status": 200,
+      "original_sha256": "523144cc14d876577b7d518cdf38fd2722b13a8f01ed5d2e20dcc38f6a2624ce",
+      "migrated_sha256": "523144cc14d876577b7d518cdf38fd2722b13a8f01ed5d2e20dcc38f6a2624ce",
+      "equal": true
+    },
+    {
+      "endpoint": "/api/themes?trade_date=2026-07-29",
+      "original_status": 200,
+      "migrated_status": 200,
+      "original_sha256": "95c2ad418f18d877d94ec2a71fe6fafd5e329b069d87a9300f7fcac92d4ba5d1",
+      "migrated_sha256": "95c2ad418f18d877d94ec2a71fe6fafd5e329b069d87a9300f7fcac92d4ba5d1",
+      "equal": true
+    },
+    {
+      "endpoint": "/api/themes/detail?code=885001.TI&trade_date=2026-07-29",
+      "original_status": 400,
+      "migrated_status": 400,
+      "original_sha256": "b11a3314b172d3ad6ba28d969fcd6a9a2a4b49e7d29ed804496a4dfecd2a364a",
+      "migrated_sha256": "b11a3314b172d3ad6ba28d969fcd6a9a2a4b49e7d29ed804496a4dfecd2a364a",
+      "equal": true
+    },
+    {
+      "endpoint": "/api/popularity?trade_date=2026-07-29",
+      "original_status": 200,
+      "migrated_status": 200,
+      "original_sha256": "e62a93c41f7c3f95c3d47f8ccaedaa809563c8dd65c04dd54b164f73e6014e94",
+      "migrated_sha256": "e62a93c41f7c3f95c3d47f8ccaedaa809563c8dd65c04dd54b164f73e6014e94",
+      "equal": true
+    },
+    {
+      "endpoint": "/api/dragon-tiger?trade_date=2026-07-29",
+      "original_status": 200,
+      "migrated_status": 200,
+      "original_sha256": "a3998b935377d5fd0673ec5b9214d1b0a64680d155c61e6cff49d0d5fcf0e843",
+      "migrated_sha256": "a3998b935377d5fd0673ec5b9214d1b0a64680d155c61e6cff49d0d5fcf0e843",
+      "equal": true
+    },
+    {
+      "endpoint": "/api/dragon-tiger/profiles",
+      "original_status": 200,
+      "migrated_status": 200,
+      "original_sha256": "dfb1e534c018ec12fd8d8ee0e7fe0f234e73dc7715d7f0a948f74ef8d12d779a",
+      "migrated_sha256": "dfb1e534c018ec12fd8d8ee0e7fe0f234e73dc7715d7f0a948f74ef8d12d779a",
+      "equal": true
+    },
+    {
+      "endpoint": "/api/seat-aliases",
+      "original_status": 200,
+      "migrated_status": 200,
+      "original_sha256": "2b0fb0a6b3e353c69158d61221c2200e4199d0d60dd0b9d99702a22eaa917a78",
+      "migrated_sha256": "2b0fb0a6b3e353c69158d61221c2200e4199d0d60dd0b9d99702a22eaa917a78",
+      "equal": true
+    }
+  ]
+}

docs/migration/evidence/slice-05/database-diff.json

@@ -0,0 +1,51 @@
+{
+  "all_equal": true,
+  "schema": {
+    "object_count": 62,
+    "original_sha256": "60a4e044f0ddb6502e44b45f65bedc1a0a31d4bd596f386d4ccfe153a6c8ddd1",
+    "migrated_sha256": "60a4e044f0ddb6502e44b45f65bedc1a0a31d4bd596f386d4ccfe153a6c8ddd1",
+    "equal": true
+  },
+  "tables": [
+    {
+      "table": "auction_factors",
+      "original_count": 511914,
+      "migrated_count": 511914,
+      "original_sha256": "3d0470787adaf7c4cf5f15f5ad9fa1d67c8fcd4807285ac264eeacdcf5054cd1",
+      "migrated_sha256": "3d0470787adaf7c4cf5f15f5ad9fa1d67c8fcd4807285ac264eeacdcf5054cd1",
+      "equal": true
+    },
+    {
+      "table": "popularity_factors",
+      "original_count": 232,
+      "migrated_count": 232,
+      "original_sha256": "3f4c61a13ceaa1ed9ecb28f86241a8a478a6757d6a44b46f432f73c0226bba7f",
+      "migrated_sha256": "3f4c61a13ceaa1ed9ecb28f86241a8a478a6757d6a44b46f432f73c0226bba7f",
+      "equal": true
+    },
+    {
+      "table": "lhb_institution_daily",
+      "original_count": 47,
+      "migrated_count": 47,
+      "original_sha256": "1f847eac34d2ba576b429da208667f3b67b591d36ee66800803605bbc447970c",
+      "migrated_sha256": "1f847eac34d2ba576b429da208667f3b67b591d36ee66800803605bbc447970c",
+      "equal": true
+    },
+    {
+      "table": "seat_aliases",
+      "original_count": 0,
+      "migrated_count": 0,
+      "original_sha256": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
+      "migrated_sha256": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
+      "equal": true
+    },
+    {
+      "table": "stock_master",
+      "original_count": 5535,
+      "migrated_count": 5535,
+      "original_sha256": "8656e2d189d3520433fb3552e17998e4bf17bcf6838403cddc6719282b23e792",
+      "migrated_sha256": "8656e2d189d3520433fb3552e17998e4bf17bcf6838403cddc6719282b23e792",
+      "equal": true
+    }
+  ]
+}

docs/migration/保真迁移状态.json

@@ -1,6 +1,6 @@
 {
   "schema_version": 1,
-  "updated_at": "2026-07-31T01:38:00+08:00",
+  "updated_at": "2026-07-31T02:42:00+08:00",
   "status": "active",
   "migration_mode": "behavior_preserving_source_migration",
   "source_of_truth": "current_original_webapp_runtime_and_source",
@@ -9,10 +9,10 @@
   "failed_roots": [
     "next"
   ],
-  "current_slice": "slice-04-ladder-rotation",
+  "current_slice": "slice-06-screener-custom-tracking",
-  "last_completed_slice": "slice-03-sentiment-pools-performance",
+  "last_completed_slice": "slice-05-auction-themes-popularity-dragon-tiger",
-  "last_checkpoint": "xiaobai-preservation-slice-03-20260731",
+  "last_checkpoint": "xiaobai-preservation-slice-05-20260731",
-  "next_action": "capture_slice-04_ladder_rotation_contracts_then_move_original_implementations",
+  "next_action": "capture_slice-06_screener_custom_selection_and_tracking_contracts_then_move_original_implementations",
   "authoritative_documents": [
     "AGENTS.md",
     "docs/migration/原版保真迁移总纲.md",

docs/migration/保真迁移账本.md

@@ -1,6 +1,6 @@
 # 小白复盘保真迁移账本
 
-> 当前状态：正式迁移，切片03“情绪周期、五类股池与涨停表现”已完成
+> 当前状态：正式迁移，切片05“集合竞价、题材库、人气热榜与龙虎榜”已完成
 
 本账本是上下文恢复和人工审计的连续记录。任何迁移提交必须在同一提交中更新本文件及
 `保真迁移状态.json`。
@@ -23,6 +23,8 @@
 | 2026-07-31 | `xiaobai-preservation-slice-01-20260731` | 启动、HTTP、账号、会员与系统管理原实现归位 | 自动差分通过，进入切片02 |
 | 2026-07-31 | `xiaobai-preservation-slice-02-20260731` | 公共行情、搜索、详情、图表与数据适配原实现归位 | 自动与浏览器差分通过，进入切片03 |
 | 2026-07-31 | `xiaobai-preservation-slice-03-20260731` | 情绪周期、五类股池与涨停表现原实现归位 | 自动、API与浏览器差分通过，进入切片04 |
+| 2026-07-31 | `xiaobai-preservation-slice-04-20260731` | 市场天梯与板块轮动原实现归位 | 自动、API与浏览器差分通过，进入切片05 |
+| 2026-07-31 | `xiaobai-preservation-slice-05-20260731` | 集合竞价、题材库、人气热榜与龙虎榜原实现归位 | 自动、API、数据库与浏览器差分通过，进入切片06 |
 
 ## 资产处置登记
 
@@ -40,6 +42,12 @@
 | `sentiment_engine.py` | 情绪周期计算 | 总览、轮动、选股 | 移动并保留兼容别名 | `app/backend/features/sentiment/engine.py` | 文件哈希与原版一致；248项Python与45项Playwright通过 | 已移动 |
 | `DashboardService`情绪及股池原因方法 | 业务服务 | 情绪页、五类股池、涨停表现 | 按职责机械移动 | `app/backend/features/sentiment/`、`app/backend/features/pools/` | 8个方法AST与原版一致；真实API完全一致 | 已移动 |
 | `ReviewDatabase`原因覆盖方法 | 持久化 | 股池原因人工覆盖 | 按职责机械移动 | `app/backend/features/pools/repository.py` | 2个方法AST与原版一致；数据库schema哈希一致 | 已移动 |
+| `DashboardService`板块轮动方法 | 业务服务 | 板块轮动页 | 按职责机械移动 | `app/backend/features/rotation/service.py` | 2个方法AST、真实API与原版一致 | 已移动 |
+| Tushare天梯与轮动构造函数 | 公共数据计算 | 市场天梯、板块轮动 | 原位置保持唯一实现 | `app/backend/data/providers/tushare_client.py` | 2个构造函数AST与原版一致 | 已归位 |
+| `MarketInsightsService` | 共享市场洞察服务 | 集合竞价、题材库、人气热榜 | 整体机械移动，保留唯一共享实现 | `app/backend/features/market/insights.py` | 22个方法AST与原版一致；根级模块为同一类对象别名 | 已移动 |
+| `DashboardService`竞价、题材、人气与龙虎榜方法 | 业务服务 | 切片05四类页面与API | 按职责机械移动 | `app/backend/features/auction/`、`themes/`、`popularity/`、`dragon_tiger/` | 8个方法AST、7个真实API与原版一致 | 已移动 |
+| `ReviewDatabase`竞价、人气与龙虎榜方法 | 持久化 | 市场洞察及后续智能选股 | 按职责机械移动并保持Mixin原接口 | `app/backend/features/auction/repository.py`、`popularity/repository.py`、`dragon_tiger/repository.py` | 7个方法AST一致；62个schema对象及5张关键表逐行一致 | 已移动 |
+| Tushare游资名录与龙虎榜实现 | 公共数据计算 | 龙虎榜与游资档案 | 原位置保持唯一实现 | `app/backend/data/providers/tushare_client.py` | 2个方法AST与原版一致 | 已归位 |
 
 处置只允许：`原样保留`、`移动`、`合并重复`、`待定`、`确认废弃`。
 
@@ -87,6 +95,26 @@
 - 回档：标签`xiaobai-preservation-slice-03-20260731`。
 - 完整证据：`docs/migration/evidence/slice-03/README.md`。
 
+已完成切片：`slice-04-ladder-rotation`。
+
+- 原版基线：提交`b3555d2`，即切片03回档点。
+- 迁移范围：2个板块轮动服务方法；市场天梯继续使用切片02已归位的原Tushare数据构造实现。
+- 兼容边界：`DashboardService`通过`RotationServiceMixin`保持所有原调用；天梯不制造空服务或第二套计算。
+- API与错误：天梯与9日轮动历史JSON完全一致；成分股两版均返回同一Tushare外部失败语义。
+- 验收：252项Python测试、4项切片源码等价测试、45项Playwright测试及两个真实页面流程通过。
+- 回档：标签`xiaobai-preservation-slice-04-20260731`。
+- 完整证据：`docs/migration/evidence/slice-04/README.md`。
+
+已完成切片：`slice-05-auction-themes-popularity-dragon-tiger`。
+
+- 原版基线：提交`814e757`，即切片04回档点。
+- 迁移范围：共享市场洞察服务、竞价/题材/人气入口、龙虎榜与游资档案服务、7个相关持久化方法。
+- 兼容边界：根级`market_insights.py`保留同一类对象别名；竞价与人气共用候选热度逻辑，不复制第二套实现。
+- API与数据库：7个真实API逐字段一致，仅排除每次请求必然变化的`request_id`；62个schema对象与5张关键表完全一致。
+- 验收：258项Python测试、6项切片源码等价测试、45项Playwright测试及四个真实页面流程通过。
+- 回档：标签`xiaobai-preservation-slice-05-20260731`。
+- 完整证据：`docs/migration/evidence/slice-05/README.md`。
+
 ## 决策记录
 
 | 日期 | 决策 | 原因 |