diff --git a/api/routers/product.py b/api/routers/product.py
new file mode 100644
index 0000000..81928d2
--- /dev/null
+++ b/api/routers/product.py
@@ -0,0 +1,49 @@
+"""产品路由：列表 + 历史业绩（匿名可访问，业务在 service/product.py）。"""
+from datetime import date
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config.deps import get_db
+from service.product import get_history, list_products
+from utils.response import success
+
+router = APIRouter()
+
+
+@router.get("/products")
+async def products(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(10, ge=1, le=100),
+    keyword: str | None = Query(None, max_length=64),
+    product_type: str | None = Query(None, max_length=32),
+    risk_level: str | None = Query(None, max_length=8),
+    status: str = Query("在售", max_length=16),
+    sort_by: str = Query("create_time", max_length=32),
+    sort_order: str = Query("desc", max_length=4),
+    db: AsyncSession = Depends(get_db),
+):
+    data = await list_products(
+        db,
+        page=page,
+        page_size=page_size,
+        keyword=keyword,
+        product_type=product_type,
+        risk_level=risk_level,
+        status=status,
+        sort_by=sort_by,
+        sort_order=sort_order,
+    )
+    return success(data)
+
+
+@router.get("/products/{product_code}/history")
+async def history(
+    product_code: str,
+    start_date: date | None = Query(None),
+    end_date: date | None = Query(None),
+    series: str = Query("nav", max_length=16),
+    db: AsyncSession = Depends(get_db),
+):
+    data = await get_history(db, product_code, start_date, end_date, series)
+    return success(data)
diff --git a/api/routers/questionnaire.py b/api/routers/questionnaire.py
new file mode 100644
index 0000000..b62ad42
--- /dev/null
+++ b/api/routers/questionnaire.py
@@ -0,0 +1,23 @@
+"""问卷路由：风险测评提交（需登录，业务在 service/questionnaire.py）。"""
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.deps import get_current_user
+from config.deps import get_db
+from model.sys_user import SysUser
+from schemas.questionnaire import RiskAssessmentSubmitReq
+from service.questionnaire import submit_assessment
+from utils.response import success
+
+router = APIRouter()
+
+
+@router.post("/risk-assessment/submit")
+async def submit(
+    body: RiskAssessmentSubmitReq,
+    user: SysUser = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    answers = [a.model_dump() for a in body.answers]
+    data = await submit_assessment(db, user.id, body.questionnaire_id, answers)
+    return success(data)
diff --git a/model/fin_customer_profile.py b/model/fin_customer_profile.py
new file mode 100644
index 0000000..a9e8389
--- /dev/null
+++ b/model/fin_customer_profile.py
@@ -0,0 +1,32 @@
+"""fin_customer_profile 客户画像主表 ORM 模型。"""
+from __future__ import annotations
+
+from datetime import datetime
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import BigInteger, DateTime, Integer, JSON, Numeric, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from model.base import Base
+
+
+class FinCustomerProfile(Base):
+    __tablename__ = "fin_customer_profile"
+    __table_args__ = {"comment": "客户画像主表（RAG/GraphRAG 个性化推荐核心上下文）"}
+
+    # 主键为 customer_id（一对一关联 sys_user.id），非自增 id
+    customer_id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    risk_level: Mapped[str | None] = mapped_column(String(16))
+    risk_score: Mapped[int | None] = mapped_column(Integer)
+    investment_experience: Mapped[str | None] = mapped_column(String(16))
+    annual_income_range: Mapped[str | None] = mapped_column(String(32))
+    total_assets: Mapped[Decimal | None] = mapped_column(Numeric(18, 2))
+    asset_allocation: Mapped[dict[str, Any] | None] = mapped_column(JSON)
+    product_preference: Mapped[dict[str, Any] | None] = mapped_column(JSON)
+    customer_level: Mapped[str | None] = mapped_column(String(16))
+    confidence_score: Mapped[Decimal] = mapped_column(Numeric(5, 2), server_default="0.50")
+    profile_version: Mapped[int] = mapped_column(Integer, server_default="1")
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime, server_default=func.now(), onupdate=func.now()
+    )
diff --git a/model/fin_product.py b/model/fin_product.py
new file mode 100644
index 0000000..f324ba9
--- /dev/null
+++ b/model/fin_product.py
@@ -0,0 +1,32 @@
+"""fin_product 公募基金产品表 ORM 模型。"""
+from __future__ import annotations
+
+from datetime import date, datetime
+from decimal import Decimal
+
+from sqlalchemy import BigInteger, Date, DateTime, Integer, Numeric, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from model.base import Base
+
+
+class FinProduct(Base):
+    __tablename__ = "fin_product"
+    __table_args__ = {"comment": "公募基金产品表"}
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    product_code: Mapped[str] = mapped_column(String(32), unique=True)
+    product_name: Mapped[str] = mapped_column(String(128))
+    product_type: Mapped[str] = mapped_column(String(32))
+    risk_level: Mapped[str] = mapped_column(String(8))
+    expected_return: Mapped[Decimal | None] = mapped_column(Numeric(7, 4))
+    nav: Mapped[Decimal | None] = mapped_column(Numeric(12, 6))
+    nav_date: Mapped[date | None] = mapped_column(Date)
+    fee_rate: Mapped[Decimal] = mapped_column(Numeric(6, 4), server_default="0.0000")
+    term_days: Mapped[int] = mapped_column(Integer, server_default="0")
+    fund_manager: Mapped[str | None] = mapped_column(String(64))
+    status: Mapped[str] = mapped_column(String(16), server_default="在售")
+    create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime, server_default=func.now(), onupdate=func.now()
+    )
diff --git a/model/fin_risk_assessment.py b/model/fin_risk_assessment.py
new file mode 100644
index 0000000..f9995ba
--- /dev/null
+++ b/model/fin_risk_assessment.py
@@ -0,0 +1,26 @@
+"""fin_risk_assessment 风险评估记录表 ORM 模型。"""
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Any
+
+from sqlalchemy import BigInteger, Date, DateTime, Integer, JSON, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from model.base import Base
+
+
+class FinRiskAssessment(Base):
+    __tablename__ = "fin_risk_assessment"
+    __table_args__ = {"comment": "风险评估记录表（购买前问卷确认客户等级）"}
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    customer_id: Mapped[int] = mapped_column(BigInteger)
+    assessment_date: Mapped[date] = mapped_column(Date)
+    question_version: Mapped[str | None] = mapped_column(String(16))
+    total_score: Mapped[int] = mapped_column(Integer)
+    risk_level: Mapped[str] = mapped_column(String(16))
+    answers: Mapped[list[Any] | None] = mapped_column(JSON)
+    assessor_type: Mapped[str] = mapped_column(String(16), server_default="AI评估")
+    valid_until: Mapped[date] = mapped_column(Date)
+    create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
diff --git a/model/fund_nav_history.py b/model/fund_nav_history.py
new file mode 100644
index 0000000..f05dcbc
--- /dev/null
+++ b/model/fund_nav_history.py
@@ -0,0 +1,23 @@
+"""fund_nav_history 基金净值历史表 ORM 模型。"""
+from __future__ import annotations
+
+from datetime import date, datetime
+from decimal import Decimal
+
+from sqlalchemy import BigInteger, Date, DateTime, Numeric, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from model.base import Base
+
+
+class FundNavHistory(Base):
+    __tablename__ = "fund_nav_history"
+    __table_args__ = {"comment": "基金净值历史（业绩指标/走势图数据源）"}
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    product_id: Mapped[int] = mapped_column(BigInteger)
+    nav_date: Mapped[date] = mapped_column(Date)
+    unit_nav: Mapped[Decimal] = mapped_column(Numeric(12, 6))
+    accumulated_nav: Mapped[Decimal | None] = mapped_column(Numeric(12, 6))
+    daily_growth: Mapped[Decimal | None] = mapped_column(Numeric(8, 4))
+    create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
diff --git a/model/ops_question.py b/model/ops_question.py
new file mode 100644
index 0000000..d7cd249
--- /dev/null
+++ b/model/ops_question.py
@@ -0,0 +1,22 @@
+"""ops_question 问卷题目表 ORM 模型。"""
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import BigInteger, Integer, JSON, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from model.base import Base
+
+
+class OpsQuestion(Base):
+    __tablename__ = "ops_question"
+    __table_args__ = {"comment": "问卷题目表（题目/选项/分值可运营配置）"}
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    questionnaire_id: Mapped[int] = mapped_column(BigInteger)
+    question_no: Mapped[int] = mapped_column(Integer)
+    content: Mapped[str] = mapped_column(String(512))
+    option_json: Mapped[list[Any] | None] = mapped_column(JSON)
+    score_json: Mapped[dict[str, Any] | None] = mapped_column(JSON)
+    sort: Mapped[int] = mapped_column(Integer, server_default="0")
diff --git a/model/ops_questionnaire.py b/model/ops_questionnaire.py
new file mode 100644
index 0000000..05bc560
--- /dev/null
+++ b/model/ops_questionnaire.py
@@ -0,0 +1,24 @@
+"""ops_questionnaire 问卷模板表 ORM 模型。"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from model.base import Base
+
+
+class OpsQuestionnaire(Base):
+    __tablename__ = "ops_questionnaire"
+    __table_args__ = {"comment": "问卷模板表"}
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    name: Mapped[str] = mapped_column(String(64))
+    q_type: Mapped[str] = mapped_column(String(16))
+    version: Mapped[str] = mapped_column(String(16), server_default="v1")
+    status: Mapped[str] = mapped_column(String(16), server_default="启用")
+    create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime, server_default=func.now(), onupdate=func.now()
+    )
diff --git a/repositories/fund_nav.py b/repositories/fund_nav.py
new file mode 100644
index 0000000..6d58e8f
--- /dev/null
+++ b/repositories/fund_nav.py
@@ -0,0 +1,32 @@
+"""fund_nav_history 净值历史仓储：按产品代码与日期区间查询走势。
+
+数据现状说明：fund_nav_history.product_id 列在 Mock 数据灌入时存的是基金代码
+（如 110001）而非 fin_product 主键 id。为不改数据库、只通过代码适配，这里用
+CAST 把 BIGINT 列按字符串与 product_code 比对。
+"""
+from __future__ import annotations
+
+from datetime import date
+
+from sqlalchemy import String, cast, select
+
+from model.fund_nav_history import FundNavHistory
+from repositories.base import BaseRepository
+
+
+class FundNavRepo(BaseRepository):
+    model = FundNavHistory
+
+    async def list_series(
+        self, product_code: str, start_date: date, end_date: date
+    ) -> list[FundNavHistory]:
+        stmt = (
+            select(FundNavHistory)
+            .where(
+                cast(FundNavHistory.product_id, String) == product_code,
+                FundNavHistory.nav_date >= start_date,
+                FundNavHistory.nav_date <= end_date,
+            )
+            .order_by(FundNavHistory.nav_date.asc())
+        )
+        return list((await self.db.scalars(stmt)).all())
diff --git a/repositories/product.py b/repositories/product.py
new file mode 100644
index 0000000..2eddc99
--- /dev/null
+++ b/repositories/product.py
@@ -0,0 +1,71 @@
+"""fin_product 产品仓储：列表搜索（关键字/类型/风险/状态筛选 + 排序 + 分页）。"""
+from __future__ import annotations
+
+from sqlalchemy import or_, select
+
+from model.fin_product import FinProduct
+from repositories.base import BaseRepository
+from utils.exceptions import ParamError
+
+# 排序字段白名单：防止 sort_by 注入，字符串映射到 ORM 列（不拼 SQL）
+SORTABLE_FIELDS = {
+    "expected_return": FinProduct.expected_return,
+    "nav": FinProduct.nav,
+    "fee_rate": FinProduct.fee_rate,
+    "risk_level": FinProduct.risk_level,
+    "create_time": FinProduct.create_time,
+}
+
+
+class ProductRepo(BaseRepository):
+    model = FinProduct
+
+    async def get_by_code(self, product_code: str) -> FinProduct | None:
+        return await self.db.scalar(
+            select(FinProduct).where(FinProduct.product_code == product_code)
+        )
+
+    async def search(
+        self,
+        *,
+        keyword: str | None = None,
+        product_type: str | None = None,
+        risk_level: str | None = None,
+        status: str | None = "在售",
+        sort_by: str = "create_time",
+        sort_order: str = "desc",
+        limit: int = 10,
+        offset: int = 0,
+    ) -> tuple[list[FinProduct], int]:
+        conds = []
+        if status and status != "全部":
+            conds.append(FinProduct.status == status)
+        if product_type:
+            conds.append(FinProduct.product_type == product_type)
+        if risk_level:
+            conds.append(FinProduct.risk_level == risk_level)
+        if keyword:
+            like = f"%{keyword}%"
+            conds.append(
+                or_(
+                    FinProduct.product_code.like(like),
+                    FinProduct.product_name.like(like),
+                    FinProduct.fund_manager.like(like),
+                )
+            )
+
+        col = SORTABLE_FIELDS.get(sort_by)
+        if col is None:
+            raise ParamError(f"不支持的排序字段: {sort_by}")
+        order_col = col.desc() if sort_order == "desc" else col.asc()
+
+        stmt = (
+            select(FinProduct)
+            .where(*conds)
+            .order_by(order_col)
+            .limit(limit)
+            .offset(offset)
+        )
+        items = list((await self.db.scalars(stmt)).all())
+        total = await self.count(where=conds)
+        return items, total
diff --git a/repositories/questionnaire.py b/repositories/questionnaire.py
new file mode 100644
index 0000000..b459b3f
--- /dev/null
+++ b/repositories/questionnaire.py
@@ -0,0 +1,24 @@
+"""问卷域仓储：问卷模板 + 题目。"""
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from model.ops_question import OpsQuestion
+from model.ops_questionnaire import OpsQuestionnaire
+from repositories.base import BaseRepository
+
+
+class QuestionnaireRepo(BaseRepository):
+    model = OpsQuestionnaire
+
+
+class QuestionRepo(BaseRepository):
+    model = OpsQuestion
+
+    async def list_by_questionnaire(self, questionnaire_id: int) -> list[OpsQuestion]:
+        stmt = (
+            select(OpsQuestion)
+            .where(OpsQuestion.questionnaire_id == questionnaire_id)
+            .order_by(OpsQuestion.question_no.asc(), OpsQuestion.sort.asc())
+        )
+        return list((await self.db.scalars(stmt)).all())
diff --git a/repositories/risk_assessment.py b/repositories/risk_assessment.py
new file mode 100644
index 0000000..20b0ced
--- /dev/null
+++ b/repositories/risk_assessment.py
@@ -0,0 +1,46 @@
+"""风评域仓储：风评记录 + 客户画像（画像主键为 customer_id）。"""
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from model.fin_customer_profile import FinCustomerProfile
+from model.fin_risk_assessment import FinRiskAssessment
+from repositories.base import BaseRepository
+
+
+class RiskAssessmentRepo(BaseRepository):
+    model = FinRiskAssessment
+
+
+class CustomerProfileRepo(BaseRepository):
+    """客户画像仓储。注意：主键是 customer_id 而非 id，不适用基类 get(pk)。"""
+
+    model = FinCustomerProfile
+
+    async def get_by_customer(self, customer_id: int) -> FinCustomerProfile | None:
+        return await self.db.scalar(
+            select(FinCustomerProfile).where(
+                FinCustomerProfile.customer_id == customer_id
+            )
+        )
+
+    async def upsert_risk(
+        self, customer_id: int, risk_level: str, risk_score: int
+    ) -> FinCustomerProfile:
+        """回写风险等级与评分：无画像则初始化，有则更新并递增画像版本号。"""
+        profile = await self.get_by_customer(customer_id)
+        if profile is None:
+            profile = FinCustomerProfile(
+                customer_id=customer_id,
+                risk_level=risk_level,
+                risk_score=risk_score,
+                profile_version=1,
+            )
+            self.db.add(profile)
+        else:
+            profile.risk_level = risk_level
+            profile.risk_score = risk_score
+            profile.profile_version = (profile.profile_version or 0) + 1
+        await self.db.commit()
+        await self.db.refresh(profile)
+        return profile
diff --git a/schemas/questionnaire.py b/schemas/questionnaire.py
new file mode 100644
index 0000000..7b0da7d
--- /dev/null
+++ b/schemas/questionnaire.py
@@ -0,0 +1,12 @@
+"""问卷提交相关 DTO。"""
+from pydantic import BaseModel, Field
+
+
+class AnswerItem(BaseModel):
+    question_no: int = Field(gt=0)
+    option: str = Field(min_length=1, max_length=8)
+
+
+class RiskAssessmentSubmitReq(BaseModel):
+    questionnaire_id: int = Field(gt=0)
+    answers: list[AnswerItem] = Field(min_length=1)
diff --git a/service/product.py b/service/product.py
new file mode 100644
index 0000000..0bb05b4
--- /dev/null
+++ b/service/product.py
@@ -0,0 +1,155 @@
+"""产品服务：列表（分页/筛选/排序）+ 历史业绩（净值/收益走势）。"""
+from __future__ import annotations
+
+from datetime import date, timedelta
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from model.fin_product import FinProduct
+from repositories.fund_nav import FundNavRepo
+from repositories.product import ProductRepo
+from utils.exceptions import NotFoundError, ParamError
+
+
+def _to_item(p: FinProduct) -> dict:
+    """ORM → 响应 dict（Decimal 显式转 float，避免序列化歧义）。"""
+    return {
+        "id": p.id,
+        "product_code": p.product_code,
+        "product_name": p.product_name,
+        "product_type": p.product_type,
+        "risk_level": p.risk_level,
+        "expected_return": float(p.expected_return) if p.expected_return is not None else None,
+        "nav": float(p.nav) if p.nav is not None else None,
+        "nav_date": p.nav_date.isoformat() if p.nav_date else None,
+        "fee_rate": float(p.fee_rate),
+        "term_days": p.term_days,
+        "fund_manager": p.fund_manager,
+        "status": p.status,
+        # 不改表：交易方式标识由现有字段推导（status=在售 可购买；term_days=0 可定投）
+        "support_purchase": p.status == "在售",
+        "support_dca": p.term_days == 0,
+    }
+
+
+async def list_products(
+    db: AsyncSession,
+    *,
+    page: int = 1,
+    page_size: int = 10,
+    keyword: str | None = None,
+    product_type: str | None = None,
+    risk_level: str | None = None,
+    status: str = "在售",
+    sort_by: str = "create_time",
+    sort_order: str = "desc",
+) -> dict:
+    if page < 1:
+        raise ParamError("page 必须 >= 1")
+    if page_size < 1 or page_size > 100:
+        raise ParamError("page_size 必须在 1-100 之间")
+    if sort_order not in ("asc", "desc"):
+        raise ParamError("sort_order 仅支持 asc/desc")
+
+    items, total = await ProductRepo(db).search(
+        keyword=keyword,
+        product_type=product_type,
+        risk_level=risk_level,
+        status=status,
+        sort_by=sort_by,
+        sort_order=sort_order,
+        limit=page_size,
+        offset=(page - 1) * page_size,
+    )
+    return {
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+        "items": [_to_item(p) for p in items],
+    }
+
+
+def _calc_metrics(rows: list) -> dict:
+    """基于净值序列计算区间汇总指标（百分比）。
+
+    - total_growth_rate：总增长率 = 每天日增长率累加
+    - max_return_rate：区间最大收益率 = 从区间内最低点买入到其后最高点的最大涨幅
+    - max_drawdown_rate：区间最大回撤率 = 从区间内最高点回落到其后最低点的最大跌幅
+    """
+    total_growth = sum(
+        float(r.daily_growth) for r in rows if r.daily_growth is not None
+    )
+
+    peak = trough = float(rows[0].unit_nav)
+    max_return = max_drawdown = 0.0
+    for r in rows:
+        nav = float(r.unit_nav)
+        peak = max(peak, nav)
+        trough = min(trough, nav)
+        max_drawdown = max(max_drawdown, (peak - nav) / peak if peak else 0.0)
+        max_return = max(max_return, (nav - trough) / trough if trough else 0.0)
+
+    return {
+        "total_growth_rate": round(total_growth, 4),
+        "max_return_rate": round(max_return * 100, 4),
+        "max_drawdown_rate": round(max_drawdown * 100, 4),
+    }
+
+
+async def get_history(
+    db: AsyncSession,
+    product_code: str,
+    start_date: date | None = None,
+    end_date: date | None = None,
+    series: str = "nav",
+) -> dict:
+    if series not in ("nav", "return"):
+        raise ParamError("series 仅支持 nav/return")
+    if await ProductRepo(db).get_by_code(product_code) is None:
+        raise NotFoundError("基金产品不存在")
+
+    end = end_date or date.today()
+    start = start_date or (end - timedelta(days=90))
+    if start > end:
+        raise ParamError("start_date 不能晚于 end_date")
+
+    rows = await FundNavRepo(db).list_series(product_code, start, end)
+    if not rows:
+        return {
+            "product_code": product_code,
+            "series": series,
+            "total_growth_rate": 0.0,
+            "max_return_rate": 0.0,
+            "max_drawdown_rate": 0.0,
+            "items": [],
+        }
+
+    if series == "nav":
+        items = [
+            {
+                "nav_date": r.nav_date.isoformat(),
+                "unit_nav": float(r.unit_nav),
+                "accumulated_nav": float(r.accumulated_nav) if r.accumulated_nav is not None else None,
+                "daily_growth": float(r.daily_growth) if r.daily_growth is not None else None,
+            }
+            for r in rows
+        ]
+    else:
+        base = float(rows[0].unit_nav)  # 区间首日为基准，累计收益实时从净值计算
+        items = [
+            {
+                "nav_date": r.nav_date.isoformat(),
+                "unit_nav": float(r.unit_nav),
+                "cum_return_pct": round((float(r.unit_nav) / base - 1) * 100, 4),
+            }
+            for r in rows
+        ]
+    metrics = _calc_metrics(rows)
+    return {
+        "product_code": product_code,
+        "series": series,
+        "total_growth_rate": metrics["total_growth_rate"],
+        "max_return_rate": metrics["max_return_rate"],
+        "max_drawdown_rate": metrics["max_drawdown_rate"],
+        "items": items,
+    }
diff --git a/service/questionnaire.py b/service/questionnaire.py
new file mode 100644
index 0000000..e1ec9e5
--- /dev/null
+++ b/service/questionnaire.py
@@ -0,0 +1,85 @@
+"""问卷服务：提交风险测评 → 完整性/选项校验 → 算分定级 → 落风评记录 → 回写画像。"""
+from __future__ import annotations
+
+from datetime import date, timedelta
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from model.fin_risk_assessment import FinRiskAssessment
+from repositories.questionnaire import QuestionRepo, QuestionnaireRepo
+from repositories.risk_assessment import CustomerProfileRepo, RiskAssessmentRepo
+from utils.exceptions import NotFoundError, ParamError
+
+# 分数 → 风险等级映射（总分 0-100，20 分一档；后续可迁移到 sys_config 运营化）
+_SCORE_LEVELS = (
+    (80, "激进"),
+    (60, "进取"),
+    (40, "平衡"),
+    (20, "稳健"),
+    (0, "保守"),
+)
+
+
+def _score_to_level(score: int) -> str:
+    for threshold, level in _SCORE_LEVELS:
+        if score >= threshold:
+            return level
+    return "保守"
+
+
+async def submit_assessment(
+    db: AsyncSession,
+    customer_id: int,
+    questionnaire_id: int,
+    answers: list[dict],
+) -> dict:
+    questionnaire = await QuestionnaireRepo(db).get(questionnaire_id)
+    if questionnaire is None:
+        raise NotFoundError("问卷不存在")
+    if questionnaire.status != "启用":
+        raise ParamError("问卷未启用")
+
+    questions = await QuestionRepo(db).list_by_questionnaire(questionnaire_id)
+    if not questions:
+        raise ParamError("问卷尚未配置题目")
+
+    # 提交答案按 question_no（题号）建索引，校验必填完整性与选项合法性
+    by_no = {a["question_no"]: a["option"] for a in answers}
+    total_score = 0
+    detail = []
+    for q in questions:
+        option = by_no.get(q.question_no)
+        if option is None:
+            raise ParamError(f"第{q.question_no}题未作答")
+        scores = q.score_json or {}
+        if option not in scores:
+            raise ParamError(f"第{q.question_no}题选项非法: {option}")
+        score = int(scores[option])
+        total_score += score
+        detail.append({"q": q.question_no, "a": option, "score": score})
+
+    risk_level = _score_to_level(total_score)
+    today = date.today()
+    valid_until = today + timedelta(days=365)  # 监管要求一年一评
+
+    record = FinRiskAssessment(
+        customer_id=customer_id,
+        assessment_date=today,
+        question_version=questionnaire.version,
+        total_score=total_score,
+        risk_level=risk_level,
+        answers=detail,
+        assessor_type="AI评估",
+        valid_until=valid_until,
+    )
+    record = await RiskAssessmentRepo(db).add(record)
+
+    # 回写画像：无画像初始化，有画像更新风险等级/评分并递增版本号
+    await CustomerProfileRepo(db).upsert_risk(customer_id, risk_level, total_score)
+
+    return {
+        "assessment_id": record.id,
+        "total_score": total_score,
+        "risk_level": risk_level,
+        "valid_until": valid_until.isoformat(),
+    }