Add files via upload

Author: yohanmarkose

notebooks/02_raw_to_harmonized/02_raw_to_harmonized.ipynb

@@ -0,0 +1,94 @@
+{
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Streamlit Notebook",
+   "name": "streamlit"
+  },
+  "lastEditStatus": {
+<<<<<<< HEAD
+   "notebookId": "dwcf2fynfosmyghponvw",
+   "authorId": "6690310009356",
+   "authorName": "YOHANMARKOSE",
+   "authorEmail": "markose.y@northeastern.edu",
+   "sessionId": "3abaf741-87a4-40b0-8c7d-dc03d4b7f2a7",
+   "lastEditTime": 1740739209269
+=======
+   "notebookId": "eit25xap76vo7ktof4y5",
+   "authorId": "6690310009356",
+   "authorName": "YOHANMARKOSE",
+   "authorEmail": "markose.y@northeastern.edu",
+   "sessionId": "cc15f23b-b6c0-4028-a56c-82718000c5b9",
+   "lastEditTime": 1740718649816
+>>>>>>> origin/main
+  }
+ },
+ "nbformat_minor": 5,
+ "nbformat": 4,
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "3775908f-ca36-4846-8f38-5adca39217f2",
+   "metadata": {
+    "language": "sql",
+    "name": "sql_get_context"
+   },
+   "outputs": [],
+   "source": [
+    "SELECT current_database() AS DATABASE_NAME, current_schema() AS SCHEMA_NAME\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "9a73e27a-c874-4e9c-a236-67aed79d8fd9",
+   "metadata": {
+    "language": "python",
+    "name": "cell1"
+   },
+   "outputs": [],
+   "source": [
+    "# Import python packages\n",
+    "import logging\n",
+    "from snowflake.snowpark.context import get_active_session\n",
+    "\n",
+    "logger = logging.getLogger(\"fred_logger\")\n",
+    "\n",
+    "current_context_df = cells.sql_get_context.to_pandas()\n",
+    "database_name = current_context_df.iloc[0,0]\n",
+    "schema_name = current_context_df.iloc[0,1]\n",
+    "\n",
+    "session = get_active_session()\n",
+    "\n",
+    "logger.info(\"02_raw_to_harmonized start\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "31927266-930c-4441-8639-6d8cefcfcee4",
+   "metadata": {
+    "collapsed": false,
+    "language": "python",
+    "name": "cell3"
+   },
+   "outputs": [],
+<<<<<<< HEAD
+   "source": "import snowflake.snowpark.functions as F\n\nenv = schema_name.split('_')[0]\ndef create_fred_table(session):\n    raw_fred_10Y = session.table(f\"{database_name}.{env}_RAW_FRED.FRED_DGS10Y\").select(\n        F.to_date(F.col('\"date\"')).alias(\"OBSERVATION_DATE\"),\n        F.when(F.col('\"value\"') == 0, None)\n         .otherwise(F.col('\"value\"'))\n         .cast(\"float\")\n         .alias(\"10Y_YIELD\")\n    )\n    \n    raw_fred_2Y = session.table(f\"{database_name}.{env}_RAW_FRED.FRED_DGS2Y\").select(\n        F.to_date(F.col('\"date\"')).alias(\"OBSERVATION_DATE\"),\n        F.when(F.col('\"value\"') == 0, None)\n         .otherwise(F.col('\"value\"'))\n         .cast(\"float\")\n         .alias(\"2Y_YIELD\")\n    )\n\n    harmonized_data = raw_fred_10Y.join(\n        raw_fred_2Y, \n        raw_fred_10Y[\"OBSERVATION_DATE\"] == raw_fred_2Y[\"OBSERVATION_DATE\"]\n    ).select(\n        raw_fred_10Y[\"OBSERVATION_DATE\"].alias(\"OBSERVATION_DATE\"),\n        F.col(\"10Y_YIELD\"),\n        F.col(\"2Y_YIELD\")\n    )\n    \n    session.use_schema(f\"{database_name}.{schema_name}\")\n    harmonized_data.write.mode(\"overwrite\").save_as_table('FRED_FLATTENED')\n\ndef create_fred_table_stream(session):\n    _ = session.sql('CREATE OR REPLACE STREAM FRED_STREAM \\\n                        ON TABLE FRED_FLATTENED \\\n                        SHOW_INITIAL_ROWS = TRUE').collect()\n    \n"
+=======
+   "source": "import snowflake.snowpark.functions as F\n\nenv = schema_name.split('_')[0]\ndef create_fred_table(session):\n    raw_fred_10Y = session.table(f\"{database_name}.{env}_RAW_FRED.FRED_DGS10Y\").select(\n                                                                                        F.to_date(F.col('\"date\"')).alias(\"OBSERVATION_DATE\"),\n                                                                                        F.col('\"value\"').cast(\"float\").alias(\"10Y_YIELD\")\n                                                                                    )\n    \n    raw_fred_2Y = session.table(f\"{database_name}.{env}_RAW_FRED.FRED_DGS2Y\").select(\n                                                                                        F.to_date(F.col('\"date\"')).alias(\"OBSERVATION_DATE\"),\n                                                                                        F.col('\"value\"').cast(\"float\").alias(\"2Y_YIELD\")\n                                                                                    )\n\n    harmonized_data = raw_fred_10Y.join(raw_fred_2Y, raw_fred_10Y[\"OBSERVATION_DATE\"] == raw_fred_2Y[\"OBSERVATION_DATE\"]).select(\n                                                                                                                                    raw_fred_10Y[\"OBSERVATION_DATE\"].alias(\"OBSERVATION_DATE\"),\n                                                                                                                                    F.col(\"10Y_YIELD\"),\n                                                                                                                                    F.col(\"2Y_YIELD\")\n                                                                                                                                )\n\n\n    \n    session.use_schema(f\"{database_name}.{schema_name}\")\n    harmonized_data.write.mode(\"overwrite\").save_as_table('FRED_FLATTENED')\n\ndef create_fred_table_stream(session):\n    _ = session.sql('CREATE OR REPLACE STREAM FRED_STREAM \\\n                        ON TABLE FRED_FLATTENED \\\n                        SHOW_INITIAL_ROWS = TRUE').collect()\n    \n"
+>>>>>>> origin/main
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "29badd8a-aead-4f7b-bdcb-e12049c0d18b",
+   "metadata": {
+    "language": "python",
+    "name": "cell4"
+   },
+   "outputs": [],
+   "source": "\ncreate_fred_table(session)\ncreate_fred_table_stream(session)"
+  }
+ ]
+}