"""南方基金指定产品行情模块。"""
# Public-source adapter retains readable request expressions; line-length checks are not useful here.
# ruff: noqa: E501
from __future__ import annotations

import json
import logging
import re
import threading
import time
from datetime import date, datetime
from datetime import time as clock_time
from decimal import Decimal, InvalidOperation
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="
TENCENT_HISTORY_API = "https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get"
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
EXCHANGE_HISTORY_LOCK = threading.Lock()
EXCHANGE_HISTORY_MIN_INTERVAL_SECONDS = 1.0
MAX_FUNDS_PER_CALL = 1000
FUND_TYPE_GROUPS = {
    "货币型": ("202308", "020480", "511810"),
    # ⚠️ 此处曾含 `160129`（南方金利定开债券C），已移除：C 类份额只在场外销售、
    # **不在交易所挂牌**，行情源对它永远返回空（下单因此走净值降级）。
    # 它的 A 类 `160128` 是上市交易的，已在本组内。
    "债券型": ("007161", "003776", "020281", "511070", "159700", "160128"),
    "混合型": ("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", "515450",
    ),
    "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] = {}
_last_exchange_history_request_at = 0.0


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: list[dict[str, Any]] = []
    try:
        records = _fetch_nav_records(
            fund_code, start_date=start.isoformat(), end_date=end.isoformat(), all_pages=True
        )
    except (httpx.HTTPError, ValueError, TypeError) as exc:
        logger.warning("东方财富历史净值接口失败 code=%s error=%s", fund_code, type(exc).__name__)

    turnover_by_date: dict[str, str] = {}
    turnover_source_by_date: dict[str, str] = {}
    nav_source_by_date: dict[str, str] = {}
    if not records:
        # Listed funds can use Tencent's exchange close as the historical price when
        # the fund NAV provider is unavailable. Amount remains the provider field.
        fallback_rows = get_southern_fund_exchange_history_tencent(
            fund_code, start_date, end_date
        )
        records = [
            {
                "FSRQ": row["trade_date"],
                "DWJZ": row["close_price"],
                "turnover_amount": row["turnover_amount"],
                "source": "tencent_hq_history",
            }
            for row in fallback_rows
        ]
        for row in fallback_rows:
            trade_date = row["trade_date"]
            turnover_by_date[trade_date] = row["turnover_amount"]
            turnover_source_by_date[trade_date] = "tencent_hq_history"
            nav_source_by_date[trade_date] = "tencent_hq_history"
    try:
        if records and not nav_source_by_date:
            for row in get_southern_fund_exchange_history(fund_code, start_date, end_date):
                if row.get("turnover_amount"):
                    trade_date = row["trade_date"]
                    turnover_by_date[trade_date] = row["turnover_amount"]
                    turnover_source_by_date[trade_date] = "eastmoney_hq_nav"
    except (httpx.HTTPError, ValueError, TypeError) as exc:
        logger.warning("历史成交额接口失败 code=%s error=%s", fund_code, type(exc).__name__)
    needed_dates = {
        str(record.get("FSRQ") or "")
        for record in records
        if record.get("FSRQ")
    }
    if needed_dates - turnover_by_date.keys():
        try:
            for row in get_southern_fund_exchange_history_tencent(
                fund_code, start_date, end_date
            ):
                trade_date = row["trade_date"]
                if trade_date not in turnover_by_date and row.get("turnover_amount"):
                    turnover_by_date[trade_date] = row["turnover_amount"]
                    turnover_source_by_date[trade_date] = "eastmoney_hq_nav+tencent_hq_history"
        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,
            "source": nav_source_by_date.get(value_date, "eastmoney_hq_nav"),
        }
        if value_date in turnover_by_date:
            row["turnover_amount"] = turnover_by_date[value_date]
            row["source"] = turnover_source_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 + 1):
        try:
            # The public endpoint intermittently resets concurrent connections.  Serialize
            # historical requests while keeping product-level async orchestration intact.
            with EXCHANGE_HISTORY_LOCK:
                _pace_exchange_history_request()
                response = httpx.get(
                    EXCHANGE_HISTORY_API,
                    params=params,
                    headers={
                        **HEADERS,
                        "Accept": "application/json, text/plain, */*",
                        "Accept-Language": "zh-CN,zh;q=0.9",
                        "Connection": "close",
                    },
                    timeout=REQUEST_TIMEOUT,
                )
            response.raise_for_status()
            payload = response.json()
            if not isinstance(payload, dict):
                raise ValueError("历史行情响应不是 JSON 对象")
            break
        except (httpx.HTTPError, ValueError, TypeError):
            if attempt == EXCHANGE_HISTORY_RETRIES:
                raise
            time.sleep(0.75 * (2**attempt))
    data = payload.get("data") or {}
    if not isinstance(data, dict):
        raise ValueError("历史行情响应 data 字段格式错误")
    records = data.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_exchange_history_tencent(
    fund_code: str, start_date: str, end_date: str
) -> list[dict[str, str]]:
    """Return Tencent daily close and turnover as the history fallback.

    Tencent's public response stores turnover in ``day[][8]`` in ten-thousand yuan.
    This function converts that explicitly documented unit to yuan; it never derives
    turnover from volume or price.
    """
    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("开始日期不能晚于结束日期")
    symbol = ("sh" if fund_code.startswith(("5", "6", "9")) else "sz") + fund_code
    result: dict[str, dict[str, str]] = {}
    for year in range(start.year, end.year + 1):
        params = {
            "_var": f"kline_day{year}",
            "param": f"{symbol},day,{year}-01-01,{year + 1}-12-31,640,",
            "r": "0.8205512681390605",
        }
        for attempt in range(EXCHANGE_HISTORY_RETRIES + 1):
            try:
                with EXCHANGE_HISTORY_LOCK:
                    _pace_exchange_history_request()
                    response = httpx.get(
                        TENCENT_HISTORY_API,
                        params=params,
                        headers={
                            "User-Agent": HEADERS["User-Agent"],
                            "Referer": "https://gu.qq.com/",
                        },
                        timeout=REQUEST_TIMEOUT,
                    )
                response.raise_for_status()
                payload_text = response.text
                _, separator, payload_text = payload_text.partition("=")
                if not separator:
                    raise ValueError("腾讯历史行情响应不是 JSONP")
                payload = json.loads(payload_text)
                if not isinstance(payload, dict):
                    raise ValueError("腾讯历史行情响应不是 JSON 对象")
                break
            except (httpx.HTTPError, ValueError, TypeError):
                if attempt == EXCHANGE_HISTORY_RETRIES:
                    raise
                time.sleep(0.75 * (2**attempt))
        data = payload.get("data") or {}
        if not isinstance(data, dict):
            raise ValueError("腾讯历史行情响应 data 字段格式错误")
        symbol_data = data.get(symbol) or {}
        if not isinstance(symbol_data, dict):
            raise ValueError("腾讯历史行情响应产品字段格式错误")
        rows = symbol_data.get("day") or []
        if not isinstance(rows, list):
            raise ValueError("腾讯历史行情响应 day 字段格式错误")
        for raw in rows:
            if not isinstance(raw, list) or len(raw) < 9:
                continue
            trade_date, close_price, raw_amount = str(raw[0]), str(raw[2]), raw[8]
            try:
                parsed_date = date.fromisoformat(trade_date)
                close = float(close_price)
                amount = Decimal(str(raw_amount)) * Decimal("10000")
                if parsed_date < start or parsed_date > end or close <= 0 or amount < 0:
                    continue
            except (InvalidOperation, TypeError, ValueError):
                continue
            result[trade_date] = {
                "fund_code": fund_code,
                "trade_date": trade_date,
                "close_price": close_price,
                "turnover_amount": format(amount, "f"),
            }
    return [result[key] for key in sorted(result)]


def _pace_exchange_history_request() -> None:
    """Avoid triggering provider-side connection resets during bulk refreshes."""
    global _last_exchange_history_request_at
    now = time.monotonic()
    wait_seconds = (
        EXCHANGE_HISTORY_MIN_INTERVAL_SECONDS
        - (now - _last_exchange_history_request_at)
    )
    if wait_seconds > 0:
        time.sleep(wait_seconds)
    _last_exchange_history_request_at = time.monotonic()


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}%"