365 lines
10 KiB
Python
365 lines
10 KiB
Python
|
"""
|
|||
|
|
策略模板生成器 — 为 AI agent 提供标准化起点
|
|||
|
|
|
|||
|
|
支持模板:
|
|||
|
|
- crossover: 均线交叉 (SMA/EMA)
|
|||
|
|
- mean_reversion: RSI/Bollinger 均值回归
|
|||
|
|
- trend_following: ADX + DI 趋势跟踪
|
|||
|
|
- breakout: Donchian 通道突破
|
|||
|
|
- custom: 空白模板
|
|||
|
|
|
|||
|
|
用法:
|
|||
|
|
from scaffold import scaffold_strategy
|
|||
|
|
path = scaffold_strategy("my_rsi", template="mean_reversion",
|
|||
|
|
description="RSI 超卖反弹策略")
|
|||
|
|
print(f"模板已生成: {path}")
|
|||
|
|
# 接着编辑 strategies/my_rsi.py 填入具体逻辑
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _to_class_name(name: str) -> str:
|
|||
|
|
"""snake_case → PascalCase (my_strategy → MyStrategy)"""
|
|||
|
|
return "".join(w.capitalize() for w in name.split("_"))
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 策略文件头部注释 (包含前视偏差警告, 提醒 AI agent 遵守)
|
|||
|
|
_HEADER = '''"""{name} — {description}
|
|||
|
|
|
|||
|
|
⚠️ 前视偏差 (Look-Ahead Bias) 注意事项:
|
|||
|
|
信号生成时只能用当前 bar 及之前的数据, 严禁使用未来 bar。
|
|||
|
|
以下模式会引入前视偏差, 必须避免:
|
|||
|
|
- .shift(-N) # 访问未来 bar (N>0)
|
|||
|
|
- close[-1] / high[-1] # 负索引访问未来
|
|||
|
|
- df.iloc[i+N:] # 切片到未来索引
|
|||
|
|
- 滚动统计后 shift 负值
|
|||
|
|
正确做法:
|
|||
|
|
- 用 cross_above / cross_below (基类已内置前视安全)
|
|||
|
|
- 信号在 bar 收盘后生成, 用 close 成交 (引擎默认 upon_bar_close=True)
|
|||
|
|
- 检测: python -m app.main check {name}
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
'''
|
|||
|
|
|
|||
|
|
_TEMPLATES = {
|
|||
|
|
|
|||
|
|
"crossover": '''"""{name} — {description}"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import numpy as np
|
|||
|
|
import raptorbt
|
|||
|
|
|
|||
|
|
from .base import Strategy, SignalResult
|
|||
|
|
|
|||
|
|
|
|||
|
|
class {ClassName}Strategy(Strategy):
|
|||
|
|
"""{description}"""
|
|||
|
|
|
|||
|
|
name = "{name}"
|
|||
|
|
|
|||
|
|
def __init__(self, fast: int = 10, slow: int = 20):
|
|||
|
|
self.fast = fast
|
|||
|
|
self.slow = slow
|
|||
|
|
|
|||
|
|
def warmup_bars(self) -> int:
|
|||
|
|
return self.slow + 1
|
|||
|
|
|
|||
|
|
def generate_signals(self, df) -> SignalResult:
|
|||
|
|
close = df["close"].values.astype(np.float64)
|
|||
|
|
ma_fast = raptorbt.sma(close, period=self.fast)
|
|||
|
|
ma_slow = raptorbt.sma(close, period=self.slow)
|
|||
|
|
|
|||
|
|
entries = self.cross_above(ma_fast, ma_slow).astype(bool)
|
|||
|
|
exits = self.cross_below(ma_fast, ma_slow).astype(bool)
|
|||
|
|
entries, exits = self.apply_warmup(entries, exits)
|
|||
|
|
|
|||
|
|
return SignalResult(
|
|||
|
|
entries=entries, exits=exits, direction=1,
|
|||
|
|
extra={{"ma_fast": ma_fast, "ma_slow": ma_slow}},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def build_config(self) -> raptorbt.PyBacktestConfig:
|
|||
|
|
config = raptorbt.PyBacktestConfig(
|
|||
|
|
initial_capital=100000.0, fees=0.001, slippage=0.0005,
|
|||
|
|
)
|
|||
|
|
config.set_fixed_stop(0.02)
|
|||
|
|
config.set_fixed_target(0.04)
|
|||
|
|
return config
|
|||
|
|
|
|||
|
|
def description(self) -> str:
|
|||
|
|
return f"SMA({{self.fast}})/SMA({{self.slow}}) 交叉, 2% 止损/4% 止盈"
|
|||
|
|
|
|||
|
|
|
|||
|
|
STRATEGY_CLASS = {ClassName}Strategy
|
|||
|
|
''',
|
|||
|
|
|
|||
|
|
"mean_reversion": '''"""{name} — {description}"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import numpy as np
|
|||
|
|
import raptorbt
|
|||
|
|
|
|||
|
|
from .base import Strategy, SignalResult
|
|||
|
|
|
|||
|
|
|
|||
|
|
class {ClassName}Strategy(Strategy):
|
|||
|
|
"""{description}"""
|
|||
|
|
|
|||
|
|
name = "{name}"
|
|||
|
|
|
|||
|
|
def __init__(self, period: int = 14, oversold: float = 30.0, overbought: float = 70.0):
|
|||
|
|
self.period = period
|
|||
|
|
self.oversold = oversold
|
|||
|
|
self.overbought = overbought
|
|||
|
|
|
|||
|
|
def warmup_bars(self) -> int:
|
|||
|
|
return self.period + 1
|
|||
|
|
|
|||
|
|
def generate_signals(self, df) -> SignalResult:
|
|||
|
|
close = df["close"].values.astype(np.float64)
|
|||
|
|
rsi = raptorbt.rsi(close, period=self.period)
|
|||
|
|
|
|||
|
|
entries = (rsi < self.oversold).astype(bool)
|
|||
|
|
exits = (rsi > self.overbought).astype(bool)
|
|||
|
|
entries, exits = self.apply_warmup(entries, exits)
|
|||
|
|
|
|||
|
|
return SignalResult(
|
|||
|
|
entries=entries, exits=exits, direction=1,
|
|||
|
|
extra={{"rsi": rsi}},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def build_config(self) -> raptorbt.PyBacktestConfig:
|
|||
|
|
config = raptorbt.PyBacktestConfig(
|
|||
|
|
initial_capital=100000.0, fees=0.001, slippage=0.0005,
|
|||
|
|
)
|
|||
|
|
config.set_trailing_stop(0.03)
|
|||
|
|
return config
|
|||
|
|
|
|||
|
|
def description(self) -> str:
|
|||
|
|
return f"RSI({{self.period}}) 均值回归, <{{self.oversold}} 买入 / >{{self.overbought}} 卖出"
|
|||
|
|
|
|||
|
|
|
|||
|
|
STRATEGY_CLASS = {ClassName}Strategy
|
|||
|
|
''',
|
|||
|
|
|
|||
|
|
"trend_following": '''"""{name} — {description}"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import numpy as np
|
|||
|
|
import raptorbt
|
|||
|
|
|
|||
|
|
from .base import Strategy, SignalResult
|
|||
|
|
|
|||
|
|
|
|||
|
|
class {ClassName}Strategy(Strategy):
|
|||
|
|
"""{description}"""
|
|||
|
|
|
|||
|
|
name = "{name}"
|
|||
|
|
|
|||
|
|
def __init__(self, adx_period: int = 14, adx_threshold: float = 25.0):
|
|||
|
|
self.adx_period = adx_period
|
|||
|
|
self.adx_threshold = adx_threshold
|
|||
|
|
|
|||
|
|
def warmup_bars(self) -> int:
|
|||
|
|
return 2 * self.adx_period + 10
|
|||
|
|
|
|||
|
|
def generate_signals(self, df) -> SignalResult:
|
|||
|
|
close = df["close"].values.astype(np.float64)
|
|||
|
|
high = df["high"].values.astype(np.float64)
|
|||
|
|
low = df["low"].values.astype(np.float64)
|
|||
|
|
|
|||
|
|
adx, plus_di, minus_di = raptorbt.adx_all(high, low, close, period=self.adx_period)
|
|||
|
|
|
|||
|
|
adx_strong = adx > self.adx_threshold
|
|||
|
|
entries = (adx_strong & (plus_di > minus_di)).astype(bool)
|
|||
|
|
exits = (adx_strong & (minus_di > plus_di)).astype(bool)
|
|||
|
|
entries, exits = self.apply_warmup(entries, exits)
|
|||
|
|
|
|||
|
|
return SignalResult(
|
|||
|
|
entries=entries, exits=exits, direction=1,
|
|||
|
|
extra={{"adx": adx, "plus_di": plus_di, "minus_di": minus_di}},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def build_config(self) -> raptorbt.PyBacktestConfig:
|
|||
|
|
config = raptorbt.PyBacktestConfig(
|
|||
|
|
initial_capital=100000.0, fees=0.001, slippage=0.0005,
|
|||
|
|
)
|
|||
|
|
config.set_atr_stop(multiplier=2.5, period=14)
|
|||
|
|
config.set_fixed_target(0.05)
|
|||
|
|
return config
|
|||
|
|
|
|||
|
|
def description(self) -> str:
|
|||
|
|
return f"ADX({{self.adx_period}})>{{self.adx_threshold}} + DI 方向确认, 2.5×ATR 止损"
|
|||
|
|
|
|||
|
|
|
|||
|
|
STRATEGY_CLASS = {ClassName}Strategy
|
|||
|
|
''',
|
|||
|
|
|
|||
|
|
"breakout": '''"""{name} — {description}"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import numpy as np
|
|||
|
|
import raptorbt
|
|||
|
|
|
|||
|
|
from .base import Strategy, SignalResult
|
|||
|
|
|
|||
|
|
|
|||
|
|
class {ClassName}Strategy(Strategy):
|
|||
|
|
"""{description}"""
|
|||
|
|
|
|||
|
|
name = "{name}"
|
|||
|
|
|
|||
|
|
def __init__(self, period: int = 20):
|
|||
|
|
self.period = period
|
|||
|
|
|
|||
|
|
def warmup_bars(self) -> int:
|
|||
|
|
return self.period + 1
|
|||
|
|
|
|||
|
|
def generate_signals(self, df) -> SignalResult:
|
|||
|
|
close = df["close"].values.astype(np.float64)
|
|||
|
|
high = df["high"].values.astype(np.float64)
|
|||
|
|
low = df["low"].values.astype(np.float64)
|
|||
|
|
|
|||
|
|
upper, middle, lower = raptorbt.donchian(high, low, period=self.period)
|
|||
|
|
|
|||
|
|
entries = (close > np.roll(upper, 1)).astype(bool)
|
|||
|
|
exits = (close < np.roll(lower, 1)).astype(bool)
|
|||
|
|
entries, exits = self.apply_warmup(entries, exits)
|
|||
|
|
|
|||
|
|
return SignalResult(
|
|||
|
|
entries=entries, exits=exits, direction=1,
|
|||
|
|
extra={{"donchian_upper": upper, "donchian_lower": lower}},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def build_config(self) -> raptorbt.PyBacktestConfig:
|
|||
|
|
config = raptorbt.PyBacktestConfig(
|
|||
|
|
initial_capital=100000.0, fees=0.001, slippage=0.0005,
|
|||
|
|
)
|
|||
|
|
config.set_trailing_stop(0.05)
|
|||
|
|
return config
|
|||
|
|
|
|||
|
|
def description(self) -> str:
|
|||
|
|
return f"Donchian({{self.period}}) 通道突破, 5% 追踪止损"
|
|||
|
|
|
|||
|
|
|
|||
|
|
STRATEGY_CLASS = {ClassName}Strategy
|
|||
|
|
''',
|
|||
|
|
|
|||
|
|
"custom": '''"""{name} — {description}"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import numpy as np
|
|||
|
|
import raptorbt
|
|||
|
|
|
|||
|
|
from .base import Strategy, SignalResult
|
|||
|
|
|
|||
|
|
|
|||
|
|
class {ClassName}Strategy(Strategy):
|
|||
|
|
"""{description}"""
|
|||
|
|
|
|||
|
|
name = "{name}"
|
|||
|
|
|
|||
|
|
def __init__(self, period: int = 14):
|
|||
|
|
self.period = period
|
|||
|
|
|
|||
|
|
def warmup_bars(self) -> int:
|
|||
|
|
return self.period + 1
|
|||
|
|
|
|||
|
|
def generate_signals(self, df) -> SignalResult:
|
|||
|
|
close = df["close"].values.astype(np.float64)
|
|||
|
|
high = df["high"].values.astype(np.float64)
|
|||
|
|
low = df["low"].values.astype(np.float64)
|
|||
|
|
volume = df.get("tick_volume", df.get("volume")).values.astype(np.float64)
|
|||
|
|
|
|||
|
|
# TODO: 在此添加指标计算和信号生成逻辑
|
|||
|
|
# 例如:
|
|||
|
|
# rsi = raptorbt.rsi(close, period=self.period)
|
|||
|
|
# entries = rsi < 30
|
|||
|
|
# exits = rsi > 70
|
|||
|
|
|
|||
|
|
entries = np.zeros(len(close), dtype=bool)
|
|||
|
|
exits = np.zeros(len(close), dtype=bool)
|
|||
|
|
entries, exits = self.apply_warmup(entries, exits)
|
|||
|
|
|
|||
|
|
return SignalResult(entries=entries, exits=exits, direction=1)
|
|||
|
|
|
|||
|
|
def build_config(self) -> raptorbt.PyBacktestConfig:
|
|||
|
|
config = raptorbt.PyBacktestConfig(
|
|||
|
|
initial_capital=100000.0, fees=0.001, slippage=0.0005,
|
|||
|
|
)
|
|||
|
|
config.set_fixed_stop(0.02)
|
|||
|
|
config.set_fixed_target(0.04)
|
|||
|
|
return config
|
|||
|
|
|
|||
|
|
def description(self) -> str:
|
|||
|
|
return "{description}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
STRATEGY_CLASS = {ClassName}Strategy
|
|||
|
|
''',
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def scaffold_strategy(
|
|||
|
|
name: str,
|
|||
|
|
template: str = "custom",
|
|||
|
|
description: str = "",
|
|||
|
|
overwrite: bool = False,
|
|||
|
|
) -> str:
|
|||
|
|
"""
|
|||
|
|
生成策略模板文件
|
|||
|
|
|
|||
|
|
参数:
|
|||
|
|
name: 策略名称 (snake_case, 如 "my_rsi")
|
|||
|
|
template: 模板类型 (crossover/mean_reversion/trend_following/breakout/custom)
|
|||
|
|
description: 策略描述文字
|
|||
|
|
overwrite: 是否覆盖已存在的文件
|
|||
|
|
|
|||
|
|
返回:
|
|||
|
|
生成的文件路径
|
|||
|
|
"""
|
|||
|
|
if template not in _TEMPLATES:
|
|||
|
|
raise ValueError(
|
|||
|
|
f"未知模板 '{template}', 可选: {', '.join(_TEMPLATES.keys())}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
class_name = _to_class_name(name)
|
|||
|
|
if not description:
|
|||
|
|
description = f"{template} 策略"
|
|||
|
|
|
|||
|
|
# 组装: 头部警告 + 模板正文 (去掉模板自带的 docstring 行)
|
|||
|
|
header = _HEADER.format(name=name, description=description)
|
|||
|
|
template_body = _TEMPLATES[template].format(
|
|||
|
|
name=name,
|
|||
|
|
ClassName=class_name,
|
|||
|
|
description=description,
|
|||
|
|
)
|
|||
|
|
# 去掉模板第一行的 docstring (已被 _HEADER 取代)
|
|||
|
|
template_body = template_body.split("\n", 1)[1] if template_body.startswith('"""') else template_body
|
|||
|
|
code = header + template_body
|
|||
|
|
|
|||
|
|
file_path = os.path.join(
|
|||
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|||
|
|
"strategies",
|
|||
|
|
f"{name}.py",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
if os.path.exists(file_path) and not overwrite:
|
|||
|
|
raise FileExistsError(
|
|||
|
|
f"策略文件已存在: {file_path}\n使用 overwrite=True 覆盖"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|||
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|||
|
|
f.write(code)
|
|||
|
|
|
|||
|
|
return file_path
|