def update_system_config(self):
# 初始化交易记录,主要获取需要订阅的合约,区分日盘和夜盘
now = self.now
tomorrow = now + timedelta(days=1)
self.night_trade_end = tomorrow.replace(hour=1, minute=0, second=0, microsecond=0)
week = now.isocalendar().week
year = now.isocalendar().year
self.cur_year = year
self.cur_week = week
print(f”{now}开始交易, 交易文件年{year}, 周{week}”)
hour = now.hour
# 判断是日盘交易还是夜盘交易
if 9 <= hour < 16:
trade_period = ‘day’
else:
trade_period = ‘night’
self.trade_day = now
self.trade_period = trade_period
if self.trade_period==’day’:
self.logger.info(f'{self.trade_day.year}-{self.trade_day.month}-{self.trade_day.day}日盘开始’)
if self.trade_period==’night’:
self.logger.info(f'{self.trade_day.year}-{self.trade_day.month}-{self.trade_day.day}夜盘开始’)
# 获取交易标的
# 仅需要标的和方向{‘au’: ‘buy’}
self.target_product, self.trading_days = self.extract_trade_target(trade_period, year, week)
print(self.target_product, self.trading_days)
self.force_close_datetime = datetime.strptime(f”{self.trading_days[-1]} 14:50:00″, ‘%Y%m%d %H:%M:%S’)
self.valid_datetime = datetime.strptime(f”{self.trading_days[-1]} 16:01:00″, ‘%Y%m%d %H:%M:%S’)
# 初始化组件
self.kline_manager = KlineManager(self.api, self.logger)
# 注册需要监控的品种
self._register_symbols()
self.signal_generator = SignalGenerator(self.kline_manager, self.logger)
print(‘信号系统初始化’)
self.signal_generator.initial_indicator()
print(‘信号系统初始化完成’)
# 交易执行器
self.trading_enabled = True
print(‘交易系统初始化’)
self.trade_executor = TradeExecutor(self.api, self.kline_manager.product_2_full_symbols, self.product_trade_info, self.target_product, logger=self.logger)
print(‘交易系统初始化完成’)
初步定位到是交易执行器部分有问题,如果注释掉行情可以正常播放
class TradeExecutor:
def __init__(self, api: TqApi, product_2_full_symbols: Dict, product_trade_info: Dict, target_product: Dict, logger=None):
self.api = api
self.product_2_full_symbols = product_2_full_symbols
self.target_product = target_product
self.product_trade_info = product_trade_info
self.logger = logger or logging.getLogger(__name__)
# 初始化交易执行器
self.trade_executor: Dict[str, TargetPosTask] = {}
self.executor_open_history: Dict[str, SignalData] = {}
for product, full_symbol in product_2_full_symbols.items():
if product not in self.trade_executor:
self.trade_executor[product] = TargetPosTask(self.api, full_symbol)
def execute_signal(self, product: str, last_kline: KlineData, signal_data: SignalData):
if signal_data is not None:
full_symbol = signal_data.symbol
direction = signal_data.signal_type
if direction == ‘close’:
# 平仓
logger.info(f’平仓{product}’)
if product in self.executor_open_history:
target_pos = 0
self.trade_executor[product].set_target_volume(target_pos)
# 从开仓历史中删除记录
logger.info(f’平仓信息: {signal_data}’)
del self.executor_open_history[product]
else:
print(f”{product} 仅开{self.target_product[product][‘direction’]}, 当前信号{direction}”)
# 开仓信号buy和sell,在没有开仓记录时,且与基本面方向一致时开仓
if (product not in self.executor_open_history) and (self.target_product[product][‘direction’]==direction):
target_notion = self.target_product[product][‘target_notion’]
target_pos = target_notion / (last_kline.close * self.product_trade_info[product][‘multipilier’])
target_pos = target_pos if direction==’buy’ else -target_pos
if target_pos > 0:
self.trade_executor[product].set_target_volume(target_pos)
self.executor_open_history[product] = signal_data
logger.info(f’开仓信息: {signal_data}, 张数{target_pos}’)
else:
logger.info(f”{product}不执行任何买卖”)
return
def force_close(self, product: str):
if product in self.executor_open_history:
target_pos = 0
self.trade_executor[product].set_target_volume(target_pos)
logger.info(f’本周最后一个交易日强平: {product}’)
del self.executor_open_history[product]
报错信息

有其他问题欢迎加入官方Q群748265037一起交流