Corporate Actions & Mathématiques d'Ajustement

-- Table de référence des corporate actions
-- Ingérée par le tap Go de la Partie 2 depuis Yahoo Finance / Polygon.io
CREATE TABLE IF NOT EXISTS raw_zone.corporate_actions (
    id              BIGSERIAL PRIMARY KEY,
    symbol          VARCHAR(20) NOT NULL,
    isin            CHAR(12),
    ex_date         DATE        NOT NULL,    -- Date ex-action (premier jour SANS droit)
    record_date     DATE,                    -- Date d'enregistrement
    ca_type         VARCHAR(20) NOT NULL,    -- 'SPLIT', 'DIVIDEND', 'SPINOFF', 'MERGER'
    -- Pour les splits : ratio = nb nouvelles actions / nb anciennes actions
    -- Split 4:1 → ratio = 4.0 (le dénominateur d'ajustement du prix)
    split_ratio     NUMERIC(10,6),
    -- Pour les dividendes : montant en devise de l'action
    dividend_amount NUMERIC(14,6),
    dividend_currency CHAR(3),
    -- Commentaire éditorial
    description     TEXT,
    provider_name   VARCHAR(50)  NOT NULL,
    ingested_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Index pour les jointures de fenêtre temporelle
CREATE INDEX idx_ca_symbol_date ON raw_zone.corporate_actions (symbol, ex_date DESC);
CREATE INDEX idx_ca_type        ON raw_zone.corporate_actions (ca_type, ex_date DESC);

{{ config(
    materialized='table',
    tags=['silver', 'corporate_actions']
) }}

WITH splits AS (
    SELECT
        symbol,
        ex_date,
        split_ratio,
        -- Validation : ratio doit être positif et non-nul
        CASE
            WHEN split_ratio IS NULL OR split_ratio <= 0 THEN NULL
            ELSE split_ratio
        END AS validated_ratio
    FROM {{ source('raw_zone', 'corporate_actions') }}
    WHERE ca_type = 'SPLIT'
      AND split_ratio IS NOT NULL
      AND split_ratio > 0
),

-- Calcul du facteur cumulatif pour chaque événement
-- La fenêtre DESC + UNBOUNDED FOLLOWING signifie :
--   "pour ce split ET tous les splits FUTURS" = ajustement rétroactif
cumulative_factors AS (
    SELECT
        symbol,
        ex_date,
        split_ratio,

        -- CRAF = produit de TOUS les ratios >= ex_date (splits futurs inclus)
        -- EXP(SUM(LN())) = produit cumulatif, numériquement stable
        EXP(SUM(LN(validated_ratio)) OVER (
            PARTITION BY symbol
            ORDER BY ex_date DESC
            ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING
        )) AS craf_from_this_event,

        -- Facteur d'ajustement AVANT cette date (pour ajuster les prix historiques)
        -- Pour les dates < ex_date, diviser par ce facteur
        EXP(SUM(LN(validated_ratio)) OVER (
            PARTITION BY symbol
            ORDER BY ex_date ASC
            ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING
        )) AS craf_for_historical_prices

    FROM splits
    WHERE validated_ratio IS NOT NULL
)

SELECT
    symbol,
    ex_date,
    split_ratio,
    craf_from_this_event,
    craf_for_historical_prices,
    -- Prix ajusté = prix brut / craf_for_historical_prices
    -- C'est ce facteur qui s'applique à toutes les barres AVANT ex_date
    ROUND(1.0 / craf_for_historical_prices, 8) AS price_adjustment_multiplier
FROM cumulative_factors
ORDER BY symbol, ex_date ASC

-- Comment appliquer le facteur d'ajustement à une barre historique
-- Utilisé en couche Gold (voir section suivante)

WITH bars AS (
    SELECT symbol, trade_date, open_price, high_price, low_price,
           close_price, adj_close_price, volume
    FROM silver_zone.fct_market_data_daily
),

-- Pour chaque barre, trouver le facteur d'ajustement cumulatif
-- = produit de tous les splits qui se sont produits APRÈS trade_date
split_factors AS (
    SELECT
        b.symbol,
        b.trade_date,
        -- COALESCE(factor, 1.0) = si pas de split après cette date → facteur = 1
        COALESCE(
            EXP(SUM(LN(ca.split_ratio)) FILTER (WHERE ca.ex_date > b.trade_date)),
            1.0
        ) AS cumulative_split_factor
    FROM bars b
    LEFT JOIN raw_zone.corporate_actions ca
           ON ca.symbol = b.symbol
          AND ca.ca_type = 'SPLIT'
          AND ca.ex_date > b.trade_date
    GROUP BY b.symbol, b.trade_date
)

-- Prix ajusté = prix brut / facteur cumulatif des splits FUTURS
SELECT
    b.symbol,
    b.trade_date,
    b.close_price                                          AS close_unadjusted,
    ROUND(b.close_price / sf.cumulative_split_factor, 4)  AS close_split_adjusted,

    -- Yahoo Finance "Adj Close" inclut déjà les dividendes ET les splits
    -- Pour un backtester total return, utiliser adj_close_price directement
    b.adj_close_price                                      AS close_total_return_adjusted,

    -- Volumes ajustés en sens inverse : un split 4:1 × volume × 4
    ROUND(b.volume * sf.cumulative_split_factor)           AS volume_adjusted,

    sf.cumulative_split_factor

FROM bars b
JOIN split_factors sf USING (symbol, trade_date)

{{ config(materialized='table', tags=['silver', 'corporate_actions']) }}

WITH dividends AS (
    SELECT
        ca.symbol,
        ca.ex_date,
        ca.dividend_amount,

        -- Prix de clôture la veille de l'ex-date (T-1)
        -- C'est le prix "cum-dividend" — avant que la bourse retire le dividende
        LAG(b.close_price, 0) OVER (
            PARTITION BY b.symbol
            ORDER BY b.trade_date
        ) AS close_price_day_before

    FROM {{ source('raw_zone', 'corporate_actions') }} ca
    JOIN {{ ref('fct_market_data_daily') }} b
      ON b.symbol = ca.symbol
      -- Clôture de la veille de l'ex-date
     AND b.trade_date = ca.ex_date - INTERVAL '1 day'
    WHERE ca.ca_type = 'DIVIDEND'
      AND ca.dividend_amount > 0
),

-- Dividend Adjustment Factor = (P_cum - D) / P_cum
daf_computed AS (
    SELECT
        symbol,
        ex_date,
        dividend_amount,
        close_price_day_before,
        CASE
            WHEN close_price_day_before > 0 THEN
                (close_price_day_before - dividend_amount) / close_price_day_before
            ELSE NULL
        END AS daf_single_event     -- Facteur pour CET événement uniquement
    FROM dividends
    WHERE close_price_day_before IS NOT NULL
      AND close_price_day_before > dividend_amount  -- sanity: dividende < prix
),

-- CRAF des dividendes : produit cumulatif de tous les DAF FUTURS
-- (même logique que pour les splits)
cumulative_daf AS (
    SELECT
        symbol,
        ex_date,
        dividend_amount,
        daf_single_event,

        EXP(SUM(LN(NULLIF(daf_single_event, 0))) OVER (
            PARTITION BY symbol
            ORDER BY ex_date ASC
            ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING
        )) AS cumulative_daf

    FROM daf_computed
    WHERE daf_single_event IS NOT NULL
      AND daf_single_event > 0
      AND daf_single_event < 1       -- Doit être entre 0 et 1
)

SELECT
    symbol,
    ex_date,
    dividend_amount,
    ROUND(daf_single_event::NUMERIC, 8)  AS daf_single,
    ROUND(cumulative_daf::NUMERIC, 8)    AS cumulative_daf,
    -- Prix ajusté total return = prix brut × cumulative_daf
    -- (multiplier < 1 → ajuste les prix historiques à la baisse)
    cumulative_daf                       AS price_multiplier_for_history
FROM cumulative_daf
ORDER BY symbol, ex_date ASC

-- Enregistrement du spin-off IBM → Kyndryl
INSERT INTO raw_zone.corporate_actions (
    symbol, isin, ex_date, record_date, ca_type,
    split_ratio, description, provider_name
) VALUES
-- L'événement affecte IBM (action mère)
(
    'IBM',
    'US4592001014',
    '2021-11-03',   -- ex-date (premier jour ex-spin-off)
    '2021-11-01',
    'SPINOFF',
    0.8955,         -- facteur d'ajustement : (120.05 / 134.04) = 0.8956
    'IBM spins off infrastructure services as Kyndryl Holdings (KD). '
    'Ratio: 1 KD share per 5 IBM shares. '
    'IBM price adjustment factor: ~0.896',
    'MANUAL_REVIEW'
),
-- Naissance du nouvel instrument Kyndryl
-- (géré via le snapshot SCD2 de dim_instruments en Partie 3)
(
    'KD',
    'US50155Q1004',
    '2021-11-03',
    '2021-11-01',
    'SPINOFF_NEW',  -- Type custom : nouvel instrument issu d'un spin-off
    NULL,
    'Kyndryl Holdings first trading day. Spun off from IBM. '
    'Opening price ~$28.10. is_tradable changes from FALSE to TRUE.',
    'MANUAL_REVIEW'
);

-- Fusion Microsoft / Activision Blizzard — clôture ATVI (octobre 2023)
INSERT INTO raw_zone.corporate_actions (
    symbol, isin, ex_date, ca_type, dividend_amount, description, provider_name
) VALUES (
    'ATVI',
    'US00507V1098',
    '2023-10-13',   -- Dernier jour de trading ATVI
    'MERGER',
    95.00,          -- Rachat tout-cash à $95/action par Microsoft
    'Microsoft acquires Activision Blizzard in all-cash deal. '
    'ATVI shareholders receive $95.00/share in cash. '
    'ATVI delisted after market close. Last price: $94.98.',
    'MANUAL_REVIEW'
);

-- Le tap Meltano détectera que ATVI disparaît de l'API Yahoo
-- Le snapshot dbt (Partie 3) avec invalidate_hard_deletes=true
-- fermera automatiquement la ligne :
--   dbt_valid_to = 2023-10-13
--   is_tradable  = FALSE (propagé depuis la source)

{{ config(
    materialized='table',
    tags=['gold', 'corporate_actions', 'ai_marts'],
    post_hook="ANALYZE {{ this }}"
) }}

WITH bars AS (
    SELECT
        symbol, trade_date,
        open_price, high_price, low_price, close_price,
        adj_close_price,   -- Yahoo Total Return (splits + dividendes)
        volume
    FROM {{ ref('fct_market_data_daily') }}
),

-- Facteurs de splits par (symbol, trade_date)
-- Pour chaque barre, on veut le produit des splits APRÈS cette date
split_factors_by_bar AS (
    SELECT
        b.symbol,
        b.trade_date,
        COALESCE(
            EXP(
                SUM(LN(ca.split_ratio))
                FILTER (WHERE ca.ex_date > b.trade_date AND ca.split_ratio > 0)
            ),
            1.0
        ) AS cumulative_split_factor

    FROM bars b
    LEFT JOIN {{ source('raw_zone', 'corporate_actions') }} ca
           ON ca.symbol = b.symbol
          AND ca.ca_type = 'SPLIT'
    GROUP BY b.symbol, b.trade_date
),

-- Facteurs de dividendes par (symbol, trade_date)
div_factors_by_bar AS (
    SELECT
        b.symbol,
        b.trade_date,
        COALESCE(
            EXP(
                SUM(LN(NULLIF(daf.cumulative_daf, 0)))
                FILTER (WHERE daf.ex_date > b.trade_date)
            ),
            1.0
        ) AS cumulative_div_factor

    FROM bars b
    LEFT JOIN {{ ref('fct_dividend_adjustment_factors') }} daf
           ON daf.symbol = b.symbol
    GROUP BY b.symbol, b.trade_date
)

SELECT
    b.symbol,
    b.trade_date,

    -- ═══ Prix BRUTS ═══
    b.open_price                                                AS open_raw,
    b.high_price                                                AS high_raw,
    b.low_price                                                 AS low_raw,
    b.close_price                                               AS close_raw,
    b.volume                                                    AS volume_raw,

    -- ═══ Ajustement splits uniquement ═══
    ROUND(b.open_price  / sf.cumulative_split_factor, 4)       AS open_split_adj,
    ROUND(b.high_price  / sf.cumulative_split_factor, 4)       AS high_split_adj,
    ROUND(b.low_price   / sf.cumulative_split_factor, 4)       AS low_split_adj,
    ROUND(b.close_price / sf.cumulative_split_factor, 4)       AS close_split_adj,
    ROUND(b.volume      * sf.cumulative_split_factor)          AS volume_split_adj,

    -- ═══ Ajustement Total Return (splits + dividendes) ═══
    -- Option A : Yahoo Adj Close (recommandé si source Yahoo)
    b.adj_close_price                                          AS close_total_return_yahoo,

    -- Option B : Calcul manuel (pour providers sans Adj Close)
    ROUND(b.close_price / sf.cumulative_split_factor
                        * df.cumulative_div_factor, 4)         AS close_total_return_manual,

    -- ═══ Métadonnées d'ajustement ═══
    sf.cumulative_split_factor,
    df.cumulative_div_factor,
    ROUND((sf.cumulative_split_factor * (1 / NULLIF(df.cumulative_div_factor, 0)))::NUMERIC, 6)
                                                               AS combined_adjustment_factor

FROM bars b
JOIN split_factors_by_bar sf USING (symbol, trade_date)
JOIN div_factors_by_bar   df USING (symbol, trade_date)

-- Validation : vérifier que les prix ajustés sont continus autour du split 4:1 AAPL
-- Le prix ajusté ne devrait pas avoir de "saut" le 31 août 2020

SELECT
    trade_date,
    close_raw,
    close_split_adj,
    -- Rendement journalier sur prix bruts (INCORRECT : montre le crash fantôme)
    ROUND(
        (close_raw - LAG(close_raw) OVER (ORDER BY trade_date)) /
        NULLIF(LAG(close_raw) OVER (ORDER BY trade_date), 0) * 100,
        2
    ) AS daily_return_raw_pct,
    -- Rendement journalier sur prix ajustés (CORRECT : continuité maintenue)
    ROUND(
        (close_split_adj - LAG(close_split_adj) OVER (ORDER BY trade_date)) /
        NULLIF(LAG(close_split_adj) OVER (ORDER BY trade_date), 0) * 100,
        2
    ) AS daily_return_adj_pct,
    cumulative_split_factor

FROM gold_zone.mart_adjusted_prices
WHERE symbol = 'AAPL'
  AND trade_date BETWEEN '2020-08-25' AND '2020-09-04'
ORDER BY trade_date;

-- Résultat attendu :
-- trade_date | close_raw | close_split_adj | daily_return_raw_pct | daily_return_adj_pct
-- 2020-08-28 |   499.23  |    124.81       | +1.2%                | +1.2%   ← normal
-- 2020-08-31 |   129.04  |    129.04       | -74.2% ← CRASH FANTÔME | +3.4% ← CORRECT
-- 2020-09-01 |   134.18  |    134.18       | +3.9%                | +3.9%   ← normal

-- Si daily_return_adj_pct montre aussi -74% : le facteur n'est pas appliqué correctement.

plugins:
  extractors:
    - name: tap-polygon-corporate-actions
      namespace: tap_polygon_corporate_actions
      pip_url: tap-polygon
      config:
        api_key: $POLYGON_API_KEY
        start_date: "1990-01-01"
        tickers_list: "{{ env_var('SECURITIES_UNIVERSE') }}"   # Liste des tickers à surveiller

      streams:
        - name: stock_splits
          endpoint: v3/reference/splits
          # Polygon retourne : ticker, ex_date, split_from, split_to
          # ratio = split_to / split_from (ex: 4:1 → ratio=4)

        - name: dividends
          endpoint: v3/reference/dividends
          # Polygon retourne : ticker, ex_dividend_date, cash_amount, declaration_date

      select:
        - stock_splits.*
        - dividends.*