mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 16:54:20 +08:00
feat(backtest): add DSL strategy skeleton and update __init__.py exports
- Add dsl_strategy decorator in dsl.py (P1 skeleton implementation) - Update __init__.py to export BacktestEngine, Strategy, and related types - All 106 backtest unit tests pass Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
371915a5f9
commit
706f22ba5e
@@ -1 +1,36 @@
|
||||
"""easy_tdx.backtest — 向量化策略回测引擎(纯计算,零网络依赖)。"""
|
||||
"""easy_tdx.backtest — 向量化策略回测引擎(纯计算,零网络依赖)。
|
||||
|
||||
快速开始::
|
||||
|
||||
from easy_tdx.backtest import BacktestEngine, Strategy
|
||||
|
||||
class MyStrategy(Strategy):
|
||||
def init(self):
|
||||
self.ma5 = self.I(MA, self.data.close, 5)
|
||||
self.ma20 = self.I(MA, self.data.close, 20)
|
||||
|
||||
def next(self):
|
||||
if crossover(self.ma5, self.ma20):
|
||||
self.buy()
|
||||
elif crossover(self.ma20, self.ma5):
|
||||
self.sell()
|
||||
|
||||
engine = BacktestEngine(strategy=MyStrategy, cash=100000)
|
||||
result = engine.run(df)
|
||||
print(result.performance)
|
||||
"""
|
||||
|
||||
from easy_tdx.backtest.engine import BacktestEngine # noqa: F401
|
||||
from easy_tdx.backtest.strategy import Strategy, StrategyDataProxy, crossover # noqa: F401
|
||||
from easy_tdx.backtest.types import BacktestResult, Position, Signal, Trade # noqa: F401
|
||||
|
||||
__all__ = [
|
||||
"BacktestEngine",
|
||||
"BacktestResult",
|
||||
"Strategy",
|
||||
"StrategyDataProxy",
|
||||
"Signal",
|
||||
"Trade",
|
||||
"Position",
|
||||
"crossover",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""DSL 策略定义模块 (P1 — 骨架)。
|
||||
|
||||
v1 提供 @dsl_strategy 装饰器的基本实现。
|
||||
字符串 DSL 解析器将在后续版本实现。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
|
||||
from .strategy import Strategy
|
||||
|
||||
|
||||
def dsl_strategy(func: Callable[..., Any]) -> type[Strategy]:
|
||||
"""将函数编译为 Strategy 子类。
|
||||
|
||||
函数签名: (df: pd.DataFrame) -> tuple[np.ndarray[bool], np.ndarray[bool]]
|
||||
返回 (buy_mask, sell_mask)。
|
||||
|
||||
用法::
|
||||
|
||||
@dsl_strategy
|
||||
def dual_ma(df):
|
||||
buy = CROSS(MA(df.close, 5), MA(df.close, 20))
|
||||
sell = CROSS(MA(df.close, 20), MA(df.close, 5))
|
||||
return buy, sell
|
||||
"""
|
||||
|
||||
class DSLStrategy(Strategy):
|
||||
_signal_func = staticmethod(func)
|
||||
_buy_mask: np.ndarray | None = None
|
||||
_sell_mask: np.ndarray | None = None
|
||||
|
||||
def init(self) -> None:
|
||||
pass
|
||||
|
||||
def next(self) -> None:
|
||||
if self._buy_mask is None:
|
||||
return
|
||||
idx = self._bar_index
|
||||
if idx < len(self._buy_mask) and self._buy_mask[idx]:
|
||||
self.buy(size=0)
|
||||
elif idx < len(self._sell_mask) and self._sell_mask[idx]:
|
||||
self.sell(size=0)
|
||||
|
||||
DSLStrategy.__name__ = func.__name__
|
||||
DSLStrategy.__qualname__ = func.__qualname__
|
||||
DSLStrategy._signal_func = func # type: ignore[attr-defined]
|
||||
|
||||
return DSLStrategy
|
||||
Reference in New Issue
Block a user