feat: add A-card operations frontend and backend foundation

backend/Dockerfile

@@ -14,6 +14,8 @@ COPY common ./common
 COPY backend ./backend
 COPY alembic.ini ./
 COPY migrations ./migrations
+COPY operations-alembic.ini ./
+COPY operations_migrations ./operations_migrations
 RUN uv sync --frozen --no-dev --package backend
 
 

backend/README.md

@@ -5,4 +5,41 @@
 模块合并，外部 REST 契约保持不变。
 
 底层走的是 `common.storage.AsyncStorageBackend` 抽象，按
-`settings.storage_backend` 切换 s3 / local 两种实现。
+`settings.storage_backend` 切换 s3 / local 两种实现。
+
+## A 卡模型运维模块
+
+运维能力作为同一 FastAPI 进程内的独立 bounded context 接入：
+
+- 平台库继续使用 `DATABASE_URL`，负责登录、角色和工作空间校验。
+- 运维库使用 `OPERATIONS_DATABASE_URL`；未配置时自动复用平台库连接信息，
+  仅将库名切换为 `model_operations`。
+- `PLATFORM_READ_DATABASE_URL` 与 `DEPLOY_DATABASE_URL` 使用独立只读
+  Session；MySQL 事务会显式执行 `SET TRANSACTION READ ONLY`。
+- 两套 SQLAlchemy engine/session factory 完全分离，运维路由只读写 `ops_*`。
+- 当前已落地 P0 ORM，以及模型列表、模型详情、指定月份监控结果三条接口。
+- `OPERATIONS_DATA_MODE=mock` 时仍走真实鉴权、Workspace校验和API响应，
+  仅将业务查询替换为后端Mock数据；切换为 `database` 后读取运维库。
+- Redis作为可选加速层接入：缓存聚合查询并承载通知 Streams；MySQL
+  Outbox仍是可靠事实源，Redis不可用时主接口继续运行。
+
+本地启动：
+
+```bash
+uv run --package backend --env-file .env uvicorn backend.main:app \
+  --host 0.0.0.0 --port 8010 --reload
+```
+
+运维模块探针：
+
+```text
+GET /api/v1/operations/health
+```
+
+真实业务接口均需登录 Cookie 和 `workspace_id` 查询参数：
+
+```text
+GET /api/v1/operations/models?workspace_id=<ULID>
+GET /api/v1/operations/models/{model_id}?workspace_id=<ULID>
+GET /api/v1/operations/models/{model_id}/monitor-results?month=YYYY-MM&workspace_id=<ULID>
+```

backend/pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "bcrypt>=4.0,<4.1",
     "loguru>=0.7.2",
     "aiofiles>=25.1.0",
+    "redis[hiredis]>=6.2,<7",
 ]
 
 [tool.uv.sources]

backend/src/backend/api/operations/__init__.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from backend.api.operations.health import router as health_router
+from backend.api.operations.models import router as models_router
+
+router = APIRouter(prefix="/api/v1/operations", tags=["operations"])
+router.include_router(health_router)
+router.include_router(models_router)
+
+__all__ = ["router"]

backend/src/backend/api/operations/_deps.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from typing import Literal
+
+from common.config import settings
+from common.db import readonly_session_scope, session_scope
+from common.db.models import Roles
+from fastapi import Depends, HTTPException, Request, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.api.dependencies import RequestContext, request_context
+from backend.services.operations import OperationsCache, OperationsEventStream
+
+OperationsRole = Literal["admin", "model_team", "business_team"]
+
+_ROLE_ALIASES: dict[str, OperationsRole] = {
+    "admin": "admin",
+    "administrator": "admin",
+    "developer": "model_team",
+    "model_team": "model_team",
+    "model": "model_team",
+    "business": "business_team",
+    "business_team": "business_team",
+    "biz": "business_team",
+}
+
+
+@dataclass(frozen=True)
+class OperationsContext:
+    request_id: str
+    user_id: str
+    workspace_id: str
+    role: OperationsRole
+
+    @property
+    def is_admin(self) -> bool:
+        return self.role == "admin"
+
+
+async def operations_database_session(
+    request: Request,
+) -> AsyncIterator[AsyncSession]:
+    factory = getattr(request.app.state, "operations_session_factory", None)
+    if factory is None:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            {"code": "OPERATIONS_DATABASE_UNAVAILABLE", "message": "运维数据库未初始化"},
+        )
+    async with session_scope(factory) as session:
+        yield session
+
+
+async def operations_platform_database_session(
+    request: Request,
+) -> AsyncIterator[AsyncSession]:
+    factory = getattr(
+        request.app.state,
+        "operations_platform_read_session_factory",
+        None,
+    )
+    if factory is None:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            {"code": "PLATFORM_DATABASE_UNAVAILABLE", "message": "平台只读库未初始化"},
+        )
+    async with readonly_session_scope(factory) as session:
+        yield session
+
+
+async def deploy_database_session(
+    request: Request,
+) -> AsyncIterator[AsyncSession]:
+    factory = getattr(request.app.state, "deploy_read_session_factory", None)
+    if factory is None:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            {"code": "DEPLOY_DATABASE_UNAVAILABLE", "message": "部署只读库未初始化"},
+        )
+    async with readonly_session_scope(factory) as session:
+        yield session
+
+
+def operations_cache(request: Request) -> OperationsCache:
+    return OperationsCache(
+        getattr(request.app.state, "operations_redis", None),
+        prefix=settings.operations_redis_prefix,
+        default_ttl_seconds=settings.operations_cache_ttl_seconds,
+    )
+
+
+def operations_event_stream(request: Request) -> OperationsEventStream:
+    return OperationsEventStream(
+        getattr(request.app.state, "operations_redis", None),
+        stream_name=settings.operations_event_stream,
+        maxlen=settings.operations_event_stream_maxlen,
+    )
+
+
+async def operations_context(
+    context: RequestContext = Depends(request_context),
+    platform_session: AsyncSession = Depends(operations_platform_database_session),
+) -> OperationsContext:
+    """Reuse login/workspace checks and normalize the three operations roles."""
+    raw_role = context.role.role_code
+    if context.is_system_admin:
+        raw_role = "admin"
+    elif context.user.platform_role_id:
+        platform_role = await platform_session.get(Roles, context.user.platform_role_id)
+        if platform_role is not None:
+            raw_role = platform_role.role_code
+
+    normalized = _ROLE_ALIASES.get(raw_role)
+    if normalized is None:
+        raise HTTPException(
+            status.HTTP_403_FORBIDDEN,
+            {
+                "code": "OPERATIONS_ROLE_UNSUPPORTED",
+                "message": f"当前登录角色不支持运维模块：{raw_role}",
+            },
+        )
+    return OperationsContext(
+        request_id=context.request_id,
+        user_id=context.user.user_id,
+        workspace_id=context.workspace.workspace_id,
+        role=normalized,
+    )
+
+
+__all__ = [
+    "OperationsContext",
+    "deploy_database_session",
+    "operations_cache",
+    "operations_context",
+    "operations_database_session",
+    "operations_event_stream",
+    "operations_platform_database_session",
+]

backend/src/backend/api/operations/health.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Request, Response, status
+from sqlalchemy import text
+
+from common.db import readonly_session_scope, session_scope
+
+router = APIRouter()
+
+
+@router.get("/health")
+async def operations_health(
+    request: Request,
+    response: Response,
+) -> dict[str, Any]:
+    checks: dict[str, dict[str, str]] = {}
+    targets = (
+        ("model_operations", "operations_session_factory", False),
+        ("model_platform", "operations_platform_read_session_factory", True),
+        ("model_deploy", "deploy_read_session_factory", True),
+    )
+    for database_name, factory_name, read_only in targets:
+        factory = getattr(request.app.state, factory_name, None)
+        if factory is None:
+            checks[database_name] = {"status": "error", "mode": "unavailable"}
+            continue
+        scope = readonly_session_scope if read_only else session_scope
+        try:
+            async with scope(factory) as session:
+                current_database = await session.scalar(text("SELECT DATABASE()"))
+            checks[database_name] = {
+                "status": "ok" if current_database == database_name else "error",
+                "mode": "read_only" if read_only else "read_write",
+            }
+        except Exception:
+            checks[database_name] = {
+                "status": "error",
+                "mode": "read_only" if read_only else "read_write",
+            }
+    ready = all(item["status"] == "ok" for item in checks.values())
+    if not ready:
+        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
+    redis_client = getattr(request.app.state, "operations_redis", None)
+    if redis_client is None:
+        redis_check = {"status": "disabled", "mode": "cache_and_streams"}
+    else:
+        try:
+            pong = await redis_client.ping()
+            redis_check = {
+                "status": "ok" if pong else "error",
+                "mode": "cache_and_streams",
+            }
+        except Exception:
+            redis_check = {"status": "error", "mode": "cache_and_streams"}
+    module_status = "ok" if ready and redis_check["status"] == "ok" else (
+        "degraded" if ready else "not_ready"
+    )
+    return {
+        "data": {
+            "status": module_status,
+            "module": "model_operations",
+            "databases": checks,
+            "redis": redis_check,
+        },
+        "meta": {},
+    }
+
+
+__all__ = ["router"]

backend/src/backend/api/operations/models.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.api.operations._deps import (
+    OperationsContext,
+    operations_context,
+    operations_database_session,
+)
+from backend.schemas.operations import ModelCategoryCode, ModelStatusLabel
+from backend.services.operations import (
+    get_model,
+    get_monthly_monitoring_result,
+    list_models,
+    parse_monitor_month,
+)
+from backend.services.operations.mock_data import (
+    get_mock_model,
+    get_mock_monthly_result,
+    list_mock_models,
+)
+from common.config import settings
+
+router = APIRouter()
+
+
+def _envelope(context: OperationsContext, data: Any) -> dict[str, Any]:
+    return {"request_id": context.request_id, "data": data, "meta": {}}
+
+
+@router.get("/models")
+async def list_operation_models(
+    bank: str | None = Query(default=None, max_length=150),
+    category: ModelCategoryCode | None = Query(default=None),
+    model_status: ModelStatusLabel | None = Query(default=None, alias="status"),
+    keyword: str | None = Query(default=None, max_length=200),
+    context: OperationsContext = Depends(operations_context),
+    session: AsyncSession = Depends(operations_database_session),
+) -> dict[str, Any]:
+    models = list_mock_models(
+        bank=bank,
+        category=category,
+        status=model_status,
+        keyword=keyword,
+    ) if settings.operations_data_mode == "mock" else await list_models(
+        session,
+        context.workspace_id,
+        bank=bank,
+        category=category,
+        status=model_status,
+        keyword=keyword,
+    )
+    return _envelope(context, models)
+
+
+@router.get("/models/{model_id}")
+async def get_operation_model(
+    model_id: str,
+    context: OperationsContext = Depends(operations_context),
+    session: AsyncSession = Depends(operations_database_session),
+) -> dict[str, Any]:
+    model = get_mock_model(model_id) if settings.operations_data_mode == "mock" else await get_model(session, context.workspace_id, model_id)
+    if model is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            {"code": "MODEL_NOT_FOUND", "message": "模型不存在"},
+        )
+    return _envelope(context, model)
+
+
+@router.get("/models/{model_id}/monitor-results")
+async def get_operation_model_monitor_result(
+    model_id: str,
+    month: str = Query(..., pattern=r"^\d{4}-\d{2}$"),
+    context: OperationsContext = Depends(operations_context),
+    session: AsyncSession = Depends(operations_database_session),
+) -> dict[str, Any]:
+    try:
+        monitor_month = parse_monitor_month(month)
+    except ValueError as exc:
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            {"code": "INVALID_MONITOR_MONTH", "message": str(exc)},
+        ) from exc
+    result = get_mock_monthly_result(model_id, month) if settings.operations_data_mode == "mock" else await get_monthly_monitoring_result(
+        session,
+        context.workspace_id,
+        model_id,
+        monitor_month,
+    )
+    if result is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            {
+                "code": "MONITOR_RESULT_NOT_FOUND",
+                "message": "该月份暂无监控结果",
+            },
+        )
+    return _envelope(context, result)
+
+
+__all__ = ["router"]

backend/src/backend/main.py

@@ -32,10 +32,12 @@ from common.storage import (
 from fastapi import Request
 from fastapi.responses import JSONResponse
 from loguru import logger
+from redis.asyncio import Redis
 
 from backend.api.admin import router as admin_router
 from backend.api.auth import router as auth_router
 from backend.api.jupyter import router as jupyter_router
+from backend.api.operations import router as operations_router
 from backend.api.platform import router as platform_router
 from backend.api.resources import router as resources_router
 from backend.api.schedules.runs import router as schedule_runs_router
@@ -65,6 +67,35 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
     # 取得它们；这样每个请求不会重复创建数据库连接或 HTTP 客户端。
     engine = create_database_engine(settings.database_url)
     app.state.session_factory = create_session_factory(engine)
+    operations_database_url = settings.operations_database_url
+    if not operations_database_url:
+        raise RuntimeError("OPERATIONS_DATABASE_URL could not be resolved")
+    operations_engine = create_database_engine(operations_database_url)
+    app.state.operations_session_factory = create_session_factory(operations_engine)
+    platform_read_database_url = settings.platform_read_database_url
+    deploy_database_url = settings.deploy_database_url
+    if not platform_read_database_url or not deploy_database_url:
+        raise RuntimeError("operations read-only database URLs could not be resolved")
+    platform_read_engine = create_database_engine(platform_read_database_url)
+    deploy_read_engine = create_database_engine(deploy_database_url)
+    app.state.operations_platform_read_session_factory = create_session_factory(
+        platform_read_engine
+    )
+    app.state.deploy_read_session_factory = create_session_factory(deploy_read_engine)
+
+    # Redis only accelerates operations reads and event delivery. It is kept
+    # optional so a cache outage never prevents MySQL-backed APIs from booting.
+    operations_redis: Redis | None = None
+    if settings.redis_url:
+        operations_redis = Redis.from_url(
+            settings.redis_url,
+            decode_responses=True,
+            socket_connect_timeout=settings.redis_socket_connect_timeout_seconds,
+            socket_timeout=settings.redis_socket_timeout_seconds,
+            health_check_interval=30,
+            retry_on_timeout=True,
+        )
+    app.state.operations_redis = operations_redis
 
     # 存储接口与业务路由运行在同一个 backend 进程中，因此直接复用存储对象，
     # 不需要再通过 HTTP 调用自己。字典键使用真实桶名，便于上传会话和存储
@@ -94,8 +125,13 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
     try:
         yield
     finally:
+        if operations_redis is not None:
+            await operations_redis.aclose()
         await rclone_http_client.aclose()
         await runtime_http_client.aclose()
+        await deploy_read_engine.dispose()
+        await platform_read_engine.dispose()
+        await operations_engine.dispose()
         await engine.dispose()
 
 
@@ -112,6 +148,7 @@ app.include_router(schedules_router)
 app.include_router(scripts_router)
 app.include_router(admin_router)
 app.include_router(platform_router)
+app.include_router(operations_router)
 
 # 内部存储接口额外加上 /internal 前缀，供后端服务间调用，不作为普通前端 API。
 app.include_router(storage_api_router, prefix="/internal")

backend/src/backend/schemas/operations/__init__.py

@@ -0,0 +1,15 @@
+from backend.schemas.operations.models import (
+    ModelCategoryCode,
+    ModelStatusLabel,
+    OperationsModelDto,
+    RankingStatusLabel,
+)
+from backend.schemas.operations.monitoring import MonthlyMonitoringResultDto
+
+__all__ = [
+    "ModelCategoryCode",
+    "ModelStatusLabel",
+    "MonthlyMonitoringResultDto",
+    "OperationsModelDto",
+    "RankingStatusLabel",
+]

backend/src/backend/schemas/operations/models.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from common.schemas import StrictModel
+
+ModelCategoryCode = Literal["std", "bai", "big", "afd"]
+ModelStatusLabel = Literal["正常", "陪跑", "陪跑结束", "下线"]
+RankingStatusLabel = Literal["相符", "不符", "—"]
+
+
+class OperationsModelDto(StrictModel):
+    model_instance_id: str
+    bank_name: str
+    model_category: ModelCategoryCode
+    model_name: str
+    model_id: str
+    model_version: str
+    model_status: ModelStatusLabel
+    last_iteration_date: str
+    ranking_result: RankingStatusLabel
+    ks: float
+    psi: float
+    ks_mom_drop: float
+    secondary_hits_6m: int
+    is_wuji_bank: bool
+    last_processed_at: str | None
+    previous_advice: str
+    common_model_name: str | None
+
+
+__all__ = [
+    "ModelCategoryCode",
+    "ModelStatusLabel",
+    "OperationsModelDto",
+    "RankingStatusLabel",
+]

backend/src/backend/schemas/operations/monitoring.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from common.schemas import StrictModel
+
+from backend.schemas.operations.models import RankingStatusLabel
+
+
+class MonthlyMonitoringResultDto(StrictModel):
+    model_instance_id: str
+    monitor_month: str
+    ranking_result: RankingStatusLabel
+    ks: float
+    psi: float
+    ks_mom_drop: float
+    secondary_hits_6m: int
+
+
+__all__ = ["MonthlyMonitoringResultDto"]

backend/src/backend/services/operations/__init__.py

@@ -0,0 +1,17 @@
+from backend.services.operations.cache import OperationsCache
+from backend.services.operations.model_queries import (
+    get_model,
+    get_monthly_monitoring_result,
+    list_models,
+    parse_monitor_month,
+)
+from backend.services.operations.streams import OperationsEventStream
+
+__all__ = [
+    "OperationsCache",
+    "OperationsEventStream",
+    "get_model",
+    "get_monthly_monitoring_result",
+    "list_models",
+    "parse_monitor_month",
+]

backend/src/backend/services/operations/cache.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping
+from typing import Any
+
+from loguru import logger
+from redis.asyncio import Redis
+from redis.exceptions import RedisError
+
+
+class OperationsCache:
+    """Best-effort JSON cache; every failure degrades to a cache miss."""
+
+    def __init__(
+        self,
+        client: Redis | None,
+        *,
+        prefix: str,
+        default_ttl_seconds: int,
+    ) -> None:
+        self._client = client
+        self._prefix = prefix.rstrip(":")
+        self._default_ttl_seconds = default_ttl_seconds
+
+    def build_key(
+        self,
+        workspace_id: str,
+        resource: str,
+        params: Mapping[str, Any] | None = None,
+    ) -> str:
+        canonical = json.dumps(
+            dict(params or {}),
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+        )
+        digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20]
+        safe_resource = resource.strip(":").replace(" ", "-")
+        return f"{self._prefix}:cache:{workspace_id}:{safe_resource}:{digest}"
+
+    async def get_json(self, key: str) -> Any | None:
+        if self._client is None:
+            return None
+        try:
+            value = await self._client.get(key)
+            return json.loads(value) if value is not None else None
+        except (RedisError, ValueError, TypeError) as exc:
+            logger.warning("operations cache get failed for {key}: {error}", key=key, error=exc)
+            return None
+
+    async def set_json(
+        self,
+        key: str,
+        value: Any,
+        *,
+        ttl_seconds: int | None = None,
+    ) -> bool:
+        if self._client is None:
+            return False
+        try:
+            payload = json.dumps(
+                value,
+                ensure_ascii=False,
+                separators=(",", ":"),
+                default=str,
+            )
+            await self._client.set(
+                key,
+                payload,
+                ex=ttl_seconds or self._default_ttl_seconds,
+            )
+            return True
+        except (RedisError, TypeError, ValueError) as exc:
+            logger.warning("operations cache set failed for {key}: {error}", key=key, error=exc)
+            return False
+
+    async def delete(self, *keys: str) -> int:
+        if self._client is None or not keys:
+            return 0
+        try:
+            return int(await self._client.unlink(*keys))
+        except RedisError as exc:
+            logger.warning("operations cache delete failed: {error}", error=exc)
+            return 0
+
+    async def invalidate_resource(self, workspace_id: str, resource: str) -> int:
+        if self._client is None:
+            return 0
+        pattern = (
+            f"{self._prefix}:cache:{workspace_id}:"
+            f"{resource.strip(':').replace(' ', '-')}:*"
+        )
+        removed = 0
+        try:
+            batch: list[str] = []
+            async for key in self._client.scan_iter(match=pattern, count=200):
+                batch.append(key)
+                if len(batch) >= 200:
+                    removed += int(await self._client.unlink(*batch))
+                    batch.clear()
+            if batch:
+                removed += int(await self._client.unlink(*batch))
+            return removed
+        except RedisError as exc:
+            logger.warning("operations cache invalidation failed: {error}", error=exc)
+            return removed
+
+
+__all__ = ["OperationsCache"]

backend/src/backend/services/operations/mock_data.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import datetime
+
+from backend.schemas.operations import MonthlyMonitoringResultDto, OperationsModelDto
+
+_MODEL_SPECS = [
+    ("江城银行", "std", "标准A卡", "JC-STD-001", "v2.3", "正常", "2026-06-18", "不符", 36.84, 27.40, 24.10, 3, True, "2026-06-18", "建议启动模型微调或重构评估", "标准A卡通用版 v2.3"),
+    ("滨海银行", "std", "标准A卡", "BH-STD-001", "v2.3", "陪跑", "2026-07-28", "相符", 44.10, 6.20, 3.40, 0, False, "2026-06-03", "维持现状，继续监控", "标准A卡通用版 v2.3"),
+    ("华东银行", "std", "标准A卡", "HD-STD-001", "v2.2", "正常", "2026-03-20", "相符", 43.50, 8.80, 6.10, 0, False, "2026-05-20", "维持现状，继续监控", "标准A卡通用版 v2.3"),
+    ("南岭银行", "std", "标准A卡", "NL-STD-001", "v1.6", "正常", "2025-08-22", "相符", 41.12, 13.60, 22.50, 1, False, "2026-06-12", "建议重点关注 PSI 与 KS 环比变化", "标准A卡通用版 v2.3"),
+    ("云岭银行", "std", "标准A卡", "YL-STD-001", "v2.1", "正常", "2026-01-09", "相符", 45.80, 5.10, 2.20, 0, False, "2026-04-28", "维持现状，继续监控", "标准A卡通用版 v2.3"),
+    ("通汇银行", "std", "标准A卡", "TH-STD-001", "v2.0", "正常", "2025-11-30", "不符", 38.73, 9.40, 11.80, 4, False, "2026-05-16", "建议开展模型微调可行性评估", "标准A卡通用版 v2.3"),
+    ("北岸银行", "std", "标准A卡", "BA-STD-001", "v1.9", "下线", "2024-12-11", "—", 0.00, 0.00, 0.00, 0, False, "2026-03-02", "模型已下线，不再参与月度监控", None),
+    ("江城银行", "big", "大额A卡", "JC-BIG-001", "v1.4", "正常", "2026-07-03", "相符", 46.30, 4.80, 1.50, 0, False, "2026-06-24", "维持现状，继续监控", "大额A卡通用版 v1.4"),
+    ("华东银行", "big", "大额A卡", "HD-BIG-001", "v1.3", "正常", "2026-04-16", "相符", 44.90, 7.30, 4.90, 0, False, "2026-05-30", "维持现状，继续监控", "大额A卡通用版 v1.4"),
+    ("云岭银行", "big", "大额A卡", "YL-BIG-001", "v1.2", "陪跑结束", "2026-08-05", "相符", 42.70, 9.60, 8.20, 0, False, None, "暂无历史处理建议", "大额A卡通用版 v1.4"),
+    ("滨海银行", "big", "大额A卡", "BH-BIG-001", "v1.1", "正常", "2025-09-25", "相符", 39.40, 11.20, 9.70, 2, False, "2026-04-19", "建议持续跟踪，下期复核", "大额A卡通用版 v1.4"),
+    ("南岭银行", "bai", "白户A卡", "NL-BAI-001", "v1.5", "正常", "2026-05-22", "不符", 34.20, 18.70, 15.30, 2, True, "2026-06-09", "建议启动模型微调或重构评估", None),
+    ("通汇银行", "bai", "白户A卡", "TH-BAI-001", "v1.4", "正常", "2026-02-11", "相符", 37.60, 12.10, 7.40, 1, False, "2026-05-11", "建议重点关注并持续跟踪", "白户A卡通用版 v1.4"),
+    ("江城银行", "bai", "白户A卡", "JC-BAI-001", "v1.3", "正常", "2025-12-05", "相符", 40.80, 9.90, 5.60, 0, False, "2026-03-26", "维持现状，继续监控", "白户A卡通用版 v1.4"),
+    ("华东银行", "afd", "反欺诈评分", "HD-AFD-001", "v2.0", "正常", "2026-04-10", "不符", 28.60, 31.50, 18.90, 3, False, "2026-06-21", "建议启动模型微调或重构评估", None),
+    ("滨海银行", "afd", "反欺诈评分", "BH-AFD-001", "v1.8", "正常", "2025-10-30", "相符", 36.40, 8.10, 4.20, 3, True, "2026-05-25", "建议重点关注并持续跟踪", "反欺诈评分通用版 v1.8"),
+    ("云岭银行", "afd", "反欺诈评分", "YL-AFD-001", "v1.7", "正常", "2025-08-19", "相符", 38.20, 14.60, 12.70, 2, False, "2026-04-15", "建议持续跟踪，下期复核", None),
+]
+
+
+def _model(spec: tuple[object, ...]) -> OperationsModelDto:
+    bank, category, name, model_id, version, model_status, iteration_date, ranking, ks, psi, drop, hits, wuji, processed, advice, common = spec
+    return OperationsModelDto(
+        model_instance_id=f"mock-{model_id}",
+        bank_name=str(bank),
+        model_category=category,  # type: ignore[arg-type]
+        model_name=str(name),
+        model_id=str(model_id),
+        model_version=str(version),
+        model_status=model_status,  # type: ignore[arg-type]
+        last_iteration_date=str(iteration_date),
+        ranking_result=ranking,  # type: ignore[arg-type]
+        ks=float(ks),
+        psi=float(psi),
+        ks_mom_drop=float(drop),
+        secondary_hits_6m=int(hits),
+        is_wuji_bank=bool(wuji),
+        last_processed_at=processed,
+        previous_advice=str(advice),
+        common_model_name=common,
+    )
+
+
+MOCK_MODELS = tuple(_model(spec) for spec in _MODEL_SPECS)
+_MODEL_BY_ID = {model.model_id: model for model in MOCK_MODELS}
+
+
+def list_mock_models(
+    *,
+    bank: str | None = None,
+    category: str | None = None,
+    status: str | None = None,
+    keyword: str | None = None,
+) -> list[OperationsModelDto]:
+    term = keyword.strip().lower() if keyword else ""
+    return [
+        model.model_copy()
+        for model in MOCK_MODELS
+        if not bank or model.bank_name == bank
+        if not category or model.model_category == category
+        if not status or model.model_status == status
+        if not term or term in " ".join((model.bank_name, model.model_name, model.model_id, model.model_version)).lower()
+    ]
+
+
+def get_mock_model(model_id: str) -> OperationsModelDto | None:
+    model = _MODEL_BY_ID.get(model_id)
+    return model.model_copy() if model else None
+
+
+def get_mock_monthly_result(model_id: str, month: str) -> MonthlyMonitoringResultDto | None:
+    model = _MODEL_BY_ID.get(model_id)
+    if model is None:
+        return None
+    if model.model_status == "下线" and month != "2026-02":
+        return None
+    try:
+        index = max(0, min(5, (datetime.date.fromisoformat(f"{month}-01").year - 2026) * 12 + datetime.date.fromisoformat(f"{month}-01").month - 2))
+    except ValueError:
+        return None
+    distance = 5 - index
+    return MonthlyMonitoringResultDto(
+        model_instance_id=model.model_instance_id,
+        monitor_month=month,
+        ranking_result=model.ranking_result,
+        ks=round(max(0, model.ks - distance * 0.32), 2),
+        psi=round(max(0, model.psi - distance * 0.74), 2),
+        ks_mom_drop=round(max(0, model.ks_mom_drop - distance * 0.9), 2),
+        secondary_hits_6m=max(0, model.secondary_hits_6m - distance),
+    )
+
+
+__all__ = [
+    "MOCK_MODELS",
+    "get_mock_model",
+    "get_mock_monthly_result",
+    "list_mock_models",
+]

backend/src/backend/services/operations/model_queries.py

@@ -0,0 +1,339 @@
+from __future__ import annotations
+
+import datetime
+from collections.abc import Mapping
+from decimal import Decimal
+from typing import Any
+
+from common.db.models.operations import (
+    OpsBank,
+    OpsModelInstance,
+    OpsModelVersion,
+    OpsMonitorBatch,
+    OpsMonitorEvaluation,
+    OpsMonitorResult,
+    OpsMonitorReview,
+)
+from sqlalchemy import Select, and_, func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.schemas.operations import (
+    MonthlyMonitoringResultDto,
+    OperationsModelDto,
+)
+
+_MODEL_STATUS_TO_API = {
+    "normal": "正常",
+    "escort": "陪跑",
+    "escort_finished": "陪跑结束",
+    "offline": "下线",
+}
+_MODEL_STATUS_TO_DB = {value: key for key, value in _MODEL_STATUS_TO_API.items()}
+_RANKING_TO_API = {
+    "matched": "相符",
+    "unmatched": "不符",
+    "not_applicable": "—",
+    None: "—",
+}
+
+
+def parse_monitor_month(value: str) -> datetime.date:
+    """Parse the API's YYYY-MM month into the DB's first-of-month date."""
+    try:
+        parsed = datetime.datetime.strptime(value, "%Y-%m")
+    except ValueError as exc:
+        raise ValueError("month must use YYYY-MM format") from exc
+    return parsed.date().replace(day=1)
+
+
+def _percent(value: Decimal | float | int | None) -> float:
+    return round(float(value or 0) * 100, 4)
+
+
+def _date_text(value: datetime.date | None) -> str:
+    return value.isoformat() if value else "—"
+
+
+def _datetime_date_text(value: datetime.datetime | None) -> str | None:
+    return value.date().isoformat() if value else None
+
+
+def _latest_result_subquery(
+    workspace_id: str,
+    monitor_month: datetime.date | None = None,
+):
+    filters = [
+        OpsMonitorBatch.workspace_id == workspace_id,
+        OpsMonitorBatch.batch_status == "published",
+        OpsMonitorBatch.is_deleted == 0,
+        OpsMonitorResult.workspace_id == workspace_id,
+        OpsMonitorResult.is_deleted == 0,
+    ]
+    if monitor_month is not None:
+        filters.append(OpsMonitorResult.monitor_month == monitor_month)
+
+    ranked = (
+        select(
+            OpsMonitorResult.monitor_result_id.label("monitor_result_id"),
+            OpsMonitorResult.model_instance_id.label("model_instance_id"),
+            OpsMonitorResult.model_version_id.label("model_version_id"),
+            OpsMonitorResult.monitor_month.label("monitor_month"),
+            OpsMonitorResult.ranking_result.label("ranking_result"),
+            OpsMonitorResult.ks_value.label("ks_value"),
+            OpsMonitorResult.psi_value.label("psi_value"),
+            func.row_number()
+            .over(
+                partition_by=OpsMonitorResult.model_instance_id,
+                order_by=(
+                    OpsMonitorResult.monitor_month.desc(),
+                    OpsMonitorBatch.revision_no.desc(),
+                    OpsMonitorBatch.published_at.desc(),
+                ),
+            )
+            .label("row_no"),
+        )
+        .join(OpsMonitorBatch, OpsMonitorBatch.batch_id == OpsMonitorResult.batch_id)
+        .where(*filters)
+        .subquery("ranked_monitor_results")
+    )
+    return (
+        select(
+            ranked.c.monitor_result_id,
+            ranked.c.model_instance_id,
+            ranked.c.model_version_id,
+            ranked.c.monitor_month,
+            ranked.c.ranking_result,
+            ranked.c.ks_value,
+            ranked.c.psi_value,
+        )
+        .where(ranked.c.row_no == 1)
+        .subquery("latest_monitor_result")
+    )
+
+
+def _latest_review_subquery():
+    return (
+        select(
+            OpsMonitorReview.monitor_result_id.label("monitor_result_id"),
+            func.max(OpsMonitorReview.handled_at).label("handled_at"),
+        )
+        .where(
+            OpsMonitorReview.is_deleted == 0,
+            OpsMonitorReview.handled_at.is_not(None),
+        )
+        .group_by(OpsMonitorReview.monitor_result_id)
+        .subquery("latest_monitor_review")
+    )
+
+
+def _model_statement(
+    workspace_id: str,
+    *,
+    bank: str | None = None,
+    category: str | None = None,
+    status: str | None = None,
+    keyword: str | None = None,
+    model_id: str | None = None,
+) -> Select[Any]:
+    latest_result = _latest_result_subquery(workspace_id)
+    latest_review = _latest_review_subquery()
+    statement = (
+        select(
+            OpsModelInstance.model_instance_id.label("model_instance_id"),
+            OpsBank.bank_name.label("bank_name"),
+            OpsModelInstance.category_code.label("model_category"),
+            OpsModelInstance.model_name.label("model_name"),
+            OpsModelInstance.model_id.label("model_id"),
+            OpsModelVersion.version_label.label("model_version"),
+            OpsModelInstance.model_status.label("model_status"),
+            OpsModelVersion.last_iteration_date.label("last_iteration_date"),
+            latest_result.c.ranking_result.label("ranking_result"),
+            latest_result.c.ks_value.label("ks_value"),
+            latest_result.c.psi_value.label("psi_value"),
+            OpsMonitorEvaluation.ks_mom_drop_rate.label("ks_mom_drop_rate"),
+            OpsMonitorEvaluation.secondary_level2_hits_6m.label(
+                "secondary_level2_hits_6m"
+            ),
+            OpsBank.is_wuji_bank.label("is_wuji_bank"),
+            latest_review.c.handled_at.label("last_processed_at"),
+            OpsMonitorEvaluation.action_snapshot.label("previous_advice"),
+            OpsModelInstance.common_model_name.label("common_model_name"),
+        )
+        .select_from(OpsModelInstance)
+        .join(OpsBank, OpsBank.bank_id == OpsModelInstance.bank_id)
+        .outerjoin(
+            OpsModelVersion,
+            and_(
+                OpsModelVersion.model_version_id
+                == OpsModelInstance.current_version_id,
+                OpsModelVersion.is_deleted == 0,
+            ),
+        )
+        .outerjoin(
+            latest_result,
+            latest_result.c.model_instance_id
+            == OpsModelInstance.model_instance_id,
+        )
+        .outerjoin(
+            OpsMonitorEvaluation,
+            and_(
+                OpsMonitorEvaluation.monitor_result_id
+                == latest_result.c.monitor_result_id,
+                OpsMonitorEvaluation.is_current == 1,
+                OpsMonitorEvaluation.is_deleted == 0,
+            ),
+        )
+        .outerjoin(
+            latest_review,
+            latest_review.c.monitor_result_id
+            == latest_result.c.monitor_result_id,
+        )
+        .where(
+            OpsModelInstance.workspace_id == workspace_id,
+            OpsModelInstance.is_deleted == 0,
+            OpsBank.workspace_id == workspace_id,
+            OpsBank.is_deleted == 0,
+        )
+    )
+    if bank:
+        statement = statement.where(OpsBank.bank_name == bank)
+    if category:
+        statement = statement.where(OpsModelInstance.category_code == category)
+    if status:
+        statement = statement.where(
+            OpsModelInstance.model_status == _MODEL_STATUS_TO_DB.get(status, status)
+        )
+    if model_id:
+        statement = statement.where(OpsModelInstance.model_id == model_id)
+    if keyword and (term := keyword.strip()):
+        like_term = f"%{term}%"
+        statement = statement.where(
+            or_(
+                OpsBank.bank_name.ilike(like_term),
+                OpsModelInstance.model_name.ilike(like_term),
+                OpsModelInstance.model_id.ilike(like_term),
+                OpsModelVersion.version_label.ilike(like_term),
+            )
+        )
+    return statement.order_by(OpsBank.bank_name, OpsModelInstance.model_id)
+
+
+def _model_dto(row: Mapping[str, Any]) -> OperationsModelDto:
+    return OperationsModelDto(
+        model_instance_id=row["model_instance_id"],
+        bank_name=row["bank_name"],
+        model_category=row["model_category"],
+        model_name=row["model_name"],
+        model_id=row["model_id"],
+        model_version=row["model_version"] or "—",
+        model_status=_MODEL_STATUS_TO_API.get(row["model_status"], "正常"),
+        last_iteration_date=_date_text(row["last_iteration_date"]),
+        ranking_result=_RANKING_TO_API.get(row["ranking_result"], "—"),
+        ks=_percent(row["ks_value"]),
+        psi=_percent(row["psi_value"]),
+        ks_mom_drop=_percent(row["ks_mom_drop_rate"]),
+        secondary_hits_6m=int(row["secondary_level2_hits_6m"] or 0),
+        is_wuji_bank=bool(row["is_wuji_bank"]),
+        last_processed_at=_datetime_date_text(row["last_processed_at"]),
+        previous_advice=row["previous_advice"] or "暂无历史处理建议",
+        common_model_name=row["common_model_name"],
+    )
+
+
+async def list_models(
+    session: AsyncSession,
+    workspace_id: str,
+    *,
+    bank: str | None = None,
+    category: str | None = None,
+    status: str | None = None,
+    keyword: str | None = None,
+) -> list[OperationsModelDto]:
+    rows = (
+        await session.execute(
+            _model_statement(
+                workspace_id,
+                bank=bank,
+                category=category,
+                status=status,
+                keyword=keyword,
+            )
+        )
+    ).mappings()
+    return [_model_dto(row) for row in rows]
+
+
+async def get_model(
+    session: AsyncSession,
+    workspace_id: str,
+    model_id: str,
+) -> OperationsModelDto | None:
+    row = (
+        await session.execute(
+            _model_statement(workspace_id, model_id=model_id).limit(1)
+        )
+    ).mappings().first()
+    return _model_dto(row) if row else None
+
+
+async def get_monthly_monitoring_result(
+    session: AsyncSession,
+    workspace_id: str,
+    model_id: str,
+    monitor_month: datetime.date,
+) -> MonthlyMonitoringResultDto | None:
+    latest_result = _latest_result_subquery(workspace_id, monitor_month)
+    statement = (
+        select(
+            OpsModelInstance.model_instance_id.label("model_instance_id"),
+            latest_result.c.monitor_month.label("monitor_month"),
+            latest_result.c.ranking_result.label("ranking_result"),
+            latest_result.c.ks_value.label("ks_value"),
+            latest_result.c.psi_value.label("psi_value"),
+            OpsMonitorEvaluation.ks_mom_drop_rate.label("ks_mom_drop_rate"),
+            OpsMonitorEvaluation.secondary_level2_hits_6m.label(
+                "secondary_level2_hits_6m"
+            ),
+        )
+        .select_from(OpsModelInstance)
+        .join(
+            latest_result,
+            latest_result.c.model_instance_id
+            == OpsModelInstance.model_instance_id,
+        )
+        .outerjoin(
+            OpsMonitorEvaluation,
+            and_(
+                OpsMonitorEvaluation.monitor_result_id
+                == latest_result.c.monitor_result_id,
+                OpsMonitorEvaluation.is_current == 1,
+                OpsMonitorEvaluation.is_deleted == 0,
+            ),
+        )
+        .where(
+            OpsModelInstance.workspace_id == workspace_id,
+            OpsModelInstance.model_id == model_id,
+            OpsModelInstance.is_deleted == 0,
+        )
+        .limit(1)
+    )
+    row = (await session.execute(statement)).mappings().first()
+    if not row:
+        return None
+    return MonthlyMonitoringResultDto(
+        model_instance_id=row["model_instance_id"],
+        monitor_month=row["monitor_month"].strftime("%Y-%m"),
+        ranking_result=_RANKING_TO_API.get(row["ranking_result"], "—"),
+        ks=_percent(row["ks_value"]),
+        psi=_percent(row["psi_value"]),
+        ks_mom_drop=_percent(row["ks_mom_drop_rate"]),
+        secondary_hits_6m=int(row["secondary_level2_hits_6m"] or 0),
+    )
+
+
+__all__ = [
+    "get_model",
+    "get_monthly_monitoring_result",
+    "list_models",
+    "parse_monitor_month",
+]

backend/src/backend/services/operations/streams.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from loguru import logger
+from redis.asyncio import Redis
+from redis.exceptions import RedisError, ResponseError
+
+
+class OperationsEventStream:
+    """Best-effort Redis Streams transport fed by the durable MySQL Outbox."""
+
+    def __init__(
+        self,
+        client: Redis | None,
+        *,
+        stream_name: str,
+        maxlen: int,
+    ) -> None:
+        self._client = client
+        self.stream_name = stream_name
+        self._maxlen = maxlen
+
+    async def publish(
+        self,
+        event_type: str,
+        payload: dict[str, Any],
+        *,
+        event_id: str,
+        idempotency_key: str = "",
+        trace_id: str = "",
+        schema_version: int = 1,
+    ) -> str | None:
+        if self._client is None:
+            return None
+        fields = {
+            "event_id": event_id,
+            "event_type": event_type,
+            "schema_version": str(schema_version),
+            "idempotency_key": idempotency_key,
+            "trace_id": trace_id,
+            "payload": json.dumps(
+                payload,
+                ensure_ascii=False,
+                separators=(",", ":"),
+                default=str,
+            ),
+        }
+        try:
+            return await self._client.xadd(
+                self.stream_name,
+                fields,
+                maxlen=self._maxlen,
+                approximate=True,
+            )
+        except RedisError as exc:
+            logger.warning(
+                "operations stream publish failed for {event_id}: {error}",
+                event_id=event_id,
+                error=exc,
+            )
+            return None
+
+    async def ensure_group(self, group_name: str) -> bool:
+        if self._client is None:
+            return False
+        try:
+            await self._client.xgroup_create(
+                self.stream_name,
+                group_name,
+                id="0-0",
+                mkstream=True,
+            )
+            return True
+        except ResponseError as exc:
+            if "BUSYGROUP" in str(exc):
+                return True
+            logger.warning("operations stream group creation failed: {error}", error=exc)
+            return False
+        except RedisError as exc:
+            logger.warning("operations stream group creation failed: {error}", error=exc)
+            return False
+
+    async def read_group(
+        self,
+        group_name: str,
+        consumer_name: str,
+        *,
+        count: int = 20,
+        block_ms: int = 5000,
+    ) -> list[tuple[str, dict[str, str]]]:
+        if self._client is None:
+            return []
+        try:
+            response = await self._client.xreadgroup(
+                group_name,
+                consumer_name,
+                {self.stream_name: ">"},
+                count=count,
+                block=block_ms,
+            )
+            if not response:
+                return []
+            return [(message_id, fields) for _, messages in response for message_id, fields in messages]
+        except RedisError as exc:
+            logger.warning("operations stream read failed: {error}", error=exc)
+            return []
+
+    async def acknowledge(self, group_name: str, *message_ids: str) -> int:
+        if self._client is None or not message_ids:
+            return 0
+        try:
+            return int(
+                await self._client.xack(
+                    self.stream_name,
+                    group_name,
+                    *message_ids,
+                )
+            )
+        except RedisError as exc:
+            logger.warning("operations stream ack failed: {error}", error=exc)
+            return 0
+
+
+__all__ = ["OperationsEventStream"]

backend/tests/test_operations_queries.py

@@ -0,0 +1,233 @@
+from __future__ import annotations
+
+import datetime
+from decimal import Decimal
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import StaticPool
+
+from backend.services.operations import (
+    get_model,
+    get_monthly_monitoring_result,
+    list_models,
+    parse_monitor_month,
+)
+from common.config import Settings
+from common.db.models.operations import (
+    OperationsBase,
+    OpsBank,
+    OpsModelInstance,
+    OpsModelVersion,
+    OpsMonitorBatch,
+    OpsMonitorEvaluation,
+    OpsMonitorResult,
+    OpsMonitorReview,
+)
+
+WORKSPACE_ID = "W" * 26
+OTHER_WORKSPACE_ID = "X" * 26
+MODEL_INSTANCE_ID = "M" * 26
+MODEL_VERSION_ID = "V" * 26
+MODEL_ID = "JC-STD-001"
+
+
+@pytest.fixture
+async def session():
+    engine = create_async_engine(
+        "sqlite+aiosqlite://",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as connection:
+        await connection.run_sync(OperationsBase.metadata.create_all)
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with factory() as database_session:
+        await _seed(database_session)
+        await database_session.commit()
+        yield database_session
+    await engine.dispose()
+
+
+async def _seed(session) -> None:
+    bank = OpsBank(
+        bank_id="B" * 26,
+        workspace_id=WORKSPACE_ID,
+        bank_code="JC",
+        bank_name="江城银行",
+        is_wuji_bank=1,
+        bank_status="active",
+        source_bank_ref="bank-jc",
+    )
+    version = OpsModelVersion(
+        model_version_id=MODEL_VERSION_ID,
+        workspace_id=WORKSPACE_ID,
+        model_instance_id=MODEL_INSTANCE_ID,
+        version_label="v2.3",
+        version_status="active",
+        last_iteration_date=datetime.date(2026, 6, 18),
+        source_version_ref="version-jc-std-v23",
+    )
+    model = OpsModelInstance(
+        model_instance_id=MODEL_INSTANCE_ID,
+        workspace_id=WORKSPACE_ID,
+        bank_id=bank.bank_id,
+        category_code="std",
+        model_id=MODEL_ID,
+        model_name="标准A卡",
+        model_status="normal",
+        current_version_id=version.model_version_id,
+        common_model_name="标准A卡通用版 v2.3",
+        source_model_ref="model-jc-std",
+    )
+    old_batch = OpsMonitorBatch(
+        batch_id="1" * 26,
+        workspace_id=WORKSPACE_ID,
+        source_batch_no="2026-07-r1",
+        monitor_month=datetime.date(2026, 7, 1),
+        revision_no=1,
+        batch_status="published",
+        published_at=datetime.datetime(2026, 7, 20, 8, 0),
+    )
+    new_batch = OpsMonitorBatch(
+        batch_id="2" * 26,
+        workspace_id=WORKSPACE_ID,
+        source_batch_no="2026-07-r2",
+        monitor_month=datetime.date(2026, 7, 1),
+        revision_no=2,
+        batch_status="published",
+        published_at=datetime.datetime(2026, 7, 21, 8, 0),
+    )
+    old_result = OpsMonitorResult(
+        monitor_result_id="3" * 26,
+        workspace_id=WORKSPACE_ID,
+        batch_id=old_batch.batch_id,
+        model_instance_id=MODEL_INSTANCE_ID,
+        model_version_id=MODEL_VERSION_ID,
+        monitor_month=datetime.date(2026, 7, 1),
+        ranking_result="matched",
+        ks_value=Decimal("0.41000000"),
+        psi_value=Decimal("0.09000000"),
+    )
+    new_result = OpsMonitorResult(
+        monitor_result_id="4" * 26,
+        workspace_id=WORKSPACE_ID,
+        batch_id=new_batch.batch_id,
+        model_instance_id=MODEL_INSTANCE_ID,
+        model_version_id=MODEL_VERSION_ID,
+        monitor_month=datetime.date(2026, 7, 1),
+        ranking_result="unmatched",
+        ks_value=Decimal("0.36840000"),
+        psi_value=Decimal("0.27400000"),
+    )
+    evaluation = OpsMonitorEvaluation(
+        evaluation_id="5" * 26,
+        workspace_id=WORKSPACE_ID,
+        monitor_result_id=new_result.monitor_result_id,
+        rule_version_id="6" * 26,
+        rule_item_id="7" * 26,
+        ks_mom_drop_rate=Decimal("0.24100000"),
+        secondary_level2_hits_6m=3,
+        abnormal_level="level3",
+        monitor_grade="C",
+        reason_code="R-C-001",
+        reason_text_snapshot="排序性不符、KS低、PSI高",
+        action_snapshot="建议启动模型微调或重构评估",
+        is_current=1,
+        evaluated_at=datetime.datetime(2026, 7, 21, 9, 0),
+    )
+    review = OpsMonitorReview(
+        review_id="8" * 26,
+        workspace_id=WORKSPACE_ID,
+        monitor_result_id=new_result.monitor_result_id,
+        evaluation_id=evaluation.evaluation_id,
+        review_stage="model_initial",
+        review_status="handled",
+        decision_code="tune_or_rebuild",
+        handling_note="建议评估重构",
+        handled_by="U" * 26,
+        handled_at=datetime.datetime(2026, 7, 22, 10, 30),
+    )
+    other_bank = OpsBank(
+        bank_id="C" * 26,
+        workspace_id=OTHER_WORKSPACE_ID,
+        bank_code="OTHER",
+        bank_name="其他银行",
+        bank_status="active",
+        source_bank_ref="bank-other",
+    )
+    other_model = OpsModelInstance(
+        model_instance_id="N" * 26,
+        workspace_id=OTHER_WORKSPACE_ID,
+        bank_id=other_bank.bank_id,
+        category_code="std",
+        model_id="OTHER-STD-001",
+        model_name="其他模型",
+        model_status="normal",
+        source_model_ref="model-other",
+    )
+    session.add_all(
+        [
+            bank,
+            version,
+            model,
+            old_batch,
+            new_batch,
+            old_result,
+            new_result,
+            evaluation,
+            review,
+            other_bank,
+            other_model,
+        ]
+    )
+
+
+async def test_list_models_uses_latest_revision_and_workspace(session) -> None:
+    models = await list_models(session, WORKSPACE_ID)
+    assert len(models) == 1
+    model = models[0]
+    assert model.model_id == MODEL_ID
+    assert model.ks == 36.84
+    assert model.psi == 27.4
+    assert model.ks_mom_drop == 24.1
+    assert model.ranking_result == "不符"
+    assert model.last_processed_at == "2026-07-22"
+    assert model.is_wuji_bank is True
+
+
+async def test_list_models_filters_and_detail(session) -> None:
+    assert len(await list_models(session, WORKSPACE_ID, bank="江城银行")) == 1
+    assert len(await list_models(session, WORKSPACE_ID, category="big")) == 0
+    assert len(await list_models(session, WORKSPACE_ID, keyword="v2.3")) == 1
+    model = await get_model(session, WORKSPACE_ID, MODEL_ID)
+    assert model is not None
+    assert model.model_version == "v2.3"
+    assert model.common_model_name == "标准A卡通用版 v2.3"
+    assert await get_model(session, OTHER_WORKSPACE_ID, MODEL_ID) is None
+
+
+async def test_get_monthly_monitoring_result(session) -> None:
+    result = await get_monthly_monitoring_result(
+        session,
+        WORKSPACE_ID,
+        MODEL_ID,
+        datetime.date(2026, 7, 1),
+    )
+    assert result is not None
+    assert result.monitor_month == "2026-07"
+    assert result.ks == 36.84
+    assert result.psi == 27.4
+    assert result.secondary_hits_6m == 3
+
+
+def test_monitor_month_and_operations_url() -> None:
+    assert parse_monitor_month("2026-07") == datetime.date(2026, 7, 1)
+    with pytest.raises(ValueError):
+        parse_monitor_month("2026-13")
+    settings = Settings(
+        database_url="mysql+asyncmy://user:pass@db:3306/model_platform?charset=utf8mb4",
+        _env_file=None,
+    )
+    assert settings.operations_database_url is not None
+    assert "/model_operations?" in settings.operations_database_url

backend/tests/test_operations_redis.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import fnmatch
+
+from backend.services.operations import OperationsCache, OperationsEventStream
+
+
+class FakeRedis:
+    def __init__(self) -> None:
+        self.values: dict[str, str] = {}
+        self.expirations: dict[str, int] = {}
+        self.stream_entries: list[tuple[str, dict[str, str]]] = []
+        self.groups: set[str] = set()
+
+    async def get(self, key: str):
+        return self.values.get(key)
+
+    async def set(self, key: str, value: str, *, ex: int):
+        self.values[key] = value
+        self.expirations[key] = ex
+        return True
+
+    async def unlink(self, *keys: str):
+        removed = 0
+        for key in keys:
+            if key in self.values:
+                removed += 1
+                self.values.pop(key)
+                self.expirations.pop(key, None)
+        return removed
+
+    async def scan_iter(self, *, match: str, count: int):
+        del count
+        for key in list(self.values):
+            if fnmatch.fnmatch(key, match):
+                yield key
+
+    async def xadd(self, stream: str, fields, *, maxlen: int, approximate: bool):
+        assert maxlen == 1000
+        assert approximate is True
+        message_id = f"1-{len(self.stream_entries)}"
+        self.stream_entries.append((message_id, dict(fields)))
+        return message_id
+
+    async def xgroup_create(self, stream: str, group: str, *, id: str, mkstream: bool):
+        assert stream == "ops:events"
+        assert id == "0-0"
+        assert mkstream is True
+        self.groups.add(group)
+        return True
+
+    async def xreadgroup(self, group, consumer, streams, *, count: int, block: int):
+        del group, consumer, count, block
+        stream = next(iter(streams))
+        return [(stream, list(self.stream_entries))]
+
+    async def xack(self, stream: str, group: str, *message_ids: str):
+        del stream, group
+        return len(message_ids)
+
+
+async def test_cache_round_trip_and_workspace_invalidation() -> None:
+    redis = FakeRedis()
+    cache = OperationsCache(
+        redis,  # type: ignore[arg-type]
+        prefix="ops",
+        default_ttl_seconds=300,
+    )
+    key = cache.build_key("W1", "model-overview", {"month": "2026-07", "bank": "江城"})
+    same_key = cache.build_key("W1", "model-overview", {"bank": "江城", "month": "2026-07"})
+    other_key = cache.build_key("W2", "model-overview", {"bank": "江城"})
+    assert key == same_key
+    assert key != other_key
+    assert await cache.set_json(key, {"total": 3, "name": "标准A卡"}) is True
+    assert redis.expirations[key] == 300
+    assert await cache.get_json(key) == {"total": 3, "name": "标准A卡"}
+    await cache.set_json(other_key, {"total": 1})
+    assert await cache.invalidate_resource("W1", "model-overview") == 1
+    assert await cache.get_json(key) is None
+    assert await cache.get_json(other_key) == {"total": 1}
+
+
+async def test_cache_without_redis_is_a_noop() -> None:
+    cache = OperationsCache(None, prefix="ops", default_ttl_seconds=300)
+    assert await cache.get_json("missing") is None
+    assert await cache.set_json("missing", {"value": 1}) is False
+    assert await cache.delete("missing") == 0
+
+
+async def test_stream_publish_group_read_and_ack() -> None:
+    redis = FakeRedis()
+    stream = OperationsEventStream(
+        redis,  # type: ignore[arg-type]
+        stream_name="ops:events",
+        maxlen=1000,
+    )
+    assert await stream.ensure_group("notifications") is True
+    message_id = await stream.publish(
+        "monitor.review.due",
+        {"review_id": "R1", "bank": "江城银行"},
+        event_id="E1",
+        idempotency_key="review:R1:due",
+        trace_id="T1",
+    )
+    assert message_id == "1-0"
+    messages = await stream.read_group("notifications", "worker-1")
+    assert messages[0][0] == "1-0"
+    assert messages[0][1]["event_type"] == "monitor.review.due"
+    assert '"review_id":"R1"' in messages[0][1]["payload"]
+    assert await stream.acknowledge("notifications", message_id) == 1