2502 多因子复合策略组合 5 个子策略融和 加持大市值国九条 54686

策略名：多因子复合策略组合

核心逻辑：

1. 策略组合架构：

 策略权重分配
"搅屎棍策略(小盘反转)" : 20
"全天候ETF策略(大类资产)" : 20
"国九条策略(政策红利)" : 20
"大市值价值策略(蓝筹)" : 20
"小市值策略(成长)" : 20

2. 动态调仓机制：

• 每月 / 周固定时点调仓（避开月初流动性紧张期）
• 子策略独立运行，互不干扰
• 剩余现金自动配置货币 ETF（511880）

3. 风险控制体系：

graph TD
    A[个股层面] --> B[涨停板管理]
    A --> C[7%硬止损]
    D[组合层面] --> E[大类资产对冲]
    D --> F[1/4月强制空仓]
    G[市场层面] --> H[中小板指数熔断]

子策略详解：

1. 搅屎棍策略：

• 类型：小盘反转策略
• 选股：中小板市值最小 20 只 + 净利润为正
• 择时：银行 / 煤炭 / 钢铁行业过热时空仓
• 频率：周调仓

2. 全天候 ETF 策略：

• 配置：国债 (45%)+ 黄金 (25%)+ 纳斯达克 (15%)+ 商品 (15%)
• 特点：抗通胀 + 跨市场对冲
• 调仓：月频再平衡

3. 国九条策略：

• 选股：符合分红要求的中小市值股
• 过滤：剔除审计不合格股票
• 特殊规则：1/4 月空仓避财报季

4. 大市值价值策略：

• 选股：PB<1+ROA>15% 的蓝筹股
• 持有期：中长期投资
• 止损：涨停打开卖出

5. 小市值策略：

• 选股：5-50 亿市值成长股
• 交易：7% 止损 + 100% 止盈
• 风控：市场大跌 5% 全仓止损

关键优化点：

1. 订单执行优化：

• 分笔下单（整手数）
• 涨跌停过滤
• 0.02% 滑点补偿

2. 资金效率提升：

• 货币 ETF 日内回转
• 动态现金管理
• 佣金优化（万 0.3）

3. 风险平价设计：

    title 风险贡献分布
    "权益类" : 40
    "固收类" : 30
    "商品类" : 20
    "现金类" : 10

4. 监控体系：

• 每日交易日志
• 持仓市值实时跟踪
• 策略绩效归因

特殊处理机制：

1. 极端行情应对：

• 中小板指数单日跌 5% 触发全仓止损
• 个股涨停板特殊管理（持有至次日观察）
• 流动性危机模式（自动切换国债 ETF）

2. 日历效应：

• 1/4 月强制降仓
• 财报季审计强化过滤
• 分红季红利因子增强

3. 调整：

• 国九条财务指标过滤
• 剔除 ST/*ST / 退市股
• 规避次新股炒作

# 导入函数库
from jqdata import *
from jqfactor import get_factor_values
import datetime
import math
from scipy.optimize import minimize


# 初始化函数，设定基准等等
def initialize(context):
    # 设定沪深300作为基准
    # set_benchmark("515080.XSHG")
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 开启动态复权模式(真实价格)
    set_option("use_real_price", True)
    # 输出内容到日志 log.info()
    log.info("初始函数开始运行且全局只运行一次")
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level("order", "error")
    # 固定滑点设置ETF 0.001(即交易对手方一档价)
    set_slippage(FixedSlippage(0.002), type="fund")
    # 股票交易总成本0.3%(含固定滑点0.02)
    set_slippage(FixedSlippage(0.02), type="stock")
    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",
    )
    # 设置货币ETF交易佣金0
    set_order_cost(
        OrderCost(
            open_tax=0,
            close_tax=0,
            open_commission=0,
            close_commission=0,
            close_today_commission=0,
            min_commission=0,
        ),
        type="mmf",
    )
    # 全局变量
    g.fill_stock = "511880.XSHG"  # 货币ETF,用于现金管理
    g.strategys = {}
    
    #g.portfolio_value_proportion = [0.0,  0.0, 0.0,1, 0.0]
    g.portfolio_value_proportion = [0.2, 0.2, 0.2, 0.2, 0.2]  # 修改权重分配，增加小市值策略
    g.positions = {i: {} for i in range(len(g.portfolio_value_proportion))}  # 记录每个子策
    # 子策略执行计划
    if g.portfolio_value_proportion[0] > 0:
        run_weekly(jsg_adjust, 1, "9:31")
        run_daily(jsg_check, "14:50")
    if g.portfolio_value_proportion[1] > 0:
        run_monthly(all_day_adjust, 1, "9:32")
    if g.portfolio_value_proportion[2] > 0:
        run_weekly(guojiutiao_adjust, 2, "9:35")  # 每周二调仓
        run_daily(guojiutiao_check, "14:52") 
    if g.portfolio_value_proportion[3] > 0:  # 新增大市值价值投资策略
        run_monthly(large_cap_value_adjust, 1, "9:33")  # 每月第一个交易日调仓
        run_daily(large_cap_value_check, "14:53")  # 每日检查涨停
    if g.portfolio_value_proportion[4] > 0:  # 新增小市值策略
        run_weekly(small_cap_adjust, 2, "9:35")  # 每周二调仓
        run_daily(small_cap_check, "14:52")  # 每日检查止损
    # 每日剩余资金购买货币ETF
    run_daily(end_trade, "14:55")
    run_daily(summary_report, '15:15')

def process_initialize(context):
    print("重启程序")
    g.strategys["搅屎棍策略"] = JSG_Strategy(context, index=0, name="搅屎棍策略")
    g.strategys["全天候策略"] = All_Day_Strategy(context, index=1, name="全天候策略")
    g.strategys["国九条策略"] = GuoJiuTiao_Strategy(context, index=2, name="国九条策略")
    g.strategys["大市值价值策略"] = Large_Cap_Value_Strategy(context, index=3, name="大市值价值策略")  # 新增大市值价值策略
    g.strategys["小市值策略"] = Small_Cap_Strategy(context, index=4, name="小市值策略")  # 新增小市值策略

# 买入货币ETF
# 尾盘处理
def end_trade(context):
    current_data = get_current_data()

    # 卖出未记录的股票（比如送股）
    keys = [key for d in g.positions.values() if isinstance(d, dict) for key in d.keys()]
    for stock in context.portfolio.positions:
        if stock not in keys and stock != g.fill_stock and current_data[stock].last_price < current_data[stock].high_limit:
            if order_target_value(stock, 0):
                log.info(f"卖出{stock}因送股未记录在持仓中")

    # 买入货币ETF
    amount = int(context.portfolio.available_cash / current_data[g.fill_stock].last_price)
    if amount >= 100:
        order(g.fill_stock, amount)


# 卖出货币ETF换现金
def get_cash(context, value):
    if g.fill_stock not in context.portfolio.positions:
        return
    current_data = get_current_data()
    amount = math.ceil(value / current_data[g.fill_stock].last_price / 100) * 100
    position = context.portfolio.positions[g.fill_stock].closeable_amount
    if amount >= 100:
        order(g.fill_stock, -min(amount, position))


def jsg_check(context):
    g.strategys["搅屎棍策略"].check()


def jsg_adjust(context):
    g.strategys["搅屎棍策略"].adjust()


def all_day_adjust(context):
    g.strategys["全天候策略"].adjust()


def guojiutiao_adjust(context):
    g.strategys["国九条策略"].adjust()

def guojiutiao_check(context):
    g.strategys["国九条策略"].check()


def large_cap_value_adjust(context):
    g.strategys["大市值价值策略"].adjust()

def large_cap_value_check(context):
    g.strategys["大市值价值策略"].check()

def small_cap_adjust(context):
    g.strategys["小市值策略"].adjust()

def small_cap_check(context):
    g.strategys["小市值策略"].check()

# 策略基类
class Strategy:

    def __init__(self, context, index, name):
        self.context = context
        self.index = index
        self.name = name
        self.stock_sum = 1
        self.hold_list = []
        self.min_volume = 2000
        self.trade_log = []  # 新增交易日志
        self.position_value = 0  # 持仓市值

    # 获取策略当前持仓市值
    def get_total_value(self):
        if not g.positions[self.index]:
            return 0
        return sum(self.context.portfolio.positions[key].price * value for key, value in g.positions[self.index].items())

    # 检查昨日涨停票
    def _check(self):
        # 获取已持有列表
        self.hold_list = list(g.positions[self.index].keys())
        # 获取昨日涨停列表
        if self.hold_list != []:
            df = get_price(
                self.hold_list,
                end_date=self.context.previous_date,
                frequency="daily",
                fields=["close", "high_limit"],
                count=1,
                panel=False,
                fill_paused=False,
            )
            df = df[df["close"] == df["high_limit"]]
            return list(df.code)
        return []

    # 调仓(等权购买target中按顺序排列固定数量的的标的)
    def _adjust(self, target):

        # 获取前stock_sum个标的
        target = target[: min(len(target), self.stock_sum)]

        # 获取已持有列表
        self.hold_list = list(g.positions[self.index].keys())
        portfolio = self.context.portfolio

        # 调仓卖出
        for stock in self.hold_list:
            if stock not in target:
                self.order_target_value_(stock, 0)

        # 调仓买入
        count = len(set(target) - set(self.hold_list))
        if count == 0 or self.stock_sum <= len(self.hold_list):
            return

        # 目标市值
        target_value = portfolio.total_value * g.portfolio_value_proportion[self.index]

        # 当前市值
        position_value = self.get_total_value()

        # 可用现金:当前现金 + 货币ETF市值
        available_cash = portfolio.available_cash + (portfolio.positions[g.fill_stock].value if g.fill_stock in portfolio.positions else 0)

        # 买入股票的总市值
        value = max(0, min(target_value - position_value, available_cash))

        # 卖出部分货币ETF获取现金
        if value > portfolio.available_cash:
            get_cash(self.context, value - portfolio.available_cash)

        # 等价值买入每一个未买入的标的
        for security in target:
            if security not in self.hold_list:
                self.order_target_value_(security, value / count)

    # 调仓2(targets为字典，key为股票代码，value为目标市值)
    def _adjust2(self, targets):

        # 获取已持有列表
        self.hold_list = list(g.positions[self.index].keys())
        current_data = get_current_data()
        portfolio = self.context.portfolio

        # 清仓被调出的
        for stock in self.hold_list:
            if stock not in targets:
                self.order_target_value_(stock, 0)

        # 先卖出
        for stock, target in targets.items():
            price = current_data[stock].last_price
            value = g.positions[self.index].get(stock, 0) * price
            if value - target > self.min_volume and value - target > price * 100:
                self.order_target_value_(stock, target)

        # 后买入
        for stock, target in targets.items():
            price = current_data[stock].last_price
            value = g.positions[self.index].get(stock, 0) * price
            if target - value > self.min_volume and target - value > price * 100:
                if target - value > portfolio.available_cash:
                    get_cash(self.context, target - value - portfolio.available_cash)
                if portfolio.available_cash > price * 100:
                    self.order_target_value_(stock, target)

    # 自定义下单(涨跌停不交易)
    # 自定义下单(涨跌停不交易)
    def order_target_value_(self, security, value):
        current_data = get_current_data()
        
        # 检查标的是否停牌、涨停、跌停
        if current_data[security].paused:
            log.info(f"{security}: 今日停牌")
            return False

        # 检查是否涨停
        if current_data[security].last_price == current_data[security].high_limit:
            log.info(f"{security}: 当前涨停")
            return False

        # 检查是否跌停
        if current_data[security].last_price == current_data[security].low_limit:
            log.info(f"{security}: 当前跌停")
            return False

        # 获取当前标的的价格
        price = current_data[security].last_price

        # 获取当前策略的持仓数量
        current_position = g.positions[self.index].get(security, 0)

        # 计算目标持仓数量
        target_position = (int(value / price) // 100) * 100 if price != 0 else 0

        # 计算需要调整的数量
        adjustment = target_position - current_position

        # 检查是否当天买入卖出
        closeable_amount = self.context.portfolio.positions[security].closeable_amount if security in self.context.portfolio.positions else 0
        if adjustment < 0 and closeable_amount == 0:
            log.info(f"{security}: 当天买入不可卖出")
            return False

        # 下单并更新持仓
        if adjustment != 0:
            o = order(security, adjustment)
            if o:
                # 记录交易明细
                trade_type = "买入" if o.is_buy else "卖出"
                self.trade_log.append({
                    'date': self.context.current_dt,
                    'symbol': security,
                    'amount': o.amount,
                    'price': o.price,
                    'type': trade_type,
                    'commission': o.commission
                })
                # 更新持仓数量
                amount = o.amount if o.is_buy else -o.amount
                g.positions[self.index][security] = amount + current_position
                # 如果目标持仓为零，移除该证券
                if target_position == 0:
                    g.positions[self.index].pop(security, None)
                # 更新持有列表
                self.hold_list = list(g.positions[self.index].keys())
                return True
        return False

        # 获取当前标的的价格
        price = current_data[security].last_price

        # 获取当前策略的持仓数量
        current_position = g.positions[self.index].get(security, 0)

        # 计算目标持仓数量
        target_position = (int(value / price) // 100) * 100 if price != 0 else 0

        # 计算需要调整的数量
        adjustment = target_position - current_position

        # 检查是否当天买入卖出
        closeable_amount = self.context.portfolio.positions[security].closeable_amount if security in self.context.portfolio.positions else 0
        if adjustment < 0 and closeable_amount == 0:
            log.info(f"{security}: 当天买入不可卖出")
            return False

        # 下单并更新持仓
        if adjustment != 0:
            o = order(security, adjustment)
            if o:
                # 更新持仓数量
                amount = o.amount if o.is_buy else -o.amount
                g.positions[self.index][security] = amount + current_position
                # 如果目标持仓为零，移除该证券
                if target_position == 0:
                    g.positions[self.index].pop(security, None)
                # 更新持有列表
                self.hold_list = list(g.positions[self.index].keys())
                return True
        return False

    # 基础过滤(过滤科创北交、ST、停牌、次新股)
    def filter_basic_stock(self, stock_list):

        current_data = get_current_data()
        return [
            stock
            for stock in stock_list
            if not current_data[stock].paused
            and 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
            and not (stock[0] == "4" or stock[0] == "8" or stock[:2] == "68")
            and not self.context.previous_date - get_security_info(stock).start_date < datetime.timedelta(375)
        ]

    # 过滤当前时间涨跌停的股票
    def filter_limitup_limitdown_stock(self, stock_list):
        current_data = get_current_data()
        return [
            stock
            for stock in stock_list
            if current_data[stock].last_price < current_data[stock].high_limit and current_data[stock].last_price > current_data[stock].low_limit
        ]

    # 判断今天是在空仓月
    def is_empty_month(self):
        month = self.context.current_dt.month
        return month in self.pass_months

# 新增收盘总结函数
def summary_report(context):
    log.info("\n====== 每日策略总结 ======")
    total_turnover = 0
    
    for strategy_name, strategy in g.strategys.items():
        # 计算策略持仓市值
        position_value = sum([p.value for p in context.portfolio.positions.values() 
                            if p.security in g.positions[strategy.index]])
        
        # 输出策略概况
        log.info(f"【{strategy_name}】")
        log.info(f"持仓市值: {position_value:.2f}元")
        log.info(f"当日交易明细（{len(strategy.trade_log)}笔）:")
        
        # 输出每笔交易详情
        strategy_turnover = 0
        for trade in strategy.trade_log:
            trade_value = trade['amount'] * trade['price']
            strategy_turnover += trade_value
            log.info(f"{trade['date']} {trade['type']} {trade['symbol']} "
                    f"数量:{trade['amount']} 价格:{trade['price']:.2f} "
                    f"金额:{trade_value:.2f}元 手续费:{trade['commission']:.2f}元")
        
        total_turnover += strategy_turnover
        log.info(f"策略成交总额: {strategy_turnover:.2f}元\n")
        strategy.trade_log = []  # 清空当日日志

    # 输出汇总信息
    log.info("=== 全局汇总 ===")
    log.info(f"总资产净值: {context.portfolio.total_value:.2f}元")
    log.info(f"现金余额: {context.portfolio.available_cash:.2f}元")
    log.info(f"当日总成交额: {total_turnover:.2f}元")
    log.info(f"持仓股票数量: {sum(len(p) for p in g.positions.values())}只")
    log.info("================\n")

# 搅屎棍策略
class JSG_Strategy(Strategy):

    def __init__(self, context, index, name):
        super().__init__(context, index, name)

        self.stock_sum = 6
        # 判断买卖点的行业数量
        self.num = 1
        # 空仓的月份
        self.pass_months = [1, 4]

    def getStockIndustry(self, stocks):
        industry = get_industry(stocks)
        return pd.Series({stock: info["sw_l1"]["industry_name"] for stock, info in industry.items() if "sw_l1" in info})

    # 获取市场宽度
    def get_market_breadth(self):
        # 指定日期防止未来数据
        yesterday = self.context.previous_date
        # 获取初始列表
        stocks = get_index_stocks("000985.XSHG")
        count = 1
        h = get_price(
            stocks,
            end_date=yesterday,
            frequency="1d",
            fields=["close"],
            count=count + 20,
            panel=False,
        )
        h["date"] = pd.DatetimeIndex(h.time).date
        df_close = h.pivot(index="code", columns="date", values="close").dropna(axis=0)
        # 计算20日均线
        df_ma20 = df_close.rolling(window=20, axis=1).mean().iloc[:, -count:]
        # 计算偏离程度
        df_bias = df_close.iloc[:, -count:] > df_ma20
        df_bias["industry_name"] = self.getStockIndustry(stocks)
        # 计算行业偏离比例
        df_ratio = ((df_bias.groupby("industry_name").sum() * 100.0) / df_bias.groupby("industry_name").count()).round()
        # 获取偏离程度最高的行业
        top_values = df_ratio.loc[:, yesterday].nlargest(self.num)
        I = top_values.index.tolist()
        return I

    # 过滤股票
    def filter(self):
        stocks = get_index_stocks("399101.XSHE")
        # stocks = get_all_securities("stock", date=self.context.previous_date).index.tolist()
        stocks = self.filter_basic_stock(stocks)
        stocks = (
            get_fundamentals(
                query(
                    valuation.code,
                )
                .filter(
                    valuation.code.in_(stocks),
                    indicator.adjusted_profit > 0,
                )
                .order_by(valuation.market_cap.asc())
            )
            .head(20)
            .code
        )
        stocks = self.filter_limitup_limitdown_stock(stocks)
        return stocks

    # 择时
    def select(self):
        I = self.get_market_breadth()
        industries = {"银行I", "煤炭I", "采掘I", "钢铁I"}
        if not industries.intersection(I) and not self.is_empty_month():
            return self.filter()
        return []

    # 调仓
    def adjust(self):
        self._adjust(self.select())

    # 获取昨日涨停票
    def check(self):
        banner_stocks = self._check()
        for stock in banner_stocks:
            self.order_target_value_(stock, 0)

全天候ETF策略 国九条策略 大市值价值投资策略 小市值策略 解锁后查看：