track idex orderbook snapshots (#1397)

* Track Idex and Oasis Orderbook Snapshots

4 files changed, 179 insertions, 15 deletions

diff --git a/packages/pipeline/src/parsers/ddex_orders/index.ts b/packages/pipeline/src/parsers/ddex_orders/index.ts
index 81132e8f0..52a998f9f 100644
--- a/packages/pipeline/src/parsers/ddex_orders/index.ts
+++ b/packages/pipeline/src/parsers/ddex_orders/index.ts
@@ -1,7 +1,8 @@
 import { BigNumber } from '@0x/utils';
-import * as R from 'ramda';
 
-import { DdexMarket, DdexOrder, DdexOrderbook } from '../../data_sources/ddex';
+import { aggregateOrders } from '../utils';
+
+import { DdexMarket, DdexOrderbook } from '../../data_sources/ddex';
 import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
 import { OrderType } from '../../types';
 
@@ -28,19 +29,6 @@ export function parseDdexOrders(
 }
 
 /**
- * Aggregates orders by price point for consistency with other exchanges.
- * Querying the Ddex API at level 3 setting returns a breakdown of
- * individual orders at each price point. Other exchanges only give total amount
- * at each price point. Returns an array of <price, amount> tuples.
- * @param ddexOrders A list of Ddex orders awaiting aggregation.
- */
-export function aggregateOrders(ddexOrders: DdexOrder[]): Array<[string, BigNumber]> {
-    const sumAmount = (acc: BigNumber, order: DdexOrder): BigNumber => acc.plus(order.amount);
-    const aggregatedPricePoints = R.reduceBy(sumAmount, new BigNumber(0), R.prop('price'), ddexOrders);
-    return Object.entries(aggregatedPricePoints);
-}
-
-/**
  * Parse a single aggregated Ddex order in order to form a tokenOrder entity
  * which can be saved into the database.
  * @param ddexMarket An object containing information about the market where these
diff --git a/packages/pipeline/src/parsers/idex_orders/index.ts b/packages/pipeline/src/parsers/idex_orders/index.ts
new file mode 100644
index 000000000..dfe27455c
--- /dev/null
+++ b/packages/pipeline/src/parsers/idex_orders/index.ts
@@ -0,0 +1,77 @@
+import { BigNumber } from '@0x/utils';
+
+import { aggregateOrders } from '../utils';
+
+import { IdexOrder, IdexOrderbook, IdexOrderParam } from '../../data_sources/idex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
+import { OrderType } from '../../types';
+
+/**
+ * Marque function of this file.
+ * 1) Takes in orders from an orderbook,
+ * 2) Aggregates them by price point,
+ * 3) Parses them into entities which are then saved into the database.
+ * @param idexOrderbook raw orderbook that we pull from the Idex API.
+ * @param observedTimestamp Time at which the orders for the market were pulled.
+ * @param source The exchange where these orders are placed. In this case 'idex'.
+ */
+export function parseIdexOrders(idexOrderbook: IdexOrderbook, observedTimestamp: number, source: string): TokenOrder[] {
+    const aggregatedBids = aggregateOrders(idexOrderbook.bids);
+    // Any of the bid orders' params will work
+    const idexBidOrder = idexOrderbook.bids[0];
+    const parsedBids =
+        aggregatedBids.length > 0
+            ? aggregatedBids.map(order => parseIdexOrder(idexBidOrder.params, observedTimestamp, 'bid', source, order))
+            : [];
+
+    const aggregatedAsks = aggregateOrders(idexOrderbook.asks);
+    // Any of the ask orders' params will work
+    const idexAskOrder = idexOrderbook.asks[0];
+    const parsedAsks =
+        aggregatedAsks.length > 0
+            ? aggregatedAsks.map(order => parseIdexOrder(idexAskOrder.params, observedTimestamp, 'ask', source, order))
+            : [];
+    return parsedBids.concat(parsedAsks);
+}
+
+/**
+ * Parse a single aggregated Idex order in order to form a tokenOrder entity
+ * which can be saved into the database.
+ * @param idexOrderParam An object containing information about the market where these
+ * trades have been placed.
+ * @param observedTimestamp The time when the API response returned back to us.
+ * @param orderType 'bid' or 'ask' enum.
+ * @param source Exchange where these orders were placed.
+ * @param idexOrder A <price, amount> tuple which we will convert to volume-basis.
+ */
+export function parseIdexOrder(
+    idexOrderParam: IdexOrderParam,
+    observedTimestamp: number,
+    orderType: OrderType,
+    source: string,
+    idexOrder: [string, BigNumber],
+): TokenOrder {
+    const tokenOrder = new TokenOrder();
+    const price = new BigNumber(idexOrder[0]);
+    const amount = idexOrder[1];
+
+    tokenOrder.source = source;
+    tokenOrder.observedTimestamp = observedTimestamp;
+    tokenOrder.orderType = orderType;
+    tokenOrder.price = price;
+    tokenOrder.baseVolume = amount;
+    tokenOrder.quoteVolume = price.times(amount);
+
+    if (orderType === 'bid') {
+        tokenOrder.baseAssetSymbol = idexOrderParam.buySymbol;
+        tokenOrder.baseAssetAddress = idexOrderParam.tokenBuy;
+        tokenOrder.quoteAssetSymbol = idexOrderParam.sellSymbol;
+        tokenOrder.quoteAssetAddress = idexOrderParam.tokenSell;
+    } else {
+        tokenOrder.baseAssetSymbol = idexOrderParam.sellSymbol;
+        tokenOrder.baseAssetAddress = idexOrderParam.tokenSell;
+        tokenOrder.quoteAssetSymbol = idexOrderParam.buySymbol;
+        tokenOrder.quoteAssetAddress = idexOrderParam.tokenBuy;
+    }
+    return tokenOrder;
+}
diff --git a/packages/pipeline/src/parsers/oasis_orders/index.ts b/packages/pipeline/src/parsers/oasis_orders/index.ts
new file mode 100644
index 000000000..7aafbf460
--- /dev/null
+++ b/packages/pipeline/src/parsers/oasis_orders/index.ts
@@ -0,0 +1,71 @@
+import { BigNumber } from '@0x/utils';
+import * as R from 'ramda';
+
+import { aggregateOrders } from '../utils';
+
+import { OasisMarket, OasisOrder } from '../../data_sources/oasis';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../entities';
+import { OrderType } from '../../types';
+
+/**
+ * Marque function of this file.
+ * 1) Takes in orders from an orderbook,
+ * 2) Aggregates them according to price point,
+ * 3) Builds TokenOrder entity with other information attached.
+ * @param oasisOrderbook A raw orderbook that we pull from the Oasis API.
+ * @param oasisMarket An object containing market data also directly from the API.
+ * @param observedTimestamp Time at which the orders for the market were pulled.
+ * @param source The exchange where these orders are placed. In this case 'oasis'.
+ */
+export function parseOasisOrders(
+    oasisOrderbook: OasisOrder[],
+    oasisMarket: OasisMarket,
+    observedTimestamp: number,
+    source: string,
+): TokenOrder[] {
+    const aggregatedBids = aggregateOrders(R.filter(R.propEq('act', 'bid'), oasisOrderbook));
+    const aggregatedAsks = aggregateOrders(R.filter(R.propEq('act', 'ask'), oasisOrderbook));
+    const parsedBids = aggregatedBids.map(order =>
+        parseOasisOrder(oasisMarket, observedTimestamp, 'bid', source, order),
+    );
+    const parsedAsks = aggregatedAsks.map(order =>
+        parseOasisOrder(oasisMarket, observedTimestamp, 'ask', source, order),
+    );
+    return parsedBids.concat(parsedAsks);
+}
+
+/**
+ * Parse a single aggregated Oasis order to form a tokenOrder entity
+ * which can be saved into the database.
+ * @param oasisMarket An object containing information about the market where these
+ * trades have been placed.
+ * @param observedTimestamp The time when the API response returned back to us.
+ * @param orderType 'bid' or 'ask' enum.
+ * @param source Exchange where these orders were placed.
+ * @param oasisOrder A <price, amount> tuple which we will convert to volume-basis.
+ */
+export function parseOasisOrder(
+    oasisMarket: OasisMarket,
+    observedTimestamp: number,
+    orderType: OrderType,
+    source: string,
+    oasisOrder: [string, BigNumber],
+): TokenOrder {
+    const tokenOrder = new TokenOrder();
+    const price = new BigNumber(oasisOrder[0]);
+    const amount = oasisOrder[1];
+
+    tokenOrder.source = source;
+    tokenOrder.observedTimestamp = observedTimestamp;
+    tokenOrder.orderType = orderType;
+    tokenOrder.price = price;
+
+    tokenOrder.baseAssetSymbol = oasisMarket.base;
+    tokenOrder.baseAssetAddress = null; // Oasis doesn't provide address information
+    tokenOrder.baseVolume = price.times(amount);
+
+    tokenOrder.quoteAssetSymbol = oasisMarket.quote;
+    tokenOrder.quoteAssetAddress = null; // Oasis doesn't provide address information
+    tokenOrder.quoteVolume = amount;
+    return tokenOrder;
+}
diff --git a/packages/pipeline/src/parsers/utils.ts b/packages/pipeline/src/parsers/utils.ts
new file mode 100644
index 000000000..860729e9f
--- /dev/null
+++ b/packages/pipeline/src/parsers/utils.ts
@@ -0,0 +1,28 @@
+import { BigNumber } from '@0x/utils';
+
+export interface GenericRawOrder {
+    price: string;
+    amount: string;
+}
+
+/**
+ * Aggregates individual orders by price point. Filters zero amount orders.
+ * @param rawOrders An array of objects that have price and amount information.
+ */
+export function aggregateOrders(rawOrders: GenericRawOrder[]): Array<[string, BigNumber]> {
+    const aggregatedOrders = new Map<string, BigNumber>();
+    rawOrders.forEach(order => {
+        const amount = new BigNumber(order.amount);
+        if (amount.isZero()) {
+            return;
+        }
+        // Use string instead of BigNum to aggregate by value instead of variable.
+        // Convert to BigNumber first to consolidate different string
+        // representations of the same number. Eg. '0.0' and '0.00'.
+        const price = new BigNumber(order.price).toString();
+
+        const existingAmount = aggregatedOrders.get(price) || new BigNumber(0);
+        aggregatedOrders.set(price, amount.plus(existingAmount));
+    });
+    return Array.from(aggregatedOrders.entries());
+}