Implement market cap metrics for currency discovery

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.59.0
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.102.2
 	github.com/code-payments/code-vm-indexer v1.2.0
-	github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260814155826-8088d9d58830
+	github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260818131607-042819236352
 	github.com/emirpasic/gods v1.12.0
 	github.com/envoyproxy/protoc-gen-validate v1.3.3
 	github.com/golang/protobuf v1.5.4

go.sum

@@ -80,8 +80,8 @@ github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
 github.com/code-payments/code-vm-indexer v1.2.0 h1:rSHpBMiT9BKgmKcXg/VIoi/h0t7jNxGx07Qz59m+6Q0=
 github.com/code-payments/code-vm-indexer v1.2.0/go.mod h1:vn91YN2qNqb+gGJeZe2+l+TNxVmEEiRHXXnIn2Y40h8=
-github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260814155826-8088d9d58830 h1:PVX61XNEbm8iBKAUw6i+NN04qOX9bbenkpRWs0R6CCc=
-github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260814155826-8088d9d58830/go.mod h1:tw6BooY5a8l6CtSZnKOruyKII0W04n89pcM4BizrgG8=
+github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260818131607-042819236352 h1:RsDDYM1VXBNcEYSZ+GvnMXCua9pMckYdIwdJnmiX+FI=
+github.com/code-payments/ocp-protobuf-api v1.14.1-0.20260818131607-042819236352/go.mod h1:tw6BooY5a8l6CtSZnKOruyKII0W04n89pcM4BizrgG8=
 github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6 h1:NmTXa/uVnDyp0TY5MKi197+3HWcnYWfnHGyaFthlnGw=
 github.com/containerd/continuity v0.0.0-20190827140505-75bee3e2ccb6/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=

ocp/currency/data_provider.go

@@ -50,6 +50,13 @@ type LiveHolderCountData struct {
 	LastWeekDelta  int64
 }
 
+// LiveMarketCapData represents live market cap data for a currency
+type LiveMarketCapData struct {
+	Mint             *common.Account
+	CurrentMarketCap float64
+	LastWeekDelta    float64
+}
+
 type cachedProtoMint struct {
 	mint          *currencypb.Mint
 	lastUpdatedAt time.Time
@@ -77,6 +84,7 @@ type MintDataProvider struct {
 	exchangeRates     *LiveExchangeRateData
 	launchpadReserves map[string]*LiveReserveStateData
 	holderCounts      map[string]*LiveHolderCountData
+	marketCaps        map[string]*LiveMarketCapData
 
 	streamsMu sync.RWMutex
 	streams   map[string]*LiveMintDataStream
@@ -131,6 +139,7 @@ func NewMintDataProvider(
 		currencyMetadata:              make(map[string]*currency.MetadataRecord),
 		launchpadReserves:             make(map[string]*LiveReserveStateData),
 		holderCounts:                  make(map[string]*LiveHolderCountData),
+		marketCaps:                    make(map[string]*LiveMarketCapData),
 		streams:                       make(map[string]*LiveMintDataStream),
 		exchangeRatesReady:            make(chan struct{}),
 		reserveStatesReady:            make(chan struct{}),
@@ -176,7 +185,7 @@ func (m *MintDataProvider) Stop() {
 func (m *MintDataProvider) ToProtoMint(
 	ctx context.Context,
 	metadataRecord *currency.MetadataRecord,
-	includeLiveReserveState, includeLiveHolderMetrics bool,
+	includeLiveReserveState, includeLiveHolderMetrics, includeLiveMarketCapMetrics bool,
 ) (*currencypb.Mint, error) {
 	mint, err := common.NewAccountFromPublicKeyString(metadataRecord.Mint)
 	if err != nil {
@@ -278,6 +287,13 @@ func (m *MintDataProvider) ToProtoMint(
 		}
 	}
 
+	if includeLiveMarketCapMetrics {
+		err = m.InjectLiveMarketCapMetrics(ctx, protoMint)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return protoMint, nil
 }
 
@@ -346,6 +362,39 @@ func SetHolderMetrics(protoMint *currencypb.Mint, holderData *LiveHolderCountDat
 	}
 }
 
+func (m *MintDataProvider) InjectLiveMarketCapMetrics(ctx context.Context, protoMint *currencypb.Mint) error {
+	if protoMint.LaunchpadMetadata == nil {
+		return errors.New("only launchpad currencies supported for market cap")
+	}
+
+	mint, err := common.NewAccountFromProto(protoMint.Address)
+	if err != nil {
+		return errors.New("invalid proto mint")
+	}
+
+	marketCapData, err := m.GetLiveMarketCap(ctx, mint)
+	if err != nil {
+		return err
+	}
+
+	SetMarketCapMetrics(protoMint, marketCapData)
+	return nil
+}
+
+// SetMarketCapMetrics applies market cap data to a proto Mint without fetching
+// from the provider.
+func SetMarketCapMetrics(protoMint *currencypb.Mint, marketCapData *LiveMarketCapData) {
+	protoMint.MarketCapMetrics = &currencypb.MarketCapMetrics{
+		CurrentMarketCap: marketCapData.CurrentMarketCap,
+		MarketCapDeltas: []*currencypb.MarketCapMetrics_DeltaMarketCap{
+			{
+				Range: currencypb.PredefinedRange_LAST_WEEK,
+				Delta: marketCapData.LastWeekDelta,
+			},
+		},
+	}
+}
+
 // GetProtoMint gets a proto Mint object. Static and infrequently updated metadata is
 // heavily cached.
 func (m *MintDataProvider) GetProtoMint(ctx context.Context, mint *common.Account) (*currencypb.Mint, error) {
@@ -411,7 +460,7 @@ func (m *MintDataProvider) GetProtoMint(ctx context.Context, mint *common.Accoun
 			return nil, currency.ErrNotFound
 		}
 
-		protoMetadata, err = m.ToProtoMint(ctx, metadataRecord, true, false)
+		protoMetadata, err = m.ToProtoMint(ctx, metadataRecord, true, false, false)
 		if err != nil {
 			return nil, err
 		}
@@ -538,6 +587,24 @@ func (m *MintDataProvider) GetAllCachedHolderCounts(ctx context.Context) (map[st
 	return out, nil
 }
 
+// GetAllCachedMarketCaps returns a snapshot of all currently cached market cap
+// data keyed by mint address. It blocks until the first successful poll has
+// completed.
+func (m *MintDataProvider) GetAllCachedMarketCaps(ctx context.Context) (map[string]*LiveMarketCapData, error) {
+	if err := m.waitForReserveStates(ctx); err != nil {
+		return nil, err
+	}
+
+	m.stateMu.RLock()
+	defer m.stateMu.RUnlock()
+
+	out := make(map[string]*LiveMarketCapData, len(m.marketCaps))
+	for k, v := range m.marketCaps {
+		out[k] = v
+	}
+	return out, nil
+}
+
 // GetLiveReserveState returns a current pre-signed live launchpad currency reserve state for a mint
 func (m *MintDataProvider) GetLiveReserveState(ctx context.Context, mint *common.Account) (*LiveReserveStateData, error) {
 	m.stateMu.RLock()
@@ -571,6 +638,40 @@ func (m *MintDataProvider) GetLiveReserveState(ctx context.Context, mint *common
 	return data, nil
 }
 
+// GetLiveMarketCap returns live market cap data for a mint, blocking until the
+// reserve poller has retrieved the state it's derived from.
+func (m *MintDataProvider) GetLiveMarketCap(ctx context.Context, mint *common.Account) (*LiveMarketCapData, error) {
+	m.stateMu.RLock()
+	data, ok := m.marketCaps[mint.PublicKey().ToBase58()]
+	m.stateMu.RUnlock()
+
+	if !ok {
+		isSupported, err := common.IsSupportedMint(ctx, m.data, mint)
+		if err != nil {
+			return nil, err
+		}
+		if !isSupported {
+			return nil, common.ErrUnsupportedMint
+		}
+	} else {
+		return data, nil
+	}
+
+	err := m.waitForReserveState(ctx, mint)
+	if err != nil {
+		return nil, err
+	}
+
+	m.stateMu.RLock()
+	defer m.stateMu.RUnlock()
+
+	data, ok = m.marketCaps[mint.PublicKey().ToBase58()]
+	if !ok {
+		return nil, errors.New("not found")
+	}
+	return data, nil
+}
+
 // waitForExchangeRates blocks until exchange rate data is available or context is cancelled
 func (m *MintDataProvider) waitForExchangeRates(ctx context.Context) error {
 	select {
@@ -937,6 +1038,23 @@ func (m *MintDataProvider) fetchAndUpdateReserveStates(ctx context.Context) {
 		return
 	}
 
+	mints := make([]string, 0, len(liveReserves))
+	for mintAddr := range liveReserves {
+		mints = append(mints, mintAddr)
+	}
+
+	// Market cap is derived from supply, so the weekly delta comes from the supply
+	// held a week ago.
+	var includeWeeklyDeltas bool
+	oneWeekAgo := time.Now().Add(-7 * 24 * time.Hour)
+	endOfWeekAgoDay := time.Date(oneWeekAgo.Year(), oneWeekAgo.Month(), oneWeekAgo.Day(), 23, 59, 59, 0, time.UTC)
+	historicalReserves, err := m.reserveStore.GetReservesForDay(ctx, mints, endOfWeekAgoDay)
+	if err != nil && err != currency.ErrNotFound {
+		m.log.With(zap.Error(err)).Warn("failed to fetch historical reserves for weekly market cap delta")
+	} else {
+		includeWeeklyDeltas = true
+	}
+
 	var updatedStates []*LiveReserveStateData
 	for mintAddr, record := range liveReserves {
 		mint, err := common.NewAccountFromPublicKeyString(mintAddr)
@@ -963,8 +1081,25 @@ func (m *MintDataProvider) fetchAndUpdateReserveStates(ctx context.Context) {
 			SignedState:       signedState,
 		}
 
+		var pastSupplyFromBonding uint64
+		if historicalReserves != nil {
+			if pastRecord, ok := historicalReserves[mintAddr]; ok {
+				pastSupplyFromBonding = pastRecord.SupplyFromBonding
+			}
+		}
+
+		currentMarketCap := CalculateMarketCap(record.SupplyFromBonding, 1.0)
+		marketCapData := &LiveMarketCapData{
+			Mint:             mint,
+			CurrentMarketCap: currentMarketCap,
+		}
+		if includeWeeklyDeltas {
+			marketCapData.LastWeekDelta = currentMarketCap - CalculateMarketCap(pastSupplyFromBonding, 1.0)
+		}
+
 		m.stateMu.Lock()
 		m.launchpadReserves[mintAddr] = stateData
+		m.marketCaps[mintAddr] = marketCapData
 		m.stateMu.Unlock()
 
 		m.markReserveStateReady(mint)

ocp/data/currency/reserve/cache/store.go

@@ -2,7 +2,7 @@
 // reserve lookups in front of a wrapped store.
 //
 // Point-in-time reads are keyed by mint and a coarse time bucket that doubles as
-// the freshness window. Range and live reads pass straight through. Live writes
+// the freshness window. Range, day and live reads pass straight through. Live writes
 // are guarded against the last successfully saved slot per mint: a write whose
 // slot is not greater is rejected with currency.ErrStaleReserveState without a
 // round-trip to the backing store. Everything else passes straight through.
@@ -75,6 +75,10 @@ func (s *store) GetReserveAtTime(ctx context.Context, mint string, t time.Time)
 	return record, nil
 }
 
+func (s *store) GetReservesForDay(ctx context.Context, mints []string, t time.Time) (map[string]*currency.ReserveRecord, error) {
+	return s.backing.GetReservesForDay(ctx, mints, t)
+}
+
 func (s *store) GetReservesInRange(ctx context.Context, mint string, interval query.Interval, start time.Time, end time.Time, ordering query.Ordering) ([]*currency.ReserveRecord, error) {
 	return s.backing.GetReservesInRange(ctx, mint, interval, start, end, ordering)
 }

ocp/data/currency/reserve/dynamodb/store.go

@@ -33,6 +33,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -60,6 +61,9 @@ const (
 	// codeConditionalCheckFailed is the DynamoDB cancellation reason code for a
 	// transaction item whose ConditionExpression evaluated false.
 	codeConditionalCheckFailed = "ConditionalCheckFailed"
+
+	// maxBatchGetItems is DynamoDB's per-call BatchGetItem key limit.
+	maxBatchGetItems = 100
 )
 
 // rollupResolutions are the coarse resolutions maintained alongside the raw
@@ -163,6 +167,64 @@ func (s *store) GetReserveAtTime(ctx context.Context, mint string, t time.Time)
 	return historyRecord(mint, out.Items[0])
 }
 
+// GetReservesForDay returns each mint's reserve state as of the UTC day of t —
+// the close of that mint's day rollup bucket. Because a rollup bucket has a
+// deterministic key (mint#day, bucketStart), this is a single batched key get
+// rather than a per-mint query. Mints with no record on that day are omitted.
+func (s *store) GetReservesForDay(ctx context.Context, mints []string, t time.Time) (map[string]*currency.ReserveRecord, error) {
+	bucket := skN(bucketStart(t, resDay))
+
+	// Dedup so a repeated mint isn't fetched twice.
+	seen := make(map[string]struct{}, len(mints))
+	keys := make([]map[string]types.AttributeValue, 0, len(mints))
+	for _, mint := range mints {
+		if _, ok := seen[mint]; ok {
+			continue
+		}
+		seen[mint] = struct{}{}
+		keys = append(keys, map[string]types.AttributeValue{
+			attrPK: avS(historyPK(mint, resDay)),
+			attrSK: bucket,
+		})
+	}
+
+	res := make(map[string]*currency.ReserveRecord, len(keys))
+	for start := 0; start < len(keys); start += maxBatchGetItems {
+		end := start + maxBatchGetItems
+		if end > len(keys) {
+			end = len(keys)
+		}
+
+		req := map[string]types.KeysAndAttributes{
+			s.historyTable: {Keys: keys[start:end]},
+		}
+		// Drain UnprocessedKeys (DynamoDB may return a partial batch under load).
+		for len(req[s.historyTable].Keys) > 0 {
+			out, err := s.client.BatchGetItem(ctx, &dynamodb.BatchGetItemInput{RequestItems: req})
+			if err != nil {
+				return nil, err
+			}
+			for _, item := range out.Responses[s.historyTable] {
+				mint, err := mintFromDayPK(item)
+				if err != nil {
+					return nil, err
+				}
+				rec, err := historyRecord(mint, item)
+				if err != nil {
+					return nil, err
+				}
+				res[mint] = rec
+			}
+			if unprocessed, ok := out.UnprocessedKeys[s.historyTable]; ok && len(unprocessed.Keys) > 0 {
+				req = map[string]types.KeysAndAttributes{s.historyTable: unprocessed}
+			} else {
+				break
+			}
+		}
+	}
+	return res, nil
+}
+
 func (s *store) GetReservesInRange(ctx context.Context, mint string, interval query.Interval, start time.Time, end time.Time, ordering query.Ordering) ([]*currency.ReserveRecord, error) {
 	if interval > query.IntervalMonth {
 		return nil, currency.ErrInvalidInterval
@@ -412,6 +474,18 @@ func bucketStart(t time.Time, res string) time.Time {
 
 func historyPK(mint, res string) string { return mint + "#" + res }
 
+// mintFromDayPK recovers the mint from a day-rollup item's pk ("<mint>#day").
+// Mints are base58, which never contains '#', so the resolution suffix is
+// unambiguous.
+func mintFromDayPK(item map[string]types.AttributeValue) (string, error) {
+	pk := asS(item[attrPK])
+	suffix := "#" + resDay
+	if !strings.HasSuffix(pk, suffix) {
+		return "", fmt.Errorf("unexpected day-rollup pk %q", pk)
+	}
+	return strings.TrimSuffix(pk, suffix), nil
+}
+
 // skN encodes t's unix-nanos as the numeric sort key, which sorts in
 // chronological order.
 func skN(t time.Time) types.AttributeValue { return avN(t.UTC().UnixNano()) }

ocp/data/currency/reserve/memory/store.go

@@ -87,6 +87,35 @@ func (s *store) GetReserveAtTime(ctx context.Context, mint string, t time.Time)
 	return results[0].Clone(), nil
 }
 
+func (s *store) GetReservesForDay(ctx context.Context, mints []string, t time.Time) (map[string]*currency.ReserveRecord, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	res := make(map[string]*currency.ReserveRecord, len(mints))
+	for _, mint := range mints {
+		// The close of t's UTC day for this mint: its most recent record that day.
+		var latest *currency.ReserveRecord
+		for _, item := range s.historical {
+			if item.Mint != mint || !sameUTCDay(item.Time, t) {
+				continue
+			}
+			if latest == nil || item.Time.After(latest.Time) {
+				latest = item
+			}
+		}
+		if latest != nil {
+			res[mint] = latest.Clone()
+		}
+	}
+	return res, nil
+}
+
+func sameUTCDay(a, b time.Time) bool {
+	ay, am, ad := a.UTC().Date()
+	by, bm, bd := b.UTC().Date()
+	return ay == by && am == bm && ad == bd
+}
+
 func (s *store) GetReservesInRange(ctx context.Context, mint string, interval query.Interval, start time.Time, end time.Time, ordering query.Ordering) ([]*currency.ReserveRecord, error) {
 	if interval > query.IntervalMonth {
 		return nil, currency.ErrInvalidInterval