pull OHLCV records from Crypto Compare (#1349)

* [WIP] pull OHLCV records from Crypto Compare

* lint

* refactor to pull logic out of script and into modules

* add entity test for ohlcv_external entity

* implement rate limit and chronological backfill for ohlcv

* add unit tests; cleanup variable names

* Fetch OHLCV pairs params from events table

* better method names

* fix outdated test

* lint

* Clean up after review

* oops

* fix failing test

* better filtering of most recent records

* fix bug when generating pairs

* fix default earliest backfill date

* fix bug with retrieving backfill time

* prettier

2 files changed, 38 insertions, 0 deletions

diff --git a/packages/pipeline/src/parsers/ohlcv_external/crypto_compare.ts b/packages/pipeline/src/parsers/ohlcv_external/crypto_compare.ts
new file mode 100644
index 000000000..3efb90384
--- /dev/null
+++ b/packages/pipeline/src/parsers/ohlcv_external/crypto_compare.ts
@@ -0,0 +1,38 @@
+import { CryptoCompareOHLCVRecord } from '../../data_sources/ohlcv_external/crypto_compare';
+import { OHLCVExternal } from '../../entities';
+
+const ONE_SECOND = 1000; // Crypto Compare uses timestamps in seconds instead of milliseconds
+
+export interface OHLCVMetadata {
+    exchange: string;
+    fromSymbol: string;
+    toSymbol: string;
+    source: string;
+    observedTimestamp: number;
+    interval: number;
+}
+/**
+ * Parses OHLCV records from Crypto Compare into an array of OHLCVExternal entities
+ * @param rawRecords an array of OHLCV records from Crypto Compare (not the full response)
+ */
+export function parseRecords(rawRecords: CryptoCompareOHLCVRecord[], metadata: OHLCVMetadata): OHLCVExternal[] {
+    return rawRecords.map(rec => {
+        const ohlcvRecord = new OHLCVExternal();
+        ohlcvRecord.exchange = metadata.exchange;
+        ohlcvRecord.fromSymbol = metadata.fromSymbol;
+        ohlcvRecord.toSymbol = metadata.toSymbol;
+        ohlcvRecord.startTime = rec.time * ONE_SECOND - metadata.interval;
+        ohlcvRecord.endTime = rec.time * ONE_SECOND;
+
+        ohlcvRecord.open = rec.open;
+        ohlcvRecord.close = rec.close;
+        ohlcvRecord.low = rec.low;
+        ohlcvRecord.high = rec.high;
+        ohlcvRecord.volumeFrom = rec.volumefrom;
+        ohlcvRecord.volumeTo = rec.volumeto;
+
+        ohlcvRecord.source = metadata.source;
+        ohlcvRecord.observedTimestamp = metadata.observedTimestamp;
+        return ohlcvRecord;
+    });
+}
diff --git a/packages/pipeline/src/parsers/ohlcv_external/index.ts b/packages/pipeline/src/parsers/ohlcv_external/index.ts
deleted file mode 100644
index e69de29bb..000000000
--- a/packages/pipeline/src/parsers/ohlcv_external/index.ts
+++ /dev/null