refactor(data): migrate data layout to raw/processed/canonical (path-only)

data/README.md

@@ -1,18 +1,41 @@
 # data/ — 数据目录
 
+## 层（语义）
+
+```text
+raw
+  ↓ preprocessing
+processed
+  ↓ canonical resolution（script/canonical，冻结契约）
+canonical
+  ↓ SFT/RL export（script/verl/sft/export、script/verl/rl/export）
+parquet（data/sft、data/rl）
+```
+
+- `raw/`：source of truth 的原始导入物（CSV/XLSX / 原始包）。**训练脚本不得直接修改**。
+  真原始表格由数据维护方在 `data/raw/<dataset>.{csv,xlsx}` 提供（preprocessing
+  `prepare` 默认查找该路径）。`data/raw/` 为 gitignored，需要时由脚本/维护方创建。
+- `processed/<dataset>/`：预处理结果（`all.json` + `train.json` / `val.json` /
+  `test.json` + `split_report.json`）。split 按 record id 隔离，是后续 canonical 的
+  输入层，也是 `data/<dataset>` 归一化源层的正式位置。
+- `canonical/<dataset>/`：canonical 契约层（`all.json` + `resolution_report.json`）。
+  **后续所有算法（SFT/RL/评估）统一消费该层**；`data_level` 仅作 provenance。
+- `sft/`、`rl/`：可再生训练派生产物（`*.parquet` + export_report.json），gitignored；
+  源码变动后可删除并重新 export。
+- `knowledge/standards_map/`：标准知识（12 份词典，入库），不属于训练数据生成产物。
+
 ## 下载
 
 数据来自 ModelScope
 
 ```bash
 export MODELSCOPE_TOKEN=xxx     # token 走环境变量，禁止写入脚本/文档/仓库
-git clone <ModelScope 数据集地址> .
+git clone <ModelScope 数据集地址> .   # 导入到 raw/（原始 CSV/XLSX）
 ```
 
-
 ## 各领域概况（2025-08-10 快照）
 
-| 领域 | 业务域 | 记录数 | train/val/test |
+| 领域 | 业务域 | 记录数(processed all.json) | train/val/test |
 |---|---|---|---|
 | finance | 信托核心系统 | 568 | 439/69/58 |
 | infra | 钢铁基建（=shougang 子集） | 64 | 40/15/9 |
@@ -22,4 +45,104 @@ git clone <ModelScope 数据集地址> .
 ## 入库内容
 
 - `knowledge/standards_map/`：12 份多领域分类分级标准词典（公开标准知识，入库）
-- 本文档：下载说明
+- 本文档：下载说明 + 分层语义
+
+## 如何使用 `data/processed`（processed → canonical → parquet → 训练/评估）
+
+> 这是把 `data/processed` 交付给他人后、从零生成训练数据的标准步骤。
+> 只需要：仓库代码（含 `cfg/task/registry` + `cfg/task/corpus`，clone 自带）+ `data/processed`
+> + Python 3.10+ 且装有 pyarrow。**不需要** `raw` / `knowledge` / 已有 `canonical`/`sft`/`rl`。
+
+每步失败会清晰报错且不产生产物（fail-fast）；已产出时重跑需 `--overwrite`。
+
+**1) canonical（processed → `data/canonical/<ds>/all.json` + resolution_report.json）**
+
+```bash
+python -m script.canonical.targets --overwrite --datasets finance infra pers_info shougang
+# 或单数据集：--dataset pers_info
+```
+
+**2) SFT parquet（canonical → `data/sft/<ds>/{train,val,test}.parquet`）**
+
+```bash
+for ds in finance infra pers_info shougang; do
+  python -m script.verl.sft.export \
+    --canonical data/canonical/$ds/all.json \
+    --split-dir data/processed/$ds \
+    --output-dir data/sft/$ds \
+    --registry cfg/task/registry/$ds.registry.json \
+    --corpus cfg/task/corpus/$ds.corpus.json \
+    --metadata-fields field_name field_description
+done
+```
+
+**3) RL parquet（canonical → `data/rl/<ds>/{train,val,test}.parquet`，五字段）**
+
+```bash
+for ds in finance infra pers_info shougang; do
+  python -m script.verl.rl.export \
+    --canonical data/canonical/$ds/all.json \
+    --split-dir data/processed/$ds \
+    --output-dir data/rl/$ds \
+    --dataset $ds \
+    --registry cfg/task/registry/$ds.registry.json \
+    --corpus cfg/task/corpus/$ds.corpus.json \
+    --metadata-fields field_name field_description
+done
+```
+
+**4) 校验**（契约 + token 预算；token 预算需要模型 tokenizer）
+
+```bash
+python -m script.verl.sft.validate --dataset-dir data/sft/pers_info \
+  --registry cfg/task/registry/pers_info.registry.json --corpus cfg/task/corpus/pers_info.corpus.json \
+  --metadata-fields field_name field_description
+python -m script.verl.sft.check_token_budget --dataset-dir data/sft/pers_info \
+  --model <hf-model> --max-length 512
+```
+
+**5) 训练 / 评估**（需要 verl 环境 + GPU + 模型）
+
+```bash
+# SFT baseline（7B LoRA）
+DATASET=pers_info DATA_DIR=data/sft/pers_info MODEL_PATH=<hf-model> \
+  bash script/verl/sft/run_baseline.sh
+# RL smoke（GRPO）
+DATASET=pers_info TRAIN_FILE=data/rl/pers_info/train.parquet VAL_FILE=data/rl/pers_info/val.parquet \
+  MODEL_PATH=<hf-model> bash script/verl/rl/grpo_smoke.sh
+# 评估（choice protocol）
+python -m script.verl.sft.evaluate_baseline --model-path <merged> \
+  --data data/sft/pers_info/test.parquet --registry cfg/task/registry/pers_info.registry.json \
+  --report tmp/eval.json
+python -m script.verl.sft.evaluate_true_e2e --model-path <merged> \
+  --data data/sft/pers_info/test.parquet --registry cfg/task/registry/pers_info.registry.json \
+  --corpus cfg/task/corpus/pers_info.corpus.json --report tmp/eval_e2e.json
+```
+
+**预期产物 / 校验**（与迁移前仓库逐字节一致的基准行数）
+
+| dataset | canonical resolved | SFT/RL train | val | test |
+|---|---|---|---|---|
+| finance | 531 | 806 | 138 | 114 |
+| infra | 64 | 80 | 30 | 18 |
+| pers_info | 176 | 280 | 36 | 36 |
+| shougang | 18393 | 29042 | 3852 | 3892 |
+
+（shougang canonical 含 1022 条 `placeholder` 不入训；finance 有 34 missing_leaf / 3 path_mismatch 不入训。）
+生成 SFT/RL parquet 为**纯 pyarrow 计算**（无 torch/GPU）；训练/评估阶段才需要 GPU 与模型。
+新数据集上线另见 `docs/新数据集运行说明.md`（raw → processed 的完整流程）。
+
+## 迁移说明（2026-08，data layout refactor）
+
+- 原 `data/<dataset>/all.json + splits`（预处理归一化源层）→ `data/processed/<dataset>/`。
+- 原 `data/<dataset>/canonical/*` → `data/canonical/<dataset>/`。
+- 新增 `data/raw/`（原始物）、`data/legacy/`（历史遗留物）。
+- **注意：data 数据文件均为 gitignored，本 PR 不会通过 git 自动迁移现有本地数据。**
+  已有旧布局（`data/<dataset>/`）的环境需要手动将 `all.json`+splits 移至
+  `data/processed/<ds>/`、`canonical/*` 移至 `data/canonical/<ds>/`，
+  或按 `data/README.md` §「如何使用 data/processed」从 raw 重新生成。
+- 删除与 `all.json` 逐字节重复的 `data/shougang/all_shougang.json`。
+- `data/<dataset>/corpus.json`（legacy finance corpus，仅 analysis 对照用）→
+  `data/legacy/finance.corpus.json`；正式 corpus 唯一来源为 `cfg/task/corpus/`。
+- 本次迁移只改目录结构与路径引用，样本 / label / split / canonical resolution /
+  prompt / reward 均未改动。

docs/SFT_BASELINE.md

@@ -28,7 +28,7 @@ VERL_ENV_DIR=<persistent-disk>/envs/verl \
 ## Export
 
 Training labels come **exclusively** from the canonical dataset contract:
-records are read from `data/<dataset>/canonical/all.json`, only
+records are read from `data/canonical/<dataset>/all.json`, only
 `resolution_status == "resolved"` records with a target whose `category_id`
 belongs to the LeafRegistry enter training, and the ground truth is always
 `target.category_id`. `classification.level_1..level_4` stay provenance only;
@@ -44,8 +44,8 @@ resolves the five candidates by category_id against the canonical corpus.
 
 ```bash
 python -m script.verl.sft.export \
-  --canonical data/pers_info/canonical/all.json \
-  --split-dir data/pers_info \
+  --canonical data/canonical/pers_info/all.json \
+  --split-dir data/processed/pers_info \
   --output-dir data/sft/pers_info \
   --registry cfg/task/registry/pers_info.registry.json \
   --corpus cfg/task/corpus/pers_info.corpus.json \

docs/design/data_contract.md

@@ -15,7 +15,7 @@ truth for registry, corpus, ground truth, SFT labels, evaluation and reward.
   `docs/design/prompt_interface.md`); they never leak back into canonical
   records or reward semantics.
 
-## 2. Canonical sample schema (`data/<dataset>/canonical/all.json`)
+## 2. Canonical sample schema (`data/canonical/<dataset>/all.json`)
 
 Input sample keeps the original fields (`classification`, `metadata`,
 `data_level`, `label_status`, …) untouched as provenance. The resolver
@@ -72,7 +72,7 @@ appends:
 Only records with `resolution_status == "resolved"` **and** `target.category_id ∈
 LeafRegistry` **and** `target.leaf_name == registry.get(category_id).name`
 enter training; violations fail fast (no silent skip). Split membership follows
-the original `data/<dataset>/{train,val,test}.json` by record id. `canonical`
+the original `data/processed/<dataset>/{train,val,test}.json` by record id. `canonical`
 resolved / trainable counts (trainable = resolved within split boundaries):
 
 | dataset | canonical resolved | trainable | outside splits |

docs/phase_reports/stage3c_report.md

@@ -1,5 +1,11 @@
 # 阶段 3C：canonical data contract 接入 SFT 数据流水线
 
+> **2026-08 data layout migration note**：本文档正文中的路径为迁移前的旧布局
+> （`data/<ds>/all.json` + `data/<ds>/canonical/`）。新布局：
+> `data/processed/<ds>/all.json`（含 splits）→ `data/canonical/<ds>/all.json`
+> （canonical 契约）→ `data/sft/<ds>/`、`data/rl/<ds>/`（parquet）。
+> 内容/语义未变，仅目录结构。
+
 ## 1. 修改/新增文件
 
 修改：

docs/新数据集运行说明.md

@@ -1,5 +1,9 @@
 # TransClass 新数据集运行说明
 
+> 本文档覆盖 `raw → processed`（数据预处理/切分/语料）。
+> 拿到 `data/processed` 之后如何生成 `canonical → SFT/RL parquet → 训练/评估` 的标准步骤
+> 见 `data/README.md` §「如何使用 data/processed」。
+
 ## 一、放置原始数据
 
 将 CSV 或 XLSX 文件放到：

script/analysis/analyze_dataset_corpus_alignment.py

@@ -2,8 +2,8 @@
 string matching (no semantic models, no fuzzy auto-fixing of labels).
 
 Reads:
-  - datasets:     data/<dataset>/all.json (normalized TransClass JSON)
-  - leaf corpora: data/<dataset>/corpus.json (level_4 + description documents)
+  - datasets:     data/processed/<dataset>/all.json (normalized TransClass JSON)
+  - leaf corpora: data/legacy/*.corpus.json (legacy/archived dataset corpus)
   - standards:    data/knowledge/standards_map/*.json
 
 Writes:
@@ -25,6 +25,10 @@
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 DEFAULT_DATA_DIR = PROJECT_ROOT / "data"
+# post-migration layout: normalized records live under processed/
+DEFAULT_PROCESSED_DIR = DEFAULT_DATA_DIR / "processed"
+# legacy / archived dataset-level artifacts (never a formal corpus source)
+DEFAULT_LEGACY_DIR = DEFAULT_DATA_DIR / "legacy"
 DEFAULT_OUT_DIR = PROJECT_ROOT / "artifacts" / "generated" / "alignment"
 
 CLASSIFICATION_FIELDS = ("level_1", "level_2", "level_3", "level_4")
@@ -158,7 +162,7 @@ def _json_safe(value: Any) -> Any:
 
 
 def load_dataset(name: str) -> list[dict[str, Any]]:
-    path = DEFAULT_DATA_DIR / name / "all.json"
+    path = DEFAULT_PROCESSED_DIR / name / "all.json"
     with path.open("r", encoding="utf-8") as file:
         data = json.load(file)
     if not isinstance(data, list):
@@ -608,7 +612,7 @@ def match_dataset_to_corpus(
 
 def discover_datasets() -> dict[str, list[dict[str, Any]]]:
     datasets: dict[str, list[dict[str, Any]]] = {}
-    for child in sorted(DEFAULT_DATA_DIR.iterdir()):
+    for child in sorted(DEFAULT_PROCESSED_DIR.iterdir()):
         all_path = child / "all.json"
         if child.is_dir() and all_path.is_file():
             datasets[child.name] = load_dataset(child.name)
@@ -617,10 +621,16 @@ def discover_datasets() -> dict[str, list[dict[str, Any]]]:
 
 def discover_corpora() -> dict[str, tuple[Path, list[dict[str, Any]]]]:
     corpora: dict[str, tuple[Path, list[dict[str, Any]]]] = {}
-    for child in sorted(DEFAULT_DATA_DIR.iterdir()):
-        corpus_path = child / "corpus.json"
-        if child.is_dir() and corpus_path.is_file():
-            corpora[f"corpus:{child.name}"] = (corpus_path, _iter_entries(corpus_path))
+    # legacy dataset-level corpus (e.g. the archive-only finance corpus, now
+    # under data/legacy/); the formal corpus is cfg/task/corpus and is covered
+    # by the dataset/corpus alignment via canonical records, not here.
+    if DEFAULT_LEGACY_DIR.is_dir():
+        for path in sorted(DEFAULT_LEGACY_DIR.glob("*.corpus.json")):
+            # 文件形如 finance.corpus.json -> dataset "finance"（避免 Path.stem
+            # 产生 finance.corpus 双后缀，与下游 build_schema_issues 的硬编码
+            # "corpus:<dataset>" 保持一致）
+            dataset = path.name.removesuffix(".corpus.json")
+            corpora[f"corpus:{dataset}"] = (path, _iter_entries(path))
     standards_dir = DEFAULT_DATA_DIR / "knowledge" / "standards_map"
     for path in sorted(standards_dir.glob("*.json")):
         if path.name == "generate_standards_map.py":
@@ -853,7 +863,7 @@ def build_schema_issues(
                     "whitespace removal but different raw text"
                 ),
                 "detail": grouped_text + " — fix in the upstream input "
-                "(data/finance/all.json), never auto-repair.",
+                "(data/processed/finance/all.json), never auto-repair.",
             }
         )
     l3 = finance["level_stats"]["level_3"]
@@ -888,7 +898,7 @@ def build_schema_issues(
     issues.append(
         {
             "severity": "high",
-            "where": "corpus:finance (data/finance/corpus.json)",
+            "where": "corpus:finance (data/legacy/finance.corpus.json)",
             "issue": "leaf-only corpus: category identity is the bare level_4 name; "
             "no path/code is kept, so any future leaf-name collision is unresolvable",
             "detail": "all 220 unique level_4 labels; no path or code fields",
@@ -1089,7 +1099,7 @@ def render_markdown(report: dict[str, Any]) -> str:
     add("# 数据对齐分析报告 (dataset ↔ corpus alignment)")
     add("")
     add("> 方法约束：仅使用精确字符串与空白归一化匹配，不使用语义模型或模糊匹配修标签。")
-    add("> 数据来源：`data/<dataset>/all.json`、`data/<dataset>/corpus.json`、`data/knowledge/standards_map/*.json`。")
+    add("> 数据来源：`data/processed/<dataset>/all.json`、`data/legacy/*.corpus.json`、`data/knowledge/standards_map/*.json`。")
     add("")
 
     # 1. datasets
@@ -1234,7 +1244,7 @@ def render_markdown(report: dict[str, Any]) -> str:
     conclusions = report["conclusions"]
     add("### 4.1 实际有哪些 dataset")
     add("")
-    add(", ".join(f"`{name}`" for name in conclusions["datasets_found"]) + "（以存在 `all.json` 为准）")
+    add(", ".join(f"`{name}`" for name in conclusions["datasets_found"]) + "（以存在 processed/all.json 为准）")
     add("")
     add("### 4.2 每个 dataset 对应哪个 corpus")
     add("")

script/canonical/cli.py

@@ -12,7 +12,7 @@
 Sources:
 - shougang / infra: data/knowledge/standards_map/guanji_dict.json
 - finance:          data/knowledge/standards_map/financial_standards_dict.json
-- pers_info:        data/<dataset>/all.json (dataset universe, no standard exists)
+- pers_info:        data/processed/<dataset>/all.json (dataset universe, no standard exists)
 
 All builds and coverage diagnostics are computed before anything is written;
 every output file is written exactly once. Artifact sources are
@@ -60,9 +60,11 @@ def _load_json(path: Path) -> Any:
 
 
 def _load_records(data_dir: Path, dataset: str) -> list[dict[str, Any]]:
-    records = _load_json(data_dir / dataset / "all.json")
+    records = _load_json(data_dir / "processed" / dataset / "all.json")
     if not isinstance(records, list):
-        raise ValueError(f"{data_dir / dataset / 'all.json'} must be a JSON list")
+        raise ValueError(
+            f"{data_dir / 'processed' / dataset / 'all.json'} must be a JSON list"
+        )
     return records
 
 
@@ -91,7 +93,7 @@ def _annotate_coverage(
     data_dir: Path,
 ) -> None:
     """Read-only registry vs resolver-ID coverage; mutates the report only."""
-    records_path = data_dir / dataset / "all.json"
+    records_path = data_dir / "processed" / dataset / "all.json"
     if not records_path.is_file():
         report.dataset_id_coverage = {"available": False}
         return

script/canonical/targets.py

@@ -3,11 +3,12 @@
 Usage:
     python -m script.canonical.targets [--dataset finance] [--datasets ...] [--overwrite]
 
-Writes per dataset:
-    data/<dataset>/canonical/all.json
+Input:   data/processed/<dataset>/all.json (normalized records)
+Writes into data/canonical/<dataset>/:
+    all.json
         every input record unchanged (classification untouched) plus
         "resolution_status" and, for resolved records, "target".
-    data/<dataset>/canonical/resolution_report.json
+    resolution_report.json
         status counts, unresolved details, registry facts, input sha256.
 
 The LeafRegistry is the final constraint: a resolver target only counts as
@@ -36,7 +37,19 @@
 
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--data-dir", type=Path, default=PROJECT_ROOT / "data")
+    parser.add_argument(
+        "--processed-dir",
+        type=Path,
+        default=PROJECT_ROOT / "data" / "processed",
+        help="Where processed records live: processed-dir/<dataset>/all.json",
+    )
+    parser.add_argument(
+        "--canonical-dir",
+        type=Path,
+        default=PROJECT_ROOT / "data" / "canonical",
+        help="Where the canonical contract is written: canonical-dir/<dataset>/",
+    )
+
     parser.add_argument("--registry-dir", type=Path, default=PROJECT_ROOT / "cfg" / "task" / "registry")
     parser.add_argument("--corpus-dir", type=Path, default=PROJECT_ROOT / "cfg" / "task" / "corpus")
     parser.add_argument("--dataset", type=str, choices=list(DEFAULT_DATASETS))
@@ -59,10 +72,10 @@ def main(argv: list[str] | None = None) -> int:
     # 1. fail fast before building/writing anything
     if not args.overwrite:
         existing = [
-            Path(args.data_dir) / dataset / "canonical" / name
+            Path(args.canonical_dir) / dataset / name
             for dataset in datasets
             for name in ("all.json", "resolution_report.json")
-            if (Path(args.data_dir) / dataset / "canonical" / name).exists()
+            if (Path(args.canonical_dir) / dataset / name).exists()
         ]
         if existing:
             raise FileExistsError(
@@ -78,7 +91,8 @@ def main(argv: list[str] | None = None) -> int:
         prepared.append(
             prepare_canonical_dataset(
                 dataset,
-                data_dir=args.data_dir,
+                processed_dir=args.processed_dir,
+                canonical_dir=args.canonical_dir,
                 registry_dir=args.registry_dir,
                 corpus_dir=args.corpus_dir,
             )