106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
"""
|
|
策略基类 — 所有策略遵循统一接口协议。
|
|
|
|
实现一个新策略只需继承 Strategy 并实现 generate_signals / build_config / warmup_bars。
|
|
策略文件放入本目录后会被 strategies/__init__.py 自动发现注册。
|
|
|
|
约定:
|
|
- df 为 pandas.DataFrame, 列: time, open, high, low, close, tick_volume
|
|
- entries / exits 为 np.ndarray[bool], 长度 = len(df)
|
|
- warmup 期内的信号位必须置 False (防前视偏差)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import raptorbt
|
|
|
|
|
|
@dataclass
|
|
class SignalResult:
|
|
"""策略生成的信号集合"""
|
|
|
|
entries: np.ndarray # bool, 入场信号
|
|
exits: np.ndarray # bool, 出场信号
|
|
direction: int = 1 # 1=做多, -1=做空
|
|
extra: dict = field(default_factory=dict) # 策略附加信息(如中间指标值)
|
|
|
|
|
|
class Strategy:
|
|
"""
|
|
策略抽象基类。
|
|
|
|
子类必须设置类属性 `name` 并实现以下方法:
|
|
- generate_signals(df) -> SignalResult
|
|
- build_config() -> raptorbt.PyBacktestConfig
|
|
- warmup_bars() -> int
|
|
|
|
可选覆盖:
|
|
- description() -> str
|
|
"""
|
|
|
|
name: str = "base"
|
|
|
|
# ── 必须实现 ──
|
|
def generate_signals(self, df: pd.DataFrame) -> SignalResult:
|
|
raise NotImplementedError
|
|
|
|
def build_config(self) -> "raptorbt.PyBacktestConfig":
|
|
raise NotImplementedError
|
|
|
|
def warmup_bars(self) -> int:
|
|
"""返回指标预热所需的最小 K 线数"""
|
|
return 0
|
|
|
|
# ── 可选覆盖 ──
|
|
def description(self) -> str:
|
|
return self.name
|
|
|
|
# ── 通用辅助 ──
|
|
@staticmethod
|
|
def to_arrays(df: pd.DataFrame) -> dict:
|
|
"""从 DataFrame 提取 raptorbt 所需的 ndarray, float64/int64"""
|
|
return {
|
|
"timestamps": df["time"].values.astype("int64"),
|
|
"open": df["open"].values.astype(np.float64),
|
|
"high": df["high"].values.astype(np.float64),
|
|
"low": df["low"].values.astype(np.float64),
|
|
"close": df["close"].values.astype(np.float64),
|
|
"volume": df.get("tick_volume", df.get("volume", pd.Series(np.ones(len(df))))).values.astype(np.float64),
|
|
}
|
|
|
|
@staticmethod
|
|
def cross_above(fast: np.ndarray, slow: np.ndarray) -> np.ndarray:
|
|
"""fast 向上穿越 slow 的信号 (bar i 收盘后判定, 无前视)"""
|
|
if len(fast) == 0:
|
|
return np.array([], dtype=bool)
|
|
# 前一根 fast <= slow, 当前根 fast > slow
|
|
prev_le = np.empty_like(fast, dtype=bool)
|
|
prev_le[0] = False # 第一根无前值, 不触发交叉
|
|
prev_le[1:] = fast[:-1] <= slow[:-1]
|
|
return (fast > slow) & prev_le
|
|
|
|
@staticmethod
|
|
def cross_below(fast: np.ndarray, slow: np.ndarray) -> np.ndarray:
|
|
"""fast 向下穿越 slow 的信号 (bar i 收盘后判定, 无前视)"""
|
|
if len(fast) == 0:
|
|
return np.array([], dtype=bool)
|
|
prev_ge = np.empty_like(fast, dtype=bool)
|
|
prev_ge[0] = False
|
|
prev_ge[1:] = fast[:-1] >= slow[:-1]
|
|
return (fast < slow) & prev_ge
|
|
|
|
def apply_warmup(self, entries: np.ndarray, exits: np.ndarray) -> tuple:
|
|
"""将 warmup 期内的信号置 False"""
|
|
w = self.warmup_bars()
|
|
if w > 0:
|
|
entries = entries.copy()
|
|
exits = exits.copy()
|
|
entries[:w] = False
|
|
exits[:w] = False
|
|
return entries, exits
|