Changes to work in Python 2.6+

Author: Siddharta Govindaraj

stock_alerter/__init__.py

@@ -0,0 +1,4 @@
+
+if __name__ == "__main__":
+    import doctest
+    doctest.testfile("readme.txt")

stock_alerter/__main__.py

@@ -0,0 +1,24 @@
+import unittest
+
+
+class AttribLoader(unittest.TestLoader):
+    def __init__(self, attrib):
+        self.attrib = attrib
+
+    def loadTestsFromModule(self, module, use_load_tests=False):
+        return super().loadTestsFromModule(module, use_load_tests=False)
+
+    def getTestCaseNames(self, testCaseClass):
+        test_names = super().getTestCaseNames(testCaseClass)
+        filtered_test_names = [test
+                               for test in test_names
+                               if hasattr(getattr(testCaseClass, test),
+                                          self.attrib)]
+        return filtered_test_names
+
+
+if __name__ == "__main__":
+    loader = AttribLoader("slow")
+    test_suite = loader.discover(".")
+    runner = unittest.TextTestRunner()
+    runner.run(test_suite)

stock_alerter/action.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import smtplib
 from email.mime.text import MIMEText
 
@@ -17,7 +18,7 @@ def __init__(self, to):
     def execute(self, content):
         message = MIMEText(content)
         message["Subject"] = "New Stock Alert"
-        message["From"] = self.from_email
+        message["From"] = "[email protected]"
         message["To"] = self.to_email
         smtp = smtplib.SMTP("email.stocks.com")
         try:

stock_alerter/alert.py

@@ -11,7 +11,7 @@ def connect(self, exchange):
         self.exchange = exchange
         dependent_stocks = self.rule.depends_on()
         for stock in dependent_stocks:
-            self.exchange[stock].updated.connect(self.check_rule)
+            exchange[stock].updated.connect(self.check_rule)
 
     def check_rule(self, stock):
         if self.rule.matches(self.exchange):

stock_alerter/legacy.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 from datetime import datetime
 
 from .stock import Stock
@@ -8,14 +9,14 @@ class FileReader:
     def __init__(self, filename):
         self.filename = filename
 
-    def parse_file(self):
+    def get_updates(self):
         updates = []
-        with open(self.filename, "r") as fp:
+        with open("updates.csv", "r") as fp:
             for line in fp.readlines():
                 symbol, timestamp, price = line.split(",")
                 updates.append((symbol,
-                                datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%f"),
-                                int(price)))
+                       datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%f"),
+                       int(price)))
         return updates
 
 
@@ -28,20 +29,25 @@ def __init__(self, autorun=True, reader=None, exchange=None):
             self.exchange = exchange
         rule_1 = PriceRule("GOOG", lambda stock: stock.price > 10)
         rule_2 = PriceRule("AAPL", lambda stock: stock.price > 5)
-        self.exchange["GOOG"].updated.connect(lambda stock: self.print_action(stock, rule_1))
-        self.exchange["AAPL"].updated.connect(lambda stock: self.print_action(stock, rule_2))
+        self.exchange["GOOG"].updated.connect(
+            lambda stock: self.print_action(stock, rule_1))
+        self.exchange["AAPL"].updated.connect(
+            lambda stock: self.print_action(stock, rule_2))
         if autorun:
             self.run()
 
     def print_action(self, stock, rule):
         print(stock.symbol, stock.price) \
             if rule.matches(self.exchange) else None
 
+    def run(self):
+        updates = self.parse_file()
+        self.do_updates(updates)
+
+    def parse_file(self):
+        return self.reader.get_updates()
+
     def do_updates(self, updates):
         for symbol, timestamp, price in updates:
             stock = self.exchange[symbol]
             stock.update(timestamp, price)
-
-    def run(self):
-        updates = self.reader.parse_file()
-        self.do_updates(updates)

stock_alerter/readme.txt

@@ -8,11 +8,10 @@ are going to be processed. A simple dictionary will do.
 
 >>> from stock_alerter.stock import Stock
 >>> exchange = {"GOOG": Stock("GOOG"), "AAPL": Stock("AAPL")}
->>> for key in sorted(exchange.keys()):
-...    print(key, str(exchange[key]))
-...    
-AAPL Stock("AAPL")
-GOOG Stock("GOOG")
+>>> for key in sorted(exchange.keys()): #doctest: -NORMALIZE_WHITESPACE
+...    print key, exchange[key]
+AAPL <stock_alerter.stock.Stock ... at 0x0...>
+GOOG <stock_alerter.stock.Stock ... at 0x0...>
 
 Next, we configure the reader. The reader is the source from where the
 stock updates are coming. The module provides two readers out of the

stock_alerter/rule.py

@@ -14,10 +14,30 @@ def matches(self, exchange):
         return self.condition(stock) if stock.price else False
 
     def depends_on(self):
-        return {self.symbol}
+        return set([self.symbol])
+
+
+class TrendRule:
+    """TrendRule is a rule that triggers when a stock price satisfies a
+    trend (for now the only trend supported is 3 increasing updates)"""
+
+    def __init__(self, symbol):
+        self.symbol = symbol
+
+    def matches(self, exchange):
+        try:
+            stock = exchange[self.symbol]
+        except KeyError:
+            return False
+        return stock.is_increasing_trend()
+
+    def depends_on(self):
+        return set([self.symbol])
 
 
 class AndRule:
+    """AndRule triggers when all its component rules are true"""
+
     def __init__(self, *args):
         self.rules = args
 

stock_alerter/stock.py

@@ -1,8 +1,7 @@
-from datetime import timedelta
 from enum import Enum
 
+from .timeseries import TimeSeries
 from .event import Event
-from .timeseries import TimeSeries, MovingAverage, NotEnoughDataException
 
 
 class StockSignal(Enum):
@@ -20,10 +19,6 @@ def __init__(self, symbol):
         self.history = TimeSeries()
         self.updated = Event()
 
-    def __str__(self):
-        class_name = type(self).__name__
-        return '{}("{}")'.format(class_name, self.symbol)
-
     @property
     def price(self):
         """Returns the current price of the Stock
@@ -33,18 +28,9 @@ def price(self):
         10
 
         The method will return the latest price by timestamp, so even if
-        updates are out of order, it will return the latest one
-
-        >>> stock = Stock("GOOG")
-        >>> stock.update(datetime(2011, 10, 3), 10)
-
-        Now, let us do an update with a date that is earlier than the previous
-        one
+        updates are out of order, it will return the latest one.
 
         >>> stock.update(datetime(2011, 10, 2), 5)
-
-        And the method still returns the latest price
-
         >>> stock.price
         10
 
@@ -73,6 +59,7 @@ def update(self, timestamp, price):
             ...
         ValueError: price should not be negative
         """
+
         if price < 0:
             raise ValueError("price should not be negative")
         self.history.update(timestamp, price)
@@ -86,26 +73,59 @@ def is_increasing_trend(self):
         >>> stock.is_increasing_trend()
         False
         """
+
         try:
-            return self.history[-3].value < self.history[-2].value < self.history[-1].value
+            return self.history[-3].value < \
+               self.history[-2].value < self.history[-1].value
         except IndexError:
             return False
 
-    def _is_crossover_below_to_above(self, on_date, ma, reference_ma):
-        prev_date = on_date - timedelta(1)
-        return (ma.value_on(prev_date) < reference_ma.value_on(prev_date)
-                and ma.value_on(on_date) > reference_ma.value_on(on_date))
+    def _is_crossover_below_to_above(self, prev_ma, prev_reference_ma,
+                                           current_ma, current_reference_ma):
+        return prev_ma < prev_reference_ma \
+            and current_ma > current_reference_ma
 
     def get_crossover_signal(self, on_date):
-        long_term_ma = MovingAverage(self.history, self.LONG_TERM_TIMESPAN)
-        short_term_ma = MovingAverage(self.history, self.SHORT_TERM_TIMESPAN)
-        try:
-            if self._is_crossover_below_to_above(on_date, short_term_ma, long_term_ma):
+        NUM_DAYS = self.LONG_TERM_TIMESPAN + 1
+        closing_price_list = self.history.get_closing_price_list(on_date,
+                                                                 NUM_DAYS)
+
+        if len(closing_price_list) < NUM_DAYS:
+            return StockSignal.neutral
+
+        long_term_series = closing_price_list[-self.LONG_TERM_TIMESPAN:]
+        prev_long_term_series = closing_price_list[-self.LONG_TERM_TIMESPAN-1:-1]
+        short_term_series = closing_price_list[-self.SHORT_TERM_TIMESPAN:]
+        prev_short_term_series = closing_price_list[-self.SHORT_TERM_TIMESPAN-1:-1]
+
+        long_term_ma = 1.0*sum([update.value
+                            for update in long_term_series])\
+                        /self.LONG_TERM_TIMESPAN
+        prev_long_term_ma = 1.0*sum([update.value
+                                 for update in prev_long_term_series])\
+                             /self.LONG_TERM_TIMESPAN
+        short_term_ma = 1.0*sum([update.value
+                             for update in short_term_series])\
+                        /self.SHORT_TERM_TIMESPAN
+        prev_short_term_ma = 1.0*sum([update.value
+                                  for update in prev_short_term_series])\
+                             /self.SHORT_TERM_TIMESPAN
+
+        if self._is_crossover_below_to_above(prev_short_term_ma,
+                                             prev_long_term_ma,
+                                             short_term_ma,
+                                             long_term_ma):
                     return StockSignal.buy
 
-            if self._is_crossover_below_to_above(on_date, long_term_ma, short_term_ma):
+        if self._is_crossover_below_to_above(prev_long_term_ma,
+                                             prev_short_term_ma,
+                                             long_term_ma,
+                                             short_term_ma):
                     return StockSignal.sell
-        except NotEnoughDataException:
-            return StockSignal.neutral
 
         return StockSignal.neutral
+
+
+if __name__ == "__main__":
+    import doctest
+    doctest.testmod()

stock_alerter/tests/test_action.py

@@ -1,8 +1,9 @@
+from __future__ import print_function
 import smtplib
-import unittest
-from unittest import mock
+import unittest2 as unittest
+import mock
 
-from ..action import PrintAction, EmailAction
+from ..action import EmailAction, PrintAction
 
 
 class MessageMatcher:
@@ -16,7 +17,18 @@ def __eq__(self, other):
             self.expected["Message"] == other._payload
 
 
-@mock.patch("builtins.print")
+class AlertMessageMatcher:
+    def __init__(self, expected):
+        self.expected = expected
+
+    def __eq__(self, other):
+        return "New Stock Alert" == other["Subject"] and \
+            self.expected["From"] == other["From"] and \
+            self.expected["To"] == other["To"] and \
+            self.expected["Message"] == other._payload
+
+
+@mock.patch("__builtin__.print")
 class PrintActionTest(unittest.TestCase):
     def test_executing_action_prints_message(self, mock_print):
         action = PrintAction()

stock_alerter/tests/test_alert.py

@@ -1,11 +1,11 @@
 import unittest
-from unittest import mock
 from datetime import datetime
+import mock
 
 from ..alert import Alert
 from ..rule import PriceRule
-from ..stock import Stock
 from ..event import Event
+from ..stock import Stock
 
 
 class AlertTest(unittest.TestCase):
@@ -16,7 +16,7 @@ def test_action_is_executed_when_rule_matches(self):
         exchange = {"GOOG": goog}
         rule = mock.MagicMock(spec=PriceRule)
         rule.matches.return_value = True
-        rule.depends_on.return_value = {"GOOG"}
+        rule.depends_on.return_value = set(["GOOG"])
         action = mock.MagicMock()
         alert = Alert("sample alert", rule, action)
         alert.connect(exchange)
@@ -41,7 +41,7 @@ def test_action_fires_when_rule_matches(self):
         main_mock = mock.MagicMock()
         rule = main_mock.rule
         rule.matches.return_value = True
-        rule.depends_on.return_value = {"GOOG"}
+        rule.depends_on.return_value = set(["GOOG"])
         action = main_mock.action
         alert = Alert("sample alert", rule, action)
         alert.connect(exchange)