决策树分析器 decision-tree-analyzer

决策树分析器是一个AI技能,能够构建决策树,计算期望货币值(EMV),分析风险,应用效用理论,进行敏感性分析,帮助在不确定性下做出最优决策。

数据分析 0 次安装 0 次浏览 更新于 2/25/2026

以下是对提供的决策树分析技能的中文翻译:

decision-tree-analyzer

你是 决策树分析器 - 一个专门用于决策树分析的技能,包括期望值计算、风险分析和效用理论应用。

概览

此技能使AI能够进行决策树分析,包括:

  • 决策树构建
  • 期望货币值(EMV)计算
  • 完美信息的期望值(EVPI)
  • 样本信息的期望值(EVSI)
  • 风险档案和敏感性
  • 效用函数应用
  • 决策回滚分析
  • 多阶段顺序决策

能力

1. 决策树构建

import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum

class NodeType(Enum):
    DECISION = "decision"
    CHANCE = "chance"
    TERMINAL = "terminal"

@dataclass
class TreeNode:
    node_id: str
    node_type: NodeType
    name: str
    value: float = 0  # 用于终端节点
    probability: float = 1.0  # 用于机会分支
    children: List['TreeNode'] = None
    parent: Optional['TreeNode'] = None

    def __post_init__(self):
        if self.children is None:
            self.children = []

def build_decision_tree(structure: dict):
    """
    从结构定义构建决策树

    structure: 嵌套字典定义树
    {
        'type': 'decision',
        'name': '初始决策',
        'branches': [
            {
                'name': '选项 A',
                'type': 'chance',
                'branches': [
                    {'name': '高', 'probability': 0.3, 'value': 100},
                    {'name': '低', 'probability': 0.7, 'value': 50}
                ]
            }
        ]
    }
    """
    def build_node(data, parent=None, node_id='root'):
        node_type = NodeType(data.get('type', 'terminal'))

        node = TreeNode(
            node_id=node_id,
            node_type=node_type,
            name=data.get('name', ''),
            value=data.get('value', 0),
            probability=data.get('probability', 1.0),
            parent=parent
        )

        if 'branches' in data:
            for i, branch in enumerate(data['branches']):
                child = build_node(branch, node, f"{node_id}_{i}")
                node.children.append(child)

        return node

    root = build_node(structure)
    return root

2. 期望货币值(EMV)

def calculate_emv(node: TreeNode):
    """
    使用回滚分析计算期望货币值
    """
    results = {}

    def rollback(n):
        if n.node_type == NodeType.TERMINAL:
            return n.value

        if n.node_type == NodeType.CHANCE:
            # EMV是结果的加权平均
            emv = sum(child.probability * rollback(child) for child in n.children)
            results[n.node_id] = {'name': n.name, 'emv': emv, 'type': 'chance'}
            return emv

        if n.node_type == NodeType.DECISION:
            # 选择最大EMV分支
            child_values = [(child, rollback(child)) for child in n.children]
            best_child, best_value = max(child_values, key=lambda x: x[1])
            results[n.node_id] = {
                'name': n.name,
                'emv': best_value,
                'type': 'decision',
                'best_choice': best_child.name,
                'all_choices': {c.name: v for c, v in child_values}
            }
            return best_value

    final_emv = rollback(node)

    return {
        "emv": round(final_emv, 2),
        "node_values": results,
        "optimal_strategy": extract_optimal_strategy(results)
    }

def extract_optimal_strategy(results):
    """提取最优决策路径"""
    strategy = []
    for node_id, data in results.items():
        if data['type'] == 'decision':
            strategy.append({
                'decision': data['name'],
                'choice': data['best_choice'],
                'emv': round(data['emv'], 2)
            })
    return strategy

3. 完美信息的期望值(EVPI)

def calculate_evpi(decision_node: TreeNode):
    """
    计算完美信息的期望值

    EVPI = 有完美信息时的EV - 无信息时的EMV
    """
    # 首先,获取无完美信息时的EMV
    emv_result = calculate_emv(decision_node)
    emv_without = emv_result['emv']

    # 计算有完美信息时的EV
    # 对于每个自然状态,选择最佳决策
    states = collect_chance_outcomes(decision_node)

    ev_with_perfect = 0
    perfect_decisions = {}

    for state, prob in states.items():
        # 对于这个状态,找到最佳决策
        best_value = float('-inf')
        best_decision = None

        for decision_branch in decision_node.children:
            value = get_value_given_state(decision_branch, state)
            if value > best_value:
                best_value = value
                best_decision = decision_branch.name

        ev_with_perfect += prob * best_value
        perfect_decisions[state] = {'decision': best_decision, 'value': best_value}

    evpi = ev_with_perfect - emv_without

    return {
        "evpi": round(evpi, 2),
        "ev_with_perfect_info": round(ev_with_perfect, 2),
        "emv_without_info": round(emv_without, 2),
        "perfect_decisions": perfect_decisions,
        "interpretation": f"为完美信息支付高达 ${round(evpi, 2)} 的价值"
    }

def collect_chance_outcomes(node, outcomes=None, current_prob=1.0):
    """收集所有机会结果及其概率"""
    if outcomes is None:
        outcomes = {}

    if node.node_type == NodeType.TERMINAL:
        return outcomes

    if node.node_type == NodeType.CHANCE:
        for child in node.children:
            outcomes[child.name] = child.probability
            collect_chance_outcomes(child, outcomes, current_prob * child.probability)

    for child in node.children:
        collect_chance_outcomes(child, outcomes, current_prob)

    return outcomes

def get_value_given_state(node, state):
    """给定特定状态发生时获取分支的价值"""
    # 简化 - 需要完整的树遍历
    for child in node.children:
        if child.name == state:
            return child.value if child.node_type == NodeType.TERMINAL else 0
        result = get_value_given_state(child, state)
        if result != 0:
            return result
    return 0

4. 风险档案分析

def create_risk_profile(decision_node: TreeNode, decision_choice: str = None):
    """
    创建显示结果概率分布的风险档案
    """
    outcomes = []

    def collect_outcomes(node, current_prob=1.0, path=None):
        if path is None:
            path = []

        if node.node_type == NodeType.TERMINAL:
            outcomes.append({
                'value': node.value,
                'probability': current_prob,
                'path': ' -> '.join(path)
            })
            return

        if node.node_type == NodeType.CHANCE:
            for child in node.children:
                collect_outcomes(child, current_prob * child.probability,
                               path + [child.name])

        elif node.node_type == NodeType.DECISION:
            if decision_choice:
                for child in node.children:
                    if child.name == decision_choice:
                        collect_outcomes(child, current_prob, path + [child.name])
            else:
                # 使用最优决策
                emv_result = calculate_emv(node)
                best = emv_result['node_values'].get(node.node_id, {}).get('best_choice')
                for child in node.children:
                    if child.name == best:
                        collect_outcomes(child, current_prob, path + [child.name])

    collect_outcomes(decision_node)

    # 按值聚合
    value_probs = {}
    for outcome in outcomes:
        v = outcome['value']
        value_probs[v] = value_probs.get(v, 0) + outcome['probability']

    # 计算统计数据
    values = [o['value'] for o in outcomes]
    probs = [o['probability'] for o in outcomes]

    expected_value = sum(v * p for v, p in zip(values, probs))
    variance = sum(p * (v - expected_value)**2 for v, p in zip(values, probs))
    std_dev = np.sqrt(variance)

    # 累积分布
    sorted_outcomes = sorted(value_probs.items())
    cumulative = 0
    cdf = []
    for value, prob in sorted_outcomes:
        cumulative += prob
        cdf.append({'value': value, 'cumulative_prob': cumulative})

    return {
        "outcomes": outcomes,
        "probability_distribution": value_probs,
        "statistics": {
            "expected_value": round(expected_value, 2),
            "variance": round(variance, 2),
            "std_deviation": round(std_dev, 2),
            "min_value": min(values),
            "max_value": max(values)
        },
        "cumulative_distribution": cdf
    }

5. 效用函数分析

def apply_utility_function(decision_node: TreeNode, risk_attitude: str = 'neutral',
                          risk_parameter: float = None):
    """
    应用效用函数转换货币值

    risk_attitude: 'neutral', 'averse', 'seeking'
    """
    def utility(x, attitude, param):
        if attitude == 'neutral':
            return x
        elif attitude == 'averse':
            # 指数效用:U(x) = 1 - e^(-x/R)
            R = param or 1000  # 风险容忍度
            return 1 - np.exp(-x / R)
        elif attitude == 'seeking':
            # 风险寻求的指数效用
            R = param or 1000
            return np.exp(x / R) - 1
        return x

    def inverse_utility(u, attitude, param):
        if attitude == 'neutral':
            return u
        elif attitude == 'averse':
            R = param or 1000
            return -R * np.log(1 - u) if u < 1 else float('inf')
        elif attitude == 'seeking':
            R = param or 1000
            return R * np.log(u + 1)
        return u

    # 将树转换为效用值
    def convert_node(n):
        if n.node_type == NodeType.TERMINAL:
            n.utility_value = utility(n.value, risk_attitude, risk_parameter)
        for child in n.children:
            convert_node(child)

    convert_node(decision_node)

    # 计算期望效用
    def expected_utility(n):
        if n.node_type == NodeType.TERMINAL:
            return n.utility_value

        if n.node_type == NodeType.CHANCE:
            return sum(child.probability * expected_utility(child) for child in n.children)

        if n.node_type == NodeType.DECISION:
            return max(expected_utility(child) for child in n.children)

    eu = expected_utility(decision_node)
    certainty_equivalent = inverse_utility(eu, risk_attitude, risk_parameter)

    # 与EMV比较
    emv_result = calculate_emv(decision_node)

    return {
        "expected_utility": round(eu, 4),
        "certainty_equivalent": round(certainty_equivalent, 2),
        "emv": emv_result['emv'],
        "risk_premium": round(emv_result['emv'] - certainty_equivalent, 2),
        "risk_attitude": risk_attitude,
        "interpretation": interpret_risk_attitude(certainty_equivalent, emv_result['emv'])
    }

def interpret_risk_attitude(ce, emv):
    if abs(ce - emv) < 1:
        return "风险中性 - 对期望值和确定等价物无差异"
    elif ce < emv:
        return f"风险规避 - 愿意接受比期望值少 ${round(emv - ce, 2)} 的确定性"
    else:
        return f"风险寻求 - 需要比期望值多 ${round(ce - emv, 2)} 的溢价"

6. 敏感性分析

def sensitivity_analysis(decision_node: TreeNode, parameter: str,
                        range_min: float, range_max: float, steps: int = 10):
    """
    分析决策对参数变化的敏感性
    """
    values = np.linspace(range_min, range_max, steps)
    results = []

    for val in values:
        # 修改参数(概率或值)
        modify_parameter(decision_node, parameter, val)
        emv_result = calculate_emv(decision_node)

        results.append({
            'parameter_value': round(val, 3),
            'emv': round(emv_result['emv'], 2),
            'best_decision': emv_result['optimal_strategy'][0]['choice']
                           if emv_result['optimal_strategy'] else None
        })

    # 查找交叉点
    crossovers = []
    for i in range(1, len(results)):
        if results[i]['best_decision'] != results[i-1]['best_decision']:
            crossovers.append({
                'value': results[i]['parameter_value'],
                'from': results[i-1]['best_decision'],
                'to': results[i]['best_decision']
            })

    return {
        "parameter": parameter,
        "range": {"min": range_min, "max": range_max},
        "results": results,
        "crossover_points": crossovers,
        "recommendation": generate_sensitivity_recommendation(crossovers, results)
    }

def modify_parameter(node, parameter, value):
    """在树中修改参数"""
    # 实现取决于参数规范
    pass

def generate_sensitivity_recommendation(crossovers, results):
    if not crossovers:
        return f"决策是稳健的 - 在整个范围内选择相同"
    return f"决策在 {len(crossovers)} 个点切换 - 需要仔细分析"

流程集成

此技能与以下流程集成:

  • multi-criteria-decision-analysis.js
  • risk-assessment-analysis.js
  • investment-analysis.js

输出格式

{
  "decision_tree": {
    "emv": 125000,
    "optimal_strategy": [
      {"decision": "Initial", "choice": "Expand", "emv": 125000}
    ]
  },
  "evpi": 15000,
  "risk_profile": {
    "expected_value": 125000,
    "std_deviation": 45000,
    "probability_of_loss": 0.15
  },
  "utility_analysis": {
    "certainty_equivalent": 110000,
    "risk_premium": 15000
  },
  "recommendation": "Choose Expand option with expected value of $125,000"
}

最佳实践

  1. 仔细构建 - 清晰的决策和机会节点
  2. 验证概率 - 必须在机会节点处总和为1
  3. 考虑所有结果 - 不要遗漏重要场景
  4. 测试敏感性 - 了解关键驱动因素
  5. 考虑风险态度 - EMV假设风险中性
  6. 记录假设 - 记录概率来源

限制

  • 需要概率估计
  • 树的复杂性增长迅速
  • 顺序决策增加不确定性
  • 效用函数是主观的