Section 6

Author: susantee

Section 6/README.md

@@ -0,0 +1,36 @@
+# Section 6
+
+- `main.py`: section 6.2, client GUI and websocket client
+- `tests/`: section 5 unit tests
+  - `test_rx.py`: section 5.1 basic reactive test
+  - `test_transaction_form.py`: section 5.2 reactive GUI test
+  - `test_server.py`: section 5.3 reactive Tornado server test
+- `server/` directory contains code from:
+    - `matching_server.py`: section 4.3 web server, section 6.1
+      managing orders on the web server
+    - `transaction_server.py`: section 5.4 cluster testing
+- `client/` directory contains code from:
+  - section 3 reactive GUIs and data flows
+  - `client.py`: section 4.4 real-time async client receiver/sender
+    (used in section 5.4 cluster testing)
+
+Server-side Streams:
+
+* orders per user
+* all orders (combination of orders per user)
+* prices (aggregate of orders grouped by stock symbol)
+
+Client-side Streams:
+
+* prices (from server)
+* orders (from user input or from bot)
+
+Cluster test:
+
+    cd server
+    python matching_server.py 8888 &
+    python matching_server.py 7777 &
+    python transaction_server.py &
+    cd ..
+    cd client
+    python client.py

Section 6/client/client.py

@@ -0,0 +1,72 @@
+import asyncio
+import datetime
+from random import choice
+
+from rx import Observable
+from rx.subjects import Subject
+from rx.concurrency import IOLoopScheduler
+
+from tornado.ioloop import IOLoop
+from tornado.websocket import websocket_connect
+
+class Client:
+    def __init__(self, host='localhost', port='8888'):
+        self._url = 'ws://{}:{}/exchange'.format(host, port)
+        self.conn = None
+        self.opened = Subject()
+        self.messages = Subject()
+        # self.messages.subscribe(lambda msg: print('received: {}'.format(msg)))
+
+    def connect(self):
+        def on_connect(conn):
+            print('on_connect')
+            self.conn = conn
+            self.opened.on_next(conn)
+            self.opened.on_completed()
+            self.opened.dispose()
+
+        def on_message_callback(message):
+            # print('on_message_callback')
+            self.messages.on_next(message)
+
+        print('connect to server')
+        future = websocket_connect(
+            self._url,
+            on_message_callback=on_message_callback,
+        )
+        Observable.from_future(future).subscribe(on_connect)
+
+    def write_message(self, message):
+        self.conn.write_message(message)
+
+if __name__ == '__main__':
+    scheduler = IOLoopScheduler(IOLoop.current())
+
+    def make_say_hello(client, client_id):
+        def say_hello():
+            print('{} client #{} is sending orders'.format(
+                datetime.datetime.now(), client_id))
+            symbols = ['ABC', 'DEF', 'GHI', 'A', 'GS', 'GO']
+            quantities = [90, 100, 110]
+            prices = [1.20, 1.21, 1.22, 1.23, 1.24, 1.25]
+            client.write_message(
+                'order,{},{},buy,{},{}'.format(client_id, choice(symbols), choice(quantities), choice(prices)))
+            client.write_message(
+                'order,{},{},sell,{},{}'.format(client_id, choice(symbols), choice(quantities), choice(prices)))
+
+        def schedule_say_hello(conn):
+            sleep = 5000
+            Observable \
+                .interval(sleep, scheduler=scheduler) \
+                .subscribe(lambda value: say_hello())
+        return schedule_say_hello
+
+    clients = []
+    for client_id in range(10):
+        client = Client(port='9999')
+        client.opened.subscribe(make_say_hello(client, client_id))
+        clients.append(client)
+
+    for client in clients:
+        client.connect()
+    IOLoop.current().start()

Section 6/client/client_window.py

@@ -0,0 +1,27 @@
+from PyQt5.QtWidgets import QWidget, QGridLayout
+
+from rx.subjects import Subject
+
+from client.transaction_form import TransactionForm
+from client.stock_overview_table import StockOverviewTable
+
+class ClientWindow(QWidget):
+    def __init__(self, *args, **kwargs):
+        stock_prices_stream = kwargs.pop('stock_prices_stream')
+        QWidget.__init__(self, *args, **kwargs)
+        self.events = Subject()
+        self._setup_window()
+        self._layout = QGridLayout(self)
+        self._overview_table = StockOverviewTable(stock_prices_stream=stock_prices_stream)
+        self._form = TransactionForm()
+        self._layout.addWidget(self._overview_table, 0, 0)
+        self._layout.addWidget(self._form, 0, 2)
+        self._layout.setColumnStretch(0, 2)
+
+    def _setup_window(self):
+        self.resize(640, 320)
+        self.move(350, 200)
+        self.setWindowTitle('client: stock exchange')
+
+    def get_orders(self):
+        return self._form.orders_stream

Section 6/client/stock_overview_table.py

@@ -0,0 +1,32 @@
+from PyQt5 import QtCore
+from PyQt5.QtWidgets import QTableWidget, QTableWidgetItem
+
+class StockOverviewTable(QTableWidget):
+    def __init__(self, *args, **kwargs):
+        stock_prices_stream = kwargs.pop('stock_prices_stream')
+        QTableWidget.__init__(self, *args, **kwargs)
+        self.setRowCount(0)
+        self.setColumnCount(4)
+        self.setHorizontalHeaderLabels(['Symbol', 'Name', 'Buy Price', 'Sell Price'])
+        self.setColumnWidth(0, 50)
+        self.setColumnWidth(1, 200)
+        self.setColumnWidth(2, 100)
+        self.setColumnWidth(3, 100)
+        self.horizontalHeader().setStretchLastSection(True)
+        self.setSortingEnabled(True)
+        stock_prices_stream.subscribe(self._create_or_update_stock_row)
+
+    def _find_matching_row_index(self, stock_row):
+        matches = self.findItems(stock_row[0], QtCore.Qt.MatchExactly)
+        if len(matches) == 0:
+            self.setRowCount(self.rowCount() + 1)
+            return self.rowCount() - 1
+        return self.indexFromItem(matches[0]).row()
+
+    def _create_or_update_stock_row(self, stock_row):
+        row = self._find_matching_row_index(stock_row)
+
+        column_index = 0
+        for column in stock_row:
+            self.setItem(row, column_index, QTableWidgetItem(str(column)))
+            column_index += 1

Section 6/client/transaction_form.py

@@ -0,0 +1,66 @@
+from PyQt5.QtWidgets import QFormLayout, QLabel, QLineEdit, QPushButton, QWidget
+
+from rx.subjects import Subject
+
+from client.transaction_validator import TransactionValidator
+
+class TransactionForm(QWidget):
+    def __init__(self, *args, **kwargs):
+        self.orders_stream = Subject()
+        self._validator = TransactionValidator()
+        QWidget.__init__(self, *args, **kwargs)
+        self._setup_form()
+        self._connect_events()
+        self._subscribe_to_streams()
+
+    def _setup_form(self):
+        self._symbol_input = QLineEdit()
+        self._symbol_error_label = QLabel()
+        self._price_input = QLineEdit()
+        self._price_error_label = QLabel()
+        self._ok_button = QPushButton('Post Order')
+        self._layout = QFormLayout(self)
+        self._layout.addRow('Symbol:', self._symbol_input)
+        self._layout.addRow(self._symbol_error_label)
+        self._layout.addRow('Price:', self._price_input)
+        self._layout.addRow(self._price_error_label)
+        self._layout.addRow(self._ok_button)
+        self._clear_form()
+
+    def _connect_events(self):
+        self._ok_button.clicked.connect(self.submit_order)
+        self._symbol_input.textChanged.connect(self._symbol_text_changed)
+        self._price_input.textChanged.connect(self._price_text_changed)
+
+    def _subscribe_to_streams(self):
+        def invalid_field(error):
+            self.set_error_label(*error)
+        self._validator.on_error(invalid_field)
+
+        def enable_order_submission(order):
+            self._ok_button.setEnabled(True)
+        self._validator.on_valid_order(enable_order_submission)
+
+    def set_error_label(self, field, text):
+        label = '_{}_error_label'.format(field)
+        getattr(self, label).setText(text)
+
+    def _clear_form(self):
+        for widget_name in ['symbol_input', 'symbol_error_label', 'price_input', 'price_error_label']:
+            widget = '_{}'.format(widget_name)
+            getattr(self, widget).setText('')
+        self._ok_button.setEnabled(False)
+
+    def submit_order(self):
+        self.orders_stream.on_next(self._validator.latest_valid_order)
+        self._clear_form()
+
+    def _symbol_text_changed(self):
+        self._symbol_error_label.setText('')
+        value = self._symbol_input.text()
+        self._validator.next_symbol(value)
+
+    def _price_text_changed(self):
+        self._price_error_label.setText('')
+        price_string = self._price_input.text()
+        self._validator.next_price(price_string)

Section 6/client/transaction_validator.py

@@ -0,0 +1,62 @@
+import re
+
+from rx import Observable
+from rx.subjects import Subject
+
+class TransactionValidator:
+    def __init__(self):
+        self._error_stream = Subject()
+        self._price_stream = Subject()
+        self._symbol_stream = Subject()
+        self.latest_valid_order = None
+
+    def next_error(self, field, error_text):
+        self._error_stream.on_next([field, error_text])
+
+    # We're going to check if the price text has changed to a valid
+    # value. If it is an empty string or not a number that's greater
+    # than 0.0 we're going to emit an error in the error stream. If
+    # it's valid, we will emit the value to the price stream.
+    def next_price(self, value):
+        if len(value) == 0:
+            self.next_error('price', 'cannot be blank')
+        else:
+            try:
+                price = float(value)
+                if price < 0.01:
+                    self.next_error('price', 'must be greater than 0.00')
+                else:
+                    self._price_stream.on_next(price)
+            except ValueError:
+                if len(re.sub('[a-zA-Z ]', '', value)) == 0:
+                    self.next_error('price', 'must be a number')
+
+    # We're going to check if the symbol text has changed.  It's a
+    # simple check, we want to make sure it isn't an empty string. If
+    # it's valid, we will emit the value on the symbol stream. If it's
+    # invalid, we'll emit the error on the error stream.
+    def next_symbol(self, value):
+        if len(value) == 0 or len(re.sub('[0-9 ]', '', value)) == 0:
+            self.next_error('symbol', 'cannot be blank')
+        else:
+            self._symbol_stream.on_next(value)
+
+    def on_error(self, func):
+        self._error_stream.subscribe(func)
+
+    def on_valid_order(self, func):
+        # Combine latest will emit items when any of the observables
+        # have an item to emit. We want the latest valid values that
+        # the user has entered in the symbol and price input
+        # boxes. When we have that, we can enable the "submit order"
+        # button. The form inputs and buttons are manipulated through
+        # the subscription function that was passed in.
+        def store_order_and_send_to_subscriber(order):
+            self.latest_valid_order = order
+            func(order)
+
+        Observable.combine_latest(
+            self._symbol_stream,
+            self._price_stream,
+            lambda symbol, price: { 'symbol': symbol, 'price': price }
+        ).subscribe(store_order_and_send_to_subscriber)

Section 6/main2.py

@@ -0,0 +1,43 @@
+import sys
+import random
+import datetime
+import asyncio
+from threading import Thread
+
+from PyQt5 import QtCore
+from PyQt5.QtWidgets import QApplication
+from tornado.ioloop import IOLoop
+
+from rx import Observable
+from rx.subjects import Subject
+from rx.concurrency import QtScheduler, AsyncIOScheduler
+
+from client.client import Client
+from client.client_window import ClientWindow
+
+import utils
+
+if __name__ == '__main__':
+    app = QApplication(sys.argv)
+    scheduler = QtScheduler(QtCore)
+    stock_prices = Subject()
+    client = Client(port='9999')
+
+    loop = asyncio.new_event_loop()
+    asyncio_scheduler = AsyncIOScheduler(loop)
+
+    def run_client():
+        asyncio.set_event_loop(loop)
+        asyncio.get_event_loop().call_soon(
+            utils.run_client_websocket, client, stock_prices)
+        asyncio.get_event_loop().run_forever()
+
+    thread = Thread(target=run_client, daemon=True)
+    thread.start()
+
+    client_window = ClientWindow(stock_prices_stream=stock_prices)
+    client_window.get_orders() \
+               .filter(utils.order_is_valid) \
+               .subscribe(lambda x: print(x), lambda x: print('error'))
+    client_window.show()
+    sys.exit(app.exec_())

Section 6/main3.py

@@ -0,0 +1,54 @@
+import sys
+import random
+import datetime
+import asyncio
+from threading import Thread
+
+from PyQt5 import QtCore
+from PyQt5.QtWidgets import QApplication
+from tornado.ioloop import IOLoop
+
+from rx import Observable
+from rx.subjects import Subject
+from rx.concurrency import QtScheduler, AsyncIOScheduler
+
+from client.client import Client
+from client.client_window import ClientWindow
+
+import utils
+
+if __name__ == '__main__':
+    app = QApplication(sys.argv)
+    scheduler = QtScheduler(QtCore)
+    stock_prices = Subject()
+    client = Client(port='9999')
+
+    loop = asyncio.new_event_loop()
+    asyncio_scheduler = AsyncIOScheduler(loop)
+
+    def run_client():
+        asyncio.set_event_loop(loop)
+        asyncio.get_event_loop().call_soon(
+            utils.run_client_websocket, client, stock_prices)
+        asyncio.get_event_loop().run_forever()
+
+    thread = Thread(target=run_client, daemon=True)
+    thread.start()
+
+    client_window = ClientWindow(stock_prices_stream=stock_prices)
+
+    def send_order(order):
+        stock_order = {
+            'stock_symbol': order['symbol'],
+            'price': order['price'],
+            'direction': 'buy',
+            'quantity': 1,
+        }
+        utils.write_order(client, stock_order)
+
+    client_window.get_orders() \
+        .filter(utils.order_is_valid) \
+        .subscribe(send_order)
+
+    client_window.show()
+    sys.exit(app.exec_())