diff --git a/backend/app/api/map_dimensions.py b/backend/app/api/map_dimensions.py
new file mode 100644
index 00000000..bf09b850
--- /dev/null
+++ b/backend/app/api/map_dimensions.py
@@ -0,0 +1,600 @@
+# -*- coding: utf-8 -*-
+"""战略地图 · 维度切片(时间 / 门店 / 产品)—— A3
+
+目的:让战略地图的 KPI 能按维度切片重算,而不是永远只显示总额/最新值。
+
+口径(与既有读路径保持一致,不另造口径):
+- 实测区:复用 `app.utils.kpi_value_scope.actual_only_clause()`,预估行不得进入切片。
+- 指标方向:优先 `kpi_definitions.direction`;缺失时退回
+ `app.utils.strategy_board.INVERSE_KPI_IDS` 显式白名单(不用 NULL 推断)。
+- 状态类:`unit == '状态'` ⇒ 1 = 达标 / 0 = 未达标,不做除法。
+- 达成率阈值沿用地图现有语义:≥90% success / ≥60% warning / <60% danger。
+
+三维可用性(以库内事实为准,不假装有数据):
+- 时间:`kpi_values.period`(YYYY / YYYY-Hn / YYYY-MM)真实存在,可切。
+- 产品:`product_sales`(仅 entity 1,2026-01…2026-08,按 `product_name`)真实存在;
+ 只对「产品级可取数」的 KPI(PRODUCT_SCOPED_CODES)生效,其余 KPI 显式不可用。
+- 门店:库内**无**门店级事实表(无 store/dept 字段,`voucher_details` 无辅助核算),
+ 仅能从 `subjects.subject_name` 反推门店名称 ⇒ 提供候选值但标记不可用;
+ 切片结果一律 `available=false`,**绝不回退总额**。
+
+空态契约:所选维度下没有某 KPI 的实测值时,返回 `actual_value=null` +
+`available=false` + `reason`;绝不把总额/其它期间的值填进来。
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.auth_middleware import require_role
+from app.database import get_db
+from app.deps import get_entity_id
+from app.models import KPIDefinition, KPIValue, MapObjective, ProductSales, StrategicMap
+from app.utils.kpi_value_scope import actual_only_clause
+from app.utils.strategy_board import INVERSE_KPI_IDS, STATUS_UNIT
+
+router = APIRouter(
+ prefix="/api/cma/maps",
+ tags=["战略地图维度切片"],
+ dependencies=[Depends(require_role("ceo", "finance"))],
+)
+
+#: 目前只有营业收入的月度总额来自 product_sales(已对拍:2026-01…07 与 KPI 值逐月相等)
+PRODUCT_SCOPED_CODES = frozenset({"F_REVENUE"})
+
+#: 流量型单位:季度/年度切片按「月份求和」
+FLOW_UNITS = frozenset({"万元", "元", "个", "次", "轮", "件", "人"})
+
+#: 达成率 → 色阶(与地图现有 legend 一致)
+SUCCESS_RATE = 0.9
+WARNING_RATE = 0.6
+
+_MONTH_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")
+_QUARTER_RE = re.compile(r"^\d{4}-Q[1-4]$")
+_YEAR_RE = re.compile(r"^\d{4}$")
+
+_STORE_SUFFIX_RE = re.compile(r"(吴忠地标店|吴忠隆泰店|新区大店|地标店|一店|二店)")
+_STORE_DISPLAY_ORDER = ["一店", "二店", "吴忠地标店", "新区大店", "吴忠隆泰店"]
+
+#: 存量/余额型指标:区间切片取「最新一期」,禁止按月求和(否则余额被累加)
+BALANCE_HINTS = ("余额", "库存", "去化", "货币资金", "借款", "应收", "应付", "资产", "负债", "占用", "头寸")
+
+#: 产品源与 KPI 实测值的允许偏差(超过即判源未对齐,不参与切片)
+PRODUCT_SOURCE_TOLERANCE = 0.01
+
+
+# ── 小工具 ────────────────────────────────────────────────
+def _obj_kpi_codes(raw: Any) -> list[str]:
+ """map_objectives.kpis 兼容四种存储形态:list[str] / json 字符串 / list[dict] / NULL。"""
+ if raw is None:
+ return []
+ data = raw
+ if isinstance(raw, str):
+ try:
+ data = json.loads(raw)
+ except (TypeError, ValueError):
+ return []
+ if not isinstance(data, list):
+ return []
+ out: list[str] = []
+ for item in data:
+ if isinstance(item, str):
+ out.append(item)
+ elif isinstance(item, dict):
+ code = item.get("kpi_code") or item.get("code")
+ if code:
+ out.append(str(code))
+ return out
+
+
+def _map_or_404(db: Session, map_id: int, entity_id: int) -> StrategicMap:
+ m = db.query(StrategicMap).filter(
+ StrategicMap.id == map_id, StrategicMap.entity_id == entity_id,
+ ).first()
+ if not m:
+ raise HTTPException(404, "战略地图不存在")
+ return m
+
+
+def _map_kpis(db: Session, map_id: int, entity_id: int):
+ """返回 (map_objectives, {kpi_code: KPIDefinition}),KPI 只取本实体(账套隔离)。"""
+ objs = db.query(MapObjective).filter(
+ MapObjective.map_id == map_id, MapObjective.entity_id == entity_id,
+ ).order_by(MapObjective.sort_order, MapObjective.id).all()
+ codes: set[str] = set()
+ for o in objs:
+ codes.update(_obj_kpi_codes(o.kpis))
+ defs: dict[str, KPIDefinition] = {}
+ if codes:
+ for k in db.query(KPIDefinition).filter(
+ KPIDefinition.entity_id == entity_id, KPIDefinition.kpi_code.in_(codes),
+ ).all():
+ defs[k.kpi_code] = k
+ return objs, defs
+
+
+def _parse_period(period: Optional[str]) -> tuple[Optional[str], Optional[str]]:
+ """返回 (粒度, 归一值):year / quarter / month;非法或空返回 (None, None)。"""
+ if not period:
+ return None, None
+ p = period.strip()
+ if _MONTH_RE.match(p):
+ return "month", p
+ if _QUARTER_RE.match(p):
+ return "quarter", p
+ if _YEAR_RE.match(p):
+ return "year", p
+ return None, None
+
+
+def _months_of(period_type: str, period: str) -> list[str]:
+ if period_type == "month":
+ return [period]
+ year = period[:4]
+ if period_type == "quarter":
+ q = int(period[-1])
+ start = (q - 1) * 3 + 1
+ return [f"{year}-{m:02d}" for m in range(start, start + 3)]
+ return [f"{year}-{m:02d}" for m in range(1, 13)]
+
+
+def _unit_kind(unit: Optional[str], kpi: Optional[KPIDefinition] = None) -> str:
+ u = (unit or "").strip()
+ if u == STATUS_UNIT:
+ return "status"
+ name = (getattr(kpi, "kpi_name", "") or "") + " " + (getattr(kpi, "kpi_code", "") or "")
+ if any(h in name for h in BALANCE_HINTS):
+ return "balance"
+ if u in FLOW_UNITS:
+ return "flow"
+ return "ratio"
+
+
+def _is_lower_better(kpi: KPIDefinition) -> bool:
+ direction = (getattr(kpi, "direction", None) or "").strip().lower()
+ if direction in ("lower_better", "lower", "desc", "down"):
+ return True
+ if direction in ("higher_better", "higher", "asc", "up"):
+ return False
+ return kpi.id in INVERSE_KPI_IDS
+
+
+def _level_from_ratio(ratio: Optional[float]) -> str:
+ if ratio is None:
+ return "gray"
+ if ratio >= SUCCESS_RATE:
+ return "green"
+ if ratio >= WARNING_RATE:
+ return "yellow"
+ return "red"
+
+
+def _base_target(kpi: KPIDefinition) -> Optional[float]:
+ for attr in ("target_monthly", "target_value"):
+ v = getattr(kpi, attr, None)
+ if v is not None:
+ try:
+ return float(v)
+ except (TypeError, ValueError):
+ continue
+ return None
+
+
+def _scaled_target(kpi: KPIDefinition, period_type: str, year_period: Optional[str], db: Session, entity_id: int) -> tuple[Optional[float], str]:
+ """目标值:年切片优先取 kpi_targets 的年度行;否则按单位缩放月度目标。
+
+ 返回 (target, target_basis)。
+ """
+ if year_period:
+ row = db.execute(text(
+ "SELECT target_value FROM kpi_targets WHERE entity_id=:e AND kpi_id=:k "
+ "AND period=:p AND target_type IN ('yearly','annual') ORDER BY target_type='yearly' DESC LIMIT 1"
+ ), {"e": entity_id, "k": kpi.id, "p": year_period}).fetchone()
+ if row and row[0] is not None:
+ return float(row[0]), f"kpi_targets.{year_period}.yearly"
+
+ base = _base_target(kpi)
+ if base is None:
+ return None, "no_target"
+ kind = _unit_kind(kpi.unit, kpi)
+ if kind == "status":
+ return None, "status_no_target"
+ if kind == "balance":
+ return base, "target_balance"
+ if period_type == "month":
+ return base, "target_monthly"
+ if period_type == "quarter":
+ return (base * 3 if kind == "flow" else base), "target_monthly×3"
+ return (base * 12 if kind == "flow" else base), "target_monthly×12"
+
+
+def _aggregate(rows: list[KPIValue], unit: Optional[str], kpi: Optional[KPIDefinition] = None) -> tuple[Optional[float], str, list[str], bool]:
+ """把区间内的实测行聚合成一个切片值。
+
+ 返回 (value, aggregation, months_used, partial)。
+ - 状态类 / 存量余额类:取最新一期(不求和)
+ - 流量型:求和
+ - 比率/天数:均值
+ """
+ if not rows:
+ return None, "none", [], True
+ ordered = sorted(rows, key=lambda v: (str(v.period), v.id))
+ months = [str(v.period) for v in ordered]
+ kind = _unit_kind(unit, kpi)
+ if kind in ("status", "balance"):
+ last = ordered[-1]
+ return (float(last.actual_value) if last.actual_value is not None else None), "latest", [str(last.period)], False
+ vals = [float(v.actual_value) for v in ordered if v.actual_value is not None]
+ if not vals:
+ return None, "none", months, True
+ if kind == "flow":
+ return round(sum(vals), 4), "sum_of_months", months, False
+ return round(sum(vals) / len(vals), 4), "mean_of_months", months, False
+
+
+# ── 维度候选值 ────────────────────────────────────────────
+def _time_options(db: Session, kpi_ids: list[int], entity_id: int) -> dict:
+ if not kpi_ids:
+ return {"available": False, "reason": "该地图未绑定 KPI", "years": [], "quarters": [], "months": []}
+ rows = db.query(KPIValue.period).filter(
+ KPIValue.entity_id == entity_id,
+ KPIValue.kpi_id.in_(kpi_ids),
+ actual_only_clause(),
+ ).distinct().all()
+ periods = [str(r[0]) for r in rows if r[0]]
+ months = sorted(p for p in periods if _MONTH_RE.match(p))
+ quarters = sorted(p for p in periods if _QUARTER_RE.match(p))
+ years = sorted({p[:4] for p in periods})
+ if months:
+ for y in sorted({m[:4] for m in months}):
+ for q in range(1, 5):
+ qq = f"{y}-Q{q}"
+ if qq not in quarters and any(m in _months_of("quarter", qq) for m in months):
+ quarters.append(qq)
+ quarters.sort()
+ return {
+ "available": bool(periods),
+ "reason": None if periods else "该地图 KPI 无实测值(kpi_values 为空)",
+ "years": years,
+ "quarters": quarters,
+ "months": months,
+ "granularity": ["year", "quarter", "month"],
+ }
+
+
+def _store_options(db: Session, entity_id: int) -> dict:
+ rows = db.execute(text(
+ "SELECT DISTINCT subject_code, subject_name FROM subjects WHERE entity_id=:e AND subject_name LIKE '%店%'"
+ ), {"e": entity_id}).fetchall()
+ found: set[str] = set()
+ for _code, name in rows:
+ for hit in _STORE_SUFFIX_RE.findall(str(name or "")):
+ found.add("吴忠地标店" if hit == "地标店" else hit)
+ values = [{"code": f"BR-{i+1:02d}", "name": n} for i, n in enumerate(
+ [n for n in _STORE_DISPLAY_ORDER if n in found] or sorted(found)
+ )]
+ return {
+ "available": False,
+ "reason": "该维度暂无数据:库内无门店级事实表(无 store/dept 字段,凭证无辅助核算),"
+ "门店名称仅能从费用类科目名称反推,无法按门店切 KPI;不回退总额",
+ "values": values,
+ "source": "subjects.subject_name 反推(仅费用类科目)",
+ "coverage": "expense_subjects_only",
+ }
+
+
+def _product_options(db: Session, entity_id: int) -> dict:
+ rows = db.query(ProductSales.product_name).filter(
+ ProductSales.entity_id == entity_id,
+ ).distinct().all()
+ names = sorted({str(r[0]) for r in rows if r[0]})
+ if not names:
+ return {"available": False, "reason": "该实体无产品级销售数据(product_sales 为空)", "values": [], "kpi_scope": sorted(PRODUCT_SCOPED_CODES)}
+ span = db.query(ProductSales.period_month).filter(
+ ProductSales.entity_id == entity_id,
+ ).distinct().all()
+ months = sorted({str(r[0]) for r in span if r[0]})
+ return {
+ "available": True,
+ "reason": None,
+ "values": [{"code": n, "name": n} for n in names],
+ "kpi_scope": sorted(PRODUCT_SCOPED_CODES),
+ "period_coverage": {"min": months[0] if months else None, "max": months[-1] if months else None, "months": months},
+ "source": "product_sales.product_name(按 period_month 聚合,元→万元)",
+ }
+
+
+# ── 切片计算 ──────────────────────────────────────────────
+def _slice_kpi_month(db: Session, kpi: KPIDefinition, entity_id: int, period: str) -> tuple[Optional[float], Optional[str]]:
+ rows = db.query(KPIValue).filter(
+ KPIValue.entity_id == entity_id,
+ KPIValue.kpi_id == kpi.id,
+ KPIValue.period == period,
+ actual_only_clause(),
+ ).order_by(KPIValue.id.desc()).all()
+ if not rows:
+ return None, f"{period} 无实测值"
+ # 同期多行:优先 verified/calculated(strategy_board 的同口径排序)
+ rank = {"verified": 3, "calculated": 2, "pending": 1}
+ best = sorted(rows, key=lambda v: rank.get((v.data_status or "").lower(), 0), reverse=True)[0]
+ return (float(best.actual_value) if best.actual_value is not None else None), None
+
+
+def _slice_kpi_period(db: Session, kpi: KPIDefinition, entity_id: int, ptype: str, period: str) -> tuple[Optional[float], str, list[str], Optional[str]]:
+ """时间切片:月=精确值;季/年=区间聚合(年优先取 YYYY 行)。"""
+ if ptype == "month":
+ val, err = _slice_kpi_month(db, kpi, entity_id, period)
+ return val, "exact", [period], err
+
+ if ptype == "year":
+ val, err = _slice_kpi_month(db, kpi, entity_id, period)
+ if val is not None:
+ return val, "exact", [period], None
+
+ months = _months_of(ptype, period)
+ rows = db.query(KPIValue).filter(
+ KPIValue.entity_id == entity_id,
+ KPIValue.kpi_id == kpi.id,
+ KPIValue.period.in_(months),
+ actual_only_clause(),
+ ).all()
+ val, agg, used, _ = _aggregate(rows, kpi.unit, kpi)
+ expected = 3 if ptype == "quarter" else 12
+ if val is None:
+ return None, "none", [], f"{period} 无实测值(区间内无月度数据)"
+ partial = len(used) < expected and agg != "latest"
+ note = f"{period} 仅 {len(used)}/{expected} 个月有实测值" if partial else None
+ return val, agg, used, note
+
+
+def _slice_kpi_product(db: Session, kpi: KPIDefinition, entity_id: int, product: str, ptype: str, period: str) -> tuple[Optional[float], str, list[str], Optional[str]]:
+ """产品切片:仅对 PRODUCT_SCOPED_CODES 生效,并按「逐月源一致性」闸门。
+
+ `product_sales` 月合计与 KPI 同月实测值必须吻合(±1%)才纳入;不吻合的月份
+ 视为源未对齐(例如 2026-08 产品表 8.08 万 vs KPI 14.45 万),
+ 不做缩放、不并入,避免用产品表冒充 KPI 总额。
+ """
+ if kpi.kpi_code not in PRODUCT_SCOPED_CODES:
+ return None, "unsupported", [], "该 KPI 无产品级取数(维度未落库),不回退总额"
+ months = _months_of(ptype, period)
+
+ # 逐月源一致性:product_sales 全量合计(元) vs KPI 实测值(万元)
+ totals = db.query(ProductSales.period_month, ProductSales.sales_amount).filter(
+ ProductSales.entity_id == entity_id, ProductSales.period_month.in_(months),
+ ).all()
+ by_month: dict[str, float] = {}
+ for pm, amt in totals:
+ by_month[str(pm)] = by_month.get(str(pm), 0.0) + float(amt or 0)
+
+ kpi_month: dict[str, float] = {}
+ for m in months:
+ v, _err = _slice_kpi_month(db, kpi, entity_id, m)
+ if v is not None:
+ kpi_month[m] = float(v) * 10000.0 # 万元 → 元
+
+ covered = []
+ misaligned = []
+ for m in months:
+ if m not in by_month or m not in kpi_month:
+ continue
+ base = kpi_month[m]
+ if base == 0:
+ continue
+ if abs(by_month[m] - base) / abs(base) <= PRODUCT_SOURCE_TOLERANCE:
+ covered.append(m)
+ else:
+ misaligned.append(m)
+
+ if not covered:
+ detail = ""
+ if misaligned:
+ detail = "(源未对齐月份:" + "、".join(misaligned) + ")"
+ return None, "uncovered", [], f"产品表与 KPI 实测值不一致{detail},该期间不做产品切片、不回退总额"
+
+ rows = db.query(ProductSales).filter(
+ ProductSales.entity_id == entity_id,
+ ProductSales.product_name == product,
+ ProductSales.period_month.in_(covered),
+ ).all()
+ if not rows:
+ return None, "none", [], f"产品「{product}」在 {'、'.join(covered)} 无销售数据"
+
+ used = sorted({r.period_month for r in rows})
+ total = sum(float(r.sales_amount or 0) for r in rows) / 10000.0 # 元 → 万元
+ note = None
+ if len(covered) < len(months):
+ note = f"仅 {'、'.join(covered)} 通过源一致性校验" + (f";未对齐:{'、'.join(misaligned)}" if misaligned else "")
+ return round(total, 4), "sum_of_product_sales", used, note
+
+
+def build_slice(db: Session, entity_id: int, map_id: int, period: Optional[str], store: Optional[str], product: Optional[str]) -> dict:
+ _map_or_404(db, map_id, entity_id)
+ objs, defs = _map_kpis(db, map_id, entity_id)
+ ptype, pnorm = _parse_period(period)
+ if period and not ptype:
+ raise HTTPException(400, "period 格式应为 YYYY / YYYY-Qn / YYYY-MM")
+ if not pnorm:
+ # 未选时间:取该地图 KPI 的最新实测期(沿用现有「最新值」语义,并显式标注)
+ latest = None
+ ids = [k.id for k in defs.values()]
+ if ids:
+ rows = db.query(KPIValue.period).filter(
+ KPIValue.entity_id == entity_id, KPIValue.kpi_id.in_(ids), actual_only_clause(),
+ ).all()
+ cand = [str(r[0]) for r in rows if r[0]]
+ months = sorted(p for p in cand if _MONTH_RE.match(p))
+ pnorm = months[-1] if months else (sorted(cand)[-1] if cand else None)
+ if pnorm:
+ ptype = "month" if _MONTH_RE.match(pnorm) else ("quarter" if _QUARTER_RE.match(pnorm) else "year")
+ if latest is None and not pnorm:
+ latest = None
+ ptype = ptype or "month"
+ pnorm = pnorm or ""
+
+ dims_meta = {
+ "time": {"available": bool(pnorm), "period": pnorm, "type": ptype,
+ "reason": None if pnorm else "该地图 KPI 无实测值"},
+ "store": _store_options(db, entity_id) if store else {"available": True, "selected": None},
+ "product": _product_options(db, entity_id) if product else {"available": True, "selected": None},
+ }
+ if store:
+ dims_meta["store"]["selected"] = store
+ if product:
+ dims_meta["product"]["selected"] = product
+
+ objectives: list[dict] = []
+ any_available = False
+ empty_reasons: list[str] = []
+
+ for obj in objs:
+ codes = [c for c in _obj_kpi_codes(obj.kpis) if c in defs]
+ kpi_rows: list[dict] = []
+ worst = "green"
+ for code in codes:
+ kpi = defs[code]
+ target = None
+ basis = "no_period"
+ actual = None
+ agg = "none"
+ used: list[str] = []
+ reason: Optional[str] = None
+
+ if store:
+ reason = "该维度暂无数据:无门店级 KPI 事实表,不回退总额"
+ elif not pnorm:
+ reason = "无可用实测期间"
+ elif product:
+ actual, agg, used, reason = _slice_kpi_product(db, kpi, entity_id, product, ptype, pnorm)
+ target, basis = _scaled_target(kpi, ptype, pnorm[:4] if ptype == "year" else None, db, entity_id)
+ # 产品切片只覆盖通过源一致性校验的月份 ⇒ 流量型目标按覆盖月数缩放
+ if actual is not None and _unit_kind(kpi.unit, kpi) == "flow" and used:
+ base = _base_target(kpi)
+ if base is not None:
+ target = round(base * len(used), 4)
+ basis = f"target_monthly×{len(used)}(覆盖月数)"
+ else:
+ actual, agg, used, note = _slice_kpi_period(db, kpi, entity_id, ptype, pnorm)
+ target, basis = _scaled_target(kpi, ptype, pnorm[:4] if ptype == "year" else None, db, entity_id)
+ reason = note
+
+ available = actual is not None
+ act = float(actual) if actual is not None else None
+ tgt = float(target) if target is not None else None
+ ratio = None
+ level = "gray"
+ if act is not None and _unit_kind(kpi.unit) == "status":
+ ratio = 1.0 if act >= 1 else 0.0
+ level = "green" if ratio == 1.0 else "red"
+ elif act is not None and tgt:
+ if _is_lower_better(kpi):
+ ratio = (tgt / act) if act > 0 else None
+ else:
+ ratio = act / tgt if tgt != 0 else None
+ level = _level_from_ratio(ratio)
+ if reason and available and agg != "exact":
+ # 部分月份可用:值可用但必须带口径说明
+ pass
+
+ if level == "red":
+ worst = "red"
+ elif level == "yellow" and worst != "red":
+ worst = "yellow"
+ elif level == "gray" and worst == "green" and not available:
+ worst = "gray"
+
+ if not available:
+ empty_reasons.append(reason or f"{code} 无数据")
+ any_available = any_available or available
+
+ kpi_rows.append({
+ "kpi_id": kpi.id,
+ "kpi_code": code,
+ "kpi_name": kpi.kpi_name,
+ "unit": kpi.unit,
+ "actual_value": actual,
+ "target_value": target,
+ "target_basis": basis,
+ "achievement_pct": (round(ratio * 100, 1) if ratio is not None else None),
+ "level": level,
+ "available": available,
+ "reason": reason,
+ "aggregation": agg,
+ "periods_used": used,
+ })
+
+ objectives.append({
+ "objective_id": obj.id,
+ "dimension_key": obj.dimension_key,
+ "name": obj.name,
+ "kpis": kpi_rows,
+ "level": worst if kpi_rows else "gray",
+ "has_data": any(k["available"] for k in kpi_rows),
+ })
+
+ empty = not any_available
+ empty_reason = None
+ if empty:
+ if store:
+ empty_reason = "该维度暂无数据:库内无门店级 KPI 事实表(G4 未落地),不回退总额"
+ elif product:
+ specific = next((r for r in empty_reasons if "产品" in r), None)
+ empty_reason = specific or "该维度暂无数据:所选产品未覆盖该地图 KPI,不回退总额"
+ elif not pnorm:
+ empty_reason = "该地图 KPI 无实测值"
+ else:
+ empty_reason = empty_reasons[0] if empty_reasons else "所选切片无实测值"
+
+ return {
+ "map_id": map_id,
+ "entity_id": entity_id,
+ "slice": {"period": pnorm, "period_type": ptype, "store": store, "product": product},
+ "empty": empty,
+ "empty_reason": empty_reason,
+ "dims": dims_meta,
+ "objectives": objectives,
+ "caliber": {
+ "actual_scope": "实测区(预估行排除)",
+ "achievement_thresholds": {"success": SUCCESS_RATE, "warning": WARNING_RATE},
+ "aggregation": {"flow": "sum_of_months", "ratio": "mean_of_months", "status": "latest"},
+ "no_fallback": "维度无数据时返回 null,不用总额/其它期间冒充",
+ },
+ }
+
+
+# ── 路由 ──────────────────────────────────────────────────
+@router.get("/{map_id}/dimension-options")
+def get_dimension_options(
+ map_id: int,
+ db: Session = Depends(get_db),
+ entity_id: int = Depends(get_entity_id),
+):
+ """战略地图三维候选值(时间真实可切;门店/产品按库内事实显式标注可用性)。"""
+ _map_or_404(db, map_id, entity_id)
+ _objs, defs = _map_kpis(db, map_id, entity_id)
+ ids = [k.id for k in defs.values()]
+ return {
+ "map_id": map_id,
+ "entity_id": entity_id,
+ "time": _time_options(db, ids, entity_id),
+ "store": _store_options(db, entity_id),
+ "product": _product_options(db, entity_id),
+ "kpi_count": len(defs),
+ }
+
+
+@router.get("/{map_id}/slice")
+def get_map_slice(
+ map_id: int,
+ period: Optional[str] = Query(None, description="YYYY / YYYY-Qn / YYYY-MM;空=最新实测期"),
+ store: Optional[str] = Query(None, description="门店名称;库内无门店级事实时显式空态"),
+ product: Optional[str] = Query(None, description="产品名称(product_sales)"),
+ db: Session = Depends(get_db),
+ entity_id: int = Depends(get_entity_id),
+):
+ """按 时间/门店/产品 切片重算地图 KPI(只读;无数据 ⇒ null + 原因,绝不回退总额)。"""
+ return build_slice(db, entity_id, map_id, period, store, product)
diff --git a/backend/app/main.py b/backend/app/main.py
index 53b24c0d..3260704e 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -7,7 +7,7 @@ from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from dotenv import load_dotenv
from app.database import init_db, get_db
-from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers, treasury, doc_center
+from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers, treasury, doc_center, map_dimensions
from app.utils.cache import clear_all as clear_cache, delete as delete_cache
from scripts.erp_sync import run_sync as run_erp_sync
from app.auth_middleware import require_auth, require_auth_401
@@ -67,6 +67,7 @@ app.include_router(cashflow_structure.compat_router)
app.include_router(cost.router)
app.include_router(predict.router)
app.include_router(growth_quality.router)
+app.include_router(map_dimensions.router)
app.include_router(products.router)
app.include_router(reports.router)
app.include_router(security.router)
diff --git a/backend/scripts/_a3_product.py b/backend/scripts/_a3_product.py
new file mode 100644
index 00000000..04303fb8
--- /dev/null
+++ b/backend/scripts/_a3_product.py
@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+"""A3 产品切片正例:2026-07(源一致性通过)逐产品核对。"""
+import os, sys
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+for line in open('/root/cma-management/backend/.env'):
+ line = line.strip()
+ if '=' in line and not line.startswith('#') and line.split('=')[0].startswith('CMA_DB'):
+ k, v = line.split('=', 1); os.environ.setdefault(k, v)
+from app.database import get_session_local
+from app.api.map_dimensions import build_slice
+db = get_session_local()()
+
+for prod in ("酣客酱酒", "禧酱", "标准版"):
+ r = build_slice(db, 1, 65, "2026-07", None, prod)
+ rev = [k for o in r["objectives"] for k in o["kpis"] if k["kpi_code"] == "F_REVENUE"][0]
+ print(prod, "| empty:", r["empty"], "|", r["empty_reason"], "| rev:", rev["actual_value"], "/", rev["target_value"], rev["achievement_pct"], rev["aggregation"], rev["periods_used"], "| note:", rev["reason"])
+
+r = build_slice(db, 1, 65, "2026-Q3", None, "酣客酱酒")
+rev = [k for o in r["objectives"] for k in o["kpis"] if k["kpi_code"] == "F_REVENUE"][0]
+print("Q3 酣客酱酒 | empty:", r["empty"], "| rev:", rev["actual_value"], "/", rev["target_value"], rev["achievement_pct"], rev["aggregation"], rev["periods_used"], "| note:", rev["reason"])
+
+r = build_slice(db, 1, 65, None, None, None)
+print("默认(最新期):", r["slice"], [ (k["kpi_code"],k["actual_value"]) for o in r["objectives"] for k in o["kpis"] if k["available"]][:6])
+print("store 空态:", build_slice(db, 1, 65, "2026-07", "一店", None)["empty_reason"])
+db.close()
diff --git a/backend/scripts/_a3_selftest.py b/backend/scripts/_a3_selftest.py
new file mode 100644
index 00000000..3f79178c
--- /dev/null
+++ b/backend/scripts/_a3_selftest.py
@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+"""A3 自测:直接调 build_slice(不启服务),校验真实库切片与空态。"""
+import os, sys, json
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+for line in open('/root/cma-management/backend/.env'):
+ line = line.strip()
+ if '=' in line and not line.startswith('#') and line.split('=')[0].startswith(('CMA_DB', 'ERP_DB')):
+ k, v = line.split('=', 1); os.environ.setdefault(k, v)
+
+from app.database import get_session_local
+from app.api.map_dimensions import build_slice, _time_options, _store_options, _product_options
+
+db = get_session_local()()
+
+def brief(r):
+ print("slice:", r["slice"], "| empty:", r["empty"], "|", r["empty_reason"])
+ for o in r["objectives"]:
+ ks = [f'{k["kpi_code"]}={k["actual_value"]}/{k["target_value"]}@{k["achievement_pct"]}%({k["level"]},{k["aggregation"]}){"" if k["available"] else "[空:"+str(k["reason"])[:36]+"]"}' for k in o["kpis"]]
+ if ks:
+ print(" ", o["dimension_key"], o["name"], "|", "; ".join(ks))
+
+print("=== 1) 2026-08 月切片 (map 65)")
+brief(build_slice(db, 1, 65, "2026-08", None, None))
+print("\n=== 2) 2026-Q3 季切片")
+brief(build_slice(db, 1, 65, "2026-Q3", None, None))
+print("\n=== 3) 2026 年切片")
+brief(build_slice(db, 1, 65, "2026", None, None))
+print("\n=== 4) 产品切片 酣客酱酒 @2026-08")
+brief(build_slice(db, 1, 65, "2026-08", None, "酣客酱酒"))
+print("\n=== 5) 产品切片 不存在产品 @2026-08")
+brief(build_slice(db, 1, 65, "2026-08", None, "不存在的产品X"))
+print("\n=== 6) 门店切片 一店 @2026-08")
+brief(build_slice(db, 1, 65, "2026-08", "一店", None))
+print("\n=== 7) entity 隔离: map65 + entity2")
+try:
+ build_slice(db, 2, 65, "2026-08", None, None)
+ print("FAIL 未拦截")
+except Exception as e:
+ print("OK 拦截:", type(e).__name__, getattr(e, "status_code", ""), getattr(e, "detail", e))
+
+print("\n=== 8) options map65")
+o = build_slice(db, 1, 65, None, None, None)
+print("time dims:", json.dumps(o["dims"]["time"], ensure_ascii=False)[:300])
+print("store:", json.dumps(o["dims"]["store"], ensure_ascii=False)[:300])
+print("product:", json.dumps(o["dims"]["product"], ensure_ascii=False)[:300])
+db.close()
diff --git a/frontend/src/components/map-canvas/MapDimensionFilter.vue b/frontend/src/components/map-canvas/MapDimensionFilter.vue
new file mode 100644
index 00000000..40b67a47
--- /dev/null
+++ b/frontend/src/components/map-canvas/MapDimensionFilter.vue
@@ -0,0 +1,242 @@
+
+