"""Central configuration: paths, constants, and domain priors for MuleGuard.""" from __future__ import annotations from pathlib import Path # ---- Paths ----------------------------------------------------------------- ROOT = Path(__file__).resolve().parents[1] DATA_DIR = ROOT / "data" ARTIFACTS_DIR = ROOT / "artifacts" REPORTS_DIR = ROOT / "reports" FIGURES_DIR = REPORTS_DIR / "figures" RAW_CSV = DATA_DIR / "DataSet.csv" CACHE_PARQUET = DATA_DIR / "dataset.parquet" # fast-load cache of the raw CSV # Persisted model artifacts PIPELINE_PATH = ARTIFACTS_DIR / "feature_pipeline.pkl" MODEL_PATH = ARTIFACTS_DIR / "model.pkl" FEATURE_LIST_PATH = ARTIFACTS_DIR / "feature_list.json" THRESHOLD_PATH = ARTIFACTS_DIR / "threshold.json" METADATA_PATH = ARTIFACTS_DIR / "metadata.json" SHAP_EXPLAINER_PATH = ARTIFACTS_DIR / "shap_explainer.pkl" TEST_SPLIT_PATH = ARTIFACTS_DIR / "test_holdout.parquet" # Lightweight sample shipped with the deployed app (so it runs without the 111MB dataset). SAMPLE_ACCOUNTS_PATH = ARTIFACTS_DIR / "sample_accounts.parquet" def stream_source() -> "Path": """Held-out set for streaming/scoring: full test split locally, sample on deploy.""" return TEST_SPLIT_PATH if TEST_SPLIT_PATH.exists() else SAMPLE_ACCOUNTS_PATH # ---- Modeling constants ---------------------------------------------------- TARGET = "F3924" INDEX_COL = "Unnamed: 0" # the leading unnamed index column in the CSV # Leakage exclusions — removed to keep the model honest and metrics trustworthy: # F3912 — binary flag aligned ~perfectly with the target (single-feature AUC # 0.987; 1 for 79/81 mules, 0 for 8998/9001 legit). A label-adjacent # "fraud-flagged" indicator, not a learned behavioral signal. # F2230 — observation month. ALL legit accounts fall in Oct25 while ALL mules # fall in Sep/Nov/Dec25 — a temporal data-collection artifact that # perfectly separates classes (AUC 1.000) but encodes no real behavior. LEAKAGE_EXCLUDE = ["F3912", "F2230"] SEED = 42 TEST_SIZE = 0.20 N_SELECT = 80 # target number of features after selection # Bank-flagged "commonly used" high-signal fraud features (domain priors). KNOWN_IMPORTANT = [ "F115", "F321", "F527", "F531", "F670", "F1692", "F2082", "F2122", "F2582", "F2678", "F2737", "F2956", "F3043", "F3836", "F3887", "F3889", "F3891", "F3894", ] # Risk tiers on the threshold-anchored 0-100 score (50 == decision threshold, # so anything >=50 is a flagged alert and tiers are always consistent). RISK_TIERS = [ (0, 50, "LOW"), (50, 70, "MEDIUM"), (70, 85, "HIGH"), (85, 101, "CRITICAL"), ] def ensure_dirs() -> None: for d in (ARTIFACTS_DIR, REPORTS_DIR, FIGURES_DIR): d.mkdir(parents=True, exist_ok=True) def prob_to_risk(prob: float, threshold: float) -> float: """Map a calibrated probability to a threshold-anchored 0-100 risk score. The decision threshold maps to exactly 50, so risk >= 50 == flagged. Below threshold scales into [0, 50]; above into [50, 100]. Keeps score, tier, and decision mutually consistent and intuitive for analysts. """ threshold = min(max(threshold, 1e-6), 1 - 1e-6) if prob < threshold: score = 50.0 * (prob / threshold) else: score = 50.0 + 50.0 * (prob - threshold) / (1.0 - threshold) return round(min(max(score, 0.0), 100.0), 1) def risk_tier(score_0_100: float) -> str: for lo, hi, name in RISK_TIERS: if lo <= score_0_100 < hi: return name return "CRITICAL"