Merge branch 'samholt-master'

build/ccxt.php

@@ -7629,7 +7629,7 @@ public function __construct ($options = array ()) {
             ),
             'markets' => array (
                 'BTC/EUR' => array ( 'id' => 'BTCEUR', 'symbol' => 'BTC/EUR', 'base' => 'BTC', 'quote' => 'EUR' ),
-                'LTC/EUR' => array ( 'id' => 'LTCEUR', 'symbol' => 'LTC/EUR', 'base' => 'LTC', 'quote' => 'EUR' ),
+                // 'LTC/EUR' => array ( 'id' => 'LTCEUR', 'symbol' => 'LTC/EUR', 'base' => 'LTC', 'quote' => 'EUR' ),
             ),
         ), $options));
     }
@@ -17107,8 +17107,9 @@ public function fetch_tickers ($symbols = null, $params = array ()) {
         for ($s = 0; $s < count ($this->symbols); $s++) {
             $symbol = $this->symbols[$s];
             $market = $this->markets[$symbol];
-            if (!$market['darkpool'])
-                $pairs[] = $market['id'];
+            if ($market['active'])
+                if (!$market['darkpool'])
+                    $pairs[] = $market['id'];
         }
         $filter = implode (',', $pairs);
         $response = $this->publicGetTicker (array_merge (array (

ccxt.js

@@ -7394,7 +7394,7 @@ var bl3p = {
     },
     'markets': {
         'BTC/EUR': { 'id': 'BTCEUR', 'symbol': 'BTC/EUR', 'base': 'BTC', 'quote': 'EUR' },
-        'LTC/EUR': { 'id': 'LTCEUR', 'symbol': 'LTC/EUR', 'base': 'LTC', 'quote': 'EUR' },
+        // 'LTC/EUR': { 'id': 'LTCEUR', 'symbol': 'LTC/EUR', 'base': 'LTC', 'quote': 'EUR' },
     },
 
     async fetchBalance (params = {}) {
@@ -16609,7 +16609,7 @@ var kraken = {
         return result;
     },
 
-    async appendInactiveMarkets (result = []) {
+    appendInactiveMarkets (result = []) {
         let precision = { 'amount': 8, 'price': 8 };
         let costLimits = { 'min': 0, 'max': undefined };
         let priceLimits = { 'min': Math.pow (10, -precision['price']), 'max': undefined };
@@ -16680,8 +16680,9 @@ var kraken = {
         for (let s = 0; s < this.symbols.length; s++) {
             let symbol = this.symbols[s];
             let market = this.markets[symbol];
-            if (!market['darkpool'])
-                pairs.push (market['id']);
+            if (market['active'])
+                if (!market['darkpool'])
+                    pairs.push (market['id']);
         }
         let filter = pairs.join (',');
         let response = await this.publicGetTicker (this.extend ({

ccxt/async/exchange.py

@@ -26,10 +26,14 @@
 
 # -----------------------------------------------------------------------------
 
-import aiohttp
 import asyncio
 import concurrent
 import socket
+import time
+import math
+import random
+
+import aiohttp
 
 # -----------------------------------------------------------------------------
 
@@ -50,7 +54,7 @@
 # -----------------------------------------------------------------------------
 
 
-class Exchange (BaseExchange):
+class Exchange(BaseExchange):
 
     def __init__(self, config={}):
         super(Exchange, self).__init__(config)
@@ -72,6 +76,26 @@ def __del__(self):
     # def run_rest_poller_loop
     #     await asyncio.sleep (exchange.rateLimit / 1000.0)
 
+    async def wait_for_token(self):
+        while self.rateLimitTokens <= 1:
+            # if self.verbose:
+            #     print('Waiting for tokens: Exchange: {0}'.format(self.id))
+            self.add_new_tokens()
+            seconds_delays = [0.01, 0.1, 0.7, 1, 1.5, 2]
+            delay = random.choice(seconds_delays)
+            await asyncio.sleep(delay)
+        self.rateLimitTokens -= 1
+
+    def add_new_tokens(self):
+        # if self.verbose:
+        #     print('Adding new tokens: Exchange: {0}'.format(self.id))
+        now = time.monotonic()
+        time_since_update = now - self.rateLimitUpdateTime
+        new_tokens = math.floor((0.8 * 1000.0 * time_since_update) / self.rateLimit)
+        if new_tokens > 1:
+            self.rateLimitTokens = min(self.rateLimitTokens + new_tokens, self.rateLimitMaxTokens)
+            self.rateLimitUpdateTime = now
+
     async def fetch(self, url, method='GET', headers=None, body=None):
         """Perform a HTTP request and return decoded JSON data"""
         headers = headers or {}
@@ -80,14 +104,16 @@ async def fetch(self, url, method='GET', headers=None, body=None):
                 headers.update({'User-Agent': self.userAgent})
             elif (type(self.userAgent) is dict) and ('User-Agent' in self.userAgent):
                 headers.update(self.userAgent)
-        if len(self.proxy):
+        if self.proxy:
             headers.update({'Origin': '*'})
         headers.update({'Accept-Encoding': 'gzip, deflate'})
         url = self.proxy + url
         if self.verbose:
             print(url, method, url, "\nRequest:", headers, body)
         encoded_body = body.encode() if body else None
         session_method = getattr(self.aiohttp_session, method.lower())
+        if self.enableRateLimit:
+            await self.wait_for_token()
         try:
             async with session_method(url, data=encoded_body, headers=headers, timeout=(self.timeout / 1000), proxy=self.aiohttp_proxy) as response:
                 text = await response.text()

ccxt/exchange.py

@@ -142,6 +142,9 @@ class Exchange(object):
     lastRestPollTimestamp = 0
     restRequestQueue = None
     restPollerLoopIsRunning = False
+    rateLimitTokens = 16
+    rateLimitMaxTokens = 16
+    rateLimitUpdateTime = 0
     last_http_response = None
     last_json_response = None
 

ccxt/exchanges.py

@@ -6077,7 +6077,7 @@ def __init__(self, config={}):
             },
             'markets': {
                 'BTC/EUR': {'id': 'BTCEUR', 'symbol': 'BTC/EUR', 'base': 'BTC', 'quote': 'EUR'},
-                'LTC/EUR': {'id': 'LTCEUR', 'symbol': 'LTC/EUR', 'base': 'LTC', 'quote': 'EUR'},
+                # 'LTC/EUR': {'id': 'LTCEUR', 'symbol': 'LTC/EUR', 'base': 'LTC', 'quote': 'EUR'},
             },
         }
         params.update(config)
@@ -14967,8 +14967,9 @@ def fetch_tickers(self, symbols=None, params={}):
         for s in range(0, len(self.symbols)):
             symbol = self.symbols[s]
             market = self.markets[symbol]
-            if not market['darkpool']:
-                pairs.append(market['id'])
+            if market['active']:
+                if not market['darkpool']:
+                    pairs.append(market['id'])
         filter = ','.join(pairs)
         response = self.publicGetTicker(self.extend({
             'pair': filter,

test/test.py

@@ -27,10 +27,12 @@ class Argv(object):
 argv = Argv()
 
 parser = argparse.ArgumentParser()
+
 parser.add_argument('--verbose', action='store_true', help='enable verbose output')
 parser.add_argument('--nonce', type=int, help='integer')
 parser.add_argument('exchange', type=str, help='exchange id in lowercase', nargs='?')
 parser.add_argument('symbol', type=str, help='symbol in uppercase', nargs='?')
+
 parser.parse_args(namespace=argv)
 
 exchanges = {}
@@ -142,8 +144,23 @@ def test_tickers(exchange, symbol):
             dump(green(exchange.id), 'failed to fetch all tickers, fetching multiple tickers at once...')
             tickers = exchange.fetch_tickers([symbol])
             dump(green(exchange.id), 'fetched', green(len(list(tickers.keys()))), 'tickers')
-    else:
-        dump(yellow(exchange.id), 'fetching all tickers at once not supported')
+
+        test_tickers_async(exchange)
+
+# ------------------------------------------------------------------------------
+
+def get_active_symbols(exchange):
+    return [symbol for symbol in exchange.symbols if is_active_symbol (exchange, symbol)]
+
+def is_active_symbol(exchange, symbol):
+    return ('.' not in symbol) and (('active' not in exchange.markets[symbol]) or (exchange.markets[symbol]['active']))
+
+def test_tickers_async(exchange):
+    dump(green(exchange.id), 'fetching all tickers by simultaneous multiple concurrent requests')
+    symbols_to_load = get_active_symbols(exchange)
+    input_coroutines = [exchange.fetchTicker(symbol) for symbol in symbols_to_load]
+    tickers = asyncio.gather(*input_coroutines)
+    dump(green(exchange.id), 'fetched', green(len(list(tickers))), 'tickers')
 
 # ------------------------------------------------------------------------------
 

test/test_async.py

@@ -28,10 +28,12 @@ class Argv(object):
 argv = Argv()
 
 parser = argparse.ArgumentParser()
+parser.add_argument('--token_bucket', action='store_true', help='enable token bucket experimental test')
 parser.add_argument('--verbose', action='store_true', help='enable verbose output')
 parser.add_argument('--nonce', type=int, help='integer')
 parser.add_argument('exchange', type=str, help='exchange id in lowercase', nargs='?')
 parser.add_argument('symbol', type=str, help='symbol in uppercase', nargs='?')
+
 parser.parse_args(namespace=argv)
 
 exchanges = {}
@@ -143,8 +145,23 @@ async def test_tickers(exchange, symbol):
             dump(green(exchange.id), 'failed to fetch all tickers, fetching multiple tickers at once...')
             tickers = await exchange.fetch_tickers([symbol])
             dump(green(exchange.id), 'fetched', green(len(list(tickers.keys()))), 'tickers')
-    else:
-        dump(yellow(exchange.id), 'fetching all tickers at once not supported')
+    elif argv.token_bucket:
+        await test_tickers_async(exchange)
+
+# ------------------------------------------------------------------------------
+
+def get_active_symbols(exchange):
+    return [symbol for symbol in exchange.symbols if is_active_symbol (exchange, symbol)]
+
+def is_active_symbol(exchange, symbol):
+    return ('.' not in symbol) and (('active' not in exchange.markets[symbol]) or (exchange.markets[symbol]['active']))
+
+async def test_tickers_async(exchange):
+    dump(green(exchange.id), 'fetching all tickers by simultaneous multiple concurrent requests')
+    symbols_to_load = get_active_symbols(exchange)
+    input_coroutines = [exchange.fetchTicker(symbol) for symbol in symbols_to_load]
+    tickers = await asyncio.gather(*input_coroutines)
+    dump(green(exchange.id), 'fetched', green(len(list(tickers))), 'tickers')
 
 # ------------------------------------------------------------------------------
 

transpile.js

@@ -495,6 +495,7 @@ function transpilePythonAsyncToSync (oldName, newName) {
                     return (
                         line.replace ('asyncio.get_event_loop().run_until_complete(main())', 'main()')
                             .replace ('import ccxt.async as ccxt', 'import ccxt')
+                            .replace (/.*token\_bucket.*/g, '')
                             .replace ('async ', '')
                             .replace ('await ', ''))
                 })