本策略的核心是在沪深300股票池中选股，选择的股票都是市值比较大，业绩比较好的白马股，根据市场相对位置以及基本面因子来选择股票，每月调仓一次，总的手续费低，策略的容量比较大，比较适合资金量大的稳健投资者和上班族。

回测数据(2020.1.1-2025.7.10)如下：

*回测数据只作测试用，不代表未来实际收益

1、策略初始化配置

定义了持股数、股票池、市场位置、调仓函数等

g.buy_stock_count = 5 # 持股数
g.check_out_lists = [] # 股票池
g.market_temperature = "mid" # 市场位置
g.month = 0 # 记录月份
# 调仓函数
run_daily(context, my_trade, time='9:45')

2、盘前处理

（1）计算市场位置

市场位置监控，获取沪深300指数过去220个交易日的收盘价，计算市场的相对位置，如果在0.2以下就是底部区域，0.9以上就是顶部区域，最近60日最高涨幅超过20%就是相对温和上涨位置。

def Market_temperature(context):
    index300 = get_history(220, frequency="1d", field="close", security_list="000300.SS").close.tolist()
    market_height = (np.mean(index300[-5:]) - np.min(index300)) / (np.max(index300) - np.min(index300))
    if market_height < 0.20:
        g.market_temperature = "low"

    elif market_height > 0.90:
        g.market_temperature = "high"

    elif np.max(index300[-60:]) / np.min(index300) > 1.20:
        g.market_temperature = "mid" 

（2）过滤科创北交、ST、停牌、当日涨停股票

all_stocks = get_index_stocks("000300.SS")
    list = []
    check_out_lists = []

    final_list = filter_st_status(all_stocks)
    final_list = filter_halt_status(final_list)
    final_list = filter_deli_status(final_list)

    for stock in final_list:
        info = get_stock_info(stock)
        if not (('ST' in info[stock]["stock_name"]) or
            ('*' in info[stock]["stock_name"]) or
            ('退' in info[stock]["stock_name"]) or
            (stock.startswith('30')) or  # 创业
            (stock.startswith('68')) or  # 科创
            (stock.startswith('8')) or  # 北交
            (stock.startswith('4'))):
              list.append(stock)

（3）根据市场位置选股

低位：选择强现金流、扣非净利润为正的破净股，以roa/pb来排序

中位：选择营收翻倍高增的破净股，以roa/pb来排序

高位：选择pb3以上当前营收3倍高增、利润20%增长的成长股, 只用roa排序

选出6只股票

if g.market_temperature == "low":
      check_out_lists = get_market_low(context, list)
elif g.market_temperature == "mid":
      check_out_lists = get_market_mid(context, list)
elif g.market_temperature == "high":
      check_out_lists = get_market_high(context, list)

def get_market_low(context, stock_list):
    df = get_fundamentals(stock_list, "valuation", fields=["pb"],
                          date=context.previous_date)
    df = df[(df["pb"] > 0) & (df["pb"] < 1)]  
    if df.empty:
        return []
    list = df.index.tolist()
    # print(list)
    subtotal_operate_cash_inflow_df = get_single_fundamentals(context, list, "cashflow_statement", "subtotal_operate_cash_inflow") 
    net_profit_cut_df = get_single_fundamentals(context, list, "profit_ability", "net_profit_cut")
    net_profit_grow_rate_df = get_single_growth_fundamentals(context, list, "income_statement", "net_profit")
    roe_cut_df = get_single_roe_cut_fundamentals(context, list)
    roa_df = get_single_roa_fundamentals(context, list)
    df = pd.concat([subtotal_operate_cash_inflow_df, net_profit_cut_df, net_profit_grow_rate_df, roe_cut_df, roa_df], axis=1)
    
    df["division"] = df["subtotal_operate_cash_inflow"] / df["net_profit_cut"]
    df = df[(df["subtotal_operate_cash_inflow"] > 0) & (df["net_profit_cut"] > 0) & (df["net_profit"] > -15) & (df["division"] > 2) & (df["roe_cut"] > 1.5)]
    if df.empty:
        return []
    filter_list = df["stocks"].iloc[:, -1].tolist()
    pb_df = get_fundamentals(filter_list, "valuation", fields=["pb"],
                          date=context.previous_date)
    stocks_pb = pb_df["pb"].tolist()
    df["pb"] = stocks_pb
    df["sort"] = df["roa"] / df["pb"]
    df = df.sort_values(by="sort", ascending=False)
    filter_list = df["stocks"].iloc[:, -1].tolist()
    print(filter_list)
    return filter_list[:6]

def get_market_mid(context, stock_list):
    # stock_list = ["601211.SS", "600036.SS"]
    df = get_fundamentals(stock_list, "valuation", fields=["pb"],
                          date=context.previous_date)                      
    df = df[(df["pb"] > 0) & (df["pb"] < 1)]  
    if df.empty:
        return []
    list = df.index.tolist()
    subtotal_operate_cash_inflow_df = get_single_fundamentals(context, list, "cashflow_statement", "subtotal_operate_cash_inflow") 
    net_profit_cut_df = get_single_fundamentals(context, list, "profit_ability", "net_profit_cut")
    net_profit_grow_rate_df = get_single_growth_fundamentals(context, list, "income_statement", "net_profit")
    roe_cut_df = get_single_roe_cut_fundamentals(context, list)
    roa_df = get_single_roa_fundamentals(context, list)
    df = pd.concat([subtotal_operate_cash_inflow_df, net_profit_cut_df, net_profit_grow_rate_df, roe_cut_df, roa_df], axis=1)
    
    df["division"] = df["subtotal_operate_cash_inflow"] / df["net_profit_cut"]
    df = df[(df["subtotal_operate_cash_inflow"] > 0) & (df["net_profit_cut"] > 0) & (df["net_profit"] > 0) & (df["division"] > 1) & (df["roe_cut"] > 2)]
    if df.empty:
        return []
    filter_list = df["stocks"].iloc[:, -1].tolist()
    pb_df = get_fundamentals(filter_list, "valuation", fields=["pb"],
                          date=context.previous_date)
    stocks_pb = pb_df["pb"].tolist()
    df["pb"] = stocks_pb
    df["sort"] = df["roa"] / df["pb"]
    df = df.sort_values(by="sort", ascending=False)
    filter_list = df["stocks"].iloc[:, -1].tolist()
    print(filter_list)
    return filter_list[:6]  

def get_market_high(context, stock_list):
    df = get_fundamentals(stock_list, "valuation", fields=["pb"],
                          date=context.previous_date)
    df = df[df["pb"] > 3]  
    if df.empty:
        return []
    list = df.index.tolist()
    subtotal_operate_cash_inflow_df = get_single_fundamentals(context, list, "cashflow_statement", "subtotal_operate_cash_inflow") 
    net_profit_cut_df = get_single_fundamentals(context, list, "profit_ability", "net_profit_cut")
    net_profit_grow_rate_df = get_single_growth_fundamentals(context, list, "income_statement", "net_profit")
    roe_cut_df = get_single_roe_cut_fundamentals(context, list)
    roa_df = get_single_roa_fundamentals(context, list)
    df = pd.concat([subtotal_operate_cash_inflow_df, net_profit_cut_df, net_profit_grow_rate_df, roe_cut_df, roa_df], axis=1)
    
    df["division"] = df["subtotal_operate_cash_inflow"] / df["net_profit_cut"]
    df = df[(df["subtotal_operate_cash_inflow"] > 0) & (df["net_profit_cut"] > 0) & (df["net_profit"] > 20) & (df["division"] > 0.5) & (df["roe_cut"] > 3)]
    if df.empty:
        return []
    df = df.sort_values(by="roa", ascending=False)
    filter_list = df["stocks"].iloc[:, -1].tolist()
    print(filter_list)
    return filter_list[:6]  

（4）动量因子打分

动量因子评分，选择评分最高的5只股票，计算两个值：

年化收益率：250天的年化收益率

高R平方值：用于评估趋势的稳定性，高R平方值意味着价格变动更符合线性趋势，策略信号更可靠，筛选出趋势明显的股票，避免在波动大或无趋势的市场中交易。

def MOM(context, stock, days):
    pre_date = context.previous_date.strftime('%Y%m%d')
    df = get_price(stock, end_date=pre_date, frequency='1d', 
                       fields=['close'], count=days)
    y = np.log(df['close'])
    n = len(y)  
    x = np.arange(n)
    weights = np.linspace(1, 2, n)  
    slope, intercept = np.polyfit(x, y, 1, w=weights)
    annualized_returns = math.pow(math.exp(slope), 250) - 1
    residuals = y - (slope * x + intercept)
    weighted_residuals = weights * residuals**2
    r_squared = 1 - (np.sum(weighted_residuals) / np.sum(weights * (y - np.mean(y))**2))
    score = annualized_returns * r_squared
    return score

def Moment_rank(context, stock_pool, days, ll, hh):
    score_list = []
    for stock in stock_pool:
        score = MOM(context, stock, days)
        score_list.append(score)
    df = pd.DataFrame(index=stock_pool, data={'score':score_list})
    df = df.sort_values(by='score', ascending=False)  # 降序 
    df = df[(df['score']>ll) & (df['score']<hh)]
    rank_list = list(df.index)    
    return rank_list

3、调仓逻辑

（1）卖出

卖出不在目标股票池中的股票

hold_list = list(context.portfolio.positions.keys())
    for stock in hold_list:
        if stock not in buy_stocks[:g.buy_stock_count]:
            log.info("调出平仓：[%s]" % (stock))
            close_position(stock)
        else:
            log.info("已持仓，本次不买入：[%s]" % (stock))

（2）买入

根据可用资金平均分配买入

    # 根据股票数量分仓
    # 此处只根据可用金额平均分配购买，不能保证每个仓位平均分配
    stocks = updatePositions(context)
    position_count = len(stocks)
    if g.buy_stock_count > position_count:
        value = context.portfolio.cash / (g.buy_stock_count - position_count)
        
        for stock in buy_stocks[:g.buy_stock_count]:
            if stock not in stocks:
                open_position(stock, value)

这篇文章主要分享沪深300指数增强策略，主要的逻辑在市场位置判断以及选股逻辑上，适合资金量大的稳健投资者。

如果有不懂的，欢迎找我一起交流，加入量化交易大家庭