Add payment risk scoring

src/application/abstractions/i_unit_of_work.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 from typing import Protocol, runtime_checkable
-from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,ISbpWithdrawalRepository,IUserRepository
+from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,ISbpWithdrawalRepository,IUserRepository,IRiskRepository
 
 
 @runtime_checkable
@@ -23,3 +23,6 @@ class IUnitOfWork(Protocol):
     @property
     def user_repository(self) -> IUserRepository: ...
 
+    @property
+    def risk_repository(self) -> IRiskRepository: ...
+

src/application/abstractions/repositories/__init__.py

@@ -1,4 +1,5 @@
 from src.application.abstractions.repositories.i_order_repository import IOrderRepository
 from src.application.abstractions.repositories.i_payment_repository import IPaymentRepository
 from src.application.abstractions.repositories.i_sbp_withdrawal_repository import ISbpWithdrawalRepository
-from src.application.abstractions.repositories.i_user_repository import IUserRepository
+from src.application.abstractions.repositories.i_user_repository import IUserRepository
+from src.application.abstractions.repositories.i_risk_repository import IRiskRepository

src/application/abstractions/repositories/i_order_repository.py

@@ -1,4 +1,5 @@
 from abc import ABC,abstractmethod
+from datetime import datetime
 
 from src.application.domain.entities.order import OrderEntity
 from src.application.domain.enums import OrderStatus
@@ -30,6 +31,12 @@ class IOrderRepository(ABC):
         raise NotImplementedError
 
 
+    @abstractmethod
+    @abstractmethod
+    async def count_recent_by_user(self,*,user_id: str,since: datetime) -> int:
+        raise NotImplementedError
+
+
     @abstractmethod
     async def update_after_itpay_payment_created(self,order: OrderEntity) -> OrderEntity:
         raise NotImplementedError

src/application/abstractions/repositories/i_risk_repository.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from src.application.domain.entities.risk import AuditEventEntity, RiskAssessmentEntity
+
+
+class IRiskRepository(ABC):
+    @abstractmethod
+    async def create_assessment(self, assessment: RiskAssessmentEntity) -> RiskAssessmentEntity:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def create_audit_event(self, event: AuditEventEntity) -> AuditEventEntity:
+        raise NotImplementedError

src/application/commands/create_order_command.py

@@ -1,14 +1,15 @@
 from __future__ import annotations
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from decimal import Decimal
 from ulid import ULID
 from src.application.abstractions import IUnitOfWork
 from src.application.contracts import ILogger
 from src.application.contracts import IItPayService
 from src.application.domain.entities.order import OrderEntity
+from src.application.domain.entities.risk import AuditEventEntity, RiskAssessmentEntity
 from src.application.domain.enums import OrderStatus
 from src.application.domain.exceptions import ApplicationException, ForbiddenException, OrderTotalOutOfRangeException, PriceChangedException
-from src.application.services import PaymentQuoteService
+from src.application.services import PaymentQuoteService, RiskScoringService
 from src.application.services.payment_quote_service import MIN_TOTAL_RUB
 from src.infrastructure.database.decorators import transactional
 from src.presentation.schemas.order import CreateOrder
@@ -23,12 +24,13 @@ class CreateOrderCommand:
             logger: ILogger,
             payment_quote_service: PaymentQuoteService,
             itpay_service: IItPayService,
+            risk_scoring_service: RiskScoringService,
     ) -> None:
         self._unit_of_work = unit_of_work
         self._logger = logger
         self._payment_quote_service = payment_quote_service
         self._itpay_service = itpay_service
-
+        self._risk_scoring_service = risk_scoring_service
 
     @transactional
     async def __call__(self, payment_data: CreateOrder, user_id: str) -> OrderEntity:
@@ -38,6 +40,27 @@ class CreateOrderCommand:
         if user.account_type == 'legal_entity':
             raise ForbiddenException(message='USDT purchase orders are not available for legal entity accounts')
 
+        recent_order_count = await self._unit_of_work.order_repository.count_recent_by_user(
+            user_id=user_id,
+            since=datetime.now(timezone.utc) - timedelta(minutes=10),
+        )
+        risk = self._risk_scoring_service.assess_order(
+            user=user,
+            total_price=payment_data.total_price,
+            recent_order_count=recent_order_count,
+        )
+        if risk.decision == 'reject' and 'kyc_not_completed' in risk.reasons:
+            await self._unit_of_work.risk_repository.create_audit_event(AuditEventEntity(
+                actor_type='user',
+                actor_id=user_id,
+                action='payment_blocked_kyc_required',
+                entity_type='user',
+                entity_id=user_id,
+                severity='warning',
+                metadata={'score': risk.score, 'decision': risk.decision, 'reasons': risk.reasons},
+            ))
+            raise ForbiddenException(message='KYC verification is required before creating payment orders')
+
         client_payment_id = str(ULID())
 
         if payment_data.total_price < MIN_TOTAL_RUB:
@@ -67,7 +90,29 @@ class CreateOrderCommand:
         )
 
         saved = await self._unit_of_work.order_repository.create(order)
-        with_itpay = await self._itpay_service.create_payment(saved,self._logger.get_trace_id())
+        await self._unit_of_work.risk_repository.create_assessment(RiskAssessmentEntity(
+            user_id=user_id,
+            order_id=saved.id,
+            subject_type=user.account_type,
+            score=risk.score,
+            decision=risk.decision,
+            reasons=risk.reasons,
+        ))
+        await self._unit_of_work.risk_repository.create_audit_event(AuditEventEntity(
+            actor_type='user',
+            actor_id=user_id,
+            action='order_risk_assessed',
+            entity_type='order',
+            entity_id=saved.id,
+            severity='warning' if risk.decision != 'allow' else 'info',
+            metadata={'score': risk.score, 'decision': risk.decision, 'reasons': risk.reasons},
+        ))
+        if risk.decision in ('manual_review', 'reject'):
+            saved.status = OrderStatus.REJECTED
+            await self._unit_of_work.order_repository.update_status(order_id=saved.id, status=OrderStatus.REJECTED)
+            return saved
+
+        with_itpay = await self._itpay_service.create_payment(saved, self._logger.get_trace_id())
         if with_itpay.status in (
             OrderStatus.CANCELLED,
             OrderStatus.REJECTED,
@@ -77,4 +122,3 @@ class CreateOrderCommand:
         else:
             await self._unit_of_work.order_repository.update_after_itpay_payment_created(with_itpay)
         return with_itpay
-

src/application/domain/entities/risk.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass(slots=True)
+class RiskAssessmentEntity:
+    id: str | None = None
+    user_id: str | None = None
+    order_id: str | None = None
+    subject_type: str = "individual"
+    score: int = 0
+    decision: str = "allow"
+    reasons: list[str] = field(default_factory=list)
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+@dataclass(slots=True)
+class AuditEventEntity:
+    id: str | None = None
+    actor_type: str = "system"
+    actor_id: str | None = None
+    action: str = ""
+    entity_type: str = ""
+    entity_id: str | None = None
+    severity: str = "info"
+    metadata: dict | None = None
+    created_at: datetime | None = None

src/application/services/__init__.py

@@ -1,4 +1,5 @@
 from src.application.services.payment_quote_service import PaymentQuote,PaymentQuoteService
+from src.application.services.risk_scoring import RiskAssessmentResult,RiskScoringService
 from src.application.services.sbp_withdrawal_quote_service import SbpWithdrawalQuote,SbpWithdrawalQuoteService
 
 

src/application/services/risk_scoring.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from decimal import Decimal
+
+from src.application.domain.entities.user import UserEntity
+
+
+@dataclass(slots=True)
+class RiskAssessmentResult:
+    score: int
+    decision: str
+    reasons: list[str]
+
+
+class RiskScoringService:
+    MANUAL_REVIEW_THRESHOLD = 60
+    REJECT_THRESHOLD = 80
+
+    def assess_order(
+        self,
+        *,
+        user: UserEntity,
+        total_price: Decimal,
+        recent_order_count: int,
+    ) -> RiskAssessmentResult:
+        if user.account_type == "legal_entity":
+            return RiskAssessmentResult(score=0, decision="skip", reasons=[])
+
+        score = 0
+        reasons: list[str] = []
+
+        if not user.kyc_verified:
+            score += 100
+            reasons.append("kyc_not_completed")
+
+        created_at = user.created_at
+        if created_at is not None:
+            if created_at.tzinfo is None:
+                created_at = created_at.replace(tzinfo=timezone.utc)
+            age = datetime.now(timezone.utc) - created_at
+            if age.total_seconds() < 24 * 60 * 60:
+                score += 20
+                reasons.append("new_account")
+
+        if total_price >= Decimal("200000.00"):
+            score += 30
+            reasons.append("large_amount")
+        elif total_price >= Decimal("50000.00"):
+            score += 15
+            reasons.append("elevated_amount")
+
+        if recent_order_count >= 5:
+            score += 20
+            reasons.append("velocity_spike")
+
+        score = min(score, 100)
+        if score >= self.REJECT_THRESHOLD:
+            decision = "reject"
+        elif score >= self.MANUAL_REVIEW_THRESHOLD:
+            decision = "manual_review"
+        else:
+            decision = "allow"
+        return RiskAssessmentResult(score=score, decision=decision, reasons=reasons)

src/infrastructure/database/models/__init__.py

@@ -3,6 +3,7 @@ from src.infrastructure.database.models.order import Order
 from src.infrastructure.database.models.payment import Payment
 from src.infrastructure.database.models.sbp_withdrawal import SbpWithdrawal
 from src.infrastructure.database.models.user import UserModel
+from src.infrastructure.database.models.risk import AuditEventModel,RiskAssessmentModel
 
 
-__all__ = ['Base','Order','Payment','SbpWithdrawal','UserModel']
+__all__ = ['Base','Order','Payment','SbpWithdrawal','UserModel','RiskAssessmentModel','AuditEventModel']

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/order_repository.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 from dataclasses import replace
 from datetime import datetime,timezone
 from decimal import Decimal
-from sqlalchemy import desc,select,update
+from sqlalchemy import desc,func,select,update
 from sqlalchemy.ext.asyncio import AsyncSession
 from src.application.abstractions.repositories.i_order_repository import IOrderRepository
 from src.application.contracts import ILogger
@@ -109,6 +109,12 @@ class OrderRepository(IOrderRepository):
         return [self._to_entity(model) for model in result.all()]
 
 
+    async def count_recent_by_user(self,*,user_id: str,since: datetime) -> int:
+        stmt = select(func.count()).select_from(Order).where(Order.user_id == user_id, Order.created_at >= since)
+        result = await self._session.execute(stmt)
+        return int(result.scalar_one())
+
+
     async def update_after_itpay_payment_created(self,order: OrderEntity) -> OrderEntity:
         if not order.id:
             raise ValueError('OrderEntity.id is required')

src/infrastructure/database/repositories/risk_repository.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+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, RiskAssessmentEntity
+from src.infrastructure.database.models.risk import AuditEventModel, RiskAssessmentModel
+
+
+class RiskRepository(IRiskRepository):
+    def __init__(self, session: AsyncSession, logger: ILogger):
+        self._session = session
+        self._logger = logger
+
+    async def create_assessment(self, assessment: RiskAssessmentEntity) -> RiskAssessmentEntity:
+        model = RiskAssessmentModel(
+            user_id=assessment.user_id,
+            order_id=assessment.order_id,
+            subject_type=assessment.subject_type,
+            score=assessment.score,
+            decision=assessment.decision,
+            reasons=assessment.reasons,
+        )
+        self._session.add(model)
+        await self._session.flush()
+        return replace(assessment, id=model.id, created_at=model.created_at, updated_at=model.updated_at)
+
+    async def create_audit_event(self, event: AuditEventEntity) -> AuditEventEntity:
+        model = AuditEventModel(
+            actor_type=event.actor_type,
+            actor_id=event.actor_id,
+            action=event.action,
+            entity_type=event.entity_type,
+            entity_id=event.entity_id,
+            severity=event.severity,
+            metadata_json=event.metadata or {},
+        )
+        self._session.add(model)
+        await self._session.flush()
+        return replace(event, id=model.id, created_at=model.created_at)

src/infrastructure/database/unit_of_work.py

@@ -1,12 +1,12 @@
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from src.application.abstractions import IUnitOfWork
-from src.application.abstractions.repositories import IOrderRepository,IPaymentRepository,ISbpWithdrawalRepository,IUserRepository
+from src.application.abstractions.repositories import IOrderRepository, IPaymentRepository, ISbpWithdrawalRepository, IUserRepository, IRiskRepository
 from src.application.contracts import ILogger
 from src.infrastructure.database.repositories.order_repository import OrderRepository
 from src.infrastructure.database.repositories.payment_repository import PaymentRepository
 from src.infrastructure.database.repositories.sbp_withdrawal_repository import SbpWithdrawalRepository
 from src.infrastructure.database.repositories.user_repository import UserRepository
-
+from src.infrastructure.database.repositories.risk_repository import RiskRepository
 
 
 class UnitOfWork(IUnitOfWork):
@@ -17,6 +17,7 @@ class UnitOfWork(IUnitOfWork):
         self._payment_repository: IPaymentRepository | None = None
         self._sbp_withdrawal_repository: ISbpWithdrawalRepository | None = None
         self._user_repository: IUserRepository | None = None
+        self._risk_repository: IRiskRepository | None = None
         self._logger: ILogger = logger
 
     async def __aenter__(self):
@@ -25,6 +26,7 @@ class UnitOfWork(IUnitOfWork):
         self._payment_repository = None
         self._sbp_withdrawal_repository = None
         self._user_repository = None
+        self._risk_repository = None
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
@@ -44,23 +46,26 @@ class UnitOfWork(IUnitOfWork):
             self._order_repository = OrderRepository(session=self._session, logger=self._logger)
         return self._order_repository
 
-
     @property
     def payment_repository(self) -> IPaymentRepository:
         if self._payment_repository is None:
             self._payment_repository = PaymentRepository(session=self._session, logger=self._logger)
         return self._payment_repository
 
-
     @property
     def sbp_withdrawal_repository(self) -> ISbpWithdrawalRepository:
         if self._sbp_withdrawal_repository is None:
-            self._sbp_withdrawal_repository = SbpWithdrawalRepository(session=self._session,logger=self._logger)
+            self._sbp_withdrawal_repository = SbpWithdrawalRepository(session=self._session, logger=self._logger)
         return self._sbp_withdrawal_repository
 
-
     @property
     def user_repository(self) -> IUserRepository:
         if self._user_repository is None:
             self._user_repository = UserRepository(session=self._session, logger=self._logger)
         return self._user_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/infrastructure/metrics.py

@@ -9,12 +9,12 @@ from typing import Any
 
 from fastapi import Request, Response
 from prometheus_client import CollectorRegistry, Counter, Histogram, CONTENT_TYPE_LATEST, generate_latest, multiprocess
-from sqlalchemy import func, select
+from sqlalchemy import func, select, text
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 from starlette.responses import Response as StarletteResponse
 
 from src.infrastructure.database.context import async_session_maker
-from src.infrastructure.database.models import Order, Payment
+from src.infrastructure.database.models import Order, Payment, RiskAssessmentModel, UserModel
 
 SERVICE_NAME = "payment"
 BUSINESS_CACHE_TTL_SECONDS = 30
@@ -102,6 +102,15 @@ async def _collect_business_metrics() -> list[str]:
         "# TYPE payments_created_total gauge",
         "# TYPE payments_succeeded_total gauge",
         "# TYPE payments_failed_total gauge",
+        "# TYPE risk_assessments_total gauge",
+        "# TYPE risk_assessments_by_decision_total gauge",
+        "# TYPE risk_score_average gauge",
+        "# TYPE risk_high_risk_total gauge",
+        "# TYPE product_funnel_registrations_total gauge",
+        "# TYPE product_funnel_kyc_started_total gauge",
+        "# TYPE product_funnel_kyc_completed_total gauge",
+        "# TYPE product_funnel_first_payment_total gauge",
+        "# TYPE product_funnel_successful_operations_total gauge",
     ]
     async with async_session_maker() as session:
         for model, metric_prefix in ((Order, "payment_orders"), (Payment, "payments")):
@@ -119,6 +128,27 @@ async def _collect_business_metrics() -> list[str]:
         failed_total = await _scalar(session, select(func.count()).select_from(Payment).where(Payment.status.in_(["web3_hash_error", "web3_balance_problem", "receipt_error"])))
         lines.append(_sample("payments_succeeded_total", paid_total, {"service": SERVICE_NAME}))
         lines.append(_sample("payments_failed_total", failed_total, {"service": SERVICE_NAME}))
+
+        risk_total = await _scalar(session, select(func.count()).select_from(RiskAssessmentModel))
+        avg_score_result = await session.execute(select(func.coalesce(func.avg(RiskAssessmentModel.score), 0)))
+        avg_score = float(avg_score_result.scalar_one() or 0)
+        high_risk = await _scalar(session, select(func.count()).select_from(RiskAssessmentModel).where(RiskAssessmentModel.score >= 60))
+        lines.append(_sample("risk_assessments_total", risk_total, {"service": SERVICE_NAME}))
+        lines.append(_sample("risk_score_average", avg_score, {"service": SERVICE_NAME}))
+        lines.append(_sample("risk_high_risk_total", high_risk, {"service": SERVICE_NAME}))
+        for decision, count in await _rows(session, select(RiskAssessmentModel.decision, func.count()).group_by(RiskAssessmentModel.decision)):
+            lines.append(_sample("risk_assessments_by_decision_total", int(count), {"service": SERVICE_NAME, "decision": str(decision)}))
+
+        registrations = await _scalar(session, select(func.count()).select_from(UserModel))
+        kyc_started = await _scalar(session, text("select count(distinct user_id) from kyc"))
+        kyc_completed = await _scalar(session, select(func.count()).select_from(UserModel).where(UserModel.kyc_verified.is_(True)))
+        first_payment = await _scalar(session, select(func.count(func.distinct(Order.user_id))))
+        successful_operations = await _scalar(session, select(func.count()).select_from(Payment).where(Payment.status == "completed"))
+        lines.append(_sample("product_funnel_registrations_total", registrations, {"service": SERVICE_NAME}))
+        lines.append(_sample("product_funnel_kyc_started_total", kyc_started, {"service": SERVICE_NAME}))
+        lines.append(_sample("product_funnel_kyc_completed_total", kyc_completed, {"service": SERVICE_NAME}))
+        lines.append(_sample("product_funnel_first_payment_total", first_payment, {"service": SERVICE_NAME}))
+        lines.append(_sample("product_funnel_successful_operations_total", successful_operations, {"service": SERVICE_NAME}))
     lines.append(_sample("payment_metrics_scrape_success", 1, {"service": SERVICE_NAME}))
     return lines
 

src/presentation/dependencies/commands.py

@@ -4,7 +4,7 @@ from src.application.abstractions import IUnitOfWork
 from src.application.commands import CreateCryptoTransferCompletedCommand,CreateOrderCommand,CreatePaymentCommand,CreateSbpWithdrawalCommand,GetOrderCommand,GetOrderStatusCommand,GetPaymentCommand,GetPaymentConfigCommand,GetPaymentQuoteCommand,GetPaymentQuoteFromRubCommand,GetSbpBanksCommand,GetSbpWithdrawalCommand,HandleSbpWithdrawalWalletEventCommand,ListClientOperationsCommand,ListOrdersCommand,ListPaymentsCommand
 from src.application.contracts import ICache,ILogger,IMozenSbpService,IQueueMessanger,IReceipt
 from src.application.contracts.i_itpay_service import IItPayService
-from src.application.services import PaymentQuoteService,SbpWithdrawalQuoteService
+from src.application.services import PaymentQuoteService,RiskScoringService,SbpWithdrawalQuoteService
 from src.infrastructure.cloud_kassir import ClaudeKassirClient
 from src.infrastructure.cloud_kassir.constants import CLOUD_KASSIR_API_BASE_URL,CLOUD_KASSIR_FAIL_URL,CLOUD_KASSIR_INN,CLOUD_KASSIR_SUCCESS_URL
 from src.infrastructure.config import settings
@@ -44,6 +44,7 @@ def get_create_order_command(
         logger=logger,
         payment_quote_service=payment_quote_service,
         itpay_service=itpay_service,
+        risk_scoring_service=RiskScoringService(),
     )
 
 

tests/test_risk_scoring.py

@@ -0,0 +1,53 @@
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+import unittest
+
+from src.application.domain.entities.user import UserEntity
+from src.application.services.risk_scoring import RiskScoringService
+
+
+class RiskScoringServiceTests(unittest.TestCase):
+    def test_individual_without_kyc_is_rejected_before_payment(self):
+        service = RiskScoringService()
+        user = UserEntity(
+            id="user_1",
+            account_type="individual",
+            kyc_verified=False,
+            created_at=datetime.now(timezone.utc) - timedelta(days=10),
+        )
+
+        assessment = service.assess_order(user=user, total_price=Decimal("1000.00"), recent_order_count=0)
+
+        self.assertEqual(assessment.decision, "reject")
+        self.assertIn("kyc_not_completed", assessment.reasons)
+
+    def test_recent_new_high_value_individual_goes_to_manual_review(self):
+        service = RiskScoringService()
+        user = UserEntity(
+            id="user_1",
+            account_type="individual",
+            kyc_verified=True,
+            created_at=datetime.now(timezone.utc) - timedelta(hours=2),
+        )
+
+        assessment = service.assess_order(user=user, total_price=Decimal("250000.00"), recent_order_count=6)
+
+        self.assertEqual(assessment.decision, "manual_review")
+        self.assertGreaterEqual(assessment.score, 60)
+        self.assertIn("new_account", assessment.reasons)
+        self.assertIn("large_amount", assessment.reasons)
+        self.assertIn("velocity_spike", assessment.reasons)
+
+    def test_legal_entity_is_not_scored(self):
+        service = RiskScoringService()
+        user = UserEntity(id="org_1", account_type="legal_entity", kyc_verified=False)
+
+        assessment = service.assess_order(user=user, total_price=Decimal("999999.00"), recent_order_count=99)
+
+        self.assertEqual(assessment.decision, "skip")
+        self.assertEqual(assessment.score, 0)
+        self.assertEqual(assessment.reasons, [])
+
+
+if __name__ == "__main__":
+    unittest.main()