别再定投 QQQ 了！分享一个 26 年翻 27 倍的“终极组合”策略 📈

很多开发者在投资时容易走两个极端：要么稳如老狗（全仓 QQQ），要么富贵险中求（满仓 TQQQ）。今天拆解一个兼顾“防御力”与“爆发力”的量化配方：60/30/10 核心卫星策略。

1️⃣ 为什么需要这个组合？

回测数据显示，1999-2025 年：

• 全仓 QQQ： 回报 13 倍，最大回撤 83%（互联网大崩盘期要等 15 年才回本）。
• 60/30/10 组合： 回报 27 倍，且在 2022 年大跌中，由于避险机制，回撤比 QQQ 还小！

2️⃣ 终极配方 (60/30/10)

• 60% 核心仓位 (QQQ/VOO)： Buy & Hold。确保你永远在车上，吃满科技长牛的 Beta 收益。
• 30% 卫星仓位 (QLD – 2倍杠杆)： SMA200 择时。利用杠杆在趋势确认时放大收益，这是增益的主引擎。
• 10% 彩票仓位 (TQQQ – 3倍杠杆)： SMA200 择时。作为“期权”博取阶级跨越的爆发力。

3️⃣ 核心算法：SMA200 + 7% 缓冲带

为了过滤震荡市的“噪音”，策略引入了安全气囊：

• 买入信号： 当 QQQ > SMA200 × 1.04
• 卖出信号： 当 QQQ < SMA200 × 0.97
• 原理： 只有趋势足够强劲才进场，趋势真的坏了才撤离。这 7% 的区间就是避风港。

4️⃣ 为什么它更安全？

这套策略最优雅的地方在于：当市场真崩盘时，那 40% 的杠杆仓位会迅速触发止损变为现金。现金在熊市就是最好的减震器，同时为你提供了下一轮牛市起跑时最猛的子弹。

5️⃣ 开发者视角：一键监控代码

我们不盯盘，只看信号。分享一个简单的 Python 核心逻辑实现：

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.ticker as mtick
import glob
import os
from datetime import datetime

# ================= 配置区域 =================
# 策略标的配置
CORE_TICKER = "QQQ"      # 核心仓位 (60%)
SAT_TICKER = "QLD"       # 卫星仓位 (30%)
LOT_TICKER = "TQQQ"      # 彩票仓位 (10%)

INITIAL_CASH = 100000    # 初始本金
CASH_RATE = 0.04         # 闲置资金年化收益 (40%部分空仓时)
MA_WINDOW = 200          # 均线周期
BUFFER_BUY = 1.04        # 买入阈值 (1 + 4%)
BUFFER_SELL = 0.97       # 卖出阈值 (1 - 3%)

DATA_DIR = "data"
RESULTS_BASE_DIR = "results"
# ===========================================

plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Arial Unicode MS', 'sans-serif']
plt.rcParams['axes.unicode_minus'] = False

def get_latest_file(symbol):
    patterns = [f"data_asset_{symbol}_*.csv", f"data_asset_{symbol.replace('^', '')}_*.csv"]
    for pattern in patterns:
        files = glob.glob(os.path.join(DATA_DIR, pattern))
        if files:
            files.sort()
            return files[-1]
    return None

def calculate_drawdown(equity_curve):
    rolling_max = equity_curve.cummax()
    drawdown = (equity_curve - rolling_max) / rolling_max
    max_drawdown = drawdown.min()
    return drawdown, max_drawdown

def run():
    print(f"--- 启动 60/30/10 组合回测 (信号源: {CORE_TICKER}) ---")

    # 1. 加载数据
    f_core = get_latest_file(CORE_TICKER)
    f_sat = get_latest_file(SAT_TICKER)
    f_lot = get_latest_file(LOT_TICKER)

    if not all([f_core, f_sat, f_lot]):
        print("❌ 缺少必要的数据文件，请检查 data 目录")
        return

    df_q = pd.read_csv(f_core, index_col=0, parse_dates=True).sort_index()
    df_ld = pd.read_csv(f_sat, index_col=0, parse_dates=True).sort_index()
    df_tq = pd.read_csv(f_lot, index_col=0, parse_dates=True).sort_index()

    # 计算信号源均线
    df_q['SMA'] = df_q['Close'].rolling(window=MA_WINDOW).mean()

    # 2. 初始化账户
    # 60% 始终持有 QQQ
    common_start = df_q['SMA'].dropna().index[0]
    df_q, df_ld, df_tq = df_q.loc[common_start:], df_ld.loc[common_start:], df_tq.loc[common_start:]

    core_shares = (INITIAL_CASH * 0.6) / df_q['Open'].iloc[0]
    dyn_cash = INITIAL_CASH * 0.4 # 用于卫星和彩票的40%资金

    ld_shares = 0
    tq_shares = 0
    in_market = False # 40% 杠杆部分的状态

    equity_curve = []
    trades = []
    dates = df_q.index

    # 3. 交易循环
    for i in range(len(df_q)):
        curr_date = dates[i]
        q_row = df_q.iloc[i]

        # 信号判断逻辑 (基于前一日收盘)
        if i > 0:
            prev_q = df_q.iloc[i-1]
            # 缓冲带逻辑
            if prev_q['Close'] > prev_q['SMA'] * BUFFER_BUY:
                signal = "BUY"
            elif prev_q['Close'] < prev_q['SMA'] * BUFFER_SELL:
                signal = "SELL"
            else:
                signal = "HOLD"

            # 执行交易 (今开成交)
            ld_open = df_ld.iloc[i]['Open']
            tq_open = df_tq.iloc[i]['Open']

            if signal == "BUY" and not in_market:
                # 40% 资金按 3:1 分配给 QLD 和 TQQQ
                ld_shares = (dyn_cash * 0.75) / ld_open
                tq_shares = (dyn_cash * 0.25) / tq_open
                trades.append({
                    'date': curr_date, 'action': '🟢 杠杆进场',
                    'q_sig': prev_q['Close'], 'q_sma': prev_q['SMA'],
                    'ld_p': ld_open, 'tq_p': tq_open, 'equity': core_shares * q_row['Open'] + dyn_cash
                })
                dyn_cash = 0
                in_market = True
            elif signal == "SELL" and in_market:
                # 卖出杠杆标的回归现金
                dyn_cash = (ld_shares * ld_open) + (tq_shares * tq_open)
                trades.append({
                    'date': curr_date, 'action': '🔴 杠杆撤离',
                    'q_sig': prev_q['Close'], 'q_sma': prev_q['SMA'],
                    'ld_p': ld_open, 'tq_p': tq_open, 'equity': core_shares * q_row['Open'] + dyn_cash
                })
                ld_shares = 0
                tq_shares = 0
                in_market = False

        # 计算当日净值
        current_core_val = core_shares * q_row['Close']
        if in_market:
            current_dyn_val = (ld_shares * df_ld.iloc[i]['Close']) + (tq_shares * df_tq.iloc[i]['Close'])
        else:
            dyn_cash += dyn_cash * (CASH_RATE / 252) # 空仓利息
            current_dyn_val = dyn_cash

        equity_curve.append(current_core_val + current_dyn_val)

    # 4. 统计与导出
    res = pd.DataFrame(index=dates)
    res['My_Equity'] = equity_curve
    # 基准：100% 死拿 QQQ
    hold_shares = INITIAL_CASH / df_q['Open'].iloc[0]
    res['Hold_Equity'] = hold_shares * df_q['Close']

    res['My_Drawdown'], my_mdd = calculate_drawdown(res['My_Equity'])
    res['Hold_Drawdown'], hold_mdd = calculate_drawdown(res['Hold_Equity'])

    # 保存路径
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    folder_name = f"result_603010_{CORE_TICKER}_{timestamp}"
    folder = os.path.join(RESULTS_BASE_DIR, folder_name)
    os.makedirs(folder, exist_ok=True)

    # 写入 Report.txt (保留原格式并升级)
    with open(os.path.join(folder, "report.txt"), "w", encoding="utf-8") as f:
        final_my = res['My_Equity'].iloc[-1]
        final_hold = res['Hold_Equity'].iloc[-1]
        summary = f"""
============================================================
📊 60/30/10 组合回测报告: {CORE_TICKER} 系列
策略: 60% QQQ死拿 | 30% QLD均线 | 10% TQQQ均线
信号: QQQ收盘 > SMA200*1.04 进场 | < SMA200*0.97 出场
------------------------------------------------------------
{'指标':<15} | {'组合策略':<15} | {'纯QQQ死拿'}
------------------------------------------------------------
{'最终资产':<15} | ${final_my:<15,.0f} | ${final_hold:,.0f}
{'收益率':<15} | {(final_my/INITIAL_CASH-1)*100:.0f}%            | {(final_hold/INITIAL_CASH-1)*100:.0f}%
{'最大回撤':<15} | {my_mdd*100:.2f}%           | {hold_mdd*100:.2f}%
============================================================
"""
        f.write(summary)
        print(summary)

        f.write("\n📋 杠杆部分(40%仓位) 交易明细:\n")
        f.write(f"{'日期':<12} | {'动作':<8} | {'QQQ信号':<10} | {'SMA':<10} | {'QLD成交':<10} | {'TQQQ成交':<10} | {'总净值'}\n")
        f.write("-" * 100 + "\n")
        for t in trades:
            f.write(f"{t['date'].strftime('%Y-%m-%d')} | {t['action']:<8} | {t['q_sig']:<10.2f} | {t['q_sma']:<10.2f} | {t['ld_p']:<10.2f} | {t['tq_p']:<10.2f} | ${t['equity']:,.0f}\n")

    # 5. 绘图 (保持三联图风格)
    fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(14, 12), gridspec_kw={'height_ratios': [2, 1, 1]}, sharex=True)

    ax1.plot(res.index, res['My_Equity'], color='#d62728', label='60/30/10 策略组合', linewidth=2)
    ax1.plot(res.index, res['Hold_Equity'], color='gray', alpha=0.5, linestyle='--', label='100% QQQ 死拿')
    ax1.set_yscale('log')
    ax1.set_title('组合资金曲线 (对数坐标)', fontsize=14)
    ax1.legend(loc='upper left')
    ax1.grid(True, alpha=0.3)
    ax1.yaxis.set_major_formatter('${x:,.0f}')

    # 画买卖点
    buy_dates = [t['date'] for t in trades if '进场' in t['action']]
    sell_dates = [t['date'] for t in trades if '撤离' in t['action']]
    if buy_dates: ax1.scatter(res.loc[buy_dates].index, res.loc[buy_dates]['My_Equity'], marker='^', color='green', s=100, zorder=5)
    if sell_dates: ax1.scatter(res.loc[sell_dates].index, res.loc[sell_dates]['My_Equity'], marker='v', color='red', s=100, zorder=5)

    ax2.plot(df_q.index, df_q['Close'], color='black', alpha=0.4, label='QQQ 价格')
    ax2.plot(df_q.index, df_q['SMA'], color='blue', label='SMA 200')
    ax2.fill_between(df_q.index, df_q['SMA']*BUFFER_SELL, df_q['SMA']*BUFFER_BUY, color='blue', alpha=0.1, label='缓冲带')
    ax2.set_title('信号源监控: QQQ vs SMA200')
    ax2.legend(loc='upper left')

    ax3.fill_between(res.index, res['My_Drawdown'], 0, color='#d62728', alpha=0.3, label='组合回撤')
    ax3.plot(res.index, res['My_Drawdown'], color='#d62728')
    ax3.set_title('最大回撤对比')
    ax3.yaxis.set_major_formatter(mtick.PercentFormatter(1.0))
    ax3.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.savefig(os.path.join(folder, "chart.png"))
    print(f"✅ 回测报告与图表已保存至: {folder}")
    plt.show()

if __name__ == "__main__":
    run()

总结： 投资不是比谁更聪明，而是比谁的体系更健壮。牛市有爆破力，熊市有缓冲垫，这才是普通人能拿得住的终极答案。