{ "cells": [ { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import sys\n", "sys.path.insert(0, '/Users/joel/src/thundr/tracker')" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "import random\n", "import traceback\n", "import json\n", "import requests\n", "import pandas as pd\n", "import numpy\n", "import matplotlib\n", "from io import StringIO\n", "from datetime import date, timedelta" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [], "source": [ "import pymysql\n", "from contextlib import contextmanager\n", "import datetime\n", "\n", "conn_kwargs = {\n", " 'host': 'unicron.cluster-cbn1zk7uet6r.eu-west-1.rds.amazonaws.com', \n", " 'user': 'unicron', \n", " 'password': 'tydz5cy9xbvchqpa249rfekhcg9sey', \n", " 'db': 'unicron', \n", " 'charset': 'utf8mb4'\n", "}\n", "\n", "@contextmanager\n", "def get_cursor(commit_after=False):\n", " connection = pymysql.connect(**conn_kwargs)\n", " try:\n", " with connection.cursor() as cur:\n", " yield cur\n", " if commit_after:\n", " connection.commit()\n", " else:\n", " connection.rollback()\n", " finally:\n", " try:\n", " connection.close()\n", " except Exception as e:\n", " print(\"Warning, failed to close connection\")\n", " print(e)\n", " \n", " \n", "def chunks(lst, n):\n", " \"\"\"Yield successive n-sized chunks from lst.\"\"\"\n", " for i in range(0, len(lst), n):\n", " yield lst[i:i + n]" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [], "source": [ "from scripts import run_spartus_daily" ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [], "source": [ "cg = run_spartus_daily.get_for_skechers()" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "h = cg.refresh_request_headers()" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "def make_url(artist_id, spartus_daily: run_spartus_daily.SpartusDaily):\n", " return (\n", " 'https://generic.wg.spotify.com/s4x-insights-api/v1'\n", " f'/artist/{spartus_daily.ARTIST_ID}/audience/timeline/streams'\n", " f'/{artist_id}'\n", " '?time-filter=since2015&aggregation-level=recording' \n", " )\n" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [], "source": [ "def medians(df, window):\n", " if (df.size < int(window*0.75)):\n", " # not enough data\n", " return [None, None, None, None]\n", " \n", " ma, mb = df.iloc[-window:].resample(f'{window // 2}D').median().num.to_numpy().tolist()\n", " return [ma, mb, mb - ma, (mb - ma) / ma if ma > 0 else None]" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [], "source": [ "def get_start(df):\n", " first_non_zero = (df.sort_index().num > 0).idxmax()\n", " return df.sort_index().loc[first_non_zero:].copy()" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "def to_csv_str(df):\n", " sio = StringIO()\n", " packed = df.copy()\n", " packed.index = (packed.index.to_series().dt.date - date(2015,1,1)).dt.days\n", " packed.to_csv(sio)\n", " return sio.getvalue()" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [], "source": [ "def request_data(artist_id, spartus_daily):\n", " url = make_url(artist_id, spartus_daily)\n", " r1 = requests.get(url, headers=spartus_daily.headers)\n", " if r1.status_code != 200:\n", " raise RuntimeError(f\"Non 200 request {url}, {r1}, {r1.reason}\")\n", " \n", " df1 = pd.DataFrame(data=r1.json()['timelinePoint'])\n", "\n", " df1['num'] = df1.num.astype(int)\n", " df1['date'] = df1.date.astype(numpy.datetime64)\n", " \n", " return r1, df1.set_index('date')" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [], "source": [ "def store_in_db(spyid, df):\n", " smaller_df = get_start(df)\n", " args = [spyid, smaller_df.index.min().date(), smaller_df.index.max().date(),]\n", " args.extend([int(smaller_df.num.iloc[i]) for i in [-1, -2, -8, -15, -29]])\n", " args.extend([int(smaller_df.num.max()), smaller_df.num.idxmax().date()])\n", " args.extend(medians(smaller_df, 14))\n", " args.extend(medians(smaller_df, 28))\n", " args.extend(medians(smaller_df, 56))\n", " args.extend(medians(smaller_df, 112))\n", " args.extend([to_csv_str(smaller_df)]) \n", "\n", " pcts = \",\".join(['%s']*len(args))\n", " \n", " sql = f'''replace into spotify_daily_streams (\n", " spyid, as_of, first_date, \n", " s0, s1, s7, s14, s28, \n", " smax, smax_date, \n", " m14a, m14b, d14, p14, \n", " m28a, m28b, d28, p28, \n", " m56a, m56b, d56, p56, \n", " m112a, m112b, d112, p112, \n", " tsdata\n", " ) values ({pcts})'''\n", " \n", " with get_cursor(True) as cur:\n", " cur.execute(sql, args)\n", " " ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [], "source": [ "def make_df_from_row(spyid, tsdata):\n", " nf = pd.read_csv(StringIO(tsdata))\n", " nf['spyid'] = spyid\n", " return nf\n", " " ] }, { "cell_type": "code", "execution_count": 14, "metadata": {}, "outputs": [], "source": [ "def load_streams(spyids:list):\n", " with get_cursor(False) as cur:\n", " pcts = \",\".join(['%s'] * len(spyids))\n", " cur.execute(f'select spyid, tsdata from spotify_daily_streams where spyid in ({pcts})', spyids)\n", " res = cur.fetchall()\n", " dfall = pd.concat([make_df_from_row(spyid, tsdata) for spyid, tsdata in res])\n", " dfall['date'] = dfall.date.apply(lambda x: date(2015,1,1) + timedelta(days=x)).astype(numpy.datetime64)\n", " return dfall.pivot(index='date', columns='spyid', values='num')" ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [], "source": [ "def fetch_and_store(artist_id, spartus_daily):\n", " r1, df1 = request_data(artist_id, cg)\n", " store_in_db(artist_id, df1)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": 16, "metadata": {}, "outputs": [], "source": [ "isomap = requests.get(\n", " ('https://gist.githubusercontent.com/ssskip/5a94bfcd2835bf1dea52/raw/'\n", " 'aeed5b0cb3a7eda19e614915c3d88ce113e4a914/ISO3166-1.alpha2.json')\n", ").json()" ] }, { "cell_type": "code", "execution_count": 17, "metadata": {}, "outputs": [], "source": [ "def store_batch(spyid_list):\n", " print(f\"Starting, {len(spyid_list)} items\")\n", " errors = 0\n", " for i, artist_id in enumerate(spyid_list):\n", " if (i % 30 == 0):\n", " print(\"refreshing headers...\")\n", " cg.refresh_request_headers()\n", " print(i, 'doing ', artist_id)\n", " try:\n", " fetch_and_store(artist_id, cg)\n", " errors = 0\n", " except Exception as e:\n", " errors += 1\n", " print(f\"Failed to get and store streams for {artist_id}, {e}\")\n", " traceback.print_exc()\n", "\n", " if errors > 5:\n", " print(\"Too many errors\")\n", " break\n", "\n", " import time; time.sleep(30 + random.randint(10,40))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## " ] }, { "cell_type": "code", "execution_count": 18, "metadata": {}, "outputs": [], "source": [ "def add_slope(dframe, days_back, roll=7):\n", " col = dframe.columns[0]\n", " rn = dframe.rolling(roll).mean().iloc[-days_back:]\n", " int_x = rn.index.astype(int)\n", " m_b = numpy.polyfit(y=rn[col], x=int_x, deg=1) \n", " fit_func = numpy.poly1d(m_b)\n", " correlation = numpy.corrcoef(x=int_x, y=rn[col])\n", " return int_x, fit_func(int_x), m_b, correlation" ] }, { "cell_type": "code", "execution_count": 19, "metadata": {}, "outputs": [], "source": [ "def calc_slope(dframe, days_back, roll=7, log=False):\n", " col = dframe.columns[0]\n", " rn = numpy.log(dframe) if log else dframe.copy()\n", " rn = rn.rolling(roll).mean().iloc[-days_back:]\n", " \n", " int_x = rn.index.astype(int) / 1e9 / 3600 / 24\n", " m, b = numpy.polyfit(y=rn[col], x=int_x, deg=1)\n", " corrcoef = numpy.corrcoef(x=int_x, y=rn[col])\n", " return {\n", " 'm': m,\n", " 'b': b,\n", " 'rsq': corrcoef[0,1]**2\n", " }" ] }, { "cell_type": "code", "execution_count": 20, "metadata": {}, "outputs": [], "source": [ "def do_curves(dframe):\n", " dfn = numpy.log(dframe[dframe.columns[0]].to_frame(f'ln_{dframe.columns[0]}'))\n", "\n", " windows = [14,28,35,56]\n", "\n", " all_fits = [\n", " (pd.DataFrame(index=fit[0], data=fit[1], columns=[f'slope{w}']), fit)\n", " for w, fit in [\n", " (w1, add_slope(dfn, w1, 7))\n", " for w1 in windows \n", " ]\n", " ]\n", "\n", "\n", " dfall = dfn.set_index(dfn.index.astype(int)).rolling(7).mean().iloc[-windows[-1]:]\n", " allfits = []\n", " for fit_frame, fit in all_fits:\n", " dfall = pd.merge(\n", " dfall, \n", " fit_frame, \n", " how='outer', \n", " left_index=True, \n", " right_index=True\n", " )\n", " all_fits.append(fit)\n", " dfall.set_index(dfall.index.astype('datetime64[ns]'), inplace=True)\n", " return dfall, allfits" ] }, { "cell_type": "code", "execution_count": 21, "metadata": {}, "outputs": [], "source": [ "def calc_all_slopes(dframe):\n", " rows = []\n", " for col in dframe.columns:\n", " windows = [14,28,42,56]\n", " data = {'spyid': col}\n", " for w in windows:\n", " slope = calc_slope(dframe[dframe[col].notna()][[col]], w, 1, False)\n", " data.update({f'{k}{w}': v for k,v in slope.items()})\n", " rows.append(data)\n", " return rows" ] }, { "cell_type": "code", "execution_count": 22, "metadata": {}, "outputs": [], "source": [ "def get_spyids():\n", " with get_cursor() as cur:\n", " cur.execute('''select spyid from spotify_daily_streams''')\n", " return [r[0] for r in cur.fetchall()]" ] }, { "cell_type": "code", "execution_count": 23, "metadata": {}, "outputs": [], "source": [ "def load_timeseries_from_db(previous_spyids=[], only_these_ids=None):\n", " all_spyids = only_these_ids or get_spyids()\n", " new_spyids = [n for n in all_spyids if n not in previous_spyids]\n", " tens = chunks(new_spyids, 100)\n", " loaded = []\n", " for ten in list(tens):\n", " print(ten)\n", " loaded.append(load_streams(ten))\n", " \n", " merged = loaded[0].copy()\n", " for df in loaded[1:]:\n", " merged = pd.merge(merged, df, left_index=True, right_index=True, how='outer')\n", " return merged" ] }, { "cell_type": "code", "execution_count": 24, "metadata": {}, "outputs": [], "source": [ "def make_csv_from_frames(slopedf, alldf, tsdf):\n", " def expand_row(row):\n", " spyid = row.loc['spyid']\n", " data_row = tsdf[spyid]\n", " data_row = data_row[~data_row.isna()]\n", " return pd.Series([\n", " data_row.max(),\n", " data_row.idxmax().date(),\n", " data_row.iloc[-1],\n", " data_row.index.max().date(),\n", " data_row.index.min().date(),\n", " data_row.resample('7D').mean().iloc[-1],\n", " data_row.iloc[-365:].astype(int).tolist()\n", " ], index=[\n", " 'Max',\n", " 'Date of Max',\n", " 'Last Streams',\n", " 'As of',\n", " 'Start Date',\n", " '7 Day Avg.',\n", " 'Daily Streams'\n", " ])\n", "\n", " finaldf = pd.merge(slopedf.set_index('spyid'), alldf, left_index=True, right_index=True)\n", " finaldf['spyid'] = finaldf.index\n", " finaldf['Link'] = finaldf.spyid.apply(lambda s: \"https://open.spotify.com/artist/\" + s)\n", " finaldf.rename({\n", " 'first': 'Artist',\n", " 'm14': 'Slope 14',\n", " 'rsq14': 'RSq 14',\n", " 'm28': 'Slope 28',\n", " 'rsq28': 'RSq 28',\n", " 'ts': 'Daily Plays',\n", " 'last_release_date': 'Last Released',\n", " 'array_accum.1': 'Genres',\n", " 'artist_popularity': 'Artist Popularity',\n", " 'monthly_listeners': 'Monthly Listeners',\n", " }, axis='columns', inplace=True)\n", " return pd.merge(finaldf, finaldf.apply(expand_row, axis=1), left_index=True, right_index=True)\n", "\n", "def create_csv(only_ids, filter_slopes=True):\n", " all_artists_fr = pd.concat([\n", " pd.read_csv('/Users/joel/Desktop/all_40_60.tsv', sep='\\t').set_index('artist_spyid'),\n", " pd.read_csv('/Users/joel/Desktop/all_61_70.tsv', sep='\\t').set_index('artist_spyid'),\n", " ])\n", " \n", " tsdf = load_timeseries_from_db(only_these_ids=only_ids)\n", " slopes = pd.DataFrame(calc_all_slopes(tsdf))\n", " if filter_slopes:\n", " slopes = slopes[(slopes.m28 > 0) & (slopes.m14 > slopes.m28 - 100) & (slopes.rsq28 > 0.3) & (slopes.rsq14 > 0.4)].sort_values('m14', ascending=False)\n", " \n", " return make_csv_from_frames(slopes, all_artists_fr, tsdf)[[\n", " 'Artist',\n", " 'Link',\n", " 'Last Streams',\n", " 'Daily Streams',\n", " 'As of',\n", " 'Max',\n", " 'Date of Max',\n", " 'Start Date',\n", " '7 Day Avg.',\n", " 'Slope 14',\n", " 'RSq 14',\n", " 'Slope 28',\n", " 'RSq 28',\n", " ]]\n", " " ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "No buildsha.py defined. sha will be 'NOT_SET'\n", "Not using secrets manager\n", "use_my_whitelist_cache False\n" ] } ], "source": [ "from tracker.bangers import query_to_sheet\n", "\n", "from tracker.sheet import Sheet" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "from tracker import bangers" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [], "source": [ "sheet = Sheet()" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "ss = sheet.client.create('Spotify Followers May 25')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Spotify Followers Slope" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "from tracker import db;" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [], "source": [ "db.setup_session()" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [], "source": [ "matrix_sql = '''\n", "with\n", "never_signed as (\n", " select\n", " artist_spyid, max(release_date) as last_release\n", " from spy_track_olap\n", " where release_date > current_date - 365\n", " and artist_popularity > 10\n", " group by 1 having not bool_or(is_signed)\n", "),\n", "\n", "followers0 as (\n", " select\n", " f0.artist_spyid,\n", " f0.followers f0,\n", " f7.followers as f7,\n", " f14.followers as f14,\n", " f0.followers - f7.followers as d7,\n", " f7.followers - f14.followers as d14\n", " from spy_artist_metrics f0\n", " join spy_artist_metrics f7 on f7.artist_spyid = f0.artist_spyid and f7.as_of::date = current_date - 7\n", " join spy_artist_metrics f14 on f14.artist_spyid = f0.artist_spyid and f14.as_of::date = current_date - 14\n", " where f0.as_of::date = current_date\n", " and f0.artist_spyid in (\n", " select artist_spyid from never_signed\n", " )\n", ")\n", "select\n", " sa.spyid,\n", " (d7)::float / f7::float as p7,\n", " (d14)::float / f14::float as p14,\n", " f.*\n", "from followers0 f\n", "join spy_artists sa on sa.spyid = f.artist_spyid\n", "where f0 between 5000 and 10000000\n", " and d7 > d14\n", " and d14 > 500\n", " and d7::float / f7::float > 0.1\n", "order by p7 desc\n", "'''" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [], "source": [ "results = db.Session.execute(matrix_sql).fetchall()" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "74" ] }, "execution_count": 13, "metadata": {}, "output_type": "execute_result" } ], "source": [ "len(results)" ] }, { "cell_type": "code", "execution_count": 27, "metadata": {}, "outputs": [], "source": [ "follower_rows = bangers._create_csv_for_statistics_matrix(results, bangers.data_from_artist_spyids_query, \n", " ['p7','p14','spyid2','f0','f7','f14','d7','d14'])" ] }, { "cell_type": "code", "execution_count": 28, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "75" ] }, "execution_count": 28, "metadata": {}, "output_type": "execute_result" } ], "source": [ "len(follower_rows)" ] }, { "cell_type": "code", "execution_count": 29, "metadata": {}, "outputs": [], "source": [ "# ss = sheet.client.create('Spotify One - Apr 3')" ] }, { "cell_type": "code", "execution_count": 17, "metadata": {}, "outputs": [], "source": [ "ss.share('joel@whtlst.in', perm_type='user', role='writer', notify=False)" ] }, { "cell_type": "code", "execution_count": 48, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "" ] }, "execution_count": 48, "metadata": {}, "output_type": "execute_result" } ], "source": [ "query_to_sheet.csv_to_sheet(with_streams, ss)" ] }, { "cell_type": "code", "execution_count": 25, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "['spyid',\n", " 'link',\n", " 'Name',\n", " 'Current Followers',\n", " 'Followers',\n", " 'Daily Follower Growth',\n", " 'spyid',\n", " 'p7',\n", " 'p14',\n", " 'spyid2',\n", " 'f0',\n", " 'f7',\n", " 'f14',\n", " 'd7',\n", " 'd14']" ] }, "execution_count": 25, "metadata": {}, "output_type": "execute_result" } ], "source": [ "follower_rows[0]" ] }, { "cell_type": "code", "execution_count": 26, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "['2amDdZfkKjK50RDrUlDcTc',\n", " 'https://bot.whtlst.in/#/artist/spy/2amDdZfkKjK50RDrUlDcTc',\n", " 'Barlito',\n", " 7179,\n", " [78,\n", " 88,\n", " 102,\n", " 108,\n", " 126,\n", " 162,\n", " 827,\n", " 736,\n", " 1696,\n", " 2854,\n", " 4067,\n", " 5348,\n", " 6025,\n", " 6511,\n", " 6903,\n", " 7179],\n", " [12, 10, 14, 6, 18, 36, 665, -91, 960, 1158, 1213, 1281, 677, 486, 392, 276],\n", " 3.2329009433962264,\n", " 18.272727272727273,\n", " '2amDdZfkKjK50RDrUlDcTc',\n", " 7179,\n", " 1696,\n", " 88,\n", " 5483,\n", " 1608]" ] }, "execution_count": 26, "metadata": {}, "output_type": "execute_result" } ], "source": [ "follower_rows[1]" ] }, { "cell_type": "code", "execution_count": 35, "metadata": {}, "outputs": [], "source": [ "from tracker.unicorn import artist_ingestion, mysql" ] }, { "cell_type": "code", "execution_count": 32, "metadata": {}, "outputs": [], "source": [ "artists_ids = [r[0] for r in results]" ] }, { "cell_type": "code", "execution_count": 40, "metadata": {}, "outputs": [], "source": [ "def get_ts_data(spyid):\n", " return [t.split(',')[1] for t in \n", " mysql.query_one('select tsdata, max from spotify_daily_streams where spyid = %s', [spyid]).splitlines()\n", " if 'num' not in t]" ] }, { "cell_type": "code", "execution_count": 42, "metadata": {}, "outputs": [], "source": [ "def add_daily_streams_to_csv(rows, has_header=True, position=4):\n", " added = []\n", " if has_header:\n", " rows[0].insert(position, 'Daily Streams')\n", " added.append(rows[0])\n", " \n", " for row in rows[(1 if has_header else 0):]:\n", " ts_data = get_ts_data(row[0])\n", " new_row = row[:]\n", " new_row.insert(position, ts_data)\n", " added.append(new_row)\n", " return added" ] }, { "cell_type": "code", "execution_count": 43, "metadata": {}, "outputs": [], "source": [ "with_streams = add_daily_streams_to_csv(follower_rows)" ] }, { "cell_type": "code", "execution_count": 34, "metadata": {}, "outputs": [], "source": [ "# artist_ingestion.store_batch_of_daily_streams(artists_ids)" ] }, { "cell_type": "code", "execution_count": 47, "metadata": {}, "outputs": [ { "data": { "text/plain": [ "15" ] }, "execution_count": 47, "metadata": {}, "output_type": "execute_result" } ], "source": [ "len(with_streams[1])" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.7.1" } }, "nbformat": 4, "nbformat_minor": 2 }