From 15dc44e8e0aaf0de0e1b73d6ee16d99212ef021d Mon Sep 17 00:00:00 2001 From: YuWuKunCheng Date: Sat, 30 May 2026 20:06:59 +0800 Subject: [PATCH] =?UTF-8?q?=E5=9B=9E=E6=B5=8B=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- strategies.py | 90 +++++++++++++++++++++++++++++++++------------------ 1 file changed, 58 insertions(+), 32 deletions(-) diff --git a/strategies.py b/strategies.py index 6a12698..5d49d1f 100644 --- a/strategies.py +++ b/strategies.py @@ -657,35 +657,56 @@ class 随机数据(bt.feeds.DataBase): class 自定义实时数据源(bt.feed.DataBase): """ 一个用于模拟实时数据推送的数据源，继承自Backtrader的DataBase。 - 在实际应用中，你需要将“模拟生成数据”的部分，替换为接收WebSocket等推送数据的代码。 + + 支持两种数据输入方式： + 1. 手动投喂：调用 投喂数据(时间戳, 开, 高, 低, 收, 量) 从 WebSocket 回调等外部代码推送 + 2. 后台线程：传入 观察员 和 魔法 参数，start() 时自动启动线程获取历史数据 + (注意：观察员 需实现 读取任意数据 方法) + + 时间戳格式：Unix 时间戳（秒），为 int 类型。 """ - def __init__(self, 数据队列: queue.Queue, 观察员: "观察者", 魔法, **魔法参数): - # 调用父类初始化方法 + def __init__(self, 数据队列: queue.Queue = None, 观察员: "观察者" = None, 魔法=None, **魔法参数): super(自定义实时数据源, self).__init__() - # 存储外部传入的数据队列，用于接收实时数据 - self.数据队列 = 数据队列 - # 定义一个标志，用于控制数据加载的停止 + self.数据队列 = 数据队列 or queue.Queue() self.正在运行 = False self.观察员 = 观察员 self.魔法 = 魔法 self.__魔法参数 = 魔法参数 self.已有数据 = False + def 投喂数据(self, 时间戳: int, 开盘价: float, 最高价: float, 最低价: float, 收盘价: float, 成交量: float, 持仓量: float = 0): + """线程安全的外部数据推送接口，供 WebSocket 回调等外部代码调用。 + + :param 时间戳: Unix 时间戳（秒，int 类型） + :param 开盘价: 开盘价 + :param 最高价: 最高价 + :param 最低价: 最低价 + :param 收盘价: 收盘价 + :param 成交量: 成交量 + :param 持仓量: 持仓量（默认为 0） + """ + self.数据队列.put((时间戳, 开盘价, 最高价, 最低价, 收盘价, 成交量, 持仓量)) + def start(self): - """ - 数据源开始工作时的初始化操作。 - 这里可以不执行任何操作，或者启动数据接收线程等。 - """ print(f"[{datetime.now()}] 自定义数据源已启动...") self.正在运行 = True - def 运行回测(): - self.观察员.读取任意数据(self.魔法, **self.__魔法参数) - self.正在运行 = False + if self.观察员 is not None and self.魔法 is not None: + if not hasattr(self.观察员, "读取任意数据"): + print(f"[{datetime.now()}] 警告: 观察员没有 '读取任意数据' 方法，后台线程不会启动") + return - 回测线程 = threading.Thread(target=运行回测, daemon=True) - 回测线程.start() + def 运行回测(): + try: + self.观察员.读取任意数据(self.魔法, **self.__魔法参数) + except Exception as e: + print(f"[{datetime.now()}] 后台数据线程异常: {e}") + finally: + self.正在运行 = False + + 回测线程 = threading.Thread(target=运行回测, daemon=True) + 回测线程.start() def stop(self): """ @@ -696,32 +717,25 @@ class 自定义实时数据源(bt.feed.DataBase): print(f"[{datetime.now()}] 自定义数据源已停止。") def _load(self): - """ - (核心方法) Backtrader会循环调用此方法来获取数据。 - 每次被调用时，都应该返回新的数据行（一个K线/一个数据点）。 - 如果没有新数据，返回 False，Backtrader会等待下次调用。 - """ + """Backtrader 核心回调 — 阻塞等待数据，有数据时返回 True，数据源停止时返回 False。""" if not self.正在运行: return False - # 当没有新数据且系统仍在运行时，进入等待 - # 从外部队列获取新数据 while True: try: data_point = self.数据队列.get(timeout=0.5) break except queue.Empty: - if not self.已有数据: - continue - else: - if self.正在运行: - continue + if not self.正在运行: return False - # 解析数据元组 - dt, o, h, l, c, v, oi = data_point - # 将数据写入Backtrader的数据线 (lines) - self.lines.datetime[0] = bt.date2num(dt) + try: + dt, o, h, l, c, v, oi = data_point + except (ValueError, TypeError) as e: + print(f"[{datetime.now()}] 自定义数据源: 数据格式错误，跳过: {e}") + return True + + self.lines.datetime[0] = bt.date2num(datetime.utcfromtimestamp(int(dt))) self.lines.open[0] = o self.lines.high[0] = h self.lines.low[0] = l @@ -729,7 +743,6 @@ class 自定义实时数据源(bt.feed.DataBase): self.lines.volume[0] = v self.lines.openinterest[0] = oi self.已有数据 = True - # 返回 True 表示成功加载一行数据 return True @@ -946,6 +959,7 @@ class 回测(高级策略基类): def __init__(self): super().__init__() self.已处理信号 = set() + print(f"{self.p.观察员.__class__.__name__}: 加载完成...") def 获取开仓限价(self, 是否做多): # 根据缠论分型计算限价，若无则返回 None 使用基类默认逻辑 @@ -986,15 +1000,21 @@ class 回测(高级策略基类): def 检查买信号(self): if self.p.观察员.笔序列: k线 = self.p.观察员.缠论K线序列[-1] + if k线.买卖点信息: + print(f"回测-首 {self.p.观察员.__class__.__name__}", k线.买卖点信息) 首 = True if k线.买卖点信息 and "买" in next(iter(k线.买卖点信息)) else False if 首: + print(k线) 原始差值 = self.p.观察员.当前K线.序号 - k线.标的K线.序号 差值 = self.p.观察员.当前缠K.序号 - k线.序号 self.日志(f"首_买入信号差值: {原始差值}, {差值}, 观察员.当前K线 时间戳: {self.p.观察员.当前K线.时间戳}") k线 = self.p.观察员.缠论K线序列[-2] + if k线.买卖点信息: + print(f"回测-尾 {self.p.观察员.__class__.__name__}", k线.买卖点信息) 尾 = True if self.p.观察员.缠论K线序列[-2].买卖点信息 and "买" in next(iter(self.p.观察员.缠论K线序列[-2].买卖点信息)) else False if 尾: + print(k线) 原始差值 = self.p.观察员.当前K线.序号 - k线.标的K线.序号 差值 = self.p.观察员.当前缠K.序号 - k线.序号 self.日志(f"尾_买入信号差值: {原始差值}, {差值}, 观察员.当前K线 时间戳: {self.p.观察员.当前K线.时间戳}") @@ -1003,15 +1023,21 @@ class 回测(高级策略基类): def 检查卖信号(self): if self.p.观察员.笔序列: k线 = self.p.观察员.缠论K线序列[-1] + if k线.买卖点信息: + print(f"回测-首 {self.p.观察员.__class__.__name__}", k线.买卖点信息) 首 = True if k线.买卖点信息 and "卖" in next(iter(k线.买卖点信息)) else False if 首: + print(k线) 原始差值 = self.p.观察员.当前K线.序号 - k线.标的K线.序号 差值 = self.p.观察员.当前缠K.序号 - k线.序号 self.日志(f"首_买入信号差值: {原始差值}, {差值}, 观察员.当前K线 时间戳: {self.p.观察员.当前K线.时间戳}") k线 = self.p.观察员.缠论K线序列[-2] + if k线.买卖点信息: + print(f"回测-尾 {self.p.观察员.__class__.__name__}", k线.买卖点信息) 尾 = True if self.p.观察员.缠论K线序列[-2].买卖点信息 and "卖" in next(iter(self.p.观察员.缠论K线序列[-2].买卖点信息)) else False if 尾: + print(k线) 原始差值 = self.p.观察员.当前K线.序号 - k线.标的K线.序号 差值 = self.p.观察员.当前缠K.序号 - k线.序号 self.日志(f"尾_买入信号差值: {原始差值}, {差值}, 观察员.当前K线 时间戳: {self.p.观察员.当前K线.时间戳}")