#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import zipfile
from copy import deepcopy
from datetime import date
from html import escape
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parent.parent
def load_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def write_json(path: Path, payload: dict[str, Any]) -> None:
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
def text(value: Any, default: str = "-") -> str:
if value is None:
return default
if isinstance(value, dict):
for key in ("zh", "text", "name", "label", "en"):
if value.get(key):
return str(value[key]).strip()
return default
if isinstance(value, list):
items = [text(item, "") for item in value]
items = [item for item in items if item]
return ";".join(items) if items else default
result = str(value).strip()
return result if result else default
def as_list(value: Any) -> list[Any]:
if value is None:
return []
if isinstance(value, list):
return value
return [value]
def number(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default
def fmt_num(value: Any) -> str:
n = number(value)
if abs(n - round(n)) < 1e-9:
return str(int(round(n)))
return f"{n:.1f}"
def fmt_pct(value: Any) -> str:
return f"{number(value) * 100:.0f}%"
def pipe_escape(value: Any) -> str:
return text(value).replace("|", "\\|").replace("\n", " ")
def h(value: Any) -> str:
return escape(text(value))
def index_by_id(items: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
return {str(item.get("id")): item for item in items if item.get("id")}
def label(ref: Any, action_map: dict[str, dict[str, Any]], default: str | None = None) -> str:
ref_text = text(ref, "")
if ref_text in action_map:
return text(action_map[ref_text].get("name"), ref_text)
for item in action_map.values():
if ref_text == text(item.get("name"), ""):
return text(item.get("name"), ref_text)
return default or ref_text or "-"
def same_ref(ref: Any, item: dict[str, Any]) -> bool:
ref_text = text(ref, "")
return ref_text in {str(item.get("id", "")), text(item.get("name"), "")}
def merge_update(base: dict[str, Any], update: dict[str, Any]) -> dict[str, Any]:
merged = deepcopy(base)
append_keys = [
"players",
"our_actions",
"opponent_actions",
"reaction_estimates",
"payoff_matrix",
"commitment_tests",
"signals",
"historical_behavior_data",
"experience_reference_analysis",
"rounds",
"scenario_triggers",
"warnings",
]
for key in append_keys:
if update.get(key):
merged.setdefault(key, [])
merged[key].extend(as_list(update[key]))
for key in ("case_context", "equilibrium", "repeated_game", "recommendation", "rationality_priors"):
if isinstance(update.get(key), dict):
merged.setdefault(key, {})
merged[key].update(update[key])
merged.setdefault("applied_updates", [])
merged["applied_updates"].append(update)
return merged
def commitment_score(item: dict[str, Any]) -> tuple[float, str]:
components = [
number(item.get("costliness"), 0.0),
number(item.get("observability"), 0.0),
1.0 - number(item.get("reversibility"), 1.0),
number(item.get("incentive_fit"), 0.0),
number(item.get("capability_backing"), 0.0),
]
present = [max(0.0, min(1.0, value)) for value in components]
score = sum(present) / len(present) if present else 0.0
return score, commitment_level(score)
def clamp(value: float, low: float = 0.0, high: float = 1.0) -> float:
return max(low, min(high, value))
def commitment_level(score: float) -> str:
if score >= 0.75:
return "可信"
if score >= 0.45:
return "部分可信"
return "弱承诺/更像口头信号"
def rationality_signal_adjustment(signal: Any) -> float:
signal_text = text(signal, "").lower()
mapping = {
"rational": 0.08,
"time_consistent": 0.1,
"disciplined": 0.08,
"bounded": -0.12,
"bounded_rational": -0.12,
"opportunistic": -0.14,
"inconsistent": -0.18,
"cheap_talk": -0.18,
"bluff": -0.2,
"irrational": -0.25,
}
return mapping.get(signal_text, 0.0)
def rationality_priors(request: dict[str, Any]) -> dict[str, float]:
raw = request.get("rationality_priors", {})
priors: dict[str, float] = {}
if isinstance(raw, dict):
for key, value in raw.items():
if isinstance(value, dict):
priors[str(key)] = clamp(number(value.get("probability"), 0.72), 0.05, 0.95)
else:
priors[str(key)] = clamp(number(value, 0.72), 0.05, 0.95)
elif isinstance(raw, list):
for item in raw:
if isinstance(item, dict) and item.get("player_id"):
priors[str(item["player_id"])] = clamp(number(item.get("probability"), 0.72), 0.05, 0.95)
return priors
def build_historical_behavior_analysis(request: dict[str, Any]) -> dict[str, Any]:
players = index_by_id(request.get("players", []))
history_items = [item for item in as_list(request.get("historical_behavior_data")) if isinstance(item, dict)]
reference_items = [item for item in as_list(request.get("experience_reference_analysis")) if isinstance(item, dict)]
priors = rationality_priors(request)
player_ids = set(players.keys()) | {text(item.get("player_id"), "") for item in history_items + reference_items}
player_ids = {player_id for player_id in player_ids if player_id}
if not player_ids:
player_ids = {"head_competitor"}
player_adjustments = []
for player_id in sorted(player_ids):
prior = priors.get(player_id, 0.72 if player_id != "us" else 0.78)
adjustments: list[float] = []
evidence_weights: list[float] = []
player_history = [item for item in history_items if text(item.get("player_id"), "") == player_id]
player_refs = [item for item in reference_items if text(item.get("player_id"), player_id) == player_id]
for item in player_history:
similarity = clamp(number(item.get("context_similarity"), 0.55), 0.0, 1.0)
follow_raw = item.get("commitment_follow_through")
follow_adjustment = 0.0
if follow_raw is not None:
follow_adjustment = (clamp(number(follow_raw), 0.0, 1.0) - 0.5) * 0.35
signal_adjustment = rationality_signal_adjustment(item.get("rationality_signal"))
adjustments.append((signal_adjustment + follow_adjustment) * similarity)
evidence_weights.append(similarity)
for item in player_refs:
confidence = clamp(number(item.get("confidence"), 0.55), 0.0, 1.0)
adjustments.append(number(item.get("rationality_adjustment"), 0.0) * confidence)
evidence_weights.append(confidence)
total_adjustment = sum(adjustments)
adjusted = clamp(prior + total_adjustment, 0.05, 0.95)
evidence_quality = sum(evidence_weights) / len(evidence_weights) if evidence_weights else 0.0
if not evidence_weights:
interpretation = "未提供真实历史行为数据;理性概率仍是模型先验,承诺和反应判断需要打折。"
elif adjusted < prior - 0.08:
interpretation = "真实历史行为显示该玩家未必时间一致或完全理性,应下调其长期威胁、承诺或信号可信度。"
elif adjusted > prior + 0.08:
interpretation = "真实历史行为显示该玩家执行较一致,可适度提高其理性反应和承诺可信度。"
else:
interpretation = "历史行为没有显著改变理性概率,但可作为反应和承诺判断的校准证据。"
player_adjustments.append(
{
"player_id": player_id,
"player": text(players.get(player_id, {}).get("name"), player_id),
"prior_rationality": round(prior, 2),
"adjusted_rationality": round(adjusted, 2),
"adjustment": round(adjusted - prior, 2),
"evidence_quality": round(evidence_quality, 2),
"history_count": len(player_history),
"reference_count": len(player_refs),
"interpretation": interpretation,
}
)
history_rows = [
{
"player_id": text(item.get("player_id")),
"player": text(players.get(text(item.get("player_id"), ""), {}).get("name"), text(item.get("player_id"))),
"event": text(item.get("event")),
"observed_action": text(item.get("observed_action")),
"context_similarity": clamp(number(item.get("context_similarity"), 0.0), 0.0, 1.0),
"commitment_follow_through": clamp(number(item.get("commitment_follow_through"), 0.0), 0.0, 1.0),
"source": text(item.get("source")),
"implication": text(item.get("implication")),
}
for item in history_items
]
reference_rows = [
{
"player_id": text(item.get("player_id"), ""),
"reference_class": text(item.get("reference_class")),
"pattern": text(item.get("pattern")),
"rationality_adjustment": round(number(item.get("rationality_adjustment"), 0.0), 2),
"confidence": clamp(number(item.get("confidence"), 0.0), 0.0, 1.0),
"implication": text(item.get("implication")),
}
for item in reference_items
]
return {
"principle": "真实历史行为优先于纯模型推演;多轮博弈中应防止高估对手完全理性和时间一致性。",
"player_adjustments": player_adjustments,
"history_rows": history_rows,
"reference_rows": reference_rows,
}
def enrich_commitments(request: dict[str, Any], historical_analysis: dict[str, Any]) -> list[dict[str, Any]]:
actions = index_by_id(request.get("our_actions", []))
opponent_actions = index_by_id(request.get("opponent_actions", []))
players = index_by_id(request.get("players", []))
rationality_map = {item.get("player_id"): item for item in historical_analysis.get("player_adjustments", [])}
enriched = []
for item in request.get("commitment_tests", []):
score, level = commitment_score(item)
action_id = text(item.get("action_id"), "")
player_id = text(item.get("player_id"), "")
if not player_id and action_id in opponent_actions:
player_id = text(opponent_actions[action_id].get("player_id"), "")
if not player_id:
player_id = "us"
adjustment = rationality_map.get(player_id, {})
prior_rationality = number(adjustment.get("prior_rationality"), 0.0)
adjusted_rationality = number(adjustment.get("adjusted_rationality"), prior_rationality)
has_history = bool(number(adjustment.get("history_count"), 0.0) or number(adjustment.get("reference_count"), 0.0))
history_adjusted_score = score
if has_history:
history_adjusted_score = clamp(score + (adjusted_rationality - prior_rationality) * 0.35)
adjusted_level = commitment_level(history_adjusted_score)
if action_id in actions:
target_name = text(actions[action_id].get("name"), action_id)
elif action_id in opponent_actions:
target_name = text(opponent_actions[action_id].get("name"), action_id)
else:
target_name = ""
if not target_name or target_name == "-":
target_name = text(players.get(player_id, {}).get("name"), player_id)
enriched.append(
{
**item,
"player_id": player_id,
"player_name": text(players.get(player_id, {}).get("name"), player_id),
"action_name": target_name,
"raw_score": score,
"raw_level": level,
"score": history_adjusted_score,
"level": adjusted_level,
"rationality_prior": prior_rationality,
"history_adjusted_rationality": adjusted_rationality,
"history_adjusted_score": history_adjusted_score,
"history_adjustment_note": text(adjustment.get("interpretation"), "未提供该玩家历史行为校正。"),
}
)
return enriched
def payoff_rows_for_action(request: dict[str, Any], our_action: dict[str, Any]) -> list[dict[str, Any]]:
return [row for row in request.get("payoff_matrix", []) if same_ref(row.get("our_action"), our_action)]
def matching_payoff(row_action: Any, opponent_action: Any) -> bool:
row = text(row_action, "")
candidate = text(opponent_action, "")
return bool(row and candidate and row == candidate)
def expected_payoffs(request: dict[str, Any]) -> list[dict[str, Any]]:
our_actions = request.get("our_actions", [])
opponent_actions = index_by_id(request.get("opponent_actions", []))
results = []
for action in our_actions:
action_rows = payoff_rows_for_action(request, action)
weighted = 0.0
weight = 0.0
detail = []
for estimate in request.get("reaction_estimates", []):
if not same_ref(estimate.get("if_we"), action):
continue
for likely in estimate.get("likely_actions", []):
p = number(likely.get("probability"), 0.0)
candidate = text(likely.get("action"), "")
row = next((r for r in action_rows if matching_payoff(r.get("opponent_action"), candidate)), None)
if row is None and candidate in opponent_actions:
row = next((r for r in action_rows if matching_payoff(r.get("opponent_action"), opponent_actions[candidate].get("name"))), None)
if row is None:
continue
payoff = number(row.get("our_payoff"), 0.0)
weighted += p * payoff
weight += p
detail.append(
{
"opponent_action": label(candidate, opponent_actions, candidate),
"probability": p,
"our_payoff": payoff,
"rationale": text(likely.get("rationale")),
}
)
if weight > 0:
expected = weighted / weight
method = "reaction-weighted"
elif action_rows:
expected = sum(number(row.get("our_payoff"), 0.0) for row in action_rows) / len(action_rows)
method = "matrix-average"
else:
expected = 0.0
method = "missing-payoff"
results.append(
{
"action_id": action.get("id"),
"action_name": text(action.get("name")),
"expected_payoff": expected,
"method": method,
"details": detail,
}
)
return sorted(results, key=lambda item: item["expected_payoff"], reverse=True)
def top_reactions(request: dict[str, Any], action_id: str | None) -> list[dict[str, Any]]:
if not action_id:
return []
opponent_actions = index_by_id(request.get("opponent_actions", []))
for estimate in request.get("reaction_estimates", []):
if text(estimate.get("if_we"), "") != action_id:
continue
likely = sorted(estimate.get("likely_actions", []), key=lambda item: number(item.get("probability"), 0.0), reverse=True)
return [
{
"opponent": text(estimate.get("opponent")),
"action": label(item.get("action"), opponent_actions),
"probability": number(item.get("probability"), 0.0),
"rationale": text(item.get("rationale")),
}
for item in likely
]
return []
def infer_framework_selection(request: dict[str, Any]) -> dict[str, Any]:
raw = json.dumps(request, ensure_ascii=False).lower()
primary = "标准型 payoff 矩阵博弈"
lenses: list[str] = ["纳什均衡检查:确认预测结果是否对各方都是最佳回应"]
excluded = []
logic = "根据玩家、策略、payoff、行动顺序和信息结构选择最轻的可解释框架。"
if any(term in raw for term in ["price", "pricing", "价格", "降价", "折扣", "price war"]):
primary = "Bertrand 价格竞争"
lenses.extend(
[
"囚徒困境:检查双方是否会被拖入互相降价的低收益均衡",
"重复博弈:检查短期降价是否破坏长期价格纪律",
]
)
logic = "价格是核心策略变量,因此先用 Bertrand 竞争解释跟随降价,再用囚徒困境和重复博弈检查价格战风险。"
if any(term in raw for term in ["channel", "渠道", "partner", "联盟", "生态"]):
lenses.extend(["联盟/渠道博弈:检查渠道或伙伴是否会因分成、风险和锁定而改变阵营"])
if any(term in raw for term in ["signal", "信号", "free", "免费", "announcement", "宣布", "roadmap"]):
lenses.extend(["信号博弈:检查对手或我们的公开动作是否成本高、可验证、难伪造"])
if any(term in raw for term in ["commitment", "承诺", "exclusive", "独家", "长期"]):
lenses.extend(["可信承诺:检查动作是否成本高、可观察、难撤回、时间一致"])
if any(term in raw for term in ["bid", "auction", "m&a", "并购", "竞价", "tender"]):
primary = "拍卖/竞价博弈"
lenses.extend(["赢家诅咒:检查中标是否意味着过度乐观估值", "谈判博弈:检查出价后条款和退出选项"])
logic = "稀缺资产和多方出价是核心结构,因此优先使用拍卖/竞价博弈。"
if any(term in raw for term in ["financing", "融资", "terms", "估值", "valuation", "batna"]):
primary = "谈判博弈"
lenses.extend(["外部选择:检查 BATNA 和保留价值", "信号博弈:检查融资信心与稀缺性信号"])
logic = "条款和剩余价值分配是核心问题,因此优先使用谈判博弈。"
if any(term in raw for term in ["entry", "进入", "retaliation", "威慑"]):
lenses.extend(["进入威慑:检查 incumbent 的威胁是否在进入后仍然理性", "子博弈精炼:过滤不可信威胁"])
if "zero" not in raw and "零和" not in raw:
excluded.append(
{
"name": "纯零和博弈",
"reason": "当前输入未证明总收益固定;如果存在渠道、客户体验、合作或差异化空间,应按混合动机博弈处理。",
}
)
return {
"primary": primary,
"secondary_lenses": list(dict.fromkeys(lenses)),
"excluded_frameworks": excluded,
"combination_logic": logic,
}
def readiness_status(score: float) -> tuple[str, str]:
if score >= 0.75:
return "ready", "关键玩家、策略、payoff、反应和承诺信息已经足够支持正式策略建议。"
if score >= 0.45:
return "nearly-ready", "当前模型可用于方向性决策,但仍有一两个信息缺口会影响执行强度。"
return "needs-more-info", "当前模型仍缺少关键玩家、payoff、反应或承诺证据,适合先做低成本验证。"
def build_sensitivity(
request: dict[str, Any],
payoffs: list[dict[str, Any]],
commitments: list[dict[str, Any]],
primary_id: str,
historical_analysis: dict[str, Any],
) -> dict[str, Any]:
if request.get("sensitivity"):
return request["sensitivity"]
best = payoffs[0] if payoffs else {}
runner_up = payoffs[1] if len(payoffs) > 1 else {}
gap = number(best.get("expected_payoff"), 0.0) - number(runner_up.get("expected_payoff"), 0.0)
primary_commitment = next((item for item in commitments if text(item.get("action_id"), "") == primary_id), None)
commitment_score_value = number(primary_commitment.get("score"), 0.0) if primary_commitment else 0.0
if gap >= 15 and commitment_score_value >= 0.75:
stability = "stable"
elif gap >= 5 or commitment_score_value >= 0.6:
stability = "mixed"
else:
stability = "fragile"
rationality_risk = next(
(
item
for item in historical_analysis.get("player_adjustments", [])
if item.get("player_id") != "us" and number(item.get("adjustment"), 0.0) <= -0.08
),
None,
)
if rationality_risk:
dangerous = f"可能高估「{text(rationality_risk.get('player'))}」的完全理性和时间一致性;真实历史行为要求下调其长期承诺或威胁可信度。"
elif gap < 5:
dangerous = "主推荐与备选方案的预期收益差距较小,少量 payoff 调整就可能改变排序。"
elif commitment_score_value < 0.75:
dangerous = "主推荐依赖承诺可信度;如果渠道或对手认为该承诺可轻易撤回,推荐会变弱。"
else:
dangerous = "最大风险是假设竞品不会把价格战升级到长期补贴或更强渠道攻击。"
return {
"stability": stability,
"payoff_gap": gap,
"base_recommendation": text(best.get("action_name"), "-"),
"runner_up": text(runner_up.get("action_name"), "无"),
"most_dangerous_assumption": dangerous,
"stress_tests": [
{
"name": "竞品更激进",
"assumption_shift": "将头部竞品跟随降价或攻击渠道的倾向上调。",
"expected_effect": "降价选项进一步变差,渠道绑定需要更强返点和交付资源支撑。",
"recommendation_impact": "通常不改变主推荐,但会提高执行成本。",
},
{
"name": "渠道收益折扣",
"assumption_shift": "将渠道绑定带来的渠道收益和我方收益下调。",
"expected_effect": "捆绑渠道与高端差异化的差距缩小。",
"recommendation_impact": "若渠道拒绝年度合作包,应提高高端差异化权重。",
},
{
"name": "承诺可信度下降",
"assumption_shift": "渠道认为年度合作包只是短期销售动作。",
"expected_effect": "渠道绑定的可信承诺分下降,竞品攻击渠道的胜率上升。",
"recommendation_impact": "需要先补强可观察投入,再正式推进绑定。",
},
{
"name": "对手理性程度下调",
"assumption_shift": "用真实历史行为替代模型默认理性假设。",
"expected_effect": "长期低价、免费版持续投入或公开威胁可能更像战术噪音,而非时间一致承诺。",
"recommendation_impact": "降低对方持续威胁可信度,但仍保留短期扰动和渠道舆论风险。",
},
],
}
def build_strategy_readiness(
request: dict[str, Any],
payoffs: list[dict[str, Any]],
commitments: list[dict[str, Any]],
framework_selection: dict[str, Any],
sensitivity: dict[str, Any],
historical_analysis: dict[str, Any],
) -> dict[str, Any]:
if request.get("strategy_readiness"):
return request["strategy_readiness"]
historical_scores = [
number(item.get("evidence_quality"), 0.0)
for item in historical_analysis.get("player_adjustments", [])
if item.get("player_id") != "us" and (item.get("history_count") or item.get("reference_count"))
]
checks = {
"players": 1.0 if len(request.get("players", [])) >= 2 else 0.3,
"strategies": 1.0 if request.get("our_actions") and request.get("opponent_actions") else 0.35,
"payoffs": 1.0 if request.get("payoff_matrix") and all(row.get("basis") for row in request.get("payoff_matrix", [])) else 0.45,
"reactions": 1.0 if any(item.get("likely_actions") for item in request.get("reaction_estimates", [])) else 0.35,
"commitments": min(1.0, max([number(item.get("score"), 0.0) for item in commitments] or [0.0]) + 0.15),
"history": max(historical_scores) if historical_scores else 0.35,
"framework": 1.0 if framework_selection.get("primary") and framework_selection.get("secondary_lenses") else 0.4,
"sensitivity": {"stable": 1.0, "mixed": 0.7, "fragile": 0.35}.get(text(sensitivity.get("stability"), ""), 0.5),
"updateability": 1.0 if request.get("scenario_triggers") and request.get("rounds") else 0.55,
}
weights = {
"players": 0.1,
"strategies": 0.1,
"payoffs": 0.14,
"reactions": 0.14,
"commitments": 0.12,
"history": 0.14,
"framework": 0.08,
"sensitivity": 0.1,
"updateability": 0.08,
}
score = sum(checks[key] * weights[key] for key in weights)
if sensitivity.get("stability") == "mixed":
score = min(score, 0.74)
elif sensitivity.get("stability") == "fragile":
score = min(score, 0.44)
gaps = []
if checks["payoffs"] < 0.8:
gaps.append("补充真实毛利、渠道转化、客户留存或竞品成本数据,替换 assumed payoff。")
if checks["reactions"] < 0.8:
gaps.append("补齐竞品跟随降价、渠道攻击、免费版本推进的证据和概率。")
if checks["commitments"] < 0.75:
gaps.append("提高主推荐动作的可观察投入、合同约束或撤回成本。")
if checks["history"] < 0.7:
gaps.append("补充对手真实历史行为数据、过往威胁兑现率和同类经验参考,校正理性概率。")
if sensitivity.get("stability") != "stable":
gaps.append("围绕最危险假设做一次小规模验证或敏感性复核。")
status, interpretation = readiness_status(score)
return {
"score": round(score, 2),
"status": status,
"interpretation": interpretation,
"remaining_gaps": gaps[:4],
"dimension_scores": {key: round(value, 2) for key, value in checks.items()},
}
def build_strategic_hygiene(
request: dict[str, Any],
commitments: list[dict[str, Any]],
framework_selection: dict[str, Any],
historical_analysis: dict[str, Any],
) -> dict[str, Any]:
if request.get("strategic_hygiene"):
return request["strategic_hygiene"]
payoff_basis_ok = bool(request.get("payoff_matrix")) and all(row.get("basis") for row in request.get("payoff_matrix", []))
reaction_rationale_ok = all(likely.get("rationale") for item in request.get("reaction_estimates", []) for likely in item.get("likely_actions", []))
commitment_ok = bool(commitments) and max(number(item.get("score"), 0.0) for item in commitments) >= 0.75
history_ok = bool(historical_analysis.get("history_rows") or historical_analysis.get("reference_rows"))
checks = [
{
"name": "payoff hygiene",
"status": "pass" if payoff_basis_ok else "watch",
"note": "payoff 已标记 observed / estimated / assumed。" if payoff_basis_ok else "仍需标记 payoff 来源,避免把估计当事实。",
},
{
"name": "reaction hygiene",
"status": "pass" if reaction_rationale_ok else "watch",
"note": "对手反应均有激励理由。" if reaction_rationale_ok else "部分对手反应缺少激励解释。",
},
{
"name": "commitment hygiene",
"status": "pass" if commitment_ok else "watch",
"note": "至少一个关键承诺通过可信度门槛。" if commitment_ok else "承诺仍可能被视为便宜话术。",
},
{
"name": "historical behavior hygiene",
"status": "pass" if history_ok else "watch",
"note": "已用真实历史行为或同类经验校正玩家理性概率。" if history_ok else "缺少历史行为校正,模型可能高估对手完全理性和时间一致性。",
},
{
"name": "framework hygiene",
"status": "pass" if framework_selection.get("primary") else "watch",
"note": "框架选择先从结构出发,而不是只套用著名博弈。" if framework_selection.get("primary") else "需要说明主框架为何适配。",
},
{
"name": "safety hygiene",
"status": "pass" if request.get("warnings") else "watch",
"note": "已标记法律、投资、反垄断或合规边界。" if request.get("warnings") else "需要补充合规和高风险边界。",
},
]
return {"checks": checks}
def build_next_information(
request: dict[str, Any],
readiness: dict[str, Any],
sensitivity: dict[str, Any],
historical_analysis: dict[str, Any],
) -> dict[str, Any]:
if request.get("next_information"):
return request["next_information"]
priorities = [
{
"item": "对手真实历史行为样本",
"why_it_matters": "AI 估算 payoff 和可信承诺时容易高估对手理性;历史行为能校正威胁兑现率和时间一致性。",
"collection_method": "整理过往价格战、免费版发布、渠道攻击、公开威胁与实际兑现记录,并标注相似度和来源。",
},
{
"item": "竞品真实降价承受能力",
"why_it_matters": "决定竞品跟随降价是否是可持续威胁,而不是短期信号。",
"collection_method": "观察竞品报价、销售话术、续费折扣和公开成本信号。",
},
{
"item": "核心渠道对年度合作包的接受阈值",
"why_it_matters": "决定渠道绑定是否能成为可信承诺。",
"collection_method": "用 2-3 个核心渠道做返点、线索、交付支持的访谈和报价测试。",
},
{
"item": "免费版本是否有真实资源投入",
"why_it_matters": "决定竞品免费版是便宜信号还是可持续进攻。",
"collection_method": "追踪产品可用性、渠道推广、转化路径和续费机制。",
},
]
if historical_analysis.get("history_rows") or historical_analysis.get("reference_rows"):
priorities = priorities[1:]
for gap in readiness.get("remaining_gaps", []):
priorities.append({"item": "策略准备度缺口", "why_it_matters": gap, "collection_method": "补齐后重新生成博弈报告。"})
if sensitivity.get("stability") == "fragile":
priorities.insert(
0,
{
"item": "最危险假设验证",
"why_it_matters": text(sensitivity.get("most_dangerous_assumption")),
"collection_method": "先做小规模渠道或客户测试,再决定是否推进不可逆动作。",
},
)
return {"priorities": priorities[:5]}
def build_report(request: dict[str, Any]) -> dict[str, Any]:
actions = index_by_id(request.get("our_actions", []))
payoffs = expected_payoffs(request)
historical_analysis = build_historical_behavior_analysis(request)
commitments = enrich_commitments(request, historical_analysis)
framework_selection = request.get("framework_selection") or infer_framework_selection(request)
recommendation = request.get("recommendation", {})
primary_id = text(recommendation.get("primary_action"), "") or (payoffs[0]["action_id"] if payoffs else "")
secondary_id = text(recommendation.get("secondary_action"), "")
avoid_id = text(recommendation.get("avoid_action"), "") or (payoffs[-1]["action_id"] if payoffs else "")
primary_name = label(primary_id, actions, primary_id)
secondary_name = label(secondary_id, actions, secondary_id) if secondary_id else ""
avoid_name = label(avoid_id, actions, avoid_id)
best_commitment = next((item for item in commitments if text(item.get("action_id"), "") == primary_id), None)
commitment_verdict = (
f"{text(best_commitment.get('commitment'))}:{best_commitment['level']}({fmt_pct(best_commitment['score'])})"
if best_commitment
else "未提供主推荐动作的承诺评分"
)
reaction_summary = top_reactions(request, primary_id)
top_reaction = reaction_summary[0] if reaction_summary else None
one_sentence = (
f"建议以「{primary_name}」为主"
+ (f"、以「{secondary_name}」为辅" if secondary_name else "")
+ f",避免「{avoid_name}」;"
+ (f"主反应风险是「{top_reaction['action']}」(约 {fmt_pct(top_reaction['probability'])})。" if top_reaction else "当前反应数据不足,需继续补齐对手动作。")
)
sensitivity = build_sensitivity(request, payoffs, commitments, primary_id, historical_analysis)
readiness = build_strategy_readiness(request, payoffs, commitments, framework_selection, sensitivity, historical_analysis)
strategic_hygiene = build_strategic_hygiene(request, commitments, framework_selection, historical_analysis)
next_information = build_next_information(request, readiness, sensitivity, historical_analysis)
return {
"title": text(request.get("title"), "博弈论策略报告"),
"generated_at": str(date.today()),
"case_context": request.get("case_context", {}),
"framework_selection": framework_selection,
"summary": {
"one_sentence": one_sentence,
"recommended_action": primary_name,
"secondary_action": secondary_name,
"avoid_action": avoid_name,
"reason": text(recommendation.get("reason"), "基于当前 payoff、对手反应和承诺可信度的综合判断。"),
"commitment_verdict": commitment_verdict,
"strategy_readiness": readiness["score"],
"stability": sensitivity["stability"],
},
"players": request.get("players", []),
"our_actions": request.get("our_actions", []),
"opponent_actions": request.get("opponent_actions", []),
"reaction_estimates": request.get("reaction_estimates", []),
"top_reactions": reaction_summary,
"payoff_matrix": request.get("payoff_matrix", []),
"expected_payoffs": payoffs,
"commitment_tests": commitments,
"historical_behavior_analysis": historical_analysis,
"strategy_readiness": readiness,
"strategic_hygiene": strategic_hygiene,
"sensitivity": sensitivity,
"next_information": next_information,
"signals": request.get("signals", []),
"equilibrium": request.get("equilibrium", {}),
"repeated_game": request.get("repeated_game", {}),
"rounds": request.get("rounds", []),
"scenario_triggers": request.get("scenario_triggers", []),
"warnings": request.get("warnings", []),
}
def md_table(headers: list[str], rows: list[list[Any]]) -> str:
if not rows:
return "_暂无数据。_"
lines = ["| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |"]
for row in rows:
lines.append("| " + " | ".join(pipe_escape(cell) for cell in row) + " |")
return "\n".join(lines)
def render_markdown(report: dict[str, Any]) -> str:
context = report.get("case_context", {})
frameworks = report.get("framework_selection", {})
history = report.get("historical_behavior_analysis", {})
md: list[str] = [
f"# {report['title']}",
"",
f"> 生成日期:{report['generated_at']}",
"",
"## 一句话结论",
"",
report["summary"]["one_sentence"],
"",
"## 推荐动作",
"",
f"- 主动作:{report['summary']['recommended_action']}",
f"- 辅助动作:{report['summary']['secondary_action'] or '无'}",
f"- 避免动作:{report['summary']['avoid_action']}",
f"- 推荐理由:{report['summary']['reason']}",
f"- 承诺判断:{report['summary']['commitment_verdict']}",
"",
"## 问题定义",
"",
f"- 决策问题:{text(context.get('decision_question'))}",
f"- 时间范围:{text(context.get('time_horizon'))}",
f"- 成功标准:{text(context.get('success_metric'))}",
"",
"## 框架选择与组合",
"",
f"- 主框架:{text(frameworks.get('primary'))}",
f"- 组合逻辑:{text(frameworks.get('combination_logic'))}",
"",
"### 辅助镜头",
"",
]
for lens in frameworks.get("secondary_lenses", []):
md.append(f"- {text(lens)}")
md.extend(
[
"",
"### 排除或降级的框架",
"",
md_table(
["框架", "原因"],
[[item.get("name"), item.get("reason")] for item in frameworks.get("excluded_frameworks", [])],
),
"",
"## 策略准备度",
"",
f"- 准备度:{fmt_pct(report.get('strategy_readiness', {}).get('score'))}",
f"- 状态:{text(report.get('strategy_readiness', {}).get('status'))}",
f"- 判断:{text(report.get('strategy_readiness', {}).get('interpretation'))}",
"",
"### 剩余缺口",
"",
]
)
for gap in report.get("strategy_readiness", {}).get("remaining_gaps", []):
md.append(f"- {text(gap)}")
md.extend(
[
"",
"## 战略卫生检查",
"",
md_table(
["检查", "状态", "说明"],
[[item.get("name"), item.get("status"), item.get("note")] for item in report.get("strategic_hygiene", {}).get("checks", [])],
),
"",
"## 历史行为与理性概率校正",
"",
text(history.get("principle")),
"",
md_table(
["玩家", "先验理性", "历史校正后", "调整", "证据质量", "解释"],
[
[
item.get("player"),
fmt_pct(item.get("prior_rationality")),
fmt_pct(item.get("adjusted_rationality")),
fmt_pct(item.get("adjustment")),
fmt_pct(item.get("evidence_quality")),
item.get("interpretation"),
]
for item in history.get("player_adjustments", [])
],
),
"",
"### 真实历史行为样本",
"",
md_table(
["玩家", "历史事件", "观察动作", "相似度", "兑现率", "来源", "启发"],
[
[
item.get("player"),
item.get("event"),
item.get("observed_action"),
fmt_pct(item.get("context_similarity")),
fmt_pct(item.get("commitment_follow_through")),
item.get("source"),
item.get("implication"),
]
for item in history.get("history_rows", [])
],
),
"",
"### 经验参考分析",
"",
md_table(
["参考类", "模式", "理性调整", "置信度", "启发"],
[
[
item.get("reference_class"),
item.get("pattern"),
fmt_pct(item.get("rationality_adjustment")),
fmt_pct(item.get("confidence")),
item.get("implication"),
]
for item in history.get("reference_rows", [])
],
),
"",
"## 敏感性与稳定性",
"",
f"- 稳定性:{text(report.get('sensitivity', {}).get('stability'))}",
f"- 收益差距:{fmt_num(report.get('sensitivity', {}).get('payoff_gap'))}",
f"- 最危险假设:{text(report.get('sensitivity', {}).get('most_dangerous_assumption'))}",
"",
md_table(
["压力测试", "假设变化", "影响", "对推荐的影响"],
[
[item.get("name"), item.get("assumption_shift"), item.get("expected_effect"), item.get("recommendation_impact")]
for item in report.get("sensitivity", {}).get("stress_tests", [])
],
),
"",
"## 下一步信息",
"",
md_table(
["信息", "为什么重要", "收集方法"],
[[item.get("item"), item.get("why_it_matters"), item.get("collection_method")] for item in report.get("next_information", {}).get("priorities", [])],
),
"",
"## 玩家地图",
"",
md_table(
["玩家", "角色", "目标", "约束"],
[[p.get("name"), p.get("role"), text(p.get("objectives")), text(p.get("constraints"))] for p in report.get("players", [])],
),
"",
"## 策略集合",
"",
md_table(
["我们的动作", "说明", "数据基础"],
[[a.get("name"), a.get("description"), a.get("data_basis")] for a in report.get("our_actions", [])],
),
"",
"## 对手反应地图",
"",
]
)
reaction_rows = []
action_map = index_by_id(report.get("our_actions", []))
opponent_action_map = index_by_id(report.get("opponent_actions", []))
for estimate in report.get("reaction_estimates", []):
for likely in estimate.get("likely_actions", []):
reaction_rows.append(
[
text(estimate.get("opponent")),
label(estimate.get("if_we"), action_map),
label(likely.get("action"), opponent_action_map),
fmt_pct(likely.get("probability")),
likely.get("rationale"),
]
)
md.append(md_table(["对手", "如果我们", "可能动作", "概率", "理由"], reaction_rows))
md.extend(
[
"",
"## Payoff 矩阵",
"",
md_table(
["我们的动作", "对方动作", "我方收益", "对方收益", "渠道收益", "基础", "说明"],
[
[
label(row.get("our_action"), action_map),
label(row.get("opponent_action"), opponent_action_map, text(row.get("opponent_action"))),
fmt_num(row.get("our_payoff")),
fmt_num(row.get("opponent_payoff")),
fmt_num(row.get("channel_payoff")),
row.get("basis"),
row.get("notes"),
]
for row in report.get("payoff_matrix", [])
],
),
"",
"## 预期收益排序",
"",
md_table(
["动作", "预期收益", "计算方式"],
[[item["action_name"], fmt_num(item["expected_payoff"]), item["method"]] for item in report.get("expected_payoffs", [])],
),
"",
"## 承诺与信号可信度",
"",
md_table(
["动作", "承诺", "可信度", "原始评分", "历史校正评分", "证据基础", "历史校正说明"],
[
[
item.get("action_name"),
item.get("commitment"),
item.get("level"),
fmt_pct(item.get("raw_score")),
fmt_pct(item.get("history_adjusted_score")),
item.get("evidence_basis"),
item.get("history_adjustment_note"),
]
for item in report.get("commitment_tests", [])
],
),
"",
]
)
if report.get("signals"):
md.extend(
[
"### 外部信号",
"",
md_table(
["发送方", "信号", "质量", "解释"],
[[s.get("sender"), s.get("signal"), s.get("quality"), s.get("interpretation")] for s in report.get("signals", [])],
),
"",
]
)
equilibrium = report.get("equilibrium", {})
md.extend(
[
"## 均衡解释",
"",
f"- 主模型:{text(equilibrium.get('primary_frame'))}",
f"- 解释:{text(equilibrium.get('interpretation'))}",
"",
]
)
for item in equilibrium.get("candidate_equilibria", []):
md.append(f"- {text(item)}")
repeated = report.get("repeated_game", {})
md.extend(
[
"",
"## 重复博弈与关系动态",
"",
text(repeated.get("interpretation")),
"",
]
)
for risk in repeated.get("relationship_risks", []):
md.append(f"- {text(risk)}")
if report.get("rounds"):
md.extend(
[
"",
"## 动态更新日志",
"",
md_table(
["轮次", "日期", "阶段", "新增信息", "判断", "状态"],
[
[r.get("round"), r.get("date"), r.get("stage"), text(r.get("new_information")), r.get("interim_judgment"), r.get("recommendation_status")]
for r in report.get("rounds", [])
],
),
"",
]
)
md.extend(
[
"## 需要重新打开报告的触发器",
"",
md_table(
["触发器", "更新规则"],
[[item.get("trigger"), item.get("update_rule")] for item in report.get("scenario_triggers", [])],
),
"",
"## 风险与边界",
"",
]
)
for warning in report.get("warnings", []):
md.append(f"- {text(warning)}")
md.extend(
[
"",
"## 自动化说明",
"",
"本报告由 `yao-gametheory-skill` 从同一份结构化 JSON 自动生成,Markdown、HTML、DOCX、PDF 与 canonical JSON 应保持同步。",
"",
]
)
return "\n".join(md)
CSS = """
:root {
--ink: #17211b;
--muted: #5c675f;
--line: #d9e0d8;
--paper: #fbfcf8;
--surface: #ffffff;
--accent: #196f5a;
--accent-soft: #e6f3ef;
--risk: #9f3a2f;
--warn: #8a6718;
}
* { box-sizing: border-box; }
body {
margin: 0;
color: var(--ink);
background: var(--paper);
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei", sans-serif;
line-height: 1.65;
}
.topbar {
position: sticky;
top: 0;
z-index: 10;
display: flex;
justify-content: space-between;
gap: 16px;
align-items: center;
padding: 10px 24px;
border-bottom: 1px solid var(--line);
background: rgba(251, 252, 248, 0.95);
}
.nav { display: flex; flex-wrap: wrap; gap: 10px; font-size: 13px; }
.nav a { color: var(--muted); text-decoration: none; }
.actions button {
border: 1px solid var(--line);
background: var(--surface);
color: var(--ink);
border-radius: 6px;
padding: 7px 10px;
cursor: pointer;
}
main { max-width: 1120px; margin: 0 auto; padding: 34px 24px 72px; }
.hero {
padding: 24px 0 18px;
border-bottom: 2px solid var(--ink);
}
.eyebrow { color: var(--accent); font-weight: 700; font-size: 13px; letter-spacing: 0; }
h1 { margin: 8px 0 10px; font-size: 34px; line-height: 1.18; letter-spacing: 0; }
h2 { margin: 34px 0 12px; font-size: 22px; letter-spacing: 0; }
h3 { margin: 22px 0 10px; font-size: 17px; letter-spacing: 0; }
.summary {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(180px, 1fr));
gap: 12px;
margin: 18px 0 6px;
}
.metric {
border: 1px solid var(--line);
border-radius: 8px;
background: var(--surface);
padding: 14px;
}
.metric .label { color: var(--muted); font-size: 12px; }
.metric .value { font-size: 18px; font-weight: 750; margin-top: 4px; }
.callout {
margin: 18px 0;
border-left: 4px solid var(--accent);
background: var(--accent-soft);
padding: 14px 16px;
}
.warn { border-left-color: var(--warn); background: #fff8e5; }
.risk { border-left-color: var(--risk); background: #fff0ed; }
.table-wrap { overflow-x: auto; border: 1px solid var(--line); border-radius: 8px; background: var(--surface); }
table { width: 100%; border-collapse: collapse; min-width: 760px; }
th, td {
padding: 10px 12px;
border-bottom: 1px solid var(--line);
text-align: left;
vertical-align: top;
overflow-wrap: anywhere;
word-break: break-word;
}
th { background: #edf3ec; font-size: 13px; }
tr:last-child td { border-bottom: 0; }
.pill {
display: inline-block;
border-radius: 999px;
padding: 2px 8px;
background: #edf3ec;
color: var(--accent);
font-size: 12px;
font-weight: 700;
}
details {
border: 1px solid var(--line);
border-radius: 8px;
background: var(--surface);
padding: 12px 14px;
margin-top: 12px;
}
summary { cursor: pointer; font-weight: 750; }
footer { color: var(--muted); font-size: 13px; margin-top: 42px; border-top: 1px solid var(--line); padding-top: 18px; }
@media (max-width: 760px) {
.topbar { position: static; align-items: flex-start; flex-direction: column; }
main { padding: 24px 16px 56px; }
h1 { font-size: 27px; }
.summary { grid-template-columns: 1fr; }
}
@media print {
@page { size: A4 landscape; margin: 12mm; }
.topbar, .actions { display: none !important; }
body { background: #fff; font-size: 10pt; line-height: 1.45; }
main { max-width: none; padding: 0; width: 100%; }
h1 { font-size: 22pt; }
h2 { font-size: 15pt; margin-top: 18px; }
h3 { font-size: 12pt; margin-top: 14px; }
.summary { grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 8px; }
.metric { padding: 8px; }
.metric .value { font-size: 11pt; overflow-wrap: anywhere; }
.callout { padding: 8px 10px; margin: 10px 0; }
.table-wrap { overflow: visible; border-radius: 4px; break-inside: avoid; page-break-inside: avoid; }
table { min-width: 0; width: 100%; table-layout: fixed; font-size: 8.5pt; }
th, td { padding: 5px 6px; white-space: normal; overflow-wrap: anywhere; word-break: break-word; }
th { font-size: 8.5pt; }
tr { break-inside: avoid; page-break-inside: avoid; }
details { break-inside: avoid; }
}
"""
def html_table(headers: list[str], rows: list[list[Any]]) -> str:
if not rows:
return "
暂无数据。
"
header_html = "".join(f"{escape(head)} " for head in headers)
body_rows = []
for row in rows:
cells = "".join(f"{h(cell)} " for cell in row)
body_rows.append(f"{cells} ")
return f'{header_html} {"".join(body_rows)}
'
def render_html(report: dict[str, Any]) -> str:
context = report.get("case_context", {})
frameworks = report.get("framework_selection", {})
readiness = report.get("strategy_readiness", {})
hygiene = report.get("strategic_hygiene", {})
history = report.get("historical_behavior_analysis", {})
sensitivity = report.get("sensitivity", {})
next_information = report.get("next_information", {})
action_map = index_by_id(report.get("our_actions", []))
opponent_action_map = index_by_id(report.get("opponent_actions", []))
reaction_rows = []
for estimate in report.get("reaction_estimates", []):
for likely in estimate.get("likely_actions", []):
reaction_rows.append(
[
estimate.get("opponent"),
label(estimate.get("if_we"), action_map),
label(likely.get("action"), opponent_action_map),
fmt_pct(likely.get("probability")),
likely.get("rationale"),
]
)
payoff_rows = [
[
label(row.get("our_action"), action_map),
label(row.get("opponent_action"), opponent_action_map, text(row.get("opponent_action"))),
fmt_num(row.get("our_payoff")),
fmt_num(row.get("opponent_payoff")),
fmt_num(row.get("channel_payoff")),
row.get("basis"),
row.get("notes"),
]
for row in report.get("payoff_matrix", [])
]
warning_items = "".join(f"{h(item)} " for item in report.get("warnings", []))
equilibrium = report.get("equilibrium", {})
candidate_items = "".join(f"{h(item)} " for item in equilibrium.get("candidate_equilibria", []))
repeated = report.get("repeated_game", {})
repeated_items = "".join(f"{h(item)} " for item in repeated.get("relationship_risks", []))
signal_rows = [[s.get("sender"), s.get("signal"), s.get("quality"), s.get("interpretation")] for s in report.get("signals", [])]
html = f"""
{h(report['title'])}
Yao Game Theory Skill
{h(report['title'])}
{h(report['summary']['one_sentence'])}
主动作
{h(report['summary']['recommended_action'])}
辅助动作
{h(report['summary']['secondary_action'] or '无')}
避免动作
{h(report['summary']['avoid_action'])}
承诺判断
{h(report['summary']['commitment_verdict'])}
策略准备度
{fmt_pct(readiness.get('score'))}
稳定性
{h(sensitivity.get('stability'))}
推荐动作
{h(report['summary']['reason'])}
决策问题 {h(context.get('decision_question'))}
时间范围 {h(context.get('time_horizon'))}
成功标准 {h(context.get('success_metric'))}
框架选择与组合
主框架 {h(frameworks.get('primary'))}
{h(frameworks.get('combination_logic'))}
辅助镜头
{"".join(f"{h(item)} " for item in frameworks.get("secondary_lenses", []))}
排除或降级的框架
{html_table(["框架", "原因"], [[item.get("name"), item.get("reason")] for item in frameworks.get("excluded_frameworks", [])])}
策略准备度
准备度 {fmt_pct(readiness.get('score'))} 状态 {h(readiness.get('status'))}
{h(readiness.get('interpretation'))}
剩余缺口
{"".join(f"{h(item)} " for item in readiness.get("remaining_gaps", []))}
战略卫生检查
{html_table(["检查", "状态", "说明"], [[item.get("name"), item.get("status"), item.get("note")] for item in hygiene.get("checks", [])])}
历史行为与理性概率校正
{h(history.get('principle'))}
{html_table(["玩家", "先验理性", "历史校正后", "调整", "证据质量", "解释"], [[item.get("player"), fmt_pct(item.get("prior_rationality")), fmt_pct(item.get("adjusted_rationality")), fmt_pct(item.get("adjustment")), fmt_pct(item.get("evidence_quality")), item.get("interpretation")] for item in history.get("player_adjustments", [])])}
真实历史行为样本
{html_table(["玩家", "历史事件", "观察动作", "相似度", "兑现率", "来源", "启发"], [[item.get("player"), item.get("event"), item.get("observed_action"), fmt_pct(item.get("context_similarity")), fmt_pct(item.get("commitment_follow_through")), item.get("source"), item.get("implication")] for item in history.get("history_rows", [])])}
经验参考分析
{html_table(["参考类", "模式", "理性调整", "置信度", "启发"], [[item.get("reference_class"), item.get("pattern"), fmt_pct(item.get("rationality_adjustment")), fmt_pct(item.get("confidence")), item.get("implication")] for item in history.get("reference_rows", [])])}
敏感性与稳定性
稳定性 {h(sensitivity.get('stability'))} 收益差距 {fmt_num(sensitivity.get('payoff_gap'))}
{h(sensitivity.get('most_dangerous_assumption'))}
{html_table(["压力测试", "假设变化", "影响", "对推荐的影响"], [[item.get("name"), item.get("assumption_shift"), item.get("expected_effect"), item.get("recommendation_impact")] for item in sensitivity.get("stress_tests", [])])}
下一步信息
{html_table(["信息", "为什么重要", "收集方法"], [[item.get("item"), item.get("why_it_matters"), item.get("collection_method")] for item in next_information.get("priorities", [])])}
玩家地图
{html_table(["玩家", "角色", "目标", "约束"], [[p.get("name"), p.get("role"), text(p.get("objectives")), text(p.get("constraints"))] for p in report.get("players", [])])}
我们的策略集合
{html_table(["动作", "说明", "数据基础"], [[a.get("name"), a.get("description"), a.get("data_basis")] for a in report.get("our_actions", [])])}
对手反应地图
{html_table(["对手", "如果我们", "可能动作", "概率", "理由"], reaction_rows)}
Payoff 矩阵与排序
{html_table(["我们的动作", "对方动作", "我方收益", "对方收益", "渠道收益", "基础", "说明"], payoff_rows)}
预期收益排序
{html_table(["动作", "预期收益", "计算方式"], [[item["action_name"], fmt_num(item["expected_payoff"]), item["method"]] for item in report.get("expected_payoffs", [])])}
承诺与信号可信度
{html_table(["动作", "承诺", "可信度", "原始评分", "历史校正评分", "证据基础", "历史校正说明"], [[item.get("action_name"), item.get("commitment"), item.get("level"), fmt_pct(item.get("raw_score")), fmt_pct(item.get("history_adjusted_score")), item.get("evidence_basis"), item.get("history_adjustment_note")] for item in report.get("commitment_tests", [])])}
外部信号
{html_table(["发送方", "信号", "质量", "解释"], signal_rows)}
均衡解释
主模型 {h(equilibrium.get('primary_frame'))}
{h(equilibrium.get('interpretation'))}
重复博弈与关系动态
{h(repeated.get('interpretation'))}
动态更新日志
{html_table(["轮次", "日期", "阶段", "新增信息", "判断", "状态"], [[r.get("round"), r.get("date"), r.get("stage"), text(r.get("new_information")), r.get("interim_judgment"), r.get("recommendation_status")] for r in report.get("rounds", [])])}
重新打开报告的触发器
{html_table(["触发器", "更新规则"], [[item.get("trigger"), item.get("update_rule")] for item in report.get("scenario_triggers", [])])}
风险与边界
附录:自动化说明
本报告由同一份结构化 JSON 自动生成,Markdown、HTML、DOCX、PDF 与 canonical JSON 应保持同步。后续对手动作可通过 update JSON 合并后重新导出。
Generated by yao-gametheory-skill on {h(report['generated_at'])}.
"""
return html
def docx_lines(report: dict[str, Any]) -> list[tuple[str, str]]:
frameworks = report.get("framework_selection", {})
lines: list[tuple[str, str]] = [
("title", report["title"]),
("normal", f"生成日期:{report['generated_at']}"),
("heading", "一句话结论"),
("normal", report["summary"]["one_sentence"]),
("heading", "推荐动作"),
("bullet", f"主动作:{report['summary']['recommended_action']}"),
("bullet", f"辅助动作:{report['summary']['secondary_action'] or '无'}"),
("bullet", f"避免动作:{report['summary']['avoid_action']}"),
("bullet", f"推荐理由:{report['summary']['reason']}"),
("bullet", f"承诺判断:{report['summary']['commitment_verdict']}"),
("heading", "框架选择与组合"),
("normal", f"主框架:{text(frameworks.get('primary'))}"),
("normal", f"组合逻辑:{text(frameworks.get('combination_logic'))}"),
("heading", "辅助镜头"),
]
for lens in frameworks.get("secondary_lenses", []):
lines.append(("bullet", text(lens)))
lines.extend(
[
("heading", "排除或降级的框架"),
]
)
for item in frameworks.get("excluded_frameworks", []):
lines.append(("bullet", f"{text(item.get('name'))}:{text(item.get('reason'))}"))
lines.extend(
[
("heading", "玩家地图"),
]
)
for p in report.get("players", []):
lines.append(("normal", f"{text(p.get('name'))}|{text(p.get('role'))}|目标:{text(p.get('objectives'))}|约束:{text(p.get('constraints'))}"))
lines.append(("heading", "对手反应地图"))
action_map = index_by_id(report.get("our_actions", []))
opponent_action_map = index_by_id(report.get("opponent_actions", []))
for estimate in report.get("reaction_estimates", []):
for likely in estimate.get("likely_actions", []):
lines.append(
(
"normal",
f"{text(estimate.get('opponent'))}|如果我们:{label(estimate.get('if_we'), action_map)}|可能动作:{label(likely.get('action'), opponent_action_map)}|概率:{fmt_pct(likely.get('probability'))}|{text(likely.get('rationale'))}",
)
)
lines.append(("heading", "Payoff 矩阵"))
for row in report.get("payoff_matrix", []):
lines.append(
(
"normal",
f"{label(row.get('our_action'), action_map)} x {label(row.get('opponent_action'), opponent_action_map, text(row.get('opponent_action')))}|我方 {fmt_num(row.get('our_payoff'))}|对方 {fmt_num(row.get('opponent_payoff'))}|渠道 {fmt_num(row.get('channel_payoff'))}|{text(row.get('notes'))}",
)
)
lines.append(("heading", "承诺可信度"))
for item in report.get("commitment_tests", []):
lines.append(("normal", f"{text(item.get('action_name'))}|{text(item.get('commitment'))}|{text(item.get('level'))}|{fmt_pct(item.get('score'))}|{text(item.get('notes'))}"))
lines.append(("heading", "动态更新日志"))
for r in report.get("rounds", []):
lines.append(("normal", f"第 {text(r.get('round'))} 轮|{text(r.get('date'))}|{text(r.get('stage'))}|{text(r.get('interim_judgment'))}|{text(r.get('recommendation_status'))}"))
lines.append(("heading", "风险与边界"))
for warning in report.get("warnings", []):
lines.append(("bullet", text(warning)))
return lines
def simple_docx_paragraph(style: str, value: str) -> str:
safe = escape(value)
if style == "title":
props = " "
run_props = " "
elif style == "heading":
props = " "
run_props = " "
elif style == "bullet":
props = " "
run_props = " "
safe = "• " + safe
else:
props = " "
run_props = " "
return f"{props}{run_props}{safe} "
def simple_docx_cell(value: Any, width: int, bold: bool = False) -> str:
run_props = ""
if bold:
run_props += " "
run_props += " "
safe = escape(text(value))
return (
""
f" "
" "
f"{run_props}{safe} "
" "
)
def simple_docx_table(headers: list[str], rows: list[list[Any]], widths: list[int] | None = None) -> str:
if not rows:
return simple_docx_paragraph("normal", "暂无数据。")
total_width = 15400
widths = widths or [total_width // len(headers)] * len(headers)
if len(widths) != len(headers):
widths = [total_width // len(headers)] * len(headers)
grid = "".join(f"" for width in widths)
header_cells = "".join(simple_docx_cell(head, widths[idx], bold=True) for idx, head in enumerate(headers))
body_rows = []
for row in rows:
padded = list(row)[: len(headers)] + [""] * max(0, len(headers) - len(row))
cells = "".join(simple_docx_cell(padded[idx], widths[idx]) for idx in range(len(headers)))
body_rows.append(f"{cells} ")
return (
""
""
""
""
""
""
""
""
""
""
""
" "
""
""
""
" "
" "
f"{grid} "
f"{header_cells} "
f"{''.join(body_rows)}"
" "
)
def simple_docx_blocks(report: dict[str, Any]) -> list[str]:
frameworks = report.get("framework_selection", {})
readiness = report.get("strategy_readiness", {})
hygiene = report.get("strategic_hygiene", {})
history = report.get("historical_behavior_analysis", {})
sensitivity = report.get("sensitivity", {})
next_information = report.get("next_information", {})
action_map = index_by_id(report.get("our_actions", []))
opponent_action_map = index_by_id(report.get("opponent_actions", []))
blocks = [
simple_docx_paragraph("title", report["title"]),
simple_docx_paragraph("normal", f"生成日期:{report['generated_at']}"),
simple_docx_paragraph("heading", "一句话结论"),
simple_docx_paragraph("normal", report["summary"]["one_sentence"]),
simple_docx_paragraph("heading", "推荐动作"),
simple_docx_paragraph("bullet", f"主动作:{report['summary']['recommended_action']}"),
simple_docx_paragraph("bullet", f"辅助动作:{report['summary']['secondary_action'] or '无'}"),
simple_docx_paragraph("bullet", f"避免动作:{report['summary']['avoid_action']}"),
simple_docx_paragraph("bullet", f"推荐理由:{report['summary']['reason']}"),
simple_docx_paragraph("bullet", f"承诺判断:{report['summary']['commitment_verdict']}"),
simple_docx_paragraph("heading", "框架选择与组合"),
simple_docx_paragraph("normal", f"主框架:{text(frameworks.get('primary'))}"),
simple_docx_paragraph("normal", f"组合逻辑:{text(frameworks.get('combination_logic'))}"),
simple_docx_paragraph("heading", "辅助镜头"),
]
for lens in frameworks.get("secondary_lenses", []):
blocks.append(simple_docx_paragraph("bullet", text(lens)))
blocks.extend(
[
simple_docx_paragraph("heading", "排除或降级的框架"),
simple_docx_table(
["框架", "原因"],
[[item.get("name"), item.get("reason")] for item in frameworks.get("excluded_frameworks", [])],
[3000, 12400],
),
simple_docx_paragraph("heading", "策略准备度"),
simple_docx_paragraph("normal", f"准备度:{fmt_pct(readiness.get('score'))}|状态:{text(readiness.get('status'))}"),
simple_docx_paragraph("normal", f"判断:{text(readiness.get('interpretation'))}"),
simple_docx_paragraph("heading", "剩余缺口"),
]
)
for gap in readiness.get("remaining_gaps", []):
blocks.append(simple_docx_paragraph("bullet", text(gap)))
blocks.extend(
[
simple_docx_paragraph("heading", "战略卫生检查"),
simple_docx_table(
["检查", "状态", "说明"],
[[item.get("name"), item.get("status"), item.get("note")] for item in hygiene.get("checks", [])],
[3400, 2200, 9800],
),
simple_docx_paragraph("heading", "历史行为与理性概率校正"),
simple_docx_paragraph("normal", text(history.get("principle"))),
simple_docx_table(
["玩家", "先验理性", "历史校正后", "调整", "证据质量", "解释"],
[
[item.get("player"), fmt_pct(item.get("prior_rationality")), fmt_pct(item.get("adjusted_rationality")), fmt_pct(item.get("adjustment")), fmt_pct(item.get("evidence_quality")), item.get("interpretation")]
for item in history.get("player_adjustments", [])
],
[1800, 1700, 1900, 1400, 1700, 6900],
),
simple_docx_paragraph("heading", "真实历史行为样本"),
simple_docx_table(
["玩家", "历史事件", "观察动作", "相似度", "兑现率", "来源", "启发"],
[
[item.get("player"), item.get("event"), item.get("observed_action"), fmt_pct(item.get("context_similarity")), fmt_pct(item.get("commitment_follow_through")), item.get("source"), item.get("implication")]
for item in history.get("history_rows", [])
],
[1400, 4200, 2200, 1200, 1200, 2200, 3000],
),
simple_docx_paragraph("heading", "经验参考分析"),
simple_docx_table(
["参考类", "模式", "理性调整", "置信度", "启发"],
[
[item.get("reference_class"), item.get("pattern"), fmt_pct(item.get("rationality_adjustment")), fmt_pct(item.get("confidence")), item.get("implication")]
for item in history.get("reference_rows", [])
],
[3000, 5200, 1700, 1500, 4000],
),
simple_docx_paragraph("heading", "敏感性与稳定性"),
simple_docx_paragraph("normal", f"稳定性:{text(sensitivity.get('stability'))}|收益差距:{fmt_num(sensitivity.get('payoff_gap'))}"),
simple_docx_paragraph("normal", f"最危险假设:{text(sensitivity.get('most_dangerous_assumption'))}"),
simple_docx_table(
["压力测试", "假设变化", "影响", "对推荐的影响"],
[
[item.get("name"), item.get("assumption_shift"), item.get("expected_effect"), item.get("recommendation_impact")]
for item in sensitivity.get("stress_tests", [])
],
[2500, 4500, 4200, 4200],
),
simple_docx_paragraph("heading", "下一步信息"),
simple_docx_table(
["信息", "为什么重要", "收集方法"],
[[item.get("item"), item.get("why_it_matters"), item.get("collection_method")] for item in next_information.get("priorities", [])],
[3600, 5600, 6200],
),
simple_docx_paragraph("heading", "玩家地图"),
simple_docx_table(
["玩家", "角色", "目标", "约束"],
[[p.get("name"), p.get("role"), text(p.get("objectives")), text(p.get("constraints"))] for p in report.get("players", [])],
[1800, 2800, 5400, 5400],
),
simple_docx_paragraph("heading", "我们的策略集合"),
simple_docx_table(
["动作", "说明", "数据基础"],
[[a.get("name"), a.get("description"), a.get("data_basis")] for a in report.get("our_actions", [])],
[2200, 10600, 2600],
),
]
)
reaction_rows = []
for estimate in report.get("reaction_estimates", []):
for likely in estimate.get("likely_actions", []):
reaction_rows.append(
[
estimate.get("opponent"),
label(estimate.get("if_we"), action_map),
label(likely.get("action"), opponent_action_map),
fmt_pct(likely.get("probability")),
likely.get("rationale"),
]
)
blocks.extend(
[
simple_docx_paragraph("heading", "对手反应地图"),
simple_docx_table(["对手", "如果我们", "可能动作", "概率", "理由"], reaction_rows, [3000, 2200, 2600, 1400, 6200]),
simple_docx_paragraph("heading", "Payoff 矩阵"),
simple_docx_table(
["我们的动作", "对方动作", "我方", "对方", "渠道", "基础", "说明"],
[
[
label(row.get("our_action"), action_map),
label(row.get("opponent_action"), opponent_action_map, text(row.get("opponent_action"))),
fmt_num(row.get("our_payoff")),
fmt_num(row.get("opponent_payoff")),
fmt_num(row.get("channel_payoff")),
row.get("basis"),
row.get("notes"),
]
for row in report.get("payoff_matrix", [])
],
[2200, 2600, 1000, 1000, 1000, 1500, 6100],
),
simple_docx_paragraph("heading", "预期收益排序"),
simple_docx_table(
["动作", "预期收益", "计算方式"],
[[item["action_name"], fmt_num(item["expected_payoff"]), item["method"]] for item in report.get("expected_payoffs", [])],
[4200, 2800, 8400],
),
simple_docx_paragraph("heading", "承诺可信度"),
simple_docx_table(
["动作", "承诺", "可信度", "原始评分", "历史校正", "证据基础", "历史校正说明"],
[
[item.get("action_name"), item.get("commitment"), item.get("level"), fmt_pct(item.get("raw_score")), fmt_pct(item.get("history_adjusted_score")), item.get("evidence_basis"), item.get("history_adjustment_note")]
for item in report.get("commitment_tests", [])
],
[1700, 2600, 1500, 1400, 1600, 1600, 5000],
),
simple_docx_paragraph("heading", "动态更新日志"),
simple_docx_table(
["轮次", "日期", "阶段", "新增信息", "判断", "状态"],
[[r.get("round"), r.get("date"), r.get("stage"), text(r.get("new_information")), r.get("interim_judgment"), r.get("recommendation_status")] for r in report.get("rounds", [])],
[900, 1800, 1900, 5200, 4400, 1200],
),
simple_docx_paragraph("heading", "风险与边界"),
]
)
for warning in report.get("warnings", []):
blocks.append(simple_docx_paragraph("bullet", text(warning)))
return blocks
def write_simple_docx(path: Path, report: dict[str, Any]) -> None:
blocks = "\n".join(simple_docx_blocks(report))
document_xml = f"""
{blocks}
"""
rels_xml = """
"""
content_types_xml = """
"""
with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as docx:
docx.writestr("[Content_Types].xml", content_types_xml)
docx.writestr("_rels/.rels", rels_xml)
docx.writestr("word/document.xml", document_xml)
def write_docx(path: Path, report: dict[str, Any]) -> None:
try:
from docx import Document
except Exception as exc: # pragma: no cover - fallback depends on environment.
write_simple_docx(path, report)
return
document = Document()
document.add_heading(report["title"], 0)
document.add_paragraph(f"生成日期:{report['generated_at']}")
document.add_heading("一句话结论", level=1)
document.add_paragraph(report["summary"]["one_sentence"])
document.add_heading("推荐动作", level=1)
for line in [
f"主动作:{report['summary']['recommended_action']}",
f"辅助动作:{report['summary']['secondary_action'] or '无'}",
f"避免动作:{report['summary']['avoid_action']}",
f"推荐理由:{report['summary']['reason']}",
f"承诺判断:{report['summary']['commitment_verdict']}",
]:
document.add_paragraph(line, style="List Bullet")
def add_table(title: str, headers: list[str], rows: list[list[Any]]) -> None:
document.add_heading(title, level=1)
if not rows:
document.add_paragraph("暂无数据。")
return
table = document.add_table(rows=1, cols=len(headers))
table.style = "Table Grid"
for idx, head in enumerate(headers):
table.rows[0].cells[idx].text = head
for row in rows:
cells = table.add_row().cells
for idx, cell in enumerate(row):
cells[idx].text = text(cell)
action_map = index_by_id(report.get("our_actions", []))
opponent_action_map = index_by_id(report.get("opponent_actions", []))
frameworks = report.get("framework_selection", {})
readiness = report.get("strategy_readiness", {})
hygiene = report.get("strategic_hygiene", {})
history = report.get("historical_behavior_analysis", {})
sensitivity = report.get("sensitivity", {})
next_information = report.get("next_information", {})
document.add_heading("框架选择与组合", level=1)
document.add_paragraph(f"主框架:{text(frameworks.get('primary'))}")
document.add_paragraph(f"组合逻辑:{text(frameworks.get('combination_logic'))}")
for lens in frameworks.get("secondary_lenses", []):
document.add_paragraph(text(lens), style="List Bullet")
add_table("排除或降级的框架", ["框架", "原因"], [[item.get("name"), item.get("reason")] for item in frameworks.get("excluded_frameworks", [])])
document.add_heading("策略准备度", level=1)
document.add_paragraph(f"准备度:{fmt_pct(readiness.get('score'))}|状态:{text(readiness.get('status'))}")
document.add_paragraph(f"判断:{text(readiness.get('interpretation'))}")
for gap in readiness.get("remaining_gaps", []):
document.add_paragraph(text(gap), style="List Bullet")
add_table("战略卫生检查", ["检查", "状态", "说明"], [[item.get("name"), item.get("status"), item.get("note")] for item in hygiene.get("checks", [])])
document.add_heading("历史行为与理性概率校正", level=1)
document.add_paragraph(text(history.get("principle")))
add_table(
"玩家理性概率校正",
["玩家", "先验理性", "历史校正后", "调整", "证据质量", "解释"],
[
[item.get("player"), fmt_pct(item.get("prior_rationality")), fmt_pct(item.get("adjusted_rationality")), fmt_pct(item.get("adjustment")), fmt_pct(item.get("evidence_quality")), item.get("interpretation")]
for item in history.get("player_adjustments", [])
],
)
add_table(
"真实历史行为样本",
["玩家", "历史事件", "观察动作", "相似度", "兑现率", "来源", "启发"],
[
[item.get("player"), item.get("event"), item.get("observed_action"), fmt_pct(item.get("context_similarity")), fmt_pct(item.get("commitment_follow_through")), item.get("source"), item.get("implication")]
for item in history.get("history_rows", [])
],
)
add_table(
"经验参考分析",
["参考类", "模式", "理性调整", "置信度", "启发"],
[[item.get("reference_class"), item.get("pattern"), fmt_pct(item.get("rationality_adjustment")), fmt_pct(item.get("confidence")), item.get("implication")] for item in history.get("reference_rows", [])],
)
document.add_heading("敏感性与稳定性", level=1)
document.add_paragraph(f"稳定性:{text(sensitivity.get('stability'))}|收益差距:{fmt_num(sensitivity.get('payoff_gap'))}")
document.add_paragraph(f"最危险假设:{text(sensitivity.get('most_dangerous_assumption'))}")
add_table(
"压力测试",
["压力测试", "假设变化", "影响", "对推荐的影响"],
[[item.get("name"), item.get("assumption_shift"), item.get("expected_effect"), item.get("recommendation_impact")] for item in sensitivity.get("stress_tests", [])],
)
add_table(
"下一步信息",
["信息", "为什么重要", "收集方法"],
[[item.get("item"), item.get("why_it_matters"), item.get("collection_method")] for item in next_information.get("priorities", [])],
)
add_table(
"玩家地图",
["玩家", "角色", "目标", "约束"],
[[p.get("name"), p.get("role"), text(p.get("objectives")), text(p.get("constraints"))] for p in report.get("players", [])],
)
reaction_rows = []
for estimate in report.get("reaction_estimates", []):
for likely in estimate.get("likely_actions", []):
reaction_rows.append(
[estimate.get("opponent"), label(estimate.get("if_we"), action_map), label(likely.get("action"), opponent_action_map), fmt_pct(likely.get("probability")), likely.get("rationale")]
)
add_table("对手反应地图", ["对手", "如果我们", "可能动作", "概率", "理由"], reaction_rows)
add_table(
"Payoff 矩阵",
["我们的动作", "对方动作", "我方收益", "对方收益", "渠道收益", "基础", "说明"],
[
[
label(row.get("our_action"), action_map),
label(row.get("opponent_action"), opponent_action_map, text(row.get("opponent_action"))),
fmt_num(row.get("our_payoff")),
fmt_num(row.get("opponent_payoff")),
fmt_num(row.get("channel_payoff")),
row.get("basis"),
row.get("notes"),
]
for row in report.get("payoff_matrix", [])
],
)
add_table(
"承诺可信度",
["动作", "承诺", "可信度", "原始评分", "历史校正评分", "历史校正说明"],
[[item.get("action_name"), item.get("commitment"), item.get("level"), fmt_pct(item.get("raw_score")), fmt_pct(item.get("history_adjusted_score")), item.get("history_adjustment_note")] for item in report.get("commitment_tests", [])],
)
add_table(
"动态更新日志",
["轮次", "日期", "阶段", "新增信息", "判断", "状态"],
[[r.get("round"), r.get("date"), r.get("stage"), text(r.get("new_information")), r.get("interim_judgment"), r.get("recommendation_status")] for r in report.get("rounds", [])],
)
document.add_heading("风险与边界", level=1)
for warning in report.get("warnings", []):
document.add_paragraph(text(warning), style="List Bullet")
document.save(path)
def write_pdf(path: Path, html: str) -> str | None:
try:
from weasyprint import HTML
except Exception as exc: # pragma: no cover - fallback depends on environment.
note = path.with_suffix(".pdf.unavailable.txt")
note.write_text(f"WeasyPrint is unavailable, so PDF export was skipped: {exc}\nUse the HTML report's Print / Save as PDF action instead.\n", encoding="utf-8")
return str(note)
HTML(string=html, base_url=str(ROOT)).write_pdf(path)
return None
def build_bundle(input_file: Path, output_dir: Path, update_file: Path | None = None, no_pdf: bool = False) -> dict[str, str]:
request = load_json(input_file)
if update_file:
request = merge_update(request, load_json(update_file))
report = build_report(request)
output_dir.mkdir(parents=True, exist_ok=True)
stem = input_file.stem.replace("_", "-")
canonical_path = output_dir / f"{stem}.canonical.json"
markdown_path = output_dir / f"{stem}.md"
html_path = output_dir / f"{stem}.html"
docx_path = output_dir / f"{stem}.docx"
pdf_path = output_dir / f"{stem}.pdf"
write_json(canonical_path, report)
markdown_path.write_text(render_markdown(report), encoding="utf-8")
html = render_html(report)
html_path.write_text(html, encoding="utf-8")
write_docx(docx_path, report)
pdf_note = None
if not no_pdf:
pdf_note = write_pdf(pdf_path, html)
return {
"canonical_json": str(canonical_path),
"markdown": str(markdown_path),
"html": str(html_path),
"docx": str(docx_path),
"pdf": str(pdf_path if pdf_path.exists() else pdf_note or ""),
}
def main() -> None:
parser = argparse.ArgumentParser(description="Generate a game theory strategy report bundle.")
parser.add_argument("input_file", help="Structured game theory request JSON.")
parser.add_argument("output_dir", help="Directory where report artifacts will be written.")
parser.add_argument("--update", help="Optional update JSON to merge before rendering.")
parser.add_argument("--no-pdf", action="store_true", help="Skip automated PDF rendering.")
args = parser.parse_args()
outputs = build_bundle(
input_file=Path(args.input_file).resolve(),
output_dir=Path(args.output_dir).resolve(),
update_file=Path(args.update).resolve() if args.update else None,
no_pdf=args.no_pdf,
)
print(json.dumps(outputs, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()