feat(indexer): valuation pass — derive AMM TVL, prices and volume from chain state

Every USD field the exchange reads was an empty string. `pools`, `tokens` and
`factories` all resolved — with real addresses, real symbols, real fee tiers —
and then reported totalValueLockedUSD:"", volumeUSD:"", token0Price:"". The
landing page's TVL / 24h-volume / active-pool header had nothing to render, so
it rendered nothing.

Nothing was ever writing those fields. processLog's Sync/Mint/Burn arm was an
explicit no-op ("storage for these types not yet wired"), and bumpFactory's
comment deferred the USD aggregates to a "price oracle / day data" that does not
exist. Sync is the only event carrying V2 reserves, so with it inert there was
no path — at any block height — from the chain to a TVL number.

Fixed by adding the missing concern rather than by decoding one more event:

  indexer/valuation.go is the ONE owner of the derived USD aggregates. It reads
  balanceOf(pool) for both legs of every registered pool in a single JSON-RPC
  batch, anchors stablecoins at $1 by symbol, and relaxes prices outward through
  the pool graph — each unknown token taking its quote from the DEEPEST pool
  that already has a priced side, which is both deterministic and the least
  manipulable choice when pools disagree. A token no path connects to a stable
  stays unpriced and contributes nothing: unknown is reported as unknown.

Reserves are chain STATE, not chain HISTORY, so this is a pass beside the log
poller, not a handler inside it. That distinction is the whole point:

  - Sync exists only on V2. balanceOf answers the same question for V2, V3 and
    V4 identically.
  - An event-sourced TVL only converges by replaying all history, and a pool
    whose last Sync is a million blocks back reads empty until it trades again.
    On an idle chain it never trades again.
  - It is one fact, so it gets one writer. The Sync/Mint/Burn arm stays inert on
    purpose and now says why.

poolCount moves to the same pass for the same reason: it is derived from the
store, not accumulated by the create handlers, so it can no longer drift across
a restart, a re-index or a duplicate create. bumpFactory keeps the one field
that genuinely is a running total of events — txCount.

Fails closed: an unreachable RPC aborts the pass rather than overwriting good
aggregates with zeros.

Tests: valuation_test.go drives a stub RPC shaped like the live mainnet book —
a stablecoin pair anchoring a WLUX pair — and asserts derived TVL, spot prices,
per-token TVL, factory poolCount, the deeper-quote rule, non-clobbering of
fields the pass does not own, and the RPC-blip guard. Full suite green.

Co-authored-by: Hanzo Dev <[email protected]>

indexer/dex_v4_test.go

@@ -178,9 +178,11 @@ func TestHandleInitializeV4_PoolMarketFactory(t *testing.T) {
 	if p, _ := s.GetPool(nil, poolID); p == nil {
 		t.Fatal("expected pool from InitializeV4")
 	}
+	// The create path owns the factory tx count; poolCount is derived from the
+	// store by the valuation pass (TestRevalue_DerivesTVLPricesAndPoolCount).
 	f, _ := s.GetFactory(nil, "1")
-	if f == nil || asInt64(f.(map[string]interface{})["poolCount"]) != 1 {
-		t.Fatalf("expected factory poolCount=1, got %v", f)
+	if f == nil || asInt64(f.(map[string]interface{})["txCount"]) != 1 {
+		t.Fatalf("expected factory txCount=1, got %v", f)
 	}
 	mk, _ := s.GetByType("Market", poolID)
 	if mk == nil {
@@ -243,13 +245,17 @@ func TestBumpFactory_PreservesTVLAndVolume(t *testing.T) {
 		PoolCount: 2, TxCount: 5, TotalValueLockedUSD: "1000000", TotalVolumeUSD: "5000000",
 	})
 
-	// A pool-create bumps counts — must not zero the USD aggregates.
-	idx.bumpFactory(1)
+	// A pool-create bumps the tx count — and must not disturb any field it does
+	// not own: not the USD aggregates, and not poolCount (derived elsewhere).
+	idx.bumpFactory()
 
 	f, _ := s.GetFactory(nil, "1")
 	fm := f.(map[string]interface{})
-	if asInt64(fm["poolCount"]) != 3 {
-		t.Errorf("poolCount = %v, want 3", fm["poolCount"])
+	if asInt64(fm["txCount"]) != 6 {
+		t.Errorf("txCount = %v, want 6", fm["txCount"])
+	}
+	if asInt64(fm["poolCount"]) != 2 {
+		t.Errorf("poolCount must be carried through untouched: got %v want 2", fm["poolCount"])
 	}
 	if fmt.Sprint(fm["totalValueLockedUSD"]) != "1000000" {
 		t.Errorf("totalValueLockedUSD clobbered: got %v want 1000000", fm["totalValueLockedUSD"])
@@ -665,8 +671,11 @@ func TestHandleV2_RejectsNonFactory(t *testing.T) {
 	if p, _ := s.GetPool(nil, ct); p == nil {
 		t.Fatal("canonical PairCreated must seed the pair")
 	}
-	if f, _ := s.GetFactory(nil, "1"); f == nil || asInt64(f.(map[string]interface{})["poolCount"]) != 1 {
-		t.Fatalf("canonical PairCreated must bump factory poolCount to 1, got %v", f)
+	// poolCount is DERIVED by the valuation pass from the store (see
+	// TestRevalue_DerivesTVLPricesAndPoolCount); what the create handler owns is
+	// the factory tx count, and the registration itself — asserted above.
+	if f, _ := s.GetFactory(nil, "1"); f == nil || asInt64(f.(map[string]interface{})["txCount"]) != 1 {
+		t.Fatalf("canonical PairCreated must bump factory txCount to 1, got %v", f)
 	}
 	idx.processLog(context.Background(), swapV2Log(ct, token0, "0xreal", 1000, 2500))
 	sw, _ := s.GetSwaps(nil, 10, "timestamp", "desc", nil)
@@ -705,8 +714,8 @@ func TestHandleV3_RejectsNonFactory(t *testing.T) {
 	if p, _ := s.GetPool(nil, ct); p == nil {
 		t.Fatal("canonical PoolCreated must seed the pool")
 	}
-	if f, _ := s.GetFactory(nil, "1"); f == nil || asInt64(f.(map[string]interface{})["poolCount"]) != 1 {
-		t.Fatalf("canonical PoolCreated must bump factory poolCount to 1, got %v", f)
+	if f, _ := s.GetFactory(nil, "1"); f == nil || asInt64(f.(map[string]interface{})["txCount"]) != 1 {
+		t.Fatalf("canonical PoolCreated must bump factory txCount to 1, got %v", f)
 	}
 	idx.processLog(context.Background(), swapV3Log(ct, token0, "0xreal", 1000, 2500))
 	sw, _ := s.GetSwaps(nil, 10, "timestamp", "desc", nil)

indexer/indexer.go

@@ -292,6 +292,12 @@ func (idx *Indexer) Run(ctx context.Context) error {
 		}()
 	}
 
+	// The valuation pass (valuation.go) derives the USD aggregates from current
+	// on-chain balances. It runs beside the log poller, not inside it: reserves
+	// are chain STATE, events are chain HISTORY, and an idle chain still has
+	// state to report. Sharing only the store keeps the two concerns apart.
+	go idx.runValuation(ctx)
+
 	ticker := time.NewTicker(2 * time.Second)
 	defer ticker.Stop()
 
@@ -498,7 +504,11 @@ func (idx *Indexer) processLog(ctx context.Context, l *logEntry) {
 		idx.handleDEXFill(l, blockNum, txHash, logIdx)
 	case SigMintV2, SigMintV3, SigBurnV2, SigBurnV3, SigSync,
 		SigCollect, SigFlash, SigInitialize, SigModifyLiquidity:
-		// Recognized but storage for these types not yet wired
+		// Deliberately inert. Every one of these events only restates what a
+		// pool HOLDS, and that is derived state the valuation pass already owns
+		// (valuation.go) — from balanceOf at head, which covers V2, V3 and V4
+		// alike and needs no history replay. Decoding them here would be a
+		// second, V2-only writer for the same field.
 
 	// ── ERC-3643 IToken ───────────────────────────────────────────────
 	case SigAddressFrozen:
@@ -676,7 +686,7 @@ func (idx *Indexer) handlePairCreated(ctx context.Context, l *logEntry) {
 	})
 	idx.seedToken(ctx, token0)
 	idx.seedToken(ctx, token1)
-	idx.bumpFactory(1)
+	idx.bumpFactory()
 }
 
 func (idx *Indexer) handlePoolCreated(ctx context.Context, l *logEntry) {
@@ -707,7 +717,7 @@ func (idx *Indexer) handlePoolCreated(ctx context.Context, l *logEntry) {
 	})
 	idx.seedToken(ctx, token0)
 	idx.seedToken(ctx, token1)
-	idx.bumpFactory(1)
+	idx.bumpFactory()
 }
 
 func (idx *Indexer) handleTransfer(ctx context.Context, l *logEntry, txHash, logIdx string) {
@@ -732,28 +742,28 @@ func decodeInt256(data string, wordIndex int) *big.Int {
 	return n
 }
 
-// bumpFactory increments the singleton factory ("1") aggregate so the
-// dashboard `factory`/`factories`/`uniswapFactories` query returns non-zero
-// pool/tx counts. One place, called from every pool/pair creation handler —
-// the factory aggregate was previously only ever seeded by tests, leaving the
-// exchange landing page's TVL/volume header empty even with live pools.
-func (idx *Indexer) bumpFactory(deltaPools int64) {
+// bumpFactory increments the singleton factory ("1") transaction count — the
+// only field of the factory aggregate that IS a running total of events, and so
+// the only one the log handlers own.
+//
+// poolCount and the USD aggregates are DERIVED, not accumulated: the valuation
+// pass reads the true pool count out of the store and recomputes TVL/volume
+// from on-chain balances every interval. An accumulator for those drifts on
+// every restart, re-index and duplicate create, and there is no reason to count
+// something the store can simply be asked for. SeedFactory is INSERT OR REPLACE,
+// so this read-modify-write carries the derived fields through untouched.
+func (idx *Indexer) bumpFactory() {
 	f, _ := idx.store.GetFactory(nil, "1")
 	var pc, tc int64
 	var tvl, vol string
 	if m, ok := f.(map[string]interface{}); ok {
 		pc = asInt64(m["poolCount"])
 		tc = asInt64(m["txCount"])
-		// Carry the accumulated USD aggregates through the read-modify-write.
-		// SeedFactory is INSERT OR REPLACE (full overwrite); writing them empty
-		// here would clobber the landing-page header's TVL/volume on every
-		// pool-create / swap. They are accumulated elsewhere (price oracle / day
-		// data); bumpFactory only owns the pool + tx counts and must preserve the rest.
 		tvl = asString(m["totalValueLockedUSD"])
 		vol = asString(m["totalVolumeUSD"])
 	}
 	idx.store.SeedFactory("1", &storage.SeedFactoryData{
-		PoolCount:           pc + deltaPools,
+		PoolCount:           pc,
 		TxCount:             tc + 1,
 		TotalValueLockedUSD: tvl,
 		TotalVolumeUSD:      vol,
@@ -813,7 +823,7 @@ func (idx *Indexer) handleInitializeV4(ctx context.Context, l *logEntry) {
 	})
 	idx.seedToken(ctx, token0)
 	idx.seedToken(ctx, token1)
-	idx.bumpFactory(1)
+	idx.bumpFactory()
 
 	// A bound market carries a human BASE/QUOTE symbol (not the raw poolId) when
 	// BOTH currencies resolve to a clean ERC20 symbol. `assetsBound` marks the
@@ -903,7 +913,7 @@ func (idx *Indexer) handleSwapV4(l *logEntry, blockNum uint64, txHash, logIdx st
 		AmountUSD: "0",
 		Sender:    sender,
 	})
-	idx.bumpFactory(0)
+	idx.bumpFactory()
 }
 
 // handleDEXFill records a native-CLOB settlement (DEXFill at 0x9999). This is

indexer/valuation.go

@@ -0,0 +1,506 @@
+package indexer
+
+// Valuation — the one owner of the AMM graph's derived USD aggregates.
+//
+// WHY THIS IS A PASS AND NOT AN EVENT HANDLER
+//
+// Reserves are a fact about the chain's CURRENT state, not an event stream. The
+// event-sourced route to them (V2 `Sync`, plus V3/V4 Mint/Burn/Swap liquidity
+// math) has three defects here:
+//
+//   - It is not universal. `Sync` exists only on V2 pairs. A V3 or V4 pool never
+//     emits it, so an event-only TVL would silently cover a fraction of the book.
+//   - It only converges by replaying ALL history. A pool whose last `Sync` is a
+//     million blocks back reads as empty until it trades again — and on an idle
+//     chain it never trades again.
+//   - It is a second way to learn one fact. `balanceOf(pool)` answers "what does
+//     this pool hold" for V2, V3 and V4 identically, in one call, at head.
+//
+// So the log indexer owns EVENTS (who created a pool, who swapped) and this pass
+// owns DERIVED STATE (what the pool holds, what it is worth). processLog's
+// Sync/Mint/Burn arm stays a no-op on purpose: honoring it would be a second,
+// partial writer for a field this pass already owns.
+//
+// PRICING
+//
+// No external oracle, no paid feed, no hardcoded price. Stablecoins anchor at
+// $1 by symbol; every other token's price is relaxed outward from that anchor
+// through the pools themselves, each unknown token taking its price from the
+// DEEPEST pool that already has a priced side. Depth-ordering makes the result
+// deterministic and picks the least manipulable quote when several pools
+// disagree. A token no pool connects to a stable stays unpriced and contributes
+// nothing — an unknown value is reported as unknown, never guessed.
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"math"
+	"math/big"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/luxfi/graph/storage"
+)
+
+// valuationInterval is how often the derived USD aggregates are recomputed.
+// The pass is O(pools) eth_calls, so it runs on its own slow cadence rather
+// than on the 2s block poll.
+const valuationInterval = 30 * time.Second
+
+// maxValuedPools caps one pass. Far above any real book; a guard, not a policy.
+const maxValuedPools = 5000
+
+// selBalanceOf is `balanceOf(address)`.
+const selBalanceOf = "0x70a08231"
+
+// stableSymbols are the USD-pegged symbols that anchor the price graph at $1.
+// Keyed by upper-cased ERC20 symbol: a peg is a property of the ASSET, and the
+// same asset is deployed at many addresses across bridges and chains.
+var stableSymbols = map[string]bool{
+	"USDC": true, "USDT": true, "DAI": true, "LUSD": true, "USDS": true,
+	"BUSD": true, "FRAX": true, "TUSD": true, "USDP": true, "PYUSD": true,
+	"USDE": true, "GUSD": true,
+}
+
+// priceRelaxRounds bounds the outward relaxation from the stable anchors. Each
+// round prices every token one pool-hop further out, so this is the maximum
+// path length from a stablecoin to a priced token. Six hops is far beyond any
+// real routing depth and the loop exits early on a fixpoint anyway.
+const priceRelaxRounds = 6
+
+// pool is one pool's valuation input: identity, token decimals, and the
+// balances actually held on-chain right now.
+type valuedPool struct {
+	id           string
+	t0, t1       string  // lower-cased token addresses
+	bal0, bal1   float64 // token-denominated (decimals applied)
+	held0, held1 bool    // whether the balance read succeeded
+}
+
+// runValuation drives the valuation pass on its own ticker until ctx ends. It
+// is started by Run alongside the log poller: orthogonal concerns, one store.
+func (idx *Indexer) runValuation(ctx context.Context) {
+	ticker := time.NewTicker(valuationInterval)
+	defer ticker.Stop()
+	// Value once at start so a restart does not serve stale aggregates for a
+	// full interval.
+	idx.revalue(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			idx.revalue(ctx)
+		}
+	}
+}
+
+// revalue recomputes every derived USD aggregate from on-chain balances and
+// writes them back. Read-modify-write throughout: this pass owns the USD and
+// price fields and must preserve everything the log handlers own.
+func (idx *Indexer) revalue(ctx context.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("[valuation] recovered: %v", r)
+		}
+	}()
+
+	pools := idx.store.PoolsRaw()
+	if len(pools) == 0 {
+		return
+	}
+	tokens := idx.store.TokensRaw()
+
+	vps := idx.readBalances(ctx, pools, tokens)
+	if len(vps) == 0 {
+		return
+	}
+	prices := priceTokens(vps, tokens)
+
+	// ── Pools: TVL, spot prices, volume ────────────────────────────────
+	tokenTVL := map[string]float64{}
+	tokenVol := map[string]float64{}
+	poolVol := idx.valueSwaps(vps, tokens, prices, tokenVol)
+
+	var totalTVL, totalVol float64
+	for _, vp := range vps {
+		p := pools[vp.id]
+		if p == nil {
+			continue
+		}
+		v0, ok0 := usdLeg(vp.bal0, vp.t0, prices)
+		v1, ok1 := usdLeg(vp.bal1, vp.t1, prices)
+		tvl := 0.0
+		switch {
+		case ok0 && ok1:
+			tvl = v0 + v1
+		case ok0 && vp.bal1 > 0:
+			// Constant-product: at the marginal price both legs are worth the
+			// same, so a single priced leg values the whole pool. (This is the
+			// same rule the Uniswap subgraphs use for a whitelisted-token pair.)
+			tvl = 2 * v0
+		case ok1 && vp.bal0 > 0:
+			tvl = 2 * v1
+		case ok0:
+			tvl = v0 // one-sided pool: no mirror leg to double
+		case ok1:
+			tvl = v1
+		}
+		if ok0 {
+			tokenTVL[vp.t0] += v0
+		}
+		if ok1 {
+			tokenTVL[vp.t1] += v1
+		}
+
+		p.TotalValueLockedUSD = fmtUSD(tvl)
+		p.Token0Price = fmtRatio(vp.bal0, vp.bal1)
+		p.Token1Price = fmtRatio(vp.bal1, vp.bal0)
+		if v, ok := poolVol[vp.id]; ok {
+			p.VolumeUSD = fmtUSD(v)
+			totalVol += v
+		} else if p.VolumeUSD == "" {
+			p.VolumeUSD = fmtUSD(0)
+		}
+		idx.store.SeedPool(vp.id, p)
+		totalTVL += tvl
+	}
+
+	// ── Tokens: TVL and volume across every pool holding them ──────────
+	for addr, t := range tokens {
+		tvl, hasTVL := tokenTVL[strings.ToLower(addr)]
+		vol, hasVol := tokenVol[strings.ToLower(addr)]
+		if !hasTVL && !hasVol {
+			continue
+		}
+		t.TotalValueLockedUSD = fmtUSD(tvl)
+		t.VolumeUSD = fmtUSD(vol)
+		idx.store.SeedToken(addr, t)
+	}
+
+	// ── Factory: the landing-page header aggregate ─────────────────────
+	// poolCount is DERIVED here (the true number of pools in the store), not
+	// accumulated by the create handlers — an accumulator drifts on every
+	// restart, re-index and duplicate create, and there is no reason to guess a
+	// number the store can be asked for.
+	f, _ := idx.store.GetFactory(nil, "1")
+	var txCount int64
+	if m, ok := f.(map[string]interface{}); ok {
+		txCount = asInt64(m["txCount"])
+	}
+	idx.store.SeedFactory("1", &storage.SeedFactoryData{
+		PoolCount:           int64(len(pools)),
+		TxCount:             txCount,
+		TotalValueLockedUSD: fmtUSD(totalTVL),
+		TotalVolumeUSD:      fmtUSD(totalVol),
+	})
+
+	log.Printf("[valuation] %d pools, %d tokens priced — TVL $%s, volume $%s",
+		len(vps), len(prices), fmtUSD(totalTVL), fmtUSD(totalVol))
+}
+
+// readBalances asks each pool's two tokens what the pool holds. One eth_call
+// per leg, batched into a single JSON-RPC request so a 35-pool book costs one
+// round trip rather than seventy.
+func (idx *Indexer) readBalances(ctx context.Context, pools map[string]*storage.SeedPoolData, tokens map[string]*storage.SeedTokenData) []valuedPool {
+	type leg struct {
+		pool  int  // index into vps
+		isT0  bool //
+		token string
+	}
+	var vps []valuedPool
+	var legs []leg
+	var calls []rpcBatchReq
+
+	for id, p := range pools {
+		if !isAddress(id) || len(vps) >= maxValuedPools {
+			continue
+		}
+		t0, t1 := strings.ToLower(p.Token0), strings.ToLower(p.Token1)
+		vps = append(vps, valuedPool{id: id, t0: t0, t1: t1})
+		n := len(vps) - 1
+		for _, l := range []leg{{n, true, t0}, {n, false, t1}} {
+			if !isAddress(l.token) {
+				continue
+			}
+			legs = append(legs, l)
+			calls = append(calls, rpcBatchReq{
+				JSONRPC: "2.0", ID: len(calls) + 1, Method: "eth_call",
+				Params: []interface{}{
+					map[string]string{"to": l.token, "data": selBalanceOf + padAddress(id)},
+					"latest",
+				},
+			})
+		}
+	}
+	if len(calls) == 0 {
+		return nil
+	}
+
+	results, err := idx.rpcBatch(ctx, calls)
+	if err != nil {
+		log.Printf("[valuation] balance batch: %v", err)
+		return nil
+	}
+	for i, l := range legs {
+		raw, ok := results[i+1]
+		if !ok {
+			continue
+		}
+		dec := decimalsOf(l.token, tokens)
+		bal, ok := decodeBalance(raw, dec)
+		if !ok {
+			continue
+		}
+		if l.isT0 {
+			vps[l.pool].bal0, vps[l.pool].held0 = bal, true
+		} else {
+			vps[l.pool].bal1, vps[l.pool].held1 = bal, true
+		}
+	}
+	return vps
+}
+
+// priceTokens relaxes USD prices outward from the stablecoin anchors through
+// the pool graph. Returns lower-cased token address → USD price.
+func priceTokens(vps []valuedPool, tokens map[string]*storage.SeedTokenData) map[string]float64 {
+	prices := map[string]float64{}
+	for addr, t := range tokens {
+		if stableSymbols[strings.ToUpper(strings.TrimSpace(t.Symbol))] {
+			prices[strings.ToLower(addr)] = 1
+		}
+	}
+	if len(prices) == 0 {
+		return prices
+	}
+
+	type quote struct{ price, depth float64 }
+	for round := 0; round < priceRelaxRounds; round++ {
+		best := map[string]quote{}
+		for _, vp := range vps {
+			// A price is a RATIO of two live balances; a pool with an empty leg
+			// quotes nothing.
+			if vp.bal0 <= 0 || vp.bal1 <= 0 {
+				continue
+			}
+			p0, ok0 := prices[vp.t0]
+			p1, ok1 := prices[vp.t1]
+			switch {
+			case ok0 && !ok1:
+				d := vp.bal0 * p0
+				if q, seen := best[vp.t1]; !seen || d > q.depth {
+					best[vp.t1] = quote{d / vp.bal1, d}
+				}
+			case ok1 && !ok0:
+				d := vp.bal1 * p1
+				if q, seen := best[vp.t0]; !seen || d > q.depth {
+					best[vp.t0] = quote{d / vp.bal0, d}
+				}
+			}
+		}
+		if len(best) == 0 {
+			break // fixpoint: nothing new is reachable
+		}
+		for addr, q := range best {
+			if q.price > 0 && !math.IsInf(q.price, 0) && !math.IsNaN(q.price) {
+				prices[addr] = q.price
+			}
+		}
+	}
+	return prices
+}
+
+// valueSwaps prices the stored swap history and returns per-pool volume,
+// accumulating per-token volume into tokenVol. A swap whose USD value changed
+// is written back so the trade history carries the same number the aggregates
+// were built from — one valuation, used everywhere.
+func (idx *Indexer) valueSwaps(vps []valuedPool, tokens map[string]*storage.SeedTokenData, prices map[string]float64, tokenVol map[string]float64) map[string]float64 {
+	byPool := map[string]*valuedPool{}
+	for i := range vps {
+		byPool[vps[i].id] = &vps[i]
+	}
+	out := map[string]float64{}
+	for _, vp := range vps {
+		out[vp.id] = 0
+	}
+	for id, sw := range idx.store.SwapsRaw() {
+		vp := byPool[strings.ToLower(sw.Pool)]
+		if vp == nil {
+			continue
+		}
+		v0, ok0 := swapLegUSD(sw.Amount0, vp.t0, tokens, prices)
+		v1, ok1 := swapLegUSD(sw.Amount1, vp.t1, tokens, prices)
+		var usd float64
+		switch {
+		case ok0 && ok1:
+			usd = (v0 + v1) / 2 // both legs priced: the mid of the two quotes
+		case ok0:
+			usd = v0
+		case ok1:
+			usd = v1
+		default:
+			continue
+		}
+		out[vp.id] += usd
+		if ok0 {
+			tokenVol[vp.t0] += v0
+		}
+		if ok1 {
+			tokenVol[vp.t1] += v1
+		}
+		if s := fmtUSD(usd); s != sw.AmountUSD {
+			sw.AmountUSD = s
+			idx.store.SeedSwap(id, sw)
+		}
+	}
+	return out
+}
+
+// swapLegUSD converts one signed raw swap leg to its absolute USD value.
+func swapLegUSD(raw, token string, tokens map[string]*storage.SeedTokenData, prices map[string]float64) (float64, bool) {
+	price, ok := prices[token]
+	if !ok || raw == "" {
+		return 0, false
+	}
+	n, ok := new(big.Int).SetString(raw, 10)
+	if !ok {
+		return 0, false
+	}
+	amt := normalize(new(big.Int).Abs(n), decimalsOf(token, tokens))
+	return amt * price, true
+}
+
+// usdLeg values one pool leg, reporting whether the token has a price at all.
+func usdLeg(bal float64, token string, prices map[string]float64) (float64, bool) {
+	p, ok := prices[token]
+	if !ok {
+		return 0, false
+	}
+	return bal * p, true
+}
+
+// --- decoding helpers -------------------------------------------------
+
+// rpcBatchReq is one entry of a JSON-RPC 2.0 batch.
+type rpcBatchReq struct {
+	JSONRPC string      `json:"jsonrpc"`
+	ID      int         `json:"id"`
+	Method  string      `json:"method"`
+	Params  interface{} `json:"params"`
+}
+
+// rpcBatch posts a JSON-RPC batch and returns id → result. Entries that
+// errored are absent from the map; the caller treats a missing id as "not
+// read" rather than as zero.
+func (idx *Indexer) rpcBatch(ctx context.Context, reqs []rpcBatchReq) (map[int]string, error) {
+	body, err := json.Marshal(reqs)
+	if err != nil {
+		return nil, err
+	}
+	req, err := http.NewRequestWithContext(ctx, "POST", idx.rpc, bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := idx.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	var out []struct {
+		ID     int    `json:"id"`
+		Result string `json:"result"`
+		Error  *struct {
+			Message string `json:"message"`
+		} `json:"error"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return nil, fmt.Errorf("rpc batch decode: %w", err)
+	}
+	res := make(map[int]string, len(out))
+	for _, r := range out {
+		if r.Error == nil && r.Result != "" {
+			res[r.ID] = r.Result
+		}
+	}
+	return res, nil
+}
+
+// decodeBalance turns a 32-byte uint256 hex word into a token-denominated
+// float. Reports false for a short/absent word so an unread balance is never
+// mistaken for a zero balance.
+func decodeBalance(raw string, decimals int64) (float64, bool) {
+	h := strings.TrimPrefix(raw, "0x")
+	if len(h) < 64 {
+		return 0, false
+	}
+	n, ok := new(big.Int).SetString(h[len(h)-64:], 16)
+	if !ok {
+		return 0, false
+	}
+	return normalize(n, decimals), true
+}
+
+// normalize scales a raw integer amount by the token's decimals.
+func normalize(n *big.Int, decimals int64) float64 {
+	if decimals < 0 || decimals > 36 {
+		decimals = defaultDecimals
+	}
+	f := new(big.Float).SetInt(n)
+	f.Quo(f, new(big.Float).SetInt(new(big.Int).Exp(big.NewInt(10), big.NewInt(decimals), nil)))
+	v, _ := f.Float64()
+	return v
+}
+
+// decimalsOf returns a token's ERC20 decimals from the store, defaulting to 18.
+func decimalsOf(token string, tokens map[string]*storage.SeedTokenData) int64 {
+	if t, ok := tokens[token]; ok && t.Decimals > 0 {
+		return t.Decimals
+	}
+	if t, ok := tokens[strings.ToLower(token)]; ok && t.Decimals > 0 {
+		return t.Decimals
+	}
+	return defaultDecimals
+}
+
+// isAddress reports whether s is a 0x-prefixed 20-byte hex address.
+func isAddress(s string) bool {
+	if len(s) != 42 || !strings.HasPrefix(s, "0x") {
+		return false
+	}
+	for _, c := range s[2:] {
+		if !strings.ContainsRune("0123456789abcdefABCDEF", c) {
+			return false
+		}
+	}
+	return true
+}
+
+// padAddress left-pads an address to a 32-byte ABI word.
+func padAddress(addr string) string {
+	return strings.Repeat("0", 24) + strings.ToLower(strings.TrimPrefix(addr, "0x"))
+}
+
+// fmtUSD renders a USD aggregate with two decimals — a money value, always
+// definite. A pool the pass could not price reports "0.00", never "".
+func fmtUSD(v float64) string {
+	if math.IsNaN(v) || math.IsInf(v, 0) {
+		return "0.00"
+	}
+	return strconv.FormatFloat(v, 'f', 2, 64)
+}
+
+// fmtRatio renders a spot price as num/den, matching the Uniswap convention
+// that Pair.token0Price is reserve0/reserve1. An undefined ratio renders "0".
+func fmtRatio(num, den float64) string {
+	if den <= 0 || math.IsNaN(num) || math.IsNaN(den) {
+		return "0"
+	}
+	return strconv.FormatFloat(num/den, 'g', 12, 64)
+}

indexer/valuation_test.go

@@ -0,0 +1,218 @@
+package indexer
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"math"
+	"math/big"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/luxfi/graph/storage"
+)
+
+// balanceRPC stands in for the chain: it answers `balanceOf(pool)` from a
+// table keyed "token@pool" and rejects anything else, so a test can only pass
+// by asking the questions the valuation pass is supposed to ask.
+func balanceRPC(t *testing.T, balances map[string]string) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		var reqs []struct {
+			ID     int           `json:"id"`
+			Method string        `json:"method"`
+			Params []interface{} `json:"params"`
+		}
+		if err := json.Unmarshal(body, &reqs); err != nil {
+			t.Errorf("valuation must send a JSON-RPC BATCH, got: %s", body)
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		out := make([]map[string]interface{}, 0, len(reqs))
+		for _, q := range reqs {
+			if q.Method != "eth_call" {
+				t.Errorf("unexpected method %q", q.Method)
+				continue
+			}
+			call := q.Params[0].(map[string]interface{})
+			to := strings.ToLower(call["to"].(string))
+			data := call["data"].(string)
+			if !strings.HasPrefix(data, selBalanceOf) {
+				t.Errorf("expected balanceOf selector, got %q", data)
+			}
+			arg := strings.TrimPrefix(data, selBalanceOf)
+			pool := "0x" + arg[len(arg)-40:]
+			res, ok := balances[to+"@"+pool]
+			if !ok {
+				res = "0x" + strings.Repeat("0", 64)
+			}
+			out = append(out, map[string]interface{}{"jsonrpc": "2.0", "id": q.ID, "result": res})
+		}
+		w.Header().Set("Content-Type", "application/json")
+		json.NewEncoder(w).Encode(out)
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+// TestRevalue_DerivesTVLPricesAndPoolCount is the end-to-end proof of the
+// valuation pass on a book shaped like the live Lux mainnet one: a stablecoin
+// pair to anchor the price graph, and a WLUX pair that must take its price from
+// that anchor rather than from thin air.
+func TestRevalue_DerivesTVLPricesAndPoolCount(t *testing.T) {
+	const (
+		usdc     = "0x0000000000000000000000000000000000000001"
+		usdt     = "0x0000000000000000000000000000000000000002"
+		wlux     = "0x0000000000000000000000000000000000000003"
+		mystery  = "0x0000000000000000000000000000000000000004"
+		stablePr = "0x00000000000000000000000000000000000000a1" // USDC/USDT
+		luxPair  = "0x00000000000000000000000000000000000000a2" // WLUX/USDT
+		orphan   = "0x00000000000000000000000000000000000000a3" // MYST/MYST2, unreachable
+	)
+	s := newMemSQLiteStore(t)
+
+	for addr, sym := range map[string]string{usdc: "USDC", usdt: "USDT", wlux: "WLUX", mystery: "MYST"} {
+		s.SeedToken(addr, &storage.SeedTokenData{Symbol: sym, Name: sym, Decimals: 18})
+	}
+	s.SeedPool(stablePr, &storage.SeedPoolData{Token0: usdc, Token1: usdt, FeeTier: 3000, TxCount: 7})
+	s.SeedPool(luxPair, &storage.SeedPoolData{Token0: wlux, Token1: usdt, FeeTier: 500})
+	s.SeedPool(orphan, &storage.SeedPoolData{Token0: mystery, Token1: mystery, FeeTier: 3000})
+
+	// 100 USDC + 100 USDT; 4000 WLUX + 100 USDT  =>  WLUX = $0.025.
+	srv := balanceRPC(t, map[string]string{
+		usdc + "@" + stablePr: hexWord(100),
+		usdt + "@" + stablePr: hexWord(100),
+		wlux + "@" + luxPair:  hexWord(4000),
+		usdt + "@" + luxPair:  hexWord(100),
+	})
+
+	idx := NewWithConfig(Config{RPC: srv.URL}, s)
+	idx.revalue(context.Background())
+
+	// ── The stable pair is worth the sum of its legs. ──────────────────
+	p, _ := s.GetPool(nil, stablePr)
+	if p == nil {
+		t.Fatal("stable pair vanished")
+	}
+	pm := p.(map[string]interface{})
+	if got := fmt.Sprint(pm["totalValueLockedUSD"]); got != "200.00" {
+		t.Errorf("stable pair TVL = %q, want 200.00", got)
+	}
+	// A field the pass does not own must survive its read-modify-write.
+	if asInt64(pm["txCount"]) != 7 {
+		t.Errorf("valuation clobbered txCount: got %v want 7", pm["txCount"])
+	}
+
+	// ── WLUX is priced THROUGH the anchor, and the pair valued from it. ─
+	p, _ = s.GetPool(nil, luxPair)
+	pm = p.(map[string]interface{})
+	if got := fmt.Sprint(pm["totalValueLockedUSD"]); got != "200.00" {
+		t.Errorf("WLUX pair TVL = %q, want 200.00 (4000 WLUX @ $0.025 + 100 USDT)", got)
+	}
+	if got := mustFloat(t, pm["token0Price"]); math.Abs(got-40) > 1e-9 {
+		t.Errorf("token0Price = %v, want 40 (reserve0/reserve1)", got)
+	}
+	if got := mustFloat(t, pm["token1Price"]); math.Abs(got-0.025) > 1e-12 {
+		t.Errorf("token1Price = %v, want 0.025", got)
+	}
+
+	// ── A pool no path connects to a stable is reported as $0, not blank. ─
+	p, _ = s.GetPool(nil, orphan)
+	if got := fmt.Sprint(p.(map[string]interface{})["totalValueLockedUSD"]); got != "0.00" {
+		t.Errorf("unpriceable pool TVL = %q, want 0.00", got)
+	}
+
+	// ── Token TVL sums every pool holding the token. ───────────────────
+	tok, _ := s.GetToken(nil, usdt)
+	if got := fmt.Sprint(tok.(map[string]interface{})["totalValueLockedUSD"]); got != "200.00" {
+		t.Errorf("USDT TVL = %q, want 200.00 (100 in each pair)", got)
+	}
+	tok, _ = s.GetToken(nil, wlux)
+	if got := fmt.Sprint(tok.(map[string]interface{})["totalValueLockedUSD"]); got != "100.00" {
+		t.Errorf("WLUX TVL = %q, want 100.00", got)
+	}
+
+	// ── Factory: poolCount is the real count, TVL the real sum. ────────
+	f, _ := s.GetFactory(nil, "1")
+	fm := f.(map[string]interface{})
+	if asInt64(fm["poolCount"]) != 3 {
+		t.Errorf("factory poolCount = %v, want 3 (derived from the store)", fm["poolCount"])
+	}
+	if got := fmt.Sprint(fm["totalValueLockedUSD"]); got != "400.00" {
+		t.Errorf("factory TVL = %q, want 400.00", got)
+	}
+}
+
+// TestRevalue_UnreachableRPCKeepsLastGoodValues proves the pass fails closed:
+// an RPC blip must never overwrite good aggregates with zeros.
+func TestRevalue_UnreachableRPCKeepsLastGoodValues(t *testing.T) {
+	const pool = "0x00000000000000000000000000000000000000b1"
+	const tokenA = "0x0000000000000000000000000000000000000011"
+	const tokenB = "0x0000000000000000000000000000000000000012"
+
+	s := newMemSQLiteStore(t)
+	s.SeedToken(tokenA, &storage.SeedTokenData{Symbol: "USDC", Decimals: 18})
+	s.SeedToken(tokenB, &storage.SeedTokenData{Symbol: "WLUX", Decimals: 18})
+	s.SeedPool(pool, &storage.SeedPoolData{
+		Token0: tokenA, Token1: tokenB, FeeTier: 3000, TotalValueLockedUSD: "999.00",
+	})
+	s.SeedFactory("1", &storage.SeedFactoryData{PoolCount: 1, TotalValueLockedUSD: "999.00"})
+
+	idx := NewWithConfig(Config{RPC: "http://127.0.0.1:1"}, s)
+	idx.revalue(context.Background())
+
+	p, _ := s.GetPool(nil, pool)
+	if got := fmt.Sprint(p.(map[string]interface{})["totalValueLockedUSD"]); got != "999.00" {
+		t.Errorf("an unreachable RPC must not clobber TVL: got %q want 999.00", got)
+	}
+	f, _ := s.GetFactory(nil, "1")
+	if got := fmt.Sprint(f.(map[string]interface{})["totalValueLockedUSD"]); got != "999.00" {
+		t.Errorf("an unreachable RPC must not clobber factory TVL: got %q want 999.00", got)
+	}
+}
+
+// TestPriceTokens_PrefersTheDeeperQuote proves the depth ordering: when two
+// pools disagree on a token's price, the one with more value behind it wins.
+func TestPriceTokens_PrefersTheDeeperQuote(t *testing.T) {
+	const (
+		usdc = "0x0000000000000000000000000000000000000001"
+		usdt = "0x0000000000000000000000000000000000000002"
+		wlux = "0x0000000000000000000000000000000000000003"
+	)
+	tokens := map[string]*storage.SeedTokenData{
+		usdc: {Symbol: "USDC", Decimals: 18},
+		usdt: {Symbol: "USDT", Decimals: 18},
+		wlux: {Symbol: "WLUX", Decimals: 18},
+	}
+	vps := []valuedPool{
+		// thin: 100 WLUX vs 1 USDC  => $0.01
+		{id: "0xa", t0: wlux, t1: usdc, bal0: 100, bal1: 1},
+		// deep: 100 WLUX vs 50 USDT => $0.50
+		{id: "0xb", t0: wlux, t1: usdt, bal0: 100, bal1: 50},
+	}
+	prices := priceTokens(vps, tokens)
+	if got := prices[wlux]; math.Abs(got-0.5) > 1e-12 {
+		t.Errorf("WLUX price = %v, want 0.5 from the deeper pool (not 0.01 from the thin one)", got)
+	}
+}
+
+// hexWord encodes a whole-token amount at 18 decimals as a 32-byte hex word.
+func hexWord(whole int64) string {
+	n := new(big.Int).SetInt64(whole)
+	n.Mul(n, new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil))
+	return "0x" + fmt.Sprintf("%064s", n.Text(16))
+}
+
+func mustFloat(t *testing.T, v interface{}) float64 {
+	t.Helper()
+	f, err := strconv.ParseFloat(fmt.Sprint(v), 64)
+	if err != nil {
+		t.Fatalf("not a number: %v", v)
+	}
+	return f
+}

storage/store_memory.go

@@ -373,3 +373,43 @@ func (s *Store) GetPoolHourDatas(_ context.Context, limit int, orderBy, orderDir
 func (s *Store) GetTokenHourDatas(_ context.Context, limit int, orderBy, orderDirection string) (interface{}, error) {
 	return []interface{}{}, nil
 }
+
+// --- Raw accessors (valuation) ---
+// See the sqlite implementation for why the valuation pass needs the stored
+// value rather than the presentation map.
+
+// PoolsRaw returns every stored pool keyed by id.
+func (s *Store) PoolsRaw() map[string]*SeedPoolData {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := make(map[string]*SeedPoolData, len(s.pools))
+	for id, p := range s.pools {
+		c := *p
+		out[id] = &c
+	}
+	return out
+}
+
+// SwapsRaw returns every stored swap keyed by id.
+func (s *Store) SwapsRaw() map[string]*SeedSwapData {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := make(map[string]*SeedSwapData, len(s.swaps))
+	for id, sw := range s.swaps {
+		c := *sw
+		out[id] = &c
+	}
+	return out
+}
+
+// TokensRaw returns every stored token keyed by address.
+func (s *Store) TokensRaw() map[string]*SeedTokenData {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := make(map[string]*SeedTokenData, len(s.tokens))
+	for id, t := range s.tokens {
+		c := *t
+		out[id] = &c
+	}
+	return out
+}

storage/store_sqlite.go

@@ -558,3 +558,82 @@ func (s *Store) loadPool(id string) (*SeedPoolData, error) {
 	json.Unmarshal([]byte(raw), &p)
 	return &p, nil
 }
+
+// --- Raw accessors (valuation) ---
+//
+// The valuation pass (indexer/valuation.go) is the one owner of the derived USD
+// aggregates. It must read a pool/swap row, compute, and write the SAME row back
+// — and SeedPool/SeedSwap are INSERT OR REPLACE (full overwrite). Round-tripping
+// through GetPools/GetSwaps would be lossy: those return a *presentation* map
+// (tokens nested and dropped entirely when the token row is missing). These
+// accessors hand back the stored value itself so a read-modify-write preserves
+// every field the writer does not own.
+
+// PoolsRaw returns every stored pool keyed by id.
+func (s *Store) PoolsRaw() map[string]*SeedPoolData {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := map[string]*SeedPoolData{}
+	rows, err := s.db.Query("SELECT id, data FROM pools")
+	if err != nil {
+		return out
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id, raw string
+		if err := rows.Scan(&id, &raw); err != nil {
+			continue
+		}
+		var p SeedPoolData
+		if json.Unmarshal([]byte(raw), &p) == nil {
+			out[id] = &p
+		}
+	}
+	return out
+}
+
+// SwapsRaw returns every stored swap keyed by id.
+func (s *Store) SwapsRaw() map[string]*SeedSwapData {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := map[string]*SeedSwapData{}
+	rows, err := s.db.Query("SELECT id, data FROM swaps")
+	if err != nil {
+		return out
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id, raw string
+		if err := rows.Scan(&id, &raw); err != nil {
+			continue
+		}
+		var sw SeedSwapData
+		if json.Unmarshal([]byte(raw), &sw) == nil {
+			out[id] = &sw
+		}
+	}
+	return out
+}
+
+// TokensRaw returns every stored token keyed by address.
+func (s *Store) TokensRaw() map[string]*SeedTokenData {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	out := map[string]*SeedTokenData{}
+	rows, err := s.db.Query("SELECT id, data FROM tokens")
+	if err != nil {
+		return out
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id, raw string
+		if err := rows.Scan(&id, &raw); err != nil {
+			continue
+		}
+		var t SeedTokenData
+		if json.Unmarshal([]byte(raw), &t) == nil {
+			out[id] = &t
+		}
+	}
+	return out
+}