diff --git a/README.md b/README.md
index 7b7142e0..7b57f431 100644
--- a/README.md
+++ b/README.md
@@ -118,6 +118,31 @@ Il toolkit non gestisce il deployment: scrive nella directory configurata via
 
 ---
 
+## Scrivere clean.sql con le macro standard
+
+Il toolkit fornisce **macro SQL DuckDB** precaricate automaticamente in ogni
+esecuzione del layer CLEAN. Servono a scrivere `clean.sql` senza riscrivere
+ogni volta `TRY_CAST`, `REPLACE` per numeri italiani, `CASE` per flag booleani.
+
+Esempio — invece di:
+
+```sql
+TRY_CAST(REPLACE(REPLACE("Importo"::VARCHAR, '.', ''), ',', '.') AS DOUBLE) AS importo,
+CASE WHEN TRIM("ETS") = 'X' THEN TRUE ELSE FALSE END AS flag_ets,
+TRIM(CAST("Denominazione" AS VARCHAR)) AS denominazione,
+```
+
+si scrive:
+
+```sql
+normalize_italian_number("Importo") AS importo,
+decode_flag("ETS", 'X') AS flag_ets,
+normalize_string("Denominazione") AS denominazione,
+```
+
+Le macro sono caricate automaticamente — **non serve importare nulla**.
+Dettaglio completo: [docs/standard-macros.md](docs/standard-macros.md).
+
 ## Configurazione (`dataset.yml`)
 
 Il cuore del toolkit è un file YAML che descrive il dataset:
@@ -153,6 +178,7 @@ esegue le trasformazioni SQL su DuckDB e produce output in `root/data/`.
 | [advanced-workflows.md](docs/advanced-workflows.md) | Resume, run parziali, profile, debug |
 | [notebook-contract.md](docs/notebook-contract.md) | Come leggere gli output nei notebook |
 | [feature-stability.md](docs/feature-stability.md) | Cosa è stabile, cosa sperimentale, cosa deprecated |
+| [standard-macros.md](docs/standard-macros.md) | Macro SQL predefinite per clean.sql |
 
 ### Plugin sorgente supportati
 
diff --git a/docs/standard-macros.md b/docs/standard-macros.md
new file mode 100644
index 00000000..4941c77c
--- /dev/null
+++ b/docs/standard-macros.md
@@ -0,0 +1,170 @@
+# Standard SQL Macros
+
+Le **macro SQL** sono funzioni DuckDB precaricate automaticamente in ogni esecuzione del layer CLEAN. Servono a eliminare il boilerplate che oggi è riscritto da zero in ogni `clean.sql` — `TRY_CAST`, `REPLACE` per numeri italiani, `CASE` per flag booleani, `TRIM` per stringhe.
+
+## Come funzionano
+
+Le macro sono definite in `toolkit/sql/macros.sql` (dentro il pacchetto, distribuite via wheel). Quando il toolkit esegue un `clean.sql`, carica automaticamente tutte le macro nella connessione DuckDB prima di eseguire la query. Ogni macro è `CREATE OR REPLACE` — sicura da eseguire più volte.
+
+**Non serve** importare nulla, includere file o modificare `dataset.yml`. Le macro sono sempre disponibili.
+
+## Elenco macro
+
+### `normalize_string(val)`
+
+`TRIM` + stringa vuota → `NULL`. Per colonne testuali.
+
+```sql
+-- PRIMA
+TRIM(CAST("Denominazione" AS VARCHAR)) AS denominazione,
+NULLIF(TRIM("Codice"::VARCHAR), '') AS codice,
+
+-- DOPO
+normalize_string("Denominazione") AS denominazione,
+normalize_string("Codice") AS codice,
+```
+
+### `cast_int(val)`
+
+`TRY_CAST(val AS INTEGER)`. Per colonne numeriche intere (32-bit).
+
+```sql
+-- PRIMA
+TRY_CAST("Prog" AS INTEGER) AS progressivo,
+CAST("Anno" AS INTEGER) AS anno,
+
+-- DOPO
+cast_int("Prog") AS progressivo,
+cast_int("Anno") AS anno,
+```
+
+### `cast_bigint(val)`
+
+`TRY_CAST(val AS BIGINT)`. Per colonne numeriche grandi (64-bit). Usato dallo scaffold per colonne mappate come `int`/`bigint`.
+
+```sql
+-- PRIMA
+TRY_CAST("Numero contribuenti" AS BIGINT) AS numero_contribuenti,
+
+-- DOPO
+cast_bigint("Numero contribuenti") AS numero_contribuenti,
+```
+
+### `cast_double(val)`
+
+`TRY_CAST(val AS DOUBLE)`. Per colonne numeriche con decimali.
+
+```sql
+-- PRIMA
+TRY_CAST(TRIM(CAST("Importo" AS VARCHAR)) AS DOUBLE) AS importo,
+
+-- DOPO
+cast_double("Importo") AS importo,
+```
+
+### `normalize_italian_number(val)`
+
+Converte un numero in formato italiano (`1.234,56` → `1234.56`). Rimuove punti migliaia, converte virgola decimale in punto. `TRY_CAST` restituisce `NULL` se la conversione fallisce.
+
+```sql
+-- PRIMA (15 righe per 4 colonne)
+TRY_CAST(REPLACE(REPLACE("Importo"::VARCHAR, '.', ''), ',', '.') AS DOUBLE) AS importo,
+TRY_CAST(REPLACE(REPLACE("Numero scelte"::VARCHAR, '.', ''), ',', '.') AS INTEGER) AS numero_scelte,
+
+-- DOPO (2 righe)
+normalize_italian_number("Importo") AS importo,
+normalize_italian_integer("Numero scelte") AS numero_scelte,
+```
+
+### `normalize_italian_integer(val)`
+
+Come `normalize_italian_number` ma restituisce `INTEGER`. DuckDB `CAST(DOUBLE AS INTEGER)` **arrotonda** (non tronca): `5.432,90` → `5433`.
+
+### `decode_flag(val, yes_value)`
+
+Decodifica un flag testuale in `BOOLEAN`. Il secondo argomento è il valore che rappresenta `TRUE`.
+
+```sql
+-- PRIMA
+CASE WHEN TRIM("ETS") = 'X' THEN TRUE ELSE FALSE END AS flag_ets,
+
+-- DOPO
+decode_flag("ETS", 'X') AS flag_ets,
+```
+
+### `remove_dot_thousands(val)`
+
+Rimuove punti migliaia da **numeri interi**. Attenzione: rimuove **tutti** i punti, incluso un eventuale separatore decimale standard. Usa solo su interi con punti migliaia. Per numeri con decimali usa `normalize_italian_number` o `cast_double`.
+
+```sql
+-- SOLO PER INTERI
+remove_dot_thousands("Popolazione") AS popolazione,  -- "1.234" → 1234.0
+
+-- NON USARE SU DECIMALI — usa invece:
+normalize_italian_number("Importo") AS importo,        -- "1.234,56" → 1234.56
+cast_double("Valore") AS valore,                       -- "1234.56" → 1234.56
+```
+
+## Esempio completo
+
+Prima (`ade-cinque-per-mille/sql/clean.sql`, 21 righe):
+
+```sql
+SELECT
+    {year}::INTEGER AS anno,
+    TRY_CAST("Prog" AS INTEGER) AS progressivo,
+    TRIM("Codice fiscale") AS codice_fiscale,
+    TRIM("Denominazione") AS denominazione,
+    TRIM("Regione") AS regione,
+    TRIM("PR") AS sigla_provincia,
+    TRIM("Comune") AS comune,
+    CASE WHEN TRIM("ETS") = 'X' THEN TRUE ELSE FALSE END AS flag_ets_onlus,
+    CASE WHEN TRIM("ASD") = 'X' THEN TRUE ELSE FALSE END AS flag_asd,
+    -- ... altri CASE WHEN identici ...
+    TRY_CAST(REPLACE(REPLACE("Numero scelte"::VARCHAR, '.', ''), ',', '.') AS INTEGER) AS numero_scelte,
+    TRY_CAST(REPLACE(REPLACE("Importo delle scelte espresse"::VARCHAR, '.', ''), ',', '.') AS DOUBLE) AS importo_scelte_espresse,
+FROM raw_input
+```
+
+Dopo (18 righe, zero boilerplate — solo logica di dominio):
+
+```sql
+SELECT
+    {year}::INTEGER AS anno,
+    cast_int("Prog") AS progressivo,
+    normalize_string("Codice fiscale") AS codice_fiscale,
+    normalize_string("Denominazione") AS denominazione,
+    normalize_string("Regione") AS regione,
+    normalize_string("PR") AS sigla_provincia,
+    normalize_string("Comune") AS comune,
+    decode_flag("ETS", 'X') AS flag_ets_onlus,
+    decode_flag("ASD", 'X') AS flag_asd,
+    -- ... altri decode_flag identici ...
+    normalize_italian_integer("Numero scelte") AS numero_scelte,
+    normalize_italian_number("Importo delle scelte espresse") AS importo_scelte_espresse,
+FROM raw_input
+```
+
+## Perché macro SQL invece di Python preprocessing?
+
+| Approccio | Dove opera | Pro |
+|---|---|---|
+| **Macro DuckDB** | Dentro `clean.sql` | Puro SQL, testabile in DuckDB CLI, zero dipendenze Python |
+| `normalize.py` (Python) | Prima di DuckDB (script/extractor) | Logica più complessa, gestione encoding, regex rename |
+
+Le macro DuckDB e le funzioni Python in `toolkit/core/normalize.py` sono complementari: `normalize.py` prepara i dati prima che entrino in DuckDB, le macro lavorano dentro DuckDB.
+
+## Verifica
+
+Per testare una macro direttamente:
+
+```bash
+python -c "
+import duckdb
+con = duckdb.connect()
+con.execute(open('toolkit/sql/macros.sql').read())
+print(con.execute(\"SELECT normalize_italian_number('1.234,56')\").fetchone())
+"
+```
+
+I test automatici: `pytest tests/test_macros_sql.py -v` (35 test).
diff --git a/project-example/sql/clean.sql b/project-example/sql/clean.sql
index 610345a6..9445d560 100644
--- a/project-example/sql/clean.sql
+++ b/project-example/sql/clean.sql
@@ -25,31 +25,20 @@ WITH base AS (
 )
 
 SELECT
-  CAST({year} AS INTEGER) AS anno,
-
-  CAST(TRIM(regione)  AS VARCHAR) AS regione,
-  CAST(TRIM(provincia) AS VARCHAR) AS provincia,
-  CAST(TRIM(comune)   AS VARCHAR) AS comune,
-
-  CAST(
-    NULLIF(
-      REPLACE(
-        REPLACE(
-          REPLACE(TRIM(CAST(pct_rd_raw AS VARCHAR)), '%', ''),
-        '.', ''),
-      ',', '.'),
-    '-')
-    AS DOUBLE
-  ) AS pct_rd,
-
-  CAST(
-    NULLIF(
-      REPLACE(
-        REPLACE(TRIM(CAST(ru_tot_t_raw AS VARCHAR)), '.', ''),
-      ',', '.'),
-    '-')
-    AS DOUBLE
-  ) AS ru_tot_t
+    cast_int({year}) AS anno,
+
+    normalize_string(regione) AS regione,
+    normalize_string(provincia) AS provincia,
+    normalize_string(comune) AS comune,
+
+    -- Il formato italiano (%, . e ,) viene normalizzato dal toolkit:
+    -- normalize_italian_number gestisce 1.234,56 → 1234.56
+    -- La % viene rimossa manualmente prima della macro
+    normalize_italian_number(
+      REPLACE(normalize_string(pct_rd_raw), '%', '')
+    ) AS pct_rd,
+
+    normalize_italian_number(ru_tot_t_raw) AS ru_tot_t
 
 FROM base
 WHERE regione  IS NOT NULL AND TRIM(regione)  <> ''
diff --git a/pyproject.toml b/pyproject.toml
index b3e1c079..c9e38bfb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -97,6 +97,9 @@ Documentation = "https://github.com/dataciviclab/toolkit"
 include = ["toolkit*"]
 exclude = ["tests*"]
 
+[tool.setuptools.package-data]
+toolkit = ["sql/*.sql"]
+
 [tool.setuptools.dynamic]
 version = { attr = "toolkit.version.__version__" }
 
diff --git a/smoke/bdap_ckan_csv/sql/clean.sql b/smoke/bdap_ckan_csv/sql/clean.sql
index c31afca9..cb68576a 100644
--- a/smoke/bdap_ckan_csv/sql/clean.sql
+++ b/smoke/bdap_ckan_csv/sql/clean.sql
@@ -1,13 +1,13 @@
 WITH base AS (
   SELECT
-    TRY_CAST(TRIM(CAST("Anno di Riferimento" AS VARCHAR)) AS INTEGER) AS anno,
-    TRY_CAST(TRIM(CAST("Codice Regione" AS VARCHAR)) AS INTEGER) AS codice_regione,
-    TRIM(CAST("Descrizione Regione" AS VARCHAR)) AS regione,
-    TRY_CAST(TRIM(CAST("Codice Ente SSN" AS VARCHAR)) AS INTEGER) AS codice_ente_ssn,
-    TRIM(CAST("Descrizione Ente" AS VARCHAR)) AS descrizione_ente,
-    TRIM(CAST("Codice Voce Contabile" AS VARCHAR)) AS codice_voce_contabile,
-    TRIM(CAST("Descrizione Voce Contabile" AS VARCHAR)) AS descrizione_voce_contabile,
-    TRY_CAST(TRIM(CAST("Importo Totale" AS VARCHAR)) AS DOUBLE) AS importo_totale
+    cast_int("Anno di Riferimento") AS anno,
+    cast_int("Codice Regione") AS codice_regione,
+    normalize_string("Descrizione Regione") AS regione,
+    cast_int("Codice Ente SSN") AS codice_ente_ssn,
+    normalize_string("Descrizione Ente") AS descrizione_ente,
+    normalize_string("Codice Voce Contabile") AS codice_voce_contabile,
+    normalize_string("Descrizione Voce Contabile") AS descrizione_voce_contabile,
+    cast_double("Importo Totale") AS importo_totale
   FROM raw_input
 )
 
diff --git a/smoke/bdap_http_csv/sql/clean.sql b/smoke/bdap_http_csv/sql/clean.sql
index 22585a24..724b1011 100644
--- a/smoke/bdap_http_csv/sql/clean.sql
+++ b/smoke/bdap_http_csv/sql/clean.sql
@@ -1,30 +1,27 @@
 WITH base AS (
   SELECT
-    TRY_CAST(TRIM(CAST("ANNO" AS VARCHAR)) AS INTEGER) AS anno,
-
-    TRY_CAST(TRIM(CAST("RISPARMIO_PUBBLICO" AS VARCHAR)) AS DOUBLE) AS risparmio_pubblico,
-    TRY_CAST(TRIM(CAST("SALDO_NETTO" AS VARCHAR)) AS DOUBLE) AS saldo_netto,
-    TRY_CAST(TRIM(CAST("INDEBITAMENTO_NETTO" AS VARCHAR)) AS DOUBLE) AS indebitamento_netto,
-    TRY_CAST(TRIM(CAST("RICORSO_MERCATO" AS VARCHAR)) AS DOUBLE) AS ricorso_mercato,
-    TRY_CAST(TRIM(CAST("AVANZO_PRIMARIO" AS VARCHAR)) AS DOUBLE) AS avanzo_primario,
-
-    TRY_CAST(TRIM(CAST("SPESE_CORRENTI" AS VARCHAR)) AS DOUBLE) AS spese_correnti,
-    TRY_CAST(TRIM(CAST("SPESE_INTERESSI" AS VARCHAR)) AS DOUBLE) AS spese_interessi,
-    TRY_CAST(TRIM(CAST("SPESE_CONTO_CAPITALE" AS VARCHAR)) AS DOUBLE) AS spese_conto_capitale,
-    TRY_CAST(TRIM(CAST("SPESE_ACQ_ATT_FINE" AS VARCHAR)) AS DOUBLE) AS spese_acq_att_fin,
-    TRY_CAST(TRIM(CAST("SPESE_RIMBORSO_PRESTITI" AS VARCHAR)) AS DOUBLE) AS spese_rimborso_prestiti,
-    TRY_CAST(TRIM(CAST("SPESE_COMPLESSIVE" AS VARCHAR)) AS DOUBLE) AS spese_complessive,
-    TRY_CAST(TRIM(CAST("SPESE_FINALI" AS VARCHAR)) AS DOUBLE) AS spese_finali,
-    TRY_CAST(TRIM(CAST("SPESE_FIN_NETTO_ATT_FIN" AS VARCHAR)) AS DOUBLE) AS spese_fin_netto_att_fin,
-
-    TRY_CAST(TRIM(CAST("ENTRATE_TRIBUTARIE" AS VARCHAR)) AS DOUBLE) AS entrate_tributarie,
-    TRY_CAST(TRIM(CAST("ENTRATE_EXTRA_TRIBUTARIE" AS VARCHAR)) AS DOUBLE) AS entrate_extra_tributarie,
-    TRY_CAST(TRIM(CAST("ENTR_ALIEN_PATR_RISCOS" AS VARCHAR)) AS DOUBLE) AS entr_alien_patr_riscos,
-    TRY_CAST(TRIM(CAST("RISCOSSIONE_CREDITI" AS VARCHAR)) AS DOUBLE) AS riscossione_crediti,
-    TRY_CAST(TRIM(CAST("ENTR_ACCENSIONE_PRESTITI" AS VARCHAR)) AS DOUBLE) AS entr_accensione_prestiti,
-    TRY_CAST(TRIM(CAST("ENTRATE_FINALI" AS VARCHAR)) AS DOUBLE) AS entrate_finali,
-    TRY_CAST(TRIM(CAST("ENTR_FIN_NETTO_RISCO_CRED" AS VARCHAR)) AS DOUBLE) AS entr_fin_netto_risco_cred,
-    TRY_CAST(TRIM(CAST("ENTRATE_CORRENTI" AS VARCHAR)) AS DOUBLE) AS entrate_correnti
+    cast_int("ANNO") AS anno,
+    cast_double("RISPARMIO_PUBBLICO") AS risparmio_pubblico,
+    cast_double("SALDO_NETTO") AS saldo_netto,
+    cast_double("INDEBITAMENTO_NETTO") AS indebitamento_netto,
+    cast_double("RICORSO_MERCATO") AS ricorso_mercato,
+    cast_double("AVANZO_PRIMARIO") AS avanzo_primario,
+    cast_double("SPESE_CORRENTI") AS spese_correnti,
+    cast_double("SPESE_INTERESSI") AS spese_interessi,
+    cast_double("SPESE_CONTO_CAPITALE") AS spese_conto_capitale,
+    cast_double("SPESE_ACQ_ATT_FINE") AS spese_acq_att_fin,
+    cast_double("SPESE_RIMBORSO_PRESTITI") AS spese_rimborso_prestiti,
+    cast_double("SPESE_COMPLESSIVE") AS spese_complessive,
+    cast_double("SPESE_FINALI") AS spese_finali,
+    cast_double("SPESE_FIN_NETTO_ATT_FIN") AS spese_fin_netto_att_fin,
+    cast_double("ENTRATE_TRIBUTARIE") AS entrate_tributarie,
+    cast_double("ENTRATE_EXTRA_TRIBUTARIE") AS entrate_extra_tributarie,
+    cast_double("ENTR_ALIEN_PATR_RISCOS") AS entr_alien_patr_riscos,
+    cast_double("RISCOSSIONE_CREDITI") AS riscossione_crediti,
+    cast_double("ENTR_ACCENSIONE_PRESTITI") AS entr_accensione_prestiti,
+    cast_double("ENTRATE_FINALI") AS entrate_finali,
+    cast_double("ENTR_FIN_NETTO_RISCO_CRED") AS entr_fin_netto_risco_cred,
+    cast_double("ENTRATE_CORRENTI") AS entrate_correnti
   FROM raw_input
 )
 
diff --git a/smoke/finanze_http_zip_2023/sql/clean.sql b/smoke/finanze_http_zip_2023/sql/clean.sql
index f4b1fc4d..4a148453 100644
--- a/smoke/finanze_http_zip_2023/sql/clean.sql
+++ b/smoke/finanze_http_zip_2023/sql/clean.sql
@@ -1,9 +1,9 @@
 SELECT
-  TRY_CAST("Anno di imposta" AS INTEGER) AS anno_imposta,
-  "Codice catastale" AS codice_catastale,
-  "Codice Istat Comune" AS codice_istat_comune,
-  "Denominazione Comune" AS comune,
-  "Sigla Provincia" AS sigla_provincia,
-  "Regione" AS regione,
-  TRY_CAST("Numero contribuenti" AS BIGINT) AS numero_contribuenti
+  cast_int("Anno di imposta") AS anno_imposta,
+  normalize_string("Codice catastale") AS codice_catastale,
+  normalize_string("Codice Istat Comune") AS codice_istat_comune,
+  normalize_string("Denominazione Comune") AS comune,
+  normalize_string("Sigla Provincia") AS sigla_provincia,
+  normalize_string("Regione") AS regione,
+  cast_bigint("Numero contribuenti") AS numero_contribuenti
 FROM raw_input
diff --git a/smoke/local_file_csv/sql/clean.sql b/smoke/local_file_csv/sql/clean.sql
index 89980c86..6c3f1167 100644
--- a/smoke/local_file_csv/sql/clean.sql
+++ b/smoke/local_file_csv/sql/clean.sql
@@ -1,9 +1,9 @@
 SELECT
-    CAST(anno AS INTEGER) AS anno,
+    cast_int(anno) AS anno,
     comune,
     provincia,
     regione,
-    CAST(codice_comune AS INTEGER) AS codice_comune,
+    cast_int(codice_comune) AS codice_comune,
     categoria,
-    CAST(valore AS DOUBLE) AS valore
+    cast_double(valore) AS valore
 FROM raw_input
diff --git a/tests/test_macros_sql.py b/tests/test_macros_sql.py
new file mode 100644
index 00000000..a76dad22
--- /dev/null
+++ b/tests/test_macros_sql.py
@@ -0,0 +1,381 @@
+"""Tests for the standard DuckDB macros in ``macros.sql``.
+
+Le macro sono caricate automaticamente dal layer CLEAN. Questi test
+verificano il comportamento di ogni macro direttamente via DuckDB,
+usando lo stesso meccanismo di caricamento del runtime.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from lab_connectors.duckdb import safe_connect
+
+_MACROS_PATH = Path(__file__).parent.parent / "toolkit" / "sql" / "macros.sql"
+_MACROS_SQL = _MACROS_PATH.read_text(encoding="utf-8")
+
+pytestmark = pytest.mark.pure_unit
+
+
+# ===========================================================================
+# Package data: le macro devono essere distribuibili via wheel
+# ===========================================================================
+
+
+class TestPackageData:
+    """Contratto: macros.sql e' incluso nel package toolkit (package data).
+
+    Fallisce se il wheel non include ``toolkit/sql/macros.sql``.
+    """
+
+    def test_macros_file_exists_at_source(self) -> None:
+        """Il file esiste nel sorgente (sviluppo / editable install)."""
+        assert _MACROS_PATH.exists(), (
+            f"macros.sql non trovato in {_MACROS_PATH}. "
+            "Serve: pip install -e . per editable install"
+        )
+
+    def test_macros_file_accessible_via_importlib(self) -> None:
+        """Verifica che macros.sql sia accessibile come resource package.
+
+        Questo e' il path che usa il runtime se il toolkit e' installato.
+        """
+        try:
+            from importlib.resources import files as resources_files
+        except ImportError:
+            resources_files = None  # Python <3.9 compat
+
+        if resources_files is not None:
+            try:
+                pkg_files = resources_files("toolkit")
+                macros = pkg_files / "sql" / "macros.sql"
+                assert macros.exists(), (
+                    "importlib.resources non trova macros.sql in toolkit/sql/. "
+                    "Verifica [tool.setuptools.package_data] in pyproject.toml: "
+                    "toolkit = ['sql/*.sql']"
+                )
+            except (ModuleNotFoundError, TypeError):
+                pass  # toolkit non installato come package (es. test sviluppo)
+
+
+# ===========================================================================
+# normalize_italian_number
+# ===========================================================================
+
+
+class TestNormalizeItalianNumber:
+    """Contratto: converte numero formato italiano (1.234,56 → 1234.56)."""
+
+    def test_simple_integer(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_number('1234')").fetchone()
+        assert result[0] == 1234.0
+
+    def test_thousands_and_comma(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_number('1.234,56')").fetchone()
+        assert result[0] == 1234.56
+
+    def test_comma_decimal_only(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_number('1234,56')").fetchone()
+        assert result[0] == 1234.56
+
+    def test_large_thousands(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_number('12.345.678')").fetchone()
+        assert result[0] == 12345678.0
+
+    def test_null_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_number(NULL)").fetchone()
+        assert result[0] is None
+
+    def test_non_numeric_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_number('abc')").fetchone()
+        assert result[0] is None
+
+    def test_empty_string_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_number('')").fetchone()
+        assert result[0] is None
+
+
+# ===========================================================================
+# normalize_italian_integer
+# ===========================================================================
+
+
+class TestNormalizeItalianInteger:
+    """Contratto: come normalize_italian_number ma restituisce INTEGER.
+
+    DuckDB CAST(DOUBLE AS INTEGER) arrotonda (non tronca).
+    """
+
+    def test_simple_integer(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_integer('1234')").fetchone()
+        assert result[0] == 1234
+
+    def test_with_thousands(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_integer('1.234')").fetchone()
+        assert result[0] == 1234
+
+    def test_comma_rounds_down(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_integer('5.432,10')").fetchone()
+        assert result[0] == 5432
+
+    def test_comma_rounds_up(self) -> None:
+        """CAST(DOUBLE AS INTEGER) arrotonda all'intero piu' vicino."""
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_integer('5.432,90')").fetchone()
+        assert result[0] == 5433
+
+    def test_null_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_italian_integer(NULL)").fetchone()
+        assert result[0] is None
+
+
+# ===========================================================================
+# decode_flag
+# ===========================================================================
+
+
+class TestDecodeFlag:
+    """Contratto: decode_flag(val, yes_value) → BOOLEAN."""
+
+    def test_yes_matches(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT decode_flag('X', 'X')").fetchone()
+        assert result[0] is True
+
+    def test_no_mismatch(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT decode_flag('', 'X')").fetchone()
+        assert result[0] is False
+
+    def test_different_yes_value(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT decode_flag('S', 'S')").fetchone()
+        assert result[0] is True
+
+    def test_case_sensitive_default(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT decode_flag('x', 'X')").fetchone()
+        assert result[0] is False
+
+    def test_trim_applied(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT decode_flag(' X ', 'X')").fetchone()
+        assert result[0] is True
+
+    def test_null_val_returns_false(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT decode_flag(NULL, 'X')").fetchone()
+        assert result[0] is False
+
+
+# ===========================================================================
+# normalize_string
+# ===========================================================================
+
+
+class TestNormalizeString:
+    """Contratto: TRIM + stringa vuota → NULL."""
+
+    def test_trim(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_string('  hello  ')").fetchone()
+        assert result[0] == "hello"
+
+    def test_empty_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_string('')").fetchone()
+        assert result[0] is None
+
+    def test_whitespace_only_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_string('   ')").fetchone()
+        assert result[0] is None
+
+    def test_null_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT normalize_string(NULL)").fetchone()
+        assert result[0] is None
+
+
+# ===========================================================================
+# cast_int
+# ===========================================================================
+
+
+class TestCastInt:
+    """Contratto: TRY_CAST(val AS INTEGER)."""
+
+    def test_integer_string(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_int('123')").fetchone()
+        assert result[0] == 123
+
+    def test_integer_value(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_int(456)").fetchone()
+        assert result[0] == 456
+
+    def test_non_numeric_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_int('abc')").fetchone()
+        assert result[0] is None
+
+    def test_null_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_int(NULL)").fetchone()
+        assert result[0] is None
+
+
+# ===========================================================================
+# cast_double
+# ===========================================================================
+
+
+class TestCastDouble:
+    """Contratto: TRY_CAST(val AS DOUBLE)."""
+
+    def test_integer_string(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_double('123')").fetchone()
+        assert result[0] == 123.0
+
+    def test_decimal_string(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_double('123.45')").fetchone()
+        assert result[0] == 123.45
+
+    def test_non_numeric_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_double('abc')").fetchone()
+        assert result[0] is None
+
+    def test_null_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT cast_double(NULL)").fetchone()
+        assert result[0] is None
+
+
+# ===========================================================================
+# remove_dot_thousands
+# ===========================================================================
+
+
+class TestRemoveDotThousands:
+    """Contratto: rimuove punti migliaia da interi SOLO.
+
+    ATTENZIONE: rimuove TUTTI i punti. Non usare su numeri con
+    separatore decimale standard (usa cast_double) o italiano
+    (usa normalize_italian_number).
+    """
+
+    def test_simple_thousands(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT remove_dot_thousands('1.234')").fetchone()
+        assert result[0] == 1234.0
+
+    def test_multiple_thousands(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT remove_dot_thousands('1.234.567')").fetchone()
+        assert result[0] == 1234567.0
+
+    def test_standard_decimal_gets_mangled(self) -> None:
+        """Precondizione: non usare su numeri con punto decimale standard
+        — il punto viene rimosso e il risultato e' sbagliato.
+        Per questi usa cast_double o normalize_italian_number."""
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            # 1234.56 → rimuove TUTTI i punti → 123456.0 (volutamente SBAGLIATO)
+            result = con.execute("SELECT remove_dot_thousands('1234.56')").fetchone()
+        assert result[0] == 123456.0
+
+    def test_plain_integer(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT remove_dot_thousands('1234')").fetchone()
+        assert result[0] == 1234.0
+
+    def test_null_returns_null(self) -> None:
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            result = con.execute("SELECT remove_dot_thousands(NULL)").fetchone()
+        assert result[0] is None
+
+
+# ===========================================================================
+# Integration: macro usate in una query realistica
+# ===========================================================================
+
+
+class TestMacroIntegration:
+    """Contratto: le macro funzionano insieme in una query realistica."""
+
+    def test_clean_sql_pattern(self) -> None:
+        """Simula un clean.sql che usa più macro insieme."""
+        with safe_connect() as con:
+            con.execute(_MACROS_SQL)
+            # Simula la vista raw_input con dati di esempio
+            con.execute("""
+                CREATE TABLE raw_input AS SELECT * FROM (VALUES
+                    ('1', '  Mario  ', 'X', '1.234,56'),
+                    ('2', '  Luisa  ', '', '9.876,50'),
+                    ('3', '  Carla  ', NULL, NULL)
+                ) AS t(prog, nome, flag, importo)
+            """)
+            result = con.execute("""
+                SELECT
+                    cast_int(prog) AS progressivo,
+                    normalize_string(nome) AS nome,
+                    decode_flag(flag, 'X') AS flag_attivo,
+                    normalize_italian_number(importo) AS importo
+                FROM raw_input
+                ORDER BY progressivo
+            """).fetchall()
+        assert len(result) == 3
+        assert result[0] == (1, "Mario", True, 1234.56)
+        assert result[1] == (2, "Luisa", False, 9876.50)
+        assert result[2] == (3, "Carla", False, None)
diff --git a/tests/test_scaffold_clean.py b/tests/test_scaffold_clean.py
index 79b3cee3..9c8bc3d0 100644
--- a/tests/test_scaffold_clean.py
+++ b/tests/test_scaffold_clean.py
@@ -26,49 +26,49 @@
 
 
 class TestSelectExpr:
-    """pure_unit: _select_expr sceglie TRIM / REPLACE / TRY_CAST per tipo."""
+    """pure_unit: _select_expr usa le macro standard (normalize_string, cast_bigint, cast_double)."""
 
     @pytest.mark.pure_unit
-    def test_varchar_gets_trim(self) -> None:
-        """VARCHAR columns use TRIM instead of unnecessary TRY_CAST."""
+    def test_varchar_gets_normalize_string(self) -> None:
+        """VARCHAR columns use normalize_string macro."""
         result = _select_expr("Nome", "VARCHAR", "nome")
-        assert result == 'trim(CAST("Nome" AS VARCHAR)) AS nome'
+        assert result == 'normalize_string("Nome") AS nome'
 
     @pytest.mark.pure_unit
-    def test_integer_gets_try_cast(self) -> None:
-        """Integer columns get TRY_CAST to BIGINT."""
+    def test_integer_gets_cast_bigint(self) -> None:
+        """Integer columns get cast_bigint macro."""
         result = _select_expr("Anno", "BIGINT", "anno")
-        assert result == 'TRY_CAST("Anno" AS BIGINT) AS anno'
+        assert result == 'cast_bigint("Anno") AS anno'
 
     @pytest.mark.pure_unit
-    def test_double_gets_try_cast(self) -> None:
-        """Double columns get TRY_CAST to DOUBLE."""
+    def test_double_gets_cast_double(self) -> None:
+        """Double columns get cast_double macro."""
         result = _select_expr("Valore", "DOUBLE", "valore")
-        assert result == 'TRY_CAST("Valore" AS DOUBLE) AS valore'
+        assert result == 'cast_double("Valore") AS valore'
 
     @pytest.mark.pure_unit
-    def test_double_gets_plain_try_cast(self) -> None:
-        """Double columns always get plain TRY_CAST (no REPLACE — handled by clean.read)."""
+    def test_double_gets_plain_cast_double(self) -> None:
+        """Double columns use cast_double (no REPLACE — handled by clean.read)."""
         result = _select_expr("Importo", "DOUBLE", "importo")
-        assert result == 'TRY_CAST("Importo" AS DOUBLE) AS importo'
+        assert result == 'cast_double("Importo") AS importo'
         assert "REPLACE" not in result
 
     @pytest.mark.pure_unit
-    def test_bigint_gets_plain_try_cast(self) -> None:
-        """BIGINT columns always get plain TRY_CAST."""
+    def test_bigint_gets_plain_cast_bigint(self) -> None:
+        """BIGINT columns use cast_bigint."""
         result = _select_expr("Anno", "BIGINT", "anno")
-        assert result == 'TRY_CAST("Anno" AS BIGINT) AS anno'
+        assert result == 'cast_bigint("Anno") AS anno'
         assert "REPLACE" not in result
 
     @pytest.mark.pure_unit
     def test_date_gets_try_cast(self) -> None:
-        """DATE columns get TRY_CAST."""
+        """DATE columns still use TRY_CAST (no standard macro)."""
         result = _select_expr("Data", "DATE", "data")
         assert 'TRY_CAST("Data" AS DATE)' in result
 
     @pytest.mark.pure_unit
     def test_boolean_gets_try_cast(self) -> None:
-        """BOOLEAN columns get TRY_CAST."""
+        """BOOLEAN columns still use TRY_CAST (no standard macro)."""
         result = _select_expr("Attivo", "BOOLEAN", "attivo")
         assert 'TRY_CAST("Attivo" AS BOOLEAN)' in result
 
@@ -270,7 +270,7 @@ class TestColumnsSpec:
 
     @pytest.mark.pure_unit
     def test_mixed_types(self) -> None:
-        """Mapping misto: VARCHAR → TRIM, numerici → TRY_CAST."""
+        """Mapping misto: VARCHAR → normalize_string, numerici → cast_bigint/cast_double."""
         profile: dict[str, Any] = {
             "mapping_suggestions": {
                 "Nome": {"type": "str"},
@@ -279,16 +279,16 @@ def test_mixed_types(self) -> None:
             },
         }
         exprs, spec = _columns_spec(profile, 2024)
-        assert 'trim(CAST("Nome" AS VARCHAR)) AS nome' in exprs
-        assert 'TRY_CAST("Anno" AS BIGINT) AS anno' in exprs
-        assert 'TRY_CAST("Valore" AS DOUBLE) AS valore' in exprs
+        assert 'normalize_string("Nome") AS nome' in exprs
+        assert 'cast_bigint("Anno") AS anno' in exprs
+        assert 'cast_double("Valore") AS valore' in exprs
         assert spec["Nome"] == "VARCHAR"
         assert spec["Anno"] == "BIGINT"
         assert spec["Valore"] == "DOUBLE"
 
     @pytest.mark.pure_unit
     def test_comma_decimal(self) -> None:
-        """Con decimal_suggested=',', colonne DOUBLE usano TRY_CAST normale
+        """Con decimal_suggested=',', colonne DOUBLE usano cast_double
         (REPLACE non serve: clean.read.decimal gestito da DuckDB)."""
         profile: dict[str, Any] = {
             "decimal_suggested": ",",
@@ -299,20 +299,20 @@ def test_comma_decimal(self) -> None:
         }
         exprs, _ = _columns_spec(profile, 2024)
         joined = "\n".join(exprs)
-        assert 'trim(CAST("Nome" AS VARCHAR)) AS nome' in joined
-        assert 'TRY_CAST("Importo" AS DOUBLE)' in joined
+        assert 'normalize_string("Nome") AS nome' in joined
+        assert 'cast_double("Importo")' in joined
         assert "REPLACE" not in joined
 
     @pytest.mark.pure_unit
     def test_no_mapping_fallback(self) -> None:
-        """Senza mapping: CAST(... AS VARCHAR) + TRIM per safety su tipi misti."""
+        """Senza mapping: normalize_string per safety su tipi misti."""
         profile: dict[str, Any] = {
             "mapping_suggestions": {},
             "columns_raw": ["Col1", "Col2"],
         }
         exprs, spec = _columns_spec(profile, 2024)
-        assert 'trim(CAST("Col1" AS VARCHAR)) AS col1' in exprs
-        assert 'trim(CAST("Col2" AS VARCHAR)) AS col2' in exprs
+        assert 'normalize_string("Col1") AS col1' in exprs
+        assert 'normalize_string("Col2") AS col2' in exprs
         assert spec == {"Col1": "VARCHAR", "Col2": "VARCHAR"}
 
     @pytest.mark.pure_unit
@@ -344,8 +344,8 @@ def test_basic_without_anno_column(self) -> None:
         assert "{year}::INTEGER AS anno" in sql
         assert "FROM raw_input" in sql
         assert "WHERE" not in sql
-        assert 'trim(CAST("Nome" AS VARCHAR))' in sql
-        assert 'TRY_CAST("Valore"' in sql
+        assert 'normalize_string("Nome")' in sql
+        assert 'cast_double("Valore")' in sql
 
     @pytest.mark.pure_unit
     def test_with_real_anno_column_adds_where(self) -> None:
@@ -360,9 +360,9 @@ def test_with_real_anno_column_adds_where(self) -> None:
         }
         sql = generate_clean_sql(profile, "test_dataset", 2024)
         assert "{year}::INTEGER" not in sql  # non injectato
-        assert 'TRY_CAST("Anno" AS BIGINT) AS anno' in sql
+        assert 'cast_bigint("Anno") AS anno' in sql
         assert 'WHERE try_cast("Anno" AS INTEGER) IS NOT NULL' in sql
-        assert 'trim(CAST("Regione" AS VARCHAR))' in sql
+        assert 'normalize_string("Regione")' in sql
 
     @pytest.mark.pure_unit
     def test_with_anno_di_imposta_adds_where(self) -> None:
@@ -390,7 +390,9 @@ def test_comma_decimal_in_header(self) -> None:
             },
         }
         sql = generate_clean_sql(profile, "test", 2024)
-        assert "REPLACE" not in sql
+        # REPLACE non deve apparire nel corpo SQL (solo nei commenti macro)
+        body = "\n".join(line for line in sql.split("\n") if not line.strip().startswith("--"))
+        assert "REPLACE" not in body
         assert "Decimal: ," in sql  # nel commento header
         assert "Encoding: utf-8" in sql
         assert "Delimiter: ;" in sql
@@ -459,8 +461,8 @@ class TestGenerateCleanSqlIntegration:
     """pure_unit: generate_clean_sql produce clean.sql coerente."""
 
     @pytest.mark.pure_unit
-    def test_varchar_gets_trim(self) -> None:
-        """generate_clean_sql applica TRIM alle colonne VARCHAR."""
+    def test_varchar_gets_normalize_string(self) -> None:
+        """generate_clean_sql usa normalize_string per colonne VARCHAR."""
         from toolkit.scaffold.clean import generate_clean_sql
 
         profile: dict[str, Any] = {
@@ -471,9 +473,9 @@ def test_varchar_gets_trim(self) -> None:
             },
         }
         sql = generate_clean_sql(profile, "candidate", 2024)
-        assert 'trim(CAST("nome" AS VARCHAR))' in sql
-        assert 'trim(CAST("categoria" AS VARCHAR))' in sql
-        assert 'TRY_CAST("valore" AS DOUBLE)' in sql
+        assert 'normalize_string("nome")' in sql
+        assert 'normalize_string("categoria")' in sql
+        assert 'cast_double("valore")' in sql
 
 
 # ---------------------------------------------------------------------------
@@ -575,7 +577,7 @@ def test_mixed_types(self) -> None:
 
     @pytest.mark.pure_unit
     def test_generate_clean_sql_from_datastore(self) -> None:
-        """Profile da DataStore produce clean.sql con colonne mappate."""
+        """Profile da DataStore produce clean.sql con macro standard."""
         from toolkit.scaffold.clean import generate_clean_sql, profile_from_datastore
 
         fields = [
@@ -585,7 +587,7 @@ def test_generate_clean_sql_from_datastore(self) -> None:
         ]
         profile = profile_from_datastore(fields)
         sql = generate_clean_sql(profile, "test_dataset", 2024)
-        assert 'trim(CAST("nome" AS VARCHAR)) AS nome' in sql
-        assert 'TRY_CAST("valore" AS DOUBLE) AS valore' in sql
+        assert 'normalize_string("nome") AS nome' in sql
+        assert 'cast_double("valore") AS valore' in sql
         assert 'TRY_CAST("data" AS DATE) AS data' in sql
         assert "{year}::INTEGER AS anno" in sql
diff --git a/tests/test_scout_infer.py b/tests/test_scout_infer.py
index 0b487487..779d0588 100644
--- a/tests/test_scout_infer.py
+++ b/tests/test_scout_infer.py
@@ -196,8 +196,8 @@ def test_with_casts(self) -> None:
             },
         }
         sql = generate_clean_sql(profile, "candidate", 2024)
-        assert 'TRY_CAST("valore" AS DOUBLE)' in sql
-        assert 'TRY_CAST("anno" AS BIGINT)' in sql
+        assert 'cast_double("valore")' in sql
+        assert 'cast_bigint("anno")' in sql
         assert '"nome"' in sql
         assert "FROM raw_input" in sql
 
@@ -211,7 +211,9 @@ def test_no_mapping(self) -> None:
             },
         }
         sql = generate_clean_sql(profile, "candidate", 2024)
-        assert "TRY_CAST" not in sql
+        # TRY_CAST non deve apparire nel corpo SQL (solo nei commenti macro)
+        body = "\n".join(line for line in sql.split("\n") if not line.strip().startswith("--"))
+        assert "TRY_CAST" not in body
         assert '"a"' in sql
         assert '"b"' in sql
         assert '"c"' in sql
@@ -329,8 +331,8 @@ def test_clean_sql_with_shorthand_types(self) -> None:
             },
         }
         sql = generate_clean_sql(profile, "candidate", 2024)
-        assert 'TRY_CAST("eta" AS BIGINT)' in sql, f"int type not recognized: {sql}"
-        assert 'TRY_CAST("reddito" AS DOUBLE)' in sql
+        assert 'cast_bigint("eta")' in sql, f"int type not recognized: {sql}"
+        assert 'cast_double("reddito")' in sql
 
 
 # ── Routing SPARQL in probe_url_routed ───────────────────────────────────────
diff --git a/toolkit/clean/sql_execute.py b/toolkit/clean/sql_execute.py
index 2f2318c8..167aaaf3 100644
--- a/toolkit/clean/sql_execute.py
+++ b/toolkit/clean/sql_execute.py
@@ -17,6 +17,30 @@
 from toolkit.core.layer_profile import profile_relation
 from toolkit.core.sql_utils import sql_path
 
+_MACROS_PATH = Path(__file__).parent.parent / "sql" / "macros.sql"
+
+
+def _load_standard_macros(con, logger) -> None:
+    """Load the standard toolkit DuckDB macros into the connection.
+
+    Reads ``macros.sql`` from the toolkit package and executes it.
+    All macros use ``CREATE OR REPLACE`` so repeated calls are safe.
+
+    Raises:
+        FileNotFoundError: se ``macros.sql`` non esiste (toolkit
+            installato senza package data o sviluppo incompleto).
+    """
+    if not _MACROS_PATH.exists():
+        raise FileNotFoundError(
+            f"Standard macros not found at {_MACROS_PATH}. "
+            "Ensure toolkit is installed with sql/*.sql package data "
+            "(pip install -e . rebuilds the package)."
+        )
+    macros_sql = _MACROS_PATH.read_text(encoding="utf-8")
+    con.execute(macros_sql)
+    if logger:
+        logger.debug("Loaded %d bytes of standard macros from %s", len(macros_sql), _MACROS_PATH)
+
 
 def _normalize_output_profile(output_profile: dict[str, Any] | int) -> dict[str, Any]:
     if isinstance(output_profile, dict):
@@ -42,10 +66,14 @@ def _run_sql(
     If ``sample_rows`` is set, appends ``LIMIT N`` to the SQL query per
     DuckDB syntax (``SELECT * FROM ({query}) AS _smoke LIMIT N``).
 
+    Standard toolkit macros (``normalize_italian_number``, ``decode_flag``,
+    etc.) are automatically loaded at the start of the connection.
+
     Returns:
         tuple of (source, params_used, output_profile)
     """
     with safe_connect() as con:
+        _load_standard_macros(con, logger)
         read_info = read_raw_to_relation(con, input_files, read_cfg, read_mode, logger)
         if sample_rows is not None:
             # Strip trailing semicolons: clean.sql spesso termina con ;
diff --git a/toolkit/cli/sql_dry_run.py b/toolkit/cli/sql_dry_run.py
index 95a7c8c0..6e051df7 100644
--- a/toolkit/cli/sql_dry_run.py
+++ b/toolkit/cli/sql_dry_run.py
@@ -7,6 +7,7 @@
 from lab_connectors.duckdb import safe_connect
 
 from toolkit.clean.run import load_clean_sql
+from toolkit.clean.sql_execute import _load_standard_macros
 from toolkit.core.config import ensure_dict
 from toolkit.core.paths import resolve_sql_path as _resolve_mart_sql_path
 from toolkit.core.support import flatten_support_template_ctx, resolve_support_payloads
@@ -186,6 +187,7 @@ def validate_sql_dry_run(cfg, *, year: int, layers: list[str], dry_run: bool = F
         return
 
     with safe_connect() as con:
+        _load_standard_macros(con, logger=None)
         if cfg.clean.sql:
             _build_clean_preview(
                 cfg, year=year, con=con, support_cfg=ensure_dict(cfg.support), dry_run=dry_run
diff --git a/toolkit/scaffold/clean.py b/toolkit/scaffold/clean.py
index 0ac10a23..4f680b76 100644
--- a/toolkit/scaffold/clean.py
+++ b/toolkit/scaffold/clean.py
@@ -215,16 +215,27 @@ def _select_expr(
     sql_type: str,
     out_name: str,
 ) -> str:
-    """Build a SELECT expression for one column with smart transformations.
+    """Build a SELECT expression for one column using standard macros.
 
-    - VARCHAR columns: TRIM with CAST AS VARCHAR (safe for columns sniffed as BIGINT)
-      Comma-decimal numbers are handled by clean.read.decimal in dataset.yml
-      (see propose_clean_read()), not by REPLACE in SQL — DuckDB's read_csv
-      native decimal support is more reliable and avoids DOUBLE→STRING round-trips.
+    Genera espressioni che usano le macro DuckDB standard del toolkit
+    (``normalize_string``, ``cast_bigint``, ``cast_double``) invece di
+    ``TRY_CAST``/``TRIM`` espliciti. Le macro sono caricate
+    automaticamente dal layer CLEAN.
+
+    Per DATE e BOOLEAN continua a usare TRY_CAST diretto (nessuna macro
+    standard per questi tipi).
     """
     if sql_type == "VARCHAR":
-        return f'trim(CAST("{raw_col}" AS VARCHAR)) AS {out_name}'
+        return f'normalize_string("{raw_col}") AS {out_name}'
+
+    macro_map = {
+        "BIGINT": "cast_bigint",
+        "DOUBLE": "cast_double",
+    }
+    if sql_type in macro_map:
+        return f'{macro_map[sql_type]}("{raw_col}") AS {out_name}'
 
+    # DATE, BOOLEAN e altri: TRY_CAST diretto
     return f'TRY_CAST("{raw_col}" AS {sql_type}) AS {out_name}'
 
 
@@ -233,13 +244,11 @@ def _columns_spec(profile: dict[str, Any], year: int) -> tuple[list[str], dict[s
     mapping = profile.get("mapping_suggestions", {})
     if not mapping:
         # Fallback to columns_raw if mapping is empty (e.g. Excel profiling,
-        # or profiling that failed to infer types). Use trim(CAST(... AS VARCHAR))
-        # to safely handle both text and numeric columns.
+        # or profiling that failed to infer types). Use normalize_string()
+        # (standard macro) to handle TEXT columns safely.
         columns_raw = profile.get("columns_raw", [])
         if columns_raw:
-            select_exprs = [
-                f'trim(CAST("{c}" AS VARCHAR)) AS {_snake_case(c)}' for c in columns_raw
-            ]
+            select_exprs = [f'normalize_string("{c}") AS {_snake_case(c)}' for c in columns_raw]
             columns_spec = {c: "VARCHAR" for c in columns_raw}
             return select_exprs, columns_spec
         return ["*"], {}
@@ -362,6 +371,14 @@ def generate_clean_sql(
         "-- CSV parsing options (delim, columns, encoding, header, skip) "
         "belong in clean.read in dataset.yml."
     )
+    header_parts.append("--")
+    header_parts.append("-- Macro SQL disponibili (caricate automaticamente):")
+    header_parts.append("--   normalize_string(col)       → TRIM + NULL per stringhe vuote")
+    header_parts.append("--   cast_int(col)               → TRY_CAST(col AS INTEGER)")
+    header_parts.append("--   cast_double(col)            → TRY_CAST(col AS DOUBLE)")
+    header_parts.append("--   normalize_italian_number(c) → REPLACE+REPLACE+TRY_CAST")
+    header_parts.append("--   normalize_italian_integer(c)→ REPLACE+REPLACE+TRY_CAST(INT)")
+    header_parts.append("--   decode_flag(col, 'X')       → CASE WHEN col='X' THEN TRUE ELSE FALSE")
     header_parts.append("-- Pass --run to execute raw and clean in one step, or run:")
     header_parts.append("--   toolkit run clean -c dataset.yml")
     header_parts.append("")
diff --git a/toolkit/sql/macros.sql b/toolkit/sql/macros.sql
new file mode 100644
index 00000000..7106e1dd
--- /dev/null
+++ b/toolkit/sql/macros.sql
@@ -0,0 +1,64 @@
+-- =============================================================================
+-- Standard Macros SQL per il layer CLEAN del toolkit DataCivicLab
+--
+-- Queste macro DuckDB sono automaticamente caricate in ogni esecuzione
+-- clean.sql. Possono essere usate liberamente nei file SQL dei dataset
+-- senza dover riscrivere il boilerplate.
+--
+-- Uso: SELECT normalize_italian_number("colonna") AS importo FROM raw_input
+--
+-- Tutte le macro sono CREATE OR REPLACE — sicure da eseguire più volte.
+-- =============================================================================
+
+-- ── normalize_italian_number ───────────────────────────────────────────────
+-- Converte un numero in formato italiano (1.234,56 → 1234.56).
+-- Rimuove punti migliaia, converte virgola decimale in punto.
+-- TRY_CAST restituisce NULL se la conversione fallisce.
+CREATE OR REPLACE MACRO normalize_italian_number(val) AS
+   TRY_CAST(REPLACE(REPLACE(val::VARCHAR, '.', ''), ',', '.') AS DOUBLE);
+
+-- ── normalize_italian_integer ──────────────────────────────────────────────
+-- Come normalize_italian_number ma restituisce INTEGER.
+-- DuckDB CAST(DOUBLE AS INTEGER) arrotonda (5432.90 → 5433).
+-- "1.234" → 1234, "5.432,10" → 5432, "5.432,90" → 5433.
+CREATE OR REPLACE MACRO normalize_italian_integer(val) AS
+   TRY_CAST(REPLACE(REPLACE(val::VARCHAR, '.', ''), ',', '.') AS INTEGER);
+
+-- ── decode_flag ────────────────────────────────────────────────────────────
+-- Decodifica un flag testuale in BOOLEAN.
+-- decode_flag("ETS", 'X') → TRUE se il valore è 'X', FALSE altrimenti.
+-- Lista di valori ammessi: passare come secondo argomento il valore che
+-- rappresenta TRUE (es. 'X', 'S', '1', 'TRUE').
+CREATE OR REPLACE MACRO decode_flag(val, yes_value) AS
+   CASE WHEN TRIM(val::VARCHAR) = yes_value::VARCHAR THEN TRUE ELSE FALSE END;
+
+-- ── normalize_string ───────────────────────────────────────────────────────
+-- Normalizza una stringa: TRIM + converte stringhe vuote in NULL.
+CREATE OR REPLACE MACRO normalize_string(val) AS
+   NULLIF(TRIM(val::VARCHAR), '');
+
+-- ── cast_int ───────────────────────────────────────────────────────────────
+-- TRY_CAST a INTEGER (32-bit). Per numeri grandi usa cast_bigint.
+CREATE OR REPLACE MACRO cast_int(val) AS
+   TRY_CAST(val AS INTEGER);
+
+-- ── cast_bigint ────────────────────────────────────────────────────────────
+-- TRY_CAST a BIGINT (64-bit). Usato dallo scaffold per colonne numeriche.
+CREATE OR REPLACE MACRO cast_bigint(val) AS
+   TRY_CAST(val AS BIGINT);
+
+-- ── cast_double ────────────────────────────────────────────────────────────
+-- TRY_CAST a DOUBLE con gestione NULL sicura.
+CREATE OR REPLACE MACRO cast_double(val) AS
+   TRY_CAST(val AS DOUBLE);
+
+-- ── remove_dot_thousands ───────────────────────────────────────────────────
+-- Rimuove punti migliaia da numeri interi (senza virgola decimale).
+-- ATTENZIONE: rimuove TUTTI i punti, incluso l'eventuale separatore
+-- decimale standard. Usa SOLO su interi con punti migliaia.
+-- Per numeri con decimali (italiani o standard) usa:
+--   normalize_italian_number()  — formato italiano (virgola)
+--   cast_double()               — formato internazionale (punto)
+-- "1.234" → 1234.0, "1.234.567" → 1234567.0
+CREATE OR REPLACE MACRO remove_dot_thousands(val) AS
+   TRY_CAST(REPLACE(val::VARCHAR, '.', '') AS DOUBLE);