4 Commits

Author SHA1 Message Date
YuWuKunCheng a8df8cd187 移除 无意义配置项 2026-06-09 15:03:34 +08:00
YuWuKunCheng 9c7686d104 Nothing 2026-06-09 01:06:52 +08:00
YuWuKunCheng 34c42ecd68 修复 异常BUG 2026-06-08 23:29:00 +08:00
YuWuKunCheng 8a2afb9ed0 修复 观察者混合扩展边界问题 2026-06-07 20:45:09 +08:00
17 changed files with 1888 additions and 453 deletions
+3
View File
@@ -1,5 +1,8 @@
name: 构建发布
permissions:
contents: read
on:
push:
tags:
+129 -73
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 chanlun.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):
"""买卖点类型 — 缠论的三类买卖点及扩展类型。
@@ -850,19 +929,6 @@ class 缠论配置:
买卖点_指标匹配_MACD: bool = True, # 买在负,卖在正!
买卖点_指标匹配_KDJ: bool = True, # 买在死叉之后,卖在金叉之后
买卖点_指标匹配_RSI: bool = True, # 买在均线之下,卖在均线之上
# 以下字段酌情废弃
买卖点_背离率: float = float("inf"),
买卖点_T2_回调阈值: float = 1.0,
买卖点_T2S_最大层级: int = 3,
买卖点_峰值条件: bool = False,
买卖点_计算方式: str = "",
买卖点_计算线段BSP1: bool = True,
买卖点_处理BSP2: bool = True,
买卖点_计算线段BSP3: bool = True,
买卖点_依赖T1: bool = True,
买卖点_中枢来源: str = "",
买卖点_调试输出: bool = False,
# 以上字段酌情废弃
线段内部背驰_MACD: bool = True,
线段内部背驰_斜率: bool = True,
线段内部背驰_测度: bool = True,
@@ -939,17 +1005,6 @@ class 缠论配置:
self.买卖点_指标匹配_MACD = 买卖点_指标匹配_MACD
self.买卖点_指标匹配_KDJ = 买卖点_指标匹配_KDJ
self.买卖点_指标匹配_RSI = 买卖点_指标匹配_RSI
self.买卖点_背离率 = 买卖点_背离率
self.买卖点_T2_回调阈值 = 买卖点_T2_回调阈值
self.买卖点_T2S_最大层级 = 买卖点_T2S_最大层级
self.买卖点_峰值条件 = 买卖点_峰值条件
self.买卖点_计算方式 = 买卖点_计算方式
self.买卖点_计算线段BSP1 = 买卖点_计算线段BSP1
self.买卖点_处理BSP2 = 买卖点_处理BSP2
self.买卖点_计算线段BSP3 = 买卖点_计算线段BSP3
self.买卖点_依赖T1 = 买卖点_依赖T1
self.买卖点_中枢来源 = 买卖点_中枢来源
self.买卖点_调试输出 = 买卖点_调试输出
self.线段内部背驰_MACD = 线段内部背驰_MACD
self.线段内部背驰_斜率 = 线段内部背驰_斜率
self.线段内部背驰_测度 = 线段内部背驰_测度
@@ -1036,17 +1091,6 @@ class 缠论配置:
"买卖点_指标匹配_MACD": {"annotation": bool, "default": True},
"买卖点_指标匹配_KDJ": {"annotation": bool, "default": True},
"买卖点_指标匹配_RSI": {"annotation": bool, "default": True},
"买卖点_背离率": {"annotation": float, "default": float("inf")},
"买卖点_T2_回调阈值": {"annotation": float, "default": 1.0},
"买卖点_T2S_最大层级": {"annotation": int, "default": 3},
"买卖点_峰值条件": {"annotation": bool, "default": False},
"买卖点_计算方式": {"annotation": str, "default": ""},
"买卖点_计算线段BSP1": {"annotation": bool, "default": True},
"买卖点_处理BSP2": {"annotation": bool, "default": True},
"买卖点_计算线段BSP3": {"annotation": bool, "default": True},
"买卖点_依赖T1": {"annotation": bool, "default": True},
"买卖点_中枢来源": {"annotation": str, "default": ""},
"买卖点_调试输出": {"annotation": bool, "default": False},
"线段内部背驰_MACD": {"annotation": bool, "default": True},
"线段内部背驰_斜率": {"annotation": bool, "default": True},
"线段内部背驰_测度": {"annotation": bool, "default": True},
@@ -2847,7 +2891,7 @@ class K线:
标识=标识,
序号=序号,
周期=周期,
时间戳=时间戳 if isinstance(时间戳, datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
时间戳=时间戳 if isinstance(时间戳, datetime_module.datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
开盘价=开盘价,
最高价=最高价,
最低价=最低价,
@@ -4527,12 +4571,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 +5175,7 @@ class 线段:
raise RuntimeError(f"线段._向序列中添加[{行号}], 之前线段[-1] not in 待添加虚线!", 之前线段)
待添加线段.序号 = 之前线段.序号 + 1
待添加线段.前一缺口 = 线段.获取缺口(之前线段)
待添加线段.前一缺口 = 线段.获取缺口(之前线段) if not 之前线段.短路修正 else None
待添加线段.前一结束位置 = 之前线段.基础序列[-1]
if 线段.四象(之前线段) in ("老阴", "老阳"):
@@ -5160,7 +5204,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 +5299,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 +5653,7 @@ class 线段:
线段._弹出扩展线段(线段序列, 当前线段, sys._getframe().f_lineno)
return 线段递归扩展分析(虚线序列, 线段序列, 配置)
线段._武终(当前线段, sys._getframe().f_lineno)
线段._武终(当前线段, sys._getframe().f_lineno) # TODO 添加错误处理机制
if 当前线段.基础序列[-1].序号 + 3 > 虚线序列[-1].序号:
return None
@@ -6359,33 +6403,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(self.混合扩展线段分析层次):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 测试_保存数据(self, root: str = None) -> str:
"""拆分各序列数据,单独存文件,文件名为对应变量名
@@ -6429,7 +6473,7 @@ class 观察者:
for i in range(self.扩展线段分析层次):
保存序列(self.扩展线段序列组[i])
保存序列(self.扩展中枢序列组[i])
for i in range(self.线段分析层次):
for i in range(self.混合扩展线段分析层次):
保存序列(self.混合扩展线段序列组[i])
保存序列(self.混合扩展中枢序列组[i])
@@ -6467,29 +6511,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(self.混合扩展线段分析层次):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 加载本地数据(self, 文件路径: str):
"""重置基础序列后加载数据文件
@@ -6511,6 +6557,14 @@ class 观察者:
:param 观察员: 可选已有观察者实例不传则自动创建
:return: 观察者实例
"""
if "_err-" in str(文件路径) and os.path.exists(str(文件路径).replace(".nb", ".json")):
异常配置 = 缠论配置.加载配置(str(文件路径).replace(".nb", ".json"))
差异 = 缠论配置().对比(异常配置)
传入差异 = 缠论配置().对比(配置)
传入差异.update(差异)
配置 = 缠论配置(**传入差异)
logger.info(f"加载异常配置+传入差异: {传入差异}")
name = Path(文件路径).name.split(".")[0]
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
if 观察员 is None:
@@ -6771,7 +6825,7 @@ class 立体分析器:
def 测试_保存数据(self, root: str = None):
"""拆分各序列数据,单独存文件。
:param root: 保存根目录默认取脚本所在目录
:param root: 保存根目录默认系统临时目录
:return: 数据保存目录路径
"""
# 生成存储根目录
@@ -6842,6 +6896,8 @@ def 测试_指标挂载(配置: 缠论配置):
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
周期 = int(周期)
观察员 = 观察者(符号, 周期, 配置)
观察员.线段分析层次 = 0
观察员.重置基础序列()
def 魔法():
启动时间 = datetime.now()
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "chanlun-py"
version = "26.6.44"
version = "26.6.47"
edition = "2024"
description = "缠论技术分析库 — Rust 高性能 Python 绑定"
authors = ["YuYuKunKun"]
@@ -12,7 +12,7 @@ crate-type = ["cdylib"]
name = "chanlun"
[dependencies]
chanlun = "26.6.2" #{ path = "../chanlun" }
chanlun = "26.6.3" # { path = "../chanlun" }
lru = "0.18"
pyo3 = { version = "0.28", features = ["experimental-inspect"] }
serde_json = "1"
+8 -4
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__) ==========
@@ -825,7 +828,8 @@ class 缠论配置:
def to_dict(self) -> Dict[str, Any]: ...
def to_json(self) -> str: ...
def 保存配置(self, path: str = "缠论配置.json") -> None: ...
def 对比(self, other: 缠论配置) -> Dict[str, Tuple[Any, Any]]: ...
def 对比(self, other: 缠论配置) -> Dict[str, Any]: ...
def model_copy(self, update: Optional[Dict[str, Any]] = None) -> 缠论配置: ...
@classmethod
def 加载配置(cls, path: str = "缠论配置.json") -> 缠论配置: ...
@classmethod
+8 -4
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__) ==========
@@ -850,7 +853,8 @@ class 缠论配置:
def to_dict(self) -> Dict[str, Any]: ...
def to_json(self) -> str: ...
def 保存配置(self, path: str = "缠论配置.json") -> None: ...
def 对比(self, other: 缠论配置) -> Dict[str, Tuple[Any, Any]]: ...
def 对比(self, other: 缠论配置) -> Dict[str, Any]: ...
def model_copy(self, update: Optional[Dict[str, Any]] = None) -> 缠论配置: ...
@classmethod
def 加载配置(cls, path: str = "缠论配置.json") -> 缠论配置: ...
@classmethod
+129 -73
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 chanlun.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):
"""买卖点类型 — 缠论的三类买卖点及扩展类型。
@@ -850,19 +929,6 @@ class 缠论配置:
买卖点_指标匹配_MACD: bool = True, # 买在负,卖在正!
买卖点_指标匹配_KDJ: bool = True, # 买在死叉之后,卖在金叉之后
买卖点_指标匹配_RSI: bool = True, # 买在均线之下,卖在均线之上
# 以下字段酌情废弃
买卖点_背离率: float = float("inf"),
买卖点_T2_回调阈值: float = 1.0,
买卖点_T2S_最大层级: int = 3,
买卖点_峰值条件: bool = False,
买卖点_计算方式: str = "",
买卖点_计算线段BSP1: bool = True,
买卖点_处理BSP2: bool = True,
买卖点_计算线段BSP3: bool = True,
买卖点_依赖T1: bool = True,
买卖点_中枢来源: str = "",
买卖点_调试输出: bool = False,
# 以上字段酌情废弃
线段内部背驰_MACD: bool = True,
线段内部背驰_斜率: bool = True,
线段内部背驰_测度: bool = True,
@@ -939,17 +1005,6 @@ class 缠论配置:
self.买卖点_指标匹配_MACD = 买卖点_指标匹配_MACD
self.买卖点_指标匹配_KDJ = 买卖点_指标匹配_KDJ
self.买卖点_指标匹配_RSI = 买卖点_指标匹配_RSI
self.买卖点_背离率 = 买卖点_背离率
self.买卖点_T2_回调阈值 = 买卖点_T2_回调阈值
self.买卖点_T2S_最大层级 = 买卖点_T2S_最大层级
self.买卖点_峰值条件 = 买卖点_峰值条件
self.买卖点_计算方式 = 买卖点_计算方式
self.买卖点_计算线段BSP1 = 买卖点_计算线段BSP1
self.买卖点_处理BSP2 = 买卖点_处理BSP2
self.买卖点_计算线段BSP3 = 买卖点_计算线段BSP3
self.买卖点_依赖T1 = 买卖点_依赖T1
self.买卖点_中枢来源 = 买卖点_中枢来源
self.买卖点_调试输出 = 买卖点_调试输出
self.线段内部背驰_MACD = 线段内部背驰_MACD
self.线段内部背驰_斜率 = 线段内部背驰_斜率
self.线段内部背驰_测度 = 线段内部背驰_测度
@@ -1036,17 +1091,6 @@ class 缠论配置:
"买卖点_指标匹配_MACD": {"annotation": bool, "default": True},
"买卖点_指标匹配_KDJ": {"annotation": bool, "default": True},
"买卖点_指标匹配_RSI": {"annotation": bool, "default": True},
"买卖点_背离率": {"annotation": float, "default": float("inf")},
"买卖点_T2_回调阈值": {"annotation": float, "default": 1.0},
"买卖点_T2S_最大层级": {"annotation": int, "default": 3},
"买卖点_峰值条件": {"annotation": bool, "default": False},
"买卖点_计算方式": {"annotation": str, "default": ""},
"买卖点_计算线段BSP1": {"annotation": bool, "default": True},
"买卖点_处理BSP2": {"annotation": bool, "default": True},
"买卖点_计算线段BSP3": {"annotation": bool, "default": True},
"买卖点_依赖T1": {"annotation": bool, "default": True},
"买卖点_中枢来源": {"annotation": str, "default": ""},
"买卖点_调试输出": {"annotation": bool, "default": False},
"线段内部背驰_MACD": {"annotation": bool, "default": True},
"线段内部背驰_斜率": {"annotation": bool, "default": True},
"线段内部背驰_测度": {"annotation": bool, "default": True},
@@ -2847,7 +2891,7 @@ class K线:
标识=标识,
序号=序号,
周期=周期,
时间戳=时间戳 if isinstance(时间戳, datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
时间戳=时间戳 if isinstance(时间戳, datetime_module.datetime) else 转化为时间戳(时间戳), # 注意此处只为兼容Rust绑定
开盘价=开盘价,
最高价=最高价,
最低价=最低价,
@@ -4527,12 +4571,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 +5175,7 @@ class 线段:
raise RuntimeError(f"线段._向序列中添加[{行号}], 之前线段[-1] not in 待添加虚线!", 之前线段)
待添加线段.序号 = 之前线段.序号 + 1
待添加线段.前一缺口 = 线段.获取缺口(之前线段)
待添加线段.前一缺口 = 线段.获取缺口(之前线段) if not 之前线段.短路修正 else None
待添加线段.前一结束位置 = 之前线段.基础序列[-1]
if 线段.四象(之前线段) in ("老阴", "老阳"):
@@ -5160,7 +5204,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 +5299,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 +5653,7 @@ class 线段:
线段._弹出扩展线段(线段序列, 当前线段, sys._getframe().f_lineno)
return 线段递归扩展分析(虚线序列, 线段序列, 配置)
线段._武终(当前线段, sys._getframe().f_lineno)
线段._武终(当前线段, sys._getframe().f_lineno) # TODO 添加错误处理机制
if 当前线段.基础序列[-1].序号 + 3 > 虚线序列[-1].序号:
return None
@@ -6359,33 +6403,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(self.混合扩展线段分析层次):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 测试_保存数据(self, root: str = None) -> str:
"""拆分各序列数据,单独存文件,文件名为对应变量名
@@ -6429,7 +6473,7 @@ class 观察者:
for i in range(self.扩展线段分析层次):
保存序列(self.扩展线段序列组[i])
保存序列(self.扩展中枢序列组[i])
for i in range(self.线段分析层次):
for i in range(self.混合扩展线段分析层次):
保存序列(self.混合扩展线段序列组[i])
保存序列(self.混合扩展中枢序列组[i])
@@ -6467,29 +6511,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(self.混合扩展线段分析层次):
线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))):
self.配置.分析扩展线段 and 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置)
self.配置.分析线段中枢 and 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0)
def 加载本地数据(self, 文件路径: str):
"""重置基础序列后加载数据文件
@@ -6511,6 +6557,14 @@ class 观察者:
:param 观察员: 可选已有观察者实例不传则自动创建
:return: 观察者实例
"""
if "_err-" in str(文件路径) and os.path.exists(str(文件路径).replace(".nb", ".json")):
异常配置 = 缠论配置.加载配置(str(文件路径).replace(".nb", ".json"))
差异 = 缠论配置().对比(异常配置)
传入差异 = 缠论配置().对比(配置)
传入差异.update(差异)
配置 = 缠论配置(**传入差异)
logger.info(f"加载异常配置+传入差异: {传入差异}")
name = Path(文件路径).name.split(".")[0]
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
if 观察员 is None:
@@ -6771,7 +6825,7 @@ class 立体分析器:
def 测试_保存数据(self, root: str = None):
"""拆分各序列数据,单独存文件。
:param root: 保存根目录默认取脚本所在目录
:param root: 保存根目录默认系统临时目录
:return: 数据保存目录路径
"""
# 生成存储根目录
@@ -6842,6 +6896,8 @@ def 测试_指标挂载(配置: 缠论配置):
符号, 周期, 起始时间戳, 结束时间戳 = name.split("-")
周期 = int(周期)
观察员 = 观察者(符号, 周期, 配置)
观察员.线段分析层次 = 0
观察员.重置基础序列()
def 魔法():
启动时间 = datetime.now()
+2 -1
View File
@@ -4,7 +4,7 @@ build-backend = "maturin"
[project]
name = "chanlun"
version = "2606.44"
version = "2606.47"
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",
]
+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<()> {
+29 -16
View File
@@ -169,8 +169,11 @@ impl 缠论配置Py {
/// 将配置导出为 Python 字典。
fn to_dict(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
let dict = PyDict::new(py);
let valid = chanlun::config::::model_fields();
for (k, v) in &self.fields {
dict.set_item(k, v.clone_ref(py))?;
if valid.contains(&k.as_str()) {
dict.set_item(k, v.clone_ref(py))?;
}
}
Ok(dict.into())
}
@@ -252,26 +255,36 @@ impl 缠论配置Py {
Ok(result.into())
}
/// 比较当前配置与另一个配置的差异
#[allow(clippy::type_complexity)]
fn (
&self,
py: Python<'_>,
other: &Bound<'_, Py>,
) -> PyResult<HashMap<String, (Py<PyAny>, Py<PyAny>)>> {
/// 创建当前配置的拷贝并可选择更新字段(对应 Python model_copy(update={...}, deep=True)
#[pyo3(signature = (update = None))]
fn model_copy(&self, py: Python<'_>, update: Option<&Bound<'_, PyDict>>) -> PyResult<Self> {
let current = self.to_dict(py)?;
if let Some(updates) = update {
for (key, value) in updates.iter() {
current.bind(py).set_item(key, value)?;
}
}
Self::from_dict(&py.get_type::<Self>(), current.bind(py))
}
/// 比较当前配置与另一个配置的差异(对应 Python 对比 → dict[字段名, 新值])
fn (&self, py: Python<'_>, other: &Bound<'_, Py>) -> PyResult<Py<PyAny>> {
let other_ref = other.borrow();
let mut diff = HashMap::new();
for (key, val) in &self.fields {
if let Some(other_val) = other_ref.fields.get(key) {
let a = val.clone_ref(py);
let dict = PyDict::new(py);
let valid = chanlun::config::::model_fields();
for key in valid {
if let (Some(self_val), Some(other_val)) =
(self.fields.get(*key), other_ref.fields.get(*key))
{
let a = self_val.clone_ref(py);
let b = other_val.clone_ref(py);
let eq = a.bind(py).eq(b.bind(py))?;
if !eq {
diff.insert(key.clone(), (val.clone_ref(py), other_val.clone_ref(py)));
dict.set_item(*key, b)?;
}
}
}
Ok(diff)
Ok(dict.into())
}
}
@@ -393,9 +406,9 @@ fn validate_field(
) -> Result<(), String> {
use serde_json::Value;
// 输入为 null → 跳过(保留默认
// 输入为 null → 保留(对应 Optional/Infinity 字段
if input.is_null() {
return Err("值为 null".into());
return Ok(());
}
// 字符串字段:检查有效值白名单
+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(())
}
+625
View File
@@ -1764,6 +1764,83 @@ class Test指标挂载(unittest.TestCase):
self.assertNotEqual(macd_before, macd_after)
# ============================================================
# 线段分析层次 = 0 时不崩溃
# ============================================================
class Test线段分析层次为零(unittest.TestCase):
"""验证 线段分析层次=0 时,各处理方法不会越界崩溃."""
@classmethod
def setUpClass(cls):
if not _has_nb():
raise unittest.SkipTest("需要 .nb 数据文件")
def test_处理数据_不崩溃(self):
"""投喂K线时 线段分析层次=0 → 跳过所有线段/扩展线段/混合扩展线段分析,不应崩溃."""
import chanlun
from chanlun import chan
# — Rust 侧 —
cfg_rs = chanlun.缠论配置()
obs_rs = chanlun.观察者("btcusd", 300, cfg_rs)
obs_rs.线段分析层次 = 0
obs_rs.重置基础序列()
for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:500]:
obs_rs.投喂原始数据(ts, o, h, l, c, v)
self.assertGreater(len(obs_rs.缠论K线序列), 0, "缠K序列应有数据")
self.assertGreater(len(obs_rs.分型序列), 0, "分型序列应有数据")
self.assertEqual(len(obs_rs.线段序列组), 0, "线段序列组应为空")
self.assertTrue(all(len(s) == 0 for s in obs_rs.混合扩展线段序列组), "混合扩展线段序列组所有条目应为空")
# — Python 侧 (chan.py) —
cfg_py = chan.缠论配置()
obs_py = chan.观察者("btcusd", 300, cfg_py)
obs_py.线段分析层次 = 0
obs_py.重置基础序列()
for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:500]:
obs_py.投喂原始数据(ts, o, h, l, c, v)
self.assertGreater(len(obs_py.缠论K线序列), 0, "chan.py 缠K序列应有数据")
self.assertGreater(len(obs_py.分型序列), 0, "chan.py 分型序列应有数据")
self.assertEqual(len(obs_py.线段序列组), 0, "chan.py 线段序列组应为空")
def test_静态重新分析_不崩溃(self):
"""静态重新分析时 线段分析层次=0 → 跳过所有线段分析,不应崩溃."""
import chanlun
from chanlun import chan
# — Rust 侧 —
cfg_rs = chanlun.缠论配置()
obs_rs = chanlun.观察者("btcusd", 300, cfg_rs)
for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:300]:
obs_rs.投喂原始数据(ts, o, h, l, c, v)
self.assertGreater(len(obs_rs.线段序列组), 0, "正常初始化后应有线段")
obs_rs.线段分析层次 = 0
obs_rs.静态重新分析()
self.assertEqual(len(obs_rs.线段序列组), 0, "静态重新分析后线段序列组应为空")
self.assertGreater(len(obs_rs.分型序列), 0, "静态重新分析后分型序列应有数据")
# — Python 侧 (chan.py) —
cfg_py = chan.缠论配置()
obs_py = chan.观察者("btcusd", 300, cfg_py)
for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:300]:
obs_py.投喂原始数据(ts, o, h, l, c, v)
self.assertGreater(len(obs_py.线段序列组), 0, "chan.py 正常初始化后应有线段")
obs_py.线段分析层次 = 0
obs_py.静态重新分析()
self.assertEqual(len(obs_py.线段序列组), 0, "chan.py 静态重新分析后线段序列组应为空")
self.assertGreater(len(obs_py.分型序列), 0, "chan.py 静态重新分析后分型序列应有数据")
# ============================================================
# 集成对比测试
# ============================================================
@@ -2151,5 +2228,553 @@ 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)
class Test缠论配置双端一致(unittest.TestCase):
"""缠论配置 to_dict / from_dict / model_copy 双端输出一致."""
def _make_configs(self):
import chanlun
from chanlun import chan
cfg_rs = chanlun.缠论配置()
cfg_py = chan.缠论配置()
return cfg_rs, cfg_py
def test_to_dict_keys_一致(self):
"""to_dict 字段名集合双端一致."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
d_rs = cfg_rs.to_dict()
d_py = cfg_py.to_dict()
self.assertEqual(set(d_rs.keys()), set(d_py.keys()), f"to_dict 字段不一致: R extra={set(d_rs.keys()) - set(d_py.keys())} P extra={set(d_py.keys()) - set(d_rs.keys())}")
def test_to_dict_values_一致(self):
"""to_dict 值双端一致."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
d_rs = cfg_rs.to_dict()
d_py = cfg_py.to_dict()
mismatches = []
for k in d_rs:
v_rs = d_rs[k]
v_py = d_py.get(k)
if v_rs is None and v_py is None:
continue
if v_rs != v_py:
mismatches.append(f" {k}: R={v_rs!r} P={v_py!r}")
self.assertEqual(len(mismatches), 0, f"to_dict 值不一致 ({len(mismatches)}处):\n" + "\n".join(mismatches[:10]))
def test_to_json_content_一致(self):
"""to_json 内容一致(JSON 解析后对比)."""
import chanlun
from chanlun import chan
import json
cfg_rs, cfg_py = self._make_configs()
j_rs = json.loads(cfg_rs.to_json())
j_py = json.loads(cfg_py.to_json())
self.assertEqual(j_rs, j_py, f"to_json 内容不一致")
def test_from_dict_roundtrip_一致(self):
"""from_dict → to_dict 往返双端一致."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
d = cfg_rs.to_dict()
cfg2_rs = chanlun.缠论配置.from_dict(d)
cfg2_py = chan.缠论配置.from_dict(d)
d2_rs = cfg2_rs.to_dict()
d2_py = cfg2_py.to_dict()
for k in d2_rs:
self.assertEqual(d2_rs[k], d2_py.get(k), f"from_dict 往返不一致: {k}")
def test_from_json_roundtrip_一致(self):
"""from_json → to_dict 往返双端一致."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
j = cfg_rs.to_json()
cfg2_rs = chanlun.缠论配置.from_json(j)
cfg2_py = chan.缠论配置.from_json(j)
# 验证标识和关键字段一致
self.assertEqual(cfg2_rs.标识, cfg2_py.标识)
self.assertEqual(cfg2_rs.笔内元素数量, cfg2_py.笔内元素数量)
self.assertEqual(cfg2_rs.买卖点偏移, cfg2_py.买卖点偏移)
self.assertEqual(cfg2_rs.指标计算方式, cfg2_py.指标计算方式)
def test_custom_values_from_dict_一致(self):
"""自定义字段 from_dict 双端一致."""
import chanlun
from chanlun import chan
data = {
"标识": "custom_test",
"缠K合并替换": True,
"笔内元素数量": 8,
"笔弱化": True,
"计算指标": False,
"指标计算方式": "高低均值",
"平滑异同移动平均线_快线周期": 12,
"买卖点偏移": 3,
"买卖点激进识别": True,
}
cfg_rs = chanlun.缠论配置.from_dict(data)
cfg_py = chan.缠论配置.from_dict(data)
d_rs = cfg_rs.to_dict()
d_py = cfg_py.to_dict()
for k in data:
self.assertEqual(d_rs.get(k), d_py.get(k), f"自定义字段 {k}: R={d_rs.get(k)} P={d_py.get(k)}")
def test_model_copy_一致(self):
"""model_copy 双端输出一致."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
update = {"标识": "copied", "推送K线": False, "笔内元素数量": 10}
copy_rs = cfg_rs.model_copy(update)
copy_py = cfg_py.model_copy(update)
self.assertEqual(copy_rs.标识, copy_py.标识)
self.assertEqual(copy_rs.笔内元素数量, copy_py.笔内元素数量)
self.assertFalse(copy_rs.推送K线)
self.assertFalse(copy_py.推送K线)
# 未更新字段保持原值一致
self.assertEqual(copy_rs.买卖点偏移, copy_py.买卖点偏移)
def test_from_dict_过滤未知字段_一致(self):
"""from_dict 过滤未知字段(兼容旧版本配置)双端一致."""
import chanlun
from chanlun import chan
data = {"标识": "test", "笔内元素数量": 7, "废弃字段_已删除": 999, "另一个旧字段": "xxx"}
cfg_rs = chanlun.缠论配置.from_dict(data)
cfg_py = chan.缠论配置.from_dict(data)
self.assertEqual(cfg_rs.标识, cfg_py.标识)
self.assertEqual(cfg_rs.笔内元素数量, cfg_py.笔内元素数量)
# 未知字段应被忽略,不影响构造
d_rs = cfg_rs.to_dict()
self.assertNotIn("废弃字段_已删除", d_rs)
def test_不推送_一致(self):
"""不推送 静态方法双端一致."""
import chanlun
from chanlun import chan
cfg_rs = chanlun.缠论配置.不推送()
cfg_py = chan.缠论配置.不推送()
self.assertFalse(cfg_rs.推送K线)
self.assertFalse(cfg_py.推送K线)
self.assertFalse(cfg_rs.图表展示)
self.assertFalse(cfg_py.图表展示)
self.assertEqual(cfg_rs.笔内元素数量, cfg_py.笔内元素数量)
def test_对比_默认一致(self):
"""默认配置 self 对比应无差异."""
import chanlun
from chanlun import chan
cfg_rs_a, _ = self._make_configs()
cfg_rs_b = chanlun.缠论配置()
diff_rs = cfg_rs_a.对比(cfg_rs_b)
self.assertIsInstance(diff_rs, dict)
self.assertEqual(len(diff_rs), 0, "默认一致配置不应有差异")
# Python side
cfg_py_a = chan.缠论配置()
cfg_py_b = chan.缠论配置()
diff_py = cfg_py_a.对比(cfg_py_b)
self.assertEqual(len(diff_py), 0)
def test_对比_有差异字段一致(self):
"""修改字段后 对比 双端输出一致."""
import chanlun
from chanlun import chan
cfg_rs_a, cfg_py_a = self._make_configs()
# 构造有差异的配置
update = {"标识": "changed", "笔内元素数量": 99, "推送K线": False}
cfg_rs_b = cfg_rs_a.model_copy(update)
cfg_py_b = cfg_py_a.model_copy(update)
diff_rs = cfg_rs_a.对比(cfg_rs_b)
diff_py = cfg_py_a.对比(cfg_py_b)
self.assertEqual(set(diff_rs.keys()), set(diff_py.keys()), f"对比字段不一致: R={set(diff_rs.keys())} P={set(diff_py.keys())}")
for k in diff_rs:
self.assertEqual(diff_rs[k], diff_py[k], f"对比[{k}] 值不一致: R={diff_rs[k]!r} P={diff_py[k]!r}")
def test_对比_往返一致(self):
"""to_dict → from_dict → 对比 应无差异."""
import chanlun
from chanlun import chan
cfg_rs, _ = self._make_configs()
d = cfg_rs.to_dict()
cfg2_rs = chanlun.缠论配置.from_dict(d)
diff = cfg_rs.对比(cfg2_rs)
self.assertEqual(len(diff), 0, f"Rust往返后对比不应有差异: {diff}")
# Python side
cfg_py = chan.缠论配置()
d_py = cfg_py.to_dict()
cfg2_py = chan.缠论配置.from_dict(d_py)
diff_py = cfg_py.对比(cfg2_py)
self.assertEqual(len(diff_py), 0)
def test_对比_与chan输出一致(self):
"""对比 输出与 chan.对比 逐项一致."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
update = {"标识": "test_x", "缠K合并替换": True, "笔内元素数量": 7, "计算指标": False, "买卖点激进识别": True, "线段_修正": True}
alt_rs = cfg_rs.model_copy(update)
alt_py = cfg_py.model_copy(update)
# Rust binding: cfg_rs.对比(alt_rs)
diff_rs = cfg_rs.对比(alt_rs)
# chan.py: cfg_py.对比(alt_py)
diff_py = cfg_py.对比(alt_py)
self.assertEqual(diff_rs, diff_py, f"对比输出不一致:\n R={diff_rs}\n P={diff_py}")
def test_对比_only_model_fields(self):
"""对比 仅比较 model_fields 字段."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
update = {"标识": "only_test"}
alt_rs = cfg_rs.model_copy(update)
alt_py = cfg_py.model_copy(update)
diff_rs = cfg_rs.对比(alt_rs)
diff_py = cfg_py.对比(alt_py)
self.assertEqual(len(diff_rs), 1)
self.assertEqual(len(diff_py), 1)
self.assertIn("标识", diff_rs)
self.assertIn("标识", diff_py)
self.assertEqual(diff_rs["标识"], "only_test")
self.assertEqual(diff_py["标识"], "only_test")
def test_对比_不推送_一致(self):
"""不推送 配置与默认配置 对比 双端一致."""
import chanlun
from chanlun import chan
cfg_rs, cfg_py = self._make_configs()
muted_rs = chanlun.缠论配置.不推送()
muted_py = chan.缠论配置.不推送()
diff_rs = cfg_rs.对比(muted_rs)
diff_py = cfg_py.对比(muted_py)
self.assertEqual(set(diff_rs.keys()), set(diff_py.keys()))
# 不推送应关闭所有推送/图表字段
for k in diff_rs:
self.assertFalse(diff_rs[k], f"不推送差异字段 {k} 应为 False")
self.assertFalse(diff_py[k], f"不推送差异字段 {k} 应为 False")
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "chanlun"
version = "26.6.2"
version = "26.6.3"
edition = "2024"
license = "MIT"
description = "基于缠论(缠中说禅)理论的量化技术分析核心库,支持流式数据处理和多周期联立分析。"
+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
+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!("{标签}:所有周期观察者全量校验全部一致"))
}
}
+330 -113
View File
@@ -261,77 +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])
for i in 0..self.线 {
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,
);
}
}
}
}
@@ -365,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.线 {
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,
);
}
}
}
}
@@ -581,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)]
@@ -1135,4 +1282,74 @@ mod tests {
assert_eq!(, "ethusd");
assert_eq!(, 7200);
}
#[test]
fn test_处理数据_线段分析层次为零_不崩溃() {
let mut config = ::default();
config. = test_data_path();
let obs = ::new("btcusd".into(), 300, config);
let mut obs_w = obs.write().unwrap();
obs_w.线 = 0;
obs_w.();
drop(obs_w);
// 逐根投喂K线,不应因 线段分析层次=0 而 panic
let data = std::fs::read(test_data_path()).unwrap();
let size = 48;
for i in 0..(data.len() / size).min(500) {
let offset = i * size;
if let Some(k线) = K线::from_bytes(&data[offset..offset + size], 300, "btcusd") {
obs.write().unwrap().K线(k线);
}
}
let obs_r = obs.read().unwrap();
assert!(obs_r.K线序列.len() > 0, "缠K序列应有数据");
assert!(obs_r..len() > 0, "分型序列应有数据");
assert!(obs_r.线.is_empty(), "线段序列组应为空");
// 混合扩展线段序列组 有 3 个空 Vec(因为 混合扩展线段分析层次 仍是 3),
// 但所有条目应为空(min(3, 0) = 0,循环未执行)
assert!(
obs_r.线.iter().all(|s| s.is_empty()),
"混合扩展线段序列组所有条目应为空"
);
info!(
"线段分析层次=0 处理数据 OK: {} 缠K, {} 分型, {} 笔",
obs_r.K线序列.len(),
obs_r..len(),
obs_r..len()
);
}
#[test]
fn test_静态重新分析_线段分析层次为零_不崩溃() {
let mut config = ::default();
config. = test_data_path();
let obs = ::new("btcusd".into(), 300, config);
// 先正常投喂数据
obs.write()
.unwrap()
.(&test_data_path(), Default::default())
.unwrap();
// 设为0后执行静态重新分析,不应 panic
let mut obs_w = obs.write().unwrap();
obs_w.线 = 0;
obs_w.();
drop(obs_w);
let obs_r = obs.read().unwrap();
assert!(obs_r..len() > 0, "静态重新分析后分型序列应有数据");
assert!(obs_r.线.is_empty(), "线段序列组应为空");
assert!(
obs_r.线.iter().all(|s| s.is_empty()),
"混合扩展线段序列组所有条目应为空"
);
info!(
"线段分析层次=0 静态重新分析 OK: {} 分型, {} 笔",
obs_r..len(),
obs_r..len()
);
}
}
+167 -28
View File
@@ -24,6 +24,10 @@
use crate::kline::bar::K线;
use std::collections::HashMap;
use tracing;
/// 事件回调类型 — fn(信号类型, 标识, 周期, 完成K线)
type = Box<dyn Fn(String, String, i64, K线) + Send + Sync>;
/// K线合成器 — 将小周期K线合成为大周期K线
pub struct K线合成器 {
@@ -31,11 +35,17 @@ 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<>,
}
impl K线合成器 {
/// 创建K线合成器,按周期升序排列,初始化当前K线合成K线列表
pub fn new(: String, : Vec<i64>) -> Self {
/// 创建K线合成器 — 对应 Python K线合成器.__init__(标识, 周期组, 事件回调=None)
pub fn new(
: String, : Vec<i64>, : Option<>
) -> Self {
let mut = ;
.sort();
@@ -51,37 +61,30 @@ impl K线合成器 {
,
K线,
K线列表,
,
}
}
/// 设置事件回调 — 对应 Python `设置事件回调`
pub fn (&mut self, : ) {
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 +94,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 +112,7 @@ impl K线合成器 {
( / ) *
}
/// 创建新K线 — 对应 Python `_创建新K线`
fn _创建新K线(&self, : i64, : i64, K: &K线) -> K线 {
let = self
.K线列表
@@ -132,6 +134,7 @@ impl K线合成器 {
)
}
/// 更新K线 — 对应 Python `_更新K线`
fn _更新K线(K线: &mut K线, : &K线) {
K线. = K线..max(.);
K线. = K线..min(.);
@@ -139,9 +142,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 +159,142 @@ 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线信号`
/// 异常安全:若回调 panic,捕获并记录错误,不中断管线
fn _产生完成K线信号(&self, : i64, K线: K线) {
if let Some(ref cb) = self. {
let = self..clone();
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
cb("K线完成".into(), , , K线);
}));
if let Err(e) = result {
let msg = e
.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| e.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "未知错误".into());
tracing::error!("K线合成器 事件回调 异常: {}", msg);
}
}
}
/// 获取指定周期当前正在合成的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());
}
}
+308 -51
View File
@@ -23,12 +23,9 @@
*/
use serde::{Deserialize, Deserializer, Serialize};
use std::collections::HashMap;
use tracing::warn;
fn is_infinite_f64(v: &f64) -> bool {
v.is_infinite()
}
/// 缠论配置 —— 控制所有分析阶段的行为
///
/// 50+ 参数集中控制缠K合并、笔/线段划分、中枢识别、买卖点生成等所有阶段。
@@ -211,29 +208,6 @@ pub struct 缠论配置 {
pub _指标匹配_KDJ: bool,
/// 买卖点指标匹配 RSI
pub _指标匹配_RSI: bool,
/// 买卖点背离率阈值(Infinity 表示不使用)
#[serde(skip_serializing_if = "is_infinite_f64")]
pub _背离率: f64,
/// 买卖点 T2 回调阈值
pub _T2_回调阈值: f64,
/// 买卖点 T2S 最大层级
pub _T2S_最大层级: i64,
/// 买卖点峰值条件
pub _峰值条件: bool,
/// 买卖点计算方式(峰/谷等)
pub _计算方式: String,
/// 是否计算线段BSP1
pub _计算线段BSP1: bool,
/// 是否处理BSP2
pub _处理BSP2: bool,
/// 是否计算线段BSP3
pub _计算线段BSP3: bool,
/// 是否依赖T1买卖点
pub _依赖T1: bool,
/// 买卖点中枢来源(实/虚/合)
pub _中枢来源: String,
/// 买卖点调试输出
pub _调试输出: bool,
// ---- 背驰 ----
/// 线段内部背驰使用 MACD
@@ -382,17 +356,6 @@ impl Default for 缠论配置 {
_指标匹配_MACD: true,
_指标匹配_KDJ: true,
_指标匹配_RSI: true,
_背离率: f64::INFINITY,
_T2_回调阈值: 1.0,
_T2S_最大层级: 3,
_峰值条件: false,
_计算方式: "".into(),
_计算线段BSP1: true,
_处理BSP2: true,
_计算线段BSP3: true,
_依赖T1: true,
_中枢来源: "".into(),
_调试输出: false,
线_MACD: true,
线_斜率: true,
线_测度: true,
@@ -445,6 +408,147 @@ impl 缠论配置 {
vec![("boll".into(), self._周期, self._标准差倍数)]
}
/// 序列化为 JSON 字典(对应 Python to_dict,仅返回 model_fields 中的字段)
pub fn to_dict(&self) -> serde_json::Value {
let full = serde_json::to_value(self).unwrap_or_default();
let valid = Self::model_fields();
if let serde_json::Value::Object(map) = full {
let filtered: serde_json::Map<_, _> = map
.into_iter()
.filter(|(k, _)| valid.contains(&k.as_str()))
.collect();
serde_json::Value::Object(filtered)
} else {
full
}
}
/// 从 JSON 字典反序列化(对应 Python from_dict / 兼容旧版本配置)
pub fn from_dict(value: &serde_json::Value) -> Result<Self, serde_json::Error> {
if let serde_json::Value::Object(map) = value {
let valid_fields = Self::model_fields();
let cleaned: serde_json::Map<_, _> = map
.iter()
.filter(|(k, _)| valid_fields.contains(&k.as_str()))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
serde_json::from_value(serde_json::Value::Object(cleaned))
} else {
serde_json::from_value(value.clone())
}
}
/// 验证并修正字段值(对应 Python _validate_all_fields
pub fn _validate_all_fields(&mut self) {
const : &[&str] = &[
"",
"",
"",
"",
"高低均值",
"高低收均值",
"开高低收均值",
];
if !.contains(&self..as_str()) {
warn!(
"[指标计算方式] = {} 值不在允许范围内,使用默认值:收",
self.
);
self. = "".into();
}
}
/// 返回字段名列表(对应 Python model_fields().keys()
pub fn model_fields() -> &'static [&'static str] {
&[
"标识",
"缠K合并替换",
"笔内元素数量",
"笔内相同终点取舍",
"笔内起始分型包含整笔",
"笔内起始分型包含整笔_包括右",
"笔内原始K线包含整笔",
"笔次级成笔",
"笔弱化",
"笔弱化_原始数量",
"线段_非缺口下穿刺",
"线段_特征序列忽视老阴老阳",
"线段_缺口后紧急修正",
"线段_修正",
"线段内部中枢图显",
"扩展线段_当下分析",
"分析笔",
"分析线段",
"分析扩展线段",
"分析笔中枢",
"分析线段中枢",
"手动终止",
"计算指标",
"计算BOLL",
"指标计算方式",
"平滑异同移动平均线_快线周期",
"平滑异同移动平均线_慢线周期",
"平滑异同移动平均线_信号周期",
"相对强弱指数_周期",
"相对强弱指数_移动平均线周期",
"相对强弱指数_超买阈值",
"相对强弱指数_超卖阈值",
"随机指标_RSV周期",
"随机指标_K值平滑周期",
"随机指标_D值平滑周期",
"随机指标_超买阈值",
"随机指标_超卖阈值",
"布林带_周期",
"布林带_标准差倍数",
"MACD_参数列表",
"RSI_周期列表",
"KDJ_参数列表",
"BOLL_参数列表",
"均线_类型列表",
"均线_周期列表",
"图表展示",
"推送K线",
"推送笔",
"推送线段",
"推送中枢",
"图表展示_笔",
"图表展示_线段",
"图表展示_扩展线段",
"图表展示_扩展线段_线段",
"图表展示_线段_线段",
"图表展示_中枢_笔",
"图表展示_中枢_线段",
"图表展示_中枢_扩展线段",
"图表展示_中枢_扩展线段_线段",
"图表展示_中枢_线段_线段",
"图表展示_中枢_线段内部",
"买卖点偏移",
"买卖点激进识别",
"买卖点与MACD柱强相关",
"买卖点错过误差值",
"买卖点_指标模式",
"买卖点_指标匹配_MACD",
"买卖点_指标匹配_KDJ",
"买卖点_指标匹配_RSI",
"线段内部背驰_MACD",
"线段内部背驰_斜率",
"线段内部背驰_测度",
"线段内部背驰_模式",
"加载文件路径",
]
}
/// 深拷贝并更新指定字段(对应 Python model_copy(update={...}, deep=True)
pub fn model_copy(&self, update: &HashMap<String, serde_json::Value>) -> Self {
let mut value = serde_json::to_value(self).unwrap_or_default();
if let serde_json::Value::Object(ref mut map) = value {
for (k, v) in update {
map.insert(k.clone(), v.clone());
}
}
serde_json::from_value(value).unwrap_or_else(|_| self.clone())
}
/// 序列化为 JSON 字符串
pub fn to_json(&self) -> String {
serde_json::to_string_pretty(self).unwrap_or_default()
@@ -467,9 +571,10 @@ impl 缠论配置 {
Ok(config)
}
/// 返回一个关闭所有推送/显示的新配置
/// 返回一个关闭所有推送/显示的新配置(对应 Python 不推送)
pub fn (&self) -> Self {
Self {
线: false,
: false,
K线: false,
: false,
@@ -524,19 +629,21 @@ impl 缠论配置 {
result
}
/// 对比两个配置,返回差异字段
pub fn (&self, other: &Self) -> Vec<String> {
let mut diffs = Vec::new();
let self_json = serde_json::to_value(self).unwrap();
let other_json = serde_json::to_value(other).unwrap();
/// 对比两个配置,返回差异字段及新值(对应 Python 对比 → dict[字段名, 新值])
pub fn (&self, other: &Self) -> HashMap<String, serde_json::Value> {
let mut diffs = HashMap::new();
let self_dict = self.to_dict();
let other_dict = other.to_dict();
if let (serde_json::Value::Object(self_map), serde_json::Value::Object(other_map)) =
(&self_json, &other_json)
(&self_dict, &other_dict)
{
for (key, self_val) in self_map {
if let Some(other_val) = other_map.get(key)
&& self_val != other_val
for key in Self::model_fields() {
let self_val = self_map.get(*key);
let other_val = other_map.get(*key);
if self_val != other_val
&& let Some(v) = other_val
{
diffs.push(key.clone());
diffs.insert(key.to_string(), v.clone());
}
}
}
@@ -562,7 +669,6 @@ mod tests {
let config = ::default();
assert_eq!(config., "bar");
assert_eq!(config., 5);
assert!(config._背离率.is_infinite());
assert_eq!(config., "");
}
@@ -609,6 +715,66 @@ mod tests {
assert_eq!(config.线_模式, "全量");
}
#[test]
fn test_to_dict_roundtrip() {
let config = ::default();
let dict = config.to_dict();
let restored = ::from_dict(&dict).unwrap();
assert_eq!(config.to_json(), restored.to_json());
}
#[test]
fn test_from_dict_filters_unknown_fields() {
// 兼容旧版本配置 — unknown fields are silently dropped
let json = serde_json::json!({
"标识": "test",
"不存在的字段": 42,
"另一个废弃字段": "xxx",
"笔内元素数量": 8,
});
let config = ::from_dict(&json).unwrap();
assert_eq!(config., "test");
assert_eq!(config., 8);
// 未指定字段使用默认值
assert_eq!(config., 1);
}
#[test]
fn test_model_fields_contains_all() {
let fields = ::model_fields();
assert!(fields.contains(&"标识"));
assert!(fields.contains(&"笔内元素数量"));
assert!(fields.contains(&"买卖点偏移"));
assert!(fields.contains(&"线段内部背驰_MACD"));
}
#[test]
fn test_model_copy() {
let mut update = std::collections::HashMap::new();
update.insert("标识".into(), serde_json::json!("custom"));
update.insert("推送K线".into(), serde_json::json!(false));
update.insert("笔内元素数量".into(), serde_json::json!(10));
let config = ::default();
let copied = config.model_copy(&update);
assert_eq!(copied., "custom");
assert!(!copied.K线);
assert_eq!(copied., 10);
// 未指定字段保持不变
assert_eq!(copied., 1);
assert!(copied.);
}
#[test]
fn test_to_dict_to_json_consistency() {
let config = ::default();
let dict = config.to_dict();
// to_dict → from_dict → to_json should equal original to_json
let restored = ::from_dict(&dict).unwrap();
assert_eq!(config.to_json(), restored.to_json());
}
#[test]
fn test_不推送() {
let config = ::default();
@@ -616,7 +782,98 @@ mod tests {
assert!(!muted.K线);
assert!(!muted.);
assert!(!muted.);
// 其他字段不变
assert_eq!(muted., 5);
}
#[test]
fn test_对比_无差异() {
let a = ::default();
let b = ::default();
let diff = a.(&b);
assert!(diff.is_empty(), "identical configs should have empty diff");
}
#[test]
fn test_对比_有差异() {
let mut a = ::default();
let mut b = ::default();
b. = "changed".into();
b. = 99;
let diff = a.(&b);
assert_eq!(diff.len(), 2);
assert_eq!(diff.get("标识").unwrap().as_str().unwrap(), "changed");
assert_eq!(diff.get("笔内元素数量").unwrap().as_i64().unwrap(), 99);
}
#[test]
fn test_对比_仅比较model_fields() {
// 仅比较 model_fields 中的字段(Python 一致行为)
let a = ::default();
let b = ::default();
let diff = a.(&b);
// 验证不包含废弃字段(如已删除的 "买卖点_背离率" 等)
assert!(!diff.contains_key("买卖点_背离率"));
assert!(diff.is_empty(), "default configs should have no diff");
}
#[test]
fn test_to_dict_excludes_non_model_fields() {
let config = ::default();
let dict = config.to_dict();
let valid = ::model_fields();
if let serde_json::Value::Object(map) = &dict {
for key in map.keys() {
assert!(
valid.contains(&key.as_str()),
"{key} should not be in to_dict output"
);
}
}
assert_eq!(
valid.len(),
dict.as_object().map(|m| m.len()).unwrap_or(0),
"to_dict should have exactly model_fields count"
);
}
#[test]
fn test_model_copy_then_对比() {
let config = ::default();
let mut update = HashMap::new();
update.insert("标识".into(), serde_json::json!("copied"));
update.insert("笔内元素数量".into(), serde_json::json!(10));
let copied = config.model_copy(&update);
let diff = config.(&copied);
assert_eq!(diff.len(), 2);
assert_eq!(diff["标识"].as_str().unwrap(), "copied");
assert_eq!(diff["笔内元素数量"].as_i64().unwrap(), 10);
}
#[test]
fn test_对比_boolean_difference() {
let mut a = ::default();
let mut b = ::default();
b.K线 = false;
b. = false;
let diff = a.(&b);
assert_eq!(diff.len(), 2);
assert_eq!(diff["推送K线"], serde_json::json!(false));
assert_eq!(diff["图表展示"], serde_json::json!(false));
}
#[test]
fn test_to_dict_from_dict_对比_roundtrip() {
let config = ::default();
let dict = config.to_dict();
let restored = ::from_dict(&dict).unwrap();
let diff = config.(&restored);
assert!(
diff.is_empty(),
"to_dict→from_dict roundtrip should produce no diff"
);
}
}