group_fqcd_jr/app/service/fund_quote_service.py

"""基金行情业务服务：合并外部实时行情和历史净值，向上层提供稳定结果。"""

import json
from datetime import UTC, date, datetime, time
from typing import Any, Literal
from zoneinfo import ZoneInfo

from app.core.contracts import RequestContext
from app.core.fund_contracts import FundQuote, FundQuoteQuery
from app.infrastructure.fund_market_adapter import FundMarketAdapter
from app.infrastructure.fund_quote_cache import FundQuoteCache

FUND_TYPE_GROUPS: dict[str, tuple[str, ...]] = {
    "货币型": ("202308", "020480", "511810"),
    "债券型": ("007161", "003776", "020281", "511070", "159700", "160128", "160129"),
    "混合型": ("018019", "014189", "018020", "160105", "160142", "160143", "501062"),
    "股票型": (
        "020553", "016449", "008854", "008264", "008736", "010592", "160127",
        "588890", "020839", "589700", "159382", "159511", "002900", "021958",
        "159948", "009059", "001421", "510500",
    ),
    "QDII": ("501018", "159329", "159615", "159687"),
}
SOUTHERN_FUND_CODES: tuple[str, ...] = tuple(
    dict.fromkeys(code for codes in FUND_TYPE_GROUPS.values() for code in codes)
)
FUND_TYPE_BY_CODE = {
    code: fund_type for fund_type, codes in FUND_TYPE_GROUPS.items() for code in codes
}
SHANGHAI_TZ = ZoneInfo("Asia/Shanghai")


class FundQuoteRuntimeConfig:
    """行情运行参数；缺失或非法配置时使用安全默认值。"""

    def __init__(
        self,
        *,
        allowed_codes: tuple[str, ...] = SOUTHERN_FUND_CODES,
        intraday_cache_ttl: int = 60,
        closing_cache_ttl: int = 900,
    ) -> None:
        self.allowed_codes = allowed_codes
        self.intraday_cache_ttl = intraday_cache_ttl
        self.closing_cache_ttl = closing_cache_ttl

    @classmethod
    def from_mapping(cls, value: object) -> "FundQuoteRuntimeConfig":
        if not isinstance(value, dict):
            return cls()
        raw_codes = value.get("allowed_codes", SOUTHERN_FUND_CODES)
        if isinstance(raw_codes, list):
            codes = tuple(
                code for code in raw_codes
                if isinstance(code, str) and len(code) == 6 and code.isdigit()
            )
        else:
            codes = SOUTHERN_FUND_CODES
        if not codes:
            codes = SOUTHERN_FUND_CODES
        def ttl(name: str, default: int) -> int:
            raw = value.get(name, default)
            return raw if isinstance(raw, int) and 1 <= raw <= 86400 else default
        return cls(
            allowed_codes=codes,
            intraday_cache_ttl=ttl("intraday_cache_ttl", 60),
            closing_cache_ttl=ttl("closing_cache_ttl", 900),
        )


class FundQuoteService:
    def __init__(
        self,
        adapter: FundMarketAdapter,
        cache: FundQuoteCache | None = None,
        runtime_config: FundQuoteRuntimeConfig | None = None,
    ) -> None:
        self.adapter = adapter
        self.cache = cache
        self.runtime_config = runtime_config or FundQuoteRuntimeConfig()

    async def query(
        self,
        query: FundQuoteQuery,
        *,
        now: datetime | None = None,
    ) -> list[FundQuote]:
        if query.fund_type is not None and query.fund_type not in FUND_TYPE_GROUPS:
            raise ValueError("基金类型必须是货币型、债券型、混合型或股票型")
        configured = set(self.runtime_config.allowed_codes)
        source_codes = (
            FUND_TYPE_GROUPS[query.fund_type]
            if query.fund_type is not None
            else SOUTHERN_FUND_CODES
        )
        available = tuple(code for code in source_codes if code in configured)
        selected = [code for code in query.fund_codes if code in available]
        if not query.fund_codes:
            selected = list(available)
        selected = selected[: query.limit]
        if not selected:
            return []
        current = (now or datetime.now(UTC)).astimezone(SHANGHAI_TZ)
        target_date = query.target_date or current.date()
        intraday = self.is_market_trading_time(current)
        cache_key = self._cache_key(query, target_date, intraday)
        if self.cache is not None:
            cached = await self.cache.get(cache_key)
            if cached.value:
                try:
                    values = [FundQuote.model_validate(item) for item in json.loads(cached.value)]
                    return [item.model_copy(update={"quote_source": "cache"}) for item in values]
                except (TypeError, ValueError):
                    pass
        names = await self.adapter.fetch_names(selected)
        live = await self.adapter.fetch_quotes(selected) if intraday else {}
        history = {
            code: await self.adapter.fetch_history(code, target_date) for code in selected
        }
        quote_time = current
        result = [
            self._merge(code, names, live, history, quote_time, intraday)
            for code in selected
        ]
        if self.cache is not None:
            ttl = (self.runtime_config.intraday_cache_ttl
                   if intraday else self.runtime_config.closing_cache_ttl)
            await self.cache.set(
                cache_key,
                "[" + ",".join(item.model_dump_json() for item in result) + "]",
                ttl,
            )
        return result

    @staticmethod
    def _cache_key(query: FundQuoteQuery, target_date: date, intraday: bool) -> str:
        codes = ",".join(query.fund_codes) if query.fund_codes else "all"
        return (
            f"fund_quote:v1:{codes}:{query.fund_type or 'all'}:"
            f"{target_date}:{int(intraday)}:{query.limit}"
        )

    @staticmethod
    def is_market_trading_time(current: datetime) -> bool:
        local = current.astimezone(SHANGHAI_TZ)
        if local.weekday() >= 5:
            return False
        value = local.time()
        return time(9, 30) <= value <= time(11, 30) or time(13) <= value <= time(15)

    @staticmethod
    def _merge(
        code: str,
        names: dict[str, str],
        live: dict[str, dict[str, Any]],
        history: dict[str, dict[str, Any]],
        quote_time: datetime,
        intraday: bool,
    ) -> FundQuote:
        current = live.get(code, {})
        previous = history.get(code, {})
        use_live = current.get("nav") is not None
        degraded = bool(previous.get("degraded")) or (intraday and not use_live)
        source: Literal["eastmoney", "cache", "degraded"] = (
            "eastmoney" if (use_live or not degraded) else "degraded"
        )
        return FundQuote(
            fund_code=code,
            fund_name=names.get(code, f"基金 {code}"),
            fund_type=FUND_TYPE_BY_CODE.get(code),
            nav=current.get("nav") if use_live else previous.get("nav"),
            nav_date=previous.get("nav_date"),
            daily_change=current.get("daily_change") if use_live else previous.get("daily_change"),
            half_year_return=previous.get("half_year_return"),
            one_year_return=previous.get("one_year_return"),
            year_to_date_return=previous.get("year_to_date_return"),
            since_inception_return=previous.get("since_inception_return"),
            quote_time=quote_time,
            quote_source=source,
            is_intraday=intraday,
            degraded=degraded,
        )


async def query_fund_quote_tool(
    arguments: FundQuoteQuery, context: RequestContext
) -> list[dict[str, Any]]:
    """ToolExecutor 使用的公共只读行情工具处理器。"""
    del context  # 权限、审计和来源由 ToolExecutor 统一处理。
    from app.service.runtime_config_service import load_fund_quote_config

    try:
        mapping = await load_fund_quote_config()
    except Exception:
        mapping = {}
    config = FundQuoteRuntimeConfig.from_mapping(mapping)
    result = await FundQuoteService(
        HqAdapterFactory.create(), runtime_config=config
    ).query(arguments)
    return [item.model_dump(mode="json") for item in result]


class HqAdapterFactory:
    """延迟创建 hq.py 适配器，避免导入底座时建立网络连接。"""

    @staticmethod
    def create() -> FundMarketAdapter:
        from app.infrastructure.fund_market_adapter import HqFundMarketAdapter

        return HqFundMarketAdapter()