python BACKTEST MULTI-STRATEGY（INCL DUAL MUMENTUM）
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python BACKTEST MULTI-STRATEGY（INCL DUAL MUMENTUM）相关的知识，希望对你有一定的参考价值。
import pandas_datareader.data as web
import matplotlib.pyplot as plt
import pandas as pd
from datetime import datetime, timezone
import itable
import ffn
import pytz
from fintools import get_DataArray,show_return_table, show_annual_returns, monthly_return_table
from fintools import Parameters,compute_weights_RS_DM,compute_weights_PMA, generate_orders
from fintools import endpoints, backtest
# %matplotlib inline
start = datetime(2000, 1, 1, 0, 0, 0, 0, pytz.utc)
end = datetime.today().replace(tzinfo=timezone.utc)
# CHOOSE EITHER ETFs OR FUNDS
# ETFs
# strategies = {
# 'RS0001': { 'assets': ['CWB','HYG','MBB','IEF','HYD'],
# 'start':start, 'end':end,
# 'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 2, 'frequency': 'M',
# 'cash_proxy': 'CASHX', 'risk_free': 0},
# 'RS0002': {'assets': ['HYD','QLTB','MBB'],
# 'start':start, 'end':end,
# 'rs_lookback': 3, 'risk_lookback': 2, 'n_top': 1, 'frequency': 'M',
# 'cash_proxy': 'CASHX', 'risk_free': 0},
# 'RS0003': {'assets': ['HYD','MBB','SHY'],
# 'start':start, 'end':end,
# 'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 1, 'frequency': 'Q',
# 'cash_proxy': 'CASHX', 'risk_free': 0},
# 'DM0001': {'assets': ['CWB','VYM','HYG','VHT','TLT','MBB','HYD','IEF'],
# 'start':start, 'end':end,
# 'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 3, 'frequency': 'M',
# 'cash_proxy': 'CASHX', 'risk_free': 'SHY'},
# 'DM0002': {'assets': ['CWB','TLT','HYG','VHT','VNQ'],
# 'start':start, 'end':end,
# 'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 5, 'frequency': 'M',
# 'cash_proxy': 'MBB', 'risk_free': 'SHY'},
# 'PMA001': {'assets': ['CWB','MBB'],
# 'start':start, 'end':end,
# 'risk_lookback': 3, 'frequency': 'M', 'allocations': [0.6, 0.4],
# 'cash_proxy': 'TLT'},
# 'PMA002': {'assets': ['CWB','VYM','HYG'],
# 'start':start, 'end':end,
# 'risk_lookback': 3, 'frequency': 'M', 'allocations': [0.6, 0.2, 0.2],
# 'cash_proxy': 'TLT'},
# 'PMA003': {'assets': ['VCVSX', 'FAGIX', 'VGHCX'],
# 'start':start, 'end':end,
# 'risk_lookback': 2, 'frequency': 'M', 'allocations': [1./3., 1./3., 1./3.],
# 'cash_proxy': 'VUSTX'}
# }
# FUNDS
strategies = {
'RS0001': { 'assets': ['VCVSX','VWEHX','VFIIX','FGOVX','VWAHX'],
'start':start, 'end':end,
'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 2, 'frequency': 'M',
'cash_proxy': 'CASHX', 'risk_free': 0},
'RS0002': {'assets': ['MMHYX','FAGIX','VFIIX'],
'start':start, 'end':end,
'rs_lookback': 3, 'risk_lookback': 2, 'n_top': 1, 'frequency': 'M',
'cash_proxy': 'CASHX', 'risk_free': 0},
'RS0003': {'assets': ['MMHYX','FAGIX','VFIIX'],
'start':start, 'end':end,
'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 1, 'frequency': 'Q',
'cash_proxy': 'CASHX', 'risk_free': 0},
'DM0001': {'assets': ['VCVSX','VWINX','VWEHX','VGHCX','VUSTX','VFIIX','VWAHX','FGOVX','FFXSX'],
'start':start, 'end':end,
'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 3, 'frequency': 'M',
'cash_proxy': 'CASHX', 'risk_free': 'FFXSX'},
'DM0002': {'assets': ['VCVSX','VUSTX','VWEHX','VFIIX','VGHCX','FRESX'],
'start':start, 'end':end,
'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 5, 'frequency': 'M',
'cash_proxy': 'VFIIX', 'risk_free': 'FFXSX'},
'PMA001': {'assets': ['VCVSX', 'VFIIX'],
'start':start, 'end':end,
'risk_lookback': 3, 'frequency': 'M', 'allocations': [0.6, 0.4],
'cash_proxy': 'VUSTX'},
'PMA002': {'assets': ['VCVSX', 'VWINX', 'VWEHX'],
'start':start, 'end':end,
'risk_lookback': 3, 'frequency': 'M', 'allocations': [0.6, 0.2, 0.2],
'cash_proxy': 'VUSTX'},
'PMA003': {'assets': ['VCVSX', 'FAGIX', 'VGHCX'],
'start':start, 'end':end,
'risk_lookback': 2, 'frequency': 'M', 'allocations': [1./3., 1./3., 1./3.],
'cash_proxy': 'VUSTX'}
}
strategy_values = {}
security_weights = {}
security_holdings = {}
prices = {}
for name in strategies:
if 'PMA' in name:
try:
s_value, s_holdings, s_weights, s_prices = compute_weights_PMA(name, strategies[name])
except:
print('*** Unable to use PMA strategy : ' + name)
name = 'None'
else:
try:
s_value, s_holdings, s_weights, s_prices = compute_weights_RS_DM(name, strategies[name])
except:
print('*** Unable to use RS_DM strategy : ' + name)
name = 'None'
if name != 'None':
strategy_values[name] = s_value
security_weights[name] = s_weights
security_holdings[name] = s_holdings
prices[name] = s_prices
columns = list(strategy_values.keys())
# strategy_values[columns].plot(figsize=(15,10), grid=True, legend=True)
strategy_values = pd.DataFrame(strategy_values).dropna()
index = strategy_values.index
rebalance_dates = endpoints(period='M', trading_days=index)
# find the set of all portfolio symbols
n = len(columns)
l = [list(security_weights[name].columns) for name in columns]
s = []
for i in range(n):
s = s + l[i]
aggregated_weights = pd.DataFrame(0, index=rebalance_dates, columns=list(set(s)))
all_prices = pd.DataFrame(0, index=index, columns=list(set(s)))
# for equally weighted strategies
strategy_weights = pd.Series([1. / n for i in range(n)], index=columns)
for name in columns :
aggregated_weights[security_weights[name].columns] += security_weights[name].loc[rebalance_dates] * strategy_weights[name]
all_prices = prices[name].loc[index].combine_first(all_prices)
p_value, p_holdings, p_weights = backtest(all_prices, aggregated_weights, 10000., offset=0, commission=10.)
# p_value.plot(figsize=(15,10), grid=True, legend=True)
# algo stats
ffn.calc_perf_stats(p_value).display()
show_return_table(p_value)
show_annual_returns(p_value)
# HOLDINGS, TRANSACTIONS AND ORDERS
holdings = p_holdings.loc[rebalance_dates].round(0)
transactions = (p_holdings - p_holdings.shift(1).fillna(0))
transactions = transactions[transactions.sum(1) != 0].round(0)
orders = generate_orders(transactions, all_prices)##################################################
# DUAL MOMENTUM
###############
strategy_prices = strategy_values.dropna().copy()
# need to add prices for cash_proxy and, if necessary, risk_free
d = get_DataArray(['FFXSX'],start,end).to_pandas().transpose(1,2,0)
strategy_prices['FFXSX'] = d[:,:,'adj close'].loc[s_prices.index]
strategies1 = {
'MUTLTI-RS': { 'assets': list(strategies.keys()), 'prices': strategy_prices,
'rs_lookback': 1, 'risk_lookback': 1, 'n_top': 8, 'frequency': 'm',
'cash_proxy': 'FFXSX', 'risk_free': 0}}
for name in strategies1 :
s_value, s_holdings, s_weights, s_prices = compute_weights_RS_DM (name, strategies1[name])
# poorer return but lower drawdown and better SR
ffn.calc_perf_stats(s_value).display()
########################################################################################################
# This to calculate the orders at each rebalance date
#####################################################◙
# get weights from backtest
strategy_weights = s_weights.loc[rebalance_dates].copy()
aggregated_weights = pd.DataFrame(0, index=rebalance_dates, columns=list(set(s + ['FFXSX'])))
all_prices = pd.DataFrame(0, index=index, columns=list(set(s + ['FFXSX'])))
prices = security_prices.copy()
for name in strategies:
aggregated_weights[security_weights[name].columns] += security_weights[name].loc[rebalance_dates].mul(
strategy_weights[name], axis=0)
all_prices = prices[name].loc[index].combine_first(all_prices)
# need to add in the cash_proxy weights
aggregated_weights['FFXSX'] = aggregated_weights['FFXSX'].add(strategy_weights['FFXSX'], axis=0)
aggregated_weights = aggregated_weights[aggregated_weights.sum(1) > 0]
values, holdings, weights = backtest(all_prices, aggregated_weights, 10000., offset=0, commission=10.)
transactions = (holdings - holdings.shift(1).fillna(0))
transactions = transactions[transactions.sum(1) != 0]
orders = generate_orders(transactions, all_prices)
以上是关于python BACKTEST MULTI-STRATEGY（INCL DUAL MUMENTUM）的主要内容，如果未能解决你的问题，请参考以下文章