diff --git a/src/ingest-options-from-polygon.mjs b/src/ingest-options-from-polygon.mjs new file mode 100644 index 0000000..441b2a8 --- /dev/null +++ b/src/ingest-options-from-polygon.mjs @@ -0,0 +1,109 @@ +//import { restClient as PolygonClient } from '@polygon.io/client-js'; +import { createClient as createClickhouseClient } from '@clickhouse/client' + +const apiKey = "H95NTsatM1iTWLUwDLxM2J5zhUVYdCEz"; +//const polygonClient = PolygonClient(apiKey, "https://api.polygon.io", {pagination: false, trace: true,}); // automatically call `next_url` if there is one + +const clickhouse = createClickhouseClient({username:'avraham', password:'buginoo'}); + +const optionAggregatesTableName = "option_aggregates"; +async function deleteClickhouseTable(){ + await clickhouse.command({ + query: `DROP TABLE IF EXISTS ${optionAggregatesTableName}`, + }) +} +async function createClickhouseTable(){ + await clickhouse.command({ + query: ` + CREATE TABLE ${optionAggregatesTableName} + ( + symbol LowCardinality(String), + tsStart DateTime32, + + expirationDate DateTime32, + optionType Enum('call', 'put'), + strike Decimal64(9), + + open Decimal64(9), + close Decimal64(9), + low Decimal64(9), + high Decimal64(9), + volume UInt64, + volume_weighted_price Decimal64(9) + ) + ENGINE MergeTree() + ORDER BY (symbol, tsStart) + `, + }) +} + +async function insertResultsIntoClickhouse(optionAggregatesResult){ + await clickhouse.insert({ + table: optionAggregatesTableName, + // structure should match the desired format, JSONEachRow in this example + values: optionAggregatesResult.results.map(r=>({ + symbol: optionAggregatesResult.ticker, + tsStart: (r.t || 0)/1000, + open: r.o, + close: r.c, + low: r.l, + high: r.h, + volume: r.v, + volume_weighted_price: r.vw + })), + format: 'JSONEachRow', + }); + console.log('inserted', optionAggregatesResult.ticker); +} + +function sleep(ms){ + return new Promise((resolve)=>{ + setTimeout(resolve, ms); + }); +} +/** + * Get the underlying price aggregates for every 10-minute period starting 2 years ago (the lookback period for the free tier) until now + */ +async function fetchOptionAggregates(symbol, fromDate, toDate, limit){ + let optionAggregatesResult = await (await fetch(`https://api.polygon.io/v2/aggs/ticker/${symbol}/range/10/minute/${fromDate}/${toDate}?adjusted=false&sort=asc&limit=${limit}&apiKey=${apiKey}`)).json(); + console.log(optionAggregatesResult.resultsCount); + if(optionAggregatesResult.status === 'OK' && typeof optionAggregatesResult.results !== "undefined" && optionAggregatesResult.results.length > 0){ + await insertResultsIntoClickhouse(optionAggregatesResult); + while(optionAggregatesResult.hasOwnProperty("next_url")){ + await sleep(13000); + optionAggregatesResult = await (await fetch(`${optionAggregatesResult.next_url}&apiKey=${apiKey}`)).json(); + if(optionAggregatesResult.status === 'OK' && typeof optionAggregatesResult.results !== "undefined" && optionAggregatesResult.results.length > 0){ + console.log(optionAggregatesResult.resultsCount); + await insertResultsIntoClickhouse(optionAggregatesResult); + } + else{ + console.log(optionAggregatesResult); + } + } + } + else{ + console.log(optionAggregatesResult); + } +} + +await deleteClickhouseTable(); +await createClickhouseTable(); +for(let symbol of ["AAPL", "GOOGL", "MSFT", "TSLA", "AMD", "NFLX", "SPY"]){ + //await fetchOptionAggregates(symbol, "2021-11-27", "2023-11-25", 50000); + await sleep(13000); +} + + +/** + * For each day (not aggregate), look-up the available option contracts (i.e. + * all unique type/strike/exp combinations). Limit is 1000, but it gives you a + * "nextUrl", so keep calling that until all results are in. + */ + +/** + * For each aggregate and the contracts on that day, get option aggregates + */ + + + +await clickhouse.close(); \ No newline at end of file