aboutsummaryrefslogtreecommitdiffstats
path: root/packages/pipeline/src/data_sources
diff options
context:
space:
mode:
Diffstat (limited to 'packages/pipeline/src/data_sources')
-rw-r--r--packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts45
-rw-r--r--packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts76
-rw-r--r--packages/pipeline/src/data_sources/contract-wrappers/utils.ts67
-rw-r--r--packages/pipeline/src/data_sources/idex/index.ts82
-rw-r--r--packages/pipeline/src/data_sources/oasis/index.ts103
5 files changed, 322 insertions, 51 deletions
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts
new file mode 100644
index 000000000..e0098122f
--- /dev/null
+++ b/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts
@@ -0,0 +1,45 @@
+import {
+ ContractWrappers,
+ ERC20TokenApprovalEventArgs,
+ ERC20TokenEvents,
+ ERC20TokenWrapper,
+} from '@0x/contract-wrappers';
+import { Web3ProviderEngine } from '@0x/subproviders';
+import { LogWithDecodedArgs } from 'ethereum-types';
+
+import { GetEventsFunc, getEventsWithPaginationAsync } from './utils';
+
+export class ERC20EventsSource {
+ private readonly _erc20Wrapper: ERC20TokenWrapper;
+ private readonly _tokenAddress: string;
+ constructor(provider: Web3ProviderEngine, networkId: number, tokenAddress: string) {
+ const contractWrappers = new ContractWrappers(provider, { networkId });
+ this._erc20Wrapper = contractWrappers.erc20Token;
+ this._tokenAddress = tokenAddress;
+ }
+
+ public async getApprovalEventsAsync(
+ startBlock: number,
+ endBlock: number,
+ ): Promise<Array<LogWithDecodedArgs<ERC20TokenApprovalEventArgs>>> {
+ return getEventsWithPaginationAsync(
+ this._getApprovalEventsForRangeAsync.bind(this) as GetEventsFunc<ERC20TokenApprovalEventArgs>,
+ startBlock,
+ endBlock,
+ );
+ }
+
+ // Gets all approval events of for a specific sub-range. This getter
+ // function will be called during each step of pagination.
+ private async _getApprovalEventsForRangeAsync(
+ fromBlock: number,
+ toBlock: number,
+ ): Promise<Array<LogWithDecodedArgs<ERC20TokenApprovalEventArgs>>> {
+ return this._erc20Wrapper.getLogsAsync<ERC20TokenApprovalEventArgs>(
+ this._tokenAddress,
+ ERC20TokenEvents.Approval,
+ { fromBlock, toBlock },
+ {},
+ );
+ }
+}
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
index 1717eb8b3..58691e2ab 100644
--- a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
+++ b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
@@ -8,78 +8,52 @@ import {
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.
+import { GetEventsFunc, getEventsWithPaginationAsync } from './utils';
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,
+ startBlock: number,
+ endBlock: number,
): Promise<Array<LogWithDecodedArgs<ExchangeFillEventArgs>>> {
- return this._getEventsAsync<ExchangeFillEventArgs>(ExchangeEvents.Fill, fromBlock, toBlock);
+ const getFillEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeFillEventArgs>(ExchangeEvents.Fill);
+ return getEventsWithPaginationAsync(getFillEventsForRangeAsync, startBlock, endBlock);
}
public async getCancelEventsAsync(
- fromBlock?: number,
- toBlock?: number,
+ startBlock: number,
+ endBlock: number,
): Promise<Array<LogWithDecodedArgs<ExchangeCancelEventArgs>>> {
- return this._getEventsAsync<ExchangeCancelEventArgs>(ExchangeEvents.Cancel, fromBlock, toBlock);
+ const getCancelEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeCancelEventArgs>(
+ ExchangeEvents.Cancel,
+ );
+ return getEventsWithPaginationAsync(getCancelEventsForRangeAsync, startBlock, endBlock);
}
public async getCancelUpToEventsAsync(
- fromBlock?: number,
- toBlock?: number,
+ startBlock: number,
+ endBlock: 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;
+ const getCancelUpToEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeCancelUpToEventArgs>(
+ ExchangeEvents.CancelUpTo,
+ );
+ return getEventsWithPaginationAsync(getCancelUpToEventsForRangeAsync, startBlock, endBlock);
}
- private async _getEventsForRangeAsync<ArgsType extends ExchangeEventArgs>(
- eventName: ExchangeEvents,
- fromBlock: number,
- toBlock: number,
- ): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
- return this._exchangeWrapper.getLogsAsync<ArgsType>(
- eventName,
- {
- fromBlock,
- toBlock,
- },
- {},
- );
+ // Returns a getter function which gets all events of a specific type for a
+ // specific sub-range. This getter function will be called during each step
+ // of pagination.
+ private _makeGetterFuncForEventType<ArgsType extends ExchangeEventArgs>(
+ eventType: ExchangeEvents,
+ ): GetEventsFunc<ArgsType> {
+ return async (fromBlock: number, toBlock: number) =>
+ this._exchangeWrapper.getLogsAsync<ArgsType>(eventType, { fromBlock, toBlock }, {});
}
}
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/utils.ts b/packages/pipeline/src/data_sources/contract-wrappers/utils.ts
new file mode 100644
index 000000000..67660a37e
--- /dev/null
+++ b/packages/pipeline/src/data_sources/contract-wrappers/utils.ts
@@ -0,0 +1,67 @@
+import { DecodedLogArgs, LogWithDecodedArgs } from 'ethereum-types';
+
+const NUM_BLOCKS_PER_QUERY = 10000; // Number of blocks to query for events at a time.
+const NUM_RETRIES = 3; // Number of retries if a request fails or times out.
+
+export type GetEventsFunc<ArgsType extends DecodedLogArgs> = (
+ fromBlock: number,
+ toBlock: number,
+) => Promise<Array<LogWithDecodedArgs<ArgsType>>>;
+
+/**
+ * Gets all events between the given startBlock and endBlock by querying for
+ * NUM_BLOCKS_PER_QUERY at a time. Accepts a getter function in order to
+ * maximize code re-use and allow for getting different types of events for
+ * different contracts. If the getter function throws with a retryable error,
+ * it will automatically be retried up to NUM_RETRIES times.
+ * @param getEventsAsync A getter function which will be called for each step during pagination.
+ * @param startBlock The start of the entire block range to get events for.
+ * @param endBlock The end of the entire block range to get events for.
+ */
+export async function getEventsWithPaginationAsync<ArgsType extends DecodedLogArgs>(
+ getEventsAsync: GetEventsFunc<ArgsType>,
+ startBlock: number,
+ endBlock: number,
+): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
+ let events: Array<LogWithDecodedArgs<ArgsType>> = [];
+ for (let fromBlock = startBlock; fromBlock <= endBlock; fromBlock += NUM_BLOCKS_PER_QUERY) {
+ const toBlock = Math.min(fromBlock + NUM_BLOCKS_PER_QUERY - 1, endBlock);
+ const eventsInRange = await _getEventsWithRetriesAsync(getEventsAsync, NUM_RETRIES, fromBlock, toBlock);
+ events = events.concat(eventsInRange);
+ }
+ return events;
+}
+
+/**
+ * Calls the getEventsAsync function and retries up to numRetries times if it
+ * throws with an error that is considered retryable.
+ * @param getEventsAsync a function that will be called on each iteration.
+ * @param numRetries the maximum number times to retry getEventsAsync if it fails with a retryable error.
+ * @param fromBlock the start of the sub-range of blocks we are getting events for.
+ * @param toBlock the end of the sub-range of blocks we are getting events for.
+ */
+export async function _getEventsWithRetriesAsync<ArgsType extends DecodedLogArgs>(
+ getEventsAsync: GetEventsFunc<ArgsType>,
+ numRetries: number,
+ fromBlock: number,
+ toBlock: number,
+): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
+ let eventsInRange: Array<LogWithDecodedArgs<ArgsType>> = [];
+ for (let i = 0; i <= numRetries; i++) {
+ try {
+ eventsInRange = await getEventsAsync(fromBlock, toBlock);
+ } catch (err) {
+ if (isErrorRetryable(err) && i < numRetries) {
+ continue;
+ } else {
+ throw err;
+ }
+ }
+ break;
+ }
+ return eventsInRange;
+}
+
+function isErrorRetryable(err: Error): boolean {
+ return err.message.includes('network timeout');
+}
diff --git a/packages/pipeline/src/data_sources/idex/index.ts b/packages/pipeline/src/data_sources/idex/index.ts
new file mode 100644
index 000000000..c1e53c08d
--- /dev/null
+++ b/packages/pipeline/src/data_sources/idex/index.ts
@@ -0,0 +1,82 @@
+import { fetchAsync } from '@0x/utils';
+
+const IDEX_BASE_URL = 'https://api.idex.market';
+const MARKETS_URL = `${IDEX_BASE_URL}/returnTicker`;
+const ORDERBOOK_URL = `${IDEX_BASE_URL}/returnOrderBook`;
+const MAX_ORDER_COUNT = 100; // Maximum based on https://github.com/AuroraDAO/idex-api-docs#returnorderbook
+export const IDEX_SOURCE = 'idex';
+
+export interface IdexMarketsResponse {
+ [marketName: string]: IdexMarket;
+}
+
+export interface IdexMarket {
+ last: string;
+ high: string;
+ low: string;
+ lowestAsk: string;
+ highestBid: string;
+ percentChange: string;
+ baseVolume: string;
+ quoteVolume: string;
+}
+
+export interface IdexOrderbook {
+ asks: IdexOrder[];
+ bids: IdexOrder[];
+}
+
+export interface IdexOrder {
+ price: string;
+ amount: string;
+ total: string;
+ orderHash: string;
+ params: IdexOrderParam;
+}
+
+export interface IdexOrderParam {
+ tokenBuy: string;
+ buySymbol: string;
+ buyPrecision: number;
+ amountBuy: string;
+ tokenSell: string;
+ sellSymbol: string;
+ sellPrecision: number;
+ amountSell: string;
+ expires: number;
+ nonce: number;
+ user: string;
+}
+
+// tslint:disable:prefer-function-over-method
+// ^ Keep consistency with other sources and help logical organization
+export class IdexSource {
+ /**
+ * Call Idex API to find out which markets they are maintaining orderbooks for.
+ */
+ public async getMarketsAsync(): Promise<string[]> {
+ const params = { method: 'POST' };
+ const resp = await fetchAsync(MARKETS_URL, params);
+ const respJson: IdexMarketsResponse = await resp.json();
+ const markets: string[] = Object.keys(respJson);
+ return markets;
+ }
+
+ /**
+ * Retrieve orderbook from Idex API for a given market.
+ * @param marketId String identifying the market we want data for. Eg. 'REP_AUG'
+ */
+ public async getMarketOrderbookAsync(marketId: string): Promise<IdexOrderbook> {
+ const params = {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({
+ market: marketId,
+ count: MAX_ORDER_COUNT,
+ }),
+ };
+ const resp = await fetchAsync(ORDERBOOK_URL, params);
+ const respJson: IdexOrderbook = await resp.json();
+ return respJson;
+ }
+}
diff --git a/packages/pipeline/src/data_sources/oasis/index.ts b/packages/pipeline/src/data_sources/oasis/index.ts
new file mode 100644
index 000000000..3b30e9dfd
--- /dev/null
+++ b/packages/pipeline/src/data_sources/oasis/index.ts
@@ -0,0 +1,103 @@
+import { fetchAsync } from '@0x/utils';
+
+const OASIS_BASE_URL = 'https://data.makerdao.com/v1';
+const OASIS_MARKET_QUERY = `query {
+ oasisMarkets(period: "1 week") {
+ nodes {
+ id
+ base
+ quote
+ buyVol
+ sellVol
+ price
+ high
+ low
+ }
+ }
+}`;
+const OASIS_ORDERBOOK_QUERY = `query ($market: String!) {
+ allOasisOrders(condition: { market: $market }) {
+ totalCount
+ nodes {
+ market
+ offerId
+ price
+ amount
+ act
+ }
+ }
+}`;
+export const OASIS_SOURCE = 'oasis';
+
+export interface OasisMarket {
+ id: string; // market symbol e.g MKRDAI
+ base: string; // base symbol e.g MKR
+ quote: string; // quote symbol e.g DAI
+ buyVol: number; // total buy volume (base)
+ sellVol: number; // total sell volume (base)
+ price: number; // volume weighted price (quote)
+ high: number; // max sell price
+ low: number; // min buy price
+}
+
+export interface OasisMarketResponse {
+ data: {
+ oasisMarkets: {
+ nodes: OasisMarket[];
+ };
+ };
+}
+
+export interface OasisOrder {
+ offerId: number; // Offer Id
+ market: string; // Market symbol (base/quote)
+ price: string; // Offer price (quote)
+ amount: string; // Offer amount (base)
+ act: string; // Action (ask|bid)
+}
+
+export interface OasisOrderbookResponse {
+ data: {
+ allOasisOrders: {
+ totalCount: number;
+ nodes: OasisOrder[];
+ };
+ };
+}
+
+// tslint:disable:prefer-function-over-method
+// ^ Keep consistency with other sources and help logical organization
+export class OasisSource {
+ /**
+ * Call Ddex API to find out which markets they are maintaining orderbooks for.
+ */
+ public async getActiveMarketsAsync(): Promise<OasisMarket[]> {
+ const params = {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ query: OASIS_MARKET_QUERY }),
+ };
+ const resp = await fetchAsync(OASIS_BASE_URL, params);
+ const respJson: OasisMarketResponse = await resp.json();
+ const markets = respJson.data.oasisMarkets.nodes;
+ return markets;
+ }
+
+ /**
+ * Retrieve orderbook from Oasis API for a given market.
+ * @param marketId String identifying the market we want data for. Eg. 'REPAUG'.
+ */
+ public async getMarketOrderbookAsync(marketId: string): Promise<OasisOrder[]> {
+ const input = {
+ market: marketId,
+ };
+ const params = {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ query: OASIS_ORDERBOOK_QUERY, variables: input }),
+ };
+ const resp = await fetchAsync(OASIS_BASE_URL, params);
+ const respJson: OasisOrderbookResponse = await resp.json();
+ return respJson.data.allOasisOrders.nodes;
+ }
+}