feat:新增基金产品列表接口,历史业绩接口,客户注册提交接口,问卷提交接口

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())

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

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())

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