From f4dc28c5d218e66b3364a799ced19735046bb46d Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Thu, 11 Jun 2026 02:25:42 +0800 Subject: [PATCH] feat: add append_klines for incremental chanlun analysis - Store previous DataFrame in ChanlunAnalyser after process_klines - Add append_klines(df_new) to concatenate and recompute - Handles datetime deduplication automatically - Raises RuntimeError if called before initial process_klines - Add 2 tests: append + recompute, error without init --- src/easy_tdx/chanlun/analyser.py | 35 +++++++++++++++++++++++++++++++ tests/unit/test_chanlun_phase3.py | 31 +++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/src/easy_tdx/chanlun/analyser.py b/src/easy_tdx/chanlun/analyser.py index ffd95e2..95d649e 100644 --- a/src/easy_tdx/chanlun/analyser.py +++ b/src/easy_tdx/chanlun/analyser.py @@ -156,6 +156,7 @@ class ChanlunAnalyser: code=code, frequency=frequency, ) + self._prev_df: pd.DataFrame | None = None @property def config(self) -> ChanlunConfig: @@ -174,6 +175,7 @@ class ChanlunAnalyser: Returns: ChanlunResult 包含所有缠论计算结果 """ + self._prev_df = df.copy() # Step 1: DataFrame → Kline 列表 klines = _df_to_klines(df) self._result.klines = klines @@ -217,6 +219,39 @@ class ChanlunAnalyser: return self._result + def append_klines(self, df_new: pd.DataFrame) -> ChanlunResult: + """增量追加 K 线数据并重新计算。 + + 将新数据追加到之前处理过的 DataFrame 后面, + 然后在完整数据上重新执行缠论计算管道。 + + 相比手动拼接 + process_klines 的优势: + - API 更简洁,无需用户管理 DataFrame 拼接 + - 未来可优化为只重新计算受影响的部分 + + Args: + df_new: 新增的 K 线 DataFrame + + Returns: + 更新后的 ChanlunResult + + Raises: + RuntimeError: 如果之前没有调用过 process_klines + """ + if self._prev_df is None: + msg = "请先调用 process_klines() 初始化,再使用 append_klines()" + raise RuntimeError(msg) + + # 拼接旧数据 + 新数据 + combined = pd.concat([self._prev_df, df_new], ignore_index=True) + + # 去重(防止重复追加) + if "datetime" in combined.columns: + combined = combined.drop_duplicates(subset=["datetime"], keep="last") + combined = combined.reset_index(drop=True) + + return self.process_klines(combined) + def get_bis(self) -> list[BI]: return self._result.bis diff --git a/tests/unit/test_chanlun_phase3.py b/tests/unit/test_chanlun_phase3.py index b38fc40..b09913b 100644 --- a/tests/unit/test_chanlun_phase3.py +++ b/tests/unit/test_chanlun_phase3.py @@ -5,6 +5,7 @@ from __future__ import annotations from datetime import datetime import pandas as pd +import pytest from easy_tdx.chanlun.bi import find_bis from easy_tdx.chanlun.fractal import find_fractals @@ -147,6 +148,36 @@ class TestIncrementalUpdate: assert count2 == 100 assert count2 > count1 + def test_append_klines(self) -> None: + """append_klines 应追加数据并重新计算。""" + from easy_tdx.chanlun.analyser import ChanlunAnalyser + + df1 = _make_df(30) + analyser = ChanlunAnalyser(code="SZ000001") + analyser.process_klines(df1) + count1 = len(analyser.result.klines) + + df_new = _make_df(20) + # 使用不重复的日期 + df_new["datetime"] = pd.date_range("2025-06-01", periods=20, freq="B") + + analyser.append_klines(df_new) + count2 = len(analyser.result.klines) + + # 追加后总数据量应增加 + assert count2 >= count1 + assert count2 == count1 + 20 + + def test_append_without_init_raises(self) -> None: + """未初始化时调用 append_klines 应抛异常。""" + from easy_tdx.chanlun.analyser import ChanlunAnalyser + + analyser = ChanlunAnalyser(code="SZ000001") + df_new = _make_df(10) + + with pytest.raises(RuntimeError, match="请先调用 process_klines"): + analyser.append_klines(df_new) + # ── 走势段测试 ──────────────────────────────────────────────────────────