diff --git a/backend/app/api/ext_input_ledgers.py b/backend/app/api/ext_input_ledgers.py
new file mode 100644
index 00000000..332e7d5d
--- /dev/null
+++ b/backend/app/api/ext_input_ledgers.py
@@ -0,0 +1,579 @@
+"""外部输入台账录入/查询 API（能力评价 M2 准入3 派生，2026-09-15）
+
+三张台账的单向录入接口（新增 / CSV 批量导入 / 按实体与期次查询）+ 台账→kpi_values 派生触发。
+
+设计口径：
+- **单向**：只提供「新增 + 批量导入 + 查询 + 派生」，不提供修改/删除接口
+  （历史登记行不可改；错登由人工在业务侧作废后重新登记一行，保持审计链）
+- **on_time 由服务端派生**：请求里带 on_time 一律忽略并在响应回报 on_time_ignored
+- **人事台账等式校验**：期末 ≠ 期初 + 入职 − 离职 → 拒绝写入并回报（HTTP 400 / 批量逐行 errors）
+- **多租户**：所有端点经 app.deps.assert_entity_scope 收敛账套（显式 entity_id 与会话账套不一致 → 403）
+- **鉴权**：router 级 require_auth（未认证 401）；写端点额外 require_role(ceo/finance/it)（角色不足 403）
+- 派生必须**幂等**（同 (entity_id, kpi_id, period) 重复运行只更新不新增），分母为 0 → 不写值
+
+CSV 列头（UTF-8，可带 BOM；列序无关，多余列忽略）：
+  申报台账  period,tax_type,legal_deadline,filed_date,evidence_no,registered_by,remark
+  人事台账  period,headcount_begin,hires,separations,headcount_end,registered_by,remark
+  证照台账  licence_name,licence_no,valid_from,valid_to,annual_check_status,issuer,owner_role,attachment_no,registered_by,remark
+"""
+from __future__ import annotations
+
+import csv
+import io
+import logging
+import re
+from datetime import date, datetime
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from sqlalchemy.orm import Session
+
+from app.auth_middleware import require_auth, require_role
+from app.database import get_db
+from app.deps import assert_entity_scope
+from app.models import HrMonthlyLedger, KPIDefinition, LicenceLedger, TaxFilingLog
+from app.models.ext_input_ledger import ANNUAL_CHECK_STATUSES, MONTH_PERIOD_RE, QUARTER_PERIOD_RE
+from app.services.ext_input_derive import (
+    ALL_KPI_CODES,
+    LICENCE_OK_CODE,
+    check_headcount_equation,
+    compute_employee_turnover,
+    derive_on_time,
+    explain_licence_ok,
+    explain_tax_timely_rate,
+    is_licence_valid,
+    month_to_quarter,
+    quarter_end_date,
+    run_derive,
+)
+
+logger = logging.getLogger("cma.ext_input_api")
+
+router = APIRouter(
+    prefix="/api/cma/ext-ledger",
+    tags=["外部输入台账"],
+    dependencies=[Depends(require_auth)],
+)
+
+# 写权限（录入/导入/派生）：ceo / finance / it（与 value_sources 等既有写端点同口径）
+write_guard = Depends(require_role("ceo", "finance", "it"))
+
+SUGGESTED_TAX_TYPES = ("增值税", "附加税", "企业所得税", "印花税", "个人所得税", "房产税", "土地使用税")
+
+# CSV 全量列（文档口径）+ 必需列（列缺失才拒绝；可选列缺省即留空）
+TAX_HEADERS = ("period", "tax_type", "legal_deadline", "filed_date", "evidence_no", "registered_by", "remark")
+HR_HEADERS = ("period", "headcount_begin", "hires", "separations", "headcount_end", "registered_by", "remark")
+LICENCE_HEADERS = ("licence_name", "licence_no", "valid_from", "valid_to", "annual_check_status",
+                   "issuer", "owner_role", "attachment_no", "registered_by", "remark")
+TAX_REQUIRED = ("period", "tax_type")
+HR_REQUIRED = ("period", "headcount_begin", "hires", "separations", "headcount_end")
+LICENCE_REQUIRED = ("licence_name", "licence_no", "annual_check_status")
+
+
+# ============================================================
+# 公共工具：账套收敛 / 字段校验
+# ============================================================
+
+def _resolve_entity(request: Request, db: Session, explicit: Optional[int]) -> int:
+    """账套收敛：会话 token 绑定实体优先；显式 entity_id 与会话不一致 → 403（文案同 deps.get_entity_id）"""
+    session_entity = assert_entity_scope(request, explicit, db)
+    if explicit is not None:
+        return explicit
+    if session_entity is not None:
+        return session_entity
+    return 1
+
+
+def _body_entity(payload: dict, query_entity: Optional[int]) -> Optional[int]:
+    raw = payload.get("entity_id") if isinstance(payload, dict) else None
+    if raw in (None, ""):
+        return query_entity
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        raise HTTPException(400, f"entity_id 非法: {raw}")
+
+
+def _req_str(value, name: str, max_len: int) -> str:
+    text = "" if value is None else str(value).strip()
+    if not text:
+        raise HTTPException(400, f"缺少必填字段: {name}")
+    if len(text) > max_len:
+        raise HTTPException(400, f"字段 {name} 超长（{len(text)} > {max_len}）")
+    return text
+
+
+def _opt_str(value, max_len: int) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    return text[:max_len]
+
+
+def _parse_date(value, name: str) -> Optional[date]:
+    if value in (None, ""):
+        return None
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    text = str(value).strip().replace("/", "-")
+    for fmt in ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y%m%d"):
+        try:
+            return datetime.strptime(text, fmt).date()
+        except ValueError:
+            continue
+    raise HTTPException(400, f"字段 {name} 日期格式非法: {value}（应为 YYYY-MM-DD）")
+
+
+def _parse_int(value, name: str, required: bool = True) -> Optional[int]:
+    if value in (None, ""):
+        if required:
+            raise HTTPException(400, f"缺少必填字段: {name}")
+        return None
+    try:
+        return int(str(value).strip())
+    except (TypeError, ValueError):
+        raise HTTPException(400, f"字段 {name} 必须为整数: {value}")
+
+
+def _month_period(value, name: str = "period") -> str:
+    text = _req_str(value, name, 20)
+    if not re.match(MONTH_PERIOD_RE, text):
+        raise HTTPException(400, f"字段 {name} 期次格式非法: {text}（应为 YYYY-MM）")
+    return text
+
+
+def _target_period(kpi_code: str, period: str) -> str:
+    """指标实际落库期间：P_LICENCE_OK 为季频（传月期间时归到所在季度）"""
+    if kpi_code == LICENCE_OK_CODE and re.match(MONTH_PERIOD_RE, period):
+        return month_to_quarter(period)
+    return period
+
+
+# ============================================================
+# 字段选项（前端下拉动态加载用；禁前端硬编码）
+# ============================================================
+
+@router.get("/options")
+def get_options():
+    """录入表单选项：年检状态枚举 + 建议税种 + 期次格式"""
+    return {
+        "annual_check_status": list(ANNUAL_CHECK_STATUSES),
+        "tax_type_suggested": list(SUGGESTED_TAX_TYPES),
+        "period_format": {"monthly": "YYYY-MM 例:2026-08", "quarterly": "YYYY-Qn 例:2026-Q2"},
+        "derivable_kpi_codes": list(ALL_KPI_CODES),
+    }
+
+
+# ============================================================
+# 1. 申报行为台账 tax_filing_log
+# ============================================================
+
+def _tax_values(payload: dict) -> dict:
+    period = _month_period(payload.get("period"))
+    tax_type = _req_str(payload.get("tax_type"), "tax_type", 50)
+    legal_deadline = _parse_date(payload.get("legal_deadline"), "legal_deadline")
+    filed_date = _parse_date(payload.get("filed_date"), "filed_date")
+    return {
+        "period": period,
+        "tax_type": tax_type,
+        "legal_deadline": legal_deadline,
+        "filed_date": filed_date,
+        # on_time 唯一来源 = 服务端派生（请求里的同名字段一律忽略）
+        "on_time": derive_on_time(filed_date, legal_deadline),
+        "evidence_no": _opt_str(payload.get("evidence_no"), 100),
+        "registered_by": _opt_str(payload.get("registered_by"), 100),
+        "remark": _opt_str(payload.get("remark"), 500),
+    }
+
+
+def _tax_json(row: TaxFilingLog) -> dict:
+    return {
+        "id": row.id,
+        "entity_id": row.entity_id,
+        "period": row.period,
+        "tax_type": row.tax_type,
+        "legal_deadline": row.legal_deadline.isoformat() if row.legal_deadline else None,
+        "filed_date": row.filed_date.isoformat() if row.filed_date else None,
+        "on_time": row.on_time,
+        "evidence_no": row.evidence_no,
+        "registered_by": row.registered_by,
+        "remark": row.remark,
+    }
+
+
+@router.get("/tax-filings")
+def list_tax_filings(
+    request: Request,
+    period: Optional[str] = Query(None, description="属期 YYYY-MM"),
+    tax_type: Optional[str] = Query(None),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """按 (entity_id, period) 查询申报台账 + 及时率口径摘要"""
+    eid = _resolve_entity(request, db, entity_id)
+    q = db.query(TaxFilingLog).filter(TaxFilingLog.entity_id == eid)
+    if period:
+        q = q.filter(TaxFilingLog.period == period)
+    if tax_type:
+        q = q.filter(TaxFilingLog.tax_type == tax_type)
+    rows = q.order_by(TaxFilingLog.period.desc(), TaxFilingLog.id.asc()).all()
+    summary = explain_tax_timely_rate([r for r in rows if not period or r.period == period])
+    return {"entity_id": eid, "data": [_tax_json(r) for r in rows], "total": len(rows),
+            "summary": {"period": period, "tax_timely_rate": summary["value"],
+                        "denominator": summary["denominator"], "timely": summary["timely"],
+                        "no_definition": summary["no_definition"]}}
+
+
+@router.post("/tax-filings", status_code=201, dependencies=[write_guard])
+def create_tax_filing(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """新增一条申报行为台账（on_time 服务端派生；唯一键冲突 → 409）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    vals = _tax_values(payload)
+    exists = db.query(TaxFilingLog).filter(
+        TaxFilingLog.entity_id == eid, TaxFilingLog.period == vals["period"],
+        TaxFilingLog.tax_type == vals["tax_type"]).first()
+    if exists:
+        raise HTTPException(409, f"已存在同 (entity_id={eid}, period={vals['period']}, "
+                                 f"tax_type={vals['tax_type']}) 的登记行 id={exists.id}")
+    row = TaxFilingLog(entity_id=eid, **vals)
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return {"data": _tax_json(row),
+            "on_time_derived": row.on_time,
+            "on_time_ignored": payload.get("on_time") not in (None, "")}
+
+
+@router.post("/tax-filings/import", dependencies=[write_guard])
+def import_tax_filings(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """CSV 批量导入申报台账。on_conflict: skip(默认)/update；dry_run=true 只校验不落库"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    on_conflict = str(payload.get("on_conflict") or "skip").lower()
+    if on_conflict not in ("skip", "update"):
+        raise HTTPException(400, f"on_conflict 非法: {on_conflict}（skip/update）")
+    dry_run = bool(payload.get("dry_run"))
+    records, errors = _read_csv(payload.get("csv"), TAX_REQUIRED, _tax_values)
+    result = _apply_import(db, TaxFilingLog, eid, records, errors, on_conflict, dry_run,
+                           key_fields=("period", "tax_type"))
+    return result
+
+
+# ============================================================
+# 2. 人事月度台账 hr_monthly_ledger
+# ============================================================
+
+def _hr_values(payload: dict) -> dict:
+    period = _month_period(payload.get("period"))
+    begin = _parse_int(payload.get("headcount_begin"), "headcount_begin")
+    hires = _parse_int(payload.get("hires"), "hires")
+    separations = _parse_int(payload.get("separations"), "separations")
+    end = _parse_int(payload.get("headcount_end"), "headcount_end")
+    err = check_headcount_equation(begin, hires, separations, end)
+    if err:
+        raise HTTPException(400, err)
+    return {
+        "period": period,
+        "headcount_begin": begin,
+        "hires": hires,
+        "separations": separations,
+        "headcount_end": end,
+        "registered_by": _opt_str(payload.get("registered_by"), 100),
+        "remark": _opt_str(payload.get("remark"), 500),
+    }
+
+
+def _hr_json(row: HrMonthlyLedger) -> dict:
+    return {
+        "id": row.id, "entity_id": row.entity_id, "period": row.period,
+        "headcount_begin": row.headcount_begin, "hires": row.hires,
+        "separations": row.separations, "headcount_end": row.headcount_end,
+        "turnover_rate": compute_employee_turnover(row),
+        "registered_by": row.registered_by, "remark": row.remark,
+    }
+
+
+@router.get("/hr-monthly")
+def list_hr_monthly(
+    request: Request,
+    period: Optional[str] = Query(None, description="期间 YYYY-MM"),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """按 (entity_id, period) 查询人事月度台账（含流失率派生值）"""
+    eid = _resolve_entity(request, db, entity_id)
+    q = db.query(HrMonthlyLedger).filter(HrMonthlyLedger.entity_id == eid)
+    if period:
+        q = q.filter(HrMonthlyLedger.period == period)
+    rows = q.order_by(HrMonthlyLedger.period.desc(), HrMonthlyLedger.id.asc()).all()
+    return {"entity_id": eid, "data": [_hr_json(r) for r in rows], "total": len(rows)}
+
+
+@router.post("/hr-monthly", status_code=201, dependencies=[write_guard])
+def create_hr_monthly(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """新增人事月度台账（等式校验不通过 → 400 拒绝写入）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    vals = _hr_values(payload)
+    exists = db.query(HrMonthlyLedger).filter(
+        HrMonthlyLedger.entity_id == eid, HrMonthlyLedger.period == vals["period"]).first()
+    if exists:
+        raise HTTPException(409, f"已存在同 (entity_id={eid}, period={vals['period']}) 的登记行 id={exists.id}")
+    row = HrMonthlyLedger(entity_id=eid, **vals)
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return {"data": _hr_json(row)}
+
+
+@router.post("/hr-monthly/import", dependencies=[write_guard])
+def import_hr_monthly(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """CSV 批量导入人事月度台账（等式不成立的行被拒绝并在 errors 回报）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    on_conflict = str(payload.get("on_conflict") or "skip").lower()
+    if on_conflict not in ("skip", "update"):
+        raise HTTPException(400, f"on_conflict 非法: {on_conflict}（skip/update）")
+    dry_run = bool(payload.get("dry_run"))
+    records, errors = _read_csv(payload.get("csv"), HR_REQUIRED, _hr_values)
+    return _apply_import(db, HrMonthlyLedger, eid, records, errors, on_conflict, dry_run,
+                         key_fields=("period",))
+
+
+# ============================================================
+# 3. 证照台账 licence_ledger
+# ============================================================
+
+def _licence_values(payload: dict) -> dict:
+    name = _req_str(payload.get("licence_name"), "licence_name", 200)
+    no = _req_str(payload.get("licence_no"), "licence_no", 100)
+    status = _req_str(payload.get("annual_check_status"), "annual_check_status", 20)
+    if status not in ANNUAL_CHECK_STATUSES:
+        raise HTTPException(400, f"字段 annual_check_status 非法: {status}（合法值: {'/'.join(ANNUAL_CHECK_STATUSES)}）")
+    return {
+        "licence_name": name,
+        "licence_no": no,
+        "valid_from": _parse_date(payload.get("valid_from"), "valid_from"),
+        "valid_to": _parse_date(payload.get("valid_to"), "valid_to"),
+        "annual_check_status": status,
+        "issuer": _opt_str(payload.get("issuer"), 200),
+        "owner_role": _opt_str(payload.get("owner_role"), 100),
+        "attachment_no": _opt_str(payload.get("attachment_no"), 100),
+        "registered_by": _opt_str(payload.get("registered_by"), 100),
+        "remark": _opt_str(payload.get("remark"), 500),
+    }
+
+
+def _licence_json(row: LicenceLedger, as_of: Optional[date] = None) -> dict:
+    return {
+        "id": row.id, "entity_id": row.entity_id, "licence_name": row.licence_name,
+        "licence_no": row.licence_no,
+        "valid_from": row.valid_from.isoformat() if row.valid_from else None,
+        "valid_to": row.valid_to.isoformat() if row.valid_to else None,
+        "annual_check_status": row.annual_check_status, "issuer": row.issuer,
+        "owner_role": row.owner_role, "attachment_no": row.attachment_no,
+        "valid_as_of": is_licence_valid(row, as_of) if as_of else None,
+        "registered_by": row.registered_by, "remark": row.remark,
+    }
+
+
+@router.get("/licences")
+def list_licences(
+    request: Request,
+    as_of: Optional[str] = Query(None, description="有效性基准日 YYYY-MM-DD（默认今天）"),
+    licence_name: Optional[str] = Query(None),
+    valid_only: bool = Query(False, description="仅返回基准日有效的证照"),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """证照台账查询（有效/过期按基准日判定）"""
+    eid = _resolve_entity(request, db, entity_id)
+    base = _parse_date(as_of, "as_of") or date.today()
+    q = db.query(LicenceLedger).filter(LicenceLedger.entity_id == eid)
+    if licence_name:
+        q = q.filter(LicenceLedger.licence_name == licence_name)
+    rows = q.order_by(LicenceLedger.id.asc()).all()
+    # 合规率分母 = 应持证照数（不受 valid_only 过滤影响）
+    summary = explain_licence_ok(rows, base)
+    shown = [r for r in rows if is_licence_valid(r, base)] if valid_only else rows
+    return {"entity_id": eid, "as_of": base.isoformat(),
+            "data": [_licence_json(r, base) for r in shown], "total": len(shown),
+            "summary": {"licence_ok": summary["value"], "denominator": summary["denominator"],
+                        "valid": summary["valid"], "no_definition": summary["no_definition"]}}
+
+
+@router.post("/licences", status_code=201, dependencies=[write_guard])
+def create_licence(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """新增证照台账（状态白名单校验；唯一键冲突 → 409）"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    vals = _licence_values(payload)
+    exists = db.query(LicenceLedger).filter(
+        LicenceLedger.entity_id == eid, LicenceLedger.licence_name == vals["licence_name"],
+        LicenceLedger.licence_no == vals["licence_no"]).first()
+    if exists:
+        raise HTTPException(409, f"已存在同 (entity_id={eid}, licence_name={vals['licence_name']}, "
+                                 f"licence_no={vals['licence_no']}) 的登记行 id={exists.id}")
+    row = LicenceLedger(entity_id=eid, **vals)
+    db.add(row)
+    db.commit()
+    db.refresh(row)
+    return {"data": _licence_json(row, date.today())}
+
+
+@router.post("/licences/import", dependencies=[write_guard])
+def import_licences(
+    payload: dict,
+    request: Request,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """CSV 批量导入证照台账"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    on_conflict = str(payload.get("on_conflict") or "skip").lower()
+    if on_conflict not in ("skip", "update"):
+        raise HTTPException(400, f"on_conflict 非法: {on_conflict}（skip/update）")
+    dry_run = bool(payload.get("dry_run"))
+    records, errors = _read_csv(payload.get("csv"), LICENCE_REQUIRED, _licence_values)
+    return _apply_import(db, LicenceLedger, eid, records, errors, on_conflict, dry_run,
+                         key_fields=("licence_name", "licence_no"))
+
+
+# ============================================================
+# 4. 台账 → kpi_values 派生
+# ============================================================
+
+@router.post("/derive", dependencies=[write_guard])
+def derive_kpis(
+    request: Request,
+    payload: dict,
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """触发台账→kpi_values 派生（幂等）。period: 月频 YYYY-MM；P_LICENCE_OK 接受 YYYY-Qn。
+    dry_run=true 只算不写。分母为 0 的指标 → 不写值，回报 skipped_denominator_zero。"""
+    eid = _resolve_entity(request, db, _body_entity(payload, entity_id))
+    period = str(payload.get("period") or "").strip()
+    if not period:
+        raise HTTPException(400, "缺少必填字段: period")
+    if not (re.match(MONTH_PERIOD_RE, period) or re.match(QUARTER_PERIOD_RE, period)):
+        raise HTTPException(400, f"period 非法: {period}（应为 YYYY-MM 或 YYYY-Qn）")
+    kpi_codes = payload.get("kpi_codes") or payload.get("kpis") or None
+    if kpi_codes is not None and not isinstance(kpi_codes, list):
+        raise HTTPException(400, "kpi_codes 必须为数组")
+    try:
+        result = run_derive(db, eid, period, kpis=kpi_codes, write=not bool(payload.get("dry_run")))
+    except ValueError as e:
+        raise HTTPException(400, str(e))
+    return result
+
+
+@router.get("/derive/status")
+def derive_status(
+    request: Request,
+    period: str = Query(..., description="月期间 YYYY-MM 或季期间 YYYY-Qn"),
+    entity_id: Optional[int] = Query(None, ge=1),
+    db: Session = Depends(get_db),
+):
+    """只读：查看该期派生指标的定义/现有 kpi_values 行（不写入）"""
+    eid = _resolve_entity(request, db, entity_id)
+    if not (re.match(MONTH_PERIOD_RE, period) or re.match(QUARTER_PERIOD_RE, period)):
+        raise HTTPException(400, f"period 非法: {period}（应为 YYYY-MM 或 YYYY-Qn）")
+    out = {}
+    from app.models import KPIValue
+    for code in ALL_KPI_CODES:
+        kpi = (db.query(KPIDefinition)
+               .filter(KPIDefinition.entity_id == eid, KPIDefinition.kpi_code == code,
+                       KPIDefinition.status == "active")
+               .order_by(KPIDefinition.id.asc()).first())
+        item = {"kpi_defined": kpi is not None, "kpi_id": kpi.id if kpi else None,
+                "target_period": _target_period(code, period)}
+        if kpi is not None:
+            row = (db.query(KPIValue)
+                   .filter(KPIValue.kpi_id == kpi.id, KPIValue.source_type == "derived",
+                           KPIValue.period == item["target_period"])
+                   .order_by(KPIValue.id.desc()).first())
+            item["latest_derived"] = ({"period": row.period, "actual_value": row.actual_value,
+                                       "data_status": row.data_status} if row else None)
+        out[code] = item
+    return {"entity_id": eid, "period": period, "items": out}
+
+
+# ============================================================
+# CSV 解析 / 批量落库（三张表共用；校验规则与单条新增完全同源）
+# ============================================================
+
+def _read_csv(csv_text, required_headers, builder):
+    """解析 CSV：返回 (records, errors)。required_headers 为必需列（缺列即 400）；其余列可选。"""
+    if not csv_text or not str(csv_text).strip():
+        raise HTTPException(400, "缺少必填字段: csv")
+    text = str(csv_text).lstrip("\ufeff")
+    reader = csv.DictReader(io.StringIO(text))
+    fieldnames = [(f or "").strip().lower() for f in (reader.fieldnames or [])]
+    if not fieldnames:
+        raise HTTPException(400, "CSV 缺少表头行")
+    missing = [h for h in required_headers if h not in fieldnames]
+    if missing:
+        raise HTTPException(400, f"CSV 缺少必需列: {missing}（必需列: {list(required_headers)}）")
+    records, errors = [], []
+    for offset, raw in enumerate(reader, start=2):  # 表头占第 1 行
+        row = {(k or "").strip().lower(): v for k, v in raw.items() if k}
+        if not any(str(v or "").strip() for v in row.values()):
+            continue  # 跳过空行
+        try:
+            records.append({"line": offset, "values": builder(row)})
+        except HTTPException as e:
+            errors.append({"line": offset, "reason": e.detail})
+    return records, errors
+
+
+def _apply_import(db: Session, model, entity_id: int, records, errors, on_conflict: str,
+                  dry_run: bool, key_fields) -> dict:
+    """逐行 upsert；返回 {inserted, updated, skipped, errors, dry_run}"""
+    inserted = updated = skipped = 0
+    for rec in records:
+        vals = rec["values"]
+        filters = [getattr(model, f) == vals[f] for f in key_fields] + [model.entity_id == entity_id]
+        existing = db.query(model).filter(*filters).first()
+        if existing is not None:
+            if on_conflict == "update":
+                for k, v in vals.items():
+                    setattr(existing, k, v)
+                updated += 1
+            else:
+                skipped += 1
+        else:
+            db.add(model(entity_id=entity_id, **vals))
+            inserted += 1
+    if dry_run:
+        db.rollback()
+    else:
+        db.commit()
+    return {"entity_id": entity_id, "dry_run": dry_run, "on_conflict": on_conflict,
+            "inserted": inserted, "updated": updated, "skipped": skipped,
+            "errors": errors, "error_count": len(errors)}
diff --git a/backend/app/main.py b/backend/app/main.py
index 647666c6..26b36f1f 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -7,7 +7,7 @@ from fastapi.responses import JSONResponse
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from dotenv import load_dotenv
 from app.database import init_db, get_db
-from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json
+from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers
 from app.utils.cache import clear_all as clear_cache, delete as delete_cache
 from scripts.erp_sync import run_sync as run_erp_sync
 from app.auth_middleware import require_auth, require_auth_401
@@ -98,6 +98,7 @@ app.include_router(cash_vouchers_json.router)
 app.include_router(data_classification.router)
 app.include_router(tax_compliance.router)
 app.include_router(verify.router)
+app.include_router(ext_input_ledgers.router)  # 外部输入台账（M2 准入3 派生，2026-09-15）
 
 @app.exception_handler(Exception)
 async def global_exception_handler(request: Request, exc: Exception):
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 2ae82a9e..889dcdf0 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -7,6 +7,7 @@ from app.models.cost_model import StandardCost, ActualCost, AbcActivity, AbcAllo
 from app.models.knowledge import KnowledgeEvent, KnowledgeSummary
 from app.models.driver_budget import DriverFactorTemplate, DriverFactorBudget
 from app.models.product_sales import ProductSales
+from app.models.ext_input_ledger import TaxFilingLog, HrMonthlyLedger, LicenceLedger
 
 
 class Entity(Base):
diff --git a/backend/app/models/ext_input_ledger.py b/backend/app/models/ext_input_ledger.py
new file mode 100644
index 00000000..4ca72b84
--- /dev/null
+++ b/backend/app/models/ext_input_ledger.py
@@ -0,0 +1,101 @@
+"""外部输入台账 — 三张**外部输入**业务表（能力评价 M2 准入3 派生，2026-09-15）
+
+背景：能力评价 6 项缺口值经只读实测确认**无一项**能从既有系统合法取证；其中 3 项
+只能来自新的人工登记台账（PRD v1.3 §4.0）。本模块只做「建表 + 结构化字段」，
+录入/派生分别见 app/api/ext_input_ledgers.py 与 app/services/ext_input_derive.py。
+
+    tax_filing_log      申报行为台账  → P_TAX_TIMELY      (A4 税务申报及时率, 月频)
+    hr_monthly_ledger   人事月度台账  → L_EMPLOYEE_TURNOVER (B5 员工流失率, 月频)
+    licence_ledger      证照台账      → P_LICENCE_OK      (B2 经营许可合规率, 季频)
+
+纪律（任务 0-P0-project-20260915-cma-eval-m2-ext-input-tables）：
+- 本模块**只新增表**，不改既有表、不改任何 KPI 定义（kpi_definitions 属准入2/4 同表串行）
+- `on_time` 是**服务端派生**字段（唯一实现 app/services/ext_input_derive.derive_on_time），
+  录入方手填值一律被忽略并回报 on_time_ignored
+- 每表 entity_id NOT NULL + 唯一键以 entity_id 打头（多租户隔离 + 索引可用性）
+"""
+from sqlalchemy import Column, Integer, String, Date, DateTime, UniqueConstraint, func
+from app.database import Base
+
+
+# 证照年检状态枚举（PRD v1.3 §4.0 原文口径，录入时白名单校验）
+ANNUAL_CHECK_STATUSES = ("未到期", "已年检", "待年检", "过期")
+
+# 期次格式（period）
+MONTH_PERIOD_RE = r"^\d{4}-(0[1-9]|1[0-2])$"   # 2026-08
+QUARTER_PERIOD_RE = r"^\d{4}-Q[1-4]$"          # 2026-Q2
+
+
+class TaxFilingLog(Base):
+    """申报行为台账 — P_TAX_TIMELY 的唯一取证来源
+
+    唯一键 (entity_id, period, tax_type)：同一账套同一属期同一税种只允许一条登记行。
+    """
+    __tablename__ = "tax_filing_log"
+    id = Column(Integer, primary_key=True, index=True)
+    entity_id = Column(Integer, nullable=False, index=True, comment="企业ID（账套）")
+    period = Column(String(20), nullable=False, comment="属期 YYYY-MM（申报所属月份）")
+    tax_type = Column(String(50), nullable=False, comment="税种: 增值税/附加税/企业所得税/印花税…")
+    legal_deadline = Column(Date, nullable=True, comment="法定申报期限")
+    filed_date = Column(Date, nullable=True, comment="实际申报日（未申报为 NULL）")
+    on_time = Column(Integer, nullable=True, comment="是否按时申报 1/0（**服务端派生**，录入方不得手填）")
+    evidence_no = Column(String(100), nullable=True, comment="申报回执/截图编号")
+    registered_by = Column(String(100), nullable=True, comment="登记人（职能岗位）")
+    remark = Column(String(500), nullable=True, comment="备注")
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+    __table_args__ = (
+        UniqueConstraint("entity_id", "period", "tax_type", name="uq_tax_filing_log_entity_period_type"),
+    )
+
+
+class HrMonthlyLedger(Base):
+    """人事月度台账 — L_EMPLOYEE_TURNOVER 的唯一取证来源
+
+    唯一键 (entity_id, period)。
+    硬校验：headcount_end == headcount_begin + hires - separations，
+    不通过时**拒绝写入并回报**（见 app/services/ext_input_derive.check_headcount_equation）。
+    """
+    __tablename__ = "hr_monthly_ledger"
+    id = Column(Integer, primary_key=True, index=True)
+    entity_id = Column(Integer, nullable=False, index=True, comment="企业ID（账套）")
+    period = Column(String(20), nullable=False, comment="期间 YYYY-MM")
+    headcount_begin = Column(Integer, nullable=False, comment="期初在职人数")
+    hires = Column(Integer, nullable=False, default=0, comment="入职人数")
+    separations = Column(Integer, nullable=False, default=0, comment="离职人数")
+    headcount_end = Column(Integer, nullable=False, comment="期末在职人数（须= 期初+入职-离职）")
+    registered_by = Column(String(100), nullable=True, comment="登记人（职能岗位）")
+    remark = Column(String(500), nullable=True, comment="备注")
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+    __table_args__ = (
+        UniqueConstraint("entity_id", "period", name="uq_hr_monthly_ledger_entity_period"),
+    )
+
+
+class LicenceLedger(Base):
+    """证照台账 — P_LICENCE_OK 的唯一取证来源
+
+    唯一键 (entity_id, licence_name, licence_no)。
+    有效性口径（派生侧）：valid_to >= 季末 且 年检状态 != '过期'；
+    valid_to 为空 = 长期有效（仍受状态约束）。
+    """
+    __tablename__ = "licence_ledger"
+    id = Column(Integer, primary_key=True, index=True)
+    entity_id = Column(Integer, nullable=False, index=True, comment="企业ID（账套）")
+    licence_name = Column(String(200), nullable=False, comment="证照名称")
+    licence_no = Column(String(100), nullable=False, comment="证照编号")
+    valid_from = Column(Date, nullable=True, comment="有效期起")
+    valid_to = Column(Date, nullable=True, comment="有效期止（空=长期有效）")
+    annual_check_status = Column(String(20), nullable=False, default="未到期",
+                                 comment="年检状态: 未到期/已年检/待年检/过期")
+    issuer = Column(String(200), nullable=True, comment="发证/主管机关")
+    owner_role = Column(String(100), nullable=True, comment="责任岗位（职能，不指定自然人）")
+    attachment_no = Column(String(100), nullable=True, comment="附件/扫描件编号")
+    registered_by = Column(String(100), nullable=True, comment="登记人（职能岗位）")
+    remark = Column(String(500), nullable=True, comment="备注")
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
+    __table_args__ = (
+        UniqueConstraint("entity_id", "licence_name", "licence_no", name="uq_licence_ledger_entity_name_no"),
+    )
diff --git a/backend/app/services/ext_input_derive.py b/backend/app/services/ext_input_derive.py
new file mode 100644
index 00000000..4db7de28
--- /dev/null
+++ b/backend/app/services/ext_input_derive.py
@@ -0,0 +1,349 @@
+"""外部输入台账 → kpi_values 派生模块（能力评价 M2 准入3 派生，2026-09-15）
+
+三张外部输入台账（app/models/ext_input_ledger.py）→ 三个能力评价指标值：
+
+    P_TAX_TIMELY          = 当期 on_time=1 的税种数 ÷ 当期应申报税种数 × 100   （月频）
+    L_EMPLOYEE_TURNOVER   = separations ÷ headcount_begin × 100                （月频）
+    P_LICENCE_OK          = 有效证照数 ÷ 应持证照数 × 100                        （季频，季末为基准日）
+
+硬规则（任务 0-P0-project-20260915-cma-eval-m2-ext-input-tables，直接进验收）：
+1. **分母为 0 必须返回 None（无定义 / 不参评），不得写 0、不得填大数。**
+   依据：F_INTEREST_COVER 已因「利息费用=0 → 分母 0」失去定义；同族错误不得在 3 张新台账上重演。
+   → 值为 None 时**不写 kpi_values**；若该期已存在旧的 derived 行则只回报 stale_row_present，不删不改。
+2. **只写 kpi_values**（entity_id / kpi_id / period / actual_value + 来源标识 source_type='derived'），
+   **不新建、不改动任何 KPI 定义**：定义缺失 → 跳过并回报 skipped_no_definition。
+3. **幂等**：同一 (entity_id, kpi_id, period, source_type='derived') 重复运行只更新不新增
+   （与 scripts/kpi_bep_deriver.py 同范式）。
+
+本模块只依赖「台账行对象」的属性（ORM 实例 / 简易对象皆可），纯计算函数可脱离数据库单测。
+"""
+from __future__ import annotations
+
+import calendar
+import logging
+import re
+from datetime import date, datetime
+from typing import Iterable, Optional, Sequence
+
+from sqlalchemy.orm import Session
+
+from app.models import KPIDefinition, KPIValue, TaxFilingLog, HrMonthlyLedger, LicenceLedger
+
+logger = logging.getLogger("cma.ext_input_derive")
+
+TAX_TIMELY_CODE = "P_TAX_TIMELY"
+EMPLOYEE_TURNOVER_CODE = "L_EMPLOYEE_TURNOVER"
+LICENCE_OK_CODE = "P_LICENCE_OK"
+
+ALL_KPI_CODES = (TAX_TIMELY_CODE, EMPLOYEE_TURNOVER_CODE, LICENCE_OK_CODE)
+
+# 数据来源标识（写入 kpi_values.source_type；与 dashboard「最新派生期间」取数口径一致）
+SOURCE_TYPE = "derived"
+
+_MONTH_RE = re.compile(r"^(\d{4})-(0[1-9]|1[0-2])$")
+_QUARTER_RE = re.compile(r"^(\d{4})-Q([1-4])$")
+
+
+# ============================================================
+# 期间工具
+# ============================================================
+
+def is_month_period(period: str) -> bool:
+    return bool(_MONTH_RE.match(str(period or "")))
+
+
+def is_quarter_period(period: str) -> bool:
+    return bool(_QUARTER_RE.match(str(period or "")))
+
+
+def month_to_quarter(period: str) -> str:
+    """2026-06 → 2026-Q2（季频指标按「所在季度」归集）"""
+    m = _MONTH_RE.match(str(period or ""))
+    if not m:
+        raise ValueError(f"非法月期间: {period}（应为 YYYY-MM）")
+    quarter = (int(m.group(2)) - 1) // 3 + 1
+    return f"{m.group(1)}-Q{quarter}"
+
+
+def quarter_end_date(period: str) -> date:
+    """季末基准日。接受 2026-Q2（季频）或 2026-06（月频，自动归到所在季度）"""
+    q = _QUARTER_RE.match(str(period or ""))
+    if q:
+        year, quarter = int(q.group(1)), int(q.group(2))
+    else:
+        q2 = re.match(r"^(\d{4})-Q([1-4])$", month_to_quarter(period))
+        if not q2:
+            raise ValueError(f"非法季期间: {period}（应为 YYYY-Qn 或 YYYY-MM）")
+        year, quarter = int(q2.group(1)), int(q2.group(2))
+    last_month = quarter * 3
+    return date(year, last_month, calendar.monthrange(year, last_month)[1])
+
+
+# ============================================================
+# 1. 申报行为台账 → P_TAX_TIMELY
+# ============================================================
+
+def derive_on_time(filed_date: Optional[date], legal_deadline: Optional[date]) -> Optional[int]:
+    """`on_time` 的**唯一**服务端派生规则（录入方不得手填）：
+
+    - 已申报且不晚于法定期限        → 1
+    - 已申报但晚于法定期限          → 0
+    - 未申报（filed_date 为空）     → 0（逾期未申报，计入分母）
+    - 已申报但缺法定期限（无法判定）→ None（该行不计入分子；分母仍计）
+    """
+    if filed_date is None:
+        return 0
+    if legal_deadline is None:
+        return None
+    return 1 if filed_date <= legal_deadline else 0
+
+
+def compute_tax_timely_rate(rows: Iterable) -> Optional[float]:
+    """P_TAX_TIMELY = on_time=1 的税种数 ÷ 应申报税种数 × 100；分母为 0 → None"""
+    rows = list(rows)
+    denominator = len(rows)
+    if denominator == 0:
+        return None
+    timely = sum(1 for r in rows if getattr(r, "on_time", None) == 1)
+    return round(timely / denominator * 100.0, 1)
+
+
+def explain_tax_timely_rate(rows: Iterable) -> dict:
+    rows = list(rows)
+    value = compute_tax_timely_rate(rows)
+    return {
+        "denominator": len(rows),
+        "timely": sum(1 for r in rows if getattr(r, "on_time", None) == 1),
+        "unknown": sum(1 for r in rows if getattr(r, "on_time", None) is None),
+        "value": value,
+        "no_definition": value is None,
+    }
+
+
+# ============================================================
+# 2. 人事月度台账 → L_EMPLOYEE_TURNOVER
+# ============================================================
+
+def check_headcount_equation(headcount_begin, hires, separations, headcount_end) -> Optional[str]:
+    """期末 = 期初 + 入职 − 离职。通过返回 None，不通过返回错误说明（调用方拒绝写入并回报）"""
+    try:
+        lhs = int(headcount_end)
+        rhs = int(headcount_begin) + int(hires) - int(separations)
+    except (TypeError, ValueError):
+        return "期初/入职/离职/期末必须为整数"
+    if lhs != rhs:
+        return (f"人事台账等式不成立: 期末({lhs}) ≠ 期初({headcount_begin}) + 入职({hires}) "
+                f"− 离职({separations}) = {rhs}（拒绝写入）")
+    return None
+
+
+def compute_employee_turnover(row) -> Optional[float]:
+    """L_EMPLOYEE_TURNOVER = separations ÷ headcount_begin × 100；期初为 0（或缺失）→ None"""
+    if row is None:
+        return None
+    try:
+        begin = int(getattr(row, "headcount_begin"))
+        separations = int(getattr(row, "separations"))
+    except (TypeError, ValueError):
+        return None
+    if begin == 0:
+        return None
+    return round(separations / begin * 100.0, 2)
+
+
+def explain_employee_turnover(row) -> dict:
+    """并列返回「期初期末平均在职」口径值，便于复核（平均在职为 0 → 该口径同样 None）"""
+    value = compute_employee_turnover(row)
+    detail = {
+        "headcount_begin": getattr(row, "headcount_begin", None),
+        "headcount_end": getattr(row, "headcount_end", None),
+        "hires": getattr(row, "hires", None),
+        "separations": getattr(row, "separations", None),
+        "value": value,
+        "no_definition": value is None,
+        "avg_headcount": None,
+        "value_on_avg": None,
+    }
+    if row is None:
+        return detail
+    try:
+        avg = (int(row.headcount_begin) + int(row.headcount_end)) / 2.0
+    except (TypeError, ValueError):
+        avg = None
+    detail["avg_headcount"] = avg
+    if avg:
+        detail["value_on_avg"] = round(int(row.separations) / avg * 100.0, 2)
+    return detail
+
+
+# ============================================================
+# 3. 证照台账 → P_LICENCE_OK
+# ============================================================
+
+def is_licence_valid(licence, as_of: date) -> bool:
+    """有效证照判定：状态非「过期」且（valid_to 为空=长期有效 或 valid_to >= 基准日）"""
+    if str(getattr(licence, "annual_check_status", "") or "").strip() == "过期":
+        return False
+    valid_to = getattr(licence, "valid_to", None)
+    if valid_to is None:
+        return True
+    if isinstance(valid_to, datetime):
+        valid_to = valid_to.date()
+    return valid_to >= as_of
+
+
+def compute_licence_ok(rows: Iterable, as_of: date) -> Optional[float]:
+    """P_LICENCE_OK = 有效证照数 ÷ 应持证照数 × 100（基准日=季末）；应持证照数为 0 → None"""
+    rows = list(rows)
+    denominator = len(rows)
+    if denominator == 0:
+        return None
+    valid = sum(1 for r in rows if is_licence_valid(r, as_of))
+    return round(valid / denominator * 100.0, 1)
+
+
+def explain_licence_ok(rows: Iterable, as_of: date) -> dict:
+    rows = list(rows)
+    value = compute_licence_ok(rows, as_of)
+    expired = [getattr(r, "licence_no", "") for r in rows if str(getattr(r, "annual_check_status", "")) == "过期"]
+    return {
+        "as_of": as_of.isoformat(),
+        "denominator": len(rows),
+        "valid": sum(1 for r in rows if is_licence_valid(r, as_of)),
+        "expired_or_outdated": expired,
+        "value": value,
+        "no_definition": value is None,
+    }
+
+
+# ============================================================
+# 落库（幂等 upsert；只写 kpi_values）
+# ============================================================
+
+def _resolve_kpi(db: Session, entity_id: int, kpi_code: str) -> Optional[KPIDefinition]:
+    """按 (entity_id, kpi_code) 取定义（仅 active）。定义唯一性在库内非全局：
+    e1/e2 可各有一条同名 kpi_code，故必须带 entity_id 过滤 + order_by(id) 取稳定行。"""
+    return (db.query(KPIDefinition)
+            .filter(KPIDefinition.entity_id == entity_id,
+                    KPIDefinition.kpi_code == kpi_code,
+                    KPIDefinition.status == "active")
+            .order_by(KPIDefinition.id.asc())
+            .first())
+
+
+def upsert_kpi_value(db: Session, entity_id: int, kpi: KPIDefinition, period: str,
+                     value: float, remark: str, batch: str) -> str:
+    """幂等写 kpi_values：命中 (kpi_id, period, source_type='derived') 则更新，返回 'insert'/'upsert'"""
+    existing = (db.query(KPIValue)
+                .filter(KPIValue.kpi_id == kpi.id,
+                        KPIValue.period == period,
+                        KPIValue.source_type == SOURCE_TYPE)
+                .order_by(KPIValue.id.asc())
+                .first())
+    if existing is not None:
+        action = "upsert"
+        existing.actual_value = value
+        existing.source_batch = batch
+        existing.data_status = "pending"
+        existing.remark = remark
+    else:
+        db.add(KPIValue(
+            entity_id=entity_id,
+            kpi_id=kpi.id,
+            period=period,
+            actual_value=value,
+            source_type=SOURCE_TYPE,
+            source_batch=batch,
+            data_status="pending",
+            remark=remark,
+        ))
+        action = "insert"
+    db.flush()
+    return action
+
+
+def run_derive(db: Session, entity_id: int, period: str,
+               kpis: Optional[Sequence[str]] = None, write: bool = True) -> dict:
+    """按实体+期间派生 3 个指标并写 kpi_values（幂等）。
+
+    period：P_TAX_TIMELY / L_EMPLOYEE_TURNOVER 用 YYYY-MM；
+            P_LICENCE_OK 用 YYYY-Qn（传 YYYY-MM 时自动归到所在季度，落库 period 为季度）。
+    write=False → 只算不写（dry-run）。
+    返回 {'entity_id', 'period', 'batch', 'items': {kpi_code: {...}}}
+    """
+    codes = list(kpis or ALL_KPI_CODES)
+    unknown = [c for c in codes if c not in ALL_KPI_CODES]
+    if unknown:
+        raise ValueError(f"不支持的指标编码: {unknown}（可用: {list(ALL_KPI_CODES)}）")
+    batch = f"ext-ledger-{datetime.now().strftime('%Y%m%d%H%M%S')}"
+    items: dict = {}
+
+    for code in codes:
+        kpi = _resolve_kpi(db, entity_id, code)
+        if kpi is None:
+            items[code] = {
+                "action": "skipped_no_definition",
+                "reason": f"entity={entity_id} 无 active 定义 {code}（定义变更属另一派单，本模块不新建）",
+                "actual_value": None,
+            }
+            continue
+
+        if code == TAX_TIMELY_CODE:
+            rows = (db.query(TaxFilingLog)
+                    .filter(TaxFilingLog.entity_id == entity_id, TaxFilingLog.period == period)
+                    .order_by(TaxFilingLog.id.asc()).all())
+            detail = explain_tax_timely_rate(rows)
+            target_period = period
+            remark = (f"台账派生: 应申报{detail['denominator']}个税种，按时{detail['timely']}个，"
+                      f"无法判定{detail['unknown']}个（tax_filing_log {period}）")
+        elif code == EMPLOYEE_TURNOVER_CODE:
+            row = (db.query(HrMonthlyLedger)
+                   .filter(HrMonthlyLedger.entity_id == entity_id, HrMonthlyLedger.period == period)
+                   .order_by(HrMonthlyLedger.id.asc()).first())
+            detail = explain_employee_turnover(row)
+            target_period = period
+            remark = (f"台账派生: 离职{detail['separations']}/期初在职{detail['headcount_begin']}×100 "
+                      f"(期初{detail['headcount_begin']}+入职{detail['hires']}-离职{detail['separations']}"
+                      f"=期末{detail['headcount_end']})；并列口径 期初期末平均在职{detail['avg_headcount']}人 "
+                      f"→ {detail['value_on_avg']}%")
+        else:  # LICENCE_OK —— 季频
+            target_period = month_to_quarter(period) if is_month_period(period) else period
+            as_of = quarter_end_date(target_period)
+            rows = (db.query(LicenceLedger)
+                    .filter(LicenceLedger.entity_id == entity_id)
+                    .order_by(LicenceLedger.id.asc()).all())
+            detail = explain_licence_ok(rows, as_of)
+            remark = (f"台账派生: 季末基准日{detail['as_of']}，有效{detail['valid']}/应持证{detail['denominator']}"
+                      f"（过期或失效: {detail['expired_or_outdated'] or '无'}）")
+
+        value = detail.get("value")
+        item = {
+            "kpi_id": kpi.id,
+            "kpi_code": code,
+            "period": target_period,
+            "actual_value": value,
+            "detail": detail,
+        }
+        if value is None:
+            # 硬规则：分母 0 → 无定义，不写 kpi_values（不写 0、不填大数）
+            stale = (db.query(KPIValue)
+                     .filter(KPIValue.kpi_id == kpi.id, KPIValue.period == target_period,
+                             KPIValue.source_type == SOURCE_TYPE)
+                     .order_by(KPIValue.id.asc()).first())
+            item["action"] = "skipped_denominator_zero"
+            item["reason"] = "分母为 0 → 无定义（不参评），不写 kpi_values"
+            item["stale_row_present"] = stale is not None
+            items[code] = item
+            continue
+
+        if write:
+            item["action"] = upsert_kpi_value(db, entity_id, kpi, target_period, value, remark, batch)
+        else:
+            item["action"] = "dry-run"
+        item["remark"] = remark
+        items[code] = item
+
+    if write:
+        db.commit()
+    return {"entity_id": entity_id, "period": period, "batch": batch,
+            "write": write, "items": items}
diff --git a/backend/scripts/migrate_ext_input_ledgers_20260915.py b/backend/scripts/migrate_ext_input_ledgers_20260915.py
new file mode 100644
index 00000000..082a07e9
--- /dev/null
+++ b/backend/scripts/migrate_ext_input_ledgers_20260915.py
@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+"""外部输入台账 DDL 迁移 — tax_filing_log / hr_monthly_ledger / licence_ledger
+（任务 0-P0-project-20260915-cma-eval-m2-ext-input-tables，2026-09-15）
+
+DDL 真源 = scripts/sql/ext_input_ledgers_20260915.sql（单一来源，本脚本读取执行，勿复制 DDL）
+
+动作：
+  --check        只读预检（默认）：三表存在性 + 缺列 + 唯一键，零写入
+  --apply        执行 CREATE TABLE IF NOT EXISTS（幂等：二次运行新增=0）
+  --down         回滚：DROP 三表（表内已有数据须显式 --yes）
+  --show-create  打印 SHOW CREATE TABLE 原文（验收证据；非 MySQL 目标自动跳过）
+  --url URL      目标库 SQLAlchemy URL 覆盖（默认 app.database.get_engine() = backend/.env）
+
+生产保护（生产资产派单纪律 + 本单「不得直连生产写入」）：
+  目标库名 == 'cma' 时，--apply/--down **一律拒绝**；须 --allow-prod 且环境变量
+  CMA_DDL_APPROVED=1（= 已拿到审批单）才放行。--check/--show-create 为只读，不受限但会打标。
+
+用法：
+  venv/bin/python3 scripts/migrate_ext_input_ledgers_20260915.py --check
+  venv/bin/python3 scripts/migrate_ext_input_ledgers_20260915.py --apply \
+      --url "mysql+pymysql://root@localhost/cma_scratch_x?unix_socket=/var/run/mysqld/mysqld.sock&charset=utf8mb4"
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sqlalchemy import create_engine, inspect, text  # noqa: E402
+
+TABLES = ("tax_filing_log", "hr_monthly_ledger", "licence_ledger")
+SQL_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sql",
+                        "ext_input_ledgers_20260915.sql")
+PROD_DB_NAMES = ("cma",)
+
+
+def load_statements() -> list:
+    """读取 .sql 真源 → 逐条语句（去注释/空行），返回可执行 DDL 列表"""
+    with open(SQL_FILE, "r", encoding="utf-8") as f:
+        raw = f.read()
+    lines = [ln for ln in raw.splitlines() if not ln.strip().startswith("--")]
+    cleaned = "\n".join(lines)
+    stmts = [s.strip() for s in cleaned.split(";") if s.strip()]
+    creates = [s for s in stmts if s.upper().startswith("CREATE TABLE")]
+    if len(creates) != len(TABLES):
+        raise RuntimeError(f"DDL 真源语句数异常: 期望 {len(TABLES)} 条 CREATE TABLE，"
+                           f"实际 {len(creates)} 条（{SQL_FILE}）")
+    return creates
+
+
+def db_name_of(engine) -> str:
+    try:
+        return engine.url.database or ""
+    except Exception:
+        return ""
+
+
+def describe(engine) -> dict:
+    """只读预检：返回 {table: {'exists': bool, 'missing_cols': [...], 'unique_keys': [...]}}"""
+    insp = inspect(engine)
+    existing_tables = set(insp.get_table_names())
+    out = {}
+    for t in TABLES:
+        info = {"exists": t in existing_tables, "missing_cols": [], "unique_keys": []}
+        if info["exists"]:
+            cols = {c["name"] for c in insp.get_columns(t)}
+            expected = EXPECTED_COLUMNS[t]
+            info["missing_cols"] = [c for c in expected if c not in cols]
+            info["unique_keys"] = [tuple(uc["column_names"]) for uc in insp.get_unique_constraints(t)]
+        out[t] = info
+    return out
+
+
+EXPECTED_COLUMNS = {
+    "tax_filing_log": ("id", "entity_id", "period", "tax_type", "legal_deadline", "filed_date",
+                       "on_time", "evidence_no", "registered_by", "remark", "created_at", "updated_at"),
+    "hr_monthly_ledger": ("id", "entity_id", "period", "headcount_begin", "hires", "separations",
+                          "headcount_end", "registered_by", "remark", "created_at", "updated_at"),
+    "licence_ledger": ("id", "entity_id", "licence_name", "licence_no", "valid_from", "valid_to",
+                       "annual_check_status", "issuer", "owner_role", "attachment_no",
+                       "registered_by", "remark", "created_at", "updated_at"),
+}
+
+
+def print_check(state: dict) -> None:
+    print("== 只读预检（information_schema / inspector，零写入）==")
+    for t in TABLES:
+        s = state[t]
+        print(f"  {t}: exists={s['exists']} missing_cols={s['missing_cols']} unique_keys={s['unique_keys']}")
+
+
+def main():
+    ap = argparse.ArgumentParser(description="外部输入台账 DDL 迁移")
+    g = ap.add_mutually_exclusive_group()
+    g.add_argument("--check", action="store_true", help="只读预检（默认）")
+    g.add_argument("--apply", action="store_true", help="执行建表（幂等）")
+    g.add_argument("--down", action="store_true", help="回滚：DROP 三表")
+    ap.add_argument("--show-create", action="store_true", help="打印 SHOW CREATE TABLE 原文")
+    ap.add_argument("--url", default=None, help="目标库 SQLAlchemy URL（默认 backend/.env）")
+    ap.add_argument("--allow-prod", action="store_true", help="生产库放行开关（须 CMA_DDL_APPROVED=1）")
+    ap.add_argument("--yes", action="store_true", help="--down 且有数据时确认删除")
+    args = ap.parse_args()
+
+    if args.url:
+        engine = create_engine(args.url, echo=False)
+    else:
+        from app.database import get_engine
+        engine = get_engine()
+
+    dbname = db_name_of(engine)
+    print(f"目标库: {dbname or '(未知)'} | dialect={engine.dialect.name} | url_host={engine.url.host or 'socket'}")
+    is_prod = dbname in PROD_DB_NAMES
+
+    if (args.apply or args.down) and is_prod:
+        if not args.allow_prod or os.getenv("CMA_DDL_APPROVED") != "1":
+            print("!!! 拒绝执行：目标库是生产库 %s。" % dbname)
+            print("!!! 生产 DDL 须审批后另派执行（本单边界：不得直连生产写入）。")
+            print("!!! 如需放行：--allow-prod 且 export CMA_DDL_APPROVED=1（持有审批单）。")
+            return 3
+
+    if args.apply:
+        stmts = load_statements()
+        with engine.begin() as conn:
+            for s in stmts:
+                conn.execute(text(s))
+        state = describe(engine)
+        missing = [t for t in TABLES if not state[t]["exists"]]
+        print(f"== APPLY 完成：DDL 真源 {os.path.basename(SQL_FILE)}，语句 {len(stmts)} 条 ==")
+        print_check(state)
+        print(f"  仍缺表: {missing}（二次运行应为 0 新增表）")
+        return 1 if missing else 0
+
+    if args.down:
+        counts = {}
+        with engine.connect() as conn:
+            for t in TABLES:
+                if t in set(inspect(engine).get_table_names()):
+                    counts[t] = conn.execute(text(f"SELECT COUNT(*) FROM {t}")).scalar()
+        if any(counts.values()) and not args.yes:
+            print(f"!!! 表内已有数据 {counts}：回滚须显式 --yes（防误删业务数据）")
+            return 3
+        with engine.begin() as conn:
+            for t in TABLES:
+                conn.execute(text(f"DROP TABLE IF EXISTS {t}"))
+        print(f"== DOWN 完成：DROP {list(TABLES)}（回滚前行数 {counts}）==")
+        print_check(describe(engine))
+        return 0
+
+    # --check（默认）或 --show-create
+    state = describe(engine)
+    print_check(state)
+    if is_prod:
+        print("  [注意] 目标是生产库：以上仅为只读预检，未做任何写入。")
+    if args.show_create:
+        if engine.dialect.name != "mysql":
+            print("== SHOW CREATE TABLE 跳过（非 MySQL 目标）==")
+        else:
+            with engine.connect() as conn:
+                for t in TABLES:
+                    if not state[t]["exists"]:
+                        print(f"== {t}: 表不存在，无 SHOW CREATE 输出 ==")
+                        continue
+                    row = conn.execute(text(f"SHOW CREATE TABLE {t}")).fetchone()
+                    print(f"== SHOW CREATE TABLE {t} ==")
+                    print(row[1])
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/backend/scripts/sql/ext_input_ledgers_20260915.sql b/backend/scripts/sql/ext_input_ledgers_20260915.sql
new file mode 100644
index 00000000..9539573a
--- /dev/null
+++ b/backend/scripts/sql/ext_input_ledgers_20260915.sql
@@ -0,0 +1,77 @@
+-- ============================================================================
+-- 外部输入台账 DDL（能力评价 M2 准入3 派生）
+-- 任务: 0-P0-project-20260915-cma-eval-m2-ext-input-tables
+-- 日期: 2026-09-15
+--
+-- 三张**外部输入**台账 = 能力评价 3 项缺口值的唯一合法取证来源（PRD v1.3 §4.0）：
+--   tax_filing_log    申报行为台账 → P_TAX_TIMELY      (A4 税务申报及时率, 月频)
+--   hr_monthly_ledger 人事月度台账 → L_EMPLOYEE_TURNOVER (B5 员工流失率, 月频)
+--   licence_ledger    证照台账     → P_LICENCE_OK      (B2 经营许可合规率, 季频)
+--
+-- 执行方式（**勿直接 mysql < 本文件**，列级/表级幂等由脚本保证）：
+--   backend/scripts/migrate_ext_input_ledgers_20260915.py --check    # 只读预检（默认）
+--   backend/scripts/migrate_ext_input_ledgers_20260915.py --apply    # 建表（幂等）
+--   backend/scripts/migrate_ext_input_ledgers_20260915.py --show-create  # 验收证据
+--   生产库（cma）执行须审批后另派：脚本对 DB_NAME=='cma' 默认拒绝 --apply/--down
+--
+-- 设计纪律：
+--   * 只新增表，不改既有表；不改任何 KPI 定义（kpi_definitions 属准入2/4 同表串行）
+--   * 每表 entity_id NOT NULL，唯一键一律以 entity_id 打头（多租户隔离 + 索引可用性）
+--   * on_time 为服务端派生字段（app/services/ext_input_derive.derive_on_time），录入方不得手填
+--   * 不建物理外键（业务关系列走软链名），保持 ORM 声明 == DDL 实况
+-- ============================================================================
+
+CREATE TABLE IF NOT EXISTS `tax_filing_log` (
+  `id` INT NOT NULL AUTO_INCREMENT,
+  `entity_id` INT NOT NULL COMMENT '企业ID（账套）',
+  `period` VARCHAR(20) NOT NULL COMMENT '属期 YYYY-MM（申报所属月份）',
+  `tax_type` VARCHAR(50) NOT NULL COMMENT '税种: 增值税/附加税/企业所得税/印花税…',
+  `legal_deadline` DATE NULL COMMENT '法定申报期限',
+  `filed_date` DATE NULL COMMENT '实际申报日（未申报为 NULL）',
+  `on_time` INT NULL COMMENT '是否按时申报 1/0（服务端派生，录入方不得手填）',
+  `evidence_no` VARCHAR(100) NULL COMMENT '申报回执/截图编号',
+  `registered_by` VARCHAR(100) NULL COMMENT '登记人（职能岗位）',
+  `remark` VARCHAR(500) NULL COMMENT '备注',
+  `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
+  `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uq_tax_filing_log_entity_period_type` (`entity_id`,`period`,`tax_type`),
+  KEY `ix_tax_filing_log_entity_id` (`entity_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='申报行为台账（P_TAX_TIMELY 取证来源）';
+
+CREATE TABLE IF NOT EXISTS `hr_monthly_ledger` (
+  `id` INT NOT NULL AUTO_INCREMENT,
+  `entity_id` INT NOT NULL COMMENT '企业ID（账套）',
+  `period` VARCHAR(20) NOT NULL COMMENT '期间 YYYY-MM',
+  `headcount_begin` INT NOT NULL COMMENT '期初在职人数',
+  `hires` INT NOT NULL DEFAULT 0 COMMENT '入职人数',
+  `separations` INT NOT NULL DEFAULT 0 COMMENT '离职人数',
+  `headcount_end` INT NOT NULL COMMENT '期末在职人数（须= 期初+入职-离职）',
+  `registered_by` VARCHAR(100) NULL COMMENT '登记人（职能岗位）',
+  `remark` VARCHAR(500) NULL COMMENT '备注',
+  `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
+  `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uq_hr_monthly_ledger_entity_period` (`entity_id`,`period`),
+  KEY `ix_hr_monthly_ledger_entity_id` (`entity_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='人事月度台账（L_EMPLOYEE_TURNOVER 取证来源）';
+
+CREATE TABLE IF NOT EXISTS `licence_ledger` (
+  `id` INT NOT NULL AUTO_INCREMENT,
+  `entity_id` INT NOT NULL COMMENT '企业ID（账套）',
+  `licence_name` VARCHAR(200) NOT NULL COMMENT '证照名称',
+  `licence_no` VARCHAR(100) NOT NULL COMMENT '证照编号',
+  `valid_from` DATE NULL COMMENT '有效期起',
+  `valid_to` DATE NULL COMMENT '有效期止（空=长期有效）',
+  `annual_check_status` VARCHAR(20) NOT NULL DEFAULT '未到期' COMMENT '年检状态: 未到期/已年检/待年检/过期',
+  `issuer` VARCHAR(200) NULL COMMENT '发证/主管机关',
+  `owner_role` VARCHAR(100) NULL COMMENT '责任岗位（职能，不指定自然人）',
+  `attachment_no` VARCHAR(100) NULL COMMENT '附件/扫描件编号',
+  `registered_by` VARCHAR(100) NULL COMMENT '登记人（职能岗位）',
+  `remark` VARCHAR(500) NULL COMMENT '备注',
+  `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
+  `updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uq_licence_ledger_entity_name_no` (`entity_id`,`licence_name`,`licence_no`),
+  KEY `ix_licence_ledger_entity_id` (`entity_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='证照台账（P_LICENCE_OK 取证来源）';
diff --git a/backend/scripts/verify_ext_input_ledgers_e2e.py b/backend/scripts/verify_ext_input_ledgers_e2e.py
new file mode 100644
index 00000000..40d9b515
--- /dev/null
+++ b/backend/scripts/verify_ext_input_ledgers_e2e.py
@@ -0,0 +1,311 @@
+#!/usr/bin/env python3
+"""外部输入台账 端到端验收电池（**试验田 = scratch 库**，零生产写入）
+任务: 0-P0-project-20260915-cma-eval-m2-ext-input-tables（2026-09-15）
+
+做什么：
+  1. root@unix_socket 建库 cma_scratch_ext_<ts>（先 DROP IF EXISTS，保证可复跑）
+  2. 跑真实迁移脚本（--check / --apply / 二次 --apply 幂等 / --show-create 证据），目标 = scratch
+  3. scratch 上建其余表 + 最小夹具（entities / users / kpi_definitions），不复制生产数据
+  4. 真实 HTTP（TestClient）跑：新增 / CSV 批量导入 ×2 / 按 (entity_id, period) 查询 /
+     派生 ×2（幂等）/ 分母 0 不写值（税务·人事·证照三口径）/ 定义缺失跳过 / 跨账套 403 / 未认证 401
+     —— **原始请求与响应 JSON 全文打印**
+  5. 收尾 DROP scratch
+
+边界自证（本单「生产库未写入」）：
+  * 全程只创建 1 个 engine，URL 指向 scratch 库；脚本不读也不写 cma 库
+  * 迁移脚本对 DB_NAME=='cma' 的 --apply/--down 默认拒绝（下方单独打印该守卫的实测输出）
+  * 不复制生产数据（未使用 mysqldump）：夹具全部脚本内合成；token 走进程内存储（不写生产 Redis）
+
+用法: cd /root/cma-management/backend && ./venv/bin/python3 scripts/verify_ext_input_ledgers_e2e.py
+"""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+BACKEND = os.path.dirname(HERE)
+sys.path.insert(0, BACKEND)
+
+import pymysql  # noqa: E402
+from sqlalchemy import create_engine  # noqa: E402
+
+SOCKET = "/var/run/mysqld/mysqld.sock"
+SCRATCH = f"cma_scratch_ext_{time.strftime('%Y%m%d%H%M%S')}"
+SCRATCH_URL = f"mysql+pymysql://root@localhost/{SCRATCH}?unix_socket={SOCKET}&charset=utf8mb4"
+MIGRATE = os.path.join(HERE, "migrate_ext_input_ledgers_20260915.py")
+VENV_PY = os.path.join(BACKEND, "venv", "bin", "python3")
+FAILURES = []
+
+
+def say(title: str, body: str = "") -> None:
+    print(f"\n===== {title} =====")
+    if body:
+        print(body)
+
+
+def check(name: str, cond: bool, detail: str = "") -> None:
+    print(f"  [{'PASS' if cond else 'FAIL'}] {name}{(' | ' + detail) if detail else ''}")
+    if not cond:
+        FAILURES.append(f"{name} | {detail}")
+
+
+def raw_req(label: str, resp) -> None:
+    """原始请求/响应全文（验收要求「真实请求与响应原文贴出」）"""
+    body = resp.text
+    if len(body) > 3000:
+        body = body[:3000] + "...(truncated)"
+    print(f"  -> {label}: HTTP {resp.status_code}\n     {body}")
+
+
+def root_conn(db=None):
+    return pymysql.connect(unix_socket=SOCKET, user="root", database=db, charset="utf8mb4")
+
+
+def count(db, model) -> int:
+    """回查行数：先 rollback 结束当前事务 —— MySQL REPEATABLE READ 下 session 的长事务
+    会固定快照，看不到 API 请求（另一连接）刚提交的行，会误判「写入 0 行」。"""
+    db.rollback()
+    return db.query(model).count()
+
+
+def main() -> int:
+    con = root_conn()
+    with con.cursor() as cur:
+        cur.execute(f"DROP DATABASE IF EXISTS {SCRATCH}")
+        cur.execute(f"CREATE DATABASE {SCRATCH} DEFAULT CHARSET utf8mb4")
+    con.close()
+    print(f"scratch 库已建: {SCRATCH}")
+
+    # ── 1. 迁移脚本：check → apply → 二次 apply（幂等）→ show-create ──────────
+    def run_migrate(*args):
+        p = subprocess.run([VENV_PY, MIGRATE, "--url", SCRATCH_URL] + list(args),
+                           capture_output=True, text=True, cwd=BACKEND)
+        return p.returncode, (p.stdout or "") + (p.stderr or "")
+
+    rc, out = run_migrate("--check")
+    say("迁移 --check（建表前，期望三表 exists=False）", out)
+    check("apply 前 --check rc=0 且三表不存在", rc == 0 and out.count("exists=False") == 3)
+
+    rc, out = run_migrate("--apply")
+    say("迁移 --apply（建 3 张外部输入台账）", out)
+    check("apply rc=0 且仍缺表为 0", rc == 0 and "仍缺表: []" in out)
+
+    rc, out2 = run_migrate("--apply")
+    say("迁移 --apply 二次（幂等复核）", out2)
+    check("二次 apply 幂等（仍缺表: []，新增 0）", "仍缺表: []" in out2)
+
+    rc, out = run_migrate("--check")
+    table_lines = [ln for ln in out.splitlines()
+                   if ln.strip().startswith(("tax_filing_log", "hr_monthly_ledger", "licence_ledger"))]
+    check("--check 三表存在且缺列=[]",
+          len(table_lines) == 3 and all("exists=True" in ln and "missing_cols=[]" in ln for ln in table_lines))
+
+    rc, out = run_migrate("--show-create")
+    say("SHOW CREATE TABLE 原文（验收证据）", out)
+
+    # ── 1b. 生产守卫实测（目标库名=cma 时 --apply 必须被拒）──────────────────
+    guard_env = dict(os.environ)
+    guard_env.pop("CMA_DDL_APPROVED", None)
+    p = subprocess.run([VENV_PY, MIGRATE, "--apply", "--url",
+                        "mysql+pymysql://root@localhost/cma?charset=utf8mb4"],
+                       capture_output=True, text=True, cwd=BACKEND, env=guard_env)
+    say("生产守卫实测（--apply 目标=cma，期望拒绝且零执行）", (p.stdout or "") + (p.stderr or ""))
+    check("生产库 --apply 被守卫拒绝（exit=3）",
+          p.returncode == 3 and "拒绝执行" in (p.stdout or ""), f"exit={p.returncode}")
+
+    # ── 2. 指向 scratch 的 engine + 建其余表 + 夹具 ─────────────────────────
+    import app.database as dbmod
+    dbmod.DATABASE_URL = SCRATCH_URL
+    dbmod._engine = None
+    dbmod._SessionLocal = None
+    engine = dbmod.get_engine()
+    check("app.database 指向 scratch（非生产）", engine.url.database == SCRATCH, str(engine.url.database))
+
+    import app.auth_middleware as am
+    am._redis_available = False        # token 走进程内内存，不碰生产 Redis
+    am._token_store.clear()
+
+    from app.database import Base
+    import app.models  # noqa: F401  注册全部模型
+    from app.models import (Entity, KPIDefinition, KPIValue, LicenceLedger, TaxFilingLog,
+                            HrMonthlyLedger, User)
+
+    Base.metadata.create_all(bind=engine)   # 三张新表已由迁移脚本建好 → checkfirst 跳过
+
+    Session = dbmod.get_session_local()
+    db = Session()
+    db.add_all([Entity(id=1, name="测试酣客", short_name="酣客", status="active"),
+                Entity(id=2, name="测试博海", short_name="博海", status="active")])
+    db.add(User(id=1, username="vfy_admin", password_hash="x", name="验收管理员", role="ceo"))
+    for code, name in (("P_TAX_TIMELY", "税务申报及时率"), ("L_EMPLOYEE_TURNOVER", "员工流失率"),
+                       ("P_LICENCE_OK", "经营许可合规率")):
+        db.add(KPIDefinition(entity_id=1, kpi_code=code, kpi_name=name, unit="%",
+                             dimension="process", status="active"))
+    db.commit()
+    H1 = {"Authorization": f"Bearer {am.create_token(1, 1)}"}   # user 1 / entity 1
+    H2 = {"Authorization": f"Bearer {am.create_token(1, 2)}"}   # user 1 / entity 2
+    say("夹具", "entity 1/2；user id=1(ceo)；entity 1 有 3 个 KPI 定义、entity 2 无定义；"
+                "token 走进程内存储（未写生产 Redis）")
+
+    from fastapi.testclient import TestClient
+    from app.main import app
+    client = TestClient(app)
+
+    # ── 3. 真实 HTTP：新增 / on_time 服务端派生 ───────────────────────────
+    r = client.post("/api/cma/ext-ledger/tax-filings", headers=H1, json={
+        "entity_id": 1, "period": "2026-08", "tax_type": "个人所得税",
+        "legal_deadline": "2026-08-15", "filed_date": "2026-08-20",
+        "on_time": 1, "evidence_no": "HZ-000", "registered_by": "财务部"})
+    raw_req("POST /tax-filings（请求带 on_time=1，实际逾期）", r)
+    b = r.json()
+    check("on_time 服务端派生 = 0（手填值被忽略并回报）",
+          r.status_code == 201 and b["data"]["on_time"] == 0 and b["on_time_ignored"] is True,
+          json.dumps(b, ensure_ascii=False))
+
+    r = client.post("/api/cma/ext-ledger/tax-filings", headers=H1, json={
+        "period": "2026-08", "tax_type": "个人所得税", "legal_deadline": "2026-08-15",
+        "filed_date": "2026-08-12"})
+    raw_req("POST /tax-filings 重复（唯一键冲突）", r)
+    check("唯一键冲突 → 409 且不新增", r.status_code == 409)
+
+    # ── 4. CSV 批量导入 ×2（幂等）────────────────────────────────────────
+    tax_csv = ("period,tax_type,legal_deadline,filed_date,evidence_no\n"
+               "2026-08,增值税,2026-08-15,2026-08-12,HZ-001\n"
+               "2026-08,附加税,2026-08-15,2026-08-12,HZ-002\n"
+               "2026-08,印花税,2026-08-15,2026-08-20,HZ-003\n"
+               "2026-07,增值税,2026-07-15,2026-07-10,HZ-004\n"
+               "2026-07,附加税,2026-07-15,2026-07-10,HZ-005\n")
+    r1 = client.post("/api/cma/ext-ledger/tax-filings/import", headers=H1,
+                     json={"csv": tax_csv, "on_conflict": "skip"})
+    raw_req("POST /tax-filings/import #1（CSV 5 行）", r1)
+    r2 = client.post("/api/cma/ext-ledger/tax-filings/import", headers=H1,
+                     json={"csv": tax_csv, "on_conflict": "skip"})
+    raw_req("POST /tax-filings/import #2（同 CSV 复跑）", r2)
+    check("CSV 导入幂等（#1 inserted=5；#2 inserted=0 skipped=5）",
+          r1.json()["inserted"] == 5 and r2.json()["inserted"] == 0 and r2.json()["skipped"] == 5)
+    n_tax = count(db, TaxFilingLog)
+    check("台账行数 = 6（1 新增 + 5 导入，无重复行）", n_tax == 6, f"rows={n_tax}")
+
+    r = client.post("/api/cma/ext-ledger/hr-monthly", headers=H1, json={
+        "period": "2026-08", "headcount_begin": 100, "hires": 2, "separations": 1,
+        "headcount_end": 99})
+    raw_req("POST /hr-monthly（等式不成立）", r)
+    check("人事等式不成立 → 400 拒绝写入", r.status_code == 400 and "等式不成立" in r.json()["detail"])
+    check("被拒行未落库", count(db, HrMonthlyLedger) == 0)
+    r = client.post("/api/cma/ext-ledger/hr-monthly/import", headers=H1, json={
+        "csv": ("period,headcount_begin,hires,separations,headcount_end\n"
+                "2026-08,120,2,4,118\n2026-07,118,3,1,120\n")})
+    raw_req("POST /hr-monthly/import（2 行合法）", r)
+    check("人事台账导入 2 行（2026-08: 120-4+2=118）", r.json()["inserted"] == 2)
+
+    r = client.post("/api/cma/ext-ledger/licences/import", headers=H1, json={
+        "csv": ("licence_name,licence_no,valid_from,valid_to,annual_check_status,issuer\n"
+                "营业执照,91610100MA0001,2020-01-01,2030-01-01,已年检,市场监管局\n"
+                "食品经营许可证,JY1610100001,2024-01-01,2026-03-31,过期,市场监管局\n"
+                "酒类流通备案,HZ-2024-01,,2030-12-31,未到期,商务局\n")})
+    raw_req("POST /licences/import（3 行，其中 1 张已过期）", r)
+    check("证照台账导入 3 行", r.json()["inserted"] == 3)
+
+    # ── 5. 按 (entity_id, period) 查询 ────────────────────────────────────
+    r = client.get("/api/cma/ext-ledger/tax-filings?period=2026-08", headers=H1)
+    raw_req("GET /tax-filings?period=2026-08", r)
+    check("查询按期间过滤；及时率 2/4 = 50.0",
+          r.json()["total"] == 4 and r.json()["summary"]["tax_timely_rate"] == 50.0,
+          json.dumps(r.json()["summary"], ensure_ascii=False))
+
+    r = client.get("/api/cma/ext-ledger/hr-monthly?period=2026-08", headers=H1)
+    raw_req("GET /hr-monthly?period=2026-08", r)
+    check("人事查询含流失率派生（4/120 = 3.33%）", r.json()["data"][0]["turnover_rate"] == 3.33)
+
+    r = client.get("/api/cma/ext-ledger/licences?as_of=2026-06-30", headers=H1)
+    raw_req("GET /licences?as_of=2026-06-30", r)
+    check("证照合规率 2/3 = 66.7（过期证照不计入有效数）",
+          r.json()["summary"]["licence_ok"] == 66.7 and r.json()["summary"]["valid"] == 2,
+          json.dumps(r.json()["summary"], ensure_ascii=False))
+
+    # ── 6. 派生 ×2（幂等）────────────────────────────────────────────────
+    d1 = client.post("/api/cma/ext-ledger/derive", headers=H1, json={"period": "2026-08"})
+    raw_req("POST /derive #1（2026-08）", d1)
+    v1 = {k: v["actual_value"] for k, v in d1.json()["items"].items()}
+    a1 = {k: v["action"] for k, v in d1.json()["items"].items()}
+    check("派生值 = 税务 50.0 / 流失率 3.33 / 合规率 66.7(2026-Q3)",
+          v1 == {"P_TAX_TIMELY": 50.0, "L_EMPLOYEE_TURNOVER": 3.33, "P_LICENCE_OK": 66.7},
+          json.dumps(v1, ensure_ascii=False))
+    check("首次派生全 insert（含季频期间归集 2026-08 → 2026-Q3）",
+          set(a1.values()) == {"insert"} and d1.json()["items"]["P_LICENCE_OK"]["period"] == "2026-Q3",
+          json.dumps(a1, ensure_ascii=False))
+    kv1 = count(db, KPIValue)
+    check("kpi_values 写入 3 行", kv1 == 3, f"rows={kv1}")
+
+    d2 = client.post("/api/cma/ext-ledger/derive", headers=H1, json={"period": "2026-08"})
+    raw_req("POST /derive #2（同期间复跑 → 幂等）", d2)
+    a2 = {k: v["action"] for k, v in d2.json()["items"].items()}
+    kv2 = count(db, KPIValue)
+    check("派生幂等（第二次全 upsert；kpi_values 行数不变）",
+          set(a2.values()) == {"upsert"} and kv1 == kv2, f"kv#1={kv1} kv#2={kv2} actions={a2}")
+
+    # 分母 0（口径一/二）：2026-09 无任何税务/人事台账行 → None 且不写值
+    d3 = client.post("/api/cma/ext-ledger/derive", headers=H1, json={
+        "period": "2026-09", "kpi_codes": ["P_TAX_TIMELY", "L_EMPLOYEE_TURNOVER"]})
+    raw_req("POST /derive（2026-09 台账空 → 分母 0）", d3)
+    check("分母为 0 → actual_value=None 且 action=skipped_denominator_zero",
+          all(v["actual_value"] is None and v["action"] == "skipped_denominator_zero"
+              for v in d3.json()["items"].values()),
+          json.dumps({k: v["action"] for k, v in d3.json()["items"].items()}, ensure_ascii=False))
+    check("分母 0 未新增 kpi_values 行（无 0、无大数）", count(db, KPIValue) == kv2)
+
+    # 分母 0（口径三）：证照台账清空后 P_LICENCE_OK 亦为 None（且回报旧行 stale）
+    db.query(LicenceLedger).delete()
+    db.commit()
+    d4 = client.post("/api/cma/ext-ledger/derive", headers=H1, json={
+        "period": "2026-08", "kpi_codes": ["P_LICENCE_OK"]})
+    raw_req("POST /derive（证照台账已清空 → 应持证照数 0）", d4)
+    it4 = d4.json()["items"]["P_LICENCE_OK"]
+    check("证照分母 0 → None 且不覆盖/不新增值",
+          it4["actual_value"] is None and it4["action"] == "skipped_denominator_zero"
+          and it4["stale_row_present"] is True,
+          json.dumps(it4, ensure_ascii=False)[:300])
+
+    # 定义缺失：不新建定义
+    d5 = client.post("/api/cma/ext-ledger/derive", headers=H2, json={"period": "2026-08"})
+    raw_req("POST /derive（entity=2 无 KPI 定义）", d5)
+    check("定义缺失 → skipped_no_definition（本模块不新建/不改动 KPI 定义）",
+          all(v["action"] == "skipped_no_definition" for v in d5.json()["items"].values()))
+    check("KPI 定义总数未变", count(db, KPIDefinition) == 3,
+          f"defs={count(db, KPIDefinition)}")
+
+    # ── 7. 边界：跨账套 403 / 未认证 401 ─────────────────────────────────
+    r = client.post("/api/cma/ext-ledger/hr-monthly?entity_id=2", headers=H1, json={
+        "period": "2026-08", "headcount_begin": 10, "hires": 0, "separations": 0, "headcount_end": 10})
+    raw_req("POST /hr-monthly?entity_id=2（会话账套=1）", r)
+    check("跨账套 → 403", r.status_code == 403 and "无权访问企业 entity_id=2" in r.json()["detail"])
+    r = client.get("/api/cma/ext-ledger/options")
+    raw_req("GET /options（无 token）", r)
+    check("未认证 → 401", r.status_code == 401)
+
+    # ── 8. 收尾 ─────────────────────────────────────────────────────────
+    say("汇总断言", f"FAIL 数 = {len(FAILURES)}" + ("" if not FAILURES else "\n  " + "\n  ".join(FAILURES)))
+    db.close()
+    engine.dispose()
+    con = root_conn()
+    with con.cursor() as cur:
+        cur.execute("SET SESSION lock_wait_timeout=8")
+        for _ in range(3):
+            try:
+                cur.execute(f"DROP DATABASE IF EXISTS {SCRATCH}")
+                break
+            except Exception:
+                time.sleep(1)
+    con.close()
+    print(f"\nscratch 库已删除: {SCRATCH}")
+    print("边界自证: 全程 engine 数=1（URL 指向 scratch）；未连接/未读写生产库 cma；"
+          "未使用 mysqldump 复制生产数据；token 未写生产 Redis")
+    return 1 if FAILURES else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/backend/tests/test_ext_input_ledgers_20260915.py b/backend/tests/test_ext_input_ledgers_20260915.py
new file mode 100644
index 00000000..c98272d8
--- /dev/null
+++ b/backend/tests/test_ext_input_ledgers_20260915.py
@@ -0,0 +1,364 @@
+"""外部输入台账 单测 — 3 张外部输入表 + 录入接口 + 台账→kpi_values 派生
+（任务 0-P0-project-20260915-cma-eval-m2-ext-input-tables，2026-09-15）
+
+覆盖任务单「验收标准」中的硬性项：
+  ① on_time 由服务端派生（传入值被忽略/拒绝）
+  ② 人事台账「期末 ≠ 期初 + 入职 − 离职」被拒
+  ③ 已过期证照不计入有效数
+  ④ 派生函数：**分母为 0 返回 None**（不写 0、不填大数）+ 3 个指标各 1 个正例
+  ⑤ 幂等：台账导入与派生各跑两次，第二次输出/行数与第一次一致
+  ⑥ 多租户：跨账套显式 entity_id → 403
+另覆盖：CSV 批量导入、按 (entity_id, period) 查询、唯一键冲突 409、季频期间归集。
+"""
+import os
+import sys
+from datetime import date
+from types import SimpleNamespace
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import pytest  # noqa: E402
+
+from app.models import KPIDefinition, KPIValue, HrMonthlyLedger, LicenceLedger, TaxFilingLog  # noqa: E402
+from app.services import ext_input_derive as D  # noqa: E402
+from tests.conftest import auth_header, create_test_kpi, create_test_user, get_token_for_user  # noqa: E402
+
+TAX_CSV = """period,tax_type,legal_deadline,filed_date,evidence_no
+2026-08,增值税,2026-08-15,2026-08-12,HZ-001
+2026-08,附加税,2026-08-15,2026-08-12,HZ-002
+2026-08,印花税,2026-08-15,2026-08-20,HZ-003
+"""
+
+HR_CSV = """period,headcount_begin,hires,separations,headcount_end
+2026-08,100,1,2,99
+2026-07,98,2,0,100
+"""
+
+LIC_CSV = """licence_name,licence_no,valid_from,valid_to,annual_check_status,issuer
+营业执照,91610100MA0001,2020-01-01,2030-01-01,已年检,市场监管局
+食品经营许可证,JY1610100001,2024-01-01,2026-03-31,过期,市场监管局
+"""
+
+
+def _rows(db, model, **filters):
+    q = db.query(model)
+    for k, v in filters.items():
+        q = q.filter(getattr(model, k) == v)
+    return q.all()
+
+
+# ============================================================
+# 纯函数：on_time 派生 / 分母为 0 / 三个指标正例
+# ============================================================
+
+def test_on_time_is_server_derived():
+    """on_time 唯一规则：按时=1 / 逾期=0 / 未申报=0 / 缺法定期限且已申报=无法判定(None)"""
+    assert D.derive_on_time(date(2026, 8, 12), date(2026, 8, 15)) == 1
+    assert D.derive_on_time(date(2026, 8, 15), date(2026, 8, 15)) == 1
+    assert D.derive_on_time(date(2026, 8, 20), date(2026, 8, 15)) == 0
+    assert D.derive_on_time(None, date(2026, 8, 15)) == 0       # 未申报
+    assert D.derive_on_time(date(2026, 8, 12), None) is None    # 无法判定
+
+
+def test_denominator_zero_returns_none():
+    """硬规则：分母为 0 → None（无定义/不参评），不得写 0、不得填大数"""
+    assert D.compute_tax_timely_rate([]) is None
+    assert D.compute_employee_turnover(SimpleNamespace(headcount_begin=0, separations=3,
+                                                      headcount_end=0, hires=0)) is None
+    assert D.compute_licence_ok([], date(2026, 6, 30)) is None
+    assert D.compute_employee_turnover(None) is None
+    # explain_* 同源，no_definition 同步为 True
+    assert D.explain_tax_timely_rate([])["no_definition"] is True
+    assert D.explain_employee_turnover(SimpleNamespace(headcount_begin=0, separations=0,
+                                                       headcount_end=0, hires=0))["no_definition"] is True
+    assert D.explain_licence_ok([], date(2026, 6, 30))["no_definition"] is True
+
+
+def test_tax_timely_rate_positive():
+    """正例：3 个应申报税种，2 个按时 → 66.7%（未知 on_time 不计入分子，仍计入分母）"""
+    rows = [SimpleNamespace(on_time=1), SimpleNamespace(on_time=1), SimpleNamespace(on_time=0)]
+    assert D.compute_tax_timely_rate(rows) == 66.7
+    rows2 = [SimpleNamespace(on_time=1), SimpleNamespace(on_time=None)]
+    assert D.compute_tax_timely_rate(rows2) == 50.0
+
+
+def test_employee_turnover_positive():
+    """正例：期初 100 离职 2 → 2.0%；并列返回期初期末平均在职口径（99 人 → 2.02%）"""
+    row = SimpleNamespace(headcount_begin=100, headcount_end=98, hires=1, separations=2)
+    assert D.compute_employee_turnover(row) == 2.0
+    detail = D.explain_employee_turnover(row)
+    assert detail["avg_headcount"] == 99.0
+    assert detail["value_on_avg"] == 2.02
+
+
+def test_licence_ok_excludes_expired():
+    """正例：4 张证照，1 张过期 + 1 张已到期 → 有效 2 张 = 50.0%（valid_to 空=长期有效）"""
+    as_of = date(2026, 6, 30)
+    rows = [
+        SimpleNamespace(annual_check_status="已年检", valid_to=date(2030, 1, 1), licence_no="A"),
+        SimpleNamespace(annual_check_status="过期", valid_to=date(2030, 1, 1), licence_no="B"),
+        SimpleNamespace(annual_check_status="未到期", valid_to=date(2026, 3, 31), licence_no="C"),
+        SimpleNamespace(annual_check_status="未到期", valid_to=None, licence_no="D"),
+    ]
+    assert D.compute_licence_ok(rows, as_of) == 50.0
+    detail = D.explain_licence_ok(rows, as_of)
+    assert detail["valid"] == 2 and detail["denominator"] == 4
+    assert detail["expired_or_outdated"] == ["B"]
+
+
+def test_headcount_equation_check():
+    assert D.check_headcount_equation(100, 1, 2, 99) is None
+    msg = D.check_headcount_equation(100, 2, 1, 99)
+    assert msg and "期末(99)" in msg and "101" in msg
+
+
+def test_quarter_helpers():
+    assert D.month_to_quarter("2026-06") == "2026-Q2"
+    assert D.month_to_quarter("2026-01") == "2026-Q1"
+    assert D.quarter_end_date("2026-Q2") == date(2026, 6, 30)
+    assert D.quarter_end_date("2026-06") == date(2026, 6, 30)
+    assert D.quarter_end_date("2026-Q4") == date(2026, 12, 31)
+
+
+# ============================================================
+# API：新增 / on_time 忽略 / 唯一键 409 / 校验拒绝
+# ============================================================
+
+def test_create_tax_filing_ignores_client_on_time(client, db):
+    """① on_time 服务端派生：请求里带 on_time=1 但 filed_date 晚于法定期限 → 落库 0 + 回报被忽略"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    resp = client.post("/api/cma/ext-ledger/tax-filings", headers=auth_header(token), json={
+        "period": "2026-08", "tax_type": "增值税",
+        "legal_deadline": "2026-08-15", "filed_date": "2026-08-20",
+        "on_time": 1, "evidence_no": "HZ-001",
+    })
+    assert resp.status_code == 201, resp.text
+    body = resp.json()
+    assert body["data"]["on_time"] == 0                     # 服务端派生（未被手填值污染）
+    assert body["on_time_derived"] == 0
+    assert body["on_time_ignored"] is True
+    row = _rows(db, TaxFilingLog, period="2026-08")[0]
+    assert (row.on_time, row.entity_id) == (0, 1)
+
+
+def test_create_tax_filing_duplicate_409(client, db):
+    """唯一键 (entity_id, period, tax_type) 冲突 → 409，不产生第二行"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    payload = {"period": "2026-08", "tax_type": "增值税",
+               "legal_deadline": "2026-08-15", "filed_date": "2026-08-12"}
+    assert client.post("/api/cma/ext-ledger/tax-filings", headers=auth_header(token),
+                       json=payload).status_code == 201
+    dup = client.post("/api/cma/ext-ledger/tax-filings", headers=auth_header(token), json=payload)
+    assert dup.status_code == 409
+    assert "已存在" in dup.json()["detail"]
+    assert len(_rows(db, TaxFilingLog)) == 1
+
+
+def test_hr_equation_rejected(client, db):
+    """② 期末 ≠ 期初 + 入职 − 离职 → 400 拒绝写入（台账零新增）"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    resp = client.post("/api/cma/ext-ledger/hr-monthly", headers=auth_header(token), json={
+        "period": "2026-08", "headcount_begin": 100, "hires": 2, "separations": 1,
+        "headcount_end": 99,
+    })
+    assert resp.status_code == 400
+    assert "等式不成立" in resp.json()["detail"]
+    assert len(_rows(db, HrMonthlyLedger)) == 0
+
+    ok = client.post("/api/cma/ext-ledger/hr-monthly", headers=auth_header(token), json={
+        "period": "2026-08", "headcount_begin": 100, "hires": 2, "separations": 1,
+        "headcount_end": 101,
+    })
+    assert ok.status_code == 201 and ok.json()["data"]["turnover_rate"] == 1.0
+
+
+def test_licence_invalid_status_rejected(client, db):
+    """证照年检状态白名单校验：非法值 → 400（枚举来自 PRD §4.0）"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    bad = client.post("/api/cma/ext-ledger/licences", headers=auth_header(token), json={
+        "licence_name": "营业执照", "licence_no": "X1", "annual_check_status": "有效"})
+    assert bad.status_code == 400 and "annual_check_status" in bad.json()["detail"]
+    assert len(_rows(db, LicenceLedger)) == 0
+
+
+def test_cross_entity_403(client, db):
+    """⑥ 多租户：会话账套 entity 1 + 显式 entity_id=2 → 403（文案同 deps.get_entity_id）"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    resp = client.post("/api/cma/ext-ledger/hr-monthly?entity_id=2", headers=auth_header(token), json={
+        "period": "2026-08", "headcount_begin": 10, "hires": 0, "separations": 0, "headcount_end": 10})
+    assert resp.status_code == 403
+    assert "无权访问企业 entity_id=2" in resp.json()["detail"]
+    assert len(_rows(db, HrMonthlyLedger)) == 0
+
+
+def test_unauthenticated_401(client):
+    """未认证 → 401（router 级 require_auth）"""
+    assert client.get("/api/cma/ext-ledger/options").status_code == 401
+    assert client.post("/api/cma/ext-ledger/derive", json={"period": "2026-08"}).status_code == 401
+
+
+# ============================================================
+# CSV 批量导入 + 查询 + 幂等
+# ============================================================
+
+def test_csv_import_tax_idempotent(client, db):
+    """CSV 导入两次：第一次 inserted=3，第二次 inserted=0/skipped=3，行数不变"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    r1 = client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token),
+                     json={"csv": TAX_CSV})
+    assert r1.status_code == 200, r1.text
+    assert (r1.json()["inserted"], r1.json()["skipped"]) == (3, 0)
+    r2 = client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token),
+                     json={"csv": TAX_CSV})
+    assert (r2.json()["inserted"], r2.json()["skipped"]) == (0, 3)
+    rows = _rows(db, TaxFilingLog, period="2026-08")
+    assert len(rows) == 3
+    assert {r.on_time for r in rows} == {1, 0}  # 印花税 08-20 > 08-15 → 0
+
+    # 按 (entity_id, period) 查询 + 及时率摘要（2/3 = 66.7）
+    q = client.get("/api/cma/ext-ledger/tax-filings?period=2026-08", headers=auth_header(token))
+    assert q.status_code == 200
+    assert q.json()["total"] == 3
+    assert q.json()["summary"]["tax_timely_rate"] == 66.7
+
+
+def test_csv_import_hr_reports_bad_rows(client, db):
+    """批量导入：等式不成立的行被拒绝并在 errors 回报，合法行照常写入"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    bad_csv = HR_CSV + "2026-06,50,3,1,51\n"
+    resp = client.post("/api/cma/ext-ledger/hr-monthly/import", headers=auth_header(token),
+                       json={"csv": bad_csv})
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["inserted"] == 2 and body["error_count"] == 1
+    assert "等式不成立" in body["errors"][0]["reason"]
+    assert body["errors"][0]["line"] == 4
+    assert len(_rows(db, HrMonthlyLedger)) == 2
+
+    q = client.get("/api/cma/ext-ledger/hr-monthly?period=2026-08", headers=auth_header(token))
+    assert q.json()["total"] == 1 and q.json()["data"][0]["turnover_rate"] == 2.0
+
+
+def test_csv_import_dry_run_writes_nothing(client, db):
+    create_test_user(db)
+    token = get_token_for_user(client)
+    resp = client.post("/api/cma/ext-ledger/licences/import", headers=auth_header(token),
+                       json={"csv": LIC_CSV, "dry_run": True})
+    assert resp.status_code == 200 and resp.json()["inserted"] == 2
+    assert len(_rows(db, LicenceLedger)) == 0
+
+
+def test_licences_query_valid_only(client, db):
+    """③ 已过期证照不计入有效数（查询侧 valid_only 与摘要口径一致）"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    assert client.post("/api/cma/ext-ledger/licences/import", headers=auth_header(token),
+                       json={"csv": LIC_CSV}).json()["inserted"] == 2
+    allq = client.get("/api/cma/ext-ledger/licences?as_of=2026-06-30", headers=auth_header(token)).json()
+    assert allq["total"] == 2
+    assert allq["summary"]["licence_ok"] == 50.0          # 1/2
+    assert allq["summary"]["valid"] == 1
+    only = client.get("/api/cma/ext-ledger/licences?as_of=2026-06-30&valid_only=true",
+                      headers=auth_header(token)).json()
+    assert only["total"] == 1 and only["data"][0]["licence_no"] == "91610100MA0001"
+
+
+def test_options_endpoint_lists_enums(client, db):
+    """表单下拉选项从系统数据动态取（禁前端硬编码）"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    body = client.get("/api/cma/ext-ledger/options", headers=auth_header(token)).json()
+    assert body["annual_check_status"] == ["未到期", "已年检", "待年检", "过期"]
+    assert set(body["derivable_kpi_codes"]) == {"P_TAX_TIMELY", "L_EMPLOYEE_TURNOVER", "P_LICENCE_OK"}
+
+
+# ============================================================
+# 派生：幂等 / 分母 0 不写 / 季频归集 / 定义缺失跳过
+# ============================================================
+
+def _seed_defs(db, entity_id=1):
+    return {
+        "P_TAX_TIMELY": create_test_kpi(db, entity_id=entity_id, kpi_code="P_TAX_TIMELY", kpi_name="税务申报及时率"),
+        "L_EMPLOYEE_TURNOVER": create_test_kpi(db, entity_id=entity_id, kpi_code="L_EMPLOYEE_TURNOVER", kpi_name="员工流失率"),
+        "P_LICENCE_OK": create_test_kpi(db, entity_id=entity_id, kpi_code="P_LICENCE_OK", kpi_name="经营许可合规率"),
+    }
+
+
+def test_derive_idempotent_and_denominator_zero(client, db):
+    """④⑤ 派生跑两次：kpi_values 行数不变（第二次全 upsert）；期初为 0 → 无定义不写值"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    _seed_defs(db)
+    # 台账：税务 2 条（1 按时）；人事期初 0（→ 无定义）；证照 2 条（1 过期）
+    client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={
+        "csv": "period,tax_type,legal_deadline,filed_date\n"
+               "2026-08,增值税,2026-08-15,2026-08-12\n"
+               "2026-08,附加税,2026-08-15,2026-08-20\n"})
+    client.post("/api/cma/ext-ledger/hr-monthly/import", headers=auth_header(token), json={
+        "csv": "period,headcount_begin,hires,separations,headcount_end\n2026-08,0,0,0,0\n"})
+    assert client.post("/api/cma/ext-ledger/licences/import", headers=auth_header(token),
+                       json={"csv": LIC_CSV}).json()["inserted"] == 2
+
+    first = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token),
+                        json={"period": "2026-08"}).json()
+    assert first["items"]["P_TAX_TIMELY"]["actual_value"] == 50.0
+    assert first["items"]["P_TAX_TIMELY"]["action"] == "insert"
+    assert first["items"]["L_EMPLOYEE_TURNOVER"]["actual_value"] is None
+    assert first["items"]["L_EMPLOYEE_TURNOVER"]["action"] == "skipped_denominator_zero"
+    assert first["items"]["P_LICENCE_OK"]["actual_value"] == 50.0
+    assert first["items"]["P_LICENCE_OK"]["period"] == "2026-Q3"   # 2026-08 → Q3
+    n1 = db.query(KPIValue).count()
+    assert n1 == 2                                                  # 无定义指标零写入
+
+    second = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token),
+                         json={"period": "2026-08"}).json()
+    assert second["items"]["P_TAX_TIMELY"]["action"] == "upsert"
+    assert second["items"]["P_LICENCE_OK"]["action"] == "upsert"
+    assert db.query(KPIValue).count() == n1                         # 幂等：行数不变
+    tax_row = _rows(db, KPIValue, period="2026-08")[0]
+    assert (tax_row.actual_value, tax_row.source_type, tax_row.entity_id) == (50.0, "derived", 1)
+
+
+def test_derive_skips_missing_definition(client, db):
+    """不新建/不改动 KPI 定义：定义缺失 → skipped_no_definition（不写值）"""
+    create_test_user(db)
+    token = get_token_for_user(client)
+    create_test_kpi(db, entity_id=1, kpi_code="P_TAX_TIMELY")
+    client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={
+        "csv": "period,tax_type,legal_deadline,filed_date\n2026-08,增值税,2026-08-15,2026-08-12\n"})
+    body = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token),
+                       json={"period": "2026-08"}).json()
+    assert body["items"]["P_TAX_TIMELY"]["action"] == "insert"
+    assert body["items"]["L_EMPLOYEE_TURNOVER"]["action"] == "skipped_no_definition"
+    assert body["items"]["P_LICENCE_OK"]["action"] == "skipped_no_definition"
+    assert db.query(KPIValue).count() == 1
+    assert db.query(KPIDefinition).count() == 1        # 定义数量未变（本模块不新建定义）
+
+
+def test_derive_dry_run_writes_nothing(client, db):
+    create_test_user(db)
+    token = get_token_for_user(client)
+    _seed_defs(db)
+    client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={
+        "csv": "period,tax_type,legal_deadline,filed_date\n2026-08,增值税,2026-08-15,2026-08-12\n"})
+    body = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token),
+                       json={"period": "2026-08", "kpi_codes": ["P_TAX_TIMELY"], "dry_run": True}).json()
+    assert body["items"]["P_TAX_TIMELY"]["action"] == "dry-run"
+    assert db.query(KPIValue).count() == 0
+
+
+def test_derive_status_readonly(client, db):
+    create_test_user(db)
+    token = get_token_for_user(client)
+    _seed_defs(db)
+    body = client.get("/api/cma/ext-ledger/derive/status?period=2026-06",
+                      headers=auth_header(token)).json()
+    assert body["items"]["P_LICENCE_OK"]["target_period"] == "2026-Q2"
+    assert body["items"]["P_LICENCE_OK"]["latest_derived"] is None