225 lines
10 KiB
Python
225 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""南方基金指定产品行情模块。"""
|
|
from __future__ import annotations
|
|
import re
|
|
import time
|
|
import logging
|
|
from datetime import date, datetime, time as clock_time
|
|
from typing import Any
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
NAV_API = "https://api.fund.eastmoney.com/f10/lsjz"
|
|
RETURN_API = "https://api.fund.eastmoney.com/pinzhong/LJSYLZS"
|
|
QUOTE_API = "https://push2.eastmoney.com/api/qt/ulist.np/get"
|
|
DETAIL_API = "https://fund.eastmoney.com/pingzhongdata/{code}.js"
|
|
REQUEST_TIMEOUT = 12.0
|
|
MAX_FUNDS_PER_CALL = 1000
|
|
SOUTHERN_FUND_CODES = tuple(dict.fromkeys("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".split()))
|
|
FUND_TYPE_GROUPS = {
|
|
"货币型": ("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"),
|
|
}
|
|
FUND_TYPE_BY_CODE = {code: fund_type for fund_type, codes in FUND_TYPE_GROUPS.items() for code in codes}
|
|
HEADERS = {"User-Agent": "Mozilla/5.0", "Referer": "https://fund.eastmoney.com/"}
|
|
_history_date: str | None = None
|
|
_history_cache: dict[str, dict[str, str | None]] = {}
|
|
_name_cache: dict[str, str] = {}
|
|
|
|
|
|
def is_market_trading_time(now: datetime | None = None) -> bool:
|
|
"""判断中国大陆工作日盘中时段。"""
|
|
current = now or datetime.now()
|
|
if current.weekday() >= 5:
|
|
return False
|
|
current_time = current.time()
|
|
return (clock_time(9, 30) <= current_time <= clock_time(11, 30)
|
|
or clock_time(13, 0) <= current_time <= clock_time(15, 0))
|
|
|
|
|
|
def get_southern_fund_market(target_date: str | None = None, limit: int | None = None, fund_type: str | None = None) -> list[dict[str, Any]]:
|
|
"""获取指定南方基金的完整行情表。
|
|
|
|
每次调用刷新整张表的实时行情；历史收益同一日期只请求一次并保存在进程缓存。
|
|
limit 不传时返回 SOUTHERN_FUND_CODES 中的全部产品。
|
|
"""
|
|
query_date = target_date or date.today().isoformat()
|
|
date.fromisoformat(query_date)
|
|
if fund_type and fund_type not in FUND_TYPE_GROUPS:
|
|
raise ValueError("基金类型必须是货币型、债券型、混合型或股票型")
|
|
available_codes = FUND_TYPE_GROUPS[fund_type] if fund_type else SOUTHERN_FUND_CODES
|
|
count = len(available_codes) if limit is None else min(max(limit, 1), MAX_FUNDS_PER_CALL)
|
|
codes = list(available_codes[:count])
|
|
names = _get_names(codes)
|
|
history = _get_history(codes, query_date)
|
|
quotes = _get_quotes(codes) if is_market_trading_time() else {}
|
|
now = time.strftime("%Y-%m-%d")
|
|
rows = []
|
|
for code in codes:
|
|
old = history.get(code, {})
|
|
live = quotes.get(code, {})
|
|
rows.append({
|
|
"基金代码": code, "基金名称": names.get(code, f"南方基金 {code}"),
|
|
"基金类型": FUND_TYPE_BY_CODE.get(code, "未分类"),
|
|
"基金净值": live.get("基金净值") or old.get("基金净值"),
|
|
"日期": live.get("日期") or old.get("日期"),
|
|
"日涨幅": live.get("日涨幅") or old.get("日涨幅"),
|
|
"最近半年": old.get("最近半年"), "最近一年": old.get("最近一年"),
|
|
"今年以来": old.get("今年以来"), "成立以来": old.get("成立以来"),
|
|
"行情时间": now,
|
|
"行情来源": "盘中实时行情" if live.get("基金净值") else "收盘后最新净值",
|
|
"是否盘中": is_market_trading_time(),
|
|
})
|
|
return rows
|
|
|
|
|
|
def _get_names(codes: list[str]) -> dict[str, str]:
|
|
for code in codes:
|
|
if code in _name_cache:
|
|
continue
|
|
try:
|
|
response = httpx.get(DETAIL_API.format(code=code), headers=HEADERS, timeout=REQUEST_TIMEOUT)
|
|
response.raise_for_status()
|
|
match = re.search(r"var\s+fS_name\s*=\s*[\"']([^\"']+)", response.text)
|
|
_name_cache[code] = match.group(1).strip() if match else f"南方基金 {code}"
|
|
except (httpx.HTTPError, UnicodeError) as exc:
|
|
logger.warning("基金名称接口失败 code=%s error=%s", code, type(exc).__name__)
|
|
_name_cache[code] = f"南方基金 {code}"
|
|
return {code: _name_cache.get(code, f"南方基金 {code}") for code in codes}
|
|
|
|
|
|
def _get_quotes(codes: list[str]) -> dict[str, dict[str, str | None]]:
|
|
secids = ",".join(("1." if code.startswith(("5", "6", "9")) else "0.") + code for code in codes)
|
|
try:
|
|
response = httpx.get(QUOTE_API, params={"fltt": 2, "invt": 2, "fields": "f12,f2,f3", "secids": secids}, headers=HEADERS, timeout=REQUEST_TIMEOUT)
|
|
response.raise_for_status()
|
|
items = ((response.json().get("data") or {}).get("diff") or [])
|
|
except (httpx.HTTPError, ValueError, TypeError) as exc:
|
|
logger.warning("实时行情接口失败 count=%s error=%s", len(codes), type(exc).__name__)
|
|
return {}
|
|
return {str(item["f12"]): {"基金净值": str(item["f2"]) if item.get("f2") not in (None, "-") else None, "日期": None, "日涨幅": f"{item.get('f3')}%" if item.get("f3") not in (None, "-") else None} for item in items if item.get("f12")}
|
|
|
|
|
|
def _get_history(codes: list[str], query_date: str) -> dict[str, dict[str, str | None]]:
|
|
global _history_date, _history_cache
|
|
if _history_date == query_date and all(code in _history_cache for code in codes):
|
|
return _history_cache
|
|
result = {}
|
|
for code in codes:
|
|
try:
|
|
result[code] = _get_history_snapshot(code, query_date)
|
|
except (httpx.HTTPError, ValueError, TypeError) as exc:
|
|
logger.warning("历史净值计算失败 code=%s error=%s", code, type(exc).__name__)
|
|
result[code] = _empty_returns()
|
|
_history_date, _history_cache = query_date, result
|
|
return result
|
|
|
|
|
|
def _get_history_snapshot(fund_code: str, query_date: str) -> dict[str, str | None]:
|
|
"""读取净值与累计收益率快照。"""
|
|
latest_records = _fetch_nav_records(fund_code, all_pages=False)
|
|
valid = _valid_records(latest_records)
|
|
if not valid:
|
|
return _empty_returns()
|
|
target = date.fromisoformat(query_date)
|
|
latest_date, _, latest = next((item for item in valid if item[0] == target), valid[0])
|
|
return {
|
|
"基金净值": latest.get("DWJZ"),
|
|
"日期": latest.get("FSRQ"),
|
|
"日涨幅": _format_percent(latest.get("JZZZL")),
|
|
"最近半年": _fetch_return_rate(fund_code, "6月"),
|
|
"最近一年": _fetch_return_rate(fund_code, "1年"),
|
|
"今年以来": _fetch_return_rate(fund_code, "今年来"),
|
|
"成立以来": _fetch_return_rate(fund_code, "成立来"),
|
|
}
|
|
|
|
|
|
def _fetch_nav_page(fund_code: str, page_index: int = 1, start_date: str | None = None, end_date: str | None = None) -> tuple[list[dict[str, Any]], int | None]:
|
|
"""读取一页历史净值，并返回接口提供的总条数。"""
|
|
response = httpx.get(NAV_API, params={"fundCode": fund_code, "pageIndex": page_index, "pageSize": 30, "startDate": start_date or "", "endDate": end_date or ""}, headers=HEADERS, timeout=REQUEST_TIMEOUT)
|
|
response.raise_for_status()
|
|
data = response.json().get("Data") or {}
|
|
records = data.get("LSJZList") or []
|
|
total = data.get("TotalCount") or data.get("totalCount")
|
|
try:
|
|
total = int(total) if total is not None else None
|
|
except (TypeError, ValueError):
|
|
total = None
|
|
return records, total
|
|
|
|
|
|
def _fetch_nav_records(fund_code: str, start_date: str | None = None, end_date: str | None = None, all_pages: bool = False) -> list[dict[str, Any]]:
|
|
"""调用历史净值接口；区间收益需要时读取完整分页，避免重复使用同一基准日。"""
|
|
records, total = _fetch_nav_page(fund_code, 1, start_date, end_date)
|
|
if not all_pages or not total or total <= len(records):
|
|
return records
|
|
page_count = min((total + 29) // 30, 40)
|
|
for page_index in range(2, page_count + 1):
|
|
page_records, _ = _fetch_nav_page(fund_code, page_index, start_date, end_date)
|
|
if not page_records:
|
|
break
|
|
records.extend(page_records)
|
|
return records
|
|
|
|
|
|
def _fetch_return_rate(fund_code: str, period: str) -> str | None:
|
|
"""通过东方财富累计收益率接口读取指定区间的最新收益率。"""
|
|
period_map = {
|
|
"1月": "m",
|
|
"3月": "q",
|
|
"6月": "hy",
|
|
"1年": "y",
|
|
"3年": "try",
|
|
"5年": "fiy",
|
|
"今年来": "sy",
|
|
"成立来": "se",
|
|
}
|
|
try:
|
|
response = httpx.get(
|
|
RETURN_API,
|
|
params={"fundCode": fund_code, "indexcode": "000300", "type": period_map[period]},
|
|
headers={"Referer": "https://fund.eastmoney.com/"},
|
|
timeout=REQUEST_TIMEOUT,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
series = (((payload.get("Data") or [{}])[0]).get("data") or [])
|
|
if not series:
|
|
return None
|
|
latest = series[-1]
|
|
if isinstance(latest, dict):
|
|
value = latest.get("y")
|
|
elif isinstance(latest, (list, tuple)) and len(latest) >= 2:
|
|
value = latest[1]
|
|
else:
|
|
value = None
|
|
if value in (None, ""):
|
|
return None
|
|
return _format_percent(value)
|
|
except (httpx.HTTPError, ValueError, TypeError, KeyError, IndexError) as exc:
|
|
logger.warning("累计收益率接口失败 code=%s period=%s error=%s", fund_code, period, type(exc).__name__)
|
|
return None
|
|
|
|
def _empty_returns() -> dict[str, str | None]:
|
|
return {key: None for key in ("基金净值", "日期", "日涨幅", "最近半年", "最近一年", "今年以来", "成立以来")}
|
|
|
|
|
|
def _valid_records(records: list[dict[str, Any]]) -> list[tuple[date, float, dict[str, Any]]]:
|
|
valid = []
|
|
for item in records:
|
|
try:
|
|
valid.append((date.fromisoformat(item["FSRQ"]), float(item["DWJZ"]), item))
|
|
except (KeyError, TypeError, ValueError):
|
|
continue
|
|
return sorted(valid, key=lambda item: item[0], reverse=True)
|
|
|
|
|
|
def _format_percent(value: Any) -> str | None:
|
|
if value in (None, ""):
|
|
return None
|
|
text = str(value).strip()
|
|
return text if text.endswith("%") else f"{text}%"
|