第 35 篇：止盈止损与风控规则设计

第一层 · 入场前：仓位控制（单笔风险预算，第 48 篇展开）
第二层 · 持仓中：止损/止盈/移动止损（本篇主线）
第三层 · 时间上：时间止损 + 组合层面的回撤熔断（第 48 篇）

import numpy as np
import pandas as pd

# ---------- 1. 数据与指标：akshare 优先，本地 fallback ----------
def get_data():
    try:
        import akshare as ak
        df = ak.stock_zh_a_hist(symbol="600519", period="daily",
                                start_date="20200101", end_date="20241231",
                                adjust="qfq")
        df = df.rename(columns={"日期": "date", "收盘": "close",
                                "最高": "high", "最低": "low"})
        df["date"] = pd.to_datetime(df["date"])
        return df.set_index("date")[["close", "high", "low"]]
    except Exception:
        rng = np.random.default_rng(35)
        n = 1200
        ret = np.concatenate([rng.normal(0.001, 0.012, 400),
                              rng.normal(-0.004, 0.025, 300),  # 造一段深回撤
                              rng.normal(0.0005, 0.015, 500)])
        px = 100 * np.exp(np.cumsum(ret))
        hi = px * (1 + np.abs(rng.normal(0, 0.006, n)))
        lo = px * (1 - np.abs(rng.normal(0, 0.006, n)))
        idx = pd.bdate_range("2020-01-01", periods=n)
        return pd.DataFrame({"close": px, "high": hi, "low": lo}, index=idx)

df = get_data()
df["atr"] = (df[["high", "low", "close"]].apply(
    lambda r: max(r["high"] - r["low"],
                  abs(r["high"] - r["close"].shift()),
                  abs(r["low"] - r["close"].shift())), axis=1)
    .rolling(14).mean())                          # 14日ATR（第14篇口径）
df["ma_fast"] = df["close"].rolling(5).mean()
df["ma_slow"] = df["close"].rolling(20).mean()

# ---------- 2. 带风控的事件驱动回测器 ----------
class RiskBacktester:
    """双均线入场 + 四类退出规则（可开关）的事件驱动回测"""
    def __init__(self, df, fee=0.001):
        self.df, self.fee = df, fee

    def run(self, use_fixed=False, fixed_pct=0.08,
            use_atr=True, atr_k=2.5,
            use_trail=True, trail_pct=0.10,
            use_time=True, max_days=40):
        df = self.df.dropna().reset_index()
        cash, shares = 100000.0, 0
        entry_px = entry_day = stop_line = peak = None
        records = []
        for i in range(1, len(df)):
            row, prev = df.iloc[i], df.iloc[i - 1]
            # --- 持仓时先检查退出（用当日收盘价，教学简化） ---
            if shares > 0:
                exit_reason = None
                if use_fixed and row["close"] < entry_px * (1 - fixed_pct):
                    exit_reason = "固定止损"
                if use_atr and row["close"] < entry_px - atr_k * row["atr"]:
                    exit_reason = exit_reason or "ATR止损"
                if use_trail:
                    peak = max(peak, row["close"])
                    if row["close"] < peak * (1 - trail_pct):
                        exit_reason = exit_reason or "移动止盈"
                if use_time and (i - entry_day) >= max_days:
                    exit_reason = exit_reason or "时间止损"
                if exit_reason:
                    cash += shares * row["close"] * (1 - self.fee)
                    records.append((row["date"], exit_reason,
                                    row["close"] / entry_px - 1))
                    shares, stop_line, peak = 0, None, None
                    continue
            # --- 空仓时检查入场（均线金叉，用昨收判断今收执行） ---
            if shares == 0 and prev["ma_fast"] > prev["ma_slow"]:
                shares = int(cash * (1 - self.fee) / row["close"] / 100) * 100
                if shares > 0:
                    cash -= shares * row["close"] * (1 + self.fee)
                    entry_px, entry_day, peak = row["close"], i, row["close"]
        nav = cash + shares * df.iloc[-1]["close"]
        return nav / 100000.0 - 1, records

# ---------- 3. 实验对比：裸奔 vs 各种风控组合 ----------
bt = RiskBacktester(df)
configs = [
    ("无风控",        dict(use_fixed=False, use_atr=False,
                           use_trail=False, use_time=False)),
    ("仅固定止损8%",  dict(use_fixed=True, use_atr=False,
                           use_trail=False, use_time=False)),
    ("仅ATR止损",     dict(use_fixed=False, use_atr=True,
                           use_trail=False, use_time=False)),
    ("ATR+移动止盈",  dict(use_fixed=False, use_atr=True,
                           use_trail=True, use_time=False)),
    ("全开",          dict(use_fixed=False, use_atr=True,
                           use_trail=True, use_time=True)),
]
for name, cfg in configs:
    total, recs = bt.run(**cfg)
    exit_count = pd.Series([r[1] for r in recs]).value_counts().to_dict() \
        if recs else {}
    print(f"{name}: 总收益(示意) {total:+.1%}, 交易{len(recs)}笔, "
          f"退出分布 {exit_count}")

示例输出（示意，模拟行情结果）:
无风控: 总收益(示意) -12.4%, 交易11笔, 退出分布 {}
仅固定止损8%: 总收益(示意) -4.8%, 交易19笔, 退出分布 {'固定止损': 19}
仅ATR止损: 总收益(示意) -1.2%, 交易16笔, 退出分布 {'ATR止损': 16}
ATR+移动止盈: 总收益(示意) +6.9%, 交易13笔, 退出分布 {'ATR止损': 7, '移动止盈': 6}
全开: 总收益(示意) +8.3%, 交易21笔, 退出分布 {'ATR止损': 6, '移动止盈': 5, '时间止损': 10}