Added live trading support

Author: ypogorelova

livetrading/broker.py

@@ -23,14 +23,16 @@ def __init__(
         self._cached_pairs: Dict[Pair] = {}
 
     def subscribe_to_ticker_events(
-            self, pair: Pair, event_handler
+            self, pair: Pair, interval: str, event_handler
     ):
         """Registers a callable that will be called every ticker.
 
+        :param bar_duration: The bar duration. One of 1s, 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d, 3d, 1w, 1M.
         :param pair: The trading pair.
         :param event_handler: A callable that receives an TickerEvent.
         """
-        event_source = TickersEventSource(pair, self.ws_cli)
+
+        event_source = TickersEventSource(pair, interval, self.ws_cli)
         channel = "ticker"
 
         self._subscribe_to_ws_channel_events(
@@ -78,7 +80,6 @@ def _subscribe_to_ws_channel_events(
     ):
         # Set the event source for the channel.
         self.ws_cli.set_channel_event_source(channel, event_source)
-        # self.ws_cli.subscribe_to_channels()
 
         # Subscribe the event handler to the event source.
         self.dispatcher.subscribe(event_source, event_handler)

livetrading/env

@@ -1,5 +1,5 @@
 {
 "ws_url": "wss://ws-feed.exchange.coinbase.com",
 "api_url": "https://api.exchange.coinbase.com/",
-"ws_timeout": 5}
+"ws_timeout": 5
 }

livetrading/event.py

@@ -7,6 +7,26 @@
 from typing import Optional
 
 
+intervals = {
+            "1s": 1,
+            "1m": 60,
+            "3m": 3 * 60,
+            "5m": 5 * 60,
+            "15m": 15 * 60,
+            "30m": 30 * 60,
+            "1h": 3600,
+            "2h": 2 * 3600,
+            "4h": 4 * 3600,
+            "6h": 6 * 3600,
+            "8h": 8 * 3600,
+            "12h": 12 * 3600,
+            "1d": 86400,
+            "3d": 3 * 86400,
+            "1w": 7 * 86400,
+            "1M": 31 * 86400
+}
+
+
 @dataclasses.dataclass
 class Bar:
     """A Bar, aka candlestick, is the summary of the trading activity in a given period.
@@ -150,14 +170,16 @@ class TickersEventSource(ChannelEventSource):
 
     :param pair: The trading pair.
     """
-    def __init__(self, pair: Pair, producer: EventProducer):
+    def __init__(self, pair: Pair, when: datetime, producer: EventProducer):
         super().__init__(producer=producer)
         self.pair: Pair = pair
+        self.when = intervals.get(when)
 
     def push_to_queue(self, message: dict):
         timestamp = message["time"]
+        dt = isoparse(timestamp) + datetime.timedelta(seconds=self.when)
         self.events.append(TickerEvent(
-            isoparse(timestamp),
+            dt,
             Ticker(self.pair, message)))
 
 
@@ -199,7 +221,7 @@ class BarEvent(Event):
     def __init__(self, when, bar: Bar):
         super().__init__(when)
 
-        self.bar = bar
+        self.data = bar
 
 
 class TickerEvent(Event):
@@ -211,4 +233,4 @@ class TickerEvent(Event):
     def __init__(self, when, ticker: Ticker):
         super().__init__(when)
 
-        self.ticker = ticker
+        self.data = ticker

livetrading/executor.py

@@ -1,9 +1,11 @@
 import time
 import datetime
 import logging
-
+from functools import partial
 from typing import Any, Dict, List, Set, Optional
 
+from backtesting import Backtest
+from .converter import ohlcv_to_dataframe
 from .event import Event, EventSource, EventProducer
 
 logger = logging.getLogger(__name__)
@@ -13,17 +15,29 @@ class EventDispatcher:
     """Responsible for connecting event sources to event handlers and dispatching events
     in the right order.
     """
-    def __init__(self):
+    def __init__(self, strategy):
         self._event_handlers: Dict[EventSource, List[Any]] = {}
         self._prefetched_events: Dict[EventSource, Optional[Event]] = {}
         self._prev_events: Dict[EventSource, datetime.datetime] = {}
         self._producers: Set[EventProducer] = set()
         self._running = False
         self._stopped = False
         self._current_event_dt = None
+        self.strategy = strategy
+        self.backtesting = None
 
     def set_strategy(self, strategy):
-        self.strategy = strategy
+        self._strategy = strategy
+
+    def set_backtesting_partial(self, cash: float = 10_000,
+                                commission: float = .0,
+                                margin: float = 1.,
+                                trade_on_close=False,
+                                hedging=False,
+                                exclusive_orders=False):
+        self.backtesting = partial(Backtest, strategy=self.strategy, cash=cash, commission=commission,
+                                   margin=margin, trade_on_close=trade_on_close,
+                                   hedging=hedging, exclusive_orders=exclusive_orders)
 
     def subscribe(self, source: EventSource, event_handler: Any):
         """Registers an callable that will be called when an event source has new events.
@@ -64,6 +78,10 @@ def _dispatch_next(self, ge_or_assert: Optional[datetime.datetime]):
         ]
         for source in sources_to_pop:
             if source.events:
+                df = ohlcv_to_dataframe([event.data for event in source.events])
+                bt = self.backtesting(data=df)
+                bt.run()
+
                 event = source.events.pop()
                 # Check that events from the same source are returned in order.
                 prev_event = self._prev_events.get(source)
@@ -92,8 +110,6 @@ def _dispatch_next(self, ge_or_assert: Optional[datetime.datetime]):
                 self._prefetched_events[source] = None
 
         self._current_event_dt = None
-        self.strategy.next()
-        self.strategy.init()
         return next_dt
 
     def stop(self):

livetrading/live_trading.py

@@ -1,32 +1,34 @@
+import pandas as pd
 import websocket
 
 from backtesting import Strategy
-from backtesting._util import _Data
 from livetrading import executor
 from livetrading.broker import Broker, Pair
 from livetrading.config import config
-from livetrading.converter import ohlcv_to_dataframe
+
+
+def SMA(arr: pd.Series, n: int) -> pd.Series:
+    """
+    Returns `n`-period simple moving average of array `arr`.
+    """
+    return pd.Series(arr).rolling(n).mean()
 
 
 class LiveStrategy(Strategy):
+    n1 = 10
+    n2 = 20
 
-    def __init__(self, broker):
-        super().__init__(broker=broker, data=[], params={})
-        self.event_data = []
+    def __init__(self, broker, data, params):
+        super().__init__(broker=broker, data=data, params=params)
 
     def init(self):
-        super().init()
-        self.set_atr_periods()
+        sma1 = self.I(SMA, self.data.Close, self.n1)
+        sma2 = self.I(SMA, self.data.Close, self.n2)
 
     def set_atr_periods(self):
         if len(self.data) > 1:
             print(self.data.High, self.data.Low)
 
-    def on_bar_event(self, event):
-        self.event_data.append(event.ticker)
-        event_df = ohlcv_to_dataframe(self.event_data)
-        self._data = _Data(event_df.copy(deep=False))
-
     def next(self):
         print(self.data)
 
@@ -46,19 +48,21 @@ def on_event(self, bar_event):
 
     websocket.enableTrace(False)
 
-    event_dis = executor.EventDispatcher()
+    event_dis = executor.EventDispatcher(LiveStrategy)
 
     exchange = Broker(event_dis, config=config)
 
     pair_info = exchange.get_pair_info('BTC-USD')
 
     position_mgr = PositionManager(exchange, 0.8)
 
-    strategy = LiveStrategy(exchange)
+    strategy = LiveStrategy(exchange, [], {})
 
     exchange.subscribe_to_ticker_events(Pair(base_symbol="UTC", quote_symbol="SDT"),
-                                        strategy.on_bar_event)
+                                        '3m', position_mgr.on_event)
 
     event_dis.set_strategy(strategy)
 
+    event_dis.set_backtesting_partial(cash=100000)
+
     event_dis.run()