ptrade量化策略揭秘,小市值择时策略，回测年化收益高达90%

"ptrade量化策略之小市值择时，回测年化90%" 这个描述指的是一种量化投资策略，该策略专注于小市值股票，并通过择时操作来优化收益。以下是对这种策略的一些基本解析：
1. "小市值股票"：小市值股票通常指的是市值较小的公司股票，这些公司往往具有较高的成长潜力，但也伴随着更高的风险。量化策略通常会寻找这些股票中的投资机会。
2. "择时操作"：择时是指投资者根据市场情况调整投资组合的时间点，以期在市场低迷时买入，在市场上涨时卖出。这种策略试图通过预测市场趋势来增加收益。
3. "回测年化90%"：回测是指使用历史数据来测试一个投资策略的性能。年化是指将策略的回报率按年度进行换算。如果一个策略的回测年化回报率为90%，这意味着在假设的测试期间，该策略的年化收益率达到了90%。
以下是一个简化的策略框架，可能用于实现这种小市值择时策略：
### 策略框架
#### 1. 数据收集 - 收集小市值股票的历史价格数据。 - 收集可能影响股票价格的经济指标和财务指标。
#### 2. 特征工程 - 对股票进行特征提取，如市盈率、市净率、流动比率等。 - 对宏观经济指标进行筛选，如GDP增长率、利率、通货膨胀率等。

相关内容：

本文是小市值择时的ptrade版本，可以直接在ptrade上回测和实盘交易，本策略仅供参考学习，不作为投资建议。

小市值的核心就是股票池中选择市值最小的并且基本面没太大问题的股票，这类股票价格波动大，有较大的价值发现空间；涨多了调出股票池，跌到止损线止损卖出，持仓数量不足则买入，来实现小波段高抛低吸。

通过择时和风控策略来实现控制回撤。

回测数据(2015.1.1-2025.6.19)如下：

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

1、策略初始化配置

定义了可交易日、目标持仓股票数、个股止损比例、空仓时持有的货币基金等

g.index = "399101.SZ"  #中小板综
g.adjust_num = True #是否动态调整持仓数量
g.trading_signal = True # 是否为可交易日
g.etf = '511880.SS' # 银华日利
g.stock_num = 4 # 目标持仓股票数
g.run_stoploss = True # 是否执行止损
g.stoploss_strategy = 3 # 止损策略：1个股止损、2市场止损、3联合策略
g.stoploss_limit = 0.07 # 个股止损比例

1.1、盘前处理

记录持仓股昨天涨停情况

# 盘前处理
def before_trading_start(context, data):
    g.hold_list = 
    g.limitup_stocks = 
    pre_date = context.previous_date.strftime('%Y%m%d')
    print('pre_date---------------')
    print(pre_date)
    # 更新当前持仓列表
    for position in list(context.portfolio.positions.keys()):
        g.hold_list.append(position)
    # 若有持仓，则记录昨日涨停股票
    print('g.hold_list-----------------')
    print(g.hold_list)
    if len(g.hold_list) != 0:
        df = get_price(g.hold_list, end_date=pre_date, frequency='1d', 
                       fields=, count=1)
        df = df == df]
        g.yesterday_HL_list = list(df.index)
        print("昨天持仓涨停股票--------------")
        print(g.yesterday_HL_list)
    else:
        
        g.yesterday_HL_list = 

2、选股逻辑

结合基本面和行情选股

（1）剔除停牌、ST、退市的股票

# 将ST、停牌、退市三种状态的股票剔除当日的股票池
final_list = df.index.tolist()
final_list = filter_st_status(final_list)
final_list = filter_halt_status(final_list)
final_list = filter_deli_status(final_list)

def filter_st_status(stocks_list):
    filter_stocks = 
    # 判断证券是否为ST、停牌或者退市
    st_status = get_stock_status(stocks_list, 'ST')
    # 将不是ST的证券筛选出来
    for i in stocks_list:
        if st_status is not True:
            filter_stocks.append(i)     
    return filter_stocks

def filter_halt_status(stocks_list):
    filter_stocks = 
    # 判断证券是否为ST、停牌或者退市
    HALT_status = get_stock_status(stocks_list, 'HALT')
    # 将不是ST的证券筛选出来
    for i in stocks_list:
        if HALT_status is not True:
            filter_stocks.append(i)     
    return filter_stocks

def filter_deli_status(stocks_list):
    filter_stocks = 
    # 判断证券是否为ST、停牌或者退市
    DELISTING_status = get_stock_status(stocks_list, 'DELISTING')
    # 将不是ST的证券筛选出来
    for i in stocks_list:
        if DELISTING_status is not True:
            filter_stocks.append(i)     
    return filter_stocks  

（2）筛选基本面良好的股票

市值在3-1000亿，母公司净利润与公司净利润都为正，营收不低于1亿

stock_list = get_index_stocks(g.index)
# 指数成分股按昨日收盘时的流通市值进行从小到大排序，截取市值最小的10个标的进行股票状态筛选（考虑回测速度）
subDf = get_fundamentals(stock_list, "income_statement", fields=,
                      date=context.previous_date)
subDf = subDf > 0) & (subDf > 0) & (subDf > 100000000)]
stock_list_tmp = subDf.index.tolist()
df = get_fundamentals(stock_list_tmp, "valuation", fields=,
                      date=context.previous_date)
df = df > 300000000) & (df < 100000000000)]
df = df.sort_values(by="total_value", ascending=True)

（3）根据大盘均线来动态调整股票池数量

通过大盘10日均线与最新收盘价的差值来代表市场的温度，200以上，表示价格显著高于均线，市场过热，减少持仓数量；-200以下，表示价格显著低于均线，市场超跌，增加持仓数量

# 获取MA函数
def get_ma(close_array, num):
    ma = close_array.mean()
    return round(ma, 2)

def adjust_stock_num(context):
    ma_para = 10
    # PTrade获取指数数据
    h = get_history(ma_para, frequency="1d", field="close", security_list="399101.SZ")
    
    close_data = h
    ma10 = get_ma(close_data, ma_para)
    # 数据完整性校验
    if close_data is None or len(close_data) < ma_para:
        log.error('获取指数数据失败，使用默认持仓数4')
        return 4
    
    diff = close_data - ma10
    print('diff--------------')
    print(diff)
    if diff >= 500:
        result = 3
    elif 200 <= diff < 500:
        result = 3
    elif -200 <= diff < 200:
        result = 4
    elif -500 <= diff < -200:
        result = 5
    else:
        result = 6
    return result

（4）1，4月空仓，规避风险

if month in g.pass_months: 
   return False
else:
   return True

3、调仓逻辑

每周固定时间调仓，在调仓之前进行止盈止损，当收益翻倍时止盈，当价格跌破止损线时止损

（1）止盈止损

if g.run_stoploss:
        current_positions = context.portfolio.positions

        # 个股止盈与止损（策略1或3）
        if g.stoploss_strategy in :
            for stock in list(current_positions.keys()):
                pos = get_position(stock)
                if pos.enable_amount <= 0:
                    log.info("股票 {} 持仓可用数量为0，跳过止盈止损".format(stock))
                    continue
                price = pos.last_sale_price
                avg_cost = pos.cost_basis
                # 当收益翻倍时止盈
                if price >= avg_cost * 2:
                    log.info("股票 {} 收益翻倍，准备止盈".format(stock))
                    if can_trade(context, stock, "sell") and close_position(context, stock):
                        log.info("股票 {} 成功止盈卖出".format(stock))
                    else:
                        log.info("股票 {} 止盈下单失败".format(stock))
                # 当价格跌破止损线时止损
                elif price < avg_cost * (1 - g.stoploss_limit):
                    log.info("股票 {} 跌破止损线，准备止损".format(stock))
                    if can_trade(context, stock, "sell") and close_position(context, stock):
                        log.info("股票 {} 成功止损卖出".format(stock))
                        # g.reason_to_sell = 'stoploss'
                    else:
                        log.info("股票 {} 止损下单失败".format(stock))

（2）动态调仓

选择市值最小的股票放在目标股票池，剔除持仓中不在目标股票池的股票并卖出，买入新进目标股票池的股票，达到目标持仓数量

g.target_list = get_stock_list(context)
     log.info("调仓目标股票列表: {}--------------------".format(g.target_list))
            
     sell_list = 
     hold_list = 
     log.info("待卖出股票: {}----------------".format(sell_list))
     log.info("保留持仓: {}------------------".format(hold_list))
            
     positions = list(context.portfolio.positions.keys())
     for stock in sell_list:
         if stock in positions:
            pos = get_position(stock)
            if pos.enable_amount > 0:
                 close_position(context, stock)
             else:
                 log.info("股票 {} 当前无可卖数量--------------".format(stock))
      buy_list = 
      sleep(30)
      buy_security(context, buy_list, len(buy_list))

（3）买卖时校验边界情况

不能卖：停牌、不在持仓中、持仓数量为0

不能买：停牌、已持仓

    halt_status = get_stock_status(, 'HALT')
    pos = get_position(security)
    positions = list(context.portfolio.positions.keys())
    if action == "sell":
        if halt_status is True:
            log.info("无法执行卖出操作：股票 {} 停牌中------------".format(security))
            return False
        if security not in positions:
            log.info("无法执行卖出操作：股票 {} 不在持仓中-------------".format(security))
            return False
        if pos.enable_amount <= 0:
            log.info("无法执行卖出操作：股票 {} 持仓可用数量为0-----------------".format(security))
            return False
        return True
    elif action == "buy":
        if halt_status is True:
            log.info("无法执行买入操作：股票 {} 停牌中---------------".format(security))
            return False
        # 如果已持有，则不执行买入操作（策略规定：只在空仓时买入ETF或首次进场）
        if security in positions and pos.amount > 0:
            log.info("无法执行买入操作：股票 {} 已持仓------------------".format(security))
            return False
        return True
    else:
        log.info("未知交易操作: {}--------------------".format(action))
        return False

这篇文章主要分享小市值+择时策略的思路，使用ptrade写法

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