diff --git a/chan.py b/chan.py index 17d9736..a5c6cd0 100644 --- a/chan.py +++ b/chan.py @@ -53,7 +53,7 @@ from typing import ( ) from collections.abc import Sequence -from chanlun.termcolor import colored +from termcolor import colored from loguru import logger @@ -6383,7 +6383,7 @@ class 观察者: 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置) 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0) - for i in range(self.混合扩展线段分析层次): + for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))): 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置) 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0) @@ -6429,7 +6429,7 @@ class 观察者: for i in range(self.扩展线段分析层次): 保存序列(self.扩展线段序列组[i]) 保存序列(self.扩展中枢序列组[i]) - for i in range(self.线段分析层次): + for i in range(self.混合扩展线段分析层次): 保存序列(self.混合扩展线段序列组[i]) 保存序列(self.混合扩展中枢序列组[i]) @@ -6487,7 +6487,7 @@ class 观察者: 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置) 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0) - for i in range(self.混合扩展线段分析层次): + for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))): 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置) 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0) @@ -6771,7 +6771,7 @@ class 立体分析器: def 测试_保存数据(self, root: str = None): """拆分各序列数据，单独存文件。 - :param root: 保存根目录，默认取脚本所在目录 + :param root: 保存根目录，默认系统临时目录 :return: 数据保存目录路径 """ # 生成存储根目录 diff --git a/chanlun-py/chanlun/chan.py b/chanlun-py/chanlun/chan.py index c5c8f92..402a726 100644 --- a/chanlun-py/chanlun/chan.py +++ b/chanlun-py/chanlun/chan.py @@ -53,7 +53,7 @@ from typing import ( ) from collections.abc import Sequence -from chanlun.termcolor import colored +from termcolor import colored from loguru import logger @@ -6383,7 +6383,7 @@ class 观察者: 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置) 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0) - for i in range(self.混合扩展线段分析层次): + for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))): 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置) 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0) @@ -6429,7 +6429,7 @@ class 观察者: for i in range(self.扩展线段分析层次): 保存序列(self.扩展线段序列组[i]) 保存序列(self.扩展中枢序列组[i]) - for i in range(self.线段分析层次): + for i in range(self.混合扩展线段分析层次): 保存序列(self.混合扩展线段序列组[i]) 保存序列(self.混合扩展中枢序列组[i]) @@ -6487,7 +6487,7 @@ class 观察者: 线段.扩展分析(self.扩展线段序列组[i - 1], self.扩展线段序列组[i], self.配置) 中枢.分析(self.扩展线段序列组[i], self.扩展中枢序列组[i], True, "", 0) - for i in range(self.混合扩展线段分析层次): + for i in range(min(self.混合扩展线段分析层次, len(self.线段序列组))): 线段.扩展分析(self.线段序列组[i], self.混合扩展线段序列组[i], self.配置) 中枢.分析(self.混合扩展线段序列组[i], self.混合扩展中枢序列组[i], True, "", 0) @@ -6771,7 +6771,7 @@ class 立体分析器: def 测试_保存数据(self, root: str = None): """拆分各序列数据，单独存文件。 - :param root: 保存根目录，默认取脚本所在目录 + :param root: 保存根目录，默认系统临时目录 :return: 数据保存目录路径 """ # 生成存储根目录 diff --git a/chanlun-py/tests/test_all.py b/chanlun-py/tests/test_all.py index 7af8142..e9d5bf0 100644 --- a/chanlun-py/tests/test_all.py +++ b/chanlun-py/tests/test_all.py @@ -1764,6 +1764,83 @@ class Test指标挂载(unittest.TestCase): self.assertNotEqual(macd_before, macd_after) +# ============================================================ +# 线段分析层次 = 0 时不崩溃 +# ============================================================ + + +class Test线段分析层次为零(unittest.TestCase): + """验证 线段分析层次=0 时，各处理方法不会越界崩溃.""" + + @classmethod + def setUpClass(cls): + if not _has_nb(): + raise unittest.SkipTest("需要 .nb 数据文件") + + def test_处理数据_不崩溃(self): + """投喂K线时 线段分析层次=0 → 跳过所有线段/扩展线段/混合扩展线段分析，不应崩溃.""" + import chanlun + from chanlun import chan + + # — Rust 侧 — + cfg_rs = chanlun.缠论配置() + obs_rs = chanlun.观察者("btcusd", 300, cfg_rs) + obs_rs.线段分析层次 = 0 + obs_rs.重置基础序列() + + for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:500]: + obs_rs.投喂原始数据(ts, o, h, l, c, v) + + self.assertGreater(len(obs_rs.缠论K线序列), 0, "缠K序列应有数据") + self.assertGreater(len(obs_rs.分型序列), 0, "分型序列应有数据") + self.assertEqual(len(obs_rs.线段序列组), 0, "线段序列组应为空") + self.assertTrue(all(len(s) == 0 for s in obs_rs.混合扩展线段序列组), "混合扩展线段序列组所有条目应为空") + + # — Python 侧 (chan.py) — + cfg_py = chan.缠论配置() + obs_py = chan.观察者("btcusd", 300, cfg_py) + obs_py.线段分析层次 = 0 + obs_py.重置基础序列() + + for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:500]: + obs_py.投喂原始数据(ts, o, h, l, c, v) + + self.assertGreater(len(obs_py.缠论K线序列), 0, "chan.py 缠K序列应有数据") + self.assertGreater(len(obs_py.分型序列), 0, "chan.py 分型序列应有数据") + self.assertEqual(len(obs_py.线段序列组), 0, "chan.py 线段序列组应为空") + + def test_静态重新分析_不崩溃(self): + """静态重新分析时 线段分析层次=0 → 跳过所有线段分析，不应崩溃.""" + import chanlun + from chanlun import chan + + # — Rust 侧 — + cfg_rs = chanlun.缠论配置() + obs_rs = chanlun.观察者("btcusd", 300, cfg_rs) + for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:300]: + obs_rs.投喂原始数据(ts, o, h, l, c, v) + + self.assertGreater(len(obs_rs.线段序列组), 0, "正常初始化后应有线段") + + obs_rs.线段分析层次 = 0 + obs_rs.静态重新分析() + self.assertEqual(len(obs_rs.线段序列组), 0, "静态重新分析后线段序列组应为空") + self.assertGreater(len(obs_rs.分型序列), 0, "静态重新分析后分型序列应有数据") + + # — Python 侧 (chan.py) — + cfg_py = chan.缠论配置() + obs_py = chan.观察者("btcusd", 300, cfg_py) + for ts, o, h, l, c, v in read_nb_bars(NB_PATH)[:300]: + obs_py.投喂原始数据(ts, o, h, l, c, v) + + self.assertGreater(len(obs_py.线段序列组), 0, "chan.py 正常初始化后应有线段") + + obs_py.线段分析层次 = 0 + obs_py.静态重新分析() + self.assertEqual(len(obs_py.线段序列组), 0, "chan.py 静态重新分析后线段序列组应为空") + self.assertGreater(len(obs_py.分型序列), 0, "chan.py 静态重新分析后分型序列应有数据") + + # ============================================================ # 集成对比测试 # ============================================================ diff --git a/chanlun/src/business/observer.rs b/chanlun/src/business/observer.rs index 93e53ab..9606ff9 100644 --- a/chanlun/src/business/observer.rs +++ b/chanlun/src/business/observer.rs @@ -322,7 +322,8 @@ impl 观察者 { } // Step 6: 混合扩展线段分析 — 3 级递归 (源 = 线段序列组[i]) - for i in 0..self.混合扩展线段分析层次 { + // NOTE: 当 线段分析层次=0 时 线段序列组 为空，用 min 避免越界 + for i in 0..self.混合扩展线段分析层次.min(self.线段序列组.len()) { let 源序列 = self.线段序列组[i].clone(); 线段::扩展分析(&源序列, &mut self.混合扩展线段序列组[i], &self.配置); 中枢::分析( @@ -422,7 +423,7 @@ impl 观察者 { ); } - for i in 0..self.混合扩展线段分析层次 { + for i in 0..self.混合扩展线段分析层次.min(self.线段序列组.len()) { let 源序列 = self.线段序列组[i].clone(); 线段::扩展分析(&源序列, &mut self.混合扩展线段序列组[i], &self.配置); 中枢::分析( @@ -1135,4 +1136,74 @@ mod tests { assert_eq!(符号, "ethusd"); assert_eq!(周期, 7200); } + + #[test] + fn test_处理数据_线段分析层次为零_不崩溃() { + let mut config = 缠论配置::default(); + config.加载文件路径 = test_data_path(); + let obs = 观察者::new("btcusd".into(), 300, config); + let mut obs_w = obs.write().unwrap(); + obs_w.线段分析层次 = 0; + obs_w.重置基础序列(); + drop(obs_w); + + // 逐根投喂K线，不应因 线段分析层次=0 而 panic + let data = std::fs::read(test_data_path()).unwrap(); + let size = 48; + for i in 0..(data.len() / size).min(500) { + let offset = i * size; + if let Some(k线) = K线::from_bytes(&data[offset..offset + size], 300, "btcusd") { + obs.write().unwrap().增加原始K线(k线); + } + } + + let obs_r = obs.read().unwrap(); + assert!(obs_r.缠论K线序列.len() > 0, "缠K序列应有数据"); + assert!(obs_r.分型序列.len() > 0, "分型序列应有数据"); + assert!(obs_r.线段序列组.is_empty(), "线段序列组应为空"); + // 混合扩展线段序列组 有 3 个空 Vec（因为 混合扩展线段分析层次 仍是 3）， + // 但所有条目应为空（min(3, 0) = 0，循环未执行） + assert!( + obs_r.混合扩展线段序列组.iter().all(|s| s.is_empty()), + "混合扩展线段序列组所有条目应为空" + ); + info!( + "线段分析层次=0 处理数据 OK: {} 缠K, {} 分型, {} 笔", + obs_r.缠论K线序列.len(), + obs_r.分型序列.len(), + obs_r.笔序列.len() + ); + } + + #[test] + fn test_静态重新分析_线段分析层次为零_不崩溃() { + let mut config = 缠论配置::default(); + config.加载文件路径 = test_data_path(); + let obs = 观察者::new("btcusd".into(), 300, config); + + // 先正常投喂数据 + obs.write() + .unwrap() + .读取数据文件(&test_data_path(), Default::default()) + .unwrap(); + + // 设为0后执行静态重新分析，不应 panic + let mut obs_w = obs.write().unwrap(); + obs_w.线段分析层次 = 0; + obs_w.静态重新分析(); + drop(obs_w); + + let obs_r = obs.read().unwrap(); + assert!(obs_r.分型序列.len() > 0, "静态重新分析后分型序列应有数据"); + assert!(obs_r.线段序列组.is_empty(), "线段序列组应为空"); + assert!( + obs_r.混合扩展线段序列组.iter().all(|s| s.is_empty()), + "混合扩展线段序列组所有条目应为空" + ); + info!( + "线段分析层次=0 静态重新分析 OK: {} 分型, {} 笔", + obs_r.分型序列.len(), + obs_r.笔序列.len() + ); + } }