研究了两年，终于研究出来一个无敌策略，不惧牛熊，各种行情都是稳定盈利！！ 有感兴趣的朋友欢迎留言，短周期策略。持仓数量十只 302132中航成飞缺少20250214以前的日K线和分钟K线数据，希望能尽快补全修复！ 1. 接口信息 接口类型：实时综合行情接口 支持品种：贵金属，商品期货，外汇，A股，港股，美股 查询方式：HTTP, WebSocket 申请密钥：https://infoway.io 官方对接文档：https://infoway.readme.io/reference/ws-subscription 2. 获取股票清单 这个接口用来查询股票的名单，比如我可以获取美股清单： import requests url = "https://data.infoway.io/common/basic/symbols?type=STOCK_HK" headers = {"accept": "application/json"} response = requests.get(url, headers=headers) print(response.text) 只需要在type中传入STOCK_US，或者STOCK_CN即可查询所有美股或A股的清单。 3. 批量获取股票K线 这个接口用来获取多只股票的K线，请求地址： https://data.infoway.io/stock/batch_kline/{klineType}/{klineNum}/{codes} # {klineType} 是K线的时间 # 1 = 1分钟k线 # 2 = 5分钟k线 # 3 = 15分钟k线 # 4 = 30分钟k线 # 5 = 1小时k线 # 6 = 2小时k线 # 7 = 4小时k线 # 8 = 1日k线 # 9 = 1周k线 # 10 = 1月k线 # 11 = 1季k线 # 12 = 1年k线 # {klineNum}是需要返回的K线数量，最大支持一次性返回500根K线 可以跨市场查询，比如同时获取A股，港股和美股的个股K线： import requests api_url = 'https://data.infoway.io/stock/batch_kline/1/10/002594.SZ%2C00285.HK%2CTSLA.US' # 申请免费密钥: https://infoway.io # 设置请求头 headers = { 'User-Agent': 'Mozilla/5.0', 'Accept': 'application/json', 'apiKey': 'yourApikey' } # 发送GET请求 response = requests.get(api_url, headers=headers) # 输出结果 print(f"HTTP code: {response.status_code}") print(f"message: {response.text}") 4. 查询股票盘口 这个接口用于查询股票的交易盘口，最大支持5档盘口查询： import requests url = "https://data.infoway.io/stock/batch_depth/002594.SZ%2C00285.HK%2CTSLA.US%2CAUDCAD%2CHK50%2CXNIUSD" # 申请免费密钥: https://infoway.io # 设置请求头 headers = { 'User-Agent': 'Mozilla/5.0', 'Accept': 'application/json', 'apiKey': 'yourApikey' } response = requests.get(url, headers=headers) print(response.text) 5. WebSocket订阅实时股票行情 以上介绍的是HTTP请求，这种方式存在一定的延时，在实盘交易中推荐使用WebSocket来获取实时的行情推送。WebSocket的优势是和服务器建立一个长连接，只要连接不断开都会收到来自服务器的数据推送，延时一般在120ms以内。而你只需要每隔一分钟向服务器发送一个心跳来保活即可。下面是WebSocket的代码示例，虽然是查询的贵金属，但原理是一样的： import json import time import schedule import threading import websocket from loguru import logger class WebsocketExample: def __init__(self): self.session = None # 申请免费API Key: https://infoway.io self.ws_url = "wss://data.infoway.io/ws?business=common&apikey=YourAPIKey" self.reconnecting = False self.last_ping_time = 0 self.max_ping_interval = 30 # 最大心跳包间隔时间 self.retry_attempts = 0 # 重试次数 self.max_retry_attempts = 5 # 最大重试次数 def connect_all(self): """建立WebSocket连接并启动自动重连机制""" try: self.connect(self.ws_url) self.start_reconnection(self.ws_url) except Exception as e: logger.error(f"Failed to connect to {self.ws_url}: {str(e)}") def start_reconnection(self, url): """启动定时重连检查""" def check_connection(): if not self.is_connected(): logger.debug("Reconnection attempt...") self.retry_attempts += 1 if self.retry_attempts <= self.max_retry_attempts: self.connect(url) else: logger.error("Exceeded max retry attempts.") # 使用线程定期检查连接状态 threading.Thread(target=lambda: schedule.every(10).seconds.do(check_connection), daemon=True).start() def is_connected(self): """检查WebSocket连接状态""" return self.session and self.session.connected def connect(self, url): """建立WebSocket连接""" try: if self.is_connected(): self.session.close() self.session = websocket.WebSocketApp( url, on_open=self.on_open, on_message=self.on_message, on_error=self.on_error, on_close=self.on_close ) # 启动WebSocket连接（非阻塞模式） threading.Thread(target=self.session.run_forever, daemon=True).start() except Exception as e: logger.error(f"Failed to connect to the server: {str(e)}") def on_open(self, ws): """WebSocket连接建立成功后的回调""" logger.info(f"Connection opened") try: # 发送贵金属实时成交明细订阅请求 trade_send_obj = { "code": 10000, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a", "data": {"codes": "XAUUSD"} # XAUUSD 为贵金属黄金（黄金/美元） } self.send_message(trade_send_obj) # 不同请求之间间隔一段时间 time.sleep(5) # 发送贵金属实时盘口数据订阅请求 depth_send_obj = { "code": 10003, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a", "data": {"codes": "XAUUSD"} # XAUUSD为黄金的实时盘口数据 } self.send_message(depth_send_obj) # 不同请求之间间隔一段时间 time.sleep(5) # 发送贵金属实时K线数据订阅请求 kline_data = { "arr": [ { "type": 1, "codes": "XAUUSD" # XAUUSD 为黄金K线数据 } ] } kline_send_obj = { "code": 10006, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a", "data": kline_data } self.send_message(kline_send_obj) # 启动定时心跳任务 threading.Thread(target=lambda: schedule.every(30).seconds.do(self.ping), daemon=True).start() except Exception as e: logger.error(f"Error sending initial messages: {str(e)}") def on_message(self, ws, message): """接收消息的回调""" try: logger.info(f"Message received: {message}") except Exception as e: logger.error(f"Error processing message: {str(e)}") def on_close(self, ws, close_status_code, close_msg): """连接关闭的回调""" logger.info(f"Connection closed: {close_status_code} - {close_msg}") def on_error(self, ws, error): """错误处理的回调""" logger.error(f"WebSocket error: {str(error)}") def send_message(self, message_obj): """发送消息到WebSocket服务器""" if self.is_connected(): try: self.session.send(json.dumps(message_obj)) except Exception as e: logger.error(f"Error sending message: {str(e)}") else: logger.warning("Cannot send message: Not connected") def ping(self): """发送心跳包""" current_time = time.time() if current_time - self.last_ping_time >= self.max_ping_interval: ping_obj = { "code": 10010, "trace": "01213e9d-90a0-426e-a380-ebed633cba7a" } self.send_message(ping_obj) self.last_ping_time = current_time else: logger.debug(f"Ping skipped: Time interval between pings is less than {self.max_ping_interval} seconds.") # 使用示例 if __name__ == "__main__": ws_client = WebsocketExample() ws_client.connect_all() # 保持主线程运行 try: while True: schedule.run_pending() time.sleep(1) except KeyboardInterrupt: logger.info("Exiting...") if ws_client.is_connected(): ws_client.session.close() 6. 常见问题 HTTP和WebSocket有什么区别？ 通过HTTP请求，返回的数据是基于请求和响应传输的。客户端向服务器发送请求，服务器返回响应。每次请求/响应都必须是完整的。而WebSocket的数据是全双工的，客户端和服务器可以随时向对方发送数据，也就是所谓的长连接。它更适用于实时通信，因为一旦连接建立，可以保持双向流量。 港股的延迟是多少？ 这个和接口的设计有关，一般通过WebSocket获取的数据，延迟在100ms左右。 为什么模拟交易里的证券名称都不显示了，已经运行一年了，之前都是正常显示 的。前几天突然证券名称都不显示了，以为是平台有更新什么东西，过两天会好，结果等到现在也没好，依然不能显示 在A股市场中，小市值股票因其成长潜力巨大而备受关注，但同时也面临着较高的波动风险。今天我们要深度解读一个表现优异的量化选股策略——小市值低回撤选股策略**。** 从回测数据来看，这个策略表现相当亮眼： 累计收益率：3740.17% 年化收益率：94.67% 最大回撤：-18.58% 回测周期：2020年1月至2025年6月（5年174天） 策略核心逻辑详解 1. 选股标准：精选小市值优质股 这个策略的选股逻辑非常严谨，主要包含以下几个维度： 市值筛选 # 设定市值范围 g.min_mv = 10 # 最小市值10亿 g.max_mv = 1e8 # 最大市值100亿 策略专注于10-100亿市值的股票，这个区间的公司通常具有： 较强的成长性和弹性 相对较好的流动性 避免了过小公司的极端风险 基本面筛选 q = query( valuation.code, valuation.market_cap, income.np_parent_company_owners, # 归属母公司净利润 income.net_profit, # 净利润 income.operating_revenue # 营业收入 ).filter( valuation.market_cap.between(g.min_mv, g.max_mv), income.np_parent_company_owners > 0, # 净利润为正 income.net_profit > 0, income.operating_revenue > 1e8 # 营收超过1亿 ) 这套筛选标准确保了： 公司盈利能力良好 具备一定的经营规模 避免了业绩亏损的地雷股 2. 风险控制：多重防护机制 技术面过滤 def filter_stocks(context, stock_list): # 过滤停牌股票 if current_data[stock].paused: continue # 过滤ST股票 if current_data[stock].is_st: continue # 过滤创业板、科创板等高风险板块 if stock.startswith('30') or stock.startswith('68'): continue # 过滤次新股（上市不满375天） if context.previous_date - start_date < timedelta(days=375): continue 涨跌停处理 def check_limit_up(context): # 对昨日涨停股票进行特殊处理 if current_data.iloc[0, 0] < current_data.iloc[0, 1]: # 涨停打开立即卖出 close_position(position) 这种处理方式非常聪明，既避免了追高风险，又能及时止盈。 策略如何实现低回撤 1. 空仓保护机制 g.pass_months = [1, 4] # 1月和4月空仓 策略在历史上表现较差的月份选择空仓，转而持有稳健的银行股或ETF。这是基于A股市场的季节性规律： 1月：年报预期影响，小盘股容易调整 4月：一季报公布期，业绩不确定性增加 2. 动态调仓机制 def adjust_stock_num(context): # 根据市场趋势调整持仓数量 ma_para = 10 # 10日均线参数 diff = last_row['close'] - last_row['ma'] result = 3 if diff >= 500 else \ 3 if 200 <= diff < 500 else \ 4 if -200 <= diff < 200 else \ 5 if -500 <= diff < -200 else \ 6 这个机制根据指数相对均线的位置，动态调整持仓数量： 市场强势时减少持仓（3-4只） 市场弱势时增加持仓（5-6只） 通过分散投资降低单股风险。 3. 止损止盈机制 def sell_stocks(context): # 个股止盈：收益达到100%时止盈 if price >= avg_cost * 2: order_target_value(stock, 0) # 个股止损：亏损超过9%时止损 elif price < avg_cost * (1 - g.stoploss_limit): order_target_value(stock, 0) # 市场止损：大盘平均跌幅超过5%时清仓 if down_ratio >= g.stoploss_market: # 全部清仓 这套止损止盈机制兼顾了个股和市场两个维度，有效控制了回撤。 策略特色亮点 1. 智能的涨停处理 策略对涨停股有独特的处理方式：涨停当天继续持有，第二天如果涨停打开则立即卖出。这样既能享受涨停收益，又能避免高位套牢。 2. 季节性调整 根据A股市场的季节性特征，在historically表现不佳的月份主动空仓，这种"主动防守"的思路值得学习。 3. 市场情绪感知 通过10日均线判断市场强弱，并相应调整持仓结构，体现了对市场情绪的敏感把握。 策略为什么能够盈利 1. 抓住了小市值股票的成长红利 10-100亿市值的公司往往处于快速成长期，业绩增长带来的估值提升是主要收益来源。 2. 严格的质量控制 通过营收、净利润等指标筛选，确保买入的都是基本面良好的公司，避免了踩雷风险。 3. 优秀的风险管理 多重止损机制和空仓保护，有效控制了下行风险，保证了策略的长期稳定性。 4. 顺应市场规律 策略的设计充分考虑了A股市场的特点，如季节性效应、涨停板制度等，做到了"因地制宜"。 交割单分析 从交割单可以看出，策略的交易频率适中，避免了过度交易的成本。让我们来看几个典型的交易案例： 成功案例分析 买入时机：通常选择在股价相对低位 持仓周期：一般持有2-4周 卖出时机：要么达到止盈线，要么触发止损 策略完整交割单查看 想要查看该策略的完整模拟交割单，可以访问：https://www.9db.com/detail/049687C4-10C7-4C21-A62F-F3798E44A396?t=quant_mm 从交割单中可以看到： 每笔交易的具体时间和价格 买卖理由和策略逻辑 持仓周期和收益情况 风险控制的实际执行 量化策略订阅服务 对于普通投资者来说，自己编写和运行量化策略代码是有一定技术门槛的。但现在您可以通过 www.9db.com 订阅现成的量化策略服务。 订阅优势： 无需编程基础，一键订阅 实时交易信号推送 邮件通知每笔交易操作 专业团队持续优化策略 一旦策略产生交易信号，您将第一时间收到邮件通知，包含具体的买卖操作建议，让您能够及时跟随策略操作。 投资建议与风险提示 适合人群 风险承受能力中等的投资者 希望获得超额收益的长期投资者 认可量化投资理念的投资者 注意事项 仓位控制：建议将此策略作为投资组合的一部分，而非全部资金 市场环境：策略在震荡市和牛市表现更佳 执行纪律：严格按照策略信号执行，不要主观调整 后市展望 随着A股市场的不断成熟，小市值优质公司的投资价值将继续凸显。该策略通过严格的风险控制和科学的选股逻辑，有望在未来继续获得稳健收益。 小市值低回撤选股策略通过精密的算法和严格的风险控制，在获得高收益的同时有效控制了回撤。对于想要参与量化投资但缺乏技术基础的投资者，订阅专业的量化策略服务不失为一个好选择。 记住，量化投资的核心不是预测市场，而是通过科学的方法在风险可控的前提下获得超额收益。这个策略正是这一理念的完美体现。 免责声明：本文所提供的一切内容（包括但不限于交易案例分析、交割单解读、策略分享）仅供学习参考，不构成任何投资建议或交易指导。用户应当独立判断并自行承担投资风险和后果。 回测日期：2014-12-5 至 2025-05-31 选股策略： 选取中证1000指数中的1000只股票 当日涨跌幅 3-5 当日换手率 3-10 当日均线 MA5 > MA10 止损止盈条件 收益 > 5 或者 收益-3卖出 # 中证1000选股策略 # 作者 但盼风雨来 # 选股策略： # 当日涨跌幅 3-5 # 当日换手率 3-10 # 当日均线 MA5 > MA10 # 止损止盈条件 收益 > 5 或者 收益-3卖出 import pandas as pd # 初始化函数,全局只运行一次 def init(context): # 设置基准收益：沪深300指数 set_benchmark('000300.SH') # 打印日志 log.info('策略开始运行,初始化函数全局只运行一次') # 设置股票每笔交易的手续费为万分之二(手续费在买卖成交后扣除,不包括税费,税费在卖出成交后扣除) set_commission(PerShare(type='stock',cost=0.0000854,min_trade_cost=0)) # 设置股票交易双边滑点0.5%,表示买入价为实际价格乘1.0025,卖出价为实际价格乘0.9975 set_slippage(PriceSlippage(0.005)) # 设置日级最大成交比例25%,分钟级最大成交比例50% # 日频运行时，下单数量超过当天真实成交量25%,则全部不成交 # 分钟频运行时，下单数量超过当前分钟真实成交量50%,则全部不成交 set_volume_limit(0.25,0.5) # 设置要操作的股票：同花顺 context.security = ['300033.SZ'] # 回测区间、初始资金、运行频率请在右上方设置 context.stock_list = ['300033.SZ','000718.SZ'] run_daily(function) #每日开盘前9:00被调用一次,用于储存自定义参数、全局变量,执行盘前选股等 def before_trading(context): # 获取日期 date = get_datetime().strftime('%Y-%m-%d %H:%M:%S') # 打印日期 log.info('{} 盘前运行'.format(date)) ## 开盘时运行函数 def handle_bar(context, bar_dict): # 获取时间 time = get_datetime().strftime('%Y-%m-%d %H:%M:%S') log.info(bar_dict) # 买卖策略 positions = context.stock_account.positions to_sell = [] for position in positions: shouyi = positions[position].profit_rate * 100 # log.info(f"{positions[position].symbol}的收益率为{shouyi}%") stock = positions[position].symbol if shouyi > 6 or shouyi <-3: to_sell.append(stock) for stock in to_sell: order_target(stock,0) log.warn(f"已清仓了{stock}") # 打印时间 log.info('{} 盘中运行'.format(time)) stock_list = context.stock_list log.info("下单的股票为",str(stock_list)) for stock in stock_list: price = get_current(stock) current_price = price[stock].open # log.info(f"股票代码{stock},价格{current_price}") order(stock,1000,price = current_price) date = get_datetime() # log.info('{} 运行'.format(date)) log.info(f"已购买{stock},价格为{current_price}") ## 收盘后运行函数,用于储存自定义参数、全局变量,执行盘后选股等 def after_trading(context): # 获取时间 time = get_datetime().strftime('%Y-%m-%d %H:%M:%S') # 打印时间 log.info('{} 盘后运行'.format(time)) log.info('一天结束') def function(context,*args): get_all_stock(context) get_rate_stock(context) change_hand(context) avg_stock(context) # 获取所有股票 def get_all_stock(context): # 获取时间 time = get_datetime() # 获取市场所有股票 # all_stock = get_all_securities(ty='stock', date=time) # df = all_stock # df = df[~df['display_name'].str.contains('ST', na=False)] # df = df[~df['order_book_id'].str.contains('BJ')] # stock_list = df['order_book_id'].tolist() # 获取中证1000 stock_list= get_index_stocks('000852.SH',time) log.info("已获取到所有股票代码") context.stock_list = stock_list log.info(f"中证1000股票数量为{len(stock_list)}") log.info("已获取中证1000股票") # 筛选符合涨幅条件的股票 def get_rate_stock(context): time = get_datetime().strftime('%Y-%m-%d') stock_list = context.stock_list null_list = [] fields = ['quote_rate','open'] current = get_price( stock_list, start_date=None, end_date=time, fre_step='1d', fields=fields, skip_paused=False, fq='pre', bar_count=1, is_panel=False, ) for stock in stock_list: if (current[stock].quote_rate[0] > 3) & (current[stock].quote_rate[0] < 5): null_list.append(stock) context.stock_list = null_list log.info(f"符合涨跌幅限制的股票数量为{len(null_list)}") log.info("执行筛选符合涨幅条件的股票") # 筛选符合换手率的股票 def change_hand(context): current_stock = context.stock_list hand_list = [] time = get_datetime().strftime('%Y%m%d') fields = ['turnover_rate'] change_hand = get_price( current_stock, start_date=None, end_date=time, fre_step='1d', fields=fields, skip_paused=False, fq='pre', bar_count=1, is_panel=False, ) for stock in current_stock: if (float(change_hand[stock]['turnover_rate'])>3)&(float(change_hand[stock]['turnover_rate'])<10): hand_list.append(stock) context.stock_list = hand_list log.info(f"符合换手率的股票数量为{len(hand_list)}") log.info("执行筛选符合换手率的股票") # 筛选符合均线的股票 def avg_stock(context): time = get_datetime().strftime('%Y%m%d') stock_list = context.stock_list avg_list = [] price = get_price( stock_list, start_date=None, end_date=time, fre_step='1d', fields=['close'], skip_paused=False, fq='pre', bar_count=20, is_panel=False, ) for stock in stock_list: MA5 = price[stock]['close'].iloc[-5:].mean() MA10 = price[stock]['close'].iloc[-10:].mean() if MA5 > MA10 : avg_list.append(stock) context.stock_list = avg_list log.info(f"符合均线策略的股票数量为{len(avg_list)}") log.info("执行筛选符合均线的股票") 背景与目的 之前我们有了策略回测代码，到实盘要经过熟悉实盘API、写代码、调试代码的环节，大概还需要1-2周的时间才能实盘，有非常多的用户到这一步 会束手无策，甚至放弃！ 现在，有了回测代码直接实盘的功能，可以省去这个步骤，让刚入门的朋友也可以直接拿回测代码进行实盘了。很棒！为我们的工程师点赞！ 此外还新增了一些接口，方便实盘交易。 不断降低实盘的门槛是我们的目标，如果您有任何好的想法意见请随时留言！ 本功能需要重启研究环境才能生效！ 策略实盘交易（回测代码1分钟实盘） ?调用方法： research_trade( name, source_code, capital_base=100000, frequency='DAILY', stock_market='STOCK', benchmark=None, trade_api=None, signal_mode=True, dry_run=False, recover_dt=False, ) ?参数说明： name：str，策略名称，会在./persist/下生成一个同名目录，用于存放持久化的策略信息 source_code：str，策略代码，可从策略研究模块中直接复制，代码置于"""..."""中 capital_base： float，初始资金量 如果接入了TradeAPI对象，且 signal_mode=False，那么此参数无意义 frequency: str，策略频率，'DAILY'或'MINUTE' stock_market: str，策略类型，默认'STOCK' benchmark: str，基准指数 trade_api: TradeAPI对象，绑定需要仿真交易的资金账号 如果不传入TradeAPI对象，即 trade_api=None，此时为模拟交易 如果传入TradeAPI对象，此时为仿真交易 signal_mode: bool，（新增）默认为 True signal_mode=True，此时策略实际上运行的时初始资金为capital_base的模拟交易，context、get_orders等方法返回的结果均为模拟交易中计算的数据，与资金账号的数据无关；策略下单在模拟交易撮合成交后，才会通过trade_api下单至柜台 signal_mode=False，此时策略中context、get_orders等方法返回的结果均为从 柜台查询，策略下单也会直接下至柜台 dry_run: bool，试运行，立即返回，默认为 False recover_dt: bool或 str，（新增）是否断点运行，默认为 False recover_dt=False，从当前时点开始执行，不从断定运行 recover_dt=True，从上次策略结束时点开始运行 recover_dt='today'，从当日开始运行，此模式下只会补执行 before_trading与 open_auction，handle_bar依旧从当前时间开始执行 recover_dt='yyyyMMdd HH:mm'，从指定时间开始运行 ?️ 返回值： RealtimeService类 ?作用： 模拟交易：撮合机制与回测相同 仿真交易：通过仿真柜台撮合，更贴近真实交易环境 ❗注意事项： 策略需在9:00前开启运行，否则在未设置recover_dt的情况下，会跳过before_trading等步骤 初始化TradeAPI时需要指定下单策略order_policy，MarketPolicy为市价下单；LimitPolicy为限价下单。如未指定，由于策略下单时使用均价，可能存在多位小数，最终实盘账户下单的时候可能产生废单 signal_mode=True时，如想在context中获得仿真账号的持仓、资金等数据，可以使用同步函数 sync_trade_api() ?示例： from tick_trade_api import TradeAPI #初始化TradeAPI时需要指定下单策略，MarketPolicy为市价下单；LimitPolicy为限价下单 trade_api=TradeAPI('69271711',order_policy=MarketPolicy) source_code=""" # 股票策略模版 def init(context): pass # 盘前执行 def before_trading(context): pass # 开盘时运行函数 def handle_bar(context, bar_dict): order_id = order('000001.SZ', 100) print(get_orders()) try: cancel_order(order_id) except: print('撤单失败') print(get_open_orders()) print(get_tradelogs()) print(context.portfolio.stock_account) print(context.portfolio.positions) """ rtrade = research_trade( '研究环境策略', source_code, frequency='MINUTE', trade_api=trade_api, signal_mode=False, recover_dt='today' ) trade_api=TradeAPI('69271711',order_policy=MarketPolicy) 中的账号是模拟资金账号或者是实盘资金账号。 其他更新 这次还增加了几个功能 策略框架中增加 ： cancel_order_all() 全撤 get_tradelogs()获取当日全部成交订单 get_orders() 获取委托，和get_order()一致，主要时和tradeapi中函数名对齐 tradeapi增加： get_open_orders() 获取当日未成订单 cancel_order_all() 全撤 感谢supermind解决了我多年的强迫症。 从mindgo到现在，兜兜转转，我坚持了两年多，感谢老张的陪伴，总算有了退休的希望。把biqquant平台的策略搬家到了这里，彻底放飞自我了。。训练集 15-22年 策略--超短龙头战法无未来函数绩效。我一直觉得 传统量化要迭代到AI-量化，AI-量化是未来的趋势。机器学习和深度学习算法在金融市场里面还是很多应用的空间的。做了很多策略才发现，用机器学习做出来的策略，就是比传统策略要亮眼一点而且超短T+1策略这一块 还是机器学习做出来的策略效果会比较好。继续实盘。感觉有希望了，今年退休！点赞都是有缘人，祝你23年暴富！需要框架的 邮箱 随缘给 没有过拟合，没有未来函数，熬了无数个深夜，想破脑袋的策略，我觉得不完美，还得继续优化，找出规律，主要代码如下： 集合竞价，9:26自动运行 def open_auction(context, bar_dict): try: get_open_sell(context, bar_dict) # 开盘卖出低开-9%的股票 get_auction_analysis(context) # 更新竞价强度 get_market_trend(context) # 缓存大盘趋势 print(f"大盘趋势: {'上升 🔴' if context.market_stop == 'up' else '下降 🟢'}") print(f"高开比例: {'强势 🟥' if context.limit_up_stop == 'up' else '弱势 🟩'}") if context.market_stop == 'down' and context.limit_up_stop == 'down': print("❌ 盘前检测到风险，今日禁止买入") return result = query_iwencai(query=get_stock_wencai(context.market_stop), domain='股票', df=True) if not result.empty: get_buy_stocks(context, result) except Exception as e: print(f"集合竞价 出错: {str(e)[:30]}") 构建问财查询语句 def get_stock_wencai(m_stop): dates = get_custom_dates(custom_days=[1, 2, 6]) today, yesterday, pastday2, pastday6 = dates if m_stop == "up": query = ( f"{today}竞价涨幅>=2%且<10%，" f"{today}竞价未匹配金额，" f"{today}竞价量/{yesterday}成交量>=2%，" f"{yesterday}成交量>{pastday6}至{pastday2}的日均成交量，" f"{yesterday}个股热度排名<=200，" f"{yesterday}的20日均线角度>20，" f"{yesterday}的10日均线>20日均线，" "非st，非创业板，非科创板，非北证a股" ) else: query = ( f"{today}竞价涨幅>=0%且<10%，" f"{today}竞价未匹配金额，" f"{today}竞价量/{yesterday}成交量>=1%，" f"{yesterday}成交量>{pastday6}至{pastday2}的日均成交量，" f"{yesterday}个股热度排名<=150，" f"{yesterday}的20日均线角度>20，" f"{yesterday}的10日均线>20日均线，" "非st，非创业板，非科创板，非北证a股" ) return query 获取热门概念成分股 def get_hot_concept(context, date): try: result_zhishu = query_iwencai(f'{date}同花顺概念指数涨幅，{date}涨停家数占比', domain='指数', df=True) excluded_concepts = ['融资融券', '深股通', '沪股通', '国企改革', '央企国企改革', '专精特新'] result_zhishu = result_zhishu[~result_zhishu['指数简称'].isin(excluded_concepts)] n = 10 if context.market_stop == "up" else 3 # 根据市场趋势确定筛选数量 concept_range = result_zhishu.nlargest(n, '指数@涨跌幅:前复权') # 获取涨幅和上涨家数占比靠前的概念 concept_ratio = result_zhishu.nlargest(n, '指数@涨停家数占比') if context.market_stop == "up": concept_all = pd.merge(concept_range, concept_ratio, how='inner', on=['指数代码', '指数简称']) # 强势市场取交集 else: concept_all = pd.concat([concept_range, concept_ratio]).drop_duplicates(subset=['指数代码', '指数简称']) # 弱势市场取并集 concept_stocks = {} for concept_code in concept_all['指数代码'].unique(): stocks = get_concept_stocks(concept_code, date=date) # 获取概念成分股 if stocks: concept_stocks[concept_code] = stocks return concept_stocks except Exception as e: print(f"获取热门概念成分股 出错: {str(e)[:30]}") return {} 集合竞价分析 def get_auction_analysis(context): t_date = get_datetime().strftime('%Y%m%d') # 当前交易日 y_date = get_previous_trading_date(get_datetime()).strftime('%Y%m%d') # 前一交易日 try: result = query_iwencai(query=f'{y_date}涨停，{t_date}竞价涨幅', domain='股票', df=True) if result.empty: context.limit_up_stop = "down" return limit_up_count = len(result) up_count = len(result[result['竞价涨幅'] > 0]) ratio = up_count / limit_up_count if limit_up_count > 0 else 0 if ratio >= 0.6: context.limit_up_stop = "up" else: context.limit_up_stop = "down" except Exception as e: print(f"集合竞价分析 出错: {str(e)[:30]}") context.limit_up_stop = "down" 判断大盘趋势 def get_market_trend(context): try: hs_index = history('399905.SZ', ['close', 'volume'], 20, '1d', True, 'pre') # 获取历史数据 if len(hs_index) < 20: context.market_stop = "normal" return hs_close = hs_index['close'] # 提取收盘价 hs_volume = hs_index['volume'] # 提取成交量 hs_ma5 = hs_close.rolling(5).mean().iloc[-1] # 计算均线 hs_ma10 = hs_close.rolling(10).mean().iloc[-1] hs_ma20 = hs_close.rolling(20).mean().iloc[-1] avg_volume_20 = hs_volume.rolling(20).mean().iloc[-1] # 计算成交量指标 last_volume = hs_volume.iloc[-1] close_prices = hs_close.values macd, signal, _ = talib.MACD(close_prices, fastperiod=6, slowperiod=12, signalperiod=5) # 计算MACD指标 is_macd_dead = macd[-1] < signal[-1] # DIFF < DEA 视为弱势 is_below_ma5 = hs_close.iloc[-1] < hs_ma5 # 是否跌破5日均线 if is_macd_dead and is_below_ma5: # 判断趋势 trend = "down" elif hs_ma10 > hs_ma20: if hs_ma5 < hs_ma10 and last_volume < avg_volume_20: trend = "down" else: trend = "up" else: trend = "down" context.market_stop = trend except Exception as e: print(f"判断大盘趋势 出错: {str(e)[:30]}") context.market_stop = "normal" 很多投资者都听说过去年实盘大赛的两位风云人物——归因和二池。归因以21.5倍的收益率夺冠，二池紧随其后取得21.3倍的成绩。今年归因从2347万做到3425万，收益率42%；二池从419万增长至577万，收益率38%。两人实力相当，都是值得学习的顶尖高手。 表面上看，他们都采用所谓的"价值投机模式"，但如果你仔细研究过两者的实际交割单，就会发现他们的操作理念和选股逻辑完全不同。今天我们就通过实际案例来剖析这两种模式的本质差异。 看之前写的归因交割单学习笔记：《从"归因"交易单看低吸战法的复利秘诀》 看之前写的二池交割单学习笔记：《10个月21倍百万杯冠军二池复利秘籍：强势股回调介入法》 归因模式：题材嗅觉的先发制人 核心特征：超强的题材反应能力 归因最让人钦佩的是他对主线题材的敏锐嗅觉。让他声名鹊起的成名战就是去年连板前一天提前埋伏了中百集团和永辉超市，成功把握住了大部分涨幅。这种看似不可思议的操作，背后体现的是对题材发展的深度理解。 从归因的实际操作来看，他在12月6日成本线下方买入中百集团和永辉超市，12月9日中百集团涨停后，他在永辉超市仍在水下时果断加仓，结果次日两股双双一字涨停。这种精准的时点把握和果断的操作，体现了他对零售板块逻辑的深度理解，他坚信胖东来的火爆会向零售板块扩散。 可查看归因具体操作案例：中百集团交割详情 | 永辉超市交割详情 年底小红书概念刚出现时，他第一时间上车天下秀和福石控股，又一次收获连板。今年云计算方向启动后，归因同样在这个赛道上持续滚动操作。几乎每个主线题材都能看到他的身影。 实战案例解析 以小红书概念为例，当归因第一时间介入时，他的逻辑很简单："小红书涌入大量外国人，所有人都能看到流量爆发，你认不认为这是个大题材？如果认可，那上车就是这么简单。" 这句话道出了归因模式的精髓：通过已经发生的现象推演未来的市场反应。不管是零售、小红书还是云计算，都是基于已发生事件的发酵逻辑，而归因能够迅速做出反应并付诸行动。 另一个经典案例是润泽科技的操作。归因在12月16日参与这只"豆包概念"的核心标的，在12月19日3次加仓，最终在12月20日获得约20%的丰厚收益。这种连续加仓的操作显示了他对题材发展节奏的精准把控。 可查看归因具体操作案例：润泽软件交割详情 归因模式的优势与挑战 优势： 有明确主线时收益如火箭般飞升 能够把握住市场最大的赚钱机会 对题材理解深刻，判断准确度高 挑战： 需要大量知识储备和惯性认知理解力 题材轮动时表现相对平庸 对普通投资者的要求极高 二池模式：拐点把握的后发制人 核心特征：在分歧中寻找机会 与归因截然不同，二池完全没有归因那种超前的反应能力。国庆前错过券商，年后错过云计算，基本所有主线题材他都没能第一时间参与。但二池的强大之处在于，他能在主线走出来后的分歧震荡阶段，根据个股表现差异进行低吸操作。 可查看二池具体操作案例：润和软件交割详情 | 蒙草生态交割详情 实战案例解析 案例一：蒙草生态的精准低吸 去年化债概念的蒙草生态，二池是在走完第一波行情调整时才介入，虽然每次只能吃到十几到二十个点的空间，但这样的"鱼尾行情"累积下来，收益并不比抓主线核心躺板差，而且回撤更小。具体上一段已有说明。 案例二：黑芝麻的完美低吸 11月28日黑芝麻开盘水下，二池果断低吸，当天即涨停，后续收获3连板。查看黑芝麻详细操作记录 这些案例展现了二池对个股技术面和资金面的精准判断能力。 二池的选股哲学 二池本人多次强调："大跌是最好的选股时机。在大跌时表现出抗跌特性的股票，往往能在市场修复时创出新高。" 典型例子是11月19日的易点天下操作。前一日该股在大盘回调情况下逆势上涨，19日低开回调正是二池的买点。 这种模式体现了"在大盘弱势时该弱不弱"的选股逻辑，等大盘回暖时这类股票能率先修复。 查看二池易点天下操作详情和分时线买卖点 两种模式的深度对比 学习难度对比 归因模式： 需要深厚的知识积累和理解力 要有超前的市场判断能力 对个人天赋和经验要求极高 学习门槛高，难以复制 二池模式： 主要基于盘面表现和技术分析 需要一定基本面知识，但门槛相对较低 只要有归纳总结能力就能学会 更适合普通投资者学习 适用环境对比 归因模式： 适合主线明确的市场环境 在题材轮动快速的行情中优势明显 需要能够全天候关注市场动态 二池模式： 适合震荡分化的市场环境 在市场调整期表现更加突出 适合无法全天盯盘的上班族 通过对比两人的实际交割单可以清晰看出这种差异：归因在主线爆发期收益飙升，二池在震荡期表现更稳健。 学习建议：选择适合自己的模式 如果你想学习归因模式 加强知识储备：需要对各行各业、政策动向有深入了解 培养敏锐嗅觉：多关注新闻热点，训练从现象推演市场反应的能力 提高反应速度：一旦识别机会要果断行动 做好心理准备：这种模式对天赋要求较高，不是所有人都能掌握 如果你想学习二池模式 掌握技术分析：重点学习均线、量能等技术指标的运用 培养耐心：等待主线走出后的分歧机会，不急于抢风头 学会逆向思维：在别人恐慌时寻找机会 控制风险：这种模式更注重稳健，回撤控制是关键 可以参考二池的银之杰做T操作和沃尔核材精准低吸案例来学习具体的操作技巧。 结语 归因和二池代表了两种截然不同的投资哲学。归因是"先发制人"的题材捕手，凭借超强的理解力和反应速度抢占先机；二池是"后发制人"的拐点猎手，在市场分歧中寻找被错杀的机会。 对于普通投资者而言，二池的模式更容易学习和掌握，因为它更多依赖于对盘面的观察和技术分析，而不需要归因那种近乎天赋般的题材嗅觉。但无论选择哪种模式，都需要持续的学习和实践才能真正掌握。 正如古语所说："弃而舍之，朽木不折；锲而不舍，金石可镂。"无论你选择哪种模式，坚持学习和实践才是成功的关键。 从两位高手21倍+的年化收益可以看出，无论是归因的先发制人还是二池的后发制人，都是经过市场验证的有效策略。关键在于找到适合自己的那一种。 注：本文分析基于公开的交割单信息，仅供学习交流使用，不构成投资建议。股市有风险，投资需谨慎。 
