Tushare Mcp

{
  "name": "tushare-mcp",
  "version": "1.0.0",
  "description": "Tushare Pro MCP 数据技能。通过 158+ 个 MCP tool 获取 A股/港股/美股/基金/期货/债券/宏观数据。替代旧版 tushare-finance skill，无需写 Python 脚本。",
  "author": "kirkluokun",
  "tags": ["金融数据", "MCP", "A股", "港股", "美股", "基金", "期货", "债券", "宏观经济", "Tushare"],
  "license": "MIT",
  "replaces": "tushare-finance-2.0.6",
  "dependencies": {
    "mcp": "tushare-pro"
  }
}

# ============================================================
# Tushare Pro MCP 环境配置
# ============================================================
# 使用方法：
#   1. 复制本文件为 .env       → cp .env.example .env
#   2. 填入你的真实 Token
#
# .env 查找顺序（优先级从高到低）：
#   1. 系统环境变量（已设置则不覆盖）
#   2. 本技能文件夹下的 .env
#   3. 上一级目录的 .env（适用于多技能共享同一份密钥）
# ============================================================

# Tushare Pro API Token（A 股金融数据核心依赖）
# 获取地址：https://tushare.pro/register（注册后在个人中心获取 Token）
# 注意：部分高级接口需要积分，详见 https://tushare.pro/document/1?doc_id=108
TUSHARE_API_KEY=your_tushare_token_here

# Python
__pycache__/
*.pyc
*.pyo
.venv/

# Data (SQLite 数据库，自动生成，不提交)
data/

# IDE
.idea/
.vscode/
*.swp

# pytest
.pytest_cache/

# OS
.DS_Store

[project]
name = "tushare-pro-mcp"
version = "0.1.0"
description = "Tushare Pro MCP Server - A股数据 MCP 工具服务"
requires-python = ">=3.11"
dependencies = [
    "fastmcp>=2.0.0",
    "tushare>=1.4.0",
    "pandas>=2.0.0",
    "tabulate>=0.9.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-asyncio>=0.24.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/tushare_mcp"]

"""Tushare Pro MCP Server — A股数据 MCP 工具服务"""

"""允许 python -m tushare_mcp.server 启动"""
from .server import main

main()

"""
Tushare MCP 异常类型定义

所有自定义异常都继承自 TushareError，方便统一捕获。
"""


class TushareError(Exception):
    """Tushare MCP 基础异常"""
    pass


class InsufficientPointsError(TushareError):
    """积分不足，无法调用该接口"""
    pass


class RateLimitError(TushareError):
    """API 调用频率超限"""
    pass


class ApiError(TushareError):
    """Tushare API 返回错误"""

    def __init__(self, api_name: str, message: str) -> None:
        self.api_name = api_name
        super().__init__(f"[{api_name}] {message}")


class TokenError(TushareError):
    """Token 未配置或无效"""
    pass

"""
DataFrame 格式转换

将 pandas DataFrame 转为 MCP tool 返回的标准格式。
支持 json 和 markdown 两种输出。
"""

import json
from typing import Any

import pandas as pd


def format_response(
    df: pd.DataFrame,
    api_name: str,
    output_format: str = "json",
    limit: int = 100,
) -> dict[str, Any]:
    """
    将 DataFrame 转换为 MCP tool 标准响应

    Args:
        df: tushare 返回的 DataFrame
        api_name: 接口名称，用于元信息
        output_format: 输出格式（json 或 markdown）
        limit: 最大返回行数

    Returns:
        标准响应字典，包含 api, total_rows, returned_rows, truncated, data
    """
    total_rows = len(df)
    truncated = total_rows > limit
    df_limited = df.head(limit)
    returned_rows = len(df_limited)

    if output_format == "markdown":
        data = _df_to_markdown(df_limited)
    else:
        data = _df_to_json(df_limited)

    return {
        "api": api_name,
        "total_rows": total_rows,
        "returned_rows": returned_rows,
        "truncated": truncated,
        "data": data,
    }


def _df_to_json(df: pd.DataFrame) -> list[dict[str, Any]]:
    """DataFrame → 字典列表，NaN 转为 None"""
    return json.loads(df.to_json(orient="records", date_format="iso"))


def _df_to_markdown(df: pd.DataFrame) -> str:
    """DataFrame → Markdown 表格"""
    if df.empty:
        return "*无数据*"
    return df.to_markdown(index=False)

"""
频率限制器

Tushare Pro API 有调用频率限制，这里用简单的时间间隔控制。
默认最小间隔 300ms，避免触发服务端限流。
"""

import time
import threading


class RateLimiter:
    """
    线程安全的频率限制器

    Args:
        min_interval: 两次调用之间的最小间隔（秒）
    """

    def __init__(self, min_interval: float = 0.3) -> None:
        self._min_interval = min_interval
        self._last_call: float = 0.0
        self._lock = threading.Lock()

    def wait(self) -> None:
        """等待直到可以发起下一次调用"""
        with self._lock:
            now = time.monotonic()
            elapsed = now - self._last_call
            if elapsed < self._min_interval:
                time.sleep(self._min_interval - elapsed)
            self._last_call = time.monotonic()

"""
Tushare Pro MCP Server 入口

通过 FastMCP 创建 MCP 服务器，注册所有 tool 模块。
启动方式：python -m tushare_mcp.server
"""

from fastmcp import FastMCP

from .tools import (
    stock, finance, index, macro, earnings, classify, news,
    margin, billboard, holder, corporate, moneyflow, concept,
    hk, hk_finance, us, us_finance, fund, etf,
    futures, forex, option, bond, research,
    dc_board, tdx_board, limit, chips, hotmoney, minutes,
    data_store,
)

# 创建 MCP 服务器实例
mcp = FastMCP(
    "Tushare Pro",
    instructions=(
        "A股金融数据服务。通过 tushare_* 系列工具获取股票行情、财务报表、指数数据等。"
        "所有日期参数支持 YYYYMMDD 或 YYYY-MM-DD 格式。"
        "返回结果默认限制 100 行，可通过 _limit 参数调整。"
    ),
)

# 注册各模块的 tools
stock.register(mcp)
finance.register(mcp)
index.register(mcp)
macro.register(mcp)
earnings.register(mcp)
classify.register(mcp)
news.register(mcp)
margin.register(mcp)
billboard.register(mcp)
holder.register(mcp)
corporate.register(mcp)
moneyflow.register(mcp)
concept.register(mcp)
hk.register(mcp)
hk_finance.register(mcp)
us.register(mcp)
us_finance.register(mcp)
fund.register(mcp)
etf.register(mcp)
futures.register(mcp)
forex.register(mcp)
option.register(mcp)
bond.register(mcp)
research.register(mcp)
dc_board.register(mcp)
tdx_board.register(mcp)
limit.register(mcp)
chips.register(mcp)
hotmoney.register(mcp)
minutes.register(mcp)
data_store.register(mcp)


def main() -> None:
    """启动 MCP 服务器（stdio 模式）"""
    mcp.run()


if __name__ == "__main__":
    main()

"""
SQLite 数据存储引擎

将 MCP 获取的 tushare 数据持久化到本地 SQLite，
建立索引方便未来数据回溯和历史查询。

两层表结构：
- data_index: 元数据索引表（每次查询一条记录）
- data_{api_name}: 每个 API 一张数据表（动态建表）
"""

import json
import logging
import sqlite3
from datetime import datetime
from typing import Any, Optional

import pandas as pd

logger = logging.getLogger(__name__)


# pandas dtype → SQLite 类型映射
_DTYPE_MAP = {
    "int64": "INTEGER",
    "Int64": "INTEGER",
    "float64": "REAL",
    "Float64": "REAL",
    "object": "TEXT",
    "bool": "INTEGER",
    "datetime64[ns]": "TEXT",
}


class DataStore:
    """
    SQLite 数据存储

    每次 API 调用的结果自动保存：
    1. 在 data_index 写一条索引记录
    2. 在 data_{api_name} 表写入数据行
    """

    def __init__(self, db_path: str) -> None:
        """
        初始化存储引擎

        Args:
            db_path: SQLite 数据库文件路径
        """
        self._db_path = db_path
        self._conn = sqlite3.connect(db_path, check_same_thread=False)
        # 启用 WAL 模式，提高并发读写性能
        self._conn.execute("PRAGMA journal_mode=WAL")
        self._ensure_index_table()

    def _ensure_index_table(self) -> None:
        """创建索引表（如果不存在）"""
        self._conn.executescript("""
            CREATE TABLE IF NOT EXISTS data_index (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                api_name TEXT NOT NULL,
                query_params TEXT,
                ts_code TEXT,
                trade_date TEXT,
                start_date TEXT,
                end_date TEXT,
                row_count INTEGER,
                columns TEXT,
                table_name TEXT,
                saved_at TEXT DEFAULT (datetime('now','localtime'))
            );
            CREATE INDEX IF NOT EXISTS idx_index_api_name ON data_index(api_name);
            CREATE INDEX IF NOT EXISTS idx_index_ts_code ON data_index(ts_code);
            CREATE INDEX IF NOT EXISTS idx_index_saved_at ON data_index(saved_at);
        """)
        self._conn.commit()

    def save(self, api_name: str, params: dict[str, Any], df: pd.DataFrame) -> int:
        """
        保存 API 返回的 DataFrame

        Args:
            api_name: 接口名称（如 "daily"、"income"）
            params: 原始查询参数（过滤 None 后的）
            df: tushare 返回的 DataFrame

        Returns:
            索引记录 id
        """
        if df.empty:
            return -1

        table_name = f"data_{api_name}"

        # 确保数据表存在且列齐全
        self._ensure_data_table(table_name, df)

        # 写索引记录
        query_id = self._insert_index(api_name, params, df, table_name)

        # 写数据行
        self._insert_data(table_name, query_id, df)

        return query_id

    def query_index(
        self,
        api_name: Optional[str] = None,
        ts_code: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        limit: int = 50,
    ) -> pd.DataFrame:
        """
        查询索引表

        Args:
            api_name: 按接口名筛选
            ts_code: 按股票代码筛选
            start_date: 保存时间起始（YYYY-MM-DD）
            end_date: 保存时间截止（YYYY-MM-DD）
            limit: 返回行数上限

        Returns:
            索引记录 DataFrame
        """
        conditions = []
        values: list[Any] = []

        if api_name:
            conditions.append("api_name = ?")
            values.append(api_name)
        if ts_code:
            conditions.append("ts_code = ?")
            values.append(ts_code)
        if start_date:
            conditions.append("saved_at >= ?")
            values.append(start_date)
        if end_date:
            conditions.append("saved_at <= ?")
            values.append(end_date + " 23:59:59")

        where = " AND ".join(conditions) if conditions else "1=1"
        sql = f"SELECT * FROM data_index WHERE {where} ORDER BY saved_at DESC LIMIT ?"
        values.append(limit)

        return pd.read_sql_query(sql, self._conn, params=values)

    def read_data(
        self,
        query_id: Optional[int] = None,
        api_name: Optional[str] = None,
        ts_code: Optional[str] = None,
        limit: int = 100,
    ) -> pd.DataFrame:
        """
        从本地数据库读取历史数据（不调用 API）

        三种查询模式：
        1. 按 query_id 精确读取某次查询的数据
        2. 按 api_name 读取某类数据（如所有 daily 数据）
        3. 按 api_name + ts_code 读取特定标的数据

        Args:
            query_id: 索引记录 id（精确匹配某次查询）
            api_name: 接口名称（如 daily, income）
            ts_code: 股票/标的代码（需配合 api_name 使用）
            limit: 最大返回行数

        Returns:
            数据 DataFrame
        """
        # 按 query_id 精确查询
        if query_id is not None:
            # 先从索引表找到 table_name
            cursor = self._conn.execute(
                "SELECT table_name FROM data_index WHERE id = ?", (query_id,)
            )
            row = cursor.fetchone()
            if row is None:
                return pd.DataFrame()
            table_name = row[0]
            sql = f'SELECT * FROM "{table_name}" WHERE _query_id = ? LIMIT ?'
            return pd.read_sql_query(sql, self._conn, params=[query_id, limit])

        # 按 api_name + 可选 ts_code 查询
        if api_name is None:
            return pd.DataFrame()

        table_name = f"data_{api_name}"

        # 检查表是否存在
        cursor = self._conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
            (table_name,),
        )
        if cursor.fetchone() is None:
            return pd.DataFrame()

        conditions = []
        values: list[Any] = []

        if ts_code:
            conditions.append('"ts_code" = ?')
            values.append(ts_code)

        where = " AND ".join(conditions) if conditions else "1=1"
        sql = f'SELECT * FROM "{table_name}" WHERE {where} ORDER BY rowid DESC LIMIT ?'
        values.append(limit)

        return pd.read_sql_query(sql, self._conn, params=values)

    def close(self) -> None:
        """关闭数据库连接"""
        self._conn.close()

    # ==================== 内部方法 ====================

    def _insert_index(
        self,
        api_name: str,
        params: dict[str, Any],
        df: pd.DataFrame,
        table_name: str,
    ) -> int:
        """写索引记录，返回自增 id"""
        # 从参数中提取常用字段方便查询
        ts_code = params.get("ts_code")
        trade_date = params.get("trade_date")
        start_date = params.get("start_date")
        end_date = params.get("end_date")
        columns = list(df.columns)

        cursor = self._conn.execute(
            """INSERT INTO data_index
               (api_name, query_params, ts_code, trade_date, start_date, end_date,
                row_count, columns, table_name)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                api_name,
                json.dumps(params, ensure_ascii=False, default=str),
                ts_code,
                trade_date,
                start_date,
                end_date,
                len(df),
                json.dumps(columns, ensure_ascii=False),
                table_name,
            ),
        )
        self._conn.commit()
        return cursor.lastrowid  # type: ignore[return-value]

    def _ensure_data_table(self, table_name: str, df: pd.DataFrame) -> None:
        """
        确保数据表存在，列齐全

        如果表不存在则创建；如果有新列则 ALTER TABLE ADD COLUMN。
        """
        # 检查表是否存在
        cursor = self._conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
            (table_name,),
        )
        if cursor.fetchone() is None:
            # 建新表
            self._create_data_table(table_name, df)
            return

        # 表已存在，检查是否需要加列
        existing_cols = self._get_table_columns(table_name)
        for col in df.columns:
            if col not in existing_cols:
                sqlite_type = self._map_dtype(df[col].dtype)
                self._conn.execute(
                    f'ALTER TABLE "{table_name}" ADD COLUMN "{col}" {sqlite_type}'
                )
                logger.info("表 %s 新增列: %s %s", table_name, col, sqlite_type)
        self._conn.commit()

    def _create_data_table(self, table_name: str, df: pd.DataFrame) -> None:
        """从 DataFrame 创建数据表"""
        col_defs = ['"_query_id" INTEGER']
        for col in df.columns:
            sqlite_type = self._map_dtype(df[col].dtype)
            col_defs.append(f'"{col}" {sqlite_type}')

        sql = f'CREATE TABLE "{table_name}" ({", ".join(col_defs)})'
        self._conn.execute(sql)
        # 为 _query_id 建索引，方便按查询批次检索
        self._conn.execute(
            f'CREATE INDEX IF NOT EXISTS "idx_{table_name}_qid" ON "{table_name}"("_query_id")'
        )
        self._conn.commit()
        logger.info("创建数据表: %s (%d 列)", table_name, len(df.columns))

    def _insert_data(self, table_name: str, query_id: int, df: pd.DataFrame) -> None:
        """批量插入数据行"""
        # 构造 INSERT 语句
        cols = ["_query_id"] + list(df.columns)
        placeholders = ", ".join(["?"] * len(cols))
        col_names = ", ".join(f'"{c}"' for c in cols)
        sql = f'INSERT INTO "{table_name}" ({col_names}) VALUES ({placeholders})'

        # 准备数据：NaN → None
        rows = []
        for _, row in df.iterrows():
            values = [query_id]
            for col in df.columns:
                val = row[col]
                # pandas NaN/NaT → None（SQLite NULL）
                if pd.isna(val):
                    values.append(None)
                else:
                    values.append(val)
            rows.append(values)

        self._conn.executemany(sql, rows)
        self._conn.commit()

    def _get_table_columns(self, table_name: str) -> set[str]:
        """获取已有表的列名集合"""
        cursor = self._conn.execute(f'PRAGMA table_info("{table_name}")')
        return {row[1] for row in cursor.fetchall()}

    @staticmethod
    def _map_dtype(dtype: Any) -> str:
        """pandas dtype → SQLite 类型"""
        dtype_str = str(dtype)
        return _DTYPE_MAP.get(dtype_str, "TEXT")

"""MCP tool 注册模块"""

"""
龙虎榜 MCP Tools

包含：top_list（龙虎榜每日明细）, top_inst（龙虎榜机构交易明细）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将龙虎榜相关 tools 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_top_list(
        trade_date: str,
        ts_code: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询龙虎榜每日交易明细,含涨跌幅、成交额、净买入等。积分要求:2000。

        Args:
            trade_date: 交易日期(YYYYMMDD 或 YYYY-MM-DD,必填)
            ts_code: 股票代码
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.top_list(
                trade_date=trade_date,
                ts_code=ts_code,
                fields=fields,
            )
            return format_response(df, "top_list", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_top_inst(
        trade_date: str,
        ts_code: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询龙虎榜机构成交明细,含营业部名称、买卖额等。积分要求:5000。

        Args:
            trade_date: 交易日期(YYYYMMDD 或 YYYY-MM-DD,必填)
            ts_code: 股票代码
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.top_inst(
                trade_date=trade_date,
                ts_code=ts_code,
                fields=fields,
            )
            return format_response(df, "top_inst", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

"""
债券/可转债 MCP Tools

包含：cb_basic（可转债基础信息）, cb_daily（可转债行情）, cb_issue（可转债发行）,
      cb_rate（票面利率）, cb_call（赎回信息）, cb_share（转股结果）,
      cb_price_chg（转股价变动）, repo_daily（债券回购日行情）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将债券/可转债相关 tools 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_cb_basic(
        ts_code: Optional[str] = None,
        list_date: Optional[str] = None,
        exchange: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询可转债基础信息。积分要求:2000。

        Args:
            ts_code: 转债代码
            list_date: 上市日期
            exchange: 交易所代码
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cb_basic(
                ts_code=ts_code,
                list_date=list_date,
                exchange=exchange,
                fields=fields,
            )
            return format_response(df, "cb_basic", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_cb_daily(
        ts_code: Optional[str] = None,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询可转债行情数据。积分要求:2000。

        Args:
            ts_code: 转债代码
            trade_date: 交易日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cb_daily(
                ts_code=ts_code,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "cb_daily", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_cb_issue(
        ts_code: Optional[str] = None,
        ann_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询可转债发行数据。积分要求:2000。

        Args:
            ts_code: 转债代码
            ann_date: 公告日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cb_issue(
                ts_code=ts_code,
                ann_date=ann_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "cb_issue", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_cb_rate(
        ts_code: str,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询可转债票面利率。积分要求:5000。

        Args:
            ts_code: 转债代码(必填)
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cb_rate(ts_code=ts_code, fields=fields)
            return format_response(df, "cb_rate", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_cb_call(
        ts_code: Optional[str] = None,
        ann_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询可转债赎回信息。积分要求:5000。

        Args:
            ts_code: 转债代码
            ann_date: 公告日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cb_call(
                ts_code=ts_code,
                ann_date=ann_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "cb_call", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_cb_share(
        ts_code: str,
        ann_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询可转债转股结果。积分要求:2000。

        Args:
            ts_code: 转债代码(必填)
            ann_date: 公告日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cb_share(
                ts_code=ts_code,
                ann_date=ann_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "cb_share", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_cb_price_chg(
        ts_code: str,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询可转债转股价变动。需单独权限,积分不足会返回错误。

        Args:
            ts_code: 转债代码(必填)
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cb_price_chg(ts_code=ts_code, fields=fields)
            return format_response(df, "cb_price_chg", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_repo_daily(
        ts_code: Optional[str] = None,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询债券回购日行情。积分要求:2000。

        Args:
            ts_code: 回购代码
            trade_date: 交易日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.repo_daily(
                ts_code=ts_code,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "repo_daily", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

"""
筹码分析 MCP Tools

包含：cyq_chips（筹码分布）, cyq_perf（筹码分布指标）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将筹码分析相关 tools 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_cyq_chips(
        ts_code: str,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询个股筹码分布数据。积分要求:5000+。

        Args:
            ts_code: 股票代码(必填)
            trade_date: 交易日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cyq_chips(
                ts_code=ts_code,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "cyq_chips", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_cyq_perf(
        ts_code: str,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询个股筹码分布指标。积分要求:5000+。

        Args:
            ts_code: 股票代码(必填)
            trade_date: 交易日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.cyq_perf(
                ts_code=ts_code,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "cyq_perf", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

"""
行业分类 MCP Tools

包含：index_classify（申万行业分类）, index_member_all（申万行业成分）,
      ci_daily（中信行业指数日行情）, ci_index_member（中信行业成分）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将行业分类相关 tools 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_index_classify(
        index_code: Optional[str] = None,
        level: Optional[str] = None,
        src: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询申万行业分类（2014版28个一级/2021版31个一级）。积分要求：2000。用于 comps 同行筛选。

        Args:
            index_code: 指数代码
            level: 行业级别（L1一级 L2二级 L3三级）
            src: 指数版本（SW2014 或 SW2021）
            fields: 返回字段
            _format: 输出格式（json/markdown）
            _limit: 最大返回行数
        """
        try:
            df = client.index_classify(
                index_code=index_code,
                level=level,
                src=src,
                fields=fields,
            )
            return format_response(df, "index_classify", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_index_member_all(
        l1_code: Optional[str] = None,
        l2_code: Optional[str] = None,
        l3_code: Optional[str] = None,
        ts_code: Optional[str] = None,
        is_new: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询申万行业成分(分级),支持一/二/三级。积分要求:2000。

        Args:
            l1_code: 一级行业代码
            l2_code: 二级行业代码
            l3_code: 三级行业代码
            ts_code: 股票代码
            is_new: 是否最新(Y/N,默认Y)
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.index_member_all(
                l1_code=l1_code,
                l2_code=l2_code,
                l3_code=l3_code,
                ts_code=ts_code,
                is_new=is_new,
                fields=fields,
            )
            return format_response(df, "index_member_all", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_ci_daily(
        ts_code: Optional[str] = None,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询中信行业指数日行情。积分要求:5000。

        Args:
            ts_code: 行业代码
            trade_date: 交易日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.ci_daily(
                ts_code=ts_code,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "ci_daily", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_ci_index_member(
        l1_code: Optional[str] = None,
        l2_code: Optional[str] = None,
        l3_code: Optional[str] = None,
        ts_code: Optional[str] = None,
        is_new: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询中信行业成分股。积分要求:5000。

        Args:
            l1_code: 一级行业代码
            l2_code: 二级行业代码
            l3_code: 三级行业代码
            ts_code: 股票代码
            is_new: 是否最新(Y/N,默认Y)
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.ci_index_member(
                l1_code=l1_code,
                l2_code=l2_code,
                l3_code=l3_code,
                ts_code=ts_code,
                is_new=is_new,
                fields=fields,
            )
            return format_response(df, "ci_index_member", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

"""
概念板块 MCP Tools

包含：concept（概念板块列表）, concept_detail（概念板块成分股）,
      ths_index（同花顺指数列表）, ths_daily（同花顺指数行情）, ths_member（同花顺概念成分）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将概念板块相关 tools 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_concept(
        trade_date: Optional[str] = None,
        ts_code: Optional[str] = None,
        name: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询概念题材列表(开盘啦数据源),每日盘后更新。积分要求:5000。

        Args:
            trade_date: 交易日期(YYYYMMDD 或 YYYY-MM-DD)
            ts_code: 题材代码(xxxxxx.KP格式)
            name: 题材名称
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.kpl_concept(
                trade_date=trade_date,
                ts_code=ts_code,
                name=name,
                fields=fields,
            )
            return format_response(df, "kpl_concept", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_concept_detail(
        trade_date: Optional[str] = None,
        ts_code: Optional[str] = None,
        con_code: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询概念题材的成分股(开盘啦数据源)。积分要求:5000。

        Args:
            trade_date: 交易日期(YYYYMMDD 或 YYYY-MM-DD)
            ts_code: 题材代码(xxxxxx.KP格式)
            con_code: 成分股代码(如 000001.SZ)
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.kpl_concept_cons(
                trade_date=trade_date,
                ts_code=ts_code,
                con_code=con_code,
                fields=fields,
            )
            return format_response(df, "kpl_concept_cons", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_ths_index(
        ts_code: Optional[str] = None,
        exchange: Optional[str] = None,
        type: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询同花顺概念和行业指数列表。积分要求:6000。

        Args:
            ts_code: 指数代码
            exchange: 市场类型(A-A股 HK-港股 US-美股)
            type: 指数类型(N-概念 I-行业 R-地域 S-特色 ST-风格 TH-主题 BB-宽基)
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.ths_index(
                ts_code=ts_code,
                exchange=exchange,
                type=type,
                fields=fields,
            )
            return format_response(df, "ths_index", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_ths_daily(
        ts_code: Optional[str] = None,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询同花顺板块指数行情。积分要求:6000。

        Args:
            ts_code: 指数代码
            trade_date: 交易日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.ths_daily(
                ts_code=ts_code,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "ths_daily", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_ths_member(
        ts_code: Optional[str] = None,
        con_code: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询同花顺概念板块成分股。积分要求:6000。

        Args:
            ts_code: 板块指数代码
            con_code: 股票代码
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.ths_member(
                ts_code=ts_code,
                con_code=con_code,
                fields=fields,
            )
            return format_response(df, "ths_member", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

"""
公司行为 MCP Tools

包含：dividend（分红送股）, repurchase（股票回购）, share_float（限售股解禁）,
      pledge_stat（股权质押统计）, pledge_detail（股权质押明细）, stk_surv（机构调研）,
      stk_rewards（管理层薪酬和持股）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将公司行为相关 tools 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_dividend(
        ts_code: Optional[str] = None,
        ann_date: Optional[str] = None,
        record_date: Optional[str] = None,
        ex_date: Optional[str] = None,
        imp_ann_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询分红送股数据,含每股分红、送转比例、除权除息日等。积分要求:2000。
        注意:参数中至少一个不能为空。

        Args:
            ts_code: 股票代码(如 000001.SZ)
            ann_date: 预案公告日
            record_date: 股权登记日期
            ex_date: 除权除息日
            imp_ann_date: 实施公告日
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.dividend(
                ts_code=ts_code,
                ann_date=ann_date,
                record_date=record_date,
                ex_date=ex_date,
                imp_ann_date=imp_ann_date,
                fields=fields,
            )
            return format_response(df, "dividend", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_repurchase(
        ann_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询上市公司回购股票数据。积分要求:600。

        Args:
            ann_date: 公告日期
            start_date: 公告开始日期
            end_date: 公告结束日期
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.repurchase(
                ann_date=ann_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "repurchase", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_share_float(
        ts_code: Optional[str] = None,
        ann_date: Optional[str] = None,
        float_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询限售股解禁数据。积分要求:120。

        Args:
            ts_code: 股票代码
            ann_date: 公告日期
            float_date: 解禁日期
            start_date: 解禁开始日期
            end_date: 解禁结束日期
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.share_float(
                ts_code=ts_code,
                ann_date=ann_date,
                float_date=float_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "share_float", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_pledge_stat(
        ts_code: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询股票质押统计数据,含质押次数、质押比例等。积分要求:500。

        Args:
            ts_code: 股票代码
            end_date: 截止日期
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.pledge_stat(
                ts_code=ts_code,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "pledge_stat", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_pledge_detail(
        ts_code: str,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询股票质押明细数据,含质押方、质押数量、起止日期等。积分要求:500。

        Args:
            ts_code: 股票代码(如 000001.SZ,必填)
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.pledge_detail(
                ts_code=ts_code,
                fields=fields,
            )
            return format_response(df, "pledge_detail", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_stk_surv(
        ts_code: Optional[str] = None,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询上市公司机构调研记录数据。积分要求:5000。

        Args:
            ts_code: 股票代码
            trade_date: 调研日期
            start_date: 调研开始日期
            end_date: 调研结束日期
            fields: 返回字段,逗号分隔
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.stk_surv(
                ts_code=ts_code,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "stk_surv", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_stk_rewards(
        ts_code: str,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询管理层薪酬和持股数据。积分要求:2000。

        Args:
            ts_code: TS股票代码(必填,支持多个)
            end_date: 报告期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.stk_rewards(
                ts_code=ts_code,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "stk_rewards", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

"""
数据存储 MCP Tools

提供本地数据查询能力：
- tushare_data_index: 查询保存记录索引
- tushare_data_read: 从本地 DB 读取历史数据（不调 API）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将数据存储查询 tool 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_data_index(
        api_name: Optional[str] = None,
        ts_code: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        _format: str = "json",
        _limit: int = 50,
    ) -> dict:
        """查询本地已保存的数据索引。每次调用 tushare_* tool 获取的数据都会自动保存到本地 SQLite。
        使用此接口可以查看历史保存了哪些数据，方便回溯。

        Args:
            api_name: 按接口名筛选（如 daily, income, index_daily）
            ts_code: 按股票/标的代码筛选（如 000001.SZ）
            start_date: 保存时间起始（YYYY-MM-DD），筛选 saved_at 字段
            end_date: 保存时间截止（YYYY-MM-DD），筛选 saved_at 字段
            _format: 输出格式（json/markdown），默认 json
            _limit: 最大返回行数，默认 50
        """
        try:
            store = client._get_store()
            df = store.query_index(
                api_name=api_name,
                ts_code=ts_code,
                start_date=start_date,
                end_date=end_date,
                limit=_limit,
            )
            return format_response(df, "data_index", _format, _limit)
        except Exception as e:
            return {"error": f"查询数据索引失败: {e}"}

    @mcp.tool()
    def tushare_data_read(
        query_id: Optional[int] = None,
        api_name: Optional[str] = None,
        ts_code: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """从本地数据库读取历史保存的数据，不调用 tushare API。
        三种查询模式：
        1. 按 query_id 读取某次查询的完整数据（最精确）
        2. 按 api_name 读取某类数据的最新记录
        3. 按 api_name + ts_code 读取特定标的的历史数据

        建议先用 tushare_data_index 查到 query_id，再用本接口读取。

        Args:
            query_id: 索引记录 id（从 tushare_data_index 获取）
            api_name: 接口名称（如 daily, income, index_daily）
            ts_code: 股票/标的代码（需配合 api_name 使用）
            _format: 输出格式（json/markdown），默认 json
            _limit: 最大返回行数，默认 100
        """
        try:
            store = client._get_store()
            df = store.read_data(
                query_id=query_id,
                api_name=api_name,
                ts_code=ts_code,
                limit=_limit,
            )
            source = f"local:{api_name or 'query_' + str(query_id)}"
            return format_response(df, source, _format, _limit)
        except Exception as e:
            return {"error": f"读取本地数据失败: {e}"}

"""
东财板块 MCP Tools

包含：dc_index（东财概念和行业指数行情）, dc_member（东财板块成分）, dc_hot（东财人气榜）
"""

from typing import Optional

from fastmcp import FastMCP

from .. import client
from ..formatter import format_response
from ..errors import TushareError


def register(mcp: FastMCP) -> None:
    """将东财板块相关 tools 注册到 MCP 实例"""

    @mcp.tool()
    def tushare_dc_index(
        ts_code: Optional[str] = None,
        name: Optional[str] = None,
        trade_date: Optional[str] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询东财概念和行业指数行情。积分要求:6000。

        Args:
            ts_code: 指数代码
            name: 指数名称
            trade_date: 交易日期
            start_date: 开始日期
            end_date: 结束日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.dc_index(
                ts_code=ts_code,
                name=name,
                trade_date=trade_date,
                start_date=start_date,
                end_date=end_date,
                fields=fields,
            )
            return format_response(df, "dc_index", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_dc_member(
        ts_code: Optional[str] = None,
        con_code: Optional[str] = None,
        trade_date: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询东财概念和行业板块成分。积分要求:6000。

        Args:
            ts_code: 板块指数代码
            con_code: 成分股代码
            trade_date: 交易日期
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.dc_member(
                ts_code=ts_code,
                con_code=con_code,
                trade_date=trade_date,
                fields=fields,
            )
            return format_response(df, "dc_member", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}

    @mcp.tool()
    def tushare_dc_hot(
        trade_date: Optional[str] = None,
        ts_code: Optional[str] = None,
        market: Optional[str] = None,
        hot_type: Optional[str] = None,
        is_new: Optional[str] = None,
        fields: Optional[str] = None,
        _format: str = "json",
        _limit: int = 100,
    ) -> dict:
        """查询东财人气榜(热股数据)。积分要求:8000。

        Args:
            trade_date: 交易日期
            ts_code: 股票代码
            market: 市场
            hot_type: 热度类型
            is_new: 是否最新
            fields: 返回字段
            _format: 输出格式(json/markdown),默认json
            _limit: 最大返回行数,默认100
        """
        try:
            df = client.dc_hot(
                trade_date=trade_date,
                ts_code=ts_code,
                market=market,
                hot_type=hot_type,
                is_new=is_new,
                fields=fields,
            )
            return format_response(df, "dc_hot", _format, _limit)
        except TushareError as e:
            return {"error": str(e)}