研究了两年，终于研究出来一个无敌策略，不惧牛熊，各种行情都是稳定盈利！！ 有感兴趣的朋友欢迎留言，短周期策略。持仓数量十只 回测日期：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("执行筛选符合均线的股票") 本教程将指导您如何通过简单的几步接入实时外汇行情接口，获取您所需的外汇数据。 1. 准备工作 在开始之前，请确保您已具备以下条件： Python 环境: 安装了 Python 编程语言。 Requests 库: Python 的 requests 库，用于发送 HTTP 请求。如果未安装，可以通过 pip install requests 命令进行安装。 API Key: 从数据服务提供商处获取一个有效的 API Key。这是访问接口的凭证。 2. 理解 API 请求结构 实时外汇行情接口通常通过 HTTP GET 请求获取数据。根据您提供的示例，API 请求 URL 结构如下： https://data.infoway.io/common/batch_kline/{Kline_type}/{Kline_Num}/{symbol} Kline_type: 数据的时间间隔 (各接口提供的间隔有所不同，Infoway API提供多种K线周期，1：1分钟k；2：5分钟k；3：15分钟k；4：30分钟k；5：1小时k；6：2小时k；7：4小时k；8：日k；9：周k；10：月k；11：季k；12：年k)。 Kline_Num: 所需的K线数量，比如10则返回10条最新的K线。如果你向接口查询一个货币对，最大可以查询最近500根k；多个货币对同时查询的话，1次只能返回最近2根K线。 symbol: 外汇交易对 (例如：GBPUSD 代表英镑/美元)。 3. 发送实时K线查询请求 在发送请求时，需要设置特定的 HTTP 请求头，其中最重要的是您的 apiKey。 import requests api_url = 'https://data.infoway.io/common/batch_kline/5/1/GBPUSD' # 设置请求头 # 申请API KEY: www.infoway.io headers = { 'User-Agent': 'Mozilla/5.0', 'Accept': 'application/json', 'apiKey': 'YOUR_API_KEY' } # 发送GET请求 response = requests.get(api_url, headers=headers) # 输出结果 print(f"HTTP code: {response.status_code}") print(f"message: {response.text}") 4. 处理 API 响应 请求发送后，您会收到一个 HTTP 响应。您需要检查响应状态码以确保请求成功，并解析返回的 JSON 数据。 { "t": "1752872400", "h": "0.74578", "o": "0.74527", "l": "0.74503", "c": "0.74503", "v": "45.0", "vw": "33.530460", "pc": "-0.09%", "pca": "-0.00065" } 5. 逐笔成交数据 如果需要查询货币对的逐笔成交数据，我们可以向下面这个地址发送请求： https://data.infoway.io/common/batch_trade/GBPUSD 返回示例如下： { "s": "USDGBP", "t": 1752875078529, "p": "0.74503", "v": "1.0", "vw": "0.745030", "td": 0 } 6. 外汇盘口 外汇盘口数据在量化交易中也是经常要用到的，请求地址是这个： https://data.infoway.io/common/batch_depth/{codes} 7. 注意事项 API Key 保密 您的 API Key 是访问数据的唯一凭证，请务必妥善保管，切勿泄露给第三方。在实际应用中，不建议将 API Key 硬编码在代码中，可以考虑使用环境变量或配置文件来管理。 请求频率限制 大多数数据接口提供商会对 API 请求频率进行限制（例如：每秒请求次数）。请务必查阅您所用接口的官方文档，了解并遵守其频率限制政策，避免因频繁请求而被暂时或永久封禁。 错误处理 在实际应用中，除了检查 HTTP 状态码外，还应增加更详细的错误处理机制。例如，对网络错误 (requests.exceptions.RequestException)、JSON 解析错误 (ValueError) 等进行捕获和处理，提高程序的健壮性。 数据准确性与延迟 实时行情数据可能存在微小延迟。对于需要高频交易或对延迟极度敏感的应用，请详细了解数据提供商的服务等级协议（SLA）和延迟保证。 def execute_buy_decision(context, data, stock, hist): name = get_security_info(stock).display_name 查询委托单异常: name 'trade_api' is not defined try: 1. 查询所有未成交委托单 orders = trade_api.get_orders() for order in orders: print( f"委托单号: {order['order_id']} | 股票: {order['symbol']} | 方向: {order['side']} | " f"价格: {order['price']} | 数量: {order['amount']} | 状态: {order['status']} | " f"下单时间: {order['datetime']}" ) for order in orders: if order['symbol'] == stock and order['status'] == '未成交': print(f"{name}({stock}) 已有未成交委托，跳过下单") return False except Exception as e: print(f"查询委托单异常: {e}") 初学者，已经有了委托单，但是发现委托单重复下了，现在需要撤销。 请大家不要客气，任何意见建议可以在这里评论提出。 被采纳后我们将奖励1G研究环境内存 3个月。 强烈建议增加一些python库，例如能访问数据库的pymysql，或者能访问网址的request等，有些策略需要用到自己的一些数据，不然开发不了... 球球各位大佬！！！现在我的所有买卖只在开盘的时候执行，用的是handle_bar函数。怎么实现每分钟的买卖呢？ 
荣君2025-07-20 发布
