| import json |
| import os |
| import time |
| import zipfile |
| from uuid import uuid4 |
|
|
| import joblib |
| import pandas as pd |
|
|
| from infra.storage import get_run_dir, resolve_model_path |
|
|
|
|
| def _safe_json_load(path: str): |
| try: |
| if path and os.path.exists(path): |
| with open(path, "r") as f: |
| return json.load(f) |
| except Exception: |
| pass |
| return {} |
|
|
|
|
| def _display_metric(metric_name: str, score_val): |
| try: |
| score_float = float(score_val) |
| except (TypeError, ValueError): |
| return "N/A" |
|
|
| metric_lower = (metric_name or "").lower() |
| if "r²" in metric_name or metric_lower in {"r2", "r2 score"}: |
| return f"{score_float / 100:.3f}" |
| if "rmse" in metric_lower or "mse" in metric_lower or "mae" in metric_lower: |
| return f"{score_float:.4f}" |
| return f"{score_float:.1f}%" |
|
|
|
|
| def _model_import_and_init(best_model_name: str, is_clf: bool): |
| if "LightGBM" in best_model_name: |
| model_class = "LGBMClassifier" if is_clf else "LGBMRegressor" |
| return ( |
| f"from lightgbm import {model_class}", |
| f"{model_class}(random_state=42, verbose=-1)", |
| ) |
| if "XGBoost" in best_model_name or "XGB" in best_model_name: |
| model_class = "XGBClassifier" if is_clf else "XGBRegressor" |
| if is_clf: |
| return ( |
| f"from xgboost import {model_class}", |
| f"{model_class}(random_state=42, n_estimators=150, eval_metric='logloss')", |
| ) |
| return ( |
| f"from xgboost import {model_class}", |
| f"{model_class}(random_state=42, n_estimators=150)", |
| ) |
| if "Forest" in best_model_name: |
| model_class = "RandomForestClassifier" if is_clf else "RandomForestRegressor" |
| return ( |
| f"from sklearn.ensemble import {model_class}", |
| f"{model_class}(n_estimators=200, random_state=42)", |
| ) |
| if "Logistic" in best_model_name: |
| return ( |
| "from sklearn.linear_model import LogisticRegression", |
| "LogisticRegression(max_iter=1000, random_state=42)", |
| ) |
| if "Linear" in best_model_name: |
| return ( |
| "from sklearn.linear_model import LinearRegression", |
| "LinearRegression()", |
| ) |
| if "Ridge" in best_model_name: |
| return ( |
| "from sklearn.linear_model import Ridge", |
| "Ridge(alpha=1.0, random_state=42)", |
| ) |
| if "Lasso" in best_model_name: |
| return ( |
| "from sklearn.linear_model import Lasso", |
| "Lasso(alpha=0.1, random_state=42)", |
| ) |
| if "SVM" in best_model_name or "SVC" in best_model_name: |
| model_class = "SVC" if is_clf else "SVR" |
| if is_clf: |
| return ( |
| f"from sklearn.svm import {model_class}", |
| f"{model_class}(probability=True)", |
| ) |
| return ( |
| f"from sklearn.svm import {model_class}", |
| f"{model_class}()", |
| ) |
| return ( |
| "# Replace with your preferred estimator import", |
| "# Replace with your preferred estimator init", |
| ) |
|
|
|
|
| def _training_script_content( |
| best_model_name: str, |
| target_name: str, |
| feature_names, |
| is_clf: bool, |
| metric_name: str, |
| execution_profile: dict, |
| preprocessor_name: str, |
| ): |
| model_import, model_init = _model_import_and_init(best_model_name, is_clf) |
| selected_features_literal = json.dumps(feature_names or []) |
| is_clf_literal = "True" if is_clf else "False" |
|
|
| return f'''""" |
| End-to-End Training + Inference Script |
| Generated by AutoML Studio |
| |
| This script mirrors the exported pipeline as closely as possible: |
| 1. Load dataset |
| 2. Clean data |
| 3. Select features and target |
| 4. Split train/test |
| 5. Build preprocessing pipeline |
| 6. Train model |
| 7. Evaluate |
| 8. Save reusable bundle |
| 9. Predict on new records |
| """ |
| |
| import json |
| import sys |
| import joblib |
| import numpy as np |
| import pandas as pd |
| |
| from sklearn.compose import ColumnTransformer |
| from sklearn.impute import SimpleImputer |
| from sklearn.metrics import ( |
| accuracy_score, |
| f1_score, |
| mean_absolute_error, |
| mean_squared_error, |
| r2_score, |
| precision_score, |
| recall_score, |
| ) |
| from sklearn.model_selection import train_test_split |
| from sklearn.pipeline import Pipeline |
| from sklearn.preprocessing import LabelEncoder, PolynomialFeatures, StandardScaler |
| |
| try: |
| from category_encoders import TargetEncoder |
| except ImportError: |
| TargetEncoder = None |
| |
| {model_import} |
| |
| |
| TARGET_COLUMN = "{target_name}" |
| SELECTED_FEATURES = {selected_features_literal} |
| IS_CLASSIFICATION = {is_clf_literal} |
| PRIMARY_METRIC = "{metric_name}" |
| EXPORT_PREPROCESSOR = "{preprocessor_name}" |
| EXECUTION_PROFILE = {json.dumps(execution_profile or {{}}, indent=4)} |
| |
| |
| class OutlierClipper: |
| def __init__(self, factor=3.0): |
| self.factor = factor |
| |
| def fit(self, X, y=None): |
| X = pd.DataFrame(X) |
| q1 = X.quantile(0.25) |
| q3 = X.quantile(0.75) |
| iqr = q3 - q1 |
| self.lower_ = q1 - self.factor * iqr |
| self.upper_ = q3 + self.factor * iqr |
| return self |
| |
| def transform(self, X): |
| X = pd.DataFrame(X) |
| return X.clip(lower=self.lower_, upper=self.upper_, axis=1).values |
| |
| |
| class SkewTransformer: |
| def fit(self, X, y=None): |
| X = pd.DataFrame(X) |
| self.skewed_cols_ = X.columns[X.skew(numeric_only=True).abs() > 0.75].tolist() |
| return self |
| |
| def transform(self, X): |
| X = pd.DataFrame(X).copy() |
| for col in self.skewed_cols_: |
| X[col] = np.log1p(np.maximum(X[col], 0)) |
| return X.values |
| |
| |
| def load_dataset(path: str) -> pd.DataFrame: |
| df = pd.read_csv(path) |
| if df is None or df.empty: |
| raise ValueError("Dataset is empty or unreadable.") |
| return df |
| |
| |
| def basic_cleaning(df: pd.DataFrame, target_column: str) -> pd.DataFrame: |
| df = df.copy() |
| placeholders = ["??", "nan", "none", "null", "unknown", "na", "n/a", "invalid", "?"] |
| |
| for col in df.columns: |
| if df[col].dtype == object: |
| df[col] = df[col].replace(placeholders, np.nan) |
| df[col] = df[col].astype(str).str.strip().replace("nan", np.nan) |
| |
| df = df.drop_duplicates() |
| df = df.dropna(subset=[target_column]) |
| |
| high_missing_cols = [c for c in df.columns if c != target_column and df[c].isna().mean() > 0.9] |
| if high_missing_cols: |
| df = df.drop(columns=high_missing_cols) |
| |
| constant_cols = [c for c in df.columns if c != target_column and df[c].nunique(dropna=False) <= 1] |
| if constant_cols: |
| df = df.drop(columns=constant_cols) |
| |
| return df |
| |
| |
| def select_training_columns(df: pd.DataFrame, target_column: str) -> pd.DataFrame: |
| requested = [c for c in SELECTED_FEATURES if c in df.columns and c != target_column] |
| if requested: |
| keep_cols = requested + [target_column] |
| return df[keep_cols].copy() |
| return df.copy() |
| |
| |
| def build_preprocessor(X: pd.DataFrame): |
| numeric_cols = X.select_dtypes(include=[np.number]).columns.tolist() |
| categorical_cols = X.select_dtypes(include=["object", "category", "bool"]).columns.tolist() |
| |
| if EXPORT_PREPROCESSOR == "full_column_transformer": |
| numeric_transformer = Pipeline( |
| steps=[ |
| ("imputer", SimpleImputer(strategy="median")), |
| ("skew_fix", SkewTransformer()), |
| ("outlier_clipper", OutlierClipper()), |
| ("interactions", PolynomialFeatures(degree=2, interaction_only=True, include_bias=False)), |
| ("scaler", StandardScaler()), |
| ] |
| ) |
| else: |
| numeric_transformer = Pipeline( |
| steps=[ |
| ("imputer", SimpleImputer(strategy="median")), |
| ("scaler", StandardScaler()), |
| ] |
| ) |
| |
| cat_steps = [("imputer", SimpleImputer(strategy="constant", fill_value="missing"))] |
| if TargetEncoder is not None and categorical_cols: |
| cat_steps.append(("target_encoder", TargetEncoder())) |
| categorical_transformer = Pipeline(steps=cat_steps) |
| |
| preprocessor = ColumnTransformer( |
| transformers=[ |
| ("num", numeric_transformer, numeric_cols), |
| ("cat", categorical_transformer, categorical_cols), |
| ] |
| ) |
| return preprocessor, numeric_cols, categorical_cols |
| |
| |
| def build_model(): |
| return {model_init} |
| |
| |
| def prepare_target(y: pd.Series): |
| encoder = None |
| if IS_CLASSIFICATION: |
| encoder = LabelEncoder() |
| y = pd.Series(encoder.fit_transform(y.astype(str)), index=y.index) |
| return y, encoder |
| |
| |
| def train_pipeline(df: pd.DataFrame): |
| df = basic_cleaning(df, TARGET_COLUMN) |
| df = select_training_columns(df, TARGET_COLUMN) |
| |
| X = df.drop(columns=[TARGET_COLUMN]) |
| y = df[TARGET_COLUMN] |
| y, label_encoder = prepare_target(y) |
| |
| split_kwargs = {{"test_size": 0.2, "random_state": 42}} |
| if IS_CLASSIFICATION and y.nunique() > 1: |
| split_kwargs["stratify"] = y |
| |
| X_train, X_test, y_train, y_test = train_test_split(X, y, **split_kwargs) |
| |
| preprocessor, numeric_cols, categorical_cols = build_preprocessor(X_train) |
| model = build_model() |
| |
| pipeline = Pipeline( |
| steps=[ |
| ("preprocessor", preprocessor), |
| ("model", model), |
| ] |
| ) |
| pipeline.fit(X_train, y_train) |
| |
| metrics = evaluate_pipeline(pipeline, X_test, y_test) |
| bundle = {{ |
| "model": pipeline, |
| "features": X.columns.tolist(), |
| "target": TARGET_COLUMN, |
| "is_classification": IS_CLASSIFICATION, |
| "label_encoder_classes": label_encoder.classes_.tolist() if label_encoder is not None else None, |
| "metadata": {{ |
| "model_type": "{best_model_name}", |
| "metric_name": PRIMARY_METRIC, |
| "preprocessor": EXPORT_PREPROCESSOR, |
| "numeric_cols": numeric_cols, |
| "categorical_cols": categorical_cols, |
| "execution_profile": EXECUTION_PROFILE, |
| }}, |
| }} |
| return bundle, metrics, X_test |
| |
| |
| def evaluate_pipeline(pipeline, X_test, y_test): |
| preds = pipeline.predict(X_test) |
| if IS_CLASSIFICATION: |
| return {{ |
| "accuracy": round(float(accuracy_score(y_test, preds)) * 100, 2), |
| "precision": round(float(precision_score(y_test, preds, average="weighted", zero_division=0)) * 100, 2), |
| "recall": round(float(recall_score(y_test, preds, average="weighted", zero_division=0)) * 100, 2), |
| "f1": round(float(f1_score(y_test, preds, average="weighted", zero_division=0)) * 100, 2), |
| }} |
| return {{ |
| "r2": round(float(r2_score(y_test, preds)), 4), |
| "rmse": round(float(np.sqrt(mean_squared_error(y_test, preds))), 4), |
| "mae": round(float(mean_absolute_error(y_test, preds)), 4), |
| }} |
| |
| |
| def save_bundle(bundle, output_path="model.pkl"): |
| joblib.dump(bundle, output_path) |
| |
| |
| def predict_from_bundle(bundle_path: str, records): |
| bundle = joblib.load(bundle_path) |
| model = bundle["model"] |
| features = bundle["features"] |
| frame = pd.DataFrame(records) |
| frame = frame[features] |
| return model.predict(frame) |
| |
| |
| if __name__ == "__main__": |
| data_path = sys.argv[1] if len(sys.argv) > 1 else "dataset.csv" |
| bundle, metrics, X_test = train_pipeline(load_dataset(data_path)) |
| save_bundle(bundle, "model.pkl") |
| print("Training complete.") |
| print(json.dumps(metrics, indent=2)) |
| if len(X_test): |
| sample = X_test.head(1) |
| pred = predict_from_bundle("model.pkl", sample.to_dict(orient="records")) |
| print("Sample prediction:", pred[0]) |
| ''' |
|
|
|
|
| def _inference_script_content(feature_names, target_name, best_model_name): |
| example_features = { |
| f: "..." for f in (feature_names or [])[: min(6, len(feature_names or []))] |
| } |
| return f'''""" |
| Inference helpers for the exported AutoML model. |
| """ |
| |
| import joblib |
| import pandas as pd |
| |
| |
| raw_obj = joblib.load("model.pkl") |
| if isinstance(raw_obj, dict) and "model" in raw_obj: |
| bundle = raw_obj |
| model = bundle["model"] |
| FEATURE_NAMES = bundle.get("features", {feature_names}) |
| TARGET_NAME = bundle.get("target", "{target_name}") |
| MODEL_TYPE = bundle.get("metadata", {{}}).get("model_type", "{best_model_name}") |
| else: |
| bundle = {{"model": raw_obj, "features": {feature_names}, "target": "{target_name}", "metadata": {{"model_type": "{best_model_name}"}}}} |
| model = raw_obj |
| FEATURE_NAMES = {feature_names} |
| TARGET_NAME = "{target_name}" |
| MODEL_TYPE = "{best_model_name}" |
| |
| |
| def predict_one(record: dict): |
| frame = pd.DataFrame([record], columns=FEATURE_NAMES) |
| pred = model.predict(frame)[0] |
| return float(pred) if hasattr(pred, "item") else pred |
| |
| |
| def predict_many(records): |
| frame = pd.DataFrame(records, columns=FEATURE_NAMES) |
| preds = model.predict(frame) |
| return [float(p) if hasattr(p, "item") else p for p in preds] |
| |
| |
| if __name__ == "__main__": |
| print("Loaded", MODEL_TYPE) |
| print("Expected target:", TARGET_NAME) |
| print("Expected features:", FEATURE_NAMES) |
| print("Example payload:", {example_features}) |
| ''' |
|
|
|
|
| def _requirements_content(best_model_name: str, preprocessor_name: str): |
| deps = [ |
| "pandas>=2.0.0", |
| "numpy>=1.24.0", |
| "scikit-learn>=1.3.0", |
| "joblib>=1.3.0", |
| ] |
|
|
| if preprocessor_name in {"lite_column_transformer", "full_column_transformer"}: |
| deps.append("category-encoders>=2.6.0") |
| if "LightGBM" in best_model_name: |
| deps.append("lightgbm>=4.0.0") |
| if "XGBoost" in best_model_name or "XGB" in best_model_name: |
| deps.append("xgboost>=2.0.0") |
|
|
| deps.extend(["shap>=0.44.0", "optuna>=3.0.0"]) |
| return "\n".join(dict.fromkeys(deps)) + "\n" |
|
|
|
|
| def _explain_script_content(): |
| return '''""" |
| Optional local explanation script. |
| Produces top feature importances for a CSV input. |
| Usage: |
| python explain.py dataset.csv |
| """ |
| |
| import json |
| import sys |
| import joblib |
| import numpy as np |
| import pandas as pd |
| |
| try: |
| import shap |
| except ImportError: |
| shap = None |
| |
| raw_obj = joblib.load("model.pkl") |
| if isinstance(raw_obj, dict) and "model" in raw_obj: |
| bundle = raw_obj |
| model = bundle["model"] |
| FEATURE_NAMES = bundle.get("features", []) |
| else: |
| bundle = {{"model": raw_obj, "features": []}} |
| model = raw_obj |
| FEATURE_NAMES = {feature_names} |
| |
| |
| def main(): |
| if shap is None: |
| print("shap is not installed. Run: pip install shap") |
| return |
| |
| input_path = sys.argv[1] if len(sys.argv) > 1 else "dataset.csv" |
| df = pd.read_csv(input_path) |
| frame = df[FEATURE_NAMES].head(50) |
| |
| pre = model.named_steps["preprocessor"] |
| estimator = model.named_steps["model"] |
| transformed = pre.transform(frame) |
| explainer = shap.Explainer(estimator, transformed) |
| shap_values = explainer(transformed) |
| values = np.abs(shap_values.values).mean(axis=0) |
| if len(values.shape) > 1: |
| values = values.mean(axis=1) |
| |
| names = pre.get_feature_names_out() |
| rows = [ |
| {"feature": str(name), "importance": round(float(val), 6)} |
| for name, val in sorted(zip(names, values), key=lambda x: x[1], reverse=True)[:20] |
| ] |
| print(json.dumps(rows, indent=2)) |
| |
| |
| if __name__ == "__main__": |
| main() |
| ''' |
|
|
|
|
| def _feature_dictionary(schema_data: dict, feature_names, profile: dict): |
| schema = (schema_data or {}).get("schema", {}) |
| col_stats = (profile or {}).get("column_stats", {}) |
| entries = [] |
| for name in feature_names or []: |
| stats = col_stats.get(name, {}) if isinstance(col_stats, dict) else {} |
| spec = schema.get(name, {}) if isinstance(schema, dict) else {} |
| entries.append( |
| { |
| "name": name, |
| "dtype": spec.get("type") or stats.get("dtype", "unknown"), |
| "missing_count": spec.get("missing", stats.get("missing", 0)), |
| "missing_pct": stats.get("missing_pct", 0), |
| "semantic_type": stats.get("semantic_type", "unknown"), |
| "notes": "Required model input field; preserve training-time meaning and units.", |
| } |
| ) |
| return json.dumps(entries, indent=2) |
|
|
|
|
| def _readme_content( |
| best_model_name: str, |
| metric_name: str, |
| score_val, |
| target_name: str, |
| feature_names, |
| execution_profile: dict, |
| tested_models, |
| training_date: str, |
| risks: list, |
| intended_use: str, |
| ): |
| display_score = _display_metric(metric_name, score_val) |
| feature_list = ( |
| "\n".join(f"- `{f}`" for f in (feature_names or [])) or "- *(not available)*" |
| ) |
| tested_lines = [] |
| for item in tested_models[:10] if isinstance(tested_models, list) else []: |
| line = ( |
| f"- `{item.get('model', 'Unknown')}` | status={item.get('status', 'n/a')}" |
| ) |
| if item.get("sweep_score") is not None: |
| line += f" | sweep={item.get('sweep_score')}%" |
| if item.get("best_cv_score") is not None: |
| line += f" | best_cv={item.get('best_cv_score')}%" |
| if item.get("holdout_score") is not None: |
| line += f" | holdout={item.get('holdout_score')}%" |
| tested_lines.append(line) |
| tested_block = "\n".join(tested_lines) if tested_lines else "- *(not available)*" |
|
|
| return f"""# AutoML Export Bundle |
| |
| ## Summary |
| | Field | Value | |
| |---|---| |
| | Best model | {best_model_name} | |
| | Target | {target_name} | |
| | Metric | {metric_name} | |
| | Score | {display_score} | |
| |
| ## Execution Profile |
| ```json |
| {json.dumps(execution_profile or {{}}, indent=2)} |
| ``` |
| |
| ## Exported Files |
| - `model.pkl`: trained pipeline bundle ready for inference |
| - `training.py`: reproducible end-to-end training script |
| - `inference.py`: small prediction helper |
| - `model_metadata.json`: saved model metadata from the run |
| - `metrics.json`: result payload from the run |
| - `schema.json`: saved data contract / schema snapshot when available |
| |
| ## Feature List |
| {feature_list} |
| |
| ## Tested Models |
| {tested_block} |
| |
| ## Model Card |
| | Field | Value | |
| |---|---| |
| | Training date | {training_date} | |
| | Intended use | {intended_use} | |
| | Key risks | {"; ".join(risks) if risks else "Model quality depends on data drift, missing-value patterns, and input consistency."} | |
| |
| ## Quick Start |
| 1. Install dependencies: `pip install pandas scikit-learn joblib category_encoders lightgbm xgboost` |
| 2. Retrain on a dataset: `python training.py path/to/dataset.csv` |
| 3. Load the trained bundle in Python with `joblib.load("model.pkl")` |
| """ |
|
|
|
|
| def create_export_bundle(job_id: str, results: dict) -> str: |
| """ |
| Creates a ZIP bundle containing the trained model and reproducible code. |
| """ |
| export_dir = os.path.join("tmp", "exports") |
| os.makedirs(export_dir, exist_ok=True) |
|
|
| run_dir = get_run_dir(job_id) |
| metrics_path = os.path.join(run_dir, "logs", "metrics.json") |
| schema_path = os.path.join(run_dir, "data", "schema.json") |
| metadata_path = os.path.join(run_dir, "artifacts", "model_metadata.json") |
| model_path = resolve_model_path(job_id) |
|
|
| model_metadata = _safe_json_load(metadata_path) |
| saved_metrics = _safe_json_load(metrics_path) |
| schema_data = _safe_json_load(schema_path) |
|
|
| merged_results = dict(saved_metrics) if isinstance(saved_metrics, dict) else {} |
| merged_results.update(results or {}) |
|
|
| feature_names = merged_results.get("feature_names", []) or model_metadata.get( |
| "feature_names", [] |
| ) |
| target_name = merged_results.get("target", "target") |
| best_model_name = merged_results.get("best_model", "Unknown") |
| is_clf = bool(merged_results.get("is_classification", True)) |
| metric_name = merged_results.get("metric_name", "Score") |
| score_val = merged_results.get("score", "N/A") |
| execution_profile = merged_results.get("execution_profile", {}) |
| tested_models = merged_results.get("tested_models", []) |
| preprocessor_name = model_metadata.get("preprocessor", "lite_column_transformer") |
| training_date = time.strftime( |
| "%Y-%m-%d", time.localtime(model_metadata.get("timestamp", time.time())) |
| ) |
|
|
| profile = {} |
| dataset_path = "" |
| job_snapshot = {} |
| try: |
| from infra.database import get_db, JobModel, DatasetModel |
|
|
| with get_db() as db: |
| job = db.query(JobModel).filter(JobModel.id == job_id).first() |
| if job: |
| try: |
| job_snapshot = { |
| "story": job.story, |
| "insights": ( |
| json.loads(job.insights_json) if job.insights_json else {} |
| ), |
| "reasoning": ( |
| json.loads(job.reasoning_json) if job.reasoning_json else [] |
| ), |
| "params": ( |
| json.loads(job.params_json) if job.params_json else {} |
| ), |
| } |
| except Exception: |
| job_snapshot = { |
| "story": job.story, |
| "insights": {}, |
| "reasoning": [], |
| "params": {}, |
| } |
| dataset = ( |
| db.query(DatasetModel) |
| .filter(DatasetModel.id == job.dataset_id) |
| .first() |
| ) |
| if dataset: |
| dataset_path = dataset.file_path or "" |
| try: |
| profile = ( |
| json.loads(dataset.profile_json) |
| if dataset.profile_json |
| else {} |
| ) |
| except Exception: |
| profile = {} |
| except Exception: |
| profile = {} |
| dataset_path = "" |
|
|
| risks = [] |
| execution_risks = [] |
| if execution_profile.get("use_full_preprocessor"): |
| execution_risks.append( |
| "Full-mode preprocessing may be more sensitive to schema drift and dependency mismatch." |
| ) |
| if (profile or {}).get("missing_pct", 0): |
| execution_risks.append( |
| "Input data with different missing-value patterns may degrade performance." |
| ) |
| if (profile or {}).get("imbalance") == "High ⚠️": |
| execution_risks.append( |
| "Class imbalance in production can affect recall and calibration." |
| ) |
| risks = execution_risks[:3] |
|
|
| zip_path = os.path.join(export_dir, f"{job_id}_bundle.zip") |
|
|
| temp_files = { |
| "training.py": _training_script_content( |
| best_model_name=best_model_name, |
| target_name=target_name, |
| feature_names=feature_names, |
| is_clf=is_clf, |
| metric_name=metric_name, |
| execution_profile=execution_profile, |
| preprocessor_name=preprocessor_name, |
| ), |
| "inference.py": _inference_script_content( |
| feature_names, target_name, best_model_name |
| ), |
| "requirements.txt": _requirements_content(best_model_name, preprocessor_name), |
| "feature_dictionary.json": _feature_dictionary( |
| schema_data, feature_names, profile |
| ), |
| "README.md": _readme_content( |
| best_model_name=best_model_name, |
| metric_name=metric_name, |
| score_val=score_val, |
| target_name=target_name, |
| feature_names=feature_names, |
| execution_profile=execution_profile, |
| tested_models=tested_models, |
| training_date=training_date, |
| risks=risks, |
| intended_use="Batch or API inference on datasets with the same feature semantics and preprocessing assumptions as training.", |
| ), |
| "import_bundle.json": json.dumps( |
| { |
| "format": "automl_studio_export_bundle_v1", |
| "job_id": job_id, |
| "profile": profile, |
| "results": merged_results, |
| "story": job_snapshot.get("story"), |
| "insights": job_snapshot.get("insights", {}), |
| "reasoning": job_snapshot.get("reasoning", []), |
| "params": job_snapshot.get("params", {}), |
| }, |
| indent=2, |
| ), |
| } |
|
|
| generated_paths = [] |
| for filename, content in temp_files.items(): |
| path = os.path.join("tmp", f"{job_id}_{filename}") |
| with open(path, "w") as f: |
| f.write(content) |
| generated_paths.append((path, filename)) |
|
|
| with zipfile.ZipFile(zip_path, "w") as zipf: |
| if model_path and os.path.exists(model_path): |
| zipf.write(model_path, arcname="model.pkl") |
|
|
| if os.path.exists(metadata_path): |
| zipf.write(metadata_path, arcname="model_metadata.json") |
| if os.path.exists(metrics_path): |
| zipf.write(metrics_path, arcname="metrics.json") |
| if os.path.exists(schema_path): |
| zipf.write(schema_path, arcname="schema.json") |
| if dataset_path and os.path.exists(dataset_path): |
| zipf.write(dataset_path, arcname="training_dataset.csv") |
|
|
| for path, arcname in generated_paths: |
| zipf.write(path, arcname=arcname) |
|
|
| return zip_path |
|
|
|
|
| def cleanup_old_exports(max_age_hours: int = 24): |
| """Remove tmp/ artifacts older than max_age_hours. Call periodically.""" |
| now = time.time() |
| for folder in ["tmp", "tmp/exports"]: |
| if not os.path.exists(folder): |
| continue |
| for fname in os.listdir(folder): |
| fpath = os.path.join(folder, fname) |
| if os.path.isfile(fpath): |
| age_hours = (now - os.path.getmtime(fpath)) / 3600 |
| if age_hours > max_age_hours: |
| try: |
| os.remove(fpath) |
| except Exception: |
| pass |
|
|
