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
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

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}
SELECT count(*) as cnt
FROM `asl-ml-immersion.stock_src.price_history`

%%with_globals
%%bigquery --project {PROJECT}
SELECT count(*) as cnt
FROM `asl-ml-immersion.stock_src.snp500`

"""
Explanation: Create time-series features and determine label based on market movement
Summary of base tables
End of explanation
"""


%%with_globals print
%%bigquery --project {PROJECT} df
CREATE OR REPLACE TABLE `{PROJECT}.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 `asl-ml-immersion.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
AVG(close) AS avg_close,
AVG(tomorrow_close) AS avg_tomo_close,
AVG(tomorrow_close) - AVG(close) AS avg_change,
COUNT(*) cnt
FROM
stock_market.price_history_delta

"""
Explanation: The stock market is going up on average.
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 symbol
                      ORDER BY Date
                      ROWS BETWEEN {n_days} PRECEDING AND 1 PRECEDING))/{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
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 `{PROJECT}.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,
       (scaled_change - s_p_scaled_change) 
           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
End of explanation
"""


%%with_globals print
%%bigquery --project {PROJECT} df
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

%%with_globals print
%%bigquery --project {PROJECT} df
CREATE OR REPLACE TABLE stock_market.percent_change_sp500
AS
SELECT *,
    CASE WHEN normalized_change < {down_thresh} THEN 'DOWN'
       WHEN normalized_change > {up_thresh} THEN 'UP'
       ELSE 'STAY'
    END AS direction
FROM stock_market.price_features_norm_per_change features
INNER JOIN `asl-ml-immersion.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 --project {PROJECT} df
SELECT *
FROM stock_market.percent_change_sp500
LIMIT 20

df.columns

"""
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 --project {PROJECT}
SELECT *
FROM `asl-ml-immersion.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 `asl-ml-immersion.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 5

%%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="%");

field = 'normalized_change'
CLIP_MIN, CLIP_MAX = -0.1, 0.1
df = get_aggregate_stats(field, round_digit=3)
values = df[field].clip(CLIP_MIN, CLIP_MAX)
counts = 100 * df['cnt'] / df['cnt'].sum()  # Percentage.
ax = values.hist(weights=counts, bins=50, 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
"""


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
"""


