myquant/finance/data/sources/akshare_source.py

"""
AkShare 数据源封装。

统一封装 AkShare 调用，返回标准化的 DataFrame。
所有网络请求都在这一层处理，包含重试和容错。

支持个股和指数两种数据接口。
"""

# 指数代码识别：.SH 后缀为主板指数，399xxx 为深证指数
# 注意区分：000001.SZ 是平安银行(个股)，000001.SH 是上证指数
def is_index_code(ts_code: str) -> bool:
    """判断是否为指数代码。排除 000xxx.SZ 个股。"""
    if not ts_code:
        return False
    code = ts_code.upper()
    # .SH 开头 000 是指数
    if code.endswith(".SH") and (code.startswith("000") or code.startswith("399")):
        return True
    # 深交所 399xxx 指数
    if code.startswith("399"):
        return True
    # 无后缀的纯数字 000xxx（上证指数常见写法）
    if code == "000001":
        return True
    return False

import time
import pandas as pd
import akshare as ak

from config.settings import AKSHARE_CONFIG


class AkShareSource:
    """AkShare 数据源。"""

    def __init__(self):
        self._timeout = AKSHARE_CONFIG["request_timeout"]
        self._retry = AKSHARE_CONFIG["retry_times"]
        self._delay = AKSHARE_CONFIG["retry_delay"]

    def _retry_call(self, fn, name: str, **kwargs):
        """带重试的 API 调用包装。重试间隔递增。"""
        last_err = None
        for i in range(self._retry):
            try:
                return fn(**kwargs)
            except Exception as e:
                last_err = e
                wait = self._delay * (i + 1)
                print(f"  [retry] {name} 失败 ({e})，{wait}s 后重试 ({i + 1}/{self._retry})...")
                if i < self._retry - 1:
                    time.sleep(wait)
        raise last_err  # type: ignore

    # ── 股票列表 ──────────────────────────────────────────

    def fetch_stock_list(self) -> pd.DataFrame:
        """获取 A 股股票列表。"""
        df = self._retry_call(ak.stock_info_a_code_name, "stock_list")
        df = df.rename(columns={
            "code": "ts_code",
            "name": "name",
        })
        return df[["ts_code", "name"]]

    # ── 日线数据 ──────────────────────────────────────────

    def fetch_daily(
        self, ts_code: str, start: str, end: str | None = None
    ) -> pd.DataFrame:
        """
        获取单只股票日线行情。

        参数:
            ts_code: 股票代码，如 '000001'（纯数字格式，AkShare 要求）
            start: 起始日期 'YYYYMMDD'
            end: 结束日期 'YYYYMMDD'，None 表示今天
        """
        symbol = ts_code.replace(".SZ", "").replace(".SH", "").replace(".BJ", "")
        end = end or time.strftime("%Y%m%d")
        df = self._retry_call(
            ak.stock_zh_a_hist,
            "daily",
            symbol=symbol,
            period="daily",
            start_date=start,
            end_date=end,
            adjust="qfq",  # 前复权
        )
        if df.empty:
            return df

        df = df.rename(columns={
            "日期": "trade_date",
            "开盘": "open",
            "收盘": "close",
            "最高": "high",
            "最低": "low",
            "成交量": "vol",
            "成交额": "amount",
            "振幅": "amplitude",
            "涨跌幅": "pct_chg",
            "涨跌额": "change",
            "换手率": "turnover_rate",
        })
        df["ts_code"] = ts_code
        # AkShare 返回 'YYYY-MM-DD'，统一转为 'YYYYMMDD'
        df["trade_date"] = df["trade_date"].astype(str).str.replace("-", "")
        return df

    # ── 指数日线 ──────────────────────────────────────────

    def fetch_index_daily(
        self, ts_code: str, start: str, end: str | None = None
    ) -> pd.DataFrame:
        """
        获取指数日线行情。

        AkShare index_zh_a_hist 接口，symbol 为纯数字（如 '000001'）。
        """
        symbol = ts_code.replace(".SH", "").replace(".SZ", "").replace(".BJ", "")
        end = end or time.strftime("%Y%m%d")
        df = self._retry_call(
            ak.index_zh_a_hist,
            "index_daily",
            symbol=symbol, period="daily",
            start_date=start, end_date=end,
        )
        if df.empty:
            return df

        df = df.rename(columns={
            "日期": "trade_date",
            "开盘": "open",
            "收盘": "close",
            "最高": "high",
            "最低": "low",
            "成交量": "vol",
            "成交额": "amount",
            "涨跌幅": "pct_chg",
            "涨跌额": "change",
        })
        df["ts_code"] = ts_code
        df["trade_date"] = df["trade_date"].astype(str)
        return df

    # ── 财务数据 ──────────────────────────────────────────

    def fetch_financial(self, ts_code: str) -> pd.DataFrame:
        """获取单只股票核心财务指标（同花顺接口）。"""
        symbol = ts_code.replace(".SZ", "").replace(".SH", "").replace(".BJ", "")
        try:
            df = ak.stock_financial_abstract_ths(symbol=symbol)
            if df.empty:
                return pd.DataFrame()

            df = df.rename(columns={
                "报告期": "end_date",
                "净利润": "net_profit",
                "净利润同比增长率": "net_profit_yoy",
                "扣非净利润": "deducted_net_profit",
                "扣非净利润同比增长率": "deducted_net_profit_yoy",
                "营业总收入": "total_revenue",
                "营业总收入同比增长率": "total_revenue_yoy",
                "基本每股收益": "eps",
                "每股净资产": "bvps",
                "每股资本公积金": "capital_reserve_ps",
                "每股未分配利润": "undistributed_profit_ps",
                "每股经营现金流": "ocf_ps",
                "销售净利率": "net_profit_margin",
                "净资产收益率": "roe",
                "净资产收益率-摊薄": "roe_diluted",
                "营业周期": "operating_cycle",
                "应收账款周转天数": "receivables_days",
                "流动比率": "current_ratio",
                "速动比率": "quick_ratio",
                "保守速动比率": "conservative_quick_ratio",
                "产权比率": "equity_ratio",
                "资产负债率": "debt_to_assets",
            })
            df["ts_code"] = ts_code
            # 日期格式统一
            df["end_date"] = df["end_date"].astype(str).str.replace("-", "")
            # 数值列清洗：去掉 万/亿/% 等单位
            for col in df.columns:
                if col in ("ts_code", "end_date"):
                    continue
                df[col] = self._parse_financial_value(df[col])
            return df
        except Exception:
            return pd.DataFrame()

    @staticmethod
    def _parse_financial_value(series: pd.Series) -> pd.Series:
        """解析财务数值字符串 '4302.00万', '64.75%', 'False' → float"""
        def _parse(v):
            if v is None or v == "False" or v == "":
                return None
            if isinstance(v, (int, float)):
                return float(v)
            s = str(v).strip()
            if not s:
                return None
            try:
                if s.endswith("%"):
                    return float(s[:-1])
                if "万" in s:
                    return float(s.replace("万", "")) * 1e4
                if "亿" in s:
                    return float(s.replace("亿", "")) * 1e8
                return float(s)
            except ValueError:
                return None
        return series.apply(_parse).astype("float64")