myquant/djapi/api/stock/getIndexs.py

import pandas as pd


# 导入当站目录的config文件
try:
    # 尝试相对导入（作为包的一部分）
    from .config import TS_TOKEN, START_DATE, END_DATE, PRECISION_CONFIG
    from .stock_utils import *
except (ImportError, SystemError):  
    # 失败则使用绝对导入（直接运行脚本）
    from config import TS_TOKEN, START_DATE, END_DATE, PRECISION_CONFIG
    from stock_utils import *

from .data_source import get_tushare_pro
pro = get_tushare_pro()

def get_index_by_name(index_name: str) -> pd.DataFrame:
    """
    通过指数名称查询指数基本信息（使用tushare的index_basic接口）

    Args:
        index_name (str): 要查询的指数名称（支持模糊匹配，如"上证"）

    Returns:
        pd.DataFrame: 包含查询结果的DataFrame，列包括:
            - ts_code: 指数代码
            - name: 指数名称
            - fullname: 指数全称
            - market: 市场
            - publisher: 发布方
            - index_type: 指数类型
            - etc.

    Raises:
        Exception: 当tushare接口调用失败时抛出异常

    Example:
        >>> df = get_index_by_name("上证50")
        >>> print(df[['ts_code', 'name']])
    """
    try:
        # 调用tushare接口查询指数信息
        df = pro.index_basic(name=index_name)

        # 检查返回结果是否为空
        if df.empty:
            print(f"未找到名称包含'{index_name}'的指数")
            return pd.DataFrame()  # 返回空DataFrame保持类型一致

        return df

    except Exception as e:
        print(f"查询指数信息失败: {str(e)}")
        return pd.DataFrame()

def get_index_daily_data(ts_code: str, start_date: str = START_DATE, end_date: str = END_DATE) -> pd.DataFrame:
    """
    通过指数代码查询日线行情数据（合并基本行情和扩展行情）

    Args:
        ts_code (str): 指数代码（如"000001.SH"）
        start_date (str): 开始日期（格式"YYYYMMDD"，默认使用config中的START_DATE）
        end_date (str): 结束日期（格式"YYYYMMDD"，默认使用config中的END_DATE）

    Returns:
        pd.DataFrame: 合并后的日线行情数据，包含以下列（示例）:
            - ts_code: 指数代码
            - trade_date: 交易日期
            - close: 收盘点位
            - open: 开盘点位
            - high: 最高点位
            - low: 最低点位
            - pe: 市盈率
            - pb: 市净率
            - etc.

    Raises:
        Exception: 当tushare接口调用失败或数据合并失败时抛出异常

    Example:
        >>> df = get_index_daily_data("000001.SH")
        >>> print(df[['trade_date', 'close', 'pe']].head())
    """
    try:
        start_date = date_format_correction(start_date)
        end_date = date_format_correction(end_date)

        # 1. 查询日线基本行情（核心数据，必须成功）
        daily_df = pro.index_daily(
            ts_code=ts_code,
            start_date=start_date,
            end_date=end_date,
            fields="ts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount"
        )
        if daily_df is None or daily_df.empty:
            print("未找到指数{}在{}到{}期间的行情数据".format(ts_code, start_date, end_date))
            return pd.DataFrame()

        # 2. 尝试查询扩展行情（PE/PB 等，需要高权限，失败不阻塞）
        try:
            daily_basic_df = pro.index_dailybasic(
                ts_code=ts_code,
                start_date=start_date,
                end_date=end_date,
                fields="ts_code,trade_date,total_mv,float_mv,pe,pe_ttm,pb,turnover_rate,turnover_rate_f"
            )
            if daily_basic_df is not None and not daily_basic_df.empty:
                daily_df = pd.merge(
                    daily_df, daily_basic_df,
                    on=['ts_code', 'trade_date'],
                    how='left'
                )
        except Exception as e:
            print("index_dailybasic 不可用（权限不足或接口变更）: {}".format(e))

        return daily_df

    except Exception as e:
        print("查询指数日线行情失败: {}".format(e))
        raise  # 抛出而非静默返回空，让 view 层返回错误信息
    
if __name__ == "__main__":
    # 测试指数查询
    print("测试指数查询:")
    index_df = get_index_by_name("中证500")
    if not index_df.empty:
        print(index_df[['ts_code', 'name']].head())
    else:
        print("未查询到指数信息")
    
    '''# 测试指数行情获取
    print("\n测试指数日线行情:")
    test_code = "000016.SH"  # 上证50指数代码
    daily_data = get_index_daily_data(test_code, start_date="20230101", end_date="20231231")

    if not daily_data.empty:
        print(f"获取到{daily_data.shape[0]}条数据")
        # 显示关键列的前5行
        print(daily_data[['trade_date', 'close', 'pe', 'pb']].head())

        # 数据完整性检查
        print("\n数据完整性检查:")
        print(daily_data[['close', 'pe']].describe())
    else:
        print(f"未获取到指数{test_code}的行情数据")'''