mirror of
https://ghfast.top/https://github.com/aeroxw/easy-tdx.git
synced 2026-09-12 16:54:17 +08:00
fix(backtest): resolve mypy and ruff lint issues
- dsl.py: use NDArray type annotations, fix None narrowing - cli.py: add type annotations, fix import sorting - strategy.py: fix UP038 isinstance, add noqa for I() method name - tests: fix E712 bool comparison assertions Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
fc0777533e
commit
04c2be1d7f
@@ -3,8 +3,8 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import click
|
||||
|
||||
@@ -53,9 +53,9 @@ def backtest(
|
||||
|
||||
easy-tdx backtest SZ 000001 --strategy-file my_strategy.py --indicators MACD,KDJ
|
||||
"""
|
||||
from ..backtest.engine import BacktestEngine
|
||||
from ..cli.conn import get_mac_client
|
||||
from ..cli.parsers import parse_adjust, parse_market, parse_period
|
||||
from ..backtest.engine import BacktestEngine
|
||||
from ..indicator import compute_indicators
|
||||
|
||||
# 1. 加载策略
|
||||
@@ -114,7 +114,6 @@ def _load_strategy(
|
||||
Returns:
|
||||
Strategy 子类
|
||||
"""
|
||||
from ..backtest.strategy import Strategy
|
||||
|
||||
if strategy_file:
|
||||
return _load_strategy_from_file(strategy_file)
|
||||
@@ -174,7 +173,7 @@ def _load_strategy_from_file(path: str) -> type:
|
||||
return strategy_classes[0]
|
||||
|
||||
|
||||
def _print_table(result) -> None:
|
||||
def _print_table(result: Any) -> None:
|
||||
"""以表格形式输出回测结果。"""
|
||||
perf = result.performance
|
||||
config = result.config
|
||||
|
||||
@@ -10,6 +10,7 @@ from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
from numpy.typing import NDArray
|
||||
|
||||
from .strategy import Strategy
|
||||
|
||||
@@ -31,23 +32,24 @@ def dsl_strategy(func: Callable[..., Any]) -> type[Strategy]:
|
||||
|
||||
class DSLStrategy(Strategy):
|
||||
_signal_func = staticmethod(func)
|
||||
_buy_mask: np.ndarray | None = None
|
||||
_sell_mask: np.ndarray | None = None
|
||||
_buy_mask: NDArray[np.bool_] | None = None
|
||||
_sell_mask: NDArray[np.bool_] | None = None
|
||||
|
||||
def init(self) -> None:
|
||||
pass
|
||||
|
||||
def next(self) -> None:
|
||||
if self._buy_mask is None:
|
||||
buy = self._buy_mask
|
||||
sell = self._sell_mask
|
||||
if buy is None or sell is None:
|
||||
return
|
||||
idx = self._bar_index
|
||||
if idx < len(self._buy_mask) and self._buy_mask[idx]:
|
||||
if idx < len(buy) and buy[idx]:
|
||||
self.buy(size=0)
|
||||
elif idx < len(self._sell_mask) and self._sell_mask[idx]:
|
||||
elif idx < len(sell) and sell[idx]:
|
||||
self.sell(size=0)
|
||||
|
||||
DSLStrategy.__name__ = func.__name__
|
||||
DSLStrategy.__qualname__ = func.__qualname__
|
||||
DSLStrategy._signal_func = func # type: ignore[attr-defined]
|
||||
|
||||
return DSLStrategy
|
||||
|
||||
@@ -5,8 +5,6 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
|
||||
@@ -100,7 +100,7 @@ class StrategyDataProxy:
|
||||
if col == "datetime":
|
||||
continue
|
||||
arr = df[col].to_numpy()
|
||||
if len(arr) > 0 and isinstance(arr[0], (np.datetime64, pd.Timestamp)):
|
||||
if len(arr) > 0 and isinstance(arr[0], np.datetime64 | pd.Timestamp):
|
||||
# datetime 列转为 int (YYYYMMDD)
|
||||
self._arrays[col] = _datetime_to_int(arr)
|
||||
else:
|
||||
@@ -250,7 +250,7 @@ class Strategy(ABC):
|
||||
|
||||
# ── 指标注册 ───────────────────────────────────────────────────────────────
|
||||
|
||||
def I(
|
||||
def I( # noqa: E743
|
||||
self, func: Callable[..., NDArray], *args: Any, **kwargs: Any
|
||||
) -> NDArray:
|
||||
"""注册指标。
|
||||
@@ -441,7 +441,7 @@ def _datetime_to_int(arr: NDArray) -> NDArray:
|
||||
"""
|
||||
result = np.zeros(len(arr), dtype=np.float64)
|
||||
for i, val in enumerate(arr):
|
||||
if isinstance(val, (np.datetime64, pd.Timestamp)):
|
||||
if isinstance(val, np.datetime64 | pd.Timestamp):
|
||||
ts = pd.Timestamp(val)
|
||||
result[i] = float(ts.strftime("%Y%m%d"))
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user