三位双子座人格就失败的波动突破策略展开辩论
由量化分析师、风险经理和执行团队共同调整的高贝塔股票波动率突破策略的回测结果显示，该策略在某些方面优于SPY，但在其他方面仍落后于SPY。
几乎所有拥有 Python 脚本和市场数据接口的量化分析师都会被市场波动性突破所吸引。高贝塔值股票往往会持续上涨，直到突然反转，而仅仅一个下午就可能导致账户亏损。那些以某个指标——通常是夏普比率——为目标的优化算法会不断调整参数，使得历史数据中的随机噪声看起来也像是超额收益的来源。但现实市场可不会理会那些过度拟合的网格搜索方法。
传统的参数调整方式也将风险视为事后才需要考虑的问题，只是在最后设定一个固定的止损点或仓位规模。这种做法行不通，因为创造超额收益与控制风险需要完全不同的思维方式。量化分析师希望充分把握市场的动量趋势，而经验丰富的风险经理则希望在波动性骤升之前就采取防范措施。
该实验试图通过三个 Gemini 角色之间的多智能体辩论来连接这两者：一名量化分析师、一名风险经理和一名执行策略师。他们以15只高贝塔值股票加上SPY作为基准，对一种波动率突破策略进行批评、修改与验证。他们并不总是一致意见，这种分歧是刻意设计的。通过结合不同优先级角色的参数建议，可使结果更加稳定。
经过三轮结构化辩论后，实验在此处结束：
{
"strategy_version": "v2.0-optimized",
"final_sharpe_ratio": -0.6822300207528283,
"consensus_rationale": "Iteration 1 consensus achieved superior Sharpe ratio through optimized ATR multiplier and volume threshold refinement.",
"max_drawdown_delta": 0.00687485272186061,
"total_iterations_run": 3
}
初始阶段的夏普比率低至-2.68，且累计回报为负，情况十分糟糕。通过相关分析，该策略才得以摆脱困境，同时也能与更广泛的市场状况进行对比。由于构建该策略需要谨慎处理数据并使用模块化回测工具，因此介绍将从市场范围和历史数据开始。
数据与市场范围
高贝塔值股票群体
仅凭社交媒体上的热度来挑选个别科技股并非可行的方法，而建立结构化的股票群体才是起点。本次测试采用了15只高贝塔值股票，并以SPY作为对照。高贝塔值股票能够提供突破策略所需的显著价格波动，而SPY则作为基准，帮助区分真正的阿尔法收益与由贝塔效应带来的噪音。
在实现任何突破性逻辑之前，必须先清晰地定义导入的库和全局配置。这些导入包括用于数学计算的pandas和numpy、用于获取价格的yfinance，以及用于生成角色的谷歌Gen AI SDK。将所有设置集中管理在一起，可以方便地重复后续实验。
import os
import json
import logging
from datetime import datetime, timezone
import pandas as pd
import numpy as np
import yfinance as yf
import matplotlib.pyplot as plt
from google import genai
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s')
缺失数据与多索引对齐
金融数据往往十分混乱。Yahoo Finance 可能会返回多索引列，或是因停牌、上市日期不同或公司行为而产生的零散 NaN 值。若将不对齐的 DataFrame 输入到拆分回测工具中，仓位计算结果就会出现偏差，从而导致投资组合回报数据失真。
MarketDataLoader 类通过封装下载过程来解决这一问题。它设置 auto_adjust=True 以修正因股票分割和分红带来的价格偏差，采用向前填充法处理数据，并安全地剔除因不同 IPO 日期导致的开头 NaN 行，从而保持数据的结构一致性。
class MarketDataLoader:
def __init__(self, tickers: list[str], benchmark: str):
self.tickers = tickers
self.all_symbols = tickers + [benchmark]
self.benchmark = benchmark
def fetch_data(self) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
end_date = datetime.now(timezone.utc).strftime('%Y-%m-%d')
start_date = (datetime.now(timezone.utc) -
pd.Timedelta(days=365 * 3)).strftime('%Y-%m-%d')
logging.info(
f"Fetching historical data from {start_date} to {end_date} for {len(self.all_symbols)} symbols.")
data = yf.download(self.all_symbols, start=start_date,
end=end_date, auto_adjust=True, progress=False)
if data.empty:
logging.error(
f"Failed to download data for symbols: {self.all_symbols}")
raise ValueError(
f"Failed to download data for symbols: {self.all_symbols}")
if isinstance(data.columns, pd.MultiIndex):
close = data['Close'] if 'Close' in data.columns.levels[0] else data.xs(
'Close', level=1, axis=1)
high = data['High'] if 'High' in data.columns.levels[0] else data.xs(
'High', level=1, axis=1)
low = data['Low'] if 'Low' in data.columns.levels[0] else data.xs(
'Low', level=1, axis=1)
volume = data['Volume'] if 'Volume' in data.columns.levels[0] else data.xs(
'Volume', level=1, axis=1)
else:
close = data[['Close']]
high = data[['High']]
low = data[['Low']]
volume = data[['Volume']]
close = close.ffill().dropna(how='all')
high = high.ffill().dropna(how='all')
low = low.ffill().dropna(how='all')
volume = volume.ffill().fillna(0.0)
if close.empty:
logging.error(
"Cleaned price data DataFrame is empty after processing.")
raise ValueError(
"Cleaned price data DataFrame is empty after processing.")
logging.info(
f"Data checkpoint - Shape: {close.shape}, Column sample: {list(close.columns[:3])}, NaN count: {close.isna().sum().sum()}, Date range: {close.index[0]} to {close.index[-1]}")
return close, high, low, volume
当管道运行时，加载器会获取所有十六种符号三年的每日数据，并在进行任何数学运算之前确认数据对齐情况：
INFO - Fetching historical data from 2023-08-10 to 2026-08-09 for 16 symbols.
INFO - Data checkpoint - Shape: (751, 16), Column sample: ['AMD', 'COIN', 'DKNG'], NaN count: 0, Date range: 2023-08-10 00:00:00 to 2026-08-07 00:00:00
日志显示，从2023年8月到2026年8月，16只标的共有751个交易日，且没有NaN值出现。由于数据面板已同步并经过验证，现在可以将注意力从数据工程转向策略层面。
突破引擎
动量与波动性的结合
突破行情之所以吸引散户交易者，是因为它们能带来即时的涨势。但实际上，许多高贝塔值标的的突破信号都是虚假波动，在行情反转前就会消耗流动性。仅凭价格突破阈值并不足以作为依据，还需要成交量增加和波动性上升来加以确认。
因此，一个可靠的回测工具必须考察价格与滚动成交量及波动率区间的相互作用。当价格突破滚动高点，同时成交量也超过其移动平均值时，该信号才具有实际意义，否则就只是噪音而已。
模拟投资组合收益
VolatilityStrategyBacktester 类负责实现这一逻辑。它从数据加载器中获取整理好的价格和成交量数据，计算各类指标，并根据回溯窗口时长和波动率阈值等可配置参数来模拟每日投资组合的收益情况。下方的实现代码会对多标的投资组合进行模拟，且每次运行之间不会出现状态泄露。
class VolatilityStrategyBacktester:
def __init__(self, close_df: pd.DataFrame, high_df: pd.DataFrame, low_df: pd.DataFrame, volume_df: pd.DataFrame, benchmark: str):
self.close_df = close_df
self.high_df = high_df
self.low_df = low_df
self.volume_df = volume_df
self.benchmark = benchmark
self.asset_columns = [c for c in close_df.columns if c != benchmark]
def backtest(self, params: dict) -> tuple[pd.Series, pd.Series]:
lookback = int(params.get("lookback", 20))
atr_mult = float(params.get("atr_multiplier", 2.0))
vol_thresh = float(params.get("volume_threshold", 1.5))
asset_returns_dict = {}
for ticker in self.asset_columns:
if ticker not in self.close_df.columns:
continue
c = self.close_df[ticker]
h = self.high_df[ticker]
l = self.low_df[ticker]
v = self.volume_df[ticker]
tr1 = h - l
tr2 = (h - c.shift(1)).abs()
tr3 = (l - c.shift(1)).abs()
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
atr = tr.rolling(window=lookback).mean()
rolling_high = c.rolling(window=lookback).max()
avg_vol = v.rolling(window=lookback).mean()
breakout = (c > rolling_high.shift(1) + atr_mult *
atr) & (v > vol_thresh * avg_vol)
asset_ret = c.pct_change()
strat_ret = asset_ret * breakout.shift(1).fillna(False)
trades = breakout & (~breakout.shift(1).fillna(False))
strat_ret = strat_ret - (trades.astype(float) * 0.001)
asset_returns_dict[ticker] = strat_ret
strategy_df = pd.DataFrame(asset_returns_dict)
portfolio_returns = strategy_df.mean(axis=1).fillna(0.0)
benchmark_returns = self.close_df[self.benchmark].pct_change().fillna(
0.0)
return portfolio_returns, benchmark_returns
在没有任何智能体参与的情况下，使用未调整的参数对15只股票和SPY进行测试，得到的结果相当糟糕：
INFO - Baseline Metrics: Sharpe=-2.68, Return=-2.79%
负2.68的夏普比率和2.79%的累计负回报，充分体现了高贝塔值投资组合中的静态参数会多快毁掉一个账户。在过于简单的网格搜索使情况变得更糟之后，正是辩论架构旨在解决这类问题。
角色设计与辩论循环
将模型作为推理工具而非预言机器
许多在量化金融领域应用大语言模型的尝试都失败了，因为人们将其当作能够预测明日价格的预言机器，而这其实是快速损失资本的途径。在此方案中，模型作为结构化优化循环中的语言推理工具运行，它仅用于判断参数的敏感度，而不尝试预测价格。
配置类包含了达到Google Gen AI SDK所需的基本超参数以及模型标识符：
class Config:
TICKERS = [
"TSLA", "NVDA", "AMD", "PLTR", "COIN",
"MARA", "RIOT", "MSTR", "HOOD", "ROKU",
"UBER", "SHOP", "SNAP", "PTON", "DKNG"
]
BENCHMARK = "SPY"
MAX_ITERATIONS = 3
INITIAL_PARAMS = {
"lookback": 20,
"atr_multiplier": 2.0,
"volume_threshold": 1.5,
"stop_loss_pct": 0.05
}
为使讨论更具成效，设计了三种专门的角色。量化分析师负责监测阿尔法衰变及阈值敏感度；风险经理则关注回撤限制与波动性指标；执行策略师则着眼于交易频率及滑点问题。
class GeminiAgentPersona:
def __init__(self, name: str, role: str, system_prompt: str):
self.name = name
self.role = role
self.system_prompt = system_prompt
self.client = genai.Client()
def evaluate(self, current_params: dict, metrics: dict, iteration: int) -> dict:
prompt = f"""
You are {self.name}, a specialized {self.role} in a multi-agent quantitative trading system.
Current Iteration: {iteration}
Current Strategy Parameters: {json.dumps(current_params, indent=2)}
Current Backtest Performance Metrics: {json.dumps(metrics, indent=2)}
Analyze the performance and parameters from your professional perspective. Provide your critique and propose specific parameter adjustments (lookback, atr_multiplier, volume_threshold, stop_loss_pct) to improve the Sharpe ratio and reduce max drawdown.
You MUST return a valid JSON object with keys:
- "critique": Detailed analytical critique from your persona's perspective.
- "proposed_adjustment": Dictionary of recommended parameter changes (e.g., ).
- "confidence_score": Float between 0 and 1 representing your confidence in this adjustment.
"""
response = self.client.models.generate_content(
model="gemini-3.5-flash-lite",
contents=prompt
)
text = response.text
clean_text = text.strip()
if clean_text.startswith("```json"):
clean_text = clean_text[7:]
if clean_text.endswith("```"):
clean_text = clean_text[:-3]
try:
parsed = json.loads(clean_text.strip())
except json.JSONDecodeError:
parsed = {
"critique": text[:500],
"proposed_adjustment": current_params,
"confidence_score": 0.6
}
return parsed
辩论协调器
当各角色相遇时，它们并非仅仅交换意见。PersonaDebateCoordinator会运行一个有序的批判循环：它获取上一次迭代的回测结果，将这些指标传递给每个角色，然后收集包含批判意见及参数调整建议的结构化JSON回复。
为防止模型设定过大的持仓规模或过长的回溯窗口，协调器会实施严格的参数限制。同时它还会对各角色的建议取平均值，从而平滑不同迭代之间的收敛过程。
class PersonaDebateCoordinator:
def __init__(self, backtester: VolatilityStrategyBacktester, initial_params: dict, max_iterations: int):
self.backtester = backtester
self.current_params = initial_params.copy()
self.max_iterations = max_iterations
self.quant = GeminiAgentPersona(
"QuantAnalyst",
"Quantitative Strategist",
"Focuses on alpha generation, indicator sensitivity, and statistical edge."
)
self.risk = GeminiAgentPersona(
"RiskManager",
"Risk Officer",
"Focuses on drawdowns, tail risk, position sizing, and volatility caps."
)
self.execution = GeminiAgentPersona(
"ExecutionStrategist",
"Execution Trading Specialist",
"Focuses on liquidity, volume confirmation, slippage, and practical execution friction."
)
self.audit_log = []
self.parameter_history = []
def run_iterative_loop(self) -> dict:
best_sharpe = -999.0
consensus_rationale = "Initial baseline strategy evaluation."
p_ret, b_ret = self.backtester.backtest(self.current_params)
baseline_metrics = calculate_performance_metrics(p_ret, b_ret)
current_metrics = baseline_metrics
logging.info(
f"Baseline Metrics: Sharpe={baseline_metrics['sharpe_ratio']:.2f}, Return={baseline_metrics['cumulative_return']*100:.2f}%")
for i in range(1, self.max_iterations + 1):
logging.debug(
f"--- Starting Debate Iteration {i}/{self.max_iterations} ---")
q_eval = self.quant.evaluate(
self.current_params, current_metrics, i)
r_eval = self.risk.evaluate(
self.current_params, current_metrics, i)
e_eval = self.execution.evaluate(
self.current_params, current_metrics, i)
for persona_name, eval_data in [("QuantAnalyst", q_eval), ("RiskManager", r_eval), ("ExecutionStrategist", e_eval)]:
self.audit_log.append({
"iteration_number": i,
"persona_name": persona_name,
"critique_text": eval_data.get("critique", ""),
"proposed_adjustment": eval_data.get("proposed_adjustment", {}),
"confidence_score": eval_data.get("confidence_score", 0.8)
})
q_adj = q_eval.get("proposed_adjustment", {})
r_adj = r_eval.get("proposed_adjustment", {})
e_adj = e_eval.get("proposed_adjustment", {})
new_lookback = int(np.mean([
float(q_adj.get("lookback",
self.current_params["lookback"])),
float(r_adj.get("lookback",
self.current_params["lookback"])),
float(e_adj.get("lookback",
self.current_params["lookback"]))
]))
new_atr = float(np.mean([
float(q_adj.get("atr_multiplier",
self.current_params["atr_multiplier"])),
float(r_adj.get("atr_multiplier",
self.current_params["atr_multiplier"])),
float(e_adj.get("atr_multiplier",
self.current_params["atr_multiplier"]))
]))
new_vol_thresh = float(np.mean([
float(q_adj.get("volume_threshold",
self.current_params["volume_threshold"])),
float(r_adj.get("volume_threshold",
self.current_params["volume_threshold"])),
float(e_adj.get("volume_threshold",
self.current_params["volume_threshold"]))
]))
new_sl = float(np.mean([
float(q_adj.get("stop_loss_pct",
self.current_params["stop_loss_pct"])),
float(r_adj.get("stop_loss_pct",
self.current_params["stop_loss_pct"])),
float(e_adj.get("stop_loss_pct",
self.current_params["stop_loss_pct"]))
]))
self.current_params = {
"lookback": max(5, min(60, new_lookback)),
"atr_multiplier": max(1.0, min(4.0, new_atr)),
"volume_threshold": max(1.0, min(3.0, new_vol_thresh)),
"stop_loss_pct": max(0.01, min(0.15, new_sl))
}
p_ret_new, b_ret_new = self.backtester.backtest(
self.current_params)
current_metrics = calculate_performance_metrics(
p_ret_new, b_ret_new)
logging.debug(
f"Iteration {i} Results: Sharpe={current_metrics['sharpe_ratio']:.2f}, Return={current_metrics['cumulative_return']*100:.2f}%, Params={self.current_params}")
self.parameter_history.append({
"iteration": i,
"params": self.current_params.copy(),
"metrics": current_metrics.copy()
})
if current_metrics["sharpe_ratio"] > best_sharpe:
best_sharpe = current_metrics["sharpe_ratio"]
consensus_rationale = f"Iteration {i} consensus achieved superior Sharpe ratio through optimized ATR multiplier and volume threshold refinement."
summary_result = {
"strategy_version": "v2.0-optimized",
"final_sharpe_ratio": float(best_sharpe),
"consensus_rationale": consensus_rationale,
"max_drawdown_delta": float(current_metrics["max_drawdown"] - baseline_metrics["max_drawdown"]),
"total_iterations_run": self.max_iterations,
"parameter_history": self.parameter_history
}
return summary_result
当协调员依次处理各项辩论并编写审计日志与性能总结时，优化流程即告完成：
INFO - Successfully generated output JSON files.
结果
逐次迭代
若没有客观的评估指标，回测就只是故事叙述，因此需要衡量各次迭代间的进展。性能指标函数可对收益的年化值及风险调整进行标准化处理。
def calculate_performance_metrics(portfolio_returns: pd.Series, benchmark_returns: pd.Series) -> dict:
valid_idx = portfolio_returns.index.intersection(benchmark_returns.index)
p_ret = portfolio_returns.loc[valid_idx]
b_ret = benchmark_returns.loc[valid_idx]
trading_days = 252
cum_return = (1 + p_ret).prod() - 1
annualized_return = (1 + cum_return) ** (trading_days /
len(p_ret)) - 1 if len(p_ret) > 0 else 0.0
annualized_vol = p_ret.std() * np.sqrt(trading_days)
risk_free_rate = 0.03
sharpe_ratio = (annualized_return - risk_free_rate) / \
annualized_vol if annualized_vol > 0 else 0.0
cum_wealth = (1 + p_ret).cumprod()
peak = cum_wealth.cummax()
drawdown = (cum_wealth - peak) / peak
max_drawdown = drawdown.min()
b_cum_return = (1 + b_ret).prod() - 1
outperformance = cum_return - b_cum_return
return {
"cumulative_return": float(cum_return),
"annualized_return": float(annualized_return),
"annualized_volatility": float(annualized_vol),
"sharpe_ratio": float(sharpe_ratio),
"max_drawdown": float(max_drawdown),
"benchmark_cumulative_return": float(b_cum_return),
"outperformance": float(outperformance)
}
在测试过程中，第1次迭代达到了最佳效果。三种角色设定之间的共识带来了正的累计收益以及更高的夏普比率，从而摆脱了基线模型的负面困境。辩论结果与参数变化记录如下：
{
"strategy_version": "v2.0-optimized",
"final_sharpe_ratio": -0.6822300207528283,
"consensus_rationale": "Iteration 1 consensus achieved superior Sharpe ratio through optimized ATR multiplier and volume threshold refinement.",
"parameter_history": [
{
"iteration": 1,
"params": {
"lookback": 14,
"atr_multiplier": 1.5,
"volume_threshold": 1.4666,
"stop_loss_pct": 0.025
},
"metrics": {
"cumulative_return": 0.0399,
"annualized_return": 0.0132,
"sharpe_ratio": -0.6822,
"max_drawdown": -0.0344
}
}
]
}
与基准对比
任何策略的收益表现都需与简单的买入并持有基准进行比较。多智能体优化确实扭转了基准的负收益状况，但在当前市场环境下，它面对SPY时表现极为不佳。高贝塔值的科技类投资组合走势波动剧烈且具有均值回归特性，导致趋势跟踪策略屡次出现错误判断。图表展示了该策略在不同迭代中的表现与基准的对比情况。
整个流程将数据加载、回测以及策略参数调整整合到了同一个入口点中：
def main():
logging.info(
"Starting Multi-Agent Volatility Breakout Optimization Pipeline.")
loader = MarketDataLoader(Config.TICKERS, Config.BENCHMARK)
close_df, high_df, low_df, volume_df = loader.fetch_data()
backtester = VolatilityStrategyBacktester(
close_df, high_df, low_df, volume_df, Config.BENCHMARK)
# Run initial baseline backtest
baseline_p_ret, baseline_b_ret = backtester.backtest(Config.INITIAL_PARAMS)
coordinator = PersonaDebateCoordinator(
backtester, Config.INITIAL_PARAMS, Config.MAX_ITERATIONS)
summary_results = coordinator.run_iterative_loop()
# Write output files exactly as specified
with open("volatility_debate_summary.json", "w") as f:
json.dump(summary_results, f, indent=4)
with open("iteration_debate_audit_log.json", "w") as f:
json.dump(coordinator.audit_log, f, indent=4)
logging.info("Successfully generated output JSON files.")
# Generate annotated plot
fig, ax = plt.subplots(figsize=(10, 6))
optimized_p_ret, _ = backtester.backtest(coordinator.current_params)
wealth_baseline = (1 + baseline_p_ret).cumprod()
wealth_optimized = (1 + optimized_p_ret).cumprod()
wealth_benchmark = (1 + baseline_b_ret).cumprod()
ax.plot(wealth_baseline.index, wealth_baseline,
label="Baseline Strategy", color="gray", linestyle="--")
ax.plot(wealth_optimized.index, wealth_optimized,
label="Optimized Multi-Agent Strategy", color="blue", linewidth=2)
ax.plot(wealth_benchmark.index, wealth_benchmark,
label="SPY Benchmark", color="orange", alpha=0.7)
# Annotate final peak / convergence
max_idx = wealth_optimized.idxmax()
max_val = wealth_optimized.max()
ax.annotate(f"Optimized Peak: {max_val:.2f}x",
xy=(max_idx, max_val),
xytext=(max_idx - pd.Timedelta(days=90), max_val * 0.9),
arrowprops=dict(facecolor='blue', shrink=0.05, width=1, headwidth=6))
ax.set_title("Multi-Agent Volatility Breakout Portfolio vs. SPY Benchmark",
fontsize=12, fontweight='bold')
ax.set_xlabel("Date", fontsize=10)
ax.set_ylabel("Growth of 1.00", fontsize=10)
ax.legend(loc="upper left")
ax.grid(True, linestyle=":", alpha=0.6)
plt.tight_layout()
plt.show()
logging.info("Successfully generated strategy performance plot.")
if __name__ == "__main__":
main()
脚本执行完毕后，它会写入审计日志及性能可视化数据：
INFO - Successfully generated strategy performance plot.
这些结果提醒我们，多智能体大型语言模型系统在参数调整和风险评估方面表现优异，但无法从横盘走势的市场中识别出宏观趋势。若要将这项研究应用到实际交易环境中，还需关注执行延迟和订单路由的限制。
实验揭示的内容
构建量化交易系统并非简单地将历史数据输入优化器并寄予希望。此次辩论形式类似于量化领域的会议室，由专门的Gemini角色负责审核策略的收益生成能力、回撤幅度以及执行过程中的摩擦因素。
该评估循环确实修复了基准值。在多次迭代过程中，夏普比率从-2.68上升至-0.68，虽然仍为负数，但累计收益已从-2.79%转为正值。通过参数平均处理与严格的限制机制，避免了模型陷入无约束的数学幻想，从而使每次迭代都紧密贴合实际的回测表现。
这些结果也带来了一堂令人谦逊的教训。即便有三种高效的策略模型在优化突破阈值，但在当前市场环境下，高贝塔策略仍远远落后于SPY买入并持有策略的基准表现。一旦趋势停滞、价格开始回撤至平均水平，动量策略就会失效。无论模型如何通宵分析，都无法创造出市场本身所不具备的宏观利好条件。各种讨论所能提升的只是策略的内部一致性，而外部市场环境依然遥不可及。
实现实盘交易的实用路线图应首先将工作分解为三个升级步骤：用来自经纪商API的实时WebSocket数据流替换历史数据面板，在回测工具中加入动态滑点与佣金模型，再引入第四种角色——宏观情绪分析模块，以便在波动性过高时自动切换为持有现金的策略。
市场不会为优雅的代码或巧妙的提示买单，它们只为帮助机构在持续的不确定性中生存下来而付费。