Repository: liampauling/flumine
Pull request #147 (2020-03-06 08:32:36)

context added to strategy
bflw upgraded
docstrings added
setup fix


Patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 49c8358..2a62bec 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -3,6 +3,19 @@
 Release History
 ---------------
 
+1.1.0 (2020-03-09)
++++++++++++++++++++
+
+**Improvements**
+
+- `context` added to strategy
+- `.start` / `.add` refactored to make more sense
+- HistoricalStream added and working but will change in the future to not use threads (example added)
+
+**Libraries**
+
+- betfairlightweight upgraded to 2.1.0
+
 1.0.0 (2020-03-02)
 +++++++++++++++++++
 
diff --git a/docs/advanced.md b/docs/advanced.md
index f426d43..9a1ecc9 100644
--- a/docs/advanced.md
+++ b/docs/advanced.md
@@ -48,6 +48,12 @@ flumine.add_worker(
 
 ### Strategies
 
+market_filter
+
+market_data_filter
+
+context
+
 
 ### Flumine
 
diff --git a/docs/index.md b/docs/index.md
index 3072fbd..f6a2b83 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -28,7 +28,7 @@ Betfair trading framework with a focus on:
 
 Support for market and custom streaming data (order, score and custom polling data in development)
 
-[Join slack group](https://betfairlightweight.herokuapp.com)
+[join slack group](https://betfairlightweight.herokuapp.com)
 
 Currently tested on Python 3.5, 3.6, 3.7 and 3.8.
 
@@ -83,7 +83,7 @@ strategy = ExampleStrategy(
     )
 )
 
-flumine.add_strategy(strategy)
+framework.add_strategy(strategy)
 ```
 
 Run framework:
diff --git a/examples/historical.py b/examples/historical.py
new file mode 100644
index 0000000..2f0b10a
--- /dev/null
+++ b/examples/historical.py
@@ -0,0 +1,39 @@
+import time
+import logging
+import betfairlightweight
+from pythonjsonlogger import jsonlogger
+
+from flumine import Flumine, BaseStrategy
+from flumine.streams.historicalstream import HistoricalStream
+
+logger = logging.getLogger()
+
+custom_format = "%(asctime) %(levelname) %(message)"
+log_handler = logging.StreamHandler()
+formatter = jsonlogger.JsonFormatter(custom_format)
+formatter.converter = time.gmtime
+log_handler.setFormatter(formatter)
+logger.addHandler(log_handler)
+logger.setLevel(logging.INFO)
+
+
+class Ex(BaseStrategy):
+    def check_market_book(self, market_book):
+        return True
+
+    def process_market_book(self, market_book):
+        print(market_book, market_book.total_matched)
+
+
+trading = betfairlightweight.APIClient("username")
+
+framework = Flumine(trading=trading, interactive=True)
+
+strategy = Ex(
+    market_filter={"markets": ["/tmp/marketdata/1.167775532"]},
+    stream_class=HistoricalStream,
+)
+
+framework.add_strategy(strategy)
+
+framework.run()
diff --git a/examples/exampleone.py b/examples/marketrecorder.py
similarity index 87%
rename from examples/exampleone.py
rename to examples/marketrecorder.py
index 4cbfabb..a5a5daf 100644
--- a/examples/exampleone.py
+++ b/examples/marketrecorder.py
@@ -31,9 +31,12 @@ strategy = S3MarketRecorder(
         # event_ids=[29671376]
     ),
     stream_class=DataStream,
-    local_dir="/tmp/flu",
-    bucket="fluminetest",
-    force_update=False,
+    context={
+        "local_dir": "/tmp",
+        "bucket": "fluminetest",
+        "force_update": False,
+        "remove_file": True,
+    },
 )
 
 framework.add_strategy(strategy)
diff --git a/examples/strategies/marketrecorder.py b/examples/strategies/marketrecorder.py
index 731a35c..fdc7b67 100644
--- a/examples/strategies/marketrecorder.py
+++ b/examples/strategies/marketrecorder.py
@@ -17,19 +17,21 @@ class MarketRecorder(BaseStrategy):
 
     MARKET_ID_LOOKUP = "id"
 
-    def __init__(self, local_dir: str, *args, **kwargs):
-        self._market_expiration = kwargs.pop("market_expiration", 3600)  # seconds
-        self._remove_file = kwargs.pop("remove_file", True)  # remove txt file after zip
-        self._force_update = kwargs.pop(
+    def __init__(self, *args, **kwargs):
+        BaseStrategy.__init__(self, *args, **kwargs)
+        self._market_expiration = self.context.get("market_expiration", 3600)  # seconds
+        self._remove_file = self.context.get(
+            "remove_file", True
+        )  # remove txt file after zip
+        self._force_update = self.context.get(
             "force_update", True
         )  # update after initial closure
-        BaseStrategy.__init__(self, *args, **kwargs)
-        self.local_dir = local_dir
+        self.local_dir = self.context.get("local_dir", "/tmp")
         self.recorder_id = create_short_uuid()
         self._loaded_markets = []  # list of marketIds
 
-    def start(self) -> None:
-        logger.info("Starting strategy %s with id %s" % (self.name, self.recorder_id))
+    def add(self) -> None:
+        logger.info("Adding strategy %s with id %s" % (self.name, self.recorder_id))
         # check local dir
         if not os.path.isdir(self.local_dir):
             raise OSError("File dir %s does not exist" % self.local_dir)
@@ -141,15 +143,15 @@ class MarketRecorder(BaseStrategy):
 
 class S3MarketRecorder(MarketRecorder):
     def __init__(self, *args, **kwargs):
-        self._bucket = kwargs.pop("bucket")
-        self._data_type = kwargs.pop("data_type", "marketdata")
         MarketRecorder.__init__(self, *args, **kwargs)
+        self._bucket = self.context["bucket"]
+        self._data_type = self.context.get("data_type", "marketdata")
         self.s3 = boto3.client("s3")
         transfer_config = TransferConfig(use_threads=False)
         self.transfer = S3Transfer(self.s3, config=transfer_config)
 
-    def start(self) -> None:
-        super().start()
+    def add(self) -> None:
+        super().add()
         self.s3.head_bucket(Bucket=self._bucket)  # validate bucket/access
 
     def _load(self, zip_file_dir: str, market_definition: dict) -> None:
diff --git a/flumine/__version__.py b/flumine/__version__.py
index d3d1d45..26e0b74 100644
--- a/flumine/__version__.py
+++ b/flumine/__version__.py
@@ -1,6 +1,6 @@
 __title__ = "flumine"
 __description__ = "Betfair trading framework"
 __url__ = "https://github.com/liampauling/flumine"
-__version__ = "1.0.0"
+__version__ = "1.1.0"
 __author__ = "Liam Pauling"
 __license__ = "MIT"
diff --git a/flumine/baseflumine.py b/flumine/baseflumine.py
index 8aa0952..4df9fc8 100644
--- a/flumine/baseflumine.py
+++ b/flumine/baseflumine.py
@@ -13,6 +13,12 @@ logger = logging.getLogger(__name__)
 
 class BaseFlumine:
     def __init__(self, trading: APIClient, interactive: bool = False):
+        """
+        Base framework class
+
+        :param trading: betfairlightweight client instance
+        :param interactive: Interactive login for client
+        """
         self.trading = trading
         self.interactive = interactive
         self._running = False
@@ -93,6 +99,8 @@ class BaseFlumine:
         # start workers
         for w in self._workers:
             w.start()
+        # start strategies
+        self.strategies.start()
         # start streams
         self.streams.start()
 
diff --git a/flumine/strategy/strategy.py b/flumine/strategy/strategy.py
index 7e85415..a9c0465 100644
--- a/flumine/strategy/strategy.py
+++ b/flumine/strategy/strategy.py
@@ -23,7 +23,11 @@ class Strategies:
 
     def __call__(self, strategy):
         self._strategies.append(strategy)
-        strategy.start()
+        strategy.add()
+
+    def start(self):
+        for s in self:
+            s.start()
 
     def __iter__(self):
         return iter(self._strategies)
@@ -41,13 +45,26 @@ class BaseStrategy:
         conflate_ms: int = None,
         stream_class: Type[BaseStream] = MarketStream,
         name: str = None,
+        context: dict = None,
     ):
+        """
+        Processes data from streams.
+
+        :param market_filter: Streaming market filter
+        :param market_data_filter: Streaming market data filter
+        :param streaming_timeout: Streaming timeout, will call snap() on cache
+        :param conflate_ms: Streaming conflation
+        :param stream_class: Can be Market or Data
+        :param name: Strategy name
+        :param context: Dictionary holding additional vars
+        """
         self.market_filter = market_filter
         self.market_data_filter = market_data_filter or DEFAULT_MARKET_DATA_FILTER
         self.streaming_timeout = streaming_timeout
         self.conflate_ms = conflate_ms
         self.stream_class = stream_class
         self._name = name
+        self.context = context
 
         self.streams = []  # list of streams strategy is subscribed
 
@@ -59,8 +76,13 @@ class BaseStrategy:
         else:
             return False
 
+    def add(self) -> None:
+        # called when strategy is added to framework
+        return
+
     def start(self) -> None:
-        # called when flumine starts e.g. subscribe to extra streams
+        # called when flumine starts but before streams start
+        # e.g. subscribe to extra streams
         return
 
     def check_market_book(self, market_book: MarketBook) -> bool:
@@ -90,6 +112,18 @@ class BaseStrategy:
     def stream_ids(self) -> list:
         return [stream.stream_id for stream in self.streams]
 
+    @property
+    def info(self):
+        return {
+            "name": self.name,
+            "market_filter": self.market_filter,
+            "market_data_filter": self.market_data_filter,
+            "streaming_timeout": self.streaming_timeout,
+            "conflate_ms": self.conflate_ms,
+            "stream_ids": self.stream_ids,
+            "context": self.context,
+        }
+
     @property
     def name(self):
         return self._name or self.__class__.__name__
diff --git a/flumine/streams/basestream.py b/flumine/streams/basestream.py
index e008b69..ebd0cff 100644
--- a/flumine/streams/basestream.py
+++ b/flumine/streams/basestream.py
@@ -14,6 +14,7 @@ logger = logging.getLogger(__name__)
 class BaseStream(threading.Thread):
 
     LISTENER = StreamListener
+    MAX_LATENCY = 0.5
 
     def __init__(
         self,
@@ -33,7 +34,9 @@ class BaseStream(threading.Thread):
         self.conflate_ms = conflate_ms
         self._stream = None
         self._output_queue = queue.Queue()
-        self._listener = self.LISTENER(output_queue=self._output_queue)
+        self._listener = self.LISTENER(
+            output_queue=self._output_queue, max_latency=self.MAX_LATENCY
+        )
         self._output_thread = threading.Thread(
             name="{0}_output_thread".format(self._name),
             target=self.handle_output,
diff --git a/flumine/streams/historicalstream.py b/flumine/streams/historicalstream.py
new file mode 100644
index 0000000..b8d1ca8
--- /dev/null
+++ b/flumine/streams/historicalstream.py
@@ -0,0 +1,56 @@
+import queue
+import logging
+from betfairlightweight import StreamListener
+from betfairlightweight import BetfairError
+from tenacity import retry, wait_exponential
+
+from .basestream import BaseStream
+from ..event.event import MarketBookEvent
+
+logger = logging.getLogger(__name__)
+
+
+class HistoricalStream(BaseStream):
+
+    MAX_LATENCY = 1e100
+
+    @retry(wait=wait_exponential(multiplier=1, min=2, max=20))
+    def run(self) -> None:
+        logger.info("Starting HistoricalStream")
+        self.stream_id = 0  # default for historical data
+
+        if not self._output_thread.is_alive():
+            logger.info("Starting output_thread {0}".format(self._output_thread))
+            self._output_thread.start()
+
+        for market in self.market_filter["markets"]:
+            self._stream = self.trading.streaming.create_historical_stream(
+                directory=market, listener=self._listener
+            )
+            try:
+                self._stream.start()
+            except BetfairError:
+                logger.error("HistoricalStream run error", exc_info=True)
+                raise
+            except Exception:
+                logger.critical("HistoricalStream run error", exc_info=True)
+                raise
+        logger.info("Stopped HistoricalStream {0}".format(self.stream_id))
+
+    def handle_output(self) -> None:
+        """Handles output from stream.
+        """
+        while self.is_alive():
+            try:
+                market_books = self._output_queue.get(
+                    block=True, timeout=self.streaming_timeout
+                )
+            except queue.Empty:
+                market_books = self._listener.snap(
+                    market_ids=self.flumine.live_markets.open_markets()
+                )
+            self.flumine.handler_queue.put(MarketBookEvent(market_books))
+
+        logger.info(
+            "Stopped output_thread (HistoricalStream {0})".format(self.stream_id)
+        )
diff --git a/flumine/streams/marketstream.py b/flumine/streams/marketstream.py
index 391283c..339d1ad 100644
--- a/flumine/streams/marketstream.py
+++ b/flumine/streams/marketstream.py
@@ -11,9 +11,6 @@ logger = logging.getLogger(__name__)
 
 
 class MarketStream(BaseStream):
-
-    LISTENER = StreamListener
-
     @retry(wait=wait_exponential(multiplier=1, min=2, max=20))
     def run(self) -> None:
         logger.info("Starting MarketStream")
diff --git a/flumine/streams/streams.py b/flumine/streams/streams.py
index 0cec735..2e65dd2 100644
--- a/flumine/streams/streams.py
+++ b/flumine/streams/streams.py
@@ -4,6 +4,7 @@ from typing import Union
 from ..strategy.strategy import BaseStrategy
 from .marketstream import MarketStream
 from .datastream import DataStream
+from .historicalstream import HistoricalStream
 
 logger = logging.getLogger(__name__)
 
@@ -18,7 +19,9 @@ class Streams:
         stream = self.add_stream(strategy)
         strategy.streams.append(stream)
 
-    def add_stream(self, strategy: BaseStrategy) -> Union[MarketStream, DataStream]:
+    def add_stream(
+        self, strategy: BaseStrategy
+    ) -> Union[MarketStream, DataStream, HistoricalStream]:
         for stream in self:  # check if market stream already exists
             if (
                 isinstance(stream, strategy.stream_class)
diff --git a/requirements.txt b/requirements.txt
index 16247c4..8a04d02 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,3 @@
-betfairlightweight>=2.0.1
+betfairlightweight==2.1.0
 tenacity==5.0.3
 python-json-logger==0.1.11
diff --git a/setup.py b/setup.py
index 1632d51..0c6296e 100644
--- a/setup.py
+++ b/setup.py
@@ -13,7 +13,7 @@ with open(os.path.join(here, "flumine", "__version__.py"), "r") as f:
 setup(
     name=about["__title__"],
     version=about["__version__"],
-    packages=find_packages(exclude=["examples"]),
+    packages=find_packages(exclude=["examples", "examples.strategies"]),
     package_dir={"flumine": "flumine"},
     install_requires=INSTALL_REQUIRES,
     url=about["__url__"],


Test patch:
diff --git a/tests/test_strategy.py b/tests/test_strategy.py
index 34769b1..c91a1d3 100644
--- a/tests/test_strategy.py
+++ b/tests/test_strategy.py
@@ -15,6 +15,12 @@ class StrategiesTest(unittest.TestCase):
         mock_strategy = mock.Mock()
         self.strategies(mock_strategy)
         self.assertEqual(self.strategies._strategies, [mock_strategy])
+        mock_strategy.add.assert_called_with()
+
+    def test_start(self):
+        mock_strategy = mock.Mock()
+        self.strategies._strategies.append(mock_strategy)
+        self.strategies.start()
         mock_strategy.start.assert_called_with()
 
     def test_iter(self):
@@ -38,6 +44,7 @@ class BaseStrategyTest(unittest.TestCase):
             conflate_ms=self.conflate_ms,
             stream_class=strategy.MarketStream,
             name="test",
+            context={"trigger": 0.123},
         )
 
     def test_init(self):
@@ -47,6 +54,7 @@ class BaseStrategyTest(unittest.TestCase):
         self.assertEqual(self.strategy.conflate_ms, self.conflate_ms)
         self.assertEqual(self.strategy.stream_class, strategy.MarketStream)
         self.assertEqual(self.strategy._name, "test")
+        self.assertEqual(self.strategy.context, {"trigger": 0.123})
 
     def test_check_market_no_subscribed(self):
         mock_market_book = mock.Mock()
@@ -86,6 +94,9 @@ class BaseStrategyTest(unittest.TestCase):
         self.assertTrue(self.strategy.check_market(mock_market_book))
         mock_check_market_book.assert_called_with(mock_market_book)
 
+    def test_add(self):
+        self.strategy.add()
+
     def test_start(self):
         self.strategy.start()
 
@@ -113,6 +124,20 @@ class BaseStrategyTest(unittest.TestCase):
         self.strategy.streams = [mock_stream]
         self.assertEqual(self.strategy.stream_ids, [321])
 
+    def test_info(self):
+        self.assertEqual(
+            self.strategy.info,
+            {
+                "conflate_ms": self.conflate_ms,
+                "market_data_filter": self.mock_market_data_filter,
+                "market_filter": self.mock_market_filter,
+                "name": "test",
+                "stream_ids": [],
+                "streaming_timeout": self.streaming_timeout,
+                "context": {"trigger": 0.123},
+            },
+        )
+
     def test_name(self):
         self.assertEqual(self.strategy.name, "test")
 
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 2ac1228..59d6ff8 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -114,6 +114,7 @@ class TestBaseStream(unittest.TestCase):
         self.assertEqual(self.stream.streaming_timeout, 0.01)
         self.assertEqual(self.stream.conflate_ms, 100)
         self.assertIsNone(self.stream._stream)
+        self.assertEqual(self.stream.MAX_LATENCY, 0.5)
 
     def test_run(self):
         with self.assertRaises(NotImplementedError):
@@ -251,3 +252,27 @@ class TestDataStream(unittest.TestCase):
         mock_on_process.assert_called_with(
             [mock_listener.stream_unique_id, 123, race_updates]
         )
+
+
+class TestHistoricalStream(unittest.TestCase):
+    def setUp(self) -> None:
+        self.mock_flumine = mock.Mock()
+        self.stream = streams.HistoricalStream(
+            self.mock_flumine, 123, {"test": "me"}, {"please": "now"}, 0.01, 100
+        )
+
+    def test_init(self):
+        self.assertEqual(self.stream.flumine, self.mock_flumine)
+        self.assertEqual(self.stream.stream_id, 123)
+        self.assertEqual(self.stream.market_filter, {"test": "me"})
+        self.assertEqual(self.stream.market_data_filter, {"please": "now"})
+        self.assertEqual(self.stream.streaming_timeout, 0.01)
+        self.assertEqual(self.stream.conflate_ms, 100)
+        self.assertIsNone(self.stream._stream)
+        self.assertEqual(self.stream.MAX_LATENCY, 1e100)
+
+    # def test_run(self):
+    #     pass
+    #
+    # def test_handle_output(self):
+    #     pass