diff --git a/.claude/skills/tdx2db-query/SKILL.md b/.claude/skills/tdx2db-query/SKILL.md index 0371592..55d85e0 100644 --- a/.claude/skills/tdx2db-query/SKILL.md +++ b/.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 diff --git a/AGENTS.md b/AGENTS.md index bd945c4..68ad416 100644 --- a/AGENTS.md +++ b/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）。 diff --git a/CLAUDE.md b/CLAUDE.md index bbb42e4..fc5db0b 100644 --- a/CLAUDE.md +++ b/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` 后检查数据库为准。 diff --git a/README.md b/README.md index 286d55b..a96e77a 100644 --- a/README.md +++ b/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。 diff --git a/scripts/migrate_block_relation.sql b/scripts/migrate_block_relation.sql new file mode 100644 index 0000000..187c69c --- /dev/null +++ b/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）会自动按新结构建表。 diff --git a/tdx2db/blocks.py b/tdx2db/blocks.py new file mode 100644 index 0000000..dad2476 --- /dev/null +++ b/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(' 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 diff --git a/tdx2db/cli.py b/tdx2db/cli.py index b179d14..eb96a4f 100644 --- a/tdx2db/cli.py +++ b/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 日志") diff --git a/tdx2db/storage.py b/tdx2db/storage.py index d2852fb..e412f8d 100644 --- a/tdx2db/storage.py +++ b/tdx2db/storage.py @@ -22,17 +22,20 @@ Base = declarative_base() class BlockStockRelation(Base): - """板块股票关系表模型""" + """板块股票关系表模型（issue #39：全量替换式快照，无增量语义） + + 唯一约束用 (block_type, block_name, code)：block_code 可空 + （沪深300 等指数板块无 880 码），且同名板块跨体系存在 + （880302 与 881002 都叫"煤炭开采"，type 不同）。 + """ __tablename__ = 'block_stock_relation' - # 唯一约束是 save_incremental 增量写入（ON CONFLICT / INSERT IGNORE）的前提， - # 缺失时 PG 会报错、MySQL/SQLite 会静默重复累积（issue #16） - __table_args__ = (UniqueConstraint('block_code', 'code', name='uq_block_code'),) + __table_args__ = (UniqueConstraint('block_type', 'block_name', 'code', name='uq_block_name_code'),) id = Column(Integer, primary_key=True) - block_code = Column(String(20), index=True) # 板块代码 - block_name = Column(String(50)) # 板块名称 - code = Column(String(10), index=True) # 股票代码 - name = Column(String(50)) # 股票名称 + block_type = Column(String(10), index=True) # 行业/概念/指数/地区/风格/特殊 + block_code = Column(String(20), index=True, nullable=True) # 880/881 板块代码，可空 + block_name = Column(String(50), index=True) # 板块名称 + code = Column(String(10), index=True) # 股票代码（6 位纯数字，与行情表口径一致） class DailyData(Base): """日线数据表模型""" @@ -191,7 +194,7 @@ class StockInfo(Base): 'stock_info', 'block_stock_relation', }) -# status 命令统计的表及其日期列（stock_info 无日期列） +# status 命令统计的表及其日期列（stock_info / block_stock_relation 无日期列） _STATS_TABLES = ( ('stock_info', None), ('daily_data', 'date'), @@ -199,6 +202,7 @@ class StockInfo(Base): ('minute15_data', 'datetime'), ('minute30_data', 'datetime'), ('minute60_data', 'datetime'), + ('block_stock_relation', None), ) @@ -641,13 +645,16 @@ def save_block_relation( to_db: bool = True, batch_size: int = 10000 ) -> Tuple[Optional[str], bool]: - """保存板块与股票的对应关系 + """保存板块与股票的对应关系（全量替换式快照） + + 板块关系没有历史概念（成分调整由通达信盘后文件直接体现）， + 每次同步以 DELETE + INSERT 反映当前快照——与行情表的增量语义不同。 Args: - df: 板块与股票对应关系DataFrame + df: 列 block_type/block_code/block_name/code to_csv: 是否保存到CSV to_db: 是否保存到数据库 - batch_size: 批处理大小，默认10000条记录 + batch_size: 批处理大小 Returns: tuple: (csv_path, db_success) @@ -659,6 +666,18 @@ def save_block_relation( csv_path = self.save_to_csv(df, 'block_stock_relation') if to_db: - db_success = self.save_to_database(df, 'block_stock_relation', batch_size=batch_size) + # DELETE + INSERT 必须同一事务：任一批次失败整体回滚， + # 保住旧快照——否则失败会留下空表/半新半旧快照（PR #40 review） + try: + with self.engine.begin() as conn: + conn.execute(text("DELETE FROM block_stock_relation")) + df.to_sql( + 'block_stock_relation', conn, + if_exists='append', index=False, chunksize=batch_size, + ) + db_success = True + except Exception as e: + logger.error(f"板块关系写入失败，已整体回滚、保留旧快照: {e}") + db_success = False return csv_path, db_success diff --git a/tests/test_blocks.py b/tests/test_blocks.py new file mode 100644 index 0000000..37ddf94 --- /dev/null +++ b/tests/test_blocks.py @@ -0,0 +1,197 @@ +"""板块解析与入库测试（issue #39） + +合成 fixture 覆盖四条数据链：infoharbor（概念/指数/风格，含 8 字节截断名、 +空板块、spblock 补齐）、tdxhy×tdxzs3（行业多级）、base.dbf×tdxzs（地区）。 +格式样本与真实文件逐字段一致（2026-07-07 实测验证）。 +""" +import struct + +import pandas as pd +import pytest +from sqlalchemy import text + +from tdx2db.blocks import ( + collect_block_relations, + parse_base_dbf_dy, + parse_infoharbor, + parse_spblock, + parse_tdxhy, + parse_zs_cfg, +) +from tdx2db.config import config +from tdx2db.storage import DataStorage + + +def _write_gbk(path, content): + path.write_bytes(content.encode('gbk')) + + +def _make_dbf(path, rows): + """最小 DBF：字段 SC(2)/GPDM(6)/DY(4)，与真实 base.dbf 同构""" + fields = [('SC', 2), ('GPDM', 6), ('DY', 4)] + hlen = 32 + 32 * len(fields) + 1 + rlen = 1 + sum(length for _, length in fields) + buf = bytearray() + buf += struct.pack('