210 lines
8.0 KiB
Python
210 lines
8.0 KiB
Python
"""基金行情业务服务：合并外部实时行情和历史净值，向上层提供稳定结果。"""
|
|
|
|
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"),
|
|
"债券型": ("007161", "003776", "020281"),
|
|
"混合型": ("018019", "014189", "018020"),
|
|
"股票型": (
|
|
"020553", "016449", "008854", "008264", "008736", "010592", "160127",
|
|
"588890", "020839", "589700", "159382", "159511", "002900", "021958",
|
|
"159948", "009059", "001421",
|
|
),
|
|
}
|
|
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(
|
|
EastmoneyAdapterFactory.create(), runtime_config=config
|
|
).query(arguments)
|
|
return [item.model_dump(mode="json") for item in result]
|
|
|
|
|
|
class EastmoneyAdapterFactory:
|
|
"""延迟创建外部 Adapter，避免导入底座时建立网络连接。"""
|
|
|
|
@staticmethod
|
|
def create() -> FundMarketAdapter:
|
|
from app.infrastructure.fund_market_adapter import EastmoneyFundAdapter
|
|
|
|
return EastmoneyFundAdapter()
|