<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Loading and Cleaning the Data
Turn on inline matplotlib plotting and import plotting dependencies.
Step1:  Import analytic depedencies. Doc code for spark-timeseries and source code for tsanalysis.
Step2:  Load wiki page view and stock price data into Spark Dataframes.
wiki_obs is a Spark dataframe of (timestamp, page, views) of types (Timestamp, String, Double). ticker_obs is a Spark dataframe of (timestamp, symbol, price) of types (Timestamp, String, Double).
Step3:  Display the first 5 elements of the wiki_obs RDD. 
wiki_obs contains Row objects with the fields (timestamp, page, views).
Step4:  Display the first 5 elements of the tickers_obs RDD.
ticker_obs contains Row objects with the fields (timestamp, symbol, views).
Step5:  Create datetime index.
Create time series RDD from observations and index. Remove time instants with NaNs.
Cache the tsrdd.
Examine the first element in the RDD.
Time series have values and a datetime index. We can create a tsrdd for hourly stock prices from an index and a Spark DataFrame of observations. ticker_tsrdd is an RDD of tuples where each tuple has the form (ticker symbol, stock prices) where ticker symbol is a string and stock prices is a 1D np.ndarray. We create a nicely formatted string representation of this pair in print_ticker_info(). Notice how we access the two elements of the tuple.
Step6:  Create a wiki page view tsrdd and set the index to match the index of ticker_tsrdd.
 Linearly interpolate to impute missing values.
wiki_tsrdd is an RDD of tuples where each tuple has the form (page title, wiki views) where page title is a string and wiki views is a 1D np.ndarray. We have cached both RDDs because we will be doing many subsequent operations on them.
Step7:  Filter out symbols with more than the minimum number of NaNs.
 Then filter out instants with NaNs.
Step8:  Linking symbols and pages
We need to join together the wiki page and ticker data, but the time series RDDs are not directly joinable on their keys. To overcome this, we have create a dict from wikipage title to stock  ticker symbol.
Create a dict from ticker symbols to page names.
Create another from page names to ticker symbols.
Step9:  Join together wiki_tsrdd and ticker_tsrdd.
First, we use this dict to look up the corresponding stock ticker symbol and rekey the wiki page view time series. We then join the data sets together. The result is an RDD of tuples where each element is of the form (ticker_symbol, (wiki_series, ticker_series)). We count the number of elements in the resulting rdd to see how many matches we have.
Step11:  Correlation and Relationships
Define a function for computing the pearson r correlation of the stock price and wiki page traffic associated with a  company.
Here we look up a specific stock and corrsponding wiki page, and provide an example of 
computing the pearson correlation locally. We use scipy.stats.stats.pearsonr to compute the pearson correlation and corresponding two sided p value. wiki_vol_corr and corr_with_offset both return this as a tuple of (corr, p_value).
Step12:  Create a plot of the joint distribution of wiki trafiic and stock prices for a specific company using seaborn's jointplot function.
Step13:  Find the companies with the highest correlation between stock prices time series and wikipedia page traffic.
Note that comparing a tuple means you compare the composite object lexicographically.
Step14:  Add in filtering out less than useful correlation results. 
There are a lot of invalid correlations that get computed, so lets filter those out.
Step15:  Find the top 10 correlations as defined by the ordering on tuples.
Step16:  Create a joint plot of some of the stronger relationships.
Step17:  Volatility
 Compute per-day volatility for each symbol.
Step18:  Make sure we don't have any NaNs.
Step19:  Visualize volatility
Plot daily volatility in stocks over time.
Step20:  What does the distribution of volatility for the whole market look like? Add volatility for individual stocks in a datetime bin.
Step21:  Find stocks with the highest average daily volatility.
Step22:  Plot stocks with the highest average daily volatility over time.
Step23:  We first map over ticker_daily_vol to find the index of the value with the highest volatility. We then relate that back to the index set on the RDD to find the corresponding datetime.
Step24:  A large number of stock symbols had their most volatile days on August 24th and August 25th of
this year.
Regress volatility against page views
Resample the wiki page view data set so we have total pageviews by day. 
 Cache the wiki page view RDD.
Resample the wiki page view data set so we have total pageviews by day. This means reindexing the time series and aggregating data together with daily buckets. We use np.nansum to add up numbers while treating nans like zero.
Step25:  Validate data by checking for nans.
Step26:  Fit a linear regression model to every pair in the joined wiki-ticker RDD and extract R^2 scores.
Step27:  Print out the symbols with the highest R^2 scores.
Step28:  Plot the results of a linear model.
Plotting a linear model always helps me understand it better. Again, seaborn is super useful with smart defaults built in.
Step29:  Box plot / Tukey outlier identification
Tukey originally proposed a method for identifying outliers in bow and whisker plots. Eseentially, we find the cut off value for the 75th percentile $P_{75} = percentile(sample, 0.75)$, and add a reasonable buffer (expressed in terms of the interquartile range) $1.5IQR = 1.5(P_{75}-P_{25})$ above that cutoff.
Write a function that returns the high value cutoff for Tukey's boxplot outlier criterion.
Step30:  Filter out any values below Tukey's boxplot criterion for outliers.
Step31:  Black Monday
Select the date range comprising Black Monday 2015.
Step32:  Which stocks saw the worst return for that day?
Step33:  Plot wiki page views for one of those stocks.
<ASSISTANT_TASK:>
Python Code:
%matplotlib inline
import matplotlib as mpl
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import seaborn as sns
Explanation: Loading and Cleaning the Data
Turn on inline matplotlib plotting and import plotting dependencies.
End of explanation
import numpy as np
import pandas as pd
import tsanalysis.loaddata as ld
import tsanalysis.tsutil as tsutil
import sparkts.timeseriesrdd as tsrdd
import sparkts.datetimeindex as dtindex
from sklearn import linear_model
Explanation: Import analytic depedencies. Doc code for spark-timeseries and source code for tsanalysis.
End of explanation
wiki_obs = ld.load_wiki_df(sqlCtx, '/user/srowen/wiki.tsv')
ticker_obs = ld.load_ticker_df(sqlCtx, '/user/srowen/ticker.tsv')
Explanation: Load wiki page view and stock price data into Spark Dataframes.
wiki_obs is a Spark dataframe of (timestamp, page, views) of types (Timestamp, String, Double). ticker_obs is a Spark dataframe of (timestamp, symbol, price) of types (Timestamp, String, Double).
End of explanation
wiki_obs.head(5)
Explanation: Display the first 5 elements of the wiki_obs RDD. 
wiki_obs contains Row objects with the fields (timestamp, page, views).
End of explanation
ticker_obs.head(5)
Explanation: Display the first 5 elements of the tickers_obs RDD.
ticker_obs contains Row objects with the fields (timestamp, symbol, views).
End of explanation
def print_ticker_info(ticker):
    print ('The first ticker symbol is: {} \nThe first 20 elements of the associated ' +\
    'series are:\n {}').format(ticker[0], ticker[1][:20])
dt_index = dtindex.uniform('2015-08-03', end='2015-09-22', freq=dtindex.HourFrequency(1, sc), sc=sc)
ticker_tsrdd = tsrdd.time_series_rdd_from_observations(dt_index, ticker_obs, 'timestamp', 'symbol',
                                                 'price') \
    .remove_instants_with_nans()
ticker_tsrdd.cache()
first_series = ticker_tsrdd.first()
    
print_ticker_info(first_series)
Explanation: Create datetime index.
Create time series RDD from observations and index. Remove time instants with NaNs.
Cache the tsrdd.
Examine the first element in the RDD.
Time series have values and a datetime index. We can create a tsrdd for hourly stock prices from an index and a Spark DataFrame of observations. ticker_tsrdd is an RDD of tuples where each tuple has the form (ticker symbol, stock prices) where ticker symbol is a string and stock prices is a 1D np.ndarray. We create a nicely formatted string representation of this pair in print_ticker_info(). Notice how we access the two elements of the tuple.
End of explanation
wiki_tsrdd = tsrdd.time_series_rdd_from_observations(ticker_tsrdd.index(), wiki_obs, 'timestamp', 'page', 'views') \
    .fill('linear')
wiki_tsrdd.cache()
Explanation: Create a wiki page view tsrdd and set the index to match the index of ticker_tsrdd.
 Linearly interpolate to impute missing values.
wiki_tsrdd is an RDD of tuples where each tuple has the form (page title, wiki views) where page title is a string and wiki views is a 1D np.ndarray. We have cached both RDDs because we will be doing many subsequent operations on them.
End of explanation
def count_nans(vec):
    return np.count_nonzero(np.isnan(vec))
ticker_min_nans = ticker_tsrdd.map(lambda x: count_nans(x[1])).min()
ticker_price_tsrdd = ticker_tsrdd.filter(lambda x: count_nans(x[1]) == ticker_min_nans) \
    .remove_instants_with_nans()
ticker_return_tsrdd = ticker_price_tsrdd.return_rates()
print_ticker_info(ticker_return_tsrdd.first())
Explanation: Filter out symbols with more than the minimum number of NaNs.
 Then filter out instants with NaNs.
End of explanation
# a dict from wiki page name to ticker symbol
page_symbols = {}
for line in open('../symbolnames.tsv').readlines():
    tokens = line[:-1].split('\t')
    page_symbols[tokens[1]] = tokens[0]
def get_page_symbol(page_series):
    if page_series[0] in page_symbols:
        return [(page_symbols[page_series[0]], page_series[1])]
    else:
        return []
# reverse keys and values. a dict from ticker symbol to wiki page name.
symbol_pages = dict(zip(page_symbols.values(), page_symbols.keys()))
print page_symbols.items()[0]
print symbol_pages.items()[0]
Explanation: Linking symbols and pages
We need to join together the wiki page and ticker data, but the time series RDDs are not directly joinable on their keys. To overcome this, we have create a dict from wikipage title to stock  ticker symbol.
Create a dict from ticker symbols to page names.
Create another from page names to ticker symbols.
End of explanation
joined = wiki_tsrdd.flatMap(get_page_symbol).join(ticker_tsrdd)
joined.cache()
joined.count()
Explanation: Join together wiki_tsrdd and ticker_tsrdd.
First, we use this dict to look up the corresponding stock ticker symbol and rekey the wiki page view time series. We then join the data sets together. The result is an RDD of tuples where each element is of the form (ticker_symbol, (wiki_series, ticker_series)). We count the number of elements in the resulting rdd to see how many matches we have.
End of explanation
from scipy.stats.stats import pearsonr
def wiki_vol_corr(page_key):
    # lookup individual time series by key.
    ticker = ticker_tsrdd.find_series(page_symbols[page_key]) # numpy array
    wiki = wiki_tsrdd.find_series(page_key) # numpy array
    return pearsonr(ticker, wiki)
def corr_with_offset(page_key, offset):
    offset is an integer that describes how many time intervals we have slid
    the wiki series ahead of the ticker series.
    ticker = ticker_tsrdd.find_series(page_symbols[page_key]) # numpy array
    wiki = wiki_tsrdd.find_series(page_key) # numpy array
    return pearsonr(ticker[offset:], wiki[:-offset])
print wiki_vol_corr('Netflix')
print corr_with_offset('Netflix', 1)
Explanation: Correlation and Relationships
Define a function for computing the pearson r correlation of the stock price and wiki page traffic associated with a  company.
Here we look up a specific stock and corrsponding wiki page, and provide an example of 
computing the pearson correlation locally. We use scipy.stats.stats.pearsonr to compute the pearson correlation and corresponding two sided p value. wiki_vol_corr and corr_with_offset both return this as a tuple of (corr, p_value).
End of explanation
def joint_plot(page_key, ticker, wiki, offset=0):
    with sns.axes_style("white"):
        sns.jointplot(x=ticker, y=wiki, kind="kde", color="b");
        plt.xlabel('Stock Price')
        plt.ylabel('Wikipedia Page Views')
        plt.title('Joint distribution of {} stock price\n and Wikipedia page views.'\
            +'\nWith a {} day offset'.format(page_key, offset), y=1.20)
def ticker_wiki_joint_plot(page_key, offset=0):
    if (offset == 0):
        ticker = ticker_tsrdd.find_series(page_symbols[page_key]) # numpy array
        wiki = wiki_tsrdd.find_series(page_key)
    else:
        ticker = ticker_tsrdd.find_series(page_symbols[page_key])[offset:]
        wiki = wiki_tsrdd.find_series(page_key)[:-offset]
    joint_plot(page_key, ticker, wiki)
        
ticker_wiki_joint_plot('Apple_Inc.', offset=0)
Explanation: Create a plot of the joint distribution of wiki trafiic and stock prices for a specific company using seaborn's jointplot function.
End of explanation
joined.mapValues(lambda wiki_ticker: pearsonr(wiki_ticker[0], wiki_ticker[1])) \
    .sortBy(keyfunc=lambda x: x[1]) \
    .collect()
Explanation: Find the companies with the highest correlation between stock prices time series and wikipedia page traffic.
Note that comparing a tuple means you compare the composite object lexicographically.
End of explanation
top_correlations = joined.mapValues(lambda wiki_ticker: pearsonr(wiki_ticker[0], wiki_ticker[1])) \
    .filter(lambda corr_tuple: not np.isnan(corr_tuple[1][0]))\
    .sortBy(keyfunc=lambda x: np.abs(x[1][0])) \
    .collect()
top_correlations
Explanation: Add in filtering out less than useful correlation results. 
There are a lot of invalid correlations that get computed, so lets filter those out.
End of explanation
top_10_corr = top_correlations[-10:]
top_10_corr_tickers = dict(top_10_corr).keys()
top_10_corr_tickers
Explanation: Find the top 10 correlations as defined by the ordering on tuples.
End of explanation
pages = [symbol_pages[sym] for sym in top_10_corr_tickers]
ticker_wiki_joint_plot(pages[0], offset=1)
Explanation: Create a joint plot of some of the stronger relationships.
End of explanation
ticker_daily_vol = tsutil.sample_daily(ticker_tsrdd, 'std') \
    .remove_instants_with_nans()
print ticker_daily_vol.first()
Explanation: Volatility
 Compute per-day volatility for each symbol.
End of explanation
ticker_daily_vol.map(lambda x: count_nans(x[1])).top(1)
Explanation: Make sure we don't have any NaNs.
End of explanation
daily_vol_df = ticker_daily_vol.to_pandas_dataframe()
daily_vol_df.plot()
Explanation: Visualize volatility
Plot daily volatility in stocks over time.
End of explanation
sns.distplot(daily_vol_df.sum(axis=0), kde=False)
plt.xlabel('Volatility')
plt.ylabel('Fraction of observations')
ax = sns.heatmap(daily_vol_df)
plt.xlabel('Company')
plt.ylabel('Date')
Explanation: What does the distribution of volatility for the whole market look like? Add volatility for individual stocks in a datetime bin.
End of explanation
most_volatile_on_ave_stocks = tsutil.sample_daily(ticker_return_tsrdd, 'std') \
    .remove_instants_with_nans() \
    .mapValues(lambda x: (x, x.mean())) \
    .top(10, lambda x: x[1][1])
[(stock_record[0], stock_record[1]) for stock_record in most_volatile_on_ave_stocks]
Explanation: Find stocks with the highest average daily volatility.
End of explanation
most_volatile_on_ave_stocks_set = set([x[0] for x in most_volatile_on_ave_stocks])
ticker_pandas_df = ticker_daily_vol.filter(lambda x: x[0] in most_volatile_on_ave_stocks_set) \
    .to_pandas_dataframe()
    
ticker_pandas_df.plot()
Explanation: Plot stocks with the highest average daily volatility over time.
End of explanation
most_volatile_days = ticker_daily_vol.map(lambda x: (np.argmax(x[1]), 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .top(10, lambda x: x[1])
def lookup_datetime(rdd, loc):
    return rdd.index().datetime_at_loc(loc)
[(lookup_datetime(ticker_daily_vol, day), count ) for day, count in most_volatile_days]
Explanation: We first map over ticker_daily_vol to find the index of the value with the highest volatility. We then relate that back to the index set on the RDD to find the corresponding datetime.
End of explanation
from numpy import nansum
wiki_tsrdd_full = tsrdd.time_series_rdd_from_observations(dt_index, wiki_obs, 'timestamp',
                                                    'page', 'views')
wiki_daily_views = tsutil.sample_daily(wiki_tsrdd_full, nansum) \
    .with_index(ticker_daily_vol.index())
wiki_daily_views.cache()
Explanation: A large number of stock symbols had their most volatile days on August 24th and August 25th of
this year.
Regress volatility against page views
Resample the wiki page view data set so we have total pageviews by day. 
 Cache the wiki page view RDD.
Resample the wiki page view data set so we have total pageviews by day. This means reindexing the time series and aggregating data together with daily buckets. We use np.nansum to add up numbers while treating nans like zero.
End of explanation
wiki_daily_views.map(lambda x: count_nans(x[1])).top(1)
most_viewy_days = wiki_daily_views.map(lambda x: (np.argmax(x[1]), 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .top(10, lambda x: x[1])
[(lookup_datetime(ticker_daily_vol, day), count ) for day, count in most_viewy_days]
Explanation: Validate data by checking for nans.
End of explanation
def regress(X, y):
    model = linear_model.LinearRegression()
    model.fit(X, y)
    score = model.score(X, y)
    return (score, model)
lag = 2
lead = 2
joined = regressions = wiki_daily_views.flatMap(get_page_symbol) \
    .join(ticker_daily_vol)
    
models = joined.mapValues(lambda x: regress(tsutil.lead_and_lag(lead, lag, x[0]), x[1][lag:-lead]))
models.cache()
models.count()
Explanation: Fit a linear regression model to every pair in the joined wiki-ticker RDD and extract R^2 scores.
End of explanation
models.map(lambda x: (x[1][0], x[0])).top(5)
Explanation: Print out the symbols with the highest R^2 scores.
End of explanation
page_views, returns = joined.filter(lambda x: x[0] == 'V').first()[1]
sns.set(color_codes=True)
sns.regplot(x=page_views, y=returns)
Explanation: Plot the results of a linear model.
Plotting a linear model always helps me understand it better. Again, seaborn is super useful with smart defaults built in.
End of explanation
def tukey_high_outlier_cutoff(sample):
    sample = sorted(sample)
    first_quartile = sample[len(sample) // 4]
    third_quartile = sample[len(sample) * 3 // 4]
    iqr = third_quartile - first_quartile
    return third_quartile + iqr * 1.5
Explanation: Box plot / Tukey outlier identification
Tukey originally proposed a method for identifying outliers in bow and whisker plots. Eseentially, we find the cut off value for the 75th percentile $P_{75} = percentile(sample, 0.75)$, and add a reasonable buffer (expressed in terms of the interquartile range) $1.5IQR = 1.5(P_{75}-P_{25})$ above that cutoff.
Write a function that returns the high value cutoff for Tukey's boxplot outlier criterion.
End of explanation
cutoff = tukey_high_outlier_cutoff(returns)
filter(lambda x: x > cutoff, returns)
cutoff = tukey_high_outlier_cutoff(page_views)
filter(lambda x: x > cutoff, page_views)
Explanation: Filter out any values below Tukey's boxplot criterion for outliers.
End of explanation
black_monday = ticker_price_tsrdd['2015-08-24 13:00:00':'2015-08-24 20:00:00']
Explanation: Black Monday
Select the date range comprising Black Monday 2015.
End of explanation
black_monday.mapValues(lambda x: (x[-1] / x[0]) - 1) \
    .takeOrdered(10, lambda x: x[1])
Explanation: Which stocks saw the worst return for that day?
End of explanation
series = wiki_daily_views \
    .to_pandas_series_rdd() \
    .flatMap(get_page_symbol) \
    .filter(lambda x: x[0] == 'HCA') \
    .first()
series[1].plot()
Explanation: Plot wiki page views for one of those stocks.
End of explanation
<END_TASK>