"""Dashboard Alpha homepage query services."""

from __future__ import annotations

import logging
from collections import defaultdict
from dataclasses import dataclass
from datetime import date
from types import SimpleNamespace
from typing import Any

from django.utils import timezone as django_timezone

from apps.account.application.repository_provider import get_portfolio_repository
from apps.account.application.use_cases import GetSizingContextUseCase
from apps.alpha.application.pool_resolver import (
    ALPHA_POOL_MODE_PRICE_COVERED,
    PortfolioAlphaPoolResolver,
)
from apps.alpha.application.services import AlphaService
from apps.alpha.application.trade_dates import resolve_recent_closed_trade_date
from apps.alpha.domain.entities import AlphaPoolScope, AlphaResult
from apps.dashboard.application.navigation import (
    build_decision_workspace_url,
    build_exit_user_action_label,
    normalize_exit_user_action,
)
from apps.dashboard.application.repository_provider import (
    AlphaRecommendationHistoryRepository,
    DashboardAlphaContextRepository,
    get_alpha_recommendation_history_repository,
    get_dashboard_alpha_context_repository,
)
from apps.decision_rhythm.application.repository_provider import (
    get_portfolio_transition_plan_repository,
    get_unified_recommendation_repository,
)
from apps.signal.application.repository_provider import get_signal_repository
from apps.simulated_trading.application.query_services import list_user_position_payloads
from apps.strategy.domain.services import DecisionPolicyEngine, PreTradeRiskGate, SizingEngine
from apps.task_monitor.application.tracking import record_pending_task

logger = logging.getLogger(__name__)

ALPHA_SCOPE_GENERAL = "general"
ALPHA_SCOPE_PORTFOLIO = "portfolio"
ALPHA_SCOPE_CHOICES = {ALPHA_SCOPE_GENERAL, ALPHA_SCOPE_PORTFOLIO}


def normalize_alpha_scope(raw_value: str | None) -> str:
    """Normalize Dashboard Alpha scope mode."""
    normalized = str(raw_value or "").strip().lower()
    if normalized in ALPHA_SCOPE_CHOICES:
        return normalized
    return ALPHA_SCOPE_PORTFOLIO


@dataclass(frozen=True)
class AlphaHomepageData:
    """Homepage Alpha payload."""

    pool: dict[str, Any]
    meta: dict[str, Any]
    top_candidates: list[dict[str, Any]]
    actionable_candidates: list[dict[str, Any]]
    exit_watchlist: list[dict[str, Any]]
    exit_watch_summary: dict[str, Any]
    pending_requests: list[dict[str, Any]]
    recent_runs: list[dict[str, Any]]
    history_run_id: int | None


class AlphaHomepageQuery:
    """Build the homepage Alpha candidate/ranking view."""

    def __init__(
        self,
        *,
        history_repo: AlphaRecommendationHistoryRepository | None = None,
        context_repo: DashboardAlphaContextRepository | None = None,
    ) -> None:
        self.history_repo = history_repo or get_alpha_recommendation_history_repository()
        self.context_repo = context_repo or get_dashboard_alpha_context_repository()
        self.portfolio_repo = get_portfolio_repository()
        self.alpha_service = AlphaService()
        self.decision_engine = DecisionPolicyEngine()
        self.sizing_engine = SizingEngine()
        self.risk_gate = PreTradeRiskGate()
        self.unified_recommendation_repo = get_unified_recommendation_repository()
        self.transition_plan_repo = get_portfolio_transition_plan_repository()

    def execute(
        self,
        *,
        user,
        top_n: int = 10,
        portfolio_id: int | None = None,
        pool_mode: str | None = None,
        alpha_scope: str | None = None,
        refresh_sizing_pulse_if_stale: bool = False,
    ) -> AlphaHomepageData:
        today = resolve_recent_closed_trade_date()
        normalized_scope = normalize_alpha_scope(alpha_scope)
        if normalized_scope == ALPHA_SCOPE_GENERAL:
            return self._execute_general(
                user=user,
                top_n=top_n,
                trade_date=today,
            )

        resolved_pool = PortfolioAlphaPoolResolver().resolve(
            user_id=user.id,
            portfolio_id=portfolio_id,
            trade_date=today,
            pool_mode=pool_mode or ALPHA_POOL_MODE_PRICE_COVERED,
        )
        scope = resolved_pool.scope

        alpha_result = self._fetch_alpha_result(
            user=user,
            scope=scope,
            trade_date=today,
            top_n=top_n,
        )
        self._attach_scope_resolution_metadata(
            result=alpha_result,
            resolved_pool=resolved_pool,
        )
        meta = self._build_meta(
            alpha_result=alpha_result,
            scope=scope,
            resolved_pool=resolved_pool,
        )

        top_scores = list(alpha_result.scores[:top_n]) if alpha_result.success else []
        stock_context = self._load_stock_context([score.code for score in top_scores])
        actionable_map = self._load_actionable_map()
        pending_map = self._load_pending_map()
        position_map, portfolio_snapshot, sizing_context = self._load_portfolio_context(
            user_id=user.id,
            portfolio_id=resolved_pool.portfolio_id,
            refresh_pulse_if_stale=refresh_sizing_pulse_if_stale,
        )
        policy_state = self._load_policy_state()

        top_candidates: list[dict[str, Any]] = []
        displayed_snapshots: dict[str, dict[str, Any]] = {}
        for score in top_scores:
            item = self._build_candidate_item(
                score=score,
                stock_context=stock_context.get(score.code, {}),
                actionable_candidate=actionable_map.get(score.code),
                pending_request=pending_map.get(score.code),
                sizing_context=sizing_context,
                portfolio_snapshot=portfolio_snapshot,
                position_map=position_map,
                policy_state=policy_state,
                meta=meta,
            )
            top_candidates.append(item)
            displayed_snapshots[item["code"]] = item

        actionable_candidates = [item for item in top_candidates if item["stage"] == "actionable"]
        exit_watchlist = self._build_exit_watchlist(
            user_id=user.id,
            trade_date=today,
        )
        exit_watch_summary = self._build_exit_watch_summary(exit_watchlist)
        pending_requests = [
            self._serialize_pending_request(
                request_model=model,
                stock_context=stock_context.get(code, {}),
            )
            for code, model in pending_map.items()
        ]

        for item in pending_requests:
            displayed_snapshots.setdefault(item["code"], item)

        history_run_id = self._persist_history(
            user_id=user.id,
            portfolio_id=resolved_pool.portfolio_id,
            portfolio_name=resolved_pool.portfolio_name,
            scope=scope,
            alpha_result=alpha_result,
            meta=meta,
            snapshots=list(displayed_snapshots.values()),
        )
        recent_runs = self._serialize_recent_runs(
            self.history_repo.list_recent_runs(
                user_id=user.id,
                portfolio_id=resolved_pool.portfolio_id,
                limit=5,
            )
        )

        return AlphaHomepageData(
            pool={
                "alpha_scope": ALPHA_SCOPE_PORTFOLIO,
                "portfolio_id": resolved_pool.portfolio_id,
                "portfolio_name": resolved_pool.portfolio_name,
                "label": scope.display_label,
                "pool_type": scope.pool_type,
                "market": scope.market,
                "pool_mode": scope.pool_mode,
                "pool_size": scope.pool_size,
                "requested_pool_mode": resolved_pool.requested_pool_mode,
                "requested_pool_size": resolved_pool.requested_pool_size,
                "scope_fallback": resolved_pool.scope_fallback,
                "fallback_reason": resolved_pool.fallback_reason,
                "selection_reason": scope.selection_reason,
                "scope_hash": scope.scope_hash,
            },
            meta=meta,
            top_candidates=top_candidates,
            actionable_candidates=actionable_candidates,
            exit_watchlist=exit_watchlist,
            exit_watch_summary=exit_watch_summary,
            pending_requests=pending_requests,
            recent_runs=recent_runs,
            history_run_id=history_run_id,
        )

    def _execute_general(self, *, user, top_n: int, trade_date: date) -> AlphaHomepageData:
        """Build a broad-universe research-only Alpha ranking payload."""
        alpha_result = self._fetch_general_alpha_result(
            user=user,
            trade_date=trade_date,
            top_n=top_n,
        )
        self._mark_general_research_only(result=alpha_result, trade_date=trade_date)
        top_scores = list(alpha_result.scores[:top_n]) if alpha_result.success else []
        scope = self._build_general_scope(
            trade_date=trade_date,
            instrument_codes=[score.code for score in top_scores],
        )
        meta = self._build_meta(alpha_result=alpha_result, scope=scope)

        stock_context = self._load_stock_context([score.code for score in top_scores])
        policy_state = self._load_policy_state()
        top_candidates = [
            self._build_candidate_item(
                score=score,
                stock_context=stock_context.get(score.code, {}),
                actionable_candidate=None,
                pending_request=None,
                sizing_context=None,
                portfolio_snapshot=None,
                position_map={},
                policy_state=policy_state,
                meta=meta,
            )
            for score in top_scores
        ]

        return AlphaHomepageData(
            pool={
                "alpha_scope": ALPHA_SCOPE_GENERAL,
                "portfolio_id": None,
                "portfolio_name": "通用研究池",
                "label": scope.display_label,
                "pool_type": scope.pool_type,
                "market": scope.market,
                "pool_mode": scope.pool_mode,
                "pool_size": scope.pool_size,
                "requested_pool_mode": scope.pool_mode,
                "requested_pool_size": scope.pool_size,
                "scope_fallback": False,
                "fallback_reason": "",
                "selection_reason": scope.selection_reason,
                "scope_hash": scope.scope_hash,
                "universe_id": "csi300",
            },
            meta=meta,
            top_candidates=top_candidates,
            actionable_candidates=[],
            exit_watchlist=[],
            exit_watch_summary=self._build_exit_watch_summary([]),
            pending_requests=[],
            recent_runs=[],
            history_run_id=None,
        )

    def _build_exit_watchlist(self, *, user_id: int, trade_date: date) -> list[dict[str, Any]]:
        positions = list_user_position_payloads(
            user_id=user_id,
            include_account_meta=True,
        )
        if not positions:
            return []

        positions_by_account: dict[int, list[dict[str, Any]]] = defaultdict(list)
        signal_ids: set[int] = set()
        for position in positions:
            account_id = self._safe_int(position.get("account_id"))
            if account_id is None:
                continue
            positions_by_account[account_id].append(position)
            signal_id = self._safe_int(position.get("signal_id"))
            if signal_id is not None:
                signal_ids.add(signal_id)

        recommendation_map_by_account: dict[int, dict[str, Any]] = {}
        transition_order_map_by_account: dict[int, dict[str, dict[str, Any]]] = {}
        for account_id in positions_by_account:
            recommendations = self._load_unified_recommendations(account_id)
            recommendation_map = self._latest_recommendations_by_security(recommendations)
            recommendation_map_by_account[account_id] = recommendation_map
            for recommendation in recommendation_map.values():
                for signal_id in getattr(recommendation, "source_signal_ids", []) or []:
                    normalized_id = self._safe_int(signal_id)
                    if normalized_id is not None:
                        signal_ids.add(normalized_id)
            transition_order_map_by_account[account_id] = self._load_transition_orders(
                account_id=account_id,
                trade_date=trade_date,
            )

        signal_payloads = self._load_signal_invalidation_payloads(sorted(signal_ids))
        watchlist = [
            self._build_exit_watch_item(
                position=position,
                recommendation_map=recommendation_map_by_account.get(account_id, {}),
                transition_order_map=transition_order_map_by_account.get(account_id, {}),
                signal_payloads=signal_payloads,
            )
            for account_id, account_positions in positions_by_account.items()
            for position in account_positions
        ]
        watchlist.sort(
            key=lambda item: (
                int(item.get("priority_rank", 99)),
                -float(item.get("market_value", 0.0) or 0.0),
                str(item.get("asset_code") or ""),
            )
        )
        return watchlist

    def _build_exit_watch_item(
        self,
        *,
        position: dict[str, Any],
        recommendation_map: dict[str, Any],
        transition_order_map: dict[str, Any],
        signal_payloads: dict[str, dict[str, Any]],
    ) -> dict[str, Any]:
        asset_code = str(position.get("asset_code") or "").upper()
        recommendation = recommendation_map.get(asset_code)
        order_entry = transition_order_map.get(asset_code, {})
        order = order_entry.get("order")
        plan_id = str(order_entry.get("plan_id") or "")
        signal_id = self._safe_int(position.get("signal_id"))
        if signal_id is None and recommendation is not None:
            signal_id = self._safe_int(
                next(iter(getattr(recommendation, "source_signal_ids", []) or []), None)
            )
        signal_payload = signal_payloads.get(str(signal_id), {}) if signal_id is not None else {}

        is_invalidated = bool(position.get("is_invalidated"))
        exit_action = "HOLD"
        exit_action_label = "继续跟踪"
        exit_source = "tracking.only"
        exit_reason_code = "TRACKING_ONLY"
        exit_reason_text = "当前未触发退出或减仓条件，继续跟踪。"
        reduce_quantity: int | None = None
        priority_rank = 2

        if is_invalidated:
            exit_action = "SELL"
            exit_action_label = "立即退出"
            exit_source = "simulated_trading.position_invalidation"
            exit_reason_code = "POSITION_INVALIDATED"
            exit_reason_text = str(position.get("invalidation_reason") or "持仓已被证伪。")
            priority_rank = 0
        elif (
            order is not None
            and getattr(order, "action", "") in {"EXIT", "REDUCE"}
            and bool(getattr(order, "is_ready_for_approval", False))
        ):
            order_action = str(getattr(order, "action", "")).upper()
            exit_action = "SELL" if order_action == "EXIT" else "REDUCE"
            exit_action_label = "立即退出" if exit_action == "SELL" else "减仓"
            exit_source = "decision_rhythm.transition_plan"
            exit_reason_code = f"TRANSITION_PLAN_{order_action}"
            exit_reason_text = str(
                getattr(order, "invalidation_description", "") or f"调仓计划建议 {order_action}"
            )
            reduce_quantity = (
                abs(int(getattr(order, "delta_qty", 0) or 0)) if exit_action == "REDUCE" else None
            )
            priority_rank = 0 if exit_action == "SELL" else 1
        elif str(getattr(recommendation, "side", "") or "").upper() == "SELL":
            exit_action = "SELL"
            exit_action_label = "立即退出"
            exit_source = "decision_rhythm.recommendation"
            exit_reason_code = "UNIFIED_RECOMMENDATION_SELL"
            exit_reason_text = str(
                getattr(recommendation, "human_rationale", "") or "统一推荐已转为 SELL。"
            )
            priority_rank = 1

        decision_side = str(getattr(recommendation, "side", "") or "HOLD").upper()
        stop_loss_price = self._normalize_decimal_string(
            getattr(order, "stop_loss_price", None)
            or getattr(recommendation, "stop_loss_price", None)
        )
        target_price_low = self._normalize_decimal_string(
            getattr(recommendation, "target_price_low", None)
        )
        target_price_high = self._normalize_decimal_string(
            getattr(recommendation, "target_price_high", None)
        )
        invalidation_summary = (
            str(position.get("invalidation_description") or "").strip()
            or str(getattr(order, "invalidation_description", "") or "").strip()
            or str(
                signal_payload.get("invalidation_description")
                or signal_payload.get("invalidation_logic")
                or ""
            ).strip()
            or "暂无持仓级证伪摘要。"
        )
        contract_ready = bool(
            stop_loss_price is not None
            and float(stop_loss_price) > 0
            and invalidation_summary != "暂无持仓级证伪摘要。"
        )

        source_signal_ids = [
            str(item)
            for item in (getattr(recommendation, "source_signal_ids", []) or [])
            if str(item or "").strip()
        ]
        if signal_id is not None and str(signal_id) not in source_signal_ids:
            source_signal_ids.insert(0, str(signal_id))

        recommendation_id = str(getattr(recommendation, "recommendation_id", "") or "")
        account_id = self._safe_int(position.get("account_id"))
        invalidation_conditions = []
        invalidation_rule = getattr(order, "invalidation_rule", None) or signal_payload.get(
            "invalidation_rule_json", {}
        )
        if isinstance(invalidation_rule, dict):
            invalidation_conditions = [
                str(item)
                for item in (invalidation_rule.get("conditions") or [])
                if str(item or "").strip()
            ]
        recommendation_user_action = ""
        recommendation_snapshot = {}
        if recommendation is not None:
            recommendation_user_action = normalize_exit_user_action(
                getattr(
                    getattr(recommendation, "user_action", None),
                    "value",
                    getattr(recommendation, "user_action", ""),
                )
            )
            recommendation_snapshot = {
                "recommendation_id": recommendation_id,
                "side": str(getattr(recommendation, "side", "") or ""),
                "status": getattr(
                    getattr(recommendation, "status", None),
                    "value",
                    getattr(recommendation, "status", ""),
                ),
                "user_action": recommendation_user_action,
                "user_action_label": build_exit_user_action_label(recommendation_user_action),
                "is_processed": recommendation_user_action in {"ADOPTED", "IGNORED"},
                "confidence": float(getattr(recommendation, "confidence", 0.0) or 0.0),
                "composite_score": float(getattr(recommendation, "composite_score", 0.0) or 0.0),
                "alpha_model_score": float(
                    getattr(recommendation, "alpha_model_score", 0.0) or 0.0
                ),
                "human_rationale": str(getattr(recommendation, "human_rationale", "") or ""),
                "reason_codes": list(getattr(recommendation, "reason_codes", []) or []),
                "target_price_low": self._normalize_decimal_string(
                    getattr(recommendation, "target_price_low", None)
                ),
                "target_price_high": self._normalize_decimal_string(
                    getattr(recommendation, "target_price_high", None)
                ),
                "stop_loss_price": self._normalize_decimal_string(
                    getattr(recommendation, "stop_loss_price", None)
                ),
            }
        transition_plan_snapshot = {}
        if order is not None:
            transition_plan_snapshot = {
                "plan_id": plan_id,
                "action": str(getattr(order, "action", "") or ""),
                "current_qty": int(getattr(order, "current_qty", 0) or 0),
                "target_qty": int(getattr(order, "target_qty", 0) or 0),
                "delta_qty": int(getattr(order, "delta_qty", 0) or 0),
                "stop_loss_price": self._normalize_decimal_string(
                    getattr(order, "stop_loss_price", None)
                ),
                "price_band_low": self._normalize_decimal_string(
                    getattr(order, "price_band_low", None)
                ),
                "price_band_high": self._normalize_decimal_string(
                    getattr(order, "price_band_high", None)
                ),
                "invalidation_description": str(
                    getattr(order, "invalidation_description", "") or ""
                ),
                "notes": [
                    str(item)
                    for item in (getattr(order, "notes", []) or [])
                    if str(item or "").strip()
                ],
                "is_ready_for_approval": bool(getattr(order, "is_ready_for_approval", False)),
            }
        signal_contract_snapshot = {
            "signal_id": signal_id,
            "invalidation_description": str(
                signal_payload.get("invalidation_description")
                or signal_payload.get("invalidation_logic")
                or ""
            ),
            "conditions": invalidation_conditions,
        }
        user_action_label = build_exit_user_action_label(recommendation_user_action)
        is_processed = recommendation_user_action in {"ADOPTED", "IGNORED"}
        workspace_step = 5 if plan_id or exit_action in {"SELL", "REDUCE"} else 4
        decision_workspace_url = build_decision_workspace_url(
            security_code=asset_code,
            source="dashboard-exit",
            step=workspace_step,
            account_id=account_id,
            action=exit_action,
        )

        return {
            "account_id": account_id,
            "account_name": str(position.get("account_name") or "默认账户"),
            "asset_code": asset_code,
            "asset_name": str(position.get("asset_name") or asset_code),
            "shares": float(position.get("shares") or 0.0),
            "market_value": float(position.get("market_value") or 0.0),
            "avg_cost": float(position.get("avg_cost") or 0.0),
            "current_price": float(position.get("current_price") or 0.0),
            "unrealized_pnl_pct": float(position.get("unrealized_pnl_pct") or 0.0),
            "opened_at": str(position.get("opened_at") or ""),
            "signal_id": signal_id,
            "source_signal_ids": source_signal_ids,
            "decision_side": decision_side,
            "decision_side_label": {
                "BUY": "继续看多/可加仓",
                "SELL": "统一推荐 SELL",
                "HOLD": "继续持有",
            }.get(decision_side, decision_side or "未覆盖"),
            "exit_action": exit_action,
            "exit_action_label": exit_action_label,
            "exit_source": exit_source,
            "exit_reason_code": exit_reason_code,
            "exit_reason_text": exit_reason_text,
            "reduce_quantity": reduce_quantity,
            "stop_loss_price": stop_loss_price,
            "target_price_low": target_price_low,
            "target_price_high": target_price_high,
            "invalidation_summary": invalidation_summary,
            "contract_ready": contract_ready,
            "contract_status_label": "已绑定退出契约" if contract_ready else "待补退出契约",
            "is_invalidated": is_invalidated,
            "invalidation_reason": str(position.get("invalidation_reason") or ""),
            "priority_rank": priority_rank,
            "priority_label": {
                0: "立即处理",
                1: "本轮评估",
                2: "持续跟踪",
            }.get(priority_rank, "持续跟踪"),
            "recommendation_id": recommendation_id,
            "user_action": recommendation_user_action,
            "user_action_label": user_action_label,
            "is_processed": is_processed,
            "recommendation_detail_url": (
                f"/api/decision/workspace/recommendations/?recommendation_id={recommendation_id}"
                if recommendation_id
                else ""
            ),
            "transition_plan_id": plan_id,
            "transition_plan_detail_url": (
                f"/api/decision/workspace/plans/{plan_id}/" if plan_id else ""
            ),
            "recommendation_snapshot": recommendation_snapshot,
            "transition_plan_snapshot": transition_plan_snapshot,
            "signal_contract_snapshot": signal_contract_snapshot,
            "account_detail_url": (
                f"/simulated-trading/my-accounts/{account_id}/" if account_id is not None else ""
            ),
            "decision_workspace_url": decision_workspace_url,
        }

    def _load_unified_recommendations(self, account_id: int) -> list[Any]:
        try:
            return list(self.unified_recommendation_repo.get_by_account(str(account_id)) or [])
        except Exception as exc:
            logger.warning(
                "Failed to load unified recommendations for dashboard exit watchlist %s: %s",
                account_id,
                exc,
            )
            return []

    def _load_transition_orders(
        self, *, account_id: int, trade_date: date
    ) -> dict[str, dict[str, Any]]:
        try:
            plan = self.transition_plan_repo.get_latest_for_account(str(account_id))
        except Exception as exc:
            logger.warning(
                "Failed to load transition plan for dashboard exit watchlist %s: %s",
                account_id,
                exc,
            )
            return {}

        if plan is None:
            return {}

        plan_as_of = getattr(plan, "as_of", None)
        plan_date = plan_as_of.date() if plan_as_of is not None else None
        if plan_date != trade_date:
            return {}

        order_map: dict[str, dict[str, Any]] = {}
        for order in getattr(plan, "orders", []) or []:
            security_code = str(getattr(order, "security_code", "") or "").upper()
            if security_code:
                order_map[security_code] = {
                    "order": order,
                    "plan_id": str(getattr(plan, "plan_id", "") or ""),
                }
        return order_map

    def _load_signal_invalidation_payloads(
        self, signal_ids: list[int]
    ) -> dict[str, dict[str, Any]]:
        if not signal_ids:
            return {}
        try:
            return get_signal_repository().get_invalidation_payloads(signal_ids)
        except Exception as exc:
            logger.warning(
                "Failed to load signal invalidation payloads for dashboard exit watchlist: %s",
                exc,
            )
            return {}

    @staticmethod
    def _build_exit_watch_summary(watchlist: list[dict[str, Any]]) -> dict[str, Any]:
        urgent_count = sum(1 for item in watchlist if item.get("priority_rank") == 0)
        sell_count = sum(1 for item in watchlist if item.get("exit_action") == "SELL")
        reduce_count = sum(1 for item in watchlist if item.get("exit_action") == "REDUCE")
        hold_count = sum(1 for item in watchlist if item.get("exit_action") == "HOLD")
        return {
            "total": len(watchlist),
            "urgent_count": urgent_count,
            "sell_count": sell_count,
            "reduce_count": reduce_count,
            "hold_count": hold_count,
        }

    @staticmethod
    def _latest_recommendations_by_security(recommendations: list[Any]) -> dict[str, Any]:
        latest_by_security: dict[str, Any] = {}
        for recommendation in recommendations:
            security_code = str(getattr(recommendation, "security_code", "") or "").upper()
            if not security_code:
                continue

            current = latest_by_security.get(security_code)
            if current is None:
                latest_by_security[security_code] = recommendation
                continue

            current_time = getattr(current, "updated_at", None) or getattr(
                current, "created_at", None
            )
            next_time = getattr(recommendation, "updated_at", None) or getattr(
                recommendation,
                "created_at",
                None,
            )
            if next_time and (current_time is None or next_time >= current_time):
                latest_by_security[security_code] = recommendation
        return latest_by_security

    @staticmethod
    def _safe_int(value: Any) -> int | None:
        try:
            if value in (None, ""):
                return None
            return int(value)
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _normalize_decimal_string(value: Any) -> str | None:
        if value in (None, "", 0, "0"):
            return None
        try:
            numeric = float(value)
        except (TypeError, ValueError):
            return None
        if numeric <= 0:
            return None
        return f"{numeric:.2f}"

    def list_history(
        self,
        *,
        user_id: int,
        portfolio_id: int | None = None,
        stock_code: str | None = None,
        stage: str | None = None,
        source: str | None = None,
        trade_date: date | None = None,
    ) -> list[dict[str, Any]]:
        runs = self.history_repo.filter_runs(
            user_id=user_id,
            portfolio_id=portfolio_id,
            stock_code=stock_code,
            stage=stage,
            source=source,
            trade_date=trade_date,
        )
        return self._serialize_recent_runs(runs)

    def get_history_detail(self, *, user_id: int, run_id: int) -> dict[str, Any] | None:
        run = self.history_repo.get_run_detail(user_id=user_id, run_id=run_id)
        if run is None:
            return None
        snapshot_codes = [
            str(snapshot.stock_code or "").strip().upper() for snapshot in run.snapshots.all()
        ]
        stock_context = self.context_repo.load_stock_context(snapshot_codes)
        snapshots = []
        for snapshot in run.snapshots.all():
            snapshot_code = str(snapshot.stock_code or "").strip().upper()
            fallback_name = (stock_context.get(snapshot_code) or {}).get("name") or snapshot_code
            stock_name = str(snapshot.stock_name or "").strip() or fallback_name
            if stock_name.upper() == snapshot_code:
                stock_name = fallback_name
            snapshots.append(
                {
                    "code": snapshot_code,
                    "name": stock_name,
                    "stage": snapshot.stage,
                    "gate_status": snapshot.gate_status,
                    "rank": snapshot.rank,
                    "alpha_score": snapshot.alpha_score,
                    "confidence": snapshot.confidence,
                    "source": snapshot.source,
                    "buy_reasons": snapshot.buy_reasons,
                    "no_buy_reasons": snapshot.no_buy_reasons,
                    "invalidation_rule": snapshot.invalidation_rule,
                    "risk_snapshot": snapshot.risk_snapshot,
                    "suggested_position_pct": snapshot.suggested_position_pct,
                    "suggested_notional": snapshot.suggested_notional,
                    "suggested_quantity": snapshot.suggested_quantity,
                    "extra_payload": snapshot.extra_payload,
                }
            )
        return {
            "id": run.id,
            "portfolio_id": run.portfolio_id,
            "portfolio_name": run.portfolio_name,
            "trade_date": run.trade_date.isoformat(),
            "scope_label": run.scope_label,
            "source": run.source,
            "provider_source": run.provider_source,
            "uses_cached_data": run.uses_cached_data,
            "cache_reason": run.cache_reason,
            "fallback_reason": run.fallback_reason,
            "requested_trade_date": (
                run.requested_trade_date.isoformat() if run.requested_trade_date else None
            ),
            "effective_asof_date": (
                run.effective_asof_date.isoformat() if run.effective_asof_date else None
            ),
            "meta": run.meta,
            "snapshots": snapshots,
        }

    def _fetch_general_alpha_result(self, *, user, trade_date: date, top_n: int):
        result = None
        for provider_name in ("qlib", "cache", "simple", "etf"):
            candidate = self.alpha_service.get_stock_scores(
                universe_id="csi300",
                intended_trade_date=trade_date,
                top_n=top_n,
                user=user,
                provider_filter=provider_name,
            )
            result = candidate
            if candidate.success and candidate.scores:
                return candidate
        return result or AlphaResult(
            success=False,
            scores=[],
            source="none",
            timestamp=trade_date.isoformat(),
            status="unavailable",
            error_message="general_alpha_unavailable",
            metadata={},
        )

    def _mark_general_research_only(self, *, result, trade_date: date) -> None:
        metadata = dict(getattr(result, "metadata", {}) or {})
        metadata.update(
            {
                "alpha_scope": ALPHA_SCOPE_GENERAL,
                "research_only": True,
                "must_not_use_for_decision": True,
                "recommendation_ready": False,
                "requested_trade_date": metadata.get("requested_trade_date")
                or trade_date.isoformat(),
                "reliability_notice": {
                    "level": "info",
                    "code": "general_alpha_research_only",
                    "title": "通用 Alpha 仅供研究",
                    "message": "通用 Alpha 使用 broader/universe 级结果，只展示研究排名，不作为账户专属可执行建议。",
                },
            }
        )
        result.metadata = metadata

    @staticmethod
    def _build_general_scope(*, trade_date: date, instrument_codes: list[str]) -> AlphaPoolScope:
        return AlphaPoolScope(
            pool_type="general_universe",
            market="CN",
            pool_mode="general",
            instrument_codes=tuple(instrument_codes),
            selection_reason="通用市场研究股票池，不绑定任何账户、组合或仓位约束。",
            trade_date=trade_date,
            display_label="通用 Alpha 研究池",
            portfolio_id=None,
            portfolio_name="通用研究池",
        )

    def _fetch_alpha_result(self, *, user, scope, trade_date: date, top_n: int):
        result = None
        broader_cache_candidate = None
        for provider_name in ("cache", "simple"):
            candidate = self.alpha_service.get_stock_scores(
                universe_id=scope.universe_id,
                intended_trade_date=trade_date,
                top_n=top_n,
                user=user,
                provider_filter=provider_name,
                pool_scope=scope,
            )
            result = candidate
            if candidate.success and candidate.scores:
                metadata = dict(getattr(candidate, "metadata", {}) or {})
                if metadata.get("derived_from_broader_cache"):
                    broader_cache_candidate = candidate
                    async_status = self._trigger_async_inference_if_needed(
                        user=user,
                        scope=scope,
                        trade_date=trade_date,
                        top_n=top_n,
                    )
                    metadata.update(
                        {
                            "refresh_triggered": bool(async_status.get("refresh_triggered", False)),
                            "refresh_status": async_status.get("refresh_status", ""),
                            "async_task_id": async_status.get("async_task_id", ""),
                            "poll_after_ms": async_status.get("poll_after_ms", 5000),
                            "auto_refresh_message": async_status.get("message", ""),
                            "auto_refresh_error": async_status.get("auto_refresh_error", ""),
                        }
                    )
                    candidate.metadata = metadata
                    continue
                return candidate
        if broader_cache_candidate is not None:
            return broader_cache_candidate
        if result is not None:
            async_status = self._trigger_async_inference_if_needed(
                user=user,
                scope=scope,
                trade_date=trade_date,
                top_n=top_n,
            )
            self._mark_no_verified_recommendation(
                result=result,
                scope=scope,
                async_status=async_status,
            )
        return result

    def _trigger_async_inference_if_needed(
        self,
        *,
        user,
        scope,
        trade_date: date,
        top_n: int,
    ) -> dict[str, Any]:
        """Queue scoped Qlib inference once per short window when verified cache is missing."""
        if user is not None and not getattr(user, "is_authenticated", False):
            return {"refresh_status": "skipped", "message": "匿名用户不自动触发账户池推理。"}

        try:
            from django.core.cache import cache

            lock_key = (
                "dashboard:alpha:auto-refresh:"
                f"{getattr(scope, 'scope_hash', '')}:{trade_date.isoformat()}:{top_n}"
            )
            if not cache.add(lock_key, "queued", timeout=300):
                return {
                    "refresh_triggered": False,
                    "refresh_status": "recently_queued",
                    "poll_after_ms": 5000,
                    "message": "账户池 Alpha 推理已在后台排队，请稍后刷新。",
                }

            from apps.alpha.application.tasks import qlib_predict_scores

            task = qlib_predict_scores.delay(
                scope.universe_id,
                trade_date.isoformat(),
                top_n,
                scope_payload=scope.to_dict(),
            )
            record_pending_task(
                task_id=task.id,
                task_name="apps.alpha.application.tasks.qlib_predict_scores",
                args=(scope.universe_id, trade_date.isoformat(), top_n),
                kwargs={"scope_payload": scope.to_dict()},
            )
            return {
                "refresh_triggered": True,
                "refresh_status": "queued",
                "async_task_id": getattr(task, "id", ""),
                "poll_after_ms": 5000,
                "message": "账户池暂无可信 Alpha cache，已自动触发后台 Qlib 推理。",
            }
        except Exception as exc:
            logger.warning("Failed to auto trigger scoped Alpha inference: %s", exc, exc_info=True)
            return {
                "refresh_triggered": False,
                "refresh_status": "failed",
                "auto_refresh_error": str(exc),
                "message": "账户池 Alpha 推理自动触发失败，请手动重试。",
            }

    def _mark_no_verified_recommendation(
        self,
        *,
        result,
        scope,
        async_status: dict[str, Any] | None = None,
    ) -> None:
        metadata = dict(getattr(result, "metadata", {}) or {})
        async_status = dict(async_status or {})
        scope_label = getattr(scope, "display_label", "") or "账户驱动 Alpha 池"
        reason = (
            f"{scope_label} 暂无真实账户池 Alpha 推理或缓存结果；系统未使用硬编码股票池、"
            "默认 ETF 或静态名单生成推荐。请触发实时推理后再查看。"
        )
        metadata.update(
            {
                "is_degraded": True,
                "uses_cached_data": False,
                "fallback_mode": "none",
                "fallback_reason": "",
                "no_recommendation_reason": reason,
                "hardcoded_fallback_used": False,
                "refresh_triggered": bool(async_status.get("refresh_triggered", False)),
                "refresh_status": async_status.get("refresh_status", ""),
                "async_task_id": async_status.get("async_task_id", ""),
                "poll_after_ms": async_status.get("poll_after_ms", 5000),
                "auto_refresh_message": async_status.get("message", ""),
                "auto_refresh_error": async_status.get("auto_refresh_error", ""),
                "scope_hash": getattr(scope, "scope_hash", ""),
                "scope_label": scope_label,
                "scope_metadata": scope.to_dict() if hasattr(scope, "to_dict") else {},
                "reliability_notice": {
                    "level": "warning",
                    "code": "no_verified_alpha_recommendation",
                    "title": "暂无可信 Alpha 推荐",
                    "message": reason,
                },
            }
        )
        result.metadata = metadata
        result.status = "unavailable"

    def _attach_scope_resolution_metadata(self, *, result, resolved_pool) -> None:
        metadata = dict(getattr(result, "metadata", {}) or {})
        metadata.update(
            {
                "requested_pool_mode": resolved_pool.requested_pool_mode,
                "requested_pool_size": resolved_pool.requested_pool_size,
                "effective_pool_mode": resolved_pool.scope.pool_mode,
                "effective_pool_size": resolved_pool.scope.pool_size,
                "scope_fallback": resolved_pool.scope_fallback,
                "scope_fallback_reason": resolved_pool.fallback_reason,
                "scope_fallback_universe_id": resolved_pool.scope.universe_id,
            }
        )
        if resolved_pool.scope_fallback and not metadata.get("reliability_notice"):
            metadata["reliability_notice"] = {
                "level": "warning",
                "code": "account_scope_widened",
                "title": "Alpha 已自动扩大账户池范围",
                "message": resolved_pool.fallback_reason,
            }
        result.metadata = metadata

    @staticmethod
    def _parse_meta_date(value: Any) -> date | None:
        if value in (None, ""):
            return None
        if isinstance(value, date):
            return value
        try:
            return date.fromisoformat(str(value)[:10])
        except ValueError:
            return None

    def _build_readiness_fields(
        self, *, alpha_result, scope, metadata: dict[str, Any]
    ) -> dict[str, Any]:
        requested_trade_date = self._parse_meta_date(metadata.get("requested_trade_date"))
        effective_asof_date = self._parse_meta_date(metadata.get("effective_asof_date"))
        result_age_days = getattr(alpha_result, "staleness_days", None)
        if result_age_days is None and requested_trade_date and effective_asof_date:
            result_age_days = max((requested_trade_date - effective_asof_date).days, 0)

        derived_from_broader_cache = bool(metadata.get("derived_from_broader_cache", False))
        scope_fallback = bool(metadata.get("scope_fallback", False))
        trade_date_adjusted = bool(metadata.get("trade_date_adjusted", False))
        latest_available_qlib_result = bool(metadata.get("latest_available_qlib_result", False))
        adjusted_to_latest_completed_session = (
            trade_date_adjusted
            and requested_trade_date is not None
            and effective_asof_date is not None
            and requested_trade_date.weekday() >= 5
            and 0 <= (requested_trade_date - effective_asof_date).days <= 3
        )
        hardcoded_fallback_used = bool(metadata.get("hardcoded_fallback_used", False))
        is_degraded = bool(metadata.get("is_degraded", False))
        research_only = bool(metadata.get("research_only", False)) or (
            metadata.get("alpha_scope") == ALPHA_SCOPE_GENERAL
        )
        broad_pool_research_only = (
            not research_only
            and getattr(scope, "pool_mode", "") in {"market", "price_covered"}
        )
        no_recommendation_reason = str(metadata.get("no_recommendation_reason") or "")
        fallback_mode = str(metadata.get("fallback_mode") or "")
        scores = list(getattr(alpha_result, "scores", []) or [])
        provider_source = (
            str(metadata.get("provider_source") or getattr(alpha_result, "source", ""))
            .strip()
            .lower()
        )
        data_driven_simple_result = (
            provider_source == "simple"
            and str(metadata.get("factor_basis") or "") in {"quote_momentum", ""}
            and bool(scores)
        )

        scope_verification_status = "verified"
        if not scores:
            scope_verification_status = "unavailable"
        elif research_only:
            scope_verification_status = "general_universe"
        elif derived_from_broader_cache:
            scope_verification_status = "derived_from_broader_cache"
        elif scope_fallback:
            scope_verification_status = "scope_fallback"

        freshness_status = "fresh"
        if not scores:
            freshness_status = "unavailable"
        elif adjusted_to_latest_completed_session:
            freshness_status = "latest_completed_session"
        elif trade_date_adjusted:
            freshness_status = "trade_date_adjusted"
        elif fallback_mode == "forward_fill_latest_qlib_cache":
            freshness_status = "forward_filled_cache"
        elif result_age_days not in (None, 0):
            freshness_status = "stale"
        elif data_driven_simple_result:
            freshness_status = "fresh"
        elif is_degraded or not latest_available_qlib_result:
            freshness_status = "degraded"

        blocked_reason = ""
        readiness_status = "ready"
        recommendation_ready = bool(scores)
        if no_recommendation_reason:
            readiness_status = "blocked_no_verified_result"
            blocked_reason = no_recommendation_reason
            recommendation_ready = False
        elif research_only:
            readiness_status = "research_only"
            blocked_reason = "通用 Alpha 仅用于研究排名；未绑定账户 scope，不能作为真实交易决策。"
            recommendation_ready = False
        elif hardcoded_fallback_used:
            readiness_status = "blocked_hardcoded_fallback"
            blocked_reason = "当前结果仍含硬编码回退痕迹，不能作为真实 Alpha 推荐。"
            recommendation_ready = False
        elif derived_from_broader_cache:
            readiness_status = "blocked_broader_scope_cache"
            blocked_reason = (
                "当前结果来自 broader-scope cache 映射，账户专属 scoped Alpha 推理尚未完成。"
            )
            recommendation_ready = False
        elif scope_fallback:
            readiness_status = "blocked_scope_fallback"
            blocked_reason = str(
                metadata.get("scope_fallback_reason")
                or "当前 Alpha 股票池已扩大到回退范围，不能视为原始账户池推荐。"
            )
            recommendation_ready = False
        elif trade_date_adjusted and not adjusted_to_latest_completed_session:
            readiness_status = "blocked_trade_date_adjusted"
            blocked_reason = str(
                ((metadata.get("reliability_notice") or {}).get("message"))
                or "请求交易日的 Alpha 数据尚未落地，当前只拿到了最新可用交易日结果。"
            )
            recommendation_ready = False
        elif broad_pool_research_only:
            readiness_status = "blocked_broad_pool_research_only"
            blocked_reason = (
                "当前 Alpha 股票池仅为市场/价格覆盖研究池，不含账户特异约束；"
                "系统不会把这类 broad pool 排名包装成账户推荐。"
            )
            recommendation_ready = False
        elif fallback_mode == "forward_fill_latest_qlib_cache":
            readiness_status = "blocked_forward_filled_cache"
            blocked_reason = str(
                metadata.get("fallback_reason")
                or "当前结果为前推缓存，尚未通过当期 Alpha 实时推理验证。"
            )
            recommendation_ready = False
        elif result_age_days not in (None, 0) and not adjusted_to_latest_completed_session:
            readiness_status = "blocked_stale"
            blocked_reason = f"当前 Alpha 结果相对请求交易日已陈旧 {result_age_days} 天。"
            recommendation_ready = False
        elif is_degraded:
            readiness_status = "blocked_degraded"
            blocked_reason = str(
                ((metadata.get("reliability_notice") or {}).get("message"))
                or "当前 Alpha 结果处于 degraded 状态，不能作为决策推荐。"
            )
            recommendation_ready = False
        elif not latest_available_qlib_result and not data_driven_simple_result:
            readiness_status = "blocked_unverified_delivery"
            blocked_reason = "当前 Alpha 输出尚未验证为请求交易日的最新 scoped Qlib 结果。"
            recommendation_ready = False

        verified_scope_hash = ""
        verified_asof_date = None
        if scores and scope_verification_status == "verified":
            verified_scope_hash = getattr(scope, "scope_hash", "") or ""
            verified_asof_date = (
                effective_asof_date.isoformat() if effective_asof_date is not None else None
            )

        return {
            "result_age_days": result_age_days,
            "freshness_status": freshness_status,
            "is_stale": result_age_days not in (None, 0)
            and not adjusted_to_latest_completed_session,
            "scope_verification_status": scope_verification_status,
            "is_scope_verified": scope_verification_status == "verified",
            "latest_available_qlib_result": latest_available_qlib_result,
            "derived_from_broader_cache": derived_from_broader_cache,
            "trade_date_adjusted": trade_date_adjusted,
            "latest_completed_session_result": adjusted_to_latest_completed_session,
            "effective_trade_date": metadata.get("effective_trade_date"),
            "recommendation_ready": recommendation_ready,
            "must_not_use_for_decision": not recommendation_ready,
            "blocked_reason": blocked_reason,
            "readiness_status": readiness_status,
            "verified_scope_hash": verified_scope_hash,
            "verified_asof_date": verified_asof_date,
        }

    def _build_meta(self, *, alpha_result, scope, resolved_pool=None) -> dict[str, Any]:
        metadata = dict(getattr(alpha_result, "metadata", {}) or {})
        scope_metadata = scope.to_dict() if hasattr(scope, "to_dict") else {}
        requested_pool_mode = metadata.get("requested_pool_mode")
        requested_pool_size = metadata.get("requested_pool_size")
        if resolved_pool is not None:
            requested_pool_mode = requested_pool_mode or resolved_pool.requested_pool_mode
            requested_pool_size = requested_pool_size or resolved_pool.requested_pool_size
        if requested_pool_mode is None:
            requested_pool_mode = getattr(scope, "pool_mode", "")
        if requested_pool_size is None:
            requested_pool_size = getattr(scope, "pool_size", 0)
        meta = {
            "alpha_scope": metadata.get("alpha_scope") or ALPHA_SCOPE_PORTFOLIO,
            "research_only": bool(metadata.get("research_only", False)),
            "status": getattr(alpha_result, "status", "unavailable"),
            "source": getattr(alpha_result, "source", "none"),
            "provider_source": metadata.get("provider_source")
            or getattr(alpha_result, "source", "none"),
            "is_degraded": bool(metadata.get("is_degraded", False)),
            "uses_cached_data": bool(metadata.get("uses_cached_data", False)),
            "requested_trade_date": metadata.get("requested_trade_date"),
            "effective_asof_date": metadata.get("effective_asof_date"),
            "cache_date": metadata.get("cache_date"),
            "cache_created_at": metadata.get("created_at"),
            "cache_reason": metadata.get("fallback_reason")
            or metadata.get("warning_message")
            or "",
            "fallback_reason": metadata.get("fallback_reason") or "",
            "fallback_from": metadata.get("fallback_from"),
            "scope_fallback": bool(metadata.get("scope_fallback", False)),
            "scope_fallback_reason": metadata.get("scope_fallback_reason") or "",
            "scope_fallback_universe_id": metadata.get("scope_fallback_universe_id"),
            "no_recommendation_reason": metadata.get("no_recommendation_reason") or "",
            "hardcoded_fallback_used": bool(metadata.get("hardcoded_fallback_used", False)),
            "refresh_status": metadata.get("refresh_status") or "",
            "async_task_id": metadata.get("async_task_id") or "",
            "poll_after_ms": metadata.get("poll_after_ms") or 5000,
            "auto_refresh_message": metadata.get("auto_refresh_message") or "",
            "auto_refresh_error": metadata.get("auto_refresh_error") or "",
            "warning_title": (metadata.get("reliability_notice") or {}).get("title"),
            "warning_message": (metadata.get("reliability_notice") or {}).get("message"),
            "warning_level": (metadata.get("reliability_notice") or {}).get("level"),
            "refresh_triggered": bool(metadata.get("refresh_triggered", False)),
            "requested_pool_mode": requested_pool_mode,
            "requested_pool_size": requested_pool_size,
            "effective_pool_mode": metadata.get("effective_pool_mode")
            or getattr(scope, "pool_mode", ""),
            "effective_pool_size": metadata.get("effective_pool_size")
            or getattr(scope, "pool_size", 0),
            "scope_hash": getattr(scope, "scope_hash", ""),
            "scope_label": getattr(scope, "display_label", ""),
            "scope_metadata": scope_metadata,
            "universe_id": getattr(scope, "universe_id", ""),
            "pool_mode": getattr(scope, "pool_mode", ""),
            "model_hash": metadata.get("model_artifact_hash", ""),
        }
        meta.update(
            self._build_readiness_fields(
                alpha_result=alpha_result,
                scope=scope,
                metadata={**metadata, **meta},
            )
        )
        return meta

    def _load_stock_context(self, codes: list[str]) -> dict[str, dict[str, Any]]:
        return self.context_repo.load_stock_context(codes)

    def _load_actionable_map(self) -> dict[str, Any]:
        return self.context_repo.load_actionable_map()

    def _load_pending_map(self) -> dict[str, Any]:
        return self.context_repo.load_pending_map()

    def _load_portfolio_context(
        self,
        *,
        user_id: int,
        portfolio_id: int | None,
        refresh_pulse_if_stale: bool,
    ) -> tuple[dict[str, float], Any | None, Any | None]:
        if portfolio_id is None:
            return {}, None, None
        position_map: dict[str, float] = {}
        portfolio_snapshot = self.portfolio_repo.get_portfolio_snapshot(portfolio_id)
        if portfolio_snapshot is not None:
            for position in portfolio_snapshot.positions:
                position_map[str(position.asset_code).upper()] = float(position.market_value)
        if (
            portfolio_snapshot is None
            or float(getattr(portfolio_snapshot, "total_value", 0.0) or 0.0) <= 0
        ):
            context_repo = getattr(self, "context_repo", None)
            account_totals = (
                context_repo.load_user_account_totals(user_id) if context_repo is not None else None
            )
            total_assets = float((account_totals or {}).get("total_assets") or 0.0)
            if total_assets > 0:
                portfolio_snapshot = SimpleNamespace(
                    total_value=total_assets,
                    positions=(
                        getattr(portfolio_snapshot, "positions", []) if portfolio_snapshot else []
                    ),
                )
        try:
            sizing_context = GetSizingContextUseCase().execute(
                portfolio_id=portfolio_id,
                user_id=user_id,
                refresh_pulse_if_stale=refresh_pulse_if_stale,
            )
        except Exception as exc:
            logger.warning("Failed to load sizing context for portfolio %s: %s", portfolio_id, exc)
            sizing_context = None
        return position_map, portfolio_snapshot, sizing_context

    def _load_policy_state(self) -> dict[str, Any]:
        return self.context_repo.load_policy_state()

    def _build_candidate_item(
        self,
        *,
        score,
        stock_context: dict[str, Any],
        actionable_candidate,
        pending_request,
        sizing_context,
        portfolio_snapshot,
        position_map: dict[str, float],
        policy_state: dict[str, Any],
        meta: dict[str, Any],
    ) -> dict[str, Any]:
        code = str(score.code).upper()
        current_price = float(stock_context.get("close") or 0.0)
        account_equity = float(getattr(portfolio_snapshot, "total_value", 0.0) or 0.0)
        current_position_value = float(position_map.get(code, 0.0))
        multiplier = (
            float(sizing_context.multiplier_result.multiplier)
            if sizing_context is not None
            else 1.0
        )
        regime_name = sizing_context.regime_name if sizing_context else "Unknown"
        regime_confidence = float(sizing_context.regime_confidence) if sizing_context else 0.0
        pulse_composite = float(sizing_context.pulse_composite) if sizing_context else 0.0
        pulse_warning = bool(sizing_context.pulse_warning) if sizing_context else False
        market_temperature_score = (
            float(getattr(sizing_context, "market_temperature_score", 0.0) or 0.0)
            if sizing_context is not None
            else 0.0
        )
        market_temperature_band = (
            str(getattr(sizing_context, "market_temperature_band", "") or "").strip().lower()
            if sizing_context is not None
            else ""
        )
        market_temperature_degraded = (
            bool(getattr(sizing_context, "market_temperature_degraded", False))
            if sizing_context is not None
            else False
        )
        market_temperature_threshold_source = (
            str(getattr(sizing_context, "market_temperature_threshold_source", "") or "")
            if sizing_context is not None
            else ""
        )
        market_temperature_blocked_reason = (
            str(getattr(sizing_context, "market_temperature_blocked_reason", "") or "")
            if sizing_context is not None
            else ""
        )
        market_temperature_blocks_new_position = (
            bool(getattr(sizing_context, "market_temperature_blocks_new_position", False))
            if sizing_context is not None
            else False
        )
        market_temperature_factor = (
            float(
                getattr(
                    getattr(sizing_context, "multiplier_result", None),
                    "market_temperature_factor",
                    1.0,
                )
                or 1.0
            )
            if sizing_context is not None
            else 1.0
        )
        signal_strength = max(min((float(score.score) + 1.0) / 2.0, 1.0), 0.0)
        action, decision_codes, decision_text, _ = self.decision_engine.evaluate(
            signal_strength=signal_strength,
            signal_direction="bullish" if float(score.score) >= 0 else "bearish",
            signal_confidence=float(score.confidence),
            regime=regime_name,
            regime_confidence=regime_confidence,
            daily_pnl_pct=0.0,
            daily_trade_count=0,
            target_regime=None,
        )

        suggested_notional = 0.0
        suggested_quantity = 0.0
        suggested_position_pct = 0.0
        sizing_explain = "账户或价格上下文不足，未生成建议仓位。"
        if account_equity > 0 and current_price > 0:
            target_notional, qty, _, _, sizing_explain = self.sizing_engine.calculate(
                method="fixed_fraction",
                account_equity=account_equity,
                current_price=current_price,
                current_position_value=current_position_value,
            )
            suggested_notional = float(target_notional) * multiplier
            suggested_quantity = int(suggested_notional / current_price) if current_price else 0
            suggested_position_pct = (
                (suggested_notional / account_equity * 100) if account_equity else 0.0
            )

        passed, violations, warnings, details = self.risk_gate.check(
            symbol=code,
            side="buy",
            qty=int(suggested_quantity or 0),
            price=current_price or 0.0,
            account_equity=account_equity or 0.0,
            current_position_value=current_position_value,
            daily_trade_count=0,
            daily_pnl_pct=0.0,
            avg_volume=float(stock_context.get("volume") or 0.0) or None,
        )
        new_position_temperature_block = (
            market_temperature_blocks_new_position and current_position_value <= 0
        )

        stage = "top_ranked"
        gate_status = "blocked"
        reliability_blocked = bool(meta.get("must_not_use_for_decision", False))
        reliability_blocked_reason = str(meta.get("blocked_reason") or "")

        if pending_request is not None:
            stage = "pending"
            gate_status = "warn"
        elif (
            not reliability_blocked and passed and action == "allow" and suggested_position_pct > 0
        ):
            stage = "actionable"
            gate_status = "passed"
        elif passed and action == "watch":
            gate_status = "warn"
        elif reliability_blocked and passed:
            gate_status = "warn"

        buy_reasons = [
            {
                "code": "MODEL_SOURCE",
                "text": (
                    f"来源 {meta.get('provider_source') or score.source}，"
                    f"评分日 {score.asof_date.isoformat() if score.asof_date else meta.get('effective_asof_date') or '未知'}，"
                    f"账户池 {meta.get('scope_hash') or '未标记'}"
                ),
            },
            {
                "code": "ALPHA_TOP_RANK",
                "text": f"Alpha 排名第 {score.rank}，评分 {float(score.score):.3f}",
            },
            {"code": "ALPHA_CONFIDENCE", "text": f"评分置信度 {float(score.confidence):.2f}"},
            {
                "code": "REGIME_CONTEXT",
                "text": f"当前 Regime {regime_name}，置信度 {regime_confidence:.0%}",
            },
            {"code": "PULSE_CONTEXT", "text": f"Pulse 综合分 {pulse_composite:+.2f}"},
        ]
        if market_temperature_band:
            temperature_text = (
                f"市场温度 {market_temperature_band} {market_temperature_score:.1f}"
                f"（系数 {market_temperature_factor:.2f}）"
            )
            if market_temperature_threshold_source == "user_override":
                temperature_text += "，使用个人阈值"
            if market_temperature_degraded:
                temperature_text += "，当前仅供参考"
            elif market_temperature_band in {"hot", "overheat", "extreme"}:
                temperature_text += "，已收缩建议仓位"
            buy_reasons.append({"code": "MARKET_TEMPERATURE_CONTEXT", "text": temperature_text})
        factor_basis = self._build_factor_basis(getattr(score, "factors", {}) or {})
        if factor_basis:
            buy_reasons.append(
                {
                    "code": "FACTOR_EVIDENCE",
                    "text": "因子依据：" + "；".join(factor_basis[:3]),
                }
            )
        if actionable_candidate is not None and getattr(actionable_candidate, "thesis", ""):
            buy_reasons.append({"code": "WORKFLOW_THESIS", "text": actionable_candidate.thesis})

        no_buy_reasons = []
        if pending_request is not None:
            no_buy_reasons.append(
                {"code": "ALREADY_PENDING", "text": "已进入待执行队列，避免重复下单。"}
            )
        if reliability_blocked and reliability_blocked_reason:
            no_buy_reasons.append(
                {
                    "code": "ALPHA_RELIABILITY_BLOCK",
                    "text": reliability_blocked_reason,
                }
            )
        if new_position_temperature_block:
            no_buy_reasons.append(
                {
                    "code": "MARKET_TEMPERATURE_BLOCK",
                    "text": market_temperature_blocked_reason
                    or "市场温度已进入 extreme，新仓建议暂时关闭。",
                }
            )
        if policy_state.get("gate_level") in {"L2", "L3"}:
            no_buy_reasons.append(
                {
                    "code": "POLICY_GATE_TIGHT",
                    "text": f"当前政策闸门 {policy_state.get('gate_level')}，新仓需要更严格审查。",
                }
            )
        if action == "watch":
            no_buy_reasons.append({"code": "DECISION_WATCH", "text": decision_text})
        if action == "deny":
            no_buy_reasons.append({"code": "DECISION_DENY", "text": decision_text})
        for violation in violations:
            no_buy_reasons.append({"code": "RISK_BLOCK", "text": violation})
        for warning in warnings:
            no_buy_reasons.append({"code": "RISK_WARN", "text": warning})
        if suggested_position_pct <= 0:
            no_buy_reasons.append(
                {"code": "NO_POSITION_SIZE", "text": "当前账户上下文未形成正向建议仓位。"}
            )

        invalidation_rule = {
            "summary": f"若跌出 Top {max(score.rank + 5, 10)}、政策/风控转差或评分跌破 0.55，则当前候选失效。",
            "conditions": [
                f"Alpha 评分跌出 Top {max(score.rank + 5, 10)}",
                "政策闸门提升至 L2/L3",
                "预交易风控由通过变为阻断",
                "当前候选进入待执行队列或被 workflow 显式否决",
            ],
        }
        if stage == "actionable" and new_position_temperature_block:
            stage = "top_ranked"
            gate_status = "warn"

        recommendation_ready = (
            stage == "actionable" and not reliability_blocked and not new_position_temperature_block
        )
        decision_usable = not reliability_blocked
        not_actionable_reason = (
            "当前已在待执行队列中。"
            if pending_request is not None
            else (
                (market_temperature_blocked_reason or "市场温度已进入 extreme，新仓建议暂时关闭。")
                if new_position_temperature_block
                else reliability_blocked_reason
                or (
                    no_buy_reasons[0]["text"]
                    if no_buy_reasons
                    else "当前候选仅供研究，不构成可执行推荐。"
                )
            )
        )

        return {
            "code": code,
            "name": stock_context.get("name") or code,
            "sector": stock_context.get("sector") or "",
            "market": stock_context.get("market") or "",
            "roe": stock_context.get("roe"),
            "debt_ratio": stock_context.get("debt_ratio"),
            "revenue_growth": stock_context.get("revenue_growth"),
            "profit_growth": stock_context.get("profit_growth"),
            "pe": stock_context.get("pe"),
            "pb": stock_context.get("pb"),
            "ps": stock_context.get("ps"),
            "dividend_yield": stock_context.get("dividend_yield"),
            "report_date": stock_context.get("report_date"),
            "valuation_trade_date": stock_context.get("valuation_trade_date"),
            "score": round(float(score.score), 4),
            "alpha_score": round(float(score.score), 4),
            "rank": int(score.rank),
            "source": score.source,
            "confidence": round(float(score.confidence), 3),
            "factors": score.factors,
            "asof_date": score.asof_date.isoformat() if score.asof_date else None,
            "trade_date": stock_context.get("trade_date"),
            "stage": stage,
            "stage_label": {
                "top_ranked": "Alpha Top 候选/排名",
                "actionable": "可行动候选",
                "pending": "待执行队列",
            }.get(stage, "Alpha Top 候选/排名"),
            "gate_status": gate_status,
            "gate_reasons": violations or warnings or decision_codes,
            "suggested_position_pct": round(suggested_position_pct, 2),
            "suggested_notional": round(suggested_notional, 2),
            "suggested_quantity": int(suggested_quantity or 0),
            "risk_snapshot": {
                "policy_gate_level": policy_state.get("gate_level"),
                "regime_name": regime_name,
                "regime_confidence": regime_confidence,
                "pulse_composite": pulse_composite,
                "pulse_warning": pulse_warning,
                "market_temperature_score": market_temperature_score,
                "market_temperature_band": market_temperature_band,
                "market_temperature_factor": market_temperature_factor,
                "market_temperature_threshold_source": market_temperature_threshold_source,
                "market_temperature_degraded": market_temperature_degraded,
                "market_temperature_blocks_new_position": market_temperature_blocks_new_position,
                "risk_checks": details,
                "sizing_explain": sizing_explain,
            },
            "buy_reasons": buy_reasons,
            "buy_reason_summary": "；".join(reason["text"] for reason in buy_reasons[:3]),
            "recommendation_basis": {
                "alpha_scope": meta.get("alpha_scope"),
                "provider_source": meta.get("provider_source") or score.source,
                "score_source": score.source,
                "universe_id": meta.get("universe_id"),
                "pool_mode": meta.get("pool_mode"),
                "scope_hash": meta.get("scope_hash"),
                "scope_label": meta.get("scope_label"),
                "asof_date": score.asof_date.isoformat() if score.asof_date else None,
                "requested_trade_date": meta.get("requested_trade_date"),
                "effective_asof_date": meta.get("effective_asof_date"),
                "rank": int(score.rank),
                "score": round(float(score.score), 4),
                "confidence": round(float(score.confidence), 3),
                "factor_basis": factor_basis,
                "scope_verification_status": meta.get("scope_verification_status"),
                "freshness_status": meta.get("freshness_status"),
                "result_age_days": meta.get("result_age_days"),
                "verified_scope_hash": meta.get("verified_scope_hash"),
                "verified_asof_date": meta.get("verified_asof_date"),
                "latest_available_qlib_result": bool(
                    meta.get("latest_available_qlib_result", False)
                ),
                "derived_from_broader_cache": bool(meta.get("derived_from_broader_cache", False)),
                "trade_date_adjusted": bool(meta.get("trade_date_adjusted", False)),
                "research_only": bool(meta.get("research_only", False)),
                "must_not_use_for_decision": bool(meta.get("must_not_use_for_decision", False)),
                "blocked_reason": reliability_blocked_reason,
                "market_temperature_score": market_temperature_score,
                "market_temperature_band": market_temperature_band,
                "market_temperature_factor": market_temperature_factor,
                "market_temperature_threshold_source": market_temperature_threshold_source,
                "market_temperature_degraded": market_temperature_degraded,
                "market_temperature_blocked_reason": market_temperature_blocked_reason,
                "market_temperature_blocks_new_position": market_temperature_blocks_new_position,
            },
            "no_buy_reasons": no_buy_reasons,
            "no_buy_reason_summary": "；".join(reason["text"] for reason in no_buy_reasons[:3]),
            "invalidation_rule": invalidation_rule,
            "invalidation_summary": invalidation_rule["summary"],
            "source_candidate_id": getattr(actionable_candidate, "id", None),
            "source_recommendation_id": getattr(pending_request, "id", None),
            "recommendation_ready": recommendation_ready,
            "must_not_treat_as_recommendation": not recommendation_ready,
            "decision_usable": decision_usable,
            "must_not_use_for_decision": not decision_usable,
            "blocked_reason": not_actionable_reason,
            "not_actionable_reason": not_actionable_reason,
            "extra_payload": {
                "decision_action": action,
                "decision_codes": decision_codes,
                "decision_text": decision_text,
            },
        }

    def _build_factor_basis(self, factors: dict[str, Any]) -> list[str]:
        basis: list[str] = []
        for key, raw_value in factors.items():
            if raw_value in (None, ""):
                continue
            try:
                value = float(raw_value)
            except (TypeError, ValueError):
                basis.append(f"{key}={raw_value}")
                continue
            basis.append(f"{key}={value:.3f}")
        return basis

    def _serialize_pending_request(
        self, *, request_model, stock_context: dict[str, Any]
    ) -> dict[str, Any]:
        code = str(request_model.asset_code).upper()
        reason = str(getattr(request_model, "reason", "") or "")
        return {
            "request_id": getattr(request_model, "request_id", ""),
            "code": code,
            "name": stock_context.get("name") or code,
            "stage": "pending",
            "stage_label": "待执行队列",
            "gate_status": "warn",
            "rank": 0,
            "alpha_score": 0.0,
            "confidence": 0.0,
            "source": "workflow",
            "buy_reasons": [{"code": "REQUEST_APPROVED", "text": "该标的已通过决策审批。"}],
            "no_buy_reasons": [{"code": "ALREADY_PENDING", "text": "当前已在待执行队列中。"}],
            "invalidation_rule": {
                "summary": "若执行失败或审批撤回，该待执行请求失效。",
                "conditions": ["审批被撤回", "执行状态转为取消/失败后未重试"],
            },
            "suggested_position_pct": float(getattr(request_model, "position_pct", 0.0) or 0.0),
            "suggested_notional": float(getattr(request_model, "notional", 0.0) or 0.0),
            "suggested_quantity": float(getattr(request_model, "quantity", 0.0) or 0.0),
            "source_recommendation_id": getattr(request_model, "id", None),
            "reason_summary": reason,
            "recommendation_ready": False,
            "must_not_use_for_decision": True,
            "blocked_reason": "当前已在待执行队列中。",
            "risk_snapshot": {
                "execution_status": getattr(request_model, "execution_status", ""),
                "reason": reason,
            },
            "extra_payload": {
                "execution_status": getattr(request_model, "execution_status", ""),
            },
        }

    def _persist_history(
        self,
        *,
        user_id: int,
        portfolio_id: int | None,
        portfolio_name: str,
        scope,
        alpha_result,
        meta: dict[str, Any],
        snapshots: list[dict[str, Any]],
    ) -> int | None:
        try:
            run = self.history_repo.upsert_run(
                user_id=user_id,
                portfolio_id=portfolio_id,
                portfolio_name=portfolio_name,
                trade_date=django_timezone.localdate(),
                scope_hash=scope.scope_hash,
                scope_label=scope.display_label,
                scope_metadata=scope.to_dict(),
                model_hash=meta.get("model_hash", ""),
                source=meta.get("source", "none"),
                provider_source=meta.get("provider_source", ""),
                requested_trade_date=(
                    date.fromisoformat(meta["requested_trade_date"])
                    if meta.get("requested_trade_date")
                    else None
                ),
                effective_asof_date=(
                    date.fromisoformat(meta["effective_asof_date"])
                    if meta.get("effective_asof_date")
                    else None
                ),
                uses_cached_data=bool(meta.get("uses_cached_data", False)),
                cache_reason=str(meta.get("cache_reason") or ""),
                fallback_reason=str(meta.get("fallback_reason") or ""),
                meta=dict(getattr(alpha_result, "metadata", {}) or {}),
            )
            self.history_repo.replace_snapshots(
                run=run,
                snapshots=[
                    {
                        "stock_code": item["code"],
                        "stock_name": item.get("name", ""),
                        "stage": item.get("stage", "top_ranked"),
                        "gate_status": item.get("gate_status", "blocked"),
                        "rank": item.get("rank", 0),
                        "alpha_score": item.get("alpha_score", 0.0),
                        "confidence": item.get("confidence", 0.0),
                        "source": item.get("source", ""),
                        "buy_reasons": item.get("buy_reasons", []),
                        "no_buy_reasons": item.get("no_buy_reasons", []),
                        "invalidation_rule": item.get("invalidation_rule", {}),
                        "risk_snapshot": item.get("risk_snapshot", {}),
                        "suggested_position_pct": item.get("suggested_position_pct", 0.0),
                        "suggested_notional": item.get("suggested_notional", 0.0),
                        "suggested_quantity": item.get("suggested_quantity", 0.0),
                        "source_candidate_id": item.get("source_candidate_id"),
                        "source_recommendation_id": item.get("source_recommendation_id"),
                        "extra_payload": item.get("extra_payload", {}),
                    }
                    for item in snapshots
                ],
            )
            return run.id
        except Exception as exc:
            logger.warning("Failed to persist alpha homepage history: %s", exc, exc_info=True)
            return None

    def _serialize_recent_runs(self, runs) -> list[dict[str, Any]]:
        payload = []
        for run in runs:
            cache_reason = str(run.cache_reason or "")
            meta = dict(getattr(run, "meta", {}) or {})
            legacy_hardcoded_fallback = (
                meta.get("fallback_mode") == "homepage_market_cache_fallback"
                or "csi300 全局缓存" in cache_reason
                or "全局缓存排名" in cache_reason
            )
            payload.append(
                {
                    "id": run.id,
                    "portfolio_id": run.portfolio_id,
                    "portfolio_name": run.portfolio_name,
                    "trade_date": run.trade_date.isoformat(),
                    "scope_label": run.scope_label,
                    "source": run.source,
                    "provider_source": run.provider_source,
                    "uses_cached_data": run.uses_cached_data,
                    "effective_asof_date": (
                        run.effective_asof_date.isoformat() if run.effective_asof_date else None
                    ),
                    "cache_reason": cache_reason,
                    "legacy_hardcoded_fallback": legacy_hardcoded_fallback,
                    "reliability_note": (
                        "旧版硬编码回退记录，仅用于审计回溯，不作为当前推荐依据。"
                        if legacy_hardcoded_fallback
                        else ""
                    ),
                }
            )
        return payload
