该策略基于市值因子进行股票轮动，通过定期调整持仓至市值较小的股票，利用小市值效应获取超额收益。策略基准为沪深 300 指数 (000300.XSHG)，通过筛选全市场股票构建投资组合。
- 市值因子轮动：定期买入市值较小的股票，卖出市值较大的股票
- 全市场选股：从全市场股票中筛选，不限于特定板块或指数
- 低频交易：每日 14:50 进行调仓，降低交易频率
- 风险控制：过滤 ST 股、退市股和停牌股
- 等权重配置：对选中的股票进行等权重资金分配
-
选股规则：
- 从全市场股票中筛选
- 过滤 ST 股票和退市股票
- 过滤停牌股票
- 选择市值最小的 400 只股票
-
买入规则：
- 每日 14:50 进行调仓
- 对选中的 400 只股票进行等权重配置
- 使用可用现金平均分配买入新入选股票
-
卖出规则：
- 卖出不在目标列表中的所有持仓股票
- 每日调仓时一次性完成买卖操作
-
资金管理：
- 使用真实价格交易
- 交易佣金为 0.03%(买卖均收取)
- 印花税 0.1%(卖出收取)
- 最低佣金 5 元
- 过滤高风险股票 (ST 股、退市股)
- 过滤停牌股票，避免流动性风险
- 分散投资于 400 只股票，降低个股风险
- 使用市值因子筛选，避免过度集中于大盘股
该策略通过市值因子轮动和严格的风险过滤，旨在获取小市值股票的超额收益，同时通过分散投资和低频交易控制风险。策略基准为沪深 300 指数，适合作为增强型指数投资策略。
# 标题：万得微盘复制策略
# 复制的万得微盘编制标准，为什么实际收益会比指数略低，同期万得微盘指数上涨260%，策略上涨220%
# 第一点，400只股票，一千万资金不够。
# 第二点，头寸不对，所有股票等权。
# 第三点，缺乏再平衡。无摩擦日间再平衡，才能匹配万得微盘指数。
# 原本代码中是可用金额平均分给新买入的股票，这个跟万得微盘指数要求的等权买入不太符合。 代码需要改成等权买入
from jqdata import *
import numpy as np
def initialize(context):
set_benchmark('000300.XSHG')
set_option('use_real_price', True)
set_option("avoid_future_data", True)
set_order_cost(OrderCost(open_tax=0, close_tax=0.001,
open_commission=0.0003, close_commission=0.0003,
min_commission=5), type='stock')
# 万得微盘指数参数
g.component_num = 400 # 成分股数量
g.rebalance_month = [1, 4, 7, 10] # 季度调仓(1/4/7/10月)
g.trade_time = '14:50' # 交易时间
# 每月初检查是否需要调仓
run_daily(check_rebalance, g.trade_time)
#run_monthly(check_rebalance, -1, time=g.trade_time)
# run_daily(trade, g.trade_time)
def check_rebalance(context):
"""检查是否到达调仓月份"""
current_month = context.current_dt.month
#if current_month in g.rebalance_month:
trade(context)
def trade(context):
"""执行调仓操作"""
target_list = get_target_list(context)
# 卖出不在目标列表中的股票
for stock in list(context.portfolio.positions.keys()):
if stock not in target_list:
order_target_value(stock, 0)
else:
order_target_value(stock, context.portfolio.total_value/len(target_list))
# 买入新成分股
available_cash = context.portfolio.available_cash
buy_list = [s for s in target_list if s not in context.portfolio.positions]
if len(buy_list) > 0:
per_value = available_cash / len(buy_list)
for stock in buy_list:
order_target_value(stock, per_value)
def get_target_list(context):
"""获取目标成分股列表"""
# 获取所有股票
all_stocks = get_all_securities(types=['stock'], date=context.previous_date).index.tolist()
# 风险过滤
filtered_stocks = filter_risk_stocks(all_stocks, context)
# 获取市值数据并排序
df = get_fundamentals(query(
valuation.code,
valuation.market_cap
).filter(
valuation.code.in_(filtered_stocks)
), date=context.previous_date)
# 选取市值最小的2000只股票
valid_stocks = df.sort_values('market_cap').head(g.component_num)['code'].tolist()
# 交易状态过滤
return filter_trading_status(valid_stocks, context)
def filter_risk_stocks(stock_list, context):
"""过滤风险股票(ST/*ST/退市)"""
st_info = get_extras('is_st', stock_list, count=1, end_date=context.previous_date)
return [s for s in stock_list if not (
st_info[s][-1] or
get_security_info(s).display_name.startswith('退')
)]
def filter_trading_status(stock_list, context):
"""过滤停牌股票"""
current_data = get_current_data()
return [s for s in stock_list if not current_data[s].paused] # 补全闭合中括号
# ----------------- 关键修复点说明 -----------------
# 1. 修复filter_trading_status函数返回语句的闭合括号
# 2. 统一注释符号为英文格式
# 3. 验证所有函数定义的语法完整性
2025-04-05
