研究了两年，终于研究出来一个无敌策略，不惧牛熊，各种行情都是稳定盈利！！ 有感兴趣的朋友欢迎留言，短周期策略。持仓数量十只 如何将研究环境从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("执行筛选符合均线的股票") 
