AI Agent金融顾问：让AI帮你管钱

一、开场：理财焦虑的时代

大家好，我是老金。

现在的理财环境：

• 4000+只基金，不知道选哪个
• 股市波动大，追涨杀跌亏不少
• 理财产品复杂，看不懂条款
• 财务规划混乱，月月光

找理财顾问？

• 私人银行：门槛1000万起步
• 独立顾问：咨询费500/小时
• 银行理财经理：总推销自家产品

AI金融顾问正在改变这一切：

• 24小时在线服务
• 中立客观的建议
• 个性化财务规划
• 成本几乎为零

今天分享AI Agent在金融顾问领域的应用。

二、金融顾问Agent架构

系统架构

┌─────────────────────────────────────────────────────────┐
│                  AI金融顾问Agent架构                     │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  用户接口层                                             │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐                │
│  │  投资咨询 │  │  财务规划 │  │  风险评估 │                │
│  └────┬────┘  └────┬────┘  └────┬────┘                │
│       └────────────┼────────────┘                       │
│                    ↓                                    │
│  ┌─────────────────────────────────────────────────┐   │
│  │              用户画像层                          │   │
│  │  • 风险偏好                                      │   │
│  │  • 财务状况                                      │   │
│  │  • 投资目标                                      │   │
│  │  • 时间周期                                      │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │              分析决策层                          │   │
│  │  • 资产配置                                      │   │
│  │  • 产品筛选                                      │   │
│  │  • 风险管理                                      │   │
│  │  • 调仓建议                                      │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │              数据层                              │   │
│  │  • 市场数据                                      │   │
│  │  • 产品信息                                      │   │
│  │  • 研报分析                                      │   │
│  │  • 新闻舆情                                      │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  ⚠️ 投资有风险，建议仅供参考                           │
│                                                         │
└─────────────────────────────────────────────────────────┘

三、风险评估Agent

风险画像评估

class RiskAssessmentAgent:
    """风险评估Agent"""

    def __init__(self, llm):
        self.llm = llm

    async def assess_risk_profile(self, questionnaire: dict) -> dict:
        """评估风险画像"""
        # 1. 计算风险承受能力
        capacity_score = self._calculate_risk_capacity(questionnaire)

        # 2. 计算风险偏好
        tolerance_score = self._calculate_risk_tolerance(questionnaire)

        # 3. 综合评估
        overall_score = (capacity_score * 0.4 + tolerance_score * 0.6)

        # 4. 确定风险类型
        risk_type = self._determine_risk_type(overall_score)

        return {
            "risk_capacity_score": capacity_score,
            "risk_tolerance_score": tolerance_score,
            "overall_score": overall_score,
            "risk_type": risk_type,
            "recommended_equity_allocation": self._get_equity_allocation(risk_type),
            "analysis": await self._generate_analysis(questionnaire, risk_type)
        }

    def _calculate_risk_capacity(self, questionnaire: dict) -> float:
        """计算风险承受能力"""
        score = 0

        # 年龄
        age = questionnaire.get('age', 30)
        if age < 30:
            score += 25
        elif age < 40:
            score += 20
        elif age < 50:
            score += 15
        elif age  1000000:
            score += 25
        elif net_worth > 500000:
            score += 20
        elif net_worth > 100000:
            score += 15
        else:
            score += 10

        # 应急资金
        emergency_fund = questionnaire.get('emergency_fund_months', 0)
        if emergency_fund >= 6:
            score += 15
        elif emergency_fund >= 3:
            score += 10
        else:
            score += 5

        # 投资经验
        experience = questionnaire.get('investment_experience', 'none')
        score += {
            "none": 5,
            "beginner": 10,
            "intermediate": 15,
            "advanced": 20
        }[experience]

        return score / 100  # 归一化到0-1

    def _calculate_risk_tolerance(self, questionnaire: dict) -> float:
        """计算风险偏好"""
        score = 0

        # 市场下跌反应
        reaction = questionnaire.get('market_drop_reaction', 'hold')
        score += {
            "buy_more": 20,
            "hold": 15,
            "sell_some": 10,
            "sell_all": 5
        }[reaction]

        # 可接受的最大亏损
        max_loss = questionnaire.get('max_acceptable_loss', 10)
        if max_loss >= 30:
            score += 20
        elif max_loss >= 20:
            score += 15
        elif max_loss >= 10:
            score += 10
        else:
            score += 5

        # 投资目标
        goal = questionnaire.get('investment_goal', 'balance')
        score += {
            "aggressive_growth": 20,
            "growth": 15,
            "balance": 10,
            "conservative": 5
        }[goal]

        # 心理承受
        stress_level = questionnaire.get('investment_stress', 'low')
        score += {
            "low": 20,
            "medium": 10,
            "high": 5
        }[stress_level]

        return score / 80  # 归一化

    def _determine_risk_type(self, score: float) -> str:
        """确定风险类型"""
        if score >= 0.8:
            return "激进型"
        elif score >= 0.6:
            return "积极型"
        elif score >= 0.4:
            return "稳健型"
        elif score >= 0.2:
            return "保守型"
        else:
            return "极度保守型"

    def _get_equity_allocation(self, risk_type: str) -> int:
        """获取股票资产配置比例"""
        allocation = {
            "激进型": 80,
            "积极型": 60,
            "稳健型": 40,
            "保守型": 20,
            "极度保守型": 10
        }
        return allocation.get(risk_type, 40)

四、资产配置Agent

智能资产配置

class AssetAllocationAgent:
    """资产配置Agent"""

    # 基础配置模板
    ALLOCATION_TEMPLATES = {
        "激进型": {
            "股票基金": 50,
            "债券基金": 15,
            "货币基金": 5,
            "另类投资": 20,
            "黄金": 10
        },
        "积极型": {
            "股票基金": 40,
            "债券基金": 25,
            "货币基金": 10,
            "另类投资": 15,
            "黄金": 10
        },
        "稳健型": {
            "股票基金": 25,
            "债券基金": 40,
            "货币基金": 15,
            "另类投资": 10,
            "黄金": 10
        },
        "保守型": {
            "股票基金": 15,
            "债券基金": 50,
            "货币基金": 25,
            "另类投资": 5,
            "黄金": 5
        },
        "极度保守型": {
            "股票基金": 5,
            "债券基金": 55,
            "货币基金": 35,
            "另类投资": 0,
            "黄金": 5
        }
    }

    async def create_portfolio(self, risk_profile: dict,
                              investment_amount: float,
                              goals: list = None) -> dict:
        """创建投资组合"""
        # 1. 获取基础配置
        base_allocation = self.ALLOCATION_TEMPLATES.get(
            risk_profile['risk_type'], 
            self.ALLOCATION_TEMPLATES["稳健型"]
        )

        # 2. 根据目标调整
        adjusted_allocation = await self._adjust_for_goals(
            base_allocation, goals
        )

        # 3. 计算具体金额
        allocation_with_amount = {}
        for asset, percentage in adjusted_allocation.items():
            allocation_with_amount[asset] = {
                "percentage": percentage,
                "amount": investment_amount * percentage / 100
            }

        # 4. 推荐具体产品
        product_recommendations = await self._recommend_products(
            adjusted_allocation, risk_profile
        )

        return {
            "total_amount": investment_amount,
            "allocation": allocation_with_amount,
            "products": product_recommendations,
            "expected_return": self._estimate_return(adjusted_allocation),
            "expected_risk": self._estimate_risk(adjusted_allocation),
            "rebalancing_frequency": "季度",
            "disclaimer": "⚠️ 投资有风险，历史业绩不代表未来表现"
        }

    async def _adjust_for_goals(self, base: dict, goals: list) -> dict:
        """根据目标调整配置"""
        if not goals:
            return base

        adjusted = base.copy()

        for goal in goals:
            # 根据目标时间调整
            years = goal.get('years', 5)
            if years  10:
                # 长期目标：可以更激进
                adjusted["股票基金"] = min(70, adjusted.get("股票基金", 0) + 10)
                adjusted["债券基金"] = max(10, adjusted.get("债券基金", 0) - 10)

        return adjusted

    async def _recommend_products(self, allocation: dict,
                                 risk_profile: dict) -> list:
        """推荐具体产品"""
        recommendations = []

        for asset_class, percentage in allocation.items():
            if percentage > 0:
                products = await self._get_top_products(asset_class, count=3)
                recommendations.append({
                    "asset_class": asset_class,
                    "allocation": percentage,
                    "products": products
                })

        return recommendations

    async def _get_top_products(self, asset_class: str, 
                               count: int = 3) -> list:
        """获取优质产品"""
        # 从产品数据库筛选
        # 实际应查询基金/产品数据库
        return [
            {"name": f"示例{asset_class}A", "code": "000001", "rating": 5},
            {"name": f"示例{asset_class}B", "code": "000002", "rating": 4},
            {"name": f"示例{asset_class}C", "code": "000003", "rating": 4}
        ]

动态再平衡

class RebalancingAgent:
    """再平衡Agent"""

    async def check_rebalance(self, portfolio: dict,
                             current_values: dict) -> dict:
        """检查是否需要再平衡"""
        # 计算当前配置
        total_value = sum(current_values.values())
        current_allocation = {
            asset: value / total_value * 100
            for asset, value in current_values.items()
        }

        # 与目标配置比较
        deviations = {}
        need_rebalance = False

        for asset, target_pct in portfolio['target_allocation'].items():
            current_pct = current_allocation.get(asset, 0)
            deviation = abs(current_pct - target_pct)
            deviations[asset] = {
                "target": target_pct,
                "current": current_pct,
                "deviation": deviation
            }

            if deviation > 5:  # 偏离超过5%
                need_rebalance = True

        # 生成调整建议
        if need_rebalance:
            adjustments = await self._generate_adjustments(
                portfolio['target_allocation'], current_allocation
            )
        else:
            adjustments = []

        return {
            "need_rebalance": need_rebalance,
            "deviations": deviations,
            "adjustments": adjustments,
            "current_total": total_value
        }

    async def _generate_adjustments(self, target: dict, 
                                   current: dict) -> list:
        """生成调整方案"""
        adjustments = []

        for asset in set(target.keys()) | set(current.keys()):
            target_pct = target.get(asset, 0)
            current_pct = current.get(asset, 0)
            diff = target_pct - current_pct

            if abs(diff) > 0.5:
                adjustments.append({
                    "asset": asset,
                    "action": "买入" if diff > 0 else "卖出",
                    "percentage": abs(diff)
                })

        return adjustments

五、基金筛选Agent

基金筛选逻辑

class FundSelectionAgent:
    """基金筛选Agent"""

    async def select_funds(self, criteria: dict, top_n: int = 10) -> list:
        """筛选基金"""
        # 从数据库获取候选基金
        candidates = await self._get_candidate_funds(criteria)

        # 多维度评分
        scored_funds = []
        for fund in candidates:
            score = await self._score_fund(fund, criteria)
            scored_funds.append({
                **fund,
                "score": score
            })

        # 排序
        sorted_funds = sorted(scored_funds, key=lambda x: x['score'], reverse=True)

        return sorted_funds[:top_n]

    async def _score_fund(self, fund: dict, criteria: dict) -> float:
        """基金评分"""
        score = 0

        # 1. 业绩评分（40%）
        performance_score = self._score_performance(fund)
        score += performance_score * 0.4

        # 2. 风险评分（30%）
        risk_score = self._score_risk(fund, criteria.get('risk_preference', 'medium'))
        score += risk_score * 0.3

        # 3. 基金经理评分（15%）
        manager_score = await self._score_manager(fund['manager'])
        score += manager_score * 0.15

        # 4. 基金公司评分（10%）
        company_score = self._score_company(fund['company'])
        score += company_score * 0.1

        # 5. 费用评分（5%）
        fee_score = self._score_fee(fund['fee'])
        score += fee_score * 0.05

        return score

    def _score_performance(self, fund: dict) -> float:
        """业绩评分"""
        score = 0

        # 近1年收益率
        return_1y = fund.get('return_1y', 0)
        if return_1y > 30:
            score += 30
        elif return_1y > 15:
            score += 25
        elif return_1y > 0:
            score += 20

        # 近3年收益率
        return_3y = fund.get('return_3y', 0)
        if return_3y > 50:
            score += 30
        elif return_3y > 30:
            score += 25
        elif return_3y > 10:
            score += 20

        # 同类排名
        rank_percentile = fund.get('rank_percentile', 50)
        if rank_percentile <= 10:
            score += 40
        elif rank_percentile <= 25:
            score += 30
        elif rank_percentile  float:
        """风险评分"""
        score = 0

        # 最大回撤
        max_drawdown = fund.get('max_drawdown', 0)
        if max_drawdown < 10:
            score += 40
        elif max_drawdown < 20:
            score += 30
        elif max_drawdown < 30:
            score += 20

        # 波动率
        volatility = fund.get('volatility', 0)
        if volatility < 10:
            score += 30
        elif volatility < 20:
            score += 20
        elif volatility  2:
            score += 30
        elif sharpe > 1:
            score += 20
        elif sharpe > 0:
            score += 10

        return score

六、财务规划Agent

综合财务规划

class FinancialPlanningAgent:
    """财务规划Agent"""

    async def create_financial_plan(self, profile: dict) -> dict:
        """创建财务规划"""
        # 1. 财务诊断
        diagnosis = await self._diagnose_finances(profile)

        # 2. 设定目标
        goals = await self._set_goals(profile)

        # 3. 制定计划
        plan = await self._create_plan(profile, goals)

        # 4. 执行建议
        actions = await self._generate_actions(plan)

        return {
            "diagnosis": diagnosis,
            "goals": goals,
            "plan": plan,
            "actions": actions,
            "projections": await self._project_wealth(profile, plan)
        }

    async def _diagnose_finances(self, profile: dict) -> dict:
        """财务诊断"""
        income = profile.get('monthly_income', 0)
        expenses = profile.get('monthly_expenses', 0)
        assets = profile.get('total_assets', 0)
        liabilities = profile.get('total_liabilities', 0)

        # 计算关键指标
        savings_rate = (income - expenses) / income * 100 if income > 0 else 0
        debt_ratio = liabilities / assets * 100 if assets > 0 else 0
        emergency_fund_months = assets * 0.2 / expenses if expenses > 0 else 0  # 假设20%为流动性资产

        issues = []
        if savings_rate  50:
            issues.append("负债率较高，建议优先偿还债务")
        if emergency_fund_months  dict:
        """制定计划"""
        monthly_savings = profile.get('monthly_income', 0) - profile.get('monthly_expenses', 0)

        plan = {
            "emergency_fund": {
                "target": profile.get('monthly_expenses', 0) * 6,
                "priority": 1,
                "timeline": "6个月内"
            },
            "insurance": {
                "recommended": ["重疾险", "意外险", "医疗险"],
                "budget": monthly_savings * 0.1,
                "priority": 2
            },
            "investment": {
                "monthly_amount": monthly_savings * 0.6,
                "allocation": await self._get_investment_allocation(profile),
                "priority": 3
            },
            "retirement": {
                "monthly_saving": monthly_savings * 0.2,
                "target": profile.get('monthly_expenses', 0) * 12 * 25,  # 25倍年支出
                "priority": 4
            }
        }

        return plan

    async def _project_wealth(self, profile: dict, plan: dict) -> dict:
        """财富预测"""
        current_wealth = profile.get('total_assets', 0) - profile.get('total_liabilities', 0)
        monthly_saving = plan['investment']['monthly_amount']

        # 简化预测（假设年化收益6%）
        projections = []
        for year in [1, 3, 5, 10, 20, 30]:
            future_value = current_wealth * (1.06 ** year) + 
                          monthly_saving * 12 * ((1.06 ** year - 1) / 0.06)
            projections.append({
                "year": year,
                "projected_wealth": round(future_value, 0)
            })

        return {
            "projections": projections,
            "assumptions": {
                "annual_return": "6%",
                "monthly_saving": monthly_saving
            }
        }

七、市场分析Agent

市场情绪分析

class MarketAnalysisAgent:
    """市场分析Agent"""

    async def analyze_market_sentiment(self) -> dict:
        """分析市场情绪"""
        # 获取多维度数据
        news_sentiment = await self._analyze_news()
        social_sentiment = await self._analyze_social_media()
        technical_indicators = await self._get_technical_indicators()

        # 综合情绪指数
        overall_sentiment = (
            news_sentiment['score'] * 0.3 +
            social_sentiment['score'] * 0.2 +
            technical_indicators['score'] * 0.5
        )

        return {
            "overall_sentiment": overall_sentiment,
            "interpretation": self._interpret_sentiment(overall_sentiment),
            "news_sentiment": news_sentiment,
            "social_sentiment": social_sentiment,
            "technical_indicators": technical_indicators,
            "recommendation": self._get_sentiment_recommendation(overall_sentiment)
        }

    def _interpret_sentiment(self, score: float) -> str:
        """解读情绪指数"""
        if score > 70:
            return "极度贪婪"
        elif score > 55:
            return "贪婪"
        elif score > 45:
            return "中性"
        elif score > 30:
            return "恐惧"
        else:
            return "极度恐惧"

    def _get_sentiment_recommendation(self, score: float) -> str:
        """情绪指标建议"""
        if score > 70:
            return "市场过热，建议谨慎，可适当减仓"
        elif score > 55:
            return "市场情绪乐观，保持关注"
        elif score > 45:
            return "市场情绪平稳，按计划执行"
        elif score > 30:
            return "市场恐慌，可能是布局机会"
        else:
            return "极度恐慌，逆向投资机会可能出现"

八、安全与合规

投资建议免责声明

class FinancialDisclaimer:
    """金融免责声明"""

    DISCLAIMER = """
⚠️ 重要声明：

1. 本AI顾问提供的建议仅供参考，不构成投资建议
2. 投资有风险，入市需谨慎
3. 过往业绩不代表未来表现
4. 请根据自身情况独立判断，必要时咨询专业人士
5. 本系统不承担投资损失的责任

投资决策应基于您的：
- 风险承受能力
- 投资目标
- 财务状况
- 投资经验
    """

    @staticmethod
    def add_to_response(response: dict) -> dict:
        """添加免责声明"""
        response['disclaimer'] = FinancialDisclaimer.DISCLAIMER
        return response

九、最佳实践

服务对比

功能边界

CAN_DO = [
    "风险评估和画像",
    "资产配置建议",
    "基金筛选排序",
    "财务规划建议",
    "市场信息解读"
]

CANNOT_DO = [
    "承诺收益",
    "代客操作",
    "内幕信息",
    "保本保收益",
    "替用户做最终决策"
]

十、总结

核心价值

• 普惠金融：让每个人都能享受专业理财服务
• 中立客观：算法驱动，无利益冲突
• 个性化：基于个人情况定制方案
• 持续服务：7×24小时在线

效果评估

• 用户满意度：85%+
• 方案采纳率：60%+
• 平均节省费用：每年数千元

下期预告

明天聊聊AI Agent法律助手——法律咨询也能AI化！

往期回顾

• AI Agent医疗问诊：AI当医生靠谱吗
• AI Agent教育辅导：个性化学习的未来
• AI Agent智能运维：让系统自己照顾自己