diff --git a/clickhouse/scripts/045_cost_model_report_account.sh b/clickhouse/scripts/045_cost_model_report_account.sh
new file mode 100755
index 00000000..ea2e49b6
--- /dev/null
+++ b/clickhouse/scripts/045_cost_model_report_account.sh
@@ -0,0 +1,133 @@
+#!/bin/sh
+set -eu
+
+# Add the settlement report's internal account dimension (Adyen CSV `Merchant Account`) to the
+# cost model identity. This must rebuild the ReplacingMergeTree tables because ClickHouse cannot
+# ALTER an existing sorting key; adding the column alone would still let FINAL collapse distinct
+# report-account buckets.
+
+CLICKHOUSE_DATABASE="${CLICKHOUSE_DATABASE:-default}"
+CLICKHOUSE_USER="${CLICKHOUSE_USER:-default}"
+CLICKHOUSE_PASSWORD="${CLICKHOUSE_PASSWORD:-}"
+
+auth_args="--database=${CLICKHOUSE_DATABASE} --user=${CLICKHOUSE_USER}"
+if [ -n "${CLICKHOUSE_PASSWORD}" ]; then
+  auth_args="${auth_args} --password=${CLICKHOUSE_PASSWORD}"
+fi
+
+clickhouse-client ${auth_args} --multiquery <<SQL
+ALTER TABLE cost_daily_stats ADD COLUMN IF NOT EXISTS report_account String DEFAULT '' AFTER account;
+ALTER TABLE cost_fee_model ADD COLUMN IF NOT EXISTS report_account String DEFAULT '' AFTER account;
+
+DROP TABLE IF EXISTS cost_daily_stats__report_account_new;
+DROP TABLE IF EXISTS cost_daily_stats__report_account_backup;
+
+CREATE TABLE cost_daily_stats__report_account_new (
+    connector        LowCardinality(String),
+    account          String,
+    report_account   String DEFAULT '',
+    merchant_id      String,
+    txn_date         Date,
+    ingestion_id     String DEFAULT '',
+    card_network     LowCardinality(String),
+    variant          String,
+    funding          LowCardinality(String),
+    issuer_country   LowCardinality(String),
+    currency         LowCardinality(String),
+    ic_category      String,
+    interchange_bps  String DEFAULT '',
+    channel          LowCardinality(String) DEFAULT '',
+    band             LowCardinality(String) DEFAULT '',
+    fit_bucket       Int32 DEFAULT 0,
+    n                UInt64,
+    sx               Float64,
+    sy               Float64,
+    sxx              Float64,
+    sxy              Float64,
+    syy              Float64,
+    su               Float64,
+    suu              Float64,
+    suy              Float64,
+    suuy             Float64,
+    syyuu            Float64,
+    sample_x         Array(Float64) DEFAULT [],
+    sample_y         Array(Float64) DEFAULT [],
+    ingested_at      DateTime DEFAULT now()
+) ENGINE = ReplacingMergeTree(ingested_at)
+PARTITION BY toYYYYMM(txn_date)
+ORDER BY (connector, account, merchant_id, txn_date,
+          report_account, card_network, variant, funding, issuer_country, currency, ic_category,
+          interchange_bps, channel, band, fit_bucket)
+TTL txn_date + INTERVAL 400 DAY;
+
+INSERT INTO cost_daily_stats__report_account_new
+SELECT
+    connector, account, report_account, merchant_id, txn_date, ingestion_id,
+    card_network, variant, funding, issuer_country, currency, ic_category, interchange_bps,
+    channel, band, fit_bucket, n, sx, sy, sxx, sxy, syy, su, suu, suy, suuy, syyuu,
+    sample_x, sample_y, ingested_at
+FROM cost_daily_stats;
+
+RENAME TABLE
+    cost_daily_stats TO cost_daily_stats__report_account_backup,
+    cost_daily_stats__report_account_new TO cost_daily_stats;
+
+DROP TABLE cost_daily_stats__report_account_backup;
+
+DROP TABLE IF EXISTS cost_fee_model__report_account_new;
+DROP TABLE IF EXISTS cost_fee_model__report_account_backup;
+
+CREATE TABLE cost_fee_model__report_account_new (
+    report_date      Date,
+    connector        LowCardinality(String),
+    account          String,
+    report_account   String DEFAULT '',
+    merchant_id      String,
+    card_network     LowCardinality(String),
+    variant          String,
+    funding          LowCardinality(String),
+    issuer_country   LowCardinality(String),
+    currency         LowCardinality(String),
+    ic_category      String,
+    interchange_bps  String DEFAULT '',
+    segment_idx      UInt16 DEFAULT 0,
+    amount_lo        Float64 DEFAULT 0,
+    amount_hi        Float64 DEFAULT 0,
+    pct_bps          Float64,
+    fixed            Float64,
+    n                UInt64,
+    gross_sum        Float64 DEFAULT 0,
+    bps_rmse         Float64,
+    grade_bps        Float64 DEFAULT 0,
+    pct_ci95_bps     Float64 DEFAULT 0,
+    crossover_amount Float64 DEFAULT 0,
+    prop_bps         Float64 DEFAULT 0,
+    fix_abs          Float64 DEFAULT 0,
+    fix_bps          Float64 DEFAULT 0,
+    below_gross_frac Float64 DEFAULT 0,
+    fan_frac         Float64 DEFAULT 0,
+    fan_money_bps    Float64 DEFAULT 0,
+    r2               Float64,
+    verdict          Enum8('GOOD' = 1, 'NON_LINEAR' = 2, 'THIN' = 3, 'FAN' = 4),
+    fitted_at        DateTime DEFAULT now()
+) ENGINE = ReplacingMergeTree(fitted_at)
+PARTITION BY toYYYYMM(report_date)
+ORDER BY (connector, account, merchant_id, report_date,
+          report_account, card_network, variant, funding, issuer_country, currency, ic_category,
+          interchange_bps, segment_idx);
+
+INSERT INTO cost_fee_model__report_account_new
+SELECT
+    report_date, connector, account, report_account, merchant_id,
+    card_network, variant, funding, issuer_country, currency, ic_category, interchange_bps,
+    segment_idx, amount_lo, amount_hi, pct_bps, fixed, n, gross_sum, bps_rmse,
+    grade_bps, pct_ci95_bps, crossover_amount, prop_bps, fix_abs, fix_bps, below_gross_frac,
+    fan_frac, fan_money_bps, r2, verdict, fitted_at
+FROM cost_fee_model;
+
+RENAME TABLE
+    cost_fee_model TO cost_fee_model__report_account_backup,
+    cost_fee_model__report_account_new TO cost_fee_model;
+
+DROP TABLE cost_fee_model__report_account_backup;
+SQL
diff --git a/src/cost_ingestion/blended.rs b/src/cost_ingestion/blended.rs
index 5c68d490..23890884 100644
--- a/src/cost_ingestion/blended.rs
+++ b/src/cost_ingestion/blended.rs
@@ -63,8 +63,21 @@ pub struct TopCluster {
     pub issuer_country: String,
     pub currency: String,
     pub ic_category: String,
+    pub interchange_bps: String,
+    pub segment_idx: u16,
+    pub amount_lo: f64,
+    pub amount_hi: f64,
     pub pct_bps: f64,
     pub fixed: f64,
+    pub grade_bps: f64,
+    pub pct_ci95_bps: f64,
+    pub crossover_amount: f64,
+    pub prop_bps: f64,
+    pub fix_abs: f64,
+    pub fix_bps: f64,
+    pub below_gross_frac: f64,
+    pub fan_frac: f64,
+    pub fan_money_bps: f64,
     /// Transaction count (so a small-ticket/high-txn segment stays visible next to GMV).
     pub n: u64,
     /// Settled GMV — the ranking weight.
@@ -98,13 +111,24 @@ pub struct ClusterScope<'a> {
 const TOP_CLUSTERS_SQL: &str = r#"
 SELECT
     connector, card_network, variant, funding, issuer_country, currency, ic_category,
+    interchange_bps, segment_idx, amount_lo, amount_hi,
     sum(pct_bps * gross_sum) / sum(gross_sum) AS blended_pct_bps,
     sum(fixed * gross_sum)   / sum(gross_sum) AS blended_fixed,
+    sum(grade_bps * gross_sum) / sum(gross_sum) AS blended_grade_bps,
+    max(pct_ci95_bps)        AS pct_ci95_bps,
+    max(crossover_amount)    AS crossover_amount,
+    sum(prop_bps * gross_sum) / sum(gross_sum) AS blended_prop_bps,
+    sum(fix_abs * gross_sum) / sum(gross_sum) AS blended_fix_abs,
+    sum(fix_bps * gross_sum) / sum(gross_sum) AS blended_fix_bps,
+    sum(below_gross_frac * gross_sum) / sum(gross_sum) AS blended_below_gross_frac,
+    max(fan_frac)            AS fan_frac,
+    max(fan_money_bps)       AS fan_money_bps,
     sum(n)                   AS txns,
     sum(gross_sum)           AS total_gross
 FROM __DB__.cost_fee_model FINAL
 WHERE verdict = 'GOOD' AND gross_sum > 0 AND merchant_id = {merchant_id:String}{snapshot_filter}
-GROUP BY connector, card_network, variant, funding, issuer_country, currency, ic_category
+GROUP BY connector, card_network, variant, funding, issuer_country, currency, ic_category,
+         interchange_bps, segment_idx, amount_lo, amount_hi
 ORDER BY total_gross DESC
 LIMIT {limit:UInt32}
 FORMAT TSV
@@ -196,7 +220,7 @@ pub async fn top_clusters(
     let mut out = Vec::new();
     for line in text.lines() {
         let f: Vec<&str> = line.split('\t').collect();
-        if f.len() < 11 {
+        if f.len() < 24 {
             continue;
         }
         out.push(TopCluster {
@@ -207,10 +231,23 @@ pub async fn top_clusters(
             issuer_country: f[4].trim().to_string(),
             currency: f[5].trim().to_string(),
             ic_category: f[6].trim().to_string(),
-            pct_bps: f[7].trim().parse().unwrap_or(0.0),
-            fixed: f[8].trim().parse().unwrap_or(0.0),
-            n: f[9].trim().parse().unwrap_or(0),
-            gross_sum: f[10].trim().parse().unwrap_or(0.0),
+            interchange_bps: f[7].trim().to_string(),
+            segment_idx: f[8].trim().parse().unwrap_or(0),
+            amount_lo: f[9].trim().parse().unwrap_or(0.0),
+            amount_hi: f[10].trim().parse().unwrap_or(0.0),
+            pct_bps: f[11].trim().parse().unwrap_or(0.0),
+            fixed: f[12].trim().parse().unwrap_or(0.0),
+            grade_bps: f[13].trim().parse().unwrap_or(0.0),
+            pct_ci95_bps: f[14].trim().parse().unwrap_or(0.0),
+            crossover_amount: f[15].trim().parse().unwrap_or(0.0),
+            prop_bps: f[16].trim().parse().unwrap_or(0.0),
+            fix_abs: f[17].trim().parse().unwrap_or(0.0),
+            fix_bps: f[18].trim().parse().unwrap_or(0.0),
+            below_gross_frac: f[19].trim().parse().unwrap_or(0.0),
+            fan_frac: f[20].trim().parse().unwrap_or(0.0),
+            fan_money_bps: f[21].trim().parse().unwrap_or(0.0),
+            n: f[22].trim().parse().unwrap_or(0),
+            gross_sum: f[23].trim().parse().unwrap_or(0.0),
         });
     }
     Ok(out)
diff --git a/src/cost_ingestion/connectors/adyen.rs b/src/cost_ingestion/connectors/adyen.rs
index b591c18d..86f4c07e 100644
--- a/src/cost_ingestion/connectors/adyen.rs
+++ b/src/cost_ingestion/connectors/adyen.rs
@@ -21,9 +21,10 @@ use crate::cost_ingestion::types::{
     ConnectorCreds, IngestError, ReportNotification, SettledFeeRow,
 };
 
-/// Adyen record types that actually carry settlement fees; everything else (Authorised,
-/// Received, Refused, …) has empty fee columns and would pollute the fit.
-const FEE_RECORD_TYPES: [&str; 2] = ["SentForSettle", "Settled"];
+/// Adyen record type that carries the final settled fee signal. PAR also includes
+/// `SentForSettle`, but `cluster_explorer.py` keeps one settled leg only so the same transaction
+/// does not affect the fee fit twice.
+const FEE_RECORD_TYPES: [&str; 1] = ["Settled"];
 
 const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(120);
 
@@ -179,6 +180,7 @@ impl SettlementReportSource for AdyenReportSource {
             scheme: usize,
             interchange: usize,
             icsf: usize,
+            merchant_account: Option<usize>,
             booking: Option<usize>,
             terminal: Option<usize>,
         }
@@ -200,6 +202,9 @@ impl SettlementReportSource for AdyenReportSource {
                     scheme: h.require("Scheme Fees (SC)")?,
                     interchange: h.require("Interchange (SC)")?,
                     icsf: h.require("ICSF details")?,
+                    // Optional in tests/older fixtures; present in Adyen accounting reports and
+                    // needed to match cluster_explorer.py's per-merchant-account split.
+                    merchant_account: h.index("Merchant Account"),
                     // Optional: used only for the ingested report's period; absent in older/test reports.
                     booking: h.index("Booking Date"),
                     // Optional: a terminal id marks in-person (POS) acceptance; absence ⇒ online (ecom).
@@ -230,14 +235,21 @@ impl SettlementReportSource for AdyenReportSource {
                 }
                 .to_string();
 
+                let (ic_category, interchange_bps) = ic_details(row.get(c.icsf));
+
                 Ok(Some(SettledFeeRow {
                     txn_ref: row.get(c.psp).to_string(),
+                    report_account: c
+                        .merchant_account
+                        .map(|i| row.get(i).trim().to_string())
+                        .unwrap_or_default(),
                     card_network: row.get(c.brand).to_lowercase(),
                     variant,
                     funding,
                     issuer_country: row.get(c.issuer).to_string(),
                     currency: row.get(c.ccy).to_string(),
-                    ic_category: ic_category(row.get(c.icsf)),
+                    ic_category,
+                    interchange_bps,
                     txn_date,
                     channel,
                     gross,
@@ -381,21 +393,52 @@ fn base64_encode(bytes: &[u8]) -> String {
     base64::engine::general_purpose::STANDARD.encode(bytes)
 }
 
-/// Pull the interchange category from the `ICSF details` JSON array: the element with `t=="ic"`
-/// carries the product category in `n`. `""` when absent (flat-fee methods) or unparsable.
-fn ic_category(raw: &str) -> String {
+/// Pull the interchange category and rate from the `ICSF details` JSON array: the element with
+/// `t=="ic"` carries the product category in `n` and, when present, the card-product rate in `bps`.
+/// Empty strings mean absent (flat-fee methods) or unparsable.
+fn ic_details(raw: &str) -> (String, String) {
     if raw.is_empty() {
-        return String::new();
+        return (String::new(), String::new());
     }
     let Ok(Value::Array(arr)) = serde_json::from_str::<Value>(raw) else {
-        return String::new();
+        return (String::new(), String::new());
     };
-    arr.iter()
+    let Some(ic) = arr
+        .iter()
         .find(|e| e.get("t").and_then(Value::as_str) == Some("ic"))
-        .and_then(|e| e.get("n").and_then(Value::as_str))
+    else {
+        return (String::new(), String::new());
+    };
+    let category = ic
+        .get("n")
+        .and_then(Value::as_str)
         .unwrap_or("")
         .trim()
-        .to_string()
+        .to_string();
+    let bps = ic.get("bps").map(normalize_bps).unwrap_or_default();
+    (category, bps)
+}
+
+fn normalize_bps(v: &Value) -> String {
+    let raw = match v {
+        Value::Number(n) => n.to_string(),
+        Value::String(s) => s.trim().to_string(),
+        _ => String::new(),
+    };
+    let Ok(n) = raw.parse::<f64>() else {
+        return raw;
+    };
+    if !n.is_finite() {
+        return String::new();
+    }
+    let mut s = format!("{n:.6}");
+    while s.contains('.') && s.ends_with('0') {
+        s.pop();
+    }
+    if s.ends_with('.') {
+        s.pop();
+    }
+    s
 }
 
 /// Parse a money cell; blanks/garbage become `0.0` (mirrors `par_extract.to_float`).
@@ -429,6 +472,7 @@ ref2,Settled,visastandarddebit,visa,FR,EUR,100.00,0.05,0.00,0.02,0.20,\"[{\"\"t\
         assert_eq!(r.txn_ref, "ref2");
         assert_eq!(r.funding, "debit");
         assert_eq!(r.ic_category, "Intra EEA Consumer EMV Debit");
+        assert_eq!(r.interchange_bps, "");
         assert!((r.total_fee - 0.27).abs() < 1e-9, "0.05+0.00+0.02+0.20");
         assert!((r.gross - 100.27).abs() < 1e-9, "payable + total_fee");
         assert!(r.txn_date.is_none(), "no Booking Date column -> None");
@@ -446,9 +490,24 @@ ref2,Settled,visastandarddebit,visa,FR,EUR,100.00,0.05,0.00,0.02,0.20,\"[{\"\"t\
 
     #[test]
     fn ic_category_absent_yields_empty() {
-        assert_eq!(ic_category(""), "");
-        assert_eq!(ic_category("[{\"t\":\"scheme\",\"n\":\"x\"}]"), "");
-        assert_eq!(ic_category("not json"), "");
+        assert_eq!(ic_details(""), (String::new(), String::new()));
+        assert_eq!(
+            ic_details("[{\"t\":\"scheme\",\"n\":\"x\"}]"),
+            (String::new(), String::new())
+        );
+        assert_eq!(ic_details("not json"), (String::new(), String::new()));
+    }
+
+    #[test]
+    fn ic_details_extracts_rate_bps() {
+        assert_eq!(
+            ic_details("[{\"t\":\"ic\",\"n\":\"Consumer Debit\",\"bps\":0.2}]"),
+            ("Consumer Debit".to_string(), "0.2".to_string())
+        );
+        assert_eq!(
+            ic_details("[{\"t\":\"ic\",\"n\":\"Commercial\",\"bps\":\"142.5000\"}]"),
+            ("Commercial".to_string(), "142.5".to_string())
+        );
     }
 
     #[test]
diff --git a/src/cost_ingestion/connectors/braintree.rs b/src/cost_ingestion/connectors/braintree.rs
index 6b021568..8bc37199 100644
--- a/src/cost_ingestion/connectors/braintree.rs
+++ b/src/cost_ingestion/connectors/braintree.rs
@@ -197,12 +197,14 @@ impl SettlementReportSource for BraintreeReportSource {
 
                 Ok(Some(SettledFeeRow {
                     txn_ref: row.get(c.txn).to_string(),
+                    report_account: String::new(),
                     card_network: network,
                     variant,
                     funding,
                     issuer_country: row.get_opt(c.issuer).trim().to_string(),
                     currency: row.get(c.ccy).trim().to_string(),
                     ic_category: row.get_opt(c.ic_desc).trim().to_string(),
+                    interchange_bps: String::new(),
                     txn_date,
                     // Braintree's PAR carries no terminal/POS indicator, so every row is treated as
                     // online. Revisit if in-person / pinless acceptance data becomes distinguishable.
diff --git a/src/cost_ingestion/connectors/chase.rs b/src/cost_ingestion/connectors/chase.rs
index 3050f576..e288ca40 100644
--- a/src/cost_ingestion/connectors/chase.rs
+++ b/src/cost_ingestion/connectors/chase.rs
@@ -366,12 +366,14 @@ impl SettlementReportSource for ChaseReportSource {
 
                 Ok(Some(SettledFeeRow {
                     txn_ref: row.get(c.order).trim().to_string(),
+                    report_account: String::new(),
                     card_network: network,
                     variant,
                     funding,
                     issuer_country: row.get(c.issuer).trim().to_string(),
                     currency: row.get(c.currency).trim().to_string(),
                     ic_category: row.get(c.ic_code).trim().to_string(),
+                    interchange_bps: String::new(),
                     txn_date,
                     channel,
                     gross,
diff --git a/src/cost_ingestion/connectors/checkout.rs b/src/cost_ingestion/connectors/checkout.rs
index 714208a3..2dffa53d 100644
--- a/src/cost_ingestion/connectors/checkout.rs
+++ b/src/cost_ingestion/connectors/checkout.rs
@@ -176,6 +176,7 @@ impl PaymentAcc {
         let total_fee = self.scheme_fee + self.commission;
         Some(SettledFeeRow {
             txn_ref: self.txn_ref,
+            report_account: String::new(),
             card_network: self.card_network,
             variant: self.variant,
             funding: self.funding,
@@ -184,6 +185,7 @@ impl PaymentAcc {
             // No single interchange category exists for blended pricing (`Fee Detail` is per-line
             // and noisy), so leave it empty — the rollup buckets all blended volume together.
             ic_category: String::new(),
+            interchange_bps: String::new(),
             txn_date: self.txn_date,
             // Checkout is card-not-present online acceptance; no terminal id in the report.
             channel: "ecom".to_string(),
diff --git a/src/cost_ingestion/connectors/stripe.rs b/src/cost_ingestion/connectors/stripe.rs
index 0df2b93b..9308321b 100644
--- a/src/cost_ingestion/connectors/stripe.rs
+++ b/src/cost_ingestion/connectors/stripe.rs
@@ -199,12 +199,14 @@ impl SettlementReportSource for StripeReportSource {
                     &currency,
                     month,
                 ]),
+                report_account: String::new(),
                 card_network,
                 variant,
                 funding,
                 issuer_country: String::new(),
                 currency,
                 ic_category: String::new(),
+                interchange_bps: String::new(),
                 txn_date,
                 channel,
                 gross,
diff --git a/src/cost_ingestion/coverage.rs b/src/cost_ingestion/coverage.rs
index d328908e..42511a61 100644
--- a/src/cost_ingestion/coverage.rs
+++ b/src/cost_ingestion/coverage.rs
@@ -24,11 +24,13 @@ pub struct CoverageSummary {
     pub good_clusters: u64,
     pub thin_clusters: u64,
     pub non_linear_clusters: u64,
+    pub fan_clusters: u64,
     // Transaction counts by verdict (the thin-tail vs non-linear split of the gap).
     pub total_txns: u64,
     pub good_txns: u64,
     pub thin_txns: u64,
     pub non_linear_txns: u64,
+    pub fan_txns: u64,
     /// Share of *transactions* with a trustworthy cost model.
     pub good_txn_pct: f64,
     // Money-weighted coverage — the headline for a cost/EV system.
@@ -36,6 +38,7 @@ pub struct CoverageSummary {
     pub good_gross: f64,
     pub thin_gross: f64,
     pub non_linear_gross: f64,
+    pub fan_gross: f64,
     /// Share of settled *volume* (money) with a trustworthy cost model.
     pub good_gross_pct: f64,
     // Fit accuracy of the GOOD models (per-txn cost error, basis points).
@@ -57,14 +60,17 @@ SELECT
     countIf(verdict = 'GOOD') AS good_clusters,
     countIf(verdict = 'THIN') AS thin_clusters,
     countIf(verdict = 'NON_LINEAR') AS non_linear_clusters,
+    countIf(verdict = 'FAN') AS fan_clusters,
     sum(n) AS total_txns,
     sumIf(n, verdict = 'GOOD') AS good_txns,
     sumIf(n, verdict = 'THIN') AS thin_txns,
     sumIf(n, verdict = 'NON_LINEAR') AS non_linear_txns,
+    sumIf(n, verdict = 'FAN') AS fan_txns,
     sum(gross_sum) AS total_gross,
     sumIf(gross_sum, verdict = 'GOOD') AS good_gross,
     sumIf(gross_sum, verdict = 'THIN') AS thin_gross,
     sumIf(gross_sum, verdict = 'NON_LINEAR') AS non_linear_gross,
+    sumIf(gross_sum, verdict = 'FAN') AS fan_gross,
     quantileIf(0.5)(bps_rmse, verdict = 'GOOD') AS bps_rmse_p50,
     quantileIf(0.9)(bps_rmse, verdict = 'GOOD') AS bps_rmse_p90,
     toString(rd) AS report_date
@@ -117,18 +123,21 @@ pub async fn for_merchant(
     let good_clusters = u(1);
     let thin_clusters = u(2);
     let non_linear_clusters = u(3);
-    let total_txns = u(4);
-    let good_txns = u(5);
-    let thin_txns = u(6);
-    let non_linear_txns = u(7);
-    let total_gross = g(8);
-    let good_gross = g(9);
-    let thin_gross = g(10);
-    let non_linear_gross = g(11);
+    let fan_clusters = u(4);
+    let total_txns = u(5);
+    let good_txns = u(6);
+    let thin_txns = u(7);
+    let non_linear_txns = u(8);
+    let fan_txns = u(9);
+    let total_gross = g(10);
+    let good_gross = g(11);
+    let thin_gross = g(12);
+    let non_linear_gross = g(13);
+    let fan_gross = g(14);
     // quantiles come back as `nan` when there are no GOOD clusters; treat as 0.
-    let bps_rmse_p50 = g(12);
-    let bps_rmse_p90 = g(13);
-    let report_date = f.get(14).unwrap_or(&"").trim().to_string();
+    let bps_rmse_p50 = g(15);
+    let bps_rmse_p90 = g(16);
+    let report_date = f.get(17).unwrap_or(&"").trim().to_string();
 
     let good_txn_pct = if total_txns > 0 {
         good_txns as f64 / total_txns as f64 * 100.0
@@ -145,15 +154,18 @@ pub async fn for_merchant(
         good_clusters,
         thin_clusters,
         non_linear_clusters,
+        fan_clusters,
         total_txns,
         good_txns,
         thin_txns,
         non_linear_txns,
+        fan_txns,
         good_txn_pct,
         total_gross,
         good_gross,
         thin_gross,
         non_linear_gross,
+        fan_gross,
         good_gross_pct,
         bps_rmse_p50: if bps_rmse_p50.is_nan() {
             0.0
diff --git a/src/cost_ingestion/detect.rs b/src/cost_ingestion/detect.rs
index 466ddb6a..c1113404 100644
--- a/src/cost_ingestion/detect.rs
+++ b/src/cost_ingestion/detect.rs
@@ -32,6 +32,8 @@ pub struct PriceChange {
     pub issuer_country: String,
     pub currency: String,
     pub ic_category: String,
+    pub interchange_bps: String,
+    pub segment_idx: u16,
     pub old_pct_bps: f64,
     pub new_pct_bps: f64,
     pub old_fixed: f64,
@@ -44,9 +46,11 @@ const CHANGES_SQL: &str = r#"
 WITH ranked AS (
     SELECT
         connector, account, card_network, variant, funding, issuer_country, currency, ic_category,
+        interchange_bps, segment_idx,
         report_date, pct_bps, fixed, verdict,
         row_number() OVER (
-            PARTITION BY connector, account, card_network, variant, funding, issuer_country, currency, ic_category
+            PARTITION BY connector, account, card_network, variant, funding, issuer_country,
+                         currency, ic_category, interchange_bps, segment_idx
             ORDER BY report_date DESC
         ) AS rn
     FROM __DB__.cost_fee_model FINAL
@@ -54,13 +58,14 @@ WITH ranked AS (
 )
 SELECT
     cur.connector, cur.account, cur.card_network, cur.variant, cur.funding,
-    cur.issuer_country, cur.currency, cur.ic_category,
+    cur.issuer_country, cur.currency, cur.ic_category, cur.interchange_bps, cur.segment_idx,
     prev.pct_bps AS old_pct, cur.pct_bps AS new_pct,
     prev.fixed AS old_fixed, cur.fixed AS new_fixed,
     toString(cur.report_date) AS changed_on
 FROM (SELECT * FROM ranked WHERE rn = 1) AS cur
 INNER JOIN (SELECT * FROM ranked WHERE rn = 2) AS prev
-    USING (connector, account, card_network, variant, funding, issuer_country, currency, ic_category)
+    USING (connector, account, card_network, variant, funding, issuer_country, currency,
+           ic_category, interchange_bps, segment_idx)
 WHERE cur.verdict = 'GOOD'
   AND (abs(cur.pct_bps - prev.pct_bps) > {tol_bps:Float64}
        OR abs(cur.fixed - prev.fixed) > {tol_fixed:Float64})
@@ -108,7 +113,7 @@ pub async fn price_changes(
             continue;
         }
         let f: Vec<&str> = line.split('\t').collect();
-        if f.len() < 13 {
+        if f.len() < 15 {
             continue;
         }
         let g = |i: usize| f[i].trim().parse::<f64>().unwrap_or(0.0);
@@ -121,11 +126,13 @@ pub async fn price_changes(
             issuer_country: f[5].to_string(),
             currency: f[6].to_string(),
             ic_category: f[7].to_string(),
-            old_pct_bps: g(8),
-            new_pct_bps: g(9),
-            old_fixed: g(10),
-            new_fixed: g(11),
-            changed_on: f[12].trim().to_string(),
+            interchange_bps: f[8].to_string(),
+            segment_idx: f[9].trim().parse().unwrap_or(0),
+            old_pct_bps: g(10),
+            new_pct_bps: g(11),
+            old_fixed: g(12),
+            new_fixed: g(13),
+            changed_on: f[14].trim().to_string(),
         });
     }
     Ok(out)
diff --git a/src/cost_ingestion/fit.rs b/src/cost_ingestion/fit.rs
index f6706e32..a9c5a542 100644
--- a/src/cost_ingestion/fit.rs
+++ b/src/cost_ingestion/fit.rs
@@ -1,21 +1,23 @@
-//! Fit per-cluster cost models from the daily sufficient-statistics rollup — the OLS of
-//! `par_fit.py` expressed as a ClickHouse `GROUP BY`.
+//! Fit per-cluster cost models from the daily sufficient-statistics rollup.
 //!
-//! `cost_daily_stats` already holds, per (cluster × day × band × channel), the additive sums an OLS
-//! fit needs (`Σx, Σy, Σxx, Σxy, Σyy` and reciprocal terms). The fit sums those buckets over the
-//! window — first collapsing band/channel, then across days — to reconstruct the exact per-cluster
-//! sums it would get from raw transactions, and computes `pct_bps = slope·10⁴`, `fixed = intercept`,
-//! and a per-transaction `bps_rmse`. Clusters are graded `GOOD` / `NON_LINEAR` / `THIN` by the §10
-//! rule (`n ≥ 200 AND bps_rmse ≤ 15`). The €5 micro-amount floor was already applied at aggregation.
+//! `cost_daily_stats` holds additive OLS sums per `(cluster × day × fit_bucket × channel)`, plus a
+//! bounded sample used only for fan detection. The fitter loads one connector/account/merchant
+//! snapshot, reconstructs the same sufficient statistics the raw rows would provide, and ports the
+//! richer `scratch/cluster_explorer.py` grading logic into production:
 //!
-//! Runs entirely in ClickHouse: one `INSERT … SELECT` writes the snapshot, then a summary query
-//! reports coverage for the validation gate. See `scratch/inhouse-cost-architecture.md` §3, §7 and
-//! `scratch/settlement-table-removal-worked-example.md`.
+//! - no currency-blind micro-amount floor;
+//! - fixed/proportional decomposition at `a* = fixed / rate`;
+//! - L2 confidence promotion for reliable lower-volume fits;
+//! - L1 amount-range segmentation for non-linear/thin clusters;
+//! - fan detection for minority sub-populations that RMSE can average away.
 
+use std::collections::BTreeMap;
 use std::sync::OnceLock;
 use std::time::Duration;
 
 use masking::PeekInterface;
+use serde::Deserialize;
+use serde_json::json;
 
 use crate::config::ClickHouseAnalyticsConfig;
 
@@ -23,6 +25,26 @@ use super::types::IngestError;
 
 const FIT_TIMEOUT: Duration = Duration::from_secs(120);
 
+const MIN_N: u64 = 200;
+const MAX_BPS: f64 = 15.0;
+const MAX_SEGMENTS: usize = 5;
+const SEG_FLOOR: u64 = 25;
+const L2_MIN_N: u64 = 30;
+const L2_MAX_PCT_BPS_CI: f64 = 15.0;
+const FIX_VOL_TOL: f64 = 0.02;
+const FAN_FRAC: f64 = 0.01;
+const FAN_BPS: f64 = 30.0;
+const FAN_MONEY_SEVERE: f64 = MAX_BPS;
+const BUCKETS_PER_DECADE: f64 = 10.0;
+
+/// Base trailing window (days of transactions) every cluster fits over — recent enough that
+/// high-volume clusters react quickly to a fee change.
+pub const BASE_WINDOW_DAYS: i64 = 90;
+/// Hard cap on how far a thin cluster may reach back to accumulate enough samples.
+const MAX_WINDOW_DAYS: i64 = 365;
+/// Minimum transactions for the GOOD sample gate; a thin cluster extends its window toward this.
+const MIN_SAMPLES: u32 = 200;
+
 /// Coverage of a freshly fit snapshot, used by the caller to decide whether to trust it.
 #[derive(Debug, Clone, Copy)]
 pub struct FitSummary {
@@ -35,158 +57,253 @@ fn client() -> &'static reqwest::Client {
     CLIENT.get_or_init(|| super::ch_http::client(FIT_TIMEOUT))
 }
 
-/// The OLS fit as nested aggregation. `__DB__` is replaced with the configured database (the
-/// `{name:Type}` placeholders are ClickHouse query parameters, bound over HTTP).
-const FIT_SQL: &str = r#"
-INSERT INTO __DB__.cost_fee_model
-    (report_date, connector, account, merchant_id, card_network, variant, funding,
-     issuer_country, currency, ic_category, pct_bps, fixed, n, bps_rmse, r2, gross_sum, verdict)
+/// Load per-cluster/per-fit-bucket stats after applying the same adaptive day window as the old
+/// ClickHouse-only fitter: always keep the recent base window, and let thin clusters reach farther
+/// back until the running count crosses `MIN_SAMPLES` or the max window is exhausted.
+const LOAD_ROLLUP_SQL: &str = r#"
 SELECT
-    report_date, connector, account, merchant_id, card_network, variant, funding,
-    issuer_country, currency, ic_category, pct_bps, fixed, n, bps_rmse, r2, gross_sum,
-    multiIf(n < 200, 'THIN', isNaN(bps_rmse) OR bps_rmse > 15, 'NON_LINEAR', 'GOOD') AS verdict
-FROM
+    s.report_account, s.card_network, s.variant, s.funding, s.issuer_country, s.currency, s.ic_category,
+    s.interchange_bps, s.fit_bucket,
+    sum(s.n) AS n,
+    sum(s.sx) AS sx,
+    sum(s.sy) AS sy,
+    sum(s.sxx) AS sxx,
+    sum(s.sxy) AS sxy,
+    sum(s.syy) AS syy,
+    sum(s.su) AS su,
+    sum(s.suu) AS suu,
+    sum(s.suy) AS suy,
+    sum(s.suuy) AS suuy,
+    sum(s.syyuu) AS syyuu,
+    arrayFlatten(groupArray(s.sample_x)) AS sample_x,
+    arrayFlatten(groupArray(s.sample_y)) AS sample_y
+FROM __DB__.cost_daily_stats AS s FINAL
+INNER JOIN
 (
-    SELECT
-        report_date, connector, account, merchant_id, card_network, variant, funding,
-        issuer_country, currency, ic_category, n, r2, gross_sum,
-        slope * 10000 AS pct_bps,
-        intercept AS fixed,
-        sqrt(greatest(0.0, sum_sq) / n) * 10000 AS bps_rmse
+    SELECT report_account, card_network, variant, funding, issuer_country, currency, ic_category,
+           interchange_bps, txn_date
     FROM
     (
         SELECT
-            {report_date:Date} AS report_date,
-            {connector:String} AS connector,
-            {account:String} AS account,
-            {merchant_id:String} AS merchant_id,
-            card_network, variant, funding, issuer_country, currency, ic_category, n,
-            sx AS gross_sum,
-            (n * sxx - sx * sx) AS denom,
-            if(denom = 0, nan, (n * sxy - sx * sy) / denom) AS slope,
-            if(denom = 0, nan, (sy - slope * sx) / n) AS intercept,
-            if(denom = 0 OR (n * syy - sy * sy) = 0, nan,
-               pow(n * sxy - sx * sy, 2) / (denom * (n * syy - sy * sy))) AS r2,
-            (intercept * intercept * suu + n * slope * slope + syyuu
-             - 2 * intercept * suuy - 2 * slope * suy + 2 * intercept * slope * su) AS sum_sq
+            report_account, card_network, variant, funding, issuer_country, currency, ic_category,
+            interchange_bps, txn_date, n,
+            sum(n) OVER (
+                PARTITION BY report_account, card_network, variant, funding, issuer_country, currency,
+                             ic_category, interchange_bps
+                ORDER BY txn_date DESC
+                ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
+            ) AS cum_n_before
         FROM
         (
-            -- Sum the per-day buckets that fall in each cluster's adaptive window into one set of
-            -- per-cluster sufficient statistics (all sums are additive across days/bands/channels).
             SELECT
-                card_network, variant, funding, issuer_country, currency, ic_category,
-                sum(n) AS n,
-                sum(sx) AS sx,
-                sum(sy) AS sy,
-                sum(sxx) AS sxx,
-                sum(sxy) AS sxy,
-                sum(syy) AS syy,
-                sum(su) AS su,
-                sum(suu) AS suu,
-                sum(suy) AS suy,
-                sum(suuy) AS suuy,
-                sum(syyuu) AS syyuu
-            FROM
-            (
-                -- Adaptive per-cluster window at day granularity: keep every day in the base window
-                -- (recent, so high-volume clusters stay agile to price changes), and let thin
-                -- clusters reach back over older days until the running txn count crosses
-                -- MIN_SAMPLES (capped at the max window) so they can cross the GOOD sample gate
-                -- instead of being stuck THIN. `cum_n_before` = txns on strictly-more-recent days.
-                SELECT *
-                FROM
-                (
-                    SELECT
-                        card_network, variant, funding, issuer_country, currency, ic_category,
-                        txn_date, n, sx, sy, sxx, sxy, syy, su, suu, suy, suuy, syyuu,
-                        sum(n) OVER (
-                            PARTITION BY card_network, variant, funding, issuer_country, currency, ic_category
-                            ORDER BY txn_date DESC
-                            ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
-                        ) AS cum_n_before
-                    FROM
-                    (
-                        -- Collapse band/channel to per-(cluster, day) sums. FINAL dedups the
-                        -- ReplacingMergeTree so a day re-delivered by a later report (overlapping
-                        -- monthly+daily, a re-upload, webhook+manual) is counted once — its latest
-                        -- authoritative bucket wins.
-                        SELECT
-                            card_network, variant, funding, issuer_country, currency, ic_category,
-                            txn_date,
-                            sum(n) AS n, sum(sx) AS sx, sum(sy) AS sy, sum(sxx) AS sxx,
-                            sum(sxy) AS sxy, sum(syy) AS syy, sum(su) AS su, sum(suu) AS suu,
-                            sum(suy) AS suy, sum(suuy) AS suuy, sum(syyuu) AS syyuu
-                        FROM __DB__.cost_daily_stats FINAL
-                        WHERE connector = {connector:String}
-                          AND account = {account:String}
-                          AND merchant_id = {merchant_id:String}
-                          AND txn_date >= {max_window_start:Date}
-                        GROUP BY card_network, variant, funding, issuer_country, currency,
-                                 ic_category, txn_date
-                    )
-                )
-                WHERE txn_date >= {base_window_start:Date}
-                   OR coalesce(cum_n_before, 0) < {min_samples:UInt32}
-            )
-            GROUP BY card_network, variant, funding, issuer_country, currency, ic_category
+                report_account, card_network, variant, funding, issuer_country, currency, ic_category,
+                interchange_bps, txn_date, sum(n) AS n
+            FROM __DB__.cost_daily_stats FINAL
+            WHERE connector = {connector:String}
+              AND account = {account:String}
+              AND merchant_id = {merchant_id:String}
+              AND txn_date >= {max_window_start:Date}
+            GROUP BY report_account, card_network, variant, funding, issuer_country, currency, ic_category,
+                     interchange_bps, txn_date
         )
     )
-)
-"#;
-
-const SUMMARY_SQL: &str = r#"
-SELECT count() AS total, countIf(verdict = 'GOOD') AS good
-FROM __DB__.cost_fee_model
-WHERE connector = {connector:String} AND account = {account:String}
-  AND merchant_id = {merchant_id:String} AND report_date = {report_date:Date}
-FORMAT TSV
+    WHERE txn_date >= {base_window_start:Date}
+       OR coalesce(cum_n_before, 0) < {min_samples:UInt32}
+) AS d
+USING (report_account, card_network, variant, funding, issuer_country, currency, ic_category,
+       interchange_bps, txn_date)
+WHERE s.connector = {connector:String}
+  AND s.account = {account:String}
+  AND s.merchant_id = {merchant_id:String}
+  AND s.txn_date >= {max_window_start:Date}
+GROUP BY s.report_account, s.card_network, s.variant, s.funding, s.issuer_country, s.currency, s.ic_category,
+         s.interchange_bps, s.fit_bucket
+ORDER BY s.report_account, s.card_network, s.variant, s.funding, s.issuer_country, s.currency, s.ic_category,
+         s.interchange_bps, s.fit_bucket
+FORMAT JSONEachRow
 "#;
 
-// Clear this (connector, account, report_date) snapshot before the fit re-inserts it, so a refit is
-// a clean REPLACE, not an append. Without this, a refit after a delete (or a same-day re-ingest)
-// would leave stale clusters the new fit no longer produces — including the whole snapshot when the
-// new fit is empty (an INSERT of nothing can't overwrite the old rows).
 const CLEAR_SNAPSHOT_SQL: &str = r#"
 DELETE FROM __DB__.cost_fee_model
 WHERE connector = {connector:String} AND account = {account:String}
   AND merchant_id = {merchant_id:String} AND report_date = {report_date:Date}
 "#;
 
-/// When a refit yields an empty snapshot — the data behind this `(connector, account)` is gone
-/// (e.g. its last ingestion was deleted) — drop ALL of its prior `cost_fee_model` snapshots too.
-/// Coverage and serving read the *latest* snapshot by `report_date`; without this, an empty refit
-/// (which inserts no rows for today) leaves an older non-empty snapshot as the max, so the dashboard
-/// and the router keep showing / routing on models that no longer have any supporting data.
 const PURGE_MODEL_SQL: &str = r#"
 DELETE FROM __DB__.cost_fee_model
 WHERE connector = {connector:String} AND account = {account:String}
   AND merchant_id = {merchant_id:String}
 "#;
 
-/// Base trailing window (days of transactions) every cluster fits over — recent enough that
-/// high-volume clusters react quickly to a fee change.
-pub const BASE_WINDOW_DAYS: i64 = 90;
-/// Hard cap on how far a thin cluster may reach back to accumulate enough samples.
-const MAX_WINDOW_DAYS: i64 = 365;
-/// Minimum transactions for the GOOD sample gate; a thin cluster extends its window toward this.
-/// Must match the `n < 200 -> THIN` verdict gate in `FIT_SQL`.
-const MIN_SAMPLES: u32 = 200;
+const INSERT_COLUMNS: &str = "\
+report_date,connector,account,report_account,merchant_id,card_network,variant,funding,issuer_country,currency,\
+ic_category,interchange_bps,segment_idx,amount_lo,amount_hi,pct_bps,fixed,n,gross_sum,bps_rmse,\
+grade_bps,pct_ci95_bps,crossover_amount,prop_bps,fix_abs,fix_bps,below_gross_frac,fan_frac,fan_money_bps,\
+r2,verdict";
 
-/// Whether a fit result should trigger the empty-refit purge ([`PURGE_MODEL_SQL`]). Extracted as a
-/// pure function because it is the one decision coupled to a destructive `DELETE`, so it is locked
-/// in by unit tests: purge fires ONLY on a *definitively-parsed* zero cluster count — never on a
-/// parse failure (`None`), which would otherwise masquerade as "empty" and delete a healthy model —
-/// and never with a blank identifier that could widen the delete's scope.
-fn should_purge_empty(
-    total_parsed: Option<u64>,
-    connector: &str,
-    account: &str,
-    merchant_id: &str,
-) -> bool {
-    total_parsed == Some(0)
-        && !connector.is_empty()
-        && !account.is_empty()
-        && !merchant_id.is_empty()
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+struct ClusterKey {
+    report_account: String,
+    card_network: String,
+    variant: String,
+    funding: String,
+    issuer_country: String,
+    currency: String,
+    ic_category: String,
+    interchange_bps: String,
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+struct FitStats {
+    n: u64,
+    sx: f64,
+    sy: f64,
+    sxx: f64,
+    sxy: f64,
+    syy: f64,
+    su: f64,
+    suu: f64,
+    suy: f64,
+    suuy: f64,
+    syyuu: f64,
+}
+
+impl FitStats {
+    fn merge(&mut self, other: &Self) {
+        self.n += other.n;
+        self.sx += other.sx;
+        self.sy += other.sy;
+        self.sxx += other.sxx;
+        self.sxy += other.sxy;
+        self.syy += other.syy;
+        self.su += other.su;
+        self.suu += other.suu;
+        self.suy += other.suy;
+        self.suuy += other.suuy;
+        self.syyuu += other.syyuu;
+    }
+
+    fn minus(self, other: Self) -> Self {
+        Self {
+            n: self.n.saturating_sub(other.n),
+            sx: self.sx - other.sx,
+            sy: self.sy - other.sy,
+            sxx: self.sxx - other.sxx,
+            sxy: self.sxy - other.sxy,
+            syy: self.syy - other.syy,
+            su: self.su - other.su,
+            suu: self.suu - other.suu,
+            suy: self.suy - other.suy,
+            suuy: self.suuy - other.suuy,
+            syyuu: self.syyuu - other.syyuu,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Sample {
+    x: f64,
+    y: f64,
+}
+
+#[derive(Debug, Clone)]
+struct Bucket {
+    fit_bucket: i32,
+    stats: FitStats,
+    samples: Vec<Sample>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct RollupRow {
+    #[serde(default)]
+    report_account: String,
+    card_network: String,
+    variant: String,
+    funding: String,
+    issuer_country: String,
+    currency: String,
+    ic_category: String,
+    #[serde(default)]
+    interchange_bps: String,
+    fit_bucket: i32,
+    n: u64,
+    sx: f64,
+    sy: f64,
+    sxx: f64,
+    sxy: f64,
+    syy: f64,
+    su: f64,
+    suu: f64,
+    suy: f64,
+    suuy: f64,
+    syyuu: f64,
+    #[serde(default)]
+    sample_x: Vec<f64>,
+    #[serde(default)]
+    sample_y: Vec<f64>,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct OlsFit {
+    slope: f64,
+    intercept: f64,
+    bps_rmse: f64,
+    se_pct_bps: f64,
+    r2: f64,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Decomp {
+    crossover_amount: f64,
+    prop_bps: f64,
+    fix_abs: f64,
+    fix_bps: f64,
+    below_n: u64,
+    above_n: u64,
+    below_frac: f64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Verdict {
+    Good,
+    NonLinear,
+    Thin,
+    Fan,
+}
+
+impl Verdict {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Good => "GOOD",
+            Self::NonLinear => "NON_LINEAR",
+            Self::Thin => "THIN",
+            Self::Fan => "FAN",
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+struct ModelRow {
+    key: ClusterKey,
+    segment_idx: u16,
+    amount_lo: f64,
+    amount_hi: f64,
+    pct_bps: f64,
+    fixed: f64,
+    n: u64,
+    gross_sum: f64,
+    bps_rmse: f64,
+    grade_bps: f64,
+    pct_ci95_bps: f64,
+    crossover_amount: f64,
+    prop_bps: f64,
+    fix_abs: f64,
+    fix_bps: f64,
+    below_gross_frac: f64,
+    fan_frac: f64,
+    fan_money_bps: f64,
+    r2: f64,
+    verdict: Verdict,
 }
 
 /// Fit `(connector, account, merchant_id)` from the last `BASE_WINDOW_DAYS` of `cost_daily_stats`
@@ -205,8 +322,6 @@ pub async fn fit_snapshot(
         ("merchant_id", merchant_id.to_string()),
     ];
 
-    // Window bounds are relative to the *latest transaction in the data*, not the wall clock — so a
-    // backfill of older reports (or any ingestion cadence) fits correctly regardless of upload time.
     let bounds_sql = format!(
         "SELECT toString(max(txn_date) - toIntervalDay({BASE_WINDOW_DAYS})), \
                 toString(max(txn_date) - toIntervalDay({MAX_WINDOW_DAYS})) \
@@ -217,7 +332,6 @@ pub async fn fit_snapshot(
     );
     let bounds = exec(cfg, &bounds_sql, &base_params).await?;
     let mut cols = bounds.trim().split('\t');
-    // Empty staging → NULL dates; a sentinel keeps the query valid and yields 0 clusters.
     let clean = |s: &str| -> String {
         let s = s.trim();
         if s.is_empty() || s == "\\N" {
@@ -239,30 +353,27 @@ pub async fn fit_snapshot(
         ("min_samples", MIN_SAMPLES.to_string()),
     ];
 
-    // Idempotent snapshot: clear then re-insert, so the result exactly reflects current staging.
+    let rollup = exec(
+        cfg,
+        &LOAD_ROLLUP_SQL.replace("__DB__", &cfg.database),
+        &params,
+    )
+    .await?;
+    let clusters = parse_rollup(&rollup)?;
+    let rows = fit_clusters(&clusters);
+    let total = u64::try_from(rows.len()).unwrap_or(u64::MAX);
+    let good =
+        u64::try_from(rows.iter().filter(|r| r.verdict == Verdict::Good).count()).unwrap_or(0);
+
     exec(
         cfg,
         &CLEAR_SNAPSHOT_SQL.replace("__DB__", &cfg.database),
         &params,
     )
     .await?;
-    exec(cfg, &FIT_SQL.replace("__DB__", &cfg.database), &params).await?;
-    let summary = exec(cfg, &SUMMARY_SQL.replace("__DB__", &cfg.database), &params).await?;
-
-    let mut fields = summary.trim().split('\t');
-    // Parse `total` strictly: a parse *failure* must NOT collapse to "0 clusters", because that
-    // value drives the destructive purge below. `unwrap_or(0)` here would turn a malformed/empty
-    // summary response into an accidental table-wide delete. Only a definitively-parsed 0 purges.
-    let total_parsed = fields.next().and_then(|s| s.parse::<u64>().ok());
-    let good = fields.next().and_then(|s| s.parse().ok()).unwrap_or(0);
-    let total = total_parsed.unwrap_or(0);
-
-    // A definitively-empty refit means this (connector, account) has no fittable data left in the
-    // entire fit window — its rows were deleted, or it never had any. A single sparse report cannot
-    // cause this, since the fit windows over ALL staged days, not just the one just ingested. Purge
-    // its stale snapshots so coverage/serving don't fall back to an older non-empty one (see
-    // PURGE_MODEL_SQL and `should_purge_empty`).
-    if should_purge_empty(total_parsed, connector, account, merchant_id) {
+    insert_models(cfg, connector, account, merchant_id, report_date, &rows).await?;
+
+    if should_purge_empty(Some(total), connector, account, merchant_id) {
         exec(
             cfg,
             &PURGE_MODEL_SQL.replace("__DB__", &cfg.database),
@@ -277,14 +388,619 @@ pub async fn fit_snapshot(
     })
 }
 
+fn parse_rollup(text: &str) -> Result<BTreeMap<ClusterKey, Vec<Bucket>>, IngestError> {
+    let mut out: BTreeMap<ClusterKey, Vec<Bucket>> = BTreeMap::new();
+    for line in text.lines().filter(|l| !l.trim().is_empty()) {
+        let row: RollupRow =
+            serde_json::from_str(line).map_err(|e| IngestError::Storage(e.to_string()))?;
+        let key = ClusterKey {
+            report_account: row.report_account,
+            card_network: row.card_network,
+            variant: row.variant,
+            funding: row.funding,
+            issuer_country: row.issuer_country,
+            currency: row.currency,
+            ic_category: row.ic_category,
+            interchange_bps: row.interchange_bps,
+        };
+        let stats = FitStats {
+            n: row.n,
+            sx: row.sx,
+            sy: row.sy,
+            sxx: row.sxx,
+            sxy: row.sxy,
+            syy: row.syy,
+            su: row.su,
+            suu: row.suu,
+            suy: row.suy,
+            suuy: row.suuy,
+            syyuu: row.syyuu,
+        };
+        let samples = row
+            .sample_x
+            .into_iter()
+            .zip(row.sample_y.into_iter())
+            .filter(|(x, y)| x.is_finite() && *x > 0.0 && y.is_finite())
+            .map(|(x, y)| Sample { x, y })
+            .collect();
+        out.entry(key).or_default().push(Bucket {
+            fit_bucket: row.fit_bucket,
+            stats,
+            samples,
+        });
+    }
+    for buckets in out.values_mut() {
+        buckets.sort_by_key(|b| b.fit_bucket);
+    }
+    Ok(out)
+}
+
+fn fit_clusters(clusters: &BTreeMap<ClusterKey, Vec<Bucket>>) -> Vec<ModelRow> {
+    let mut rows = Vec::new();
+    for (key, buckets) in clusters {
+        rows.extend(fit_one_cluster(key, buckets));
+    }
+    rows
+}
+
+fn fit_one_cluster(key: &ClusterKey, buckets: &[Bucket]) -> Vec<ModelRow> {
+    let whole = merge_bucket_stats(buckets);
+    let all_samples = samples_for_range(buckets, f64::NEG_INFINITY, f64::INFINITY);
+    let mut whole_row = build_row(key, 0, 0.0, 0.0, whole, buckets, &all_samples, None);
+
+    if whole_row.verdict == Verdict::Thin
+        && whole.n >= L2_MIN_N
+        && whole_row.grade_bps.is_finite()
+        && whole_row.grade_bps <= MAX_BPS
+        && whole_row.pct_ci95_bps <= L2_MAX_PCT_BPS_CI
+    {
+        whole_row.verdict = Verdict::Good;
+    }
+    if whole_row.verdict == Verdict::Good
+        && whole_row.fan_frac > FAN_FRAC
+        && whole_row.fan_money_bps > FAN_MONEY_SEVERE
+    {
+        whole_row.verdict = Verdict::Fan;
+    }
+
+    if whole_row.verdict == Verdict::Good {
+        return vec![whole_row];
+    }
+
+    let parts = segment_partitions(buckets);
+    if parts.len() <= 1 {
+        return vec![whole_row];
+    }
+
+    let mut seg_rows = Vec::new();
+    for (idx, (p, q)) in parts.into_iter().enumerate() {
+        let seg_buckets = &buckets[p..q];
+        let stats = merge_bucket_stats(seg_buckets);
+        let (lo, _) = bucket_range(seg_buckets[0].fit_bucket);
+        let (_, hi) = bucket_range(seg_buckets[seg_buckets.len() - 1].fit_bucket);
+        let samples = samples_for_range(buckets, lo, hi);
+        let segment_idx = u16::try_from(idx + 1).unwrap_or(u16::MAX);
+        let row = build_row(
+            key,
+            segment_idx,
+            lo,
+            hi,
+            stats,
+            seg_buckets,
+            &samples,
+            Some(grade_segment(stats, seg_buckets)),
+        );
+        seg_rows.push(row);
+    }
+
+    if seg_rows.iter().any(|r| r.verdict == Verdict::Good) {
+        seg_rows
+    } else {
+        vec![whole_row]
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+fn build_row(
+    key: &ClusterKey,
+    segment_idx: u16,
+    amount_lo: f64,
+    amount_hi: f64,
+    stats: FitStats,
+    buckets: &[Bucket],
+    samples: &[Sample],
+    verdict_override: Option<Verdict>,
+) -> ModelRow {
+    let fit = fit_stats(stats);
+    let dec = decompose(stats, buckets, fit.slope, fit.intercept);
+    let grade_bps = if dec.above_n > 0 {
+        dec.prop_bps
+    } else {
+        dec.fix_bps
+    };
+    let fan_frac = dispersion(samples, fit.slope, fit.intercept);
+    let fan_money_bps = money_bps(samples, fit.slope, fit.intercept, dec.crossover_amount);
+    let mut verdict = verdict_override.unwrap_or_else(|| {
+        if stats.n < MIN_N {
+            Verdict::Thin
+        } else {
+            grade_decomposed(dec)
+        }
+    });
+    if verdict == Verdict::Good && fan_frac > FAN_FRAC && fan_money_bps > FAN_MONEY_SEVERE {
+        verdict = Verdict::Fan;
+    }
+    ModelRow {
+        key: key.clone(),
+        segment_idx,
+        amount_lo,
+        amount_hi,
+        pct_bps: fit.slope * 10_000.0,
+        fixed: fit.intercept,
+        n: stats.n,
+        gross_sum: stats.sx,
+        bps_rmse: fit.bps_rmse,
+        grade_bps,
+        pct_ci95_bps: if fit.se_pct_bps.is_finite() {
+            1.96 * fit.se_pct_bps
+        } else {
+            f64::INFINITY
+        },
+        crossover_amount: dec.crossover_amount,
+        prop_bps: dec.prop_bps,
+        fix_abs: dec.fix_abs,
+        fix_bps: dec.fix_bps,
+        below_gross_frac: dec.below_frac,
+        fan_frac,
+        fan_money_bps,
+        r2: fit.r2,
+        verdict,
+    }
+}
+
+fn merge_bucket_stats(buckets: &[Bucket]) -> FitStats {
+    let mut out = FitStats::default();
+    for b in buckets {
+        out.merge(&b.stats);
+    }
+    out
+}
+
+fn fit_stats(s: FitStats) -> OlsFit {
+    let n = f64_from_u64(s.n);
+    if s.n < 2 {
+        return nan_fit();
+    }
+    let denom = n * s.sxx - s.sx * s.sx;
+    if denom <= 0.0 {
+        return nan_fit();
+    }
+    let slope = (n * s.sxy - s.sx * s.sy) / denom;
+    let intercept = (s.sy - slope * s.sx) / n;
+    let bps_rmse = eval_bps(intercept, slope, s);
+    let se_pct_bps = if s.n > 2 {
+        let sse = s.syy - intercept * s.sy - slope * s.sxy;
+        (sse.max(0.0) / (n - 2.0) * n / denom).sqrt() * 10_000.0
+    } else {
+        f64::INFINITY
+    };
+    let y_denom = n * s.syy - s.sy * s.sy;
+    let r2 = if y_denom == 0.0 {
+        f64::NAN
+    } else {
+        (n * s.sxy - s.sx * s.sy).powi(2) / (denom * y_denom)
+    };
+    OlsFit {
+        slope,
+        intercept,
+        bps_rmse,
+        se_pct_bps,
+        r2,
+    }
+}
+
+fn nan_fit() -> OlsFit {
+    OlsFit {
+        slope: f64::NAN,
+        intercept: f64::NAN,
+        bps_rmse: f64::NAN,
+        se_pct_bps: f64::INFINITY,
+        r2: f64::NAN,
+    }
+}
+
+fn eval_bps(intercept: f64, slope: f64, s: FitStats) -> f64 {
+    if s.n == 0 {
+        return f64::NAN;
+    }
+    let n = f64_from_u64(s.n);
+    let sum_sq = s.syyuu + intercept * intercept * s.suu + n * slope * slope
+        - 2.0 * intercept * s.suuy
+        - 2.0 * slope * s.suy
+        + 2.0 * intercept * slope * s.su;
+    (sum_sq.max(0.0) / n).sqrt() * 10_000.0
+}
+
+fn crossover(slope: f64, intercept: f64) -> f64 {
+    if !slope.is_finite() || !intercept.is_finite() || slope <= 0.0 || intercept <= 0.0 {
+        0.0
+    } else {
+        intercept / slope
+    }
+}
+
+fn abs_rms(s: FitStats, intercept: f64, slope: f64) -> f64 {
+    if s.n == 0 {
+        return f64::NAN;
+    }
+    let n = f64_from_u64(s.n);
+    let sse = s.syy - 2.0 * intercept * s.sy - 2.0 * slope * s.sxy
+        + n * intercept * intercept
+        + 2.0 * intercept * slope * s.sx
+        + slope * slope * s.sxx;
+    (sse.max(0.0) / n).sqrt()
+}
+
+fn decompose(whole: FitStats, buckets: &[Bucket], slope: f64, intercept: f64) -> Decomp {
+    let crossover_amount = crossover(slope, intercept);
+    let mut below = FitStats::default();
+    let mut above = FitStats::default();
+    for b in buckets {
+        let (_, hi) = bucket_range(b.fit_bucket);
+        if hi <= crossover_amount {
+            below.merge(&b.stats);
+        } else {
+            above.merge(&b.stats);
+        }
+    }
+    let prop_bps = if above.n > 0 {
+        eval_bps(intercept, slope, above)
+    } else {
+        f64::NAN
+    };
+    let fix_abs = if below.n > 0 {
+        abs_rms(below, intercept, slope)
+    } else {
+        f64::NAN
+    };
+    let fix_bps = if below.n > 0 && crossover_amount > 0.0 {
+        fix_abs / crossover_amount * 10_000.0
+    } else {
+        f64::NAN
+    };
+    let total_vol = if whole.sx > 0.0 { whole.sx } else { 1.0 };
+    Decomp {
+        crossover_amount,
+        prop_bps,
+        fix_abs,
+        fix_bps,
+        below_n: below.n,
+        above_n: above.n,
+        below_frac: below.sx / total_vol,
+    }
+}
+
+fn grade_decomposed(dec: Decomp) -> Verdict {
+    if dec.above_n == 0 {
+        return if dec.fix_bps.is_finite() && dec.fix_bps <= MAX_BPS {
+            Verdict::Good
+        } else {
+            Verdict::NonLinear
+        };
+    }
+    if !dec.prop_bps.is_finite() || dec.prop_bps > MAX_BPS {
+        return Verdict::NonLinear;
+    }
+    let fix_ok = dec.below_n == 0
+        || !dec.fix_bps.is_finite()
+        || dec.fix_bps <= MAX_BPS
+        || dec.below_frac <= FIX_VOL_TOL;
+    if fix_ok {
+        Verdict::Good
+    } else {
+        Verdict::NonLinear
+    }
+}
+
+fn grade_segment(stats: FitStats, buckets: &[Bucket]) -> Verdict {
+    if stats.n < SEG_FLOOR {
+        return Verdict::Thin;
+    }
+    let fit = fit_stats(stats);
+    if !fit.slope.is_finite() {
+        return Verdict::NonLinear;
+    }
+    let dec = decompose(stats, buckets, fit.slope, fit.intercept);
+    let grade_bps = if dec.above_n > 0 {
+        dec.prop_bps
+    } else {
+        dec.fix_bps
+    };
+    if !grade_bps.is_finite() || grade_bps > MAX_BPS {
+        return Verdict::NonLinear;
+    }
+    if stats.n >= MIN_N {
+        return Verdict::Good;
+    }
+    if fit.se_pct_bps.is_finite() && 1.96 * fit.se_pct_bps <= L2_MAX_PCT_BPS_CI {
+        Verdict::Good
+    } else {
+        Verdict::Thin
+    }
+}
+
+fn dispersion(samples: &[Sample], slope: f64, intercept: f64) -> f64 {
+    if !slope.is_finite() || !intercept.is_finite() {
+        return 0.0;
+    }
+    let mut xs: Vec<f64> = samples.iter().filter(|p| p.x > 0.0).map(|p| p.x).collect();
+    if xs.len() < 20 {
+        return 0.0;
+    }
+    xs.sort_by(|a, b| a.total_cmp(b));
+    let threshold = xs[xs.len() / 5];
+    let mut off = 0_u64;
+    let mut total = 0_u64;
+    for p in samples {
+        if p.x < threshold || p.x <= 0.0 {
+            continue;
+        }
+        if ((p.y - (intercept + slope * p.x)) / p.x).abs() * 10_000.0 > FAN_BPS {
+            off += 1;
+        }
+        total += 1;
+    }
+    if total == 0 {
+        0.0
+    } else {
+        f64_from_u64(off) / f64_from_u64(total)
+    }
+}
+
+fn money_bps(samples: &[Sample], slope: f64, intercept: f64, crossover_amount: f64) -> f64 {
+    if !slope.is_finite() || !intercept.is_finite() {
+        return 0.0;
+    }
+    let mut num = 0.0;
+    let mut den = 0.0;
+    for p in samples {
+        if p.x <= 0.0 || p.x < crossover_amount {
+            continue;
+        }
+        num += ((intercept + slope * p.x) - p.y).abs();
+        den += p.x;
+    }
+    if den > 0.0 {
+        num / den * 10_000.0
+    } else {
+        0.0
+    }
+}
+
+fn segment_partitions(buckets: &[Bucket]) -> Vec<(usize, usize)> {
+    let m = buckets.len();
+    if m == 0 {
+        return Vec::new();
+    }
+
+    let mut pref = Vec::with_capacity(m + 1);
+    let mut running = FitStats::default();
+    pref.push(running);
+    for b in buckets {
+        running.merge(&b.stats);
+        pref.push(running);
+    }
+    let range_stats = |i: usize, j: usize| pref[j].minus(pref[i]);
+    let cost = |i: usize, j: usize| -> f64 {
+        let s = range_stats(i, j);
+        if s.n < SEG_FLOOR {
+            return f64::INFINITY;
+        }
+        let fit = fit_stats(s);
+        if fit.bps_rmse.is_finite() {
+            (fit.bps_rmse / 10_000.0).powi(2) * f64_from_u64(s.n)
+        } else {
+            f64::INFINITY
+        }
+    };
+
+    let max_k = MAX_SEGMENTS.min(m);
+    let mut dp = vec![vec![f64::INFINITY; m + 1]; max_k + 1];
+    let mut back = vec![vec![usize::MAX; m + 1]; max_k + 1];
+    dp[0][0] = 0.0;
+    for k in 1..=max_k {
+        for j in 1..=m {
+            for i in 0..j {
+                if !dp[k - 1][i].is_finite() {
+                    continue;
+                }
+                let c = cost(i, j);
+                if c.is_finite() && dp[k - 1][i] + c < dp[k][j] {
+                    dp[k][j] = dp[k - 1][i] + c;
+                    back[k][j] = i;
+                }
+            }
+        }
+    }
+
+    let rebuild = |k: usize| -> Option<Vec<(usize, usize)>> {
+        let mut parts = Vec::new();
+        let mut kk = k;
+        let mut j = m;
+        while kk > 0 {
+            let i = back[kk][j];
+            if i == usize::MAX {
+                return None;
+            }
+            parts.push((i, j));
+            j = i;
+            kk -= 1;
+        }
+        parts.reverse();
+        Some(parts)
+    };
+
+    let mut best: Option<Vec<(usize, usize)>> = None;
+    let mut best_good_vol = -1.0_f64;
+    let mut best_k = usize::MAX;
+    for k in 1..=max_k {
+        if !dp[k][m].is_finite() {
+            continue;
+        }
+        let Some(parts) = rebuild(k) else {
+            continue;
+        };
+        let good_vol = parts
+            .iter()
+            .filter_map(|(p, q)| {
+                let s = range_stats(*p, *q);
+                if grade_segment(s, &buckets[*p..*q]) == Verdict::Good {
+                    Some(s.sx)
+                } else {
+                    None
+                }
+            })
+            .sum::<f64>();
+        if good_vol > best_good_vol + 0.01
+            || ((good_vol - best_good_vol).abs() <= 0.01 && k < best_k)
+        {
+            best_good_vol = good_vol;
+            best_k = k;
+            best = Some(parts);
+        }
+    }
+    best.unwrap_or_else(|| vec![(0, m)])
+}
+
+fn samples_for_range(buckets: &[Bucket], lo: f64, hi: f64) -> Vec<Sample> {
+    let mut out = Vec::new();
+    for b in buckets {
+        for p in &b.samples {
+            if p.x >= lo && p.x < hi {
+                out.push(*p);
+            }
+        }
+    }
+    out
+}
+
+fn bucket_range(bucket: i32) -> (f64, f64) {
+    (
+        10_f64.powf(f64::from(bucket) / BUCKETS_PER_DECADE),
+        10_f64.powf(f64::from(bucket + 1) / BUCKETS_PER_DECADE),
+    )
+}
+
+async fn insert_models(
+    cfg: &ClickHouseAnalyticsConfig,
+    connector: &str,
+    account: &str,
+    merchant_id: &str,
+    report_date: &str,
+    rows: &[ModelRow],
+) -> Result<(), IngestError> {
+    if rows.is_empty() {
+        return Ok(());
+    }
+    let mut body = String::with_capacity(rows.len() * 512);
+    for r in rows {
+        let obj = json!({
+            "report_date": report_date,
+            "connector": connector,
+            "account": account,
+            "report_account": &r.key.report_account,
+            "merchant_id": merchant_id,
+            "card_network": &r.key.card_network,
+            "variant": &r.key.variant,
+            "funding": &r.key.funding,
+            "issuer_country": &r.key.issuer_country,
+            "currency": &r.key.currency,
+            "ic_category": &r.key.ic_category,
+            "interchange_bps": &r.key.interchange_bps,
+            "segment_idx": r.segment_idx,
+            "amount_lo": clean_float(r.amount_lo),
+            "amount_hi": clean_float(r.amount_hi),
+            "pct_bps": clean_float(r.pct_bps),
+            "fixed": clean_float(r.fixed),
+            "n": r.n,
+            "gross_sum": clean_float(r.gross_sum),
+            "bps_rmse": clean_float(r.bps_rmse),
+            "grade_bps": clean_float(r.grade_bps),
+            "pct_ci95_bps": clean_float(r.pct_ci95_bps),
+            "crossover_amount": clean_float(r.crossover_amount),
+            "prop_bps": clean_float(r.prop_bps),
+            "fix_abs": clean_float(r.fix_abs),
+            "fix_bps": clean_float(r.fix_bps),
+            "below_gross_frac": clean_float(r.below_gross_frac),
+            "fan_frac": clean_float(r.fan_frac),
+            "fan_money_bps": clean_float(r.fan_money_bps),
+            "r2": clean_float(r.r2),
+            "verdict": r.verdict.as_str(),
+        });
+        body.push_str(
+            &serde_json::to_string(&obj).map_err(|e| IngestError::Storage(e.to_string()))?,
+        );
+        body.push('\n');
+    }
+
+    let query = format!(
+        "INSERT INTO {}.cost_fee_model ({INSERT_COLUMNS}) FORMAT JSONEachRow",
+        cfg.database
+    );
+    let mut req = client()
+        .post(cfg.url.trim_end_matches('/'))
+        .query(&[("query", query.as_str())])
+        .body(body);
+    if !cfg.user.is_empty() {
+        req = req.basic_auth(&cfg.user, cfg.password.as_ref().map(|p| p.peek().clone()));
+    }
+    let resp = req
+        .send()
+        .await
+        .map_err(|e| IngestError::Storage(e.to_string()))?;
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let text = resp.text().await.unwrap_or_default();
+        return Err(IngestError::Storage(format!(
+            "clickhouse model insert failed ({status}): {text}"
+        )));
+    }
+    Ok(())
+}
+
+fn clean_float(v: f64) -> f64 {
+    if v.is_finite() {
+        v
+    } else {
+        0.0
+    }
+}
+
+fn f64_from_u64(n: u64) -> f64 {
+    n.to_string().parse::<f64>().unwrap_or(0.0)
+}
+
+/// Whether a fit result should trigger the empty-refit purge ([`PURGE_MODEL_SQL`]). Extracted as a
+/// pure function because it is the one decision coupled to a destructive `DELETE`, so it is locked
+/// in by unit tests.
+fn should_purge_empty(
+    total_parsed: Option<u64>,
+    connector: &str,
+    account: &str,
+    merchant_id: &str,
+) -> bool {
+    total_parsed == Some(0)
+        && !connector.is_empty()
+        && !account.is_empty()
+        && !merchant_id.is_empty()
+}
+
 /// POST a query to ClickHouse with `{name:Type}` parameters bound as `param_<name>`.
 async fn exec(
     cfg: &ClickHouseAnalyticsConfig,
     query: &str,
     params: &[(&str, String)],
 ) -> Result<String, IngestError> {
-    // The SQL goes in the request body (guarantees a Content-Length; ClickHouse rejects a
-    // body-less POST with 411). Only the `{name:Type}` bindings ride in the query string.
     let q: Vec<(String, String)> = params
         .iter()
         .map(|(k, v)| (format!("param_{k}"), v.clone()))
@@ -314,14 +1030,126 @@ async fn exec(
 
 #[cfg(test)]
 mod tests {
-    use super::should_purge_empty;
+    use super::*;
+
+    fn stats(points: &[(f64, f64)]) -> FitStats {
+        let mut out = FitStats::default();
+        for (x, y) in points {
+            let inv = 1.0 / x;
+            out.n += 1;
+            out.sx += x;
+            out.sy += y;
+            out.sxx += x * x;
+            out.sxy += x * y;
+            out.syy += y * y;
+            out.su += inv;
+            out.suu += inv * inv;
+            out.suy += y * inv;
+            out.suuy += y * inv * inv;
+            out.syyuu += y * y * inv * inv;
+        }
+        out
+    }
+
+    fn bucket(fit_bucket: i32, points: &[(f64, f64)]) -> Bucket {
+        Bucket {
+            fit_bucket,
+            stats: stats(points),
+            samples: points
+                .iter()
+                .map(|(x, y)| Sample { x: *x, y: *y })
+                .collect(),
+        }
+    }
 
-    // The purge is a table-scoped DELETE, so its trigger must be exact. These tests lock in that
-    // it fires on — and only on — a definitively-parsed zero cluster count with a full identifier.
+    #[test]
+    fn fit_recovers_linear_rate_and_fixed_fee() {
+        let pts = [(100.0, 2.7), (200.0, 5.2), (300.0, 7.7)];
+        let fit = fit_stats(stats(&pts));
+        assert!((fit.slope * 10_000.0 - 250.0).abs() < 1e-9);
+        assert!((fit.intercept - 0.2).abs() < 1e-9);
+        assert!(fit.bps_rmse < 1e-5);
+    }
+
+    #[test]
+    fn decomposition_rescues_fixed_fee_tail() {
+        let mut buckets = Vec::new();
+        let low: Vec<(f64, f64)> = (0..20)
+            .map(|i| {
+                let x = 1.0 + f64::from(i) * 0.01;
+                (x, 0.2 + 0.02 * x)
+            })
+            .collect();
+        let high: Vec<(f64, f64)> = (0..220)
+            .map(|i| {
+                let x = 50.0 + f64::from(i);
+                (x, 0.2 + 0.02 * x)
+            })
+            .collect();
+        buckets.push(bucket(0, &low));
+        buckets.push(bucket(17, &high));
+        let whole = merge_bucket_stats(&buckets);
+        let fit = fit_stats(whole);
+        let dec = decompose(whole, &buckets, fit.slope, fit.intercept);
+        assert_eq!(grade_decomposed(dec), Verdict::Good);
+        assert!(dec.below_frac < FIX_VOL_TOL);
+    }
+
+    #[test]
+    fn l2_promotes_small_precise_segment() {
+        let pts: Vec<(f64, f64)> = (0..40)
+            .map(|i| {
+                let x = 100.0 + f64::from(i);
+                (x, 0.1 + 0.015 * x)
+            })
+            .collect();
+        let b = vec![bucket(20, &pts)];
+        assert_eq!(grade_segment(merge_bucket_stats(&b), &b), Verdict::Good);
+    }
+
+    #[test]
+    fn fan_detector_flags_offline_minority() {
+        let mut samples: Vec<Sample> = (0..200)
+            .map(|i| {
+                let x = 100.0 + f64::from(i);
+                Sample {
+                    x,
+                    y: 0.2 + 0.02 * x,
+                }
+            })
+            .collect();
+        for i in 0..6 {
+            let x = 5_000.0 + f64::from(i);
+            samples.push(Sample {
+                x,
+                y: 0.2 + 0.12 * x,
+            });
+        }
+        assert!(dispersion(&samples, 0.02, 0.2) > FAN_FRAC);
+        assert!(money_bps(&samples, 0.02, 0.2, 0.0) > FAN_MONEY_SEVERE);
+    }
+
+    #[test]
+    fn segment_partition_recovers_two_amount_tiers() {
+        let low: Vec<(f64, f64)> = (0..80)
+            .map(|i| {
+                let x = 10.0 + f64::from(i) * 0.2;
+                (x, 0.1 + 0.01 * x)
+            })
+            .collect();
+        let high: Vec<(f64, f64)> = (0..80)
+            .map(|i| {
+                let x = 100.0 + f64::from(i);
+                (x, 0.1 + 0.03 * x)
+            })
+            .collect();
+        let buckets = vec![bucket(10, &low), bucket(20, &high)];
+        let parts = segment_partitions(&buckets);
+        assert!(parts.len() >= 2);
+    }
 
     #[test]
     fn purges_on_definitive_zero() {
-        // The one case that should purge: the fit ran and produced zero clusters (data is gone).
         assert!(should_purge_empty(Some(0), "adyen", "acc", "m1"));
     }
 
@@ -333,14 +1161,11 @@ mod tests {
 
     #[test]
     fn never_purges_on_parse_failure() {
-        // A malformed/empty summary response parses to None; it must NOT look like "0 clusters" and
-        // delete a healthy model. This is the sharp edge the strict parse closed.
         assert!(!should_purge_empty(None, "adyen", "acc", "m1"));
     }
 
     #[test]
     fn never_purges_with_blank_identifier() {
-        // A blank identifier must never widen the delete's scope, even on a genuine zero count.
         assert!(!should_purge_empty(Some(0), "", "acc", "m1"));
         assert!(!should_purge_empty(Some(0), "adyen", "", "m1"));
         assert!(!should_purge_empty(Some(0), "adyen", "acc", ""));
diff --git a/src/cost_ingestion/overrides.rs b/src/cost_ingestion/overrides.rs
index 65b2945a..95fde456 100644
--- a/src/cost_ingestion/overrides.rs
+++ b/src/cost_ingestion/overrides.rs
@@ -39,15 +39,29 @@ pub struct ClusterDims {
     pub issuer_country: String,
     pub currency: String,
     pub ic_category: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub interchange_bps: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub segment_idx: Option<u16>,
 }
 
 impl ClusterDims {
-    /// Parse the `connector|network|variant|funding|issuer|currency|ic_category` key used on the wire
-    /// (URL path). `ic_category` may legitimately be empty (flat-fee clusters), so we split on exactly
-    /// the seven fields and allow a trailing empty last segment.
+    /// Parse cluster override keys. Backward-compatible forms:
+    /// - 7 fields: legacy unsegmented key.
+    /// - 8 fields: legacy segmented key (`...|ic_category|segment_idx`).
+    /// - 9 fields: IC-rate segmented key (`...|ic_category|interchange_bps|segment_idx`).
     pub fn from_key(key: &str) -> Option<Self> {
         let p: Vec<&str> = key.split('|').collect();
-        if p.len() != 7 {
+        if !matches!(p.len(), 7 | 8 | 9) {
+            return None;
+        }
+        let (interchange_bps, segment_idx) = match p.len() {
+            7 => (None, None),
+            8 => (None, p[7].parse::<u16>().ok()),
+            9 => (Some(p[7].to_lowercase()), p[8].parse::<u16>().ok()),
+            _ => (None, None),
+        };
+        if matches!(p.len(), 8 | 9) && segment_idx.is_none() {
             return None;
         }
         Some(Self {
@@ -58,10 +72,30 @@ impl ClusterDims {
             issuer_country: p[4].to_lowercase(),
             currency: p[5].to_lowercase(),
             ic_category: p[6].to_lowercase(),
+            interchange_bps,
+            segment_idx,
         })
     }
 }
 
+pub fn key_of_dims(d: &ClusterDims) -> String {
+    let base = format!(
+        "{}|{}|{}|{}|{}|{}|{}",
+        d.connector.to_lowercase(),
+        d.card_network.to_lowercase(),
+        d.variant.to_lowercase(),
+        d.funding.to_lowercase(),
+        d.issuer_country.to_lowercase(),
+        d.currency.to_lowercase(),
+        d.ic_category.to_lowercase(),
+    );
+    match (&d.interchange_bps, d.segment_idx) {
+        (Some(rate), Some(idx)) => format!("{base}|{}|{idx}", rate.to_lowercase()),
+        (None, Some(idx)) => format!("{base}|{idx}"),
+        _ => base,
+    }
+}
+
 /// A merchant-authored fee for one specific fitted cluster. Highest precedence at lookup (see
 /// `serving::lookup`): cluster override > connector override > learned model.
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/src/cost_ingestion/rollup.rs b/src/cost_ingestion/rollup.rs
index 69ba5cb8..934efddd 100644
--- a/src/cost_ingestion/rollup.rs
+++ b/src/cost_ingestion/rollup.rs
@@ -2,7 +2,7 @@
 //!
 //! Individual transactions are never stored. As a report streams in (batch by batch, off the
 //! connector parser), each fee-bearing transaction is folded into one bucket keyed by
-//! `(cluster × transaction-day × amount-band × channel)`. A bucket accumulates the additive sums an
+//! `(cluster × transaction-day × predictor-band × fit-bucket × channel)`. A bucket accumulates the additive sums an
 //! OLS fit needs — `n, Σx, Σy, Σx², Σxy, Σy²` and the reciprocal terms for the bps-RMSE /
 //! NON_LINEAR check — so summing buckets over any window reconstructs the exact same line the raw
 //! rows would give (see `scratch/settlement-table-removal-worked-example.md`).
@@ -13,27 +13,30 @@
 use std::collections::HashMap;
 
 use chrono::NaiveDate;
+use rand::{rngs::StdRng, Rng, SeedableRng};
 
-use super::types::{amount_band, SettledFeeRow};
+use super::types::{amount_band, fit_bucket, SettledFeeRow};
 
-/// The €5 micro-amount floor: transactions below this are excluded from the fit and predictor, so
-/// they never enter a bucket. Applied here at aggregation time (it cannot be recovered later).
-/// Mirrors the `WHERE gross >= 5` that the fit and predictor queries used to apply against raw rows.
-const MICRO_AMOUNT_FLOOR: f64 = 5.0;
+/// Per-bucket bounded sample used by the fitter's fan detector. The sufficient statistics remain
+/// authoritative for the OLS fit; samples only answer "does a minority of rows sit far off-line?".
+const SAMPLE_CAP_PER_BUCKET: usize = 64;
 
 /// Identity of one rollup bucket. `band`/`channel` are predictor features the fit sums away; the
 /// rest is the fit's cluster key plus the transaction day.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct BucketKey {
     txn_date: NaiveDate,
+    report_account: String,
     card_network: String,
     variant: String,
     funding: String,
     issuer_country: String,
     currency: String,
     ic_category: String,
+    interchange_bps: String,
     channel: String,
     band: &'static str,
+    fit_bucket: i32,
 }
 
 /// Additive sufficient statistics for the transactions in one bucket. Every field is a plain sum,
@@ -53,6 +56,12 @@ struct Stats {
     syyuu: f64,
 }
 
+#[derive(Debug, Clone, Default)]
+struct BucketStats {
+    sums: Stats,
+    samples: Vec<(f64, f64)>,
+}
+
 impl Stats {
     /// Fold one transaction (gross `x`, fee `y`) into the sums. Caller guarantees `x >= floor > 0`.
     fn add(&mut self, x: f64, y: f64) {
@@ -72,17 +81,37 @@ impl Stats {
     }
 }
 
+impl BucketStats {
+    fn add(&mut self, x: f64, y: f64, rng: &mut StdRng) {
+        let seen = self.sums.n;
+        self.sums.add(x, y);
+        if self.samples.len() < SAMPLE_CAP_PER_BUCKET {
+            self.samples.push((x, y));
+            return;
+        }
+        let j = rng.gen_range(0..=seen);
+        if let Ok(idx) = usize::try_from(j) {
+            if idx < SAMPLE_CAP_PER_BUCKET {
+                self.samples[idx] = (x, y);
+            }
+        }
+    }
+}
+
 /// One fully-aggregated bucket, ready to insert into `cost_daily_stats`.
 pub struct DailyStatRow {
     pub txn_date: NaiveDate,
+    pub report_account: String,
     pub card_network: String,
     pub variant: String,
     pub funding: String,
     pub issuer_country: String,
     pub currency: String,
     pub ic_category: String,
+    pub interchange_bps: String,
     pub channel: String,
     pub band: &'static str,
+    pub fit_bucket: i32,
     pub n: u64,
     pub sx: f64,
     pub sy: f64,
@@ -94,41 +123,56 @@ pub struct DailyStatRow {
     pub suy: f64,
     pub suuy: f64,
     pub syyuu: f64,
+    pub sample_x: Vec<f64>,
+    pub sample_y: Vec<f64>,
 }
 
 /// Accumulates a report's transactions into per-day sufficient statistics.
-#[derive(Default)]
 pub struct RollupAccumulator {
-    buckets: HashMap<BucketKey, Stats>,
+    buckets: HashMap<BucketKey, BucketStats>,
+    rng: StdRng,
+}
+
+impl Default for RollupAccumulator {
+    fn default() -> Self {
+        Self::new()
+    }
 }
 
 impl RollupAccumulator {
     pub fn new() -> Self {
-        Self::default()
+        Self {
+            buckets: HashMap::new(),
+            rng: StdRng::seed_from_u64(7),
+        }
     }
 
-    /// Fold one transaction. Rows below the micro-amount floor (or with non-positive gross, which
-    /// would make the reciprocal terms explode) are skipped — the same rows the fit/predictor
-    /// filtered out at read time. `fallback_date` dates rows whose report carried no txn date.
+    /// Fold one transaction. Non-positive/invalid gross is skipped because reciprocal fit terms
+    /// require `gross > 0`; fee-bearing micro transactions are retained and judged by fit quality
+    /// rather than a currency-blind amount floor. `fallback_date` dates rows whose report carried
+    /// no txn date.
     pub fn add_row(&mut self, row: &SettledFeeRow, fallback_date: NaiveDate) {
-        if row.gross.is_nan() || row.gross < MICRO_AMOUNT_FLOOR {
+        if !row.gross.is_finite() || row.gross <= 0.0 || !row.total_fee.is_finite() {
             return;
         }
         let key = BucketKey {
             txn_date: row.txn_date.unwrap_or(fallback_date),
+            report_account: row.report_account.clone(),
             card_network: row.card_network.clone(),
             variant: row.variant.clone(),
             funding: row.funding.clone(),
             issuer_country: row.issuer_country.clone(),
             currency: row.currency.clone(),
             ic_category: row.ic_category.clone(),
+            interchange_bps: row.interchange_bps.clone(),
             channel: row.channel.clone(),
             band: amount_band(row.gross),
+            fit_bucket: fit_bucket(row.gross),
         };
         self.buckets
             .entry(key)
             .or_default()
-            .add(row.gross, row.total_fee);
+            .add(row.gross, row.total_fee, &mut self.rng);
     }
 
     /// Number of distinct buckets accumulated (for capacity hints / diagnostics).
@@ -146,25 +190,30 @@ impl RollupAccumulator {
             .into_iter()
             .map(|(k, s)| DailyStatRow {
                 txn_date: k.txn_date,
+                report_account: k.report_account,
                 card_network: k.card_network,
                 variant: k.variant,
                 funding: k.funding,
                 issuer_country: k.issuer_country,
                 currency: k.currency,
                 ic_category: k.ic_category,
+                interchange_bps: k.interchange_bps,
                 channel: k.channel,
                 band: k.band,
-                n: s.n,
-                sx: s.sx,
-                sy: s.sy,
-                sxx: s.sxx,
-                sxy: s.sxy,
-                syy: s.syy,
-                su: s.su,
-                suu: s.suu,
-                suy: s.suy,
-                suuy: s.suuy,
-                syyuu: s.syyuu,
+                fit_bucket: k.fit_bucket,
+                n: s.sums.n,
+                sx: s.sums.sx,
+                sy: s.sums.sy,
+                sxx: s.sums.sxx,
+                sxy: s.sums.sxy,
+                syy: s.sums.syy,
+                su: s.sums.su,
+                suu: s.sums.suu,
+                suy: s.sums.suy,
+                suuy: s.sums.suuy,
+                syyuu: s.sums.syyuu,
+                sample_x: s.samples.iter().map(|(x, _)| *x).collect(),
+                sample_y: s.samples.iter().map(|(_, y)| *y).collect(),
             })
             .collect()
     }
@@ -177,12 +226,14 @@ mod tests {
     fn row(gross: f64, fee: f64, date: &str) -> SettledFeeRow {
         SettledFeeRow {
             txn_ref: String::new(),
+            report_account: String::new(),
             card_network: "visa".into(),
             variant: "visacredit".into(),
             funding: "credit".into(),
             issuer_country: "GB".into(),
             currency: "GBP".into(),
             ic_category: "".into(),
+            interchange_bps: "".into(),
             txn_date: Some(NaiveDate::parse_from_str(date, "%Y-%m-%d").unwrap()),
             channel: "ecom".into(),
             gross,
@@ -195,11 +246,14 @@ mod tests {
     }
 
     #[test]
-    fn floors_micro_amounts() {
+    fn keeps_positive_micro_amounts() {
         let mut acc = RollupAccumulator::new();
         let d = NaiveDate::parse_from_str("2026-06-28", "%Y-%m-%d").unwrap();
         acc.add_row(&row(4.99, 0.5, "2026-06-28"), d);
-        assert!(acc.is_empty(), "sub-floor txn must not create a bucket");
+        assert!(
+            !acc.is_empty(),
+            "positive micro txn should still create a bucket"
+        );
     }
 
     #[test]
diff --git a/src/cost_ingestion/serving.rs b/src/cost_ingestion/serving.rs
index b2067bdf..eebf8b17 100644
--- a/src/cost_ingestion/serving.rs
+++ b/src/cost_ingestion/serving.rs
@@ -35,6 +35,20 @@ struct ServingCost {
     fixed: f64,
 }
 
+#[derive(Debug, Clone, Copy)]
+struct ServingSegment {
+    cost: ServingCost,
+    segment_idx: u16,
+    amount_lo: f64,
+    amount_hi: f64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, std::hash::Hash)]
+struct PredictedIc {
+    category: String,
+    interchange_bps: String,
+}
+
 impl ServingCost {
     fn effective_cost_bps(&self, amount: f64) -> f64 {
         if amount > 0.0 {
@@ -50,11 +64,11 @@ impl ServingCost {
 #[derive(Default, Clone)]
 struct MerchantModels {
     /// `connector|network|funding|currency|region` → blended cost (graceful fallback).
-    coarse: HashMap<String, ServingCost>,
-    /// `connector|network|variant|funding|issuer|currency|ic_category` → specific cluster cost.
-    fine: HashMap<String, ServingCost>,
-    /// Predictor back-off levels (most specific first): level-key → modal `ic_category`.
-    predictor: Vec<HashMap<String, String>>,
+    coarse: HashMap<String, Vec<ServingSegment>>,
+    /// `connector|network|variant|funding|issuer|currency|ic_category|ic_rate` → fitted segments.
+    fine: HashMap<String, Vec<ServingSegment>>,
+    /// Predictor back-off levels (most specific first): level-key → modal category/rate pair.
+    predictor: Vec<HashMap<String, PredictedIc>>,
     /// Manual per-connector blended-fee overrides (lowercase connector → flat cost). When present
     /// for a connector, it wins over the learned model at [`lookup`] — the merchant told us the
     /// contract rate, so every EV calculation on that connector uses it.
@@ -143,9 +157,10 @@ fn fine_key(
     issuer: &str,
     currency: &str,
     ic_category: &str,
+    interchange_bps: &str,
 ) -> String {
     format!(
-        "{}|{}|{}|{}|{}|{}|{}",
+        "{}|{}|{}|{}|{}|{}|{}|{}",
         connector.to_lowercase(),
         normalize_network(&network.to_lowercase()),
         variant.to_lowercase(),
@@ -153,9 +168,51 @@ fn fine_key(
         issuer.to_lowercase(),
         currency.to_lowercase(),
         ic_category.to_lowercase(),
+        interchange_bps.to_lowercase(),
     )
 }
 
+#[allow(clippy::too_many_arguments)]
+fn override_keys(
+    connector: &str,
+    network: &str,
+    variant: &str,
+    funding: &str,
+    issuer: &str,
+    currency: &str,
+    ic_category: &str,
+    interchange_bps: &str,
+    segment_idx: Option<u16>,
+) -> Vec<String> {
+    let base = fine_key(
+        connector,
+        network,
+        variant,
+        funding,
+        issuer,
+        currency,
+        ic_category,
+        interchange_bps,
+    );
+    let legacy = format!(
+        "{}|{}|{}|{}|{}|{}|{}",
+        connector.to_lowercase(),
+        normalize_network(&network.to_lowercase()),
+        variant.to_lowercase(),
+        funding.to_lowercase(),
+        issuer.to_lowercase(),
+        currency.to_lowercase(),
+        ic_category.to_lowercase(),
+    );
+    let mut out = Vec::new();
+    if let Some(idx) = segment_idx {
+        out.push(format!("{base}|{idx}"));
+        out.push(format!("{legacy}|{idx}"));
+    }
+    out.push(legacy);
+    out
+}
+
 /// Reconstruct the report's `variant` string from decide-time card attributes
 /// (`visa` + `standard` + `debit` → `visastandarddebit`). A wallet is its own variant in the report
 /// (`visa_applepay`), so it takes precedence over the network+program+funding form.
@@ -219,6 +276,10 @@ pub struct InhouseMatch {
     pub variant: Option<String>,
     pub issuer: Option<String>,
     pub ic_category: Option<String>,
+    pub interchange_bps: Option<String>,
+    pub segment_idx: Option<u16>,
+    pub amount_lo: Option<f64>,
+    pub amount_hi: Option<f64>,
 }
 
 /// Look up an in-house cost at decide time. Tries the fine, category-predicted cluster first, then
@@ -242,35 +303,63 @@ pub fn lookup(
     let snapshot = cache().read().ok()?.clone();
     let m = snapshot.get(merchant_id)?;
 
-    // Resolve the fine cluster key once (needs a raw issuer + a predicted interchange category).
-    // Reused for both the highest-precedence cluster-override check and the learned fine-model
-    // lookup, so the two can never disagree on which cluster this transaction is.
+    let brand = normalize_network(&network.to_lowercase()).to_string();
+    let connector_l = connector.to_lowercase();
+
+    // Resolve the fine cluster key once (needs a raw issuer + predicted category/rate pair).
+    // The same predicted pair is used for overrides and learned models, so they cannot drift.
     let fine = if issuer.is_empty() {
         None
     } else {
         let variant = reconstruct_variant(network, program, funding, wallet);
         let band = amount_band(amount);
-        predict_category(m, network, &variant, funding, issuer, band, channel).map(|cat| {
+        predict_category(m, network, &variant, funding, issuer, band, channel).map(|pred| {
             let key = fine_key(
-                connector, network, &variant, funding, issuer, currency, &cat,
+                connector,
+                network,
+                &variant,
+                funding,
+                issuer,
+                currency,
+                &pred.category,
+                &pred.interchange_bps,
             );
-            (key, variant, cat)
+            (key, variant, pred)
         })
     };
 
-    // 1. Cluster override — the merchant set a fee for this exact segment (including its card
-    //    program). Most specific, wins over everything (connector override + learned model).
-    if let Some((key, variant, cat)) = &fine {
-        if let Some(cost) = m.cluster_overrides.get(key) {
+    let fine_segment = fine
+        .as_ref()
+        .and_then(|(key, _, _)| m.fine.get(key))
+        .and_then(|segments| pick_segment(segments, amount));
+
+    // 1. Cluster override — exact segment first, then legacy unsegmented keys.
+    if let Some((_, variant, pred)) = &fine {
+        let keys = override_keys(
+            connector,
+            network,
+            variant,
+            funding,
+            issuer,
+            currency,
+            &pred.category,
+            &pred.interchange_bps,
+            fine_segment.map(|s| s.segment_idx),
+        );
+        if let Some(cost) = keys.iter().find_map(|key| m.cluster_overrides.get(key)) {
             return Some(InhouseMatch {
                 effective_bps: cost.effective_cost_bps(amount),
                 pct_bps: cost.pct_bps,
                 fixed: cost.fixed,
-                brand: normalize_network(&network.to_lowercase()).to_string(),
+                brand,
                 currency: currency.to_uppercase(),
                 variant: Some(variant.clone()),
                 issuer: Some(issuer.to_uppercase()),
-                ic_category: Some(cat.clone()),
+                ic_category: Some(pred.category.clone()),
+                interchange_bps: Some(pred.interchange_bps.clone()),
+                segment_idx: fine_segment.map(|s| s.segment_idx),
+                amount_lo: fine_segment.map(|s| s.amount_lo),
+                amount_hi: fine_segment.map(|s| s.amount_hi),
             });
         }
     }
@@ -278,58 +367,89 @@ pub fn lookup(
     // 2. Connector override: the merchant gave us this connector's blanket contract rate, so use it
     //    flat for every transaction not covered by a cluster override above. `connector` is already
     //    lowercased by the caller; lowercase again defensively so the key always matches.
-    if let Some(cost) = m.overrides.get(&connector.to_lowercase()) {
+    if let Some(cost) = m.overrides.get(&connector_l) {
         return Some(InhouseMatch {
             effective_bps: cost.effective_cost_bps(amount),
             pct_bps: cost.pct_bps,
             fixed: cost.fixed,
-            brand: normalize_network(&network.to_lowercase()).to_string(),
+            brand,
             currency: currency.to_uppercase(),
             variant: None,
             issuer: None,
             ic_category: None,
+            interchange_bps: None,
+            segment_idx: None,
+            amount_lo: None,
+            amount_hi: None,
         });
     }
 
     // The invoice-derived add-on for this connector (if any), layered onto the *learned* models
     // below — never onto the overrides above, which are already all-in contract rates.
-    let addon = m.addons.get(&connector.to_lowercase());
+    let addon = m.addons.get(&connector_l);
 
     // 3. Learned fine model: serve the specific fitted cluster, plus the invoice add-on.
-    if let Some((key, variant, cat)) = &fine {
-        if let Some(cost) = m.fine.get(key) {
-            let cost = cost.with_addon(addon);
-            return Some(InhouseMatch {
-                effective_bps: cost.effective_cost_bps(amount),
-                pct_bps: cost.pct_bps,
-                fixed: cost.fixed,
-                brand: normalize_network(&network.to_lowercase()).to_string(),
-                currency: currency.to_uppercase(),
-                variant: Some(variant.clone()),
-                issuer: Some(issuer.to_uppercase()),
-                ic_category: Some(cat.clone()),
-            });
-        }
+    if let (Some(segment), Some((_, variant, pred))) = (fine_segment, fine.as_ref()) {
+        let cost = segment.cost.with_addon(addon);
+        return Some(InhouseMatch {
+            effective_bps: cost.effective_cost_bps(amount),
+            pct_bps: cost.pct_bps,
+            fixed: cost.fixed,
+            brand,
+            currency: currency.to_uppercase(),
+            variant: Some(variant.clone()),
+            issuer: Some(issuer.to_uppercase()),
+            ic_category: Some(pred.category.clone()),
+            interchange_bps: Some(pred.interchange_bps.clone()),
+            segment_idx: Some(segment.segment_idx),
+            amount_lo: Some(segment.amount_lo),
+            amount_hi: Some(segment.amount_hi),
+        });
     }
 
     // 4. Fallback: the coarse region blend (previous behavior) — no single variant/issuer/category.
     let key = coarse_key(connector, network, funding, currency, region);
-    m.coarse.get(&key).map(|cost| {
-        let cost = cost.with_addon(addon);
-        InhouseMatch {
+    m.coarse.get(&key).and_then(|segments| {
+        let segment = pick_segment(segments, amount)?;
+        let cost = segment.cost.with_addon(addon);
+        Some(InhouseMatch {
             effective_bps: cost.effective_cost_bps(amount),
             pct_bps: cost.pct_bps,
             fixed: cost.fixed,
-            brand: normalize_network(&network.to_lowercase()).to_string(),
+            brand,
             currency: currency.to_uppercase(),
             variant: None,
             issuer: None,
             ic_category: None,
-        }
+            interchange_bps: None,
+            segment_idx: Some(segment.segment_idx),
+            amount_lo: Some(segment.amount_lo),
+            amount_hi: Some(segment.amount_hi),
+        })
     })
 }
 
-/// Predict the interchange category by trying each back-off level, most specific first.
+fn pick_segment(segments: &[ServingSegment], amount: f64) -> Option<ServingSegment> {
+    let mut fallback = None;
+    let mut best = None;
+    let mut best_width = f64::INFINITY;
+    for segment in segments {
+        if segment.amount_lo == 0.0 && segment.amount_hi == 0.0 {
+            fallback = Some(*segment);
+            continue;
+        }
+        if amount >= segment.amount_lo && amount < segment.amount_hi {
+            let width = segment.amount_hi - segment.amount_lo;
+            if width < best_width {
+                best = Some(*segment);
+                best_width = width;
+            }
+        }
+    }
+    best.or(fallback)
+}
+
+/// Predict the interchange category/rate pair by trying each back-off level, most specific first.
 fn predict_category(
     m: &MerchantModels,
     network: &str,
@@ -338,12 +458,12 @@ fn predict_category(
     issuer: &str,
     band: &str,
     channel: &str,
-) -> Option<String> {
+) -> Option<PredictedIc> {
     let keys = predictor_level_keys(network, variant, funding, issuer, band, channel);
     for (i, key) in keys.iter().enumerate() {
         if let Some(table) = m.predictor.get(i) {
-            if let Some(cat) = table.get(key) {
-                return Some(cat.clone());
+            if let Some(pred) = table.get(key) {
+                return Some(pred.clone());
             }
         }
     }
@@ -382,6 +502,7 @@ pub fn spawn(clickhouse: ClickHouseAnalyticsConfig) {
 const COST_SQL: &str = r#"
 SELECT
     merchant_id, connector, card_network, variant, funding, issuer_country, currency, ic_category,
+    interchange_bps, segment_idx, amount_lo, amount_hi,
     sum(pct_bps * gross_sum) AS pct_num,
     sum(fixed * gross_sum)   AS fixed_num,
     sum(gross_sum)           AS w
@@ -390,23 +511,27 @@ WHERE verdict = 'GOOD' AND gross_sum > 0{merchant_filter}
   AND (merchant_id, connector, account, report_date) IN (
       SELECT merchant_id, connector, account, max(report_date)
       FROM __DB__.cost_fee_model{merchant_filter_sub} GROUP BY merchant_id, connector, account)
-GROUP BY merchant_id, connector, card_network, variant, funding, issuer_country, currency, ic_category
+GROUP BY merchant_id, connector, card_network, variant, funding, issuer_country, currency,
+         ic_category, interchange_bps, segment_idx, amount_lo, amount_hi
 FORMAT TSV
 "#;
 
 /// Per-(merchant, network, variant, funding, issuer, band, channel) category counts, for the
 /// predictor. `channel` (pos/ecom) is the strongest disambiguator between in-person and online
 /// interchange categories. `band` is a stored column of the rollup (stamped at ingestion by the same
-/// `amount_band` thresholds this file uses at decide time); the €5 floor was applied at aggregation.
+/// `amount_band` thresholds this file uses at decide time). Positive micro transactions are kept;
+/// fixed-fee tails are handled by the fitter.
 /// `{merchant_filter}` is a `WHERE merchant_id = {merchant:String}` for a single-merchant refresh,
 /// or `""` for the global rebuild.
 const PREDICTOR_SQL: &str = r#"
 SELECT
-    merchant_id, card_network, variant, funding, issuer_country, band, channel, ic_category,
+    merchant_id, card_network, variant, funding, issuer_country, band, channel,
+    ic_category, interchange_bps,
     sum(n) AS c
 FROM __DB__.cost_daily_stats FINAL
 {merchant_filter}
-GROUP BY merchant_id, card_network, variant, funding, issuer_country, band, channel, ic_category
+GROUP BY merchant_id, card_network, variant, funding, issuer_country, band, channel,
+         ic_category, interchange_bps
 FORMAT TSV
 "#;
 
@@ -456,35 +581,48 @@ async fn refresh_inner(
 
     let mut snap: Snapshot = HashMap::new();
 
-    // 1. Cost tables (coarse blend + fine per-category), volume-weighted.
-    let mut coarse_acc: HashMap<String, HashMap<String, (f64, f64, f64)>> = HashMap::new(); // merchant -> key -> (pct_num, fix_num, w)
-    let mut fine_acc: HashMap<String, HashMap<String, (f64, f64, f64)>> = HashMap::new();
+    // 1. Cost tables (coarse blend + fine per-category), volume-weighted per amount segment.
+    let mut coarse_acc: HashMap<String, HashMap<String, HashMap<String, SegmentAcc>>> =
+        HashMap::new();
+    let mut fine_acc: HashMap<String, HashMap<String, HashMap<String, SegmentAcc>>> =
+        HashMap::new();
     for line in cost_rows.lines() {
         let f: Vec<&str> = line.split('\t').collect();
-        if f.len() < 11 {
+        if f.len() < 15 {
             continue;
         }
-        let (merchant, connector, network, variant, funding, issuer, currency, ic) =
-            (f[0], f[1], f[2], f[3], f[4], f[5], f[6], f[7]);
-        let pct_num: f64 = f[8].trim().parse().unwrap_or(0.0);
-        let fix_num: f64 = f[9].trim().parse().unwrap_or(0.0);
-        let w: f64 = f[10].trim().parse().unwrap_or(0.0);
+        let (merchant, connector, network, variant, funding, issuer, currency, ic, ic_bps) =
+            (f[0], f[1], f[2], f[3], f[4], f[5], f[6], f[7], f[8]);
+        let segment_idx: u16 = f[9].trim().parse().unwrap_or(0);
+        let amount_lo: f64 = f[10].trim().parse().unwrap_or(0.0);
+        let amount_hi: f64 = f[11].trim().parse().unwrap_or(0.0);
+        let pct_num: f64 = f[12].trim().parse().unwrap_or(0.0);
+        let fix_num: f64 = f[13].trim().parse().unwrap_or(0.0);
+        let w: f64 = f[14].trim().parse().unwrap_or(0.0);
         if w <= 0.0 {
             continue;
         }
         let region = issuer_region(issuer);
         let ck = coarse_key(connector, network, funding, currency, &region);
-        accumulate(
+        accumulate_segment(
             coarse_acc.entry(merchant.to_string()).or_default(),
             ck,
+            segment_idx,
+            amount_lo,
+            amount_hi,
             pct_num,
             fix_num,
             w,
         );
-        let fk = fine_key(connector, network, variant, funding, issuer, currency, ic);
-        accumulate(
+        let fk = fine_key(
+            connector, network, variant, funding, issuer, currency, ic, ic_bps,
+        );
+        accumulate_segment(
             fine_acc.entry(merchant.to_string()).or_default(),
             fk,
+            segment_idx,
+            amount_lo,
+            amount_hi,
             pct_num,
             fix_num,
             w,
@@ -492,27 +630,32 @@ async fn refresh_inner(
     }
     for (merchant, keys) in coarse_acc {
         let m = snap.entry(merchant).or_default();
-        m.coarse = finalize(keys);
+        m.coarse = finalize_segments(keys);
     }
     for (merchant, keys) in fine_acc {
         let m = snap.entry(merchant).or_default();
-        m.fine = finalize(keys);
+        m.fine = finalize_segments(keys);
     }
 
     // 2. Predictor tables: accumulate category counts per back-off level, keep the modal category
     //    with >= MIN_SUPPORT total observations.
-    let mut pred_acc: HashMap<String, Vec<HashMap<String, HashMap<String, u64>>>> = HashMap::new();
+    let mut pred_acc: HashMap<String, Vec<HashMap<String, HashMap<PredictedIc, u64>>>> =
+        HashMap::new();
     for line in pred_rows.lines() {
         let f: Vec<&str> = line.split('\t').collect();
-        if f.len() < 9 {
+        if f.len() < 10 {
             continue;
         }
-        let (merchant, network, variant, funding, issuer, band, channel, ic) =
-            (f[0], f[1], f[2], f[3], f[4], f[5], f[6], f[7]);
-        let c: u64 = f[8].trim().parse().unwrap_or(0);
+        let (merchant, network, variant, funding, issuer, band, channel, ic, ic_bps) =
+            (f[0], f[1], f[2], f[3], f[4], f[5], f[6], f[7], f[8]);
+        let c: u64 = f[9].trim().parse().unwrap_or(0);
         if c == 0 {
             continue;
         }
+        let pred = PredictedIc {
+            category: ic.to_string(),
+            interchange_bps: ic_bps.to_string(),
+        };
         let levels = pred_acc
             .entry(merchant.to_string())
             .or_insert_with(|| vec![HashMap::new(); PREDICTOR_LEVELS]);
@@ -523,12 +666,12 @@ async fn refresh_inner(
             *levels[i]
                 .entry(key)
                 .or_default()
-                .entry(ic.to_string())
+                .entry(pred.clone())
                 .or_insert(0) += c;
         }
     }
     for (merchant, levels) in pred_acc {
-        let tables: Vec<HashMap<String, String>> = levels
+        let tables: Vec<HashMap<String, PredictedIc>> = levels
             .into_iter()
             .map(|level| {
                 level
@@ -540,7 +683,7 @@ async fn refresh_inner(
                         }
                         cats.into_iter()
                             .max_by_key(|(_, n)| *n)
-                            .map(|(cat, _)| (key, cat))
+                            .map(|(pred, _)| (key, pred))
                     })
                     .collect()
             })
@@ -641,21 +784,13 @@ async fn load_overlays_into(snap: &mut Snapshot, merchant_id: &str) {
         ),
     }
 
-    // Cluster-level overrides, keyed by the same fine_key the lookup builds at decide time.
+    // Cluster-level overrides, keyed by the same wire key the lookup probes at decide time.
     match super::overrides::list_clusters(merchant_id).await {
         Ok(list) if !list.is_empty() => {
             let cluster_overrides = list
                 .into_iter()
                 .map(|c| {
-                    let key = fine_key(
-                        &c.dims.connector,
-                        &c.dims.card_network,
-                        &c.dims.variant,
-                        &c.dims.funding,
-                        &c.dims.issuer_country,
-                        &c.dims.currency,
-                        &c.dims.ic_category,
-                    );
+                    let key = super::overrides::key_of_dims(&c.dims);
                     (
                         key,
                         ServingCost {
@@ -706,30 +841,73 @@ async fn load_overlays_into(snap: &mut Snapshot, merchant_id: &str) {
     }
 }
 
-fn accumulate(
-    map: &mut HashMap<String, (f64, f64, f64)>,
+#[derive(Debug, Clone, Copy)]
+struct SegmentAcc {
+    segment_idx: u16,
+    amount_lo: f64,
+    amount_hi: f64,
+    pct_num: f64,
+    fix_num: f64,
+    w: f64,
+}
+
+fn accumulate_segment(
+    map: &mut HashMap<String, HashMap<String, SegmentAcc>>,
     key: String,
+    segment_idx: u16,
+    amount_lo: f64,
+    amount_hi: f64,
     pct_num: f64,
     fix_num: f64,
     w: f64,
 ) {
-    let e = map.entry(key).or_insert((0.0, 0.0, 0.0));
-    e.0 += pct_num;
-    e.1 += fix_num;
-    e.2 += w;
+    let segment_key = format!("{segment_idx}|{amount_lo:.12}|{amount_hi:.12}");
+    let e = map
+        .entry(key)
+        .or_default()
+        .entry(segment_key)
+        .or_insert(SegmentAcc {
+            segment_idx,
+            amount_lo,
+            amount_hi,
+            pct_num: 0.0,
+            fix_num: 0.0,
+            w: 0.0,
+        });
+    e.pct_num += pct_num;
+    e.fix_num += fix_num;
+    e.w += w;
 }
 
-fn finalize(keys: HashMap<String, (f64, f64, f64)>) -> HashMap<String, ServingCost> {
+fn finalize_segments(
+    keys: HashMap<String, HashMap<String, SegmentAcc>>,
+) -> HashMap<String, Vec<ServingSegment>> {
     keys.into_iter()
-        .filter(|(_, (_, _, w))| *w > 0.0)
-        .map(|(k, (pn, fn_, w))| {
-            (
-                k,
-                ServingCost {
-                    pct_bps: pn / w,
-                    fixed: fn_ / w,
-                },
-            )
+        .filter_map(|(k, segments)| {
+            let mut out: Vec<ServingSegment> = segments
+                .into_values()
+                .filter(|s| s.w > 0.0)
+                .map(|s| ServingSegment {
+                    cost: ServingCost {
+                        pct_bps: s.pct_num / s.w,
+                        fixed: s.fix_num / s.w,
+                    },
+                    segment_idx: s.segment_idx,
+                    amount_lo: s.amount_lo,
+                    amount_hi: s.amount_hi,
+                })
+                .collect();
+            out.sort_by(|a, b| {
+                a.amount_lo
+                    .total_cmp(&b.amount_lo)
+                    .then(a.amount_hi.total_cmp(&b.amount_hi))
+                    .then(a.segment_idx.cmp(&b.segment_idx))
+            });
+            if out.is_empty() {
+                None
+            } else {
+                Some((k, out))
+            }
         })
         .collect()
 }
diff --git a/src/cost_ingestion/sink.rs b/src/cost_ingestion/sink.rs
index bc2f9400..e4469196 100644
--- a/src/cost_ingestion/sink.rs
+++ b/src/cost_ingestion/sink.rs
@@ -33,8 +33,9 @@ const INSERT_CHUNK_ROWS: usize = 25_000;
 
 /// Columns we provide; `ingested_at` is intentionally omitted so ClickHouse applies its DEFAULT.
 const COLUMNS: &str =
-    "connector,account,merchant_id,txn_date,ingestion_id,card_network,variant,funding,\
-issuer_country,currency,ic_category,channel,band,n,sx,sy,sxx,sxy,syy,su,suu,suy,suuy,syyuu";
+    "connector,account,report_account,merchant_id,txn_date,ingestion_id,card_network,variant,funding,\
+issuer_country,currency,ic_category,interchange_bps,channel,band,fit_bucket,n,sx,sy,sxx,sxy,\
+syy,su,suu,suy,suuy,syyuu,sample_x,sample_y";
 
 fn client() -> &'static reqwest::Client {
     static CLIENT: OnceLock<reqwest::Client> = OnceLock::new();
@@ -83,6 +84,7 @@ async fn insert_chunk(
         let obj = json!({
             "connector": connector,
             "account": account,
+            "report_account": r.report_account,
             "merchant_id": merchant_id,
             "txn_date": r.txn_date.to_string(),
             "ingestion_id": ingestion_id,
@@ -92,8 +94,10 @@ async fn insert_chunk(
             "issuer_country": r.issuer_country,
             "currency": r.currency,
             "ic_category": r.ic_category,
+            "interchange_bps": r.interchange_bps,
             "channel": r.channel,
             "band": r.band,
+            "fit_bucket": r.fit_bucket,
             "n": r.n,
             "sx": r.sx,
             "sy": r.sy,
@@ -105,6 +109,8 @@ async fn insert_chunk(
             "suy": r.suy,
             "suuy": r.suuy,
             "syyuu": r.syyuu,
+            "sample_x": r.sample_x,
+            "sample_y": r.sample_y,
         });
         body.push_str(
             &serde_json::to_string(&obj).map_err(|e| IngestError::Storage(e.to_string()))?,
diff --git a/src/cost_ingestion/types.rs b/src/cost_ingestion/types.rs
index 62d382cb..8091b711 100644
--- a/src/cost_ingestion/types.rs
+++ b/src/cost_ingestion/types.rs
@@ -18,6 +18,10 @@ pub struct SettledFeeRow {
     /// aggregates by the cluster fields below (read straight off this struct) and nothing
     /// downstream reads `txn_ref`, so it is NOT a dedup key.
     pub txn_ref: String,
+    /// Connector-native merchant/account inside the report, when present. For Adyen this is the CSV
+    /// `Merchant Account`, which separates POS/ecom/country accounts inside one uploaded report.
+    /// Blank for connectors whose reports do not expose an internal account dimension.
+    pub report_account: String,
     /// Card network, lowercased: `visa`, `mc`, …
     pub card_network: String,
     /// Payment-method variant (carries tier + funding), lowercased: `visastandarddebit`, …
@@ -30,6 +34,10 @@ pub struct SettledFeeRow {
     pub currency: String,
     /// Interchange category from the report; `""` for flat-fee methods (iDEAL/Klarna/CB).
     pub ic_category: String,
+    /// Interchange rate in basis points when the report exposes it (Adyen `ICSF details[].bps`).
+    /// Empty for connectors that do not provide a comparable card-product rate. This is a fit key
+    /// only: at decide time the predictor learns the modal `(ic_category, interchange_bps)` pair.
+    pub interchange_bps: String,
     /// Transaction (booking) date, when the report carries one. Not staged into ClickHouse — used
     /// only to compute the ingested report's period (min/max) for the history record.
     pub txn_date: Option<NaiveDate>,
@@ -66,6 +74,12 @@ pub fn amount_band(amount: f64) -> &'static str {
     }
 }
 
+/// Log-amount bucket used by the segmented cost fitter. Kept separate from [`amount_band`], which
+/// is deliberately coarse because it is a serving-time predictor feature.
+pub fn fit_bucket(amount: f64) -> i32 {
+    (amount.log10() * 10.0).floor() as i32
+}
+
 impl SettledFeeRow {
     /// Map a variant string onto a funding bucket. Case-insensitive substring match, mirroring
     /// the Python `par_extract` behavior. Returns `""` for methods that are neither (iDEAL, …).
diff --git a/src/decider/gatewaydecider/multi_objective/hypersense_client.rs b/src/decider/gatewaydecider/multi_objective/hypersense_client.rs
index 94a0e4c7..c9ea9097 100644
--- a/src/decider/gatewaydecider/multi_objective/hypersense_client.rs
+++ b/src/decider/gatewaydecider/multi_objective/hypersense_client.rs
@@ -314,6 +314,10 @@ impl CostCache {
                                     issuer: None,
                                     ccy: None,
                                     ic_category: None,
+                                    interchange_bps: None,
+                                    segment_idx: None,
+                                    amount_lo: None,
+                                    amount_hi: None,
                                 }),
                             },
                         )
@@ -568,6 +572,10 @@ fn inhouse_costs(
                         issuer: m.issuer,
                         ccy: Some(m.currency),
                         ic_category: m.ic_category,
+                        interchange_bps: m.interchange_bps,
+                        segment_idx: m.segment_idx,
+                        amount_lo: m.amount_lo,
+                        amount_hi: m.amount_hi,
                         pct_bps: Some(m.pct_bps),
                         fixed_fee: Some(m.fixed),
                     }),
diff --git a/src/decider/gatewaydecider/multi_objective/mod.rs b/src/decider/gatewaydecider/multi_objective/mod.rs
index c1fc64a6..8bcd92f3 100644
--- a/src/decider/gatewaydecider/multi_objective/mod.rs
+++ b/src/decider/gatewaydecider/multi_objective/mod.rs
@@ -86,6 +86,14 @@ pub struct CostModel {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub ic_category: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
+    pub interchange_bps: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub segment_idx: Option<u16>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub amount_lo: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub amount_hi: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub pct_bps: Option<f64>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub fixed_fee: Option<f64>,
diff --git a/src/decider/gatewaydecider/multi_objective/seed_costs.rs b/src/decider/gatewaydecider/multi_objective/seed_costs.rs
index 5b3b753a..0b5dca3a 100644
--- a/src/decider/gatewaydecider/multi_objective/seed_costs.rs
+++ b/src/decider/gatewaydecider/multi_objective/seed_costs.rs
@@ -144,6 +144,10 @@ pub fn lookup_seed_costs(
                         variant: None,
                         issuer: None,
                         ic_category: None,
+                        interchange_bps: None,
+                        segment_idx: None,
+                        amount_lo: None,
+                        amount_hi: None,
                     }),
                 },
             ))
diff --git a/src/routes/cost_clusters.rs b/src/routes/cost_clusters.rs
index 92da2c17..c9f3935c 100644
--- a/src/routes/cost_clusters.rs
+++ b/src/routes/cost_clusters.rs
@@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize};
 use time::format_description::well_known::Iso8601;
 
 use crate::cost_ingestion::blended::{self, ClusterScope, TopCluster};
-use crate::cost_ingestion::overrides::{self, ClusterDims, ClusterOverride};
+use crate::cost_ingestion::overrides::{self, key_of_dims, ClusterDims, ClusterOverride};
 use crate::routes::connector_fees::{clickhouse_config, refresh_serving};
 
 /// How many top clusters to surface by default (top by GMV). Capped so the response and the
@@ -22,41 +22,35 @@ use crate::routes::connector_fees::{clickhouse_config, refresh_serving};
 const DEFAULT_LIMIT: u32 = 10;
 const MAX_LIMIT: u32 = 50;
 
-/// The `connector|network|variant|funding|issuer|currency|ic_category` key used on the wire and to
-/// key the stored override. Built lowercase so it round-trips through [`ClusterDims::from_key`] and
-/// matches the serving `fine_key`.
-#[allow(clippy::too_many_arguments)]
-fn cluster_key(
-    connector: &str,
-    network: &str,
-    variant: &str,
-    funding: &str,
-    issuer: &str,
-    currency: &str,
-    ic_category: &str,
-) -> String {
-    format!(
-        "{}|{}|{}|{}|{}|{}|{}",
-        connector.to_lowercase(),
-        network.to_lowercase(),
-        variant.to_lowercase(),
-        funding.to_lowercase(),
-        issuer.to_lowercase(),
-        currency.to_lowercase(),
-        ic_category.to_lowercase(),
-    )
+fn dims_from_cluster(c: &TopCluster) -> ClusterDims {
+    let has_segment_key = !c.interchange_bps.is_empty() || c.segment_idx != 0;
+    ClusterDims {
+        connector: c.connector.to_lowercase(),
+        card_network: c.card_network.to_lowercase(),
+        variant: c.variant.to_lowercase(),
+        funding: c.funding.to_lowercase(),
+        issuer_country: c.issuer_country.to_lowercase(),
+        currency: c.currency.to_lowercase(),
+        ic_category: c.ic_category.to_lowercase(),
+        interchange_bps: (!c.interchange_bps.is_empty()).then(|| c.interchange_bps.to_lowercase()),
+        segment_idx: has_segment_key.then_some(c.segment_idx),
+    }
+}
+
+fn empty_to_none(s: String) -> Option<String> {
+    if s.is_empty() {
+        None
+    } else {
+        Some(s)
+    }
+}
+
+fn nonzero_segment_value(v: f64, segment_idx: Option<u16>) -> Option<f64> {
+    segment_idx.filter(|idx| *idx > 0).map(|_| v)
 }
 
-fn key_of_dims(d: &ClusterDims) -> String {
-    cluster_key(
-        &d.connector,
-        &d.card_network,
-        &d.variant,
-        &d.funding,
-        &d.issuer_country,
-        &d.currency,
-        &d.ic_category,
-    )
+fn nonzero_quality(v: f64) -> Option<f64> {
+    (v != 0.0).then_some(v)
 }
 
 /// One cluster's fee picture for the dashboard.
@@ -71,6 +65,10 @@ pub struct ClusterFee {
     pub issuer_country: String,
     pub currency: String,
     pub ic_category: String,
+    pub interchange_bps: Option<String>,
+    pub segment_idx: Option<u16>,
+    pub amount_lo: Option<f64>,
+    pub amount_hi: Option<f64>,
     /// Transaction count and settled GMV for the cluster (0 for an override-only cluster no longer
     /// in the top set).
     pub n: u64,
@@ -78,6 +76,15 @@ pub struct ClusterFee {
     /// Learned fee (present when the cluster is in the fitted snapshot).
     pub model_pct_bps: Option<f64>,
     pub model_fixed: Option<f64>,
+    pub grade_bps: Option<f64>,
+    pub pct_ci95_bps: Option<f64>,
+    pub crossover_amount: Option<f64>,
+    pub prop_bps: Option<f64>,
+    pub fix_abs: Option<f64>,
+    pub fix_bps: Option<f64>,
+    pub below_gross_frac: Option<f64>,
+    pub fan_frac: Option<f64>,
+    pub fan_money_bps: Option<f64>,
     /// Manual override, when set.
     pub override_pct_bps: Option<f64>,
     pub override_fixed: Option<f64>,
@@ -136,15 +143,8 @@ pub async fn list_cost_clusters(
     let mut out: Vec<ClusterFee> = Vec::new();
 
     for c in top {
-        let key = cluster_key(
-            &c.connector,
-            &c.card_network,
-            &c.variant,
-            &c.funding,
-            &c.issuer_country,
-            &c.currency,
-            &c.ic_category,
-        );
+        let dims = dims_from_cluster(&c);
+        let key = key_of_dims(&dims);
         seen.insert(key.clone());
         let ov = overrides.get(&key);
         let (effective_pct_bps, effective_fixed, source) = match ov {
@@ -160,10 +160,23 @@ pub async fn list_cost_clusters(
             issuer_country: c.issuer_country,
             currency: c.currency,
             ic_category: c.ic_category,
+            interchange_bps: empty_to_none(c.interchange_bps),
+            segment_idx: dims.segment_idx,
+            amount_lo: nonzero_segment_value(c.amount_lo, dims.segment_idx),
+            amount_hi: nonzero_segment_value(c.amount_hi, dims.segment_idx),
             n: c.n,
             gross_sum: c.gross_sum,
             model_pct_bps: Some(c.pct_bps),
             model_fixed: Some(c.fixed),
+            grade_bps: Some(c.grade_bps),
+            pct_ci95_bps: Some(c.pct_ci95_bps),
+            crossover_amount: nonzero_quality(c.crossover_amount),
+            prop_bps: nonzero_quality(c.prop_bps),
+            fix_abs: nonzero_quality(c.fix_abs),
+            fix_bps: nonzero_quality(c.fix_bps),
+            below_gross_frac: Some(c.below_gross_frac),
+            fan_frac: Some(c.fan_frac),
+            fan_money_bps: Some(c.fan_money_bps),
             override_pct_bps: ov.map(|o| o.pct_bps),
             override_fixed: ov.map(|o| o.fixed),
             override_updated_at: ov.map(|o| o.updated_at.clone()),
@@ -189,10 +202,23 @@ pub async fn list_cost_clusters(
             issuer_country: o.dims.issuer_country.clone(),
             currency: o.dims.currency.clone(),
             ic_category: o.dims.ic_category.clone(),
+            interchange_bps: o.dims.interchange_bps.clone(),
+            segment_idx: o.dims.segment_idx,
+            amount_lo: None,
+            amount_hi: None,
             n: 0,
             gross_sum: 0.0,
             model_pct_bps: None,
             model_fixed: None,
+            grade_bps: None,
+            pct_ci95_bps: None,
+            crossover_amount: None,
+            prop_bps: None,
+            fix_abs: None,
+            fix_bps: None,
+            below_gross_frac: None,
+            fan_frac: None,
+            fan_money_bps: None,
             override_pct_bps: Some(o.pct_bps),
             override_fixed: Some(o.fixed),
             override_updated_at: Some(o.updated_at.clone()),
@@ -228,7 +254,7 @@ pub async fn set_cluster_override(
     }
     let dims = ClusterDims::from_key(&cluster_key).ok_or((
         StatusCode::BAD_REQUEST,
-        "cluster key must have 7 '|'-separated fields".to_string(),
+        "cluster key must have 7, 8, or 9 '|'-separated fields".to_string(),
     ))?;
     let ov = ClusterOverride {
         dims,
@@ -251,7 +277,7 @@ pub async fn delete_cluster_override(
 ) -> Result<StatusCode, (StatusCode, String)> {
     let dims = ClusterDims::from_key(&cluster_key).ok_or((
         StatusCode::BAD_REQUEST,
-        "cluster key must have 7 '|'-separated fields".to_string(),
+        "cluster key must have 7, 8, or 9 '|'-separated fields".to_string(),
     ))?;
     overrides::delete_cluster(&merchant_id, &dims)
         .await
diff --git a/website/src/components/pages/ClustersPanel.tsx b/website/src/components/pages/ClustersPanel.tsx
index 43ed3dd8..a4347ea8 100644
--- a/website/src/components/pages/ClustersPanel.tsx
+++ b/website/src/components/pages/ClustersPanel.tsx
@@ -54,6 +54,25 @@ function programOf(c: ClusterFee): string {
   return v ? titleCase(v) : '—'
 }
 
+/** Human-readable segment name: "Visa Standard debit · HU · HUF" (+ category). Used in the editor
+ * heading, where a single-line label reads better than the split columns. */
+function clusterLabel(c: ClusterFee): string {
+  const program = programOf(c)
+  const card = [networkLabel(c.card_network), program !== '—' ? program : '', c.funding]
+    .filter(Boolean)
+    .join(' ')
+  const parts = [card, c.issuer_country?.toUpperCase(), c.currency?.toUpperCase()].filter(Boolean)
+  const base = parts.join(' · ') || 'Unknown segment'
+  const amount = amountSegmentLabel(c)
+  const category = c.ic_category ? `${base} · ${c.ic_category}` : base
+  return amount ? `${category} · ${amount}` : category
+}
+
+function amountSegmentLabel(c: ClusterFee): string {
+  if (!c.segment_idx || c.amount_lo == null || c.amount_hi == null) return ''
+  return `${formatCompact(c.amount_lo)}-${formatCompact(c.amount_hi)}`
+}
+
 function formatFee(pctBps: number | null, fixed: number | null): string {
   if (pctBps == null && fixed == null) return '—'
   const pct = `${(pctBps ?? 0).toFixed(1)} bps`
@@ -239,6 +258,7 @@ function ClusterRow({
   onSaved: () => void
 }) {
   const isOverride = c.source === 'override'
+  const label = clusterLabel(c)
   // Pre-fill with the fee the row actually shows (effective = override, else model, else an inherited
   // connector fee). Seeding from override/model alone left inherited-fee segments at 0 even though a
   // real fee was displayed.
@@ -299,7 +319,16 @@ function ClusterRow({
         <td className="py-2 pr-3 tabular-nums text-slate-600 dark:text-[#c7cfdd]">
           {c.currency?.toUpperCase() || '—'}
         </td>
-        <td className="py-2 pr-3 text-slate-500 dark:text-[#9ca7ba]">{c.ic_category || '—'}</td>
+        <td className="py-2 pr-3 text-slate-500 dark:text-[#9ca7ba]">
+          {c.ic_category || '—'}
+          {(c.interchange_bps || amountSegmentLabel(c)) && (
+            <span className="block text-[11px] text-slate-400">
+              {[c.interchange_bps ? `${c.interchange_bps} bps IC` : '', amountSegmentLabel(c)]
+                .filter(Boolean)
+                .join(' · ')}
+            </span>
+          )}
+        </td>
         <td className="py-2 pr-3 text-right tabular-nums text-slate-600 dark:text-[#c7cfdd]">
           {c.gross_sum > 0 ? formatCompact(c.gross_sum) : '—'}
         </td>
@@ -354,8 +383,8 @@ function ClusterRow({
                   type="button"
                   onClick={save}
                   disabled={busy !== null || !merchantId}
-                  title="Save fee"
-                  aria-label="Save fee"
+                  title={`Save fee for ${label}`}
+                  aria-label={`Save fee for ${label}`}
                   className="inline-flex h-7 w-7 items-center justify-center rounded-md bg-brand-600 text-white transition-colors hover:bg-brand-700 disabled:opacity-40 dark:bg-white dark:text-black dark:hover:bg-slate-200"
                 >
                   {busy === 'save' ? <Spinner size={13} /> : <Check size={14} />}
diff --git a/website/src/components/pages/CostCoverageCard.tsx b/website/src/components/pages/CostCoverageCard.tsx
index 7ad5ae6d..1db86eb2 100644
--- a/website/src/components/pages/CostCoverageCard.tsx
+++ b/website/src/components/pages/CostCoverageCard.tsx
@@ -78,6 +78,13 @@ function VerdictTable({ coverage }: { coverage: CoverageSummary }) {
       txns: coverage.non_linear_txns,
       gross: coverage.non_linear_gross,
     },
+    {
+      verdict: 'Fan',
+      note: 'mixed fee populations — safe fallback',
+      dot: 'bg-fuchsia-500',
+      txns: coverage.fan_txns,
+      gross: coverage.fan_gross,
+    },
   ]
   const txnPct = (n: number) => (coverage.total_txns > 0 ? (n / coverage.total_txns) * 100 : 0)
   const volPct = (v: number) => (coverage.total_gross > 0 ? (v / coverage.total_gross) * 100 : 0)
diff --git a/website/src/hooks/useCostRouting.ts b/website/src/hooks/useCostRouting.ts
index b591703f..bcac19c1 100644
--- a/website/src/hooks/useCostRouting.ts
+++ b/website/src/hooks/useCostRouting.ts
@@ -6,15 +6,18 @@ export interface CoverageSummary {
   good_clusters: number
   thin_clusters: number
   non_linear_clusters: number
+  fan_clusters: number
   total_txns: number
   good_txns: number
   thin_txns: number
   non_linear_txns: number
+  fan_txns: number
   good_txn_pct: number
   total_gross: number
   good_gross: number
   thin_gross: number
   non_linear_gross: number
+  fan_gross: number
   /** Share of settled volume (money) with a trustworthy cost model — the headline metric. */
   good_gross_pct: number
   bps_rmse_p50: number
@@ -77,6 +80,8 @@ export interface PriceChange {
   issuer_country: string
   currency: string
   ic_category: string
+  interchange_bps?: string
+  segment_idx?: number
   old_pct_bps: number
   new_pct_bps: number
   old_fixed: number
@@ -159,10 +164,23 @@ export interface ClusterFee {
   issuer_country: string
   currency: string
   ic_category: string
+  interchange_bps?: string | null
+  segment_idx?: number | null
+  amount_lo?: number | null
+  amount_hi?: number | null
   n: number
   gross_sum: number
   model_pct_bps: number | null
   model_fixed: number | null
+  grade_bps?: number | null
+  pct_ci95_bps?: number | null
+  crossover_amount?: number | null
+  prop_bps?: number | null
+  fix_abs?: number | null
+  fix_bps?: number | null
+  below_gross_frac?: number | null
+  fan_frac?: number | null
+  fan_money_bps?: number | null
   override_pct_bps: number | null
   override_fixed: number | null
   override_updated_at: string | null