প্রজেক্ট: end-to-end ML pipeline

-- transactions table (PostgreSQL)
CREATE TABLE transactions (
  txn_id          BIGSERIAL PRIMARY KEY,
  sender_user_id  BIGINT NOT NULL,
  receiver_msisdn VARCHAR(15) NOT NULL,
  amount_bdt      NUMERIC(14,2) NOT NULL CHECK (amount_bdt > 0),
  txn_type        VARCHAR(20),  -- send_money, cash_out, payment
  status          VARCHAR(20),  -- pending, success, failed
  channel         VARCHAR(20),  -- app, ussd, agent
  device_id       VARCHAR(64),
  ip_address      INET,
  geo_lat         NUMERIC(9,6),
  geo_lon         NUMERIC(9,6),
  created_at      TIMESTAMPTZ DEFAULT NOW(),
  updated_at      TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_txn_sender_time ON transactions(sender_user_id, created_at DESC);
CREATE INDEX idx_txn_status_time ON transactions(status, created_at DESC);

-- WAL-level logical replication enabled for CDC
ALTER SYSTEM SET wal_level = 'logical';

{
  "name": "daakpay-postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg-primary.daakpay.internal",
    "database.port": "5432",
    "database.user": "debezium_reader",
    "database.password": "${file:/secrets/db.pwd}",
    "database.dbname": "daakpay_prod",
    "topic.prefix": "daakpay",
    "table.include.list": "public.transactions,public.users,public.accounts",
    "plugin.name": "pgoutput",
    "snapshot.mode": "initial",
    "decimal.handling.mode": "double",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-changes.daakpay",
    "transforms": "unwrap,route",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "daakpay\\.public\\.(.*)",
    "transforms.route.replacement": "cdc.daakpay.$1"
  }
}

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp, to_date

spark = (SparkSession.builder
    .appName("bronze_txn_ingest")
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
    .config("spark.sql.catalog.lake", "org.apache.iceberg.spark.SparkCatalog")
    .config("spark.sql.catalog.lake.type", "glue")
    .config("spark.sql.catalog.lake.warehouse", "s3://daakpay-lake/")
    .getOrCreate())

raw_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "cdc.daakpay.transactions")
    .option("startingOffsets", "latest")
    .option("failOnDataLoss", "false")
    .load())

# Kafka value JSON parse + ingestion metadata
parsed = (raw_stream
    .selectExpr("CAST(value AS STRING) AS payload",
                "topic", "partition", "offset", "timestamp")
    .withColumn("ingest_ts", current_timestamp())
    .withColumn("ingest_date", to_date("ingest_ts")))

(parsed.writeStream
    .format("iceberg")
    .outputMode("append")
    .option("path", "lake.bronze.transactions_raw")
    .option("checkpointLocation", "s3://daakpay-checkpoints/bronze_txn/")
    .partitionBy("ingest_date")
    .trigger(processingTime="1 minute")
    .start())

from pyspark.sql.functions import from_json, col, to_timestamp
from pyspark.sql.types import (StructType, StructField, LongType,
                               StringType, DoubleType, TimestampType)

txn_schema = StructType([
    StructField("txn_id", LongType()),
    StructField("sender_user_id", LongType()),
    StructField("receiver_msisdn", StringType()),
    StructField("amount_bdt", DoubleType()),
    StructField("txn_type", StringType()),
    StructField("status", StringType()),
    StructField("channel", StringType()),
    StructField("device_id", StringType()),
    StructField("ip_address", StringType()),
    StructField("geo_lat", DoubleType()),
    StructField("geo_lon", DoubleType()),
    StructField("created_at", StringType()),
])

bronze = spark.readStream.format("iceberg").load("lake.bronze.transactions_raw")

silver = (bronze
    .withColumn("data", from_json(col("payload"), txn_schema))
    .select("data.*", "ingest_ts")
    .withColumn("created_at", to_timestamp("created_at"))
    .filter(col("amount_bdt").isNotNull() & (col("amount_bdt") > 0))
    .filter(col("status").isin("success", "failed", "pending"))
    .dropDuplicates(["txn_id"])
    .withColumn("event_date", col("created_at").cast("date")))

(silver.writeStream
    .format("iceberg")
    .outputMode("append")
    .option("path", "lake.silver.transactions")
    .option("checkpointLocation", "s3://daakpay-checkpoints/silver_txn/")
    .partitionBy("event_date")
    .trigger(processingTime="2 minutes")
    .start())

-- models/marts/fact_transactions.sql
{{
  config(
    materialized='incremental',
    unique_key='txn_id',
    on_schema_change='fail',
    incremental_strategy='merge',
    partition_by={'field': 'event_date', 'data_type': 'date'},
    cluster_by=['sender_user_id'],
    meta={
      'owner': 'risk-data-team@daakpay.com.bd',
      'pii_level': 'restricted',
      'retention_days': 2555,
      'sla_hours': 2
    }
  )
}}

WITH txn AS (
  SELECT * FROM {{ ref('stg_silver_transactions') }}
  {% if is_incremental() %}
    WHERE event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 2 DAY)
  {% endif %}
),

users AS (SELECT * FROM {{ ref('dim_users') }}),

agents AS (SELECT * FROM {{ ref('dim_agents') }})

SELECT
  t.txn_id,
  t.created_at AS event_at,
  CAST(t.created_at AS DATE) AS event_date,
  t.sender_user_id,
  u.user_segment,
  u.kyc_level,
  u.account_age_days,
  t.receiver_msisdn,
  t.amount_bdt,
  t.txn_type,
  t.status,
  t.channel,
  t.device_id,
  t.geo_lat,
  t.geo_lon,
  -- pre-compute risk signals (cheap features)
  CASE
    WHEN t.amount_bdt >= 25000 THEN 'high'
    WHEN t.amount_bdt >= 5000  THEN 'medium'
    ELSE 'low'
  END AS amount_band,
  TIMESTAMP_DIFF(t.created_at, u.last_login_at, MINUTE) AS minutes_since_login
FROM txn t
LEFT JOIN users u USING (sender_user_id)
LEFT JOIN agents a ON t.receiver_msisdn = a.agent_msisdn

# features/user_velocity.py
from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, RedisOnlineStore
from feast.types import Float32, Int64

user = Entity(name="user", join_keys=["sender_user_id"])

# Offline source — Iceberg/Snowflake table
txn_source = FileSource(
    name="silver_txn_source",
    path="s3://daakpay-lake/silver/transactions/",
    timestamp_field="event_at",
    created_timestamp_column="ingest_ts",
)

user_velocity_fv = FeatureView(
    name="user_velocity",
    entities=[user],
    ttl=timedelta(days=1),
    schema=[
        Field(name="txn_count_30d",   dtype=Int64),
        Field(name="txn_amount_30d",  dtype=Float32),
        Field(name="txn_count_1h",    dtype=Int64),
        Field(name="distinct_recv_24h", dtype=Int64),
        Field(name="avg_amount_30d",  dtype=Float32),
        Field(name="velocity_score",  dtype=Float32),
    ],
    source=txn_source,
    online=True,
)

import mlflow
import xgboost as xgb
import pandas as pd
from sklearn.metrics import roc_auc_score, average_precision_score
from feast import FeatureStore
from datetime import datetime, timedelta

mlflow.set_tracking_uri("http://mlflow.daakpay.internal:5000")
mlflow.set_experiment("fraud_detection_daily")

# 1. Training labels — last 60 days, fraud confirmed
labels = pd.read_sql("""
    SELECT txn_id, sender_user_id, event_at,
           CASE WHEN fraud_label IS NOT NULL THEN 1 ELSE 0 END AS y
    FROM analytics.fact_transactions f
    LEFT JOIN risk.fraud_confirmed c USING (txn_id)
    WHERE event_date BETWEEN DATE_SUB(CURRENT_DATE(), INTERVAL 60 DAY)
                         AND DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
""", conn)

# 2. Get features at the time of each transaction (point-in-time correct!)
fs = FeatureStore(repo_path="features/")
training_df = fs.get_historical_features(
    entity_df=labels,
    features=[
        "user_velocity:txn_count_30d",
        "user_velocity:txn_amount_30d",
        "user_velocity:distinct_recv_24h",
        "user_velocity:velocity_score",
        "user_profile:account_age_days",
        "user_profile:kyc_level",
        "device_risk:device_score",
    ],
).to_df()

X = training_df.drop(columns=["txn_id", "sender_user_id", "event_at", "y"])
y = training_df["y"]

# 3. Train + log
with mlflow.start_run(run_name=f"daily_{datetime.now():%Y%m%d}"):
    model = xgb.XGBClassifier(
        n_estimators=400, max_depth=6, learning_rate=0.05,
        scale_pos_weight=50,  # imbalanced — fraud rare
        eval_metric="aucpr", tree_method="hist",
    )
    model.fit(X, y)

    auc  = roc_auc_score(y, model.predict_proba(X)[:, 1])
    aupr = average_precision_score(y, model.predict_proba(X)[:, 1])

    mlflow.log_metric("auc", auc)
    mlflow.log_metric("aupr", aupr)
    mlflow.log_param("rows", len(X))
    mlflow.xgboost.log_model(model, "model",
        registered_model_name="fraud_detector")

    # Auto-promote if metrics threshold met
    if aupr > 0.65:
        client = mlflow.MlflowClient()
        v = client.get_latest_versions("fraud_detector",
                                       stages=["None"])[0].version
        client.transition_model_version_stage(
            "fraud_detector", v, "Production",
            archive_existing_versions=True)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from feast import FeatureStore
import mlflow.xgboost
import time, logging

app = FastAPI(title="Daak Pay fraud scorer")
logger = logging.getLogger("fraud_api")

fs = FeatureStore(repo_path="/app/features")
model = mlflow.xgboost.load_model("models:/fraud_detector/Production")

class TxnIn(BaseModel):
    txn_id: int
    sender_user_id: int
    amount_bdt: float
    device_id: str
    ip_address: str

@app.post("/score")
async def score(txn: TxnIn):
    t0 = time.perf_counter()

    # 1. fetch online features (Redis ~5ms)
    feats = fs.get_online_features(
        features=[
            "user_velocity:txn_count_30d",
            "user_velocity:txn_amount_30d",
            "user_velocity:distinct_recv_24h",
            "user_velocity:velocity_score",
            "user_profile:account_age_days",
            "user_profile:kyc_level",
        ],
        entity_rows=[{"sender_user_id": txn.sender_user_id}],
    ).to_dict()

    # 2. Build feature vector
    row = [
        feats["txn_count_30d"][0] or 0,
        feats["txn_amount_30d"][0] or 0,
        feats["distinct_recv_24h"][0] or 0,
        feats["velocity_score"][0] or 0,
        feats["account_age_days"][0] or 0,
        feats["kyc_level"][0] or 0,
        device_risk_lookup(txn.device_id),  # external Redis
    ]

    # 3. Predict
    score = float(model.predict_proba([row])[0][1])
    decision = "block" if score > 0.85 else ("review" if score > 0.45 else "approve")

    latency_ms = (time.perf_counter() - t0) * 1000
    logger.info(f"txn={txn.txn_id} score={score:.3f} dec={decision} lat={latency_ms:.1f}ms")

    return {
        "txn_id": txn.txn_id,
        "score": round(score, 4),
        "decision": decision,
        "latency_ms": round(latency_ms, 2),
        "model_version": "fraud_detector_v_prod",
    }

@app.get("/health")
async def health():
    return {"status": "ok"}

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "risk-data-team",
    "retries": 2,
    "retry_delay": timedelta(minutes=10),
    "email_on_failure": True,
    "email": ["risk-oncall@daakpay.com.bd"],
}

with DAG(
    dag_id="fraud_pipeline_daily",
    default_args=default_args,
    schedule="0 2 * * *",          # 02:00 BST daily
    start_date=datetime(2025, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=["fraud", "ml", "production"],
) as dag:

    # 1. dbt build silver → gold
    dbt_run = DbtCloudRunJobOperator(
        task_id="dbt_marts",
        dbt_cloud_conn_id="dbt_cloud",
        job_id=42,
        check_interval=60,
        timeout=3600,
    )

    # 2. Materialize Feast features
    feast_apply = KubernetesPodOperator(
        task_id="feast_materialize",
        image="daakpay/feast-runner:1.4.0",
        cmds=["bash", "-c"],
        arguments=["feast materialize-incremental $(date -u +%Y-%m-%dT%H:%M:%S)"],
        get_logs=True,
    )

    # 3. Train model
    train = KubernetesPodOperator(
        task_id="train_model",
        image="daakpay/ml-trainer:0.9.2",
        cmds=["python", "-m", "training.daily"],
        resources={"requests": {"memory": "16Gi", "cpu": "4"}},
        get_logs=True,
    )

    # 4. Validate before promotion
    validate = PythonOperator(
        task_id="validate_model",
        python_callable=lambda **ctx: assert_min_aupr(threshold=0.65),
    )

    # 5. Promote + restart inference pods (rolling)
    promote = KubernetesPodOperator(
        task_id="promote_and_rollout",
        image="daakpay/k8s-helper:1.0",
        cmds=["bash", "-c"],
        arguments=["kubectl rollout restart deploy/fraud-api -n risk"],
    )

    # 6. Daily metrics report
    report = PythonOperator(
        task_id="slack_summary",
        python_callable=post_slack_summary,
    )

    dbt_run >> feast_apply >> train >> validate >> promote >> report