Le JSON brut est toxique pour les modèles quantitatifs. Construisez un pipeline rigoureux avec dbt : vues Bronze, modèles incrémentiels Silver, snapshots SCD Type 2 automatisés, et marts Gold prêts pour l'IA.
Dans la Partie 1, nous avons conçu le schéma PostgreSQL institutionnel avec SCD Type 2, tables de référence et partitionnement. Dans la Partie 2, des extracteurs Go concurrents orchestrés par Nomad et Meltano pompent les APIs vers notre raw_zone. Maintenant, nous faisons face au chaos.
La raw_zone déborde de JSON brut. L'API Casablanca Bourse produit des propriétés imbriquées non documentées. Yahoo Finance renvoie des CSV plats avec des conventions de colonnes changeantes. Binance utilise BTCUSDT là où d'autres attendent BTC-USD. Si un chercheur quant interroge ces données directement, votre fonds perd de l'argent à cause de mauvaises données.
La solution : l'Architecture Médaillon implémentée avec dbt Core.
Le pattern Médaillon (Bronze, Silver, Gold) n'est pas un buzzword — c'est un mécanisme de défense ingénierie. Il isole le chaos des fournisseurs externes de la logique fragile de vos algorithmes de trading. Chaque couche a un contrat strict.
Matérialisation : view
Schema-on-Read. Extrait les clés JSON, caste en types PostgreSQL stricts (NUMERIC(14,4)), normalise les noms de colonnes entre providers. Zéro coût de stockage.
Couche de déduplication
Matérialisation : incremental + snapshot
Source de vérité. Déduplication, résolution des conflits entre providers, et tracking temporel SCD Type 2. Milliards de barres OHLCV fusionnées de façon incrémentielle.
Point-in-time correct
Matérialisation : table
Schémas en étoile dé-normalisés. Indicateurs techniques pré-calculés (ATR, Volatilité Rolling, Volume Dollar). Un agent LLM ou un outil BI interroge le Gold layer sans écrire de fenêtres complexes.
Interface IA prête
Avant d'écrire le moindre modèle SQL, la structure de répertoires est critique. dbt utilise une convention de nommage pour déduire la couche Médaillon et appliquer les configurations par défaut définies dans dbt_project.yml.
name: 'securities_dbt'
version: '1.7.0'
config-version: 2
profile: 'securities_master'
model-paths: ["models"]
snapshot-paths: ["snapshots"]
test-paths: ["tests"]
macro-paths: ["macros"]
models:
securities_dbt:
# Couche Bronze : vue légère, pas de stockage
bronze:
+materialized: view
+schema: bronze_zone
+tags: ['bronze']
# Couche Silver : tables incrémentielles persistantes
silver:
+materialized: incremental
+schema: silver_zone
+tags: ['silver']
# Couche Gold : tables rebuild complètes chaque nuit
gold:
+materialized: table
+schema: gold_zone
+tags: ['gold']
snapshots:
securities_dbt:
+target_schema: silver_zone
+tags: ['scd2', 'silver']La configuration centralisée dans dbt_project.yml évite de répéter materialized dans chaque fichier SQL. Chaque modèle dans models/bronze/ hérite automatiquement de view. Vous pouvez toujours écraser localement avec {{ config(materialized='table') }}.
version: 2
sources:
- name: raw_zone
description: "Données brutes ingérées par les taps Meltano/Go (Part 2)"
database: securities_master
schema: raw_zone
tables:
- name: raw_instruments
description: "Payloads JSON des instruments financiers par provider"
columns:
- name: external_id
description: "Identifiant du provider (ticker, ISIN, etc.)"
tests:
- not_null
- name: raw_payload
description: "JSONB complet du provider"
tests:
- not_null
- name: provider_name
description: "Nom du provider (BVC, YAHOO, BINANCE)"
tests:
- not_null
- accepted_values:
values: ['BVC', 'YAHOO', 'BINANCE', 'EURONEXT', 'CBOE']
- name: _sdc_extracted_at
description: "Timestamp d'ingestion Meltano (Singer _sdc_extracted_at)"
tests:
- not_null
- name: raw_market_data
description: "Données OHLCV brutes par provider"
columns:
- name: symbol
tests: [not_null]
- name: trade_date
tests: [not_null]
- name: raw_payload
tests: [not_null]{{ source('raw_zone', 'raw_instruments') }} permet à dbt de tracer le lineage complet de la donnée : de la table raw ingérée par le tap Go jusqu'au mart Gold final. Le dbt docs generate produit un graphe de dépendances interactif. Les tests sur les sources (not_null, accepted_values) créent un firewall de qualité à l'entrée du pipeline.
On n'écrit jamais d'instructions INSERT avec du parsing JSON manuel. On définit des vues dbt sur les tables raw. Si un provider change Adj Close en adjusted_close demain, on met à jour uniquement la vue Bronze. Tout le pipeline aval reste protégé.
-- Héritage de dbt_project.yml : materialized='view', schema='bronze_zone'
{{ config(tags=['bronze', 'reference_data']) }}
WITH raw_bvc AS (
-- {{ source() }} documente le lineage dans dbt docs
SELECT *
FROM {{ source('raw_zone', 'raw_instruments') }}
WHERE provider_name = 'BVC'
),
extracted AS (
SELECT
-- Métadonnées Singer/Meltano (injectées par le tap Go en Part 2)
_sdc_extracted_at AS ingested_at,
-- Extraction JSONB PostgreSQL + casting strict
external_id::VARCHAR(50) AS provider_tick,
(raw_payload->>'codeISIN')::CHAR(12) AS isin,
(raw_payload->>'instrumentName')::TEXT AS name,
(raw_payload->>'currency')::CHAR(3) AS currency,
-- JSONB imbriqué : tableaux de secteurs
-- jsonb_array_element_text() extrait l'index 0 d'un tableau JSON
jsonb_array_element_text(
raw_payload->'sectors', 0
)::TEXT AS sector_primary,
-- Standardisation vers notre modèle de données (Part 1)
UPPER(raw_payload->>'instrumentType')::VARCHAR(50) AS instrument_type,
(raw_payload->>'isTradable')::BOOLEAN AS is_tradable,
-- Coordonnées boursières
(raw_payload->>'marketCap')::NUMERIC(20,2) AS market_cap_mad,
(raw_payload->>'floatRatio')::NUMERIC(6,4) AS float_ratio
FROM raw_bvc
),
-- ⚠ Déduplication PostgreSQL correcte : CTE + ROW_NUMBER
-- NOTE: QUALIFY n'existe PAS en PostgreSQL (BigQuery/Snowflake uniquement).
-- Pattern correct : CTE intermédiaire + WHERE rn = 1
deduped AS (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY isin
ORDER BY ingested_at DESC -- garder l'ingestion la plus récente
) AS rn
FROM extracted
WHERE isin IS NOT NULL -- filtrer les instruments sans ISIN
)
SELECT
ingested_at,
provider_tick,
isin,
name,
currency,
sector_primary,
instrument_type,
is_tradable,
market_cap_mad,
float_ratio
FROM deduped
WHERE rn = 1 -- strictement le payload le plus récent par ISINLa syntaxe QUALIFY ROW_NUMBER() OVER (...) = 1 est exclusive à BigQuery, Snowflake, et DuckDB. Elle lève une erreur de syntaxe sur PostgreSQL.
Le pattern correct PostgreSQL est un CTE avec ROW_NUMBER() calculé dans une sous-requête, puis filtré avec WHERE rn = 1 dans la requête externe. dbt rend ce pattern naturel grâce aux CTEs enchaînées.
-> retourne un objet JSONB (avec les guillemets JSON inclus). ->> retourne une chaîne de texte, ce qui permet le cast direct : (raw_payload->>'marketCap')::NUMERIC. Utiliser -> avec un cast numérique provoque une erreur de type. Pour les tableaux JSON imbriqués, utiliser jsonb_array_element_text(payload->'arrayKey', index).
{{ config(tags=['bronze', 'market_data']) }}
WITH raw_yahoo AS (
SELECT *
FROM {{ source('raw_zone', 'raw_market_data') }}
WHERE provider_name = 'YAHOO'
),
extracted AS (
SELECT
_sdc_extracted_at AS ingested_at,
(raw_payload->>'symbol')::VARCHAR(20) AS symbol,
-- Yahoo renvoie trade_date comme string ISO 8601 'YYYY-MM-DD'
(raw_payload->>'date')::DATE AS trade_date,
-- OHLCV : Yahoo utilise 'Open', 'High', 'Low', 'Close' (capitalised)
(raw_payload->>'Open')::NUMERIC(14,4) AS open_price,
(raw_payload->>'High')::NUMERIC(14,4) AS high_price,
(raw_payload->>'Low')::NUMERIC(14,4) AS low_price,
(raw_payload->>'Close')::NUMERIC(14,4) AS close_price,
(raw_payload->>'Adj Close')::NUMERIC(14,4) AS adj_close_price,
(raw_payload->>'Volume')::BIGINT AS volume,
-- VWAP n'est pas fourni par Yahoo, on l'estime (HLC/3)
NULL::NUMERIC(14,4) AS vwap,
-- Validation de base : filtrer les barres sans prix valide
CASE
WHEN (raw_payload->>'Close') IS NULL THEN FALSE
WHEN (raw_payload->>'Volume') = '0' THEN FALSE
ELSE TRUE
END AS is_valid
FROM raw_yahoo
),
deduped AS (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY symbol, trade_date
ORDER BY ingested_at DESC
) AS rn
FROM extracted
WHERE is_valid = TRUE
AND trade_date IS NOT NULL
AND symbol IS NOT NULL
)
SELECT
ingested_at, symbol, trade_date,
open_price, high_price, low_price, close_price, adj_close_price,
volume, vwap
FROM deduped
WHERE rn = 1Avec 10 000 actions mondiales et 25 ans d'historique journalier, nous traitons 60+ millions de lignes. Les données intraday (barres 1 minute) atteignent plusieurs dizaines de milliards de lignes. Matérialiser cela en table dbt — qui exécute un DROP TABLE puis CREATE TABLE AS SELECT à chaque run — serait catastrophique.
La matérialisation incremental de dbt résout ce problème. Au premier run, elle construit la table complète. À chaque run suivant, dbt génère un INSERT ... ON CONFLICT DO UPDATE, ne traitant que les nouvelles données.
-- Configuration du modèle incrémentiel
-- ⚠ PostgreSQL : PAS de partition_by dans config dbt
-- (c'est une option BigQuery-specific)
-- Le partitionnement PostgreSQL est défini au niveau DDL (fait en Part 1)
{{ config(
materialized='incremental',
unique_key=['symbol', 'trade_date'],
on_schema_change='fail',
tags=['silver', 'market_data']
) }}
WITH combined_bronze AS (
-- Union de toutes les sources Bronze normalisées
SELECT * FROM {{ ref('stg_yahoo_bars') }}
UNION ALL
SELECT * FROM {{ ref('stg_bvc_bars') }}
UNION ALL
SELECT * FROM {{ ref('stg_binance_bars') }}
),
enriched AS (
SELECT
symbol,
trade_date,
open_price,
high_price,
low_price,
close_price,
adj_close_price,
volume,
-- VWAP proxy si le provider ne le fournit pas (HLC/3)
COALESCE(
vwap,
(high_price + low_price + close_price) / 3.0
) AS vwap_proxy,
-- True Range pour calcul ATR ultérieur en Gold
high_price - low_price AS true_range_hl,
ingested_at,
CURRENT_TIMESTAMP AS dbt_updated_at
FROM combined_bronze
WHERE close_price > 0 -- sanity check
AND volume > 0
AND trade_date <= CURRENT_DATE -- pas de barres futures
)
SELECT * FROM enriched
{% if is_incremental() %}
-- ⚡ Block INCRÉMENTIEL : exécuté uniquement sur les runs de mise à jour
-- Lookback de 3 jours pour capturer :
-- - Les corrections tardives de volumes envoyées par les exchanges
-- - Les dividendes ajustés rétroactivement par Yahoo Finance
-- - Les splits qui recalculent l'historique adj_close
WHERE trade_date >= (
SELECT MAX(trade_date) - INTERVAL '3 days'
FROM {{ this }}
)
{% endif %}Les données financières arrivent avec un délai. Yahoo Finance corrige silencieusement ses prix ajustés des dividendes plusieurs jours après le paiement. Les exchanges envoient parfois un volume révisé le lendemain. Avec > MAX(trade_date), ces corrections ne seraient jamais ingérées.
La fenêtre de 3 jours garantit qu'on retraite les 3 derniers jours. Avec unique_key=['symbol', 'trade_date'], dbt génère un vrai upsert : si la ligne existe, elle est mise à jour ; sinon elle est insérée. On capture les corrections tout en traitant 400× moins de données qu'un rebuild complet.
La clé partition_by={'field': 'trade_date', 'data_type': 'date', 'granularity': 'month'} dans le bloc config() dbt est exclusive à BigQuery. Sur PostgreSQL, elle est silencieusement ignorée ou provoque des erreurs selon la version de l'adaptateur.
Le partitionnement PostgreSQL est défini au niveau DDL PARTITION BY RANGE (trade_date) lors de la création initiale de la table (fait dans la Partie 1). dbt insère ensuite dans la bonne partition automatiquement — sans configuration supplémentaire dans le modèle.
{{ config(
materialized='incremental',
unique_key=['symbol', 'bar_timestamp'],
-- Sur 50+ milliards de lignes, on vérifie d'abord si le schema a changé
on_schema_change='fail',
tags=['silver', 'intraday']
) }}
WITH raw_intraday AS (
SELECT *
FROM {{ source('raw_zone', 'raw_market_data') }}
WHERE provider_name IN ('BINANCE', 'BVC_INTRADAY')
AND (raw_payload->>'interval') = '1m'
),
extracted AS (
SELECT
(raw_payload->>'symbol')::VARCHAR(20) AS symbol,
-- Timestamp complet avec timezone
TO_TIMESTAMP(
(raw_payload->>'open_time')::BIGINT / 1000.0
) AT TIME ZONE 'UTC' AS bar_timestamp,
(raw_payload->>'open')::NUMERIC(14,8) AS open_price,
(raw_payload->>'high')::NUMERIC(14,8) AS high_price,
(raw_payload->>'low')::NUMERIC(14,8) AS low_price,
(raw_payload->>'close')::NUMERIC(14,8) AS close_price,
(raw_payload->>'volume')::NUMERIC(24,8) AS volume,
(raw_payload->>'quote_asset_volume')::NUMERIC(24,8) AS quote_volume,
(raw_payload->>'number_of_trades')::INTEGER AS num_trades,
_sdc_extracted_at AS ingested_at
FROM raw_intraday
),
deduped AS (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY symbol, bar_timestamp
ORDER BY ingested_at DESC
) AS rn
FROM extracted
WHERE close_price > 0
)
SELECT
symbol, bar_timestamp, open_price, high_price, low_price,
close_price, volume, quote_volume, num_trades, ingested_at
FROM deduped
WHERE rn = 1
{% if is_incremental() %}
-- Pour les barres intraday, lookback de 2 heures suffit
-- (les exchanges corrigent rarement au-delà d'une heure)
AND bar_timestamp >= (
SELECT MAX(bar_timestamp) - INTERVAL '2 hours'
FROM {{ this }}
)
{% endif %}Dans la Partie 1, nous avons conçu le framework théorique : quand les métadonnées fondamentales changent (Apple change de secteur GICS, une action est renommée suite à une fusion), on ne doit pas écraser la ligne. Il faut fermer le valid_to de l'ancienne version et insérer une nouvelle version avec valid_from = NOW(). Coder ces merges SQL manuellement est notoire pour produire des bugs subtils.
dbt Snapshots automatise entièrement cette logique. Vous déclarez quelles colonnes surveiller ; dbt calcule les hashes, détecte les mutations, et émet les UPDATE/INSERT appropriés.
{% snapshot dim_instruments_snapshot %}
{{
config(
target_schema='silver_zone',
unique_key='isin',
-- Stratégie 'check' : dbt compare un hash des colonnes listées
-- Si le hash change → nouvelle version SCD2
strategy='check',
check_cols=[
'name',
'sector_primary',
'instrument_type',
'is_tradable',
'currency',
'market_cap_mad', -- Un changement de taille matérielle est documenté
],
-- Colonnes dbt_scd_id, dbt_updated_at, dbt_valid_from, dbt_valid_to
-- sont automatiquement ajoutées par dbt
invalidate_hard_deletes=true, -- Marquer dbt_valid_to si l'ISIN disparaît
)
}}
-- Source : la vue Bronze qui agrège tous les providers d'instruments
SELECT
isin,
provider_tick,
name,
instrument_type,
sector_primary,
currency,
is_tradable,
market_cap_mad,
float_ratio,
ingested_at
FROM {{ ref('stg_all_instruments') }}
-- On n'inclut que les instruments avec ISIN valide
WHERE isin IS NOT NULL
AND LENGTH(TRIM(isin)) = 12
{% endsnapshot %}À chaque run de dbt snapshot, dbt exécute un MERGE en deux étapes : d'abord, il compare le hash des check_cols de chaque ligne entrante avec la version active (où dbt_valid_to IS NULL). Si le hash change, dbt émet un UPDATE pour fermer l'ancienne ligne (dbt_valid_to = NOW()) puis un INSERT pour la nouvelle version (dbt_valid_from = NOW(), dbt_valid_to = NULL). Si le hash est identique, rien ne se passe. C'est exactement l'architecture point-in-time que nous avons codée manuellement en Partie 1, ici générée automatiquement.
La table générée silver_zone.dim_instruments_snapshot produit automatiquement les colonnes suivantes :
| Colonne | Type | Description | Valeur si actif |
|---|---|---|---|
| dbt_scd_id | TEXT | Hash unique de la version | UUID généré |
| dbt_updated_at | TIMESTAMP | Dernière mise à jour de la ligne | Timestamp du run |
| dbt_valid_from | TIMESTAMP | Début de validité de cette version | Date de création |
| dbt_valid_to | TIMESTAMP | Fin de validité (NULL = version actuelle) | NULL |
| isin + tous vos champs | Vos types | Données d'instrument à ce moment | Valeurs actuelles |
-- Exemple : Quel secteur avait Apple (ISIN US0378331005) au 15 janvier 2023 ?
-- Requête point-in-time correcte grâce aux snapshots dbt
SELECT
isin,
name,
sector_primary,
instrument_type,
dbt_valid_from,
dbt_valid_to
FROM silver_zone.dim_instruments_snapshot
WHERE isin = 'US0378331005'
-- Point-in-time : la version valide à cette date
AND dbt_valid_from <= '2023-01-15'::TIMESTAMP
AND (dbt_valid_to > '2023-01-15'::TIMESTAMP OR dbt_valid_to IS NULL)
ORDER BY dbt_valid_from;
-- Pour obtenir uniquement la version actuelle (toutes les actions)
SELECT *
FROM silver_zone.dim_instruments_snapshot
WHERE dbt_valid_to IS NULL
AND is_tradable = TRUE
ORDER BY market_cap_mad DESC NULLS LAST;-- Vue de commodité : expose uniquement la version actuelle de chaque instrument
-- Les utilisateurs en aval n'ont pas à gérer dbt_valid_to IS NULL
{{ config(materialized='view', tags=['silver', 'reference_data']) }}
SELECT
isin,
provider_tick,
name,
instrument_type,
sector_primary,
currency,
is_tradable,
market_cap_mad,
float_ratio,
dbt_valid_from AS last_updated
FROM {{ ref('dim_instruments_snapshot') }}
WHERE dbt_valid_to IS NULLSi vous passez des NaN à un backtester Python, il plante. Si vous passez des données erronées à un algorithme de trading live, il perd de l'argent. Nous utilisons les assertions de test dbt comme une porte CI/CD stricte. Les données n'atteignent pas la couche Gold si des tests échouent.
version: 2
models:
- name: fct_market_data_daily
description: "Barres OHLCV journalières — source de vérité du Securities Master"
columns:
- name: symbol
description: "Symbole boursier normalisé"
tests:
- not_null
- relationships:
to: ref('dim_instruments_current')
field: provider_tick
- name: trade_date
description: "Date de la séance boursière"
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_between:
min_value: "'1990-01-01'::DATE"
max_value: "CURRENT_DATE"
- name: close_price
description: "Prix de clôture en devise du marché"
tests:
- not_null
# Prix ne peut être négatif ou zéro
- dbt_expectations.expect_column_values_to_be_between:
min_value: 0.0001
strictly: false
- name: volume
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_between:
min_value: 0
- name: high_price
tests:
# High doit toujours être >= Close
- dbt_utils.expression_is_true:
expression: ">= close_price"
- name: low_price
tests:
# Low doit toujours être <= Close
- dbt_utils.expression_is_true:
expression: "<= close_price"
tests:
# Unicité composite : une seule barre par symbole par jour
- unique:
column_name: "symbol || '-' || trade_date::TEXT"
- dbt_utils.unique_combination_of_columns:
combination_of_columns:
- symbol
- trade_date
- name: dim_instruments_current
description: "Version courante de tous les instruments actifs"
columns:
- name: isin
tests:
- not_null
- unique
- dbt_expectations.expect_column_value_lengths_to_equal:
value: 12
- name: instrument_type
tests:
# Rejet si un provider invente une nouvelle classe d'actif non documentée
- accepted_values:
values: ['EQUITY', 'FUTURE', 'FXFORWARD', 'IRS', 'OPTIONVOL', 'CRYPTO', 'ETF']
- name: currency
tests:
- not_null
- dbt_expectations.expect_column_value_lengths_to_equal:
value: 3Pour les tests métier complexes, dbt supporte des tests SQL personnalisés dans tests/. Ces tests doivent retourner 0 ligne pour passer.
-- Test personnalisé : vérifie la cohérence OHLCV interne
-- Un chandelier valide doit respecter : Low <= Open, Close <= High
-- Ce test retourne les lignes invalides — dbt échoue si résultat non vide
SELECT
symbol,
trade_date,
open_price,
high_price,
low_price,
close_price,
'OHLCV_INCONSISTENCY' AS failure_reason
FROM {{ ref('fct_market_data_daily') }}
WHERE
-- Open en dehors du range High/Low
open_price > high_price
OR open_price < low_price
-- Close en dehors du range High/Low
OR close_price > high_price
OR close_price < low_price
-- High inférieur à Low (impossible physiquement)
OR high_price < low_price-- Les marchés actions sont fermés le weekend
-- Ce test détecte les barres Saturday/Sunday pour les actions (pas crypto)
-- Retourne les violations → dbt échoue si non vide
SELECT
f.symbol,
f.trade_date,
EXTRACT(DOW FROM f.trade_date) AS day_of_week, -- 0=Sun, 6=Sat
'WEEKEND_BAR' AS failure_reason
FROM {{ ref('fct_market_data_daily') }} f
JOIN {{ ref('dim_instruments_current') }} d
ON f.symbol = d.provider_tick
WHERE d.instrument_type = 'EQUITY' -- Uniquement les actions
AND EXTRACT(DOW FROM f.trade_date) IN (0, 6) -- Weekend| Test | Type | Modèle | Sévérité | Action si échec |
|---|---|---|---|---|
| not_null sur symbol/date | Built-in | Bronze + Silver | error | Pipeline bloqué |
| Prix positif (> 0.0001) | dbt-expectations | Silver | error | Pipeline bloqué |
| Cohérence OHLCV | Custom SQL | Silver | warn | Alerte, pipeline continue |
| Unicité symbol+date | Built-in | Silver | error | Pipeline bloqué |
| instrument_type accepté | Built-in | Silver | error | Pipeline bloqué |
| Barres weekend (actions) | Custom SQL | Silver | warn | Log + investigation |
| Longueur ISIN = 12 | dbt-expectations | Silver | error | Pipeline bloqué |
| Relation instruments actifs | Built-in | Silver | warn | Alerte, pipeline continue |
La couche Silver est parfaitement structurée mais normalisée. Demander à un Data Scientist ou un LLM d'écrire dynamiquement des fonctions de fenêtre pour calculer des ATR sur 20 jours en traversant des dimensions SCD2 est masochiste et source d'erreurs.
La couche Gold abandonne la normalisation. Nous utilisons dbt pour pré-calculer des tables larges en schéma étoile, optimisées pour des SELECT rapides. Quand votre scanner systématique cherche chaque matin des "Candidats Breakout Haute Volatilité Très Liquides", il frappe la couche Gold.
{{ config(
materialized='table',
tags=['gold', 'ai_marts'],
post_hook="ANALYZE {{ this }}" -- Mettre à jour les stats PostgreSQL après rebuild
) }}
WITH active_instruments AS (
-- Cacher la complexité SCD2 aux utilisateurs downstream
-- On expose uniquement CE QUI est tradable AUJOURD'HUI
SELECT *
FROM {{ ref('dim_instruments_current') }}
WHERE is_tradable = TRUE
),
-- Derniers 252 jours de données (1 an de trading)
recent_bars AS (
SELECT *
FROM {{ ref('fct_market_data_daily') }}
WHERE trade_date >= CURRENT_DATE - INTERVAL '252 days'
),
technicals AS (
SELECT
symbol,
trade_date,
close_price,
volume,
adj_close_price,
-- ══ Indicateurs de Liquidité ══
-- Average Daily Volume en devise (20 jours)
AVG(volume * close_price) OVER w_20d AS adv_20d_usd,
-- ══ Indicateurs de Volatilité ══
-- Volatilité réalisée 20 jours (écart-type des rendements log)
STDDEV(
LN(close_price / LAG(close_price, 1) OVER w_all)
) OVER w_20d * SQRT(252) AS realized_vol_20d_annualized,
-- True Range pour ATR
GREATEST(
high_price - low_price,
ABS(high_price - LAG(close_price, 1) OVER w_all),
ABS(low_price - LAG(close_price, 1) OVER w_all)
) AS true_range,
-- ══ Indicateurs de Tendance ══
AVG(close_price) OVER w_20d AS sma_20d,
AVG(close_price) OVER w_50d AS sma_50d,
AVG(close_price) OVER w_200d AS sma_200d,
-- Distance par rapport aux moyennes (momentum)
(close_price / NULLIF(AVG(close_price) OVER w_50d, 0) - 1) * 100
AS pct_above_sma50,
-- ══ Performance Periods ══
-- Rendement 1 mois (21 jours de trading)
(close_price / NULLIF(
LAG(close_price, 21) OVER w_all, 0
) - 1) * 100 AS perf_1m_pct,
-- Rendement 3 mois (63 jours de trading)
(close_price / NULLIF(
LAG(close_price, 63) OVER w_all, 0
) - 1) * 100 AS perf_3m_pct
FROM {{ ref('fct_market_data_daily') }}
JOIN active_instruments USING (symbol)
WINDOW
-- Fenêtres nommées réutilisées : lisibles et optimisées
w_all AS (PARTITION BY symbol ORDER BY trade_date
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW),
w_20d AS (PARTITION BY symbol ORDER BY trade_date
ROWS BETWEEN 20 PRECEDING AND CURRENT ROW),
w_50d AS (PARTITION BY symbol ORDER BY trade_date
ROWS BETWEEN 50 PRECEDING AND CURRENT ROW),
w_200d AS (PARTITION BY symbol ORDER BY trade_date
ROWS BETWEEN 200 PRECEDING AND CURRENT ROW)
),
-- ATR 14 jours : moyenne des true ranges
atr_calc AS (
SELECT
symbol,
trade_date,
AVG(true_range) OVER (
PARTITION BY symbol ORDER BY trade_date
ROWS BETWEEN 14 PRECEDING AND CURRENT ROW
) AS atr_14d
FROM technicals
),
-- Snapshot du dernier jour uniquement pour la table Gold
latest_day AS (
SELECT DISTINCT ON (symbol)
t.*,
a.atr_14d
FROM technicals t
JOIN atr_calc a USING (symbol, trade_date)
ORDER BY symbol, t.trade_date DESC
)
-- ══ Table Gold finale : large, dé-normalisée, IA-ready ══
SELECT
-- Dimensions de référence (depuis Silver SCD2)
ref.isin,
ref.provider_tick AS symbol,
ref.name AS company_name,
ref.sector_primary AS sector,
ref.instrument_type,
ref.currency,
ref.market_cap_mad,
-- Prix et date
ld.trade_date AS last_price_date,
ld.close_price,
ld.adj_close_price,
-- Liquidité
ROUND(ld.adv_20d_usd::NUMERIC, 0) AS adv_20d_usd,
-- Volatilité et ATR
ROUND(ld.realized_vol_20d_annualized::NUMERIC, 4) AS realized_vol_20d,
ROUND(ld.atr_14d::NUMERIC, 4) AS atr_14d,
-- Tendances
ROUND(ld.sma_20d::NUMERIC, 4) AS sma_20d,
ROUND(ld.sma_50d::NUMERIC, 4) AS sma_50d,
ROUND(ld.sma_200d::NUMERIC, 4) AS sma_200d,
ROUND(ld.pct_above_sma50::NUMERIC, 2) AS pct_above_sma50,
-- Performance
ROUND(ld.perf_1m_pct::NUMERIC, 2) AS perf_1m_pct,
ROUND(ld.perf_3m_pct::NUMERIC, 2) AS perf_3m_pct,
-- ══ Tags de régime IA : pré-classifiés ══
CASE
WHEN ld.close_price > ld.sma_200d
AND ld.sma_50d > ld.sma_200d THEN 'Bullish — Golden Cross'
WHEN ld.close_price > ld.sma_50d THEN 'Bullish — Above SMA50'
WHEN ld.close_price < ld.sma_200d
AND ld.sma_50d < ld.sma_200d THEN 'Bearish — Death Cross'
ELSE 'Bearish — Below SMA50'
END AS trend_regime,
-- Filtre liquidité institutionnel
CASE
WHEN ld.adv_20d_usd >= 50000000 THEN 'Institutionnel'
WHEN ld.adv_20d_usd >= 2000000 THEN 'Retail Eligible'
WHEN ld.adv_20d_usd >= 500000 THEN 'Micro Cap'
ELSE 'Illiquid'
END AS liquidity_class,
CURRENT_TIMESTAMP AS mart_refreshed_at
FROM active_instruments ref
JOIN latest_day ld ON ref.provider_tick = ld.symbol
-- ══ Filtre de qualité Gold : exclure les instruments illiquides ══
WHERE ld.adv_20d_usd >= 500000 -- min 500K$/jour de volume
AND ld.atr_14d IS NOT NULL -- assez d'historique pour ATR 14j
AND ld.sma_50d IS NOT NULL -- assez d'historique pour SMA 50jSELECT DISTINCT ON (symbol) est une extension PostgreSQL qui sélectionne la première ligne de chaque groupe selon l'ORDER BY. C'est souvent 2-5× plus rapide qu'un CTE avec ROW_NUMBER() ... WHERE rn = 1 sur de grands volumes, car PostgreSQL peut utiliser un index sur (symbol, trade_date DESC). Les deux approches sont correctes — DISTINCT ON est idiomatique pour "dernier enregistrement par partition".
{{ config(materialized='table', tags=['gold', 'sector_analysis']) }}
-- Agrège la performance par secteur pour détecter la rotation sectorielle
-- Alimentera les dashboards Evidence.dev de la Partie 5
SELECT
sector,
instrument_type,
COUNT(DISTINCT isin) AS num_instruments,
ROUND(AVG(perf_1m_pct)::NUMERIC, 2) AS avg_perf_1m,
ROUND(AVG(perf_3m_pct)::NUMERIC, 2) AS avg_perf_3m,
ROUND(AVG(realized_vol_20d)::NUMERIC, 4) AS avg_vol_20d,
ROUND(SUM(adv_20d_usd)::NUMERIC, 0) AS total_sector_adv,
ROUND(AVG(pct_above_sma50)::NUMERIC, 2) AS avg_pct_above_sma50,
-- Score momentum : combinaison performance 1M + momentum SMA50
ROUND((
COALESCE(AVG(perf_1m_pct), 0) * 0.6 +
COALESCE(AVG(pct_above_sma50), 0) * 0.4
)::NUMERIC, 2) AS sector_momentum_score,
-- Classement par momentum (utile pour les APIs MCP de la Partie 5)
RANK() OVER (
PARTITION BY instrument_type
ORDER BY (
COALESCE(AVG(perf_1m_pct), 0) * 0.6 +
COALESCE(AVG(pct_above_sma50), 0) * 0.4
) DESC
) AS momentum_rank,
CURRENT_DATE AS as_of_date
FROM {{ ref('mart_strategy_universe') }}
WHERE sector IS NOT NULL
GROUP BY sector, instrument_typeComme pour les pipelines d'ingestion de la Partie 2, nous utilisons Nomad pour orchestrer les runs dbt. Nomad gère le cycle de vie complet : injection des credentials de base de données depuis Vault, séquençage des étapes (Bronze → Silver → Snapshots → Tests → Gold), et retry automatique en cas d'échec.
dbt Cloud propose un scheduler intégré, mais il n'a pas accès à notre infrastructure on-premise (PostgreSQL en réseau privé, Vault pour les secrets). Le scheduler dbt open source est basique. Nomad s'intègre nativement avec Vault (injection de secrets), le réseau privé, et les alertes Prometheus — et nous l'utilisons déjà pour les taps de la Partie 2.
Un seul système d'orchestration pour toute la stack : les taps d'ingestion (periodic batch) et les runs dbt (periodic batch) sont visibles dans la même UI Nomad, avec les mêmes patterns de logs et d'alertes.
job "dbt-nightly" {
datacenters = ["dc1"]
type = "batch"
namespace = "securities-master"
# Run chaque nuit à 02:00 UTC, après la fin des ingestions (Partie 2)
periodic {
crons = ["0 2 * * *"]
prohibit_overlap = true # Ne pas lancer un nouveau run si le précédent tourne encore
time_zone = "UTC"
}
group "transforms" {
count = 1
# 4 Go de RAM pour les window functions PostgreSQL volumineuses
resources {
cpu = 2000 # MHz
memory = 4096 # MB
}
task "dbt-run" {
driver = "docker"
config {
image = "registry.internal/securities/dbt:1.7.0"
command = "/bin/sh"
args = ["-c", "/app/run_pipeline.sh"]
}
# Injection sécurisée des credentials depuis Vault
template {
destination = "local/dbt_env"
env = true
data = <#!/bin/bash
set -euo pipefail # Arrêt immédiat sur erreur
echo "=== dbt Securities Master Pipeline — $(date -u) ==="
# ── Étape 1 : Tester les sources (raw_zone) ──────────────────────────────────
echo "→ [1/5] Test des sources raw_zone"
dbt test --select "source:raw_zone" --profiles-dir /local
# ── Étape 2 : Construire la couche Bronze (vues) ─────────────────────────────
echo "→ [2/5] Build Bronze (vues JSON)"
dbt run --select "tag:bronze" --profiles-dir /local
# ── Étape 3 : Build Silver (modèles incrémentiels) ───────────────────────────
echo "→ [3/5] Build Silver (incrémentiel)"
dbt run --select "tag:silver" --profiles-dir /local
# ── Étape 4 : Snapshots SCD Type 2 ───────────────────────────────────────────
echo "→ [4/5] Snapshots SCD2"
dbt snapshot --profiles-dir /local
# ── Étape 5 : Tests de qualité Silver ────────────────────────────────────────
echo "→ [5/5] Tests qualité Silver"
dbt test --select "tag:silver" \
--store-failures \ # Stocker les lignes en échec dans la DB
--profiles-dir /local
# ── Étape 6 : Build Gold (uniquement si les tests Silver passent) ─────────────
echo "→ [6/6] Build Gold (marts IA)"
dbt run --select "tag:gold" --profiles-dir /local
# ── Étape 7 : Tests Gold ─────────────────────────────────────────────────────
echo "→ [7/7] Tests Gold"
dbt test --select "tag:gold" --profiles-dir /local
echo "✅ Pipeline dbt terminé avec succès — $(date -u)"Avec --store-failures, dbt ne se contente pas d'indiquer qu'un test a échoué — il stocke les lignes en défaut dans un schéma dédié (dbt_test__audit). Après une alerte, vous pouvez directement requêter dbt_test__audit.not_null_fct_market_data_daily_symbol pour voir exactement quels instruments ont déclenché l'échec. Inestimable pour le debugging sur un système de production.
Les données sont maintenant propres, validées, et historisées. Mais la math est encore fausse. Si Apple fait un split 4-pour-1 demain, notre graphe affichera un crash de 75%. Il faut ajuster les prix historiques.
Cliquez sur chaque question pour révéler la réponse.
QUALIFY ROW_NUMBER() OVER (...) = 1 est-elle invalide en PostgreSQL ?WITH deduped AS (SELECT *, ROW_NUMBER() OVER (...) AS rn FROM ...) SELECT * FROM deduped WHERE rn = 1. Utiliser QUALIFY dans un modèle dbt PostgreSQL provoque une erreur de syntaxe au runtime.
-> et ->> en PostgreSQL ?-> retourne un objet JSONB (avec les guillemets JSON préservés, ex: "42.5" reste de type jsonb). ->> retourne une chaîne TEXT, ce qui permet le cast direct : (payload->>'price')::NUMERIC. Pour les casts numériques, utilisez toujours ->>. Pour accéder à des objets imbriqués que vous voulez encore manipuler avec des opérateurs JSON, utilisez ->. Pour les tableaux JSON, jsonb_array_element_text(payload->'key', index) retourne directement le texte de l'élément.
WHERE trade_date > MAX(trade_date) dans un modèle incrémentiel ?> MAX(trade_date), ces corrections ne seraient jamais ingérées. Le lookback de 3 jours garantit qu'on retraite les derniers jours — et comme unique_key est défini, dbt génère un vrai upsert : les lignes existantes sont mises à jour, les nouvelles sont insérées. Coût : traiter ~3 jours au lieu de 1, soit toujours 400× moins qu'un rebuild complet.
partition_by={'field': 'trade_date', 'data_type': 'date'} dans la config dbt pour PostgreSQL ?dbt-postgres. Le partitionnement PostgreSQL (PARTITION BY RANGE (trade_date)) est une propriété DDL définie à la création de la table — ce qui est fait en Partie 1 via le schéma SQL initial. dbt insère ensuite dans les partitions automatiquement sans aucune configuration supplémentaire dans le modèle. Pour BigQuery, l'option dbt est nécessaire car c'est dbt qui génère le DDL PARTITION BY. Les deux systèmes ont des mécanismes fondamentalement différents.
check ?check_cols. À chaque run de dbt snapshot, il compare ce hash pour chaque ligne entrante avec le hash stocké de la version active (où dbt_valid_to IS NULL). Si les hashes diffèrent (une colonne a changé) : dbt émet un UPDATE pour fermer l'ancienne ligne (dbt_valid_to = NOW()) puis un INSERT pour la nouvelle version (dbt_valid_from = NOW(), dbt_valid_to = NULL). Si les hashes sont identiques, rien ne se passe. Si une clé unique disparaît et que invalidate_hard_deletes=true, la ligne est fermée avec la date du run.
DISTINCT ON (col) et ROW_NUMBER() OVER (...) WHERE rn = 1 en PostgreSQL ?DISTINCT ON est une extension PostgreSQL-native qui peut utiliser un index B-tree sur (partition_col, order_col DESC) directement — très efficace. ROW_NUMBER() via CTE est du SQL standard ANSI, portable sur BigQuery/Snowflake, mais nécessite un scan complet + tri avant le filtre. Pour du code PostgreSQL pur en production (couche Gold), DISTINCT ON est préférable. Pour du code qui doit rester portable entre moteurs, utilisez le pattern CTE avec ROW_NUMBER().
post_hook="ANALYZE {{ this }}" dans la config du mart Gold ?table), les statistiques sont obsolètes. L'optimiseur peut choisir de mauvais plans de requête (ex: Seq Scan au lieu d'Index Scan) car il ne sait pas encore que la distribution a changé. La commande ANALYZE met à jour ces statistiques en échantillonnant la table. En l'ajoutant comme post_hook dans dbt, elle s'exécute automatiquement après chaque rebuild du mart Gold — garantissant des plans de requête optimaux dès la mise en production.