




# 2026年机器人概念热度股筛选策略from jqdata import *import pandas as pdimport numpy as npfrom datetime import datetime, timedelta# ============ 初始化设置 ============def initialize(context):# 设置基准set_benchmark('000300.XSHG')# 开启动态复权set_option('use_real_price', True)# 设置滑点set_slippage(FixedSlippage(0.002))# 设置交易成本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))# 设置持仓数量g.hold_count = 10# 设置调仓周期(每月第一个交易日)g.trade_days = 0# 机器人概念股池(按相关度分类)# Tier 1: 核心零部件供应商(最高相关度)g.tier1_stocks = ['603728.XSHG', # 鸣志电器 - 运动控制/空心杯电机'688017.XSHG', # 绿的谐波 - 谐波减速器龙头'002050.XSHE', # 三花智控 - 热管理/执行器'601689.XSHG', # 拓普集团 - 执行器/电驱'003021.XSHE', # 兆威机电 - 微型传动'688160.XSHG', # 步科股份 - 无框力矩电机'002472.XSHE', # 双环传动 - 精密减速器]# Tier 2: 重要零部件及系统供应商g.tier2_stocks = ['300124.XSHE', # 汇川技术 - 伺服系统'002747.XSHE', # 埃斯顿 - 工业机器人'002896.XSHE', # 中大力德 - 减速器'603662.XSHG', # 柯力传感 - 力矩传感器'300503.XSHE', # 昊志机电 - 电主轴/谐波'603009.XSHG', # 北特科技 - 丝杠'603667.XSHG', # 五洲新春 - 轴承/丝杠'301368.XSHE', # 丰立智能 - 减速器]# Tier 3: 整机及应用端g.tier3_stocks = ['688218.XSHG', # 江苏北人 - 工业机器人系统'002698.XSHE', # 博实股份 - 特种机器人'300607.XSHE', # 拓斯达 - 工业机器人'688255.XSHG', # 凯尔达 - 焊接机器人'300024.XSHE', # 机器人 - 新松机器人]# 合并所有股票池g.all_stocks = g.tier1_stocks + g.tier2_stocks + g.tier3_stocks# 运行定时任务(每月第一个交易日调仓)run_monthly(rebalance, 1, time='09:30')# ============ 选股与评分函数 ============def get_robot_score(context, stock_list):"""计算机器人相关度综合评分评分维度:1. 基本面:营收增速、研发投入2. 技术面:动量、波动率3. 热度:成交量变化、资金流入4. 产业链位置权重(Tier1权重最高)"""scores = {}for stock in stock_list:try:score = 0# ===== 1. 产业链位置权重(基础分)=====if stock in g.tier1_stocks:base_score = 40 # 核心零部件最高分elif stock in g.tier2_stocks:base_score = 30 # 重要供应商else:base_score = 20 # 整机应用端score += base_score# ===== 2. 技术面评分(动量)=====# 获取历史价格hist = attribute_history(stock, 60, '1d', ['close', 'volume'], skip_paused=True)if len(hist) < 20:continue# 计算20日、60日涨跌幅returns_20 = (hist['close'][-1] / hist['close'][-20] - 1) * 100returns_60 = (hist['close'][-1] / hist['close'][0] - 1) * 100# 动量得分(偏好温和上涨,避免暴涨暴跌)if 0 < returns_20 < 30:score += 15elif 30 <= returns_20 < 50:score += 20elif returns_20 >= 50:score += 10 # 涨幅过大,降低分数# 趋势持续性if returns_20 > returns_60 / 3: # 短期强于长期趋势的1/3score += 10# ===== 3. 成交量热度 ======vol_mean_20 = hist['volume'][-20:].mean()vol_mean_60 = hist['volume'][:20].mean()vol_ratio = vol_mean_20 / vol_mean_60 if vol_mean_60 > 0 else 1if 1.2 <= vol_ratio < 3: # 成交量温和放大score += 15elif vol_ratio >= 3: # 过度放量score += 5# ===== 4. 基本面过滤 ======# 获取财务数据(季度营收增速)q = query(indicator).filter(indicator.code == stock)df = get_fundamentals(q)if not df.empty:revenue_growth = df['indicator.inc_revenue'][0]if revenue_growth and revenue_growth > 20: # 营收增速>20%score += 10elif revenue_growth and revenue_growth > 0:score += 5# ===== 5. 市值与流动性过滤 ======current_data = get_current_data()[stock]market_cap = current_data.market_cap # 总市值(亿)if market_cap and market_cap > 50: # 市值大于50亿score += 5if market_cap and market_cap > 200: # 大市值龙头加分score += 5scores[stock] = scoreexcept Exception as e:log.error(f"计算{stock}分数出错: {str(e)}")continuereturn scores# ============ 调仓逻辑 ============def rebalance(context):# 获取评分scores = get_robot_score(context, g.all_stocks)if not scores:log.info("未获取到有效评分,跳过本次调仓")return# 按评分排序sorted_stocks = sorted(scores.items(), key=lambda x: x[1], reverse=True)# 选择前N只股票target_stocks = [stock for stock, score in sorted_stocks[:g.hold_count]]# 输出选股结果log.info("========== 本周期机器人概念股排名 ==========")for i, (stock, score) in enumerate(sorted_stocks[:15], 1):name = get_security_info(stock).display_nametier = "T1" if stock in g.tier1_stocks else ("T2" if stock in g.tier2_stocks else "T3")log.info(f"{i}. {name}({stock}) - {tier} - 评分: {score}")# 调仓执行adjust_positions(context, target_stocks)def adjust_positions(context, target_stocks):"""调整持仓:卖出不在目标列表的股票,买入目标股票"""current_holdings = list(context.portfolio.positions.keys())# 1. 卖出不在目标列表的股票for stock in current_holdings:if stock not in target_stocks:order_target(stock, 0)log.info(f"卖出: {stock}")# 2. 计算每只目标股票的仓位if target_stocks:weight = 1.0 / len(target_stocks)cash_per_stock = context.portfolio.total_value * weightfor stock in target_stocks:current_data = get_current_data()[stock]if current_data.paused or current_data.is_st or current_data.is_ht: # 停牌、ST、退市整理continue# 获取当前价格current_price = current_data.day_openif current_price == 0:current_price = current_data.last_price# 计算目标股数target_value = context.portfolio.total_value * weightcurrent_value = context.portfolio.positions[stock].value if stock in context.portfolio.positions else 0if abs(target_value - current_value) / context.portfolio.total_value > 0.02: # 仓位变化>2%才调整order_target_value(stock, target_value)log.info(f"买入/调整: {stock}, 目标市值: {target_value:.2f}")# ============ 盘中风控(可选)===========def handle_data(context, data):"""盘中风控:跌停板风险控制根据记忆:用户之前要求增加"跌停收盘"条件,股票必须收在跌停价才算翘板成功"""current_holdings = list(context.portfolio.positions.keys())for stock in current_holdings:current_data = get_current_data()[stock]# 获取跌停价(A股跌停为前收盘价的90%)pre_close = current_data.pre_closelimit_down = pre_close * 0.9# 如果当前价格触及跌停且封单较大,考虑卖出current_price = current_data.last_priceif current_price <= limit_down * 1.005: # 接近跌停(考虑浮点误差)# 检查是否封死跌停(买一价=跌停价且买量很小)# 这里简化处理:如果当日跌幅>9.5%且是持仓股,风控卖出log.info(f"风控卖出跌停股: {stock}")order_target(stock, 0)# ============ 盘后分析 ============def after_trading_end(context):# 记录当日持仓log.info("========== 当日持仓 ==========")for stock in context.portfolio.positions:pos = context.portfolio.positions[stock]name = get_security_info(stock).display_namelog.info(f"{name}({stock}): 市值{pos.value:.2f}, 收益率{pos.pnl_ratio*100:.2f}%")
聚宽实盘可以看这个:
如果您对某个想法或者策略感兴趣,想获得更详细的落地方案,欢迎留言或私信我。我们可以进行一对一的免费初步交流,期待为您提供真正有价值的帮助。

