feat(strategy): add importable minute price-volume strategy

Signed-off-by: shy3130 <shy3130@qq.com>

backend/tests/fixtures/strategies_minute/custom_minute_price_volume.py

@@ -0,0 +1,117 @@
+"""单分钟放量拉升，一分钟自定义策略。
+
+将本文件导入策略库，或放入 data/strategies/custom/ 后重新加载策略。
+选择分钟周期运行，输入须来自已配置分钟数据源同步的分钟 K 线。
+
+默认条件为单分钟涨幅至少 2%，且成交量至少为此前 20 根分钟均量的 3 倍。
+涨幅相对上一根收盘价，均量不含当前根。窗口须连续，不跨午休或交易日。
+只判断最新交易日每个标的最后一根已完成 K 线，当前分钟留待下一分钟评估。
+本文件提供筛选计算，自动监控提醒需由应用的分钟监控入口另行接入。
+"""
+
+from datetime import datetime, time, timedelta, timezone
+
+import polars as pl
+
+META = {
+    "id": "custom_minute_price_volume",
+    "name": "单分钟放量拉升",
+    "description": "单分钟涨幅≥2%，且分钟量≥此前20分钟均量的3倍",
+    "tags": ["分钟", "放量", "拉升"],
+    "asset_types": ["stock"],
+    "timeframes": ["1m"],
+    "basic_filter": {"enabled": False},
+    "params": [
+        {"id": "min_change_pct", "label": "单分钟涨幅下限(%)", "type": "float",
+         "default": 2.0, "min": 0.0, "max": 30.0, "step": 0.1},
+        {"id": "volume_window", "label": "均量窗口(分钟)", "type": "int",
+         "default": 20, "min": 1, "max": 120, "step": 1},
+        {"id": "volume_multiple", "label": "最低放量倍数", "type": "float",
+         "default": 3.0, "min": 1.0, "max": 100.0, "step": 0.1},
+    ],
+    "scoring": {},
+    "order_by": "minute_volume_ratio",
+    "descending": True,
+    "limit": 0,
+}
+
+EXECUTION_BACKEND = "minute_filter"
+ENTRY_SIGNALS = []
+EXIT_SIGNALS = []
+
+RULES = """
+1. 本分钟收盘价相对上一分钟收盘价上涨至少2%，使用同一价系。
+2. 本分钟成交量至少为此前20根分钟K线平均成交量的3倍，均量不含本分钟。
+3. 两项同时满足，且窗口连续、均量大于零，才输出该标的。
+4. 午休或跨日后重新积累窗口；数据不足、无效或未完成的K线不触发。
+"""
+
+
+def filter_minute_history(df: pl.DataFrame, params: dict) -> pl.DataFrame:
+    """返回每个标的最新已完成分钟的命中行，涨幅输出为小数比例。"""
+    min_change_pct = float(params.get("min_change_pct", 2.0))
+    window = int(params.get("volume_window", 20))
+    volume_multiple = float(params.get("volume_multiple", 3.0))
+    if not (0 <= min_change_pct <= 30 and 1 <= window <= 120 and 1 <= volume_multiple <= 100):
+        raise ValueError("参数范围无效，请检查涨幅、均量窗口和放量倍数")
+
+    empty = pl.DataFrame(schema={
+        "symbol": pl.Utf8, "close": pl.Float64, "last_datetime": pl.Datetime,
+        "minute_change_pct": pl.Float64, "minute_volume_ratio": pl.Float64,
+        "previous_mean_volume": pl.Float64,
+    })
+    if df.is_empty():
+        return empty
+    if not {"symbol", "datetime", "close", "volume"}.issubset(df.columns):
+        raise ValueError("分钟K线缺少 symbol/datetime/close/volume 必需列")
+
+    timestamp = pl.col("datetime")
+    if df.schema["datetime"].time_zone is not None:
+        timestamp = timestamp.dt.convert_time_zone("Asia/Shanghai").dt.replace_time_zone(None)
+    cutoff = datetime.now(timezone(timedelta(hours=8))).replace(
+        tzinfo=None, second=0, microsecond=0,
+    )
+    bars = df.select(
+        pl.col("symbol").cast(pl.Utf8), timestamp.alias("datetime"),
+        pl.col("close").cast(pl.Float64, strict=False),
+        pl.col("volume").cast(pl.Float64, strict=False),
+    ).drop_nulls(["symbol", "datetime"]).filter(pl.col("datetime") < cutoff)
+    bar_time = pl.col("datetime").dt.time()
+    bars = bars.filter(
+        bar_time.is_between(time(9, 30), time(11, 30))
+        | bar_time.is_between(time(13, 0), time(15, 0))
+    )
+    if bars.is_empty():
+        return empty
+
+    bars = bars.with_columns(
+        pl.col("datetime").dt.date().alias("date"),
+        (pl.col("datetime").dt.hour() >= 13).alias("session"),
+        pl.when(pl.col("volume").is_finite() & (pl.col("volume") >= 0))
+        .then(pl.col("volume")).otherwise(None).alias("volume"),
+    ).filter(pl.col("date") == pl.col("date").max())
+    bars = bars.unique(subset=["symbol", "datetime"], keep="last").sort(["symbol", "datetime"])
+    groups = ["symbol", "date", "session"]
+    bars = bars.with_columns(
+        pl.col("close").shift(1).over(groups).alias("previous_close"),
+        pl.col("volume").shift(1).rolling_mean(window, min_samples=window)
+        .over(groups).alias("previous_mean_volume"),
+        pl.col("datetime").shift(window).over(groups).alias("window_start"),
+    )
+
+    # 先取最新根再筛选，避免把早先已失效的命中保留在当前结果中。
+    latest = bars.group_by("symbol", maintain_order=True).tail(1)
+    matched = latest.filter(
+        pl.col("close").is_finite() & (pl.col("close") > 0)
+        & pl.col("previous_close").is_finite() & (pl.col("previous_close") > 0)
+        & (pl.col("previous_mean_volume") > 0)
+        & (pl.col("datetime") - pl.col("window_start") == pl.duration(minutes=window))
+        & (pl.col("close") >= pl.col("previous_close") * (1 + min_change_pct / 100))
+        & (pl.col("volume") >= pl.col("previous_mean_volume") * volume_multiple)
+    )
+    return matched.select(
+        "symbol", "close", pl.col("datetime").alias("last_datetime"),
+        (pl.col("close") / pl.col("previous_close") - 1).alias("minute_change_pct"),
+        (pl.col("volume") / pl.col("previous_mean_volume")).alias("minute_volume_ratio"),
+        "previous_mean_volume",
+    )

backend/tests/test_minute_price_volume_strategy.py

@@ -0,0 +1,247 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+from datetime import UTC, date, datetime, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+
+import polars as pl
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.api import definitions as definitions_api
+from app.api.strategy import StrategyCodeSaveRequest, save_strategy_code_core
+from app.config import settings
+from app.strategy.ai_generator import AIStrategyGenerator
+from app.strategy.engine import StrategyDataContext, StrategyEngine
+
+STRATEGY_PATH = (
+    Path(__file__).parent / "fixtures" / "strategies_minute"
+    / "custom_minute_price_volume.py"
+)
+START = datetime(2026, 10, 20, 9, 31)
+
+
+@pytest.fixture
+def strategy():
+    spec = importlib.util.spec_from_file_location("minute_price_volume", STRATEGY_PATH)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _bars(
+    *,
+    symbol="600001.SH",
+    start=START,
+    count=21,
+    last_close=10.2,
+    last_volume=300.0,
+):
+    return pl.DataFrame({
+        "symbol": [symbol] * count,
+        "datetime": [start + timedelta(minutes=i) for i in range(count)],
+        "open": [10.0] * count,
+        "high": [max(10.0, last_close)] * count,
+        "low": [min(10.0, last_close)] * count,
+        "close": [10.0] * (count - 1) + [last_close],
+        "volume": [100.0] * (count - 1) + [last_volume],
+        "amount": [100_000.0] * count,
+    })
+
+
+def test_exact_thresholds_and_previous_twenty_bar_baseline(strategy):
+    result = strategy.filter_minute_history(_bars(), {})
+
+    assert result["symbol"].to_list() == ["600001.SH"]
+    row = result.row(0, named=True)
+    assert row["minute_change_pct"] == pytest.approx(0.02)
+    assert row["minute_volume_ratio"] == pytest.approx(3.0)
+    assert row["previous_mean_volume"] == 100.0
+    assert row["last_datetime"] == START + timedelta(minutes=20)
+
+
+@pytest.mark.parametrize(
+    "last_close,last_volume", [(10.19, 300.0), (10.2, 299.0), (10.19, 299.0)],
+)
+def test_both_conditions_are_required(strategy, last_close, last_volume):
+    bars = _bars(last_close=last_close, last_volume=last_volume)
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+def test_baseline_is_twenty_bar_average_not_previous_minute(strategy):
+    bars = _bars(last_close=10.25, last_volume=600.0).with_columns(
+        pl.when(pl.col("datetime") == START + timedelta(minutes=19))
+        .then(300.0).otherwise(pl.col("volume")).alias("volume")
+    )
+    row = strategy.filter_minute_history(bars, {}).row(0, named=True)
+
+    assert row["previous_mean_volume"] == 110.0
+    assert row["minute_volume_ratio"] == pytest.approx(600.0 / 110.0)
+
+
+def test_parameters_can_be_adjusted(strategy):
+    params = {"min_change_pct": 1.0, "volume_window": 5, "volume_multiple": 2.0}
+    bars = _bars(count=6, last_close=10.1, last_volume=200.0)
+    assert strategy.filter_minute_history(bars, params).height == 1
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+def test_only_latest_bar_can_match(strategy):
+    bars = pl.concat([
+        _bars(),
+        _bars(start=START + timedelta(minutes=21), count=1, last_close=10.2, last_volume=100.0),
+    ])
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+def test_latest_day_and_symbols_are_isolated(strategy):
+    bars = pl.concat([
+        _bars(start=START - timedelta(days=1)),
+        _bars(count=10),
+        _bars(symbol="600002.SH"),
+        _bars(symbol="600003.SH", last_volume=299.0),
+    ]).reverse()
+    result = strategy.filter_minute_history(bars, {})
+    assert result["symbol"].to_list() == ["600002.SH"]
+
+
+def test_window_restarts_after_lunch(strategy):
+    bars = pl.concat([
+        _bars(start=datetime(2026, 10, 20, 11, 10)),
+        _bars(start=datetime(2026, 10, 20, 13, 0), count=10),
+    ])
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+def test_missing_minute_does_not_extend_the_baseline(strategy):
+    bars = _bars(count=22).filter(pl.col("datetime") != START + timedelta(minutes=10))
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+def test_duplicate_minutes_do_not_change_the_baseline(strategy):
+    bars = _bars()
+    result = strategy.filter_minute_history(pl.concat([bars, bars]), {})
+    assert result.height == 1
+    assert result["previous_mean_volume"][0] == 100.0
+
+
+@pytest.mark.parametrize("bad_value", [None, -1.0, float("nan"), float("inf")])
+def test_invalid_historical_volume_fails_closed(strategy, bad_value):
+    bars = _bars().with_columns(
+        pl.when(pl.col("datetime") == START + timedelta(minutes=10))
+        .then(pl.lit(bad_value, dtype=pl.Float64))
+        .otherwise(pl.col("volume")).alias("volume")
+    )
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+@pytest.mark.parametrize("column", ["close", "volume"])
+@pytest.mark.parametrize("bad_value", [None, -1.0, float("nan"), float("inf")])
+def test_invalid_latest_price_or_volume_fails_closed(strategy, column, bad_value):
+    bars = _bars().with_columns(
+        pl.when(pl.col("datetime") == START + timedelta(minutes=20))
+        .then(pl.lit(bad_value, dtype=pl.Float64))
+        .otherwise(pl.col(column)).alias(column)
+    )
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+def test_zero_baseline_fails_closed(strategy):
+    bars = _bars().with_columns(
+        pl.when(pl.col("datetime") < START + timedelta(minutes=20))
+        .then(0.0).otherwise(pl.col("volume")).alias("volume")
+    )
+    assert strategy.filter_minute_history(bars, {}).is_empty()
+
+
+def test_empty_input_preserves_output_schema(strategy):
+    result = strategy.filter_minute_history(_bars().head(0), {})
+    assert result.is_empty()
+    assert {"symbol", "close", "last_datetime", "minute_change_pct"} <= set(result.columns)
+
+
+def test_current_minute_is_excluded_and_timezone_is_beijing(strategy, monkeypatch):
+    class FrozenDateTime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return datetime(2026, 10, 20, 1, 51, 30, tzinfo=UTC).astimezone(tz)
+
+    monkeypatch.setattr(strategy, "datetime", FrozenDateTime)
+    assert strategy.filter_minute_history(_bars(), {}).is_empty()
+
+    # 09:51 这一根留待下一分钟评估，上一根完成线满足时仍可正常命中。
+    earlier = _bars(start=START - timedelta(minutes=1))
+    result = strategy.filter_minute_history(earlier, {})
+    assert result.height == 1
+    assert result["last_datetime"][0].hour == 9
+
+
+def test_timezone_aware_data_is_normalized(strategy):
+    bars = _bars().with_columns(
+        pl.col("datetime").dt.replace_time_zone("Asia/Shanghai").dt.convert_time_zone("UTC")
+    )
+    result = strategy.filter_minute_history(bars, {})
+    assert result.height == 1
+    assert result["last_datetime"][0] == START + timedelta(minutes=20)
+
+
+@pytest.mark.parametrize(
+    "params", [{"volume_window": 0}, {"volume_window": 121},
+               {"min_change_pct": -1}, {"volume_multiple": 0}],
+)
+def test_invalid_parameters_are_rejected(strategy, params):
+    with pytest.raises(ValueError):
+        strategy.filter_minute_history(_bars(), params)
+
+
+def test_import_save_and_engine_execution(tmp_path):
+    code = STRATEGY_PATH.read_text(encoding="utf-8")
+    AIStrategyGenerator._validate_safety(code)
+    engine = StrategyEngine([tmp_path / "strategies" / "custom"])
+    saved = save_strategy_code_core(
+        StrategyCodeSaveRequest(
+            strategy_id="custom_minute_price_volume", code=code,
+            target_source="custom", mode="create",
+        ), engine, tmp_path,
+    )
+    assert saved["ok"] is True
+    context = StrategyDataContext(
+        asset_type="stock", timeframe="1m", as_of=date(2026, 10, 20),
+        current=pl.DataFrame({"symbol": ["600001.SH"], "name": ["测试标的"]}),
+        history=_bars(),
+    )
+    result = engine.run("custom_minute_price_volume", context)
+    assert result.total == 1
+    assert result.rows[0]["name"] == "测试标的"
+    assert result.rows[0]["minute_volume_ratio"] == 3.0
+
+
+def test_py_file_import_through_resource_api(tmp_path, monkeypatch):
+    monkeypatch.setattr(settings, "data_dir", tmp_path)
+    engine = StrategyEngine([tmp_path / "strategies" / "custom"])
+    app = FastAPI()
+    app.state.repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path))
+    app.state.strategy_engine = engine
+    app.include_router(definitions_api.router)
+    files = [("files", (STRATEGY_PATH.name, STRATEGY_PATH.read_bytes(), "text/x-python"))]
+
+    with TestClient(app) as client:
+        response = client.post("/api/definitions/resources/inspect", files=files)
+        assert response.status_code == 200, response.text
+        preview = response.json()
+        assert preview["roots"] == ["strategy:custom_minute_price_volume"]
+        assert preview["entries"][0]["issues"] == []
+
+        response = client.post(
+            "/api/definitions/resources/apply", files=files,
+            data={"selected": json.dumps(preview["roots"]), "revision": preview["revision"]},
+        )
+        assert response.status_code == 200, response.text
+        assert response.json()["imported"] == 1
+
+    imported = tmp_path / "strategies" / "custom" / STRATEGY_PATH.name
+    assert imported.read_bytes() == STRATEGY_PATH.read_bytes()
+    assert engine.get("custom_minute_price_volume").execution_backend == "minute_filter"