2383 动态股债平衡多因子轮动模拟盘策略 年化39% 量化交易策略 40259

1. 牛熊市判断体系：

• 使用中小 300 指数（399008.XSHE）10 日均线作为趋势判断基准
• 当前价突破均线 1.003 倍判定为牛市，低于均线判定为熊市
• 牛市满仓股票（100% 仓位），熊市股债平衡（50% 股票 + 50% 债券）

2. 多因子选股框架：

• 外部信号驱动：每日读取 CSV 文件获取多因子选股结果
• 最大持仓 5 只股票，等权重分配可用资金
• 强制冷却机制：卖出股票后 5 天内禁止再次买入
• 持仓时间约束：买入后至少持有 10 个交易日

3. 动态风控机制：

• 强制止盈：持仓个股从最高点回撤 3% 立即平仓
• 到期清仓：持仓满 10 个交易日后自动卖出
• 债券对冲：熊市阶段配置 50% 国债 ETF（511010）对冲风险
• 双向平衡：牛熊转换时自动完成股债比例再平衡

4. 交易执行优化：

• 尾盘集中交易：14:55 执行调仓减少日内波动影响
• 滑点控制：设置 1% 的滑点缓冲冲击成本
• 黑名单管理：记录卖出历史避免短期重复交易
• 名称映射优化：预加载全市场股票名称提升执行效率

# 导入函数库
from jqdata import *
from jqfactor import *
import pandas as pd
import numpy as np
from jqdata import finance
import math
import time
from six import BytesIO

# 初始化函数，设定基准等等
def initialize(context):
    # 最大持仓数量
    g.max_stock_count = 5
    
    # 牛熊函数对应参数
    g.MA = ['399008.XSHE', 10] # 均线择时，中小300指数
    g.threshold = 0.003 # 牛熊切换阈值
    g.isbull = False # 是否牛市
    g.bearpercent = 0.5 # 熊市仓位 original=30%
    

    # 买入后，最短持仓时间，才能卖出
    g.hold_interval = 10
    # 距离最近一次卖出时间大于N天，才能再次买入
    g.selldate_interval = 5
    # 最高点回撤5%离场
    g.top_withdraw_ratio =0.03
    
    
    # 要操作的债券
    g.bond = '511010.XSHG'

    # 用这个DF做统计
    g.statistics_df= pd.DataFrame(columns=['code','name','date_buy','price_buy','date_sell','price_sell','ratio','result'])
    
    # 用这个df做卖出的历史记录，防止短期内再买入
    g.sell_history_df= pd.DataFrame(columns=['code','name','last_date_sell'])
    
    #翻译中文名称用，一次性读到内存里面，提高效率 
    g.stocks_allnames_df = get_all_securities()
    
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    set_option("avoid_future_data", True)
    
    # 输出内容到日志 log.info()
    log.info('初始函数开始运行且全局只运行一次')
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')

    ### 股票相关设定 ###
   
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
   ## 运行函数（reference_security为运行时间的参考标的；传入的标的只做种类区分，因此传入'000300.XSHG'或'510300.XSHG'是一样的）

      # 开盘前运行
    run_daily(before_market_open, time='before_open', reference_security='000300.XSHG')
      # 开盘时运行
    run_daily(market_open, time='14:55', reference_security='000300.XSHG')
      # 收盘后运行
    run_daily(after_market_close, time='after_close', reference_security='000300.XSHG')

## 开盘前运行函数
def before_market_open(context):
    # 输出运行时间
    # log.info('函数运行时间(before_market_open)：'+str(context.current_dt.time()))
    # 占位，省了报错
    test = 0

    

## 开盘时运行函数
def market_open(context):
    # log.info('函数运行时间(market_open):'+str(context.current_dt.time()))
    date = context.current_dt.strftime("%Y-%m-%d")
    # 判断牛熊趋势
    get_bull_bear_signal_minute()

    # 建仓程序
    buy_df = get_df_fromfile(context)
    buylist = buy_df['code'].tolist()
    if len(buylist) == 0:
        log.info('当日无建仓信号')
    else:
        # print(buylist, type(buylist))
        log.info('今日建仓列表:'+ str(buy_df))
        # 调仓
        adjust_position(context, buylist)


## 收盘后运行函数
def after_market_close(context):
    # log.info(str('函数运行时间(after_market_close):'+str(context.current_dt.time())))
    #得到当天所有成交记录
    # trades = get_trades()
    # for _trade in trades.values():
    #     log.info('成交记录：'+str(_trade))
    log.info('############################一天结束###############################')
    

def on_strategy_end(context):
    date = context.current_dt.strftime("%Y-%m-%d")
    log.info('策略执行结束总资产 = ' + str(context.portfolio.total_value))

    # ******************************************************************************# 
    # ****************************核心函数都在下面**********************************# 
    # ******************************************************************************# 
# 函数2：获取当天的买入信号CSV文件，读取到df中，返回待买入股票列表的df
def get_df_fromfile(context):
    date = context.current_dt.strftime("%Y-%m-%d")
    file_name = '8.Mutifactors/Signal_for_trade_' + date + '.csv'
    log.info('读取买入信号文件：'+ str(file_name))
    result_df=pd.read_csv(BytesIO(read_file(file_name)))
        
    return result_df

关键函数解锁后查看：

# 函数，检验某只股票距离上次卖出时间是否大于N天  
# selldate_interval
def check_selldate_interval(date,code):
    if code in g.sell_history_df.values:
        last_date_sell = g.sell_history_df.loc[(g.sell_history_df.code == code)].last_date_sell.values
        print(g.sell_history_df)
        print('last_date_sell:',last_date_sell)
        start =str(last_date_sell)
        end =date
        start = time.mktime(time.strptime(start,'[%Y-%m-%d]'))
        end = time.mktime(time.strptime(end,'%Y-%m-%d'))
        count_days = int((end - start)/(24*60*60))
        if count_days < g.selldate_interval:
            log.info('距离最近卖出时间为：'+ str(count_days) +'天,无法再次购买:'+ str(code))
            return False
        else:
            log.info('距离最近卖出时间为：'+ str(count_days) +'天,可以再次购买:'+ str(code))
            return True
    
    else:
        log.info('卖出历史中无记录，可以建仓:'+ str(code))
        return True

# 函数，检验某只股票买入后持有时间是否足够  
# g.hold_interval = 5
def check_hold_interval(context,code,hold_interval):
    # 此股票的建仓时间
    init_time = context.portfolio.positions[code].init_time
    # date
    current_date = context.current_dt
    start = init_time
    end = current_date
    cname = get_stock_name(code)
    # print('start:',start,type(start))
    # print('end:',end,type(end))
    count_days = (end - start).days
    # print('count_days:',count_days,type(count_days))
    if count_days > hold_interval:
        log.info('持仓时间为：' + str(count_days)+ '天，可以卖出，code：' + str(code) + str(cname))
        return True
    else:
        log.info('持仓时间仅为：' + str(count_days)+ '天，拒绝卖出，code：' + str(code) + str(cname))
        return False



# 获取前n个单位时间当时的收盘价
def get_close_price(security, n, unit='1d'):
    return attribute_history(security, n, unit, 'close')['close'][0]            
# 恢复为原版分钟级数据，否则会得到相反的结论！
def get_bull_bear_signal_minute():
    # 中小300指数的最近1分钟数据
    nowindex = get_close_price(g.MA[0], 1, '1m')
    # 中小300指数的（过去10日收盘价+当前最新价格）的均值
    MAold = (attribute_history(g.MA[0], g.MA[1] - 1, '1d', 'close', True)['close'].sum() + nowindex) / g.MA[1]
    if g.isbull:
        if nowindex * (1 + g.threshold) <= MAold:
            g.isbull = False
    else:
        # 当前指数价格高于历史均值比例超过阈值时，则为牛市
        if nowindex > MAold * (1 + g.threshold):
            g.isbull = True
            
# 返回单只股票的中文名称，为了提高效率，改为一开始做一个g变量
def get_stock_name(stock_code):
    if g.stocks_allnames_df.loc[g.stocks_allnames_df.index == stock_code].empty:
        stock_name = '名称未知'
    else:
        stock_name = g.stocks_allnames_df.loc[g.stocks_allnames_df.index == stock_code]['display_name'][0]
    return stock_name  
    
    


# # 基于指数的动量趋势判断牛熊方向，而非历史分位
# def get_bull_bear_signal_minute():
#     # 中小300指数的最近1分钟数据
#     nowindex = attribute_history(g.MA[0], 1, '1m', 'close')['close'][0]
#     # 中小300指数的（过去10日收盘价+当前最新价格）的均值
#     MAold = (attribute_history(g.MA[0], g.MA[1] - 1, '1d', 'close', True)['close'].sum() + nowindex) / g.MA[1]
#     if g.isbull:
#         if nowindex * (1 + g.threshold) <= MAold:
#             log.info('今日趋势转熊')
#             g.isbull = False
#     else:
#         # 当前指数价格高于历史均值比例超过阈值时，则为牛市
#         if nowindex > MAold * (1 + g.threshold):
#             log.info('今日趋势转牛')
#             g.isbull = True

2025-02-23