量化研究--年化70%四大搅屎棍策略原理深度分析

*******7387_x00JA · 发表于昨天 13:38

文章声明:本内容为个人的业余研究，和任何单位，机构没有关系，文章出现的股票代码，全部只是测试例子，不做投资参考，投资有风险，代码学习使用，不做商业用途

今天在上线量化使用使用，我给了很多源代码，看起来也伤脑筋，我全部上线了网页，搭载一个强大的量化学习网页http://14.103.193.242:9999/xms_quants.html

视频很多，可以仔细学习

昨天谢了怎么样使用跟单系统跟踪好策略量化研究--年化60%四大搅屎棍策略实盘设置 ,可以下载

今天就详细的解析一下策略的原理，回测的结果，网页上线了一个专题选择好的策略

回测的结果

策略详细分析

这是一个基于行业轮动和涨停股管理的股票策略，名为“四大搅屎棍策略”。下面我将详细分析该策略的原理和各个模块的功能。

策略总体思路：

通过行业宽度分析，选择近期表现较好的行业，但排除“四大搅屎棍”行业（银行、有色金属、钢铁、煤炭）。
在选定的行业中，选择中小市值且基本面良好（ROE>15%, ROA>10%）的股票，构建投资组合。
每周调整一次持仓，卖出不在新选股列表中的股票（除非昨日涨停），并买入新选出的股票。
对于昨日涨停的股票，在当天14:00检查，如果涨停打开则卖出，否则继续持有。

下面分模块详细说明：

一、初始化函数 initialize(context)

设置基准为000985.XSHG（中证全指）
使用真实价格交易，避免未来函数，设置滑点为0，交易成本为万分之三。
初始化全局变量：g.stock_num（每行业选股数）、g.hold_list（当前持仓）、g.yesterday_HL_list（昨日涨停股票）、g.num（选取的行业数量，这里为1）
定时运行函数：每天9:05准备股票列表，每周一9:30调整持仓，每天14:00检查涨停股。

二、行业字典SW1

定义了申万一级行业的代码和名称映射。

三、准备股票池函数 prepare_stock_list(context)

更新当前持仓列表g.hold_list。
检查持仓中昨日涨停的股票，更新g.yesterday_HL_list。

四、行业分析相关函数

industry(stockList, industry_code, date): 计算每个行业在给定股票列表中的股票数量。
getStockIndustry(p_stocks, p_industries_type, p_day): 获取股票对应的行业代码。

五、选股模块 get_stock_list(context)

获取中证全指（000985.XSHG）的成分股作为初始股票池。
计算每个股票过去20日的收盘价，并计算其是否在20日均线之上（df_bias）。
按照行业分组，计算每个行业在20日均线之上的股票比例（df_ratio）。
选取比例最高的g.num（即1）个行业。
如果这Top1行业不是银行、有色金属、钢铁、煤炭，则进行选股，否则返回空列表。
选股条件：从中小板指（399101.XSHE）成分股中，过滤掉科创北交股票、ST股票、次新股、停牌股票、涨停股票和跌停股票，然后选择ROE>15%且ROA>10%的中小市值股票（按市值升序，取前g.stock_num=10只）。

六、调仓函数 weekly_adjustment(context)

获取目标股票列表target_B。
卖出当前持仓中不在目标列表且昨日未涨停的股票。
计算需要买入的股票数量，然后等分现金买入目标股票，直到达到目标数量（g.stock_num * g.num，即10只）或现金用完。

七、涨停股检查函数 check_limit_up(context)

在14:00检查昨日涨停的股票，如果当前价格低于涨停价（即涨停打开），则卖出；否则继续持有。

八、交易模块

自定义下单函数order_target_value_：用于下单并记录日志。
开仓函数open_position：按目标价值下单，如果成交则返回成功。
平仓函数close_position：平掉持仓。

九、股票过滤函数

filter_paused_stock: 过滤停牌股票。
filter_st_stock: 过滤ST和退市风险股票。
filter_kcbj_stock: 过滤科创板、北交所和创业板股票（代码以4、8、68、3开头）。
filter_limitup_stock: 过滤涨停股票（持仓中的股票不过滤）。
filter_limitdown_stock: 过滤跌停股票（持仓中的股票不过滤）。
filter_new_stock: 过滤上市时间小于375天的次新股。

策略逻辑总结：

行业轮动：每周选择20日均线以上股票比例最高的行业（排除四大搅屎棍行业），然后在该行业内选股。
选股标准：中小市值、ROE>15%、ROA>10%，并且通过一系列过滤（停牌、ST、涨跌停、次新等）。
持仓调整：每周调整一次，卖出不符合条件的股票，买入新选出的股票，但昨日涨停的股票即使不在新选股列表中也不卖出（直到14:00检查涨停打开情况）。
涨停股管理：昨日涨停的股票，在当天14:00检查，如果涨停打开则卖出，否则持有。

风险控制：

排除四大行业（银行、有色金属、钢铁、煤炭）是因为这些行业通常市值较大，波动性较高，可能对策略产生较大影响。
通过基本面指标（ROE、ROA）和市值筛选，选择质量较好的中小市值股票。
过滤停牌、ST、涨跌停和次新股，减少非交易性风险。

策略核心思想

1. 行业轮动与市场宽度分析

策略通过分析各行业板块的"市场宽度"来选择投资方向：

计算每个行业板块中股价在20日均线以上的股票比例
选择表现最好的行业进行投资
排除"四大搅屎棍"行业（银行、有色金属、钢铁、煤炭）

2. 选股逻辑

行业筛选：

使用申万一级行业分类
计算各行业在20日均线上方的股票比例
选择表现最佳的行业（g.num=1，即选择最好的一个行业）

个股筛选条件：

从中小板指数(399101.XSHE)成分股中选股
排除科创板、北交所、创业板股票
排除ST股和退市风险股
要求ROE > 15%，ROA > 10%
选择市值较小的股票（按市值升序排列）
排除次新股（上市时间超过375天）
排除停牌、涨停、跌停股票

3. 持仓管理机制

调仓频率：

每周进行一次主要调仓（周一9:30）

卖出规则：

不在新选股列表中的股票
非昨日涨停的股票
昨日涨停但今日涨停打开的股票（在14:00检查）

买入规则：

等分现金买入新选出的股票
控制总持仓数量为g.stock_num * g.num = 10只

4. 涨停股特殊处理

策略对涨停股有特殊的管理逻辑：

昨日涨停的股票享有"豁免权"，即使不在新选股列表中也不会被立即卖出
在14:00检查这些涨停股，如果涨停打开则卖出，否则继续持有

策略特点分析

优势：

行业轮动：跟随强势行业，捕捉行业轮动机会
质量筛选：通过财务指标筛选优质公司
小市值偏好：选择小市值股票，可能获得更高收益
风险控制：排除高风险行业和股票
涨停股管理：合理处理涨停股，避免错过连续涨停机会

策略逻辑流程

计算行业宽度 → 选择最佳行业 → 排除四大行业 → 财务筛选 → 市值筛选 → 技术过滤 → 构建组合 → 持仓管理 → 涨停股监控

这个策略结合了行业轮动、基本面分析和技术分析，通过系统化的方法在控制风险的同时追求超额收益。

下面的例子方便理解

核心目标

这个策略的目标很简单：找到市场上最热门的“好学生”（股票），买入并持有，一旦他们表现不好或者有更好的出现，就果断换掉。

第一步：挑“好班级”（选行业）

市场就像一所学校，里面有好多班级（行业），比如“科技班”、“医药班”、“白酒班”等等。

怎么看哪个班好？
- 策略会看每个班里，最近成绩在平均线以上的学生有多少。
- 如果一个班里大部分学生成绩都超过了平均线，说明这个班整体势头很旺，是个“好班级”。
排除“捣蛋鬼”班级（四大搅屎棍）

策略认为有四个班级（银行、有色金属、钢铁、煤炭）比较特殊，它们一发力，反而可能把整个市场的资金都吸走，导致其他班没表现。
所以，只要今天最好的班级不是这四个“捣蛋鬼”，我们就进去选学生。如果是它们，那我们今天就休息不玩了，以防被“搅局”。

第二步：在“好班级”里挑“好学生”（选股票）

现在我们已经锁定了一个最热门的班级，接下来要在这个班里挑出最有潜力的学生。

挑学生的标准非常严格：

基础要好： 要看学习成绩（财务数据），比如ROE（赚钱效率） 和ROA（资产利用率） 必须达到优秀线。
不能太张扬： 不选那些家里太有钱、名气太大的（不选大市值公司），偏爱那些还有潜力、规模还不算最大的（选小市值）。
不能有“问题”：
- 不能正在被老师罚站（不是停牌股）。
- 不能是戴着“ST”帽子的差生（不是ST风险股）。
- 不能是刚转学来的，还不熟悉情况（不是刚上市的新股）。
价格要合适：
今天价格不能已经“封死涨停”买不进去了。
今天价格也不能“跌停”动不了了。

通过以上层层筛选，最后会选出10个最符合条件的学生。

第三步：管理你的“学生队伍”（持仓管理）

你不是选完就完事了，还要每天管理他们。

每周调整一次：
- 每个星期一，重新执行第一、第二步，选出新的“好班级”和新的10个“好学生”。
- 卖掉你手里旧的、但不在新名单里的学生。
- 买入新名单里你还没有的学生。
对“尖子生”特殊照顾（处理涨停股）：

这是一个非常聪明的规则。如果一个学生昨天表现极好，涨停了（价格涨到最高限），那么他今天就享有“豁免权”。
即使今天的新名单里没有他，我们也不会立刻卖掉他，而是给他一个机会，观察到下午2点。
如果到下午2点，他还在涨停板上，我们就继续持有，希望他明天继续涨。
如果到下午2点，他的涨停板打开了（价格掉下来了），说明他后劲不足，我们就立刻卖掉他。

总结一下这个策略的精髓：

跟趋势： 永远跟着当前最热门的行业走。
避风头： 躲开那几个可能破坏市场节奏的“大家伙”行业。
选精英： 在热门行业里，用严格的标准挑选优质的小公司。
勤换血： 每周更新一次组合，保持队伍的先进性。
保利润： 对涨停的股票特殊处理，尽可能地让利润奔跑。

简单来说，这就是一个“谁热门就跟谁玩，但只跟班里踏实肯干的好学生玩，并且一旦发现苗头不对或者有更好的，立马换人”的策略。

源代码我全部上传了可以直接下载使用

不懂的问我就可以，加我备注入群可以加入量化群

# 克隆自聚宽文章：https://www.joinquant.com/post/49085
# 标题：四大搅屎棍策略
# 作者：MarioC


# 初始化函数
def initialize(context):
    # 设定基准
    set_benchmark('000985.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 将滑点设置为0
    set_slippage(FixedSlippage(0))
    # 设置交易成本万分之三，不同滑点影响可在归因分析中查看
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003,
                             close_today_commission=0, min_commission=5), type='stock')
    # 过滤order中低于error级别的日志
    log.set_level('order', 'error')
    # 初始化全局变量
    g.stock_num = 10
    g.hold_list = []  # 当前持仓的全部股票
    g.yesterday_HL_list = []  # 记录持仓中昨日涨停的股票
    g.num=1
    # 设置交易运行时间
    run_daily(prepare_stock_list, '9:05')
    run_weekly(weekly_adjustment, 1, '9:30')
    run_daily(check_limit_up, '14:00')  # 检查持仓中的涨停股是否需要卖出




SW1 = {
    '801010': '农林牧渔I',
    '801020': '采掘I',
    '801030': '化工I',
    '801040': '钢铁I',
    '801050': '有色金属I',
    '801060': '建筑建材I',
    '801070': '机械设备I',
    '801080': '电子I',
    '801090': '交运设备I',
    '801100': '信息设备I',
    '801110': '家用电器I',
    '801120': '食品饮料I',
    '801130': '纺织服装I',
    '801140': '轻工制造I',
    '801150': '医药生物I',
    '801160': '公用事业I',
    '801170': '交通运输I',
    '801180': '房地产I',
    '801190': '金融服务I',
    '801200': '商业贸易I',
    '801210': '休闲服务I',
    '801220': '信息服务I',
    '801230': '综合I',
    '801710': '建筑材料I',
    '801720': '建筑装饰I',
    '801730': '电气设备I',
    '801740': '国防军工I',
    '801750': '计算机I',
    '801760': '传媒I',
    '801770': '通信I',
    '801780': '银行I',
    '801790': '非银金融I',
    '801880': '汽车I',
    '801890': '机械设备I',
    '801950': '煤炭I',
    '801960': '石油石化I',
    '801970': '环保I',
    '801980': '美容护理I'
}


# 1-1 准备股票池
def prepare_stock_list(context):
    # 获取已持有列表
    g.hold_list = []
    for position in list(context.portfolio.positions.values()):
        stock = position.security
        g.hold_list.append(stock)
    # 获取昨日涨停列表
    if g.hold_list != []:
        df = get_price(g.hold_list, end_date=context.previous_date, frequency='daily', fields=['close', 'high_limit'],
                       count=1, panel=False, fill_paused=False)
        df = df[df['close'] == df['high_limit']]
        g.yesterday_HL_list = list(df.code)
    else:
        g.yesterday_HL_list = []


industry_code = ['801010','801020','801030','801040','801050','801080','801110','801120','801130','801140','801150',\
                    '801160','801170','801180','801200','801210','801230','801710','801720','801730','801740','801750',\
                   '801760','801770','801780','801790','801880','801890']


def industry(stockList,industry_code,date):
    i_Constituent_Stocks={}
    for i in industry_code:
        temp = get_industry_stocks(i, date)
        i_Constituent_Stocks[i] = list(set(temp).intersection(set(stockList)))
    count_dict = {}
    for name, content_list in i_Constituent_Stocks.items():
        count = len(content_list)
        count_dict[name] = count
    return count_dict


def getStockIndustry(p_stocks, p_industries_type, p_day):
    dict_stk_2_ind = {}
    stocks_industry_dict = get_industry(p_stocks, date=p_day)
    for stock in stocks_industry_dict:
        if p_industries_type in stocks_industry_dict[stock]:
            dict_stk_2_ind[stock] = stocks_industry_dict[stock][p_industries_type]['industry_code']
    return pd.Series(dict_stk_2_ind)
# 1-2 选股模块
def get_stock_list(context):
    # 指定日期防止未来数据
    yesterday = context.previous_date
    today = context.current_dt
    final_list =[]
    # 获取初始列表
    initial_list = get_index_stocks('000985.XSHG', today)
    p_count=1
    p_industries_type='sw_l1'
    h = get_price(initial_list, end_date=yesterday, frequency='1d', fields=['close'], count=p_count + 20, panel=False)
    h['date'] = pd.DatetimeIndex(h.time).date
    df_close = h.pivot(index='code', columns='date', values='close').dropna(axis=0)
    df_ma20 = df_close.rolling(window=20, axis=1).mean().iloc[:, -p_count:]
    df_bias = (df_close.iloc[:, -p_count:] > df_ma20) 
    s_stk_2_ind = getStockIndustry(p_stocks=initial_list, p_industries_type=p_industries_type, p_day=yesterday)
    df_bias['industry_code'] = s_stk_2_ind
    df_ratio = ((df_bias.groupby('industry_code').sum() * 100.0) / df_bias.groupby(
        'industry_code').count()).round()  
    column_names = df_ratio.columns.tolist()
    top_values = df_ratio[datetime.date(yesterday.year, yesterday.month, yesterday.day)].nlargest(g.num)
    I   =  top_values.index.tolist()
    sum_of_top_values = df_ratio.sum()
    TT = sum_of_top_values[datetime.date(yesterday.year, yesterday.month, yesterday.day)]
    name_list = [SW1[code] for code in I]
    print(name_list)
    print('全市场宽度：',np.array(df_ratio.sum(axis=0).mean()))
    if '801780' not in I and '801050' not in I and '801950' not in I and '801040' not in I:
        #《银行、有色金属、钢铁、煤炭》搅屎棍不在，开仓
        S_stocks = get_index_stocks('399101.XSHE', today)
        stocks = filter_kcbj_stock(S_stocks)
        choice = filter_st_stock(stocks)
        choice = filter_new_stock(context, choice)
        BIG_stock_list = get_fundamentals(query(
                valuation.code,
            ).filter(
                valuation.code.in_(choice),
                indicator.roe > 0.15,
                indicator.roa > 0.10,
            ).order_by(
        valuation.market_cap.asc()).limit(g.stock_num)).set_index('code').index.tolist()
        BIG_stock_list = filter_paused_stock(BIG_stock_list)
        BIG_stock_list = filter_limitup_stock(context,BIG_stock_list)
        L = filter_limitdown_stock(context,BIG_stock_list)
    else:
        print('跑')
        L=[]
    return L


# 1-3 整体调整持仓
def weekly_adjustment(context):
    target_B = get_stock_list(context)
    # 调仓卖出
    for stock in g.hold_list:
        if (stock not in target_B) and (stock not in g.yesterday_HL_list):
            position = context.portfolio.positions[stock]
            close_position(position)
    position_count = len(context.portfolio.positions)
    target_num = len(target_B)
    if target_num > position_count:
        buy_num = min(len(target_B), g.stock_num*g.num - position_count)
        value = context.portfolio.cash / buy_num
        for stock in target_B:
            if stock not in list(context.portfolio.positions.keys()):
                if open_position(stock, value):
                    if len(context.portfolio.positions) == target_num:
                        break


def check_limit_up(context):
    now_time = context.current_dt
    if g.yesterday_HL_list != []:
        # 对昨日涨停股票观察到尾盘如不涨停则提前卖出，如果涨停即使不在应买入列表仍暂时持有
        for stock in g.yesterday_HL_list:
            current_data = get_price(stock, end_date=now_time, frequency='1m', fields=['close', 'high_limit'],
                                     skip_paused=False, fq='pre', count=1, panel=False, fill_paused=True)
            if current_data.iloc[0, 0] < current_data.iloc[0, 1]:
                log.info("[%s]涨停打开，卖出" % (stock))
                position = context.portfolio.positions[stock]
                close_position(position)
            else:
                log.info("[%s]涨停，继续持有" % (stock))


# 3-1 交易模块-自定义下单
def order_target_value_(security, value):
    if value == 0:
        log.debug("Selling out %s" % (security))
    else:
        log.debug("Order %s to value %f" % (security, value))
    return order_target_value(security, value)




# 3-2 交易模块-开仓
def open_position(security, value):
    order = order_target_value_(security, value)
    if order != None and order.filled > 0:
        return True
    return False




# 3-3 交易模块-平仓
def close_position(position):
    security = position.security
    order = order_target_value_(security, 0)  # 可能会因停牌失败
    if order != None:
        if order.status == OrderStatus.held and order.filled == order.amount:
            return True
    return False




# 2-1 过滤停牌股票
def filter_paused_stock(stock_list):
    current_data = get_current_data()
    return [stock for stock in stock_list if not current_data[stock].paused]




# 2-2 过滤ST及其他具有退市标签的股票
def filter_st_stock(stock_list):
    current_data = get_current_data()
    return [stock for stock in stock_list
            if not current_data[stock].is_st
            and 'ST' not in current_data[stock].name
            and '*' not in current_data[stock].name
            and '退' not in current_data[stock].name]




# 2-3 过滤科创北交股票
def filter_kcbj_stock(stock_list):
    for stock in stock_list[:]:
        if stock[0] == '4' or stock[0] == '8' or stock[:2] == '68' or stock[0] == '3':
            stock_list.remove(stock)
    return stock_list




# 2-4 过滤涨停的股票
def filter_limitup_stock(context, stock_list):
    last_prices = history(1, unit='1m', field='close', security_list=stock_list)
    current_data = get_current_data()
    return [stock for stock in stock_list if stock in context.portfolio.positions.keys()
            or last_prices[stock][-1] < current_data[stock].high_limit]




# 2-5 过滤跌停的股票
def filter_limitdown_stock(context, stock_list):
    last_prices = history(1, unit='1m', field='close', security_list=stock_list)
    current_data = get_current_data()
    return [stock for stock in stock_list if stock in context.portfolio.positions.keys()
            or last_prices[stock][-1] > current_data[stock].low_limit]




# 2-6 过滤次新股
def filter_new_stock(context, stock_list):
    yesterday = context.previous_date
    return [stock for stock in stock_list if
            not yesterday - get_security_info(stock).start_date < datetime.timedelta(days=375)]