The event-driven approach is a way to think about software architecture. Developing and maintaining a considerable trading application while thinking in procedures and functions paradigm could be inefficient and even impossible.
We can think about trading software as a system that reacts to some random events in the real world. Something happens in the market; prices change, which is an event. We can process it inside our system and produce another event, order, for example, it conditions are met. We can also create a logging or notification event based on our needs.
The event-driven approach helps us better understand what is happening inside the system, maintain control, and support clearer and more efficient software architecture.
Events are the fundamental element of our architecture. You can think of events as data classes or data structures. They can contain a lot of information like the event type, timestamp, specific data, etc.
Example:
class Event(object):
#base class
pass
class AccountUpdateEvent(Event):
def __init__(self, timestamp, username, total):
self.type = 'ACCOUNT_UPDATE'
self.timestamp = timestamp
self.username = username
self.total = float(total)
Now we will build a simple application that can produce and process events in less than 100 lines of python code!
Our first module is going to make API calls and produce events.
Example code(whole class is available in my GitHub repo):
class GetAccountData:
def create_event(self, timestamp, username, total):
self.events.put(AccountUpdateEvent(timestamp=timestamp, username=username, total=total))
The second module processes events.
Example code(whole class is available in my GitHub repo):
class ProcessAccountData:
def update_account_data(self, event):
print('Account update received at: ', event.timestamp, 'Username: ', event.username, 'Total balance: ', event.total)
And, finally, our app. It will contain both GetAccountData and ProcessAccountData and a queue of events.
Example code(whole class is available in my GitHub repo):
class App:
def __init__(self):
self.events = Queue()
self.get_data_account = GetAccountData(self.events)
self.process_data_account = ProcessAccountData()
This app contains a loop to process events. Once a new event is in our queue, we take it out and process it.
Example code(whole class is available in my GitHub repo):
def event_loop(self):
while True:
try:
event = self.events.get(False)
except Exception as e:
time.sleep(1)
break
else:
try:
if event is not None:
if event.type == 'ACCOUNT_UPDATE':
self.process_data_account.update_account_data(event)
except Exception as e:
print('event_loop error:', e)
This is an example of a simple event-driven application. You can find the source code in my GitHub repo.
Top comments (2)
Such a detailed and helpful blog!!! Thanks!!
thanks!