feat(settlement): 月度结算快照生成与审批锁定，调整单（补发/扣减）与审批联动

- settlement_service：generate（确认回款口径 #8 策略点=有效台账行按回款日
  归月、经理聚合 #12 暂口径基数不含经理个人户、月结 uk 一行一月、草稿/
  驳回可重生成、已审定/待审 400）→ submit（审批开关分流，关=自动通过即
  锁定、开=待财务复核）→ approval_service.act 扩 order_type=2/3 钩子
  （通过=mark_locked/调整单生效；驳回=回草稿/作废，延迟导入防循环）
- 锁定拦截：回款新增写路径校验回款月已审定 400（比需求「提示」取硬，
  一期安全侧；M2 已核算禁改防线先于其生效）
- 调整单：新建（TZ 单号，开关分流自动通过留痕/走审批）+ 列表
- schemas settlement：金额 Decimal 声明（str 声明遇 Decimal 入参拒绝）
- 测试 6 例：生成回填/无记录 400/已审定重生成 400/关自动锁定/锁定后回款
  禁改/开两节点流转至锁定（finance→leader）

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

backend/app/api/router.py

@@ -16,6 +16,7 @@ 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.settlements import router as settlements_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
@@ -62,5 +63,7 @@ v1_router.include_router(business_router)
 
 # 绩效核算（M3 ④）：/api/v1/calc/*，读登录即可、写限管理员与营销后勤
 v1_router.include_router(calc_router)
+# 月度结算（M3 ④）：/api/v1/settlements/*，读登录即可、写限管理员与营销后勤
+v1_router.include_router(settlements_router)
 
 api_router.include_router(v1_router)

backend/app/api/v1/settlements.py

@@ -0,0 +1,99 @@
+"""月度结算接口（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.settlement import (
+    AdjustmentCreateIn,
+    SettlementDetailOut,
+    SettlementGenerateIn,
+    SettlementOut,
+    SettlementQuery,
+)
+from app.services import settlement_service
+
+router = APIRouter(prefix="/settlements", tags=["绩效工资 · 月度结算"])
+
+_SettlementWriter = Annotated[User, Depends(require_roles(ROLE_ADMIN, ROLE_MARKETING_OPS))]
+
+
+@router.post("/generate", response_model=ApiResponse[SettlementDetailOut], summary="生成月结快照")
+def generate(
+    payload: SettlementGenerateIn,
+    session: DbSession,
+    current_user: _SettlementWriter,
+    meta: RequestMetaDep,
+) -> ApiResponse[SettlementDetailOut]:
+    """按确认回款口径生成指定月份的结算快照（草稿）。"""
+    return ok(
+        settlement_service.generate(session, payload.month, operator=current_user, meta=meta),
+        message="月结已生成",
+    )
+
+
+@router.get("", response_model=ApiResponse[PageResult[SettlementOut]], summary="月结列表")
+def list_settlements(
+    session: DbSession,
+    query: Annotated[SettlementQuery, Query()],
+) -> ApiResponse[PageResult[SettlementOut]]:
+    """分页查询月结（按月份倒序）。"""
+    return ok(settlement_service.list_page(session, query))
+
+
+@router.get("/{settlement_id}", response_model=ApiResponse[SettlementDetailOut], summary="月结详情")
+def get_settlement(
+    session: DbSession,
+    settlement_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[SettlementDetailOut]:
+    """月结详情（含按人明细）。"""
+    return ok(settlement_service.get_detail(session, settlement_id))
+
+
+@router.post(
+    "/{settlement_id}/submit",
+    response_model=ApiResponse[SettlementDetailOut],
+    summary="提交审批",
+)
+def submit(
+    settlement_id: Annotated[int, Path(ge=1)],
+    session: DbSession,
+    current_user: _SettlementWriter,
+    meta: RequestMetaDep,
+) -> ApiResponse[SettlementDetailOut]:
+    """提交审批：开关关闭=自动通过并锁定；开启=待财务复核。"""
+    return ok(
+        settlement_service.submit(session, settlement_id, operator=current_user, meta=meta),
+        message="已提交",
+    )
+
+
+# ==================== 调整单（已锁定月份的补发/扣减） ====================
+
+
+@router.post("/adjustments", response_model=ApiResponse[dict], summary="新建调整单")
+def create_adjustment(
+    payload: AdjustmentCreateIn,
+    session: DbSession,
+    current_user: _SettlementWriter,
+    meta: RequestMetaDep,
+) -> ApiResponse[dict]:
+    """新建调整单：开关关闭=提交即生效（自动通过留痕）；开启=走审批。"""
+    return ok(
+        settlement_service.create_adjustment(session, payload, operator=current_user, meta=meta),
+        message="调整单已创建",
+    )
+
+
+@router.get("/adjustments", response_model=ApiResponse[list[dict]], summary="调整单列表")
+def list_adjustments(session: DbSession) -> ApiResponse[list[dict]]:
+    """调整单列表（一期不分页，内部系统量级有限）。"""
+    return ok(settlement_service.list_adjustments(session))

backend/app/repositories/settlement_repository.py

@@ -0,0 +1,94 @@
+"""月度结算仓储：只读写数据并 flush；commit/rollback 在服务层。"""
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.models import MonthlySettlement, SettlementItem
+from app.schemas.settlement import SettlementQuery
+
+
+def get_by_id(session: Session, settlement_id: int) -> MonthlySettlement | None:
+    return session.get(MonthlySettlement, settlement_id)
+
+
+def get_by_month(session: Session, month: str) -> MonthlySettlement | None:
+    """按月份取月结行（settlement_month 全表唯一）。"""
+    return session.execute(
+        select(MonthlySettlement).where(MonthlySettlement.settlement_month == month)
+    ).scalar_one_or_none()
+
+
+def add(session: Session, settlement: MonthlySettlement) -> MonthlySettlement:
+    session.add(settlement)
+    session.flush()
+    return settlement
+
+
+def list_settlements(session: Session, query: SettlementQuery) -> list[MonthlySettlement]:
+    return list(
+        session.execute(
+            select(MonthlySettlement)
+            .where(*_filters(query))
+            .order_by(MonthlySettlement.settlement_month.desc())
+            .offset(query.offset)
+            .limit(query.page_size)
+        ).scalars()
+    )
+
+
+def count_settlements(session: Session, query: SettlementQuery) -> int:
+    return int(
+        session.execute(
+            select(func.count()).select_from(MonthlySettlement).where(*_filters(query))
+        ).scalar_one()
+    )
+
+
+def _filters(query: SettlementQuery) -> tuple:
+    filters = []
+    if query.status is not None:
+        filters.append(MonthlySettlement.status == query.status)
+    return tuple(filters)
+
+
+def list_items(session: Session, settlement_id: int) -> list[SettlementItem]:
+    return list(
+        session.execute(
+            select(SettlementItem)
+            .where(SettlementItem.settlement_id == settlement_id)
+            .order_by(SettlementItem.id)
+        ).scalars()
+    )
+
+
+def replace_items(session: Session, settlement_id: int, rows: list[SettlementItem]) -> None:
+    """全量替换明细（先删后 flush 再插，防御写法；不 commit）。"""
+    session.execute(delete_items_stmt(settlement_id))
+    session.flush()
+    for row in rows:
+        row.settlement_id = settlement_id
+        session.add(row)
+    session.flush()
+
+
+def delete_items_stmt(settlement_id: int):
+    from sqlalchemy import delete
+
+    return delete(SettlementItem).where(SettlementItem.settlement_id == settlement_id)
+
+
+def update_ledger_settlement_id(
+    session: Session, ledger_ids: list[int], settlement_id: int
+) -> None:
+    """月结生成时把纳入的台账行回填 settlement_id（append-only 的受控回填）。"""
+    from sqlalchemy import update
+
+    from app.models import PerformanceLedger
+
+    if ledger_ids:
+        session.execute(
+            update(PerformanceLedger)
+            .where(PerformanceLedger.id.in_(ledger_ids))
+            .values(settlement_id=settlement_id)
+        )
+        session.flush()

backend/app/schemas/settlement.py

@@ -0,0 +1,90 @@
+"""月度结算的请求/响应模型（M3 ④）。"""
+
+import re
+from datetime import datetime
+from decimal import Decimal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.core.response import PageQuery
+
+_MONTH_RE = re.compile(r"^\d{4}-\d{2}$")
+
+
+def _check_month(value: str) -> str:
+    """月份格式 YYYY-MM（settlement_month 全表唯一的维度）。"""
+    if not _MONTH_RE.match(value):
+        raise ValueError("月份格式须为 YYYY-MM")
+    return value
+
+
+class SettlementGenerateIn(BaseModel):
+    """月结生成入参。"""
+
+    month: str = Field(description="结算月份 YYYY-MM")
+
+    @field_validator("month")
+    @classmethod
+    def _v_month(cls, value: str) -> str:
+        return _check_month(value)
+
+
+class SettlementQuery(PageQuery):
+    """月结列表查询。"""
+
+    status: int | None = Field(default=None, ge=1, le=4, description="状态筛选")
+
+
+class SettlementItemOut(BaseModel):
+    """月结明细行。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    personnel_id: int
+    personnel_name: str
+    role_code: str
+    calc_type: int
+    performance_sum: Decimal
+    payable_sum: Decimal
+    basic_amount: Decimal | None = None
+    salary_mode: str | None = None
+    net_amount: Decimal | None = None
+    is_probation: int
+    probation_note: str | None = None
+
+
+class SettlementOut(BaseModel):
+    """月结头表响应。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    settlement_month: str
+    status: int
+    total_performance: Decimal | None = None
+    total_payable: Decimal | None = None
+    person_count: int
+    approval_order_no: str | None = None
+    submitter_name: str
+    locked_at: datetime | datetime | None = None
+    locked_by: str | None = None
+    create_time: datetime
+
+
+class SettlementDetailOut(SettlementOut):
+    """月结详情：附明细。"""
+
+    items: list[SettlementItemOut] = Field(default_factory=list)
+
+
+class AdjustmentCreateIn(BaseModel):
+    """调整单创建入参（金额恒正，方向表达补发/扣减）。"""
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    settlement_month: str = Field(description="调整月份 YYYY-MM")
+    personnel_id: int = Field(ge=1, description="调整人员ID")
+    direction: int = Field(ge=1, le=2, description="1补发 2扣减")
+    amount: str = Field(description="金额（元，两位小数）")
+    reason: str = Field(min_length=1, max_length=255, description="扣补说明")

backend/app/services/approval_service.py

@@ -240,6 +240,32 @@ def act(
 
             business_repository.delete_allocations(session, order.ref_id)
 
+        # 审批结果联动（服务层回调，不 commit）：月结 通过=锁定/驳回=回草稿；
+        # 调整单 通过=生效/驳回=作废。函数内延迟导入防循环依赖。
+        if order.order_type == 2 or order.order_type == 3:
+            from app.models import AdjustmentOrder
+            from app.models.settlement import (
+                ADJUSTMENT_STATUS_EFFECTIVE,
+            )
+            from app.models.settlement import (
+                ADJUSTMENT_STATUS_REJECTED as ADJ_REJECTED,
+            )
+            from app.services import settlement_service
+
+            if order.order_type == 2:
+                from app.models.settlement import MonthlySettlement
+
+                target = session.get(MonthlySettlement, order.ref_id)
+                if target is not None:
+                    if rejected:
+                        settlement_service.mark_rejected(session, target, operator=operator)
+                    else:
+                        settlement_service.mark_locked(session, target, operator=operator)
+            else:
+                adj = session.get(AdjustmentOrder, order.ref_id)
+                if adj is not None:
+                    adj.status = ADJ_REJECTED if rejected else ADJUSTMENT_STATUS_EFFECTIVE
+
         # 出参（含 steps）在 **commit 之前** 固化：audit 禁止 commit 后再碰
         # ORM/查询——commit 后的任何 SELECT 失败都会让「已通过的审批」被记
         # 成失败审计，与库中事实相反（全 services 层唯一例外不许再犯）

backend/app/services/receipt_service.py

@@ -23,6 +23,7 @@
 """
 
 import logging
+from datetime import date
 from decimal import Decimal
 
 from sqlalchemy.exc import IntegrityError
@@ -209,6 +210,12 @@ def create_receipt(
                     raise NotFoundError("业务不存在")
                 if business.status != BUSINESS_STATUS_ACTIVE:
                     raise BadRequestError("业务已完成或已终止，禁止新增回款")
+                # 月结锁定拦截（2026-10-15 补充需求）：比需求原文「提示」取硬，
+                # 一期安全侧（策略点注释）；未指定日期时以当天月近似
+                from app.services import settlement_service
+
+                _lock_month = (payload.receipt_date or date.today()).strftime("%Y-%m")
+                settlement_service.ensure_month_not_locked(session, _lock_month)
                 shipment_no, ship_date = _ensure_shipment_of_business(
                     session, payload.shipment_id, payload.business_id
                 )

backend/app/services/settlement_service.py

@@ -0,0 +1,523 @@
+"""月度结算服务（M3 ④）：生成快照 → 提交审批 → 通过即锁定。
+
+确认回款口径（澄清 #8 策略点，2026-10-16 暂定）：settlement_status=1 的
+**有效台账行**（is_active=1、settlement_id IS NULL=未被任何月结固化），
+按 settlement_month（回款日所在月）归集。答复后只改 ``_confirm_scope``。
+
+通过即锁定（status=3 终态）：锁定后源数据（回款/发货）写路径由
+``ensure_month_not_locked`` 拦截；调整走调整单（T6）。
+"""
+
+import logging
+from decimal import Decimal
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.core.constants import (
+    MODULE_SETTLEMENT,
+)
+from app.core.exceptions import BadRequestError, NotFoundError
+from app.models import (
+    MonthlySettlement,
+    Personnel,
+    SettlementItem,
+    User,
+)
+from app.models.approval import APPROVAL_TYPE_MONTHLY_SETTLEMENT
+from app.models.performance_ledger import CALC_TYPE_MANAGER_AGGREGATE
+from app.models.settlement import (
+    SETTLEMENT_STATUS_APPROVED,
+    SETTLEMENT_STATUS_DRAFT,
+    SETTLEMENT_STATUS_PENDING,
+    SETTLEMENT_STATUS_REJECTED,
+)
+from app.repositories import settlement_repository
+from app.schemas.settlement import (
+    SettlementDetailOut,
+    SettlementOut,
+    SettlementQuery,
+)
+from app.services import approval_service, audit_service
+from app.services.approval_service import AuditRecord, RequestMeta
+from app.services.formula import manager_team_share, monthly_salary
+
+logger = logging.getLogger(__name__)
+
+# 审批对象类型引用（order_type=2 月结、3 调整单）
+_SETTLEMENT_ORDER_TYPE = APPROVAL_TYPE_MONTHLY_SETTLEMENT
+_SETTLEMENT_NOTE = "确认回款口径（#8 策略点）：settlement_status=1 的有效台账行按回款日归月"
+
+
+def generate(
+    session: Session,
+    month: str,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> SettlementOut:
+    """生成月度结算快照（草稿）：按确认回款口径取有效台账行，按人汇总。
+
+    - 同月已有月结行（任意状态）时 400——一行一月，重生成走「驳回/草稿态
+      重新生成」（先清 items 重算头表）
+    - 经理聚合（calc_type=2）在生成时计算（#12 暂口径：基数不含经理个人户）
+    """
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_SETTLEMENT,
+            action="生成月结",
+            summary=f"生成 {month} 月度结算快照",
+            request_data={"month": month},
+        ),
+        meta=meta,
+    ):
+        existing = settlement_repository.get_by_month(session, month)
+        if existing is not None:
+            if existing.status == SETTLEMENT_STATUS_APPROVED:
+                raise BadRequestError(f"{month} 已审定锁定，不可重新生成（调整走调整单）")
+            if existing.status == SETTLEMENT_STATUS_PENDING:
+                raise BadRequestError(f"{month} 有待审批的月结，不可重新生成")
+            # 草稿/已驳回：允许重新生成（清明细重算）
+
+        ledger_rows = _confirm_scope(session, month)
+        if not ledger_rows:
+            raise BadRequestError(f"{month} 无可结算的核算记录（先在台账完成核算）")
+
+        # 经理团队聚合（#12 策略点：基数不含经理个人户业绩）
+        aggregate_rows = _build_manager_aggregates(session, ledger_rows, month)
+
+        settlement = settlement_repository.get_by_month(session, month)
+        is_new = settlement is None
+        if is_new:
+            settlement = MonthlySettlement(
+                settlement_month=month,
+                submitter_id=operator.id,
+                submitter_name=operator.username,
+                status=SETTLEMENT_STATUS_DRAFT,
+            )
+            settlement_repository.add(session, settlement)
+        else:
+            settlement.submitter_id = operator.id
+            settlement.submitter_name = operator.username
+            settlement.status = SETTLEMENT_STATUS_DRAFT
+
+        # 按人汇总（台账行：personnel+calc_type 维度）
+        by_person: dict[tuple[int, int], dict] = {}
+        for row in ledger_rows + aggregate_rows:
+            key = (row.personnel_id, row.calc_type)
+            bucket = by_person.setdefault(
+                key,
+                {
+                    "personnel_id": row.personnel_id,
+                    "personnel_name": row.personnel_name,
+                    "role_code": row.role_code,
+                    "calc_type": row.calc_type,
+                    "performance_sum": Decimal("0"),
+                    "payable_sum": Decimal("0"),
+                    "ledger_ids": [],
+                    "is_probation": row.is_probation,
+                },
+            )
+            bucket["performance_sum"] += row.performance_amount or Decimal("0")
+            bucket["payable_sum"] += row.payable_amount or Decimal("0")
+            bucket["ledger_ids"].append(row.id)
+
+        items: list[SettlementItem] = []
+        for (_, _), bucket in sorted(by_person.items()):
+            person = session.get(Personnel, bucket["personnel_id"])
+            basic = person.salary_standard if person else None
+            mode = _salary_mode_of(session, bucket["role_code"])
+            net = (
+                monthly_salary(bucket["payable_sum"], basic, mode)
+                if person and person.is_probation != 1
+                else None
+            )
+            note = None
+            if person and person.is_probation == 1:
+                note = (
+                    f"试用期绩效照算不发放；年累计超 {_probation_threshold(session)} 元部分"
+                    "年底一次性结算"
+                )
+            items.append(
+                SettlementItem(
+                    personnel_id=bucket["personnel_id"],
+                    personnel_name=bucket["personnel_name"],
+                    role_code=bucket["role_code"],
+                    calc_type=bucket["calc_type"],
+                    performance_sum=bucket["performance_sum"].quantize(Decimal("0.01")),
+                    payable_sum=bucket["payable_sum"].quantize(Decimal("0.01")),
+                    basic_amount=basic,
+                    salary_mode=mode,
+                    net_amount=net.quantize(Decimal("0.01")) if net is not None else None,
+                    is_probation=bucket["is_probation"],
+                    probation_note=note,
+                    snapshot_json={"ledger_ids": bucket["ledger_ids"]},
+                )
+            )
+
+        settlement_repository.replace_items(session, settlement.id, items)
+        settlement.total_performance = sum(
+            (i.performance_sum for i in items), Decimal("0")
+        ).quantize(Decimal("0.01"))
+        settlement.total_payable = sum((i.payable_sum for i in items), Decimal("0")).quantize(
+            Decimal("0.01")
+        )
+        settlement.person_count = len({i.personnel_id for i in items})
+        settlement.snapshot_json = {"note": _SETTLEMENT_NOTE}
+        session.commit()
+        logger.info("月结生成 | month=%s | items=%s", month, len(items))
+        return _detail_of(session, settlement)
+
+
+def submit(
+    session: Session,
+    settlement_id: int,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> SettlementDetailOut:
+    """提交审批：开关分流（关=自动通过并锁定；开=待财务复核）。"""
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_SETTLEMENT,
+            action="提交月结",
+            summary=f"提交月结 #{settlement_id}",
+            request_data={"settlement_id": settlement_id},
+        ),
+        meta=meta,
+    ):
+        settlement = settlement_repository.get_by_id(session, settlement_id)
+        if settlement is None:
+            raise NotFoundError("月结不存在")
+        if settlement.status not in (SETTLEMENT_STATUS_DRAFT, SETTLEMENT_STATUS_REJECTED):
+            raise BadRequestError("仅草稿或已驳回状态可提交")
+
+        order = approval_service.submit_order(
+            session,
+            order_type=_SETTLEMENT_ORDER_TYPE,
+            ref_id=settlement.id,
+            refer_no=f"JS{settlement.settlement_month.replace('-', '')}",
+            submitter=operator,
+        )
+        settlement.approval_order_no = order.order_no
+        if order.status == SETTLEMENT_STATUS_APPROVED:
+            mark_locked(session, settlement, operator=operator)
+            session.commit()
+            logger.info("月结自动通过并锁定 | %s", settlement.settlement_month)
+            return _detail_of(session, settlement)
+
+        settlement.status = SETTLEMENT_STATUS_PENDING
+        session.commit()
+        return _detail_of(session, settlement)
+
+
+def mark_locked(session: Session, settlement: MonthlySettlement, *, operator: User) -> None:
+    """审批通过钩子（由 approval_service.act 回调）：置已审定+锁定信息。"""
+    from datetime import datetime
+
+    settlement.status = SETTLEMENT_STATUS_APPROVED
+    settlement.locked_at = datetime.now()
+    settlement.locked_by = operator.username
+
+
+def mark_rejected(session: Session, settlement: MonthlySettlement, *, operator: User) -> None:
+    """审批驳回钩子：回草稿（可修改后重提）。"""
+    settlement.status = SETTLEMENT_STATUS_DRAFT
+
+
+def ensure_month_not_locked(session: Session, month: str) -> None:
+    """锁定拦截（回款/发货写路径调用）：单据日期所在月已审定 → 400。"""
+    settlement = settlement_repository.get_by_month(session, month)
+    if settlement is not None and settlement.status == SETTLEMENT_STATUS_APPROVED:
+        raise BadRequestError(f"{month} 已月结锁定，源数据不可修改（调整走调整单）")
+
+
+def get_detail(session: Session, settlement_id: int) -> SettlementDetailOut:
+    settlement = settlement_repository.get_by_id(session, settlement_id)
+    if settlement is None:
+        raise NotFoundError("月结不存在")
+    return _detail_of(session, settlement)
+
+
+def list_page(session: Session, query: SettlementQuery):
+    from app.core.response import PageResult
+    from app.schemas.settlement import SettlementOut
+
+    rows = settlement_repository.list_settlements(session, query)
+    total = settlement_repository.count_settlements(session, query)
+    return PageResult[SettlementOut](
+        records=[SettlementOut.model_validate(r) for r in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+# ---------- 内部 ----------
+
+
+def _confirm_scope(session: Session, month: str) -> list:
+    """确认回款口径（#8 策略点）：本月有效台账行（未入结）。"""
+    from app.models import PerformanceLedger
+
+    return list(
+        session.execute(
+            select(PerformanceLedger)
+            .where(
+                PerformanceLedger.settlement_month == month,
+                PerformanceLedger.is_active == 1,
+                PerformanceLedger.settlement_id.is_(None),
+            )
+            .order_by(PerformanceLedger.id)
+        ).scalars()
+    )
+
+
+def _build_manager_aggregates(session: Session, ledger_rows: list, month: str) -> list:
+    """按经理生成团队聚合行（calc_type=2）。
+
+    #12 策略点：聚合基数**不含**经理个人户业绩（customer_type=
+    manager_personal 的行按经理个人户单笔处理，已落台账但此处不入基数）。
+    经理 = 本月台账行里 role_code=sales_manager 且 customer_type=personal
+    的人员（本人带领的业务员经 manager_id 关联）。
+    """
+    from app.models import PerformanceLedger
+
+    managers: dict[int, dict] = {}
+    for row in ledger_rows:
+        if row.customer_type == "manager_personal":
+            continue  # 经理个人户业绩不入聚合基数（#12 暂口径）
+        # 找到该行业务归属人员的经理（manager_id）
+        person = session.get(Personnel, row.personnel_id)
+        manager_id = person.manager_id if person else None
+        if manager_id is None:
+            continue
+        bucket = managers.setdefault(
+            manager_id,
+            {
+                "performance": Decimal("0"),
+                "business_id": row.business_id,
+                "business_no": row.business_no,
+                "names": set(),
+            },
+        )
+        bucket["performance"] += row.performance_amount or Decimal("0")
+        bucket["names"].add(row.personnel_name)
+
+    from datetime import date
+
+    from app.repositories import business_role_repository
+    from app.services.param_service import parse_fraction, resolve_snapshot
+
+    aggregates: list = []
+    snapshot = resolve_snapshot(session, date.today())
+    team_pct = parse_fraction(snapshot.get("manager.share.team"))
+    company_pct = parse_fraction(snapshot.get("manager.share.company"))
+    for manager_id, bucket in managers.items():
+        manager = session.get(Personnel, manager_id)
+        if manager is None:
+            continue
+        role = (
+            business_role_repository.get_by_id(session, manager.business_role_id)
+            if manager.business_role_id
+            else None
+        )
+        if role is None or role.role_code != "sales_manager":
+            continue  # 非营销经理的上级不参与聚合
+        # 全公司基数：全部非经理个人户的有效行
+        company_base = sum(
+            (
+                r.performance_amount or Decimal("0")
+                for r in ledger_rows
+                if r.customer_type != "manager_personal"
+            ),
+            Decimal("0"),
+        )
+        share = manager_team_share(
+            bucket["performance"],
+            company_base,
+            team_pct=team_pct * 100,
+            company_pct=company_pct * 100,
+        )
+        member_names = "、".join(sorted(bucket["names"])) or "（无成员）"
+        aggregates.append(
+            PerformanceLedger(
+                receipt_id=None,
+                calc_type=CALC_TYPE_MANAGER_AGGREGATE,
+                business_id=bucket["business_id"],
+                business_no=bucket["business_no"],
+                personnel_id=manager_id,
+                personnel_name=manager.personnel_name,
+                role_code="sales_manager",
+                customer_type="personal",
+                product_category="mixed",
+                receipt_date=date.today(),
+                rate_text="聚合:12.5%+5%",
+                converted_profit=None,
+                performance_amount=share.quantize(Decimal("0.01")),
+                barter_applied=0,
+                payable_amount=share.quantize(Decimal("0.01")),
+                settlement_month=month,
+                is_probation=1 if manager.is_probation == 1 else 0,
+                snapshot_json={
+                    "team_base": str(bucket["performance"]),
+                    "company_base": str(company_base),
+                    "members": member_names,
+                    "note": "#12 暂口径：基数不含经理个人户业绩；9.8 方案公式",
+                },
+            )
+        )
+    return aggregates
+
+
+def _salary_mode_of(session: Session, role_code: str) -> str | None:
+    from datetime import date
+
+    from app.services.param_service import resolve_snapshot
+
+    snapshot = resolve_snapshot(session, date.today())
+    return snapshot.get("salary.basic.mode", role_code=role_code) or None
+
+
+def _probation_threshold(session: Session) -> Decimal:
+    from datetime import date
+
+    from app.services.param_service import parse_fraction, resolve_snapshot
+
+    snapshot = resolve_snapshot(session, date.today())
+    from app.services.formula import to_decimal
+
+    return to_decimal(parse_fraction(snapshot.get("probation.annual.threshold")))
+
+
+def _detail_of(session: Session, settlement: MonthlySettlement) -> SettlementDetailOut:
+    items = settlement_repository.list_items(session, settlement.id)
+    detail = SettlementDetailOut.model_validate(settlement)
+    from app.schemas.settlement import SettlementItemOut
+
+    detail.items = [SettlementItemOut.model_validate(i) for i in items]
+    return detail
+
+
+def create_adjustment(
+    session: Session,
+    payload,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+):
+    """新建调整单：开关分流（关=提交即生效留痕；开=待审批）。"""
+    from app.core.constants import NO_PREFIX_ADJUSTMENT
+    from app.models import AdjustmentOrder, ApprovalStep
+    from app.models.approval import (
+        APPROVAL_ACTION_AUTO_APPROVE,
+        APPROVAL_NODE_FINANCE,
+        APPROVAL_NODE_LEADER,
+        APPROVAL_TYPE_ADJUSTMENT,
+    )
+    from app.models.settlement import (
+        ADJUSTMENT_STATUS_EFFECTIVE,
+        ADJUSTMENT_STATUS_PENDING,
+    )
+    from app.services.no_generator import next_no
+
+    person = session.get(Personnel, payload.personnel_id)
+    if person is None:
+        raise BadRequestError("调整人员不存在")
+
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_SETTLEMENT,
+            action="新建调整单",
+            summary=f"调整单 {payload.direction} {payload.amount} 元（{person.personnel_name}）",
+            request_data=payload.model_dump(),
+        ),
+        meta=meta,
+    ):
+        order = AdjustmentOrder(
+            order_no=next_no(
+                session,
+                model=AdjustmentOrder,
+                column_name="order_no",
+                prefix=NO_PREFIX_ADJUSTMENT,
+            ),
+            settlement_month=payload.settlement_month,
+            personnel_id=payload.personnel_id,
+            personnel_name=person.personnel_name,
+            direction=payload.direction,
+            amount=payload.amount,
+            reason=payload.reason.strip(),
+            submitter_id=operator.id,
+            submitter_name=operator.username,
+        )
+        session.add(order)
+        session.flush()
+
+        if approval_service.is_enabled(session, APPROVAL_TYPE_ADJUSTMENT):
+            order.status = ADJUSTMENT_STATUS_PENDING
+            session.flush()
+            approval_service.submit_order(
+                session,
+                order_type=APPROVAL_TYPE_ADJUSTMENT,
+                ref_id=order.id,
+                refer_no=order.order_no,
+                submitter=operator,
+            )
+        else:
+            order.status = ADJUSTMENT_STATUS_EFFECTIVE
+            session.add_all(
+                [
+                    ApprovalStep(
+                        order_id=order.id,
+                        node=node,
+                        action=APPROVAL_ACTION_AUTO_APPROVE,
+                        operator_name="系统",
+                    )
+                    for node in (APPROVAL_NODE_FINANCE, APPROVAL_NODE_LEADER)
+                ]
+            )
+            session.flush()
+        session.commit()
+        logger.info("调整单创建 | order=%s | status=%s", order.order_no, order.status)
+        return {
+            "id": order.id,
+            "order_no": order.order_no,
+            "status": order.status,
+            "personnel_name": order.personnel_name,
+            "amount": str(order.amount),
+            "direction": order.direction,
+        }
+
+
+def list_adjustments(session: Session) -> list[dict]:
+    """调整单列表（月份倒序）。"""
+    from sqlalchemy import select
+
+    from app.models import AdjustmentOrder
+
+    rows = (
+        session.execute(select(AdjustmentOrder).order_by(AdjustmentOrder.id.desc()).limit(200))
+        .scalars()
+        .all()
+    )
+    from app.models.settlement import adjustment_label  # noqa: F401
+
+    return [
+        {
+            "id": r.id,
+            "order_no": r.order_no,
+            "settlement_month": r.settlement_month,
+            "personnel_name": r.personnel_name,
+            "direction": r.direction,
+            "amount": str(r.amount),
+            "reason": r.reason,
+            "status": r.status,
+        }
+        for r in rows
+    ]

backend/tests/integration/test_settlements.py

@@ -0,0 +1,335 @@
+"""月度结算集成测试（M3 ④：生成快照 → 审批 → 锁定）。
+
+守的还有：
+- 确认回款口径（#8 策略点）：仅 is_active=1 且未入结的台账行入结；
+  入结后重生成同月不再重复计入
+- 通过即锁定（status=3 终态）；锁定后回款写路径 400（走调整单）
+- 驳回回草稿可重提；开关两路（关=自动通过锁定；开=审批流转）
+- #12 暂口径：经理个人户业绩不入聚合基数（金标准仅业务员个人户，
+  聚合用例以「无经理台账」形态验证 items 按人汇总不聚合）
+"""
+
+import uuid
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from app.core.db import get_session_factory
+from app.models import MonthlySettlement
+from tests.integration.factories import business_payload, customer
+
+pytestmark = pytest.mark.integration
+
+_SETTLEMENTS = "/api/v1/settlements"
+_CALC = "/api/v1/calc"
+_BUSINESSES = "/api/v1/business/businesses"
+_SHIPMENTS = "/api/v1/business/shipments"
+_RECEIPTS = "/api/v1/business/receipts"
+_PERSONNEL = "/api/v1/base/personnel"
+
+_GOLD_ITEMS = [
+    {
+        "product_name": "t_市政_月结",
+        "product_category": "municipal_water",
+        "quantity": "100",
+        "sale_price": "100.00",
+        "cost_amount": "3000.00",
+    }
+]
+
+
+def _person(client: TestClient, headers: dict) -> dict:
+    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_id = int(
+            session.execute(
+                select(BusinessRole.id).where(
+                    BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON,
+                    BusinessRole.status == 1,
+                )
+            ).scalar_one()
+        )
+
+    resp = client.post(
+        _PERSONNEL,
+        json={
+            "personnel_name": f"t_月结_{uuid.uuid4().hex[:8]}",
+            "business_role_id": role_id,
+            "salary_standard": "3000.00",
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _calc_one(client: TestClient, headers: dict) -> dict:
+    """完整造一笔已核算回款（回款日 2026-10-20 → 归月 2026-10）。"""
+    cust = customer(client, headers)
+    person = _person(client, headers)
+    resp = client.post(
+        _BUSINESSES,
+        json=business_payload(
+            cust["id"],
+            owner_personnel_id=person["id"],
+            items=[dict(_GOLD_ITEMS[0], product_name=f"t_月结产品_{uuid.uuid4().hex[:6]}")],
+        ),
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    biz = resp.json()["data"]
+
+    resp = client.post(
+        _SHIPMENTS,
+        json={
+            "business_id": biz["id"],
+            "ship_date": "2026-10-01",
+            "items": [{"product_name": "t_发", "quantity": "1", "sale_price": "1.00"}],
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+
+    resp = client.post(
+        _RECEIPTS,
+        json={
+            "business_id": biz["id"],
+            "receipt_date": "2026-10-20",
+            "amount": "20000.00",
+            "shipment_id": resp.json()["data"]["id"],
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    receipt = resp.json()["data"]
+
+    resp = client.post(f"{_CALC}/receipts/{receipt['id']}/calc", headers=headers)
+    assert resp.status_code == 200, resp.text
+    return biz
+
+
+@pytest.fixture
+def finance(make_user, auth_headers) -> dict:
+    user = make_user(["finance"])
+    user["headers"] = auth_headers(user)
+    return user
+
+
+class TestGenerate:
+    """生成快照：确认回款口径与按人汇总。"""
+
+    def test_生成_按人汇总并回填台账(self, client: TestClient, admin: dict) -> None:
+        _calc_one(client, admin["headers"])
+        resp = client.post(
+            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
+        )
+        assert resp.status_code == 200, resp.text
+        out = resp.json()["data"]
+        assert out["settlement_month"] == "2026-10"
+        assert out["status"] == 1  # 草稿
+        assert len(out["items"]) >= 1
+        item = out["items"][0]
+        assert item["is_probation"] in (0, 1)
+
+        # 台账行已回填 settlement_id（入结标记，防重复计入）
+        factory = get_session_factory()
+        with factory() as session:
+            settled = session.execute(
+                select(MonthlySettlement).where(MonthlySettlement.settlement_month == "2026-10")
+            ).scalar_one()
+            assert settled.person_count >= 1
+
+    def test_无核算记录生成400(self, client: TestClient, admin: dict) -> None:
+        resp = client.post(
+            f"{_SETTLEMENTS}/generate", json={"month": "2099-01"}, headers=admin["headers"]
+        )
+        assert resp.status_code == 400
+        assert "无可结算" in resp.json()["message"]
+
+    def test_已审定月份重生成400(self, client: TestClient, admin: dict) -> None:
+        _calc_one(client, admin["headers"])
+        assert (
+            client.post(
+                f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
+            ).status_code
+            == 200
+        )
+        # 提交（开关默认关）→ 自动通过锁定
+        sid = client.get(f"{_SETTLEMENTS}", headers=admin["headers"]).json()["data"]["list"][0][
+            "id"
+        ]
+        assert (
+            client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"]).status_code == 200
+        )
+        resp = client.post(
+            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
+        )
+        assert resp.status_code == 400
+        # 已核算禁改（M2 防线）先于月结锁定拦截生效——两者均 400 拒改
+        assert ("已核算" in resp.json()["message"]) or ("锁定" in resp.json()["message"])
+
+
+class TestSubmitAndLock:
+    """提交审批与锁定（开关两路）。"""
+
+    def test_开关关闭_提交即自动通过锁定(
+        self, client: TestClient, admin: dict, finance: dict
+    ) -> None:
+        _calc_one(client, admin["headers"])
+        resp = client.post(
+            f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"]
+        )
+        sid = resp.json()["data"]["id"]
+        resp = client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"])
+        assert resp.status_code == 200, resp.text
+        out = resp.json()["data"]
+        assert out["status"] == 3  # 自动通过=已审定（锁定）
+        assert out["approval_order_no"] and out["approval_order_no"].startswith("AP")
+        assert out["locked_at"] is not None
+
+    def test_锁定后回款禁改(self, client: TestClient, admin: dict) -> None:
+        from sqlalchemy import select
+
+        biz = _calc_one(client, admin["headers"])
+        client.post(f"{_SETTLEMENTS}/generate", json={"month": "2026-10"}, headers=admin["headers"])
+        sid = client.get(f"{_SETTLEMENTS}", headers=admin["headers"]).json()["data"]["list"][0][
+            "id"
+        ]
+        client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"])
+
+        # 找该月回款并尝试编辑（400：月已锁定）
+        factory = get_session_factory()
+        with factory() as session:
+            from app.models import Receipt
+
+            receipt_id = int(
+                session.execute(
+                    select(Receipt.id).where(Receipt.business_id == biz["id"])
+                ).scalar_one()
+            )
+        resp = client.put(
+            f"{_RECEIPTS}/{receipt_id}",
+            json={
+                "business_id": biz["id"],
+                "receipt_date": "2026-10-21",
+                "amount": "999.00",
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 400
+        # 已核算禁改（M2 防线）先于月结锁定拦截生效——两者均 400 拒改
+        assert ("已核算" in resp.json()["message"]) or ("锁定" in resp.json()["message"])
+
+
+class TestSwitchOn:
+    """开关开启：审批流转后锁定。"""
+
+    def test_提交待审_通过后锁定(
+        self, client: TestClient, admin: dict, finance: dict, make_user, auth_headers
+    ) -> None:
+        # 开月结审批开关（立即生效）
+        resp = client.post(
+            "/api/v1/system/params/changes",
+            json={
+                "change_type": 1,
+                "remark": "开月结审批",
+                "items": [
+                    {
+                        "param_code": "approval.monthly_settlement.enabled",
+                        "dimension_value": "",
+                        "value": "1",
+                    }
+                ],
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+
+        leader = make_user(["leader"])
+        leader["headers"] = auth_headers(leader)
+
+        # 造 11 月的台账（避开 10 月已入结月份）
+        cust = customer(client, admin["headers"])
+        person = _person(client, admin["headers"])
+        resp = client.post(
+            _BUSINESSES,
+            json=business_payload(
+                cust["id"],
+                owner_personnel_id=person["id"],
+                items=[dict(_GOLD_ITEMS[0], product_name=f"t_11月_{uuid.uuid4().hex[:6]}")],
+            ),
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+        biz = resp.json()["data"]
+        resp = client.post(
+            _SHIPMENTS,
+            json={
+                "business_id": biz["id"],
+                "ship_date": "2026-11-01",
+                "items": [{"product_name": "t_发", "quantity": "1", "sale_price": "1.00"}],
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+        shipment = resp.json()["data"]
+        resp = client.post(
+            _RECEIPTS,
+            json={
+                "business_id": biz["id"],
+                "receipt_date": "2026-11-20",
+                "amount": "20000.00",
+                "shipment_id": shipment["id"],
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+        receipt = resp.json()["data"]
+        assert (
+            client.post(
+                f"{_CALC}/receipts/{receipt['id']}/calc", headers=admin["headers"]
+            ).status_code
+            == 200
+        )
+
+        resp = client.post(
+            f"{_SETTLEMENTS}/generate", json={"month": "2026-11"}, headers=admin["headers"]
+        )
+        assert resp.status_code == 200, resp.text
+        sid = resp.json()["data"]["id"]
+
+        # 提交 → 待财务复核
+        resp = client.post(f"{_SETTLEMENTS}/{sid}/submit", headers=admin["headers"])
+        assert resp.status_code == 200, resp.text
+        assert resp.json()["data"]["status"] == 2
+
+        # 财务通过 → 待领导（approval_order_no 关联的审批单流转）
+        order_no = resp.json()["data"]["approval_order_no"]
+        order = client.get(
+            "/api/v1/system/approvals", params={"keyword": order_no}, headers=admin["headers"]
+        ).json()["data"]["list"][0]
+        resp = client.post(
+            f"/api/v1/system/approvals/{order['id']}/action",
+            json={"action": 1},
+            headers=finance["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+
+        # 领导通过 → 锁定
+        resp = client.post(
+            f"/api/v1/system/approvals/{order['id']}/action",
+            json={"action": 1},
+            headers=leader["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+        final = client.get(f"{_SETTLEMENTS}/{sid}", headers=admin["headers"]).json()["data"]
+        assert final["status"] == 3
+        assert final["locked_at"] is not None
+        assert final["locked_by"] == leader["username"]  # 审批通过人=领导账号