# Point-in-time join : le Feature Store gere automatiquement
# Pour chaque entity_id + event_timestamp, il retourne
# les features les plus recentes AVANT event_timestamp

training_data = store.get_historical_features(
    entity_df=pd.DataFrame({
        "customer_id": ["C001", "C002", "C003"],
        "event_timestamp": [
            datetime(2026, 1, 1),  # Features au 1er janvier
            datetime(2026, 1, 15), # Features au 15 janvier
            datetime(2026, 2, 1),  # Features au 1er fevrier
        ]
    }),
    features=[
        "customer_features:total_purchases_30d",
        "customer_features:avg_order_value_7d",
        "customer_features:days_since_last_order",
    ]
).to_df()

# Installation
pip install feast[redis,snowflake,gcp]

# Initialiser un projet Feast
feast init my_feature_repo
cd my_feature_repo

# Structure generee
my_feature_repo/
├── feature_repo/
│   ├── __init__.py
│   ├── feature_store.yaml    # Configuration
│   ├── entities.py           # Entites (customer, product)
│   ├── features.py           # Feature Views
│   └── data/                 # Donnees sources
└── README.md

project: ecommerce_features
provider: gcp  # ou aws, local
registry: gs://my-bucket/feast/registry.pb
online_store:
  type: redis
  connection_string: "redis://localhost:6379"
offline_store:
  type: bigquery
entity_key_serialization_version: 2

from feast import Entity, FeatureView, Field, FileSource, BigQuerySource
from feast.types import Float64, Int64, String
from datetime import timedelta

# Entite = la cle de jointure (ex: customer_id, product_id)
customer = Entity(
    name="customer",
    join_keys=["customer_id"],
    description="Customer entity for e-commerce"
)

# Source de donnees (BigQuery dans cet exemple)
customer_source = BigQuerySource(
    name="customer_stats_source",
    table="project.dataset.customer_daily_stats",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# Feature View = un ensemble de features pour une entite
customer_features = FeatureView(
    name="customer_features",
    entities=[customer],
    ttl=timedelta(days=3),  # Time-to-live dans l'online store
    schema=[
        Field(name="total_purchases_30d", dtype=Int64),
        Field(name="avg_order_value_7d", dtype=Float64),
        Field(name="days_since_last_order", dtype=Int64),
        Field(name="favorite_category", dtype=String),
        Field(name="lifetime_value", dtype=Float64),
        Field(name="churn_risk_score", dtype=Float64),
    ],
    source=customer_source,
    online=True,  # Materialiser dans l'online store
    tags={"team": "ml-platform", "version": "v2"},
)

from feast import FeatureStore
import pandas as pd
from datetime import datetime

store = FeatureStore(repo_path="./feature_repo")

# 1. Appliquer les definitions (deploy)
# $ feast apply

# 2. Materialiser dans l'online store (batch)
# $ feast materialize 2026-01-01T00:00:00 2026-02-01T00:00:00

# 3. Recuperer les features pour le TRAINING (offline)
training_df = store.get_historical_features(
    entity_df=pd.DataFrame({
        "customer_id": ["C001", "C002", "C003"],
        "event_timestamp": [
            datetime(2026, 1, 1),
            datetime(2026, 1, 15),
            datetime(2026, 2, 1),
        ]
    }),
    features=[
        "customer_features:total_purchases_30d",
        "customer_features:avg_order_value_7d",
        "customer_features:days_since_last_order",
    ]
).to_df()

print(training_df)
# customer_id | event_timestamp | total_purchases_30d | avg_order_value_7d | ...

# 4. Recuperer les features pour l'INFERENCE (online, < 10ms)
online_features = store.get_online_features(
    features=[
        "customer_features:total_purchases_30d",
        "customer_features:avg_order_value_7d",
        "customer_features:churn_risk_score",
    ],
    entity_rows=[
        {"customer_id": "C001"},
        {"customer_id": "C002"},
    ]
).to_dict()

# Passer directement au modele ML
prediction = model.predict(online_features)

-- Feature engineering SQL pour un Feature Store
-- Calculer les features client pour le churn prediction

WITH customer_orders AS (
    SELECT
        customer_id,
        COUNT(*) AS total_orders_30d,
        AVG(order_total) AS avg_order_value_30d,
        MAX(order_total) AS max_order_value_30d,
        SUM(order_total) AS total_revenue_30d,
        COUNT(DISTINCT product_category) AS unique_categories_30d,
        DATEDIFF('day', MAX(order_date), CURRENT_DATE) AS days_since_last_order,
        -- Recency, Frequency, Monetary (RFM)
        NTILE(5) OVER (ORDER BY DATEDIFF('day', MAX(order_date), CURRENT_DATE)) AS recency_quintile,
        NTILE(5) OVER (ORDER BY COUNT(*)) AS frequency_quintile,
        NTILE(5) OVER (ORDER BY SUM(order_total)) AS monetary_quintile
    FROM orders
    WHERE order_date >= DATEADD('day', -30, CURRENT_DATE)
    GROUP BY customer_id
),
customer_sessions AS (
    SELECT
        customer_id,
        COUNT(*) AS sessions_7d,
        AVG(session_duration_sec) AS avg_session_duration_7d,
        SUM(CASE WHEN device = 'mobile' THEN 1 ELSE 0 END)::FLOAT
            / NULLIF(COUNT(*), 0) AS mobile_ratio_7d
    FROM sessions
    WHERE session_date >= DATEADD('day', -7, CURRENT_DATE)
    GROUP BY customer_id
)
SELECT
    o.customer_id,
    CURRENT_TIMESTAMP AS event_timestamp,
    o.total_orders_30d,
    o.avg_order_value_30d,
    o.days_since_last_order,
    o.recency_quintile * o.frequency_quintile * o.monetary_quintile AS rfm_score,
    COALESCE(s.sessions_7d, 0) AS sessions_7d,
    COALESCE(s.avg_session_duration_7d, 0) AS avg_session_duration_7d,
    COALESCE(s.mobile_ratio_7d, 0) AS mobile_ratio_7d
FROM customer_orders o
LEFT JOIN customer_sessions s ON o.customer_id = s.customer_id;

# DVC (Data Version Control) - Versionner les datasets ML

# 1. Initialiser DVC dans le repo
# $ dvc init
# $ dvc remote add -d myremote s3://my-bucket/dvc-data

# 2. Tracker un dataset
# $ dvc add data/training_dataset_v3.parquet
# → Cree data/training_dataset_v3.parquet.dvc (pointeur)

# 3. Push vers le remote storage
# $ dvc push

# 4. Recuperer une version specifique
# $ git checkout v2.0  # Version Git du .dvc file
# $ dvc pull            # Telecharge le dataset correspondant

# Pipeline DVC pour reproduire le training
# dvc.yaml
stages:
  prepare:
    cmd: python src/prepare_data.py
    deps:
      - src/prepare_data.py
      - data/raw/
    outs:
      - data/processed/training.parquet

  train:
    cmd: python src/train.py
    deps:
      - src/train.py
      - data/processed/training.parquet
    params:
      - train.n_estimators
      - train.learning_rate
    metrics:
      - metrics.json:
          cache: false
    outs:
      - models/model.pkl

import mlflow
import mlflow.sklearn
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import accuracy_score, f1_score, roc_auc_score

# Configuration MLflow
mlflow.set_tracking_uri("http://mlflow-server:5000")
mlflow.set_experiment("churn-prediction")

# Training avec MLflow tracking
with mlflow.start_run(run_name="gbm-v3") as run:
    # Log des parametres
    params = {"n_estimators": 200, "learning_rate": 0.1, "max_depth": 5}
    mlflow.log_params(params)

    # Training
    model = GradientBoostingClassifier(**params)
    model.fit(X_train, y_train)

    # Predictions et metriques
    y_pred = model.predict(X_test)
    y_proba = model.predict_proba(X_test)[:, 1]

    metrics = {
        "accuracy": accuracy_score(y_test, y_pred),
        "f1_score": f1_score(y_test, y_pred),
        "auc_roc": roc_auc_score(y_test, y_proba),
    }
    mlflow.log_metrics(metrics)

    # Log du modele + signature
    from mlflow.models import infer_signature
    signature = infer_signature(X_test, y_pred)
    mlflow.sklearn.log_model(
        model, "churn-model",
        signature=signature,
        registered_model_name="churn-predictor"
    )

    # Log artifacts supplementaires
    mlflow.log_artifact("feature_importance.png")
    mlflow.log_dict({"features": feature_names}, "feature_list.json")

    print(f"Run ID: {run.info.run_id}")
    print(f"Metrics: {metrics}")

from mlflow import MlflowClient

client = MlflowClient()

# Promouvoir un modele en Staging
client.transition_model_version_stage(
    name="churn-predictor",
    version=3,
    stage="Staging"
)

# Apres validation, promouvoir en Production
client.transition_model_version_stage(
    name="churn-predictor",
    version=3,
    stage="Production"
)

# Charger le modele en production pour le serving
import mlflow.pyfunc
model = mlflow.pyfunc.load_model("models:/churn-predictor/Production")
predictions = model.predict(new_data)

# Monitoring du data drift avec Evidently
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset

# Comparer les donnees de reference (training) avec la production
drift_report = Report(metrics=[
    DataDriftPreset(),
    DataQualityPreset(),
])

drift_report.run(
    reference_data=training_data,  # Donnees d'entrainement
    current_data=production_data,  # Donnees recentes en prod
)

# Generer le rapport HTML
drift_report.save_html("drift_report.html")

# Verifier programmatiquement si drift detecte
result = drift_report.as_dict()
dataset_drift = result["metrics"][0]["result"]["dataset_drift"]
if dataset_drift:
    print("ALERT: Data drift detected! Retraining recommended.")
    # Trigger retraining pipeline via Airflow/Dagster

-- Installation de pgvector
CREATE EXTENSION vector;

-- Creer une table avec une colonne vector
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    title TEXT,
    content TEXT,
    metadata JSONB,
    embedding vector(1536),  -- OpenAI text-embedding-3-large
    created_at TIMESTAMP DEFAULT NOW()
);

-- Creer un index HNSW (recommande pour < 10M vecteurs)
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

-- Inserer des documents avec embeddings
INSERT INTO documents (title, content, embedding)
VALUES (
    'Guide FinOps',
    'Strategies pour optimiser les couts cloud...',
    '[0.023, -0.045, 0.067, ...]'::vector
);

-- Recherche des 5 documents les plus similaires (cosine)
SELECT
    id, title,
    1 - (embedding <=> query_embedding) AS similarity
FROM documents
ORDER BY embedding <=> '[0.021, -0.048, 0.069, ...]'::vector
LIMIT 5;

-- Hybrid search : vecteur + filtre metadonnees
SELECT id, title,
    1 - (embedding <=> query_embedding) AS similarity
FROM documents
WHERE metadata->>'category' = 'architecture'
  AND created_at > '2026-01-01'
ORDER BY embedding <=> query_embedding
LIMIT 5;

from langchain_community.vectorstores import PGVector
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Configuration
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
connection_string = "postgresql://user:pass@localhost:5432/vectordb"

# Charger et decouper les documents
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " "]
)
chunks = text_splitter.split_documents(documents)

# Creer le vector store
vectorstore = PGVector.from_documents(
    documents=chunks,
    embedding=embeddings,
    connection_string=connection_string,
    collection_name="architecture_docs",
)

# Recherche semantique
results = vectorstore.similarity_search_with_score(
    query="Comment optimiser les couts Snowflake ?",
    k=5
)
for doc, score in results:
    print(f"Score: {score:.3f} | {doc.page_content[:100]}...")

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import PGVector
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 1. Chunking avec overlap
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,       # ~250 mots
    chunk_overlap=200,     # 20% overlap pour le contexte
    separators=["\n\n", "\n", ". ", " "],
    length_function=len,
)
chunks = text_splitter.split_documents(raw_documents)
print(f"{len(raw_documents)} docs → {len(chunks)} chunks")

# 2. Embedding et stockage
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = PGVector.from_documents(
    documents=chunks,
    embedding=embeddings,
    connection_string=CONNECTION_STRING,
    collection_name="company_docs",
)

# 3. RAG Chain avec prompt custom
prompt_template = """Tu es un assistant expert en data architecture.
Utilise UNIQUEMENT le contexte ci-dessous pour repondre.
Si tu ne sais pas, dis "Je n'ai pas cette information dans mes sources."

Contexte:
{context}

Question: {question}

Reponse detaillee:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# 4. Chain RAG complete
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 5}
    ),
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True,
)

# 5. Utilisation
result = qa_chain.invoke({"query": "Comment optimiser les couts Snowflake ?"})
print(result["result"])
for doc in result["source_documents"]:
    print(f"  Source: {doc.metadata['source']}")

═══ STRATEGIES D'OPTIMISATION (par impact) ═══

1. SEMANTIC CACHING (economie: 30-60%)
   Stocker les reponses dans un vector DB.
   Si une question similaire arrive (cosine > 0.95),
   retourner la reponse cachee au lieu d'appeler le LLM.
   → Cout: 1 embedding ($0.0001) vs 1 appel GPT-4 ($0.03)

2. SMART ROUTING (economie: 40-70%)
   Questions simples → Modele petit (GPT-3.5, Mistral 7B)
   Questions complexes → Modele puissant (GPT-4o, Claude)
   Classification automatique de la complexite.
   → GPT-3.5: $0.001/req vs GPT-4: $0.03/req

3. PROMPT OPTIMIZATION (economie: 20-40%)
   Reduire la taille des prompts systeme.
   Utiliser des exemples concis (few-shot).
   Limiter le contexte RAG aux chunks les plus pertinents.
   → Moins de tokens = moins cher

4. BATCHING (economie: 15-25%)
   Regrouper les requetes non-urgentes.
   Traitement batch pendant les heures creuses.
   → Meilleur throughput, meilleur prix

5. QUANTIZATION (economie: 50-70% sur GPU)
   Self-hosted: utiliser des modeles quantizes (GPTQ, AWQ).
   Llama 3.1 70B en 4-bit: 40GB VRAM vs 140GB en FP16.
   → Meme GPU, modele 3x plus grand

# prompts/rag_assistant/v3.yaml
name: "rag_data_architect"
version: "3.2.0"
model: "gpt-4o"
temperature: 0.1
max_tokens: 2000

system_prompt: |
  Tu es un assistant expert en data architecture.
  Regles:
  - Reponds UNIQUEMENT avec le contexte fourni
  - Cite tes sources (nom du document, section)
  - Si tu ne sais pas, dis-le clairement
  - Utilise des exemples concrets
  - Structure ta reponse avec des bullet points

user_template: |
  Contexte:
  {context}

  Question de l'utilisateur:
  {question}

  Instructions supplementaires:
  - Niveau de detail: {detail_level}
  - Langue: francais

metadata:
  author: "data-platform-team"
  created: "2026-01-15"
  tags: ["rag", "data-architecture", "production"]
  tests:
    - "test_no_hallucination"
    - "test_source_citation"
    - "test_language_fr"

{"messages": [
  {"role": "system", "content": "Tu es un expert Data Architect qui repond de facon concise et technique."},
  {"role": "user", "content": "Quand utiliser Snowflake vs BigQuery ?"},
  {"role": "assistant", "content": "**Snowflake** : SQL standard, separation compute/storage, ideal si multi-cloud ou data sharing. **BigQuery** : serverless natif, ideal si full GCP, ML integre avec Vertex AI. Choisir selon : cloud existant, pattern de workload (concurrency vs burst), et budget (Snowflake = credit-based, BigQuery = pay-per-query)."}
]}
{"messages": [
  {"role": "system", "content": "Tu es un expert Data Architect qui repond de facon concise et technique."},
  {"role": "user", "content": "Comment implementer un Data Mesh ?"},
  {"role": "assistant", "content": "1. **Identifier les domaines** metier (sales, product, marketing). 2. **Definir les data products** par domaine avec SLAs et documentation. 3. **Creer la self-serve platform** (infra commune : data lake, orchestration, CI/CD). 4. **Implementer la federated governance** : standards globaux (naming, quality) + autonomie locale. Prerequis : maturite org, >100 personnes data, domaines clairement definis."}
]}

# Guardrails basiques pour un LLM en production
from dataclasses import dataclass

@dataclass
class GuardrailConfig:
    max_tokens: int = 2000
    blocked_topics: list = None  # ["politique", "religion"]
    pii_detection: bool = True
    source_required: bool = True
    language: str = "fr"

def apply_guardrails(prompt: str, response: str, config: GuardrailConfig):
    """Verifier la reponse avant de la retourner a l'utilisateur."""
    issues = []

    # 1. PII Detection
    if config.pii_detection:
        pii_patterns = [r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
                       r'\b[A-Z]{2}\d{2}[A-Z\d]{10,30}\b']  # IBAN
        for pattern in pii_patterns:
            if re.search(pattern, response):
                issues.append("PII detected in response")

    # 2. Blocked topics
    if config.blocked_topics:
        for topic in config.blocked_topics:
            if topic.lower() in response.lower():
                issues.append(f"Blocked topic: {topic}")

    # 3. Length check
    if len(response.split()) > config.max_tokens:
        issues.append("Response too long")

    # 4. Source citation check
    if config.source_required and "source" not in response.lower():
        issues.append("No source citation")

    return {"safe": len(issues) == 0, "issues": issues}

rag-pipeline-lab/
├── README.md
├── docker-compose.yml          # PostgreSQL + pgvector
├── requirements.txt
├── .env.example                # API keys (never commit .env!)
├── docs/                       # Documents sources
│   ├── architecture/
│   ├── runbooks/
│   └── adrs/
├── src/
│   ├── ingest.py              # Ingestion + chunking + embedding
│   ├── retriever.py           # Recherche vectorielle + reranking
│   ├── generator.py           # LLM generation avec prompt
│   ├── pipeline.py            # RAG chain complete
│   └── evaluate.py            # Evaluation automatique
├── prompts/
│   └── rag_assistant_v1.yaml  # Prompt versionne
├── eval/
│   ├── test_questions.json    # Dataset d'evaluation
│   └── results/
├── tests/
│   └── test_pipeline.py
└── Makefile

"""Pipeline RAG complet avec reranking et evaluation."""
import os
from langchain_community.vectorstores import PGVector
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

class RAGPipeline:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-large"
        )
        self.llm = ChatOpenAI(
            model="gpt-4o",
            temperature=0.1,
            max_tokens=2000
        )
        self.connection_string = os.getenv("PGVECTOR_CONNECTION")

    def ingest(self, docs_path: str):
        """Ingerer les documents : charger, chunker, embedder, stocker."""
        # Charger
        loader = DirectoryLoader(docs_path, glob="**/*.md", loader_cls=TextLoader)
        documents = loader.load()
        print(f"Loaded {len(documents)} documents")

        # Chunker
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=200,
            separators=["\n## ", "\n### ", "\n\n", "\n", ". "]
        )
        chunks = splitter.split_documents(documents)
        print(f"Split into {len(chunks)} chunks")

        # Embedder et stocker
        self.vectorstore = PGVector.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            connection_string=self.connection_string,
            collection_name="tech_docs",
            pre_delete_collection=True,
        )
        print("Ingestion complete!")
        return len(chunks)

    def query(self, question: str, k: int = 5) -> dict:
        """Interroger le pipeline RAG."""
        retriever = self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": k}
        )
        prompt = PromptTemplate(
            template="""Contexte:\n{context}\n\nQuestion: {question}\n
Reponds en francais, cite tes sources.""",
            input_variables=["context", "question"]
        )
        chain = RetrievalQA.from_chain_type(
            llm=self.llm, retriever=retriever,
            chain_type_kwargs={"prompt": prompt},
            return_source_documents=True,
        )
        return chain.invoke({"query": question})

# Usage
if __name__ == "__main__":
    rag = RAGPipeline()
    rag.ingest("./docs")
    result = rag.query("Comment optimiser les couts Snowflake ?")
    print(result["result"])

@prefix ex: <http://example.org/> .
@prefix foaf: <http://xmlns.com/foaf/0.1/> .
@prefix schema: <http://schema.org/> .

# Personnes
ex:alice a foaf:Person ;
    foaf:name "Alice Martin" ;
    ex:role "Data Architect" ;
    ex:competence ex:snowflake, ex:dbt, ex:kafka ;
    ex:employe_de ex:datacorp ;
    ex:certifie ex:cdmp, ex:snowpro .

ex:bob a foaf:Person ;
    foaf:name "Bob Dupont" ;
    ex:role "Data Engineer" ;
    ex:competence ex:python, ex:airflow, ex:spark ;
    ex:employe_de ex:datacorp .

# Entreprise
ex:datacorp a schema:Organization ;
    schema:name "DataCorp" ;
    schema:location ex:paris ;
    ex:utilise ex:snowflake, ex:dbt, ex:airflow ;
    ex:industrie "E-commerce" .

# Technologies
ex:snowflake a ex:Technology ;
    ex:categorie "Cloud Data Warehouse" ;
    ex:vendor "Snowflake Inc." .

ex:dbt a ex:Technology ;
    ex:categorie "Transformation" ;
    ex:licence "Open Source" .

# Requete 1 : Trouver toutes les competences d'Alice
SELECT ?competence WHERE {
    ex:alice ex:competence ?competence .
}

# Requete 2 : Trouver les personnes qui maitrisent Snowflake
SELECT ?personne ?nom WHERE {
    ?personne ex:competence ex:snowflake ;
              foaf:name ?nom .
}

# Requete 3 : Trouver les technologies open source utilisees par DataCorp
SELECT ?tech ?categorie WHERE {
    ex:datacorp ex:utilise ?tech .
    ?tech ex:licence "Open Source" ;
          ex:categorie ?categorie .
}

# Requete 4 : Recommandation - qui peut aider sur Kafka ?
SELECT ?personne ?nom ?role WHERE {
    ?personne ex:competence ex:kafka ;
              foaf:name ?nom ;
              ex:role ?role .
}

# Requete 5 : Chemin entre Alice et une technologie (graph traversal)
SELECT ?tech ?categorie WHERE {
    ex:alice ex:employe_de ?company .
    ?company ex:utilise ?tech .
    ?tech ex:categorie ?categorie .
    FILTER NOT EXISTS { ex:alice ex:competence ?tech }
}
# → Technologies utilisees par l'entreprise qu'Alice ne maitrise pas encore

// Creer des noeuds et relations
CREATE (alice:Person {name: "Alice", role: "Data Architect", salary: 95000})
CREATE (bob:Person {name: "Bob", role: "Data Engineer", salary: 75000})
CREATE (sf:Technology {name: "Snowflake", category: "Cloud DWH"})
CREATE (dbt:Technology {name: "dbt", category: "Transformation", license: "OSS"})
CREATE (dc:Company {name: "DataCorp", industry: "E-commerce"})

CREATE (alice)-[:WORKS_AT {since: 2024}]->(dc)
CREATE (bob)-[:WORKS_AT {since: 2025}]->(dc)
CREATE (alice)-[:SKILLED_IN {level: "expert"}]->(sf)
CREATE (alice)-[:SKILLED_IN {level: "expert"}]->(dbt)
CREATE (bob)-[:SKILLED_IN {level: "intermediate"}]->(sf)
CREATE (dc)-[:USES]->(sf)
CREATE (dc)-[:USES]->(dbt)
CREATE (alice)-[:MENTORS]->(bob);

// Requete 1 : Trouver les experts Snowflake
MATCH (p:Person)-[s:SKILLED_IN]->(t:Technology {name: "Snowflake"})
WHERE s.level = "expert"
RETURN p.name, p.role;

// Requete 2 : Technologies non maitrisees par Bob mais utilisees par son entreprise
MATCH (bob:Person {name: "Bob"})-[:WORKS_AT]->(c:Company)-[:USES]->(t:Technology)
WHERE NOT (bob)-[:SKILLED_IN]->(t)
RETURN t.name AS tech_to_learn, t.category;

// Requete 3 : Chemin le plus court entre deux personnes
MATCH path = shortestPath(
    (alice:Person {name: "Alice"})-[*]-(bob:Person {name: "Bob"})
)
RETURN path;

// Requete 4 : Recommandation - personnes avec competences similaires
MATCH (alice:Person {name: "Alice"})-[:SKILLED_IN]->(tech)<-[:SKILLED_IN]-(similar)
WHERE alice <> similar
RETURN similar.name, COUNT(tech) AS common_skills
ORDER BY common_skills DESC;

// Modeliser un Data Catalog dans Neo4j
CREATE (db:Database {name: "analytics_dwh", type: "Snowflake"})
CREATE (t1:Table {name: "raw_orders", schema: "bronze", row_count: 5000000})
CREATE (t2:Table {name: "stg_orders", schema: "silver"})
CREATE (t3:Table {name: "fct_orders", schema: "gold"})
CREATE (d1:Dashboard {name: "Revenue Dashboard", tool: "Metabase"})
CREATE (team:Team {name: "Sales Analytics"})

CREATE (db)-[:CONTAINS]->(t1)
CREATE (db)-[:CONTAINS]->(t2)
CREATE (db)-[:CONTAINS]->(t3)
CREATE (t1)-[:FEEDS]->(t2)
CREATE (t2)-[:FEEDS]->(t3)
CREATE (t3)-[:CONSUMED_BY]->(d1)
CREATE (t3)-[:OWNED_BY]->(team)

// Impact Analysis : si raw_orders change, qui est impacte ?
MATCH (source:Table {name: "raw_orders"})-[:FEEDS*]->(downstream)
RETURN downstream.name, labels(downstream)[0] AS type;

# glossary/business_terms.yaml
terms:
  revenue:
    display_name: "Revenue"
    description: "Total des ventes nettes (hors taxes, hors retours)"
    formula: "SUM(order_total) - SUM(returns) - SUM(taxes)"
    source_table: "gold.fct_orders"
    owner: "finance-team@company.com"
    synonyms: ["chiffre d'affaires", "CA", "sales revenue"]
    related_terms: ["gross_revenue", "net_revenue", "arr"]
    data_type: "decimal"
    unit: "EUR"
    refresh: "daily, 6:00 UTC"
    sla: "available by 7:00 UTC"

  active_customer:
    display_name: "Client actif"
    description: "Client ayant effectue au moins 1 achat dans les 90 derniers jours"
    formula: "customer WHERE last_order_date >= CURRENT_DATE - 90"
    source_table: "gold.dim_customers"
    owner: "marketing-team@company.com"
    synonyms: ["client engage", "utilisateur actif"]
    related_terms: ["churned_customer", "new_customer"]

  churn_rate:
    display_name: "Taux de churn"
    description: "Pourcentage de clients perdus sur une periode"
    formula: "(clients_debut - clients_fin) / clients_debut * 100"
    source_table: "gold.fct_customer_metrics"
    owner: "product-team@company.com"
    granularity: "monthly"