Implied confidential share, weekly

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;

1"""Simple value-capture model: Intents volume -> NEAR captured revenue (buyback) -> yield on NEAR market cap.
2
3  python -m pipeline.model
4
5Observed side (computed): trailing-window public volume from the pipeline, total volume / fees / captured revenue
6from the DefiLlama snapshot, gas burned by intents.near, NEAR price and total supply.
7Scenario side: data/reference/model_assumptions.json (volume multiples, take-rate multiples).
8Writes data/metrics/model_observed.csv and data/metrics/model_scenarios.csv.
9"""
10import json
11
12import pandas as pd
13
14from pipeline import config
15from pipeline.assets import REF
16
17M = config.ROOT / "data" / "metrics"
18
19
20def confidential_residual(vol: pd.DataFrame, dl: pd.DataFrame, end: pd.Timestamp) -> None:
21    """Implied confidential share = 1 - public on-chain leg-sum / DefiLlama reported volume (a residual, not a measurement).
22
23    Writes derived ratios only (never DefiLlama's own series): weekly share, and one row of facts about where the share
24    breaks out of its pre-September level, so the thesis can test the "migration to Confidential Intents" explanation.
25    """
26    v = vol[vol["date"] <= end].copy()
27    v["pub"] = v["leg_sum_priced_usd"] + v["leg_sum_imputed_usd"].fillna(0)
28    df = v.merge(dl[["date", "volume_usd"]], on="date").query("volume_usd > 0").set_index("date").sort_index()
29    wk = df.groupby(df.index - pd.to_timedelta(df.index.weekday, unit="D")).agg(days=("pub", "size"), pub=("pub", "sum"), dl=("volume_usd", "sum"))
30    wk = wk[wk["days"] == 7]
31    pd.DataFrame({"week_start": wk.index.date, "days": wk["days"], "implied_confidential_share": 1 - wk["pub"] / wk["dl"]}) \
32        .to_csv(M / "confidential_weekly.csv", index=False, float_format="%.6g")
33
34    share = lambda x: 1 - x["pub"].sum() / x["volume_usd"].sum()
35    daily = 1 - df["pub"] / df["volume_usd"]
36    brk = None
37    for d in df.index[df.index >= df.index.min() + pd.Timedelta(days=14)]:
38        before, after = df.loc[d - pd.Timedelta(days=14): d - pd.Timedelta(days=1)], df.loc[d: d + pd.Timedelta(days=13)]
39        if len(after) == 14 and daily[d] >= 0.05 and abs(share(before)) < 0.02 and share(after) >= 0.05:
40            brk = (d, before, after)
41            break
42    rows = {}
43    if brk:
44        d, before, after = brk
45        later = df.loc[d + pd.Timedelta(days=14): d + pd.Timedelta(days=41)]
46        rows = {
47            "break_date": d.date(), "share_14d_before": share(before), "share_14d_after": share(after),
48            "public_per_day_change": after["pub"].mean() / before["pub"].mean() - 1,
49            "reported_per_day_change": after["volume_usd"].mean() / before["volume_usd"].mean() - 1,
50            "intents_per_day_14d_before": before["intents"].mean(), "intents_per_day_14d_after": after["intents"].mean(),
51            "intents_per_day_days_15_to_42": later["intents"].mean(),
52            "share_month_before_break": share(df[df.index.strftime("%Y-%m") == (d - pd.Timedelta(days=1)).strftime("%Y-%m")]),
53        }
54    pd.Series(rows, name="value").to_frame().to_csv(M / "confidential_break.csv", float_format="%.6g", index_label="key")
55
56
57def main() -> None:
58    a = json.loads((REF / "model_assumptions.json").read_text())
59    vol = pd.read_csv(M / "daily_volume.csv", parse_dates=["date"])
60    gas = pd.read_csv(M / "fees_and_gas.csv", parse_dates=["date"])
61    dl = pd.read_csv(REF / "defillama_near_intents.csv", parse_dates=["date"])
62    sup = json.loads((REF / "near_supply.json").read_text())
63    supply = sup["total_supply_near"]
64
65    end = vol["date"].max() - pd.Timedelta(days=1)  # last cache day is partial; drop it
66    start = end - pd.Timedelta(days=a["window_days"] - 1)
67    win = lambda d: d[(d["date"] >= start) & (d["date"] <= end)]
68    v, g, d = win(vol), win(gas), win(dl)
69    n = a["window_days"]
70    near_px = sup["near_price_usd"]  # DefiLlama price at the window end (see near_price_date)
71    mcap = near_px * supply
72
73    pipe_public_legsum = v["leg_sum_priced_usd"].sum() + v["leg_sum_imputed_usd"].fillna(0).sum()
74    dl_total = d["volume_usd"].sum()
75    fees, rev = d["fees_usd"].sum(), d["revenue_usd"].sum()
76    obs = {
77        "window_start": start.date(), "window_end": end.date(), "days": n,
78        "pipeline_public_legsum_usd": pipe_public_legsum,
79        "pipeline_public_net_swap_usd": pipe_public_legsum / 2,
80        "defillama_total_volume_usd": dl_total,
81        "implied_confidential_share": 1 - pipe_public_legsum / dl_total,
82        "gross_fees_usd": fees, "captured_revenue_usd": rev,
83        "gross_fee_bps_of_reported_volume": 1e4 * fees / dl_total,
84        "captured_share_of_gross_fees": rev / fees,
85        "captured_bps_of_reported_volume": 1e4 * rev / dl_total,
86        "gas_burned_near": g["tokens_burnt_near"].sum(), "gas_burned_usd": g["tokens_burnt_usd"].sum(),
87        "near_price_usd": near_px, "near_total_supply": supply, "near_mcap_usd": mcap,
88        "annualized_revenue_usd": rev * 365 / n,
89        "annualized_buyback_yield_on_mcap": rev * 365 / n / mcap,
90        "annualized_gas_burn_usd": g["tokens_burnt_usd"].sum() * 365 / n,
91    }
92    pd.DataFrame([obs]).T.rename(columns={0: "value"}).to_csv(M / "model_observed.csv", float_format="%.8g")
93
94    rows = []
95    base_vol = dl_total * 365 / n
96    base_take = rev / dl_total
97    for name, mult in a["revenue_take_rate_scenarios"].items():
98        take = base_take * (1.0 if mult is None else mult)
99        for vm in a["volume_multiples"]:
100            for pm in a["near_price_multiples"]:
101                annual_rev = base_vol * vm * take
102                rows.append({"take_rate_case": name, "volume_multiple": vm, "near_price_multiple": pm,
103                             "annual_volume_usd": base_vol * vm, "take_bps": 1e4 * take,
104                             "annual_buyback_usd": annual_rev,
105                             "buyback_pct_of_mcap": annual_rev / (mcap * pm)})
106    pd.DataFrame(rows).to_csv(M / "model_scenarios.csv", index=False, float_format="%.8g")
107    confidential_residual(vol, dl, end)
108    print(pd.DataFrame([obs]).T.to_string(header=False))
109
110
111if __name__ == "__main__":
112    main()