**Proyecto**: DYGSOM Fraud Scoring API
Sign in to like and favorite skills
Proyecto: DYGSOM Fraud Scoring API
Framework: FastAPI + Python 3.11
Database: PostgreSQL 15 con Prisma
Cache: Redis 7
ML: scikit-learn + XGBoost
Backend (ali1):
ML (Alicia):
anyprint()# ✅ Correcto: snake_case transaction_repository.py fraud_service.py create_transaction_dto.py # ❌ Incorrecto TransactionRepository.py fraudService.py
# ✅ Correcto: PascalCase class TransactionRepository: class FraudService: class CreateTransactionDto:
# ✅ Correcto: snake_case def get_transaction_by_id(): async def score_transaction(): customer_email = "[email protected]"
# ✅ Correcto: UPPER_SNAKE_CASE MAX_TRANSACTION_AMOUNT = 1000000 DEFAULT_FRAUD_THRESHOLD = 0.5 CACHE_TTL_SECONDS = 3600
src/ ├── repositories/ │ ├── __init__.py │ ├── base_repository.py │ └── transaction_repository.py ├── services/ │ ├── __init__.py │ └── fraud_service.py ├── schemas/ │ ├── __init__.py │ └── transaction_schemas.py └── models/ ├── __init__.py └── transaction.py
from typing import Optional, List from prisma import Prisma from src.models.transaction import Transaction class TransactionRepository: def __init__(self, prisma: Prisma): self.prisma = prisma async def find_by_id(self, transaction_id: str) -> Optional[Transaction]: """Find transaction by ID""" result = await self.prisma.transaction.find_unique( where={"transaction_id": transaction_id} ) return Transaction(**result) if result else None async def create(self, data: dict) -> Transaction: """Create new transaction""" result = await self.prisma.transaction.create(data=data) return Transaction(**result) async def find_all(self, skip: int = 0, limit: int = 100) -> List[Transaction]: """Find all transactions with pagination""" results = await self.prisma.transaction.find_many( skip=skip, take=limit, order={"created_at": "desc"} ) return [Transaction(**r) for r in results]
from typing import Dict import logging from src.repositories.transaction_repository import TransactionRepository logger = logging.getLogger(__name__) class FraudService: def __init__(self, transaction_repo: TransactionRepository): self.transaction_repo = transaction_repo async def score_transaction(self, data: dict) -> Dict: """Score transaction for fraud risk""" logger.info(f"Scoring transaction: {data.get('transaction_id')}") # TODO: Implement fraud scoring logic return { "fraud_score": 0.0, "risk_level": "LOW", "recommendation": "APPROVE" }
from pydantic import BaseModel, Field, validator from typing import Optional from datetime import datetime class CustomerData(BaseModel): id: Optional[str] = None email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$') phone: Optional[str] = None ip_address: str class CreateTransactionDto(BaseModel): transaction_id: str amount: float = Field(..., gt=0, le=1000000) currency: str = Field(default="PEN") timestamp: datetime customer: CustomerData @validator('amount') def amount_must_be_positive(cls, v): if v <= 0: raise ValueError('Amount must be positive') return v
model Transaction { id String @id @default(uuid()) transaction_id String @unique amount Decimal @db.Decimal(12, 2) currency String @default("PEN") timestamp DateTime @default(now()) // Customer customer_id String? customer_email String? customer_ip String? // Payment card_bin String? card_last4 String? card_brand String? // Fraud Detection fraud_score Decimal? @db.Decimal(5, 4) risk_level String? decision String? created_at DateTime @default(now()) @@index([customer_email]) @@index([fraud_score]) }
Archivos implementados:
src/repositories/base_repository.pysrc/repositories/transaction_repository.pyArchivos implementados:
src/services/fraud_service.pyArchivos implementados:
src/schemas/transaction_schemas.pyArchivos implementados:
src/scripts/seed_transactions.pyValidaciones completadas:
validations/verify_day2.pyvalidations/CODE_QUALITY_REPORT.mdAlta: Repository + Service layer ✅
Alta: Seed script funcional ✅
Media: DTOs completos ✅
Baja: Tests (viene en Día 5) ⏳
Backend (ali1): 100% COMPLETADO ✅
Próximo paso: Día 3 - FastAPI Endpoints + ML Integration
dygsom-fraud-api/ ├── src/ │ ├── repositories/ ✅ COMPLETADO │ │ ├── base_repository.py │ │ └── transaction_repository.py │ ├── services/ ✅ COMPLETADO │ │ └── fraud_service.py │ ├── schemas/ ✅ COMPLETADO │ │ └── transaction_schemas.py │ └── scripts/ ✅ COMPLETADO │ └── seed_transactions.py ├── prisma/ │ └── schema.prisma ✅ COMPLETADO (4 tablas) ├── validations/ ✅ NUEVA CARPETA │ ├── verify_day2.py │ ├── CODE_QUALITY_REPORT.md │ └── README.md ├── requirements.txt ✅ ACTUALIZADO (black, flake8) └── .copilot-instructions.md ✅ ACTUALIZADO