Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| import joblib | |
| import numpy as np | |
| import pandas as pd | |
| import gradio as gr | |
| from pathlib import Path | |
| # Modellarning lokal manzili | |
| MODELS_DIR = Path("/home/user/app/outputs/models") | |
| _fraud_predictor = None | |
| _credit_predictor = None | |
| def load_predictors(): | |
| global _fraud_predictor, _credit_predictor | |
| if _fraud_predictor is None: | |
| # Modellarni to'g'ridan-to'g'ri ko'rsatilgan papkadan yuklaymiz | |
| fraud_dir = MODELS_DIR / "fraud" | |
| credit_dir = MODELS_DIR / "credit" | |
| _fraud_predictor = SimpleFraudPredictor(fraud_dir) | |
| _credit_predictor = SimpleCreditPredictor(credit_dir) | |
| class SimpleFraudPredictor: | |
| def __init__(self, model_dir): | |
| self.fe_artifacts = joblib.load(model_dir / "fe_artifacts.pkl") | |
| self.prep_artifacts = joblib.load(model_dir / "prep_artifacts.pkl") | |
| self.fs_artifacts = joblib.load(model_dir / "fs_artifacts.pkl") | |
| self.model = joblib.load(model_dir / "fraud_model.pkl") | |
| self.xgb = joblib.load(model_dir / "tuned_xgb.pkl") | |
| self.lgb = joblib.load(model_dir / "tuned_lgb.pkl") | |
| self.cat = joblib.load(model_dir / "tuned_cat.pkl") | |
| with open(model_dir / "fraud_model_metadata.json") as f: | |
| meta = json.load(f) | |
| self.threshold = meta["threshold"] | |
| self.model_name = meta["model_name"] | |
| self.final_feats = self.fs_artifacts["final_features"] | |
| def predict(self, transaction: dict) -> dict: | |
| import sys | |
| sys.path.insert(0, "/home/user/app") | |
| from src.features.ieee_cis.feature_engineer import feature_engineer_test | |
| from src.data.ieee_cis.preprocessor import preprocess_test | |
| from sklearn.linear_model import LogisticRegression | |
| from sklearn.pipeline import Pipeline | |
| df = pd.DataFrame([transaction]) | |
| df = feature_engineer_test(df, self.fe_artifacts) | |
| df = preprocess_test(df, self.prep_artifacts) | |
| df = df.apply(pd.to_numeric, errors="coerce").fillna(0.0) | |
| missing = [c for c in self.final_feats if c not in df.columns] | |
| for col in missing: | |
| df[col] = 0.0 | |
| X = df[self.final_feats] | |
| if isinstance(self.model, (LogisticRegression, Pipeline)): | |
| meta_input = np.column_stack([ | |
| self.xgb.predict_proba(X)[:, 1], | |
| self.lgb.predict_proba(X)[:, 1], | |
| self.cat.predict_proba(X)[:, 1], | |
| ]) | |
| prob = self.model.predict_proba(meta_input)[:, 1][0] | |
| else: | |
| prob = self.model.predict_proba(X)[:, 1][0] | |
| return { | |
| "fraud_probability": round(float(prob), 4), | |
| "is_fraud": bool(prob >= self.threshold), | |
| "threshold": self.threshold, | |
| "model": self.model_name, | |
| } | |
| class SimpleCreditPredictor: | |
| def __init__(self, model_dir): | |
| self.fe_artifacts = joblib.load(model_dir / "fe_artifacts.pkl") | |
| self.prep_artifacts = joblib.load(model_dir / "prep_artifacts.pkl") | |
| self.fs_artifacts = joblib.load(model_dir / "fs_artifacts.pkl") | |
| self.model = joblib.load(model_dir / "credit_model.pkl") | |
| self.xgb = joblib.load(model_dir / "tuned_xgb.pkl") | |
| self.lgb = joblib.load(model_dir / "tuned_lgb.pkl") | |
| self.cat = joblib.load(model_dir / "tuned_cat.pkl") | |
| with open(model_dir / "credit_model_metadata.json") as f: | |
| meta = json.load(f) | |
| self.threshold = meta["threshold"] | |
| self.model_name = meta["model_name"] | |
| self.final_feats = self.fs_artifacts["final_features"] | |
| def predict(self, applicant: dict) -> dict: | |
| import sys | |
| sys.path.insert(0, "/home/user/app") | |
| from src.features.home_credit.feature_engineer import feature_engineer_test | |
| from src.data.home_credit.preprocessor import preprocess_test | |
| from sklearn.linear_model import LogisticRegression | |
| from sklearn.pipeline import Pipeline | |
| df = pd.DataFrame([applicant]) | |
| empty_tables = { | |
| "bureau": pd.DataFrame(), "bureau_balance": pd.DataFrame(), | |
| "previous_app": pd.DataFrame(), "pos_cash": pd.DataFrame(), | |
| "installments": pd.DataFrame(), "credit_card": pd.DataFrame(), | |
| } | |
| df = feature_engineer_test(df, empty_tables, self.fe_artifacts) | |
| df = preprocess_test(df, self.prep_artifacts) | |
| df = df.apply(pd.to_numeric, errors="coerce").fillna(0.0) | |
| missing = [c for c in self.final_feats if c not in df.columns] | |
| for col in missing: | |
| df[col] = 0.0 | |
| X = df[self.final_feats] | |
| if isinstance(self.model, (LogisticRegression, Pipeline)): | |
| meta_input = np.column_stack([ | |
| self.xgb.predict_proba(X)[:, 1], | |
| self.lgb.predict_proba(X)[:, 1], | |
| self.cat.predict_proba(X)[:, 1], | |
| ]) | |
| prob = self.model.predict_proba(meta_input)[:, 1][0] | |
| else: | |
| prob = self.model.predict_proba(X)[:, 1][0] | |
| return { | |
| "default_probability": round(float(prob), 4), | |
| "will_default": bool(prob >= self.threshold), | |
| "threshold": self.threshold, | |
| "model": self.model_name, | |
| } | |
| def predict_fraud(txn_id, txn_dt, txn_amt, prod_cd, card4, card6, p_email, r_email, dev_type): | |
| load_predictors() | |
| payload = { | |
| "TransactionID": int(txn_id), "TransactionDT": int(txn_dt), | |
| "TransactionAmt": float(txn_amt), "ProductCD": prod_cd, | |
| "card4": card4, "card6": card6, | |
| "P_emaildomain": p_email, "R_emaildomain": r_email, | |
| "DeviceType": dev_type, | |
| } | |
| result = _fraud_predictor.predict(payload) | |
| color = "RED — FRAUD" if result["is_fraud"] else "GREEN — LEGITIMATE" | |
| output = f"Decision: {color}\nFraud Probability: {result['fraud_probability']}\nThreshold: {result['threshold']}\nModel: {result['model']}" | |
| return output, round(result["fraud_probability"] * 100, 2) | |
| def predict_credit(sk_id, amt_credit, amt_income, amt_annuity, days_birth, days_employed, ext1, ext2, ext3): | |
| load_predictors() | |
| payload = { | |
| "SK_ID_CURR": int(sk_id), "AMT_CREDIT": float(amt_credit), | |
| "AMT_INCOME_TOTAL": float(amt_income), "AMT_ANNUITY": float(amt_annuity), | |
| "DAYS_BIRTH": int(days_birth), "DAYS_EMPLOYED": int(days_employed), | |
| } | |
| if ext1: payload["EXT_SOURCE_1"] = float(ext1) | |
| if ext2: payload["EXT_SOURCE_2"] = float(ext2) | |
| if ext3: payload["EXT_SOURCE_3"] = float(ext3) | |
| result = _credit_predictor.predict(payload) | |
| color = "RED — HIGH RISK" if result["will_default"] else "GREEN — LOW RISK" | |
| output = f"Decision: {color}\nDefault Probability: {result['default_probability']}\nThreshold: {result['threshold']}\nModel: {result['model']}" | |
| return output, round(result["default_probability"] * 100, 2) | |
| # Gradio Interfeysi | |
| with gr.Blocks(title="FinRiskGuard", theme=gr.themes.Soft()) as demo: | |
| gr.Markdown("# FinRiskGuard\n### Fraud Detection and Credit Default Scoring") | |
| with gr.Tabs(): | |
| with gr.Tab("Fraud Detection"): | |
| gr.Markdown("#### IEEE-CIS | AUC-ROC: 0.9258 | Threshold: 0.44") | |
| with gr.Row(): | |
| with gr.Column(): | |
| txn_id = gr.Number(label="TransactionID", value=2987004, precision=0) | |
| txn_dt = gr.Number(label="TransactionDT", value=86400, precision=0) | |
| txn_amt = gr.Number(label="TransactionAmt", value=117.5) | |
| prod_cd = gr.Dropdown(["W","H","C","S","R"], label="ProductCD", value="W") | |
| card4 = gr.Dropdown(["visa","mastercard","american express","discover"], label="card4", value="visa") | |
| card6 = gr.Dropdown(["debit","credit","debit or credit","charge card"], label="card6", value="debit") | |
| p_email = gr.Textbox(label="P_emaildomain", value="gmail.com") | |
| r_email = gr.Textbox(label="R_emaildomain", value="gmail.com") | |
| dev_type = gr.Dropdown(["desktop","mobile"], label="DeviceType", value="desktop") | |
| btn_f = gr.Button("Predict", variant="primary") | |
| with gr.Column(): | |
| fraud_result = gr.Textbox(label="Result") | |
| fraud_gauge = gr.Number(label="Fraud Probability (%)") | |
| btn_f.click(predict_fraud, | |
| inputs=[txn_id, txn_dt, txn_amt, prod_cd, card4, card6, p_email, r_email, dev_type], | |
| outputs=[fraud_result, fraud_gauge]) | |
| with gr.Tab("Credit Scoring"): | |
| gr.Markdown("#### Home Credit | AUC-ROC: 0.7849 | Threshold: 0.50") | |
| with gr.Row(): | |
| with gr.Column(): | |
| sk_id = gr.Number(label="SK_ID_CURR", value=100002, precision=0) | |
| amt_cred = gr.Number(label="AMT_CREDIT", value=406597.5) | |
| amt_inc = gr.Number(label="AMT_INCOME_TOTAL", value=202500.0) | |
| amt_ann = gr.Number(label="AMT_ANNUITY", value=24700.5) | |
| d_birth = gr.Number(label="DAYS_BIRTH", value=-9461, precision=0) | |
| d_emp = gr.Number(label="DAYS_EMPLOYED", value=-637, precision=0) | |
| ext1 = gr.Number(label="EXT_SOURCE_1", value=None) | |
| ext2 = gr.Number(label="EXT_SOURCE_2", value=0.651) | |
| ext3 = gr.Number(label="EXT_SOURCE_3", value=0.493) | |
| btn_c = gr.Button("Predict", variant="primary") | |
| with gr.Column(): | |
| credit_result = gr.Textbox(label="Result") | |
| credit_gauge = gr.Number(label="Default Probability (%)") | |
| btn_c.click(predict_credit, | |
| inputs=[sk_id, amt_cred, amt_inc, amt_ann, d_birth, d_emp, ext1, ext2, ext3], | |
| outputs=[credit_result, credit_gauge]) | |
| gr.Markdown("---\nFinRiskGuard | XGBoost · LightGBM · CatBoost · Stacking · SHAP") | |
| if __name__ == "__main__": | |
| demo.launch() |