Files

581 lines
23 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""南方基金指定产品行情模块。"""
# 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}</a>'
position = content.find(marker)
if position < 0:
return None
start = content.rfind("<tr", 0, position)
end = content.find("</tr>", position)
if start < 0 or end < 0:
return None
row = content[start:end + len("</tr>")]
name_match = re.search(r'class="name" title="([^"]+)"', row)
cells = [
_html_cell_text(match.group(1))
for match in re.finditer(r"<td[^>]*>(.*?)</td>", 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}%"