app/services/market_data.py

"""Unified stock-data service built on top of multiple AKShare sources."""

from __future__ import annotations

import math
import os
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta, timezone
from io import StringIO
from typing import Any, Callable
from urllib.parse import urlencode

import akshare as ak
import numpy as np
import pandas as pd
import requests

from app.core.cache import CacheEntry, cache
from app.core.config import settings
from app.services.source_runner import AllSourcesFailed, SourceCallable, run_sources
from app.utils.serialization import dataframe_to_records, to_jsonable
from app.utils.stock_code import NormalizedStockCode, normalize_stock_code


UTC = timezone.utc
CN_TZ = timezone(timedelta(hours=8))


TTL = {
    "stock_quote": 300,
    "etf_quote": 300,
    "stock_order_book": 60,
    "stock_daily": 200 * 24 * 3600,
    "stock_technical": 6 * 3600,
    "stock_chip": 24 * 3600,
    "chip_cyq": 24 * 3600,
    "stock_fund_flow": 3600,
    "stock_f10": 2 * 24 * 3600,
    "stock_research_report": 6 * 3600,
    "fund_flow_rank": 1800,
    "limit_pool": 600,
    "board_flow": 1800,
    "market_breadth": 1800,
    "market_temperature": 900,
    "market_indices": 300,
    "market_moves": 300,
    "longhubang": 1800,
    "index_fund_flow": 900,
    "etf_spot": 1800,
    "etf_premium": 1800,
    "etf_daily": 24 * 3600,
    "fund_spot": 6 * 3600,
    "fund_nav": 24 * 3600,
    "news": 3600,
    "notice": 6 * 3600,
    "financial": 2 * 24 * 3600,
    "macro": 2 * 24 * 3600,
    "margin_stock": 4 * 3600,
    "margin_market": 4 * 3600,
    "northbound_hist": 3600,
    "northbound_realtime": 120,
    "northbound_holdings": 1800,
    "shareholders": 12 * 3600,
    "shareholder_top": 12 * 3600,
    "income": 2 * 24 * 3600,
    "balancesheet": 2 * 24 * 3600,
    "cashflow": 2 * 24 * 3600,
    "dividends": 2 * 24 * 3600,
    "equity_history": 2 * 24 * 3600,
    "freeholders": 2 * 24 * 3600,
    "daily_basic": 6 * 3600,
    "trade_calendar": 24 * 3600,
    "bond_yield": 24 * 3600,
    "hk_short_selling": 6 * 3600,
    "big_deal": 300,
    "shenwan_industry": 900,
    "fund_holdings": 4 * 3600,
    "fund_structure": 12 * 3600,
    "us_indices": 120,
    "us_sectors": 900,
    "us_market_summary": 300,
    "us_stock_quote": 120,
    "x_timeline": 300,
}


HTTP_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Accept": "application/json,text/plain,*/*",
}


EASTMONEY_F10_ENDPOINTS: dict[str, dict[str, Any]] = {
    "balance": {
        "api_family": "data_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("REPORT_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"type": "RPT_F10_FINANCE_GBALANCE", "sty": "F10_FINANCE_GBALANCE"},
    },
    "cashflow_quarterly": {
        "api_family": "data_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("REPORT_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"type": "RPT_F10_FINANCE_GCASHFLOWQC", "sty": "PC_F10_GCASHFLOWQC"},
    },
    "cashflow_ytd": {
        "api_family": "data_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("REPORT_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"type": "RPT_F10_FINANCE_GCASHFLOW", "sty": "APP_F10_GCASHFLOW"},
    },
    "dividend_allotment": {
        "api_family": "data_v1_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("NOTICE_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"reportName": "RPT_F10_DIVIDEND_ALLOTMENT", "columns": "ALL"},
    },
    "dividend_main": {
        "api_family": "data_v1_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("NOTICE_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"reportName": "RPT_F10_DIVIDEND_MAIN", "columns": "ALL"},
    },
    "equity_history": {
        "api_family": "data_v1_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("NOTICE_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"reportName": "RPT_F10_EH_EQUITY", "columns": "ALL"},
    },
    "freeholders": {
        "api_family": "data_v1_get",
        "date_field": "END_DATE",
        "sort_fields": ("END_DATE", "HOLDER_RANK"),
        "sort_types": ("-1", "1"),
        "fixed_params": {
            "reportName": "RPT_F10_EH_FREEHOLDERS",
            "columns": (
                "SECUCODE,SECURITY_CODE,END_DATE,HOLDER_RANK,HOLDER_NEW,HOLDER_NAME,"
                "HOLDER_TYPE,SHARES_TYPE,HOLD_NUM,FREE_HOLDNUM_RATIO,HOLD_NUM_CHANGE,"
                "CHANGE_RATIO"
            ),
        },
    },
    "income_quarterly": {
        "api_family": "data_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("REPORT_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"type": "RPT_F10_FINANCE_GINCOMEQC", "sty": "PC_F10_GINCOMEQC"},
    },
    "income_ytd": {
        "api_family": "data_get",
        "date_field": "NOTICE_DATE",
        "sort_fields": ("REPORT_DATE", "SECURITY_CODE"),
        "sort_types": ("-1", "-1"),
        "fixed_params": {"type": "RPT_F10_FINANCE_GINCOME", "sty": "APP_F10_GINCOME"},
    },
}

EASTMONEY_DATA_GET_URL = "https://datacenter.eastmoney.com/securities/api/data/get"
EASTMONEY_DATA_V1_GET_URL = "https://datacenter.eastmoney.com/securities/api/data/v1/get"
THS_LIMIT_UP_POOL_URL = "https://data.10jqka.com.cn/dataapi/limit_up/limit_up_pool"
THS_LIMIT_UP_POOL_FIELD = (
    "199112,10,9001,330323,330324,330325,9002,330329,133971,133970,"
    "1968584,3475914,9003,9004"
)
CHINABOND_HISTORY_QUERY_URL = "https://yield.chinabond.com.cn/cbweb-czb-web/czb/historyQuery"
EASTMONEY_HK_SELLSHORT_URL = "https://hk.eastmoney.com/sellshort.html"
CHINABOND_YIELD_FIELDS: tuple[tuple[str, str], ...] = (
    ("threeMonth", "three_month_yield_pct"),
    ("sixMonth", "six_month_yield_pct"),
    ("oneYear", "one_year_yield_pct"),
    ("twoYear", "two_year_yield_pct"),
    ("threeYear", "three_year_yield_pct"),
    ("fiveYear", "five_year_yield_pct"),
    ("sevenYear", "seven_year_yield_pct"),
    ("tenYear", "ten_year_yield_pct"),
    ("fifteenYear", "fifteen_year_yield_pct"),
    ("twentyYear", "twenty_year_yield_pct"),
    ("thirtyYear", "thirty_year_yield_pct"),
)


class MarketDataService:
    # Proxy handling for AKShare vs. direct HTTP sources is managed by
    # source_runner._invoke_source so that requests-based calls can still
    # use the system proxy when available.

    def catalog(self) -> dict[str, Any]:
        return {
            "groups": {
                "stocks": [
                    "/stocks/{stock_code}/quote",
                    "/stocks/{stock_code}/order-book",
                    "/stocks/{stock_code}/daily",
                    "/stocks/{stock_code}/technical",
                    "/stocks/{stock_code}/chip",
                    "/stocks/{stock_code}/chip-simple",
                    "/stocks/{stock_code}/fund-flow",
                    "/stocks/{stock_code}/margin",
                    "/stocks/{stock_code}/shareholders",
                    "/stocks/{stock_code}/shareholder-top",
                    "/stocks/{stock_code}/news",
                    "/stocks/{stock_code}/notices",
                    "/stocks/{stock_code}/financial",
                    "/stocks/{stock_code}/income",
                    "/stocks/{stock_code}/balancesheet",
                    "/stocks/{stock_code}/cashflow",
                    "/stocks/{stock_code}/dividends",
                    "/stocks/{stock_code}/equity-history",
                    "/stocks/{stock_code}/freeholders",
                    "/stocks/{stock_code}/daily-basic",
                    "/stocks/{stock_code}/f10/company",
                    "/stocks/{stock_code}/research-reports",
                ],
                "market": [
                    "/market/indices",
                    "/market/moves",
                    "/market/limit-up",
                    "/market/limit-down",
                    "/market/breadth",
                    "/market/temperature",
                    "/market/longhubang",
                    "/market/margin",
                    "/market/northbound",
                    "/market/northbound/realtime",
                    "/market/northbound/holdings",
                    "/market/shenwan-industry",
                    "/market/fund-holdings",
                    "/market/fund-structure",
                    "/market/trade-calendar",
                ],
                "boards": [
                    "/boards/flow",
                    "/boards/concepts/flow",
                    "/boards/industries/flow",
                ],
                "indices": [
                    "/indices/{index_code}/fund-flow",
                ],
                "etfs": [
                    "/etfs/spot",
                    "/etfs/premium",
                    "/etfs/{fund_code}/quote",
                    "/etfs/{fund_code}/premium",
                    "/etfs/{fund_code}/daily",
                ],
                "hk": [
                    "/hk/stocks/{stock_code}/short-selling",
                ],
                "funds": [
                    "/funds/open/spot",
                    "/funds/money/spot",
                    "/funds/{fund_code}/nav",
                ],
                "capital": [
                    "/fund-flow/rank",
                    "/capital/big-deal",
                ],
                "macro": [
                    "/macro/china/{indicator}",
                    "/macro/chinabond/yield-curve",
                ],
                "us": [
                    "/us/indices",
                    "/us/sectors",
                    "/us/market-summary",
                    "/us/stocks/{symbol}/quote",
                    "/us/stocks/{symbol}/daily",
                ],
                "social": [
                    "/social/x/timeline",
                ],
                "news": [
                    "/news/global",
                ],
                "search": [
                    "/search",
                ],
            },
            "ttl_seconds": TTL,
            "cache": cache.stats(),
        }

    def stock_quote(self, stock_code: str) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        return self._cached(
            "stock_quote",
            {"stock_code": code.display},
            TTL["stock_quote"],
            [
                ("sina.hq.realtime", lambda: self._sina_stock_quote_payload(code, include_order_book=False)),
                ("eastmoney.push2.stock.quote", lambda: self._stock_quote_eastmoney_push2(code)),
                ("tencent.qt.quote", lambda: self._stock_quote_tencent(code)),
                ("akshare.stock_zh_a_spot_em", lambda: self._stock_quote_from_spot_em(code)),
                ("akshare.stock_zh_a_hist.latest", lambda: self._stock_quote_from_daily(code)),
                ("akshare.stock_zh_a_daily.sina.latest", lambda: self._stock_quote_from_sina_daily(code)),
                ("yahoo.chart.quote", lambda: self._stock_quote_yahoo(code)),
            ],
        )

    def stock_order_book(self, stock_code: str) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        return self._cached(
            "stock_order_book",
            {"stock_code": code.display},
            TTL["stock_order_book"],
            [
                ("sina.hq.order_book", lambda: self._sina_stock_quote_payload(code, include_order_book=True)),
                ("akshare.stock_zh_a_spot_em.quote_only", lambda: self._stock_quote_from_spot_em(code)),
            ],
        )

    def stock_daily(
        self,
        stock_code: str,
        days: int = 60,
        start_date: str | None = None,
        end_date: str | None = None,
        adjust: str = "qfq",
    ) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        days = max(1, min(int(days), 5000))
        start, end = self._date_range(days, start_date, end_date)
        return self._cached(
            "stock_daily",
            {"stock_code": code.display, "days": days, "start_date": start, "end_date": end, "adjust": adjust},
            self._daily_ttl(end),
            [
                ("yahoo.chart.daily", lambda: self._stock_daily_yahoo(code, days, adjust, start, end)),
                ("eastmoney.push2his.stock_kline", lambda: self._stock_daily_eastmoney_direct_payload(code, start, end, adjust, days)),
                ("tencent.ifzq.fqkline", lambda: self._stock_daily_tencent(code, days, adjust, start, end)),
                (
                    "akshare.stock_zh_a_hist.eastmoney",
                    lambda: self._daily_payload_from_df(
                        code,
                        self._stock_daily_hist_df(code, start, end, adjust),
                        days,
                        source_style="eastmoney",
                    ),
                ),
                (
                    "akshare.stock_zh_a_daily.sina",
                    lambda: self._daily_payload_from_df(
                        code,
                        self._stock_daily_sina_df(code, start, end, adjust),
                        days,
                        source_style="sina",
                    ),
                ),
                ("baostock.query_history_k_data_plus", lambda: self._stock_daily_baostock_payload(code, start, end, adjust, days)),
            ],
        )

    def stock_technical(self, stock_code: str, days: int = 120, history_days: int = 0, adjust: str = "qfq") -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        days = max(30, min(int(days), 5000))
        history_days = max(0, min(int(history_days), 120))
        start, end = self._date_range(days, None, None)
        return self._cached(
            "stock_technical",
            {"stock_code": code.display, "days": days, "history_days": history_days, "adjust": adjust},
            TTL["stock_technical"],
            [
                (
                    "akshare.stock_zh_a_daily.sina",
                    lambda: self._technical_payload(code, self._stock_daily_sina_df(code, start, end, adjust), days, history_days),
                ),
                (
                    "akshare.stock_zh_a_hist.eastmoney",
                    lambda: self._technical_payload(code, self._stock_daily_hist_df(code, start, end, adjust), days, history_days),
                ),
            ],
        )

    def stock_chip_simple(self, stock_code: str, lookback_days: int = 60, adjust: str = "qfq") -> dict[str, Any]:
        """简化版筹码估算"""
        code = normalize_stock_code(stock_code)
        lookback_days = max(10, min(int(lookback_days), 500))
        start, end = self._date_range(lookback_days, None, None)
        return self._cached(
            "stock_chip",
            {"stock_code": code.display, "lookback_days": lookback_days, "adjust": adjust},
            TTL["stock_chip"],
            [
                (
                    "akshare.stock_zh_a_daily.sina",
                    lambda: self._chip_payload(code, self._stock_daily_sina_df(code, start, end, adjust), lookback_days),
                ),
                (
                    "akshare.stock_zh_a_hist.eastmoney",
                    lambda: self._chip_payload(code, self._stock_daily_hist_df(code, start, end, adjust), lookback_days),
                ),
            ],
        )

    def stock_chip(self, stock_code: str, adjust: str = "qfq") -> dict[str, Any]:
        """筹码分布（精算版 CYQ）"""
        code = normalize_stock_code(stock_code)
        adjust = adjust if adjust in {"qfq", "hfq", ""} else "qfq"
        return self._cached(
            "chip_cyq",
            {"stock_code": code.display, "adjust": adjust},
            TTL["chip_cyq"],
            [
                ("eastmoney.cyq_em.push2delay", lambda: self._chip_cyq_em(code, adjust)),
            ],
            timeout_seconds=30,
        )

    def stock_fund_flow(self, stock_code: str, days: int = 10) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        days = max(1, min(int(days), 120))
        return self._cached(
            "stock_fund_flow",
            {"stock_code": code.display, "days": days},
            TTL["stock_fund_flow"],
            [
                ("scrapling.ths.stock_fflow", lambda: self._stock_fund_flow_ths_payload(code, days)),
                ("scrapling.eastmoney.stock_fflow", lambda: self._stock_fund_flow_scrapling_payload(code, days)),
                ("akshare.stock_individual_fund_flow.eastmoney", lambda: self._stock_fund_flow_payload(code, days)),
                ("eastmoney.push2his.stock_fflow", lambda: self._stock_fund_flow_direct_payload(code, days, "https://push2his.eastmoney.com")),
                ("eastmoney.push2.stock_fflow", lambda: self._stock_fund_flow_direct_payload(code, days, "https://push2.eastmoney.com")),
                ("eastmoney.push2delay.stock_fflow", lambda: self._stock_fund_flow_direct_payload(code, days, "https://push2delay.eastmoney.com")),
            ],
            retry_attempts=3,
            min_rows=1,
        )

    def stock_f10_company(self, stock_code: str) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        return self._cached(
            "stock_f10_company",
            {"stock_code": code.display},
            TTL["stock_f10"],
            [
                ("eastmoney.hsf10.company_survey", lambda: self._f10_company_payload(code)),
            ],
        )

    def stock_research_reports(self, stock_code: str, limit: int = 20) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 100))
        return self._cached(
            "stock_research_reports",
            {"stock_code": code.display, "limit": limit},
            TTL["stock_research_report"],
            [
                ("eastmoney.reportapi.stock_reports", lambda: self._research_reports_payload(code, limit)),
                ("akshare.stock_research_report_em.fallback", lambda: self._research_reports_akshare_payload(code, limit)),
                ("scrapling.eastmoney.research_reports", lambda: self._research_reports_scrapling_payload(code, limit)),
            ],
            retry_attempts=2,
            timeout_seconds=30,
        )

    def fund_flow_rank(self, indicator: str = "5日", limit: int = 100) -> dict[str, Any]:
        indicator = indicator if indicator in {"今日", "3日", "5日", "10日"} else "5日"
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "fund_flow_rank",
            {"indicator": indicator, "limit": limit},
            TTL["fund_flow_rank"],
            [
                (
                    "scrapling.eastmoney.fund_flow_rank",
                    lambda: self._fund_flow_rank_scrapling_payload(indicator, limit),
                ),
                (
                    "eastmoney.push2delay.stock_fund_flow_rank",
                    lambda: self._fund_flow_rank_direct_payload(indicator, limit, "https://push2delay.eastmoney.com"),
                ),
                (
                    "eastmoney.push2.stock_fund_flow_rank",
                    lambda: self._fund_flow_rank_direct_payload(indicator, limit, "https://push2.eastmoney.com"),
                ),
                (
                    "akshare.stock_individual_fund_flow_rank.eastmoney",
                    lambda: self._records_payload(ak.stock_individual_fund_flow_rank(indicator=indicator), limit, "records"),
                ),
                (
                    "ths.stock_fund_flow_individual.fast_page",
                    lambda: self._fund_flow_rank_ths_fast_payload(indicator, limit),
                ),
                (
                    "akshare.stock_main_fund_flow.eastmoney",
                    lambda: self._fund_flow_rank_main_payload(indicator, limit),
                ),
                (
                    "akshare.ths.stock_fund_flow_individual",
                    lambda: self._fund_flow_rank_ths_payload(indicator, limit),
                ),
            ],
            retry_attempts=1,
        )

    def big_deal(self, limit: int = 100) -> dict[str, Any]:
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "big_deal",
            {"limit": limit},
            TTL["big_deal"],
            [
                ("eastmoney.push2delay.big_deal", lambda: self._big_deal_direct(limit)),
                ("scrapling.ths.big_deal", lambda: self._big_deal_scrapling(limit)),
                ("akshare.stock_fund_flow_big_deal", lambda: self._big_deal_akshare(limit)),
                ("akshare.stock_individual_fund_flow_rank", lambda: self._big_deal_rank(limit)),
            ],
            timeout_seconds=180,
        )

    def limit_pool(self, side: str, date: str | None = None, limit: int = 100) -> dict[str, Any]:
        trade_date = self._compact_date(date) if date else self._now_cn().strftime("%Y%m%d")
        side = "up" if side == "up" else "down"
        limit = max(1, min(int(limit), 1000))
        source_name = "akshare.stock_zt_pool_em" if side == "up" else "akshare.stock_zt_pool_dtgc_em"
        func = ak.stock_zt_pool_em if side == "up" else ak.stock_zt_pool_dtgc_em
        return self._cached(
            f"limit_pool_{side}",
            {"side": side, "date": trade_date, "limit": limit},
            TTL["limit_pool"],
            [
                ("eastmoney.push2ex.limit_pool", lambda: self._eastmoney_limit_pool_payload(trade_date, side, limit)),
                *(
                    [("ths.limit_up_pool.direct", lambda: self._ths_limit_up_pool_payload(trade_date, limit))]
                    if side == "up"
                    else []
                ),
                (source_name, lambda: self._limit_pool_payload(func(date=trade_date), trade_date, side, limit)),
            ],
            timeout_seconds=8,
        )

    def concept_flow(self, symbol: str = "即时", limit: int = 1000) -> dict[str, Any]:
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "concept_flow",
            {"symbol": symbol, "limit": limit},
            TTL["board_flow"],
            [
                ("eastmoney.push2delay.board_flow.concept", lambda: self._board_flow_direct_key_payload("concept", "concepts", limit, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2.board_flow.concept", lambda: self._board_flow_direct_key_payload("concept", "concepts", limit, "https://push2.eastmoney.com")),
                ("akshare.stock_fund_flow_concept.eastmoney", lambda: self._board_flow_payload(ak.stock_fund_flow_concept(symbol=symbol), "concepts", limit)),
                ("akshare.stock_board_concept_name_em.eastmoney", lambda: self._board_flow_payload(ak.stock_board_concept_name_em(), "concepts", limit)),
            ],
            retry_attempts=2,
            timeout_seconds=10,
        )

    def industry_flow(self, symbol: str = "即时", limit: int = 1000) -> dict[str, Any]:
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "industry_flow",
            {"symbol": symbol, "limit": limit},
            TTL["board_flow"],
            [
                ("eastmoney.push2delay.board_flow.industry", lambda: self._board_flow_direct_key_payload("industry", "industries", limit, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2.board_flow.industry", lambda: self._board_flow_direct_key_payload("industry", "industries", limit, "https://push2.eastmoney.com")),
                ("akshare.stock_fund_flow_industry.eastmoney", lambda: self._board_flow_payload(ak.stock_fund_flow_industry(symbol=symbol), "industries", limit)),
                ("akshare.stock_board_industry_name_em.eastmoney", lambda: self._board_flow_payload(ak.stock_board_industry_name_em(), "industries", limit)),
            ],
            timeout_seconds=10,
        )

    def board_flow(self, category: str = "industry", limit: int = 100) -> dict[str, Any]:
        category = self._normalize_board_category(category)
        limit = max(1, min(int(limit), 200))
        fallback_sources: dict[str, tuple[str, SourceCallable]] = {
            "concept": (
                "akshare.stock_fund_flow_concept.eastmoney",
                lambda: self._board_flow_payload(ak.stock_fund_flow_concept(symbol="即时"), "items", limit),
            ),
            "industry": (
                "akshare.stock_fund_flow_industry.eastmoney",
                lambda: self._board_flow_payload(ak.stock_fund_flow_industry(symbol="即时"), "items", limit),
            ),
            "region": (
                "akshare.stock_board_industry_name_em.eastmoney",
                lambda: self._board_flow_payload(ak.stock_board_industry_name_em(), "items", limit),
            ),
        }
        direct_sources: list[tuple[str, SourceCallable]] = [
            (f"eastmoney.push2delay.board_flow.{category}", lambda: self._board_flow_direct_payload(category, limit, "https://push2delay.eastmoney.com")),
            (f"eastmoney.push2.board_flow.{category}", lambda: self._board_flow_direct_payload(category, limit, "https://push2.eastmoney.com")),
        ]
        sources = [*direct_sources, fallback_sources[category]]
        return self._cached(
            "board_flow_direct",
            {"category": category, "limit": limit},
            TTL["board_flow"],
            sources,
        )

    def market_indices(self, limit: int = 100) -> dict[str, Any]:
        limit = max(1, min(int(limit), 1000))
        return self._cached(
            "market_indices",
            {"limit": limit},
            TTL["market_indices"],
            [
                ("sina.indices.realtime", lambda: self._market_indices_sina_payload(limit)),
                ("akshare.stock_zh_index_spot_sina", lambda: self._records_payload(ak.stock_zh_index_spot_sina(), limit, "indices")),
                ("akshare.stock_zh_index_spot_em", lambda: self._records_payload(ak.stock_zh_index_spot_em(), limit, "indices")),
                ("yahoo.chart.indices", lambda: self._market_indices_yahoo(limit)),
            ],
            retry_attempts=2,
        )

    def market_moves(self, move_type: str = "surge", limit: int = 50) -> dict[str, Any]:
        move_type = self._normalize_stock_move_type(move_type)
        limit = max(1, min(int(limit), 200))
        return self._cached(
            "market_moves",
            {"move_type": move_type, "limit": limit},
            TTL["market_moves"],
            [
                ("eastmoney.push2delay.stock_moves", lambda: self._market_moves_payload(move_type, limit, "https://push2delay.eastmoney.com")),
                ("eastmoney.82.push2delay.stock_moves", lambda: self._market_moves_payload(move_type, limit, "https://82.push2delay.eastmoney.com")),
                ("eastmoney.84.push2delay.stock_moves", lambda: self._market_moves_payload(move_type, limit, "https://84.push2delay.eastmoney.com")),
                ("eastmoney.push2.stock_moves", lambda: self._market_moves_payload(move_type, limit, "https://push2.eastmoney.com")),
                ("akshare.ths.stock_rank", lambda: self._market_moves_ths_rank_payload(move_type, limit)),
                ("akshare.stock_hot_rank_em", lambda: self._market_moves_hot_rank_payload(move_type, limit)),
            ],
            timeout_seconds=8,
        )

    def longhubang(self, date: str | None = None, limit: int = 50, page: int = 1) -> dict[str, Any]:
        trade_date = self._iso_date(date) if date else ""
        limit = max(1, min(int(limit), 200))
        page = max(1, int(page))
        return self._cached(
            "longhubang",
            {"date": trade_date, "limit": limit, "page": page},
            TTL["longhubang"],
            [
                ("eastmoney.datacenter.longhubang", lambda: self._longhubang_payload(trade_date, limit, page)),
            ],
        )

    def index_fund_flow(self, index_code: str, interval: str = "1m", limit: int = 120) -> dict[str, Any]:
        interval = self._normalize_fund_flow_interval(interval)
        limit = max(1, min(int(limit), 2000))
        secid = self._index_secid(index_code)
        return self._cached(
            "index_fund_flow",
            {"index_code": index_code, "secid": secid, "interval": interval, "limit": limit},
            TTL["index_fund_flow"],
            [
                ("eastmoney.push2delay.index_fflow_kline", lambda: self._index_fund_flow_payload(secid, interval, limit, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2his.index_fflow_kline", lambda: self._index_fund_flow_payload(secid, interval, limit, "https://push2his.eastmoney.com")),
                ("eastmoney.push2.index_fflow_kline", lambda: self._index_fund_flow_payload(secid, interval, limit, "https://push2.eastmoney.com")),
                ("akshare.stock_hsgt_fund_flow_summary_em", lambda: self._index_fund_flow_hsgt_fallback_payload(index_code, interval, limit)),
            ],
        )

    def etf_spot(self, limit: int = 200) -> dict[str, Any]:
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "etf_spot",
            {"limit": limit},
            TTL["etf_spot"],
            [
                ("eastmoney.push2delay.etf_spot", lambda: self._etf_spot_direct_payload(limit, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2.etf_spot", lambda: self._etf_spot_direct_payload(limit, "https://push2.eastmoney.com")),
                ("akshare.fund_etf_spot_em", lambda: self._records_payload(ak.fund_etf_spot_em(), limit, "etfs")),
                ("akshare.fund_etf_spot_ths", lambda: self._records_payload(ak.fund_etf_spot_ths(), limit, "etfs")),
                ("akshare.fund_etf_category_sina", lambda: self._records_payload(ak.fund_etf_category_sina(symbol="ETF基金"), limit, "etfs")),
            ],
        )

    def etf_premium(self, limit: int = 200, sort: str = "abs") -> dict[str, Any]:
        limit = max(1, min(int(limit), 5000))
        sort = sort if sort in {"abs", "premium", "discount", "code"} else "abs"
        return self._cached(
            "etf_premium",
            {"limit": limit, "sort": sort},
            TTL["etf_premium"],
            [
                ("eastmoney.push2delay.etf_premium", lambda: self._etf_premium_direct_payload(limit, sort, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2.etf_premium", lambda: self._etf_premium_direct_payload(limit, sort, "https://push2.eastmoney.com")),
                ("akshare.fund_etf_spot_em.premium", lambda: self._etf_premium_from_akshare(limit, sort)),
            ],
            timeout_seconds=8,
        )

    def etf_premium_detail(self, fund_code: str) -> dict[str, Any]:
        code = self._etf_code(fund_code)
        return self._cached(
            "etf_premium_detail",
            {"fund_code": code.display},
            TTL["etf_quote"],
            [
                ("eastmoney.push2delay.etf_premium.detail", lambda: self._etf_premium_detail_direct_payload(code, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2.etf_premium.detail", lambda: self._etf_premium_detail_direct_payload(code, "https://push2.eastmoney.com")),
                ("akshare.fund_etf_spot_em.premium.detail", lambda: self._etf_premium_detail_from_akshare(code)),
            ],
            timeout_seconds=8,
        )

    def etf_daily(
        self,
        fund_code: str,
        days: int = 120,
        start_date: str | None = None,
        end_date: str | None = None,
        adjust: str = "",
    ) -> dict[str, Any]:
        code = self._fund_code(fund_code)
        days = max(1, min(int(days), 5000))
        start, end = self._date_range(days, start_date, end_date)
        return self._cached(
            "etf_daily",
            {"fund_code": code, "days": days, "start_date": start, "end_date": end, "adjust": adjust},
            self._daily_ttl(end),
            [
                ("akshare.fund_etf_fund_info_em", lambda: self._fund_daily_payload(ak.fund_etf_fund_info_em(fund=code, start_date=start, end_date=end), code, days, "records")),
                ("akshare.fund_etf_hist_em", lambda: self._fund_daily_payload(ak.fund_etf_hist_em(symbol=code, period="daily", start_date=start, end_date=end, adjust=adjust or ""), code, days, "records")),
                ("akshare.fund_etf_hist_sina", lambda: self._fund_daily_payload(ak.fund_etf_hist_sina(symbol=self._sina_fund_symbol(code)), code, days, "records")),
                ("yahoo.chart.etf_daily", lambda: self._stock_daily_yahoo(self._etf_code(code), days, adjust or "")),
            ],
        )

    def fund_open_spot(self, limit: int = 200) -> dict[str, Any]:
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "fund_open_spot",
            {"limit": limit},
            TTL["fund_spot"],
            [
                ("eastmoney.fundcode_search", lambda: self._fund_name_direct_payload(limit)),
                ("akshare.fund_open_fund_rank_em", lambda: self._records_payload(ak.fund_open_fund_rank_em(symbol="全部"), limit, "funds")),
                ("akshare.fund_open_fund_daily_em", lambda: self._records_payload(ak.fund_open_fund_daily_em(), limit, "funds")),
            ],
        )

    def fund_money_spot(self, limit: int = 200) -> dict[str, Any]:
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "fund_money_spot",
            {"limit": limit},
            TTL["fund_spot"],
            [
                ("eastmoney.money_fund_html", lambda: self._fund_money_spot_direct(limit)),
                ("akshare.fund_money_fund_daily_em", lambda: self._records_payload(ak.fund_money_fund_daily_em(), limit, "funds")),
            ],
            timeout_seconds=8,
        )

    def fund_nav(self, fund_code: str, fund_type: str = "open", limit: int = 300) -> dict[str, Any]:
        code = self._fund_code(fund_code)
        fund_type = fund_type if fund_type in {"open", "money", "etf"} else "open"
        limit = max(1, min(int(limit), 5000))
        source_map: dict[str, list[tuple[str, SourceCallable]]] = {
            "open": [
                ("eastmoney.fund.lsjz", lambda: self._fund_nav_direct_payload(code, fund_type, limit)),
                ("akshare.fund_open_fund_info_em.nav", lambda: self._fund_info_payload(ak.fund_open_fund_info_em(symbol=code, indicator="单位净值走势"), code, fund_type, limit)),
                ("akshare.fund_open_fund_info_em.acc_nav", lambda: self._fund_info_payload(ak.fund_open_fund_info_em(symbol=code, indicator="累计净值走势"), code, fund_type, limit)),
            ],
            "money": [
                ("eastmoney.fund.lsjz", lambda: self._fund_nav_direct_payload(code, fund_type, limit)),
                ("akshare.fund_money_fund_info_em", lambda: self._fund_info_payload(ak.fund_money_fund_info_em(symbol=code), code, fund_type, limit)),
            ],
            "etf": [
                ("eastmoney.fund.lsjz", lambda: self._fund_nav_direct_payload(code, fund_type, limit)),
                ("akshare.fund_etf_fund_info_em", lambda: self._fund_info_payload(ak.fund_etf_fund_info_em(fund=code), code, fund_type, limit)),
            ],
        }
        return self._cached(
            "fund_nav",
            {"fund_code": code, "fund_type": fund_type, "limit": limit},
            TTL["fund_nav"],
            source_map[fund_type],
        )

    def hk_short_selling(self, stock_code: str, limit: int = 100, pages: int = 2) -> dict[str, Any]:
        code = self._hk_stock_code(stock_code)
        limit = max(1, min(int(limit), 1000))
        pages = max(1, min(int(pages), 20))
        return self._cached(
            "hk_short_selling",
            {"stock_code": code, "limit": limit, "pages": pages},
            TTL["hk_short_selling"],
            [
                ("eastmoney.hk.sellshort.html", lambda: self._hk_short_selling_direct_payload(code, limit, pages)),
            ],
            timeout_seconds=max(12, pages * 8),
        )

    def market_breadth(self) -> dict[str, Any]:
        return self._cached(
            "market_breadth",
            {},
            TTL["market_breadth"],
            [
                ("akshare.stock_board_industry_summary_ths", self._market_breadth_from_industry_summary),
                ("eastmoney.push2delay.market_breadth", self._market_breadth_direct),
                ("akshare.stock_zh_a_spot_em", self._market_breadth_from_spot_em),
            ],
            timeout_seconds=35,
        )

    def market_temperature(self, date: str | None = None) -> dict[str, Any]:
        trade_date = self._compact_date(date) if date else self._now_cn().strftime("%Y%m%d")
        return self._cached(
            "market_temperature",
            {"date": trade_date},
            TTL["market_temperature"],
            [
                ("combined.limit_pool_and_breadth", lambda: self._market_temperature_payload(trade_date)),
            ],
            timeout_seconds=35,
        )

    def global_news(self, limit: int = 50) -> dict[str, Any]:
        limit = max(1, min(int(limit), 200))
        return self._cached(
            "global_news",
            {"limit": limit},
            TTL["news"],
            [
                ("akshare.stock_info_global_em.eastmoney", lambda: self._news_payload(ak.stock_info_global_em(), limit, "news")),
                ("akshare.stock_info_global_sina.sina", lambda: self._news_payload(ak.stock_info_global_sina(), limit, "news")),
                ("akshare.stock_info_global_futu.futu", lambda: self._news_payload(ak.stock_info_global_futu(), limit, "news")),
                ("akshare.stock_info_global_ths.ths", lambda: self._news_payload(ak.stock_info_global_ths(), limit, "news")),
                ("akshare.stock_info_global_cls.cls", lambda: self._news_payload(ak.stock_info_global_cls(symbol="全部"), limit, "news")),
            ],
        )

    def stock_news(self, stock_code: str, limit: int = 30) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        return self._cached(
            "stock_news",
            {"stock_code": code.display, "limit": limit},
            TTL["news"],
            [
                ("akshare.stock_news_em.eastmoney", lambda: self._news_payload(ak.stock_news_em(symbol=code.code), limit, "news", code.display)),
            ],
        )

    def stock_notices(self, stock_code: str, limit: int = 30) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        return self._cached(
            "stock_notices",
            {"stock_code": code.display, "limit": limit},
            TTL["notice"],
            [
                ("eastmoney.notice.stock_announcements", lambda: self._stock_announcements_payload(code, limit)),
                ("scrapling.eastmoney.notices", lambda: self._notices_scrapling_payload(code, limit)),
            ],
        )

    def stock_financial(self, stock_code: str, kind: str = "abstract", limit: int = 20, report_date: str | None = None, start_year: str | None = None, end_year: str | None = None) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        kind = kind if kind in {"abstract", "indicators", "forecast", "express"} else "abstract"
        limit = max(1, min(int(limit), 200))
        explicit_report_date = report_date is not None
        report_dates: list[str] = [self._report_date(report_date)] if explicit_report_date else []
        effective_start_year = str(start_year or "2018")
        effective_end_year = str(end_year or self._now_cn().year)

        def _financial_df_filtered(df: pd.DataFrame, stock_code: str, kind: str, limit: int) -> dict[str, Any]:
            df = self._require_df(df, kind)
            date_col = None
            for col in df.columns:
                col_lower = str(col).lower()
                if "日期" in col_lower or "报告期" in col_lower or "report_date" in col_lower or "report date" in col_lower:
                    date_col = col
                    break
            if date_col is not None:
                df[date_col] = pd.to_datetime(df[date_col], errors="coerce")
                df = df.dropna(subset=[date_col])
                try:
                    start_dt = pd.Timestamp(f"{effective_start_year}-01-01")
                    end_dt = pd.Timestamp(f"{effective_end_year}-12-31")
                    df = df[(df[date_col] >= start_dt) & (df[date_col] <= end_dt)]
                except Exception:
                    pass
                df = df.sort_values(date_col, ascending=False)
            records = dataframe_to_records(df.head(limit), limit)
            return {"stock_code": stock_code, "kind": kind, "count": len(records), "records": records}

        def _recent_report_dates(n: int = 8) -> list[str]:
            today = self._now_cn().date()
            quarters = []
            year = today.year
            q_end_dates = [(3, 31), (6, 30), (9, 30), (12, 31)]
            while len(quarters) < n:
                for m, d in reversed(q_end_dates):
                    candidate = datetime(year, m, d, tzinfo=CN_TZ).date()
                    if candidate > today:
                        continue
                    quarters.append(f"{year}{m:02d}{d:02d}")
                    if len(quarters) >= n:
                        break
                year -= 1
            return quarters

        if not report_dates:
            report_dates = _recent_report_dates()
        primary_report_date = report_dates[0]

        source_map: dict[str, list[tuple[str, SourceCallable]]] = {
            "abstract": [
                (
                    "akshare.stock_financial_analysis_indicator_em.eastmoney",
                    lambda: _financial_df_filtered(
                        ak.stock_financial_analysis_indicator_em(symbol=code.display, indicator="按报告期"),
                        code.display,
                        kind,
                        limit,
                    ),
                ),
                (
                    "akshare.stock_financial_analysis_indicator.eastmoney",
                    lambda: _financial_df_filtered(
                        ak.stock_financial_analysis_indicator(symbol=code.code, start_year=effective_start_year),
                        code.display,
                        "indicators",
                        limit,
                    ),
                ),
                (
                    "akshare.stock_financial_abstract_ths.ths",
                    lambda: _financial_df_filtered(
                        ak.stock_financial_abstract_ths(symbol=code.code, indicator="按报告期"),
                        code.display,
                        kind,
                        limit,
                    ),
                ),
            ],
            "indicators": [
                (
                    "akshare.stock_financial_analysis_indicator_em.eastmoney",
                    lambda: _financial_df_filtered(
                        ak.stock_financial_analysis_indicator_em(symbol=code.display, indicator="按报告期"),
                        code.display,
                        kind,
                        limit,
                    ),
                ),
                (
                    "akshare.stock_financial_analysis_indicator.eastmoney",
                    lambda: _financial_df_filtered(
                        ak.stock_financial_analysis_indicator(symbol=code.code, start_year=effective_start_year),
                        code.display,
                        kind,
                        limit,
                    ),
                ),
            ],
            "forecast": [
                (
                    "akshare.stock_yjyg_em.eastmoney",
                    lambda: self._financial_recent_filtered_payload(
                        lambda date: ak.stock_yjyg_em(date=date),
                        code,
                        kind,
                        limit,
                        report_dates,
                        optional=False,
                    ),
                ),
            ],
            "express": [
                (
                    "akshare.stock_yjkb_em.eastmoney",
                    lambda: self._financial_recent_filtered_payload(
                        lambda date: ak.stock_yjkb_em(date=date),
                        code,
                        kind,
                        limit,
                        report_dates,
                        optional=True,
                    ),
                ),
            ],
        }
        return self._cached(
            "stock_financial",
            {
                "stock_code": code.display,
                "kind": kind,
                "limit": limit,
                "report_date": primary_report_date,
                "auto_report_date": not explicit_report_date,
                "start_year": effective_start_year,
                "end_year": effective_end_year,
            },
            TTL["financial"],
            source_map[kind],
        )

    def china_macro(self, indicator: str, limit: int = 200) -> dict[str, Any]:
        indicator_map: dict[str, Callable[[], pd.DataFrame]] = {
            "cpi_monthly": ak.macro_china_cpi_monthly,
            "cpi_yearly": ak.macro_china_cpi_yearly,
            "ppi_yearly": ak.macro_china_ppi_yearly,
            "gdp_yearly": ak.macro_china_gdp_yearly,
            "pmi_yearly": ak.macro_china_pmi_yearly,
            "money_supply": ak.macro_china_money_supply,
        }
        jin10_map = {
            "cpi_monthly": ("72", "\u4e2d\u56fdCPI\u6708\u7387\u62a5\u544a"),
            "cpi_yearly": ("56", "\u4e2d\u56fdCPI\u5e74\u7387\u62a5\u544a"),
            "ppi_yearly": ("60", "\u4e2d\u56fdPPI\u5e74\u7387\u62a5\u544a"),
            "gdp_yearly": ("57", "\u4e2d\u56fdGDP\u5e74\u7387\u62a5\u544a"),
            "pmi_yearly": ("65", "\u4e2d\u56fd\u5b98\u65b9\u5236\u9020\u4e1aPMI"),
        }
        if indicator not in indicator_map:
            raise ValueError(f"unsupported macro indicator: {indicator}")
        limit = max(1, min(int(limit), 5000))
        sources: list[tuple[str, SourceCallable]] = []
        if indicator in jin10_map:
            attr_id, symbol = jin10_map[indicator]
            sources.append((f"jin10.reports.{indicator}", lambda: self._jin10_macro_payload(attr_id, symbol, limit)))
        sources.append((f"akshare.{indicator_map[indicator].__name__}", lambda: self._macro_records_payload(indicator_map[indicator](), limit)))
        return self._cached(
            "china_macro",
            {"indicator": indicator, "limit": limit},
            TTL["macro"],
            sources,
            timeout_seconds=20,
        )

    def china_bond_yield_curve(
        self,
        start_date: str | None = None,
        end_date: str | None = None,
        days: int = 30,
        limit: int = 500,
    ) -> dict[str, Any]:
        end = self._iso_date(end_date) if end_date else self._now_cn().strftime("%Y-%m-%d")
        if start_date:
            start = self._iso_date(start_date)
        else:
            lookback_days = max(int(days) * 2, int(days) + 10)
            start = (datetime.fromisoformat(end) - timedelta(days=lookback_days)).strftime("%Y-%m-%d")
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "bond_yield",
            {"start_date": start, "end_date": end, "limit": limit},
            TTL["bond_yield"],
            [
                ("chinabond.government_bond.history_query", lambda: self._chinabond_yield_curve_payload(start, end, limit)),
            ],
            timeout_seconds=30,
        )

    def trade_calendar(self, start_date: str | None = None, end_date: str | None = None, limit: int = 5000) -> dict[str, Any]:
        start = self._iso_date(start_date) if start_date else None
        end = self._iso_date(end_date) if end_date else None
        limit = max(1, min(int(limit), 10000))
        return self._cached(
            "trade_calendar",
            {"start_date": start or "", "end_date": end or "", "limit": limit},
            TTL["trade_calendar"],
            [
                ("akshare.tool_trade_date_hist_sina", lambda: self._trade_calendar_sina_payload(start, end, limit)),
            ],
            timeout_seconds=30,
        )

    # ---- 股指期货基差 ----

    def futures_basis(self, days: int = 1) -> dict[str, Any]:
        days = max(1, min(int(days), 120))
        return self._cached(
            "futures_basis",
            {"days": days},
            600,
            [
                ("sina.cffex_futures_basis", lambda: self._futures_basis_payload(days)),
            ],
            timeout_seconds=30,
        )

    def us_indices(self, limit: int = 20) -> dict[str, Any]:
        limit = max(1, min(int(limit), 100))
        return self._cached(
            "us_indices",
            {"limit": limit},
            TTL["us_indices"],
            [
                ("tencent.qt.us_indices", lambda: self._us_indices_tencent_payload(limit)),
                ("sina.hq.us_indices", lambda: self._us_indices_sina_payload(limit)),
                ("yahoo.chart.us_indices", lambda: self._us_indices_yahoo_payload(limit)),
            ],
            timeout_seconds=12,
            retry_attempts=2,
        )
    def us_sectors(self, limit: int = 20) -> dict[str, Any]:
        limit = max(1, min(int(limit), 50))
        return self._cached(
            "us_sectors",
            {"limit": limit},
            TTL["us_sectors"],
            [
                ("yahoo.chart.us_sector_etfs", lambda: self._us_sectors_yahoo_payload(limit)),
            ],
            timeout_seconds=20,
        )
    def us_market_summary(self) -> dict[str, Any]:
        return self._cached(
            "us_market_summary",
            {},
            TTL["us_market_summary"],
            [
                ("composite.us_market_summary", lambda: self._us_market_summary_payload()),
            ],
            timeout_seconds=25,
        )
    def us_stock_quote(self, symbol: str) -> dict[str, Any]:
        ticker = self._normalize_us_symbol(symbol)
        return self._cached(
            "us_stock_quote",
            {"symbol": ticker},
            TTL["us_stock_quote"],
            [
                ("yahoo.chart.us_quote", lambda: self._us_stock_quote_yahoo(ticker)),
                ("sina.hq.us_quote", lambda: self._us_stock_quote_sina(ticker)),
            ],
            timeout_seconds=12,
            retry_attempts=2,
        )
    def us_stock_daily(self, symbol: str, days: int = 60) -> dict[str, Any]:
        ticker = self._normalize_us_symbol(symbol)
        days = max(1, min(int(days), 2000))
        return self._cached(
            "us_stock_daily",
            {"symbol": ticker, "days": days},
            TTL["stock_daily"] if days > 30 else 6 * 3600,
            [
                ("yahoo.chart.us_daily", lambda: self._us_stock_daily_yahoo(ticker, days)),
            ],
            timeout_seconds=15,
        )
    def x_timeline(self, accounts: str, limit: int = 5, hydrate: bool = True) -> dict[str, Any]:
        handles = self._parse_x_handles(accounts)
        if not handles:
            raise ValueError("accounts is required, e.g. elonmusk,OpenAI")
        limit = max(1, min(int(limit), 10))
        return self._cached(
            "x_timeline",
            {"accounts": ",".join(handles), "limit": limit, "hydrate": bool(hydrate)},
            TTL["x_timeline"],
            [
                ("openai_compatible.x_watchlist", lambda: self._x_timeline_via_model(handles, limit, hydrate)),
                ("x.com.html.timeline", lambda: self._x_timeline_via_html(handles, limit)),
            ],
            timeout_seconds=90,
        )


    def _futures_basis_payload(self, days: int = 1) -> dict[str, Any]:
        """计算股指期货基差：现货指数 - 期货主力合约收盘价

        days=1 时返回当日基差；days>1 时返回历史 N 天的基差序列。
        """
        contracts = [
            {"symbol": "IF0", "index_code": "sh000300", "name": "沪深300", "short": "IF"},
            {"symbol": "IH0", "index_code": "sh000016", "name": "上证50", "short": "IH"},
            {"symbol": "IC0", "index_code": "sh000905", "name": "中证500", "short": "IC"},
            {"symbol": "IM0", "index_code": "sh000852", "name": "中证1000", "short": "IM"},
        ]

        if days == 1:
            # ---- 当日基差（实时） ----
            spot_symbols = ",".join(c["index_code"] for c in contracts)
            spot_text = self._request_text(
                f"http://hq.sinajs.cn/list={spot_symbols}",
                headers={"Referer": "http://finance.sina.com.cn"},
                encoding="gbk",
            )
            spot_prices: dict[str, float] = {}
            for line in spot_text.strip().split("\n"):
                if '=""' in line or "=" not in line:
                    continue
                parts = line.split("=", 1)
                code = parts[0].split("_")[-1]
                vals = parts[1].strip().strip(";").strip('"').split(",")
                if len(vals) >= 2:
                    try:
                        spot_prices[code] = float(vals[1])
                    except (ValueError, TypeError):
                        pass

            contract_order = {c["short"]: idx for idx, c in enumerate(contracts)}

            def current_record(c: dict[str, str]) -> dict[str, Any]:
                rows = self._sina_futures_main_daily_rows(c["symbol"])
                latest = rows[-1]
                fut_close = self._float(latest.get("c"))
                fut_date = str(latest.get("d") or "")[:10]
                spot = spot_prices.get(c["index_code"])
                if fut_close is None or spot is None:
                    raise ValueError(f"futures basis current data missing: {c['symbol']}")
                basis = round(spot - fut_close, 2)
                basis_pct = round(basis / spot * 100, 4) if spot else None
                return {
                    "contract": c["short"], "name": c["name"],
                    "spot": round(spot, 2), "futures": round(fut_close, 2),
                    "basis": basis, "basis_pct": basis_pct, "date": fut_date,
                }

            records = []
            with ThreadPoolExecutor(max_workers=len(contracts)) as executor:
                futures = [executor.submit(current_record, c) for c in contracts]
                for future in as_completed(futures):
                    try:
                        records.append(future.result())
                    except Exception:
                        continue
            records.sort(key=lambda row: contract_order.get(row["contract"], 999))
            if not records:
                raise ValueError("futures basis data empty")

            # ---- 日内基差走势（5分钟级） ----
            def contract_intraday(c: dict[str, str]) -> tuple[str, list[dict[str, Any]]]:
                fut_rows = self._sina_futures_minute_rows(c["symbol"], "5")
                kline_url = (
                    "http://money.finance.sina.com.cn/quotes_service/api/"
                    "json_v2.php/CN_MarketData.getKLineData"
                    f"?symbol={c['index_code']}&scale=5&ma=no&datalen=60"
                )
                kline_text = self._request_text(
                    kline_url,
                    headers={"Referer": "http://finance.sina.com.cn"},
                    timeout=8,
                )
                kline_data = json.loads(kline_text)

                spot_minutes: dict[str, float] = {}
                for rec in kline_data:
                    dt_str = rec.get("day", "")
                    spot_close = self._float(rec.get("close"))
                    if dt_str and spot_close is not None:
                        spot_minutes[dt_str] = spot_close

                intraday = []
                for row in fut_rows:
                    dt_str = str(row.get("d") or "")
                    spot_close = spot_minutes.get(dt_str)
                    fut_close = self._float(row.get("c"))
                    if spot_close is None or fut_close is None:
                        continue
                    basis = round(spot_close - fut_close, 2)
                    intraday.append({
                        "time": dt_str,
                        "spot": round(spot_close, 2),
                        "futures": round(fut_close, 2),
                        "basis": basis,
                    })
                return c["short"], intraday

            intraday_by_contract = {}
            with ThreadPoolExecutor(max_workers=len(contracts)) as executor:
                futures = [executor.submit(contract_intraday, c) for c in contracts]
                for future in as_completed(futures):
                    try:
                        contract, intraday = future.result()
                    except Exception:
                        continue
                    if intraday:
                        intraday_by_contract[contract] = intraday

            result = {"count": len(records), "records": records}
            if intraday_by_contract:
                result["intraday"] = intraday_by_contract
            return result

        else:
            # ---- 历史基差序列 ----
            def contract_history(c: dict[str, str]) -> list[dict[str, Any]]:
                fut_rows = self._sina_futures_main_daily_rows(c["symbol"])[-days:]
                fut_dates = {}
                for row in fut_rows:
                    d = str(row.get("d") or "")[:10]
                    fut = self._float(row.get("c"))
                    if d and fut is not None:
                        fut_dates[d] = fut

                kline_url = (
                    "http://money.finance.sina.com.cn/quotes_service/api/"
                    "json_v2.php/CN_MarketData.getKLineData"
                    f"?symbol={c['index_code']}&scale=240&ma=no&datalen={days + 5}"
                )
                kline_text = self._request_text(
                    kline_url,
                    headers={"Referer": "http://finance.sina.com.cn"},
                    timeout=8,
                )
                kline_data = json.loads(kline_text)
                spot_dates: dict[str, float] = {}
                for rec in kline_data:
                    d = rec.get("day", "")[:10]
                    spot = self._float(rec.get("close"))
                    if d and spot is not None:
                        spot_dates[d] = spot

                records = []
                for d in sorted(fut_dates.keys())[-days:]:
                    spot = spot_dates.get(d)
                    fut = fut_dates.get(d)
                    if spot is None or fut is None:
                        continue
                    basis = round(spot - fut, 2)
                    basis_pct = round(basis / spot * 100, 4) if spot else None
                    records.append({
                        "contract": c["short"], "name": c["name"],
                        "spot": round(spot, 2), "futures": round(fut, 2),
                        "basis": basis, "basis_pct": basis_pct, "date": d,
                    })
                return records

            all_records = []
            with ThreadPoolExecutor(max_workers=len(contracts)) as executor:
                futures = [executor.submit(contract_history, c) for c in contracts]
                for future in as_completed(futures):
                    try:
                        all_records.extend(future.result())
                    except Exception:
                        continue

            if not all_records:
                raise ValueError("futures basis history data empty")
            all_records.sort(key=lambda row: (row["date"], row["contract"]))

            # 按日期分组，方便前端画趋势
            by_date: dict[str, list] = {}
            for r in all_records:
                by_date.setdefault(r["date"], []).append(r)

            summary = []
            for d in sorted(by_date.keys()):
                summary.append({"date": d, "contracts": by_date[d]})

            return {"days": days, "count": len(all_records), "by_date": summary, "records": all_records}

    # ---- 融资融券 ----

    def stock_margin(self, stock_code: str, days: int = 30) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        days = max(1, min(int(days), 500))
        return self._cached(
            "margin_stock",
            {"stock_code": code.display, "days": days},
            TTL["margin_stock"],
            [
                ("eastmoney.datacenter.rzrq_individual", lambda: self._stock_margin_datacenter(code, days)),
                ("eastmoney.push2his.stock_rrg", lambda: self._stock_margin_direct(code, days)),
                ("akshare.stock_margin_detail_sse", lambda: self._stock_margin_akshare_sse(code, days)),
                ("akshare.stock_margin_detail_szse", lambda: self._stock_margin_akshare_szse(code, days)),
            ],
            timeout_seconds=30,
        )

    def market_margin(self, date: str | None = None, limit: int = 50) -> dict[str, Any]:
        trade_date = self._compact_date(date) if date else self._prev_trade_date()
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "margin_market",
            {"date": trade_date, "limit": limit},
            TTL["margin_market"],
            [
                ("akshare.stock_margin_sse", lambda: self._market_margin_akshare_sse(trade_date, limit)),
                ("eastmoney.datacenter.rzrq_summary", lambda: self._market_margin_direct(trade_date, limit)),
            ],
        )

    # ---- 北向资金 ----

    def northbound_hist(self, days: int = 30) -> dict[str, Any]:
        days = max(1, min(int(days), 500))
        return self._cached(
            "northbound_hist",
            {"days": days},
            TTL["northbound_hist"],
            [
                ("eastmoney.push2his.kamt_kline", lambda: self._northbound_hist_direct(days)),
                ("akshare.stock_hsgt_hist_em", lambda: self._northbound_hist_akshare(days)),
            ],
        )

    def northbound_realtime(self) -> dict[str, Any]:
        return self._cached(
            "northbound_realtime",
            {},
            TTL["northbound_realtime"],
            [
                ("eastmoney.push2delay.kamt_rtmin", lambda: self._northbound_realtime_direct("https://push2delay.eastmoney.com")),
                ("akshare.stock_hsgt_fund_min_em", lambda: self._northbound_realtime_akshare()),
                ("eastmoney.push2.kamt_rtmin", lambda: self._northbound_realtime_direct("https://push2.eastmoney.com")),
            ],
            timeout_seconds=8,
        )

    def northbound_holdings(self, stock_code: str = "", limit: int = 50) -> dict[str, Any]:
        limit = max(1, min(int(limit), 500))
        if stock_code:
            code = normalize_stock_code(stock_code)
            return self._cached(
                "northbound_holdings",
                {"stock_code": code.display, "limit": limit},
                TTL["northbound_holdings"],
                [
                    ("eastmoney.datacenter.northbound_individual", lambda: self._northbound_holdings_individual_direct(code, limit)),
                    ("akshare.stock_hsgt_individual_em", lambda: self._northbound_holdings_individual(code, limit)),
                ],
            )
        return self._cached(
            "northbound_holdings",
            {"stock_code": "", "limit": limit},
            TTL["northbound_holdings"],
            [
                ("eastmoney.datacenter.rpt_mutual_market_sta", lambda: self._northbound_holdings_direct(limit)),
            ],
        )

    # ---- 申万行业 & 基金持仓 ----

    def shenwan_industry(self, limit: int = 50) -> dict[str, Any]:
        limit = max(1, min(int(limit), 200))
        return self._cached(
            "shenwan_industry",
            {"limit": limit},
            TTL["shenwan_industry"],
            [
                ("eastmoney.push2delay.sw_industry", lambda: self._shenwan_industry_direct(limit, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2.sw_industry", lambda: self._shenwan_industry_direct(limit, "https://push2.eastmoney.com")),
                ("akshare.index_realtime_sw", lambda: self._shenwan_industry_akshare(limit)),
            ],
            timeout_seconds=60,
        )

    def fund_holdings(self, date: str = "20260331", limit: int = 100) -> dict[str, Any]:
        date = self._compact_date(date)
        limit = max(1, min(int(limit), 5000))
        return self._cached(
            "fund_holdings",
            {"date": date, "limit": limit},
            TTL["fund_holdings"],
            [
                ("eastmoney.dataapi.zlsj_fund_hold", lambda: self._fund_holdings_zlsj_direct(date, limit)),
                ("akshare.fund_report_stock_cninfo", lambda: self._fund_holdings_akshare(date, limit)),
                ("eastmoney.datacenter.rpt_fund_main_position", lambda: self._fund_holdings_direct(date, limit)),
            ],
        )

    def fund_structure(self) -> dict[str, Any]:
        return self._cached(
            "fund_structure",
            {},
            TTL["fund_structure"],
            [
                ("eastmoney.funddata.hypzDetail", lambda: self._fund_structure_direct()),
                ("akshare.fund_hold_structure_em", lambda: self._fund_structure_akshare()),
                ("scrapling.eastmoney.fund_structure", lambda: self._fund_structure_scrapling()),
            ],
        )

    # ---- 股东户数 ----

    def stock_shareholders(self, stock_code: str, limit: int = 12) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 50))
        return self._cached(
            "shareholders",
            {"stock_code": code.display, "limit": limit},
            TTL["shareholders"],
            [
                ("eastmoney.datacenter.holdernum", lambda: self._shareholders_direct(code, limit)),
                ("eastmoney.f10.pageajax", lambda: self._shareholders_f10(code, limit)),
                ("akshare.stock_zh_a_gdhs_detail_em", lambda: self._shareholders_akshare(code, limit)),
            ],
        )

    def stock_shareholder_top(self, stock_code: str, date: str | None = None) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        iso_date = self._iso_date(date) if date else None
        return self._cached(
            "shareholder_top",
            {"stock_code": code.display, "date": iso_date or ""},
            TTL["shareholder_top"],
            [
                ("akshare.stock_main_stock_holder", lambda: self._shareholder_top_akshare(code)),
                (
                    "eastmoney.f10.freeholders.fallback",
                    lambda: self._eastmoney_f10_payload(
                        "freeholders",
                        code,
                        50,
                        start_date=iso_date,
                        end_date=iso_date,
                    ),
                ),
            ],
        )

    def stock_income(self, stock_code: str, limit: int = 10, kind: str = "ytd") -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        kind = "quarterly" if kind in {"quarterly", "sq", "single_quarter"} else "ytd"
        endpoint = "income_quarterly" if kind == "quarterly" else "income_ytd"
        return self._cached(
            "income",
            {"stock_code": code.display, "limit": limit, "kind": kind},
            TTL["income"],
            [
                (f"eastmoney.f10.{endpoint}", lambda: self._eastmoney_f10_payload(endpoint, code, limit)),
                ("eastmoney.datacenter.income.legacy", lambda: self._income_akshare(code, limit)),
            ],
        )

    def stock_balancesheet(self, stock_code: str, limit: int = 10) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        return self._cached(
            "balancesheet",
            {"stock_code": code.display, "limit": limit},
            TTL["balancesheet"],
            [
                ("eastmoney.f10.balance", lambda: self._eastmoney_f10_payload("balance", code, limit)),
                ("akshare.stock_balance_sheet_by_report_em", lambda: self._balancesheet_akshare(code, limit)),
            ],
        )

    def stock_cashflow(self, stock_code: str, limit: int = 10, kind: str = "ytd") -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        kind = "quarterly" if kind in {"quarterly", "sq", "single_quarter"} else "ytd"
        endpoint = "cashflow_quarterly" if kind == "quarterly" else "cashflow_ytd"
        return self._cached(
            "cashflow",
            {"stock_code": code.display, "limit": limit, "kind": kind},
            TTL["cashflow"],
            [
                (f"eastmoney.f10.{endpoint}", lambda: self._eastmoney_f10_payload(endpoint, code, limit)),
            ],
        )

    def stock_dividends(self, stock_code: str, limit: int = 20, kind: str = "main") -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        kind = "allotment" if kind in {"allotment", "all"} else "main"
        endpoint = "dividend_allotment" if kind == "allotment" else "dividend_main"
        return self._cached(
            "dividends",
            {"stock_code": code.display, "limit": limit, "kind": kind},
            TTL["dividends"],
            [
                (f"eastmoney.f10.{endpoint}", lambda: self._eastmoney_f10_payload(endpoint, code, limit, allow_empty=kind == "allotment")),
            ],
        )

    def stock_equity_history(self, stock_code: str, limit: int = 20) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        return self._cached(
            "equity_history",
            {"stock_code": code.display, "limit": limit},
            TTL["equity_history"],
            [
                ("eastmoney.f10.equity_history", lambda: self._eastmoney_f10_payload("equity_history", code, limit)),
            ],
        )

    def stock_freeholders(self, stock_code: str, limit: int = 20, end_date: str | None = None) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        limit = max(1, min(int(limit), 200))
        iso_end_date = self._iso_date(end_date) if end_date else None
        return self._cached(
            "freeholders",
            {"stock_code": code.display, "limit": limit, "end_date": iso_end_date or ""},
            TTL["freeholders"],
            [
                (
                    "eastmoney.f10.freeholders",
                    lambda: self._eastmoney_f10_payload(
                        "freeholders",
                        code,
                        limit,
                        start_date=iso_end_date,
                        end_date=iso_end_date,
                    ),
                ),
                ("akshare.stock_main_stock_holder", lambda: self._shareholder_top_akshare(code)),
            ],
        )

    def stock_daily_basic(self, stock_code: str, days: int = 30) -> dict[str, Any]:
        code = normalize_stock_code(stock_code)
        days = max(1, min(int(days), 500))
        return self._cached(
            "daily_basic",
            {"stock_code": code.display, "days": days},
            TTL["daily_basic"],
            [
                ("eastmoney.push2.daily_basic", lambda: self._daily_basic_lg(code, days)),
            ],
        )

    def legacy_board_concept_hot(self, lookback_days: int = 1, limit: int = 30) -> dict[str, Any]:
        limit = max(1, min(int(limit), 100))
        return self._cached(
            "legacy_board_concept_hot",
            {"lookback_days": lookback_days, "limit": limit},
            TTL["board_flow"],
            [
                ("akshare.stock_fund_flow_concept.eastmoney", lambda: self._format_board_flow_text(self._concept_flow_df(), "今日概念板块热度排行", limit)),
                ("akshare.stock_fund_flow_industry.eastmoney", lambda: self._format_board_flow_text(self._industry_flow_df(), "今日行业板块热度排行（备用数据源）", limit)),
            ],
        )

    def legacy_board_sector_flow(self, limit: int = 30) -> dict[str, Any]:
        limit = max(1, min(int(limit), 100))
        return self._cached(
            "legacy_board_sector_flow",
            {"limit": limit},
            TTL["board_flow"],
            [
                ("akshare.stock_fund_flow_industry.eastmoney", lambda: self._format_board_flow_text(self._industry_flow_df(), "行业板块资金流向排名", limit)),
                ("akshare.stock_fund_flow_concept.eastmoney", lambda: self._format_board_flow_text(self._concept_flow_df(), "概念板块资金流向（备用数据源）", limit)),
            ],
        )

    def legacy_board_temperature(self, date: str | None = None) -> dict[str, Any]:
        trade_date = self._compact_date(date) if date else self._now_cn().strftime("%Y%m%d")
        return self._cached(
            "legacy_board_temperature",
            {"date": trade_date},
            TTL["market_temperature"],
            [
                ("combined.limit_pool_and_breadth", lambda: self._format_temperature_text(self._market_temperature_payload(trade_date))),
            ],
            timeout_seconds=35,
        )

    def legacy_board_news(self, stock_code: str = "000300", data_type: str = "global_news", limit: int = 20) -> dict[str, Any]:
        limit = max(1, min(int(limit), 100))
        data_type = data_type if data_type in {"news", "global_news", "notice", "important_news", "hotspot_news"} else "global_news"
        return self._cached(
            "legacy_board_news",
            {"stock_code": stock_code, "data_type": data_type, "limit": limit},
            TTL["news" if data_type != "notice" else "notice"],
            [
                ("legacy.board_news", lambda: self._format_legacy_news(stock_code, data_type, limit)),
            ],
        )

    def legacy_sector_ambush(self, lookback_days: int = 60, limit: int = 10) -> dict[str, Any]:
        limit = max(1, min(int(limit), 50))
        lookback_days = max(1, min(int(lookback_days), 250))
        return self._cached(
            "legacy_sector_ambush",
            {"lookback_days": lookback_days, "limit": limit},
            TTL["board_flow"],
            [
                ("eastmoney.push2delay.board_flow.industry.legacy", lambda: self._format_sector_ambush(self._legacy_board_flow_direct_df("industry", 200, "https://push2delay.eastmoney.com"), lookback_days, limit)),
                ("eastmoney.push2delay.board_flow.concept.legacy", lambda: self._format_sector_ambush(self._legacy_board_flow_direct_df("concept", 200, "https://push2delay.eastmoney.com"), lookback_days, limit)),
                ("eastmoney.push2delay.board_flow.combined.legacy", lambda: self._format_sector_ambush(self._legacy_board_flow_combined_df("https://push2delay.eastmoney.com"), lookback_days, limit)),
                ("akshare.stock_fund_flow_concept.eastmoney", lambda: self._format_sector_ambush(self._concept_flow_df(), lookback_days, limit)),
                ("akshare.stock_fund_flow_industry.eastmoney", lambda: self._format_sector_ambush(self._industry_flow_df(), lookback_days, limit)),
                ("eastmoney.push2.board_flow.concept.legacy", lambda: self._format_sector_ambush(self._legacy_board_flow_direct_df("concept", 200, "https://push2.eastmoney.com"), lookback_days, limit)),
            ],
            timeout_seconds=8,
        )

    def legacy_sector_leaders(self, sector_name: str, lookback_days: int = 5, top_n: int = 3) -> dict[str, Any]:
        top_n = max(1, min(int(top_n), 10))
        lookback_days = max(1, min(int(lookback_days), 250))
        return self._cached(
            "legacy_sector_leaders",
            {"sector_name": sector_name, "lookback_days": lookback_days, "top_n": top_n},
            TTL["board_flow"],
            [
                ("eastmoney.push2delay.board_flow.combined.legacy", lambda: self._format_sector_leaders(self._legacy_board_flow_combined_df("https://push2delay.eastmoney.com"), sector_name, lookback_days, top_n)),
                ("eastmoney.push2delay.board_flow.concept.legacy", lambda: self._format_sector_leaders(self._legacy_board_flow_direct_df("concept", 500, "https://push2delay.eastmoney.com"), sector_name, lookback_days, top_n)),
                ("eastmoney.push2delay.board_flow.industry.legacy", lambda: self._format_sector_leaders(self._legacy_board_flow_direct_df("industry", 300, "https://push2delay.eastmoney.com"), sector_name, lookback_days, top_n)),
                ("akshare.stock_fund_flow_concept.eastmoney", lambda: self._format_sector_leaders(self._concept_flow_df(), sector_name, lookback_days, top_n)),
                ("akshare.stock_fund_flow_industry.eastmoney", lambda: self._format_sector_leaders(self._industry_flow_df(), sector_name, lookback_days, top_n)),
                ("eastmoney.push2.board_flow.concept.legacy", lambda: self._format_sector_leaders(self._legacy_board_flow_direct_df("concept", 500, "https://push2.eastmoney.com"), sector_name, lookback_days, top_n)),
            ],
            timeout_seconds=8,
        )

    def legacy_leader_frequency(self, stock_name: str, lookback_days: int = 5) -> dict[str, Any]:
        lookback_days = max(1, min(int(lookback_days), 250))
        return self._cached(
            "legacy_leader_frequency",
            {"stock_name": stock_name, "lookback_days": lookback_days},
            TTL["board_flow"],
            [
                ("eastmoney.push2delay.board_flow.concept.legacy", lambda: self._format_leader_frequency(self._legacy_board_flow_direct_df("concept", 500, "https://push2delay.eastmoney.com"), stock_name)),
                ("eastmoney.push2delay.board_flow.combined.legacy", lambda: self._format_leader_frequency(self._legacy_board_flow_combined_df("https://push2delay.eastmoney.com"), stock_name)),
                ("eastmoney.push2delay.board_flow.industry.legacy", lambda: self._format_leader_frequency(self._legacy_board_flow_direct_df("industry", 300, "https://push2delay.eastmoney.com"), stock_name)),
                ("akshare.stock_fund_flow_concept.eastmoney", lambda: self._format_leader_frequency(self._concept_flow_df(), stock_name)),
                ("akshare.stock_fund_flow_industry.eastmoney", lambda: self._format_leader_frequency(self._industry_flow_df(), stock_name)),
                ("eastmoney.push2.board_flow.concept.legacy", lambda: self._format_leader_frequency(self._legacy_board_flow_direct_df("concept", 500, "https://push2.eastmoney.com"), stock_name)),
            ],
            timeout_seconds=8,
        )

    def _cached(
        self,
        namespace: str,
        params: dict[str, Any],
        ttl_seconds: int,
        sources: list[tuple[str, SourceCallable]],
        retry_attempts: int | None = None,
        timeout_seconds: int | None = None,
        min_rows: int | None = None,
        circuit_retries: int = 0,
    ) -> dict[str, Any]:
        key = cache.make_key(namespace, params)
        cached = cache.get(key)
        if cached:
            return self._envelope(
                namespace,
                cached.payload,
                cached.source,
                key,
                ttl_seconds,
                cache_hit=True,
                stale=False,
                attempts=[],
                cache_entry=cached,
            )

        try:
            result, attempts = run_sources(
                namespace,
                sources,
                timeout_seconds or settings.source_timeout_seconds,
                retry_attempts=retry_attempts or settings.source_retry_attempts,
                retry_backoff_seconds=settings.source_retry_backoff_seconds,
                retry_max_backoff_seconds=settings.source_retry_max_backoff_seconds,
                retry_jitter_seconds=settings.source_retry_jitter_seconds,
                min_rows=min_rows,
                circuit_retries=circuit_retries,
            )
            payload = to_jsonable(result.data)
            row_count = self._row_count(payload)
            saved = cache.set(key, namespace, payload, result.source, ttl_seconds, row_count=row_count)
            return self._envelope(
                namespace,
                payload,
                result.source,
                key,
                ttl_seconds,
                cache_hit=False,
                stale=False,
                attempts=attempts,
                cache_entry=saved,
            )
        except AllSourcesFailed as exc:
            stale = cache.get(key, allow_expired=True)
            if stale:
                return self._envelope(
                    namespace,
                    stale.payload,
                    stale.source,
                    key,
                    ttl_seconds,
                    cache_hit=True,
                    stale=True,
                    attempts=exc.attempts,
                    cache_entry=stale,
                )
            raise

    def _envelope(
        self,
        endpoint: str,
        data: Any,
        source: str,
        cache_key: str,
        ttl_seconds: int,
        cache_hit: bool,
        stale: bool,
        attempts: list[dict[str, Any]],
        cache_entry: CacheEntry | None,
    ) -> dict[str, Any]:
        now = datetime.now(UTC)
        expires_at = cache_entry.expires_at.isoformat() if cache_entry else (now + timedelta(seconds=ttl_seconds)).isoformat()
        created_at = cache_entry.created_at.isoformat() if cache_entry else now.isoformat()
        row_count = cache_entry.row_count if cache_entry else self._row_count(data)
        return {
            "ok": True,
            "data": data,
            "meta": {
                "endpoint": endpoint,
                "source": source,
                "generated_at": now.isoformat(),
                "cache": {
                    "hit": cache_hit,
                    "stale": stale,
                    "key": cache_key,
                    "ttl_seconds": ttl_seconds,
                    "created_at": created_at,
                    "expires_at": expires_at,
                    "row_count": row_count,
                },
                "attempts": attempts,
            },
        }

    def _http_get(
        self,
        url: str,
        params: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
        timeout: float | None = None,
    ) -> Any:
        """HTTP GET using curl_cffi when available to impersonate a real browser.

        Eastmoney and similar endpoints often block plain urllib3/requests TLS
        fingerprints; curl_cffi makes the same call look like Chrome and is
        accepted. Falls back to standard requests if curl_cffi is unavailable.
        """
        timeout_value = timeout or settings.source_timeout_seconds
        try:
            import curl_cffi.requests as curl_requests

            return curl_requests.get(
                url,
                params=params,
                headers=headers,
                impersonate="chrome131",
                timeout=timeout_value,
            )
        except Exception:
            session = requests.Session()
            return session.get(
                url,
                params=params,
                headers=headers,
                timeout=timeout_value,
            )

    def _request_json(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        params: dict[str, Any] | None = None,
        timeout: float | None = None,
    ) -> dict[str, Any]:
        merged_headers = dict(HTTP_HEADERS)
        if headers:
            merged_headers.update(headers)
        response = self._http_get(url, params=params, headers=merged_headers, timeout=timeout)
        response.raise_for_status()
        content = response.content
        for enc in ("utf-8-sig", "utf-8", "gbk", "gb2312"):
            try:
                text = content.decode(enc)
                break
            except UnicodeDecodeError:
                continue
        else:
            text = content.decode("utf-8-sig", errors="replace")
        text = text.strip()
        if text.startswith(("jQuery", "callback")) and "(" in text and text.endswith(")"):
            text = text[text.find("(") + 1 : -1]
        return json.loads(text)

    def _request_text(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        encoding: str | None = None,
        timeout: float | None = None,
    ) -> str:
        merged_headers = dict(HTTP_HEADERS)
        if headers:
            merged_headers.update(headers)
        response = self._http_get(url, headers=merged_headers, timeout=timeout)
        response.raise_for_status()
        content = response.content
        if encoding:
            return content.decode(encoding, errors="replace")
        # Try robust decoding
        for enc in ("utf-8-sig", "utf-8", "gbk", "gb2312", "latin-1"):
            try:
                return content.decode(enc)
            except UnicodeDecodeError:
                continue
        return content.decode("utf-8-sig", errors="replace")

    def _date_range(self, days: int, start_date: str | None, end_date: str | None) -> tuple[str, str]:
        end = self._compact_date(end_date) if end_date else self._now_cn().strftime("%Y%m%d")
        if start_date:
            start = self._compact_date(start_date)
        else:
            lookback = max(days * 2, days + 30)
            start = (self._now_cn() - timedelta(days=lookback)).strftime("%Y%m%d")
        return start, end

    def _daily_ttl(self, end_date: str) -> int:
        today = self._now_cn().strftime("%Y%m%d")
        if end_date < today:
            return TTL["stock_daily"]
        return 6 * 3600

    def _now_cn(self) -> datetime:
        return datetime.now(CN_TZ)

    def _compact_date(self, value: str) -> str:
        compact = str(value).replace("-", "").replace("/", "").strip()
        if len(compact) != 8 or not compact.isdigit():
            raise ValueError(f"invalid date: {value}, expected YYYYMMDD or YYYY-MM-DD")
        return compact

    def _iso_date(self, value: str) -> str:
        compact = self._compact_date(value)
        return f"{compact[:4]}-{compact[4:6]}-{compact[6:8]}"

    def _fund_code(self, value: str) -> str:
        code = str(value or "").strip().lower()
        if "." in code:
            code = code.split(".", 1)[0]
        if code.startswith(("sh", "sz", "bj")):
            code = code[2:]
        if len(code) != 6 or not code.isdigit():
            raise ValueError(f"invalid fund code: {value}")
        return code

    def _hk_stock_code(self, value: str) -> str:
        text = str(value or "").strip().lower()
        if "." in text:
            text = text.split(".", 1)[0]
        if text.startswith("hk"):
            text = text[2:]
        digits = "".join(ch for ch in text if ch.isdigit())
        if not digits or len(digits) > 5:
            raise ValueError(f"invalid hk stock code: {value}")
        return digits.zfill(5)

    def _sina_fund_symbol(self, code: str) -> str:
        if code.startswith(("5", "6")):
            return f"sh{code}"
        return f"sz{code}"

    def _etf_code(self, value: str) -> NormalizedStockCode:
        """Normalize an ETF code to a NormalizedStockCode-like object.

        Accepts 510300, 510300.SH, sh510300, 159048.SZ, etc.
        Raises ValueError for codes that cannot be mapped to a SH/SZ ETF.
        """
        raw = str(value or "").strip().lower()
        if not raw:
            raise ValueError(f"invalid fund code: {value}")
        code = raw
        if "." in code:
            code = code.split(".", 1)[0]
        if code.startswith(("sh", "sz", "bj")):
            code = code[2:]
        if len(code) != 6 or not code.isdigit():
            raise ValueError(f"invalid fund code: {value}")
        if code.startswith(("5", "6", "58", "56")):
            market = "sh"
        elif code.startswith(("15", "16")):
            market = "sz"
        else:
            raise ValueError(f"invalid fund code: {value}; expected SH/SZ ETF")
        suffix = market.upper()
        return NormalizedStockCode(
            code=code,
            market=market,
            suffix=suffix,
            prefixed=f"{market}{code}",
            display=f"{code}.{suffix}",
        )

    def _etf_secid(self, code: NormalizedStockCode) -> str:
        market = 1 if code.market == "sh" else 0
        return f"{market}.{code.code}"

    def etf_quote(self, fund_code: str) -> dict[str, Any]:
        code = self._etf_code(fund_code)
        return self._cached(
            "etf_quote",
            {"fund_code": code.display},
            TTL["etf_quote"],
            [
                ("sina.hq.realtime", lambda: self._sina_etf_quote_payload(code)),
                ("eastmoney.push2delay.etf_quote", lambda: self._eastmoney_etf_quote_clist(code, "https://push2delay.eastmoney.com")),
                ("eastmoney.push2.etf_quote", lambda: self._eastmoney_etf_quote_clist(code, "https://push2.eastmoney.com")),
                ("akshare.fund_etf_spot_em", lambda: self._etf_quote_from_spot_em(code)),
                ("akshare.fund_etf_spot_ths", lambda: self._etf_quote_from_spot_ths(code)),
                ("akshare.fund_etf_hist_em.latest", lambda: self._etf_quote_from_hist_em(code)),
                ("yahoo.chart.etf_quote", lambda: self._stock_quote_yahoo(code)),
            ],
        )

    def _etf_quote_from_spot_em(self, code: NormalizedStockCode) -> dict[str, Any]:
        df = self._require_df(ak.fund_etf_spot_em(), "fund_etf_spot_em")
        row_df = df[df["代码"].astype(str).str.strip().str.lower() == code.code]
        if row_df.empty:
            raise ValueError(f"ETF not found in spot list: {code.display}")
        row = row_df.iloc[0]
        return self._format_etf_quote(
            code,
            name=row.get("名称"),
            price=row.get("最新价"),
            change_pct=row.get("涨跌幅"),
            change_amount=row.get("涨跌额"),
            volume=row.get("成交量"),
            amount=row.get("成交额"),
            open=row.get("今开"),
            high=row.get("最高"),
            low=row.get("最低"),
            pre_close=row.get("昨收"),
            turnover=row.get("换手率"),
        )

    def _sina_etf_quote_payload(self, code: NormalizedStockCode) -> dict[str, Any]:
        text = self._request_text(
            f"http://hq.sinajs.cn/rn={int(self._now_cn().timestamp() * 1000)}&list={code.prefixed}",
            headers={"Referer": "http://finance.sina.com.cn"},
            encoding="gbk",
        )
        if '=""' in text:
            raise ValueError(f"sina hq returned empty quote: {code.display}")
        value = text.split("=", 1)[1].strip().strip(";").strip('"')
        fields = value.split(",")
        if len(fields) < 10:
            raise ValueError(f"sina hq returned malformed quote: {code.display}")
        pre_close = self._float(fields[2])
        price = self._float(fields[3])
        change_amount = (price - pre_close) if price is not None and pre_close else None
        return self._format_etf_quote(
            code,
            name=fields[0],
            price=price,
            change_pct=round(change_amount / pre_close * 100, 4) if change_amount is not None and pre_close else None,
            change_amount=round(change_amount, 4) if change_amount is not None else None,
            volume=self._float(fields[8]),
            amount=self._float(fields[9]),
            open=self._float(fields[1]),
            high=self._float(fields[4]),
            low=self._float(fields[5]),
            pre_close=pre_close,
            turnover=None,
            date=fields[30] if len(fields) > 30 else None,
            time=fields[31] if len(fields) > 31 else None,
        )

    def _eastmoney_etf_quote_clist(self, code: NormalizedStockCode, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        raw = self._request_json(
            f"{base_url.rstrip('/')}/api/qt/ulist.np/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "fltt": "2",
                "invt": "2",
                "secids": self._etf_secid(code),
                "fields": "f12,f14,f2,f3,f4,f5,f6,f8,f15,f16,f17,f18,f124",
                "ut": "b2884a393a59ad64002292a3e90d46a5",
            },
        )
        rows = (raw.get("data") or {}).get("diff") or []
        row = next((r for r in rows if str(r.get("f12") or "").zfill(6) == code.code), None)
        if row is None:
            raise ValueError(f"eastmoney etf quote returned no data: {code.display}")
        return self._format_etf_quote(
            code,
            name=row.get("f14"),
            price=row.get("f2"),
            change_pct=row.get("f3"),
            change_amount=row.get("f4"),
            volume=row.get("f5"),
            amount=row.get("f6"),
            open=row.get("f17"),
            high=row.get("f15"),
            low=row.get("f16"),
            pre_close=row.get("f18"),
            turnover=row.get("f8"),
            update_time=self._eastmoney_time(row.get("f124")),
        )

    def _etf_quote_from_spot_ths(self, code: NormalizedStockCode) -> dict[str, Any]:
        df = self._require_df(ak.fund_etf_spot_ths(), "fund_etf_spot_ths")
        code_col = next((c for c in df.columns if "代码" in str(c)), None)
        if not code_col:
            raise ValueError("fund_etf_spot_ths returned unexpected columns")
        row_df = df[df[code_col].astype(str).str[-6:].str.lower() == code.code]
        if row_df.empty:
            raise ValueError(f"ETF not found in THS spot list: {code.display}")
        row = row_df.iloc[0]
        return self._format_etf_quote(
            code,
            name=row.get("基金简称") or row.get("名称") or row.get("基金名称"),
            price=row.get("最新价") or row.get("最新"),
            change_pct=row.get("涨跌幅"),
            change_amount=row.get("涨跌额"),
            volume=row.get("成交量") or row.get("成交"),
            amount=row.get("成交额"),
            open=row.get("今开") or row.get("开盘价"),
            high=row.get("最高") or row.get("最高价"),
            low=row.get("最低") or row.get("最低价"),
            pre_close=row.get("昨收") or row.get("昨收价"),
            turnover=row.get("换手率"),
        )

    def _etf_quote_from_hist_em(self, code: NormalizedStockCode) -> dict[str, Any]:
        start, end = self._date_range(8, None, None)
        df = self._require_df(
            ak.fund_etf_hist_em(symbol=code.display, period="daily", start_date=start, end_date=end, adjust=""),
            "fund_etf_hist_em",
        )
        records = self._normalize_daily_records(df, "eastmoney")
        if not records:
            raise ValueError("fund_etf_hist_em returned empty data")
        latest = records[-1]
        prev = records[-2] if len(records) >= 2 else None
        pre_close = prev["close"] if prev else latest["open"]
        price = latest["close"]
        change_amount = (price or 0) - (pre_close or 0)
        change_pct = change_amount / pre_close * 100 if pre_close else None
        return self._format_etf_quote(
            code,
            name=code.code,
            price=price,
            change_pct=round(change_pct, 4) if change_pct is not None else None,
            change_amount=round(change_amount, 4),
            volume=latest.get("volume"),
            amount=latest.get("amount"),
            open=latest.get("open"),
            high=latest.get("high"),
            low=latest.get("low"),
            pre_close=pre_close,
            turnover=latest.get("turnover"),
            date=latest.get("date"),
        )

    def _format_etf_quote(
        self,
        code: NormalizedStockCode,
        *,
        name: Any,
        price: Any,
        change_pct: Any,
        change_amount: Any,
        volume: Any,
        amount: Any,
        open: Any,
        high: Any,
        low: Any,
        pre_close: Any,
        turnover: Any,
        date: str | None = None,
        time: str | None = None,
        update_time: str | None = None,
    ) -> dict[str, Any]:
        now = self._now_cn()
        return {
            "fund_code": code.display,
            "name": name if name is not None else code.code,
            "price": self._float(price),
            "change_pct": self._float(change_pct),
            "change_amount": self._float(change_amount),
            "volume": self._float(volume),
            "amount": self._float(amount),
            "open": self._float(open),
            "high": self._float(high),
            "low": self._float(low),
            "pre_close": self._float(pre_close),
            "turnover": self._float(turnover),
            "date": date or now.date().isoformat(),
            "time": time or update_time,
        }

    # ---- Sina futures helpers ----

    def _sina_jsonp_list(self, text: str, label: str) -> list[dict[str, Any]]:
        start = text.find("([")
        end = text.rfind("])")
        if start < 0 or end <= start:
            raise ValueError(f"{label} returned malformed jsonp")
        rows = json.loads(text[start + 1 : end + 1])
        if not isinstance(rows, list):
            raise ValueError(f"{label} returned non-list payload")
        return rows

    def _sina_futures_main_daily_rows(self, symbol: str) -> list[dict[str, Any]]:
        trade_date = "2021_08_17"
        text = self._request_text(
            (
                "https://stock2.finance.sina.com.cn/futures/api/jsonp.php/"
                f"var%20_{symbol}{trade_date}=/InnerFuturesNewService.getDailyKLine"
                f"?symbol={symbol}&_={trade_date}"
            ),
            headers={"Referer": "https://finance.sina.com.cn"},
            timeout=8,
        )
        rows = self._sina_jsonp_list(text, "sina futures daily")
        if not rows:
            raise ValueError(f"sina futures daily returned empty data: {symbol}")
        return rows

    def _sina_futures_minute_rows(self, symbol: str, period: str = "5") -> list[dict[str, Any]]:
        text = self._request_text(
            (
                "https://stock2.finance.sina.com.cn/futures/api/jsonp.php/=/"
                "InnerFuturesNewService.getFewMinLine"
                f"?symbol={symbol}&type={period}"
            ),
            headers={"Referer": "https://finance.sina.com.cn"},
            timeout=8,
        )
        rows = self._sina_jsonp_list(text, "sina futures minute")
        if not rows:
            raise ValueError(f"sina futures minute returned empty data: {symbol}")
        return rows

    # ---- 融资融券 helpers ----

    def _stock_margin_direct(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://push2his.eastmoney.com/api/qt/stock/rrg/kline/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "secid": self._stock_secid(code),
                "lmt": days,
                "klt": "101",
                "fields1": "f1,f2,f3,f7",
                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63",
                "ut": "b2884a393a59ad64002292a3e90d46a5",
            },
        )
        data = raw.get("data") or {}
        klines = data.get("klines") or []
        records = []
        for line in klines[-days:]:
            parts = str(line).split(",")
            if len(parts) < 7:
                continue
            records.append({
                "date": parts[0],
                "margin_balance": self._float(parts[1]),
                "margin_buy": self._float(parts[2]),
                "margin_repay": self._float(parts[3]),
                "short_balance": self._float(parts[4]),
                "short_sell": self._float(parts[5]),
                "short_repay": self._float(parts[6]),
                "total_balance": self._float(parts[1]) + self._float(parts[4]) if self._float(parts[1]) is not None and self._float(parts[4]) is not None else None,
            })
        if not records:
            raise ValueError("eastmoney stock margin kline returned empty data")
        return {"stock_code": code.display, "days": len(records), "records": records}

    def _stock_margin_datacenter(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://datacenter-web.eastmoney.com/api/data/v1/get",
            headers={"Referer": "https://data.eastmoney.com/"},
            params={
                "reportName": "RPTA_WEB_RZRQ_GGMX",
                "columns": "DATE,SCODE,SECNAME,RZYE,RQYE,RZRQYE,RZMRE,RQMCL",
                "filter": f'(SCODE="{code.code}")',
                "pageNumber": 1,
                "pageSize": days,
                "sortColumns": "DATE",
                "sortTypes": "-1",
                "source": "WEB",
                "client": "WEB",
            },
        )
        result = raw.get("result") or {}
        rows = result.get("data") or []
        records = []
        for row in rows[:days]:
            records.append({
                "date": str(row.get("DATE", ""))[:10],
                "margin_balance": self._float(row.get("RZYE")),
                "short_balance": self._float(row.get("RQYE")),
                "total_balance": self._float(row.get("RZRQYE")),
                "margin_buy": self._float(row.get("RZMRE")),
                "short_sell": self._float(row.get("RQMCL")),
            })
        if not records:
            raise ValueError("eastmoney datacenter margin returned empty data")
        return {"stock_code": code.display, "days": len(records), "records": records}

    def _stock_margin_akshare_sse(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        if code.market != "sh":
            raise ValueError("SSE margin data only for SH stocks")
        now = self._now_cn()
        records = []
        for offset in range(0, days * 3 + 10):
            date_str = (now - timedelta(days=offset)).strftime("%Y%m%d")
            try:
                df = ak.stock_margin_detail_sse(date=date_str)
            except Exception:
                continue
            if df is None or df.empty:
                continue
            if len(df.columns) >= 9:
                df = df.copy()
                df.columns = [
                    "信用交易日期",
                    "标的证券代码",
                    "标的证券简称",
                    "融资余额",
                    "融资买入额",
                    "融资偿还额",
                    "融券余量",
                    "融券卖出量",
                    "融券偿还量",
                ]
            code_col = "标的证券代码"
            if code_col not in df.columns:
                continue
            filtered = df[df[code_col].astype(str).str.strip() == code.code]
            if not filtered.empty:
                rec = dataframe_to_records(filtered, 1)
                if rec:
                    records.append(rec[0])
            if len(records) >= days:
                break
        if not records:
            raise ValueError("SSE margin detail data empty for this stock")
        return {"stock_code": code.display, "days": len(records), "records": records[:days]}


    def _stock_margin_akshare_szse(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        if code.market != "sz":
            raise ValueError("SZSE margin data only for SZ stocks")
        now = self._now_cn()
        # 尝试最近5个交易日
        for offset in range(1, 6):
            date_str = (now - timedelta(days=offset)).strftime("%Y%m%d")
            df = ak.stock_margin_detail_szse(date=date_str)
            if df is not None and not df.empty:
                code_col = next((c for c in df.columns if "代码" in str(c)), None)
                if code_col:
                    filtered = df[df[code_col].astype(str) == code.code]
                    if not filtered.empty:
                        records = dataframe_to_records(filtered, days)
                        return {"stock_code": code.display, "days": len(records), "records": records}
        raise ValueError("SZSE margin data empty for this stock")

    def _market_margin_direct(self, trade_date: str, limit: int) -> dict[str, Any]:
        date_iso = f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}"
        raw = self._request_json(
            "https://datacenter-web.eastmoney.com/api/data/v1/get",
            headers={"Referer": "https://data.eastmoney.com/"},
            params={
                "reportName": "RPTA_WEB_RZRQ_GGMX",
                "columns": "ALL",
                "filter": f"(TRADE_DATE='{date_iso}')",
                "pageNumber": 1,
                "pageSize": limit,
                "sortColumns": "RZYE",
                "sortTypes": "-1",
                "source": "WEB",
                "client": "WEB",
            },
        )
        result = raw.get("result") or {}
        rows = result.get("data") or []
        records = []
        for row in rows[:limit]:
            records.append({
                "stock_code": row.get("SECURITY_CODE"),
                "name": row.get("SECURITY_NAME_ABBR"),
                "date": str(row.get("TRADE_DATE", ""))[:10],
                "margin_balance": self._float(row.get("RZYE")),
                "margin_buy": self._float(row.get("RZMRE")),
                "short_balance": self._float(row.get("RQYE")),
                "short_sell": self._float(row.get("RQMCL")),
                "total_balance": self._float(row.get("RZRQYE")),
            })
        if not records:
            raise ValueError("eastmoney margin summary returned empty data")
        return {"date": date_iso, "count": len(records), "total": result.get("count"), "records": records}

    def _market_margin_akshare_sse(self, trade_date: str, limit: int) -> dict[str, Any]:
        start = (datetime.strptime(trade_date, "%Y%m%d") - timedelta(days=7)).strftime("%Y%m%d")
        df = self._require_df(ak.stock_margin_sse(start_date=start, end_date=trade_date), "stock_margin_sse")
        records = dataframe_to_records(df.tail(limit), limit)
        if not records:
            raise ValueError("SSE market margin data empty")
        return {"date": trade_date, "count": len(records), "records": records}

    # ---- 北向资金 helpers ----

    def _northbound_hist_akshare(self, days: int) -> dict[str, Any]:
        df = self._require_df(ak.stock_hsgt_hist_em(symbol="北向资金"), "stock_hsgt_hist_em")
        # Map Chinese column names to English for API consumers
        col_map = {}
        for col in df.columns:
            col_str = str(col)
            if "日期" in col_str:
                col_map[col] = "date"
            elif "当日成交净买额" in col_str:
                col_map[col] = "north_net"
            elif "沪股通" in col_str and "净买额" in col_str and "历史" not in col_str:
                col_map[col] = "sh_net"
            elif "深股通" in col_str and "净买额" in col_str and "历史" not in col_str:
                col_map[col] = "sz_net"
            elif "历史累计净买额" in col_str:
                col_map[col] = "cumulative_net"
            elif "资金净流入" in col_str:
                col_map[col] = "north_net"
            elif "持股名称" in col_str:
                col_map[col] = "leading_stock"
            elif "涨跌幅" in col_str and "领涨股" in col_str:
                col_map[col] = "leading_stock_change_pct"
            elif "沪深300" == col_str:
                col_map[col] = "csi300"
            elif "沪深300-涨跌幅" in col_str:
                col_map[col] = "csi300_change_pct"
            elif "代码" in col_str and "领涨" in col_str:
                col_map[col] = "leading_stock_code"
        if col_map:
            df = df.rename(columns=col_map)
        records = dataframe_to_records(df.tail(days), days)
        if not records:
            raise ValueError("northbound hist data empty")
        return {"count": len(records), "records": records}

    def _northbound_hist_direct(self, days: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://push2his.eastmoney.com/api/qt/kamt.kline/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "fields1": "f1,f2,f3,f4",
                "fields2": "f51,f52,f53,f54,f55,f56",
                "klt": "101",
                "lmt": days,
                "ut": "b2884a393a59ad64002292a3e90d46a5",
            },
        )
        data = raw.get("data") or {}

        # EastMoney API changed format: old uses "s2n", new uses "hk2sh"/"hk2sz"
        s2n = data.get("s2n")
        hk2sh = data.get("hk2sh")
        hk2sz = data.get("hk2sz")

        points = []
        if s2n:
            # Legacy format: date,north_net,sh_net,sz_net,north_buy,north_sell
            for line in s2n[-days:]:
                parts = str(line).split(",")
                if len(parts) < 4:
                    continue
                points.append({
                    "date": parts[0],
                    "north_net": self._float(parts[1]),
                    "sh_net": self._float(parts[2]),
                    "sz_net": self._float(parts[3]),
                    "north_buy": self._float(parts[4]) if len(parts) > 4 else None,
                    "north_sell": self._float(parts[5]) if len(parts) > 5 else None,
                })
        elif hk2sh or hk2sz:
            # New format: hk2sh and hk2sz each have date,field1,field2,cumulative
            # Merge by date to compute northbound net flow
            by_date: dict[str, dict[str, float]] = {}
            for line in (hk2sh or [])[-days:]:
                parts = str(line).split(",")
                if len(parts) >= 2:
                    d = parts[0]
                    by_date.setdefault(d, {})["sh_net"] = self._float(parts[1])
            for line in (hk2sz or [])[-days:]:
                parts = str(line).split(",")
                if len(parts) >= 2:
                    d = parts[0]
                    by_date.setdefault(d, {})["sz_net"] = self._float(parts[1])
            for d in sorted(by_date)[-days:]:
                vals = by_date[d]
                sh = vals.get("sh_net") or 0
                sz = vals.get("sz_net") or 0
                points.append({
                    "date": d,
                    "north_net": sh + sz if sh or sz else None,
                    "sh_net": vals.get("sh_net"),
                    "sz_net": vals.get("sz_net"),
                })
        if not points:
            raise ValueError("eastmoney northbound kline returned empty data")
        return {"count": len(points), "records": points}

    def _northbound_realtime_akshare(self) -> dict[str, Any]:
        df = self._require_df(ak.stock_hsgt_fund_min_em(symbol="北向资金"), "stock_hsgt_fund_min_em")
        records = dataframe_to_records(df, 240)
        if not records:
            raise ValueError("northbound realtime data empty")
        return {"count": len(records), "records": records}

    def _northbound_realtime_direct(self, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        raw = self._request_json(
            f"{base_url.rstrip('/')}/api/qt/kamt.rtmin/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "fields1": "f1,f2,f3,f4",
                "fields2": "f51,f52,f53,f54,f55,f56",
                "fields3": "f1,f2",
                "ut": "b2884a393a59ad64002292a3e90d46a5",
            },
            timeout=6,
        )
        data = raw.get("data") or {}
        points = []

        s2n = data.get("s2n")
        hk2sh = data.get("hk2sh")
        hk2sz = data.get("hk2sz")

        if s2n:
            for line in s2n:
                parts = str(line).split(",")
                if len(parts) < 4:
                    continue
                points.append({
                    "time": parts[0],
                    "north_net": self._float(parts[1]),
                    "sh_net": self._float(parts[2]),
                    "sz_net": self._float(parts[3]),
                    "north_buy": self._float(parts[4]) if len(parts) > 4 else None,
                })
        elif hk2sh or hk2sz:
            by_time: dict[str, dict[str, float]] = {}
            for line in (hk2sh or []):
                parts = str(line).split(",")
                if len(parts) >= 2:
                    by_time.setdefault(parts[0], {})["sh_net"] = self._float(parts[1])
            for line in (hk2sz or []):
                parts = str(line).split(",")
                if len(parts) >= 2:
                    by_time.setdefault(parts[0], {})["sz_net"] = self._float(parts[1])
            for t in sorted(by_time):
                vals = by_time[t]
                sh = vals.get("sh_net") or 0
                sz = vals.get("sz_net") or 0
                points.append({
                    "time": t,
                    "north_net": sh + sz if sh or sz else None,
                    "sh_net": vals.get("sh_net"),
                    "sz_net": vals.get("sz_net"),
                })
        if not points:
            raise ValueError("eastmoney northbound rtmin returned empty data")
        return {"count": len(points), "records": points}

    def _northbound_holdings_individual(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        df = self._require_df(
            ak.stock_hsgt_individual_em(symbol=code.code),
            "stock_hsgt_individual_em",
        )
        date_col = None
        for col in df.columns:
            col_str = str(col)
            if "日期" in col_str or "date" in col_str.lower():
                date_col = col
                break
        if date_col is not None:
            df[date_col] = pd.to_datetime(df[date_col], errors="coerce")
            df = df.dropna(subset=[date_col]).sort_values(date_col, ascending=False)
            # Check staleness: if latest date is more than 30 days old, reject
            latest = df[date_col].max()
            if pd.notna(latest) and (pd.Timestamp.now() - latest).days > 30:
                raise ValueError(f"northbound data too stale, latest: {latest.date()}")
        records = dataframe_to_records(df.head(limit), limit)
        if not records:
            raise ValueError("northbound individual stock data empty")
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _northbound_holdings_individual_direct(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://datacenter-web.eastmoney.com/api/data/v1/get",
            headers={"Referer": "https://data.eastmoney.com/"},
            params={
                "reportName": "RPT_MUTUAL_HOLDSTOCKNORTH_STA",
                "columns": "ALL",
                "filter": f'(SECURITY_CODE="{code.code}")',
                "pageNumber": 1,
                "pageSize": limit,
                "sortColumns": "TRADE_DATE",
                "sortTypes": "-1",
                "source": "WEB",
                "client": "WEB",
            },
        )
        result = raw.get("result") or {}
        rows = result.get("data") or []
        records = []
        for row in rows[:limit]:
            records.append({
                "date": str(row.get("TRADE_DATE", ""))[:10],
                "close": self._float(row.get("CLOSE_PRICE")),
                "change_pct": self._float(row.get("CHANGE_RATE")),
                "hold_shares": self._float(row.get("HOLD_SHARES")),
                "hold_market_cap": self._float(row.get("HOLD_MARKET_CAP")),
                "hold_ratio_a": self._float(row.get("HOLD_SHARES_RATIO")),
                "hold_ratio_free": self._float(row.get("FREE_SHARES_RATIO")),
            })
        if not records:
            return {"stock_code": code.display, "count": 0, "records": [], "note": "该股票不在沪深港通标的池内，无北向持股数据"}
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _northbound_holdings_direct(self, limit: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://datacenter-web.eastmoney.com/api/data/v1/get",
            headers={"Referer": "https://data.eastmoney.com/"},
            params={
                "reportName": "RPT_MUTUAL_MARKET_STA",
                "columns": "ALL",
                "pageNumber": 1,
                "pageSize": limit,
                "sortColumns": "HOLD_MARKET_CAP",
                "sortTypes": "-1",
                "source": "WEB",
                "client": "WEB",
            },
        )
        result = raw.get("result") or {}
        rows = result.get("data") or []
        records = []
        for row in rows[:limit]:
            records.append({
                "market_code": row.get("MARKET_CODE"),
                "date": str(row.get("HOLD_DATE", ""))[:10],
                "change_pct": self._float(row.get("CHANGE_RATE")),
                "add_market_cap": self._float(row.get("ADD_MARKET_CAP")),
                "add_ratio": self._float(row.get("ADD_MARKET_RATE")),
                "hold_market_cap": self._float(row.get("HOLD_MARKET_CAP")),
            })
        if not records:
            raise ValueError("eastmoney northbound market stats returned empty data")
        return {"count": len(records), "records": records}

    # ---- 申万行业 & 基金持仓 helpers ----

    def _shenwan_industry_direct(self, limit: int, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        raw = self._request_json(
            f"{base_url.rstrip('/')}/api/qt/clist/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "pn": "1",
                "pz": limit,
                "po": "1",
                "np": "1",
                "ut": "bd1d9ddb04089700cf9c27f6f7426281",
                "fltt": "2",
                "invt": "2",
                "fid": "f3",
                "fs": "m:90+t:2",
                "fields": "f12,f14,f2,f3,f9,f23,f115,f116,f117,f130,f131,f132,f133,f134",
            },
        )
        data = raw.get("data") or {}
        diff = data.get("diff") or []
        records = []
        for idx, row in enumerate(diff[:limit], 1):
            records.append({
                "rank": idx,
                "code": row.get("f12"),
                "name": row.get("f14"),
                "price": self._float(row.get("f2")),
                "change_pct": self._float(row.get("f3")),
                "pe_dynamic": self._float(row.get("f9")),
                "pb_mrq": self._float(row.get("f23")),
                "pe_min": self._float(row.get("f130")),
                "pe_max": self._float(row.get("f131")),
                "pe_median": self._float(row.get("f132")),
                "pe_percentile": self._float(row.get("f133")),
                "pb_percentile": self._float(row.get("f134")),
            })
        if not records:
            raise ValueError("eastmoney shenwan industry returned empty data")
        return {"count": len(records), "records": records}

    def _shenwan_industry_akshare(self, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.index_realtime_sw(), "index_realtime_sw")
        # AKShare 返回的列名是中文，使用位置映射
        # 标准列顺序: 指数代码, 指数名称, 昨收, 现价, 最新价, 成交额, 成交量, 最高, 最低
        records = []
        for _, row in df.head(limit).iterrows():
            vals = row.tolist()
            if len(vals) >= 9:
                records.append({
                    "code": str(vals[0]) if vals[0] else None,
                    "name": str(vals[1]) if vals[1] else None,
                    "pre_close": self._float(vals[2]),
                    "price": self._float(vals[3]) or self._float(vals[4]),
                    "change_pct": round((self._float(vals[3] or vals[4]) - self._float(vals[2])) / self._float(vals[2]) * 100, 2) if self._float(vals[2]) and (self._float(vals[3]) or self._float(vals[4])) else None,
                    "amount": self._float(vals[5]),
                    "volume": self._float(vals[6]),
                    "high": self._float(vals[7]),
                    "low": self._float(vals[8]),
                })
        if not records:
            raise ValueError("shenwan industry data empty")
        return {"count": len(records), "records": records, "fallback_note": "AKShare申万指数实时数据，无PE/PB历史分位信息"}

    def _fund_holdings_akshare(self, date: str, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.fund_report_stock_cninfo(date=date), "fund_report_stock_cninfo")
        records = dataframe_to_records(df.head(limit), limit)
        if not records:
            raise ValueError("fund holdings data empty")
        return {"date": date, "count": len(records), "records": records}

    def _fund_holdings_zlsj_direct(self, date: str, limit: int) -> dict[str, Any]:
        report_date = self._iso_date(date)
        raw = self._request_json(
            "https://data.eastmoney.com/dataapi/zlsj/list",
            headers={"Referer": "https://data.eastmoney.com/zlsj/"},
            params={
                "date": report_date,
                "type": "1",
                "zjc": "0",
                "sortField": "HOULD_NUM",
                "sortDirec": "1",
                "pageNum": "1",
                "pageSize": str(min(max(int(limit), 1), 500)),
                "p": "1",
                "pageNo": "1",
            },
        )
        rows = raw.get("data") if isinstance(raw, dict) else None
        if not isinstance(rows, list) or not rows:
            raise ValueError("eastmoney zlsj fund holdings returned empty data")

        records = []
        for idx, row in enumerate(rows[:limit], 1):
            if not isinstance(row, dict):
                continue
            secucode = str(row.get("SECUCODE") or "")
            code = str(row.get("SECURITY_CODE") or (secucode.split(".", 1)[0] if secucode else ""))
            records.append(
                {
                    "rank": idx,
                    "stock_code": secucode or code,
                    "code": code or None,
                    "name": row.get("SECURITY_NAME_ABBR"),
                    "report_date": row.get("REPORT_DATE") or report_date,
                    "holder_type": row.get("ORG_TYPE_NAME"),
                    "fund_count": self._float(row.get("HOULD_NUM")),
                    "hold_shares": self._float(row.get("TOTAL_SHARES")),
                    "hold_market_cap": self._float(row.get("HOLD_VALUE")),
                    "free_cap_ratio": self._float(row.get("FREESHARES_RATIO")),
                    "total_shares_ratio": self._float(row.get("TOTALSHARES_RATIO")),
                    "hold_change": row.get("HOLDCHA"),
                    "hold_change_shares": self._float(row.get("HOLDCHA_NUM")),
                    "hold_change_ratio": self._float(row.get("HOLDCHA_RATIO")),
                    "hold_change_value": self._float(row.get("HOLDCHA_VALUE")),
                    "quarter_change_rate": self._float(row.get("QCHANGE_RATE")),
                }
            )
        if not records:
            raise ValueError("eastmoney zlsj fund holdings records empty")
        return {
            "date": date,
            "report_date": report_date,
            "count": len(records),
            "total_pages": raw.get("pages") if isinstance(raw, dict) else None,
            "records": records,
        }

    def _fund_holdings_direct(self, date: str, limit: int) -> dict[str, Any]:
        # 将 YYYYMMDD 转换为中文报告期格式
        year = date[:4]
        month = date[4:6]
        if month in ("03", "04"):
            report_name = f"{year}年1季报"
        elif month in ("06", "07"):
            report_name = f"{year}年2季报"
        elif month in ("09", "10"):
            report_name = f"{year}年3季报"
        else:
            report_name = f"{year}年4季报"

        raw = self._request_json(
            "https://datacenter-web.eastmoney.com/api/data/v1/get",
            headers={"Referer": "https://data.eastmoney.com/"},
            params={
                "reportName": "RPT_FUND_MAIN_POSITION",
                "columns": "ALL",
                "filter": f'(REPORT_DATE_NAME="{report_name}")',
                "pageNumber": 1,
                "pageSize": limit,
                "sortColumns": "HOLD_MARKET_CAP",
                "sortTypes": "-1",
                "source": "WEB",
                "client": "WEB",
            },
        )
        result = raw.get("result") or {}
        rows = result.get("data") or []
        records = []
        for row in rows[:limit]:
            records.append({
                "stock_code": row.get("SECURITY_CODE"),
                "name": row.get("SECURITY_NAME_ABBR"),
                "industry": row.get("SECURITY_INDUSTRY"),
                "hold_market_cap": self._float(row.get("HOLD_MARKET_CAP")),
                "fund_count": self._float(row.get("FUNDS_HOLDNUM")),
                "hold_shares": self._float(row.get("HOLD_SHARES")),
                "free_cap_ratio": self._float(row.get("FREECAP_HOLD_RATIO")),
                "report_date": row.get("REPORT_DATE_NAME"),
            })
        if not records:
            raise ValueError("eastmoney fund holdings returned empty data")
        return {"report_date": report_name, "count": len(records), "total": result.get("count"), "records": records}

    def _fund_structure_direct(self) -> dict[str, Any]:
        params = {
            "dt": "11",
            "pi": "1",
            "pn": "50",
            "mc": "hypzDetail",
            "st": "desc",
            "sc": "reportdate",
        }
        text = self._request_text(
            f"https://fund.eastmoney.com/data/FundDataPortfolio_Interface.aspx?{urlencode(params)}",
            headers={"Referer": "https://fund.eastmoney.com/data/"},
        ).strip()
        if "=" not in text:
            raise ValueError("eastmoney fund structure returned malformed payload")
        body = text.split("=", 1)[1].strip().rstrip(";")
        data_marker = body.find("data:")
        if data_marker < 0:
            raise ValueError("eastmoney fund structure missing data field")
        array_start = body.find("[", data_marker)
        array_end = body.find("],record", array_start)
        if array_start < 0:
            raise ValueError("eastmoney fund structure missing data array")
        if array_end < 0:
            array_end = body.rfind("]")
        array_text = body[array_start : array_end + 1]
        rows = json.loads(array_text)

        def _meta_value(name: str) -> str | None:
            marker = f'{name}:"'
            start = body.find(marker)
            if start < 0:
                return None
            start += len(marker)
            end = body.find('"', start)
            return body[start:end] if end >= 0 else None

        records = []
        for row in rows:
            if not isinstance(row, list) or len(row) < 6:
                continue
            records.append(
                {
                    "report_date": row[0],
                    "fund_count": self._float(row[1]),
                    "institution_ratio": self._float(row[2]),
                    "individual_ratio": self._float(row[3]),
                    "internal_ratio": self._float(row[4]),
                    "total_share": self._float(row[5]),
                }
            )
        if not records:
            raise ValueError("eastmoney fund structure returned empty data")
        return {
            "count": len(records),
            "total": self._float(_meta_value("record")),
            "pages": self._float(_meta_value("pages")),
            "records": records,
        }

    def _fund_structure_akshare(self) -> dict[str, Any]:
        df = self._require_df(ak.fund_hold_structure_em(), "fund_hold_structure_em")
        records = dataframe_to_records(df, 50)
        if not records:
            raise ValueError("fund structure data empty")
        return {"count": len(records), "records": records}

    def _fund_structure_scrapling(self) -> dict[str, Any]:
        """使用 Scrapling 从东方财富网页抓取基金持仓结构数据"""
        try:
            from scrapling.fetchers import Fetcher

            # 东方财富基金持仓结构页面
            page = Fetcher.get(
                'https://data.eastmoney.com/zlsj/zlsjNew.html',
                headers={'Referer': 'https://data.eastmoney.com/'}
            )

            if page.status != 200:
                raise ValueError(f"scrapling fetch failed: HTTP {page.status}")

            # 解析表格数据
            rows = page.css('table tbody tr')
            if not rows:
                raise ValueError("scrapling found no table rows")

            records = []
            for row in rows[:50]:
                cells = row.css('td')
                if len(cells) >= 6:
                    records.append({
                        "日期": cells[0].text.strip() if cells[0].text else None,
                        "持有基金数": self._float(cells[1].text.strip() if cells[1].text else None),
                        "基金份额": self._float(cells[2].text.strip() if cells[2].text else None),
                        "机构投资者持有份额占比": self._float(cells[3].text.strip().replace('%', '') if cells[3].text else None),
                        "个人投资者持有份额占比": self._float(cells[4].text.strip().replace('%', '') if cells[4].text else None),
                        "内部持有份额占比": self._float(cells[5].text.strip().replace('%', '') if cells[5].text else None),
                        "总份额": self._float(cells[6].text.strip() if len(cells) > 6 and cells[6].text else None),
                    })

            if not records:
                raise ValueError("scrapling parsed no records")

            return {"count": len(records), "records": records, "source_note": "Scrapling网页抓取"}

        except ImportError:
            raise ValueError("scrapling not installed")
        except Exception as e:
            raise ValueError(f"scrapling failed: {e}")

    # ---- 股东户数 helpers ----

    def _shareholders_akshare(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.stock_zh_a_gdhs_detail_em(symbol=code.code), "stock_zh_a_gdhs_detail_em")
        date_col = None
        for col in df.columns:
            col_str = str(col)
            if any(kw in col_str for kw in ["截止", "统计"]):
                date_col = col
                break
        if date_col is not None:
            df[date_col] = pd.to_datetime(df[date_col], errors="coerce")
            df = df.dropna(subset=[date_col]).sort_values(date_col, ascending=False)
        records = dataframe_to_records(df.head(limit), limit)
        if not records:
            raise ValueError("shareholder count data empty")
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _shareholders_direct(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://datacenter-web.eastmoney.com/api/data/v1/get",
            headers={"Referer": "https://data.eastmoney.com/"},
            params={
                "reportName": "RPT_F10_EH_HOLDERNUM",
                "columns": "ALL",
                "filter": f'(SECURITY_CODE="{code.code}")',
                "pageNumber": 1,
                "pageSize": limit,
                "sortColumns": "END_DATE",
                "sortTypes": "-1",
                "source": "WEB",
                "client": "WEB",
            },
        )
        result = raw.get("result") or {}
        rows = result.get("data") or []
        records = []
        for row in rows[:limit]:
            records.append({
                "end_date": str(row.get("END_DATE", ""))[:10],
                "notice_date": str(row.get("NOTICE_DATE", ""))[:10],
                "shareholder_count": self._float(row.get("HOLDER_TOTAL_NUM")),
                "prev_count": self._float(row.get("HOLDER_TOTAL_NUMCHANGE")),
                "change_pct": self._float(row.get("TOTAL_NUM_RATIO")),
                "avg_shares": self._float(row.get("AVG_FREE_SHARES")),
                "avg_market_cap": self._float(row.get("AVG_HOLD_AMT")),
            })
        if not records:
            raise ValueError("eastmoney shareholder count returned empty data")
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _shareholders_f10(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        secid = f"{code.suffix.upper()}{code.code}"
        raw = self._request_json(
            "https://emweb.securities.eastmoney.com/PC_HSF10/ShareholderResearch/PageAjax",
            headers={"Referer": "https://emweb.securities.eastmoney.com/"},
            params={"code": secid},
        )
        rows = raw.get("gdrs") or []
        records = []
        for row in rows[:limit]:
            records.append({
                "end_date": str(row.get("END_DATE", ""))[:10],
                "shareholder_count": self._float(row.get("HOLDER_TOTAL_NUM")),
                "change_pct": self._float(row.get("TOTAL_NUM_RATIO")),
                "avg_shares": self._float(row.get("AVG_FREE_SHARES")),
                "avg_market_cap": self._float(row.get("AVG_HOLD_AMT")),
                "price": self._float(row.get("PRICE")),
            })
        if not records:
            raise ValueError("eastmoney f10 shareholder data empty")
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _shareholder_top_akshare(self, code: NormalizedStockCode) -> dict[str, Any]:
        df = self._require_df(ak.stock_main_stock_holder(stock=code.code), "stock_main_stock_holder")
        records = dataframe_to_records(df, 50)
        if not records:
            raise ValueError("shareholder top data empty")
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _eastmoney_f10_payload(
        self,
        endpoint_key: str,
        code: NormalizedStockCode,
        limit: int,
        start_date: str | None = None,
        end_date: str | None = None,
        allow_empty: bool = False,
    ) -> dict[str, Any]:
        rows = self._eastmoney_f10_rows(endpoint_key, code, limit, start_date, end_date)
        records = rows[:limit]
        if not records and not allow_empty:
            raise ValueError(f"eastmoney f10 {endpoint_key} returned empty data")
        return {
            "stock_code": code.display,
            "dataset": endpoint_key,
            "count": len(records),
            "records": records,
        }

    def _eastmoney_f10_rows(
        self,
        endpoint_key: str,
        code: NormalizedStockCode,
        limit: int,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> list[dict[str, Any]]:
        config = EASTMONEY_F10_ENDPOINTS.get(endpoint_key)
        if config is None:
            raise ValueError(f"unsupported eastmoney f10 endpoint: {endpoint_key}")

        api_family = str(config["api_family"])
        url = EASTMONEY_DATA_GET_URL if api_family == "data_get" else EASTMONEY_DATA_V1_GET_URL
        params = {
            **dict(config["fixed_params"]),
            "filter": self._eastmoney_f10_filter(config, code, start_date, end_date),
            "source": "HSF10",
            "client": "PC",
        }
        page_size = max(1, min(max(int(limit), 1), 500))
        sort_fields = ",".join(config["sort_fields"])
        sort_types = ",".join(config["sort_types"])
        if api_family == "data_get":
            params.update({"p": "1", "ps": str(page_size), "st": sort_fields, "sr": sort_types})
        elif api_family == "data_v1_get":
            params.update(
                {
                    "pageNumber": "1",
                    "pageSize": str(page_size),
                    "sortColumns": sort_fields,
                    "sortTypes": sort_types,
                    "quoteColumns": "",
                }
            )
        else:
            raise ValueError(f"unsupported eastmoney f10 api family: {api_family}")

        raw = self._request_json(
            url,
            headers={"Referer": "https://emweb.securities.eastmoney.com/"},
            params=params,
        )
        result = raw.get("result")
        if not isinstance(result, dict):
            return []
        data = result.get("data")
        if not isinstance(data, list):
            return []
        return [dict(row) for row in data if isinstance(row, dict)]

    def _eastmoney_f10_filter(
        self,
        config: dict[str, Any],
        code: NormalizedStockCode,
        start_date: str | None,
        end_date: str | None,
    ) -> str:
        clauses = [f'(SECUCODE="{code.display}")']
        date_field = str(config["date_field"])
        if start_date:
            clauses.append(f"({date_field}>='{start_date}')")
        if end_date:
            clauses.append(f"({date_field}<='{end_date}')")
        return "".join(clauses)

    def _income_akshare(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        import requests as _req
        # Use EastMoney datacenter API for income statement (proven to work on HF)
        url = "https://datacenter-web.eastmoney.com/api/data/v1/get"
        params = {
            "reportName": "RPT_DMSK_FN_INCOME",
            "columns": "ALL",
            "filter": f'(SECURITY_CODE="{code.code}")',
            "pageNumber": 1,
            "pageSize": limit,
            "sortColumns": "REPORT_DATE",
            "sortTypes": -1,
            "source": "WEB",
            "client": "WEB",
        }
        resp = _req.get(url, params=params, headers=HTTP_HEADERS, timeout=15)
        resp.raise_for_status()
        body = resp.json()
        if not body.get("result") or not body["result"].get("data"):
            raise ValueError(f"income statement data empty for {code.code}")
        raw = body["result"]["data"]
        records = []
        for r in raw[:limit]:
            records.append({
                "report_date": r.get("REPORT_DATE", "")[:10],
                "name": r.get("SECURITY_NAME_ABBR", ""),
                "total_revenue": r.get("TOTAL_OPERATE_INCOME"),
                "revenue": r.get("OPERATE_INCOME"),
                "operating_cost": r.get("OPERATE_COST"),
                "net_profit": r.get("NETPROFIT"),
                "parent_net_profit": r.get("PARENT_NETPROFIT"),
                "basic_eps": r.get("BASIC_EPS"),
                "diluted_eps": r.get("DILUTED_EPS"),
                "rd_expense": r.get("RESEARCH_EXPENSE"),
                "finance_expense": r.get("FINANCE_EXPENSE"),
            })
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _balancesheet_akshare(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        import akshare as ak
        prefixed = code.suffix + code.code  # e.g. "SH600519"
        df = self._require_df(ak.stock_balance_sheet_by_report_em(symbol=prefixed), "stock_balance_sheet_by_report_em")
        cols = [c for c in [
            "REPORT_DATE", "SECURITY_NAME_ABBR",
            "TOTAL_ASSETS", "TOTAL_LIABILITIES", "TOTAL_EQUITY",
            "TOTAL_PARENT_EQUITY", "MINORITY_EQUITY",
            "MONETARYFUNDS", "ACCOUNTS_RECE", "INVENTORY",
            "FIXED_ASSET", "TOTAL_CURRENT_ASSETS", "TOTAL_NONCURRENT_ASSETS",
            "TOTAL_CURRENT_LIAB", "TOTAL_NONCURRENT_LIAB",
        ] if c in df.columns]
        df = df[cols].head(limit)
        records = dataframe_to_records(df, 200)
        if not records:
            raise ValueError("balance sheet data empty")
        return {"stock_code": code.display, "count": len(records), "records": records}

    def _daily_basic_lg(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        import requests as _req
        secid = f"{1 if code.market == 'sh' else 0}.{code.code}"
        params = {
            "secid": secid,
            "fields": "f57,f58,f43,f169,f170,f46,f44,f45,f168,f116,f117,f162,f163,f164,f167,f127",
            "ut": "fa5fd1943c7b386f172d6893dbbd1035",
        }
        last_err = None
        for host in ("push2delay.eastmoney.com", "push2.eastmoney.com"):
            try:
                url = f"https://{host}/api/qt/stock/get"
                resp = _req.get(url, params=params, headers=HTTP_HEADERS, timeout=10)
                resp.raise_for_status()
                d = resp.json().get("data")
                if d:
                    def _v(key, div=100):
                        val = d.get(key)
                        return round(val / div, 2) if val is not None and val != "-" else None
                    return {
                        "stock_code": code.display,
                        "count": 1,
                        "records": [{
                            "date": d.get("f57", ""),
                            "name": d.get("f58", ""),
                            "price": _v("f43"),
                            "pe_dynamic": _v("f162"),
                            "pe_ttm": _v("f163"),
                            "pb": _v("f167"),
                            "total_mv": _v("f116", div=1),
                            "circ_mv": _v("f117", div=1),
                            "change_pct": _v("f170"),
                        }],
                    }
            except Exception as e:
                last_err = e
                continue
        raise ValueError(f"daily basic data unavailable: {last_err}")

    def _daily_basic_spot(self, code: NormalizedStockCode) -> dict[str, Any]:
        import akshare as ak
        df = self._require_df(ak.stock_zh_a_spot_em(), "stock_zh_a_spot_em")
        row = df[df["代码"] == code.code]
        if row.empty:
            raise ValueError(f"stock {code.code} not found in spot data")
        r = row.iloc[0]
        return {
            "stock_code": code.display,
            "name": str(r.get("名称", "")),
            "price": float(r.get("最新价", 0)) if r.get("最新价") is not None else None,
            "pe_dynamic": float(r.get("市盈率-动态", 0)) if r.get("市盈率-动态") is not None else None,
            "pb": float(r.get("市净率", 0)) if r.get("市净率") is not None else None,
            "total_mv": float(r.get("总市值", 0)) if r.get("总市值") is not None else None,
            "circ_mv": float(r.get("流通市值", 0)) if r.get("流通市值") is not None else None,
            "turnover": float(r.get("换手率", 0)) if r.get("换手率") is not None else None,
            "change_pct": float(r.get("涨跌幅", 0)) if r.get("涨跌幅") is not None else None,
        }

    def _prev_trade_date(self) -> str:
        now = self._now_cn()
        offset = 3 if now.weekday() == 0 else 1
        prev = now - timedelta(days=offset)
        return prev.strftime("%Y%m%d")

    def _report_date(self, value: str | None) -> str:
        if value:
            return self._compact_date(value)
        now = self._now_cn()
        quarters = [(3, 31), (6, 30), (9, 30), (12, 31)]
        for month, day in reversed(quarters):
            if (now.month, now.day) >= (month, day):
                return f"{now.year}{month:02d}{day:02d}"
        return f"{now.year - 1}1231"

    def _stock_daily_hist_df(self, code: NormalizedStockCode, start: str, end: str, adjust: str) -> pd.DataFrame:
        df = ak.stock_zh_a_hist(symbol=code.code, period="daily", start_date=start, end_date=end, adjust=adjust or "")
        return self._require_df(df, "stock_zh_a_hist")

    def _concept_flow_df(self) -> pd.DataFrame:
        return self._require_df(ak.stock_fund_flow_concept(symbol="即时"), "stock_fund_flow_concept")

    def _industry_flow_df(self) -> pd.DataFrame:
        return self._require_df(ak.stock_fund_flow_industry(symbol="即时"), "stock_fund_flow_industry")

    def _stock_daily_sina_df(self, code: NormalizedStockCode, start: str, end: str, adjust: str) -> pd.DataFrame:
        df = ak.stock_zh_a_daily(symbol=code.prefixed, start_date=start, end_date=end, adjust=adjust or "")
        return self._require_df(df, "stock_zh_a_daily")

    def _stock_daily_eastmoney_direct_payload(
        self,
        code: NormalizedStockCode,
        start: str,
        end: str,
        adjust: str,
        days: int,
        base_url: str = "https://push2his.eastmoney.com",
    ) -> dict[str, Any]:
        fqt = {"qfq": "1", "hfq": "2", "": "0"}.get(adjust or "", "1")
        raw = self._request_json(
            f"{base_url.rstrip('/')}/api/qt/stock/kline/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "secid": self._stock_secid(code),
                "fields1": "f1,f2,f3,f4,f5,f6",
                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61",
                "klt": "101",
                "fqt": fqt,
                "beg": start,
                "end": end,
                "lmt": max(days * 3, days + 20),
                "ut": "fa5fd1943c7b386f172d6893dbbd1035",
            },
            timeout=6,
        )
        data = raw.get("data") or {}
        records = []
        for line in data.get("klines") or []:
            parts = str(line).split(",")
            if len(parts) < 11:
                continue
            records.append(
                {
                    "date": parts[0],
                    "open": self._float(parts[1]),
                    "close": self._float(parts[2]),
                    "high": self._float(parts[3]),
                    "low": self._float(parts[4]),
                    "volume": self._float(parts[5]),
                    "amount": self._float(parts[6]),
                    "amplitude": self._float(parts[7]),
                    "change_pct": self._float(parts[8]),
                    "change_amount": self._float(parts[9]),
                    "turnover": self._float(parts[10]),
                }
            )
        if not records:
            raise ValueError(f"eastmoney stock kline returned empty data: {code.display}")
        records = records[-days:]
        return {"stock_code": code.display, "days": len(records), "records": records}

    def _stock_daily_baostock_payload(
        self,
        code: NormalizedStockCode,
        start: str,
        end: str,
        adjust: str,
        days: int,
    ) -> dict[str, Any]:
        import baostock as bs

        adjustflag = {"qfq": "2", "hfq": "1", "": "3"}.get(adjust or "", "3")
        login = bs.login()
        if login.error_code != "0":
            raise ValueError(f"baostock login failed: {login.error_msg}")
        try:
            rs = bs.query_history_k_data_plus(
                f"{code.market}.{code.code}",
                "date,open,high,low,close,preclose,volume,amount,turn,pctChg",
                start_date=self._iso_date(start),
                end_date=self._iso_date(end),
                frequency="d",
                adjustflag=adjustflag,
            )
            rows: list[list[str]] = []
            while rs.error_code == "0" and rs.next():
                rows.append(rs.get_row_data())
            if rs.error_code != "0":
                raise ValueError(f"baostock query failed: {rs.error_msg}")
            if not rows:
                raise ValueError("baostock returned empty data")
            df = pd.DataFrame(rows, columns=rs.fields)
        finally:
            bs.logout()

        records: list[dict[str, Any]] = []
        for _, row in df.tail(days).iterrows():
            close = self._float(row.get("close"))
            pre_close = self._float(row.get("preclose"))
            change_amount = (close - pre_close) if close is not None and pre_close is not None else None
            records.append(
                {
                    "date": row.get("date"),
                    "open": self._float(row.get("open")),
                    "high": self._float(row.get("high")),
                    "low": self._float(row.get("low")),
                    "close": close,
                    "pre_close": pre_close,
                    "change_amount": round(change_amount, 6) if change_amount is not None else None,
                    "change_pct": self._float(row.get("pctChg")),
                    "volume": self._float(row.get("volume")),
                    "amount": self._float(row.get("amount")),
                    "turnover": self._float(row.get("turn")),
                }
            )
        return {"stock_code": code.display, "days": len(records), "records": records}

    def _daily_payload_from_df(self, code: NormalizedStockCode, df: pd.DataFrame, days: int, source_style: str) -> dict[str, Any]:
        records = self._normalize_daily_records(df, source_style)
        records = records[-days:]
        return {
            "stock_code": code.display,
            "days": len(records),
            "records": records,
        }

    def _stock_quote_from_spot_em(self, code: NormalizedStockCode) -> dict[str, Any]:
        df = self._require_df(ak.stock_zh_a_spot_em(), "stock_zh_a_spot_em")
        row_df = df[df["代码"].astype(str) == code.code]
        if row_df.empty:
            raise ValueError(f"stock not found in spot list: {code.display}")
        row = row_df.iloc[0]
        return {
            "stock_code": code.display,
            "name": row.get("名称"),
            "price": self._float(row.get("最新价")),
            "change_pct": self._float(row.get("涨跌幅")),
            "change_amount": self._float(row.get("涨跌额")),
            "volume": self._float(row.get("成交量")),
            "amount": self._float(row.get("成交额")),
            "open": self._float(row.get("今开")),
            "high": self._float(row.get("最高")),
            "low": self._float(row.get("最低")),
            "pre_close": self._float(row.get("昨收")),
            "turnover": self._float(row.get("换手率")),
            "date": self._now_cn().date().isoformat(),
        }

    def _sina_stock_quote_payload(self, code: NormalizedStockCode, include_order_book: bool) -> dict[str, Any]:
        text = self._request_text(
            f"http://hq.sinajs.cn/rn={int(self._now_cn().timestamp() * 1000)}&list={code.prefixed}",
            headers={"Referer": "http://finance.sina.com.cn"},
            encoding="gbk",
        )
        if '=""' in text:
            raise ValueError(f"sina hq returned empty quote: {code.display}")
        value = text.split("=", 1)[1].strip().strip(";").strip('"')
        fields = value.split(",")
        if len(fields) < 32:
            raise ValueError(f"sina hq returned malformed quote: {code.display}")
        pre_close = self._float(fields[2])
        price = self._float(fields[3])
        change_amount = (price - pre_close) if price is not None and pre_close else None
        payload: dict[str, Any] = {
            "stock_code": code.display,
            "name": fields[0],
            "price": price,
            "change_pct": round(change_amount / pre_close * 100, 4) if change_amount is not None and pre_close else None,
            "change_amount": round(change_amount, 4) if change_amount is not None else None,
            "volume": self._float(fields[8]),
            "amount": self._float(fields[9]),
            "open": self._float(fields[1]),
            "high": self._float(fields[4]),
            "low": self._float(fields[5]),
            "pre_close": pre_close,
            "date": fields[30],
            "time": fields[31],
        }
        if include_order_book:
            bids = []
            asks = []
            for i in range(5):
                bid_base = 10 + i * 2
                ask_base = 20 + i * 2
                bids.append({"level": i + 1, "price": self._float(fields[bid_base + 1]), "volume": self._float(fields[bid_base])})
                asks.append({"level": i + 1, "price": self._float(fields[ask_base + 1]), "volume": self._float(fields[ask_base])})
            payload["order_book"] = {"bids": bids, "asks": asks}
        return payload

    def _stock_quote_from_daily(self, code: NormalizedStockCode) -> dict[str, Any]:
        start, end = self._date_range(8, None, None)
        df = self._stock_daily_hist_df(code, start, end, "qfq")
        return self._quote_from_daily_df(code, df, "eastmoney")

    def _stock_quote_from_sina_daily(self, code: NormalizedStockCode) -> dict[str, Any]:
        start, end = self._date_range(8, None, None)
        df = self._stock_daily_sina_df(code, start, end, "qfq")
        return self._quote_from_daily_df(code, df, "sina")

    def _quote_from_daily_df(self, code: NormalizedStockCode, df: pd.DataFrame, source_style: str) -> dict[str, Any]:
        records = self._normalize_daily_records(df, source_style)
        if not records:
            raise ValueError("daily data is empty")
        latest = records[-1]
        prev = records[-2] if len(records) >= 2 else None
        pre_close = prev["close"] if prev else latest["open"]
        change_amount = (latest["close"] or 0) - (pre_close or 0)
        change_pct = change_amount / pre_close * 100 if pre_close else None
        return {
            "stock_code": code.display,
            "name": code.code,
            "price": latest["close"],
            "change_pct": round(change_pct, 4) if change_pct is not None else None,
            "change_amount": round(change_amount, 4),
            "volume": latest["volume"],
            "amount": latest.get("amount"),
            "open": latest["open"],
            "high": latest["high"],
            "low": latest["low"],
            "pre_close": pre_close,
            "turnover": latest.get("turnover"),
            "date": latest["date"],
        }

    def _normalize_daily_records(self, df: pd.DataFrame, source_style: str) -> list[dict[str, Any]]:
        work = df.copy()
        if source_style == "eastmoney":
            mapping = {
                "日期": "date",
                "开盘": "open",
                "最高": "high",
                "最低": "low",
                "收盘": "close",
                "成交量": "volume",
                "成交额": "amount",
                "振幅": "amplitude",
                "涨跌幅": "change_pct",
                "涨跌额": "change_amount",
                "换手率": "turnover",
            }
            work = work.rename(columns=mapping)
        work["date"] = pd.to_datetime(work["date"], errors="coerce")
        work = work.dropna(subset=["date"]).sort_values("date")
        keep = ["date", "open", "high", "low", "close", "volume", "amount", "turnover", "change_pct", "change_amount", "amplitude"]
        records: list[dict[str, Any]] = []
        for _, row in work.iterrows():
            item = {}
            for col in keep:
                if col not in work.columns:
                    continue
                value = row.get(col)
                item[col] = self._float(value) if col != "date" else pd.to_datetime(value).date().isoformat()
            records.append(item)
        return records

    def _technical_payload(self, code: NormalizedStockCode, df: pd.DataFrame, days: int, history_days: int) -> dict[str, Any]:
        records = self._normalize_daily_records(df, "eastmoney" if "日期" in df.columns else "sina")
        if len(records) < 5:
            raise ValueError("not enough daily data for technical indicators")
        norm_df = pd.DataFrame(records).tail(days).reset_index(drop=True)
        latest = self._compute_technical_indicators(norm_df)
        payload: dict[str, Any] = {
            "stock_code": code.display,
            "days": len(norm_df),
            "latest": latest,
        }
        if history_days > 0:
            payload["history"] = self._technical_history(norm_df, history_days)
        return payload

    def _compute_technical_indicators(self, df: pd.DataFrame) -> dict[str, Any]:
        close = df["close"].astype(float)
        high = df["high"].astype(float)
        low = df["low"].astype(float)
        volume = df["volume"].astype(float)
        result: dict[str, Any] = {}

        for period in [5, 10, 20, 60, 120]:
            result[f"MA{period}"] = self._last(close.rolling(period).mean()) if len(close) >= period else None
        for period in [5, 10, 20, 60]:
            result[f"EMA{period}"] = self._last(close.ewm(span=period, adjust=False).mean()) if len(close) >= period else None

        if len(close) >= 15:
            delta = close.diff()
            gain = delta.where(delta > 0, 0).rolling(14).mean()
            loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
            rs = gain / loss.replace(0, np.nan)
            rsi = 100 - (100 / (1 + rs))
            result["RSI"] = self._last(rsi)
        else:
            result["RSI"] = None

        if len(close) >= 9:
            low_9 = low.rolling(9).min()
            high_9 = high.rolling(9).max()
            rsv = (close - low_9) / (high_9 - low_9).replace(0, np.nan) * 100
            k = rsv.ewm(com=2, adjust=False).mean()
            d = k.ewm(com=2, adjust=False).mean()
            j = 3 * k - 2 * d
            result["KDJ_K"] = self._last(k)
            result["KDJ_D"] = self._last(d)
            result["KDJ_J"] = self._last(j)
        else:
            result["KDJ_K"] = result["KDJ_D"] = result["KDJ_J"] = None

        if len(close) >= 26:
            ema12 = close.ewm(span=12, adjust=False).mean()
            ema26 = close.ewm(span=26, adjust=False).mean()
            diff = ema12 - ema26
            dea = diff.ewm(span=9, adjust=False).mean()
            macd = 2 * (diff - dea)
            result["DIFF"] = self._last(diff)
            result["DEA"] = self._last(dea)
            result["MACD"] = self._last(macd)
        else:
            result["DIFF"] = result["DEA"] = result["MACD"] = None

        if len(close) >= 20:
            ma20 = close.rolling(20).mean()
            std20 = close.rolling(20).std()
            result["BollUp"] = self._last(ma20 + 2 * std20)
            result["BollDown"] = self._last(ma20 - 2 * std20)
        else:
            result["BollUp"] = result["BollDown"] = None

        result["OBV"] = self._last((np.sign(close.diff()) * volume).fillna(0).cumsum()) if len(close) >= 2 else None
        if len(close) >= 15:
            tr = pd.concat([high - low, (high - close.shift()).abs(), (low - close.shift()).abs()], axis=1).max(axis=1)
            result["ATR14"] = self._last(tr.rolling(14).mean())
        else:
            result["ATR14"] = None

        for period in [5, 10, 20]:
            if len(close) >= period:
                ma = close.rolling(period).mean()
                result[f"BIAS{period}"] = self._last((close - ma) / ma * 100)
            else:
                result[f"BIAS{period}"] = None

        if len(close) >= 14:
            tp = (high + low + close) / 3
            ma_tp = tp.rolling(14).mean()
            md = tp.rolling(14).apply(lambda x: np.abs(x - x.mean()).mean())
            result["CCI"] = self._last((tp - ma_tp) / (0.015 * md))
        else:
            result["CCI"] = None

        if "amount" in df.columns and len(df) >= 1:
            amount = df["amount"].astype(float)
            result["VWAP"] = self._last(amount / volume.replace(0, np.nan))
        else:
            result["VWAP"] = None

        return to_jsonable({k: (round(v, 6) if isinstance(v, float) else v) for k, v in result.items()})

    def _technical_history(self, df: pd.DataFrame, days: int) -> list[dict[str, Any]]:
        history: list[dict[str, Any]] = []
        for i in range(max(5, len(df) - days + 1), len(df) + 1):
            window = df.iloc[:i]
            indicators = self._compute_technical_indicators(window)
            history.append(
                {
                    "date": str(window.iloc[-1]["date"]),
                    "RSI": indicators.get("RSI"),
                    "MACD": indicators.get("MACD"),
                    "DIFF": indicators.get("DIFF"),
                    "DEA": indicators.get("DEA"),
                    "KDJ_K": indicators.get("KDJ_K"),
                    "KDJ_D": indicators.get("KDJ_D"),
                    "KDJ_J": indicators.get("KDJ_J"),
                    "OBV": indicators.get("OBV"),
                }
            )
        return history[-days:]

    def _chip_payload(self, code: NormalizedStockCode, df: pd.DataFrame, lookback_days: int) -> dict[str, Any]:
        records = self._normalize_daily_records(df, "eastmoney" if "日期" in df.columns else "sina")[-lookback_days:]
        if not records:
            raise ValueError("daily data is empty")
        work = pd.DataFrame(records)
        close = work["close"].astype(float)
        volume = work["volume"].astype(float)
        current_close = float(close.iloc[-1])
        total_vol = float(volume.sum()) if volume.sum() > 0 else 1.0
        avg_cost = float((close * volume).sum() / total_vol)
        profitable_vol = float(volume[close <= current_close].sum())
        profitability_ratio = profitable_vol / total_vol * 100.0
        price_range = float(close.max() - close.min())
        dispersion = price_range / max(1e-6, abs(avg_cost))
        concentration = max(0.0, min(100.0, 100.0 * (1.0 - min(1.0, dispersion))))
        clusters: list[dict[str, Any]] = []
        if float(close.max()) > float(close.min()):
            weights, edges = np.histogram(close, bins=10, range=(float(close.min()), float(close.max())), weights=volume)
            total_w = float(weights.sum()) if weights.sum() > 0 else 1.0
            top_idx = np.argsort(weights)[-3:][::-1]
            for idx in top_idx:
                clusters.append(
                    {
                        "range": [round(float(edges[idx]), 2), round(float(edges[idx + 1]), 2)],
                        "share": round(float(weights[idx] / total_w * 100.0), 2),
                    }
                )
        return {
            "stock_code": code.display,
            "lookback_days": len(records),
            "current_close": round(current_close, 2),
            "avg_cost": round(avg_cost, 2),
            "profitability_ratio": round(profitability_ratio, 2),
            "concentration_score": round(concentration, 2),
            "clusters": clusters,
        }

    def _chip_cyq_em(self, code: NormalizedStockCode, adjust: str) -> dict[str, Any]:
        """筹码分布精算版 — 用 AKShare 日 K + 本地 CYQ 算法（绕过 push2 封锁）"""
        # Step 1: Get 210 days of K-line via AKShare (proven to work on HF)
        start, end = self._date_range(210, None, None)
        try:
            df = self._stock_daily_hist_df(code, start, end, adjust)
        except Exception:
            df = self._stock_daily_sina_df(code, start, end, adjust)

        kdata = []
        for _, row in df.iterrows():
            try:
                date_val = str(row.get("date", row.get("日期", "")))[:10]
                o = float(row.get("open", row.get("开盘", 0)))
                c = float(row.get("close", row.get("收盘", 0)))
                h = float(row.get("high", row.get("最高", 0)))
                l = float(row.get("low", row.get("最低", 0)))
                v = float(row.get("volume", row.get("成交量", 0)))
                hsl_raw = row.get("turnover", row.get("换手率", None))
                hsl = float(hsl_raw) if hsl_raw is not None and str(hsl_raw).strip() else 0
                kdata.append({"date": date_val, "open": o, "close": c, "high": h, "low": l, "volume": v, "hsl": hsl})
            except (ValueError, TypeError):
                continue
        if len(kdata) < 10:
            raise ValueError("chip cyq insufficient kline data")

        # Estimate turnover if not available from data source
        if all(r["hsl"] == 0 for r in kdata):
            avg_vol = sum(r["volume"] for r in kdata) / len(kdata)
            for r in kdata:
                r["hsl"] = (r["volume"] / avg_vol * 5) if avg_vol > 0 else 3

        # Step 2: Run CYQ algorithm (EastMoney's chip distribution model)
        factor = 150
        rng = 120
        results = []
        for idx in range(len(kdata)):
            win_start = max(0, idx - rng + 1)
            window = kdata[win_start:idx + 1]

            maxprice = max(r["high"] for r in window)
            minprice = min(r["low"] for r in window)
            if maxprice <= minprice:
                maxprice = minprice + 0.01
            accuracy = max(0.01, (maxprice - minprice) / (factor - 1))

            xdata = [0.0] * factor
            for k in window:
                o, c, h, l = k["open"], k["close"], k["high"], k["low"]
                avg = (o + c + h + l) / 4.0
                turnover = min(1.0, k["hsl"] / 100.0)
                H = min(int((h - minprice) / accuracy), factor - 1)
                L_raw = (l - minprice) / accuracy
                L = max(0, int(L_raw) if L_raw == int(L_raw) else int(L_raw) + 1)
                L = min(L, H, factor - 1)

                for n in range(factor):
                    xdata[n] *= (1 - turnover)

                if h == l:
                    mid = max(0, min(int((avg - minprice) / accuracy), factor - 1))
                    xdata[mid] += (factor - 1) * turnover / 2.0
                else:
                    gpoint = 2.0 / (h - l)
                    for j in range(L, H + 1):
                        curprice = minprice + accuracy * j
                        if curprice <= avg:
                            xdata[j] += ((curprice - l) / (avg - l) * gpoint * turnover) if abs(avg - l) > 1e-8 else gpoint * turnover
                        else:
                            xdata[j] += ((h - curprice) / (h - avg) * gpoint * turnover) if abs(h - avg) > 1e-8 else gpoint * turnover

            current_price = kdata[idx]["close"]
            total = sum(xdata)
            if total <= 0:
                continue

            below = sum(xdata[i] for i in range(factor) if current_price >= minprice + i * accuracy)
            benefit = below / total

            half = total * 0.5
            s = 0.0
            avg_cost = minprice
            for i in range(factor):
                s += xdata[i]
                if s >= half:
                    avg_cost = minprice + i * accuracy
                    break

            def pct_range(pct):
                lo_t = total * (1 - pct) / 2
                hi_t = total * (1 + pct) / 2
                lo_v = hi_v = 0.0
                s2 = 0.0
                for i in range(factor):
                    s2 += xdata[i]
                    if s2 >= lo_t and lo_v == 0:
                        lo_v = minprice + i * accuracy
                    if s2 >= hi_t:
                        hi_v = minprice + i * accuracy
                        break
                conc = (hi_v - lo_v) / (hi_v + lo_v) if (hi_v + lo_v) > 0 else 0
                return round(lo_v, 2), round(hi_v, 2), round(conc, 4)

            lo70, hi70, con70 = pct_range(0.7)
            lo90, hi90, con90 = pct_range(0.9)

            results.append({
                "date": kdata[idx]["date"],
                "close": round(current_price, 2),
                "benefit_ratio": round(benefit, 4),
                "avg_cost": round(avg_cost, 2),
                "cost_70_low": lo70, "cost_70_high": hi70, "concentration_70": con70,
                "cost_90_low": lo90, "cost_90_high": hi90, "concentration_90": con90,
            })

        if not results:
            raise ValueError("chip cyq computation empty")
        return {"stock_code": code.display, "count": len(results[-90:]), "records": results[-90:]}

    def _stock_fund_flow_payload(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        df = self._require_df(ak.stock_individual_fund_flow(stock=code.code, market=code.market), "stock_individual_fund_flow")
        if "日期" in df.columns:
            df["日期"] = pd.to_datetime(df["日期"], errors="coerce")
            df = df.dropna(subset=["日期"]).sort_values("日期", ascending=False)
        recent = df.head(days)
        daily = dataframe_to_records(recent)
        min_required = 1
        if len(daily) < min_required:
            raise ValueError(f"stock_individual_fund_flow returned only {len(daily)} days, required at least {min_required}")
        total_main = self._sum_col(recent, "主力净流入-净额")
        total_super = self._sum_col(recent, "超大单净流入-净额")
        total_large = self._sum_col(recent, "大单净流入-净额")
        total_medium = self._sum_col(recent, "中单净流入-净额")
        total_small = self._sum_col(recent, "小单净流入-净额")
        return {
            "stock_code": code.display,
            "days": len(daily),
            "summary": {
                "main_force_net": round(total_main, 2),
                "super_large_net": round(total_super, 2),
                "large_net": round(total_large, 2),
                "medium_net": round(total_medium, 2),
                "small_net": round(total_small, 2),
                "main_trend": "净流入" if total_main > 0 else "净流出",
            },
            "daily": daily,
        }

    def _stock_fund_flow_direct_payload(self, code: NormalizedStockCode, days: int, base_url: str = "https://push2his.eastmoney.com") -> dict[str, Any]:
        base_url = str(base_url or "https://push2his.eastmoney.com").rstrip("/")
        raw = self._request_json(
            f"{base_url}/api/qt/stock/fflow/daykline/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "lmt": days,
                "klt": "101",
                "secid": self._stock_secid(code),
                "fields1": "f1,f2,f3,f7",
                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63",
                "ut": "b2884a393a59ad64002292a3e90d46a5",
            },
        )
        data = raw.get("data") or {}
        klines = data.get("klines") or []
        rows = []
        for line in klines[-days:]:
            parts = str(line).split(",")
            if len(parts) < 6:
                continue
            rows.append(
                {
                    "date": parts[0],
                    "main_force_net": self._float(parts[1]),
                    "super_large_net": self._float(parts[2]),
                    "large_net": self._float(parts[3]),
                    "medium_net": self._float(parts[4]),
                    "small_net": self._float(parts[5]),
                }
            )
        if not rows:
            raise ValueError("eastmoney stock fund flow returned empty data")
        return {
            "stock_code": code.display,
            "days": len(rows),
            "summary": {
                "main_force_net": round(sum((row.get("main_force_net") or 0) for row in rows), 2),
                "super_large_net": round(sum((row.get("super_large_net") or 0) for row in rows), 2),
                "large_net": round(sum((row.get("large_net") or 0) for row in rows), 2),
                "medium_net": round(sum((row.get("medium_net") or 0) for row in rows), 2),
                "small_net": round(sum((row.get("small_net") or 0) for row in rows), 2),
            },
            "daily": rows,
        }

    def _stock_fund_flow_ths_payload(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        """从同花顺网页抓取个股资金流数据（约30天），不依赖东方财富API"""
        try:
            from scrapling.fetchers import Fetcher

            page = Fetcher.get(
                f"https://stockpage.10jqka.com.cn/{code.code}/funds/",
                headers={
                    "Referer": "https://stockpage.10jqka.com.cn/",
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
                },
                timeout=10, retries=1, stealthy_headers=True,
            )
            if page.status != 200:
                raise ValueError(f"THS HTTP {page.status}")

            tables = page.css("table")
            data_table = None
            for table in tables:
                rows = table.css("tr")
                if len(rows) > 10:
                    data_table = table
                    break
            if data_table is None:
                raise ValueError("THS fund flow table not found")

            rows = data_table.css("tr")
            records = []
            for row in rows[1:]:
                cells = row.css("td")
                if len(cells) < 4:
                    continue
                date_str = (cells[0].text or "").strip()
                if not date_str or len(date_str) < 8:
                    continue
                try:
                    # THS amounts are in 万元, convert to 元
                    flow_val = float((cells[3].text or "0").strip().replace(",", "")) * 10000
                    main_val = float((cells[5].text or "0").strip().replace(",", "")) * 10000
                    medium_val = float((cells[7].text or "0").strip().replace(",", "")) * 10000
                    small_val = float((cells[9].text or "0").strip().replace(",", "")) * 10000
                    large_val = main_val - medium_val  # THS doesn't separate large from main
                except (ValueError, IndexError):
                    continue
                records.append({
                    "date": date_str,
                    "main_force_net": round(main_val, 2),
                    "super_large_net": 0,  # THS doesn't separate super-large
                    "large_net": round(large_val, 2),
                    "medium_net": round(medium_val, 2),
                    "small_net": round(small_val, 2),
                })
            if not records:
                raise ValueError("THS fund flow no valid records")
            records = records[:days]
            return {
                "stock_code": code.display,
                "days": len(records),
                "summary": {
                    "main_force_net": round(sum(r["main_force_net"] for r in records), 2),
                    "super_large_net": 0,
                    "large_net": round(sum(r["large_net"] for r in records), 2),
                    "medium_net": round(sum(r["medium_net"] for r in records), 2),
                    "small_net": round(sum(r["small_net"] for r in records), 2),
                },
                "daily": records,
            }
        except ImportError:
            raise ValueError("scrapling not installed")
        except Exception as e:
            raise ValueError(f"THS fund flow failed: {e}")

    def _stock_fund_flow_scrapling_payload(self, code: NormalizedStockCode, days: int) -> dict[str, Any]:
        """使用 Scrapling 抓取东方财富个股资金流数据，绕过 HF IP 限制"""
        try:
            from scrapling.fetchers import Fetcher
            import os as _os

            proxy_url = _os.getenv("FUND_FLOW_PROXY_URL", "").strip().rstrip("/")
            if proxy_url:
                base_urls = [proxy_url]
            else:
                base_urls = [
                    "https://push2his.eastmoney.com",
                    "https://push2.eastmoney.com",
                    "https://push2delay.eastmoney.com",
                ]
            last_error = None
            for base_url in base_urls:
                try:
                    page = Fetcher.get(
                        f"{base_url}/api/qt/stock/fflow/daykline/get",
                        timeout=5,
                        retries=1,
                        params={
                            "lmt": str(days),
                            "klt": "101",
                            "secid": self._stock_secid(code),
                            "fields1": "f1,f2,f3,f7",
                            "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63",
                            "ut": "b2884a393a59ad64002292a3e90d46a5",
                        },
                        headers={"Referer": "https://quote.eastmoney.com/"},
                        stealthy_headers=True,
                    )
                    if page.status != 200:
                        last_error = f"HTTP {page.status}"
                        continue
                    body = page.body
                    if not body:
                        last_error = "empty body"
                        continue
                    import json as _json
                    raw = _json.loads(body.decode("utf-8", errors="replace"))
                    data = raw.get("data") or {}
                    klines = data.get("klines") or []
                    rows = []
                    for line in klines[-days:]:
                        parts = str(line).split(",")
                        if len(parts) < 6:
                            continue
                        rows.append({
                            "date": parts[0],
                            "main_force_net": self._float(parts[1]),
                            "super_large_net": self._float(parts[2]),
                            "large_net": self._float(parts[3]),
                            "medium_net": self._float(parts[4]),
                            "small_net": self._float(parts[5]),
                        })
                    if not rows:
                        last_error = "empty klines"
                        continue
                    return {
                        "stock_code": code.display,
                        "days": len(rows),
                        "summary": {
                            "main_force_net": round(sum((r.get("main_force_net") or 0) for r in rows), 2),
                            "super_large_net": round(sum((r.get("super_large_net") or 0) for r in rows), 2),
                            "large_net": round(sum((r.get("large_net") or 0) for r in rows), 2),
                            "medium_net": round(sum((r.get("medium_net") or 0) for r in rows), 2),
                            "small_net": round(sum((r.get("small_net") or 0) for r in rows), 2),
                        },
                        "daily": rows,
                    }
                except Exception as e:
                    last_error = str(e)
                    continue
            raise ValueError(f"scrapling fund flow all sources failed: {last_error}")
        except ImportError:
            raise ValueError("scrapling not installed")

    def _records_payload(self, df: pd.DataFrame, limit: int, key: str) -> dict[str, Any]:
        df = self._require_df(df, key)
        records = dataframe_to_records(df, limit)
        return {"count": len(records), key: records}

    def _macro_records_payload(self, df: pd.DataFrame, limit: int) -> dict[str, Any]:
        """Macro data payload: sort by date descending so newest comes first."""
        df = self._require_df(df, "records")
        # Find the date column by checking each column for date-like values
        for col in df.columns:
            sample = df[col].dropna().head(3)
            if sample.empty:
                continue
            first_val = str(sample.iloc[0])
            # Match standard dates (2024-01-20) or Chinese month format (2026年05月)
            if "20" in first_val and ("-" in first_val or "年" in first_val):
                try:
                    df = df.sort_values(col, ascending=False)
                except Exception:
                    pass
                break
        records = dataframe_to_records(df, limit)
        return {"count": len(records), "records": records}

    def _jin10_macro_payload(self, attr_id: str, symbol: str, limit: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://datacenter-api.jin10.com/reports/list_v2",
            headers={
                "Referer": "https://datacenter.jin10.com/",
                "x-app-id": "rU6QIu7JHe2gOUeR",
                "x-csrf-token": "x-csrf-token",
                "x-version": "1.0.0",
            },
            params={
                "max_date": "",
                "category": "ec",
                "attr_id": attr_id,
                "_": str(int(time.time() * 1000)),
            },
            timeout=8,
        )
        rows = ((raw.get("data") or {}).get("values") or [])[:limit]
        records = []
        for row in rows:
            values = list(row) + [None] * 4
            records.append(
                {
                    "商品": symbol,
                    "日期": values[0],
                    "今值": self._float(values[1]),
                    "预测值": self._float(values[2]),
                    "前值": self._float(values[3]),
                }
            )
        if not records:
            raise ValueError("jin10 macro source returned empty data")
        return {"count": len(records), "records": records}

    def _fund_flow_rank_scrapling_payload(self, indicator: str, limit: int) -> dict[str, Any]:
        """Scrapling fallback for fund flow rank"""
        try:
            from scrapling.fetchers import Fetcher
            config = self._fund_flow_rank_direct_config(indicator)
            base_urls = [
                "https://push2delay.eastmoney.com",
                "https://push2.eastmoney.com",
                "https://push2his.eastmoney.com",
            ]
            last_error = None
            for base_url in base_urls:
                try:
                    page = Fetcher.get(
                        f"{base_url}/api/qt/clist/get",
                        params={
                            "fid": config["fid"],
                            "po": "1",
                            "pz": str(limit),
                            "pn": "1",
                            "np": "1",
                            "fs": "m:0+t:6+f:!2,m:0+t:13+f:!2,m:0+t:80+f:!2,m:1+t:2+f:!2,m:1+t:23+f:!2,m:0+t:7+f:!2,m:1+t:3+f:!2",
                            "fields": config["fields"],
                            "ut": "b2884a393a59ad64002292a3e90d46a5",
                        },
                        headers={"Referer": "https://data.eastmoney.com/zjlx/detail.html"},
                        timeout=5, retries=1, stealthy_headers=True,
                    )
                    if page.status != 200 or not page.body:
                        last_error = f"HTTP {page.status}"
                        continue
                    import json as _json
                    raw = _json.loads(page.body.decode("utf-8", errors="replace"))
                    rows = (raw.get("data") or {}).get("diff") or []
                    records = []
                    for idx, row in enumerate(rows[:limit], 1):
                        records.append({
                            "rank": idx,
                            "code": str(row.get("f12") or ""),
                            "name": row.get("f14"),
                            "price": self._float(row.get("f2")),
                            "change_pct": self._float(row.get(config["change_pct"])),
                            "main_net_inflow": self._float(row.get(config["main_net"])),
                            "main_net_inflow_ratio": self._float(row.get(config["main_ratio"])),
                        })
                    if not records:
                        last_error = "empty data"
                        continue
                    return {"indicator": indicator, "count": len(records), "records": records}
                except Exception as e:
                    last_error = str(e)
                    continue
            raise ValueError(f"scrapling fund flow rank failed: {last_error}")
        except ImportError:
            raise ValueError("scrapling not installed")

    def _fund_flow_rank_direct_payload(self, indicator: str, limit: int, base_url: str) -> dict[str, Any]:
        config = self._fund_flow_rank_direct_config(indicator)
        raw = self._eastmoney_clist(
            fs="m:0+t:6+f:!2,m:0+t:13+f:!2,m:0+t:80+f:!2,m:1+t:2+f:!2,m:1+t:23+f:!2,m:0+t:7+f:!2,m:1+t:3+f:!2",
            fields=config["fields"],
            limit=limit,
            fid=config["fid"],
            po="1",
            referer="https://data.eastmoney.com/zjlx/detail.html",
            ut="b2884a393a59ad64002292a3e90d46a5",
            base_url=base_url,
        )
        rows = (raw.get("data") or {}).get("diff") or []
        records = []
        for idx, row in enumerate(rows[:limit], 1):
            records.append(
                {
                    "rank": idx,
                    "code": str(row.get("f12") or ""),
                    "name": row.get("f14"),
                    "price": self._float(row.get("f2")),
                    "change_pct": self._float(row.get(config["change_pct"])),
                    "main_net_inflow": self._float(row.get(config["main_net"])),
                    "main_net_inflow_ratio": self._float(row.get(config["main_ratio"])),
                    "super_net_inflow": self._float(row.get(config["super_net"])),
                    "super_net_inflow_ratio": self._float(row.get(config["super_ratio"])),
                    "large_net_inflow": self._float(row.get(config["large_net"])),
                    "large_net_inflow_ratio": self._float(row.get(config["large_ratio"])),
                    "medium_net_inflow": self._float(row.get(config["medium_net"])),
                    "medium_net_inflow_ratio": self._float(row.get(config["medium_ratio"])),
                    "small_net_inflow": self._float(row.get(config["small_net"])),
                    "small_net_inflow_ratio": self._float(row.get(config["small_ratio"])),
                    "update_time": self._eastmoney_time(row.get("f124")),
                }
            )
        if not records:
            raise ValueError("eastmoney fund flow rank returned empty data")
        return {"indicator": indicator, "count": len(records), "records": records}

    def _fund_flow_rank_ths_fast_payload(self, indicator: str, limit: int) -> dict[str, Any]:
        import py_mini_racer
        from akshare.datasets import get_ths_js

        with open(get_ths_js("ths.js"), encoding="utf-8") as file:
            js_content = file.read()
        js_code = py_mini_racer.MiniRacer()
        js_code.eval(js_content)
        headers = {
            "Accept": "text/html, */*; q=0.01",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "hexin-v": js_code.call("v"),
            "Host": "data.10jqka.com.cn",
            "Pragma": "no-cache",
            "Referer": "http://data.10jqka.com.cn/funds/ggzjl/",
            "User-Agent": HTTP_HEADERS["User-Agent"],
            "X-Requested-With": "XMLHttpRequest",
        }
        board = {"今日": "", "3日": "board/3/", "5日": "board/5/", "10日": "board/10/"}[indicator]
        text = self._request_text(
            f"http://data.10jqka.com.cn/funds/ggzjl/{board}field/jlr/order/desc/page/1/ajax/1/free/1/",
            headers=headers,
        )
        tables = pd.read_html(StringIO(text))
        if not tables:
            raise ValueError("ths fast fund-flow page returned no table")
        return self._fund_flow_rank_ths_records(tables[0], indicator, limit, "ths_fast")

    def _fund_flow_rank_main_payload(self, indicator: str, limit: int) -> dict[str, Any]:
        if indicator not in {"今日", "5日", "10日"}:
            raise ValueError(f"stock_main_fund_flow does not support indicator: {indicator}")
        df = self._require_df(ak.stock_main_fund_flow(symbol="全部股票"), "stock_main_fund_flow")
        records = []
        prefix = {"今日": "今日排行榜", "5日": "5日排行榜", "10日": "10日排行榜"}[indicator]
        for idx, (_, row) in enumerate(df.head(limit).iterrows(), 1):
            records.append(
                {
                    "rank": idx,
                    "code": str(row.get("代码") or ""),
                    "name": row.get("名称"),
                    "price": self._float(row.get("最新价")),
                    "change_pct": self._float(row.get(f"{prefix}-{'今日' if indicator == '今日' else indicator}涨跌")),
                    "main_net_inflow_ratio": self._float(row.get(f"{prefix}-主力净占比")),
                    "source_rank": self._float(row.get(f"{prefix}-{'今日' if indicator == '今日' else indicator}排名")),
                    "sector": row.get("所属板块"),
                }
            )
        if not records:
            raise ValueError("stock_main_fund_flow returned empty data")
        return {"indicator": indicator, "count": len(records), "records": records}

    def _fund_flow_rank_ths_payload(self, indicator: str, limit: int) -> dict[str, Any]:
        symbol = {"今日": "即时", "3日": "3日排行", "5日": "5日排行", "10日": "10日排行"}[indicator]
        df = self._require_df(ak.stock_fund_flow_individual(symbol=symbol), "stock_fund_flow_individual")
        return self._fund_flow_rank_ths_records(df, indicator, limit, "akshare_ths")

    def _fund_flow_rank_ths_records(self, df: pd.DataFrame, indicator: str, limit: int, source_style: str) -> dict[str, Any]:
        df = self._require_df(df, "stock_fund_flow_individual").head(limit)
        records = []
        for idx, (_, row) in enumerate(df.head(limit).iterrows(), 1):
            net_value = row.get("净额", row.get("资金流入净额"))
            records.append(
                {
                    "rank": idx,
                    "code": str(row.get("股票代码") or ""),
                    "name": row.get("股票简称"),
                    "price": self._float(row.get("最新价")),
                    "change_pct": self._float(row.get("涨跌幅", row.get("阶段涨跌幅"))),
                    "turnover": self._float(row.get("换手率", row.get("连续换手率"))),
                    "main_net_inflow": self._amount_value(net_value),
                    "main_net_inflow_text": None if net_value is None else str(net_value),
                    "amount": self._amount_value(row.get("成交额")),
                    "source_style": source_style,
                }
            )
        if not records:
            raise ValueError("ths fund-flow rank returned empty data")
        return {"indicator": indicator, "count": len(records), "records": records}

    def _fund_flow_rank_direct_config(self, indicator: str) -> dict[str, str]:
        base = {
            "今日": ("f62", "f3", "f62", "f184", "f66", "f69", "f72", "f75", "f78", "f81", "f84", "f87"),
            "3日": ("f267", "f127", "f267", "f268", "f269", "f270", "f271", "f272", "f273", "f274", "f275", "f276"),
            "5日": ("f164", "f109", "f164", "f165", "f166", "f167", "f168", "f169", "f170", "f171", "f172", "f173"),
            "10日": ("f174", "f160", "f174", "f175", "f176", "f177", "f178", "f179", "f180", "f181", "f182", "f183"),
        }[indicator]
        fields = {
            "今日": "f12,f14,f2,f3,f62,f184,f66,f69,f72,f75,f78,f81,f84,f87,f204,f205,f124",
            "3日": "f12,f14,f2,f127,f267,f268,f269,f270,f271,f272,f273,f274,f275,f276,f257,f258,f124",
            "5日": "f12,f14,f2,f109,f164,f165,f166,f167,f168,f169,f170,f171,f172,f173,f257,f258,f124",
            "10日": "f12,f14,f2,f160,f174,f175,f176,f177,f178,f179,f180,f181,f182,f183,f260,f261,f124",
        }[indicator]
        names = [
            "fid",
            "change_pct",
            "main_net",
            "main_ratio",
            "super_net",
            "super_ratio",
            "large_net",
            "large_ratio",
            "medium_net",
            "medium_ratio",
            "small_net",
            "small_ratio",
        ]
        return {**dict(zip(names, base)), "fields": fields}

    def _big_deal_akshare(self, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.stock_fund_flow_big_deal(), "stock_fund_flow_big_deal")
        records = dataframe_to_records(df.head(limit), limit)
        if not records:
            raise ValueError("big deal data empty")
        return {"count": len(records), "records": records}

    def _big_deal_scrapling(self, limit: int) -> dict[str, Any]:
        """使用 Scrapling 从同花顺网页抓取大单成交数据"""
        try:
            from scrapling.fetchers import Fetcher

            # 计算需要抓取的页数（每页50条）
            pages_needed = min((limit + 49) // 50, 5)  # 最多5页

            all_records = []
            for page_num in range(1, pages_needed + 1):
                url = f'https://data.10jqka.com.cn/funds/ddzz/order/desc/page/{page_num}/ajax/1/free/1/'
                page = Fetcher.get(url, headers={
                    'Referer': 'https://data.10jqka.com.cn/funds/ddzz/',
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
                })

                if page.status != 200:
                    continue

                # 解析表格
                tables = page.css('table')
                if not tables:
                    continue

                rows = tables[0].css('tbody tr')
                for row in rows:
                    cells = row.css('td')
                    if len(cells) >= 8:
                        record = {
                            '成交时间': cells[0].text.strip() if cells[0].text else None,
                            '股票代码': cells[1].text.strip() if cells[1].text else None,
                            '股票名称': cells[2].text.strip() if cells[2].text else None,
                            '成交价格': self._float(cells[3].text.strip() if cells[3].text else None),
                            '成交量(手)': self._float(cells[4].text.strip() if cells[4].text else None),
                            '成交金额(万元)': self._float(cells[5].text.strip() if cells[5].text else None),
                            '买卖方向': cells[6].text.strip() if cells[6].text else None,
                            '涨跌幅': cells[7].text.strip() if cells[7].text else None,
                        }
                        all_records.append(record)

                        if len(all_records) >= limit:
                            break

                if len(all_records) >= limit:
                    break

            if not all_records:
                raise ValueError("scrapling found no data")

            return {"count": len(all_records), "records": all_records[:limit], "source_note": "Scrapling网页抓取"}

        except ImportError:
            raise ValueError("scrapling not installed")
        except Exception as e:
            raise ValueError(f"scrapling failed: {e}")

    def _big_deal_rank(self, limit: int) -> dict[str, Any]:
        try:
            df = ak.stock_individual_fund_flow_rank(indicator="今日")
            if df is None or df.empty:
                raise ValueError("fund flow rank returned empty")
            # Normalize column names to match expected format
            rename = {}
            for col in df.columns:
                cl = str(col).lower()
                if "代码" in cl or "code" in cl:
                    rename[col] = "股票代码"
                elif "名称" in cl or "name" in cl:
                    rename[col] = "股票名称"
                elif "最新价" in cl:
                    rename[col] = "最新价"
                elif "涨跌幅" in cl:
                    rename[col] = "涨跌幅"
            if rename:
                df = df.rename(columns=rename)
            records = dataframe_to_records(df.head(limit), limit)
            if not records:
                raise ValueError("fund flow rank records empty")
            return {"count": len(records), "records": records, "fallback_note": "数据来源为资金流排名，非逐笔大单成交"}
        except Exception as e:
            raise ValueError(f"fund flow rank failed: {e}")

    def _big_deal_direct(self, limit: int) -> dict[str, Any]:
        import requests as _req
        # Use push2delay (works on HF), fallback to push2
        params = {
            "pn": 1,
            "pz": limit,
            "po": 1,
            "np": 1,
            "ut": "bd1d9ddb04089700cf9c27f6f7426281",
            "fltt": 2,
            "invt": 2,
            "fid": "f62",
            "fs": "m:0+t:6,m:0+t:80,m:1+t:2,m:1+t:23,m:0+t:7,m:1+t:3",
            "fields": "f12,f14,f2,f3,f62,f184,f66,f69,f72,f75,f78,f81,f84,f87,f124",
        }
        last_err = None
        for host in ("push2delay.eastmoney.com", "push2.eastmoney.com"):
            try:
                resp = _req.get(
                    f"https://{host}/api/qt/clist/get",
                    params=params,
                    headers={**HTTP_HEADERS, "Referer": "https://data.eastmoney.com/"},
                    timeout=15,
                )
                resp.raise_for_status()
                raw = resp.json()
                data = raw.get("data") or {}
                diff = data.get("diff") or []
                if diff:
                    records = []
                    for idx, row in enumerate(diff[:limit], 1):
                        records.append({
                            "rank": idx,
                            "code": row.get("f12"),
                            "name": row.get("f14"),
                            "price": self._float(row.get("f2")),
                            "change_pct": self._float(row.get("f3")),
                            "main_net_inflow": self._float(row.get("f62")),
                            "main_net_inflow_ratio": self._float(row.get("f184")),
                        })
                    return {"count": len(records), "records": records, "source_note": "东方财富主力资金排行（非逐笔大单）"}
            except Exception as e:
                last_err = e
                continue
        raise ValueError(f"eastmoney fund flow unavailable: {last_err}")

    def _limit_pool_payload(self, df: pd.DataFrame, trade_date: str, side: str, limit: int) -> dict[str, Any]:
        if df is None:
            raise ValueError("limit pool source returned None")
        records = dataframe_to_records(df, limit)
        return {"date": trade_date, "side": side, "count": len(df), "stocks": records}

    def _eastmoney_limit_pool_payload(self, trade_date: str, side: str, limit: int) -> dict[str, Any]:
        path = "getTopicZTPool" if side == "up" else "getTopicDTPool"
        params = {
            "ut": "7eea3edcaed734bea9cbfc24409ed989",
            "dpt": "wz.ztzt",
            "Pageindex": "0",
            "pagesize": str(min(max(int(limit), 1), 10000)),
            "sort": "fbt:asc" if side == "up" else "fund:asc",
            "date": trade_date,
        }
        raw = self._request_json(
            f"https://push2ex.eastmoney.com/{path}",
            headers={"Referer": "https://quote.eastmoney.com/ztb/detail"},
            params=params,
            timeout=5,
        )
        if raw.get("rc") not in (0, None):
            raise ValueError(f"eastmoney limit pool returned rc={raw.get('rc')}")
        data = raw.get("data") or {}
        rows = data.get("pool") or []
        total = self._float(data.get("tc"))
        if not rows and total:
            raise ValueError(f"eastmoney limit pool returned empty page with total={total}")

        def _time_text(value: Any) -> str | None:
            text = str(value or "").split(".", 1)[0].zfill(6)
            if len(text) != 6 or not text.isdigit():
                return None
            return f"{text[:2]}:{text[2:4]}:{text[4:6]}"

        stocks = []
        for idx, row in enumerate(rows[:limit], 1):
            code = str(row.get("c") or "").zfill(6)
            item = {
                "rank": idx,
                "code": code,
                "stock_code": self._display_code_from_digits(code) if len(code) == 6 and code.isdigit() else code,
                "name": row.get("n"),
                "price": round(self._float(row.get("p"), 0.0) / 1000, 4) if row.get("p") is not None else None,
                "change_pct": self._float(row.get("zdp")),
                "amount": self._float(row.get("amount")),
                "free_market_cap": self._float(row.get("ltsz")),
                "total_market_cap": self._float(row.get("tshare")),
                "turnover_rate": self._float(row.get("hs")),
                "sealed_amount": self._float(row.get("fund")),
                "industry": row.get("hybk"),
            }
            if side == "up":
                stat = row.get("zttj") if isinstance(row.get("zttj"), dict) else {}
                item.update(
                    {
                        "limit_up_count": self._float(row.get("lbc")),
                        "first_limit_time": _time_text(row.get("fbt")),
                        "last_limit_time": _time_text(row.get("lbt")),
                        "open_count": self._float(row.get("zbc")),
                        "limit_up_days": self._float(stat.get("days")),
                        "limit_up_total": self._float(stat.get("ct")),
                    }
                )
            else:
                item.update(
                    {
                        "pe_dynamic": self._float(row.get("pe")),
                        "last_limit_time": _time_text(row.get("lbt")),
                        "board_amount": self._float(row.get("fba")),
                        "consecutive_limit_down": self._float(row.get("days")),
                        "open_count": self._float(row.get("oc")),
                    }
                )
            stocks.append(item)
        return {
            "date": str(data.get("qdate") or trade_date),
            "side": side,
            "count": len(stocks),
            "total": int(total) if total is not None else None,
            "stocks": stocks,
        }

    def _ths_limit_up_pool_payload(self, trade_date: str, limit: int) -> dict[str, Any]:
        params = {
            "page": "1",
            "limit": str(min(max(int(limit), 1), 200)),
            "field": THS_LIMIT_UP_POOL_FIELD,
            "filter": "HS,GEM2STAR",
            "order_field": "330324",
            "order_type": "0",
            "date": trade_date,
            "_": str(int(self._now_cn().timestamp() * 1000)),
        }
        raw = self._request_json(
            THS_LIMIT_UP_POOL_URL,
            headers={"Referer": "https://data.10jqka.com.cn/"},
            params=params,
        )
        if raw.get("status_code") != 0:
            raise ValueError(f"ths limit_up_pool returned {raw.get('status_code')}: {raw.get('status_msg')}")
        data = raw.get("data")
        if not isinstance(data, dict):
            raise ValueError("ths limit_up_pool data is not an object")
        rows = data.get("info")
        if not isinstance(rows, list):
            raise ValueError("ths limit_up_pool info is not a list")

        stocks = []
        for row in rows[:limit]:
            if not isinstance(row, dict):
                continue
            item = dict(row)
            raw_code = str(item.get("code") or "")
            if len(raw_code) == 6 and raw_code.isdigit():
                item.setdefault("stock_code", self._display_code_from_digits(raw_code))
            for key in ("first_limit_up_time", "last_limit_up_time"):
                if key in item:
                    item[f"{key}_at"] = self._eastmoney_time(item.get(key))
            stocks.append(item)
        if not stocks:
            raise ValueError("ths limit_up_pool returned empty data")
        return {
            "date": trade_date,
            "side": "up",
            "count": len(stocks),
            "stocks": stocks,
            "stats": {
                "limit_up_count": data.get("limit_up_count"),
                "limit_down_count": data.get("limit_down_count"),
                "trade_status": data.get("trade_status"),
                "source_response_date": data.get("date"),
            },
        }

    def _trade_calendar_sina_payload(self, start_date: str | None, end_date: str | None, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.tool_trade_date_hist_sina(), "tool_trade_date_hist_sina")
        column = "trade_date" if "trade_date" in df.columns else df.columns[0]
        values = pd.to_datetime(df[column], errors="coerce").dropna()
        dates = sorted({value.strftime("%Y-%m-%d") for value in values})
        if start_date:
            dates = [value for value in dates if value >= start_date]
        if end_date:
            dates = [value for value in dates if value <= end_date]
        dates = dates[-limit:]
        if not dates:
            raise ValueError("sina trade calendar returned empty data")
        return {
            "start_date": start_date or dates[0],
            "end_date": end_date or dates[-1],
            "count": len(dates),
            "trade_dates": dates,
        }

    def _chinabond_yield_curve_payload(self, start_date: str, end_date: str, limit: int) -> dict[str, Any]:
        raw = self._request_json(
            CHINABOND_HISTORY_QUERY_URL,
            headers={"Referer": "https://yield.chinabond.com.cn/"},
            params={
                "startDate": start_date,
                "endDate": end_date,
                "gjqx": "0",
                "locale": "cn_ZH",
                "qxmc": "1",
            },
        )
        if str(raw.get("flag")) != "0":
            raise ValueError(f"chinabond returned flag={raw.get('flag')!r}")
        rows = raw.get("heList")
        if not isinstance(rows, list):
            raise ValueError("chinabond heList is not a list")

        records = []
        for row in rows:
            if not isinstance(row, dict):
                continue
            work_date = row.get("workTime")
            if not work_date:
                continue
            record = {
                "trade_date": str(work_date)[:10],
                "curve_name": row.get("qxmc"),
            }
            for raw_key, target_key in CHINABOND_YIELD_FIELDS:
                record[target_key] = self._float(row.get(raw_key))
            records.append(record)
        records.sort(key=lambda item: str(item.get("trade_date") or ""))
        records = records[-limit:]
        if not records:
            raise ValueError("chinabond returned empty yield curve data")
        return {
            "start_date": start_date,
            "end_date": end_date,
            "count": len(records),
            "records": records,
        }

    def _board_flow_payload(self, df: pd.DataFrame, key: str, limit: int) -> dict[str, Any]:
        df = self._require_df(df, key)
        records = dataframe_to_records(df, limit)
        return {"count": len(records), key: records}

    def _legacy_board_flow_direct_df(self, category: str, limit: int, base_url: str) -> pd.DataFrame:
        raw = self._eastmoney_clist(
            fs=self._board_flow_fs(category),
            fields="f12,f14,f2,f3,f8,f20,f62,f104,f105,f128,f140,f136,f207,f208,f124",
            limit=limit,
            fid="f3",
            po="1",
            referer="https://quote.eastmoney.com/center/boardlist.html",
            base_url=base_url,
        )
        rows = (raw.get("data") or {}).get("diff") or []
        records = []
        for idx, row in enumerate(rows[:limit], 1):
            leader = row.get("f128") or row.get("f207") or ""
            records.append(
                {
                    "序号": idx,
                    "代码": row.get("f12"),
                    "行业": row.get("f14"),
                    "行业指数": self._float(row.get("f2")),
                    "行业-涨跌幅": self._float(row.get("f3")),
                    "换手率": self._float(row.get("f8")),
                    "流通市值": self._float(row.get("f20")),
                    "净额": self._amount_value(row.get("f62"), 0.0) / 100000000,
                    "公司家数": self._float(row.get("f104")),
                    "上涨家数": self._float(row.get("f104")),
                    "下跌家数": self._float(row.get("f105")),
                    "领涨股": leader,
                    "领涨股代码": row.get("f140") or row.get("f208"),
                    "领涨股-涨跌幅": self._float(row.get("f136")),
                    "当前价": None,
                    "数据源分类": category,
                    "更新时间": self._eastmoney_time(row.get("f124")),
                }
            )
        if not records:
            raise ValueError(f"eastmoney {category} board legacy flow returned empty data")
        return pd.DataFrame(records)

    def _legacy_board_flow_combined_df(self, base_url: str) -> pd.DataFrame:
        frames = []
        errors = []
        for category, limit in (("concept", 500), ("industry", 300)):
            try:
                frames.append(self._legacy_board_flow_direct_df(category, limit, base_url))
            except Exception as exc:
                errors.append(f"{category}: {type(exc).__name__}: {exc}")
        if not frames:
            raise ValueError(f"eastmoney combined board legacy flow returned no data; {'; '.join(errors)}")
        return pd.concat(frames, ignore_index=True)

    def _board_flow_direct_payload(self, category: str, limit: int, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        raw = self._eastmoney_clist(
            fs=self._board_flow_fs(category),
            fields="f12,f14,f2,f3,f62,f184,f66,f69,f72,f75,f78,f81,f84,f87,f124",
            limit=limit,
            fid="f62",
            po="1",
            referer="https://data.eastmoney.com/",
            base_url=base_url,
        )
        data = raw.get("data") or {}
        rows = data.get("diff") or []
        items = []
        update_time = ""
        for idx, row in enumerate(rows[:limit], 1):
            timestamp = self._eastmoney_time(row.get("f124"))
            if timestamp and not update_time:
                update_time = timestamp
            items.append(
                {
                    "rank": idx,
                    "code": row.get("f12"),
                    "name": row.get("f14"),
                    "price": self._float(row.get("f2")),
                    "change_pct": self._float(row.get("f3")),
                    "main_net_inflow": self._float(row.get("f62")),
                    "main_net_inflow_ratio": self._float(row.get("f184")),
                    "super_net_inflow": self._float(row.get("f66")),
                    "super_net_inflow_ratio": self._float(row.get("f69")),
                    "large_net_inflow": self._float(row.get("f72")),
                    "large_net_inflow_ratio": self._float(row.get("f75")),
                    "medium_net_inflow": self._float(row.get("f78")),
                    "medium_net_inflow_ratio": self._float(row.get("f81")),
                    "small_net_inflow": self._float(row.get("f84")),
                    "small_net_inflow_ratio": self._float(row.get("f87")),
                    "update_time": timestamp,
                }
            )
        if not items:
            raise ValueError("eastmoney board flow returned empty data")
        return {"category": category, "count": len(items), "total": data.get("total"), "update_time": update_time, "items": items}

    def _board_flow_direct_key_payload(self, category: str, key: str, limit: int, base_url: str) -> dict[str, Any]:
        payload = self._board_flow_direct_payload(category, limit, base_url)
        payload[key] = payload.get("items") or []
        return payload

    def _market_indices_sina_payload(self, limit: int) -> dict[str, Any]:
        symbols = ["s_sh000001", "s_sz399001", "s_sz399006", "s_sh000688", "s_sz399005", "s_sz399300"]
        text = self._request_text(
            f"http://hq.sinajs.cn/rn={int(self._now_cn().timestamp() * 1000)}&list={','.join(symbols)}",
            headers={"Referer": "http://finance.sina.com.cn"},
            encoding="gbk",
        )
        items = []
        for line in text.splitlines():
            if "=" not in line:
                continue
            symbol = line.split("=", 1)[0].split("_")[-1]
            value = line.split("=", 1)[1].strip().strip(";").strip('"')
            parts = value.split(",")
            if len(parts) < 6 or not parts[0]:
                continue
            items.append(
                {
                    "code": symbol,
                    "name": parts[0],
                    "price": self._float(parts[1]),
                    "change_amount": self._float(parts[2]),
                    "change_pct": self._float(parts[3]),
                    "volume": self._float(parts[4]),
                    "amount": self._float(parts[5]),
                }
            )
        if not items:
            raise ValueError("sina index quote returned empty data")
        return {"count": min(len(items), limit), "indices": items[:limit]}

    def _market_moves_payload(self, move_type: str, limit: int, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        fid, po = self._stock_move_sort(move_type)
        raw = self._eastmoney_clist(
            fs="m:0 t:6,m:0 t:80,m:1 t:2,m:1 t:23",
            fields="f12,f14,f2,f3,f22,f8,f5,f6,f62,f184,f15,f16,f17,f18,f124",
            limit=limit,
            fid=fid,
            po=po,
            referer="https://quote.eastmoney.com/",
            base_url=base_url,
            timeout=5,
        )
        data = raw.get("data") or {}
        rows = data.get("diff") or []
        items = []
        update_time = ""
        for idx, row in enumerate(rows[:limit], 1):
            timestamp = self._eastmoney_time(row.get("f124"))
            if timestamp and not update_time:
                update_time = timestamp
            items.append(
                {
                    "rank": idx,
                    "code": row.get("f12"),
                    "name": row.get("f14"),
                    "price": self._float(row.get("f2")),
                    "change_pct": self._float(row.get("f3")),
                    "speed": self._float(row.get("f22")),
                    "turnover_rate": self._float(row.get("f8")),
                    "volume": self._float(row.get("f5")),
                    "amount": self._float(row.get("f6")),
                    "main_net_inflow": self._float(row.get("f62")),
                    "main_net_inflow_ratio": self._float(row.get("f184")),
                    "high": self._float(row.get("f15")),
                    "low": self._float(row.get("f16")),
                    "open": self._float(row.get("f17")),
                    "pre_close": self._float(row.get("f18")),
                    "update_time": timestamp,
                }
            )
        if not items:
            raise ValueError("eastmoney stock moves returned empty data")
        return {"move_type": move_type, "count": len(items), "total": data.get("total"), "update_time": update_time, "items": items}

    def _market_moves_hot_rank_payload(self, move_type: str, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.stock_hot_rank_em(), "stock_hot_rank_em").copy()
        if "涨跌幅" in df.columns:
            df["涨跌幅"] = pd.to_numeric(df["涨跌幅"], errors="coerce")
        if move_type in {"drop", "change_down"} and "涨跌幅" in df.columns:
            df = df.sort_values("涨跌幅", ascending=True)
        elif move_type in {"change_up", "surge"} and "涨跌幅" in df.columns:
            df = df.sort_values("涨跌幅", ascending=False)
        records = []
        for idx, (_, row) in enumerate(df.head(limit).iterrows(), 1):
            code = str(row.get("代码", "")).strip()
            records.append(
                {
                    "rank": idx,
                    "code": code[-6:] if len(code) >= 6 else code,
                    "market_code": code,
                    "name": row.get("股票名称"),
                    "price": self._float(row.get("最新价")),
                    "change_amount": self._float(row.get("涨跌额")),
                    "change_pct": self._float(row.get("涨跌幅")),
                    "hot_rank": self._float(row.get("当前排名")),
                    "fallback_note": "EastMoney stock-move source failed; this row is derived from EastMoney hot-rank data via AKShare.",
                }
            )
        if not records:
            raise ValueError("stock_hot_rank_em returned empty data")
        return {"move_type": move_type, "count": len(records), "items": records}

    def _market_moves_ths_rank_payload(self, move_type: str, limit: int) -> dict[str, Any]:
        if move_type in {"drop", "change_down"}:
            df = self._require_df(ak.stock_rank_xxtp_ths(symbol="500日均线"), "stock_rank_xxtp_ths")
            fallback_kind = "ths_down_break"
        elif move_type in {"turnover", "mainflow"}:
            df = self._require_df(ak.stock_rank_lxsz_ths(), "stock_rank_lxsz_ths")
            fallback_kind = "ths_continue_rise"
        else:
            df = self._require_df(ak.stock_rank_cxg_ths(symbol="创月新高"), "stock_rank_cxg_ths")
            fallback_kind = "ths_new_high"
        records = []
        for idx, (_, row) in enumerate(df.head(limit).iterrows(), 1):
            records.append(
                {
                    "rank": idx,
                    "code": str(row.get("股票代码", "")).strip(),
                    "name": row.get("股票简称"),
                    "price": self._float(row.get("最新价", row.get("收盘价"))),
                    "change_pct": self._float(row.get("涨跌幅", row.get("连续涨跌幅", row.get("阶段涨幅")))),
                    "turnover_rate": self._float(row.get("换手率", row.get("累计换手率"))),
                    "amount": row.get("成交额"),
                    "volume": row.get("成交量"),
                    "industry": row.get("所属行业"),
                    "fallback_kind": fallback_kind,
                    "fallback_note": "EastMoney stock-move source failed; this row is derived from TongHuaShun technical rank data via AKShare.",
                }
            )
        if not records:
            raise ValueError("ths rank returned empty data")
        return {"move_type": move_type, "count": len(records), "items": records}

    def _etf_spot_direct_payload(self, limit: int, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        raw = self._eastmoney_clist(
            fs="b:MK0021,b:MK0022,b:MK0023,b:MK0024,b:MK0827",
            fields="f12,f14,f2,f3,f4,f5,f6,f8,f9,f10,f15,f16,f17,f18,f20,f21,f124,f297,f402,f441",
            limit=limit,
            fid="f12",
            po="1",
            referer="https://quote.eastmoney.com/",
            base_url=base_url,
        )
        data = raw.get("data") or {}
        rows = data.get("diff") or []
        items = []
        for row in rows[:limit]:
            items.append(
                {
                    "code": row.get("f12"),
                    "name": row.get("f14"),
                    "price": self._float(row.get("f2")),
                    "change_pct": self._float(row.get("f3")),
                    "change_amount": self._float(row.get("f4")),
                    "volume": self._float(row.get("f5")),
                    "amount": self._float(row.get("f6")),
                    "turnover_rate": self._float(row.get("f8")),
                    "pe": self._float(row.get("f9")),
                    "volume_ratio": self._float(row.get("f10")),
                    "high": self._float(row.get("f15")),
                    "low": self._float(row.get("f16")),
                    "open": self._float(row.get("f17")),
                    "pre_close": self._float(row.get("f18")),
                    "total_market_value": self._float(row.get("f20")),
                    "circulating_market_value": self._float(row.get("f21")),
                    "premium_discount_rate": self._float(row.get("f402")),
                    "iopv_realtime_value": self._float(row.get("f441")),
                    "data_date": row.get("f297"),
                    "update_time": self._eastmoney_time(row.get("f124")),
                }
            )
        if not items:
            raise ValueError("eastmoney etf spot returned empty data")
        return {"count": len(items), "total": data.get("total"), "etfs": items}

    def _sort_etf_premium_items(self, items: list[dict[str, Any]], sort: str) -> list[dict[str, Any]]:
        if sort == "premium":
            return sorted(items, key=lambda item: item.get("premium_discount_rate") if item.get("premium_discount_rate") is not None else -999999, reverse=True)
        if sort == "discount":
            return sorted(items, key=lambda item: item.get("premium_discount_rate") if item.get("premium_discount_rate") is not None else 999999)
        if sort == "code":
            return sorted(items, key=lambda item: str(item.get("code") or item.get("fund_code") or ""))
        return sorted(items, key=lambda item: abs(item.get("premium_discount_rate") or 0), reverse=True)

    def _etf_premium_direct_payload(self, limit: int, sort: str, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        payload = self._etf_spot_direct_payload(5000, base_url)
        items = []
        for row in payload.get("etfs") or []:
            rate = row.get("premium_discount_rate")
            if rate is None:
                continue
            items.append(
                {
                    "fund_code": row.get("code"),
                    "code": row.get("code"),
                    "name": row.get("name"),
                    "price": row.get("price"),
                    "iopv_realtime_value": row.get("iopv_realtime_value"),
                    "premium_discount_rate": rate,
                    "change_pct": row.get("change_pct"),
                    "amount": row.get("amount"),
                    "turnover_rate": row.get("turnover_rate"),
                    "data_date": row.get("data_date"),
                    "update_time": row.get("update_time"),
                }
            )
        if not items:
            raise ValueError("eastmoney etf premium source returned empty data")
        records = self._sort_etf_premium_items(items, sort)[:limit]
        return {"count": len(records), "total": len(items), "sort": sort, "records": records}

    def _etf_premium_detail_direct_payload(self, code: NormalizedStockCode, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        raw = self._request_json(
            f"{base_url.rstrip('/')}/api/qt/ulist.np/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "fltt": "2",
                "invt": "2",
                "secids": self._etf_secid(code),
                "fields": "f12,f14,f2,f3,f4,f5,f6,f8,f15,f16,f17,f18,f20,f21,f124,f297,f402,f441",
                "ut": "b2884a393a59ad64002292a3e90d46a5",
            },
            timeout=8,
        )
        rows = (raw.get("data") or {}).get("diff") or []
        row = next((item for item in rows if str(item.get("f12") or "").zfill(6) == code.code), None)
        if row is None:
            raise ValueError(f"eastmoney etf premium source returned no data: {code.display}")
        return {
            "fund_code": code.display,
            "code": code.code,
            "name": row.get("f14"),
            "price": self._float(row.get("f2")),
            "iopv_realtime_value": self._float(row.get("f441")),
            "premium_discount_rate": self._float(row.get("f402")),
            "change_pct": self._float(row.get("f3")),
            "amount": self._float(row.get("f6")),
            "turnover_rate": self._float(row.get("f8")),
            "data_date": row.get("f297"),
            "update_time": self._eastmoney_time(row.get("f124")),
        }

    def _etf_premium_from_akshare(self, limit: int, sort: str) -> dict[str, Any]:
        df = self._require_df(ak.fund_etf_spot_em(), "fund_etf_spot_em")
        items = []
        for _, row in df.iterrows():
            rate = self._float(row.get("基金折价率"))
            if rate is None:
                continue
            code = str(row.get("代码") or "").strip().zfill(6)
            items.append(
                {
                    "fund_code": code,
                    "code": code,
                    "name": row.get("名称"),
                    "price": self._float(row.get("最新价")),
                    "iopv_realtime_value": self._float(row.get("IOPV实时估值")),
                    "premium_discount_rate": rate,
                    "change_pct": self._float(row.get("涨跌幅")),
                    "amount": self._float(row.get("成交额")),
                    "turnover_rate": self._float(row.get("换手率")),
                    "data_date": row.get("数据日期"),
                    "update_time": self._eastmoney_time(row.get("更新时间")),
                }
            )
        if not items:
            raise ValueError("akshare etf premium source returned empty data")
        records = self._sort_etf_premium_items(items, sort)[:limit]
        return {"count": len(records), "total": len(items), "sort": sort, "records": records}

    def _etf_premium_detail_from_akshare(self, code: NormalizedStockCode) -> dict[str, Any]:
        payload = self._etf_premium_from_akshare(5000, "code")
        row = next((item for item in payload.get("records") or [] if str(item.get("code") or "").zfill(6) == code.code), None)
        if row is None:
            raise ValueError(f"akshare etf premium source returned no data: {code.display}")
        return {**row, "fund_code": code.display, "code": code.code}

    def _index_fund_flow_payload(self, secid: str, interval: str, limit: int, base_url: str = "https://push2.eastmoney.com") -> dict[str, Any]:
        raw = self._request_json(
            f"{base_url}/api/qt/stock/fflow/kline/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "lmt": "0",
                "klt": interval,
                "fields1": "f1,f2,f3,f7",
                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65",
                "ut": "b2884a393a59ad64002292a3e90d46a5",
                "secid": secid,
            },
        )
        data = raw.get("data") or {}
        points = []
        for line in (data.get("klines") or [])[-limit:]:
            parts = str(line).split(",")
            if len(parts) < 6:
                continue
            points.append(
                {
                    "time": parts[0],
                    "main_net_inflow": self._float(parts[1]),
                    "super_net_inflow": self._float(parts[2]),
                    "large_net_inflow": self._float(parts[3]),
                    "medium_net_inflow": self._float(parts[4]),
                    "small_net_inflow": self._float(parts[5]),
                }
            )
        if not points:
            raise ValueError("eastmoney index fund flow returned empty data")
        return {
            "code": data.get("code"),
            "name": data.get("name"),
            "market": data.get("market"),
            "secid": secid,
            "interval": interval,
            "count": len(points),
            "records": points,
        }

    def _index_fund_flow_hsgt_fallback_payload(self, index_code: str, interval: str, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.stock_hsgt_fund_flow_summary_em(), "stock_hsgt_fund_flow_summary_em")
        records = dataframe_to_records(df.tail(limit), limit)
        return {
            "code": index_code,
            "interval": interval,
            "count": len(records),
            "records": records,
            "fallback_note": (
                "EastMoney index fund-flow K-line source failed; this fallback returns "
                "沪深港通资金摘要, not per-minute index main-force K-lines."
            ),
        }

    def _longhubang_payload(self, trade_date: str, limit: int, page: int) -> dict[str, Any]:
        params = {
            "sortColumns": "TRADE_DATE,BILLBOARD_NET_AMT",
            "sortTypes": "-1,-1",
            "pageSize": limit,
            "pageNumber": page,
            "reportName": "RPT_DAILYBILLBOARD_DETAILSNEW",
            "columns": (
                "SECURITY_CODE,SECUCODE,SECURITY_NAME_ABBR,TRADE_DATE,EXPLAIN,CLOSE_PRICE,CHANGE_RATE,"
                "BILLBOARD_NET_AMT,BILLBOARD_BUY_AMT,BILLBOARD_SELL_AMT,BILLBOARD_DEAL_AMT,ACCUM_AMOUNT,"
                "DEAL_NET_RATIO,DEAL_AMOUNT_RATIO,TURNOVERRATE,FREE_MARKET_CAP,EXPLANATION,"
                "D1_CLOSE_ADJCHRATE,D2_CLOSE_ADJCHRATE,D5_CLOSE_ADJCHRATE,D10_CLOSE_ADJCHRATE,SECURITY_TYPE_CODE"
            ),
            "source": "WEB",
            "client": "WEB",
        }
        if trade_date:
            params["filter"] = f"(TRADE_DATE='{trade_date}')"
        raw = self._request_json(
            "https://datacenter-web.eastmoney.com/api/data/v1/get",
            headers={"Referer": "https://data.eastmoney.com/"},
            params=params,
        )
        result = raw.get("result") or {}
        rows = result.get("data") or []
        items = []
        for row in rows:
            trade_time = str(row.get("TRADE_DATE") or "")
            items.append(
                {
                    "trade_date": trade_time[:10],
                    "code": row.get("SECURITY_CODE"),
                    "secucode": row.get("SECUCODE"),
                    "name": row.get("SECURITY_NAME_ABBR"),
                    "close": self._float(row.get("CLOSE_PRICE")),
                    "change_pct": self._float(row.get("CHANGE_RATE")),
                    "net_buy_amount": self._float(row.get("BILLBOARD_NET_AMT")),
                    "buy_amount": self._float(row.get("BILLBOARD_BUY_AMT")),
                    "sell_amount": self._float(row.get("BILLBOARD_SELL_AMT")),
                    "deal_amount": self._float(row.get("BILLBOARD_DEAL_AMT")),
                    "turnover_rate": self._float(row.get("TURNOVERRATE")),
                    "free_market_cap": self._float(row.get("FREE_MARKET_CAP")),
                    "reason": row.get("EXPLAIN"),
                    "reason_detail": row.get("EXPLANATION"),
                    "d1_change": self._float(row.get("D1_CLOSE_ADJCHRATE")),
                    "d2_change": self._float(row.get("D2_CLOSE_ADJCHRATE")),
                    "d5_change": self._float(row.get("D5_CLOSE_ADJCHRATE")),
                    "d10_change": self._float(row.get("D10_CLOSE_ADJCHRATE")),
                }
            )
        if not items:
            raise ValueError("eastmoney longhubang returned empty data")
        return {"date": trade_date or None, "page": page, "count": len(items), "total": result.get("count"), "items": items}

    def _stock_announcements_payload(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://np-anotice-stock.eastmoney.com/api/security/ann",
            headers={"Referer": "https://notice.eastmoney.com/"},
            params={
                "ann_type": "A",
                "client_source": "web",
                "stock_list": code.code,
                "page_index": 1,
                "page_size": limit,
            },
        )
        data = raw.get("data") or {}
        rows = data.get("list") or []
        items = [
            {
                "title": row.get("title"),
                "notice_date": row.get("notice_date") or row.get("display_time"),
                "type": row.get("ann_type"),
                "columns": row.get("columns"),
                "art_code": row.get("art_code"),
            }
            for row in rows[:limit]
        ]
        if not items:
            raise ValueError("eastmoney announcements returned empty data")
        return {"stock_code": code.display, "count": len(items), "total": data.get("total"), "notices": items}

    def _notices_scrapling_payload(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        """Scrapling fallback for stock notices"""
        try:
            from scrapling.fetchers import Fetcher
            page = Fetcher.get(
                "https://np-anotice-stock.eastmoney.com/api/security/ann",
                params={
                    "ann_type": "A",
                    "client_source": "web",
                    "stock_list": code.code,
                    "page_index": "1",
                    "page_size": str(limit),
                },
                headers={"Referer": "https://notice.eastmoney.com/"},
                timeout=5, retries=1, stealthy_headers=True,
            )
            if page.status != 200 or not page.body:
                raise ValueError(f"scrapling notices HTTP {page.status}")
            import json as _json
            raw = _json.loads(page.body.decode("utf-8", errors="replace"))
            data = raw.get("data") or {}
            rows = data.get("list") or []
            items = [
                {
                    "title": row.get("title"),
                    "notice_date": row.get("notice_date") or row.get("display_time"),
                    "type": row.get("ann_type"),
                    "columns": row.get("columns"),
                    "art_code": row.get("art_code"),
                }
                for row in rows[:limit]
            ]
            if not items:
                raise ValueError("scrapling notices returned empty")
            return {"stock_code": code.display, "count": len(items), "total": data.get("total"), "notices": items}
        except ImportError:
            raise ValueError("scrapling not installed")
        except Exception as e:
            raise ValueError(f"scrapling notices failed: {e}")

    def _research_reports_payload(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        end = self._now_cn().date()
        begin = end - timedelta(days=365)
        raw = self._request_json(
            "https://reportapi.eastmoney.com/report/list",
            headers={"Referer": "https://data.eastmoney.com/"},
            params={
                "pageSize": limit,
                "beginTime": begin.isoformat(),
                "endTime": end.isoformat(),
                "pageNo": 1,
                "qType": 0,
                "code": code.code,
            },
        )
        rows = raw.get("data") or raw.get("Data") or []
        reports = []
        for row in rows[:limit]:
            reports.append(
                {
                    "title": row.get("title") or row.get("TITLE"),
                    "org_name": row.get("orgName") or row.get("ORG_NAME"),
                    "publish_date": row.get("publishDate") or row.get("PUBLISH_DATE"),
                    "author": row.get("author") or row.get("AUTHOR"),
                    "rating": row.get("emRatingName") or row.get("rating") or row.get("RATING"),
                    "industry": row.get("industryName") or row.get("INDUSTRY_NAME"),
                    "url": row.get("url") or row.get("attachUrl") or row.get("INFO_CODE"),
                }
            )
        if not reports:
            raise ValueError("eastmoney research reports returned empty data")
        return {"stock_code": code.display, "count": len(reports), "reports": reports}

    def _research_reports_akshare_payload(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        df = self._require_df(ak.stock_research_report_em(symbol=code.code), "stock_research_report_em")
        records = []
        for _, row in df.head(limit).iterrows():
            records.append(
                {
                    "title": row.get("报告名称") or row.get("标题") or row.get("title"),
                    "org_name": row.get("研究机构") or row.get("机构") or row.get("orgName"),
                    "publish_date": row.get("发布日期") or row.get("日期") or row.get("publishDate"),
                    "author": row.get("研究员") or row.get("作者") or row.get("author"),
                    "rating": row.get("最新评级") or row.get("评级") or row.get("emRatingName"),
                    "industry": row.get("行业") or row.get("industryName"),
                    "url": row.get("报告链接") or row.get("url") or row.get("attachUrl"),
                }
            )
        if not records:
            raise ValueError("akshare stock_research_report_em returned empty data")
        return {"stock_code": code.display, "count": len(records), "reports": records}

    def _research_reports_scrapling_payload(self, code: NormalizedStockCode, limit: int) -> dict[str, Any]:
        """Scrapling fallback for research reports"""
        try:
            from scrapling.fetchers import Fetcher
            end = self._now_cn().date()
            begin = end - timedelta(days=365)
            page = Fetcher.get(
                "https://reportapi.eastmoney.com/report/list",
                params={
                    "pageSize": str(limit),
                    "beginTime": begin.isoformat(),
                    "endTime": end.isoformat(),
                    "pageNo": "1",
                    "qType": "0",
                    "code": code.code,
                },
                headers={"Referer": "https://data.eastmoney.com/"},
                timeout=5, retries=1, stealthy_headers=True,
            )
            if page.status != 200 or not page.body:
                raise ValueError(f"scrapling reports HTTP {page.status}")
            import json as _json
            raw = _json.loads(page.body.decode("utf-8", errors="replace"))
            rows = raw.get("data") or raw.get("Data") or []
            reports = []
            for row in rows[:limit]:
                reports.append({
                    "title": row.get("title") or row.get("TITLE"),
                    "org_name": row.get("orgName") or row.get("ORG_NAME"),
                    "publish_date": row.get("publishDate") or row.get("PUBLISH_DATE"),
                    "author": row.get("author") or row.get("AUTHOR"),
                    "rating": row.get("emRatingName") or row.get("rating") or row.get("RATING"),
                    "industry": row.get("industryName") or row.get("INDUSTRY_NAME"),
                    "url": row.get("url") or row.get("attachUrl") or row.get("INFO_CODE"),
                })
            if not reports:
                raise ValueError("scrapling reports returned empty")
            return {"stock_code": code.display, "count": len(reports), "reports": reports}
        except ImportError:
            raise ValueError("scrapling not installed")
        except Exception as e:
            raise ValueError(f"scrapling reports failed: {e}")

    def _f10_company_payload(self, code: NormalizedStockCode) -> dict[str, Any]:
        raw = self._request_json(
            "https://emweb.securities.eastmoney.com/PC_HSF10/CompanySurvey/PageAjax",
            headers={"Referer": "https://emweb.securities.eastmoney.com/"},
            params={"code": self._eastmoney_f10_code(code)},
        )
        if not raw:
            raise ValueError("eastmoney f10 company returned empty data")
        return {"stock_code": code.display, "company": raw}

    def _fund_daily_payload(self, df: pd.DataFrame, code: str, days: int, key: str) -> dict[str, Any]:
        df = self._require_df(df, "fund_daily")
        records = dataframe_to_records(df.tail(days), days)
        return {"fund_code": code, "days": len(records), key: records}

    def _fund_info_payload(self, df: pd.DataFrame, code: str, fund_type: str, limit: int) -> dict[str, Any]:
        df = self._require_df(df, "fund_info")
        records = dataframe_to_records(df.tail(limit), limit)
        return {"fund_code": code, "fund_type": fund_type, "count": len(records), "records": records}

    def _fund_nav_direct_payload(self, code: str, fund_type: str, limit: int) -> dict[str, Any]:
        raw = self._request_json(
            "https://api.fund.eastmoney.com/f10/lsjz",
            headers={"Referer": f"https://fundf10.eastmoney.com/jjjz_{code}.html"},
            params={
                "fundCode": code,
                "pageIndex": "1",
                "pageSize": str(min(max(int(limit), 1), 500)),
                "startDate": "",
                "endDate": "",
                "_": str(int(self._now_cn().timestamp() * 1000)),
            },
        )
        rows = ((raw.get("Data") or {}).get("LSJZList") or [])[:limit]
        records = []
        for row in rows:
            records.append(
                {
                    "date": row.get("FSRQ"),
                    "unit_nav": self._float(row.get("DWJZ")),
                    "accumulated_nav": self._float(row.get("LJJZ")),
                    "daily_growth_rate": self._float(row.get("JZZZL")),
                    "actual_yield": self._float(row.get("ACTUALSYI")),
                    "subscribe_status": row.get("SGZT"),
                    "redeem_status": row.get("SHZT"),
                    "nav_type": row.get("NAVTYPE"),
                }
            )
        if not records:
            raise ValueError("eastmoney fund nav returned empty data")
        return {
            "fund_code": code,
            "fund_type": fund_type,
            "count": len(records),
            "total": raw.get("TotalCount"),
            "records": records,
        }

    def _fund_name_direct_payload(self, limit: int) -> dict[str, Any]:
        text = self._request_text(
            "https://fund.eastmoney.com/js/fundcode_search.js",
            headers={"Referer": "https://fund.eastmoney.com/"},
            encoding="utf-8-sig",
        ).strip()
        prefix = "var r = "
        if text.startswith(prefix):
            text = text[len(prefix):]
        if text.endswith(";"):
            text = text[:-1]
        rows = json.loads(text)
        funds = []
        for row in rows[:limit]:
            if len(row) < 5:
                continue
            funds.append(
                {
                    "基金代码": row[0],
                    "拼音缩写": row[1],
                    "基金简称": row[2],
                    "基金类型": row[3],
                    "拼音全称": row[4],
                }
            )
        if not funds:
            raise ValueError("eastmoney fundcode search returned empty data")
        return {"count": len(funds), "funds": funds}

    def _fund_money_spot_direct(self, limit: int) -> dict[str, Any]:
        text = self._request_text(
            "https://fund.eastmoney.com/HBJJ_pjsyl.html",
            headers={"Referer": "https://fund.eastmoney.com/"},
            encoding="gb2312",
            timeout=5,
        )
        tables = pd.read_html(StringIO(text))
        if len(tables) < 2:
            raise ValueError("eastmoney money fund page returned no data table")
        raw = tables[1]
        if raw.shape[0] < 3 or raw.shape[1] < 16:
            raise ValueError("eastmoney money fund table shape unexpected")

        show_day = [str(value) for value in raw.iloc[0, 5:11].tolist()]
        df = raw.iloc[2:, [3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]].copy()
        df.columns = [
            "基金代码",
            "基金简称",
            f"{show_day[0]}-万份收益",
            f"{show_day[1]}-7日年化%",
            f"{show_day[2]}-单位净值",
            f"{show_day[3]}-万份收益",
            f"{show_day[4]}-7日年化%",
            f"{show_day[5]}-单位净值",
            "日涨幅",
            "成立日期",
            "基金经理",
            "手续费",
            "可购全部",
        ]
        df["基金简称"] = df["基金简称"].astype(str).str.replace("基金吧档案", "", regex=False).str.strip()
        records = dataframe_to_records(df, limit)
        if not records:
            raise ValueError("eastmoney money fund table returned empty data")
        return {"count": len(records), "funds": records}

    def _hk_short_selling_direct_payload(self, code: str, limit: int, pages: int) -> dict[str, Any]:
        records: list[dict[str, Any]] = []
        seen: set[tuple[str | None, str | None]] = set()
        last_url = None
        for page in range(1, pages + 1):
            params = {"code": code}
            if page > 1:
                params["page"] = str(page)
            last_url = f"{EASTMONEY_HK_SELLSHORT_URL}?{urlencode(params)}"
            text = self._request_text(
                last_url,
                headers={"Referer": "https://hk.eastmoney.com/", "Accept": "text/html,application/xhtml+xml"},
                encoding="utf-8",
                timeout=8,
            )
            tables = pd.read_html(StringIO(text))
            if not tables:
                break
            table = tables[0]
            if table.empty:
                break
            for _, row in table.iterrows():
                row_code = self._hk_stock_code(row.get("股票代码"))
                date = self._text_or_none(row.get("日期"))
                key = (row_code, date)
                if key in seen:
                    continue
                seen.add(key)
                raw_short_amount = row.get("沽空金额(港元)")
                raw_total_turnover = row.get("总成交金额(港元)")
                raw_ratio = row.get("沽空占成交比例")
                records.append(
                    {
                        "stock_code": row_code,
                        "stock_name": self._text_or_none(row.get("股票名称")),
                        "latest_price": self._float(row.get("最新价")),
                        "short_volume": self._float(row.get("沽空数量(股)")),
                        "short_avg_price": self._float(row.get("沽空平均价")),
                        "short_amount_hkd": self._amount_value(raw_short_amount),
                        "total_turnover_hkd": self._amount_value(raw_total_turnover),
                        "short_turnover_ratio_pct": self._float(raw_ratio),
                        "date": date,
                        "raw_short_amount": self._text_or_none(raw_short_amount),
                        "raw_total_turnover": self._text_or_none(raw_total_turnover),
                        "raw_short_turnover_ratio": self._text_or_none(raw_ratio),
                    }
                )
                if len(records) >= limit:
                    break
            if len(records) >= limit:
                break
        if not records:
            raise ValueError(f"eastmoney hk sellshort returned empty data: {code}")
        return {
            "stock_code": code,
            "count": len(records),
            "pages_requested": pages,
            "source_url": last_url or EASTMONEY_HK_SELLSHORT_URL,
            "records": records[:limit],
        }

    def _market_breadth_direct(self) -> dict[str, Any]:
        # EastMoney caps clist page size to 100. Count positive/negative
        # stocks by binary-searching sorted change_pct pages instead of
        # downloading every page.
        page_size = 100
        page_cache: dict[tuple[int, str], tuple[list[dict[str, Any]], int]] = {}

        def fetch_page(page: int, order: str) -> tuple[list[dict[str, Any]], int]:
            key = (page, order)
            if key in page_cache:
                return page_cache[key]
            raw = self._request_json(
                "https://push2delay.eastmoney.com/api/qt/clist/get",
                headers={"Referer": "https://quote.eastmoney.com/"},
                params={
                    "np": "1",
                    "fltt": "2",
                    "invt": "2",
                    "po": order,
                    "fid": "f3",
                    "pn": str(page),
                    "pz": str(page_size),
                    "fs": "m:0 t:6,m:0 t:80,m:1 t:2,m:1 t:23",
                    "fields": "f12,f3",
                    "ut": "8dec03ba335b81bf4ebdf7b29ec27d15",
                },
            )
            data = raw.get("data") or {}
            batch = data.get("diff") or []
            raw_total = data.get("total")
            total = int(raw_total) if raw_total is not None else 0
            page_cache[key] = (batch, total)
            return page_cache[key]

        def count_ordered(order: str, predicate: Callable[[float], bool]) -> tuple[int, int]:
            first_rows, total = fetch_page(1, order)
            if not first_rows or not total:
                return 0, total
            pages = math.ceil(total / page_size)
            lo, hi = 1, pages
            last_full_match = 0
            while lo <= hi:
                mid = (lo + hi) // 2
                rows, total = fetch_page(mid, order)
                values = [
                    value
                    for value in (self._float(row.get("f3")) for row in rows)
                    if value is not None
                ]
                if not values:
                    hi = mid - 1
                    continue
                if all(predicate(value) for value in values):
                    last_full_match = mid
                    lo = mid + 1
                    continue
                if predicate(values[0]):
                    before = (mid - 1) * page_size
                    return min(before + sum(1 for value in values if predicate(value)), total), total
                hi = mid - 1
            return min(last_full_match * page_size, total), total

        up_count, total = count_ordered("1", lambda value: value > 0)
        down_count, total_from_down = count_ordered("0", lambda value: value < 0)
        total_count = total or total_from_down
        if not total_count:
            raise ValueError("eastmoney market breadth returned empty data")
        flat_count = max(int(total_count) - up_count - down_count, 0)
        return {
            "date": self._now_cn().strftime("%Y%m%d"),
            "up_count": up_count,
            "down_count": down_count,
            "flat_count": flat_count,
            "total_count": int(total_count),
            "turnover": None,
            "source_note": "EastMoney stock-level breadth counted by binary search over sorted change_pct pages",
        }

    def _market_breadth_from_industry_summary(self) -> dict[str, Any]:
        df = self._require_df(ak.stock_board_industry_summary_ths(), "stock_board_industry_summary_ths")
        for col in ("上涨家数", "下跌家数", "总成交额", "涨跌幅"):
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors="coerce")
        data: dict[str, Any] = {
            "date": self._now_cn().strftime("%Y%m%d"),
            "up_count": int(df["上涨家数"].sum()) if "上涨家数" in df.columns else None,
            "down_count": int(df["下跌家数"].sum()) if "下跌家数" in df.columns else None,
            "turnover": self._float(df["总成交额"].sum()) if "总成交额" in df.columns else None,
            "industry_up": int((df["涨跌幅"] > 0).sum()) if "涨跌幅" in df.columns else None,
            "industry_down": int((df["涨跌幅"] < 0).sum()) if "涨跌幅" in df.columns else None,
        }
        return data

    def _market_breadth_from_spot_em(self) -> dict[str, Any]:
        df = self._require_df(ak.stock_zh_a_spot_em(), "stock_zh_a_spot_em")
        df["涨跌幅"] = pd.to_numeric(df.get("涨跌幅"), errors="coerce")
        amount = pd.to_numeric(df.get("成交额"), errors="coerce") if "成交额" in df.columns else pd.Series(dtype=float)
        return {
            "date": self._now_cn().strftime("%Y%m%d"),
            "up_count": int((df["涨跌幅"] > 0).sum()),
            "down_count": int((df["涨跌幅"] < 0).sum()),
            "flat_count": int((df["涨跌幅"] == 0).sum()),
            "turnover": self._float(amount.sum() / 100000000) if not amount.empty else None,
        }

    def _market_temperature_payload(self, trade_date: str) -> dict[str, Any]:
        breadth: dict[str, Any] = {}
        try:
            breadth = self._market_breadth_from_industry_summary()
        except Exception:
            try:
                breadth = self._market_breadth_from_spot_em()
            except Exception:
                breadth = {}
        pool_status: dict[str, Any] = {}

        def _pool_count(side: str, fetcher: Callable[[], pd.DataFrame]) -> int:
            try:
                payload = self._eastmoney_limit_pool_payload(trade_date, side, 1)
                count = payload.get("total")
                if count is None:
                    count = payload.get("count")
                pool_status[side] = {"source": "eastmoney.push2ex.limit_pool", "ok": True}
                return int(count or 0)
            except Exception as direct_exc:
                try:
                    df = fetcher()
                    count = len(df) if df is not None else 0
                    pool_status[side] = {
                        "source": "akshare.limit_pool",
                        "ok": True,
                        "direct_error": f"{type(direct_exc).__name__}: {direct_exc}",
                    }
                    return count
                except Exception as ak_exc:
                    pool_status[side] = {
                        "source": "limit_pool",
                        "ok": False,
                        "direct_error": f"{type(direct_exc).__name__}: {direct_exc}",
                        "akshare_error": f"{type(ak_exc).__name__}: {ak_exc}",
                    }
                    return 0

        lu_count = _pool_count("up", lambda: ak.stock_zt_pool_em(date=trade_date))
        ld_count = _pool_count("down", lambda: ak.stock_zt_pool_dtgc_em(date=trade_date))
        if lu_count > 80 and ld_count < 20:
            rating = "火热"
        elif lu_count > 50 and ld_count < 30:
            rating = "偏热"
        elif lu_count > 30 and ld_count < 50:
            rating = "温和"
        elif lu_count > 15:
            rating = "偏冷"
        else:
            rating = "冰冷"
        return {
            "date": trade_date,
            "limit_up_count": lu_count,
            "limit_down_count": ld_count,
            "breadth": breadth,
            "temperature": rating,
            "pool_sources": pool_status,
        }

    def _news_payload(self, df: pd.DataFrame, limit: int, key: str, stock_code: str | None = None) -> dict[str, Any]:
        if df is None:
            raise ValueError("news source returned None")
        records = dataframe_to_records(df, limit)
        payload = {"count": len(records), key: records}
        if stock_code:
            payload["stock_code"] = stock_code
        return payload

    def _notice_payload(self, df: pd.DataFrame, stock_code: str, limit: int) -> dict[str, Any]:
        if df is None:
            raise ValueError("notice source returned None")
        records = dataframe_to_records(df, limit)
        return {"stock_code": stock_code, "count": len(records), "notices": records}

    def _financial_df_payload(self, df: pd.DataFrame, stock_code: str, kind: str, limit: int) -> dict[str, Any]:
        df = self._require_df(df, kind)
        records = dataframe_to_records(df.tail(limit), limit)
        return {"stock_code": stock_code, "kind": kind, "count": len(records), "records": records}

    def _financial_recent_filtered_payload(
        self,
        fetcher: Callable[[str], pd.DataFrame],
        code: NormalizedStockCode,
        kind: str,
        limit: int,
        report_dates: list[str],
        optional: bool = False,
    ) -> dict[str, Any]:
        checked: list[str] = []
        last_payload: dict[str, Any] | None = None
        last_error: Exception | None = None
        for report_date in report_dates:
            checked.append(report_date)
            try:
                if optional:
                    payload = self._financial_optional_filtered_payload(
                        lambda date=report_date: fetcher(date),
                        code,
                        kind,
                        limit,
                        report_date,
                    )
                else:
                    payload = self._financial_filtered_payload(fetcher(report_date), code, kind, limit, report_date)
            except Exception as exc:
                last_error = exc
                continue
            payload["report_dates_checked"] = checked.copy()
            if int(payload.get("count") or 0) > 0:
                return payload
            last_payload = payload
        if last_payload is not None:
            last_payload["report_dates_checked"] = checked
            last_payload["source_note"] = (
                f"{kind} source returned no rows for stock={code.display}; "
                f"checked report_dates={','.join(checked)}"
            )
            return last_payload
        if last_error is not None:
            raise last_error
        return {
            "stock_code": code.display,
            "kind": kind,
            "report_date": report_dates[0] if report_dates else "",
            "report_dates_checked": checked,
            "count": 0,
            "records": [],
            "source_note": f"{kind} source returned no rows",
        }

    def _financial_optional_filtered_payload(
        self,
        fetcher: Callable[[], pd.DataFrame],
        code: NormalizedStockCode,
        kind: str,
        limit: int,
        report_date: str,
    ) -> dict[str, Any]:
        try:
            df = fetcher()
        except TypeError as exc:
            if "NoneType" not in str(exc):
                raise
            return {
                "stock_code": code.display,
                "kind": kind,
                "report_date": report_date,
                "count": 0,
                "records": [],
                "source_note": f"{kind} source returned no rows for report_date={report_date}",
            }
        if df is None or (isinstance(df, pd.DataFrame) and df.empty):
            return {
                "stock_code": code.display,
                "kind": kind,
                "report_date": report_date,
                "count": 0,
                "records": [],
                "source_note": f"{kind} source returned empty data for report_date={report_date}",
            }
        return self._financial_filtered_payload(df, code, kind, limit, report_date)

    def _financial_filtered_payload(
        self,
        df: pd.DataFrame,
        code: NormalizedStockCode,
        kind: str,
        limit: int,
        report_date: str,
    ) -> dict[str, Any]:
        df = self._require_df(df, kind)
        code_cols = [col for col in df.columns if "代码" in str(col)]
        if code_cols:
            mask = pd.Series([False] * len(df), index=df.index)
            for col in code_cols:
                mask = mask | (df[col].astype(str).str.zfill(6) == code.code)
            df = df[mask]
        records = dataframe_to_records(df, limit)
        return {"stock_code": code.display, "kind": kind, "report_date": report_date, "count": len(records), "records": records}

    def _format_board_flow_text(self, df: pd.DataFrame, title: str, limit: int) -> str:
        df = self._require_df(df, title).copy()
        for col in ("行业-涨跌幅", "流入资金", "流出资金", "净额", "公司家数", "领涨股-涨跌幅"):
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors="coerce")
        sort_col = "净额" if "净额" in df.columns else ("行业-涨跌幅" if "行业-涨跌幅" in df.columns else df.columns[0])
        df_sorted = df.sort_values(by=sort_col, ascending=False)
        today = self._now_cn().strftime("%Y-%m-%d %H:%M:%S")
        text = f"{title}\n数据获取时间: {today}\n\n"
        text += f"{'排名':<6} {'板块/行业':<18} {'涨跌幅':<10} {'净流入(亿)':<14} {'领涨股':<12} {'领涨幅':<8}\n"
        text += "-" * 78 + "\n"
        for idx, (_, row) in enumerate(df_sorted.head(limit).iterrows(), 1):
            name = str(row.get("行业", row.get("板块名称", "N/A")))
            chg = self._float(row.get("行业-涨跌幅", row.get("涨跌幅")), 0) or 0
            net = self._float(row.get("净额", row.get("主力净流入-净额")), 0) or 0
            leader = str(row.get("领涨股", "N/A"))
            leader_chg = self._float(row.get("领涨股-涨跌幅"), 0) or 0
            text += f"{idx:<6} {name:<18} {chg:<+10.2f} {net:<14.2f} {leader:<12} {leader_chg:<+.2f}%\n"
        text += f"\n【数据来源：{title} / AKShare 在线接口】\n"
        return text

    def _format_temperature_text(self, payload: dict[str, Any]) -> str:
        breadth = payload.get("breadth") or {}
        text = f"A股市场温度数据（数据日期：{payload.get('date')}）\n\n"
        text += "=== 市场温度计 ===\n"
        text += f"涨停家数: {payload.get('limit_up_count', 0)} 只\n"
        text += f"跌停家数: {payload.get('limit_down_count', 0)} 只\n"
        if breadth:
            text += f"上涨家数: {breadth.get('up_count', 0)} 只\n"
            text += f"下跌家数: {breadth.get('down_count', 0)} 只\n"
            if breadth.get("turnover") is not None:
                text += f"成交额: {breadth.get('turnover'):.2f} 亿\n"
        text += f"\n温度评级: {payload.get('temperature', '未知')}\n"
        text += "\n【数据来源：涨跌停池 + 市场宽度在线接口】\n"
        return text

    def _format_legacy_news(self, stock_code: str, data_type: str, limit: int) -> str:
        if data_type == "notice":
            payload = self.stock_notices(stock_code, limit).get("data") or {}
            records = payload.get("notices") or []
            text = f"股票 {stock_code} 最新公告（前{len(records)}条）：\n\n"
            for row in records:
                text += f"{row.get('公告日期', row.get('date', ''))} {row.get('公告标题', row.get('title', ''))}\n"
            return text
        if data_type in {"news"}:
            payload = self.stock_news(stock_code, limit).get("data") or {}
            records = payload.get("news") or []
            text = f"股票 {stock_code} 最新资讯（前{len(records)}条）：\n\n"
        else:
            payload = self.global_news(limit).get("data") or {}
            records = payload.get("news") or []
            text = f"财经新闻（前{len(records)}条）：\n\n"
        keywords = ["重要", "利好", "重磅", "突发", "关注", "涨停", "跌停", "重组", "并购", "业绩"]
        if data_type in {"important_news", "hotspot_news"}:
            records = [
                row for row in records
                if any(kw in f"{row.get('标题', row.get('title', ''))} {row.get('内容', row.get('content', row.get('摘要', '')))}" for kw in keywords)
            ][:limit]
        for idx, row in enumerate(records[:limit], 1):
            title = row.get("标题", row.get("新闻标题", row.get("title", "")))
            content = row.get("内容", row.get("新闻内容", row.get("摘要", row.get("content", ""))))
            time_value = row.get("发布时间", row.get("新闻发布时间", row.get("time", "")))
            text += f"【{idx}】{title}\n"
            if time_value:
                text += f"   时间：{time_value}\n"
            if content:
                text += f"   内容：{str(content)[:200]}{'...' if len(str(content)) > 200 else ''}\n"
            text += "\n"
        return text

    def _format_sector_ambush(self, df: pd.DataFrame, lookback_days: int, limit: int) -> str:
        df = self._require_df(df, "stock_fund_flow_concept").copy()
        for col in ("行业-涨跌幅", "净额", "领涨股-涨跌幅"):
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors="coerce")
        candidates = []
        for _, row in df.head(100).iterrows():
            name = str(row.get("行业", "")).strip()
            chg = self._float(row.get("行业-涨跌幅"), 0) or 0
            net = self._float(row.get("净额"), 0) or 0
            leader = str(row.get("领涨股", "")).strip()
            leader_chg = self._float(row.get("领涨股-涨跌幅"), 0) or 0
            if net < 0 and -1 < chg < 1:
                reason = "资金流出但跌幅收窄，可能筑底"
            elif 0 < net < 5 and chg < 1.5:
                reason = "资金小幅流入，可能在吸筹"
            else:
                continue
            candidates.append((name, chg, net, leader, leader_chg, reason))
            if len(candidates) >= limit:
                break
        text = f"🟦 低位埋伏板块分析（近{lookback_days}日回溯）\n\n"
        text += "| 板块名称 | 当日涨幅 | 净流入(亿) | 领涨股 | 领涨幅 | 埋伏逻辑 |\n"
        text += "|----------|----------|------------|--------|--------|----------|\n"
        if not candidates:
            text += "| 暂无 | - | - | - | - | 当前暂无满足条件的低位资金异动板块 |\n"
        for name, chg, net, leader, leader_chg, reason in candidates:
            text += f"| {name} | {chg:+.2f}% | {net:.2f} | {leader} | {leader_chg:+.2f}% | {reason} |\n"
        return text

    def _format_sector_leaders(self, df: pd.DataFrame, sector_name: str, lookback_days: int, top_n: int) -> str:
        matched = self._match_concept_rows(df, sector_name)
        today = self._now_cn().strftime("%Y-%m-%d")
        if matched.empty:
            return f"❌ 未找到板块 '{sector_name}' 的数据。请检查板块名称是否正确。"
        leaders = []
        seen = set()
        matched["领涨股-涨跌幅"] = pd.to_numeric(matched.get("领涨股-涨跌幅"), errors="coerce")
        for _, row in matched.sort_values("领涨股-涨跌幅", ascending=False).iterrows():
            name = str(row.get("领涨股", "")).strip()
            if not name or name in seen:
                continue
            seen.add(name)
            leaders.append(row)
            if len(leaders) >= top_n:
                break
        text = f"🏆 【{sector_name}】板块龙头股分析报告\n"
        text += f"📅 数据基准日期: {today}\n📊 分析区间: 最近{lookback_days}个交易日\n\n"
        text += "| 排名 | 股票名称 | 所属概念 | 概念涨幅 | 领涨幅 | 概念净流入(亿) |\n"
        text += "|------|----------|----------|----------|--------|----------------|\n"
        for idx, row in enumerate(leaders, 1):
            text += (
                f"| {idx} | {row.get('领涨股', '')} | {row.get('行业', '')} | "
                f"{self._float(row.get('行业-涨跌幅'), 0):+.2f}% | "
                f"{self._float(row.get('领涨股-涨跌幅'), 0):+.2f}% | "
                f"{self._float(row.get('净额'), 0):.2f} |\n"
            )
        return text

    def _format_leader_frequency(self, df: pd.DataFrame, stock_name: str) -> str:
        details = []
        for _, row in df.iterrows():
            leader = str(row.get("领涨股", "")).strip()
            if leader == stock_name or (len(stock_name) >= 2 and stock_name in leader):
                details.append(row)
        today = self._now_cn().strftime("%Y-%m-%d")
        if not details:
            return f"📊 【{stock_name}】多概念龙头分析报告\n📅 数据基准日期: {today}\n\n该股票当前未在任何概念中被识别为领涨股。"
        concepts = {str(row.get("行业", "")).strip() for row in details if str(row.get("行业", "")).strip()}
        text = f"👑 【{stock_name}】多概念龙头分析报告\n"
        text += f"📅 数据基准日期: {today}\n🔥 龙头出现总频次: {len(details)} 次\n🌐 覆盖概念数量: {len(concepts)} 个\n\n"
        text += "| 概念名称 | 概念涨幅 | 领涨股涨幅 | 净流入(亿) |\n"
        text += "|----------|----------|------------|------------|\n"
        for row in details[:30]:
            text += (
                f"| {row.get('行业', '')} | {self._float(row.get('行业-涨跌幅'), 0):+.2f}% | "
                f"{self._float(row.get('领涨股-涨跌幅'), 0):+.2f}% | {self._float(row.get('净额'), 0):.2f} |\n"
            )
        return text

    def _match_concept_rows(self, df: pd.DataFrame, sector_name: str) -> pd.DataFrame:
        keywords = {sector_name, f"{sector_name}概念", f"{sector_name}板块"}
        alias = {
            "AI": "人工智能",
            "芯片": "芯片",
            "半导体": "半导体",
            "新能源": "新能源",
            "机器人": "机器人",
            "CPO": "CPO",
            "算力": "算力",
        }
        if sector_name in alias:
            keywords.add(alias[sector_name])
        mask = pd.Series([False] * len(df), index=df.index)
        for keyword in keywords:
            mask = mask | df["行业"].astype(str).str.contains(keyword, na=False)
        return df[mask].copy()

    def _require_df(self, df: pd.DataFrame | None, label: str) -> pd.DataFrame:
        if df is None:
            raise ValueError(f"{label} returned None")
        if df.empty:
            raise ValueError(f"{label} returned empty DataFrame")
        return df

    def _row_count(self, payload: Any) -> int | None:
        if isinstance(payload, list):
            return len(payload)
        if isinstance(payload, dict):
            for key in (
                "records",
                "data",
                "daily",
                "stocks",
                "concepts",
                "industries",
                "items",
                "etfs",
                "funds",
                "indices",
                "reports",
                "news",
                "notices",
            ):
                value = payload.get(key)
                if isinstance(value, list):
                    return len(value)
            count = payload.get("count")
            if isinstance(count, int):
                return count
        return None

    def _text_or_none(self, value: Any) -> str | None:
        if value is None:
            return None
        try:
            if pd.isna(value):
                return None
        except (TypeError, ValueError):
            pass
        text = str(value).strip()
        if not text or text.lower() in {"nan", "none", "nat"} or text in {"-", "--"}:
            return None
        if text.endswith(".0") and text[:-2].isdigit():
            return text[:-2]
        return text

    def _float(self, value: Any, default: float | None = None) -> float | None:
        if isinstance(value, str):
            value = value.strip().replace(",", "")
            if value.endswith("%"):
                value = value[:-1]
        try:
            out = float(value)
        except (TypeError, ValueError):
            return default
        return out if math.isfinite(out) else None

    def _amount_value(self, value: Any, default: float | None = None) -> float | None:
        if value is None:
            return default
        if isinstance(value, (int, float)):
            return self._float(value, default)
        text = str(value).strip().replace(",", "")
        if not text or text in {"-", "--", "None", "nan"}:
            return default
        multiplier = 1.0
        if text.endswith("亿"):
            multiplier = 100000000.0
            text = text[:-1]
        elif text.endswith("万"):
            multiplier = 10000.0
            text = text[:-1]
        value_float = self._float(text, default)
        return round(value_float * multiplier, 4) if value_float is not None else default

    def _last(self, series: pd.Series) -> float | None:
        if series.empty:
            return None
        return self._float(series.iloc[-1])

    def _sum_col(self, df: pd.DataFrame, col: str) -> float:
        if col not in df.columns:
            return 0.0
        return float(pd.to_numeric(df[col], errors="coerce").fillna(0).sum())

    def _eastmoney_clist(
        self,
        fs: str,
        fields: str,
        limit: int,
        fid: str,
        po: str,
        referer: str,
        ut: str = "8dec03ba335b81bf4ebdf7b29ec27d15",
        base_url: str = "https://push2.eastmoney.com",
        timeout: float | None = None,
    ) -> dict[str, Any]:
        return self._request_json(
            f"{base_url.rstrip('/')}/api/qt/clist/get",
            headers={"Referer": referer},
            params={
                "np": "1",
                "fltt": "2",
                "invt": "2",
                "po": po,
                "fid": fid,
                "pn": "1",
                "pz": limit,
                "fs": fs,
                "fields": fields,
                "ut": ut,
            },
            timeout=timeout,
        )

    def _normalize_board_category(self, category: str) -> str:
        text = str(category or "").strip().lower()
        if text in {"concept", "concepts", "gn", "theme"}:
            return "concept"
        if text in {"region", "area", "province", "diqu"}:
            return "region"
        return "industry"

    def _board_flow_fs(self, category: str) -> str:
        if category == "concept":
            return "m:90 t:3"
        if category == "region":
            return "m:90 t:1"
        return "m:90 s:4"

    def _normalize_stock_move_type(self, move_type: str) -> str:
        text = str(move_type or "").strip().lower()
        if text in {"drop", "speed_down", "speeddown", "rapid_down", "down"}:
            return "drop"
        if text in {"change_up", "rise", "up_change"}:
            return "change_up"
        if text in {"change_down", "fall", "down_change"}:
            return "change_down"
        if text in {"mainflow", "fund", "capital"}:
            return "mainflow"
        if text in {"turnover", "activity", "active"}:
            return "turnover"
        return "surge"

    def _stock_move_sort(self, move_type: str) -> tuple[str, str]:
        if move_type == "drop":
            return "f22", "0"
        if move_type == "change_up":
            return "f3", "1"
        if move_type == "change_down":
            return "f3", "0"
        if move_type == "mainflow":
            return "f62", "1"
        if move_type == "turnover":
            return "f8", "1"
        return "f22", "1"

    def _normalize_fund_flow_interval(self, interval: str) -> str:
        text = str(interval or "").strip().lower()
        mapping = {
            "1": "1",
            "1m": "1",
            "1min": "1",
            "min": "1",
            "5": "5",
            "5m": "5",
            "15": "15",
            "15m": "15",
            "30": "30",
            "30m": "30",
            "60": "60",
            "60m": "60",
            "101": "101",
            "1d": "101",
            "day": "101",
            "daily": "101",
        }
        return mapping.get(text, "1")

    def _stock_secid(self, code: NormalizedStockCode) -> str:
        market = 1 if code.market == "sh" else 0
        return f"{market}.{code.code}"

    def _display_code_from_digits(self, code: str) -> str:
        if code.startswith(("600", "601", "603", "605", "688")):
            return f"{code}.SH"
        if code.startswith(("920", "8", "4")):
            return f"{code}.BJ"
        return f"{code}.SZ"

    # ---- Yahoo Finance helpers ----

    def _to_yahoo_ticker(self, code: NormalizedStockCode) -> str:
        """Convert internal stock code to Yahoo Finance ticker format."""
        if code.market == "sh":
            return f"{code.code}.SS"
        return f"{code.code}.SZ"

    def _yahoo_index_ticker(self, code: str) -> str:
        """Convert index code to Yahoo Finance ticker format."""
        raw = str(code or "").strip().lower()
        value = raw
        if raw.startswith(("sh", "sz")):
            value = raw[2:]
        elif "." in raw:
            value = raw.split(".", 1)[0]
        if value.startswith(("000", "880")):
            return f"{value}.SS"
        if value.startswith("399"):
            return f"{value}.SZ"
        return f"{value}.SS"

    def _yahoo_etf_ticker(self, code: NormalizedStockCode) -> str:
        """Convert ETF code to Yahoo Finance ticker format."""
        if code.market == "sh":
            return f"{code.code}.SS"
        return f"{code.code}.SZ"

    def _yahoo_chart(
        self,
        ticker: str,
        range_str: str = "5d",
        interval: str = "1d",
        period1: int | None = None,
        period2: int | None = None,
    ) -> dict[str, Any]:
        """Fetch chart data from Yahoo Finance."""
        params: dict[str, Any] = {"interval": interval}
        if period1 is not None and period2 is not None:
            params["period1"] = period1
            params["period2"] = period2
        else:
            params["range"] = range_str
        url = f"https://query1.finance.yahoo.com/v8/finance/chart/{ticker}?{urlencode(params)}"
        response = self._http_get(url, headers={"User-Agent": "Mozilla/5.0"})
        response.raise_for_status()
        return response.json()

    def _stock_daily_yahoo(
        self,
        code: NormalizedStockCode,
        days: int,
        adjust: str,
        start: str | None = None,
        end: str | None = None,
    ) -> dict[str, Any]:
        """Stock daily K-line from Yahoo Finance."""
        ticker = self._to_yahoo_ticker(code)
        # Yahoo range: 1d,5d,1mo,3mo,6mo,1y,2y,5y,10y,ytd
        period1 = None
        period2 = None
        if start and end:
            start_dt = datetime.strptime(start, "%Y%m%d").replace(tzinfo=CN_TZ)
            end_dt = datetime.strptime(end, "%Y%m%d").replace(tzinfo=CN_TZ) + timedelta(days=1)
            period1 = int((start_dt - timedelta(days=7)).timestamp())
            period2 = int(end_dt.timestamp())
            range_str = "5d"
        elif days <= 5:
            range_str = "5d"
        elif days <= 30:
            range_str = "1mo"
        elif days <= 90:
            range_str = "3mo"
        elif days <= 180:
            range_str = "6mo"
        elif days <= 365:
            range_str = "1y"
        elif days <= 730:
            range_str = "2y"
        else:
            range_str = "5y"

        data = self._yahoo_chart(ticker, range_str=range_str, period1=period1, period2=period2)
        result = data.get("chart", {}).get("result", [])
        if not result:
            raise ValueError(f"yahoo chart returned no data for {ticker}")

        r = result[0]
        timestamps = r.get("timestamp", [])
        quote = r.get("indicators", {}).get("quote", [{}])[0]

        closes = quote.get("close", [])
        opens = quote.get("open", [])
        highs = quote.get("high", [])
        lows = quote.get("low", [])
        volumes = quote.get("volume", [])

        records = []
        for i, ts in enumerate(timestamps):
            if i >= len(closes):
                break
            dt = datetime.fromtimestamp(ts, tz=CN_TZ)
            c = closes[i]
            if c is None:
                continue
            o = opens[i] if i < len(opens) and opens[i] is not None else c
            h = highs[i] if i < len(highs) and highs[i] is not None else c
            l = lows[i] if i < len(lows) and lows[i] is not None else c
            v = volumes[i] if i < len(volumes) and volumes[i] is not None else 0
            prev_c = closes[i - 1] if i > 0 and closes[i - 1] is not None else o
            change_pct = round((c - prev_c) / prev_c * 100, 4) if prev_c else None
            records.append({
                "date": dt.strftime("%Y-%m-%d"),
                "open": round(o, 2),
                "high": round(h, 2),
                "low": round(l, 2),
                "close": round(c, 2),
                "volume": int(v),
                "amount": None,
                "turnover": None,
                "change_pct": change_pct,
            })

        if start and end:
            start_iso = self._iso_date(start)
            end_iso = self._iso_date(end)
            records = [row for row in records if start_iso <= str(row.get("date")) <= end_iso]
        if not records:
            raise ValueError(f"yahoo chart parsed no records for {ticker}")
        records = records[-days:]
        return {"stock_code": code.display, "days": len(records), "records": records, "source": "yahoo"}

    def _stock_quote_yahoo(self, code: NormalizedStockCode) -> dict[str, Any]:
        """Stock realtime quote from Yahoo Finance."""
        ticker = self._to_yahoo_ticker(code)
        data = self._yahoo_chart(ticker, range_str="2d", interval="1d")
        result = data.get("chart", {}).get("result", [])
        if not result:
            raise ValueError(f"yahoo chart returned no data for {ticker}")

        r = result[0]
        meta = r.get("meta", {})
        timestamps = r.get("timestamp", [])
        quote = r.get("indicators", {}).get("quote", [{}])[0]

        closes = quote.get("close", [])
        opens = quote.get("open", [])
        highs = quote.get("high", [])
        lows = quote.get("low", [])
        volumes = quote.get("volume", [])

        if not closes:
            raise ValueError(f"yahoo no close data for {ticker}")

        idx = -1
        c = closes[idx]
        while c is None and abs(idx) < len(closes):
            idx -= 1
            c = closes[idx]

        if c is None:
            raise ValueError(f"yahoo all closes are None for {ticker}")

        o = opens[idx] if idx < len(opens) and opens[idx] is not None else c
        h = highs[idx] if idx < len(highs) and highs[idx] is not None else c
        l = lows[idx] if idx < len(lows) and lows[idx] is not None else c
        v = volumes[idx] if idx < len(volumes) and volumes[idx] is not None else 0

        prev_c = closes[idx - 1] if abs(idx) < len(closes) and closes[idx - 1] is not None else o
        change_pct = round((c - prev_c) / prev_c * 100, 4) if prev_c else None
        change_amt = round(c - prev_c, 4) if prev_c else None

        dt = datetime.fromtimestamp(timestamps[idx], tz=CN_TZ) if abs(idx) <= len(timestamps) else self._now_cn()

        return {
            "stock_code": code.display,
            "name": meta.get("shortName", meta.get("symbol", code.display)),
            "price": round(c, 2),
            "change_pct": change_pct,
            "change_amount": change_amt,
            "volume": int(v),
            "amount": None,
            "open": round(o, 2),
            "high": round(h, 2),
            "low": round(l, 2),
            "pre_close": round(prev_c, 2),
            "turnover": None,
            "date": dt.strftime("%Y-%m-%d"),
        }

    def _market_indices_yahoo(self, limit: int) -> dict[str, Any]:
        """Market indices from Yahoo Finance."""
        major_indices = [
            ("000001.SS", "上证指数"),
            ("399001.SZ", "深证成指"),
            ("399006.SZ", "创业板指"),
            ("000300.SS", "沪深300"),
            ("000016.SS", "上证50"),
            ("000905.SS", "中证500"),
            ("000688.SS", "科创50"),
        ]
        records = []
        for ticker, name in major_indices:
            try:
                data = self._yahoo_chart(ticker, range_str="2d")
                result = data.get("chart", {}).get("result", [])
                if not result:
                    continue
                meta = result[0].get("meta", {})
                quote = result[0].get("indicators", {}).get("quote", [{}])[0]
                closes = quote.get("close", [])
                if not closes:
                    continue
                c = closes[-1]
                prev = closes[-2] if len(closes) >= 2 and closes[-2] is not None else c
                change_pct = round((c - prev) / prev * 100, 4) if prev else None
                records.append({
                    "code": ticker,
                    "name": name,
                    "price": round(c, 2),
                    "change_pct": change_pct,
                    "volume": meta.get("regularMarketVolume"),
                })
            except Exception:
                continue
        if not records:
            raise ValueError("yahoo indices returned no data")
        return {"count": len(records), "indices": records[:limit]}

    def _index_secid(self, code: str) -> str:
        raw = str(code or "").strip().lower()
        if not raw:
            raise ValueError("index_code is required")
        if raw.startswith(("sh", "sz", "bj")):
            market = 1 if raw.startswith("sh") else 0
            value = raw[2:8]
        elif "." in raw:
            value, suffix = raw.split(".", 1)
            market = 1 if suffix.lower().startswith("sh") else 0
        else:
            value = raw[:6]
            market = 0 if value.startswith(("399", "159", "16")) else 1
        if len(value) != 6 or not value.isdigit():
            raise ValueError(f"invalid index_code: {code}")
        return f"{market}.{value}"

    def _eastmoney_f10_code(self, code: NormalizedStockCode) -> str:
        if code.market == "sh":
            return f"SH{code.code}"
        if code.market == "bj":
            return f"BJ{code.code}"
        return f"SZ{code.code}"

    def _eastmoney_time(self, value: Any) -> str | None:
        try:
            ts = int(value)
        except (TypeError, ValueError):
            return None
        text = str(ts)
        try:
            if len(text) == 13:
                return datetime.fromtimestamp(ts / 1000, CN_TZ).strftime("%Y-%m-%d %H:%M:%S")
            if len(text) == 10:
                return datetime.fromtimestamp(ts, CN_TZ).strftime("%Y-%m-%d %H:%M:%S")
            if len(text) == 8:
                return f"{text[:4]}-{text[4:6]}-{text[6:8]}"
        except (OSError, ValueError):
            return None
        return None



    def _stock_quote_eastmoney_push2(self, code: NormalizedStockCode) -> dict[str, Any]:
        raw = self._request_json(
            "https://push2.eastmoney.com/api/qt/stock/get",
            headers={"Referer": "https://quote.eastmoney.com/"},
            params={
                "secid": self._stock_secid(code),
                "fields": "f43,f57,f58,f60,f169,f170,f171,f168,f46,f44,f45,f47,f48,f50",
                "ut": "fa5fd1943c7b386f172d6893dbbd1035",
            },
            timeout=8,
        )
        data = raw.get("data") or {}
        if not data:
            raise ValueError(f"eastmoney push2 quote empty: {code.display}")

        def price(v: Any) -> float | None:
            n = self._float(v)
            return None if n is None else round(n / 100.0, 2)

        def pct(v: Any) -> float | None:
            n = self._float(v)
            return None if n is None else round(n / 100.0, 4)

        price_val = price(data.get("f43"))
        pre_close = price(data.get("f60"))
        if price_val is None and pre_close is None:
            raise ValueError(f"eastmoney push2 quote missing prices: {code.display}")
        return {
            "stock_code": code.display,
            "name": data.get("f58") or code.code,
            "price": price_val,
            "change_pct": pct(data.get("f170")),
            "change_amount": price(data.get("f169")),
            "volume": self._float(data.get("f47")),
            "amount": self._float(data.get("f48")),
            "open": price(data.get("f46")),
            "high": price(data.get("f44")),
            "low": price(data.get("f45")),
            "pre_close": pre_close,
            "turnover": None,
            "volume_ratio": self._float(data.get("f50")),
            "amplitude_pct": pct(data.get("f171")),
            "date": self._now_cn().strftime("%Y-%m-%d"),
        }

    def _stock_quote_tencent(self, code: NormalizedStockCode) -> dict[str, Any]:
        text = self._request_text(
            f"https://qt.gtimg.cn/q={code.prefixed}",
            headers={"Referer": "https://gu.qq.com/", "User-Agent": HTTP_HEADERS["User-Agent"]},
            encoding="gbk",
            timeout=8,
        )
        parts = text.split('="', 1)[-1].rstrip('";\n').split("~")
        if len(parts) < 38:
            raise ValueError(f"tencent qt quote malformed: {code.display}")
        price = self._float(parts[3])
        prev_close = self._float(parts[4])
        open_price = self._float(parts[5])
        high = self._float(parts[33])
        low = self._float(parts[34])
        if price is None or prev_close is None:
            raise ValueError(f"tencent qt quote missing prices: {code.display}")
        change_amount = round(price - prev_close, 4)
        change_pct = round(change_amount / prev_close * 100, 4) if prev_close else None
        amount = self._float(parts[37])
        return {
            "stock_code": code.display,
            "name": parts[1] or code.code,
            "price": price,
            "change_pct": change_pct,
            "change_amount": change_amount,
            "volume": self._float(parts[6]),
            "amount": None if amount is None else amount * 10000,
            "open": open_price,
            "high": high,
            "low": low,
            "pre_close": prev_close,
            "turnover": None,
            "date": self._now_cn().strftime("%Y-%m-%d"),
            "time": parts[30] if len(parts) > 30 else None,
        }

    def _stock_daily_tencent(
        self,
        code: NormalizedStockCode,
        days: int,
        adjust: str,
        start: str | None = None,
        end: str | None = None,
    ) -> dict[str, Any]:
        adj = "qfq" if (adjust or "qfq") == "qfq" else "day"
        count = max(days + 10, days * 2, 30)
        raw = self._request_json(
            "https://ifzq.gtimg.cn/appstock/app/fqkline/get",
            headers={"Referer": "https://gu.qq.com/", "User-Agent": HTTP_HEADERS["User-Agent"]},
            params={"param": f"{code.prefixed},day,,,{count},{adj}"},
            timeout=10,
        )
        node = ((raw.get("data") or {}).get(code.prefixed) or {})
        klines = node.get("qfqday") or node.get("day") or []
        if not klines:
            raise ValueError(f"tencent fqkline empty: {code.display}")
        records: list[dict[str, Any]] = []
        for row in klines:
            if not isinstance(row, (list, tuple)) or len(row) < 6:
                continue
            records.append(
                {
                    "date": str(row[0]),
                    "open": self._float(row[1]),
                    "close": self._float(row[2]),
                    "high": self._float(row[3]),
                    "low": self._float(row[4]),
                    "volume": self._float(row[5]),
                    "amount": None,
                    "turnover": None,
                    "change_pct": None,
                }
            )
        for i in range(1, len(records)):
            prev = records[i - 1].get("close")
            cur = records[i].get("close")
            if prev and cur:
                records[i]["change_pct"] = round((cur - prev) / prev * 100, 4)
        if start and end:
            start_iso = self._iso_date(start)
            end_iso = self._iso_date(end)
            records = [r for r in records if start_iso <= str(r.get("date")) <= end_iso]
        if not records:
            raise ValueError(f"tencent fqkline parsed empty: {code.display}")
        records = records[-days:]
        return {"stock_code": code.display, "days": len(records), "records": records, "source": "tencent"}

    # ---- US market helpers ----

    def _normalize_us_symbol(self, symbol: str) -> str:
        raw = str(symbol or "").strip().upper()
        if not raw:
            raise ValueError("symbol is required")
        if raw.startswith("^"):
            return raw
        if raw.startswith("US"):
            raw = raw[2:]
        cleaned = "".join(ch for ch in raw if ch.isalnum() or ch in {".", "-", "_"})
        if not cleaned:
            raise ValueError(f"invalid US symbol: {symbol}")
        aliases = {
            "DJI": "^DJI",
            "DJIA": "^DJI",
            "IXIC": "^IXIC",
            "NASDAQ": "^IXIC",
            "GSPC": "^GSPC",
            "SPX": "^GSPC",
            "INX": "^GSPC",
        }
        return aliases.get(cleaned, cleaned)

    def _us_index_defs(self) -> list[dict[str, str]]:
        return [
            {"key": "dow", "tencent": "usDJI", "sina": "gb_dji", "yahoo": "^DJI", "name": "道琼斯指数"},
            {"key": "nas", "tencent": "usIXIC", "sina": "gb_ixic", "yahoo": "^IXIC", "name": "纳斯达克指数"},
            {"key": "spx", "tencent": "usINX", "sina": "gb_inx", "yahoo": "^GSPC", "name": "标普500指数"},
        ]

    def _us_sector_defs(self) -> list[dict[str, Any]]:
        return [
            {"key": "semiconductors", "symbol": "XSD", "label": "半导体", "a_share_mapping": ["半导体"]},
            {"key": "software_services", "symbol": "XSW", "label": "软件服务", "a_share_mapping": ["软件服务"]},
            {"key": "telecom", "symbol": "XTL", "label": "电信", "a_share_mapping": ["运营商"]},
            {"key": "retail", "symbol": "XRT", "label": "零售", "a_share_mapping": ["零售"]},
            {"key": "homebuilders", "symbol": "XHB", "label": "住宅建筑", "a_share_mapping": ["住宅产业链"]},
            {"key": "transportation", "symbol": "XTN", "label": "运输", "a_share_mapping": ["交通运输"]},
            {"key": "aerospace_defense", "symbol": "XAR", "label": "航空航天与国防", "a_share_mapping": ["军工"]},
            {"key": "regional_banks", "symbol": "KRE", "label": "地区银行", "a_share_mapping": ["银行"]},
            {"key": "capital_markets", "symbol": "KCE", "label": "资本市场", "a_share_mapping": ["券商"]},
            {"key": "insurance", "symbol": "KIE", "label": "保险", "a_share_mapping": ["保险"]},
            {"key": "biotechnology", "symbol": "XBI", "label": "生物科技", "a_share_mapping": ["创新药"]},
            {"key": "pharmaceuticals", "symbol": "XPH", "label": "制药", "a_share_mapping": ["制药"]},
            {"key": "healthcare_equipment", "symbol": "XHE", "label": "医疗设备", "a_share_mapping": ["医疗器械"]},
            {"key": "healthcare_services", "symbol": "XHS", "label": "医疗服务", "a_share_mapping": ["医疗服务"]},
            {"key": "oil_gas_exploration", "symbol": "XOP", "label": "油气勘探", "a_share_mapping": ["油气开采"]},
            {"key": "oil_gas_services", "symbol": "XES", "label": "油服设备", "a_share_mapping": ["油服"]},
            {"key": "metals_mining", "symbol": "XME", "label": "金属矿业", "a_share_mapping": ["金属矿业"]},
        ]

    def _us_indices_tencent_payload(self, limit: int) -> dict[str, Any]:
        defs = self._us_index_defs()
        codes = [item["tencent"] for item in defs]
        text = self._request_text(
            "https://qt.gtimg.cn/q=" + ",".join(codes),
            headers={"Referer": "https://finance.qq.com/", "User-Agent": HTTP_HEADERS["User-Agent"]},
            encoding="gbk",
            timeout=8,
        )
        by_code: dict[str, list[str]] = {}
        for line in text.strip().splitlines():
            if '="' not in line:
                continue
            left, right = line.split('="', 1)
            code = left.split("_", 1)[-1]
            parts = right.rstrip('";').split("~")
            by_code[code] = parts
        items: list[dict[str, Any]] = []
        for item in defs:
            parts = by_code.get(item["tencent"]) or []
            if len(parts) < 5:
                continue
            price = self._float(parts[3])
            prev = self._float(parts[4])
            if price is None or prev is None or prev <= 0:
                continue
            change = round(price - prev, 4)
            items.append(
                {
                    "key": item["key"],
                    "symbol": item["yahoo"],
                    "name": item["name"],
                    "price": price,
                    "prev_close": prev,
                    "change": change,
                    "change_pct": round(change / prev * 100, 4),
                    "time": parts[30] if len(parts) > 30 else "",
                    "source": "tencent.qt",
                }
            )
        if len(items) < 2:
            raise ValueError("tencent us indices insufficient")
        return {"count": len(items[:limit]), "items": items[:limit], "generated_at": self._now_cn().isoformat()}

    def _us_indices_sina_payload(self, limit: int) -> dict[str, Any]:
        defs = self._us_index_defs()
        codes = [item["sina"] for item in defs]
        text = self._request_text(
            "https://hq.sinajs.cn/list=" + ",".join(codes),
            headers={"Referer": "https://finance.sina.com.cn", "User-Agent": HTTP_HEADERS["User-Agent"]},
            encoding="gbk",
            timeout=8,
        )
        by_code: dict[str, list[str]] = {}
        for line in text.strip().splitlines():
            if '="' not in line:
                continue
            left, right = line.split('="', 1)
            code = left.split("_")[-1] if "hq_str_" not in left else left.split("hq_str_", 1)[-1]
            # var hq_str_gb_dji="...
            if "hq_str_" in left:
                code = left.split("hq_str_", 1)[-1]
            parts = right.rstrip('";').split(",")
            by_code[code] = parts
        items: list[dict[str, Any]] = []
        for item in defs:
            parts = by_code.get(item["sina"]) or []
            if len(parts) < 3:
                continue
            # gb_*: name, price, change_pct ...
            price = self._float(parts[1]) if len(parts) > 1 else None
            change_pct = self._float(parts[2]) if len(parts) > 2 else None
            if price is None:
                continue
            prev = None
            if change_pct is not None and abs(change_pct) < 1000:
                # sina may return absolute change in some fields; prefer pct when small
                try:
                    prev = price / (1 + change_pct / 100.0) if change_pct != -100 else None
                except Exception:
                    prev = None
            items.append(
                {
                    "key": item["key"],
                    "symbol": item["yahoo"],
                    "name": parts[0] or item["name"],
                    "price": price,
                    "prev_close": None if prev is None else round(prev, 4),
                    "change": None if prev is None else round(price - prev, 4),
                    "change_pct": change_pct,
                    "time": "",
                    "source": "sina.hq",
                }
            )
        if len(items) < 2:
            raise ValueError("sina us indices insufficient")
        return {"count": len(items[:limit]), "items": items[:limit], "generated_at": self._now_cn().isoformat()}

    def _us_indices_yahoo_payload(self, limit: int) -> dict[str, Any]:
        items: list[dict[str, Any]] = []
        for item in self._us_index_defs():
            quote = self._yahoo_us_daily_quote(item["yahoo"])
            if not quote:
                continue
            items.append(
                {
                    "key": item["key"],
                    "symbol": item["yahoo"],
                    "name": item["name"],
                    "price": quote["price"],
                    "prev_close": quote["prev_close"],
                    "change": quote["change"],
                    "change_pct": quote["change_pct"],
                    "time": quote.get("time") or "",
                    "source": "yahoo.chart",
                }
            )
        if len(items) < 2:
            raise ValueError("yahoo us indices insufficient")
        return {"count": len(items[:limit]), "items": items[:limit], "generated_at": self._now_cn().isoformat()}

    def _yahoo_us_daily_quote(self, symbol: str) -> dict[str, Any] | None:
        data = self._yahoo_chart(symbol, range_str="5d", interval="1d")
        result = (((data.get("chart") or {}).get("result") or []) + [None])[0]
        if not isinstance(result, dict):
            return None
        meta = result.get("meta") if isinstance(result.get("meta"), dict) else {}
        closes = (((result.get("indicators") or {}).get("quote") or [{}])[0] or {}).get("close") or []
        close_values = [n for v in closes if (n := self._float(v)) is not None and n > 0]
        price = self._float(meta.get("regularMarketPrice")) or (close_values[-1] if close_values else None)
        prev_close = (
            (close_values[-2] if len(close_values) >= 2 else None)
            or self._float(meta.get("previousClose"))
            or self._float(meta.get("chartPreviousClose"))
        )
        if price is None or prev_close is None or prev_close <= 0:
            return None
        ts = self._float(meta.get("regularMarketTime"))
        time_text = ""
        if ts:
            time_text = datetime.fromtimestamp(float(ts), CN_TZ).strftime("%Y-%m-%d %H:%M:%S")
        return {
            "symbol": symbol,
            "price": round(price, 4),
            "prev_close": round(prev_close, 4),
            "change": round(price - prev_close, 4),
            "change_pct": round((price / prev_close - 1) * 100, 4),
            "time": time_text,
            "name": meta.get("shortName") or meta.get("symbol") or symbol,
        }

    def _us_sectors_yahoo_payload(self, limit: int) -> dict[str, Any]:
        items: list[dict[str, Any]] = []
        errors: list[str] = []

        def build(defn: dict[str, Any]) -> dict[str, Any] | None:
            try:
                quote = self._yahoo_us_daily_quote(str(defn["symbol"]))
            except Exception as exc:
                errors.append(f"{defn.get('symbol')}:{type(exc).__name__}")
                return None
            if not quote:
                return None
            pct = quote.get("change_pct")
            return {
                "key": defn.get("key"),
                "symbol": defn.get("symbol"),
                "label": defn.get("label"),
                "price": quote.get("price"),
                "prev_close": quote.get("prev_close"),
                "change": quote.get("change"),
                "change_pct": pct,
                "change_pct_text": self._fmt_pct(pct),
                "time": quote.get("time") or "",
                "a_share_mapping": list(defn.get("a_share_mapping") or []),
            }

        with ThreadPoolExecutor(max_workers=min(6, len(self._us_sector_defs()))) as pool:
            for item in pool.map(build, self._us_sector_defs()):
                if item:
                    items.append(item)
        if not items:
            raise ValueError("yahoo us sectors empty: " + ";".join(errors[:3]))
        items.sort(key=lambda row: abs(float(row.get("change_pct") or 0)), reverse=True)
        return {
            "count": len(items[:limit]),
            "items": items[:limit],
            "generated_at": self._now_cn().strftime("%Y-%m-%d %H:%M:%S"),
            "errors": errors[:5] if errors else [],
        }

    def _fmt_pct(self, value: Any) -> str:
        n = self._float(value)
        if n is None:
            return "--"
        sign = "+" if n > 0 else ""
        return f"{sign}{n:.2f}%"

    def _us_market_summary_payload(self) -> dict[str, Any]:
        indices = self.us_indices(limit=10).get("data") or self._us_indices_yahoo_payload(10)
        # when called inside source runner, us_indices returns envelope; support both
        if isinstance(indices, dict) and "items" not in indices and "data" in indices:
            indices = indices.get("data") or {}
        try:
            sectors_env = self.us_sectors(limit=8)
            sectors = sectors_env.get("data") if isinstance(sectors_env, dict) and "data" in sectors_env else sectors_env
        except Exception:
            sectors = {"items": []}
        items = [x for x in (indices.get("items") or []) if isinstance(x, dict)]
        by_key = {str(x.get("key")): x for x in items}
        index_metrics = []
        for key, label in (("dow", "道琼斯指数"), ("nas", "纳斯达克指数"), ("spx", "标普500指数")):
            row = by_key.get(key)
            if not row:
                continue
            pct = self._float(row.get("change_pct"))
            index_metrics.append(
                {
                    "key": key,
                    "label": label,
                    "value": row.get("price"),
                    "change_pct": pct,
                    "change_pct_text": self._fmt_pct(pct),
                    "time": row.get("time") or "",
                }
            )
        pcts = [self._float(m.get("change_pct")) for m in index_metrics]
        pcts = [p for p in pcts if p is not None]
        available = len(pcts) >= 2
        avg = sum(pcts) / len(pcts) if pcts else 0.0
        positives = sum(1 for p in pcts if p > 0.15)
        negatives = sum(1 for p in pcts if p < -0.15)
        if avg <= -1.0:
            tone, tone_label, tone_reason = "defensive", "防守", "隔夜美股明显承压，今日先把风险预算降下来。"
        elif avg < -0.25 or negatives >= 2:
            tone, tone_label, tone_reason = "cautious", "谨慎", "隔夜美股偏弱或分化，今日不急着追高。"
        elif avg >= 0.55 and positives >= 2:
            tone, tone_label, tone_reason = "offensive", "进攻", "隔夜美股风险偏好回暖，今日可以更积极寻找确认后的机会。"
        elif avg > 0.05 and positives >= 2:
            tone, tone_label, tone_reason = "balanced", "平衡", "隔夜美股整体偏暖，今日按结构性机会处理。"
        else:
            tone, tone_label, tone_reason = "neutral", "中性", "隔夜美股方向不强，今日以 A 股自身竞价和资金流为准。"
        index_line = "、".join(f"{m['label']} {m['change_pct_text']}" for m in index_metrics) or "三大指数数据暂缺"
        sector_items = [x for x in ((sectors or {}).get("items") or []) if isinstance(x, dict)]
        sector_mappings = []
        for row in sector_items[:5]:
            pct = self._float(row.get("change_pct"))
            if pct is None:
                bias = "观察"
            elif pct >= 0.35:
                bias = "正映射"
            elif pct <= -0.35:
                bias = "负映射"
            else:
                bias = "观察"
            sector_mappings.append(
                {
                    "us_sector": row.get("label"),
                    "proxy": row.get("symbol"),
                    "change_pct": pct,
                    "change_pct_text": row.get("change_pct_text") or self._fmt_pct(pct),
                    "bias": bias,
                    "a_share_mapping": row.get("a_share_mapping") or [],
                }
            )
        now = self._now_cn()
        target_us = now.date() - timedelta(days=1)
        while target_us.weekday() >= 5:
            target_us -= timedelta(days=1)
        summary = f"{target_us:%Y-%m-%d} 美股收盘：{index_line}。{tone_reason}"
        if not available:
            summary = f"{target_us:%Y-%m-%d} 美股盘面数据暂不完整，今日先按中性外盘背景处理。"
        return {
            "available": available,
            "target_cn_date": now.strftime("%Y-%m-%d"),
            "target_us_date": target_us.strftime("%Y-%m-%d"),
            "date_rule": "周一显示上周五美股盘面；其他日期显示前一美股交易日。",
            "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
            "tone": tone,
            "tone_label": tone_label,
            "summary": summary,
            "metrics": index_metrics,
            "sector_mappings": sector_mappings,
            "guidance_lines": [
                tone_reason,
                "买入节奏：先看 A 股竞价强弱、资金流和板块联动，不因外盘单独加仓。",
                "选股方向：外盘正映射仅作加分项，必须叠加 A 股自身确认。",
            ],
        }

    def _us_stock_quote_yahoo(self, symbol: str) -> dict[str, Any]:
        quote = self._yahoo_us_daily_quote(symbol)
        if not quote:
            raise ValueError(f"yahoo us quote empty: {symbol}")
        return {
            "symbol": symbol,
            "name": quote.get("name") or symbol,
            "price": quote.get("price"),
            "prev_close": quote.get("prev_close"),
            "change": quote.get("change"),
            "change_pct": quote.get("change_pct"),
            "time": quote.get("time") or "",
            "source": "yahoo.chart",
        }

    def _us_stock_quote_sina(self, symbol: str) -> dict[str, Any]:
        code = "gb_" + symbol.lower().lstrip("^").replace(".", "")
        text = self._request_text(
            f"https://hq.sinajs.cn/list={code}",
            headers={"Referer": "https://finance.sina.com.cn", "User-Agent": HTTP_HEADERS["User-Agent"]},
            encoding="gbk",
            timeout=8,
        )
        if '=""' in text or '="' not in text:
            raise ValueError(f"sina us quote empty: {symbol}")
        value = text.split('="', 1)[1].rstrip('";\n')
        parts = value.split(",")
        if len(parts) < 3:
            raise ValueError(f"sina us quote malformed: {symbol}")
        price = self._float(parts[1])
        change_pct = self._float(parts[2])
        if price is None:
            raise ValueError(f"sina us quote missing price: {symbol}")
        return {
            "symbol": symbol,
            "name": parts[0] or symbol,
            "price": price,
            "prev_close": None,
            "change": None,
            "change_pct": change_pct,
            "time": "",
            "source": "sina.hq",
        }

    def _us_stock_daily_yahoo(self, symbol: str, days: int) -> dict[str, Any]:
        if days <= 5:
            range_str = "5d"
        elif days <= 30:
            range_str = "1mo"
        elif days <= 90:
            range_str = "3mo"
        elif days <= 180:
            range_str = "6mo"
        elif days <= 365:
            range_str = "1y"
        elif days <= 730:
            range_str = "2y"
        else:
            range_str = "5y"
        data = self._yahoo_chart(symbol, range_str=range_str, interval="1d")
        result = (((data.get("chart") or {}).get("result") or []) + [None])[0]
        if not isinstance(result, dict):
            raise ValueError(f"yahoo us daily empty: {symbol}")
        timestamps = result.get("timestamp") or []
        quote = ((result.get("indicators") or {}).get("quote") or [{}])[0] or {}
        closes = quote.get("close") or []
        opens = quote.get("open") or []
        highs = quote.get("high") or []
        lows = quote.get("low") or []
        volumes = quote.get("volume") or []
        records: list[dict[str, Any]] = []
        for i, ts in enumerate(timestamps):
            c = self._float(closes[i]) if i < len(closes) else None
            if c is None:
                continue
            o = self._float(opens[i]) if i < len(opens) else c
            h = self._float(highs[i]) if i < len(highs) else c
            l = self._float(lows[i]) if i < len(lows) else c
            v = self._float(volumes[i]) if i < len(volumes) else 0
            prev = self._float(closes[i - 1]) if i > 0 and i - 1 < len(closes) else o
            change_pct = round((c - prev) / prev * 100, 4) if prev else None
            records.append(
                {
                    "date": datetime.fromtimestamp(ts, CN_TZ).strftime("%Y-%m-%d"),
                    "open": None if o is None else round(o, 4),
                    "high": None if h is None else round(h, 4),
                    "low": None if l is None else round(l, 4),
                    "close": round(c, 4),
                    "volume": None if v is None else int(v),
                    "change_pct": change_pct,
                }
            )
        if not records:
            raise ValueError(f"yahoo us daily parsed empty: {symbol}")
        records = records[-days:]
        return {"symbol": symbol, "days": len(records), "records": records, "source": "yahoo"}

    # ---- X / Twitter timeline helpers ----

    def _parse_x_handles(self, accounts: str) -> list[str]:
        raw = str(accounts or "")
        parts = [p.strip().lstrip("@") for p in raw.replace(";", ",").replace(" ", ",").split(",")]
        handles: list[str] = []
        seen: set[str] = set()
        for part in parts:
            handle = "".join(ch for ch in part if ch.isalnum() or ch == "_")
            if not handle:
                continue
            key = handle.lower()
            if key in seen:
                continue
            seen.add(key)
            handles.append(handle)
        return handles[:20]

    def _x_timeline_via_model(self, handles: list[str], limit: int, hydrate: bool) -> dict[str, Any]:
        if not settings.search_api_base_url or not settings.search_api_key:
            raise ValueError("search api not configured for x timeline")
        account_text = ", ".join("@" + h for h in handles)
        prompt = f"""
请使用你可用的实时 X/Twitter 能力，获取以下账号每个账号最近 {limit} 条公开推文的最小列表：{account_text}。

严格返回 JSON，不要 markdown，不要解释。格式：
{{"accounts":[{{"handle":"wallstreetbets","display_name":"昵称","posts":[{{"post_id":"数字ID或唯一ID","time":"YYYY-MM-DD HH:mm:ss 北京时间","chinese_text":"完整正文；外文完整翻译成中文；中文保留原文","conversation_type":"original|reply|quote|repost|unknown","media":[]}}]}}]}}

要求：
- 必须返回上述每个账号，即使没有抓到也要给 posts: []。
- 每个账号最多 {limit} 条，优先最新推文。
- 必须包含 post_id；没有数字 ID 时用可稳定去重的唯一字符串。
- 必须包含发布时间；尽量使用北京时间。
- 不要省略推文正文。
"""
        content = self._openai_chat_text(prompt, max_tokens=min(4000, 800 + 400 * len(handles)), timeout=60)
        payload = self._extract_json_object(content)
        accounts = payload.get("accounts") if isinstance(payload, dict) else None
        if not isinstance(accounts, list):
            raise ValueError("model x timeline returned invalid accounts")
        normalized = []
        for account in accounts:
            if not isinstance(account, dict):
                continue
            handle = str(account.get("handle") or "").lstrip("@")
            posts = account.get("posts") if isinstance(account.get("posts"), list) else []
            clean_posts = []
            for post in posts[:limit]:
                if not isinstance(post, dict):
                    continue
                post_id = str(post.get("post_id") or "").strip()
                clean_posts.append(
                    {
                        "post_id": post_id,
                        "time": post.get("time") or "",
                        "text": post.get("chinese_text") or post.get("full_text") or post.get("text") or "",
                        "conversation_type": post.get("conversation_type") or "unknown",
                        "url": f"https://x.com/{handle}/status/{post_id}" if post_id.isdigit() else "",
                        "media": post.get("media") if isinstance(post.get("media"), list) else [],
                    }
                )
            if hydrate:
                for post in clean_posts:
                    if post.get("post_id") and post["post_id"].isdigit():
                        html_post = self._x_fetch_status_html(handle, post["post_id"])
                        if html_post:
                            if not post.get("text"):
                                post["text"] = html_post.get("text") or ""
                            if html_post.get("media"):
                                post["media"] = html_post["media"]
            normalized.append(
                {
                    "handle": handle,
                    "display_name": account.get("display_name") or handle,
                    "posts": clean_posts,
                }
            )
        if not normalized:
            raise ValueError("model x timeline empty")
        return {
            "accounts": normalized,
            "count": sum(len(a.get("posts") or []) for a in normalized),
            "generated_at": self._now_cn().strftime("%Y-%m-%d %H:%M:%S"),
            "source": "openai_compatible.x_watchlist",
        }

    def _x_timeline_via_html(self, handles: list[str], limit: int) -> dict[str, Any]:
        accounts = []
        for handle in handles:
            posts = self._x_fetch_profile_posts_html(handle, limit)
            accounts.append({"handle": handle, "display_name": handle, "posts": posts})
        if not any(a.get("posts") for a in accounts):
            raise ValueError("x html timeline empty")
        return {
            "accounts": accounts,
            "count": sum(len(a.get("posts") or []) for a in accounts),
            "generated_at": self._now_cn().strftime("%Y-%m-%d %H:%M:%S"),
            "source": "x.com.html",
        }

    def _x_fetch_status_html(self, handle: str, post_id: str) -> dict[str, Any] | None:
        import re

        for domain in ("x.com", "twitter.com"):
            url = f"https://{domain}/{handle}/status/{post_id}"
            try:
                text = self._request_text(
                    url,
                    headers={
                        "User-Agent": HTTP_HEADERS["User-Agent"],
                        "Accept": "text/html,application/xhtml+xml",
                    },
                    timeout=8,
                )
            except Exception:
                continue
            title = ""
            for pattern in (
                r'<meta[^>]+property=["\']og:title["\'][^>]+content=["\']([^"\']*)["\']',
                r'<meta[^>]+content=["\']([^"\']*)["\'][^>]+property=["\']og:title["\']',
                r"<title>([^<]+)</title>",
            ):
                match = re.search(pattern, text, flags=re.I)
                if match:
                    title = match.group(1).strip()
                    break
            media = []
            for match in re.finditer(
                r'https://pbs\.twimg\.com/(?:media|ext_tw_video_thumb|tweet_video_thumb)/[^"\'\\\s<>]+',
                text,
            ):
                media.append({"type": "image", "url": match.group(0)})
            if title or media:
                return {"text": title, "media": media[:8], "url": url}
        return None

    def _x_fetch_profile_posts_html(self, handle: str, limit: int) -> list[dict[str, Any]]:
        import re

        posts: list[dict[str, Any]] = []
        for domain in ("x.com", "twitter.com"):
            url = f"https://{domain}/{handle}"
            try:
                text = self._request_text(
                    url,
                    headers={
                        "User-Agent": HTTP_HEADERS["User-Agent"],
                        "Accept": "text/html,application/xhtml+xml",
                    },
                    timeout=10,
                )
            except Exception:
                continue
            ids = []
            for match in re.finditer(rf'/{re.escape(handle)}/status/(\d+)', text, flags=re.I):
                post_id = match.group(1)
                if post_id not in ids:
                    ids.append(post_id)
            for post_id in ids[:limit]:
                detail = self._x_fetch_status_html(handle, post_id) or {}
                posts.append(
                    {
                        "post_id": post_id,
                        "time": "",
                        "text": detail.get("text") or "",
                        "conversation_type": "unknown",
                        "url": f"https://x.com/{handle}/status/{post_id}",
                        "media": detail.get("media") or [],
                    }
                )
            if posts:
                break
        return posts

    def _openai_chat_text(self, prompt: str, max_tokens: int = 2000, timeout: int = 60) -> str:
        base = settings.search_api_base_url.rstrip("/")
        if base.endswith("/chat/completions"):
            url = base
        else:
            url = base + "/chat/completions"
        body = {
            "model": settings.search_api_model,
            "messages": [
                {"role": "system", "content": "You are a precise market data assistant. Return only valid JSON when asked."},
                {"role": "user", "content": prompt},
            ],
            "max_tokens": max_tokens,
            "temperature": 0.1,
            "stream": False,
        }
        response = self._http_get  # placate type checkers
        import requests as _requests

        resp = _requests.post(
            url,
            headers={
                "Authorization": f"Bearer {settings.search_api_key}",
                "Content-Type": "application/json",
                "Accept": "application/json",
                "User-Agent": "stock-data-api/niuone-x",
            },
            json=body,
            timeout=timeout,
        )
        resp.raise_for_status()
        data = resp.json()
        content = (((data.get("choices") or [{}])[0].get("message") or {}).get("content")) or ""
        if not str(content).strip():
            raise ValueError("openai compatible model returned empty content")
        return str(content).strip()

    def _extract_json_object(self, content: str) -> dict[str, Any]:
        import re

        text = str(content or "").strip()
        if text.startswith("```"):
            text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.I).strip()
            text = re.sub(r"\s*```$", "", text).strip()
        if not text.startswith("{"):
            match = re.search(r"\{[\s\S]*\}", text)
            if match:
                text = match.group(0)
        data = json.loads(text)
        if not isinstance(data, dict):
            raise ValueError("json payload must be object")
        return data

market_data_service = MarketDataService()