File size: 2,681 Bytes
a1f48d9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | 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()
|