IEX plugin improvements

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))

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,
+}