Feature/bitcoin demo

concord/src/CoinbaseMatchedOrderMovingAverage.py

@@ -0,0 +1,57 @@
+import json
+import sys
+import time
+import concord
+from collections import deque
+from concord.computation import (Computation, Metadata, serve_computation,
+                                 StreamGrouping)
+from models.CoinbaseOrder import CoinbaseOrder
+from utils.time_utils import (time_millis, bottom_of_current_second,
+                              nseconds_from_now_in_millis)
+from utils.kafka_utils import local_kafka_producer
+
+
+# TODO(agallego): use pandas / numpy
+# stores 1k numbers as the avg. simpler, but def not correct
+class MovingAvg(deque):
+    def __init__(self, size=1000):
+        super(MovingAvg, self).__init__(maxlen=size)
+
+    @property
+    def average(self):
+        if len(self) <= 0: return 0
+        return sum(self)/len(self)
+
+class CoinbaseMatchedOrderMovingAverage(Computation):
+    def __init__(self):
+        self.moving_average = MovingAvg()
+        self.producer = local_kafka_producer('match-avg')
+
+    def init(self, ctx):
+        self.concord_logger.info("Matched order moving avg init")
+
+    def process_record(self, ctx, record):
+        order = CoinbaseOrder(record.data)
+        if order.type == 'match':
+            self.concord_logger.info("Found matched order at price: %s",
+                                     str(order.price))
+            # it has to include 0's in the millisecond and micro second
+            # parts of the time to be reasonable for updates to kafka
+            sec = (bottom_of_current_second() * 1000) + 1000
+            self.moving_average.append(order.price)
+            ctx.set_timer(str(sec), sec)
+
+    def process_timer(self, ctx, key, time):
+        d = {
+            'time': int(key),
+            'avg': self.moving_average.average
+        }
+        self.producer.produce(json.dumps(d))
+        self.concord_logger.info("Saving to kafka for time: %s, avg price: %s",
+                                 key, str(self.moving_average.average))
+
+    def metadata(self):
+        return Metadata(name='match-orders-sec-avg',
+                        istreams=[('btcusd', StreamGrouping.GROUP_BY)])
+
+serve_computation(CoinbaseMatchedOrderMovingAverage())

concord/src/CoinbasePricePrinter.py

@@ -0,0 +1,104 @@
+import json
+import sys
+import unicodedata
+import logging
+import concord
+from concord.computation import (Computation, Metadata, serve_computation)
+import cachetools
+
+def time_millis(): return int(round(time.time() * 1000))
+def next_second(sec=1): return time_millis() + (sec * 1000)
+
+
+
+class CoinbasePricePrinter(Computation):
+    def __init__(self):
+        from cachetools import TTLCache
+        def missing_order(order_type):
+            return CoinbaseOrder(
+                {'time': '1970-01-01T00:00:0Z',
+                 'sequence': 0,
+                 'price': 0,
+                 'type': order_type})
+
+        self.order_cache = TTLCache(10, 300, missing=missing_order)
+        self.volume_cache = TTLCache(10, 300, missing=lambda x: 0)
+
+
+    def update_volume(self, coinbase_order, context):
+        current = self.volume_cache[coinbase_order.time]
+        self.volume_cache[c.time] = (current +
+                                     (coinbase_order.price * coinbase.volume))
+        context.set_state(CoinbasePricePrinter.volume_key(coinbase_order.time),
+                          CoinbaseOrder.to_json(self.volume_cache[c.time]))
+
+
+    # TODO(agallego) map(filter(test,cache)cache)
+    def update_orders(self, coinbase_order, context):
+        if not coinbase_order.valid(): return
+        current = self.order_cache[coinbase_order.type]
+        if not current.valid():
+            self.order_cache[coinbase_order.type] = coinbase_order
+        else if CoinbaseOrder.is_combinable(coinbase_order, current):
+            self.order_cache[coinbase_order.type] = CoinbaseOrder.combie(
+                current, coinbase_order)
+        else:
+            raise Exception("Couldn't combine with any orders", coinbase_order)
+
+        context.set_state(CoinbasePricePrinter.durable_key(coinbase_order),
+                          CoinbasePricePrinter.to_json(
+                              self.order_cache[coinbase_order.type]))
+
+    def init(self, ctx):
+        log.info("Price Printer initialized")
+        ctx.set_timer("loop", next_second())
+
+    def process_record(self, ctx, record):
+        try:
+            c = CoinbaseOrder(record.data)
+            self.update_volume(c)
+            self.update_orders(c)
+        except as e:
+            log.exception(e)
+
+    def process_timer(self, ctx, key, time):
+        ctx.produce_record('btcusd-volume', self.volume_key(time),
+                           self.cache(self.volume_key(time)))
+        ctx.produce_record('btcusd-buy', self.buy_key(time),
+                           self.cache(self.buy_key(time)))
+        ctx.produce_record('btcusd-sell', self.sell_key(time),
+                           self.cache(self.sell_key(time)))
+        ctx.set_timer(key, next_second())
+
+    def metadata(self):
+        return Metadata(name='coinbase-price-printer',
+                        istreams=['btcusd'], ostreams=[])
+
+    @staticmethod
+    def durable_key(coinbase_order):
+        key_map = {
+            'sell' = CoinbasePricePrinter.sell_key(coinbase_order.time),
+            'buy' = CoinbasePricePrinter.buy_key(coinbase_order.time),
+        }
+        return key_map[coinbase_order.type]
+
+    @staticmethod
+    def to_bottom_of_second(time_stamp_millis):
+        return time_stamp_millis - (time_stamp_millis % 1000)
+
+    @staticmethod
+    def volume_key(time_stamp):
+        return 'volume-' + str(
+            CoinbasePricePrinter.to_bottom_of_second(time_stamp))
+
+    @staticmethod
+    def sell_key(time_stamp):
+        return 'sell-' + str(
+            CoinbasePricePrinter.to_bottom_of_second(time_stamp))
+
+    @staticmethod
+    def buy_key( time_stamp):
+        return 'buy-' + str(
+            CoinbasePricePrinter.to_bottom_of_second(time_stamp))
+
+serve_computation(CoinbasePricePrinter())

CoinbaseSource.py → concord/src/CoinbaseSource.py

@@ -41,7 +41,7 @@ def __init__(self):
         self.queue = Queue()
     def init(self, ctx):
         ctx.set_timer('loop', time_millis() + 1000) # start in 1 sec
-        log.info("Coinbase initialized")
+        self.concord_logger.info("Coinbase initialized")
     def process_timer(self, ctx, key, time):
         while not self.queue.empty():
             ctx.produce_record('btcusd', 'empty', self.queue.get())

concord/src/LatestMatchPrice.py

@@ -0,0 +1,31 @@
+import json
+import sys
+import time
+import concord
+from concord.computation import (Computation, Metadata, serve_computation,
+                                 StreamGrouping)
+from models.CoinbaseOrder import CoinbaseOrder
+from utils.time_utils import (time_millis, bottom_of_current_second,
+                              nseconds_from_now_in_millis)
+from utils.kafka_utils import local_kafka_producer
+
+class LatestMatchPrice(Computation):
+    def __init__(self):
+        self.producer = local_kafka_producer('latest-match-price')
+
+    def init(self, ctx):
+        self.concord_logger.info("Latest price init")
+
+    def process_record(self, ctx, record):
+        order = CoinbaseOrder(record.data)
+        if order.type == 'match':
+            self.concord_logger.info("Emitting latest price: %s",
+                                     str(order.price))
+            d = {'time': time_millis(), 'price': order.price}
+            self.producer.produce(json.dumps(d))
+
+    def metadata(self):
+        return Metadata(name='latest-price',
+                        istreams=[('btcusd', StreamGrouping.GROUP_BY)])
+
+serve_computation(LatestMatchPrice())

concord/src/coinbase_latest_price.json

@@ -0,0 +1,11 @@
+{
+  "zookeeper_hosts": "localhost:2181",
+  "zookeeper_path": "/concord",
+  "executable_arguments": ["LatestMatchPrice.py"],
+  "executable_name": "runner.bash",
+  "compress_files": ["runner.bash", "."],
+  "exclude_compress_files": [".*\\.pyc$", ".*\\.log$", ".*\\.json"],
+  "computation_name": "latest-price",
+  "mem": 256,
+  "cpus": 0.5
+}

concord/src/coinbase_moving_avg.json

@@ -0,0 +1,11 @@
+{
+  "zookeeper_hosts": "localhost:2181",
+  "zookeeper_path": "/concord",
+  "executable_arguments": ["CoinbaseMatchedOrderMovingAverage.py"],
+  "executable_name": "runner.bash",
+  "compress_files": ["runner.bash", "."],
+  "exclude_compress_files": [".*\\.pyc$", ".*\\.log$", ".*\\.json"],
+  "computation_name": "match-orders-sec-avg",
+  "mem": 1024,
+  "cpus": 1.0
+}

coinbase_source.json → concord/src/coinbase_source.json

@@ -1,9 +1,10 @@
 {
   "zookeeper_hosts": "localhost:2181",
-  "zookeeper_path": "/bolt",
+  "zookeeper_path": "/concord",
   "executable_arguments": ["CoinbaseSource.py"],
   "executable_name": "runner.bash",
-  "compress_files": ["CoinbaseSource.py", "runner.bash", "requirements.txt"],
+  "compress_files": ["runner.bash", "."],
+  "exclude_compress_files": [".*\\.pyc$", ".*\\.log$", ".*\\.json"],
   "computation_name": "coinbase-indx",
   "mem": 1024,
   "cpus": 1.0

concord/src/models/Avg.py

@@ -0,0 +1,11 @@
+class Avg:
+    def __init__(self):
+        self.array = []
+        self.count = 0.0
+    def avg(self):
+        if self.count == 0:
+            return 0
+        return float(sum(self.array))/self.count
+    def append(self, point):
+        self.array.append(point)
+        self.count = self.count + 1

concord/src/models/CoinbaseOrder.py

@@ -0,0 +1,49 @@
+# coding: utf-8
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+from __future__ import unicode_literals
+
+import time
+import json
+import dateutil
+
+class CoinbaseOrder:
+    def __init__(self, json_data):
+        r = json.loads(json_data)
+        for key in r: setattr(self, key, r[key])
+        from dateutil.parser import parse
+        self.time = parse(self.time)
+        self.sequence = int(self.sequence)
+        if self.type in ['match', 'open', 'close']:
+            self.price = float(self.price)
+
+    def valid(self):
+        return self.sequence > 0
+
+
+    @staticmethod
+    def to_json(coinbase_order):
+        return json.dumps(coinbase_order,
+                          default=lambda o: o.__dict__,
+                          sort_keys=True)
+
+    @staticmethod
+    def from_json(byte_array):
+        return CoinbaseOrder(json.loads(byte_array))
+
+    @staticmethod
+    def combine(lhs_coinbase, rhs_coinbase):
+        if not is_combinable(lhs_coinbase, rhs_coinbase):
+            raise Exception("Cannot combine orders")
+        ret = lhs_coinbase
+        ret.price += rhs_coinbase.price
+        return ret
+
+    @staticmethod
+    def is_combinable(lhs_coinbase, rhs_coinbase):
+        try:
+            return (lhs_coinbase.type == rhs_coinbase.type and
+                    lhs_coinbase.side == rhs_coinbase.side)
+        except:
+            return false

requirements.txt → concord/src/requirements.txt

@@ -1,7 +1,8 @@
 autobahn==0.10.5.post2
 backports.ssl-match-hostname==3.4.0.2
+cachetools==1.0.3
 cffi==1.1.2
-concord-py==0.2.5
+concord-py==0.2.21
 cryptography==1.0
 enum34==1.0.4
 graphviz==0.4.3
@@ -10,8 +11,11 @@ ipaddress==1.0.14
 kazoo==2.0
 pyasn1==0.1.8
 pycparser==2.14
+pykafka==1.1.1
 pyOpenSSL==0.15.1
+python-dateutil==2.4.2
 six==1.9.0
+tabulate==0.7.5
 thrift==0.9.2
 trace2html==0.2.1
 Twisted==15.3.0

concord/src/runner.bash

@@ -0,0 +1,39 @@
+#!/bin/bash
+# set -x
+
+SOURCE="${BASH_SOURCE[0]}"
+while [ -h "$SOURCE" ]; do # resolve $SOURCE until the file is no longer a symlink
+    exec_dir="$( cd -P "$( dirname "$SOURCE" )" && pwd )"
+    SOURCE="$(readlink "$SOURCE")"
+    [[ $SOURCE != /* ]] && SOURCE="$exec_dir/$SOURCE"
+    # if $SOURCE was a relative symlink, we need to resolve it
+    # relative to the path where the symlink file was located
+done
+exec_dir="$( cd -P "$( dirname "$SOURCE" )" && pwd )"
+original=${PWD}
+requirements=$(find $exec_dir -name requirements.txt)
+
+echo "Mesos directory: ${PWD}"
+echo "Exec directory: ${exec_dir}"
+
+if [[ -f $requirements ]]; then
+    # need to overcome pip 128 chars path - software... :'(
+    work_dir=$(mktemp -d -p $original)
+    symlink_dir=$(mktemp -d)
+    ln -s $work_dir $symlink_dir/concord
+    dir=$symlink_dir/concord
+    cd $dir
+    echo "Installing venv in $dir"
+    virtualenv $dir/env
+    $dir/env/bin/pip install -r $requirements
+    cd $exec_dir
+    exec $dir/env/bin/python "$original/$@"
+else
+    exec python "$original/$@"
+fi
+
+rc = $?
+if [[ $rc != 0 ]]; then
+    echo "Client exited with: $rc"
+    exit $rc
+fi