import time import os import json from simulation_trader import SimulationTrader from xtquant import xtconstant from logger_config import get_logger # 获取日志记录器 logger = get_logger('strategy') # 策略仓位管理 strategy_positions = { 'real': {}, # 存储实盘策略持仓 'simulation': {} # 存储模拟交易策略持仓 } strategy_trades = { 'real': {}, # 存储实盘策略交易记录 'simulation': {} # 存储模拟交易策略交易记录 } pending_orders = { 'real': {}, # 存储实盘未完成委托 'simulation': {} # 存储模拟交易未完成委托 } class StrategyPositionManager: """策略持仓管理器，负责管理不同策略的持仓情况""" @staticmethod def get_trader_type(trader): """根据交易实例确定交易类型 Args: trader: 交易实例 Returns: str: 'simulation'或'real' """ return 'simulation' if isinstance(trader, SimulationTrader) else 'real' @staticmethod def update_strategy_position(trader, strategy_name, code, direction, amount): """更新策略持仓 Args: trader: 交易实例 strategy_name: 策略名称 code: 股票代码 direction: 'buy'或'sell' amount: 交易数量 """ if not strategy_name: return # 判断交易类型 trader_type = StrategyPositionManager.get_trader_type(trader) # 确保策略在字典中 if strategy_name not in strategy_positions[trader_type]: strategy_positions[trader_type][strategy_name] = {} try: # 获取交易实例持仓情况 actual_positions = trader.get_positions() code_position = next((pos for pos in actual_positions if pos.get('stock_code') == code), None) # 记录实际持仓总量 actual_total = code_position.get('volume', 0) if code_position else 0 actual_can_use = code_position.get('can_use_volume', 0) if code_position else 0 logger.info(f"实际持仓 - 代码: {code}, 总量: {actual_total}, 可用: {actual_can_use}") # 如果股票代码在持仓字典中不存在，初始化它 if code not in strategy_positions[trader_type][strategy_name]: strategy_positions[trader_type][strategy_name][code] = { 'total_amount': 0, 'closeable_amount': 0 } # 直接使用实际持仓数据更新策略持仓 strategy_positions[trader_type][strategy_name][code]['total_amount'] = actual_total strategy_positions[trader_type][strategy_name][code]['closeable_amount'] = actual_can_use logger.info(f"更新策略持仓 - 交易类型: {trader_type}, 策略: {strategy_name}, 代码: {code}, 方向: {direction}, 数量: {amount}, 总量: {strategy_positions[trader_type][strategy_name][code]['total_amount']}, 可用: {strategy_positions[trader_type][strategy_name][code]['closeable_amount']}") except Exception as e: logger.error(f"获取实际持仓失败: {str(e)}") # 异常情况下只记录错误，不尝试更新持仓 # 移除total_amount为0的持仓 if code in strategy_positions[trader_type][strategy_name] and strategy_positions[trader_type][strategy_name][code]['total_amount'] <= 0: del strategy_positions[trader_type][strategy_name][code] @staticmethod def update_pending_orders(trader): """更新未完成委托状态 Args: trader: 交易实例 """ try: # 判断当前交易类型 trader_type = StrategyPositionManager.get_trader_type(trader) # 获取今日委托 today_entrusts = trader.get_today_entrust() # 更新委托状态 for order_id, order_info in list(pending_orders[trader_type].items()): entrust = next((e for e in today_entrusts if e.get('order_id') == order_id), None) if entrust: if entrust.get('order_status') in [xtconstant.ORDER_SUCCEEDED, xtconstant.ORDER_PART_SUCC]: # 成交量计算 traded_amount = int(entrust.get('traded_volume', 0)) # 更新策略持仓 StrategyPositionManager.update_strategy_position( trader, order_info['strategy_name'], order_info['code'], order_info['direction'], traded_amount ) # 如果完全成交，从待处理列表中移除 if entrust.get('order_status') == xtconstant.ORDER_SUCCEEDED: del pending_orders[trader_type][order_id] # 如果已撤单、废单等终态，也从待处理列表中移除 elif entrust.get('order_status') in [xtconstant.ORDER_CANCELED, xtconstant.ORDER_JUNK]: del pending_orders[trader_type][order_id] except Exception as e: logger.error(f"更新未完成委托状态失败: {str(e)}") @staticmethod def add_pending_order(trader, order_id, strategy_name, code, price, amount, direction): """添加未完成委托 Args: trader: 交易实例 order_id: 委托编号 strategy_name: 策略名称 code: 股票代码 price: 委托价格 amount: 委托数量 direction: 交易方向，'buy'或'sell' """ if not order_id or order_id == 'simulation': return # 判断当前交易类型 trader_type = StrategyPositionManager.get_trader_type(trader) # 添加到未完成委托列表 pending_orders[trader_type][order_id] = { 'strategy_name': strategy_name, 'code': code, 'price': price, 'amount': amount, 'direction': direction, 'created_time': time.time() } # 同时记录到交易历史 if strategy_name: if strategy_name not in strategy_trades[trader_type]: strategy_trades[trader_type][strategy_name] = [] strategy_trades[trader_type][strategy_name].append({ 'time': time.strftime('%Y-%m-%d %H:%M:%S'), 'type': direction, 'code': code, 'price': price, 'amount': amount, 'order_id': order_id, 'status': 'pending' }) logger.info(f"添加未完成委托: {order_id}, 交易类型: {trader_type}, 策略: {strategy_name}, 代码: {code}, 方向: {direction}") @staticmethod def clean_timeout_orders(): """清理超时委托""" current_time = time.time() # 遍历实盘和模拟两种类型的委托 for trader_type in ['real', 'simulation']: for order_id, order_info in list(pending_orders[trader_type].items()): # 超过24小时的委托视为超时 if current_time - order_info['created_time'] > 24 * 60 * 60: del pending_orders[trader_type][order_id] @staticmethod def load_strategy_data(): """加载策略数据""" global strategy_positions, strategy_trades, pending_orders try: if os.path.exists('strategy_data.json'): with open('strategy_data.json', 'r') as f: data = json.load(f) # 直接使用新版数据结构，不再兼容旧版格式 strategy_positions = data.get('positions', {'real': {}, 'simulation': {}}) strategy_trades = data.get('trades', {'real': {}, 'simulation': {}}) pending_orders = data.get('pending_orders', {'real': {}, 'simulation': {}}) # 确保数据结构完整 if 'real' not in strategy_positions: strategy_positions['real'] = {} if 'simulation' not in strategy_positions: strategy_positions['simulation'] = {} if 'real' not in strategy_trades: strategy_trades['real'] = {} if 'simulation' not in strategy_trades: strategy_trades['simulation'] = {} if 'real' not in pending_orders: pending_orders['real'] = {} if 'simulation' not in pending_orders: pending_orders['simulation'] = {} logger.info("已加载策略数据") logger.info(f"实盘策略数: {len(strategy_positions['real'])}, 模拟策略数: {len(strategy_positions['simulation'])}") except Exception as e: logger.error(f"加载策略数据失败: {str(e)}") # 初始化空数据结构 strategy_positions = {'real': {}, 'simulation': {}} strategy_trades = {'real': {}, 'simulation': {}} pending_orders = {'real': {}, 'simulation': {}} @staticmethod def save_strategy_data(): """保存策略数据""" try: with open('strategy_data.json', 'w') as f: json.dump({ 'positions': strategy_positions, 'trades': strategy_trades, 'pending_orders': pending_orders }, f) except Exception as e: logger.error(f"保存策略数据失败: {str(e)}") @staticmethod def get_strategy_positions(trader, strategy_name=None): """获取策略持仓 Args: trader: 交易实例 strategy_name: 策略名称，如果为None，返回所有持仓 Returns: 如果strategy_name为None，返回交易实例的所有持仓 否则返回指定策略的持仓 """ # 判断当前交易类型 trader_type = StrategyPositionManager.get_trader_type(trader) # 如果指定了策略名称，返回该策略的持仓 if strategy_name: # 获取真实账户持仓，用于计算可交易量 real_positions = trader.get_positions() real_positions_map = {} for pos in real_positions: # 使用xt_trader返回的字段名 if 'stock_code' in pos and 'can_use_volume' in pos: real_positions_map[pos['stock_code']] = pos # 如果该策略没有记录，返回空列表 if strategy_name not in strategy_positions[trader_type]: logger.info(f"Strategy {strategy_name} has no positions in {trader_type} mode") return [] # 合并策略持仓和真实持仓的可交易量 result = [] for code, pos_info in strategy_positions[trader_type][strategy_name].items(): # 忽略total_amount为0的持仓 if pos_info['total_amount'] <= 0: continue # 使用真实账户的可交易量作为策略的可交易量上限 real_pos = real_positions_map.get(code, {}) closeable = min(pos_info['total_amount'], real_pos.get('can_use_volume', 0)) result.append({ code: { 'total_amount': pos_info['total_amount'], 'closeable_amount': closeable } }) logger.info(f"Strategy {strategy_name} positions in {trader_type} mode: {result}") return result # 否则返回原始持仓 positions = trader.get_positions() logger.info(f"Positions in {trader_type} mode: {positions}") return positions @staticmethod def clear_strategy(trader, strategy_name): """清除指定策略的持仓管理数据 Args: trader: 交易实例 strategy_name: 策略名称 Returns: tuple: (success, message) success: 是否成功清除 message: 提示信息 """ if not strategy_name: return False, "缺少策略名称参数" # 判断当前交易类型 trader_type = StrategyPositionManager.get_trader_type(trader) # 检查策略是否存在于当前交易类型中 if strategy_name in strategy_positions[trader_type]: # 从策略持仓字典中删除该策略 del strategy_positions[trader_type][strategy_name] # 清除该策略的交易记录 if strategy_name in strategy_trades[trader_type]: del strategy_trades[trader_type][strategy_name] # 清除与该策略相关的未完成委托 for order_id, order_info in list(pending_orders[trader_type].items()): if order_info.get('strategy_name') == strategy_name: del pending_orders[trader_type][order_id] # 保存更新后的策略数据 StrategyPositionManager.save_strategy_data() logger.info(f"成功清除策略持仓数据: {strategy_name} (交易类型: {trader_type})") return True, f"成功清除策略 '{strategy_name}' 的持仓数据 (交易类型: {trader_type})" else: logger.info(f"策略不存在或没有持仓数据: {strategy_name} (交易类型: {trader_type})") return True, f"策略 '{strategy_name}' 不存在或没有持仓数据 (交易类型: {trader_type})"