本策略的核心是在沪深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 scoredef 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_list3、调仓逻辑
（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指数增强策略，主要的逻辑在市场位置判断以及选股逻辑上，适合资金量大的稳健投资者。
如果有不懂的，欢迎找我一起交流，加入量化交易大家庭
