五金门店AI智能体分析体系（新架构版）

-- MySQL源库创建DTS专用账号
CREATE USER 'dts_sync'@'%' IDENTIFIED BY 'DtsPass@2026';
GRANT SELECT, LOCK TABLES ON *.* TO 'dts_sync'@'%';
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'dts_sync'@'%';
GRANT ALL PRIVILEGES ON `inventory_db`.* TO 'dts_sync'@'%';
GRANT ALL PRIVILEGES ON `sales_db`.* TO 'dts_sync'@'%';
FLUSH PRIVILEGES;

-- 订单表ODS（保留CDC操作类型）
CREATE TABLE ods.orders
(
    order_id       String,
    store_id       String,
    customer_id    String,
    salesperson_id String,
    total_amount   Decimal(12,2),
    cost_amount    Decimal(12,2),
    discount_amount Decimal(12,2),
    payment_method Enum8('cash'=1,'wechat'=2,'alipay'=3,'credit'=4,'mixed'=5),
    order_status   Enum8('pending'=1,'confirmed'=2,'shipped'=3,'completed'=4,'cancelled'=5),
    order_date     Date,
    created_at     DateTime,
    updated_at     DateTime,
    _ts_ms         DateTime64(3) DEFAULT now64(3),
    _op            Enum8('insert'=1,'update'=2,'delete'=3)
)
ENGINE = ReplicatedReplacingMergeTree(_ts_ms)
PARTITION BY toYYYYMM(order_date)
ORDER BY (store_id, order_id, _ts_ms)
TTL order_date + INTERVAL 2 YEAR;

-- 订单明细ODS
CREATE TABLE ods.order_detail
(
    detail_id    String,
    order_id     String,
    product_id   String,
    product_code String,
    product_name String,
    category_id  String,
    warehouse_id String,
    quantity     UInt32,
    unit_price   Decimal(10,2),
    cost_price   Decimal(10,2),
    line_amount  Decimal(12,2),
    created_at   DateTime
)
ENGINE = ReplicatedReplacingMergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (order_id, detail_id)
TTL created_at + INTERVAL 2 YEAR;

-- 商品表ODS
CREATE TABLE ods.products
(
    product_id    String,
    product_code  String,
    product_name  String,
    category_id   String,
    category_name String,
    brand         String,
    spec          String,
    unit          String,
    price         Decimal(10,2),
    cost_price    Decimal(10,2),
    barcode       String,
    safe_stock    UInt32,
    _ts_ms        DateTime64(3) DEFAULT now64(3),
    _op           Enum8('insert'=1,'update'=2,'delete'=3)
)
ENGINE = ReplicatedReplacingMergeMergeTree(_ts_ms)
ORDER BY (category_id, product_id);

-- 库存表ODS
CREATE TABLE ods.inventory_stock
(
    warehouse_id   String,
    warehouse_name String,
    product_id     String,
    stock_qty      Int32,
    available_qty Int32,
    reserved_qty  Int32,
    update_time    DateTime,
    _ts_ms         DateTime64(3) DEFAULT now64(3)
)
ENGINE = ReplicatedReplacingMergeTree(_ts_ms)
PARTITION BY toYYYYMM(update_time)
ORDER BY (warehouse_id, product_id);

-- 客户表ODS
CREATE TABLE ods.customers
(
    customer_id   String,
    customer_code String,
    customer_name String,
    customer_type Enum8('retail'=1,'wholesale'=2,'engineering'=3),
    credit_limit  Decimal(12,2),
    contact_person String,
    phone         String,
    address       String,
    _ts_ms        DateTime64(3) DEFAULT now64(3)
)
ENGINE = ReplicatedReplacingMergeTree(_ts_ms)
ORDER BY (customer_type, customer_id);

-- 供应商表ODS
CREATE TABLE ods.suppliers
(
    supplier_id   String,
    supplier_code String,
    supplier_name String,
    contact_person String,
    phone         String,
    address       String,
    _ts_ms        DateTime64(3) DEFAULT now64(3)
)
ENGINE = ReplicatedReplacingMergeTree(_ts_ms)
ORDER BY (supplier_id);

-- 订单宽表（去删除标记，合并更新）
CREATE TABLE dwd.order_fact
(
    order_id         String,
    store_id         String,
    store_name       String,
    customer_id      String,
    customer_name    String,
    customer_type    String,
    salesperson_id   String,
    salesperson_name String,
    order_date       Date,
    order_hour       UInt8,
    weekday          UInt8,
    total_amount     Decimal(12,2),
    cost_amount      Decimal(12,2),
    profit_amount    Decimal(12,2),
    profit_margin    Decimal(5,2),
    discount_amount  Decimal(12,2),
    payment_method   String,
    order_status     String,
    is_credit        UInt8,
    created_at       DateTime,
    updated_at       DateTime
)
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(order_date)
ORDER BY (store_id, order_date, order_id)
TTL order_date + INTERVAL 2 YEAR;

-- 库存宽表（最新余额）
CREATE TABLE dwd.inventory_fact
(
    warehouse_id   String,
    warehouse_name String,
    product_id     String,
    product_code   String,
    product_name   String,
    category_id    String,
    stock_qty      Int32,
    available_qty  Int32,
    reserved_qty   Int32,
    update_time    DateTime
)
ENGINE = ReplacingMergeTree(update_time)
ORDER BY (warehouse_id, product_id);

-- 日粒度订单汇总（加速日报）
CREATE MATERIALIZED VIEW dws.mv_orders_daily
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(order_date)
ORDER BY (store_id, category_id, order_date)
AS
SELECT
    store_id,
    category_id,
    order_date,
    count()            AS order_count,
    uniqExact(order_id) AS order_num,
    sum(total_amount)  AS daily_sales,
    sum(cost_amount)   AS daily_cost,
    sum(profit_amount)  AS daily_profit,
    sum(profit_amount) / NULLIF(sum(total_amount), 0) * 100 AS profit_margin_pct,
    uniqExact(customer_id) AS customer_count
FROM dwd.order_fact
WHERE order_status NOT IN ('cancelled')
GROUP BY store_id, category_id, order_date;

-- 月粒度订单汇总（加速月报）
CREATE MATERIALIZED VIEW dws.mv_orders_monthly
ENGINE = SummingMergeTree()
ORDER BY (store_id, category_id, order_year, order_month)
AS
SELECT
    store_id,
    category_id,
    toYear(order_date)    AS order_year,
    toMonth(order_date)   AS order_month,
    count()               AS order_count,
    sum(total_amount)     AS monthly_sales,
    sum(profit_amount)    AS monthly_profit
FROM dwd.order_fact
WHERE order_status NOT IN ('cancelled')
GROUP BY store_id, category_id, order_year, order_month;

-- 商品销售排行汇总（加速爆款分析）
CREATE MATERIALIZED VIEW dws.mv_product_sales_rank
ENGINE = AggregatingMergeTree()
ORDER BY (rank_date, category_id, product_id)
AS
SELECT
    toDate(created_at) AS rank_date,
    category_id,
    product_id,
    product_name,
    sum(line_amount)   AS total_sales,
    sum(quantity)      AS total_qty,
    count()            AS order_count
FROM ods.order_detail
GROUP BY rank_date, category_id, product_id, product_name;

-- 库存余额汇总
CREATE MATERIALIZED VIEW dws.mv_inventory_latest
ENGINE = ReplacingMergeTree(update_time)
ORDER BY (warehouse_id, product_id)
AS
SELECT
    warehouse_id,
    warehouse_name,
    product_id,
    product_code,
    product_name,
    category_id,
    stock_qty,
    available_qty,
    update_time
FROM ods.inventory_stock;

-- 客户应收账款汇总
CREATE MATERIALIZED VIEW dws.mv_customer_ar
ENGINE = SummingMergeTree()
ORDER BY (customer_id, stat_month)
AS
SELECT
    customer_id,
    toStartOfMonth(created_at) AS stat_month,
    sum(total_amount)     AS total_sales,
    sum(profit_amount)    AS total_profit,
    count()               AS order_count
FROM dwd.order_fact
WHERE is_credit = 1 AND order_status NOT IN ('cancelled')
GROUP BY customer_id, stat_month;

-- 每日核心KPI（查询物化视图，秒级响应）
SELECT
    order_date,
    sum(order_num)      AS order_count,
    sum(daily_sales)    AS total_sales,
    sum(daily_profit)   AS total_profit,
    sum(daily_profit) / NULLIF(sum(daily_sales), 0) * 100 AS profit_margin_pct,
    sum(customer_count) AS customer_count
FROM dws.mv_orders_daily
WHERE order_date = today() - 1
GROUP BY order_date;

-- 环比计算
WITH today AS (
    SELECT sum(daily_sales) AS sales FROM dws.mv_orders_daily WHERE order_date = today() - 1
),
yesterday AS (
    SELECT sum(daily_sales) AS sales FROM dws.mv_orders_daily WHERE order_date = today() - 2
)
SELECT
    (t.sales - y.sales) / NULLIF(y.sales, 0) * 100 AS growth_pct
FROM today t, yesterday y;

-- 近30天爆款商品TOP10（含库存可售天数）
SELECT
    product_id,
    product_name,
    sum(total_sales)    AS total_sales,
    sum(total_qty)      AS total_qty,
    argMax(stock_qty, update_time) AS current_stock,
    argMax(stock_qty, update_time) / NULLIF(sum(total_qty) / 30, 0) AS stock_days_remaining
FROM dws.mv_product_sales_rank d
LEFT JOIN dws.mv_inventory_latest i ON d.product_id = i.product_id
WHERE rank_date BETWEEN today() - 30 AND today() - 1
GROUP BY product_id, product_name
ORDER BY total_sales DESC
LIMIT 10;

-- 库存预警：缺货 + 临期 + 积压
SELECT
    i.product_id,
    i.product_name,
    i.stock_qty,
    p.safe_stock,
    CASE
        WHEN i.stock_qty = 0 THEN '🔴 缺货'
        WHEN i.stock_qty < p.safe_stock THEN '🟠 库存不足'
        WHEN i.stock_qty > p.safe_stock * 3 THEN '🟡 积压'
        ELSE '🟢 正常'
    END AS alert_status
FROM dws.mv_inventory_latest i
JOIN ods.products p ON i.product_id = p.product_id
WHERE i.warehouse_id = 'MAIN_WH'
  AND (
    i.stock_qty = 0
    OR i.stock_qty < p.safe_stock
    OR i.stock_qty > p.safe_stock * 3
  )
ORDER BY alert_status, i.stock_qty ASC;

-- 员工当日绩效排名
SELECT
    e.employee_name,
    e.department,
    count()            AS order_count,
    sum(o.total_amount) AS total_sales,
    sum(o.profit_amount) AS total_profit,
    sum(o.profit_amount) / NULLIF(sum(o.total_amount), 0) * 100 AS profit_margin,
    uniqExact(o.customer_id) AS customer_count
FROM dwd.order_fact o
JOIN ods.employees e ON o.salesperson_id = e.employee_id
WHERE o.order_date = today() - 1
  AND o.order_status NOT IN ('cancelled')
GROUP BY e.employee_name, e.department
ORDER BY total_sales DESC
LIMIT 10;

WITH daily_sales AS (
    SELECT
        product_id,
        toDate(created_at) AS sale_date,
        sum(quantity) AS daily_qty
    FROM ods.order_detail
    WHERE created_at >= today() - 90
    GROUP BY product_id, toDate(created_at)
),
sales_stats AS (
    SELECT
        product_id,
        avg(daily_qty)           AS avg_daily_sales,
        stddevPop(daily_qty)    AS demand_stddev
    FROM daily_sales
    GROUP BY product_id
),
lead_times AS (
    SELECT product_id, avg(lead_time_days) AS avg_lead_time
    FROM ods.supplier_lead_times
    GROUP BY product_id
),
safe_stock_calc AS (
    SELECT
        s.product_id,
        p.product_name,
        s.avg_daily_sales,
        s.demand_stddev,
        l.avg_lead_time,
        -- 安全库存 97.5%服务水平
        ROUND(1.96 * s.demand_stddev * sqrt(l.avg_lead_time)) AS safe_stock,
        -- 再订货点
        ROUND(s.avg_daily_sales * l.avg_lead_time + 1.96 * s.demand_stddev * sqrt(l.avg_lead_time)) AS reorder_point
    FROM sales_stats s
    LEFT JOIN lead_times l ON s.product_id = l.product_id
    JOIN ods.products p ON s.product_id = p.product_id
)
SELECT
    *,
    GREATEST(0, reorder_point + safe_stock - current_stock) AS suggested_order_qty
FROM safe_stock_calc
ORDER BY suggested_order_qty DESC;

-- 7天/14天缺货风险预测
WITH future_demand AS (
    SELECT
        product_id,
        sum(quantity) / 7 AS avg_daily_7d,
        sum(quantity)     AS demand_7d
    FROM ods.order_detail
    WHERE created_at >= today() - 7
    GROUP BY product_id
)
SELECT
    p.product_name,
    i.stock_qty,
    f.avg_daily_7d,
    ROUND(i.stock_qty / NULLIF(f.avg_daily_7d, 0)) AS stock_days_remaining,
    CASE
        WHEN i.stock_qty <= f.demand_7d THEN '🔴 7天内缺货'
        WHEN i.stock_qty <= f.demand_7d * 2 THEN '🟠 14天内缺货'
        ELSE '🟢 库存充足'
    END AS stockout_risk
FROM ods.products p
LEFT JOIN dws.mv_inventory_latest i ON p.product_id = i.product_id
LEFT JOIN future_demand f ON p.product_id = f.product_id
WHERE i.stock_qty <= f.demand_7d * 2 OR i.stock_qty = 0
ORDER BY i.stock_qty ASC;

SELECT
    p.product_name,
    i.stock_qty,
    today() - toDate(last_out_date) AS days_since_last_sale,
    today() - toDate(last_in_date)  AS days_in_stock,
    CASE
        WHEN days_since_last_sale > 180 THEN '🆕 死库存'
        WHEN days_in_stock > 365 THEN '⚠️ 超龄库存'
        WHEN days_since_last_sale > 90 THEN '🔴 长期滞销'
        WHEN days_in_stock > 90 THEN '🟡 积压'
        ELSE '🟢 正常'
    END AS slow_moving_type
FROM dws.mv_inventory_latest i
JOIN ods.products p ON i.product_id = p.product_id
WHERE i.stock_qty > 0
  AND (days_since_last_sale > 90 OR days_in_stock > 180)
ORDER BY days_since_last_sale DESC
LIMIT 50;

SELECT
    sp.supplier_id,
    si.supplier_name,
    sum(purchase_amount)     AS total_purchase,
    sum(return_amount)        AS total_return,
    sum(purchase_amount) - sum(return_amount) AS net_purchase,
    sum(CASE WHEN payment_status = 'unpaid' THEN purchase_amount ELSE 0 END) AS accounts_payable,
    max(due_date)             AS latest_due_date,
    today() - max(due_date)   AS aging_days,
    CASE
        WHEN aging_days <= 30 THEN '🟢 正常'
        WHEN aging_days <= 60 THEN '🟡 预警'
        WHEN aging_days <= 90 THEN '🟠 警告'
        ELSE '🔴 危险'
    END AS aging_level
FROM ods.supplier_purchases sp
JOIN ods.suppliers si ON sp.supplier_id = si.supplier_id
WHERE purchase_date BETWEEN '{start_date}' AND '{end_date}'
GROUP BY sp.supplier_id, si.supplier_name
ORDER BY net_purchase DESC;

WITH delivery_score AS (
    SELECT
        supplier_id,
        ROUND(sum(is_on_time) * 100.0 / count(), 2) AS on_time_rate
    FROM ods.supplier_delivery
    WHERE actual_delivery_date BETWEEN '{start_date}' AND '{end_date}'
    GROUP BY supplier_id
),
quality_score AS (
    SELECT
        supplier_id,
        ROUND((1 - sum(return_quantity) / sum(quantity)) * 100, 2) AS quality_rate
    FROM ods.supplier_quality
    WHERE return_date BETWEEN '{start_date}' AND '{end_date}'
    GROUP BY supplier_id
)
SELECT
    si.supplier_name,
    ds.on_time_rate,
    qs.quality_rate,
    ROUND(ds.on_time_rate * 0.3 + qs.quality_rate * 0.3 + coop_score * 0.4, 2) AS total_score,
    CASE WHEN total_score >= 90 THEN 'A' WHEN total_score >= 75 THEN 'B' WHEN total_score >= 60 THEN 'C' ELSE 'D' END AS grade
FROM ods.suppliers si
LEFT JOIN delivery_score ds ON si.supplier_id = ds.supplier_id
LEFT JOIN quality_score qs ON si.supplier_id = qs.supplier_id
ORDER BY total_score DESC;

WITH yearly_sales AS (
    SELECT
        customer_id,
        sum(total_amount) AS annual_sales
    FROM dwd.order_fact
    WHERE order_date BETWEEN '{year}-01-01' AND '{year}-12-31'
      AND order_status NOT IN ('cancelled')
    GROUP BY customer_id
),
sales_with_cumsum AS (
    SELECT
        customer_id,
        annual_sales,
        sum(annual_sales) OVER() AS total_sales,
        sum(annual_sales) OVER(ORDER BY annual_sales DESC) AS cumsum_sales
    FROM yearly_sales
)
SELECT
    ci.customer_name,
    s.annual_sales,
    ROUND(s.annual_sales / s.total_sales * 100, 2) AS sales_pct,
    ROUND(s.cumsum_sales / s.total_sales * 100, 2) AS cumsum_pct,
    CASE WHEN cumsum_pct <= 80 THEN 'A' WHEN cumsum_pct <= 95 THEN 'B' ELSE 'C' END AS abc_class
FROM sales_with_cumsum s
JOIN ods.customers ci ON s.customer_id = ci.customer_id
ORDER BY s.annual_sales DESC;

SELECT
    ci.customer_name,
    ci.customer_type,
    max(o.order_date)  AS last_transaction_date,
    today() - max(o.order_date) AS days_since_transaction,
    CASE
        WHEN max(o.order_date) IS NULL THEN '🔴 严重预警'
        WHEN today() - max(o.order_date) > 90 THEN '🔴 严重预警'
        WHEN today() - max(o.order_date) > 60 THEN '🟠 中度预警'
        WHEN today() - max(o.order_date) > 30 THEN '🟡 轻度预警'
        ELSE '🟢 正常'
    END AS warning_level
FROM ods.customers ci
LEFT JOIN dwd.order_fact o ON ci.customer_id = o.customer_id
GROUP BY ci.customer_id, ci.customer_name, ci.customer_type
HAVING days_since_transaction > 30 OR days_since_transaction IS NULL
ORDER BY days_since_transaction DESC NULLS FIRST;

SELECT
    e.employee_name,
    e.department,
    sum(o.total_amount)   AS total_sales,
    sum(CASE WHEN o.payment_method != 'credit' THEN o.total_amount ELSE 0 END) AS cash_sales,
    sum(o.profit_amount)  AS total_profit,
    ROUND(sum(o.profit_amount) / NULLIF(sum(o.total_amount), 0) * 100, 2) AS profit_margin,
    -- 综合提成 = 销售额 × 回款率系数 × 毛利率系数 × 基础提成率
    ROUND(
        sum(o.total_amount)
        * (sum(CASE WHEN o.payment_method != 'credit' THEN o.total_amount ELSE 0 END) / sum(o.total_amount))
        * (sum(o.profit_amount) / sum(o.total_amount))
        * 0.15,
    2) AS commission_amount
FROM dwd.order_fact o
JOIN ods.employees e ON o.salesperson_id = e.employee_id
WHERE o.order_date BETWEEN '{start_date}' AND '{end_date}'
  AND o.order_status NOT IN ('cancelled')
GROUP BY e.employee_name, e.department
ORDER BY commission_amount DESC;

WITH monthly_data AS (
    SELECT
        toStartOfMonth(order_date) AS month,
        sum(total_amount)  AS gross_sales,
        sum(profit_amount) AS gross_profit,
        sum(profit_amount) / NULLIF(sum(total_amount), 0) * 100 AS gross_margin_pct
    FROM dwd.order_fact
    WHERE order_date >= toStartOfMonth(addMonths(today(), -12))
      AND order_status NOT IN ('cancelled', 'returned')
    GROUP BY toStartOfMonth(order_date)
),
expense_data AS (
    SELECT
        toStartOfMonth(expense_date) AS month,
        sum(amount) AS total_expense
    FROM ods.expense_record
    WHERE expense_date >= toStartOfMonth(addMonths(today(), -12))
    GROUP BY toStartOfMonth(expense_date)
)
SELECT
    m.month,
    m.gross_sales,
    m.gross_profit,
    m.gross_margin_pct,
    (m.gross_profit - e.total_expense) AS net_profit,
    (m.gross_profit - e.total_expense) / NULLIF(m.gross_sales, 0) * 100 AS net_margin_pct,
    e.total_expense,
    e.total_expense / NULLIF(m.gross_sales, 0) * 100 AS expense_ratio_pct
FROM monthly_data m
LEFT JOIN expense_data e ON m.month = e.month
ORDER BY m.month DESC;

WITH sales_data AS (
    SELECT
        toStartOfMonth(created_at) AS month,
        sum(cost_amount) AS cogs
    FROM ods.order_detail
    WHERE created_at >= toStartOfMonth(addMonths(today(), -12))
    GROUP BY toStartOfMonth(created_at)
),
avg_inventory AS (
    SELECT
        toStartOfMonth(update_time) AS month,
        avg(stock_qty * cost_price) AS avg_inventory_value
    FROM dws.mv_inventory_latest i
    JOIN ods.products p ON i.product_id = p.product_id
    WHERE update_time >= toStartOfMonth(addMonths(today(), -12))
    GROUP BY toStartOfMonth(update_time)
)
SELECT
    s.month,
    a.avg_inventory_value,
    s.cogs,
    a.avg_inventory_value / NULLIF(s.cogs / 30, 0) AS inventory_turnover_days,
    s.cogs / NULLIF(a.avg_inventory_value, 0) AS inventory_turnover_rate
FROM sales_data s
LEFT JOIN avg_inventory a ON s.month = a.month
ORDER BY s.month DESC;

SELECT
    c.customer_code,
    c.customer_name,
    sum(CASE WHEN aging_days <= 0 THEN outstanding_amount ELSE 0 END) AS not_due_amount,
    sum(CASE WHEN aging_days BETWEEN 1 AND 30 THEN outstanding_amount ELSE 0 END) AS overdue_1_30,
    sum(CASE WHEN aging_days BETWEEN 31 AND 60 THEN outstanding_amount ELSE 0 END) AS overdue_31_60,
    sum(CASE WHEN aging_days BETWEEN 61 AND 90 THEN outstanding_amount ELSE 0 END) AS overdue_61_90,
    sum(CASE WHEN aging_days > 90 THEN outstanding_amount ELSE 0 END) AS overdue_90_plus,
    sum(outstanding_amount) AS total_outstanding
FROM (
    SELECT
        ar.customer_id,
        ar.invoice_amount - ar.paid_amount AS outstanding_amount,
        toDate(ar.due_date) - today() AS aging_days
    FROM ods.ar_record ar
    WHERE ar.invoice_amount > ar.paid_amount
) t
JOIN ods.customers c ON t.customer_id = c.customer_id
GROUP BY c.customer_code, c.customer_name
ORDER BY total_outstanding DESC
LIMIT 50;