import streamlit as st import numpy as np import pandas as pd import pickle #from keras.models import model_from_json #from keras.optimizers import Adam # ───────────────────────────────────────────────────────────────────────────── # Neural Network Model Laden (aus JSON + Weights) # ───────────────────────────────────────────────────────────────────────────── def load_nn_model(config_path, weights_path): with open(config_path, "r") as f: model_json = f.read() model = model_from_json(model_json) model.load_weights(weights_path) model.compile(optimizer=Adam(), loss="mse", metrics=["mse", "mae"]) return model # ───────────────────────────────────────────────────────────────────────────── # SVD Model & Trainset Laden # ───────────────────────────────────────────────────────────────────────────── @st.cache_resource def load_svd_model(path="models/svd_model.pkl"): with open(path, "rb") as f: return pickle.load(f) @st.cache_resource def load_trainset(path="models/trainset.pkl"): with open(path, "rb") as f: return pickle.load(f) # ───────────────────────────────────────────────────────────────────────────── # Encodings (dict mit user/movie Encodings) laden # ───────────────────────────────────────────────────────────────────────────── def load_encodings(path="encodings.pkl"): with open(path, "rb") as f: return pickle.load(f) # ───────────────────────────────────────────────────────────────────────────── # SVD Recommendation # ───────────────────────────────────────────────────────────────────────────── def fold_in_new_user(model, trainset, user_ratings, reg=5): n_factors = model.n_factors A = np.zeros((n_factors, n_factors)) b = np.zeros(n_factors) bias_numerator, bias_denominator = 0, 0 global_mean = trainset.global_mean for movie_id, rating in user_ratings.items(): try: movie_id = int(movie_id) inner_iid = trainset.to_inner_iid(movie_id) qi = model.qi[inner_iid] bi = model.bi[inner_iid] A += np.outer(qi, qi) b += qi * (rating - global_mean - bi) bias_numerator += (rating - global_mean - bi) bias_denominator += 1 except ValueError: continue A += reg * np.eye(n_factors) new_user_factors = np.linalg.solve(A, b) if np.linalg.det(A) != 0 else np.linalg.pinv(A) @ b new_user_bias = bias_numerator / bias_denominator if bias_denominator > 0 else 0 return new_user_factors, new_user_bias def recommend_with_svd(model, trainset, ratings_df, user_ratings, top_n=10): new_user_factors, new_user_bias = fold_in_new_user(model, trainset, user_ratings) global_mean = trainset.global_mean movie_predictions = [] for inner_iid in range(trainset.n_items): movie_id = trainset.to_raw_iid(inner_iid) if movie_id not in user_ratings: pred = global_mean + new_user_bias + model.bi[inner_iid] + np.dot(new_user_factors, model.qi[inner_iid]) movie_predictions.append({'movieId': int(movie_id), 'rating': pred}) df = pd.DataFrame(movie_predictions) df["rating"] = df["rating"].round(4) # optional: Ratings runden für mehr Stabilität df = df.sort_values(["rating", "movieId"], ascending=[False, True]).head(top_n) return df # ───────────────────────────────────────────────────────────────────────────── # Neural Network Recommendation # ───────────────────────────────────────────────────────────────────────────── def recommend_with_nn(user_ratings, model, available_movies, top_n=10): """ Args: user_ratings: dict of movieId → rating model: compiled Keras model available_movies: list of movieIds top_n: number of recommendations Returns: DataFrame with movieId and predicted rating """ if not available_movies: return pd.DataFrame(columns=["movieId", "rating"]) user_id = max(user_ratings.keys(), default=0) + 100000 # Dummy new user user_vector = np.array([user_id] * len(available_movies)) movie_vector = np.array(available_movies) predictions = model.predict([user_vector, movie_vector], verbose=0) df = pd.DataFrame({ 'movieId': available_movies, 'rating': predictions.flatten() }) df = df[~df["movieId"].isin(user_ratings.keys())] return df.sort_values("rating", ascending=False).head(top_n)