small_cap.py

# 标题：小市值轮动v1.0：每5天调仓，回撤可控

# 导入聚宽函数库
from jqdata import *
import pandas as pd
import numpy as np
from datetime import timedelta

def initialize(context):
    """
    初始化函数：小市值轮动策略
    逻辑：每5个交易日，买入主要指数成分股中流通市值最小的5只股票，等权持有
    """
    # 设置基准
    set_benchmark('000905.XSHG')  # 中证500
    # 使用真实价格
    set_option('use_real_price', True)
    # 设置滑点（千分之一）
    set_slippage(PriceRelatedSlippage(0.001))
    # 设置手续费：买入万三，卖出万三+千一印花税
    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')
    
    # 策略参数
    g.stock_num = 5                      # 持仓数量：5只
    g.rebalance_days = 5                 # 调仓周期：5个交易日
    g.last_trade_date = None             # 上次调仓日期
    
    # 定时任务
    run_daily(trade, time='10:00')
    run_daily(record_holdings, time='15:05')


def get_target_stocks(context):
    """
    获取目标股票：流通市值最小的5只股票
    过滤条件：ST、停牌、PE>0、上市满365天、负债率<70%、ROE>5%、经营现金流>0
    """
    current_date = context.current_dt.date()
    
    # ========== 第一步：获取主要指数成分股 ==========
    index_stocks = set()
    indexes = [
        "000300.XSHG",  # 沪深300
        "000905.XSHG",  # 中证500
        "000852.XSHG",  # 中证1000
        "000001.XSHG",  # 上证指数
        "399001.XSHE",  # 深证成指
        "399006.XSHE",  # 创业板指
        "000016.XSHG",  # 上证50
        "000688.XSHG",  # 科创50
        "399330.XSHE",  # 深证100
    ]
    
    for index_code in indexes:
        try:
            stocks = get_index_stocks(index_code, date=current_date)
            index_stocks.update(stocks)
        except:
            continue
    
    # 获取当前数据用于过滤
    current_data = get_current_data()
    
    valid_stocks = []
    for stock in index_stocks:
        try:
            # 过滤ST股
            if current_data[stock].is_st:
                continue
            if 'ST' in current_data[stock].name or '*' in current_data[stock].name:
                continue
            
            # 过滤停牌
            if current_data[stock].paused:
                continue
            
            valid_stocks.append(stock)
        except:
            continue
    
    log.info(f"指数成分股筛选后: {len(valid_stocks)}只")
    
    if len(valid_stocks) < g.stock_num:
        return []
    
    # ========== 第二步：获取基本面数据 ==========
    # 聚宽 valuation 表常用字段：
    # - pe_ratio: 市盈率
    # - circulating_market_cap: 流通市值
    # - market_cap: 总市值
    # indicator 表字段：
    # - roe: 净资产收益率
    # balance 表字段：
    # - total_assets: 总资产
    # - total_liability: 总负债
    # cash_flow 表字段：
    # - net_operate_cash_flow: 经营活动现金流

    q = query(
        valuation.code,
        valuation.circulating_market_cap,  # 流通市值
        valuation.pe_ratio,                # 市盈率
        indicator.roe,                     # 净资产收益率
        balance.total_assets,              # 总资产
        balance.total_liability,           # 总负债
        cash_flow.net_operate_cash_flow    # 经营现金流
    ).filter(
        valuation.code.in_(valid_stocks)
    )
    df = get_fundamentals(q, date=current_date)
    
    if df.empty:
        log.info("获取基本面数据失败")
        return []
    
    log.info(f"获取基本面数据 {len(df)}只")
    
    # 应用过滤条件：市盈率为正
    df = df[df['pe_ratio'] > 0]

    log.info(f"PE>0: {len(df)}只")

    # 计算资产负债率
    df['debt_ratio'] = df['total_liability'] / df['total_assets'] * 100

    # 财务风险过滤
    # 1. 资产负债率 < 70%（排除高负债公司）
    df = df[df['debt_ratio'] < 70]
    log.info(f"资产负债率<70%: {len(df)}只")

    # 2. ROE > 5%（排除盈利能力差的公司）
    df = df[df['roe'] > 5]
    log.info(f"ROE>5%: {len(df)}只")

    # 3. 经营现金流 > 0（排除现金流为负的公司）
    df = df[df['net_operate_cash_flow'] > 0]
    log.info(f"经营现金流>0: {len(df)}只")
    
    if len(df) < g.stock_num:
        return []
    
    # 计算上市天数（通过股票信息）
    listed_days_list = []
    for stock in df['code'].tolist():
        try:
            info = get_security_info(stock)
            days = (current_date - info.start_date).days
            listed_days_list.append(days)
        except:
            listed_days_list.append(0)
    
    df['list_days'] = listed_days_list
    
    # 上市满365天
    df = df[df['list_days'] > 365]
    
    log.info(f"上市满365天: {len(df)}只")
    
    if len(df) < g.stock_num:
        return []
    
    # 按流通市值排序（从小到大）
    df = df.sort_values('circulating_market_cap')
    
    # 取市值最小的3只
    target_stocks = df['code'].head(g.stock_num).tolist()
    
    # 打印目标股票信息
    log.info(f"目标股票（市值最小{g.stock_num}只）:")
    for i, (idx, row) in enumerate(df.head(g.stock_num).iterrows()):
        try:
            name = get_security_info(row['code']).display_name
            log.info(f"  {i+1}. {row['code']} {name}, "
                    f"流通市值: {row['circulating_market_cap']/1e8:.2f}亿, "
                    f"PE: {row['pe_ratio']:.2f}, "
                    f"负债率: {row['debt_ratio']:.1f}%, "
                    f"ROE: {row['roe']:.1f}%, "
                    f"上市天数: {row['list_days']}")
        except:
            log.info(f"  {i+1}. {row['code']}, 流通市值: {row['circulating_market_cap']/1e8:.2f}亿")
    
    return target_stocks


def trade(context):
    """
    交易函数：每5个交易日调仓一次
    """
    current_date = context.current_dt.date()
    
    # 检查是否需要调仓
    if g.last_trade_date is None:
        need_rebalance = True
    else:
        trade_days = get_trade_days(start_date=g.last_trade_date, end_date=current_date)
        need_rebalance = len(trade_days) >= g.rebalance_days
    
    if not need_rebalance:
        return
    
    log.info(f"【调仓日】{current_date}，开始执行调仓")
    
    # 获取目标持仓
    target_stocks = get_target_stocks(context)
    
    if not target_stocks:
        log.info("无目标股票，跳过调仓")
        return
    
    # 获取当前持仓
    current_holdings = [s for s, p in context.portfolio.positions.items() if p.total_amount > 0]
    
    # 卖出不在目标列表中的股票
    to_sell = [s for s in current_holdings if s not in target_stocks]
    for stock in to_sell:
        order_target_value(stock, 0)
        log.info(f"卖出: {stock}")
    
    # 买入新目标（等权分配）
    to_buy = [s for s in target_stocks if s not in current_holdings]
    if to_buy:
        cash_per_stock = context.portfolio.available_cash / len(to_buy)
        for stock in to_buy:
            current_price = get_current_data()[stock].last_price
            # 确保至少买入100股
            if cash_per_stock >= current_price * 100:
                order_target_value(stock, cash_per_stock)
                log.info(f"买入: {stock}, 金额: {cash_per_stock:.2f}")
            else:
                log.info(f"资金不足，跳过买入: {stock}")
    
    # 更新调仓日期
    g.last_trade_date = current_date
    log.info(f"调仓完成，目标持仓: {target_stocks}")


def record_holdings(context):
    """记录持仓信息"""
    positions = context.portfolio.positions
    current_holdings = [s for s in positions if positions[s].total_amount > 0]
    
    log.info("=" * 60)
    log.info(f"【{context.current_dt.date()}】持仓情况")
    log.info(f"持仓数量: {len(current_holdings)}/{g.stock_num}支")
    log.info(f"总资产: {context.portfolio.portfolio_value:.2f}")
    
    if len(current_holdings) > 0:
        for stock in current_holdings:
            pos = positions[stock]
            profit = (pos.price - pos.avg_cost) / pos.avg_cost * 100 if pos.avg_cost > 0 else 0
            try:
                name = get_security_info(stock).display_name
                log.info(f"  {stock}({name}): 数量{pos.total_amount}, "
                        f"成本{pos.avg_cost:.2f}, 现价{pos.price:.2f}, "
                        f"盈亏{profit:.2f}%")
            except:
                log.info(f"  {stock}: 数量{pos.total_amount}, "
                        f"成本{pos.avg_cost:.2f}, 现价{pos.price:.2f}, "
                        f"盈亏{profit:.2f}%")
    else:
        log.info("当前无持仓")
    log.info("=" * 60)