量化策略之白马股价投回测40倍

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

回测数据(2013.1.1-2024.12.31)如下：

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

1、策略初始化配置

定义了持股数、股票池、市场温度、调仓函数等

g.buy_stock_count = 5 # 持股数
g.check_out_lists = [] # 股票池
g.market_temperature = "warm" # 市场温度
# 盘前数据处理
run_monthly(before_market_open, 1, time='5:00', reference_security='000300.XSHG')
# 调仓函数
run_monthly(my_trade, 1, time='9:45', reference_security='000300.XSHG')

2、盘前处理

（1）计算市场温度

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

def Market_temperature(context):
    # 获取沪深300指数过去220个交易日的收盘价
    index300 = attribute_history('000300.XSHG', 220, '1d', ('close'), df=False)['close']
    # 计算市场的相对位置
    market_height = (mean(index300[-5:]) - min(index300)) / (max(index300) - min(index300))
    # 底部区域
    if market_height < 0.20:
        g.market_temperature = "cold"
    # 顶部区域
    elif market_height > 0.90:
        g.market_temperature = "hot"
    # 相对合理位置
    elif max(index300[-60:]) / min(index300) > 1.20:
        g.market_temperature = "warm"

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

all_stocks = get_index_stocks("000300.XSHG")
    # 过滤创业板、ST、停牌、当日涨停
    all_stocks = [stock for stock in all_stocks if not (
            (current_data[stock].day_open == current_data[stock].high_limit) or  # 涨停开盘
            (current_data[stock].day_open == current_data[stock].low_limit) or  # 跌停开盘
            current_data[stock].paused or  # 停牌
            current_data[stock].is_st or  # ST
            ('ST' in current_data[stock].name) or
            ('*' in current_data[stock].name) or
            ('退' in current_data[stock].name) or
            (stock.startswith('30')) or  # 创业
            (stock.startswith('68')) or  # 科创
            (stock.startswith('8')) or  # 北交
            (stock.startswith('4'))   # 北交
    )]

（3）根据市场温度选股

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

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

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

选出6只股票

if g.market_temperature == "cold":
        q = query(
            valuation.code, 
            ).filter(
            valuation.pb_ratio > 0,
            valuation.pb_ratio < 1,
            cash_flow.subtotal_operate_cash_inflow > 0,
            indicator.adjusted_profit > 0,
            cash_flow.subtotal_operate_cash_inflow/indicator.adjusted_profit>2.0,
            indicator.inc_return > 1.5,
            indicator.inc_net_profit_year_on_year > -15,
            valuation.code.in_(all_stocks)
            ).order_by(
            (indicator.roa/valuation.pb_ratio).desc()
        ).limit(
            g.buy_stock_count + 1
        )
    elif g.market_temperature == "warm":
        q = query(
            valuation.code, 
            ).filter(
            valuation.pb_ratio > 0,
            valuation.pb_ratio < 1,
            cash_flow.subtotal_operate_cash_inflow > 0,
            indicator.adjusted_profit > 0,
            cash_flow.subtotal_operate_cash_inflow/indicator.adjusted_profit>1.0,
            indicator.inc_return > 2.0,
            indicator.inc_net_profit_year_on_year > 0,
            valuation.code.in_(all_stocks)
            ).order_by(
            (indicator.roa/valuation.pb_ratio).desc()
        ).limit(
            g.buy_stock_count + 1
        )
    elif g.market_temperature == "hot":
        q = query(
            valuation.code, 
            ).filter(
 
            valuation.pb_ratio > 3,
            cash_flow.subtotal_operate_cash_inflow > 0,
            indicator.adjusted_profit > 0,
            cash_flow.subtotal_operate_cash_inflow/indicator.adjusted_profit>0.5,
            indicator.inc_return > 3.0,
            indicator.inc_net_profit_year_on_year > 20,
            valuation.code.in_(all_stocks)
            ).order_by(
            indicator.roa.desc()
        ).limit(
            g.buy_stock_count + 1
        )

（4）动量因子打分

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

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

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

def MOM(stock,days):
    df = attribute_history(stock, days, '1d', ['close'], df= False)
    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)
    # 250天年化收益率
    annualized_returns = math.pow(math.exp(slope), 250) - 1
    residuals = y - (slope * x + intercept)
    weighted_residuals = weights * residuals**2
    # 高R平方值
    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(stock_pool, days,ll, hh):
    score_list = []
    for stock in stock_pool:
        score = MOM(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）卖出

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

nosell_1 = context.portfolio.positions[stock].price >= current_data[stock].high_limit
sell_2 = stock not in buy_stocks
        if sell_2 and not nosell_1:
            log.info("调出平仓：[%s]" % (stock))
            position = context.portfolio.positions[stock]
            close_position(position)
        else:
            log.info("已持仓，本次不买入：[%s]" % (stock))

（2）买入

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

position_count = len(context.portfolio.positions)
    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 context.portfolio.positions:
                if open_position(stock, value):
                    if len(context.portfolio.positions) >= g.buy_stock_count:
                        break