完善 测试类

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
@@ -2202,6 +2204,9 @@ class 缠论K线(object):
         return 序列[序列.index(始) : 序列.index(终) + 1]
 
 
+分型模式 = True
+
+
 class 分型(object):
     """分型 — 由左中右三根缠论K线构成的顶/底分型结构。
 
@@ -2216,7 +2221,7 @@ class 分型(object):
     :ivar 与MACD柱子分型匹配: 是否与MACD柱子分型匹配
     """
 
-    __slots__ = ["左", "中", "右", "结构", "时间戳", "分型特征值"]
+    __slots__ = ["左", "中", "右", "_结构", "_时间戳", "_分型特征值"]
 
     def __init__(self, 左: Optional[缠论K线], 中: 缠论K线, 右: Optional[缠论K线]):
         """
@@ -2229,9 +2234,9 @@ class 分型(object):
         self.左: Optional[缠论K线] = 左
         self.中: 缠论K线 = 中
         self.右: Optional[缠论K线] = 右
-        self.结构 = 中.分型
-        self.时间戳 = 中.时间戳
-        self.分型特征值 = 中.分型特征值
+        self._结构 = 中.分型
+        self._时间戳 = 中.时间戳
+        self._分型特征值 = 中.分型特征值
 
     def __str__(self):
         return f"{self.中.分型}<{self.时间戳}, {self.分型特征值:g}, None: {self.左 is None}, None: {self.右 is None}>"
@@ -2239,6 +2244,24 @@ class 分型(object):
     def __repr__(self):
         return f"{self.中.分型}<{self.时间戳}, {self.分型特征值:g}, None: {self.左 is None}, None: {self.右 is None}>"
 
+    @property
+    def 时间戳(self):
+        if 分型模式:
+            return self._时间戳
+        return self.中.时间戳
+
+    @property
+    def 分型特征值(self):
+        if 分型模式:
+            return self._分型特征值
+        return self.中.分型特征值
+
+    @property
+    def 结构(self):
+        if 分型模式:
+            return self._结构
+        return self.中.分型
+
     @property
     def 关系组(self) -> Optional[Tuple[相对方向, 相对方向, 相对方向]]:
         """左、中、右三对相对方向关系
@@ -5462,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.86"
+version = "26.5.94"
 edition = "2021"
 description = "缠论技术分析库 — Rust 高性能 Python 绑定"
 authors = ["YuYuKunKun"]
@@ -12,7 +12,7 @@ crate-type = ["cdylib"]
 name = "chanlun"
 
 [dependencies]
-chanlun = "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.86"
+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/src/algorithm_py.rs

@@ -445,6 +445,7 @@ impl 笔Py {
                 &mut bi_seq,
                 &ck_list,
                 &bar_list,
+                递归层次,
                 &config,
             )),
             None => Ok(递归层次),
@@ -480,7 +481,7 @@ impl 笔Py {
             .map(|k| k.bind(py).borrow().inner.clone())
             .collect();
         let config = 配置.borrow().to_rust_config(py)?;
-        Ok(chanlun::algorithm::bi::笔::分析递归(
+        Ok(chanlun::algorithm::bi::笔::分析(
             Arc::clone(&当前分型.borrow().inner),
             &mut fr_seq,
             &mut bi_seq,
@@ -749,9 +750,11 @@ impl 线段Py {
     )> {
         let borrowed = 段.borrow();
         let (a, b, c, d) = if let Some(hub_bound) = 所属中枢 {
-            if let Ok(mut hub_ref) = hub_bound.extract::<PyRefMut<'_, 中枢Py>>() {
-                let inner_mut = Arc::make_mut(&mut hub_ref.inner);
-                chanlun::algorithm::segment::线段::分割序列(&borrowed.inner, Some(inner_mut))
+            if let Ok(hub_ref) = hub_bound.extract::<PyRef<'_, 中枢Py>>() {
+                chanlun::algorithm::segment::线段::分割序列(
+                    &borrowed.inner,
+                    Some(&hub_ref.inner),
+                )
             } else {
                 chanlun::algorithm::segment::线段::分割序列(&borrowed.inner, None)
             }

chanlun-py/src/business_py.rs

@@ -48,7 +48,12 @@ use crate::types_py::买卖点类型Py;
 ///   有效性: bool — 买卖点是否仍有效
 ///   与MACD柱子匹配: bool|None — 是否与MACD柱状图方向匹配
 ///   与MACD柱子分型匹配: bool|None — 是否与MACD柱分型匹配
-#[pyclass(name = "基础买卖点", module = "chanlun._chanlun", from_py_object)]
+#[pyclass(
+    name = "基础买卖点",
+    module = "chanlun._chanlun",
+    subclass,
+    from_py_object
+)]
 #[derive(Clone)]
 pub struct 基础买卖点Py {
     pub(crate) inner: chanlun::business::bsp::基础买卖点,
@@ -71,6 +76,7 @@ impl 基础买卖点Py {
                 Arc::clone(&买卖点分型.borrow().inner),
                 备注,
                 中枢破位值,
+                当前K线.borrow().inner.序号,
             ),
         }
     }
@@ -199,19 +205,17 @@ impl 基础买卖点Py {
 
 // ========== 买卖点 ==========
 
-/// 买卖点 — 静态方法容器，提供各类买卖点的构造算法（不存储数据）。
+/// 买卖点 — 继承 基础买卖点，添加工厂类方法。
 ///
-/// 类方法（均返回对应的买卖点对象）:
+/// 类方法（均返回 买卖点 实例）:
 ///   一卖点(...) / 一买点(...) / 二卖点(...) / 二买点(...) / 三卖点(...) / 三买点(...)
-///   生成买卖点(特征, 序号, 级别, 分型, 当前缠K, 备注?) -> 买卖点
-///      — 根据特征字符串自动路由到对应的一/二/三类买卖点构造函数
-#[pyclass(name = "买卖点", module = "chanlun._chanlun")]
+///   生成买卖点(特征, 序号, 级别, 分型, 当前缠K) -> 买卖点
+#[pyclass(name = "买卖点", module = "chanlun._chanlun", extends=基础买卖点Py)]
 pub struct 买卖点Py;
 
 #[pymethods]
 impl 买卖点Py {
     #[classmethod]
-    /// :param 买卖点分型: 买卖点对应的分型
     fn 一卖点(
         _cls: &Bound<'_, PyType>,
         买卖点分型: &Bound<'_, 分型Py>,
@@ -219,20 +223,23 @@ impl 买卖点Py {
         标识: &str,
         备注: String,
         中枢破位值: f64,
-    ) -> 基础买卖点Py {
-        基础买卖点Py {
+        py: Python<'_>,
+    ) -> PyResult<Py<Self>> {
+        let base = 基础买卖点Py {
             inner: chanlun::business::bsp::买卖点::一卖点(
                 Arc::clone(&买卖点分型.borrow().inner),
                 当前K线.borrow().inner.clone(),
                 标识,
                 备注,
                 中枢破位值,
+                当前K线.borrow().inner.序号,
             ),
-        }
+        };
+        let init = PyClassInitializer::from(base).add_subclass(买卖点Py);
+        Ok(Bound::new(py, init)?.unbind())
     }
 
     #[classmethod]
-    /// :param 买卖点分型: 买卖点对应的分型
     fn 一买点(
         _cls: &Bound<'_, PyType>,
         买卖点分型: &Bound<'_, 分型Py>,
@@ -240,20 +247,23 @@ impl 买卖点Py {
         标识: &str,
         备注: String,
         中枢破位值: f64,
-    ) -> 基础买卖点Py {
-        基础买卖点Py {
+        py: Python<'_>,
+    ) -> PyResult<Py<Self>> {
+        let base = 基础买卖点Py {
             inner: chanlun::business::bsp::买卖点::一买点(
                 Arc::clone(&买卖点分型.borrow().inner),
                 当前K线.borrow().inner.clone(),
                 标识,
                 备注,
                 中枢破位值,
+                当前K线.borrow().inner.序号,
             ),
-        }
+        };
+        let init = PyClassInitializer::from(base).add_subclass(买卖点Py);
+        Ok(Bound::new(py, init)?.unbind())
     }
 
     #[classmethod]
-    /// :param 买卖点分型: 买卖点对应的分型
     fn 二卖点(
         _cls: &Bound<'_, PyType>,
         买卖点分型: &Bound<'_, 分型Py>,
@@ -261,20 +271,23 @@ impl 买卖点Py {
         标识: &str,
         备注: String,
         中枢破位值: f64,
-    ) -> 基础买卖点Py {
-        基础买卖点Py {
+        py: Python<'_>,
+    ) -> PyResult<Py<Self>> {
+        let base = 基础买卖点Py {
             inner: chanlun::business::bsp::买卖点::二卖点(
                 Arc::clone(&买卖点分型.borrow().inner),
                 当前K线.borrow().inner.clone(),
                 标识,
                 备注,
                 中枢破位值,
+                当前K线.borrow().inner.序号,
             ),
-        }
+        };
+        let init = PyClassInitializer::from(base).add_subclass(买卖点Py);
+        Ok(Bound::new(py, init)?.unbind())
     }
 
     #[classmethod]
-    /// :param 买卖点分型: 买卖点对应的分型
     fn 二买点(
         _cls: &Bound<'_, PyType>,
         买卖点分型: &Bound<'_, 分型Py>,
@@ -282,20 +295,23 @@ impl 买卖点Py {
         标识: &str,
         备注: String,
         中枢破位值: f64,
-    ) -> 基础买卖点Py {
-        基础买卖点Py {
+        py: Python<'_>,
+    ) -> PyResult<Py<Self>> {
+        let base = 基础买卖点Py {
             inner: chanlun::business::bsp::买卖点::二买点(
                 Arc::clone(&买卖点分型.borrow().inner),
                 当前K线.borrow().inner.clone(),
                 标识,
                 备注,
                 中枢破位值,
+                当前K线.borrow().inner.序号,
             ),
-        }
+        };
+        let init = PyClassInitializer::from(base).add_subclass(买卖点Py);
+        Ok(Bound::new(py, init)?.unbind())
     }
 
     #[classmethod]
-    /// :param 买卖点分型: 买卖点对应的分型
     fn 三卖点(
         _cls: &Bound<'_, PyType>,
         买卖点分型: &Bound<'_, 分型Py>,
@@ -303,20 +319,23 @@ impl 买卖点Py {
         标识: &str,
         备注: String,
         中枢破位值: f64,
-    ) -> 基础买卖点Py {
-        基础买卖点Py {
+        py: Python<'_>,
+    ) -> PyResult<Py<Self>> {
+        let base = 基础买卖点Py {
             inner: chanlun::business::bsp::买卖点::三卖点(
                 Arc::clone(&买卖点分型.borrow().inner),
                 当前K线.borrow().inner.clone(),
                 标识,
                 备注,
                 中枢破位值,
+                当前K线.borrow().inner.序号,
             ),
-        }
+        };
+        let init = PyClassInitializer::from(base).add_subclass(买卖点Py);
+        Ok(Bound::new(py, init)?.unbind())
     }
 
     #[classmethod]
-    /// :param 买卖点分型: 买卖点对应的分型
     fn 三买点(
         _cls: &Bound<'_, PyType>,
         买卖点分型: &Bound<'_, 分型Py>,
@@ -324,20 +343,24 @@ impl 买卖点Py {
         标识: &str,
         备注: String,
         中枢破位值: f64,
-    ) -> 基础买卖点Py {
-        基础买卖点Py {
+        py: Python<'_>,
+    ) -> PyResult<Py<Self>> {
+        let base = 基础买卖点Py {
             inner: chanlun::business::bsp::买卖点::三买点(
                 Arc::clone(&买卖点分型.borrow().inner),
                 当前K线.borrow().inner.clone(),
                 标识,
                 备注,
                 中枢破位值,
+                当前K线.borrow().inner.序号,
             ),
-        }
+        };
+        let init = PyClassInitializer::from(base).add_subclass(买卖点Py);
+        Ok(Bound::new(py, init)?.unbind())
     }
 
     #[classmethod]
-    /// :param 特征: 特征字符串
+    #[pyo3(signature = (特征, 序号, 级别, 买卖点分型, 当前缠K))]
     fn 生成买卖点(
         _cls: &Bound<'_, PyType>,
         特征: &str,
@@ -345,8 +368,9 @@ impl 买卖点Py {
         级别: &str,
         买卖点分型: &Bound<'_, 分型Py>,
         当前缠K: &Bound<'_, 缠论K线Py>,
-    ) -> 基础买卖点Py {
-        基础买卖点Py {
+        py: Python<'_>,
+    ) -> PyResult<Py<Self>> {
+        let base = 基础买卖点Py {
             inner: chanlun::business::bsp::买卖点::生成买卖点(
                 特征,
                 序号,
@@ -354,7 +378,9 @@ impl 买卖点Py {
                 Arc::clone(&买卖点分型.borrow().inner),
                 Arc::clone(&当前缠K.borrow().inner),
             ),
-        }
+        };
+        let init = PyClassInitializer::from(base).add_subclass(买卖点Py);
+        Ok(Bound::new(py, init)?.unbind())
     }
 }
 

chanlun-py/src/kline_py.rs

@@ -450,12 +450,12 @@ impl 缠论K线Py {
     }
 
     #[getter]
-    fn 分型(&self) -> Option<crate::types_py::分型结构Py> {
+    fn 分型(&self, py: Python<'_>) -> Option<Py<crate::types_py::分型结构Py>> {
         self.inner
             .分型
             .read()
             .unwrap()
-            .map(|f| crate::types_py::分型结构Py { inner: f })
+            .map(|f| crate::types_py::获取分型结构单例(py, f))
     }
 
     #[getter]
@@ -506,7 +506,7 @@ impl 缠论K线Py {
         dict.set_item("与RSI匹配", self.与RSI匹配())?;
         dict.set_item("与KDJ匹配", self.与KDJ匹配())?;
 
-        if let Some(v) = self.分型() {
+        if let Some(v) = self.分型(py) {
             dict.set_item("分型", v)?;
         }
         Ok(dict.into())

chanlun-py/src/lib.rs

@@ -23,6 +23,7 @@
  */
 
 use pyo3::prelude::*;
+use std::sync::atomic::Ordering;
 
 mod algorithm_py;
 mod business_py;
@@ -32,10 +33,24 @@ mod kline_py;
 mod structure_py;
 mod types_py;
 
+/// 分型模式 — True 时使用构造时缓存值，False 时从 中 缠K 实时读取
+#[pyfunction]
+fn get_分型模式() -> bool {
+    chanlun::structure::fractal_obj::分型模式.load(Ordering::Relaxed)
+}
+
+/// 设置 分型模式
+#[pyfunction]
+fn set_分型模式(value: bool) {
+    chanlun::structure::fractal_obj::分型模式.store(value, Ordering::Relaxed);
+}
+
 /// 缠论技术分析库 — Rust 高性能实现
 #[pymodule]
 /// 缠论技术分析库 — Rust 高性能实现
 fn _chanlun(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_function(wrap_pyfunction!(get_分型模式, m)?)?;
+    m.add_function(wrap_pyfunction!(set_分型模式, m)?)?;
     // 阶段 1: 枚举和基础类型
     types_py::register(m)?;
     // 阶段 2: 配置

chanlun-py/src/structure_py.rs

@@ -184,20 +184,32 @@ impl 分型Py {
     }
 
     #[getter]
-    fn 结构(&self) -> 分型结构Py {
-        分型结构Py {
-            inner: self.inner.结构,
-        }
+    fn 结构(&self, py: Python<'_>) -> Py<分型结构Py> {
+        let inner = if chanlun::structure::fractal_obj::分型模式.load(Ordering::Relaxed) {
+            self.inner.结构
+        } else {
+            self.inner
+                .中
+                .分型
+                .read()
+                .unwrap()
+                .unwrap_or(chanlun::types::分型结构::散)
+        };
+        crate::types_py::获取分型结构单例(py, inner)
     }
 
     #[getter]
     fn 时间戳(&self) -> i64 {
-        self.inner.时间戳
+        self.inner.时间戳()
     }
 
     #[getter]
     fn 分型特征值(&self) -> f64 {
-        self.inner.分型特征值
+        if chanlun::structure::fractal_obj::分型模式.load(Ordering::Relaxed) {
+            self.inner.分型特征值
+        } else {
+            self.inner.中.分型特征值.get()
+        }
     }
 
     fn __str__(&self) -> String {
@@ -247,7 +259,7 @@ impl 分型Py {
     #[getter]
     fn __dict__(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
         let dict = PyDict::new(py);
-        dict.set_item("结构", self.结构())?;
+        dict.set_item("结构", self.结构(py))?;
         dict.set_item("时间戳", self.时间戳())?;
         dict.set_item("分型特征值", self.分型特征值())?;
         dict.set_item("强度", self.强度())?;
@@ -1264,17 +1276,15 @@ impl 特征分型Py {
     }
 
     #[getter]
-    fn 结构(&self) -> 分型结构Py {
-        分型结构Py {
-            inner: self.inner.结构,
-        }
+    fn 结构(&self, py: Python<'_>) -> Py<分型结构Py> {
+        crate::types_py::获取分型结构单例(py, self.inner.结构)
     }
 
     /// pandas 兼容 — 返回关键标量字段构成的字典
     #[getter]
     fn __dict__(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
         let dict = PyDict::new(py);
-        dict.set_item("结构", self.结构())?;
+        dict.set_item("结构", self.结构(py))?;
         Ok(dict.into())
     }
 

chanlun-py/src/types_py.rs

@@ -22,10 +22,45 @@
  * SOFTWARE.
  */
 
+use std::collections::HashMap;
+use std::sync::Mutex;
+
 use pyo3::basic::CompareOp;
 use pyo3::prelude::*;
 use pyo3::types::{PyDict, PyType};
 
+// ========== 单例缓存 ==========
+
+static 分型结构_单例缓存: Mutex<Option<HashMap<u8, Py<分型结构Py>>>> = Mutex::new(None);
+
+pub fn 获取分型结构单例(
+    py: Python<'_>,
+    inner: chanlun::types::分型结构,
+) -> Py<分型结构Py> {
+    let mut guard = 分型结构_单例缓存.lock().unwrap();
+    if let Some(ref map) = *guard {
+        return map[&(inner as u8)].clone_ref(py);
+    }
+
+    // 首次访问时从类属性加载单例
+    let module = py.import("chanlun._chanlun").unwrap();
+    let class = module.getattr("分型结构").unwrap();
+    let mut map = HashMap::new();
+    for (name, variant) in &[
+        ("上", chanlun::types::分型结构::上),
+        ("下", chanlun::types::分型结构::下),
+        ("顶", chanlun::types::分型结构::顶),
+        ("底", chanlun::types::分型结构::底),
+        ("散", chanlun::types::分型结构::散),
+    ] {
+        let instance: Py<分型结构Py> = class.getattr(*name).unwrap().extract().unwrap();
+        map.insert(*variant as u8, instance);
+    }
+    let result = map[&(inner as u8)].clone_ref(py);
+    *guard = Some(map);
+    result
+}
+
 // ========== 买卖点类型 ==========
 
 /// 买卖点类型 — 缠论的三类买卖点及扩展类型。

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"

chanlun/src/algorithm/bi.rs

@@ -336,7 +336,7 @@ impl 笔 {
     /// 核心笔分析 — 使用显式栈模拟递归
     ///
     /// 返回: 递归层次数
-    pub fn 分析(
+    pub fn 分析_显式栈(
         初始分型: Arc<分型>,
         分型序列: &mut Vec<Arc<分型>>,
         笔序列: &mut Vec<Arc<虚线>>,
@@ -415,7 +415,8 @@ impl 笔 {
             let 之前分型 = Arc::clone(分型序列.last().unwrap());
 
             // Python line 2330-2335: 清理无效数据
-            if 之前分型.时间戳 == 当前分型.时间戳
+            if 之前分型.中.时间戳.load(Ordering::Relaxed)
+                == 当前分型.中.时间戳.load(Ordering::Relaxed)
                 || matches!(之前分型.结构, 分型结构::上 | 分型结构::下)
             {
                 Self::弹出旧笔(分型序列, 笔序列);
@@ -430,7 +431,8 @@ impl 笔 {
             let 之前分型 = Arc::clone(分型序列.last().unwrap());
 
             // Python line 2338: 时序检查 — skip out-of-order fractals
-            if 之前分型.时间戳 > 当前分型.时间戳
+            if 之前分型.中.时间戳.load(Ordering::Relaxed)
+                > 当前分型.中.时间戳.load(Ordering::Relaxed)
                 && 之前分型.中.序号.load(Ordering::Relaxed)
                     - 当前分型.中.序号.load(Ordering::Relaxed)
                     > 1
@@ -608,7 +610,7 @@ impl 笔 {
     /// 核心笔分析 — 递归实现，逐句对照 chan.py 笔.分析 / 笔递归分析
     ///
     /// 返回: 递归层次数
-    pub fn 分析递归(
+    pub fn 分析(
         当前分型: Arc<分型>,
         分型序列: &mut Vec<Arc<分型>>,
         笔序列: &mut Vec<Arc<虚线>>,
@@ -637,7 +639,7 @@ impl 笔 {
 
         // Python line 2329-2335: 清理无效数据
         let 之前分型 = Arc::clone(分型序列.last().unwrap());
-        if 之前分型.时间戳 == 当前分型.时间戳
+        if 之前分型.中.时间戳.load(Ordering::Relaxed) == 当前分型.中.时间戳.load(Ordering::Relaxed)
             || matches!(之前分型.结构, 分型结构::上 | 分型结构::下)
         {
             Self::弹出旧笔(分型序列, 笔序列);
@@ -651,7 +653,7 @@ impl 笔 {
 
         // Python line 2337-2341: 时序检查
         let 之前分型 = Arc::clone(分型序列.last().unwrap());
-        if 之前分型.时间戳 > 当前分型.时间戳
+        if 之前分型.中.时间戳.load(Ordering::Relaxed) > 当前分型.中.时间戳.load(Ordering::Relaxed)
             && 之前分型.中.序号.load(Ordering::Relaxed) - 当前分型.中.序号.load(Ordering::Relaxed)
                 > 1
         {
@@ -674,7 +676,7 @@ impl 笔 {
                         && 当前分型.结构 == 分型结构::顶);
                 if 破位 {
                     Self::弹出旧笔(分型序列, 笔序列);
-                    return Self::分析递归(
+                    return Self::分析(
                         当前分型,
                         分型序列,
                         笔序列,
@@ -712,7 +714,7 @@ impl 笔 {
                             if let Some(临时分型) =
                                 分型::从缠K序列中获取分型(缠K序列, 文官_k)
                             {
-                                let 递归层次 = Self::分析递归(
+                                let 递归层次 = Self::分析(
                                     Arc::new(临时分型),
                                     分型序列,
                                     笔序列,
@@ -721,7 +723,7 @@ impl 笔 {
                                     递归层次 + 1,
                                     配置,
                                 );
-                                return Self::分析递归(
+                                return Self::分析(
                                     当前分型,
                                     分型序列,
                                     笔序列,
@@ -770,7 +772,7 @@ impl 笔 {
                     if let Some(ref 右) = 当前分型.右 {
                         if let Some(临时分型) = 分型::从缠K序列中获取分型(缠K序列, 右)
                         {
-                            return Self::分析递归(
+                            return Self::分析(
                                 Arc::new(临时分型),
                                 分型序列,
                                 笔序列,
@@ -812,7 +814,7 @@ impl 笔 {
                             let 临时分型_rc = Arc::new(临时分型);
 
                             if !分型序列.is_empty() {
-                                let mut 递归层次 = Self::分析递归(
+                                let mut 递归层次 = Self::分析(
                                     Arc::clone(&临时分型_rc),
                                     分型序列,
                                     笔序列,
@@ -841,7 +843,7 @@ impl 笔 {
                                                     )
                                                 {
                                                     let 错过分型_rc = Arc::new(错过分型);
-                                                    递归层次 = Self::分析递归(
+                                                    递归层次 = Self::分析(
                                                         Arc::clone(&错过分型_rc),
                                                         分型序列,
                                                         笔序列,
@@ -856,7 +858,7 @@ impl 笔 {
                                     }
                                 }
 
-                                return Self::分析递归(
+                                return Self::分析(
                                     当前分型,
                                     分型序列,
                                     笔序列,
@@ -873,7 +875,7 @@ impl 笔 {
                 } else if 分型序列.is_empty() {
                     分型::向序列中添加(分型序列, 当前分型);
                 } else {
-                    return Self::分析递归(
+                    return Self::分析(
                         当前分型,
                         分型序列,
                         笔序列,

chanlun/src/algorithm/divergence.rs

@@ -69,14 +69,14 @@ impl 背驰分析 {
 
     /// 斜率背驰 — 价格斜率背驰
     pub fn 斜率背驰(进入段: &虚线, 离开段: &虚线) -> bool {
-        let dx = (进入段.武.read().unwrap().时间戳 - 进入段.文.时间戳) as f64;
+        let dx = (进入段.武.read().unwrap().时间戳() - 进入段.文.时间戳()) as f64;
         if dx == 0.0 {
             return false;
         }
         let dy = 进入段.武.read().unwrap().分型特征值 - 进入段.文.分型特征值;
         let 进入斜率 = dy / dx;
 
-        let dx = (离开段.武.read().unwrap().时间戳 - 离开段.文.时间戳) as f64;
+        let dx = (离开段.武.read().unwrap().时间戳() - 离开段.文.时间戳()) as f64;
         if dx == 0.0 {
             return false;
         }
@@ -92,11 +92,11 @@ impl 背驰分析 {
 
     /// 测度背驰 — 价格时间测度背驰
     pub fn 测度背驰(进入段: &虚线, 离开段: &虚线) -> bool {
-        let dx = (进入段.武.read().unwrap().时间戳 - 进入段.文.时间戳) as f64;
+        let dx = (进入段.武.read().unwrap().时间戳() - 进入段.文.时间戳()) as f64;
         let dy = 进入段.武.read().unwrap().分型特征值 - 进入段.文.分型特征值;
         let 进入测度 = (dx * dx + dy * dy).sqrt();
 
-        let dx = (离开段.武.read().unwrap().时间戳 - 离开段.文.时间戳) as f64;
+        let dx = (离开段.武.read().unwrap().时间戳() - 离开段.文.时间戳()) as f64;
         let dy = 离开段.武.read().unwrap().分型特征值 - 离开段.文.分型特征值;
         let 离开测度 = (dx * dx + dy * dy).sqrt();
 

chanlun/src/algorithm/hub.rs

@@ -165,9 +165,9 @@ impl 中枢 {
             self.标识.read().unwrap(),
             self.序号.load(Ordering::Relaxed),
             self.级别.load(Ordering::Relaxed),
-            self.文().时间戳,
+            self.文().时间戳(),
             crate::utils::format_f64_g(self.文().分型特征值),
-            self.武().时间戳,
+            self.武().时间戳(),
             crate::utils::format_f64_g(self.武().分型特征值),
             第三买卖线_str,
             本级_第三买卖线_str,
@@ -387,7 +387,7 @@ impl 中枢 {
 
     /// 向中枢序列尾部添加
     pub fn 向中枢序列尾部添加(
-        中枢序列: &mut Vec<Arc<中枢>>, mut 待添加中枢: Arc<中枢>
+        中枢序列: &mut Vec<Arc<中枢>>, 待添加中枢: Arc<中枢>
     ) {
         if let Some(前一个) = 中枢序列.last() {
             待添加中枢

chanlun/src/algorithm/segment.rs

@@ -106,7 +106,7 @@ impl 线段 {
             return;
         }
         if 段.武.read().unwrap().分型特征值 == 武.分型特征值
-            && 段.武.read().unwrap().时间戳 != 武.时间戳
+            && 段.武.read().unwrap().时间戳() != 武.时间戳()
         {
             eprintln!(
                 "线段.武斗[{}], 发现特征值相等但时间戳不同 {} {}",
@@ -474,7 +474,7 @@ impl 线段 {
     /// 分割序列 — 将线段的基础序列分为前、后、第三买卖线、贯穿伤
     pub fn 分割序列(
         段: &虚线,
-        mut 所属中枢: Option<&mut 中枢>,
+        所属中枢: Option<&中枢>,
     ) -> (
         Vec<Arc<虚线>>,
         Vec<Arc<虚线>>,
@@ -517,7 +517,7 @@ impl 线段 {
 
         let mut 状态 = None;
 
-        if let Some(ref mut 中枢) = 所属中枢 {
+        if let Some(ref 中枢) = 所属中枢 {
             *中枢.本级_第三买卖线.write().unwrap() = None;
             let 尾部 = if let Some(ref 后笔) = 后.last() {
                 后笔.武.read().unwrap().clone()
@@ -548,8 +548,6 @@ impl 线段 {
                     } else {
                         break;
                     }
-                } else {
-                    break;
                 }
             }
         }
@@ -568,15 +566,13 @@ impl 线段 {
                     } else {
                         break;
                     }
-                } else {
-                    break;
                 }
             }
         }
 
         if !第三买卖线.is_empty() {
             第三买卖线.reverse();
-            if let Some(ref mut 中枢) = 所属中枢 {
+            if let Some(ref 中枢) = 所属中枢 {
                 *中枢.本级_第三买卖线.write().unwrap() = Some(Arc::clone(&第三买卖线[0]));
             }
         }

chanlun/src/business/bsp.rs

@@ -42,6 +42,8 @@ pub struct 基础买卖点 {
     pub 终结K线: Option<Arc<K线>>,
     pub 破位值: f64,
     pub 结构: Option<分型结构>,
+    /// 当前缠K的序号 — 与 Python 一致，偏移计算使用缠论K线序号而非标的K线序号
+    pub 当前序号: i64,
 }
 
 impl 基础买卖点 {
@@ -51,6 +53,7 @@ impl 基础买卖点 {
         买卖点分型: Arc<分型>,
         备注: String,
         中枢破位值: f64,
+        当前序号: i64,
     ) -> Self {
         let 买卖点K线 = Arc::clone(&买卖点分型.中);
         Self {
@@ -63,12 +66,13 @@ impl 基础买卖点 {
             终结K线: None,
             破位值: 中枢破位值,
             结构: None,
+            当前序号,
         }
     }
 
-    /// 偏移 — 当前K线与买卖点K线的序号差
+    /// 偏移 — 当前缠K序号与买卖点K线序号的差（对齐 Python，使用缠论K线序号）
     pub fn 偏移(&self) -> i64 {
-        self.当前K线.序号 - self.买卖点K线.序号.load(Ordering::Relaxed)
+        self.当前序号 - self.买卖点K线.序号.load(Ordering::Relaxed)
     }
 
     /// 失效偏移
@@ -118,8 +122,16 @@ impl 买卖点 {
         _标识: &str,
         备注: String,
         中枢破位值: f64,
+        当前序号: i64,
     ) -> 基础买卖点 {
-        基础买卖点::new(买卖点类型::一卖, 当前K线, 买卖点分型, 备注, 中枢破位值)
+        基础买卖点::new(
+            买卖点类型::一卖,
+            当前K线,
+            买卖点分型,
+            备注,
+            中枢破位值,
+            当前序号,
+        )
     }
 
     pub fn 一买点(
@@ -128,8 +140,16 @@ impl 买卖点 {
         _标识: &str,
         备注: String,
         中枢破位值: f64,
+        当前序号: i64,
     ) -> 基础买卖点 {
-        基础买卖点::new(买卖点类型::一买, 当前K线, 买卖点分型, 备注, 中枢破位值)
+        基础买卖点::new(
+            买卖点类型::一买,
+            当前K线,
+            买卖点分型,
+            备注,
+            中枢破位值,
+            当前序号,
+        )
     }
 
     pub fn 二卖点(
@@ -138,8 +158,16 @@ impl 买卖点 {
         _标识: &str,
         备注: String,
         中枢破位值: f64,
+        当前序号: i64,
     ) -> 基础买卖点 {
-        基础买卖点::new(买卖点类型::二卖, 当前K线, 买卖点分型, 备注, 中枢破位值)
+        基础买卖点::new(
+            买卖点类型::二卖,
+            当前K线,
+            买卖点分型,
+            备注,
+            中枢破位值,
+            当前序号,
+        )
     }
 
     pub fn 二买点(
@@ -148,8 +176,16 @@ impl 买卖点 {
         _标识: &str,
         备注: String,
         中枢破位值: f64,
+        当前序号: i64,
     ) -> 基础买卖点 {
-        基础买卖点::new(买卖点类型::二买, 当前K线, 买卖点分型, 备注, 中枢破位值)
+        基础买卖点::new(
+            买卖点类型::二买,
+            当前K线,
+            买卖点分型,
+            备注,
+            中枢破位值,
+            当前序号,
+        )
     }
 
     pub fn 三卖点(
@@ -158,8 +194,16 @@ impl 买卖点 {
         _标识: &str,
         备注: String,
         中枢破位值: f64,
+        当前序号: i64,
     ) -> 基础买卖点 {
-        基础买卖点::new(买卖点类型::三卖, 当前K线, 买卖点分型, 备注, 中枢破位值)
+        基础买卖点::new(
+            买卖点类型::三卖,
+            当前K线,
+            买卖点分型,
+            备注,
+            中枢破位值,
+            当前序号,
+        )
     }
 
     pub fn 三买点(
@@ -168,8 +212,16 @@ impl 买卖点 {
         _标识: &str,
         备注: String,
         中枢破位值: f64,
+        当前序号: i64,
     ) -> 基础买卖点 {
-        基础买卖点::new(买卖点类型::三买, 当前K线, 买卖点分型, 备注, 中枢破位值)
+        基础买卖点::new(
+            买卖点类型::三买,
+            当前K线,
+            买卖点分型,
+            备注,
+            中枢破位值,
+            当前序号,
+        )
     }
 
     /// 生成买卖点 — 根据参数自动选择类型
@@ -190,6 +242,8 @@ impl 买卖点 {
 
         // 当前K线 — 从缠K获取其标的K线
         let 当前K线 = Arc::clone(&*当前缠K.标的K线.read().unwrap());
+        // 当前序号 — 使用缠论K线序号（对齐 Python 偏移计算）
+        let 当前序号 = 当前缠K.序号.load(Ordering::Relaxed);
 
         let 类型 = match (序号, 买卖) {
             ("一", "买") => 买卖点类型::一买,
@@ -201,6 +255,6 @@ impl 买卖点 {
             _ => 买卖点类型::一买, // fallback
         };
 
-        基础买卖点::new(类型, 当前K线, 买卖点分型, 备注, 破位值)
+        基础买卖点::new(类型, 当前K线, 买卖点分型, 备注, 破位值, 当前序号)
     }
 }

chanlun/src/business/observer.rs

@@ -177,6 +177,7 @@ impl 观察者 {
                 &mut self.笔序列,
                 &self.缠论K线序列,
                 &self.普通K线序列,
+                0,
                 &self.配置,
             );
         }
@@ -289,6 +290,7 @@ impl 观察者 {
                 &mut self.笔序列,
                 &self.缠论K线序列,
                 &self.普通K线序列,
+                0,
                 &self.配置,
             );
         }
@@ -434,7 +436,7 @@ impl 观察者 {
                 format!(
                     "分型, {}, {}, {:?}, {}, {}, {}",
                     i,
-                    fx.时间戳,
+                    fx.时间戳(),
                     fx.结构,
                     fx.分型特征值,
                     fx.中.时间戳.load(Ordering::Relaxed),
@@ -640,7 +642,7 @@ mod tests {
             let 文_ptr = Arc::as_ptr(&bi.文);
             let 文_found = obs_ref.分型序列.iter().any(|f| Arc::as_ptr(f) == 文_ptr);
             if !文_found {
-                println!("笔 {}: 文(时间戳={}) 不在分型序列中!", i, bi.文.时间戳);
+                println!("笔 {}: 文(时间戳={}) 不在分型序列中!", i, bi.文.时间戳());
             }
 
             let 武_ptr = Arc::as_ptr(&*bi.武.read().unwrap());
@@ -649,7 +651,7 @@ mod tests {
                 println!(
                     "笔 {}: 武(时间戳={}) 不在分型序列中!",
                     i,
-                    bi.武.read().unwrap().时间戳
+                    bi.武.read().unwrap().时间戳()
                 );
             }
         }

chanlun/src/structure/dash_line.rs

@@ -239,9 +239,9 @@ impl 虚线 {
                 self.标识.read().unwrap(),
                 self.序号.load(Ordering::Relaxed),
                 self.级别.load(Ordering::Relaxed),
-                self.文.时间戳,
+                self.文.时间戳(),
                 format_f64_g(self.文.分型特征值),
-                self.武.read().unwrap().时间戳,
+                self.武.read().unwrap().时间戳(),
                 format_f64_g(self.武.read().unwrap().分型特征值),
                 if self.有效性.load(Ordering::Relaxed) {
                     "True"
@@ -330,9 +330,9 @@ impl 虚线 {
             self.标识.read().unwrap(),
             self.序号.load(Ordering::Relaxed),
             self.级别.load(Ordering::Relaxed),
-            self.文.时间戳,
+            self.文.时间戳(),
             format_f64_g(self.文.分型特征值),
-            self.武.read().unwrap().时间戳,
+            self.武.read().unwrap().时间戳(),
             format_f64_g(self.武.read().unwrap().分型特征值),
             if self.有效性.load(Ordering::Relaxed) { "True" } else { "False" },
             self.基础序列.read().unwrap().len(),
@@ -1232,7 +1232,7 @@ mod tests {
         assert_eq!(Arc::as_ptr(&笔.文), 文_ptr_before);
 
         // 但方向变了（因为武从底1变成底2）
-        let 新武耗时 = 笔.武.read().unwrap().时间戳;
+        let 新武耗时 = 笔.武.read().unwrap().时间戳();
         assert_eq!(新武耗时, 300);
     }
 }

chanlun/src/structure/fractal_obj.rs

@@ -25,9 +25,13 @@
 use crate::kline::chan_kline::缠论K线;
 use crate::types::分型结构;
 use crate::types::相对方向;
+use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
+/// 分型模式 — True 时使用构造时缓存值（默认），False 时从 中 缠K 实时读取
+pub static 分型模式: AtomicBool = AtomicBool::new(true);
+
 /// 分型 — 由三根缠K构成（可能缺左或右）
 #[derive(Debug, Clone)]
 pub struct 分型 {
@@ -56,6 +60,15 @@ impl 分型 {
         }
     }
 
+    /// 时间戳 — 根据 分型模式 决定返回缓存值（True）或实时值（False）
+    pub fn 时间戳(&self) -> i64 {
+        if 分型模式.load(Ordering::Relaxed) {
+            self.时间戳
+        } else {
+            self.中.时间戳.load(Ordering::Relaxed)
+        }
+    }
+
     /// 左中右三组关系
     pub fn 关系组(&self) -> Option<(相对方向, 相对方向, 相对方向)> {
         let 左 = self.左.as_ref()?;
@@ -215,7 +228,7 @@ impl std::fmt::Display for 分型 {
                 .read()
                 .unwrap()
                 .unwrap_or(crate::types::分型结构::散),
-            self.时间戳,
+            self.时间戳(),
             crate::utils::format_f64_g(self.分型特征值),
             if self.左.is_none() { "True" } else { "False" },
             if self.右.is_none() { "True" } else { "False" },

chanlun/src/structure/segment_feat.rs

@@ -63,7 +63,7 @@ impl 线段特征 {
                         .分型特征值
                         .partial_cmp(&b.文.分型特征值)
                         .unwrap_or(std::cmp::Ordering::Equal)
-                        .then_with(|| a.文.时间戳.cmp(&b.文.时间戳))
+                        .then_with(|| a.文.时间戳().cmp(&b.文.时间戳()))
                 })
                 .map(|x| Arc::clone(&x.文))
                 .unwrap_or_else(|| Arc::clone(&self.元素[0].文))
@@ -75,7 +75,7 @@ impl 线段特征 {
                         .分型特征值
                         .partial_cmp(&b.文.分型特征值)
                         .unwrap_or(std::cmp::Ordering::Equal)
-                        .then_with(|| b.文.时间戳.cmp(&a.文.时间戳))
+                        .then_with(|| b.文.时间戳().cmp(&a.文.时间戳()))
                 })
                 .map(|x| Arc::clone(&x.文))
                 .unwrap_or_else(|| Arc::clone(&self.元素[0].文))
@@ -99,8 +99,8 @@ impl 线段特征 {
                             a.武
                                 .read()
                                 .unwrap()
-                                .时间戳
-                                .cmp(&b.武.read().unwrap().时间戳)
+                                .时间戳()
+                                .cmp(&b.武.read().unwrap().时间戳())
                         })
                 })
                 .map(|x| x.武.read().unwrap().clone())
@@ -119,8 +119,8 @@ impl 线段特征 {
                             b.武
                                 .read()
                                 .unwrap()
-                                .时间戳
-                                .cmp(&a.武.read().unwrap().时间戳)
+                                .时间戳()
+                                .cmp(&a.武.read().unwrap().时间戳())
                         })
                 })
                 .map(|x| x.武.read().unwrap().clone())
@@ -527,7 +527,7 @@ mod tests {
 
         let 文 = feat.文();
         // 向上取最大特征值：都是100 → tiebreaker取后时间戳 → 笔2.文(300)
-        assert_eq!(文.时间戳, 300);
+        assert_eq!(文.时间戳(), 300);
     }
 
     #[test]
@@ -544,7 +544,7 @@ mod tests {
 
         let 武 = feat.武();
         // 向上取最大特征值：都是80 → tiebreaker取后时间戳 → 笔2.武(400)
-        assert_eq!(武.时间戳, 400);
+        assert_eq!(武.时间戳(), 400);
     }
 
     // ============================================================