决策逻辑变更的风险链：

  修改规则阈值            修改决策树分支           修改约束条件
       │                      │                      │
       ▼                      ▼                      ▼
  ┌─────────────────────────────────────────────────────┐
  │              未经验证直接上线                          │
  │  → 审批通过率异常升高/降低                              │
  │  → 合规风险（如信贷误拒）                               │
  │  → 业务损失（如库存分配不均）                            │
  └─────────────────────────────────────────────────────┘

Dry-Run 模式概览：

  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐
  │ Shadow Mode  │    │ What-If Mode │    │ Replay Mode  │
  │ 影子模式      │    │ 假设分析      │    │ 回放模式      │
  └──────┬───────┘    └──────┬───────┘    └──────┬───────┘
         │                   │                   │
         ▼                   ▼                   ▼
  生产流量并行执行       手动构造场景         历史数据回放
  新旧逻辑对比          模拟推演             批量验证
  零风险                 探索式              统计级对比

from __future__ import annotations
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any


@dataclass
class ShadowConfig:
    """影子模式配置"""
    shadow_id: str
    description: str
    primary_version: str           # 当前生产版本
    shadow_version: str            # 待验证版本
    sample_rate: float = 1.0       # 采样率 0.0-1.0
    enabled: bool = True
    start_time: datetime | None = None
    end_time: datetime | None = None
    tags: dict[str, str] = field(default_factory=dict)


@dataclass
class ShadowResult:
    """影子比较结果"""
    request_id: str
    primary_decision: str
    shadow_decision: str
    primary_confidence: float
    shadow_confidence: float
    is_divergent: bool
    divergence_details: dict[str, Any] = field(default_factory=dict)
    primary_latency_ms: float = 0.0
    shadow_latency_ms: float = 0.0
    timestamp: datetime = field(default_factory=datetime.utcnow)


class ShadowModeExecutor:
    """影子模式执行器"""

    def __init__(self, primary_engine, shadow_engine,
                 config: ShadowConfig):
        self._primary = primary_engine
        self._shadow = shadow_engine
        self._config = config
        self._results: list[ShadowResult] = []
        self._divergence_count = 0
        self._total_count = 0

    async def execute(self, context) -> tuple[Any, ShadowResult | None]:
        """执行影子模式决策"""
        import time
        import random

        # 主引擎始终执行（返回给用户）
        start = time.monotonic()
        primary_result = self._primary.evaluate(context)
        primary_latency = (time.monotonic() - start) * 1000

        # 影子引擎按采样率执行
        shadow_result_obj = None
        if (self._config.enabled and
                random.random() < self._config.sample_rate):

            start = time.monotonic()
            try:
                shadow_result = self._shadow.evaluate(context)
                shadow_latency = (time.monotonic() - start) * 1000

                # 比较结果
                shadow_result_obj = self._compare(
                    context, primary_result, shadow_result,
                    primary_latency, shadow_latency
                )
                self._results.append(shadow_result_obj)
                self._total_count += 1

                if shadow_result_obj.is_divergent:
                    self._divergence_count += 1

            except Exception as e:
                shadow_result_obj = ShadowResult(
                    request_id=context.context_id,
                    primary_decision=primary_result.decision,
                    shadow_decision=f"ERROR: {e}",
                    primary_confidence=primary_result.confidence,
                    shadow_confidence=0.0,
                    is_divergent=True,
                    divergence_details={"error": str(e)},
                    primary_latency_ms=primary_latency,
                )

        return primary_result, shadow_result_obj

    def _compare(self, context, primary, shadow,
                 p_latency: float, s_latency: float) -> ShadowResult:
        is_divergent = primary.decision != shadow.decision
        details = {}

        if is_divergent:
            details["decision_change"] = {
                "from": primary.decision,
                "to": shadow.decision,
            }
            details["confidence_delta"] = (
                shadow.confidence - primary.confidence
            )

        conf_threshold = 0.1
        if abs(primary.confidence - shadow.confidence) > conf_threshold:
            details["confidence_shift"] = {
                "primary": primary.confidence,
                "shadow": shadow.confidence,
                "delta": shadow.confidence - primary.confidence,
            }

        return ShadowResult(
            request_id=context.context_id,
            primary_decision=primary.decision,
            shadow_decision=shadow.decision,
            primary_confidence=primary.confidence,
            shadow_confidence=shadow.confidence,
            is_divergent=is_divergent,
            divergence_details=details,
            primary_latency_ms=p_latency,
            shadow_latency_ms=s_latency,
        )

    @property
    def divergence_rate(self) -> float:
        if self._total_count == 0:
            return 0.0
        return self._divergence_count / self._total_count

    def generate_report(self) -> ShadowReport:
        return ShadowReportGenerator.generate(
            self._config, self._results
        )

@dataclass
class ShadowReport:
    """影子模式分析报告"""
    shadow_id: str
    total_requests: int
    divergent_requests: int
    divergence_rate: float
    avg_primary_latency_ms: float
    avg_shadow_latency_ms: float
    decision_distribution_primary: dict[str, int]
    decision_distribution_shadow: dict[str, int]
    top_divergence_patterns: list[dict]
    recommendation: str


class ShadowReportGenerator:
    """影子报告生成器"""

    @staticmethod
    def generate(config: ShadowConfig,
                 results: list[ShadowResult]) -> ShadowReport:
        total = len(results)
        divergent = [r for r in results if r.is_divergent]

        # 决策分布
        primary_dist: dict[str, int] = {}
        shadow_dist: dict[str, int] = {}
        for r in results:
            primary_dist[r.primary_decision] = (
                primary_dist.get(r.primary_decision, 0) + 1
            )
            shadow_dist[r.shadow_decision] = (
                shadow_dist.get(r.shadow_decision, 0) + 1
            )

        # 差异模式分析
        patterns: dict[str, int] = {}
        for r in divergent:
            key = f"{r.primary_decision} -> {r.shadow_decision}"
            patterns[key] = patterns.get(key, 0) + 1

        top_patterns = sorted(
            [{"pattern": k, "count": v} for k, v in patterns.items()],
            key=lambda x: x["count"], reverse=True
        )[:10]

        # 生成建议
        div_rate = len(divergent) / total if total > 0 else 0
        if div_rate < 0.01:
            rec = "差异率极低（<1%），建议推进上线。"
        elif div_rate < 0.05:
            rec = "差异率可接受（1-5%），建议人工审查差异样本后上线。"
        elif div_rate < 0.15:
            rec = "差异率较高（5-15%），建议深入分析差异模式，调整后重新验证。"
        else:
            rec = "差异率过高（>15%），不建议上线，请回退检查逻辑变更。"

        return ShadowReport(
            shadow_id=config.shadow_id,
            total_requests=total,
            divergent_requests=len(divergent),
            divergence_rate=div_rate,
            avg_primary_latency_ms=(
                sum(r.primary_latency_ms for r in results) / total
                if total > 0 else 0
            ),
            avg_shadow_latency_ms=(
                sum(r.shadow_latency_ms for r in results) / total
                if total > 0 else 0
            ),
            decision_distribution_primary=primary_dist,
            decision_distribution_shadow=shadow_dist,
            top_divergence_patterns=top_patterns,
            recommendation=rec,
        )

影子模式分析报告：
═══════════════════════════════════

  配置: shadow-credit-v3 (采样率: 100%)
  时间: 2026-03-20 08:00 ~ 2026-03-20 20:00

  总请求数:      12,847
  差异请求数:        641
  差异率:          4.99%

  决策分布对比:
  ─────────────────────────────────
  决策类型           | 生产版本  | 影子版本
  approve           |  6,421   |  6,892  (+7.3%)
  conditional       |  3,208   |  2,891  (-9.9%)
  reject            |  3,218   |  3,064  (-4.8%)

  Top 差异模式:
  ─────────────────────────────────
  reject -> conditional    :  312 次 (48.7%)
  conditional -> approve   :  198 次 (30.9%)
  approve -> conditional   :   89 次 (13.9%)
  conditional -> reject    :   42 次 ( 6.5%)

  延迟对比:
  ─────────────────────────────────
  生产引擎 P50:  2.1ms    影子引擎 P50:  2.3ms
  生产引擎 P99:  8.7ms    影子引擎 P99:  9.1ms

  建议: 差异率可接受（1-5%），建议人工审查差异样本后上线。

@dataclass
class WhatIfScenario:
    """假设场景"""
    scenario_id: str
    name: str
    base_inputs: dict[str, Any]
    modifications: list[WhatIfModification]
    description: str = ""


@dataclass
class WhatIfModification:
    """单个假设修改"""
    variable: str
    original_value: Any
    hypothetical_value: Any
    description: str = ""


@dataclass
class WhatIfResult:
    """假设分析结果"""
    scenario_id: str
    original_decision: str
    original_confidence: float
    hypothetical_decision: str
    hypothetical_confidence: float
    decision_changed: bool
    modifications_applied: list[WhatIfModification]
    sensitivity: dict[str, float]    # 每个变量的敏感度


class WhatIfAnalyzer:
    """What-If 分析器"""

    def __init__(self, decision_engine):
        self._engine = decision_engine

    def analyze(self, scenario: WhatIfScenario,
                context_builder) -> WhatIfResult:
        """执行假设分析"""
        # 原始决策
        original_ctx = context_builder(scenario.base_inputs)
        original = self._engine.evaluate(original_ctx)

        # 应用修改后的决策
        modified_inputs = dict(scenario.base_inputs)
        for mod in scenario.modifications:
            modified_inputs[mod.variable] = mod.hypothetical_value

        modified_ctx = context_builder(modified_inputs)
        modified = self._engine.evaluate(modified_ctx)

        # 计算敏感度
        sensitivity = self._compute_sensitivity(
            scenario.base_inputs, scenario.modifications, context_builder
        )

        return WhatIfResult(
            scenario_id=scenario.scenario_id,
            original_decision=original.decision,
            original_confidence=original.confidence,
            hypothetical_decision=modified.decision,
            hypothetical_confidence=modified.confidence,
            decision_changed=original.decision != modified.decision,
            modifications_applied=scenario.modifications,
            sensitivity=sensitivity,
        )

    def _compute_sensitivity(self, base: dict,
                             modifications: list[WhatIfModification],
                             builder) -> dict[str, float]:
        """计算每个变量的独立敏感度"""
        sensitivities = {}
        base_ctx = builder(base)
        base_result = self._engine.evaluate(base_ctx)

        for mod in modifications:
            single_mod = dict(base)
            single_mod[mod.variable] = mod.hypothetical_value
            mod_ctx = builder(single_mod)
            mod_result = self._engine.evaluate(mod_ctx)

            conf_delta = abs(mod_result.confidence - base_result.confidence)
            decision_change = 1.0 if mod_result.decision != base_result.decision else 0.0
            sensitivities[mod.variable] = conf_delta + decision_change

        return sensitivities

    def batch_analyze(self, scenarios: list[WhatIfScenario],
                      builder) -> list[WhatIfResult]:
        """批量假设分析"""
        return [self.analyze(s, builder) for s in scenarios]

    def sweep(self, base_inputs: dict, variable: str,
              values: list, builder) -> list[dict]:
        """变量扫描：单变量从 min 到 max 的决策变化"""
        results = []
        for val in values:
            inputs = dict(base_inputs)
            inputs[variable] = val
            ctx = builder(inputs)
            result = self._engine.evaluate(ctx)
            results.append({
                "value": val,
                "decision": result.decision,
                "confidence": result.confidence,
            })
        return results

变量扫描：credit_score 从 400 到 800

  credit_score | 决策              | 置信度
  ─────────────|──────────────────|────────
  400          | reject           | 0.95
  450          | reject           | 0.92
  500          | reject           | 0.88
  550          | reject           | 0.78  ← 开始不确定
  600          | conditional      | 0.65
  620          | conditional      | 0.72
  650          | conditional      | 0.80
  700          | approve          | 0.88  ← 决策翻转点
  750          | approve          | 0.93
  800          | approve          | 0.97

  决策翻转点: credit_score = 700
  敏感区间: [550, 700] — 在此区间内微小变化会显著影响决策

from datetime import datetime, timedelta


@dataclass
class ReplayConfig:
    """回放配置"""
    replay_id: str
    source: str                      # "database", "file", "kafka"
    time_range: tuple[datetime, datetime]
    new_engine_version: str
    batch_size: int = 1000
    max_records: int = 100_000


@dataclass
class ReplayStats:
    """回放统计"""
    total: int = 0
    same_decision: int = 0
    different_decision: int = 0
    new_approve_rate: float = 0.0
    old_approve_rate: float = 0.0
    latency_improvement_pct: float = 0.0
    error_count: int = 0


class ReplayEngine:
    """历史决策回放引擎"""

    def __init__(self, old_engine, new_engine, decision_store):
        self._old = old_engine
        self._new = new_engine
        self._store = decision_store

    async def replay(self, config: ReplayConfig) -> ReplayStats:
        """回放历史决策"""
        stats = ReplayStats()
        old_approves = 0
        new_approves = 0

        records = await self._store.query(
            start=config.time_range[0],
            end=config.time_range[1],
            limit=config.max_records,
        )

        for batch_start in range(0, len(records), config.batch_size):
            batch = records[batch_start:batch_start + config.batch_size]

            for record in batch:
                try:
                    old_result = record["original_decision"]
                    new_result = self._new.evaluate(
                        self._rebuild_context(record)
                    )

                    stats.total += 1

                    if old_result == new_result.decision:
                        stats.same_decision += 1
                    else:
                        stats.different_decision += 1

                    if old_result in ("approve", "conditional_approve"):
                        old_approves += 1
                    if new_result.decision in ("approve", "conditional_approve"):
                        new_approves += 1

                except Exception:
                    stats.error_count += 1

        if stats.total > 0:
            stats.old_approve_rate = old_approves / stats.total
            stats.new_approve_rate = new_approves / stats.total

        return stats

    def _rebuild_context(self, record: dict):
        """从历史记录重建 DecisionContext"""
        builder = DecisionContextBuilder(record["domain"])
        for k, v in record.get("inputs", {}).items():
            builder = builder.with_inputs(**{k: v})
        return builder.build()

历史回放报告：
═══════════════════════════════════

  回放 ID: replay-credit-v3-upgrade
  时间范围: 2026-02-01 ~ 2026-03-01
  总记录数: 89,421

  决策一致性:
  ─────────────────────────────────
  一致:    84,602 (94.6%)
  不一致:   4,819 ( 5.4%)

  审批率对比:
  ─────────────────────────────────
  旧版本审批率:  62.3%
  新版本审批率:  64.8%  (+2.5pp)

  不一致详情:
  ─────────────────────────────────
  reject -> approve        :  2,104 (43.7%)
  reject -> conditional    :  1,287 (26.7%)
  conditional -> approve   :    891 (18.5%)
  approve -> conditional   :    412 ( 8.5%)
  approve -> reject        :    125 ( 2.6%)

  风险评估:
  - 新增审批 2,104 笔，需评估违约风险
  - 降级为有条件审批 412 笔，影响客户体验
  - 新增拒绝 125 笔，需合规审查

syntax = "proto3";
package onto.decision.dryrun.v1;

service DryRunService {
    rpc CreateShadow(CreateShadowRequest) returns (CreateShadowResponse);
    rpc StopShadow(StopShadowRequest) returns (StopShadowResponse);
    rpc GetShadowReport(GetReportRequest) returns (ShadowReportResponse);
    rpc WhatIf(WhatIfRequest) returns (WhatIfResponse);
    rpc VariableSweep(SweepRequest) returns (SweepResponse);
    rpc StartReplay(ReplayRequest) returns (stream ReplayProgress);
}

message WhatIfRequest {
    string domain = 1;
    map<string, string> base_inputs = 2;
    repeated Modification modifications = 3;
}

message Modification {
    string variable = 1;
    string original_value = 2;
    string hypothetical_value = 3;
}

message WhatIfResponse {
    string original_decision = 1;
    double original_confidence = 2;
    string hypothetical_decision = 3;
    double hypothetical_confidence = 4;
    bool decision_changed = 5;
    map<string, double> sensitivity = 6;
}

class DryRunServiceImpl:
    """Dry-Run gRPC 服务"""

    def __init__(self, engine_registry, decision_store):
        self._registry = engine_registry
        self._store = decision_store
        self._active_shadows: dict[str, ShadowModeExecutor] = {}

    async def CreateShadow(self, request, context):
        primary = self._registry.get(request.primary_version)
        shadow = self._registry.get(request.shadow_version)

        config = ShadowConfig(
            shadow_id=request.shadow_id,
            description=request.description,
            primary_version=request.primary_version,
            shadow_version=request.shadow_version,
            sample_rate=request.sample_rate,
        )

        executor = ShadowModeExecutor(primary, shadow, config)
        self._active_shadows[config.shadow_id] = executor

        return {"shadow_id": config.shadow_id, "status": "active"}

    async def WhatIf(self, request, context):
        engine = self._registry.get_current()
        analyzer = WhatIfAnalyzer(engine)

        modifications = [
            WhatIfModification(
                variable=m.variable,
                original_value=m.original_value,
                hypothetical_value=self._parse(m.hypothetical_value),
            )
            for m in request.modifications
        ]

        scenario = WhatIfScenario(
            scenario_id=f"whatif-{id(request)}",
            name="ad-hoc",
            base_inputs={k: self._parse(v)
                         for k, v in request.base_inputs.items()},
            modifications=modifications,
        )

        result = analyzer.analyze(
            scenario,
            lambda inputs: DecisionContextBuilder(request.domain)
            .with_inputs(**inputs).build()
        )

        return {
            "original_decision": result.original_decision,
            "original_confidence": result.original_confidence,
            "hypothetical_decision": result.hypothetical_decision,
            "hypothetical_confidence": result.hypothetical_confidence,
            "decision_changed": result.decision_changed,
            "sensitivity": result.sensitivity,
        }

    def _parse(self, s: str):
        try:
            return int(s)
        except ValueError:
            pass
        try:
            return float(s)
        except ValueError:
            return s

class ShadowGuard:
    """影子模式安全守护"""

    def __init__(self, max_latency_ratio: float = 2.0,
                 max_error_rate: float = 0.05):
        self._max_latency_ratio = max_latency_ratio
        self._max_error_rate = max_error_rate
        self._error_count = 0
        self._total_count = 0

    def check(self, result: ShadowResult) -> bool:
        """检查影子执行是否安全"""
        self._total_count += 1

        # 延迟检查
        if result.primary_latency_ms > 0:
            ratio = result.shadow_latency_ms / result.primary_latency_ms
            if ratio > self._max_latency_ratio:
                return False

        # 错误率检查
        if result.shadow_decision.startswith("ERROR"):
            self._error_count += 1

        error_rate = self._error_count / self._total_count
        if error_rate > self._max_error_rate:
            return False

        return True

    def should_disable(self) -> bool:
        """是否应该禁用影子模式"""
        if self._total_count < 100:
            return False
        error_rate = self._error_count / self._total_count
        return error_rate > self._max_error_rate * 2

# .gitlab-ci.yml 中的决策验证阶段
decision-dry-run:
  stage: validate
  script:
    - python -m onto.decision.dryrun replay
        --config replay-config.yaml
        --new-version $CI_COMMIT_SHA
        --time-range "7d"
        --max-records 50000
    - python -m onto.decision.dryrun check
        --max-divergence-rate 0.10
        --max-new-reject-increase 0.02
  artifacts:
    paths:
      - reports/dryrun-report.json
  rules:
    - if: $CI_MERGE_REQUEST_TARGET_BRANCH == "main"
      changes:
        - decision-trees/**/*
        - intelligence-Layer/reasoning/**/*

class DryRunQualityGate:
    """Dry-Run 质量门"""

    def __init__(self, max_divergence: float = 0.10,
                 max_reject_increase: float = 0.02,
                 max_latency_increase_pct: float = 20.0):
        self._max_div = max_divergence
        self._max_reject = max_reject_increase
        self._max_latency = max_latency_increase_pct

    def evaluate(self, report: ShadowReport | ReplayStats) -> dict:
        """评估是否通过质量门"""
        checks = []

        if isinstance(report, ShadowReport):
            checks.append({
                "name": "divergence_rate",
                "value": report.divergence_rate,
                "threshold": self._max_div,
                "passed": report.divergence_rate <= self._max_div,
            })

        if isinstance(report, ReplayStats):
            reject_increase = (
                (1 - report.new_approve_rate) -
                (1 - report.old_approve_rate)
            )
            checks.append({
                "name": "reject_increase",
                "value": reject_increase,
                "threshold": self._max_reject,
                "passed": reject_increase <= self._max_reject,
            })

        all_passed = all(c["passed"] for c in checks)
        return {
            "passed": all_passed,
            "checks": checks,
            "recommendation": (
                "PROCEED" if all_passed else "BLOCK"
            ),
        }

class EngineRegistry:
    """决策引擎版本注册表"""

    def __init__(self):
        self._versions: dict[str, Any] = {}
        self._current: str = ""

    def register(self, version: str, engine) -> None:
        self._versions[version] = engine

    def set_current(self, version: str) -> None:
        if version not in self._versions:
            raise ValueError(f"Version {version} not registered")
        self._current = version

    def get(self, version: str):
        engine = self._versions.get(version)
        if engine is None:
            raise ValueError(f"Version {version} not found")
        return engine

    def get_current(self):
        return self.get(self._current)

    def list_versions(self) -> list[str]:
        return list(self._versions.keys())

采样率选择指南：

  QPS 级别       | 建议采样率  | 每日样本量
  ──────────────|──────────|──────────
  < 100 QPS     | 100%     | ~8.6M
  100-1000 QPS  | 10-50%   | ~4.3M-43M
  1000-10000 QPS| 1-10%    | ~864K-8.6M
  > 10000 QPS   | 0.1-1%   | ~864K

# 场景：信贷规则阈值调整验证

# 1. 创建影子模式
config = ShadowConfig(
    shadow_id="shadow-credit-threshold-adjust",
    description="将 credit_score 阈值从 650 调整为 620",
    primary_version="credit-v2.1",
    shadow_version="credit-v2.2-candidate",
    sample_rate=0.5,     # 50% 采样
)

executor = ShadowModeExecutor(
    primary_engine=registry.get("credit-v2.1"),
    shadow_engine=registry.get("credit-v2.2-candidate"),
    config=config,
)

# 2. 执行 7 天后查看报告
report = executor.generate_report()
print(f"差异率: {report.divergence_rate:.1%}")
print(f"建议: {report.recommendation}")

# 3. What-If 分析边界案例
analyzer = WhatIfAnalyzer(registry.get("credit-v2.2-candidate"))
sweep_results = analyzer.sweep(
    base_inputs={"credit_score": 580, "debt_ratio": 0.45,
                 "annual_income": 200000},
    variable="credit_score",
    values=list(range(550, 750, 10)),
    builder=lambda inputs: DecisionContextBuilder("credit")
    .with_inputs(**inputs).build()
)

# 4. 质量门检查
gate = DryRunQualityGate(max_divergence=0.10)
result = gate.evaluate(report)
# {"passed": True, "recommendation": "PROCEED"}