研究了两年,终于研究出来一个无敌策略,不惧牛熊,各种行情都是稳定盈利!! 有感兴趣的朋友欢迎留言,短周期策略。持仓数量十只 如何将研究环境从3切换成3.8?现在打开研究环境就默认进入3, 目前在运作的量化模拟盘,策略是每天交易,一直运行正常,消息也正常推送。 但今天没有发生任何动静,日志也停留在昨天,请问是什么原因? 之前我分享过一个小工具网站,支持国内主流量化平台,可以让 AI 直接帮你写各个平台的策略代码,直接生成可运行的策略代码,代码质量远高于直接使用 DeepSeek、Trae 等平台。上线之后获得了非常多朋友的好评。 大家可以直接用描述策略,然后一键生成可运行的完整策略代码,也可以把它当做一个API 查询工具。 AI工具平台:https://iris.findtruman.io/ai/tool/ai-quantitative-trading/ 我看平台正在开发SuperMind支持,很快就能支持同花顺了 亲测最好用的AI编写量化策略工具,可以让 AI 直接写各个平台的策略代码,直接生成可运行的策略代码,代码质量远高于直接使用 DeepSeek、Trae 等平台。 大家可以直接用描述策略,然后一键生成可运行的完整策略代码,也可以把它当做一个API 查询工具。 最新消息,已经支持SuperMind等主流量化平台啦,并且实盘亲测过了,很适合小白用户,上线之后获得了非常多朋友的好评。 **🚀️ AI工具平台:https://iris.findtruman.io/ai/tool/ai-quantitative-trading/** 大家好,我想和大家分享一个我最近开发的项目——一款面向量化交易的 AI 智能助手工具网站。它可以帮助大家快速生成高质量、可直接复制运行的量化策略代码,无论你是量化小白还是策略开发者,都能从中受益。 核心亮点: 1.多平台支持:目前已支持 PTrade、QMT、miniQMT、聚宽等,并计划不断扩展更多平台。 2.策略生成高效:用户只需选择平台并输入策略想法,AI 即可生成可运行的量化策略代码。 3.快速入门与优化: • 对量化小白:轻松生成可直接运行的策略,快速上手交易。 • 对策略开发者:帮助完善、优化已有策略,节省开发时间。 • 对文档需求者:可作为量化平台的 API 文档问答机器人,方便查询和使用。 4.业内首创:这是首个面向多平台的量化交易 AI 助手,解决了现有 Deepseek 或 Trae 等 AI 工具因缺乏平台知识库而生成代码无法运行的问题。 使用方式:登录 → 选择你使用的平台 → 输入策略想法 → 生成可运行的策略代码。 我希望这个工具能帮助大家更高效地进行策略开发和量化交易,也欢迎大家在帖子里分享使用体验和建议。 网站链接:https://iris.findtruman.io/ai/tool/ai-quantitative-trading/ 如果大家有任何问题或功能需求,也可以在帖子里留言,我会持续优化和更新,让它成为量化交易领域最实用的 AI 助手! 我现在可以每天获取到一个股票代码了。 然后我希望这个股票在当日已开盘价买入,次日以收盘价卖出。如果涨停了则继续观察,知道最高收益回落-5%直接卖出;用python3该怎么写 背景与目的 之前我们有了策略回测代码,到实盘要经过熟悉实盘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() 全撤 回测日期: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("执行筛选符合均线的股票")