From af005d9fe4caef51c8d9519bc39a23ac5699a0eb Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Thu, 11 Jun 2026 01:56:59 +0800 Subject: [PATCH] feat: auto-bridge chanlun analysis into backtest strategies - Add chanlun_level param to BacktestEngine constructor - When set, auto-create ChanlunAnalyser and compute ChanlunResult - Manual chanlun_result in run() takes priority over auto-compute - Update Strategy.chanlun type to Any (accepts ChanlunResult or dict) - Add 2 tests: auto-bridge and manual override priority --- src/easy_tdx/backtest/engine.py | 14 +++++++- src/easy_tdx/backtest/strategy.py | 8 +++-- tests/unit/test_backtest_engine.py | 56 ++++++++++++++++++++++++++++++ 3 files changed, 75 insertions(+), 3 deletions(-) diff --git a/src/easy_tdx/backtest/engine.py b/src/easy_tdx/backtest/engine.py index e1fa2a3..005570a 100644 --- a/src/easy_tdx/backtest/engine.py +++ b/src/easy_tdx/backtest/engine.py @@ -56,6 +56,7 @@ class BacktestEngine: position_mode: str = "full", reject_policy: str = "reduce", benchmark: pd.DataFrame | None = None, + chanlun_level: str | None = None, ): """Initialize engine. @@ -70,6 +71,8 @@ class BacktestEngine: position_mode: Position mode ('full', 'long_only', 'short_only') reject_policy: Reject policy ('reduce', 'reject') benchmark: Benchmark data for performance comparison + chanlun_level: Auto-compute chanlun analysis at this level + (e.g. 'DAILY', '30MIN'). Strategy accesses via self.chanlun. """ self._strategy_cls = strategy if isinstance(strategy, type) else type(strategy) self._strategy_instance = strategy if isinstance(strategy, Strategy) else None @@ -83,13 +86,15 @@ class BacktestEngine: self._position_mode = position_mode self._reject_policy = reject_policy self._benchmark = benchmark + self._chanlun_level = chanlun_level def run(self, df: pd.DataFrame, chanlun_result: Any | None = None) -> BacktestResult: """Run backtest. Args: df: Price data with OHLCV columns - chanlun_result: Optional chanlun analysis result for strategy + chanlun_result: Optional chanlun analysis result for strategy. + When provided, takes priority over auto-computed result. Returns: BacktestResult with performance, equity_curve, trades, positions, config @@ -97,6 +102,13 @@ class BacktestEngine: if len(df) == 0: return self._empty_result() + # Auto-compute chanlun if chanlun_level is set and no manual result + if chanlun_result is None and self._chanlun_level is not None: + from easy_tdx.chanlun.analyser import ChanlunAnalyser + + analyser = ChanlunAnalyser(frequency=self._chanlun_level) + chanlun_result = analyser.process_klines(df) + # Step 1: Signal generation signals = self._generate_signals(df, chanlun_result) diff --git a/src/easy_tdx/backtest/strategy.py b/src/easy_tdx/backtest/strategy.py index fa2ea9e..0ad45e6 100644 --- a/src/easy_tdx/backtest/strategy.py +++ b/src/easy_tdx/backtest/strategy.py @@ -368,8 +368,12 @@ class Strategy(ABC): return {"size": self._position_size} @property - def chanlun(self) -> dict[str, Any] | None: - """缠论分析结果（预留）。""" + def chanlun(self) -> Any: + """缠论分析结果。 + + 通过 BacktestEngine(chanlun_level=...) 自动注入 ChanlunResult， + 或通过 engine.run(chanlun_result=...) 手动注入任意对象。 + """ return self._chanlun_result # ── 内部方法（引擎调用） ───────────────────────────────────────────────────── diff --git a/tests/unit/test_backtest_engine.py b/tests/unit/test_backtest_engine.py index 98b6488..1216f96 100644 --- a/tests/unit/test_backtest_engine.py +++ b/tests/unit/test_backtest_engine.py @@ -496,3 +496,59 @@ def test_stop_loss_takes_priority_over_strategy_sell(): # Should have exactly 1 SELL (from SL, not the manual one at bar 15) assert len(sell_trades) == 1, f"Expected 1 SL sell, got {len(sell_trades)}" assert sell_trades.iloc[0]["price"] == 95.0 + + +# ── Chanlun Auto-Bridge ────────────────────────────────────────────────────── + + +class ChanlunAwareStrategy(Strategy): + """Strategy that buys when chanlun analysis has at least one bi.""" + + def init(self) -> None: + pass + + def next(self) -> None: + if self.chanlun is not None and self._bar_index == 15 and self.position["size"] == 0: + # Strategy uses chanlun result to make trading decisions + bis = self.chanlun.bis if hasattr(self.chanlun, "bis") else [] + if len(bis) > 0: + self.buy(size=0) + + +def test_chanlun_auto_bridge(): + """Test chanlun_level auto-computes and injects analysis into strategy.""" + df = _make_df(n=100) + engine = BacktestEngine(ChanlunAwareStrategy, cash=100000, chanlun_level="DAILY") + result = engine.run(df) + + # Strategy should have received chanlun result (100 bars → at least some bis) + trades = result.trades[~result.trades["rejected"]] + buy_trades = trades[trades["direction"] == "BUY"] + + # With 100 bars of random data, ChanlunAnalyser should produce bis, + # so the strategy should trigger a BUY at bar 15 + assert len(buy_trades) >= 1, "Expected chanlun-aware BUY" + + +def test_chanlun_manual_result_overrides_auto(): + """Test explicit chanlun_result takes priority over chanlun_level.""" + df = _make_df(n=50) + + class CheckerStrategy(Strategy): + received: object = None + + def init(self) -> None: + pass + + def next(self) -> None: + if self._bar_index == 10: + CheckerStrategy.received = self.chanlun + self.buy(size=10) + + # Pass explicit result — should NOT auto-compute + manual_result = {"manual": True} + engine = BacktestEngine(CheckerStrategy, cash=100000, chanlun_level="DAILY") + result = engine.run(df, chanlun_result=manual_result) + + # Strategy should have received the manual result, not auto-computed one + assert CheckerStrategy.received == manual_result