How deeply and for how long did USDT, USDC and DAI depeg?

-- Candidate only: stablecoin_depeg_event_history.v1
-- Fixed scope: Ethereum USDT, USDC and DAI using Coinpaprika-backed hourly
-- observations from prices_external.hour. This query does not score reserves,
-- solvency, redemption capacity or future depeg probability.

WITH asset_registry(asset, blockchain, contract_address) AS (
  VALUES
    ('USDT', 'ethereum', 0xdac17f958d2ee523a2206206994597c13d831ec7),
    ('USDC', 'ethereum', 0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48),
    ('DAI', 'ethereum', 0x6b175474e89094c44da98b954eedeac495271d0f)
),
event_policy AS (
  SELECT
    CAST(1.0 AS DOUBLE) AS peg_usd,
    CAST(1.0 AS DOUBLE) AS threshold_pct,
    CAST(2 AS BIGINT) AS minimum_consecutive_hours,
    CAST(3.0 AS DOUBLE) AS severe_single_hour_threshold_pct
),
price_points AS (
  SELECT
    r.asset,
    r.blockchain,
    r.contract_address,
    p.timestamp,
    p.price AS price_usd
  FROM prices_external.hour p
  JOIN asset_registry r
    ON p.blockchain = r.blockchain
   AND p.contract_address = r.contract_address
  WHERE p.blockchain = 'ethereum'
    AND p.contract_address IN (
      0xdac17f958d2ee523a2206206994597c13d831ec7,
      0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48,
      0x6b175474e89094c44da98b954eedeac495271d0f
    )
    AND p.timestamp >= TIMESTAMP '2020-01-01 00:00:00'
    AND p.timestamp < CAST(date_trunc('hour', current_timestamp) AS timestamp)
    AND p.price IS NOT NULL
    AND p.price > 0
),
coverage AS (
  SELECT
    asset,
    blockchain,
    contract_address,
    MIN(timestamp) AS observed_start_utc,
    MAX(timestamp) AS observed_through_utc
  FROM price_points
  GROUP BY 1, 2, 3
),
classified AS (
  SELECT
    p.*,
    CASE
      WHEN p.price_usd <= e.peg_usd * (1 - e.threshold_pct / 100)
        THEN 'discount'
      WHEN p.price_usd >= e.peg_usd * (1 + e.threshold_pct / 100)
        THEN 'premium'
      ELSE NULL
    END AS event_side
  FROM price_points p
  CROSS JOIN event_policy e
),
sequenced AS (
  SELECT
    c.*,
    LAG(c.timestamp) OVER (
      PARTITION BY c.asset
      ORDER BY c.timestamp
    ) AS previous_timestamp,
    LAG(c.event_side) OVER (
      PARTITION BY c.asset
      ORDER BY c.timestamp
    ) AS previous_event_side,
    LEAD(c.timestamp) OVER (
      PARTITION BY c.asset
      ORDER BY c.timestamp
    ) AS next_timestamp,
    LEAD(c.event_side) OVER (
      PARTITION BY c.asset
      ORDER BY c.timestamp
    ) AS next_event_side
  FROM classified c
),
flagged AS (
  SELECT
    s.*,
    CASE
      WHEN s.event_side IS NOT NULL
       AND (
         s.previous_event_side IS DISTINCT FROM s.event_side
         OR s.previous_timestamp IS NULL
         OR date_diff('hour', s.previous_timestamp, s.timestamp) <> 1
       )
        THEN 1
      ELSE 0
    END AS event_start_flag
  FROM sequenced s
),
grouped AS (
  SELECT
    f.*,
    SUM(f.event_start_flag) OVER (
      PARTITION BY f.asset
      ORDER BY f.timestamp
      ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS event_group
  FROM flagged f
  WHERE f.event_side IS NOT NULL
),
event_aggregates AS (
  SELECT
    asset,
    blockchain,
    contract_address,
    event_side,
    event_group,
    MIN(timestamp) AS event_start_utc,
    MAX(timestamp) AS event_end_utc,
    COUNT(*) AS observation_count,
    MIN(price_usd) AS min_price_usd,
    MAX(price_usd) AS max_price_usd
  FROM grouped
  GROUP BY 1, 2, 3, 4, 5
),
qualified_events AS (
  SELECT
    a.*,
    date_diff('hour', a.event_start_utc, a.event_end_utc) + 1
      AS duration_hours,
    CASE
      WHEN a.event_side = 'discount'
        THEN (e.peg_usd - a.min_price_usd) / e.peg_usd * 100
      ELSE (a.max_price_usd - e.peg_usd) / e.peg_usd * 100
    END AS peak_abs_deviation_pct
  FROM event_aggregates a
  CROSS JOIN event_policy e
  WHERE a.observation_count >= e.minimum_consecutive_hours
     OR CASE
          WHEN a.event_side = 'discount'
            THEN (e.peg_usd - a.min_price_usd) / e.peg_usd * 100
          ELSE (a.max_price_usd - e.peg_usd) / e.peg_usd * 100
        END >= e.severe_single_hour_threshold_pct
),
coverage_records AS (
  SELECT
    'coverage' AS record_type,
    c.asset,
    c.blockchain,
    CONCAT('0x', LOWER(TO_HEX(c.contract_address))) AS contract_address,
    c.observed_start_utc,
    c.observed_through_utc,
    CAST(NULL AS VARCHAR) AS event_side,
    CAST(NULL AS TIMESTAMP) AS event_start_utc,
    CAST(NULL AS TIMESTAMP) AS event_end_utc,
    CAST(NULL AS TIMESTAMP) AS recovery_timestamp_utc,
    CAST(NULL AS BIGINT) AS duration_hours,
    CAST(NULL AS BIGINT) AS observation_count,
    CAST(NULL AS DOUBLE) AS min_price_usd,
    CAST(NULL AS DOUBLE) AS max_price_usd,
    CAST(NULL AS DOUBLE) AS peak_abs_deviation_pct,
    CAST(NULL AS BOOLEAN) AS recovered_to_band,
    e.threshold_pct,
    e.minimum_consecutive_hours,
    e.severe_single_hour_threshold_pct,
    'coinpaprika' AS price_source,
    'dune:prices_external.hour' AS source,
    'stablecoin_depeg_event_history.v1' AS source_contract_version
  FROM coverage c
  CROSS JOIN event_policy e
),
event_records AS (
  SELECT
    'event' AS record_type,
    q.asset,
    q.blockchain,
    CONCAT('0x', LOWER(TO_HEX(q.contract_address))) AS contract_address,
    c.observed_start_utc,
    c.observed_through_utc,
    q.event_side,
    q.event_start_utc,
    q.event_end_utc,
    CASE
      WHEN endpoint.next_timestamp = date_add('hour', 1, q.event_end_utc)
       AND endpoint.next_event_side IS NULL
        THEN endpoint.next_timestamp
      ELSE NULL
    END AS recovery_timestamp_utc,
    q.duration_hours,
    q.observation_count,
    q.min_price_usd,
    q.max_price_usd,
    q.peak_abs_deviation_pct,
    CASE
      WHEN endpoint.next_timestamp = date_add('hour', 1, q.event_end_utc)
       AND endpoint.next_event_side IS NULL
        THEN TRUE
      ELSE FALSE
    END AS recovered_to_band,
    e.threshold_pct,
    e.minimum_consecutive_hours,
    e.severe_single_hour_threshold_pct,
    'coinpaprika' AS price_source,
    'dune:prices_external.hour' AS source,
    'stablecoin_depeg_event_history.v1' AS source_contract_version
  FROM qualified_events q
  JOIN coverage c
    ON q.asset = c.asset
  JOIN sequenced endpoint
    ON q.asset = endpoint.asset
   AND q.event_end_utc = endpoint.timestamp
  CROSS JOIN event_policy e
)
SELECT
  record_type,
  asset,
  blockchain,
  contract_address,
  observed_start_utc,
  observed_through_utc,
  event_side,
  event_start_utc,
  event_end_utc,
  recovery_timestamp_utc,
  duration_hours,
  observation_count,
  min_price_usd,
  max_price_usd,
  peak_abs_deviation_pct,
  recovered_to_band,
  threshold_pct,
  minimum_consecutive_hours,
  severe_single_hour_threshold_pct,
  price_source,
  source,
  source_contract_version
FROM coverage_records
UNION ALL
SELECT
  record_type,
  asset,
  blockchain,
  contract_address,
  observed_start_utc,
  observed_through_utc,
  event_side,
  event_start_utc,
  event_end_utc,
  recovery_timestamp_utc,
  duration_hours,
  observation_count,
  min_price_usd,
  max_price_usd,
  peak_abs_deviation_pct,
  recovered_to_band,
  threshold_pct,
  minimum_consecutive_hours,
  severe_single_hour_threshold_pct,
  price_source,
  source,
  source_contract_version
FROM event_records
ORDER BY asset, record_type, event_start_utc;