策略工坊

查看内置策略源码；在「我的」中编辑/新建策略（存于 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。

突破轮动 breakout

收盘时选择突破 N 日最高价且动量最强的 ETF；若无一突破则空仓。

来源：内置（只读）

去回测 →

from __future__ import annotations

from typing import Any

import pandas as pd

from app.strategies import base

STRATEGY_ID = "breakout"
NAME = "突破轮动"
DESCRIPTION = "收盘时选择突破 N 日最高价且动量最强的 ETF；若无一突破则空仓。"
PARAMS: dict[str, Any] = {
    "lookback": 20,
    "momentum_window": 10,
}


def generate_signals(panel: dict[str, pd.DataFrame], params: dict[str, Any]) -> pd.Series:
    lookback = int(params["lookback"])
    mom = int(params["momentum_window"])
    dates = base.aligned_dates(panel)
    scores: dict[pd.Timestamp, tuple[str | None, float]] = {}

    for code, df in panel.items():
        tmp = df.set_index("date").sort_index()
        close = tmp["close"]
        high_n = close.rolling(lookback).max().shift(1)
        momentum = close.pct_change(mom)
        for dt in dates:
            if dt not in close.index:
                continue
            if pd.isna(high_n.loc[dt]):
                continue
            if close.loc[dt] <= high_n.loc[dt]:
                continue
            m = momentum.loc[dt] if dt in momentum.index else float("nan")
            if pd.isna(m):
                continue
            prev = scores.get(dt)
            if prev is None or m > prev[1]:
                scores[dt] = (code, float(m))

    result = pd.Series(index=dates, dtype=object)
    for dt in dates:
        pick = scores.get(dt)
        result.loc[dt] = pick[0] if pick else None
    return result