gavindiaz/chanlun.rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复 异常BUG

Browse Source
This commit is contained in:
YuWuKunCheng
2026-06-08 23:29:00 +08:00
parent 8a2afb9ed0
commit 34c42ecd68
14 changed files with 1113 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chan.py
+125 -34
View File
@@ -34,6 +34,7 @@ import os
import struct
import sys
import tempfile
import datetime as datetime_module
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
@@ -48,12 +49,10 @@ from typing import (
Dict,
Any,
Union,
Sequence,
Callable,
Set,
)
from collections.abc import Sequence
from termcolor import colored
from loguru import logger
@@ -96,6 +95,8 @@ __all__ = [
"线段特征相等",
"中枢相等",
"虚线相等",
"观察者相等",
"立体分析器相等",
]
# 日志级别映射: 名称 → loguru 级别名
@@ -446,6 +447,84 @@ def 虚线相等(A, B, 浮点容差: float = 1e-9) -> tuple[bool, str]:
return True, f"{标签}：全字段、所有嵌套子结构校验全部一致"
def 观察者相等(A, B, 浮点容差: float = 1e-9) -> tuple[bool, str]:
"""观察者：全量序列逐项比对（笔/线段/中枢等），双端一致性验证专用。
:param A: 观察者 A (Rust binding)
:param B: 观察者 B (chan.py)
:param 浮点容差: 浮点容差
:return: (一致, 详情)
"""
标签 = f"观察者校验[A={A.标识},B={B.标识}]"
# 基础序列长度
if len(A.缠论K线序列) != len(B.缠论K线序列):
return False, f"{标签}: 缠K序列长度不一致 A={len(A.缠论K线序列)},B={len(B.缠论K线序列)}"
if len(A.分型序列) != len(B.分型序列):
return False, f"{标签}: 分型序列长度不一致 A={len(A.分型序列)},B={len(B.分型序列)}"
if len(A.笔序列) != len(B.笔序列):
return False, f"{标签}: 笔序列长度不一致 A={len(A.笔序列)},B={len(B.笔序列)}"
# 笔序列逐项虚线相等
for i, (a笔, b笔) in enumerate(zip(A.笔序列, B.笔序列)):
eq, msg = 虚线相等(a笔, b笔, 浮点容差)
if not eq:
return False, f"{标签}: 笔#{i}不一致 >> {msg}"
# 笔中枢逐项
if len(A.笔_中枢序列) != len(B.笔_中枢序列):
return False, f"{标签}: 笔中枢序列长度不一致 A={len(A.笔_中枢序列)},B={len(B.笔_中枢序列)}"
for i, (a中, b中) in enumerate(zip(A.笔_中枢序列, B.笔_中枢序列)):
eq, msg = 中枢相等(a中, b中, 浮点容差)
if not eq:
return False, f"{标签}: 笔中枢#{i}不一致 >> {msg}"
# 线段序列组逐层比对
for level in range(A.线段分析层次):
a_segs = A.线段序列组[level]
b_segs = B.线段序列组[level]
if len(a_segs) != len(b_segs):
return False, f"{标签}: 线段序列组[{level}]长度不一致 A={len(a_segs)},B={len(b_segs)}"
for i, (a_seg, b_seg) in enumerate(zip(a_segs, b_segs)):
eq, msg = 虚线相等(a_seg, b_seg, 浮点容差)
if not eq:
return False, f"{标签}: 线段序列组[{level}]#{i}不一致 >> {msg}"
a_hubs = A.中枢序列组[level]
b_hubs = B.中枢序列组[level]
if len(a_hubs) != len(b_hubs):
return False, f"{标签}: 中枢序列组[{level}]长度不一致 A={len(a_hubs)},B={len(b_hubs)}"
for i, (a_hub, b_hub) in enumerate(zip(a_hubs, b_hubs)):
eq, msg = 中枢相等(a_hub, b_hub, 浮点容差)
if not eq:
return False, f"{标签}: 中枢序列组[{level}]#{i}不一致 >> {msg}"
return True, f"{标签}：全量序列、所有嵌套子结构校验全部一致"
def 立体分析器相等(A, B, 浮点容差: float = 1e-9) -> tuple[bool, str]:
"""立体分析器：各周期观察者全量比对，双端一致性验证专用。
:param A: 立体分析器 A (Rust binding)
:param B: 立体分析器 B (chan.py)
:param 浮点容差: 浮点容差
:return: (一致, 详情)
"""
标签 = f"立体分析器校验[A={A.周期组},B={B.周期组}]"
if A.周期组 != B.周期组:
return False, f"{标签}: 周期组不一致 A={A.周期组},B={B.周期组}"
for 周期 in A.周期组:
a_obs = A.单体分析器[周期] if hasattr(A, "单体分析器") else A._单体分析器[周期]
b_obs = B.单体分析器[周期] if hasattr(B, "单体分析器") else B._单体分析器[周期]
eq, msg = 观察者相等(a_obs, b_obs, 浮点容差)
if not eq:
return False, f"{标签}: 周期{周期} >> {msg}"
return True, f"{标签}：所有周期观察者全量校验全部一致"
class 买卖点类型(str, Enum):
"""买卖点类型 — 缠论的三类买卖点及扩展类型。
@@ -2847,7 +2926,7 @@ class K线:
标识=标识,
序号=序号,
周期=周期,
时间戳=时间戳 if isinstance(时间戳, datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
时间戳=时间戳 if isinstance(时间戳, datetime_module.datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
开盘价=开盘价,
最高价=最高价,
最低价=最低价,
@@ -4527,12 +4606,12 @@ class 线段特征:
def __str__(self):
if not len(self):
return colored(f"{self.标识}<{self.线段方向}, 空>", "green")
return f"{self.标识}<{self.线段方向}, 空>"
return f"{self.标识}<{self.线段方向}, {self.文}, {self.武}, {len(self)}>"
def __repr__(self):
if not len(self):
return colored(f"{self.标识}<{self.线段方向}, 空>", "green")
return f"{self.标识}<{self.线段方向}, 空>"
return f"{self.标识}<{self.线段方向}, {self.文}, {self.武}, {len(self)}>"
@property
@@ -5131,7 +5210,7 @@ class 线段:
raise RuntimeError(f"线段._向序列中添加[{行号}], 之前线段[-1] not in 待添加虚线!", 之前线段)
待添加线段.序号 = 之前线段.序号 + 1
待添加线段.前一缺口 = 线段.获取缺口(之前线段)
待添加线段.前一缺口 = 线段.获取缺口(之前线段) if not 之前线段.短路修正 else None
待添加线段.前一结束位置 = 之前线段.基础序列[-1]
if 线段.四象(之前线段) in ("老阴", "老阳"):
@@ -5160,7 +5239,7 @@ class 线段:
if 右 is not None:
结构 = 分型结构.分析(左, 中, 右, True, True)
if 结构 in (分型结构.顶, 分型结构.底) and not 相对方向.分析(左.高, 左.低, 中.高, 中.低).是否缺口():
logger.warning(f"{colored(f'[警告<{行号}>]:', 'yellow')} {colored('线段._从序列中删除 发现分型完毕, 且特征序列无缺口', 'red')} {待弹出线段}")
logger.warning(f"警告<{行号}>] 线段._从序列中删除 发现分型完毕, 且特征序列无缺口 {待弹出线段}")
线段序列.pop()
待弹出线段.前一结束位置 = None
@@ -5255,7 +5334,7 @@ class 线段:
return False
# 执行修正
logger.warning(f"{colored(f'[警告<{sys._getframe().f_lineno}, {层级}>]:', 'yellow')} {colored('线段.修复贯穿伤', 'red')} {贯穿伤} {基础序列}") # 异常弹出
logger.warning(f"[警告<{sys._getframe().f_lineno}, {层级}>]: {当前线段.标识}.修复贯穿伤, 序号:{当前线段.序号} {贯穿伤} {基础序列}") # 异常弹出
基础序列 = 当前线段.基础序列[:]
线段._弹出线段(线段序列, 当前线段, 配置, f"{sys._getframe().f_lineno}, {层级}")
@@ -5609,7 +5688,7 @@ class 线段:
线段._弹出扩展线段(线段序列, 当前线段, sys._getframe().f_lineno)
return 线段递归扩展分析(虚线序列, 线段序列, 配置)
线段._武终(当前线段, sys._getframe().f_lineno)
线段._武终(当前线段, sys._getframe().f_lineno) # TODO 添加错误处理机制
if 当前线段.基础序列[-1].序号 + 3 > 虚线序列[-1].序号:
return None
@@ -6359,33 +6438,33 @@ class 观察者:
if 当前分型 is None:
return
笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
self.配置.分析笔 and 笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
if not self.分型序列:
return
中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
self.配置.分析笔中枢 and 中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
if not self.笔序列:
return
for i in range(self.线段分析层次):
if i == 0:
线段.分析(self.笔序列, self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.笔序列, self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
continue
线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
for i in range(self.扩展线段分析层次):
if i == 0:
线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
continue
线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 测试_保存数据(self, root: str = None) -> str:
"""拆分各序列数据，单独存文件，文件名为对应变量名
@@ -6467,29 +6546,31 @@ class 观察者:
for i in range(1, len(self.缠论K线序列) - 1):
当前分型 = 分型(self.缠论K线序列[i - 1], self.缠论K线序列[i], self.缠论K线序列[i + 1])
笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
self.配置.分析笔 and 笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
if not self.笔序列:
return
self.配置.分析笔中枢 and 中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
for i in range(self.线段分析层次):
if i == 0:
线段.分析(self.笔序列, self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.笔序列, self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
continue
线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
for i in range(self.扩展线段分析层次):
if i == 0:
线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
continue
线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 加载本地数据(self, 文件路径: str):
"""重置基础序列后加载数据文件
@@ -6511,6 +6592,14 @@ class 观察者:
:param 观察员: 可选，已有观察者实例；不传则自动创建
:return: 观察者实例
"""
if "_err-" in str(文件路径) and os.path.exists(str(文件路径).replace(".nb", ".json")):
异常配置 = 缠论配置.加载配置(str(文件路径).replace(".nb", ".json"))
差异 = 缠论配置().对比(异常配置)
传入差异 = 缠论配置().对比(配置)
传入差异.update(差异)
配置 = 缠论配置(**传入差异)
print("加载异常配置+传入差异", 传入差异)
name = Path(文件路径).name.split(".")[0]
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
if 观察员 is None:
@@ -6842,6 +6931,8 @@ def 测试_指标挂载(配置: 缠论配置):
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
周期 = int(周期)
观察员 = 观察者(符号, 周期, 配置)
观察员.线段分析层次 = 0
观察员.重置基础序列()
def 魔法():
启动时间 = datetime.now()
chanlun-py/Cargo.toml
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "chanlun-py"
version = "26.6.44"
version = "26.6.45"
edition = "2024"
description = "缠论技术分析库 — Rust 高性能 Python 绑定"
authors = ["YuYuKunKun"]
@@ -12,7 +12,7 @@ crate-type = ["cdylib"]
name = "chanlun"
[dependencies]
chanlun = "26.6.2" #{ path = "../chanlun" }
chanlun = { path = "../chanlun" }
lru = "0.18"
pyo3 = { version = "0.28", features = ["experimental-inspect"] }
serde_json = "1"
chanlun-py/chanlun.pyi
+6 -3
View File
@@ -19,6 +19,8 @@ def 缺口相等(A: 缺口, B: 缺口, 浮点容差: float = 1e-9) -> Tuple[bool
def 线段特征相等(A: 线段特征, B: 线段特征, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 中枢相等(A: 中枢, B: 中枢, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 虚线相等(A: 虚线, B: 虚线, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 观察者相等(A: 观察者, B: 观察者, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 立体分析器相等(A: 立体分析器, B: 立体分析器, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
# ========== Enum-like types ==========
@@ -806,16 +808,17 @@ class K线合成器:
def 标识(self) -> str: ...
@property
def 周期组(self) -> List[int]: ...
def 投喂K线(self, 普K: K线) -> List[Tuple[int, K线]]: ...
def 投喂(self, 时间戳: int, 开: float, 高: float, 低: float, 收: float, 量: float) -> List[Tuple[int, K线]]: ...
def 投喂K线(self, 普K: K线) -> None: ...
def 投喂(self, 时间戳: int, 开: float, 高: float, 低: float, 收: float, 量: float) -> None: ...
def 获取当前K线(self, 周期: int) -> Optional[K线]: ...
class 立体分析器:
def __init__(self, 符号: str, 周期组: List[int], 配置: Optional[缠论配置] = None, 配置组: Optional[Dict[int, 缠论配置]] = None) -> None: ...
@property
def 周期组(self) -> List[int]: ...
@property
def _单体分析器(self) -> Dict[int, 观察者]: ...
def 投喂K线(self, 普K: K线) -> None: ...
def 获取观察者(self, 周期: int) -> Optional[观察者]: ...
def 测试_保存数据(self, root: Optional[str] = None) -> None: ...
# ========== 缠论配置 (fields via __getattr__/__setattr__) ==========
chanlun-py/chanlun/__init__.pyi
+6 -3
View File
@@ -18,6 +18,8 @@ def 缺口相等(A: 缺口, B: 缺口, 浮点容差: float = 1e-9) -> Tuple[bool
def 线段特征相等(A: 线段特征, B: 线段特征, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 中枢相等(A: 中枢, B: 中枢, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 虚线相等(A: 虚线, B: 虚线, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 观察者相等(A: 观察者, B: 观察者, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
def 立体分析器相等(A: 立体分析器, B: 立体分析器, 浮点容差: float = 1e-9) -> Tuple[bool, str]: ...
# ========== Enum-like types ==========
@@ -831,16 +833,17 @@ class K线合成器:
def 标识(self) -> str: ...
@property
def 周期组(self) -> List[int]: ...
def 投喂K线(self, 普K: K线) -> List[Tuple[int, K线]]: ...
def 投喂(self, 时间戳: int, 开: float, 高: float, 低: float, 收: float, 量: float) -> List[Tuple[int, K线]]: ...
def 投喂K线(self, 普K: K线) -> None: ...
def 投喂(self, 时间戳: int, 开: float, 高: float, 低: float, 收: float, 量: float) -> None: ...
def 获取当前K线(self, 周期: int) -> Optional[K线]: ...
class 立体分析器:
def __init__(self, 符号: str, 周期组: List[int], 配置: Optional[缠论配置] = None, 配置组: Optional[Dict[int, 缠论配置]] = None) -> None: ...
@property
def 周期组(self) -> List[int]: ...
@property
def _单体分析器(self) -> Dict[int, 观察者]: ...
def 投喂K线(self, 普K: K线) -> None: ...
def 获取观察者(self, 周期: int) -> Optional[观察者]: ...
def 测试_保存数据(self, root: Optional[str] = None) -> None: ...
# ========== 缠论配置 (fields via __getattr__/__setattr__) ==========
chanlun-py/chanlun/chan.py
+125 -34
View File
@@ -34,6 +34,7 @@ import os
import struct
import sys
import tempfile
import datetime as datetime_module
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
@@ -48,12 +49,10 @@ from typing import (
Dict,
Any,
Union,
Sequence,
Callable,
Set,
)
from collections.abc import Sequence
from termcolor import colored
from loguru import logger
@@ -96,6 +95,8 @@ __all__ = [
"线段特征相等",
"中枢相等",
"虚线相等",
"观察者相等",
"立体分析器相等",
]
# 日志级别映射: 名称 → loguru 级别名
@@ -446,6 +447,84 @@ def 虚线相等(A, B, 浮点容差: float = 1e-9) -> tuple[bool, str]:
return True, f"{标签}：全字段、所有嵌套子结构校验全部一致"
def 观察者相等(A, B, 浮点容差: float = 1e-9) -> tuple[bool, str]:
"""观察者：全量序列逐项比对（笔/线段/中枢等），双端一致性验证专用。
:param A: 观察者 A (Rust binding)
:param B: 观察者 B (chan.py)
:param 浮点容差: 浮点容差
:return: (一致, 详情)
"""
标签 = f"观察者校验[A={A.标识},B={B.标识}]"
# 基础序列长度
if len(A.缠论K线序列) != len(B.缠论K线序列):
return False, f"{标签}: 缠K序列长度不一致 A={len(A.缠论K线序列)},B={len(B.缠论K线序列)}"
if len(A.分型序列) != len(B.分型序列):
return False, f"{标签}: 分型序列长度不一致 A={len(A.分型序列)},B={len(B.分型序列)}"
if len(A.笔序列) != len(B.笔序列):
return False, f"{标签}: 笔序列长度不一致 A={len(A.笔序列)},B={len(B.笔序列)}"
# 笔序列逐项虚线相等
for i, (a笔, b笔) in enumerate(zip(A.笔序列, B.笔序列)):
eq, msg = 虚线相等(a笔, b笔, 浮点容差)
if not eq:
return False, f"{标签}: 笔#{i}不一致 >> {msg}"
# 笔中枢逐项
if len(A.笔_中枢序列) != len(B.笔_中枢序列):
return False, f"{标签}: 笔中枢序列长度不一致 A={len(A.笔_中枢序列)},B={len(B.笔_中枢序列)}"
for i, (a中, b中) in enumerate(zip(A.笔_中枢序列, B.笔_中枢序列)):
eq, msg = 中枢相等(a中, b中, 浮点容差)
if not eq:
return False, f"{标签}: 笔中枢#{i}不一致 >> {msg}"
# 线段序列组逐层比对
for level in range(A.线段分析层次):
a_segs = A.线段序列组[level]
b_segs = B.线段序列组[level]
if len(a_segs) != len(b_segs):
return False, f"{标签}: 线段序列组[{level}]长度不一致 A={len(a_segs)},B={len(b_segs)}"
for i, (a_seg, b_seg) in enumerate(zip(a_segs, b_segs)):
eq, msg = 虚线相等(a_seg, b_seg, 浮点容差)
if not eq:
return False, f"{标签}: 线段序列组[{level}]#{i}不一致 >> {msg}"
a_hubs = A.中枢序列组[level]
b_hubs = B.中枢序列组[level]
if len(a_hubs) != len(b_hubs):
return False, f"{标签}: 中枢序列组[{level}]长度不一致 A={len(a_hubs)},B={len(b_hubs)}"
for i, (a_hub, b_hub) in enumerate(zip(a_hubs, b_hubs)):
eq, msg = 中枢相等(a_hub, b_hub, 浮点容差)
if not eq:
return False, f"{标签}: 中枢序列组[{level}]#{i}不一致 >> {msg}"
return True, f"{标签}：全量序列、所有嵌套子结构校验全部一致"
def 立体分析器相等(A, B, 浮点容差: float = 1e-9) -> tuple[bool, str]:
"""立体分析器：各周期观察者全量比对，双端一致性验证专用。
:param A: 立体分析器 A (Rust binding)
:param B: 立体分析器 B (chan.py)
:param 浮点容差: 浮点容差
:return: (一致, 详情)
"""
标签 = f"立体分析器校验[A={A.周期组},B={B.周期组}]"
if A.周期组 != B.周期组:
return False, f"{标签}: 周期组不一致 A={A.周期组},B={B.周期组}"
for 周期 in A.周期组:
a_obs = A.单体分析器[周期] if hasattr(A, "单体分析器") else A._单体分析器[周期]
b_obs = B.单体分析器[周期] if hasattr(B, "单体分析器") else B._单体分析器[周期]
eq, msg = 观察者相等(a_obs, b_obs, 浮点容差)
if not eq:
return False, f"{标签}: 周期{周期} >> {msg}"
return True, f"{标签}：所有周期观察者全量校验全部一致"
class 买卖点类型(str, Enum):
"""买卖点类型 — 缠论的三类买卖点及扩展类型。
@@ -2847,7 +2926,7 @@ class K线:
标识=标识,
序号=序号,
周期=周期,
时间戳=时间戳 if isinstance(时间戳, datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
时间戳=时间戳 if isinstance(时间戳, datetime_module.datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
开盘价=开盘价,
最高价=最高价,
最低价=最低价,
@@ -4527,12 +4606,12 @@ class 线段特征:
def __str__(self):
if not len(self):
return colored(f"{self.标识}<{self.线段方向}, 空>", "green")
return f"{self.标识}<{self.线段方向}, 空>"
return f"{self.标识}<{self.线段方向}, {self.文}, {self.武}, {len(self)}>"
def __repr__(self):
if not len(self):
return colored(f"{self.标识}<{self.线段方向}, 空>", "green")
return f"{self.标识}<{self.线段方向}, 空>"
return f"{self.标识}<{self.线段方向}, {self.文}, {self.武}, {len(self)}>"
@property
@@ -5131,7 +5210,7 @@ class 线段:
raise RuntimeError(f"线段._向序列中添加[{行号}], 之前线段[-1] not in 待添加虚线!", 之前线段)
待添加线段.序号 = 之前线段.序号 + 1
待添加线段.前一缺口 = 线段.获取缺口(之前线段)
待添加线段.前一缺口 = 线段.获取缺口(之前线段) if not 之前线段.短路修正 else None
待添加线段.前一结束位置 = 之前线段.基础序列[-1]
if 线段.四象(之前线段) in ("老阴", "老阳"):
@@ -5160,7 +5239,7 @@ class 线段:
if 右 is not None:
结构 = 分型结构.分析(左, 中, 右, True, True)
if 结构 in (分型结构.顶, 分型结构.底) and not 相对方向.分析(左.高, 左.低, 中.高, 中.低).是否缺口():
logger.warning(f"{colored(f'[警告<{行号}>]:', 'yellow')} {colored('线段._从序列中删除 发现分型完毕, 且特征序列无缺口', 'red')} {待弹出线段}")
logger.warning(f"警告<{行号}>] 线段._从序列中删除 发现分型完毕, 且特征序列无缺口 {待弹出线段}")
线段序列.pop()
待弹出线段.前一结束位置 = None
@@ -5255,7 +5334,7 @@ class 线段:
return False
# 执行修正
logger.warning(f"{colored(f'[警告<{sys._getframe().f_lineno}, {层级}>]:', 'yellow')} {colored('线段.修复贯穿伤', 'red')} {贯穿伤} {基础序列}") # 异常弹出
logger.warning(f"[警告<{sys._getframe().f_lineno}, {层级}>]: {当前线段.标识}.修复贯穿伤, 序号:{当前线段.序号} {贯穿伤} {基础序列}") # 异常弹出
基础序列 = 当前线段.基础序列[:]
线段._弹出线段(线段序列, 当前线段, 配置, f"{sys._getframe().f_lineno}, {层级}")
@@ -5609,7 +5688,7 @@ class 线段:
线段._弹出扩展线段(线段序列, 当前线段, sys._getframe().f_lineno)
return 线段递归扩展分析(虚线序列, 线段序列, 配置)
线段._武终(当前线段, sys._getframe().f_lineno)
线段._武终(当前线段, sys._getframe().f_lineno) # TODO 添加错误处理机制
if 当前线段.基础序列[-1].序号 + 3 > 虚线序列[-1].序号:
return None
@@ -6359,33 +6438,33 @@ class 观察者:
if 当前分型 is None:
return
笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
self.配置.分析笔 and 笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
if not self.分型序列:
return
中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
self.配置.分析笔中枢 and 中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
if not self.笔序列:
return
for i in range(self.线段分析层次):
if i == 0:
线段.分析(self.笔序列, self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.笔序列, self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
continue
线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
for i in range(self.扩展线段分析层次):
if i == 0:
线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
continue
线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 测试_保存数据(self, root: str = None) -> str:
"""拆分各序列数据，单独存文件，文件名为对应变量名
@@ -6467,29 +6546,31 @@ class 观察者:
for i in range(1, len(self.缠论K线序列) - 1):
当前分型 = 分型(self.缠论K线序列[i - 1], self.缠论K线序列[i], self.缠论K线序列[i + 1])
笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
self.配置.分析笔 and 笔.分析(当前分型, self.分型序列, self.笔序列, self.缠论K线序列, self.普通K线序列, 0, self.配置)
中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
if not self.笔序列:
return
self.配置.分析笔中枢 and 中枢.分析(self.笔序列, self.笔_中枢序列, True, "", 0)
for i in range(self.线段分析层次):
if i == 0:
线段.分析(self.笔序列, self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.笔序列, self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
continue
线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
self.配置.分析线段 and 线段.分析(self.线段序列组[i - 1], self.线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.线段序列组[i], self.中枢序列组[i], True, "", 0)
for i in range(self.扩展线段分析层次):
if i == 0:
线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.笔序列, self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
continue
线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 加载本地数据(self, 文件路径: str):
"""重置基础序列后加载数据文件
@@ -6511,6 +6592,14 @@ class 观察者:
:param 观察员: 可选，已有观察者实例；不传则自动创建
:return: 观察者实例
"""
if "_err-" in str(文件路径) and os.path.exists(str(文件路径).replace(".nb", ".json")):
异常配置 = 缠论配置.加载配置(str(文件路径).replace(".nb", ".json"))
差异 = 缠论配置().对比(异常配置)
传入差异 = 缠论配置().对比(配置)
传入差异.update(差异)
配置 = 缠论配置(**传入差异)
print("加载异常配置+传入差异", 传入差异)
name = Path(文件路径).name.split(".")[0]
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
if 观察员 is None:
@@ -6842,6 +6931,8 @@ def 测试_指标挂载(配置: 缠论配置):
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
周期 = int(周期)
观察员 = 观察者(符号, 周期, 配置)
观察员.线段分析层次 = 0
观察员.重置基础序列()
def 魔法():
启动时间 = datetime.now()
chanlun-py/pyproject.toml
+2 -1
View File
@@ -4,7 +4,7 @@ build-backend = "maturin"
[project]
name = "chanlun"
version = "2606.44"
version = "2606.45"
description = "缠论技术分析库 — Rust 高性能实现"
readme = { file = "README.md", content-type = "text/markdown" }
license = { file = "LICENSE", content-type = "text/plain" }
@@ -28,6 +28,7 @@ classifiers = [
requires-python = ">=3.9"
dependencies = [
"termcolor>=3.0",
"parse>=1.2",
"loguru>=0.6",
"backtrader==1.9.78.123",
]
chanlun-py/src/business_py.rs
+22 -49
View File
@@ -1293,9 +1293,8 @@ impl 观察者Py {
/// K线合成器(标识, 周期组) — 周期组为升序排列的整数列表（如 [60, 300, 900]）
///
/// 方法:
/// 投喂K线(普K) -> list[(周期, K线)] — 喂入普通K线，返回合成后的大周期K线
/// 投喂(时间戳, 开盘价, 最高价, 最低价, 收盘价, 成交量) -> list[(周期, K线)]
/// — 快捷入口，免去构造K线对象
/// 投喂K线(普K) — 喂入普通K线，完成事件通过回调分发
/// 投喂(时间戳, 开盘价, 最高价, 最低价, 收盘价, 成交量) — 快捷入口
/// 获取当前K线(周期) -> K线|None — 获取指定周期的当前合成结果
#[pyclass(name = "K线合成器", module = "chanlun._chanlun")]
pub struct K线合成器Py {
@@ -1307,51 +1306,18 @@ impl K线合成器Py {
#[new]
fn new(标识: String, 周期组: Vec<i64>) -> Self {
Self {
inner: chanlun::business::synthesizer::K线合成器::new(标识, 周期组),
inner: chanlun::business::synthesizer::K线合成器::new(标识, 周期组, None),
}
}
/// 统一入口 — 投喂最小周期K线，自动合成大周期并分发给各周期观察者
fn 投喂K线(
&mut self,
普K: &Bound<'_, K线Py>,
py: Python<'_>,
) -> PyResult<Vec<(i64, Py<K线Py>)>> {
let results = self.inner.投喂K线((*普K.borrow().inner).clone());
Ok(results
.into_iter()
.map(|(周期, k)| (周期, bar_to_py(py, Arc::new(k))))
.collect())
/// 投喂K线 — 输入最小周期K线，合成为所有目标周期，完成事件通过回调分发
fn 投喂K线(&mut self, 普K: &Bound<'_, K线Py>) {
self.inner.投喂K线((*普K.borrow().inner).clone());
}
/// 投喂原始tick数据
fn 投喂(
&mut self,
时间戳: i64,
开: f64,
高: f64,
低: f64,
收: f64,
量: f64,
py: Python<'_>,
) -> Vec<(i64, Py<K线Py>)> {
let min_cycle = self.inner.周期组.iter().copied().min().unwrap_or(1);
let k = chanlun::kline::bar::K线::创建普K(
&self.inner.标识,
时间戳,
开,
高,
低,
收,
量,
0,
min_cycle,
);
let results = self.inner.投喂K线(k);
results
.into_iter()
.map(|(周期, k2)| (周期, bar_to_py(py, Arc::new(k2))))
.collect()
fn 投喂(&mut self, 时间戳: i64, 开: f64, 高: f64, 低: f64, 收: f64, 量: f64) {
self.inner.投喂(时间戳, 开, 高, 低, 收, 量);
}
/// 获取指定周期当前正在合成的K线
@@ -1430,13 +1396,6 @@ impl 立体分析器Py {
self.inner.投喂K线((*普K.borrow().inner).clone());
}
fn 获取观察者(&self, 周期: i64) -> Option<观察者Py> {
self.inner.获取观察者(周期).map(|rc| 观察者Py {
inner: Some(rc),
配置缓存: std::sync::Mutex::new(None),
})
}
/// 拆分各序列数据，单独存文件，文件名为对应变量名
fn 测试_保存数据(&self, root: Option<&str>) {
self.inner.测试_保存数据(root);
@@ -1446,6 +1405,20 @@ impl 立体分析器Py {
fn 周期组(&self) -> Vec<i64> {
self.inner.周期组.clone()
}
/// _单体分析器 — 对应 Python 立体分析器._单体分析器: dict[周期, 观察者]
#[getter(_单体分析器)]
fn get_单体分析器(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
let dict = pyo3::types::PyDict::new(py);
for (周期, obs_rc) in &self.inner.单体分析器 {
let obs_py = 观察者Py {
inner: Some(obs_rc.clone()),
配置缓存: std::sync::Mutex::new(None),
};
dict.set_item(周期, obs_py)?;
}
Ok(dict.into())
}
}
pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
chanlun-py/src/equality_py.rs
+46
View File
@@ -23,6 +23,9 @@
*/
use std::num::NonZeroUsize;
use crate::business_py::立体分析器Py;
use crate::business_py::观察者Py;
use std::sync::Mutex;
use lru::LruCache;
@@ -768,6 +771,47 @@ fn 虚线相等(
Ok((true, format!("{标签}: 全字段所有嵌套子结构校验一致")))
}
// ========== 观察者相等 ==========
#[pyfunction]
fn 观察者相等(
a: &Bound<'_, 观察者Py>,
b: &Bound<'_, 观察者Py>,
浮点容差: Option<f64>,
) -> PyResult<(bool, String)> {
let 容差 = 浮点容差.unwrap_or(1e-9);
let arc_a = a
.borrow()
.inner
.clone()
.ok_or_else(|| pyo3::exceptions::PyValueError::new_err("观察者A 内部为空"))?;
let arc_b = b
.borrow()
.inner
.clone()
.ok_or_else(|| pyo3::exceptions::PyValueError::new_err("观察者B 内部为空"))?;
let obs_a = arc_a.read().unwrap();
let obs_b = arc_b.read().unwrap();
Ok(obs_a.相等(&obs_b, 容差))
}
// ========== 立体分析器相等 ==========
#[pyfunction]
fn 立体分析器相等(
a: &Bound<'_, 立体分析器Py>,
b: &Bound<'_, 立体分析器Py>,
浮点容差: Option<f64>,
) -> PyResult<(bool, String)> {
let 容差 = 浮点容差.unwrap_or(1e-9);
let result = {
let ref_a = a.borrow();
let ref_b = b.borrow();
ref_a.inner.相等(&ref_b.inner, 容差)
};
Ok(result)
}
pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_function(wrap_pyfunction!(K线相等, m)?)?;
m.add_function(wrap_pyfunction!(缠论K线相等, m)?)?;
@@ -776,5 +820,7 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_function(wrap_pyfunction!(线段特征相等, m)?)?;
m.add_function(wrap_pyfunction!(中枢相等, m)?)?;
m.add_function(wrap_pyfunction!(虚线相等, m)?)?;
m.add_function(wrap_pyfunction!(观察者相等, m)?)?;
m.add_function(wrap_pyfunction!(立体分析器相等, m)?)?;
Ok(())
}
chanlun-py/tests/test_all.py
+283
View File
@@ -2228,5 +2228,288 @@ class Test导出函数双端等效(unittest.TestCase):
# all passed (or vacuously true if no hubs with >=9 segments)
class TestK线合成器(unittest.TestCase):
"""K线合成器 模块测试."""
def test_构造(self):
"""K线合成器 初始状态."""
import chanlun
s = chanlun.K线合成器("btcusd", [60, 300])
self.assertEqual(s.标识, "btcusd")
self.assertEqual(s.周期组, [60, 300])
self.assertIsNone(s.获取当前K线(60))
self.assertIsNone(s.获取当前K线(300))
def test_投喂单周期(self):
"""投喂单周期K线."""
import chanlun
s = chanlun.K线合成器("btcusd", [300])
bar = chanlun.K线.创建普K("btcusd", 300, 100, 110, 90, 105, 1000, 0, 60)
s.投喂K线(bar)
cur = s.获取当前K线(300)
self.assertIsNotNone(cur)
self.assertEqual(cur.周期, 300)
self.assertAlmostEqual(cur.高, 110)
def test_投喂多周期(self):
"""投喂生成多周期K线."""
import chanlun
s = chanlun.K线合成器("btcusd", [60, 300])
bar = chanlun.K线.创建普K("btcusd", 60, 100, 110, 90, 105, 1000, 0, 60)
s.投喂K线(bar)
self.assertIsNotNone(s.获取当前K线(60))
self.assertIsNotNone(s.获取当前K线(300))
def test_便捷投喂(self):
"""便捷投喂方法."""
import chanlun
s = chanlun.K线合成器("btcusd", [300])
s.投喂(1218124800, 100, 110, 90, 105, 1000)
cur = s.获取当前K线(300)
self.assertIsNotNone(cur)
class TestK线合成器双端一致(unittest.TestCase):
"""K线合成器 Rust vs chan.py 运行时行为一致 — 每步对比."""
_TEST_COUNT = 200
@classmethod
def setUpClass(cls):
if not _has_nb():
raise unittest.SkipTest("需要 .nb 数据文件")
cls.bars = read_nb_bars(NB_PATH, cls._TEST_COUNT)
def _make_synthesizers(self):
import chanlun
from chanlun import chan
return chanlun.K线合成器("btcusd", [60, 300]), chan.K线合成器("btcusd", [60, 300])
def test_合成K线逐笔OHLC一致(self):
"""每投喂一根K线后，大周期当前K线OHLC双端一致."""
import chanlun
from chanlun import chan
s_rs, s_py = self._make_synthesizers()
mismatches = []
for i, (ts, o, h, l, c, v) in enumerate(self.bars):
bar_rs = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 60)
bar_py = chan.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 60)
s_rs.投喂K线(bar_rs)
s_py.投喂K线(bar_py)
cur_rs = s_rs.获取当前K线(300)
cur_py = s_py.获取当前K线(300)
if cur_rs is None and cur_py is None:
continue
if (cur_rs is None) != (cur_py is None):
mismatches.append(f"#{i} ts={ts}: R={cur_rs} P={cur_py}")
continue
if abs(cur_rs.高 - cur_py.高) > 1e-6 or abs(cur_rs.低 - cur_py.低) > 1e-6 or abs(cur_rs.开盘价 - cur_py.开盘价) > 1e-6 or abs(cur_rs.收盘价 - cur_py.收盘价) > 1e-6:
mismatches.append(f"#{i} ts={ts}: R(o={cur_rs.开盘价} h={cur_rs.高} l={cur_rs.低} c={cur_rs.收盘价}) P(o={cur_py.开盘价} h={cur_py.高} l={cur_py.低} c={cur_py.收盘价})")
self.assertEqual(len(mismatches), 0, f"合成K线不一致 ({len(mismatches)}处):\n" + "\n".join(mismatches[:10]))
def test_合成K线逐笔时间戳一致(self):
"""每投喂一根K线后，大周期当前K线时间戳双端一致."""
import chanlun
from chanlun import chan
s_rs, s_py = self._make_synthesizers()
mismatches = []
for i, (ts, o, h, l, c, v) in enumerate(self.bars):
bar_rs = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 60)
bar_py = chan.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 60)
s_rs.投喂K线(bar_rs)
s_py.投喂K线(bar_py)
cur_rs = s_rs.获取当前K线(300)
cur_py = s_py.获取当前K线(300)
if cur_rs is None and cur_py is None:
continue
if (cur_rs is None) != (cur_py is None):
mismatches.append(f"#{i} ts={ts}: R={cur_rs} P={cur_py}")
continue
if int(cur_rs.时间戳) != int(cur_py.时间戳):
mismatches.append(f"#{i}: R={int(cur_rs.时间戳)} P={int(cur_py.时间戳)}")
self.assertEqual(len(mismatches), 0, f"时间戳不一致 ({len(mismatches)}处):\n" + "\n".join(mismatches[:10]))
class Test立体分析器(unittest.TestCase):
"""立体分析器 模块测试."""
@classmethod
def setUpClass(cls):
if not _has_nb():
raise unittest.SkipTest("需要 .nb 数据文件")
cls.bars = read_nb_bars(NB_PATH) # [:300]
def test_构造(self):
"""立体分析器 构造."""
import chanlun
cfg = chanlun.缠论配置()
ma = chanlun.立体分析器("btcusd", [60, 300], cfg)
self.assertEqual(ma.周期组, [60, 300])
def test_单体分析器(self):
"""单体分析器字典包含各周期观察者."""
import chanlun
cfg = chanlun.缠论配置()
ma = chanlun.立体分析器("btcusd", [60, 300], cfg)
d = ma._单体分析器
self.assertIn(60, d)
self.assertIn(300, d)
self.assertEqual(d[60].周期, 60)
self.assertEqual(d[300].周期, 300)
def test_投喂K线生成各级别数据(self):
"""投喂K线后各周期有分析数据."""
import chanlun
cfg = chanlun.缠论配置()
ma = chanlun.立体分析器("btcusd", [300, 300 * 5], cfg)
for ts, o, h, l, c, v in self.bars:
bar = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
ma.投喂K线(bar)
obs_300 = ma._单体分析器[300]
self.assertGreater(len(obs_300.缠论K线序列), 0, "300周期无缠K")
self.assertGreater(len(obs_300.普通K线序列), 0, "300周期无普K")
class Test立体分析器双端一致(unittest.TestCase):
"""立体分析器 Rust vs chan.py 运行时行为一致 — 每步对比 + 数据内容对比."""
_TEST_COUNT = 500
@classmethod
def setUpClass(cls):
if not _has_nb():
raise unittest.SkipTest("需要 .nb 数据文件")
cls.bars = read_nb_bars(NB_PATH, cls._TEST_COUNT)
def _make_analyzers(self):
import chanlun
from chanlun import chan
cfg_rs = chanlun.缠论配置()
cfg_py = chan.缠论配置()
return (chanlun.立体分析器("btcusd", [300, 300 * 5], cfg_rs), chan.立体分析器("btcusd", [300, 300 * 5], cfg_py))
def test_立体分析逐笔笔序列增长一致(self):
"""每投喂K线后，各周期笔序列长度双端一致."""
import chanlun
from chanlun import chan
ma_rs, ma_py = self._make_analyzers()
mismatches = []
for i, (ts, o, h, l, c, v) in enumerate(self.bars):
bar_rs = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
bar_py = chan.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
ma_rs.投喂K线(bar_rs)
ma_py.投喂K线(bar_py)
for period in [300, 300 * 5]:
obs_rs = ma_rs._单体分析器[period]
obs_py = ma_py._单体分析器[period]
if len(obs_rs.笔序列) != len(obs_py.笔序列):
mismatches.append(f"#{i} ts={ts} 周期{period}: R笔={len(obs_rs.笔序列)} P笔={len(obs_py.笔序列)}")
if len(obs_rs.分型序列) != len(obs_py.分型序列):
mismatches.append(f"#{i} ts={ts} 周期{period}: R分型={len(obs_rs.分型序列)} P分型={len(obs_py.分型序列)}")
eq, msg = chan.观察者相等(obs_py, obs_rs)
self.assertTrue(eq, msg)
self.assertEqual(len(mismatches), 0, f"立体分析不一致 ({len(mismatches)}处):\n" + "\n".join(mismatches[:10]))
def test_立体分析逐笔缠K序列增长一致(self):
"""每投喂K线后，各周期缠论K线序列长度双端一致."""
import chanlun
from chanlun import chan
ma_rs, ma_py = self._make_analyzers()
mismatches = []
for i, (ts, o, h, l, c, v) in enumerate(self.bars):
bar_rs = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
bar_py = chan.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
ma_rs.投喂K线(bar_rs)
ma_py.投喂K线(bar_py)
for period in [300, 300 * 5]:
obs_rs = ma_rs._单体分析器[period]
obs_py = ma_py._单体分析器[period]
if len(obs_rs.缠论K线序列) != len(obs_py.缠论K线序列):
mismatches.append(f"#{i} ts={ts} 周期{period}: R缠K={len(obs_rs.缠论K线序列)} P缠K={len(obs_py.缠论K线序列)}")
self.assertEqual(len(mismatches), 0, f"缠K序列不一致 ({len(mismatches)}处):\n" + "\n".join(mismatches[:10]))
def test_立体分析逐笔线段序列增长一致(self):
"""每投喂K线后，显示周期线段序列长度双端一致."""
import chanlun
from chanlun import chan
ma_rs, ma_py = self._make_analyzers()
mismatches = []
for i, (ts, o, h, l, c, v) in enumerate(self.bars):
bar_rs = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
bar_py = chan.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
ma_rs.投喂K线(bar_rs)
ma_py.投喂K线(bar_py)
obs_rs = ma_rs._单体分析器[300 * 5]
obs_py = ma_py._单体分析器[300 * 5]
if len(obs_rs.线段序列) != len(obs_py.线段序列):
mismatches.append(f"#{i} ts={ts}: R线段={len(obs_rs.线段序列)} P线段={len(obs_py.线段序列)}")
self.assertEqual(len(mismatches), 0, f"线段序列不一致 ({len(mismatches)}处):\n" + "\n".join(mismatches[:10]))
def test_立体分析器相等(self):
"""立体分析后 chan.立体分析器相等 全量数据对比一致."""
import chanlun
from chanlun import chan
ma_rs, ma_py = self._make_analyzers()
for ts, o, h, l, c, v in self.bars:
bar_rs = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
bar_py = chan.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
ma_rs.投喂K线(bar_rs)
ma_py.投喂K线(bar_py)
eq, msg = chan.立体分析器相等(ma_rs, ma_py)
self.assertTrue(eq, msg)
def test_立体分析观察者相等(self):
"""立体分析后主周期 chan.观察者相等 全量数据对比一致."""
import chanlun
from chanlun import chan
ma_rs, ma_py = self._make_analyzers()
for ts, o, h, l, c, v in self.bars:
bar_rs = chanlun.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
bar_py = chan.K线.创建普K("btcusd", ts, o, h, l, c, v, 0, 300)
ma_rs.投喂K线(bar_rs)
ma_py.投喂K线(bar_py)
eq, msg = chan.观察者相等(ma_rs._单体分析器[300 * 5], ma_py._单体分析器[300 * 5])
self.assertTrue(eq, msg)
if __name__ == "__main__":
unittest.main()
chanlun/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "chanlun"
version = "26.6.2"
version = "26.6.3"
edition = "2024"
license = "MIT"
description = "基于缠论（缠中说禅）理论的量化技术分析核心库，支持流式数据处理和多周期联立分析。"
chanlun/src/algorithm/segment.rs
+13 -11
View File
@@ -800,7 +800,12 @@ impl 线段 {
seg.序号
.store(之前线段.序号.load(Ordering::Relaxed) + 1, Ordering::Relaxed);
*seg.前一缺口.write().unwrap() = Self::获取缺口(之前线段);
*seg.前一缺口.write().unwrap() = if 之前线段.短路修正.load(Ordering::Relaxed)
{
None
} else {
Self::获取缺口(之前线段)
};
*seg.前一结束位置.write().unwrap() = Some(Arc::clone(
之前线段.基础序列.read().unwrap().last().unwrap(),
));
@@ -970,6 +975,7 @@ impl 线段 {
}
let 左 = Arc::clone(&基础序列[基础序列.len() - 3]);
let 中 = Arc::clone(&基础序列[基础序列.len() - 2]);
let 右 = Arc::clone(&基础序列[基础序列.len() - 1]);
// 方向条件
@@ -987,7 +993,8 @@ impl 线段 {
基础序列
);
let 原始基础序列 = 当前线段.基础序列.read().unwrap().clone();
// Reassign to full copy (matching Python pattern)
let 基础序列 = 当前线段.基础序列.read().unwrap().clone();
Self::_弹出线段(
线段序列,
&Arc::clone(线段序列.last().unwrap()),
@@ -1021,13 +1028,13 @@ impl 线段 {
cur.特征序列.write().unwrap()[2] = None;
let 开始笔 = Arc::clone(cur.基础序列.read().unwrap().last().unwrap());
let 开始序号 = 原始基础序列
let 开始序号 = 基础序列
.iter()
.position(|x| Arc::as_ptr(x) == Arc::as_ptr(&开始笔));
开始序号_opt = 开始序号;
if let Some(序号) = 开始序号 {
待添加元素 = 原始基础序列[序号 + 1..].to_vec();
待添加元素 = 基础序列[序号 + 1..].to_vec();
} else {
待添加元素 = Vec::new();
}
@@ -1047,19 +1054,14 @@ impl 线段 {
let 当前线段 = Arc::clone(&线段序列[idx]);
当前线段.短路修正.store(true, Ordering::Relaxed);
if 当前线段.特征序列.read().unwrap()[2].is_some() {
let 段 = 虚线::创建线段(&[
Arc::clone(&基础序列[基础序列.len() - 3]),
Arc::clone(&基础序列[基础序列.len() - 2]),
Arc::clone(&基础序列[基础序列.len() - 1]),
]);
let 段 = 虚线::创建线段(&[Arc::clone(&左), Arc::clone(&中), Arc::clone(&右)]);
let 段_rc = Arc::new(段);
Self::_添加线段(线段序列, 段_rc, 配置, format!("{}, {}", line!(), 层级));
// Set feature sequence [0]
let 新段 = Self::取段(线段序列.last_mut().unwrap());
let 中笔 = Arc::clone(&基础序列[基础序列.len() - 2]);
新段.特征序列.write().unwrap()[0] =
Some(Arc::new(线段特征::新建(vec![中笔], 新段.方向())));
Some(Arc::new(线段特征::新建(vec![中], 新段.方向())));
}
true
chanlun/src/business/multi_frame.rs
+66 -27
View File
@@ -32,18 +32,15 @@ use std::sync::RwLock;
use tracing::{error, info};
/// 立体分析器 — 多周期协调器
///
/// 包含一个K线合成器和每周期一个观察者。
/// 输入最小周期K线，合成大周期后分发到对应观察者。
pub struct 立体分析器 {
pub 周期组: Vec<i64>,
输入周期: i64,
K线合成器: K线合成器,
单体分析器: HashMap<i64, Arc<RwLock<观察者>>>,
pub K线合成器: K线合成器,
pub 单体分析器: HashMap<i64, Arc<RwLock<观察者>>>,
}
impl 立体分析器 {
/// 创建立体分析器，自动创建K线合成器 + 每周期一个观察者
/// 创建立体分析器 — 对应 Python 立体分析器.__init__
pub fn new(
符号: String,
周期组: Vec<i64>,
@@ -58,9 +55,7 @@ impl 立体分析器 {
let 默认配置 = 配置.unwrap_or_default();
let 配置组 = 配置组.unwrap_or_default();
let K线合成器 = K线合成器::new(符号.clone(), 周期组.clone());
let mut 单体分析器 = HashMap::new();
let mut 单体分析器: HashMap<i64, Arc<RwLock<观察者>>> = HashMap::new();
for &周期 in &周期组 {
let mut 当前配置 = 配置组
.get(&周期)
@@ -101,6 +96,18 @@ impl 立体分析器 {
}
}
// 对应 Python: K线合成器(符号, 周期组, self.__K线回调)
let 单体分析器_回调 = 单体分析器.clone();
let K线合成器 = K线合成器::new(
符号.clone(),
周期组.clone(),
Some(Box::new(
move |_信号类型: String, _标识: String, 周期: i64, 完成K线: K线| {
立体分析器::__K线回调_调度(&单体分析器_回调, 周期, 完成K线);
},
)),
);
Self {
周期组,
输入周期,
@@ -109,8 +116,28 @@ impl 立体分析器 {
}
}
/// 投喂K线 — 统一入口，接收最小周期K线
/// 匹配 Python __K线回调：合成器完成K线时喂给观察者
/// __K线回调 — 对应 Python 立体分析器.__K线回调
fn __K线回调(&self, _信号类型: String, _标识: String, 周期: i64, 完成K线: K线) {
if let Some(观察员) = self.单体分析器.get(&周期) {
let mut obs = 观察员.write().unwrap();
obs.增加原始K线(完成K线);
// 对应 Python: if 当前K线 := self._K线合成器.获取当前K线(周期)
// _完成K线刚清空当前K线，获取当前K线返回 None，所以这里不添加
}
}
/// 静态调度版本 — 用于回调闭包
fn __K线回调_调度(
单体分析器: &HashMap<i64, Arc<RwLock<观察者>>>,
周期: i64,
完成K线: K线,
) {
if let Some(观察员) = 单体分析器.get(&周期) {
观察员.write().unwrap().增加原始K线(完成K线);
}
}
/// 投喂K线 — 对应 Python 立体分析器.投喂K线
pub fn 投喂K线(&mut self, 普K: K线) {
if 普K.周期 != self.输入周期 {
panic!(
@@ -118,19 +145,7 @@ impl 立体分析器 {
普K.周期, self.输入周期
);
}
// Feed to synthesizer, get completion events
let 完成事件 = self.K线合成器.投喂K线(普K);
// Dispatch on completion events (matching Python's __K线回调)
for (周期, 完成K线) in 完成事件 {
if let Some(观察员) = self.单体分析器.get(&周期) {
观察员.write().unwrap().增加原始K线(完成K线);
if let Some(当前K线) = self.K线合成器.获取当前K线(周期) {
观察员.write().unwrap().增加原始K线(当前K线.clone());
}
}
}
self.K线合成器.投喂K线(普K);
}
/// 获取指定周期的观察者
@@ -138,8 +153,7 @@ impl 立体分析器 {
self.单体分析器.get(&周期).cloned()
}
/// 测试_保存数据 — 多级别数据拆分保存
/// 创建父目录 PyM_{标识}_{起始时间}_{结束时间}，各周期观察者保存到子目录
/// 测试_保存数据 — 对应 Python 立体分析器.测试_保存数据
pub fn 测试_保存数据(&self, root: Option<&str>) {
let 根目录 = match root {
Some(r) => std::path::PathBuf::from(r),
@@ -163,7 +177,6 @@ impl 立体分析器 {
.get(&self.输入周期)
.map(|o| o.read().unwrap().符号.clone())
.unwrap_or_default();
let 周期 = self
.单体分析器
.get(&self.输入周期)
@@ -189,4 +202,30 @@ impl 立体分析器 {
info!("多级别数据拆分保存完成，目录：{}", 保存路径.display());
}
/// 相等 — 各周期观察者全量比对，对应 Python `立体分析器相等`
pub fn 相等(&self, other: &Self, 浮点容差: f64) -> (bool, String) {
let 标签 = format!("立体分析器校验[A={:?},B={:?}]", self.周期组, other.周期组);
if self.周期组 != other.周期组 {
return (false, format!("{标签}: 周期组不一致"));
}
for 周期 in &self.周期组 {
let a_obs = match self.单体分析器.get(周期) {
Some(o) => o.read().unwrap(),
None => return (false, format!("{标签}: 周期{周期} 观察者不存在 (A)")),
};
let b_obs = match other.单体分析器.get(周期) {
Some(o) => o.read().unwrap(),
None => return (false, format!("{标签}: 周期{周期} 观察者不存在 (B)")),
};
let (eq, msg) = a_obs.相等(&b_obs, 浮点容差);
if !eq {
return (false, format!("{标签}: 周期{周期} >> {msg}"));
}
}
(true, format!("{标签}：所有周期观察者全量校验全部一致"))
}
}
chanlun/src/business/observer.rs
+260 -114
View File
@@ -261,78 +261,103 @@ impl 观察者 {
None => return,
};
// Step 2: 笔分析（无条件）
笔::分析(
当前分型,
&mut self.分型序列,
&mut self.笔序列,
&self.缠论K线序列,
&self.普通K线序列,
0,
&self.配置,
);
// Step 2: 笔分析
if self.配置.分析笔 {
笔::分析(
当前分型,
&mut self.分型序列,
&mut self.笔序列,
&self.缠论K线序列,
&self.普通K线序列,
0,
&self.配置,
);
}
if self.分型序列.is_empty() {
return;
}
// Step 3: 笔中枢分析（无条件）
中枢::分析(&self.笔序列, &mut self.笔_中枢序列, true, "", 0);
// Step 3: 笔中枢分析
if self.配置.分析笔中枢 {
中枢::分析(&self.笔序列, &mut self.笔_中枢序列, true, "", 0);
}
if self.笔序列.is_empty() {
return;
}
// Step 4: 线段分析 — 3 级递归
for i in 0..self.线段分析层次 {
if i == 0 {
线段::分析(
&self.笔序列,
&mut self.线段序列组[i],
&self.配置,
0,
&[相对方向::向上, 相对方向::向下],
);
} else {
let 源序列 = self.线段序列组[i - 1].clone();
线段::分析(
&源序列,
&mut self.线段序列组[i],
&self.配置,
0,
&[相对方向::向上, 相对方向::向下],
);
if self.配置.分析线段 || self.配置.分析线段中枢 {
for i in 0..self.线段分析层次 {
if i == 0 {
if self.配置.分析线段 {
线段::分析(
&self.笔序列,
&mut self.线段序列组[i],
&self.配置,
0,
&[相对方向::向上, 相对方向::向下],
);
}
} else {
if self.配置.分析线段 {
let 源序列 = self.线段序列组[i - 1].clone();
线段::分析(
&源序列,
&mut self.线段序列组[i],
&self.配置,
0,
&[相对方向::向上, 相对方向::向下],
);
}
}
if self.配置.分析线段中枢 {
中枢::分析(&self.线段序列组[i], &mut self.中枢序列组[i], true, "", 0);
}
}
中枢::分析(&self.线段序列组[i], &mut self.中枢序列组[i], true, "", 0);
}
// Step 5: 扩展线段分析 — 3 级递归
for i in 0..self.扩展线段分析层次 {
if i == 0 {
线段::扩展分析(&self.笔序列, &mut self.扩展线段序列组[i], &self.配置);
} else {
let 源序列 = self.扩展线段序列组[i - 1].clone();
线段::扩展分析(&源序列, &mut self.扩展线段序列组[i], &self.配置);
if self.配置.分析扩展线段 || self.配置.分析线段中枢 {
for i in 0..self.扩展线段分析层次 {
if i == 0 {
if self.配置.分析扩展线段 {
线段::扩展分析(&self.笔序列, &mut self.扩展线段序列组[i], &self.配置);
}
} else {
if self.配置.分析扩展线段 {
let 源序列 = self.扩展线段序列组[i - 1].clone();
线段::扩展分析(&源序列, &mut self.扩展线段序列组[i], &self.配置);
}
}
if self.配置.分析线段中枢 {
中枢::分析(
&self.扩展线段序列组[i],
&mut self.扩展中枢序列组[i],
true,
"",
0,
);
}
}
中枢::分析(
&self.扩展线段序列组[i],
&mut self.扩展中枢序列组[i],
true,
"",
0,
);
}
// Step 6: 混合扩展线段分析 — 3 级递归 (源 = 线段序列组[i])
// NOTE: 当 线段分析层次=0 时 线段序列组 为空，用 min 避免越界
for i in 0..self.混合扩展线段分析层次.min(self.线段序列组.len()) {
let 源序列 = self.线段序列组[i].clone();
线段::扩展分析(&源序列, &mut self.混合扩展线段序列组[i], &self.配置);
中枢::分析(
&self.混合扩展线段序列组[i],
&mut self.混合扩展中枢序列组[i],
true,
"",
0,
);
// Step 6: 混合扩展线段分析 — 3 级递归
if self.配置.分析扩展线段 || self.配置.分析线段中枢 {
for i in 0..self.混合扩展线段分析层次.min(self.线段序列组.len()) {
if self.配置.分析扩展线段 {
let 源序列 = self.线段序列组[i].clone();
线段::扩展分析(&源序列, &mut self.混合扩展线段序列组[i], &self.配置);
}
if self.配置.分析线段中枢 {
中枢::分析(
&self.混合扩展线段序列组[i],
&mut self.混合扩展中枢序列组[i],
true,
"",
0,
);
}
}
}
}
@@ -366,73 +391,103 @@ impl 观察者 {
self.混合扩展中枢序列组.push(Vec::new());
}
for i in 1..self.缠论K线序列.len() - 1 {
let 当前分型 = 分型::new(
Some(Arc::clone(&self.缠论K线序列[i - 1])),
Arc::clone(&self.缠论K线序列[i]),
Some(Arc::clone(&self.缠论K线序列[i + 1])),
);
笔::分析(
Arc::new(当前分型),
&mut self.分型序列,
&mut self.笔序列,
&self.缠论K线序列,
&self.普通K线序列,
0,
&self.配置,
);
}
中枢::分析(&self.笔序列, &mut self.笔_中枢序列, true, "", 0);
for i in 0..self.线段分析层次 {
if i == 0 {
线段::分析(
&self.笔序列,
&mut self.线段序列组[i],
&self.配置,
0,
&[相对方向::向上, 相对方向::向下],
if self.配置.分析笔 {
for i in 1..self.缠论K线序列.len() - 1 {
let 当前分型 = 分型::new(
Some(Arc::clone(&self.缠论K线序列[i - 1])),
Arc::clone(&self.缠论K线序列[i]),
Some(Arc::clone(&self.缠论K线序列[i + 1])),
);
} else {
let 源序列 = self.线段序列组[i - 1].clone();
线段::分析(
&源序列,
&mut self.线段序列组[i],
&self.配置,
笔::分析(
Arc::new(当前分型),
&mut self.分型序列,
&mut self.笔序列,
&self.缠论K线序列,
&self.普通K线序列,
0,
&[相对方向::向上, 相对方向::向下],
&self.配置,
);
}
中枢::分析(&self.线段序列组[i], &mut self.中枢序列组[i], true, "", 0);
}
for i in 0..self.扩展线段分析层次 {
if i == 0 {
线段::扩展分析(&self.笔序列, &mut self.扩展线段序列组[i], &self.配置);
} else {
let 源序列 = self.扩展线段序列组[i - 1].clone();
线段::扩展分析(&源序列, &mut self.扩展线段序列组[i], &self.配置);
if self.笔序列.is_empty() {
return;
}
if self.配置.分析笔中枢 {
中枢::分析(&self.笔序列, &mut self.笔_中枢序列, true, "", 0);
}
if self.配置.分析线段 || self.配置.分析线段中枢 {
for i in 0..self.线段分析层次 {
if i == 0 {
if self.配置.分析线段 {
线段::分析(
&self.笔序列,
&mut self.线段序列组[i],
&self.配置,
0,
&[相对方向::向上, 相对方向::向下],
);
}
} else {
if self.配置.分析线段 {
let 源序列 = self.线段序列组[i - 1].clone();
线段::分析(
&源序列,
&mut self.线段序列组[i],
&self.配置,
0,
&[相对方向::向上, 相对方向::向下],
);
}
}
if self.配置.分析线段中枢 {
中枢::分析(&self.线段序列组[i], &mut self.中枢序列组[i], true, "", 0);
}
}
中枢::分析(
&self.扩展线段序列组[i],
&mut self.扩展中枢序列组[i],
true,
"",
0,
);
}
for i in 0..self.混合扩展线段分析层次.min(self.线段序列组.len()) {
let 源序列 = self.线段序列组[i].clone();
线段::扩展分析(&源序列, &mut self.混合扩展线段序列组[i], &self.配置);
中枢::分析(
&self.混合扩展线段序列组[i],
&mut self.混合扩展中枢序列组[i],
true,
"",
0,
);
if self.配置.分析扩展线段 || self.配置.分析线段中枢 {
for i in 0..self.扩展线段分析层次 {
if i == 0 {
if self.配置.分析扩展线段 {
线段::扩展分析(&self.笔序列, &mut self.扩展线段序列组[i], &self.配置);
}
} else {
if self.配置.分析扩展线段 {
let 源序列 = self.扩展线段序列组[i - 1].clone();
线段::扩展分析(&源序列, &mut self.扩展线段序列组[i], &self.配置);
}
}
if self.配置.分析线段中枢 {
中枢::分析(
&self.扩展线段序列组[i],
&mut self.扩展中枢序列组[i],
true,
"",
0,
);
}
}
}
if self.配置.分析扩展线段 || self.配置.分析线段中枢 {
for i in 0..self.混合扩展线段分析层次.min(self.线段序列组.len()) {
if self.配置.分析扩展线段 {
let 源序列 = self.线段序列组[i].clone();
线段::扩展分析(&源序列, &mut self.混合扩展线段序列组[i], &self.配置);
}
if self.配置.分析线段中枢 {
中枢::分析(
&self.混合扩展线段序列组[i],
&mut self.混合扩展中枢序列组[i],
true,
"",
0,
);
}
}
}
}
@@ -582,6 +637,97 @@ impl 观察者 {
self.配置 = 配置;
self.加载本地数据(文件路径)
}
/// 相等 — 全量序列逐项比对，双端一致性验证，对应 Python `观察者相等`
pub fn 相等(&self, other: &Self, 浮点容差: f64) -> (bool, String) {
let 标签 = format!("观察者校验[A={},B={}]", self.标识(), other.标识());
if self.缠论K线序列.len() != other.缠论K线序列.len() {
return (
false,
format!(
"{标签}: 缠K序列长度不一致 A={},B={}",
self.缠论K线序列.len(),
other.缠论K线序列.len()
),
);
}
if self.分型序列.len() != other.分型序列.len() {
return (
false,
format!(
"{标签}: 分型序列长度不一致 A={},B={}",
self.分型序列.len(),
other.分型序列.len()
),
);
}
if self.笔序列.len() != other.笔序列.len() {
return (
false,
format!(
"{标签}: 笔序列长度不一致 A={},B={}",
self.笔序列.len(),
other.笔序列.len()
),
);
}
for (i, (a, b)) in self.笔序列.iter().zip(other.笔序列.iter()).enumerate() {
let (eq, msg) = a.相等(b, 浮点容差);
if !eq {
return (false, format!("{标签}: 笔#{i}不一致 >> {msg}"));
}
}
if self.笔_中枢序列.len() != other.笔_中枢序列.len() {
return (false, format!("{标签}: 笔中枢序列长度不一致"));
}
for (i, (a, b)) in self
.笔_中枢序列
.iter()
.zip(other.笔_中枢序列.iter())
.enumerate()
{
let (eq, msg) = a.相等(b, 浮点容差);
if !eq {
return (false, format!("{标签}: 笔中枢#{i}不一致 >> {msg}"));
}
}
for level in 0..self.线段分析层次.min(other.线段分析层次) {
let a_segs = &self.线段序列组[level];
let b_segs = &other.线段序列组[level];
if a_segs.len() != b_segs.len() {
return (false, format!("{标签}: 线段序列组[{level}]长度不一致"));
}
for (i, (a, b)) in a_segs.iter().zip(b_segs.iter()).enumerate() {
let (eq, msg) = a.相等(b, 浮点容差);
if !eq {
return (
false,
format!("{标签}: 线段序列组[{level}]#{i}不一致 >> {msg}"),
);
}
}
let a_hubs = &self.中枢序列组[level];
let b_hubs = &other.中枢序列组[level];
if a_hubs.len() != b_hubs.len() {
return (false, format!("{标签}: 中枢序列组[{level}]长度不一致"));
}
for (i, (a, b)) in a_hubs.iter().zip(b_hubs.iter()).enumerate() {
let (eq, msg) = a.相等(b, 浮点容差);
if !eq {
return (
false,
format!("{标签}: 中枢序列组[{level}]#{i}不一致 >> {msg}"),
);
}
}
}
(true, format!("{标签}：全量序列校验全部一致"))
}
}
#[cfg(test)]
chanlun/src/business/synthesizer.rs
+156 -28
View File
@@ -31,11 +31,19 @@ pub struct K线合成器 {
pub 周期组: Vec<i64>,
pub 当前K线: HashMap<i64, Option<K线>>,
pub 合成K线列表: HashMap<i64, Vec<K线>>,
/// 事件回调 — K线完成时触发，对应 Python K线合成器.事件回调
/// 签名: fn(信号类型: str, 标识: str, 周期: i64, 完成K线: K线)
/// 在 _完成K线 清空当前K线后、新K线创建前触发
事件回调: Option<Box<dyn Fn(String, String, i64, K线) + Send + Sync>>,
}
impl K线合成器 {
/// 创建K线合成器，按周期升序排列，初始化当前K线和合成K线列表
pub fn new(标识: String, 周期组: Vec<i64>) -> Self {
/// 创建K线合成器 — 对应 Python K线合成器.__init__(标识, 周期组, 事件回调=None)
pub fn new(
标识: String,
周期组: Vec<i64>,
事件回调: Option<Box<dyn Fn(String, String, i64, K线) + Send + Sync>>,
) -> Self {
let mut 周期组 = 周期组;
周期组.sort();
@@ -51,37 +59,33 @@ impl K线合成器 {
周期组,
当前K线,
合成K线列表,
事件回调,
}
}
/// 设置事件回调 — 对应 Python `设置事件回调`
pub fn 设置事件回调(
&mut self,
回调: Box<dyn Fn(String, String, i64, K线) + Send + Sync>,
) {
self.事件回调 = Some(回调);
}
/// 投喂 — 便捷入口，直接从 OHLCV 创建 K线 并投喂
pub fn 投喂(
&mut self,
时间戳: i64,
开: f64,
高: f64,
低: f64,
收: f64,
量: f64,
) -> Vec<(i64, K线)> {
pub fn 投喂(&mut self, 时间戳: i64, 开: f64, 高: f64, 低: f64, 收: f64, 量: f64) {
let 普K = K线::创建普K(&self.标识, 时间戳, 开, 高, 低, 收, 量, 0, 0);
self.投喂K线(普K)
self.投喂K线(普K);
}
/// 投喂K线 — 输入最小周期K线，合成为所有目标周期
/// 返回本次投喂完成了哪些周期的K线（周期 → 完成K线）
pub fn 投喂K线(&mut self, 普K: K线) -> Vec<(i64, K线)> {
let mut 完成记录 = Vec::new();
pub fn 投喂K线(&mut self, 普K: K线) {
let 周期组 = self.周期组.clone();
for 周期 in 周期组 {
if let Some(完成K线) = self._处理单个周期(周期, &普K) {
完成记录.push((周期, 完成K线));
}
self._处理单个周期(周期, &普K);
}
完成记录
}
fn _处理单个周期(&mut self, 周期: i64, 普K: &K线) -> Option<K线> {
fn _处理单个周期(&mut self, 周期: i64, 普K: &K线) {
let 目标时间戳 = self._对齐时间戳(普K.时间戳, 周期);
let 相同时间 = self.当前K线[&周期]
.as_ref()
@@ -91,19 +95,17 @@ impl K线合成器 {
if self.当前K线[&周期].is_none() {
let 新K线 = self._创建新K线(周期, 目标时间戳, 普K);
self.当前K线.insert(周期, Some(新K线));
None
} else if 相同时间 {
let ent = self.当前K线.get_mut(&周期).unwrap();
Self::_更新K线(ent.as_mut().unwrap(), 普K);
None
} else {
let 完成K线 = self._完成K线(周期);
self._完成K线(周期);
let 新K线 = self._创建新K线(周期, 目标时间戳, 普K);
self.当前K线.insert(周期, Some(新K线));
完成K线
}
}
/// 对齐时间戳到周期边界 — 对应 Python `_对齐时间戳`
fn _对齐时间戳(&self, 时间戳: i64, 周期: i64) -> i64 {
if 周期 == 0 {
panic!("_对齐时间戳: 周期不能为0");
@@ -111,6 +113,7 @@ impl K线合成器 {
(时间戳 / 周期) * 周期
}
/// 创建新K线 — 对应 Python `_创建新K线`
fn _创建新K线(&self, 周期: i64, 时间戳: i64, 普K: &K线) -> K线 {
let 序号 = self
.合成K线列表
@@ -132,6 +135,7 @@ impl K线合成器 {
)
}
/// 更新K线 — 对应 Python `_更新K线`
fn _更新K线(当前K线: &mut K线, 新数据: &K线) {
当前K线.高 = 当前K线.高.max(新数据.高);
当前K线.低 = 当前K线.低.min(新数据.低);
@@ -139,9 +143,14 @@ impl K线合成器 {
当前K线.成交量 += 新数据.成交量;
}
fn _完成K线(&mut self, 周期: i64) -> Option<K线> {
/// 完成K线 — 对应 Python `_完成K线`
/// 清空当前K线后，触发事件回调（此时获取当前K线返回 None）
fn _完成K线(&mut self, 周期: i64) {
let ent = self.当前K线.get_mut(&周期).unwrap();
let mut k线 = ent.take()?;
let mut k线 = match ent.take() {
Some(k) => k,
None => return,
};
k线.序号 = self
.合成K线列表
.get(&周期)
@@ -151,11 +160,130 @@ impl K线合成器 {
let 完成K线 = k线.clone();
self.合成K线列表.get_mut(&周期).unwrap().push(k线);
Some(完成K线)
// 对应 Python _完成K线：清空当前K线后、新K线创建前触发回调
self._产生完成K线信号(周期, 完成K线);
}
/// 获取指定周期当前正在合成的K线
/// 产生完成K线信号 — 对应 Python `_产生完成K线信号`
fn _产生完成K线信号(&self, 周期: i64, 完成K线: K线) {
if let Some(ref cb) = self.事件回调 {
cb("K线完成".into(), self.标识.clone(), 周期, 完成K线);
}
}
/// 获取指定周期当前正在合成的K线 — 对应 Python `获取当前K线`
pub fn 获取当前K线(&self, 周期: i64) -> Option<&K线> {
self.当前K线.get(&周期).and_then(|k| k.as_ref())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_创建合成器_初始状态正确() {
let synth = K线合成器::new("btcusd".into(), vec![60, 300], None);
assert_eq!(synth.标识, "btcusd");
assert_eq!(synth.周期组, vec![60, 300]);
assert!(synth.事件回调.is_none());
assert!(synth.当前K线[&60].is_none());
assert!(synth.当前K线[&300].is_none());
}
#[test]
fn test_设置事件回调() {
let mut synth = K线合成器::new("btcusd".into(), vec![60], None);
assert!(synth.事件回调.is_none());
synth.设置事件回调(Box::new(|_, _, _, _| {}));
assert!(synth.事件回调.is_some());
}
#[test]
fn test_对齐时间戳() {
let synth = K线合成器::new("t".into(), vec![300], None);
assert_eq!(synth._对齐时间戳(1218124800, 300), 1218124800);
assert_eq!(synth._对齐时间戳(1218124801, 300), 1218124800);
assert_eq!(synth._对齐时间戳(1218125099, 300), 1218124800);
assert_eq!(synth._对齐时间戳(1218125100, 300), 1218125100);
}
#[test]
fn test_创建新K线_序号递进() {
let mut synth = K线合成器::new("btcusd".into(), vec![300], None);
{
let first = K线::创建普K("btcusd", 0, 100.0, 110.0, 90.0, 105.0, 1000.0, 0, 300);
synth.合成K线列表.get_mut(&300).unwrap().push(first);
}
let new_bar = K线::创建普K("btcusd", 100, 200.0, 210.0, 190.0, 205.0, 500.0, 0, 60);
let created = synth._创建新K线(300, 300, &new_bar);
assert_eq!(created.序号, 1);
assert_eq!(created.时间戳, 300);
assert_eq!(created.开盘价, 200.0);
}
#[test]
fn test_更新K线_高低更新() {
let mut current = K线::创建普K("t", 0, 100.0, 110.0, 90.0, 105.0, 100.0, 0, 300);
let new_data = K线::创建普K("t", 0, 102.0, 115.0, 85.0, 108.0, 50.0, 0, 60);
K线合成器::_更新K线(&mut current, &new_data);
assert_eq!(current.高, 115.0);
assert_eq!(current.低, 85.0);
assert_eq!(current.收盘价, 108.0);
assert_eq!(current.成交量, 150.0);
}
#[test]
fn test_完成K线_返回完成K并将当前置空() {
let mut synth = K线合成器::new("btcusd".into(), vec![300], None);
let bar = K线::创建普K("btcusd", 300, 100.0, 110.0, 90.0, 105.0, 1000.0, 0, 300);
synth.当前K线.insert(300, Some(bar));
synth._完成K线(300);
assert!(synth.当前K线[&300].is_none());
assert_eq!(synth.合成K线列表[&300].len(), 1);
}
#[test]
fn test_完成K线_事件回调触发() {
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
let callback_fired = Arc::new(AtomicBool::new(false));
let cb_flag = Arc::clone(&callback_fired);
let mut synth = K线合成器::new(
"btcusd".into(),
vec![300],
Some(Box::new(move |信号类型, 标识, 周期, _完成K线| {
assert_eq!(信号类型, "K线完成");
assert_eq!(标识, "btcusd");
assert_eq!(周期, 300);
cb_flag.store(true, Ordering::SeqCst);
})),
);
let bar1 = K线::创建普K("btcusd", 0, 100.0, 110.0, 90.0, 105.0, 1000.0, 0, 300);
synth.当前K线.insert(300, Some(bar1));
let bar2 = K线::创建普K("btcusd", 400, 200.0, 210.0, 190.0, 205.0, 500.0, 0, 60);
synth.投喂K线(bar2);
assert!(callback_fired.load(Ordering::SeqCst));
}
#[test]
fn test_投喂K线_多周期合成() {
let mut synth = K线合成器::new("btcusd".into(), vec![60, 300], None);
synth.投喂K线(K线::创建普K(
"btcusd", 60, 100.0, 110.0, 90.0, 105.0, 100.0, 0, 60,
));
assert!(synth.获取当前K线(60).is_some());
assert!(synth.获取当前K线(300).is_some());
}
#[test]
fn test_投喂_便捷方法() {
let mut synth = K线合成器::new("btcusd".into(), vec![300], None);
synth.投喂(1218124800, 100.0, 110.0, 90.0, 105.0, 1000.0);
assert!(synth.获取当前K线(300).is_some());
}
}