A股投资入门到量化选股

Appearance

均线交叉策略 ​

均线交叉（Moving Average Crossover）是量化交易中最经典、最基础的策略。虽然简单，但它包含了趋势跟踪策略的所有核心要素：趋势识别、信号生成、持仓管理。正因为简单，它的每一个环节都清晰可理解，是学习量化策略开发的最佳起点。

本文将带你完整实现一个均线交叉策略，从数据获取到绩效评估，并进行参数对比实验。建议你在阅读的同时，在自己的 Python 环境中运行每一行代码。

策略原理 ​

什么是均线交叉 ​

均线交叉策略的核心逻辑只有一句话：当短期均线上穿长期均线时买入（金叉），当短期均线下穿长期均线时卖出（死叉）。

这背后的经济学直觉是：

  • 短期均线反映近期价格趋势，长期均线反映中长期趋势
  • 短期均线上穿长期均线（金叉），说明短期趋势开始强于长期趋势，价格可能进入上升通道
  • 短期均线下穿长期均线（死叉），说明短期趋势开始弱于长期趋势，价格可能进入下降通道

策略参数 ​

均线交叉策略只有两个参数：

参数含义常见取值
fast_period短期均线周期5、10、20
slow_period长期均线周期20、30、60、120

参数选择的原则：fast_period < slow_period，且两者差距不能太小（否则信号太频繁）也不能太大（否则信号太滞后）。

第一步：数据获取 ​

我们以沪深300成分股为例，获取足够长的历史数据来确保回测的统计意义：

python
import akshare as ak
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# 获取沪深300ETF（510300）的日线数据
# 使用ETF而非个股，避免个股特异性和退市风险
df = ak.fund_etf_hist_em(
    symbol="510300",
    period="daily",
    start_date="20180101",
    end_date="20241231",
    adjust="qfq"
)

df = df.rename(columns={
    "日期": "Date", "开盘": "Open", "收盘": "Close",
    "最高": "High", "最低": "Low", "成交量": "Volume"
})
df["Date"] = pd.to_datetime(df["Date"])
df = df.set_index("Date")
df = df[["Open", "High", "Low", "Close", "Volume"]]

print(f"数据范围: {df.index[0].date()} ~ {df.index[-1].date()}")
print(f"总交易日: {len(df)}")
print(f"起止价格: {df['Close'].iloc[0]:.3f} → {df['Close'].iloc[-1]:.3f}")
print(df.describe())

为什么用 ETF 而不是个股？ ETF 代表一篮子股票的表现，能有效分散个股风险。沪深300ETF 跟踪沪深300指数，涵盖A股市值最大的300家公司，是A股市场的核心宽基指数。

第二步：计算均线 ​

python
def calc_moving_averages(df, fast_period, slow_period):
    """
    计算短期和长期移动均线

    参数:
        df: DataFrame，包含 Close 列
        fast_period: int，短期均线周期
        slow_period: int，长期均线周期

    返回:
        DataFrame，新增 MA_fast 和 MA_slow 列
    """
    df = df.copy()
    df["MA_fast"] = df["Close"].rolling(window=fast_period).mean()
    df["MA_slow"] = df["Close"].rolling(window=slow_period).mean()
    return df

# 计算 10日 和 30日 均线
df = calc_moving_averages(df, fast_period=10, slow_period=30)
print(df[["Close", "MA_fast", "MA_slow"]].tail(10))

可视化均线 ​

python
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

fig, ax = plt.subplots(figsize=(14, 6))
ax.plot(df.index, df["Close"], color="gray", alpha=0.5, label="收盘价")
ax.plot(df.index, df["MA_fast"], color="red", linewidth=1.2, label="MA10（短期）")
ax.plot(df.index, df["MA_slow"], color="blue", linewidth=1.2, label="MA30（长期）")
ax.set_title("沪深300ETF - 均线走势")
ax.set_ylabel("价格（元）")
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

第三步：生成买卖信号 ​

信号生成的核心逻辑：检测均线交叉点。

python
def generate_signals(df):
    """
    根据均线交叉生成买卖信号

    信号定义:
        1 = 金叉买入（短期均线从下方穿越长期均线）
        -1 = 死叉卖出（短期均线从上方穿越长期均线）
        0 = 无信号

    返回:
        DataFrame，新增 signal 列
    """
    df = df.copy()

    # 判断当日短期均线与长期均线的关系
    df["fast_above_slow"] = df["MA_fast"] > df["MA_slow"]

    # 检测交叉：当日关系与前一日关系不同
    df["crossover"] = df["fast_above_slow"] != df["fast_above_slow"].shift(1)

    # 金叉：从"短期<长期"变为"短期>长期"
    df["signal"] = 0
    df.loc[
        (df["fast_above_slow"] == True) &
        (df["fast_above_slow"].shift(1) == False),
        "signal"
    ] = 1  # 金叉

    # 死叉：从"短期>长期"变为"短期<长期"
    df.loc[
        (df["fast_above_slow"] == False) &
        (df["fast_above_slow"].shift(1) == True),
        "signal"
    ] = -1  # 死叉

    # 移除均线未计算完成的初始行
    df = df.dropna(subset=["MA_fast", "MA_slow"])

    return df

df = generate_signals(df)

# 查看信号统计
buy_count = (df["signal"] == 1).sum()
sell_count = (df["signal"] == -1).sum()
print(f"金叉买入信号: {buy_count} 次")
print(f"死叉卖出信号: {sell_count} 次")
print(f"\n最近5次信号:")
print(df[df["signal"] != 0][["Close", "MA_fast", "MA_slow", "signal"]].tail(5))

理解信号生成 ​

需要注意一个重要细节：信号在收盘时确认，交易在次日开盘时执行。 这是因为均线是用收盘价计算的，只有收盘后才能确定当天是否发生了交叉。这种"当日确认、次日执行"的方式更接近实际交易。

第四步：回测 ​

回测是用历史数据模拟策略执行，计算策略的实际收益和风险指标。

完整回测引擎 ​

python
def backtest(df, initial_capital=100000, commission_rate=0.0003, slippage=0.001):
    """
    均线交叉策略回测引擎

    参数:
        df: DataFrame，包含信号和价格数据
        initial_capital: float，初始资金
        commission_rate: float，佣金费率（双边），默认万三
        slippage: float，滑点比例，默认千分之一

    返回:
        dict，包含交易记录、净值曲线和绩效指标
    """
    capital = initial_capital
    position = 0        # 持仓数量（股数）
    entry_price = 0     # 买入价格
    trades = []          # 交易记录
    portfolio_values = [] # 每日净值

    for i in range(len(df)):
        date = df.index[i]
        close = df["Close"].iloc[i]
        signal = df["signal"].iloc[i]

        # 获取次日开盘价作为成交价
        # 这里简化处理：用当日收盘价近似
        # 实际中应使用次日开盘价
        if i + 1 < len(df):
            next_open = df["Open"].iloc[i + 1]
        else:
            next_open = close

        # 执行交易
        if signal == 1 and position == 0:
            # 金叉买入
            buy_price = next_open * (1 + slippage)
            shares = int(capital / (buy_price * (1 + commission_rate)))
            shares = shares // 100 * 100  # 凑整百

            if shares > 0:
                cost = shares * buy_price * (1 + commission_rate)
                capital -= cost
                position = shares
                entry_price = buy_price

                trades.append({
                    "date": date,
                    "action": "BUY",
                    "price": round(buy_price, 3),
                    "shares": shares,
                    "cost": round(cost, 2)
                })

        elif signal == -1 and position > 0:
            # 死叉卖出
            sell_price = next_open * (1 - slippage)
            revenue = position * sell_price * (1 - commission_rate)
            profit = revenue - position * entry_price
            profit_rate = profit / (position * entry_price)

            capital += revenue

            trades.append({
                "date": date,
                "action": "SELL",
                "price": round(sell_price, 3),
                "shares": position,
                "revenue": round(revenue, 2),
                "profit": round(profit, 2),
                "profit_rate": round(profit_rate * 100, 2)
            })

            position = 0

        # 记录每日净值
        portfolio_value = capital + position * close
        portfolio_values.append({
            "date": date,
            "portfolio_value": portfolio_value
        })

    # 计算绩效
    values_df = pd.DataFrame(portfolio_values).set_index("date")
    final_value = values_df["portfolio_value"].iloc[-1]

    return {
        "trades": trades,
        "values": values_df,
        "final_value": final_value,
        "total_return": (final_value / initial_capital - 1) * 100
    }

# 运行回测
result = backtest(df)
print(f"初始资金: 100,000 元")
print(f"最终资产: {result['final_value']:,.0f} 元")
print(f"总收益率: {result['total_return']:.2f}%")
print(f"交易次数: {len(result['trades'])} 笔")

第五步：绩效评估 ​

核心指标计算 ​

python
def evaluate_performance(result, initial_capital=100000):
    """
    计算策略的综合绩效指标

    参数:
        result: backtest 函数的返回值
        initial_capital: 初始资金

    返回:
        dict，各项绩效指标
    """
    values = result["values"]["portfolio_value"]

    # 计算每日收益率
    daily_returns = values.pct_change().dropna()

    # 总收益率
    total_return = (values.iloc[-1] / initial_capital - 1)

    # 年化收益率
    trading_days = len(values)
    years = trading_days / 252  # 一年约252个交易日
    annual_return = (1 + total_return) ** (1 / years) - 1

    # 最大回撤
    cummax = values.cummax()
    drawdown = (values - cummax) / cummax
    max_drawdown = drawdown.min()

    # 夏普比率（假设无风险利率为3%）
    risk_free_rate = 0.03 / 252  # 日化无风险利率
    excess_returns = daily_returns - risk_free_rate
    sharpe_ratio = np.sqrt(252) * excess_returns.mean() / excess_returns.std() if excess_returns.std() > 0 else 0

    # 交易统计
    trades = result["trades"]
    sell_trades = [t for t in trades if t["action"] == "SELL"]
    win_trades = [t for t in sell_trades if t.get("profit", 0) > 0]

    win_rate = len(win_trades) / len(sell_trades) if sell_trades else 0

    profits = [t["profit_rate"] for t in sell_trades if t.get("profit_rate") is not None]
    avg_profit = np.mean(profits) if profits else 0
    avg_win = np.mean([p for p in profits if p > 0]) if any(p > 0 for p in profits) else 0
    avg_loss = abs(np.mean([p for p in profits if p < 0])) if any(p < 0 for p in profits) else 0
    profit_loss_ratio = avg_win / avg_loss if avg_loss > 0 else float('inf')

    metrics = {
        "总收益率": f"{total_return * 100:.2f}%",
        "年化收益率": f"{annual_return * 100:.2f}%",
        "最大回撤": f"{max_drawdown * 100:.2f}%",
        "夏普比率": f"{sharpe_ratio:.2f}",
        "交易次数（买卖对）": f"{len(sell_trades)}",
        "胜率": f"{win_rate * 100:.1f}%",
        "平均每笔收益": f"{avg_profit:.2f}%",
        "盈亏比": f"{profit_loss_ratio:.2f}",
        "回测年数": f"{years:.1f}"
    }

    return metrics

# 评估策略表现
metrics = evaluate_performance(result)
print("=== 策略绩效 ===")
for k, v in metrics.items():
    print(f"  {k}: {v}")

与基准对比 ​

python
# 计算买入持有策略的收益（作为基准）
buy_hold_return = (df["Close"].iloc[-1] / df["Close"].iloc[0] - 1) * 100
print(f"\n买入持有收益率: {buy_hold_return:.2f}%")
print(f"策略收益率: {result['total_return']:.2f}%")

if result['total_return'] > buy_hold_return:
    print("策略跑赢基准")
else:
    print("策略跑输基准 — 这并不意味着策略没有价值")

重要提醒：均线交叉策略在震荡市中表现不佳，会产生大量虚假信号（频繁买卖），每次交易都会被手续费侵蚀。它的优势在于捕捉大趋势。因此，在某些时期跑输基准是完全正常的。

第六步：参数对比实验 ​

不同参数组合对策略表现的影响很大。下面我们对比三组常用参数：

组合短期均线长期均线特点
5/205日20日信号灵敏，交易频繁
10/3010日30日均衡型，本文默认参数
20/6020日60日信号稀少，适合大趋势
python
def run_ma_crossover(df, fast_period, slow_period, initial_capital=100000):
    """
    运行完整均线交叉策略（数据获取→计算均线→生成信号→回测）
    """
    # 计算
    df_copy = calc_moving_averages(df[["Open", "High", "Low", "Close", "Volume"]].copy(),
                                    fast_period, slow_period)
    df_copy = generate_signals(df_copy)
    result = backtest(df_copy, initial_capital)
    metrics = evaluate_performance(result, initial_capital)

    return {
        "params": f"MA{fast_period}/MA{slow_period}",
        "result": result,
        "metrics": metrics
    }

# 获取原始数据
raw_df = ak.fund_etf_hist_em(
    symbol="510300", period="daily",
    start_date="20180101", end_date="20241231", adjust="qfq"
)
raw_df = raw_df.rename(columns={
    "日期": "Date", "开盘": "Open", "收盘": "Close",
    "最高": "High", "最低": "Low", "成交量": "Volume"
})
raw_df["Date"] = pd.to_datetime(raw_df["Date"])
raw_df = raw_df.set_index("Date")
raw_df = raw_df[["Open", "High", "Low", "Close", "Volume"]]

# 三组参数对比
param_sets = [
    {"fast_period": 5, "slow_period": 20},
    {"fast_period": 10, "slow_period": 30},
    {"fast_period": 20, "slow_period": 60},
]

results = []
for params in param_sets:
    r = run_ma_crossover(raw_df, **params)
    results.append(r)
    print(f"\n=== {r['params']} ===")
    for k, v in r["metrics"].items():
        print(f"  {k}: {v}")

参数对比可视化 ​

python
fig, axes = plt.subplots(2, 1, figsize=(14, 10), height_ratios=[2, 1])

# 净值曲线对比
ax1 = axes[0]
colors = ["red", "blue", "green"]

for r, color in zip(results, colors):
    ax1.plot(
        r["result"]["values"].index,
        r["result"]["values"]["portfolio_value"],
        color=color,
        linewidth=1.2,
        label=r["params"]
    )

# 添加买入持有基准线
benchmark = raw_df["Close"] / raw_df["Close"].iloc[0] * 100000
ax1.plot(benchmark.index, benchmark.values, color="gray",
         linewidth=1, linestyle="--", label="买入持有")

ax1.set_title("均线交叉策略 - 参数对比（净值曲线）", fontsize=14)
ax1.set_ylabel("资产净值（元）")
ax1.legend()
ax1.grid(True, alpha=0.3)

# 回撤对比
ax2 = axes[1]
for r, color in zip(results, colors):
    values = r["result"]["values"]["portfolio_value"]
    cummax = values.cummax()
    drawdown = (values - cummax) / cummax * 100
    ax2.fill_between(drawdown.index, drawdown.values, 0,
                     color=color, alpha=0.3, label=r["params"])

ax2.set_title("回撤对比", fontsize=12)
ax2.set_ylabel("回撤（%）")
ax2.legend()
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

不同市场环境下的表现 ​

均线交叉策略的表现高度依赖市场环境：

趋势行情（策略优势期） ​

  • 单边上涨：策略能跟上大部分涨幅，金叉后持有直到死叉才卖出
  • 单边下跌：策略能及时清仓回避，死叉信号帮助避开大部分跌幅

震荡行情（策略劣势期） ​

  • 横盘震荡：均线反复交叉，产生大量虚假信号。每次金叉买入后价格回落，死叉卖出后价格又上涨，频繁亏损
  • 这种现象被称为**"均线缠绕"**，是均线策略最大的弱点

用数据验证 ​

python
# 将回测期划分为不同市场阶段
# 这里简化为按年划分
raw_df["year"] = raw_df.index.year

print("=== 各年度策略表现 ===")
print(f"{'年份':<8} {'5/20':<12} {'10/30':<12} {'20/60':<12}")
print("-" * 44)

for year in sorted(raw_df["year"].unique()):
    year_data = raw_df[raw_df["year"] == year]
    if len(year_data) < 30:  # 数据太少跳过
        continue

    row = f"{year:<8}"
    for params in param_sets:
        r = run_ma_crossover(year_data, **params)
        row += f" {r['metrics']['总收益率']:<12}"
    print(row)

通过分年度数据，可以清晰地看到策略在不同市场环境下的表现差异。这有助于判断策略是否适合当前市场环境。

改进思路 ​

基础均线交叉策略有明显的局限性，以下是几个实用的改进方向：

1. 添加过滤器：减少虚假信号 ​

python
def generate_signals_with_filter(df, trend_period=60):
    """
    添加趋势过滤器的信号生成

    仅在长期趋势向上时才执行金叉买入
    避免在下降趋势中的反弹金叉买入
    """
    df = df.copy()

    # 计算趋势判断均线
    df["MA_trend"] = df["Close"].rolling(window=trend_period).mean()

    # 基础信号
    df = generate_signals(df)

    # 过滤：只在收盘价 > 趋势均线时才执行买入
    df.loc[
        (df["signal"] == 1) & (df["Close"] < df["MA_trend"]),
        "signal"
    ] = 0  # 取消在下跌趋势中的金叉买入信号

    return df

2. 用 EMA 替代 SMA ​

指数移动平均（EMA）比简单移动平均（SMA）对近期价格更敏感，信号更及时：

python
# 将 SMA 替换为 EMA
df["MA_fast"] = df["Close"].ewm(span=fast_period, adjust=False).mean()
df["MA_slow"] = df["Close"].ewm(span=slow_period, adjust=False).mean()

3. 加入止损机制 ​

python
# 在回测引擎中加入止损逻辑
max_loss = -0.05  # 单笔最大亏损5%

if position > 0:
    current_loss = (close - entry_price) / entry_price
    if current_loss <= max_loss:
        # 触发止损
        sell_price = close * (1 - slippage)
        revenue = position * sell_price * (1 - commission_rate)
        capital += revenue
        position = 0

4. 多时间周期确认 ​

在日线出现金叉时，检查周线是否也处于上升趋势。只有多周期共振时才执行交易：

python
# 简化示例：要求 MA20 也在上升
df["MA_slow_rising"] = df["MA_slow"] > df["MA_slow"].shift(5)

df.loc[
    (df["signal"] == 1) & (df["MA_slow_rising"] == False),
    "signal"
] = 0

5. 加入成交量确认 ​

python
# 金叉时要求成交量放大
df["vol_ma5"] = df["Volume"].rolling(5).mean()
df["vol_surge"] = df["Volume"] > df["vol_ma5"] * 1.2

df.loc[
    (df["signal"] == 1) & (df["vol_surge"] == False),
    "signal"
] = 0

以上改进可以单独使用，也可以组合使用。但要注意：每增加一个过滤条件，都会减少交易次数。如果过滤得太严格，可能会错过真正的大趋势。改进的目标是提高信号质量，而不是完全消除亏损——那是不可能的。

策略的局限性 ​

在结束之前，需要坦诚地指出均线交叉策略的几个固有局限：

  1. 滞后性：均线是基于历史价格计算的，任何基于均线的信号都必然滞后于实际价格转折点。参数越大的均线，滞后越严重。

  2. 震荡市磨损：在横盘震荡市场中，策略会产生大量虚假信号，每次交易的手续费和滑点会持续侵蚀资金。这是均线策略最大的成本。

  3. 不适用于所有股票：均线策略适合流动性好、趋势性强的标的（如宽基ETF），不适合波动剧烈的小盘股或庄股。

  4. 参数敏感性：不同参数在不同市场环境下表现差异巨大，不存在"最优参数"。今天回测最好的参数，明天可能就不好用了。

  5. 单独使用风险较高：均线交叉策略最好与其他指标（如成交量、RSI、MACD）结合使用，形成综合判断。

小结 ​

本文完整实现了均线交叉策略的开发流程：

  1. 数据获取：使用 akshare 获取沪深300ETF 的历史数据
  2. 计算均线：用 pandas 的 rolling 方法计算简单移动平均
  3. 生成信号：检测均线交叉点，金叉买入、死叉卖出
  4. 回测：模拟策略执行，考虑了手续费和滑点
  5. 绩效评估：计算总收益、最大回撤、夏普比率等核心指标
  6. 参数对比：三组参数的表现对比和可视化
  7. 改进思路：趋势过滤、EMA替代、止损机制等方向

均线交叉策略虽然简单，但它教会了我们量化策略开发的核心方法论。掌握了这套流程，你就可以将同样的方法应用到更复杂的策略上。

免责声明：本文所有代码和策略仅用于技术教学目的，不构成任何投资建议。均线交叉策略存在固有局限性，在实盘交易中可能产生亏损。投资有风险，入市需谨慎。

仅供学习交流，不构成任何投资建议