feat: enhance cost estimation fit logic

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

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)

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]

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.