YuWuKunCheng
2166 lines
90 KiB
Python
2166 lines
90 KiB
Python
import importlib.metadata
|
||
import platform
|
||
import random
|
||
import os
|
||
import math
|
||
import asyncio
|
||
import io
|
||
import json
|
||
import ast
|
||
import signal
|
||
import struct
|
||
import sys
|
||
import time
|
||
import queue
|
||
import traceback
|
||
import threading
|
||
from datetime import datetime, timedelta
|
||
from enum import Enum
|
||
from pathlib import Path
|
||
from random import seed, randint, uniform, choice, choices
|
||
from threading import Thread
|
||
from typing import (
|
||
List,
|
||
Optional,
|
||
Tuple,
|
||
SupportsInt,
|
||
Generator,
|
||
SupportsIndex,
|
||
Union,
|
||
Self,
|
||
Final,
|
||
Dict,
|
||
Any,
|
||
final,
|
||
)
|
||
|
||
|
||
import requests
|
||
from fastapi import FastAPI, WebSocketDisconnect, Request, WebSocket
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.templating import Jinja2Templates
|
||
|
||
from jinja2 import Environment, FileSystemLoader
|
||
|
||
from pydantic import BaseModel
|
||
import backtrader as bt
|
||
|
||
import chanlun
|
||
from chanlun import *
|
||
from strategies import *
|
||
|
||
chanlun.set_扩展线段模式(False)
|
||
|
||
import chanlun.chan_external as cet
|
||
|
||
|
||
def Nil(*args, **kwargs):
|
||
return None
|
||
|
||
|
||
def 获取模块版本():
|
||
versions = {}
|
||
|
||
# 1.
|
||
try:
|
||
versions["chanlun"] = importlib.metadata.version("chanlun")
|
||
except importlib.metadata.PackageNotFoundError:
|
||
pass
|
||
# 2.
|
||
try:
|
||
versions["fastapi"] = importlib.metadata.version("fastapi")
|
||
except importlib.metadata.PackageNotFoundError:
|
||
pass
|
||
try:
|
||
versions["requests"] = importlib.metadata.version("requests")
|
||
except importlib.metadata.PackageNotFoundError:
|
||
pass
|
||
|
||
# 3. 回测框架（你在用 backtrader 或类似）
|
||
try:
|
||
versions["backtrader"] = importlib.metadata.version("backtrader")
|
||
except importlib.metadata.PackageNotFoundError:
|
||
pass
|
||
|
||
# 4. 配置/模型（你这个缠论配置用了 pydantic）
|
||
try:
|
||
versions["pydantic"] = importlib.metadata.version("pydantic")
|
||
except importlib.metadata.PackageNotFoundError:
|
||
pass
|
||
|
||
return versions
|
||
|
||
|
||
def 收集异常信息(exception: Exception, 上下文: dict = None):
|
||
"""
|
||
万能异常收集函数
|
||
:param exception: 捕获到的异常
|
||
:param 上下文: 自定义环境数据（当前K线、品种、配置等）
|
||
:return: 完整错误报告
|
||
"""
|
||
# 1. 基础信息
|
||
exc_type, exc_obj, exc_tb = sys.exc_info()
|
||
文件名 = exc_tb.tb_frame.f_code.co_filename
|
||
行号 = exc_tb.tb_lineno
|
||
函数名 = exc_tb.tb_frame.f_code.co_name
|
||
|
||
# 2. 完整堆栈
|
||
堆栈 = "".join(traceback.format_exception(exc_type, exc_obj, exc_tb))
|
||
|
||
# 3. 代码片段
|
||
代码行 = open(文件名, "r", encoding="utf-8").readlines()[行号 - 1].strip()
|
||
|
||
# 4. 系统信息
|
||
系统信息 = {
|
||
"时间": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"Python版本": sys.version,
|
||
"系统": platform.platform(),
|
||
}
|
||
|
||
# 5. 组装最终报告
|
||
错误报告 = f"""
|
||
==================== 程序异常 ====================
|
||
异常类型: {exc_type.__name__}
|
||
异常信息: {str(exception)}
|
||
文件: {文件名}
|
||
行号: {行号}
|
||
函数: {函数名}
|
||
代码行: {代码行}
|
||
|
||
-------------------- 堆栈信息 --------------------
|
||
{堆栈}
|
||
|
||
-------------------- 系统信息 --------------------
|
||
{系统信息}
|
||
-------------------- 模块信息 --------------------
|
||
{获取模块版本()}
|
||
-------------------- 上下文信息 --------------------
|
||
{上下文 or "无"}
|
||
==================================================
|
||
"""
|
||
return 错误报告
|
||
|
||
|
||
class 时间周期:
|
||
def __init__(self, 秒: int, 是否单笔交易: bool = False):
|
||
self._秒 = 秒
|
||
self.是否单笔交易 = 是否单笔交易
|
||
|
||
def __repr__(self):
|
||
return f"时间周期<{self._秒}, {self.是否单笔交易}>"
|
||
|
||
def __str__(self):
|
||
return f"时间周期<{self._秒}, {self.是否单笔交易}>"
|
||
|
||
def __int__(self):
|
||
return int(self._秒)
|
||
|
||
@classmethod
|
||
def BitstampSupport(cls):
|
||
return {60, 180, 300, 900, 1800, 3600, 7200, 14400, 21600, 43200, 86400, 259200}
|
||
|
||
@classmethod
|
||
def 秒(cls, value: int):
|
||
return cls(value)
|
||
|
||
@classmethod
|
||
def 分(cls, value: int):
|
||
return cls(60 * value)
|
||
|
||
@classmethod
|
||
def 时(cls, value: int):
|
||
return cls(60 * 60 * value)
|
||
|
||
@classmethod
|
||
def 天(cls, value: int):
|
||
return cls(60 * 60 * 24 * value)
|
||
|
||
@classmethod
|
||
def 周(cls, value: int):
|
||
return cls(60 * 60 * 24 * 7 * value)
|
||
|
||
@classmethod
|
||
def 找到最大可整除周期(cls, 输入秒数: int) -> str:
|
||
"""
|
||
输入秒数 → 返回 最大可整除的周期秒数
|
||
周期范围：
|
||
1~59分钟、1~23小时、1~28天
|
||
"""
|
||
周期列表 = []
|
||
|
||
# 1~59分钟
|
||
for m in range(1, 60):
|
||
周期列表.append(m * 60)
|
||
|
||
# 1~23小时
|
||
for h in range(1, 24):
|
||
周期列表.append(h * 3600)
|
||
|
||
# 1~28天
|
||
for d in range(1, 29):
|
||
周期列表.append(d * 86400)
|
||
|
||
# 从大到小排序
|
||
周期列表.sort(reverse=True)
|
||
|
||
# 找第一个能整除的
|
||
for 周期秒 in 周期列表:
|
||
if 输入秒数 % 周期秒 == 0:
|
||
return cls.秒数转周期(周期秒)
|
||
|
||
return "1"
|
||
|
||
@classmethod
|
||
def 秒数转周期(cls, 秒数: int) -> str:
|
||
"""
|
||
智能选择最精确的周期单位
|
||
|
||
Args:
|
||
秒数: 输入的秒数值，必须是正整数
|
||
|
||
Returns:
|
||
周期字符串，格式为：数字 + 单位（H=小时，D=天，W=周，M=月）
|
||
"""
|
||
if not isinstance(秒数, int) or 秒数 <= 0:
|
||
raise ValueError("秒数必须是正整数")
|
||
|
||
一分钟秒数 = 60
|
||
一小时秒数 = 一分钟秒数 * 60 # 3600
|
||
一天秒数 = 24 * 一小时秒数
|
||
一周秒数 = 7 * 一天秒数
|
||
一月秒数 = 30 * 一天秒数
|
||
|
||
# 计算各周期单位
|
||
月数 = 秒数 // 一月秒数
|
||
月余秒 = 秒数 % 一月秒数
|
||
|
||
周数 = 秒数 // 一周秒数
|
||
周余秒 = 秒数 % 一周秒数
|
||
|
||
天数 = 秒数 // 一天秒数
|
||
天余秒 = 秒数 % 一天秒数
|
||
|
||
小时数 = 秒数 // 一小时秒数
|
||
时余数 = 秒数 % 一小时秒数
|
||
|
||
分钟数 = 秒数 // 一分钟秒数
|
||
分钟余数 = 秒数 % 一分钟秒数
|
||
|
||
if 分钟余数:
|
||
return str(秒数)
|
||
|
||
# 选择最精确的单位
|
||
# 优先选择余数为0的单位，如果没有，选择余数最小的单位
|
||
|
||
# 找出所有可能的表示方式及其余数
|
||
选项 = [(月数, "M", 月余秒), (周数, "W", 周余秒), (天数, "D", 天余秒), (小时数, "H", 时余数), (分钟数, "", 分钟余数)]
|
||
|
||
# 过滤掉数值为0的选项（小时除外）
|
||
有效选项 = [(值, 单位, 余数) for 值, 单位, 余数 in 选项 if 值 > 0]
|
||
|
||
# 如果没有有效选项，使用小时
|
||
if not 有效选项:
|
||
return str(秒数)
|
||
|
||
# 优先选择余数为0的选项
|
||
无余数选项 = [(值, 单位, 余数) for 值, 单位, 余数 in 有效选项 if 余数 == 0]
|
||
if 无余数选项:
|
||
# 选择单位最大的无余数选项
|
||
值, 单位, _ = min(无余数选项, key=lambda x: (x[0], len(x[1])))
|
||
return f"{值}{单位}"
|
||
|
||
# 如果没有无余数选项，选择余数最小的选项
|
||
值, 单位, _ = min(有效选项, key=lambda x: x[2])
|
||
return f"{值}{单位}"
|
||
|
||
@classmethod
|
||
def 周期转秒数(cls, 周期字符串: str) -> int:
|
||
"""
|
||
将带单位的周期字符串转换为秒数
|
||
|
||
Args:
|
||
周期字符串: 带单位的周期字符串，如 "15M", "60", "25H", "2D", "1W"
|
||
|
||
Returns:
|
||
对应的秒数值
|
||
|
||
Raises:
|
||
ValueError: 如果输入格式不正确或包含无效字符
|
||
"""
|
||
# 去除字符串两端的空格
|
||
周期字符串 = 周期字符串.strip().upper()
|
||
|
||
# 如果字符串为空，抛出异常
|
||
if not 周期字符串:
|
||
raise ValueError("周期字符串不能为空")
|
||
|
||
# 检查字符串是否以单位结尾
|
||
if 周期字符串[-1].isalpha():
|
||
# 提取数值部分和单位部分
|
||
单位 = 周期字符串[-1]
|
||
数值部分 = 周期字符串[:-1]
|
||
|
||
# 验证数值部分是否为有效数字
|
||
if not 数值部分.isdigit():
|
||
raise ValueError(f"无效的数值部分: {数值部分}")
|
||
|
||
数值 = int(数值部分)
|
||
|
||
# 根据单位计算秒数
|
||
if 单位 == "M": # 月
|
||
return 数值 * 2592000
|
||
elif 单位 == "H": # 小时
|
||
return 数值 * 3600
|
||
elif 单位 == "D": # 天
|
||
return 数值 * 86400
|
||
elif 单位 == "W": # 周
|
||
return 数值 * 604800
|
||
else:
|
||
raise ValueError(f"不支持的单位: {单位}")
|
||
else:
|
||
# 没有单位，默认为分钟
|
||
if not 周期字符串.isdigit():
|
||
raise ValueError(f"无效的数值: {周期字符串}")
|
||
|
||
数值 = int(周期字符串)
|
||
return 数值 * 60
|
||
|
||
|
||
class 指令:
|
||
增: Final[str] = "APPEND"
|
||
改: Final[str] = "MODIFY"
|
||
删: Final[str] = "REMOVE"
|
||
|
||
def __init__(self, 命令: str, 备注: str) -> None:
|
||
self.指令 = 命令
|
||
self.备注 = 备注
|
||
|
||
def __str__(self):
|
||
return f"{self.指令.upper()}"
|
||
|
||
def __repr__(self):
|
||
return f"{self.指令.upper()}"
|
||
|
||
@classmethod
|
||
def 添加(cls, 标识: str) -> Self:
|
||
return cls(cls.增, 标识)
|
||
|
||
@classmethod
|
||
def 修改(cls, 标识: str) -> Self:
|
||
return cls(cls.改, 标识)
|
||
|
||
@classmethod
|
||
def 删除(cls, 标识: str) -> Self:
|
||
return cls(cls.删, 标识)
|
||
|
||
|
||
class 图表展示序列(list):
|
||
def __init__(self, 观察员: "观察者"):
|
||
super().__init__()
|
||
self.观察员 = 观察员
|
||
self.序号 = 0
|
||
self.__类型标识 = None
|
||
|
||
def append(self, __object):
|
||
if self.序号 > 0:
|
||
if __object.标识 != self.__类型标识:
|
||
...
|
||
self.图表刷新(self[-1], sys._getframe().f_lineno)
|
||
|
||
else:
|
||
self.__类型标识 = __object.标识
|
||
super().append(__object)
|
||
self.图表添加(__object, sys._getframe().f_lineno)
|
||
self.序号 += 1
|
||
|
||
if __object.标识 in ("线段", "线段<线段>"):
|
||
if self.观察员 and self.观察员.配置.线段内部中枢图显:
|
||
段: 虚线 = __object
|
||
段.合_中枢序列 = 图表展示序列(self.观察员)
|
||
段.实_中枢序列 = 图表展示序列(self.观察员)
|
||
段.虚_中枢序列 = 图表展示序列(self.观察员)
|
||
|
||
def pop(self, __index: SupportsIndex = -1):
|
||
弹出 = super().pop(__index)
|
||
self.图表移除(弹出, sys._getframe().f_lineno)
|
||
self.序号 -= 1
|
||
return 弹出
|
||
|
||
def clear(self) -> None:
|
||
self.序号 = 0
|
||
super().clear()
|
||
|
||
def 尾部刷新(self, 行号: int):
|
||
if self.序号:
|
||
self.图表刷新(self[-1], 行号)
|
||
|
||
def 图表添加(self, 实线: Union["虚线", "中枢"], 行号: int):
|
||
self.观察员 and self.观察员.报信(实线, 指令.添加(实线.标识), 行号)
|
||
|
||
def 图表移除(self, 实线: Union["虚线", "中枢"], 行号: int):
|
||
self.观察员 and self.观察员.报信(实线, 指令.删除(实线.标识), 行号)
|
||
|
||
def 图表刷新(self, 实线: Union["虚线", "中枢"], 行号: int):
|
||
self.观察员 and self.观察员.报信(实线, 指令.修改(实线.标识), 行号)
|
||
|
||
|
||
class 观察者(观察者):
|
||
当前事件循环: Any = None # if __name__ == "__main__" else asyncio.get_event_loop()
|
||
延迟时间: float = 0.01
|
||
|
||
def __init__(self, 符号: str, 周期: int, 数据通道: Optional[WebSocket], 配置: 缠论配置, 数据队列: Optional[queue.Queue] = None):
|
||
self.数据通道: Optional[Any] = 数据通道 # WebSocket
|
||
self.数据队列: queue.Queue = 数据队列
|
||
super().__init__(符号, 周期, 配置)
|
||
self.__终止时间戳: Optional[datetime] = 转化为时间戳(self.配置.手动终止) if self.配置.手动终止 else None
|
||
self.买卖点字典 = dict()
|
||
|
||
@final
|
||
def 增加原始K线(self, 普K: K线):
|
||
if self.__终止时间戳 and 普K.时间戳 > self.__终止时间戳:
|
||
return
|
||
|
||
if self.配置.展示标签("RawBar"):
|
||
self.报信(普K, 指令.添加("RawBar"), sys._getframe().f_lineno, 周期=普K.周期)
|
||
|
||
try:
|
||
super().增加原始K线(普K)
|
||
self.数据队列 and self.数据队列.put((普K.时间戳, 普K.开盘价, 普K.高, 普K.低, 普K.收盘价, 普K.成交量, 0))
|
||
if self.数据通道 is not None and self.配置.图表展示:
|
||
time.sleep(self.延迟时间)
|
||
try:
|
||
self.图表刷新()
|
||
self.识别买卖点()
|
||
except:
|
||
print("~~~~~~~~~~~~~~", self.当前K线)
|
||
traceback.print_exc()
|
||
except Exception as e:
|
||
路径 = f"./templates/{self.符号}_err-{self.周期}-{int(self.普通K线序列[0].时间戳)}-{int(self.普通K线序列[-1].时间戳)}"
|
||
K线.保存到DAT文件(
|
||
路径 + ".nb",
|
||
self.普通K线序列,
|
||
)
|
||
self.配置.保存配置(路径 + ".json")
|
||
|
||
with open(路径 + ".log", "w") as f:
|
||
f.write(收集异常信息(e))
|
||
|
||
traceback.print_exc()
|
||
print(f"K线数据已保存在: {路径}.nb")
|
||
print(f"当前配置已保存在: {路径}.json")
|
||
print(f"详细错误信息已保存在: {路径}.log")
|
||
raise e
|
||
|
||
def 重置基础序列(self):
|
||
self.买卖点字典 = dict()
|
||
super().重置基础序列()
|
||
return
|
||
self.笔序列: List[虚线] = [] if not self.配置.图表展示 else 图表展示序列(self)
|
||
self.笔_中枢序列: List[中枢] = [] if not self.配置.图表展示 else 图表展示序列(self)
|
||
|
||
self.线段序列组: List[List[虚线],] = [] # 线段, 线段<线段>，线段<线段<线段>>...
|
||
self.中枢序列组: List[List[中枢],] = []
|
||
for i in range(self.线段分析层次):
|
||
self.线段序列组.append(list() if not self.配置.图表展示 else 图表展示序列(self))
|
||
self.中枢序列组.append(list() if not self.配置.图表展示 else 图表展示序列(self))
|
||
|
||
self.扩展线段序列组: List[List[虚线],] = [] # 扩展线段, 扩展线段<扩展线段>, 扩展线段<扩展线段<扩展线段>>...
|
||
self.扩展中枢序列组: List[List[中枢],] = []
|
||
for i in range(self.扩展线段分析层次):
|
||
self.扩展线段序列组.append(list() if not self.配置.图表展示 else 图表展示序列(self))
|
||
self.扩展中枢序列组.append(list() if not self.配置.图表展示 else 图表展示序列(self))
|
||
|
||
self.混合扩展线段序列组: List[List[虚线],] = [] # 扩展线段<线段>, 扩展线段<线段<线段>>, 扩展线段<线段<线段<线段>>>...
|
||
self.混合扩展中枢序列组: List[List[中枢],] = []
|
||
for i in range(self.混合扩展线段分析层次):
|
||
self.混合扩展线段序列组.append(list() if not self.配置.图表展示 else 图表展示序列(self))
|
||
self.混合扩展中枢序列组.append(list() if not self.配置.图表展示 else 图表展示序列(self))
|
||
|
||
def 读取任意数据(self, 魔法, **魔法参数):
|
||
魔法(**魔法参数)
|
||
return self
|
||
|
||
def 加载本地数据(self, 文件路径: str):
|
||
self.重置基础序列()
|
||
with open(文件路径, "rb") as f:
|
||
buffer = f.read()
|
||
size = struct.calcsize(">6d")
|
||
for i in range(len(buffer) // size):
|
||
k线 = K线.读取大端字节数组(buffer[i * size : i * size + size], self.周期, self.标识)
|
||
self.增加原始K线(k线)
|
||
|
||
def 静态重新分析(self):
|
||
self.买卖点字典 = dict()
|
||
super().静态重新分析()
|
||
|
||
def 添加买卖点(self, 特征: str, 买卖点分型: 分型, 序号: str, 级别: str):
|
||
当前买卖点: 买卖点 = 买卖点.生成买卖点(特征, 序号, 级别, 买卖点分型, self.当前缠K)
|
||
if "事后" in 特征:
|
||
当前买卖点.失效K线 = self.当前缠K
|
||
偏移 = self.配置.买卖点偏移
|
||
if 当前买卖点.偏移 > 偏移 and "事后" not in 特征:
|
||
return
|
||
|
||
买卖点序列 = self.买卖点字典.get(特征, set())
|
||
self.买卖点字典[特征] = 买卖点序列
|
||
活跃序列 = [点 for 点 in 买卖点序列 if 点.失效K线 is None]
|
||
活跃时间戳序列 = [点.买卖点K线.时间戳 for 点 in 活跃序列]
|
||
|
||
if self.配置.买卖点与MACD柱强相关 and not 买卖点分型.中.与MACD柱子匹配:
|
||
return
|
||
分型匹配 = 买卖点分型.与MACD柱子分型匹配
|
||
柱子匹配 = 买卖点分型.中.与MACD柱子匹配
|
||
|
||
rsi匹配 = 买卖点分型.中.与RSI匹配
|
||
kdj匹配 = 买卖点分型.中.与KDJ匹配
|
||
|
||
当前买卖点.备注 = f"{self.标识}" + 当前买卖点.备注
|
||
当前买卖点.备注 = 当前买卖点.备注 + f"_{买卖点分型.强度}"
|
||
if 分型匹配 is not None and not 分型匹配:
|
||
当前买卖点.备注 = 当前买卖点.备注 + "_非MACD分型"
|
||
|
||
if not 柱子匹配:
|
||
当前买卖点.备注 = 当前买卖点.备注 + "_非普K柱子匹配"
|
||
|
||
if rsi匹配 is not None and not rsi匹配:
|
||
当前买卖点.备注 = 当前买卖点.备注 + "_非RSI匹配"
|
||
|
||
if kdj匹配 is not None and not kdj匹配:
|
||
当前买卖点.备注 = 当前买卖点.备注 + "_非KDJ匹配"
|
||
|
||
if not self.配置.买卖点激进识别 and not 买卖点分型.右:
|
||
pass # return
|
||
|
||
if 当前买卖点.买卖点K线.时间戳 not in 活跃时间戳序列:
|
||
买卖点序列.add(当前买卖点)
|
||
当前买卖点.买卖点K线.买卖点信息.add(当前买卖点.备注)
|
||
print(当前买卖点, type(当前买卖点), 当前买卖点.备注)
|
||
self.报信(当前买卖点, 指令.添加(当前买卖点.备注), sys._getframe().f_lineno)
|
||
|
||
def 图表刷新(self):
|
||
def 报信(序列):
|
||
getattr(序列, "尾部刷新", Nil)(行号=-1)
|
||
|
||
报信(self.笔序列)
|
||
报信(self.笔_中枢序列)
|
||
for i in range(self.线段分析层次):
|
||
报信(self.线段序列组[i])
|
||
报信(self.中枢序列组[i])
|
||
for i in range(self.扩展线段分析层次):
|
||
报信(self.扩展线段序列组[i])
|
||
报信(self.扩展中枢序列组[i])
|
||
for i in range(self.混合扩展线段分析层次):
|
||
报信(self.混合扩展线段序列组[i])
|
||
报信(self.混合扩展中枢序列组[i])
|
||
return
|
||
|
||
def 报信(序列):
|
||
for 对象 in 序列[-3:]:
|
||
self.报信(对象, 指令.添加(对象.标识), 0)
|
||
|
||
报信(self.笔序列)
|
||
报信(self.笔_中枢序列)
|
||
|
||
for i in range(self.线段分析层次):
|
||
报信(self.线段序列组[i])
|
||
报信(self.中枢序列组[i])
|
||
for i in range(self.扩展线段分析层次):
|
||
报信(self.扩展线段序列组[i])
|
||
报信(self.扩展中枢序列组[i])
|
||
for i in range(self.混合扩展线段分析层次):
|
||
报信(self.混合扩展线段序列组[i])
|
||
报信(self.混合扩展中枢序列组[i])
|
||
|
||
# self.将图表数据固化到本地()
|
||
|
||
def 报信(self, 对象: Any, 命令: 指令, 行号, **kwargs) -> None:
|
||
if self.数据通道 is None or not self.配置.图表展示:
|
||
return
|
||
|
||
message = dict()
|
||
|
||
if type(对象) is K线:
|
||
message["type"] = "realtime"
|
||
message["timestamp"] = str(对象.时间戳)
|
||
message["open"] = 对象.开盘价
|
||
message["high"] = 对象.高
|
||
message["low"] = 对象.低
|
||
message["close"] = 对象.收盘价
|
||
message["volume"] = 对象.成交量
|
||
|
||
配色表 = {
|
||
"笔": "#6C4D7E",
|
||
"线段": "#FEC187",
|
||
"线段<线段>": "#8F6048", # 以线段为基础的特征序列线段
|
||
"扩展线段": "#09a4ff", # 以笔为基础的
|
||
"扩展线段<线段>": "#07d59e", # 以线段为基础的
|
||
"扩展线段<扩展线段>": "#ff29e3",
|
||
"扩展线段<扩展线段<线段>>": "#07d59e",
|
||
}
|
||
for k, v in list(配色表.items()):
|
||
配色表[f"中枢<{k}>"] = v
|
||
|
||
if type(对象) is 买卖点:
|
||
message["type"] = "shape"
|
||
message["cmd"] = 命令.指令.upper()
|
||
message["id"] = str(id(对象))
|
||
message["name"] = "arrow_down" if 对象.类型.是卖点 else "arrow_up"
|
||
message["points"] = [{"time": int(对象.买卖点K线.时间戳), "price": 对象.买卖点K线.分型特征值}]
|
||
arrowColor = "#FF2800" if 对象.类型.是卖点 else "#00FF22"
|
||
text = f"{str(对象.偏移)}, {对象.破位值}, {对象.备注}"
|
||
message["overrides"] = {
|
||
"color": "#CC62FF",
|
||
"arrowColor": arrowColor,
|
||
"text": text,
|
||
"title": 对象.备注.split("_")[0],
|
||
"showLabel": False if 对象.偏移 <= 1 else True,
|
||
}
|
||
|
||
if type(对象) in (虚线, 中枢, 线段特征) and not self.配置.展示标签(对象.标识):
|
||
return
|
||
|
||
if type(对象) in (虚线, 中枢, 线段特征):
|
||
图标 = 对象.图表标题
|
||
message["type"] = "shape"
|
||
message["cmd"] = 命令.指令.upper()
|
||
message["id"] = 图标
|
||
message["name"] = "trend_line" if type(对象) is not 中枢 else "rectangle"
|
||
if 命令.指令 != 指令.删:
|
||
message["points"] = [
|
||
{"time": int(缠论K线.时间戳对齐(self.基础缠K序列, 对象.文.中)), "price": 对象.文.分型特征值 if type(对象) is not 中枢 else 对象.高},
|
||
{"time": int(缠论K线.时间戳对齐(self.基础缠K序列, 对象.武.中)), "price": 对象.武.分型特征值 if type(对象) is not 中枢 else 对象.低},
|
||
]
|
||
linewidths = {"笔": 1, "线段": 2, "走势": 3, "线段特征": 2}
|
||
message["overrides"] = {
|
||
"bold": True,
|
||
"linecolor": 配色表.get(对象.标识, 配色表["笔"]),
|
||
"textcolor": "#000000",
|
||
"text": "",
|
||
"title": 图标,
|
||
"linewidth": linewidths.get(对象.标识, 2) if type(对象) is not 中枢 else linewidths.get(对象.基础序列[0].标识, 2),
|
||
"backgroundColor": "rgba(242, 54, 69, 0.2)" if 对象.方向 is 相对方向.向下 else "rgba(76, 175, 80, 0.2)", # 上下上 为 红色，反之为 绿色,
|
||
"color": 配色表.get(对象.标识, 配色表["笔"]) if type(对象) is not 中枢 else 配色表.get(对象.基础序列[0].标识, 配色表["笔"]),
|
||
"textColor": 配色表.get(对象.标识, 配色表["笔"]) if type(对象) is not 中枢 else 配色表.get(对象.基础序列[0].标识, 配色表["笔"]),
|
||
"visible": False,
|
||
}
|
||
|
||
if 对象.标识 in ("笔", "线段", "线段<线段>", "中枢<笔>", "中枢<线段>"):
|
||
message["overrides"]["visible"] = True
|
||
|
||
if type(对象) is not 线段特征:
|
||
message["overrides"]["text"] = f"{对象.标识} {对象.序号} 周期:{self.周期} {getattr(对象, '四象', '')} {getattr(对象, '特征序列状态', '')} {getattr(对象, '级别', '')} {getattr(对象, '备注', '')}"
|
||
|
||
if type(对象) is 中枢:
|
||
message["overrides"]["text"] = f"{对象.标识} {对象.序号} 周期:{self.周期} 基础序列数量: {len(对象.基础序列)}"
|
||
|
||
if 对象.标识 in ("线段", "线段<线段>"):
|
||
message["overrides"]["text"] = f"{对象.标识} {对象.序号} 周期:{self.周期} {线段.四象(对象)} {线段.特征序列状态(对象)} {getattr(对象, '级别', '')} {getattr(对象, '备注', '')}"
|
||
|
||
if 对象.标识 in ("线段", "线段<线段>", "线段<线段<线段>>"):
|
||
message["overrides"]["text"] += f" 内部中枢数量:{len(对象.实_中枢序列)}"
|
||
|
||
if type(对象) is 线段特征:
|
||
message["overrides"].update({"linecolor": "#F1C40F" if 对象.方向 is 相对方向.向下 else "#fbc02d", "linewidth": 4, "linestyle": 1})
|
||
message["overrides"]["visible"] = True
|
||
|
||
if type(对象) is 中枢:
|
||
del message["overrides"]["textcolor"]
|
||
del message["overrides"]["linecolor"]
|
||
else:
|
||
del message["overrides"]["textColor"]
|
||
del message["overrides"]["backgroundColor"]
|
||
del message["overrides"]["color"]
|
||
|
||
if len(message) < 3:
|
||
return
|
||
|
||
if self.数据通道 is not None and self.配置.图表展示:
|
||
asyncio.set_event_loop(观察者.当前事件循环)
|
||
asyncio.ensure_future(self.数据通道.send_text(json.dumps(message)))
|
||
return
|
||
|
||
def 将图表数据固化到本地(self, static_shapes=None):
|
||
template_path = "./templates/static.html"
|
||
# 初始化 Jinja2 环境，模板目录为当前目录
|
||
env = Environment(loader=FileSystemLoader(os.path.dirname(template_path) or "."))
|
||
template = env.get_template(os.path.basename(template_path))
|
||
resolution = 时间周期.找到最大可整除周期(self.周期)
|
||
static_data = {"bars": [[int(k.时间戳), k.开盘价, k.高, k.低, k.收盘价, k.成交量] for k in self.普通K线序列]}
|
||
|
||
配色表 = {
|
||
"笔": "#6C4D7E",
|
||
"线段": "#FEC187",
|
||
"线段<线段>": "#8F6048", # 以线段为基础的特征序列线段
|
||
"扩展线段": "#09a4ff", # 以笔为基础的
|
||
"扩展线段<线段>": "#07d59e", # 以线段为基础的
|
||
"扩展线段<扩展线段>": "#ff29e3",
|
||
"扩展线段<扩展线段<线段>>": "#07d59e",
|
||
}
|
||
|
||
for k, v in list(配色表.items()):
|
||
配色表[f"中枢<{k}>"] = v
|
||
|
||
if not static_shapes:
|
||
static_shapes = []
|
||
全部 = []
|
||
for o in dir(self):
|
||
if "序列" in o and "K线序列" not in o and "分型" not in o:
|
||
oo = getattr(self, o)
|
||
if isinstance(oo, list):
|
||
全部.extend(oo)
|
||
for o in self.买卖点字典.values():
|
||
全部.extend(o)
|
||
|
||
for 对象 in 全部:
|
||
if type(对象) in (笔, 线段, 中枢, 线段特征):
|
||
message = dict()
|
||
图标 = 对象.图表标题
|
||
message["type"] = "shape"
|
||
message["id"] = 图标
|
||
message["shapeType"] = "trend_line" if type(对象) is not 中枢 else "rectangle"
|
||
message["points"] = [
|
||
{"time": int(缠论K线.时间戳对齐(self.缠论K线序列, 对象.文.中)), "price": 对象.文.分型特征值 if type(对象) is not 中枢 else 对象.高},
|
||
{"time": int(缠论K线.时间戳对齐(self.缠论K线序列, 对象.武.中)), "price": 对象.武.分型特征值 if type(对象) is not 中枢 else 对象.低},
|
||
]
|
||
linewidths = {"笔": 1, "线段": 2, "走势": 3, "线段特征": 2}
|
||
message["overrides"] = {
|
||
"bold": True,
|
||
"linecolor": 配色表.get(对象.标识, 配色表["笔"]),
|
||
"textcolor": "#000000",
|
||
"text": "",
|
||
"title": 图标,
|
||
"linewidth": linewidths.get(对象.标识, 2) if type(对象) is not 中枢 else linewidths.get(对象[0].标识, 2),
|
||
"backgroundColor": "rgba(242, 54, 69, 0.2)" if 对象.方向 is 相对方向.向下 else "rgba(76, 175, 80, 0.2)",
|
||
# 上下上 为 红色，反之为 绿色,
|
||
"color": 配色表.get(对象.标识, 配色表["笔"]) if type(对象) is not 中枢 else 配色表.get(对象[0].标识, 配色表["笔"]),
|
||
"textColor": 配色表.get(对象.标识, 配色表["笔"]) if type(对象) is not 中枢 else 配色表.get(对象[0].标识, 配色表["笔"]),
|
||
}
|
||
|
||
if type(对象) is not 线段特征:
|
||
message["overrides"]["text"] = f"{对象.标识} {对象.序号} 周期:{self.周期} {getattr(对象, '四象', '')} {getattr(对象, '特征序列状态', '')} {getattr(对象, '级别', '')} "
|
||
|
||
if type(对象) is 线段:
|
||
message["overrides"]["text"] += f" 内部中枢数量:{len(对象.实_中枢序列)}"
|
||
|
||
if type(对象) is 线段特征:
|
||
message["overrides"].update({"linecolor": "#F1C40F" if 对象.方向 is 相对方向.向下 else "#fbc02d", "linewidth": 4, "linestyle": 1})
|
||
|
||
if type(对象) is 中枢:
|
||
del message["overrides"]["textcolor"]
|
||
del message["overrides"]["linecolor"]
|
||
else:
|
||
del message["overrides"]["textColor"]
|
||
del message["overrides"]["backgroundColor"]
|
||
del message["overrides"]["color"]
|
||
static_shapes.append(message)
|
||
continue
|
||
|
||
if type(对象) is 买卖点:
|
||
message = dict()
|
||
message["type"] = "shape"
|
||
message["id"] = str(id(对象))
|
||
message["shapeType"] = "arrow_down" if 对象.类型.是卖点 else "arrow_up"
|
||
message["points"] = [{"time": int(对象.买卖点K线.时间戳), "price": 对象.买卖点K线.分型特征值}]
|
||
arrowColor = "#FF2800" if 对象.类型.是卖点 else "#00FF22"
|
||
text = f"{str(对象.偏移)}, {对象.破位值}, {对象.备注}"
|
||
message["overrides"] = {
|
||
"color": "#CC62FF",
|
||
"arrowColor": arrowColor,
|
||
"text": text,
|
||
"title": 对象.备注.split("_")[0],
|
||
"showLabel": False,
|
||
}
|
||
static_shapes.append(message)
|
||
continue
|
||
|
||
else:
|
||
print(type(对象), 对象)
|
||
for item in static_shapes:
|
||
if item.get("overrides") and item["overrides"].get("intervalsVisibilities"):
|
||
del item["overrides"]["intervalsVisibilities"]
|
||
# 渲染
|
||
rendered_html = template.render(static_data=static_data, static_shapes=static_shapes, symbol=self.符号, interval=resolution, chan_config=self.配置.to_dict())
|
||
|
||
output_file = "./new.html"
|
||
# 写入输出文件
|
||
with open(output_file, "w", encoding="utf-8") as f:
|
||
f.write(rendered_html)
|
||
|
||
print(f"✅ 成功生成文件: {output_file}, 需要另行开启服务器 如 python -m http.server 8081")
|
||
|
||
@classmethod
|
||
def 读取数据文件(cls, 文件路径: str, ws=None, 配置=缠论配置(), *, 观察员: Optional[观察者] = None) -> Self:
|
||
"""加载数据文件
|
||
:param 文件路径: 数据文件路径 格式如: btcusd-300-1631772074-1632222374.nb
|
||
:param 配置: 缠论配置
|
||
:param 观察员: 可选，已有观察者实例；不传则自动创建
|
||
:return: 观察者实例
|
||
"""
|
||
if "_err-" in str(文件路径) and os.path.exists(str(文件路径).replace(".nb", ".json")):
|
||
异常配置 = 缠论配置.加载配置(str(文件路径).replace(".nb", ".json"))
|
||
差异 = 缠论配置().对比(异常配置)
|
||
传入差异 = 缠论配置().对比(配置)
|
||
传入差异.update(差异)
|
||
配置 = 缠论配置(**传入差异)
|
||
print(f"加载异常配置+传入差异: {传入差异}")
|
||
|
||
name = Path(文件路径).name.split(".")[0]
|
||
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
|
||
if 观察员 is None:
|
||
观察员 = cls(符号=符号, 周期=int(周期), 数据通道=ws, 配置=配置)
|
||
else:
|
||
观察员.符号 = 符号
|
||
观察员.周期 = int(周期)
|
||
观察员.配置 = 配置
|
||
观察员.加载本地数据(文件路径)
|
||
return 观察员
|
||
|
||
def 识别买卖点(self):
|
||
"""
|
||
简单买卖策略
|
||
"""
|
||
if not self.笔序列:
|
||
return
|
||
if self.分型序列[-1].中.序号 + 2 < self.当前缠K.序号:
|
||
return
|
||
if self.分型序列[-1].强度 not in "强中":
|
||
pass
|
||
|
||
__代码执行器_全局声明__ = dir()
|
||
|
||
|
||
def 随机配置(随机源: Optional[random.Random] = None):
|
||
"""生成随机缠论配置，可传入独立的 Random 实例以保证线程安全"""
|
||
rng = 随机源 if 随机源 is not None else random.Random()
|
||
return 缠论配置.不推送().from_dict(
|
||
{
|
||
"缠K合并替换": rng.choice((True, False)),
|
||
"笔内元素数量": rng.randint(3, 9),
|
||
"笔内相同终点取舍": rng.choice((True, False)),
|
||
"笔内起始分型包含整笔": rng.choice((True, False)),
|
||
"笔内原始K线包含整笔": rng.choice((True, False)),
|
||
"笔次级成笔": rng.choice((True, False)),
|
||
"笔弱化": rng.choice((True, False)),
|
||
"笔弱化_原始数量": rng.randint(3, 9),
|
||
"线段_非缺口下穿刺": rng.choice((True, False)),
|
||
"线段_特征序列忽视老阴老阳": rng.choice((True, False)),
|
||
"线段_修正": rng.choice((True, False)),
|
||
"线段_缺口后紧急修正": rng.choice((True, False)),
|
||
"扩展线段_当下分析": rng.choice((True, False)),
|
||
"买卖点激进识别": rng.choice((True, False)),
|
||
"买卖点与MACD柱强相关": rng.choice((True, False)),
|
||
}
|
||
)
|
||
|
||
|
||
def 测试_随机生成(symbol: str = "btcusd", limit: int = 5000, freq: SupportsInt = 时间周期.分(5), ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置()):
|
||
def 魔法():
|
||
随机生成实例 = 观察者(symbol + "_gen", 周期=int(freq), 数据通道=ws, 配置=配置)
|
||
dt = datetime(2008, 8, 8)
|
||
原始K线 = K线.创建普K("随机", int(dt.timestamp()), 8888.55, 10000.00, 9000.22, 9527.33, 888, 0, int(freq))
|
||
随机生成实例.增加原始K线(原始K线)
|
||
for 方向 in 相对方向.从序列中机选(
|
||
int(limit),
|
||
[相对方向.向上, 相对方向.向上缺口, 相对方向.衔接向上, 相对方向.向下, 相对方向.向下缺口, 相对方向.衔接向下],
|
||
):
|
||
原始K线 = 原始K线.根据当前K线生成新K线(方向)
|
||
随机生成实例.增加原始K线(原始K线)
|
||
|
||
折线 = [元素.文.分型特征值 for 元素 in 随机生成实例.笔序列]
|
||
折线.append(随机生成实例.笔序列[-1].武.分型特征值)
|
||
print(折线)
|
||
|
||
return 随机生成实例
|
||
|
||
return 魔法
|
||
|
||
|
||
class Bitstamp:
|
||
@classmethod
|
||
def init(cls, 观察员_, size):
|
||
观察员 = 观察员_
|
||
left_date_timestamp = int(datetime.now().timestamp() * 1000)
|
||
left = int(left_date_timestamp / 1000) - 观察员.周期 * size
|
||
if left < 0:
|
||
raise RuntimeError
|
||
_next = left
|
||
while 1:
|
||
data = cls.ohlc(观察员.符号, 观察员.周期, _next, _next := _next + 观察员.周期 * 1000)
|
||
if not data.get("data"):
|
||
print(data)
|
||
raise ValueError("")
|
||
for bar in data["data"]["ohlc"]:
|
||
K = K线.创建普K(
|
||
观察员.符号,
|
||
int(bar["timestamp"]),
|
||
float(bar["open"]),
|
||
float(bar["high"]),
|
||
float(bar["low"]),
|
||
float(bar["close"]),
|
||
float(bar["volume"]),
|
||
0,
|
||
观察员.周期,
|
||
)
|
||
观察员.增加原始K线(K)
|
||
|
||
# start = int(data["data"]["ohlc"][0]["timestamp"])
|
||
end = int(data["data"]["ohlc"][-1]["timestamp"])
|
||
|
||
_next = end
|
||
if len(data["data"]["ohlc"]) < 100:
|
||
break
|
||
折线 = [元素.文.分型特征值 for 元素 in 观察员.笔序列]
|
||
折线.append(观察员.笔序列[-1].武.分型特征值)
|
||
# print(折线)
|
||
K线.保存到DAT文件(
|
||
f"./templates/{观察员.符号}-{观察员.周期}-{int(观察员.普通K线序列[0].时间戳)}-{int(观察员.普通K线序列[-1].时间戳)}.nb",
|
||
观察员.普通K线序列,
|
||
)
|
||
K线.保存到DAT文件(
|
||
"./templates/last.nb",
|
||
观察员.普通K线序列,
|
||
)
|
||
|
||
@staticmethod
|
||
def 获取K线数据(数量: int, 符号: str, 周期: int, obj):
|
||
end_ts = int(datetime.now().timestamp())
|
||
left = end_ts - 周期 * 数量
|
||
if left < 0:
|
||
raise RuntimeError
|
||
_next = left
|
||
while 1:
|
||
data = Bitstamp.ohlc(符号, 周期, _next, _next := _next + 周期 * 1000)
|
||
if not data.get("data"):
|
||
print(data)
|
||
raise ValueError
|
||
for bar in data["data"]["ohlc"]:
|
||
K = K线.创建普K(
|
||
符号,
|
||
转化为时间戳(int(bar["timestamp"])),
|
||
float(bar["open"]),
|
||
float(bar["high"]),
|
||
float(bar["low"]),
|
||
float(bar["close"]),
|
||
float(bar["volume"]),
|
||
0,
|
||
周期,
|
||
)
|
||
obj.投喂K线(K)
|
||
|
||
# start = int(data["data"]["ohlc"][0]["timestamp"])
|
||
end = int(data["data"]["ohlc"][-1]["timestamp"])
|
||
|
||
_next = end
|
||
if len(data["data"]["ohlc"]) < 100:
|
||
break
|
||
|
||
@staticmethod
|
||
def ohlc(pair: str, step: int, start: int, end: int, length: int = 1000, retries: int = 3) -> Dict:
|
||
"""执行HTTP请求，带重试机制"""
|
||
url = f"https://www.bitstamp.net/api/v2/ohlc/{pair}/"
|
||
session = requests.Session()
|
||
session.headers = {
|
||
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:144.0) Gecko/20100101 Firefox/144.0",
|
||
# "content-type": "application/json",
|
||
}
|
||
"""proxies = {
|
||
"http": "http://127.0.0.1:10808",
|
||
"https": "http://127.0.0.1:10808",
|
||
}"""
|
||
|
||
params = {"step": step, "limit": length, "start": start, "end": end}
|
||
|
||
for attempt in range(retries):
|
||
try:
|
||
# resp = session.get(url, params=params, timeout=10, proxies=proxies)
|
||
resp = session.get(url, params=params, timeout=10)
|
||
resp.raise_for_status()
|
||
return resp.json()
|
||
except Exception as e:
|
||
print(f"请求失败 (尝试 {attempt + 1}/{retries}): {e}")
|
||
if attempt == retries - 1:
|
||
raise
|
||
time.sleep(2**attempt) # 指数退避
|
||
|
||
|
||
def 同步_跟踪回测(观察员: 观察者, 数据源: bt.feed.DataBase):
|
||
cerebro = bt.Cerebro()
|
||
cerebro.addstrategy(回测, 观察员=观察员)
|
||
|
||
# 收益与风险指标
|
||
cerebro.addanalyzer(bt.analyzers.TimeReturn, _name="时间收益率") # 需要指定timeframe? 默认用数据源的时间周期
|
||
cerebro.addanalyzer(bt.analyzers.AnnualReturn, _name="年度收益率")
|
||
cerebro.addanalyzer(bt.analyzers.Returns, _name="总体收益率")
|
||
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="夏普比率")
|
||
cerebro.addanalyzer(bt.analyzers.SharpeRatio_A, _name="年化夏普比率")
|
||
cerebro.addanalyzer(bt.analyzers.Calmar, _name="卡尔玛比率")
|
||
cerebro.addanalyzer(bt.analyzers.SQN, _name="系统质量指数")
|
||
cerebro.addanalyzer(bt.analyzers.VWR, _name="变异性加权回报")
|
||
|
||
# 风险与资金管理
|
||
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="回撤分析")
|
||
cerebro.addanalyzer(bt.analyzers.TimeDrawDown, _name="时间周期回撤") # 需要timeframe参数，下面会重设
|
||
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="交易分析")
|
||
cerebro.addanalyzer(bt.analyzers.PeriodStats, _name="周期统计") # 需要timeframe
|
||
cerebro.addanalyzer(bt.analyzers.Transactions, _name="交易记录")
|
||
cerebro.addanalyzer(bt.analyzers.PyFolio, _name="pyfolio导出")
|
||
|
||
# 其他
|
||
cerebro.addanalyzer(bt.analyzers.LogReturnsRolling, _name="滚动对数收益率") # 需要timeframe和period
|
||
|
||
cerebro.adddata(数据源)
|
||
|
||
cerebro.broker.setcash(1000000)
|
||
cerebro.broker.setcommission(commission=0.001) # 0.1%佣金
|
||
|
||
初始资金 = cerebro.broker.getvalue()
|
||
print("初始资金:", 初始资金)
|
||
results = cerebro.run(live=True)
|
||
最终资金 = cerebro.broker.getvalue()
|
||
|
||
strat = results[0]
|
||
print("回测结束，分析结果如下：")
|
||
print("=" * 60)
|
||
|
||
# 定义打印函数，安全获取分析结果
|
||
def 打印分析(名称, 分析器对象):
|
||
return
|
||
try:
|
||
result = 分析器对象.get_analysis()
|
||
print(f"\n【{名称}】")
|
||
# 格式化输出，如果是字典则打印键值对
|
||
if isinstance(result, dict):
|
||
for k, v in result.items():
|
||
print(f" {k}: {v}")
|
||
else:
|
||
print(f" {result}")
|
||
except Exception as e:
|
||
print(f"【{名称}】获取失败: {e}")
|
||
|
||
# 逐一打印各分析器结果
|
||
打印分析("时间收益率", strat.analyzers.时间收益率)
|
||
打印分析("年度收益率", strat.analyzers.年度收益率)
|
||
打印分析("总体收益率", strat.analyzers.总体收益率)
|
||
打印分析("夏普比率", strat.analyzers.夏普比率)
|
||
打印分析("年化夏普比率", strat.analyzers.年化夏普比率)
|
||
打印分析("卡尔玛比率", strat.analyzers.卡尔玛比率)
|
||
打印分析("系统质量指数", strat.analyzers.系统质量指数)
|
||
打印分析("变异性加权回报", strat.analyzers.变异性加权回报)
|
||
打印分析("回撤分析", strat.analyzers.回撤分析)
|
||
打印分析("时间周期回撤", strat.analyzers.时间周期回撤)
|
||
打印分析("交易分析", strat.analyzers.交易分析)
|
||
print(strat.analyzers.交易分析.get_analysis())
|
||
打印分析("周期统计", strat.analyzers.周期统计)
|
||
打印分析("交易记录", strat.analyzers.交易记录)
|
||
# pyfolio 分析器不直接打印，需额外调用导出函数，此处略
|
||
打印分析("滚动对数收益率", strat.analyzers.滚动对数收益率)
|
||
|
||
# 最终资金
|
||
print(f"\n最终账户价值: {cerebro.broker.getvalue():.2f}")
|
||
print("最终资金:", 最终资金, (最终资金 - 初始资金) / 初始资金)
|
||
|
||
|
||
def 测试_读取数据(观察员, ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置(线段内部中枢图显=False)):
|
||
def 魔法():
|
||
启动时间 = datetime.now()
|
||
观察者.读取数据文件(配置.加载文件路径, ws, 配置, 观察员=观察员)
|
||
# 观察员.分部分析()
|
||
消耗用时 = datetime.now() - 启动时间
|
||
print(消耗用时)
|
||
观察员.图表刷新()
|
||
return 观察员
|
||
|
||
return 魔法
|
||
|
||
|
||
def 测试_邮局数据(symbol: str = "btcusd", limit: int = 500, freq: SupportsInt = 时间周期.分(5), ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置(线段内部中枢图显=False)):
|
||
def 魔法():
|
||
观察员 = 观察者(symbol, int(freq), ws, 配置)
|
||
Bitstamp.init(观察员, int(limit))
|
||
观察员.图表刷新()
|
||
return 观察员
|
||
|
||
return 魔法
|
||
|
||
|
||
def 测试_读取上一次数据(名称: str = "btcusd", 数量: int = 500, 周期: SupportsInt = 时间周期.分(5), ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置(线段内部中枢图显=False)):
|
||
def 魔法():
|
||
观察员 = 观察者(名称, int(周期), ws, 配置)
|
||
观察员.加载本地数据("./templates/last.nb")
|
||
观察员.图表刷新()
|
||
return 观察员
|
||
|
||
return 魔法
|
||
|
||
|
||
def 测试_读取上一次数据_回测(名称: str = "btcusd", 数量: int = 500, 周期: SupportsInt = 时间周期.分(5), ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置(线段内部中枢图显=False)):
|
||
def 魔法():
|
||
数据队列 = queue.Queue(1)
|
||
观察员 = 观察者(名称, int(周期), ws, 配置, 数据队列)
|
||
数据源 = 自定义实时数据源(数据队列, 观察员, 观察员.加载本地数据, 文件路径="./templates/last.nb")
|
||
同步_跟踪回测(观察员, 数据源)
|
||
观察员.图表刷新()
|
||
return 观察员
|
||
|
||
return 魔法
|
||
|
||
|
||
def 测试_邮局数据_同步回测(symbol: str = "btcusd", limit: int = 500, freq: SupportsInt = 时间周期.分(5), ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置()):
|
||
def 魔法():
|
||
数据队列 = queue.Queue(1)
|
||
观察员 = 观察者(symbol, int(freq), ws, 配置)
|
||
观察员.数据队列 = 数据队列
|
||
数据源 = 自定义实时数据源(数据队列, 观察员, Bitstamp.init, size=int(limit), 观察员_=观察员)
|
||
同步_跟踪回测(观察员, 数据源)
|
||
观察员.图表刷新()
|
||
return 观察员
|
||
|
||
return 魔法
|
||
|
||
|
||
def 测试_周期合成(symbol: str = "btcusd", limit: int = 500, freq: SupportsInt = 时间周期.分(5), ws: Optional[WebSocket] = None, 配置: 缠论配置 = 缠论配置(), 配置组: Dict[int, 缠论配置] = None):
|
||
def 魔法():
|
||
周期组 = [int(freq), int(freq) * 5, int(freq) * 5 * 6]
|
||
多级别分析 = 立体分析器(symbol, 周期组, ws, 配置, 配置组)
|
||
Bitstamp.获取K线数据(int(limit), symbol, 周期组[0], 多级别分析)
|
||
return 多级别分析
|
||
|
||
return 魔法
|
||
|
||
|
||
app = FastAPI()
|
||
# 添加CORS中间件
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
app.mount(
|
||
"/charting_library",
|
||
StaticFiles(directory="charting_library"),
|
||
name="charting_library",
|
||
)
|
||
templates = Jinja2Templates(directory="templates")
|
||
|
||
|
||
class 代码执行器:
|
||
"""
|
||
在母体进程中安全执行 Python 代码（受限环境）。
|
||
支持超时（Unix 信号机制）、重置、帮助、历史。
|
||
警告：无法完全阻止恶意代码访问主进程，请仅用于可信环境！
|
||
"""
|
||
|
||
def __init__(self, 用户标识: str, 默认超时: float = 5.0):
|
||
self.图表观察员 = None
|
||
self.用户标识 = 用户标识
|
||
self.超时 = 默认超时
|
||
self.历史记录: List[Dict[str, Any]] = []
|
||
# 安全内置函数白名单
|
||
self.安全内置函数 = {
|
||
# 基础函数
|
||
"print": print,
|
||
"len": len,
|
||
"range": range,
|
||
"int": int,
|
||
"str": str,
|
||
"float": float,
|
||
"bool": bool,
|
||
"list": list,
|
||
"dict": dict,
|
||
"set": set,
|
||
"tuple": tuple,
|
||
"abs": abs,
|
||
"round": round,
|
||
"sum": sum,
|
||
"min": min,
|
||
"max": max,
|
||
"enumerate": enumerate,
|
||
"zip": zip,
|
||
"sorted": sorted,
|
||
"reversed": reversed,
|
||
"isinstance": isinstance,
|
||
"type": type,
|
||
"id": id,
|
||
"chr": chr,
|
||
"ord": ord,
|
||
"bin": bin,
|
||
"hex": hex,
|
||
"oct": oct,
|
||
"all": all,
|
||
"any": any,
|
||
"next": next,
|
||
"iter": iter,
|
||
# 常量
|
||
"True": True,
|
||
"False": False,
|
||
"None": None,
|
||
"dir": dir,
|
||
"math": math,
|
||
"random": random,
|
||
"datetime": datetime,
|
||
"timedelta": timedelta,
|
||
"time": __import__("time"),
|
||
"help": self.获取帮助,
|
||
"clear": self.重置,
|
||
}
|
||
self.安全内置函数.update({k: globals()[k] for k in __代码执行器_全局声明__ if k[0] != "_"})
|
||
# 初始化命名空间
|
||
self.重置()
|
||
|
||
def 设置图表观察员(self, observer):
|
||
self.图表观察员 = observer
|
||
self.全局命名空间["观察员"] = observer
|
||
|
||
def _代码安全检查(self, 代码字符串: str) -> Optional[str]:
|
||
"""
|
||
使用 AST 检查代码是否包含危险属性访问（如 .__class__ 或 ._xxx）。
|
||
返回 None 表示安全，否则返回错误信息。
|
||
"""
|
||
危险属性列表 = ["__class__", "__bases__", "__subclasses__", "__globals__", "__builtins__", "__import__", "__getattribute__", "__setattr__", "__delattr__", "__reduce__", "__reduce_ex__", "__code__"]
|
||
try:
|
||
树 = ast.parse(代码字符串)
|
||
except SyntaxError as e:
|
||
return f"语法错误: {e}"
|
||
for 节点 in ast.walk(树):
|
||
if isinstance(节点, ast.Attribute):
|
||
if 节点.attr in 危险属性列表 or 节点.attr.startswith("__"):
|
||
return f"禁止访问属性 '{节点.attr}'"
|
||
if isinstance(节点, ast.Call):
|
||
# 禁止调用内置的 __import__
|
||
if isinstance(节点.func, ast.Name) and 节点.func.id == "__import__":
|
||
return "禁止调用 __import__"
|
||
# 禁止 eval/exec
|
||
if isinstance(节点.func, ast.Name) and 节点.func.id in ("eval", "exec"):
|
||
return f"禁止使用 {节点.func.id}"
|
||
return None
|
||
|
||
def _超时处理(self, 信号编号, 帧):
|
||
"""信号处理函数，抛出超时异常"""
|
||
raise TimeoutError(f"代码执行超时（超过 {self.超时} 秒）")
|
||
|
||
def 执行(self, 代码字符串: str) -> Dict[str, Optional[str]]:
|
||
"""
|
||
在主进程中执行代码，返回 {"标准输出": str, "错误输出": str, "异常信息": str or None}
|
||
"""
|
||
# 安全检查
|
||
检查结果 = self._代码安全检查(代码字符串)
|
||
if 检查结果:
|
||
return {
|
||
"success": False,
|
||
"output": "",
|
||
"error": {"type": "安全检查", "message": 检查结果, "traceback": ""},
|
||
"stdout": "",
|
||
"stderr": "",
|
||
"print_output": "",
|
||
"execution_time": datetime.now().isoformat(),
|
||
}
|
||
|
||
# 重定向输出
|
||
原始stdout = sys.stdout
|
||
原始stderr = sys.stderr
|
||
stdout缓冲区 = io.StringIO()
|
||
stderr缓冲区 = io.StringIO()
|
||
sys.stdout = stdout缓冲区
|
||
sys.stderr = stderr缓冲区
|
||
|
||
异常信息 = None
|
||
# 保存原有信号处理（仅 Unix）
|
||
原有信号处理 = None
|
||
if hasattr(signal, "SIGALRM"):
|
||
原有信号处理 = signal.signal(signal.SIGALRM, self._超时处理)
|
||
signal.alarm(int(self.超时) + 1) # 设置超时秒数，多给1秒宽松
|
||
|
||
try:
|
||
# 使用受限命名空间执行
|
||
# 注意：每次执行使用同一个 self.全局命名空间 和 self.局部命名空间，以实现变量持久化
|
||
exec(代码字符串, self.全局命名空间, self.局部命名空间)
|
||
except TimeoutError as e:
|
||
异常信息 = traceback.format_exc()
|
||
异常信息 = {"type": type(e).__name__, "message": str(e), "traceback": traceback.format_exc()}
|
||
except Exception as e:
|
||
异常信息 = traceback.format_exc()
|
||
异常信息 = {"type": type(e).__name__, "message": str(e), "traceback": traceback.format_exc()}
|
||
finally:
|
||
# 取消超时报警
|
||
if hasattr(signal, "SIGALRM"):
|
||
signal.alarm(0)
|
||
if 原有信号处理:
|
||
signal.signal(signal.SIGALRM, 原有信号处理)
|
||
# 恢复输出
|
||
sys.stdout = 原始stdout
|
||
sys.stderr = 原始stderr
|
||
# 获取捕获的输出
|
||
标准输出 = stdout缓冲区.getvalue()
|
||
错误输出 = stderr缓冲区.getvalue()
|
||
# 记录历史
|
||
self.历史记录.append({"代码": 代码字符串, "结果": {"标准输出": 标准输出, "错误输出": 错误输出, "异常信息": 异常信息}})
|
||
结果 = {
|
||
"success": not 异常信息,
|
||
"output": 标准输出,
|
||
"error": 异常信息,
|
||
"stdout": 标准输出,
|
||
"stderr": 错误输出,
|
||
"print_output": 标准输出,
|
||
"execution_time": datetime.now().isoformat(),
|
||
}
|
||
return 结果
|
||
|
||
def 重置(self) -> None:
|
||
"""重置命名空间，清除所有已定义的变量"""
|
||
self.全局命名空间 = {
|
||
"__builtins__": self.安全内置函数,
|
||
"__name__": "__沙箱__",
|
||
}
|
||
self.局部命名空间 = {}
|
||
print("执行环境已重置")
|
||
|
||
def 获取帮助(self) -> str:
|
||
"""返回帮助信息"""
|
||
帮助文本 = "可用的内置函数/类型：\n"
|
||
for 名称 in sorted(self.安全内置函数.keys()):
|
||
if not 名称.startswith("__"): # 过滤内部名称
|
||
帮助文本 += f" - {名称}\n"
|
||
帮助文本 += "\n注意：不支持文件 I/O、系统命令、网络请求、属性访问（如 .__class__）。\n"
|
||
帮助文本 += f"当前超时设置：{self.超时} 秒\n"
|
||
帮助文本 += "使用 重置() 可清空变量，使用 设置超时(秒) 可修改超时。"
|
||
return 帮助文本
|
||
|
||
def 设置超时(self, 秒数: float) -> None:
|
||
"""动态修改超时时间"""
|
||
self.超时 = max(0.5, 秒数) # 至少0.5秒
|
||
print(f"超时已设置为 {self.超时} 秒")
|
||
|
||
def 获取历史(self, 最近条数: int = None) -> List[Dict]:
|
||
"""返回执行历史"""
|
||
if 最近条数 is None:
|
||
return self.历史记录.copy()
|
||
return self.历史记录[-最近条数:]
|
||
|
||
def 清空历史(self) -> None:
|
||
"""清空历史记录（不影响当前变量）"""
|
||
self.历史记录.clear()
|
||
|
||
def 关闭(self) -> None:
|
||
"""清理（预留）"""
|
||
pass
|
||
|
||
|
||
class 连接管理器:
|
||
def __init__(self):
|
||
self.活跃连接字典: Dict[str, WebSocket] = {}
|
||
self.环境字典: Dict[str, 代码执行器] = {}
|
||
self.图表观察员字典: Dict[str, 观察者] = {}
|
||
|
||
async def 进行连接(self, 用户标识: str, websocket: WebSocket):
|
||
await websocket.accept()
|
||
self.活跃连接字典[用户标识] = websocket
|
||
|
||
if 用户标识 not in self.环境字典:
|
||
self.环境字典[用户标识] = 代码执行器(用户标识)
|
||
|
||
print(f"[连接] 用户 {用户标识} 已连接")
|
||
|
||
def 切断连接(self, 用户标识: str):
|
||
if 用户标识 in self.活跃连接字典:
|
||
del self.活跃连接字典[用户标识]
|
||
if 用户标识 in self.环境字典:
|
||
del self.环境字典[用户标识]
|
||
if 用户标识 in self.图表观察员字典:
|
||
del self.图表观察员字典[用户标识]
|
||
|
||
print(f"[断开] 用户 {用户标识} 已断开")
|
||
|
||
async def 发送信息(self, 用户标识: str, message: Dict[str, Any]):
|
||
if 用户标识 in self.活跃连接字典:
|
||
try:
|
||
await self.活跃连接字典[用户标识].send_json(message)
|
||
except Exception as e:
|
||
print(f"[错误] 发送消息到 {用户标识} 失败: {e}")
|
||
|
||
def 设置图表观察员(self, 用户标识: str, observer):
|
||
self.图表观察员字典[用户标识] = observer
|
||
|
||
if 用户标识 in self.环境字典:
|
||
self.环境字典[用户标识].设置图表观察员(observer)
|
||
|
||
def 获取图表观察员(self, 用户标识: str):
|
||
return self.图表观察员字典.get(用户标识)
|
||
|
||
def 获取执行环境(self, 用户标识: str):
|
||
if 用户标识 not in self.环境字典:
|
||
self.环境字典[用户标识] = 代码执行器(用户标识)
|
||
return self.环境字典[用户标识]
|
||
|
||
|
||
全局连接管理器 = 连接管理器()
|
||
|
||
# 全局线程变量
|
||
主线程 = None
|
||
|
||
|
||
# ============ WebSocket端点 ============
|
||
@app.websocket("/ws/{user_id}")
|
||
async def 全局消息分发器(websocket: WebSocket, user_id: str):
|
||
"""统一的WebSocket端点，处理所有类型的消息"""
|
||
用户标识 = user_id
|
||
await 全局连接管理器.进行连接(用户标识, websocket)
|
||
|
||
try:
|
||
# 发送欢迎消息
|
||
await 全局连接管理器.发送信息(
|
||
用户标识,
|
||
{
|
||
"type": "connected",
|
||
"message": "✅ 已连接到服务器",
|
||
"用户标识": 用户标识,
|
||
"timestamp": datetime.now().isoformat(),
|
||
"endpoint": "unified",
|
||
},
|
||
)
|
||
|
||
while True:
|
||
try:
|
||
消息字典 = json.loads(await websocket.receive_text())
|
||
except WebSocketDisconnect:
|
||
全局连接管理器.切断连接(用户标识)
|
||
break
|
||
# 获取消息类型
|
||
消息类型 = 消息字典.get("type", "")
|
||
模块 = 消息字典.get("module", "chart") # 默认是chart模块
|
||
|
||
print(f"[消息] 用户 {用户标识} | 模块: {模块} | 类型: {消息类型}")
|
||
|
||
if 模块 == "python":
|
||
# Python执行相关消息
|
||
await 处理代码消息(用户标识, 消息字典)
|
||
elif 模块 == "chart":
|
||
# 图表相关消息
|
||
await 处理图表消息(用户标识, 消息字典, websocket)
|
||
else:
|
||
print(模块, 消息字典)
|
||
|
||
except WebSocketDisconnect:
|
||
全局连接管理器.切断连接(用户标识)
|
||
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
print(f"[错误] WebSocket处理异常: {e}")
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "error", "message": f"服务器错误: {str(e)}", "timestamp": datetime.now().isoformat()})
|
||
全局连接管理器.切断连接(用户标识)
|
||
|
||
|
||
async def 处理图表消息(用户标识: str, 消息字典: Dict, websocket: WebSocket):
|
||
"""处理图表消息"""
|
||
消息类型 = 消息字典.get("type", "")
|
||
|
||
if 消息类型 == "ready":
|
||
# 初始化分析器
|
||
symbol = 消息字典.get("symbol", "btcusd")
|
||
freq = 消息字典.get("freq", 300)
|
||
limit = 消息字典.get("limit", 500)
|
||
generator = 消息字典.get("generator", "True")
|
||
|
||
config = 消息字典.get("config", dict())
|
||
print("RAW config:", repr(config))
|
||
当前配置 = 缠论配置.from_dict(config)
|
||
print("", 当前配置.to_dict())
|
||
|
||
差异 = 缠论配置().对比(当前配置)
|
||
print(差异)
|
||
配置组 = 缠论配置.按序号重组字典(当前配置, config)
|
||
print(配置组)
|
||
|
||
# 停止现有线程
|
||
global 主线程
|
||
if 主线程 is not None:
|
||
主线程.join(1)
|
||
time.sleep(1)
|
||
主线程 = None
|
||
|
||
观察员 = 观察者("", 60, websocket, 当前配置)
|
||
# 创建新的分析器
|
||
if generator == "zqhc":
|
||
魔法 = 测试_周期合成(symbol=symbol, freq=freq, limit=limit, ws=websocket, 配置=当前配置, 配置组=配置组)
|
||
elif generator == "hc":
|
||
魔法 = 测试_邮局数据_同步回测(symbol=symbol, freq=freq, limit=limit, ws=websocket, 配置=当前配置)
|
||
|
||
elif generator == "ex":
|
||
魔法 = 测试_读取数据(观察员=观察员, ws=websocket, 配置=当前配置)
|
||
elif generator == "last":
|
||
魔法 = 测试_读取上一次数据(名称=symbol, 数量=limit, 周期=freq, ws=websocket, 配置=当前配置)
|
||
|
||
elif generator == "lasthc":
|
||
魔法 = 测试_读取上一次数据_回测(名称=symbol, 数量=limit, 周期=freq, ws=websocket, 配置=当前配置)
|
||
|
||
else:
|
||
魔法 = 测试_邮局数据(symbol=symbol, freq=freq, limit=limit, ws=websocket, 配置=当前配置)
|
||
|
||
def 数据加载线程():
|
||
try:
|
||
if generator == "ex":
|
||
全局连接管理器.设置图表观察员(用户标识, 观察员)
|
||
魔法()
|
||
else:
|
||
全局连接管理器.设置图表观察员(用户标识, 魔法())
|
||
print(f"[分析器] 用户 {用户标识} 的分析器已启动")
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
print(f"[分析器错误] {e}")
|
||
|
||
主线程 = Thread(target=数据加载线程, daemon=True)
|
||
主线程.start()
|
||
|
||
await 全局连接管理器.发送信息(
|
||
用户标识,
|
||
{
|
||
"type": "ready_ack",
|
||
"message": "图表分析器已启动",
|
||
"symbol": symbol,
|
||
"freq": freq,
|
||
"timestamp": datetime.now().isoformat(),
|
||
},
|
||
)
|
||
|
||
elif 消息类型 == "query_by_index":
|
||
观察员: 观察者 = 全局连接管理器.获取图表观察员(用户标识)
|
||
if 观察员 is not None:
|
||
符号, 周期, 数据类型, 序号 = 消息字典.get("index").split(":")
|
||
序号 = int(序号)
|
||
print(符号, 周期, 数据类型, 序号)
|
||
|
||
if type(观察员) is 立体分析器:
|
||
观察员 = 观察员._单体分析器[int(周期)]
|
||
|
||
try:
|
||
待发送消息 = {}
|
||
if 数据类型 == "中枢<笔>":
|
||
待发送消息.update({"index": 序号, "data": str(观察员.笔_中枢序列[序号])})
|
||
if 数据类型 == "笔":
|
||
待发送消息.update({"index": 序号, "data": str(观察员.笔序列[序号])})
|
||
|
||
if "中枢" in 数据类型 and 数据类型 != "中枢<笔>":
|
||
for i in range(观察员.中枢分析层次):
|
||
if 观察员.中枢序列组[i] and 观察员.中枢序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.中枢序列组[i][序号])})
|
||
for i in range(观察员.扩展中枢分析层次):
|
||
if 观察员.扩展中枢序列组[i] and 观察员.扩展中枢序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.扩展中枢序列组[i][序号])})
|
||
for i in range(观察员.混合扩展中枢分析层次):
|
||
if 观察员.混合扩展中枢序列组[i] and 观察员.混合扩展中枢序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.混合扩展中枢序列组[i][序号])})
|
||
|
||
elif "线段" in 数据类型 and 数据类型 != "笔":
|
||
for i in range(观察员.线段分析层次):
|
||
if 观察员.线段序列组[i] and 观察员.线段序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.线段序列组[i][序号])})
|
||
段 = 观察员.线段序列组[i][序号]
|
||
if 段._特征序列_显示:
|
||
段._特征序列_显示 = False
|
||
for 特征 in 段.特征序列:
|
||
if 特征 is not None:
|
||
观察员 and 观察员.报信(特征, 指令.删除(特征.标识), sys._getframe().f_lineno)
|
||
|
||
else:
|
||
段._特征序列_显示 = True
|
||
序号 = 0
|
||
for 特征 in 段.特征序列:
|
||
if 特征 is not None:
|
||
特征.序号 = 序号
|
||
特征.标识 = f"{段.文.中.标识}:{段.文.中.周期}:{段.标识}_特征序列_{序号}:{段.序号}"
|
||
观察员 and 观察员.报信(特征, 指令.添加(特征.标识), sys._getframe().f_lineno)
|
||
序号 += 1
|
||
|
||
for i in range(观察员.扩展线段分析层次):
|
||
if 观察员.扩展线段序列组[i] and 观察员.扩展线段序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.扩展线段序列组[i][序号])})
|
||
for i in range(观察员.混合扩展线段分析层次):
|
||
if 观察员.混合扩展线段序列组[i] and 观察员.混合扩展线段序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.混合扩展线段序列组[i][序号])})
|
||
|
||
if "_" in 数据类型 and "中枢" in 数据类型: # 线段_0_实_中枢<笔>
|
||
数据类型, 线序, 虚实合, 类型 = 数据类型.split("_")
|
||
|
||
段序号 = int(线序)
|
||
if 数据类型 == "线段":
|
||
段: 虚线 = 观察员.线段序列[段序号]
|
||
zs = getattr(段, f"{虚实合}_中枢序列")[序号]
|
||
待发送消息.update({"index": 序号, "data": str(zs)})
|
||
|
||
if 数据类型 == "线段<线段>":
|
||
段: 虚线 = 观察员.线段_线段序列[段序号]
|
||
zs = getattr(段, f"{虚实合}_中枢序列")[序号]
|
||
待发送消息.update({"index": 序号, "data": str(zs)})
|
||
|
||
for i in range(观察员.线段分析层次):
|
||
if 观察员.线段序列组[i] and 观察员.线段序列组[i][0].标识 == 数据类型:
|
||
段 = 观察员.线段序列组[i][段序号]
|
||
zs = getattr(段, f"{虚实合}_中枢序列")[序号]
|
||
待发送消息.update({"index": 序号, "data": str(zs)})
|
||
|
||
for i in range(观察员.扩展线段分析层次):
|
||
if 观察员.扩展线段序列组[i] and 观察员.扩展线段序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.扩展线段序列组[i][序号])})
|
||
段 = 观察员.扩展线段序列组[i][序号]
|
||
zs = getattr(段, f"{虚实合}_中枢序列")[序号]
|
||
待发送消息.update({"index": 序号, "data": str(zs)})
|
||
|
||
for i in range(观察员.混合扩展线段分析层次):
|
||
if 观察员.混合扩展线段序列组[i] and 观察员.混合扩展线段序列组[i][0].标识 == 数据类型:
|
||
待发送消息.update({"index": 序号, "data": str(观察员.混合扩展线段序列组[i][序号])})
|
||
段 = 观察员.混合扩展线段序列组[i][序号]
|
||
zs = getattr(段, f"{虚实合}_中枢序列")[序号]
|
||
待发送消息.update({"index": 序号, "data": str(zs)})
|
||
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "query_result", "success": True, "data_type": 数据类型, "data": 待发送消息})
|
||
|
||
except IndexError:
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "query_result", "success": False, "message": f"索引 {序号} 超出范围"})
|
||
except Exception as e:
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "query_result", "success": False, "message": str(e)})
|
||
else:
|
||
print(f"[query_by_index] 用户 {用户标识} 没有分析器！")
|
||
|
||
elif 消息类型 == "save_path":
|
||
print(f"[保存路径] 用户 {用户标识}: {消息字典}")
|
||
await 全局连接管理器.发送信息(
|
||
用户标识,
|
||
{
|
||
"type": "path_saved",
|
||
"message": "路径已保存",
|
||
"index": 消息字典.get("index"),
|
||
"timestamp": datetime.now().isoformat(),
|
||
},
|
||
)
|
||
|
||
elif 消息类型 == "sync_shape_overrides":
|
||
shapes_data = 消息字典["data"]
|
||
观察员: 观察者 = 全局连接管理器.获取图表观察员(用户标识)
|
||
if 观察员:
|
||
观察员.将图表数据固化到本地(shapes_data)
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "sync_response", "status": "received", "count": len(shapes_data)})
|
||
else:
|
||
print(f"[sync_shape_overrides] 用户 {用户标识} 没有分析器！")
|
||
|
||
elif 消息类型 == "ping":
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "pong", "timestamp": datetime.now().isoformat()})
|
||
|
||
else:
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "error", "message": f"未知的图表消息类型: {消息类型}", "timestamp": datetime.now().isoformat()})
|
||
|
||
|
||
async def 处理代码消息(用户标识: str, 消息字典: Dict):
|
||
"""处理Python执行消息"""
|
||
command = 消息字典.get("command", "")
|
||
|
||
if command == "execute":
|
||
code = 消息字典.get("code", "").strip()
|
||
|
||
if not code:
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "execution_result", "success": False, "message": "❌ 代码不能为空", "module": "python"})
|
||
return
|
||
|
||
当前执行环境 = 全局连接管理器.获取执行环境(用户标识)
|
||
result = 当前执行环境.执行(code)
|
||
|
||
response = {
|
||
"type": "execution_result",
|
||
"success": result["success"],
|
||
"timestamp": datetime.now().isoformat(),
|
||
"execution_time": result.get("execution_time"),
|
||
"module": "python",
|
||
}
|
||
|
||
if result["success"]:
|
||
response.update({"message": "✅ 执行成功", "output": result.get("output", ""), "has_output": bool(result.get("output"))})
|
||
else:
|
||
response.update(
|
||
{
|
||
"message": f"❌ 执行失败: {result.get('error', {}).get('message', '未知错误')}",
|
||
"error": result.get("error"),
|
||
"output": result.get("output", ""),
|
||
}
|
||
)
|
||
|
||
await 全局连接管理器.发送信息(用户标识, response)
|
||
|
||
elif command == "reset":
|
||
当前执行环境 = 全局连接管理器.获取执行环境(用户标识)
|
||
当前执行环境.重置()
|
||
|
||
await 全局连接管理器.发送信息(
|
||
用户标识,
|
||
{
|
||
"type": "environment_reset",
|
||
"message": "🔄 Python执行环境已重置",
|
||
"timestamp": datetime.now().isoformat(),
|
||
"module": "python",
|
||
},
|
||
)
|
||
|
||
elif command == "help":
|
||
当前执行环境 = 全局连接管理器.获取执行环境(用户标识)
|
||
help_text = 当前执行环境.获取帮助()
|
||
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "help_response", "help": help_text, "timestamp": datetime.now().isoformat(), "module": "python"})
|
||
|
||
elif command == "ping":
|
||
await 全局连接管理器.发送信息(用户标识, {"type": "pong", "timestamp": datetime.now().isoformat(), "module": "python"})
|
||
|
||
else:
|
||
await 全局连接管理器.发送信息(
|
||
用户标识,
|
||
{
|
||
"type": "error",
|
||
"message": f"❌ 未知命令: {command}",
|
||
"timestamp": datetime.now().isoformat(),
|
||
"module": "python",
|
||
},
|
||
)
|
||
|
||
|
||
# ============ HTTP端点 ============
|
||
@app.get("/")
|
||
async def 主页(
|
||
request: Request,
|
||
nol: str = "network",
|
||
exchange: str = "bitstamp",
|
||
symbol: str = "btcusd",
|
||
step: int = 300,
|
||
limit: int = 500,
|
||
generator: str = "True",
|
||
):
|
||
"""主页面"""
|
||
观察者.当前事件循环 = asyncio.get_event_loop()
|
||
resolutions = {
|
||
60: "1",
|
||
180: "3",
|
||
300: "5",
|
||
900: "15",
|
||
1800: "30",
|
||
2400: "40",
|
||
3600: "1H",
|
||
7200: "2H",
|
||
14400: "4H",
|
||
21600: "6H",
|
||
43200: "12H",
|
||
86400: "1D",
|
||
259200: "3D",
|
||
604800: "1W",
|
||
}
|
||
|
||
if step not in resolutions:
|
||
return {"error": "不支持的时间周期", "支持的周期": list(resolutions.keys())}
|
||
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"index.html",
|
||
context={
|
||
"request": request,
|
||
"exchange": exchange,
|
||
"symbol": symbol,
|
||
"interval": resolutions.get(step),
|
||
"limit": str(limit),
|
||
"step": str(step),
|
||
"generator": generator,
|
||
},
|
||
)
|
||
|
||
|
||
chanlun.set_rs_log_level("error")
|
||
chanlun.set_log_level("error")
|
||
set_log_level("error")
|
||
|
||
|
||
def _检测nb周期(nb_path: str) -> Optional[int]:
|
||
"""从 .nb 文件前两根K线的时间差检测周期（秒）"""
|
||
import struct
|
||
|
||
try:
|
||
with open(nb_path, "rb") as f:
|
||
buf = f.read(96)
|
||
if len(buf) < 96:
|
||
return None
|
||
t0 = struct.unpack(">d", buf[0:8])[0]
|
||
t1 = struct.unpack(">d", buf[48:56])[0]
|
||
return int(t1 - t0)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def 运行信号驱动策略(
|
||
symbol: str = "btcusd",
|
||
bars: int = 500,
|
||
nb_path: Optional[str] = None,
|
||
max_batches: int = 5,
|
||
warmup: int = 50,
|
||
base_period: int = 86400,
|
||
higher_periods: tuple = (604800,),
|
||
):
|
||
"""信号驱动策略快速启动 — Backtrader 回测
|
||
|
||
使用方式::
|
||
|
||
# CLI: python main.py --mode strategy --symbol btcusd --bars 1000
|
||
# 代码: 运行信号驱动策略(symbol="btcusd", bars=500, nb_path="templates/last.nb")
|
||
|
||
开平仓信号使用 MA 均线多头策略（看多向上开多，看空平多）。
|
||
|
||
:param symbol: 交易品种
|
||
:param bars: 回测 K 线数
|
||
:param nb_path: .nb 数据文件路径，为 None 时使用随机数据
|
||
:param max_batches: 最大同时持仓批次数
|
||
:param warmup: 数据预热周期
|
||
:param base_period: 基础周期（秒），默认 86400（日线）
|
||
:param higher_periods: 高级周期组（秒），默认 (604800,)（周线）
|
||
"""
|
||
start = datetime.now()
|
||
|
||
# ---- 数据源（先检测周期，再构建策略） ----
|
||
if nb_path and os.path.exists(nb_path):
|
||
nb_period = _检测nb周期(nb_path)
|
||
if nb_period:
|
||
base_period = nb_period
|
||
higher_periods = tuple(p for p in higher_periods if p > nb_period)
|
||
if not higher_periods:
|
||
higher_periods = (nb_period * 5,)
|
||
print(f"数据源: {nb_path} (周期={base_period}s, 高级周期={higher_periods})")
|
||
data = NB数据源(nb_path, 最大条数=bars)
|
||
else:
|
||
print(f"数据源: 随机生成 ({bars} 根)")
|
||
data = 随机数据(start_date=datetime(2020, 1, 1), 最大条数=bars)
|
||
|
||
# ---- Backtrader 回测 ----
|
||
cerebro = bt.Cerebro(stdstats=False)
|
||
cerebro.adddata(data)
|
||
|
||
cerebro.addstrategy(
|
||
信号驱动策略,
|
||
符号=symbol,
|
||
基础周期=base_period,
|
||
高级周期组=higher_periods,
|
||
持仓=[
|
||
cet.Position(
|
||
symbol=symbol,
|
||
name="中枢",
|
||
opens=[
|
||
cet.Event(
|
||
cet.Operate.LO,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三买")]),
|
||
],
|
||
)
|
||
],
|
||
exits=[
|
||
cet.Event(
|
||
cet.Operate.LE,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三卖")]),
|
||
],
|
||
)
|
||
],
|
||
),
|
||
cet.Position(
|
||
symbol=symbol,
|
||
name="中枢",
|
||
exits=[
|
||
cet.Event(
|
||
cet.Operate.LO,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三买")]),
|
||
],
|
||
)
|
||
],
|
||
opens=[
|
||
cet.Event(
|
||
cet.Operate.LE,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三卖")]),
|
||
],
|
||
)
|
||
],
|
||
),
|
||
],
|
||
投喂预热数=2,
|
||
指标计算=True,
|
||
)
|
||
|
||
cerebro.broker.setcash(100000.0)
|
||
print(f"初始资金: {cerebro.broker.getvalue():,.0f}")
|
||
|
||
results = cerebro.run()
|
||
strat = results[0]
|
||
elapsed = datetime.now() - start
|
||
|
||
# ---- 结果输出 ----
|
||
print(f"\n{'=' * 60}")
|
||
print(f" 信号驱动策略 回测结果")
|
||
print(f"{'=' * 60}")
|
||
print(f" 品种: {symbol} 周期: {base_period}s K线数: {strat.投喂计数}")
|
||
print(f" 耗时: {elapsed}")
|
||
最终资金 = cerebro.broker.getvalue()
|
||
收益 = (最终资金 - 100000.0) / 100000.0 * 100
|
||
print(f"\n 最终资金: {最终资金:,.0f} | 收益率: {收益:+.2f}%")
|
||
|
||
|
||
def 运行同步信号驱动策略(
|
||
symbol: str = "btcusd",
|
||
bars: int = 500,
|
||
nb_path: Optional[str] = None,
|
||
max_batches: int = 5,
|
||
warmup: int = 50,
|
||
base_period: int = 86400,
|
||
higher_periods: tuple = (604800,),
|
||
):
|
||
"""同步回测 — 实盘运算模式。
|
||
|
||
数据逐根到达，``信号驱动策略`` 在每根 bar 上只能看到已投喂的数据。
|
||
与 :func:`运行信号驱动策略` 使用**同一个** :class:`信号驱动策略`，
|
||
区别仅在于数据投喂方式：
|
||
|
||
- **strategy 模式**：Backtrader 预加载全部数据，一次性回测
|
||
- **sync 模式**：观察者在后台线程逐根喂 K 线 → 队列 → Backtrader 逐根消费
|
||
|
||
数据流::
|
||
|
||
后台线程: 数据源 → 观察员.增加原始K线(k) → 数据队列.put(OHLCV)
|
||
主线程: 自定义实时数据源._load() → 队列.get() → bar
|
||
→ 信号驱动策略.next()
|
||
→ 立体分析器.投喂K线(k)
|
||
→ 信号交易员.更新()
|
||
→ 仓位变化 → 订单执行
|
||
|
||
使用方式::
|
||
|
||
python main.py --mode sync --nb templates/last.nb
|
||
"""
|
||
start = datetime.now()
|
||
|
||
# ---- 先检测周期再构建策略 ----
|
||
if nb_path and os.path.exists(nb_path):
|
||
nb_period = _检测nb周期(nb_path)
|
||
if nb_period:
|
||
base_period = nb_period
|
||
higher_periods = tuple(p for p in higher_periods if p > nb_period)
|
||
if not higher_periods:
|
||
higher_periods = (nb_period * 5,)
|
||
|
||
# ---- 观察者 + 数据队列 (仿 测试_邮局数据_同步回测) ----
|
||
配置 = 缠论配置()
|
||
配置.计算指标 = True
|
||
|
||
数据队列 = queue.Queue()
|
||
观察员 = 观察者(symbol, base_period, None, 配置, 数据队列)
|
||
|
||
if nb_path and os.path.exists(nb_path):
|
||
print(f"数据源: {nb_path}")
|
||
|
||
def _后台喂K线():
|
||
观察员.加载本地数据(nb_path)
|
||
else:
|
||
print(f"数据源: 随机生成 ({bars} 根)")
|
||
|
||
def _后台喂K线():
|
||
dt = datetime(2020, 1, 1)
|
||
price = 100.0
|
||
import random as _r
|
||
|
||
for i in range(bars):
|
||
dt += timedelta(days=1)
|
||
price += _r.gauss(0, 2)
|
||
k = K线.创建普K(symbol, dt, price - _r.random(), price + abs(_r.gauss(0, 1)), price - abs(_r.gauss(0, 1)), price, abs(_r.gauss(100, 50)), 0, base_period)
|
||
观察员.增加原始K线(k)
|
||
|
||
数据源 = 自定义实时数据源(数据队列, 观察员, _后台喂K线)
|
||
|
||
# ---- Backtrader (与 strategy 模式使用同一个 信号驱动策略) ----
|
||
cerebro = bt.Cerebro(stdstats=False)
|
||
cerebro.adddata(数据源)
|
||
|
||
cerebro.addstrategy(
|
||
信号驱动策略,
|
||
符号=symbol,
|
||
基础周期=base_period,
|
||
持仓=[
|
||
cet.Position(
|
||
symbol=symbol,
|
||
name="中枢",
|
||
opens=[
|
||
cet.Event(
|
||
cet.Operate.LO,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三买")]),
|
||
],
|
||
)
|
||
],
|
||
exits=[
|
||
cet.Event(
|
||
cet.Operate.LE,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三卖")]),
|
||
],
|
||
)
|
||
],
|
||
),
|
||
cet.Position(
|
||
symbol=symbol,
|
||
name="中枢",
|
||
exits=[
|
||
cet.Event(
|
||
cet.Operate.LO,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三买")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三买")]),
|
||
],
|
||
)
|
||
],
|
||
opens=[
|
||
cet.Event(
|
||
cet.Operate.LE,
|
||
factors=[
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="中枢段DEA穿越2V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="DEA穿越0轴V230602", v2="三卖")]),
|
||
cet.Factor(signals_all=[cet.Signal(k1=str(base_period), k2="D1MO3", k3="首次穿越0轴V230602", v2="三卖")]),
|
||
],
|
||
)
|
||
],
|
||
),
|
||
],
|
||
投喂预热数=2,
|
||
指标计算=True,
|
||
)
|
||
|
||
cerebro.broker.setcash(100000.0)
|
||
print(f"初始资金: {cerebro.broker.getvalue():,.0f}")
|
||
|
||
results = cerebro.run(live=True)
|
||
strat = results[0]
|
||
elapsed = datetime.now() - start
|
||
|
||
print(f"\n{'=' * 60}")
|
||
print(f" 同步回测结果")
|
||
print(f"{'=' * 60}")
|
||
print(f" 品种: {symbol} 周期: {base_period}s 耗时: {elapsed}")
|
||
print(f" 观察者: {len(观察员.笔序列)} 笔, {len(观察员.线段序列)} 线段, {len(观察员.中枢序列)} 中枢")
|
||
print(f" 策略投喂: {strat.投喂计数} 根K线")
|
||
|
||
最终资金 = cerebro.broker.getvalue()
|
||
收益 = (最终资金 - 100000.0) / 100000.0 * 100
|
||
print(f"\n 最终资金: {最终资金:,.0f} | 收益率: {收益:+.2f}%")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="缠论多线程随机回测 / 信号交易")
|
||
parser.add_argument("--threads", type=int, default=50, help="线程数 (默认 50)")
|
||
parser.add_argument("--limit", type=int, default=10000, help="每线程 K 线数 (默认 10000)")
|
||
parser.add_argument("--log", choices=["off", "simple", "tracing"], default="off", help="日志模式 (默认 off)")
|
||
parser.add_argument("--cache", choices=["thread_local", "global"], default="thread_local", help="缓存模式 (默认 thread_local)")
|
||
parser.add_argument("--mode", choices=["backtest", "trade", "strategy", "sync"], default="backtest", help="运行模式: backtest=随机回测, trade=信号交易, strategy=信号驱动策略, sync=同步信号驱动策略")
|
||
parser.add_argument("--symbol", default="btcusd", help="交易品种 (默认 btcusd)")
|
||
parser.add_argument("--nb", default=None, help=".nb 数据文件路径 (strategy/sync 模式可选)")
|
||
parser.add_argument("--bars", type=int, default=500, help="strategy/sync 模式回测K线数 (默认 500)")
|
||
parser.add_argument("--period", type=int, default=None, help="基础周期秒数 (默认从.nb自动检测, 随机数据默认86400)")
|
||
args = parser.parse_args()
|
||
|
||
# 初始化日志和缓存
|
||
chanlun.set_log_mode(args.log)
|
||
if args.log != "off":
|
||
chanlun.set_log_level("debug")
|
||
chanlun.set_cache_mode(args.cache)
|
||
|
||
if args.mode == "trade":
|
||
# === 信号交易模式 ===
|
||
周期组 = [300, 900, 3600]
|
||
分析器 = 立体分析器(args.symbol, 周期组, 缠论配置())
|
||
|
||
计算器 = cet.SignalOrchestrator(分析器, 信号模块="chanlun.signals")
|
||
计算器.从信号列表提取配置(
|
||
[
|
||
f"300_D1MO3_中枢第三买卖点V230602_中枢段DEA穿越2_三买_偏移1_0",
|
||
f"300_D1MO3_中枢第三买卖点V230602_中枢段DEA穿越2_三卖_偏移1_0",
|
||
]
|
||
)
|
||
|
||
start = datetime.now()
|
||
end_ts = int(datetime.now().timestamp())
|
||
left = end_ts - 周期组[0] * args.limit
|
||
_next = left
|
||
投喂次数 = 0
|
||
while 投喂次数 < args.limit:
|
||
data = Bitstamp.ohlc(args.symbol, 周期组[0], _next, _next := _next + 周期组[0] * 1000)
|
||
if not data.get("data"):
|
||
break
|
||
for bar in data["data"]["ohlc"]:
|
||
k = K线.创建普K(args.symbol, 转化为时间戳(int(bar["timestamp"])), float(bar["open"]), float(bar["high"]), float(bar["low"]), float(bar["close"]), float(bar["volume"]), 0, 周期组[0])
|
||
分析器.投喂K线(k)
|
||
投喂次数 += 1
|
||
if 投喂次数 < 2:
|
||
continue
|
||
计算器.更新()
|
||
if 投喂次数 >= args.limit:
|
||
break
|
||
print(计算器.信号字典)
|
||
_next = int(data["data"]["ohlc"][-1]["timestamp"])
|
||
print(f"交易完成 | 耗时 {datetime.now() - start}")
|
||
|
||
elif args.mode == "strategy":
|
||
# === 信号驱动策略模式 — Backtrader 回测 ===
|
||
kwargs = dict(symbol=args.symbol, bars=args.bars, nb_path=args.nb)
|
||
if args.period is not None:
|
||
kwargs["base_period"] = args.period
|
||
运行信号驱动策略(**kwargs)
|
||
|
||
elif args.mode == "sync":
|
||
# === 同步信号驱动策略模式 — 观察者驱动的实时信号交易 ===
|
||
kwargs = dict(symbol=args.symbol, bars=args.bars, nb_path=args.nb)
|
||
if args.period is not None:
|
||
kwargs["base_period"] = args.period
|
||
运行同步信号驱动策略(**kwargs)
|
||
|
||
else:
|
||
# === 随机回测模式 ===
|
||
def 运行单个回测(线程编号: int):
|
||
try:
|
||
本地随机 = random.Random(os.urandom(64))
|
||
配置 = 随机配置(本地随机)
|
||
print(f"[线程{线程编号:02d}] 开始 | 差异: {缠论配置().对比(配置)}")
|
||
测试函数 = 测试_随机生成(symbol="btcusd", limit=args.limit, freq=时间周期.分(5), ws=None, 配置=配置)
|
||
结果 = 测试函数()
|
||
print(f"[线程{线程编号:02d}] 完成 | 笔: {len(结果.笔序列)} 段: {len(结果.线段序列)} 中枢: {len(结果.中枢序列)}")
|
||
except Exception as e:
|
||
print(f"[线程{线程编号:02d}] 异常: {e}")
|
||
traceback.print_exc()
|
||
|
||
start = datetime.now()
|
||
print(f"启动 {args.threads} 线程 × {args.limit} K线 | 缓存: {chanlun.get_cache_mode()} | 日志: {chanlun.get_log_mode()}")
|
||
|
||
线程列表 = []
|
||
for i in range(1, args.threads + 1):
|
||
线程 = threading.Thread(target=运行单个回测, args=(i,), name=f"回测-{i}")
|
||
线程列表.append(线程)
|
||
线程.start()
|
||
|
||
for 线程 in 线程列表:
|
||
线程.join()
|
||
|
||
elapsed = datetime.now() - start
|
||
total = args.threads * (args.limit + 1)
|
||
print(f"\n全部 {args.threads} 线程已完成 | 总 {total} K线 | 耗时 {elapsed} | {total / elapsed.total_seconds():.0f} K/s")
|