| from pathlib import Path |
| import argparse, json |
| import joblib |
| import pandas as pd |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| MODELS = ROOT / "models" |
|
|
| def load_artifacts(): |
| metadata = json.loads((MODELS/"model_metadata.json").read_text(encoding="utf-8")) |
| model = joblib.load(MODELS/"risk_probability_model.joblib") |
| return metadata, model |
|
|
| def predict_cancellation(features: dict) -> dict: |
| metadata, model = load_artifacts() |
| required = metadata["feature_columns"] |
| missing = [c for c in required if c not in features] |
| extra = [c for c in features if c not in required] |
| if missing: |
| raise ValueError(f"Missing required features: {missing}") |
|
|
| row = pd.DataFrame([{c: features.get(c) for c in required}]) |
| |
| |
| probability = float(model.predict_proba(row)[:, 1][0]) |
| threshold = float(metadata["risk_threshold"]) |
| prediction = int(probability >= threshold) |
|
|
| level = "LOW" if probability < 0.35 else ("MEDIUM" if probability < 0.65 else "HIGH") |
| return { |
| "target": metadata["target"], |
| "cancel_probability": round(probability, 6), |
| "threshold": round(threshold, 6), |
| "prediction": prediction, |
| "prediction_label": "CANCEL_RISK" if prediction else "NOT_CANCEL_RISK", |
| "risk_level": level, |
| "ignored_extra_fields": extra, |
| "model_name": metadata["risk_probability_model"], |
| "synthetic_training_data_warning": metadata["synthetic_training_data"], |
| } |
|
|
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument("json_file") |
| args = parser.parse_args() |
| payload = json.loads(Path(args.json_file).read_text(encoding="utf-8")) |
| print(json.dumps( |
| predict_cancellation(payload.get("features", payload)), |
| ensure_ascii=False, indent=2 |
| )) |
|
|
| if __name__ == "__main__": |
| main() |
|
|
