diff --git a/contrib/iex/backfill/backfill.go b/contrib/iex/backfill/backfill.go index 0c956e060..7aa67dcba 100644 --- a/contrib/iex/backfill/backfill.go +++ b/contrib/iex/backfill/backfill.go @@ -13,6 +13,7 @@ import ( "time" "github.com/alpacahq/marketstore/contrib/calendar" + "github.com/alpacahq/marketstore/contrib/iex/filter" "github.com/alpacahq/marketstore/contrib/ondiskagg/aggtrigger" "github.com/alpacahq/marketstore/executor" "github.com/alpacahq/marketstore/plugins/trigger" @@ -25,9 +26,11 @@ import ( ) var ( - dir string - from string - to string + dir string + from string + to string + filterName string + symbolFilter filter.SymbolFilter // NY timezone NY, _ = time.LoadLocation("America/New_York") @@ -38,6 +41,7 @@ func init() { flag.StringVar(&dir, "dir", "/project/data", "mktsdb directory to backfill to") flag.StringVar(&from, "from", time.Now().Add(-365*24*time.Hour).Format(format), "backfill from date (YYYY-MM-DD)") flag.StringVar(&to, "to", time.Now().Format(format), "backfill from date (YYYY-MM-DD)") + flag.StringVar(&filterName, "filter", "", "symbols filter (SPY)") flag.Parse() } @@ -55,6 +59,11 @@ func main() { log.Fatal(err.Error()) } + sf, found := filter.Filters[filterName] + if found { + symbolFilter = sf + } + log.Info("backfilling from %v to %v", start.Format(format), end.Format(format)) sem := make(chan struct{}, runtime.NumCPU()) @@ -182,7 +191,7 @@ func writeBars(bars []*consolidator.Bar) error { for i := range bars { batch, index := nextBatch(bars, i) - if len(batch) > 0 { + if len(batch) > 0 && symbolFilter(batch[0].Symbol) { tbk := NewTimeBucketKeyFromString(fmt.Sprintf("%s/1Min/OHLCV", batch[0].Symbol)) epoch := make([]int64, len(batch)) diff --git a/contrib/iex/filter/filter.go b/contrib/iex/filter/filter.go new file mode 100644 index 000000000..d0e791e2e --- /dev/null +++ b/contrib/iex/filter/filter.go @@ -0,0 +1,9 @@ +// Package filter serves as a utility for filtering IEX symbols +package filter + +// SymbolFilter defines a function type for filtering symbols +type SymbolFilter func(string) bool + +var Filters = map[string]SymbolFilter{ + "SPY": SPY, +} diff --git a/contrib/iex/filter/spy.go b/contrib/iex/filter/spy.go new file mode 100644 index 000000000..19f75156f --- /dev/null +++ b/contrib/iex/filter/spy.go @@ -0,0 +1,522 @@ +package filter + +// SPY is a function which indicates if the given symbols is within S&P500 +// symbols +func SPY(symbol string) bool { + for _, s := range SPYSymbols { + if s == symbol { + return true + } + } + + return false +} + +// SPYSymbols is array symbols which are within S&P500 +var SPYSymbols = []string{ + "MMM", + "AOS", + "ABT", + "ABBV", + "ACN", + "ATVI", + "AYI", + "ADBE", + "AAP", + "AMD", + "AES", + "AET", + "AMG", + "AFL", + "A", + "APD", + "AKAM", + "ALK", + "ALB", + "ARE", + "ALXN", + "ALGN", + "ALLE", + "AGN", + "ADS", + "LNT", + "ALL", + "GOOGL", + "GOOG", + "MO", + "AMZN", + "AEE", + "AAL", + "AEP", + "AXP", + "AIG", + "AMT", + "AWK", + "AMP", + "ABC", + "AME", + "AMGN", + "APH", + "APC", + "ADI", + "ANDV", + "ANSS", + "ANTM", + "AON", + "APA", + "AIV", + "AAPL", + "AMAT", + "APTV", + "ADM", + "ARNC", + "AJG", + "AIZ", + "T", + "ADSK", + "ADP", + "AZO", + "AVB", + "AVY", + "BHGE", + "BLL", + "BAC", + "BAX", + "BBT", + "BDX", + "BRK.B", + "BBY", + "BIIB", + "BLK", + "HRB", + "BA", + "BKNG", + "BWA", + "BXP", + "BSX", + "BHF", + "BMY", + "AVGO", + "BF.B", + "CHRW", + "CA", + "COG", + "CDNS", + "CPB", + "COF", + "CAH", + "KMX", + "CCL", + "CAT", + "CBOE", + "CBRE", + "CBS", + "CELG", + "CNC", + "CNP", + "CTL", + "CERN", + "CF", + "SCHW", + "CHTR", + "CVX", + "CMG", + "CB", + "CHD", + "CI", + "XEC", + "CINF", + "CTAS", + "CSCO", + "C", + "CFG", + "CTXS", + "CME", + "CMS", + "KO", + "CTSH", + "CL", + "CMCSA", + "CMA", + "CAG", + "CXO", + "COP", + "ED", + "STZ", + "GLW", + "COST", + "COTY", + "CCI", + "CSRA", + "CSX", + "CMI", + "CVS", + "DHI", + "DHR", + "DRI", + "DVA", + "DE", + "DAL", + "XRAY", + "DVN", + "DLR", + "DFS", + "DISCA", + "DISCK", + "DISH", + "DG", + "DLTR", + "D", + "DOV", + "DWDP", + "DPS", + "DTE", + "DUK", + "DRE", + "DXC", + "ETFC", + "EMN", + "ETN", + "EBAY", + "ECL", + "EIX", + "EW", + "EA", + "EMR", + "ETR", + "EVHC", + "EOG", + "EQT", + "EFX", + "EQIX", + "EQR", + "ESS", + "EL", + "RE", + "ES", + "EXC", + "EXPE", + "EXPD", + "ESRX", + "EXR", + "XOM", + "FFIV", + "FB", + "FAST", + "FRT", + "FDX", + "FIS", + "FITB", + "FE", + "FISV", + "FLIR", + "FLS", + "FLR", + "FMC", + "FL", + "F", + "FTV", + "FBHS", + "BEN", + "FCX", + "GPS", + "GRMN", + "IT", + "GD", + "GE", + "GGP", + "GIS", + "GM", + "GPC", + "GILD", + "GPN", + "GS", + "GT", + "GWW", + "HAL", + "HBI", + "HOG", + "HRS", + "HIG", + "HAS", + "HCA", + "HCP", + "HP", + "HSIC", + "HES", + "HPE", + "HLT", + "HOLX", + "HD", + "HON", + "HRL", + "HST", + "HPQ", + "HUM", + "HBAN", + "HII", + "IDXX", + "INFO", + "ITW", + "ILMN", + "INCY", + "IR", + "INTC", + "ICE", + "IBM", + "IP", + "IPG", + "IFF", + "INTU", + "ISRG", + "IVZ", + "IPGP", + "IQV", + "IRM", + "JBHT", + "JEC", + "SJM", + "JNJ", + "JCI", + "JPM", + "JNPR", + "KSU", + "K", + "KEY", + "KMB", + "KIM", + "KMI", + "KLAC", + "KSS", + "KHC", + "KR", + "LB", + "LLL", + "LH", + "LRCX", + "LEG", + "LEN", + "LUK", + "LLY", + "LNC", + "LKQ", + "LMT", + "L", + "LOW", + "LYB", + "MTB", + "MAC", + "M", + "MRO", + "MPC", + "MAR", + "MMC", + "MLM", + "MAS", + "MA", + "MAT", + "MKC", + "MCD", + "MCK", + "MDT", + "MRK", + "MET", + "MTD", + "MGM", + "KORS", + "MCHP", + "MU", + "MSFT", + "MAA", + "MHK", + "TAP", + "MDLZ", + "MON", + "MNST", + "MCO", + "MS", + "MSI", + "MYL", + "NDAQ", + "NOV", + "NAVI", + "NKTR", + "NTAP", + "NFLX", + "NWL", + "NFX", + "NEM", + "NWSA", + "NWS", + "NEE", + "NLSN", + "NKE", + "NI", + "NBL", + "JWN", + "NSC", + "NTRS", + "NOC", + "NCLH", + "NRG", + "NUE", + "NVDA", + "ORLY", + "OXY", + "OMC", + "OKE", + "ORCL", + "PCAR", + "PKG", + "PH", + "PAYX", + "PYPL", + "PNR", + "PBCT", + "PEP", + "PKI", + "PRGO", + "PFE", + "PCG", + "PM", + "PSX", + "PNW", + "PXD", + "PNC", + "RL", + "PPG", + "PPL", + "PX", + "PFG", + "PG", + "PGR", + "PLD", + "PRU", + "PEG", + "PSA", + "PHM", + "PVH", + "QRVO", + "QCOM", + "PWR", + "DGX", + "RRC", + "RJF", + "RTN", + "O", + "RHT", + "REG", + "REGN", + "RF", + "RSG", + "RMD", + "RHI", + "ROK", + "COL", + "ROP", + "ROST", + "RCL", + "SPGI", + "CRM", + "SBAC", + "SCG", + "SLB", + "STX", + "SEE", + "SRE", + "SHW", + "SPG", + "SWKS", + "SLG", + "SNA", + "SO", + "LUV", + "SWK", + "SBUX", + "STT", + "SRCL", + "SYK", + "STI", + "SIVB", + "SYMC", + "SYF", + "SNPS", + "SYY", + "TROW", + "TTWO", + "TPR", + "TGT", + "TEL", + "FTI", + "TXN", + "TXT", + "BK", + "CLX", + "COO", + "HSY", + "MOS", + "TRV", + "DIS", + "TMO", + "TIF", + "TWX", + "TJX", + "TMK", + "TSS", + "TSCO", + "TDG", + "TRIP", + "FOXA", + "FOX", + "TSN", + "USB", + "UDR", + "ULTA", + "UAA", + "UA", + "UNP", + "UAL", + "UNH", + "UPS", + "URI", + "UTX", + "UHS", + "UNM", + "VFC", + "VLO", + "VAR", + "VTR", + "VRSN", + "VRSK", + "VZ", + "VRTX", + "VIAB", + "V", + "VNO", + "VMC", + "WMT", + "WBA", + "WM", + "WAT", + "WEC", + "WFC", + "WELL", + "WDC", + "WU", + "WRK", + "WY", + "WHR", + "WMB", + "WLTW", + "WYN", + "WYNN", + "XEL", + "XRX", + "XLNX", + "XL", + "XYL", + "YUM", + "ZBH", + "ZION", + "ZTS", +} diff --git a/contrib/iex/iex.go b/contrib/iex/iex.go index 712922cb8..e6214d29b 100644 --- a/contrib/iex/iex.go +++ b/contrib/iex/iex.go @@ -8,7 +8,9 @@ import ( "sync" "time" + "github.com/alpacahq/marketstore/contrib/calendar" "github.com/alpacahq/marketstore/contrib/iex/api" + "github.com/alpacahq/marketstore/contrib/iex/filter" "github.com/alpacahq/marketstore/executor" "github.com/alpacahq/marketstore/plugins/bgworker" "github.com/alpacahq/marketstore/utils/io" @@ -37,6 +39,9 @@ type FetcherConfig struct { Intraday bool // list of symbols to poll - queries all if empty Symbols []string + // Filter filters symbols based on predefined set of symbols. Currently + // only SPY filter is supported + Filter string // API Token Token string // True for sandbox @@ -77,6 +82,21 @@ func NewBgWorker(conf map[string]interface{}) (bgworker.BgWorker, error) { } } + // filter symbols if we have configured filter + if config.Filter != "" { + f, found := filter.Filters[config.Filter] + if found { + var filtered []string + for _, symbol := range config.Symbols { + if f(symbol) { + filtered = append(filtered, symbol) + } + } + + config.Symbols = filtered + } + } + return &IEXFetcher{ backfillM: &sync.Map{}, config: config, @@ -104,8 +124,13 @@ func (f *IEXFetcher) Run() { // loop forever over the batches for batch := range f.queue { - f.pollIntraday(batch) - f.pollDaily(batch) + if f.config.Intraday { + f.pollIntraday(batch) + } + + if f.config.Daily && !calendar.Nasdaq.IsMarketOpen(time.Now()) { + f.pollDaily(batch) + } <-time.After(limiter()) f.queue <- batch