Use pathos.

jesse/config.py

@@ -10,7 +10,7 @@ config = {
         },
 
         'caching': {
-            'driver': 'pickle'
+            'driver': None
         },
 
         'logging': {
@@ -96,6 +96,9 @@ config = {
                     {'asset': 'BTC', 'balance': 0},
                 ],
             },
+            'Polygon': {
+                'api_key': '',
+            },
         },
 
         # changes the metrics output of the backtest

jesse/helpers.py

@@ -251,6 +251,10 @@ def get_config(keys: str, default=None):
     return CACHED_CONFIG[keys]
 
 
+def get_nan_indices(array):
+    return np.where(np.isnan(array))[0]
+
+
 def get_strategy_class(strategy_name):
     from pydoc import locate
 
@@ -520,6 +524,12 @@ def readable_duration(seconds, granularity=2):
     return ', '.join(result[:granularity])
 
 
+def reinsert_nan(array, nan_indices):
+    for i in range(nan_indices.shape[0]):
+        array = np.concatenate((array[:nan_indices[i]], [np.nan], array[nan_indices[i]:]))
+    return array
+
+
 def relative_to_absolute(path: str) -> str:
     return os.path.abspath(path)
 

jesse/models/Candle.py

@@ -5,16 +5,14 @@ from jesse.services.db import db
 
 
 class Candle(peewee.Model):
-    """
 
-    """
     id = peewee.UUIDField(primary_key=True)
     timestamp = peewee.BigIntegerField()
-    open = peewee.FloatField()
+    open = peewee.FloatField(null = True)
-    close = peewee.FloatField()
+    close = peewee.FloatField(null = True)
-    high = peewee.FloatField()
+    high = peewee.FloatField(null = True)
-    low = peewee.FloatField()
+    low = peewee.FloatField(null = True)
-    volume = peewee.FloatField()
+    volume = peewee.FloatField(null = True)
     symbol = peewee.CharField()
     exchange = peewee.CharField()
 

jesse/modes/import_candles_mode/__init__.py

@@ -12,6 +12,8 @@ from jesse.models import Candle
 from jesse.modes.import_candles_mode.drivers import drivers
 from jesse.modes.import_candles_mode.drivers.interface import CandleExchange
 
+import numpy as np
+
 
 def run(exchange: str, symbol: str, start_date_str: str, skip_confirmation=False):
     try:
@@ -108,8 +110,11 @@ def run(exchange: str, symbol: str, start_date_str: str, skip_confirmation=False
                         run(exchange, symbol, jh.timestamp_to_time(first_existing_timestamp)[:10], True)
                         return
 
-                # fill absent candles (if there's any)
+                if driver.stock_mode:
-                candles = _fill_absent_candles(candles, temp_start_timestamp, temp_end_timestamp)
+                    candles = _fill_absent_candles_with_nan(candles, temp_start_timestamp, temp_end_timestamp)
+                else:
+                    # fill absent candles (if there's any)
+                    candles = _fill_absent_candles(candles, temp_start_timestamp, temp_end_timestamp)
 
                 # store in the database
                 if skip_confirmation:
@@ -245,6 +250,46 @@ def _get_candles_from_backup_exchange(
         return total_candles
 
 
+def _fill_absent_candles_with_nan(temp_candles, start_timestamp, end_timestamp):
+    if len(temp_candles) == 0:
+        raise CandleNotFoundInExchange(
+            'No candles exists in the market for this day: {} \n'
+            'Try another start_date'.format(
+                jh.timestamp_to_time(start_timestamp)[:10],
+            )
+        )
+
+    symbol = temp_candles[0]['symbol']
+    exchange = temp_candles[0]['exchange']
+    candles = []
+    loop_length = ((end_timestamp - start_timestamp) / 60000) + 1
+
+    i = 0
+    while i < loop_length:
+        candle_for_timestamp = pydash.find(
+            temp_candles, lambda c: c['timestamp'] == start_timestamp)
+
+        if candle_for_timestamp is None:
+            candles.append({
+                'id': jh.generate_unique_id(),
+                'symbol': symbol,
+                'exchange': exchange,
+                'timestamp': start_timestamp,
+                'open': np.nan,
+                'high': np.nan,
+                'low': np.nan,
+                'close': np.nan,
+                'volume': np.nan
+            })
+        # candle is present
+        else:
+            candles.append(candle_for_timestamp)
+
+        start_timestamp += 60000
+        i += 1
+    return candles
+
+
 def _fill_absent_candles(temp_candles, start_timestamp, end_timestamp):
     if len(temp_candles) == 0:
         raise CandleNotFoundInExchange(
@@ -305,3 +350,4 @@ def _fill_absent_candles(temp_candles, start_timestamp, end_timestamp):
 
 def _insert_to_database(candles):
     Candle.insert_many(candles).on_conflict_ignore().execute()
+

jesse/modes/import_candles_mode/drivers/__init__.py

@@ -3,6 +3,7 @@ from .binance_futures import BinanceFutures
 from .bitfinex import Bitfinex
 from .coinbase import Coinbase
 from .testnet_binance_futures import TestnetBinanceFutures
+from .polygon import Polygon
 
 drivers = {
     'Binance': Binance,
@@ -10,4 +11,5 @@ drivers = {
     'Testnet Binance Futures': TestnetBinanceFutures,
     'Bitfinex': Bitfinex,
     'Coinbase': Coinbase,
+    'Polygon': Polygon,
 }

jesse/modes/import_candles_mode/drivers/interface.py

@@ -5,11 +5,12 @@ class CandleExchange(ABC):
     """
 
     """
-    def __init__(self, name, count, sleep_time):
+    def __init__(self, name, count, sleep_time, stock_mode = False):
         self.name = name
         self.count = count
         self.sleep_time = sleep_time
         self.backup_exchange: CandleExchange = None
+        self.stock_mode = stock_mode
 
     @abstractmethod
     def init_backup_exchange(self):

jesse/modes/import_candles_mode/drivers/polygon.py

@@ -0,0 +1,64 @@
+from polygon import RESTClient
+from requests import HTTPError
+
+import jesse.helpers as jh
+from .interface import CandleExchange
+
+
+class Polygon(CandleExchange):
+
+    def __init__(self):
+        super().__init__('Polygon', 5000, 0.5, stock_mode=True)
+
+        try:
+            api_key = jh.get_config('env.exchanges.Polygon.api_key')
+        except:
+            raise ValueError("Polygon api_key missing in config.py")
+
+        self.restclient = RESTClient(api_key)
+
+    def init_backup_exchange(self):
+        self.backup_exchange = None
+
+    def get_starting_time(self, symbol):
+
+        return None
+
+    def fetch(self, symbol, start_timestamp):
+
+        base = jh.base_asset(symbol)
+        # Check if symbol exists. Raises HTTP 404 if it doesn't.
+        try:
+            details = self.restclient.reference_ticker_details(base)
+        except HTTPError:
+            raise ValueError("Symbol ({}) probably doesn't exist.".format(base))
+
+        payload = {
+            'unadjusted': 'false',
+            'sort': 'asc',
+            'limit': self.count,
+        }
+
+        # Polygon takes string dates not timestamps
+        start = jh.timestamp_to_date(start_timestamp)
+        end = jh.timestamp_to_date(start_timestamp + (self.count) * 60000)
+        response = self.restclient.stocks_equities_aggregates(base, 1, 'minute', start, end, **payload)
+
+        data = response.results
+
+        candles = []
+
+        for d in data:
+            candles.append({
+                'id': jh.generate_unique_id(),
+                'symbol': symbol,
+                'exchange': self.name,
+                'timestamp': int(d['t']),
+                'open': float(d['o']),
+                'close': float(d['c']),
+                'high': float(d['h']),
+                'low': float(d['l']),
+                'volume': int(d['v'])
+            })
+
+        return candles

jesse/services/candle.py

@@ -21,11 +21,11 @@ def generate_candle_from_one_minutes(timeframe,
 
     return np.array([
         candles[0][0],
-        candles[0][1],
+        candles[0][1] if not np.isnan(candles[:, 1]).any() else np.nan,
-        candles[-1][2],
+        candles[-1][2] if not np.isnan(candles[:, 2]).any() else np.nan,
-        candles[:, 3].max(),
+        candles[:, 3].max() if not np.isnan(candles[:, 3]).any() else np.nan,
-        candles[:, 4].min(),
+        candles[:, 4].min() if not np.isnan(candles[:, 4]).any() else np.nan,
-        candles[:, 5].sum(),
+        candles[:, 5].sum() if not np.isnan(candles[:, 5]).any() else np.nan,
     ])
 
 

jesse/services/project_maker/project_template/config.py

@@ -90,6 +90,10 @@ config = {
                 {'asset': 'BTC', 'balance': 0},
             ],
         },
+            'Polygon': {
+                'api_key': '',
+            },
+
     },
 
     # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #

requirements.txt

@@ -7,6 +7,7 @@ newtulipy~=0.4.2
 numpy~=1.19.4
 pandas==1.2.0
 peewee~=3.14.0
+polygon-api-client~=0.1.9
 psycopg2-binary~=2.8.6
 pydash~=4.9.0
 pytest==6.2.1

setup.py

@@ -13,6 +13,7 @@ REQUIRED_PACKAGES = [
     'numpy',
     'pandas',
     'peewee',
+    'polygon-api-client',
     'psycopg2-binary',
     'pydash',
     'pytest',

tests/test_helpers.py

@@ -211,6 +211,12 @@ def test_get_config():
     assert jh.get_config('env.logging.order_submission', 2020) is True
 
 
+def test_get_nan_indices():
+    arr = np.array([0, 11, 22, np.nan, 33, 44, 54, 55, np.nan, np.nan, 20])
+    ind = jh.get_nan_indices(arr)
+    assert (ind == np.array([3, 8, 9])).all()
+
+
 def test_get_strategy_class():
     from jesse.strategies import Strategy
     assert issubclass(jh.get_strategy_class("Test01"), Strategy)
@@ -405,6 +411,14 @@ def test_readable_duration():
     assert jh.readable_duration(604312) == "6 days, 23 hours"
 
 
+def test_reinsert_nan():
+    arr = np.array([0, 11, 22, np.nan, 33, 44, 54, 55, np.nan, np.nan, 20])
+    arr_without_nan = arr[~np.isnan(arr)]
+    ind = jh.get_nan_indices(arr)
+    arr_with_nan = jh.reinsert_nan(arr_without_nan, ind)
+    assert ((arr == arr_with_nan) | (np.isnan(arr) & np.isnan(arr_with_nan))).all()
+
+
 def test_relative_to_absolute():
     from pathlib import Path
     assert jh.relative_to_absolute("tests/test_helpers.py") == str(Path(__file__).absolute())