策略工坊

查看内置策略源码；在「我的」中编辑/新建策略（存于 data/user_strategies/）。回测时同 id 优先用我的版本。

新策略开发指南（必读）

每个策略是一个独立的 Python 文件，放在 data/user_strategies/<id>.py，必须包含以下接口：

名称	类型	说明
`STRATEGY_ID`	str	唯一 id，仅字母/数字/下划线，不能以数字开头；须与文件名一致
`NAME`	str	展示名称
`DESCRIPTION`	str	策略逻辑简述
`PARAMS`	dict	可调参数字典，回测与信号页直接读取，不在回测页单独填表
`generate_signals(panel, params)`	函数	核心信号函数（见下）

`generate_signals` 约定

入参 panel：dict[str, pd.DataFrame]，键为标的 code，值为含 date/open/high/low/close/volume 的日线表。
入参 params：即本策略的 PARAMS（可被用户策略文件内覆盖）。
返回值：pd.Series，索引为交易日（pd.Timestamp），值为目标持仓 code 或 None（空仓）。
eval-v1 规则：T 日收盘产生信号 → T+1 开盘成交；最多持 1 只；整手 100 股。

可用工具函数（`from app.strategies import base`）

base.aligned_dates(panel) — 合并所有标的交易日，得到统一日期索引。
base.apply_hold_inertia_from_candidates(...) — 带最小持有期与换仓阈值的轮动辅助。

最小模板

from __future__ import annotations
from typing import Any
import pandas as pd
from app.strategies import base

STRATEGY_ID = "my_strategy"
NAME = "我的策略"
DESCRIPTION = "策略说明"
PARAMS: dict[str, Any] = {"window": 20}

def generate_signals(panel: dict[str, pd.DataFrame], params: dict[str, Any]) -> pd.Series:
    dates = base.aligned_dates(panel)
    result = pd.Series(index=dates, dtype=object)
    for dt in dates:
        result.loc[dt] = None  # 或某只 code
    return result

保存前可点「校验」；建议从内置策略「另存为我的」后修改。参考实现：dual_ma、breakout。

双均线轮动 dual_ma

对池内每只 ETF 计算快慢均线，收盘时在快线位于慢线上方的标的中按动量排序；带持仓惯性：未满最小持有 bar 数或领先幅度不足时不换仓；趋势失效则空仓。

来源：内置（只读）

去回测 →

from __future__ import annotations

from typing import Any

import pandas as pd

from app.strategies import base

STRATEGY_ID = "dual_ma"
NAME = "双均线轮动"
DESCRIPTION = (
    "对池内每只 ETF 计算快慢均线，收盘时在快线位于慢线上方的标的中按动量排序；"
    "带持仓惯性：未满最小持有 bar 数或领先幅度不足时不换仓；趋势失效则空仓。"
)
PARAMS: dict[str, Any] = {
    # HL Round 5：保留 20/60，min_hold 8→6 以恢复往返样本量
    "fast_window": 20,
    "slow_window": 60,
    "momentum_window": 20,
    "min_hold_bars": 6,
    "switch_margin": 0.02,
}


def generate_signals(panel: dict[str, pd.DataFrame], params: dict[str, Any]) -> pd.Series:
    fast = int(params["fast_window"])
    slow = int(params["slow_window"])
    mom = int(params["momentum_window"])
    min_hold = int(params.get("min_hold_bars", 1))
    switch_margin = float(params.get("switch_margin", 0.0))
    if fast >= slow:
        raise ValueError("fast_window 必须小于 slow_window")

    dates = base.aligned_dates(panel)
    candidates_by_date: dict[pd.Timestamp, dict[str, float]] = {dt: {} for dt in dates}

    for code, df in panel.items():
        tmp = df.set_index("date").sort_index()
        close = tmp["close"]
        fast_ma = close.rolling(fast).mean()
        slow_ma = close.rolling(slow).mean()
        momentum = close.pct_change(mom)
        for dt in dates:
            if dt not in close.index:
                continue
            if pd.isna(fast_ma.loc[dt]) or pd.isna(slow_ma.loc[dt]):
                continue
            if fast_ma.loc[dt] <= slow_ma.loc[dt]:
                continue
            m = momentum.loc[dt] if dt in momentum.index else float("nan")
            if pd.isna(m):
                continue
            candidates_by_date[dt][code] = float(m)

    return base.apply_hold_inertia_from_candidates(
        candidates_by_date,
        dates,
        min_hold_bars=min_hold,
        switch_margin=switch_margin,
    )