完善 测试类

chan.py

@@ -28,8 +28,10 @@ SOFTWARE.
 
 import json
 import math
+import os
 import struct
 import sys
+import tempfile
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
@@ -5483,10 +5485,10 @@ class 立体分析器:
         if 当前K线 := self._K线合成器.获取当前K线(周期):
             self._单体分析器[周期].增加原始K线(当前K线)
 
-    def 测试_保存数据(self):
+    def 测试_保存数据(self, root: str = None):
         """拆分各序列数据，单独存文件，文件名为对应变量名"""
         # 生成存储根目录
-        脚本目录 = Path(__file__).parent  # 取当前脚本所在文件夹
+        脚本目录 = Path(__file__).parent if not root else root  # 取当前脚本所在文件夹
         起始时间 = int(self._单体分析器[self.__输入周期].普通K线序列[0].时间戳.timestamp())
         结束时间 = int(self._单体分析器[self.__输入周期].普通K线序列[-1].时间戳.timestamp())
         目录标识 = f"PyM_{self._单体分析器[self.__输入周期].标识}_{起始时间}_{结束时间}"

chanlun-py/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "chanlun-py"
-version = "26.5.93"
+version = "26.5.94"
 edition = "2021"
 description = "缠论技术分析库 — Rust 高性能 Python 绑定"
 authors = ["YuYuKunKun"]
@@ -12,7 +12,7 @@ crate-type = ["cdylib"]
 name = "chanlun"
 
 [dependencies]
-chanlun = { version = "26.5.3", path = "../chanlun" }
+chanlun = "26.5.4" # { path = "../chanlun" }
 pyo3 = { version = "0.28", features = ["extension-module", "experimental-inspect"] }
 serde_json = "1"
 chrono = "0.4"

chanlun-py/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "chanlun"
-version = "2605.93"
+version = "2605.94"
 description = "缠论技术分析库 — Rust 高性能实现"
 readme = { file = "README.md", content-type = "text/markdown" }
 license = { file = "LICENSE", content-type = "text/plain" }
@@ -37,3 +37,7 @@ features = ["pyo3/extension-module"]
 python-source = "."
 module-name = "chanlun._chanlun"
 manifest-path = "Cargo.toml"
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+addopts = ["-v", "--tb=short", "--durations=10"]

chanlun-py/tests/helpers/__init__.py

@@ -0,0 +1,25 @@
+"""pyo3_test_helpers — 可复用的 PyO3 测试工具包。
+
+提供四个核心模块:
+
+    rc_identity      — Rc/Arc 指针身份一致性测试 Mixin
+    subclass         — PyO3 #[pyclass(subclass)] 子类化兼容性测试 Mixin
+    type_shape       — 返回值类型形状验证工具
+    api_consistency  — 两个模块间 API 描述符类型一致性测试 Mixin
+
+所有 Mixin 都是纯 Python，不依赖 pytest，与 unittest.TestCase 配合使用。
+下游项目复制此目录即可复用。
+"""
+
+from .api_consistency import ApiConsistencyMixin
+from .rc_identity import RcIdentityMixin
+from .subclass import PyO3SubclassMixin
+from .type_shape import assert_type_shape, TypeShapeAssertions
+
+__all__ = [
+    "ApiConsistencyMixin",
+    "RcIdentityMixin",
+    "PyO3SubclassMixin",
+    "assert_type_shape",
+    "TypeShapeAssertions",
+]

chanlun-py/tests/helpers/api_consistency.py

@@ -0,0 +1,196 @@
+"""API 一致性测试 Mixin。
+
+验证两个模块中同名类的公开成员描述符类型一致。
+典型用途：对比 Python 参考实现 (chan.py) 与 Rust/PyO3 移植 (chanlun) 的 API 兼容性。
+
+用法::
+
+    class TestApi一致性(ApiConsistencyMixin, unittest.TestCase):
+        reference_module = mylib.ref      # Python 参考实现
+        target_module = mylib             # Rust/PyO3 移植
+
+        # 可选: 已知差异（不会报错）
+        known_missing_in_target = {
+            "SomeClass": {"old_deprecated_method"},
+        }
+        known_descriptor_diffs = {
+            # (class_name, member, ref_type, target_type)
+        }
+
+        # 可选: 成员名过滤（匹配则跳过，支持前缀用 "prefix_" 表示）
+        noise_filters = ["model_", "parse_", "from_orm"]
+"""
+
+import unittest
+
+
+def _classify_member(cls, attr_name):
+    """返回描述符类型: property / classmethod / staticmethod / regular_method / None(data)."""
+    # 优先检查元类字典中的描述符
+    for klass in type(cls).__mro__:
+        if attr_name in klass.__dict__:
+            raw = klass.__dict__[attr_name]
+            if isinstance(raw, property):
+                return "property"
+            elif isinstance(raw, classmethod):
+                return "classmethod"
+            elif isinstance(raw, staticmethod):
+                return "staticmethod"
+            break
+    try:
+        attr = getattr(cls, attr_name)
+    except Exception:
+        return None
+    if callable(attr):
+        return "regular_method"
+    return None
+
+
+def _is_noise(name, filters):
+    for pat in filters:
+        if pat == name:
+            return True
+        if pat.endswith("_") and name.startswith(pat):
+            return True
+    return False
+
+
+def _get_classes(mod):
+    """获取模块中所有公开的 type."""
+    return {n: getattr(mod, n) for n in dir(mod) if not n.startswith("_") and isinstance(getattr(mod, n), type)}
+
+
+class ApiConsistencyMixin:
+    """API 一致性测试 Mixin。
+
+    子类必须定义:
+        reference_module: 参考模块 (Python 实现)
+        target_module:   目标模块 (Rust/PyO3 移植)
+
+    子类可选定义:
+        known_missing_in_target: dict[str, set[str]]  — 已知 target 中缺失的成员
+        known_descriptor_diffs: set[tuple]             — 已知描述符类型差异
+        noise_filters: list[str]                       — 噪音成员名过滤
+    """
+
+    reference_module = None
+    target_module = None
+    known_missing_in_target: dict = {}
+    known_descriptor_diffs: set = set()
+    noise_filters: list = []
+
+    @classmethod
+    def setUpClass(cls):
+        if cls.reference_module is None or cls.target_module is None:
+            raise unittest.SkipTest(f"{cls.__name__} 未定义 reference_module / target_module")
+
+    # ---- 描述符类型一致性 ----
+
+    def test_共有成员描述符类型一致(self):
+        """同名类的同名成员，描述符类型 (property/classmethod/staticmethod/regular) 一致."""
+        ref_classes = _get_classes(self.reference_module)
+        tgt_classes = _get_classes(self.target_module)
+        shared = sorted(set(ref_classes) & set(tgt_classes))
+
+        failures = []
+        for cls_name in shared:
+            ref_cls = ref_classes[cls_name]
+            tgt_cls = tgt_classes[cls_name]
+
+            ref_members = {}
+            tgt_members = {}
+
+            for attr_name in sorted(dir(ref_cls)):
+                if attr_name.startswith("_") or _is_noise(attr_name, self.noise_filters):
+                    continue
+                cat = _classify_member(ref_cls, attr_name)
+                if cat:
+                    ref_members[attr_name] = cat
+
+            for attr_name in sorted(dir(tgt_cls)):
+                if attr_name.startswith("_") or _is_noise(attr_name, self.noise_filters):
+                    continue
+                cat = _classify_member(tgt_cls, attr_name)
+                if cat:
+                    tgt_members[attr_name] = cat
+
+            shared_members = sorted(set(ref_members) & set(tgt_members))
+            for member in shared_members:
+                ref_cat = ref_members[member]
+                tgt_cat = tgt_members[member]
+                if ref_cat != tgt_cat:
+                    diff_key = (cls_name, member, ref_cat, tgt_cat)
+                    if diff_key not in self.known_descriptor_diffs:
+                        failures.append(f"{cls_name}.{member}: ref={ref_cat}, tgt={tgt_cat}")
+
+        if failures:
+            self.fail("描述符类型不一致:\n  " + "\n  ".join(failures))
+
+    # ---- 缺失成员检查 ----
+
+    def test_参考模块成员在目标模块中存在(self):
+        """chan 中的关键公开成员在 chanlun 中均有对应."""
+        ref_classes = _get_classes(self.reference_module)
+        tgt_classes = _get_classes(self.target_module)
+        shared = sorted(set(ref_classes) & set(tgt_classes))
+
+        failures = []
+        for cls_name in shared:
+            if cls_name not in self.known_missing_in_target:
+                continue
+            ref_cls = ref_classes[cls_name]
+            tgt_cls = tgt_classes[cls_name]
+
+            expected_missing = self.known_missing_in_target.get(cls_name, set())
+
+            ref_members = set()
+            for attr_name in sorted(dir(ref_cls)):
+                if attr_name.startswith("_") or _is_noise(attr_name, self.noise_filters):
+                    continue
+                cat = _classify_member(ref_cls, attr_name)
+                if cat and attr_name not in expected_missing:
+                    ref_members.add(attr_name)
+
+            tgt_members = set()
+            for attr_name in sorted(dir(tgt_cls)):
+                if attr_name.startswith("_") or _is_noise(attr_name, self.noise_filters):
+                    continue
+                cat = _classify_member(tgt_cls, attr_name)
+                if cat:
+                    tgt_members.add(attr_name)
+
+            missing = ref_members - tgt_members - expected_missing
+            for member in sorted(missing):
+                failures.append(f"{cls_name}.{member}: ref={_classify_member(ref_cls, member)}, tgt=未导出")
+
+        if failures:
+            self.fail("参考模块中的成员在目标模块中缺失:\n  " + "\n  ".join(failures))
+
+    # ---- 方法可调用性 ----
+
+    def test_共有方法均可调用(self):
+        """所有共有 regular_method 在两边都是 callable."""
+        ref_classes = _get_classes(self.reference_module)
+        tgt_classes = _get_classes(self.target_module)
+        shared = sorted(set(ref_classes) & set(tgt_classes))
+
+        failures = []
+        for cls_name in shared:
+            ref_cls = ref_classes[cls_name]
+            tgt_cls = tgt_classes[cls_name]
+
+            for attr_name in sorted(dir(ref_cls)):
+                if attr_name.startswith("_") or _is_noise(attr_name, self.noise_filters):
+                    continue
+                ref_cat = _classify_member(ref_cls, attr_name)
+                tgt_cat = _classify_member(tgt_cls, attr_name)
+                if ref_cat == "regular_method" and tgt_cat == "regular_method":
+                    ref_obj = getattr(ref_cls, attr_name)
+                    tgt_obj = getattr(tgt_cls, attr_name)
+                    if not callable(ref_obj):
+                        failures.append(f"{cls_name}.{attr_name}: ref 不是 callable")
+                    if not callable(tgt_obj):
+                        failures.append(f"{cls_name}.{attr_name}: tgt 不是 callable")
+
+        if failures:
+            self.fail("方法不可调用:\n  " + "\n  ".join(failures))

chanlun-py/tests/helpers/rc_identity.py

@@ -0,0 +1,126 @@
+"""Rc/Arc 指针身份一致性测试 Mixin。
+
+验证：同一个 Rust Rc<T>/Arc<T> 无论通过哪条路径到达 Python，
+始终返回相同的 PyObject（`a is b` 为 True）。
+
+用法::
+
+    class TestMyLib(RcIdentityMixin, unittest.TestCase):
+        # 必须: 创建被测对象实例（每个 test_ 调用一次）
+        @staticmethod
+        def target_factory():
+            return make_fresh_instance()
+
+        # 必须: 序列 getter —— (名称, target → list)
+        # Mixin 会验证: 同一 getter 调用两次，list[i] is list[j]
+        sequence_getters = {
+            "主序列": lambda t: t.items,
+            "子序列": lambda t: t.children,
+        }
+
+        # 可选: 跨路径身份断言 —— (名称, (target → obj_a, target → obj_b))
+        # Mixin 会验证: obj_a is obj_b
+        cross_path_assertions = [
+            ("序列[0] 与 首元素.父", lambda t: t.items[0], lambda t: t.items[0].parent),
+        ]
+
+        # 可选: getter 稳定性 —— (名称, target → obj)
+        # Mixin 会验证: obj is obj (两次调用返回同一对象)
+        stable_getters = {
+            "首元素.属性": lambda t: t.items[0].attr,
+        }
+
+        # 可选: 序列长度检查的最小值（默认不检查，设为 >0 开启）
+        min_sequence_lengths = {
+            "主序列": 3,
+            "子序列": 2,
+        }
+"""
+
+import unittest
+
+
+class RcIdentityMixin:
+    """Rc/Arc 指针身份一致性测试 Mixin。
+
+    子类必须定义:
+        target_factory: Callable[[], Any]
+        sequence_getters: dict[str, Callable[[Any], list]]
+
+    子类可选定义:
+        cross_path_assertions: list[tuple[str, Callable, Callable]]
+        stable_getters: dict[str, Callable]
+        min_sequence_lengths: dict[str, int]
+    """
+
+    target_factory = None
+    sequence_getters: dict = {}
+    cross_path_assertions: list = []
+    stable_getters: dict = {}
+    min_sequence_lengths: dict = {}
+
+    def _get_target(self):
+        """惰性获取 target，首次调用后缓存在类上。避免 setUpClass MRO 冲突."""
+        cls = type(self)
+        # 每次测试重新创建——但这会太慢。用类级别缓存。
+        # 子类应在 setUpClass 中调用 self._get_target() 或自己设置 cls._cached_target。
+        if not hasattr(cls, "_cached_target"):
+            if cls.target_factory is None:
+                raise unittest.SkipTest(f"{cls.__name__} 未定义 target_factory")
+            cls._cached_target = cls.target_factory()
+        return cls._cached_target
+
+    # ---- 序列 getter 稳定性 ----
+
+    def test_序列重复获取身份一致(self):
+        """同一序列 getter 调用两次，对应位置元素 is 相同."""
+        t = self._get_target()
+        for name, getter in self.sequence_getters.items():
+            seq1 = getter(t)
+            seq2 = getter(t)
+            self.assertEqual(len(seq1), len(seq2), f"{name}: 两次获取长度不同")
+            check_n = min(len(seq1), 10)
+            for i in range(check_n):
+                self.assertIs(seq1[i], seq2[i], f"{name}[{i}] 身份不一致")
+
+    def test_序列最小长度(self):
+        """序列长度至少达到配置的最小值."""
+        t = self._get_target()
+        for name, getter in self.sequence_getters.items():
+            if name in self.min_sequence_lengths:
+                min_len = self.min_sequence_lengths[name]
+                actual = len(getter(t))
+                self.assertGreaterEqual(actual, min_len, f"{name} 长度 {actual} < {min_len}")
+
+    # ---- 跨路径身份 ----
+
+    def test_跨路径身份一致(self):
+        """不同访问路径到达的同一 Rust 对象在 Python 侧 is 相同."""
+        t = self._get_target()
+        for i, (label, path_a, path_b) in enumerate(self.cross_path_assertions):
+            obj_a = path_a(t)
+            obj_b = path_b(t)
+            self.assertIsNotNone(obj_a, f"[{i}] {label}: path_a 返回 None")
+            self.assertIsNotNone(obj_b, f"[{i}] {label}: path_b 返回 None")
+            self.assertIs(obj_a, obj_b, f"[{i}] {label}: 身份不一致")
+
+    # ---- getter 稳定性 ----
+
+    def test_getter重复调用身份一致(self):
+        """同一 getter 调用两次返回同一 PyObject."""
+        t = self._get_target()
+        for name, getter in self.stable_getters.items():
+            obj1 = getter(t)
+            obj2 = getter(t)
+            self.assertIs(obj1, obj2, f"{name}: 两次调用返回不同对象")
+
+    # ---- list.index 基于 is ----
+
+    def test_list_index_基于身份(self):
+        """list.index(elem) 正常工作（依赖 __eq__ 基于 is 比较）."""
+        t = self._get_target()
+        for name, getter in self.sequence_getters.items():
+            seq = getter(t)
+            if len(seq) >= 2:
+                self.assertEqual(seq.index(seq[0]), 0, f"{name}: index(seq[0]) != 0")
+                self.assertEqual(seq.index(seq[-1]), len(seq) - 1, f"{name}: index(seq[-1]) != {len(seq) - 1}")

chanlun-py/tests/helpers/subclass.py

@@ -0,0 +1,247 @@
+"""PyO3 #[pyclass(subclass)] 子类化兼容性测试 Mixin。
+
+验证: Python 端可以正常子类化 PyO3 导出的类，__new__/__init__ 协作、
+super() 委托、MRO 链、property/method 重写等全部正确。
+
+用法::
+
+    class TestMyObserver(PyO3SubclassMixin, unittest.TestCase):
+        base_class = mylib.Observer
+        constructor_args = ("symbol", 300)
+        constructor_kwargs = {}
+
+        # 可选: 用 kwargs 的构造
+        constructor_with_config = ("symbol", 300, {"配置": mylib.Config()})
+
+        # 可选: 序列 getter 名称列表（重写测试会检查这些 getter 可被覆盖）
+        sequence_getter_names = [
+            "普通K线序列", "高级序列",
+        ]
+
+        # 可选: 需要 .nb 数据文件才能运行的测试会检查这个
+        @staticmethod
+        def has_data_file():
+            return os.path.isfile("data.nb")
+
+        # 可选: 创建一个"喂了一根K线"的 target
+        @staticmethod
+        def make_target_with_data():
+            obs = mylib.Observer("sym", 300)
+            k = mylib.KLine(...)
+            obs.feed(k)
+            return obs
+
+        # 可选: 创建一个"喂了一根K线"的子类实例
+        @staticmethod
+        def make_sub_with_data():
+            class Sub(mylib.Observer):
+                pass
+            obs = Sub("sym", 300)
+            k = mylib.KLine(...)
+            obs.feed(k)
+            return obs
+"""
+
+import unittest
+
+
+class PyO3SubclassMixin:
+    """PyO3 子类化兼容性测试 Mixin。
+
+    子类必须定义:
+        base_class: type
+        constructor_args: tuple
+        constructor_kwargs: dict
+
+    子类可选定义:
+        sequence_getter_names: list[str]
+        has_data_file: Callable[[], bool]
+        make_target_with_data: Callable[[], Any]
+        make_sub_with_data: Callable[[], Any]
+        make_data_item: Callable[[], Any]          # 创建一根可喂入的数据项
+        feed_method_name: str                       # 喂数据的方法名，默认 "增加原始K线"
+        property_getters: list[str]                 # 需要逐一下覆写的 property 名
+        method_overrides: list[str]                 # 需要逐一重写的方法名
+    """
+
+    base_class: type = None
+    constructor_args: tuple = ()
+    constructor_kwargs: dict = {}
+    sequence_getter_names: list = []
+
+    # 可选 hooks
+    has_data_file = None
+    make_target_with_data = None
+    make_sub_with_data = None
+    make_data_item = None
+    feed_method_name = "增加原始K线"
+    property_getters: list = []
+    method_overrides: list = []
+
+    @classmethod
+    def setUpClass(cls):
+        if cls.base_class is None:
+            raise unittest.SkipTest(f"{cls.__name__} 未定义 base_class")
+
+    # ---- 基础子类化 ----
+
+    def test_子类可实例化(self):
+        """子类可创建，isinstance 正确."""
+        Base = self.base_class
+
+        class Sub(Base):
+            pass
+
+        obs = Sub(*self.constructor_args, **self.constructor_kwargs)
+        self.assertIsInstance(obs, Base)
+        self.assertEqual(type(obs).__name__, "Sub")
+
+    def test_子类_init_可添加自定义属性(self):
+        """子类 __init__ 可添加自定义属性，基类字段不受影响."""
+        Base = self.base_class
+        args = self.constructor_args
+        kwargs = self.constructor_kwargs
+
+        class Sub(Base):
+            def __init__(self, *a, **kw):
+                self.tag = "custom"
+                self.count = 0
+
+        obs = Sub(*args, **kwargs)
+        self.assertEqual(obs.tag, "custom")
+        self.assertEqual(obs.count, 0)
+
+    def test_子类_new_过滤_kwargs(self):
+        """__new__ 过滤子类专属参数，只把父类需要的传给 super().__new__."""
+        Base = self.base_class
+        args = self.constructor_args
+
+        class Sub(Base):
+            def __new__(cls, *a, extra=None, **kw):
+                return super().__new__(cls, *a)
+
+            def __init__(self, *a, extra=None, **kw):
+                self.extra = extra
+
+        obs = Sub(*args, extra={"debug": True})
+        self.assertEqual(obs.extra, {"debug": True})
+
+        obs2 = Sub(*args)
+        self.assertIsNone(obs2.extra)
+
+    # ---- 方法重写 ----
+
+    def test_方法重写_super调用(self):
+        """重写方法，super() 调用父类."""
+        if self.make_target_with_data is None or self.make_sub_with_data is None:
+            self.skipTest("未定义 make_target_with_data / make_sub_with_data")
+
+        base_obs = self.make_target_with_data()
+        sub_obs = self.make_sub_with_data()
+
+        for attr in self.sequence_getter_names:
+            base_len = len(getattr(base_obs, attr))
+            sub_len = len(getattr(sub_obs, attr))
+            self.assertEqual(base_len, sub_len, f"{attr}: base={base_len}, sub={sub_len}")
+
+    def test_方法完全重写不调super(self):
+        """完全重写方法不调 super()，基类逻辑不执行."""
+        Base = self.base_class
+        args = self.constructor_args
+        kwargs = self.constructor_kwargs
+
+        class Sub(Base):
+            def __init__(self, *a, **kw):
+                self.log = []
+
+        obs = Sub(*args, **kwargs)
+        self.assertEqual(obs.log, [])
+
+    # ---- property 重写 ----
+
+    def test_property_重写_super调用(self):
+        """重写 @property getter，super() 取基类值."""
+        Base = self.base_class
+        args = self.constructor_args
+        kwargs = self.constructor_kwargs
+
+        class Sub(Base):
+            pass
+
+        obs = Sub(*args, **kwargs)
+        # 验证实例创建成功即可，具体 getter 覆盖由子类测试
+        self.assertIsInstance(obs, Base)
+
+    def test_str_repr_重写(self):
+        """重写 __str__ / __repr__."""
+        Base = self.base_class
+        args = self.constructor_args
+        kwargs = self.constructor_kwargs
+
+        class Sub(Base):
+            def __str__(self):
+                return f"Custom({id(self)})"
+
+            def __repr__(self):
+                return self.__str__()
+
+        obs = Sub(*args, **kwargs)
+        self.assertIn("Custom", str(obs))
+        self.assertEqual(str(obs), repr(obs))
+
+    # ---- 多层继承 MRO ----
+
+    def test_多层继承_MRO链完整(self):
+        """多层继承，MRO 调用链完整."""
+        if self.make_data_item is None:
+            self.skipTest("未定义 make_data_item")
+
+        Base = self.base_class
+        args = self.constructor_args
+        kwargs = self.constructor_kwargs
+        feed_name = self.feed_method_name
+
+        class L1(Base):
+            def __init__(self, *a, **kw):
+                self._l1_called = False
+
+        class L2(L1):
+            def __init__(self, *a, **kw):
+                super().__init__(*a, **kw)
+                self._l2_called = True
+
+        obs = L2(*args, **kwargs)
+        self.assertTrue(obs._l2_called)
+
+    def test_未重写方法直接继承(self):
+        """未重写的方法从基类直接继承."""
+        if self.make_data_item is None:
+            self.skipTest("未定义 make_data_item")
+
+        Base = self.base_class
+        args = self.constructor_args
+        kwargs = self.constructor_kwargs
+
+        class Sub(Base):
+            pass
+
+        obs = Sub(*args, **kwargs)
+        self.assertIsInstance(obs, Base)
+
+    # ---- 重写后实例行为与基类一致 ----
+
+    def test_同名继承行为一致(self):
+        """同名继承（零重写），行为与基类完全一致."""
+        if self.make_target_with_data is None:
+            self.skipTest("未定义 make_target_with_data")
+
+        Base = self.base_class
+        args = self.constructor_args
+        kwargs = self.constructor_kwargs
+
+        class Sub(Base):
+            pass
+
+        base = Base(*args, **kwargs)
+        sub = Sub(*args, **kwargs)
+        self.assertIsInstance(sub, Base)

chanlun-py/tests/helpers/type_shape.py

@@ -0,0 +1,88 @@
+"""PyO3 返回值的 Python 类型形状验证工具。
+
+验证 PyO3 导出的函数/方法返回值类型正确：
+- int 不是 str/float
+- list 元素是 tuple 不是 list
+- 方法是 callable 不是 property
+- 返回值结构（嵌套类型）符合预期
+
+用法::
+
+    from helpers.type_shape import assert_type_shape
+
+    result = mylib.compute(some_input)
+    assert_type_shape(result, {
+        "count": int,
+        "ratio": float,
+        "label": str,
+        "items": [(int, str, bool)],    # list of 3-tuples
+        "nested": {"key": int},
+    })
+"""
+
+import unittest
+
+
+def assert_type_shape(obj, schema, path=""):
+    """验证 obj 的类型形状与 schema 一致。
+
+    schema 支持:
+        - type: obj 必须是该类型实例
+        - [inner]: obj 必须是 list，每个元素验证 inner
+        - (t1, t2, ...): obj 必须是 tuple，每字段验证对应类型
+        - {key: inner}: obj 必须是 dict，递归验证
+        - callable: obj 必须是 callable（函数/方法）
+    """
+    if isinstance(schema, type):
+        _check_type(obj, schema, path)
+    elif isinstance(schema, list):
+        _check_list(obj, schema, path)
+    elif isinstance(schema, tuple):
+        _check_tuple(obj, schema, path)
+    elif isinstance(schema, dict):
+        _check_dict(obj, schema, path)
+    elif schema is callable:
+        _check_callable(obj, path)
+    else:
+        raise ValueError(f"{path}: 不支持的 schema 类型 {type(schema)}")
+
+
+def _check_type(obj, expected, path):
+    assert isinstance(obj, expected), f"{path}: 期望 {expected.__name__}, 实际 {type(obj).__name__}"
+
+
+def _check_list(obj, schema, path):
+    assert isinstance(obj, list), f"{path}: 期望 list, 实际 {type(obj).__name__}"
+    if len(schema) == 1:
+        inner = schema[0]
+        for i, item in enumerate(obj):
+            assert_type_shape(item, inner, f"{path}[{i}]")
+
+
+def _check_tuple(obj, schema, path):
+    assert isinstance(obj, tuple), f"{path}: 期望 tuple, 实际 {type(obj).__name__}"
+    assert len(obj) == len(schema), f"{path}: 期望 tuple 长度 {len(schema)}, 实际 {len(obj)}"
+    for i, (item, inner) in enumerate(zip(obj, schema)):
+        assert_type_shape(item, inner, f"{path}[{i}]")
+
+
+def _check_dict(obj, schema, path):
+    assert isinstance(obj, dict), f"{path}: 期望 dict, 实际 {type(obj).__name__}"
+    for key, inner in schema.items():
+        assert key in obj, f"{path}: 缺少键 '{key}'"
+        assert_type_shape(obj[key], inner, f"{path}['{key}']")
+
+
+def _check_callable(obj, path):
+    assert callable(obj), f"{path}: 期望 callable, 实际 {type(obj).__name__}"
+
+
+# ---- TestCase mixin ----
+
+
+class TypeShapeAssertions:
+    """提供 assert_type_shape 便捷方法的 mixin."""
+
+    def assertTypeShape(self, obj, schema, path=""):
+        """断言 obj 的类型形状与 schema 一致."""
+        assert_type_shape(obj, schema, path)

chanlun-py/tests/test_identity.py

@@ -1,267 +0,0 @@
-#!/usr/bin/env python3
-"""
-对象标识测试：验证同一 Rc 底层数据通过不同路径访问时，
-Python 侧始终返回相同的 PyObject（`is` 比较为 True）。
-
-涉及的 Rc 包装类型：
-  - K线 (Rc<K线>)           — 原始OHLCV数据
-  - 缠论K线 (Rc<缠论K线>)     — 包含处理后的K线
-  - 分型 (Rc<分型>)          — 顶底分型
-  - 虚线 (Rc<虚线>)          — 笔/线段的通用抽象
-  - 中枢 (Rc<中枢>)          — 三段虚线重叠区间
-  - 线段特征 (Rc<线段特征>)   — 线段特征序列元素
-  - 特征分型 (Rc<特征分型>)   — 特征序列的分型
-
-路径示例：
-  - 缠K序列[i]  vs  分型序列[j].中  (同一根缠K)
-  - 分型序列[i]  vs  笔序列[j].文    (同一个分型)
-  - 笔序列[i]    vs  中枢[k].基础序列[m]  (同一条虚线)
-"""
-
-import chanlun
-import math
-
-
-def create_observer(symbol="btcusd", period=14400, n_bars=500):
-    """创建观察者并喂入模拟K线数据。"""
-    cfg = chanlun.缠论配置()
-    obs = chanlun.观察者(symbol, period, cfg)
-
-    for i in range(n_bars):
-        trend = i * 3
-        wave = math.sin(i * 0.05) * 2000
-        mid = 68000.0 + trend + wave
-        high = mid + abs(math.cos(i * 0.3)) * 400 + 100
-        low = mid - abs(math.sin(i * 0.5)) * 400 - 100
-        k = chanlun.K线(
-            标识=symbol,
-            周期=period,
-            时间戳=1771675200 + i * period,
-            开盘价=mid - 50,
-            高=high,
-            低=low,
-            收盘价=mid + 50,
-            成交量=abs(math.sin(i)) * 1000,
-        )
-        obs.增加原始K线(k)
-
-    return obs
-
-
-class Test缠K身份:
-    """缠论K线: 从序列、分型、笔端点、中枢等不同路径访问。"""
-
-    def test_序列重复获取(self):
-        """同一序列获取两次，元素应相同。"""
-        obs = create_observer()
-        seq1 = obs.缠论K线序列
-        seq2 = obs.缠论K线序列
-        for i in range(min(len(seq1), 10)):
-            assert seq1[i] is seq2[i], f"缠K序列[{i}] 身份不一致"
-
-    def test_分型中K(self):
-        """分型.中 与 缠K序列 对应元素应相同。"""
-        obs = create_observer()
-        seq = obs.缠论K线序列
-        分序 = obs.分型序列
-        for fx in 分序[:10]:
-            中 = fx.中
-            for ck in seq:
-                if ck.时间戳 == 中.时间戳:
-                    assert ck is 中, f"分型.中 (ts={中.时间戳}) 与序列中元素不匹配"
-                    break
-
-    def test_笔端点钟K(self):
-        """笔的端点分型的中间K线应与序列元素相同。"""
-        obs = create_observer()
-        seq = obs.缠论K线序列
-        for bi in obs.笔序列:
-            for nm, getter in [("文", lambda b=bi: b.文), ("武", lambda b=bi: b.武)]:
-                ep = getter()
-                if ep is None:
-                    continue
-                中 = ep.中
-                for ck in seq:
-                    if ck.时间戳 == 中.时间戳:
-                        assert ck is 中, f"笔.{nm}.中 (ts={中.时间戳}) 与序列中元素不匹配"
-                        break
-
-    def test_getter重复调用(self):
-        """同一getter调用两次返回同一对象。"""
-        obs = create_observer()
-        for fx in obs.分型序列[:5]:
-            中1 = fx.中
-            中2 = fx.中
-            assert 中1 is 中2, "分型.中 两次调用返回不同对象"
-
-
-class Test分型身份:
-    """分型: 从分型序列、笔/线段端点、买卖点等不同路径访问。"""
-
-    def test_序列重复获取(self):
-        """同一序列获取两次，元素应相同。"""
-        obs = create_observer()
-        seq1 = obs.分型序列
-        seq2 = obs.分型序列
-        for i in range(min(len(seq1), 9)):
-            assert seq1[i] is seq2[i], f"分型序列[{i}] 身份不一致"
-
-    def test_笔端点与序列(self):
-        """笔.文 / 笔.武 应与分型序列中对应元素相同。"""
-        obs = create_observer()
-        分序 = obs.分型序列
-        for bi in obs.笔序列:
-            for nm in ["文", "武"]:
-                ep = getattr(bi, nm)
-                if ep is None:
-                    continue
-                matched = False
-                for fx in 分序:
-                    if fx.时间戳 == ep.时间戳 and fx.结构 == ep.结构:
-                        assert fx is ep, f"笔.{nm} (ts={ep.时间戳}) 与分型序列中元素不匹配"
-                        matched = True
-                        break
-                assert matched, f"笔.{nm} (ts={ep.时间戳}) 在分型序列中未找到"
-
-    def test_段端点与序列(self):
-        """段.文 / 段.武 应与分型序列中对应元素相同。"""
-        obs = create_observer()
-        分序 = obs.分型序列
-        for duan in obs.线段序列:
-            for nm in ["文", "武"]:
-                ep = getattr(duan, nm)
-                if ep is None:
-                    continue
-                matched = False
-                for fx in 分序:
-                    if fx.时间戳 == ep.时间戳 and fx.结构 == ep.结构:
-                        assert fx is ep, f"段.{nm} (ts={ep.时间戳}) 与分型序列中元素不匹配"
-                        matched = True
-                        break
-                assert matched, f"段.{nm} (ts={ep.时间戳}) 在分型序列中未找到"
-
-    def test_getter重复调用(self):
-        """同一getter调用两次返回同一对象。"""
-        obs = create_observer()
-        for bi in obs.笔序列:
-            文1 = bi.文
-            文2 = bi.文
-            assert 文1 is 文2, "笔.文 两次调用返回不同对象"
-            武1 = bi.武
-            武2 = bi.武
-            assert 武1 is 武2, "笔.武 两次调用返回不同对象"
-            break  # 只测第一笔
-
-
-class Test虚线身份:
-    """虚线(笔/线段): 从笔序列、线段序列、中枢内部序列等不同路径访问。"""
-
-    def test_笔序列重复获取(self):
-        obs = create_observer()
-        seq1 = obs.笔序列
-        seq2 = obs.笔序列
-        for i in range(min(len(seq1), 8)):
-            assert seq1[i] is seq2[i], f"笔序列[{i}] 身份不一致"
-
-    def test_线段序列重复获取(self):
-        obs = create_observer()
-        seq1 = obs.线段序列
-        seq2 = obs.线段序列
-        for i in range(min(len(seq1), 5)):
-            assert seq1[i] is seq2[i], f"线段序列[{i}] 身份不一致"
-
-    def test_多个扩展序列(self):
-        """扩展线段的不同序列获取同一虚线应相同。"""
-        obs = create_observer()
-        s1 = obs.扩展线段序列
-        s2 = obs.扩展线段序列_线段
-        s3 = obs.扩展线段序列_扩展线段
-        # 这些序列可能包含不同的虚线，但如果同一个 Rc 出现在两个序列中应该相同
-        for d1 in s1:
-            for d2 in s2:
-                if d1.序号 == d2.序号:
-                    assert d1 is d2, f"扩展线段序列[{d1.序号}] 跨序列身份不一致"
-                    break
-
-
-class TestK线身份:
-    """原始K线: 从序列、买卖点、缠K标的等不同路径访问。"""
-
-    def test_序列重复获取(self):
-        obs = create_observer()
-        seq1 = obs.普通K线序列
-        seq2 = obs.普通K线序列
-        for i in range(min(len(seq1), 10)):
-            assert seq1[i] is seq2[i], f"普K序列[{i}] 身份不一致"
-
-
-class Test中枢身份:
-    """中枢: 从中枢序列、分型关联、笔中枢/线段中枢等不同路径访问。"""
-
-    def test_序列重复获取(self):
-        obs = create_observer(period=3600, n_bars=800)
-        seq1 = obs.中枢序列
-        seq2 = obs.中枢序列
-        for i in range(min(len(seq1), 5)):
-            assert seq1[i] is seq2[i], f"中枢序列[{i}] 身份不一致"
-
-    def test_笔中枢与线段中枢(self):
-        obs = create_observer(period=3600, n_bars=800)
-        笔中 = obs.笔_中枢序列
-        段中 = obs.线段_中枢序列
-        扩展中 = obs.扩展中枢序列
-        # 验证同一次获取内的身份
-        for zs in 笔中:
-            文1 = zs.文
-            文2 = zs.文
-            assert 文1 is 文2, f"笔中枢.文 两次调用不同"
-            break
-        for zs in 段中:
-            文1 = zs.文
-            文2 = zs.文
-            assert 文1 is 文2, f"段中枢.文 两次调用不同"
-            break
-
-
-class Test整体身份:
-    """跨类型综合身份测试。"""
-
-    def test_买卖点分型(self):
-        """验证买卖点的关联分型身份。"""
-        obs = create_observer(period=3600, n_bars=800)
-        # 尝试访问可用的结构
-        分序 = obs.分型序列
-        笔序 = obs.笔序列
-        assert len(分序) >= 0 and len(笔序) >= 0  # 至少不崩溃
-
-    def test_全链路一致性(self):
-        """缠K → 分型 → 笔 → 段 链路中所有对象身份一致。"""
-        obs = create_observer()
-        seq = obs.缠论K线序列
-
-        for bi in obs.笔序列:
-            # 笔的端点分型
-            for nm, getter in [("文", lambda b=bi: b.文), ("武", lambda b=bi: b.武)]:
-                ep = getter()
-                if ep is None:
-                    continue
-                # ep 中的 中 是一根缠K，应能在序列中找到相同对象
-                中 = ep.中
-                for ck in seq:
-                    if ck.时间戳 == 中.时间戳:
-                        assert ck is 中
-                        break
-                # 左也应该是可访问的
-                左 = ep.左
-                if 左 is not None:
-                    for ck in seq:
-                        if ck.时间戳 == 左.时间戳:
-                            assert ck is 左
-                            break
-                # 右也应该是可访问的
-                右 = ep.右
-                if 右 is not None:
-                    for ck in seq:
-                        if ck.时间戳 == 右.时间戳:
-                            assert ck is 右
-                            break

chanlun/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "chanlun"
-version = "26.5.3"
+version = "26.5.4"
 edition = "2021"
 rust-version = "1.70"
 license = "MIT"