当前位置: 首页 > article >正文

6. 马科维茨资产组合模型+政策意图AI金融智能体(DeepSeek-V3)增强方案(理论+Python实战)

目录

    • 0. 承前
    • 1. 幻方量化 & DeepSeek
      • 1.1 What is 幻方量化
      • 1.2 What is DeepSeek
    • 2. 重写AI金融智能体函数
    • 3. 汇总代码
    • 4. 反思
      • 4.1 不足之处
      • 4.2 提升思路
    • 5. 启后

0. 承前

本篇博文是对上一篇文章,链接:
5. 马科维茨资产组合模型+政策意图AI金融智能体(Qwen-Max)增强方案(理论+Python实战)
AI金融智能体更改为幻方量化DeepSeek-V3的尝试。

本文与上一篇文章唯一区别之处在于上文中的get_ai_weights函数,如果需要整体金融工程的思路流程,可以在上文直接跳转,本文中会直接给出修改地方,与能够直接运行的汇总代码。

本文主旨:

  • 用文章向大家展示低耦合开发的优点,如本文与上一篇文章的转换只需要重写一个函数即可;
  • 使用实际代码,向大家展示最常用的AI金融智能体接口:openai、dashcode;

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

1. 幻方量化 & DeepSeek

1.1 What is 幻方量化

幻方量化(High-Flyer Quant)是中国领先的量化投资机构之一,成立于2008年,专注于利用人工智能(AI)和机器学习技术进行量化投资。其名称“幻方”源自古代数学中的幻方矩阵,象征着数学与科学的逻辑魅力。幻方量化的核心业务是通过大数据分析、深度学习模型和强大的计算能力,开发量化交易策略,为投资者提供资产管理服务。

1.2 What is DeepSeek

  • 低成本训练:DeepSeek通过创新的算法和硬件优化,大幅降低了模型训练成本。例如,其最新模型DeepSeek-V3的训练成本仅为557万美元,远低于OpenAI等巨头的数十亿美元投入。

  • 创新架构:DeepSeek采用了MLA(多头潜在注意力机制)和DeepSeekMoE(混合专家模型)等创新架构,显著提升了模型的训练效率和推理性能。

  • 高性能表现:DeepSeek-V3在多项基准测试中表现优异,尤其在数学推理、代码生成和中文处理能力上超越了众多开源和闭源模型,甚至与GPT-4等顶尖模型媲美。

2. 重写AI金融智能体函数

登录DeepSeek官网获取 api,并使用OpenAI格式的对话函数。代码实现:

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)}
      ]
  
  client = OpenAI(api_key=api_key, base_url="https://api.deepseek.com")

  response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    stream=False
  )

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

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

  # 对AI输出结果进行归一化
  weights_sum = sum(portfolio_weights.values())
  portfolio_weights = {key: value/weights_sum for key, value in portfolio_weights.items()}

  # 将字典中的值修改为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'	# DeepSeek-V3 API

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)}
      ]
  
  client = OpenAI(api_key=api_key, base_url="https://api.deepseek.com")

  response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    stream=False
  )

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

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

  # 对AI输出结果进行归一化
  weights_sum = sum(portfolio_weights.values())
  portfolio_weights = {key: value/weights_sum for key, value in portfolio_weights.items()}

  # 将字典中的值修改为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金融智能(DeepSeek-V3)体调仓后权重:

{'601398.SH': 0.166525, '601328.SH': 0.165165, '600919.SH': 0.134011, '600036.SH': 0.112955, '601169.SH': 0.092157, 
'600016.SH': 0.086663, '601166.SH': 0.084316, '601288.SH': 0.062868, '600908.SH': 0.053454, '600926.SH': 0.041885}

与Qwen-Max进行对比:

股票代码股票占比(Qwen-Max)股票占比(DeepSeek-V3)
601398.SH0.1630250.166525
601328.SH0.1616230.165165
600919.SH0.1292520.134011
600036.SH0.1073720.112955
601169.SH0.0957640.092157
600016.SH0.0900460.086663
601166.SH0.0876060.084316
601288.SH0.0653230.062868
600908.SH0.0555410.053454
600926.SH0.0444490.041885

DeepSeek-V3相较于Qwen-Max,通过保守的风格调整资产权重,降低高波动性资产的占比,保持重点资产的较高权重,展现出较低的风险偏好和对稳健收益的追求。它注重分散化投资和长期价值,旨在实现投资组合的稳定增长,适合寻求风险控制与稳定回报的投资者。

其他尝试:或许我们能够直接在人设提示词工程,让AI金融智能体更加趋向于高波动&高收益。

4. 反思

4.1 不足之处

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

4.2 提升思路

  1. 更换提示词工程
  2. 工作流接入金融工程内部,实现真正全自动

5. 启后

  • 优化,使用深度研报,增加AI获取数据量:,可参考下一篇文章:
    7. 马科维茨资产组合模型+金融研报AI长文本智能体(Qwen-Long)增强方案(理论+Python实战)

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


http://www.kler.cn/a/517064.html

相关文章:

  • 数据库性能优化(sql优化)_SQL执行计划02_yxy
  • linux系统中的 scp的使用方法
  • HarmonyOS Next构建工具 lycium 原理介绍
  • 差分进化算法 (Differential Evolution) 算法详解及案例分析
  • 从零到一:Spring Boot 与 RocketMQ 的完美集成指南
  • jenkins-k8s pod方式动态生成slave节点
  • (详细)Springboot 整合动态多数据源 这里有mysql(分为master 和 slave) 和oracle,根据不同路径适配不同数据源
  • Redis线上阻塞要如何排查
  • Java面向对象专题
  • 【leetcode100】二叉搜索树中第k小的元素
  • python远程获取数据库中的相关数据并存储至json文件
  • MySQL中的关联查询:方式、区别及示例
  • Python 爬虫——爬取Web页面图片
  • 03垃圾回收篇(D3_垃圾收集器的选择及相关参数)
  • 2K高刷电竞显示器怎么选?
  • 记忆层增强的 Transformer 架构:通过可训练键值存储提升 LLM 性能的创新方法
  • Django 静态文件配置实战指南
  • <keep-alive> <component ></component> </keep-alive>缓存的组件实现组件,实现组件切换时每次都执行指定方法
  • 详解Redis的Zset类型及相关命令
  • AviatorScript用法
  • 解决docker: ‘buildx‘ is not a docker command.
  • Golang初识
  • vue3中为什么引入setup,引入setup是为了解决什么问题,setup的执行时机是什么?返回值是什么
  • linux数据压缩
  • 14-6-1C++的list
  • Elixir语言的数据结构