adelorenzo/stable
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fixing backtesting and perisstence

Browse Source
This commit is contained in:
Gert Wohlgemuth 2018-05-23 00:25:38 -07:00
parent c25aa22690
commit 9fd735a3a0
12 changed files with 355 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
freqtrade/aws/backtesting_lambda.py
View File

@ -1,13 +1,11 @@
import logging
import boto3
import os
from freqtrade.arguments import Arguments
from freqtrade.configuration import Configuration
from freqtrade.optimize.backtesting import Backtesting
import boto3
import simplejson as json
from boto3.dynamodb.conditions import Key, Attr
from boto3.dynamodb.conditions import Key
from freqtrade.aws.tables import get_trade_table, get_strategy_table
from freqtrade.optimize.backtesting import Backtesting
db = boto3.resource('dynamodb')
@ -30,6 +28,9 @@ def backtest(event, context):
'exchange' : name of the exchange we should be using
}
it should be invoked by SNS only to avoid abuse of the system!
:param context:
standard AWS context, so pleaes ignore for now!
:return:
@ -40,13 +41,16 @@ def backtest(event, context):
event['body'] = json.loads(event['body'])
name = event['body']['name']
user = event['body']['user']
# technically we can get all these from teh strategy table
stake_currency = event['body']['stake_currency'].upper()
asset = event['body']['asset']
exchange = event['body']['exchange']
assets = list(map(lambda x: "{}/{}".format(x, stake_currency).upper(), asset))
table = db.Table(os.environ['strategyTable'])
trade_table = get_trade_table()
table = get_strategy_table()
response = table.query(
KeyConditionExpression=Key('user').eq(user) &
@ -104,27 +108,33 @@ def backtest(event, context):
print("persist data in dynamo")
print(result)
for index, row in result.iterrows():
if row['loss'] > 0 or row['profit'] > 0:
item = {
"id": "{}.{}:{}".format(user, name, row['pair']),
"pair": row['pair'],
"count_profit": row['profit'],
"count_loss": row['loss'],
"avg_duration": row['avg duration'],
"avg profit": row['avg profit %'],
"total profit": row['total profit {}'.format(stake_currency)]
data = {
"id": "{}.{}:{}".format(user, name, row['currency']),
"trade": "{} to {}".format(row['entry'].strftime('%Y-%m-%d %H:%M:%S'),
row['exit'].strftime('%Y-%m-%d %H:%M:%S')),
"pair": row['currency'],
"duration": row['duration'],
"profit_percent": row['profit_percent'],
"profit_stake": row['profit_BTC'],
"entry_date": row['entry'].strftime('%Y-%m-%d %H:%M:%S'),
"exit_date": row['exit'].strftime('%Y-%m-%d %H:%M:%S')
}
print(item)
data = json.dumps(data, use_decimal=True)
data = json.loads(data, use_decimal=True)
print(data)
# persist data
trade_table.put_item(Item=data)
else:
raise Exception("sorry we did not find any matching strategy for user {} and name {}".format(user, name))
raise Exception(
"sorry we did not find any matching strategy for user {} and name {}".format(user, name))
else:
raise Exception("no body provided")
def submit(event, context):
def cron(event, context):
"""
this functions submits a new strategy to the backtesting queue
@ -133,8 +143,56 @@ def submit(event, context):
:param context:
:return:
"""
# if topic exists, we just reuse it
client = boto3.client('sns')
topic_arn = client.create_topic(Name=os.environ['topic'])['TopicArn']
table = get_strategy_table()
response = table.scan()
def fetch(response, table):
"""
fetches all strategies from the server
technically code duplications
TODO refacture
:param response:
:param table:
:return:
"""
for i in response['Items']:
# fire a message to our queue
print(i)
message = {
"user": i['user'],
"name": i['name']
}
serialized = json.dumps(message, use_decimal=True)
# submit item to queue for routing to the correct persistence
result = client.publish(
TopicArn=topic_arn,
Message=json.dumps({'default': serialized}),
Subject="schedule backtesting",
MessageStructure='json'
)
print(result)
if 'LastEvaluatedKey' in response:
return table.scan(
ExclusiveStartKey=response['LastEvaluatedKey']
)
else:
return {}
# do/while simulation
response = fetch(response, table)
while 'LastEvaluatedKey' in response:
response = fetch(response, table)
pass
if __name__ == '__main__':
backtest({}, {})

14
freqtrade/aws/exchange.py Normal file
View File

@ -0,0 +1,14 @@
def fetch_pairs (events, context):
"""
fetches the pairs for the given exchange name and currency
requires:
name: name of the exchange
stake_currency: name of the stake currency
:param events:
:param context:
:return:
"""

66
freqtrade/aws/schemas.py
View File

@ -8,7 +8,7 @@ __SUBMIT_STRATEGY_SCHEMA__ = {
"user": {
"$id": "/properties/user",
"type": "string",
"title": "the associated Isaac user",
"title": "The User Schema ",
"default": "",
"examples": [
"GCU4LW2XXZW3A3FM2XZJTEJHNWHTWDKY2DIJLCZJ5ULVZ4K7LZ7D23TG"
@ -17,29 +17,56 @@ __SUBMIT_STRATEGY_SCHEMA__ = {
"description": {
"$id": "/properties/description",
"type": "string",
"title": "a brief description",
"title": "The Description Schema ",
"default": "",
"examples": [
"simple test strategy"
]
},
"exchange": {
"$id": "/properties/exchange",
"type": "object",
"properties": {
"name": {
"$id": "/properties/exchange/properties/name",
"type": "string",
"title": "The Name Schema ",
"default": "",
"examples": [
"binance"
]
},
"stake": {
"$id": "/properties/exchange/properties/stake",
"type": "string",
"title": "The Stake Schema ",
"default": "",
"examples": [
"usdt"
]
},
"pairs": {
"$id": "/properties/exchange/properties/pairs",
"type": "array",
"items": {
"$id": "/properties/exchange/properties/pairs/items",
"type": "string",
"title": "The 0th Schema ",
"default": "",
"examples": [
"btc/usdt"
]
}
}
}
},
"name": {
"$id": "/properties/name",
"type": "string",
"title": "the name of your strategy",
"title": "The Name Schema ",
"default": "",
"examples": [
"TestStrategy"
]
},
"public": {
"$id": "/properties/public",
"type": "boolean",
"title": "Will this strategy be public",
"default": "false",
"examples": [
"true",
"false"
"MyFancyTestStrategy"
]
},
"content": {
@ -48,7 +75,16 @@ __SUBMIT_STRATEGY_SCHEMA__ = {
"title": "The Content Schema ",
"default": "",
"examples": [
"IyAtLS0gRG8gbm90IHJlbW92ZSB0aGVzZSBsaWJzIC0tLQpmcm9tIGZyZXF0cmFkZS5zdHJhdGVneS5pbnRlcmZhY2UgaW1wb3J0IElTdHJhdGVneQpmcm9tIHR5cGluZyBpbXBvcnQgRGljdCwgTGlzdApmcm9tIGh5cGVyb3B0IGltcG9ydCBocApmcm9tIGZ1bmN0b29scyBpbXBvcnQgcmVkdWNlCmZyb20gcGFuZGFzIGltcG9ydCBEYXRhRnJhbWUKIyAtLS0tLS0tLS0tLS0tLS0tLS0tLS0tLS0tLS0tLS0tLQoKaW1wb3J0IHRhbGliLmFic3RyYWN0IGFzIHRhCmltcG9ydCBmcmVxdHJhZGUudmVuZG9yLnF0cHlsaWIuaW5kaWNhdG9ycyBhcyBxdHB5bGliCgpjbGFzcyBUZXN0U3RyYXRlZ3koSVN0cmF0ZWd5KToKICAgIG1pbmltYWxfcm9pID0gewogICAgICAgICIwIjogMC41CiAgICB9CiAgICBzdG9wbG9zcyA9IC0wLjIKICAgIHRpY2tlcl9pbnRlcnZhbCA9ICc1bScKCiAgICBkZWYgcG9wdWxhdGVfaW5kaWNhdG9ycyhzZWxmLCBkYXRhZnJhbWU6IERhdGFGcmFtZSkgLT4gRGF0YUZyYW1lOgogICAgICAgIG1hY2QgPSB0YS5NQUNEKGRhdGFmcmFtZSkKICAgICAgICBkYXRhZnJhbWVbJ21hU2hvcnQnXSA9IHRhLkVNQShkYXRhZnJhbWUsIHRpbWVwZXJpb2Q9OCkKICAgICAgICBkYXRhZnJhbWVbJ21hTWVkaXVtJ10gPSB0YS5FTUEoZGF0YWZyYW1lLCB0aW1lcGVyaW9kPTIxKQogICAgICAgIHJldHVybiBkYXRhZnJhbWUKCiAgICBkZWYgcG9wdWxhdGVfYnV5X3RyZW5kKHNlbGYsIGRhdGFmcmFtZTogRGF0YUZyYW1lKSAtPiBEYXRhRnJhbWU6CiAgICAgICAgZGF0YWZyYW1lLmxvY1sKICAgICAgICAgICAgKAogICAgICAgICAgICAgICAgcXRweWxpYi5jcm9zc2VkX2Fib3ZlKGRhdGFmcmFtZVsnbWFTaG9ydCddLCBkYXRhZnJhbWVbJ21hTWVkaXVtJ10pCiAgICAgICAgICAgICksCiAgICAgICAgICAgICdidXknXSA9IDEKCiAgICAgICAgcmV0dXJuIGRhdGFmcmFtZQoKICAgIGRlZiBwb3B1bGF0ZV9zZWxsX3RyZW5kKHNlbGYsIGRhdGFmcmFtZTogRGF0YUZyYW1lKSAtPiBEYXRhRnJhbWU6CiAgICAgICAgZGF0YWZyYW1lLmxvY1sKICAgICAgICAgICAgKAogICAgICAgICAgICAgICAgcXRweWxpYi5jcm9zc2VkX2Fib3ZlKGRhdGFmcmFtZVsnbWFNZWRpdW0nXSwgZGF0YWZyYW1lWydtYVNob3J0J10pCiAgICAgICAgICAgICksCiAgICAgICAgICAgICdzZWxsJ10gPSAxCiAgICAgICAgcmV0dXJuIGRhdGFmcmFtZQoKCiAgICAgICAg"
"IyAtLS0gRG8gbm90IHJlbW92ZSB0aGVzZSBsaWJzIC0tLQpmcm9tIGZyZXF0cmFkZS5zdHJhdGVneS5pbnRlcmZhY2UgaW1wb3J0IElTdHJhdGVneQpmcm9tIHR5cGluZyBpbXBvcnQgRGljdCwgTGlzdApmcm9tIGh5cGVyb3B0IGltcG9ydCBocApmcm9tIGZ1bmN0b29scyBpbXBvcnQgcmVkdWNlCmZyb20gcGFuZGFzIGltcG9ydCBEYXRhRnJhbWUKIyAtLS0tLS0tLS0tLS0tLS0tLS0tLS0tLS0tLS0tLS0tLQoKaW1wb3J0IHRhbGliLmFic3RyYWN0IGFzIHRhCmltcG9ydCBmcmVxdHJhZGUudmVuZG9yLnF0cHlsaWIuaW5kaWNhdG9ycyBhcyBxdHB5bGliCgpjbGFzcyBNeUZhbmN5VGVzdFN0cmF0ZWd5KElTdHJhdGVneSk6CiAgICBtaW5pbWFsX3JvaSA9IHsKICAgICAgICAiMCI6IDAuNQogICAgfQogICAgc3RvcGxvc3MgPSAtMC4yCiAgICB0aWNrZXJfaW50ZXJ2YWwgPSAnNW0nCgogICAgZGVmIHBvcHVsYXRlX2luZGljYXRvcnMoc2VsZiwgZGF0YWZyYW1lOiBEYXRhRnJhbWUpIC0-IERhdGFGcmFtZToKICAgICAgICBtYWNkID0gdGEuTUFDRChkYXRhZnJhbWUpCiAgICAgICAgZGF0YWZyYW1lWydtYVNob3J0J10gPSB0YS5FTUEoZGF0YWZyYW1lLCB0aW1lcGVyaW9kPTgpCiAgICAgICAgZGF0YWZyYW1lWydtYU1lZGl1bSddID0gdGEuRU1BKGRhdGFmcmFtZSwgdGltZXBlcmlvZD0yMSkKICAgICAgICByZXR1cm4gZGF0YWZyYW1lCgogICAgZGVmIHBvcHVsYXRlX2J1eV90cmVuZChzZWxmLCBkYXRhZnJhbWU6IERhdGFGcmFtZSkgLT4gRGF0YUZyYW1lOgogICAgICAgIGRhdGFmcmFtZS5sb2NbCiAgICAgICAgICAgICgKICAgICAgICAgICAgICAgIHF0cHlsaWIuY3Jvc3NlZF9hYm92ZShkYXRhZnJhbWVbJ21hU2hvcnQnXSwgZGF0YWZyYW1lWydtYU1lZGl1bSddKQogICAgICAgICAgICApLAogICAgICAgICAgICAnYnV5J10gPSAxCgogICAgICAgIHJldHVybiBkYXRhZnJhbWUKCiAgICBkZWYgcG9wdWxhdGVfc2VsbF90cmVuZChzZWxmLCBkYXRhZnJhbWU6IERhdGFGcmFtZSkgLT4gRGF0YUZyYW1lOgogICAgICAgIGRhdGFmcmFtZS5sb2NbCiAgICAgICAgICAgICgKICAgICAgICAgICAgICAgIHF0cHlsaWIuY3Jvc3NlZF9hYm92ZShkYXRhZnJhbWVbJ21hTWVkaXVtJ10sIGRhdGFmcmFtZVsnbWFTaG9ydCddKQogICAgICAgICAgICApLAogICAgICAgICAgICAnc2VsbCddID0gMQogICAgICAgIHJldHVybiBkYXRhZnJhbWUKCgogICAgICAgIA=="
]
},
"public": {
"$id": "/properties/public",
"type": "boolean",
"title": "The Public Schema ",
"default": False,
"examples": [
False
]
}
}

79
freqtrade/aws/service/Persistence.py
View File

@ -1,79 +0,0 @@
import boto3
import simplejson as json
import decimal
class DecimalEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, decimal.Decimal):
if o % 1 > 0:
return float(o)
else:
return int(o)
return super(DecimalEncoder, self).default(o)
class Persistence:
"""
simplistic persistence framework
"""
def __init__(self, table):
"""
creates a new object with the associated table
:param table:
"""
self.table = table
self.db = boto3.resource('dynamodb')
def list(self):
table = self.db.Table(self.table)
response = table.scan()
result = response['Items']
while 'LastEvaluatedKey' in response:
for i in response['Items']:
result.append(i)
response = table.scan(
ExclusiveStartKey=response['LastEvaluatedKey']
)
return result
def load(self, sample):
"""
loads a given object from the database storage
:param sample:
:return:
"""
table = self.db.Table(self.table)
result = table.get_item(
Key={
'id': sample
}
)
if 'Item' in result:
return result['Item']
else:
return None
def save(self, object):
"""
saves and object to the database storage with the specific key
:param object:
:return:
"""
table = self.db.Table(self.table)
# force serialization to deal with decimal number tag
data = json.dumps(object, use_decimal=True)
data = json.loads(data, use_decimal=True)
print(data)
return table.put_item(Item=data)

47
freqtrade/aws/service/Queue.py
View File

@ -1,47 +0,0 @@
import simplejson as json
import os
import boto3
class Queue:
"""
abstraction of the underlaying queuing system to schedule a message to the backend for processing
"""
def submit(self, object, routingKey):
"""
submits the given object to the queue associated with the
routing key.
The routing lambda function will than make sure it will be delivered to the right destination
:param object:
:param routingKey:
:return:
"""
# get topic refrence
client = boto3.client('sns')
# if topic exists, we just reuse it
topic_arn = client.create_topic(Name=os.environ['topic'])['TopicArn']
serialized = json.dumps(object, use_decimal=True)
# submit item to queue for routing to the correct persistence
result = client.publish(
TopicArn=topic_arn,
Message=json.dumps({'default': serialized}),
Subject="route:" + routingKey,
MessageStructure='json',
MessageAttributes={
'route': {
'DataType': 'String',
'StringValue': routingKey
}
},
)
return {
"statusCode": result['ResponseMetadata']['HTTPStatusCode'],
"body": serialized
}

0
freqtrade/aws/service/__init__.py
View File

14
freqtrade/aws/strategy.py
View File

@ -8,6 +8,7 @@ from boto3.dynamodb.conditions import Key, Attr
from jsonschema import validate
from freqtrade.aws.schemas import __SUBMIT_STRATEGY_SCHEMA__
from freqtrade.aws.tables import get_strategy_table
from freqtrade.strategy.resolver import StrategyResolver
import requests
@ -21,7 +22,7 @@ def names(event, context):
:param context:
:return:
"""
table = db.Table(os.environ['strategyTable'])
table = get_strategy_table()
response = table.scan()
result = response['Items']
@ -63,8 +64,7 @@ def get(event, context):
assert 'user' in event['pathParameters']
assert 'name' in event['pathParameters']
table = db.Table(os.environ['strategyTable'])
table = get_strategy_table()
response = table.query(
KeyConditionExpression=Key('user').eq(event['pathParameters']['user']) &
Key('name').eq(event['pathParameters']['name'])
@ -112,8 +112,7 @@ def code(event, context):
user = event['path']['user']
name = event['path']['name']
table = db.Table(os.environ['strategyTable'])
table = get_strategy_table()
response = table.query(
KeyConditionExpression=Key('user').eq(user) &
Key('name').eq(name)
@ -191,7 +190,7 @@ def __evaluate(data):
# force serialization to deal with decimal number
data = json.dumps(data, use_decimal=True)
data = json.loads(data, use_decimal=True)
table = db.Table(os.environ['strategyTable'])
table = get_strategy_table()
result = table.put_item(Item=data)
return result
@ -223,7 +222,8 @@ def submit_github(event, context):
# generate simple id
# submit it
# submit it - we should be able to support multiple repositories
# maybe another database table, where we can map these?
try:
__evaluate({
"name": x['path'].split("/")[-1].split(".py")[0],

93
freqtrade/aws/tables.py Normal file
View File

@ -0,0 +1,93 @@
import os
import boto3
from boto.dynamodb2.exceptions import ResourceInUseException
db = boto3.resource('dynamodb')
def get_trade_table():
"""
provides access to the trade table and if it doesn't exists
creates it for us
:return:
"""
table_name = os.environ['tradeTable']
existing_tables = boto3.client('dynamodb').list_tables()['TableNames']
if table_name not in existing_tables:
try:
db.create_table(
TableName=table_name,
KeySchema=[
{
'AttributeName': 'id',
'KeyType': 'HASH'
},
{
'AttributeName': 'trade',
'KeyType': 'RANGE'
}
],
AttributeDefinitions=[
{
'AttributeName': 'id',
'AttributeType': 'S'
}, {
'AttributeName': 'trade',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 1,
'WriteCapacityUnits': 1
}
)
except ResourceInUseException as e:
print("table already exist {}".format(e))
return db.Table(table_name)
def get_strategy_table():
"""
provides us access to the strategy table and if it doesn't exists creates it for us
:return:
"""
table_name = os.environ['strategyTable']
existing_tables = boto3.client('dynamodb').list_tables()['TableNames']
existing_tables = boto3.client('dynamodb').list_tables()['TableNames']
if table_name not in existing_tables:
try:
db.create_table(
TableName=os.environ[table_name],
KeySchema=[
{
'AttributeName': 'user',
'KeyType': 'HASH'
},
{
'AttributeName': 'name',
'KeyType': 'RANGE'
}
],
AttributeDefinitions=[
{
'AttributeName': 'user',
'AttributeType': 'S'
}, {
'AttributeName': 'name',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 1,
'WriteCapacityUnits': 1
}
)
except ResourceInUseException as e:
print("table already exist {}".format(e))
return db.Table(table_name)

12
freqtrade/optimize/backtesting.py
View File

@ -143,7 +143,9 @@ class Backtesting(object):
pair,
trade.calc_profit_percent(rate=sell_row.close),
trade.calc_profit(rate=sell_row.close),
(sell_row.date - buy_row.date).seconds // 60
(sell_row.date - buy_row.date).seconds // 60,
buy_row.date,
sell_row.date
), \
sell_row.date
return None
@ -200,6 +202,7 @@ class Backtesting(object):
if ret:
row2, trade_entry, next_date = ret
lock_pair_until = next_date
trades.append(trade_entry)
if record:
# Note, need to be json.dump friendly
@ -214,7 +217,8 @@ class Backtesting(object):
if record and record.find('trades') >= 0:
logger.info('Dumping backtest results')
file_dump_json('backtest-result.json', records)
labels = ['currency', 'profit_percent', 'profit_BTC', 'duration']
labels = ['currency', 'profit_percent', 'profit_BTC', 'duration', 'entry', 'exit']
return DataFrame.from_records(trades, columns=labels)
def start(self):
@ -287,8 +291,8 @@ class Backtesting(object):
)
# return date for data storage
temp = self.aggregate(data, results)
return DataFrame(data=temp[2][:-1], columns=temp[1])
self.aggregate(data, results)
return results
def setup_configuration(args: Namespace) -> Dict[str, Any]:

65
freqtrade/tests/aws/test_backtest.py
View File

@ -2,7 +2,7 @@ from base64 import urlsafe_b64encode
import pytest
import simplejson as json
from freqtrade.aws.backtesting_lambda import backtest
from freqtrade.aws.backtesting_lambda import backtest, cron
from freqtrade.aws.strategy import submit
@ -73,3 +73,66 @@ class MyFancyTestStrategy(IStrategy):
}
backtest({"body": json.dumps(request)}, {})
def test_cron(lambda_context):
""" test the scheduling to the queue"""
content = """# --- Do not remove these libs ---
from freqtrade.strategy.interface import IStrategy
from typing import Dict, List
from hyperopt import hp
from functools import reduce
from pandas import DataFrame
# --------------------------------
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
class MyFancyTestStrategy(IStrategy):
minimal_roi = {
"0": 0.5
}
stoploss = -0.2
ticker_interval = '5m'
def populate_indicators(self, dataframe: DataFrame) -> DataFrame:
macd = ta.MACD(dataframe)
dataframe['maShort'] = ta.EMA(dataframe, timeperiod=8)
dataframe['maMedium'] = ta.EMA(dataframe, timeperiod=21)
return dataframe
def populate_buy_trend(self, dataframe: DataFrame) -> DataFrame:
dataframe.loc[
(
qtpylib.crossed_above(dataframe['maShort'], dataframe['maMedium'])
),
'buy'] = 1
return dataframe
def populate_sell_trend(self, dataframe: DataFrame) -> DataFrame:
dataframe.loc[
(
qtpylib.crossed_above(dataframe['maMedium'], dataframe['maShort'])
),
'sell'] = 1
return dataframe
"""
request = {
"user": "GCU4LW2XXZW3A3FM2XZJTEJHNWHTWDKY2DIJLCZJ5ULVZ4K7LZ7D23TG",
"description": "simple test strategy",
"name": "MyFancyTestStrategy",
"content": urlsafe_b64encode(content.encode('utf-8')),
"public": False
}
# now we add an entry
submit({
"body": json.dumps(request)
}, {})
print("evaluating cron job")
cron({}, {})

2
freqtrade/tests/aws/test_strategy_lambda.py
View File

@ -99,6 +99,8 @@ class TestStrategy(IStrategy):
"public": True
}
print(json.dumps(request))
aws.submit({
"body": json.dumps(request)
}, {})

29
freqtrade/tests/conftest.py
View File

@ -609,33 +609,8 @@ def lambda_context():
client = session.client('sns')
dynamodb = boto3.resource('dynamodb')
os.environ["strategyTable"] = "StrategyTable"
dynamodb.create_table(
TableName=os.environ["strategyTable"],
KeySchema=[
{
'AttributeName': 'user',
'KeyType': 'HASH'
},
{
'AttributeName': 'name',
'KeyType': 'RANGE'
}
],
AttributeDefinitions=[
{
'AttributeName': 'user',
'AttributeType': 'S'
}, {
'AttributeName': 'name',
'AttributeType': 'S'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 1,
'WriteCapacityUnits': 1
}
)
os.environ["tradeTable"] = "TradeTable"
os.environ["topic"] = "UnitTestTopic"
import responses