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
This commit is contained in:
Justin Gu
2026-06-11 01:56:59 +08:00
parent 815b3ddf7c
commit af005d9fe4
3 changed files with 75 additions and 3 deletions
+13 -1
View File
@@ -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)
+6 -2
View File
@@ -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
# ── 内部方法(引擎调用) ─────────────────────────────────────────────────────
+56
View File
@@ -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