aitrader对接mini qmt实盘教程
实盘代码、qmt、ptrade、聚宽等
3
帖子
2
发布者
415
浏览
-
AI量化实验室星球专栏的文章的地址:
https://articles.zsxq.com/id_ep1u4ddsxpls.html前提条件:本实盘教程运行时间为14:50~14:57之间。也可在9:30之后验证前一天的执行结果,
运行流程
初始化日志模块初始化qmt模块
初始化回测引擎
判断运行时间
运行回测任务
执行实盘交易
# 实盘运行文件.py # 获取运行小时数 14点50之前 获取前一交易日的日期为end_date 之后end_date为当天日期 now = datetime.now() if now.hour > 14: end_date = now.strftime('%Y%m%d') elif now.hour == 14 and now.minute >= 50: end_date = now.strftime('%Y%m%d') else: end_date = (now - pd.Timedelta(days=1)).strftime('%Y%m%d') # 初始化任务 task = tasks.ma_energy_rolling() # task.symbols = xtdata.get_stock_list_in_sector('沪深300') task.buy_at_least_count = 3 # 同时满足3个因子才买入 task.symbols = ["516510.SH","513060.SH","513330.SH","515790.SH","159998.SZ","515030.SH","159995.SZ","159985.SZ","515880.SH","513520.SH","162719.SZ","512100.SH","512660.SH","159949.SZ","512510.SH","513500.SH","159928.SZ","513100.SH", "159819.SZ"] task.start_date = '20241101' task.end_date = end_date # 设置结束日期 task.current_day_trade = True # 获取当天交易信号 task.initial_capital = 50_000 # 设置初始资金 res = self.engine.run(task)
-
生成交易操作
使用上一步的res, res.get_transactions() 获取最终持仓比例通过qmt获取当前可用持仓
通过回测获取最终持仓比例
通过qmt获取上述股票的当前价格
通过当前可用持仓和最终持仓比例和单价计算出需要买入和卖出的股票和数量
# 计算交易操作.py 由deepseek生成 def dynamic_position_a_stock(current_positions, target_positions, prices, total_capital, min_lot=100, safety_margin=0.1): """ A股动态仓位计算(考虑最小交易单位100股) 参数: current_positions (dict) : 当前持仓数量,如{'A': 10000, 'B': 20000} target_positions (dict) : 目标持仓百分比,如{'A': 0.4, 'C': 0.6} prices (dict) : 当前股价,如{'A': 10.5, 'B': 20.0, 'C': 15.0} total_capital (float) : 总可用资金 min_lot (int) : 最小交易单位(默认100股) safety_margin (float) : 安全资金比例 返回: (目标持仓, 买卖数量, 所需资金, 买卖信息) """ available_capital = total_capital * (1 - safety_margin) # 阶段1:计算理论目标股数 target_values = {k: available_capital * v for k, v in target_positions.items()} theory_shares = {k: int(v / prices[k]) for k, v in target_values.items()} # 阶段2:整百处理(向下取整保证不超资金) target_shares = { k: max((s // min_lot) * min_lot, 0) # 至少0股 for k, s in theory_shares.items() } # 阶段3:计算交易数量(考虑当前持仓) delta = {} for k in set(current_positions.keys()).union(target_shares.keys()): current = current_positions.get(k, 0) target = target_shares.get(k, 0) delta[k] = target - current # 阶段4:交易数量整百处理(严格遵循交易规则) adjusted_delta = {} for k, d in delta.items(): if d == 0: adj = 0 elif d > 0: # 买入 adj = (d // min_lot) * min_lot # 向下取整 else: # 卖出 adj = -((abs(d) // min_lot) * min_lot) # 向下取整绝对值 adjusted_delta[k] = adj # 阶段5:计算实际目标仓位 final_target = {k: current_positions.get(k, 0) + adjusted_delta.get(k, 0) for k in set(current_positions.keys()).union(adjusted_delta.keys())} # 阶段6:资金验证 needed_cash = sum(final_target.get(k, 0) * prices.get(k, 0) for k in set(final_target.keys()).union(prices.keys())) overflow = needed_cash - available_capital # 阶段7:资金不足时的自动缩减 if overflow > 0: scale = (available_capital - sum(current_positions.get(k, 0) * prices.get(k, 0) for k in current_positions.keys())) / overflow scaled_delta = {k: int(d * scale) // min_lot * min_lot for k, d in adjusted_delta.items()} final_target = {k: current_positions.get(k, 0) + scaled_delta.get(k, 0) for k in set(current_positions.keys()).union(scaled_delta.keys())} adjusted_delta = scaled_delta # 阶段8:生成买卖信息 buy_sell_info = {} for k, d in adjusted_delta.items(): if d > 0: buy_sell_info[k] = f"买入 {d} 股" elif d < 0: buy_sell_info[k] = f"卖出 {abs(d)} 股" else: buy_sell_info[k] = "无操作" return final_target, adjusted_delta, needed_cash, buy_sell_info # 使用示例 if __name__ == "__main__": current = {'Aa': 10000, 'B': 20000} # 当前持仓 target = {'Aa': 0.5, 'C': 0.5} # 目标持仓百分比 prices = {'Aa': 1.12, 'B': 0.56, 'C': 0.75} # 当前股价 capital = 50000 # 总可用资金 final_target, delta, needed, buy_sell_info = dynamic_position_a_stock(current, target, prices, capital) print(f"当前持仓: {current}") print(f"目标持仓: {final_target}") print(f"交易数量: {delta}") print(f"所需资金: {needed}元") print(f"买卖信息: {buy_sell_info}")
-