很多开发者在投资时容易走两个极端:要么稳如老狗(全仓 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()
总结: 投资不是比谁更聪明,而是比谁的体系更健壮。牛市有爆破力,熊市有缓冲垫,这才是普通人能拿得住的终极答案。
猜你喜欢
发表评论
电子邮件地址不会被公开。 必填项已用*标注