"""SQLAlchemy 2.0 typed ORM models for the fraud-detection persistence schema. Seven tables: - `users` -- login accounts (officer or client role), bcrypt password hashes. - `sessions` -- server-side session tokens (auth-04): opaque token -> user_id + expires_at, checked on every authenticated request. Chosen over JWT so access can be instantly revoked (delete the row) rather than waiting out a token's expiry. - `clients` -- a bank customer's account: banking fields (starting_balance, account_type) plus a 1:1 FK to the `users` row that logs into it, and `created_by_officer_id` recording which officer created the account. A client's *name* lives on `users.name`, not duplicated here. - `transactions` -- every scored transaction, both officer-submitted (via /predict, /transactions/batch -- backfill/ testing, `client_id` NULL) and client-submitted (via /client/transactions, `client_id` set). One unified table so the fraud alert queue is a single source of truth: `/officer/alerts` sees everything, `/client/alerts` is the same query filtered to the caller's own `client_id`. - `fraud_predictions` -- one row per scored transaction: probability, risk tier, model version, and a structured top-SHAP- features JSON column (a list of `{"feature": str, "shap_value": float}` objects, ordered by descending absolute SHAP value) so later phases can query/filter by feature name instead of storing an opaque blob. - `fraud_alerts` -- one row per prediction whose probability crosses FRAUD_ALERT_THRESHOLD; carries `disposition` (open/reviewed/dismissed/rejected) and a nullable `reviewed_at` so a future analyst-labeling / retraining pipeline can consume reviewed alerts without a schema rewrite (DB-04). Rejecting an alert also flips `transactions.reversed` on the underlying transaction, excluding it from balance calculations -- the officer's way of saying "this was fraud, undo it." `isFlaggedFraud` from the raw PaySim dataset is intentionally NOT modeled as a feature or a label anywhere in this schema. It has near-zero recall against true `isFraud` (PITFALLS.md Pitfall 2) and must never be treated as a usable signal -- it is simply absent from `Transaction` below, not even retained as a reference column, to remove any risk of it re-entering a feature matrix via a wildcard column selection later. """ from __future__ import annotations import enum from datetime import datetime from decimal import Decimal from sqlalchemy import ( JSON, Boolean, DateTime, ForeignKey, Integer, Numeric, String, ) from sqlalchemy import Enum as SAEnum from sqlalchemy import func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): """Shared declarative base for every ORM model in this schema.""" class AlertDisposition(str, enum.Enum): """Analyst-facing lifecycle state for a `fraud_alerts` row (DB-04). Structured now, ahead of the analyst-labeling UI, so a future retraining pipeline can select on `disposition` / `reviewed_at` without a schema rewrite. """ OPEN = "open" REVIEWED = "reviewed" DISMISSED = "dismissed" REJECTED = "rejected" class UserRole(str, enum.Enum): """A user is either a banking officer (back-office) or a client (self-service, scoped to their own account).""" OFFICER = "officer" CLIENT = "client" class User(Base): """A login-capable account. Password is always stored as a bcrypt hash (`app/auth.py` via passlib) -- never plaintext, never reversible. """ __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(100), nullable=False) email: Mapped[str] = mapped_column(String(255), nullable=False, unique=True, index=True) password_hash: Mapped[str] = mapped_column(String(255), nullable=False) role: Mapped[UserRole] = mapped_column( SAEnum( UserRole, native_enum=True, length=10, name="user_role", values_callable=lambda enum_cls: [member.value for member in enum_cls], ), nullable=False, index=True, ) created_at: Mapped[datetime] = mapped_column( DateTime, server_default=func.now(), nullable=False ) sessions: Mapped[list["Session"]] = relationship( back_populates="user", cascade="all, delete-orphan" ) client_profile: Mapped["Client | None"] = relationship( back_populates="user", foreign_keys="Client.user_id", uselist=False, cascade="all, delete-orphan", ) class Session(Base): """A server-side session token (opaque, random -- not a JWT), checked on every authenticated request via a DB lookup. Deleting the row is an instant, unconditional revocation (used by logout, and available for a future "disable this client" officer action) -- the tradeoff a JWT wouldn't give without an extra denylist table. """ __tablename__ = "sessions" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) token: Mapped[str] = mapped_column(String(64), nullable=False, unique=True, index=True) user_id: Mapped[int] = mapped_column( ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True ) created_at: Mapped[datetime] = mapped_column( DateTime, server_default=func.now(), nullable=False ) expires_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, index=True) user: Mapped["User"] = relationship(back_populates="sessions") class Client(Base): """A bank customer's account -- banking fields only; identity (name, email, password) lives on the linked `users` row (`user_id`, 1:1). `created_by_officer_id` records which officer created the account, for audit purposes. """ __tablename__ = "clients" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column( ForeignKey("users.id", ondelete="CASCADE"), nullable=False, unique=True, index=True ) created_by_officer_id: Mapped[int] = mapped_column( ForeignKey("users.id", ondelete="RESTRICT"), nullable=False ) starting_balance: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False) account_type: Mapped[str] = mapped_column(String(30), nullable=False) created_at: Mapped[datetime] = mapped_column( DateTime, server_default=func.now(), nullable=False ) user: Mapped["User"] = relationship(back_populates="client_profile", foreign_keys=[user_id]) created_by: Mapped["User"] = relationship(foreign_keys=[created_by_officer_id]) transactions: Mapped[list["Transaction"]] = relationship(back_populates="client") class Transaction(Base): """A transaction scored for fraud risk -- both officer-submitted (`/predict`, `/transactions/batch`; `client_id` NULL) and client-submitted (`/client/transactions`; `client_id` set) share this one table, so the fraud alert queue has a single source of truth. Mirrors the PaySim column set needed for scoring: step, type, amount, nameOrig, oldbalanceOrg, newbalanceOrig, nameDest, oldbalanceDest, newbalanceDest, and the isFraud label. `is_fraud` is nullable because a freshly-submitted real-time transaction has no known ground-truth label at score time; historical PaySim rows loaded for reference/training carry the known label. """ __tablename__ = "transactions" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) client_id: Mapped[int | None] = mapped_column( ForeignKey("clients.id", ondelete="SET NULL"), nullable=True, index=True ) step: Mapped[int] = mapped_column(Integer, nullable=False, index=True) type: Mapped[str] = mapped_column(String(20), nullable=False, index=True) amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False) name_orig: Mapped[str] = mapped_column(String(50), nullable=False, index=True) oldbalance_org: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False) newbalance_orig: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False) name_dest: Mapped[str] = mapped_column(String(50), nullable=False, index=True) oldbalance_dest: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False) newbalance_dest: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False) is_fraud: Mapped[bool | None] = mapped_column(Boolean, nullable=True) reversed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="0") created_at: Mapped[datetime] = mapped_column( DateTime, server_default=func.now(), nullable=False ) client: Mapped["Client | None"] = relationship(back_populates="transactions") predictions: Mapped[list["FraudPrediction"]] = relationship( back_populates="transaction", cascade="all, delete-orphan" ) class FraudPrediction(Base): """One row per scored transaction. `top_features` is a structured, queryable JSON column -- a list of `{"feature": , "shap_value": }` objects ordered by descending absolute SHAP value (top-N) -- never an opaque blob. `model_version` records the artifact identifier (e.g. "model_v1") the prediction was produced with, so the UI can show which model produced a given result (PITFALLS.md UX Pitfalls: stale cached predictions after a redeploy). """ __tablename__ = "fraud_predictions" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) transaction_id: Mapped[int] = mapped_column( ForeignKey("transactions.id", ondelete="CASCADE"), nullable=False, index=True ) probability: Mapped[float] = mapped_column( Numeric(6, 5, asdecimal=False), nullable=False ) risk_tier: Mapped[str] = mapped_column(String(10), nullable=False, index=True) model_version: Mapped[str] = mapped_column(String(50), nullable=False) top_features: Mapped[list[dict]] = mapped_column(JSON, nullable=False) created_at: Mapped[datetime] = mapped_column( DateTime, server_default=func.now(), nullable=False, index=True ) transaction: Mapped["Transaction"] = relationship(back_populates="predictions") alerts: Mapped[list["FraudAlert"]] = relationship( back_populates="prediction", cascade="all, delete-orphan" ) class FraudAlert(Base): """One row per prediction whose probability crossed FRAUD_ALERT_THRESHOLD. `disposition` is constrained to open/reviewed/dismissed (DB-04); `reviewed_at` is nullable and set when an officer moves the alert out of `open`. This shape lets a future labeling/retraining pipeline select reviewed alerts (and treat disposition as a soft label signal) without altering the schema. Reviewer identity isn't tracked per-alert yet (single-officer-role assumption still holds even with multiple officer accounts) -- see threat register T-02-04. """ __tablename__ = "fraud_alerts" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) prediction_id: Mapped[int] = mapped_column( ForeignKey("fraud_predictions.id", ondelete="CASCADE"), nullable=False, index=True, ) risk_tier: Mapped[str] = mapped_column(String(10), nullable=False, index=True) disposition: Mapped[AlertDisposition] = mapped_column( SAEnum( AlertDisposition, native_enum=True, length=20, name="alert_disposition", # Use the enum *values* ("open"/"reviewed"/"dismissed") as the # DB-level ENUM labels rather than the Python member names # ("OPEN"/"REVIEWED"/"DISMISSED") -- otherwise server_default # below (which is a value) would not match any label the # database actually accepts. values_callable=lambda enum_cls: [member.value for member in enum_cls], ), nullable=False, default=AlertDisposition.OPEN, server_default=AlertDisposition.OPEN.value, index=True, ) reviewed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime, server_default=func.now(), nullable=False, index=True ) prediction: Mapped["FraudPrediction"] = relationship(back_populates="alerts")