双均线与动量轮动策略实战：参数稳健性比收益率更重要

# strategies/ma_cross.py
import pandas as pd


def ma_cross_signal(close: pd.Series, fast: int = 20, slow: int = 60) -> pd.Series:
    """返回目标仓位序列：1 满仓 / 0 空仓。只在收盘后生成，次日生效。"""
    ma_fast = close.rolling(fast).mean()
    ma_slow = close.rolling(slow).mean()
    sig = (ma_fast > ma_slow).astype(float)
    # 均线未成型期间不持仓
    sig[ma_slow.isna()] = 0.0
    return sig

# strategies/run_ma.py
import pandas as pd

from backtest.engine import BacktestConfig, run_backtest
from strategies.ma_cross import ma_cross_signal


def run_ma_backtest(symbol: str, start: str, end: str,
                    fast: int = 20, slow: int = 60) -> dict:
    from src.data.loader import load_bars

    df = load_bars([symbol], start, end, adjust="qfq")
    close = df.pivot(index="trade_date", columns="symbol", values="close")
    volume = df.pivot(index="trade_date", columns="symbol", values="volume")
    pct = load_bars([symbol], start, end, adjust="none")
    pct = pct.pivot(index="trade_date", columns="symbol", values="pct_chg")

    sig = ma_cross_signal(close[symbol], fast, slow)
    target_w = pd.DataFrame({symbol: sig}, index=close.index)
    # 引擎内部会做 shift(1)：sig 在 T 日收盘生成，T+1 日承担收益
    res = run_backtest(close, volume, target_w, pct_chg=pct)
    return res.metrics

# strategies/momentum_rotation.py
from __future__ import annotations

import numpy as np
import pandas as pd


def momentum_score(close: pd.DataFrame, lookback: int = 20,
                   vol_window: int = 20) -> pd.DataFrame:
    """风险调整动量：过去 N 日收益 / 过去 N 日波动。截面排名用。"""
    ret = close / close.shift(lookback) - 1
    vol = close.pct_change().rolling(vol_window).std()
    return (ret / vol).replace([np.inf, -np.inf], np.nan)


def build_target_weights(close: pd.DataFrame, score: pd.DataFrame,
                         top_k: int = 2, rebalance: str = "M",
                         min_score: float = 0.0) -> pd.DataFrame:
    """月末选动量最强的 top_k 只等权持有，其余现金。

    rebalance='M' 表示只在每月最后一个交易日调仓，中间日权重保持不变。
    """
    idx = close.index
    if rebalance == "M":
        # 每月最后一个交易日。这里统一转成 Timestamp，避免 datetime64 与 Timestamp
        # 在 set 里哈希不一致导致"一个调仓日都没命中"的静默 bug。
        month_end = pd.Series(idx, index=idx).groupby(idx.to_period("M")).max()
        rebal_set = {pd.Timestamp(d).normalize() for d in month_end}
    else:
        rebal_set = {pd.Timestamp(d).normalize() for d in idx}

    target = pd.DataFrame(0.0, index=idx, columns=close.columns)
    last_w = pd.Series(0.0, index=close.columns)
    for date, row in score.iterrows():
        if pd.Timestamp(date).normalize() in rebal_set:
            valid = row.dropna()
            # 只买动量为正的标的，否则空仓
            valid = valid[valid > min_score].sort_values(ascending=False)
            picks = valid.head(top_k).index
            last_w = pd.Series(0.0, index=close.columns)
            if len(picks):
                last_w[picks] = 1.0 / top_k
        target.loc[date] = last_w
    return target

POOL = {
    "510300": "沪深300ETF",
    "510500": "中证500ETF",
    "159915": "创业板ETF",
    "513100": "纳指ETF",
    "518880": "黄金ETF",
    "511010": "国债ETF",
}

# scripts/grid_robustness.py
import itertools

import pandas as pd

from strategies.momentum_rotation import build_target_weights, momentum_score
from strategies.run_ma import run_ma_backtest  # noqa: F401  (示意：复用取数/引擎)


def grid_search(close: pd.DataFrame, volume: pd.DataFrame, pct: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for lookback, top_k, rebal in itertools.product([10, 20, 40, 60], [1, 2, 3], ["M", "W"]):
        score = momentum_score(close, lookback=lookback)
        w = build_target_weights(close, score, top_k=top_k, rebalance=rebal)
        res = run_backtest(close, volume, w, pct_chg=pct)
        rows.append({"lookback": lookback, "top_k": top_k, "rebal": rebal,
                     **{k: res.metrics[k] for k in ("年化收益", "夏普", "最大回撤", "年化换手")}})
    return pd.DataFrame(rows)


def robustness_table(df: pd.DataFrame) -> pd.DataFrame:
    """汇总：多少比例的参数组合夏普 > 0.5，多少 > 0.3。"""
    return pd.DataFrame({
        "夏普>0.8": [(df["夏普"] > 0.8).mean()],
        "夏普>0.5": [(df["夏普"] > 0.5).mean()],
        "夏普>0.3": [(df["夏普"] > 0.3).mean()],
        "夏普中位数": [df["夏普"].median()],
        "最差夏普": [df["夏普"].min()],
    })