通达信数据下载V8.0极速小工具 免费数据下载工具(电脑版)

核心代码:（Python ）

# -*- coding: utf-8 -*-

import os

import struct

import time

import datetime

import requests

from concurrent.futures import ThreadPoolExecutor, as_completed

import logging

# 配置日志

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

TDX_DATA_DIR = r"D:\new_tdx\vipdoc"  #←←← 请改成你的通达信安装路径

def get_market_prefix(code):

    if code.startswith(('6', '9')):

        return 'sh'

    elif code.startswith(('0', '3')):

        return 'sz'

    else:

        return None

def fetch_sina_daily(code):

    """

    返回: [(date, open, high, low, close, vol), ...]

    """

    prefix = get_market_prefix(code)

    if not prefix:

        return []

    url = f"http://money.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketData.getKLineData?symbol={prefix}{code}&scale=240&ma=5&datetype=1&candletype=1"

    try:

        resp = requests.get(url, timeout=10)

        if resp.status_code != 200:

            return []

        data = resp.json()

        if not isinstance(data, list):

            return []

        result = []

        for item in data:

            date_str = item['day'].replace('-', '')

            date_int = int(date_str)

            open_p = float(item['open'])

            high = float(item['high'])

            low = float(item['low'])

            close = float(item['close'])

            vol = int(float(item['volume']) / 100)  

            result.append((date_int, open_p, high, low, close, vol))

        return result

    except Exception as e:

        logging.error(f"获取 {code} 失败: {e}")

        return []

# 将数据写入通达信 .day 文件

def write_tdx_day_file(code, klines):

    prefix = get_market_prefix(code)

    if not prefix:

        return

    filename = os.path.join(TDX_DATA_DIR, f"{prefix}{code}.day")

    os.makedirs(TDX_DATA_DIR, exist_ok=True)

    # 读取已有数据（避免重复）

    existing_dates = set()

    if os.path.exists(filename):

        with open(filename, "rb") as f:

            while True:

                buf = f.read(32)

                if len(buf) < 32:

                    break

                date, = struct.unpack('<I', buf[:4])

                existing_dates.add(date)

    # 过滤新数据

    new_klines = [k for k in klines if k[0] not in existing_dates]

    if not new_klines:

        return

    # 写入

    with open(filename, "ab") as f:

        for k in sorted(new_klines, key=lambda x: x[0]): 

            date, open_p, high, low, close, vol = k

            buf = struct.pack(

                '<IffffII',

                date,

                open_p * 100,   

                high * 100,

                low * 100,

                close * 100,

                vol,

                0 

            )

            f.write(buf)

    logging.info(f"更新 {code} 新增 {len(new_klines)} 条记录")

# 下载

def download_stock(code):

    try:

        klines = fetch_sina_daily(code)

        if klines:

            write_tdx_day_file(code, klines)

    except Exception as e:

        logging.error(f"处理 {code} 异常: {e}")

def main():

    stock_list = [

        '600000', '000001', '300750', '601318', '002475',

        '600519', '000858', '603259', '300059', '688981'

    ]  

    logging.info(f"开始下载 {len(stock_list)}...")

    with ThreadPoolExecutor(max_workers=12) as executor:

        futures = {executor.submit(download_stock, code): code for code in stock_list}

        for future in as_completed(futures):

            code = futures[future]

            try:

                future.result()

            except Exception as e:

                logging.error(f"{code}: {e}")

    logging.info("全部下载完成！")

if __name__ == "__main__":

    main()