<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Label and feature engineering
This lab is optional. It demonstrates advanced SQL queries for time-series engineering. For real-world problems, this type of feature engineering code is essential. If you are pursuing a time-series project for open project week, feel free to use this code as a template. 
Learning objectives
Step1:  Create time-series features and determine label based on market movement
Summary of base tables
TODO
Step2:  Label engineering
Ultimately, we need to end up with a single label for each day. The label takes on 3 values
Step3:  TODO
Step5:  Add time series features
<h3><font color="#4885ed">Compute price features using analytics functions</font> </h3>
In addition, we will also build time-series features using the min, max, mean, and std (can you think of any over functions to use?). To do this, let's use analytic functions in BigQuery (also known as window functions). 
An analytic function is a function that computes aggregate values over a group of rows. Unlike aggregate functions, which return a single aggregate value for a group of rows, analytic functions return a single value for each row by computing the function over a group of input rows.
 Using the AVG analytic function, we can compute the average close price of a given symbol over the past week (5 business days)
Step6:  Compute percentage change, then self join with prices from S&P index.
We will also compute price change of S&P index, GSPC. We do this so we can compute the normalized percentage change.
<h3><font color="#4885ed">Compute normalized price change (%)</font> </h3>
Before we can create our labels we need to normalize the price change using the S&P 500 index. The normalization using the S&P index fund helps ensure that the future price of a stock is not due to larger market effects. Normalization helps us isolate the factors contributing to the performance of a stock_market.
Let's use the normalization scheme from by subtracting the scaled difference in the S&P 500 index during the same time period.
In Python
Step7:  Compute normalized price change (shown above).
Let's join scaled price change (tomorrow_close / close) with the gspc symbol (symbol for the S&P index). Then we can normalize using the scheme described above.
Learning objective 3
TODO
Step8:  Verify results
Step9:  <h3><font color="#4885ed">Join with S&P 500 table and Create labels
Step10:  TODO
Step11:  The dataset is still quite large and the majority of the days the market STAYs. Let's focus our analysis on dates where earnings per share (EPS) information is released by the companies. The EPS data has 3 key columns surprise, reported_EPS, and consensus_EPS
Step12:  The surprise column indicates the difference between the expected (consensus expected eps by analysts) and the reported eps. We can join this table with our derived table to focus our analysis during earnings periods
Step14:  Feature exploration
Now that we have created our recent movements of the company’s stock price, let's visualize our features. This will help us understand the data better and possibly spot errors we may have made during our calculations.
As a reminder, we calculated the scaled prices 1 week, 1 month, and 1 year before the date that we are predicting at. 
Let's write a re-usable function for aggregating our features.
Learning objective 2
Step15:  TODO Use the get_aggregate_stats from above to visualize the normalized_change column.
Step16:  Let's look at results by day-of-week, month, etc.
<ASSISTANT_TASK:>
Python Code:
PROJECT = 'your-gcp-project'  # Replace with your project ID.
import pandas as pd
from google.cloud import bigquery
from IPython.core.magic import register_cell_magic
from IPython import get_ipython
bq = bigquery.Client(project = PROJECT)
# Allow you to easily have Python variables in SQL query.
@register_cell_magic('with_globals')
def with_globals(line, cell):
    contents = cell.format(**globals())
    if 'print' in line:
        print(contents)
    get_ipython().run_cell(contents)
def create_dataset():
    dataset = bigquery.Dataset(bq.dataset("stock_market"))
    try:
        bq.create_dataset(dataset)  # Will fail if dataset already exists.
        print("Dataset created")
    except:
        print("Dataset already exists")
create_dataset()
Explanation: Label and feature engineering
This lab is optional. It demonstrates advanced SQL queries for time-series engineering. For real-world problems, this type of feature engineering code is essential. If you are pursuing a time-series project for open project week, feel free to use this code as a template. 
Learning objectives:
Learn how to use BigQuery to build time-series features and labels for forecasting
Learn how to visualize and explore features.
Learn effective scaling and normalizing techniques to improve our modeling results
Note: In the previous lab we explored the data, if you haven’t run the previous notebook, go back to optional_1_data_exploration.ipynb and run it.
Now that we have explored the data, let's start building our features, so we can build a model.
<h3><font color="#4885ed">Feature Engineering</font> </h3>
Use the price_history table, we can look at past performance of a given stock, to try to predict it's future stock price. In this notebook we will be focused on cleaning and creating features from this table. 
There are typically two different approaches to creating features with time-series data. 
One approach is aggregate the time-series into "static" features, such as "min_price_over_past_month" or "exp_moving_avg_past_30_days". Using this approach, we can use a deep neural network or a more "traditional" ML model to train. Notice we have essentially removed all sequention information after aggregating. This assumption can work well in practice. 
A second approach is to preserve the ordered nature of the data and use a sequential model, such as a recurrent neural network. This approach has a nice benefit that is typically requires less feature engineering. Although, training sequentially models typically takes longer.
In this notebook, we will build features and also create rolling windows of the ordered time-series data.
<h3><font color="#4885ed">Label Engineering</font> </h3>
We are trying to predict if the stock will go up or down. In order to do this we will need to "engineer" our label by looking into the future and using that as the label. We will be using the LAG function in BigQuery to do this. Visually this looks like:
Import libraries; setup
End of explanation
%%with_globals
%%bigquery --project {PROJECT}
--# TODO
%%with_globals
%%bigquery --project {PROJECT}
--# TODO
Explanation: Create time-series features and determine label based on market movement
Summary of base tables
TODO: How many rows are in our base tables price_history and snp500?
End of explanation
%%with_globals print
%%bigquery df --project {PROJECT}
CREATE OR REPLACE TABLE `stock_market.price_history_delta`
AS
(
WITH shifted_price AS
  (
  SELECT *,
  (LAG(close, 1) OVER (PARTITION BY symbol order by Date DESC)) AS tomorrow_close
  FROM `stock_src.price_history`
  WHERE Close > 0
  )
SELECT a.*,
(tomorrow_close - Close) AS tomo_close_m_close
FROM shifted_price a
)
%%with_globals
%%bigquery --project {PROJECT}
SELECT *
FROM stock_market.price_history_delta
ORDER by Date
LIMIT 100
Explanation: Label engineering
Ultimately, we need to end up with a single label for each day. The label takes on 3 values: {down, stay, up}, where down and up indicates the normalized price (more on this below) went down 1% or more and up 1% or more, respectively. stay indicates the stock remained within 1%.
The steps are:
Compare close price and open price
Compute price features using analytics functions
Compute normalized price change (%)
Join with S&P 500 table
Create labels (up, down, stay)
<h3><font color="#4885ed">Compare close price and open price</font> </h3>
For each row, get the close price of yesterday and the open price of tomorrow using the LAG function. We will determine tomorrow's close - today's close.
Shift to get tomorrow's close price.
Learning objective 1
End of explanation
%%with_globals print
%%bigquery --project {PROJECT}
SELECT
--# TODO: verify the stock market is going up -- on average.
FROM
stock_market.price_history_delta
Explanation: TODO: Historically, we know that the stock market has been going up. Can you think of a way to verify this using our newly created table price_history_delta?
Learning objective 2
End of explanation
def get_window_fxn(agg_fxn, n_days):
    Generate a time-series feature. 
    
    E.g., Compute the average of the price over the past 5 days.
    SCALE_VALUE = 'close'
    sql = '''
    ({agg_fxn}(close) OVER (PARTITION BY (# TODO)
                      ORDER BY (# TODO)
                      ROWS BETWEEN {n_days} (# TODO)))/{scale}
                      AS close_{agg_fxn}_prior_{n_days}_days'''.format(
                          agg_fxn=agg_fxn, n_days=n_days, scale=SCALE_VALUE)
    return sql
WEEK = 5
MONTH = 20
YEAR = 52*5
agg_funcs = ('MIN', 'MAX', 'AVG', 'STDDEV')
lookbacks = (WEEK, MONTH, YEAR)
sqls = []
for fxn in agg_funcs:
    for lookback in lookbacks:
        sqls.append(get_window_fxn(fxn, lookback))
time_series_features_sql = ','.join(sqls)  # SQL string.
def preview_query():
    print(time_series_features_sql[0:1000])
preview_query()
%%with_globals print
%%bigquery --project {PROJECT}
CREATE OR REPLACE TABLE stock_market.price_features_delta
AS
SELECT * 
FROM 
    (SELECT *,
    {time_series_features_sql},
    -- Also get the raw time-series values; will be useful for the RNN model.
    (ARRAY_AGG(close) OVER (PARTITION BY symbol
                    ORDER BY Date
                    ROWS BETWEEN 260 PRECEDING AND 1 PRECEDING))
                    AS close_values_prior_260,
    ROW_NUMBER() OVER (PARTITION BY symbol ORDER BY Date) AS days_on_market
    FROM stock_market.price_history_delta)
WHERE days_on_market > {YEAR}
%%bigquery --project {PROJECT}
SELECT *
FROM stock_market.price_features_delta 
ORDER BY symbol, Date
LIMIT 10
Explanation: Add time series features
<h3><font color="#4885ed">Compute price features using analytics functions</font> </h3>
In addition, we will also build time-series features using the min, max, mean, and std (can you think of any over functions to use?). To do this, let's use analytic functions in BigQuery (also known as window functions). 
An analytic function is a function that computes aggregate values over a group of rows. Unlike aggregate functions, which return a single aggregate value for a group of rows, analytic functions return a single value for each row by computing the function over a group of input rows.
 Using the AVG analytic function, we can compute the average close price of a given symbol over the past week (5 business days):
python
 (AVG(close) OVER (PARTITION BY symbol
  ORDER BY Date
  ROWS BETWEEN 5 PRECEDING AND 1 PRECEDING)) / close
  AS close_avg_prior_5_days
Learning objective 1
TODO: Please fill in the # TODOs in the below query
End of explanation
scaled_change = (50.59 - 50.69) / 50.69
scaled_s_p = (939.38 - 930.09) / 930.09
normalized_change = scaled_change - scaled_s_p
print('''
scaled change: {:2.3f}
scaled_s_p: {:2.3f}
normalized_change: {:2.3f}
'''.format(scaled_change, scaled_s_p, normalized_change))
Explanation: Compute percentage change, then self join with prices from S&P index.
We will also compute price change of S&P index, GSPC. We do this so we can compute the normalized percentage change.
<h3><font color="#4885ed">Compute normalized price change (%)</font> </h3>
Before we can create our labels we need to normalize the price change using the S&P 500 index. The normalization using the S&P index fund helps ensure that the future price of a stock is not due to larger market effects. Normalization helps us isolate the factors contributing to the performance of a stock_market.
Let's use the normalization scheme from by subtracting the scaled difference in the S&P 500 index during the same time period.
In Python: 
```python
Example calculation.
scaled_change = (50.59 - 50.69) / 50.69
scaled_s_p = (939.38 - 930.09) / 930.09
normalized_change = scaled_change - scaled_s_p
assert normalized_change == ~1.2%
```
End of explanation
snp500_index = 'gspc'  
%%with_globals print
%%bigquery --project {PROJECT}
CREATE OR REPLACE TABLE stock_market.price_features_norm_per_change
AS
WITH 
all_percent_changes AS
    (
    SELECT *, (tomo_close_m_close / Close) AS scaled_change
    FROM `stock_market.price_features_delta`
    ),
s_p_changes AS
    (SELECT
     scaled_change AS s_p_scaled_change,
     date
     FROM all_percent_changes
     WHERE symbol="{snp500_index}")
SELECT all_percent_changes.*,
       s_p_scaled_change,
       (# TODO) AS normalized_change          
FROM 
    all_percent_changes LEFT JOIN s_p_changes
    --# Add S&P change to all rows
    ON all_percent_changes.date = s_p_changes.date
Explanation: Compute normalized price change (shown above).
Let's join scaled price change (tomorrow_close / close) with the gspc symbol (symbol for the S&P index). Then we can normalize using the scheme described above.
Learning objective 3
TODO: Please fill in the # TODO in the code below.
End of explanation
%%with_globals print
%%bigquery df --project {PROJECT}
SELECT *
FROM stock_market.price_features_norm_per_change
LIMIT 10
df.head()
Explanation: Verify results
End of explanation
down_thresh = -0.01
up_thresh = 0.01
Explanation: <h3><font color="#4885ed">Join with S&P 500 table and Create labels: {`up`, `down`, `stay`}</font> </h3>
Join the table with the list of S&P 500. This will allow us to limit our analysis to S&P 500 companies only.
Finally we can create labels. The following SQL statement should do:
sql
  CASE WHEN normalized_change &lt; -0.01 THEN 'DOWN'
       WHEN normalized_change &gt; 0.01 THEN 'UP'
       ELSE 'STAY'
  END
Learning objective 1
End of explanation
%%with_globals print
%%bigquery df --project {PROJECT}
CREATE OR REPLACE TABLE stock_market.percent_change_sp500
AS
SELECT *,
    CASE 
    --# TODO
    END AS direction
FROM stock_market.price_features_norm_per_change features
INNER JOIN `stock_src.snp500`
USING (symbol)
%%with_globals print
%%bigquery --project {PROJECT}
SELECT direction, COUNT(*) as cnt
FROM stock_market.percent_change_sp500
GROUP BY direction
%%with_globals print
%%bigquery df --project {PROJECT}
SELECT *
FROM stock_market.percent_change_sp500
LIMIT 20
df.columns
Explanation: TODO: Please fill in the CASE function below.
End of explanation
%%with_globals print
%%bigquery --project {PROJECT}
SELECT *
FROM `stock_src.eps`
LIMIT 10
Explanation: The dataset is still quite large and the majority of the days the market STAYs. Let's focus our analysis on dates where earnings per share (EPS) information is released by the companies. The EPS data has 3 key columns surprise, reported_EPS, and consensus_EPS:
End of explanation
%%with_globals print
%%bigquery --project {PROJECT}
CREATE OR REPLACE TABLE stock_market.eps_percent_change_sp500
AS
SELECT a.*, b.consensus_EPS, b.reported_EPS, b.surprise
FROM stock_market.percent_change_sp500 a
INNER JOIN `stock_src.eps` b
ON a.Date = b.date
AND a.symbol = b.symbol
%%with_globals print
%%bigquery --project {PROJECT}
SELECT *
FROM stock_market.eps_percent_change_sp500
LIMIT 20
%%with_globals print
%%bigquery --project {PROJECT}
SELECT direction, COUNT(*) as cnt
FROM stock_market.eps_percent_change_sp500
GROUP BY direction
Explanation: The surprise column indicates the difference between the expected (consensus expected eps by analysts) and the reported eps. We can join this table with our derived table to focus our analysis during earnings periods:
End of explanation
def get_aggregate_stats(field, round_digit=2):
    Run SELECT ... GROUP BY field, rounding to nearest digit.
    df = bq.query('''
    SELECT {field}, COUNT(*) as cnt
    FROM
    (SELECT ROUND({field}, {round_digit}) AS {field}
    FROM stock_market.eps_percent_change_sp500) rounded_field
    GROUP BY {field}
    ORDER BY {field}'''.format(field=field,
                             round_digit=round_digit,
                             PROJECT=PROJECT)).to_dataframe()
    return df.dropna()
field = 'close_AVG_prior_260_days'
CLIP_MIN, CLIP_MAX = 0.1, 4.
df = get_aggregate_stats(field)
values = df[field].clip(CLIP_MIN, CLIP_MAX)
counts = 100 * df['cnt'] / df['cnt'].sum()  # Percentage.
ax = values.hist(weights=counts, bins=30, figsize=(10, 5))
ax.set(xlabel=field, ylabel="%");
Explanation: Feature exploration
Now that we have created our recent movements of the company’s stock price, let's visualize our features. This will help us understand the data better and possibly spot errors we may have made during our calculations.
As a reminder, we calculated the scaled prices 1 week, 1 month, and 1 year before the date that we are predicting at. 
Let's write a re-usable function for aggregating our features.
Learning objective 2
End of explanation
field = 'normalized_change'
# TODO
Explanation: TODO Use the get_aggregate_stats from above to visualize the normalized_change column.
End of explanation
VALID_GROUPBY_KEYS = ('DAYOFWEEK', 'DAY', 'DAYOFYEAR',
                   'WEEK', 'MONTH', 'QUARTER', 'YEAR')
DOW_MAPPING = {1: 'Sun', 2: 'Mon', 3: 'Tues', 4: 'Wed',
                5: 'Thur', 6: 'Fri', 7: 'Sun'}
def groupby_datetime(groupby_key, field):
    if groupby_key not in VALID_GROUPBY_KEYS:
        raise Exception('Please use a valid groupby_key.')
    sql = '''
    SELECT {groupby_key}, AVG({field}) as avg_{field}
    FROM
    (SELECT {field},
     EXTRACT({groupby_key} FROM date) AS {groupby_key}
    FROM stock_market.eps_percent_change_sp500) foo
    GROUP BY {groupby_key}
    ORDER BY {groupby_key} DESC'''.format(groupby_key=groupby_key,
                                      field=field,
                                      PROJECT=PROJECT)
    print(sql)
    df = bq.query(sql).to_dataframe()
    if groupby_key == 'DAYOFWEEK':
        df.DAYOFWEEK = df.DAYOFWEEK.map(DOW_MAPPING)
    return df.set_index(groupby_key).dropna()
field = 'normalized_change'
df = groupby_datetime('DAYOFWEEK', field)
ax = df.plot(kind='barh', color='orange', alpha=0.7)
ax.grid(which='major', axis='y', linewidth=0)
field = 'close'
df = groupby_datetime('DAYOFWEEK', field)
ax = df.plot(kind='barh', color='orange', alpha=0.7)
ax.grid(which='major', axis='y', linewidth=0)
field = 'normalized_change'
df = groupby_datetime('MONTH', field)
ax = df.plot(kind='barh', color='blue', alpha=0.7)
ax.grid(which='major', axis='y', linewidth=0)
field = 'normalized_change'
df = groupby_datetime('QUARTER', field)
ax = df.plot(kind='barh', color='green', alpha=0.7)
ax.grid(which='major', axis='y', linewidth=0)
field = 'close'
df = groupby_datetime('YEAR', field)
ax = df.plot(kind='line', color='purple', alpha=0.7)
ax.grid(which='major', axis='y', linewidth=0)
field = 'normalized_change'
df = groupby_datetime('YEAR', field)
ax = df.plot(kind='line', color='purple', alpha=0.7)
ax.grid(which='major', axis='y', linewidth=0)
Explanation: Let's look at results by day-of-week, month, etc.
End of explanation
<END_TASK>