Feature/Add_usd_price_and_amount_to_tradelogs

cmd/tradelogs/main.go

@@ -8,12 +8,14 @@ import (
 	"os"
 	"time"
 
+	"github.com/KyberNetwork/go-binance/v2"
 	"github.com/KyberNetwork/tradelogs/pkg/dune"
 	"github.com/KyberNetwork/tradelogs/pkg/parser"
 	"github.com/KyberNetwork/tradelogs/pkg/parser/bebop"
 	"github.com/KyberNetwork/tradelogs/pkg/parser/oneinch"
 	"github.com/KyberNetwork/tradelogs/pkg/parser/oneinchv6"
 	"github.com/KyberNetwork/tradelogs/pkg/parser/uniswapx"
+	"github.com/KyberNetwork/tradelogs/pkg/pricefiller"
 	"github.com/KyberNetwork/tradelogs/pkg/rpcnode"
 	"github.com/KyberNetwork/tradelogs/pkg/tracecall"
 
@@ -50,6 +52,7 @@ func main() {
 	app.Flags = append(app.Flags, libapp.HTTPServerFlags()...)
 	app.Flags = append(app.Flags, libapp.BigqueryFlags()...)
 	app.Flags = append(app.Flags, libapp.RPCNodeFlags()...)
+	app.Flags = append(app.Flags, pricefiller.PriceFillerFlags()...)
 
 	if err := app.Run(os.Args); err != nil {
 		log.Panic(err)
@@ -111,8 +114,15 @@ func run(c *cli.Context) error {
 		bebop.MustNewParser(traceCalls),
 	}
 
+	binanceClient := binance.NewClient(c.String(pricefiller.BinanceAPIKeyFlag.Name), c.String(pricefiller.BinanceSecretKeyFlag.Name))
+	priceFiller, err := pricefiller.NewPriceFiller(binanceClient, s)
+	if err != nil {
+		l.Errorw("Error while init price filler")
+		return err
+	}
+
 	tradeLogChan := make(chan storage.TradeLog, 1000)
-	w, err := worker.New(l, s, listener, tradeLogChan, parsers...)
+	w, err := worker.New(l, s, listener, priceFiller, tradeLogChan, parsers...)
 	if err != nil {
 		l.Errorw("Error while init worker")
 		return err

cmd/tradelogs/migrations/00008_add_price_usd_column.up.sql

@@ -0,0 +1,10 @@
+CREATE TYPE tradelog_states AS ENUM ('new', 'processed');
+
+ALTER TABLE tradelogs
+    ADD COLUMN maker_token_price FLOAT NOT NULL DEFAULT 0,
+    ADD COLUMN taker_token_price FLOAT NOT NULL DEFAULT 0,
+    ADD COLUMN maker_usd_amount FLOAT NOT NULL DEFAULT 0,
+    ADD COLUMN taker_usd_amount FLOAT NOT NULL DEFAULT 0,
+    ADD COLUMN state tradelog_states NOT NULL DEFAULT 'new';
+
+CREATE INDEX tradelogs_state_idx ON tradelogs (state);

go.mod

@@ -7,6 +7,8 @@ toolchain go1.22.2
 require (
 	cloud.google.com/go/bigquery v1.56.0
 	github.com/KyberNetwork/cclog v1.1.0
+	github.com/KyberNetwork/go-binance/v2 v2.0.3
+	github.com/KyberNetwork/tradinglib v0.4.36
 	github.com/TheZeroSlave/zapsentry v1.20.2
 	github.com/ethereum/go-ethereum v1.13.14
 	github.com/getsentry/sentry-go v0.26.0
@@ -33,13 +35,13 @@ require (
 	cloud.google.com/go/compute/metadata v0.2.3 // indirect
 	cloud.google.com/go/iam v1.1.5 // indirect
 	github.com/DataDog/zstd v1.5.2 // indirect
-	github.com/KyberNetwork/tradinglib v0.4.36 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
+	github.com/adshao/go-binance/v2 v2.5.1 // indirect
 	github.com/andybalholm/brotli v1.0.5 // indirect
 	github.com/apache/arrow/go/v12 v12.0.0 // indirect
 	github.com/apache/thrift v0.16.0 // indirect
+	github.com/bitly/go-simplejson v0.5.0 // indirect
 	github.com/bits-and-blooms/bitset v1.13.0 // indirect
-	github.com/btcsuite/btcd/chaincfg/chainhash v1.0.2 // indirect
 	github.com/bytedance/sonic v1.10.2 // indirect
 	github.com/cespare/cp v1.1.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect

go.sum

@@ -35,6 +35,8 @@ github.com/KyberNetwork/cclog v1.1.0 h1:3gqKpSayABuTjS4J7H8qcQHrlfyggpjmHUnCYQT0
 github.com/KyberNetwork/cclog v1.1.0/go.mod h1:vf9+yocFGEyqqObn4Gr9rtj4ffnuKVpKWiSCtfsxIHg=
 github.com/KyberNetwork/evmlistener v0.4.7 h1:SlJwzqngj2N2ot/1M391GbwE/A6wyKfV9z1RIlxZpyI=
 github.com/KyberNetwork/evmlistener v0.4.7/go.mod h1:7ylrHTrF9bRJRcVdw02f4P2k3ohc1/gKGQ3HgaYtfE8=
+github.com/KyberNetwork/go-binance/v2 v2.0.3 h1:VsBBHXAJTxryBsdO6U9MyK+oHEJ9qZtQpZvmEzpHmGA=
+github.com/KyberNetwork/go-binance/v2 v2.0.3/go.mod h1:ihNzjOXDgxZ4cSENV2sQfXnO30wobXQPpqTuKxxUVjw=
 github.com/KyberNetwork/tradinglib v0.4.36 h1:h8+pKMPiqnR/3Q+wGq1JWYxP3qFJ3qXqsPJQETuJcFw=
 github.com/KyberNetwork/tradinglib v0.4.36/go.mod h1:3Ciie58Qmd0mFSnUWAg6phbalao6f2GGcAZ6nxw/lcQ=
 github.com/Masterminds/squirrel v1.5.4 h1:uUcX/aBc8O7Fg9kaISIUsHXdKuqehiXAMQTYX8afzqM=
@@ -47,6 +49,8 @@ github.com/TheZeroSlave/zapsentry v1.20.2 h1:llgC91ZJdoU/OzGxYpUlEhKinf65mw9hJ2K
 github.com/TheZeroSlave/zapsentry v1.20.2/go.mod h1:D1YMfSuu6xnkhwFXxrronesmsiyDhIqo+86I3Ok+r64=
 github.com/VictoriaMetrics/fastcache v1.12.1 h1:i0mICQuojGDL3KblA7wUNlY5lOK6a4bwt3uRKnkZU40=
 github.com/VictoriaMetrics/fastcache v1.12.1/go.mod h1:tX04vaqcNoQeGLD+ra5pU5sWkuxnzWhEzLwhP9w653o=
+github.com/adshao/go-binance/v2 v2.5.1 h1:3CV9iyzEpHFqZFVzjnxI3GtZdIWH9Z/ewR66+q+mCk0=
+github.com/adshao/go-binance/v2 v2.5.1/go.mod h1:41Up2dG4NfMXpCldrDPETEtiOq+pHoGsFZ73xGgaumo=
 github.com/ajg/form v1.5.1/go.mod h1:uL1WgH+h2mgNtvBq0339dVnzXdBETtL2LeUXaIv25UY=
 github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs=
 github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
@@ -58,8 +62,12 @@ github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5
 github.com/aymerick/raymond v2.0.3-0.20180322193309-b565731e1464+incompatible/go.mod h1:osfaiScAUVup+UC9Nfq76eWqDhXlp+4UYaA8uhTBO6g=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/bitly/go-simplejson v0.5.0 h1:6IH+V8/tVMab511d5bn4M7EwGXZf9Hj6i2xSwkNEM+Y=
+github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
 github.com/bits-and-blooms/bitset v1.13.0 h1:bAQ9OPNFYbGHV6Nez0tmNI0RiEu7/hxlYJRUA0wFAVE=
 github.com/bits-and-blooms/bitset v1.13.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
+github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
+github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
 github.com/btcsuite/btcd/btcec/v2 v2.3.2 h1:5n0X6hX0Zk+6omWcihdYvdAlGf2DfasC0GMf7DClJ3U=
 github.com/btcsuite/btcd/btcec/v2 v2.3.2/go.mod h1:zYzJ8etWJQIv1Ogk7OzpWjowwOdXY1W/17j2MW85J04=
 github.com/btcsuite/btcd/chaincfg/chainhash v1.0.2 h1:KdUfX2zKommPRa+PD0sWZUyXe9w277ABlgELO7H04IM=
@@ -528,6 +536,7 @@ github.com/status-im/keycard-go v0.2.0 h1:QDLFswOQu1r5jsycloeQh3bVU8n/NatHHaZobt
 github.com/status-im/keycard-go v0.2.0/go.mod h1:wlp8ZLbsmrF6g6WjugPAx+IzoLrkdf9+mHxBEeo3Hbg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=

internal/worker/worker.go

@@ -9,6 +9,7 @@ import (
 	"github.com/KyberNetwork/tradelogs/pkg/convert"
 	"github.com/KyberNetwork/tradelogs/pkg/evmlistenerclient"
 	"github.com/KyberNetwork/tradelogs/pkg/parser"
+	"github.com/KyberNetwork/tradelogs/pkg/pricefiller"
 	"github.com/KyberNetwork/tradelogs/pkg/storage"
 	"go.uber.org/zap"
 )
@@ -18,10 +19,12 @@ type Worker struct {
 	l            *zap.SugaredLogger
 	s            *storage.Storage
 	p            map[string]parser.Parser
+	priceFiller  *pricefiller.PriceFiller
 	tradeLogChan chan storage.TradeLog
 }
 
-func New(l *zap.SugaredLogger, s *storage.Storage, listener *evmlistenerclient.Client, tradeLogChan chan storage.TradeLog,
+func New(l *zap.SugaredLogger, s *storage.Storage, listener *evmlistenerclient.Client,
+	priceFiller *pricefiller.PriceFiller, tradeLogChan chan storage.TradeLog,
 	parsers ...parser.Parser) (*Worker, error) {
 	p := make(map[string]parser.Parser)
 	for _, ps := range parsers {
@@ -34,6 +37,7 @@ func New(l *zap.SugaredLogger, s *storage.Storage, listener *evmlistenerclient.C
 		l:            l,
 		s:            s,
 		p:            p,
+		priceFiller:  priceFiller,
 		tradeLogChan: tradeLogChan,
 	}, nil
 }
@@ -115,6 +119,7 @@ func (w *Worker) processMessages(m []evmlistenerclient.Message) error {
 		if err := w.s.Delete(deleteBlocks); err != nil {
 			return err
 		}
+		w.priceFiller.FullFillTradeLogs(insertOrders)
 		if err := w.s.Insert(insertOrders); err != nil {
 			return err
 		}
@@ -164,6 +169,7 @@ func (w *Worker) retryParseLog() error {
 		insertOrders = append(insertOrders, order)
 	}
 
+	w.priceFiller.FullFillTradeLogs(insertOrders)
 	if err := w.s.Insert(insertOrders); err != nil {
 		return err
 	}

pkg/pricefiller/flags.go

@@ -0,0 +1,22 @@
+package pricefiller
+
+import (
+	"github.com/urfave/cli"
+)
+
+var BinanceAPIKeyFlag = cli.StringFlag{
+	Name:   "binance-api-key",
+	EnvVar: "BINANCE_API_KEY",
+}
+
+var BinanceSecretKeyFlag = cli.StringFlag{
+	Name:   "binance-secret-key",
+	EnvVar: "BINANCE_SECRET_KEY",
+}
+
+func PriceFillerFlags() []cli.Flag {
+	return []cli.Flag{
+		BinanceAPIKeyFlag,
+		BinanceSecretKeyFlag,
+	}
+}

pkg/pricefiller/ks_client.go

@@ -0,0 +1,105 @@
+package pricefiller
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+const ksSettingUrl = "https://ks-setting.kyberswap.com/api/v1"
+
+type KsClient struct {
+	client  *http.Client
+	baseURL string
+}
+
+func NewKsClient() *KsClient {
+	return &KsClient{
+		client:  &http.Client{},
+		baseURL: ksSettingUrl,
+	}
+}
+
+func (c *KsClient) DoRequest(ctx context.Context, method, path string, jsonData interface{}, out interface{}) error {
+	req, err := createRequest(ctx, method, path, jsonData)
+	if err != nil {
+		return err
+	}
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	bb, err := readResponse(resp.Body, out)
+	if err != nil {
+		return fmt.Errorf("readResponse error: %w, data: %s", err, string(bb))
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("server return %d - %v", resp.StatusCode, string(bb))
+	}
+
+	return nil
+}
+
+func createRequest(ctx context.Context, method, url string, jsonData interface{}) (*http.Request, error) {
+	var buf io.Reader
+	if jsonData != nil {
+		body, err := json.Marshal(jsonData)
+		if err != nil {
+			return nil, err
+		}
+		buf = bytes.NewBuffer(body)
+	}
+	req, err := http.NewRequestWithContext(ctx, method, url, buf)
+	if err != nil {
+		return nil, err
+	}
+	if jsonData != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	return req, nil
+}
+
+func readResponse(data io.Reader, dataField interface{}) ([]byte, error) {
+	if dataField == nil {
+		return nil, fmt.Errorf("nil data")
+	}
+	bb, err := io.ReadAll(data)
+	if err != nil {
+		return nil, err
+	}
+	return bb, json.Unmarshal(bb, dataField)
+}
+
+type TokenCatalogResp struct {
+	Code    int64  `json:"code"`
+	Message string `json:"message"`
+	Data    struct {
+		Tokens []TokenCatalog `json:"tokens"`
+	}
+}
+
+type TokenCatalog struct {
+	Decimals int64 `json:"decimals"`
+}
+
+func (c *KsClient) GetTokenCatalog(address string) (TokenCatalogResp, error) {
+	var resp TokenCatalogResp
+	err := c.DoRequest(context.Background(), http.MethodGet,
+		fmt.Sprintf("%s/tokens?chainIds=%d&query=%s", c.baseURL, NetworkETHChanID, address),
+		nil, &resp)
+	if err != nil {
+		return TokenCatalogResp{}, err
+	}
+
+	if resp.Code != 0 {
+		return TokenCatalogResp{}, fmt.Errorf("invalid response code: %d", resp.Code)
+	}
+
+	return resp, nil
+}