36 浏览
0

    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]

报错信息

chaos 已回答的问题 11小时 前
0

这个得看完整报错,大概率还是代码问题,先根据报错内容检查一下代码

chaos 发表新评论 11小时 前

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

您正在查看1个答案中的1个,单击此处查看所有答案。