من Notebook لمشروع

price-predictor/
    data/
        apartments.csv
    models/                  # trained models (not in git if large)
    src/
        train.py             # train and save the model
        api.py               # serve predictions (next lesson)
    tests/
        test_api.py
    requirements.txt
    Dockerfile
    .env                     # secrets (in .gitignore!)
    .gitignore
    README.md

import json
import logging
import sys
from datetime import datetime

import joblib
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s",
                    handlers=[logging.StreamHandler(sys.stdout)])
log = logging.getLogger("train")

NUMERIC = ["area", "rooms", "floor", "age", "metro_km"]
CATEGORICAL = ["city"]
MODEL_VERSION = "1.0.0"

def main():
    df = pd.read_csv("apartments.csv")
    log.info("loaded %d rows", len(df))

    X_train, X_test, y_train, y_test = train_test_split(
        df[NUMERIC + CATEGORICAL], df["price_k"], test_size=0.2, random_state=42)

    pipe = Pipeline([
        ("prep", ColumnTransformer([
            ("num", StandardScaler(), NUMERIC),
            ("cat", OneHotEncoder(handle_unknown="ignore"), CATEGORICAL),
        ])),
        ("model", LinearRegression()),
    ])
    pipe.fit(X_train, y_train)

    mae = mean_absolute_error(y_test, pipe.predict(X_test))
    log.info("test MAE = %.1fK EGP", mae)

    joblib.dump(pipe, "price_model.joblib")
    with open("metrics.json", "w") as f:
        json.dump({"version": MODEL_VERSION, "mae_k": round(mae, 1),
                   "trained_at": datetime.now().date().isoformat()}, f)
    log.info("saved model v%s", MODEL_VERSION)

if __name__ == "__main__":
    main()