test_kafka.py

from __future__ import print_function
import sys, os
# sys.path.append('..')
# sys.path.append(os.path.abspath(os.path.dirname(__file__)) + '..' + 'orderbook')
sys.path.append(os.path.join(os.path.dirname(__file__) , '..' ))

import unittest
from unittest import TestCase
from run_bond_hedge import *
import time
import json
from pprint import pprint
from kafka import KafkaConsumer


print ("Connecting to Kafka on group_id = group3")

f = open("results.txt", "w")

# To consume latest messages and auto-commit offsets
consumer = KafkaConsumer('transaction-sink',
                         group_id='group3',
                         bootstrap_servers=['kafka:9092'],
                         client_id='python-client')

consumer.poll(1)

for message in consumer:
    # with open('../resources/json_sample') as data_file:
    #     data = json.load(data_file)
    data = json.load(message)
    data_dict = data['orderMarkdownDetail'] # gets the dict back

    # 1. Create a trade object
    tr = create_trade(data_dict)

    # 2. Create a dummy hedge
    hedge = create_hedge()

    # 3. run the markout calculations (hedge/unhedge)
    hedge_markout = run_bond_hedge(tr, hedge)

    # 4. output to json format //TODO: Need to write back to kafka queue
    json.dump(hedge_markout.__dict__,f)

f.close()