4730c37915
- Created FastAPI application with transaction ingestion endpoints - Built dynamic rule engine supporting velocity checks and aggregations - Implemented real-time and batch screening capabilities - Added rule management with versioning and rollback functionality - Created comprehensive audit and reporting endpoints with pagination - Set up SQLite database with proper migrations using Alembic - Added intelligent caching for aggregate computations - Included extensive API documentation and example rule definitions - Configured CORS, health endpoints, and proper error handling - Added support for time-windowed aggregations (sum, count, avg, max, min) - Built background processing for high-volume batch screening - Implemented field-agnostic rule conditions with flexible operators Features include transaction ingestion, rule CRUD operations, real-time screening, batch processing, aggregation computations, and comprehensive reporting capabilities suitable for fintech fraud monitoring systems.
24 lines
1.2 KiB
Python
24 lines
1.2 KiB
Python
from sqlalchemy import Column, Integer, String, Float, DateTime, Text
|
|
from sqlalchemy.sql import func
|
|
from app.db.base import Base
|
|
|
|
class Transaction(Base):
|
|
__tablename__ = "transactions"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
transaction_id = Column(String, unique=True, index=True, nullable=False)
|
|
user_id = Column(String, index=True, nullable=False)
|
|
account_id = Column(String, index=True, nullable=False)
|
|
amount = Column(Float, nullable=False)
|
|
currency = Column(String, default="NGN")
|
|
transaction_type = Column(String, nullable=False) # debit, credit, transfer
|
|
merchant_id = Column(String, index=True, nullable=True)
|
|
merchant_category = Column(String, nullable=True)
|
|
channel = Column(String, nullable=False) # web, mobile, atm, pos
|
|
location = Column(String, nullable=True)
|
|
ip_address = Column(String, nullable=True)
|
|
device_id = Column(String, index=True, nullable=True)
|
|
status = Column(String, default="pending") # pending, completed, failed
|
|
metadata = Column(Text, nullable=True) # JSON string for additional fields
|
|
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
|
updated_at = Column(DateTime(timezone=True), onupdate=func.now()) |