PART FIVE: Error Handling — Building a Production-Ready Algorithmic Trading Framework in Python
Nothing in the data world is certain. Data can come in triangle shapes, while your system was built with square holes. Software is designed to do what it's told, and sometimes if you make it do something it was not designed to do, it will error out or, in Python's case, throw an exception.
As an aspiring data lord designing your ETL processes, error handling is necessary. Errors should ripple up through the different layers of the program, allowing you to know where the hunt and debug should start.
Adding this layering effect of errors passing through to each class allows a developer to sense-check the flow of the program. If you are expecting a value and None is returned, it's a great time to log an error and capture the following:
- Date & Time
- Error Message
- Error Codes
- What's Running
- Stacktrace
As always, here is some code to get you started and some boilerplate code to demonstrate what's happening. A prerequisite for this software is my other article about logging. I am writing a series that walks you through how to build an Algorithmic trading platform. Check it out!
Let me know in the comments if you want to see logging into an SQL database or into an Azure App Insights.
The Code
import inspect
import datetime as dt
from logging import Logger
class ErrorHandling:
def __init__(self, logger: Logger):
self.logger = logger
# Global class variables, will be available through inheritance
ErrorList: list = []
ErrorsDetectedCount: int = 0
ErrorsDetected: bool = False
def error_details(self, message, stack_trace: str = str(0), error_code: str = str(0)):
"""
Used to create error information in your code.
:param stack_trace:
:param message: Log information.
:param error_code: The returned error code. Useful when making HTTP requests.
:return: Error details in an array.
"""
self.ErrorsDetectedCount = + 1
self.logger.error(message)
return [str(dt.datetime.now()), str(message), str(inspect.stack()[1][3]), str(self.ErrorsDetectedCount), stack_trace, error_code]
def print_all_errors(self):
self.logger.error([f"\n\n--- ERRORS DETECTED: {error} ---- \n\n" for error in self.ErrorList])If you save the following code as ‘ErrorHandling.py’, we can start with the boilerplate code below. The code below shows a function that will calculate the returns over a number of days from a chosen starting index.
import numpy
from scipy import signal
from Logger import log_maker
from ErrorHandling import ErrorHandling
import numpy as np
logger = log_maker('Calculation')
class Calculation(ErrorHandling):
def calculate_period_return(self, pct_change_data: np.ndarray, start_index: int, num_days: int=3):
"""
:param pct_change_data: an array of returns data
:param start_index: number to start summing from
:param num_days: how many days into the future to sum
:return: summed returns over time
"""
# Create a None type object for return value,
# will be used to ripple errors up as it can be checked if None
period_return = None
try:
# Check my input data has information
if pct_change_data.shape != (0,0):
# Select the pct_change_data array from the start_index
data = pct_change_data[start_index:]
# Calculate the cumulative returns over the specified number of days
cumulative_returns = np.cumprod(1 + data[-num_days:]) - 1
# Calculate the sum of the returns over the specified number of days
period_return = np.sum(cumulative_returns)
# Final check to be certain the final answer has value
if period_return == 0 or period_return == None:
self.ErrorList.append(self.ErrorDetail(f"Error: calculate_period_return -> The Return Value Is null"))
else:
# Using inheritence to utilise the Error handling classes objects and functions to create logging system wide
self.ErrorList.append(self.ErrorDetail(f"Error: calculate_period_return -> The Input Value Is null {pct_change_data.shape}"))
except TypeError as ty:
self.ErrorsDetected = True
self.ErrorList.append(self.ErrorDetail(f"Error: calculate_period_return -> {ty}"))
except Exception as err_:
self.ErrorsDetected = True
# when in an exception we can use try / except to catch and print error messages and stack traces
self.ErrorList.append(self.ErrorDetail(f"Error: calculate_period_return -> {err_}\n{traceback.format_exc()}"))
return period_return
if __name__ == '__main__':
# I inputted a wrong value, and the errors popped off
test = PeaksValleys().calculate_period_return(np.empty((0,0)))
if test == None:
print('Big Problem!!')If all goes well (or not ehe), you should see something like this:
If you did, that's great, you now have a way to triage your ETL processes. For further development, you should learn the different kinds of Errors that can be excepted in Python. You will figure out which ones are the best for your program with experience and Trial&Error.
Wrapping up
And that’s all she wrote, folks. Please learn things along the way and help me to improve this framework. I could be better, and I always love to improve my software. So please drop me a message here or on git if anything in this project can be improved. All the code will be available there when the series is finished.
I will add links to each article as they become available alongside the bullet points above. So please follow me and give me a clap if you are excited about this project as I am to write about it.
If you want to help buy me a coffee that fuels these articles’ late nights of research and development, please consider donating to my PayPal link below. Thanks so much, and have a great day.
May you forever be profitable.
