5. 马科维茨资产组合模型+AI金融智能体(qwen-max)识别政策意图方案（理论+Python实战）

news2025/4/25 13:17:42

目录

    • 0. 承前
    • 1. AI金融智能体
      • 1.1 What is AI金融智能体
      • 1.2 Why is AI金融智能体
      • 1.3 How to AI金融智能体
    • 2. 数据要素&计算流程
      • 2.1 参数集设置
      • 2.2 数据获取&预处理
      • 2.3 收益率计算
      • 2.4 因子构建与预期收益率计算
      • 2.5 协方差矩阵计算
      • 2.6 投资组合优化
      • 2.7 持仓筛选
      • 2.8 AI金融智能体调仓函数
    • 3. 汇总代码
    • 4. 反思
      • 4.1 不足之处
      • 4.2 提升思路
    • 5. 启后

0. 承前

本篇博文是对上一篇文章，链接:
4. 马科维茨资产组合模型+Fama-French五因子优化方案（理论+Python实战）
的资产权重进行AI干预。
本文首先使用Fama-French五因子计算出资产组合模型权重，再把权重结合政策信息输入AI模型，目的是：

  • 在金融工程中，实现AI功能在金融模型的落地的尝试；
  • AI模型对政策信息描述内容，情绪分析等进行分析，实现政策信息对金融模型的主动影响。

本文主要要素：

  • 马科维茨资产组合模型；
  • Fama-French五因子模型预期收益率；
  • AI金融智能体（通义千问：qwen-max），提示词工程；
  • 政策信息通过AI转化影响预期收益率。

如果想更加全面清晰地了解金融资产组合模型进化论的体系架构，可参考：
0. 金融资产组合模型进化全图鉴

1. AI金融智能体

1.1 What is AI金融智能体

AI金融智能体是指利用人工智能技术，特别是机器学习、深度学习和自然语言处理等先进技术，来模拟人类分析师的行为，以执行复杂的金融分析任务的软件。本文主要是通过提示词工程，结合政策信息，给予AI一定的范围权限去影响资产组合权重，属于金融工程构建的尝试性实验。

1.2 Why is AI金融智能体

  • 自动接入全网，实时搜索并分析海量数据：AI金融智能体能够无缝连接互联网，迅速处理和解析大量的金融市场信息，为用户提供即时的洞察和支持。

  • 拥有远超人类的计算能力，揭示潜在的市场盲点：凭借其卓越的计算力，AI智能体可以深入挖掘数据，识别出那些基于常识可能被忽视的市场机会或风险。

  • 顺应技术潮流，推动更多AI工具在金融领域的实际应用：随着技术的进步，AI在金融行业的落地已成为必然趋势。金融机构需积极探索和实施更多的AI解决方案，以保持竞争力和服务创新。

1.3 How to AI金融智能体

  • 参数集设置：

    1. ts.set_token：设置Tushare的API访问令牌
    2. industry：选择目标行业，如"银行"
    3. end_date：回测结束日期，格式为’YYYYMMDD’
    4. years：回测年限，默认5年
    5. risk_free_rate：无风险利率，默认0.03
    6. top_holdings：投资组合持仓数量，默认10只股票
    7. index_code：市场指数代码，默认’000300.SH’
    8. api_key：通义千问API
    9. character：AI人设提示词工程
    10. policy_info：政策信息（5条）

  • 数据准备：

    1. 股票行业数据：通过tushare获取指定行业的股票列表
    2. 历史价格数据：获取指定时间段内的股票日线数据
    3. 市场指数数据：获取指定时间段内的市场指数数据
    4. 因子数据：获取市值(Size)和账面市值比(B/M)数据
    5. 财务数据：获取ROE和资产增长率数据
    6. 无风险利率：设定无风险利率参数

  • 计算流程：

    1. 数据获取：获取股票、市场指数和因子数据
    2. 收益率计算：计算月度对数收益率
    3. 因子构建：构建SMB、HML、RMW和CMA因子
    4. 因子载荷计算：计算每只股票对五个因子的敏感度
    5. FF5预期收益：使用五因子模型计算预期收益率
    6. 组合优化：最大化夏普比率得到最优权重
    7. 持仓筛选：选取权重最大的N只股票并归一化
    8. AI函数：结合权重数据、政策信息、提示词工程，对权重实现智能调整

2. 数据要素&计算流程

2.1 参数集设置

设置模型所需的基本参数，包括数据获取、回测区间和优化约束等。

# 参数集
ts.set_token('token')
pro = ts.pro_api()
industry = '银行'
end_date = '20240101'
years = 5   # 数据时长
risk_free_rate = 0.03  # 无风险利率参数
top_holdings = 10      # 持仓数量参数
index_code = '000300.SH'  # 市场指数代码参数
api_key='sk-api_key'	# 通义千问API

# AI人设提示词工程
character = f'''
你是一名专业的金融数据与政策分析师，擅长解读金融市场动态和政策导向，并据此调整资产组合的权重分布，以优化投资策略。你的主要任务是对给定的资产组合进行权重调整，确保：
1. 权重之和精确为1；
2. 每个资产调整后的权重只能在原有基础上增减最多10%；
3. 每个资产调整完毕后，如果权重之和不等于1，则归一化使权重之和精确为1；
4. 数据对应的日期是{end_date}，在思考过程中，切勿根据该日期之后的信息进行思考。
5. 输出的数据格式需与输入保持一致，仅提供数据而不做额外解释；

当你接收到具体的资产组合及其权重时，请根据最新的金融数据和政策信息对其进行合理调整。
'''

# 通过工作流获取的政策信息
policy_info = '''
| 日期 | 政策简述 |
|------|----------|
| 2023-12-29 | 央行发布《关于优化商业银行存款利率监管有关事项的通知》，取消定期存款利率浮动上限，允许银行自主协调存贷款利率 |
| 2023-11-17 | 央行、银保监会联合发布《关于做好当前商业银行房地产贷款投放管理的通知》，优化房地产信贷政策，支持刚性和改善性住房需求 |
| 2023-09-25 | 银保监会发布《关于进一步加强银行业金融机构流动性风险管理的通知》，要求银行加强流动性风险管理，完善风险监测预警机制 |
| 2023-08-31 | 央行、银保监会宣布下调全国首套住房贷款利率下限，各地可自主决定下调幅度，二套房贷款利率政策与首套相同 |
| 2023-07-21 | 十四届全国人大常委会第四次会议表决通过《中华人民共和国金融稳定法》，建立健全金融风险防范化解制度体系 |
'''

2.2 数据获取&预处理

获取股票、市场指数、因子数据和财务数据，并进行必要的数据清洗和格式转换。

def get_industry_stocks(industry):
    """获取指定行业的股票列表"""
    df = pro.stock_basic(fields=["ts_code", "name", "industry"])
    industry_stocks = df[df["industry"]==industry].copy()
    industry_stocks.sort_values(by='ts_code', inplace=True)
    industry_stocks.reset_index(drop=True, inplace=True)
    return industry_stocks['ts_code'].tolist()

def get_data(code_list, end_date, years):
    """获取指定行业名称的历史收盘价数据"""
    ts_code_list = code_list
    end_date_dt = datetime.strptime(end_date, '%Y%m%d')
    start_date_dt = end_date_dt - timedelta(days=years*365)
    start_date = start_date_dt.strftime('%Y%m%d')
    
    all_data = []
    for stock in ts_code_list:
        df = pro.daily(ts_code=stock, start_date=start_date, end_date=end_date)
        all_data.append(df)
    
    combined_df = pd.concat(all_data).sort_values(by=['ts_code', 'trade_date'])
    combined_df.reset_index(drop=True, inplace=True)
    combined_df.rename(columns={'trade_date': 'date'}, inplace=True)
    
    return combined_df

def get_market_data(index_code='000300.SH', start_date=None, end_date=None):
    """获取市场指数数据用于计算贝塔"""
    df_market = pro.index_daily(ts_code=index_code, 
                              start_date=start_date, 
                              end_date=end_date,
                              fields=['trade_date', 'close'])
    df_market['date'] = pd.to_datetime(df_market['trade_date'])
    df_market.set_index('date', inplace=True)
    df_market = df_market.sort_index()
    
    monthly_last_close = df_market['close'].resample('M').last()
    monthly_log_returns = np.log(monthly_last_close).diff().dropna()
    return monthly_log_returns

def get_factor_data(stock_codes, start_date=None, end_date=None):
    """获取指定股票的因子数据（市值和PB）"""
    all_factor_data = []
    for stock in stock_codes:
        try:
            df = pro.daily_basic(
                ts_code=stock,
                start_date=start_date,
                end_date=end_date,
                fields=['ts_code', 'trade_date', 'total_mv', 'pb']
            )
            all_factor_data.append(df)
        except Exception as e:
            print(f"获取股票 {stock} 的因子数据失败: {str(e)}")
            continue
    
    factor_data = pd.concat(all_factor_data, ignore_index=True)
    factor_data['trade_date'] = pd.to_datetime(factor_data['trade_date'])
    return factor_data

def get_fina_data(stock_codes, start_date=None, end_date=None):
    """获取指定股票的财务指标数据（ROE和资产增长率）"""
    all_fina_data = []
    for stock in stock_codes:
        try:
            df = pro.fina_indicator(
                ts_code=stock,
                start_date=start_date,
                end_date=end_date,
                fields=['ts_code', 'end_date', 'roe_dt', 'assets_yoy', 'update_flag']
            )
            all_fina_data.append(df)
        except Exception as e:
            print(f"获取股票 {stock} 的财务数据失败: {str(e)}")
            continue
    
    # 合并数据
    fina_data = pd.concat(all_fina_data, ignore_index=True)
    
    # 处理update_flag，保留最新数据
    fina_data = (fina_data.groupby(['ts_code', 'end_date'])
                         .agg({'roe_dt': 'first', 
                              'assets_yoy': 'first',
                              'update_flag': 'max'})
                         .reset_index())
    
    # 将end_date转换为datetime
    fina_data['end_date'] = pd.to_datetime(fina_data['end_date'])
    
    # 创建季度到月度的映射
    monthly_data = []
    for _, row in fina_data.iterrows():
        quarter_end = row['end_date']
        if quarter_end.month == 3:  # Q1
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        elif quarter_end.month == 6:  # Q2
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        elif quarter_end.month == 9:  # Q3
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        else:  # Q4
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        
        for month in months:
            monthly_data.append({
                'ts_code': row['ts_code'],
                'trade_date': month,
                'roe_dt': row['roe_dt'],
                'assets_yoy': row['assets_yoy']
            })
    
    monthly_df = pd.DataFrame(monthly_data)
    return monthly_df

2.3 收益率计算

计算月度对数收益率，为后续的因子构建和优化计算做准备。

def calculate_monthly_log_returns(df):
    """计算每月的对数收益率"""
    df['date'] = pd.to_datetime(df['date'])
    monthly_last_close = df.groupby(['ts_code', pd.Grouper(key='date', freq='M')])['close'].last().unstack(level=-1)
    monthly_log_returns = np.log(monthly_last_close).diff().dropna()
    return monthly_log_returns.T

2.4 因子构建与预期收益率计算

构建SMB、HML、RMW和CMA因子，并使用五因子模型计算预期收益率。

def calculate_expected_returns(monthly_log_returns):
    """使用Fama-French五因子模型计算各股票的预期收益率"""
    start_date = monthly_log_returns.index.min().strftime('%Y%m%d')
    end_date = monthly_log_returns.index.max().strftime('%Y%m%d')
    
    # 获取财务数据时，将start_date往前推一个季度，以确保有完整的季度数据
    fina_start_date = (datetime.strptime(start_date, '%Y%m%d') - timedelta(days=90)).strftime('%Y%m%d')
    
    # 获取市场收益率
    market_returns = get_market_data(index_code, start_date, end_date)
    
    # 获取股票的市值和PB数据
    stock_data = get_factor_data(
        monthly_log_returns.columns.tolist(),
        start_date,
        end_date
    )
    
    # 获取财务指标数据，使用提前的start_date
    fina_data = get_fina_data(
        monthly_log_returns.columns.tolist(),
        fina_start_date,
        end_date
    )
    
    # 确保所有数据的日期对齐
    aligned_dates = monthly_log_returns.index.intersection(market_returns.index)
    market_returns = market_returns[aligned_dates]
    stock_returns = monthly_log_returns.loc[aligned_dates].copy()  # 使用copy()避免SettingWithCopyWarning
    
    def calculate_size_factor(date):
        date_data = stock_data[stock_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        median_mv = date_data['total_mv'].median()
        small_returns = stock_returns.loc[date, date_data[date_data['total_mv'] <= median_mv]['ts_code']]
        big_returns = stock_returns.loc[date, date_data[date_data['total_mv'] > median_mv]['ts_code']]
        return small_returns.mean() - big_returns.mean()
    
    def calculate_value_factor(date):
        date_data = stock_data[stock_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        # 创建date_data的副本并计算bm_ratio
        date_data = date_data.copy()
        date_data.loc[:, 'bm_ratio'] = 1 / date_data['pb']
        
        median_bm = date_data['bm_ratio'].median()
        high_returns = stock_returns.loc[date, date_data[date_data['bm_ratio'] > median_bm]['ts_code']]
        low_returns = stock_returns.loc[date, date_data[date_data['bm_ratio'] <= median_bm]['ts_code']]
        return high_returns.mean() - low_returns.mean()
    
    def calculate_profitability_factor(date):
        date_data = fina_data[fina_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        
        median_roe = date_data['roe_dt'].median()
        robust_returns = stock_returns.loc[date, date_data[date_data['roe_dt'] > median_roe]['ts_code']]
        weak_returns = stock_returns.loc[date, date_data[date_data['roe_dt'] <= median_roe]['ts_code']]
        return robust_returns.mean() - weak_returns.mean()
    
    def calculate_investment_factor(date):
        date_data = fina_data[fina_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        
        median_growth = date_data['assets_yoy'].median()
        conservative_returns = stock_returns.loc[date, date_data[date_data['assets_yoy'] <= median_growth]['ts_code']]
        aggressive_returns = stock_returns.loc[date, date_data[date_data['assets_yoy'] > median_growth]['ts_code']]
        return conservative_returns.mean() - aggressive_returns.mean()
    
    # 计算每个月的因子收益
    smb_factor = pd.Series([calculate_size_factor(date) for date in aligned_dates], index=aligned_dates)
    hml_factor = pd.Series([calculate_value_factor(date) for date in aligned_dates], index=aligned_dates)
    rmw_factor = pd.Series([calculate_profitability_factor(date) for date in aligned_dates], index=aligned_dates)
    cma_factor = pd.Series([calculate_investment_factor(date) for date in aligned_dates], index=aligned_dates)
    
    # 使用OLS回归计算每个股票的因子载荷
    factor_loadings = {}
    for stock in stock_returns.columns:
        X = sm.add_constant(pd.concat([
            market_returns - risk_free_rate,
            smb_factor,
            hml_factor,
            rmw_factor,
            cma_factor
        ], axis=1))
        y = stock_returns[stock] - risk_free_rate
        
        model = sm.OLS(y, X).fit()
        factor_loadings[stock] = model.params[1:]
    
    # 计算因子风险溢价
    market_premium = market_returns.mean() - risk_free_rate
    smb_premium = smb_factor.mean()
    hml_premium = hml_factor.mean()
    rmw_premium = rmw_factor.mean()
    cma_premium = cma_factor.mean()
    
    # 使用FF5模型计算预期收益率
    expected_returns = pd.Series({
        stock: (risk_free_rate + 
                loadings.iloc[0] * market_premium +
                loadings.iloc[1] * smb_premium + 
                loadings.iloc[2] * hml_premium +
                loadings.iloc[3] * rmw_premium +
                loadings.iloc[4] * cma_premium)
        for stock, loadings in factor_loadings.items()
    })
    
    return expected_returns

2.5 协方差矩阵计算

计算收益率的协方差矩阵，用于评估资产间的相关性和波动性。

def calculate_covariance_matrix(monthly_log_returns):
    """计算收益率协方差矩阵"""
    return monthly_log_returns.cov()

2.6 投资组合优化

通过最大化夏普比率来寻找最优权重配置。

def max_sharpe_ratio(mean_returns, cov_matrix, risk_free_rate):
    """计算最大夏普比率的投资组合权重"""
    num_assets = len(mean_returns)
    args = (mean_returns, cov_matrix, risk_free_rate)
    constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
    bounds = tuple((0, 1) for asset in range(num_assets))
    result = minimize(negative_sharpe_ratio, num_assets*[1./num_assets], args=args,
                      method='SLSQP', bounds=bounds, constraints=constraints)
    return result.x

2.7 持仓筛选

选取权重最大的N只股票并重新归一化权重。

def calculate_top_holdings_weights(optimal_weights, monthly_log_returns_columns, top_n):
    """计算前N大持仓的权重占比"""
    result_dict = {asset: weight for asset, weight in zip(monthly_log_returns_columns, optimal_weights)}
    top_n_holdings = sorted(result_dict.items(), key=lambda item: item[1], reverse=True)[:top_n]
    top_n_sum = sum(value for _, value in top_n_holdings)
    updated_result = {key: value / top_n_sum for key, value in top_n_holdings}
    return updated_result

2.8 AI金融智能体调仓函数

def get_ai_weights(character, policy_info, updated_result, api_key):
  # 定义发送对话内容
  messages = [
      {'role': 'system', 'content': character},
      {'role': 'user', 'content': policy_info},
      {'role': 'user', 'content': json.dumps(updated_result, ensure_ascii=False)}
      ]
  response = dashscope.Generation.call(
      api_key=api_key,
      model="qwen-max",
      messages=messages,
      result_format='message',
      enable_search=True,
      top_p=0.01
      )

  # 提取content内容
  content = response['output']['choices'][0]['message']['content']

  # 将JSON字符串转换为Python字典
  portfolio_weights = json.loads(content)

  # 将字典中的值修改为6位小数
  portfolio_weights = {k: round(v, 6) for k, v in portfolio_weights.items()}

  return portfolio_weights

3. 汇总代码

以下即为全量代码，修改参数集中内容即可跑出个性化数据。

import tushare as ts
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from scipy.optimize import minimize
import backtrader as bt
import statsmodels.api as sm
import os
import json
import dashscope

# 参数集##############################################################################
ts.set_token('token')
pro = ts.pro_api()
industry = '银行'
end_date = '20240101'
years = 5   # 数据时长
risk_free_rate = 0.03  # 无风险利率参数
top_holdings = 10      # 持仓数量参数
index_code = '000300.SH'  # 市场指数代码参数
api_key='sk-api_key'	# 通义千问API

# AI人设提示词工程
character = f'''
你是一名专业的金融数据与政策分析师，擅长解读金融市场动态和政策导向，并据此调整资产组合的权重分布，以优化投资策略。你的主要任务是对给定的资产组合进行权重调整，确保：
1. 权重之和精确为1；
2. 每个资产调整后的权重只能在原有基础上增减最多10%；
3. 每个资产调整完毕后，如果权重之和不等于1，则归一化使权重之和精确为1；
4. 数据对应的日期是{end_date}，在思考过程中，切勿根据该日期之后的信息进行思考。
5. 输出的数据格式需与输入保持一致，仅提供数据而不做额外解释；

当你接收到具体的资产组合及其权重时，请根据最新的金融数据和政策信息对其进行合理调整。
'''

# 通过工作流获取的政策信息
policy_info = '''
| 日期 | 政策简述 |
|------|----------|
| 2023-12-29 | 央行发布《关于优化商业银行存款利率监管有关事项的通知》，取消定期存款利率浮动上限，允许银行自主协调存贷款利率 |
| 2023-11-17 | 央行、银保监会联合发布《关于做好当前商业银行房地产贷款投放管理的通知》，优化房地产信贷政策，支持刚性和改善性住房需求 |
| 2023-09-25 | 银保监会发布《关于进一步加强银行业金融机构流动性风险管理的通知》，要求银行加强流动性风险管理，完善风险监测预警机制 |
| 2023-08-31 | 央行、银保监会宣布下调全国首套住房贷款利率下限，各地可自主决定下调幅度，二套房贷款利率政策与首套相同 |
| 2023-07-21 | 十四届全国人大常委会第四次会议表决通过《中华人民共和国金融稳定法》，建立健全金融风险防范化解制度体系 |
'''
# 参数集##############################################################################

def get_industry_stocks(industry):
    """获取指定行业的股票列表"""
    df = pro.stock_basic(fields=["ts_code", "name", "industry"])
    industry_stocks = df[df["industry"]==industry].copy()
    industry_stocks.sort_values(by='ts_code', inplace=True)
    industry_stocks.reset_index(drop=True, inplace=True)
    return industry_stocks['ts_code'].tolist()

def get_data(code_list, end_date, years):
    """获取指定行业名称的历史收盘价数据"""
    ts_code_list = code_list
    end_date_dt = datetime.strptime(end_date, '%Y%m%d')
    start_date_dt = end_date_dt - timedelta(days=years*365)
    start_date = start_date_dt.strftime('%Y%m%d')
    
    all_data = []
    for stock in ts_code_list:
        df = pro.daily(ts_code=stock, start_date=start_date, end_date=end_date)
        all_data.append(df)
    
    combined_df = pd.concat(all_data).sort_values(by=['ts_code', 'trade_date'])
    combined_df.reset_index(drop=True, inplace=True)
    combined_df.rename(columns={'trade_date': 'date'}, inplace=True)
    
    return combined_df

def get_market_data(index_code='000300.SH', start_date=None, end_date=None):
    """获取市场指数数据用于计算贝塔"""
    df_market = pro.index_daily(ts_code=index_code, 
                              start_date=start_date, 
                              end_date=end_date,
                              fields=['trade_date', 'close'])
    df_market['date'] = pd.to_datetime(df_market['trade_date'])
    df_market.set_index('date', inplace=True)
    df_market = df_market.sort_index()
    
    monthly_last_close = df_market['close'].resample('M').last()
    monthly_log_returns = np.log(monthly_last_close).diff().dropna()
    return monthly_log_returns

def get_factor_data(stock_codes, start_date=None, end_date=None):
    """获取指定股票的因子数据（市值和PB）"""
    all_factor_data = []
    for stock in stock_codes:
        try:
            df = pro.daily_basic(
                ts_code=stock,
                start_date=start_date,
                end_date=end_date,
                fields=['ts_code', 'trade_date', 'total_mv', 'pb']
            )
            all_factor_data.append(df)
        except Exception as e:
            print(f"获取股票 {stock} 的因子数据失败: {str(e)}")
            continue
    
    factor_data = pd.concat(all_factor_data, ignore_index=True)
    factor_data['trade_date'] = pd.to_datetime(factor_data['trade_date'])
    return factor_data

def get_fina_data(stock_codes, start_date=None, end_date=None):
	"""获取指定股票的财务指标数据（ROE和资产增长率）"""
    all_fina_data = []
    for stock in stock_codes:
        try:
            df = pro.fina_indicator(
                ts_code=stock,
                start_date=start_date,
                end_date=end_date,
                fields=['ts_code', 'end_date', 'roe_dt', 'assets_yoy', 'update_flag']
            )
            all_fina_data.append(df)
        except Exception as e:
            print(f"获取股票 {stock} 的财务数据失败: {str(e)}")
            continue
    
    # 合并数据
    fina_data = pd.concat(all_fina_data, ignore_index=True)
    
    # 处理update_flag，保留最新数据
    fina_data = (fina_data.groupby(['ts_code', 'end_date'])
                         .agg({'roe_dt': 'first', 
                              'assets_yoy': 'first',
                              'update_flag': 'max'})
                         .reset_index())
    
    # 将end_date转换为datetime
    fina_data['end_date'] = pd.to_datetime(fina_data['end_date'])
    
    # 创建季度到月度的映射
    monthly_data = []
    for _, row in fina_data.iterrows():
        quarter_end = row['end_date']
        if quarter_end.month == 3:  # Q1
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        elif quarter_end.month == 6:  # Q2
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        elif quarter_end.month == 9:  # Q3
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        else:  # Q4
            months = [quarter_end + pd.DateOffset(months=i) for i in range(1, 4)]
        
        for month in months:
            monthly_data.append({
                'ts_code': row['ts_code'],
                'trade_date': month,
                'roe_dt': row['roe_dt'],
                'assets_yoy': row['assets_yoy']
            })
    
    monthly_df = pd.DataFrame(monthly_data)
    return monthly_df

def calculate_monthly_log_returns(df):
    """计算每月的对数收益率"""
    df['date'] = pd.to_datetime(df['date'])
    monthly_last_close = df.groupby(['ts_code', pd.Grouper(key='date', freq='M')])['close'].last().unstack(level=-1)
    monthly_log_returns = np.log(monthly_last_close).diff().dropna()
    return monthly_log_returns.T

def calculate_expected_returns(monthly_log_returns):
    """使用Fama-French五因子模型计算各股票的预期收益率"""
    start_date = monthly_log_returns.index.min().strftime('%Y%m%d')
    end_date = monthly_log_returns.index.max().strftime('%Y%m%d')
    
    # 获取财务数据时，将start_date往前推一个季度，以确保有完整的季度数据
    fina_start_date = (datetime.strptime(start_date, '%Y%m%d') - timedelta(days=90)).strftime('%Y%m%d')
    
    # 获取市场收益率
    market_returns = get_market_data(index_code, start_date, end_date)
    
    # 获取股票的市值和PB数据
    stock_data = get_factor_data(
        monthly_log_returns.columns.tolist(),
        start_date,
        end_date
    )
    
    # 获取财务指标数据，使用提前的start_date
    fina_data = get_fina_data(
        monthly_log_returns.columns.tolist(),
        fina_start_date,
        end_date
    )
    
    # 确保所有数据的日期对齐
    aligned_dates = monthly_log_returns.index.intersection(market_returns.index)
    market_returns = market_returns[aligned_dates]
    stock_returns = monthly_log_returns.loc[aligned_dates].copy()  # 使用copy()避免SettingWithCopyWarning
    
    def calculate_size_factor(date):
        date_data = stock_data[stock_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        median_mv = date_data['total_mv'].median()
        small_returns = stock_returns.loc[date, date_data[date_data['total_mv'] <= median_mv]['ts_code']]
        big_returns = stock_returns.loc[date, date_data[date_data['total_mv'] > median_mv]['ts_code']]
        return small_returns.mean() - big_returns.mean()
    
    def calculate_value_factor(date):
        date_data = stock_data[stock_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        # 创建date_data的副本并计算bm_ratio
        date_data = date_data.copy()
        date_data.loc[:, 'bm_ratio'] = 1 / date_data['pb']
        
        median_bm = date_data['bm_ratio'].median()
        high_returns = stock_returns.loc[date, date_data[date_data['bm_ratio'] > median_bm]['ts_code']]
        low_returns = stock_returns.loc[date, date_data[date_data['bm_ratio'] <= median_bm]['ts_code']]
        return high_returns.mean() - low_returns.mean()
    
    def calculate_profitability_factor(date):
        date_data = fina_data[fina_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        
        median_roe = date_data['roe_dt'].median()
        robust_returns = stock_returns.loc[date, date_data[date_data['roe_dt'] > median_roe]['ts_code']]
        weak_returns = stock_returns.loc[date, date_data[date_data['roe_dt'] <= median_roe]['ts_code']]
        return robust_returns.mean() - weak_returns.mean()
    
    def calculate_investment_factor(date):
        date_data = fina_data[fina_data['trade_date'].dt.to_period('M') == date.to_period('M')]
        
        median_growth = date_data['assets_yoy'].median()
        conservative_returns = stock_returns.loc[date, date_data[date_data['assets_yoy'] <= median_growth]['ts_code']]
        aggressive_returns = stock_returns.loc[date, date_data[date_data['assets_yoy'] > median_growth]['ts_code']]
        return conservative_returns.mean() - aggressive_returns.mean()
    
    # 计算每个月的因子收益
    smb_factor = pd.Series([calculate_size_factor(date) for date in aligned_dates], index=aligned_dates)
    hml_factor = pd.Series([calculate_value_factor(date) for date in aligned_dates], index=aligned_dates)
    rmw_factor = pd.Series([calculate_profitability_factor(date) for date in aligned_dates], index=aligned_dates)
    cma_factor = pd.Series([calculate_investment_factor(date) for date in aligned_dates], index=aligned_dates)
    
    # 使用OLS回归计算每个股票的因子载荷
    factor_loadings = {}
    for stock in stock_returns.columns:
        X = sm.add_constant(pd.concat([
            market_returns - risk_free_rate,
            smb_factor,
            hml_factor,
            rmw_factor,
            cma_factor
        ], axis=1))
        y = stock_returns[stock] - risk_free_rate
        
        model = sm.OLS(y, X).fit()
        factor_loadings[stock] = model.params[1:]
    
    # 计算因子风险溢价
    market_premium = market_returns.mean() - risk_free_rate
    smb_premium = smb_factor.mean()
    hml_premium = hml_factor.mean()
    rmw_premium = rmw_factor.mean()
    cma_premium = cma_factor.mean()
    
    # 使用FF5模型计算预期收益率
    expected_returns = pd.Series({
        stock: (risk_free_rate + 
                loadings.iloc[0] * market_premium +
                loadings.iloc[1] * smb_premium + 
                loadings.iloc[2] * hml_premium +
                loadings.iloc[3] * rmw_premium +
                loadings.iloc[4] * cma_premium)
        for stock, loadings in factor_loadings.items()
    })
    
    return expected_returns

def calculate_covariance_matrix(monthly_log_returns):
    """计算收益率协方差矩阵"""
    return monthly_log_returns.cov()

def portfolio_performance(weights, mean_returns, cov_matrix):
    """计算投资组合的表现"""
    returns = np.sum(mean_returns * weights) 
    std_dev = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
    return returns, std_dev

def negative_sharpe_ratio(weights, mean_returns, cov_matrix, risk_free_rate):
    """计算负夏普比率"""
    p_ret, p_std = portfolio_performance(weights, mean_returns, cov_matrix)
    sharpe_ratio = (p_ret - risk_free_rate) / p_std
    return -sharpe_ratio

def max_sharpe_ratio(mean_returns, cov_matrix, risk_free_rate):
    """计算最大夏普比率的投资组合权重"""
    num_assets = len(mean_returns)
    args = (mean_returns, cov_matrix, risk_free_rate)
    constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
    bounds = tuple((0, 1) for asset in range(num_assets))
    result = minimize(negative_sharpe_ratio, num_assets*[1./num_assets], args=args,
                      method='SLSQP', bounds=bounds, constraints=constraints)
    return result.x

def calculate_top_holdings_weights(optimal_weights, monthly_log_returns_columns, top_n):
    """计算前N大持仓的权重占比"""
    result_dict = {asset: weight for asset, weight in zip(monthly_log_returns_columns, optimal_weights)}
    top_n_holdings = sorted(result_dict.items(), key=lambda item: item[1], reverse=True)[:top_n]
    top_n_sum = sum(value for _, value in top_n_holdings)
    updated_result = {key: value / top_n_sum for key, value in top_n_holdings}
    return updated_result

def get_ai_weights(character, policy_info, updated_result, api_key):
  # 定义发送对话内容
  messages = [
      {'role': 'system', 'content': character},
      {'role': 'user', 'content': policy_info},
      {'role': 'user', 'content': json.dumps(updated_result, ensure_ascii=False)}
      ]
  response = dashscope.Generation.call(
      api_key=api_key,
      model="qwen-max",
      messages=messages,
      result_format='message',
      enable_search=True,
      top_p=0.01
      )

  # 提取content内容
  content = response['output']['choices'][0]['message']['content']

  # 将JSON字符串转换为Python字典
  portfolio_weights = json.loads(content)

  # 将字典中的值修改为6位小数
  portfolio_weights = {k: round(v, 6) for k, v in portfolio_weights.items()}

  return portfolio_weights

def main():
    # 获取数据
    code_list = get_industry_stocks(industry)
    df = get_data(code_list, end_date, years)

    # 计算每月的对数收益率
    monthly_log_returns = calculate_monthly_log_returns(df)
    
    # 使用FF5模型计算预期收益率
    mean_returns = calculate_expected_returns(monthly_log_returns)
    
    # 计算收益率协方差矩阵
    cov_matrix = calculate_covariance_matrix(monthly_log_returns)

    # 优化权重
    optimal_weights = max_sharpe_ratio(mean_returns, cov_matrix, risk_free_rate)
    
    # 计算前N大持仓权重
    updated_result = calculate_top_holdings_weights(
        optimal_weights, 
        monthly_log_returns.columns, 
        top_holdings
    )
    
	# 计算AI调仓后的持仓权重
	updated_result = get_ai_weights(character, policy_info, updated_result, api_key)
    
    # 打印更新后的资产占比
    print(f"\n{end_date}最优资产前{top_holdings}占比:")
    print(updated_result)

if __name__ == "__main__":
    main()

运行结果：

AI金融智能体调仓前权重：
在这里插入图片描述

股票代码占比
601398.SH0.16318772568631026
601328.SH0.16177476392789242
600919.SH0.12936894301756055
600036.SH0.10747174637443846
601169.SH0.0958427702817229
600016.SH0.09012906474680284
601166.SH0.08768928377548085
601288.SH0.06538512327994642
600908.SH0.05559150377594274
600926.SH0.043559075133902475

AI金融智能体调仓后权重：

{'601398.SH': 0.179499, '601328.SH': 0.177949, '600919.SH': 0.142309, '600036.SH': 0.118218, 
'601169.SH': 0.105428, '600016.SH': 0.099142, '601166.SH': 0.096456, '601288.SH': 0.071921, 
'600908.SH': 0.061151, '600926.SH': 0.048936}
股票代码占比
601398.SH0.179499
601328.SH0.177949
600919.SH0.142309
600036.SH0.118218
601169.SH0.105428
600016.SH0.099142
601166.SH0.096456
601288.SH0.071921
600908.SH0.061151
600926.SH0.048936

可见，AI金融智能体通过对政策信息的了解，加大对银行业投资的信心，特别是对大型国有银行和部分地方性银行。这种调整反映了AI对于市场动态的理解以及对未来收益预期的优化。
我们后验的经验也证明：在2024年银行行业是一个值得投资的行业，如果在20240101投资银行行业，将会获得不俗的收益。

4. 反思

4.1 不足之处

  1. 政策信息获取：获取政策信息方案仍为半手动
  2. AI逻辑缜密度：AI可能未能完全按照提示词工程执行

4.2 提升思路

  1. 更换AI智能体：使用由幻方量化开发的DeepSeek-V3 模型
  2. 工作流接入金融工程内部，实现真正全自动

5. 启后

  • 优化，下一篇文章将会尝试使用由幻方量化开发的DeepSeek_V3模型：，可参考下一篇文章：
    pass

  • 量化回测实现，可参考下一篇文章：
    pass

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2280407.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

PostMan最新版本及离线安装指南

PostMan最新版本及离线安装指南

本文还有配套的精品资源&#xff0c;点击获取 简介&#xff1a;PostMan是一款流行的API测试工具&#xff0c;它提供了一个直观的用户界面&#xff0c;方便Web开发者和测试人员进行接口测试。本文将指导你如何安装最新版的PostMan&#xff0c;包括在线安装和离线安装两种方法。…
阅读更多...
记录一次k8s起不来的排查过程

记录一次k8s起不来的排查过程

我在k8s集群&#xff0c;重启了一个node宿主机&#xff0c;竟然发现kubelet起不来了&#xff01;报错如下 这个报错很模糊&#xff0c;怎么排查呢。这样&#xff0c;开两个界面&#xff0c;一个重启kubelet&#xff0c;一个看系统日志(/var/log/message:centos&#xff0c;/va…
阅读更多...
grafana + Prometheus + node_exporter搭建监控大屏

grafana + Prometheus + node_exporter搭建监控大屏

本文介绍生产系统监控大屏的搭建&#xff0c;比较实用也是实际应用比较多的方式&#xff0c;希望能够帮助大家对监控系统有一定的认识。 0、规划 grafana主要是展示和报警&#xff0c;Prometheus用于保存监控数据&#xff0c;node_exporter用于实时采集各个应用服务器的事实状…
阅读更多...
2024年博客之星主题创作｜从零到一：我的技术成长与创作之路

2024年博客之星主题创作｜从零到一：我的技术成长与创作之路

2024年博客之星主题创作&#xff5c;从零到一&#xff1a;我的技术成长与创作之路 个人简介个人主页个人成就热门专栏 历程回顾初来CSDN&#xff1a;怀揣憧憬&#xff0c;开启创作之旅成长之路&#xff1a;从平凡到榜一的蜕变持续分享&#xff1a;打卡基地与成长复盘四年历程&a…
阅读更多...
Golang的网络编程安全

Golang的网络编程安全

Golang的网络编程安全 一、Golang网络编程的基本概念 作为一种现代化的编程语言&#xff0c;具有优秀的并发特性和网络编程能力。在Golang中&#xff0c;网络编程是非常常见的需求&#xff0c;可以用于开发各种类型的网络应用&#xff0c;比如Web服务、API服务、消息队列等。Go…
阅读更多...
【2024年华为OD机试】(C/D卷,200分)- 5G网络建设 （JavaScriptJava PythonC/C++）

【2024年华为OD机试】(C/D卷,200分)- 5G网络建设 （JavaScriptJava PythonC/C++）

一、问题描述 题目描述 现需要在某城市进行5G网络建设&#xff0c;已经选取N个地点设置5G基站&#xff0c;编号固定为1到N。接下来需要各个基站之间使用光纤进行连接以确保基站能互联互通。不同基站之间假设光纤的成本各不相同&#xff0c;且有些节点之间已经存在光纤相连。 …
阅读更多...
消息队列篇--原理篇--RabbitMQ和Kafka对比分析

消息队列篇--原理篇--RabbitMQ和Kafka对比分析

RabbitMQ和Kafka是两种非常流行的消息队列系统&#xff0c;但它们的设计哲学、架构特点和适用场景存在显著差异。对比如下。 1、架构设计 RabbitMQ&#xff1a; 基AMQP协议&#xff1a;RabbitMQ是基于AMQP&#xff08;高级消息队列协议&#xff09;构建的&#xff0c;支持多…
阅读更多...
玻璃样式的登录界面

玻璃样式的登录界面

AI越来越火了，我们想要不被淘汰就得主动拥抱。推荐一个人工智能学习网站，通俗易懂，风趣幽默，最重要的屌图甚多，忍不住分享一下给大家。点击跳转到网站 先看样式： 源码： <div class="wrapper">
阅读更多...
Python数据可视化（够用版）：懂基础 + 专业的图表抛给Tableau等专业绘图工具

Python数据可视化（够用版）：懂基础 + 专业的图表抛给Tableau等专业绘图工具

我先说说文章标题中的“够用版”啥意思&#xff0c;为什么这么写。 按照我个人观点&#xff0c;在使用Python进行数据分析时&#xff0c;我们有时候肯定要结合到图表去进行分析&#xff0c;去直观展现数据的规律和特定&#xff0c;那么我们肯定要做一些简单的可视化&#xff0…
阅读更多...
物联网网关Web服务器--CGI开发实例BMI计算

物联网网关Web服务器--CGI开发实例BMI计算

本例子通一个计算体重指数的程序来演示Web服务器CGI开发。 硬件环境&#xff1a;飞腾派开发板&#xff08;国产E2000处理器&#xff09; 软件环境&#xff1a;飞腾派OS&#xff08;Phytium Pi OS&#xff09; 硬件平台参考另一篇博客&#xff1a;国产化ARM平台-飞腾派开发板…
阅读更多...
HTML新春烟花

HTML新春烟花

系列文章 序号目录1HTML满屏跳动的爱心&#xff08;可写字&#xff09;2HTML五彩缤纷的爱心3HTML满屏漂浮爱心4HTML情人节快乐5HTML蓝色爱心射线6HTML跳动的爱心&#xff08;简易版&#xff09;7HTML粒子爱心8HTML蓝色动态爱心9HTML跳动的爱心&#xff08;双心版&#xff09;1…
阅读更多...
从结构嵌套的幻梦里：递归与数据构建的精巧和鸣

从结构嵌套的幻梦里：递归与数据构建的精巧和鸣

大家好啊&#xff0c;我是小象٩(๑ω๑)۶ 我的博客&#xff1a;Xiao Xiangζั͡ޓއއ 很高兴见到大家&#xff0c;希望能够和大家一起交流学习&#xff0c;共同进步。 这一节我们来学习递归的相关知识 函数递归 一、什么是递归1.1 递归的思想 二、递归的限制条件三、递归的…
阅读更多...
【Linux系统】—— 编译器 gcc/g++ 的使用

【Linux系统】—— 编译器 gcc/g++ 的使用

【Linux系统】—— 编译器 gcc/g 的使用 1 用 gcc 直接编译2 翻译环境2.1 预处理&#xff08;进行宏替换&#xff09;2.2 编译&#xff08;生成汇编&#xff09;2.3 汇编&#xff08;生成机器可识别代码&#xff09;2.4 链接2.5 记忆小技巧2.6 编译方式2.7 几个问题2.7.1 如何理…
阅读更多...
【Unity3D】3D物体摆放、场景优化案例Demo

【Unity3D】3D物体摆放、场景优化案例Demo

目录 PlaceManager.cs(放置管理类) Ground.cs(地板类) 和 GroundData.cs(地板数据类) 额外知识点说明 1、MeshFilter和MeshRenderer的Bounds区别 2、Gizmos 绘制一个平行于斜面的立方体 通过网盘分享的文件&#xff1a;PlaceGameDemo2.unitypackage 链接: https://pan.baid…
阅读更多...
智能系统的感知和决策

智能系统的感知和决策

智能系统在感知和决策过程中具备的关键能力表现在智能感知/自主判定上&#xff0c;下面可以从感知的本质、自主判断的含义及其在智能系统中的作用进行深入分析。 1、智能感知&#xff1a;信息获取与理解 智能感知是指智能系统通过传感器或其他数据采集手段获取环境中的信息&…
阅读更多...
Spring 中的事件驱动模型

Spring 中的事件驱动模型

事件驱动的基本了解 事件模式也就是观察者模式&#xff0c;当一个对象改变的时候&#xff0c;所有依赖的对象都会收到一个通知。 Subject&#xff1a;抽象主题 Observer&#xff1a;具体主题 Concrete Subject&#xff1a;抽象观察者&#xff0c;在得到更新通知之后去更新自…
阅读更多...
linux-FTP服务配置与应用

linux-FTP服务配置与应用

也许你对FTP不陌生&#xff0c;但是你是否了解FTP到底是个什么玩意&#xff1f; FTP 是File Transfer Protocol&#xff08;文件传输协议&#xff09;的英文简称&#xff0c;而中文简称为 “文传协议” 用于Internet上的控制文件的双向传输。同时&#xff0c;它也是一个应用程序…
阅读更多...
linux-NFS网络共享存储服务配置

linux-NFS网络共享存储服务配置

1.NFS服务原理 NFS会经常用到&#xff0c;用于在网络上共享存储&#xff0c;这样讲&#xff0c;你对NFS可能不太了解&#xff0c;举一个例子&#xff0c; 加入有三台机器A,B,C&#xff0c;它们需要访问同一个目录&#xff0c;目录中都是图片&#xff0c;传统的做法是把这些 图…
阅读更多...
Jenkins 启动

Jenkins 启动

废话 这一阵子感觉空虚&#xff0c;心里空捞捞的&#xff0c;总想找点事情做&#xff0c;即使这是一件微小的事情&#xff0c;空余时间除了骑车、打球&#xff0c;偶尔朋友聚会 … 还能干什么呢&#xff1f; 当独自一人时&#xff0c;究竟可以做点什么&#xff0c;填补这空虚…
阅读更多...
消息队列篇--原理篇--Pulsar（Namespace，BookKeeper，类似Kafka甚至更好的消息队列）

消息队列篇--原理篇--Pulsar（Namespace，BookKeeper，类似Kafka甚至更好的消息队列）

Apache Pulusar是一个分布式、多租户、高性能的发布/订阅&#xff08;Pub/Sub&#xff09;消息系统&#xff0c;最初由Yahoo开发并开源。它结合了Kafka和传统消息队列的优点&#xff0c;提供高吞吐量、低延迟、强一致性和可扩展的消息传递能力&#xff0c;适用于大规模分布式系…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023