第九节：量化投资应用：模拟场景、策略与回测

from pathlib import Path
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

data_path = Path("data/hf_600519_5min_2024.csv")
bars = pd.read_csv(data_path, parse_dates=["datetime"])
bars = bars.sort_values("datetime").reset_index(drop=True)

print("数据形状：", bars.shape)
print("日期范围：", bars["date"].min(), "至", bars["date"].max())
print("股票：", bars["ticker"].unique())
print("收盘价是否全部为正：", (bars["close"] > 0).all())
print(bars[["datetime", "open", "high", "low", "close", "volume"]].head(3))

fig, ax = plt.subplots(figsize=(11, 4.8))
ax.plot(bars["datetime"], bars["close"], color="#1f6b65", linewidth=1)
ax.set(title="600519.SH：课程样本的 5 分钟收盘价", xlabel="日期", ylabel="价格（元）")
ax.grid(alpha=0.25)
fig.tight_layout()
plt.show()

bars["bar_return"] = bars["close"].pct_change().fillna(0.0)
bars["buyhold_nav"] = (1 + bars["bar_return"]).cumprod()

print("买入并持有期末净值：", round(bars["buyhold_nav"].iloc[-1], 4))
print("买入并持有样本区间收益：",
      f"{(bars['buyhold_nav'].iloc[-1] - 1):.2%}")

# 6 个 5 分钟 bar = 30 分钟；24 个 bar = 120 分钟
bars["sma_fast"] = bars["close"].rolling(6, min_periods=6).mean()
bars["sma_slow"] = bars["close"].rolling(24, min_periods=24).mean()
bars["trend_signal"] = (bars["sma_fast"] > bars["sma_slow"]).astype(int)

# shift(1) 是避免“用收盘价在同一收盘价交易”的关键
bars["trend_position"] = bars["trend_signal"].shift(1).fillna(0)
bars["trend_turnover"] = bars["trend_position"].diff().abs().fillna(bars["trend_position"])

print(bars[["datetime", "close", "sma_fast", "sma_slow",
            "trend_signal", "trend_position"]].iloc[22:28].round(3))

# 每个交易日从头累计，不能把昨天的成交额带到今天
bars["cum_amount"] = bars.groupby("date")["amount"].cumsum()
bars["cum_volume"] = bars.groupby("date")["volume"].cumsum()
bars["vwap"] = bars["cum_amount"] / bars["cum_volume"].replace(0, np.nan)

bars["vwap_gap"] = bars["close"] - bars["vwap"]
bars["gap_std"] = bars.groupby("date")["vwap_gap"].transform(
    lambda x: x.rolling(12, min_periods=12).std()
)
bars["zscore"] = bars["vwap_gap"] / bars["gap_std"]
bars["meanrev_signal"] = (bars["zscore"] < -1.0).astype(int)
bars["meanrev_position"] = bars["meanrev_signal"].shift(1).fillna(0)
bars["meanrev_turnover"] = bars["meanrev_position"].diff().abs().fillna(bars["meanrev_position"])

print(bars[["datetime", "close", "vwap", "zscore",
            "meanrev_signal", "meanrev_position"]].dropna().head(4).round(3))

ONE_WAY_COST = 0.0005  # 单边 5bp，只用于本节演示

for strategy in ["trend", "meanrev"]:
    bars[f"{strategy}_net_return"] = (
        bars[f"{strategy}_position"] * bars["bar_return"]
        - bars[f"{strategy}_turnover"] * ONE_WAY_COST
    )
    bars[f"{strategy}_nav"] = (1 + bars[f"{strategy}_net_return"]).cumprod()

print("趋势策略交易次数：", int(bars["trend_turnover"].sum()))
print("VWAP 策略交易次数：", int(bars["meanrev_turnover"].sum()))
print("趋势策略期末净值：", round(bars["trend_nav"].iloc[-1], 4))
print("VWAP 策略期末净值：", round(bars["meanrev_nav"].iloc[-1], 4))

def daily_returns(bar_returns, dates):
    """把同一交易日的 5 分钟收益复利为一个日收益。"""
    return pd.DataFrame({"date": dates, "return": bar_returns}).groupby("date")["return"].apply(
        lambda x: (1 + x).prod() - 1
    )

def performance_report(nav, bar_returns, dates):
    daily = daily_returns(bar_returns, dates)
    annual_return = (1 + daily).prod() ** (252 / len(daily)) - 1
    annual_vol = daily.std(ddof=1) * np.sqrt(252)
    max_drawdown = (nav / nav.cummax() - 1).min()
    return {
        "样本区间收益": nav.iloc[-1] - 1,
        "年化收益": annual_return,
        "年化波动率": annual_vol,
        "夏普比率（rf=0）": annual_return / annual_vol,
        "最大回撤": max_drawdown,
    }

reports = {
    "买入并持有": performance_report(bars["buyhold_nav"], bars["bar_return"], bars["date"]),
    "趋势跟随（6/24）": performance_report(bars["trend_nav"], bars["trend_net_return"], bars["date"]),
    "VWAP 均值回归": performance_report(bars["meanrev_nav"], bars["meanrev_net_return"], bars["date"]),
}
report = pd.DataFrame(reports).T
print(report.applymap(lambda x: f"{x:.2%}"))

fig, ax = plt.subplots(figsize=(11, 5.5))
for column, label, color in [
    ("buyhold_nav", "买入并持有", "#71808a"),
    ("trend_nav", "趋势跟随（6/24）", "#1f6b65"),
    ("meanrev_nav", "VWAP 均值回归", "#a66a3d"),
]:
    ax.plot(bars["datetime"], bars[column], label=label, color=color, linewidth=2)
ax.set(title="含单边 5bp 成本后的策略净值", xlabel="日期", ylabel="初始净值 = 1")
ax.legend(); ax.grid(alpha=0.25)
fig.tight_layout(); plt.show()

fig, ax = plt.subplots(figsize=(11, 5.5))
for column, label, color in [
    ("buyhold_nav", "买入并持有", "#71808a"),
    ("trend_nav", "趋势跟随（6/24）", "#1f6b65"),
    ("meanrev_nav", "VWAP 均值回归", "#a66a3d"),
]:
    drawdown = bars[column] / bars[column].cummax() - 1
    ax.plot(bars["datetime"], drawdown, label=label, color=color, linewidth=1.8)
ax.axhline(0, color="#71808a", linewidth=0.8)
ax.set(title="从各自历史峰值计算的回撤", xlabel="日期", ylabel="回撤")
ax.legend(); ax.grid(alpha=0.25)
fig.tight_layout(); plt.show()