Merge branch 'development' of https://github.com/0xProject/0x-monorepo into development

2 files changed, 43 insertions, 31 deletions

diff --git a/packages/pipeline/src/data_sources/ohlcv_external/crypto_compare.ts b/packages/pipeline/src/data_sources/ohlcv_external/crypto_compare.ts
index 6b10c29c5..8804c34d0 100644
--- a/packages/pipeline/src/data_sources/ohlcv_external/crypto_compare.ts
+++ b/packages/pipeline/src/data_sources/ohlcv_external/crypto_compare.ts
@@ -7,9 +7,10 @@ import * as R from 'ramda';
 import { TradingPair } from '../../utils/get_ohlcv_trading_pairs';
 
 export interface CryptoCompareOHLCVResponse {
-    Data: Map<string, CryptoCompareOHLCVRecord[]>;
+    Data: CryptoCompareOHLCVRecord[];
     Response: string;
     Message: string;
+    Type: number;
 }
 
 export interface CryptoCompareOHLCVRecord {
@@ -35,7 +36,9 @@ export interface CryptoCompareOHLCVParams {
 const ONE_WEEK = 7 * 24 * 60 * 60 * 1000; // tslint:disable-line:custom-no-magic-numbers
 const ONE_HOUR = 60 * 60 * 1000; // tslint:disable-line:custom-no-magic-numbers
 const ONE_SECOND = 1000;
+const ONE_HOUR_AGO = new Date().getTime() - ONE_HOUR;
 const HTTP_OK_STATUS = 200;
+const CRYPTO_COMPARE_VALID_EMPTY_RESPONSE_TYPE = 96;
 
 export class CryptoCompareOHLCVSource {
     public readonly interval = ONE_WEEK; // the hourly API returns data for one week at a time
@@ -68,17 +71,21 @@ export class CryptoCompareOHLCVSource {
 
         const response = await Promise.resolve(fetchPromise);
         if (response.status !== HTTP_OK_STATUS) {
-            // tslint:disable-next-line:no-console
-            console.log(`Error scraping ${url}`);
-            return [];
+            throw new Error(`HTTP error while scraping Crypto Compare: [${response}]`);
         }
         const json: CryptoCompareOHLCVResponse = await response.json();
-        if (json.Response === 'Error' || Object.keys(json.Data).length === 0) {
-            // tslint:disable-next-line:no-console
-            console.log(`Error scraping ${url}: ${json.Message}`);
-            return [];
+        if (
+            (json.Response === 'Error' || json.Data.length === 0) &&
+            json.Type !== CRYPTO_COMPARE_VALID_EMPTY_RESPONSE_TYPE
+        ) {
+            throw new Error(JSON.stringify(json));
         }
-        return Object.values(json.Data).filter(rec => rec.time * ONE_SECOND >= pair.latestSavedTime);
+        return json.Data.filter(rec => {
+            return (
+                // Crypto Compare takes ~30 mins to finalise records
+                rec.time * ONE_SECOND < ONE_HOUR_AGO && rec.time * ONE_SECOND > pair.latestSavedTime && hasData(rec)
+            );
+        });
     }
     public generateBackfillIntervals(pair: TradingPair): TradingPair[] {
         const now = new Date().getTime();
@@ -92,3 +99,14 @@ export class CryptoCompareOHLCVSource {
         return R.unfold(f, pair);
     }
 }
+
+function hasData(record: CryptoCompareOHLCVRecord): boolean {
+    return (
+        record.close !== 0 ||
+        record.open !== 0 ||
+        record.high !== 0 ||
+        record.low !== 0 ||
+        record.volumefrom !== 0 ||
+        record.volumeto !== 0
+    );
+}
diff --git a/packages/pipeline/src/scripts/pull_ohlcv_cryptocompare.ts b/packages/pipeline/src/scripts/pull_ohlcv_cryptocompare.ts
index 6979cd10e..7377a64d8 100644
--- a/packages/pipeline/src/scripts/pull_ohlcv_cryptocompare.ts
+++ b/packages/pipeline/src/scripts/pull_ohlcv_cryptocompare.ts
@@ -10,11 +10,9 @@ import { fetchOHLCVTradingPairsAsync, TradingPair } from '../utils/get_ohlcv_tra
 
 const SOURCE_NAME = 'CryptoCompare';
 const TWO_HOURS_AGO = new Date().getTime() - 2 * 60 * 60 * 1000; // tslint:disable-line:custom-no-magic-numbers
-const ONE_HOUR_AGO = new Date().getTime() - 60 * 60 * 1000; // tslint:disable-line:custom-no-magic-numbers
-const ONE_SECOND = 1000;
 
 const MAX_CONCURRENT_REQUESTS = parseInt(process.env.CRYPTOCOMPARE_MAX_CONCURRENT_REQUESTS || '14', 10); // tslint:disable-line:custom-no-magic-numbers
-const EARLIEST_BACKFILL_DATE = process.env.OHLCV_EARLIEST_BACKFILL_DATE || '2010-09-01'; // the time when BTC/USD info starts appearing on Crypto Compare
+const EARLIEST_BACKFILL_DATE = process.env.OHLCV_EARLIEST_BACKFILL_DATE || '2014-06-01';
 const EARLIEST_BACKFILL_TIME = new Date(EARLIEST_BACKFILL_DATE).getTime();
 
 let connection: Connection;
@@ -60,28 +58,24 @@ async function fetchAndSaveAsync(
         if (pair.latestSavedTime > TWO_HOURS_AGO) {
             break;
         }
-        const rawRecords = await source.getHourlyOHLCVAsync(pair);
-        const records = rawRecords.filter(rec => {
-            return rec.time * ONE_SECOND < ONE_HOUR_AGO && rec.time * ONE_SECOND > pair.latestSavedTime;
-        }); // Crypto Compare can take ~30mins to finalise records
-        if (records.length === 0) {
-            console.log(`No more records, stopping task for ${JSON.stringify(pair)}`);
-            break;
-        }
-        const metadata: OHLCVMetadata = {
-            exchange: source.default_exchange,
-            fromSymbol: pair.fromSymbol,
-            toSymbol: pair.toSymbol,
-            source: SOURCE_NAME,
-            observedTimestamp: jobTime,
-            interval: source.intervalBetweenRecords,
-        };
-        const parsedRecords = parseRecords(records, metadata);
         try {
-            await saveRecordsAsync(repository, parsedRecords);
+            const records = await source.getHourlyOHLCVAsync(pair);
+            console.log(`Retrieved ${records.length} records for ${JSON.stringify(pair)}`);
+            if (records.length > 0) {
+                const metadata: OHLCVMetadata = {
+                    exchange: source.default_exchange,
+                    fromSymbol: pair.fromSymbol,
+                    toSymbol: pair.toSymbol,
+                    source: SOURCE_NAME,
+                    observedTimestamp: jobTime,
+                    interval: source.intervalBetweenRecords,
+                };
+                const parsedRecords = parseRecords(records, metadata);
+                await saveRecordsAsync(repository, parsedRecords);
+            }
             i++;
         } catch (err) {
-            console.log(`Error saving OHLCVRecords, stopping task for ${JSON.stringify(pair)} [${err}]`);
+            console.log(`Error scraping OHLCVRecords, stopping task for ${JSON.stringify(pair)} [${err}]`);
             break;
         }
     }