Pair OHLCV

-- Trade fact: one row per fill, sign-carrying for reorg collapse.
CREATE TABLE recipe_pair_ohlcv.fact_swaps
(
    vendor_event_id     String,
    ingested_at         DateTime64(3),
    chain_id            UInt32,
    block_hash          String,
    block_number        UInt64,
    event_ts            DateTime64(3),
    pool_address        String,        -- pair (pool_fill) or aggregator
    token0_address      String,
    token1_address      String,
    amount0             String,        -- raw uint256
    amount1             String,        -- raw uint256
    trader_address      String,
    price_usd           Nullable(String),   -- bought-leg (token1) USD price
    notional_usd        Nullable(String),   -- signed token1 amount × price
    side                LowCardinality(String),   -- buy | sell
    source_kind         LowCardinality(String),   -- aggregator | pool_fill
    protocol            LowCardinality(String),
    sign                Int8
)
ENGINE = ReplicatedCollapsingMergeTree(
    '/clickhouse/tables/{database}/fact_swaps', '{replica}', sign)
PARTITION BY (chain_id, toYYYYMM(event_ts))
ORDER BY (chain_id, pool_address, event_ts, vendor_event_id);

-- Candle accelerator: one state-row per (chain, pool, hour).
-- Additive measures are sign-weighted so reorg −1 rows subtract cleanly.
CREATE TABLE recipe_pair_ohlcv.candles_agg
(
    chain_id        UInt32,
    pool_address    String,
    bucket_start    DateTime,                       -- toStartOfHour(event_ts)
    open_state      AggregateFunction(argMin, Float64, DateTime64(3)),
    high_state      AggregateFunction(max,    Float64),
    low_state       AggregateFunction(min,    Float64),
    close_state     AggregateFunction(argMax, Float64, DateTime64(3)),
    volume_state    AggregateFunction(sum,    Float64),
    trades_state    AggregateFunction(sum,    Int64)
)
ENGINE = ReplicatedAggregatingMergeTree(
    '/clickhouse/tables/{database}/candles_agg', '{replica}')
PARTITION BY (chain_id, toYYYYMM(bucket_start))
ORDER BY (chain_id, pool_address, bucket_start);

-- candles VIEW: finalizes the aggregate state to scalars (the fast surface).
CREATE VIEW recipe_pair_ohlcv.candles AS
SELECT
    chain_id,
    pool_address,
    bucket_start,
    argMinMerge(open_state)  AS open,
    maxMerge(high_state)     AS high,
    minMerge(low_state)      AS low,
    argMaxMerge(close_state) AS close,
    sumMerge(volume_state)   AS volume,
    sumMerge(trades_state)   AS trades
FROM recipe_pair_ohlcv.candles_agg
GROUP BY chain_id, pool_address, bucket_start
HAVING trades > 0;

-- candles_exact VIEW: reorg-EXACT — FINAL collapses each +1/−1 pair before
-- aggregating, so OHLC reflects only canonical trades. Slower; use after a reorg
-- may have retracted an hour's all-time high/low.
CREATE VIEW recipe_pair_ohlcv.candles_exact AS
SELECT
    chain_id,
    pool_address,
    toStartOfHour(event_ts)                      AS bucket_start,
    argMin(toFloat64OrZero(price_usd), event_ts) AS open,
    max(toFloat64OrZero(price_usd))              AS high,
    min(toFloat64OrZero(price_usd))              AS low,
    argMax(toFloat64OrZero(price_usd), event_ts) AS close,
    sum(abs(toFloat64OrZero(notional_usd)))      AS volume,  -- absolute traded USD
    count()                                      AS trades
FROM recipe_pair_ohlcv.fact_swaps FINAL
WHERE price_usd IS NOT NULL AND price_usd != ''
GROUP BY chain_id, pool_address, bucket_start;

-- Flat one-row-per-trade landing table (the Swaps by Pair ingest).
CREATE TABLE public.swaps (
  position           BIGINT      NOT NULL,
  log_index          BIGINT,                 -- NULL for aggregator rows
  transaction_index  BIGINT,
  block_number       BIGINT      NOT NULL,
  block_timestamp    BIGINT      NOT NULL,    -- unix seconds
  tx_hash            TEXT        NOT NULL,
  vendor_event_id    TEXT        NOT NULL,
  trader_address     TEXT        NOT NULL,
  pool_address       TEXT        NOT NULL,    -- pair (pool_fill) or aggregator
  token0_address     TEXT        NOT NULL,
  token1_address     TEXT        NOT NULL,
  amount0            NUMERIC(76, 0)  NOT NULL,
  amount1            NUMERIC(76, 0)  NOT NULL,
  price_usd          NUMERIC(38, 18),
  notional_usd       NUMERIC(38, 18),
  side               TEXT        NOT NULL,
  source_kind        TEXT        NOT NULL,    -- aggregator | pool_fill
  protocol           TEXT        NOT NULL
);

-- Candle aggregation reads pool_address + block_timestamp.
CREATE INDEX swaps_pool_ts_idx ON public.swaps (pool_address, block_timestamp);
CREATE INDEX swaps_block_idx   ON public.swaps (block_number);

-- candles: always-fresh 1-hour OHLCV VIEW (the read surface).
-- Only priced trades participate; open/close use the array_agg-ORDER-BY idiom.
CREATE VIEW public.candles AS
SELECT
  pool_address,
  date_trunc('hour', to_timestamp(block_timestamp)) AS bucket_start,
  (array_agg(price_usd ORDER BY block_timestamp ASC,  vendor_event_id ASC))[1]  AS open,
  max(price_usd)                                     AS high,
  min(price_usd)                                     AS low,
  (array_agg(price_usd ORDER BY block_timestamp DESC, vendor_event_id DESC))[1] AS close,
  sum(abs(notional_usd))                             AS volume,  -- absolute traded USD
  count(*)                                           AS trades
FROM public.swaps
WHERE price_usd IS NOT NULL
GROUP BY pool_address, date_trunc('hour', to_timestamp(block_timestamp));

-- OPTIONAL accelerator: a REFRESHable matview when read-time aggregation gets
-- expensive. Refresh on a schedule (cron / pg_cron) after the backfill lands.
-- CREATE MATERIALIZED VIEW public.candles_mat AS
--   SELECT * FROM public.candles WITH NO DATA;
-- CREATE UNIQUE INDEX candles_mat_pk ON public.candles_mat (pool_address, bucket_start);
-- SELECT public.refresh_candles();   -- REFRESH MATERIALIZED VIEW CONCURRENTLY