通达信和同花顺本地数据解析笔记

Slug: /通达信和同花顺本地数据解析笔记.html 阅读: 读取中
标签: 量化 本地数据 通达信 同花顺

背景

做 A 股回测时，经常需要三类本地数据：

  • 历史某一天的总市值、流通市值
  • 股票属于哪些通达信概念/板块
  • 股票属于哪些同花顺概念/行业板块

这些数据不一定要一开始就接第三方数据库。通达信和同花顺客户端本地缓存里已经有一部分可用数据，可以先用它们搭一个自动化流程。

下面记录的是我本机验证过的路径和解析方式。

通达信历史市值

相关文件

通达信安装目录：

D:\dev\tongdaxin(上证指数)

股本变动文件：

D:\dev\tongdaxin(上证指数)\T0002\hq_cache\gbbq

日线文件：

D:\dev\tongdaxin(上证指数)\vipdoc\sh\lday\sh601857.day
D:\dev\tongdaxin(上证指数)\vipdoc\sz\lday\sz000001.day

gbbq 是事件表，不是每日快照。它记录除权除息、送配股上市、股本变化、增发、回购等事件。要得到某一天的股本，需要取这一天之前最近的一条股本相关事件。

我本机当前 gbbq 的解析结果：

全部事件: 196941 条
全部事件日期范围: 1990-03-01 到 2026-06-11
股本相关事件: 132627 条
股本相关事件日期范围: 1990-12-10 到 2026-06-10

几个例子：

000001: 股本记录从 1991-04-03 开始
600028: 股本记录从 2001-11-08 开始
601857: 股本记录从 2007-11-05 开始

gbbq 字段含义

可以直接用 pytdx 的 GbbqReader 读取：

from pathlib import Path
from pytdx.reader.gbbq_reader import GbbqReader

tdx = next(p for p in Path(r"D:\dev").iterdir() if p.is_dir() and p.name.startswith("tongdaxin("))
gbbq_path = tdx / "T0002" / "hq_cache" / "gbbq"

df = GbbqReader().get_df(str(gbbq_path))
print(df.head())

主要字段：

market
code
datetime
category
hongli_panqianliutong
peigujia_qianzongguben
songgu_qianzongguben
peigu_houzongguben

常用的 category：

1  除权除息
2  送配股上市
3  非流通股上市
4  未知股本变动
5  股本变化
6  增发新股
7  股份回购
8  增发新股上市
9  转配股上市
10 可转债上市
11 扩缩股
12 非流通股缩股

注意：category=1 是分红除权行，这一类行的字段含义不是股本，不应该拿来算总市值。

对股本相关行，可以按下面理解：

hongli_panqianliutong   变动前流通股本
peigujia_qianzongguben  变动前总股本
songgu_qianzongguben    变动后流通股本
peigu_houzongguben      变动后总股本

单位看起来是“万股”。例如中国石油 601857 的总股本字段出现 18302098.0，对应：

18302098 万股 = 183020980000 股

这和中国石油约 1830 亿股总股本一致。

day 文件结构

通达信 .day 日线文件是 32 字节一条记录：

import struct

RECORD = struct.Struct("<IIIIIfII")

def read_day_file(path):
    rows = []
    with open(path, "rb") as f:
        data = f.read()

    for offset in range(0, len(data), RECORD.size):
        date, open_, high, low, close, amount, volume, _ = RECORD.unpack_from(data, offset)
        rows.append({
            "date": date,
            "open": open_ / 100,
            "high": high / 100,
            "low": low / 100,
            "close": close / 100,
            "amount": amount,
            "volume": volume,
        })

    return rows

推导历史市值

如果要算 2023-01-31 的总市值：

  1. 从 .day 里取目标日或目标日前最近交易日的收盘价。
  2. 从 gbbq 里取 datetime <= 20230131 的最后一条股本相关记录。
  3. 用收盘价乘总股本。

公式：

总市值 = close * peigu_houzongguben * 10000
流通市值 = close * songgu_qianzongguben * 10000

如果要用“亿元”为单位：

总市值_亿元 = close * peigu_houzongguben * 10000 / 100000000
流通市值_亿元 = close * songgu_qianzongguben * 10000 / 100000000

示例代码：

from pathlib import Path
import struct
import pandas as pd
from pytdx.reader.gbbq_reader import GbbqReader

DAY_RECORD = struct.Struct("<IIIIIfII")
SHARE_CATEGORIES = {2, 3, 4, 5, 6, 7, 8, 9, 11, 12}

def find_tdx_root():
    return next(p for p in Path(r"D:\dev").iterdir() if p.is_dir() and p.name.startswith("tongdaxin("))

def read_day_close(path, target_date):
    target_date = int(target_date)
    last = None

    with open(path, "rb") as f:
        data = f.read()

    for offset in range(0, len(data), DAY_RECORD.size):
        date, open_, high, low, close, amount, volume, _ = DAY_RECORD.unpack_from(data, offset)
        if date <= target_date:
            last = {
                "trade_date": date,
                "close": close / 100,
                "amount": amount,
                "volume": volume,
            }
        else:
            break

    return last

def market_from_code(code):
    if code.startswith(("5", "6", "9")):
        return "sh"
    return "sz"

def day_path(tdx_root, code):
    market = market_from_code(code)
    return tdx_root / "vipdoc" / market / "lday" / f"{market}{code}.day"

def latest_share_table(tdx_root, target_date):
    gbbq = GbbqReader().get_df(str(tdx_root / "T0002" / "hq_cache" / "gbbq"))
    gbbq = gbbq[gbbq["category"].isin(SHARE_CATEGORIES)]
    gbbq = gbbq[gbbq["datetime"] <= int(target_date)]
    gbbq = gbbq[gbbq["peigu_houzongguben"] > 0]
    gbbq = gbbq.sort_values(["code", "datetime"])
    return gbbq.groupby("code", as_index=False).tail(1)

def rank_market_cap(target_date="20230131", topn=100):
    tdx = find_tdx_root()
    shares = latest_share_table(tdx, target_date)

    rows = []
    for row in shares.itertuples(index=False):
        code = row.code
        p = day_path(tdx, code)
        if not p.exists():
            continue

        day = read_day_close(p, target_date)
        if not day:
            continue

        total_mv_yi = day["close"] * row.peigu_houzongguben * 10000 / 100000000
        float_mv_yi = day["close"] * row.songgu_qianzongguben * 10000 / 100000000

        rows.append({
            "code": code,
            "trade_date": day["trade_date"],
            "close": day["close"],
            "total_share_10k": row.peigu_houzongguben,
            "float_share_10k": row.songgu_qianzongguben,
            "total_mv_yi": total_mv_yi,
            "float_mv_yi": float_mv_yi,
            "share_event_date": row.datetime,
        })

    result = pd.DataFrame(rows)
    return result.sort_values("total_mv_yi", ascending=False).head(topn)

if __name__ == "__main__":
    print(rank_market_cap("20230131", 100).to_string(index=False))

这个方法适合推导历史市值。它的关键优点是股本来自历史事件，不是当前快照。

通达信概念板块

相关文件

通达信概念、风格、指数/板块文件在：

D:\dev\tongdaxin(上证指数)\T0002\hq_cache\block_gn.dat
D:\dev\tongdaxin(上证指数)\T0002\hq_cache\block_fg.dat
D:\dev\tongdaxin(上证指数)\T0002\hq_cache\block_zs.dat

行业分类文件：

D:\dev\tongdaxin(上证指数)\T0002\hq_cache\tdxhy.cfg

几个文件的用途：

block_gn.dat  概念板块
block_fg.dat  风格板块
block_zs.dat  指数/板块
tdxhy.cfg     股票到通达信行业分类代码的映射

block_gn.dat 可以用 pytdx.reader.block_reader.BlockReader 直接解析：

from pathlib import Path
from pytdx.reader.block_reader import BlockReader

tdx = next(p for p in Path(r"D:\dev").iterdir() if p.is_dir() and p.name.startswith("tongdaxin("))
path = tdx / "T0002" / "hq_cache" / "block_gn.dat"

df = BlockReader().get_df(str(path))
print(df.head())

输出字段：

blockname   板块名
block_type  板块类型
code_index  股票在板块内的序号
code        股票代码

如果想查一只股票属于哪些通达信概念：

def tdx_concepts_of(code):
    tdx = next(p for p in Path(r"D:\dev").iterdir() if p.is_dir() and p.name.startswith("tongdaxin("))
    path = tdx / "T0002" / "hq_cache" / "block_gn.dat"
    df = BlockReader().get_df(str(path))
    return sorted(df.loc[df["code"] == code, "blockname"].unique())

print(tdx_concepts_of("601857"))

tdxhy.cfg 是文本格式，一行一个股票：

0|000001|T1001|||X500102
0|000002|T110201|||X530101

其中：

第 1 列: 市场
第 2 列: 股票代码
第 3 列: 通达信行业代码
第 6 列: 另一个行业/分类代码

它可以做行业代码映射，但如果要行业名称，还需要再找通达信本地的行业树/配置文件做代码到名称的映射。单纯拿概念板块时，block_gn.dat 更直接。

回测注意事项

通达信 block_gn.dat、block_fg.dat、block_zs.dat 是当前客户端缓存下来的板块快照。它们适合做当前股票标签，不适合直接当作历史某日概念成分。

如果在历史回测里使用当前概念成分，会有未来函数。除非你的策略定义就是“用当前概念标签回看历史表现”，否则需要历史概念成分数据源。

同花顺概念和成分股

相关文件

同花顺安装目录：

C:\同花顺软件\同花顺

概念板块：

C:\同花顺软件\同花顺\BlockUpdate\block_conception.ini

行业板块：

C:\同花顺软件\同花顺\BlockUpdate\block_industry.ini

同花顺方案里的全集缓存：

C:\同花顺软件\同花顺\system\同花顺方案\StockBlock.ini

我本机当前验证结果：

block_conception.ini:
  概念名称数: 396
  有成分股的概念板块: 388

block_industry.ini:
  行业名称数: 356
  有成分股的行业板块: 257

StockBlock.ini:
  名称数: 2880
  有成分股的板块: 2209

优先级：

只要同花顺概念: block_conception.ini
只要同花顺行业: block_industry.ini
想要混合标签全集: StockBlock.ini

文件结构

block_conception.ini 是 GBK/ANSI 文本。核心两段：

[BLOCK_NAME_MAP_TABLE]
CBE8=氢能源
...

[BLOCK_STOCK_CONTEXT]
CBE8=33:000009,33:000027,17:600028,17:601857,...
...

含义：

[BLOCK_NAME_MAP_TABLE]   板块 ID 到板块名称
[BLOCK_STOCK_CONTEXT]    板块 ID 到成分股列表

成分股前面的数字是市场/证券类别标记。常见例子：

33:000001  深市股票
17:600000  沪市股票

做股票标签时，可以先只取冒号后面的 6 位股票代码。

解析代码

from pathlib import Path

def parse_ths_block_ini(path):
    path = Path(path)
    text = path.read_text(encoding="gbk", errors="replace")

    section = None
    names = {}
    members = {}

    for raw in text.splitlines():
        line = raw.strip()
        if not line or line.startswith(";"):
            continue

        if line.startswith("[") and line.endswith("]"):
            section = line[1:-1]
            continue

        if "=" not in line:
            continue

        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip()

        if section == "BLOCK_NAME_MAP_TABLE" and value:
            names[key] = value
        elif section == "BLOCK_STOCK_CONTEXT" and value:
            codes = []
            for item in value.split(","):
                item = item.strip()
                if not item:
                    continue
                codes.append(item.split(":")[-1])
            members[key] = codes

    rows = []
    for block_id, codes in members.items():
        block_name = names.get(block_id, block_id)
        for code in codes:
            rows.append({
                "block_id": block_id,
                "block_name": block_name,
                "code": code,
            })

    return rows

def ths_concepts_of(code):
    root = Path(r"C:\同花顺软件\同花顺")
    rows = parse_ths_block_ini(root / "BlockUpdate" / "block_conception.ini")
    return sorted({row["block_name"] for row in rows if row["code"] == code})

print(ths_concepts_of("601857"))

我本机解析 601857 中国石油得到的同花顺概念包括：

氢能源
融资融券
中字头股票
证金持股
参股保险
央企国企改革
沪股通
一带一路
天然气
页岩气
俄乌冲突概念
碳交易
国企改革
同花顺中特估100
高股息精选

同花顺行业示例：

601857 -> 石油加工
600519 -> 白酒Ⅲ
600028 -> 石油加工

建议的数据管线

如果目标是做回测，可以按下面组织：

1. 每天或每次启动前同步客户端数据
2. 用通达信 vipdoc 读取日线收盘价
3. 用通达信 gbbq 还原历史总股本/流通股本
4. 生成每日 total_mv / float_mv
5. 用通达信或同花顺板块文件生成当前股票标签
6. 回测时明确区分“历史可得数据”和“当前静态标签”

市值排序这类历史指标，优先使用：

通达信 gbbq + 通达信 day

概念/行业标签，当前快照可以使用：

通达信 block_gn.dat
同花顺 block_conception.ini
同花顺 block_industry.ini

需要特别注意：

gbbq 是历史事件数据，适合推历史市值。
概念板块文件多数是当前快照，不等于历史成分。
目录
build:   20260615