NSIF-X (National Secure Identity Fabric – Experimental) is an identity risk scoring service for Nigeria’s digital ecosystem.

It combines:

• MSISDN – phone numbers.
• Device fingerprint – stable device hash from apps/web.
• SIM-swap events – SIM change / replacement signals.
• Event context – type (login/transfer/etc.), amount, channel, geo.

and outputs:

• risk_score (0–100)
• risk_level (low / medium / high / critical)
• recommendation (allow / step_up_auth)
• risk_factors (human-readable reasons)

Recommended repo layout:

nsifx/
├── app/
│   ├── main.py
│   ├── api/
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── endpoints/
│   │       │   ├── risk.py
│   │       │   ├── device.py
│   │       │   └── sim.py
│   │       └── router.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── db/
│   │   ├── base.py
│   │   ├── models.py
│   │   ├── session.py
│   │   └── migrations/
│   ├── schemas/
│   │   ├── risk.py
│   │   ├── device.py
│   │   └── sim.py
│   ├── services/
│   │   ├── risk_engine.py
│   │   ├── device_service.py
│   │   └── sim_service.py
│   └── utils/
│       └── logging.py
├── infra/
│   ├── docker-compose.yml
│   ├── Dockerfile.backend
│   └── .env.example
├── tests/
│   └── test_risk_api.py
├── requirements.txt
└── README.md

Key ideas:

• app/services holds business logic (risk engine, SIM logic).
• app/api is thin: validate → call services → respond.
• app/db centralises ORM models and DB session handling.
• infra is where all Docker/deployment config lives.

• Git
• Docker & docker-compose
• Python 3.11+ (only needed if running without Docker)

git clone <REPO_URL> nsifx
cd nsifx
cp infra/.env.example infra/.env

cd infra
docker-compose up --build

Services exposed:

• API: http://localhost:8000
• Docs (Swagger): http://localhost:8000/api/v1/docs
• Postgres: localhost:5432
• Redis: localhost:6379
• Adminer: http://localhost:8080

Environment file: infra/.env

PROJECT_NAME="NSIF-X Risk Engine"
API_V1_STR="/api/v1"

POSTGRES_DSN=postgresql+psycopg2://nsifx_user:nsifx_password@db:5432/nsifx
REDIS_URL=redis://redis:6379/0

RISK_BASELINE=10
SIM_SWAP_RISK_BUMP=40
NEW_DEVICE_RISK_BUMP=25
MAX_RISK_SCORE=100
SIM_SWAP_COOLDOWN_HOURS=48

Settings class: app/core/config.py

from pydantic import BaseSettings, AnyUrl

class Settings(BaseSettings):
    PROJECT_NAME: str = "NSIF-X Risk Engine"
    API_V1_STR: str = "/api/v1"

    POSTGRES_DSN: AnyUrl
    REDIS_URL: str

    RISK_BASELINE: int = 10
    SIM_SWAP_RISK_BUMP: int = 40
    NEW_DEVICE_RISK_BUMP: int = 25
    MAX_RISK_SCORE: int = 100
    SIM_SWAP_COOLDOWN_HOURS: int = 48

    class Config:
        env_file = "infra/.env"

settings = Settings()

Engine & Session: app/db/session.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from app.core.config import settings

engine = create_engine(settings.POSTGRES_DSN, pool_pre_ping=True, future=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True)
Base = declarative_base()

Key entities:

• Device – registered devices (hash, trust_level, metadata).
• PhoneIdentity – phone numbers & risk state.
• PhoneDeviceLink – mapping phone ↔ device.
• SimEvent – SIM swap/replacement history.
• RiskEvent – audit trail of risk changes.

Example models (partial):

from sqlalchemy import Column, String, Integer, DateTime, JSON, ForeignKey, UniqueConstraint
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.sql import func
import uuid
from app.db.session import Base

class Device(Base):
    __tablename__ = "devices"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    device_hash = Column(String(255), unique=True, nullable=False)
    first_seen_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    last_seen_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    trust_level = Column(String(20), nullable=False, default="unknown")
    metadata = Column(JSON)

class PhoneIdentity(Base):
    __tablename__ = "phone_identities"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    msisdn = Column(String(20), unique=True, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    last_seen_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    risk_score = Column(Integer, nullable=False, default=0)
    risk_level = Column(String(20), nullable=False, default="low")
    cooldown_until = Column(DateTime(timezone=True), nullable=True)

class PhoneDeviceLink(Base):
    __tablename__ = "phone_device_links"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    phone_id = Column(UUID(as_uuid=True), ForeignKey("phone_identities.id", ondelete="CASCADE"), nullable=False)
    device_id = Column(UUID(as_uuid=True), ForeignKey("devices.id", ondelete="CASCADE"), nullable=False)
    first_seen_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
    last_seen_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)

    __table_args__ = (
        UniqueConstraint("phone_id", "device_id", name="phone_device_unique"),
    )

File: app/services/risk_engine.py

Responsibilities:

• Ensure PhoneIdentity exists for an msisdn.
• Register new device hashes.
• Link phones ↔ devices.
• Apply SIM swap logic (risk bump + cooldown).
• Compute final risk_score, risk_level & recommendation.

Example (simplified):

from datetime import datetime, timedelta
from typing import Tuple, List
from sqlalchemy.orm import Session
import redis

from app.core.config import settings
from app.db import models

r = redis.from_url(settings.REDIS_URL, decode_responses=True)

def _compute_risk_level(score: int) -> str:
    if score < 30:
        return "low"
    if score < 60:
        return "medium"
    if score < 85:
        return "high"
    return "critical"

def register_device(db: Session, device_hash: str, metadata: dict | None = None) -> models.Device:
    device = db.query(models.Device).filter_by(device_hash=device_hash).first()
    if not device:
        device = models.Device(device_hash=device_hash, metadata=metadata or {})
        db.add(device)
    device.last_seen_at = datetime.utcnow()
    db.commit()
    db.refresh(device)
    return device

def ensure_phone(db: Session, msisdn: str) -> models.PhoneIdentity:
    phone = db.query(models.PhoneIdentity).filter_by(msisdn=msisdn).first()
    if not phone:
        phone = models.PhoneIdentity(msisdn=msisdn, risk_score=settings.RISK_BASELINE)
        db.add(phone)
        db.commit()
        db.refresh(phone)
    phone.last_seen_at = datetime.utcnow()
    db.commit()
    db.refresh(phone)
    return phone

def apply_sim_swap_event(db: Session, msisdn: str) -> models.PhoneIdentity:
    phone = ensure_phone(db, msisdn)
    new_score = min(settings.MAX_RISK_SCORE, phone.risk_score + settings.SIM_SWAP_RISK_BUMP)
    phone.risk_score = new_score
    phone.risk_level = _compute_risk_level(new_score)
    phone.cooldown_until = datetime.utcnow() + timedelta(hours=settings.SIM_SWAP_COOLDOWN_HOURS)
    db.commit()
    return phone

def calculate_risk_for_event(
    db: Session, msisdn: str, device_hash: str, event_type: str, amount: float | None = None
) -> Tuple[int, str, str, List[str]]:
    factors: List[str] = []
    phone = ensure_phone(db, msisdn)
    device = register_device(db, device_hash, metadata=None)

    factors.append("baseline_identity_risk")
    score = phone.risk_score

    if phone.cooldown_until and phone.cooldown_until > datetime.utcnow():
        factors.append("recent_sim_swap_cooldown")
        score = min(settings.MAX_RISK_SCORE, score + 20)

    if event_type == "transfer" and amount and amount > 100000:
        factors.append("high_value_transfer")
        score = min(settings.MAX_RISK_SCORE, score + 10)

    phone.risk_score = score
    phone.risk_level = _compute_risk_level(score)
    db.commit()
    db.refresh(phone)

    if phone.risk_level in ("high", "critical"):
        recommendation = "step_up_auth"
    else:
        recommendation = "allow"

    return phone.risk_score, phone.risk_level, recommendation, factors

Main app: app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.api.v1.router import api_router

app = FastAPI(
    title=settings.PROJECT_NAME,
    version="0.1.0",
    openapi_url=f"{settings.API_V1_STR}/openapi.json",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(api_router, prefix=settings.API_V1_STR)

@app.get("/")
def healthcheck():
    return {"status": "ok", "service": settings.PROJECT_NAME}

Router: app/api/v1/router.py

from fastapi import APIRouter
from .endpoints import risk, device, sim

api_router = APIRouter()
api_router.include_router(risk.router, tags=["risk"])
api_router.include_router(device.router, tags=["device"])
api_router.include_router(sim.router, tags=["sim"])

@router.post("/risk-score", response_model=RiskScoreResponse)
def get_risk_score(payload: RiskScoreRequest, db: Session = Depends(get_db)):
    score, level, recommendation, factors = risk_engine.calculate_risk_for_event(
        db=db,
        msisdn=payload.msisdn,
        device_hash=payload.device_hash,
        event_type=payload.event_type,
        amount=payload.amount,
    )
    return RiskScoreResponse(
        risk_score=score,
        risk_level=level,
        recommendation=recommendation,
        risk_factors=factors,
    )

Risk schemas: app/schemas/risk.py

from pydantic import BaseModel, Field
from typing import Optional, List

class RiskScoreRequest(BaseModel):
    msisdn: str = Field(..., example="+2348012345678")
    device_hash: str = Field(..., example="abcdef1234567890")
    event_type: str = Field(..., example="login")
    amount: Optional[float] = Field(None, example=200000.0)
    channel: Optional[str] = Field(None, example="mobile-app")
    geo: Optional[str] = Field(None, example="Lagos")

class RiskScoreResponse(BaseModel):
    risk_score: int
    risk_level: str
    recommendation: str
    risk_factors: List[str]

Device schemas: app/schemas/device.py

from pydantic import BaseModel, Field
from typing import Optional, Dict

class DeviceRegisterRequest(BaseModel):
    device_hash: str = Field(..., example="abcdef1234567890")
    metadata: Optional[Dict[str, str]] = None

class DeviceRegisterResponse(BaseModel):
    device_id: str
    trust_level: str

SIM schemas: app/schemas/sim.py

from pydantic import BaseModel, Field
from typing import Optional, Dict

class SimEventRequest(BaseModel):
    msisdn: str = Field(..., example="+2348012345678")
    event_type: str = Field(..., example="SIM_SWAP")
    channel: Optional[str] = Field(None, example="retail_agent")
    metadata: Optional[Dict[str, str]] = None

Once the stack is running, you can test APIs directly.

curl http://localhost:8000/

curl -X POST http://localhost:8000/api/v1/device/register \
  -H "Content-Type: application/json" \
  -d '{
    "device_hash": "test-device-abc123",
    "metadata": {"platform": "android"}
  }'

curl -X POST http://localhost:8000/api/v1/sim/event \
  -H "Content-Type: application/json" \
  -d '{
    "msisdn": "+2348012345678",
    "event_type": "SIM_SWAP",
    "channel": "retail_agent"
  }'

curl -X POST http://localhost:8000/api/v1/risk-score \
  -H "Content-Type: application/json" \
  -d '{
    "msisdn": "+2348012345678",
    "device_hash": "test-device-abc123",
    "event_type": "login",
    "amount": 200000,
    "channel": "mobile-app",
    "geo": "Lagos"
  }'

Goal: Let stakeholders see risk scoring in real time.

Frontend stack:

• Next.js 13+ (App Router) or React SPA.
• Tailwind CSS for quick styling.

Environment variables:

• NEXT_PUBLIC_API_BASE → e.g. http://localhost:8000/api/v1
• NEXT_PUBLIC_API_KEY → for auth (later).

Core UI sections:

• Identity panel – MSISDN + device hash input.
• Buttons – “Register Device”, “Simulate SIM Swap”, “Get Risk Score”.
• Result panel – big score, level badge, recommendation, risk factors list.

For demos you can keep it light; for production, you must add:

• API authentication (API keys or JWT/OAuth2) in security.py.
• Locked-down CORS (specific origins only).
• Rate limiting for public endpoints (Redis-backed).
• Request/response logging with correlation IDs.
• Audit logging for risk changes and SIM events.
• Alembic migrations for DB schema evolution.

Recommended workflow:

• Branches:
  • main – stable.
  • dev – integration.
  • feature branches – e.g. feature/velocity-metrics.
• PR rules:
  • All tests pass (pytest).
  • Public APIs documented if changed.
  • Security impact briefly assessed.
• Tests:
  • Unit tests for risk_engine.
  • Integration tests for endpoints (FastAPI TestClient).

After MVP stabilises:

• Velocity & behaviour:
  • Count logins / transfers in sliding windows.
  • Detect unusual device changes per MSISDN.
• NIN/BVN integration:
  • Optional linking to national identifiers for deeper risk context.
• Admin portal:
  • RBAC, institution-level views, risk configuration per client.
• Analytics & reporting:
  • Fraud heatmaps, SIM-swap clusters, high-risk device insights.
• Multi-tenant design:
  • Institution separation for banks/telcos.
  • Shared intelligence, private data slices.