Day 14 已經鎖定一筆資料能否進入系統的邊界：模型只能讀取初次檢傷當下的 14 個候選欄位，護理師級數、專家標籤與檢傷後結果都被隔離。通過這道檢查後，我們仍然不能立刻訓練模型，因為「欄位可以使用」不代表「每一列都有值」，也不代表五個急迫級都有足夠樣本。

以呼吸困難案例為例。假設主訴、呼吸速率與心率都有紀錄，血氧飽和度卻是空白。空白只告訴我們「目前沒有讀到數值」，不能推論病患血氧正常，更不能擅自填成 0。0% 的血氧不是「沒量到」，而是另一個完全不同、甚至極端異常的數值語意。

疼痛分數又更複雜。同樣是空白，若 Pain = 0，可能表示病患沒有疼痛，因此疼痛數字評分不適用；若 Pain = 1，空白才可能代表應該詢問但沒有記錄。只用一個平均值填滿這兩種空白，就會把不同原因混成同一件事。

下圖是概念示意。閱讀時請注意兩個動作：左側資料卡的空白保持未知，沒有被自動改成正常值；右側不同急迫級的資料，則被分配到多個驗證折，而不是任由少數類別集中在某一折。

上圖把今天的兩個核心原則放在一起：先如實保留資料不知道的部分，再建立每次都能重現的公平切分。本篇沿用韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale, KTAS）公開資料，但不會訓練大型語言模型（Large Language Model, LLM），也不需要啟動 Ollama 或下載模型；我們要產生的是缺失概況、五級標籤分布與固定的交叉驗證分派表。

今天要完成什麼？

讀完並跟著操作後，你會完成以下五件事：

1. 用實際 1,267 筆資料計算七個數值欄位的缺失率，並拆開疼痛分數缺失的兩種語意。
2. 區分缺失值、結構上不合法的數值，以及罕見但仍可能成立的極端值。
3. 看懂五級標籤不平衡如何製造「整體準確率看似不差」的錯覺。
4. 建立五次重複、每次五折，而且同時維持急迫級與場域比例的紀錄層級切分。
5. 把缺失處理、類別權重與折外預測的順序寫成可執行契約，避免驗證資料洩漏回訓練流程。

本篇會用到的名詞

本篇會使用多個資料科學名詞，先用表格建立快速索引。後面的正文仍會以實際 KTAS 資料逐一解釋用途。

先盤點缺失，不急著補值

缺失值（Missing Value）是資料表中沒有可用觀測值的位置。它只描述「值不存在於目前資料」，不自動說明原因。原因可能是沒有測量、量測失敗、未記錄、轉檔遺失，或該欄位根本不適用。

本篇讀取 Day 12 產生的 data/interim/ktas-v1/model-input-candidates.csv。data/ 是保存本機資料的資料夾，interim/ 子目錄保存已完成編碼、欄位名稱與角色正規化，但還沒有做模型補值的中介資料。這個逗號分隔值（Comma-Separated Values, CSV）檔共有 1,267 列；除了負責對齊的 record_index，還有 Day 14 允許的 14 個候選輸入欄位。

本篇集中檢查七個可能缺失的數值欄位：

• 疼痛數字評定量表（Numeric Rating Scale, NRS）的 NRS_pain。
• 收縮壓（Systolic Blood Pressure, SBP）。
• 舒張壓（Diastolic Blood Pressure, DBP）。
• 心率（Heart Rate, HR）。
• 呼吸速率（Respiratory Rate, RR）。
• 體溫（Body Temperature, BT）。
• 來源欄位 Saturation 所表示的血氧飽和度。

缺失率的計算方式是：

[
\text{欄位缺失率}=\frac{\text{該欄位缺失筆數}}{\text{全部紀錄筆數}}
]

分母是 1,267 筆符合資料契約的紀錄；分子則逐欄計算空值數量。例如 Saturation 缺失 697 筆，所以：

[
\frac{697}{1267}\times 100% \approx 55.0%
]

數值越大，表示該欄位在目前公開資料中越常沒有觀測值。它不能直接證明臨床現場也有相同比例，更不能告訴我們為什麼缺失。

實際盤點結果如下：

下圖把七個欄位放在同一個尺度上，右側再拆開 NRS_pain 缺失的語意。這些數字直接由本篇公開聚合結果產生，不是手動抄進圖片。

上圖顯示，1,267 筆中有 706 筆至少缺一項生命徵象，16 筆六項生命徵象全部缺失。這不是可以略過的小問題。若只保留生命徵象完整的紀錄，資料量、病患組成與急迫級比例都可能改變，因此完整案例刪除不會成為預設策略。

臨床預測模型透明報告指引加人工智慧擴充版（Transparent Reporting of a Multivariable Prediction Model for Individual Prognosis or Diagnosis Plus Artificial Intelligence, TRIPOD+AI）也要求清楚報告每個預測變數的缺失數量、處理方法與假設，而不是只寫一句「已清理資料」。TRIPOD+AI 擴充檢查表

同樣是空白，疼痛分數有兩種意思

不適用（Not Applicable）表示某個欄位在目前條件下本來就不該有值。公開資料同時提供 Pain 與 NRS_pain：前者記錄是否有疼痛，後者才是 1 到 10 的疼痛強度。

在 556 筆 NRS_pain 缺失紀錄中：

• 553 筆同時是 Pain = 0，本系列表示為「不適用」。
• 3 筆同時是 Pain = 1，本系列表示為「未知／未記錄」。
• 沒有出現 Pain = 0 卻填入疼痛分數的紀錄。

因此，後續建立檢索增強生成（Retrieval-Augmented Generation, RAG）輸入文字時，兩種情況要使用不同語句：

Pain = 0 且 NRS_pain 缺失 → 疼痛：無；疼痛分數：不適用
Pain = 1 且 NRS_pain 缺失 → 疼痛：有；疼痛分數：未記錄

RAG 是先檢索外部知識，再讓生成模型根據檢索內容回答的方法。對 RAG 而言，「未記錄」與「不適用」本身就是語意，不需要先捏造一個數值才能寫進文字。

若後續的傳統機器學習（Machine Learning, ML）基準模型只接受數值，才考慮在每個訓練折內建立補值器，並搭配缺失指示欄（Missing Indicator）。缺失指示欄通常用 1 表示原值缺失、0 表示原值存在；它不是補出的生命徵象，而是保存資料取得狀態。

缺失、非法值與極端值不能混為一談

清理資料時常見的錯誤，是看到罕見數值就直接刪除。可是「少見」不等於「錯誤」，尤其在急診資料中，極端值可能正是危急病患的重要訊號。

本篇把數值分成三類：

1. 缺失值：沒有數值，保留缺失狀態。
2. 結構上不合法的值：明確違反欄位基本限制，例如血氧小於 0、疼痛分數大於 10，直接讓資料品質檢查失敗。
3. 結構合法但罕見的值：仍在允許範圍內，只標記供人工審查與敏感度分析，不自動刪除、截尾或改寫。

實際觀測範圍如下：

這張表只能證明數值通過目前寫下的結構規則。例如收縮壓 275 或血氧 20 都很極端，但公開檔案沒有提供足夠證據判定是量測錯誤、輸入錯誤或真實臨床狀況。因此本篇不會僅憑「看起來不尋常」就刪除它們。

五級標籤其實非常不平均

類別不平衡（Class Imbalance）是不同標籤的樣本數量差距明顯。本系列的主要參考標籤是專家重新判定的 KTAS_expert；KTAS 以第一級表示最緊急、第五級表示最不緊急。

實際分布如下：

第一、二級合計 246 筆，占 19.4%；最大類別第三級有 487 筆，與只有 26 筆的第一級相差約 18.7 倍。

下圖左側呈現五級實際分布，右側則是一個完全不讀取病患資料、永遠猜第三級的基準。閱讀時請比較整體準確率與平衡準確率，而不是只看其中一個數字。

上圖中的「全猜第三級」可以得到 38.4% 整體準確率（Accuracy），也就是 1,267 筆中預測正確的比例，因為資料本來就有 38.4% 是第三級。然而這個方法對第一、二、四、五級的召回率（Recall）全部是 0%，只有第三級召回率是 100%。某一級的召回率，是該級真實紀錄中被模型正確找回的比例。

平衡準確率（Balanced Accuracy）會先計算每一級召回率，再對五級取平均：

[
\text{平衡準確率}=\frac{R_1+R_2+R_3+R_4+R_5}{5}
]

其中 (R_i) 代表第 (i) 級的召回率。全猜第三級時：

[
\frac{0+0+1+0+0}{5}=0.2=20%
]

這就是為什麼 Day 7 已把各級召回率、平衡準確率、巨集 F1 分數（Macro F1-score）與檢傷不足率列為主要指標，整體準確率只作次要參考。平衡準確率的用途正是避免不平衡資料讓多數類別支配評估。Brodersen 等人：The Balanced Accuracy and Its Posterior Distribution

類別不平衡也不能靠切分前複製第一級資料解決。若同一筆複製資料同時落入訓練折與驗證折，模型等於看過驗證答案。任何重新抽樣或類別權重，只能在每一個訓練折內根據該折資料建立，驗證折保持原始分布。

為什麼不能做病患層級切分？

理想情況下，同一位病患的多次急診就醫應該全部留在同一側，避免模型從訓練資料記住個人特徵，再到驗證資料遇到同一人。這稱為病患層級分組切分。

然而，公開 Kaggle KTAS 資料沒有可用病患識別碼。record_index 只是本系列為 1,267 列依序建立的對齊鍵，不能辨認兩列是否來自同一位病患。因此本篇只能建立「急診就醫紀錄層級」切分，不能宣稱病患層級切分已完成。

這個限制有三個後果：

1. 同一病患若重複就醫，紀錄可能落入不同折。
2. 內部交叉驗證分數可能因未知相依性而過度樂觀。
3. 後續重抽樣也只能以 record_index 為單位，不能假裝是病患層級信賴區間。

公開資料也沒有能支援嚴格時間切分的逐筆日期，所以本篇不會把 1,267 筆切成早期訓練與晚期測試。資料來自兩所醫院，但 Group 只有場域代碼；本篇會用它維持各折場域比例，卻不把同一份資料內的場域分層稱為跨醫院外部驗證。

重複分層五折如何運作？

分層 K 折交叉驗證（Stratified K-Fold Cross-Validation）先把資料分成 K 個互斥部分，再輪流用其中一折驗證、其餘 K−1 折訓練。分層的目的，是讓每一折盡量保留原始標籤比例，避免只有 26 筆的第一級偶然集中在少數折。

本篇設定 K = 5。每次切分都會進行五輪：

1. 第 1 折當驗證資料，第 2 到第 5 折當訓練資料。
2. 第 2 折當驗證資料，其餘四折當訓練資料。
3. 依序輪替，直到五折都各自當過一次驗證資料。

因此，每筆紀錄在一次五折中會參與四輪訓練，並且恰好在一輪作為驗證資料。它在沒有參與該輪訓練時得到的結果，就是折外預測（Out-of-Fold Prediction, OOF Prediction）。

只做一次五折，結果仍可能依賴某一次隨機分配。重複分層 K 折交叉驗證（Repeated Stratified K-Fold Cross-Validation）會使用不同亂數種子重新切分。本篇固定五個亂數種子 42、43、44、45、46，因此共有：

[
5\text{ 次重複}\times 5\text{ 折}=25\text{ 組訓練／驗證}
]

固定亂數種子是讓偽隨機程序在相同輸入下重現相同結果的起始值，不代表資料沒有隨機化。scikit-learn 官方文件也把重複分層 K 折定義為使用不同隨機化重複類別分層 K 折，並建議以固定整數 random_state 重現切分。scikit-learn：RepeatedStratifiedKFold

本篇比只看標籤再多一層：先把 KTAS_expert 與兩所醫院的 Group 組成十個分層，再把每個分層中的紀錄輪流分到五折。Group 只在建立切分時使用，不會寫入模型輸入或分派檔。

下圖左側是第一次重複的實際五個驗證折。每折共有 251 到 255 筆，第三、第四級占多數，但每一級都存在。右側則顯示五次重複如何讓每筆紀錄累積五個折外預測。

上圖中第一次重複的第一級筆數依序為 4、6、5、6、5；五折合計仍是原始的 26 筆。程式也檢查每一個「級數 × 場域」分層在五折之間最多只相差一筆。這能減少切分造成的比例波動，但不能增加第一級資料量，也不能解決病患識別碼缺失的問題。官方文件同樣提醒，分層主要是工程上的切分工具，不是消除小樣本統計不確定性的方案。scikit-learn：Cross-validation

切分一定要早於補值與類別權重

資料洩漏不只發生在欄位中，也可能發生在操作順序。假設先用全部 1,267 筆計算血壓中位數，再做五折切分，每個驗證折的分布已經參與補值器，評估就不再是完全未見資料。

正確順序是：

1. 先只使用 record_index、KTAS_expert 與 Group 建立固定分派表。
2. 依某次重複與某一折，把紀錄分成 training fold 與 validation fold。
3. 只用 training fold 學習補值統計量、缺失指示欄規則、尺度轉換或類別權重。
4. 用已在 training fold 學到的規則轉換 validation fold，不重新計算。
5. 模型只用 training fold 訓練，再對 validation fold 產生折外預測。
6. 五折全部完成後，才能組回該次重複的 1,267 筆折外預測。

training fold 是當輪可供模型學習的四折；validation fold 是完全不參與該輪學習、只用來產生折外預測的一折。兩者在每輪會交換角色，但任何一輪都不能讓 validation fold 反向影響前處理。

scikit-learn 的資料洩漏指南也要求先切分，再讓 fit 或 fit_transform 只接觸訓練資料；驗證資料只能呼叫已學好規則的 transform。scikit-learn：Common pitfalls and recommended practices 臨床預測模型的缺失資料研究也強調，部署或驗證時的缺失處理必須只依賴開發資料，才能套用到下一位未知結果的個案。Sisk 等人：Handling missing data when developing and validating clinical prediction models

把決策寫進設定，而不是只留在文章裡

本篇接下來會建立 configs/data/day-15-quality-and-split-contract.json。configs/ 是保存可審查參數與規則的資料夾，data/ 子目錄集中資料取得、欄位角色與切分契約；.json 表示 JavaScript 物件表示法（JavaScript Object Notation, JSON）檔案。

這份契約讀取三份 Day 12 中介資料：

• model-input-candidates.csv：14 個候選輸入與 record_index，用來盤點缺失與結構值域。
• audit-strata.csv：場域代碼 Group，只用來稽核與分層。
• reference-labels.csv：主要標籤 KTAS_expert，只用來建立分層與後續評估。

契約固定以下決策：

• 缺失不能被假設為正常。
• NRS_pain 必須依 Pain 拆成「不適用」與「未知／未記錄」。
• 不因數值極端就自動刪除。
• 切分前禁止重新抽樣，validation fold 也禁止重新抽樣。
• 補值器、缺失指示欄與類別權重只能在 training fold 內學習。
• 五折、五次重複與五個亂數種子固定。
• 分派表只能包含 record_index、repeat_id、fold_id 與 seed，不能夾帶標籤或場域。

以下是實際設定的核心節錄。為了聚焦操作順序，省略了完整欄位清單、輸入路徑與輸出政策，所以不能把這段單獨存檔後直接執行。

{
  "missingness_policy": {
    "rag_representation": "保留缺失狀態，明確寫成未記錄或不適用，不得轉寫為正常。",
    "reference_label_must_not_be_used_for_imputation": true,
    "complete_case_deletion_is_not_default": true
  },
  "split_policy": {
    "method": "repeated_stratified_five_fold",
    "fold_count": 5,
    "repeat_count": 5,
    "repeat_seeds": [42, 43, 44, 45, 46],
    "stratification_fields": ["KTAS_expert", "Group"],
    "preprocessing_fit_scope": "training_fold_only"
  }
}

設定檔不會自行處理資料，它的功能是讓程式、測試與文章共用同一份規則。若未來要更換折數或種子，必須更新契約版本、重新產生分派表並重跑所有方法，不能只修改某個模型的訓練程式。

核心程式、測試與執行入口如何分工？

可重複使用的資料品質與切分邏輯位於 src/triage_rag/data/quality.py。src/ 是存放專案核心 Python 套件的資料夾，triage_rag/data/ 子目錄專門放資料驗證與轉換邏輯。這支程式接收三個 Pandas 資料表與 Day 15 契約，產生聚合報告及分派表，不負責命令列參數或執行目錄。

核心程式依序完成：

1. 檢查三份資料的 record_index 唯一、非空且順序相同。
2. 計算七欄缺失率、疼痛分數缺失語意與結構值域。
3. 計算五級標籤分布、危急級比例及多數類別基準。
4. 依「KTAS_expert × Group」與固定種子建立五次五折分派。
5. 驗證每筆紀錄在每次重複只出現一次、所有折都存在，而且同分層的折間筆數最多差一。

以下片段是實際建立單一分層分派的核心。輸入是某個級數與場域內排序後的 record_index；固定種子打亂順序後，程式輪流放入五折。輸出只寫索引、重複編號、折編號與種子。

generator.shuffle(record_indices)
fold_order = list(range(1, fold_count + 1))
generator.shuffle(fold_order)

for position, record_index in enumerate(record_indices):
    rows.append(
        {
            "record_index": int(record_index),
            "repeat_id": repeat_id,
            "fold_id": fold_order[position % fold_count],
            "seed": seed,
        }
    )

自動化測試位於 tests/test_quality_and_splits.py。tests/ 是保存不接觸真實病患資料之合成測試的資料夾。本篇新增五項測試：相同契約必須產生完全相同分派、疼痛分數缺失必須拆成兩種語意、重複 record_index 必須失敗、超出結構範圍的血氧必須失敗，以及任何級數與場域分層少於五筆時必須失敗。

讀者真正從終端機啟動的是 scripts/prepare_day15_splits.py。scripts/ 保存可執行的自動化入口；這支 .py 程式讀取契約與三份中介資料，呼叫核心模組，再把結果分別寫到本機資料、公開聚合摘要與單次執行紀錄。它不會下載資料；若 Day 12 產物尚未建立，要先執行資料管線。

先在自己的專案資料夾建立本篇完整檔案

接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾，依下列順序建立檔案；每個程式碼區塊都是該檔案的完整內容，不含省略號。

本篇沿用 Day 12 的三份處理後 CSV、Day 13 的重現性工具與 Day 14 的 record_index 對齊契約；以下是 Day 15 新增的完整檔案。

先從專案根目錄建立需要的資料夾：

mkdir -p configs/data scripts src/triage_rag/data tests

如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案，把對應區塊完整貼入後儲存。

檔案 1：建立 configs/data/day-15-quality-and-split-contract.json

固定缺失、極端值、類別不平衡、五次五折與不確定性處理政策。

請在文字編輯器建立 configs/data/day-15-quality-and-split-contract.json，貼入以下完整內容並儲存：

{
  "schema_version": 1,
  "contract_id": "day-15-ktas-quality-and-record-level-splits-v1",
  "status": "implemented-and-source-validated",
  "source_dataset_card": "configs/data/day-12-kaggle-ktas-dataset-card.json",
  "source_data_contract": "configs/data/day-14-ktas-data-contract.json",
  "inputs": {
    "model_inputs": "data/interim/ktas-v1/model-input-candidates.csv",
    "audit_strata": "data/interim/ktas-v1/audit-strata.csv",
    "reference_labels": "data/interim/ktas-v1/reference-labels.csv"
  },
  "row_link_key": "record_index",
  "missingness_policy": {
    "profile_fields": [
      "NRS_pain",
      "SBP",
      "DBP",
      "HR",
      "RR",
      "BT",
      "Saturation"
    ],
    "vital_sign_fields": [
      "SBP",
      "DBP",
      "HR",
      "RR",
      "BT",
      "Saturation"
    ],
    "nrs_pain_rule": {
      "applicability_field": "Pain",
      "not_applicable_value": 0,
      "expected_observed_value": 1,
      "missing_when_not_applicable_representation": "not_applicable",
      "missing_when_expected_representation": "unknown_not_recorded",
      "never_fill_missing_with_zero_by_default": true
    },
    "rag_representation": "保留缺失狀態，明確寫成未記錄或不適用，不得轉寫為正常。",
    "traditional_ml_policy": "若基準模型需要補值，補值器與缺失指示欄位只能在每個 training fold 內 fit，再套用到該 fold 的 validation 資料。",
    "reference_label_must_not_be_used_for_imputation": true,
    "complete_case_deletion_is_not_default": true
  },
  "structural_value_policy": {
    "automatic_extreme_value_deletion": false,
    "strictly_positive_fields": [
      "SBP",
      "DBP",
      "HR",
      "RR",
      "BT",
      "Saturation"
    ],
    "bounded_fields": {
      "NRS_pain": {
        "minimum": 0,
        "maximum": 10
      },
      "Saturation": {
        "minimum": 0,
        "maximum": 100
      }
    },
    "extreme_but_structurally_valid_values": "只列入人工審查與敏感度分析；在缺少單位錯誤或量測錯誤證據時，不因罕見就刪除、截尾或改寫。"
  },
  "class_imbalance_policy": {
    "label_field": "KTAS_expert",
    "label_order": [1, 2, 3, 4, 5],
    "resampling_before_split_forbidden": true,
    "validation_fold_resampling_forbidden": true,
    "training_fold_class_weighting": "允許作為預先標示的基準模型設定，但只能由 training fold 計算。",
    "primary_metrics": [
      "balanced_accuracy",
      "per_class_recall",
      "macro_f1",
      "severe_undertriage_rate"
    ],
    "overall_accuracy_is_secondary": true
  },
  "split_policy": {
    "unit": "record_level_emergency_encounter",
    "patient_identifier_available": false,
    "patient_level_split_claim_allowed": false,
    "method": "repeated_stratified_five_fold",
    "fold_count": 5,
    "repeat_count": 5,
    "repeat_seeds": [42, 43, 44, 45, 46],
    "stratification_fields": [
      "KTAS_expert",
      "Group"
    ],
    "assignment_output_fields": [
      "record_index",
      "repeat_id",
      "fold_id",
      "seed"
    ],
    "each_record_is_validation_once_per_repeat": true,
    "preprocessing_fit_scope": "training_fold_only",
    "locked_test_role": "五折中的 validation fold 只產生折外預測；設定鎖定後不得使用其結果反向調參。",
    "limitation": "同一病患若有多次就醫，可能被分到不同 folds；公開資料沒有病患識別碼，無法量化或排除此相依性。"
  },
  "uncertainty_policy": {
    "status": "planned-for-day-28-before-final-model-comparison",
    "method": "paired_record_level_bootstrap_on_aggregated_out_of_fold_predictions",
    "resampling_unit": "record_index",
    "aggregate_repeated_predictions_before_resampling": true,
    "do_not_resample_folds_as_independent_observations": true,
    "patient_level_bootstrap_available": false,
    "limitation": "缺少病患識別碼，因此只能進行紀錄層級重抽樣；同一病患若有多次就醫，信賴區間可能低估相依性造成的不確定性。"
  },
  "output_policy": {
    "assignment_path": "data/processed/ktas-v1/day-15-split-assignments.csv",
    "public_summary_path": "results/public/day-15-data-quality-and-splits.json",
    "run_output_root": "results/runs/day-15",
    "scope": "data_quality_and_split_engineering_not_model_evaluation",
    "aggregate_public_output_only": true
  }
}

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 2：建立 src/triage_rag/data/quality.py

實作缺失概況、結構值域、五級分布、重複分層切分與驗證。

請在文字編輯器建立 src/triage_rag/data/quality.py，貼入以下完整內容並儲存：

"""Profile Day 15 data quality and build reproducible record-level folds."""

from __future__ import annotations

import hashlib
import random
from typing import Any, Mapping

import pandas as pd


JsonObject = dict[str, Any]


class QualityContractError(ValueError):
    """Raised when data quality or split invariants are violated."""


def _rounded_rate(count: int, total: int) -> float:
    return round(count / total, 6) if total else 0.0


def _require_columns(
    frame: pd.DataFrame, required: list[str], frame_name: str
) -> None:
    missing = [name for name in required if name not in frame.columns]
    if missing:
        raise QualityContractError(f"{frame_name} 缺少必要欄位：{missing}")


def validate_aligned_frames(
    model_inputs: pd.DataFrame,
    audit_strata: pd.DataFrame,
    reference_labels: pd.DataFrame,
    row_link_key: str,
) -> None:
    """Require unique and identical record indices across the three prepared tables."""

    frames = {
        "model_inputs": model_inputs,
        "audit_strata": audit_strata,
        "reference_labels": reference_labels,
    }
    for name, frame in frames.items():
        _require_columns(frame, [row_link_key], name)
        if frame[row_link_key].isna().any():
            raise QualityContractError(f"{name} 的 {row_link_key} 不可為空")
        if frame[row_link_key].duplicated().any():
            raise QualityContractError(f"{name} 的 {row_link_key} 必須唯一")

    expected = model_inputs[row_link_key].tolist()
    for name, frame in frames.items():
        if frame[row_link_key].tolist() != expected:
            raise QualityContractError(
                f"{name} 的 {row_link_key} 順序與 model_inputs 不一致"
            )


def build_missingness_profile(
    model_inputs: pd.DataFrame, contract: Mapping[str, Any]
) -> JsonObject:
    """Return aggregate missingness and structural-value checks."""

    policy = contract["missingness_policy"]
    fields = [str(name) for name in policy["profile_fields"]]
    vital_fields = [str(name) for name in policy["vital_sign_fields"]]
    nrs_rule = policy["nrs_pain_rule"]
    applicability_field = str(nrs_rule["applicability_field"])
    _require_columns(
        model_inputs,
        [applicability_field, *fields],
        "model_inputs",
    )

    total = len(model_inputs)
    field_profiles = []
    for name in fields:
        missing = int(model_inputs[name].isna().sum())
        field_profiles.append(
            {
                "field": name,
                "missing_count": missing,
                "missing_rate": _rounded_rate(missing, total),
                "observed_count": total - missing,
            }
        )

    nrs_missing = model_inputs["NRS_pain"].isna()
    pain = model_inputs[applicability_field]
    nrs_not_applicable = int(
        (nrs_missing & pain.eq(nrs_rule["not_applicable_value"])).sum()
    )
    nrs_unknown = int(
        (nrs_missing & pain.eq(nrs_rule["expected_observed_value"])).sum()
    )
    nrs_contradictory = int(
        ((~nrs_missing) & pain.eq(nrs_rule["not_applicable_value"])).sum()
    )

    structural_policy = contract["structural_value_policy"]
    structural_checks: JsonObject = {}
    for name in structural_policy["strictly_positive_fields"]:
        values = pd.to_numeric(model_inputs[name], errors="coerce").dropna()
        nonpositive_count = int((values <= 0).sum())
        structural_checks[str(name)] = {
            "observed_minimum": float(values.min()) if len(values) else None,
            "observed_maximum": float(values.max()) if len(values) else None,
            "nonpositive_count": nonpositive_count,
        }
        if nonpositive_count:
            raise QualityContractError(
                f"{name} 出現 {nonpositive_count} 筆小於或等於 0 的結構非法值"
            )

    for name, bounds in structural_policy["bounded_fields"].items():
        values = pd.to_numeric(model_inputs[name], errors="coerce").dropna()
        outside = values.lt(bounds["minimum"]) | values.gt(bounds["maximum"])
        current = structural_checks.setdefault(
            str(name),
            {
                "observed_minimum": float(values.min()) if len(values) else None,
                "observed_maximum": float(values.max()) if len(values) else None,
            },
        )
        current["declared_minimum"] = bounds["minimum"]
        current["declared_maximum"] = bounds["maximum"]
        current["outside_declared_bounds_count"] = int(outside.sum())
        if outside.any():
            raise QualityContractError(
                f"{name} 出現 {int(outside.sum())} 筆超出 "
                f"{bounds['minimum']} 到 {bounds['maximum']} 的結構非法值"
            )

    any_missing = int(model_inputs[fields].isna().any(axis=1).sum())
    any_vital_missing = int(model_inputs[vital_fields].isna().any(axis=1).sum())
    all_vital_missing = int(model_inputs[vital_fields].isna().all(axis=1).sum())
    return {
        "record_count": total,
        "field_profiles": field_profiles,
        "rows_with_any_profiled_missing": any_missing,
        "rows_with_any_vital_sign_missing": any_vital_missing,
        "rows_with_all_vital_signs_missing": all_vital_missing,
        "nrs_pain_missing_interpretation": {
            "total_missing": int(nrs_missing.sum()),
            "not_applicable_when_pain_is_zero": nrs_not_applicable,
            "unknown_when_pain_is_one": nrs_unknown,
            "observed_score_when_pain_is_zero": nrs_contradictory,
        },
        "structural_value_checks": structural_checks,
        "automatic_extreme_value_deletion": False,
    }


def build_class_distribution(
    reference_labels: pd.DataFrame, contract: Mapping[str, Any]
) -> JsonObject:
    policy = contract["class_imbalance_policy"]
    label_field = str(policy["label_field"])
    label_order = [int(value) for value in policy["label_order"]]
    _require_columns(reference_labels, [label_field], "reference_labels")
    if reference_labels[label_field].isna().any():
        raise QualityContractError(f"{label_field} 不可為空")

    observed = set(reference_labels[label_field].astype(int).unique())
    unexpected = sorted(observed.difference(label_order))
    if unexpected:
        raise QualityContractError(f"{label_field} 出現未允許級數：{unexpected}")

    total = len(reference_labels)
    counts = {
        str(level): int(reference_labels[label_field].eq(level).sum())
        for level in label_order
    }
    nonzero_counts = [count for count in counts.values() if count]
    majority_level = max(label_order, key=lambda level: counts[str(level)])
    majority_count = counts[str(majority_level)]
    high_acuity_count = counts.get("1", 0) + counts.get("2", 0)
    return {
        "label_field": label_field,
        "label_direction": "1 最緊急，5 最不緊急",
        "counts": counts,
        "rates": {
            level: _rounded_rate(count, total) for level, count in counts.items()
        },
        "majority_level": majority_level,
        "majority_class_accuracy": _rounded_rate(majority_count, total),
        "majority_class_balanced_accuracy": round(1 / len(label_order), 6),
        "largest_to_smallest_class_ratio": round(
            max(nonzero_counts) / min(nonzero_counts), 6
        ),
        "high_acuity_level_1_or_2_count": high_acuity_count,
        "high_acuity_level_1_or_2_rate": _rounded_rate(high_acuity_count, total),
    }


def _combined_split_metadata(
    audit_strata: pd.DataFrame,
    reference_labels: pd.DataFrame,
    row_link_key: str,
    stratification_fields: list[str],
) -> pd.DataFrame:
    metadata = audit_strata.merge(
        reference_labels,
        on=row_link_key,
        how="inner",
        validate="one_to_one",
    )
    _require_columns(metadata, [row_link_key, *stratification_fields], "split metadata")
    if metadata[stratification_fields].isna().any().any():
        raise QualityContractError("分層欄位不可為空")
    metadata = metadata[[row_link_key, *stratification_fields]].copy()
    metadata["_stratum"] = metadata[stratification_fields].astype(str).agg("|".join, axis=1)
    return metadata


def build_repeated_stratified_assignments(
    audit_strata: pd.DataFrame,
    reference_labels: pd.DataFrame,
    contract: Mapping[str, Any],
) -> tuple[pd.DataFrame, JsonObject]:
    """Assign each record to one validation fold per repeat."""

    policy = contract["split_policy"]
    row_link_key = str(contract["row_link_key"])
    stratification_fields = [str(name) for name in policy["stratification_fields"]]
    fold_count = int(policy["fold_count"])
    repeat_count = int(policy["repeat_count"])
    seeds = [int(seed) for seed in policy["repeat_seeds"]]
    if len(seeds) != repeat_count or len(set(seeds)) != repeat_count:
        raise QualityContractError("repeat_seeds 必須和 repeat_count 等長且不可重複")
    if fold_count < 2:
        raise QualityContractError("fold_count 必須至少為 2")

    metadata = _combined_split_metadata(
        audit_strata,
        reference_labels,
        row_link_key,
        stratification_fields,
    )
    stratum_sizes = metadata["_stratum"].value_counts().sort_index()
    too_small = stratum_sizes[stratum_sizes < fold_count].to_dict()
    if too_small:
        raise QualityContractError(
            f"每個分層至少需要 {fold_count} 筆，筆數不足={too_small}"
        )

    rows: list[JsonObject] = []
    for repeat_id, seed in enumerate(seeds, start=1):
        generator = random.Random(seed)
        for stratum in sorted(stratum_sizes.index):
            record_indices = sorted(
                metadata.loc[metadata["_stratum"].eq(stratum), row_link_key].tolist()
            )
            generator.shuffle(record_indices)
            fold_order = list(range(1, fold_count + 1))
            generator.shuffle(fold_order)
            for position, record_index in enumerate(record_indices):
                rows.append(
                    {
                        row_link_key: int(record_index),
                        "repeat_id": repeat_id,
                        "fold_id": fold_order[position % fold_count],
                        "seed": seed,
                    }
                )

    assignments = pd.DataFrame(rows).sort_values(
        ["repeat_id", "fold_id", row_link_key], ignore_index=True
    )
    validation = validate_split_assignments(assignments, metadata, contract)
    return assignments, validation


def validate_split_assignments(
    assignments: pd.DataFrame,
    metadata: pd.DataFrame,
    contract: Mapping[str, Any],
) -> JsonObject:
    policy = contract["split_policy"]
    row_link_key = str(contract["row_link_key"])
    expected_columns = [str(name) for name in policy["assignment_output_fields"]]
    if assignments.columns.tolist() != expected_columns:
        raise QualityContractError(
            f"split assignment 欄位應為 {expected_columns}，實際為 {assignments.columns.tolist()}"
        )

    fold_count = int(policy["fold_count"])
    repeat_count = int(policy["repeat_count"])
    expected_ids = set(metadata[row_link_key].tolist())
    for repeat_id in range(1, repeat_count + 1):
        current = assignments.loc[assignments["repeat_id"].eq(repeat_id)]
        if len(current) != len(metadata):
            raise QualityContractError(f"repeat {repeat_id} 的筆數不等於原始資料")
        if current[row_link_key].duplicated().any():
            raise QualityContractError(f"repeat {repeat_id} 有重複 record_index")
        if set(current[row_link_key]) != expected_ids:
            raise QualityContractError(f"repeat {repeat_id} 沒有完整涵蓋所有紀錄")
        if set(current["fold_id"]) != set(range(1, fold_count + 1)):
            raise QualityContractError(f"repeat {repeat_id} 的 fold_id 不完整")

    joined = assignments.merge(
        metadata,
        on=row_link_key,
        how="left",
        validate="many_to_one",
    )
    max_stratum_difference = 0
    for (_, _), group in joined.groupby(["repeat_id", "_stratum"]):
        counts = group["fold_id"].value_counts().reindex(
            range(1, fold_count + 1), fill_value=0
        )
        difference = int(counts.max() - counts.min())
        max_stratum_difference = max(max_stratum_difference, difference)
        if difference > 1:
            raise QualityContractError("同一分層在 folds 間相差超過一筆")

    fold_sizes = joined.groupby(["repeat_id", "fold_id"]).size()
    return {
        "every_record_once_per_repeat": True,
        "all_fold_ids_present": True,
        "maximum_within_stratum_fold_count_difference": max_stratum_difference,
        "minimum_fold_size": int(fold_sizes.min()),
        "maximum_fold_size": int(fold_sizes.max()),
        "assignment_contains_no_label_or_site": not any(
            name in assignments.columns
            for name in contract["split_policy"]["stratification_fields"]
        ),
    }


def build_fold_summary(
    assignments: pd.DataFrame,
    audit_strata: pd.DataFrame,
    reference_labels: pd.DataFrame,
    contract: Mapping[str, Any],
) -> list[JsonObject]:
    row_link_key = str(contract["row_link_key"])
    label_field = str(contract["class_imbalance_policy"]["label_field"])
    site_field = "Group"
    joined = (
        assignments.merge(
            reference_labels[[row_link_key, label_field]],
            on=row_link_key,
            validate="many_to_one",
        )
        .merge(
            audit_strata[[row_link_key, site_field]],
            on=row_link_key,
            validate="many_to_one",
        )
    )
    label_order = [int(value) for value in contract["class_imbalance_policy"]["label_order"]]
    summaries = []
    for (repeat_id, fold_id), group in joined.groupby(["repeat_id", "fold_id"], sort=True):
        summaries.append(
            {
                "repeat_id": int(repeat_id),
                "fold_id": int(fold_id),
                "record_count": len(group),
                "label_counts": {
                    str(level): int(group[label_field].eq(level).sum())
                    for level in label_order
                },
                "site_counts": {
                    str(site): int(count)
                    for site, count in group[site_field].value_counts().sort_index().items()
                },
            }
        )
    return summaries


def build_day15_report(
    model_inputs: pd.DataFrame,
    audit_strata: pd.DataFrame,
    reference_labels: pd.DataFrame,
    contract: Mapping[str, Any],
) -> tuple[JsonObject, pd.DataFrame]:
    """Build the aggregate Day 15 report and local split assignment table."""

    if contract.get("schema_version") != 1:
        raise QualityContractError("目前只支援 schema_version=1 的 Day 15 契約")
    row_link_key = str(contract["row_link_key"])
    validate_aligned_frames(
        model_inputs,
        audit_strata,
        reference_labels,
        row_link_key,
    )
    missingness = build_missingness_profile(model_inputs, contract)
    distribution = build_class_distribution(reference_labels, contract)
    assignments, validation = build_repeated_stratified_assignments(
        audit_strata,
        reference_labels,
        contract,
    )
    csv_bytes = assignments.to_csv(index=False, lineterminator="\n").encode("utf-8")
    report = {
        "schema_version": 1,
        "contract_id": contract["contract_id"],
        "status": "passed",
        "scope": contract["output_policy"]["scope"],
        "record_count": len(model_inputs),
        "missingness": missingness,
        "class_distribution": distribution,
        "split_plan": {
            "unit": contract["split_policy"]["unit"],
            "patient_identifier_available": contract["split_policy"]["patient_identifier_available"],
            "patient_level_split_claim_allowed": contract["split_policy"]["patient_level_split_claim_allowed"],
            "method": contract["split_policy"]["method"],
            "fold_count": contract["split_policy"]["fold_count"],
            "repeat_count": contract["split_policy"]["repeat_count"],
            "repeat_seeds": contract["split_policy"]["repeat_seeds"],
            "stratification_fields": contract["split_policy"]["stratification_fields"],
            "assignment_row_count": len(assignments),
            "assignment_sha256": hashlib.sha256(csv_bytes).hexdigest(),
            "validation": validation,
            "folds": build_fold_summary(
                assignments,
                audit_strata,
                reference_labels,
                contract,
            ),
        },
        "decisions": {
            "missing_is_never_assumed_normal": True,
            "preprocessing_is_fit_inside_training_fold": True,
            "resampling_before_split_is_forbidden": True,
            "overall_accuracy_is_secondary": True,
            "extreme_values_are_not_automatically_deleted": True,
        },
        "limitations": [
            contract["split_policy"]["limitation"],
            "這是資料品質與切分工程結果，不是模型效能或臨床安全結果。",
        ],
    }
    return report, assignments

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 3：建立 scripts/prepare_day15_splits.py

讀取設定與三份 CSV，產生本機分派表、公開聚合摘要與 run manifest。

請在文字編輯器建立 scripts/prepare_day15_splits.py，貼入以下完整內容並儲存：

#!/usr/bin/env python3
"""Profile Day 15 data quality and create repeated stratified folds."""

from __future__ import annotations

import argparse
import sys
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

from triage_rag.data.quality import build_day15_report
from triage_rag.reproducibility import (
    canonical_json_bytes,
    file_record,
    git_state,
    load_json,
    sha256_bytes,
    sha256_file,
    write_json,
)


PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONTRACT = "configs/data/day-15-quality-and-split-contract.json"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="建立 Day 15 缺失概況、類別分布與重複分層五折切分。"
    )
    parser.add_argument("--contract", default=DEFAULT_CONTRACT)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    contract_path = PROJECT_ROOT / args.contract
    contract = load_json(contract_path)
    input_paths = {
        name: PROJECT_ROOT / path for name, path in contract["inputs"].items()
    }
    model_inputs = pd.read_csv(input_paths["model_inputs"])
    audit_strata = pd.read_csv(input_paths["audit_strata"])
    reference_labels = pd.read_csv(input_paths["reference_labels"])

    report, assignments = build_day15_report(
        model_inputs,
        audit_strata,
        reference_labels,
        contract,
    )
    report_sha256 = sha256_bytes(canonical_json_bytes(report))

    assignment_path = PROJECT_ROOT / contract["output_policy"]["assignment_path"]
    assignment_path.parent.mkdir(parents=True, exist_ok=True)
    assignments.to_csv(assignment_path, index=False, lineterminator="\n")
    if sha256_file(assignment_path) != report["split_plan"]["assignment_sha256"]:
        raise RuntimeError("split assignment 寫入後的 SHA-256 與報告不一致")

    public_summary_path = PROJECT_ROOT / contract["output_policy"]["public_summary_path"]
    write_json(public_summary_path, report)

    started_at = datetime.now(timezone.utc)
    run_id = f"{started_at.strftime('%Y%m%dT%H%M%S%fZ')}-{report_sha256[:8]}"
    run_directory = PROJECT_ROOT / contract["output_policy"]["run_output_root"] / run_id
    run_directory.mkdir(parents=True, exist_ok=False)
    report_path = run_directory / "data-quality-and-splits.json"
    write_json(report_path, report)

    tracked_inputs = [
        args.contract,
        contract["source_dataset_card"],
        contract["source_data_contract"],
        *contract["inputs"].values(),
    ]
    manifest = {
        "manifest_schema_version": 1,
        "run_id": run_id,
        "started_at_utc": started_at.isoformat().replace("+00:00", "Z"),
        "command": [sys.executable, *sys.argv],
        "documented_command": [
            "poetry",
            "run",
            "python",
            "scripts/prepare_day15_splits.py",
        ],
        "git": git_state(PROJECT_ROOT),
        "inputs": [file_record(PROJECT_ROOT, path) for path in tracked_inputs],
        "parameters": {
            "fold_count": contract["split_policy"]["fold_count"],
            "repeat_count": contract["split_policy"]["repeat_count"],
            "repeat_seeds": contract["split_policy"]["repeat_seeds"],
            "stratification_fields": contract["split_policy"]["stratification_fields"],
        },
        "outputs": [
            file_record(
                PROJECT_ROOT,
                str(report_path.relative_to(PROJECT_ROOT)),
            ),
            file_record(
                PROJECT_ROOT,
                str(public_summary_path.relative_to(PROJECT_ROOT)),
            ),
            file_record(
                PROJECT_ROOT,
                str(assignment_path.relative_to(PROJECT_ROOT)),
            ),
        ],
        "scope": contract["output_policy"]["scope"],
        "privacy": "公開摘要只含聚合計數；split assignment 只有 record_index 與 fold 編號，且留在 gitignored data/。",
    }
    manifest_path = run_directory / "run-manifest.json"
    write_json(manifest_path, manifest)

    missing = {
        item["field"]: item["missing_count"]
        for item in report["missingness"]["field_profiles"]
    }
    print("Day 15 資料品質與切分：通過")
    print(f"資料筆數：{report['record_count']:,}")
    print(f"缺失最多欄位：Saturation，共 {missing['Saturation']:,} 筆")
    print(
        "專家級數分布："
        + ", ".join(
            f"第{level}級 {count} 筆"
            for level, count in report["class_distribution"]["counts"].items()
        )
    )
    print(
        f"切分：{report['split_plan']['repeat_count']} 次重複 × "
        f"{report['split_plan']['fold_count']} folds，"
        f"共 {report['split_plan']['assignment_row_count']:,} 筆 assignment"
    )
    print(f"公開聚合摘要：{public_summary_path.relative_to(PROJECT_ROOT)}")
    print(f"本機切分檔：{assignment_path.relative_to(PROJECT_ROOT)}")
    print(f"執行目錄：{run_directory.relative_to(PROJECT_ROOT)}")
    print(f"穩定報告摘要：{report_sha256}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 4：建立 tests/test_quality_and_splits.py

以合成資料驗證切分可重現、疼痛缺失語意、非法值與錯誤分層都會被處理。

請在文字編輯器建立 tests/test_quality_and_splits.py，貼入以下完整內容並儲存：

from __future__ import annotations

import copy
import unittest
from pathlib import Path

import pandas as pd

from triage_rag.data.quality import (
    QualityContractError,
    build_day15_report,
)
from triage_rag.reproducibility import load_json


ROOT = Path(__file__).resolve().parents[1]
CONTRACT_PATH = ROOT / "configs" / "data" / "day-15-quality-and-split-contract.json"


class Day15QualityAndSplitTests(unittest.TestCase):
    def setUp(self) -> None:
        self.contract = load_json(CONTRACT_PATH)
        records = []
        audits = []
        labels = []
        record_index = 1
        for level in range(1, 6):
            for group in (1, 2):
                for position in range(6):
                    pain = 0 if position % 2 == 0 else 1
                    nrs = None if pain == 0 else 4
                    if level == 1 and group == 1 and position == 1:
                        nrs = None
                    records.append(
                        {
                            "record_index": record_index,
                            "Pain": pain,
                            "NRS_pain": nrs,
                            "SBP": 120,
                            "DBP": 80,
                            "HR": 80,
                            "RR": 18,
                            "BT": 36.5,
                            "Saturation": None if position == 0 else 98,
                        }
                    )
                    audits.append({"record_index": record_index, "Group": group})
                    labels.append(
                        {"record_index": record_index, "KTAS_expert": level}
                    )
                    record_index += 1
        self.model_inputs = pd.DataFrame(records)
        self.audit_strata = pd.DataFrame(audits)
        self.reference_labels = pd.DataFrame(labels)

    def test_report_and_assignments_are_deterministic(self) -> None:
        first_report, first_assignments = build_day15_report(
            self.model_inputs,
            self.audit_strata,
            self.reference_labels,
            self.contract,
        )
        second_report, second_assignments = build_day15_report(
            self.model_inputs,
            self.audit_strata,
            self.reference_labels,
            self.contract,
        )
        self.assertEqual(first_report, second_report)
        pd.testing.assert_frame_equal(first_assignments, second_assignments)
        self.assertEqual(len(first_assignments), 60 * 5)
        self.assertNotIn("KTAS_expert", first_assignments.columns)
        self.assertNotIn("Group", first_assignments.columns)
        self.assertTrue(
            first_report["split_plan"]["validation"][
                "every_record_once_per_repeat"
            ]
        )

    def test_nrs_missing_is_split_into_not_applicable_and_unknown(self) -> None:
        report, _ = build_day15_report(
            self.model_inputs,
            self.audit_strata,
            self.reference_labels,
            self.contract,
        )
        interpretation = report["missingness"]["nrs_pain_missing_interpretation"]
        self.assertEqual(interpretation["not_applicable_when_pain_is_zero"], 30)
        self.assertEqual(interpretation["unknown_when_pain_is_one"], 1)
        self.assertEqual(interpretation["observed_score_when_pain_is_zero"], 0)

    def test_duplicate_record_index_is_rejected(self) -> None:
        invalid = self.model_inputs.copy()
        invalid.loc[1, "record_index"] = invalid.loc[0, "record_index"]
        with self.assertRaisesRegex(QualityContractError, "必須唯一"):
            build_day15_report(
                invalid,
                self.audit_strata,
                self.reference_labels,
                self.contract,
            )

    def test_structurally_invalid_saturation_is_rejected(self) -> None:
        invalid = self.model_inputs.copy()
        invalid.loc[0, "Saturation"] = 101
        with self.assertRaisesRegex(QualityContractError, "超出 0 到 100"):
            build_day15_report(
                invalid,
                self.audit_strata,
                self.reference_labels,
                self.contract,
            )

    def test_stratum_smaller_than_fold_count_is_rejected(self) -> None:
        keep = ~(
            self.reference_labels["KTAS_expert"].eq(1)
            & self.audit_strata["Group"].eq(1)
        ) | self.reference_labels["record_index"].le(4)
        model_inputs = self.model_inputs.loc[keep].reset_index(drop=True)
        audit_strata = self.audit_strata.loc[keep].reset_index(drop=True)
        reference_labels = self.reference_labels.loc[keep].reset_index(drop=True)
        with self.assertRaisesRegex(QualityContractError, "筆數不足"):
            build_day15_report(
                model_inputs,
                audit_strata,
                reference_labels,
                copy.deepcopy(self.contract),
            )


if __name__ == "__main__":
    unittest.main()

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

完成後產生必要檔案並做靜態檢查

以下命令會在需要時產生套件鎖定檔，接著檢查設定格式與 Python 語法；它們不會啟動模型，也不會把病患資料送到網路：

poetry run python -m json.tool configs/data/day-15-quality-and-split-contract.json
poetry run python -m py_compile src/triage_rag/data/quality.py scripts/prepare_day15_splits.py tests/test_quality_and_splits.py

每個命令都應正常結束。若 JSON 顯示行號，先檢查貼上時是否遺漏逗號、引號或括號；若 py_compile 報錯，先依行號修正縮排或漏貼內容。靜態檢查通過後，再執行本文後面的正式步驟。

實際執行 Day 15

本篇沿用 Python 相依套件管理與封裝工具（Python dependency management and packaging tool, Poetry）。Poetry 依 poetry.lock 使用 Day 13 鎖定的環境。以下命令都從自己的專案根目錄執行，而且不需要 Ollama 模型。

步驟 1：確認中介資料已準備完成

• 目的：讓 Day 15 讀到 Day 12 正規化且依角色拆分的 1,267 筆資料。
• 前置條件：已安裝 Poetry 並完成 poetry sync；若尚未取得 Kaggle 資料，需要可連線下載公開 ZIP。
• 輸入：Day 11 欄位政策、Day 12 資料卡，以及 Kaggle 公開來源。
• 操作：若 data/interim/ktas-v1/ 尚未存在，執行資料管線。

poetry run python scripts/run_ktas_pipeline.py

• 預期輸出：終端機顯示 1,267 筆、24 個來源欄位，並產生候選輸入、稽核分層、參考標籤與排除欄位四份產物。
• 驗證方式：確認 data/interim/ktas-v1/ 至少存在 model-input-candidates.csv、audit-strata.csv 與 reference-labels.csv。若來源安全雜湊演算法 256 位元版本（Secure Hash Algorithm 256-bit, SHA-256）不同，管線應停止，不要直接改掉預期摘要。

步驟 2：執行全部自動化測試

• 目的：先用合成資料證明品質檢查與切分不只走得通，也能拒絕錯誤輸入。
• 前置條件：Poetry 環境已同步；這一步不需要真實 KTAS CSV。
• 輸入：tests/ 中的合成案例、Day 13 至 Day 15 核心模組與設定。
• 操作：執行完整測試套件。

poetry run python -m unittest discover -s tests -v

• 預期輸出：全部十三項測試顯示 ok，最後出現 Ran 13 tests 與 OK。
• 驗證方式：任何 FAIL 或 ERROR 都代表工程規則尚未通過，不應繼續產生切分或模型結果。

本次實際執行結果為：

Ran 13 tests in 0.084s

OK

執行秒數可能因電腦而略有不同，真正要核對的是測試數量與 OK。

步驟 3：建立缺失報告與固定分派表

• 目的：用實際 1,267 筆公開資料執行缺失、值域、類別分布與五次五折切分。
• 前置條件：步驟 1 的三份中介 CSV 存在，步驟 2 全部通過。
• 輸入：Day 15 契約、Day 12 資料卡、Day 14 資料契約與三份中介 CSV。
• 操作：執行本篇入口程式。

poetry run python scripts/prepare_day15_splits.py

• 預期輸出：終端機顯示資料品質與切分通過、1,267 筆、Saturation 缺失 697 筆、五級筆數，以及 5 次重複 × 5 折共 6,335 筆分派。
• 驗證方式：確認下列三種輸出都建立成功。

data/processed/ 是保存切分後索引與未來模型處理產物的本機資料夾，已由 Git 忽略。results/public/ 只保存可公開聚合資訊；results/runs/ 則按每次執行建立獨立目錄。執行產物清單（Run Manifest，下文簡稱 run manifest）會記錄命令、時間、Git 狀態、輸入雜湊與輸出雜湊，讓日後知道某份結果是怎麼產生的。

本次實際得到的穩定聚合報告摘要為：

f30c189620559060d9904fae2d8e55160652a88d3ebfbc2d71c3cfeecc53c3b7

分派 CSV 自己的 SHA-256 則是：

4e254967c612063c57aca7e3c1f8960828135b1c762401438b4b52f279a44c51

兩個摘要用途不同：前者驗證聚合 JSON 內容，後者驗證逐列分派順序。它們都不是 Kaggle ZIP、模型權重或效能結果的摘要。

步驟 4：獨立檢查公開結果

• 目的：不只相信終端機的「通過」，而是直接讀取程式寫出的公開 JSON。
• 前置條件：步驟 3 已成功完成。
• 輸入：results/public/day-15-data-quality-and-splits.json。
• 操作：下列 Python 片段只讀取聚合結果，不會修改檔案。

poetry run python - <<'PY'
import json
from pathlib import Path

report = json.loads(
    Path("results/public/day-15-data-quality-and-splits.json")
    .read_text(encoding="utf-8")
)
validation = report["split_plan"]["validation"]

print("資料筆數：", report["record_count"])
print("分派筆數：", report["split_plan"]["assignment_row_count"])
print("每筆每次只出現一次：", validation["every_record_once_per_repeat"])
print("分派檔不含標籤或場域：", validation["assignment_contains_no_label_or_site"])
print("最小／最大折：", validation["minimum_fold_size"], validation["maximum_fold_size"])
PY

• 預期輸出：資料筆數為 1,267、分派筆數為 6,335、兩項布林值都是 True，所有 25 個折的最小與最大筆數分別為 251 與 256。
• 驗證方式：若布林值不是 True、分派筆數不是 (1267\times 5)，或折大小超出報告範圍，代表分派檔與契約不一致，應停止後續實驗。

Bootstrap 策略現在只鎖原則，不假裝已有區間

重抽樣（Bootstrap Resampling）是從既有樣本中有放回地重複抽取資料，形成許多同樣大小的重抽樣資料集，再觀察評估指標如何波動。它會在 Day 28 用來估計模型比較的不確定性；Day 15 還沒有模型預測，因此現在不能產生 95% 信賴區間。

本篇先鎖定以下原則：

1. 同一筆紀錄跨五次重複得到的折外預測，要先依預先定義的規則彙整成一筆。
2. 後續比較兩個方法時，以同一批 record_index 做成對重抽樣，不能替兩個方法各抽各的資料。
3. 抽樣單位是紀錄，不是 25 個 folds；折不是彼此獨立的病患樣本。
4. 缺少病患識別碼，所以只能稱紀錄層級 bootstrap，並保留可能低估相依性的限制。
5. 重抽樣次數、信賴區間算法與亂數種子要在查看最終模型差異前鎖定。

這樣做的目的，是避免先看到哪種算法會產生較漂亮的區間，再倒過來挑選設定。現在鎖定的是分析順序與單位，不是尚未發生的模型結果。

本篇結果證明了什麼，又沒有證明什麼？

本篇實際完成並驗證的內容包括：

• 1,267 筆資料與三份輸入表的索引完全對齊。
• 七個欄位的缺失筆數、缺失率與結構值域已有可重跑報告。
• 疼痛分數的 556 筆缺失已拆成 553 筆不適用與 3 筆未知／未記錄。
• 五級專家標籤分布已有聚合統計，第一級到第五級都被保留。
• 五次重複、每次五折的 6,335 筆分派可以由固定種子重建。
• 每筆紀錄在每次重複恰好進入一個 validation fold。
• 同一級數與場域在五折間最多相差一筆。
• 分派檔沒有 KTAS_expert、Group 或任何病患輸入值。

這些結果沒有證明：

• 缺失是隨機發生，或血氧缺失與病患狀況無關。
• 極端值一定正確，只是目前沒有足夠證據自動刪除。
• 交叉驗證等同獨立醫院或