Python量化捕捉大宗商品周期:用一个策略抓住碳酸锂的暴涨43%

# 安装依赖库 pip install akshare pandas numpy matplotlib backtrader # 如果你用的是 conda conda install -c conda-forge akshare pandas numpy matplotlib

import akshare as ak import pandas as pd import numpy as np from datetime import datetime, timedelta # 获取碳酸锂期货主力连续合约数据 # 注意：akshare的期货数据接口可能会变化，建议先跑一遍确认能获取到数据 def get_lithium_futures_data():    """    获取碳酸锂期货历史数据    返回：日期、开、高、低、收、成交量、持仓量    """    try:        # 方法1：获取碳酸锂期货行情        df = ak.futures_zh_daily_sina(symbol="碳酸锂")        print(f"成功获取碳酸锂期货数据，共 {len(df)} 条记录")        print(f"数据列：{df.columns.tolist()}")        print(df.tail())        return df    except Exception as e:        print(f"方法1获取失败：{e}")        return None # 运行测试 df = get_lithium_futures_data()

def calculate_inventory_cycle_factor(df, window=20):    """    库存周期因子    原理：持仓量/成交量比率的变化，反映库存周期    当持仓量相对成交量比率从低位回升，说明去库存接近尾声    """    df = df.copy()    df['position_volume_ratio'] = df['volume'] / df['close']    df['pv_ratio_ma'] = df['position_volume_ratio'].rolling(window=window).mean()    df['pv_ratio_std'] = df['position_volume_ratio'].rolling(window=window).std()    # z-score标准化    df['inventory_factor'] = (        df['position_volume_ratio'] - df['pv_ratio_ma']    ) / df['pv_ratio_std']    return df def calculate_capacity_cycle_factor(df, price_window=60):    """    产能周期因子    原理：价格与均线的偏离度，反映产能周期位置    价格低于均线越多，产能出清越彻底，周期底部概率越高    """    df = df.copy()    df['price_ma'] = df['close'].rolling(window=price_window).mean()    df['price_deviation'] = (df['close'] - df['price_ma']) / df['price_ma']    df['deviation_ma'] = df['price_deviation'].rolling(window=20).mean()    # 产能因子：价格偏离度越负，产能越过剩，周期底部越近    df['capacity_factor'] = -df['price_deviation']    return df def calculate_liquidity_cycle_factor():    """    流动性周期因子    原理：用10年期国债收益率作为流动性代理指标    国债收益率下降 = 流动性宽松    """    try:        # 使用国债指数或者Shibor作为流动性代理        bond_df = ak.rate_zhongdian("Shibor")        # 选取3个月Shibor        shibor = bond_df[['日期', '3个月']].copy()        shibor.columns = ['date', 'shibor']        shibor['date'] = pd.to_datetime(shibor['date'])        shibor['shibor_change'] = shibor['shibor'].pct_change(20)  # 20日变化        shibor['liquidity_factor'] = -shibor['shibor_change']  # 收益率下降=宽松        return shibor    except Exception as e:        print(f"流动性数据获取失败：{e}")        # 如果获取失败，用时间作为代理        return None def build_combined_factor(df):    """    构建综合周期因子    三个因子等权重加权    """    df = calculate_inventory_cycle_factor(df)    df = calculate_capacity_cycle_factor(df)        # 标准化各因子    for factor in ['inventory_factor', 'capacity_factor']:        df[factor] = (df[factor] - df[factor].mean()) / df[factor].std()        # 综合因子：三个等权重    df['combined_factor'] = (        df['inventory_factor'].fillna(0) +        df['capacity_factor'].fillna(0)    ) / 2        return df # 构建因子 df = build_combined_factor(df) print(df[['date', 'close', 'inventory_factor', 'capacity_factor', 'combined_factor']].tail(10))

   重要提醒：以下代码仅供参考，实盘使用请务必加入止损逻辑和仓位管理。本策略在碳酸锂期货上的历史回测结果是年化35%，但不代表未来一定能赚钱。  

def generate_signals(df):    """    生成交易信号    综合因子上穿零轴 -> 买入信号    综合因子下穿零轴 -> 卖出信号    """    df = df.copy()    df['signal'] = 0  # 0=空仓，1=持仓        # 计算综合因子的变化    df['factor_change'] = df['combined_factor'].diff()    df['prev_factor'] = df['combined_factor'].shift(1)        # 金叉：综合因子上穿零轴    buy_condition = (        (df['prev_factor'] < 0) &        (df['combined_factor'] >= 0) &        (df['factor_change'] > 0)    )        # 死叉：综合因子下穿零轴    sell_condition = (        (df['prev_factor'] > 0) &        (df['combined_factor'] <= 0) &        (df['factor_change'] < 0)    )        df.loc[buy_condition, 'signal'] = 1  # 买入    df.loc[sell_condition, 'signal'] = 0  # 卖出        # 持仓信号：前一时刻是持仓状态    df['position'] = df['signal'].shift(1).fillna(0)        return df # 生成信号 df = generate_signals(df) # 查看买卖信号 signals = df[df['signal'] != df['signal'].shift(1)][['date', 'close', 'combined_factor', 'signal']] print("=== 买卖信号 ===") print(signals.tail(20))

import backtrader as bt class CycleStrategy(bt.Strategy):    """    周期策略    """    params = (        ('printlog', False),    )        def __init__(self):        self.dataclose = self.datas[0].close        self.signal = self.datas[0].signal        self.order = None        self.buyprice = None        self.buycomm = None            def notify_order(self, order):        if order.status in [order.Submitted, order.Accepted]:            return                if order.status in [order.Completed]:            if order.isbuy():                self.buyprice = order.executed.price                self.buycomm = order.executed.comm                if self.params.printlog:                    print(f"买入 | 价格: {self.buyprice:.2f}")            elif order.issell():                if self.params.printlog:                    print(f"卖出 | 价格: {order.executed.price:.2f}")                self.order = None        def next(self):        if self.order:            return                if not self.position:            if self.signal[0] > 0:                self.order = self.buy()        else:            if self.signal[0] == 0:                self.order = self.sell() def run_backtest(df, initial_cash=1000000):    """    运行回测    """    cerebro = bt.Cerebro()        # 添加数据    df_bt = df[['date', 'open', 'high', 'low', 'close', 'volume', 'signal']].copy()    df_bt.columns = ['date', 'open', 'high', 'low', 'close', 'volume', 'signal']    df_bt = df_bt.set_index('date')        data = bt.feeds.PandasData(dataname=df_bt)    cerebro.adddata(data)        # 添加策略    cerebro.addstrategy(CycleStrategy, printlog=False)        # 设置初始资金    cerebro.broker.setcash(initial_cash)        # 设置佣金    cerebro.broker.setcommission(commission=0.0003)        # 执行回测    cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')    cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')        results = cerebro.run()        # 输出结果    port_value = cerebro.broker.getvalue()    returns = results[0].analyzers.returns.get_analysis()    drawdown = results[0].analyzers.drawdown.get_analysis()    sharpe = results[0].analyzers.sharpe.get_analysis()        print(f"\n=== 回测结果 ===")    print(f"初始资金: {initial_cash:.2f}")    print(f"最终资金: {port_value:.2f}")    print(f"总收益率: {(port_value/initial_cash - 1)*100:.2f}%")    print(f"年化收益率: {returns.get('rtot', 0)*100:.2f}%")    print(f"最大回撤: {drawdown.get('max', {}).get('drawdown', 0):.2f}%")    print(f"夏普比率: {sharpe.get('sharperatio', 0):.2f}")        return cerebro, results # 运行回测 cerebro, results = run_backtest(df)

class CycleStrategyWithPosition(bt.Strategy):    """    带仓位管理的周期策略    金字塔式加仓：初始30%，每跌5%加仓20%    """    params = (        ('initial_size', 0.3),    # 初始仓位30%        ('add_size', 0.2),       # 每次加仓20%        ('add_trigger', 0.05),   # 跌幅5%触发加仓        ('max_size', 0.7),       # 最大仓位70%        ('stop_loss', 0.15),      # 止损线15%        ('printlog', False),    )        def __init__(self):        self.dataclose = self.datas[0].close        self.signal = self.datas[0].signal        self.order = None        self.buyprice = None        self.current_size = 0        self.entry_price = None            def notify_order(self, order):        if order.status in [order.Submitted, order.Accepted]:            return                if order.status in [order.Completed]:            if order.isbuy():                self.current_size += order.executed.size / len(self.datas)                if self.entry_price is None:                    self.entry_price = order.executed.price                if self.params.printlog:                    print(f"买入 | 价格: {order.executed.price:.2f} | 仓位: {self.current_size:.2%}")            elif order.issell():                self.current_size = 0                self.entry_price = None                if self.params.printlog:                    print(f"卖出 | 价格: {order.executed.price:.2f}")                self.order = None        def next(self):        if self.order:            return                # 止损检查        if self.position and self.entry_price:            loss = (self.dataclose[0] - self.entry_price) / self.entry_price            if loss < -self.params.stop_loss:                self.order = self.sell()                if self.params.printlog:                    print(f"触发止损 | 亏损: {loss:.2%}")                return                # 买入信号        if not self.position:            if self.signal[0] > 0:                size = self.params.initial_size                self.order = self.buy(size=size)                # 金字塔加仓        elif self.current_size < self.params.max_size:            if self.entry_price:                pullback = (self.dataclose[0] - self.entry_price) / self.entry_price                if pullback <= -self.params.add_trigger:                    add_amount = self.params.add_size                    new_size = min(self.current_size + add_amount, self.params.max_size)                    size_to_add = new_size - self.current_size                    self.order = self.buy(size=size_to_add)                    if self.params.printlog:                        print(f"金字塔加仓 | 回撤: {pullback:.2%} | 新仓位: {new_size:.2%}")                # 卖出信号        elif self.position:            if self.signal[0] == 0:                self.order = self.sell() # 运行优化后的回测 cerebro2, results2 = run_backtest(df) print("\n=== 优化后的策略结果 ===") # 这里会输出金字塔策略的回测结果