{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceType":"competition","sourceId":7391,"databundleVersionId":44328}],"dockerImageVersionId":31328,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col\n\n# init spark\nspark = SparkSession.builder \\\n    .appName(\"IE212_EDA\") \\\n    .config(\"spark.sql.shuffle.partitions\", \"200\") \\\n    .config(\"spark.driver.memory\", \"16g\") \\\n    .getOrCreate()\n\nspark.sparkContext.setLogLevel(\"ERROR\")\n\n# new path for 27 stores\nparquet_path = \"/kaggle/working/train_filtered_27_stores.parquet\"\n\n# if not exists, extract, read raw CSV, filter, and save to parquet\nif not os.path.exists(parquet_path):\n    print(\"Parquet not found. Extracting and processing raw CSV...\")\n    if not os.path.exists(\"/kaggle/working/train.csv\"):\n        !7z x /kaggle/input/competitions/favorita-grocery-sales-forecasting/train.csv.7z -o/kaggle/working/ -y\n    \n    # load raw data\n    df_raw = spark.read.csv(\"/kaggle/working/train.csv\", header=True, inferSchema=True)\n    \n    # early filtering: keep 1 year + 27 stores\n    valid_stores = [i for i in range(1, 28)]\n    df_filtered = df_raw.filter(\n        (col(\"date\") >= \"2016-08-01\") & \n        (col(\"date\") <= \"2017-08-15\") &\n        (col(\"store_nbr\").isin(valid_stores)) \n    )\n    \n    # execute and save to parquet\n    df_filtered.write.mode(\"overwrite\").parquet(parquet_path)\n    print(\"Saved to Parquet for future runs!\")\n\n# CRITICAL FIX: Cut the lineage! Read directly from the highly compressed Parquet.\nprint(\"Loading filtered data from Parquet...\")\ndf_train = spark.read.parquet(parquet_path)\n\n# cache to ram for faster joins later\ndf_train.cache()\n\n# check rows (now it counts from Parquet instantly, no Py4JError)\nprint(f\"Total rows: {df_train.count()}\")\ndf_train.printSchema()","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2026-05-31T00:19:56.795316Z","iopub.execute_input":"2026-05-31T00:19:56.795649Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import col, count, when, isnan\nexprs = []\nfor c, t in df_train.dtypes:\n    if t in ('double', 'float'):\n        exprs.append(count(when(isnan(col(c)) | col(c).isNull(), c)).alias(c))\n    else:\n        exprs.append(count(when(col(c).isNull(), c)).alias(c))\n#count missing values\nprint(\"Missing Values in Train:\")\ndf_train.select(exprs).show()\n\n# check unique values in categorical column\nprint(\"Categorical distribution (onpromotion):\")\ndf_train.groupBy(\"onpromotion\").count().show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Stats basics\nprint(\"Stats unit_sales:\")\ndf_train.select(\"unit_sales\").summary(\"count\", \"mean\", \"stddev\", \"min\", \"max\").show()\n\n# negative sale\nnegative_sales = df_train.filter(col(\"unit_sales\") < 0).count()\nprint(f\"Negative sales: {negative_sales}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\n\n# extract files if not exist\naux_files = ['oil.csv.7z', 'holidays_events.csv.7z', 'items.csv.7z', 'stores.csv.7z']\nfor f in aux_files:\n    csv_name = f.replace('.7z', '')\n    if not os.path.exists(f\"/kaggle/working/{csv_name}\"):\n        print(f\"Extracting {f}...\")\n        os.system(f\"7z x /kaggle/input/competitions/favorita-grocery-sales-forecasting/{f} -o/kaggle/working/ -y\")\n\n# load macro data\ndf_oil = spark.read.csv(\"/kaggle/working/oil.csv\", header=True, inferSchema=True)\ndf_hol = spark.read.csv(\"/kaggle/working/holidays_events.csv\", header=True, inferSchema=True)\n\n# check missing values in oil \nprint(\"Missing values in Oil:\")\ndf_oil.select([count(when(col(c).isNull(), c)).alias(c) for c in df_oil.columns]).show()\n\n# check holiday types & transferred flags\nprint(\"Holiday types distribution:\")\ndf_hol.groupBy(\"type\", \"transferred\").count().show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.window import Window\nfrom pyspark.sql.functions import last, first, to_date\n\n# --- 1. PROCESS OIL  ---\ndf_oil = spark.read.csv(\"/kaggle/working/oil.csv\", header=True, inferSchema=True)\ndf_oil = df_oil.withColumn(\"date\", to_date(col(\"date\")))\n\n# TDense Calendar from Train \ndf_dates = df_train.select(\"date\").distinct()\ndf_oil_dense = df_dates.join(df_oil, \"date\", \"left\")\n\n# ffill n bfill \nwindow_ffill = Window.orderBy(\"date\").rowsBetween(Window.unboundedPreceding, 0)\nwindow_bfill = Window.orderBy(\"date\").rowsBetween(0, Window.unboundedFollowing)\n\ndf_oil_dense = df_oil_dense.withColumn(\"dcoilwtico\", last(\"dcoilwtico\", ignorenulls=True).over(window_ffill))\ndf_oil_dense = df_oil_dense.withColumn(\"dcoilwtico\", first(\"dcoilwtico\", ignorenulls=True).over(window_bfill))\n\n# --- 2. PROCESS HOLIDAYS  ---\ndf_hol = spark.read.csv(\"/kaggle/working/holidays_events.csv\", header=True, inferSchema=True)\ndf_hol = df_hol.withColumn(\"date\", to_date(col(\"date\")))\ndf_hol = df_hol.filter(col(\"transferred\") == False)\n\n# Drop duplicate\ndf_hol = df_hol.dropDuplicates([\"date\"])\n\ndf_hol = df_hol.withColumnRenamed(\"type\", \"holiday_type\")\ndf_hol = df_hol.select(\"date\", \"holiday_type\", \"locale\")\n\nprint(\"Oil and Holidays processed.\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import broadcast, when, col\n\ndf_items = spark.read.csv(\"/kaggle/working/items.csv\", header=True, inferSchema=True)\ndf_stores = spark.read.csv(\"/kaggle/working/stores.csv\", header=True, inferSchema=True)\n\n# CRITICAL FIX\ndf_train_clipped = df_train.withColumn(\"unit_sales\", when(col(\"unit_sales\") < 0, 0.0).otherwise(col(\"unit_sales\")))\n\n# Join data sử dụng bản df_train_clipped\ndf_base = df_train_clipped.join(broadcast(df_stores), \"store_nbr\", \"left\") \\\n                  .join(broadcast(df_items), \"item_nbr\", \"left\") \\\n                  .join(broadcast(df_oil_dense), \"date\", \"left\") \\\n                  .join(broadcast(df_hol), \"date\", \"left\")\n\n# Fill null for normal day\ndf_base = df_base.fillna({\"holiday_type\": \"Normal\", \"locale\": \"National\"})\n\nprint(f\"Total rows after joins: {df_base.count()}\")\nprint(\"Check nulls :\")\ndf_base.select([count(when(col(c).isNull(), c)).alias(c) for c in [\"dcoilwtico\", \"holiday_type\"]]).show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!pip install pykalman -q\n\nimport pandas as pd\nimport matplotlib.pyplot as plt\nfrom pykalman import KalmanFilter\nfrom pyspark.sql.functions import col\n\n# enable arrow for faster pandas udf\nspark.conf.set(\"spark.sql.execution.arrow.pyspark.enabled\", \"true\")\nspark.conf.set(\"spark.sql.execution.arrow.maxRecordsPerBatch\", \"10000\")\n\n# build safe ddl schema\nschema_ddl = \", \".join([f\"{f.name} {f.dataType.simpleString()}\" for f in df_base.schema.fields]) + \", kalman_sales float\"\n\ndef apply_kalman_grouped(pdf: pd.DataFrame) -> pd.DataFrame:\n    pdf = pdf.sort_values(\"date\")\n    if len(pdf) < 2:\n        pdf[\"kalman_sales\"] = pdf[\"unit_sales\"]\n        return pdf\n        \n    kf = KalmanFilter(transition_matrices=[1], observation_matrices=[1],\n                      initial_state_mean=pdf[\"unit_sales\"].iloc[0], initial_state_covariance=1,\n                      observation_covariance=5, transition_covariance=0.1)\n    \n    state_means, _ = kf.filter(pdf[\"unit_sales\"].values)\n    pdf[\"kalman_sales\"] = state_means.flatten()\n    return pdf\n\n# apply udf with repartition\ndf_kalman = df_base.repartition(200, \"store_nbr\", \"item_nbr\") \\\n    .groupBy(\"store_nbr\", \"item_nbr\") \\\n    .applyInPandas(apply_kalman_grouped, schema=schema_ddl)\n\n# new path for 27 stores version\noutput_path = \"/kaggle/working/base_processed_27_stores.parquet\"\ndf_kalman.write.mode(\"overwrite\").parquet(output_path)\n\n# check plot\ndf_check = spark.read.parquet(output_path)\npdf_sample = df_check.filter((col(\"store_nbr\") == 1) & (col(\"item_nbr\") == 1503844)).orderBy(\"date\").toPandas()\n\nif not pdf_sample.empty:\n    plt.figure(figsize=(14, 5))\n    plt.plot(pdf_sample['date'], pdf_sample['unit_sales'], label='raw sales', alpha=0.5, color='gray')\n    plt.plot(pdf_sample['date'], pdf_sample['kalman_sales'], label='kalman', color='red', linewidth=2)\n    plt.title('kalman denoising (store 1, item 1503844)')\n    plt.xlabel('date')\n    plt.ylabel('sales')\n    plt.legend()\n    plt.grid(True)\n    # save plot with new name\n    plt.savefig(\"/kaggle/working/kalman_report_27_stores.png\", dpi=300, bbox_inches='tight')\n    plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!pip install PyWavelets -q\n\nimport pandas as pd\nimport pywt\nimport numpy as np\nfrom pyspark.sql.functions import col\nimport warnings\n\n# read from the new kalman parquet (27 stores)\ndf_base = spark.read.parquet(\"/kaggle/working/base_processed_27_stores.parquet\")\n\n# build safe ddl schema for wavelet\nschema_wavelet = \", \".join([f\"{f.name} {f.dataType.simpleString()}\" for f in df_base.schema.fields]) + \", wavelet_soft_sales float, wavelet_hard_sales float\"\n\ndef apply_wavelet_grouped(pdf: pd.DataFrame) -> pd.DataFrame:\n    warnings.filterwarnings(\"ignore\", category=UserWarning, module=\"pywt\")\n    pdf = pdf.sort_values(\"date\")\n    if len(pdf) < 2:\n        pdf[\"wavelet_soft_sales\"] = pdf[\"unit_sales\"]\n        pdf[\"wavelet_hard_sales\"] = pdf[\"unit_sales\"]\n        return pdf\n        \n    signal = pdf[\"unit_sales\"].values\n    wavelet = 'db4'\n    level = 2\n    # check length for wavelet level\n    if len(signal) < 2**level:\n        level = 1\n        if len(signal) < 2: level = 0\n    \n    coeffs = pywt.wavedec(signal, wavelet, level=level)\n    sigma = np.median(np.abs(coeffs[-1])) / 0.6745\n    uthresh = sigma * np.sqrt(2 * np.log(len(signal)))\n    \n    coeffs_soft = [pywt.threshold(c, value=uthresh, mode='soft') for c in coeffs]\n    reconstructed_soft = pywt.waverec(coeffs_soft, wavelet)\n    \n    coeffs_hard = [pywt.threshold(c, value=uthresh, mode='hard') for c in coeffs]\n    reconstructed_hard = pywt.waverec(coeffs_hard, wavelet)\n    \n    pdf[\"wavelet_soft_sales\"] = reconstructed_soft[:len(signal)]\n    pdf[\"wavelet_hard_sales\"] = reconstructed_hard[:len(signal)]\n    \n    return pdf\n\ndf_wavelet = df_base.repartition(200, \"store_nbr\", \"item_nbr\") \\\n    .groupBy(\"store_nbr\", \"item_nbr\") \\\n    .applyInPandas(apply_wavelet_grouped, schema=schema_wavelet)\n\n# new path for wavelet output\nout_wavelet = \"/kaggle/working/features_wavelet_27_stores.parquet\"\ndf_wavelet.write.mode(\"overwrite\").parquet(out_wavelet)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.window import Window\nfrom pyspark.sql.functions import lag, mean, stddev, coalesce, lit, datediff\n\ndf_features = spark.read.parquet(\"/kaggle/working/features_wavelet_27_stores.parquet\")\n\n# create time_idx for TFT model (Track 3)\ndf_features = df_features.withColumn(\"time_idx\", datediff(col(\"date\"), lit(\"2016-08-01\")))\n\nwindow_ts = Window.partitionBy(\"store_nbr\", \"item_nbr\").orderBy(\"date\")\n\n# generate lags for all signal variations (raw, kalman, wavelet)\ntarget_cols = [\"unit_sales\", \"kalman_sales\", \"wavelet_soft_sales\"]\nfor t_col in target_cols:\n    df_features = df_features.withColumn(f\"lag_1_{t_col}\", lag(t_col, 1).over(window_ts))\n    df_features = df_features.withColumn(f\"lag_7_{t_col}\", lag(t_col, 7).over(window_ts))\n    df_features = df_features.withColumn(f\"lag_16_{t_col}\", lag(t_col, 16).over(window_ts))\n\nwindow_roll_7 = window_ts.rowsBetween(-7, -1)\nwindow_roll_30 = window_ts.rowsBetween(-30, -1)\n\n# rolling features on raw sales (baseline)\ndf_features = df_features.withColumn(\"rolling_mean_7\", mean(\"unit_sales\").over(window_roll_7))\ndf_features = df_features.withColumn(\"rolling_std_7\", coalesce(stddev(\"unit_sales\").over(window_roll_7), lit(0.0)))\ndf_features = df_features.withColumn(\"rolling_mean_30\", mean(\"unit_sales\").over(window_roll_30))\n\nout_master = \"/kaggle/working/engineered_features_27_stores.parquet\"\ndf_features.write.mode(\"overwrite\").parquet(out_master)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.feature import StringIndexer\n\n# read from the 27 stores master data\ndf_features = spark.read.parquet(\"/kaggle/working/engineered_features_27_stores.parquet\")\n\n# string indexer for categorical columns\ncat_cols = [\"city\", \"state\", \"type\", \"cluster\", \"family\", \"class\", \"holiday_type\", \"locale\"]\nout_cols = [c + \"_idx\" for c in cat_cols]\nindexer = StringIndexer(inputCols=cat_cols, outputCols=out_cols, handleInvalid=\"keep\")\ndf_features = indexer.fit(df_features).transform(df_features)\n\n# drop strings to save ram\ndf_features = df_features.drop(*cat_cols)\n\n# new path for final encoded data (27 stores)\nout_final = \"/kaggle/working/engineered_features_encoded_27_stores.parquet\"\ndf_features.repartition(200, \"store_nbr\", \"item_nbr\").write.mode(\"overwrite\").parquet(out_final)\nprint(f\"saved final encoded data to {out_final}\")\n\n# check final schema\ndf_features.printSchema()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport seaborn as sns\nfrom pyspark.sql.functions import col, round, when\nfrom pyspark.ml.feature import VectorAssembler\nfrom pyspark.ml.stat import Correlation\n\n# read raw data from 27 stores version\ndf_raw = spark.read.parquet(\"/kaggle/working/train_filtered_27_stores.parquet\")\n\n# clip negatives to 0 on spark\ndf_clipped = df_raw.withColumn(\"clipped_sales\", when(col(\"unit_sales\") < 0, 0).otherwise(col(\"unit_sales\")))\n\n# create int bins for plotting\ndf_bins_raw = df_raw.withColumn(\"bin\", round(col(\"unit_sales\"), 0))\ndf_bins_clipped = df_clipped.withColumn(\"bin\", round(col(\"clipped_sales\"), 0))\n\n# filter range and aggregate exact counts (very safe for RAM)\npdf_raw_counts = df_bins_raw.filter(col(\"bin\").between(-20, 50)) \\\n                            .groupBy(\"bin\").count().orderBy(\"bin\").toPandas()\n\npdf_clipped_counts = df_bins_clipped.filter(col(\"bin\").between(-20, 50)) \\\n                                    .groupBy(\"bin\").count().orderBy(\"bin\").toPandas()\n\n# plot 1: dist comparison\nfig, axes = plt.subplots(1, 2, figsize=(15, 5))\n\naxes[0].bar(pdf_raw_counts['bin'], pdf_raw_counts['count'], color='gray', width=1.0)\naxes[0].set_title('Raw Distribution (Exact Counts)')\naxes[0].set_xlim(-20, 50)\naxes[0].set_yscale('log') \n\naxes[1].bar(pdf_clipped_counts['bin'], pdf_clipped_counts['count'], color='blue', width=1.0)\naxes[1].set_title('Clipped Distribution (Negatives mapped to 0)')\naxes[1].set_xlim(-20, 50)\naxes[1].set_yscale('log')\n\nplt.tight_layout()\n# save plot with new name\nplt.savefig(\"/kaggle/working/sales_distribution_exact_27_stores.png\", dpi=300)\nplt.show()\n\n# load processed data for macro corr (27 stores)\ndf_base = spark.read.parquet(\"/kaggle/working/base_processed_27_stores.parquet\")\n\n# create binary hol flag and drop nulls\ndf_corr = df_base.withColumn(\"is_hol\", when(col(\"holiday_type\") != \"Normal\", 1).otherwise(0))\ndf_corr = df_corr.na.drop(subset=[\"unit_sales\", \"dcoilwtico\", \"is_hol\", \"onpromotion\"])\n\n# vector assemble\nassembler = VectorAssembler(inputCols=[\"unit_sales\", \"dcoilwtico\", \"is_hol\", \"onpromotion\"], outputCol=\"features\")\ndf_vec = assembler.transform(df_corr).select(\"features\")\n\n# compute pearson matrix\ncorr_mat = Correlation.corr(df_vec, \"features\", \"pearson\").collect()[0][0].toArray()\n\n# plot 2: heatmap\nplt.figure(figsize=(6, 5))\ncols = [\"Sales\", \"Oil\", \"Holiday\", \"Promo\"]\nsns.heatmap(corr_mat, annot=True, xticklabels=cols, yticklabels=cols, cmap='coolwarm', fmt=\".2f\")\nplt.title(\"Macro Correlation\")\n# save plot with new name\nplt.savefig(\"/kaggle/working/macro_corr_27_stores.png\", bbox_inches='tight')\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport seaborn as sns\nfrom pyspark.sql.functions import col\n\n# read from 27 stores version\ndf_raw = spark.read.parquet(\"/kaggle/working/train_filtered_27_stores.parquet\")\n\n# count total negative returns for reference\nn_neg = df_raw.filter(col(\"unit_sales\") < 0).count()\n\n# sample 1% of data in range [-20, 50] to avoid OOM in pandas\npdf_dist = df_raw.filter(col(\"unit_sales\").between(-20, 50)) \\\n                 .sample(fraction=0.01, seed=42) \\\n                 .select(\"unit_sales\").toPandas()\n\n# clip negatives to 0\npdf_dist['clipped'] = pdf_dist['unit_sales'].clip(lower=0)\n\n# plot dist comparison\nfig, ax = plt.subplots(1, 2, figsize=(12, 4))\n\nsns.histplot(pdf_dist['unit_sales'], bins=50, ax=ax[0], color='gray')\nax[0].set_title(f'Raw Data (1% Sample, Ref: {n_neg} total returns)')\n\nsns.histplot(pdf_dist['clipped'], bins=50, ax=ax[1], color='blue')\nax[1].set_title('Clipped to 0 (1% Sample)')\n\n# save plot with new name for 27 stores\nplt.savefig(\"/kaggle/working/sales_dist_sample_1pct_27_stores.png\", bbox_inches='tight')\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport pandas as pd\nfrom pyspark.sql.functions import col, sum\n\nprint(\"Finding a top-selling item to plot signals...\")\n\n# read from 27 stores master features\ndf_plot = spark.read.parquet(\"/kaggle/working/engineered_features_27_stores.parquet\")\n\n# auto-find top item to ensure dense data for plotting\ntop_item = df_plot.groupBy(\"store_nbr\", \"item_nbr\") \\\n                  .agg(sum(\"unit_sales\").alias(\"total_sales\")) \\\n                  .orderBy(col(\"total_sales\").desc()) \\\n                  .first()\n\nbest_store = top_item[\"store_nbr\"]\nbest_item = top_item[\"item_nbr\"]\nprint(f\"Top selling item found: Store {best_store}, Item {best_item}\")\n\n# extract data to pandas\npdf_sample = df_plot.filter((col(\"store_nbr\") == best_store) & (col(\"item_nbr\") == best_item)) \\\n                    .select(\"date\", \"unit_sales\", \"kalman_sales\", \"wavelet_soft_sales\") \\\n                    .orderBy(\"date\").toPandas()\n\n# plot all signals\nplt.figure(figsize=(16, 6))\nplt.plot(pdf_sample['date'], pdf_sample['unit_sales'], label='Raw Sales', alpha=0.3, color='gray')\nplt.plot(pdf_sample['date'], pdf_sample['kalman_sales'], label='Kalman Filter', alpha=0.8, color='blue', linewidth=1.5)\nplt.plot(pdf_sample['date'], pdf_sample['wavelet_soft_sales'], label='Wavelet (Soft)', alpha=0.8, color='green', linewidth=1.5)\n\nplt.title(f'Signal Denoising Comparison (Store {best_store}, Item {best_item})')\nplt.xlabel('Date')\nplt.ylabel('Sales')\nplt.legend()\nplt.grid(True, linestyle='--', alpha=0.6)\n\n# save plot with new name\nplt.savefig(\"/kaggle/working/signal_denoising_comparison_27_stores.png\", dpi=300, bbox_inches='tight')\nplt.show()\nprint(\"Saved plot to signal_denoising_comparison_27_stores.png\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# load final 27 stores dataset\ndf_final = spark.read.parquet(\"/kaggle/working/engineered_features_encoded_27_stores.parquet\")\n\n# check dims\nnum_rows = df_final.count()\nnum_cols = len(df_final.columns)\nprint(f\"Validation Result - Rows: {num_rows}, Columns: {num_cols}\")\n\n# verify no strings left for dl models\nstring_features = [f.name for f in df_final.schema.fields if f.dataType.simpleString() == \"string\"]\nprint(f\"Verification - String columns remaining: {len(string_features)}\")\n\n# strict qa check\nassert len(string_features) == 0, \"qa failed: strings detected!\"\nassert num_cols == 32, f\"qa failed: expected 32 cols, got {num_cols}\"\nprint(\"QA Status: PASSED. Dataset is ready for model input.\")\n\n# print final schema for report\ndf_final.printSchema()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport pandas as pd\nfrom pyspark.sql.functions import col, sum\n\n# read from 27 stores master features\ndf_plot = spark.read.parquet(\"/kaggle/working/engineered_features_27_stores.parquet\")\n\nprint(\"Dynamically finding the best store and item...\")\n\n# auto-find top item to avoid hardcoding and missing data\ntop_item = df_plot.groupBy(\"store_nbr\", \"item_nbr\") \\\n                  .agg(sum(\"unit_sales\").alias(\"total_sales\")) \\\n                  .orderBy(col(\"total_sales\").desc()) \\\n                  .first()\n\nbest_store = top_item[\"store_nbr\"]\nbest_item = top_item[\"item_nbr\"]\nprint(f\"-> Selected for plotting: Store {best_store}, Item {best_item}\")\n\n# extract sample\npdf_sample = df_plot.filter((col(\"store_nbr\") == best_store) & (col(\"item_nbr\") == best_item)) \\\n                    .select(\"date\", \"unit_sales\", \"kalman_sales\", \"wavelet_soft_sales\", \"wavelet_hard_sales\") \\\n                    .orderBy(\"date\").toPandas()\n\n# plot all 4 signals\nplt.figure(figsize=(16, 6))\nplt.plot(pdf_sample['date'], pdf_sample['unit_sales'], label='Raw Sales', alpha=0.3, color='gray')\nplt.plot(pdf_sample['date'], pdf_sample['kalman_sales'], label='Kalman Filter', alpha=0.8, color='blue', linewidth=1.5)\nplt.plot(pdf_sample['date'], pdf_sample['wavelet_soft_sales'], label='Wavelet (Soft)', alpha=0.8, color='green', linewidth=1.5)\nplt.plot(pdf_sample['date'], pdf_sample['wavelet_hard_sales'], label='Wavelet (Hard)', alpha=0.8, color='purple', linewidth=1.5)\n\nplt.title(f'Signal Denoising (Soft vs Hard) - Store {best_store}, Item {best_item}')\nplt.xlabel('Date')\nplt.ylabel('Sales')\nplt.legend()\nplt.grid(True, linestyle='--', alpha=0.6)\n\n# save plot with new name\nplt.savefig(\"/kaggle/working/signal_denoising_with_hard_27_stores.png\", dpi=300, bbox_inches='tight')\nplt.show()\nprint(\"Saved plot to signal_denoising_with_hard_27_stores.png\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# load final encoded dataset\ndf_final = spark.read.parquet(\"/kaggle/working/engineered_features_encoded_27_stores.parquet\")\n\n# check dataset dimensions\nnum_rows = df_final.count()\nnum_cols = len(df_final.columns)\nprint(f\"Validation Result - Rows: {num_rows}, Columns: {num_cols}\")\n\n# verify zero string columns remain in schema\nstring_features = [f.name for f in df_final.schema.fields if f.dataType.simpleString() == \"string\"]\nprint(f\"Verification - String columns remaining: {len(string_features)}\")\n# trigger strict type assertion\nassert len(string_features) == 0, \"QA Failed: String columns detected!\"\nassert num_cols == 32, f\"QA Failed: Expected 32 columns, got {num_cols}\"\nprint(\"QA Status: PASSED. Dataset is ready for model input.\")\n\n# print full final schema for report documentation\ndf_final.printSchema()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-31T01:47:50.347070Z","iopub.execute_input":"2026-05-31T01:47:50.347474Z","iopub.status.idle":"2026-05-31T01:47:50.366255Z","shell.execute_reply.started":"2026-05-31T01:47:50.347444Z","shell.execute_reply":"2026-05-31T01:47:50.364541Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\n\ncmd = \"zip -rq /kaggle/working/output.zip /kaggle/working/engineered_features_encoded_27_stores.parquet\"\nos.system(cmd)\n\nprint(\"Compression completed successfully.\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}