{ "metadata": { "kernelspec": { "name": "jupyter", "display_name": "Jupyter Notebook" } }, "nbformat_minor": 5, "nbformat": 4, "cells": [ { "cell_type": "code", "id": "78433c43-97e1-402d-b78c-e1a6d46d9a67", "metadata": { "language": "sql", "name": "DB & Schema Setup", "title": "DB & Schema Setup", "resultVariableName": "dataframe_1" }, "source": "%%sql -r dataframe_1\nuse database FACTS;\nuse schema DEV;", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "be4c5dcb-8b43-4023-b4ba-c0e46302ed99", "metadata": { "language": "python", "name": "Pip Imports", "title": "Pip Imports" }, "source": "!pip install -U ipywidgets", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "c380f535-fcb7-4b30-b043-e1ec21aba543", "metadata": { "language": "python" }, "source": "# Imports \n# General python packages\nimport numpy as np\nimport pandas as pd\nimport datetime as dt\nimport matplotlib.pyplot as plt\n\n# Import Snowpark\nfrom snowflake.snowpark.context import get_active_session\nsession = get_active_session()\n\nfrom snowflake.ml.registry import Registry\nregistry = Registry(session=session, database_name=\"FACTS\", schema_name=\"DEV\")\n\nmodel_name = 'TADAS_30_FYQ225_NOAMAZON'\ncurrent_tadas_model = registry.get_model(model_name).version(\"DEV_V1\")\n\n# Import my stuff\nimport data_generation as dg\nimport days_trending_helpers as dth\n\nimport global_tadas_variables as gtv\nall_geos_str = gtv.all_geos_str\nsmall_geos = gtv.small_geos\nsmaller_geos = gtv.smaller_geos\nx_cols_no_amazon = gtv.x_cols_no_amazon\nlist_of_cols = gtv.list_of_cols\n\ndow_df = session.sql(\"\"\"select * from DEV_ENGINEERING.TSTOWE.DAY_OF_WEEK_MODEL\"\"\").to_pandas()", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "f847a0aa-d116-41ea-9434-386f7e6730d5", "metadata": { "language": "python", "name": "Data Generation Function", "title": "Data Generation Function" }, "source": "def data_generation_loop(current_date_str):\n # Make dates.\n yesterday_str = (dt.datetime.strptime(current_date_str, '%Y-%m-%d') - dt.timedelta(days=1)).strftime('%Y-%m-%d')\n seven_days_ago_str = (dt.datetime.strptime(current_date_str, '%Y-%m-%d') - dt.timedelta(days=7)).strftime('%Y-%m-%d')\n\n # Select Tracks.\n print('Selecting Tracks for', current_date_str, ':', dt.datetime.now().strftime(\"%H:%M:%S\"))\n select_tracks_q = \"\"\"select isrc, country_code, sum(STREAMS_ACTIVE) as lean_forward_streams\n from facts.prod.V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY\n where download_activity_date = '\"\"\" + current_date_str + \"\"\"'\n and country_code in \"\"\" + all_geos_str + \"\"\"\n group by 1, 2\n having sum(STREAMS_ACTIVE) > 5000;\"\"\"\n this_date_selected_tracks_df = session.sql(select_tracks_q).to_pandas()\n print(f'Tracks Selected.', len(this_date_selected_tracks_df), 'tracks selected.', dt.datetime.now().strftime(\"%H:%M:%S\"))\n\n # Add unique id.\n this_date_selected_tracks_df['ISRC_GEO'] = this_date_selected_tracks_df['ISRC'].astype(str) + '_' + this_date_selected_tracks_df['COUNTRY_CODE'].astype(str)\n\n # Make list of unique ids.\n this_date_selected_tracks = str(tuple(this_date_selected_tracks_df['ISRC_GEO'].unique()))\n\n # get those beastly sql queries\n sq, ttq = dg.get_streaming_data_sql(current_date_str, yesterday_str, seven_days_ago_str, this_date_selected_tracks)\n\n # Collect data.\n print('Collecting Streaming Data for', current_date_str, ':', dt.datetime.now().strftime(\"%H:%M:%S\"))\n streaming_df = session.sql(sq).to_pandas()\n print('Streaming Data Collected:', dt.datetime.now().strftime(\"%H:%M:%S\"))\n \n print('Collecting Socials Data for', current_date_str, ':', dt.datetime.now().strftime(\"%H:%M:%S\"))\n tiktok_df = session.sql(ttq).to_pandas()\n print('Socials Data Collected:', dt.datetime.now().strftime(\"%H:%M:%S\"))\n\n # merge data.\n merged_df = streaming_df.merge(tiktok_df, how='left', on='ISRC_GEO')\n print('Data Merged.', len(merged_df), 'rows of data. Saving...')\n\n merged_df['DOWNLOAD_ACTIVITY_DATE'] = current_date_str\n \n # Save data.\n session.write_pandas(\n merged_df,\n 'TADAS_2026_DATA_GENERATION_HISTORICAL',\n database='DEV_ENGINEERING',\n schema='TSTOWE',\n auto_create_table=True,\n overwrite=False,\n use_logical_type=True\n )\n print('Save Complete for ', current_date_str, ':', dt.datetime.now().strftime(\"%H:%M:%S\"))\n print('------------------------------------------------')\n\n\nstart_date_str = '2025-08-01'\nend_date_str = '2026-01-31'\n\n# Create a list of date objects\ndate_list = []\ncurrent_date = start_date_str\nwhile current_date <= end_date_str:\n date_list.append(current_date)\n current_date = (dt.datetime.strptime(current_date, '%Y-%m-%d') + dt.timedelta(days=1)).strftime('%Y-%m-%d')\n\nfor d in date_list:\n data_generation_loop(d)", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "6330fad2-3993-4d7a-b706-aa0ce33b9b20", "metadata": { "language": "python", "name": "Days Trending Raw Generator", "title": "Days Trending Raw Generator" }, "source": "# So, to be able to get Days Trending data we dont need to re-run it for every freaking day.\n# We can just collect data from 10 weeks before November 1 through now, run the days trending on that\n# and save the results, then reference them when we need them, either for predictions or truth.\n\n# get isrc_geo from the dg table\n\nmonths = ['2025-08-01', '2025-09-01', '2025-10-01', '2025-11-01', '2025-12-01', '2026-01-01', '2026-02-01']\n\nisrc_geo_q = \"\"\"select isrc_geo \n from DEV_ENGINEERING.TSTOWE.TADAS_2026_DATA_GENERATION_HISTORICAL \n group by 1\"\"\"\nisrc_geo_df = session.sql(isrc_geo_q).to_pandas()\nprint(len(isrc_geo_df))\ntracks_list = list(isrc_geo_df['ISRC_GEO'].unique())\n\ntop_10_geos = ['MX', 'BR', 'DE', 'IN', 'GB', 'ES', 'FR', 'AR', 'CA']\n\n\nfor g in top_10_geos:\n this_geo_tracks = [x for x in tracks_list if x.endswith(g)]\n tracks_tuple = str(tuple(this_geo_tracks))\n \n for i in range(len(months)):\n j = i + 1\n month_start = months[i]\n try:\n month_end = months[j]\n \n print('running query for', month_start, 'in', g, 'on', len(this_geo_tracks), 'tracks.')\n q = dg.get_daysT_q(tracks_tuple, g, month_start, month_end)\n one_month_df = session.sql(q).to_pandas()\n \n print(month_start, g, 'query complete.')\n \n if (g == \"US\") and (month_start == \"2025-08-01\"):\n print('restarting the table.')\n overwrite_choice = True\n else:\n overwrite_choice = False\n \n session.write_pandas(\n one_month_df,\n 'TADAS_2026_DAYS_TRENDING_HISTORICAL',\n database='DEV_ENGINEERING',\n schema='TSTOWE',\n auto_create_table=True,\n overwrite = overwrite_choice,\n use_logical_type=True\n )\n except:\n print('moving to the next geo.')\n", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "090b48ca-a7db-485e-ba0c-8b2b0f439d9d", "metadata": { "language": "python", "name": "Make Days Trending Model", "title": "Make Days Trending Model" }, "source": "# get timeseries data.\noverwrite_choice = True\ntop_10_geos = ['US', 'MX', 'BR', 'DE', 'IN', 'GB', 'ES', 'FR', 'AR', 'CA']\n\nfor g in top_10_geos:\n print('processing', g)\n q = \"\"\"select *\n from DEV_ENGINEERING.TSTOWE.TADAS_2026_DAYS_TRENDING_HISTORICAL\n where RIGHT(isrc_geo, 2) = '\"\"\" + g + \"\"\"'\"\"\"\n \n timeseries_df = session.sql(q).to_pandas()\n timeseries_df['geo_country'] = timeseries_df['ISRC_GEO'].str[-2:]\n \n timeseries_df.rename(columns={'DOWNLOAD_ACTIVITY_DATE':'report_date',\n 'ISRC_GEO': 'pfn_geo',\n 'TOTAL_STREAMS_ACTIVE': 'streams'}, inplace=True)\n \n # add simple columns\n print('--- ', dt.datetime.now().strftime(\"%H:%M:%S\"), ': prepping the time series -', len(timeseries_df), 'rows of data.')\n timeseries_df = dth.prep_timeseries(timeseries_df, small_geos, smaller_geos)\n \n # run the iron out function.\n print('--- ', dt.datetime.now().strftime(\"%H:%M:%S\"), ': ironing out trends')\n ironed_out_intl_df = dth.iron_out_trends(timeseries_df)\n \n ironed_out_intl_df['vs_forecast_lift'] = ironed_out_intl_df['streams'] / ironed_out_intl_df['combined_forecast'] - 1.0\n \n print('saving', g)\n session.write_pandas(\n ironed_out_intl_df,\n 'TADAS_2026_DAYS_TRENDING_HISTORICAL_PROCESSED',\n database='DEV_ENGINEERING',\n schema='TSTOWE',\n auto_create_table=True,\n overwrite = overwrite_choice,\n use_logical_type=True\n )\n \n overwrite_choice = False\n print('save complete for', g)\n print('---------------------------')\n", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "621a2a70-d07c-4b32-8cc4-65f7d1f4ab3c", "metadata": { "language": "python", "name": "Make Date List", "title": "Make Date List" }, "source": "start_date_str = '2025-11-01'\nend_date_str = '2026-01-31'\n\n# Create a list of date objects\ndate_list = []\ncurrent_date = start_date_str\nwhile current_date <= end_date_str:\n date_list.append(current_date)\n current_date = (dt.datetime.strptime(current_date, '%Y-%m-%d') + dt.timedelta(days=1)).strftime('%Y-%m-%d')", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "4365abfc-c1cc-4263-815f-1420bcd4beb5", "metadata": { "language": "python", "name": "Get Predictions", "title": "Get Predictions" }, "source": "# P4: Get Predictions\n# for each day and store them. Include geography.\n\noverwrite_choice = True\n\nfor d in date_list:\n print('Running Predictions for', d)\n # query from data_generation.\n q = \"\"\"select * \n from DEV_ENGINEERING.TSTOWE.TADAS_2026_DATA_GENERATION_HISTORICAL \n where DOWNLOAD_ACTIVITY_DATE = '\"\"\" + d + \"\"\"';\"\"\"\n data_generation_df = session.sql(q).to_pandas()\n\n # query days trending\n q = \"\"\"select * from DEV_ENGINEERING.TSTOWE.TADAS_2026_DAYS_TRENDING_HISTORICAL_PROCESSED where \"report_date\" = '\"\"\" + d + \"\"\"';\"\"\"\n days_trending_df = session.sql(q).to_pandas()\n days_trending_df.rename(columns = {'pfn_geo': 'ISRC_GEO'}, inplace=True)\n \n # combine inner.\n combined_df = data_generation_df.merge(days_trending_df, how='inner', on='ISRC_GEO')\n print('--- got both datasets and combined.')\n\n combined_df.rename(columns={'consecutive_trend': 'CONSECUTIVE_TREND'}, inplace=True)\n\n # run through model to get predictions.\n for x in list_of_cols:\n yest_x = 'YEST_'+x\n seven_x = 'SEVEN_'+x\n wow_new_col = 'WOW_'+x\n dod_new_col = 'DOD_'+x\n combined_df[wow_new_col] = combined_df[x] / combined_df[seven_x] - 1.0\n combined_df[dod_new_col] = combined_df[x] / combined_df[yest_x] - 1.0\n\n combined_df['REPORT_DATE_STR'] = combined_df['report_date']\n combined_df['REPORT_DATE_DT'] = pd.to_datetime(combined_df['REPORT_DATE_STR'])\n combined_df['WEEKDAY_NUM'] = combined_df['REPORT_DATE_DT'].dt.dayofweek # 0=Mon, ..., 6=Sun\n \n # # Merge in expected changes\n combined_df = combined_df.merge(dow_df, on='WEEKDAY_NUM', how='left')\n \n for x in combined_df.columns:\n if x[0:4] == 'DOD_':\n adj_col = 'ADJ_'+x\n # 2. Index vs expected (ratio; handle divide by zero)\n combined_df[adj_col] = combined_df[x] / combined_df['PCT_CHANGE_VS_PREV'].replace(0, pd.NA)\n \n # Make the data usable/modelable.\n combined_df = combined_df.replace([np.inf], 1000.0)\n combined_df = combined_df.replace([-np.inf], -1000.0)\n combined_df = combined_df.fillna(0.0)\n\n print('--- extra columns made. Starting Predictions.')\n\n trending_df = combined_df[combined_df['CONSECUTIVE_TREND'] > 0]\n nottrending_df = combined_df[combined_df['CONSECUTIVE_TREND'] == 0]\n\n # Prep trending to go into the model.\n trending_df = trending_df.set_index('ISRC_GEO')\n \n X = trending_df[x_cols_no_amazon]\n\n # combine predictions back onto df.\n predictions = current_tadas_model.run(X, function_name=\"PREDICT_PROBA\")\n\n # add the predictions onto the df.\n trending_df['TADAS_30'] = predictions['output_feature_1'].values\n\n # bring back isrc_geo.\n trending_df = trending_df.reset_index()\n\n # TADAS 30 is 0 for all non trending tracks.\n nottrending_df['TADAS_30'] = 0.0\n\n print('--- Predictions complete. Saving.')\n\n recombined_df = pd.concat([trending_df, nottrending_df], ignore_index=True)\n \n # save.\n session.write_pandas(\n recombined_df,\n 'TADAS_2026_PREDICTIONS_RAW',\n database='DEV_ENGINEERING',\n schema='TSTOWE',\n auto_create_table=True,\n overwrite = overwrite_choice,\n use_logical_type=True\n )\n\n overwrite_choice = False\n print('--- Successfully saved.')\n print('-----------------------------------')\n ", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "2f7a9508-8ed3-4587-9672-d1a933144456", "metadata": { "language": "python", "name": "Get Truth Labels.", "title": "Get Truth Labels." }, "source": "# P5: Get Truth Labels.\n\n# ok, lets requery the above table for the consecutive trend and tadas scores.\n# q = \"\"\"select ISRC_GEO, \"geo_country\", \"report_date\", CONSECUTIVE_TREND, TADAS_30\n# from DEV_ENGINEERING.TSTOWE.TADAS_2026_PREDICTIONS_RAW;\"\"\"\n\n# conclusion_df = session.sql(q).to_pandas()\n", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "7fb57dba-9b1e-473d-a608-8ec5d4119675", "metadata": { "language": "python", "name": "Alternate Shift 30 Method Start", "title": "Alternate Shift 30 Method Start" }, "source": "q = \"\"\"select ISRC_GEO, \"geo_country\", \"report_date\", CONSECUTIVE_TREND, TADAS_30\n from DEV_ENGINEERING.TSTOWE.TADAS_2026_PREDICTIONS_RAW\n where CONSECUTIVE_TREND > 0;\"\"\"\n\ntrending_df = session.sql(q).to_pandas()\n\ntrending_df['report_date'] = trending_df['report_date'].dt.strftime('%Y-%m-%d')\ntrending_df['report_date_dt'] = pd.to_datetime(trending_df['report_date'])\ntrending_df['lookup_key'] = trending_df['ISRC_GEO'] + '_' + trending_df['report_date']\ntrending_df['target_date'] = (trending_df['report_date_dt'] + pd.Timedelta(days=30)).dt.strftime('%Y-%m-%d')\ntrending_df['target_key'] = trending_df['ISRC_GEO'] + '_' + trending_df['target_date']\n\ntarget_df = trending_df[['lookup_key', 'CONSECUTIVE_TREND']]\ntarget_df['is_30'] = 0\ntarget_df['is_30'][target_df['CONSECUTIVE_TREND'] >= 28] = 1\ndel target_df['CONSECUTIVE_TREND']\ntarget_df.rename(columns={'lookup_key':'target_key'}, inplace=True)\ntarget_df = target_df[target_df['is_30'] > 0]\n\nresults_df = trending_df.merge(target_df, how='left', on='target_key')\nresults_df['is_30'][results_df['is_30'].isnull()] = 0\n\nresults_df.reset_index(drop=True, inplace=True)\n\nsession.write_pandas(\n results_df,\n 'TADAS_2026_MODEL_MONITORING_OUTPUT',\n database='DEV_ENGINEERING',\n schema='TSTOWE',\n auto_create_table=True,\n overwrite = True,\n use_logical_type=True\n )", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "2385837c-cb09-4daf-b172-b679c9b81cd3", "metadata": { "language": "python", "name": "Old Shift 30 Method", "title": "Old Shift 30 Method" }, "source": "# # Make an Is_30 column\n# conclusion_df['is_30'] = 0\n# conclusion_df['is_30'][conclusion_df['CONSECUTIVE_TREND'] >= 30] = 1\n\n# # shift is_30 back 30\n# conclusion_df['shift_30'] = conclusion_df['is_30'].shift(-29)\n\n# # Give is_30 some wiggle room for 1 day fails.\n# conclusion_df['shift_30_34'] = conclusion_df['is_30'].shift(-30)\n# conclusion_df['shift_30_34'][conclusion_df['shift_30_34'] == 0] = conclusion_df['is_30'].shift(-31)\n# conclusion_df['shift_30_34'][conclusion_df['shift_30_34'] == 0] = conclusion_df['is_30'].shift(-32)\n# conclusion_df['shift_30_34'][conclusion_df['shift_30_34'] == 0] = conclusion_df['is_30'].shift(-33)\n\n# min_dates = conclusion_df.groupby('ISRC_GEO')['report_date'].transform('min')\n# days_since_start = (pd.to_datetime(conclusion_df['report_date']) - pd.to_datetime(min_dates)).dt.days\n# conclusion_df.loc[days_since_start < 30, 'shift_30_34'] = np.nan\n\n# del conclusion_df['is_30']\n# del conclusion_df['shift_30']", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "9807b842-c562-420a-b08e-0082980dc3be", "metadata": { "language": "python", "name": "Filter & Save", "title": "Filter & Save" }, "source": "# filtered_df = conclusion_df[~conclusion_df['shift_30_34'].isnull()]\n# filtered_df = filtered_df[filtered_df['CONSECUTIVE_TREND'] > 0]\n# filtered_df = filtered_df[filtered_df['TADAS_30'] > 0]\n# filtered_df['ground_truth'] = 0\n# filtered_df['ground_truth'][filtered_df['shift_30_34'] > 0] = 1\n# filtered_df.reset_index(drop=True, inplace=True)\n\n# # REPORT_DATE | ISRC_GEO | GEO_COUNTRY | TADAS_30 | CONSECUTIVE_TREND | SHIFT_3032\n\n# session.write_pandas(\n# filtered_df,\n# 'TADAS_2026_MODEL_MONITORING_OUTPUT',\n# database='DEV_ENGINEERING',\n# schema='TSTOWE',\n# auto_create_table=True,\n# overwrite = True,\n# use_logical_type=True\n# )\n", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "e8a14297-a409-4d94-9ba3-b11aaa4bcbf8", "metadata": { "language": "python", "name": "Show Calibration Curve", "title": "Show Calibration Curve" }, "source": "bins = np.arange(0, 1.1, 0.1)\nresults_df['tadas_bucket'] = pd.cut(results_df['TADAS_30'], bins=bins, right=False)\n\ncal = results_df.groupby('tadas_bucket', observed=False).agg(\n success_rate=('is_30', 'mean'),\n count=('is_30', 'count')\n).reset_index()\n\ncal['bucket_mid'] = [b.mid for b in cal['tadas_bucket']]\n\nfig, ax1 = plt.subplots(figsize=(10, 6))\n\nax1.bar(cal['bucket_mid'], cal['count'], width=0.08, alpha=0.3, color='steelblue', label='Count')\nax1.set_ylabel('Count', color='steelblue')\nax1.tick_params(axis='y', labelcolor='steelblue')\n\nax2 = ax1.twinx()\nax2.plot(cal['bucket_mid'], cal['success_rate'], 'o-', color='darkorange', linewidth=2, label='Actual Success Rate')\nax2.plot([0, 1], [0, 1], '--', color='gray', label='Perfect Calibration')\nax2.set_ylabel('P(ground_truth = 1)', color='darkorange')\nax2.tick_params(axis='y', labelcolor='darkorange')\nax2.set_ylim(0, 1)\n\nax1.set_xlabel('TADAS_30 Bucket')\nax1.set_title('Calibration Curve: TADAS_30 vs Actual Success Rate')\nax1.set_xticks(np.arange(0.05, 1.05, 0.1))\nax1.set_xticklabels([f'{int(b*100-5)}-{int(b*100+5)}%' for b in np.arange(0.05, 1.05, 0.1)], rotation=45)\n\nlines1, labels1 = ax1.get_legend_handles_labels()\nlines2, labels2 = ax2.get_legend_handles_labels()\nax2.legend(lines1 + lines2, labels1 + labels2, loc='upper left')\n\nplt.tight_layout()\nplt.show()", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "f2024692-231c-49ea-935a-266cc704cab1", "metadata": { "language": "python" }, "source": "", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "1b146cdf-16e0-4a27-a874-a45f93573826", "metadata": { "language": "sql", "resultVariableName": "dataframe_2" }, "source": "%%sql -r dataframe_2\nCREATE OR REPLACE MODEL MONITOR FACTS.DEV.TRAVIS_TEST_TADAS_MONITOR\nWITH\n MODEL=TADAS_30_FYQ225_NOAMAZON\n VERSION=DEV_V1\n FUNCTION=predict\n SOURCE=DEV_ENGINEERING.TSTOWE.TADAS_2026_MODEL_MONITORING_OUTPUT\n TIMESTAMP_COLUMN=\"REPORT_DATE_DT\"\n PREDICTION_CLASS_COLUMNS=(TADAS_30)\n ACTUAL_CLASS_COLUMNS=(is_30)\n ID_COLUMNS=(ISRC_GEO)\n WAREHOUSE=SNOWFLAKE_NOTEBOOKS_WAREHOUSE\n REFRESH_INTERVAL='3 hours'\n AGGREGATION_WINDOW='1 day';", "outputs": [], "execution_count": null }, { "cell_type": "code", "id": "1232aa39-8396-47c7-a143-14b8aa59bbc7", "metadata": { "language": "sql", "resultVariableName": "dataframe_3" }, "source": "%%sql -r dataframe_3\n", "outputs": [], "execution_count": null } ] }