Merge remote-tracking branch 'origin/sub_desktop_1' into main_laptop_1

B09_Estimation/B09_Estimation_Guards.py

@@ -1,17 +1,20 @@
-"""B09 원가계산 — 이중계상 감시 (거울 테스트 3종).
+"""B09 원가계산 — 이중계상 감시 (거울 테스트 4종).
 
 **왜 있는가** — 수량(B08)과 원가(B09)의 담당이 갈렸다가 합쳐졌다가 다시 갈리는 동안,
 「할증을 두 번 붙인다·20 m 운반을 또 센다·콘크리트를 두 번 쪼갠다」 세 자리가 반복해서
 위험 항목으로 올라왔다(PLAN 8-7 금지 규칙). 주석은 읽히지 않으므로 **수치로 깨지는
 검사**를 두어, 규칙을 어기면 계산이 멈추게 한다.
 
-세 규칙 (원문 = PLAN 8-7 ㉠㉡㉢)
+네 규칙 (㉠㉡㉢ 원문 = PLAN 8-7 · ㉣ = PLAN 9-6, 2026-09-07 구현 중 발견)
   ㉠ **할증은 자재총괄에서 딱 한 번.** 일위대가 재료비 구성은 **할증 전** 값을 쓴다
      (품셈 1-3-1 「할증 중복 적용 금지」).
   ㉡ **소운반 20 m 이내(`free_haul`)는 내역 줄에 단가를 붙이지 않는다.** 품에 이미
      포함돼 있고, 품셈에 20 m 이내 운반 품목 자체가 없다(1-2-7 · 인력운반 10-6).
   ㉢ **콘크리트·모르터는 한 번만 쪼갠다.** 원단위표는 「㎥」까지 내고, 시멘트·모래
      분해는 일위대가에서 한 번만 한다.
+  ㉣ **작업효율 `E` 를 두 곳에 넣지 않는다.** 품셈은 작업량 산정에 넣는다
+     (`Q = n·q·f·E`, 8-1-4). 시간당 사용료는 1일 8시간으로 나눈다(8-1-6).
+     실무 관측이 사용료 쪽에 미리 넣어 두어, 그 값을 보고 나눗수를 고치는 사고가 난다.
 """
 
 from __future__ import annotations
@@ -109,3 +112,44 @@ def check_mix_decomposed_once(
             f"{cement_per_m3_kg} kg/㎥ = {expected:,.2f} kg 를 넘습니다 — 배합을 두 번 "
             "쪼갰습니다. 원단위표는 「콘크리트 ㎥」까지만 냅니다 (PLAN 8-7 ㉢)."
         )
+
+
+def reject_efficiency_in_hourly_rate(factor: Decimal | None, *, where: str) -> None:
+    """㉣ 작업효율 `E`(실작업시간율)가 **시간당 사용료 쪽**에 들어오면 멈춘다.
+
+    품셈은 작업효율을 **작업량 산정**에 넣는다 — `Q = n·q·f·E` (8-1-4).
+    시간당 사용료는 **1일 8시간**으로 나눈다 (8-1-6 「관리비는 1일 8시간 초과해도
+    8시간으로 계산」). **양쪽에 다 넣으면 같은 효율이 두 번 곱해진다.**
+
+    실무 관측(STC 2024)은 효율을 사용료 쪽에 미리 넣어 굴착기 0.7㎥ 노무가
+    55,700 원/hr(일당 ÷ 약 5.09시간)로 나온다. 그 값을 보고 「우리가 틀렸다」며
+    나눗수를 고치는 것이 정확히 이 자리에서 일어난다.
+    """
+    if factor is None:
+        return
+    if Decimal(0) < Decimal(str(factor)) < Decimal(1):
+        raise DoubleCountError(
+            f"{where}: 작업효율({factor})이 시간당 사용료 계산에 들어왔습니다 — "
+            "효율은 작업량 산정(Q = n·q·f·E)에만 넣습니다. 양쪽에 넣으면 두 번 곱해집니다 "
+            "(PLAN 9-6 ㉣)."
+        )
+
+
+def check_operator_hours_basis(
+    *,
+    labor_per_hour: Decimal,
+    daily_wage: Decimal,
+    person_days: Decimal = Decimal(1),
+    hours_per_day: int = 8,
+) -> None:
+    """㉣ 보조 — 조종원 노임 나눗수가 8시간인가.
+
+    나눗수를 몰래 줄이는 것이 곧 효율을 사용료에 넣는 것과 같다.
+    """
+    expected = daily_wage * person_days / Decimal(hours_per_day)
+    if abs(labor_per_hour - expected) > _TOLERANCE:
+        raise DoubleCountError(
+            f"조종원 시간당 노무비 {labor_per_hour:,.2f} 가 "
+            f"{daily_wage:,.0f} × {person_days} ÷ {hours_per_day}h = {expected:,.2f} 와 다릅니다 — "
+            "나눗수를 줄이면 작업효율을 사용료에 넣은 것이 됩니다 (PLAN 9-6 ㉣)."
+        )

B09_Estimation/B09_Estimation_MachineCost.py

@@ -27,12 +27,28 @@ from dataclasses import dataclass, field
 from decimal import Decimal
 from typing import Any
 
+from B09_Estimation.B09_Estimation_Guards import (
+    check_operator_hours_basis,
+    reject_efficiency_in_hourly_rate,
+)
 from B09_Estimation.B09_Estimation_PriceBook import Money3
 
 _CATALOG_SUBPATH = ("resources", "data_cost_input_value")
 _THOUSAND = Decimal(1000)
 
 
+#: 조종원 노임을 시간당으로 바꿀 때 나누는 시간.
+#: 근거 — 품셈 8-1-6 기계손료 단서 「**관리비는 1일 8시간 초과해도 8시간으로 계산**」이
+#: 1일을 8시간으로 잡는다. 노임표도 `hours_per_day: 8` 이다.
+#: ⚠ **실무 관측은 이보다 짧은 시간으로 나눈 사례가 있다** — STC 2024 관측
+#: 굴착기 0.7㎥ 노무 55,700 원/hr 은 건설기계운전사 일당 283,297 을 **약 5.09시간**으로
+#: 나눈 값이다(작업효율 `E`(실작업시간율)를 시간당 사용료에 미리 반영한 것으로 보임).
+#: 품셈에서 작업효율은 **작업량 산정**(`Q = n·q·f·E`, 8-1-4)에 들어가지 시간당 사용료에
+#: 들어가지 않으므로 **8시간을 유지한다.** 관측에 맞추려 나눗수를 바꾸지 않는다
+#: (★법대로 PLAN 8-10). TODO(미결 PLAN 9-6): 발주처가 실가동시간을 요구하는 사례 확인.
+OPERATOR_HOURS_PER_DAY = 8
+
+
 class MachineCostError(LookupError):
     """기계경비를 세울 수 없는 경우. 0 으로 때우지 않고 멈춘다."""
 
@@ -147,7 +163,8 @@ def hourly_machine_cost(
     fuel_liters_per_hour: Decimal | None = None,
     fuel_price_per_liter: Decimal | None = None,
     operator_daily_wage: Decimal | None = None,
-    operator_hours_per_day: int = 8,
+    operator_hours_per_day: int = OPERATOR_HOURS_PER_DAY,
+    efficiency_factor: Decimal | None = None,
 ) -> HourlyMachineCost:
     """시간당 사용료 한 시간분.
 
@@ -155,6 +172,11 @@ def hourly_machine_cost(
     0 으로 때우지 않고 `gaps` 에 적어 돌려준다** — 빠진 채로 총액에 섞이는 것이
     이 자리에서 제일 위험하다.
     """
+    # ㉣ 작업효율이 이 자리에 들어오면 멈춘다 — 효율은 작업량 산정에만 들어간다.
+    reject_efficiency_in_hourly_rate(
+        efficiency_factor, where=f"{machine.display_name} 시간당 사용료"
+    )
+
     gaps: list[str] = []
     expense = hourly_loss_cost(machine)
 
@@ -171,6 +193,12 @@ def hourly_machine_cost(
         gaps.append("운전사 직종 매핑 미확보 — 노무비 성분 비어 있음")
     else:
         labor = operator_daily_wage / Decimal(operator_hours_per_day)
+        # ㉣ 보조 — 나눗수를 몰래 줄이면 효율을 사용료에 넣은 것이 된다.
+        check_operator_hours_basis(
+            labor_per_hour=labor,
+            daily_wage=operator_daily_wage,
+            hours_per_day=operator_hours_per_day,
+        )
 
     return HourlyMachineCost(
         machine=machine,

B09_Estimation/B09_Estimation_MachineOperating.py

@@ -0,0 +1,350 @@
+"""B09 원가계산 — 운전경비(연료·잡재료·조종원)를 품셈 8-4 에서 뽑는다.
+
+`mach_base_2026.json` 의 `mach_fuel_rate`·`mach_operator_map` 이 **0건**이라 시간당
+중기사용료의 **재료비·노무비 성분이 비어 있었다**(PLAN 9-3). 그 값은 건설품셈
+**8-4 운전경비 산정** 표에 있다.
+
+**파생 파일로 낸다** — 기준자료(`resources/data_cost_input_value/`)를 고치지 않는다.
+그 파일은 `_manifest.json` 지문으로 재현성을 거는 자리이고 원천 xlsx 에서 다시
+생성되는 물건이라, 값을 채워 넣으면 ① 이미 그 판으로 계산한 스냅샷과 대조가 깨지고
+② 다음 재생성 때 날아가고 ③ 주인이 겹친다.
+⚠ **길게 보면 기준자료 원본이 채워지는 것이 맞다** — `mach_fuel_rate` 가 그 파일 안에
+**빈 변수로 선언**돼 있다는 것이 「원래 거기 들어갈 값」이라는 뜻이다. 원천 재생성 몫이라
+미결로 올려 둔다.
+
+⚠ **표가 열 단위로 뭉쳐 있다.** PDF 에서 뽑히며 한 열이 한 칸에 공백으로 이어 붙었다 —
+`분류번호` 칸 하나에 코드 31개, `주연료` 칸 하나에 값 31개가 들어 있다. 그래서
+**위치로 짝짓고, 짝이 안 맞으면 그 표를 통째로 버린다.** 어긋난 채로 짝지으면 다른
+기종의 연료값이 조용히 붙는다.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import dataclass, field
+from decimal import Decimal
+from typing import Any
+
+_CATALOG_SUBPATH = ("resources", "data_cost_input_value")
+_OUTPUT_SUBPATH = ("resources", "data_cost_machine_operating")
+
+#: 품셈 8-4 운전경비 표의 머리글 — 이 여섯이 다 있어야 그 표로 본다.
+_REQUIRED_HEADERS = ("분류번호", "기계명", "규격", "주연료", "잡재료", "조종원")
+
+_RE_CODE_FULL = re.compile(r"^(\d{4})-(\d{4})$")
+_RE_CODE_TAIL = re.compile(r"^\d{4}$")
+_RE_DECIMAL = re.compile(r"^\d+(?:\.\d+)?$")
+
+#: 연료 종류가 값 앞에 붙는 경우 — 「휘발유0.7」·「중유487.2」.
+_RE_FUEL_WITH_KIND = re.compile(r"^(휘발유|중유|경유)?(\d+(?:\.\d+)?)$")
+
+#: 조종원 직종 — 품셈 표는 「인/일」 수만 주고 직종명을 안 준다.
+#: TODO(미결 PLAN 9-6): 기종별 직종이 품셈 다른 장에 있다. 아래는 `aliases` 기반 **잠정**이며
+#: 결과에 `operator_mapping_is_provisional: true` 로 드러난다.
+_OPERATOR_TRUCK_WORDS = ("덤프트럭", "트럭", "트레일러", "화물")
+_OPERATOR_ALIAS_TRUCK = "labor_op_truck"
+_OPERATOR_ALIAS_CONSTRUCTION = "labor_op_const"
+
+
+class OperatingCostError(ValueError):
+    """운전경비 표를 못 읽은 경우. 어긋난 채로 짝짓지 않는다."""
+
+
+def _project_root() -> str:
+    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _read_json(*parts: str) -> dict[str, Any]:
+    with open(os.path.join(_project_root(), *parts), encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+@dataclass(frozen=True)
+class OperatingRecord:
+    """기종 하나의 운전경비 원단위."""
+
+    machine_code: str
+    machine_name: str
+    specification: str
+    fuel_liters_per_hour: Decimal | None
+    fuel_kind: str
+    misc_material_percent: Decimal | None
+    operator_person_days: Decimal | None
+    operator_occupation_code: str = ""
+    operator_mapping_is_provisional: bool = True
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "machine_code": self.machine_code,
+            "machine_name": self.machine_name,
+            "specification": self.specification,
+            "fuel_liters_per_hour": (
+                None if self.fuel_liters_per_hour is None else str(self.fuel_liters_per_hour)
+            ),
+            "fuel_kind": self.fuel_kind,
+            "misc_material_percent": (
+                None if self.misc_material_percent is None else str(self.misc_material_percent)
+            ),
+            "operator_person_days": (
+                None if self.operator_person_days is None else str(self.operator_person_days)
+            ),
+            "operator_occupation_code": self.operator_occupation_code,
+            "operator_mapping_is_provisional": self.operator_mapping_is_provisional,
+        }
+
+
+@dataclass
+class OperatingParseResult:
+    records: list[OperatingRecord] = field(default_factory=list)
+    #: 짝이 안 맞아 버린 표 — 조용히 넘기지 않고 센다.
+    dropped_tables: list[str] = field(default_factory=list)
+
+
+def _tokens(cell: str) -> list[str]:
+    return [t for t in re.split(r"\s+", str(cell or "").strip()) if t]
+
+
+def expand_codes(tokens: list[str]) -> list[str]:
+    """`0201-0012 0020 0040` → `0201-0012 · 0201-0020 · 0201-0040`.
+
+    뒤 코드는 앞 코드의 **앞 네 자리를 이어받는다**. 이어받을 앞자리가 없으면 버린다.
+    """
+    codes: list[str] = []
+    prefix = ""
+    for token in tokens:
+        full = _RE_CODE_FULL.match(token)
+        if full:
+            prefix = full.group(1)
+            codes.append(token)
+            continue
+        if _RE_CODE_TAIL.match(token) and prefix:
+            codes.append(f"{prefix}-{token}")
+            continue
+        return []  # 코드 열이 아닌 표
+    return codes
+
+
+def _parse_fuel(token: str) -> tuple[Decimal | None, str]:
+    """「11.6」·「휘발유0.7」·「-」 를 값과 연료 종류로 가른다."""
+    text = token.strip()
+    if text in ("-", "–", ""):
+        return None, ""
+    match = _RE_FUEL_WITH_KIND.match(text)
+    if not match:
+        return None, ""
+    kind = match.group(1) or "경유"
+    return Decimal(match.group(2)), kind
+
+
+def _parse_percent(token: str) -> Decimal | None:
+    text = token.strip().rstrip("%")
+    return Decimal(text) if _RE_DECIMAL.match(text) else None
+
+
+def _parse_person_days(token: str) -> Decimal | None:
+    text = token.strip()
+    return Decimal(text) if _RE_DECIMAL.match(text) else None
+
+
+def _operator_code(machine_name: str, aliases: dict[str, str]) -> str:
+    """기종 이름으로 운전사 직종을 고른다 — **잠정 규칙**.
+
+    품셈 8-4 표는 「조종원 인/일」 수만 주고 직종명을 안 준다. 노임표의 `aliases` 가
+    운전사 직종 셋(`labor_op_const`·`labor_op_truck`·`labor_op_general`)을 들고 있어
+    트럭 계열만 화물차운전사로, 나머지는 건설기계운전사로 **잠정** 매핑한다.
+    """
+    key = (
+        _OPERATOR_ALIAS_TRUCK
+        if any(word in machine_name for word in _OPERATOR_TRUCK_WORDS)
+        else _OPERATOR_ALIAS_CONSTRUCTION
+    )
+    return aliases.get(key, "")
+
+
+def parse_operating_tables(
+    pum: dict[str, Any],
+    aliases: dict[str, str],
+) -> OperatingParseResult:
+    """품셈 표 뭉치에서 8-4 운전경비 표만 골라 기종별 원단위를 만든다."""
+    result = OperatingParseResult()
+    for table in pum.get("tables", []):
+        headers = table.get("headers") or []
+        joined = " ".join(headers)
+        if not all(word in joined for word in _REQUIRED_HEADERS):
+            continue
+
+        for row in table.get("rows") or []:
+            if len(row) < 6:
+                continue
+            codes = expand_codes(_tokens(row[0]))
+            names = _tokens(row[1])
+            specs = _tokens(row[2])
+            fuels = _tokens(row[3])
+            miscs = _tokens(row[4])
+            operators = _tokens(row[5])
+
+            # ⚠ 위치로 짝짓는다 — 개수가 안 맞으면 그 줄을 통째로 버린다.
+            if not codes or not (len(codes) == len(specs) == len(fuels) == len(operators)):
+                result.dropped_tables.append(f"{table.get('section', '')[:40]} (개수 불일치)")
+                continue
+
+            # ⚠ 이름 칸은 **표 전체의 기종명이 한 덩어리로** 들어온다
+            # (「불도저(무한궤도)불도저(타이어)습지불도저굴착기(무한궤도)…」).
+            # 코드별로 못 가르므로 **이름은 여기서 안 쓴다** — 613 기종 카탈로그에서
+            # `machine_code` 로 찾아 붙인다(`enrich_with_catalog`).
+            _ = names
+            for index, code in enumerate(codes):
+                fuel, kind = _parse_fuel(fuels[index])
+                misc = _parse_percent(miscs[index]) if index < len(miscs) else None
+                result.records.append(
+                    OperatingRecord(
+                        machine_code=code,
+                        machine_name="",  # 카탈로그에서 채운다
+                        specification=specs[index],
+                        fuel_liters_per_hour=fuel,
+                        fuel_kind=kind,
+                        misc_material_percent=misc,
+                        operator_person_days=_parse_person_days(operators[index]),
+                        operator_occupation_code="",  # 이름을 안 뒤 정한다
+                    )
+                )
+    return result
+
+
+def enrich_with_catalog(
+    result: OperatingParseResult,
+    catalog: Any,
+    aliases: dict[str, str],
+) -> OperatingParseResult:
+    """코드로 613 기종 카탈로그에서 **이름·규격을 가져와** 채운다.
+
+    품셈 표의 이름 칸은 통째로 뭉쳐 와 못 쓴다. 코드가 정본이다.
+    카탈로그에 없는 코드는 **버리고 목록에 남긴다** — 이름 없는 기종은 단가가 못 붙는다.
+    """
+    enriched: list[OperatingRecord] = []
+    for record in result.records:
+        machine = catalog.machines.get(record.machine_code)
+        if machine is None:
+            result.dropped_tables.append(f"{record.machine_code} (기종 카탈로그에 없음)")
+            continue
+        enriched.append(
+            OperatingRecord(
+                machine_code=record.machine_code,
+                machine_name=machine.name,
+                specification=machine.specification or record.specification,
+                fuel_liters_per_hour=record.fuel_liters_per_hour,
+                fuel_kind=record.fuel_kind,
+                misc_material_percent=record.misc_material_percent,
+                operator_person_days=record.operator_person_days,
+                operator_occupation_code=_operator_code(machine.name, aliases),
+            )
+        )
+    result.records = enriched
+    return result
+
+
+def load_operating_records(
+    pum_file: str = "pum_const_2026.json",
+    labor_file: str = "labor_const_2026-01-01.json",
+) -> OperatingParseResult:
+    from B09_Estimation.B09_Estimation_MachineCost import load_machine_catalog
+
+    pum = _read_json(*_CATALOG_SUBPATH, pum_file)["variables"]["pum"]
+    aliases = _read_json(*_CATALOG_SUBPATH, labor_file)["variables"].get("aliases", {})
+    parsed = parse_operating_tables(pum, aliases)
+    return enrich_with_catalog(parsed, load_machine_catalog(), aliases)
+
+
+def write_operating_records(
+    result: OperatingParseResult,
+    *,
+    source_dataset_version: dict[str, str],
+    output_dir: str | None = None,
+) -> str:
+    """파생 파일로 낸다. **어느 기준자료 판에서 파생됐는지**를 함께 적는다."""
+    directory = output_dir or os.path.join(_project_root(), *_OUTPUT_SUBPATH)
+    os.makedirs(directory, exist_ok=True)
+    path = os.path.join(directory, "machine_operating_2026.json")
+
+    payload = {
+        "schema_version": "1.0",
+        "dataset_id": "machine_operating_derived",
+        "derived_from": source_dataset_version,
+        "note": (
+            "건설품셈 8-4 운전경비 산정에서 뽑은 파생본. 기준자료가 아니다 — "
+            "`mach_fuel_rate`·`mach_operator_map` 이 채워지면 이 파일은 걷어낸다."
+        ),
+        "policy": {
+            "operator_mapping_is_provisional": True,
+            "operator_mapping_rule": "트럭 계열 = 화물차운전사, 그 밖 = 건설기계운전사 (잠정)",
+        },
+        "stats": {"records": len(result.records), "dropped": len(result.dropped_tables)},
+        "records": [r.as_dict() for r in result.records],
+        "dropped_tables": result.dropped_tables,
+    }
+    with open(path, "w", encoding="utf-8", newline="\n") as handle:
+        json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
+    return path
+
+
+def load_fuel_price(oil_file: str = "oil_2026-08-14.json") -> tuple[Decimal, dict[str, str]]:
+    """경유 단가와 그 판의 신원.
+
+    ⚠ **전국평균이다.** 품셈 8-1-7 5호는 「유류가격은 **해당 지역의 가격**」이라
+    규정하므로 나중에 현장 소재지 값으로 갈아끼울 자리다 (TODO 미결 PLAN 9-6).
+    지역 파라미터 자리만 뚫어 두고 지금은 전국평균을 잠정으로 쓴다.
+    """
+    payload = _read_json(*_CATALOG_SUBPATH, oil_file)
+    diesel = payload["variables"]["oil_diesel"]
+    return Decimal(str(diesel["value"])), {
+        "dataset_id": payload.get("dataset_id", ""),
+        "effective_date": payload.get("effective_date", ""),
+        "scope": diesel.get("scope", ""),
+    }
+
+
+def load_operator_wages(labor_file: str = "labor_const_2026-01-01.json") -> dict[str, Decimal]:
+    """직종코드 → 일당."""
+    records = _read_json(*_CATALOG_SUBPATH, labor_file)["variables"]["labor_rate"]["records"]
+    return {
+        str(r["occupation_code"]): Decimal(str(r["daily_wage_krw"]))
+        for r in records
+        if "daily_wage_krw" in r
+    }
+
+
+def hourly_cost_of(machine_code: str, *, region: str | None = None):
+    """기종 하나의 **시간당 사용료 3분할**을 완성해 돌려준다.
+
+    재료비 = 주연료 × 유가 + 잡재료(주연료의 %) / 노무비 = 조종원 일당 ÷ 8시간 /
+    경비 = 손료. `region` 은 유가 지역값 자리 — 지금은 전국평균만 있어 무시된다.
+    """
+    from B09_Estimation.B09_Estimation_MachineCost import hourly_machine_cost, load_machine_catalog
+
+    catalog = load_machine_catalog()
+    machine = catalog.get(machine_code)
+    records = {r.machine_code: r for r in load_operating_records().records}
+    record = records.get(machine_code)
+    if record is None:
+        return hourly_machine_cost(machine)
+
+    fuel_price, _ = load_fuel_price()
+    wages = load_operator_wages()
+
+    liters = record.fuel_liters_per_hour
+    if liters is not None and record.misc_material_percent is not None:
+        # 잡재료는 **주연료의 %** 라 유가와 같이 움직인다(PLAN 8-18 유가 민감분).
+        liters = liters * (Decimal(1) + record.misc_material_percent / Decimal(100))
+
+    wage = wages.get(record.operator_occupation_code)
+    if record.operator_person_days is not None and wage is not None:
+        wage = wage * record.operator_person_days
+
+    return hourly_machine_cost(
+        machine,
+        fuel_liters_per_hour=liters,
+        fuel_price_per_liter=fuel_price if liters is not None else None,
+        operator_daily_wage=wage,
+    )

B09_Estimation/B09_Estimation_UnitPrice.py

@@ -0,0 +1,232 @@
+"""B09 원가계산 — ③ 단가산출·일위대가 조립 (PLAN 9-3 · 9-5).
+
+자원 축(`resource_axis`)이 「이 공종 1단위에 무엇이 얼마나」를 갖고 있고, 카탈로그가
+「그 자원 하나가 얼마」를 갖고 있다. 이 모듈이 둘을 곱해 **일위대가 한 줄**을 만든다.
+
+층은 그대로 쌓는다 (PLAN 9-3):
+
+    S 취득가 · L 노임 · M 자재  →  X 시간당 중기사용료  →  B 일위대가
+
+`PriceBook` 에 제목·상세로 앉히므로 **표를 따로 만들지 않는다.**
+
+**부르는 가드** (함수만 있고 안 부르면 없는 것과 같다)
+  - ㉠ 자재는 **할증 전** 값 — 할증은 자재총괄 한 곳뿐 (`check_surcharge_once`).
+  - ㉣ 작업효율은 사용료 쪽에 안 넣음 (`reject_efficiency_in_hourly_rate`,
+    `B09_Estimation_MachineCost` 안에서 호출됨).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from decimal import Decimal
+
+from B09_Estimation.B09_Estimation_Guards import check_surcharge_once
+from B09_Estimation.B09_Estimation_MachineCost import load_machine_catalog
+from B09_Estimation.B09_Estimation_MachineOperating import (
+    load_fuel_price,
+    load_operating_records,
+    load_operator_wages,
+)
+from B09_Estimation.B09_Estimation_PriceBook import (
+    PriceBook,
+    PriceDetail,
+    PriceKind,
+    PriceTitle,
+)
+from B09_Estimation.B09_Estimation_ResourceAxis import (
+    AxisResult,
+    build_resource_axis,
+    load_combined_catalog,
+    load_labor_catalog,
+    load_work_item_master,
+)
+
+_ZERO = Decimal(0)
+#: 연료는 자재 카탈로그에 없어 합성 코드로 세운다 — 코드가 있어야 조인이 성립한다.
+FUEL_CODE_PREFIX = "M-FUEL-"
+
+
+def _slots(value: Decimal) -> list[Decimal | None]:
+    """6번(적용 단가) 슬롯에만 값을 넣는다 — 유료 물가지 미구독 상태의 기본 모양."""
+    slots: list[Decimal | None] = [None] * 6
+    slots[5] = value
+    return slots
+
+
+@dataclass
+class UnitPriceBuild:
+    book: PriceBook = field(default_factory=PriceBook)
+    #: 세우지 못한 공종 — 값이 안 서는 것을 빈 줄로 두지 않는다.
+    skipped: list[str] = field(default_factory=list)
+    #: 기계 성분이 비어 있는 채로 쓰인 기종 (연료·조종원 미확보).
+    incomplete_machines: list[str] = field(default_factory=list)
+
+
+def _add_labor_titles(book: PriceBook, wages: dict[str, Decimal]) -> None:
+    catalog = load_labor_catalog()
+    for entry in catalog.entries:
+        wage = wages.get(entry.code)
+        if wage is None or entry.code in book.titles:
+            continue
+        book.add_title(
+            PriceTitle(
+                code=entry.code,
+                kind=PriceKind.LABOR,
+                name=entry.name,
+                unit="인",
+                slots=_slots(wage),
+            )
+        )
+
+
+def _add_machine_layers(book: PriceBook, machine_codes: set[str]) -> list[str]:
+    """`S`(취득가) · `L`(운전사) · `M`(연료) 을 세우고 그 위에 `X` 를 올린다.
+
+    시간당 사용료를 **미리 계산해 넣지 않는다** — 층을 실제로 쌓아야 화면이
+    「무엇으로 이루어졌나」를 보일 수 있다(PLAN 8-13 계산 과정을 감추지 않음).
+    """
+    catalog = load_machine_catalog()
+    operating = {r.machine_code: r for r in load_operating_records().records}
+    fuel_price, _ = load_fuel_price()
+    wages = load_operator_wages()
+    incomplete: list[str] = []
+
+    fuel_code = f"{FUEL_CODE_PREFIX}경유"
+    if fuel_code not in book.titles:
+        book.add_title(
+            PriceTitle(
+                code=fuel_code,
+                kind=PriceKind.MATERIAL,
+                name="경유",
+                unit="L",
+                slots=_slots(fuel_price),
+            )
+        )
+
+    for code in sorted(machine_codes):
+        machine = catalog.machines.get(code)
+        record = operating.get(code)
+        if machine is None or machine.loss_coefficient_per_hour is None or record is None:
+            incomplete.append(code)
+            continue
+
+        base_code = f"S-{code}"
+        hourly_code = f"X-{code}"
+        if hourly_code in book.titles:
+            continue
+
+        # S — 취득가에서 나온 시간당 손료. 경비 성분만 갖는다.
+        book.add_title(
+            PriceTitle(
+                code=base_code,
+                kind=PriceKind.MACHINE_BASE,
+                name=machine.name,
+                spec=machine.specification,
+                unit="hr",
+                slots=_slots(
+                    machine.price_thousand_krw * Decimal(1000) * machine.loss_coefficient_per_hour
+                ),
+            )
+        )
+        book.add_title(
+            PriceTitle(
+                code=hourly_code,
+                kind=PriceKind.MACHINE_HOURLY,
+                name=machine.name,
+                spec=machine.specification,
+                unit="hr",
+            )
+        )
+        book.add_detail(PriceDetail(hourly_code, base_code, Decimal(1), note="시간당 손료"))
+
+        liters = record.fuel_liters_per_hour
+        if liters is not None:
+            if record.misc_material_percent is not None:
+                # 잡재료는 **주연료의 %** — 유가와 같이 움직인다.
+                liters = liters * (Decimal(1) + record.misc_material_percent / Decimal(100))
+            book.add_detail(PriceDetail(hourly_code, fuel_code, liters, note="주연료 + 잡재료"))
+        else:
+            incomplete.append(f"{code} (연료소모량 없음)")
+
+        wage_code = record.operator_occupation_code
+        if wage_code and wage_code in wages and record.operator_person_days is not None:
+            # ㉣ 나눗수는 8시간 — `PriceDetail` 수량이 「1시간분 인」이 된다.
+            per_hour_person = record.operator_person_days / Decimal(8)
+            if wage_code not in book.titles:
+                book.add_title(
+                    PriceTitle(
+                        code=wage_code,
+                        kind=PriceKind.LABOR,
+                        name="조종원",
+                        unit="인",
+                        slots=_slots(wages[wage_code]),
+                    )
+                )
+            book.add_detail(
+                PriceDetail(hourly_code, wage_code, per_hour_person, note="조종원 (1일 8시간)")
+            )
+        else:
+            incomplete.append(f"{code} (조종원 없음)")
+
+    return incomplete
+
+
+def build_unit_prices(axis: AxisResult | None = None) -> UnitPriceBuild:
+    """자원 축을 일위대가(`B`)로 조립한다.
+
+    공종 하나에 붙은 자원 줄들을 그 공종의 상세로 삼는다. 자원이 하나도 안 붙은
+    공종은 **빈 줄로 세우지 않고 건너뛴다** — 0 원 일위대가가 내역에 서면 안 된다.
+    """
+    if axis is None:
+        axis = build_resource_axis(load_work_item_master(), load_combined_catalog())
+
+    build = UnitPriceBuild()
+    wages = load_operator_wages()
+    _add_labor_titles(build.book, wages)
+
+    machine_codes = {r.resource_code for r in axis.rows if r.resource_kind == "machine"}
+    build.incomplete_machines = _add_machine_layers(build.book, machine_codes)
+
+    by_item: dict[str, list] = {}
+    for row in axis.rows:
+        by_item.setdefault(row.work_item_code, []).append(row)
+
+    for work_item_code, rows in sorted(by_item.items()):
+        title_code = f"B-{work_item_code}"
+        if title_code in build.book.titles:
+            continue
+        unit = next((r.amount_unit for r in rows if r.amount_unit), "")
+        build.book.add_title(
+            PriceTitle(code=title_code, kind=PriceKind.UNIT_PRICE, name=work_item_code, unit=unit)
+        )
+        added = 0
+        for row in rows:
+            ref = row.resource_code if row.resource_kind == "labor" else f"X-{row.resource_code}"
+            if ref not in build.book.titles:
+                continue
+            build.book.add_detail(PriceDetail(title_code, ref, row.amount))
+            added += 1
+        if added == 0:
+            build.skipped.append(work_item_code)
+    return build
+
+
+def material_total_before_surcharge(build: UnitPriceBuild, code: str) -> Decimal:
+    """일위대가 한 줄의 **할증 전** 재료비 합계 — ㉠ 가드에 넘길 값."""
+    return build.book.resolve(code).material
+
+
+def verify_surcharge_once(
+    build: UnitPriceBuild,
+    code: str,
+    *,
+    material_summary_total: Decimal,
+    surcharge_rate_percent: Decimal,
+) -> None:
+    """㉠ 자재총괄 합과 대조한다 — 할증이 두 번 붙었으면 여기서 멈춘다."""
+    check_surcharge_once(
+        material_summary_total=material_summary_total,
+        unit_price_material_total=material_total_before_surcharge(build, code),
+        surcharge_rate_percent=surcharge_rate_percent,
+        label=code,
+    )