基于Backtrader的量化回测系统开发与过拟合防范实战（附完整可视化py代码）

基于Backtrader的量化回测系统开发与过拟合防范实战（附完整可视化py代码）

一、量化回测入门：先搞懂这些核心概念

1.1 回测（Backtest）

• 是什么：用历史数据模拟策略交易，验证策略在过去的有效性。
• 目的：判断策略能否在历史行情中盈利，暴露潜在风险（如大幅亏损、波动过大）。

1.2 多资产回测（Multi-Asset Backtest）

• 特点：同时交易多个品种（如股票、期货、外汇等），分散单一资产风险。
• 关键数据：
  • Final Portfolio Value（最终组合价值）：回测结束后账户总资金，对比初始资金（假设初始100万）可判断整体盈亏。
  • Sharpe Ratio（夏普比率）：
    公式：（策略收益 - 无风险利率）/ 收益标准差
    意义：衡量承担单位风险获得的超额收益。
    • 正数：收益超过风险，值越大越好（>1较好，>2优秀）；
    • 负数：策略收益无法覆盖风险（如你数据中的-3.266，说明风险极高且亏损）。
  • Max Drawdown（最大回撤）：
    定义：历史最大亏损比例（从峰值到谷底的跌幅）。
    例子：若峰值100万，谷底70万，最大回撤30%。数值越小，策略抗跌性越强。
• 举个生活例子：
  你有100万，想买稳健的国债（低风险低波动）和激进的股票（高风险高波动）。如果按“风险平价策略”，会给国债多分点钱（因为它波动小，风险贡献低），股票少分点钱（波动大，风险贡献高），让两者对整体风险的影响差不多。

1.3 过拟合：回测时的“甜蜜陷阱”

假如你用过去10年数据开发了一个策略，发现“每年都能赚50%”，但实盘时却亏了20%——这就是过拟合。

• 原因：策略记住了历史数据中的“噪声”（比如某只股票偶然的大涨），而不是真正的规律。
• 类比：就像学生只背会了课本上的例题，遇到新题目就不会做了。

二、手把手搭建回测系统：从0到1实现核心功能

2.1 Backtrader框架：量化回测的“万能工具箱”

这是一个专门做回测的Python库，帮你处理数据、执行策略、计算绩效。

2.1.1 策略编写：让电脑学会“调仓”

class MultiAssetPortfolio(bt.Strategy):
    # 参数设置：30天再平衡一次，目标年化波动率15%
    params = (('rebalance_days', 30), ('risk_parity_vol', 0.15))
    
    def __init__(self):
        # 保存所有资产数据，比如['AAPL', 'BTC-USD']
        self.assets = self.datas  
        # 记录过了多少天，判断是否该调仓
        self.day_counter = 0      

    def next(self):
        self.day_counter += 1
        # 每30天执行一次调仓
        if self.day_counter % self.p.rebalance_days != 0: return
        
        # 计算每只资产最近30天的年化波动率（波动大=风险高）
        vols = []
        for data in self.assets:
            # 标准差*√252：把30天的波动换算成年化
            vol = np.std(data.close.get(size=30)) * np.sqrt(252)  
            vols.append(vol)
        vols = np.array(vols)
        
        # 风险平价权重：波动越小，权重越高（比如波动10%的资产，权重是波动20%资产的2倍）
        weights = (1/vols) / np.sum(1/vols)  
        
        # 调整持仓：让每只资产的价值等于目标权重*总资金
        for i, data in enumerate(self.assets):
            target_value = self.broker.getvalue() * weights[i]  # 目标价值
            current_value = self.broker.getvalue([data])  # 当前价值
            if target_value > current_value:  # 买不够，补仓
                self.buy(data, size=(target_value - current_value)/data.close[0])
            else:  # 买太多，减仓
                self.sell(data, size=(current_value - target_value)/data.close[0])

2.1.2 数据与交易设置：模拟真实交易环境

cerebro = bt.Cerebro()  # 创建回测引擎

# 1. 加载数据（假设你有AAPL.csv和BTC-USD.csv）
for symbol in ['AAPL', 'BTC-USD']:
    data = bt.feeds.PandasData(
        dataname=pd.read_csv(f'{symbol}.csv', parse_dates=True),  # 读取CSV数据
        name=symbol  # 给数据命名，方便识别
    )
    cerebro.adddata(data)  # 把数据加入引擎

# 2. 设置初始资金、手续费、滑点（模拟真实交易成本）
cerebro.broker.setcash(1000000)  # 初始100万
cerebro.broker.setcommission(commission=0.001, commtype=bt.CommInfoBase.COMM_PERC)  # 千分之一手续费
cerebro.broker.set_slippage_fixed(0.001)  # 滑点：成交价比预期差0.1%（比如想买100元，实际花100.1元）

2.2 自定义回测引擎：不依赖框架，自己控制流程

适合想深入理解回测逻辑的同学，比如手动处理订单成交、记录交易历史。

class Backtester:
    def __init__(self, data, initial_capital=1e6):
        self.data = data  # 数据格式：{symbol: DataFrame}，比如{'AAPL': df, 'BTC': df}
        self.capital = initial_capital  # 剩余现金
        self.positions = {}  # 持仓：{symbol: 数量}
        self.history = []  # 交易记录：每次买卖的详细信息

    def execute_order(self, symbol, qty, price, timestamp):
        # 模拟真实成交：可能只成交95%（滑点和流动性影响）
        filled_qty = int(qty * 0.95)  
        # 计算成本：价格*数量+手续费（0.03%）
        cost = filled_qty * price * (1 + 0.0003)  
        self.capital -= cost  # 扣除现金
        self.positions[symbol] = self.positions.get(symbol, 0) + filled_qty  # 更新持仓
        # 记录交易
        self.history.append({
            'timestamp': timestamp,
            'symbol': symbol,
            'qty': filled_qty,
            'price': price,
            'cost': cost
        })

    def run_backtest(self, strategy):
        # 收集所有数据的时间戳，按顺序处理
        all_timestamps = sorted(set(ts for df in self.data.values() for ts in df.index))  
        for ts in all_timestamps:
            # 获取当前所有资产的收盘价
            prices = {sym: df.loc[ts]['close'] for sym, df in self.data.items() if ts in df.index}  
            # 策略生成交易信号（比如“买100股AAPL”）
            signals = strategy.generate_signals(prices, self.positions, self.capital)  
            # 执行所有信号
            for sym, signal in signals.items():
                self.execute_order(sym, signal['qty'], prices[sym], ts)

三、过拟合防范：让策略经得起“实战”考验

3.1 Walk-Forward分析：用“时间旅行”验证策略

把历史数据分成多个“训练期+测试期”，比如：

• 先用2018-2020年训练策略，再用2020-2021年测试
• 然后用2019-2021年训练，2021-2022年测试
• ……以此类推

为什么有效？

如果策略在多个测试期都表现好（比如夏普比率稳定在1以上），说明它真的抓住了市场规律，而不是偶然拟合历史数据。

代码核心逻辑：

def walk_forward_analysis(data, train_months=24, test_months=6):
    results = []
    # 计算所有月份的边界（比如从2018-01到2023-12，每月一个点）
    month_dates = pd.date_range(start=data.index.min(), end=data.index.max(), freq='M')  
    
    for i in range(len(month_dates) - train_months - test_months + 1):
        # 划分训练集和测试集（按时间顺序，不随机划分）
        train_start = month_dates[i]
        train_end = month_dates[i + train_months]
        test_start = train_end
        test_end = month_dates[i + train_months + test_months]
        
        train_data = data[train_start:train_end]
        test_data = data[test_start:test_end]
        
        # 训练策略（比如用风险平价策略）
        model = train_strategy(train_data)  
        # 测试策略，得到夏普比率和最大回撤
        sharpe, max_dd = test_strategy(model, test_data)  
        results.append({
            'train_period': f'{train_start.date()} - {train_end.date()}',
            'test_period': f'{test_start.date()} - {test_end.date()}',
            'sharpe': sharpe,
            'max_dd': max_dd
        })
    return pd.DataFrame(results)

3.2 Lasso因子筛选：剔除“没用的因子”

如果你用100个指标（比如市盈率、成交量增长率等）选股，其中很多指标可能是“噪音”。Lasso回归能自动剔除对预测收益贡献小的指标，只保留真正有用的。

• 类比：
  就像老师批改试卷，给每个题目（因子）打分，把得分低（系数接近0）的题目删掉，只保留高分题目，这样复习效率更高。

• 代码实现：

from sklearn.linear_model import LassoCV

# 假设X是1000个样本×100个因子的矩阵，y是未来收益率
model = LassoCV(cv=5)  # 5折交叉验证，自动选最佳正则化强度
model.fit(X, y)  

# 找出系数不为0的因子（即被保留的有效因子）
selected_factors = np.where(model.coef_ != 0)[0]  
print(f"从100个因子中筛选出{len(selected_factors)}个有效因子")

四、图表怎么看？新手必学的分析技巧

4.1 Backtrader多资产回测可视化（4个子图）

（注：实际图可能因数据不同而变化，以下为通用解读）

子图1：净值曲线 + 基准对比（顶部蓝线+红线）

• 横轴：时间（日期，可能因数据格式问题显示不全，检查数据是否包含datetime索引）
• 蓝线（关键）：策略净值曲线（初始资金=100万，若上升表示盈利，下降表示亏损）
• 红线（可选）：基准指数（如标普500，代码中未显式添加，可能为默认对比线）
• 观察点：
  • 蓝线是否长期高于红线？→ 策略跑赢大盘
  • 是否有陡峭下跌？→ 对应“最大回撤”指标（代码中计算的Max Drawdown）

子图2-3：资产价格走势（中间K线图）

• 每子图对应1个资产（如AAPL、GOOG的日线K线）
• 柱子：K线（开盘-最高-最低-收盘）
• 黄色线：收盘价均线（默认50日均线，可在策略中自定义）
• 红绿三角（关键）：
  • 红色向下三角：卖出信号（策略执行self.sell()）
  • 绿色向上三角：买入信号（策略执行self.buy()）
• 你遇到的问题：
  • 第四部分无三角：可能该资产（如BTC-USD）在回测期内未触发交易信号（始终持有/未达调仓条件）
  • 横坐标缺失：右键点击图表→选择“Configure”→勾选“X-axis”显示日期

4.2 自定义回测引擎净值曲线（类似心电图的蓝线）

• 横轴：时间（按数据时间戳排序，可能为日线/分钟线）
• 纵轴：总权益（现金+持仓市值，初始100万）
• 曲线特征：
  • 平稳上升：策略稳健（如风险平价策略）
  • 剧烈波动：可能过度交易或滑点设置不合理
  • 长期横盘：交易成本过高（手续费吃掉收益）
• 结合代码看：
  • 来自performance_report函数，由plt.plot(df['cum_pnl'])生成
  • 图标题显示的“夏普比率”和“最大回撤”是核心指标：
    • 夏普>1：承担风险值得（类比：上班时薪>市场价）
    • 最大回撤<15%：抗跌性良好（类比：考试分数波动小）

4.3 Lasso正则化路径图（黑线+红虚线）

• 横轴（左到右）：正则化强度alpha（对数刻度，值越大→筛选越严格）
• 纵轴：均方误差（MSE，值越小→模型预测越准）
• 黑线：不同alpha下的MSE均值（5折交叉验证结果）
• 红色虚线：最佳alpha（通过LassoCV自动选择，此时模型复杂度最优）
• 关键判断：
  • 左侧（alpha小）：MSE低但保留很多因子（可能过拟合，类似“死记硬背所有知识”）
  • 右侧（alpha大）：MSE高但因子少（可能欠拟合，类似“只学皮毛”）
  • 红虚线位置：找到“MSE较低+因子较少”的平衡点（类似“学会核心考点”）
• 代码对应：

  plt.semilogx(model.alphas_, model.mse_path_.mean(axis=1), 'k')  # 黑线
  plt.axvline(model.alpha_, color='red', linestyle='--')  # 红虚线
  

五、回测数据分析指南

5.1 走步分析（Walk-Forward Analysis）

• 是什么：将数据分成多个「训练期+测试期」，逐段验证策略适应性（避免过拟合）。
  • 训练期：用历史数据优化策略参数；
  • 测试期：用未用过的新数据检验策略效果。
• 目的：观察策略在不同市场环境下的稳定性（如牛熊转换时是否失效）。

5.2 数据表格解读

由于采取的随机数据，每次图表不一样，本阶内容以其中1次回测数据为例。

1. 多资产回测结果

Running Multi-Asset Backtest with Backtrader...
Final Portfolio Value: 937,544.96 （初始假设100万，亏损6.25%）
Sharpe Ratio: -3.266 （风险远大于收益，策略不可取）
Max Drawdown: 27.18% （过程中最大亏损近28%，风险很高）

策略绩效分析:
夏普比率: -0.394 （比整体回测稍好，可能不同计算方式或细分资产表现）
最大回撤: -22.7% （细分后最大亏损略小，但仍需警惕）

• 结论：该策略在多资产组合中整体亏损，风险调整后收益极差，最大回撤接近30%，说明策略可能存在明显缺陷（如入场/离场逻辑错误、仓位管理不当）。

2. 走步分析表格

• 核心观察点：
  1. 夏普比率波动大：从-0.6到1.69再到-0.75，说明策略在不同时间段表现极不稳定（可能过度依赖特定市场环境）。
  2. 测试期亏损为主：多数区间夏普为负，仅第4、5行短暂盈利，说明策略泛化能力差（训练期拟合好，但无法适应新数据）。
  3. 最大回撤可控：多数区间回撤小于15%，但夏普持续为负，说明策略即使风险不高，也无法稳定盈利。

5.3 如何通过数据判断策略有效性（3步走）

1. 先看整体回测：是否“赚钱+低风险”？

• 赚钱：Final Portfolio Value > 初始资金（如第二次回测1,075,454.45 > 100万，盈利7.5%）。
• 低风险：
  • 夏普比率 持续为正（如第二次回测夏普0.493，优于第一次的负数）；
  • 最大回撤 小于20%（第一次27.18%偏高，第二次19.71%勉强可接受）。

2. 再看走步分析：是否“稳定适应市场”？

• 理想状态：
  • 多数测试期夏普 ≥0.5，且波动小（说明策略不依赖特定行情）；
  • 最大回撤在不同区间 保持一致（如始终＜10%，无突然放大的风险）。
• 你的数据问题：
  第一次走步分析中，夏普从-0.6到1.69再到-0.75，说明策略可能存在「过拟合」（在训练期过度优化，测试期失效）。例如第5行夏普1.69可能是偶然运气，而非真实能力。

3. 结合图表

• 第一张图（多资产组合分析）：
  • 上半部分：蓝线（资产净值曲线）、红线（基准指数/无风险利率），若蓝线长期低于红线，说明策略跑输大盘；
  • 下半部分：各资产价格走势（类似股票K线），无红绿三角可能表示无明确买卖信号（或信号在其他图中）。
• 第二张图（净值曲线细节）：
  • 心电图式蓝线：可能是策略每日净值波动，陡峭下跌对应最大回撤区间（如27.18%时的暴跌段）。
• 第三张图（走步分析结果）：
  • 黑线：各测试期夏普比率趋势（从左到右爬升可能表示后期优化有效）；
  • 红色虚线：关键时间点（如策略参数调整日、市场重大事件日），观察虚线两侧夏普是否突变。

六、初学者实操建议（从数据到决策）

6.1 策略改进方向

• 若夏普为负：
  • 检查入场逻辑（是否频繁追涨杀跌）；
  • 增加止损/止盈规则（降低最大回撤）；
  • 分散资产相关性（避免多资产同时暴跌）。
• 若走步分析波动大：
  • 延长训练期（获取更多历史数据）；
  • 简化策略参数（减少过度拟合可能）；
  • 加入风险控制指标（如波动率过滤）。

6.2 图表辅助判断

• 净值曲线：理想状态是「稳步上升，回撤浅且快速修复」，避免“大起大落”；
• 夏普比率趋势：在走步分析图中，若多数测试期夏普在0轴上方，且波动区间小（如0.3-0.8），策略更可靠；
• 最大回撤标注：在净值曲线中标注回撤区域，查看是否因特定事件（如股灾）导致，判断是策略问题还是市场极端情况。

6.3 数据对比技巧

• 对比两次回测结果：第二次Final Portfolio Value更高（107万 vs 93万），夏普转正（0.493 vs -3.266），说明调整策略后有改善，但仍需优化（如最大回撤19.71%仍较高）；
• 关注走步分析中的“异常值”：如第一次走步第5行夏普1.69，需检查对应测试期是否有特殊行情（如某资产单边上涨），避免被偶然好成绩误导。

6.4 从数据到决策的核心逻辑

1. 先看“能不能赚钱”：最终组合价值是否增长，夏普比率是否为正；
2. 再看“会不会亏钱”：最大回撤是否在承受范围内（如散户通常接受＜20%）；
3. 最后看“稳不稳定”：走步分析中各测试期表现是否一致，避免依赖特定行情的“运气策略”。

七、从回测到实盘：新手必知的3个注意事项

7.1 先在模拟盘跑3个月

比如用米筐、FMZ等平台，观察：

• 实际成交价格是否和回测一致（验证滑点模型）
• 手续费是否正确扣除（避免回测时低估成本）

7.2 渐进式资金投入

• 初始投入5%资金（比如实盘100万，先投5万）
• 如果3个月内最大回撤<2%，再加仓5%
• 以此类推，直到达到目标仓位（比如50%）

7.3 数据质量是核心

• 股票数据：需要前复权（处理分红、拆股）
• 期货数据：要连续合约（避免换月跳空）
• 加密货币数据：检查交易所是否可靠（避免异常成交记录）

总结：新手也能掌握的量化回测三步法

1. 搭系统：用Backtrader快速实现多资产策略，用自定义引擎理解底层逻辑
2. 防过拟合：用Walk-Forward看稳定性，用Lasso筛选因子
3. 看图表：净值曲线看收益，滚动夏普看稳健性，回撤分布看风险

通过以上步骤，即使看不懂图表细节，也能通过关键数据快速判断策略的优缺点，进而针对性优化。量化交易的核心不是追求“完美策略”，而是找到风险与收益的平衡点，让数据成为决策的“指南针。
通过这套系统，你可以像“数据科学家”一样，用历史数据验证想法，再一步步走向实盘。记住：回测不是终点，而是量化投资的起点！

完整代码

import matplotlib
import matplotlib.pyplot as plt
import os
# 消除Qt警告
os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = ""
from sklearn.linear_model import LassoCV
from datetime import datetime, timedelta
import matplotlib.font_manager as fm

# 设置 matplotlib 中文字体
plt.rcParams["font.sans-serif"] = ["Microsoft YaHei"]  # 使用微软雅黑
plt.rcParams["axes.unicode_minus"] = True  # 正确显示负号

# 设置 matplotlib 中文字体
plt.rcParams["font.sans-serif"] = ["Microsoft YaHei"]  # 使用微软雅黑
plt.rcParams["axes.unicode_minus"] = True  # 正确显示负号


# ================
# 多资产组合回测策略
# ================
class MultiAssetPortfolio(bt.Strategy):
    params = (
        ("rebalance_days", 30),  # 每月再平衡
        ("risk_parity_vol", 0.15),  # 目标年化波动率
    )

    def __init__(self):
        self.assets = self.datas
        self.symbols = [d._name for d in self.assets]
        self.day_counter = 0

    def next(self):
        self.day_counter += 1
        if self.day_counter % self.p.rebalance_days != 0:
            return

        # 计算波动率调整权重
        vols = np.array(
            [np.std(d.close.get(size=30)) * np.sqrt(252) for d in self.assets]
        )
        weights = (1 / vols) / sum(1 / vols)

        # 执行再平衡
        for i, d in enumerate(self.assets):
            target_value = self.broker.getvalue() * weights[i]
            current_value = self.broker.getvalue([d])
            delta = target_value - current_value
            if delta > 0:
                self.buy(d, size=delta / d.close[0])
            else:
                self.sell(d, size=-delta / d.close[0])


# =================
# 自定义回测引擎
# =================
class Backtester:
    def __init__(self, data, initial_capital=1e6):
        self.data = data  # Dict格式：{symbol: DataFrame}
        self.positions = {}  # 当前持仓
        self.capital = initial_capital
        self.history = []  # 交易记录
        self.equity_curve = []  # 净值曲线
        self.current_date = None

    def execute_order(self, symbol, qty, price, timestamp):
        # 处理部分成交与滑点
        filled_qty = int(qty * 0.95)  # 假设95%成交率
        cost = filled_qty * price * (1 + 0.0003)  # 包含手续费
        self.capital -= cost
        self.positions[symbol] = self.positions.get(symbol, 0) + filled_qty
        self.history.append(
            {
                "timestamp": timestamp,
                "symbol": symbol,
                "qty": filled_qty,
                "price": price,
                "cost": cost,
            }
        )

    def calculate_equity(self, prices):
        """计算当前总权益"""
        position_value = sum(
            [self.positions.get(sym, 0) * prices.get(sym, 0) for sym in self.positions]
        )
        return self.capital + position_value

    def run_backtest(self, strategy):
        """运行回测"""
        # 按时间戳统一遍历
        all_timestamps = sorted(set(ts for df in self.data.values() for ts in df.index))

        for ts in all_timestamps:
            self.current_date = ts
            # 获取当前所有资产价格
            prices = {
                sym: df.loc[ts]["close"]
                for sym, df in self.data.items()
                if ts in df.index
            }

            # 记录净值曲线
            equity = self.calculate_equity(prices)
            self.equity_curve.append({"timestamp": ts, "equity": equity})

            # 执行策略逻辑
            signals = strategy.generate_signals(prices, self.positions, self.capital)

            # 处理订单
            for sym, signal in signals.items():
                if sym in prices:  # 确保有价格数据
                    self.execute_order(sym, signal["qty"], prices[sym], ts)


# ================
# 简单测试策略（用于自定义回测引擎）
# ================
class SimpleMomentumStrategy:
    def __init__(self, lookback=20, threshold=0.05):
        self.lookback = lookback
        self.threshold = threshold
        self.price_history = {}

    def generate_signals(self, prices, positions, capital):
        signals = {}

        # 更新价格历史
        for sym, price in prices.items():
            if sym not in self.price_history:
                self.price_history[sym] = []
            self.price_history[sym].append(price)
            # 保持历史长度一致
            if len(self.price_history[sym]) > self.lookback:
                self.price_history[sym] = self.price_history[sym][-self.lookback :]

        # 生成信号
        for sym, price_list in self.price_history.items():
            if len(price_list) < self.lookback:
                continue

            # 计算动量
            momentum = price_list[-1] / price_list[0] - 1

            # 持仓量
            current_position = positions.get(sym, 0)

            # 基于动量生成信号
            if momentum > self.threshold and current_position <= 0:
                # 买入信号 - 分配20%资金
                target_value = capital * 0.2
                signals[sym] = {"qty": int(target_value / price_list[-1])}
            elif momentum < -self.threshold and current_position >= 0:
                # 卖出信号 - 清空持仓
                signals[sym] = {"qty": -current_position}

        return signals


# ================
# Walk-Forward分析
# ================
def walk_forward_analysis(data, strategy_func, train_months=24, test_months=6):
    """
    执行Walk-Forward分析

    参数:
    data: 完整数据集
    strategy_func: 策略函数，输入训练数据，返回训练好的策略
    train_months: 训练窗口长度(月)
    test_months: 测试窗口长度(月)
    """
    start_date = data.index.min()
    end_date = data.index.max()
    current_date = start_date + pd.DateOffset(months=train_months)

    results = []
    while current_date + pd.DateOffset(months=test_months) <= end_date:
        # 划分数据集
        train_data = data.loc[start_date:current_date]
        test_data = data.loc[
            current_date : current_date + pd.DateOffset(months=test_months)
        ]

        # 策略训练与测试
        model = strategy_func(train_data)
        sharpe, max_dd = test_strategy(model, test_data)

        results.append(
            {
                "train_start": start_date,
                "train_end": current_date,
                "test_start": current_date,
                "test_end": current_date + pd.DateOffset(months=test_months),
                "sharpe": sharpe,
                "max_drawdown": max_dd,
            }
        )

        # 滚动窗口
        start_date += pd.DateOffset(months=test_months)
        current_date += pd.DateOffset(months=test_months)

    return pd.DataFrame(results)


def test_strategy(model, data):
    """测试策略表现，返回夏普比率和最大回撤"""
    # 这里简化处理，实际应根据模型类型和数据格式具体实现
    returns = np.random.randn(len(data)) * 0.01  # 示例随机收益
    sharpe = returns.mean() / returns.std() * np.sqrt(252)

    # 计算最大回撤
    cumulative = (1 + returns).cumprod()
    running_max = np.maximum.accumulate(cumulative)
    drawdown = (cumulative / running_max) - 1
    max_dd = drawdown.min()

    return sharpe, max_dd


# ================
# Lasso回归因子筛选
# ================
def lasso_factor_selection(X, y, alphas=None):
    """使用LassoCV进行因子筛选"""
    if alphas is None:
        alphas = np.logspace(-4, 0, 50)

    # Lasso交叉验证
    model = LassoCV(cv=5, alphas=alphas)
    model.fit(X, y)

    # 筛选有效因子
    selected_factors = np.where(model.coef_ != 0)[0]
    print(f"Selected {len(selected_factors)} factors out of {X.shape[1]}")

    return model, selected_factors


# ================
# 绩效统计
# ================
def performance_report(history, initial_capital):
    """生成绩效报告"""
    if not history:
        print("无交易记录可供分析(No trading history to analyze).")
        return None

    df = pd.DataFrame(history)
    if "timestamp" not in df.columns:
        print("无效的交易记录格式(Invalid history format). 缺少 'timestamp' 列.")
        return None

    df["date"] = pd.to_datetime(df["timestamp"])
    df = df.set_index("date")

    # 计算每日净值
    daily_equity = df.groupby(df.index.date)["cost"].sum().cumsum() + initial_capital

    # 如果没有交易记录，使用初始资金
    if daily_equity.empty:
        daily_equity = pd.Series([initial_capital], index=[df.index[0].date()])

    # 计算收益率
    returns = daily_equity.pct_change().dropna()

    # 关键指标计算
    if len(returns) > 1:
        sharpe = returns.mean() / returns.std() * np.sqrt(252)
    else:
        sharpe = 0

    cumulative = (1 + returns).cumprod()
    running_max = np.maximum.accumulate(cumulative)
    drawdown = (cumulative / running_max) - 1
    max_drawdown = drawdown.min() if len(drawdown) > 0 else 0

    # 可视化
    plt.figure(figsize=(12, 6))
    plt.plot(daily_equity, label="策略净值曲线(Strategy Equity Curve)")
    plt.title(
        f"净值曲线 - 夏普比率(Sharpe): {sharpe:.2f} | 最大回撤(Max DD): {max_drawdown:.1%}"
    )
    plt.legend()
    plt.grid(alpha=0.3)
    plt.tight_layout()
    plt.show()

    print("策略绩效分析(Strategy Performance Analysis):")
    print(f"夏普比率(Sharpe Ratio): {sharpe:.3f}")
    print(f"最大回撤(Max Drawdown): {max_drawdown:.1%}")
    print(f"策略净值曲线(Strategy Equity Curve)已生成图表")

    return {
        "sharpe_ratio": sharpe,
        "max_drawdown": max_drawdown,
        "returns": returns,
        "equity_curve": daily_equity,
    }


# 修改 Walk-Forward 分析的可视化部分
def visualize_walk_forward_results(walk_forward_results):
    """可视化 Walk-Forward 分析结果"""
    plt.figure(figsize=(14, 6))

    # 夏普比率趋势
    plt.subplot(121)
    plt.plot(walk_forward_results["sharpe"], marker="o", color="#2ca02c")
    plt.axhline(y=1.0, color="red", linestyle="--")
    plt.title("滚动夏普比率趋势(Rolling Sharpe Ratio Trend)")
    plt.xlabel("窗口序号(Window Index)")
    plt.ylabel("夏普比率(Sharpe Ratio)")
    plt.grid(alpha=0.3)

    # 最大回撤分布
    plt.subplot(122)
    plt.hist(walk_forward_results["max_drawdown"], bins=20, color="#d62728", alpha=0.7)
    plt.title("最大回撤分布(Max Drawdown Distribution)")
    plt.xlabel("回撤幅度(Drawdown Ratio)")
    plt.ylabel("出现频率(Frequency)")
    plt.grid(alpha=0.3)

    plt.tight_layout()
    plt.show()


# 修改 Lasso 回归的图表标题
def visualize_lasso_path(model):
    """可视化 Lasso 正则化路径"""
    plt.figure(figsize=(10, 6))
    plt.semilogx(model.alphas_, model.mse_path_.mean(axis=1), "k")
    plt.axvline(model.alpha_, color="red", linestyle="--")
    plt.title("Lasso正则化路径(Lasso Regularization Path)")
    plt.xlabel("正则化强度(Alpha)")
    plt.ylabel("均方误差(MSE)")
    plt.grid(alpha=0.3)
    plt.show()


# ================
# 主函数 - 运行示例
# ================
def main():
    """运行回测示例"""

    # 创建示例数据
    def create_sample_data(symbol, start_date, periods=365, volatility=0.02):
        """创建示例价格数据"""
        dates = pd.date_range(start=start_date, periods=periods)
        returns = np.random.normal(0, volatility, periods)
        prices = 100 * (1 + returns).cumprod()

        return pd.DataFrame(
            {
                "date": dates,
                "open": prices * 0.99,
                "high": prices * 1.02,
                "low": prices * 0.98,
                "close": prices,
                "volume": np.random.randint(1000, 100000, periods),
            }
        ).set_index("date")

    # 1. 使用Backtrader进行多资产回测
    print("Running Multi-Asset Backtest with Backtrader...")
    cerebro = bt.Cerebro()
    cerebro.broker.setcash(1_000_000)
    cerebro.broker.set_slippage_fixed(0.001)  # 设置滑点

    # 添加示例数据
    symbols = ["AAPL", "GOOG", "BTC-USD"]
    for symbol in symbols:
        data = bt.feeds.PandasData(
            dataname=create_sample_data(symbol, "2020-01-01", 730),
            datetime=None,
            open=0,
            high=1,
            low=2,
            close=3,
            volume=4,
            openinterest=-1,
        )
        cerebro.adddata(data, name=symbol)

    # 设置手续费
    cerebro.broker.setcommission(
        commission=0.001,  # 基础费率
        mult=1.0,  # 合约乘数
        margin=None,  # 保证金要求
        commtype=bt.CommInfoBase.COMM_PERC,
        stocklike=True,
    )

    # 添加策略和分析器
    cerebro.addstrategy(MultiAssetPortfolio)
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")

    # 运行回测
    results = cerebro.run()
    strat = results[0]

    # 打印结果
    print(f"Final Portfolio Value: {cerebro.broker.getvalue():,.2f}")
    print(f"Sharpe Ratio: {strat.analyzers.sharpe.get_analysis()['sharperatio']:.3f}")
    print(
        f"Max Drawdown: {strat.analyzers.drawdown.get_analysis()['max']['drawdown']:.2f}%"
    )

    # 绘制结果
    cerebro.plot()

    # 2. 使用自定义回测引擎
    print("\nRunning Custom Backtester...")
    data = {
        "AAPL": create_sample_data("AAPL", "2020-01-01", 365),
        "GOOG": create_sample_data("GOOG", "2020-01-01", 365),
        "TSLA": create_sample_data("TSLA", "2020-01-01", 365),
    }

    backtester = Backtester(data, initial_capital=1e6)
    strategy = SimpleMomentumStrategy(lookback=20, threshold=0.05)
    backtester.run_backtest(strategy)

    # 生成绩效报告
    performance_report(backtester.history, 1e6)

    # 3. Walk-Forward分析示例
    print("\nRunning Walk-Forward Analysis...")
    # 创建更长期的示例数据
    long_data = create_sample_data("SPY", "2018-01-01", 1095)

    # 定义一个简单的策略训练函数
    def train_sample_strategy(train_data):
        # 这里只是示例，实际应训练一个真正的策略
        class SampleStrategy:
            def generate_signals(self, prices, positions, capital):
                signals = {}
                for sym, price in prices.items():
                    # 简单随机信号
                    if np.random.random() > 0.5:
                        signals[sym] = {"qty": int(capital * 0.1 / price)}
                    elif sym in positions:
                        signals[sym] = {"qty": -positions[sym]}
                return signals

        return SampleStrategy()

    # 执行Walk-Forward分析
    wf_results = walk_forward_analysis(
        long_data, train_sample_strategy, train_months=12, test_months=3
    )
    print(wf_results)

    # 4. Lasso因子筛选示例
    print("\nRunning Lasso Factor Selection...")
    # 创建示例数据 - 100个因子，1000个样本
    X = np.random.randn(1000, 100)
    # 假设只有前10个因子有真实效应
    y = (
        2 * X[:, 0]
        + 1.5 * X[:, 1]
        + X[:, 2]
        + 0.5 * X[:, 3]
        + 0.3 * X[:, 4]
        + 0.2 * X[:, 5]
        + 0.1 * X[:, 6]
        + 0.05 * X[:, 7]
        + 0.02 * X[:, 8]
        + 0.01 * X[:, 9]
        + np.random.randn(1000) * 0.1
    )

    model, selected = lasso_factor_selection(X, y)

    # 可视化正则化路径
    visualize_lasso_path(model)


if __name__ == "__main__":
    main()