mirror of
https://github.com/status-im/status-go.git
synced 2025-01-21 12:11:44 +00:00
feat: convert history balance tokens to fiat
Add history.exchange sub-package with following responsibilities
- fetch and caches daily exchange rates for tokens
- Partial update if missing form yesterday back
- Implement all time fetching special case
- Fetches the price of the token after merging entries for the selected
chains
history.service package changes
- Return ValuePoint instead of DataPoint
- Value point contains the value in fiat as float64 instead
- Convert merged values from tokens balance (wei) to fiat
Other changes
- Move start/stop balance history to startWallet/stopWallet
- Move cryptocompare to thirdparty package to avoid recursive import
- Rename DataPoint.Value to DataPoint.Balance
- Don't merge entries for single chain
This commit is contained in:
Stefan
Stefan Dunca
parent
0b2f0ef289
commit
f4f6b25302
@ -19,6 +19,7 @@ import (
|
||||
"github.com/status-im/status-go/services/wallet/bridge"
|
||||
"github.com/status-im/status-go/services/wallet/chain"
|
||||
"github.com/status-im/status-go/services/wallet/history"
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
"github.com/status-im/status-go/services/wallet/token"
|
||||
"github.com/status-im/status-go/services/wallet/transfer"
|
||||
)
|
||||
@ -129,27 +130,15 @@ func (api *API) GetTokensBalancesForChainIDs(ctx context.Context, chainIDs []uin
|
||||
return api.s.tokenManager.GetBalances(ctx, clients, accounts, addresses)
|
||||
}
|
||||
|
||||
func (api *API) StartBalanceHistory(ctx context.Context) error {
|
||||
api.s.history.StartBalanceHistory()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (api *API) StopBalanceHistory(ctx context.Context) error {
|
||||
api.s.history.Stop()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (api *API) UpdateVisibleTokens(ctx context.Context, symbols []string) error {
|
||||
api.s.history.UpdateVisibleTokens(symbols)
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetBalanceHistory retrieves token balance history for token identity on multiple chains
|
||||
// TODO: pass parameters by GetBalanceHistoryParameters struct
|
||||
// TODO: expose endTimestamp parameter
|
||||
func (api *API) GetBalanceHistory(ctx context.Context, chainIDs []uint64, address common.Address, currency string, timeInterval history.TimeInterval) ([]*history.DataPoint, error) {
|
||||
func (api *API) GetBalanceHistory(ctx context.Context, chainIDs []uint64, address common.Address, tokenSymbol string, currencySymbol string, timeInterval history.TimeInterval) ([]*history.ValuePoint, error) {
|
||||
endTimestamp := time.Now().UTC().Unix()
|
||||
return api.s.history.GetBalanceHistory(ctx, chainIDs, address, currency, endTimestamp, timeInterval)
|
||||
return api.s.history.GetBalanceHistory(ctx, chainIDs, address, tokenSymbol, currencySymbol, endTimestamp, timeInterval)
|
||||
}
|
||||
|
||||
func (api *API) GetTokens(ctx context.Context, chainID uint64) ([]*token.Token, error) {
|
||||
@ -353,24 +342,24 @@ func (api *API) GetCachedPrices(ctx context.Context) (map[string]map[string]floa
|
||||
return api.s.priceManager.GetCachedPrices()
|
||||
}
|
||||
|
||||
func (api *API) FetchMarketValues(ctx context.Context, symbols []string, currencies []string) (map[string]map[string]MarketCoinValues, error) {
|
||||
func (api *API) FetchMarketValues(ctx context.Context, symbols []string, currencies []string) (map[string]map[string]thirdparty.MarketCoinValues, error) {
|
||||
log.Debug("call to FetchMarketValues")
|
||||
return api.s.cryptoCompare.fetchTokenMarketValues(symbols, currencies)
|
||||
return api.s.cryptoCompare.FetchTokenMarketValues(symbols, currencies)
|
||||
}
|
||||
|
||||
func (api *API) GetHourlyMarketValues(ctx context.Context, symbol string, currency string, limit int, aggregate int) ([]TokenHistoricalPairs, error) {
|
||||
func (api *API) GetHourlyMarketValues(ctx context.Context, symbol string, currency string, limit int, aggregate int) ([]thirdparty.TokenHistoricalPairs, error) {
|
||||
log.Debug("call to GetHourlyMarketValues")
|
||||
return api.s.cryptoCompare.fetchHourlyMarketValues(symbol, currency, limit, aggregate)
|
||||
return api.s.cryptoCompare.FetchHourlyMarketValues(symbol, currency, limit, aggregate)
|
||||
}
|
||||
|
||||
func (api *API) GetDailyMarketValues(ctx context.Context, symbol string, currency string, limit int, allData bool, aggregate int) ([]TokenHistoricalPairs, error) {
|
||||
func (api *API) GetDailyMarketValues(ctx context.Context, symbol string, currency string, limit int, allData bool, aggregate int) ([]thirdparty.TokenHistoricalPairs, error) {
|
||||
log.Debug("call to GetDailyMarketValues")
|
||||
return api.s.cryptoCompare.fetchDailyMarketValues(symbol, currency, limit, allData, aggregate)
|
||||
return api.s.cryptoCompare.FetchDailyMarketValues(symbol, currency, limit, allData, aggregate)
|
||||
}
|
||||
|
||||
func (api *API) FetchTokenDetails(ctx context.Context, symbols []string) (map[string]Coin, error) {
|
||||
func (api *API) FetchTokenDetails(ctx context.Context, symbols []string) (map[string]thirdparty.Coin, error) {
|
||||
log.Debug("call to FetchTokenDetails")
|
||||
return api.s.cryptoCompare.fetchTokenDetails(symbols)
|
||||
return api.s.cryptoCompare.FetchTokenDetails(symbols)
|
||||
}
|
||||
|
||||
func (api *API) GetSuggestedFees(ctx context.Context, chainID uint64) (*SuggestedFees, error) {
|
||||
|
||||
@ -87,9 +87,9 @@ type DataSource interface {
|
||||
}
|
||||
|
||||
type DataPoint struct {
|
||||
Value *hexutil.Big `json:"value"`
|
||||
Timestamp uint64 `json:"time"`
|
||||
BlockNumber *hexutil.Big `json:"blockNumber"`
|
||||
Balance *hexutil.Big
|
||||
Timestamp uint64
|
||||
BlockNumber *hexutil.Big
|
||||
}
|
||||
|
||||
func strideDuration(timeInterval TimeInterval) time.Duration {
|
||||
@ -113,7 +113,7 @@ func (b *Balance) fetchAndCache(ctx context.Context, source DataSource, address
|
||||
return nil, nil, err
|
||||
}
|
||||
return &DataPoint{
|
||||
Value: (*hexutil.Big)(cached[0].balance),
|
||||
Balance: (*hexutil.Big)(cached[0].balance),
|
||||
Timestamp: uint64(cached[0].timestamp),
|
||||
BlockNumber: (*hexutil.Big)(cached[0].block),
|
||||
}, blockNo, nil
|
||||
@ -156,7 +156,7 @@ func (b *Balance) fetchAndCache(ctx context.Context, source DataSource, address
|
||||
}
|
||||
|
||||
var dataPoint DataPoint
|
||||
dataPoint.Value = (*hexutil.Big)(currentBalance)
|
||||
dataPoint.Balance = (*hexutil.Big)(currentBalance)
|
||||
dataPoint.Timestamp = uint64(timestamp)
|
||||
return &dataPoint, blockNo, nil
|
||||
}
|
||||
@ -241,7 +241,7 @@ func (b *Balance) get(ctx context.Context, chainID uint64, currency string, addr
|
||||
points := make([]*DataPoint, 0, len(cached)+1)
|
||||
for _, entry := range cached {
|
||||
dataPoint := DataPoint{
|
||||
Value: (*hexutil.Big)(entry.balance),
|
||||
Balance: (*hexutil.Big)(entry.balance),
|
||||
Timestamp: uint64(entry.timestamp),
|
||||
BlockNumber: (*hexutil.Big)(entry.block),
|
||||
}
|
||||
@ -254,7 +254,7 @@ func (b *Balance) get(ctx context.Context, chainID uint64, currency string, addr
|
||||
}
|
||||
if len(lastCached) > 0 && len(cached) > 0 && lastCached[0].block.Cmp(cached[len(cached)-1].block) > 0 {
|
||||
points = append(points, &DataPoint{
|
||||
Value: (*hexutil.Big)(lastCached[0].balance),
|
||||
Balance: (*hexutil.Big)(lastCached[0].balance),
|
||||
Timestamp: uint64(lastCached[0].timestamp),
|
||||
BlockNumber: (*hexutil.Big)(lastCached[0].block),
|
||||
})
|
||||
|
||||
@ -497,7 +497,7 @@ func TestBalanceHistoryValidateBalanceValuesAndCacheHit(t *testing.T) {
|
||||
n := reqBlkNos[i]
|
||||
|
||||
if value, contains := requestedBalance[n]; contains {
|
||||
require.Equal(t, value.Cmp(balanceData[resIdx].Value.ToInt()), 0)
|
||||
require.Equal(t, value.Cmp(balanceData[resIdx].Balance.ToInt()), 0)
|
||||
resIdx++
|
||||
}
|
||||
blockHeaderRequestCount := dataSource.requestedBlocks[n].headerInfoRequests
|
||||
@ -508,7 +508,7 @@ func TestBalanceHistoryValidateBalanceValuesAndCacheHit(t *testing.T) {
|
||||
|
||||
// Check that balance values are in order
|
||||
for i := 1; i < len(balanceData); i++ {
|
||||
require.Greater(t, balanceData[i].Value.ToInt().Cmp(balanceData[i-1].Value.ToInt()), 0, "expected balanceData[%d] > balanceData[%d] for interval %d", i, i-1, testInput.interval)
|
||||
require.Greater(t, balanceData[i].Balance.ToInt().Cmp(balanceData[i-1].Balance.ToInt()), 0, "expected balanceData[%d] > balanceData[%d] for interval %d", i, i-1, testInput.interval)
|
||||
}
|
||||
requestedBalance = make(map[int64]*big.Int)
|
||||
})
|
||||
|
||||
176
services/wallet/history/exchange.go
Normal file
176
services/wallet/history/exchange.go
Normal file
@ -0,0 +1,176 @@
|
||||
package history
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
)
|
||||
|
||||
type tokenType = string
|
||||
type currencyType = string
|
||||
type yearType = int
|
||||
|
||||
type allTimeEntry struct {
|
||||
value float32
|
||||
startTimestamp int64
|
||||
endTimestamp int64
|
||||
}
|
||||
|
||||
// Exchange caches conversion rates in memory on a daily basis
|
||||
type Exchange struct {
|
||||
// year map keeps a list of values with days as index in the slice for the corresponding year (key) starting from the first to the last available
|
||||
cache map[tokenType]map[currencyType]map[yearType][]float32
|
||||
// special case for all time information
|
||||
allTimeCache map[tokenType]map[currencyType][]allTimeEntry
|
||||
fetchMutex sync.Mutex
|
||||
|
||||
cryptoCompare *thirdparty.CryptoCompare
|
||||
}
|
||||
|
||||
func NewExchange(cryptoCompare *thirdparty.CryptoCompare) *Exchange {
|
||||
return &Exchange{
|
||||
cache: make(map[tokenType]map[currencyType]map[yearType][]float32),
|
||||
cryptoCompare: cryptoCompare,
|
||||
}
|
||||
}
|
||||
|
||||
// GetExchangeRate returns the exchange rate from token to currency in the day of the given date
|
||||
// if none exists returns "missing <element>" error
|
||||
func (e *Exchange) GetExchangeRateForDay(token tokenType, currency currencyType, date time.Time) (float32, error) {
|
||||
e.fetchMutex.Lock()
|
||||
defer e.fetchMutex.Unlock()
|
||||
|
||||
currencyMap, found := e.cache[token]
|
||||
if !found {
|
||||
return 0, errors.New("missing token")
|
||||
}
|
||||
|
||||
yearsMap, found := currencyMap[currency]
|
||||
if !found {
|
||||
return 0, errors.New("missing currency")
|
||||
}
|
||||
|
||||
year := date.Year()
|
||||
valueForDays, found := yearsMap[year]
|
||||
if !found {
|
||||
// Search closest in all time
|
||||
allCurrencyMap, found := e.allTimeCache[token]
|
||||
if !found {
|
||||
return 0, errors.New("missing token in all time data")
|
||||
}
|
||||
|
||||
allYearsMap, found := allCurrencyMap[currency]
|
||||
if !found {
|
||||
return 0, errors.New("missing currency in all time data")
|
||||
}
|
||||
for _, entry := range allYearsMap {
|
||||
if entry.startTimestamp <= date.Unix() && entry.endTimestamp > date.Unix() {
|
||||
return entry.value, nil
|
||||
}
|
||||
}
|
||||
return 0, errors.New("missing entry")
|
||||
}
|
||||
|
||||
day := date.YearDay()
|
||||
if day >= len(valueForDays) {
|
||||
return 0, errors.New("missing day")
|
||||
}
|
||||
return valueForDays[day], nil
|
||||
}
|
||||
|
||||
// fetchAndCacheRates fetches and in memory cache exchange rates for this and last year
|
||||
func (e *Exchange) FetchAndCacheMissingRates(token tokenType, currency currencyType) error {
|
||||
// Protect REST calls also to prevent fetching the same token/currency twice
|
||||
e.fetchMutex.Lock()
|
||||
defer e.fetchMutex.Unlock()
|
||||
|
||||
// Allocate missing values
|
||||
currencyMap, found := e.cache[token]
|
||||
if !found {
|
||||
currencyMap = make(map[currencyType]map[yearType][]float32)
|
||||
e.cache[token] = currencyMap
|
||||
}
|
||||
|
||||
yearsMap, found := currencyMap[currency]
|
||||
if !found {
|
||||
yearsMap = make(map[yearType][]float32)
|
||||
currencyMap[currency] = yearsMap
|
||||
}
|
||||
|
||||
currentTime := time.Now().UTC()
|
||||
endOfPrevYearTime := time.Date(currentTime.Year()-1, 12, 31, 23, 0, 0, 0, time.UTC)
|
||||
|
||||
daysToFetch := extendDaysSliceForYear(yearsMap, endOfPrevYearTime)
|
||||
|
||||
curYearTime := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), 0, 0, 0, 0, time.UTC)
|
||||
daysToFetch += extendDaysSliceForYear(yearsMap, curYearTime)
|
||||
if daysToFetch == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
res, err := e.cryptoCompare.FetchDailyMarketValues(token, currency, daysToFetch, false, 1)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for i := 0; i < len(res); i++ {
|
||||
t := time.Unix(res[i].Timestamp, 0).UTC()
|
||||
yearDayIndex := t.YearDay() - 1
|
||||
yearValues, found := yearsMap[t.Year()]
|
||||
if found && yearDayIndex < len(yearValues) {
|
||||
yearValues[yearDayIndex] = float32(res[i].Value)
|
||||
}
|
||||
}
|
||||
|
||||
// Fetch all time
|
||||
allTime, err := e.cryptoCompare.FetchDailyMarketValues(token, currency, 1, true, 30)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if e.allTimeCache == nil {
|
||||
e.allTimeCache = make(map[tokenType]map[currencyType][]allTimeEntry)
|
||||
}
|
||||
_, found = e.allTimeCache[token]
|
||||
if !found {
|
||||
e.allTimeCache[token] = make(map[currencyType][]allTimeEntry)
|
||||
}
|
||||
|
||||
// No benefit to fetch intermendiate values, overwrite historical
|
||||
e.allTimeCache[token][currency] = make([]allTimeEntry, 0)
|
||||
|
||||
for i := 0; i < len(allTime) && allTime[i].Timestamp < res[0].Timestamp; i++ {
|
||||
if allTime[i].Value > 0 {
|
||||
var endTimestamp int64
|
||||
if i+1 < len(allTime) {
|
||||
endTimestamp = allTime[i+1].Timestamp
|
||||
} else {
|
||||
endTimestamp = res[0].Timestamp
|
||||
}
|
||||
e.allTimeCache[token][currency] = append(e.allTimeCache[token][currency],
|
||||
allTimeEntry{
|
||||
value: float32(allTime[i].Value),
|
||||
startTimestamp: allTime[i].Timestamp,
|
||||
endTimestamp: endTimestamp,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func extendDaysSliceForYear(yearsMap map[yearType][]float32, untilTime time.Time) (daysToFetch int) {
|
||||
year := untilTime.Year()
|
||||
_, found := yearsMap[year]
|
||||
if !found {
|
||||
yearsMap[year] = make([]float32, untilTime.YearDay())
|
||||
return untilTime.YearDay()
|
||||
}
|
||||
|
||||
// Just extend the slice if needed
|
||||
missingDays := untilTime.YearDay() - len(yearsMap[year])
|
||||
yearsMap[year] = append(yearsMap[year], make([]float32, missingDays)...)
|
||||
return missingDays
|
||||
}
|
||||
@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"math"
|
||||
"math/big"
|
||||
"sort"
|
||||
"sync"
|
||||
@ -21,6 +22,7 @@ import (
|
||||
"github.com/status-im/status-go/rpc/network"
|
||||
|
||||
"github.com/status-im/status-go/services/wallet/chain"
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
"github.com/status-im/status-go/services/wallet/token"
|
||||
"github.com/status-im/status-go/services/wallet/walletevent"
|
||||
)
|
||||
@ -44,14 +46,16 @@ type Service struct {
|
||||
serviceContext context.Context
|
||||
cancelFn context.CancelFunc
|
||||
|
||||
exchange *Exchange
|
||||
|
||||
timer *time.Timer
|
||||
visibleTokenSymbols []string
|
||||
visibleTokenSymbolsMutex sync.Mutex // Protects access to visibleSymbols
|
||||
visibleTokenSymbolsMutex sync.Mutex
|
||||
}
|
||||
|
||||
type chainIdentity uint64
|
||||
|
||||
func NewService(db *sql.DB, eventFeed *event.Feed, rpcClient *statusrpc.Client, tokenManager *token.Manager) *Service {
|
||||
func NewService(db *sql.DB, eventFeed *event.Feed, rpcClient *statusrpc.Client, tokenManager *token.Manager, cryptoCompare *thirdparty.CryptoCompare) *Service {
|
||||
return &Service{
|
||||
balance: NewBalance(NewBalanceDB(db)),
|
||||
db: db,
|
||||
@ -59,6 +63,7 @@ func NewService(db *sql.DB, eventFeed *event.Feed, rpcClient *statusrpc.Client,
|
||||
rpcClient: rpcClient,
|
||||
networkManager: rpcClient.NetworkManager,
|
||||
tokenManager: tokenManager,
|
||||
exchange: NewExchange(cryptoCompare),
|
||||
}
|
||||
}
|
||||
|
||||
@ -78,7 +83,7 @@ func (s *Service) triggerEvent(eventType walletevent.EventType, account statusty
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Service) StartBalanceHistory() {
|
||||
func (s *Service) Start() {
|
||||
go func() {
|
||||
s.serviceContext, s.cancelFn = context.WithCancel(context.Background())
|
||||
s.timer = time.NewTimer(balanceHistoryUpdateInterval)
|
||||
@ -198,40 +203,125 @@ func (src *tokenChainClientSource) BalanceAt(ctx context.Context, account common
|
||||
return balance, err
|
||||
}
|
||||
|
||||
type ValuePoint struct {
|
||||
Value float64 `json:"value"`
|
||||
Timestamp uint64 `json:"time"`
|
||||
BlockNumber *hexutil.Big `json:"blockNumber"`
|
||||
}
|
||||
|
||||
// GetBalanceHistory returns token count balance
|
||||
// TODO: fetch token to FIAT exchange rates and return FIAT balance
|
||||
func (s *Service) GetBalanceHistory(ctx context.Context, chainIDs []uint64, address common.Address, currency string, endTimestamp int64, timeInterval TimeInterval) ([]*DataPoint, error) {
|
||||
func (s *Service) GetBalanceHistory(ctx context.Context, chainIDs []uint64, address common.Address, tokenSymbol string, currencySymbol string, endTimestamp int64, timeInterval TimeInterval) ([]*ValuePoint, error) {
|
||||
// Retrieve cached data for all chains
|
||||
allData := make(map[chainIdentity][]*DataPoint)
|
||||
for _, chainID := range chainIDs {
|
||||
data, err := s.balance.get(ctx, chainID, currency, address, endTimestamp, timeInterval)
|
||||
data, err := s.balance.get(ctx, chainID, tokenSymbol, address, endTimestamp, timeInterval)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(data) > 0 {
|
||||
allData[chainIdentity(chainID)] = data
|
||||
} else {
|
||||
return make([]*ValuePoint, 0), nil
|
||||
}
|
||||
}
|
||||
|
||||
return mergeDataPoints(allData, strideDuration(timeInterval))
|
||||
data, err := mergeDataPoints(allData, strideDuration(timeInterval))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if len(data) == 0 {
|
||||
return make([]*ValuePoint, 0), nil
|
||||
}
|
||||
|
||||
// Check if historical exchange rate for data point is present and fetch remaining if not
|
||||
lastDayTime := time.Unix(int64(data[len(data)-1].Timestamp), 0).UTC()
|
||||
currentTime := time.Now().UTC()
|
||||
currentDayStart := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), 0, 0, 0, 0, time.UTC)
|
||||
if lastDayTime.After(currentDayStart) {
|
||||
// No chance to have today, use the previous day value for the last data point
|
||||
lastDayTime = lastDayTime.AddDate(0, 0, -1)
|
||||
}
|
||||
|
||||
_, err = s.exchange.GetExchangeRateForDay(tokenSymbol, currencySymbol, lastDayTime)
|
||||
if err != nil {
|
||||
err := s.exchange.FetchAndCacheMissingRates(tokenSymbol, currencySymbol)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
decimals, err := s.decimalsForToken(tokenSymbol, chainIDs[0])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
weisInOneMain := big.NewFloat(math.Pow(10, float64(decimals)))
|
||||
|
||||
var res []*ValuePoint
|
||||
for _, d := range data {
|
||||
dayTime := time.Unix(int64(d.Timestamp), 0).UTC()
|
||||
if dayTime.After(currentDayStart) {
|
||||
// No chance to have today, use the previous day value for the last data point
|
||||
dayTime = lastDayTime
|
||||
}
|
||||
dayValue, err := s.exchange.GetExchangeRateForDay(tokenSymbol, currencySymbol, dayTime)
|
||||
if err != nil {
|
||||
log.Warn("Echange rate missing for", dayTime, "- err", err)
|
||||
continue
|
||||
}
|
||||
|
||||
// The big.Int values are discarded, hence copy the original values
|
||||
res = append(res, &ValuePoint{
|
||||
Timestamp: d.Timestamp,
|
||||
Value: tokenToValue((*big.Int)(d.Balance), dayValue, weisInOneMain),
|
||||
BlockNumber: d.BlockNumber,
|
||||
})
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (s *Service) decimalsForToken(tokenSymbol string, chainID uint64) (int, error) {
|
||||
network := s.networkManager.Find(chainID)
|
||||
if network == nil {
|
||||
return 0, errors.New("network not found")
|
||||
}
|
||||
token := s.tokenManager.FindToken(network, tokenSymbol)
|
||||
if token == nil {
|
||||
return 0, errors.New("token not found")
|
||||
}
|
||||
return int(token.Decimals), nil
|
||||
}
|
||||
|
||||
func tokenToValue(tokenCount *big.Int, mainDenominationValue float32, weisInOneMain *big.Float) float64 {
|
||||
weis := new(big.Float).SetInt(tokenCount)
|
||||
mainTokens := new(big.Float).Quo(weis, weisInOneMain)
|
||||
mainTokenValue := new(big.Float).SetFloat64(float64(mainDenominationValue))
|
||||
res, accuracy := new(big.Float).Mul(mainTokens, mainTokenValue).Float64()
|
||||
if res == 0 && accuracy == big.Below {
|
||||
return math.SmallestNonzeroFloat64
|
||||
} else if res == math.Inf(1) && accuracy == big.Above {
|
||||
return math.Inf(1)
|
||||
}
|
||||
|
||||
return res
|
||||
}
|
||||
|
||||
// mergeDataPoints merges close in time block numbers. Drops the ones that are not in a stride duration
|
||||
// this should improve merging balance data from different chains which are incompatible due to different timelines
|
||||
// and block length
|
||||
func mergeDataPoints(data map[chainIdentity][]*DataPoint, stride time.Duration) ([]*DataPoint, error) {
|
||||
// Special cases
|
||||
if len(data) == 0 {
|
||||
return make([]*DataPoint, 0), nil
|
||||
}
|
||||
|
||||
pos := make(map[chainIdentity]int)
|
||||
for k := range data {
|
||||
pos[k] = 0
|
||||
} else if len(data) == 1 {
|
||||
for k := range data {
|
||||
return data[k], nil
|
||||
}
|
||||
}
|
||||
|
||||
res := make([]*DataPoint, 0)
|
||||
strideStart := findFirstStrideWindow(data, stride)
|
||||
strideStart, pos := findFirstStrideWindow(data, stride)
|
||||
for {
|
||||
strideEnd := strideStart + int64(stride.Seconds())
|
||||
|
||||
// - Gather all points in the stride window starting with current pos
|
||||
var strideIdentities map[chainIdentity][]timeIdentity
|
||||
strideIdentities, pos = dataInStrideWindowAndNextPos(data, pos, strideEnd)
|
||||
@ -249,7 +339,7 @@ func mergeDataPoints(data map[chainIdentity][]*DataPoint, stride time.Duration)
|
||||
for chainID, identities := range strideIdentities {
|
||||
for _, identity := range identities {
|
||||
_, exists := chainMaxBalance[chainID]
|
||||
if exists && (*big.Int)(identity.dataPoint(data).Value).Cmp((*big.Int)(chainMaxBalance[chainID].Value)) <= 0 {
|
||||
if exists && (*big.Int)(identity.dataPoint(data).Balance).Cmp((*big.Int)(chainMaxBalance[chainID].Balance)) <= 0 {
|
||||
continue
|
||||
}
|
||||
chainMaxBalance[chainID] = identity.dataPoint(data)
|
||||
@ -257,11 +347,17 @@ func mergeDataPoints(data map[chainIdentity][]*DataPoint, stride time.Duration)
|
||||
}
|
||||
balance := big.NewInt(0)
|
||||
for _, chainBalance := range chainMaxBalance {
|
||||
balance.Add(balance, (*big.Int)(chainBalance.Value))
|
||||
balance.Add(balance, (*big.Int)(chainBalance.Balance))
|
||||
}
|
||||
|
||||
// if last stride, the timestamp might be in the future
|
||||
if strideEnd > time.Now().UTC().Unix() {
|
||||
strideEnd = time.Now().UTC().Unix()
|
||||
}
|
||||
|
||||
res = append(res, &DataPoint{
|
||||
Timestamp: uint64(strideEnd),
|
||||
Value: (*hexutil.Big)(balance),
|
||||
Balance: (*hexutil.Big)(balance),
|
||||
BlockNumber: (*hexutil.Big)(getBlockID(chainMaxBalance)),
|
||||
})
|
||||
}
|
||||
@ -313,19 +409,17 @@ func allPastEnd(data map[chainIdentity][]*DataPoint, pos map[chainIdentity]int)
|
||||
return true
|
||||
}
|
||||
|
||||
// findFirstStrideWindow returns the start of the first stride window
|
||||
// Tried to implement finding an optimal stride window but it was becoming too complicated and not worth it given that it will
|
||||
// potentially save the first and last stride but it is not guaranteed. Current implementation should give good results
|
||||
// as long as the the DataPoints are regular enough
|
||||
func findFirstStrideWindow(data map[chainIdentity][]*DataPoint, stride time.Duration) int64 {
|
||||
pos := make(map[chainIdentity]int)
|
||||
// findFirstStrideWindow returns the start of the first stride window (timestamp and all positions)
|
||||
//
|
||||
// Note: tried to implement finding an optimal stride window but it was becoming too complicated and not worth it given that it will potentially save the first and last stride but it is not guaranteed. Current implementation should give good results as long as the the DataPoints are regular enough
|
||||
func findFirstStrideWindow(data map[chainIdentity][]*DataPoint, stride time.Duration) (firstTimestamp int64, pos map[chainIdentity]int) {
|
||||
pos = make(map[chainIdentity]int)
|
||||
for k := range data {
|
||||
pos[k] = 0
|
||||
}
|
||||
|
||||
// Identify the current oldest and newest block
|
||||
cur := sortTimeAsc(data, pos)
|
||||
return int64(cur[0].dataPoint(data).Timestamp)
|
||||
return int64(cur[0].dataPoint(data).Timestamp), pos
|
||||
}
|
||||
|
||||
func copyMap[K comparable, V any](original map[K]V) map[K]V {
|
||||
|
||||
@ -1,15 +1,50 @@
|
||||
package history
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"math/big"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/common/hexutil"
|
||||
gethrpc "github.com/ethereum/go-ethereum/rpc"
|
||||
|
||||
"github.com/golang/mock/gomock"
|
||||
|
||||
"github.com/status-im/status-go/appdatabase"
|
||||
"github.com/status-im/status-go/params"
|
||||
statusRPC "github.com/status-im/status-go/rpc"
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
"github.com/status-im/status-go/transactions/fake"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func setupDummyServiceNoDependencies(t *testing.T) (service *Service, closeFn func()) {
|
||||
db, err := appdatabase.InitializeDB(":memory:", "wallet-history-service-tests", 1)
|
||||
require.NoError(t, err)
|
||||
cryptoCompare := thirdparty.NewCryptoCompare()
|
||||
|
||||
// Creating a dummy status node to simulate what it's done in get_status_node.go
|
||||
upstreamConfig := params.UpstreamRPCConfig{
|
||||
URL: "https://mainnet.infura.io/v3/800c641949d64d768a5070a1b0511938",
|
||||
Enabled: true,
|
||||
}
|
||||
|
||||
txServiceMockCtrl := gomock.NewController(t)
|
||||
server, _ := fake.NewTestServer(txServiceMockCtrl)
|
||||
client := gethrpc.DialInProc(server)
|
||||
|
||||
rpcClient, err := statusRPC.NewClient(client, 1, upstreamConfig, nil, db)
|
||||
require.NoError(t, err)
|
||||
|
||||
return NewService(db, nil, rpcClient, nil, cryptoCompare), func() {
|
||||
require.NoError(t, db.Close())
|
||||
}
|
||||
}
|
||||
|
||||
type TestDataPoint struct {
|
||||
value int64
|
||||
timestamp uint64
|
||||
@ -29,7 +64,7 @@ func prepareTestData(data []TestDataPoint) map[chainIdentity][]*DataPoint {
|
||||
res[entry.chainID] = append(res[entry.chainID], &DataPoint{
|
||||
BlockNumber: (*hexutil.Big)(big.NewInt(data[i].blockNumber)),
|
||||
Timestamp: data[i].timestamp,
|
||||
Value: (*hexutil.Big)(big.NewInt(data[i].value)),
|
||||
Balance: (*hexutil.Big)(big.NewInt(data[i].value)),
|
||||
})
|
||||
}
|
||||
return res
|
||||
@ -51,7 +86,7 @@ func getBlockNumbers(data []*DataPoint) []int64 {
|
||||
func getValues(data []*DataPoint) []int64 {
|
||||
res := make([]int64, 0)
|
||||
for _, entry := range data {
|
||||
res = append(res, entry.Value.ToInt().Int64())
|
||||
res = append(res, entry.Balance.ToInt().Int64())
|
||||
}
|
||||
return res
|
||||
}
|
||||
@ -148,7 +183,7 @@ func TestServiceMergeDataPointsOneChain(t *testing.T) {
|
||||
require.Equal(t, 3, len(res))
|
||||
require.Equal(t, []int64{105, 115, 125}, getBlockNumbers(res))
|
||||
require.Equal(t, []int64{1, 2, 3}, getValues(res))
|
||||
require.Equal(t, []int64{115, 125, 135}, getTimestamps(res))
|
||||
require.Equal(t, []int64{105, 115, 125}, getTimestamps(res), "Expect no merging for one chain")
|
||||
}
|
||||
|
||||
func TestServiceMergeDataPointsDropAll(t *testing.T) {
|
||||
@ -185,8 +220,9 @@ func TestServiceFindFirstStrideWindowFirstForAllChainInOneStride(t *testing.T) {
|
||||
{value: 1, timestamp: 110, blockNumber: 103, chainID: 2},
|
||||
})
|
||||
|
||||
startTimestamp := findFirstStrideWindow(testData, strideDuration)
|
||||
startTimestamp, pos := findFirstStrideWindow(testData, strideDuration)
|
||||
require.Equal(t, testData[1][0].Timestamp, uint64(startTimestamp))
|
||||
require.Equal(t, map[chainIdentity]int{1: 0, 2: 0, 3: 0}, pos)
|
||||
}
|
||||
|
||||
func TestServiceSortTimeAsc(t *testing.T) {
|
||||
@ -214,3 +250,32 @@ func TestServiceAtEnd(t *testing.T) {
|
||||
sorted = sortTimeAsc(testData, map[chainIdentity]int{1: 1, 2: 0})
|
||||
require.True(t, sorted[1].atEnd(testData))
|
||||
}
|
||||
|
||||
func TestServiceTokenToValue(t *testing.T) {
|
||||
weisInOneMain := big.NewFloat(math.Pow(10, 18.0))
|
||||
res := tokenToValue(big.NewInt(12345), 1000, weisInOneMain)
|
||||
require.Equal(t, 0.000000000012345, res)
|
||||
|
||||
in, ok := new(big.Int).SetString("1234567890000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", 10)
|
||||
require.True(t, ok)
|
||||
res = tokenToValue(in, 10000, weisInOneMain)
|
||||
require.Equal(t, 1.23456789e+112, res)
|
||||
|
||||
res = tokenToValue(big.NewInt(1000000000000000000), 1.0, weisInOneMain)
|
||||
require.Equal(t, 1.0, res)
|
||||
|
||||
res = tokenToValue(big.NewInt(1), 1.23456789, weisInOneMain)
|
||||
require.InEpsilonf(t, 1.23456789e-18, res, 1.0e-8, "Expects error for handling such low values")
|
||||
|
||||
res = tokenToValue(new(big.Int).Exp(big.NewInt(10), big.NewInt(254), nil), 100000, weisInOneMain)
|
||||
require.Equal(t, 1e+241, res, "Expect exponent 254-18+5")
|
||||
}
|
||||
|
||||
func TestServiceGetBalanceHistoryNoData(t *testing.T) {
|
||||
service, closeFn := setupDummyServiceNoDependencies(t)
|
||||
defer closeFn()
|
||||
|
||||
res, err := service.GetBalanceHistory(context.Background(), []uint64{777}, common.HexToAddress(`0x1`), "ETH", "EUR", time.Now().Unix(), BalanceHistory1Year)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 0, len(res))
|
||||
}
|
||||
|
||||
@ -4,21 +4,23 @@ import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
)
|
||||
|
||||
type PricesPerTokenAndCurrency = map[string]map[string]float64
|
||||
|
||||
type PriceManager struct {
|
||||
db *sql.DB
|
||||
cryptoCompare *CryptoCompare
|
||||
cryptoCompare *thirdparty.CryptoCompare
|
||||
}
|
||||
|
||||
func NewPriceManager(db *sql.DB, cryptoCompare *CryptoCompare) *PriceManager {
|
||||
func NewPriceManager(db *sql.DB, cryptoCompare *thirdparty.CryptoCompare) *PriceManager {
|
||||
return &PriceManager{db: db, cryptoCompare: cryptoCompare}
|
||||
}
|
||||
|
||||
func (pm *PriceManager) FetchPrices(symbols []string, currencies []string) (PricesPerTokenAndCurrency, error) {
|
||||
result, err := pm.cryptoCompare.fetchPrices(symbols, currencies)
|
||||
result, err := pm.cryptoCompare.FetchPrices(symbols, currencies)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@ -6,12 +6,13 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/status-im/status-go/appdatabase"
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
)
|
||||
|
||||
func setupTestPriceDB(t *testing.T) (*PriceManager, func()) {
|
||||
db, err := appdatabase.InitializeDB(":memory:", "wallet-price-tests-", 1)
|
||||
require.NoError(t, err)
|
||||
return NewPriceManager(db, NewCryptoCompare()), func() {
|
||||
return NewPriceManager(db, thirdparty.NewCryptoCompare()), func() {
|
||||
require.NoError(t, db.Close())
|
||||
}
|
||||
}
|
||||
|
||||
@ -13,6 +13,7 @@ import (
|
||||
"github.com/status-im/status-go/rpc"
|
||||
"github.com/status-im/status-go/services/wallet/async"
|
||||
"github.com/status-im/status-go/services/wallet/chain"
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
"github.com/status-im/status-go/services/wallet/token"
|
||||
"github.com/status-im/status-go/services/wallet/walletevent"
|
||||
)
|
||||
@ -25,7 +26,7 @@ func getFixedCurrencies() []string {
|
||||
return []string{"USD"}
|
||||
}
|
||||
|
||||
func NewReader(rpcClient *rpc.Client, tokenManager *token.Manager, priceManager *PriceManager, cryptoCompare *CryptoCompare, accountsDB *accounts.Database, walletFeed *event.Feed) *Reader {
|
||||
func NewReader(rpcClient *rpc.Client, tokenManager *token.Manager, priceManager *PriceManager, cryptoCompare *thirdparty.CryptoCompare, accountsDB *accounts.Database, walletFeed *event.Feed) *Reader {
|
||||
return &Reader{rpcClient, tokenManager, priceManager, cryptoCompare, accountsDB, walletFeed, nil}
|
||||
}
|
||||
|
||||
@ -33,7 +34,7 @@ type Reader struct {
|
||||
rpcClient *rpc.Client
|
||||
tokenManager *token.Manager
|
||||
priceManager *PriceManager
|
||||
cryptoCompare *CryptoCompare
|
||||
cryptoCompare *thirdparty.CryptoCompare
|
||||
accountsDB *accounts.Database
|
||||
walletFeed *event.Feed
|
||||
cancel context.CancelFunc
|
||||
@ -179,8 +180,8 @@ func (r *Reader) GetWalletToken(ctx context.Context, addresses []common.Address)
|
||||
var (
|
||||
group = async.NewAtomicGroup(ctx)
|
||||
prices = map[string]map[string]float64{}
|
||||
tokenDetails = map[string]Coin{}
|
||||
tokenMarketValues = map[string]map[string]MarketCoinValues{}
|
||||
tokenDetails = map[string]thirdparty.Coin{}
|
||||
tokenMarketValues = map[string]map[string]thirdparty.MarketCoinValues{}
|
||||
balances = map[uint64]map[common.Address]map[common.Address]*hexutil.Big{}
|
||||
)
|
||||
|
||||
@ -193,7 +194,7 @@ func (r *Reader) GetWalletToken(ctx context.Context, addresses []common.Address)
|
||||
})
|
||||
|
||||
group.Add(func(parent context.Context) error {
|
||||
tokenDetails, err = r.cryptoCompare.fetchTokenDetails(tokenSymbols)
|
||||
tokenDetails, err = r.cryptoCompare.FetchTokenDetails(tokenSymbols)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -201,7 +202,7 @@ func (r *Reader) GetWalletToken(ctx context.Context, addresses []common.Address)
|
||||
})
|
||||
|
||||
group.Add(func(parent context.Context) error {
|
||||
tokenMarketValues, err = r.cryptoCompare.fetchTokenMarketValues(tokenSymbols, currencies)
|
||||
tokenMarketValues, err = r.cryptoCompare.FetchTokenMarketValues(tokenSymbols, currencies)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@ -17,6 +17,7 @@ import (
|
||||
"github.com/status-im/status-go/services/stickers"
|
||||
"github.com/status-im/status-go/services/wallet/chain"
|
||||
"github.com/status-im/status-go/services/wallet/history"
|
||||
"github.com/status-im/status-go/services/wallet/thirdparty"
|
||||
"github.com/status-im/status-go/services/wallet/token"
|
||||
"github.com/status-im/status-go/services/wallet/transfer"
|
||||
"github.com/status-im/status-go/services/wallet/walletevent"
|
||||
@ -53,10 +54,10 @@ func NewService(
|
||||
savedAddressesManager := &SavedAddressesManager{db: db}
|
||||
transactionManager := &TransactionManager{db: db, transactor: transactor, gethManager: gethManager, config: config, accountsDB: accountsDB}
|
||||
transferController := transfer.NewTransferController(db, rpcClient, accountFeed, walletFeed)
|
||||
cryptoCompare := NewCryptoCompare()
|
||||
cryptoCompare := thirdparty.NewCryptoCompare()
|
||||
priceManager := NewPriceManager(db, cryptoCompare)
|
||||
reader := NewReader(rpcClient, tokenManager, priceManager, cryptoCompare, accountsDB, walletFeed)
|
||||
history := history.NewService(db, walletFeed, rpcClient, tokenManager)
|
||||
history := history.NewService(db, walletFeed, rpcClient, tokenManager, cryptoCompare)
|
||||
return &Service{
|
||||
db: db,
|
||||
accountsDB: accountsDB,
|
||||
@ -102,7 +103,7 @@ type Service struct {
|
||||
feed *event.Feed
|
||||
signals *walletevent.SignalsTransmitter
|
||||
reader *Reader
|
||||
cryptoCompare *CryptoCompare
|
||||
cryptoCompare *thirdparty.CryptoCompare
|
||||
history *history.Service
|
||||
}
|
||||
|
||||
@ -110,6 +111,7 @@ type Service struct {
|
||||
func (s *Service) Start() error {
|
||||
s.transferController.Start()
|
||||
err := s.signals.Start()
|
||||
s.history.Start()
|
||||
s.started = true
|
||||
return err
|
||||
}
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
package wallet
|
||||
package thirdparty
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
@ -119,7 +119,7 @@ func (c *CryptoCompare) DoQuery(url string) (*http.Response, error) {
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (c *CryptoCompare) fetchPrices(symbols []string, currencies []string) (map[string]map[string]float64, error) {
|
||||
func (c *CryptoCompare) FetchPrices(symbols []string, currencies []string) (map[string]map[string]float64, error) {
|
||||
chunks := chunkSymbols(symbols)
|
||||
result := make(map[string]map[string]float64)
|
||||
realCurrencies := renameSymbols(currencies)
|
||||
@ -153,7 +153,7 @@ func (c *CryptoCompare) fetchPrices(symbols []string, currencies []string) (map[
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (c *CryptoCompare) fetchTokenDetails(symbols []string) (map[string]Coin, error) {
|
||||
func (c *CryptoCompare) FetchTokenDetails(symbols []string) (map[string]Coin, error) {
|
||||
url := fmt.Sprintf("%s/data/all/coinlist", cryptocompareURL)
|
||||
resp, err := c.DoQuery(url)
|
||||
if err != nil {
|
||||
@ -181,7 +181,7 @@ func (c *CryptoCompare) fetchTokenDetails(symbols []string) (map[string]Coin, er
|
||||
return coins, nil
|
||||
}
|
||||
|
||||
func (c *CryptoCompare) fetchTokenMarketValues(symbols []string, currencies []string) (map[string]map[string]MarketCoinValues, error) {
|
||||
func (c *CryptoCompare) FetchTokenMarketValues(symbols []string, currencies []string) (map[string]map[string]MarketCoinValues, error) {
|
||||
realCurrencies := renameSymbols(currencies)
|
||||
realSymbols := renameSymbols(symbols)
|
||||
item := map[string]map[string]MarketCoinValues{}
|
||||
@ -214,7 +214,7 @@ func (c *CryptoCompare) fetchTokenMarketValues(symbols []string, currencies []st
|
||||
return item, nil
|
||||
}
|
||||
|
||||
func (c *CryptoCompare) fetchHourlyMarketValues(symbol string, currency string, limit int, aggregate int) ([]TokenHistoricalPairs, error) {
|
||||
func (c *CryptoCompare) FetchHourlyMarketValues(symbol string, currency string, limit int, aggregate int) ([]TokenHistoricalPairs, error) {
|
||||
item := []TokenHistoricalPairs{}
|
||||
|
||||
url := fmt.Sprintf("%s/data/v2/histohour?fsym=%s&tsym=%s&aggregate=%d&limit=%d&extraParams=Status.im", cryptocompareURL, getRealSymbol(symbol), currency, aggregate, limit)
|
||||
@ -240,7 +240,7 @@ func (c *CryptoCompare) fetchHourlyMarketValues(symbol string, currency string,
|
||||
return item, nil
|
||||
}
|
||||
|
||||
func (c *CryptoCompare) fetchDailyMarketValues(symbol string, currency string, limit int, allData bool, aggregate int) ([]TokenHistoricalPairs, error) {
|
||||
func (c *CryptoCompare) FetchDailyMarketValues(symbol string, currency string, limit int, allData bool, aggregate int) ([]TokenHistoricalPairs, error) {
|
||||
item := []TokenHistoricalPairs{}
|
||||
|
||||
url := fmt.Sprintf("%s/data/v2/histoday?fsym=%s&tsym=%s&aggregate=%d&limit=%d&allData=%v&extraParams=Status.im", cryptocompareURL, getRealSymbol(symbol), currency, aggregate, limit, allData)
|
||||
Loading…
x
Reference in New Issue
Block a user