configuration/kubernetes/deployment.yaml

@@ -119,10 +119,10 @@ spec:
         imagePullPolicy: Always
         resources:
           requests:
-            cpu: 100m
+            cpu: 64m
             memory: 64Mi
           limits:
-            cpu: 500m
+            cpu: 150m
             memory: 256Mi
         securityContext:
           capabilities:

src/tweets/collector.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-import os, sys, json, uuid
+import os, sys, json
 
 from datetime import datetime, timedelta
 from time import sleep, time
@@ -17,7 +17,6 @@ from src.utils.spamFilter import callSpamFilter
 from src.utils.sentimentAnalyser import callSentimentAnalyser
 from src.utils.activemqConnect import activeMQSender
 from src.utils.jsonLogger import log
-from src.utils.whitelistedWords import filterOutTweetsWithNoneWhitelistedWords
 
 from http.client import IncompleteRead
 from urllib3.exceptions import ProtocolError
@@ -34,20 +33,20 @@ class keys():
         self.access_token = os.getenv("ACCESS_TOKEN")
         self.access_secret = os.getenv("ACCESS_SECRET")
 
-def sendToArtemis(syncId, pos, neu, neg, compound, type):
+def sendToArtemis(pos, neu, neg, compound, type):
 
     timestamp = datetime.now() + timedelta(hours=1)
     strippedTimestamp = timestamp.replace(minute=0, second=0, microsecond=0)
     timestamp = strippedTimestamp.strftime('%Y-%m-%dT%H:%M:%S')
 
-    message = { "timestamp" : timestamp, "syncId": str(syncId), "pos" : pos, "neu" : neu, "neg" : neg, "compound" : compound, "type": type }
+    message = { "timestamp" : timestamp, "pos" : pos, "neu" : neu, "neg" : neg, "compound" : compound, "type": type }
 
     messageJson = json.dumps(message, indent = 4)
 
-    log("Sending message to TweetSave queue for SyncId [{}]".format(syncId), 'INFO')
+    log("Sending message to TweetSave queue", 'INFO')
     log("Message: {}".format(message), 'INFO')
 
-    activeMQSender(messageJson, syncId)
+    activeMQSender(messageJson)
 
 class Streamer():
 
@@ -97,23 +96,18 @@ class Listener(StreamListener):
                 if 'extended_tweet' in data['retweeted_status']:
                     #if tweet is over the 140 word limit
                     text = data['retweeted_status']['extended_tweet']['full_text']
-                    text = filterOutTweetsWithNoneWhitelistedWords(text)
                     dumpStack.append({'type': self.hashtag, 'tweet': text})
                 else:
                     text = data['retweeted_status']['text']
-                    text = filterOutTweetsWithNoneWhitelistedWords(text)
                     dumpStack.append({'type': self.hashtag, 'tweet': text})
             else:
                 # Else if a normal Tweeet
                 if 'extended_tweet' in data:
                     # If tweet is over 140 word limit
                     text = data['extended_tweet']['full_text']
-                    text = filterOutTweetsWithNoneWhitelistedWords(text)
                     dumpStack.append({'type': self.hashtag, 'tweet': text})
 
-def processTweet(syncId):
+def processTweet():
-
-    log(len(dumpStack), 'INFO')
 
     processStack = dumpStack.copy()
     dumpStack.clear()
@@ -123,7 +117,7 @@ def processTweet(syncId):
     if len(processStack) != 0:
         for tweet in processStack:
 
-            removedLines = fixLines(tweet["tweet"])
+            removedLines = fixLines(str(tweet))
 
             removedSpecialChars = cleanTweet(removedLines)
             removedSpacing = removeSpacing(removedSpecialChars[0])
@@ -141,12 +135,12 @@ def processTweet(syncId):
 
                     cleanedTweet = tweetText + ' ' + removedSpecialChars[1]
 
-                    if callSpamFilter(cleanedTweet, syncId) != 'spam':
+                    if callSpamFilter(cleanedTweet) != 'spam':
 
-                        pos, neu, neg, compound = callSentimentAnalyser(cleanedTweet, syncId)
+                        pos, neu, neg, compound = callSentimentAnalyser(cleanedTweet)
 
-                        if compound != 0.0 and neu <= 0.6:
+                        if compound != 0.0 or neu <= 0.8:
-                            hourTweet = {'pos': pos, 'neu': neu, 'neg': neg, 'compound': compound, 'type': tweet["type"]}
+                            hourTweet = {'pos': 1, 'neu': 1, 'neg': 1, 'compound': 1, 'type': ""}
 
                             hourStack.append(hourTweet)
         processStack.clear()
@@ -182,9 +176,7 @@ def createHourJob():
     global timeF
     timeF = timeFunction()
 
-    syncId = uuid.uuid4()
+    processTweet()
-
-    processTweet(syncId)
 
     processStack = hourStack.copy()
     hourStack.clear()
@@ -193,7 +185,7 @@ def createHourJob():
 
     if len(processStack) != 0:
         log("Process stack size is :: [{}]".format(len(processStack)), 'INFO')
-        for item in processStack:
+        for item in hourStack:
             ovPos = ovPos + item['pos']
             ovNeu = ovNeu + item['neu']
             ovNeg = ovNeg + item['neg']
@@ -210,7 +202,7 @@ def createHourJob():
 
         processStack.clear()
 
-        sendToArtemis(syncId, pos, neu, neg, compound, type)
+        sendToArtemis(pos, neu, neg, compound, type)
     else:
         log("Stack is empty", 'WARN')
 
@@ -224,7 +216,6 @@ def collectorMain(hashtag):
     for i in range(len(hashtag)):
         Thread(target=collector, args=[hashtag[i]]).start()
 
-    sleep(2)
     createHourJob()
 
     while True:

src/utils/activemqConnect.py

@@ -17,20 +17,14 @@ class keys():
     def returnKeys(self):
         return self.addr, self.port, self.amqU, self.amqP
 
-def activeMQSender(message, syncId):
+def activeMQSender(message):
     addr, port, mqUser, mqPass = keys().returnKeys()
 
     log("Attempting Connection to Artemis...", 'INFO')
     con = stomp.Connection([(addr, port)], auto_content_length=False)
     con.connect( mqUser, mqPass, wait=True)
 
-    con.send("TweetSave",
+    con.send("TweetSave", message, content_type="application/json", headers={"Content-Type":"application/json"})
-             message,
-             content_type="application/json",
-             headers={
-                 "Content-Type":"application/json",
-                 "X-CRYPTO-Sync-ID":syncId
-             })
 
     con.disconnect()
 

src/utils/jsonLogger.py

@@ -28,13 +28,13 @@ def setup_logging(log_level='INFO'):
     logHandler.setFormatter(formatter)
     logger.addHandler(logHandler)
 
-def log(message, level, syncId=""):
+def log(message, level):
     logger = logging.getLogger(__name__)
     if level == 'INFO':
-        logger.info(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.info(message)
     elif level == 'WARN':
-        logger.warn(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.warn(message)
     elif level == 'ERR':
-        logger.error(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.error(message)
     elif level == 'DEBUG':
-        logger.debug(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.debug(message)

src/utils/sentimentAnalyser.py

@@ -9,14 +9,11 @@ class keys():
     def __init__(self):
         self.sentiment_analyser_uri = os.getenv("SENTIMENT_URL")
 
-def callSentimentAnalyser(tweet, syncId):
+def callSentimentAnalyser(tweet):
-    headers = {
+    # log("Calling Sentiment Analyser for [{}]".format(tweet), 'INFO')
-        "content-type":"text",
-        "X-CRYPTO-Sync-ID" : str(syncId)
-    }
     try:
         uri = keys().sentiment_analyser_uri + "/sentiment?tweet="+tweet
-        response = requests.request("GET", url=uri, headers=headers)
+        response = requests.request("GET", uri)
 
         response = json.loads(response.text)
 
@@ -24,5 +21,5 @@ def callSentimentAnalyser(tweet, syncId):
 
         return scores["pos"], scores["neu"], scores["neg"], scores["compound"]
     except:
-        log("Could not call Sentiment Analyser Service with syncId of [{}]".format(syncId), 'ERR', syncId)
+        log("Could not call Sentiment Analyser Service", 'ERR')
         return 0, 0, 0, 0

src/utils/spamFilter.py

@@ -9,18 +9,16 @@ class keys():
     def __init__(self):
         self.spamFilter_uri = os.getenv("FILTER_URL")
 
-def callSpamFilter(tweet, syncId):
+def callSpamFilter(tweet):
-    headers = {
-        "content-type":"text",
-        "X-CRYPTO-Sync-ID" : str(syncId)
-    }
     try:
         uri = keys().spamFilter_uri + "/predict?tweet="+tweet
-        response = requests.request("GET", url=uri, headers=headers)
+        response = requests.request("GET", uri)
 
         response = json.loads(response.text)
 
+        # log("Spam Filter result for [{}] is [{}]".format(tweet, response["result"]), 'INFO')
+
         return response["result"]
     except:
-        log("Could not call spam filter service with syncId of [{}]".format(syncId), 'ERR', syncId)
+        log("Could not call spam filter service", 'ERR')
         return ""

src/utils/whitelistedWords.py

@@ -1,152 +0,0 @@
-#!/usr/bin/env python
-
-whitelist = [
-    "bull",
-    "bear",
-    "bullish",
-    "bearish",
-    "up",
-    "down",
-    "high",
-    "low",
-    "higher",
-    "lower",
-    "absconded",
-    "maximalists",
-    "regulate",
-    "infamous",
-    "tradehigher",
-    "tradelower",
-    "revival",
-    "centralized",
-    "decentralized",
-    "centralised",
-    "decentralised",
-    "decentralization",
-    "decentralisation",
-    "centralization",
-    "centralisation",
-    "bans",
-    "hodl",
-    "ambiguity",
-    "revolutionize",
-    "revolutionise",
-    "consolidation",
-    "shorts",
-    "longs",
-    "long",
-    "short",
-    "shorting",
-    "grow",
-    "volatile",
-    "rally",
-    "rallying",
-    "noob",
-    "noobs",
-    "innovation",
-    "bottom",
-    "top",
-    "topped",
-    "bottomed",
-    "upwards",
-    "downwards",
-    "invest",
-    "raging",
-    "rocketing",
-    "swing",
-    "swinging",
-    "stake",
-    "whale",
-    "whales",
-    "lull",
-    "moon",
-    "choppy",
-    "buy",
-    "buying",
-    "sell",
-    "selling",
-    "startselling",
-    "stopselling",
-    "startbuying",
-    "stopbuying",
-    "bitcoin",
-    "btc",
-    "eth",
-    "xmr",
-    "xrp",
-    "ripple",
-    "block",
-    "reward",
-    "airdrop",
-    "drop",
-    "raise",
-    "stack",
-    "stake",
-    "invest",
-    "pull",
-    "push",
-    "token",
-    "sale",
-    "unhappy",
-    "happy",
-    "expert",
-    "novice"
-    "passed",
-    "mark",
-    "decline",
-    "incline",
-    "fees",
-    "crypto",
-    "wallet",
-    "price",
-    "history",
-    "reached",
-    "upward",
-    "downward",
-    "trading",
-    "mining",
-    "defi",
-    "finance",
-    "blockchain",
-    "interest",
-    "alt",
-    "alts",
-    "fiat",
-    "fiat",
-    "currency",
-    "currencies",
-    "wealth",
-    "hype",
-    "hyped",
-    "achievement",
-    "platform",
-    "incremental",
-    "increment",
-    "decrement",
-    "decremental",
-    "success",
-    "loss",
-    "win",
-    "lose",
-    "worth",
-    "strongest",
-    "weakest",
-    "strong",
-    "weak",
-    "trade",
-    "popping",
-    "sucking",
-    "shard",
-    "sharding",
-    "industry",
-    "powerful",
-    "better",
-    "worse"
-]
-
-def filterOutTweetsWithNoneWhitelistedWords(text):
-    if any(x in text for x in whitelist):
-        return text
-    else:
-        # log("Tweet [{}] did not contain any keywords for it to be considered crypto related".format(text), 'WARN')
-        return ""