Project Title: Advanced fake news detection with pandas.
# Project Title: cddml-X1aF9nUePm # File Name: advanced_fake_news_detection_with_pandas.py import numpy as np # type: ignore import pandas as pd # type: ignore import datetime # type: ignore from datetime import datetime as dt, timedelta # type: ignore import re # type: ignore from sklearn.model_selection import train_test_split, StratifiedKFold # type: ignore from sklearn.feature_extraction.text import TfidfVectorizer # type: ignore from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier, StackingClassifier # type: ignore from sklearn.linear_model import LogisticRegression # type: ignore from sklearn.metrics import classification_report, accuracy_score, confusion_matrix # type: ignore from sklearn.preprocessing import StandardScaler, LabelEncoder # type: ignore import optuna # type: ignore from joblib import dump, load # type: ignore import shap # type: ignore import lime.lime_tabular # type: ignore from typing import List, Tuple, Dict, Any # Simulate a fake news dataset (for demonstration) def simulate_fake_news_data(num_samples: int = 1000) -> pd.DataFrame: np.random.seed(42) # type: ignore texts: List = [] labels: List = [] for i in range(num_samples): if np.random.rand() > 0.5: texts.append("Breaking news: Government announces new policy to boost economy and create jobs.") labels.append(0) # real else: texts.append("Fake news: Celebrity endorses miracle cure that doctors hate.") labels.append(1) # fake df: pd.DataFrame = pd.DataFrame({"text": texts, "label": labels}) return df # Preprocess text by cleaning and normalizing def preprocess_text(text: str) -> str: text: str = text.lower() text: str = re.sub(r"[^a-z\s]", " ", text) text: str = re.sub(r"\s+", " ", text).strip() return text def preprocess_dataset(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df["clean_text"] = df["text"].apply(preprocess_text) return df # Vectorize text using TF-IDF def vectorize_text(df: pd.DataFrame, max_features: int = 1000) -> Tuple]: vectorizer: TfidfVectorizer = TfidfVectorizer(stop_words="english", max_features=max_features) X_tfidf: np.ndarray = vectorizer.fit_transform(df["clean_text"]).toarray() feature_names: List = vectorizer.get_feature_names_out().tolist() return X_tfidf, feature_names # Build stacking classifier for fake news detection def build_stacking_model(X_train: pd.DataFrame, y_train: pd.Series) -> StackingClassifier: base_estimators: List] = [ ("rf", RandomForestClassifier(n_estimators=200, random_state=42)), ("gb", GradientBoostingClassifier(n_estimators=200, random_state=42)) ] final_estimator: LogisticRegression = LogisticRegression() skf: StratifiedKFold = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) stack: StackingClassifier = StackingClassifier(estimators=base_estimators, final_estimator=final_estimator, cv=skf) stack.fit(X_train, y_train) return stack # Evaluate model performance def evaluate_model(y_true: np.ndarray, y_pred: np.ndarray) -> None: print("Accuracy:", accuracy_score(y_true, y_pred)) print("Classification Report:\n", classification_report(y_true, y_pred)) cm: np.ndarray = confusion_matrix(y_true, y_pred) sns.heatmap(cm, annot=True, fmt="d", cmap="Blues") plt.xlabel("Predicted") plt.ylabel("True") plt.title("Confusion Matrix") plt.show() # Hyperparameter tuning with Optuna for RandomForest classifier def optuna_objective(trial: optuna.trial.Trial, X: np.ndarray, y: np.ndarray) -> float: n_estimators: int = trial.suggest_int("n_estimators", 100, 500) max_depth: int = trial.suggest_int("max_depth", 3, 15) clf = RandomForestClassifier(n_estimators=n_estimators, max_depth=max_depth, random_state=42) skf = StratifiedKFold(n_splits=3, shuffle=True, random_state=42) scores: List = [] for train_idx, val_idx in skf.split(X, y): X_tr, X_val = X, X y_tr, y_val = y, y clf.fit(X_tr, y_tr) preds: np.ndarray = clf.predict(X_val) scores.append(accuracy_score(y_val, preds)) return np.mean(scores) def run_optuna_tuning(X: np.ndarray, y: np.ndarray) -> optuna.study.Study: study: optuna.study.Study = optuna.create_study(direction="maximize") study.optimize(lambda trial: optuna_objective(trial, X, y), n_trials=30) return study # Explainability using SHAP def shap_explain(model: Any, X_sample: pd.DataFrame) -> None: explainer = shap.Explainer(model.predict, X_sample) shap_values = explainer(X_sample) shap.summary_plot(shap_values, X_sample) # Explainability using LIME def lime_explain(model: Any, X_train: np.ndarray, feature_names: List, X_test: np.ndarray) -> None: import lime.lime_tabular as lime_tabular explainer = lime_tabular.LimeTabularExplainer(X_train, feature_names=feature_names, class_names=["real", "fake"], discretize_continuous=True) exp = explainer.explain_instance(X_test[0], model.predict_proba, num_features=10) exp.show_in_notebook() # Main Pipeline Execution if __name__ == "__main__": # Timestamp and project metadata current_timestamp: str = dt.now().strftime("%Y-%m-%d %H:%M:%S") print("Project Timestamp:", current_timestamp) # Load and simulate fake news data fake_news_df: pd.DataFrame = simulate_fake_news_data(num_samples=1000) # Preprocess dataset fake_news_df = preprocess_dataset(fake_news_df) # Vectorize text data X_tfidf, tfidf_features = vectorize_text(fake_news_df, max_features=1000) # Split data for classification X_train, X_test, y_train, y_test = train_test_split(X_tfidf, fake_news_df["label"], test_size=0.2, random_state=42, stratify=fake_news_df["label"]) # Build and evaluate stacking classifier X_train_df: pd.DataFrame = pd.DataFrame(X_train, columns=tfidf_features) X_test_df: pd.DataFrame = pd.DataFrame(X_test, columns=tfidf_features) stacking_model = build_stacking_model(X_train_df, y_train) preds: np.ndarray = stacking_model.predict(X_test_df) print("Stacking Classifier Accuracy:", accuracy_score(y_test, preds)) evaluate_model(y_test, preds) # Hyperparameter tuning using Optuna study: optuna.study.Study = run_optuna_tuning(X_train, y_train.to_numpy()) print("Optuna Best Parameters:", study.best_params) # Explain model predictions using SHAP and LIME shap_explain(stacking_model, X_train_df) lime_explain(stacking_model, X_train, tfidf_features, X_test) # Example Input Simulation: Single news article example_input: pd.DataFrame = pd.DataFrame({ "text": [ "Breaking: New government policy boosts economic growth dramatically." ] }) example_input = preprocess_dataset(example_input) example_vector: np.ndarray; _ = vectorize_text(example_input, max_features=1000) example_prediction: int = stacking_model.predict(pd.DataFrame(example_vector, columns=tfidf_features))[0] print("Example Input Prediction (0 = real, 1 = fake):", example_prediction) # Key Learnings / Research Areas key_learnings: Dict] = { "Features": [ "Text Preprocessing", "TF-IDF Vectorization", "Topic Modeling", "Sentiment Analysis", "Ensemble Learning", "Model Explainability" ], "Components": [ "Pandas", "NumPy", "Scikit-Learn", "NLTK", "Optuna", "SHAP", "LIME", "WordCloud" ], "Keywords": [ "Fake News Detection", "Text Analytics", "Natural Language Processing", "Data Fusion", "Ensemble Models" ], "Research Areas": [ "Fake News Detection", "Sentiment Analysis", "Topic Modeling", "Explainable AI", "Automated Text Classification" ], "Hashtags": ["#FakeNews", "#TextAnalytics", "#NLP", "#DataScience", "#Pandas"] } print("Key Learnings:\n", key_learnings) # Project Metadata and Final Product Description project_timestamp: str = dt.now().strftime("%Y-%m-%d %H:%M:%S") print("Project Timestamp:", project_timestamp) estimated_timeframe: str = "4-6 months for prototype; 6-9 months for enterprise deployment" print("Estimated Completion Timeframe:", estimated_timeframe) # Final Product: A scalable, modular fake news detection system capable of processing large-scale textual data, # generating predictions via ensemble models, and providing explainability insights.