feat: blocks 命令 — 本地板块文件同步（行业/概念/指数/地区）

.claude/skills/tdx2db-query/SKILL.md

@@ -49,6 +49,22 @@ ORDER BY datetime;
 
 **周线/月线**：无独立表，取日线后 pandas `resample('W')`/`resample('ME')` 聚合（open=first, high=max, low=min, close=last, volume/amount=sum）。
 
+**板块成分 / 个股板块归属**（block_type ∈ 行业/概念/指数/地区/风格/特殊）
+```sql
+-- 板块 → 成分（行业为 881 研究行业，一/二/三级各一行，按名或 block_code 定位）
+SELECT code FROM block_stock_relation
+WHERE block_type = '行业' AND block_name = '煤炭开采';
+
+-- 个股 → 全部板块归属
+SELECT block_type, block_code, block_name FROM block_stock_relation WHERE code = :code6;
+
+-- 板块成分 JOIN 行情（北交所成员在行情表无数据，JOIN 自然过滤）
+SELECT d.code, d.close, d.ma233
+FROM block_stock_relation b JOIN daily_data d ON d.code = b.code
+WHERE b.block_type = '概念' AND b.block_name = '人形机器人' AND d.date = :d;
+```
+板块表为全量快照（随 sync 更新，无历史版本）；板块名以 tdxzs.cfg 官方全名为准，跨口径对齐用 block_code。
+
 **覆盖度自检**（跑批前防"同步不完整静默算子集"）
 ```sql
 SELECT

AGENTS.md

@@ -28,6 +28,7 @@ tdx2db stock-list --db-only  # 同步股票列表
 | `daily_data` | (code, date) | 日线 OHLCV + 11 条均线 |
 | `minute{5,15,30,60}_data` | (code, datetime) | 分钟线；15/30/60 由 5 分钟重采样 |
 | `stock_info` | code | 股票列表（name 是占位符，见"陷阱"） |
+| `block_stock_relation` | (block_type, block_name, code) | 板块-个股关系，全量快照；block_type ∈ 行业/概念/指数/地区/风格/特殊 |
 
 - 行情表通用列：`code, market, datetime, date, open, high, low, close, volume, amount, ma5, ma10, ma13, ma21, ma34, ma55, ma60, ma89, ma144, ma233, ma250`
 - 均线：`ma13/21/34/55/89/144/233` 是斐波那契窗口（缠论强弱分析主力），`ma233` 常被当作年线；上市天数不足窗口的行为 NULL
@@ -51,8 +52,18 @@ WHERE date = '2026-07-03' AND ma233 IS NOT NULL;
 
 -- 3. 最新交易日探测（加 ma5 条件确保当日指标已计算完成）
 SELECT MAX(date) FROM daily_data WHERE ma5 IS NOT NULL;
+
+-- 4. 按板块取成分（行业含一/二/三级，用 block_name 或 block_code 定位）
+SELECT code FROM block_stock_relation
+WHERE block_type = '行业' AND block_name = '煤炭开采';
+
+-- 5. 按个股查板块归属
+SELECT block_type, block_code, block_name FROM block_stock_relation
+WHERE code = '000001';
 ```
 
+板块注意：行业是 881"研究行业"口径（多级，一股多行）；板块名以 `tdxzs.cfg` 官方全名为准，与通达信软件导出 CSV 偶有变体（如"碳纤维概念" vs 导出的"碳纤维"），跨口径对齐用 `block_code`；关系表为全量快照，无历史版本。
+
 ## 陷阱（每一条都有真实事故背书）
 
 1. **code 格式跨表不一致**：`stock_info.code` 带前缀（`sz000001`），行情表是 6 位纯数字（`000001`）。用 `LIKE 'sh688%'` 查 `daily_data` 会静默零匹配。跨表需 `RIGHT(stock_info.code, 6)`——但内部实践根本不用 `stock_info`（见 3）。

CLAUDE.md

@@ -26,8 +26,13 @@ tdx2db stock-list --db-only
 
 # 数据库状态一览（只读，不需要 TDX_PATH；--json 机器可读）
 tdx2db status
+
+# 板块-个股关系同步（sync 已自动包含，也可单独跑）
+tdx2db blocks --db-only
 ```
 
+板块数据链（blocks.py，issue #39）：概念/指数/风格 ← `T0002/hq_cache/infoharbor_block.dat`（文本，板块名 8 字节截断需经 880 码还原全名）；行业 ← `tdxhy.cfg` X 码 × `tdxzs3.cfg` 类别 12（881 研究行业多级）；跨市场指数成分 ← `spblock.dat`；地区 ← `base.dbf` DY 字段 × `tdxzs.cfg` 类别 3。全量替换式快照（DELETE+INSERT），与行情表的增量语义不同。旧二进制 block_zs/gn/fg.dat 在新版通达信不存在，pytdx BlockReader 不适用。
+
 `python main.py <子命令>` 与 `tdx2db <子命令>` 等价（main.py 是薄封装，保留老用户习惯）。包目录为 `tdx2db/`（v0.3.0 起从 `src/` 改名，发布到 PyPI）。psycopg2-binary/pymysql 为可选依赖（extras: postgres/mysql/all），默认安装仅支持 SQLite。
 
 测试：`pytest tests/`。CI 矩阵：ubuntu + windows × 3.9/3.11（+ ubuntu 3.10）——Windows 是用户主场景必须覆盖。`tests/fixtures/` 存 .day/.lc5 二进制解析 fixture（合成字节，生成器 `make_fixtures.py`；断言值与生成参数一一对应）。数据正确性验证仍以运行 `sync` 后检查数据库为准。

README.md

@@ -175,6 +175,9 @@ tdx2db minutes --csv-only
 | `daily_data` | (code, date) | 日线 OHLCV + 均线 |
 | `minute5_data` / `minute15_data` / `minute30_data` / `minute60_data` | (code, datetime) | 分钟线 OHLCV + 均线（15/30/60 由 5 分钟重采样） |
 | `stock_info` | code | 股票列表 |
+| `block_stock_relation` | (block_type, block_name, code) | 板块-个股关系（行业/概念/指数/地区/风格/特殊），全量快照 |
+
+**板块数据**：来自通达信本地板块文件（`T0002/hq_cache/`），随 `sync` 自动更新，也可单独 `tdx2db blocks --db-only`。行业为 881 研究行业（一/二/三级各一行）；中证500/1000 等跨市场指数成分完整；每次同步为全量替换快照（无历史版本）。老用户需执行一次 `scripts/migrate_block_relation.sql`（表结构变更，原表从未有写入路径）。
 
 **均线列**：`ma5 / ma10 / ma60 / ma250` 为常规窗口，`ma13 / ma21 / ma34 / ma55 / ma89 / ma144 / ma233` 为斐波那契窗口（服务缠论类分析，不需要可忽略）。上市不足对应窗口天数的行为 NULL。
 

scripts/migrate_block_relation.sql

@@ -0,0 +1,10 @@
+-- issue #39: block_stock_relation 表结构变更
+--   新增 block_type 列；唯一约束从 (block_code, code) 改为 (block_type, block_name, code)；
+--   移除 name 列（板块文件不含股票名称，与"行情表不存名称"契约一致）。
+--
+-- 该表在本次变更前从未有 CLI 写入路径（功能未实现），正常情况下为空表，直接重建：
+-- PostgreSQL / MySQL / SQLite 通用。执行前请确认表中没有你自行写入的数据。
+
+DROP TABLE IF EXISTS block_stock_relation;
+
+-- 之后运行任意写库命令（如 tdx2db blocks --db-only）会自动按新结构建表。

tdx2db/blocks.py

@@ -0,0 +1,227 @@
+"""板块数据解析模块（issue #39）
+
+四条数据链，均为通达信本地文件（盘后数据下载时自动更新）：
+
+- 概念/指数/风格: T0002/hq_cache/infoharbor_block.dat（文本 GBK）
+- 行业:          tdxhy.cfg（个股 X 码）× tdxzs3.cfg 类别 12（881 研究行业，多级前缀匹配）
+- 指数补充/特殊:  spblock.dat（中证500/1000 等跨市场指数成分仅此处有）
+- 地区:          base.dbf 的 DY 字段 × tdxzs.cfg 类别 3
+
+历史注记：旧版通达信的二进制 block_zs/gn/fg.dat 在新版已不存在，
+pytdx BlockReader 不适用；以上文件均自行解析，无新增依赖。
+格式均经真实文件与导出 CSV 全量比对验证（2026-07-07，issue #39）。
+"""
+import struct
+from pathlib import Path
+from typing import Dict, List, Optional, Set, Tuple
+
+import pandas as pd
+
+from .logger import logger
+
+# infoharbor section 前缀 → block_type
+_IH_TYPE = {'GN': '概念', 'ZS': '指数', 'FG': '风格'}
+
+
+def parse_infoharbor(path: Path) -> List[dict]:
+    """解析 infoharbor_block.dat
+
+    格式：``#GN_板块名,成员数,880码,创建日,更新日,,`` + 成员行 ``市场#6位code,...``
+    板块名按 GBK 8 字节截断（如"人形机器人"→"人形机器"），
+    调用方应用 880 码从 tdxzs/tdxzs3 还原全名。
+
+    Returns:
+        list[dict]: {'type', 'name', 'block_code'(可为 None), 'codes': set}
+    """
+    sections: List[dict] = []
+    cur: Optional[dict] = None
+    for line in path.read_text(encoding='gbk', errors='replace').splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        if line.startswith('#'):
+            head = line[1:].split(',')
+            if '_' not in head[0]:
+                cur = None
+                continue
+            typ, name = head[0].split('_', 1)
+            cur = {
+                'type': _IH_TYPE.get(typ, typ),
+                'name': name,
+                'block_code': head[2] if len(head) > 2 and head[2] else None,
+                'codes': set(),
+            }
+            sections.append(cur)
+        elif cur is not None:
+            for member in line.split(','):
+                if '#' in member:
+                    cur['codes'].add(member.split('#', 1)[1].strip())
+    return sections
+
+
+def parse_zs_cfg(path: Path, category: str) -> Dict[str, Tuple[str, str]]:
+    """解析 tdxzs.cfg / tdxzs3.cfg，按类别过滤
+
+    行格式：``板块名|880码|类别|级别|_|key``。
+    类别实测分布：2=TDX行业(T码key) 3=地区(数字key) 4=概念 5=风格 12=研究行业(X码key)
+
+    Returns:
+        dict: key -> (板块名, 板块代码)
+    """
+    out: Dict[str, Tuple[str, str]] = {}
+    for line in path.read_text(encoding='gbk', errors='replace').splitlines():
+        p = line.strip().split('|')
+        if len(p) >= 6 and p[2] == category and p[5]:
+            out[p[5]] = (p[0], p[1])
+    return out
+
+
+def parse_block_code_names(path: Path) -> Dict[str, str]:
+    """tdxzs/tdxzs3: 板块代码 -> 全名（用于还原 infoharbor 的 8 字节截断名）"""
+    out: Dict[str, str] = {}
+    for line in path.read_text(encoding='gbk', errors='replace').splitlines():
+        p = line.strip().split('|')
+        if len(p) >= 2 and p[1]:
+            out[p[1]] = p[0]
+    return out
+
+
+def parse_tdxhy(path: Path) -> Dict[str, str]:
+    """解析 tdxhy.cfg：6位code -> X行业码
+
+    行格式：``市场|code|T码|||X码``
+    """
+    out: Dict[str, str] = {}
+    for line in path.read_text(encoding='gbk', errors='replace').splitlines():
+        p = line.strip().split('|')
+        if len(p) >= 6 and p[1] and p[5]:
+            out[p[1]] = p[5]
+    return out
+
+
+def parse_spblock(path: Path) -> Dict[str, Set[str]]:
+    """解析 spblock.dat：``#板块名`` + 每行一个 ``市场前缀+6位code``"""
+    boards: Dict[str, Set[str]] = {}
+    cur: Optional[str] = None
+    for line in path.read_text(encoding='gbk', errors='replace').splitlines():
+        line = line.strip()
+        if line.startswith('#'):
+            cur = line[1:]
+            boards[cur] = set()
+        elif line and cur is not None and len(line) == 7:
+            boards[cur].add(line[1:])
+    return boards
+
+
+def parse_base_dbf_dy(path: Path) -> Dict[str, str]:
+    """解析 base.dbf，仅取 GPDM(股票代码)/DY(地域码) 两列
+
+    标准 DBF：头 32 字节（记录数@4、头长@8、记录长@10），
+    之后每 32 字节一个字段描述（名 11B + 类型 1B + 4B + 长度 1B），0x0D 结束。
+    """
+    data = path.read_bytes()
+    nrec = struct.unpack_from('<I', data, 4)[0]
+    hlen, rlen = struct.unpack_from('<HH', data, 8)
+
+    fields = []
+    offset = 1  # 每条记录首字节是删除标记
+    p = 32
+    while p < len(data) and data[p] != 0x0D:
+        name = data[p:p + 11].split(b'\x00')[0].decode('ascii', 'replace')
+        flen = data[p + 16]
+        fields.append((name, offset, flen))
+        offset += flen
+        p += 32
+    fmap = {n: (o, l) for n, o, l in fields}
+    if 'GPDM' not in fmap or 'DY' not in fmap:
+        raise ValueError(f"base.dbf 缺少 GPDM/DY 字段，实际字段: {sorted(fmap)[:12]}")
+
+    dm_o, dm_l = fmap['GPDM']
+    dy_o, dy_l = fmap['DY']
+    out: Dict[str, str] = {}
+    for i in range(nrec):
+        rec = data[hlen + i * rlen: hlen + (i + 1) * rlen]
+        if len(rec) < rlen or rec[0:1] == b'*':  # 删除标记
+            continue
+        code = rec[dm_o:dm_o + dm_l].decode('gbk', 'replace').strip()
+        dy = rec[dy_o:dy_o + dy_l].decode('gbk', 'replace').strip()
+        if code:
+            out[code] = dy
+    return out
+
+
+def collect_block_relations(tdx_path) -> pd.DataFrame:
+    """汇总四条数据链为板块-个股关系 DataFrame
+
+    单链文件缺失时 WARNING 跳过该类，不阻塞其他链。
+
+    Returns:
+        DataFrame: 列 block_type / block_code / block_name / code
+    """
+    hq = Path(tdx_path) / 'T0002' / 'hq_cache'
+    rows: List[Tuple[str, Optional[str], str, str]] = []
+
+    # 板块代码 -> 全名（还原 infoharbor 截断名）
+    fullname: Dict[str, str] = {}
+    for fname in ('tdxzs.cfg', 'tdxzs3.cfg'):
+        f = hq / fname
+        if f.exists():
+            fullname.update(parse_block_code_names(f))
+
+    # --- 概念/指数/风格（infoharbor）+ 指数补充（spblock） ---
+    ih_file = hq / 'infoharbor_block.dat'
+    sp_file = hq / 'spblock.dat'
+    sections = parse_infoharbor(ih_file) if ih_file.exists() else []
+    sp_boards = parse_spblock(sp_file) if sp_file.exists() else {}
+    if not ih_file.exists():
+        logger.warning(f"缺少 {ih_file}，跳过概念/指数/风格板块")
+    if not sp_file.exists():
+        logger.warning(f"缺少 {sp_file}，中证500 等跨市场指数成分将缺失")
+
+    ih_names = set()
+    for s in sections:
+        ih_names.add(s['name'])
+        # infoharbor 空板块（如中证500）优先用 spblock 同名成分补齐
+        codes = s['codes'] or sp_boards.get(s['name'], set())
+        if not codes:
+            continue
+        name = fullname.get(s['block_code'], s['name']) if s['block_code'] else s['name']
+        rows.extend((s['type'], s['block_code'], name, c) for c in codes)
+
+    # spblock 中未被 infoharbor 收录的板块（融资融券等）归为"特殊"
+    for name, codes in sp_boards.items():
+        if name not in ih_names:
+            rows.extend(('特殊', None, name, c) for c in codes)
+
+    # --- 行业（tdxhy X 码 × tdxzs3 类别 12，多级前缀匹配各入一行） ---
+    hy_file = hq / 'tdxhy.cfg'
+    zs3_file = hq / 'tdxzs3.cfg'
+    if hy_file.exists() and zs3_file.exists():
+        key2board = parse_zs_cfg(zs3_file, '12')
+        keys = sorted(key2board, key=len, reverse=True)
+        for code, x in parse_tdxhy(hy_file).items():
+            for k in keys:
+                if x.startswith(k):
+                    name, bcode = key2board[k]
+                    rows.append(('行业', bcode, name, code))
+    else:
+        logger.warning(f"缺少 {hy_file.name}/{zs3_file.name}，跳过行业板块")
+
+    # --- 地区（base.dbf DY × tdxzs 类别 3） ---
+    dbf_file = hq / 'base.dbf'
+    zs_file = hq / 'tdxzs.cfg'
+    if dbf_file.exists() and zs_file.exists():
+        dy2board = parse_zs_cfg(zs_file, '3')
+        for code, dy in parse_base_dbf_dy(dbf_file).items():
+            if dy in dy2board:
+                name, bcode = dy2board[dy]
+                rows.append(('地区', bcode, name, code))
+    else:
+        logger.warning(f"缺少 {dbf_file.name}/{zs_file.name}，跳过地区板块")
+
+    df = pd.DataFrame(rows, columns=['block_type', 'block_code', 'block_name', 'code'])
+    df = df.drop_duplicates(subset=['block_type', 'block_name', 'code'])
+    if not df.empty:
+        counts = df.groupby('block_type')['block_name'].nunique().to_dict()
+        logger.info(f"板块解析完成: {len(df)} 条关系，板块数 {counts}")
+    return df

tdx2db/cli.py

@@ -288,8 +288,13 @@ def parse_args() -> Namespace:
     min_parser.add_argument('--incremental', action='store_true', help='增量同步模式，跳过重复数据')
 
 
+    # 板块-个股关系同步
+    blocks_parser = subparsers.add_parser('blocks', help='同步板块-个股关系（行业/概念/指数/地区，来自通达信本地板块文件）')
+    blocks_parser.add_argument('--csv-only', action='store_true', help='仅保存到CSV')
+    blocks_parser.add_argument('--db-only', action='store_true', help='仅保存到数据库')
+
     # 一键同步（日线 + 分钟线增量同步到数据库）
-    subparsers.add_parser('sync', help='一键增量同步所有数据到数据库（日线 + 5/15/30/60分钟线）')
+    subparsers.add_parser('sync', help='一键增量同步所有数据到数据库（日线 + 5/15/30/60分钟线 + 板块关系）')
 
     # 数据库状态一览（只读，不需要 TDX_PATH）
     status_parser = subparsers.add_parser('status', help='数据库状态一览：每表行数/覆盖股票数/日期范围（只读）')
@@ -329,6 +334,23 @@ def update_config(args: Namespace) -> None:
     if args.no_tqdm:
         config.use_tqdm = False
 
+def sync_blocks(storage: DataStorage, to_csv: bool = False, to_db: bool = True) -> bool:
+    """同步板块-个股关系（全量替换式快照）
+
+    单链文件缺失在解析层降级为 WARNING；全部缺失（df 为空）不写库。
+    """
+    from .blocks import collect_block_relations
+
+    df = collect_block_relations(config.tdx_path)
+    if df.empty:
+        logger.warning("未解析到任何板块关系（板块文件缺失？），跳过写入")
+        return False
+    csv_path, db_success = storage.save_block_relation(df, to_csv=to_csv, to_db=to_db)
+    if to_db and db_success:
+        logger.info(f"板块关系已入库: {len(df)} 条")
+    return (not to_db) or db_success
+
+
 def _init_storage(create_tables: bool = True) -> Optional[DataStorage]:
     """初始化 DataStorage，失败时打印指引并返回 None"""
     try:
@@ -561,6 +583,18 @@ def main() -> int:
             logger.error(f"获取分钟线数据时出错: {e}")
             return 1
 
+    elif args.command == 'blocks':
+        try:
+            success = sync_blocks(
+                storage,
+                to_csv=not args.db_only,
+                to_db=not args.csv_only,
+            )
+            return 0 if success else 1
+        except Exception as e:
+            logger.error(f"同步板块关系时出错: {e}")
+            return 1
+
     elif args.command == 'sync':
         # 一键增量同步所有数据
         import time
@@ -592,6 +626,16 @@ def main() -> int:
             logger.error(f"同步分钟线数据时出错: {e}")
             has_error = True
 
+        # 3. 同步板块关系（秒级；文件缺失在解析层降级，不阻塞行情结果）
+        try:
+            logger.info("=== 同步板块关系 ===")
+            if not sync_blocks(storage):
+                logger.error("同步板块关系时出错")
+                has_error = True
+        except Exception as e:
+            logger.error(f"同步板块关系时出错: {e}")
+            has_error = True
+
         elapsed = time.monotonic() - t0
         if has_error:
             logger.warning(f"同步完成但有错误（耗时 {elapsed/60:.1f} 分钟），请回看上方 ERROR 日志")