Skip to content

chore(jobs): drop volume accumulator from on-chain coin stats #1003

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rickyrombo merged 1 commit into from
Aug 3, 2026
Merged

chore(jobs): drop volume accumulator from on-chain coin stats #1003

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 0 additions & 6 deletions database/seed.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -534,12 +534,6 @@ var (
"timestamp": nil,
"price": 0.0,
},
"artist_coin_volume_accumulator": {
"mint": nil,
"last_processed_slot": 0,
"total_volume": 0.0,
"total_volume_usd": 0.0,
},
"sol_meteora_dbc_pools": {
"account": nil,
"slot": 1,
Expand Down
8 changes: 8 additions & 0 deletions ddl/migrations/0234_drop_artist_coin_volume_accumulator.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
-- Drop the artist coin volume accumulator. The all-time volume stat it fed was
-- removed from the product (unverifiable vendor figure; no other aggregator even
-- reports it), and CoinStatsOnchainJob no longer computes volume.
BEGIN;

DROP TABLE IF EXISTS artist_coin_volume_accumulator;

COMMIT;
89 changes: 4 additions & 85 deletions jobs/coin_stats_onchain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,12 +95,6 @@ type coinAgg struct {
Liquidity float64 `db:"liquidity"`
}

type volumeRow struct {
Mint string `db:"mint"`
TotalVolume float64 `db:"total_volume"`
TotalVolumeUsd float64 `db:"total_volume_usd"`
}

type priceRow struct {
Mint string `db:"mint"`
Price float64 `db:"price"`
Expand Down Expand Up @@ -138,16 +132,7 @@ func (j *CoinStatsOnchainJob) run(ctx context.Context) error {
return fmt.Errorf("error computing aggregates: %w", err)
}

// 2. Advance the all-time volume accumulator with trades in new slots.
if err := j.updateVolumeAccumulator(ctx, audioPrice); err != nil {
return fmt.Errorf("error updating volume accumulator: %w", err)
}
volumes, err := j.readVolumeAccumulator(ctx)
if err != nil {
return fmt.Errorf("error reading volume accumulator: %w", err)
}

// 3. Snapshot current on-chain prices (hourly bin) and read the ~24h-ago price.
// 2. Snapshot current on-chain prices (hourly bin) and read the ~24h-ago price.
if err := j.snapshotPrices(ctx, now); err != nil {
return fmt.Errorf("error snapshotting prices: %w", err)
}
Expand Down Expand Up @@ -194,9 +179,7 @@ func (j *CoinStatsOnchainJob) run(ctx context.Context) error {
}
}

vol := volumes[mint] // zero value if absent

if err := j.upsertStats(ctx, mint, agg, marketCap, totalSupply, vol, history, priceChange); err != nil {
if err := j.upsertStats(ctx, mint, agg, marketCap, totalSupply, history, priceChange); err != nil {
j.logger.Error("error upserting onchain stats", zap.String("mint", mint), zap.Error(err))
}
}
Expand Down Expand Up @@ -301,65 +284,6 @@ func (j *CoinStatsOnchainJob) queryAggregates(ctx context.Context, audioPrice fl
return out, nil
}

// updateVolumeAccumulator adds trading volume from balance changes in new slots
// (per-coin watermark) to the running accumulator, valued in USD at the current
// AUDIO price so historical USD isn't repriced at today's rate.
func (j *CoinStatsOnchainJob) updateVolumeAccumulator(ctx context.Context, audioPrice float64) error {
sql := `
WITH vaults AS (
SELECT base_mint AS mint, quote_vault AS vault FROM sol_meteora_dbc_pools
UNION
SELECT token_a_mint AS mint, token_b_vault AS vault FROM sol_meteora_damm_v2_pools
),
delta AS (
SELECT
v.mint,
SUM(ABS(c.change))::numeric / POWER(10, 8) AS vol_audio,
MAX(c.slot) AS max_slot
FROM vaults v
JOIN sol_token_account_balance_changes c
ON c.account = v.vault AND c.mint = @audio_mint
WHERE c.slot > COALESCE(
(SELECT last_processed_slot FROM artist_coin_volume_accumulator acc WHERE acc.mint = v.mint),
0
)
GROUP BY v.mint
)
INSERT INTO artist_coin_volume_accumulator (mint, last_processed_slot, total_volume, total_volume_usd, updated_at)
SELECT d.mint, d.max_slot, d.vol_audio, d.vol_audio * @audio_price, NOW()
FROM delta d
WHERE d.max_slot IS NOT NULL
ON CONFLICT (mint) DO UPDATE SET
total_volume = artist_coin_volume_accumulator.total_volume + EXCLUDED.total_volume,
total_volume_usd = artist_coin_volume_accumulator.total_volume_usd + EXCLUDED.total_volume_usd,
last_processed_slot = EXCLUDED.last_processed_slot,
updated_at = NOW()
`
_, err := j.pool.Exec(ctx, sql, pgx.NamedArgs{
"audio_mint": j.audioMint,
"audio_price": audioPrice,
})
return err
}

func (j *CoinStatsOnchainJob) readVolumeAccumulator(ctx context.Context) (map[string]volumeRow, error) {
rows, err := j.pool.Query(ctx,
`SELECT mint, total_volume, total_volume_usd FROM artist_coin_volume_accumulator`,
)
if err != nil {
return nil, err
}
list, err := pgx.CollectRows(rows, pgx.RowToStructByName[volumeRow])
if err != nil {
return nil, err
}
out := make(map[string]volumeRow, len(list))
for _, r := range list {
out[r.Mint] = r
}
return out, nil
}

// snapshotPrices records the current on-chain USD price for each coin in an
// hourly bin, used to compute the 24h price change.
func (j *CoinStatsOnchainJob) snapshotPrices(ctx context.Context, now time.Time) error {
Expand Down Expand Up @@ -481,18 +405,17 @@ func (j *CoinStatsOnchainJob) upsertStats(
agg coinAgg,
marketCap *float64,
totalSupply *float64,
vol volumeRow,
history24h *float64,
priceChange *float64,
) error {
sql := `
INSERT INTO artist_coin_stats_onchain (
mint, price, market_cap, liquidity, holder, total_supply,
total_volume, total_volume_usd, history_24h_price, price_change_24h_percent,
history_24h_price, price_change_24h_percent,
created_at, updated_at
) VALUES (
@mint, @price, @market_cap, @liquidity, @holder, @total_supply,
@total_volume, @total_volume_usd, @history_24h_price, @price_change_24h_percent,
@history_24h_price, @price_change_24h_percent,
NOW(), NOW()
)
ON CONFLICT (mint) DO UPDATE SET
Expand All @@ -502,8 +425,6 @@ func (j *CoinStatsOnchainJob) upsertStats(
liquidity = EXCLUDED.liquidity,
holder = EXCLUDED.holder,
total_supply = COALESCE(EXCLUDED.total_supply, artist_coin_stats_onchain.total_supply),
total_volume = EXCLUDED.total_volume,
total_volume_usd = EXCLUDED.total_volume_usd,
history_24h_price = COALESCE(EXCLUDED.history_24h_price, artist_coin_stats_onchain.history_24h_price),
price_change_24h_percent = COALESCE(EXCLUDED.price_change_24h_percent, artist_coin_stats_onchain.price_change_24h_percent),
updated_at = NOW()
Expand All @@ -515,8 +436,6 @@ func (j *CoinStatsOnchainJob) upsertStats(
"liquidity": agg.Liquidity,
"holder": agg.Holder,
"total_supply": totalSupply,
"total_volume": vol.TotalVolume,
"total_volume_usd": vol.TotalVolumeUsd,
"history_24h_price": history24h,
"price_change_24h_percent": priceChange,
})
Expand Down
15 changes: 2 additions & 13 deletions jobs/coin_stats_onchain_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,13 +79,6 @@ func TestCoinStatsOnchainJob(t *testing.T) {
{"account": "acct5", "mint": coinMint, "owner": "owner1", "balance": 3, "slot": 1},
{"account": "acct4", "mint": coinMint, "owner": "owner4", "balance": 0, "slot": 1},
},
// Volume through the DBC quote vault (AUDIO leg): |1e8| + |-2e8| = 3e8 -> 3 AUDIO.
"sol_token_account_balance_changes": {
{"signature": "sig1", "mint": audioMint, "owner": "trader1", "account": quoteVault,
"change": 100_000_000, "balance": 100_000_000, "slot": 10, "block_timestamp": "2024-01-01 00:00:00"},
{"signature": "sig2", "mint": audioMint, "owner": "trader2", "account": quoteVault,
"change": -200_000_000, "balance": 0, "slot": 11, "block_timestamp": "2024-01-01 00:01:00"},
},
// Active DBC pool -> liquidity = base_reserve*price + quote_reserve*audioPrice
// = (1e9/1e6)*4.0 + (5e10/1e8)*2.0 = 1000*4 + 500*2 = 5000.
"sol_meteora_dbc_pools": {
Expand Down Expand Up @@ -119,26 +112,22 @@ func TestCoinStatsOnchainJob(t *testing.T) {
liquidity float64
totalSupply float64
marketCap float64
totalVolume float64
totalVolUSD float64
history24h float64
priceChange24 float64
)
err := pool.QueryRow(ctx, `
SELECT price, holder, liquidity, total_supply, market_cap,
total_volume, total_volume_usd, history_24h_price, price_change_24h_percent
history_24h_price, price_change_24h_percent
FROM artist_coin_stats_onchain WHERE mint = $1`, coinMint).
Scan(&price, &holder, &liquidity, &totalSupply, &marketCap,
&totalVolume, &totalVolUSD, &history24h, &priceChange24)
&history24h, &priceChange24)
require.NoError(t, err)

assert.InDelta(t, 4.0, price, 1e-9, "price from pools_price_usd")
assert.Equal(t, 4, holder, "token accounts with balance > 0 (owner1 has two)")
assert.InDelta(t, 5000.0, liquidity, 1e-6, "TVL = base_usd + quote_usd")
assert.InDelta(t, 1000.0, totalSupply, 1e-9, "supply from RPC")
assert.InDelta(t, 4000.0, marketCap, 1e-6, "price * supply")
assert.InDelta(t, 3.0, totalVolume, 1e-9, "AUDIO volume through vault")
assert.InDelta(t, 6.0, totalVolUSD, 1e-9, "volume * audioPrice")
assert.InDelta(t, 2.0, history24h, 1e-9, "24h-ago snapshot")
assert.InDelta(t, 100.0, priceChange24, 1e-6, "24h percent change")

Expand Down
28 changes: 0 additions & 28 deletions sql/01_schema.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7735,26 +7735,6 @@ CREATE VIEW public.artist_coin_stats_comparison AS
COMMENT ON VIEW public.artist_coin_stats_comparison IS 'Compares Birdeye artist_coin_stats vs on-chain artist_coin_stats_onchain per coin (values + % diff) to validate CoinStatsOnchainJob before cutover.';


--
-- Name: artist_coin_volume_accumulator; Type: TABLE; Schema: public; Owner: -
--

CREATE TABLE public.artist_coin_volume_accumulator (
mint text NOT NULL,
last_processed_slot bigint DEFAULT 0 NOT NULL,
total_volume double precision DEFAULT 0 NOT NULL,
total_volume_usd double precision DEFAULT 0 NOT NULL,
updated_at timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL
);


--
-- Name: TABLE artist_coin_volume_accumulator; Type: COMMENT; Schema: public; Owner: -
--

COMMENT ON TABLE public.artist_coin_volume_accumulator IS 'Running per-coin trading volume (AUDIO + USD) accumulated from pool quote-vault balance changes, watermarked by last_processed_slot. Written by CoinStatsOnchainJob.';


--
-- Name: associated_wallets; Type: TABLE; Schema: public; Owner: -
--
Expand Down Expand Up @@ -11387,14 +11367,6 @@ ALTER TABLE ONLY public.artist_coin_stats
ADD CONSTRAINT artist_coin_stats_pkey PRIMARY KEY (mint);


--
-- Name: artist_coin_volume_accumulator artist_coin_volume_accumulator_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--

ALTER TABLE ONLY public.artist_coin_volume_accumulator
ADD CONSTRAINT artist_coin_volume_accumulator_pkey PRIMARY KEY (mint);


--
-- Name: artist_coins artist_coins_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
Expand Down
Loading