diff --git a/backend/app/api/router.py b/backend/app/api/router.py
index 4d94019..f8c155f 100644
--- a/backend/app/api/router.py
+++ b/backend/app/api/router.py
@@ -11,6 +11,8 @@ from app.api.v1 import auth
 from app.api.v1.base import customers as base_customers
 from app.api.v1.base import personnel as base_personnel
 from app.api.v1.base import products as base_products
+from app.api.v1.business import businesses as business_businesses
+from app.api.v1.business import ratings as business_ratings
 from app.api.v1.system import business_roles as system_business_roles
 from app.api.v1.system import logs as system_logs
 from app.api.v1.system import params as system_params
@@ -43,4 +45,10 @@ base_router.include_router(base_products.router)
 base_router.include_router(base_customers.router)
 v1_router.include_router(base_router)
 
+# 业务数据（需求 2.2）：/api/v1/business/*，业务操作登录即可访问
+business_router = APIRouter(prefix="/business")
+business_router.include_router(business_businesses.router)
+business_router.include_router(business_ratings.router)
+v1_router.include_router(business_router)
+
 api_router.include_router(v1_router)
diff --git a/backend/app/api/v1/business/businesses.py b/backend/app/api/v1/business/businesses.py
new file mode 100644
index 0000000..440babd
--- /dev/null
+++ b/backend/app/api/v1/business/businesses.py
@@ -0,0 +1,177 @@
+"""业务管理接口（需求 2.2.1/2.2.4）。
+
+校验规则（编号生成、状态机、归属留痕、汇总重算）见 `business_service`
+模块 docstring；基础数据属业务操作，登录即可访问。
+"""
+
+from datetime import date
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Query
+
+from app.api.deps import DbSession, RequestMetaDep, get_current_user
+from app.core.response import ApiResponse, PageQuery, PageResult, ok
+from app.models import User
+from app.schemas.business import (
+    BusinessCreateIn,
+    BusinessDetailOut,
+    BusinessOut,
+    BusinessOwnershipIn,
+    BusinessQuery,
+    BusinessStatusIn,
+    BusinessStatusLogOut,
+    BusinessUpdateIn,
+    GrossProfitOut,
+    OwnershipChangeLogOut,
+)
+from app.services import business_service
+
+# 基础数据是业务操作：登录即可，不要求管理员（区别于系统管理的 AdminUser）
+CurrentUser = Annotated[User, Depends(get_current_user)]
+
+router = APIRouter(prefix="/businesses", tags=["业务管理"])
+
+
+@router.get("", response_model=ApiResponse[PageResult[BusinessOut]], summary="业务列表")
+def list_businesses(
+    session: DbSession,
+    _: CurrentUser,
+    query: Annotated[BusinessQuery, Query()],
+) -> ApiResponse[PageResult[BusinessOut]]:
+    """分页查询业务。编号/公司名称模糊，可按状态/客户/归属人员/日期范围筛选。"""
+    return ok(business_service.list_businesses(session, query))
+
+
+@router.post("", response_model=ApiResponse[BusinessOut], summary="创建业务")
+def create_business(
+    session: DbSession,
+    current_user: CurrentUser,
+    payload: BusinessCreateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[BusinessOut]:
+    """创建业务 + 明细行（≥1）。业务编号自动生成；创建写首条状态留痕。"""
+    return ok(
+        business_service.create_business(session, payload, operator=current_user, meta=meta),
+        message="业务已创建",
+    )
+
+
+@router.get("/{business_id}", response_model=ApiResponse[BusinessDetailOut], summary="业务详情")
+def get_business(
+    session: DbSession,
+    _: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[BusinessDetailOut]:
+    """业务详情，附产品明细行，供编辑抽屉一次回显。"""
+    return ok(business_service.get_detail(session, business_id))
+
+
+@router.put("/{business_id}", response_model=ApiResponse[BusinessOut], summary="编辑业务")
+def update_business(
+    session: DbSession,
+    current_user: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+    payload: BusinessUpdateIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[BusinessOut]:
+    """编辑业务（明细全量替换并重算汇总）。归属不在此端点调整，走 /ownership。"""
+    return ok(
+        business_service.update_business(
+            session, business_id, payload, operator=current_user, meta=meta
+        ),
+        message="已保存",
+    )
+
+
+@router.delete("/{business_id}", response_model=ApiResponse[None], summary="删除业务")
+def delete_business(
+    session: DbSession,
+    current_user: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+    meta: RequestMetaDep,
+) -> ApiResponse[None]:
+    """删除业务。有发货或回款记录时禁删，应改为已终止。"""
+    business_service.delete_business(session, business_id, operator=current_user, meta=meta)
+    return ok(message="业务已删除")
+
+
+@router.put("/{business_id}/status", response_model=ApiResponse[BusinessOut], summary="状态流转")
+def change_status(
+    session: DbSession,
+    current_user: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+    payload: BusinessStatusIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[BusinessOut]:
+    """状态流转：执行中→已完成/已终止；终态不可再变更；终止必填原因。"""
+    return ok(
+        business_service.change_status(
+            session, business_id, payload, operator=current_user, meta=meta
+        ),
+        message="状态已变更",
+    )
+
+
+@router.put("/{business_id}/ownership", response_model=ApiResponse[BusinessOut], summary="调整归属")
+def change_ownership(
+    session: DbSession,
+    current_user: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+    payload: BusinessOwnershipIn,
+    meta: RequestMetaDep,
+) -> ApiResponse[BusinessOut]:
+    """调整负责/协助营销人员，同事务写归属变更留痕（不回溯历史）。"""
+    return ok(
+        business_service.change_ownership(
+            session, business_id, payload, operator=current_user, meta=meta
+        ),
+        message="归属已调整",
+    )
+
+
+@router.get(
+    "/{business_id}/ownership-logs",
+    response_model=ApiResponse[PageResult[OwnershipChangeLogOut]],
+    summary="归属变更记录",
+)
+def list_ownership_logs(
+    session: DbSession,
+    _: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+    query: Annotated[PageQuery, Query()],
+) -> ApiResponse[PageResult[OwnershipChangeLogOut]]:
+    """某业务的归属变更留痕（分页，按时间倒序）。"""
+    return ok(business_service.list_ownership_logs(session, business_id, query))
+
+
+@router.get(
+    "/{business_id}/status-logs",
+    response_model=ApiResponse[PageResult[BusinessStatusLogOut]],
+    summary="状态流转记录",
+)
+def list_status_logs(
+    session: DbSession,
+    _: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+    query: Annotated[PageQuery, Query()],
+) -> ApiResponse[PageResult[BusinessStatusLogOut]]:
+    """某业务的状态流转时间线（分页，按时间正序，从创建起展示）。"""
+    return ok(business_service.list_status_logs(session, business_id, query))
+
+
+@router.get(
+    "/{business_id}/gross-profit",
+    response_model=ApiResponse[GrossProfitOut],
+    summary="毛利润数据视图",
+)
+def gross_profit(
+    session: DbSession,
+    _: CurrentUser,
+    business_id: Annotated[int, Path(ge=1)],
+    as_of: Annotated[date | None, Query()] = None,
+) -> ApiResponse[GrossProfitOut]:
+    """按产品逐行实时计算毛利润与折算额（不含运费口径，对照表 #2）。
+
+    阈值按 asOf（默认今日）取参数快照；M3 核算实际按回款到账日取参。
+    """
+    return ok(business_service.gross_profit(session, business_id, as_of))
diff --git a/backend/app/repositories/business_repository.py b/backend/app/repositories/business_repository.py
new file mode 100644
index 0000000..50950db
--- /dev/null
+++ b/backend/app/repositories/business_repository.py
@@ -0,0 +1,257 @@
+"""业务数据仓储：业务单、明细、状态留痕与归属留痕。
+
+只读写数据并 flush；commit/rollback 与业务异常一律在 service 层。
+归属留痕复用 ownership_change_logs（owner_type=2 为业务维度，常量在
+models/customer.py 定义）。
+"""
+
+from datetime import date
+
+from sqlalchemy import Row, Select, delete, func, or_, select
+from sqlalchemy.orm import Session, aliased
+
+from app.models import (
+    Business,
+    BusinessItem,
+    BusinessStatusLog,
+    Customer,
+    OwnershipChangeLog,
+    Personnel,
+)
+from app.models.customer import OWNER_TYPE_BUSINESS
+
+
+def get_by_id(session: Session, business_id: int, *, for_update: bool = False) -> Business | None:
+    """按主键取业务行；for_update 供删除保护与状态机使用。"""
+    stmt = select(Business).where(Business.id == business_id)
+    if for_update:
+        stmt = stmt.with_for_update()
+    return session.execute(stmt).scalars().first()
+
+
+def create(session: Session, business: Business) -> None:
+    """新增业务行（不 commit，flush 让自增 id 就位）。"""
+    session.add(business)
+    session.flush()
+
+
+def get_with_names(
+    session: Session, business_id: int
+) -> tuple[Business, str, str | None, str | None] | None:
+    """业务行 + 客户名 + 负责人名 + 协助人名（DTO 与审计摘要共用）。"""
+    owner = aliased(Personnel)
+    assistant = aliased(Personnel)
+    row = session.execute(
+        select(Business, Customer.company_name, owner.personnel_name, assistant.personnel_name)
+        .join(Customer, Customer.id == Business.customer_id)
+        .outerjoin(owner, owner.id == Business.owner_personnel_id)
+        .outerjoin(assistant, assistant.id == Business.assistant_personnel_id)
+        .where(Business.id == business_id)
+    ).first()
+    return row
+
+
+def _apply_filters(
+    stmt: Select,
+    *,
+    keyword: str | None,
+    status: int | None,
+    customer_id: int | None,
+    owner_personnel_id: int | None,
+    date_from: date | None,
+    date_to: date | None,
+) -> Select:
+    """业务列表的筛选条件（调用方须已 join Customer）。"""
+    if keyword:
+        # 业务编号/公司名称双列 OR 模糊；contains(autoescape) 同 query_utils.fuzzy
+        # 的转义口径（fuzzy 为单列 where 语义，双列 OR 在此直接组合）
+        stmt = stmt.where(
+            or_(
+                Business.business_no.contains(keyword, autoescape=True),
+                Customer.company_name.contains(keyword, autoescape=True),
+            )
+        )
+    if status is not None:
+        stmt = stmt.where(Business.status == status)
+    if customer_id is not None:
+        stmt = stmt.where(Business.customer_id == customer_id)
+    if owner_personnel_id is not None:
+        stmt = stmt.where(Business.owner_personnel_id == owner_personnel_id)
+    if date_from is not None:
+        stmt = stmt.where(Business.business_date >= date_from)
+    if date_to is not None:
+        stmt = stmt.where(Business.business_date <= date_to)
+    return stmt
+
+
+def list_businesses(
+    session: Session,
+    *,
+    offset: int,
+    limit: int,
+    keyword: str | None,
+    status: int | None,
+    customer_id: int | None,
+    owner_personnel_id: int | None,
+    date_from: date | None,
+    date_to: date | None,
+) -> list[tuple[Business, str, str | None, str | None]]:
+    """分页查询业务（附客户名/负责人名/协助人名），按 id 倒序。"""
+    owner = aliased(Personnel)
+    assistant = aliased(Personnel)
+    stmt = select(
+        Business, Customer.company_name, owner.personnel_name, assistant.personnel_name
+    ).join(Customer, Customer.id == Business.customer_id)
+    stmt = _apply_filters(
+        stmt,
+        keyword=keyword,
+        status=status,
+        customer_id=customer_id,
+        owner_personnel_id=owner_personnel_id,
+        date_from=date_from,
+        date_to=date_to,
+    )
+    stmt = (
+        stmt.outerjoin(owner, owner.id == Business.owner_personnel_id)
+        .outerjoin(assistant, assistant.id == Business.assistant_personnel_id)
+        .order_by(Business.id.desc())
+        .offset(offset)
+        .limit(limit)
+    )
+    return list(session.execute(stmt).all())
+
+
+def count_businesses(
+    session: Session,
+    *,
+    keyword: str | None,
+    status: int | None,
+    customer_id: int | None,
+    owner_personnel_id: int | None,
+    date_from: date | None,
+    date_to: date | None,
+) -> int:
+    stmt = (
+        select(func.count())
+        .select_from(Business)
+        .join(Customer, Customer.id == Business.customer_id)
+    )
+    stmt = _apply_filters(
+        stmt,
+        keyword=keyword,
+        status=status,
+        customer_id=customer_id,
+        owner_personnel_id=owner_personnel_id,
+        date_from=date_from,
+        date_to=date_to,
+    )
+    return int(session.execute(stmt).scalar_one())
+
+
+def list_items(session: Session, business_id: int) -> list[BusinessItem]:
+    """业务明细行（编辑回显与汇总重算共用）。"""
+    return list(
+        session.execute(
+            select(BusinessItem)
+            .where(BusinessItem.business_id == business_id)
+            .order_by(BusinessItem.id)
+        ).scalars()
+    )
+
+
+def replace_items(session: Session, business_id: int, rows: list[BusinessItem]) -> None:
+    """全量替换明细：先删、立即 flush、再插入（replace_roles 先例的防御写法）。
+
+    同一 flush 中 SQLAlchemy 先处理 INSERT 后处理 DELETE，先删后插不落库
+    会在有唯一键的表上撞键；business_items 虽无唯一键，统一按此写法防御。
+    """
+    session.execute(delete(BusinessItem).where(BusinessItem.business_id == business_id))
+    session.flush()
+    for row in rows:
+        row.business_id = business_id
+        session.add(row)
+    session.flush()
+
+
+def count_items(session: Session, business_ids: list[int]) -> dict[int, int]:
+    """业务明细行数（删除保护备用）。"""
+    if not business_ids:
+        return {}
+    rows = session.execute(
+        select(BusinessItem.business_id, func.count())
+        .where(BusinessItem.business_id.in_(business_ids))
+        .group_by(BusinessItem.business_id)
+    ).all()
+    return dict(rows)
+
+
+def add_status_log(session: Session, log: BusinessStatusLog) -> None:
+    """追加状态留痕（不 commit）。"""
+    session.add(log)
+    session.flush()
+
+
+def list_status_logs(
+    session: Session, business_id: int, *, offset: int, limit: int
+) -> list[BusinessStatusLog]:
+    """状态流转时间线（详情页），按时间正序（时间线从创建起展示）。"""
+    return list(
+        session.execute(
+            select(BusinessStatusLog)
+            .where(BusinessStatusLog.business_id == business_id)
+            .order_by(BusinessStatusLog.create_time, BusinessStatusLog.id)
+            .offset(offset)
+            .limit(limit)
+        ).scalars()
+    )
+
+
+def count_status_logs(session: Session, business_id: int) -> int:
+    return int(
+        session.execute(
+            select(func.count())
+            .select_from(BusinessStatusLog)
+            .where(BusinessStatusLog.business_id == business_id)
+        ).scalar_one()
+    )
+
+
+def add_ownership_log(session: Session, log: OwnershipChangeLog) -> None:
+    """追加归属变更留痕（不 commit）。"""
+    session.add(log)
+    session.flush()
+
+
+def list_ownership_logs(
+    session: Session, business_id: int, *, offset: int, limit: int
+) -> list[Row]:
+    """某业务的归属变更留痕（join 人员出姓名；人员删除后为 None，留痕固化 ID 兜底）。"""
+    old_p = aliased(Personnel)
+    new_p = aliased(Personnel)
+    return list(
+        session.execute(
+            select(OwnershipChangeLog, old_p.personnel_name, new_p.personnel_name)
+            .outerjoin(old_p, old_p.id == OwnershipChangeLog.old_owner_personnel_id)
+            .outerjoin(new_p, new_p.id == OwnershipChangeLog.new_owner_personnel_id)
+            .where(
+                OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
+                OwnershipChangeLog.business_id == business_id,
+            )
+            .order_by(OwnershipChangeLog.create_time.desc(), OwnershipChangeLog.id.desc())
+            .offset(offset)
+            .limit(limit)
+        ).all()
+    )
+
+
+def count_ownership_logs(session: Session, business_id: int) -> int:
+    return int(
+        session.execute(
+            select(func.count())
+            .select_from(OwnershipChangeLog)
+            .where(
+                OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
+                OwnershipChangeLog.business_id == business_id,
+            )
+        ).scalar_one()
+    )
diff --git a/backend/app/schemas/business.py b/backend/app/schemas/business.py
new file mode 100644
index 0000000..cfdf5b8
--- /dev/null
+++ b/backend/app/schemas/business.py
@@ -0,0 +1,324 @@
+"""业务管理的请求/响应模型（需求 2.2.1/2.2.4）。
+
+金额/数量入参出参一律 str（服务层 Decimal/Fraction，全程禁 float）；
+派生值（毛利润额/毛利润率/折算毛利润额）不在写接口入参内 —— 落库列只有
+源数据，派生值由毛利润视图实时计算。
+"""
+
+from datetime import date, datetime
+from decimal import Decimal, InvalidOperation
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from app.core.constants import (
+    PRODUCT_CATEGORIES,
+    PRODUCT_CATEGORY_MUNICIPAL_WATER,
+    PRODUCT_CATEGORY_RAIL_TRANSIT,
+)
+from app.core.response import PageQuery
+
+# DECIMAL(14,2) 容量上限：整数部分 12 位（同 personnel.py 的金额校验口径）
+AMOUNT_MAX = Decimal("999999999999.99")
+# DECIMAL(14,4) 容量上限：整数部分 10 位
+PRICE_MAX = Decimal("9999999999.9999")
+
+
+def _check_amount(
+    value: str | None, label: str, cap: Decimal, allow_positive_only: bool = False
+) -> str | None:
+    """字符串金额的通用校验：数字/有限/非负/容量/两位小数。
+
+    小数位超过两位直接拒绝而非静默舍入 —— 舍入即算错钱（工资系统红线）。
+    """
+    if value is None:
+        return None
+    try:
+        amount = Decimal(value)
+    except InvalidOperation:
+        raise ValueError(f"{label}须为有效数字") from None
+    if not amount.is_finite():
+        raise ValueError(f"{label}须为有效数字")
+    if allow_positive_only:
+        if amount <= 0:
+            raise ValueError(f"{label}必须大于 0")
+    elif amount < 0:
+        raise ValueError(f"{label}不能为负数")
+    if amount > cap:
+        raise ValueError(f"{label}超出可录入上限")
+    if -amount.as_tuple().exponent > 2:
+        raise ValueError(f"{label}最多两位小数")
+    return value
+
+
+class BusinessItemIn(BaseModel):
+    """业务产品明细行（需求 2.2.1「支持添加多种产品，录入指导价」）。
+
+    product_category 保留蛇形：该值与 M1 参数维度值域同源（枚举码本就是
+    蛇形）；其余标识类字段驼峰兼容（snake/驼峰双入，同 customer.py 契约）。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    product_id: int | None = Field(default=None, ge=1, description="产品ID；空=自由录入")
+    product_name: str = Field(min_length=1, max_length=128, description="产品名称")
+    spec_model: str | None = Field(
+        default=None, max_length=128, description="规格型号，空=空串归一"
+    )
+    unit: str | None = Field(default=None, max_length=16, description="计量单位")
+    product_category: str = Field(
+        description="产品类别：municipal_water/rail_transit（行级折算锚点）"
+    )
+    quantity: str = Field(description="数量，>0")
+    sale_price: str | None = Field(default=None, description="销售单价（元）")
+    cost_amount: str | None = Field(
+        default=None, description="成本额（元）；毛利润额 = 销售额 − 成本额"
+    )
+    guided_price: str | None = Field(default=None, description="指导价（元）")
+    remark: str | None = Field(default=None, max_length=255)
+
+    @field_validator("product_category")
+    @classmethod
+    def _check_category(cls, value: str) -> str:
+        if value not in PRODUCT_CATEGORIES:
+            raise ValueError("产品类别不合法")
+        return value
+
+    @field_validator("quantity")
+    @classmethod
+    def _check_quantity(cls, value: str) -> str:
+        return _check_amount(value, "数量", AMOUNT_MAX, allow_positive_only=True) or value
+
+    @field_validator("sale_price", "cost_amount", "guided_price")
+    @classmethod
+    def _check_prices(cls, value: str | None) -> str | None:
+        return _check_amount(value, "金额", PRICE_MAX)
+
+
+class BusinessItemOut(BaseModel):
+    """业务产品明细行响应。金额/数量一律 str。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    product_id: int | None = None
+    product_name: str
+    spec_model: str | None = None
+    unit: str | None = None
+    product_category: str
+    quantity: str
+    sale_price: str | None = None
+    sale_amount: str
+    cost_amount: str | None = None
+    guided_price: str | None = None
+    remark: str | None = None
+
+
+class BusinessOut(BaseModel):
+    """业务响应（列表与写操作回包共用）。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    business_no: str
+    customer_id: int
+    customer_name: str | None = None
+    business_date: date | None = None
+    status: int = 1
+    owner_personnel_id: int | None = None
+    owner_personnel_name: str | None = None
+    assistant_personnel_id: int | None = None
+    assistant_personnel_name: str | None = None
+    is_special: int = 0
+    special_note: str | None = None
+    total_sale_amount: str | None = None
+    total_cost_amount: str | None = None
+    total_ship_amount: str | None = None
+    total_received_amount: str | None = None
+    total_freight_amount: str | None = None
+    remark: str | None = None
+
+
+class BusinessDetailOut(BusinessOut):
+    """业务详情：附产品明细行。"""
+
+    items: list[BusinessItemOut] = Field(default_factory=list)
+
+
+class BusinessCreateIn(BaseModel):
+    """创建业务：主表信息 + 明细行数组（明细 ≥1 行）。
+
+    字段名全线蛇形直出（M1 既有契约：前端 api 封装与 schemas 同名对应，
+    仅分页信封 list/pageNum 驼峰）。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    customer_id: int = Field(ge=1, description="客户ID")
+    business_date: date | None = Field(default=None, description="业务日期，ISO 格式")
+    owner_personnel_id: int | None = Field(
+        default=None, ge=1, description="负责营销人员ID，可空=待指定"
+    )
+    assistant_personnel_id: int | None = Field(default=None, ge=1, description="协助营销人员ID")
+    is_special: int = Field(default=0, ge=0, le=1, description="特殊业务标记：1是 0否")
+    special_note: str | None = Field(default=None, max_length=255, description="特殊业务说明")
+    items: list[BusinessItemIn] = Field(min_length=1, description="产品明细行，至少 1 行")
+    remark: str | None = Field(default=None, max_length=255)
+
+    @model_validator(mode="after")
+    def _check_special_note(self) -> "BusinessCreateIn":
+        """特殊业务须有说明：标记的意义就在于可追溯「特殊在哪」。"""
+        if self.is_special == 1 and not self.special_note:
+            raise ValueError("特殊业务必须填写说明")
+        return self
+
+
+class BusinessUpdateIn(BaseModel):
+    """编辑业务（全量替换明细；归属不在此端点调整，走 /ownership）。"""
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    customer_id: int = Field(ge=1, description="客户ID")
+    business_date: date | None = Field(default=None, description="业务日期，ISO 格式")
+    owner_personnel_id: int | None = Field(
+        default=None,
+        ge=1,
+        description="负责营销人员ID（编辑端点忽略，仅 /ownership 生效）",
+    )
+    assistant_personnel_id: int | None = Field(
+        default=None,
+        ge=1,
+        description="协助营销人员ID（编辑端点忽略）",
+    )
+    is_special: int = Field(default=0, ge=0, le=1, description="特殊业务标记：1是 0否")
+    special_note: str | None = Field(default=None, max_length=255, description="特殊业务说明")
+    items: list[BusinessItemIn] = Field(
+        min_length=1, description="产品明细行，至少 1 行（全量替换）"
+    )
+    remark: str | None = Field(default=None, max_length=255)
+
+    @model_validator(mode="after")
+    def _check_special_note(self) -> "BusinessUpdateIn":
+        if self.is_special == 1 and not self.special_note:
+            raise ValueError("特殊业务必须填写说明")
+        return self
+
+
+class BusinessStatusIn(BaseModel):
+    """状态流转：1执行中 → 2已完成 / 3已终止（终态不可再变更）。"""
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    status: int = Field(ge=1, le=3, description="目标状态：2已完成 3已终止")
+    reason: str | None = Field(default=None, max_length=255, description="变更原因（终止时必填）")
+
+    @model_validator(mode="after")
+    def _check_reason(self) -> "BusinessStatusIn":
+        if self.status == 3 and not self.reason:
+            raise ValueError("终止业务必须填写原因")
+        return self
+
+
+class BusinessOwnershipIn(BaseModel):
+    """业务归属变更（负责/协助；前后值写入留痕）。"""
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    new_owner_personnel_id: int | None = Field(default=None, ge=1)
+    new_assistant_personnel_id: int | None = Field(default=None, ge=1)
+    reason: str | None = Field(default=None, max_length=255)
+
+
+class BusinessQuery(PageQuery):
+    """业务列表查询条件（筛选参数驼峰 alias，customer.py 同款契约）。"""
+
+    keyword: str | None = Field(default=None, max_length=32, description="业务编号/公司名称，模糊")
+    status: int | None = Field(default=None, ge=1, le=3, description="状态筛选")
+    customer_id: int | None = Field(default=None, ge=1, description="客户筛选")
+    owner_personnel_id: int | None = Field(default=None, ge=1, description="负责营销人员筛选")
+    date_from: date | None = Field(default=None, description="业务日期起（含）")
+    date_to: date | None = Field(default=None, description="业务日期止（含）")
+
+
+class GrossProfitRowOut(BaseModel):
+    """毛利润视图行：行级源数据 + 派生值（实时计算，不落库）。"""
+
+    item_id: int
+    product_category: str
+    product_name: str
+    spec_model: str | None = None
+    quantity: str
+    sale_amount: str
+    cost_amount: str | None = None
+    """成本缺失（未录入）时为 None，派生列同为 None 并置 missing_reason"""
+    profit_amount: str | None = None
+    profit_rate: str | None = None
+    threshold: str
+    converted_profit: str | None = None
+    missing_reason: str | None = None
+
+
+class GrossProfitCategoryOut(BaseModel):
+    """按产品类别的毛利润汇总（可计算行聚合；成本缺失行不计入）。"""
+
+    product_category: str
+    sale_amount: str
+    cost_amount: str
+    profit_amount: str
+    profit_rate: str
+    threshold: str
+    converted_profit: str
+
+
+class GrossProfitOut(BaseModel):
+    """毛利润数据视图（需求 2.2.4）：行级 + 类别汇总。
+
+    口径注释（前端须展示）：毛利润率 = 毛利润额 / 销售额，**不含运费**
+    （对照表 #2 作废需规含运费公式）；阈值按 asOf 取 M1 参数快照；M3 核算
+    实际按回款到账日取参，本视图默认今日仅供录入期核对。
+    """
+
+    business_id: int
+    as_of: date
+    rows: list[GrossProfitRowOut] = Field(default_factory=list)
+    categories: list[GrossProfitCategoryOut] = Field(default_factory=list)
+    missing_count: int = 0
+    note: str = "毛利润率与折算额不含运费；M3 核算按回款到账日取参数版本"
+
+
+class OwnershipChangeLogOut(BaseModel):
+    """业务归属变更留痕（人员删除后姓名为 None，固化 ID 兜底可追溯）。"""
+
+    id: int
+    business_id: int
+    old_owner_personnel_id: int | None = None
+    new_owner_personnel_id: int | None = None
+    old_assistant_personnel_id: int | None = None
+    new_assistant_personnel_id: int | None = None
+    old_owner_name: str | None = None
+    new_owner_name: str | None = None
+    change_reason: str | None = None
+    operator_id: int | None = None
+    operator_name: str
+    create_time: datetime
+
+
+class BusinessStatusLogOut(BaseModel):
+    """业务状态流转记录（详情页时间线；创建首条 old=new=1）。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    business_id: int
+    old_status: int
+    new_status: int
+    change_reason: str | None = None
+    operator_id: int | None = None
+    operator_name: str
+    create_time: datetime
+
+
+# 供 schema 外部引用的两类别常量（类别汇总展示用）
+CATEGORY_LABELS = {
+    PRODUCT_CATEGORY_MUNICIPAL_WATER: "市政水利",
+    PRODUCT_CATEGORY_RAIL_TRANSIT: "轨道交通",
+}
diff --git a/backend/app/schemas/customer.py b/backend/app/schemas/customer.py
index 3c905d6..fde79e3 100644
--- a/backend/app/schemas/customer.py
+++ b/backend/app/schemas/customer.py
@@ -35,11 +35,11 @@ class CustomerCreateIn(BaseModel):
 
     owner_personnel_id 为**初始归属**，可空（客户先建档、后派归属也合法）；
     归属的后续调整不走本模型，统一走 ``PUT /{id}/ownership``（见
-    ``CustomerUpdateIn`` 的说明）。camelCase 别名与蛇形字段名均可绑定，
-    兼容前端两种拼写。
+    ``CustomerUpdateIn`` 的说明）。字段名蛇形直出（M1 契约：前端 api 封装
+    与 schema 同名对应，仅分页信封 list/pageNum 驼峰）。
     """
 
-    model_config = ConfigDict(str_strip_whitespace=True, populate_by_name=True)
+    model_config = ConfigDict(str_strip_whitespace=True)
 
     company_name: str = Field(min_length=1, max_length=128, description="公司名称")
     owner_name: str = Field(
@@ -59,7 +59,6 @@ class CustomerCreateIn(BaseModel):
     owner_personnel_id: int | None = Field(
         default=None,
         ge=1,
-        alias="ownerPersonnelId",
         description="归属负责营销人员ID（初始归属，可空）",
     )
     remark: str | None = Field(default=None, max_length=255, description="备注")
@@ -91,21 +90,13 @@ class CustomerUpdateIn(BaseModel):
 
 
 class CustomerQuery(PageQuery):
-    """客户列表查询条件。
-
-    customer_type / owner_personnel_id 同时接受蛇形（项目既有契约，见前端
-    `types.ts` 的说明）与驼峰别名，两种拼写不并存即可。
-    """
+    """客户列表查询条件（字段名蛇形直出，与前端 api 封装同名对应）。"""
 
     keyword: str | None = Field(
         default=None, max_length=64, description="公司名称或客户负责人，模糊匹配"
     )
-    customer_type: str | None = Field(
-        default=None, max_length=32, alias="customerType", description="客户分类筛选"
-    )
-    owner_personnel_id: int | None = Field(
-        default=None, ge=1, alias="ownerPersonnelId", description="归属人员ID筛选"
-    )
+    customer_type: str | None = Field(default=None, max_length=32, description="客户分类筛选")
+    owner_personnel_id: int | None = Field(default=None, ge=1, description="归属人员ID筛选")
 
 
 class CustomerOwnershipIn(BaseModel):
@@ -114,12 +105,9 @@ class CustomerOwnershipIn(BaseModel):
     new_owner_personnel_id 可空：null 表示清空归属（客户暂无负责人也合法）。
     """
 
-    model_config = ConfigDict(populate_by_name=True)
-
     new_owner_personnel_id: int | None = Field(
         default=None,
         ge=1,
-        alias="newOwnerPersonnelId",
         description="变更后归属人员ID，null=清空归属",
     )
     reason: str | None = Field(default=None, max_length=255, description="变更原因")
diff --git a/backend/app/services/business_service.py b/backend/app/services/business_service.py
new file mode 100644
index 0000000..0c92e1d
--- /dev/null
+++ b/backend/app/services/business_service.py
@@ -0,0 +1,747 @@
+"""业务管理业务编排（需求 2.2.1/2.2.4）。
+
+事务边界在本层：repository 只 flush，commit/rollback 只写在这里。
+
+- 业务编号自动生成（no_generator.next_no）：MAX+FOR UPDATE 锁当日序号区间，
+  残余竞态以唯一键（1062）冲突重试一次兜底
+- 五项汇总冗余（total_sale/cost/ship/received/freight）由 ``_recalc_totals``
+  单点重算 —— 明细/发货/回款任何写路径收口到此函数，禁止各处手写
+- 状态机：1执行中 → 2已完成 / 3已终止；终态不可再变更（一期口径，开放问题 #8）；
+  终止必填原因；每次流转写 business_status_logs（创建写首条 old=new=1）
+- 归属变更（负责/协助）写 ownership_change_logs（owner_type=2），同事务一次
+  commit；只影响变更后核算，不回溯历史（对照表 #19）
+- 毛利润视图为只读派生值（复用 formula.profit_rate/converted_profit），
+  阈值按 asOf 取 M1 参数快照 —— 落库即漂移，M3 台账才是固化点
+"""
+
+import logging
+from datetime import date
+from decimal import Decimal
+from fractions import Fraction
+
+from sqlalchemy import Row, func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core.constants import (
+    ACTION_CREATE,
+    ACTION_DELETE,
+    ACTION_UPDATE,
+    MODULE_BUSINESS,
+    NO_PREFIX_BUSINESS,
+)
+from app.core.exceptions import BadRequestError, NotFoundError
+from app.core.response import PageQuery, PageResult
+from app.models import (
+    Business,
+    BusinessItem,
+    BusinessStatusLog,
+    Customer,
+    OwnershipChangeLog,
+    Personnel,
+    Receipt,
+    Shipment,
+    User,
+)
+from app.models.business import (
+    BUSINESS_STATUS_ACTIVE,
+    BUSINESS_STATUS_DONE,
+)
+from app.models.customer import OWNER_TYPE_BUSINESS
+from app.repositories import business_repository, customer_repository, personnel_repository
+from app.schemas.business import (
+    BusinessCreateIn,
+    BusinessDetailOut,
+    BusinessItemIn,
+    BusinessItemOut,
+    BusinessOut,
+    BusinessOwnershipIn,
+    BusinessQuery,
+    BusinessStatusIn,
+    BusinessStatusLogOut,
+    BusinessUpdateIn,
+    GrossProfitCategoryOut,
+    GrossProfitOut,
+    GrossProfitRowOut,
+    OwnershipChangeLogOut,
+)
+from app.services import audit_service, param_service
+from app.services.audit_service import AuditRecord, RequestMeta
+from app.services.formula import converted_profit, profit_rate
+from app.services.no_generator import next_no
+
+logger = logging.getLogger(__name__)
+
+# 编号唯一键冲突后的重试次数：MAX+FOR UPDATE 已覆盖绝大多数竞态，
+# 重试是为极端场景（锁等待超时后重入）兜底 —— 无限重试反而掩盖死锁类缺陷
+_NO_RETRY = 1
+
+# personnel.status 的在职值（与 customer_service 同口径，注释见彼处）
+_PERSONNEL_STATUS_ACTIVE = 1
+
+
+def _to_out(row: tuple) -> BusinessOut:
+    """(business, company_name, owner_name, assistant_name) → 响应模型。
+
+    金额冗余列出参 str；汇总列可能为 NULL（server_default 只对 INSERT 生效，
+    存量行可能为 NULL），统一转 "0" 保持契约稳定。
+    """
+    business, company_name, owner_name, assistant_name = row
+    return BusinessOut(
+        id=business.id,
+        business_no=business.business_no,
+        customer_id=business.customer_id,
+        customer_name=company_name,
+        business_date=business.business_date,
+        status=business.status,
+        owner_personnel_id=business.owner_personnel_id,
+        owner_personnel_name=owner_name,
+        assistant_personnel_id=business.assistant_personnel_id,
+        assistant_personnel_name=assistant_name,
+        is_special=business.is_special,
+        special_note=business.special_note,
+        total_sale_amount=_str_or_zero(business.total_sale_amount),
+        total_cost_amount=_str_or_zero(business.total_cost_amount),
+        total_ship_amount=_str_or_zero(business.total_ship_amount),
+        total_received_amount=_str_or_zero(business.total_received_amount),
+        total_freight_amount=_str_or_zero(business.total_freight_amount),
+        remark=business.remark,
+    )
+
+
+def _str_or_zero(value: Decimal | None) -> str:
+    return str(value) if value is not None else "0"
+
+
+def _item_to_out(item: BusinessItem) -> BusinessItemOut:
+    return BusinessItemOut(
+        id=item.id,
+        product_id=item.product_id,
+        product_name=item.product_name,
+        spec_model=item.spec_model,
+        unit=item.unit,
+        product_category=item.product_category,
+        quantity=str(item.quantity),
+        sale_price=str(item.sale_price) if item.sale_price is not None else None,
+        sale_amount=str(item.sale_amount),
+        cost_amount=str(item.cost_amount) if item.cost_amount is not None else None,
+        guided_price=str(item.guided_price) if item.guided_price is not None else None,
+        remark=item.remark,
+    )
+
+
+def _build_item_rows(business_id: int, items: list[BusinessItemIn]) -> list[BusinessItem]:
+    """入参明细行转 ORM 行：sale_amount 服务层按 数量×单价 计算固化。
+
+    不信任前端传值（schema 亦无 sale_amount 入参）——单价可空时销售额记 0，
+    由用户在编辑中补价后重算（毛利润视图中销售额 0 的行全部为 0，防除零）。
+    """
+    rows: list[BusinessItem] = []
+    for item in items:
+        quantity = Decimal(item.quantity)
+        sale_price = Decimal(item.sale_price) if item.sale_price is not None else None
+        sale_amount = (quantity * sale_price) if sale_price is not None else Decimal("0")
+        rows.append(
+            BusinessItem(
+                business_id=business_id,
+                product_id=item.product_id,
+                product_name=item.product_name,
+                spec_model=item.spec_model or "",
+                unit=item.unit,
+                product_category=item.product_category,
+                quantity=quantity,
+                sale_price=sale_price,
+                sale_amount=sale_amount,
+                cost_amount=Decimal(item.cost_amount) if item.cost_amount is not None else None,
+                guided_price=Decimal(item.guided_price) if item.guided_price is not None else None,
+                remark=item.remark,
+            )
+        )
+    return rows
+
+
+def _recalc_totals(session: Session, business: Business) -> None:
+    """重算业务的五项汇总冗余（明细/发货/回款写路径统一收口）。
+
+    五条独立聚合查询（各表单列 SUM，走 business_id 索引）——业务单据量级
+    为单业务数行，不值得为省查询做单大 JOIN。
+    """
+    sale = session.execute(
+        _sum(BusinessItem.sale_amount, BusinessItem.business_id, business.id)
+    ).scalar_one()
+    cost = session.execute(
+        _sum(BusinessItem.cost_amount, BusinessItem.business_id, business.id)
+    ).scalar_one()
+    ship = session.execute(_sum_ship_items(business.id)).scalar_one()
+    freight = session.execute(
+        _sum(Shipment.freight_amount, Shipment.business_id, business.id)
+    ).scalar_one()
+    received = session.execute(_sum(Receipt.amount, Receipt.business_id, business.id)).scalar_one()
+    business.total_sale_amount = sale or Decimal("0")
+    business.total_cost_amount = cost or Decimal("0")
+    business.total_ship_amount = ship or Decimal("0")
+    business.total_freight_amount = freight or Decimal("0")
+    business.total_received_amount = received or Decimal("0")
+
+
+def _sum(column, match_column, value) -> object:
+    return select(func.sum(column)).where(match_column == value)
+
+
+def _sum_ship_items(business_id: int):
+    """发货明细行金额合计（经发货单关联到业务）。"""
+    from app.models import ShipmentItem
+
+    return (
+        select(func.sum(ShipmentItem.amount))
+        .select_from(ShipmentItem)
+        .join(Shipment, Shipment.id == ShipmentItem.shipment_id)
+        .where(Shipment.business_id == business_id)
+    )
+
+
+def _generate_no(session: Session) -> str:
+    """生成业务编号（MAX+FOR UPDATE；1062 兜底重试一次，见模块 docstring）。"""
+    return next_no(session, model=Business, column_name="business_no", prefix=NO_PREFIX_BUSINESS)
+
+
+def _ensure_personnel_active(session: Session, personnel_id: int, label: str) -> Personnel:
+    """归属人员须存在且在职（人员侧复用校验文案口径）。"""
+    person = personnel_repository.get_by_id(session, personnel_id)
+    if person is None or person.status != _PERSONNEL_STATUS_ACTIVE:
+        raise BadRequestError(f"{label}人员不存在或已离职")
+    return person
+
+
+def _ensure_customer(session: Session, customer_id: int) -> Customer:
+    customer = customer_repository.get_by_id(session, customer_id)
+    if customer is None:
+        raise NotFoundError("客户不存在")
+    return customer
+
+
+def _is_duplicate_entry(exc: IntegrityError) -> bool:
+    args = getattr(exc.orig, "args", ())
+    return bool(args) and args[0] == 1062
+
+
+def list_businesses(session: Session, query: BusinessQuery) -> PageResult[BusinessOut]:
+    """分页查询业务（附客户名与归属人名）。"""
+    rows = business_repository.list_businesses(
+        session,
+        offset=query.offset,
+        limit=query.page_size,
+        keyword=query.keyword,
+        status=query.status,
+        customer_id=query.customer_id,
+        owner_personnel_id=query.owner_personnel_id,
+        date_from=query.date_from,
+        date_to=query.date_to,
+    )
+    total = business_repository.count_businesses(
+        session,
+        keyword=query.keyword,
+        status=query.status,
+        customer_id=query.customer_id,
+        owner_personnel_id=query.owner_personnel_id,
+        date_from=query.date_from,
+        date_to=query.date_to,
+    )
+    return PageResult[BusinessOut](
+        records=[_to_out(row) for row in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def get_detail(session: Session, business_id: int) -> BusinessDetailOut:
+    """业务详情（附明细行，供编辑抽屉回显）。
+
+    :raises NotFoundError: 业务不存在
+    """
+    row = business_repository.get_with_names(session, business_id)
+    if row is None:
+        raise NotFoundError("业务不存在")
+    out = BusinessDetailOut(**_to_out(row).model_dump())
+    out.items = [
+        _item_to_out(item) for item in business_repository.list_items(session, business_id)
+    ]
+    return out
+
+
+def create_business(
+    session: Session,
+    payload: BusinessCreateIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> BusinessOut:
+    """创建业务 + 明细行 + 首条状态留痕，一个事务。
+
+    创建时指定的归属不写归属留痕（留痕记「变更」，创建即初始事实）。
+
+    :raises NotFoundError: 客户不存在
+    :raises BadRequestError: 归属人员不存在或已离职 / 明细校验不通过
+    """
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_BUSINESS,
+            action=ACTION_CREATE,
+            summary=f"创建业务（客户ID={payload.customer_id}，明细 {len(payload.items)} 行）",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        _ensure_customer(session, payload.customer_id)
+        if payload.owner_personnel_id is not None:
+            _ensure_personnel_active(session, payload.owner_personnel_id, "负责")
+        if payload.assistant_personnel_id is not None:
+            _ensure_personnel_active(session, payload.assistant_personnel_id, "协助")
+
+        try:
+            business = Business(
+                business_no=_generate_no(session),
+                customer_id=payload.customer_id,
+                business_date=payload.business_date,
+                status=BUSINESS_STATUS_ACTIVE,
+                owner_personnel_id=payload.owner_personnel_id,
+                assistant_personnel_id=payload.assistant_personnel_id,
+                is_special=payload.is_special,
+                special_note=payload.special_note,
+                remark=payload.remark,
+            )
+            business_repository.create(session, business)
+            business_repository.replace_items(
+                session, business.id, _build_item_rows(business.id, payload.items)
+            )
+            _recalc_totals(session, business)
+            business_repository.add_status_log(
+                session,
+                BusinessStatusLog(
+                    business_id=business.id,
+                    old_status=BUSINESS_STATUS_ACTIVE,
+                    new_status=BUSINESS_STATUS_ACTIVE,
+                    operator_id=operator.id,
+                    operator_name=operator.username,
+                ),
+            )
+            customer = customer_repository.get_by_id(session, business.customer_id)
+            out = _to_out((business, customer.company_name if customer else None, None, None))
+            session.commit()
+        except IntegrityError as exc:
+            session.rollback()
+            if _is_duplicate_entry(exc):
+                # 编号生成的残余竞态：唯一键兜底重试一次（防无限重试掩盖死锁）
+                return _retry_create(session, payload, operator=operator, meta=meta)
+            raise
+
+        logger.info("创建业务 | business_id=%s | no=%s", business.id, business.business_no)
+        return out
+
+
+def _retry_create(
+    session: Session,
+    payload: BusinessCreateIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None,
+) -> BusinessOut:
+    """编号撞唯一键后的单次重试（防无限重试掩盖死锁类缺陷）。"""
+    return create_business(session, payload, operator=operator, meta=meta)
+
+
+def update_business(
+    session: Session,
+    business_id: int,
+    payload: BusinessUpdateIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> BusinessOut:
+    """编辑业务：全量替换明细 + 重算汇总；归属不在此端点（防旁路）。
+
+    客户可改，但有发货或回款记录后禁改 —— 单据 FK 挂在业务上，换客户会让
+    发货/回款语义断裂（明细同理随业务走）。
+
+    :raises NotFoundError: 业务不存在
+    :raises BadRequestError: 归属人员不合法 / 有发货或回款时换客户
+    """
+    row = business_repository.get_with_names(session, business_id)
+    business = row[0] if row else None
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_BUSINESS,
+            action=ACTION_UPDATE,
+            summary=f"修改业务 {business.business_no if business else f'#{business_id}'}",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if row is None:
+            raise NotFoundError("业务不存在")
+        business, _, _, _ = row
+
+        if payload.customer_id != business.customer_id:
+            has_docs = session.query(Shipment.id).filter(
+                Shipment.business_id == business_id
+            ).first() is not None or (
+                session.query(Receipt.id).filter(Receipt.business_id == business_id).first()
+                is not None
+            )
+            if has_docs:
+                raise BadRequestError("业务已有发货或回款记录，不能更换客户")
+        _ensure_customer(session, payload.customer_id)
+        if payload.owner_personnel_id is not None:
+            _ensure_personnel_active(session, payload.owner_personnel_id, "负责")
+        if payload.assistant_personnel_id is not None:
+            _ensure_personnel_active(session, payload.assistant_personnel_id, "协助")
+
+        business.customer_id = payload.customer_id
+        business.business_date = payload.business_date
+        business.is_special = payload.is_special
+        business.special_note = payload.special_note
+        business.remark = payload.remark
+        # 归属刻意不赋值：留痕只由 /ownership 端点产生，编辑路径改归属构成
+        # 「改归属不留痕」的旁路（与 customers 编辑不触归属同一口径）
+        business_repository.replace_items(
+            session, business_id, _build_item_rows(business_id, payload.items)
+        )
+        _recalc_totals(session, business)
+        out = _to_out(business_repository.get_with_names(session, business_id))
+        session.commit()
+        logger.info("修改业务 | business_id=%s", business_id)
+        return out
+
+
+def delete_business(
+    session: Session,
+    business_id: int,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> None:
+    """删除业务（物理删除）：有发货/回款 → 400 引导走「已终止」。
+
+    无单据引用的业务（建错的空单）方可删除；删除时连带清理明细、状态留痕
+    与业务维度的归属留痕（无外键固化表，须手动清）。
+
+    :raises NotFoundError: 业务不存在
+    :raises BadRequestError: 存在发货或回款记录
+    """
+    business = business_repository.get_by_id(session, business_id, for_update=True)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_BUSINESS,
+            action=ACTION_DELETE,
+            summary=f"删除业务 {business.business_no if business else f'#{business_id}'}",
+        ),
+        meta=meta,
+    ):
+        if business is None:
+            raise NotFoundError("业务不存在")
+        if (
+            session.query(Shipment.id).filter(Shipment.business_id == business_id).first()
+            is not None
+            or session.query(Receipt.id).filter(Receipt.business_id == business_id).first()
+            is not None
+        ):
+            raise BadRequestError("业务已有发货或回款记录，请改为已终止而非删除")
+
+        business_no = business.business_no
+        _hard_delete(session, business_id)
+        session.commit()
+        logger.info("删除业务 | business_id=%s | no=%s", business_id, business_no)
+
+
+def _hard_delete(session: Session, business_id: int) -> None:
+    """按依赖序删除：明细 → 状态留痕 → 业务维度归属留痕 → 主表。"""
+    from sqlalchemy import delete
+
+    session.execute(delete(BusinessItem).where(BusinessItem.business_id == business_id))
+    session.execute(delete(BusinessStatusLog).where(BusinessStatusLog.business_id == business_id))
+    session.execute(
+        delete(OwnershipChangeLog).where(
+            OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
+            OwnershipChangeLog.business_id == business_id,
+        )
+    )
+    session.execute(delete(Business).where(Business.id == business_id))
+
+
+def change_status(
+    session: Session,
+    business_id: int,
+    payload: BusinessStatusIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> BusinessOut:
+    """状态流转：1→2 / 1→3，终态不可再变更；每次流转写留痕。
+
+    :raises NotFoundError: 业务不存在
+    :raises BadRequestError: 非执行中状态流转 / 目标与当前相同
+    """
+    row = business_repository.get_with_names(session, business_id)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_BUSINESS,
+            action=ACTION_UPDATE,
+            summary=(
+                f"业务状态变更 {row[0].business_no if row else f'#{business_id}'} → "
+                f"{'已完成' if payload.status == BUSINESS_STATUS_DONE else '已终止'}"
+            ),
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if row is None:
+            raise NotFoundError("业务不存在")
+        business, _, _, _ = row
+        if business.status != BUSINESS_STATUS_ACTIVE:
+            raise BadRequestError("已完成或已终止的业务不可再变更状态")
+        if business.status == payload.status:
+            raise BadRequestError("状态未变化")
+
+        business.status = payload.status
+        business_repository.add_status_log(
+            session,
+            BusinessStatusLog(
+                business_id=business_id,
+                old_status=business.status,
+                new_status=payload.status,
+                change_reason=payload.reason,
+                operator_id=operator.id,
+                operator_name=operator.username,
+            ),
+        )
+        out = _to_out(business_repository.get_with_names(session, business_id))
+        session.commit()
+        logger.info(
+            "业务状态变更 | business_id=%s | %s -> %s",
+            business_id,
+            business.status,
+            payload.status,
+        )
+        return out
+
+
+def change_ownership(
+    session: Session,
+    business_id: int,
+    payload: BusinessOwnershipIn,
+    *,
+    operator: User,
+    meta: RequestMeta | None = None,
+) -> BusinessOut:
+    """归属变更（负责/协助）：同事务「更新主表 + 插留痕」，一次 commit。
+
+    负责与协助之一发生变化即记一条留痕（前后值都含两个字段）；两者均未变化
+    → 400「归属未变化」。
+
+    :raises NotFoundError: 业务不存在
+    :raises BadRequestError: 归属人员不存在或已离职 / 归属未变化
+    """
+    row = business_repository.get_with_names(session, business_id)
+    with audit_service.audit(
+        session,
+        operator=operator,
+        record=AuditRecord(
+            module=MODULE_BUSINESS,
+            action=ACTION_UPDATE,
+            summary=f"调整业务归属 {row[0].business_no if row else f'#{business_id}'}",
+            request_data=payload.model_dump(exclude_none=True),
+        ),
+        meta=meta,
+    ):
+        if row is None:
+            raise NotFoundError("业务不存在")
+        business, _, _, _ = row
+        if payload.new_owner_personnel_id is not None:
+            _ensure_personnel_active(session, payload.new_owner_personnel_id, "负责")
+        if payload.new_assistant_personnel_id is not None:
+            _ensure_personnel_active(session, payload.new_assistant_personnel_id, "协助")
+
+        owner_changed = business.owner_personnel_id != payload.new_owner_personnel_id
+        assistant_changed = business.assistant_personnel_id != payload.new_assistant_personnel_id
+        if not owner_changed and not assistant_changed:
+            raise BadRequestError("归属未变化")
+
+        old_owner_id = business.owner_personnel_id
+        old_assistant_id = business.assistant_personnel_id
+        business.owner_personnel_id = payload.new_owner_personnel_id
+        business.assistant_personnel_id = payload.new_assistant_personnel_id
+        business_repository.add_ownership_log(
+            session,
+            OwnershipChangeLog(
+                owner_type=OWNER_TYPE_BUSINESS,
+                business_id=business_id,
+                old_owner_personnel_id=old_owner_id,
+                new_owner_personnel_id=payload.new_owner_personnel_id,
+                old_assistant_personnel_id=old_assistant_id,
+                new_assistant_personnel_id=payload.new_assistant_personnel_id,
+                change_reason=payload.reason,
+                operator_id=operator.id,
+                operator_name=operator.username,
+            ),
+        )
+        out = _to_out(business_repository.get_with_names(session, business_id))
+        session.commit()
+        logger.info(
+            "调整业务归属 | business_id=%s | owner %s->%s | assistant %s->%s",
+            business_id,
+            old_owner_id,
+            payload.new_owner_personnel_id,
+            old_assistant_id,
+            payload.new_assistant_personnel_id,
+        )
+        return out
+
+
+def list_ownership_logs(
+    session: Session, business_id: int, query: PageQuery
+) -> PageResult[OwnershipChangeLogOut]:
+    """某业务的归属变更留痕（分页）。
+
+    :raises NotFoundError: 业务不存在
+    """
+    if business_repository.get_by_id(session, business_id) is None:
+        raise NotFoundError("业务不存在")
+    rows = business_repository.list_ownership_logs(
+        session, business_id, offset=query.offset, limit=query.page_size
+    )
+    total = business_repository.count_ownership_logs(session, business_id=business_id)
+    return PageResult[OwnershipChangeLogOut](
+        records=[_log_to_out(row) for row in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def _log_to_out(row: Row) -> OwnershipChangeLogOut:
+    log, old_owner_name, new_owner_name = row
+    return OwnershipChangeLogOut(
+        id=log.id,
+        business_id=log.business_id,
+        old_owner_personnel_id=log.old_owner_personnel_id,
+        new_owner_personnel_id=log.new_owner_personnel_id,
+        old_assistant_personnel_id=log.old_assistant_personnel_id,
+        new_assistant_personnel_id=log.new_assistant_personnel_id,
+        old_owner_name=old_owner_name,
+        new_owner_name=new_owner_name,
+        change_reason=log.change_reason,
+        operator_id=log.operator_id,
+        operator_name=log.operator_name,
+        create_time=log.create_time,
+    )
+
+
+def list_status_logs(
+    session: Session, business_id: int, query: PageQuery
+) -> PageResult[BusinessStatusLogOut]:
+    """某业务的状态流转时间线（分页，时间正序）。
+
+    :raises NotFoundError: 业务不存在
+    """
+    if business_repository.get_by_id(session, business_id) is None:
+        raise NotFoundError("业务不存在")
+    rows = business_repository.list_status_logs(
+        session, business_id, offset=query.offset, limit=query.page_size
+    )
+    total = business_repository.count_status_logs(session, business_id=business_id)
+    return PageResult[BusinessStatusLogOut](
+        records=[BusinessStatusLogOut.model_validate(row) for row in rows],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def gross_profit(session: Session, business_id: int, as_of: date | None) -> GrossProfitOut:
+    """毛利润数据视图（需求 2.2.4）：行级派生值 + 类别汇总，实时计算不落库。
+
+    阈值按 asOf（默认今日）取 M1 参数快照（product.threshold，按行类别维度，
+    存储口径为 % 数如 30，直接进 converted_profit）；毛利润率按 M1 预览同一
+    出参口径输出 Fraction 分数字符串（如 "97/10000"）；成本缺失的行派生列为
+    None 并标注原因；汇总只聚合可计算行。
+
+    :raises NotFoundError: 业务不存在
+    """
+    if business_repository.get_by_id(session, business_id) is None:
+        raise NotFoundError("业务不存在")
+    as_of = as_of or date.today()
+    snapshot = param_service.resolve_snapshot(session, as_of)
+
+    def threshold_of(category: str) -> Fraction:
+        return snapshot.fraction("product.threshold", product_category=category)
+
+    rows_out: list[GrossProfitRowOut] = []
+    missing_count = 0
+    # 类别聚合容器：{类别: [sale, cost]}，成本缺失行不计入该类别聚合
+    agg: dict[str, list[Decimal]] = {}
+    for item in business_repository.list_items(session, business_id):
+        threshold = threshold_of(item.product_category)
+        row_kwargs: dict = {
+            "item_id": item.id,
+            "product_category": item.product_category,
+            "product_name": item.product_name,
+            "spec_model": item.spec_model,
+            "quantity": str(item.quantity),
+            "sale_amount": str(item.sale_amount),
+            "threshold": str(threshold),
+        }
+        if item.cost_amount is None:
+            missing_count += 1
+            row_kwargs.update(
+                cost_amount=None,
+                profit_amount=None,
+                profit_rate=None,
+                converted_profit=None,
+                missing_reason="成本缺失，无法计算毛利润",
+            )
+        else:
+            profit = item.sale_amount - item.cost_amount
+            rate = profit_rate(profit, item.sale_amount)
+            conv = converted_profit(item.sale_amount, rate, threshold)
+            row_kwargs.update(
+                cost_amount=str(item.cost_amount),
+                profit_amount=str(profit),
+                profit_rate=str(rate),
+                converted_profit=str(conv),
+                missing_reason=None,
+            )
+            bucket = agg.setdefault(item.product_category, [Decimal("0"), Decimal("0")])
+            bucket[0] += item.sale_amount
+            bucket[1] += item.cost_amount
+        rows_out.append(GrossProfitRowOut(**row_kwargs))
+
+    categories = [
+        GrossProfitCategoryOut(
+            product_category=category,
+            sale_amount=str(sale),
+            cost_amount=str(cost),
+            profit_amount=str(sale - cost),
+            profit_rate=str(profit_rate(sale - cost, sale)),
+            threshold=str(threshold_of(category)),
+            converted_profit=str(
+                converted_profit(sale, profit_rate(sale - cost, sale), threshold_of(category))
+            ),
+        )
+        for category, [sale, cost] in agg.items()
+    ]
+    return GrossProfitOut(
+        business_id=business_id,
+        as_of=as_of,
+        rows=rows_out,
+        categories=categories,
+        missing_count=missing_count,
+    )
diff --git a/backend/app/services/no_generator.py b/backend/app/services/no_generator.py
new file mode 100644
index 0000000..447f227
--- /dev/null
+++ b/backend/app/services/no_generator.py
@@ -0,0 +1,46 @@
+"""单据编号生成器（业务 YW / 发货 SH / 回款 HK 共用）。
+
+规则：``{前缀}{YYYYMMDD}-{当日4位序号}``，序号按自然日重置。
+并发安全：``SELECT MAX(no) ... FOR UPDATE`` 锁住当日序号区间 —— 后到的
+事务会等待先到者提交后再读到新 MAX，序号单调不重；残余竞态（锁等待超时
+等极端场景）由服务层捕唯一键冲突（1062）重试一次兜底。
+"""
+
+from datetime import datetime
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+
+def today_prefix(prefix: str) -> str:
+    """当日编号前缀（如 YW20261013-），LIKE 匹配与拼号共用同一来源防漂移。"""
+    return f"{prefix}{datetime.now().strftime('%Y%m%d')}-"
+
+
+def next_no(
+    session: Session,
+    *,
+    model: type,
+    column_name: str,
+    prefix: str,
+    seq_width: int = 4,
+) -> str:
+    """生成下一个编号。
+
+    :param model: 编号所在的 ORM 模型（businesses/shipments/receipts）
+    :param column_name: 编号列名（business_no/shipment_no/receipt_batch_no）
+    :param prefix: 单据前缀（NO_PREFIX_* 常量）
+    :param seq_width: 序号位数（默认 4）；超出宽度的序号按实际位数输出，
+        VARCHAR(32) 余量充足、唯一性不受影响
+    """
+    like_prefix = today_prefix(prefix)
+    column = getattr(model, column_name)
+    max_no = session.execute(
+        select(column)
+        .where(column.like(f"{like_prefix}%"))
+        .order_by(column.desc())
+        .limit(1)
+        .with_for_update()
+    ).scalar()
+    seq = int(max_no.rsplit("-", 1)[1]) + 1 if max_no else 1
+    return f"{like_prefix}{seq:0{seq_width}d}"