User volume touching NEAR itself

1-- Base tables. Inputs: raw parquet cache ({RAW}) plus the `assets` and `prices` tables registered by metrics.py
2-- from data/reference/. Nothing here is aggregated; every metric file below builds on these.
3
4-- forward-filled daily price per key (calendar x key), so a missing day reuses the last known price
5CREATE OR REPLACE TABLE prices_filled AS
6WITH cal AS (SELECT CAST(d AS DATE) AS date FROM generate_series(DATE '2024-12-31', DATE '2026-10-30', INTERVAL 1 DAY) t(d)),
7     keys AS (SELECT DISTINCT price_key FROM prices)
8SELECT k.price_key, c.date,
9       last_value(p.price IGNORE NULLS) OVER (PARTITION BY k.price_key ORDER BY c.date) AS price
10FROM keys k CROSS JOIN cal c LEFT JOIN prices p ON p.price_key = k.price_key AND p.date = c.date;
11
12-- one row per (token_diff entry, asset); entry_id identifies one token_diff entry (one account's side of a swap)
13CREATE OR REPLACE TABLE legs AS
14SELECT entry_id, CAST(ts AS DATE) AS date, ts, failed, intent_hash, account_id, receipt_id,
15       e.key AS asset, TRY_CAST(e.value AS DOUBLE) AS amount_raw
16FROM (SELECT row_number() OVER () AS entry_id, *,
17             unnest(map_entries(CAST(diff::JSON AS MAP(VARCHAR, VARCHAR)))) AS e
18      FROM read_parquet('{RAW}/swaps/*.parquet'));
19
20-- usd: priced from the price table. usd_imputed: for an entry with exactly ONE unpriced asset, the value implied
21-- by value conservation (minus the sum of the priced legs of the same entry), only if its sign matches the leg.
22-- usd_best = usd, else usd_imputed. Imputed values are always reported separately, never silently mixed in.
23CREATE OR REPLACE TABLE legs_usd AS
24WITH base AS (
25  SELECT l.*, a.symbol, a.origin_chain,
26         l.amount_raw / pow(10, a.decimals) AS qty,
27         CASE WHEN a.price_key = 'USD' THEN 1.0 ELSE p.price END AS price,
28         l.amount_raw / pow(10, a.decimals) * CASE WHEN a.price_key = 'USD' THEN 1.0 ELSE p.price END AS usd
29  FROM legs l
30  LEFT JOIN assets a ON a.asset = l.asset
31  LEFT JOIN prices_filled p ON p.price_key = a.price_key AND p.date = l.date),
32e AS (
33  SELECT *, sum(usd) OVER (PARTITION BY entry_id) AS priced_sum,
34         count(*) FILTER (WHERE usd IS NULL) OVER (PARTITION BY entry_id) AS n_unpriced
35  FROM base)
36SELECT * EXCLUDE (priced_sum, n_unpriced),
37       CASE WHEN usd IS NULL AND n_unpriced = 1 AND priced_sum IS NOT NULL
38                 AND sign(-priced_sum) = sign(amount_raw) THEN -priced_sum END AS usd_imputed,
39       coalesce(usd, CASE WHEN usd IS NULL AND n_unpriced = 1 AND priced_sum IS NOT NULL
40                               AND sign(-priced_sum) = sign(amount_raw) THEN -priced_sum END) AS usd_best
41FROM e;
42
43-- fees_collected per entry, priced the same way
44CREATE OR REPLACE TABLE fees_usd AS
45SELECT CAST(f.ts AS DATE) AS date, f.failed, f.e.key AS asset,
46       TRY_CAST(f.e.value AS DOUBLE) / pow(10, a.decimals) *
47         CASE WHEN a.price_key = 'USD' THEN 1.0 ELSE p.price END AS usd
48FROM (SELECT ts, failed, unnest(map_entries(CAST(fees::JSON AS MAP(VARCHAR, VARCHAR)))) AS e
49      FROM read_parquet('{RAW}/swaps/*.parquet') WHERE fees <> '{}') f
50LEFT JOIN assets a ON a.asset = f.e.key
51LEFT JOIN prices_filled p ON p.price_key = a.price_key AND p.date = CAST(f.ts AS DATE);
52
53-- NEP-245 mint/burn (deposits, withdrawals, refunds); token_ids and amounts are parallel arrays
54CREATE OR REPLACE TABLE mt_usd AS
55SELECT CAST(m.ts AS DATE) AS date, m.failed, m.event, m.memo, m.owner_id, m.asset,
56       TRY_CAST(m.amt AS DOUBLE) / pow(10, a.decimals) *
57         CASE WHEN a.price_key = 'USD' THEN 1.0 ELSE p.price END AS usd,
58       a.origin_chain, a.symbol, a.asset IS NULL AS unknown_asset
59FROM (SELECT ts, failed, event, memo, owner_id,
60             unnest(from_json(token_ids::JSON, '["VARCHAR"]')) AS asset,
61             unnest(from_json(amounts::JSON, '["VARCHAR"]')) AS amt
62      FROM read_parquet('{RAW}/mt/*.parquet')) m
63LEFT JOIN assets a ON a.asset = m.asset
64LEFT JOIN prices_filled p ON p.price_key = a.price_key AND p.date = CAST(m.ts AS DATE)
65WHERE m.asset LIKE 'nep141:%' OR m.asset LIKE 'nep245:%';
66
67
68-- Swap pairing: a receipt whose successful entries belong to exactly two accounts is one user<->solver swap.
69CREATE OR REPLACE TABLE pair_receipts AS
70SELECT receipt_id, min(account_id) AS a, max(account_id) AS b
71FROM (SELECT DISTINCT receipt_id, account_id FROM legs WHERE NOT failed)
72GROUP BY receipt_id HAVING count(*) = 2;
73
74-- Solver heuristic (documented in METHODOLOGY): an account is a solver if it faced >= 100 distinct counterparties
75-- across the window. Users face a handful of solvers; solvers face thousands of users. Not an on-chain label.
76CREATE OR REPLACE TABLE account_roles AS
77WITH sides AS (SELECT a AS account_id, b AS cp FROM pair_receipts UNION ALL SELECT b, a FROM pair_receipts)
78SELECT account_id, count(DISTINCT cp) AS n_counterparties, count(*) AS n_swaps,
79       CASE WHEN count(DISTINCT cp) >= 100 THEN 'solver' ELSE 'user' END AS role
80FROM sides GROUP BY account_id;