add(xignitefeeder): feature to run data feeding when the market is cl…

…osed (#488)

* add(xignitefeeder): feature to run data feeding when the market is closed.

contrib/xignitefeeder/README.md

@@ -40,6 +40,9 @@ bgworkers:
       timeout: 10
       # Interval [sec] to call Xignite API
       interval: 10
+      # If a non-zero value is set for off_hours_interval,
+      # the data-feeding is executed every off_hours_interval[minute] even when the market is closed.
+      off_hours_interval: 5
       # XigniteFeeder runs from openTime ~ closeTime (UTC)
       openTime: "23:00:00" # 08:00 (JST)
       closeTime: "06:10:00" # 15:10 (JST)

contrib/xignitefeeder/configs/config.go

@@ -33,7 +33,10 @@ type DefaultConfig struct {
 	ClosedDaysOfTheWeek []time.Weekday
 	ClosedDays          []time.Time
 	Interval            int `json:"interval"`
-	Backfill            struct {
+	// If a non-zero value is set for OffHoursInterval,
+	// the data-feeding is executed every offHoursInterval[minute] even when the market is closed.
+	OffHoursInterval int `json:"off_hours_interval"`
+	Backfill         struct {
 		Enabled   bool      `json:"enabled"`
 		Since     CustomDay `json:"since"`
 		Timeframe string    `json:"timeframe"`

contrib/xignitefeeder/feed/interval.go

@@ -0,0 +1,40 @@
+package feed
+
+import (
+	"time"
+
+	"github.com/alpacahq/marketstore/v4/utils/log"
+)
+
+// IntervalMarketTimeChecker is used where periodic processing is needed to run even when the market is closed.
+type IntervalMarketTimeChecker struct {
+	MarketTimeChecker
+	// LastTime holds the last time that IntervalTimeChceker.IsOpen returned true.
+	LastTime time.Time
+	Interval time.Duration
+}
+
+func NewIntervalMarketTimeChecker(
+	mtc MarketTimeChecker,
+	interval time.Duration,
+) *IntervalMarketTimeChecker {
+	return &IntervalMarketTimeChecker{
+		MarketTimeChecker: mtc,
+		LastTime:          time.Time{},
+		Interval:          interval,
+	}
+}
+
+// IsOpen returns true when the market is open or the interval elapsed since LastTime.
+func (c *IntervalMarketTimeChecker) IsOpen(t time.Time) bool {
+	return c.MarketTimeChecker.IsOpen(t) || c.intervalElapsed(t)
+}
+
+func (c *IntervalMarketTimeChecker) intervalElapsed(t time.Time) bool {
+	elapsed := t.Sub(c.LastTime) >= c.Interval
+	if elapsed {
+		c.LastTime = t
+		log.Debug("[Xignite Feeder] interval elapsed since last time: " + t.String())
+	}
+	return elapsed
+}

contrib/xignitefeeder/feed/interval_test.go

@@ -0,0 +1,87 @@
+package feed_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/alpacahq/marketstore/v4/contrib/xignitefeeder/feed"
+)
+
+var (
+	jst                      = time.FixedZone("Asia/Tokyo", 9*60*60)
+	exampleDate              = time.Date(2021, 8, 20, 0, 0, 0, 0, time.UTC)
+	exampleDatePlus5min      = time.Date(2021, 8, 20, 0, 5, 0, 0, time.UTC)
+	exampleDatePlus4min59s   = time.Date(2021, 8, 20, 0, 4, 59, 0, time.UTC)
+	exampleDatePlus5minInJST = time.Date(2021, 8, 20, 9, 5, 0, 0, jst)
+)
+
+type mockMarketTimeChecker struct {
+	isOpen bool
+}
+
+func (m *mockMarketTimeChecker) IsOpen(_ time.Time) bool {
+	return m.isOpen
+}
+func (m *mockMarketTimeChecker) Sub(_ time.Time, _ int) (time.Time, error) {
+	panic("not implemented")
+}
+
+func TestIntervalMarketTimeChecker_IsOpen(t *testing.T) {
+	t.Parallel()
+	tests := map[string]struct {
+		MarketTimeChecker feed.MarketTimeChecker
+		Interval          time.Duration
+		CurrentTime       time.Time
+		LastTime          time.Time
+		want              bool
+	}{
+		"ok: IsOpen returns true if the interval elapsed": {
+			MarketTimeChecker: &mockMarketTimeChecker{isOpen: false},
+			Interval:          5 * time.Minute,
+			LastTime:          exampleDate,
+			CurrentTime:       exampleDatePlus5min,
+			want:              true,
+		},
+		"ok: IsOpen returns false if the interval has not yet elapsed": {
+			MarketTimeChecker: &mockMarketTimeChecker{isOpen: false},
+			Interval:          5 * time.Minute,
+			LastTime:          exampleDate,
+			CurrentTime:       exampleDatePlus4min59s,
+			want:              false,
+		},
+		"ok: time in any location can be passed": {
+			MarketTimeChecker: &mockMarketTimeChecker{isOpen: false},
+			Interval:          5 * time.Minute,
+			LastTime:          exampleDate,
+			CurrentTime:       exampleDatePlus5minInJST,
+			want:              true,
+		},
+		"ok: always Open if the base market time checker is IsOpen=true": {
+			MarketTimeChecker: &mockMarketTimeChecker{isOpen: true},
+			Interval:          5 * time.Minute,
+			LastTime:          exampleDate,
+			CurrentTime:       exampleDate,
+			want:              true,
+		},
+	}
+	for name := range tests {
+		tt := tests[name]
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			// --- given ---
+			c := feed.NewIntervalMarketTimeChecker(
+				tt.MarketTimeChecker,
+				tt.Interval,
+			)
+			c.LastTime = tt.LastTime
+
+			// --- when ---
+			got := c.IsOpen(tt.CurrentTime)
+
+			// --- then ---
+			if got != tt.want {
+				t.Errorf("IsOpen() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

contrib/xignitefeeder/feed/time_checker.go

@@ -33,7 +33,12 @@ type DefaultMarketTimeChecker struct {
 }
 
 // NewDefaultMarketTimeChecker initializes the DefaultMarketTimeChecker object with the specifier parameters.s
-func NewDefaultMarketTimeChecker(closedDaysOfTheWeek []time.Weekday, closedDays []time.Time, openTime time.Time, closeTime time.Time) *DefaultMarketTimeChecker {
+func NewDefaultMarketTimeChecker(
+	closedDaysOfTheWeek []time.Weekday,
+	closedDays []time.Time,
+	openTime time.Time,
+	closeTime time.Time,
+) *DefaultMarketTimeChecker {
 	return &DefaultMarketTimeChecker{
 		ClosedDaysOfTheWeek: closedDaysOfTheWeek,
 		ClosedDays:          closedDays,

contrib/xignitefeeder/xignitefeeder.go

@@ -34,11 +34,22 @@ func NewBgWorker(conf map[string]interface{}) (bgworker.BgWorker, error) {
 	apiClient := api.NewDefaultAPIClient(config.APIToken, config.Timeout)
 
 	// init Market Time Checker
-	timeChecker := feed.NewDefaultMarketTimeChecker(
+	var timeChecker feed.MarketTimeChecker
+	timeChecker = feed.NewDefaultMarketTimeChecker(
 		config.ClosedDaysOfTheWeek,
 		config.ClosedDays,
 		config.OpenTime,
 		config.CloseTime)
+	if config.OffHoursInterval != 0 {
+		log.Info(fmt.Sprintf("[Xignite Feeder] off_hours_interval=%dmin is set. "+
+			"The data will be retrieved every %d minutes even when the market is closed.",
+			config.OffHoursInterval, config.OffHoursInterval),
+		)
+		timeChecker = feed.NewIntervalMarketTimeChecker(
+			timeChecker,
+			time.Duration(config.OffHoursInterval)*time.Minute,
+		)
+	}
 
 	ctx := context.Background()
 	// init Symbols Manager to...