Adjust risk scoring for established users

Dockerfile

@@ -21,8 +21,9 @@ COPY --from=builder /app/src /app/src
 ENV PATH="/app/.venv/bin:$PATH" \
     PYTHONUNBUFFERED=1 \
     PYTHONDONTWRITEBYTECODE=1 \
-    PYTHONPATH=/app
+    PYTHONPATH=/app \
+    PROMETHEUS_MULTIPROC_DIR=/tmp/prometheus_multiproc
 
 EXPOSE 8000
 
-CMD ["sh", "-c", "granian --interface asgi ${APP_MODULE:-src.main:app} --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000} --workers ${APP_WORKERS:-2} --loop uvloop"]
+CMD mkdir -p "$PROMETHEUS_MULTIPROC_DIR" && rm -f "$PROMETHEUS_MULTIPROC_DIR"/* && exec granian --interface asgi ${APP_MODULE:-src.main:app} --host ${APP_HOST:-0.0.0.0} --port ${APP_PORT:-8000} --workers ${APP_WORKERS:-2} --loop uvloop

alter_sbp_withdrawals_drop_wallet_addresses.sql

@@ -1,7 +0,0 @@
-BEGIN;
-
-ALTER TABLE sbp_withdrawals
-    DROP COLUMN IF EXISTS wallet_address,
-    DROP COLUMN IF EXISTS sender_wallet_address;
-
-COMMIT;

docker-compose.yml

@@ -12,10 +12,10 @@ services:
       APP_HOST: "0.0.0.0"
       APP_PORT: "8000"
       APP_WORKERS: "2"
-      KEYDB_REMOTE_HOST: "pay_keydb"
+      KEYDB_HOST: "pay_keydb"
-      KEYDB_REMOTE_PORT: "${KEYDB_PORT:-6380}"
+      KEYDB_PORT: "${KEYDB_PORT:-6380}"
-      KEYDB_REMOTE_PASSWORD: "${REDIS_PASSWORD}"
+      KEYDB_PASSWORD: "${REDIS_PASSWORD}"
-      KEYDB_REMOTE_DB: "${REDIS_DB:-0}"
+      KEYDB_DB: "${REDIS_DB:-0}"
     env_file:
       - .env
     depends_on:

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "hvac==2.4.0",
     "itsdangerous==2.2.0",
     "orjson==3.11.7",
+    "prometheus-client==0.23.1",
     "pydantic-settings==2.12.0",
     "python-jose==3.5.0",
     "python-ulid==3.1.0",

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

@@ -2,3 +2,4 @@ from src.application.abstractions.repositories.i_order_repository import IOrderR
 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_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/commands/payment_read_commands.py

@@ -25,6 +25,15 @@ class ClientOperationResult:
     withdrawal: SbpWithdrawalEntity | None = None
 
 
+def _parse_cursor(cursor: str | None) -> datetime | None:
+    if not cursor:
+        return None
+    parsed = datetime.fromisoformat(cursor.replace('Z','+00:00'))
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
 class GetPaymentConfigCommand:
     def __init__(self, *, payment_quote_service: PaymentQuoteService, logger: ILogger):
         self._payment_quote_service = payment_quote_service
@@ -68,11 +77,12 @@ class ListOrdersCommand:
 
 
     @transactional
-    async def __call__(self, *, user_id: str, limit: int, offset: int) -> list[OrderPaymentResult]:
+    async def __call__(self, *, user_id: str, limit: int, offset: int, cursor: str | None = None) -> list[OrderPaymentResult]:
         orders = await self._unit_of_work.order_repository.list_by_user_id(
             user_id=user_id,
             limit=limit,
             offset=offset,
+            cursor_created_at=_parse_cursor(cursor),
         )
         items: list[OrderPaymentResult] = []
         for order in orders:
@@ -80,7 +90,7 @@ class ListOrdersCommand:
                 continue
             payment = await self._unit_of_work.payment_repository.get_by_order_id(order.id)
             items.append(OrderPaymentResult(order=order,payment=payment))
-        self._logger.info({'event':'orders_list_requested','user_id':user_id,'limit':limit,'offset':offset})
+        self._logger.info({'event':'orders_list_requested','user_id':user_id,'limit':limit,'offset':offset,'cursor':cursor})
         return items
 
 
@@ -108,17 +118,20 @@ class ListClientOperationsCommand:
 
 
     @transactional
-    async def __call__(self, *, user_id: str, limit: int, offset: int) -> list[ClientOperationResult]:
+    async def __call__(self, *, user_id: str, limit: int, offset: int, cursor: str | None = None) -> list[ClientOperationResult]:
-        fetch_limit=limit+offset
+        cursor_created_at=_parse_cursor(cursor)
+        fetch_limit=limit if cursor_created_at is not None else limit+offset
         payments=await self._unit_of_work.payment_repository.list_by_user_id(
             user_id=user_id,
             limit=fetch_limit,
             offset=0,
+            cursor_created_at=cursor_created_at,
         )
         withdrawals=await self._unit_of_work.sbp_withdrawal_repository.list_by_user_id(
             user_id=user_id,
             limit=fetch_limit,
             offset=0,
+            cursor_created_at=cursor_created_at,
         )
         min_dt=datetime.min.replace(tzinfo=timezone.utc)
         items=[
@@ -130,8 +143,8 @@ class ListClientOperationsCommand:
             for withdrawal in withdrawals
         )
         items.sort(key=lambda item:item.created_at,reverse=True)
-        self._logger.info({'event':'client_operations_list_requested','user_id':user_id,'limit':limit,'offset':offset})
+        self._logger.info({'event':'client_operations_list_requested','user_id':user_id,'limit':limit,'offset':offset,'cursor':cursor})
-        return items[offset:offset+limit]
+        return items[:limit] if cursor_created_at is not None else items[offset:offset+limit]
 
 
 class GetPaymentCommand:

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,67 @@
+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")
+
+        is_new_account = False
+        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:
+                is_new_account = True
+                score += 20
+                reasons.append("new_account")
+
+        if is_new_account and total_price >= Decimal("200000.00"):
+            score += 30
+            reasons.append("new_account_large_amount")
+        elif is_new_account and total_price >= Decimal("50000.00"):
+            score += 15
+            reasons.append("new_account_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/cache/client.py

@@ -4,7 +4,7 @@ from src.infrastructure.config import settings
 
 
 def create_redis_client(url:str|None=None) -> Redis:
-    redis_url = url or settings.KEYDB_REMOTE_URL
+    redis_url = url or settings.KEYDB_MARKET_URL
     return redis.from_url(
         redis_url,
         max_connections=50,

src/infrastructure/config/settings.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
-from functools import lru_cache
-from typing import List, Literal
 import os
-from dotenv import load_dotenv, find_dotenv
+from functools import lru_cache
-from pydantic import AliasChoices, Field, field_validator, model_validator
+from typing import List,Literal
-from pydantic_settings import BaseSettings, SettingsConfigDict
+from dotenv import find_dotenv,load_dotenv
+from pydantic import AliasChoices,Field,field_validator,model_validator
+from pydantic_settings import BaseSettings,SettingsConfigDict
 from src.infrastructure.vault import create_hvac_client_from_approle, read_kv2_secret
 
-env_file = find_dotenv(".env")
+env_file = find_dotenv('.env')
 if env_file:
     load_dotenv(env_file)
 
@@ -80,6 +80,10 @@ class Settings(BaseSettings):
     KEYDB_REMOTE_PORT: int | None = None
     KEYDB_REMOTE_PASSWORD: str | None = None
     KEYDB_REMOTE_DB: int | None = None
+    KEYDB_HOST: str | None = Field(default=None,validation_alias=AliasChoices('KEYDB_HOST','KEYDB_CACHE_HOST'))
+    KEYDB_PORT: int | None = Field(default=None,validation_alias=AliasChoices('KEYDB_PORT','KEYDB_CACHE_PORT'))
+    KEYDB_PASSWORD: str | None = Field(default=None,validation_alias=AliasChoices('KEYDB_PASSWORD','KEYDB_CACHE_PASSWORD'))
+    KEYDB_DB: int | None = Field(default=None,validation_alias=AliasChoices('KEYDB_DB','KEYDB_CACHE_DB'))
 
     RABBIT_HOST: str = "localhost"
     RABBIT_PORT: int = 5672
@@ -128,7 +132,7 @@ class Settings(BaseSettings):
             return v
         return normalize_vault_base_url(v)
 
-    @model_validator(mode="before")
+    @model_validator(mode='before')
     @classmethod
     def load_from_vault(cls, data: dict):
         if not isinstance(data, dict):
@@ -284,19 +288,6 @@ class Settings(BaseSettings):
         else:
             data['KEYDB_REMOTE_PASSWORD'] = None
 
-        keydb_host_override = os.getenv('KEYDB_REMOTE_HOST') or data.get('KEYDB_REMOTE_HOST')
-        keydb_port_override = os.getenv('KEYDB_REMOTE_PORT') or data.get('KEYDB_REMOTE_PORT')
-        keydb_db_override = os.getenv('KEYDB_REMOTE_DB') or data.get('KEYDB_REMOTE_DB')
-        keydb_password_override = os.getenv('KEYDB_REMOTE_PASSWORD') or data.get('KEYDB_REMOTE_PASSWORD')
-        if keydb_host_override is not None and str(keydb_host_override).strip():
-            data['KEYDB_REMOTE_HOST'] = str(keydb_host_override).strip()
-        if keydb_port_override is not None and str(keydb_port_override).strip():
-            data['KEYDB_REMOTE_PORT'] = int(keydb_port_override)
-        if keydb_db_override is not None and str(keydb_db_override).strip():
-            data['KEYDB_REMOTE_DB'] = int(keydb_db_override)
-        if keydb_password_override is not None and str(keydb_password_override).strip():
-            data['KEYDB_REMOTE_PASSWORD'] = str(keydb_password_override).strip()
-
         itpay_public_id = data.get('ITPAY_PUBLIC_ID') or os.getenv('ITPAY_PUBLIC_ID')
         itpay_api_secret = data.get('ITPAY_API_SECRET') or os.getenv('ITPAY_API_SECRET')
         if itpay_public_id is not None and str(itpay_public_id).strip() and itpay_api_secret is not None and str(itpay_api_secret).strip():
@@ -348,20 +339,41 @@ class Settings(BaseSettings):
 
     @property
     def REDIS_URL(self) -> str:
-        return self.KEYDB_REMOTE_URL
+        return self.KEYDB_MARKET_URL
 
     @staticmethod
     def _redis_url(*, host: str, port: int, password: str | None, db: int) -> str:
         auth = f':{password}@' if password else ''
         return f'redis://{auth}{host}:{port}/{db}'
 
+    @property
+    def KEYDB_MARKET_URL(self) -> str:
+        if self.KEYDB_REMOTE_HOST is None or self.KEYDB_REMOTE_PORT is None or self.KEYDB_REMOTE_DB is None:
+            raise RuntimeError('Vault KeyDB settings are required for market data')
+        return self._redis_url(
+            host=self.KEYDB_REMOTE_HOST,
+            port=int(self.KEYDB_REMOTE_PORT),
+            password=self.KEYDB_REMOTE_PASSWORD,
+            db=int(self.KEYDB_REMOTE_DB),
+        )
+
     @property
     def KEYDB_REMOTE_URL(self) -> str:
-        host = self.KEYDB_REMOTE_HOST or self.REDIS_HOST
+        return self.KEYDB_MARKET_URL
-        port = int(self.KEYDB_REMOTE_PORT) if self.KEYDB_REMOTE_PORT is not None else int(self.REDIS_PORT)
+
-        password = self.KEYDB_REMOTE_PASSWORD if self.KEYDB_REMOTE_PASSWORD is not None else self.REDIS_PASSWORD
+    @property
-        db = int(self.KEYDB_REMOTE_DB) if self.KEYDB_REMOTE_DB is not None else int(self.REDIS_DB)
+    def KEYDB_CACHE_URL(self) -> str | None:
-        return self._redis_url(host=host, port=port, password=password, db=db)
+        if self.KEYDB_HOST is None or not self.KEYDB_HOST.strip():
+            return None
+        if self.KEYDB_PORT is None:
+            return None
+        db = int(self.KEYDB_DB) if self.KEYDB_DB is not None else 0
+        return self._redis_url(
+            host=self.KEYDB_HOST.strip(),
+            port=int(self.KEYDB_PORT),
+            password=self.KEYDB_PASSWORD,
+            db=db,
+        )
 
     @property
     def RABBIT_URL(self) -> str:
@@ -370,7 +382,7 @@ class Settings(BaseSettings):
 
     @property
     def EXCLUDED_PATHS(self) -> List[str]:
-        return ["/docs", "/redoc", "/openapi.json", "/ping", "/health"]
+        return ["/docs", "/redoc", "/openapi.json", "/ping", "/health", '/healthcheck']
 
 
 @lru_cache(maxsize=1)

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
@@ -94,18 +94,27 @@ class OrderRepository(IOrderRepository):
         return self._to_entity(model)
 
 
-    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[OrderEntity]:
+    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int,cursor_created_at=None) -> list[OrderEntity]:
         stmt=(
             select(Order)
             .where(Order.user_id==user_id)
             .order_by(desc(Order.created_at))
             .limit(limit)
-            .offset(offset)
         )
+        if cursor_created_at is not None:
+            stmt=stmt.where(Order.created_at < cursor_created_at)
+        else:
+            stmt=stmt.offset(offset)
         result=await self._session.scalars(stmt)
         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/payment_repository.py

@@ -38,14 +38,12 @@ class PaymentRepository(IPaymentRepository):
 
 
     async def create_completed(self,*,user_id:str,order_id:str,itpay_payment_id:str,itpay_paid_amount:str|None,transaction_id:str|None,paid_at:str|None,expired_date:str|None) -> bool:
-        stmt=select(Payment).where(Payment.order_id==order_id)
-        existing=await self._session.scalar(stmt)
-        if existing is not None:
-            return False
         paid_at_dt=datetime.fromisoformat(paid_at.replace('Z','+00:00')) if paid_at else None
         expired_dt=datetime.fromisoformat(expired_date.replace('Z','+00:00')) if expired_date else None
         paid_amount_dec=Decimal(str(itpay_paid_amount)) if itpay_paid_amount is not None else None
-        model=Payment(
+        stmt=(
+            insert(Payment)
+            .values(
                 user_id=user_id,
                 order_id=order_id,
                 status=PaymentStatus.MONEY_ACCEPTED,
@@ -57,9 +55,11 @@ class PaymentRepository(IPaymentRepository):
                 paid_at=paid_at_dt,
                 expired_date=expired_dt,
             )
-        self._session.add(model)
+            .on_conflict_do_nothing(index_elements=[Payment.order_id])
+        )
+        result=await self._session.execute(stmt)
         await self._session.flush()
-        return True
+        return bool(result.rowcount)
 
 
     async def update_crypto_transfer_completed(self,*,order_id:str,web3_transaction_hash:str|None) -> None:
@@ -111,14 +111,17 @@ class PaymentRepository(IPaymentRepository):
         return self._to_entity(model)
 
 
-    async def list_by_user_id(self,*,user_id:str,limit:int,offset:int) -> list[PaymentEntity]:
+    async def list_by_user_id(self,*,user_id:str,limit:int,offset:int,cursor_created_at=None) -> list[PaymentEntity]:
         stmt=(
             select(Payment)
             .where(Payment.user_id==user_id)
             .order_by(desc(Payment.created_at))
             .limit(limit)
-            .offset(offset)
         )
+        if cursor_created_at is not None:
+            stmt=stmt.where(Payment.created_at < cursor_created_at)
+        else:
+            stmt=stmt.offset(offset)
         result=await self._session.scalars(stmt)
         return [self._to_entity(model) for model in result.all()]
 

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/repositories/sbp_withdrawal_repository.py

@@ -84,14 +84,17 @@ class SbpWithdrawalRepository(ISbpWithdrawalRepository):
         return self._to_entity(model)
 
 
-    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int) -> list[SbpWithdrawalEntity]:
+    async def list_by_user_id(self,*,user_id: str,limit: int,offset: int,cursor_created_at=None) -> list[SbpWithdrawalEntity]:
         stmt=(
             select(SbpWithdrawal)
             .where(SbpWithdrawal.user_id==user_id)
             .order_by(desc(SbpWithdrawal.created_at))
             .limit(limit)
-            .offset(offset)
         )
+        if cursor_created_at is not None:
+            stmt=stmt.where(SbpWithdrawal.created_at < cursor_created_at)
+        else:
+            stmt=stmt.offset(offset)
         result=await self._session.scalars(stmt)
         return [self._to_entity(model) for model in result.all()]
 

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

@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+from collections.abc import Awaitable, Callable
+from datetime import timedelta
+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, 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, RiskAssessmentModel, UserModel
+
+SERVICE_NAME = "payment"
+BUSINESS_CACHE_TTL_SECONDS = 30
+LATENCY_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
+WINDOWS = (("all", None), ("30d", timedelta(days=30)), ("7d", timedelta(days=7)), ("24h", timedelta(days=1)), ("1h", timedelta(hours=1)))
+
+HTTP_REQUESTS = Counter("payment_http_requests_total", "Total HTTP requests handled by payment service.", ("service", "method", "route", "status_code"))
+HTTP_LATENCY = Histogram("payment_http_request_duration_seconds", "HTTP request duration in seconds for payment service.", ("service", "method", "route"), buckets=LATENCY_BUCKETS)
+
+
+def _route_template(request: Request) -> str:
+    route = request.scope.get("route")
+    path = getattr(route, "path", None)
+    return str(path) if path else "__unmatched__"
+
+
+class PrometheusMetricsMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> StarletteResponse:
+        if request.url.path == "/metrics":
+            return await call_next(request)
+        started = time.perf_counter()
+        status_code = "500"
+        try:
+            response = await call_next(request)
+            status_code = str(response.status_code)
+            return response
+        finally:
+            elapsed = time.perf_counter() - started
+            route = _route_template(request)
+            HTTP_REQUESTS.labels(SERVICE_NAME, request.method, route, status_code).inc()
+            HTTP_LATENCY.labels(SERVICE_NAME, request.method, route).observe(elapsed)
+
+
+class CachedBusinessMetrics:
+    def __init__(self, ttl_seconds: int, collector: Callable[[], Awaitable[list[str]]]) -> None:
+        self._ttl_seconds = ttl_seconds
+        self._collector = collector
+        self._expires_at = 0.0
+        self._lines: list[str] = []
+        self._lock = asyncio.Lock()
+
+    async def get(self) -> list[str]:
+        now = time.monotonic()
+        if now < self._expires_at:
+            return self._lines
+        async with self._lock:
+            now = time.monotonic()
+            if now < self._expires_at:
+                return self._lines
+            self._lines = await self._collector()
+            self._expires_at = now + self._ttl_seconds
+            return self._lines
+
+
+def _labels(labels: dict[str, str]) -> str:
+    escaped = {key: value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") for key, value in labels.items()}
+    return ",".join(f'{key}="{value}"' for key, value in escaped.items())
+
+
+def _sample(name: str, value: int | float, labels: dict[str, str] | None = None) -> str:
+    return f"{name}{{{_labels(labels)}}} {value}" if labels else f"{name} {value}"
+
+
+async def _scalar(session: Any, statement: Any) -> int:
+    result = await session.execute(statement)
+    return int(result.scalar_one() or 0)
+
+
+async def _rows(session: Any, statement: Any) -> list[Any]:
+    result = await session.execute(statement)
+    return list(result.all())
+
+
+def _status_value(status: Any) -> str:
+    return getattr(status, "value", str(status))
+
+
+async def _collect_business_metrics() -> list[str]:
+    lines: list[str] = [
+        "# TYPE payment_orders_total gauge",
+        "# TYPE payment_orders_by_status_total gauge",
+        "# TYPE payment_orders_created_total gauge",
+        "# TYPE payments_total gauge",
+        "# TYPE payments_by_status_total gauge",
+        "# 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")):
+            total = await _scalar(session, select(func.count()).select_from(model))
+            lines.append(_sample(f"{metric_prefix}_total", total, {"service": SERVICE_NAME}))
+            for status, count in await _rows(session, select(model.status, func.count()).group_by(model.status)):
+                lines.append(_sample(f"{metric_prefix}_by_status_total", int(count), {"service": SERVICE_NAME, "status": _status_value(status)}))
+            for window_name, delta in WINDOWS:
+                conditions = []
+                if delta is not None:
+                    conditions.append(model.created_at >= func.now() - delta)
+                created = await _scalar(session, select(func.count()).select_from(model).where(*conditions))
+                lines.append(_sample(f"{metric_prefix}_created_total", created, {"service": SERVICE_NAME, "window": window_name}))
+        paid_total = await _scalar(session, select(func.count()).select_from(Payment).where(Payment.paid_at.is_not(None)))
+        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
+
+
+_business_metrics_cache = CachedBusinessMetrics(BUSINESS_CACHE_TTL_SECONDS, _collect_business_metrics)
+
+
+def _client_metrics() -> bytes:
+    if os.environ.get("PROMETHEUS_MULTIPROC_DIR"):
+        registry = CollectorRegistry()
+        multiprocess.MultiProcessCollector(registry)
+        return generate_latest(registry)
+    return generate_latest()
+
+
+async def metrics_response() -> Response:
+    lines: list[str] = []
+    try:
+        lines.extend(await _business_metrics_cache.get())
+    except Exception:
+        lines.append(_sample("payment_metrics_scrape_success", 0, {"service": SERVICE_NAME}))
+    body = _client_metrics().decode("utf-8") + "\n".join(lines) + "\n"
+    return Response(body, media_type=CONTENT_TYPE_LATEST)

src/main.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 from contextlib import asynccontextmanager
 import secrets
 from typing import AsyncGenerator
-from fastapi import Depends, FastAPI
+from fastapi import Depends, FastAPI, Response
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
 from fastapi.responses import HTMLResponse
@@ -13,6 +13,7 @@ from src.infrastructure.config.settings import get_settings
 from src.infrastructure.vault import JwtKeyStore, start_jwt_keys_scheduler
 from src.infrastructure.utils import generate_instance_id
 from src.infrastructure.logger import logger
+from src.infrastructure.metrics import PrometheusMetricsMiddleware, metrics_response
 from src.infrastructure.config import settings
 from src.presentation.handler import application_exception_handler, unhandled_exception_handler
 from src.presentation.messaging import crypto_transfer_router,sbp_withdrawal_messaging_router
@@ -40,16 +41,33 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     logger.set_instance_id(instance_id)
     logger.info(f'Users service instance started with id {instance_id}')
 
-    app.state.redis_remote = create_redis_client(settings.KEYDB_REMOTE_URL)
+    app.state.redis_market = create_redis_client(settings.KEYDB_MARKET_URL)
-    app.state.redis = app.state.redis_remote
+    app.state.redis_remote = app.state.redis_market
+    app.state.redis_cache = None
+    app.state.redis = None
     try:
-        await app.state.redis_remote.ping()
+        await app.state.redis_market.ping()
-        logger.info('KeyDB connection established')
+        logger.info('Market KeyDB connection established')
     except Exception as exception:
-        logger.error(f'KeyDB connection failed: {exception}')
+        logger.error(f'Market KeyDB connection failed: {exception}')
-        await app.state.redis_remote.aclose()
+        await app.state.redis_market.aclose()
         raise
 
+    cache_url = settings.KEYDB_CACHE_URL
+    if cache_url:
+        app.state.redis_cache = create_redis_client(cache_url)
+        app.state.redis = app.state.redis_cache
+        try:
+            await app.state.redis_cache.ping()
+            logger.info('Cache KeyDB connection established')
+        except Exception as exception:
+            logger.error(f'Cache KeyDB connection failed: {exception}')
+            await app.state.redis_cache.aclose()
+            await app.state.redis_market.aclose()
+            raise
+    else:
+        logger.info('Cache KeyDB is not configured')
+
     jwt_store = JwtKeyStore(
         vault_addr=settings.VAULT_ADDR,
         vault_role_id=settings.VAULT_ROLE_ID,
@@ -72,7 +90,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         sched = getattr(app.state,'jwt_keys_scheduler',None)
         if sched:
             sched.shutdown(wait=False)
-        await app.state.redis_remote.aclose()
+        redis_cache = getattr(app.state,'redis_cache',None)
+        if redis_cache is not None:
+            await redis_cache.aclose()
+        await app.state.redis_market.aclose()
         logger.info(f'Pay service instance ended with id {instance_id}')
 
 
@@ -86,6 +107,8 @@ app: FastAPI = FastAPI(
 app.add_exception_handler(ApplicationException, application_exception_handler)
 app.add_exception_handler(Exception, unhandled_exception_handler)
 
+app.add_middleware(PrometheusMetricsMiddleware)
+
 app.include_router(order_router)
 app.include_router(orders_router)
 app.include_router(payment_router)
@@ -137,6 +160,11 @@ async def custom_redoc_html(_credentials: HTTPBasicCredentials = Depends(verify_
     )
 
 
+@app.get('/metrics', include_in_schema=False)
+async def metrics() -> Response:
+    return await metrics_response()
+
+
 @app.post('/ping')
 async def ping() -> dict[str, str]:
     return {
@@ -144,6 +172,7 @@ async def ping() -> dict[str, str]:
         'status': 'ok',
     }
 
+@app.get('/healthcheck')
 @app.get('/health')
 async def health() -> dict[str, str]:
     return {

src/presentation/dependencies/cache.py

@@ -9,19 +9,22 @@ from src.infrastructure.cache import KeydbCache,RemoteCache
 
 
 def get_redis_remote(request: Request) -> Redis:
-    return request.app.state.redis_remote
+    return request.app.state.redis_market
 
 
 def get_redis(request: Request) -> Redis:
-    return request.app.state.redis_remote
+    redis_cache = getattr(request.app.state,'redis_cache',None)
+    if redis_cache is None:
+        raise RuntimeError('Cache KeyDB is not configured')
+    return redis_cache
 
 
-def get_cache_remote(redis_client: Redis = Depends(get_redis_remote)) -> ICache:
+def get_cache_remote(request: Request) -> ICache:
-    return KeydbCache(redis_client)
+    return KeydbCache(get_redis(request))
 
 
-def get_remote_cache(redis_client: Redis = Depends(get_redis_remote)) -> ICache:
+def get_remote_cache(request: Request) -> ICache:
-    return RemoteCache(redis_client)
+    return RemoteCache(get_redis_remote(request))
 
 
 def get_cache(cache: ICache = Depends(get_cache_remote)) -> ICache:

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

src/presentation/routing/order.py

@@ -10,7 +10,7 @@ from src.application.commands import ClientOperationResult,CreateOrderCommand,Cr
 from src.application.contracts import IJwtService,ILogger
 from src.application.domain.dto import AccessTokenPayload,AuthContext
 from src.application.domain.entities import OrderEntity,PaymentEntity,SbpWithdrawalEntity
-from src.application.domain.enums import OrderStatus
+from src.application.domain.enums import OrderStatus, PaymentStatus
 from src.application.domain.exceptions import ApplicationException,ConflictException
 from src.application.services import PaymentQuote
 from src.infrastructure.context_vars import trace_id_var
@@ -21,6 +21,18 @@ from src.presentation.dependencies.security import get_jwt_service
 from src.presentation.schemas.order import ClientOperationResponse,ClientOperationsResponse,CreateOrder,CreateOrderResponse,ErrorResponse,OrderDetailResponse,OrderPaymentResponse,OrdersResponse,OrderStatusResponse,OrderWithPaymentResponse,PaymentConfigResponse,PaymentDetailResponse,PaymentQuoteResponse,PaymentResponse,SbpWithdrawalOperationResponse
 from src.presentation.schemas.itpay_payment_models import ItpayPaymentData
 
+ORDER_EVENTS_CACHE_TTL_SECONDS = 10
+ORDER_EVENTS_TERMINAL_CACHE_TTL_SECONDS = 300
+ORDER_EVENTS_IDLE_SLEEP_SECONDS = 2
+TERMINAL_ORDER_STATUSES = {OrderStatus.COMPLETED, OrderStatus.CANCELLED, OrderStatus.ERROR, OrderStatus.REJECTED}
+TERMINAL_PAYMENT_STATUSES = {
+    PaymentStatus.COMPLETED,
+    PaymentStatus.USDT_DELIVERED,
+    PaymentStatus.WEB3_HASH_ERROR,
+    PaymentStatus.WEB3_BALANCE_PROBLEM,
+    PaymentStatus.RECEIPT_ERROR,
+}
+
 order_router = APIRouter(prefix='/order', tags=['orders'])
 orders_router = APIRouter(tags=['orders'])
 payment_router = APIRouter(prefix='/payment', tags=['payments'])
@@ -170,6 +182,23 @@ def _order_status_response(order: OrderEntity, payment: PaymentEntity | None) ->
     )
 
 
+def _is_terminal_order_status(order: OrderEntity, payment: PaymentEntity | None) -> bool:
+    if order.status in TERMINAL_ORDER_STATUSES:
+        return True
+    return bool(payment and payment.status in TERMINAL_PAYMENT_STATUSES)
+
+
+def _is_terminal_status_payload(payload: dict) -> bool:
+    data = payload.get('data') if isinstance(payload, dict) else None
+    if not isinstance(data, dict):
+        return False
+    order_status = data.get('order_status')
+    payment_status = data.get('payment_status')
+    terminal_order_values = {status.value for status in TERMINAL_ORDER_STATUSES}
+    terminal_payment_values = {status.value for status in TERMINAL_PAYMENT_STATUSES}
+    return order_status in terminal_order_values or payment_status in terminal_payment_values
+
+
 def _extract_websocket_access_token(websocket: WebSocket) -> str | None:
     token = websocket.cookies.get('access_token')
     if token:
@@ -289,12 +318,14 @@ async def payment_list_orders(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListOrdersCommand = Depends(get_list_orders_command),
 ) -> OrdersResponse:
-    orders = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    orders = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     items = [_order_with_payment_response(item.order,item.payment) for item in orders]
-    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset)
+    next_cursor = items[-1].order.created_at if items and items[-1].order.created_at else None
+    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset,next_cursor=next_cursor)
 
 
 @payment_router.get(
@@ -308,15 +339,17 @@ async def payment_list_payments(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListClientOperationsCommand = Depends(get_list_client_operations_command),
 ) -> ClientOperationsResponse:
-    operations = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    operations = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     return ClientOperationsResponse(
         status_code=200,
         operations=[_client_operation_response(operation) for operation in operations],
         limit=limit,
         offset=offset,
+        next_cursor=operations[-1].created_at.isoformat() if operations else None,
     )
 
 
@@ -348,12 +381,14 @@ async def list_orders(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListOrdersCommand = Depends(get_list_orders_command),
 ) -> OrdersResponse:
-    orders = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    orders = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     items = [_order_with_payment_response(item.order,item.payment) for item in orders]
-    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset)
+    next_cursor = items[-1].order.created_at if items and items[-1].order.created_at else None
+    return OrdersResponse(status_code=200,orders=items,limit=limit,offset=offset,next_cursor=next_cursor)
 
 
 @payments_router.get(
@@ -367,15 +402,17 @@ async def list_payments(
         request: Request,
         limit: int = Query(default=20, ge=1, le=100),
         offset: int = Query(default=0, ge=0),
+        cursor: str | None = Query(default=None),
         auth: AuthContext = Depends(require_access_token),
         command: ListClientOperationsCommand = Depends(get_list_client_operations_command),
 ) -> ClientOperationsResponse:
-    operations = await command(user_id=auth.user_id,limit=limit,offset=offset)
+    operations = await command(user_id=auth.user_id,limit=limit,offset=offset,cursor=cursor)
     return ClientOperationsResponse(
         status_code=200,
         operations=[_client_operation_response(operation) for operation in operations],
         limit=limit,
         offset=offset,
+        next_cursor=operations[-1].created_at.isoformat() if operations else None,
     )
 
 
@@ -413,8 +450,19 @@ async def order_events(
             return
         await websocket.accept()
         logger.info({'event':'order_events_connected','order_id':order_id,'user_id':auth.user_id})
+        redis_cache = getattr(websocket.app.state, 'redis_cache', None)
+        cache_key = f'order_status:{auth.user_id}:{order_id}'
         last_payload: dict | None = None
         while True:
+            payload: dict | None = None
+            if redis_cache is not None:
+                try:
+                    cached = await redis_cache.get(cache_key)
+                    if cached:
+                        payload = orjson.loads(cached)
+                except Exception as exception:
+                    logger.warning({'event':'order_events_cache_read_failed','order_id':order_id,'error':str(exception)})
+            if payload is None:
                 try:
                     result = await command(order_id=order_id,user_id=auth.user_id)
                 except ApplicationException as exception:
@@ -423,10 +471,19 @@ async def order_events(
                     return
                 status_payload = _order_status_response(result.order,result.payment).model_dump(mode='json')
                 payload = {'event':'order_status','data':status_payload}
+                if redis_cache is not None:
+                    ttl = ORDER_EVENTS_TERMINAL_CACHE_TTL_SECONDS if _is_terminal_order_status(result.order,result.payment) else ORDER_EVENTS_CACHE_TTL_SECONDS
+                    try:
+                        await redis_cache.set(cache_key,orjson.dumps(payload).decode(),ex=ttl)
+                    except Exception as exception:
+                        logger.warning({'event':'order_events_cache_write_failed','order_id':order_id,'error':str(exception)})
             if payload != last_payload:
                 await websocket.send_text(orjson.dumps(payload).decode())
                 last_payload = payload
-            await asyncio.sleep(2)
+            if _is_terminal_status_payload(payload):
+                await websocket.close(code=1000)
+                return
+            await asyncio.sleep(ORDER_EVENTS_IDLE_SLEEP_SECONDS)
     except WebSocketDisconnect:
         logger.info({'event':'order_events_disconnected','order_id':order_id})
     finally:

src/presentation/schemas/order.py

@@ -120,6 +120,7 @@ class OrdersResponse(BaseModel):
     orders: list[OrderWithPaymentResponse]
     limit: int
     offset: int
+    next_cursor: str | None = None
 
 
 class PaymentsResponse(BaseModel):
@@ -134,6 +135,7 @@ class ClientOperationsResponse(BaseModel):
     operations: list[ClientOperationResponse]
     limit: int
     offset: int
+    next_cursor: str | None = None
 
 
 class PaymentDetailResponse(BaseModel):

tests/test_risk_scoring.py

@@ -0,0 +1,69 @@
+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("new_account_large_amount", assessment.reasons)
+        self.assertIn("velocity_spike", assessment.reasons)
+
+
+    def test_old_verified_individual_can_pay_large_amount_without_risk_penalty(self):
+        service = RiskScoringService()
+        user = UserEntity(
+            id="user_1",
+            account_type="individual",
+            kyc_verified=True,
+            created_at=datetime.now(timezone.utc) - timedelta(days=90),
+        )
+
+        assessment = service.assess_order(user=user, total_price=Decimal("999999.00"), recent_order_count=0)
+
+        self.assertEqual(assessment.decision, "allow")
+        self.assertEqual(assessment.score, 0)
+        self.assertEqual(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()

uv.lock

@@ -398,9 +398,7 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/65/8b/3669ad3b3f247a791b2b4aceb3aa5a31f5f6817bf547e4e1ff712338145a/greenlet-3.4.0-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:1a54a921561dd9518d31d2d3db4d7f80e589083063ab4d3e2e950756ef809e1a", size = 286902, upload-time = "2026-04-08T15:52:12.138Z" },
     { url = "https://files.pythonhosted.org/packages/38/3e/3c0e19b82900873e2d8469b590a6c4b3dfd2b316d0591f1c26b38a4879a5/greenlet-3.4.0-cp312-cp312-manylinux_2_24_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:16dec271460a9a2b154e3b1c2fa1050ce6280878430320e85e08c166772e3f97", size = 606099, upload-time = "2026-04-08T16:24:38.408Z" },
     { url = "https://files.pythonhosted.org/packages/b5/33/99fef65e7754fc76a4ed14794074c38c9ed3394a5bd129d7f61b705f3168/greenlet-3.4.0-cp312-cp312-manylinux_2_24_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:90036ce224ed6fe75508c1907a77e4540176dcf0744473627785dd519c6f9996", size = 618837, upload-time = "2026-04-08T16:30:58.298Z" },
-    { url = "https://files.pythonhosted.org/packages/44/57/eae2cac10421feae6c0987e3dc106c6d86262b1cb379e171b017aba893a6/greenlet-3.4.0-cp312-cp312-manylinux_2_24_s390x.manylinux_2_28_s390x.whl", hash = "sha256:6f0def07ec9a71d72315cf26c061aceee53b306c36ed38c35caba952ea1b319d", size = 624901, upload-time = "2026-04-08T16:40:38.981Z" },
     { url = "https://files.pythonhosted.org/packages/36/f7/229f3aed6948faa20e0616a0b8568da22e365ede6a54d7d369058b128afd/greenlet-3.4.0-cp312-cp312-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a1c4f6b453006efb8310affb2d132832e9bbb4fc01ce6df6b70d810d38f1f6dc", size = 615062, upload-time = "2026-04-08T15:56:33.766Z" },
-    { url = "https://files.pythonhosted.org/packages/6a/8a/0e73c9b94f31d1cc257fe79a0eff621674141cdae7d6d00f40de378a1e42/greenlet-3.4.0-cp312-cp312-manylinux_2_39_riscv64.whl", hash = "sha256:0e1254cf0cbaa17b04320c3a78575f29f3c161ef38f59c977108f19ffddaf077", size = 423927, upload-time = "2026-04-08T16:43:05.293Z" },
     { url = "https://files.pythonhosted.org/packages/08/97/d988180011aa40135c46cd0d0cf01dd97f7162bae14139b4a3ef54889ba5/greenlet-3.4.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:9b2d9a138ffa0e306d0e2b72976d2fb10b97e690d40ab36a472acaab0838e2de", size = 1573511, upload-time = "2026-04-08T16:26:20.058Z" },
     { url = "https://files.pythonhosted.org/packages/d4/0f/a5a26fe152fb3d12e6a474181f6e9848283504d0afd095f353d85726374b/greenlet-3.4.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:8424683caf46eb0eb6f626cb95e008e8cc30d0cb675bdfa48200925c79b38a08", size = 1640396, upload-time = "2026-04-08T15:57:30.88Z" },
     { url = "https://files.pythonhosted.org/packages/42/cf/bb2c32d9a100e36ee9f6e38fad6b1e082b8184010cb06259b49e1266ca01/greenlet-3.4.0-cp312-cp312-win_amd64.whl", hash = "sha256:a0a53fb071531d003b075c444014ff8f8b1a9898d36bb88abd9ac7b3524648a2", size = 238892, upload-time = "2026-04-08T17:03:10.094Z" },
@@ -513,6 +511,7 @@ dependencies = [
     { name = "hvac" },
     { name = "itsdangerous" },
     { name = "orjson" },
+    { name = "prometheus-client" },
     { name = "pydantic-settings" },
     { name = "python-jose" },
     { name = "python-ulid" },
@@ -535,6 +534,7 @@ requires-dist = [
     { name = "hvac", specifier = "==2.4.0" },
     { name = "itsdangerous", specifier = "==2.2.0" },
     { name = "orjson", specifier = "==3.11.7" },
+    { name = "prometheus-client", specifier = "==0.23.1" },
     { name = "pydantic-settings", specifier = "==2.12.0" },
     { name = "python-jose", specifier = "==3.5.0" },
     { name = "python-ulid", specifier = "==3.1.0" },
@@ -543,6 +543,15 @@ requires-dist = [
     { name = "uvloop", marker = "sys_platform != 'win32'", specifier = "==0.22.1" },
 ]
 
+[[package]]
+name = "prometheus-client"
+version = "0.23.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/23/53/3edb5d68ecf6b38fcbcc1ad28391117d2a322d9a1a3eff04bfdb184d8c3b/prometheus_client-0.23.1.tar.gz", hash = "sha256:6ae8f9081eaaaf153a2e959d2e6c4f4fb57b12ef76c8c7980202f1e57b48b2ce", size = 80481, upload-time = "2025-09-18T20:47:25.043Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b8/db/14bafcb4af2139e046d03fd00dea7873e48eafe18b7d2797e73d6681f210/prometheus_client-0.23.1-py3-none-any.whl", hash = "sha256:dd1913e6e76b59cfe44e7a4b83e01afc9873c1bdfd2ed8739f1e76aeca115f99", size = 61145, upload-time = "2025-09-18T20:47:23.875Z" },
+]
+
 [[package]]
 name = "propcache"
 version = "0.4.1"