Merge branch 'development' into feature/contracts-monorepo-7

8 files changed, 582 insertions, 0 deletions

diff --git a/packages/pipeline/src/data_sources/bloxy/index.ts b/packages/pipeline/src/data_sources/bloxy/index.ts
new file mode 100644
index 000000000..94468d25a
--- /dev/null
+++ b/packages/pipeline/src/data_sources/bloxy/index.ts
@@ -0,0 +1,133 @@
+import axios from 'axios';
+import * as R from 'ramda';
+
+// URL to use for getting dex trades from Bloxy.
+export const BLOXY_DEX_TRADES_URL = 'https://bloxy.info/api/dex/trades';
+// Number of trades to get at once. Must be less than or equal to MAX_OFFSET.
+const TRADES_PER_QUERY = 10000;
+// Maximum offset supported by the Bloxy API.
+const MAX_OFFSET = 100000;
+// Buffer to subtract from offset. This means we will request some trades twice
+// but we have less chance on missing out on any data.
+const OFFSET_BUFFER = 1000;
+// Maximum number of days supported by the Bloxy API.
+const MAX_DAYS = 30;
+// Buffer used for comparing the last seen timestamp to the last returned
+// timestamp. Increasing this reduces chances of data loss but also creates more
+// redundancy and can impact performance.
+// tslint:disable-next-line:custom-no-magic-numbers
+const LAST_SEEN_TIMESTAMP_BUFFER_MS = 1000 * 60 * 30; // 30 minutes
+
+// tslint:disable-next-line:custom-no-magic-numbers
+const millisecondsPerDay = 1000 * 60 * 60 * 24; // ms/d = ms/s * s/m * m/h * h/d
+
+export interface BloxyTrade {
+    tx_hash: string;
+    tx_time: string;
+    tx_date: string;
+    tx_sender: string;
+    smart_contract_id: number;
+    smart_contract_address: string;
+    contract_type: string;
+    maker: string;
+    taker: string;
+    amountBuy: number;
+    makerFee: number;
+    buyCurrencyId: number;
+    buySymbol: string;
+    amountSell: number;
+    takerFee: number;
+    sellCurrencyId: number;
+    sellSymbol: string;
+    maker_annotation: string;
+    taker_annotation: string;
+    protocol: string;
+    buyAddress: string | null;
+    sellAddress: string | null;
+}
+
+interface BloxyError {
+    error: string;
+}
+
+type BloxyResponse<T> = T | BloxyError;
+type BloxyTradeResponse = BloxyResponse<BloxyTrade[]>;
+
+function isError<T>(response: BloxyResponse<T>): response is BloxyError {
+    return (response as BloxyError).error !== undefined;
+}
+
+export class BloxySource {
+    private readonly _apiKey: string;
+
+    constructor(apiKey: string) {
+        this._apiKey = apiKey;
+    }
+
+    /**
+     * Gets all latest trades between the lastSeenTimestamp (minus some buffer)
+     * and the current time. Note that because the Bloxy API has some hard
+     * limits it might not always be possible to get *all* the trades in the
+     * desired time range.
+     * @param lastSeenTimestamp The latest timestamp for trades that have
+     * already been seen.
+     */
+    public async getDexTradesAsync(lastSeenTimestamp: number): Promise<BloxyTrade[]> {
+        let allTrades: BloxyTrade[] = [];
+
+        // Clamp numberOfDays so that it is always between 1 and MAX_DAYS (inclusive)
+        const numberOfDays = R.clamp(1, MAX_DAYS, getDaysSinceTimestamp(lastSeenTimestamp));
+
+        // Keep getting trades until we hit one of the following conditions:
+        //
+        //  1. Offset hits MAX_OFFSET (we can't go back any further).
+        //  2. There are no more trades in the response.
+        //  3. We see a tx_time equal to or earlier than lastSeenTimestamp (plus
+        //     some buffer).
+        //
+        for (let offset = 0; offset <= MAX_OFFSET; offset += TRADES_PER_QUERY - OFFSET_BUFFER) {
+            const trades = await this._getTradesWithOffsetAsync(numberOfDays, offset);
+            if (trades.length === 0) {
+                // There are no more trades left for the days we are querying.
+                // This means we are done.
+                return filterDuplicateTrades(allTrades);
+            }
+            const sortedTrades = R.reverse(R.sortBy(trade => trade.tx_time, trades));
+            allTrades = allTrades.concat(sortedTrades);
+
+            // Check if lastReturnedTimestamp < lastSeenTimestamp
+            const lastReturnedTimestamp = new Date(sortedTrades[0].tx_time).getTime();
+            if (lastReturnedTimestamp < lastSeenTimestamp - LAST_SEEN_TIMESTAMP_BUFFER_MS) {
+                // We are at the point where we have already seen trades for the
+                // timestamp range that is being returned. We're done.
+                return filterDuplicateTrades(allTrades);
+            }
+        }
+        return filterDuplicateTrades(allTrades);
+    }
+
+    private async _getTradesWithOffsetAsync(numberOfDays: number, offset: number): Promise<BloxyTrade[]> {
+        const resp = await axios.get<BloxyTradeResponse>(BLOXY_DEX_TRADES_URL, {
+            params: {
+                key: this._apiKey,
+                days: numberOfDays,
+                limit: TRADES_PER_QUERY,
+                offset,
+            },
+        });
+        if (isError(resp.data)) {
+            throw new Error(`Error in Bloxy API response: ${resp.data.error}`);
+        }
+        return resp.data;
+    }
+}
+
+// Computes the number of days between the given timestamp and the current
+// timestamp (rounded up).
+function getDaysSinceTimestamp(timestamp: number): number {
+    const msSinceTimestamp = Date.now() - timestamp;
+    const daysSinceTimestamp = msSinceTimestamp / millisecondsPerDay;
+    return Math.ceil(daysSinceTimestamp);
+}
+
+const filterDuplicateTrades = R.uniqBy((trade: BloxyTrade) => trade.tx_hash);
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
new file mode 100644
index 000000000..1717eb8b3
--- /dev/null
+++ b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
@@ -0,0 +1,85 @@
+import {
+    ContractWrappers,
+    ExchangeCancelEventArgs,
+    ExchangeCancelUpToEventArgs,
+    ExchangeEventArgs,
+    ExchangeEvents,
+    ExchangeFillEventArgs,
+    ExchangeWrapper,
+} from '@0x/contract-wrappers';
+import { Web3ProviderEngine } from '@0x/subproviders';
+import { Web3Wrapper } from '@0x/web3-wrapper';
+import { LogWithDecodedArgs } from 'ethereum-types';
+
+import { EXCHANGE_START_BLOCK } from '../../utils';
+
+const BLOCK_FINALITY_THRESHOLD = 10; // When to consider blocks as final. Used to compute default toBlock.
+const NUM_BLOCKS_PER_QUERY = 20000; // Number of blocks to query for events at a time.
+
+export class ExchangeEventsSource {
+    private readonly _exchangeWrapper: ExchangeWrapper;
+    private readonly _web3Wrapper: Web3Wrapper;
+    constructor(provider: Web3ProviderEngine, networkId: number) {
+        this._web3Wrapper = new Web3Wrapper(provider);
+        const contractWrappers = new ContractWrappers(provider, { networkId });
+        this._exchangeWrapper = contractWrappers.exchange;
+    }
+
+    public async getFillEventsAsync(
+        fromBlock?: number,
+        toBlock?: number,
+    ): Promise<Array<LogWithDecodedArgs<ExchangeFillEventArgs>>> {
+        return this._getEventsAsync<ExchangeFillEventArgs>(ExchangeEvents.Fill, fromBlock, toBlock);
+    }
+
+    public async getCancelEventsAsync(
+        fromBlock?: number,
+        toBlock?: number,
+    ): Promise<Array<LogWithDecodedArgs<ExchangeCancelEventArgs>>> {
+        return this._getEventsAsync<ExchangeCancelEventArgs>(ExchangeEvents.Cancel, fromBlock, toBlock);
+    }
+
+    public async getCancelUpToEventsAsync(
+        fromBlock?: number,
+        toBlock?: number,
+    ): Promise<Array<LogWithDecodedArgs<ExchangeCancelUpToEventArgs>>> {
+        return this._getEventsAsync<ExchangeCancelUpToEventArgs>(ExchangeEvents.CancelUpTo, fromBlock, toBlock);
+    }
+
+    private async _getEventsAsync<ArgsType extends ExchangeEventArgs>(
+        eventName: ExchangeEvents,
+        fromBlock: number = EXCHANGE_START_BLOCK,
+        toBlock?: number,
+    ): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
+        const calculatedToBlock =
+            toBlock === undefined
+                ? (await this._web3Wrapper.getBlockNumberAsync()) - BLOCK_FINALITY_THRESHOLD
+                : toBlock;
+        let events: Array<LogWithDecodedArgs<ArgsType>> = [];
+        for (let currFromBlock = fromBlock; currFromBlock <= calculatedToBlock; currFromBlock += NUM_BLOCKS_PER_QUERY) {
+            events = events.concat(
+                await this._getEventsForRangeAsync<ArgsType>(
+                    eventName,
+                    currFromBlock,
+                    Math.min(currFromBlock + NUM_BLOCKS_PER_QUERY - 1, calculatedToBlock),
+                ),
+            );
+        }
+        return events;
+    }
+
+    private async _getEventsForRangeAsync<ArgsType extends ExchangeEventArgs>(
+        eventName: ExchangeEvents,
+        fromBlock: number,
+        toBlock: number,
+    ): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
+        return this._exchangeWrapper.getLogsAsync<ArgsType>(
+            eventName,
+            {
+                fromBlock,
+                toBlock,
+            },
+            {},
+        );
+    }
+}
diff --git a/packages/pipeline/src/data_sources/ddex/index.ts b/packages/pipeline/src/data_sources/ddex/index.ts
new file mode 100644
index 000000000..2bbd8c29b
--- /dev/null
+++ b/packages/pipeline/src/data_sources/ddex/index.ts
@@ -0,0 +1,78 @@
+import { fetchAsync, logUtils } from '@0x/utils';
+
+const DDEX_BASE_URL = 'https://api.ddex.io/v2';
+const ACTIVE_MARKETS_URL = `${DDEX_BASE_URL}/markets`;
+const NO_AGGREGATION_LEVEL = 3; // See https://docs.ddex.io/#get-orderbook
+const ORDERBOOK_ENDPOINT = `/orderbook?level=${NO_AGGREGATION_LEVEL}`;
+export const DDEX_SOURCE = 'ddex';
+
+export interface DdexActiveMarketsResponse {
+    status: number;
+    desc: string;
+    data: {
+        markets: DdexMarket[];
+    };
+}
+
+export interface DdexMarket {
+    id: string;
+    quoteToken: string;
+    quoteTokenDecimals: number;
+    quoteTokenAddress: string;
+    baseToken: string;
+    baseTokenDecimals: number;
+    baseTokenAddress: string;
+    minOrderSize: string;
+    maxOrderSize: string;
+    pricePrecision: number;
+    priceDecimals: number;
+    amountDecimals: number;
+}
+
+export interface DdexOrderbookResponse {
+    status: number;
+    desc: string;
+    data: {
+        orderBook: DdexOrderbook;
+    };
+}
+
+export interface DdexOrderbook {
+    marketId: string;
+    bids: DdexOrder[];
+    asks: DdexOrder[];
+}
+
+export interface DdexOrder {
+    price: string;
+    amount: string;
+    orderId: string;
+}
+
+// tslint:disable:prefer-function-over-method
+// ^ Keep consistency with other sources and help logical organization
+export class DdexSource {
+    /**
+     * Call Ddex API to find out which markets they are maintaining orderbooks for.
+     */
+    public async getActiveMarketsAsync(): Promise<DdexMarket[]> {
+        logUtils.log('Getting all active DDEX markets');
+        const resp = await fetchAsync(ACTIVE_MARKETS_URL);
+        const respJson: DdexActiveMarketsResponse = await resp.json();
+        const markets = respJson.data.markets;
+        logUtils.log(`Got ${markets.length} markets.`);
+        return markets;
+    }
+
+    /**
+     * Retrieve orderbook from Ddex API for a given market.
+     * @param marketId String identifying the market we want data for. Eg. 'REP/AUG'
+     */
+    public async getMarketOrderbookAsync(marketId: string): Promise<DdexOrderbook> {
+        logUtils.log(`${marketId}: Retrieving orderbook.`);
+        const marketOrderbookUrl = `${ACTIVE_MARKETS_URL}/${marketId}${ORDERBOOK_ENDPOINT}`;
+        const resp = await fetchAsync(marketOrderbookUrl);
+        const respJson: DdexOrderbookResponse = await resp.json();
+        return respJson.data.orderBook;
+    }
+}
diff --git a/packages/pipeline/src/data_sources/ohlcv_external/crypto_compare.ts b/packages/pipeline/src/data_sources/ohlcv_external/crypto_compare.ts
new file mode 100644
index 000000000..85042501b
--- /dev/null
+++ b/packages/pipeline/src/data_sources/ohlcv_external/crypto_compare.ts
@@ -0,0 +1,110 @@
+// tslint:disable:no-duplicate-imports
+import { fetchAsync } from '@0x/utils';
+import Bottleneck from 'bottleneck';
+import { stringify } from 'querystring';
+import * as R from 'ramda';
+
+import { TradingPair } from '../../utils/get_ohlcv_trading_pairs';
+
+export interface CryptoCompareOHLCVResponse {
+    Data: CryptoCompareOHLCVRecord[];
+    Response: string;
+    Message: string;
+    Type: number;
+}
+
+export interface CryptoCompareOHLCVRecord {
+    time: number; // in seconds, not milliseconds
+    close: number;
+    high: number;
+    low: number;
+    open: number;
+    volumefrom: number;
+    volumeto: number;
+}
+
+export interface CryptoCompareOHLCVParams {
+    fsym: string;
+    tsym: string;
+    e?: string;
+    aggregate?: string;
+    aggregatePredictableTimePeriods?: boolean;
+    limit?: number;
+    toTs?: number;
+}
+
+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;
+const MAX_PAGE_SIZE = 2000;
+
+export class CryptoCompareOHLCVSource {
+    public readonly intervalBetweenRecords = ONE_HOUR;
+    public readonly defaultExchange = 'CCCAGG';
+    public readonly interval = this.intervalBetweenRecords * MAX_PAGE_SIZE; // the hourly API returns data for one interval at a time
+    private readonly _url: string = 'https://min-api.cryptocompare.com/data/histohour?';
+
+    // rate-limit for all API calls through this class instance
+    private readonly _limiter: Bottleneck;
+    constructor(maxReqsPerSecond: number) {
+        this._limiter = new Bottleneck({
+            minTime: ONE_SECOND / maxReqsPerSecond,
+            reservoir: 30,
+            reservoirRefreshAmount: 30,
+            reservoirRefreshInterval: ONE_SECOND,
+        });
+    }
+
+    // gets OHLCV records starting from pair.latest
+    public async getHourlyOHLCVAsync(pair: TradingPair): Promise<CryptoCompareOHLCVRecord[]> {
+        const params = {
+            e: this.defaultExchange,
+            fsym: pair.fromSymbol,
+            tsym: pair.toSymbol,
+            limit: MAX_PAGE_SIZE,
+            toTs: Math.floor((pair.latestSavedTime + this.interval) / ONE_SECOND), // CryptoCompare uses timestamp in seconds. not ms
+        };
+        const url = this._url + stringify(params);
+        const response = await this._limiter.schedule(() => fetchAsync(url));
+        if (response.status !== HTTP_OK_STATUS) {
+            throw new Error(`HTTP error while scraping Crypto Compare: [${response}]`);
+        }
+        const json: CryptoCompareOHLCVResponse = await response.json();
+        if (
+            (json.Response === 'Error' || json.Data.length === 0) &&
+            json.Type !== CRYPTO_COMPARE_VALID_EMPTY_RESPONSE_TYPE
+        ) {
+            throw new Error(JSON.stringify(json));
+        }
+        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();
+        const f = (p: TradingPair): false | [TradingPair, TradingPair] => {
+            if (p.latestSavedTime > now) {
+                return false;
+            } else {
+                return [p, R.merge(p, { latestSavedTime: p.latestSavedTime + this.interval })];
+            }
+        };
+        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/data_sources/paradex/index.ts b/packages/pipeline/src/data_sources/paradex/index.ts
new file mode 100644
index 000000000..46d448f4b
--- /dev/null
+++ b/packages/pipeline/src/data_sources/paradex/index.ts
@@ -0,0 +1,92 @@
+import { fetchAsync, logUtils } from '@0x/utils';
+
+const PARADEX_BASE_URL = 'https://api.paradex.io/consumer/v0';
+const ACTIVE_MARKETS_URL = `${PARADEX_BASE_URL}/markets`;
+const ORDERBOOK_ENDPOINT = `${PARADEX_BASE_URL}/orderbook`;
+const TOKEN_INFO_ENDPOINT = `${PARADEX_BASE_URL}/tokens`;
+export const PARADEX_SOURCE = 'paradex';
+
+export type ParadexActiveMarketsResponse = ParadexMarket[];
+
+export interface ParadexMarket {
+    id: string;
+    symbol: string;
+    baseToken: string;
+    quoteToken: string;
+    minOrderSize: string;
+    maxOrderSize: string;
+    priceMaxDecimals: number;
+    amountMaxDecimals: number;
+    // These are not native to the Paradex API response. We tag them on later
+    // by calling the token endpoint and joining on symbol.
+    baseTokenAddress?: string;
+    quoteTokenAddress?: string;
+}
+
+export interface ParadexOrderbookResponse {
+    marketId: number;
+    marketSymbol: string;
+    bids: ParadexOrder[];
+    asks: ParadexOrder[];
+}
+
+export interface ParadexOrder {
+    amount: string;
+    price: string;
+}
+
+export type ParadexTokenInfoResponse = ParadexTokenInfo[];
+
+export interface ParadexTokenInfo {
+    name: string;
+    symbol: string;
+    address: string;
+}
+
+export class ParadexSource {
+    private readonly _apiKey: string;
+
+    constructor(apiKey: string) {
+        this._apiKey = apiKey;
+    }
+
+    /**
+     * Call Paradex API to find out which markets they are maintaining orderbooks for.
+     */
+    public async getActiveMarketsAsync(): Promise<ParadexActiveMarketsResponse> {
+        logUtils.log('Getting all active Paradex markets.');
+        const resp = await fetchAsync(ACTIVE_MARKETS_URL, {
+            headers: { 'API-KEY': this._apiKey },
+        });
+        const markets: ParadexActiveMarketsResponse = await resp.json();
+        logUtils.log(`Got ${markets.length} markets.`);
+        return markets;
+    }
+
+    /**
+     * Call Paradex API to find out their token information.
+     */
+    public async getTokenInfoAsync(): Promise<ParadexTokenInfoResponse> {
+        logUtils.log('Getting token information from Paradex.');
+        const resp = await fetchAsync(TOKEN_INFO_ENDPOINT, {
+            headers: { 'API-KEY': this._apiKey },
+        });
+        const tokens: ParadexTokenInfoResponse = await resp.json();
+        logUtils.log(`Got information for ${tokens.length} tokens.`);
+        return tokens;
+    }
+
+    /**
+     * Retrieve orderbook from Paradex API for a given market.
+     * @param marketSymbol String representing the market we want data for.
+     */
+    public async getMarketOrderbookAsync(marketSymbol: string): Promise<ParadexOrderbookResponse> {
+        logUtils.log(`${marketSymbol}: Retrieving orderbook.`);
+        const marketOrderbookUrl = `${ORDERBOOK_ENDPOINT}?market=${marketSymbol}`;
+        const resp = await fetchAsync(marketOrderbookUrl, {
+            headers: { 'API-KEY': this._apiKey },
+        });
+        const orderbookResponse: ParadexOrderbookResponse = await resp.json();
+        return orderbookResponse;
+    }
+}
diff --git a/packages/pipeline/src/data_sources/relayer-registry/index.ts b/packages/pipeline/src/data_sources/relayer-registry/index.ts
new file mode 100644
index 000000000..8133f5eae
--- /dev/null
+++ b/packages/pipeline/src/data_sources/relayer-registry/index.ts
@@ -0,0 +1,33 @@
+import axios from 'axios';
+
+export interface RelayerResponse {
+    name: string;
+    homepage_url: string;
+    app_url: string;
+    header_img: string;
+    logo_img: string;
+    networks: RelayerResponseNetwork[];
+}
+
+export interface RelayerResponseNetwork {
+    networkId: number;
+    sra_http_endpoint?: string;
+    sra_ws_endpoint?: string;
+    static_order_fields?: {
+        fee_recipient_addresses?: string[];
+        taker_addresses?: string[];
+    };
+}
+
+export class RelayerRegistrySource {
+    private readonly _url: string;
+
+    constructor(url: string) {
+        this._url = url;
+    }
+
+    public async getRelayerInfoAsync(): Promise<Map<string, RelayerResponse>> {
+        const resp = await axios.get<Map<string, RelayerResponse>>(this._url);
+        return resp.data;
+    }
+}
diff --git a/packages/pipeline/src/data_sources/trusted_tokens/index.ts b/packages/pipeline/src/data_sources/trusted_tokens/index.ts
new file mode 100644
index 000000000..552739fb9
--- /dev/null
+++ b/packages/pipeline/src/data_sources/trusted_tokens/index.ts
@@ -0,0 +1,29 @@
+import axios from 'axios';
+
+export interface ZeroExTrustedTokenMeta {
+    address: string;
+    name: string;
+    symbol: string;
+    decimals: number;
+}
+
+export interface MetamaskTrustedTokenMeta {
+    address: string;
+    name: string;
+    erc20: boolean;
+    symbol: string;
+    decimals: number;
+}
+
+export class TrustedTokenSource<T> {
+    private readonly _url: string;
+
+    constructor(url: string) {
+        this._url = url;
+    }
+
+    public async getTrustedTokenMetaAsync(): Promise<T> {
+        const resp = await axios.get<T>(this._url);
+        return resp.data;
+    }
+}
diff --git a/packages/pipeline/src/data_sources/web3/index.ts b/packages/pipeline/src/data_sources/web3/index.ts
new file mode 100644
index 000000000..45a9ea161
--- /dev/null
+++ b/packages/pipeline/src/data_sources/web3/index.ts
@@ -0,0 +1,22 @@
+import { Web3ProviderEngine } from '@0x/subproviders';
+import { Web3Wrapper } from '@0x/web3-wrapper';
+import { BlockWithoutTransactionData, Transaction } from 'ethereum-types';
+
+export class Web3Source {
+    private readonly _web3Wrapper: Web3Wrapper;
+    constructor(provider: Web3ProviderEngine) {
+        this._web3Wrapper = new Web3Wrapper(provider);
+    }
+
+    public async getBlockInfoAsync(blockNumber: number): Promise<BlockWithoutTransactionData> {
+        const block = await this._web3Wrapper.getBlockIfExistsAsync(blockNumber);
+        if (block == null) {
+            return Promise.reject(new Error(`Could not find block for given block number: ${blockNumber}`));
+        }
+        return block;
+    }
+
+    public async getTransactionInfoAsync(txHash: string): Promise<Transaction> {
+        return this._web3Wrapper.getTransactionByHashAsync(txHash);
+    }
+}