xiaobaifupan/app/backend/data/providers/tushare_client.py

from __future__ import annotations

import json
import math
import re
import urllib.error
import urllib.request
from collections import Counter
from dataclasses import dataclass
from datetime import datetime, time as dt_time, timedelta
from threading import Lock
from typing import Any, ClassVar

from backend.features.sentiment.engine import apply_sentiment_to_dashboard


TUSHARE_URL = "http://api.tushare.pro"


class TushareError(RuntimeError):
    pass


@dataclass
class TushareClient:
    token: str
    timeout: int = 30
    _realtime_reference_cache: ClassVar[dict[str, dict[str, Any]]] = {}
    _realtime_reference_lock: ClassVar[Lock] = Lock()
    _capital_cache: ClassVar[dict[str, dict[str, Any]]] = {}
    _latest_realtime_market: ClassVar[dict[str, dict[str, Any]]] = {}
    _stock_activity_cache: ClassVar[dict[str, dict[str, Any]]] = {}
    _stock_listing_cache: ClassVar[dict[str, Any]] = {}
    _stock_listing_lock: ClassVar[Lock] = Lock()
    _suspension_cache: ClassVar[dict[str, dict[str, str] | None]] = {}
    _suspension_lock: ClassVar[Lock] = Lock()

    def query(
        self,
        api_name: str,
        params: dict[str, Any] | None = None,
        fields: str = "",
    ) -> list[dict[str, Any]]:
        payload = json.dumps(
            {
                "api_name": api_name,
                "token": self.token,
                "params": params or {},
                "fields": fields,
            }
        ).encode("utf-8")
        request = urllib.request.Request(
            TUSHARE_URL,
            data=payload,
            headers={"Content-Type": "application/json", "User-Agent": "XiaobaiReviewWeb/0.2"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=self.timeout) as response:
                result = json.loads(response.read().decode("utf-8"))
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
            raise TushareError(f"Tushare request failed: {exc}") from exc

        if result.get("code") != 0:
            raise TushareError(result.get("msg") or "Tushare returned an unknown error")

        data = result.get("data") or {}
        columns = data.get("fields") or []
        return [dict(zip(columns, item)) for item in data.get("items") or []]

    def dashboard(self, requested_date: str) -> dict[str, Any]:
        trade_date, previous_trade_date = self.resolve_trade_context(requested_date)
        if self.should_use_realtime(requested_date, trade_date):
            return self._realtime_dashboard(
                requested_date,
                trade_date,
                previous_trade_date,
            )

        daily = self._load_daily(trade_date)
        if (
            not daily
            and requested_date == datetime.now().astimezone().strftime("%Y%m%d")
            and trade_date == requested_date
            and datetime.now().astimezone().time().replace(tzinfo=None) >= dt_time(9, 15)
        ):
            return self._realtime_dashboard(
                requested_date,
                trade_date,
                previous_trade_date,
            )
        if not daily:
            raise TushareError(f"No daily data returned for {trade_date}")

        notices: list[str] = []
        try:
            limit_rows = self._load_limit_lists(trade_date)
            previous_limit_rows = self._load_limit_type(previous_trade_date, "U")
            if not limit_rows:
                notices.append("涨跌停高级接口当日数据尚未更新，已使用日线数据推算。")
                limit_rows = self._derive_limits(trade_date, daily)
        except TushareError as exc:
            notices.append(f"涨跌停高级接口不可用，已使用日线数据推算：{exc}")
            limit_rows = self._derive_limits(trade_date, daily)
            previous_daily = self._load_daily(previous_trade_date)
            previous_limit_rows = [
                row for row in self._derive_limits(previous_trade_date, previous_daily)
                if row.get("limit_type") == "U"
            ]

        up_rows = [row for row in limit_rows if row.get("limit_type") == "U"]
        down_rows = [row for row in limit_rows if row.get("limit_type") == "D"]
        broken_rows = [row for row in limit_rows if row.get("limit_type") == "Z"]
        limits = [self._normalize_limit(row, "涨停") for row in up_rows]
        broken = [self._normalize_limit(row, "炸板") for row in broken_rows]
        down_limits = [self._normalize_limit(row, "跌停") for row in down_rows]
        previous_limits = [self._normalize_limit(row, "涨停") for row in previous_limit_rows]
        yesterday_limits = _build_yesterday_performance(
            previous_limits,
            daily,
            limits,
            broken,
            down_limits,
        )
        sectors = _build_sectors(limits)
        previous_sectors = _build_sectors(previous_limits)

        dashboard = {
            "meta": {
                "requested_date": _display_date(requested_date),
                "trade_date": _display_date(trade_date),
                "previous_trade_date": _display_date(previous_trade_date),
                "source": "tushare",
                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
                "notice": "；".join(notices),
            },
            "overview": _build_overview(daily, up_rows, down_rows, broken_rows),
            "limits": limits,
            "broken": broken,
            "down_limits": down_limits,
            "yesterday_limits": yesterday_limits,
            "limit_performance": _build_limit_performance(yesterday_limits),
            "ladders": _build_ladders(limits),
            "sectors": sectors,
            "sector_rotation": _build_sector_rotation(sectors, previous_sectors),
        }
        return apply_sentiment_to_dashboard(dashboard)

    @staticmethod
    def should_use_realtime(requested_date: str, trade_date: str) -> bool:
        """Use rt_k for today's open market until end-of-day datasets settle."""
        now = datetime.now().astimezone()
        today = now.strftime("%Y%m%d")
        return (
            requested_date == today
            and trade_date == today
            and dt_time(9, 15) <= now.time().replace(tzinfo=None) < dt_time(16, 30)
        )

    def _realtime_dashboard(
        self,
        requested_date: str,
        trade_date: str,
        previous_trade_date: str,
    ) -> dict[str, Any]:
        reference = self._load_realtime_reference(trade_date, previous_trade_date)
        basic_rows = list(reference["basic_rows"])
        codes = ",".join(
            str(row.get("ts_code") or "") for row in basic_rows if row.get("ts_code")
        )
        if not codes:
            raise TushareError("No active stock codes available for rt_k")
        quotes = self.query("rt_k", {"ts_code": codes})
        if not quotes:
            raise TushareError(f"No realtime data returned for {trade_date}")

        basic_map = {str(row.get("ts_code") or ""): row for row in basic_rows}
        daily: list[dict[str, Any]] = []
        for quote in quotes:
            close = _number(quote.get("close"))
            previous_close = _number(quote.get("pre_close"))
            if close <= 0 or previous_close <= 0:
                continue
            basic = basic_map.get(str(quote.get("ts_code") or ""), {})
            daily.append(
                {
                    **quote,
                    "trade_date": trade_date,
                    "name": str(quote.get("name") or basic.get("name") or "--").strip(),
                    "industry": basic.get("industry") or "其他",
                    "pct_chg": round((close / previous_close - 1) * 100, 4),
                    "amount_unit": "yuan",
                }
            )
        with self._realtime_reference_lock:
            self._latest_realtime_market[trade_date] = {
                "rows": daily,
                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
            }
            if len(self._latest_realtime_market) > 3:
                oldest = next(iter(self._latest_realtime_market))
                self._latest_realtime_market.pop(oldest, None)

        limit_rows = self._derive_limits(
            trade_date,
            daily,
            price_limits=list(reference["price_limits"]),
            basic_rows=basic_rows,
            previous_limit_rows=list(reference["previous_limit_rows"]),
            capital_rows=list(reference["capital_rows"]),
        )
        previous_limit_rows = list(reference["previous_limit_rows"])
        up_rows = [row for row in limit_rows if row.get("limit_type") == "U"]
        down_rows = [row for row in limit_rows if row.get("limit_type") == "D"]
        broken_rows = [row for row in limit_rows if row.get("limit_type") == "Z"]
        limits = [self._normalize_limit(row, "涨停") for row in up_rows]
        broken = [self._normalize_limit(row, "炸板") for row in broken_rows]
        down_limits = [self._normalize_limit(row, "跌停") for row in down_rows]
        previous_limits = [self._normalize_limit(row, "涨停") for row in previous_limit_rows]
        yesterday_limits = _build_yesterday_performance(
            previous_limits,
            daily,
            limits,
            broken,
            down_limits,
        )
        sectors = _build_sectors(limits)
        previous_sectors = _build_sectors(previous_limits)
        now = datetime.now().astimezone()
        market_status = _realtime_market_status(now.time().replace(tzinfo=None))
        dashboard = {
            "meta": {
                "requested_date": _display_date(requested_date),
                "trade_date": _display_date(trade_date),
                "previous_trade_date": _display_date(previous_trade_date),
                "source": "tushare",
                "mode": "realtime",
                "realtime": True,
                "market_status": market_status,
                "refresh_mode": "manual",
                "auto_refresh": False,
                "quote_count": len(daily),
                "updated_at": now.isoformat(timespec="seconds"),
                "notice": "盘中行情由 Tushare rt_k 实时计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。",
            },
            "overview": _build_overview(daily, up_rows, down_rows, broken_rows),
            "limits": limits,
            "broken": broken,
            "down_limits": down_limits,
            "yesterday_limits": yesterday_limits,
            "limit_performance": _build_limit_performance(yesterday_limits),
            "ladders": _build_ladders(limits),
            "sectors": sectors,
            "sector_rotation": _build_sector_rotation(sectors, previous_sectors),
        }
        return apply_sentiment_to_dashboard(dashboard)

    def _load_realtime_reference(
        self,
        trade_date: str,
        previous_trade_date: str,
    ) -> dict[str, Any]:
        cache_key = f"{trade_date}:{previous_trade_date}"
        with self._realtime_reference_lock:
            cached = self._realtime_reference_cache.get(cache_key)
        if cached:
            return cached

        basic_rows = self.query(
            "stock_basic",
            {"exchange": "", "list_status": "L"},
            "ts_code,name,industry,market,list_date",
        )
        price_limits = self.query(
            "stk_limit",
            {"trade_date": trade_date},
            "ts_code,trade_date,up_limit,down_limit",
        )
        previous_limit_rows = self._load_limit_type(previous_trade_date, "U")
        capital_rows = self.query(
            "daily_basic",
            {"trade_date": previous_trade_date},
            "ts_code,trade_date,total_share,float_share,free_share,total_mv,circ_mv",
        )
        if not basic_rows or not price_limits:
            raise TushareError(f"Realtime reference data is incomplete for {trade_date}")
        result = {
            "basic_rows": basic_rows,
            "price_limits": price_limits,
            "previous_limit_rows": previous_limit_rows,
            "capital_rows": capital_rows,
        }
        with self._realtime_reference_lock:
            self._realtime_reference_cache[cache_key] = result
            if len(self._realtime_reference_cache) > 3:
                oldest = next(iter(self._realtime_reference_cache))
                self._realtime_reference_cache.pop(oldest, None)
        return result

    def realtime_stock_quote(
        self,
        ts_code: str,
        reference_date: str = "",
    ) -> dict[str, Any]:
        rows = self.query("rt_k", {"ts_code": ts_code})
        if not rows:
            raise TushareError(f"No realtime quote returned for {ts_code}")
        row = rows[0]
        close = _number(row.get("close"))
        previous_close = _number(row.get("pre_close"))
        if close <= 0 or previous_close <= 0:
            raise TushareError(f"Realtime quote is unavailable for {ts_code}")

        basic: dict[str, Any] = {}
        with self._realtime_reference_lock:
            references = list(self._realtime_reference_cache.values())
        for reference in reversed(references):
            basic = next(
                (
                    item for item in reference.get("basic_rows") or []
                    if str(item.get("ts_code") or "") == ts_code
                ),
                {},
            )
            if basic:
                break
        if not basic:
            basics = self.query(
                "stock_basic",
                {"ts_code": ts_code},
                "ts_code,name,industry,market,list_date",
            )
            basic = basics[0] if basics else {}
        capital = self._latest_capital(ts_code, reference_date)
        float_share = _number(capital.get("float_share"))
        # rt_k volume is shares; daily_basic float_share is reported in 10k shares.
        turnover_rate = _number(row.get("vol")) / float_share / 100 if float_share else 0
        market_date = reference_date or datetime.now().astimezone().strftime("%Y%m%d")
        self._ensure_realtime_market_cache(market_date)
        with self._realtime_reference_lock:
            market_rows = list((self._latest_realtime_market.get(market_date) or {}).get("rows") or [])
            references = list(self._realtime_reference_cache.values())
        capital_map: dict[str, dict[str, Any]] = {}
        for reference in reversed(references):
            capital_map = {
                str(item.get("ts_code") or ""): item
                for item in reference.get("capital_rows") or []
            }
            if capital_map:
                break
        market_amounts = [_number(item.get("amount")) for item in market_rows if _number(item.get("amount")) > 0]
        amount_percentile = _value_percentile(_number(row.get("amount")), market_amounts)
        market_turnovers = []
        for item in market_rows:
            item_capital = capital_map.get(str(item.get("ts_code") or ""), {})
            item_float_share = _number(item_capital.get("float_share"))
            if item_float_share:
                market_turnovers.append(_number(item.get("vol")) / item_float_share / 100)
        market_turnover = (
            sum(market_turnovers) / len(market_turnovers) if market_turnovers else 0
        )
        turnover_relative = turnover_rate / market_turnover if market_turnover else 0
        activity = self._stock_activity_metrics(
            ts_code,
            market_date,
            _number(row.get("vol")) / 100,
        )
        return {
            "code": ts_code.split(".")[0],
            "ts_code": ts_code,
            "name": str(row.get("name") or basic.get("name") or "--").strip(),
            "sector": basic.get("industry") or "其他",
            "price": round(close, 3),
            "change": round((close / previous_close - 1) * 100, 4),
            "open": round(_number(row.get("open")), 3),
            "high": round(_number(row.get("high")), 3),
            "low": round(_number(row.get("low")), 3),
            "previous_close": round(previous_close, 3),
            "amount_billion": round(_number(row.get("amount")) / 100000000, 3),
            "volume": _number(row.get("vol")),
            "trade_count": int(_number(row.get("num"))),
            "turnover_rate": round(turnover_rate, 4),
            "market_turnover_rate": round(market_turnover, 4),
            "turnover_relative": round(turnover_relative, 4),
            "amount_percentile": round(amount_percentile * 100, 2),
            "volume_activity_ratio": activity.get("volume_activity_ratio", 0),
            "activity_history_date": activity.get("history_trade_date", ""),
            "activity_source": activity.get("source", "unavailable"),
            "float_share_10k": float_share,
            "capital_trade_date": str(capital.get("trade_date") or ""),
            "turnover_source": "rt_volume/latest_float_share" if float_share else "unavailable",
            "data_source": "tushare",
            "realtime": True,
        }

    def _stock_activity_metrics(
        self,
        ts_code: str,
        reference_date: str,
        current_volume_lots: float,
    ) -> dict[str, Any]:
        cache_key = f"{ts_code}:{reference_date}"
        with self._realtime_reference_lock:
            history = self._stock_activity_cache.get(cache_key)
        if history is None:
            try:
                end = datetime.strptime(reference_date, "%Y%m%d")
            except ValueError:
                end = datetime.now().astimezone().replace(tzinfo=None)
            rows = self.query(
                "daily",
                {
                    "ts_code": ts_code,
                    "start_date": (end - timedelta(days=30)).strftime("%Y%m%d"),
                    "end_date": reference_date,
                },
                "ts_code,trade_date,vol,amount",
            )
            completed = [
                item for item in rows
                if str(item.get("trade_date") or "") < reference_date and _number(item.get("vol")) > 0
            ]
            completed.sort(key=lambda item: str(item.get("trade_date") or ""))
            recent = completed[-5:]
            history = {
                "average_volume_lots": (
                    sum(_number(item.get("vol")) for item in recent) / len(recent)
                    if recent else 0
                ),
                "history_trade_date": str(recent[-1].get("trade_date") or "") if recent else "",
            }
            with self._realtime_reference_lock:
                self._stock_activity_cache[cache_key] = history
                if len(self._stock_activity_cache) > 256:
                    oldest = next(iter(self._stock_activity_cache))
                    self._stock_activity_cache.pop(oldest, None)
        average_volume = _number(history.get("average_volume_lots"))
        progress = _trading_session_progress(datetime.now().astimezone().time().replace(tzinfo=None))
        expected_volume = average_volume * progress
        ratio = current_volume_lots / expected_volume if expected_volume else 0
        return {
            **history,
            "volume_activity_ratio": round(ratio, 4),
            "session_progress": round(progress, 4),
            "source": "rt_volume/5d_average_at_same_progress" if expected_volume else "unavailable",
        }

    def realtime_factor_snapshot(self, requested_date: str) -> dict[str, Any]:
        trade_date, previous_trade_date = self.resolve_trade_context(requested_date)
        reference = self._load_realtime_reference(trade_date, previous_trade_date)
        codes = [
            str(row.get("ts_code") or "")
            for row in reference.get("basic_rows") or []
            if row.get("ts_code")
        ]
        quotes = self.query("rt_k", {"ts_code": ",".join(codes)}, "")
        capital_map = {
            str(row.get("ts_code") or ""): row
            for row in reference.get("capital_rows") or []
        }
        rows = []
        for quote in quotes:
            ts_code = str(quote.get("ts_code") or "")
            close = _number(quote.get("close"))
            previous_close = _number(quote.get("pre_close"))
            if not ts_code or close <= 0 or previous_close <= 0:
                continue
            capital = capital_map.get(ts_code, {})
            float_share = _number(capital.get("float_share"))
            rows.append(
                {
                    "ts_code": ts_code,
                    "trade_date": trade_date,
                    "open": _number(quote.get("open")),
                    "high": _number(quote.get("high")),
                    "low": _number(quote.get("low")),
                    "close": close,
                    "pct_chg": (close / previous_close - 1) * 100,
                    "vol": _number(quote.get("vol")) / 100,
                    "amount": _number(quote.get("amount")),
                    "turnover_rate": (
                        _number(quote.get("vol")) / float_share / 100 if float_share else 0
                    ),
                    "capital_trade_date": str(capital.get("trade_date") or ""),
                }
            )
        if not rows:
            raise TushareError(f"No realtime factor snapshot returned for {trade_date}")
        return {
            "trade_date": trade_date,
            "previous_trade_date": previous_trade_date,
            "source": "tushare_rt_k",
            "realtime": True,
            "rows": rows,
        }

    def _ensure_realtime_market_cache(self, requested_date: str) -> list[dict[str, Any]]:
        with self._realtime_reference_lock:
            cached = list(
                (self._latest_realtime_market.get(requested_date) or {}).get("rows") or []
            )
        if cached:
            return cached
        trade_date, previous_trade_date = self.resolve_trade_context(requested_date)
        if trade_date != requested_date:
            return []
        reference = self._load_realtime_reference(trade_date, previous_trade_date)
        codes = [
            str(row.get("ts_code") or "")
            for row in reference.get("basic_rows") or []
            if row.get("ts_code")
        ]
        quotes = self.query("rt_k", {"ts_code": ",".join(codes)}, "")
        rows = [
            row for row in quotes
            if _number(row.get("close")) > 0 and _number(row.get("pre_close")) > 0
        ]
        with self._realtime_reference_lock:
            self._latest_realtime_market[trade_date] = {
                "rows": rows,
                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
            }
        return rows

    def _latest_capital(self, ts_code: str, reference_date: str = "") -> dict[str, Any]:
        end_date = reference_date or datetime.now().astimezone().strftime("%Y%m%d")
        cache_key = f"{ts_code}:{end_date}"
        with self._realtime_reference_lock:
            cached = self._capital_cache.get(cache_key)
        if cached:
            return cached
        try:
            end = datetime.strptime(end_date, "%Y%m%d")
        except ValueError:
            end = datetime.now().astimezone().replace(tzinfo=None)
            end_date = end.strftime("%Y%m%d")
        start_date = (end - timedelta(days=20)).strftime("%Y%m%d")
        rows = self.query(
            "daily_basic",
            {"ts_code": ts_code, "start_date": start_date, "end_date": end_date},
            "ts_code,trade_date,turnover_rate,volume_ratio,total_share,float_share,"
            "free_share,total_mv,circ_mv",
        )
        rows.sort(key=lambda item: str(item.get("trade_date") or ""))
        result = rows[-1] if rows else {}
        with self._realtime_reference_lock:
            self._capital_cache[cache_key] = result
            if len(self._capital_cache) > 256:
                oldest = next(iter(self._capital_cache))
                self._capital_cache.pop(oldest, None)
        return result

    def market_indices(self, requested_date: str, lookback_days: int = 45) -> dict[str, Any]:
        trade_date, _ = self.resolve_trade_context(requested_date)
        end = datetime.strptime(trade_date, "%Y%m%d")
        start_date = (end - timedelta(days=max(30, lookback_days * 2))).strftime("%Y%m%d")
        index_names = {
            "000001.SH": "上证指数",
            "399001.SZ": "深证成指",
            "399006.SZ": "创业板指",
        }
        indices = []
        for ts_code, name in index_names.items():
            rows = self.query(
                "index_daily",
                {"ts_code": ts_code, "start_date": start_date, "end_date": trade_date},
                "ts_code,trade_date,close,pct_chg,vol,amount",
            )
            rows.sort(key=lambda item: str(item.get("trade_date") or ""))
            if not rows:
                continue
            latest = rows[-1]
            close = _number(latest.get("close"))
            close_5d = _number(rows[-6].get("close")) if len(rows) >= 6 else _number(rows[0].get("close"))
            close_20d = _number(rows[-21].get("close")) if len(rows) >= 21 else _number(rows[0].get("close"))
            indices.append(
                {
                    "ts_code": ts_code,
                    "name": name,
                    "trade_date": str(latest.get("trade_date") or trade_date),
                    "close": close,
                    "pct_chg": round(_number(latest.get("pct_chg")), 3),
                    "return_5d": round((close / close_5d - 1) * 100, 3) if close_5d else 0,
                    "return_20d": round((close / close_20d - 1) * 100, 3) if close_20d else 0,
                    "amount_billion": round(_number(latest.get("amount")) / 100000, 2),
                }
            )
        if not indices:
            raise TushareError(f"No index data returned for {trade_date}")
        return {
            "trade_date": trade_date,
            "source": "tushare",
            "realtime": False,
            "precise": all(item["trade_date"] == trade_date for item in indices),
            "indices": indices,
            "aggregate": {
                "average_pct_chg": round(sum(item["pct_chg"] for item in indices) / len(indices), 3),
                "average_return_5d": round(sum(item["return_5d"] for item in indices) / len(indices), 3),
                "average_return_20d": round(sum(item["return_20d"] for item in indices) / len(indices), 3),
            },
        }

    def realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
        trade_date, _ = self.resolve_trade_context(requested_date)
        index_names = {
            "000001.SH": "上证指数",
            "399001.SZ": "深证成指",
            "399006.SZ": "创业板指",
        }
        rows = self.query("rt_idx_k", {"ts_code": ",".join(index_names)}, "")
        row_map = {str(row.get("ts_code") or ""): row for row in rows}
        indices = []
        for ts_code, name in index_names.items():
            row = row_map.get(ts_code)
            if not row:
                continue
            close = _number(row.get("close"))
            previous_close = _number(row.get("pre_close"))
            if close <= 0 or previous_close <= 0:
                continue
            history = self.query(
                "index_daily",
                {
                    "ts_code": ts_code,
                    "start_date": (datetime.strptime(trade_date, "%Y%m%d") - timedelta(days=20)).strftime("%Y%m%d"),
                    "end_date": trade_date,
                },
                "ts_code,trade_date,close,pct_chg",
            )
            history.sort(key=lambda item: str(item.get("trade_date") or ""))
            previous_closes = [
                _number(item.get("close")) for item in history
                if str(item.get("trade_date") or "") < trade_date and _number(item.get("close")) > 0
            ]
            close_5d = previous_closes[-5] if len(previous_closes) >= 5 else previous_closes[0] if previous_closes else previous_close
            indices.append(
                {
                    "ts_code": ts_code,
                    "name": str(row.get("name") or name).strip(),
                    "trade_date": trade_date,
                    "close": close,
                    "pct_chg": round((close / previous_close - 1) * 100, 3),
                    "return_5d": round((close / close_5d - 1) * 100, 3) if close_5d else 0,
                    "amount_billion": round(_number(row.get("amount")) / 100000000, 2),
                }
            )
        if len(indices) != len(index_names):
            raise TushareError("Realtime index quotes are incomplete")
        return {
            "trade_date": trade_date,
            "source": "tushare_rt_idx_k",
            "realtime": True,
            "precise": True,
            "indices": indices,
            "aggregate": {
                "average_pct_chg": round(sum(item["pct_chg"] for item in indices) / len(indices), 3),
                "average_return_5d": round(sum(item["return_5d"] for item in indices) / len(indices), 3),
                "average_return_20d": 0,
            },
        }

    def sw_stock_industry(self, ts_code: str, trade_date: str) -> dict[str, Any]:
        """Return the Shenwan industry active for a stock on trade_date."""
        rows = []
        for is_new in ("Y", "N"):
            rows.extend(
                self.query(
                    "index_member_all",
                    {"ts_code": ts_code, "is_new": is_new},
                    "l1_code,l1_name,l2_code,l2_name,l3_code,l3_name,"
                    "ts_code,name,in_date,out_date,is_new",
                )
            )
        rows = _reconcile_membership_rows(rows)
        matched = [row for row in rows if _membership_active_on(row, trade_date)]
        if not matched:
            matched = [
                row for row in rows
                if row.get("is_new") == "Y"
                and str(row.get("in_date") or "") <= trade_date
            ]
        if not matched:
            raise TushareError(f"No Shenwan industry returned for {ts_code}")
        row = max(
            matched,
            key=lambda item: (
                str(item.get("in_date") or ""),
                1 if item.get("is_new") == "Y" else 0,
                str(item.get("l3_code") or item.get("l2_code") or ""),
            ),
        )
        return {
            "l1_code": str(row.get("l1_code") or ""),
            "l1_name": str(row.get("l1_name") or ""),
            "l2_code": str(row.get("l2_code") or ""),
            "l2_name": str(row.get("l2_name") or ""),
            "l3_code": str(row.get("l3_code") or ""),
            "l3_name": str(row.get("l3_name") or ""),
            "in_date": str(row.get("in_date") or ""),
            "out_date": str(row.get("out_date") or ""),
            "is_new": str(row.get("is_new") or ""),
        }

    def sw_sector_snapshot(
        self,
        ts_code: str,
        requested_date: str,
        realtime_expected: bool = False,
        allow_realtime_close: bool = False,
    ) -> dict[str, Any]:
        """Build the single Shenwan L2 sector context used by heaven trend."""
        trade_date, previous_trade_date = self.resolve_trade_context(requested_date)
        industry = self.sw_stock_industry(ts_code, trade_date)
        sector_code = str(industry.get("l2_code") or "")
        if not sector_code:
            raise TushareError(f"Shenwan L2 code is unavailable for {ts_code}")
        members = self._sw_sector_members(sector_code, trade_date)
        if not members:
            raise TushareError(f"No Shenwan members returned for {sector_code}")
        raw_member_count = len(members)
        members, excluded_members = _filter_members_by_listing(
            members,
            self._stock_listing_reference(),
            trade_date,
        )
        if not members:
            raise TushareError(f"No listed Shenwan members returned for {sector_code}")

        if realtime_expected:
            snapshot = self._sw_realtime_sector_snapshot(
                industry,
                members,
                trade_date,
                previous_trade_date,
                finalized=False,
            )
            snapshot.update({
                "raw_member_count": raw_member_count,
                "excluded_member_count": len(excluded_members),
                "excluded_members": excluded_members,
            })
            return snapshot

        member_set = {str(item.get("ts_code") or "") for item in members}
        member_names = {
            str(item.get("ts_code") or ""): str(item.get("name") or "")
            for item in members
        }
        member_rows = [
            row for row in self._load_daily(trade_date)
            if str(row.get("ts_code") or "") in member_set
        ]
        quoted_codes = {str(row.get("ts_code") or "") for row in member_rows}
        suspended_members = self._confirmed_suspended_members(
            members, quoted_codes, trade_date
        )
        up_count = sum(_number(row.get("pct_chg")) > 0 for row in member_rows)
        down_count = sum(_number(row.get("pct_chg")) < 0 for row in member_rows)
        leader = max(member_rows, key=lambda row: _number(row.get("pct_chg")), default={})
        leader_code = str(leader.get("ts_code") or "")
        equal_change = (
            sum(_number(row.get("pct_chg")) for row in member_rows) / len(member_rows)
            if member_rows else 0
        )
        coverage = len(member_rows) / max(len(members), 1) * 100
        explained_count = len(member_rows) + len(suspended_members)
        explained_coverage = explained_count / max(len(members), 1) * 100
        coverage_issue = _sector_coverage_issue(
            len(members),
            len(member_rows),
            explained_coverage,
            explained_count,
        )
        inner_precise = not coverage_issue
        inner_error = coverage_issue
        amount_billion = sum(_number(row.get("amount")) for row in member_rows) / 100000
        rows = self.query(
            "sw_daily",
            {"ts_code": sector_code, "trade_date": trade_date},
            "ts_code,trade_date,name,close,pct_change,vol,amount,pe,pb,float_mv,total_mv",
        )
        daily = rows[0] if rows else {}
        actual_trade_date = str(daily.get("trade_date") or "")
        outer_precise = actual_trade_date == trade_date
        outer_error = "" if outer_precise else (
            f"No Shenwan daily returned for {sector_code} on {trade_date}"
        )
        if not outer_precise and allow_realtime_close:
            try:
                return self._sw_realtime_sector_snapshot(
                    industry,
                    members,
                    trade_date,
                    previous_trade_date,
                    finalized=True,
                )
            except TushareError as exc:
                outer_error = f"{outer_error}; realtime close fallback failed: {exc}"

        official_change = _number(daily.get("pct_change")) if outer_precise else None
        return {
            "code": sector_code,
            "name": industry.get("l2_name") or daily.get("name") or sector_code,
            "leader": str(leader.get("name") or member_names.get(leader_code) or "--"),
            "leader_code": leader_code,
            "leading_pct": round(_number(leader.get("pct_chg")), 3),
            "change": round(official_change, 3) if official_change is not None else None,
            "member_equal_change": round(equal_change, 3),
            "turnover_rate": 0,
            "up_count": up_count,
            "down_count": down_count,
            "flat_count": len(member_rows) - up_count - down_count,
            "member_count": len(members),
            "raw_member_count": raw_member_count,
            "excluded_member_count": len(excluded_members),
            "excluded_members": excluded_members,
            "quote_count": len(member_rows),
            "coverage": round(coverage, 1),
            "explained_count": explained_count,
            "explained_coverage": round(explained_coverage, 1),
            "suspended_count": len(suspended_members),
            "suspended_members": suspended_members,
            "strength": round(max(0, min(100, 50 + (official_change if official_change is not None else equal_change) * 5)), 1),
            "amount_billion": round(amount_billion, 2),
            "count": 0,
            "max_streak": 0,
            "source": "tushare_sw_daily+member_daily" if outer_precise else "tushare_member_daily",
            "inner_source": "tushare_member_daily",
            "outer_source": "tushare_sw_daily" if outer_precise else "unavailable",
            "taxonomy": "sw_l2",
            "industry": industry,
            "trade_date": trade_date,
            "inner_trade_date": trade_date if member_rows else "",
            "outer_trade_date": actual_trade_date,
            "realtime": False,
            "finalized": True,
            "inner_precise": inner_precise,
            "outer_precise": outer_precise,
            "precise": inner_precise and outer_precise,
            "inner_error": inner_error,
            "outer_error": outer_error,
            "schema_version": 6,
            "methodology": "外显使用申万二级行业官方日线；内核独立使用当日成分日线宽度与等权涨跌聚合",
        }

    def _sw_sector_members(
        self,
        sector_code: str,
        trade_date: str,
    ) -> list[dict[str, Any]]:
        rows = []
        for is_new in ("Y", "N"):
            rows.extend(
                self.query(
                    "index_member_all",
                    {"l2_code": sector_code, "is_new": is_new},
                    "l2_code,l2_name,ts_code,name,in_date,out_date,is_new",
                )
            )
        deduped: dict[str, dict[str, Any]] = {}
        for row in _reconcile_membership_rows(rows):
            code = str(row.get("ts_code") or "")
            if code and _membership_active_on(row, trade_date):
                current = deduped.get(code)
                if current is None or str(row.get("in_date") or "") > str(current.get("in_date") or ""):
                    deduped[code] = row
        return list(deduped.values())

    def sw_sector_members(self, sector_code: str, trade_date: str) -> list[dict[str, Any]]:
        """Return constituents active in a Shenwan L2 industry on the target date."""
        return self._sw_sector_members(sector_code, trade_date)

    def _stock_listing_reference(self) -> dict[str, dict[str, Any]]:
        now = datetime.now().astimezone()
        with self._stock_listing_lock:
            loaded_at = self._stock_listing_cache.get("loaded_at")
            cached = self._stock_listing_cache.get("rows")
            if (
                isinstance(loaded_at, datetime)
                and isinstance(cached, dict)
                and now - loaded_at < timedelta(hours=6)
            ):
                return cached

        rows: list[dict[str, Any]] = []
        try:
            for status in ("L", "D", "P"):
                rows.extend(self.query(
                    "stock_basic",
                    {"list_status": status},
                    "ts_code,name,list_status,list_date,delist_date",
                ))
        except TushareError:
            # Unknown status must remain in the denominator so a reference-data
            # failure cannot silently improve coverage.
            return {}
        reference = {
            str(row.get("ts_code") or ""): dict(row)
            for row in rows
            if row.get("ts_code")
        }
        with self._stock_listing_lock:
            type(self)._stock_listing_cache = {"loaded_at": now, "rows": reference}
        return reference

    def _confirmed_suspended_members(
        self,
        members: list[dict[str, Any]],
        quoted_codes: set[str],
        trade_date: str,
    ) -> list[dict[str, str]]:
        suspended: list[dict[str, str]] = []
        for member in members:
            code = str(member.get("ts_code") or "")
            if not code or code in quoted_codes:
                continue
            cache_key = f"{trade_date}:{code}"
            with self._suspension_lock:
                cached = self._suspension_cache.get(cache_key, "missing")
            if cached == "missing":
                try:
                    rows = self.query(
                        "suspend_d",
                        {"ts_code": code},
                        "ts_code,suspend_date,resume_date,ann_date,suspend_reason,reason_type",
                    )
                except TushareError:
                    rows = []
                active = [
                    row for row in rows
                    if str(row.get("suspend_date") or "")
                    and str(row.get("suspend_date") or "") <= trade_date
                    and (
                        not str(row.get("resume_date") or "")
                        or trade_date < str(row.get("resume_date") or "")
                    )
                ]
                row = max(
                    active,
                    key=lambda item: str(item.get("suspend_date") or ""),
                    default=None,
                )
                cached = ({
                    "ts_code": code,
                    "name": str(member.get("name") or code),
                    "suspend_date": str(row.get("suspend_date") or ""),
                    "resume_date": str(row.get("resume_date") or ""),
                    "reason": str(row.get("suspend_reason") or row.get("reason_type") or "已确认停牌"),
                } if row else None)
                with self._suspension_lock:
                    type(self)._suspension_cache[cache_key] = cached
            if isinstance(cached, dict):
                suspended.append(cached)
        return suspended

    def _sw_realtime_sector_snapshot(
        self,
        industry: dict[str, Any],
        members: list[dict[str, Any]],
        trade_date: str,
        previous_trade_date: str,
        finalized: bool = False,
    ) -> dict[str, Any]:
        sector_code = str(industry.get("l2_code") or "")
        sw_rows = self.query(
            "rt_sw_k",
            {"ts_code": sector_code},
            "ts_code,name,trade_time,close,pre_close,high,open,low,vol,amount,pct_change",
        )
        sw_row = sw_rows[0] if sw_rows else {}
        trade_time = str(sw_row.get("trade_time") or "")
        quote_date = trade_time[:10].replace("-", "")
        quote_clock = trade_time[11:19] if len(trade_time) >= 19 else ""
        outer_precise = bool(sw_row and quote_date == trade_date)
        if finalized and (not quote_clock or quote_clock < "15:00:00"):
            outer_precise = False
        official_change = _number(sw_row.get("pct_change"))
        if not official_change:
            close = _number(sw_row.get("close"))
            pre_close = _number(sw_row.get("pre_close"))
            official_change = (close / pre_close - 1) * 100 if close and pre_close else 0
        if not outer_precise:
            official_change = None
        outer_error = ""
        if not sw_row:
            outer_error = f"No Shenwan realtime index returned for {sector_code}"
        elif quote_date != trade_date:
            outer_error = f"Shenwan realtime index date is {quote_date or 'unknown'}, expected {trade_date}"
        elif finalized and (not quote_clock or quote_clock < "15:00:00"):
            outer_error = f"Shenwan realtime index is not a close snapshot ({trade_time})"

        valid: list[dict[str, Any]] = []
        codes: list[str] = []
        reference: dict[str, Any] = {}
        inner_error = ""
        try:
            reference = self._load_realtime_reference(trade_date, previous_trade_date)
            active_codes = {
                str(row.get("ts_code") or "")
                for row in reference.get("basic_rows") or []
                if row.get("ts_code")
            }
            codes = [
                str(row.get("ts_code") or "")
                for row in members
                if str(row.get("ts_code") or "") in active_codes
            ]
            if codes:
                quotes = self.query("rt_k", {"ts_code": ",".join(codes)}, "")
                for row in quotes:
                    close = _number(row.get("close"))
                    previous_close = _number(row.get("pre_close"))
                    if close <= 0 or previous_close <= 0:
                        continue
                    valid.append({**row, "change": (close / previous_close - 1) * 100})
            else:
                inner_error = f"No active Shenwan members returned for {sector_code}"
        except TushareError as exc:
            inner_error = str(exc)

        coverage = len(valid) / max(len(codes), 1) * 100
        valid_codes = {str(item.get("ts_code") or "") for item in valid}
        suspended_members = self._confirmed_suspended_members(
            members, valid_codes, trade_date
        )
        explained_count = len(valid) + len(suspended_members)
        explained_coverage = explained_count / max(len(codes), 1) * 100
        coverage_issue = _sector_coverage_issue(
            len(codes), len(valid), explained_coverage, explained_count
        )
        inner_precise = bool(codes) and not coverage_issue
        if not inner_precise and not inner_error:
            inner_error = coverage_issue or "申万实时有效成分为空"
        up_count = sum(item["change"] > 0 for item in valid)
        down_count = sum(item["change"] < 0 for item in valid)
        leader = max(valid, key=lambda item: item["change"], default={})
        leader_code = str(leader.get("ts_code") or "")
        member_names = {
            str(item.get("ts_code") or ""): str(item.get("name") or "")
            for item in members
        }
        equal_change = sum(item["change"] for item in valid) / len(valid) if valid else 0
        amount_billion = sum(_number(item.get("amount")) for item in valid) / 100000000
        try:
            self._ensure_realtime_market_cache(trade_date)
            with self._realtime_reference_lock:
                market_rows = list(
                    (self._latest_realtime_market.get(trade_date) or {}).get("rows") or []
                )
        except TushareError as exc:
            market_rows = []
            inner_precise = False
            inner_error = inner_error or str(exc)
        capital_map = {
            str(item.get("ts_code") or ""): item
            for item in reference.get("capital_rows") or []
        }
        sector_turnovers = []
        for item in valid:
            capital = capital_map.get(str(item.get("ts_code") or ""), {})
            float_share = _number(capital.get("float_share"))
            if float_share:
                sector_turnovers.append(_number(item.get("vol")) / float_share / 100)
        market_turnovers = []
        for item in market_rows:
            capital = capital_map.get(str(item.get("ts_code") or ""), {})
            float_share = _number(capital.get("float_share"))
            if float_share:
                market_turnovers.append(_number(item.get("vol")) / float_share / 100)
        average_turnover = sum(sector_turnovers) / len(sector_turnovers) if sector_turnovers else 0
        market_turnover = sum(market_turnovers) / len(market_turnovers) if market_turnovers else 0
        relative_turnover = average_turnover / market_turnover if market_turnover else 0
        if not relative_turnover:
            inner_precise = False
            inner_error = inner_error or "Shenwan member relative turnover is unavailable"
        return {
            "code": sector_code,
            "name": str(industry.get("l2_name") or sw_row.get("name") or ""),
            "leader": str(leader.get("name") or member_names.get(leader_code) or "--").strip(),
            "leader_code": leader_code,
            "leading_pct": round(_number(leader.get("change")), 3),
            "change": round(official_change, 3) if official_change is not None else None,
            "member_equal_change": round(equal_change, 3),
            "turnover_rate": round(average_turnover, 4),
            "market_turnover_rate": round(market_turnover, 4),
            "relative_turnover": round(relative_turnover, 4),
            "up_count": up_count,
            "down_count": down_count,
            "flat_count": len(valid) - up_count - down_count,
            "member_count": len(codes),
            "quote_count": len(valid),
            "coverage": round(coverage, 1),
            "explained_count": explained_count,
            "explained_coverage": round(explained_coverage, 1),
            "suspended_count": len(suspended_members),
            "suspended_members": suspended_members,
            "strength": round(max(0, min(100, 50 + (official_change if official_change is not None else equal_change) * 5)), 1),
            "amount_billion": round(amount_billion, 2),
            "count": sum(item["change"] >= 9.5 for item in valid),
            "max_streak": 0,
            "source": "tushare_rt_sw_k+sw_members_rt_k",
            "inner_source": "tushare_sw_members+rt_k",
            "outer_source": "tushare_rt_sw_k",
            "taxonomy": "sw_l2",
            "industry": industry,
            "trade_date": trade_date,
            "inner_trade_date": trade_date if valid else "",
            "outer_trade_date": quote_date,
            "trade_time": trade_time,
            "realtime": True,
            "finalized": finalized,
            "inner_precise": inner_precise,
            "outer_precise": outer_precise,
            "precise": inner_precise and outer_precise,
            "inner_error": inner_error,
            "outer_error": outer_error,
            "schema_version": 6,
            "methodology": "外显使用申万官方 rt_sw_k；内核独立使用申万成分 rt_k 宽度与相对换手聚合",
        }

    def sector_snapshot(
        self,
        identifier: str,
        requested_date: str,
        realtime_expected: bool | None = None,
    ) -> dict[str, Any]:
        trade_date, _ = self.resolve_trade_context(requested_date)
        raw_identifier = identifier.strip()
        if not raw_identifier:
            raise TushareError("Sector identifier is empty")
        errors = []
        now = datetime.now().astimezone()
        if realtime_expected is None:
            realtime_expected = (
                trade_date == now.strftime("%Y%m%d")
                and dt_time(9, 15) <= now.time().replace(tzinfo=None) <= dt_time(15, 5)
            )
        try:
            dc_params = {"trade_date": trade_date}
            if re.fullmatch(r"[A-Z0-9.]+", raw_identifier.upper()) and "." in raw_identifier:
                dc_params["ts_code"] = raw_identifier.upper()
            else:
                dc_params["name"] = raw_identifier
            dc_rows = self.query(
                "dc_index",
                dc_params,
                "ts_code,trade_date,name,leading,leading_code,pct_change,leading_pct,"
                "total_mv,turnover_rate,up_num,down_num",
            )
            if not dc_rows and "name" in dc_params:
                dc_rows = self.query(
                    "dc_index",
                    {"trade_date": trade_date},
                    "ts_code,trade_date,name,leading,leading_code,pct_change,leading_pct,"
                    "total_mv,turnover_rate,up_num,down_num",
                )
            dc_row = _match_sector_row(dc_rows, raw_identifier)
            if dc_row and not realtime_expected:
                change = _number(dc_row.get("pct_change"))
                actual_trade_date = str(dc_row.get("trade_date") or "")
                return {
                    "code": dc_row.get("ts_code") or "",
                    "name": dc_row.get("name") or raw_identifier,
                    "leader": dc_row.get("leading") or "--",
                    "leader_code": dc_row.get("leading_code") or "",
                    "leading_pct": _number(dc_row.get("leading_pct")),
                    "change": change,
                    "turnover_rate": _number(dc_row.get("turnover_rate")),
                    "up_count": int(_number(dc_row.get("up_num"))),
                    "down_count": int(_number(dc_row.get("down_num"))),
                    "total_mv": _number(dc_row.get("total_mv")),
                    "strength": round(max(0, min(100, 50 + change * 5)), 1),
                    "amount_billion": 0,
                    "count": 0,
                    "max_streak": 0,
                    "source": "tushare_dc",
                    "trade_date": actual_trade_date,
                    "realtime": False,
                    "precise": actual_trade_date == trade_date,
                }
        except TushareError as exc:
            errors.append(f"DC: {exc}")

        ts_code = raw_identifier.upper()
        if re.fullmatch(r"\d{6}", ts_code):
            ts_code = f"{ts_code}.TI"
        try:
            if re.fullmatch(r"\d{6}\.TI", ts_code):
                index_rows = self.query(
                    "ths_index",
                    {"ts_code": ts_code},
                    "ts_code,name,count,exchange,list_date,type",
                )
            else:
                index_rows = self.query(
                    "ths_index",
                    {},
                    "ts_code,name,count,exchange,list_date,type",
                )
            basic = _match_sector_row(index_rows, raw_identifier)
            if not basic:
                raise TushareError(f"No THS sector returned for {raw_identifier}")
        except TushareError as exc:
            errors.append(f"THS: {exc}")
            raise TushareError("; ".join(errors)) from exc
        actual_code = str(basic.get("ts_code") or ts_code)
        if realtime_expected:
            try:
                realtime_sector = self._realtime_sector_snapshot(actual_code, basic, trade_date)
                if realtime_sector:
                    return realtime_sector
            except TushareError as exc:
                errors.append(f"THS realtime members: {exc}")
        daily_rows = self.query(
            "ths_daily",
            {"ts_code": actual_code, "trade_date": trade_date},
            "ts_code,trade_date,close,pct_change,vol,turnover_rate,total_mv,float_mv",
        )
        daily = daily_rows[0] if daily_rows else {}
        actual_trade_date = str(daily.get("trade_date") or "")
        change = _number(daily.get("pct_change"))
        return {
            "code": actual_code,
            "name": basic.get("name") or raw_identifier,
            "leader": "--",
            "change": change,
            "leading_pct": change,
            "turnover_rate": _number(daily.get("turnover_rate")),
            "up_count": 0,
            "down_count": 0,
            "strength": round(max(0, min(100, 50 + change * 5)), 1),
            "amount_billion": 0,
            "count": 0,
            "max_streak": 0,
            "source": "tushare_ths",
            "trade_date": actual_trade_date,
            "realtime": False,
            "precise": actual_trade_date == trade_date,
        }

    def _realtime_sector_snapshot(
        self,
        sector_code: str,
        basic: dict[str, Any],
        trade_date: str,
    ) -> dict[str, Any] | None:
        members = self.query(
            "ths_member",
            {"ts_code": sector_code, "is_new": "Y"},
            "ts_code,con_code,con_name,is_new",
        )
        codes = [str(row.get("con_code") or "") for row in members if row.get("con_code")]
        if not codes:
            return None
        quotes = self.query("rt_k", {"ts_code": ",".join(codes)}, "")
        valid = []
        for row in quotes:
            close = _number(row.get("close"))
            previous_close = _number(row.get("pre_close"))
            if close <= 0 or previous_close <= 0:
                continue
            valid.append(
                {
                    **row,
                    "change": (close / previous_close - 1) * 100,
                }
            )
        minimum = max(1, math.ceil(len(codes) * 0.9))
        if len(valid) < minimum:
            raise TushareError(
                f"Realtime sector coverage is insufficient ({len(valid)}/{len(codes)})"
            )
        up_count = sum(item["change"] > 0 for item in valid)
        down_count = sum(item["change"] < 0 for item in valid)
        flat_count = len(valid) - up_count - down_count
        leader = max(valid, key=lambda item: item["change"])
        change = sum(item["change"] for item in valid) / len(valid)
        amount_billion = sum(_number(item.get("amount")) for item in valid) / 100000000
        self._ensure_realtime_market_cache(trade_date)
        with self._realtime_reference_lock:
            references = list(self._realtime_reference_cache.values())
            market_rows = list((self._latest_realtime_market.get(trade_date) or {}).get("rows") or [])
        capital_map: dict[str, dict[str, Any]] = {}
        for reference in reversed(references):
            capital_map = {
                str(item.get("ts_code") or ""): item
                for item in reference.get("capital_rows") or []
            }
            if capital_map:
                break
        sector_turnovers = []
        for item in valid:
            capital = capital_map.get(str(item.get("ts_code") or ""), {})
            float_share = _number(capital.get("float_share"))
            if float_share:
                sector_turnovers.append(_number(item.get("vol")) / float_share / 100)
        market_turnovers = []
        for item in market_rows:
            capital = capital_map.get(str(item.get("ts_code") or ""), {})
            float_share = _number(capital.get("float_share"))
            if float_share:
                market_turnovers.append(_number(item.get("vol")) / float_share / 100)
        average_turnover = sum(sector_turnovers) / len(sector_turnovers) if sector_turnovers else 0
        market_turnover = sum(market_turnovers) / len(market_turnovers) if market_turnovers else 0
        relative_turnover = average_turnover / market_turnover if market_turnover else 0
        return {
            "code": sector_code,
            "name": basic.get("name") or sector_code,
            "leader": str(leader.get("name") or "--").strip(),
            "leader_code": leader.get("ts_code") or "",
            "leading_pct": round(leader["change"], 3),
            "change": round(change, 3),
            "turnover_rate": round(average_turnover, 4),
            "market_turnover_rate": round(market_turnover, 4),
            "relative_turnover": round(relative_turnover, 4),
            "up_count": up_count,
            "down_count": down_count,
            "flat_count": flat_count,
            "member_count": len(codes),
            "quote_count": len(valid),
            "coverage": round(len(valid) / len(codes) * 100, 1),
            "strength": round(max(0, min(100, 50 + change * 5)), 1),
            "amount_billion": round(amount_billion, 2),
            "count": sum(item["change"] >= 9.5 for item in valid),
            "max_streak": 0,
            "source": "tushare_rt_ths_members",
            "trade_date": trade_date,
            "realtime": True,
            "precise": True,
            "methodology": "同花顺行业最新成分股的 rt_k 等权涨跌、宽度与成交额聚合",
        }

    def hot_money_profiles(self) -> dict[str, Any]:
        rows = self.query("hm_list", {}, "name,desc,orgs")
        profiles: list[dict[str, Any]] = []
        seen_names: set[str] = set()
        for row in rows:
            name = str(row.get("name") or "").strip()
            if not name or name in seen_names:
                continue
            seen_names.add(name)
            description = _text(row.get("desc"))
            organization_text = _text(row.get("orgs"))
            parsed_organizations: Any = None
            if organization_text.startswith("["):
                try:
                    parsed_organizations = json.loads(organization_text)
                except json.JSONDecodeError:
                    parsed_organizations = None
            organization_parts = (
                parsed_organizations
                if isinstance(parsed_organizations, list)
                else re.split(r"[,，;；\n]+", organization_text)
            )
            organizations = list(dict.fromkeys(
                _text(part)
                for part in organization_parts
                if _text(part)
            ))
            profiles.append(
                {
                    "id": f"hot-money-profile-{len(profiles) + 1}",
                    "name": name,
                    "description": description,
                    "organizations": organizations,
                    "organization_count": len(organizations),
                }
            )
        return {
            "meta": {
                "source": "tushare",
                "status": "success" if profiles else "empty",
                "schema_version": 1,
                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
                "notice": "",
            },
            "summary": {
                "profile_count": len(profiles),
                "described_count": sum(bool(item["description"]) for item in profiles),
                "organization_count": sum(item["organization_count"] for item in profiles),
            },
            "profiles": profiles,
        }

    def dragon_tiger(self, requested_date: str) -> dict[str, Any]:
        trade_date, _ = self.resolve_trade_context(requested_date)
        detail_rows = self.query(
            "hm_detail",
            {"trade_date": trade_date},
            "trade_date,ts_code,ts_name,buy_amount,sell_amount,net_amount,"
            "hm_name,hm_orgs,tag",
        )

        notices: list[str] = []
        try:
            directory_rows = self.query("hm_list", {}, "name,desc,orgs")
        except TushareError as exc:
            directory_rows = []
            notices.append(f"游资名录暂不可用：{exc}")
        directory = {
            str(row.get("name") or "").strip(): {
                "description": _text(row.get("desc")),
                "orgs": _text(row.get("orgs")),
            }
            for row in directory_rows
            if str(row.get("name") or "").strip()
        }

        # 个股龙虎榜仅用于补充涨幅和上榜原因，不参与游资身份识别。
        try:
            top_rows = self.query(
                "top_list",
                {"trade_date": trade_date},
                "trade_date,ts_code,name,pct_change,reason",
            )
        except TushareError as exc:
            top_rows = []
            notices.append(f"个股龙虎榜辅助信息暂不可用：{exc}")
        stock_context: dict[str, dict[str, Any]] = {}
        for row in top_rows:
            ts_code = str(row.get("ts_code") or "")
            if ts_code and ts_code not in stock_context:
                stock_context[ts_code] = row

        groups: dict[str, dict[str, Any]] = {}
        for row in detail_rows:
            trader_name = str(row.get("hm_name") or "未命名游资").strip()
            ts_code = str(row.get("ts_code") or "").strip()
            stock = stock_context.get(ts_code, {})
            directory_item = directory.get(trader_name, {})
            seat_name = _text(row.get("hm_orgs")) or directory_item.get("orgs") or "--"
            buy = round(_number(row.get("buy_amount")) / 1000000, 2)
            sell = round(_number(row.get("sell_amount")) / 1000000, 2)
            net_buy = round(_number(row.get("net_amount")) / 1000000, 2)
            group = groups.setdefault(
                trader_name,
                {
                    "name": trader_name,
                    "description": directory_item.get("description") or "",
                    "directory_orgs": directory_item.get("orgs") or "",
                    "identity_type": "trader",
                    "identity_source": "tushare_hm",
                    "recognized": True,
                    "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
            if seat_name != "--":
                group["seat_names"].add(seat_name)
            code = ts_code.split(".")[0]
            if code:
                group["stock_codes"].add(code)
            group["operations"].append(
                {
                    "code": code,
                    "ts_code": ts_code,
                    "name": row.get("ts_name") or stock.get("name") or "--",
                    "change": (
                        _number(stock.get("pct_change"))
                        if stock.get("pct_change") is not None
                        else None
                    ),
                    "direction": "买入" if net_buy > 0 else "卖出" if net_buy < 0 else "持平",
                    "buy_million": buy,
                    "sell_million": sell,
                    "net_buy_million": net_buy,
                    "seat_name": seat_name,
                    "seat_alias": trader_name,
                    "tag": _text(row.get("tag")) or "--",
                    "reason": _text(stock.get("reason")) or "--",
                }
            )

        traders = list(groups.values())
        traders.sort(key=lambda item: abs(item["net_buy_million"]), reverse=True)
        for index, group in enumerate(traders, start=1):
            group["id"] = f"hot-money-{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
            )

        operation_count = sum(item["operation_count"] for item in traders)
        active_stocks = {
            operation["code"] for item in traders for operation in item["operations"]
            if operation["code"]
        }
        net_buy_total = round(sum(item["net_buy_million"] for item in traders), 2)
        status = "success" if detail_rows else "partial" if top_rows else "empty"
        if not detail_rows:
            notices.insert(
                0,
                f"当日有 {len(stock_context)} 只股票上榜，但未返回可识别的游资每日明细。"
                if top_rows
                else "该交易日未返回龙虎榜或游资每日明细。",
            )
        return {
            "meta": {
                "requested_date": _display_date(requested_date),
                "trade_date": _display_date(trade_date),
                "source": "tushare",
                "status": status,
                "schema_version": 3,
                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
                "notice": "；".join(notices),
            },
            "summary": {
                "trader_count": len(traders),
                "identity_count": len(traders),
                "operation_count": operation_count,
                "active_stock_count": len(active_stocks),
                "seat_net_buy_million": net_buy_total,
                "unclassified_count": 0,
                "directory_count": len(directory),
                "official_stock_count": len(stock_context),
            },
            "traders": traders,
            "unclassified_seats": [],
            "rows": [],
        }

    def stock_detail(self, ts_code: str, requested_date: str) -> dict[str, Any]:
        trade_date, _ = self.resolve_trade_context(requested_date)
        end = datetime.strptime(trade_date, "%Y%m%d")
        start_date = (end - timedelta(days=190)).strftime("%Y%m%d")
        daily = self.query(
            "daily",
            {"ts_code": ts_code, "start_date": start_date, "end_date": trade_date},
            "ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
        )
        factors = self.query(
            "adj_factor",
            {"ts_code": ts_code, "start_date": start_date, "end_date": trade_date},
            "ts_code,trade_date,adj_factor",
        )
        basics = self.query(
            "stock_basic",
            {"ts_code": ts_code},
            "ts_code,symbol,name,area,industry,market,list_date",
        )
        daily_basics = self.query(
            "daily_basic",
            {"ts_code": ts_code, "trade_date": trade_date},
            "ts_code,trade_date,turnover_rate,volume_ratio,total_mv,circ_mv",
        )
        moneyflow = self.query(
            "moneyflow",
            {"ts_code": ts_code, "trade_date": trade_date},
            "ts_code,trade_date,buy_sm_amount,sell_sm_amount,buy_md_amount,sell_md_amount,"
            "buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount",
        )
        factor_map = {row["trade_date"]: _number(row.get("adj_factor"), 1) for row in factors}
        latest_factor = max(factor_map.values(), default=1) or 1
        prices = []
        for row in sorted(daily, key=lambda item: item.get("trade_date", ""))[-90:]:
            factor = factor_map.get(row.get("trade_date"), latest_factor)
            ratio = factor / latest_factor
            prices.append(
                {
                    "trade_date": _display_date(str(row.get("trade_date", ""))),
                    "open": round(_number(row.get("open")) * ratio, 3),
                    "high": round(_number(row.get("high")) * ratio, 3),
                    "low": round(_number(row.get("low")) * ratio, 3),
                    "close": round(_number(row.get("close")) * ratio, 3),
                    "change": _number(row.get("pct_chg")),
                    "volume": _number(row.get("vol")),
                    "amount_billion": round(_number(row.get("amount")) / 100000, 2),
                }
            )
        flow = moneyflow[0] if moneyflow else {}
        basic = basics[0] if basics else {}
        daily_basic = daily_basics[0] if daily_basics else {}
        latest = prices[-1] if prices else {}
        actual_trade_date = max(
            (str(row.get("trade_date") or "") for row in daily),
            default=trade_date,
        ) or trade_date
        return {
            "meta": {
                "requested_date": _display_date(requested_date),
                "trade_date": _display_date(actual_trade_date),
                "source": "tushare",
                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
                "notice": "",
            },
            "stock": {
                "code": ts_code.split(".")[0],
                "ts_code": ts_code,
                "name": basic.get("name") or "--",
                "industry": basic.get("industry") or "其他",
                "area": basic.get("area") or "--",
                "market": basic.get("market") or "--",
                "list_date": _display_date(str(basic.get("list_date") or "")),
                "price": latest.get("close", 0),
                "change": latest.get("change", 0),
                "turnover_rate": _number(daily_basic.get("turnover_rate")),
                "volume_ratio": _number(daily_basic.get("volume_ratio")),
                "amount_billion": latest.get("amount_billion", 0),
            },
            "prices": prices,
            "moneyflow": {
                "net_million": round(_number(flow.get("net_mf_amount")) / 100, 2),
                "large_million": round(
                    (_number(flow.get("buy_lg_amount")) + _number(flow.get("buy_elg_amount"))
                     - _number(flow.get("sell_lg_amount")) - _number(flow.get("sell_elg_amount"))) / 100,
                    2,
                ),
                "medium_million": round(
                    (_number(flow.get("buy_md_amount")) - _number(flow.get("sell_md_amount"))) / 100,
                    2,
                ),
                "small_million": round(
                    (_number(flow.get("buy_sm_amount")) - _number(flow.get("sell_sm_amount"))) / 100,
                    2,
                ),
            },
        }

    def stock_intraday(self, ts_code: str, requested_date: str) -> dict[str, Any]:
        trade_date, _ = self.resolve_trade_context(requested_date)
        display_date = _display_date(trade_date)
        rows = self.query(
            "stk_mins",
            {
                "ts_code": ts_code,
                "freq": "1min",
                "start_date": f"{display_date} 09:00:00",
                "end_date": f"{display_date} 15:30:00",
            },
            "ts_code,trade_time,open,close,high,low,vol,amount",
        )
        points = []
        for row in sorted(rows, key=lambda item: str(item.get("trade_time") or "")):
            trade_time = str(row.get("trade_time") or "")
            if not trade_time:
                continue
            points.append(
                {
                    "time": trade_time[-8:-3] if len(trade_time) >= 8 else trade_time,
                    "open": round(_number(row.get("open")), 3),
                    "high": round(_number(row.get("high")), 3),
                    "low": round(_number(row.get("low")), 3),
                    "close": round(_number(row.get("close")), 3),
                    "volume": _number(row.get("vol")),
                    "amount": _number(row.get("amount")),
                }
            )
        return {"trade_date": display_date, "points": points}

    def resolve_trade_context(self, requested: str) -> tuple[str, str]:
        requested_rows = self.query(
            "trade_cal",
            {"exchange": "SSE", "start_date": requested, "end_date": requested},
            "cal_date,is_open,pretrade_date",
        )
        if not requested_rows:
            trade_date = requested
        else:
            row = requested_rows[0]
            trade_date = row["cal_date"] if row.get("is_open") == 1 else row.get("pretrade_date", requested)

        resolved_rows = self.query(
            "trade_cal",
            {"exchange": "SSE", "start_date": trade_date, "end_date": trade_date},
            "cal_date,is_open,pretrade_date",
        )
        previous = resolved_rows[0].get("pretrade_date") if resolved_rows else ""
        return trade_date, previous or trade_date

    def _load_daily(self, trade_date: str) -> list[dict[str, Any]]:
        return self.query(
            "daily",
            {"trade_date": trade_date},
            "ts_code,trade_date,open,high,low,close,pct_chg,amount",
        )

    def _load_limit_type(self, trade_date: str, limit_type: str) -> list[dict[str, Any]]:
        fields = (
            "trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount,"
            "float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
            "open_times,up_stat,limit_times"
        )
        rows = self.query(
            "limit_list_d",
            {"trade_date": trade_date, "limit_type": limit_type},
            fields,
        )
        for row in rows:
            row["limit_type"] = limit_type
            row["amount_unit"] = "yuan"
        return rows

    def _load_limit_lists(self, trade_date: str) -> list[dict[str, Any]]:
        rows: list[dict[str, Any]] = []
        for limit_type in ("U", "D", "Z"):
            rows.extend(self._load_limit_type(trade_date, limit_type))
        return rows

    def _derive_limits(
        self,
        trade_date: str,
        daily: list[dict[str, Any]],
        price_limits: list[dict[str, Any]] | None = None,
        basic_rows: list[dict[str, Any]] | None = None,
        previous_limit_rows: list[dict[str, Any]] | None = None,
        capital_rows: list[dict[str, Any]] | None = None,
    ) -> list[dict[str, Any]]:
        if price_limits is None:
            price_limits = self.query(
                "stk_limit",
                {"trade_date": trade_date},
                "ts_code,trade_date,up_limit,down_limit",
            )
        limit_map = {row["ts_code"]: row for row in price_limits}
        if basic_rows is None:
            basic_rows = self.query(
                "stock_basic",
                {"list_status": "L"},
                "ts_code,name,industry",
            )
        basic_map = {row["ts_code"]: row for row in basic_rows}
        previous_limit_map = {
            str(row.get("ts_code") or ""): row for row in (previous_limit_rows or [])
        }
        capital_map = {
            str(row.get("ts_code") or ""): row for row in (capital_rows or [])
        }

        result: list[dict[str, Any]] = []
        for row in daily:
            bounds = limit_map.get(row.get("ts_code"))
            if not bounds or row.get("close") is None:
                continue
            limit_type = ""
            if _prices_equal(row["close"], bounds.get("up_limit")):
                limit_type = "U"
            elif _prices_equal(row["close"], bounds.get("down_limit")):
                limit_type = "D"
            elif _prices_equal(row.get("high"), bounds.get("up_limit")):
                limit_type = "Z"
            if not limit_type:
                continue
            basic = basic_map.get(row["ts_code"], {})
            previous_limit = previous_limit_map.get(str(row.get("ts_code") or ""), {})
            streak = (
                max(1, int(_number(previous_limit.get("limit_times"), 1)) + 1)
                if limit_type == "U" and previous_limit
                else 1
            )
            item = {
                    **row,
                    "name": basic.get("name", "--"),
                    "industry": basic.get("industry") or "其他",
                    "limit_type": limit_type,
                    "limit_times": streak,
                    "open_times": 1 if limit_type == "Z" else 0,
                    "amount_unit": row.get("amount_unit") or "thousand_yuan",
                }
            if row.get("amount_unit") == "yuan":
                capital = capital_map.get(str(row.get("ts_code") or ""), {})
                if not capital and capital_rows is None:
                    capital = self._latest_capital(str(row.get("ts_code") or ""), trade_date)
                float_share = _number(capital.get("float_share"))
                item["turnover_ratio"] = (
                    _number(row.get("vol")) / float_share / 100 if float_share else 0
                )
                item["turnover_source"] = (
                    "rt_volume/latest_float_share" if float_share else "unavailable"
                )
                item["capital_trade_date"] = str(capital.get("trade_date") or "")
            result.append(item)
        return result

    @staticmethod
    def _normalize_limit(row: dict[str, Any], status: str) -> dict[str, Any]:
        amount = _number(row.get("amount"))
        if row.get("amount_unit") == "thousand_yuan":
            amount_billion = amount / 100000
        else:
            amount_billion = amount / 100000000
        return {
            "code": str(row.get("ts_code", "")).split(".")[0],
            "ts_code": row.get("ts_code", ""),
            "name": row.get("name") or "--",
            "price": _number(row.get("close")),
            "change": _number(row.get("pct_chg")),
            "sector": row.get("industry") or "其他",
            "reason": row.get("industry") or "待补充",
            "first_time": _display_time(row.get("first_time")),
            "last_time": _display_time(row.get("last_time")),
            "open_times": int(_number(row.get("open_times"))),
            "streak": max(1, int(_number(row.get("limit_times"), 1))),
            "turnover_rate": _number(row.get("turnover_ratio")),
            "turnover_source": row.get("turnover_source") or "provider",
            "capital_trade_date": row.get("capital_trade_date") or "",
            "amount_billion": round(amount_billion, 2),
            "seal_amount_million": round(_number(row.get("fd_amount")) / 10000, 0),
            "float_mv_billion": round(_number(row.get("float_mv")) / 100000000, 1),
            "status": status,
        }


def _number(value: Any, default: float = 0.0) -> float:
    try:
        number = float(value)
        return number if math.isfinite(number) else default
    except (TypeError, ValueError):
        return default


def _text(value: Any) -> str:
    if isinstance(value, (list, tuple, set)):
        return "、".join(str(item).strip() for item in value if str(item).strip())
    return str(value or "").strip()


def _filter_members_by_listing(
    members: list[dict[str, Any]],
    listing_reference: dict[str, dict[str, Any]],
    trade_date: str,
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
    eligible: list[dict[str, Any]] = []
    excluded: list[dict[str, str]] = []
    for member in members:
        code = str(member.get("ts_code") or "")
        listing = listing_reference.get(code)
        if not listing:
            eligible.append(member)
            continue
        list_date = str(listing.get("list_date") or "")
        delist_date = str(listing.get("delist_date") or "")
        reason = ""
        effective_date = ""
        if delist_date and delist_date <= trade_date:
            reason = "目标日期前已退市"
            effective_date = delist_date
        elif list_date and list_date > trade_date:
            reason = "目标日期尚未上市"
            effective_date = list_date
        if not reason:
            eligible.append(member)
            continue
        excluded.append({
            "ts_code": code,
            "name": str(member.get("name") or listing.get("name") or code),
            "reason": reason,
            "effective_date": effective_date,
        })
    return eligible, excluded


def _sector_coverage_issue(
    member_count: int,
    quote_count: int,
    coverage: float | None = None,
    explained_count: int | None = None,
) -> str:
    members = max(0, int(member_count or 0))
    quotes = max(0, min(int(quote_count or 0), members))
    if members <= 0:
        if coverage is not None and float(coverage) >= 90:
            return ""
        if coverage is not None:
            return "行业成分行情覆盖率低于90%"
        return "申万有效成分为空"
    explained = quotes if explained_count is None else max(
        quotes, min(int(explained_count or 0), members)
    )
    actual_coverage = (
        float(coverage)
        if coverage is not None
        else explained / members * 100
    )
    missing = members - explained
    if members <= 7 and missing:
        return f"小型行业有效成分状态仅确认 {explained}/{members}，要求全部可解释"
    if members <= 20 and (actual_coverage < 90 or missing > 1):
        return f"中型行业有效成分状态仅确认 {explained}/{members}，要求覆盖率至少90%且最多缺1只"
    if members > 20 and actual_coverage < 90:
        return f"行业有效成分状态仅确认 {explained}/{members}，覆盖率低于90%"
    return ""


def _membership_active_on(row: dict[str, Any], trade_date: str) -> bool:
    start = str(row.get("in_date") or "")
    end = str(row.get("out_date") or "")
    return (not start or start <= trade_date) and (not end or end > trade_date)


def _reconcile_membership_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Merge duplicate Y/N membership rows before evaluating their date interval."""
    reconciled: dict[tuple[str, str, str, str, str], dict[str, Any]] = {}
    for raw in rows:
        row = dict(raw)
        key = (
            str(row.get("ts_code") or ""),
            str(row.get("l1_code") or ""),
            str(row.get("l2_code") or ""),
            str(row.get("l3_code") or ""),
            str(row.get("in_date") or ""),
        )
        current = reconciled.get(key)
        if current is None:
            reconciled[key] = row
            continue
        current_end = str(current.get("out_date") or "")
        candidate_end = str(row.get("out_date") or "")
        if candidate_end and not current_end:
            current["out_date"] = candidate_end
            current["is_new"] = row.get("is_new") or current.get("is_new")
        for field, value in row.items():
            if not current.get(field) and value not in (None, ""):
                current[field] = value
    return list(reconciled.values())


def _match_sector_row(rows: list[dict[str, Any]], identifier: str) -> dict[str, Any] | None:
    if not rows:
        return None
    target = identifier.strip().upper()
    code_match = next(
        (row for row in rows if str(row.get("ts_code") or "").strip().upper() == target),
        None,
    )
    if code_match:
        return code_match

    def normalized(value: Any) -> str:
        text = str(value or "").strip().replace(" ", "")
        for suffix in ("板块", "概念", "行业"):
            text = text.removesuffix(suffix)
        aliases = {
            "元器件": "元件",
            "电子元器件": "元件",
        }
        return aliases.get(text, text)

    target_name = normalized(identifier)
    exact = [row for row in rows if normalized(row.get("name")) == target_name]
    if exact:
        return min(exact, key=_sector_match_priority)
    fuzzy = [
        row for row in rows
        if target_name and (
            target_name in normalized(row.get("name"))
            or normalized(row.get("name")) in target_name
        )
    ]
    return min(
        fuzzy,
        key=lambda row: (len(normalized(row.get("name"))), *_sector_match_priority(row)),
    ) if fuzzy else None


def _sector_match_priority(row: dict[str, Any]) -> tuple[int, int, int]:
    code = str(row.get("ts_code") or "")
    exchange = str(row.get("exchange") or "").upper()
    return (
        0 if exchange == "A" else 1,
        0 if code.startswith("881") else 1,
        0 if _number(row.get("count")) > 0 else 1,
    )


def _prices_equal(left: Any, right: Any) -> bool:
    if left is None or right is None:
        return False
    return abs(_number(left) - _number(right)) < 0.005


def _value_percentile(value: float, population: list[float]) -> float:
    valid = sorted(item for item in population if item >= 0)
    if not valid:
        return 0.0
    below = sum(item < value for item in valid)
    equal = sum(item == value for item in valid)
    return (below + equal * 0.5) / len(valid)


def _trading_session_progress(current_time: dt_time) -> float:
    morning_start = dt_time(9, 30)
    morning_end = dt_time(11, 30)
    afternoon_start = dt_time(13, 0)
    afternoon_end = dt_time(15, 0)
    if current_time <= morning_start:
        return 0.05
    if current_time <= morning_end:
        minutes = (current_time.hour * 60 + current_time.minute) - (9 * 60 + 30)
        return max(0.05, min(0.5, minutes / 240))
    if current_time < afternoon_start:
        return 0.5
    if current_time <= afternoon_end:
        minutes = (current_time.hour * 60 + current_time.minute) - 13 * 60
        return max(0.5, min(1.0, 0.5 + minutes / 240))
    return 1.0


def _display_time(value: Any) -> str:
    raw = str(value or "").replace(":", "").zfill(6)
    if not raw.strip("0"):
        return "--"
    return f"{raw[:2]}:{raw[2:4]}:{raw[4:6]}"


def _display_date(value: str) -> str:
    return f"{value[:4]}-{value[4:6]}-{value[6:8]}" if len(value) == 8 else value


def _realtime_market_status(current_time: dt_time) -> str:
    if current_time < dt_time(9, 25):
        return "pre_open"
    if current_time < dt_time(9, 30):
        return "auction"
    if current_time <= dt_time(11, 30) or dt_time(13, 0) <= current_time <= dt_time(15, 0):
        return "trading"
    if current_time < dt_time(13, 0):
        return "lunch_break"
    return "closed"


def _build_overview(
    daily: list[dict[str, Any]],
    up_rows: list[dict[str, Any]],
    down_rows: list[dict[str, Any]],
    broken_rows: list[dict[str, Any]],
) -> dict[str, Any]:
    up_count = sum(1 for row in daily if _number(row.get("pct_chg")) > 0)
    down_count = sum(1 for row in daily if _number(row.get("pct_chg")) < 0)
    flat_count = len(daily) - up_count - down_count
    amount_billion = sum(
        _number(row.get("amount"))
        / (100000000 if row.get("amount_unit") == "yuan" else 100000)
        for row in daily
    )
    limit_count = len(up_rows)
    broken_count = len(broken_rows)
    seal_rate = round(limit_count / max(limit_count + broken_count, 1) * 100, 1)
    return {
        "up_count": up_count,
        "down_count": down_count,
        "flat_count": flat_count,
        "limit_up_count": limit_count,
        "limit_down_count": len(down_rows),
        "broken_count": broken_count,
        "amount_billion": round(amount_billion, 1),
        "seal_rate": seal_rate,
    }


def _build_ladders(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    groups: dict[int, list[dict[str, Any]]] = {}
    for row in rows:
        groups.setdefault(int(row.get("streak") or 1), []).append(row)
    return [
        {
            "level": level,
            "label": "首板" if level == 1 else f"{level}板",
            "count": len(stocks),
            "stocks": sorted(stocks, key=lambda item: item.get("first_time") or "99:99:99"),
        }
        for level, stocks in sorted(groups.items(), reverse=True)
    ]


def _build_sectors(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    counts = Counter(row.get("sector") or "其他" for row in rows)
    result: list[dict[str, Any]] = []
    for name, count in counts.most_common(20):
        stocks = [row for row in rows if (row.get("sector") or "其他") == name]
        max_streak = max(item.get("streak", 1) for item in stocks)
        leader = max(stocks, key=lambda item: (item.get("streak", 1), item.get("amount_billion", 0)))
        result.append(
            {
                "name": name,
                "count": count,
                "strength": min(100, 44 + count * 8 + max_streak * 5),
                "amount_billion": round(sum(item.get("amount_billion", 0) for item in stocks), 1),
                "leader": leader.get("name", "--"),
                "change": round(sum(item.get("change", 0) for item in stocks) / count, 2),
                "max_streak": max_streak,
            }
        )
    return result


def _build_yesterday_performance(
    previous_limits: list[dict[str, Any]],
    daily: list[dict[str, Any]],
    current_limits: list[dict[str, Any]],
    current_broken: list[dict[str, Any]],
    current_down: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    daily_map = {str(row.get("ts_code", "")).split(".")[0]: row for row in daily}
    limit_map = {row["code"]: row for row in current_limits}
    broken_codes = {row["code"] for row in current_broken}
    down_codes = {row["code"] for row in current_down}
    result = []
    for previous in previous_limits:
        code = previous["code"]
        daily_row = daily_map.get(code, {})
        current = limit_map.get(code)
        if current:
            outcome = "晋级"
        elif code in broken_codes:
            outcome = "炸板"
        elif code in down_codes:
            outcome = "跌停"
        else:
            outcome = "断板"
        result.append(
            {
                "code": code,
                "name": previous["name"],
                "prior_streak": previous.get("streak", 1),
                "current_streak": current.get("streak", 0) if current else 0,
                "current_change": _number(daily_row.get("pct_chg")),
                "current_price": _number(daily_row.get("close")),
                "sector": previous.get("sector", "其他"),
                "reason": previous.get("reason", "待补充"),
                "outcome": outcome,
            }
        )
    return result


def _build_limit_performance(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    result = []
    for level in sorted({int(row.get("prior_streak") or 1) for row in rows}, reverse=True):
        group = [row for row in rows if int(row.get("prior_streak") or 1) == level]
        advanced = sum(row.get("outcome") == "晋级" for row in group)
        positive = sum(_number(row.get("current_change")) > 0 for row in group)
        result.append(
            {
                "level": level,
                "label": "昨日首板" if level == 1 else f"昨日{level}板",
                "count": len(group),
                "advanced": advanced,
                "advance_rate": round(advanced / len(group) * 100, 1),
                "positive_rate": round(positive / len(group) * 100, 1),
                "average_change": round(sum(_number(row.get("current_change")) for row in group) / len(group), 2),
            }
        )
    return result


def _build_sector_rotation(
    current: list[dict[str, Any]], previous: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    previous_map = {row["name"]: row for row in previous}
    result = []
    for index, sector in enumerate(current, start=1):
        previous_count = int(previous_map.get(sector["name"], {}).get("count", 0))
        delta = int(sector["count"]) - previous_count
        result.append(
            {
                **sector,
                "rank": index,
                "previous_count": previous_count,
                "delta": delta,
                "trend": "升温" if delta > 0 else "降温" if delta < 0 else "持平",
            }
        )
    return result