Add risk and audit admin APIs

src/application/abstractions/i_unit_of_work.py

@@ -11,6 +11,7 @@ from src.application.abstractions.repositories import (
     IUserRepository,
     IWalletRepository,
     IOrderRepository,
+    IRiskRepository,
 )
 
 
@@ -45,3 +46,6 @@ class IUnitOfWork(Protocol):
 
     @property
     def order_repository(self) -> IOrderRepository: ...
+
+    @property
+    def risk_repository(self) -> IRiskRepository: ...

src/application/abstractions/repositories/__init__.py

@@ -6,3 +6,4 @@ from src.application.abstractions.repositories.i_organization_wallet_repository
 from src.application.abstractions.repositories.i_purchase_request_repository import IPurchaseRequestRepository
 from src.application.abstractions.repositories.i_wallets_repository import IWalletRepository
 from src.application.abstractions.repositories.i_order_repository import IOrderRepository
+from src.application.abstractions.repositories.i_risk_repository import IRiskRepository

src/application/abstractions/repositories/i_risk_repository.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from src.application.domain.entities.risk import AuditEventEntity, FunnelEntity, RiskAssessmentEntity, RiskSummaryEntity
+
+
+class IRiskRepository(ABC):
+    @abstractmethod
+    async def list_assessments(self, *, limit: int, offset: int, decision: str | None, user_id: str | None, order_id: str | None) -> tuple[list[RiskAssessmentEntity], int]:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def summary(self) -> RiskSummaryEntity:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def list_audit_events(self, *, limit: int, offset: int, severity: str | None, action: str | None, entity_type: str | None) -> tuple[list[AuditEventEntity], int]:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def funnel(self) -> FunnelEntity:
+        raise NotImplementedError

src/application/commands/__init__.py

@@ -44,6 +44,7 @@ from src.application.commands.orders_commands import (
     ListOrdersByDateCommand,
     GetOrderCommand,
 )
+from src.application.commands.risk_commands import GetFunnelCommand,GetRiskSummaryCommand,ListAuditEventsCommand,ListRiskAssessmentsCommand
 from src.application.commands.analytics_commands import (
     GetPurchaseAnalyticsForYearCommand,
     GetOrdersAnalyticsForYearCommand,
@@ -92,4 +93,8 @@ __all__ = [
     'GetOrdersAnalyticsForMonthCommand',
     'GetPurchaseAnalyticsForWeekCommand',
     'GetOrdersAnalyticsForWeekCommand',
+    'ListRiskAssessmentsCommand',
+    'GetRiskSummaryCommand',
+    'ListAuditEventsCommand',
+    'GetFunnelCommand',
 ]

src/application/commands/risk_commands.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from src.application.abstractions import IUnitOfWork
+from src.application.domain.entities.risk import AuditEventEntity, FunnelEntity, RiskAssessmentEntity, RiskSummaryEntity
+from src.infrastructure.database.decorators import transactional
+
+
+class ListRiskAssessmentsCommand:
+    def __init__(self, uow: IUnitOfWork):
+        self._uow = uow
+
+    @transactional
+    async def __call__(self, *, limit: int, offset: int, decision: str | None, user_id: str | None, order_id: str | None) -> tuple[list[RiskAssessmentEntity], int]:
+        return await self._uow.risk_repository.list_assessments(limit=limit, offset=offset, decision=decision, user_id=user_id, order_id=order_id)
+
+
+class GetRiskSummaryCommand:
+    def __init__(self, uow: IUnitOfWork):
+        self._uow = uow
+
+    @transactional
+    async def __call__(self) -> RiskSummaryEntity:
+        return await self._uow.risk_repository.summary()
+
+
+class ListAuditEventsCommand:
+    def __init__(self, uow: IUnitOfWork):
+        self._uow = uow
+
+    @transactional
+    async def __call__(self, *, limit: int, offset: int, severity: str | None, action: str | None, entity_type: str | None) -> tuple[list[AuditEventEntity], int]:
+        return await self._uow.risk_repository.list_audit_events(limit=limit, offset=offset, severity=severity, action=action, entity_type=entity_type)
+
+
+class GetFunnelCommand:
+    def __init__(self, uow: IUnitOfWork):
+        self._uow = uow
+
+    @transactional
+    async def __call__(self) -> FunnelEntity:
+        return await self._uow.risk_repository.funnel()

src/application/domain/entities/risk.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass(slots=True)
+class RiskAssessmentEntity:
+    id: str
+    user_id: str
+    order_id: str | None
+    subject_type: str
+    score: int
+    decision: str
+    reasons: list[str] = field(default_factory=list)
+    created_at: datetime | None = None
+
+
+@dataclass(slots=True)
+class AuditEventEntity:
+    id: str
+    actor_type: str
+    actor_id: str | None
+    action: str
+    entity_type: str
+    entity_id: str | None
+    severity: str
+    metadata: dict
+    created_at: datetime | None = None
+
+
+@dataclass(slots=True)
+class RiskSummaryEntity:
+    total: int
+    allow: int
+    manual_review: int
+    reject: int
+    average_score: float
+    high_risk: int
+
+
+@dataclass(slots=True)
+class FunnelEntity:
+    registrations: int
+    kyc_started: int
+    kyc_completed: int
+    first_payment: int
+    successful_operations: int

src/infrastructure/database/models/__init__.py

@@ -7,6 +7,7 @@ from src.infrastructure.database.models.organization_wallet import OrganizationW
 from src.infrastructure.database.models.purchase_request import PurchaseRequestModel
 from src.infrastructure.database.models.wallets import WalletModel
 from src.infrastructure.database.models.order import OrderModel
+from src.infrastructure.database.models.risk import AuditEventModel, RiskAssessmentModel
 
 __all__ = [
     'Base',

src/infrastructure/database/models/risk.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func, text
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.infrastructure.database.models.base import Base
+from src.infrastructure.database.models.mixins import AuditTimestampsMixin, UlidPrimaryKeyMixin
+
+
+class RiskAssessmentModel(Base, UlidPrimaryKeyMixin, AuditTimestampsMixin):
+    __tablename__ = "risk_assessments"
+
+    user_id: Mapped[str] = mapped_column(String(26), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False, index=True)
+    order_id: Mapped[str | None] = mapped_column(String(26), ForeignKey("orders.id", ondelete="SET NULL"), nullable=True, index=True)
+    subject_type: Mapped[str] = mapped_column(String(32), nullable=False, default="individual", server_default="individual", index=True)
+    score: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    decision: Mapped[str] = mapped_column(String(32), nullable=False, default="allow", index=True)
+    reasons: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list, server_default=text("'[]'::jsonb"))
+
+
+class AuditEventModel(Base, UlidPrimaryKeyMixin):
+    __tablename__ = "audit_events"
+
+    actor_type: Mapped[str] = mapped_column(String(32), nullable=False, default="system", server_default="system", index=True)
+    actor_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    action: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
+    entity_type: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    entity_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    severity: Mapped[str] = mapped_column(String(32), nullable=False, default="info", server_default="info", index=True)
+    metadata_json: Mapped[dict] = mapped_column("metadata", JSONB, nullable=False, default=dict, server_default=text("'{}'::jsonb"))
+    created_at: Mapped[DateTime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())

src/infrastructure/database/repositories/__init__.py

@@ -6,6 +6,7 @@ from src.infrastructure.database.repositories.organization_wallet_repository imp
 from src.infrastructure.database.repositories.purchase_request_repository import PurchaseRequestRepository
 from src.infrastructure.database.repositories.wallets_repository import WalletRepository
 from src.infrastructure.database.repositories.orders_repository import OrderRepository
+from src.infrastructure.database.repositories.risk_repository import RiskRepository
 
 __all__ = [
     'AdminUserRepository',

src/infrastructure/database/repositories/risk_repository.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from sqlalchemy import Integer, func, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.application.abstractions.repositories.i_risk_repository import IRiskRepository
+from src.application.contracts import ILogger
+from src.application.domain.entities.risk import AuditEventEntity, FunnelEntity, RiskAssessmentEntity, RiskSummaryEntity
+from src.infrastructure.database.models.order import OrderModel
+from src.infrastructure.database.models.risk import AuditEventModel, RiskAssessmentModel
+from src.infrastructure.database.models.user import UserModel
+
+
+class RiskRepository(IRiskRepository):
+    def __init__(self, session: AsyncSession, logger: ILogger):
+        self._session = session
+        self._logger = logger
+
+    @staticmethod
+    def _assessment(model: RiskAssessmentModel) -> RiskAssessmentEntity:
+        return RiskAssessmentEntity(
+            id=model.id,
+            user_id=model.user_id,
+            order_id=model.order_id,
+            subject_type=model.subject_type,
+            score=model.score,
+            decision=model.decision,
+            reasons=list(model.reasons or []),
+            created_at=model.created_at,
+        )
+
+    @staticmethod
+    def _event(model: AuditEventModel) -> AuditEventEntity:
+        return AuditEventEntity(
+            id=model.id,
+            actor_type=model.actor_type,
+            actor_id=model.actor_id,
+            action=model.action,
+            entity_type=model.entity_type,
+            entity_id=model.entity_id,
+            severity=model.severity,
+            metadata=dict(model.metadata_json or {}),
+            created_at=model.created_at,
+        )
+
+    async def list_assessments(self, *, limit: int, offset: int, decision: str | None, user_id: str | None, order_id: str | None) -> tuple[list[RiskAssessmentEntity], int]:
+        stmt = select(RiskAssessmentModel).order_by(RiskAssessmentModel.created_at.desc()).limit(limit).offset(offset)
+        count_stmt = select(func.count()).select_from(RiskAssessmentModel)
+        filters = []
+        if decision:
+            filters.append(RiskAssessmentModel.decision == decision)
+        if user_id:
+            filters.append(RiskAssessmentModel.user_id == user_id)
+        if order_id:
+            filters.append(RiskAssessmentModel.order_id == order_id)
+        if filters:
+            stmt = stmt.where(*filters)
+            count_stmt = count_stmt.where(*filters)
+        rows = (await self._session.scalars(stmt)).all()
+        total = int((await self._session.execute(count_stmt)).scalar_one())
+        return [self._assessment(row) for row in rows], total
+
+    async def summary(self) -> RiskSummaryEntity:
+        stmt = select(
+            func.count(),
+            func.coalesce(func.sum((RiskAssessmentModel.decision == "allow").cast(Integer)), 0),
+            func.coalesce(func.sum((RiskAssessmentModel.decision == "manual_review").cast(Integer)), 0),
+            func.coalesce(func.sum((RiskAssessmentModel.decision == "reject").cast(Integer)), 0),
+            func.coalesce(func.avg(RiskAssessmentModel.score), 0),
+            func.coalesce(func.sum((RiskAssessmentModel.score >= 60).cast(Integer)), 0),
+        )
+        row = (await self._session.execute(stmt)).one()
+        return RiskSummaryEntity(
+            total=int(row[0] or 0),
+            allow=int(row[1] or 0),
+            manual_review=int(row[2] or 0),
+            reject=int(row[3] or 0),
+            average_score=float(row[4] or 0),
+            high_risk=int(row[5] or 0),
+        )
+
+    async def list_audit_events(self, *, limit: int, offset: int, severity: str | None, action: str | None, entity_type: str | None) -> tuple[list[AuditEventEntity], int]:
+        stmt = select(AuditEventModel).order_by(AuditEventModel.created_at.desc()).limit(limit).offset(offset)
+        count_stmt = select(func.count()).select_from(AuditEventModel)
+        filters = []
+        if severity:
+            filters.append(AuditEventModel.severity == severity)
+        if action:
+            filters.append(AuditEventModel.action == action)
+        if entity_type:
+            filters.append(AuditEventModel.entity_type == entity_type)
+        if filters:
+            stmt = stmt.where(*filters)
+            count_stmt = count_stmt.where(*filters)
+        rows = (await self._session.scalars(stmt)).all()
+        total = int((await self._session.execute(count_stmt)).scalar_one())
+        return [self._event(row) for row in rows], total
+
+    async def funnel(self) -> FunnelEntity:
+        registrations = int((await self._session.execute(select(func.count()).select_from(UserModel))).scalar_one())
+        kyc_started = int((await self._session.execute(text("select count(distinct user_id) from kyc"))).scalar_one())
+        kyc_completed = int((await self._session.execute(select(func.count()).select_from(UserModel).where(UserModel.kyc_verified.is_(True)))).scalar_one())
+        first_payment = int((await self._session.execute(select(func.count(func.distinct(OrderModel.user_id))))).scalar_one())
+        successful_operations = int((await self._session.execute(text("select count(*) from payments where status = 'completed'"))).scalar_one())
+        return FunnelEntity(
+            registrations=registrations,
+            kyc_started=kyc_started,
+            kyc_completed=kyc_completed,
+            first_payment=first_payment,
+            successful_operations=successful_operations,
+        )

src/infrastructure/database/unit_of_work.py

@@ -10,6 +10,7 @@ from src.application.abstractions.repositories import (
     IUserRepository,
     IWalletRepository,
     IOrderRepository,
+    IRiskRepository,
 )
 from src.application.contracts import ILogger
 from src.application.domain.exceptions import RefreshConcurrentException
@@ -22,6 +23,7 @@ from src.infrastructure.database.repositories import (
     UserRepository,
     WalletRepository,
     OrderRepository,
+    RiskRepository,
 )
 
 
@@ -37,10 +39,12 @@ class UnitOfWork(IUnitOfWork):
         self._purchase_request_repository: IPurchaseRequestRepository | None = None
         self._wallet_repository: IWalletRepository | None = None
         self._order_repository: IOrderRepository | None = None
+        self._risk_repository: IRiskRepository | None = None
         self._logger: ILogger = logger
 
     async def __aenter__(self):
         self._session = self.session_factory()
+        self._risk_repository = None
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
@@ -114,3 +118,9 @@ class UnitOfWork(IUnitOfWork):
                 session=self._session, logger=self._logger
             )
         return self._order_repository
+
+    @property
+    def risk_repository(self) -> IRiskRepository:
+        if self._risk_repository is None:
+            self._risk_repository = RiskRepository(session=self._session, logger=self._logger)
+        return self._risk_repository

src/presentation/dependencies/commands.py

@@ -42,6 +42,10 @@ from src.application.commands import (
     GetOrdersAnalyticsForMonthCommand,
     GetPurchaseAnalyticsForWeekCommand,
     GetOrdersAnalyticsForWeekCommand,
+    ListRiskAssessmentsCommand,
+    GetRiskSummaryCommand,
+    ListAuditEventsCommand,
+    GetFunnelCommand,
 )
 from src.application.contracts import IHashService, IJwtService, ILogger
 from src.infrastructure.config import settings
@@ -320,3 +324,27 @@ def get_get_orders_analytics_for_week_command(
     logger: ILogger = Depends(get_logger),
 ) -> GetOrdersAnalyticsForWeekCommand:
     return GetOrdersAnalyticsForWeekCommand(uow, logger)
+
+
+def get_list_risk_assessments_command(
+    uow: IUnitOfWork = Depends(get_unit_of_work),
+) -> ListRiskAssessmentsCommand:
+    return ListRiskAssessmentsCommand(uow)
+
+
+def get_get_risk_summary_command(
+    uow: IUnitOfWork = Depends(get_unit_of_work),
+) -> GetRiskSummaryCommand:
+    return GetRiskSummaryCommand(uow)
+
+
+def get_list_audit_events_command(
+    uow: IUnitOfWork = Depends(get_unit_of_work),
+) -> ListAuditEventsCommand:
+    return ListAuditEventsCommand(uow)
+
+
+def get_get_funnel_command(
+    uow: IUnitOfWork = Depends(get_unit_of_work),
+) -> GetFunnelCommand:
+    return GetFunnelCommand(uow)

src/presentation/routing/__init__.py

@@ -9,6 +9,7 @@ from src.presentation.routing.users import users_router
 from src.presentation.routing.search import search_router
 from src.presentation.routing.orders import orders_router
 from src.presentation.routing.analytics import analytics_router
+from src.presentation.routing.risk import audit_router, risk_router
 
 v1_router = APIRouter(prefix='/v1')
 v1_router.include_router(auth_router)
@@ -20,3 +21,5 @@ v1_router.include_router(users_router)
 v1_router.include_router(search_router)
 v1_router.include_router(orders_router)
 v1_router.include_router(analytics_router)
+v1_router.include_router(risk_router)
+v1_router.include_router(audit_router)

src/presentation/routing/analytics.py

@@ -9,11 +9,13 @@ from src.presentation.dependencies.commands import (
     get_get_orders_analytics_for_month_command,
     get_get_purchase_analytics_for_week_command,
     get_get_orders_analytics_for_week_command,
+    get_get_funnel_command,
 )
 from src.presentation.schemas.analytics import (
     OrderAnalyticsResponse,
     PurchaseRequestAnalyticsResponse,
 )
+from src.presentation.schemas.risk import FunnelResponse
 from src.application.commands.analytics_commands import (
     GetPurchaseAnalyticsForYearCommand,
     GetOrdersAnalyticsForYearCommand,
@@ -22,6 +24,7 @@ from src.application.commands.analytics_commands import (
     GetPurchaseAnalyticsForWeekCommand,
     GetOrdersAnalyticsForWeekCommand,
 )
+from src.application.commands.risk_commands import GetFunnelCommand
 
 analytics_router = APIRouter(prefix='/analytics', tags=['analytics'])
 
@@ -158,3 +161,12 @@ async def get_orders_analytics_for_week(
         total_summed_service_fees=total_sum,
         period=period,
     )
+
+
+@analytics_router.get('/funnel', response_model=FunnelResponse)
+async def get_funnel(
+    auth: AdminAuthContext = Depends(require_admin_access),
+    command: GetFunnelCommand = Depends(get_get_funnel_command),
+):
+    item = await command()
+    return FunnelResponse(**item.__dict__)

src/presentation/routing/risk.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, Query
+
+from src.application.commands.risk_commands import GetRiskSummaryCommand, ListAuditEventsCommand, ListRiskAssessmentsCommand
+from src.application.domain.dto import AdminAuthContext
+from src.presentation.decorators.admin_auth import require_admin_access
+from src.presentation.dependencies.commands import get_get_risk_summary_command, get_list_audit_events_command, get_list_risk_assessments_command
+from src.presentation.schemas.risk import AuditEventResponse, AuditEventsResponse, RiskAssessmentResponse, RiskAssessmentsResponse, RiskSummaryResponse
+
+risk_router = APIRouter(prefix="/risk", tags=["risk"])
+audit_router = APIRouter(prefix="/audit", tags=["audit"])
+
+
+@risk_router.get("/assessments", response_model=RiskAssessmentsResponse)
+async def list_risk_assessments(
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
+    decision: str | None = Query(default=None, min_length=1, max_length=32),
+    user_id: str | None = Query(default=None, min_length=1, max_length=64),
+    order_id: str | None = Query(default=None, min_length=1, max_length=64),
+    auth: AdminAuthContext = Depends(require_admin_access),
+    command: ListRiskAssessmentsCommand = Depends(get_list_risk_assessments_command),
+):
+    items, total = await command(limit=limit, offset=offset, decision=decision, user_id=user_id, order_id=order_id)
+    return RiskAssessmentsResponse(
+        items=[
+            RiskAssessmentResponse(
+                id=x.id,
+                user_id=x.user_id,
+                order_id=x.order_id,
+                subject_type=x.subject_type,
+                score=x.score,
+                decision=x.decision,
+                reasons=x.reasons,
+                created_at=x.created_at.isoformat() if x.created_at else None,
+            )
+            for x in items
+        ],
+        total=total,
+    )
+
+
+@risk_router.get("/summary", response_model=RiskSummaryResponse)
+async def get_risk_summary(
+    auth: AdminAuthContext = Depends(require_admin_access),
+    command: GetRiskSummaryCommand = Depends(get_get_risk_summary_command),
+):
+    item = await command()
+    return RiskSummaryResponse(**item.__dict__)
+
+
+@audit_router.get("/events", response_model=AuditEventsResponse)
+async def list_audit_events(
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
+    severity: str | None = Query(default=None, min_length=1, max_length=32),
+    action: str | None = Query(default=None, min_length=1, max_length=128),
+    entity_type: str | None = Query(default=None, min_length=1, max_length=64),
+    auth: AdminAuthContext = Depends(require_admin_access),
+    command: ListAuditEventsCommand = Depends(get_list_audit_events_command),
+):
+    items, total = await command(limit=limit, offset=offset, severity=severity, action=action, entity_type=entity_type)
+    return AuditEventsResponse(
+        items=[
+            AuditEventResponse(
+                id=x.id,
+                actor_type=x.actor_type,
+                actor_id=x.actor_id,
+                action=x.action,
+                entity_type=x.entity_type,
+                entity_id=x.entity_id,
+                severity=x.severity,
+                metadata=x.metadata,
+                created_at=x.created_at.isoformat() if x.created_at else None,
+            )
+            for x in items
+        ],
+        total=total,
+    )

src/presentation/schemas/risk.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class RiskAssessmentResponse(BaseModel):
+    id: str
+    user_id: str
+    order_id: str | None = None
+    subject_type: str
+    score: int
+    decision: str
+    reasons: list[str]
+    created_at: str | None = None
+
+
+class RiskAssessmentsResponse(BaseModel):
+    items: list[RiskAssessmentResponse]
+    total: int
+
+
+class RiskSummaryResponse(BaseModel):
+    total: int
+    allow: int
+    manual_review: int
+    reject: int
+    average_score: float
+    high_risk: int
+
+
+class AuditEventResponse(BaseModel):
+    id: str
+    actor_type: str
+    actor_id: str | None = None
+    action: str
+    entity_type: str
+    entity_id: str | None = None
+    severity: str
+    metadata: dict
+    created_at: str | None = None
+
+
+class AuditEventsResponse(BaseModel):
+    items: list[AuditEventResponse]
+    total: int
+
+
+class FunnelResponse(BaseModel):
+    registrations: int
+    kyc_started: int
+    kyc_completed: int
+    first_payment: int
+    successful_operations: int