| from pathlib import Path |
| import argparse,json |
| import pandas as pd |
| import joblib |
|
|
| 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/"final_model.joblib") |
| medians={} |
| median_file=MODELS/"catboost_numeric_medians.json" |
| if median_file.exists(): |
| medians=json.loads(median_file.read_text(encoding="utf-8")) |
| preprocessor=None |
| pp=MODELS/"final_preprocessor.joblib" |
| if pp.exists(): |
| preprocessor=joblib.load(pp) |
| return metadata,model,medians,preprocessor |
|
|
| def _prepare_catboost(features,metadata,medians): |
| required=metadata["feature_columns"] |
| row=pd.DataFrame([{c:features.get(c) for c in required}]) |
| for c in metadata["numeric_columns"]: |
| row[c]=pd.to_numeric(row[c],errors="coerce").fillna(float(medians[c])) |
| for c in metadata["categorical_columns"]: |
| row[c]=row[c].astype("object").where(row[c].notna(),"MISSING").astype(str) |
| return row[required] |
|
|
| def predict_next7d_occupancy(features:dict)->dict: |
| metadata,model,medians,preprocessor=load_artifacts() |
| required=metadata["feature_columns"] |
| missing=[c for c in required if c not in features] |
| if missing: |
| raise ValueError(f"Missing required features: {missing}") |
|
|
| if metadata["model_kind"]=="catboost_native": |
| row=_prepare_catboost(features,metadata,medians) |
| pred=float(model.predict(row)[0]) |
| else: |
| row=pd.DataFrame([{c:features.get(c) for c in required}]) |
| pred=float(model.predict(preprocessor.transform(row))[0]) |
|
|
| pred=max(0.0,min(1.0,pred)) |
| on_books=float(features["on_books_occupancy_next_7d"]) |
| expected_increment=max(0.0,pred-on_books) |
|
|
| return { |
| "predicted_next_7d_occupancy_rate":round(pred,6), |
| "predicted_next_7d_occupancy_percent":round(pred*100,2), |
| "on_books_occupancy_rate":round(on_books,6), |
| "expected_additional_occupancy_rate":round(expected_increment,6), |
| "model_name":metadata["selected_algorithm"], |
| "forecast_window":"next 7 days", |
| "decision_support_only":True, |
| "synthetic_training_data_warning":metadata["synthetic_training_data"], |
| "production_warning":metadata["production_warning"] |
| } |
|
|
| 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_next7d_occupancy(payload.get("features",payload)), |
| ensure_ascii=False,indent=2 |
| )) |
|
|
| if __name__=="__main__": |
| main() |
|
|
