lux/graph
2
0
Fork 0
mirror of https://github.com/luxfi/graph.git synced 2026-09-24 00:26:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

All-time volume is a sum over the swap table, not over a window

Browse Source 
v1.2.47 replaced one window sum with the day series added up. Those rows are
folded from the same window, so their sum is that window again — a change that
read correctly and moved nothing. Live proof: the day series summed to exactly
the pass's own figure, and its trade counts to exactly the window size.

The value of a trade has always lived inside the swap's JSON document, where SQL
cannot reach it — this driver's amalgamation carries no JSON1, so json_extract
is not an escape either. So amountUSD joins timestamp and pool as a column: the
queryable projection of the document, written by both paths that touch a swap,
summed by one reader. What the protocol has traded is now one scan of one column
and covers every trade ever indexed.

A database that predates the column gets it on the start that introduces it,
filled from the rows themselves. Empty, it would have zeroed every trade older
than the current window — the fix erasing the number it exists to correct — so
that is the case the test pins, and the ALTER failing is how a later start knows
the fill is done.

txCount is left where it is. It counts every interaction, mints and burns among
them, which is a different question from what has traded.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zeekay
2026-08-14 06:48:41 -07:00
co-authored by Hanzo Dev
parent 84d71e5a98
commit d3e86b3eb8
6 changed files with 298 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
LLM.md
+23
View File
@@ -63,6 +63,29 @@ which is exactly what `healSwapTimes` needs to look the real time up. That is
`Dated()`, and both the heal and the day series ask it. The heal runs once at
`Run` and stops when every row has a time.
## A swap's columns are its queryable projection, and all-time volume is a SUM
`swaps` is `(id, data JSON, timestamp, pool, amountUSD)`. The three scalars are
copies of fields inside `data`, promoted so the database can answer questions Go
would otherwise answer by reading the table into memory: a window (`timestamp`),
a pool's own history (`pool`), and what the protocol has traded (`amountUSD`).
`SeedSwap` and `ValueSwaps` write both homes; `Store.Traded()` is the only reader
of the sum. 🪤 There is no `json_extract` escape hatch — this driver's
amalgamation is built without JSON1, which is why the column had to exist at all.
A database written before the column gets it in `Init`, filled once from the rows
themselves; the ALTER failing is how a later start knows not to do it again.
Arriving empty would have zeroed every trade older than the current window — the
fix erasing the number it exists to correct.
**Total volume is not the pass's total.** `maxValuedSwaps` bounds what one
valuation pass values, because the swap table grows forever. Summing that window
answers for the newest trades only. Summing the day series does not escape it
either: those rows are folded from the same window, so their sum IS the window —
a fix that looked right and changed nothing. `Factory.txCount` is a separate
question (every interaction, mints and burns included) and stays with the
handlers that see them.
## The day series (series.go) is a SECOND FOLD over the valuation pass's trades
`tokenDayDatas` / `poolDayDatas` / `uniswapDayDatas` are rolled out of the swaps
indexer/supply_test.go
+19 -14
View File
@@ -111,29 +111,34 @@ func TestNativeSupplySurvivesAStaleCopy(t *testing.T) {
}
}
// Volume for all time is every day added up, not the newest window summed.
// Volume for all time is the whole table, not the window a pass reads.
//
// A pass values a bounded slice of the swap table because that table grows
// forever. The bound is right for the pass and wrong for the field the wire
// calls total volume: on a chain with a million swaps it reported the last
// twelve percent and called it everything. The day rows are written once and
// kept, so their sum is the whole history at the cost of a row per day.
func TestFactoryVolumeCountsEveryDay(t *testing.T) {
// calls total volume. Summing the day series does not escape it — those rows
// are folded from the same window, so their sum is that window again. This
// gives the store more trades than any window would carry and asks it what has
// traded.
func TestFactoryVolumeCoversEveryTrade(t *testing.T) {
s := newMemSQLiteStore(t)
idx := NewWithConfig(Config{RPC: chainAt(t, 1767225600, nil).URL}, s)
// Three days the indexer has seen, only one of which a recent window covers.
for i, v := range []string{"1000.00", "2000.00", "500.00"} {
s.SetEntity(storage.FactoryDay, fmt.Sprintf("1-%d", 20000+i), map[string]interface{}{
"id": fmt.Sprintf("1-%d", 20000+i), "date": (20000 + i) * 86400, "volumeUSD": v,
for i, usd := range []string{"1000.00", "2000.00", "500.00"} {
s.SeedSwap(fmt.Sprintf("0xabc#%d", i), &storage.SeedSwapData{
Timestamp: int64(1767225600 + i*86400),
Block: uint64(100 + i),
Pool: "0xpool",
AmountUSD: usd,
})
}
got, err := idx.factoryVolume()
traded, volume, err := s.Traded()
if err != nil {
t.Fatalf("factoryVolume: %v", err)
t.Fatalf("Traded: %v", err)
}
if got != 3500 {
t.Errorf("volume = %v, want 3500 — a day outside the window still traded", got)
if traded != 3 {
t.Errorf("trades = %d, want 3", traded)
}
if volume != 3500 {
t.Errorf("volume = %v, want 3500 \u2014 a trade outside the window still traded", volume)
}
}
indexer/valuation.go
+20 -54
View File
@@ -270,27 +270,26 @@ func (idx *Indexer) revalue(parent context.Context) {
// 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.
//
// txCount counts every interaction — mints and burns as well as trades — and
// is carried through from the handlers that see them. Volume is a different
// question and gets a different answer: what the whole table has traded.
//
// A pass values a bounded slice of that table, newest first, because it grows
// forever and rescanning it would cost more every day the chain lives. The
// bound is right for the pass and wrong for a figure the wire calls all-time.
// Summing the day series does not escape it either — those rows are folded
// from the same window, so their sum IS the window. Summing a column costs
// one scan and answers for every trade ever indexed.
f, _ := idx.store.GetFactory(nil, "1")
var txCount int64
if m, ok := f.(map[string]interface{}); ok {
txCount = asInt64(m["txCount"])
}
// Volume for all time is the days added up, not one window summed.
//
// A pass values a bounded slice of the swap table, newest first, because the
// table grows forever and rescanning it would cost more every day the chain
// lives. That bound is right for the pass and wrong for this field: the wire
// calls it total volume, and totalling a window answers for the newest
// trades only — on a chain with a million swaps it reported the last
// twelve percent and called it everything.
//
// The day series already holds each day's volume, written once and kept, so
// the sum of those rows is the whole history at the cost of reading a row
// per day. The window stays exactly as it is; only what is claimed of it
// changes.
volume := totalVol
if allDays, err := idx.factoryVolume(); err == nil && allDays > volume {
volume = allDays
traded, volume, err := idx.store.Traded()
if err != nil {
idx.logf("[valuation] all-time volume unavailable, reporting this window: %v", err)
traded, volume = int64(len(trades)), totalVol
}
idx.store.SeedFactory("1", &storage.SeedFactoryData{
PoolCount: int64(len(pools)),
@@ -312,10 +311,6 @@ func (idx *Indexer) revalue(parent context.Context) {
tokens: tokens,
})
window := ""
if len(trades) >= maxValuedSwaps {
window = fmt.Sprintf(" [volume covers the most recent %d swaps only]", maxValuedSwaps)
}
// Say what the swap write actually did. A pass that reports its pools and
// then writes nothing to the trades underneath them is how a pool came to
// claim thousands in volume over a list of swaps each worth zero.
@@ -323,9 +318,12 @@ func (idx *Indexer) revalue(parent context.Context) {
if valuedErr != nil {
priced = fmt.Sprintf(", SWAP PRICES NOT WRITTEN: %v", valuedErr)
}
idx.logf("[valuation] %d/%d pools valued, %d tokens priced, %d swaps — TVL $%s, volume $%s%s (%s)%s",
// Both volumes, because they answer different questions and a reader who
// sees one alone cannot tell which he has. The first is what this pass just
// valued; the second is what the chain has traded and what is served.
idx.logf("[valuation] %d/%d pools valued, %d tokens priced, %d swaps — TVL $%s, volume $%s of $%s over %d trades%s (%s)",
len(vps), len(pools), len(prices), len(trades), fmtUSD(totalTVL), fmtUSD(totalVol),
priced, time.Since(started).Round(time.Millisecond), window)
fmtUSD(volume), traded, priced, time.Since(started).Round(time.Millisecond))
}
// readBalances asks each pool's two tokens what the pool holds. One eth_call
@@ -682,35 +680,3 @@ func fmtRatio(num, den float64) string {
}
return strconv.FormatFloat(num/den, 'g', 12, 64)
}
// factoryVolume adds up every day the protocol has traded.
//
// Each row was written by the pass that saw that day and has been kept since,
// so this is the whole history without rereading a swap. A day the indexer
// never saw contributes nothing, which is honest: it can only report what it
// has been shown.
func (idx *Indexer) factoryVolume() (float64, error) {
rows, err := idx.store.Entities(storage.FactoryDay, maxValuedPools, "date", "asc", nil)
if err != nil {
return 0, err
}
list, ok := rows.([]interface{})
if !ok {
return 0, nil
}
var sum float64
for _, r := range list {
m, ok := r.(map[string]interface{})
if !ok {
continue
}
// The same overflow guard the rest of this pass uses (1e12): a tiny pool
// can produce a decimal artefact that would otherwise become the headline.
v, err := strconv.ParseFloat(fmt.Sprint(m["volumeUSD"]), 64)
if err != nil || !(v > 0) || v > 1e12 {
continue
}
sum += v
}
return sum, nil
}
storage/store_memory.go
+23
View File
@@ -5,7 +5,9 @@ package storage
import (
"context"
"encoding/json"
"math"
"sort"
"strconv"
"strings"
"sync"
)
@@ -439,6 +441,27 @@ func (s *Store) PoolsRaw() map[string]*SeedPoolData {
return out
}
// Traded reports every trade the store holds and what they were worth.
func (s *Store) Traded() (int64, float64, error) {
s.mu.RLock()
defer s.mu.RUnlock()
var volume float64
for _, sw := range s.swaps {
volume += dollars(sw.AmountUSD)
}
return int64(len(s.swaps)), volume, nil
}
// dollars reads a formatted dollar figure back as a number. Anything unreadable
// is worth nothing, which is what an unpriced swap already says.
func dollars(s string) float64 {
v, err := strconv.ParseFloat(s, 64)
if err != nil || math.IsNaN(v) || math.IsInf(v, 0) {
return 0
}
return v
}
// RecentSwapsRaw returns the `limit` most recent stored swaps keyed by id.
// See the sqlite implementation for why the bound exists.
func (s *Store) RecentSwapsRaw(limit int) map[string]*SeedSwapData {
storage/store_sqlite.go
+102 -7
View File
@@ -7,6 +7,7 @@ import (
"database/sql"
"encoding/json"
"fmt"
"math"
"os"
"path/filepath"
"strconv"
@@ -68,7 +69,7 @@ func (s *Store) Init(_ context.Context) error {
CREATE TABLE IF NOT EXISTS bundles (id TEXT PRIMARY KEY, data JSON);
CREATE TABLE IF NOT EXISTS tokens (id TEXT PRIMARY KEY, data JSON);
CREATE TABLE IF NOT EXISTS pools (id TEXT PRIMARY KEY, data JSON);
CREATE TABLE IF NOT EXISTS swaps (id TEXT PRIMARY KEY, data JSON, timestamp INTEGER, pool TEXT);
CREATE TABLE IF NOT EXISTS swaps (id TEXT PRIMARY KEY, data JSON, timestamp INTEGER, pool TEXT, amountUSD REAL NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS entities (type TEXT, id TEXT, data JSON, PRIMARY KEY(type, id));
CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT);
@@ -76,8 +77,102 @@ func (s *Store) Init(_ context.Context) error {
CREATE INDEX IF NOT EXISTS idx_swaps_pool ON swaps(pool);
CREATE INDEX IF NOT EXISTS idx_entities_type ON entities(type);
`
_, err := s.db.Exec(schema)
return err
if _, err := s.db.Exec(schema); err != nil {
return err
}
// A swap's columns are the queryable projection of its document. Timestamp
// and pool were promoted so a window and a pool's history could be asked of
// the database instead of of Go; amountUSD is promoted for the same reason.
// What the protocol has traded for all time is a SUM over this table, and a
// figure that exists only inside a JSON document cannot be summed — this
// driver's amalgamation is built without JSON1, so json_extract is not an
// option here either (see ValueSwaps).
//
// A database written before the column gets it now, filled once from the
// rows themselves. Every write since keeps the two in step, so the fill runs
// exactly once per database and the ALTER failing IS how it knows.
if _, err := s.db.Exec(`ALTER TABLE swaps ADD COLUMN amountUSD REAL NOT NULL DEFAULT 0`); err == nil {
return s.projectAmounts()
}
return nil
}
// projectAmounts copies each stored swap's dollar value out of its document and
// into the column beside it.
func (s *Store) projectAmounts() error {
rows, err := s.db.Query(`SELECT id, data FROM swaps`)
if err != nil {
return fmt.Errorf("storage: read swaps to project: %w", err)
}
type valued struct {
id string
usd float64
}
// Drained whole before the write opens, because a read held open across a
// write on the same file is how SQLite is asked to deadlock with itself.
var priced []valued
for rows.Next() {
var id, raw string
if err := rows.Scan(&id, &raw); err != nil {
continue
}
var d SeedSwapData
if json.Unmarshal([]byte(raw), &d) != nil {
continue
}
if usd := dollars(d.AmountUSD); usd != 0 {
priced = append(priced, valued{id, usd})
}
}
rows.Close()
if len(priced) == 0 {
return nil
}
tx, err := s.db.Begin()
if err != nil {
return fmt.Errorf("storage: project amounts: %w", err)
}
stmt, err := tx.Prepare(`UPDATE swaps SET amountUSD = ? WHERE id = ?`)
if err != nil {
tx.Rollback()
return fmt.Errorf("storage: project amounts: %w", err)
}
defer stmt.Close()
for _, v := range priced {
if _, err := stmt.Exec(v.usd, v.id); err != nil {
tx.Rollback()
return fmt.Errorf("storage: project %s: %w", v.id, err)
}
}
return tx.Commit()
}
// dollars reads a formatted dollar figure back as a number. Anything unreadable
// is worth nothing, which is what an unpriced swap already says.
func dollars(s string) float64 {
v, err := strconv.ParseFloat(s, 64)
if err != nil || math.IsNaN(v) || math.IsInf(v, 0) {
return 0
}
return v
}
// Traded reports every trade the store holds and what they were worth.
//
// Over the whole table, not over a window: a valuation pass reads a bounded
// slice of this table because the table grows forever, and that bound is right
// for the pass and wrong for a figure the wire calls all-time. Summing a column
// costs one scan and answers for every trade ever indexed.
func (s *Store) Traded() (int64, float64, error) {
s.mu.RLock()
defer s.mu.RUnlock()
var count int64
var volume float64
err := s.db.QueryRow(`SELECT COUNT(*), COALESCE(SUM(amountUSD), 0) FROM swaps`).Scan(&count, &volume)
if err != nil {
return 0, 0, fmt.Errorf("storage: total traded: %w", err)
}
return count, volume, nil
}
// Close closes the database.
@@ -182,8 +277,8 @@ func (s *Store) SeedSwap(id string, d *SeedSwapData) {
s.mu.Lock()
defer s.mu.Unlock()
data, _ := json.Marshal(d)
s.db.Exec("INSERT OR REPLACE INTO swaps(id, data, timestamp, pool) VALUES(?, ?, ?, ?)",
id, string(data), d.Timestamp, d.Pool)
s.db.Exec("INSERT OR REPLACE INTO swaps(id, data, timestamp, pool, amountUSD) VALUES(?, ?, ?, ?, ?)",
id, string(data), d.Timestamp, d.Pool, dollars(d.AmountUSD))
}
// ValueSwaps writes what each trade was worth onto the stored swap.
@@ -220,7 +315,7 @@ func (s *Store) ValueSwaps(usd map[string]string) (int64, error) {
return 0, fmt.Errorf("prepare read: %w", err)
}
defer read.Close()
write, err := tx.Prepare(`UPDATE swaps SET data = ? WHERE id = ?`)
write, err := tx.Prepare(`UPDATE swaps SET data = ?, amountUSD = ? WHERE id = ?`)
if err != nil {
tx.Rollback()
return 0, fmt.Errorf("prepare write: %w", err)
@@ -248,7 +343,7 @@ func (s *Store) ValueSwaps(usd map[string]string) (int64, error) {
tx.Rollback()
return 0, fmt.Errorf("encode %s: %w", id, err)
}
res, err := write.Exec(string(next), id)
res, err := write.Exec(string(next), dollars(v), id)
if err != nil {
tx.Rollback()
return 0, fmt.Errorf("write %s: %w", id, err)
storage/traded_test.go
+111
View File
@@ -0,0 +1,111 @@
//go:build !nosqlite
package storage
import (
"context"
"encoding/json"
"testing"
)
// A database written before the column has to arrive holding its history.
//
// The dollar value of a trade has always been stored inside the swap's JSON
// document, where SQL cannot reach it — this driver's amalgamation carries no
// JSON1, so json_extract is not available to read it either. Promoting it to a
// column is what lets the protocol's all-time volume be a SUM. A chain that has
// been indexing for a year meets that column for the first time on the start
// that introduces it, and if it arrived empty every trade before that moment
// would read as worthless: the fix would zero the very number it exists to
// correct.
func TestTradedSurvivesTheColumnArriving(t *testing.T) {
dir := t.TempDir()
// A store as it was: swaps with no amountUSD column, values in the document.
old, err := New(dir)
if err != nil {
t.Fatalf("open store: %v", err)
}
if _, err := old.db.Exec(`
CREATE TABLE swaps (id TEXT PRIMARY KEY, data JSON, timestamp INTEGER, pool TEXT);
CREATE INDEX idx_swaps_timestamp ON swaps(timestamp);
`); err != nil {
t.Fatalf("old schema: %v", err)
}
for id, usd := range map[string]string{"s1": "1000.00", "s2": "2000.00", "s3": "500.00"} {
row, _ := json.Marshal(&SeedSwapData{Timestamp: 100, Pool: "0xpool", AmountUSD: usd})
if _, err := old.db.Exec(`INSERT INTO swaps(id, data, timestamp, pool) VALUES(?,?,?,?)`,
id, string(row), 100, "0xpool"); err != nil {
t.Fatalf("seed %s: %v", id, err)
}
}
old.Close()
s, err := New(dir)
if err != nil {
t.Fatalf("reopen store: %v", err)
}
defer s.Close()
if err := s.Init(context.Background()); err != nil {
t.Fatalf("init schema: %v", err)
}
traded, volume, err := s.Traded()
if err != nil {
t.Fatalf("Traded: %v", err)
}
if traded != 3 {
t.Errorf("trades = %d, want 3", traded)
}
if volume != 3500 {
t.Errorf("volume = %v, want 3500 — the history was already in the rows", volume)
}
// And the fill runs once: a second Init must not undo a value the pass has
// since written, which is what re-reading the documents on every start would
// do to any row whose column has moved ahead of them.
if _, err := s.ValueSwaps(map[string]string{"s1": "1500.00"}); err != nil {
t.Fatalf("ValueSwaps: %v", err)
}
if err := s.Init(context.Background()); err != nil {
t.Fatalf("re-init: %v", err)
}
if _, volume, _ = s.Traded(); volume != 4000 {
t.Errorf("volume = %v, want 4000 after repricing s1", volume)
}
}
// Pricing a trade moves the column with the document. They are one value in two
// places — the document is the wire shape, the column is what SQL can add up —
// and a write that moves only one of them makes the total disagree with the
// trades it is a total of.
func TestValueSwapsMovesTheColumn(t *testing.T) {
s, err := New(t.TempDir())
if err != nil {
t.Fatalf("open store: %v", err)
}
defer s.Close()
if err := s.Init(context.Background()); err != nil {
t.Fatalf("init schema: %v", err)
}
s.SeedSwap("s1", &SeedSwapData{Timestamp: 100, Pool: "0xpool", AmountUSD: "0"})
if _, _, err := traded(t, s); err != nil {
t.Fatal(err)
}
if _, err := s.ValueSwaps(map[string]string{"s1": "42.50"}); err != nil {
t.Fatalf("ValueSwaps: %v", err)
}
count, volume, err := traded(t, s)
if err != nil {
t.Fatal(err)
}
if count != 1 || volume != 42.5 {
t.Errorf("traded = %d/%v, want 1/42.5", count, volume)
}
}
func traded(t *testing.T, s *Store) (int64, float64, error) {
t.Helper()
return s.Traded()
}