Automate Financial Asset Tracking
Automate tracking financial assets with Python, Docker, AWS & CRON.
Being a programmer is about spending a week to automate something that takes 5 seconds to do manually — and this project is testimony!
I work fully remotely and that means I do a LOT of traveling. As a consequence,
Foreign Exchange Rates Matter 💵 💶 💷
I’m not trying to predict Forex assets, but monitoring exchange rates (appears) to help me get slightly better prices (although as a statistician I understand my best bet is to buy at random consistently).
Whatever your motives, monitoring financial assets is tedious so why not automate it?
How to Automate Financial Asset Tracking with Python, Docker, AWS & CRON
Project Architecture
The project requires 4 steps:
- Fetch & analyse up-to-date forex data (Python).
- Store and configure the results (AWS S3).
- Packaging the code and dependencies (Docker).
- Automate the pipeline (CRON & Shell Scripts).
Fetch & Analyse Forex Data (Python)
I’ve written a Python Library to fetch & model financial data. The Python package bundles dependencies, API queries and statistical functions.
ForexFlaggr
The package:
- Fetches financial data with the
yfinance(Yahoo Finance) API. - Generates plots.
- Fits statistical models (GAMs & LOESS) to the data.
- Generates an output HTML file to share the results.
The package is git installable (hosted on PyPi) & the source code is available on GitHub.
Intentionally very high level, the purpose of the package is to abstract functionality making it easy to run in many different environments. Usage is self-explanatory:
I. Fetching & structuring data
Taking a look at the source code, the package is designed to be simple but extendable — making it easy to add modules in future. The src contains a ForexFlaggr class that fetches the data and constructs a data frame containing raw data and moving averages.
"""
------------------------------------------------------------------------------------------------
forex_flagger.py
ForexFlagger class.
- fetch data from yfinance
- construct dataframe
- compute (weighted) moving averages
- generate plotly graph
: 18.11.23
: Zach Wolpe
: zach.wolpe@mlxgo.com
------------------------------------------------------------------------------------------------
"""
from .dependencies import (dt, yf, go, np)
class ForexFlaggr:
def __init__(self) -> None:
self.data = None
@staticmethod
def get_price(ff):
"""
Return the most recent price data.
Returns:
datetime, timezone, close, open, high, low
"""
x = ff.df_all.iloc[-1, :]
return str(x.name), str(x.name.tz), x['Close'], x['Open'], x['High'], x['Low']
@staticmethod
def moving_average(signal, period=20):
return signal.rolling(window=period).mean()
@staticmethod
def weighted_moving_average(signal, period=14):
weights = np.linspace(0,1,period)
sum_weights = np.sum(weights)
return signal\
.rolling(window=period)\
.apply(lambda x: np.sum(weights*x)/sum_weights)
def filter_by_date(self, start_date, end_date):
return self.data[start_date:end_date]
def fetch_data(self, stock="USDZAR=X", sample_interval='1h', n_days=500):
self.title = stock
start_date = dt.datetime.today() - dt.timedelta(n_days)
end_date = dt.datetime.today()
self.data = yf.download(stock, start_date, end_date, interval=sample_interval)
return self
def join_df(self, ma, wma):
self.ma = ma
self.wma = wma
self.wma.name = 'wma'
self.ma.name = 'ma'
self.df_all = self.data.join(ma).join(wma)
def plot_signal(self, MA_periods=None):
if self.data is None:
return self
if MA_periods is None:
MA_periods = 12*15 # three weeks (exclude weekends)
ma = ForexFlaggr.moving_average(self.data.Close, MA_periods)
wma = ForexFlaggr.weighted_moving_average(self.data.Close, MA_periods)
fig = go.Figure()
fig.add_trace(go.Line(x=self.data.index, y=self.data.Close, name=self.title))
fig.add_trace(go.Line(x=self.data.index, y=ma, name='Moving avg.'))
fig.add_trace(go.Line(x=self.data.index, y=wma, name='Weighted ma.'))
fig.update_layout(template='none', title=self.title)
self.join_df(ma, wma)
self.fig = fig
return self
@staticmethod
def consol_log():
msg = """
______________________________________________________________________________________________________________________________________________
______________________________________________________________________________________________________________________________________________
/$$ /$$$$$$$$ /$$$$$$$$ /$$
/$$$$$$ | $$_____/ | $$_____/| $$
/$$__ $$ | $$ /$$$$$$ /$$$$$$ /$$$$$$ /$$ /$$ | $$ | $$ /$$$$$$ /$$$$$$ /$$$$$$ /$$$$$$
| $$ \__/ | $$$$$ /$$__ $$ /$$__ $$ /$$__ $$| $$ /$$/ | $$$$$ | $$ |____ $$ /$$__ $$ /$$__ $$ /$$__ $$
| $$$$$$ | $$__/| $$ \ $$| $$ \__/| $$$$$$$$ \ $$$$/ | $$__/ | $$ /$$$$$$$| $$ \ $$| $$ \ $$| $$ \__/
\____ $$ | $$ | $$ | $$| $$ | $$_____/ $$ $$ | $$ | $$ /$$__ $$| $$ | $$| $$ | $$| $$
/$$ \ $$ | $$ | $$$$$$/| $$ | $$$$$$$ /$$/\ $$ | $$ | $$| $$$$$$$| $$$$$$$| $$$$$$$| $$
| $$$$$$/ |__/ \______/ |__/ \_______/|__/ \__/ |__/ |__/ \_______/ \____ $$ \____ $$|__/
\_ $$_/ /$$ \ $$ /$$ \ $$
\__/ | $$$$$$/| $$$$$$/
\______/ \______/
______________________________________________________________________________________________________________________________________________
______________________________________________________________________________________________________________________________________________
"""
return msgII. Statistical models
The data can be used to fit statistical models. We want to ensure that new models can be implemented easily without breaking the pipeline or requiring substantial changes to downstream code. We also want to ensure that retraining can be done regularly, and weights can be cached. This is achieved by implementing a base class from which contrete models are built.
A model interface defines the expected behaviour of a concrete model instance.
"""
------------------------------------------------------------------------------------------------
stochastic_model_interface.py
A base class that provides an interface for stochastic models.
: 18.11.23
: Zach Wolpe
: zach.wolpe@mlxgo.com
------------------------------------------------------------------------------------------------
"""
from .dependencies import (dt)
class Stochastic_Model_Interface:
"""Interface for stochastic models"""
def __init__(self, X, y, model, model_name):
self.X = X
self.y = y
self.model = model
self.model_name = model_name
@staticmethod
def internal_logger(msg):
print(f'[{dt.datetime.now()}] {msg}')
@staticmethod
def internal_tracing(func):
"""Decorator for internal tracing (timing & logging)."""
def wrapper(*args, **kwargs):
# Stochastic_Model_Interface.internal_logger(f'Running {func.__name__}.')
start = dt.datetime.now()
result = func(*args, **kwargs)
end = dt.datetime.now()
s = ''
l = max(0,15-len(func.__name__))
Stochastic_Model_Interface.internal_logger(f'<{func.__name__}> finished in {s:<{l}}{end-start}.')
return result
return wrapper
def check_Xy(self,X,y):
if X is None:
X = self.X
if y is None:
y = self.y
self.X, self.y = X,y
return X,y
def fit(self, X, y):
raise NotImplementedError
def transform(self):
raise NotImplementedError
def fit_transform(self, X, y):
raise NotImplementedError
def predict(self, x_pred):
raise NotImplementedError
def extrapolate(self, x_pred, n_steps=500):
raise NotImplementedError
def build(self, X=None, y=None, n_steps=500):
"""Implement fit & extrapolate."""
raise NotImplementedError
def plot_prediction(self, X=None, y=None, n_steps=500):
"""Implement plot."""
raise NotImplementedErrorA GAM & LOESS are fit using the model interface as a framework. Here is the GAM concrete implementation, see the source code for the LOESS implementation.
"""
------------------------------------------------------------------------------------------------
gam_model.py
A GAM model class, building of the stochastic model interface.
: 18.11.23
: Zach Wolpe
: zach.wolpe@mlxgo.com
------------------------------------------------------------------------------------------------
"""
from .dependencies import (LinearGAM, go, np)
from .stochastic_model_interface import Stochastic_Model_Interface
class GAM_Model(Stochastic_Model_Interface):
"""
Generalized Additive Model
documentation: https://pygam.readthedocs.io/en/latest/api/linearGAM.html
"""
n_samples = 500
def __init__(self, X, y) -> None:
super().__init__(X, y, LinearGAM, 'GAM')
self._n_splines = 25
@property
def n_splines(self):
return self._n_splines
@n_splines.setter
def n_splines(self, n):
self._n_splines = n
self.model = LinearGAM(n_splines=n).gridsearch(self.X, self.y)
@Stochastic_Model_Interface.internal_tracing
def fit(self, X=None, y=None):
X,y = self.check_Xy(X,y)
self.model = LinearGAM(n_splines=self.n_splines).gridsearch(X, y)
self.XX = self.model.generate_X_grid(term=0, n=GAM_Model.n_samples)
return self
@Stochastic_Model_Interface.internal_tracing
def transform(self):
self.yhat, self.CI = self.predict(self.XX)
self.CI_lower, self.CI_upper = self.CI[:,0], self.CI[:,1]
return self
def fit_transform(self, X=None, y=None):
self.fit(X,y)
self.transform()
return self
def predict(self, x_pred, CI_width=.95):
"""returns the mean prediction & confidence intervals"""
return self.model.predict(x_pred), self.model.prediction_intervals(x_pred, width=CI_width)
@Stochastic_Model_Interface.internal_tracing
def extrapolate(self, CI_width=0.95, n_steps=500):
"""Extrapolate forward n_steps."""
# self.internal_logger('Extrapolating {n_steps} forward...')
# m = self.X.min()
M = self.X.max()
self.Xforward = np.linspace(M, M + n_steps, n_steps)
self.yhatforward, self.CIforward = self.predict(self.Xforward, CI_width)
self.CIf_lower, self.CIf_upper = self.CIforward[:,0], self.CIforward[:,1]
return self
@Stochastic_Model_Interface.internal_tracing
def build(self, X=None, y=None, n_steps=500):
self.check_Xy(X,y)
self.fit(X=X, y=y)\
.transform()\
.extrapolate(n_steps=n_steps)\
.plot_prediction()
return self
@Stochastic_Model_Interface.internal_tracing
def plot_prediction(self, c1='orange', c2='darkblue', c3='lightblue', *args, **kwargs):
"""Plot the prediction & confidence intervals."""
fig = go.Figure()
fig.add_trace(go.Scatter(x=self.X.flatten(), y=self.y.flatten(), name='raw data', line=dict(color=c1)))
fig.add_trace(go.Scatter(x=self.XX.flatten(), y=self.yhat, name='prediction', line=dict(color=c2)))
fig.add_trace(go.Scatter(x=self.XX.flatten(), y=self.CI_lower, name='confidence', line=dict(color=c3)))
fig.add_trace(go.Scatter(x=self.XX.flatten(), y=self.CI_upper, name=self.model_name, line=dict(color=c3), showlegend=False))
fig.update_layout(template='none', title=f'{self.model_name} ~ (USD/ZAR)', yaxis_title='Exchange Rate', xaxis_title='Date')
try:
# if extrapolated
# make lines dashed
fig.add_trace(go.Scatter(x=self.Xforward, y=self.yhatforward, name='GAM', line=dict(color='black', dash='dash'), showlegend=False))
fig.add_trace(go.Scatter(x=self.Xforward, y=self.CIf_lower, name='GAM', line=dict(color='black', dash='dash'), showlegend=False))
fig.add_trace(go.Scatter(x=self.Xforward, y=self.CIf_upper, name='GAM', line=dict(color='black', dash='dash'), showlegend=False))
except Exception:
pass
self.fig = fig
return selfIII. Convert to HTML
Finally, the html_config.py module transforms all of the raw plots and statistical figures into a readable html document. This provides a clean way of interacting with the results of our analysis.
Statistical Financial Analysis: HTML Doc
I am interested in modelling USD/ZAR. My instance of ForexFlaggr has been configured to:
- Provide recent price data of USD/ZAR.
- Model USD/ZAR as a function of various financial indicators.
- Make recommendations on buying or selling USD/ZAR.
I have configured an API to serve this purpose, which is contained in the package source code. Thus I only need to run the following to produce the HTML report.
import forexflaggr as fxr
import numpy as np
if __name__ == '__main__':
print('fxr version : ', fxr.__version__)
np.int = np.int64 # workaround for deprecated np.int
fxr.Build_HTML_Config(_store_path='output')Here are samples from the output HTML document, stored in forexflaggr.html.
We have generated the entire report.
We now want to automatically schedule the build, host the results & send a notification when the runtime completes.
Hosting (AWS S3 Bucket)
The easiest way to host an HTML document is on AWS. Any S3 bucket can be configured to host static websites. Once a bucket is configured and you have logged in to the AWS CLI, a shell script can be used to put the files to S3.
Our analysis is contained in the forexflaggr.html document. The additional files are supporting documents needed to generate the HTML report.
echo 'Writing to S3 MY_BUCKET...'
MY_DIR=output
MY_BUCKET=<$S3-BUCKET-PATH>
MY_POLICY=public-read
aws s3 cp ${MY_DIR}/3d_hyperplane_scatter.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/3d_hyperplane_scatter.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/3d_scatter_prediction.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/3d_scatter.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/forexflaggr.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/gam.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/gam.png ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/loess.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/pie_chart_recommendation.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/price_data.json ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/sp500.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/usdzar.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/sp500.html ${MY_BUCKET} --acl ${MY_POLICY}
aws s3 cp ${MY_DIR}/UStreasury.html ${MY_BUCKET} --acl ${MY_POLICY}
echo Write to S3 MY_BUCKET=$MY_BUCKET successful!Notification (Slack)
It would be nice to be notified when an upload is completed. This could be configured with AWS SNS if you plan on running the whole pipeline on AWS, but another solution is to use Slack webhooks to send a message to Slack directly.
The Slack API can be used to configure a bot. This sh script is used to send the <MESSAGE> on Slack.
echo 'Writing Slack notification...'
URL='<*************************************************>'
WEBHOOK='*************************************************'
PRICE=$(cat output/price_data.json | jq '.[2]' | cut -c 1-6)
MESSAGE="*************************************************"
curl -X POST -H 'Content-type: application/json' --data "{'text': '${MESSAGE}'}" $WEBHOOK
echo 'Slack notification successful!'Orchestration: Build Once, Run Anyway (Docker)
Our app now contains a Python package & several shell scripts and API connections that all need to be run sequentially. One minimal orchestration option would place all the code in a single directory and use an orchestrator script to manage execution.
Alternatively (used here), we can use Docker to package the application so that it is portable and can run either locally or on the cloud.
Create a directory to store the docker runtime. All of our code is placed in an ./app directory.
dockerfile.yml
The Dockerfile builds the docker image. A Debian Linux distribution is used to build a python==3.10 environment. The environment installs the AWS CLI, Python packages (including our own ForexFlaggr) and configures the file structure.
# Use Alpine Linux as base image
# FROM python:3.10.13-alpine3.18
FROM python:3
# Install AWS CLI within the container
RUN apt-get update && \
apt-get install -y awscli
# install jq
RUN apt-get update && apt-get install -y jq
# # Install Python dependencies
# RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --upgrade pip
RUN pip install kaleido
RUN pip install forexflaggr==0.0.5
RUN pip install pyyaml
RUN pip install numpy==1.26.2
# copy python script into image
RUN mkdir app
COPY ./app/* ./app
RUN mkdir app/output
# change working directory to app
WORKDIR /app
RUN echo ">> DOCKER BUILD COMPLETE. <<"Build the docker image with:
docker image build -f dockerfile.yml -t <IMAGE:TAG> <LOCATION>
# in my case:
docker build -f dockerfile.yml -t forexflaggr:0.0.1 .docker-compose.yml
docker-compose.yml is used to launch a new container from the image. We connect our local files to the files in the docker image using a mount. We also need to configure our AWS credentials here if we want to run the app locally.
The command specifies which commands to run on launch. This orchestrates our entire process, calling each stage of our deployment:
run.py: builds the ForexFlaggr module../aws-put.sh: pushed our data to S3../slack-notification.sh: sends a message to Slack.
Note: chmod +x <file> is used to grant system execution access to the <file>. and can be removed after the initial run.
version: '3'
services:
forexflaggr:
image: forexflaggr:0.0.1
environment:
AWS_ACCESS_KEY_ID: ***************************************
AWS_SECRET_ACCESS_KEY: ***************************************
AWS_DEFAULT_REGION: ***************************************
volumes:
- ./app:/app
command: >
bash -c "echo 'launching docker compose...' &&
chmod +x ./aws-put.sh &&
chmod +x ./slack-notify.sh &&
python run.py &&
./aws-put.sh &&
./slack-notify.sh &&
echo 'docker compose finished'
"Build
Voila! Now running our entire pipeline is as simple as
docker compose upThis command launches a new Docker container that:
- Run ForexFlaggr to fetch and model the Forex data.
- Pushes the results to aws/s3.
- Send a Slack notification with the link to the results.
CRON Scheduling
Although the process now requires a single line in the terminal to execute, why not automate the entire process?
CRON can be used to schedule code on UNIX-based systems (Mac/Linux). Using crontab shell commands can be scheduled. Install crontab via HomeBrew, then enter the crontab by running crontab -e in a terminal. This will launch vim.
To launch the docker compose file every day at 10 am, insert the command in your Vim editor
0 10 * * * /usr/local/bin/docker compose -f $PATH_TO_DIRECTORY/docker-compose.yml upCRON Configuration Tips:
- You will need to grant execution access to all shell/docker scripts with
chmod +=. - CRON may not be able to access your docker daemon due to environment variable conflicts. Run
which dockerto see where your docker daemon is stored (in my case,/usr/local/binand use the full path. - CRON may not have access to the file system. You may be confused to see that even after granting access to CRON (via Mac/privacy settings) CRON is unable to local files. Ensure you grant access to your terminal (in my case,
iTerm2),crontabapplication & CRON. - Access cron logs by starting a shell and
cat /var/mail/<usr>orlog show — process cron -debug | grep <usr>. - Vim: Add a command to Vim by pressing
i(insert). Write (save)wand quitqVim by running:wq!. Delete a line in vim by runningdd.
Result
Every morning at 10 am I get a Slack message from ForexFlaggr with the latest USD/ZAR price as well as a link to the full analysis.
The report is publically accessible.
