feat: block_stock_relation 增加 block_level（行业层级标注）

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

@@ -55,6 +55,10 @@ ORDER BY datetime;
 SELECT code FROM block_stock_relation
 WHERE block_type = '行业' AND block_name = '煤炭开采';
 
+-- 全部二级行业（≈ 通达信导出 CSV 的行业口径，板块强弱迭代用这个）
+SELECT DISTINCT block_name FROM block_stock_relation
+WHERE block_type = '行业' AND block_level = 2;
+
 -- 个股 → 全部板块归属
 SELECT block_type, block_code, block_name FROM block_stock_relation WHERE code = :code6;
 

AGENTS.md

@@ -53,10 +53,13 @@ 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 定位）
+-- 4. 按板块取成分（行业含一/二/三级；block_level=2 ≈ 通达信导出 CSV 的行业口径）
 SELECT code FROM block_stock_relation
 WHERE block_type = '行业' AND block_name = '煤炭开采';
 
+SELECT DISTINCT block_name FROM block_stock_relation
+WHERE block_type = '行业' AND block_level = 2;   -- 全部二级行业（板块强弱迭代口径）
+
 -- 5. 按个股查板块归属
 SELECT block_type, block_code, block_name FROM block_stock_relation
 WHERE code = '000001';

tdx2db/blocks.py

@@ -159,7 +159,8 @@ def collect_block_relations(tdx_path) -> pd.DataFrame:
         DataFrame: 列 block_type / block_code / block_name / code
     """
     hq = Path(tdx_path) / 'T0002' / 'hq_cache'
-    rows: List[Tuple[str, Optional[str], str, str]] = []
+    # (block_type, block_code, block_name, block_level, code)
+    rows: List[Tuple[str, Optional[str], str, Optional[int], str]] = []
 
     # 板块代码 -> 全名（还原 infoharbor 截断名）
     fullname: Dict[str, str] = {}
@@ -186,24 +187,27 @@ def collect_block_relations(tdx_path) -> pd.DataFrame:
         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)
+        rows.extend((s['type'], s['block_code'], name, None, 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)
+            rows.extend(('特殊', None, name, None, 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)
+        # 层级 = 自身及前缀 key 的数量（X10→1，X1001→2，X100101→3），
+        # 不依赖 key 长度约定，对分类体系调整鲁棒
+        key_level = {k: sum(1 for o in key2board if k.startswith(o)) for k in key2board}
         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))
+                    rows.append(('行业', bcode, name, key_level[k], code))
     else:
         logger.warning(f"缺少 {hy_file.name}/{zs3_file.name}，跳过行业板块")
 
@@ -215,11 +219,11 @@ def collect_block_relations(tdx_path) -> pd.DataFrame:
         for code, dy in parse_base_dbf_dy(dbf_file).items():
             if dy in dy2board:
                 name, bcode = dy2board[dy]
-                rows.append(('地区', bcode, name, code))
+                rows.append(('地区', bcode, name, None, code))
     else:
         logger.warning(f"缺少 {dbf_file.name}/{zs_file.name}，跳过地区板块")
 
-    df = pd.DataFrame(rows, columns=['block_type', 'block_code', 'block_name', 'code'])
+    df = pd.DataFrame(rows, columns=['block_type', 'block_code', 'block_name', 'block_level', '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()

tdx2db/storage.py

@@ -35,6 +35,7 @@ class BlockStockRelation(Base):
     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)  # 板块名称
+    block_level = Column(Integer, nullable=True)  # 行业层级 1/2/3（二级≈通达信导出 CSV 口径），其他类型为 NULL
     code = Column(String(10), index=True)  # 股票代码（6 位纯数字，与行情表口径一致）
 
 class DailyData(Base):

tests/test_blocks.py

@@ -138,10 +138,14 @@ def test_all_chains(self, hq_cache):
         sp = df[(df.block_type == '特殊') & (df.block_name == '融资融券')]
         assert set(sp.code) == {'000001'}
 
-        # 行业：X100101 命中三级各一行
+        # 行业：X100101 命中三级各一行，block_level 标注层级
         hy = df[(df.block_type == '行业') & (df.code == '000552')]
         assert set(hy.block_name) == {'煤炭', '煤炭开采', '动力煤'}
         assert set(hy.block_code) == {'881001', '881002', '881003'}
+        assert dict(zip(hy.block_name, hy.block_level)) == {'煤炭': 1, '煤炭开采': 2, '动力煤': 3}
+
+        # 非行业类型 block_level 为空
+        assert df[df.block_type != '行业'].block_level.isna().all()
         # 无匹配 key 的 X9999 不产生行
         assert df[(df.block_type == '行业') & (df.code == '600000')].empty