Merge pull request #1377 from 0xProject/feature/pipeline-cleanup-mega-rebase

Merge all pipeline code into development

1 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;
+    });
+}