"""南方基金指定产品行情模块。""" # Public-source adapter retains readable request expressions; line-length checks are not useful here. # ruff: noqa: E501 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" EXCHANGE_HISTORY_API = "https://push2his.eastmoney.com/api/qt/stock/kline/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 EXCHANGE_HISTORY_RETRIES = 2 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 ) turnover_by_date: dict[str, str] = {} try: turnover_by_date = { row["trade_date"]: row["turnover_amount"] for row in get_southern_fund_exchange_history(fund_code, start_date, end_date) if row.get("turnover_amount") } except (httpx.HTTPError, ValueError, TypeError) as exc: logger.warning("历史成交额接口失败 code=%s error=%s", fund_code, type(exc).__name__) 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 row = {"fund_code": fund_code, "trade_date": value_date, "nav": nav} if value_date in turnover_by_date: row["turnover_amount"] = turnover_by_date[value_date] observations.append(row) return sorted(observations, key=lambda item: item["trade_date"]) def get_southern_fund_exchange_history( fund_code: str, start_date: str, end_date: str ) -> list[dict[str, str]]: """Return exchange daily close and turnover for a listed fund. The NAV endpoint has no trading amount. Eastmoney's exchange K-line endpoint exposes amount as field f56, which is the only source accepted for historical liquidity calculations here. """ 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("开始日期不能晚于结束日期") secid = ("1." if fund_code.startswith(("5", "6", "9")) else "0.") + fund_code params: dict[str, str | int] = { "secid": secid, "klt": 101, "fqt": 1, "beg": start.strftime("%Y%m%d"), "end": end.strftime("%Y%m%d"), "fields1": "f1,f2,f3,f4,f5,f6", "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61", } for attempt in range(EXCHANGE_HISTORY_RETRIES): try: response = httpx.get( EXCHANGE_HISTORY_API, params=params, headers=HEADERS, timeout=REQUEST_TIMEOUT, ) response.raise_for_status() break except httpx.HTTPError: if attempt == EXCHANGE_HISTORY_RETRIES - 1: raise time.sleep(0.2 * (attempt + 1)) payload = response.json() records = ((payload.get("data") or {}).get("klines") or []) result: list[dict[str, str]] = [] for raw in records: fields = str(raw).split(",") if len(fields) < 7: continue trade_date, close_price, turnover_amount = fields[0], fields[2], fields[6] try: parsed_date = date.fromisoformat(trade_date) if parsed_date < start or parsed_date > end: continue if float(close_price) <= 0 or float(turnover_amount) < 0: continue except ValueError: continue result.append({ "fund_code": fund_code, "trade_date": trade_date, "close_price": close_price, "turnover_amount": turnover_amount, }) return result 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}%"