feat(calc): 单笔回款核算引擎与绩效台账

- calc_service.calc_receipt：锁回款行→settlement_status=0 校验→发货批次/
  归属前置→resolve_snapshot(回款日)→按 calc_template 分派（与 _preview_rate
  同构，‰÷1000 防线）→items 行级折算（行类别阈值，混合按行汇总）→
  barter/起付/R→台账 v1→CAS 置已核算，单 audit 事务
- 策略点集中：_personal_rate（#12 经理个人户单笔落 0 待答复）、_row_profit
  （#20 不含税原样）；核算粒度 #25 一期口径=业务级一次性计提（重复回款
  repeated=True 仅登记进度）
- 台账仓储：生效行查询/业务级判定/版本号；schemas LedgerOut/Query
- api /api/v1/calc/*：台账分页/详情/核算触发（写限 admin+marketing_ops）
- 测试 8 例：金标准双类别复算到分（DB 锚定 4000.00/39.11/39.11）、
  一次性计提、已核算 400、未指定批次、成本缺失行保守 -3000、负绩效照算、
  准入、列表筛选；行级明细折算两位固化
- conftest 清理顺序修正：V7 台账/月结清理移到回款/人员删除之前
  （RESTRICT 外键不挡序）

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/app/api/router.py

@@ -15,6 +15,7 @@ from app.api.v1.business import businesses as business_businesses
 from app.api.v1.business import ratings as business_ratings
 from app.api.v1.business import receipts as business_receipts
 from app.api.v1.business import shipments as business_shipments
+from app.api.v1.calc import router as calc_router
 from app.api.v1.system import approvals as system_approvals
 from app.api.v1.system import business_roles as system_business_roles
 from app.api.v1.system import logs as system_logs
@@ -59,4 +60,7 @@ business_router.include_router(business_receipts.router)
 business_router.include_router(business_ratings.router)
 v1_router.include_router(business_router)
 
+# 绩效核算（M3 ④）：/api/v1/calc/*，读登录即可、写限管理员与营销后勤
+v1_router.include_router(calc_router)
+
 api_router.include_router(v1_router)

backend/app/api/v1/calc.py

@@ -0,0 +1,57 @@
+"""绩效核算接口（M3 ④）：台账查询与单笔核算触发。
+
+核算动作 = 把回款固化为台账行（每笔回款仅一次；重算走独立入口并授权）。
+准入：读=登录即可；写（核算/重算/批量）=管理员与营销后勤
+（权限清单待澄清 #6，答复后细化）。
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from app.api.deps import DbSession, RequestMetaDep, require_roles
+from app.core.constants import ROLE_ADMIN, ROLE_MARKETING_OPS
+from app.core.response import ApiResponse, PageResult, ok
+from app.models import User
+from app.schemas.calc import CalcReceiptOut, LedgerOut, LedgerQuery
+from app.services import calc_service
+
+router = APIRouter(prefix="/calc", tags=["绩效工资 · 绩效核算"])
+
+_CalcWriter = Annotated[User, Depends(require_roles(ROLE_ADMIN, ROLE_MARKETING_OPS))]
+
+
+@router.get("/ledger", response_model=ApiResponse[PageResult[LedgerOut]], summary="绩效台账")
+def list_ledger(
+    session: DbSession,
+    query: Annotated[LedgerQuery, Query()],
+) -> ApiResponse[PageResult[LedgerOut]]:
+    """分页查询绩效核算台账（默认仅生效行，可筛选历史版本）。"""
+    return ok(calc_service.list_ledger_page(session, query))
+
+
+@router.get("/ledger/{ledger_id}", response_model=ApiResponse[LedgerOut], summary="台账详情")
+def get_ledger(
+    session: DbSession,
+    ledger_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[LedgerOut]:
+    """台账行详情（含计算快照）。"""
+    return ok(calc_service.get_ledger(session, ledger_id))
+
+
+@router.post(
+    "/receipts/{receipt_id}/calc",
+    response_model=ApiResponse[CalcReceiptOut],
+    summary="核算回款",
+)
+def calc_receipt(
+    receipt_id: Annotated[int, Path(ge=1)],
+    session: DbSession,
+    current_user: _CalcWriter,
+    meta: RequestMetaDep,
+) -> ApiResponse[CalcReceiptOut]:
+    """把回款固化为台账行（每笔回款仅一次；业务级一次性计提，重复回款仅登记进度）。"""
+    return ok(
+        calc_service.calc_receipt(session, receipt_id, operator=current_user, meta=meta),
+        message="核算完成",
+    )

backend/app/repositories/performance_ledger_repository.py

@@ -0,0 +1,114 @@
+"""绩效核算台账仓储：只读写数据并 flush；commit/rollback 在服务层。"""
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.models import PerformanceLedger
+from app.schemas.calc import LedgerQuery
+
+
+def get_by_id(session: Session, ledger_id: int) -> PerformanceLedger | None:
+    """按主键取台账行。"""
+    return session.get(PerformanceLedger, ledger_id)
+
+
+def get_active_by_receipt(
+    session: Session, receipt_id: int, *, calc_type: int = 1
+) -> PerformanceLedger | None:
+    """取某回款的当前生效核算行（is_active=1；无则 None）。"""
+    return session.execute(
+        select(PerformanceLedger).where(
+            PerformanceLedger.receipt_id == receipt_id,
+            PerformanceLedger.calc_type == calc_type,
+            PerformanceLedger.is_active == 1,
+        )
+    ).scalar_one_or_none()
+
+
+def has_active_for_business(session: Session, business_id: int, *, calc_type: int = 1) -> bool:
+    """该业务是否已有生效核算行（核算粒度 #25：业务级一次性计提的判定）。"""
+    return (
+        session.execute(
+            select(func.count())
+            .select_from(PerformanceLedger)
+            .where(
+                PerformanceLedger.business_id == business_id,
+                PerformanceLedger.calc_type == calc_type,
+                PerformanceLedger.is_active == 1,
+            )
+        ).scalar_one()
+        > 0
+    )
+
+
+def get_active_by_receipt_of_business(
+    session: Session, business_id: int, *, calc_type: int = 1
+) -> PerformanceLedger:
+    """取该业务当前生效的核算行（重复回款登记进度时回显既有台账）。"""
+    return (
+        session.execute(
+            select(PerformanceLedger)
+            .where(
+                PerformanceLedger.business_id == business_id,
+                PerformanceLedger.calc_type == calc_type,
+                PerformanceLedger.is_active == 1,
+            )
+            .order_by(PerformanceLedger.id.desc())
+        )
+        .scalars()
+        .first()
+    )
+
+
+def add(session: Session, row: PerformanceLedger) -> PerformanceLedger:
+    """落台账行（flush 取 id；commit 在服务层）。"""
+    session.add(row)
+    session.flush()
+    return row
+
+
+def mark_inactive(session: Session, row: PerformanceLedger) -> None:
+    """重算时把旧版本行翻为已替代（全表唯一的受控 UPDATE 例外；不 commit）。"""
+    row.is_active = 0
+    session.flush()
+
+
+def next_version(session: Session, receipt_id: int) -> int:
+    """某回款的下一个核算版本号（max+1；无行则 1）。"""
+    current = session.execute(
+        select(func.max(PerformanceLedger.calc_version)).where(
+            PerformanceLedger.receipt_id == receipt_id
+        )
+    ).scalar_one()
+    return int(current) + 1
+
+
+def list_ledger(session: Session, query: LedgerQuery) -> list[PerformanceLedger]:
+    """分页查询台账（筛选：月/人员/业务/生效/类型）。"""
+    stmt = select(PerformanceLedger).where(*_filters(query))
+    stmt = stmt.order_by(PerformanceLedger.id.desc()).offset(query.offset).limit(query.page_size)
+    return list(session.execute(stmt).scalars())
+
+
+def count_ledger(session: Session, query: LedgerQuery) -> int:
+    """台账计数（与 list_ledger 同筛选）。"""
+    return int(
+        session.execute(
+            select(func.count()).select_from(PerformanceLedger).where(*_filters(query))
+        ).scalar_one()
+    )
+
+
+def _filters(query: LedgerQuery) -> tuple:
+    filters = []
+    if query.settlement_month:
+        filters.append(PerformanceLedger.settlement_month == query.settlement_month)
+    if query.personnel_id is not None:
+        filters.append(PerformanceLedger.personnel_id == query.personnel_id)
+    if query.business_id is not None:
+        filters.append(PerformanceLedger.business_id == query.business_id)
+    if query.is_active is not None:
+        filters.append(PerformanceLedger.is_active == query.is_active)
+    if query.calc_type is not None:
+        filters.append(PerformanceLedger.calc_type == query.calc_type)
+    return tuple(filters)

backend/app/schemas/calc.py

@@ -0,0 +1,62 @@
+"""绩效核算的请求/响应模型（M3 ④）。"""
+
+from datetime import date, datetime
+from decimal import Decimal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.core.response import PageQuery
+
+
+class LedgerQuery(PageQuery):
+    """台账列表查询条件。"""
+
+    settlement_month: str | None = Field(default=None, max_length=7, description="归属月 YYYY-MM")
+    personnel_id: int | None = Field(default=None, ge=1, description="归属人员筛选")
+    business_id: int | None = Field(default=None, ge=1, description="业务筛选")
+    is_active: int | None = Field(default=None, ge=0, le=1, description="生效标记筛选")
+    calc_type: int | None = Field(default=None, ge=1, le=2, description="核算类型筛选")
+
+
+class LedgerOut(BaseModel):
+    """台账行响应（金额/比例一律 str）。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    receipt_id: int | None = None
+    calc_version: int
+    is_active: int
+    calc_type: int
+    business_id: int
+    business_no: str
+    personnel_id: int
+    personnel_name: str
+    role_code: str
+    customer_type: str
+    product_category: str
+    ship_date: date | None = None
+    receipt_date: date
+    payment_days: int | None = None
+    rate_text: str
+    converted_profit: Decimal | None = None
+    performance_amount: Decimal | None = None
+    barter_applied: int
+    weight_r: int | None = None
+    payment_weight: Decimal | None = None
+    payable_amount: Decimal | None = None
+    settlement_month: str
+    is_probation: int
+    missing_cost: int
+    snapshot_json: dict | None = None
+    settlement_id: int | None = None
+    calc_reason: str | None = None
+    create_time: datetime
+
+
+class CalcReceiptOut(BaseModel):
+    """单笔核算结果：台账行 + 重复计提标记（核算粒度 #25 一期口径）。"""
+
+    ledger: LedgerOut
+    #: True=该业务此前已核算过，本笔回款仅登记进度、未重复计提（#25 一期口径）
+    repeated: bool = False

backend/app/services/calc_service.py

@@ -0,0 +1,391 @@
+"""绩效核算服务（M3 ④）：回款 → 台账的固化流水线。
+
+单笔核算（calc_receipt）与计算预览（param_service.preview）**共用同一
+公式路径与取值防线**：resolve_snapshot 按回款到账日取参（裁决 #11）、按
+岗位模板分派计提比例（‰ 存储口径 ÷1000 转纯比例）、行级折算（混合业务
+按行汇总，对照表 #14）。两者的任何口径修改必须同步——预览给业务对数、
+核算才是算钱的法定固化点。
+
+策略点（待业务答复，全部集中为单函数，答复后只改函数 + 按回款批量重算）：
+- ``_personal_rate``：#12 经理个人户单笔落台账 rate=0/绩效 0（12.5%+5%
+  聚合口径在月结时由 manager_team_share 计算）；#20 在 ``_row_profit``
+- 核算粒度 #25：业务级一次性计提——首个触发核算的回款计算全额应发，
+  后续回款仅置 settlement_status=1（repeated=True），不重复计提
+"""
+
+import logging
+from dataclasses import dataclass
+from decimal import Decimal
+from fractions import Fraction
+from typing import Any
+
+from sqlalchemy import update
+from sqlalchemy.orm import Session
+
+from app.core.constants import MODULE_CALC
+from app.core.exceptions import BadRequestError
+from app.models import (
+    Business,
+    BusinessItem,
+    Customer,
+    PerformanceLedger,
+    Receipt,
+    Shipment,
+    User,
+)
+from app.models.performance_ledger import CALC_TYPE_RECEIPT
+from app.models.receipt import SETTLEMENT_DONE, SETTLEMENT_PENDING
+from app.repositories import business_repository, performance_ledger_repository
+from app.schemas.calc import CalcReceiptOut
+from app.services import audit_service, param_service, personnel_service
+from app.services.audit_service import AuditRecord, RequestMeta
+from app.services.formula import (
+    apply_barter,
+    clamp_weight_r,
+    converted_profit,
+    payment_days,
+    payment_weight,
+    performance_amount,
+    profit_rate,
+    salesperson_rate,
+    start_payment_ok,
+)
+from app.services.param_service import parse_fraction, resolve_snapshot
+
+logger = logging.getLogger(__name__)
+
+# snapshot_json 里的策略点注记（答复回填时的检索锚点）
+_NOTE_MANAGER_PENDING = "经理个人户单笔不计提（#12 待答复）：聚合口径在月结时由团队聚合计算"
+
+
+@dataclass(frozen=True, slots=True)
+class _RateDecision:
+    """单笔计提比例的分派结果。"""
+
+    rate: Fraction  # 纯比例（str() 产出分数原文如 88/9000，直接落台账 rate_text）
+    rate_note: str  # 人读说明（入 snapshot）
+    base_override: Decimal | None  # 非 None 时绩效基数用它（信息员=销售额）
+    manager_pending: bool  # True=经理聚合口径，单笔落 0（#12 策略点）
+
+
+def calc_receipt(
+    session: Session,
+    receipt_id: int,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> CalcReceiptOut:
+    """单笔回款核算：回款 → 发货批次 → 业务 → 明细行逐行折算 → 台账固化。
+
+    防重前两层在本函数（FOR UPDATE 锁行 + settlement_status=0 校验），
+    第三层是 CAS UPDATE（rowcount=0 即并发撞车 400）。全程一个 audit
+    事务（MODULE_CALC），出参在 commit 前固化。
+
+    :raises BadRequestError: 已核算 / 未指定发货批次 / 归属未指定 /
+        无明细行 / 并发撞车
+    """
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_CALC,
+            action="核算",
+            summary=f"核算回款 #{receipt_id}",
+            request_data={"receipt_id": receipt_id},
+        ),
+        meta=meta,
+    ):
+        receipt = session.get(Receipt, receipt_id, with_for_update=True)
+        if receipt is None:
+            raise BadRequestError("回款不存在")
+        if receipt.settlement_status == SETTLEMENT_DONE:
+            raise BadRequestError("该回款已核算，禁止重复核算（重算请走重算入口）")
+
+        business = session.get(Business, receipt.business_id, with_for_update=True)
+        if business is None:
+            raise BadRequestError("回款关联的业务不存在")
+        if business.owner_personnel_id is None:
+            raise BadRequestError("业务未指定负责营销人员，无法核算（先在业务归属中指定）")
+
+        # 核算粒度 #25（一期口径）：业务级一次性计提——已有生效台账行时，
+        # 本笔回款仅置已核算状态（支付进度），不重复计提
+        repeated = performance_ledger_repository.has_active_for_business(
+            session, business.id, calc_type=CALC_TYPE_RECEIPT
+        )
+        if repeated:
+            _mark_settled(session, receipt)
+            row = performance_ledger_repository.get_active_by_receipt_of_business(
+                session, business.id
+            )
+            session.flush()
+            out = CalcReceiptOut(ledger=_to_out(row), repeated=True)
+            session.commit()
+            logger.info(
+                "回款登记进度（业务已核算）| receipt=%s | business=%s", receipt_id, business.id
+            )
+            return out
+
+        if receipt.shipment_id is None:
+            raise BadRequestError("回款未指定发货批次，无法确定发货日期（先指定批次再核算）")
+        shipment = session.get(Shipment, receipt.shipment_id)
+        if shipment is None or shipment.ship_date is None:
+            raise BadRequestError("发货批次缺少发货日期，无法核算")
+
+        person = personnel_service.ensure_active(
+            session, business.owner_personnel_id, label="核算归属"
+        )
+        if person.business_role_id is None:
+            raise BadRequestError("归属人员未绑定岗位角色，无法确定计提公式")
+
+        receipt_date = receipt.receipt_date
+        snapshot = resolve_snapshot(session, receipt_date)
+        items = business_repository.list_items(session, business.id)
+        if not items:
+            raise BadRequestError("业务无产品明细行，无法核算")
+
+        # 逐行折算（行级 product_category 阈值；成本缺失行毛利按 0 计并标记）
+        total_revenue = Decimal("0")
+        total_profit = Decimal("0")
+        converted_total = Decimal("0")
+        missing_cost = False
+        line_details: list[dict[str, Any]] = []
+        for item in items:
+            row_revenue = item.sale_amount
+            if item.cost_amount is None:
+                missing_cost = True
+                row_profit = Decimal("0")
+            else:
+                row_profit = _row_profit(row_revenue, item.cost_amount)
+            total_revenue += row_revenue
+            total_profit += row_profit
+            threshold = parse_fraction(
+                snapshot.get(
+                    "product.threshold",
+                    role_code=role_code_of(person),
+                    product_category=item.product_category,
+                )
+            )
+            row_converted = converted_profit(
+                row_revenue, profit_rate(row_profit, row_revenue), threshold
+            )
+            converted_total += row_converted
+            line_details.append(
+                {
+                    "item_id": item.id,
+                    "product_category": item.product_category,
+                    "revenue": str(row_revenue),
+                    "profit": str(row_profit),
+                    # 行级折算同样固化为两位（to_decimal 对整数值无小数位，
+                    # 快照/台账口径需与列一致）
+                    "converted": str(row_converted.quantize(Decimal("0.01"))),
+                }
+            )
+
+        # 天数只依赖发货/回款日期，先于比例分派算出（业务员递减公式的输入）
+        days = payment_days(shipment.ship_date, receipt_date)
+        decision = _personal_rate(session, snapshot, person, business, days)
+
+        if decision.base_override is not None:
+            converted = decision.base_override  # 信息员：基数=本人销售额
+        else:
+            converted = converted_total
+
+        performance = (
+            Decimal("0")
+            if decision.manager_pending
+            else performance_amount(converted, decision.rate)
+        )
+        barter_applied = receipt.is_barter == 1
+        base_amount = performance
+        if barter_applied:
+            discount = parse_fraction(
+                snapshot.get("barter.discount", role_code=role_code_of(person))
+            )
+            base_amount = apply_barter(performance, discount)
+
+        received_total = business.total_received_amount or Decimal("0")
+        start_ratio = parse_fraction(
+            snapshot.get("payment.start.ratio", role_code=role_code_of(person))
+        )
+        start_ok = start_payment_ok(received_total, total_revenue, start_ratio)
+        weight_r = clamp_weight_r(received_total, total_revenue)
+        weight = payment_weight(weight_r)
+        payable = param_service.to_decimal(Fraction(base_amount) * weight)
+
+        settlement_month = receipt_date.strftime("%Y-%m")
+        snapshot_json: dict[str, Any] = {
+            "params_as_of": str(receipt_date),
+            "line_details": line_details,
+            "start_payment_ok": start_ok,
+            "rate_note": decision.rate_note,
+            "revenue_total": str(total_revenue),
+            "profit_total": str(total_profit),
+        }
+        if decision.manager_pending:
+            snapshot_json["strategy_note"] = _NOTE_MANAGER_PENDING
+
+        def q2(value: Decimal | None) -> Decimal | None:
+            """金额统一两位小数（HALF_UP）：内存/DB/序列化三方一致。"""
+            return value.quantize(Decimal("0.01")) if value is not None else None
+
+        row = performance_ledger_repository.add(
+            session,
+            PerformanceLedger(
+                receipt_id=receipt.id,
+                calc_type=CALC_TYPE_RECEIPT,
+                business_id=business.id,
+                business_no=business.business_no,
+                personnel_id=person.id,
+                personnel_name=person.personnel_name,
+                role_code=role_code_of(person),
+                customer_type=_customer_type(session, business),
+                product_category=_category_label(items),
+                ship_date=shipment.ship_date,
+                receipt_date=receipt_date,
+                payment_days=days,
+                rate_text=str(decision.rate),
+                converted_profit=q2(converted),
+                performance_amount=q2(performance),
+                barter_applied=1 if barter_applied else 0,
+                weight_r=weight_r,
+                payment_weight=param_service.to_decimal(weight),
+                payable_amount=q2(payable),
+                settlement_month=settlement_month,
+                is_probation=1 if person.is_probation == 1 else 0,
+                missing_cost=1 if missing_cost else 0,
+                snapshot_json=snapshot_json,
+            ),
+        )
+        _mark_settled(session, receipt)
+        session.flush()
+        out = CalcReceiptOut(ledger=_to_out(row), repeated=False)
+        session.commit()
+        logger.info("核算完成 | receipt=%s | ledger=%s | payable=%s", receipt_id, row.id, payable)
+        return out
+
+
+def _personal_rate(
+    session: Session,
+    snapshot,
+    person,
+    business: Business,
+    days: int,
+) -> _RateDecision:
+    """按岗位模板分派单笔计提比例（与 param_service._preview_rate 同构）。
+
+    #12 策略点：经理个人户（owner 岗位绑 TPL_MANAGER_SHARE）单笔不计提
+    —— 12.5%+5% 是月度团队聚合口径，单笔落 rate=0/绩效 0 的台账行保事实。
+    """
+    role = _role_of(session, person)
+    customer_type = _customer_type(session, business)
+
+    if customer_type == "enterprise":
+        # 存储口径 ‰ → 纯比例（漏换算放大 1000 倍，与预览同防线）
+        rate = parse_fraction(snapshot.get("enterprise.rate")) / 1000
+        return _RateDecision(rate, "企业户固定计提比例", None, False)
+
+    template = role.calc_template if role else "TPL_SALESPERSON_DECAY"
+    if template == "TPL_MANAGER_SHARE":
+        return _RateDecision(
+            Decimal("0"),
+            "经理个人户：12.5%+5% 聚合计提（月结时计算），单笔落 0",
+            None,
+            True,
+        )
+    if template == "TPL_INFO_SALES_FIXED":
+        # 信息员基数=本人销售额（覆盖折算与阈值），‰ 换算同防线
+        rate = parse_fraction(snapshot.get("info.rate")) / 1000
+        return _RateDecision(rate, "信息员：本人销售额 × 计提比例", None, False)
+    if template == "TPL_SALESPERSON_DECAY":
+        start = parse_fraction(snapshot.get("salesperson.rate.start"))
+        slope = parse_fraction(snapshot.get("salesperson.rate.slope"))
+        floor = parse_fraction(snapshot.get("salesperson.rate.floor"))
+        boundary = int(snapshot.get("salesperson.days.boundary"))
+        rate = salesperson_rate(days, start=start, slope=slope, floor=floor, boundary_days=boundary)
+        note = f"业务员递减：{start}‰ 起，{days} 天（分界 {boundary} 天，超界取 {floor}‰）"
+        return _RateDecision(rate, note, None, False)
+    raise BadRequestError(f"公式模板 {template} 暂不支持核算")
+
+
+def _role_of(session: Session, person):
+    from app.repositories import business_role_repository
+
+    if person.business_role_id is None:
+        return None
+    return business_role_repository.get_by_id(session, person.business_role_id)
+
+
+def role_code_of(person) -> str:
+    """人员岗位编码（台账冗余固化用；无岗位为空串——前置校验已拦）。"""
+    from app.repositories import business_role_repository
+
+    if person.business_role_id is None:
+        return ""
+    session = Session.object_session(person)
+    role = business_role_repository.get_by_id(session, person.business_role_id)
+    return role.role_code if role else ""
+
+
+def _customer_type(session: Session, business: Business) -> str:
+    customer = session.get(Customer, business.customer_id)
+    return customer.customer_type if customer else "personal"
+
+
+def _category_label(items: list[BusinessItem]) -> str:
+    """单类别直出；混合业务记 mixed（行级明细在 snapshot_json）。"""
+    categories = {item.product_category for item in items}
+    return categories.pop() if len(categories) == 1 else "mixed"
+
+
+def _mark_settled(session: Session, receipt: Receipt) -> None:
+    """CAS 置已核算：并发下另一事务先行置 1 时 rowcount=0 → 400。"""
+    result = session.execute(
+        update(Receipt)
+        .where(Receipt.id == receipt.id, Receipt.settlement_status == SETTLEMENT_PENDING)
+        .values(settlement_status=SETTLEMENT_DONE)
+    )
+    if result.rowcount == 0:
+        raise BadRequestError("该回款已被并发核算，请刷新后查看台账")
+
+
+def _row_profit(revenue: Decimal, cost: Decimal) -> Decimal:
+    """行毛利 = 行销售额 − 行成本。
+
+    #20 策略点（待业务答复，2026-10-16 暂定口径）：金额暂按**不含税**原样
+    相减；答复若为含税口径，只改本函数（含税换算/取不含税金额列）+ 按回款
+    批量重算回填。
+    """
+    return revenue - cost
+
+
+def list_ledger_page(session: Session, query):
+    """分页查询台账（api 层转发）。"""
+    rows = performance_ledger_repository.list_ledger(session, query)
+    total = performance_ledger_repository.count_ledger(session, query)
+    from app.core.response import PageResult
+    from app.schemas.calc import LedgerOut
+
+    return PageResult[LedgerOut](
+        records=[LedgerOut.model_validate(r) for r in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def get_ledger(session: Session, ledger_id: int):
+    """台账详情（api 层转发）。"""
+    row = performance_ledger_repository.get_by_id(session, ledger_id)
+    if row is None:
+        from app.core.exceptions import NotFoundError
+
+        raise NotFoundError("台账记录不存在")
+    from app.schemas.calc import LedgerOut
+
+    return LedgerOut.model_validate(row)
+
+
+def _to_out(row: PerformanceLedger):
+    from app.schemas.calc import LedgerOut
+
+    return LedgerOut.model_validate(row)

backend/tests/integration/conftest.py

@@ -101,6 +101,14 @@ def _purge_test_data() -> None:
             session.execute(delete(UserRole).where(UserRole.role_id.in_(role_ids)))
             session.execute(delete(Role).where(Role.id.in_(role_ids)))
 
+        # ---- M3-core：核算台账 / 月结 / 调整单（必须先于回款/业务/人员删除——
+        # ledger.receipt_id/personnel_id/settlement_id 外键 RESTRICT 会挡住后面
+        # 的单据与人员清理；仅测试写入，全量清）----
+        session.execute(delete(PerformanceLedger))
+        session.execute(delete(SettlementItem))
+        session.execute(delete(MonthlySettlement))
+        session.execute(delete(AdjustmentOrder).where(AdjustmentOrder.order_no.like("TZ%")))
+
         # ---- M2 业务数据（T11）：业务/发货/回款三类单据 ----
         # 编号系统生成、无 t_ 前缀可用，按业务编号前缀全量清（三类单据表目前
         # 只有测试会写）。外键均为 RESTRICT，删除顺序不能倒：回款 → 发货
@@ -182,12 +190,6 @@ def _purge_test_data() -> None:
         # 审批单无 t_ 前缀可用，按 AP 编号前缀全量清（steps 外键 CASCADE 随删；
         # ref_id 固化无外键，业务已在上一步删除，无悬挂影响）
         session.execute(delete(ApprovalOrder).where(ApprovalOrder.order_no.like("AP%")))
-        # ---- M3-core：核算台账 / 月结 / 调整单（仅测试写入，全量清；
-        # 删除顺序：台账→明细→月结头→调整单，TZ 前缀全量清）----
-        session.execute(delete(PerformanceLedger))
-        session.execute(delete(SettlementItem))
-        session.execute(delete(MonthlySettlement))
-        session.execute(delete(AdjustmentOrder).where(AdjustmentOrder.order_no.like("TZ%")))
         # 开关参数恢复种子基线：测试提交的变更批次条目 id 均大于种子终值 22
         # （V3 十九条 + V6 两条 + V7 调整单开关一条），删新条目与新批次后解析
         # 回退到种子值（开关=0）

backend/tests/integration/test_calc.py

@@ -0,0 +1,370 @@
+"""单笔回款核算集成测试（M3 ④：金标准复算到分）。
+
+金标准场景（双类别混合业务，覆盖行级折算与汇总）：
+- 行1 市政水利：100×100.00 = 销售 10000，成本 3000 → 毛利率 0.7，阈值 30%
+  → 折算 10000×(0.7−0.3) = 4000
+- 行2 轨道交通：50×200.00 = 销售 10000，成本 8000 → 毛利率 0.2，阈值 20%
+  → 折算 10000×(0.2−0.2) = 0
+- 折算合计 4000；发货 10-01 → 回款 10-20，days=20
+- 率 = (10 − 20×4/360)‰ = 88/9000；绩效 = 4000×88/9000 ≈ 39.11
+- 回款 20000（=销售额 100%）→ R=100、权重 1 → 应付 = 39.11
+
+守的还有：核算粒度 #25（业务级一次性计提——第二笔回款仅登记进度）、
+已核算再核算 400、未指定批次 400、成本缺失行毛利按 0 计、负绩效照算、
+准入（normal 403 / marketing_ops 可用）。
+"""
+
+import uuid
+from decimal import Decimal
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from app.core.db import get_session_factory
+from app.models import PerformanceLedger, Receipt
+from app.models.receipt import SETTLEMENT_DONE
+from tests.integration.factories import business_payload, customer
+
+pytestmark = pytest.mark.integration
+
+_CALC = "/api/v1/calc"
+_BUSINESSES = "/api/v1/business/businesses"
+_SHIPMENTS = "/api/v1/business/shipments"
+_RECEIPTS = "/api/v1/business/receipts"
+_PERSONNEL = "/api/v1/base/personnel"
+
+
+def _person(client: TestClient, headers: dict) -> dict:
+    resp = client.post(
+        _PERSONNEL,
+        json={
+            "personnel_name": f"t_核算_{uuid.uuid4().hex[:8]}",
+            "business_role_id": _role_id("salesperson"),
+            "salary_standard": "3000.00",
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _role_id(code: str) -> int:
+    from sqlalchemy import select
+
+    from app.core.constants import BUSINESS_ROLE_SALESPERSON
+    from app.core.db import get_session_factory
+    from app.models import BusinessRole
+
+    factory = get_session_factory()
+    with factory() as session:
+        role_code = BUSINESS_ROLE_SALESPERSON if code == "salesperson" else code
+        return int(
+            session.execute(
+                select(BusinessRole.id).where(
+                    BusinessRole.role_code == role_code, BusinessRole.status == 1
+                )
+            ).scalar_one()
+        )
+
+
+def _create_full_chain(
+    client: TestClient,
+    headers: dict,
+    *,
+    items: list[dict],
+    ship_date: str,
+    receipt_date: str,
+    amount: str,
+) -> tuple[dict, dict]:
+    """客户 → 人员 → 业务（双行明细+归属）→ 发货 → 回款（指定批次）。"""
+    cust = customer(client, headers)
+    person = _person(client, headers)
+    resp = client.post(
+        _BUSINESSES,
+        json=business_payload(cust["id"], owner_personnel_id=person["id"], items=items),
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    biz = resp.json()["data"]
+
+    resp = client.post(
+        _SHIPMENTS,
+        json={
+            "business_id": biz["id"],
+            "ship_date": ship_date,
+            "items": [{"product_name": "t_发货物", "quantity": "1", "sale_price": "1.00"}],
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    shipment = resp.json()["data"]
+
+    resp = client.post(
+        _RECEIPTS,
+        json={
+            "business_id": biz["id"],
+            "receipt_date": receipt_date,
+            "amount": amount,
+            "shipment_id": shipment["id"],
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    receipt = resp.json()["data"]
+    return biz, receipt
+
+
+def _calc(client: TestClient, headers: dict, receipt_id: int):
+    return client.post(f"{_CALC}/receipts/{receipt_id}/calc", headers=headers)
+
+
+_GOLD_ITEMS = [
+    {
+        "product_name": "t_市政_金标",
+        "product_category": "municipal_water",
+        "quantity": "100",
+        "sale_price": "100.00",
+        "cost_amount": "3000.00",
+    },
+    {
+        "product_name": "t_轨交_金标",
+        "product_category": "rail_transit",
+        "quantity": "50",
+        "sale_price": "200.00",
+        "cost_amount": "8000.00",
+    },
+]
+
+
+class TestGoldenStandard:
+    """金标准：双类别混合业务复算到分（Decimal 精确断言）。"""
+
+    def test_复算到分_全字段固化(self, client: TestClient, admin: dict) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="20000.00",
+        )
+
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 200, resp.text
+        out = resp.json()["data"]
+        assert out["repeated"] is False
+
+        ledger = out["ledger"]
+        assert ledger["payment_days"] == 20
+        assert ledger["settlement_month"] == "2026-10"
+        # 率 = (10 − 20×4/360)‰ = 88/9000，Fraction 约分原文 11/1125
+        assert ledger["rate_text"] == "11/1125"
+        assert ledger["weight_r"] == 100
+        assert ledger["missing_cost"] == 0
+        assert ledger["snapshot_json"]["line_details"][0]["converted"] == "4000.00"
+
+        # 金额锚定 DB 事实源（NUMERIC(14,2) 读回恒两位；响应 JSON 的 Decimal
+        # 序列化 scale 不作断言——归前端展示问题）
+        factory = get_session_factory()
+        with factory() as session:
+            row = session.execute(
+                select(PerformanceLedger).where(
+                    PerformanceLedger.receipt_id == receipt["id"],
+                    PerformanceLedger.is_active == 1,
+                )
+            ).scalar_one()
+        assert row.converted_profit == Decimal("4000.00")
+        # 绩效 = 4000 × 88/9000 = 39.111… → 落库两位 39.11
+        assert row.performance_amount == Decimal("39.11")
+        # 应付 = 绩效 × 权重 1 = 39.11
+        assert row.payable_amount == Decimal("39.11")
+
+        # 回款状态被 CAS 置已核算
+        detail = client.get(f"{_RECEIPTS}/{receipt['id']}", headers=admin["headers"]).json()["data"]
+        assert detail["settlement_status"] == 1
+
+    def test_已核算回款再核算400(self, client: TestClient, admin: dict) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="100.00",
+        )
+        assert _calc(client, admin["headers"], receipt["id"]).status_code == 200
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 400
+        assert "已核算" in resp.json()["message"]
+
+    def test_业务级一次性计提_第二笔仅登记进度(self, client: TestClient, admin: dict) -> None:
+        """核算粒度 #25 一期口径：同业务第二笔回款不重复计提。"""
+        biz, first = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="10000.00",
+        )
+        assert _calc(client, admin["headers"], first["id"]).status_code == 200
+
+        resp = client.post(
+            _RECEIPTS,
+            json={
+                "business_id": biz["id"],
+                "receipt_date": "2026-10-25",
+                "amount": "5000.00",
+                "shipment_id": None,
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+        second = resp.json()["data"]
+
+        resp = _calc(client, admin["headers"], second["id"])
+        assert resp.status_code == 200, resp.text
+        out = resp.json()["data"]
+        assert out["repeated"] is True
+
+        factory = get_session_factory()
+        with factory() as session:
+            rows = (
+                session.execute(
+                    select(PerformanceLedger).where(
+                        PerformanceLedger.business_id == biz["id"],
+                        PerformanceLedger.is_active == 1,
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert len(rows) == 1, "第二笔回款不得新增台账行（一次性计提）"
+            receipts = (
+                session.execute(select(Receipt).where(Receipt.business_id == biz["id"]))
+                .scalars()
+                .all()
+            )
+            assert all(r.settlement_status == SETTLEMENT_DONE for r in receipts)
+
+
+class TestGuards:
+    """校验链与策略点。"""
+
+    def test_未指定批次400(self, client: TestClient, admin: dict) -> None:
+        cust = customer(client, admin["headers"])
+        person = _person(client, admin["headers"])
+        resp = client.post(
+            _BUSINESSES,
+            json=business_payload(cust["id"], owner_personnel_id=person["id"]),
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200
+        biz = resp.json()["data"]
+        resp = client.post(
+            _RECEIPTS,
+            json={
+                "business_id": biz["id"],
+                "receipt_date": "2026-10-20",
+                "amount": "100.00",
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200
+        receipt = resp.json()["data"]
+
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 400
+        assert "发货批次" in resp.json()["message"]
+
+    def test_成本缺失行_毛利按0计并标记(self, client: TestClient, admin: dict) -> None:
+        items = [
+            {
+                "product_name": "t_无成本_市政",
+                "product_category": "municipal_water",
+                "quantity": "100",
+                "sale_price": "100.00",
+                "cost_amount": None,
+            }
+        ]
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=items,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="10000.00",
+        )
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 200, resp.text
+        ledger = resp.json()["data"]["ledger"]
+        # 成本缺失 → 毛利按 0 → 折算 = 销售额×(0−30%) = **−3000**（保守少算
+        # 方向：缺成本的行宁少算不多发），missing_cost=1
+        assert ledger["missing_cost"] == 1
+        assert float(ledger["converted_profit"]) == -3000.0
+        assert float(ledger["performance_amount"]) == pytest.approx(-29.33, abs=0.01)
+
+    def test_负绩效照算不归零(self, client: TestClient, admin: dict) -> None:
+        items = [
+            {
+                "product_name": "t_亏损_市政",
+                "product_category": "municipal_water",
+                "quantity": "100",
+                "sale_price": "100.00",
+                "cost_amount": "9000.00",
+            }
+        ]
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=items,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="10000.00",
+        )
+        resp = _calc(client, admin["headers"], receipt["id"])
+        assert resp.status_code == 200, resp.text
+        ledger = resp.json()["data"]["ledger"]
+        # 毛利 1000、率 0.1 < 阈值 30% → 折算 10000×(0.1−0.3) = −2000 → 负绩效
+        assert float(ledger["converted_profit"]) == -2000.0
+        assert float(ledger["performance_amount"]) < 0
+
+    def test_准入_normal403_marketing_ops可用(
+        self, client: TestClient, admin: dict, normal: dict, make_user, auth_headers
+    ) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="100.00",
+        )
+        resp = _calc(client, normal["headers"], receipt["id"])
+        assert resp.status_code == 403
+
+        ops = make_user(["marketing_ops"])
+        ops["headers"] = auth_headers(ops)
+        resp = _calc(client, ops["headers"], receipt["id"])
+        assert resp.status_code == 200, resp.text
+
+    def test_台账列表与筛选(self, client: TestClient, admin: dict) -> None:
+        biz, receipt = _create_full_chain(
+            client,
+            admin["headers"],
+            items=_GOLD_ITEMS,
+            ship_date="2026-10-01",
+            receipt_date="2026-10-20",
+            amount="100.00",
+        )
+        _calc(client, admin["headers"], receipt["id"])
+
+        resp = client.get(
+            f"{_CALC}/ledger",
+            params={"settlement_month": "2026-10", "is_active": 1},
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200
+        rows = resp.json()["data"]["list"]
+        assert rows and rows[0]["business_no"] == biz["business_no"]