捕捉能源风险溢价

“该策略交易10种能源期货，使用七个基于风格的多空子投资组合，标准化信号，按信号值加权头寸，等权组合投资组合，并每月重新平衡以获得多元化回报。”

资产类别: 期货 | 地区: 美国 | 周期: 每月 | 市场: 大宗商品 | 关键词: 能源

I. 策略概要

该策略交易10种美国交易所上市的能源期货，包括原油、天然气和电力。前端合约持有至到期前一个月，然后展期至次近月合约。根据七种风格创建七个多空子投资组合：展期收益、对冲压力、投机压力、过往表现、价值、流动性和偏度。每种风格的信号（详见表1）在子投资组合内通过减去均值并除以标准差进行标准化。具有正信号的能源期货做多，具有负信号的做空，头寸按信号值加权。最终投资组合等权组合七个子投资组合，并每月重新平衡，确保风格多元化。

II. 策略合理性

首先，最终策略中使用的所有风格都被广泛认为是期货的成功预测因子。当投资者依赖于聚合各种投资风格信息的复合信号时，他能够更可靠地预测随后的价格变化，从而更有能力捕捉能源期货市场中存在的风险溢价。

最后，本研究的发现对各种稳健性检验都是稳健的；在考虑交易成本、集成投资组合的替代规范、数据窥探检验和经济子周期分析后，它们仍然成立。

III. 来源论文

Capturing Energy Risk Premia [点击查看论文]

• Adrian Fernandez-Perez、Ana-Maria Fuertes 和 Joëlle Miffre。都柏林大学学院 (UCD) – 银行与金融系。伦敦城市大学贝叶斯商学院。南特高等商学院

<摘要>

本文研究了能源期货风险溢价，可以通过多空投资组合来提取这些溢价，这些投资组合利用合约在各种特征或信号以及其整合方面的异质性。投资者可以通过利用与对冲者净头寸和展期收益特征相关的能源期货合约风险，分别每年赚取约8%和12%的可观溢价，这与对冲压力假说和储存理论的预测一致。同时利用各种信号进行风格整合，并采用替代加权方案，进一步提高了溢价。特别是，等权所有信号的风格整合投资组合表现最为出色。研究结果对交易成本、数据挖掘和子周期分析具有稳健性。

IV. 回测表现

年化回报	12.38%
波动率	13.75%
β值	0.053
夏普比率	0.9
索提诺比率	-0.164
最大回撤	-22.32%
胜率	16%

V. 完整的 Python 代码

from AlgorithmImports import *
import data_tools
from typing import Dict
#endregion
class HedgersEffectCommodities(QCAlgorithm):
    
    def Initialize(self):
        self.SetStartDate(2005, 1, 1)
        self.SetCash(100000)
 
        tickers:Dict[str, str] = {
            'CME_CL1' : Futures.Energies.CrudeOilWTI,   # Crude Oil Futures, Continuous Contract
            'CME_NG1' : Futures.Energies.NaturalGas,    # Natural Gas (Henry Hub) Physical Futures, Continuous Contract
            'ICE_O1'  : Futures.Energies.HeatingOil,    # Heating Oil Futures, Continuous Contract
            'CME_CU1' : Futures.Energies.Ethanol,       # Chicago Ethanol (Platts) Futures
        }
        # Weekly hedging pressure data.
        self.data:Dict[Symbol, data_tools.SymbolData] = {}
        self.futures_data:Dict[str, data_tools.FuturesData] = {}
        
        min_expiration_days:int = 0
        max_expiration_days:int = 360
        self.min_futures:int = 2
        self.leverage:int = 5
        self.total_portfolios:int = 7
        self.one_year_period:int = 252
        self.period:int = int(252 * 5.5)
        self.cot_period:int = 52
        self.volume_period:int = 42
        for qp_ticker, qc_ticker in tickers.items():
            # Add quantpedia back-adjusted data.
            security = self.AddData(data_tools.QuantpediaFutures, qp_ticker, Resolution.Daily)
            security.SetFeeModel(data_tools.CustomFeeModel())
            security.SetLeverage(self.leverage)
            qp_symbol:Symbol = security.Symbol
            
            cot_ticker:str = 'Q' + qp_ticker.split('_')[1][:-1]
            cot_symbol:Symbol = self.AddData(data_tools.CommitmentsOfTraders, cot_ticker, Resolution.Daily).Symbol
            # QC futures
            future:Future = self.AddFuture(qc_ticker, Resolution.Daily, dataNormalizationMode=DataNormalizationMode.Raw)
            future.SetFilter(timedelta(days=min_expiration_days), timedelta(days=max_expiration_days))
            future_ticker:str = future.Symbol.Value
            self.futures_data[future_ticker] = data_tools.FuturesData(self.one_year_period, self.volume_period)
            self.data[qp_symbol] = data_tools.SymbolData(cot_symbol, future_ticker,
                self.period, self.cot_period)
        self.recent_month:int = -1
        self.settings.daily_precise_end_time = False
        self.settings.minimum_order_margin_portfolio_percentage = 0.
    def FindAndUpdateContracts(self, futures_chain, ticker:str) -> None:
        near_contract:FuturesContract = None
        dist_contract:FuturesContract = None
        if ticker in futures_chain:
            contracts:list[:FuturesContract] = [contract for contract in futures_chain[ticker] if contract.Expiry.date() > self.Time.date()]
            if len(contracts) >= 2:
                contracts:list[:FuturesContract] = sorted(contracts, key=lambda x: x.Expiry, reverse=False)
                near_contract = contracts[0]
                dist_contract = contracts[1]
        self.futures_data[ticker].update_contracts(near_contract, dist_contract)
        
    def OnData(self, data):
        curr_date:datetime.date = self.Time.date()
        for qp_symbol, symbol_obj in self.data.items():
            # store daily price
            if qp_symbol in data and data[qp_symbol]:
                price:float = data[qp_symbol].Value
                symbol_obj.update_qp_prices(price)
            cot_symbol:Symbol = symbol_obj.cot_symbol
            if cot_symbol in data and data[cot_symbol]:
                speculator_long_count:float = data[cot_symbol].GetProperty('LARGE_SPECULATOR_LONG')
                speculator_short_count:float = data[cot_symbol].GetProperty('LARGE_SPECULATOR_SHORT')
                hedgers_long_count:float = data[cot_symbol].GetProperty('COMMERCIAL_HEDGER_LONG')
                hedgers_short_count:float = data[cot_symbol].GetProperty('COMMERCIAL_HEDGER_SHORT')
                
                if speculator_long_count != 0 and speculator_short_count != 0 and hedgers_long_count != 0 and hedgers_short_count != 0:
                    hedging_pressure_value:float = (hedgers_short_count - hedgers_long_count) / (hedgers_long_count + hedgers_short_count)
                    speculative_pressure_value:float = (speculator_long_count - speculator_short_count) / (speculator_long_count + speculator_short_count)
                    
                    symbol_obj.update_pressures(hedging_pressure_value, speculative_pressure_value)
         # daily update qc future data
        if data.FutureChains.Count > 0:
            for ticker, future_obj in self.futures_data.items():
                # check if near contract is expired or is not initialized
                if not future_obj.is_initialized() or \
                    (future_obj.is_initialized() and future_obj.near_contract.Expiry.date() == curr_date):
                    self.FindAndUpdateContracts(data.FutureChains, ticker)
                # update QC futures rolling return
                if future_obj.is_initialized():
                    near_c:FuturesContract = future_obj.near_contract
                    dist_c:FuturesContract = future_obj.distant_contract
                    if near_c.Symbol in data and data[near_c.Symbol] and dist_c.Symbol in data and data[dist_c.Symbol]:
                        near_price:float = data[near_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
                        dist_price:float = data[dist_c.Symbol].Value * self.Securities[ticker].SymbolProperties.PriceMagnifier
                        volume:float = near_c.Volume
                        if near_price != 0 and dist_price != 0 and volume != 0:
                            future_obj.update_prices_and_volumes(near_price, dist_price, volume)
        if self.recent_month == curr_date.month:
            return
        self.recent_month = curr_date.month
        self.Liquidate()
        last_qp_price_by_symbol:Dict[Symbol, float] = {}
        portfolios:list[Dict[Symbol, float]] = [
            {} for i in range(self.total_portfolios)
        ]
        for qp_symbol, symbol_obj in self.data.items():
            future_ticker:str = symbol_obj.future_ticker
            if any([self.securities[symbol].get_last_data() and self.time.date() > data_tools.LastDateHandler.get_last_update_date()[symbol] for symbol in [qp_symbol, symbol_obj.cot_symbol]]):
                self.liquidate()
                return
                
            if symbol_obj.qp_prices_ready() and symbol_obj.pressures_ready() and self.futures_data[future_ticker].prices_and_volumes_ready():
                latest_near_contract_price:float = self.futures_data[future_ticker].get_latest_near_contract_price()
                near_contract_volumes:list[float] = self.futures_data[future_ticker].get_near_contract_volumes()
                portfolios[0][qp_symbol] = self.futures_data[future_ticker].get_roll_yield()
                portfolios[1][qp_symbol] = symbol_obj.get_hedging_pressure()
                portfolios[2][qp_symbol] = symbol_obj.get_speculative_pressure()
                portfolios[3][qp_symbol] = symbol_obj.get_momentum(self.one_year_period)
                portfolios[4][qp_symbol] = symbol_obj.get_value(self.one_year_period, latest_near_contract_price)
                portfolios[5][qp_symbol] = symbol_obj.get_liquidity(near_contract_volumes)
                portfolios[6][qp_symbol] = symbol_obj.get_skewness(self.one_year_period)
                last_qp_price_by_symbol[qp_symbol] = symbol_obj.get_last_price()
        if len(portfolios[0]) < self.min_futures:
            return
        
        # trade execution
        for portfolio in portfolios:
            # signal mean and std
            portfolio_mean = np.mean([x for x in portfolio.values()])
            portfolio_std = np.std([x for x in portfolio.values()])
            
            # signal standardization
            portfolio = { x[0] : (x[1] - portfolio_mean) / portfolio_std for x in portfolio.items() }
            
            long_leg = [x for x in portfolio.items() if x[1] > 0]
            short_leg = [x for x in portfolio.items() if x[1] < 0]
            total_signal_long = sum(abs(x[1]) for x in long_leg)
            total_signal_short = sum(abs(x[1]) for x in short_leg)
            sub_portfolio_weight = self.Portfolio.TotalPortfolioValue / self.total_portfolios
            
            for symbol, signal in long_leg:
                symbol_quantity:float = np.floor((sub_portfolio_weight*(signal/total_signal_long)) / last_qp_price_by_symbol[symbol])
                self.MarketOrder(symbol, symbol_quantity)
            for symbol, signal in short_leg:
                symbol_quantity:float = np.floor((sub_portfolio_weight*(signal/total_signal_short)) / last_qp_price_by_symbol[symbol])
                self.MarketOrder(symbol, symbol_quantity)
# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters):
        fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))