"""南方基金指定产品行情模块。"""
from __future__ import annotations
import logging
import re
import time
from datetime import date, datetime
from datetime import time as clock_time
from html import unescape
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"
TENCENT_QUOTE_API = "https://qt.gtimg.cn/q="
DETAIL_API = "https://fund.eastmoney.com/pingzhongdata/{code}.js"
SOUTHERN_COMPANY_API = "https://fund.eastmoney.com/company/80000220.html"
REQUEST_TIMEOUT = 12.0
MAX_FUNDS_PER_CALL = 1000
FUND_TYPE_GROUPS = {
"货币型": ("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(
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
}
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] = {}
class ExchangeQuoteSourceError(RuntimeError):
"""A public exchange quote provider could not supply a usable response."""
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,
fund_codes: list[str] | tuple[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
if fund_codes is not None:
requested = tuple(dict.fromkeys(fund_codes))
if any(code not in SOUTHERN_FUND_CODES for code in requested):
raise ValueError("基金代码不在南方基金白名单内")
available_codes = tuple(code for code in requested if code in available_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_southern_fund_nav_history(
fund_code: str, start_date: str, end_date: str
) -> list[dict[str, str]]:
"""Return validated historical unit-NAV observations for an allowed fund."""
if fund_code not in SOUTHERN_FUND_CODES:
raise ValueError("基金代码不在南方基金白名单内")
start = date.fromisoformat(start_date)
end = date.fromisoformat(end_date)
if start > end:
raise ValueError("开始日期不能晚于结束日期")
records = _fetch_nav_records(
fund_code, start_date=start.isoformat(), end_date=end.isoformat(), all_pages=True
)
observations: list[dict[str, str]] = []
for record in records:
value_date = str(record.get("FSRQ") or "")
nav = str(record.get("DWJZ") or "").strip()
try:
parsed_date = date.fromisoformat(value_date)
if parsed_date < start or parsed_date > end or float(nav) <= 0:
continue
except ValueError:
continue
observations.append({"fund_code": fund_code, "trade_date": value_date, "nav": nav})
return sorted(observations, key=lambda item: item["trade_date"])
def get_southern_fund_catalog(
fund_codes: list[str] | tuple[str, ...],
) -> list[dict[str, Any]]:
"""Return Southern Fund's public catalogue rows.
The company directory supplies fund type and reported asset scale, which are
needed to select test products. It is reference data only: product risk is
intentionally not inferred here because the source does not publish a
channel-independent R1-R5 suitability rating.
"""
requested = tuple(dict.fromkeys(fund_codes))
if not requested:
return []
response = httpx.get(SOUTHERN_COMPANY_API, headers=HEADERS, timeout=REQUEST_TIMEOUT)
response.raise_for_status()
content = response.content.decode("utf-8")
scale_as_of_date = _company_scale_as_of_date(content)
rows = []
for code in requested:
row = _company_catalog_row(content, code)
if row is not None:
row["scale_as_of_date"] = scale_as_of_date
rows.append(row)
return rows
def get_southern_exchange_catalog(
fund_codes: list[str] | tuple[str, ...],
) -> list[dict[str, Any]]:
"""Backward-compatible alias for callers that only request exchange codes."""
return get_southern_fund_catalog(fund_codes)
def _company_catalog_row(content: str, fund_code: str) -> dict[str, Any] | None:
marker = f'class="code">{fund_code}'
position = content.find(marker)
if position < 0:
return None
start = content.rfind("
", position)
if start < 0 or end < 0:
return None
row = content[start:end + len("
")]
name_match = re.search(r'class="name" title="([^"]+)"', row)
cells = [
_html_cell_text(match.group(1))
for match in re.finditer(r"]*>(.*?) | ", row, flags=re.DOTALL)
]
if name_match is None or len(cells) < 10:
return None
return {
"fund_code": fund_code,
"fund_name": unescape(name_match.group(1)).strip(),
"fund_type": cells[2],
"nav_date": _catalog_nav_date(cells[3]),
"nav": cells[4],
"fund_asset_scale_billion": cells[9],
"trading_venue": cells[11] if len(cells) > 11 else "场内交易",
}
def _company_scale_as_of_date(content: str) -> str | None:
match = re.search(r"(?:数据截止|截止日期)[::]\s*(20\d{2}-\d{2}-\d{2})", content)
return match.group(1) if match else None
def _catalog_nav_date(value: str) -> str | None:
match = re.fullmatch(r"(\d{2})-(\d{2})", value)
if match is None:
return None
return f"{date.today().year}-{match.group(1)}-{match.group(2)}"
def _html_cell_text(value: str) -> str:
return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", unescape(value))).strip()
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_southern_exchange_quotes(
fund_codes: list[str] | tuple[str, ...],
) -> dict[str, dict[str, str | None]]:
"""Backward-compatible Eastmoney quote lookup that degrades to an empty result."""
try:
return fetch_southern_exchange_quotes_eastmoney(fund_codes)
except ExchangeQuoteSourceError as exc:
logger.warning("exchange quote endpoint failed error=%s", type(exc).__name__)
return {}
def fetch_southern_exchange_quotes_eastmoney(
fund_codes: list[str] | tuple[str, ...],
) -> dict[str, dict[str, str | None]]:
"""Return Eastmoney exchange quotes and raise on a provider-level failure."""
requested = tuple(dict.fromkeys(fund_codes))
if any(code not in SOUTHERN_FUND_CODES for code in requested):
raise ValueError("fund code is not in the Southern Fund whitelist")
if not requested:
return {}
secids = ",".join(
("1." if code.startswith(("5", "6", "9")) else "0.") + code for code in requested
)
try:
response = httpx.get(
QUOTE_API,
params={
"fltt": 2,
"invt": 2,
"fields": "f12,f2,f3,f5,f6,f17,f18",
"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:
raise ExchangeQuoteSourceError("eastmoney quote request failed") from exc
def value(item: dict[str, Any], field: str) -> str | None:
raw = item.get(field)
return str(raw) if raw not in (None, "-") else None
return {
str(item["f12"]): {
"last_price": value(item, "f2"),
"change_pct": value(item, "f3"),
"volume": value(item, "f5"),
"turnover_amount": value(item, "f6"),
"open_price": value(item, "f17"),
"previous_close": value(item, "f18"),
}
for item in items
if item.get("f12")
}
def fetch_southern_exchange_quotes_tencent(
fund_codes: list[str] | tuple[str, ...],
) -> dict[str, dict[str, str | None]]:
"""Return Tencent Finance exchange quotes as an independent fallback source.
Tencent publishes volume in lots. Turnover is deliberately left unset
because its payload does not provide a field with compatible semantics.
"""
requested = tuple(dict.fromkeys(fund_codes))
if any(code not in SOUTHERN_FUND_CODES for code in requested):
raise ValueError("fund code is not in the Southern Fund whitelist")
if not requested:
return {}
symbols = ",".join(
("sh" if code.startswith(("5", "6", "9")) else "sz") + code
for code in requested
)
try:
response = httpx.get(
TENCENT_QUOTE_API + symbols,
headers={"User-Agent": HEADERS["User-Agent"]},
timeout=REQUEST_TIMEOUT,
)
response.raise_for_status()
except httpx.HTTPError as exc:
raise ExchangeQuoteSourceError("tencent quote request failed") from exc
return _parse_tencent_exchange_quotes(response.content, requested)
def _parse_tencent_exchange_quotes(
content: bytes, requested: tuple[str, ...],
) -> dict[str, dict[str, str | None]]:
try:
payload = content.decode("gbk")
except UnicodeDecodeError as exc:
raise ExchangeQuoteSourceError("tencent quote response decoding failed") from exc
result: dict[str, dict[str, str | None]] = {}
for match in re.finditer(r'v_(?:sh|sz)(\d{6})="([^"]*)"', payload):
code, raw = match.groups()
if code not in requested:
continue
fields = raw.split("~")
if len(fields) < 7 or not _valid_quote_number(fields[3]):
continue
previous_close = fields[4] if _valid_quote_number(fields[4]) else None
change_pct = fields[33] if len(fields) > 33 and _valid_quote_number(fields[33]) else None
if change_pct is None and previous_close is not None:
change_pct = _quote_change_pct(fields[3], previous_close)
result[code] = {
"last_price": fields[3],
"previous_close": previous_close,
"change_pct": change_pct,
"volume": fields[6] if _valid_quote_number(fields[6]) else None,
"turnover_amount": None,
}
if not result:
raise ExchangeQuoteSourceError("tencent quote response had no usable quotes")
return result
def _valid_quote_number(value: str) -> bool:
try:
return float(value) > 0
except ValueError:
return False
def _quote_change_pct(last_price: str, previous_close: str) -> str | None:
try:
return str(round((float(last_price) / float(previous_close) - 1) * 100, 4))
except (ValueError, ZeroDivisionError):
return None
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 (total is not None and total <= len(records)):
return records
if not (start_date and end_date):
return records
page_size = max(len(records), 1)
page_count = min((total + page_size - 1) // page_size, 40) if total else 40
seen_dates = {str(item.get("FSRQ") or "") for item in records}
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
new_records = [
item for item in page_records if str(item.get("FSRQ") or "") not in seen_dates
]
if not new_records:
break
seen_dates.update(str(item.get("FSRQ") or "") for item in new_records)
records.extend(new_records)
if len(page_records) < page_size:
break
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}%"