add mark fusion for oneinchv6 parser

v2/cmd/backfill/main.go

@@ -27,7 +27,7 @@ import (
 	promotion1inchv2 "github.com/KyberNetwork/tradelogs/v2/pkg/promotionparser/oneinchv2"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/rpcnode"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/backfill"
-	promoteeTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
+	promotee_storage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/state"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs"
 	bebopStorage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/bebop"
@@ -99,7 +99,7 @@ func run(c *cli.Context) error {
 	manager := tradelogs.NewManager(l, storages)
 
 	//promotee storage
-	promoteeStorage := promoteeTypes.New(l, db)
+	promoteeStorage := promotee_storage.New(l, db)
 
 	// backfill storage
 	backfillStorage := backfill.New(l, db)
@@ -132,7 +132,7 @@ func run(c *cli.Context) error {
 		paraswap.MustNewParser(),
 		kyberswaprfq.MustNewParser(),
 		hashflowv3.MustNewParser(),
-		oneinchv6.MustNewParser(),
+		oneinchv6.MustNewParser(promoteeStorage),
 		uniswapx.MustNewParser(),
 		bebop.MustNewParser(),
 		zxrfqv3.MustNewParserWithDeployer(l, zxv3DeployStorage, ethClients[0], common.HexToAddress(constant.Deployer0xV3)),

v2/cmd/parse_log/main.go

@@ -27,7 +27,7 @@ import (
 	"github.com/KyberNetwork/tradelogs/v2/pkg/promotionparser"
 	promotion1inchv2 "github.com/KyberNetwork/tradelogs/v2/pkg/promotionparser/oneinchv2"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/rpcnode"
-	promoteeTypes "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
+	promotee_storage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/promotees"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/state"
 	"github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs"
 	bebopStorage "github.com/KyberNetwork/tradelogs/v2/pkg/storage/tradelogs/bebop"
@@ -98,7 +98,7 @@ func run(c *cli.Context) error {
 	})
 
 	//promotee storage
-	promoteeStorage := promoteeTypes.New(l, db)
+	promoteeStorage := promotee_storage.New(l, db)
 
 	// state storage
 	s := state.New(l, db)
@@ -128,7 +128,7 @@ func run(c *cli.Context) error {
 		paraswap.MustNewParser(),
 		kyberswaprfq.MustNewParser(),
 		hashflowv3.MustNewParser(),
-		oneinchv6.MustNewParser(),
+		oneinchv6.MustNewParser(promoteeStorage),
 		uniswapx.MustNewParser(),
 		bebop.MustNewParser(),
 		zxrfqv3.MustNewParserWithDeployer(l, zxv3DeployStorage, ethClients[0], common.HexToAddress(constant.Deployer0xV3)),

v2/cmd/price_filler/main.go

@@ -27,8 +27,6 @@ import (
 	"go.uber.org/zap"
 )
 
-// This week I will deploy the new price filler of tradelog v2, which calls my new mark to market.
-// After deploying, I will have data to continue creating competition dashboard.
 func main() {
 	app := libapp.NewApp()
 	app.Name = "trade logs crawler service"
@@ -80,7 +78,12 @@ func run(c *cli.Context) error {
 		l.Errorw("Error while init price filler")
 		return err
 	}
-	priceFiller.Run()
+	fillPriceInterval := c.Duration(libapp.FillPriceTimeIntervalFlag.Name)
+	if fillPriceInterval <= 0 {
+		l.Errorw("invalid value of fill-price-time-interval-flag")
+		return fmt.Errorf("invalid value of fill-price-time-interval-flag")
+	}
+	priceFiller.Run(fillPriceInterval)
 	return nil
 }
 

v2/pkg/app/price_filler.go

@@ -1,6 +1,8 @@
 package app
 
 import (
+	"time"
+
 	"github.com/urfave/cli"
 )
 
@@ -13,15 +15,23 @@ var BinanceSecretKeyFlag = cli.StringFlag{
 	Name:   "binance-secret-key",
 	EnvVar: "BINANCE_SECRET_KEY",
 }
+
 var MarkToMarketURLFlag = cli.StringFlag{
 	Name:   "mark-to-market-url",
 	EnvVar: "MARK_TO_MARKET_URL",
 }
 
+var FillPriceTimeIntervalFlag = cli.DurationFlag{
+	Name:   "fill-price-time-interval",
+	EnvVar: "FILL_PRICE_TIME_INTERVAL",
+	Value:  time.Minute,
+}
+
 func PriceFillerFlags() []cli.Flag {
 	return []cli.Flag{
 		BinanceAPIKeyFlag,
 		BinanceSecretKeyFlag,
 		MarkToMarketURLFlag,
+		FillPriceTimeIntervalFlag,
 	}
 }

v2/pkg/handler/trade_logs.go

@@ -73,12 +73,12 @@ func (h *TradeLogHandler) ProcessBlockWithExclusion(blockHash string, blockNumbe
 		return fmt.Errorf("delete blocks error: %w", err)
 	}
 
-	err = h.processForTradelog(calls, blockHash, blockNumber, timestamp, exclusions)
+	err = h.processForPromotion(calls, blockHash, blockNumber, timestamp)
 	if err != nil {
 		return fmt.Errorf("error when process block: %d", blockNumber)
 	}
 
-	err = h.processForPromotion(calls, blockHash, blockNumber, timestamp)
+	err = h.processForTradelog(calls, blockHash, blockNumber, timestamp, exclusions)
 	if err != nil {
 		return fmt.Errorf("error when process block: %d", blockNumber)
 	}

v2/pkg/mtm/client.go

@@ -2,14 +2,17 @@ package mtm
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
-
-	"github.com/KyberNetwork/tradinglib/pkg/httpclient"
 )
 
+var ErrRateLimit = errors.New("rate limit exceeded")
+
 type MtmClient struct {
 	baseURL    string
 	httpClient *http.Client
@@ -35,28 +38,24 @@ type TokenResponse struct {
 
 func (m *MtmClient) GetListTokens(ctx context.Context) ([]Token, error) {
 	const path = "/tokens"
-
-	httpReq, err := httpclient.NewRequestWithContext(
-		ctx,
-		http.MethodGet,
-		m.baseURL,
-		path,
-		nil,
-		nil)
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, m.baseURL+path, nil)
 	if err != nil {
 		return nil, fmt.Errorf("new request error: %w", err)
 	}
+	req.Header.Set("Accept", "application/json")
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("do request error: %w", err)
+	}
 
-	var resp TokenResponse
-	if _, err := httpclient.DoHTTPRequest(
-		m.httpClient,
-		httpReq,
-		&resp,
-		httpclient.WithStatusCode(http.StatusOK)); err != nil {
-		return nil, fmt.Errorf("do http request error: %w", err)
+	var tokens TokenResponse
+
+	if err := json.NewDecoder(resp.Body).Decode(&tokens); err != nil {
+		return nil, fmt.Errorf("decode error: %w", err)
 	}
 
-	return resp.Data, nil
+	return tokens.Data, nil
 }
 
 type RateV3Response struct {
@@ -71,28 +70,29 @@ func (m *MtmClient) GetHistoricalRate(
 	ctx context.Context, base, quote string, chainId int64, ts time.Time,
 ) (float64, error) {
 	const path = "/v3/historical"
-	httpReq, err := httpclient.NewRequestWithContext(
-		ctx,
-		http.MethodGet,
-		m.baseURL,
-		path,
-		httpclient.NewQuery().
-			SetString("base", base).
-			SetString("quote", quote).
-			Int64("chain_id", chainId).
-			Int64("time", ts.Unix()),
-		nil)
+	params := "?base=" + base +
+		"&quote=" + quote +
+		"&chain_id=" + strconv.FormatInt(chainId, 10) +
+		"&time=" + strconv.FormatInt(ts.Unix(), 10)
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, m.baseURL+path+params, nil)
 	if err != nil {
 		return 0, fmt.Errorf("new request error: %w", err)
 	}
+	req.Header.Set("Accept", "application/json")
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if resp.StatusCode == http.StatusTooManyRequests { // 429
+		return 0, ErrRateLimit
+	}
+	if err != nil {
+		return 0, fmt.Errorf("do request error: %w", err)
+	}
+	defer resp.Body.Close()
 
 	var rate RateV3Response
-	if _, err := httpclient.DoHTTPRequest(
-		m.httpClient,
-		httpReq,
-		&rate,
-		httpclient.WithStatusCode(http.StatusOK)); err != nil {
-		return 0, fmt.Errorf("do http request error: %w", err)
+
+	if err := json.NewDecoder(resp.Body).Decode(&rate); err != nil {
+		return 0, fmt.Errorf("decode error: %w", err)
 	}
 
 	return rate.Data.Price, nil

v2/pkg/mtm/client_test.go

@@ -2,7 +2,6 @@ package mtm
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"testing"
 	"time"
@@ -16,10 +15,11 @@ func TestNewMtmClient(t *testing.T) {
 	MTM_URL := ""
 	httpClient := &http.Client{}
 	client := NewMtmClient(MTM_URL, httpClient)
-	rate, err := client.GetHistoricalRate(context.Background(), "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", "0xdac17f958d2ee523a2206206994597c13d831ec7", 1, time.UnixMilli(1732870268000))
+	rate, err := client.GetHistoricalRate(context.Background(), "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", "0xdac17f958d2ee523a2206206994597c13d831ec7", 1, time.Now().Add(-time.Hour))
 	require.NoError(t, err)
-	fmt.Println("historical rate", rate)
+	t.Log("historical rate", rate)
 
-	_, err = client.GetListTokens(context.Background())
+	tokens, err := client.GetListTokens(context.Background())
 	require.NoError(t, err)
+	t.Log("tokens", tokens)
 }