2411 小盘三正超分散动态平衡策略 低风险中等收益策略 48789

1. 小盘股优选机制：

• 基础池筛选：以深证民营指数（399317）为基准，剔除 ST 股票
• 市值分层：选取全市场市值最小的前 10% 个股作为候选池
• 财务三正筛选：要求市净率 > 0、累计收益增长 > 0、经营现金流 / 营收 > 0
• 持仓延续性：优先保留已有持仓个股，降低换手率

2. 组合管理机制：

• 超分散配置：设置 100 个持仓单位，单只个股占比 1%
• 股债动态平衡：配置国债 ETF (511220)、黄金 ETF (518880)、标普 500ETF (513500) 作为现金管理工具
• 渐进调仓：每日调整 20% 仓位，维持组合结构稳定

3. 交易执行规则：

• 开盘集中交易：9:35 完成主要调仓操作
• 双向市价单：卖出使用 99% 现价保护，买入使用 101% 现价追单
• 流动性保障：每笔交易至少 100 股起，保证最小成交单位

4. 风险控制体系：

• 财务安全边际：严格筛选盈利增长与现金流正向个股
• 极端行情缓冲：配置 30% 海外资产对冲系统性风险
• 实时监控机制：盘后生成详细持仓报告，监控组合风险敞口

策略代码

import pandas as pd
import json

def initialize(context):
    # setting system
    log.set_level('order', 'error')
    set_option('use_real_price', True)
    set_option('avoid_future_data', True)
    # setting strategy
    run_daily(iUpdate, 'before_open')
    run_daily(iTrader, '9:35')
    run_daily(iReport, 'after_close')
    g.days = 0 # day counter

def iUpdate(context):
    # parameters
    nposition = 100 # number of positions
    nchoice = 30
    # daily update
    g.days = g.days + 1
    g.stocks = _choice_small(context, nchoice)
    g.funds = _choice_funds(context)
    g.position_size = 1.0/nposition * context.portfolio.total_value

关键函数解锁后查看：

def iReport(context):
    # table of positions
    cdata = get_current_data()
    tvalue = context.portfolio.total_value
    ptable = pd.DataFrame(columns=['amount', 'value', 'weight', 'name'])
    for s in context.portfolio.positions:
        ps = context.portfolio.positions[s]
        ptable.loc[s] = [ps.total_amount, int(ps.value), 100*ps.value/tvalue, cdata[s].name]
    ptable = ptable.sort_values(by='weight', ascending=False)
    # daily report
    pd.set_option('display.max_rows', None)
    log.info('  positions', len(ptable), '\n', ptable.head())
    log.info('  total win %i, return %.2f%%', \
            int(tvalue - context.portfolio.inout_cash), 100*context.portfolio.returns)
    log.info('  total value %.2f, cash %.2f', \
            context.portfolio.total_value/10000, context.portfolio.available_cash/10000)
    log.info('running days', g.days)

def _choice_small(context, nchoice):
    # parameters
    index = '399317.XSHE'
    # stocks
    dt_now = context.current_dt.date()
    stocks = get_index_stocks(index, dt_now)
    # non-ST
    cdata = get_current_data()
    stocks = [s for s in stocks if not cdata[s].is_st]
    # small stocks, 10%
    m = int(0.1*len(stocks))
    df = get_fundamentals(query(
            valuation.code,
            valuation.market_cap,
            valuation.pb_ratio,
            indicator.inc_return,
            indicator.ocf_to_revenue,
        ).filter(
            valuation.code.in_(stocks),
        ).order_by(valuation.market_cap.asc()
        ).limit(m)
        ).dropna().set_index('code')
    # qualify, 三正
    df = df[(df.pb_ratio > 0) & (df.inc_return > 0) & (df.ocf_to_revenue > 0)]
    # choice
    n = int(1.2 * nchoice) # buffer 20%
    stocks = df.head(n).index.tolist()
    # united
    stocks_0 = [s for s in stocks if s in context.portfolio.positions]
    stocks_1 = [s for s in stocks if s not in context.portfolio.positions]
    choice = (stocks_0 + stocks_1)[:nchoice]
    # report
    df = df[['market_cap']].loc[choice]
    df['name'] = [cdata[s].name for s in df.index]
    log.info('small-quality stocks', len(choice), '\n', df.head())
    # reuslt
    return choice

def _choice_funds(context):
    # load funds
    #funds = json.loads(read_file('funds'))
    funds = ['511220.XSHG', '518880.XSHG', '513500.XSHG']
    # filter
    cdata = get_current_data()
    funds = [s for s in funds if not cdata[s].paused]
    if len(funds) == 0:
        funds = ['000012.XSHG'] # default
    # results
    return funds
# end

2025-02-24