通达信TdxQuant 实战：一招搞定“外部调用”与“数据清洗”两大痛点

import sys
import os

# 1. 定义通达信安装根目录
tdx_install_path = r"D:\lwj\new_tdx"  # 请修改为你自己的通达信路径

# 2. 拼接出 PYPlugins/user 的绝对路径
pyplugins_user_path = os.path.join(tdx_install_path, "PYPlugins", "user")

# 3. 将该路径插入到 sys.path 的第一位，确保优先加载
sys.path.insert(0, pyplugins_user_path)

# 4. 现在可以愉快地导入了
from tqcenter import tq

def raw_data_to_dataframe_complete(raw_data):
    """
    通用转换函数：将原始字典数据转换为标准的 DataFrame
    """
    if not raw_data:
        return pd.DataFrame()

    fields = list(raw_data.keys())
    # 取第一个字段获取股票列表和日期索引
    first_field = fields[0]
    first_df = raw_data[first_field]

    # 确保原始数据是DataFrame格式
    if isinstance(first_df, pd.DataFrame) and hasattr(first_df, 'columns'):
        stock_codes = first_df.columns.tolist()
        df_list = []

        # 遍历每一只股票
        for stock_code in stock_codes:
            stock_data = {}

            # 遍历每一个字段，提取该股票的数据
            for field in fields:
                field_data = raw_data[field]

                # 安全获取数据
                if isinstance(field_data, pd.DataFrame) and stock_code in field_data.columns:
                    stock_data[field] = field_data[stock_code].values
                else:
                    stock_data[field] = np.nan

            # 构建该股票的临时DataFrame
            stock_df = pd.DataFrame(stock_data)
            stock_df['股票代码'] = stock_code

            # 补充日期索引
            if hasattr(first_df, 'index'):
                stock_df['日期'] = first_df.index

            df_list.append(stock_df)

        # 合并所有股票数据
        if df_list:
            final_df = pd.concat(df_list, ignore_index=True)
            # 调整列顺序：代码、日期、指标...
            col_order = ['股票代码', '日期'] + fields
            return final_df[col_order]

    return pd.DataFrame()

# -*- coding: utf-8 -*-
import sys
import os
import pandas as pd
import numpy as np

# ---------------------------------------------------------
# 关键步骤1: 将通达信的PYPlugins/user目录添加到Python的搜索路径中
# 【外部调用解决方案】
tdx_install_path = r"D:\lwj\new_tdx"  # <<--- 请修改为你的通达信安装根目录
pyplugins_user_path = os.path.join(tdx_install_path, "PYPlugins", "user")
sys.path.insert(0, pyplugins_user_path)
# ---------------------------------------------------------

# ---------------------------------------------------------
# 关键步骤2: 导入tqcenter模块
from tqcenter import tq
# ---------------------------------------------------------

def raw_data_to_dataframe_complete(raw_data):
    """
    将tq.get_market_data返回的原始数据完整转换为DataFrame
    支持多股票、多字段的完整转换
    【Series转DataFrame/数据清洗解决方案】
    """
    # 检查数据是否为空
    if not raw_data:
        print("原始数据为空")
        return pd.DataFrame()

    # 获取所有字段
    fields = list(raw_data.keys())

    # 获取第一个字段的DataFrame来提取股票代码信息
    first_field = fields[0]
    first_df = raw_data[first_field]

    if isinstance(first_df, pd.DataFrame) and hasattr(first_df, 'columns'):
        # 获取所有股票代码
        stock_codes = first_df.columns.tolist()

        # 创建合并后的DataFrame
        df_list = []

        for stock_code in stock_codes:
            stock_data = {}

            # 提取每个字段下该股票的数据
            for field in fields:
                field_data = raw_data[field]

                if isinstance(field_data, pd.DataFrame) and stock_code in field_data.columns:
                    stock_data[field] = field_data[stock_code].values
                elif isinstance(field_data, pd.Series):
                    # 如果是Series，尝试通过索引获取
                    stock_data[field] = field_data.values
                else:
                    stock_data[field] = np.nan  # 填充缺失值

            # 创建该股票的DataFrame
            if stock_data[first_field] is not np.nan:
                stock_df = pd.DataFrame(stock_data)
                stock_df['股票代码'] = stock_code

                # 添加日期索引
                if hasattr(first_df, 'index'):
                    stock_df['日期'] = first_df.index
                else:
                    stock_df['日期'] = range(len(stock_df))

                df_list.append(stock_df)

        # 合并所有股票的数据
        if df_list:
            final_df = pd.concat(df_list, ignore_index=True)
            # 重新排序列
            col_order = ['股票代码', '日期'] + fields
            final_df = final_df[col_order]
            return final_df

    return pd.DataFrame()


def main():
    # ---------------------------------------------------------
    # 关键步骤3: 初始化与通达信的连接
    try:
        # __file__ 表示当前脚本路径，用于让通达信定位调用源
        tq.initialize(__file__)
        print("成功与通达信建立连接！")
    except Exception as e:
        print(f"初始化连接失败，请检查通达信是否已运行。错误信息: {e}")
        return
    # ---------------------------------------------------------

    # 定义参数
    stock_list = ["600519.SH"]  # 贵州茅台
    start_date = "20260105"    # 起始日期 (建议改为近期有数据的日期)
    end_date = "20260120"      # 结束日期
    period = '1d'              # K线周期：日线

    # ---------------------------------------------------------
    # 关键步骤4: 调用数据接口获取数据
    try:
        # 刷新行情缓存，确保获取最新数据
        refresh_cache = tq.refresh_cache()
        print(f"刷新行情缓存结果: {refresh_cache}")

        # 获取原始数据
        raw_data = tq.get_market_data(
            stock_list=stock_list,
            start_time=start_date,
            end_time=end_date,
            period=period,
            dividend_type='none',
            field_list=['Open', 'High', 'Low', 'Close', 'Amount'],
            fill_data=True
        )

        # -----------------------------------------------------
        # 关键步骤5: 转换数据格式
        # -----------------------------------------------------
        df_standard = raw_data_to_dataframe_complete(raw_data)
        print("\n标准格式DataFrame (前5行):")
        print(df_standard.head())

    except Exception as e:
        print(f"获取或处理数据时出错: {e}")
        import traceback
        traceback.print_exc()

    finally:
        # ---------------------------------------------------------
        # 关键步骤6: 清理资源，防止通达信卡死
        try:
            tq.close()
            print("通达信连接已清理")
        except:
            pass


if __name__ == "__main__":
    main()

如果我的分享对你有所帮助， 不吝啬给个点赞呗