2332 多因子选股策略，非线性关系市值（不是小市值）4只绩优股组合 自动化量化交易策略代码

策略名称：多因子量化选股策略

策略概述：

该策略通过结合财务指标、动量因子和市值因子，筛选出具有良好基本面、成长性和流动性的股票。策略的核心逻辑是通过多因子评分模型对股票进行综合评分，并根据评分结果动态调整持仓，以实现长期稳定的超额收益。

---

策略核心逻辑：

1. 财务指标筛选：

   • 使用市盈率（PE）、市净率（PB）、股息率、营业收入同比增长率、净利润同比增长率等财务指标，筛选出基本面良好的股票。

   • 剔除ST股票、停牌股票、涨停股票、跌停股票等不符合条件的股票。

2. 多因子评分模型：

   • 结合当前价格、成交量、涨幅、流通市值、总市值等因子，计算每只股票的综合评分。

   • 根据评分对股票进行排序，选择评分最高的股票作为投资标的。

3. 动态调仓：

   • 每月初根据最新的财务数据和市场表现，重新筛选股票并调整持仓。

   • 根据评分结果，卖出低评分股票，买入高评分股票。

4. 风险控制：

   • 通过过滤ST股票、停牌股票、涨停股票、跌停股票等，避免投资风险。

   • 设置最大持仓比例，避免单只股票仓位过重。

---

策略特点：

• 多因子结合：通过财务指标、动量因子和市值因子的结合，筛选出具有良好基本面和成长性的股票。

• 动态调仓：每月初根据最新的市场数据和财务数据，动态调整持仓，确保投资组合的优化。

• 风险控制：通过严格的股票过滤和仓位控制，降低投资风险。

• 自动化交易：策略实现全自动化交易，减少人为干预。

---

策略参数：

参数名称	参数值	说明
最大持仓数	4	每次调仓时持有的股票数量
股票评分阈值	7	股票评分的最低阈值
市值筛选百分比	70%	筛选市值前70%的股票
成交量计算天数	5	计算成交量的周期
涨幅计算天数	60	计算涨幅的周期
评分权重	[2,1,1,4,4]	[当前价格，成交量，涨幅，流通市值，总市值]

---

策略流程：

1. 初始化：

   • 设置基准指数（沪深300）。

   • 定义财务指标筛选条件。

   • 初始化多因子评分模型的参数。

2. 每月调仓：

   • 使用财务指标筛选出符合条件的股票。

   • 根据多因子评分模型对股票进行评分和排序。

   • 根据评分结果，卖出低评分股票，买入高评分股票。

3. 多因子评分模型：

   • 计算每只股票的当前价格、成交量、涨幅、流通市值、总市值等因子。

   • 根据评分权重计算每只股票的综合评分。

   • 根据评分对股票进行排序，选择评分最高的股票。

4. 动态调仓：

   • 卖出当前持仓中评分较低的股票。

   • 买入评分最高的股票，确保持仓数量不超过最大持仓数。

---

策略风险：

1. 市场风险：

   • 如果市场整体下跌，多因子模型可能失效，导致策略表现不佳。

2. 财务数据滞后：

   • 财务数据的更新可能存在滞后，导致策略的选股结果不准确。

3. 流动性风险：

   • 部分股票的流动性不足可能导致买卖价差较大，影响策略收益。

4. 参数敏感性：

   • 策略对评分权重、市值筛选百分比等参数较为敏感，参数设置不当可能导致策略失效。

---

策略优化方向：

1. 多因子优化：

   • 引入更多因子（如波动率、动量、质量等）优化选股模型。

2. 动态参数调整：

   • 根据市场环境动态调整评分权重和市值筛选百分比。

3. 风险控制增强：

   • 加入止损机制，进一步控制回撤。

4. 行业轮动：

   • 结合行业轮动策略，优化资产配置。

---

策略适用场景：

• 市场环境：适用于市场风格偏向价值投资、成长性股票具备反弹潜力的环境。

• 投资者类型：适合风险承受能力中等、追求长期稳定收益的投资者。

---

策略回测表现：

• 年化收益率：根据历史回测数据，策略年化收益率较高（具体数值需通过回测验证）。

• 最大回撤：通过多因子评分模型和动态调仓，策略能够有效控制回撤。

---

总结：

“多因子量化选股策略”通过结合财务指标、动量因子和市值因子，实现了对股票的动态筛选和优化配置。策略具备较强的自动化能力和风险控制机制，适合在价值投资和成长性股票表现良好的市场环境中运行。通过进一步优化和扩展，策略有望在更多市场环境中表现出色。

策略代码

# 标题：非线性市值（非小市值）组合4只

import math
from jqdata import *
from pandas.core.frame import DataFrame
def initialize(context):
    # 设置系统参数
    set_option('use_real_price', True)    # 用真实价格交易
    set_slippage(PriceRelatedSlippage(0.00))  # 设定滑点为百分比滑点
    # 设置交易成本
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5),type='stock')
    # 初始化全局变量
    g.chosen_stock_list = []  # 选出的股票列表
    g.sold_stock = {}         # 已卖出股票列表
    g.buy_stock_count = 4    # 购买股票数量
#    g.increase1d = 0.06       # 1日涨幅限制
#    g.tradeday = 12          # 上市天数限制
#    g.buyagain = 31            # 再次购买间隔天数
    g.score = 7               # 股票评分
    g.buyrank = g.buy_stock_count * 2   # 输出可买入列表的个数
    g.sellrank = g.buy_stock_count * 2  # 筛选时保留的股票个数
    g.stock_selection_percent = 0.7     # 设置选取市值最大股票的百分比，1为全部。
    g.volume_days = 5                   # 成交量天数
    g.increase_days = 60                # 涨幅天数
#    g.score_weights = [10,9,1,10,10]# [当前价格，成交量，涨幅天数，流通市值，总市值]
    g.score_weights = [2,1,1,4,4]# [当前价格，成交量，涨幅天数，流通市值，总市值]
    # 设置定时任务
    run_monthly(before_trading,1,'09:29')  # 每月第一个交易日运行
    run_daily(mysell,'09:30')
    run_daily(mybuy,'09:31')
    # 设置日志级别
    log.set_level('order', 'error')   # 过滤order中低于error级别的日志
    log.set_level('system', 'error')
    log.set_level('history', 'error')

def select_top_percent_stocks(df, percent):
    """
    选择前N%的股票
    :param df: DataFrame，包含股票数据的数据帧
    :param percent: float，要选择的股票百分比，范围为0到1
    :return: DataFrame，包含所选股票的数据帧
    """
    top_percent = int(len(df) * percent)  # 计算前N%的股票数量
    return df.head(top_percent)  # 返回前N%的股票

# 每月运行，筛选股票
def before_trading(context):
#    temp = g.sold_stock  # 临时存储已卖出股票
#    g.sold_stock = {}  # 清空已卖出股票列表
#    for stock in temp.keys():# 判断已经卖出的股票是否达到了再次购买的条件
#        if temp[stock] >= g.buyagain - 1:
#            pass
#        else:
#            g.sold_stock[stock] = temp[stock] + 1
    g.chosen_stock_list = get_stock_list(context)  # 获取筛选后的股票列表

# 获取和筛选股票列表
def get_stock_list(context):
    # 使用 query 函数从 valuation 表中获取股票代码，并添加过滤条件
    q = query(
        valuation.code,  # 股票代码
        valuation.pe_ratio,  # 市盈率
        valuation.pb_ratio,  # 市净率
        indicator.inc_return,  # 股息率
        indicator.inc_total_revenue_year_on_year,  # 营业收入同比增长率
        indicator.inc_net_profit_year_on_year,  # 净利润同比增长率
        valuation.market_cap   # 市值
    ).filter(
        valuation.pe_ratio > 0,  # 过滤市盈率大于0的股票
        valuation.pb_ratio > 0,  # 过滤市净率大于0的股票
        indicator.inc_return > 0,  # 过滤股息率大于0的股票
        indicator.inc_total_revenue_year_on_year > 0,  # 过滤营业收入同比增长率大于0的股票
        indicator.inc_net_profit_year_on_year > 0  # 过滤净利润同比增长率大于0的股票
    )
    # 执行查询并获取数据
    df = get_fundamentals(q)
    # 将查询结果转换为 DataFrame
    df = pd.DataFrame(df)
    # 删除包含 NaN 值的行
    df = df.dropna()
    # 按市值从大到小排序
    df = df.sort_values(by='market_cap', ascending=False)  
    print('本月股票总数: %s' % len(df))  # 输出符合条件的股票总数
    df = select_top_percent_stocks(df, g.stock_selection_percent)  # 调用全局函数，选取前N%的股票
    print('本月选中股票总数: {}% ({})'.format(g.stock_selection_percent * 100, len(df)))  # 输出选取的股票总数
    stock_list = list(df['code'])      # 将股票代码转换为列表
    # 过滤股票
    stock_list = filter_st_stock(stock_list)  # 过滤ST股票
    stock_list = filter_paused_stock(stock_list)  # 过滤停牌股票
    # stock_list = filter_new_stock(context, stock_list)  # 过滤新股（注释掉）
    stock_list = filter_limitup_stock(context, stock_list)  # 过滤涨停股票
    stock_list = filter_limitdown_stock(context, stock_list)  # 过滤跌停股票
    # stock_list = filter_increase1d(stock_list)  # 过滤连续涨停股票（注释掉）
    # stock_list = filter_buyagain(stock_list)  # 过滤近期可买入股票（注释掉）
    stock_list = filter_kcbj_stock(stock_list)  # 过滤科创板股票
    stock_list = ffscore_stock(context, g.score, stock_list, context.current_dt.date())  # 根据财务评分过滤股票
    print('本月股票池 %s 个' % len(stock_list))
#    log.info("——————————————————————————————————")
#    for i, stock in enumerate(stock_list):
#        rank=i+1
#        name = get_security_info(stock).display_name
#        print("本月股票池第 {}：{} {}".format(rank, stock, name))
#    log.info("——————————————————————————————————")
    return stock_list

#   定义调仓策略：调整持仓至设定的仓位比例
def my_adjust_position(context, hold_stocks):
    # 获取账户总资产价值
    free_value = context.portfolio.total_value
    # 计算每只股票的最大持仓比例
    maxpercent = 1.3 / g.buy_stock_count
    # 计算每只股票的买入金额
    buycash = free_value / g.buy_stock_count
    # 遍历当前持仓的每只股票
    for stock in context.portfolio.positions.keys():
        # 获取当前的股票市场数据
        current_data = get_current_data()
        # 获取该股票最近一日的收盘价
        price1d = get_close_price(stock, 1)
        # 判断股票是否处于不可卖出状态（例如涨停）
        nosell_1 = context.portfolio.positions[stock].price >= current_data[stock].high_limit
        # 判断股票是否不在持有列表中
        sell_2 = stock not in hold_stocks
        # 如果股票不在持有列表中且不是处于不可卖出状态，则清仓该股票
        if sell_2 and not nosell_1:
            close_position(stock)
        # 否则，计算当前股票持仓占总资产的比例
        else:
            current_percent = context.portfolio.positions[stock].value / context.portfolio.total_value
            # 如果当前持仓比例超过最大持仓比例，则调整至目标买入金额
            if current_percent > maxpercent:
                order_target_value(stock, buycash)

#   卖出函数
def mysell(context):
    # 调用 get_stock_rank_m_m 函数对全局变量 g.chosen_stock_list 中的股票进行排名
    # 这个函数可能基于某种策略（例如动量、价值或其他财务指标）对股票进行排序
    # g.chosen_stock_list 存储了当前选择持有的股票列表
    g.chosen_stock_list = get_stock_rank_m_m(g.chosen_stock_list)
    
    # 调用 my_adjust_position 函数调整持仓
    # 这个函数可能会根据预设的仓位比例和当前的持仓情况进行调整
    # 例如，如果某只股票的持仓比例超出了设定的范围，该函数可能会卖出部分股票
    # 如果某只股票不在 g.chosen_stock_list 中，也可能会被卖出
    # 该函数确保最终的持仓符合策略的要求
    my_adjust_position(context, g.chosen_stock_list)

关键函数解锁后查看：

# 定义平仓，卖出指定持仓
def close_position(code):
    order = order_target_value(code, 0)    # 下单清仓 ；可能会因停牌失败
    if order != None and order.status == OrderStatus.held:
        g.sold_stock[code] = 0

# 定义过滤停牌股票
def filter_paused_stock(stock_list):
	current_data = get_current_data()   # 获取当前所有股票的数据
	return [stock for stock in stock_list if not current_data[stock].paused]   # 使用列表推导式，返回不在停牌状态的股票列表

# 定义过滤ST及其他具有退市标签的股票        
def filter_st_stock(stock_list):
    current_data = get_current_data()   # 获取当前所有股票的数据
    return [stock for stock in stock_list if not current_data[stock].is_st and 'ST' not in current_data[stock].name and '*' not in current_data[stock].name and '退' not in current_data[stock].name]

# 定义过滤涨停的股票
def filter_limitup_stock(context, stock_list):
    last_prices = history(1, unit='1m', field='close', security_list=stock_list)  # 获取最近一分钟的收盘价
    current_data = get_current_data()   # 获取当前所有股票的数据
    return [stock for stock in stock_list if stock in context.portfolio.positions.keys() 
        or last_prices[stock][-1] < current_data[stock].high_limit]

# 定义过滤跌停的股票
def filter_limitdown_stock(context, stock_list):
    last_prices = history(1, unit='1m', field='close', security_list=stock_list)  # 获取最近一分钟的收盘价
    current_data = get_current_data()   # 获取当前所有股票的数据
    return [stock for stock in stock_list if stock in context.portfolio.positions.keys()
			or last_prices[stock][-1] > current_data[stock].low_limit]

# 定义过滤次新股
def filter_new_stock(context, stock_list):
    return [stock for stock in stock_list if (context.previous_date - datetime.timedelta(days=g.tradeday)) > get_security_info(stock).start_date]

# 定义过滤昨日涨幅过高的股票    
def filter_increase1d(stock_list):
    return [stock for stock in stock_list if get_close_price(stock, 1) / get_close_price(stock, 2) < (1 + g.increase1d)]

# 定义过滤买过的股票
def filter_buyagain(stock_list):
    return [stock for stock in stock_list if stock not in g.sold_stock.keys()]

# 定义过滤科创北交股票
def filter_kcbj_stock(stock_list):
    for stock in stock_list[:]:
        if stock[0] == '4' or stock[0] == '8' or stock[:2] == '68':  # 根据股票代码判断是否属于科创板或北交所
            stock_list.remove(stock)
    return stock_list

# 定义过滤基本面股票    
def ffscore_stock(context,score,security_list,date):
    # 定义观察日期，即用于获取历史财务数据的日期
    my_watch_date = date
    # 计算一年前的日期
    one_year_ago = my_watch_date - datetime.timedelta(days=365)
    # 获取过去五个季度的历史财务数据,security_list 是股票列表，用于过滤股票,指定要获取的财务指标列表
    h = get_history_fundamentals(security_list,
                             [indicator.adjusted_profit,  # 调整后的净利润
                              balance.total_current_assets,  # 流动资产总额
                              balance.total_assets,  # 资产总额
                              balance.total_current_liability,  # 流动负债总额
                              balance.total_non_current_liability,  # 非流动负债总额
                              cash_flow.net_operate_cash_flow,  # 经营活动产生的现金流量净额
                              income.operating_revenue,  # 营业收入
                              income.operating_cost,  # 营业成本
                              ],
                             watch_date=my_watch_date, count=5).dropna()  # 连续的5个季度
    # 定义计算最近四个季度总和的函数
    def ttm_sum(x):
        return x.iloc[1:].sum()
    # 定义计算最近四个季度平均值的函数
    def ttm_avg(x):
        return x.iloc[1:].mean()
    # 定义计算前四个季度总和的函数
    def pre_ttm_sum(x):
        return x.iloc[:-1].sum()
    # 定义计算前四个季度平均值的函数
    def pre_ttm_avg(x):
        return x.iloc[:-1].mean()
    # 定义获取最新季度值的函数
    def val_1(x):
        return x.iloc[-1]
    # 定义获取前一个季度值的函数
    def val_2(x):
        if len(x.index) > 1:
            return x.iloc[-2]
        else:
            return nan
    # 扣非利润
    adjusted_profit_ttm = h.groupby('code')['adjusted_profit'].apply(ttm_sum)
    adjusted_profit_ttm_pre = h.groupby('code')['adjusted_profit'].apply(pre_ttm_sum)
    # 总资产平均
    total_assets_avg = h.groupby('code')['total_assets'].apply(ttm_avg)
    total_assets_avg_pre = h.groupby('code')['total_assets'].apply(pre_ttm_avg)
    # 经营活动产生的现金流量净额
    net_operate_cash_flow_ttm = h.groupby('code')['net_operate_cash_flow'].apply(ttm_sum)
    # 长期负债率: 长期负债/总资产
    long_term_debt_ratio = h.groupby('code')['total_non_current_liability'].apply(val_1) / h.groupby('code')['total_assets'].apply(val_1)
    long_term_debt_ratio_pre = h.groupby('code')['total_non_current_liability'].apply(val_2) / h.groupby('code')['total_assets'].apply(val_2)
    # 流动比率：流动资产/流动负债
    current_ratio = h.groupby('code')['total_current_assets'].apply(val_1) / h.groupby('code')['total_current_liability'].apply(val_1)
    current_ratio_pre = h.groupby('code')['total_current_assets'].apply(val_2) / h.groupby('code')['total_current_liability'].apply(val_2)
    # 营业收入
    operating_revenue_ttm = h.groupby('code')['operating_revenue'].apply(ttm_sum)
    operating_revenue_ttm_pre = h.groupby('code')['operating_revenue'].apply(pre_ttm_sum)
    # 营业成本
    operating_cost_ttm = h.groupby('code')['operating_cost'].apply(ttm_sum)
    operating_cost_ttm_pre = h.groupby('code')['operating_cost'].apply(pre_ttm_sum)
    # 1. ROA 资产收益率
    roa = adjusted_profit_ttm / total_assets_avg
    roa_pre = adjusted_profit_ttm_pre / total_assets_avg_pre
    # 2. OCFOA 经营活动产生的现金流量净额/总资产
    ocfoa = net_operate_cash_flow_ttm / total_assets_avg
    # 3. ROA_CHG 资产收益率变化
    roa_chg = roa - roa_pre
    # 4. OCFOA_ROA 应计收益率: 经营活动产生的现金流量净额/总资产 -资产收益率
    ocfoa_roa = ocfoa - roa
    # 5. LTDR_CHG 长期负债率变化 (长期负债率=长期负债/总资产)
    ltdr_chg = long_term_debt_ratio - long_term_debt_ratio_pre
    # 6. CR_CHG 流动比率变化 (流动比率=流动资产/流动负债)
    cr_chg = current_ratio - current_ratio_pre
    # 8. GPM_CHG 毛利率变化 (毛利率=1-营业成本/营业收入)
    gpm_chg = operating_cost_ttm_pre/operating_revenue_ttm_pre - operating_cost_ttm/operating_revenue_ttm
    # 9. TAT_CHG 资产周转率变化(资产周转率=营业收入/总资产)
    tat_chg = operating_revenue_ttm/total_assets_avg - operating_revenue_ttm_pre/total_assets_avg_pre
    # 查询特定股票的股本变动信息
    spo_list = list(set(finance.run_query(
        query(
            finance.STK_CAPITAL_CHANGE.code       # 查询股票代码
        ).filter(
            finance.STK_CAPITAL_CHANGE.code.in_(security_list),  # 筛选特定股票列表
            finance.STK_CAPITAL_CHANGE.pub_date.between(one_year_ago, my_watch_date),  # 筛选特定日期范围内的变动
            finance.STK_CAPITAL_CHANGE.change_reason_id == 306004)  # 筛选特定变动原因
    )['code']))
    spo_score = pd.Series(True, index = security_list)  # 初始化一个 True 值的 Series，用于标记股票是否通过初步筛选
    # 如果存在股本变动的股票，将它们标记为 False（不通过初步筛选）
    if spo_list:
        spo_score[spo_list] = False
    # 创建一个空的 DataFrame 用于存储股票评分
    df_scores = pd.DataFrame(index=security_list)# 1
    # 根据一系列财务指标和市场行为为股票打分
    df_scores['roa'] = roa>0.0   # 1. 净资产收益率大于0
    df_scores['ocfoa'] = ocfoa>0  # 2. 经营活动产生的现金流量净额大于0
    df_scores['roa_chg'] = roa_chg>0   # 3. 净资产收益率同比增长率大于0
    df_scores['ocfoa_roa'] = ocfoa_roa>0   # 4. 经营活动产生的现金流量净额与净资产收益率的比值大于0
    df_scores['ltdr_chg'] = ltdr_chg<=0   # 5. 长期负债同比增长率小于等于0
    df_scores['cr_chg'] = cr_chg>0   # 6. 流动负债同比增长率大于0
    df_scores['spo'] = spo_score  > 0   # 7. 股本变动筛选结果
    df_scores['gpm_chg'] = gpm_chg>0  # 8. 毛利率同比增长率大于0
    df_scores['tat_chg'] = tat_chg>0    # 9. 合计增长率大于0
    df_scores = df_scores.dropna() # 删除包含 NaN 值的行
    # 计算每只股票的总评分
    df_scores['total'] = df_scores['roa'] + df_scores['ocfoa'] + df_scores['roa_chg'] + \
        df_scores['ocfoa_roa'] + df_scores['ltdr_chg'] + df_scores['cr_chg'] + \
        df_scores['spo'] + df_scores['gpm_chg'] + df_scores['tat_chg']
    # 根据总评分筛选股票，并按评分降序排序
    res  = df_scores.loc[lambda df_scores: df_scores['total'] > score].sort_values(by = 'total',ascending=False).index
    # 获取筛选后股票的 ROE（净资产收益率）数据，并按 ROE 降序排序
    q = get_fundamentals(query(indicator.code,indicator.roe).filter(indicator.code.in_(res)).order_by(indicator.roe.desc()).limit(len(res)))
    # 将最终筛选出的股票代码转换为列表
    res = list(q['code'])
    return res   # 返回最终筛选出的股票列表

2025-02-21