{"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, when, count, isnan\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\nparquet_path = \"/kaggle/working/train_filtered.parquet\"\n\n# check if parquet exists to skip unzip & filter steps\nif os.path.exists(parquet_path):\n    print(\"Loading filtered data from Parquet...\")\n    df_train = spark.read.parquet(parquet_path)\nelse:\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 and filter\n    df_train = spark.read.csv(\"/kaggle/working/train.csv\", header=True, inferSchema=True)\n    df_train = df_train.filter((col(\"date\") >= \"2016-08-01\") & (col(\"date\") <= \"2017-08-15\"))\n    \n    # save to parquet for future runs\n    df_train.write.mode(\"overwrite\").parquet(parquet_path)\n    print(\"Saved to Parquet for future runs!\")\n\n# check result\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-30T04:38:59.982019Z","iopub.execute_input":"2026-05-30T04:38:59.982451Z","iopub.status.idle":"2026-05-30T04:39:22.511308Z","shell.execute_reply.started":"2026-05-30T04:38:59.98238Z","shell.execute_reply":"2026-05-30T04:39:22.508017Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# save filtered data to parquet \noutput_path = \"/kaggle/working/train_filtered.parquet\"\ndf_train.write.mode(\"overwrite\").parquet(output_path)\nprint(f\"Data saved to {output_path}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-30T04:39:22.51339Z","iopub.execute_input":"2026-05-30T04:39:22.516569Z","iopub.status.idle":"2026-05-30T04:39:22.52583Z","shell.execute_reply.started":"2026-05-30T04:39:22.516494Z","shell.execute_reply":"2026-05-30T04:39:22.524783Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\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,"execution":{"iopub.status.busy":"2026-05-30T04:39:22.529051Z","iopub.execute_input":"2026-05-30T04:39:22.529524Z","iopub.status.idle":"2026-05-30T04:39:30.526432Z","shell.execute_reply.started":"2026-05-30T04:39:22.529471Z","shell.execute_reply":"2026-05-30T04:39:30.517792Z"}},"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,"execution":{"iopub.status.busy":"2026-05-30T04:39:30.527631Z","iopub.execute_input":"2026-05-30T04:39:30.528028Z","iopub.status.idle":"2026-05-30T04:39:34.926812Z","shell.execute_reply.started":"2026-05-30T04:39:30.527983Z","shell.execute_reply":"2026-05-30T04:39:34.915739Z"}},"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,"execution":{"iopub.status.busy":"2026-05-30T04:39:34.935785Z","iopub.execute_input":"2026-05-30T04:39:34.93632Z","iopub.status.idle":"2026-05-30T04:39:37.064384Z","shell.execute_reply.started":"2026-05-30T04:39:34.936265Z","shell.execute_reply":"2026-05-30T04:39:37.063496Z"}},"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,"execution":{"iopub.status.busy":"2026-05-30T04:39:37.065461Z","iopub.execute_input":"2026-05-30T04:39:37.067032Z","iopub.status.idle":"2026-05-30T04:39:38.005233Z","shell.execute_reply.started":"2026-05-30T04:39:37.066982Z","shell.execute_reply":"2026-05-30T04:39:38.003796Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.sql.functions import broadcast\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# Join data\ndf_base = df_train.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,"execution":{"iopub.status.busy":"2026-05-30T04:57:27.810433Z","iopub.execute_input":"2026-05-30T04:57:27.811143Z","iopub.status.idle":"2026-05-30T04:57:38.878959Z","shell.execute_reply.started":"2026-05-30T04:57:27.811085Z","shell.execute_reply":"2026-05-30T04:57:38.877749Z"}},"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\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# save output\noutput_path = \"/kaggle/working/base_processed.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    plt.savefig(\"/kaggle/working/kalman_report.png\", dpi=300, bbox_inches='tight')\n    plt.show()\n\n!pip install PyWavelets -q\n\nimport pandas as pd\nimport pywt\nimport numpy as np\nfrom pyspark.sql.functions import col\n\ndf_base = spark.read.parquet(\"/kaggle/working/base_processed.parquet\")\n\n# use ddl schema\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    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    \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\nout_wavelet = \"/kaggle/working/features_wavelet.parquet\"\ndf_wavelet.write.mode(\"overwrite\").parquet(out_wavelet)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-30T04:57:39.494858Z","iopub.execute_input":"2026-05-30T04:57:39.495844Z","iopub.status.idle":"2026-05-30T06:23:09.674009Z","shell.execute_reply.started":"2026-05-30T04:57:39.495805Z","shell.execute_reply":"2026-05-30T06:23:09.669096Z"}},"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.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.parquet\"\ndf_features.write.mode(\"overwrite\").parquet(out_master)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-30T06:38:48.147808Z","iopub.execute_input":"2026-05-30T06:38:48.149059Z","iopub.status.idle":"2026-05-30T06:42:54.543693Z","shell.execute_reply.started":"2026-05-30T06:38:48.149005Z","shell.execute_reply":"2026-05-30T06:42:54.54209Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pyspark.ml.feature import StringIndexer\n\n# load saved master data\ndf_features = spark.read.parquet(\"/kaggle/working/engineered_features.parquet\")\n\n# apply string indexer for categories\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 original strings to save memory\ndf_features = df_features.drop(*cat_cols)\n\n# save final encoded data for model input\nout_final = \"/kaggle/working/engineered_features_encoded.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# print schema for report\ndf_features.printSchema()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-30T06:46:08.929699Z","iopub.execute_input":"2026-05-30T06:46:08.930251Z","iopub.status.idle":"2026-05-30T06:48:52.583445Z","shell.execute_reply.started":"2026-05-30T06:46:08.93021Z","shell.execute_reply":"2026-05-30T06:48:52.582209Z"}},"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# load raw data\ndf_raw = spark.read.parquet(\"/kaggle/working/train_filtered.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 integer 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 (100% data)\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 (exact counts)\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 100% Data)')\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()\nplt.savefig(\"/kaggle/working/sales_distribution_exact.png\", dpi=300)\nplt.show()\n\n# load processed data for macro corr\ndf_base = spark.read.parquet(\"/kaggle/working/base_processed.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\")\nplt.savefig(\"/kaggle/working/macro_corr.png\", bbox_inches='tight')\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-30T06:49:54.437393Z","iopub.execute_input":"2026-05-30T06:49:54.437896Z","iopub.status.idle":"2026-05-30T06:51:13.629111Z","shell.execute_reply.started":"2026-05-30T06:49:54.437858Z","shell.execute_reply":"2026-05-30T06:51:13.626126Z"}},"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# load raw data\ndf_raw = spark.read.parquet(\"/kaggle/working/train_filtered.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\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 in pandas\npdf_dist['clipped'] = pdf_dist['unit_sales'].clip(lower=0)\n\n# plot dist comparison (sample)\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 with distinct filename\nplt.savefig(\"/kaggle/working/sales_dist_sample_1pct.png\", bbox_inches='tight')\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-30T06:52:01.517764Z","iopub.execute_input":"2026-05-30T06:52:01.518615Z","iopub.status.idle":"2026-05-30T06:52:05.309526Z","shell.execute_reply.started":"2026-05-30T06:52:01.518546Z","shell.execute_reply":"2026-05-30T06:52:05.308192Z"}},"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# Load the final data that contains both kalman and wavelet\ndf_plot = spark.read.parquet(\"/kaggle/working/engineered_features.parquet\")\n\n# Auto-find the store and item with the highest total sales (ensures a dense, non-empty time series)\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 \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 image \nplt.savefig(\"/kaggle/working/signal_denoising_comparison.png\", dpi=300, bbox_inches='tight')\nplt.show()\nprint(\"Saved plot to signal_denoising_comparison.png\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-05-30T06:59:37.263802Z","iopub.execute_input":"2026-05-30T06:59:37.264462Z","iopub.status.idle":"2026-05-30T06:59:42.396418Z","shell.execute_reply.started":"2026-05-30T06:59:37.264419Z","shell.execute_reply":"2026-05-30T06:59:42.394834Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# load final encoded dataset\ndf_final = spark.read.parquet(\"/kaggle/working/engineered_features_encoded.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\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-30T07:05:29.651307Z","iopub.execute_input":"2026-05-30T07:05:29.651876Z","iopub.status.idle":"2026-05-30T07:05:30.90103Z","shell.execute_reply.started":"2026-05-30T07:05:29.651828Z","shell.execute_reply":"2026-05-30T07:05:30.899885Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport pandas as pd\nfrom pyspark.sql.functions import col\n\n# load finalized data\ndf_plot = spark.read.parquet(\"/kaggle/working/engineered_features.parquet\")\n\n# use the exact store and item from the previous plot for fair comparison\nbest_store = 44\nbest_item = 1503844\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 with new name\nplt.savefig(\"/kaggle/working/signal_denoising_with_hard.png\", dpi=300, bbox_inches='tight')\nplt.show()\nprint(\"Saved plot to signal_denoising_with_hard.png\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}