refactor(strategy): 分钟红7由内置策略改为自定义策略

按用户要求, 分钟红7不再是产品内置形态, 以自定义策略形态交付:
- 移除 app/strategy/builtin/minute_red_streak.py; 运行时副本落在
  data/strategies/custom/ (gitignore, 用户可自行修改调参)
- 策略 id (minute_red_streak) 不变: 参数覆盖 / 策略池引用 / minute_filter
  契约 (daily_history_bars + daily= 注入) 全部不受位置影响
- 仓库保留参考实现作为测试夹具 (tests/fixtures/strategies/),
  25 个行为测试改由夹具加载, 引擎加载测试同时断言 source == "custom"
- builtin 相关不变量更新: 内置 19 个全部 matrix_native, 无 minute_filter

本地验证: /api/strategies/reload 后 source=custom, 08-25 分区分钟扫描
命中 5 只 (用户实盘参数: 6根开5红+最高2红+20日涨停过+沪深主板50-200亿)。
This commit is contained in:
shy3130
2026-08-30 19:05:11 +08:00
parent 8ccb681214
commit 9593883c59
4 changed files with 20 additions and 16 deletions
@@ -319,13 +319,10 @@ def test_builtin_matrix_strategies_use_their_declared_formula_modules():
path for path in strategy_dir.glob("*.py") if path.name != "__init__.py"
)
# 非 matrix 后端的内置策略白名单 (当前仅分钟形态策略)
non_matrix = {"minute_red_streak"}
assert len(strategy_files) == 19 + len(non_matrix)
# 分钟形态策略 (minute_red_streak) 已迁至自定义策略目录, 内置策略全部 matrix 后端
assert len(strategy_files) == 19
for strategy_path in strategy_files:
strategy = StrategyEngine._load_file(strategy_path)
if strategy_path.stem in non_matrix:
continue
assert strategy.execution_backend == "matrix_native"
assert strategy.matrix_strategy is not None
assert strategy.matrix_strategy.__class__.__module__ == strategy_path.stem
+14 -5
View File
@@ -12,15 +12,24 @@
from __future__ import annotations
import datetime as _dt
import importlib.util
from datetime import date, datetime
from pathlib import Path
import polars as pl
from app.services.screener import ScreenerService
from app.strategy.builtin import minute_red_streak
from app.strategy.engine import StrategyDataContext, StrategyEngine
# 分钟红7 已从内置策略改为自定义策略 (运行时 data/strategies/custom/, 不入库);
# 测试通过仓库内的参考实现夹具加载, 覆盖同一份策略逻辑。
STRATEGY_FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "strategies"
_spec = importlib.util.spec_from_file_location(
"minute_red_streak_fixture", STRATEGY_FIXTURE_DIR / "minute_red_streak.py"
)
minute_red_streak = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(minute_red_streak)
def _bars(symbol: str, candles: list[tuple[float, float, float]], start_hour: int = 9) -> pl.DataFrame:
"""candles: (open, close, high) 序列, 时间从 start_hour:30 起每分钟一根。"""
@@ -261,15 +270,15 @@ def test_pattern_limit_up_disabled_ignores_daily():
# ── 引擎加载与运行 ──────────────────────────────────────────────────
def test_builtin_minute_strategy_loads_with_minute_filter_backend():
engine = StrategyEngine(
strategy_dirs=[Path(__file__).resolve().parent.parent / "app" / "strategy" / "builtin"]
)
def test_custom_minute_strategy_loads_with_minute_filter_backend():
# 自定义策略与内置策略共用同一加载器: 夹具目录即一个 custom 目录
engine = StrategyEngine(strategy_dirs=[STRATEGY_FIXTURE_DIR])
assert not [e for e in engine.load_errors() if "minute" in e["file"]]
s = engine.get("minute_red_streak")
assert s.execution_backend == "minute_filter"
assert s.filter_minute_history_fn is not None
assert s.meta["timeframes"] == ["1m"]
assert s.source == "custom"
def _minute_code(sid: str, timeframes: str = '["1m"]', extra: str = "") -> str:
+4 -6
View File
@@ -41,8 +41,8 @@ def test_all_builtin_strategies_declare_asset_types_and_timeframes():
assert engine.load_errors() == []
for meta in engine.list_strategies():
assert meta["asset_types"]
# 分钟策略 timeframes 为 ["1m"], 日线内置策略为 ["1d"]
assert meta["timeframes"] in (["1d"], ["1m"])
# 分钟红7已迁至自定义策略目录, 内置策略均为日线
assert meta["timeframes"] == ["1d"]
def test_all_builtin_strategies_use_matrix_backend_only():
@@ -54,10 +54,8 @@ def test_all_builtin_strategies_use_matrix_backend_only():
assert all(s.matrix_strategy is not None for s in matrix_strategies)
assert all(s.filter_fn is None for s in matrix_strategies)
assert all(s.filter_history_fn is None for s in matrix_strategies)
# 分钟形态策略 (minute_filter) 不参与日线矩阵不变量
assert [s.meta["id"] for s in strategies if s.execution_backend == "minute_filter"] == [
"minute_red_streak"
]
# 分钟形态策略 (minute_filter) 已迁至自定义策略目录, 不在 builtin 加载范围
assert [s.meta["id"] for s in strategies if s.execution_backend == "minute_filter"] == []
def test_all_builtin_matrix_formulas_accept_base_market_matrix():