| import joblib |
| import pandas as pd |
| from flask import Flask, jsonify, request |
|
|
| |
| app = Flask("ExtraaLearn Lead Conversion Predictor") |
| model = joblib.load("extraalearn_lead_model_v1_0.joblib") |
|
|
| |
| FEATURES = ["age", "current_occupation", "first_interaction", "profile_completed", "website_visits", |
| "time_spent_on_website", "page_views_per_visit", "last_activity", "print_media_type1", |
| "print_media_type2", "digital_media", "educational_channels", "referral"] |
| NUMERIC = {"age": int, "website_visits": int, "time_spent_on_website": int, "page_views_per_visit": float} |
| CATEGORIES = { |
| "current_occupation": ["Professional", "Unemployed", "Student"], |
| "first_interaction": ["Website", "Mobile App"], |
| "profile_completed": ["Low", "Medium", "High"], |
| "last_activity": ["Email Activity", "Phone Activity", "Website Activity"], |
| **{flag: ["Yes", "No"] for flag in ["print_media_type1", "print_media_type2", "digital_media", |
| "educational_channels", "referral"]}, |
| } |
|
|
|
|
| def validate(record): |
| """Check that a lead record has every feature with a valid value; return a list of errors.""" |
| errors = [] |
| for f in FEATURES: |
| if f not in record: |
| errors.append(f"missing field '{f}'") |
| elif f in NUMERIC: |
| try: |
| if NUMERIC[f](record[f]) < 0: |
| errors.append(f"'{f}' must be non-negative") |
| except (TypeError, ValueError): |
| errors.append(f"'{f}' must be numeric") |
| elif record[f] not in CATEGORIES[f]: |
| errors.append(f"'{f}' must be one of {CATEGORIES[f]}") |
| return errors |
|
|
|
|
| def to_frame(records): |
| """Build a model-ready DataFrame with the right column order and numeric types.""" |
| df = pd.DataFrame(records)[FEATURES] |
| for f, cast in NUMERIC.items(): |
| df[f] = df[f].astype(cast) |
| return df |
|
|
|
|
| @app.get("/") |
| def home(): |
| """Health check.""" |
| return "Welcome to the ExtraaLearn Lead Conversion Prediction API!" |
|
|
|
|
| @app.post("/v1/lead") |
| def predict_lead(): |
| """Predict the conversion of a single lead sent as a JSON object.""" |
| record = request.get_json(silent=True) |
| if not isinstance(record, dict): |
| return jsonify({"error": "Request body must be a JSON object with the lead's features"}), 400 |
| errors = validate(record) |
| if errors: |
| return jsonify({"error": "Invalid input", "details": errors}), 400 |
|
|
| df = to_frame([record]) |
| probability = float(model.predict_proba(df)[0, 1]) |
| prediction = int(model.predict(df)[0]) |
| return jsonify({ |
| "prediction": prediction, |
| "prediction_label": "Likely to convert" if prediction == 1 else "Unlikely to convert", |
| "conversion_probability": round(probability, 4), |
| }) |
|
|
|
|
| @app.post("/v1/leadbatch") |
| def predict_lead_batch(): |
| """Predict the conversion of every lead in an uploaded CSV file (form field 'file').""" |
| if "file" not in request.files: |
| return jsonify({"error": "Upload a CSV file in the form field 'file'"}), 400 |
| batch = pd.read_csv(request.files["file"]) |
| missing = [f for f in FEATURES if f not in batch.columns] |
| if missing: |
| return jsonify({"error": f"CSV is missing columns: {missing}"}), 400 |
|
|
| df = to_frame(batch) |
| ids = batch["ID"].astype(str).tolist() if "ID" in batch.columns else [str(i) for i in range(len(batch))] |
| results = [ |
| {"ID": i, "prediction": int(p), "conversion_probability": round(float(pr), 4)} |
| for i, p, pr in zip(ids, model.predict(df), model.predict_proba(df)[:, 1]) |
| ] |
| return jsonify(results) |
|
|
|
|
| if __name__ == "__main__": |
| app.run(host="0.0.0.0", port=7860, debug=False) |
|
|
