[WIP] work on returnTicker

alembic/versions/e032ff832b82_create_ticker_table.py

@@ -0,0 +1,38 @@
+"""create ticker table
+
+Revision ID: e032ff832b82
+Revises: 4ce9876eea54
+Create Date: 2018-02-10 20:04:45.809762
+
+"""
+from alembic import op
+from sqlalchemy import Column, DateTime, Integer, text
+
+import os.path, sys
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from common_types import SA_TYPE_ADDR, SA_TYPE_TXHASH, SA_TYPE_VALUE, UUID
+
+
+# revision identifiers, used by Alembic.
+revision = 'e032ff832b82'
+down_revision = '4ce9876eea54'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "tickers",
+        Column("token_address", SA_TYPE_ADDR, primary_key=True, nullable=False),
+        Column("quote_volume", SA_TYPE_VALUE, nullable=False),
+        Column("base_volume", SA_TYPE_VALUE, nullable=False),
+        Column("last", SA_TYPE_VALUE, nullable=False),
+        Column("percent_change", SA_TYPE_VALUE, nullable=False),
+        Column("bid", SA_TYPE_VALUE, nullable=False),
+        Column("ask", SA_TYPE_VALUE, nullable=False),
+        Column("modified", DateTime, nullable=False)
+    ) 
+
+
+def downgrade():
+    op.drop_table("tickers")

app/app.py

@@ -1,7 +1,6 @@
 import asyncio
 import asyncpg
 import app.config as config
-from huey import RedisHuey
 import logging
 from os import environ
 from web3 import Web3, HTTPProvider
@@ -28,7 +27,6 @@ class __App:
         def __init__(self):
             self.config = config
             self.db = DB(config)
-            self.huey = RedisHuey(host="redis", result_store=False)
             self.web3 = Web3(HTTPProvider(config.HTTP_PROVIDER_URL))
 
         def __str__(self):

app/services/ticker_observer.py

@@ -0,0 +1,153 @@
+'''
+TODO?
+After the first time this is run,
+we could change it to only grab orders/trades
+on an interval so that it only has to process
+the new orders/trades from the last time this
+was run?
+'''
+
+from aiohttp import web
+from ..app import App
+import asyncio
+from ..src.erc20_token import ERC20Token
+import logging
+from ..src.order_enums import OrderState
+from time import time
+import socketio
+from web3 import Web3
+from datetime import datetime
+
+sio = socketio.AsyncServer()
+app = web.Application()
+sio.attach(app)
+
+ZERO_ADDR = "0x0000000000000000000000000000000000000000"
+ZERO_ADDR_BYTES = Web3.toBytes(hexstr=ZERO_ADDR)
+
+logger = logging.getLogger('ticker_observer')
+
+async def get_tickers():
+
+    async with App().db.acquire_connection() as conn:
+        return await conn.fetch(
+            """
+            SELECT *
+            FROM tickers
+            """
+
+# Maybe there is a fancy SQL statement here that will make this faster? 
+async def get_orders():
+
+    async with App().db.acquire_connection() as conn:
+        return await conn.fetch(
+            """
+            SELECT token_give, amount_give, token_get, amount_get, expires
+            FROM orders
+            """
+
+# TODO: Are trades pruned or do I need to add a WHERE here by block number?    
+async def get_trades():
+
+    async with App().db.acquire_connection() as conn:
+        return await conn.fetch(
+            """
+            SELECT token_give, amount_give, token_get, amount_get
+            FROM trades
+            ORDER BY block_number DESC, date DESC
+            """
+
+# Scan passed in orders and update the `tickers` table. Bid, Ask, quoteVolume, and baseVolume is updated.
+# TODO: quoteVolume, baseVolume
+async def parse_orders(tickers, orders): 
+
+    # This is the object we will use to update `tickers` with.
+    ticker_updates = {}
+
+    for order in orders:
+
+        # Check if this is a buy or sell order by seeing if the given token is ETH
+        side = "buy" if order["token_give"] == ZERO_ADDR_BYTES else "sell"
+
+        if side == "buy":
+
+            this_orders_bid = order["amount_give"] / order["amount_get"]
+
+            # If ticker updates doesn't already have this token, add it to our object to update.
+            if not hasattr(ticker_updates, order["token_get"]):
+
+                ticker_updates[order["token_get"]] = {
+                    'bid': this_orders_bid,
+                }
+
+            else:
+                # If there isn't a current recorded bid OR current recorded bid is lower than the bid in this order.
+                if not has_attr(ticker_updates[order["token_get"]], 'bid') || ticker_updates[order["token_get"]]["bid"] < this_orders_bid:
+                    ticker_updates[order["token_get"]]["bid"] = this_orders_bid
+
+        else:
+
+            this_orders_ask = order["amount_get"] / order["amount_give"]
+
+            # If ticker updates doesn't already have this token, add it to our object to update.
+            if not hasattr(ticker_updates, order["token_give"]):
+
+                ticker_updates[order["token_give"]] = {
+                    'ask': this_orders_ask,
+                }
+
+            else:                    
+                # If there is a current recorded ask OR if current recorded ask is higher than the ask in this order.
+                if not has_attr(ticker_updates[order["token_give"]], 'ask') || ticker_updates[order["token_give"]]["ask"] > this_orders_ask:
+                    ticker_updates[order["token_give"]]["ask"] = this_orders_ask
+
+    # TODO: Submit the ticker_updates to `tickers`
+    # TODO: Fill modified date with "modified": datetime.now().isoformat()
+
+    return
+
+# Scan passed in trades and update the `tickers` table. Last and percentChange is updated.
+# TODO: percentChange
+async def parse_trades(tickers, trades): 
+
+    # This is the object we will use to update `tickers` with.
+    ticker_updates = {}
+
+    for trade in trades:
+
+        # Check if this is a buy or sell order by seeing if the given token is ETH
+        side = "buy" if trade["token_give"] == ZERO_ADDR_BYTES else "sell"
+
+        if side == "buy"
+
+            # If ticker updates doesn't already have this token, add it to our object to update.
+            if not hasattr(ticker_updates, trade["token_get"]):
+                ticker_updates[trade["token_get"]] = {
+                    'last': trade["amount_give"] / trade["amount_get"],
+                    'percentChange': '',
+                }
+
+        else:
+
+            # If ticker updates doesn't already have this token, add it to our object to update.
+            if not hasattr(ticker_updates, trade["token_give"]):
+                ticker_updates[trade["token_give"]] = {
+                    'last': trade["amount_get"] / trade["amount_give"],
+                    'percentChange': '',
+                }
+
+    # TODO: Submit the ticker_updates to `tickers`
+    # TODO: Fill modified date with "modified": datetime.now().isoformat()
+
+    return
+
+# Main function. Grabs orders and trades, then scans through orders and trades to update `tickers`
+async def update_tickers():
+
+    orders = await get_orders();
+    trades = await get_trades();
+    tickers = await get_tickers()
+
+    parse_orders(tickers, orders);
+    parse_trades(tickers, trades);
+

app/services/websocket_server.py

@@ -8,6 +8,7 @@
 import socketio
 from web3 import Web3
 import websockets
+from datetime import datetime
 
 sio = socketio.AsyncServer()
 app = web.Application()
@@ -22,6 +23,55 @@
 def connect(sid, environ):
     print("connect ", sid)
 
+# get returnTicker, grabs data from tickers table
+async def get_tickers(token_hexstr):
+
+    # init
+    where = ''
+    placeholder_args = []
+
+    # if token is passed in, add where addr = token
+    if token_hexstr:
+        where = '("token_address" = $1)'
+        placeholder_args = [Web3.toBytes(hexstr=token_hexstr), ]
+
+    # connect to db and grab data from tickers table
+    async with App().db.acquire_connection() as conn:
+        return await conn.fetch(
+            """
+            SELECT *
+            FROM tickers
+            WHERE {}
+            ORDER BY token_address DESC
+            """.format(where),
+            *placeholder_args)
+
+# format ticker information to camelcase from underscore, sanitize, add modified date
+"""
+    sample return data:
+    ETH_0x8f3470a7388c05ee4e7af3d01d8c722b0ff52374:
+   { tokenAddr: '0x8f3470a7388c05ee4e7af3d01d8c722b0ff52374',
+     quoteVolume: 1000.1,
+     baseVolume: 212.3,
+     last: 0.245,
+     percentChange: 0.0047,
+     bid: 0.243,
+     ask: 0.246,
+     modified: 2018-01-15T15:53:00},
+"""
+def format_ticker(ticker):
+    contract = ERC20Token(ticker["token_address"])
+    return { "{}_{}".format("ETH", Web3.toHex(ticker["token_address"]) : {
+        "tokenAddr": Web3.toHex(ticker["token_address"]),
+        "quoteVolume": str(contract.denormalize_value(ticker["quote_volume"])),
+        "baseVolume": str(contract.denormalize_value(ticker["base_volume"])),
+        "last": str(contract.denormalize_value(ticker["last"])),
+        "percentChange": str(contract.denormalize_value(ticker["percent_change"])),
+        "bid": str(contract.denormalize_value(ticker["bid"])),
+        "ask": str(contract.denormalize_value(ticker["ask"])),
+        "modified": ticker["modified"]
+    }}
+
 def format_trade(trade):
     contract_give = ERC20Token(trade["token_give"])
     contract_get = ERC20Token(trade["token_get"])
@@ -202,24 +252,40 @@ async def get_market(sid, data):
     token = data["token"] if "token" in data and Web3.isAddress(data["token"]) else None
     user = data["user"] if "user" in data and Web3.isAddress(data["user"]) else None
 
+    # response vars
     trades = []
     my_trades = []
     my_funds = []
+
+    # get all tickers
+    tickers = await get_tickers()
+
+    # if token is passed in
     if token:
+
+        # get all trades
         trades = await get_trades(token)
+
+        # get all buy orders
         orders_buys = await get_orders(ZERO_ADDR, token,
                                         state=OrderState.OPEN.name,
                                         sort="(amount_give / amount_get) DESC",
                                         expires_after=current_block)
+
+        # get all sell orders
         orders_sells = await get_orders(token, ZERO_ADDR,
                                         state=OrderState.OPEN.name,
                                         sort="(amount_get / amount_give) ASC",
                                         expires_after=current_block)
+
+        # if user is also passed in 
         if user:
             my_trades = await get_trades(token, user)
             my_funds = await get_transfers(token, user)
 
+    # return this variable
     response = {
+        "returnTicker": [format_ticker(ticker) for ticker in tickers],
         "trades": [format_trade(trade) for trade in trades],
         "myTrades": [format_trade(trade) for trade in my_trades],
         "myFunds": [format_transfer(transfer) for transfer in my_funds],