<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Copyright 2020 Google Inc.
Licensed under the Apache License, Version 2.0 (the "License").
<!--
    Licensed to the Apache Software Foundation (ASF) under one
    or more contributor license agreements.  See the NOTICE file
    distributed with this work for additional information
    regarding copyright ownership.  The ASF licenses this file
    to you under the Apache License, Version 2.0 (the
    "License"); you may not use this file except in compliance
    with the License.  You may obtain a copy of the License at
      http
Step1:  Using the new Apache Beam time series java framework in Java to compute metrics at scale in GCP Dataflow
In this case we call a gradle command to execute the Java pipeline to read and compute the metrics in scope at scale in Dataflow, PLEASE
Step4:  --resampleSec and --windowSec parameters indicate the sample period in seconds, e.g., 5, and the rolling window period to use to compute the metric in seconds, e.g., 60.
Reading results from BigQuery and loading in Pandas Dataframes
With the above command we submitted a Beam Java pipeline to Dataflow to calculate all the metrics currently implemented, including the ones in scope (Moving Average and Standard Deviation), in the next steps we would query BigQuery and load the results in a Pandas Dataframe for comparison.
We first need to be able to authenticate in GCP to read the data from BigQuery (Please ensure you have Google Cloud SDK installed and configured with the correct project)
Step5:  Using the Pandas Dataframe and Numpy APIs to compute the metrics
In this case we will leverage the data processing Python libraries directly, we first read the original dataset.
Step6:  In order to calculate the metrics in Pandas using the same constraints we need to
Step7:  Once we have the moving average and standard deviation calculated in both frameworks we can do
Step8:  We can also sample the joined dataframes to compare the two
Step9:  We can see at a first glance that the Standard Deviation metric is almost identical between the two frameworks, the moving average deviates slightly more.
This is caused by the round-off error between Java BigDecimals vs Python Float numerical data types. In the case of moving average the difference is higher as there are values at both sides of the decimal point, for more details The Perils of Floating Point.
We can also use the Bokeh library to enable interactive graphs, so we can interact with the results to find areas where the delta is bigger.
Step10:  
Step12:  Filling the gaps
As wrap up, let's check how the gap forward filling feature is applied in both frameworks before the metrics are calculated
Step13:  We can see how the unsampled data points are less frequent, so the resampled data points are forward filled correctly, in this case using the LAST measure from the samples.
Let's repeat the exercise with the Pandas resample this time
<ASSISTANT_TASK:>
Python Code:
!conda install -c conda-forge google-cloud-bigquery google-cloud-bigquery-storage pyarrow pandas numpy matplotlib bokeh -y
Explanation: Copyright 2020 Google Inc.
Licensed under the Apache License, Version 2.0 (the "License").
<!--
    Licensed to the Apache Software Foundation (ASF) under one
    or more contributor license agreements.  See the NOTICE file
    distributed with this work for additional information
    regarding copyright ownership.  The ASF licenses this file
    to you under the Apache License, Version 2.0 (the
    "License"); you may not use this file except in compliance
    with the License.  You may obtain a copy of the License at
      http://www.apache.org/licenses/LICENSE-2.0
    Unless required by applicable law or agreed to in writing,
    software distributed under the License is distributed on an
    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    KIND, either express or implied.  See the License for the
    specific language governing permissions and limitations
    under the License.
-->
Introduction
This notebook compares the following time series metrics calculated using two different frameworks:
- 60 seconds Moving Average
- Standard Deviation over 60 second rolling window
Using one day worth of forex tick data, for the EUR/USD currency pair, extracted from a public API and resampled to 5 seconds periods.
The two frameworks compared are:
1. The new Apache Beam time series framework, which includes
the metrics library available out of the box in Java SDK
2. The Pandas and Numpy libraries available in Python SDK, as the defacto tools for data manipulation of time series data in vectorised form
Prerequisites
GCP project with the BigQuery API enabled to store results, and Dataflow API enabled to compute time series metrics at scale
Google Cloud SDK installed and configured 
Setup
Assuming you are reading this from Github:
- Clone this repository locally with the correct tag:
git clone -b tsflow-0.3.2-sample https://github.com/GoogleCloudPlatform/dataflow-sample-applications
Install Java 1.8, Gradle and Python 3.x
Create Python Virtual or Conda Environment, e.g. for Conda: conda create --name timeseriesanalysis-is-cool python=3.7 jupyterlab
Run Jupyter notebook or lab to open this notebook
Execute the below cell to install the required dependencies
End of explanation
!gradle -p ../../timeseries-java-applications forex_example --args='--resampleSec=5 --windowSec=60 --runner=DataflowRunner --workerMachineType=n1-standard-4 --project=<GCPPROJECT> --region=<REGION> --bigQueryTableForTSAccumOutputLocation=<GCPPROJECT>:<DATASET>.<TABLE> --gcpTempLocation=<GCPTMPLOCATION> --tempLocation=<TMPLOCATION> --inputPath=<INPUTDATASETLOCATION>'
Explanation: Using the new Apache Beam time series java framework in Java to compute metrics at scale in GCP Dataflow
In this case we call a gradle command to execute the Java pipeline to read and compute the metrics in scope at scale in Dataflow, PLEASE: Replace &lt;GCPPROJECT&gt;, &lt;REGION&gt;, &lt;DATASET&gt;, &lt;TABLE&gt; , &lt;GCPTMPLOCATION&gt; and &lt;TMPLOCATION&gt; parameters accordingly:
End of explanation
import google.auth
from google.cloud import bigquery
from google.cloud import bigquery_storage
import pandas as pd
# Explicitly create a credentials object. This allows you to use the same
# credentials for both the BigQuery and BigQuery Storage clients, avoiding
# unnecessary API calls to fetch duplicate authentication tokens.
credentials, your_project_id = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
# Make clients.
bqclient = bigquery.Client(credentials=credentials, project=your_project_id,)
bqstorageclient = bigquery_storage.BigQueryReadClient(credentials=credentials)
# Download results from BigQuery after the Beam pipeline finishes processing the time series data
query_string_60m = 
WITH BID_MA60_T AS (
    SELECT lower_window_boundary, upper_window_boundary, 
    (SELECT dbl_data from UNNEST(data) WHERE metric = 'SIMPLE_MOVING_AVERAGE') AS BID_MA60_BQ
    FROM `<GCPPROJECT>.<DATASET>.<TABLE>`
    WHERE timeseries_minor_key = 'BID'
)
SELECT * from BID_MA60_T
bid_ma60_df = (
    bqclient.query(query_string_60m)
    .result()
    .to_dataframe(bqstorage_client=bqstorageclient)
).sort_values(by=['upper_window_boundary'])
bid_ma60_df.index = pd.to_datetime(bid_ma60_df['upper_window_boundary'])
bid_ma60_df.drop(columns=['upper_window_boundary', 'lower_window_boundary'],inplace=True)
query_string_stddev = 
WITH BID_STDDEV_T AS (
    SELECT lower_window_boundary, upper_window_boundary, 
    (SELECT dbl_data from UNNEST(data) WHERE metric = 'STANDARD_DEVIATION') AS BID_STDDEV_BQ
    FROM `<GCPPROJECT>.<DATASET>.<TABLE>`
    WHERE timeseries_minor_key = 'BID'
)
SELECT * from BID_STDDEV_T
bid_std_dev_df = (
    bqclient.query(query_string_stddev)
    .result()
    .to_dataframe(bqstorage_client=bqstorageclient)
).sort_values(by=['upper_window_boundary'])
bid_std_dev_df.index = pd.to_datetime(bid_std_dev_df['upper_window_boundary'])
bid_std_dev_df.drop(columns=['upper_window_boundary', 'lower_window_boundary'],inplace=True)
# Showing Moving Average sample
bid_ma60_df
# Showing Moving Average sample
bid_std_dev_df
Explanation: --resampleSec and --windowSec parameters indicate the sample period in seconds, e.g., 5, and the rolling window period to use to compute the metric in seconds, e.g., 60.
Reading results from BigQuery and loading in Pandas Dataframes
With the above command we submitted a Beam Java pipeline to Dataflow to calculate all the metrics currently implemented, including the ones in scope (Moving Average and Standard Deviation), in the next steps we would query BigQuery and load the results in a Pandas Dataframe for comparison.
We first need to be able to authenticate in GCP to read the data from BigQuery (Please ensure you have Google Cloud SDK installed and configured with the correct project)
End of explanation
import pandas as pd
eurusd = pd.read_csv("../../timeseries-java-applications/Examples/src/main/resources/EURUSD-2020-05-11_2020-05-11.csv",
                     index_col=1, names=["Pair", "Timestamp", "Ask", "Bid", "Ask Volume", "Bid Volume"], header=None)
eurusd.index = pd.to_datetime(eurusd.index)
Explanation: Using the Pandas Dataframe and Numpy APIs to compute the metrics
In this case we will leverage the data processing Python libraries directly, we first read the original dataset.
End of explanation
eurusd_bid_ma60 = eurusd.rename(columns={"Bid": "Bid MA60 Pandas"})\
                        .resample("5S")\
                        .fillna(method='ffill')['Bid MA60 Pandas']\
                        .rolling(window=12, min_periods=1)\
                        .apply(lambda x: np.mean(x))
eurusd_bid_stddev = eurusd.rename(columns={"Bid": "Bid StdDev Pandas"})\
                          .resample("5S")\
                          .fillna(method='ffill')['Bid StdDev Pandas']\
                          .rolling(window=12, min_periods=1)\
                          .apply(lambda x: np.std(x))
Explanation: In order to calculate the metrics in Pandas using the same constraints we need to:
1. resample() the time series datapoints to every 5 seconds
2. Do forward filling with the fillna(method='ffill') method to fill the gaps accordingly
3. Apply rolling() window
4. Aggregate by applying the Numpy mean() and std() methods respectively
End of explanation
bid_ma60_df_utc = bid_ma60_df.tz_convert(None)
bid_ma60_df_utc_join = bid_ma60_df_utc.join(eurusd_bid_ma60)
bid_ma60_df_utc_join['Delta'] = bid_ma60_df_utc_join["Bid MA60 Pandas"] - bid_ma60_df_utc_join["BID_MA60_BQ"]
bid_ma60_df_utc_join[['BID_MA60_BQ','Bid MA60 Pandas']].plot(figsize=(15,6))
bid_std_dev_df_utc = bid_std_dev_df.tz_convert(None)
bid_std_dev_df_utc_join = bid_std_dev_df_utc.join(eurusd_bid_stddev)
bid_std_dev_df_utc_join['Delta'] = bid_std_dev_df_utc_join["Bid StdDev Pandas"] - bid_std_dev_df_utc_join["BID_STDDEV_BQ"]
bid_std_dev_df_utc_join[['BID_STDDEV_BQ','Bid StdDev Pandas']].plot(figsize=(15,6))
Explanation: Once we have the moving average and standard deviation calculated in both frameworks we can do:
1. Convert the dataframes generated from BigQuery to UTC
2. Join the results from the two frameworks for comparisson
3. Create a Delta metric to visualise how close the values are
4. Quick visualization to compare the two metrics
Comparing results
End of explanation
bid_ma60_df_utc_join
bid_std_dev_df_utc_join
Explanation: We can also sample the joined dataframes to compare the two
End of explanation
import numpy as np
from bokeh.io import show, output_notebook
from bokeh.layouts import column
from bokeh.models import RangeTool
from bokeh.plotting import figure
dates = np.array(bid_ma60_df_utc_join.index, dtype=np.datetime64)
p1 = figure(plot_height=300, plot_width=800, tools="xpan", toolbar_location=None,
           x_axis_type="datetime", x_axis_location="above",
           background_fill_color="#efefef", x_range=(dates[1500], dates[2500]))
p1.line(dates, bid_ma60_df_utc_join['BID_MA60_BQ'], color='#A6CEE3', legend_label='Bid MA60 BQ')
p1.line(dates, bid_ma60_df_utc_join['Bid MA60 Pandas'], color='#FB9A99', legend_label='Bid MA60 Pandas')
p1.yaxis.axis_label = 'Price MA60'
p2 = figure(plot_height=300, plot_width=800, tools="xpan", toolbar_location=None,
           x_axis_type="datetime", x_axis_location="above",
           background_fill_color="#efefef", x_range=p1.x_range)
p2.line(dates, bid_ma60_df_utc_join['Delta'], color='#A6CEE3', legend_label='Delta')
p2.yaxis.axis_label = 'Delta'
select = figure(title="Drag the middle and edges of the selection box to change the range above",
                plot_height=130, plot_width=800, y_range=p1.y_range,
                x_axis_type="datetime", y_axis_type=None,
                tools="", toolbar_location=None, background_fill_color="#efefef")
range_tool = RangeTool(x_range=p1.x_range)
range_tool.overlay.fill_color = "navy"
range_tool.overlay.fill_alpha = 0.2
select.line(dates, bid_ma60_df_utc_join['BID_MA60_BQ'], color='#A6CEE3')
select.line(dates, bid_ma60_df_utc_join['Bid MA60 Pandas'], color='#FB9A99')
select.ygrid.grid_line_color = None
select.add_tools(range_tool)
select.toolbar.active_multi = range_tool
output_notebook()
show(column(p1, p2, select))
Explanation: We can see at a first glance that the Standard Deviation metric is almost identical between the two frameworks, the moving average deviates slightly more.
This is caused by the round-off error between Java BigDecimals vs Python Float numerical data types. In the case of moving average the difference is higher as there are values at both sides of the decimal point, for more details The Perils of Floating Point.
We can also use the Bokeh library to enable interactive graphs, so we can interact with the results to find areas where the delta is bigger.
End of explanation
import numpy as np
from bokeh.io import show, output_notebook
from bokeh.layouts import column
from bokeh.models import RangeTool
from bokeh.plotting import figure
dates = np.array(bid_std_dev_df_utc_join.index, dtype=np.datetime64)
p1 = figure(plot_height=300, plot_width=800, tools="xpan", toolbar_location=None,
           x_axis_type="datetime", x_axis_location="above",
           background_fill_color="#efefef", x_range=(dates[1500], dates[2500]))
p1.line(dates, bid_std_dev_df_utc_join['BID_STDDEV_BQ'], color='#A6CEE3', legend_label='Bid STDDEV BQ')
p1.line(dates, bid_std_dev_df_utc_join['Bid StdDev Pandas'], color='#FB9A99', legend_label='Bid StdDev Pandas')
p1.yaxis.axis_label = 'Standard Deviation'
p2 = figure(plot_height=300, plot_width=800, tools="xpan", toolbar_location=None,
           x_axis_type="datetime", x_axis_location="above",
           background_fill_color="#efefef", x_range=p1.x_range)
p2.line(dates, bid_std_dev_df_utc_join['Delta'], color='#A6CEE3', legend_label='Delta')
p2.yaxis.axis_label = 'Delta'
select = figure(title="Drag the middle and edges of the selection box to change the range above",
                plot_height=130, plot_width=800, y_range=p1.y_range,
                x_axis_type="datetime", y_axis_type=None,
                tools="", toolbar_location=None, background_fill_color="#efefef")
range_tool = RangeTool(x_range=p1.x_range)
range_tool.overlay.fill_color = "navy"
range_tool.overlay.fill_alpha = 0.2
select.line(dates, bid_std_dev_df_utc_join['BID_STDDEV_BQ'], color='#A6CEE3')
select.line(dates, bid_std_dev_df_utc_join['Bid StdDev Pandas'], color='#FB9A99')
select.ygrid.grid_line_color = None
select.add_tools(range_tool)
select.toolbar.active_multi = range_tool
output_notebook()
show(column(p1, p2, select))
Explanation: 
End of explanation
query_string_last = 
WITH LAST AS (
    SELECT lower_window_boundary, upper_window_boundary, 
    (SELECT dbl_data from UNNEST(data) WHERE metric = 'LAST') AS LAST
    FROM `<GCPPROJECT>.<DATASET>.<TABLE>`
    WHERE timeseries_minor_key = 'BID'
)
SELECT * from LAST
last_bq = (
    bqclient.query(query_string_last)
    .result()
    .to_dataframe(bqstorage_client=bqstorageclient)
).sort_values(by=['upper_window_boundary'])
last_bq.index = pd.to_datetime(last_bq['upper_window_boundary'])
last_bq.drop(columns=['upper_window_boundary', 'lower_window_boundary'],inplace=True)
last_bq
import numpy as np
from bokeh.io import show, output_notebook
from bokeh.layouts import column
from bokeh.models import RangeTool
from bokeh.plotting import figure
dates_unsampled = np.array(eurusd.index, dtype=np.datetime64)
dates_resampled = np.array(last_bq.index, dtype=np.datetime64)
# We position the graph towards the end of day to show a quiet period when forward filling is necessary
p1 = figure(plot_height=300, plot_width=800, tools="xpan", toolbar_location=None,
           x_axis_type="datetime", x_axis_location="above",
           background_fill_color="#efefef", x_range=(dates_unsampled[dates_unsampled.size-100], dates_unsampled[dates_unsampled.size-1]))
p1.circle(dates_unsampled, eurusd['Bid'], color='#A6CEE3', legend_label='Bid unsampled')
p1.circle(dates_resampled, last_bq['LAST'], color='#FB9A99', legend_label='Bid downsampled')
p1.yaxis.axis_label = 'Downsampling'
select = figure(title="Drag the middle and edges of the selection box to change the range above",
                plot_height=130, plot_width=800, y_range=p1.y_range,
                x_axis_type="datetime", y_axis_type=None,
                tools="", toolbar_location=None, background_fill_color="#efefef")
range_tool = RangeTool(x_range=p1.x_range)
range_tool.overlay.fill_color = "navy"
range_tool.overlay.fill_alpha = 0.2
select.line(dates_unsampled, eurusd['Bid'], color='#A6CEE3')
select.ygrid.grid_line_color = None
select.add_tools(range_tool)
select.toolbar.active_multi = range_tool
output_notebook()
show(column(p1, select))
# from bokeh.io import export_png
# export_png(column(p1, select), filename="img/FILLINGBQ.png")
Explanation: Filling the gaps
As wrap up, let's check how the gap forward filling feature is applied in both frameworks before the metrics are calculated:
We start first with the results from the Time Series Beam framework stored in BigQuery already
End of explanation
last_pandas = eurusd.resample("5S").fillna(method='ffill')['Bid']
import numpy as np
from bokeh.io import show, output_notebook
from bokeh.layouts import column
from bokeh.models import RangeTool
from bokeh.plotting import figure
dates_unsampled = np.array(eurusd.index, dtype=np.datetime64)
dates_resampled = np.array(last_pandas.index, dtype=np.datetime64)
# We position the graph towards the end of day to show a quiet period when forward filling is necessary
p1 = figure(plot_height=300, plot_width=800, tools="xpan", toolbar_location=None,
           x_axis_type="datetime", x_axis_location="above",
           background_fill_color="#efefef", x_range=(dates_unsampled[dates_unsampled.size-100], dates_unsampled[dates_unsampled.size-1]))
p1.circle(dates_unsampled, eurusd['Bid'], color='#A6CEE3', legend_label='Bid unsampled')
p1.circle(dates_resampled, last_pandas, color='#FB9A99', legend_label='Bid downsampled')
p1.yaxis.axis_label = 'Downsampling'
select = figure(title="Drag the middle and edges of the selection box to change the range above",
                plot_height=130, plot_width=800, y_range=p1.y_range,
                x_axis_type="datetime", y_axis_type=None,
                tools="", toolbar_location=None, background_fill_color="#efefef")
range_tool = RangeTool(x_range=p1.x_range)
range_tool.overlay.fill_color = "navy"
range_tool.overlay.fill_alpha = 0.2
select.line(dates_unsampled, eurusd['Bid'], color='#A6CEE3')
select.ygrid.grid_line_color = None
select.add_tools(range_tool)
select.toolbar.active_multi = range_tool
output_notebook()
show(column(p1, select))
# from bokeh.io import export_png
# export_png(column(p1, select), filename="img/FILLINGPANDAS.png")
Explanation: We can see how the unsampled data points are less frequent, so the resampled data points are forward filled correctly, in this case using the LAST measure from the samples.
Let's repeat the exercise with the Pandas resample this time
End of explanation
<END_TASK>