fix(scheduler): company & security mismatch

also enhanced empty float values handling
fcote · Feb 22, 2023 · f3e92e3 · f3e92e3
1 parent 4abd647
commit f3e92e3
Show file tree

Hide file tree

Showing 15 changed files with 1,419 additions and 166 deletions.
diff --git a/scheduler/config/config.go b/scheduler/config/config.go
@@ -63,7 +63,7 @@ func getBaseConfig() *Config {
 		Job: JobConfig{
 			FullSync: CronConfig{
 				Enabled: true,
-				Rule:    "0 0 0 * * *",
+				Rule:    "* * * * * *",
 			},
 			NewsSync: CronConfig{
 				Enabled: true,

diff --git a/scheduler/go.mod b/scheduler/go.mod
@@ -1,10 +1,10 @@
 module github.com/fcote/merlin/sheduler
 
-go 1.19
+go 1.20
 
 require (
 	github.com/go-co-op/gocron v1.18.0
-	github.com/jackc/pgx/v4 v4.18.0
+	github.com/jackc/pgx/v5 v5.3.0
 	github.com/newrelic/go-agent/v3 v3.20.3
 	github.com/newrelic/go-agent/v3/integrations/logcontext-v2/nrzerolog v1.0.0
 	github.com/rs/zerolog v1.29.0
@@ -17,14 +17,9 @@ require (
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
-	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
-	github.com/jackc/pgconn v1.14.0 // indirect
-	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
-	github.com/jackc/pgproto3/v2 v2.3.2 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
-	github.com/jackc/pgtype v1.14.0 // indirect
-	github.com/jackc/puddle v1.3.0 // indirect
+	github.com/jackc/puddle/v2 v2.2.0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mattn/go-colorable v0.1.12 // indirect
 	github.com/mattn/go-isatty v0.0.14 // indirect

diff --git a/scheduler/go.sum b/scheduler/go.sum
diff --git a/scheduler/internal/domain/earning.go b/scheduler/internal/domain/earning.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/fcote/merlin/sheduler/pkg/math"
 	"github.com/fcote/merlin/sheduler/pkg/pointer"
 	"github.com/fcote/merlin/sheduler/pkg/slices"
 )
@@ -102,7 +103,7 @@ func (earnings Earnings) Times() []string {
 
 func (earnings Earnings) EpsEstimates() []*float64 {
 	return slices.Map(earnings, func(e Earning) *float64 {
-		if e.EpsEstimate == 0 {
+		if math.IsEmpty(e.EpsEstimate) {
 			return nil
 		}
 		return pointer.To(e.EpsEstimate)
@@ -111,7 +112,7 @@ func (earnings Earnings) EpsEstimates() []*float64 {
 
 func (earnings Earnings) Epss() []*float64 {
 	return slices.Map(earnings, func(e Earning) *float64 {
-		if e.Eps == 0 {
+		if math.IsEmpty(e.Eps) {
 			return nil
 		}
 		return pointer.To(e.Eps)
@@ -120,7 +121,7 @@ func (earnings Earnings) Epss() []*float64 {
 
 func (earnings Earnings) EpsSurprisePercents() []*float64 {
 	return slices.Map(earnings, func(e Earning) *float64 {
-		if e.EpsSurprisePercent == 0 {
+		if math.IsEmpty(e.EpsSurprisePercent) {
 			return nil
 		}
 		return pointer.To(e.EpsSurprisePercent)
@@ -129,7 +130,7 @@ func (earnings Earnings) EpsSurprisePercents() []*float64 {
 
 func (earnings Earnings) RevenueEstimates() []*float64 {
 	return slices.Map(earnings, func(e Earning) *float64 {
-		if e.RevenueEstimate == 0 {
+		if math.IsEmpty(e.RevenueEstimate) {
 			return nil
 		}
 		return pointer.To(e.RevenueEstimate)
@@ -138,7 +139,7 @@ func (earnings Earnings) RevenueEstimates() []*float64 {
 
 func (earnings Earnings) Revenues() []*float64 {
 	return slices.Map(earnings, func(e Earning) *float64 {
-		if e.Revenue == 0 {
+		if math.IsEmpty(e.Revenue) {
 			return nil
 		}
 		return pointer.To(e.Revenue)
@@ -147,7 +148,7 @@ func (earnings Earnings) Revenues() []*float64 {
 
 func (earnings Earnings) RevenueSurprisePercents() []*float64 {
 	return slices.Map(earnings, func(e Earning) *float64 {
-		if e.RevenueSurprisePercent == 0 {
+		if math.IsEmpty(e.RevenueSurprisePercent) {
 			return nil
 		}
 		return pointer.To(e.RevenueSurprisePercent)

diff --git a/scheduler/internal/domain/financial.go b/scheduler/internal/domain/financial.go
@@ -3,6 +3,7 @@ package domain
 import (
 	"fmt"
 
+	"github.com/fcote/merlin/sheduler/pkg/math"
 	"github.com/fcote/merlin/sheduler/pkg/pointer"
 	"github.com/fcote/merlin/sheduler/pkg/slices"
 )
@@ -81,7 +82,7 @@ type Financials []Financial
 
 func (financials Financials) Values() []*float64 {
 	return slices.Map(financials, func(f Financial) *float64 {
-		if f.Value == 0 {
+		if math.IsEmpty(f.Value) {
 			return nil
 		}
 		return pointer.To(f.Value)

diff --git a/scheduler/internal/domain/historical_price.go b/scheduler/internal/domain/historical_price.go
@@ -50,7 +50,7 @@ func (prices HistoricalPrices) Dates() []string {
 
 func (prices HistoricalPrices) Opens() []*float64 {
 	return slices.Map(prices, func(i HistoricalPrice) *float64 {
-		if i.Open == 0 {
+		if math.IsEmpty(i.Open) {
 			return nil
 		}
 		return pointer.To(i.Open)
@@ -59,7 +59,7 @@ func (prices HistoricalPrices) Opens() []*float64 {
 
 func (prices HistoricalPrices) Highs() []*float64 {
 	return slices.Map(prices, func(i HistoricalPrice) *float64 {
-		if i.High == 0 {
+		if math.IsEmpty(i.High) {
 			return nil
 		}
 		return pointer.To(i.High)
@@ -68,7 +68,7 @@ func (prices HistoricalPrices) Highs() []*float64 {
 
 func (prices HistoricalPrices) Lows() []*float64 {
 	return slices.Map(prices, func(i HistoricalPrice) *float64 {
-		if i.Low == 0 {
+		if math.IsEmpty(i.Low) {
 			return nil
 		}
 		return pointer.To(i.Low)
@@ -77,7 +77,7 @@ func (prices HistoricalPrices) Lows() []*float64 {
 
 func (prices HistoricalPrices) Closes() []*float64 {
 	return slices.Map(prices, func(i HistoricalPrice) *float64 {
-		if i.Close == 0 {
+		if math.IsEmpty(i.Close) {
 			return nil
 		}
 		return pointer.To(i.Close)
@@ -86,7 +86,7 @@ func (prices HistoricalPrices) Closes() []*float64 {
 
 func (prices HistoricalPrices) Volumes() []*float64 {
 	return slices.Map(prices, func(i HistoricalPrice) *float64 {
-		if i.Volume == 0 {
+		if math.IsEmpty(i.Volume) {
 			return nil
 		}
 		return pointer.To(i.Volume)
@@ -95,7 +95,7 @@ func (prices HistoricalPrices) Volumes() []*float64 {
 
 func (prices HistoricalPrices) Changes() []*float64 {
 	return slices.Map(prices, func(i HistoricalPrice) *float64 {
-		if i.Change == 0 {
+		if math.IsEmpty(i.Change) {
 			return nil
 		}
 		return pointer.To(i.Change)
@@ -104,7 +104,7 @@ func (prices HistoricalPrices) Changes() []*float64 {
 
 func (prices HistoricalPrices) ChangePercents() []*float64 {
 	return slices.Map(prices, func(i HistoricalPrice) *float64 {
-		if i.ChangePercent == 0 {
+		if math.IsEmpty(i.ChangePercent) {
 			return nil
 		}
 		return pointer.To(i.ChangePercent)

diff --git a/scheduler/internal/repository/fmp/company.go b/scheduler/internal/repository/fmp/company.go
@@ -6,17 +6,21 @@ import (
 
 	"github.com/fcote/merlin/sheduler/internal/domain"
 	"github.com/fcote/merlin/sheduler/pkg/fmp"
+	"github.com/fcote/merlin/sheduler/pkg/maps"
 )
 
 func (r Repository) Companies(ctx context.Context, tickers []string) ([]domain.CompanyBase, error) {
 	companies, err := r.client.BatchCompanies(ctx, tickers)
 	if err != nil {
 		return nil, err
 	}
+	companiesByTicker := maps.GroupBy(companies, func(c fmp.Company) string {
+		return c.Symbol
+	})
 
-	result := make([]domain.CompanyBase, len(companies))
-	for i, company := range companies {
-		result[i] = CompanyBaseFromFMP(company)
+	result := make([]domain.CompanyBase, len(tickers))
+	for i, ticker := range tickers {
+		result[i] = CompanyBaseFromFMP(companiesByTicker[ticker])
 	}
 
 	return result, nil

diff --git a/scheduler/internal/repository/fmp/security.go b/scheduler/internal/repository/fmp/security.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/fcote/merlin/sheduler/internal/domain"
 	"github.com/fcote/merlin/sheduler/pkg/fmp"
+	"github.com/fcote/merlin/sheduler/pkg/maps"
 	"github.com/fcote/merlin/sheduler/pkg/pointer"
 )
 
@@ -28,15 +29,21 @@ func (r Repository) Securities(ctx context.Context, tickers []string) ([]domain.
 	if err != nil {
 		return nil, err
 	}
+	companiesByTicker := maps.GroupBy(companies, func(c fmp.Company) string {
+		return c.Symbol
+	})
 
 	stocks, err := r.client.BatchStocks(ctx, tickers)
 	if err != nil {
 		return nil, err
 	}
+	stocksByTicker := maps.GroupBy(stocks, func(s fmp.Stock) string {
+		return s.Symbol
+	})
 
 	result := make([]domain.SecurityBase, len(tickers))
-	for i := range stocks {
-		result[i] = SecurityBaseFromFMP(stocks[i], companies[i])
+	for i, ticker := range tickers {
+		result[i] = SecurityBaseFromFMP(stocksByTicker[ticker], companiesByTicker[ticker])
 	}
 
 	return result, nil

diff --git a/scheduler/internal/repository/pg/base.go b/scheduler/internal/repository/pg/base.go
@@ -4,8 +4,8 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/jackc/pgx/v4"
-	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/fcote/merlin/sheduler/internal/usecase"
 )

diff --git a/scheduler/internal/usecase/earning.go b/scheduler/internal/usecase/earning.go
@@ -52,8 +52,7 @@ func (uc EarningUsecase) sync(ctx context.Context, task domain.SecurityTask) (do
 			Map(rawEarnings, func(p domain.EarningBase) domain.Earning {
 				return domain.EarningFromBase(p, task.SecurityId)
 			})
-		result, err := s.
-			BatchInsertEarnings(ctx, earningInputs)
+		result, err := s.BatchInsertEarnings(ctx, earningInputs)
 		if err != nil {
 			return err
 		}

diff --git a/scheduler/main.go b/scheduler/main.go
@@ -4,7 +4,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/rs/zerolog/log"
 
 	"github.com/go-co-op/gocron"
@@ -36,7 +36,7 @@ func main() {
 	if err != nil {
 		logger.Fatal().Msgf("failed to initialize database config: %v", err)
 	}
-	dbPool, err := pgxpool.ConnectConfig(context.Background(), pgConf)
+	dbPool, err := pgxpool.NewWithConfig(context.Background(), pgConf)
 	if err != nil {
 		logger.Fatal().Msgf("failed to initialize database: %v", err)
 	}

diff --git a/scheduler/pkg/fmp/rate_limit_timer.go b/scheduler/pkg/fmp/rate_limit_timer.go
@@ -36,10 +36,10 @@ func (t *RateLimitTimer) Wait() {
 	defer t.mu.Unlock()
 
 	t.nRequests++
-	glog.Get().Debug().Msgf("fmp | rate limit | %d/%d", t.nRequests, t.maxRequestsPerMin)
 
 	if t.nRequests >= t.maxRequestsPerMin {
 		timeToWait := time.Until(t.end)
+
 		glog.Get().Debug().Msgf("fmp | rate limit | waiting %.2fs", timeToWait.Seconds())
 
 		<-time.After(timeToWait)
@@ -53,8 +53,6 @@ func (t *RateLimitTimer) reset() {
 
 	t.nRequests = 0
 
-	glog.Get().Debug().Msg("fmp | rate limit | reset")
-
 	t.timer.Reset(timerDuration)
 	t.end = time.Now().Add(timerDuration)
 }
diff --git a/scheduler/pkg/maps/group.go b/scheduler/pkg/maps/group.go
@@ -0,0 +1,10 @@
+package maps
+
+func GroupBy[T any, C comparable](s []T, fn func(a T) C) map[C]T {
+	ret := make(map[C]T)
+	for _, input := range s {
+		key := fn(input)
+		ret[key] = input
+	}
+	return ret
+}
diff --git a/scheduler/pkg/math/empty.go b/scheduler/pkg/math/empty.go
@@ -0,0 +1,10 @@
+package math
+
+import "math"
+
+func IsEmpty(f float64) bool {
+	return f == 0 ||
+		math.IsNaN(f) ||
+		math.IsInf(f, 0) ||
+		math.IsInf(f, -1)
+}