第03章 风险偏好分析：AI如何读懂客户的投资心理
第03章 风险偏好分析：AI如何读懂客户的投资心理
“每个人都知道自己的风险承受能力，直到账户真正亏损的那一天。” —— 行为金融学经典观察
风险偏好是理财服务的基石。但传统的风险测评问卷有一个巨大的缺陷：它测的是客户"认为自己"的风险承受能力，而不是真实的风险承受能力。Hermes 通过持续观察客户的行为模式，构建更真实、更动态的风险画像。
3.1 传统风险测评的问题
# ====================================================
# 风险偏好分析系统
# ====================================================
"""
传统风险测评的三大缺陷：
缺陷1：静态测评 vs 动态现实
・2021年市场大牛市，客户填问卷时：激进型（想赚更多）
・2022年市场大熊市，账户亏损30%，同一位客户：实际表现：保守型（每天问能不能赎回）
・测评结果不会自动更新，但客户的真实风险承受能力每天都在变化
缺陷2：自我报告偏差（Self-Report Bias）
・客户倾向于高估自己的风险承受能力
・在市场好的时候，大多数人认为自己能承受更大风险
・实际亏损体验和想象中的承受能力差距巨大
典型问题："如果您的投资亏损20%，您会怎么做？"
客户回答："继续持有，等待回升" （理想情况）
客户实际行为：频繁咨询+要求赎回 （真实情况）
缺陷3：问卷无法捕捉行为信号
・客户的实际操作记录才是最真实的风险偏好信息
・"说的"和"做的"往往不一致
・行为数据：历次赎回发生在什么市场条件下？咨询频率如何随市场变动？
Hermes 的解决方案：
・维护一个动态风险评分（0-100），而非静态等级
・结合问卷结果、行为数据、市场时机做综合评估
・每次市场大幅波动后，自动重新评估风险分数
・识别"口头偏好"和"行为偏好"的差距，给理财经理提示
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import date, datetime
from enum import Enum
class BehaviorSignal(str, Enum):
"""行为风险信号类型"""
PANIC_REDEEM = "恐慌性赎回" # 市场下跌时大额赎回
PANIC_INQUIRE = "频繁问询" # 市场下跌时咨询频率剧增
CHASING_HIGH = "高位追涨" # 市场高点大额买入
TAKE_PROFIT = "及时止盈" # 达到目标收益时赎回
HOLD_CALMLY = "下跌持有" # 市场下跌时保持冷静
BOTTOM_FISHING = "逢低加仓" # 市场低点追加投资
@dataclass
class BehaviorRecord:
"""客户行为记录"""
date: date
signal_type: BehaviorSignal
market_change_pct: float # 当时市场涨跌幅（%）
client_action: str # 具体行为描述
impact_on_risk_score: int # 对风险评分的影响（-10 到 +10）
@dataclass
class RiskAssessment:
"""动态风险评估结果"""
client_id: str
assessment_date: date
# 问卷分数（0-100）
questionnaire_score: int
# 行为分数（0-100，基于历史行为计算）
behavioral_score: int
# 综合分数（加权平均）
composite_score: int
# 关键发现
key_findings: List[str]
# 建议的调整
recommendations: List[str]
@property
def risk_level(self) -> str:
if self.composite_score <= 30:
return "保守型"
elif self.composite_score <= 50:
return "谨慎型"
elif self.composite_score <= 65:
return "稳健型"
elif self.composite_score <= 80:
return "积极型"
else:
return "激进型"
class DynamicRiskAnalyzer:
"""
动态风险偏好分析器
结合问卷结果和行为数据，构建更真实的风险画像
"""
# 行为信号对风险评分的影响权重
BEHAVIOR_WEIGHTS = {
BehaviorSignal.PANIC_REDEEM: -15, # 强烈降低风险评分
BehaviorSignal.PANIC_INQUIRE: -5,
BehaviorSignal.CHASING_HIGH: +5,
BehaviorSignal.TAKE_PROFIT: +3,
BehaviorSignal.HOLD_CALMLY: +8, # 提升风险评分
BehaviorSignal.BOTTOM_FISHING: +10,
}
def analyze(
self,
questionnaire_score: int,
behavior_records: List[BehaviorRecord],
client_id: str,
) -> RiskAssessment:
"""综合评估客户的真实风险偏好"""
# 计算行为分数
behavioral_score = self._calculate_behavioral_score(
questionnaire_score, behavior_records
)
# 综合分数（问卷权重 40%，行为权重 60%）
composite_score = int(
questionnaire_score * 0.4 + behavioral_score * 0.6
)
composite_score = max(0, min(100, composite_score))
# 生成关键发现
findings, recommendations = self._generate_insights(
questionnaire_score,
behavioral_score,
composite_score,
behavior_records,
)
return RiskAssessment(
client_id=client_id,
assessment_date=date.today(),
questionnaire_score=questionnaire_score,
behavioral_score=behavioral_score,
composite_score=composite_score,
key_findings=findings,
recommendations=recommendations,
)
def _calculate_behavioral_score(
self,
base_score: int,
records: List[BehaviorRecord],
) -> int:
"""基于行为记录调整风险分数"""
score = base_score
# 近6个月的行为权重更高（时间衰减）
recent_cutoff = date.today().replace(month=date.today().month - 6
if date.today().month > 6
else date.today().month + 6)
for record in records:
impact = self.BEHAVIOR_WEIGHTS.get(record.signal_type, 0)
# 恐慌性行为发生在大跌时，放大负面影响
if record.signal_type == BehaviorSignal.PANIC_REDEEM:
if record.market_change_pct < -10: # 大跌时恐慌
impact *= 1.5
elif record.market_change_pct > -3: # 小跌也恐慌
impact *= 2.0 # 更严重，说明风险承受能力极低
score += int(impact)
return max(10, min(90, score))
def _generate_insights(
self,
questionnaire: int,
behavioral: int,
composite: int,
records: List[BehaviorRecord],
) -> Tuple[List[str], List[str]]:
"""生成洞察和建议"""
findings = []
recommendations = []
gap = questionnaire - behavioral
if gap > 20:
findings.append(f"⚠️ 口头风险偏好（{questionnaire}分）显著高于行为风险偏好（{behavioral}分）")
findings.append("客户实际风险承受能力可能低于问卷所示")
recommendations.append("建议下次沟通时，使用情景模拟方式重新确认风险承受能力")
recommendations.append("持仓配置应参考行为评分，而非问卷评分")
panic_count = sum(1 for r in records if r.signal_type == BehaviorSignal.PANIC_REDEEM)
if panic_count >= 2:
findings.append(f"过去记录中有 {panic_count} 次恐慌性赎回")
recommendations.append("考虑降低权益类资产比例，增加固收类缓冲")
calm_count = sum(1 for r in records if r.signal_type == BehaviorSignal.HOLD_CALMLY)
if calm_count >= 3:
findings.append(f"过去市场下跌时保持冷静 {calm_count} 次，风险承受能力较强")
recommendations.append("可以适当提高波动较大资产的配置比例")
return findings, recommendations
# 演示
analyzer = DynamicRiskAnalyzer()
sample_behaviors = [
BehaviorRecord(date(2022, 4, 26), BehaviorSignal.PANIC_INQUIRE, -8.5, "一天内连续咨询3次", -5),
BehaviorRecord(date(2022, 10, 31), BehaviorSignal.PANIC_REDEEM, -12.0, "赎回基金仓位50%", -15),
BehaviorRecord(date(2023, 3, 15), BehaviorSignal.BOTTOM_FISHING, -5.0, "主动追加20万", +10),
BehaviorRecord(date(2024, 9, 20), BehaviorSignal.HOLD_CALMLY, -7.0, "未主动联系，保持观望", +8),
]
assessment = analyzer.analyze(
questionnaire_score=70, # 问卷显示稳健偏积极
behavior_records=sample_behaviors,
client_id="C001",
)
print(f"=== 张先生风险评估报告 ===")
print(f"问卷评分：{assessment.questionnaire_score} → 建议等级：积极型")
print(f"行为评分：{assessment.behavioral_score}")
print(f"综合评分：{assessment.composite_score} → 实际等级：{assessment.risk_level}")
print(f"\n关键发现：")
for f in assessment.key_findings:
print(f" {f}")
print(f"\n调整建议：")
for r in assessment.recommendations:
print(f" • {r}")
3.2 AI 辅助的智能风险问卷
# ====================================================
# AI 辅助的自适应风险评估问卷
# ====================================================
"""
传统问卷的另一个问题：
・问题是固定的，无法根据前序回答调整后续问题
・标准问题无法捕捉个人情况的细节
AI 增强的问卷设计：
・根据回答动态调整下一个问题（自适应）
・从客户的自然语言回答中提取风险信号
・比较回答和历史行为的一致性
"""
import asyncio
import json
from openai import AsyncOpenAI
client = AsyncOpenAI()
# 核心风险评估问题库
RISK_QUESTIONS = [
{
"id": "Q1",
"question": "您最主要的投资目标是什么？",
"options": [
"A. 保住本金，不亏损最重要",
"B. 跑赢通胀，稳定增值",
"C. 积累财富，实现中长期增值目标",
"D. 争取最大回报，可接受较大波动",
],
"risk_scores": {"A": 10, "B": 35, "C": 60, "D": 85},
},
{
"id": "Q2",
"question": "如果您的投资组合在1个月内下跌了15%，您最可能的反应是？",
"options": [
"A. 立即赎回，止损最重要",
"B. 感到担忧，可能减少一部分仓位",
"C. 冷静等待，相信会回升",
"D. 认为是买入机会，考虑追加投资",
],
"risk_scores": {"A": 5, "B": 30, "C": 65, "D": 90},
},
{
"id": "Q3",
"question": "您的这笔资金最长可以锁定几年不用？",
"options": [
"A. 不到1年",
"B. 1-3年",
"C. 3-5年",
"D. 5年以上",
],
"risk_scores": {"A": 15, "B": 40, "C": 65, "D": 85},
},
]
async def adaptive_risk_assessment(
previous_answers: List[Dict],
current_question_id: str,
user_response: str,
) -> Dict:
"""
使用 AI 分析自然语言回答，并决定下一个问题
用户可以用自己的话回答，而不必局限于选项
"""
context = "\n".join([
f"问题{a['question_id']}：{a['user_answer']}"
for a in previous_answers
])
current_q = next(
(q for q in RISK_QUESTIONS if q["id"] == current_question_id),
RISK_QUESTIONS[0],
)
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""分析以下理财客户的风险评估对话。
当前问题：{current_q['question']}
问题选项：{json.dumps(current_q['options'], ensure_ascii=False)}
客户回答："{user_response}"
已有对话记录：
{context if context else '这是第一个问题'}
请以JSON格式输出：
{{
"matched_option": "A/B/C/D",
"confidence": 0.0-1.0,
"risk_signal": "保守/中性/激进",
"key_phrase": "客户回答中最能反映风险态度的关键词",
"follow_up_needed": true/false,
"follow_up_question": "如果需要追问，写出追问内容"
}}"""
}],
response_format={"type": "json_object"},
max_tokens=200,
)
return json.loads(response.choices[0].message.content)
print("自适应风险评估系统已就绪")
print("支持：自然语言回答解析·关键信号提取·自适应追问")
本章小结
-
风险偏好是动态的，不是静态标签：一个人在牛市和熊市的风险承受能力可以天壤之别。Hermes 维护的不是一个固定的"风险等级"，而是一个随市场状况和客户行为不断更新的动态评分。
-
行为数据比问卷答案更可靠：客户说"我可以承受30%的亏损"，但历史上每次下跌5%就打电话——行为才是真正的答案。权重60%给行为分、40%给问卷分是更接近真实风险承受能力的方案。
-
识别"口头偏好"和"行为偏好"的差距是核心能力：当两者差距超过20分，就要警惕——客户可能被放在了超出其真实承受能力的风险敞口中。这种情况既损害客户利益，也埋下合规隐患。
-
AI 自适应问卷比固定选项问卷更真实：允许客户用自己的语言回答，再由 AI 分析风险信号，比强迫客户从 A/B/C/D 中选一个更能反映真实想法。很多时候，客户的"但是…"才是最重要的信息。
-
风险评估结果直接驱动后续所有服务：资产配置建议、产品推荐、报告的风险提示措辞——所有这些都应该基于最新的风险评估结果。定期（至少每年一次，大波动后立即）重新评估，保持服务与客户当前状态的一致性。
# 核心行动：为你的每位客户计算一次"行为 vs 问卷"差距
def quick_gap_analysis(questionnaire_score: int, panic_actions_last_year: int):
"""
快速评估客户的口头偏好vs行为偏好差距
panic_actions: 过去一年内恐慌性赎回或频繁焦虑咨询的次数
"""
behavior_deduction = panic_actions_last_year * 12
behavioral_estimate = max(10, questionnaire_score - behavior_deduction)
gap = questionnaire_score - behavioral_estimate
if gap > 20:
print(f"⚠️ 差距较大（{gap}分）：建议下次会面重新确认风险偏好")
elif gap > 10:
print(f"注意：存在一定差距（{gap}分）：持仓宜偏保守")
else:
print(f"✅ 差距在合理范围（{gap}分）：问卷结果可信度高")
# 用真实数据替换
quick_gap_analysis(questionnaire_score=70, panic_actions_last_year=2)
本章提示词模板
【模板1：客户风险重评估提示词】
我有一位客户，风险测评信息如下：
历史问卷评分：{questionnaire_score}（当时等级：{risk_level}）
评测时间：{assessment_date}
最近市场变化：{market_context}
客户最近行为表现：{recent_behaviors}
请帮我：
1. 评估客户的风险承受能力是否需要重新评估
2. 基于行为表现，预测真实风险偏好的大致范围
3. 提供3个合适的追问问题，以非正式的方式在电话中了解
4. 如果风险等级需要调整，建议如何调整持仓配置
5. 如何向客户解释风险重评估的必要性（不让客户感到被质疑）
【模板2：情景模拟风险测试提示词】
我需要用情景模拟的方式评估客户的真实风险承受能力。
客户背景：{client_background}
当前持仓：{current_portfolio}
历史行为：{behavior_history}
请设计3个渐进式的情景问题：
1. 温和情景：市场小幅下跌（5-8%），问客户的反应
2. 中等情景：市场大幅下跌（15-20%），问客户的应对
3. 极端情景：黑天鹅事件（30%以上），问客户的底线
每个情景需要：
- 具体描述（有画面感，不是抽象数字）
- 3个可选反应（对应不同风险偏好）
- 如何从回答中识别真实的风险态度